Merge remote-tracking branch 'origin/main-v4.4' into merge-v44-into-v45-b

.github/workflows/apps_version_check.yaml

@@ -21,6 +21,10 @@ jobs:
           fetch-depth: 0 # need full history
       - name: fix-git-unsafe-repository
         run: git config --global --add safe.directory /__w/emqx/emqx
+      - name: Check relup version DB
+        run: |
+          PKG_VSN=$(./pkg-vsn.sh)
+          ./scripts/relup-base-vsns.escript check-vsn-db $PKG_VSN ./data/relup-paths.eterm
       - name: Check relup (ce)
         if: endsWith(github.repository, 'emqx')
         run: ./scripts/update-appup.sh emqx --check

.github/workflows/build_packages.yaml

@@ -316,7 +316,7 @@ jobs:
       id: meta
       with:
         images: ${{ matrix.registry }}/${{ github.repository_owner }}/${{ matrix.profile }}
-        ## only stable tag is latest
+        ## only 5.0 is latest
         flavor: |
           latest=false # latest is now 5.0
         tags: |

CHANGES-4.3.md

@@ -8,7 +8,31 @@ File format:
 
 - Use weight-2 heading for releases
 - One list item per change topic
-  Change log ends with a list of github PRs
+  Change log ends with a list of GitHub PRs
+
+## v4.3.19
+
+### Enhancements
+
+- Improve error message for LwM2M plugin when object ID is not valid [#8654](https://github.com/emqx/emqx/pull/8654).
+- Add tzdata apk package to alpine docker image. [#8671](https://github.com/emqx/emqx/pull/8671)
+- Add node evacuation and cluster rebalancing features [#8597](https://github.com/emqx/emqx/pull/8597)
+- Refine Rule Engine error log. RuleId will be logged when take action failed. [#8737](https://github.com/emqx/emqx/pull/8737)
+- Close ExProto client process immediately if it's keepalive timeouted. [#8725](https://github.com/emqx/emqx/pull/8725)
+- Upgrade grpc-erl driver to 0.6.7 to support batch operation in sending stream. [#8725](https://github.com/emqx/emqx/pull/8725)
+- Improved jwt authentication module initialization process.[#8736](https://github.com/emqx/emqx/pull/8736)
+
+### Bug fixes
+
+- Fix rule SQL compare to null values always returns false. [#8743](https://github.com/emqx/emqx/pull/8743)
+  Before this change, the following SQL failed to match on the WHERE clause (`clientid != foo` returns false):
+  `SELECT 'some_var' as clientid FROM "t" WHERE clientid != foo`.
+  The `foo` variable is a null value, so `clientid != foo` should be evaluated as true.
+- Fix GET `/auth_clientid` and `/auth_username` counts. [#8655](https://github.com/emqx/emqx/pull/8655)
+- Add an idle timer for ExProto UDP client to avoid client leaking [#8628](https://github.com/emqx/emqx/pull/8628)
+- Fix ExHook can't be un-hooked if the grpc service stop first. [#8725](//github.com/emqx/emqx/pull/8725)
+- Fix GET `/listeners/` crashes when listener is not ready. [#8752](https://github.com/emqx/emqx/pull/8752)
+
 
 ## v4.3.18
 

apps/emqx_auth_jwt/src/emqx_auth_jwt.app.src

@@ -1,6 +1,6 @@
 {application, emqx_auth_jwt,
  [{description, "EMQ X Authentication with JWT"},
-  {vsn, "4.4.3"}, % strict semver, bump manually!
+  {vsn, "4.4.4"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_auth_jwt_sup]},
   {applications, [kernel,stdlib,jose]},

apps/emqx_auth_jwt/src/emqx_auth_jwt.appup.src

@@ -1,11 +1,13 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.2",[{load_module,emqx_auth_jwt_svr,brutal_purge,soft_purge,[]},
+  [{"4.4.3",[{load_module,emqx_auth_jwt_svr,brutal_purge,soft_purge,[]}]},
+   {"4.4.2",[{load_module,emqx_auth_jwt_svr,brutal_purge,soft_purge,[]},
              {load_module,emqx_auth_jwt,brutal_purge,soft_purge,[]}]},
    {<<"4\\.4\\.[0-1]">>,[{restart_application,emqx_auth_jwt}]},
    {<<".*">>,[]}],
-  [{"4.4.2",[{load_module,emqx_auth_jwt_svr,brutal_purge,soft_purge,[]},
+  [{"4.4.3",[{load_module,emqx_auth_jwt_svr,brutal_purge,soft_purge,[]}]},
+   {"4.4.2",[{load_module,emqx_auth_jwt_svr,brutal_purge,soft_purge,[]},
              {load_module,emqx_auth_jwt,brutal_purge,soft_purge,[]}]},
    {<<"4\\.4\\.[0-1]">>,[{restart_application,emqx_auth_jwt}]},
    {<<".*">>,[]}]}.

apps/emqx_auth_jwt/src/emqx_auth_jwt_svr.erl

@@ -73,8 +73,9 @@ verify(JwsCompacted) when is_binary(JwsCompacted) ->
 init([Options]) ->
     ok = jose:json_module(jiffy),
     _ = ets:new(?TAB, [set, protected, named_table]),
-    {Static, Remote} = do_init_jwks(Options),
+    Static = do_init_jwks(Options),
-    true = ets:insert(?TAB, [{static, Static}, {remote, Remote}]),
+    to_request_jwks(Options),
+    true = ets:insert(?TAB, [{static, Static}, {remote, undefined}]),
     Intv = proplists:get_value(interval, Options, ?INTERVAL),
     {ok, reset_timer(
            #state{
@@ -83,29 +84,13 @@ init([Options]) ->
 
 %% @private
 do_init_jwks(Options) ->
-    K2J = fun(K, F) ->
+    OctJwk = key2jwt_value(secret,
-              case proplists:get_value(K, Options) of
+                           fun(V) ->
-                  undefined -> undefined;
-                  V ->
-                     try F(V) of
-                         {error, Reason} ->
-                             ?LOG(warning, "Build ~p JWK ~p failed: {error, ~p}~n",
-                                  [K, V, Reason]),
-                             undefined;
-                         J -> J
-                     catch T:R ->
-                         ?LOG(warning, "Build ~p JWK ~p failed: {~p, ~p}~n",
-                              [K, V, T, R]),
-                         undefined
-                     end
-              end
-          end,
-    OctJwk = K2J(secret, fun(V) ->
                                    jose_jwk:from_oct(list_to_binary(V))
-                         end),
+                           end,
-    PemJwk = K2J(pubkey, fun jose_jwk:from_pem_file/1),
+                           Options),
-    Remote = K2J(jwks_addr, fun request_jwks/1),
+    PemJwk = key2jwt_value(pubkey, fun jose_jwk:from_pem_file/1, Options),
-    {[J ||J <- [OctJwk, PemJwk], J /= undefined], Remote}.
+    [J ||J <- [OctJwk, PemJwk], J /= undefined].
 
 handle_call(_Req, _From, State) ->
     {reply, ok, State}.
@@ -122,6 +107,11 @@ handle_info({timeout, _TRef, refresh}, State = #state{addr = Addr}) ->
              end,
     {noreply, reset_timer(NState)};
 
+handle_info({request_jwks, Options}, State) ->
+    Remote = key2jwt_value(jwks_addr, fun request_jwks/1, Options),
+    true = ets:insert(?TAB, {remote, Remote}),
+    {noreply, State};
+
 handle_info(_Info, State) ->
     {noreply, State}.
 
@@ -249,3 +239,23 @@ do_check_claim([{K, F}|More], Claims) ->
         _ ->
             do_check_claim(More, Claims)
     end.
+
+to_request_jwks(Options) ->
+    erlang:send(self(), {request_jwks, Options}).
+
+key2jwt_value(Key, Func, Options) ->
+    case proplists:get_value(Key, Options) of
+        undefined -> undefined;
+        V ->
+            try Func(V) of
+                {error, Reason} ->
+                    ?LOG(warning, "Build ~p JWK ~p failed: {error, ~p}~n",
+                         [Key, V, Reason]),
+                    undefined;
+                J -> J
+            catch T:R ->
+                    ?LOG(warning, "Build ~p JWK ~p failed: {~p, ~p}~n",
+                         [Key, V, T, R]),
+                    undefined
+            end
+    end.

apps/emqx_auth_mnesia/src/emqx_auth_mnesia.app.src

@@ -1,6 +1,6 @@
 {application, emqx_auth_mnesia,
  [{description, "EMQ X Authentication with Mnesia"},
-  {vsn, "4.3.7"}, % strict semver, bump manually
+  {vsn, "4.3.8"}, % strict semver, bump manually
   {modules, []},
   {registered, []},
   {applications, [kernel,stdlib,mnesia]},

apps/emqx_auth_mnesia/src/emqx_auth_mnesia.appup.src

@@ -1,7 +1,9 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{<<"4\\.3\\.[5-6]">>,
+  [{<<"4\\.3\\.7">>,
+    [{load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[5-6]">>,
     [{load_module,emqx_auth_mnesia_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]},
@@ -28,7 +30,8 @@
      {load_module,emqx_acl_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_app,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{<<"4\\.3\\.[5-6]">>,
+  [{"4.3.7",[{load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[5-6]">>,
     [{load_module,emqx_auth_mnesia_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]},

apps/emqx_auth_mnesia/src/emqx_auth_mnesia_api.erl

@@ -132,7 +132,11 @@
 
 list_clientid(_Bindings, Params) ->
     SortFun = fun(#{created_at := C1}, #{created_at := C2}) -> C1 > C2 end,
-    return({ok, emqx_mgmt_api:node_query(node(), Params, ?CLIENTID_SCHEMA, ?query_clientid, SortFun)}).
+    CountFun = fun() ->
+        MatchSpec = [{{?TABLE, {clientid, '_'}, '_', '_'}, [], [true]}],
+        ets:select_count(?TABLE, MatchSpec)
+               end,
+    return({ok, emqx_mgmt_api:node_query(node(), Params, ?CLIENTID_SCHEMA, ?query_clientid, SortFun, CountFun)}).
 
 lookup_clientid(#{clientid := Clientid}, _Params) ->
     return({ok, format(emqx_auth_mnesia_cli:lookup_user({clientid, urldecode(Clientid)}))}).
@@ -182,7 +186,11 @@ delete_clientid(#{clientid := Clientid}, _) ->
 
 list_username(_Bindings, Params) ->
     SortFun = fun(#{created_at := C1}, #{created_at := C2}) -> C1 > C2 end,
-    return({ok, emqx_mgmt_api:node_query(node(), Params, ?USERNAME_SCHEMA, ?query_username, SortFun)}).
+    CountFun = fun() ->
+        MatchSpec = [{{?TABLE, {username, '_'}, '_', '_'}, [], [true]}],
+        ets:select_count(?TABLE, MatchSpec)
+               end,
+    return({ok, emqx_mgmt_api:node_query(node(), Params, ?USERNAME_SCHEMA, ?query_username, SortFun, CountFun)}).
 
 lookup_username(#{username := Username}, _Params) ->
     return({ok, format(emqx_auth_mnesia_cli:lookup_user({username, urldecode(Username)}))}).

apps/emqx_auth_mnesia/test/emqx_auth_mnesia_SUITE.erl

@@ -272,7 +272,8 @@ t_clientid_rest_api(_Config) ->
     clean_all_users(),
 
     {ok, Result1} = request_http_rest_list(["auth_clientid"]),
-    [] = get_http_data(Result1),
+    ?assertMatch(#{<<"data">> := [], <<"meta">> := #{<<"count">> := 0}},
+        emqx_json:decode(Result1, [return_maps])),
 
     Params1 = #{<<"clientid">> => ?CLIENTID, <<"password">> => ?PASSWORD},
     {ok, _} = request_http_rest_add(["auth_clientid"], Params1),
@@ -295,8 +296,28 @@ t_clientid_rest_api(_Config) ->
                   }, get_http_data(Result3)),
 
     {ok, Result4} = request_http_rest_list(["auth_clientid"]),
+    #{<<"data">> := Data4, <<"meta">> := #{<<"count">> := Count4}}
+        = emqx_json:decode(Result4, [return_maps]),
 
-    ?assertEqual(3, length(get_http_data(Result4))),
+    ?assertEqual(3, Count4),
+    ?assertEqual([<<"client2">>, <<"clientid1">>, ?CLIENTID],
+        lists:sort(lists:map(fun(#{<<"clientid">> := C}) -> C end, Data4))),
+
+    UserNameParams = [#{<<"username">> => <<"username1">>, <<"password">> => ?PASSWORD}
+        , #{<<"username">> => <<"username2">>, <<"password">> => ?PASSWORD}
+    ],
+    {ok, _} = request_http_rest_add(["auth_username"], UserNameParams),
+
+    {ok, Result41} = request_http_rest_list(["auth_clientid"]),
+    %% the count clientid is not affected by username count.
+    ?assertEqual(Result4, Result41),
+
+    {ok, Result42} = request_http_rest_list(["auth_username"]),
+    #{<<"data">> := Data42, <<"meta">> := #{<<"count">> := Count42}}
+        = emqx_json:decode(Result42, [return_maps]),
+    ?assertEqual(2, Count42),
+    ?assertEqual([<<"username1">>, <<"username2">>],
+        lists:sort(lists:map(fun(#{<<"username">> := U}) -> U end, Data42))),
 
     {ok, Result5} = request_http_rest_list(["auth_clientid?_like_clientid=id"]),
     ?assertEqual(2, length(get_http_data(Result5))),

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.app.src

@@ -1,6 +1,6 @@
 {application, emqx_bridge_mqtt,
  [{description, "EMQ X Bridge to MQTT Broker"},
-  {vsn, "4.3.5"}, % strict semver, bump manually!
+  {vsn, "4.3.6"}, % strict semver, bump manually!
   {modules, []},
   {registered, []},
   {applications, [kernel,stdlib,replayq,emqtt]},

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt.appup.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.3.4",
+  [{<<"4\\.3\\.[4-5]">>,
     [{load_module,emqx_bridge_mqtt_actions,brutal_purge,soft_purge,[]}]},
    {"4.3.3",
     [{load_module,emqx_bridge_mqtt_actions,brutal_purge,soft_purge,[]},
@@ -14,7 +14,7 @@
      {load_module,emqx_bridge_worker,brutal_purge,soft_purge,[]},
      {load_module,emqx_bridge_mqtt_actions,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.3.4",
+  [{<<"4\\.3\\.[4-5]">>,
     [{load_module,emqx_bridge_mqtt_actions,brutal_purge,soft_purge,[]}]},
    {"4.3.3",
     [{load_module,emqx_bridge_mqtt_actions,brutal_purge,soft_purge,[]},

apps/emqx_bridge_mqtt/src/emqx_bridge_mqtt_actions.erl

@@ -433,7 +433,7 @@ test_resource_status(PoolName) ->
     try
         Status = [
             receive {Pid, R} -> R
-            after 1000 -> %% get_worker_status/1 should be a quick operation
+            after 10000 -> %% get_worker_status/1 should be a quick operation
                 throw({timeout, Pid})
             end || Pid <- Pids],
         lists:any(fun(St) -> St =:= true end, Status)
@@ -444,13 +444,28 @@ test_resource_status(PoolName) ->
             false
     end.
 
+-define(RETRY_TIMES, 4).
+
 get_worker_status(Worker) ->
+    get_worker_status(Worker, ?RETRY_TIMES).
+
+get_worker_status(_Worker, 0) ->
+    false;
+get_worker_status(Worker, Times) ->
     case ecpool_worker:client(Worker) of
         {ok, Bridge} ->
             try emqx_bridge_worker:status(Bridge) of
-                connected -> true;
+                connected ->
-                _ -> false
+                    true;
-            catch _Error:_Reason ->
+                idle ->
+                    ?LOG(info, "MQTT Bridge get status idle. Should not ignore this."),
+                    timer:sleep(100),
+                    get_worker_status(Worker, Times - 1);
+                ErrorStatus ->
+                    ?LOG(error, "MQTT Bridge get status ~p", [ErrorStatus]),
+                    false
+            catch Error:Reason:ST ->
+                    ?LOG(error, "MQTT Bridge get status error: ~p reason: ~p stacktrace: ~p", [Error, Reason, ST]),
                     false
             end;
         {error, _} ->

apps/emqx_exhook/src/emqx_exhook.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_exhook,
  [{description, "EMQ X Extension for Hook"},
-  {vsn, "4.4.2"},
+  {vsn, "4.4.3"},
   {modules, []},
   {registered, []},
   {mod, {emqx_exhook_app, []}},

apps/emqx_exhook/src/emqx_exhook.appup.src

@@ -1,8 +1,10 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.1",
+  [
-    [{load_module, emqx_exhook_server, brutal_purge, soft_purge, []}]},
+   {<<"4\\.4\\.[1-2]">>,
+    [{load_module, emqx_exhook_server, brutal_purge, soft_purge, []},
+     {load_module,emqx_exhook_mngr,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
     [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]},
      {load_module,emqx_exhook,brutal_purge,soft_purge,[]},
@@ -11,8 +13,9 @@
      {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[]},
      {update, emqx_exhook_mngr, {advanced, ["4.4.0"]}}]},
    {<<".*">>,[]}],
-  [{"4.4.1",
+  [{<<"4\\.4\\.[1-2]">>,
-    [{load_module, emqx_exhook_server, brutal_purge, soft_purge, []}]},
+    [{load_module, emqx_exhook_server, brutal_purge, soft_purge, []},
+     {load_module,emqx_exhook_mngr,brutal_purge,soft_purge,[]}]},
    {"4.4.0",
     [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]},
      {load_module,emqx_exhook,brutal_purge,soft_purge,[]},

apps/emqx_exhook/src/emqx_exhook_mngr.erl

@@ -198,6 +198,17 @@ handle_info({timeout, _Ref, {reload, Name}}, State) ->
             {noreply, NState};
         {error, not_found} ->
             {noreply, NState};
+        {error, {already_started, Pid}} ->
+            ?LOG(warning, "Server ~s already started on ~p, try to restart it", [Name, Pid]),
+            case server(Name) of
+                undefined ->
+                    %% force close grpc client pool
+                    grpc_client_sup:stop_channel_pool(Name);
+                ServerState ->
+                   emqx_exhook_server:unload(ServerState)
+            end,
+            %% try again immediately
+            handle_info({timeout, _Ref, {reload, Name}}, State);
         {error, Reason} ->
             ?LOG(warning, "Failed to reload exhook callback server \"~s\", "
                           "Reason: ~0p", [Name, Reason]),
@@ -208,11 +219,11 @@ handle_info(_Info, State) ->
     {noreply, State}.
 
 terminate(_Reason, State = #state{running = Running}) ->
+    _ = unload_exhooks(),
     _ = maps:fold(fun(Name, _, AccIn) ->
             {ok, NAccIn} = do_unload_server(Name, AccIn),
             NAccIn
         end, State, Running),
-    _ = unload_exhooks(),
     ok.
 
 %% in the emqx_exhook:v4.3.5, we have added one new field in the state last:
@@ -332,7 +343,7 @@ get_pool_size() ->
 
 save(Name, ServerState) ->
     Saved = persistent_term:get(?APP, []),
-    persistent_term:put(?APP, lists:reverse([Name | Saved])),
+    persistent_term:put(?APP, lists:usort(lists:reverse([Name | Saved]))),
     persistent_term:put({?APP, Name}, ServerState).
 
 unsave(Name) ->

apps/emqx_exhook/src/emqx_exhook_server.erl

@@ -153,13 +153,15 @@ format_http_uri(Scheme, Host0, Port) ->
 
 -spec unload(server()) -> ok.
 unload(#server{name = Name, options = ReqOpts, hookspec = HookSpecs}) ->
-    _ = do_deinit(Name, ReqOpts),
     _ = may_unload_hooks(HookSpecs),
+    _ = do_deinit(Name, ReqOpts),
     _ = emqx_exhook_sup:stop_grpc_client_channel(Name),
     ok.
 
 do_deinit(Name, ReqOpts) ->
-    _ = do_call(Name, 'on_provider_unloaded', #{}, ReqOpts),
+    %% Using shorter timeout to deinit grpc server to avoid emqx_exhook_mngr
+    %% force killed by upper supervisor
+    _ = do_call(Name, 'on_provider_unloaded', #{}, ReqOpts#{timeout => 3000}),
     ok.
 
 do_init(ChannName, ReqOpts) ->

apps/emqx_exproto/src/emqx_exproto.app.src

@@ -1,6 +1,6 @@
 {application, emqx_exproto,
  [{description, "EMQ X Extension for Protocol"},
-  {vsn, "4.3.9"}, %% 4.3.3 is used by ee
+  {vsn, "4.3.10"}, %% 4.3.3 is used by ee
   {modules, []},
   {registered, []},
   {mod, {emqx_exproto_app, []}},

apps/emqx_exproto/src/emqx_exproto.appup.src

@@ -1,8 +1,12 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{<<"4\\.3\\.[2-8]">>,
+  [{"4.3.9",
-    [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[2-8]">>,
+    [{load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
      {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
    {<<"4\\.3\\.[0-1]">>,
     [{load_module,emqx_exproto_gsvr,brutal_purge,soft_purge,[]},
@@ -10,8 +14,12 @@
      {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
      {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{<<"4\\.3\\.[2-8]">>,
+  [{"4.3.9",
-    [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[2-8]">>,
+    [{load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
      {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
    {<<"4\\.3\\.[0-1]">>,
     [{load_module,emqx_exproto_gsvr,brutal_purge,soft_purge,[]},

apps/emqx_exproto/src/emqx_exproto_channel.erl

@@ -76,7 +76,8 @@
 
 -define(TIMER_TABLE, #{
           alive_timer => keepalive,
-          force_timer => force_close
+          force_timer => force_close,
+          idle_timer => force_close_idle
          }).
 
 -define(INFO_KEYS, [conninfo, conn_state, clientinfo, session, will_msg]).
@@ -94,6 +95,8 @@
          awaiting_rel_max
         ]).
 
+-define(DEFAULT_IDLE_TIMEOUT, 30000).
+
 %%--------------------------------------------------------------------
 %% Info, Attrs and Caps
 %%--------------------------------------------------------------------
@@ -148,9 +151,13 @@ init(ConnInfo = #{socktype := Socktype,
                   peercert := Peercert}, Options) ->
     GRpcChann = proplists:get_value(handler, Options),
     NConnInfo = default_conninfo(ConnInfo),
-    ClientInfo = default_clientinfo(ConnInfo),
+    ClientInfo = default_clientinfo(NConnInfo),
+
+    IdleTimeout = proplists:get_value(idle_timeout, Options, ?DEFAULT_IDLE_TIMEOUT),
+
+    NConnInfo1 = NConnInfo#{idle_timeout => IdleTimeout},
     Channel = #channel{gcli = #{channel => GRpcChann},
-                       conninfo = NConnInfo,
+                       conninfo = NConnInfo1,
                        clientinfo = ClientInfo,
                        conn_state = accepted,
                        timers = #{}
@@ -165,7 +172,8 @@ init(ConnInfo = #{socktype := Socktype,
                              #{socktype => socktype(Socktype),
                                peername => address(Peername),
                                sockname => address(Sockname)})},
-            try_dispatch(on_socket_created, wrap(Req), Channel)
+            start_idle_checking_timer(
+              try_dispatch(on_socket_created, wrap(Req), Channel))
     end.
 
 register_the_anonymous_client(ClientInfo, ConnInfo) ->
@@ -184,6 +192,12 @@ register_the_anonymous_client(ClientInfo, ConnInfo) ->
 unregister_the_anonymous_client(ClientId) ->
     emqx_cm:unregister_channel(ClientId).
 
+start_idle_checking_timer(Channel = #channel{conninfo = #{socktype := udp}}) ->
+    ensure_timer(idle_timer, Channel);
+
+start_idle_checking_timer(Channel) ->
+    Channel.
+
 %% @private
 peercert(NoSsl, ConnInfo) when NoSsl == nossl;
                                NoSsl == undefined ->
@@ -261,12 +275,17 @@ handle_timeout(_TRef, {keepalive, StatVal},
         {error, timeout} ->
             Req = #{type => 'KEEPALIVE'},
             NChannel = clean_timer(alive_timer, Channel),
-            {ok, try_dispatch(on_timer_timeout, wrap(Req), NChannel)}
+            %% close connection if keepalive timeout
+            Replies = [{event, disconnected}, {close, normal}],
+            {ok, Replies, try_dispatch(on_timer_timeout, wrap(Req), NChannel)}
     end;
 
 handle_timeout(_TRef, force_close, Channel = #channel{closed_reason = Reason}) ->
     {shutdown, Reason, Channel};
 
+handle_timeout(_TRef, force_close_idle, Channel) ->
+    {shutdown, idle_timeout, Channel};
+
 handle_timeout(_TRef, Msg, Channel) ->
     ?WARN("Unexpected timeout: ~p", [Msg]),
     {ok, Channel}.
@@ -328,7 +347,8 @@ handle_call({start_timer, keepalive, Interval},
     NConnInfo = ConnInfo#{keepalive => Interval},
     NClientInfo = ClientInfo#{keepalive => Interval},
     NChannel = Channel#channel{conninfo = NConnInfo, clientinfo = NClientInfo},
-    {reply, ok, [{event, updated}], ensure_keepalive(NChannel)};
+    {reply, ok, [{event, updated}],
+     ensure_keepalive(cancel_timer(idle_timer, NChannel))};
 
 handle_call({subscribe, TopicFilter, Qos},
             Channel = #channel{
@@ -363,7 +383,7 @@ handle_call({publish, Topic, Qos, Payload},
     end;
 
 handle_call(kick, Channel) ->
-    {shutdown, kicked, ok, Channel};
+    {reply, ok, [{event, disconnected}, {close, kicked}], Channel};
 
 handle_call(discard, Channel) ->
     {shutdown, discarded, ok, Channel};
@@ -561,6 +581,12 @@ reset_timer(Name, Channel) ->
 clean_timer(Name, Channel = #channel{timers = Timers}) ->
     Channel#channel{timers = maps:remove(Name, Timers)}.
 
+cancel_timer(Name, Channel = #channel{timers = Timers}) ->
+    emqx_misc:cancel_timer(maps:get(Name, Timers, undefined)),
+    clean_timer(Name, Channel).
+
+interval(idle_timer, #channel{conninfo = #{idle_timeout := IdleTimeout}}) ->
+    IdleTimeout;
 interval(force_timer, _) ->
     15000;
 interval(alive_timer, #channel{keepalive = Keepalive}) ->
@@ -609,9 +635,10 @@ enrich_clientinfo(InClientInfo = #{proto_name := ProtoName}, ClientInfo) ->
     NClientInfo = maps:merge(ClientInfo, maps:with(Ks, InClientInfo)),
     NClientInfo#{protocol => ProtoName}.
 
-default_conninfo(ConnInfo) ->
+default_conninfo(ConnInfo =
+                 #{peername := {PeerHost, PeerPort}}) ->
     ConnInfo#{clean_start => true,
-              clientid => undefined,
+              clientid => anonymous_clientid(PeerHost, PeerPort),
               username => undefined,
               conn_props => #{},
               connected => true,
@@ -622,13 +649,15 @@ default_conninfo(ConnInfo) ->
               receive_maximum => 0,
               expiry_interval => 0}.
 
-default_clientinfo(#{peername := {PeerHost, PeerPort},
+default_clientinfo(#{peername := {PeerHost, _},
-                     sockname := {_, SockPort}}) ->
+                     sockname := {_, SockPort},
+                     clientid := ClientId
+                    }) ->
     #{zone         => external,
       protocol     => exproto,
       peerhost     => PeerHost,
       sockport     => SockPort,
-      clientid     => anonymous_clientid(PeerHost, PeerPort),
+      clientid     => ClientId,
       username     => undefined,
       is_bridge    => false,
       is_superuser => false,

apps/emqx_exproto/src/emqx_exproto_gcli.erl

@@ -54,7 +54,13 @@ start_link(Pool, Id) ->
                           ?MODULE, [Pool, Id], []).
 
 async_call(FunName, Req = #{conn := Conn}, Options) ->
-    cast(pick(Conn), {rpc, FunName, Req, Options, self()}).
+    case pick(Conn) of
+        false ->
+            ?LOG(error, "No available grpc client for ~s: ~p",
+                 [FunName, Req]);
+        Pid when is_pid(Pid) ->
+            cast(Pid, {rpc, FunName, Req, Options, self()})
+    end.
 
 %%--------------------------------------------------------------------
 %% cast, pick
@@ -65,6 +71,7 @@ async_call(FunName, Req = #{conn := Conn}, Options) ->
 cast(Deliver, Msg) ->
     gen_server:cast(Deliver, Msg).
 
+-spec pick(term()) -> pid() | false.
 pick(Conn) ->
     gproc_pool:pick_worker(exproto_gcli_pool, Conn).
 

apps/emqx_exproto/test/emqx_exproto_SUITE.erl

@@ -213,11 +213,13 @@ t_keepalive_timeout(Cfg) ->
     send(Sock, ConnBin),
     {ok, ConnAckBin} = recv(Sock, 5000),
 
-    DisconnectBin = frame_disconnect(),
+    %% Timed out connections are closed immediately,
-    {ok, DisconnectBin} = recv(Sock, 10000),
+    %% so there may not be a disconnect message here
+    %%DisconnectBin = frame_disconnect(),
+    %%{ok, DisconnectBin} = recv(Sock, 10000),
 
     SockType =/= udp andalso begin
-        {error, closed} = recv(Sock, 5000)
+        {error, closed} = recv(Sock, 10000)
     end, ok.
 
 t_hook_connected_disconnected(Cfg) ->

apps/emqx_lwm2m/src/emqx_lwm2m.app.src

@@ -1,6 +1,6 @@
 {application,emqx_lwm2m,
  [{description,"EMQ X LwM2M Gateway"},
-  {vsn, "4.3.7"}, % strict semver, bump manually!
+  {vsn, "4.3.8"}, % strict semver, bump manually!
   {modules,[]},
   {registered,[emqx_lwm2m_sup]},
   {applications,[kernel,stdlib,lwm2m_coap]},

apps/emqx_lwm2m/src/emqx_lwm2m.appup.src

@@ -1,29 +1,75 @@
 %% -*- mode: erlang -*-
+%% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{<<"4\\.3\\.[0-1]">>,
+  [{"4.3.7",
-    [{restart_application,emqx_lwm2m}]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[0-1]">>,[{restart_application,emqx_lwm2m}]},
    {"4.3.2",
-    [{load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {<<"4\\.3\\.[3-4]">>,
-    [{load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {"4.3.5",
-    [{load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {"4.3.6",
-    [ %% There are only changes to the schema file, so we don't need any
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
-      %% commands here
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
-    ]}],
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
-  [{<<"4\\.3\\.[0-1]">>,
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
-    [{restart_application,emqx_lwm2m}]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]}]}],
+  [{"4.3.7",
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]}]},
+   {<<"4\\.3\\.[0-1]">>,[{restart_application,emqx_lwm2m}]},
    {"4.3.2",
-    [{load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {<<"4\\.3\\.[3-4]">>,
-    [{load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_protocol,brutal_purge,soft_purge,[]},
      {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
    {"4.3.5",
-    [{load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
-   {"4.3.6", []}]}.
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_api,brutal_purge,soft_purge,[]}]},
+   {"4.3.6",
+    [{load_module,emqx_lwm2m_xml_object_db,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_xml_object,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_message,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_json,brutal_purge,soft_purge,[]},
+     {load_module,emqx_lwm2m_cmd_handler,brutal_purge,soft_purge,[]}]}]}.

apps/emqx_lwm2m/src/emqx_lwm2m_cmd_handler.erl

@@ -106,9 +106,11 @@ coap_read_to_mqtt({ok, SuccessCode}, CoapPayload, Format, Ref) ->
         Result = coap_content_to_mqtt_payload(CoapPayload, Format, Ref),
         make_response(SuccessCode, Ref, Format, Result)
     catch
-        error:not_implemented -> make_response(not_implemented, Ref);
+        throw : {bad_request, Reason} ->
+            ?LOG(error, "bad_request, reason=~p, payload=~p", [Reason, CoapPayload]),
+            make_response(bad_request, Ref);
         C:R:Stack ->
-            ?LOG(error, "~p, bad payload format: ~p, stacktrace: ~p", [{C, R}, CoapPayload, Stack]),
+            ?LOG(error, "bad_request, error=~p, stacktrace=~p~npayload=~p", [{C, R}, Stack, CoapPayload]),
             make_response(bad_request, Ref)
     end.
 

apps/emqx_lwm2m/src/emqx_lwm2m_json.erl

@@ -29,7 +29,7 @@
 tlv_to_json(BaseName, TlvData) ->
     DecodedTlv = emqx_lwm2m_tlv:parse(TlvData),
     ObjectId = object_id(BaseName),
-    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def(ObjectId, true),
+    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def_assertive(ObjectId, true),
     case DecodedTlv of
         [#{tlv_resource_with_value:=Id, value:=Value}] ->
             TrueBaseName = basename(BaseName, undefined, undefined, Id, 3),
@@ -315,7 +315,7 @@ encode_int(Int) -> binary:encode_unsigned(Int).
 
 text_to_json(BaseName, Text) ->
     {ObjectId, ResourceId} = object_resource_id(BaseName),
-    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def(ObjectId, true),
+    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def_assertive(ObjectId, true),
     {K, V} = text_value(Text, ResourceId, ObjDefinition),
     #{bn=>BaseName, e=>[#{K=>V}]}.
 

apps/emqx_lwm2m/src/emqx_lwm2m_message.erl

@@ -33,7 +33,7 @@
 tlv_to_json(BaseName, TlvData) ->
     DecodedTlv = emqx_lwm2m_tlv:parse(TlvData),
     ObjectId = object_id(BaseName),
-    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def(ObjectId, true),
+    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def_assertive(ObjectId, true),
     case DecodedTlv of
         [#{tlv_resource_with_value:=Id, value:=Value}] ->
             TrueBaseName = basename(BaseName, undefined, undefined, Id, 3),
@@ -289,7 +289,7 @@ path([H|T], Acc) ->
 
 text_to_json(BaseName, Text) ->
     {ObjectId, ResourceId} = object_resource_id(BaseName),
-    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def(ObjectId, true),
+    ObjDefinition = emqx_lwm2m_xml_object:get_obj_def_assertive(ObjectId, true),
     Val = text_value(Text, ResourceId, ObjDefinition),
     [#{path => BaseName, value => Val}].
 

apps/emqx_lwm2m/src/emqx_lwm2m_xml_object.erl

@@ -20,6 +20,7 @@
 -include_lib("xmerl/include/xmerl.hrl").
 
 -export([ get_obj_def/2
+        , get_obj_def_assertive/2
         , get_object_id/1
         , get_object_name/1
         , get_object_and_resource_id/2
@@ -31,15 +32,19 @@
 -define(LOG(Level, Format, Args),
     logger:Level("LWM2M-OBJ: " ++ Format, Args)).
 
-% This module is for future use. Disabled now.
+get_obj_def_assertive(ObjectId, IsInt) ->
+    case get_obj_def(ObjectId, IsInt) of
+        {error, no_xml_definition} ->
+            erlang:throw({bad_request, {unknown_object_id, ObjectId}});
+        Xml ->
+            Xml
+    end.
 
 get_obj_def(ObjectIdInt, true) ->
     emqx_lwm2m_xml_object_db:find_objectid(ObjectIdInt);
 get_obj_def(ObjectNameStr, false) ->
     emqx_lwm2m_xml_object_db:find_name(ObjectNameStr).
 
-
-
 get_object_id(ObjDefinition) ->
     [#xmlText{value=ObjectId}] = xmerl_xpath:string("ObjectID/text()", ObjDefinition),
     ObjectId.

apps/emqx_lwm2m/src/emqx_lwm2m_xml_object_db.erl

@@ -69,12 +69,9 @@ find_name(Name) ->
                  end,
     case ets:lookup(?LWM2M_OBJECT_NAME_TO_ID_TAB, NameBinary) of
         [] ->
-            undefined;
+            {error, no_xml_definition};
         [{NameBinary, ObjectId}] ->
-            case ets:lookup(?LWM2M_OBJECT_DEF_TAB, ObjectId) of
+            find_objectid(ObjectId)
-                [] -> undefined;
-                [{ObjectId, Xml}] -> Xml
-            end
     end.
 
 stop() ->

apps/emqx_lwm2m/test/emqx_lwm2m_SUITE.erl

@@ -689,6 +689,60 @@ case10_read(Config) ->
                              }),
     ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)).
 
+case10_read_bad_request(Config) ->
+    UdpSock = ?config(sock, Config),
+    Epn = "urn:oma:lwm2m:oma:3",
+    MsgId1 = 15,
+    RespTopic = list_to_binary("lwm2m/"++Epn++"/up/resp"),
+    emqtt:subscribe(?config(emqx_c, Config), RespTopic, qos0),
+    timer:sleep(200),
+    % step 1, device register ...
+    test_send_coap_request( UdpSock,
+                            post,
+                            sprintf("coap://127.0.0.1:~b/rd?ep=~s&lt=345&lwm2m=1", [?PORT, Epn]),
+                            #coap_content{content_format = <<"text/plain">>,
+                            payload = <<"</lwm2m>;rt=\"oma.lwm2m\";ct=11543,</lwm2m/1/0>,</lwm2m/2/0>,</lwm2m/3/0>">>},
+                            [],
+                            MsgId1),
+    #coap_message{method = Method1} = test_recv_coap_response(UdpSock),
+    ?assertEqual({ok,created}, Method1),
+    test_recv_mqtt_response(RespTopic),
+
+    % step2,  send a READ command to device
+    CmdId = 206,
+    CommandTopic = <<"lwm2m/", (list_to_binary(Epn))/binary, "/dn/dm">>,
+    Command =   #{
+                    <<"requestID">> => CmdId, <<"cacheID">> => CmdId,
+                    <<"msgType">> => <<"read">>,
+                    <<"data">> => #{
+                        <<"path">> => <<"/3333/0/0">>
+                    }
+                },
+    CommandJson = emqx_json:encode(Command),
+    ?LOGT("CommandJson=~p", [CommandJson]),
+    test_mqtt_broker:publish(CommandTopic, CommandJson, 0),
+    timer:sleep(50),
+    Request2 = test_recv_coap_request(UdpSock),
+    #coap_message{method = Method2, payload=Payload2} = Request2,
+    ?LOGT("LwM2M client got ~p", [Request2]),
+    ?assertEqual(get, Method2),
+    ?assertEqual(<<>>, Payload2),
+    timer:sleep(50),
+
+    test_send_coap_response(UdpSock, "127.0.0.1", ?PORT, {ok, content}, #coap_content{content_format = <<"text/plain">>, payload = <<"EMQ">>}, Request2, true),
+    timer:sleep(100),
+
+    ReadResult = emqx_json:encode(#{  <<"requestID">> => CmdId, <<"cacheID">> => CmdId,
+                                <<"msgType">> => <<"read">>,
+                                <<"data">> => #{
+                                    <<"code">> => <<"4.00">>,
+                                    <<"codeMsg">> => <<"bad_request">>,
+                                    <<"reqPath">> => <<"/3333/0/0">>
+                                }
+                             }),
+    ?assertEqual(ReadResult, test_recv_mqtt_response(RespTopic)).
+
+
 case10_read_separate_ack(Config) ->
     UdpSock = ?config(sock, Config),
     Epn = "urn:oma:lwm2m:oma:3",

apps/emqx_management/src/emqx_mgmt_api.erl

@@ -24,6 +24,7 @@
 -export([ params2qs/2
         , node_query/4
         , node_query/5
+        , node_query/6
         , cluster_query/3
         , traverse_table/5
         , select_table/5
@@ -75,6 +76,11 @@ query_handle(Tables) ->
     Handles = lists:foldl(Fold, [], Tables),
     qlc:append(lists:reverse(Handles)).
 
+count_size(Table, undefined) ->
+    count(Table);
+count_size(_Table, CountFun) ->
+    CountFun().
+
 count(Table) when is_atom(Table) ->
     ets:info(Table, size);
 
@@ -112,9 +118,12 @@ limit(Params) ->
 %%--------------------------------------------------------------------
 
 node_query(Node, Params, {Tab, QsSchema}, QueryFun) ->
-    node_query(Node, Params, {Tab, QsSchema}, QueryFun, undefined).
+    node_query(Node, Params, {Tab, QsSchema}, QueryFun, undefined, undefined).
 
 node_query(Node, Params, {Tab, QsSchema}, QueryFun, SortFun) ->
+    node_query(Node, Params, {Tab, QsSchema}, QueryFun, SortFun, undefined).
+
+node_query(Node, Params, {Tab, QsSchema}, QueryFun, SortFun, CountFun) ->
     {CodCnt, Qs} = params2qs(Params, QsSchema),
     Limit = limit(Params),
     Page  = page(Params),
@@ -124,7 +133,7 @@ node_query(Node, Params, {Tab, QsSchema}, QueryFun, SortFun) ->
     {_, Rows} = do_query(Node, Qs, QueryFun, Start, Limit+1),
     Meta = #{page => Page, limit => Limit},
     NMeta = case CodCnt =:= 0 of
-                true -> Meta#{count => count(Tab), hasnext => length(Rows) > Limit};
+                true -> Meta#{count => count_size(Tab, CountFun), hasnext => length(Rows) > Limit};
                 _ -> Meta#{count => -1, hasnext => length(Rows) > Limit}
             end,
     Data0 = lists:sublist(Rows, Limit),

apps/emqx_management/src/emqx_mgmt_api_listeners.erl

@@ -72,5 +72,4 @@ format(Listeners) when is_list(Listeners) ->
     [ Info#{listen_on => list_to_binary(esockd:to_string(ListenOn))}
      || Info = #{listen_on := ListenOn} <- Listeners ];
 
-format({error, Reason}) -> [{error, Reason}].
+format({error, Reason}) -> [{error, iolist_to_binary(io_lib:format("~p", [Reason]))}].
-

apps/emqx_management/test/emqx_auth_mnesia_data_export_import_SUITE.erl

@@ -70,4 +70,10 @@ remove_all_users_and_acl() ->
     mnesia:delete_table(emqx_user),
     mnesia:delete_table(emqx_acl).
 
+-else.
+
+%% opensource edition
+
+all() -> [].
+
 -endif.

apps/emqx_management/test/emqx_mgmt_api_SUITE.erl

@@ -25,15 +25,12 @@
 -include_lib("emqx/include/emqx_mqtt.hrl").
 -include_lib("emqx_management/include/emqx_mgmt.hrl").
 
--define(CONTENT_TYPE, "application/x-www-form-urlencoded").
+-import(emqx_mgmt_api_test_helpers,
-
+        [request_api/3,
--define(HOST, "http://127.0.0.1:8081/").
+         request_api/4,
-
+         request_api/5,
--elvis([{elvis_style, line_length, disable}]).
+         auth_header_/0,
-
+         api_path/1]).
--define(API_VERSION, "v4").
-
--define(BASE_PATH, "api").
 
 all() ->
     emqx_ct:all(?MODULE).
@@ -790,49 +787,6 @@ t_keepalive(_Config) ->
     application:stop(emqx_dashboard),
     ok.
 
-request_api(Method, Url, Auth) ->
-    request_api(Method, Url, [], Auth, []).
-
-request_api(Method, Url, QueryParams, Auth) ->
-    request_api(Method, Url, QueryParams, Auth, []).
-
-request_api(Method, Url, QueryParams, Auth, []) ->
-    NewUrl = case QueryParams of
-                 "" -> Url;
-                 _ -> Url ++ "?" ++ QueryParams
-             end,
-    do_request_api(Method, {NewUrl, [Auth]});
-request_api(Method, Url, QueryParams, Auth, Body) ->
-    NewUrl = case QueryParams of
-                 "" -> Url;
-                 _ -> Url ++ "?" ++ QueryParams
-             end,
-    do_request_api(Method, {NewUrl, [Auth], "application/json", emqx_json:encode(Body)}).
-
-do_request_api(Method, Request)->
-    ct:pal("Method: ~p, Request: ~p", [Method, Request]),
-    case httpc:request(Method, Request, [], []) of
-        {error, socket_closed_remotely} ->
-            {error, socket_closed_remotely};
-        {ok, {{"HTTP/1.1", Code, _}, _, Return} }
-            when Code =:= 200 orelse Code =:= 201 ->
-            {ok, Return};
-        {ok, {Reason, _, _}} ->
-            {error, Reason}
-    end.
-
-auth_header_() ->
-    AppId = <<"admin">>,
-    AppSecret = <<"public">>,
-    auth_header_(binary_to_list(AppId), binary_to_list(AppSecret)).
-
-auth_header_(User, Pass) ->
-    Encoded = base64:encode_to_string(lists:append([User,":",Pass])),
-    {"Authorization","Basic " ++ Encoded}.
-
-api_path(Parts)->
-    ?HOST ++ filename:join([?BASE_PATH, ?API_VERSION] ++ Parts).
-
 filter(List, Key, Value) ->
     lists:filter(fun(Item) ->
         maps:get(Key, Item) == Value

apps/emqx_management/test/emqx_mgmt_api_test_helpers.erl

@@ -0,0 +1,69 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-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.
+%%--------------------------------------------------------------------
+
+-module(emqx_mgmt_api_test_helpers).
+
+-compile(export_all).
+-compile(nowarn_export_all).
+
+-define(HOST, "http://127.0.0.1:8081/").
+
+-define(API_VERSION, "v4").
+
+-define(BASE_PATH, "api").
+
+request_api(Method, Url, Auth) ->
+    request_api(Method, Url, [], Auth, []).
+
+request_api(Method, Url, QueryParams, Auth) ->
+    request_api(Method, Url, QueryParams, Auth, []).
+
+request_api(Method, Url, QueryParams, Auth, []) ->
+    NewUrl = case QueryParams of
+                 "" -> Url;
+                 _ -> Url ++ "?" ++ QueryParams
+             end,
+    do_request_api(Method, {NewUrl, [Auth]});
+request_api(Method, Url, QueryParams, Auth, Body) ->
+    NewUrl = case QueryParams of
+                 "" -> Url;
+                 _ -> Url ++ "?" ++ QueryParams
+             end,
+    do_request_api(Method, {NewUrl, [Auth], "application/json", emqx_json:encode(Body)}).
+
+do_request_api(Method, Request)->
+    ct:pal("Method: ~p, Request: ~p", [Method, Request]),
+    case httpc:request(Method, Request, [], []) of
+        {error, socket_closed_remotely} ->
+            {error, socket_closed_remotely};
+        {ok, {{"HTTP/1.1", Code, _}, _, Return} }
+            when Code =:= 200 orelse Code =:= 201 ->
+            {ok, Return};
+        {ok, {Reason, _, _}} ->
+            {error, Reason}
+    end.
+
+auth_header_() ->
+    AppId = <<"admin">>,
+    AppSecret = <<"public">>,
+    auth_header_(binary_to_list(AppId), binary_to_list(AppSecret)).
+
+auth_header_(User, Pass) ->
+    Encoded = base64:encode_to_string(lists:append([User,":",Pass])),
+    {"Authorization","Basic " ++ Encoded}.
+
+api_path(Parts)->
+    ?HOST ++ filename:join([?BASE_PATH, ?API_VERSION] ++ Parts).

apps/emqx_retainer/src/emqx_retainer.app.src

@@ -1,6 +1,6 @@
 {application, emqx_retainer,
  [{description, "EMQ X Retainer"},
-  {vsn, "4.4.1"}, % strict semver, bump manually!
+  {vsn, "4.4.2"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_retainer_sup]},
   {applications, [kernel,stdlib]},

apps/emqx_retainer/src/emqx_retainer.appup.src

@@ -1,7 +1,12 @@
 %% -*- mode: erlang -*-
+%% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.0",[{load_module,emqx_retainer_cli,brutal_purge,soft_purge,[]}]},
+  [{"4.4.1",[{load_module,emqx_retainer_sup,brutal_purge,soft_purge,[]}]},
+   {"4.4.0",[{load_module,emqx_retainer_cli,brutal_purge,soft_purge,[]},
+             {load_module,emqx_retainer_sup,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.4.0",[{load_module,emqx_retainer_cli,brutal_purge,soft_purge,[]}]},
+  [{"4.4.1",[{load_module,emqx_retainer_sup,brutal_purge,soft_purge,[]}]},
+   {"4.4.0",[{load_module,emqx_retainer_sup,brutal_purge,soft_purge,[]},
+            {load_module,emqx_retainer_cli,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}]
 }.

apps/emqx_retainer/src/emqx_retainer_sup.erl

@@ -32,5 +32,8 @@ init([Env]) ->
              restart  => permanent,
              shutdown => 5000,
              type     => worker,
-             modules  => [emqx_retainer]}]}}.
+             modules  => [emqx_retainer]} || not is_managed_by_modules()]}}.
 
+is_managed_by_modules() ->
+    %% always false for opensource edition
+    false.

apps/emqx_retainer/test/emqx_retainer_SUITE.erl

@@ -31,11 +31,12 @@ all() -> emqx_ct:all(?MODULE).
 %%--------------------------------------------------------------------
 
 init_per_suite(Config) ->
-    emqx_ct_helpers:start_apps([emqx_retainer]),
+    emqx_retainer_ct_helper:ensure_start(),
     Config.
 
 end_per_suite(_Config) ->
-    emqx_ct_helpers:stop_apps([emqx_retainer]).
+    emqx_retainer_ct_helper:ensure_stop(),
+    ok.
 
 init_per_testcase(TestCase, Config) ->
     emqx_retainer:clean(<<"#">>),
@@ -207,4 +208,3 @@ receive_messages(Count, Msgs) ->
     after 2000 ->
             Msgs
     end.
-

apps/emqx_retainer/test/emqx_retainer_cli_SUITE.erl

@@ -24,11 +24,12 @@
 all() -> emqx_ct:all(?MODULE).
 
 init_per_suite(Config) ->
-    emqx_ct_helpers:start_apps([emqx_retainer]),
+    emqx_retainer_ct_helper:ensure_start(),
     Config.
 
 end_per_suite(_Config) ->
-    emqx_ct_helpers:stop_apps([emqx_retainer]).
+    emqx_retainer_ct_helper:ensure_stop(),
+    ok.
 
 init_per_testcase(_TestCase, Config) ->
     Config.

apps/emqx_retainer/test/emqx_retainer_ct_helper.erl

@@ -0,0 +1,46 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-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.
+%%--------------------------------------------------------------------
+-module(emqx_retainer_ct_helper).
+
+
+%% API
+-export([ensure_start/0, ensure_stop/0]).
+-ifdef(EMQX_ENTERPRISE).
+ensure_start() ->
+    application:stop(emqx_modules),
+    init_conf(),
+    emqx_ct_helpers:start_apps([emqx_retainer]),
+    ok.
+
+-else.
+
+ensure_start() ->
+    init_conf(),
+    emqx_ct_helpers:start_apps([emqx_retainer]),
+    ok.
+
+-endif.
+
+ensure_stop() ->
+    emqx_ct_helpers:stop_apps([emqx_retainer]).
+
+
+init_conf() ->
+    application:set_env(emqx_retainer, expiry_interval, 0),
+    application:set_env(emqx_retainer, max_payload_size, 1024000),
+    application:set_env(emqx_retainer, max_retained_messages, 0),
+    application:set_env(emqx_retainer, storage_type, ram),
+    ok.

apps/emqx_retainer/test/mqtt_protocol_v5_SUITE.erl

@@ -27,12 +27,13 @@ init_per_suite(Config) ->
     %% Meck emqtt
     ok = meck:new(emqtt, [non_strict, passthrough, no_history, no_link]),
     %% Start Apps
-    emqx_ct_helpers:start_apps([emqx_retainer]),
+    emqx_retainer_ct_helper:ensure_start(),
     Config.
 
 end_per_suite(_Config) ->
     ok = meck:unload(emqtt),
-    emqx_ct_helpers:stop_apps([emqx_retainer]).
+    emqx_retainer_ct_helper:ensure_stop().
+
 
 %%--------------------------------------------------------------------
 %% Helpers

apps/emqx_rule_engine/include/rule_actions.hrl

@@ -1,3 +1,19 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-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.
+%%--------------------------------------------------------------------
+
 -compile({parse_transform, emqx_rule_actions_trans}).
 
 -type selected_data() :: map().
@@ -6,6 +22,9 @@
 
 -define(BINDING_KEYS, '__bindings__').
 
+-define(LOG_RULE_ACTION(Level, Metadata, Fmt, Args),
+        emqx_rule_utils:log_action(Level, Metadata, Fmt, Args)).
+
 -define(bound_v(Key, ENVS0),
     maps:get(Key,
         maps:get(?BINDING_KEYS, ENVS0, #{}))).

apps/emqx_rule_engine/src/emqx_rule_actions.erl

@@ -171,9 +171,13 @@ on_action_create_republish(Id, Params = #{
 on_action_republish(_Selected, Envs = #{
             topic := Topic,
             headers := #{republish_by := ActId},
-            ?BINDING_KEYS := #{'Id' := ActId}
+            ?BINDING_KEYS := #{'Id' := ActId},
+            metadata := Metadata
         }) ->
-    ?LOG(error, "[republish] recursively republish detected, msg topic: ~p, target topic: ~p",
+    ?LOG_RULE_ACTION(
+        error,
+        Metadata,
+        "[republish] recursively republish detected, msg topic: ~p, target topic: ~p",
         [Topic, ?bound_v('TargetTopic', Envs)]),
     emqx_rule_metrics:inc_actions_error(?bound_v('Id', Envs)),
     {badact, recursively_republish};
@@ -186,8 +190,9 @@ on_action_republish(Selected, _Envs = #{
                 'TargetQoS' := TargetQoS,
                 'TopicTks' := TopicTks,
                 'PayloadTks' := PayloadTks
-            } = Bindings}) ->
+            } = Bindings,
-    ?LOG(debug, "[republish] republish to: ~p, Selected: ~p", [TargetTopic, Selected]),
+            metadata := Metadata}) ->
+    ?LOG_RULE_ACTION(debug, Metadata, "[republish] republish to: ~p, Selected: ~p", [TargetTopic, Selected]),
     TargetRetain = maps:get('TargetRetain', Bindings, false),
     Message =
         #message{
@@ -210,8 +215,9 @@ on_action_republish(Selected, _Envs = #{
                 'TargetQoS' := TargetQoS,
                 'TopicTks' := TopicTks,
                 'PayloadTks' := PayloadTks
-            } = Bindings}) ->
+            } = Bindings,
-    ?LOG(debug, "[republish] republish to: ~p, Selected: ~p", [TargetTopic, Selected]),
+            metadata := Metadata}) ->
+    ?LOG_RULE_ACTION(debug, Metadata, "[republish] republish to: ~p, Selected: ~p", [TargetTopic, Selected]),
     TargetRetain = maps:get('TargetRetain', Bindings, false),
     Message =
         #message{

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -1,10 +1,16 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{"4.4.7",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+  [{<<"4\\.4\\.[6-7]">>,
-   {"4.4.6",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+   [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+    {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
+    {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.5",
     [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.4.4",
@@ -73,11 +79,17 @@
      {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{"4.4.7",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+  [{<<"4\\.4\\.[6-7]">>,
-   {"4.4.6",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+   [{load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+    {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
+    {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
+    {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.4.5",
     [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_utils,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_validator,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]},
      {load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]}]},
    {"4.4.4",
     [{load_module,emqx_rule_engine_api,brutal_purge,soft_purge,[]},

apps/emqx_rule_engine/src/emqx_rule_runtime.erl

@@ -17,6 +17,7 @@
 -module(emqx_rule_runtime).
 
 -include("rule_engine.hrl").
+-include("rule_actions.hrl").
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/logger.hrl").
 
@@ -54,36 +55,37 @@ apply_rules([Rule|More], Input) ->
     apply_rule(Rule, Input),
     apply_rules(More, Input).
 
-apply_rule(Rule = #rule{id = RuleID}, Input) ->
+apply_rule(Rule = #rule{id = RuleId}, Input) ->
     clear_rule_payload(),
-    ok = emqx_rule_metrics:inc_rules_matched(RuleID),
+    ok = emqx_rule_metrics:inc_rules_matched(RuleId),
-    try do_apply_rule(Rule, add_metadata(Input, #{rule_id => RuleID}))
+    %% Add metadata here caused we need support `metadata` and `rule_id` in SQL
+    try do_apply_rule(Rule, emqx_rule_utils:add_metadata(Input, #{rule_id => RuleId}))
     catch
         %% ignore the errors if select or match failed
         _:Reason = {select_and_transform_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
+            emqx_rule_metrics:inc_rules_exception(RuleId),
             ?LOG(warning, "SELECT clause exception for ~s failed: ~p",
-                 [RuleID, Error]),
+                 [RuleId, Error]),
             {error, Reason};
         _:Reason = {match_conditions_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
+            emqx_rule_metrics:inc_rules_exception(RuleId),
             ?LOG(warning, "WHERE clause exception for ~s failed: ~p",
-                 [RuleID, Error]),
+                 [RuleId, Error]),
             {error, Reason};
         _:Reason = {select_and_collect_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
+            emqx_rule_metrics:inc_rules_exception(RuleId),
             ?LOG(warning, "FOREACH clause exception for ~s failed: ~p",
-                 [RuleID, Error]),
+                 [RuleId, Error]),
             {error, Reason};
         _:Reason = {match_incase_error, Error} ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
+            emqx_rule_metrics:inc_rules_exception(RuleId),
             ?LOG(warning, "INCASE clause exception for ~s failed: ~p",
-                 [RuleID, Error]),
+                 [RuleId, Error]),
             {error, Reason};
         _:Error:StkTrace ->
-            emqx_rule_metrics:inc_rules_exception(RuleID),
+            emqx_rule_metrics:inc_rules_exception(RuleId),
             ?LOG(error, "Apply rule ~s failed: ~p. Stacktrace:~n~p",
-                 [RuleID, Error, StkTrace]),
+                 [RuleId, Error, StkTrace]),
             {error, {Error, StkTrace}}
     end.
 
@@ -216,10 +218,8 @@ match_conditions({}, _Data) ->
     true.
 
 %% comparing numbers against strings
-compare(Op, undefined, undefined) ->
+compare(Op, L, R) when L == undefined; R == undefined ->
-    do_compare(Op, undefined, undefined);
+    do_compare(Op, L, R);
-compare(_Op, L, R) when L == undefined; R == undefined ->
-    false;
 compare(Op, L, R) when is_number(L), is_binary(R) ->
     do_compare(Op, L, number(R));
 compare(Op, L, R) when is_binary(L), is_number(R) ->
@@ -232,10 +232,14 @@ compare(Op, L, R) ->
     do_compare(Op, L, R).
 
 do_compare('=', L, R) -> L == R;
+do_compare('>', L, R) when L == undefined; R == undefined -> false;
 do_compare('>', L, R) -> L > R;
+do_compare('<', L, R) when L == undefined; R == undefined -> false;
 do_compare('<', L, R) -> L < R;
-do_compare('<=', L, R) -> L =< R;
+do_compare('<=', L, R) ->
-do_compare('>=', L, R) -> L >= R;
+    do_compare('=', L, R) orelse do_compare('<', L, R);
+do_compare('>=', L, R) ->
+    do_compare('=', L, R) orelse do_compare('>', L, R);
 do_compare('<>', L, R) -> L /= R;
 do_compare('!=', L, R) -> L /= R;
 do_compare('=~', T, F) -> emqx_topic:match(T, F).
@@ -245,9 +249,10 @@ number(Bin) ->
     catch error:badarg -> binary_to_float(Bin)
     end.
 
-%% Step3 -> Take actions
+%% %% Step3 -> Take actions
+%% fallback actions already have `rule_id` in `metadata`
 take_actions(Actions, Selected, Envs, OnFailed) ->
-    [take_action(ActInst, Selected, Envs, OnFailed, ?ActionMaxRetry)
+    [take_action(ActInst, Selected, emqx_rule_utils:add_metadata(Envs, ActInst), OnFailed, ?ActionMaxRetry)
      || ActInst <- Actions].
 
 take_action(#action_instance{id = Id, name = ActName, fallbacks = Fallbacks} = ActInst,
@@ -312,12 +317,12 @@ wait_action_on(Id, RetryN) ->
             end
     end.
 
-handle_action_failure(continue, Id, Fallbacks, Selected, Envs, Reason) ->
+handle_action_failure(continue, _Id, Fallbacks, Selected, Envs = #{metadata := Metadata}, Reason) ->
-    ?LOG(error, "Take action ~p failed, continue next action, reason: ~0p", [Id, Reason]),
+    ?LOG_RULE_ACTION(error, Metadata, "Continue next action, reason: ~0p", [Reason]),
     _ = take_actions(Fallbacks, Selected, Envs, continue),
     failed;
-handle_action_failure(stop, Id, Fallbacks, Selected, Envs, Reason) ->
+handle_action_failure(stop, Id, Fallbacks, Selected, Envs = #{metadata := Metadata}, Reason) ->
-    ?LOG(error, "Take action ~p failed, skip all actions, reason: ~0p", [Id, Reason]),
+    ?LOG_RULE_ACTION(error, Metadata, "Skip all actions, reason: ~0p", [Reason]),
     _ = take_actions(Fallbacks, Selected, Envs, continue),
     error({take_action_failed, {Id, Reason}}).
 
@@ -429,10 +434,6 @@ do_apply_func(Name, Args, Input) ->
         Result -> Result
     end.
 
-add_metadata(Input, Metadata) when is_map(Input), is_map(Metadata) ->
-    NewMetadata = maps:merge(maps:get(metadata, Input, #{}), Metadata),
-    Input#{metadata => NewMetadata}.
-
 %%------------------------------------------------------------------------------
 %% Internal Functions
 %%------------------------------------------------------------------------------

apps/emqx_rule_engine/src/emqx_rule_utils.erl

@@ -16,6 +16,9 @@
 
 -module(emqx_rule_utils).
 
+-include("rule_engine.hrl").
+-include_lib("emqx/include/logger.hrl").
+
 -export([ replace_var/2
         ]).
 
@@ -59,6 +62,10 @@
         , can_topic_match_oneof/2
         ]).
 
+-export([ add_metadata/2
+        , log_action/4
+        ]).
+
 -compile({no_auto_import,
           [ float/1
           ]}).
@@ -371,3 +378,30 @@ can_topic_match_oneof(Topic, Filters) ->
     lists:any(fun(Fltr) ->
         emqx_topic:match(Topic, Fltr)
     end, Filters).
+
+add_metadata(Envs, Metadata) when is_map(Envs), is_map(Metadata) ->
+    NMetadata = maps:merge(maps:get(metadata, Envs, #{}), Metadata),
+    Envs#{metadata => NMetadata};
+add_metadata(Envs, Action) when is_map(Envs), is_record(Action, action_instance)->
+    Metadata = gen_metadata_from_action(Action),
+    NMetadata = maps:merge(maps:get(metadata, Envs, #{}), Metadata),
+    Envs#{metadata => NMetadata}.
+
+gen_metadata_from_action(#action_instance{name = Name, args = undefined}) ->
+    #{action_name => Name, resource_id => undefined};
+gen_metadata_from_action(#action_instance{name = Name, args = Args})
+  when is_map(Args) ->
+    #{action_name => Name, resource_id => maps:get(<<"$resource">>, Args, undefined)};
+gen_metadata_from_action(#action_instance{name = Name}) ->
+    #{action_name => Name, resource_id => undefined}.
+
+log_action(Level, Metadata, Fmt, Args) ->
+    ?LOG(Level,
+         "Rule: ~p; Action: ~p; Resource: ~p. " ++ Fmt,
+         metadata_values(Metadata) ++ Args).
+
+metadata_values(Metadata) ->
+    RuleId = maps:get(rule_id, Metadata, undefined),
+    ActionName = maps:get(action_name, Metadata, undefined),
+    ResourceName = maps:get(resource_id, Metadata, undefined),
+    [RuleId, ActionName, ResourceName].

apps/emqx_rule_engine/test/emqx_rule_engine_SUITE.erl

@@ -126,6 +126,11 @@ groups() ->
        t_sqlparse_array_range_1,
        t_sqlparse_array_range_2,
        t_sqlparse_true_false,
+       t_sqlparse_compare_undefined,
+       t_sqlparse_compare_null_null,
+       t_sqlparse_compare_null_notnull,
+       t_sqlparse_compare_notnull_null,
+       t_sqlparse_compare,
        t_sqlparse_new_map,
        t_sqlparse_invalid_json
       ]},
@@ -2514,6 +2519,235 @@ t_sqlparse_true_false(_Config) ->
                    <<"c">> := [true]
                    }, Res00).
 
+-define(TEST_SQL(SQL),
+    emqx_rule_sqltester:test(
+        #{<<"rawsql">> => SQL,
+          <<"ctx">> => #{<<"payload">> => <<"{}">>,
+                         <<"topic">> => <<"t/a">>}})).
+
+t_sqlparse_compare_undefined(_Config) ->
+    Sql00 = "select "
+            " * "
+            "from \"t/#\" "
+            "where dev != undefined ",
+    %% no match
+    ?assertMatch({error, nomatch}, ?TEST_SQL(Sql00)),
+
+    Sql01 = "select "
+            " 'd' as dev "
+            "from \"t/#\" "
+            "where dev != undefined ",
+    {ok, Res01} = ?TEST_SQL(Sql01),
+    %% pass
+    ?assertMatch(#{}, Res01),
+
+    Sql02 = "select "
+            " * "
+            "from \"t/#\" "
+            "where dev != 'undefined' ",
+    {ok, Res02} = ?TEST_SQL(Sql02),
+    %% pass
+    ?assertMatch(#{}, Res02).
+
+t_sqlparse_compare_null_null(_Config) ->
+    %% test undefined == undefined
+    Sql00 = "select "
+            " a = b as c "
+            "from \"t/#\" ",
+    {ok, Res00} = ?TEST_SQL(Sql00),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res00),
+
+    %% test undefined != undefined
+    Sql01 = "select "
+            " a != b as c "
+            "from \"t/#\" ",
+    {ok, Res01} = ?TEST_SQL(Sql01),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res01),
+
+    %% test undefined > undefined
+    Sql02 = "select "
+            " a > b as c "
+            "from \"t/#\" ",
+    {ok, Res02} = ?TEST_SQL(Sql02),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res02),
+
+    %% test undefined < undefined
+    Sql03 = "select "
+            " a < b as c "
+            "from \"t/#\" ",
+    {ok, Res03} = ?TEST_SQL(Sql03),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res03),
+
+    %% test undefined <= undefined
+    Sql04 = "select "
+            " a <= b as c "
+            "from \"t/#\" ",
+    {ok, Res04} = ?TEST_SQL(Sql04),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res04),
+
+    %% test undefined >= undefined
+    Sql05 = "select "
+            " a >= b as c "
+            "from \"t/#\" ",
+    {ok, Res05} = ?TEST_SQL(Sql05),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res05).
+
+t_sqlparse_compare_null_notnull(_Config) ->
+    %% test undefined == b
+    Sql00 = "select "
+            " 'b' as b, a = b as c "
+            "from \"t/#\" ",
+    {ok, Res00} = ?TEST_SQL(Sql00),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res00),
+
+    %% test undefined != b
+    Sql01 = "select "
+            " 'b' as b, a != b as c "
+            "from \"t/#\" ",
+    {ok, Res01} = ?TEST_SQL(Sql01),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res01),
+
+    %% test undefined > b
+    Sql02 = "select "
+            " 'b' as b, a > b as c "
+            "from \"t/#\" ",
+    {ok, Res02} = ?TEST_SQL(Sql02),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res02),
+
+    %% test undefined < b
+    Sql03 = "select "
+            " 'b' as b, a < b as c "
+            "from \"t/#\" ",
+    {ok, Res03} = ?TEST_SQL(Sql03),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res03),
+
+    %% test undefined <= b
+    Sql04 = "select "
+            " 'b' as b, a <= b as c "
+            "from \"t/#\" ",
+    {ok, Res04} = ?TEST_SQL(Sql04),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res04),
+
+    %% test undefined >= b
+    Sql05 = "select "
+            " 'b' as b, a >= b as c "
+            "from \"t/#\" ",
+    {ok, Res05} = ?TEST_SQL(Sql05),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res05).
+
+t_sqlparse_compare_notnull_null(_Config) ->
+    %% test 'a' == undefined
+    Sql00 = "select "
+            " 'a' as a, a = b as c "
+            "from \"t/#\" ",
+    {ok, Res00} = ?TEST_SQL(Sql00),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res00),
+
+    %% test 'a' != undefined
+    Sql01 = "select "
+            " 'a' as a, a != b as c "
+            "from \"t/#\" ",
+    {ok, Res01} = ?TEST_SQL(Sql01),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res01),
+
+    %% test 'a' > undefined
+    Sql02 = "select "
+            " 'a' as a, a > b as c "
+            "from \"t/#\" ",
+    {ok, Res02} = ?TEST_SQL(Sql02),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res02),
+
+    %% test 'a' < undefined
+    Sql03 = "select "
+            " 'a' as a, a < b as c "
+            "from \"t/#\" ",
+    {ok, Res03} = ?TEST_SQL(Sql03),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res03),
+
+    %% test 'a' <= undefined
+    Sql04 = "select "
+            " 'a' as a, a <= b as c "
+            "from \"t/#\" ",
+    {ok, Res04} = ?TEST_SQL(Sql04),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res04),
+
+    %% test 'a' >= undefined
+    Sql05 = "select "
+            " 'a' as a, a >= b as c "
+            "from \"t/#\" ",
+    {ok, Res05} = ?TEST_SQL(Sql05),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res05).
+
+t_sqlparse_compare(_Config) ->
+    Sql00 = "select "
+            " 'a' as a, 'a' as b, a = b as c "
+            "from \"t/#\" ",
+    {ok, Res00} = ?TEST_SQL(Sql00),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res00),
+
+    Sql01 = "select "
+            " is_null(a) as c "
+            "from \"t/#\" ",
+    {ok, Res01} = ?TEST_SQL(Sql01),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res01),
+
+    Sql02 = "select "
+            " 1 as a, 2 as b, a < b as c "
+            "from \"t/#\" ",
+    {ok, Res02} = ?TEST_SQL(Sql02),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res02),
+
+    Sql03 = "select "
+            " 1 as a, 2 as b, a > b as c "
+            "from \"t/#\" ",
+    {ok, Res03} = ?TEST_SQL(Sql03),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res03),
+
+    Sql04 = "select "
+            " 1 as a, 2 as b, a = b as c "
+            "from \"t/#\" ",
+    {ok, Res04} = ?TEST_SQL(Sql04),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res04),
+
+    %% test 'a' >= undefined
+    Sql05 = "select "
+            " 1 as a, 2 as b, a >= b as c "
+            "from \"t/#\" ",
+    {ok, Res05} = ?TEST_SQL(Sql05),
+    ?assertMatch(#{<<"c">> := false
+                   }, Res05),
+
+    %% test 'a' >= undefined
+    Sql06 = "select "
+            " 1 as a, 2 as b, a <= b as c "
+            "from \"t/#\" ",
+    {ok, Res06} = ?TEST_SQL(Sql06),
+    ?assertMatch(#{<<"c">> := true
+                   }, Res06).
+
 t_sqlparse_new_map(_Config) ->
     %% construct a range without 'as'
     Sql00 = "select "

apps/emqx_sn/examples/simple_example2.erl

@@ -6,6 +6,7 @@
 -define(PORT, 1884).
 
 -export([start/0]).
+-export([gen_register_packet/2]).
 
 start() ->
     io:format("start to connect ~p:~p~n", [?HOST, ?PORT]),

apps/emqx_sn/examples/simple_example3.erl

@@ -6,6 +6,7 @@
 -define(PORT, 1884).
 
 -export([start/0]).
+-export([gen_register_packet/2]).
 
 start() ->
     io:format("start to connect ~p:~p~n", [?HOST, ?PORT]),

apps/emqx_sn/examples/simple_example4.erl

@@ -5,7 +5,7 @@
 -define(HOST, {127,0,0,1}).
 -define(PORT, 1884).
 
--export([start/0]).
+-export([start/1]).
 
 start(LoopTimes) ->
     io:format("start to connect ~p:~p~n", [?HOST, ?PORT]),

apps/emqx_web_hook/src/emqx_web_hook.app.src

@@ -1,6 +1,6 @@
 {application, emqx_web_hook,
  [{description, "EMQ X WebHook Plugin"},
-  {vsn, "4.3.13"}, % strict semver, bump manually!
+  {vsn, "4.3.14"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_web_hook_sup]},
   {applications, [kernel,stdlib,ehttpc]},

apps/emqx_web_hook/src/emqx_web_hook.appup.src

@@ -1,12 +1,7 @@
 %% -*- mode: erlang -*-
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
-  [{<<"4\\.3\\.[0-2]">>,
+  [{<<"4\\.3\\.[0-7]">>,
-    [{apply,{application,stop,[emqx_web_hook]}},
-     {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {<<"4\\.3\\.[3-7]">>,
     [{apply,{application,stop,[emqx_web_hook]}},
      {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook,brutal_purge,soft_purge,[]},
@@ -26,15 +21,10 @@
    {"4.3.11",
     [{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook,brutal_purge,soft_purge,[]}]},
-   {"4.3.12",
+   {<<"4\\.3\\.1[2-3]">>,
     [{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
-  [{<<"4\\.3\\.[0-2]">>,
+  [{<<"4\\.3\\.[0-7]">>,
-    [{apply,{application,stop,[emqx_web_hook]}},
-     {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook,brutal_purge,soft_purge,[]},
-     {load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
-   {<<"4\\.3\\.[3-7]">>,
     [{apply,{application,stop,[emqx_web_hook]}},
      {load_module,emqx_web_hook_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook,brutal_purge,soft_purge,[]},
@@ -54,6 +44,6 @@
    {"4.3.11",
     [{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]},
      {load_module,emqx_web_hook,brutal_purge,soft_purge,[]}]},
-   {"4.3.12",
+   {<<"4\\.3\\.1[2-3]">>,
     [{load_module,emqx_web_hook_actions,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}]}.

apps/emqx_web_hook/src/emqx_web_hook_actions.erl

@@ -259,25 +259,27 @@ on_action_data_to_webserver(Selected, _Envs =
                                 'BodyTokens' := BodyTokens,
                                 'Pool' := Pool,
                                 'RequestTimeout' := RequestTimeout},
-                              clientid := ClientID}) ->
+                              clientid := ClientID,
+                              metadata := Metadata}) ->
     NBody = format_msg(BodyTokens, clear_user_property_header(Selected)),
     NPath = emqx_rule_utils:proc_tmpl(PathTokens, Selected),
     Req = create_req(Method, NPath, Headers, NBody),
     case ehttpc:request({Pool, ClientID}, Method, Req, RequestTimeout) of
         {ok, StatusCode, _} when StatusCode >= 200 andalso StatusCode < 300 ->
+            ?LOG_RULE_ACTION(debug, Metadata, "HTTP Request succeeded with path: ~p status code ~p", [NPath, StatusCode]),
             emqx_rule_metrics:inc_actions_success(Id);
         {ok, StatusCode, _, _} when StatusCode >= 200 andalso StatusCode < 300 ->
             emqx_rule_metrics:inc_actions_success(Id);
         {ok, StatusCode, _} ->
-            ?LOG(warning, "HTTP request failed with path: ~p status code: ~p", [NPath, StatusCode]),
+            ?LOG_RULE_ACTION(warning, Metadata, "HTTP request failed with path: ~p status code: ~p", [NPath, StatusCode]),
             emqx_rule_metrics:inc_actions_error(Id),
             {badact, StatusCode};
         {ok, StatusCode, _, _} ->
-            ?LOG(warning, "HTTP request failed with path: ~p status code: ~p", [NPath, StatusCode]),
+            ?LOG_RULE_ACTION(warning, Metadata, "HTTP request failed with path: ~p status code: ~p", [NPath, StatusCode]),
             emqx_rule_metrics:inc_actions_error(Id),
             {badact, StatusCode};
         {error, Reason} ->
-            ?LOG(error, "HTTP request failed path: ~p error: ~p", [NPath, Reason]),
+            ?LOG_RULE_ACTION(error, Metadata, "HTTP request failed path: ~p error: ~p", [NPath, Reason]),
             emqx_rule_metrics:inc_actions_error(Id),
             {badact, Reason}
     end.

bin/emqx

@@ -38,9 +38,20 @@ export PROGNAME="erl"
 DYNLIBS_DIR="$RUNNER_ROOT_DIR/dynlibs"
 ERTS_LIB_DIR="$ERTS_DIR/../lib"
 
+# Fix bin permission for all erts bin files
+# the 'x' attributes may get lost if the files are extracted from a relup package
+find "$BINDIR" -exec chmod a+x {} \;
+
 # Echo to stderr on errors
 echoerr() { echo "$*" 1>&2; }
 
+die() {
+    set +x
+    echoerr "ERROR: $1"
+    errno=${2:-1}
+    exit "$errno"
+}
+
 assert_node_alive() {
     if ! relx_nodetool "ping" > /dev/null; then
         die "node_is_not_running!" 1
@@ -48,9 +59,8 @@ assert_node_alive() {
 }
 
 check_erlang_start() {
-    # Fix bin permission
+    # set ERL_CRASH_DUMP_BYTES to zero so it will not write a crash dump file
-    find "$BINDIR" ! -executable -exec chmod a+x {} \;
+    env ERL_CRASH_DUMP_BYTES=0 "$BINDIR/$PROGNAME" -boot "$REL_DIR/start_clean" -eval "crypto:start(),halt()"
-    "$BINDIR/$PROGNAME" -boot "$REL_DIR/start_clean" -eval "crypto:start(),halt()"
 }
 
 if ! check_erlang_start >/dev/null 2>&1; then

build

@@ -66,36 +66,32 @@ make_rel() {
     ./rebar3 as "$PROFILE" tar
 }
 
+relup_db() {
+    ./scripts/relup-base-vsns.escript "$@" ./data/relup-paths.eterm
+}
+
 ## unzip previous version .zip files to _build/$PROFILE/rel/emqx/releases before making relup
 make_relup() {
     local lib_dir="_build/$PROFILE/rel/emqx/lib"
     local releases_dir="_build/$PROFILE/rel/emqx/releases"
-    local name_pattern
+    local zip_file
-    # 4.5.0 upgraded OTP from 24.1.5-3 to 24.3.4.2-1, so we need to
-    # fool the matcher to make it pick up the previous 4.4.X releases
-    # with older OTP.
-    if [[ "${PKG_VSN}" == 4.5.0* ]]; then
-      name_pattern="${PROFILE}-$(env OTP_VSN=24.1.5-3 ./scripts/pkg-full-vsn.sh 'vsn_matcher')"
-    else
-      name_pattern="${PROFILE}-$(./scripts/pkg-full-vsn.sh 'vsn_matcher')"
-    fi
     mkdir -p "$lib_dir" "$releases_dir" '_upgrade_base'
     local releases=()
     if [ -d "$releases_dir" ]; then
-        while read -r zip; do
+        for BASE_VSN in $(relup_db base-vsns "$PKG_VSN"); do
-            local base_vsn
+            OTP_BASE=$(relup_db otp-vsn-for "$PKG_VSN")
-            base_vsn="$(echo "$zip" | grep -oE "[0-9]+\.[0-9]+\.[0-9]+(-[0-9a-f]{8})?" | head -1)"
+            zip_file="_upgrade_base/${PROFILE}-$(env OTP_VSN="$OTP_BASE" PKG_VSN="$BASE_VSN" ./scripts/pkg-full-vsn.sh 'vsn_exact').zip"
-            if [ ! -d "$releases_dir/$base_vsn" ]; then
+            if [ ! -d "$releases_dir/$BASE_VSN" ]; then
                 local tmp_dir
                 tmp_dir="$(mktemp -d -t emqx.XXXXXXX)"
-                unzip -q "$zip" "emqx/releases/*" -d "$tmp_dir"
+                unzip -q "$zip_file" "emqx/releases/*" -d "$tmp_dir"
-                unzip -q "$zip" "emqx/lib/*" -d "$tmp_dir"
+                unzip -q "$zip_file" "emqx/lib/*" -d "$tmp_dir"
                 cp -r -n "$tmp_dir/emqx/releases"/* "$releases_dir" || true
                 cp -r -n "$tmp_dir/emqx/lib"/* "$lib_dir" || true
                 rm -rf "$tmp_dir"
             fi
-            releases+=( "$base_vsn" )
+            releases+=( "$BASE_VSN" )
-        done < <("$FIND" _upgrade_base -maxdepth 1 -name "${name_pattern}.zip" -type f)
+        done
     fi
     if [ ${#releases[@]} -eq 0 ]; then
         log "No upgrade base found, relup ignored"
@@ -188,10 +184,9 @@ make_zip() {
             esac
             ;;
     esac
-    ## one can only upgrade to 4.5.X from 4.4.8 or later, when that's
+    # shellcheck disable=SC2207
-    ## released.  Until that's released, there's no relup.
+    bases=($(relup_db base-vsns "$PKG_VSN"))
-    v44_bases="$(git tag -l "*4.4*" | grep -c -E '4\.4\.([8-9]|1[0-9])' || true)"
+    if [[ "${#bases[@]}" -eq 0 ]]; then
-    if [[ "${v44_bases}" -eq 0 ]]; then
         has_relup='no'
     fi
     if [ "$has_relup" = 'yes' ]; then

data/relup-paths.eterm

@@ -0,0 +1,32 @@
+%% -*- mode: erlang; -*-
+
+{<<"4.4.0">>,#{from_versions => [],otp => <<"24.1.5-3">>}}.
+{<<"4.4.1">>,#{from_versions => [<<"4.4.0">>],otp => <<"24.1.5-3">>}}.
+{<<"4.4.2">>,
+ #{from_versions => [<<"4.4.0">>,<<"4.4.1">>],otp => <<"24.1.5-3">>}}.
+{<<"4.4.3">>,
+ #{from_versions => [<<"4.4.0">>,<<"4.4.1">>,<<"4.4.2">>],
+   otp => <<"24.1.5-3">>}}.
+{<<"4.4.4">>,
+ #{from_versions => [<<"4.4.0">>,<<"4.4.1">>,<<"4.4.2">>,<<"4.4.3">>],
+   otp => <<"24.1.5-3">>}}.
+{<<"4.4.5">>,
+ #{from_versions =>
+       [<<"4.4.0">>,<<"4.4.1">>,<<"4.4.2">>,<<"4.4.3">>,<<"4.4.4">>],
+   otp => <<"24.1.5-3">>}}.
+{<<"4.4.6">>,
+ #{from_versions =>
+       [<<"4.4.0">>,<<"4.4.1">>,<<"4.4.2">>,<<"4.4.3">>,<<"4.4.4">>,
+        <<"4.4.5">>],
+   otp => <<"24.1.5-3">>}}.
+{<<"4.4.7">>,
+ #{from_versions =>
+       [<<"4.4.0">>,<<"4.4.1">>,<<"4.4.2">>,<<"4.4.3">>,<<"4.4.4">>,
+        <<"4.4.5">>,<<"4.4.6">>],
+   otp => <<"24.1.5-3">>}}.
+{<<"4.4.8">>,
+ #{from_versions =>
+       [<<"4.4.0">>,<<"4.4.1">>,<<"4.4.2">>,<<"4.4.3">>,<<"4.4.4">>,
+        <<"4.4.5">>,<<"4.4.6">>,<<"4.4.7">>],
+   otp => <<"24.1.5-3">>}}.
+{<<"4.5.0">>,#{from_versions => [<<"4.4.8">>],otp => <<"24.3.4.2-1">>}}.

deploy/docker/Dockerfile

@@ -16,6 +16,7 @@ RUN apk add --no-cache \
     libc-dev \
     libstdc++ \
     bash \
+    tzdata \
     jq
 
 COPY . /emqx
@@ -37,7 +38,7 @@ COPY deploy/docker/docker-entrypoint.sh /usr/bin/
 COPY --from=builder /emqx/_build/$EMQX_NAME/rel/emqx /opt/emqx
 
 RUN ln -s /opt/emqx/bin/* /usr/local/bin/
-RUN apk add --no-cache curl ncurses-libs openssl sudo libstdc++ bash
+RUN apk add --no-cache curl ncurses-libs openssl sudo libstdc++ bash tzdata
 
 WORKDIR /opt/emqx
 

lib-ce/emqx_dashboard/src/emqx_dashboard.app.src

@@ -1,6 +1,6 @@
 {application, emqx_dashboard,
  [{description, "EMQX Web Dashboard"},
-  {vsn, "4.4.6"}, % strict semver, bump manually!
+  {vsn, "4.4.7"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_dashboard_sup]},
   {applications, [kernel,stdlib,mnesia,minirest]},

rebar.config

@@ -41,7 +41,7 @@
     [ {gpb, "4.11.2"} %% gpb only used to build, but not for release, pin it here to avoid fetching a wrong version due to rebar plugins scattered in all the deps
     , {redbug, "2.0.7"}
     , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.2.0"}}}
-    , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.7.2"}}}
+    , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.7.3"}}}
     , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
     , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
     , {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.0"}}}
@@ -62,7 +62,7 @@
     , {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.1"}}}
     , {mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.13"}}}
     , {epgsql, {git, "https://github.com/emqx/epgsql.git", {tag, "4.6.0"}}}
-    , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.6"}}}
+    , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.7"}}}
     ]}.
 
 {xref_ignores,

rebar.config.erl

@@ -51,6 +51,8 @@ overrides() ->
     [ {add, [ {extra_src_dirs, [{"etc", [{recursive,true}]}]}
             , {erl_opts, [{compile_info, [{emqx_vsn, get_vsn()}]}]}
             ]}
+
+    , {add, relx, [{erl_opts, [{d, 'RLX_LOG', rlx_log}]}]}
     , {add, snabbkaffe,
        [{erl_opts, common_compile_opts()}]}
     ] ++ community_plugin_overrides().
@@ -88,7 +90,7 @@ project_app_dirs() ->
     ["apps/*", alternative_lib_dir() ++ "/*", "."].
 
 plugins(HasElixir) ->
-    [ {relup_helper,{git,"https://github.com/emqx/relup_helper", {tag, "2.0.0"}}}
+    [ {relup_helper,{git,"https://github.com/emqx/relup_helper", {tag, "2.1.0"}}}
     , {er_coap_client, {git, "https://github.com/emqx/er_coap_client", {tag, "v1.0.4"}}}
       %% emqx main project does not require port-compiler
       %% pin at root level for deterministic

scripts/apps-version-check.sh

@@ -29,6 +29,14 @@ check_apps() {
         app_path="."
     fi
     src_file="$app_path/src/$(basename "$app").app.src"
+
+    old_app_exists=0
+    git show "$latest_release":"$src_file" >/dev/null 2>&1 || old_app_exists="$?"
+    if [ "$old_app_exists" != "0" ]; then
+        echo "$app is new, skipping version check"
+        continue
+    fi
+
     old_app_version="$(git show "$latest_release":"$src_file" | grep vsn | grep -oE '"[0-9]+\.[0-9]+\.[0-9]+"' | tr -d '"')"
     now_app_version=$(grep -E 'vsn' "$src_file" | grep -oE '"[0-9]+\.[0-9]+\.[0-9]+"' | tr -d '"')
     if [ "$old_app_version" = "$now_app_version" ]; then
@@ -58,7 +66,7 @@ check_apps() {
         now_app_version_semver=($(parse_semver "$now_app_version"))
         if  [ "${old_app_version_semver[0]}" = "${now_app_version_semver[0]}" ] && \
             [ "${old_app_version_semver[1]}" = "${now_app_version_semver[1]}" ] && \
-            [ "$(( "${old_app_version_semver[2]}" + 1 ))" = "${now_app_version_semver[2]}" ]; then
+            [ "$(( old_app_version_semver[2] + 1 ))" = "${now_app_version_semver[2]}" ]; then
             true
         else
             echo "$src_file: non-strict semver version bump from $old_app_version to $now_app_version"

scripts/get-dashboard.sh

@@ -8,13 +8,13 @@ cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."
 PKG_VSN="${PKG_VSN:-$(./pkg-vsn.sh)}"
 case "${PKG_VSN}" in
     4.3*)
-        EMQX_CE_DASHBOARD_VERSION='v4.3.9'
+        EMQX_CE_DASHBOARD_VERSION='v4.3.10'
-        EMQX_EE_DASHBOARD_VERSION='v4.3.23'
+        EMQX_EE_DASHBOARD_VERSION='v4.3.24'
         ;;
     4.4*)
         # keep the above 4.3 untouched, otherwise conflicts!
-        EMQX_CE_DASHBOARD_VERSION='v4.4.4'
+        EMQX_CE_DASHBOARD_VERSION='v4.4.5'
-        EMQX_EE_DASHBOARD_VERSION='v4.4.14'
+        EMQX_EE_DASHBOARD_VERSION='v4.4.15'
         ;;
     4.5*)
         # keep the above 4.3 untouched, otherwise conflicts!

scripts/relup-base-packages.sh

@@ -63,17 +63,8 @@ cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."
 mkdir -p _upgrade_base
 pushd _upgrade_base
 
-# For 4.5+, we upgrade from OTP 24.1.5-3 to 24.3.4.2-1, so we must manually
-# check the old OTP releases.
 otp_vsn_for() {
-  case "${1#[e|v]}" in
+    ../scripts/relup-base-vsns.escript otp-vsn-for "${1#[e|v]}" ../data/relup-paths.eterm
-    4.4.*)
-      echo "24.1.5-3"
-      ;;
-    4.5.*)
-      echo "$OTP_VSN"
-      ;;
-  esac
 }
 
 for tag in $(../scripts/relup-base-vsns.sh $EDITION | xargs echo -n); do
@@ -90,6 +81,8 @@ for tag in $(../scripts/relup-base-vsns.sh $EDITION | xargs echo -n); do
             ## https://askubuntu.com/questions/1202208/checking-sha256-checksum
             echo "${SUMSTR}  ${filename}" | $SHASUM -c || exit 1
         fi
+    else
+        echo "file $filename already downloaded or doesn't exist in the archives; skipping it"
     fi
 done
 

scripts/relup-base-vsns.escript

@@ -0,0 +1,285 @@
+#!/usr/bin/env escript
+%% -*- mode: erlang; -*-
+
+-mode(compile).
+
+-define(RED, "\e[31m").
+-define(RESET, "\e[39m").
+
+usage() ->
+"A script to manage the released versions of EMQX for relup and hot
+upgrade/downgrade.
+
+We store a \"database\" of released versions as an `eterm' file, which
+is a mapping from a given version `Vsn' to its OTP version and a list
+of previous versions from which one can upgrade to `Vsn' (the
+\"from_versions\" list).  That allow us to more easily/explicitly keep
+track of allowed version upgrades/downgrades, as well as OTP changes
+between releases
+
+In the examples below, `VERSION_DB_PATH' represents the path to the
+`eterm' file containing the version database to be used.
+
+Usage:
+
+  * List the previous base versions from which `TO_VSN' may be
+    upgraded to.  Used to list versions for which relup files are to
+    be made.
+
+    relup-base-vsns.escript base-vsns TO_VSN VERSION_DB_PATH
+
+
+  * Show the OTP version with which `Vsn' was built.
+
+    relup-base-vsns.escript otp-vsn-for VSN VERSION_DB_PATH
+
+
+  * Automatically inserts a new version into the database.  Previous
+    versions with the same Major and Minor numbers as `Vsn' are
+    considered to be upgradeable from, and versions with higher Major
+    and Minor numbers will automatically include `Vsn' in their
+    \"from_versions\" list.
+
+    For example, if inserting 4.4.8 when 4.5.0 and 4.5.1 exists,
+    versions `BASE_FROM_VSN'...4.4.7 will be considered 4.4.8's
+    \"from_versions\", and 4.4.8 will be included into 4.5.0 and
+    4.5.1's from versions.
+
+    relup-base-vsns.escript insert-new-vsn NEW_VSN BASE_FROM_VSN OTP_VSN VERSION_DB_PATH
+
+  * Check if the version database is consistent considering `VSN'.
+
+    relup-base-vsns.escript check-vsn-db VSN VERSION_DB_PATH
+".
+
+main(["base-vsns", To0, VsnDB]) ->
+    VsnMap = read_db(VsnDB),
+    To = strip_pre_release(To0),
+    #{from_versions := Froms} = fetch_version(To, VsnMap),
+    AvailableVersionsIndex = available_versions_index(),
+    lists:foreach(
+     fun(From) ->
+       io:format(user, "~s~n", [From])
+     end,
+     filter_froms(Froms, AvailableVersionsIndex)),
+    halt(0);
+main(["otp-vsn-for", Vsn0, VsnDB]) ->
+    VsnMap = read_db(VsnDB),
+    Vsn = strip_pre_release(Vsn0),
+    #{otp := OtpVsn} = fetch_version(Vsn, VsnMap),
+    io:format(user, "~s~n", [OtpVsn]),
+    halt(0);
+main(["insert-new-vsn", NewVsn0, BaseFromVsn0, OtpVsn0, VsnDB]) ->
+    VsnMap = read_db(VsnDB),
+    NewVsn = strip_pre_release(NewVsn0),
+    validate_version(NewVsn),
+    BaseFromVsn = strip_pre_release(BaseFromVsn0),
+    validate_version(BaseFromVsn),
+    OtpVsn = list_to_binary(OtpVsn0),
+    case VsnMap of
+      #{NewVsn := _} ->
+          print_warning("Version ~s already in DB!~n", [NewVsn]),
+          halt(1);
+      #{BaseFromVsn := _} ->
+          ok;
+      _ ->
+          print_warning("Version ~s not found in DB!~n", [BaseFromVsn]),
+          halt(1)
+    end,
+    NewVsnMap = insert_new_vsn(VsnMap, NewVsn, OtpVsn, BaseFromVsn),
+    NewVsnList =
+        lists:sort(
+         fun({Vsn1, _}, {Vsn2, _}) ->
+           parse_vsn(Vsn1) < parse_vsn(Vsn2)
+         end, maps:to_list(NewVsnMap)),
+    {ok, FD} = file:open(VsnDB, [write]),
+    io:format(FD, "%% -*- mode: erlang; -*-\n\n", []),
+    lists:foreach(
+      fun(Entry) ->
+        io:format(FD, "~p.~n", [Entry])
+      end,
+      NewVsnList),
+    file:close(FD),
+    halt(0);
+main(["check-vsn-db", NewVsn0, VsnDB]) ->
+    VsnMap = read_db(VsnDB),
+    NewVsn = strip_pre_release(NewVsn0),
+    case check_all_vsns_schema(VsnMap) of
+        [] -> ok;
+        Problems ->
+            print_warning("Invalid Version DB ~s!~n", [VsnDB]),
+            print_warning("Problems found:~n"),
+            lists:foreach(
+              fun(Problem) ->
+                print_warning("  ~p~n", [Problem])
+              end, Problems),
+            halt(1)
+    end,
+    case VsnMap of
+        #{NewVsn := _} ->
+            io:format(user, "ok~n", []),
+            halt(0);
+        _ ->
+            Candidates = find_insertion_candidates(NewVsn, VsnMap),
+            print_warning("Version ~s not found in the version DB!~n", [NewVsn]),
+            [] =/= Candidates
+                andalso print_warning("Candidates for to insert this version into:~n"),
+            lists:foreach(
+              fun(Vsn) ->
+                io:format(user, "  ~s~n", [Vsn])
+              end, Candidates),
+            print_warning(
+              "To insert this version automatically, run:~n"
+              "./scripts/relup-base-vsns insert-new-vsn NEW-VSN BASE-FROM-VSN NEW-OTP-VSN ~s~n"
+              "And commit the results.  Be sure to revise the changes.~n"
+              "Otherwise, edit the file manually~n",
+              [VsnDB]),
+            halt(1)
+    end;
+main(_) ->
+    io:format(user, usage(), []),
+    halt(1).
+
+strip_pre_release(Vsn0) ->
+    case re:run(Vsn0, "[0-9]+\\.[0-9]+\\.[0-9]+", [{capture, all, binary}]) of
+        {match, [Vsn]} ->
+            Vsn;
+        _ ->
+            print_warning("Invalid Version: ~s ~n", [Vsn0]),
+            halt(1)
+    end.
+
+fetch_version(Vsn, VsnMap) ->
+    case VsnMap of
+        #{Vsn := VsnData} ->
+            VsnData;
+        _ ->
+            print_warning("Version not found in releases: ~s ~n", [Vsn]),
+            halt(1)
+    end.
+
+filter_froms(Froms0, AvailableVersionsIndex) ->
+    Froms1 =
+        case os:getenv("SYSTEM") of
+            %% debian11 is introduced since v4.4.2 and e4.4.2
+            %% exclude tags before them
+            "debian11" ->
+                lists:filter(
+                  fun(Vsn) ->
+                          not lists:member(Vsn, [<<"4.4.0">>, <<"4.4.1">>])
+                  end, Froms0);
+            _ ->
+                Froms0
+        end,
+    lists:filter(
+      fun(V) -> maps:get(V, AvailableVersionsIndex, false) end,
+      Froms1).
+
+%% assumes that's X.Y.Z, without pre-releases
+parse_vsn(VsnBin) ->
+    {match, [Major0, Minor0, Patch0]} = re:run(VsnBin, "([0-9]+)\\.([0-9]+)\\.([0-9]+)",
+                                               [{capture, all_but_first, binary}]),
+    [Major, Minor, Patch] = lists:map(fun binary_to_integer/1, [Major0, Minor0, Patch0]),
+    {Major, Minor, Patch}.
+
+parsed_vsn_to_bin({Major, Minor, Patch}) ->
+    iolist_to_binary(io_lib:format("~b.~b.~b", [Major, Minor, Patch])).
+
+find_insertion_candidates(NewVsn, VsnMap) ->
+    ParsedNewVsn = parse_vsn(NewVsn),
+    [Vsn
+     || Vsn <- maps:keys(VsnMap),
+        ParsedVsn <- [parse_vsn(Vsn)],
+        ParsedVsn > ParsedNewVsn].
+
+check_all_vsns_schema(VsnMap) ->
+    maps:fold(
+      fun(Vsn, Val, Acc) ->
+        Problems =
+          [{Vsn, should_be_binary} || not is_binary(Vsn)] ++
+          [{Vsn, must_have_map_value} || not is_map(Val)] ++
+          [{Vsn, {must_contain_keys, [otp, from_versions]}}
+           || case Val of
+                  #{otp := _, from_versions := _} ->
+                      false;
+                  _ ->
+                      true
+              end] ++
+          [{Vsn, otp_version_must_be_binary}
+           || case Val of
+                  #{otp := Otp} when is_binary(Otp) ->
+                      false;
+                  _ ->
+                      true
+              end] ++
+          [{Vsn, versions_must_be_list_of_binaries}
+           || case Val of
+                  #{from_versions := Froms} when is_list(Froms) ->
+                      not lists:all(fun is_binary/1, Froms);
+                  _ ->
+                      true
+              end],
+        Problems ++ Acc
+      end,
+      [],
+      VsnMap).
+
+insert_new_vsn(VsnMap0, NewVsn, OtpVsn, BaseFromVsn) ->
+    ParsedNewVsn = parse_vsn(NewVsn),
+    ParsedBaseFromVsn = parse_vsn(BaseFromVsn),
+    %% candidates to insert this version into (they are "future" versions)
+    Candidates = find_insertion_candidates(NewVsn, VsnMap0),
+    %% Past versions we can upgrade from
+    Froms = [Vsn || Vsn <- maps:keys(VsnMap0),
+                    ParsedVsn <- [parse_vsn(Vsn)],
+                    ParsedVsn >= ParsedBaseFromVsn,
+                    ParsedVsn < ParsedNewVsn],
+    VsnMap1 =
+        lists:foldl(
+          fun(FutureVsn, Acc) ->
+            FutureData0 = #{from_versions := Froms0} = maps:get(FutureVsn, Acc),
+            FutureData = FutureData0#{from_versions => lists:usort(Froms0 ++ [NewVsn])},
+            Acc#{FutureVsn => FutureData}
+          end,
+          VsnMap0,
+          Candidates),
+    VsnMap1#{NewVsn => #{otp => OtpVsn, from_versions => Froms}}.
+
+validate_version(Vsn) ->
+    ParsedVsn = parse_vsn(Vsn),
+    VsnBack = parsed_vsn_to_bin(ParsedVsn),
+    case VsnBack =:= Vsn of
+        true -> ok;
+        false ->
+            print_warning("Invalid version ~p !~n", [Vsn]),
+            print_warning("Versions MUST be of the form X.Y.Z "
+                          "and not prefixed by `e` or `v`~n"),
+            halt(1)
+    end.
+
+available_versions_index() ->
+    Output = os:cmd("git tag -l"),
+    AllVersions =
+        lists:filtermap(
+          fun(Line) ->
+            case re:run(Line, "^[ve]([0-9]+)\\.([0-9]+)\\.([0-9]+)$",
+                        [{capture, all_but_first, binary}]) of
+                {match, [Major, Minor, Patch]} ->
+                    Vsn = iolist_to_binary(io_lib:format("~s.~s.~s", [Major, Minor, Patch])),
+                    {true, Vsn};
+                _ -> false
+            end
+          end, string:split(Output, "\n", all)),
+    %% FIXME: `maps:from_keys' is available only in OTP 24, but we
+    %% still build with 23.  Switch to that once we drop OTP 23.
+    maps:from_list([{Vsn, true} || Vsn <- AllVersions]).
+
+read_db(VsnDB) ->
+    {ok, VsnList} = file:consult(VsnDB),
+    maps:from_list(VsnList).
+
+print_warning(Msg) ->
+    print_warning(Msg, []).
+
+print_warning(Msg, Args) ->
+    io:format(user, ?RED ++ Msg ++ ?RESET, Args).

scripts/relup-base-vsns.sh

@@ -41,14 +41,6 @@ if [ "${#CUR_SEMVER[@]}" -lt 3 ]; then
     usage
 fi
 
-## when the current version has no suffix such as -abcdef00
-## it is a formal release
-if [ "${#CUR_SEMVER[@]}" -eq 3 ]; then
-    IS_RELEASE=true
-else
-    IS_RELEASE=false
-fi
-
 case "${EDITION}" in
     *enterprise*)
         GIT_TAG_PREFIX="e"
@@ -62,28 +54,10 @@ esac
 TAGS=( 'dummy' )
 TAGS_EXCLUDE=( 'dummy' )
 
-while read -r git_tag; do
+while read -r vsn; do
     # shellcheck disable=SC2207
-    semver=($(parse_semver "$git_tag"))
+    TAGS+=($(git tag -l "${GIT_TAG_PREFIX}${vsn}"))
-    if [ "${#semver[@]}" -eq 3 ] && [ "${semver[2]}" -le "${CUR_SEMVER[2]}" ]; then
+done < <(./scripts/relup-base-vsns.escript base-vsns "$CUR" ./data/relup-paths.eterm)
-        if [ ${IS_RELEASE} = true ] && [ "${semver[2]}" -eq "${CUR_SEMVER[2]}" ] ; then
-            # do nothing
-            # exact match, do not print current version
-            # because current version is not an upgrade base
-            true
-        else
-            TAGS+=( "$git_tag" )
-        fi
-    fi
-done < <(git tag -l "${GIT_TAG_PREFIX}${CUR_SEMVER[0]}.${CUR_SEMVER[1]}.*")
-
-# debian11 is introduced since v4.4.2 and e4.4.2
-# exclude tags before them
-SYSTEM="${SYSTEM:-$(./scripts/get-distro.sh)}"
-if [ "$SYSTEM" = 'debian11' ]; then
-    TAGS_EXCLUDE+=( 'v4.4.0' 'v4.4.1' )
-    TAGS_EXCLUDE+=( 'e4.4.0' 'e4.4.1' )
-fi
 
 for tag_to_del in "${TAGS_EXCLUDE[@]}"; do
     TAGS=( "${TAGS[@]/$tag_to_del}" )

src/emqx.appup.src

@@ -2,21 +2,39 @@
 %% Unless you know what you are doing, DO NOT edit manually!!
 {VSN,
   [{"4.4.7",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_relup,brutal_purge,soft_purge,[]}]},
-   {"4.4.6",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_relup,brutal_purge,soft_purge,[]}]},
-   {"4.4.5",
     [{load_module,emqx_relup,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
+   {"4.4.6",
+    [{load_module,emqx_relup,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]}]},
+   {"4.4.5",
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_relup,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]}]},
    {"4.4.4",
-    [{load_module,emqx_access_control,brutal_purge,soft_purge,[]},
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx_broker,brutal_purge,soft_purge,[]},
@@ -32,7 +50,8 @@
      {load_module,emqx_metrics,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]}]},
    {"4.4.3",
-    [{add_module,emqx_calendar},
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {add_module,emqx_calendar},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_broker,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
@@ -47,6 +66,7 @@
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_frame,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
@@ -79,6 +99,7 @@
      {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
      {load_module,emqx_frame,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
@@ -139,7 +160,6 @@
      {update,emqx_os_mon,{advanced,[]}},
      {load_module,emqx_hooks,brutal_purge,soft_purge,[]},
      {load_module,emqx_listeners,brutal_purge,soft_purge,[]},
-     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_flapping,brutal_purge,soft_purge,[]},
      {load_module,emqx_pmon,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
@@ -152,6 +172,7 @@
      {add_module,emqx_relup},
      {load_module,emqx_vm_mon,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_connection,brutal_purge,soft_purge,[]},
      {load_module,emqx_ws_connection,brutal_purge,soft_purge,[]},
@@ -166,24 +187,43 @@
      {load_module,emqx_limiter,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
   [{"4.4.7",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_relup,brutal_purge,soft_purge,[]},
-     {load_module,emqx_relup,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]}]},
    {"4.4.6",
-    [{load_module,emqx_app,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_relup,brutal_purge,soft_purge,[]},
-     {load_module,emqx_relup,brutal_purge,soft_purge,[]}]},
-   {"4.4.5",
-    [{update,emqx_broker_sup,supervisor},
-     {load_module,emqx_relup,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx,brutal_purge,soft_purge,[]}]},
+   {"4.4.5",
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
+     {update,emqx_broker_sup,supervisor},
+     {load_module,emqx_app,brutal_purge,soft_purge,[]},
+     {load_module,emqx_session,brutal_purge,soft_purge,[]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
+     {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
+     {load_module,emqx_channel,brutal_purge,soft_purge,[]},
+     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
+     {load_module,emqx_relup,brutal_purge,soft_purge,[]}]},
+   {"4.4.4",
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_control,brutal_purge,soft_purge,[]},
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
-     {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]}]},
+     {load_module,emqx_misc,brutal_purge,soft_purge,[]},
-   {"4.4.4",
-    [{load_module,emqx_access_control,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx_broker,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
      {load_module,emqx_mqtt_caps,brutal_purge,soft_purge,[]},
@@ -193,10 +233,10 @@
      {load_module,emqx_relup,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
-     {load_module,emqx_session,brutal_purge,soft_purge,[]},
      {load_module,emqx_metrics,brutal_purge,soft_purge,[]}]},
    {"4.4.3",
-    [{load_module,emqx_broker,brutal_purge,soft_purge,[]},
+    [{load_module,emqx,brutal_purge,soft_purge,[]},
+     {load_module,emqx_broker,brutal_purge,soft_purge,[]},
      {update,emqx_broker_sup,supervisor},
      {load_module,emqx_ctl,brutal_purge,soft_purge,[]},
      {load_module,emqx_mqtt_caps,brutal_purge,soft_purge,[]},
@@ -211,6 +251,7 @@
      {load_module,emqx_misc,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
      {load_module,emqx_frame,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_access_rule,brutal_purge,soft_purge,[]},
      {load_module,emqx_alarm_handler,brutal_purge,soft_purge,[]},
@@ -241,6 +282,7 @@
      {update,emqx_os_mon,{advanced,[]}},
      {load_module,emqx_plugins,brutal_purge,soft_purge,[]},
      {load_module,emqx_frame,brutal_purge,soft_purge,[]},
+     {load_module,emqx_cm,brutal_purge,soft_purge,[]},
      {load_module,emqx_channel,brutal_purge,soft_purge,[]},
      {load_module,emqx_sys,brutal_purge,soft_purge,[]},
      {load_module,emqx_shared_sub,brutal_purge,soft_purge,[]},
@@ -321,7 +363,6 @@
      {load_module,emqx_session,brutal_purge,soft_purge,[]},
      {load_module,emqx,brutal_purge,soft_purge,[]},
      {load_module,emqx_app,brutal_purge,soft_purge,[]},
-     {load_module,emqx_message,brutal_purge,soft_purge,[]},
      {load_module,emqx_limiter,brutal_purge,soft_purge,[]},
      {delete_module,emqx_relup}]},
    {<<".*">>,[]}]}.

src/emqx_channel.erl

@@ -22,6 +22,8 @@
 -include("logger.hrl").
 -include("types.hrl").
 
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+
 -logger_header("[Channel]").
 
 -ifdef(TEST).
@@ -316,7 +318,7 @@ handle_in(?CONNECT_PACKET(ConnPkt) = Packet, Channel) ->
                                         },
             case enhanced_auth(?CONNECT_PACKET(NConnPkt), NChannel1) of
                 {ok, Properties, NChannel2} ->
-                    process_connect(Properties, ensure_connected(NChannel2));
+                    process_connect(Properties, NChannel2);
                 {continue, Properties, NChannel2} ->
                     handle_out(auth, {?RC_CONTINUE_AUTHENTICATION, Properties}, NChannel2);
                 {error, ReasonCode, NChannel2} ->
@@ -332,7 +334,7 @@ handle_in(Packet = ?AUTH_PACKET(?RC_CONTINUE_AUTHENTICATION, _Properties),
         {ok, NProperties, NChannel} ->
             case ConnState of
                 connecting ->
-                    process_connect(NProperties, ensure_connected(NChannel));
+                    process_connect(NProperties, NChannel);
                 connected ->
                     handle_out(auth, {?RC_SUCCESS, NProperties}, NChannel);
                 _ ->
@@ -522,14 +524,14 @@ process_connect(AckProps, Channel = #channel{conninfo = ConnInfo,
     case emqx_cm:open_session(CleanStart, ClientInfo, ConnInfo) of
         {ok, #{session := Session, present := false}} ->
             NChannel = Channel#channel{session = Session},
-            handle_out(connack, {?RC_SUCCESS, sp(false), AckProps}, NChannel);
+            handle_out(connack, {?RC_SUCCESS, sp(false), AckProps}, ensure_connected(NChannel));
         {ok, #{session := Session, present := true, pendings := Pendings}} ->
             Pendings1 = lists:usort(lists:append(Pendings, emqx_misc:drain_deliver())),
             NChannel = Channel#channel{session  = Session,
                                        resuming = true,
                                        pendings = Pendings1
                                       },
-            handle_out(connack, {?RC_SUCCESS, sp(true), AckProps}, NChannel);
+            handle_out(connack, {?RC_SUCCESS, sp(true), AckProps}, ensure_connected(NChannel));
         {error, client_id_unavailable} ->
             handle_out(connack, ?RC_CLIENT_IDENTIFIER_NOT_VALID, Channel);
         {error, Reason} ->
@@ -875,11 +877,14 @@ handle_out(disconnect, ReasonCode, Channel) when is_integer(ReasonCode) ->
     ReasonName = disconnect_reason(ReasonCode),
     handle_out(disconnect, {ReasonCode, ReasonName}, Channel);
 
-handle_out(disconnect, {ReasonCode, ReasonName}, Channel = ?IS_MQTT_V5) ->
+handle_out(disconnect, {ReasonCode, ReasonName}, Channel) ->
-    Packet = ?DISCONNECT_PACKET(ReasonCode),
+    handle_out(disconnect, {ReasonCode, ReasonName, #{}}, Channel);
+
+handle_out(disconnect, {ReasonCode, ReasonName, Props}, Channel = ?IS_MQTT_V5) ->
+    Packet = ?DISCONNECT_PACKET(ReasonCode, Props),
     {ok, [{outgoing, Packet}, {close, ReasonName}], Channel};
 
-handle_out(disconnect, {_ReasonCode, ReasonName}, Channel) ->
+handle_out(disconnect, {_ReasonCode, ReasonName, _Props}, Channel) ->
     {ok, {close, ReasonName}, Channel};
 
 handle_out(auth, {ReasonCode, Properties}, Channel) ->
@@ -979,11 +984,15 @@ handle_call({takeover, 'begin'}, Channel = #channel{session = Session}) ->
     reply(Session, Channel#channel{takeover = true});
 
 handle_call({takeover, 'end'}, Channel = #channel{session  = Session,
-                                                  pendings = Pendings}) ->
+                                                  pendings = Pendings,
+                                                  conninfo = #{clientid := ClientId}}) ->
     ok = emqx_session:takeover(Session),
     %% TODO: Should not drain deliver here (side effect)
     Delivers = emqx_misc:drain_deliver(),
     AllPendings = lists:append(Delivers, Pendings),
+    ?tp(debug,
+        emqx_channel_takeover_end,
+        #{clientid => ClientId}),
     disconnect_and_shutdown(takeovered, AllPendings, Channel);
 
 handle_call(list_acl_cache, Channel) ->
@@ -1057,6 +1066,9 @@ handle_info(clean_acl_cache, Channel) ->
     ok = emqx_acl_cache:empty_acl_cache(),
     {ok, Channel};
 
+handle_info({disconnect, ReasonCode, ReasonName, Props}, Channel) ->
+    handle_out(disconnect, {ReasonCode, ReasonName, Props}, Channel);
+
 handle_info(Info, Channel) ->
     ?LOG(error, "Unexpected info: ~p", [Info]),
     {ok, Channel}.

src/emqx_cm.erl

@@ -22,6 +22,8 @@
 -include("emqx.hrl").
 -include("logger.hrl").
 -include("types.hrl").
+
+-include_lib("stdlib/include/qlc.hrl").
 -include_lib("stdlib/include/ms_transform.hrl").
 -include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
@@ -61,7 +63,9 @@
         , lookup_channels/2
         ]).
 
--export([all_channels/0]).
+-export([all_channels/0,
+         channel_with_session_table/0,
+         live_connection_table/0]).
 
 %% gen_server callbacks
 -export([ init/1
@@ -158,8 +162,11 @@ connection_closed(ClientId) ->
     connection_closed(ClientId, self()).
 
 -spec(connection_closed(emqx_types:clientid(), chan_pid()) -> true).
-connection_closed(ClientId, ChanPid) ->
+connection_closed(_ClientId, _ChanPid) ->
-    ets:delete_object(?CHAN_CONN_TAB, {ClientId, ChanPid}).
+    %% We can't clean CHAN_CONN_TAB because records for dead connections
+    %% are required for `get_chann_conn_mod/1` function, and `get_chann_conn_mod/1`
+    %% is used for takeover.
+    true.
 
 %% @doc Get info of a channel.
 -spec(get_chan_info(emqx_types:clientid()) -> maybe(emqx_types:infos())).
@@ -435,6 +442,38 @@ all_channels() ->
     Pat = [{{'_', '$1'}, [], ['$1']}],
     ets:select(?CHAN_TAB, Pat).
 
+%% @doc Get clientinfo for all clients with sessions
+channel_with_session_table() ->
+    Ms = ets:fun2ms(
+           fun({{ClientId, _ChanPid},
+                Info,
+                _Stats}) ->
+                   {ClientId, Info}
+           end),
+    Table = ets:table(?CHAN_INFO_TAB, [{traverse, {select, Ms}}]),
+    qlc:q([ {ClientId, ConnState, ConnInfo, ClientInfo}
+            || {ClientId,
+                #{conn_state := ConnState,
+                  clientinfo := ClientInfo,
+                  conninfo := #{clean_start := false} = ConnInfo}} <- Table
+          ]).
+
+%% @doc Get all local connection query handle
+live_connection_table() ->
+    Ms = ets:fun2ms(
+           fun({{ClientId, ChanPid}, _}) ->
+                   {ClientId, ChanPid}
+           end),
+    Table = ets:table(?CHAN_CONN_TAB, [{traverse, {select, Ms}}]),
+    qlc:q([{ClientId, ChanPid} || {ClientId, ChanPid} <- Table, is_channel_connected(ClientId, ChanPid)]).
+
+is_channel_connected(ClientId, ChanPid) when node(ChanPid) =:= node() ->
+    case get_chan_info(ClientId, ChanPid) of
+        #{conn_state := disconnected} -> false;
+        _ -> true
+    end;
+is_channel_connected(_ClientId, _ChanPid) -> false.
+
 %% @doc Lookup channels.
 -spec(lookup_channels(emqx_types:clientid()) -> list(chan_pid())).
 lookup_channels(ClientId) ->

src/emqx_plugins.erl

@@ -217,20 +217,47 @@ load_plugin_conf(AppName, PluginDir) ->
 ensure_file(File) ->
     case filelib:is_file(File) of
         false ->
-            DefaultPlugins = [ {emqx_management, true}
+            DefaultPlugins = default_plugins(),
-                             , {emqx_dashboard, true}
-                             , {emqx_modules, false}
-                             , {emqx_recon, true}
-                             , {emqx_retainer, true}
-                             , {emqx_telemetry, true}
-                             , {emqx_rule_engine, true}
-                             , {emqx_bridge_mqtt, false}
-                             ],
             write_loaded(DefaultPlugins);
         true ->
             ok
     end.
 
+-ifndef(EMQX_ENTERPRISE).
+%% default plugins see rebar.config.erl
+default_plugins() ->
+    [
+        {emqx_management, true},
+        {emqx_dashboard, true},
+        %% emqx_modules is not a plugin, but a normal application starting when boots.
+        {emqx_modules, false},
+        {emqx_retainer, true},
+        {emqx_recon, true},
+        {emqx_telemetry, true},
+        {emqx_rule_engine, true},
+        {emqx_bridge_mqtt, false}
+    ].
+
+-else.
+
+default_plugins() ->
+    [
+        {emqx_management, true},
+        {emqx_dashboard, true},
+        %% enterprise version of emqx_modules is a plugin
+        {emqx_modules, true},
+        %% retainer is managed by emqx_modules.
+        %% default is true in data/load_modules. **NOT HERE**
+        {emqx_retainer, false},
+        {emqx_recon, true},
+        %% emqx_telemetry is not exist in enterprise.
+        %% {emqx_telemetry, false},
+        {emqx_rule_engine, true},
+        {emqx_bridge_mqtt, false}
+    ].
+
+-endif.
+
 with_loaded_file(File, SuccFun) ->
     case read_loaded(File) of
         {ok, Names0} ->
@@ -265,7 +292,7 @@ load_plugins(Names, Persistent) ->
         []       -> ok;
         NotFound -> ?LOG(alert, "cannot_find_plugins: ~p", [NotFound])
     end,
-    NeedToLoad = Names -- NotFound -- names(started_app),
+    NeedToLoad = (Names -- NotFound) -- names(started_app),
     lists:foreach(fun(Name) ->
                       Plugin = find_plugin(Name, Plugins),
                       load_plugin(Plugin#plugin.name, Persistent)

test/emqx_broker_sup_SUITE.erl

@@ -61,5 +61,5 @@ t_restart_shared_sub(Config) when is_list(Config) ->
        after 2000 ->
                  false
        end);
-t_restart_shared_sub({'end', Config}) ->
+t_restart_shared_sub({'end', _Config}) ->
     emqx:unsubscribe(<<"$share/grpa/t/a">>).

test/emqx_node_helpers.erl

@@ -0,0 +1,83 @@
+%%--------------------------------------------------------------------
+%% 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.
+%%--------------------------------------------------------------------
+
+-module(emqx_node_helpers).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(SLAVE_START_APPS, [emqx]).
+
+-export([start_slave/1,
+         start_slave/2,
+         stop_slave/1]).
+
+start_slave(Name) ->
+    start_slave(Name, #{}).
+
+start_slave(Name, Opts) ->
+    {ok, Node} = ct_slave:start(list_to_atom(atom_to_list(Name) ++ "@" ++ host()),
+                                [{kill_if_fail, true},
+                                 {monitor_master, true},
+                                 {init_timeout, 10000},
+                                 {startup_timeout, 10000},
+                                 {erl_flags, ebin_path()}]),
+
+    pong = net_adm:ping(Node),
+    setup_node(Node, Opts),
+    Node.
+
+stop_slave(Node) ->
+    rpc:call(Node, ekka, leave, []),
+    ct_slave:stop(Node).
+
+host() ->
+    [_, Host] = string:tokens(atom_to_list(node()), "@"), Host.
+
+ebin_path() ->
+    string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " ").
+
+is_lib(Path) ->
+    string:prefix(Path, code:lib_dir()) =:= nomatch.
+
+setup_node(Node, #{} = Opts) ->
+    Listeners = maps:get(listeners, Opts, []),
+    StartApps = maps:get(start_apps, Opts, ?SLAVE_START_APPS),
+    DefaultEnvHandler =
+        fun(emqx) ->
+                application:set_env(
+                  emqx,
+                  listeners,
+                  Listeners),
+                application:set_env(gen_rpc, port_discovery, stateless),
+                ok;
+           (_) ->
+                ok
+        end,
+    EnvHandler = maps:get(env_handler, Opts, DefaultEnvHandler),
+
+    [ok = rpc:call(Node, application, load, [App]) || App <- [gen_rpc, emqx]],
+    ok = rpc:call(Node, emqx_ct_helpers, start_apps, [StartApps, EnvHandler]),
+
+    rpc:call(Node, ekka, join, [node()]),
+
+    %% Sanity check. Assert that `gen_rpc' is set up correctly:
+    ?assertEqual( Node
+                , gen_rpc:call(Node, erlang, node, [])
+                ),
+    ?assertEqual( node()
+                , gen_rpc:call(Node, gen_rpc, call, [node(), erlang, node, []])
+                ),
+    ok.

test/emqx_plugins_SUITE.erl

@@ -89,6 +89,34 @@ t_load(_) ->
     ?assertEqual(ignore, emqx_plugins:load()),
     ?assertEqual(ignore, emqx_plugins:unload()).
 
+-ifndef(EMQX_ENTERPRISE).
+default_plugins() ->
+    [
+        {emqx_bridge_mqtt, false},
+        {emqx_dashboard, true},
+        {emqx_management, true},
+        {emqx_modules, false},
+        {emqx_recon, true},
+        {emqx_retainer, true},
+        {emqx_rule_engine, true},
+        {emqx_telemetry, true}
+    ].
+
+-else.
+
+default_plugins() ->
+    [
+        {emqx_bridge_mqtt, false},
+        {emqx_dashboard, true},
+        {emqx_management, true},
+        {emqx_modules, true},
+        {emqx_recon, true},
+        {emqx_retainer, false},
+        {emqx_rule_engine, true},
+        {emqx_telemetry, true}
+    ].
+
+-endif.
 t_ensure_default_loaded_plugins_file(Config) ->
     %% this will trigger it to write the default plugins to the
     %% inexistent file; but it won't truly load them in this test
@@ -96,17 +124,8 @@ t_ensure_default_loaded_plugins_file(Config) ->
     TmpFilepath = ?config(tmp_filepath, Config),
     ok = emqx_plugins:load(),
     {ok, Contents} = file:consult(TmpFilepath),
-    ?assertEqual(
+    DefaultPlugins = default_plugins(),
-       [ {emqx_bridge_mqtt, false}
+    ?assertEqual(DefaultPlugins, lists:sort(Contents)),
-       , {emqx_dashboard, true}
-       , {emqx_management, true}
-       , {emqx_modules, false}
-       , {emqx_recon, true}
-       , {emqx_retainer, true}
-       , {emqx_rule_engine, true}
-       , {emqx_telemetry, true}
-       ],
-       lists:sort(Contents)),
     ok.
 
 t_init_config(_) ->

test/emqx_shared_sub_SUITE.erl

@@ -380,7 +380,7 @@ t_local(_) ->
 
     emqtt:stop(ConnPid1),
     emqtt:stop(ConnPid2),
-    stop_slave(Node),
+    emqx_node_helpers:stop_slave(Node),
 
     ?assertEqual(local, emqx_shared_sub:strategy(<<"local_group">>)),
     ?assertEqual(local, RemoteLocalGroupStrategy),
@@ -415,7 +415,7 @@ t_local_fallback(_) ->
     {true, UsedSubPid2} = last_message(<<"hello2">>, [ConnPid1]),
 
     emqtt:stop(ConnPid1),
-    stop_slave(Node),
+    emqx_node_helpers:stop_slave(Node),
 
     ?assertEqual(UsedSubPid1, UsedSubPid2),
     ok.
@@ -536,55 +536,8 @@ recv_msgs(Count, Msgs) ->
     end.
 
 start_slave(Name, Port) ->
-    {ok, Node} = ct_slave:start(list_to_atom(atom_to_list(Name) ++ "@" ++ host()),
+    Listeners = [#{listen_on => {{127,0,0,1}, Port},
-                                [{kill_if_fail, true},
-                                 {monitor_master, true},
-                                 {init_timeout, 10000},
-                                 {startup_timeout, 10000},
-                                 {erl_flags, ebin_path()}]),
-
-    pong = net_adm:ping(Node),
-    ok = setup_node(Node, Port),
-    Node.
-
-stop_slave(Node) ->
-    rpc:call(Node, ekka, leave, []),
-    ct_slave:stop(Node).
-
-host() ->
-    [_, Host] = string:tokens(atom_to_list(node()), "@"), Host.
-
-ebin_path() ->
-    string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " ").
-
-is_lib(Path) ->
-    string:prefix(Path, code:lib_dir()) =:= nomatch.
-
-setup_node(Node, Port) ->
-    EnvHandler =
-        fun(emqx) ->
-                application:set_env(
-                  emqx,
-                  listeners,
-                  [#{listen_on => {{127,0,0,1},Port},
                    name => "internal",
                    opts => [{zone,internal}],
-                     proto => tcp}]),
+                   proto => tcp}],
-                application:set_env(gen_rpc, port_discovery, stateless),
+    emqx_node_helpers:start_slave(Name, #{listeners => Listeners}).
-                ok;
-           (_) ->
-                ok
-        end,
-
-    [ok = rpc:call(Node, application, load, [App]) || App <- [gen_rpc, emqx]],
-    ok = rpc:call(Node, emqx_ct_helpers, start_apps, [[emqx], EnvHandler]),
-    rpc:call(Node, ekka, join, [node()]),
-
-    %% Sanity check. Assert that `gen_rpc' is set up correctly:
-    ?assertEqual( Node
-                , gen_rpc:call(Node, erlang, node, [])
-                ),
-    ?assertEqual( node()
-                , gen_rpc:call(Node, gen_rpc, call, [node(), erlang, node, []])
-                ),
-    ok.