Merge pull request #12660 from zmstone/sync-5.5.1

Sync 5.5.1

.github/workflows/build_and_push_docker_images.yaml

@@ -112,8 +112,8 @@ jobs:
       fail-fast: false
       matrix:
         profile:
-          - ${{ inputs.profile }}
-          - ${{ inputs.profile }}-elixir
+          - ["${{ inputs.profile }}", "${{ inputs.profile == 'emqx' && 'docker.io,public.ecr.aws' || 'docker.io' }}"]
+          - ["${{ inputs.profile }}-elixir", "${{ inputs.profile == 'emqx' && 'docker.io,public.ecr.aws' || 'docker.io' }}"]
 
     steps:
       - uses: actions/checkout@b4ffde65f46336ab88eb53be808477a3936bae11 # v4.1.1
@@ -121,7 +121,7 @@ jobs:
           ref: ${{ github.event.inputs.ref }}
       - uses: actions/download-artifact@eaceaf801fd36c7dee90939fad912460b18a1ffe # v4.1.2
         with:
-          pattern: "${{ matrix.profile }}-*.tar.gz"
+          pattern: "${{ matrix.profile[0] }}-*.tar.gz"
           path: _packages
           merge-multiple: true
 
@@ -158,8 +158,8 @@ jobs:
 
       - name: Build docker image
         env:
-          PROFILE: ${{ matrix.profile }}
-          DOCKER_REGISTRY: 'docker.io,public.ecr.aws'
+          PROFILE: ${{ matrix.profile[0] }}
+          DOCKER_REGISTRY: ${{ matrix.profile[1] }}
           DOCKER_ORG: ${{ github.repository_owner }}
           DOCKER_LATEST: ${{ inputs.latest }}
           DOCKER_PUSH: false

apps/emqx/src/emqx.app.src

