Merge pull request #6884 from emqx/copy_of_main-v4.3

Sync code from main-v4.3 to main-v4.4

.github/workflows/build_packages.yaml

@@ -32,14 +32,12 @@ jobs:
         shell: bash
         working-directory: source
         run: |
-          vsn="$(./pkg-vsn.sh)"
-          base_vsn_prefix="$(echo $vsn | grep -oE '^[0-9]+\.[0-9]+')"
           if make emqx-ee --dry-run > /dev/null 2>&1; then
-            old_vsns="$(git tag -l | grep -E "^e${base_vsn_prefix}\.[0-9]+$" | grep -v "e${vsn}" | xargs)"
+            old_vsns="$(./scripts/relup-base-vsns.sh enterprise | xargs)"
             echo "::set-output name=old_vsns::$old_vsns"
             echo "::set-output name=profiles::[\"emqx-ee\"]"
           else
-            old_vsns="$(git tag -l | grep -E "^v${base_vsn_prefix}\.[0-9]+$" | grep -v "v${vsn}" | xargs)"
+            old_vsns="$(./scripts/relup-base-vsns.sh community | xargs)"
             echo "::set-output name=old_vsns::$old_vsns"
             echo "::set-output name=profiles::[\"emqx\", \"emqx-edge\"]"
           fi
@@ -96,6 +94,7 @@ jobs:
       working-directory: source
       run: |
         $env:PATH = "${{ steps.install_erlang.outputs.erlpath }}\bin;$env:PATH"
+        erl -eval "erlang:display(crypto:info_lib())" -s init stop
 
         $version = $( "${{ github.ref }}" -replace "^(.*)/(.*)/" )
         if ($version -match "^v[0-9]+\.[0-9]+(\.[0-9]+)?") {

.github/workflows/run_fvt_tests.yaml

@@ -247,16 +247,13 @@ jobs:
           run: |
             cd emqx
             vsn="$(./pkg-vsn.sh)"
-            pre_vsn="$(echo $vsn | grep -oE '^[0-9]+.[0-9]')"
-
             if make emqx-ee --dry-run > /dev/null 2>&1; then
               profile="emqx-ee"
-              old_vsns="$(git tag -l "e$pre_vsn.[0-9]" | xargs echo -n | sed "s/e$vsn//")"
+              old_vsns="$(./scripts/relup-base-vsns.sh enterprise | xargs)"
               broker="emqx-ee"
-
             else
               profile="emqx"
-              old_vsns="$(git tag -l "v$pre_vsn.[0-9]" | xargs echo -n | sed "s/v$vsn//")"
+              old_vsns="$(./scripts/relup-base-vsns.sh community | xargs)"
               broker="emqx-ce"
             fi
 

CHANGES-4.3.md

@@ -16,6 +16,7 @@ File format:
 ### Minor changes
 * Fix updating `emqx_auth_mnesia.conf` password and restarting the new password does not take effect [#6717]
 * Fix import data crash when emqx_auth_mnesia's record is not empty [#6717]
+* Fix `os_mon.sysmem_high_watermark` may not alert after reboot.
 
 ## v4.3.11
 

Windows.md

@@ -25,17 +25,17 @@ C:\Program Files (x86)\Microsoft Visual Studio\2019\Community\VC\Auxiliary\Build
 
 Depending on your visual studio version and OS, the paths may differ.
 The first path is for rebar3 port compiler to find `cl.exe` and `link.exe`
-The second path is for Powershell or CMD to setup environment variables.
+The second path is for CMD to setup environment variables.
 
 ### Erlang/OTP
 
 Install Erlang/OTP 23.2 from https://www.erlang.org/downloads
 You may need to edit the `Path` environment variable to allow running
-Erlang commands such as `erl` from powershell.
+Erlang commands such as `erl` from CMD.
 
-To validate Erlang installation in CMD or powershell:
+To validate Erlang installation in CMD :
 
-* Start (or restart) CMD or powershell
+* Start (or restart) CMD
 
 * Execute `erl` command to enter Erlang shell
 
@@ -63,7 +63,7 @@ Cygwin is what we tested with.
   to `Path` list.
 
 * Validate installation.
-  Start (restart) CMD or powershell console and execute `which bash`, it should
+  Start (restart) CMD console and execute `which bash`, it should
   print out `/usr/bin/bash`
 
 ### Other tools
@@ -88,7 +88,7 @@ scoop install git curl make jq zip unzip
 
 * Clone the repo: `git clone https://github.com/emqx/emqx.git`
 
-* Start CMD or Powershell
+* Start CMD
 
 * Execute `vcvarsall.bat x86_amd64` to load environment variables
 

apps/emqx_auth_mnesia/src/emqx_auth_mnesia.erl

@@ -42,11 +42,11 @@ init(#{clientid_list := ClientidList, username_list := UsernameList}) ->
             {attributes, record_info(fields, emqx_user)},
             {storage_properties, [{ets, [{read_concurrency, true}]}]}]),
     lists:foreach(fun({Clientid, Password}) ->
-    emqx_auth_mnesia_cli:add_default_user(clientid, iolist_to_binary(Clientid), iolist_to_binary(Password))
+        emqx_auth_mnesia_cli:add_default_user(clientid, iolist_to_binary(Clientid), iolist_to_binary(Password))
     end, ClientidList),
 
     lists:foreach(fun({Username, Password}) ->
-    emqx_auth_mnesia_cli:add_default_user(username, iolist_to_binary(Username), iolist_to_binary(Password))
+        emqx_auth_mnesia_cli:add_default_user(username, iolist_to_binary(Username), iolist_to_binary(Password))
     end, UsernameList),
 
     ok = ekka_mnesia:copy_table(?TABLE, disc_copies).

apps/emqx_auth_mnesia/src/emqx_auth_mnesia_cli.erl

@@ -78,14 +78,14 @@ add_default_user(Type, Key, Password) ->
                             username -> user
                         end,
                     ?LOG(warning,
-                        "[Auth Mnesia] auth.client.x.~p=~s's password in the emqx_auth_mnesia.conf\n"
+                        "[Auth Mnesia] auth.client.x.~p=~s password in the emqx_auth_mnesia.conf\n"
                         "does not match the password in the database(mnesia).\n"
                         "1. If you have already changed the password via the HTTP API, this warning has no effect.\n"
-                        "You can remove the warning from emqx_auth_mnesia.conf to resolve the warning.\n"
+                        "You can remove the `auth.client.x.~p=~s` from emqx_auth_mnesia.conf to resolve this warning.\n"
                         "2. If you just want to update the password by manually changing the configuration file,\n"
                         "you need to delete the old user and password using `emqx_ctl ~p delete ~s` first\n"
                         "the new password in emqx_auth_mnesia.conf can take effect after reboot.",
-                        [Type, Key, TypeCtl, Key]),
+                        [Type, Key, Type, Key, TypeCtl, Key]),
                     ok
             end;
         Error -> Error

