Merge pull request #10713 from zhongwencool/put-webhook-request-timeo…

…ut-into-resource-opts

feat: update wehbook's request_timeout into resource_opts

apps/emqx_bridge/src/emqx_bridge.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_bridge, [
     {description, "EMQX bridges"},
-    {vsn, "0.1.18"},
+    {vsn, "0.1.19"},
     {registered, [emqx_bridge_sup]},
     {mod, {emqx_bridge_app, []}},
     {applications, [

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -892,10 +892,17 @@ fill_defaults(Type, RawConf) ->
 pack_bridge_conf(Type, RawConf) ->
     #{<<"bridges">> => #{bin(Type) => #{<<"foo">> => RawConf}}}.
 
+%% Hide webhook's resource_opts.request_timeout from user.
+filter_raw_conf(<<"webhook">>, RawConf0) ->
+    emqx_utils_maps:deep_remove([<<"resource_opts">>, <<"request_timeout">>], RawConf0);
+filter_raw_conf(_TypeBin, RawConf) ->
+    RawConf.
+
 unpack_bridge_conf(Type, PackedConf) ->
+    TypeBin = bin(Type),
     #{<<"bridges">> := Bridges} = PackedConf,
-    #{<<"foo">> := RawConf} = maps:get(bin(Type), Bridges),
-    RawConf.
+    #{<<"foo">> := RawConf} = maps:get(TypeBin, Bridges),
+    filter_raw_conf(TypeBin, RawConf).
 
 is_ok(ok) ->
     ok;

apps/emqx_bridge/src/emqx_bridge_resource.erl

@@ -165,20 +165,20 @@ create(BridgeId, Conf) ->
 create(Type, Name, Conf) ->
     create(Type, Name, Conf, #{}).
 
-create(Type, Name, Conf, Opts0) ->
+create(Type, Name, Conf, Opts) ->
     ?SLOG(info, #{
         msg => "create bridge",
         type => Type,
         name => Name,
         config => emqx_utils:redact(Conf)
     }),
-    Opts = override_start_after_created(Conf, Opts0),
+    TypeBin = bin(Type),
     {ok, _Data} = emqx_resource:create_local(
         resource_id(Type, Name),
         <<"emqx_bridge">>,
         bridge_to_resource_type(Type),
-        parse_confs(bin(Type), Name, Conf),
-        Opts
+        parse_confs(TypeBin, Name, Conf),
+        parse_opts(Conf, Opts)
     ),
     ok.
 
@@ -189,7 +189,7 @@ update(BridgeId, {OldConf, Conf}) ->
 update(Type, Name, {OldConf, Conf}) ->
     update(Type, Name, {OldConf, Conf}, #{}).
 
-update(Type, Name, {OldConf, Conf}, Opts0) ->
+update(Type, Name, {OldConf, Conf}, Opts) ->
     %% TODO: sometimes its not necessary to restart the bridge connection.
     %%
     %% - if the connection related configs like `servers` is updated, we should restart/start
@@ -198,7 +198,6 @@ update(Type, Name, {OldConf, Conf}, Opts0) ->
     %% the `method` or `headers` of a WebHook is changed, then the bridge can be updated
     %% without restarting the bridge.
     %%
-    Opts = override_start_after_created(Conf, Opts0),
     case emqx_utils_maps:if_only_to_toggle_enable(OldConf, Conf) of
         false ->
             ?SLOG(info, #{
@@ -241,11 +240,12 @@ recreate(Type, Name, Conf) ->
     recreate(Type, Name, Conf, #{}).
 
 recreate(Type, Name, Conf, Opts) ->
+    TypeBin = bin(Type),
     emqx_resource:recreate_local(
         resource_id(Type, Name),
         bridge_to_resource_type(Type),
-        parse_confs(bin(Type), Name, Conf),
-        Opts
+        parse_confs(TypeBin, Name, Conf),
+        parse_opts(Conf, Opts)
     ).
 
 create_dry_run(Type, Conf0) ->
@@ -402,6 +402,9 @@ bin(Bin) when is_binary(Bin) -> Bin;
 bin(Str) when is_list(Str) -> list_to_binary(Str);
 bin(Atom) when is_atom(Atom) -> atom_to_binary(Atom, utf8).
 
+parse_opts(Conf, Opts0) ->
+    override_start_after_created(Conf, Opts0).
+
 override_start_after_created(Config, Opts) ->
     Enabled = maps:get(enable, Config, true),
     StartAfterCreated = Enabled andalso maps:get(start_after_created, Opts, Enabled),

apps/emqx_bridge/src/schema/emqx_bridge_schema.erl

@@ -238,36 +238,10 @@ webhook_bridge_converter(Conf0, _HoconOpts) ->
             )
     end.
 
+%% We hide resource_opts.request_timeout from user.
 do_convert_webhook_config(
-    #{<<"request_timeout">> := ReqT, <<"resource_opts">> := #{<<"request_timeout">> := ReqT}} = Conf
+    #{<<"request_timeout">> := ReqT, <<"resource_opts">> := ResOpts} = Conf
 ) ->
-    %% ok: same values
-    Conf;
-do_convert_webhook_config(
-    #{
-        <<"request_timeout">> := ReqTRootRaw,
-        <<"resource_opts">> := #{<<"request_timeout">> := ReqTResourceRaw}
-    } = Conf0
-) ->
-    %% different values; we set them to the same, if they are valid
-    %% durations
-    MReqTRoot = emqx_schema:to_duration_ms(ReqTRootRaw),
-    MReqTResource = emqx_schema:to_duration_ms(ReqTResourceRaw),
-    case {MReqTRoot, MReqTResource} of
-        {{ok, ReqTRoot}, {ok, ReqTResource}} ->
-            {_Parsed, ReqTRaw} = max({ReqTRoot, ReqTRootRaw}, {ReqTResource, ReqTResourceRaw}),
-            Conf1 = emqx_utils_maps:deep_merge(
-                Conf0,
-                #{
-                    <<"request_timeout">> => ReqTRaw,
-                    <<"resource_opts">> => #{<<"request_timeout">> => ReqTRaw}
-                }
-            ),
-            Conf1;
-        _ ->
-            %% invalid values; let the type checker complain about
-            %% that.
-            Conf0
-    end;
+    Conf#{<<"resource_opts">> => ResOpts#{<<"request_timeout">> => ReqT}};
 do_convert_webhook_config(Conf) ->
     Conf.

apps/emqx_bridge/src/schema/emqx_bridge_webhook_schema.erl

@@ -40,12 +40,15 @@ fields("put") ->
 fields("get") ->
     emqx_bridge_schema:status_fields() ++ fields("post");
 fields("creation_opts") ->
-    lists:filter(
-        fun({K, _V}) ->
-            not lists:member(K, unsupported_opts())
-        end,
-        emqx_resource_schema:fields("creation_opts")
-    ).
+    [
+        hidden_request_timeout()
+        | lists:filter(
+            fun({K, _V}) ->
+                not lists:member(K, unsupported_opts())
+            end,
+            emqx_resource_schema:fields("creation_opts")
+        )
+    ].
 
 desc("config") ->
     ?DESC("desc_config");
@@ -163,7 +166,8 @@ unsupported_opts() ->
     [
         enable_batch,
         batch_size,
-        batch_time
+        batch_time,
+        request_timeout
     ].
 
 %%======================================================================================
@@ -190,3 +194,13 @@ name_field() ->
 
 method() ->
     enum([post, put, get, delete]).
+
+hidden_request_timeout() ->
+    {request_timeout,
+        mk(
+            hoconsc:union([infinity, emqx_schema:duration_ms()]),
+            #{
+                required => false,
+                importance => ?IMPORTANCE_HIDDEN
+            }
+        )}.

apps/emqx_bridge/test/emqx_bridge_api_SUITE.erl

@@ -1284,21 +1284,43 @@ t_inconsistent_webhook_request_timeouts(Config) ->
                 <<"resource_opts">> => #{<<"request_timeout">> => <<"2s">>}
             }
         ),
