feat(bridges): enable async query mode for all bridges with buffer wo…

…rkers Fixes https://emqx.atlassian.net/browse/EMQX-9130 Since buffer workers always support async calls ("outer calls"), we should decouple those two call modes (inner and outer), and avoid exposing the inner call configuration to user to avoid complexity. For bridges that currently only allow sync query modes, we should allow them to be configured with async. That means basically all bridge types except Kafka Producer.
emqx · Apr 3, 2023 · f3ffc02 · f3ffc02
1 parent 082dfc0
commit f3ffc02
Show file tree

Hide file tree

Showing 21 changed files with 639 additions and 292 deletions.
diff --git a/apps/emqx_connector/src/emqx_connector_mysql.erl b/apps/emqx_connector/src/emqx_connector_mysql.erl
@@ -172,10 +172,15 @@ on_query(
                     %% not return result, next loop will try again
                     on_query(InstId, {TypeOrKey, SQLOrKey, Params, Timeout}, State);
                 {error, Reason} ->
-                    LogMeta = #{connector => InstId, sql => SQLOrKey, state => State},
-                    ?SLOG(
+                    ?tp(
                         error,
-                        LogMeta#{msg => "mysql_connector_do_prepare_failed", reason => Reason}
+                        "mysql_connector_do_prepare_failed",
+                        #{
+                            connector => InstId,
+                            sql => SQLOrKey,
+                            state => State,
+                            reason => Reason
+                        }
                     ),
                     {error, Reason}
             end;
@@ -417,12 +422,10 @@ on_sql_query(
             ),
             do_sql_query(SQLFunc, Conn, SQLOrKey, Params, Timeout, LogMeta);
         {error, disconnected} ->
-            ?SLOG(
+            ?tp(
                 error,
-                LogMeta#{
-                    msg => "mysql_connector_do_sql_query_failed",
-                    reason => worker_is_disconnected
-                }
+                "mysql_connector_do_sql_query_failed",
+                LogMeta#{reason => worker_is_disconnected}
             ),
             {error, {recoverable_error, disconnected}}
     end.

diff --git a/apps/emqx_connector/src/emqx_connector_pgsql.erl b/apps/emqx_connector/src/emqx_connector_pgsql.erl
@@ -44,7 +44,8 @@
     execute_batch/3
 ]).
 
--export([do_get_status/1]).
+%% for ecpool workers usage
+-export([do_get_status/1, prepare_sql_to_conn/2]).
 
 -define(PGSQL_HOST_OPTIONS, #{
     default_port => ?PGSQL_DEFAULT_PORT

diff --git a/apps/emqx_resource/src/schema/emqx_resource_schema.erl b/apps/emqx_resource/src/schema/emqx_resource_schema.erl
@@ -30,18 +30,6 @@ namespace() -> "resource_schema".
 
 roots() -> [].
 
-fields("resource_opts_sync_only") ->
-    [
-        {resource_opts,
-            mk(
-                ref(?MODULE, "creation_opts_sync_only"),
-                resource_opts_meta()
-            )}
-    ];
-fields("creation_opts_sync_only") ->
-    Fields = fields("creation_opts"),
-    QueryMod = {query_mode, fun query_mode_sync_only/1},
-    lists:keyreplace(query_mode, 1, Fields, QueryMod);
 fields("resource_opts") ->
     [
         {resource_opts,
@@ -117,12 +105,6 @@ query_mode(default) -> async;
 query_mode(required) -> false;
 query_mode(_) -> undefined.
 
-query_mode_sync_only(type) -> enum([sync]);
-query_mode_sync_only(desc) -> ?DESC("query_mode_sync_only");
-query_mode_sync_only(default) -> sync;
-query_mode_sync_only(required) -> false;
-query_mode_sync_only(_) -> undefined.
-
 request_timeout(type) -> hoconsc:union([infinity, emqx_schema:duration_ms()]);
 request_timeout(desc) -> ?DESC("request_timeout");
 request_timeout(default) -> <<"15s">>;
@@ -167,7 +149,4 @@ max_queue_bytes(default) -> ?DEFAULT_QUEUE_SIZE_RAW;
 max_queue_bytes(required) -> false;
 max_queue_bytes(_) -> undefined.
 
-desc("creation_opts") ->
-    ?DESC("creation_opts");
-desc("creation_opts_sync_only") ->
-    ?DESC("creation_opts").
+desc("creation_opts") -> ?DESC("creation_opts").
diff --git a/apps/emqx_rule_engine/test/emqx_rule_funcs_SUITE.erl b/apps/emqx_rule_engine/test/emqx_rule_funcs_SUITE.erl
@@ -687,10 +687,6 @@ t_jq(_) ->
                 got_timeout
         end,
     ConfigRootKey = emqx_rule_engine_schema:namespace(),
-    DefaultTimeOut = emqx_config:get([
-        ConfigRootKey,
-        jq_function_default_timeout
-    ]),
     ?assertThrow(
         {jq_exception, {timeout, _}},
         apply_func(jq, [TOProgram, <<"-2">>])

diff --git a/changes/ce/feat-10306.en.md b/changes/ce/feat-10306.en.md
@@ -0,0 +1,3 @@
+Add support for `async` query mode for most bridges.
+
+Before this change, some bridges (Cassandra, MongoDB, MySQL, Postgres, Redis, RocketMQ, TDengine) were only allowed to be created with a `sync` query mode.
diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_cassa.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_cassa.erl
@@ -86,21 +86,10 @@ fields("config") ->
             mk(
                 binary(),
                 #{desc => ?DESC("local_topic"), default => undefined}
-            )},
-        {resource_opts,
-            mk(
-                ref(?MODULE, "creation_opts"),
-                #{
-                    required => false,
-                    default => #{},
-                    desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
-                }
             )}
-    ] ++
+    ] ++ emqx_resource_schema:fields("resource_opts") ++
         (emqx_ee_connector_cassa:fields(config) --
             emqx_connector_schema_lib:prepare_statement_fields());