apps/emqx_stomp/src/emqx_stomp_frame.erl

@@ -123,6 +123,8 @@ parse(<<>>, Parser) ->
 
 parse(Bytes, #{phase := body, length := Len, state := State}) ->
     parse(body, Bytes, State, Len);
+parse(Bytes, #{phase := Phase, state := State}) when Phase =/= none ->
+    parse(Phase, Bytes, State);
 
 parse(Bytes, Parser = #{pre := Pre}) ->
     parse(<<Pre/binary, Bytes/binary>>, maps:without([pre], Parser));
@@ -153,6 +155,8 @@ parse(command, <<?LF, Rest/binary>>, State = #parser_state{acc = Acc}) ->
     parse(headers, Rest, State#parser_state{cmd = Acc, acc = <<>>});
 parse(command, <<Ch:8, Rest/binary>>, State) ->
     parse(command, Rest, acc(Ch, State));
+parse(command, <<>>, State) ->
+    {more, #{phase => command, state => State}};
 
 parse(headers, <<?LF, Rest/binary>>, State) ->
     parse(body, Rest, State, content_len(State#parser_state{acc = <<>>}));
@@ -165,11 +169,15 @@ parse(hdname, <<?COLON, Rest/binary>>, State = #parser_state{acc = Acc}) ->
     parse(hdvalue, Rest, State#parser_state{hdname = Acc, acc = <<>>});
 parse(hdname, <<Ch:8, Rest/binary>>, State) ->
     parse(hdname, Rest, acc(Ch, State));
+parse(hdname, <<>>, State) ->
+    {more, #{phase => hdname, state => State}};
 
 parse(hdvalue, <<?LF, Rest/binary>>, State = #parser_state{headers = Headers, hdname = Name, acc = Acc}) ->
     parse(headers, Rest, State#parser_state{headers = add_header(Name, Acc, Headers), hdname = undefined, acc = <<>>});
 parse(hdvalue, <<Ch:8, Rest/binary>>, State) ->
-    parse(hdvalue, Rest, acc(Ch, State)).
+    parse(hdvalue, Rest, acc(Ch, State));
+parse(hdvalue, <<>>, State) ->
+    {more, #{phase => hdvalue, state => State}}.
 
 %% @private
 parse(body, <<>>, State, Length) ->

apps/emqx_stomp/test/emqx_stomp_SUITE.erl

@@ -17,6 +17,7 @@
 -module(emqx_stomp_SUITE).
 
 -include_lib("emqx_stomp/include/emqx_stomp.hrl").
+-include_lib("eunit/include/eunit.hrl").
 
 -compile(export_all).
 -compile(nowarn_export_all).
@@ -324,6 +325,40 @@ t_ack(_) ->
                                                   body    = _}, _} = parse(Data4)
                     end).
 
+t_1000_msg_send(_) ->
+    with_connection(fun(Sock) ->
+        gen_tcp:send(Sock, serialize(<<"CONNECT">>,
+                                     [{<<"accept-version">>, ?STOMP_VER},
+                                      {<<"host">>, <<"127.0.0.1:61613">>},
+                                      {<<"login">>, <<"guest">>},
+                                      {<<"passcode">>, <<"guest">>},
+                                      {<<"heart-beat">>, <<"0,0">>}])),
+        {ok, Data} = gen_tcp:recv(Sock, 0),
+        {ok, #stomp_frame{command = <<"CONNECTED">>,
+                          headers = _,
+                          body    = _}, _} = parse(Data),
+
+        Topic = <<"/queue/foo">>,
+        SendFun = fun() ->
+            gen_tcp:send(Sock, serialize(<<"SEND">>,
+                                        [{<<"destination">>, Topic}],
+                                        <<"msgtest">>))
+        end,
+
+        RecvFun = fun() ->
+            receive
+                {deliver, Topic, _Msg}->
+                    ok
+            after 100 ->
+                      ?assert(false, "waiting message timeout")
+            end
+        end,
+
+        emqx:subscribe(Topic),
+        lists:foreach(fun(_) -> SendFun() end, lists:seq(1, 1000)),
+        lists:foreach(fun(_) -> RecvFun() end, lists:seq(1, 1000))
+    end).
+
 with_connection(DoFun) ->
     {ok, Sock} = gen_tcp:connect({127, 0, 0, 1},
                                  61613,

apps/emqx_web_hook/src/emqx_web_hook.erl

@@ -93,7 +93,6 @@ on_client_connect(ConnInfo = #{clientid := ClientId, username := Username, peern
               , ipaddress => iolist_to_binary(ntoa(Peerhost))
               , keepalive => maps:get(keepalive, ConnInfo)
               , proto_ver => maps:get(proto_ver, ConnInfo)
-              , connected_at => maps:get(connected_at, ConnInfo)
               },
     send_http_request(ClientId, Params).
 

apps/emqx_web_hook/test/props/prop_webhook_hooks.erl

@@ -52,8 +52,7 @@ prop_client_connect() ->
                       username => maybe(maps:get(username, ConnInfo)),
                       ipaddress => peer2addr(maps:get(peername, ConnInfo)),
                       keepalive => maps:get(keepalive, ConnInfo),
-                      proto_ver => maps:get(proto_ver, ConnInfo),
+                      proto_ver => maps:get(proto_ver, ConnInfo)
-                      connected_at => maps:get(connected_at, ConnInfo)
                      }),
            true
        end).

bin/emqx

@@ -348,6 +348,39 @@ bootstrapd() {
     fi
 }
 
+# check if a PID is down
+is_down() {
+    PID="$1"
+    if ps -p "$PID" >/dev/null; then
+        # still around
+        # shellcheck disable=SC2009 # this grep pattern is not a part of the progra names
+        if ps -p "$PID" | grep -q 'defunct'; then
+            return 0
+        fi
+        return 1
+    fi
+    # it's gone
+    return 0
+}
+
+wait_for() {
+    local WAIT_TIME
+    local CMD
+    WAIT_TIME="$1"
+    shift
+    CMD="$*"
+    while true; do
+        if $CMD >/dev/null 2>&1; then
+            return 0
+        fi
+        if [ "$WAIT_TIME" -le 0 ]; then
+            return 1
+        fi
+        WAIT_TIME=$((WAIT_TIME - 1))
+        sleep 1
+    done
+}
+
 # Use $CWD/etc/sys.config if exists
 if [ -z "$RELX_CONFIG_PATH" ]; then
     if [ -f "$RUNNER_ETC_DIR/sys.config" ]; then
@@ -530,13 +563,13 @@ case "$1" in
             echoerr "Graceful shutdown failed PID=[$PID]"
             exit 1
         fi
-        WAIT_TIME="${WAIT_FOR_ERLANG_STOP:-120}"
+        WAIT_TIME="${EMQX_WAIT_FOR_STOP:-120}"
         if ! wait_for "$WAIT_TIME" 'is_down' "$PID"; then
             msg="dangling after ${WAIT_TIME} seconds"
             # also log to syslog
             logger -t "${REL_NAME}[${PID}]" "STOP: $msg"
             # log to user console
-            echoerr "stop failed, $msg"
+            echoerr "Stop failed, $msg"
             echo "ERROR: $PID is still around"
             ps -p "$PID"
             exit 1

scripts/relup-base-vsns.sh

@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+## This script prints the relup upgrade base versions
+## for the given EMQ X edition (specified as first arg)
+##
+## The second argument is the current release version
+## if not provided, it's taken from pkg-vsn.sh
+
+usage() {
+    echo "Usage: $0 <EMQX_PROFILE> [<CURRENT_VERSION>]"
+    echo "e.g.   $0 enterprise 4.3.10"
+    exit 1
+}
+
+parse_semver() {
+    echo "$1" | tr '.|-' ' '
+}
+
+PROFILE="${1:-}"
+[ -z "${PROFILE}" ] && usage
+
+## Get the current release version
+## e.g.
+## 5.0.0                 when GA
+## 5.0.0-beta.3          when pre-release
+## 5.0.0-beta.3.abcdef00 when developing
+CUR="${2:-}"
+if [ -z "${CUR}" ]; then
+    CUR="$(./pkg-vsn.sh)"
+fi
+
+# shellcheck disable=SC2207
+CUR_SEMVER=($(parse_semver "$CUR"))
+
+if [ "${#CUR_SEMVER[@]}" -lt 3 ]; then
+    echo "$CUR is not Major.Minor.Patch"
+    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 "${PROFILE}" in
+    *enterprise*)
+        GIT_TAG_PREFIX="e"
+        ;;
+    *)
+        GIT_TAG_PREFIX="v"
+        ;;
+esac
+
+while read -r git_tag; do
+    # shellcheck disable=SC2207
+    semver=($(parse_semver "$git_tag"))
+    if [ "${#semver[@]}" -eq 3 ] && [ "${semver[2]}" -le "${CUR_SEMVER[2]}" ]; then
+        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
+            echo "$git_tag"
+        fi
+    fi
+done < <(git tag -l "${GIT_TAG_PREFIX}${CUR_SEMVER[0]}.${CUR_SEMVER[1]}.*")

src/emqx.appup.src

@@ -12,10 +12,16 @@
     , {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}
     , {load_module,emqx_mod_sup,brutal_purge,soft_purge,[]}
     , {load_module,emqx_session,brutal_purge,soft_purge,[]}
+    , {load_module,emqx_os_mon,brutal_purge,soft_purge,[]}
+    , {load_module,emqx,brutal_purge,soft_purge,[]}
+    , {load_module,emqx_app,brutal_purge,soft_purge,[]}
+    , {load_module,emqx_limiter,brutal_purge,soft_purge,[]}
     ]},
-   {<<".*">>,[]}],
+   {<<".*">>,[]}
+  ],
   [{"4.4.0",
     [ {load_module,emqx_metrics,brutal_purge,soft_purge,[]}
+    , {apply,{emqx_metrics,assign_acl_stats_from_ets_to_counter,[]}}
     , {load_module,emqx_access_control,brutal_purge,soft_purge,[]}
     , {load_module,emqx_channel,brutal_purge,soft_purge,[]}
     , {load_module,emqx_connection,brutal_purge,soft_purge,[]}
@@ -25,6 +31,11 @@
     , {load_module,emqx_slow_subs_api,brutal_purge,soft_purge,[]}
     , {load_module,emqx_mod_sup,brutal_purge,soft_purge,[]}
     , {load_module,emqx_session,brutal_purge,soft_purge,[]}
+    , {load_module,emqx_os_mon,brutal_purge,soft_purge,[]}
+    , {load_module,emqx,brutal_purge,soft_purge,[]}
+    , {load_module,emqx_app,brutal_purge,soft_purge,[]}
+    , {load_module,emqx_limiter,brutal_purge,soft_purge,[]}
     ]},
-   {<<".*">>,[]}]
+   {<<".*">>,[]}
+  ]
 }.

