Merge remote-tracking branch 'ce/main-v4.3' into merge-main-v4.3-into-v4.4

.github/workflows/apps_version_check.yaml

@@ -6,7 +6,22 @@ jobs:
   check_apps_version:
     runs-on: ubuntu-20.04
 
+    strategy:
+      matrix:
+        erl_otp:
+        - erl23.2.7.2-emqx-3
+        os:
+        - ubuntu20.04
+
+    container: emqx/build-env:${{ matrix.erl_otp }}-${{ matrix.os }}
+
     steps:
       - uses: actions/checkout@v1
       - name: Check apps version
         run: ./scripts/apps-version-check.sh
+      - name: Check relup (ce)
+        if: endsWith(github.repository, 'emqx')
+        run: ./scripts/update-appup.sh emqx --check
+      - name: Check relup (ee)
+        if: endsWith(github.repository, 'enterprise')
+        run: ./scripts/update-appup.sh emqx-ee --check

CHANGES-4.3.md

@@ -36,6 +36,7 @@ File format:
 * Fix Stomp client can not trigger `$event/client_connection` message [#7096]
 * Fix system memory false alarm at boot
 * Fix the MQTT-SN message replay when the topic is not registered to the client [#6970]
+* Fix rpc get node info maybe crash when other nodes is not ready.
 
 ## v4.3.12
 ### Important changes

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.5"}, % strict semver, bump manually
+  {vsn, "4.3.6"}, % strict semver, bump manually
   {modules, []},
   {registered, []},
   {applications, [kernel,stdlib,mnesia]},

apps/emqx_auth_mnesia/src/emqx_auth_mnesia.appup.src

@@ -13,10 +13,14 @@
      {load_module,emqx_acl_mnesia_api,brutal_purge,soft_purge,[]},
      {load_module,emqx_acl_mnesia_cli,brutal_purge,soft_purge,[]}]},
    {<<"4.3.4">>,
-    [{load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_cli,brutal_purge,soft_purge,[]},
      {load_module,emqx_acl_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_app,brutal_purge,soft_purge,[]}]},
+   {<<"4.3.5">>,
+    [{load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}],
   [{<<"4.3.[0-3]">>,
     [{load_module,emqx_auth_mnesia_cli,brutal_purge,soft_purge,[]},
@@ -31,8 +35,12 @@
      {delete_module,emqx_acl_mnesia_migrator},
      {delete_module,emqx_acl_mnesia_db}]},
    {<<"4.3.4">>,
-    [{load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]},
+    [{load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_cli,brutal_purge,soft_purge,[]},
      {load_module,emqx_acl_mnesia,brutal_purge,soft_purge,[]},
      {load_module,emqx_auth_mnesia_app,brutal_purge,soft_purge,[]}]},
+   {<<"4.3.5">>,
+    [{load_module,emqx_auth_mnesia_api,brutal_purge,soft_purge,[]},
+     {load_module,emqx_auth_mnesia,brutal_purge,soft_purge,[]}]},
    {<<".*">>,[]}]}.

apps/emqx_auth_mnesia/src/emqx_auth_mnesia.erl

@@ -1,5 +1,5 @@
 %%--------------------------------------------------------------------
