Merge pull request #10375 from zhongwencool/trace-conf

feat: deprecated trace config

apps/emqx/src/emqx_trace/emqx_trace.hrl

@@ -24,6 +24,8 @@
     filter ::
         emqx_types:topic() | emqx_types:clientid() | emqx_trace:ip_address() | undefined | '_',
     enable = true :: boolean() | '_',
+    payload_encode = text :: hex | text | hidden | '_',
+    extra = #{} :: map() | '_',
     start_at :: integer() | undefined | '_',
     end_at :: integer() | undefined | '_'
 }).

apps/emqx/src/emqx_schema.erl

@@ -229,7 +229,7 @@ roots(low) ->
         {"trace",
             sc(
                 ref("trace"),
-                #{}
+                #{importance => ?IMPORTANCE_HIDDEN}
             )},
         {"crl_cache",
             sc(
@@ -1853,6 +1853,8 @@ fields("trace") ->
         {"payload_encode",
             sc(hoconsc:enum([hex, text, hidden]), #{
                 default => text,
+                deprecated => {since, "5.0.22"},
+                importance => ?IMPORTANCE_HIDDEN,
                 desc => ?DESC(fields_trace_payload_encode)
             })}
     ].

apps/emqx/src/emqx_trace/emqx_trace.erl

@@ -21,6 +21,7 @@
 -include_lib("emqx/include/logger.hrl").
 -include_lib("kernel/include/file.hrl").
 -include_lib("snabbkaffe/include/trace.hrl").