@@ -2,7 +2,7 @@
 {application, emqx, [
     {id, "emqx"},
     {description, "EMQX Core"},
-    {vsn, "5.1.20"},
+    {vsn, "5.2.0"},
     {modules, []},
     {registered, []},
     {applications, [

apps/emqx/src/emqx_connection.erl

@@ -186,6 +186,8 @@
 -define(LIMITER_BYTES_IN, bytes).
 -define(LIMITER_MESSAGE_IN, messages).
 
+-define(LOG(Level, Data), ?SLOG(Level, (Data)#{tag => "MQTT"})).
+
 -dialyzer({no_match, [info/2]}).
 -dialyzer(
     {nowarn_function, [
@@ -282,7 +284,7 @@ async_set_keepalive(OS, Pid, Idle, Interval, Probes) ->
         {ok, Options} ->
             async_set_socket_options(Pid, Options);
         {error, {unsupported_os, OS}} ->
-            ?SLOG(warning, #{
+            ?LOG(warning, #{
                 msg => "Unsupported operation: set TCP keepalive",
                 os => OS
             }),
@@ -774,7 +776,7 @@ handle_timeout(TRef, Msg, State) ->
 %% Parse incoming data
 -compile({inline, [when_bytes_in/3]}).
 when_bytes_in(Oct, Data, State) ->
-    ?SLOG(debug, #{
+    ?LOG(debug, #{
         msg => "raw_bin_received",
         size => Oct,
         bin => binary_to_list(binary:encode_hex(Data)),
@@ -810,7 +812,7 @@ parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
             parse_incoming(Rest, [Packet | Packets], NState)
     catch
         throw:{?FRAME_PARSE_ERROR, Reason} ->
-            ?SLOG(info, #{
+            ?LOG(info, #{
                 reason => Reason,
                 at_state => emqx_frame:describe_state(ParseState),
                 input_bytes => Data,
@@ -818,7 +820,7 @@ parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
             }),
             {[{frame_error, Reason} | Packets], State};
         error:Reason:Stacktrace ->
-            ?SLOG(error, #{
+            ?LOG(error, #{
                 at_state => emqx_frame:describe_state(ParseState),
                 input_bytes => Data,
                 parsed_packets => Packets,
@@ -873,7 +875,7 @@ serialize_and_inc_stats_fun(#state{serialize = Serialize}) ->
     fun(Packet) ->
         try emqx_frame:serialize_pkt(Packet, Serialize) of
             <<>> ->
-                ?SLOG(warning, #{
+                ?LOG(warning, #{
                     msg => "packet_is_discarded",
                     reason => "frame_is_too_large",
                     packet => emqx_packet:format(Packet, hidden)
@@ -889,13 +891,13 @@ serialize_and_inc_stats_fun(#state{serialize = Serialize}) ->
         catch
             %% Maybe Never happen.
             throw:{?FRAME_SERIALIZE_ERROR, Reason} ->
-                ?SLOG(info, #{
+                ?LOG(info, #{
                     reason => Reason,
                     input_packet => Packet
                 }),
                 erlang:error({?FRAME_SERIALIZE_ERROR, Reason});
             error:Reason:Stacktrace ->
-                ?SLOG(error, #{
+                ?LOG(error, #{
                     input_packet => Packet,
                     exception => Reason,
                     stacktrace => Stacktrace
@@ -1018,7 +1020,7 @@ check_limiter(
             WhenOk(Data, Msgs, State#state{limiter = Limiter2});
         {pause, Time, Limiter2} ->
             ?SLOG(debug, #{
-                msg => "pause_time_dueto_rate_limit",
+                msg => "pause_time_due_to_rate_limit",
                 needs => Needs,
                 time_in_ms => Time
             }),
@@ -1070,7 +1072,7 @@ retry_limiter(#state{limiter = Limiter} = State) ->
             );
         {pause, Time, Limiter2} ->
             ?SLOG(debug, #{
-                msg => "pause_time_dueto_rate_limit",
+                msg => "pause_time_due_to_rate_limit",
                 types => Types,
                 time_in_ms => Time
             }),

apps/emqx/src/emqx_logger_textfmt.erl

@@ -22,7 +22,11 @@
 
 check_config(X) -> logger_formatter:check_config(X).
 
+%% Principle here is to delegate the formatting to logger_formatter:format/2
+%% as much as possible, and only enrich the report with clientid, peername, topic, username
 format(#{msg := {report, ReportMap}, meta := Meta} = Event, Config) when is_map(ReportMap) ->
+    %% The most common case, when entering from SLOG macro
+    %% i.e. logger:log(Level, #{msg => "my_msg", foo => bar})
     ReportList = enrich_report(ReportMap, Meta),
     Report =
         case is_list_report_acceptable(Meta) of
@@ -33,13 +37,17 @@ format(#{msg := {report, ReportMap}, meta := Meta} = Event, Config) when is_map(
         end,
     logger_formatter:format(Event#{msg := {report, Report}}, Config);
 format(#{msg := {string, String}} = Event, Config) ->
+    %% copied from logger_formatter:format/2
+    %% unsure how this case is triggered
     format(Event#{msg => {"~ts ", [String]}}, Config);
-%% trace
 format(#{msg := Msg0, meta := Meta} = Event, Config) ->
+    %% For format strings like logger:log(Level, "~p", [Var])
+    %% and logger:log(Level, "message", #{key => value})
     Msg1 = enrich_client_info(Msg0, Meta),
     Msg2 = enrich_topic(Msg1, Meta),
     logger_formatter:format(Event#{msg := Msg2}, Config).
 
+%% Other report callbacks may only accept map() reports such as gen_server formatter
 is_list_report_acceptable(#{report_cb := Cb}) ->
     Cb =:= fun logger:format_otp_report/1 orelse Cb =:= fun logger:format_report/1;
 is_list_report_acceptable(_) ->
@@ -61,19 +69,21 @@ enrich_report(ReportRaw, Meta) ->
     ClientId = maps:get(clientid, Meta, undefined),
     Peer = maps:get(peername, Meta, undefined),
     Msg = maps:get(msg, ReportRaw, undefined),
+    Tag = maps:get(tag, ReportRaw, undefined),
     %% turn it into a list so that the order of the fields is determined
     lists:foldl(
         fun
             ({_, undefined}, Acc) -> Acc;
             (Item, Acc) -> [Item | Acc]
         end,
-        maps:to_list(maps:without([topic, msg, clientid, username], ReportRaw)),
+        maps:to_list(maps:without([topic, msg, clientid, username, tag], ReportRaw)),
         [
-            {username, try_format_unicode(Username)},
             {topic, try_format_unicode(Topic)},
-            {clientid, try_format_unicode(ClientId)},
+            {username, try_format_unicode(Username)},
             {peername, Peer},
-            {msg, Msg}
+            {msg, Msg},
+            {clientid, try_format_unicode(ClientId)},
+            {tag, Tag}
         ]
     ).
 

apps/emqx/src/emqx_ws_connection.erl

@@ -128,6 +128,8 @@
 -dialyzer({no_match, [info/2]}).
 -dialyzer({nowarn_function, [websocket_init/1]}).
 
+-define(LOG(Level, Data), ?SLOG(Level, (Data)#{tag => "MQTT"})).
+
 %%--------------------------------------------------------------------
 %% Info, Stats
 %%--------------------------------------------------------------------
@@ -401,7 +403,7 @@ get_peer_info(Type, Listener, Req, Opts) ->
 websocket_handle({binary, Data}, State) when is_list(Data) ->
     websocket_handle({binary, iolist_to_binary(Data)}, State);
 websocket_handle({binary, Data}, State) ->
-    ?SLOG(debug, #{
+    ?LOG(debug, #{
         msg => "raw_bin_received",
         size => iolist_size(Data),
         bin => binary_to_list(binary:encode_hex(Data)),
@@ -428,7 +430,7 @@ websocket_handle({Frame, _}, State) when Frame =:= ping; Frame =:= pong ->
     return(State);
 websocket_handle({Frame, _}, State) ->
     %% TODO: should not close the ws connection
-    ?SLOG(error, #{msg => "unexpected_frame", frame => Frame}),
+    ?LOG(error, #{msg => "unexpected_frame", frame => Frame}),
     shutdown(unexpected_ws_frame, State).
 websocket_info({call, From, Req}, State) ->
     handle_call(From, Req, State);
@@ -714,7 +716,7 @@ parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
             parse_incoming(Rest, [{incoming, Packet} | Packets], NState)
     catch
         throw:{?FRAME_PARSE_ERROR, Reason} ->
-            ?SLOG(info, #{
+            ?LOG(info, #{
                 reason => Reason,
                 at_state => emqx_frame:describe_state(ParseState),
                 input_bytes => Data
@@ -722,7 +724,7 @@ parse_incoming(Data, Packets, State = #state{parse_state = ParseState}) ->
             FrameError = {frame_error, Reason},
             {[{incoming, FrameError} | Packets], State};
         error:Reason:Stacktrace ->
-            ?SLOG(error, #{
+            ?LOG(error, #{
                 at_state => emqx_frame:describe_state(ParseState),
                 input_bytes => Data,
                 exception => Reason,
@@ -812,7 +814,7 @@ serialize_and_inc_stats_fun(#state{serialize = Serialize}) ->
     fun(Packet) ->
         try emqx_frame:serialize_pkt(Packet, Serialize) of
             <<>> ->
-                ?SLOG(warning, #{
+                ?LOG(warning, #{
                     msg => "packet_discarded",
                     reason => "frame_too_large",
                     packet => emqx_packet:format(Packet)
@@ -828,13 +830,13 @@ serialize_and_inc_stats_fun(#state{serialize = Serialize}) ->
         catch
             %% Maybe Never happen.
             throw:{?FRAME_SERIALIZE_ERROR, Reason} ->
-                ?SLOG(info, #{
+                ?LOG(info, #{
                     reason => Reason,
                     input_packet => Packet
                 }),
                 erlang:error({?FRAME_SERIALIZE_ERROR, Reason});
             error:Reason:Stacktrace ->
-                ?SLOG(error, #{
+                ?LOG(error, #{
                     input_packet => Packet,
                     exception => Reason,
                     stacktrace => Stacktrace

apps/emqx_bridge_http/src/emqx_bridge_http.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_http, [
     {description, "EMQX HTTP Bridge and Connector Application"},
-    {vsn, "0.2.3"},
+    {vsn, "0.2.4"},
     {registered, []},
     {applications, [kernel, stdlib, emqx_resource, ehttpc]},
     {env, [{emqx_action_info_modules, [emqx_bridge_http_action_info]}]},

apps/emqx_bridge_iotdb/src/emqx_bridge_iotdb.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge_iotdb, [
     {description, "EMQX Enterprise Apache IoTDB Bridge"},
-    {vsn, "0.1.6"},
+    {vsn, "0.1.7"},
     {modules, [
         emqx_bridge_iotdb,
         emqx_bridge_iotdb_connector

apps/emqx_connector/src/emqx_connector.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_connector, [
     {description, "EMQX Data Integration Connectors"},
-    {vsn, "0.1.39"},
+    {vsn, "0.2.0"},
     {registered, []},
     {mod, {emqx_connector_app, []}},
     {applications, [

apps/emqx_management/src/emqx_management.app.src

@@ -2,7 +2,7 @@
 {application, emqx_management, [
     {description, "EMQX Management API and CLI"},
     % strict semver, bump manually!
-    {vsn, "5.0.38"},
+    {vsn, "5.1.0"},
     {modules, []},
     {registered, [emqx_management_sup]},
     {applications, [

apps/emqx_prometheus/src/emqx_prometheus.app.src

@@ -2,7 +2,7 @@
 {application, emqx_prometheus, [
     {description, "Prometheus for EMQX"},
     % strict semver, bump manually!
-    {vsn, "5.0.20"},
+    {vsn, "5.1.0"},
     {modules, []},
     {registered, [emqx_prometheus_sup]},
     {applications, [kernel, stdlib, prometheus, emqx, emqx_auth, emqx_resource, emqx_management]},

apps/emqx_rule_engine/test/emqx_rule_funcs_SUITE.erl

@@ -1192,7 +1192,18 @@ t_parse_date_errors(_) ->
     ?assertEqual(
         UnixTsLeap2,
         emqx_rule_funcs:date_to_unix_ts(second, <<"%Y-%m-%d %H:%M:%S">>, <<"2024-03-04 06:56:27">>)
-    ).
+    ),
+
+    %% None zero zone shift with millisecond level precision
+    Tz1 = calendar:rfc3339_to_system_time("2024-02-23T15:00:00.123+08:00", [{unit, second}]),
+    ?assertEqual(
+        Tz1,
+        emqx_rule_funcs:date_to_unix_ts(
+            second, <<"%Y-%m-%d %H:%M:%S.%3N%:z">>, <<"2024-02-23 15:00:00.123+08:00">>
+        )
+    ),
+
+    ok.
 
 %%------------------------------------------------------------------------------
 %% Utility functions

apps/emqx_utils/src/emqx_utils.app.src

@@ -2,7 +2,7 @@
 {application, emqx_utils, [
     {description, "Miscellaneous utilities for EMQX apps"},
     % strict semver, bump manually!
-    {vsn, "5.0.16"},
+    {vsn, "5.1.0"},
     {modules, [
         emqx_utils,
         emqx_utils_api,

apps/emqx_utils/src/emqx_utils_calendar.erl

@@ -507,7 +507,7 @@ do_parse(DateStr, Unit, Formatter) ->
             (nanosecond, V, Res) ->
                 Res + V;
             (parsed_offset, V, Res) ->
-                Res - V
+                Res - V * Precise
         end,
     Count = maps:fold(Counter, 0, DateInfo) - (?SECONDS_PER_DAY * Precise),
     erlang:convert_time_unit(Count, PrecisionUnit, Unit).

build

@@ -470,10 +470,8 @@ make_docker() {
     )
     :> ./.emqx_docker_image_tags
     for r in "${DOCKER_REGISTRIES[@]}"; do
-        if ! is_ecr_and_enterprise "$r" "$PROFILE"; then
-            DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_IMAGE_TAG}")
-            echo "$r/${EMQX_IMAGE_TAG}" >> ./.emqx_docker_image_tags
-        fi
+        DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_IMAGE_TAG}")
+        echo "$r/${EMQX_IMAGE_TAG}" >> ./.emqx_docker_image_tags
     done
     if [ "${DOCKER_BUILD_NOCACHE:-false}" = true ]; then
         DOCKER_BUILDX_ARGS+=(--no-cache)
@@ -483,14 +481,12 @@ make_docker() {
     fi
     if [ "${DOCKER_LATEST:-false}" = true ]; then
         for r in "${DOCKER_REGISTRIES[@]}"; do
-            if ! is_ecr_and_enterprise "$r" "$PROFILE"; then
-                DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_BASE_DOCKER_TAG}:latest${SUFFIX}")
-                echo "$r/${EMQX_BASE_DOCKER_TAG}:latest${SUFFIX}" >> ./.emqx_docker_image_tags
-                DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}${SUFFIX}")
-                echo "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}${SUFFIX}" >> ./.emqx_docker_image_tags
-                DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}.${VSN_PATCH}${SUFFIX}")
-                echo "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}.${VSN_PATCH}${SUFFIX}" >> ./.emqx_docker_image_tags
-            fi
+          DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_BASE_DOCKER_TAG}:latest${SUFFIX}")
+          echo "$r/${EMQX_BASE_DOCKER_TAG}:latest${SUFFIX}" >> ./.emqx_docker_image_tags
+          DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}${SUFFIX}")
+          echo "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}${SUFFIX}" >> ./.emqx_docker_image_tags
+          DOCKER_BUILDX_ARGS+=(--tag "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}.${VSN_PATCH}${SUFFIX}")
+          echo "$r/${EMQX_BASE_DOCKER_TAG}:${VSN_MAJOR}.${VSN_MINOR}.${VSN_PATCH}${SUFFIX}" >> ./.emqx_docker_image_tags
         done
     fi
     if [ "${DOCKER_PLATFORMS:-default}" != 'default' ]; then

changes/ce/feat-12641.en.md

@@ -0,0 +1,3 @@
+Improve text log formatter fields order.
+
+`tag` > `clientid` > `msg` > `peername` > `username` > `topic` > [other fields]

changes/ce/fix-12646.en.md

@@ -0,0 +1,3 @@
+Fix rule engine date time string parser.
+
+Prior to this fix, time zone shift can only work when date time string is at second level precision.

changes/e5.5.1.en.md

@@ -8,8 +8,6 @@
 
 - [#12471](https://github.com/emqx/emqx/pull/12471) Fixed an issue that data integration configurations failed to load correctly during upgrades from EMQX version 5.0.2 to newer releases.
 
-- [#12542](https://github.com/emqx/emqx/pull/12542) Redacted authorization headers to exclude basic authorization credentials from debug logs in the HTTP Server connector, mitigating potential security risks.
-
 - [#12598](https://github.com/emqx/emqx/pull/12598) Fixed an issue that users were unable to subscribe to or unsubscribe from shared topic filters via HTTP API.
 
   The affected APIs include:
@@ -24,6 +22,10 @@
 
 - [#12606](https://github.com/emqx/emqx/pull/12606) The Prometheus API experienced crashes when the specified SSL certificate file did not exist in the given path. Now, when an SSL certificate file is missing, the `emqx_cert_expiry_at` metric will report a value of 0, indicating the non-existence of the certificate.
 
+- [#12620](https://github.com/emqx/emqx/pull/12620) Redacted sensitive information in HTTP headers to exclude authentication and authorization credentials from `debug` level logs in the HTTP Server connector, mitigating potential security risks.
+
+- [#12632](https://github.com/emqx/emqx/pull/12632) Fixed an issue where the rule engine's SQL built-in function `date_to_unix_ts` produced incorrect results for dates starting from March 1st on leap years.
+
 - [#12608](https://github.com/emqx/emqx/pull/12608) Fixed a `function_clause` error in the IoTDB action caused by the absence of a `payload` field in query data.
 
 - [#12610](https://github.com/emqx/emqx/pull/12610) Fixed an issue where connections to the LDAP connector could unexpectedly disconnect after a certain period of time.

changes/v5.5.1.en.md

@@ -4,8 +4,6 @@
 
 - [#12471](https://github.com/emqx/emqx/pull/12471) Fixed an issue that data integration configurations failed to load correctly during upgrades from EMQX version 5.0.2 to newer releases.
 
-- [#12542](https://github.com/emqx/emqx/pull/12542) Redacted authorization headers to exclude basic authorization credentials from debug logs in the HTTP Server connector, mitigating potential security risks.
-
 - [#12598](https://github.com/emqx/emqx/pull/12598) Fixed an issue that users were unable to subscribe to or unsubscribe from shared topic filters via HTTP API.
 
   The affected APIs include:
@@ -19,3 +17,8 @@
 - [#12601](https://github.com/emqx/emqx/pull/12601) Fixed an issue where logs of the LDAP driver were not being captured. Now, all logs are recorded at the `info` level.
 
 - [#12606](https://github.com/emqx/emqx/pull/12606) The Prometheus API experienced crashes when the specified SSL certificate file did not exist in the given path. Now, when an SSL certificate file is missing, the `emqx_cert_expiry_at` metric will report a value of 0, indicating the non-existence of the certificate.
+
+- [#12620](https://github.com/emqx/emqx/pull/12620) Redacted sensitive information in HTTP headers to exclude authentication and authorization credentials from `debug` level logs in the HTTP Server connector, mitigating potential security risks.
+
+- [#12632](https://github.com/emqx/emqx/pull/12632) Fixed an issue where the rule engine's SQL built-in function `date_to_unix_ts` produced incorrect results for dates starting from March 1st on leap years.
+

scripts/git-hook-pre-commit.sh

@@ -2,6 +2,10 @@
 
 set -euo pipefail
 
+if [ -n "${FORCE:-}" ]; then
+    exit 0
+fi
+
 OPT="${1:--c}"
 
 # mix format check is quite fast