fix(webhook): keep `resource_opts.request_timeout` for webhook bridge (r5.0)

Fixes https://emqx.atlassian.net/browse/EMQX-10001

Recently, we unified request_timeout in a single field located at the
webhook connector schema.  However, the correct fix would be to use
the resource_opts.request_timeout one, as that’s the only one that
allows infinity timeout.

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -915,9 +915,6 @@ 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.
 

apps/emqx_bridge/src/emqx_bridge_resource.erl

@@ -311,7 +311,6 @@ parse_confs(
         url := Url,
         method := Method,
         headers := Headers,
-        request_timeout := ReqTimeout,
         max_retries := Retry
     } = Conf
 ) ->
@@ -325,6 +324,10 @@ parse_confs(
                 Reason1 = emqx_utils:readable_error_msg(Reason),
                 invalid_data(<<"Invalid URL: ", Url1/binary, ", details: ", Reason1/binary>>)
         end,
+    RequestTimeout = emqx_utils_maps:deep_get(
+        [resource_opts, request_timeout],
+        Conf
+    ),
     Conf#{
         base_url => BaseUrl1,
         request =>
@@ -333,7 +336,7 @@ parse_confs(
                 method => Method,
                 body => maps:get(body, Conf, undefined),
                 headers => Headers,
-                request_timeout => ReqTimeout,
+                request_timeout => RequestTimeout,
                 max_retries => Retry
             }
     };

apps/emqx_bridge/src/schema/emqx_bridge_schema.erl

@@ -251,25 +251,6 @@ node_name() ->
     {"node", mk(binary(), #{desc => ?DESC("desc_node_name"), example => "emqx@127.0.0.1"})}.
 
 webhook_bridge_converter(Conf0, _HoconOpts) ->
-    Conf1 = emqx_bridge_compatible_config:upgrade_pre_ee(
+    emqx_bridge_compatible_config:upgrade_pre_ee(
         Conf0, fun emqx_bridge_compatible_config:webhook_maybe_upgrade/1
-    ),
-    case Conf1 of
-        undefined ->
-            undefined;
-        _ ->
-            maps:map(
-                fun(_Name, Conf) ->
-                    do_convert_webhook_config(Conf)
-                end,
-                Conf1
-            )
-    end.
-
-%% We hide resource_opts.request_timeout from user.
-do_convert_webhook_config(
-    #{<<"request_timeout">> := ReqT, <<"resource_opts">> := ResOpts} = Conf
-) ->
-    Conf#{<<"resource_opts">> => ResOpts#{<<"request_timeout">> => ReqT}};
-do_convert_webhook_config(Conf) ->
-    Conf.
+    ).

apps/emqx_bridge/src/schema/emqx_bridge_webhook_schema.erl

@@ -40,15 +40,12 @@ fields("put") ->
 fields("get") ->
     emqx_bridge_schema:status_fields() ++ fields("post");
 fields("creation_opts") ->
-    [
-        hidden_request_timeout()
-        | lists:filter(
-            fun({K, _V}) ->
-                not lists:member(K, unsupported_opts())
-            end,
-            emqx_resource_schema:fields("creation_opts")
-        )
-    ].
+    lists:filter(
+        fun({K, _V}) ->
+            not lists:member(K, unsupported_opts())
+        end,
+        emqx_resource_schema:fields("creation_opts")
+    ).
 
 desc("config") ->
     ?DESC("desc_config");
@@ -144,6 +141,7 @@ request_config() ->
                 emqx_schema:duration_ms(),
                 #{
                     default => <<"15s">>,
+                    deprecated => {since, "v5.0.26"},
                     desc => ?DESC("config_request_timeout")
                 }
             )}
@@ -166,8 +164,7 @@ unsupported_opts() ->
     [
         enable_batch,
         batch_size,
-        batch_time,
-        request_timeout
+        batch_time
     ].
 
 %%======================================================================================
@@ -194,13 +191,3 @@ 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

@@ -1390,18 +1390,20 @@ t_inconsistent_webhook_request_timeouts(Config) ->
                 <<"resource_opts">> => #{<<"request_timeout">> => <<"2s">>}
             }
         ),
-    {ok, 201, #{
-        <<"request_timeout">> := <<"1s">>,
-        <<"resource_opts">> := ResourceOpts
-    }} =
+    %% root request_timeout is deprecated for bridge.
+    {ok, 201,
+        #{
+            <<"resource_opts">> := ResourceOpts
+        } = Response} =
         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),
+    ?assertNot(maps:is_key(<<"request_timeout">>, Response)),
+    ?assertMatch(#{<<"request_timeout">> := <<"2s">>}, ResourceOpts),
+    validate_resource_request_timeout(proplists:get_value(group, Config), 2000, Name),
     ok.
 
 t_cluster_later_join_metrics(Config) ->

apps/emqx_bridge/test/emqx_bridge_compatible_config_tests.erl

@@ -65,15 +65,13 @@ webhook_config_test() ->
                 <<"the_name">> :=
                     #{
                         <<"method">> := get,
-                        <<"request_timeout">> := RequestTime,
                         <<"resource_opts">> := ResourceOpts,
                         <<"body">> := <<"${payload}">>
                     }
             }
         }
     } = check(Conf3),
-    ?assertEqual(60_000, RequestTime),
-    ?assertMatch(#{<<"request_timeout">> := 60_000}, ResourceOpts),
+    ?assertMatch(#{<<"request_timeout">> := infinity}, ResourceOpts),
     ok.
 
 up(#{<<"bridges">> := Bridges0} = Conf0) ->
@@ -196,7 +194,7 @@ full_webhook_v5019_hocon() ->
     "      pool_type = \"random\"\n"
     "      request_timeout = \"1m\"\n"
     "      resource_opts = {\n"
-    "        request_timeout = \"7s\"\n"
+    "        request_timeout = \"infinity\"\n"
     "      }\n"
     "      ssl {\n"
     "        ciphers = \"\"\n"

apps/emqx_connector/src/emqx_connector_http.erl

@@ -58,6 +58,7 @@
 ]).
 
 -define(DEFAULT_PIPELINE_SIZE, 100).
+-define(DEFAULT_REQUEST_TIMEOUT_MS, 30_000).
 
 %%=====================================================================
 %% Hocon schema
@@ -477,7 +478,7 @@ preprocess_request(
         path => emqx_plugin_libs_rule:preproc_tmpl(Path),
         body => maybe_preproc_tmpl(body, Req),
         headers => wrap_auth_header(preproc_headers(Headers)),
-        request_timeout => maps:get(request_timeout, Req, 30000),
+        request_timeout => maps:get(request_timeout, Req, ?DEFAULT_REQUEST_TIMEOUT_MS),
         max_retries => maps:get(max_retries, Req, 2)
     }.