Merge pull request #10817 from HJianBo/fix-restart-timer-interval

fix: fix auto_restart_interval checker

apps/emqx_bridge/test/emqx_bridge_api_SUITE.erl

@@ -79,7 +79,8 @@ groups() ->
     SingleOnlyTests = [
         t_broken_bpapi_vsn,
         t_old_bpapi_vsn,
-        t_bridges_probe
+        t_bridges_probe,
+        t_auto_restart_interval
     ],
     ClusterLaterJoinOnlyTCs = [t_cluster_later_join_metrics],
     [
@@ -550,6 +551,89 @@ t_http_crud_apis(Config) ->
 
     {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), Config).
 
+t_auto_restart_interval(Config) ->
+    Port = ?config(port, Config),
+    %% assert we there's no bridges at first
+    {ok, 200, []} = request_json(get, uri(["bridges"]), Config),
+
+    meck:new(emqx_resource, [passthrough]),
+    meck:expect(emqx_resource, call_start, fun(_, _, _) -> {error, fake_error} end),
+
+    %% then we add a webhook bridge, using POST
+    %% POST /bridges/ will create a bridge
+    URL1 = ?URL(Port, "path1"),
+    Name = ?BRIDGE_NAME,
+    BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE_HTTP, Name),
+    BridgeParams = ?HTTP_BRIDGE(URL1, Name)#{
+        <<"resource_opts">> => #{<<"auto_restart_interval">> => "1s"}
+    },
+    ?check_trace(
+        begin
+            ?assertMatch(
+                {ok, 201, #{
+                    <<"type">> := ?BRIDGE_TYPE_HTTP,
+                    <<"name">> := Name,
+                    <<"enable">> := true,
+                    <<"status">> := _,
+                    <<"node_status">> := [_ | _],
+                    <<"url">> := URL1
+                }},
+                request_json(
+                    post,
+                    uri(["bridges"]),
+                    BridgeParams,
+                    Config
+                )
+            ),
+            {ok, _} = ?block_until(#{?snk_kind := resource_disconnected_enter}),
+            {ok, _} = ?block_until(#{?snk_kind := resource_auto_reconnect}, 1500)
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(resource_auto_reconnect, Trace0),
+            ?assertMatch([#{}], Trace),
+            ok
+        end
+    ),
+    %% delete the bridge
+    {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), Config),
+    {ok, 200, []} = request_json(get, uri(["bridges"]), Config),
+
+    %% auto_retry_interval=infinity
+    BridgeParams1 = BridgeParams#{
+        <<"resource_opts">> => #{<<"auto_restart_interval">> => "infinity"}
+    },
+    ?check_trace(
+        begin
+            ?assertMatch(
+                {ok, 201, #{
+                    <<"type">> := ?BRIDGE_TYPE_HTTP,
+                    <<"name">> := Name,
+                    <<"enable">> := true,
+                    <<"status">> := _,
+                    <<"node_status">> := [_ | _],
+                    <<"url">> := URL1
+                }},
+                request_json(
+                    post,
+                    uri(["bridges"]),
+                    BridgeParams1,
+                    Config
+                )
+            ),
+            {ok, _} = ?block_until(#{?snk_kind := resource_disconnected_enter}),
+            ?assertEqual(timeout, ?block_until(#{?snk_kind := resource_auto_reconnect}, 1500))
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(resource_auto_reconnect, Trace0),
+            ?assertMatch([], Trace),
+            ok
+        end
+    ),
+    %% delete the bridge
+    {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), Config),
+    {ok, 200, []} = request_json(get, uri(["bridges"]), Config),
+    meck:unload(emqx_resource).
+
 t_http_bridges_local_topic(Config) ->
     Port = ?config(port, Config),
     %% assert we there's no bridges at first

apps/emqx_resource/include/emqx_resource.hrl

@@ -66,7 +66,7 @@
     start_after_created => boolean(),
     %% If the resource disconnected, we can set to retry starting the resource
     %% periodically.
-    auto_restart_interval => pos_integer(),
+    auto_restart_interval => pos_integer() | infinity,
     batch_size => pos_integer(),
     batch_time => pos_integer(),
     max_buffer_bytes => pos_integer(),

apps/emqx_resource/src/emqx_resource_manager.erl

@@ -389,8 +389,10 @@ handle_event(state_timeout, health_check, connected, Data) ->
 %% State: DISCONNECTED
 handle_event(enter, _OldState, disconnected = State, Data) ->
     ok = log_state_consistency(State, Data),
+    ?tp(resource_disconnected_enter, #{}),
     {keep_state_and_data, retry_actions(Data)};
 handle_event(state_timeout, auto_retry, disconnected, Data) ->
+    ?tp(resource_auto_reconnect, #{}),
     start_resource(Data, undefined);
 %% State: STOPPED
 %% The stopped state is entered after the resource has been explicitly stopped
@@ -450,6 +452,8 @@ retry_actions(Data) ->
     case maps:get(auto_restart_interval, Data#data.opts, ?AUTO_RESTART_INTERVAL) of
         undefined ->
             [];
+        infinity ->
+            [];
         RetryInterval ->
             [{state_timeout, RetryInterval, auto_retry}]
     end.

apps/emqx_resource/src/schema/emqx_resource_schema.erl

@@ -124,6 +124,8 @@ auto_restart_interval(required) -> false;
 auto_restart_interval(validator) -> fun auto_restart_interval_range/1;
 auto_restart_interval(_) -> undefined.
 
+auto_restart_interval_range(infinity) ->
+    ok;
 auto_restart_interval_range(AutoRestartInterval) when
     is_integer(AutoRestartInterval) andalso
         AutoRestartInterval >= ?AUTO_RESTART_INTERVAL_RANGE_MIN andalso

changes/ce/fix-10817.en.md

@@ -0,0 +1 @@
+Fix the error of not being able to configure `auto_restart_interval` as infinity