-%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% 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.
@@ -72,7 +72,8 @@ check(ClientInfo = #{ clientid := Clientid
         List ->
             case match_password(NPassword, HashType, List)  of
                 false ->
-                    ?LOG(error, "[Mnesia] Auth from mnesia failed: ~p", [ClientInfo]),
+                    Info = maps:without([password], ClientInfo),
+                    ?LOG(info, "[Mnesia] Auth from mnesia failed: ~p", [Info]),
                     emqx_metrics:inc(?AUTH_METRICS(failure)),
                     {stop, AuthResult#{anonymous => false, auth_result => password_error}};
                 _ ->

apps/emqx_rule_engine/src/emqx_rule_engine.app.src

@@ -1,6 +1,6 @@
 {application, emqx_rule_engine,
  [{description, "EMQ X Rule Engine"},
-  {vsn, "4.4.1"}, % strict semver, bump manually!
+  {vsn, "4.4.2"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_rule_engine_sup, emqx_rule_registry]},
   {applications, [kernel,stdlib,rulesql,getopt]},

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -1,15 +1,23 @@
 %% -*- mode: erlang -*-
 {VSN,
-  [{"4.4.0",
+  [{"4.4.1",
-    [ {update, emqx_rule_metrics, {advanced, ["4.4.0"]}}
+    [ {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}
+    ]},
+   {"4.4.0",
+    [ {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}
+    , {update, emqx_rule_metrics, {advanced, ["4.4.0"]}}
     , {load_module,emqx_rule_events,brutal_purge,soft_purge,[]}
     , {load_module,emqx_rule_engine,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.0",
+  [{"4.4.1",
-    [ {update, emqx_rule_metrics, {advanced, ["4.4.0"]}}
+    [ {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}
+    ]},
+   {"4.4.0",
+    [ {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}
+    , {update, emqx_rule_metrics, {advanced, ["4.4.0"]}}
     , {load_module,emqx_rule_events,brutal_purge,soft_purge,[]}
     , {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]}
     , {load_module,emqx_rule_runtime,brutal_purge,soft_purge,[]}

apps/emqx_sn/rebar.config

@@ -2,7 +2,7 @@
 {plugins, [rebar3_proper]}.
 
 {deps,
- [{esockd, {git, "https://github.com/emqx/esockd", {tag, "5.7.4"}}},
+ [{esockd, {git, "https://github.com/emqx/esockd", {tag, "5.8.5"}}},
   {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v3.0.0"}}}
  ]}.
 

bin/install_upgrade.escript

@@ -144,9 +144,10 @@ parse_arguments([VersionStr|Rest], Acc) ->
     parse_arguments(Rest, [{version, Version}] ++ Acc).
 
 unpack_release(RelName, TargetNode, Version) ->
+    StartScriptExists = filelib:is_dir(filename:join(["releases", Version, "start.boot"])),
     WhichReleases = which_releases(TargetNode),
     case proplists:get_value(Version, WhichReleases) of
-        undefined ->
+        Res when Res =:= undefined; (Res =:= unpacked andalso not StartScriptExists) ->
             %% not installed, so unpack tarball:
             %% look for a release package with the intended version in the following order:
             %%      releases/<relname>-<version>.tar.gz
@@ -161,10 +162,39 @@ unpack_release(RelName, TargetNode, Version) ->
                     case rpc:call(TargetNode, release_handler, unpack_release,
                                   [ReleasePackageLink], ?TIMEOUT) of
                         {ok, Vsn} -> {ok, Vsn};
+                        {error, {existing_release, Vsn}} ->
+                            %% sometimes the user may have removed the release/<vsn> dir
+                            %% for an `unpacked` release, then we need to re-unpack it from
+                            %% the .tar ball
+                            untar_for_unpacked_release(str(RelName), Vsn),
+                            {ok, Vsn};
                         {error, _} = Error -> Error
                     end
             end;
-        Other -> Other
+        Other ->
+            Other
+    end.
+
+untar_for_unpacked_release(RelName, Vsn) ->
+    {ok, Root} = file:get_cwd(),
+    RelDir = filename:join([Root, "releases"]),
+    %% untar the .tar file, so release/<vsn> will be created
+    Tar = filename:join([RelDir, Vsn, RelName ++ ".tar.gz"]),
+    extract_tar(Root, Tar),
+
+    %% create RELEASE file
+    RelFile = filename:join([RelDir, Vsn, RelName ++ ".rel"]),
+    release_handler:create_RELEASES(Root, RelFile),
+
+    %% Clean release
+    _ = file:delete(Tar),
+    _ = file:delete(RelFile).
+
+extract_tar(Cwd, Tar) ->
+    case erl_tar:extract(Tar, [keep_old_files, {cwd, Cwd}, compressed]) of
+        ok -> ok;
+        {error, {Name, Reason}} ->		% New erl_tar (R3A).
+            throw({error, {cannot_extract_file, Name, Reason}})
     end.
 
 %% 1. look for a release package tarball with the provided version in the following order:
@@ -391,3 +421,10 @@ erts_vsn() ->
     {ok, Str} = file:read_file(filename:join(["releases", "start_erl.data"])),
     [ErtsVsn, _] = string:tokens(binary_to_list(Str), " "),
     ErtsVsn.
+
+str(A) when is_atom(A) ->
+    atom_to_list(A);
+str(A) when is_binary(A) ->
+    binary_to_list(A);
+str(A) when is_list(A) ->
+    (A).

deploy/charts/emqx/templates/configmap.env.yaml

@@ -11,7 +11,7 @@ metadata:
     app.kubernetes.io/managed-by: {{ .Release.Service }}
 data:
   {{- range $index, $value := .Values.emqxConfig }}
-  {{- if ne $value nil }}
+  {{- if $value }}
   {{- $key := (regexReplaceAllLiteral "\\." (regexReplaceAllLiteral "EMQX[_\\.]" (upper (trimAll " " $index)) "") "__") }}
   {{ print "EMQX_" $key }}: "{{ tpl (printf "%v" $value) $ }}"
   {{- end }}

rebar.config

@@ -42,7 +42,7 @@
     , {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"}}}
-    , {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.8.4"}}}
+    , {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.8.5"}}}
     , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.8.1.8"}}}
     , {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "2.7.0"}}}
     , {cuttlefish, {git, "https://github.com/emqx/cuttlefish", {tag, "v3.3.6"}}}

scripts/update-appup.sh

@@ -0,0 +1,63 @@
+#!/usr/bin/env bash
+
+## This script wrapps update_appup.escript,
+## it provides a more commonly used set of default args.
+
+## Arg1: EMQX PROFILE
+
+set -euo pipefail
+
+usage() {
+    echo "$0 PROFILE PREV_VERSION"
+}
+# ensure dir
+cd -P -- "$(dirname -- "${BASH_SOURCE[0]}")/.."
+
+PROFILE="${1:-}"
+case "$PROFILE" in
+    emqx-ee)
+        DIR='enterprise'
+        TAG_PREFIX='e'
+        ;;
+    emqx)
+        DIR='broker'
+        TAG_PREFIX='v'
+        ;;
+    emqx-edge)
+        DIR='edge'
+        TAG_PREFIX='v'
+        ;;
+    *)
+        echo "Unknown profile $PROFILE"
+        usage
+        exit 1
+        ;;
+esac
+
+PREV_VERSION="$(git describe --tag --match "${TAG_PREFIX}*" | grep -oE "${TAG_PREFIX}4\.[0-9]+\.[0-9]+")"
+PREV_VERSION="${PREV_VERSION#[e|v]}"
+
+shift 1
+ESCRIPT_ARGS="$*"
+
+SYSTEM="${SYSTEM:-$(./scripts/get-distro.sh)}"
+if [ -z "${ARCH:-}" ]; then
+    UNAME="$(uname -m)"
+    case "$UNAME" in
+        x86_64)
+            ARCH='amd64'
+            ;;
+        aarch64)
+            ARCH='arm64'
+            ;;
+        arm*)
+            ARCH='arm'
+            ;;
+    esac
+fi
+
+PACKAGE_NAME="${PROFILE}-${SYSTEM}-${PREV_VERSION}-${ARCH}.zip"
+DOWNLOAD_URL="https://www.emqx.com/downloads/${DIR}/v${PREV_VERSION}/${PACKAGE_NAME}"
+
+# shellcheck disable=SC2086
+./scripts/update_appup.escript --make-command "make ${PROFILE}-rel" --binary-rel-url "$DOWNLOAD_URL" $ESCRIPT_ARGS "$PREV_VERSION"

scripts/update_appup.escript

@@ -171,9 +171,11 @@ download_prev_release(Tag, #{binary_rel_url := {ok, URL0}, clone_url := Repo}) -
     BaseDir = "/tmp/emqx-baseline-bin/",
     Dir = filename:basename(Repo, ".git") ++ [$-|Tag],
     Filename = filename:join(BaseDir, Dir),
-    Script = "echo \"Download: ${OUTFILE}\" &&
+    Script = "mkdir -p ${OUTFILE} &&
-              mkdir -p ${OUTFILE} &&
+              if [ ! -f \"${OUTFILE}.zip\" ]; then \
-              curl -f -L -o ${OUTFILE}.zip ${URL} &&
+                echo \"Download: ${OUTFILE}\" && \
+                curl -f -L -o \"${OUTFILE}.zip\" \"${URL}\"; \
+              fi &&
               unzip -q -n -d ${OUTFILE} ${OUTFILE}.zip",
     Env = [{"TAG", Tag}, {"OUTFILE", Filename}, {"URL", URL}],
     bash(Script, Env),
@@ -540,20 +542,31 @@ diff_app(UpOrDown, App,
                  , NewModules
                  ),
     Deleted = maps:keys(maps:without(maps:keys(NewModules), OldModules)),
-    NChanges = length(New) + length(Changed) + length(Deleted),
+    Changes = lists:filter(fun({_T, L}) -> length(L) > 0 end,
+                           [{added, New}, {changed, Changed}, {deleted, Deleted}]),
     case NewVersion =:= OldVersion of
-        true when NChanges =:= 0 ->
+        true when Changes =:= [] ->
             %% no change
             ok;
         true ->
             set_invalid(),
-            log("ERROR: Application '~p' contains changes, but its version is not updated~n", [App]);
+            case UpOrDown =:= up of
+                true ->
+                    %% only log for the upgrade case because it would be the same result
+                    log("ERROR: Application '~p' contains changes, but its version is not updated. ~s",
+                        [App, format_changes(Changes)]);
+                false ->
+                    ok
+            end;
         false ->
             log("INFO: Application '~p' has been updated: ~p --[~p]--> ~p~n", [App, OldVersion, UpOrDown, NewVersion]),
             ok
     end,
     {New, Changed, Deleted}.
 
+format_changes(Changes) ->
+    lists:map(fun({Tag, List}) -> io_lib:format("~p: ~p~n", [Tag, List]) end, Changes).
+
 -spec hashsums(file:filename()) -> #{module() => binary()}.
 hashsums(EbinDir) ->
     maps:from_list(lists:map(
@@ -614,7 +627,9 @@ locate(src, App, Suffix) ->
         [File] ->
             {ok, File};
         [] ->
-            undefined
+            undefined;
+        Files ->
+            error({more_than_one_app_found, Files})
     end.
 
 find_app(Pattern) ->

src/emqx_sys.erl

@@ -94,7 +94,15 @@ sysdescr() -> emqx_app:get_description().
 %% @doc Get sys uptime
 -spec(uptime() -> string()).
 uptime() ->
-    gen_server:call(?SYS, uptime).
+    {TotalWallClock, _} = erlang:statistics(wall_clock),
+    uptime(TotalWallClock div 1000).
+
+uptime(Seconds) ->
+    {D, {H, M, S}} = calendar:seconds_to_daystime(Seconds),
+    L0 = [{D, " days"}, {H, " hours"}, {M, " minutes"}, {S, " seconds"}],
+    L1 = lists:dropwhile(fun({K, _}) ->  K =:= 0 end, L0),
+    L2 = lists:map(fun({Time, Unit}) -> [integer_to_list(Time), Unit] end, L1),
+    lists:flatten(lists:join(", ", L2)).
 
 %% @doc Get sys datetime
 -spec(datetime() -> string()).
@@ -137,9 +145,6 @@ heartbeat(State) ->
 tick(State) ->
     State#state{ticker = start_timer(sys_interval(), tick)}.
 
-handle_call(uptime, _From, State) ->
-    {reply, uptime(State), State};
-
 handle_call(Req, _From, State) ->
     ?LOG(error, "Unexpected call: ~p", [Req]),
     {reply, ignored, State}.
@@ -149,7 +154,7 @@ handle_cast(Msg, State) ->
     {noreply, State}.
 
 handle_info({timeout, TRef, heartbeat}, State = #state{heartbeat = TRef}) ->
-    publish_any(uptime, iolist_to_binary(uptime(State))),
+    publish_any(uptime, iolist_to_binary(uptime())),
     publish_any(datetime, iolist_to_binary(datetime())),
     {noreply, heartbeat(State)};
 
@@ -173,24 +178,6 @@ terminate(_Reason, #state{heartbeat = TRef1, ticker = TRef2}) ->
 %% Internal functions
 %%-----------------------------------------------------------------------------
 
-uptime(#state{start_time = Ts}) ->
-    Secs = timer:now_diff(erlang:timestamp(), Ts) div 1000000,
-    lists:flatten(uptime(seconds, Secs)).
-uptime(seconds, Secs) when Secs < 60 ->
-    [integer_to_list(Secs), " seconds"];
-uptime(seconds, Secs) ->
-    [uptime(minutes, Secs div 60), integer_to_list(Secs rem 60), " seconds"];
-uptime(minutes, M) when M < 60 ->
-    [integer_to_list(M), " minutes, "];
-uptime(minutes, M) ->
-    [uptime(hours, M div 60), integer_to_list(M rem 60), " minutes, "];
-uptime(hours, H) when H < 24 ->
-    [integer_to_list(H), " hours, "];
-uptime(hours, H) ->
-    [uptime(days, H div 24), integer_to_list(H rem 24), " hours, "];
-uptime(days, D) ->
-    [integer_to_list(D), " days, "].
-
 publish_any(Name, Value) ->
     _ = publish(Name, Value),
     ok.

test/emqx_sys_SUITE.erl

@@ -42,10 +42,23 @@ end_per_suite(_Config) ->
 %     error('TODO').
 
 t_uptime(_) ->
-    ?assertEqual(<<"1 seconds">>, iolist_to_binary(emqx_sys:uptime(seconds, 1))),
+    ?assert(is_list(emqx_sys:uptime())),
-    ?assertEqual(<<"1 minutes, 0 seconds">>, iolist_to_binary(emqx_sys:uptime(seconds, 60))),
+    ?assertEqual(<<"1 seconds">>, iolist_to_binary(emqx_sys:uptime(1))),
-    ?assertEqual(<<"1 hours, 0 minutes, 0 seconds">>, iolist_to_binary(emqx_sys:uptime(seconds, 3600))),
+    ?assertEqual(<<"1 minutes, 0 seconds">>, iolist_to_binary(emqx_sys:uptime(60))),
-    ?assertEqual(<<"1 days, 0 hours, 0 minutes, 0 seconds">>, iolist_to_binary(emqx_sys:uptime(seconds, 86400))).
+    ?assertEqual(<<"1 hours, 0 minutes, 0 seconds">>, iolist_to_binary(emqx_sys:uptime(3600))),
+    ?assertEqual(<<"1 hours, 1 minutes, 1 seconds">>, iolist_to_binary(emqx_sys:uptime(3661))),
+    ?assertEqual(<<"1 days, 0 hours, 0 minutes, 0 seconds">>, iolist_to_binary(emqx_sys:uptime(86400))),
+    lists:map(fun({D, H, M, S}) ->
+        Expect = <<
+            (integer_to_binary(D))/binary, " days, ",
+            (integer_to_binary(H))/binary, " hours, ",
+            (integer_to_binary(M))/binary, " minutes, ",
+            (integer_to_binary(S))/binary, " seconds"
+        >>,
+        Actual = iolist_to_binary(emqx_sys:uptime(D * 86400 + H * 3600 + M * 60 + S)),
+        ?assertEqual(Expect, Actual)
+              end,
+        [{1, 2, 3, 4}, {10, 20, 30, 40}, {2222, 3, 56, 59}, {59, 23, 59, 59}]).
 
 % t_datetime(_) ->
 %     error('TODO').