+-include_lib("emqx/include/emqx_trace.hrl").
 
 -export([
     publish/1,
@@ -54,8 +55,6 @@
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
 
--include("emqx_trace.hrl").
-
 -ifdef(TEST).
 -export([
     log_file/2,
@@ -147,7 +146,11 @@ list(Enable) ->
 
 -spec create([{Key :: binary(), Value :: binary()}] | #{atom() => binary()}) ->
     {ok, #?TRACE{}}
-    | {error, {duplicate_condition, iodata()} | {already_existed, iodata()} | iodata()}.
+    | {error,
+        {duplicate_condition, iodata()}
+        | {already_existed, iodata()}
+        | {bad_type, any()}
+        | iodata()}.
 create(Trace) ->
     case mnesia:table_info(?TRACE, size) < ?MAX_SIZE of
         true ->
@@ -222,14 +225,16 @@ format(Traces) ->
 
 init([]) ->
     erlang:process_flag(trap_exit, true),
+    Fields = record_info(fields, ?TRACE),
     ok = mria:create_table(?TRACE, [
         {type, set},
         {rlog_shard, ?SHARD},
         {storage, disc_copies},
         {record_name, ?TRACE},
-        {attributes, record_info(fields, ?TRACE)}
+        {attributes, Fields}
     ]),
     ok = mria:wait_for_tables([?TRACE]),
+    maybe_migrate_trace(Fields),
     {ok, _} = mnesia:subscribe({table, ?TRACE, simple}),
     ok = filelib:ensure_dir(filename:join([trace_dir(), dummy])),
     ok = filelib:ensure_dir(filename:join([zip_dir(), dummy])),
@@ -358,9 +363,10 @@ start_trace(Trace) ->
         name = Name,
         type = Type,
         filter = Filter,
-        start_at = Start
+        start_at = Start,
+        payload_encode = PayloadEncode
     } = Trace,
-    Who = #{name => Name, type => Type, filter => Filter},
+    Who = #{name => Name, type => Type, filter => Filter, payload_encode => PayloadEncode},
     emqx_trace_handler:install(Who, debug, log_file(Name, Start)).
 
 stop_trace(Finished, Started) ->
@@ -490,6 +496,8 @@ to_trace(#{type := ip_address, ip_address := Filter} = Trace, Rec) ->
     end;
 to_trace(#{type := Type}, _Rec) ->
     {error, io_lib:format("required ~s field", [Type])};
+to_trace(#{payload_encode := PayloadEncode} = Trace, Rec) ->
+    to_trace(maps:remove(payload_encode, Trace), Rec#?TRACE{payload_encode = PayloadEncode});
 to_trace(#{start_at := StartAt} = Trace, Rec) ->
     {ok, Sec} = to_system_second(StartAt),
     to_trace(maps:remove(start_at, Trace), Rec#?TRACE{start_at = Sec});
@@ -573,3 +581,29 @@ filter_cli_handler(Names) ->
 
 now_second() ->
     os:system_time(second).
+
+maybe_migrate_trace(Fields) ->
+    case mnesia:table_info(emqx_trace, attributes) =:= Fields of
+        true ->
+            ok;
+        false ->
+            TransFun = fun(Trace) ->
+                case Trace of
+                    {?TRACE, Name, Type, Filter, Enable, StartAt, EndAt} ->
+                        #?TRACE{
+                            name = Name,
+                            type = Type,
+                            filter = Filter,
+                            enable = Enable,
+                            start_at = StartAt,
+                            end_at = EndAt,
+                            payload_encode = text,
+                            extra = #{}
+                        };
+                    #?TRACE{} ->
+                        Trace
+                end
+            end,
+            {atomic, ok} = mnesia:transform_table(?TRACE, TransFun, Fields, ?TRACE),
+            ok
+    end.

apps/emqx/src/emqx_trace/emqx_trace_handler.erl

@@ -44,7 +44,8 @@
 -type tracer() :: #{
     name := binary(),
     type := clientid | topic | ip_address,
-    filter := emqx_types:clientid() | emqx_types:topic() | emqx_trace:ip_address()
+    filter := emqx_types:clientid() | emqx_types:topic() | emqx_trace:ip_address(),
+    payload_encode := text | hidden | hex
 }.
 
 -define(CONFIG(_LogFile_), #{
@@ -70,7 +71,12 @@
     LogFilePath :: string()
 ) -> ok | {error, term()}.
 install(Name, Type, Filter, Level, LogFile) ->
-    Who = #{type => Type, filter => ensure_bin(Filter), name => ensure_bin(Name)},
+    Who = #{
+        type => Type,
+        filter => ensure_bin(Filter),
+        name => ensure_bin(Name),
+        payload_encode => payload_encode()
+    },
     install(Who, Level, LogFile).
 
 -spec install(
@@ -160,14 +166,14 @@ filters(#{type := topic, filter := Filter, name := Name}) ->
 filters(#{type := ip_address, filter := Filter, name := Name}) ->
     [{ip_address, {fun ?MODULE:filter_ip_address/2, {ensure_list(Filter), Name}}}].
 
-formatter(#{type := _Type}) ->
+formatter(#{type := _Type, payload_encode := PayloadEncode}) ->
     {emqx_trace_formatter, #{
         %% template is for ?SLOG message not ?TRACE.
         template => [time, " [", level, "] ", msg, "\n"],
         single_line => true,
         max_size => unlimited,
         depth => unlimited,
-        payload_encode => payload_encode()
+        payload_encode => PayloadEncode
     }}.
 
 filter_traces(#{id := Id, level := Level, dst := Dst, filters := Filters}, Acc) ->

apps/emqx/test/emqx_trace_SUITE.erl

@@ -22,10 +22,9 @@
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("emqx/include/emqx.hrl").
+-include_lib("emqx/include/emqx_trace.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
--record(emqx_trace, {name, type, filter, enable = true, start_at, end_at}).
-
 %%--------------------------------------------------------------------
 %% Setups
 %%--------------------------------------------------------------------
@@ -97,7 +96,9 @@ t_base_create_delete(_Config) ->
             type => clientid,
             name => <<"name1">>,
             start_at => Now,
-            end_at => Now + 30 * 60
+            end_at => Now + 30 * 60,
+            payload_encode => text,
+            extra => #{}
         }
     ],
     ?assertEqual(ExpectFormat, emqx_trace:format([TraceRec])),
@@ -385,6 +386,48 @@ t_find_closed_time(_Config) ->
     ?assertEqual(1000, emqx_trace:find_closest_time(Traces, Now)),
     ok.
 
+t_migrate_trace(_Config) ->
+    build_new_trace_data(),
+    build_old_trace_data(),
+    reload(),
+    Traces = emqx_trace:format(emqx_trace:list()),
+    ?assertEqual(2, erlang:length(Traces)),
+    lists:foreach(
+        fun(#{name := Name, enable := Enable}) ->
+            ?assertEqual(true, Enable, Name)
+        end,
+        Traces
+    ),
+    LoggerIds = logger:get_handler_ids(),
+    lists:foreach(
+        fun(Id) ->
+            ?assertEqual(true, lists:member(Id, LoggerIds), LoggerIds)
+        end,
+        [
+            trace_topic_test_topic_migrate_new,
+            trace_topic_test_topic_migrate_old
+        ]
+    ),
+    ok.
+
+build_new_trace_data() ->
+    Now = erlang:system_time(second),
+    {ok, _} = emqx_trace:create([
+        {<<"name">>, <<"test_topic_migrate_new">>},
+        {<<"type">>, topic},
+        {<<"topic">>, <<"/test/migrate/new">>},
+        {<<"start_at">>, Now - 10}
+    ]).
+
+build_old_trace_data() ->
+    Now = erlang:system_time(second),
+    OldAttrs = [name, type, filter, enable, start_at, end_at],
+    {atomic, ok} = mnesia:transform_table(emqx_trace, ignore, OldAttrs, emqx_trace),
+    OldTrace =
+        {emqx_trace, <<"test_topic_migrate_old">>, topic, <<"topic">>, true, Now - 10, Now + 100},
+    ok = mnesia:dirty_write(OldTrace),
+    ok.
+
 reload() ->
     catch ok = gen_server:stop(emqx_trace),
     {ok, _Pid} = emqx_trace:start_link().

apps/emqx_management/src/emqx_mgmt_api_trace.erl

@@ -94,7 +94,8 @@ schema("/trace") ->
                 409 => emqx_dashboard_swagger:error_codes(
                     [
                         'ALREADY_EXISTS',
-                        'DUPLICATE_CONDITION'
+                        'DUPLICATE_CONDITION',
+                        'BAD_TYPE'
                     ],
                     <<"trace already exists">>
                 )
@@ -265,6 +266,19 @@ fields(trace) ->
                     example => running
                 }
             )},
+        {payload_encode,
+            hoconsc:mk(hoconsc:enum([hex, text, hidden]), #{
+                desc =>
+                    ""
+                    "Determine the format of the payload format in the trace file.<br/>\n"
+                    "`text`: Text-based protocol or plain text protocol.\n"
+                    " It is recommended when payload is JSON encoded.<br/>\n"
+                    "`hex`: Binary hexadecimal encode."
+                    "It is recommended when payload is a custom binary protocol.<br/>\n"
+                    "`hidden`: payload is obfuscated as `******`"
+                    "",
+                default => text
+            })},
         {start_at,
             hoconsc:mk(
                 emqx_datetime:epoch_second(),
@@ -421,6 +435,11 @@ trace(post, #{body := Param}) ->
                 code => 'DUPLICATE_CONDITION',
                 message => ?TO_BIN([Name, " Duplication Condition"])
             }};
+        {error, {bad_type, _}} ->
+            {409, #{
+                code => 'BAD_TYPE',
+                message => <<"Rolling upgrade in progress, create failed">>
+            }};
         {error, Reason} ->
             {400, #{
                 code => 'INVALID_PARAMS',

apps/emqx_management/test/emqx_mgmt_api_trace_SUITE.erl

@@ -174,6 +174,13 @@ t_create_failed(_Config) ->
         {error, {"HTTP/1.1", 409, _}},
         request_api(post, api_path("trace"), [GoodName2 | Trace])
     ),
+    %% new name but bad payload-encode
+    GoodName3 = {<<"name">>, <<"test-name-2">>},
+    PayloadEncode = {<<"payload_encode">>, <<"bad">>},
+    ?assertMatch(
+        {error, {"HTTP/1.1", 400, _}},
+        request_api(post, api_path("trace"), [GoodName3, PayloadEncode | Trace])
+    ),
 
     unload(),
     emqx_trace:clear(),

changes/ce/feat-10373.en.md

@@ -0,0 +1,2 @@
+Deprecate the trace.payload_encode configuration.
+Add payload_encode=[text,hidden,hex] option when creating a trace via HTTP API.