Merge pull request #13305 from kjellwinblad/kjell/redis_conn_always_time_out_no_username_password/EMQX-12557

fix: redis connector should not timeout because no username and password

apps/emqx_bridge/test/emqx_bridge_v2_testlib.erl

@@ -945,6 +945,7 @@ t_on_get_status(Config, Opts) ->
     ProxyHost = ?config(proxy_host, Config),
     ProxyName = ?config(proxy_name, Config),
     FailureStatus = maps:get(failure_status, Opts, disconnected),
+    NormalStatus = maps:get(normal_status, Opts, connected),
     ?assertMatch({ok, _}, create_bridge(Config)),
     ResourceId = resource_id(Config),
     %% Since the connection process is async, we give it some time to
@@ -952,7 +953,7 @@ t_on_get_status(Config, Opts) ->
     ?retry(
         _Sleep = 1_000,
         _Attempts = 20,
-        ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+        ?assertEqual({ok, NormalStatus}, emqx_resource_manager:health_check(ResourceId))
     ),
     case ProxyHost of
         undefined ->
@@ -971,7 +972,7 @@ t_on_get_status(Config, Opts) ->
             ?retry(
                 _Sleep = 1_000,
                 _Attempts = 20,
-                ?assertEqual({ok, connected}, emqx_resource_manager:health_check(ResourceId))
+                ?assertEqual({ok, NormalStatus}, emqx_resource_manager:health_check(ResourceId))
             )
     end,
     ok.

apps/emqx_bridge_redis/test/emqx_bridge_v2_redis_SUITE.erl

@@ -19,6 +19,7 @@
 
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 -define(BRIDGE_TYPE, redis).
 -define(BRIDGE_TYPE_BIN, <<"redis">>).
@@ -46,6 +47,7 @@ matrix_testcases() ->
         t_start_stop,
         t_create_via_http,
         t_on_get_status,
+        t_on_get_status_no_username_pass,
         t_sync_query,
         t_map_to_redis_hset_args
     ].
