CERBERUS/mtproto_proxy
1
0
Fork 0
mirror of https://github.com/seriyps/mtproto_proxy.git synced 2026-08-28 05:07:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Use "hut" for logger abstraction

Browse source
This commit is contained in:
Сергей Прохоров 2019-05-22 00:08:32 +02:00
parent c8d8edfb27
commit e2308a6e57
No known key found for this signature in database
GPG key ID: 1C570244E4EF3337
10 changed files with 71 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

2
rebar.config
View file

@ -1,8 +1,10 @@
% -*- mode: erlang -*-
{erl_opts, [debug_info,
{d, 'HUT_LAGER'},
{parse_transform, lager_transform}]}.
{deps, [{ranch, "1.7.0"},
{hut, "1.3.0"},
{lager, "3.6.3"},
{psq, {git, "https://github.com/eryx67/psq.git", {branch, "master"}}}
]}.

2
rebar.lock
View file

@ -1,5 +1,6 @@
{"1.1.0",
[{<<"goldrush">>,{pkg,<<"goldrush">>,<<"0.1.9">>},1},
{<<"hut">>,{pkg,<<"hut">>,<<"1.3.0">>},0},
{<<"lager">>,{pkg,<<"lager">>,<<"3.6.3">>},0},
{<<"psq">>,
{git,"https://github.com/eryx67/psq.git",
@ -9,6 +10,7 @@
[
{pkg_hash,[
{<<"goldrush">>, <<"F06E5D5F1277DA5C413E84D5A2924174182FB108DABB39D5EC548B27424CD106">>},
{<<"hut">>, <<"71F2F054E657C03F959CF1ACC43F436EA87580696528CA2A55C8AFB1B06C85E7">>},
{<<"lager">>, <<"FE78951D174616273F87F0DBC3374D1430B1952E5EFC4E1C995592D30A207294">>},
{<<"ranch">>, <<"9583F47160CA62AF7F8D5DB11454068EAA32B56EEADF984D4F46E61A076DF5F2">>}]}
].

15
src/mtp_config.erl
View file

@ -30,6 +30,8 @@
-type dc_id() :: integer().
-type netloc() :: {inet:ip4_address(), inet:port_number()}.
-include_lib("hut/include/hut.hrl").
-define(TAB, ?MODULE).
-define(IPS_KEY(DcId), {id, DcId}).
-define(IDS_KEY, dc_ids).
@ -142,7 +144,7 @@ handle_call(_Request, _From, State) ->
handle_cast(update, #state{timer = Timer} = State) ->
update(State, soft),
lager:info("Config updated"),
?log(info, "Config updated"),
Timer1 = gen_timeout:bump(
gen_timeout:reset(Timer)),
{noreply, State#state{timer = Timer1}}.
@ -151,7 +153,7 @@ handle_info(timeout, #state{timer = Timer} =State) ->
case gen_timeout:is_expired(Timer) of
true ->
update(State, soft),
lager:info("Config updated"),
?log(info, "Config updated"),
Timer1 = gen_timeout:bump(
gen_timeout:reset(Timer)),
{noreply, State#state{timer = Timer1}};
@ -174,9 +176,8 @@ update(#state{tab = Tab}, force) ->
update(State, _) ->
try update(State, force)
catch ?WITH_STACKTRACE(Class, Reason, Stack)
lager:error(
"Err updating proxy settings: ~s",
[lager:pr_stacktrace(Stack, {Class, Reason})])
?log(error, "Err updating proxy settings: ~s",
[lager:pr_stacktrace(Stack, {Class, Reason})]) %XXX lager-specific
end.
update_key(Tab) ->
@ -247,8 +248,8 @@ update_ip([Url | Fallbacks]) ->
{ok, _} = inet:parse_ipv4strict_address(IpStr), %assert
application:set_env(?APP, external_ip, IpStr)
catch ?WITH_STACKTRACE(Class, Reason, Stack)
lager:error("Failed to update IP with ~s service: ~s",
[Url, lager:pr_stacktrace(Stack, {Class, Reason})]),
?log(error, "Failed to update IP with ~s service: ~s",
[Url, lager:pr_stacktrace(Stack, {Class, Reason})]), %XXX - lager-specific
update_ip(Fallbacks)
end;
update_ip([]) ->

10
src/mtp_dc_pool.erl
View file

@ -28,6 +28,8 @@
terminate/2, code_change/3]).
-export_type([status/0]).
-include_lib("hut/include/hut.hrl").
-define(SERVER, ?MODULE).
-define(APP, mtproto_proxy).
-define(BURST_MAX, 10).
@ -181,14 +183,12 @@ handle_down(MonRef, Pid, Reason, #state{downstreams = Ds,
{Pid, DsM1} ->
Pending1 = lists:delete(Pid, Pending),
Ds1 = ds_remove(Pid, Ds),
lager:error("Downstream=~p is down. reason=~p",
[Pid, Reason]),
?log(error, "Downstream=~p is down. reason=~p", [Pid, Reason]),
St#state{pending_downstreams = Pending1,
downstreams = Ds1,
downstream_monitors = DsM1};
_ ->
lager:error("Unexpected DOWN. ref=~p, pid=~p, reason=~p",
[MonRef, Pid, Reason]),
?log(error, "Unexpected DOWN. ref=~p, pid=~p, reason=~p", [MonRef, Pid, Reason]),
St
end
end.
@ -289,7 +289,7 @@ ds_return(Pid, St) ->
{ok, St1} ->
St1;
undefined ->
lager:warning("Attempt to release unknown connection ~p", [Pid]),
?log(warning, "Attempt to release unknown connection ~p", [Pid]),
St
end.

28
src/mtp_down_conn.erl
View file

@ -27,6 +27,8 @@
terminate/2, code_change/3]).
-export_type([handle/0, upstream_opts/0]).
-include_lib("hut/include/hut.hrl").
-define(SERVER, ?MODULE).
-define(APP, mtproto_proxy).
-define(CONN_TIMEOUT, 10000).
@ -115,7 +117,7 @@ handle_call({set_config, Name, Value}, _From, State) ->
ok = inet:setopts(State#state.sock, [{buffer, Value}]),
{ok, OldSize};
_ ->
lager:warning("set_config ~p=~p ignored", [Name, Value]),
?log(warning, "set_config ~p=~p ignored", [Name, Value]),
ignored
end,
{reply, Result, State}.
@ -145,16 +147,16 @@ handle_info(do_connect, #state{dc_id = DcId} = State) ->
{ok, St1} = connect(DcId, State),
{noreply, St1}
catch ?WITH_STACKTRACE(Class, Reason, Stack)
lager:error("Down connect error: ~s",
[lager:pr_stacktrace(Stack, {Class, Reason})]),
?log(error, "Down connect error: ~s",
[lager:pr_stacktrace(Stack, {Class, Reason})]), %XXX lager-specific
erlang:send_after(300, self(), do_connect),
{noreply, State}
end.
terminate(_Reason, #state{upstreams = Ups}) ->
%% Should I do this or dc_pool? Maybe only when reason is 'normal'?
lager:warning("Downstream terminates with reason ~p; len(upstreams)=~p",
[_Reason, map_size(Ups)]),
?log(warning, "Downstream terminates with reason ~p; len(upstreams)=~p",
[_Reason, map_size(Ups)]),
Self = self(),
lists:foreach(
fun(Upstream) ->
@ -172,7 +174,7 @@ handle_send(Data, Upstream, #state{upstreams = Ups,
Packet = mtp_rpc:encode_packet({data, Data}, {UpstreamStatic, ProxyAddr}),
down_send(Packet, St);
_ ->
lager:warning("Upstream=~p not found", [Upstream]),
?log(warning, "Upstream=~p not found", [Upstream]),
{{error, unknown_upstream}, St}
end.
@ -185,7 +187,7 @@ handle_upstream_new(Upstream, Opts, #state{upstreams = Ups,
UpsStatic = {ConnId, iolist_to_binary(mtp_rpc:encode_ip_port(Ip, Port)), AdTag},
Ups1 = Ups#{Upstream => {UpsStatic, 0, 0}},
UpsRev1 = UpsRev#{ConnId => Upstream},
lager:debug("New upstream=~p conn_id=~p", [Upstream, ConnId]),
?log(debug, "New upstream=~p conn_id=~p", [Upstream, ConnId]),
St#state{upstreams = Ups1,
upstreams_rev = UpsRev1}.
@ -203,7 +205,7 @@ handle_upstream_closed(Upstream, #state{upstreams = Ups,
down_send(Packet, St2);
error ->
%% It happens when we get rpc_close_ext
lager:info("Unknown upstream ~p", [Upstream]),
?log(info, "Unknown upstream ~p", [Upstream]),
{ok, St}
end.
@ -253,7 +255,7 @@ handle_rpc({close_ext, ConnId}, St) ->
St2#state{upstreams = Ups1,
upstreams_rev = UpsRev1};
error ->
lager:warning("Unknown upstream ~p", [ConnId]),
?log(warning, "Unknown upstream ~p", [ConnId]),
St1
end;
handle_rpc({simple_ack, ConnId, Confirm}, S) ->
@ -261,7 +263,7 @@ handle_rpc({simple_ack, ConnId, Confirm}, S) ->
-spec down_send(iodata(), #state{}) -> {ok, #state{}}.
down_send(Packet, #state{sock = Sock, codec = Codec, dc_id = DcId} = St) ->
%% lager:debug("Up>Down: ~w", [Packet]),
%% ?log(debug, "Up>Down: ~w", [Packet]),
{Encoded, Codec1} = mtp_codec:encode_packet(Packet, Codec),
mtp_metric:rt(
[?APP, downstream_send_duration, seconds],
@ -285,7 +287,7 @@ up_send(Packet, ConnId, #state{upstreams_rev = UpsRev} = St) ->
St
end;
error ->
lager:warning("Unknown connection_id=~w", [ConnId]),
?log(warning, "Unknown connection_id=~w", [ConnId]),
%% WHY!!!?
%% ClosedPacket = mtp_rpc:encode_packet(remote_closed, ConnId),
%% {ok, St1} = down_send(ClosedPacket, St),
@ -393,7 +395,7 @@ connect(DcId, S) ->
mtp_metric:count_inc([?APP, out_connect_ok, total], 1,
#{labels => [DcId]}),
AddrStr = inet:ntoa(Host),
lager:info("~s:~p: TCP connected", [AddrStr, Port]),
?log(info, "~s:~p: TCP connected", [AddrStr, Port]),
down_handshake1(S#state{sock = Sock,
netloc = {Host, Port}});
{error, Reason} = Err ->
@ -489,7 +491,7 @@ down_handshake3(Pkt, #state{stage_state = PrevSenderPid, pool = Pool,
{handshake, _SenderPid, PeerPid} = mtp_rpc:decode_handshake(Pkt),
(PeerPid == PrevSenderPid) orelse error({wrong_sender_pid, PeerPid}),
ok = mtp_dc_pool:ack_connected(Pool, self()),
lager:info("~s:~w: handshake complete", [inet:ntoa(Addr), Port]),
?log(info, "~s:~w: handshake complete", [inet:ntoa(Addr), Port]),
{ok, S#state{stage = tunnel,
stage_state = undefined}}.

39
src/mtp_handler.erl
View file

@ -22,6 +22,8 @@
-type handle() :: pid().
-include_lib("hut/include/hut.hrl").
-define(MAX_SOCK_BUF_SIZE, 1024 * 50). % Decrease if CPU is cheaper than RAM
-define(MAX_UP_INIT_BUF_SIZE, 1024 * 1024). %1mb
@ -93,7 +95,7 @@ init({Socket, Transport, [Name, Secret, Tag]}) ->
mtp_metric:count_inc([?APP, in_connection, total], 1, #{labels => [Name]}),
case Transport:peername(Socket) of
{ok, {Ip, Port}} ->
lager:info("~s: new connection ~s:~p", [Name, inet:ntoa(Ip), Port]),
?log(info, "~s: new connection ~s:~p", [Name, inet:ntoa(Ip), Port]),
{TimeoutKey, TimeoutDefault} = state_timeout(init),
Timer = gen_timeout:new(
#{timeout => {env, ?APP, TimeoutKey, TimeoutDefault}}),
@ -110,7 +112,7 @@ init({Socket, Transport, [Name, Secret, Tag]}) ->
{ok, State};
{error, Reason} ->
mtp_metric:count_inc([?APP, in_connection_closed, total], 1, #{labels => [Name]}),
lager:info("Can't read peername: ~p", [Reason]),
?log(info, "Can't read peername: ~p", [Reason]),
{stop, error}
end.
@ -124,14 +126,14 @@ handle_cast({proxy_ans, Down, Data}, #state{down = Down} = S) ->
ok = mtp_down_conn:ack(Down, 1, iolist_size(Data)),
maybe_check_health(bump_timer(S1));
handle_cast({close_ext, Down}, #state{down = Down, sock = USock, transport = UTrans} = S) ->
lager:debug("asked to close connection by downstream"),
?log(debug, "asked to close connection by downstream"),
ok = UTrans:close(USock),
{stop, normal, S#state{down = undefined}};
handle_cast({simple_ack, Down, Confirm}, #state{down = Down} = S) ->
lager:info("Simple ack: ~p, ~p", [Down, Confirm]),
?log(info, "Simple ack: ~p, ~p", [Down, Confirm]),
{noreply, S};
handle_cast(Other, State) ->
lager:warning("Unexpected msg ~p", [Other]),
?log(warning, "Unexpected msg ~p", [Other]),
{noreply, State}.
handle_info({tcp, Sock, Data}, #state{sock = Sock, transport = Transport,
@ -146,18 +148,18 @@ handle_info({tcp, Sock, Data}, #state{sock = Sock, transport = Transport,
%% Consider checking health here as well
{noreply, bump_timer(S1)};
{error, Reason} ->
lager:info("handle_data error ~p", [Reason]),
?log(info, "handle_data error ~p", [Reason]),
{stop, normal, S}
catch error:{protocol_error, Type, Extra} ->
mtp_metric:count_inc([?APP, protocol_error, total], 1, #{labels => [Type]}),
lager:warning("protocol_error ~p ~p", [Type, Extra]),
?log(warning, "protocol_error ~p ~p", [Type, Extra]),
{stop, normal, maybe_close_down(S)}
end;
handle_info({tcp_closed, Sock}, #state{sock = Sock} = S) ->
lager:debug("upstream sock closed"),
?log(debug, "upstream sock closed"),
{stop, normal, maybe_close_down(S)};
handle_info({tcp_error, Sock, Reason}, #state{sock = Sock} = S) ->
lager:warning("upstream sock error: ~p", [Reason]),
?log(warning, "upstream sock error: ~p", [Reason]),
{stop, normal, maybe_close_down(S)};
handle_info(timeout, #state{timer = Timer, timer_state = TState, listener = Listener} = S) ->
@ -165,7 +167,7 @@ handle_info(timeout, #state{timer = Timer, timer_state = TState, listener = List
true when TState == stop;
TState == init ->
mtp_metric:count_inc([?APP, inactive_timeout, total], 1, #{labels => [Listener]}),
lager:info("inactive timeout in state ~p", [TState]),
?log(info, "inactive timeout in state ~p", [TState]),
{stop, normal, S};
true when TState == hibernate ->
mtp_metric:count_inc([?APP, inactive_hibernate, total], 1, #{labels => [Listener]}),
@ -175,7 +177,7 @@ handle_info(timeout, #state{timer = Timer, timer_state = TState, listener = List
{noreply, S#state{timer = Timer1}}
end;
handle_info(Other, S) ->
lager:warning("Unexpected msg ~p", [Other]),
?log(warning, "Unexpected msg ~p", [Other]),
{noreply, S}.
terminate(_Reason, #state{started_at = Started, listener = Listener} = S) ->
@ -185,7 +187,7 @@ terminate(_Reason, #state{started_at = Started, listener = Listener} = S) ->
mtp_metric:histogram_observe(
[?APP, session_lifetime, seconds],
erlang:convert_time_unit(Lifetime, millisecond, native), #{labels => [Listener]}),
lager:info("terminate ~p", [_Reason]),
?log(info, "terminate ~p", [_Reason]),
ok.
code_change(_OldVsn, State, _Extra) ->
@ -280,7 +282,7 @@ up_send(Packet, #state{stage = tunnel,
sock = Sock,
transport = Transport,
listener = Listener} = S) ->
%% lager:debug(">Up: ~p", [Packet]),
%% ?log(debug, ">Up: ~p", [Packet]),
{Encoded, UpCodec1} = mtp_codec:encode_packet(Packet, UpCodec),
mtp_metric:rt([?APP, upstream_send_duration, seconds],
fun() ->
@ -295,14 +297,14 @@ up_send(Packet, #state{stage = tunnel,
mtp_metric:count_inc(
[?APP, upstream_send_error, total], 1,
#{labels => [Listener, Reason]}),
lager:warning("Upstream send error: ~p", [Reason]),
?log(warning, "Upstream send error: ~p", [Reason]),
throw({stop, normal, S})
end
end, #{labels => [Listener]}),
{ok, S#state{codec = UpCodec1}}.
down_send(Packet, #state{down = Down} = S) ->
%% lager:debug(">Down: ~p", [Packet]),
%% ?log(debug, ">Down: ~p", [Packet]),
case mtp_down_conn:send(Down, Packet) of
ok ->
{ok, S};
@ -316,7 +318,7 @@ handle_unknown_upstream(#state{down = Down, sock = USock, transport = UTrans} =
ok = UTrans:close(USock),
receive
{'$gen_cast', {close_ext, Down}} ->
lager:debug("asked to close connection by downstream"),
?log(debug, "asked to close connection by downstream"),
throw({stop, normal, S#state{down = undefined}})
after 0 ->
throw({stop, got_unknown_upstream, S})
@ -370,7 +372,7 @@ check_health() ->
do_check_health([{qlen, Limit} | _], #{message_queue_len := QLen} = Health) when QLen > Limit ->
mtp_metric:count_inc([?APP, healthcheck, total], 1,
#{labels => [message_queue_len]}),
lager:warning("Upstream too large queue_len=~w, health=~p", [QLen, Health]),
?log(warning, "Upstream too large queue_len=~w, health=~p", [QLen, Health]),
overflow;
do_check_health([{gc, Limit} | Other], #{total_mem := TotalMem}) when TotalMem > Limit ->
%% Maybe it doesn't makes sense to do GC if queue len is more than, eg, 50?
@ -383,8 +385,7 @@ do_check_health([{total_mem, Limit} | _Other], #{total_mem := TotalMem} = Health
TotalMem > Limit ->
mtp_metric:count_inc([?APP, healthcheck, total], 1,
#{labels => [total_memory]}),
lager:warning("Process too large total_mem=~p, health=~p",
[TotalMem / 1024, Health]),
?log(warning, "Process too large total_mem=~p, health=~p", [TotalMem / 1024, Health]),
overflow;
do_check_health([_Ok | Other], Health) ->
do_check_health(Other, Health);

3
src/mtp_session_storage.erl
View file

@ -23,6 +23,7 @@
terminate/2, code_change/3]).
-include_lib("stdlib/include/ms_transform.hrl").
-include_lib("hut/include/hut.hrl").
-define(DATA_TAB, ?MODULE).
-define(HISTOGRAM_TAB, mtp_session_storage_histogram).
@ -108,7 +109,7 @@ handle_info(timeout, #state{data_tab = DataTab, histogram_tab = HistTab, clean_t
#{max_age_minutes => 360}),
Cleans = clean_storage(DataTab, HistTab, Opts),
Remaining = ets:info(DataTab, size),
lager:info("storage cleaned: ~p; remaining: ~p", [Cleans, Remaining]),
?log(info, "storage cleaned: ~p; remaining: ~p", [Cleans, Remaining]),
gen_timeout:bump(gen_timeout:reset(Timer0));
false ->
gen_timeout:reset(Timer0)

9
src/mtproto_proxy_app.erl
View file

@ -10,8 +10,11 @@
%% Application callbacks
-export([start/2, prep_stop/1, stop/1, config_change/3]).
-export([mtp_listeners/0, running_ports/0, start_proxy/1]).
-define(APP, mtproto_proxy).
-include_lib("hut/include/hut.hrl").
-type proxy_port() :: #{name := any(),
port := inet:port_number(),
secret := binary(),
@ -134,15 +137,15 @@ config_changed(Action, ports, Ports) when Action == new; Action == changed ->
ok;
config_changed(Action, K, V) ->
%% Most of the other config options are applied automatically without extra work
lager:info("Config ~p ~p to ~p ignored", [K, Action, V]),
?log(info, "Config ~p ~p to ~p ignored", [K, Action, V]),
ok.
-ifdef(TEST).
report(Fmt, Args) ->
lager:debug(Fmt, Args).
?log(debug, Fmt, Args).
-else.
report(Fmt, Args) ->
io:format(Fmt, Args),
lager:info(Fmt, Args).
?log(info, Fmt, Args).
-endif.

4
test/mtp_test_middle_server.erl
View file

@ -171,14 +171,14 @@ handle_event(info, {tcp_closed, _Sock}, _EventName, _S) ->
hs_send(Packet, #hs_state{transport = Transport, sock = Sock,
codec = Codec} = St) ->
%% lager:debug("Up>Down: ~w", [Packet]),
%% ?log(debug, "Up>Down: ~w", [Packet]),
{Encoded, Codec1} = mtp_codec:encode_packet(Packet, Codec),
ok = Transport:send(Sock, Encoded),
{ok, St#hs_state{codec = Codec1}}.
t_send(Packet, #t_state{transport = Transport, sock = Sock,
codec = Codec} = St) ->
%% lager:debug("Up>Down: ~w", [Packet]),
%% ?log(debug, "Up>Down: ~w", [Packet]),
{Encoded, Codec1} = mtp_codec:encode_packet(Packet, Codec),
ok = Transport:send(Sock, Encoded),
{ok, St#t_state{codec = Codec1}}.

10
test/prop_mtp_statefull.erl
View file

@ -231,11 +231,19 @@ ask_for_close(Id) ->
gen_rpc_close([], _ConnId, St) ->
{close, St}.
-ifdef(OTP_RELEASE).
disable_log() ->
logger:set_primary_config(level, critical).
-else.
disable_log() ->
ok.
-endif.
%% Setup / teardown
setup(DcCfg0) ->
application:ensure_all_started(lager),
lager:set_loglevel(lager_console_backend, critical),
lager:set_loglevel(lager_console_backend, critical), %XXX lager-specific
disable_log(),
{ok, Pid} = mtp_test_metric:start_link(),
PubKey = crypto:strong_rand_bytes(128),
DcId = ?DC_ID,