-    ?assertMatch(
-        {ok, 201, #{
-            %% note: same value on both fields
-            <<"request_timeout">> := <<"2s">>,
-            <<"resource_opts">> := #{<<"request_timeout">> := <<"2s">>}
-        }},
+    {ok, 201, #{
+        <<"request_timeout">> := <<"1s">>,
+        <<"resource_opts">> := ResourceOpts
+    }} =
         request_json(
             post,
             uri(["bridges"]),
             BadBridgeParams,
             Config
-        )
-    ),
+        ),
+    ?assertNot(maps:is_key(<<"request_timeout">>, ResourceOpts)),
+    validate_resource_request_timeout(proplists:get_value(group, Config), 1000, Name),
     ok.
 
+validate_resource_request_timeout(single, Timeout, Name) ->
+    SentData = #{payload => <<"Hello EMQX">>, timestamp => 1668602148000},
+    BridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE_HTTP, Name),
+    ResId = emqx_bridge_resource:resource_id(<<"webhook">>, Name),
+    ?check_trace(
+        begin
+            {ok, Res} =
+                ?wait_async_action(
+                    emqx_bridge:send_message(BridgeID, SentData),
+                    #{?snk_kind := async_query},
+                    1000
+                ),
+            ?assertMatch({ok, #{id := ResId, query_opts := #{timeout := Timeout}}}, Res)
+        end,
+        fun(Trace0) ->
+            Trace = ?of_kind(async_query, Trace0),
+            ?assertMatch([#{query_opts := #{timeout := Timeout}}], Trace),
+            ok
+        end
+    );
+validate_resource_request_timeout(_Cluster, _Timeout, _Name) ->
+    ignore.
+
 %%
 
 request(Method, URL, Config) ->

apps/emqx_bridge/test/emqx_bridge_compatible_config_tests.erl

@@ -59,27 +59,21 @@ webhook_config_test() ->
         },
         check(Conf2)
     ),
-
-    %% the converter should pick the greater of the two
-    %% request_timeouts and place them in the root and inside
-    %% resource_opts.
-    ?assertMatch(
-        #{
-            <<"bridges">> := #{
-                <<"webhook">> := #{
-                    <<"the_name">> :=
-                        #{
-                            <<"method">> := get,
-                            <<"request_timeout">> := 60_000,
-                            <<"resource_opts">> := #{<<"request_timeout">> := 60_000},
-                            <<"body">> := <<"${payload}">>
-                        }
-                }
+    #{
+        <<"bridges">> := #{
+            <<"webhook">> := #{
+                <<"the_name">> :=
+                    #{
+                        <<"method">> := get,
+                        <<"request_timeout">> := RequestTime,
+                        <<"resource_opts">> := ResourceOpts,
+                        <<"body">> := <<"${payload}">>
+                    }
             }
-        },
-        check(Conf3)
-    ),
-
+        }
+    } = check(Conf3),
+    ?assertEqual(60_000, RequestTime),
+    ?assertMatch(#{<<"request_timeout">> := 60_000}, ResourceOpts),
     ok.
 
 up(#{<<"bridges">> := Bridges0} = Conf0) ->

changes/ce/feat-10713.en.md

@@ -0,0 +1,3 @@
+We hide the request_timeout in resource_option of the webhook to keep it consistent with the http request_timeout of the webhook.
+From now on, when configuring a webhook through API or configuration files,
+it is no longer necessary to configure the request_timeout of the resource. Only configuring the http request_timeout is sufficient, and the request_timeout in the resource will automatically be consistent with the http request_timeout.