Merge pull request #9346 from zmstone/1110-sync-v44

1110 sync v44
2022-11-10 18:21:13 +01:00 · 2022-11-10 18:21:13 +01:00 · dae7371f3b
parent 2fe841c451 c1d61a6f30
commit dae7371f3b
11 changed files with 76 additions and 32 deletions
--- a/.github/workflows/build_packages.yaml
+++ b/.github/workflows/build_packages.yaml
@ -132,8 +132,8 @@ jobs:
        apple_developer_id_bundle_password: ${{ secrets.APPLE_DEVELOPER_ID_BUNDLE_PASSWORD }}
    - uses: actions/upload-artifact@v3
      with:
-        name: ${EMQX_NAME}-${{ matrix.otp }}
-        path: _packages/${EMQX_NAME}/
+        name: ${{ env.EMQX_NAME }}-${{ matrix.otp }}
+        path: _packages/${{ env.EMQX_NAME }}/

  linux:
    runs-on: ubuntu-20.04
--- a/2
+++ b/2
@ -21,7 +21,7 @@ REL_PROFILES := emqx emqx-edge
 PKG_PROFILES := emqx-pkg emqx-edge-pkg
 PROFILES := $(REL_PROFILES) $(PKG_PROFILES) default

-CT_READABLE ?= false
+CT_READABLE ?= true

 export REBAR_GIT_CLONE_OPTIONS += --depth=1

--- a/apps/emqx_rule_engine/src/emqx_rule_engine.erl
+++ b/apps/emqx_rule_engine/src/emqx_rule_engine.erl
@ -79,18 +79,19 @@
             , action_instance_params/0
             ]).

-define(T_RETRY, 60000).
-
 %% redefine this macro to confine the appup scope
 -undef(RAISE).
 -define(RAISE(_EXP_, _ERROR_CONTEXT_),
+        ?RAISE(_EXP_, do_nothing, _ERROR_CONTEXT_)).
+-define(RAISE(_EXP_, _EXP_ON_FAIL_, _ERROR_CONTEXT_),
        fun() ->
            try (_EXP_)
            catch
                throw : Reason ->
                    throw({_ERROR_CONTEXT_, Reason});
                _EXCLASS_:_EXCPTION_:_ST_ ->
-                    throw({_ERROR_CONTEXT_, {_EXCPTION_, _EXCPTION_, _ST_}})
+                    _EXP_ON_FAIL_,
+                    throw({_ERROR_CONTEXT_, {_EXCLASS_, _EXCPTION_, _ST_}})
            end
        end()).

