diff --git a/apps/emqx/src/emqx_trace/emqx_trace.erl b/apps/emqx/src/emqx_trace/emqx_trace.erl index e6ee4260f..e9866fb16 100644 --- a/apps/emqx/src/emqx_trace/emqx_trace.erl +++ b/apps/emqx/src/emqx_trace/emqx_trace.erl @@ -51,10 +51,7 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --define(TRACE, ?MODULE). --define(SHARD, ?COMMON_SHARD). --define(MAX_SIZE, 30). --define(OWN_KEYS, [level, filters, filter_default, handlers]). +-include("emqx_trace.hrl"). -ifdef(TEST). -export([ @@ -66,15 +63,6 @@ -export_type([ip_address/0]). -type ip_address() :: string(). --record(?TRACE, { - name :: binary() | undefined | '_', - type :: clientid | topic | ip_address | undefined | '_', - filter :: emqx_types:topic() | emqx_types:clientid() | ip_address() | undefined | '_', - enable = true :: boolean() | '_', - start_at :: integer() | undefined | '_', - end_at :: integer() | undefined | '_' -}). - publish(#message{topic = <<"$SYS/", _/binary>>}) -> ignore; publish(#message{from = From, topic = Topic, payload = Payload}) when @@ -172,13 +160,7 @@ create(Trace) -> -spec delete(Name :: binary()) -> ok | {error, not_found}. delete(Name) -> - Tran = fun() -> - case mnesia:read(?TRACE, Name) of - [_] -> mnesia:delete(?TRACE, Name, write); - [] -> mnesia:abort(not_found) - end - end, - transaction(Tran). + transaction(fun emqx_trace_dl:delete/1, [Name]). -spec clear() -> ok | {error, Reason :: term()}. clear() -> @@ -190,20 +172,7 @@ clear() -> -spec update(Name :: binary(), Enable :: boolean()) -> ok | {error, not_found | finished}. update(Name, Enable) -> - Tran = fun() -> - case mnesia:read(?TRACE, Name) of - [] -> - mnesia:abort(not_found); - [#?TRACE{enable = Enable}] -> - ok; - [Rec] -> - case erlang:system_time(second) >= Rec#?TRACE.end_at of - false -> mnesia:write(?TRACE, Rec#?TRACE{enable = Enable}, write); - true -> mnesia:abort(finished) - end - end - end, - transaction(Tran). + transaction(fun emqx_trace_dl:update/2, [Name, Enable]). check() -> gen_server:call(?MODULE, check). @@ -211,13 +180,7 @@ check() -> -spec get_trace_filename(Name :: binary()) -> {ok, FileName :: string()} | {error, not_found}. get_trace_filename(Name) -> - Tran = fun() -> - case mnesia:read(?TRACE, Name, read) of - [] -> mnesia:abort(not_found); - [#?TRACE{start_at = Start}] -> {ok, filename(Name, Start)} - end - end, - transaction(Tran). + transaction(fun emqx_trace_dl:get_trace_filename/1, [Name]). -spec trace_file(File :: file:filename_all()) -> {ok, Node :: list(), Binary :: binary()} @@ -309,23 +272,7 @@ code_change(_, State, _Extra) -> {ok, State}. insert_new_trace(Trace) -> - Tran = fun() -> - case mnesia:read(?TRACE, Trace#?TRACE.name) of - [] -> - #?TRACE{start_at = StartAt, type = Type, filter = Filter} = Trace, - Match = #?TRACE{_ = '_', start_at = StartAt, type = Type, filter = Filter}, - case mnesia:match_object(?TRACE, Match, read) of - [] -> - ok = mnesia:write(?TRACE, Trace, write), - {ok, Trace}; - [#?TRACE{name = Name}] -> - mnesia:abort({duplicate_condition, Name}) - end; - [#?TRACE{name = Name}] -> - mnesia:abort({already_existed, Name}) - end - end, - transaction(Tran). + transaction(fun emqx_trace_dl:insert_new_trace/1, [Trace]). update_trace(Traces) -> Now = erlang:system_time(second), @@ -347,9 +294,7 @@ stop_all_trace_handler() -> get_enabled_trace() -> {atomic, Traces} = - mria:ro_transaction(?SHARD, fun() -> - mnesia:match_object(?TRACE, #?TRACE{enable = true, _ = '_'}, read) - end), + mria:ro_transaction(?SHARD, fun emqx_trace_dl:get_enabled_trace/0), Traces. find_closest_time(Traces, Now) -> @@ -372,17 +317,7 @@ closest(Time, Now, Closest) -> min(Time - Now, Closest). disable_finished([]) -> ok; disable_finished(Traces) -> - transaction(fun() -> - lists:map( - fun(#?TRACE{name = Name}) -> - case mnesia:read(?TRACE, Name, write) of - [] -> ok; - [Trace] -> mnesia:write(?TRACE, Trace#?TRACE{enable = false}, write) - end - end, - Traces - ) - end). + transaction(fun emqx_trace_dl:delete_finished/1, [Traces]). start_trace(Traces, Started0) -> Started = lists:map(fun(#{name := Name}) -> Name end, Started0), @@ -586,8 +521,8 @@ filename(Name, Start) -> [Time, _] = string:split(calendar:system_time_to_rfc3339(Start), "T", leading), lists:flatten(["trace_", binary_to_list(Name), "_", Time, ".log"]). -transaction(Tran) -> - case mria:transaction(?COMMON_SHARD, Tran) of +transaction(Fun, Args) -> + case mria:transaction(?COMMON_SHARD, Fun, Args) of {atomic, Res} -> Res; {aborted, Reason} -> {error, Reason} end. diff --git a/apps/emqx/src/emqx_trace/emqx_trace.hrl b/apps/emqx/src/emqx_trace/emqx_trace.hrl new file mode 100644 index 000000000..a00a37132 --- /dev/null +++ b/apps/emqx/src/emqx_trace/emqx_trace.hrl @@ -0,0 +1,35 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%%-------------------------------------------------------------------- +-ifndef(EMQX_TRACE_HRL). +-define(EMQX_TRACE_HRL, true). + +-define(TRACE, emqx_trace). + +-record(?TRACE, { + name :: binary() | undefined | '_', + type :: clientid | topic | ip_address | undefined | '_', + filter :: + emqx_types:topic() | emqx_types:clientid() | emqx_trace:ip_address() | undefined | '_', + enable = true :: boolean() | '_', + start_at :: integer() | undefined | '_', + end_at :: integer() | undefined | '_' +}). + +-define(SHARD, ?COMMON_SHARD). +-define(MAX_SIZE, 30). +-define(OWN_KEYS, [level, filters, filter_default, handlers]). + +-endif. diff --git a/apps/emqx/src/emqx_trace/emqx_trace_dl.erl b/apps/emqx/src/emqx_trace/emqx_trace_dl.erl new file mode 100644 index 000000000..3f96e1531 --- /dev/null +++ b/apps/emqx/src/emqx_trace/emqx_trace_dl.erl @@ -0,0 +1,103 @@ +%%-------------------------------------------------------------------- +%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%%-------------------------------------------------------------------- + +%% Data layer for emqx_trace +-module(emqx_trace_dl). + +%% API: +-export([ + update/2, + insert_new_trace/1, + delete/1, + get_trace_filename/1, + delete_finished/1, + get_enabled_trace/0 +]). + +-include("emqx_trace.hrl"). + +%%================================================================================ +%% API funcions +%%================================================================================ + +%% Introduced in 5.0 +-spec update(Name :: binary(), Enable :: boolean()) -> + ok. +update(Name, Enable) -> + case mnesia:read(?TRACE, Name) of + [] -> + mnesia:abort(not_found); + [#?TRACE{enable = Enable}] -> + ok; + [Rec] -> + case erlang:system_time(second) >= Rec#?TRACE.end_at of + false -> mnesia:write(?TRACE, Rec#?TRACE{enable = Enable}, write); + true -> mnesia:abort(finished) + end + end. + +%% Introduced in 5.0 +insert_new_trace(Trace) -> + case mnesia:read(?TRACE, Trace#?TRACE.name) of + [] -> + #?TRACE{start_at = StartAt, type = Type, filter = Filter} = Trace, + Match = #?TRACE{_ = '_', start_at = StartAt, type = Type, filter = Filter}, + case mnesia:match_object(?TRACE, Match, read) of + [] -> + ok = mnesia:write(?TRACE, Trace, write), + {ok, Trace}; + [#?TRACE{name = Name}] -> + mnesia:abort({duplicate_condition, Name}) + end; + [#?TRACE{name = Name}] -> + mnesia:abort({already_existed, Name}) + end. + +%% Introduced in 5.0 +-spec delete(Name :: binary()) -> ok. +delete(Name) -> + case mnesia:read(?TRACE, Name) of + [_] -> mnesia:delete(?TRACE, Name, write); + [] -> mnesia:abort(not_found) + end. + +%% Introduced in 5.0 +-spec get_trace_filename(Name :: binary()) -> {ok, string()}. +get_trace_filename(Name) -> + case mnesia:read(?TRACE, Name, read) of + [] -> mnesia:abort(not_found); + [#?TRACE{start_at = Start}] -> {ok, emqx_trace:filename(Name, Start)} + end. + +%% Introduced in 5.0 +delete_finished(Traces) -> + lists:map( + fun(#?TRACE{name = Name}) -> + case mnesia:read(?TRACE, Name, write) of + [] -> ok; + [Trace] -> mnesia:write(?TRACE, Trace#?TRACE{enable = false}, write) + end + end, + Traces + ). + +%% Introduced in 5.0 +get_enabled_trace() -> + mnesia:match_object(?TRACE, #?TRACE{enable = true, _ = '_'}, read). + +%%================================================================================ +%% Internal functions +%%================================================================================