test: plugin refactor

apps/emqx_plugins/src/emqx_plugins.erl

@@ -18,8 +18,9 @@
 
 -feature(maybe_expr, enable).
 
--include_lib("emqx/include/logger.hrl").
 -include("emqx_plugins.hrl").
+-include_lib("emqx/include/logger.hrl").
+-include_lib("snabbkaffe/include/trace.hrl").
 
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
@@ -281,11 +282,15 @@ ensure_started(NameVsn) ->
 %% @doc Stop all plugins before broker stops.
 -spec ensure_stopped() -> ok.
 ensure_stopped() ->
-    Fun = fun(#{name_vsn := NameVsn, enable := true}) ->
+    Fun = fun
-        case ensure_stopped(NameVsn) of
+        (#{name_vsn := NameVsn, enable := true}) ->
-            ok -> [];
+            case ensure_stopped(NameVsn) of
-            {error, Reason} -> [{NameVsn, Reason}]
+                ok -> [];
-        end
+                {error, Reason} -> [{NameVsn, Reason}]
+            end;
+        (#{name_vsn := NameVsn, enable := false}) ->
+            ?SLOG(debug, #{msg => "plugin_disabled", action => stop_plugin, name_vsn => NameVsn}),
+            []
     end,
     ok = for_plugins(Fun).
 
@@ -732,18 +737,22 @@ do_get_from_cluster(NameVsn) ->
             ok = file:write_file(pkg_file_path(NameVsn), TarContent),
             ok = do_ensure_installed(NameVsn);
         {error, NodeErrors} when Nodes =/= [] ->
-            ?SLOG(error, #{
+            ErrMeta = #{
-                msg => "failed_to_copy_plugin_from_other_nodes",
+                error_msg => "failed_to_copy_plugin_from_other_nodes",
                 name_vsn => NameVsn,
-                node_errors => NodeErrors
+                node_errors => NodeErrors,
-            }),
+                reason => not_found
-            {error, #{reason => not_found, name_vsn => NameVsn}};
+            },
+            ?SLOG(error, ErrMeta),
+            {error, ErrMeta};
         {error, _} ->
-            ?SLOG(error, #{
+            ErrMeta = #{
-                msg => "no_nodes_to_copy_plugin_from",
+                error_msg => "no_nodes_to_copy_plugin_from",
-                name_vsn => NameVsn
+                name_vsn => NameVsn,
-            }),
+                reason => not_found
-            {error, #{reason => not_found, name_vsn => NameVsn}}
+            },
+            ?SLOG(error, ErrMeta),
+            {error, ErrMeta}
     end.
 
 get_from_any_node([], _NameVsn, Errors) ->
@@ -1088,7 +1097,12 @@ for_plugins(ActionFun) ->
         [] ->
             ok;
         Errors ->
-            ?SLOG(error, #{function => ActionFun, errors => Errors}),
+            ErrMeta = #{function => ActionFun, errors => Errors},
+            ?tp(
+                for_plugins_action_error_occurred,
+                ErrMeta
+            ),
+            ?SLOG(error, ErrMeta),
             ok
     end.
 

apps/emqx_plugins/test/emqx_plugins_SUITE.erl

@@ -21,6 +21,7 @@
 
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("common_test/include/ct.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 -define(EMQX_PLUGIN_APP_NAME, my_emqx_plugin).
 -define(EMQX_PLUGIN_TEMPLATE_RELEASE_NAME, atom_to_list(?EMQX_PLUGIN_APP_NAME)).
@@ -273,9 +274,15 @@ t_start_restart_and_stop(Config) ->
     %% fake enable bar-2
     ok = ensure_state(Bar2, rear, true),
     %% should cause an error
-    ?assertError(
+    ?check_trace(
-        #{function := _, errors := [_ | _]},
+        emqx_plugins:ensure_started(),
-        emqx_plugins:ensure_started()
+        fun(Trace) ->
+            ?assertMatch(
+                [#{function := _, errors := [_ | _]}],
+                ?of_kind(for_plugins_action_error_occurred, Trace)
+            ),
+            ok
+        end
     ),
     %% but demo plugin should still be running
     assert_app_running(?EMQX_PLUGIN_APP_NAME, true),
@@ -337,7 +344,7 @@ t_enable_disable({'end', Config}) ->
 t_enable_disable(Config) ->
     NameVsn = proplists:get_value(name_vsn, Config),
     ok = emqx_plugins:ensure_installed(NameVsn),
-    ?assertEqual([], emqx_plugins:configured()),
+    ?assertEqual([#{name_vsn => NameVsn, enable => false}], emqx_plugins:configured()),
     ok = emqx_plugins:ensure_enabled(NameVsn),
     ?assertEqual([#{name_vsn => NameVsn, enable => true}], emqx_plugins:configured()),
     ok = emqx_plugins:ensure_disabled(NameVsn),
@@ -379,9 +386,10 @@ t_bad_tar_gz(Config) ->
         }},
         emqx_plugins:ensure_installed("fake-vsn")
     ),
+    %% the plugin tarball can not be found on any nodes
     ?assertMatch(
         {error, #{
-            error_msg := "failed_to_extract_plugin_package",
+            error_msg := "no_nodes_to_copy_plugin_from",
             reason := not_found
         }},
         emqx_plugins:ensure_installed("nonexisting")
@@ -556,7 +564,7 @@ t_load_config_from_cli({'end', Config}) ->
 t_load_config_from_cli(Config) when is_list(Config) ->
     NameVsn = ?config(name_vsn, Config),
     ok = emqx_plugins:ensure_installed(NameVsn),
-    ?assertEqual([], emqx_plugins:configured()),
+    ?assertEqual([#{name_vsn => NameVsn, enable => false}], emqx_plugins:configured()),
     ok = emqx_plugins:ensure_enabled(NameVsn),
     ok = emqx_plugins:ensure_started(NameVsn),
     Params0 = unused,
@@ -687,6 +695,14 @@ group_t_copy_plugin_to_a_new_node(Config) ->
     %% see: emqx_conf_app:init_conf/0
     ok = rpc:call(CopyToNode, application, stop, [emqx_plugins]),
     {ok, _} = rpc:call(CopyToNode, application, ensure_all_started, [emqx_plugins]),
+
+    %% Plugin config should be synced from `CopyFromNode`
+    %% by application `emqx` and `emqx_conf`
+    %% FIXME: in test case, we manually do it here
+    ok = rpc:call(CopyToNode, emqx_plugins, put_config_internal, [[states], CopyFromPluginsState]),
+    ok = rpc:call(CopyToNode, emqx_plugins, ensure_installed, []),
+    ok = rpc:call(CopyToNode, emqx_plugins, ensure_started, []),
+
     ?assertMatch(
         {ok, #{running_status := running, config_status := enabled}},
         rpc:call(CopyToNode, emqx_plugins, describe, [NameVsn])
@@ -739,6 +755,16 @@ group_t_copy_plugin_to_a_new_node_single_node(Config) ->
     ct:pal("~p install_dir:\n  ~p", [
         CopyToNode, erpc:call(CopyToNode, file, list_dir, [ToInstallDir])
     ]),
+
+    %% Plugin config should be synced from `CopyFromNode`
+    %% by application `emqx` and `emqx_conf`
+    %% FIXME: in test case, we manually do it here
+    ok = rpc:call(CopyToNode, emqx_plugins, put_config_internal, [
+        [states], [#{enable => true, name_vsn => NameVsn}]
+    ]),
+    ok = rpc:call(CopyToNode, emqx_plugins, ensure_installed, []),
+    ok = rpc:call(CopyToNode, emqx_plugins, ensure_started, []),
+
     ?assertMatch(
         {ok, #{running_status := running, config_status := enabled}},
         rpc:call(CopyToNode, emqx_plugins, describe, [NameVsn])
@@ -785,6 +811,11 @@ group_t_cluster_leave(Config) ->
     ok = erpc:call(N1, emqx_plugins, ensure_installed, [NameVsn]),
     ok = erpc:call(N1, emqx_plugins, ensure_started, [NameVsn]),
     ok = erpc:call(N1, emqx_plugins, ensure_enabled, [NameVsn]),
+
+    ok = erpc:call(N2, emqx_plugins, ensure_installed, [NameVsn]),
+    ok = erpc:call(N2, emqx_plugins, ensure_started, [NameVsn]),
+    ok = erpc:call(N2, emqx_plugins, ensure_enabled, [NameVsn]),
+
     Params = unused,
     %% 2 nodes running
     ?assertMatch(