@ -498,7 +499,12 @@ refresh_resource(Type) when is_atom(Type) ->
 refresh_resource(#resource{id = ResId, type = Type, config = Config}) ->
    {ok, #resource_type{on_create = {M, F}}} =
        emqx_rule_registry:find_resource_type(Type),
-    ok = emqx_rule_engine:init_resource_with_retrier(M, F, ResId, Config).
+    try
+        init_resource_with_retrier(M, F, ResId, Config)
+    catch
+        throw:Reason ->
+            ?LOG_SENSITIVE(warning, "refresh_resource failed: ~0p", [Reason])
+    end.

 -spec(refresh_rules_when_boot() -> ok).
 refresh_rules_when_boot() ->
@ -677,17 +683,12 @@ init_resource(Module, OnCreate, ResId, Config) ->
    emqx_rule_registry:add_resource_params(ResParams).

 init_resource_with_retrier(Module, OnCreate, ResId, Config) ->
-    try
-        Params = Module:OnCreate(ResId, Config),
-        ResParams = #resource_params{id = ResId,
-                                    params = Params,
-                                    status = #{is_alive => true}},
-        emqx_rule_registry:add_resource_params(ResParams)
-    catch Class:Reason:ST ->
-        Interval = persistent_term:get({emqx_rule_engine, resource_restart_interval}, ?T_RETRY),
-        emqx_rule_monitor:ensure_resource_retrier(ResId, Interval),
-        erlang:raise(Class, {init_resource, Reason}, ST)
-    end.
+    Params = ?RAISE(Module:OnCreate(ResId, Config),
+                emqx_rule_monitor:ensure_resource_retrier(ResId), {Module, OnCreate}),
+    ResParams = #resource_params{id = ResId,
+                                 params = Params,
+                                 status = #{is_alive => true}},
+    emqx_rule_registry:add_resource_params(ResParams).

 init_action(Module, OnCreate, ActionInstId, Params) ->
    ok = emqx_rule_metrics:create_metrics(ActionInstId),
--- a/apps/emqx_rule_engine/src/emqx_rule_monitor.erl
+++ b/apps/emqx_rule_engine/src/emqx_rule_monitor.erl
@ -32,10 +32,18 @@
 -export([ start_link/0
        , stop/0
        , async_refresh_resources_rules/0
-        , ensure_resource_retrier/2
+        , ensure_resource_retrier/1
        , retry_loop/3
        ]).

+%% fot test
+-export([ put_retry_interval/1
+        , get_retry_interval/0
+        , erase_retry_interval/0
+        ]).
+
+-define(T_RETRY, 60000).
+
 start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

@ -46,10 +54,22 @@ init([]) ->
    _ = erlang:process_flag(trap_exit, true),
    {ok, #{retryers => #{}}}.

+put_retry_interval(I) when is_integer(I) andalso I >= 10 ->
+    _ = persistent_term:put({?MODULE, resource_restart_interval}, I),
+    ok.
+
+erase_retry_interval() ->
+    _ = persistent_term:erase({?MODULE, resource_restart_interval}),
+    ok.
+
+get_retry_interval() ->
+    persistent_term:get({?MODULE, resource_restart_interval}, ?T_RETRY).
+
 async_refresh_resources_rules() ->
    gen_server:cast(?MODULE, async_refresh).

-ensure_resource_retrier(ResId, Interval) ->
+ensure_resource_retrier(ResId) ->
+    Interval = get_retry_interval(),
    gen_server:cast(?MODULE, {create_restart_handler, resource, ResId, Interval}).

 handle_call(_Msg, _From, State) ->
@ -111,11 +131,12 @@ update_object(Tag, Obj, Retryer, State) ->
    }.

 create_restart_handler(Tag, Obj, Interval) ->
-    ?LOG(info, "keep restarting ~p ~p, interval: ~p", [Tag, Obj, Interval]),
+    ?LOG(info, "starting_a_retry_loop for ~p ~p, with delay interval: ~p", [Tag, Obj, Interval]),
    %% spawn a dedicated process to handle the restarting asynchronously
    spawn_link(?MODULE, retry_loop, [Tag, Obj, Interval]).

 retry_loop(resource, ResId, Interval) ->
+    timer:sleep(Interval),
    case emqx_rule_registry:find_resource(ResId) of
        {ok, #resource{type = Type, config = Config}} ->
            try
@ -124,10 +145,15 @@ retry_loop(resource, ResId, Interval) ->
                ok = emqx_rule_engine:init_resource(M, F, ResId, Config),
                refresh_and_enable_rules_of_resource(ResId)
            catch
-                Err:Reason:ST ->
-                    ?LOG(warning, "init_resource failed: ~p, ~0p",
-                        [{Err, Reason}, ST]),
-                    timer:sleep(Interval),
+                Err:Reason:Stacktrace ->
+                    %% do not log stacktrace if it's a throw
+                    LogContext =
+                        case Err of
+                            throw -> Reason;
+                            _ -> {Reason, Stacktrace}
+                        end,
+                    ?LOG_SENSITIVE(warning, "init_resource_retry_failed ~p, ~0p", [ResId, LogContext]),
+                    %% keep looping
                    ?MODULE:retry_loop(resource, ResId, Interval)
            end;
        not_found ->
--- a/apps/emqx_rule_engine/test/emqx_rule_monitor_SUITE.erl
+++ b/apps/emqx_rule_engine/test/emqx_rule_monitor_SUITE.erl
@ -48,7 +48,7 @@ end_per_suite(_Config) ->
    ok.

 init_per_testcase(t_restart_resource, Config) ->
-    persistent_term:put({emqx_rule_engine, resource_restart_interval}, 100),
+    emqx_rule_monitor:put_retry_interval(100),
    Opts = [public, named_table, set, {read_concurrency, true}],
    _ = ets:new(?RES_PARAMS_TAB, [{keypos, #resource_params.id}|Opts]),
    ets:new(t_restart_resource, [named_table, public]),
@ -77,7 +77,6 @@ init_per_testcase(_, Config) ->
    Config.

 end_per_testcase(t_restart_resource, Config) ->
-    persistent_term:put({emqx_rule_engine, resource_restart_interval}, 60000),
    ets:delete(t_restart_resource),
    common_end_per_testcases(),
    Config;
@ -91,7 +90,9 @@ end_per_testcase(_, Config) ->

 common_init_per_testcase() ->
    {ok, _} = emqx_rule_monitor:start_link().
+
 common_end_per_testcases() ->
+    emqx_rule_monitor:erase_retry_interval(),
    emqx_rule_monitor:stop().

 t_restart_resource(_) ->
--- a/apps/emqx_stomp/test/emqx_stomp_SUITE.erl
+++ b/apps/emqx_stomp/test/emqx_stomp_SUITE.erl
@ -349,7 +349,7 @@ t_1000_msg_send(_) ->
            receive
                {deliver, Topic, _Msg}->
                    ok
-            after 100 ->
+            after 5000 ->
                      ?assert(false, "waiting message timeout")
            end
        end,
--- a/11
+++ b/11
@ -62,7 +62,18 @@ log() {
    echo "===< $msg"
 }

+delete_unwanted_file() {
+    if [ -e "${1}" ]; then
+        log "Deleting file: ${1}"
+        rm -f "${1}"
+    else
+        log "Cannot delete file: ${1} -- file not found"
+    fi
+}
+
 make_rel() {
+    ./rebar3 as "$PROFILE" release
+    delete_unwanted_file _build/"${PROFILE}"/rel/emqx/lib/certifi*/priv/cacerts.pem
    ./rebar3 as "$PROFILE" tar
 }

--- a/changes/v4.3.22-en.md
+++ b/changes/v4.3.22-en.md
@ -36,13 +36,15 @@
  For example: `acl_order = jwt,http`, this will make sure `jwt` is always checked before `http`,
  meaning if JWT is not found (or no `acl` cliam) for a client, then the ACL check will fallback to use the HTTP backend.

-
 - Added configurations to enable more `client.disconnected` events (and counter bumps) [#9267](https://github.com/emqx/emqx/pull/9267).
  Prior to this change, the `client.disconnected` event (and counter bump) is triggered when a client
  performs a 'normal' disconnect, or is 'kicked' by system admin, but NOT triggered when a
  stale connection had to be 'discarded' (for clean session) or 'takenover' (for non-clean session).
  Now it is possible to set configs `broker.client_disconnect_discarded` and `broker.client_disconnect_takenover` to `on` to enable the event in these scenarios.

+- For Rule-Engine resource creation failure, delay before the first retry [#9313](https://github.com/emqx/emqx/pull/9313).
+  Prior to this change, the retry delay was added *after* the retry failure.
+
 ## Bug fixes

 - Fix that after uploading a backup file with an non-ASCII filename, HTTP API `GET /data/export` fails with status code 500 [#9224](https://github.com/emqx/emqx/pull/9224).
--- a/changes/v4.3.22-zh.md
+++ b/changes/v4.3.22-zh.md
@ -37,6 +37,9 @@
  但不会在旧 session 被废弃 (clean_session = true) 或旧 session 被接管 (clean_session = false) 时被触发。
  可将 `broker.client_disconnect_discarded` 和 `broker.client_disconnect_takovered` 选项设置为 `on` 来启用此场景下的客户端断连事件。

+- 规则引擎资源创建失败后，第一次重试前增加一个延迟 [#9313](https://github.com/emqx/emqx/pull/9313)。
+  在此之前，重试的延迟发生在重试失败之后。
+
 ## 修复

 - 修复若上传的备份文件名中包含非 ASCII 字符，`GET /data/export` HTTP 接口返回 500 错误 [#9224](https://github.com/emqx/emqx/pull/9224)。
--- a/deploy/charts/emqx/Chart.yaml
+++ b/deploy/charts/emqx/Chart.yaml
@ -13,8 +13,8 @@ type: application

 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
-version: 4.4.10
+version: 4.4.11

 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application.
-appVersion: 4.4.10
+appVersion: 4.4.11
--- a/include/emqx_release.hrl
+++ b/include/emqx_release.hrl
@ -29,7 +29,7 @@

 -ifndef(EMQX_ENTERPRISE).

-define(EMQX_RELEASE, {opensource, "4.4.10"}).
+-define(EMQX_RELEASE, {opensource, "4.4.11-alpha.1"}).

 -else.