Merge pull request #11514 from zhongwencool/stop-load-application-before-reboot

fix: stop otel deps appication before reboot

apps/emqx_machine/src/emqx_machine_boot.erl

@@ -62,6 +62,8 @@ stop_apps() ->
     ?SLOG(notice, #{msg => "stopping_emqx_apps"}),
     _ = emqx_alarm_handler:unload(),
     ok = emqx_conf_app:unset_config_loaded(),
+    %% Mute otel deps application.
+    _ = emqx_otel:stop_otel(),
     lists:foreach(fun stop_one_app/1, lists:reverse(sorted_reboot_apps())).
 
 %% Those port apps are terminated after the main apps

apps/emqx_machine/test/emqx_machine_SUITE.erl

@@ -42,7 +42,7 @@ init_per_suite(Config) ->
     %%   Unload emqx_authz to avoid reboot this application
     %%
     application:unload(emqx_authz),
-    emqx_common_test_helpers:start_apps([emqx_conf]),
+    emqx_common_test_helpers:start_apps([emqx_conf, emqx_opentelemetry]),
     application:set_env(emqx_machine, applications, [
         emqx_prometheus,
         emqx_modules,
@@ -56,12 +56,13 @@ init_per_suite(Config) ->
         emqx_exhook,
         emqx_authn,
         emqx_authz,
-        emqx_plugin
+        emqx_plugin,
+        emqx_opentelemetry
     ]),
     Config.
 
 end_per_suite(_Config) ->
-    emqx_common_test_helpers:stop_apps([]).
+    emqx_common_test_helpers:stop_apps([emqx_opentelemetry, emqx_conf]).
 
 init_per_testcase(t_custom_shard_transports, Config) ->
     OldConfig = application:get_env(emqx_machine, custom_shard_transports),

apps/emqx_opentelemetry/src/emqx_otel.erl

@@ -17,10 +17,30 @@
 -module(emqx_otel).
 -include_lib("emqx/include/logger.hrl").
 
--export([start_link/1, cleanup/0]).
+-export([start_otel/1, stop_otel/0]).
 -export([get_cluster_gauge/1, get_stats_gauge/1, get_vm_gauge/1, get_metric_counter/1]).
+-export([start_link/1]).
 -export([init/1, handle_continue/2, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
 
+-define(SUPERVISOR, emqx_otel_sup).
+
+start_otel(Conf) ->
+    Spec = emqx_otel_sup:worker_spec(?MODULE, Conf),
+    assert_started(supervisor:start_child(?SUPERVISOR, Spec)).
+
+stop_otel() ->
+    ok = cleanup(),
+    case erlang:whereis(?SUPERVISOR) of
+        undefined ->
+            ok;
+        Pid ->
+            case supervisor:terminate_child(Pid, ?MODULE) of
+                ok -> supervisor:delete_child(Pid, ?MODULE);
+                {error, not_found} -> ok;
+                Error -> Error
+            end
+    end.
+
 start_link(Conf) ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, Conf, []).
 
@@ -47,7 +67,7 @@ setup(Conf = #{enable := true}) ->
     ensure_apps(Conf),
     create_metric_views();
 setup(_Conf) ->
-    cleanup(),
+    ok = cleanup(),
     ok.
 
 ensure_apps(Conf) ->
@@ -225,3 +245,8 @@ create_counter(Meter, Counters, CallBack) ->
 
 normalize_name(Name) ->
     list_to_existing_atom(lists:flatten(string:replace(atom_to_list(Name), "_", ".", all))).
+
+assert_started({ok, _Pid}) -> ok;
+assert_started({ok, _Pid, _Info}) -> ok;
+assert_started({error, {already_started, _Pid}}) -> ok;
+assert_started({error, Reason}) -> {error, Reason}.

apps/emqx_opentelemetry/src/emqx_otel_config.erl

@@ -52,7 +52,7 @@ post_config_update(_ConfPath, _Req, _NewConf, _OldConf, _AppEnvs) ->
     ok.
 
 ensure_otel(#{enable := true} = Conf) ->
-    _ = emqx_otel_sup:stop_otel(),
+    _ = emqx_otel:stop_otel(),
-    emqx_otel_sup:start_otel(Conf);
+    emqx_otel:start_otel(Conf);
 ensure_otel(#{enable := false}) ->
-    emqx_otel_sup:stop_otel().
+    emqx_otel:stop_otel().

apps/emqx_opentelemetry/src/emqx_otel_sup.erl

@@ -19,36 +19,21 @@
 
 -export([start_link/0]).
 -export([init/1]).
--export([start_otel/1]).
+-export([worker_spec/2]).
--export([stop_otel/0]).
 
--define(CHILD(Mod, Opts), #{
+worker_spec(Mod, Opts) ->
-    id => Mod,
+    #{
-    start => {Mod, start_link, [Opts]},
+        id => Mod,
-    restart => permanent,
+        start => {Mod, start_link, [Opts]},
-    shutdown => 5000,
+        restart => permanent,
-    type => worker,
+        shutdown => 5000,
-    modules => [Mod]
+        type => worker,
-}).
+        modules => [Mod]
-
+    }.
--define(WORKER, emqx_otel).
 
 start_link() ->
     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
--spec start_otel(map()) -> ok.
-start_otel(Conf) ->
-    assert_started(supervisor:start_child(?MODULE, ?CHILD(?WORKER, Conf))).
-
--spec stop_otel() -> ok | {error, term()}.
-stop_otel() ->
-    ok = emqx_otel:cleanup(),
-    case supervisor:terminate_child(?MODULE, ?WORKER) of
-        ok -> supervisor:delete_child(?MODULE, ?WORKER);
-        {error, not_found} -> ok;
-        Error -> Error
-    end.
-
 init([]) ->
     SupFlags = #{
         strategy => one_for_one,
@@ -58,11 +43,6 @@ init([]) ->
     Children =
         case emqx_conf:get([opentelemetry]) of
             #{enable := false} -> [];
-            #{enable := true} = Conf -> [?CHILD(?WORKER, Conf)]
+            #{enable := true} = Conf -> [worker_spec(emqx_otel, Conf)]
         end,
     {ok, {SupFlags, Children}}.
-
-assert_started({ok, _Pid}) -> ok;
-assert_started({ok, _Pid, _Info}) -> ok;
-assert_started({error, {already_started, _Pid}}) -> ok;
-assert_started({error, Reason}) -> {error, Reason}.