Merge pull request #8532 from zhongwencool/sysmon-update-conf

feat: add sysmon config handler to reload conf

apps/emqx/src/emqx_config.erl

@@ -556,10 +556,12 @@ save_to_override_conf(RawConf, Opts) ->
 
 add_handlers() ->
     ok = emqx_config_logger:add_handler(),
+    emqx_sys_mon:add_handler(),
     ok.
 
 remove_handlers() ->
     ok = emqx_config_logger:remove_handler(),
+    emqx_sys_mon:remove_handler(),
     ok.
 
 load_hocon_file(FileName, LoadType) ->

apps/emqx/src/emqx_os_mon.erl

@@ -35,6 +35,8 @@
     current_sysmem_percent/0
 ]).
 
+-export([update/1]).
+
 %% gen_server callbacks
 -export([
     init/1,
@@ -52,6 +54,9 @@
 start_link() ->
     gen_server:start_link({local, ?OS_MON}, ?MODULE, [], []).
 
+update(OS) ->
+    erlang:send(?MODULE, {monitor_conf_update, OS}).
+
 %%--------------------------------------------------------------------
 %% API
 %%--------------------------------------------------------------------
@@ -87,18 +92,24 @@ current_sysmem_percent() ->
 init([]) ->
     %% memsup is not reliable, ignore
     memsup:set_sysmem_high_watermark(1.0),
+    SysHW = init_os_monitor(),
+    _ = start_mem_check_timer(),
+    _ = start_cpu_check_timer(),
+    {ok, #{sysmem_high_watermark => SysHW}}.
+
+init_os_monitor() ->
+    init_os_monitor(emqx:get_config([sysmon, os])).
+
+init_os_monitor(OS) ->
     #{
         sysmem_high_watermark := SysHW,
         procmem_high_watermark := PHW,
         mem_check_interval := MCI
-    } = emqx:get_config([sysmon, os]),
-
+    } = OS,
     set_procmem_high_watermark(PHW),
     set_mem_check_interval(MCI),
     ok = update_mem_alarm_status(SysHW),
-    _ = start_mem_check_timer(),
-    _ = start_cpu_check_timer(),
-    {ok, #{sysmem_high_watermark => SysHW}}.
+    SysHW.
 
 handle_call(get_sysmem_high_watermark, _From, #{sysmem_high_watermark := HWM} = State) ->
     {reply, HWM, State};
@@ -147,6 +158,9 @@ handle_info({timeout, _Timer, cpu_check}, State) ->
     end,
     ok = start_cpu_check_timer(),
     {noreply, State};
+handle_info({monitor_conf_update, OS}, _State) ->
+    SysHW = init_os_monitor(OS),
+    {noreply, #{sysmem_high_watermark => SysHW}};
 handle_info(Info, State) ->
     ?SLOG(error, #{msg => "unexpected_info", info => Info}),
     {noreply, State}.

apps/emqx/src/emqx_sys_mon.erl

@@ -35,32 +35,52 @@
     terminate/2,
     code_change/3
 ]).
+-export([add_handler/0, remove_handler/0, post_config_update/5]).
+-export([update/1]).
 
 -define(SYSMON, ?MODULE).
+-define(SYSMON_CONF_ROOT, [sysmon]).
 
 %% @doc Start the system monitor.
 -spec start_link() -> startlink_ret().
 start_link() ->
     gen_server:start_link({local, ?SYSMON}, ?MODULE, [], []).
 
+add_handler() ->
+    ok = emqx_config_handler:add_handler(?SYSMON_CONF_ROOT, ?MODULE),
+    ok.
+
+remove_handler() ->
+    ok = emqx_config_handler:remove_handler(?SYSMON_CONF_ROOT),
+    ok.
+
+post_config_update(_, _Req, NewConf, OldConf, _AppEnvs) ->
+    #{os := OS1, vm := VM1} = OldConf,
+    #{os := OS2, vm := VM2} = NewConf,
+    VM1 =/= VM2 andalso ?MODULE:update(VM2),
+    OS1 =/= OS2 andalso emqx_os_mon:update(OS2),
+    ok.
+
+update(VM) ->
+    erlang:send(?MODULE, {monitor_conf_update, VM}).
+
 %%--------------------------------------------------------------------
 %% gen_server callbacks
 %%--------------------------------------------------------------------
 
 init([]) ->
-    _ = erlang:system_monitor(self(), sysm_opts()),
     emqx_logger:set_proc_metadata(#{sysmon => true}),
+    init_system_monitor(),
 
     %% Monitor cluster partition event
     ekka:monitor(partition, fun handle_partition_event/1),
-
     {ok, start_timer(#{timer => undefined, events => []})}.
 
 start_timer(State) ->
     State#{timer := emqx_misc:start_timer(timer:seconds(2), reset)}.
 
-sysm_opts() ->
-    sysm_opts(maps:to_list(emqx:get_config([sysmon, vm])), []).
+sysm_opts(VM) ->
+    sysm_opts(maps:to_list(VM), []).
 sysm_opts([], Acc) ->
     Acc;
 sysm_opts([{_, disabled} | Opts], Acc) ->
@@ -176,12 +196,16 @@ handle_info({monitor, SusPid, busy_dist_port, Port}, State) ->
     );
 handle_info({timeout, _Ref, reset}, State) ->
     {noreply, State#{events := []}, hibernate};
+handle_info({monitor_conf_update, VM}, State) ->
+    init_system_monitor(VM),
+    {noreply, State#{events := []}, hibernate};
 handle_info(Info, State) ->
     ?SLOG(error, #{msg => "unexpected_info", info => Info}),
     {noreply, State}.
 
 terminate(_Reason, #{timer := TRef}) ->
-    emqx_misc:cancel_timer(TRef).
+    emqx_misc:cancel_timer(TRef),
+    ok.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
@@ -237,3 +261,11 @@ safe_publish(Event, WarnMsg) ->
 sysmon_msg(Topic, Payload) ->
     Msg = emqx_message:make(?SYSMON, Topic, Payload),
     emqx_message:set_flag(sys, Msg).
+
+init_system_monitor() ->
+    VM = emqx:get_config([sysmon, vm]),
+    init_system_monitor(VM).
+
+init_system_monitor(VM) ->
+    _ = erlang:system_monitor(self(), sysm_opts(VM)),
+    ok.

apps/emqx/src/emqx_vm_mon.erl

@@ -86,7 +86,7 @@ handle_info({timeout, _Timer, check}, State) ->
                 },
                 Message
             );
-        _Precent ->
+        _Percent ->
             ok
     end,
     _ = start_check_timer(),

apps/emqx_management/test/emqx_mgmt_api_configs_SUITE.erl

@@ -50,6 +50,7 @@ t_update(_Config) ->
     {ok, SysMon1} = get_config(<<"sysmon">>),
     #{<<"vm">> := #{<<"busy_port">> := BusyPort1}} = SysMon1,
     ?assertEqual(BusyPort, not BusyPort1),
+    assert_busy_port(BusyPort1),
 
     %% update failed
     ErrorSysMon = emqx_map_lib:deep_put([<<"vm">>, <<"busy_port">>], SysMon, "123"),
@@ -64,6 +65,7 @@ t_update(_Config) ->
     ok = reset_config(<<"sysmon">>, "conf_path=vm.busy_port"),
     {ok, SysMon3} = get_config(<<"sysmon">>),
     ?assertMatch(#{<<"vm">> := #{<<"busy_port">> := true}}, SysMon3),
+    assert_busy_port(true),
 
     %% reset no_default_value config
     NewSysMon1 = emqx_map_lib:deep_put([<<"vm">>, <<"busy_port">>], SysMon, false),
@@ -73,6 +75,11 @@ t_update(_Config) ->
     ?assertMatch(#{<<"vm">> := #{<<"busy_port">> := false}}, SysMon4),
     ok.
 
+assert_busy_port(BusyPort) ->
+    {_Pid, Monitors} = erlang:system_monitor(),
+    RealBusyPort = proplists:get_value(busy_port, Monitors, false),
+    ?assertEqual(BusyPort, RealBusyPort).
+
 t_log(_Config) ->
     {ok, Log} = get_config("log"),
     File = "log/emqx-test.log",