-fields("creation_opts") ->
-    emqx_resource_schema:fields("creation_opts_sync_only");
 fields("post") ->
     fields("post", cassandra);
 fields("put") ->
@@ -115,8 +104,6 @@ desc("config") ->
     ?DESC("desc_config");
 desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
     ["Configuration for Cassandra using `", string:to_upper(Method), "` method."];
-desc("creation_opts" = Name) ->
-    emqx_resource_schema:desc(Name);
 desc(_) ->
     undefined.
 

diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mongodb.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mongodb.erl
@@ -38,7 +38,7 @@ fields("config") ->
         {enable, mk(boolean(), #{desc => ?DESC("enable"), default => true})},
         {collection, mk(binary(), #{desc => ?DESC("collection"), default => <<"mqtt">>})},
         {payload_template, mk(binary(), #{required => false, desc => ?DESC("payload_template")})}
-    ] ++ emqx_resource_schema:fields("resource_opts_sync_only");
+    ] ++ emqx_resource_schema:fields("resource_opts");
 fields(mongodb_rs) ->
     emqx_connector_mongo:fields(rs) ++ fields("config");
 fields(mongodb_sharded) ->

diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_mysql.erl
@@ -79,21 +79,10 @@ fields("config") ->
             mk(
                 binary(),
                 #{desc => ?DESC("local_topic"), default => undefined}
-            )},
-        {resource_opts,
-            mk(
-                ref(?MODULE, "creation_opts"),
-                #{
-                    required => false,
-                    default => #{},
-                    desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
-                }
             )}
-    ] ++
+    ] ++ emqx_resource_schema:fields("resource_opts") ++
         (emqx_connector_mysql:fields(config) --
             emqx_connector_schema_lib:prepare_statement_fields());
-fields("creation_opts") ->
-    emqx_resource_schema:fields("creation_opts_sync_only");
 fields("post") ->
     [type_field(), name_field() | fields("config")];
 fields("put") ->
@@ -105,8 +94,6 @@ desc("config") ->
     ?DESC("desc_config");
 desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
     ["Configuration for MySQL using `", string:to_upper(Method), "` method."];
-desc("creation_opts" = Name) ->
-    emqx_resource_schema:desc(Name);
 desc(_) ->
     undefined.
 

diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_pgsql.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_pgsql.erl
@@ -81,21 +81,10 @@ fields("config") ->
             mk(
                 binary(),
                 #{desc => ?DESC("local_topic"), default => undefined}
-            )},
-        {resource_opts,
-            mk(
-                ref(?MODULE, "creation_opts"),
-                #{
-                    required => false,
-                    default => #{},
-                    desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
-                }
             )}
-    ] ++
+    ] ++ emqx_resource_schema:fields("resource_opts") ++
         (emqx_connector_pgsql:fields(config) --
             emqx_connector_schema_lib:prepare_statement_fields());
-fields("creation_opts") ->
-    emqx_resource_schema:fields("creation_opts_sync_only");
 fields("post") ->
     fields("post", pgsql);
 fields("put") ->
@@ -110,8 +99,6 @@ desc("config") ->
     ?DESC("desc_config");
 desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
     ["Configuration for PostgreSQL using `", string:to_upper(Method), "` method."];
-desc("creation_opts" = Name) ->
-    emqx_resource_schema:desc(Name);
 desc(_) ->
     undefined.
 

diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_redis.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_redis.erl
@@ -180,10 +180,10 @@ resource_fields(Type) ->
 resource_creation_fields("redis_cluster") ->
     % TODO
     % Cluster bridge is currently incompatible with batching.
-    Fields = emqx_resource_schema:fields("creation_opts_sync_only"),
+    Fields = emqx_resource_schema:fields("creation_opts"),
     lists:foldl(fun proplists:delete/2, Fields, [batch_size, batch_time, enable_batch]);
 resource_creation_fields(_) ->
-    emqx_resource_schema:fields("creation_opts_sync_only").
+    emqx_resource_schema:fields("creation_opts").
 
 desc("config") ->
     ?DESC("desc_config");

diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_rocketmq.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_rocketmq.erl
@@ -80,21 +80,10 @@ fields("config") ->
             mk(
                 binary(),
                 #{desc => ?DESC("local_topic"), required => false}
-            )},
-        {resource_opts,
-            mk(
-                ref(?MODULE, "creation_opts"),
-                #{
-                    required => false,
-                    default => #{<<"request_timeout">> => ?DEFFAULT_REQ_TIMEOUT},
-                    desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
-                }
             )}
-    ] ++
+    ] ++ emqx_resource_schema:fields("resource_opts") ++
         (emqx_ee_connector_rocketmq:fields(config) --
             emqx_connector_schema_lib:prepare_statement_fields());
-fields("creation_opts") ->
-    emqx_resource_schema:fields("creation_opts_sync_only");
 fields("post") ->
     [type_field(), name_field() | fields("config")];
 fields("put") ->
@@ -106,8 +95,6 @@ desc("config") ->
     ?DESC("desc_config");
 desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
     ["Configuration for RocketMQ using `", string:to_upper(Method), "` method."];
-desc("creation_opts" = Name) ->
-    emqx_resource_schema:desc(Name);
 desc(_) ->
     undefined.
 

diff --git a/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_tdengine.erl b/lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_tdengine.erl
@@ -80,19 +80,8 @@ fields("config") ->
             mk(
                 binary(),
                 #{desc => ?DESC("local_topic"), default => undefined}
-            )},
-        {resource_opts,
-            mk(
-                ref(?MODULE, "creation_opts"),
-                #{
-                    required => false,
-                    default => #{},
-                    desc => ?DESC(emqx_resource_schema, <<"resource_opts">>)
-                }
             )}
-    ] ++ emqx_ee_connector_tdengine:fields(config);
-fields("creation_opts") ->
-    emqx_resource_schema:fields("creation_opts_sync_only");
+    ] ++ emqx_resource_schema:fields("resource_opts") ++ emqx_ee_connector_tdengine:fields(config);
 fields("post") ->
     [type_field(), name_field() | fields("config")];
 fields("put") ->
@@ -104,8 +93,6 @@ desc("config") ->
     ?DESC("desc_config");
 desc(Method) when Method =:= "get"; Method =:= "put"; Method =:= "post" ->
     ["Configuration for TDengine using `", string:to_upper(Method), "` method."];
-desc("creation_opts" = Name) ->
-    emqx_resource_schema:desc(Name);
 desc(_) ->
     undefined.