yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make emqx_tracer more reliable 

Remove the emqx_tracer process as the trace-handlers stored in state would be lost in case of crash.
This commit is contained in:
terry-xiaoyu 2019-08-04 01:01:21 +08:00
parent be2ce93a2c
commit 1a5d8ca3fd
3 changed files with 75 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/emqx_kernel_sup.erl
View File

@ -30,8 +30,7 @@ init([]) ->
child_spec(emqx_stats, worker), child_spec(emqx_stats, worker),
child_spec(emqx_metrics, worker), child_spec(emqx_metrics, worker),
child_spec(emqx_ctl, worker), child_spec(emqx_ctl, worker),
child_spec(emqx_zone, worker), child_spec(emqx_zone, worker)]}}.
child_spec(emqx_tracer, worker)]}}.
child_spec(M, worker) -> child_spec(M, worker) ->
#{id => M, #{id => M,

182
src/emqx_tracer.erl
View File

@ -14,34 +14,19 @@
-module(emqx_tracer). -module(emqx_tracer).
-behaviour(gen_server).
-include("emqx.hrl"). -include("emqx.hrl").
-include("logger.hrl"). -include("logger.hrl").
-logger_header("[Tracer]"). -logger_header("[Tracer]").
%% APIs %% APIs
-export([start_link/0]).
-export([ trace/2 -export([ trace/2
, start_trace/3 , start_trace/3
, lookup_traces/0 , lookup_traces/0
, stop_trace/1 , stop_trace/1
]). ]).
%% gen_server callbacks -type(trace_who() :: {client_id | topic, binary() | list()}).
-export([ init/1
, handle_call/3
, handle_cast/2
, handle_info/2
, terminate/2
, code_change/3
]).
-record(state, {traces}).
-type(trace_who() :: {client_id | topic, binary()}).
-define(TRACER, ?MODULE). -define(TRACER, ?MODULE).
-define(FORMAT, {emqx_logger_formatter, -define(FORMAT, {emqx_logger_formatter,
@ -55,120 +40,100 @@
[peername," "], [peername," "],
[]}]}, []}]},
msg,"\n"]}}). msg,"\n"]}}).
-define(TOPIC_TRACE_ID(T), "trace_topic_"++T).
-define(CLIENT_TRACE_ID(C), "trace_clientid_"++C).
-define(TOPIC_TRACE(T), {topic,T}).
-define(CLIENT_TRACE(C), {client_id,C}).
-define(is_log_level(L),
L =:= emergency orelse
L =:= alert orelse
L =:= critical orelse
L =:= error orelse
L =:= warning orelse
L =:= notice orelse
L =:= info orelse
L =:= debug).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% APIs %% APIs
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
-spec(start_link() -> {ok, pid()} | ignore | {error, term()}).
start_link() ->
gen_server:start_link({local, ?TRACER}, ?MODULE, [], []).
trace(publish, #message{topic = <<"$SYS/", _/binary>>}) -> trace(publish, #message{topic = <<"$SYS/", _/binary>>}) ->
%% Dont' trace '$SYS' publish %% Do not trace '$SYS' publish
ignore; ignore;
trace(publish, #message{from = From, topic = Topic, payload = Payload}) trace(publish, #message{from = From, topic = Topic, payload = Payload})
when is_binary(From); is_atom(From) -> when is_binary(From); is_atom(From) ->
emqx_logger:info(#{topic => Topic}, "PUBLISH to ~s: ~p", [Topic, Payload]). emqx_logger:info(#{topic => Topic}, "PUBLISH to ~s: ~p", [Topic, Payload]).
%%------------------------------------------------------------------------------
%% Start/Stop trace
%%------------------------------------------------------------------------------
%% @doc Start to trace client_id or topic. %% @doc Start to trace client_id or topic.
-spec(start_trace(trace_who(), logger:level(), string()) -> ok | {error, term()}). -spec(start_trace(trace_who(), logger:level(), string()) -> ok | {error, term()}).
start_trace({client_id, ClientId}, Level, LogFile) -> start_trace(Who, all, LogFile) ->
do_start_trace({client_id, ClientId}, Level, LogFile); start_trace(Who, debug, LogFile);
start_trace({topic, Topic}, Level, LogFile) -> start_trace(Who, Level, LogFile) ->
do_start_trace({topic, Topic}, Level, LogFile). case ?is_log_level(Level) of
true ->
do_start_trace(Who, Level, LogFile) -> #{level := PrimaryLevel} = logger:get_primary_config(),
#{level := PrimaryLevel} = logger:get_primary_config(), try logger:compare_levels(Level, PrimaryLevel) of
try logger:compare_levels(log_level(Level), PrimaryLevel) of lt ->
lt -> {error, io_lib:format("Cannot trace at a log level (~s) lower than the primary log level (~s)", [Level, PrimaryLevel])};
{error, io_lib:format("Cannot trace at a log level (~s) lower than the primary log level (~s)", [Level, PrimaryLevel])}; _GtOrEq ->
_GtOrEq -> install_trace_handler(Who, Level, LogFile)
gen_server:call(?MODULE, {start_trace, Who, Level, LogFile}, 5000) catch
catch _:Error ->
_:Error -> {error, Error}
{error, Error} end;
false -> {error, {invalid_log_level, Level}}
end. end.
%% @doc Stop tracing client_id or topic. %% @doc Stop tracing client_id or topic.
-spec(stop_trace(trace_who()) -> ok | {error, term()}). -spec(stop_trace(trace_who()) -> ok | {error, term()}).
stop_trace({client_id, ClientId}) -> stop_trace(Who) ->
gen_server:call(?MODULE, {stop_trace, {client_id, ClientId}}); uninstall_trance_handler(Who).
stop_trace({topic, Topic}) ->
gen_server:call(?MODULE, {stop_trace, {topic, Topic}}).
%% @doc Lookup all traces %% @doc Lookup all traces
-spec(lookup_traces() -> [{Who :: trace_who(), LogFile :: string()}]). -spec(lookup_traces() -> [{Who :: trace_who(), LogFile :: string()}]).
lookup_traces() -> lookup_traces() ->
gen_server:call(?TRACER, lookup_traces). lists:foldl(fun filter_traces/2, [], emqx_logger:get_log_handlers()).
%%------------------------------------------------------------------------------ install_trace_handler(Who, Level, LogFile) ->
%% gen_server callbacks
%%------------------------------------------------------------------------------
init([]) ->
{ok, #state{traces = #{}}}.
handle_call({start_trace, Who, Level, LogFile}, _From, State = #state{traces = Traces}) ->
case logger:add_handler(handler_id(Who), logger_disk_log_h, case logger:add_handler(handler_id(Who), logger_disk_log_h,
#{level => Level, #{level => Level,
formatter => ?FORMAT, formatter => ?FORMAT,
filesync_repeat_interval => no_repeat, filesync_repeat_interval => no_repeat,
config => #{type => halt, file => LogFile}, config => #{type => halt, file => LogFile},
filter_default => stop, filter_default => stop,
filters => [{meta_key_filter, filters => [{meta_key_filter,
{fun filter_by_meta_key/2, Who} }]}) of {fun filter_by_meta_key/2, Who}}]})
of
ok -> ok ->
?LOG(info, "Start trace for ~p", [Who]), ?LOG(info, "Start trace for ~p", [Who]);
{reply, ok, State#state{traces = maps:put(Who, {Level, LogFile}, Traces)}};
{error, Reason} -> {error, Reason} ->
?LOG(error, "Start trace for ~p failed, error: ~p", [Who, Reason]), ?LOG(error, "Start trace for ~p failed, error: ~p", [Who, Reason]),
{reply, {error, Reason}, State} {error, Reason}
end; end.
handle_call({stop_trace, Who}, _From, State = #state{traces = Traces}) -> uninstall_trance_handler(Who) ->
case maps:find(Who, Traces) of case logger:remove_handler(handler_id(Who)) of
{ok, _LogFile} -> ok ->
case logger:remove_handler(handler_id(Who)) of ?LOG(info, "Stop trace for ~p", [Who]);
ok -> {error, Reason} ->
?LOG(info, "Stop trace for ~p", [Who]); ?LOG(error, "Stop trace for ~p failed, error: ~p", [Who, Reason]),
{error, Reason} -> {error, Reason}
?LOG(error, "Stop trace for ~p failed, error: ~p", [Who, Reason]) end.
end,
{reply, ok, State#state{traces = maps:remove(Who, Traces)}};
error ->
{reply, {error, not_found}, State}
end;
handle_call(lookup_traces, _From, State = #state{traces = Traces}) -> filter_traces({Id, Level, Dst}, Acc) ->
{reply, [{Who, LogFile} || {Who, LogFile} <- maps:to_list(Traces)], State}; case atom_to_list(Id) of
?TOPIC_TRACE_ID(T)->
[{?TOPIC_TRACE(T), {Level,Dst}} | Acc];
?CLIENT_TRACE_ID(C) ->
[{?CLIENT_TRACE(C), {Level,Dst}} | Acc];
_ -> Acc
end.
handle_call(Req, _From, State) -> handler_id(?TOPIC_TRACE(Topic)) ->
?LOG(error, "Unexpected call: ~p", [Req]), list_to_atom(?TOPIC_TRACE_ID(str(Topic)));
{reply, ignored, State}. handler_id(?CLIENT_TRACE(ClientId)) ->
list_to_atom(?CLIENT_TRACE_ID(str(ClientId))).
handle_cast(Msg, State) ->
?LOG(error, "Unexpected cast: ~p", [Msg]),
{noreply, State}.
handle_info(Info, State) ->
?LOG(error, "Unexpected info: ~p", [Info]),
{noreply, State}.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
handler_id({topic, Topic}) ->
list_to_atom("topic_" ++ binary_to_list(Topic));
handler_id({client_id, ClientId}) ->
list_to_atom("clientid_" ++ binary_to_list(ClientId)).
filter_by_meta_key(#{meta:=Meta}=LogEvent, {MetaKey, MetaValue}) -> filter_by_meta_key(#{meta:=Meta}=LogEvent, {MetaKey, MetaValue}) ->
case maps:find(MetaKey, Meta) of case maps:find(MetaKey, Meta) of
@ -181,13 +146,6 @@ filter_by_meta_key(#{meta:=Meta}=LogEvent, {MetaKey, MetaValue}) ->
_ -> ignore _ -> ignore
end. end.
log_level(emergency) -> emergency; str(Bin) when is_binary(Bin) -> binary_to_list(Bin);
log_level(alert) -> alert; str(Atom) when is_atom(Atom) -> atom_to_list(Atom);
log_level(critical) -> critical; str(Str) when is_list(Str) -> Str.
log_level(error) -> error;
log_level(warning) -> warning;
log_level(notice) -> notice;
log_level(info) -> info;
log_level(debug) -> debug;
log_level(all) -> debug;
log_level(_) -> throw(invalid_log_level).

8
test/emqx_tracer_SUITE.erl
View File

@ -43,7 +43,7 @@ start_traces(_Config) ->
emqx_logger:set_log_level(debug), emqx_logger:set_log_level(debug),
ok = emqx_tracer:start_trace({client_id, <<"client">>}, debug, "tmp/client.log"), ok = emqx_tracer:start_trace({client_id, <<"client">>}, debug, "tmp/client.log"),
ok = emqx_tracer:start_trace({client_id, <<"client2">>}, all, "tmp/client2.log"), ok = emqx_tracer:start_trace({client_id, <<"client2">>}, all, "tmp/client2.log"),
{error, invalid_log_level} = emqx_tracer:start_trace({client_id, <<"client3">>}, bad_level, "tmp/client3.log"), {error, {invalid_log_level, bad_level}} = emqx_tracer:start_trace({client_id, <<"client3">>}, bad_level, "tmp/client3.log"),
ok = emqx_tracer:start_trace({topic, <<"a/#">>}, all, "tmp/topic_trace.log"), ok = emqx_tracer:start_trace({topic, <<"a/#">>}, all, "tmp/topic_trace.log"),
ct:sleep(100), ct:sleep(100),
@ -53,9 +53,9 @@ start_traces(_Config) ->
?assert(filelib:is_regular("tmp/topic_trace.log")), ?assert(filelib:is_regular("tmp/topic_trace.log")),
%% Get current traces %% Get current traces
?assertEqual([{{client_id,<<"client">>},{debug,"tmp/client.log"}}, ?assertEqual([{{client_id,"client"},{debug,"tmp/client.log"}},
{{client_id,<<"client2">>},{all,"tmp/client2.log"}}, {{client_id,"client2"},{debug,"tmp/client2.log"}},
{{topic,<<"a/#">>},{all,"tmp/topic_trace.log"}}], emqx_tracer:lookup_traces()), {{topic,"a/#"},{debug,"tmp/topic_trace.log"}}], emqx_tracer:lookup_traces()),
%% set the overall log level to debug %% set the overall log level to debug
emqx_logger:set_log_level(debug), emqx_logger:set_log_level(debug),