src/emqx_os_mon.erl

@@ -147,12 +147,12 @@ handle_info({timeout, Timer, check}, State = #{timer := Timer,
     case emqx_vm:cpu_util() of %% TODO: should be improved?
         0 ->
             State#{timer := undefined};
-        Busy when Busy >= CPUHighWatermark ->
+        Busy when Busy > CPUHighWatermark ->
             emqx_alarm:activate(high_cpu_usage, #{usage => Busy,
                                                   high_watermark => CPUHighWatermark,
                                                   low_watermark => CPULowWatermark}),
             ensure_check_timer(State);
-        Busy when Busy =< CPULowWatermark ->
+        Busy when Busy < CPULowWatermark ->
             emqx_alarm:deactivate(high_cpu_usage),
             ensure_check_timer(State);
         _Busy ->
@@ -191,7 +191,7 @@ ensure_system_memory_alarm(HW) ->
         undefined -> ok;
         _Pid ->
             {Allocated, Total, _Worst} = memsup:get_memory_data(),
-            case Total =/= 0 andalso Allocated/Total * 100 >= HW of
+            case Total =/= 0 andalso Allocated/Total * 100 > HW of
                 true -> emqx_alarm:activate(high_system_memory_usage, #{high_watermark => HW});
                 false -> ok
             end

src/emqx_vm_mon.erl

@@ -112,7 +112,7 @@ handle_info({timeout, Timer, check},
                       process_low_watermark := ProcLowWatermark}) ->
     ProcessCount = erlang:system_info(process_count),
     case ProcessCount / erlang:system_info(process_limit) * 100 of
