feat: implement plugin health check

docker-compose.yml

@@ -1,6 +1,8 @@
 services:
   emqx:
     image: emqx/emqx-enterprise:5.8.5
+    # TODO: uncomment when 5.9.0 image is released
+    # image: docker.io/emqx/emqx-enterprise:5.9.0
     container_name: emqx
     environment:
       EMQX_LOG__CONSOLE__LEVEL: debug

src/emqx_offline_message_plugin_app.erl

@@ -17,7 +17,10 @@
 ]).
 
 %% EMQX Plugin callbacks
--export([on_config_changed/2]).
+-export([
+    on_config_changed/2,
+    on_health_check/1
+]).
 
 start(_StartType, _StartArgs) ->
     {ok, Sup} = emqx_omp_sup:start_link(),
@@ -28,3 +31,6 @@ stop(_State) ->
 
 on_config_changed(OldConf, NewConf) ->
     emqx_omp:on_config_changed(OldConf, NewConf).
+
+on_health_check(_Options) ->
+    emqx_omp:on_health_check().

src/emqx_omp.erl

@@ -19,7 +19,8 @@
 
 %% Plugin callbacks
 -export([
-    on_config_changed/2
+    on_config_changed/2,
+    on_health_check/0
 ]).
 
 %% gen_server callbacks
@@ -39,11 +40,14 @@
 %%--------------------------------------------------------------------
 
 -record(state, {}).
+
 -record(on_config_changed, {
     old_conf :: map(),
     new_conf :: map()
 }).
 
+-record(on_health_check, {}).
+
 %%--------------------------------------------------------------------
 %% API
 %%--------------------------------------------------------------------
@@ -68,7 +72,20 @@ child_spec() ->
 %%--------------------------------------------------------------------
 
 on_config_changed(OldConf, NewConf) ->
