Merge pull request #14033 from emqx/fix-log-packet-format

fix: incorrect format when printing packets in log

apps/emqx/src/emqx_logger_textfmt.erl

@@ -190,18 +190,19 @@ try_encode_meta(Report, Config) ->
     ).
 
 try_encode_meta(payload, #{payload := Payload} = Report, #{payload_encode := Encode}) ->
-    Report#{payload := encode_payload(Payload, Encode)};
+    {Payload1, Encode1} = format_payload(Payload, Encode),
+    Report#{payload => Payload1, payload_encode => Encode1};
 try_encode_meta(packet, #{packet := Packet} = Report, #{payload_encode := Encode}) when
     is_tuple(Packet)
 ->
-    Report#{packet := emqx_packet:format(Packet, Encode)};
+    Report#{packet := try_format_unicode(emqx_packet:format(Packet, Encode))};
 try_encode_meta(_, Report, _Config) ->
     Report.
 
-encode_payload(Payload, text) ->
-    Payload;
-encode_payload(_Payload, hidden) ->
-    "******";
-encode_payload(Payload, hex) ->
-    Bin = emqx_utils_conv:bin(Payload),
-    binary:encode_hex(Bin).
+format_payload(undefined, Type) ->
+    {"", Type};
+format_payload(Payload, Type) when is_binary(Payload) ->
+    {Payload1, Type1} = emqx_packet:format_payload(Payload, Type),
+    {try_format_unicode(Payload1), Type1};
+format_payload(Payload, Type) ->
+    {Payload, Type}.

apps/emqx/src/emqx_packet.erl

@@ -65,6 +65,7 @@
 -type publish() :: #mqtt_packet_publish{}.
 -type subscribe() :: #mqtt_packet_subscribe{}.
 -type unsubscribe() :: #mqtt_packet_unsubscribe{}.
+-type payload_encode() :: hex | text | hidden.
 
 %%--------------------------------------------------------------------
 %% MQTT Packet Type and Flags.
@@ -481,7 +482,7 @@ will_msg(#mqtt_packet_connect{
     }.
 
 %% @doc Format packet
--spec format(emqx_types:packet(), hex | text | hidden) -> iolist().
+-spec format(emqx_types:packet(), payload_encode()) -> iolist().
 format(#mqtt_packet{header = Header, variable = Variable, payload = Payload}, PayloadEncode) ->
     HeaderIO = format_header(Header),
     case format_variable(Variable, Payload, PayloadEncode) of
@@ -618,12 +619,14 @@ format_password(<<>>) -> "";
 format_password(_Password) -> "******".
 
 format_payload_label(Payload, Type) ->
-    ["Payload=", format_payload(Payload, Type)].
+    {FPayload, Type1} = format_payload(Payload, Type),
+    [io_lib:format("Payload(~s)=", [Type1]), FPayload].
 
+-spec format_payload(binary(), payload_encode()) -> {iolist(), payload_encode()}.
 format_payload(_, hidden) ->
-    "******";
-format_payload(<<>>, _) ->
-    "";
+    {"******", hidden};
+format_payload(<<>>, Type) ->
+    {"", Type};
 format_payload(Payload, Type) when ?MAX_PAYLOAD_FORMAT_LIMIT(Payload) ->
     %% under the 1KB limit
     format_payload_limit(Type, Payload, size(Payload));
@@ -635,25 +638,28 @@ format_payload_limit(Type0, Payload, Limit) when size(Payload) > Limit ->
     {Type, Part, TruncatedBytes} = truncate_payload(Type0, Limit, Payload),
     case TruncatedBytes > 0 of
         true ->
-            [do_format_payload(Type, Part), "...(", integer_to_list(TruncatedBytes), " bytes)"];
+            {
+                [do_format_payload(Type, Part), "...(", integer_to_list(TruncatedBytes), " bytes)"],
+                Type
+            };
         false ->
-            do_format_payload(Type, Payload)
+            {do_format_payload(Type, Payload), Type}
     end;
 format_payload_limit(text, Payload, _Limit) ->
     case is_utf8(Payload) of
         true ->
-            do_format_payload(text, Payload);
+            {do_format_payload(text, Payload), text};
         false ->
-            do_format_payload(hex, Payload)
+            {do_format_payload(hex, Payload), hex}
     end;
 format_payload_limit(hex, Payload, _Limit) ->
-    do_format_payload(hex, Payload).
+    {do_format_payload(hex, Payload), hex}.
 
 do_format_payload(text, Bytes) ->
     %% utf8 ensured
     Bytes;
 do_format_payload(hex, Bytes) ->
-    ["hex:", binary:encode_hex(Bytes)].
+    binary:encode_hex(Bytes).
 
 is_utf8(Bytes) ->
     case trim_utf8(size(Bytes), Bytes) of

apps/emqx/src/emqx_trace/emqx_trace_formatter.erl

@@ -54,7 +54,13 @@ format_meta_map(Meta, _Encode, []) ->
 format_meta_map(Meta, Encode, [{Name, FormatFun} | Rest]) ->
     case Meta of
         #{Name := Value} ->
-            NewMeta = Meta#{Name => FormatFun(Value, Encode)},
+            NewMeta =
+                case FormatFun(Value, Encode) of
+                    {NewValue, NewMeta0} ->
+                        maps:merge(Meta#{Name => NewValue}, NewMeta0);
+                    NewValue ->
+                        Meta#{Name => NewValue}
+                end,
             format_meta_map(NewMeta, Encode, Rest);
         #{} ->
             format_meta_map(Meta, Encode, Rest)
@@ -96,12 +102,13 @@ format_packet(Packet, Encode) ->
             Packet
     end.
 
-format_payload(undefined, _) ->
-    "";
+format_payload(undefined, Type) ->
+    {"", #{payload_encode => Type}};
 format_payload(Payload, Type) when is_binary(Payload) ->
-    emqx_packet:format_payload(Payload, Type);
-format_payload(Payload, _) ->
-    Payload.
+    {Payload1, Type1} = emqx_packet:format_payload(Payload, Type),
+    {Payload1, #{payload_encode => Type1}};
+format_payload(Payload, Type) ->
+    {Payload, #{payload_encode => Type}}.
 
 to_iolist(Atom) when is_atom(Atom) -> atom_to_list(Atom);
 to_iolist(Int) when is_integer(Int) -> integer_to_list(Int);

apps/emqx/src/emqx_trace/emqx_trace_json_formatter.erl

@@ -86,7 +86,17 @@ do_maybe_format_msg(Msg, Meta, Config) ->
     emqx_logger_jsonfmt:format_msg(Msg, Meta, Config).
 
 prepare_log_data(LogMap, PEncode) when is_map(LogMap) ->
-    NewKeyValuePairs = [prepare_key_value(K, V, PEncode) || {K, V} <- maps:to_list(LogMap)],
+    NewKeyValuePairs = lists:flatmap(
+        fun({K, V}) ->
+            case prepare_key_value(K, V, PEncode) of
+                KV when is_tuple(KV) ->
+                    [KV];
+                KVs when is_list(KVs) ->
+                    KVs
+            end
+        end,
+        maps:to_list(LogMap)
+    ),
     maps:from_list(NewKeyValuePairs);
 prepare_log_data(V, PEncode) when is_list(V) ->
     [prepare_log_data(Item, PEncode) || Item <- V];
@@ -118,14 +128,14 @@ prepare_key_value(host, {I1, I2, I3, I4, I5, I6, I7, I8} = IP, _PEncode) when
     %% We assume this is an IP address
     {host, unicode:characters_to_binary(inet:ntoa(IP))};
 prepare_key_value(payload = K, V, PEncode) ->
-    NewV =
+    {NewV, PEncode1} =
         try
             format_payload(V, PEncode)
         catch
             _:_ ->
                 V
         end,
-    {K, NewV};
+    [{K, NewV}, {payload_encode, PEncode1}];
 prepare_key_value(<<"payload">>, V, PEncode) ->
     prepare_key_value(payload, V, PEncode);
 prepare_key_value(packet = K, V, PEncode) ->
@@ -183,12 +193,12 @@ prepare_key_value(K, V, PEncode) ->
 format_packet(undefined, _) -> "";
 format_packet(Packet, Encode) -> emqx_packet:format(Packet, Encode).
 
-format_payload(undefined, _) ->
-    "";
+format_payload(undefined, Type) ->
+    {"", Type};
 format_payload(Payload, Type) when is_binary(Payload) ->
     emqx_packet:format_payload(Payload, Type);
-format_payload(Payload, _) ->
-    Payload.
+format_payload(Payload, Type) ->
+    {Payload, Type}.
 
 format_map_set_to_list(Map) ->
     Items = [

apps/emqx/test/emqx_packet_tests.erl

@@ -19,53 +19,52 @@
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("emqx/include/emqx_mqtt.hrl").
 
+-define(HEX(P), {bin(P), hex}).
+-define(HIDDEN(P), {bin(P), hidden}).
+-define(TEXT(P), {bin(P), text}).
+
 format_payload_test_() ->
-    Hidden = fun(Payload) -> emqx_packet:format_payload(Payload, hidden) end,
-    Hex = fun(Payload) -> bin(emqx_packet:format_payload(Payload, hex)) end,
+    Hidden = format_payload(hidden),
+    Hex = format_payload(hex),
     [
-        {"hidden", fun() -> ?assertEqual("******", Hidden(<<>>)) end},
-        {"hex empty", fun() -> ?assertEqual(<<"">>, Hex(<<"">>)) end},
-        {"hex short", fun() -> ?assertEqual(<<"hex:303030">>, Hex(<<"000">>)) end},
+        {"hidden", fun() -> ?assertEqual(?HIDDEN("******"), Hidden(<<>>)) end},
+        {"hex empty", fun() -> ?assertEqual(?HEX(<<"">>), Hex(<<"">>)) end},
+        {"hex short", fun() -> ?assertEqual(?HEX(<<"303030">>), Hex(<<"000">>)) end},
         {"hex at limit", fun() ->
             Payload = bin(lists:duplicate(?MAX_PAYLOAD_FORMAT_SIZE, 0)),
-            Expected = bin(
-                [
-                    "hex:",
-                    binary:encode_hex(bin(lists:duplicate(?MAX_PAYLOAD_FORMAT_SIZE, 0)))
-                ]
-            ),
-            ?assertEqual(Expected, Hex(Payload))
+            Expected = binary:encode_hex(bin(lists:duplicate(?MAX_PAYLOAD_FORMAT_SIZE, 0))),
+            ?assertEqual(?HEX(Expected), Hex(Payload))
         end},
         {"hex long", fun() ->
             Payload = bin(lists:duplicate(?MAX_PAYLOAD_FORMAT_SIZE + 2, 0)),
             Prefix = binary:encode_hex(bin(lists:duplicate(?TRUNCATED_PAYLOAD_SIZE, 0))),
             Lost = size(Payload) - ?TRUNCATED_PAYLOAD_SIZE,
-            Expected = bin(["hex:", Prefix, "...(", integer_to_list(Lost), " bytes)"]),
-            ?assertEqual(Expected, Hex(Payload))
+            Expected = [Prefix, "...(", integer_to_list(Lost), " bytes)"],
+            ?assertEqual(?HEX(Expected), Hex(Payload))
         end}
     ].
 
 format_payload_utf8_test_() ->
-    Fmt = fun(P) -> bin(emqx_packet:format_payload(P, text)) end,
+    Fmt = format_payload(text),
     [
-        {"empty", fun() -> ?assertEqual(<<"">>, Fmt(<<>>)) end},
-        {"short ascii", fun() -> ?assertEqual(<<"abc">>, Fmt(<<"abc">>)) end},
-        {"short unicode", fun() -> ?assertEqual(<<"日志"/utf8>>, Fmt(<<"日志"/utf8>>)) end},
+        {"empty", fun() -> ?assertEqual(?TEXT(<<"">>), Fmt(<<>>)) end},
+        {"short ascii", fun() -> ?assertEqual(?TEXT(<<"abc">>), Fmt(<<"abc">>)) end},
+        {"short unicode", fun() -> ?assertEqual(?TEXT(<<"日志"/utf8>>), Fmt(<<"日志"/utf8>>)) end},
         {"unicode at limit", fun() ->
             Payload = bin(lists:duplicate(?MAX_PAYLOAD_FORMAT_SIZE div 2, <<"¢"/utf8>>)),
-            Expected = bin(["", Payload]),
-            ?assertEqual(Expected, Fmt(Payload))
+            Expected = ["", Payload],
+            ?assertEqual(?TEXT(Expected), Fmt(Payload))
         end}
     ].
 
 format_payload_utf8_cutoff_test_() ->
-    Fmt = fun(P) -> bin(emqx_packet:format_payload(P, text)) end,
+    Fmt = format_payload(text),
     Check = fun(MultiBytesChar) ->
         Prefix = [lists:duplicate(?TRUNCATED_PAYLOAD_SIZE - 1, $a), MultiBytesChar],
         Payload = bin([Prefix, MultiBytesChar, lists:duplicate(?MAX_PAYLOAD_FORMAT_SIZE, $b)]),
         Lost = size(Payload) - iolist_size(Prefix),
-        Expected = bin([Prefix, "...(", integer_to_list(Lost), " bytes)"]),
-        ?assertEqual(Expected, Fmt(Payload))
+        Expected = [Prefix, "...(", integer_to_list(Lost), " bytes)"],
+        ?assertEqual(?TEXT(Expected), Fmt(Payload))
     end,
     [
         {"utf8 1B", fun() -> Check(<<"x"/utf8>>) end},
@@ -75,8 +74,9 @@ format_payload_utf8_cutoff_test_() ->
     ].
 
 invalid_utf8_fallback_test() ->
+    Fmt = format_payload(text),
     %% trucate after the first byte of a utf8 encoded unicode character
-    <<FirstByte:8, Last3Bytes/binary>> = <<"𐍈"/utf8>>,
+    <<FirstByte:8, _Last3Bytes/binary>> = <<"𐍈"/utf8>>,
     Prefix = iolist_to_binary([lists:duplicate(?TRUNCATED_PAYLOAD_SIZE - 1, $a), FirstByte]),
     %% invalidate utf8 byte sequence, so it should fallback to hex
     InvalidUtf8 = 255,
@@ -85,10 +85,16 @@ invalid_utf8_fallback_test() ->
     ]),
     Lost = size(Payload) - iolist_size(Prefix),
     Expected = iolist_to_binary([
-        "hex:", binary:encode_hex(Prefix), "...(", integer_to_list(Lost), " bytes)"
+        binary:encode_hex(Prefix), "...(", integer_to_list(Lost), " bytes)"
     ]),
-    ?assertEqual(Expected, bin(emqx_packet:format_payload(Payload, text))),
+    ?assertEqual(?HEX(Expected), Fmt(Payload)),
     ok.
 
+format_payload(Encode) ->
+    fun(Payload) ->
+        {Payload1, Encode1} = emqx_packet:format_payload(Payload, Encode),
+        {bin(Payload1), Encode1}
+    end.
+
 bin(X) ->
     unicode:characters_to_binary(X).