-        Percent when Percent >= ProcHighWatermark ->
+        Percent when Percent > ProcHighWatermark ->
             emqx_alarm:activate(too_many_processes, #{usage => Percent,
                                                       high_watermark => ProcHighWatermark,
                                                       low_watermark => ProcLowWatermark});

test/emqx_alarm_SUITE.erl

@@ -67,7 +67,8 @@ t_alarm(_) ->
     ok = emqx_alarm:deactivate(unknown_alarm),
     {error, not_found} = emqx_alarm:deactivate(unknown_alarm),
     ?assertEqual({error, not_found}, get_alarm(unknown_alarm, emqx_alarm:get_alarms(activated))),
-    ?assertNotEqual({error, not_found}, get_alarm(unknown_alarm, emqx_alarm:get_alarms(deactivated))),
+    ?assertNotEqual({error, not_found},
+        get_alarm(unknown_alarm, emqx_alarm:get_alarms(deactivated))),
 
     emqx_alarm:delete_all_deactivated_alarms(),
     ?assertEqual({error, not_found}, get_alarm(unknown_alarm, emqx_alarm:get_alarms(deactivated))).
@@ -78,7 +79,8 @@ t_deactivate_all_alarms(_) ->
     ?assertNotEqual({error, not_found}, get_alarm(unknown_alarm, emqx_alarm:get_alarms(activated))),
 
     emqx_alarm:deactivate_all_alarms(),
-    ?assertNotEqual({error, not_found}, get_alarm(unknown_alarm, emqx_alarm:get_alarms(deactivated))),
+    ?assertNotEqual({error, not_found},
+        get_alarm(unknown_alarm, emqx_alarm:get_alarms(deactivated))),
 
     emqx_alarm:delete_all_deactivated_alarms(),
     ?assertEqual({error, not_found}, get_alarm(unknown_alarm, emqx_alarm:get_alarms(deactivated))).