diff --git a/rebar.config b/rebar.config index 7eef606..d43439a 100644 --- a/rebar.config +++ b/rebar.config @@ -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"}}} ]}. diff --git a/rebar.lock b/rebar.lock index 4c75028..da27c6c 100644 --- a/rebar.lock +++ b/rebar.lock @@ -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">>}]} ]. diff --git a/src/mtp_config.erl b/src/mtp_config.erl index 169469c..291c3b7 100644 --- a/src/mtp_config.erl +++ b/src/mtp_config.erl @@ -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([]) -> diff --git a/src/mtp_dc_pool.erl b/src/mtp_dc_pool.erl index 0b2d711..bd1aa1a 100644 --- a/src/mtp_dc_pool.erl +++ b/src/mtp_dc_pool.erl @@ -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. diff --git a/src/mtp_down_conn.erl b/src/mtp_down_conn.erl index f7f9526..1b4865e 100644 --- a/src/mtp_down_conn.erl +++ b/src/mtp_down_conn.erl @@ -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}}. diff --git a/src/mtp_handler.erl b/src/mtp_handler.erl index 11ad75f..ec25c20 100644 --- a/src/mtp_handler.erl +++ b/src/mtp_handler.erl @@ -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); diff --git a/src/mtp_session_storage.erl b/src/mtp_session_storage.erl index 577d3a0..4a3f27e 100644 --- a/src/mtp_session_storage.erl +++ b/src/mtp_session_storage.erl @@ -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) diff --git a/src/mtproto_proxy_app.erl b/src/mtproto_proxy_app.erl index 44eac5a..62bc387 100644 --- a/src/mtproto_proxy_app.erl +++ b/src/mtproto_proxy_app.erl @@ -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. diff --git a/test/mtp_test_middle_server.erl b/test/mtp_test_middle_server.erl index 1fc8d92..014f382 100644 --- a/test/mtp_test_middle_server.erl +++ b/test/mtp_test_middle_server.erl @@ -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}}. diff --git a/test/prop_mtp_statefull.erl b/test/prop_mtp_statefull.erl index 5a19dba..be9fabe 100644 --- a/test/prop_mtp_statefull.erl +++ b/test/prop_mtp_statefull.erl @@ -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,