-    gen_server:call(?SERVER, #on_config_changed{old_conf = OldConf, new_conf = NewConf}, ?TIMEOUT).
+    try
+        gen_server:call(?SERVER, #on_config_changed{old_conf = OldConf, new_conf = NewConf}, ?TIMEOUT)
+    catch
+        exit:{noproc, _} ->
+            ok
+    end.
+
+on_health_check() ->
+    try
+        gen_server:call(?SERVER, #on_health_check{}, ?TIMEOUT)
+    catch
+        exit:{noproc, _} ->
+            {error, <<"Plugin is not running">>}
+    end.
 
 %%--------------------------------------------------------------------
 %% gen_server callbacks
@@ -83,6 +100,8 @@ init([]) ->
 
 handle_call(#on_config_changed{old_conf = OldConf, new_conf = NewConf}, _From, State) ->
     {reply, handle_on_config_changed(OldConf, NewConf), State};
+handle_call(#on_health_check{}, _From, State) ->
+    {reply, handle_on_health_check(), State};
 handle_call(Request, From, State) ->
     ?SLOG(error, #{
         msg => "offline_message_plugin_unexpected_call", request => Request, from => From
@@ -125,9 +144,32 @@ handle_on_config_changed(OldConf, NewConf) ->
     ok = emqx_omp_redis:on_config_changed(OldRedisConf, NewRedisConf),
     ok.
 
+handle_on_health_check() ->
+    Config = current_config(),
+    MysqlConf = maps:get(<<"mysql">>, Config, #{}),
+    RedisConf = maps:get(<<"redis">>, Config, #{}),
+    MysqlStatus = emqx_omp_mysql:on_health_check(MysqlConf),
+    RedisStatus = emqx_omp_redis:on_health_check(RedisConf),
+    Errors = status_to_error_list(MysqlStatus) ++ status_to_error_list(RedisStatus),
+    case Errors of
+        [] ->
+            ok;
+        Errors ->
+            {error, iolist_to_binary(lists:join(",", Errors))}
+    end.
+
+status_to_error_list(ok) -> [];
+status_to_error_list({error, Error}) -> [Error].
+
 current_config() ->
-    {ok, Config} = emqx_plugins:get_config(?PLUGIN_NAME_VSN),
-    Config.
+    case emqx_plugins:get_config(?PLUGIN_NAME_VSN) of
+        %% Pre 5.9.0
+        {ok, Config} when is_map(Config) ->
+            Config;
+        %% 5.9.0 and later
+        Config when is_map(Config) ->
+            Config
+    end.
 
 init_metrics() ->
     ?SLOG(info, #{msg => "omp_init_metrics"}),

src/emqx_omp_mysql.erl

@@ -11,7 +11,8 @@
 -include("emqx_omp.hrl").
 
 -export([
-    on_config_changed/2
+    on_config_changed/2,
+    on_health_check/1
 ]).
 
 -export([
@@ -78,6 +79,12 @@ on_config_changed(#{<<"enable">> := true} = _OldConf, #{<<"enable">> := false} =
 on_config_changed(#{<<"enable">> := false} = _OldConf, #{<<"enable">> := true} = NewConf) ->
     ok = start(NewConf).
 
+-spec on_health_check(map()) -> ok | {error, binary()}.
+on_health_check(#{<<"enable">> := false}) ->
+    ok;
+on_health_check(#{<<"enable">> := true}) ->
+    emqx_omp_utils:resource_health_status(<<"MySQL">>, ?RESOURCE_ID).
+
 %%-------------------------------------------------------------------
 %% start/stop
 %%-------------------------------------------------------------------

src/emqx_omp_redis.erl

@@ -11,7 +11,8 @@
 -include("emqx_omp.hrl").
 
 -export([
-    on_config_changed/2
+    on_config_changed/2,
+    on_health_check/1
 ]).
 
 -export([
@@ -49,6 +50,12 @@ on_config_changed(#{<<"enable">> := true} = _OldConf, #{<<"enable">> := false} =
 on_config_changed(#{<<"enable">> := false} = _OldConf, #{<<"enable">> := true} = NewConf) ->
     ok = start(NewConf).
 
+-spec on_health_check(map()) -> ok | {error, binary()}.
+on_health_check(#{<<"enable">> := false}) ->
+    ok;
+on_health_check(#{<<"enable">> := true}) ->
+    emqx_omp_utils:resource_health_status(<<"Redis">>, ?RESOURCE_ID).
+
 %%--------------------------------------------------------------------
 %% start/stop
 %%--------------------------------------------------------------------

src/emqx_omp_utils.erl

@@ -13,7 +13,8 @@
     deliver_messages/2,
     induce_subscriptions/1,
     need_persist_message/2,
-    topic_filters/1
+    topic_filters/1,
+    resource_health_status/2
 ]).
 
 fix_ssl_config(#{<<"ssl">> := SslConfig0} = RawConfig) ->
@@ -90,3 +91,13 @@ is_message_qos_nonzero(Message) ->
 does_message_topic_match(Message, TopicFilters) ->
     Topic = emqx_message:topic(Message),
     lists:any(fun(Filter) -> emqx_topic:match(Topic, Filter) end, TopicFilters).
+
+resource_health_status(Name, ResourceId) ->
+    case emqx_resource:health_check(ResourceId) of
+        {ok, connected} ->
+            ok;
+        {ok, OtherStatus} ->
+            {error, iolist_to_binary(io_lib:format("Resource ~s is not connected, status: ~p", [Name, OtherStatus]))};
+        {error, Reason} ->
+            {error, iolist_to_binary(io_lib:format("Resource ~s health check failed: ~p", [Name, Reason]))}
+    end.

test/emqx_omp_SUITE.erl

@@ -33,14 +33,17 @@ groups() ->
         [mysql_tcp, mysql_ssl, redis_tcp, redis_ssl],
         [sync, async],
         [buffered, unbuffered],
-        emqx_omp_test_helpers:all(?MODULE)
+        %% TODO: restore t_health_check when 5.9.0 image is released
+        %% and used in the Docker Compose file
+        emqx_omp_test_helpers:all(?MODULE) -- [t_health_check]
     ]).
 
 init_per_suite(Config) ->
     ok = emqx_omp_test_helpers:start(),
 
     %% clean up
     ok = emqx_omp_test_api_helpers:delete_all_plugins(),
+    ok = emqx_omp_test_helpers:allow_plugin_install(),
 
     %% install plugin
     {PluginId, Filename} = emqx_omp_test_api_helpers:find_plugin(),
@@ -211,6 +214,25 @@ t_subscribition_persistence(_Config) ->
     ok = emqtt:stop(ClientPub),
     ok = emqtt:stop(ClientSub2).
 
+t_health_check(Config) ->
+    PluginId = ?config(plugin_id, Config),
+    ?assertMatch(
+        #{
+            <<"running_status">> :=
+                [#{<<"health_status">> := #{<<"status">> := <<"ok">>}}]
+        },
+        emqx_omp_test_api_helpers:get_plugin(PluginId)
+    ),
+    Config0 = ?config(plugin_config, Config),
+    Config1 = emqx_utils_maps:deep_put([mysql, server], Config0, <<"bad-host:3306">>),
+    Config2 = emqx_utils_maps:deep_put([redis, servers], Config1, <<"bad-host:6379">>),
+    ok = emqx_omp_test_api_helpers:configure_plugin(PluginId, Config2),
+    ?assertMatch(
+        #{<<"running_status">> := [#{<<"health_status">> := #{<<"status">> := <<"error">>}}]},
+        emqx_omp_test_api_helpers:get_plugin(PluginId)
+    ),
+    ok.
+
 t_message_order(_Config) ->
     Topic = unique_topic(),
 

test/emqx_omp_test_api_helpers.erl

@@ -17,6 +17,7 @@
     find_plugin/0,
     upload_plugin/1,
     list_plugins/0,
+    get_plugin/1,
     start_plugin/1,
     stop_plugin/1,
     delete_plugin/1,
@@ -106,6 +107,14 @@ list_plugins() ->
             error(Error)
     end.
 
+get_plugin(PluginId) ->
+    case emqx_omp_test_helpers:api_get({plugins, PluginId}) of
+        {ok, Plugin} ->
+            Plugin;
+        {error, Error} ->
+            error(Error)
+    end.
+
 plugin_id(#{<<"name">> := Name, <<"rel_vsn">> := RelVsn}) ->
     <<Name/binary, "-", RelVsn/binary>>.
 

test/emqx_omp_test_helpers.erl

@@ -7,6 +7,8 @@
 -compile(nowarn_export_all).
 -compile(export_all).
 
+-include("emqx_omp.hrl").
+
 all(Suite) ->
     lists:usort([
         F
@@ -33,6 +35,13 @@ start() ->
 stop() ->
     ok.
 
+allow_plugin_install() ->
+    Command = "docker compose exec emqx /opt/emqx/bin/emqx ctl plugins allow " ++ binary_to_list(?PLUGIN_NAME_VSN),
+    ct:print("Command: ~s~n", [Command]),
+    os:cmd(Command),
+    timer:sleep(1000),
+    ok.
+
 api_get(Path) ->
     Result = make_request({get, Path}),
     handle_result(Result).
@@ -94,7 +103,9 @@ handle_result({error, Reason}) ->
 
 handle_result_raw({ok, Code, _Headers, ClientRef}) when Code >= 200 andalso Code < 300 ->
     hackney:body(ClientRef);
-handle_result_raw({ok, Code, _Headers, _Body}) ->
+handle_result_raw({ok, Code, _Headers, ClientRef}) ->
+    {ok, Body} = hackney:body(ClientRef),
+    ct:pal("Response body:~n~s~n", [Body]),
     {error, {http_status, Code}};
 handle_result_raw({error, Reason}) ->
     {error, Reason}.