@@ -325,6 +327,43 @@ t_on_get_status(Config) when is_list(Config) ->
     emqx_bridge_v2_testlib:t_on_get_status(Config, #{failure_status => connecting}),
     ok.
 
+t_on_get_status_no_username_pass(matrix) ->
+    {on_get_status, [
+        [single, tcp],
+        [cluster, tcp],
+        [sentinel, tcp]
+    ]};
+t_on_get_status_no_username_pass(Config0) when is_list(Config0) ->
+    ConnectorConfig0 = ?config(connector_config, Config0),
+    ConnectorConfig1 = emqx_utils_maps:deep_put(
+        [<<"parameters">>, <<"password">>], ConnectorConfig0, <<"">>
+    ),
+    ConnectorConfig2 = emqx_utils_maps:deep_put(
+        [<<"parameters">>, <<"username">>], ConnectorConfig1, <<"">>
+    ),
+    Config1 = proplists:delete(connector_config, Config0),
+    Config2 = [{connector_config, ConnectorConfig2} | Config1],
+    ?check_trace(
+        emqx_bridge_v2_testlib:t_on_get_status(
+            Config2,
+            #{
+                failure_status => disconnected,
+                normal_status => disconnected
+            }
+        ),
+        fun(ok, Trace) ->
+            case ?config(redis_type, Config2) of
+                single ->
+                    ?assertMatch([_ | _], ?of_kind(emqx_redis_auth_required_error, Trace));
+                sentinel ->
+                    ?assertMatch([_ | _], ?of_kind(emqx_redis_auth_required_error, Trace));
+                cluster ->
+                    ok
+            end
+        end
+    ),
+    ok.
+
 t_sync_query(matrix) ->
     {sync_query, [
         [single, tcp],

apps/emqx_redis/src/emqx_redis.app.src

@@ -1,6 +1,6 @@
 {application, emqx_redis, [
     {description, "EMQX Redis Database Connector"},
-    {vsn, "0.1.5"},
+    {vsn, "0.1.6"},
     {registered, []},
     {applications, [
         kernel,

apps/emqx_redis/src/emqx_redis.erl

@@ -19,6 +19,8 @@
 -include_lib("typerefl/include/types.hrl").
 -include_lib("hocon/include/hoconsc.hrl").
 -include_lib("emqx/include/logger.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
+-include_lib("emqx_resource/include/emqx_resource.hrl").
 
 -export([namespace/0, roots/0, fields/1, redis_fields/0, desc/1]).
 
@@ -231,7 +233,7 @@ is_unrecoverable_error({error, invalid_cluster_command}) ->
 is_unrecoverable_error(_) ->
     false.
 
-on_get_status(_InstId, #{type := cluster, pool_name := PoolName}) ->
+on_get_status(_InstId, #{type := cluster, pool_name := PoolName} = State) ->
     case eredis_cluster:pool_exists(PoolName) of
         true ->
             %% eredis_cluster has null slot even pool_exists when emqx start before redis cluster.
@@ -240,26 +242,51 @@ on_get_status(_InstId, #{type := cluster, pool_name := PoolName}) ->
             %% In this case, we can directly consider it as a disconnect and then proceed to reconnect.
             case eredis_cluster_monitor:get_all_pools(PoolName) of
                 [] ->
-                    disconnected;
+                    ?status_disconnected;
                 [_ | _] ->
-                    Health = eredis_cluster:ping_all(PoolName),
-                    status_result(Health)
+                    do_cluster_status_check(PoolName, State)
             end;
         false ->
-            disconnected
+            ?status_disconnected
     end;
-on_get_status(_InstId, #{pool_name := PoolName}) ->
-    Health = emqx_resource_pool:health_check_workers(PoolName, fun ?MODULE:do_get_status/1),
-    status_result(Health).
-
-do_get_status(Conn) ->
-    case eredis:q(Conn, ["PING"]) of
-        {ok, _} -> true;
-        _ -> false
+on_get_status(_InstId, #{pool_name := PoolName} = State) ->
+    HealthCheckResoults = emqx_resource_pool:health_check_workers(
+        PoolName,
+        fun ?MODULE:do_get_status/1,
+        emqx_resource_pool:health_check_timeout(),
+        #{return_values => true}
+    ),
+    case HealthCheckResoults of
+        {ok, Results} ->
+            sum_worker_results(Results, State);
+        Error ->
+            {?status_disconnected, State, Error}
     end.
 
-status_result(_Status = true) -> connected;
-status_result(_Status = false) -> connecting.
+do_cluster_status_check(Pool, State) ->
+    Pongs = eredis_cluster:qa(Pool, [<<"PING">>]),
+    sum_worker_results(Pongs, State).
+
+do_get_status(Conn) ->
+    eredis:q(Conn, ["PING"]).
+
+sum_worker_results([], _State) ->
+    ?status_connected;
+sum_worker_results([{error, <<"NOAUTH Authentication required.">>} = Error | _Rest], State) ->
+    ?tp(emqx_redis_auth_required_error, #{}),
+    %% This requires user action to fix so we set the status to disconnected
+    {?status_disconnected, State, {unhealthy_target, Error}};
+sum_worker_results([{ok, _} | Rest], State) ->
+    sum_worker_results(Rest, State);
+sum_worker_results([Error | _Rest], State) ->
+    ?SLOG(
+        warning,
+        #{
+            msg => "emqx_redis_check_status_error",
+            error => Error
+        }
+    ),
+    {?status_connecting, State, Error}.
 
 do_cmd(PoolName, cluster, {cmd, Command}) ->
     eredis_cluster:q(PoolName, Command);

changes/ee/fix-13305.en.md

@@ -0,0 +1 @@
+Improved error handling for Redis connectors. Previously, Redis connectors of type single or sentinel would always encounter a timeout error during the connector test in the dashboard if no username or password was provided. This update ensures that users now receive an informative error message in such scenarios. Additionally, more detailed error information has been added for all Redis connector types to enhance diagnostics and troubleshooting.