Merge pull request #10270 from kjellwinblad/kjell/clickhouse/on_get_status

fix: on_get_status clickhouse just reporting connecting

changes/ee/fix-10270.en.md

@@ -0,0 +1 @@
+Clickhouse has got a fix that makes the error message better when users click the test button in the settings dialog.

changes/ee/fix-10270.zh.md

@@ -0,0 +1 @@
+Clickhouse 已经修复了一个问题，当用户在设置对话框中点击测试按钮时，错误信息会更清晰。

lib-ee/emqx_ee_connector/i18n/emqx_ee_connector_clickhouse.conf

@@ -12,4 +12,15 @@ emqx_ee_connector_clickhouse {
             }
     }
 
+    connect_timeout {
+        desc {
+          en: "The timeout when connecting to the Clickhouse server."
+          zh: "连接HTTP服务器的超时时间。"
+        }
+        label: {
+              en: "Clickhouse Timeout"
+              zh: "连接超时"
+            }
+    }
+
 }

lib-ee/emqx_ee_connector/rebar.config

@@ -3,7 +3,7 @@
   {hstreamdb_erl, {git, "https://github.com/hstreamdb/hstreamdb_erl.git", {tag, "0.2.5"}}},
   {influxdb, {git, "https://github.com/emqx/influxdb-client-erl", {tag, "1.1.9"}}},
   {tdengine, {git, "https://github.com/emqx/tdengine-client-erl", {tag, "0.1.5"}}},
-  {clickhouse, {git, "https://github.com/emqx/clickhouse-client-erl", {tag, "0.2"}}},
+  {clickhouse, {git, "https://github.com/emqx/clickhouse-client-erl", {tag, "0.3"}}},
   {erlcloud, {git, "https://github.com/emqx/erlcloud.git", {tag,"3.5.16-emqx-1"}}},
   {rocketmq, {git, "https://github.com/emqx/rocketmq-client-erl.git", {tag, "v0.5.1"}}},
   {emqx, {path, "../../apps/emqx"}}

lib-ee/emqx_ee_connector/src/emqx_ee_connector_clickhouse.erl

@@ -53,7 +53,6 @@
 
 %% Internal exports used to execute code with ecpool worker
 -export([
-    check_database_status/1,
     execute_sql_in_clickhouse_server_using_connection/2
 ]).
 
@@ -102,6 +101,14 @@ fields(config) ->
                     end,
                     desc => ?DESC("base_url")
                 }
+            )},
+        {connect_timeout,
+            hoconsc:mk(
+                emqx_schema:duration_ms(),
+                #{
+                    default => <<"15s">>,
+                    desc => ?DESC("connect_timeout")
+                }
             )}
     ] ++ emqx_connector_schema_lib:relational_db_fields().
 
@@ -137,7 +144,8 @@ on_start(
     #{
         url := URL,
         database := DB,
-        pool_size := PoolSize
+        pool_size := PoolSize,
+        connect_timeout := ConnectTimeout
     } = Config
 ) ->
     ?SLOG(info, #{
@@ -155,7 +163,10 @@ on_start(
         {pool_size, PoolSize},
         {pool, PoolName}
     ],
-    InitState = #{poolname => PoolName},
+    InitState = #{
+        poolname => PoolName,
+        connect_timeout => ConnectTimeout
+    },
     try
         Templates = prepare_sql_templates(Config),
         State = maps:merge(InitState, #{templates => Templates}),
@@ -282,18 +293,52 @@ on_stop(ResourceID, #{poolname := PoolName}) ->
 %% on_get_status emqx_resouce callback and related functions
 %% -------------------------------------------------------------------
 
-on_get_status(_ResourceID, #{poolname := Pool} = _State) ->
-    case
-        emqx_plugin_libs_pool:health_check_ecpool_workers(Pool, fun ?MODULE:check_database_status/1)
-    of
-        true ->
-            connected;
-        false ->
-            connecting
+on_get_status(
+    _InstId,
+    #{
+        poolname := PoolName,
+        connect_timeout := Timeout
+    } = State
+) ->
+    case do_get_status(PoolName, Timeout) of
+        ok ->
+            {connected, State};
+        {error, Reason} ->
+            {disconnected, State, Reason}
     end.
 
-check_database_status(Connection) ->
-    clickhouse:status(Connection).
+do_get_status(PoolName, Timeout) ->
+    Workers = [Worker || {_WorkerName, Worker} <- ecpool:workers(PoolName)],
+    DoPerWorker =
+        fun(Worker) ->
+            case ecpool_worker:exec(Worker, fun clickhouse:detailed_status/1, Timeout) of
+                ok ->
+                    ok;
+                {error, Reason} = Error ->
+                    ?SLOG(error, #{
+                        msg => "clickhouse_connector_get_status_failed",
+                        reason => Reason,
+                        worker => Worker
+                    }),
+                    Error
+            end
+        end,
+    try emqx_misc:pmap(DoPerWorker, Workers, Timeout) of
+        Results ->
+            case [E || {error, _} = E <- Results] of
+                [] ->
+                    ok;
+                Errors ->
+                    hd(Errors)
+            end
+    catch
+        exit:timeout ->
+            ?SLOG(error, #{
+                msg => "clickhouse_connector_pmap_failed",
+                reason => timeout
+            }),
+            {error, timeout}
+    end.
 
 %% -------------------------------------------------------------------
 %% on_query emqx_resouce callback and related functions

lib-ee/emqx_ee_connector/test/ee_connector_clickhouse_SUITE.erl

@@ -190,7 +190,8 @@ clickhouse_config() ->
                         ?CLICKHOUSE_DEFAULT_PORT
                     ]
                 )
-            )
+            ),
+            connect_timeout => 10000
         },
     #{<<"config">> => Config}.