refactor(statsd): optimize default value

apps/emqx_statsd/include/emqx_statsd.hrl

@@ -5,5 +5,5 @@
 -define(DEFAULT_PREFIX, undefined).
 -define(DEFAULT_TAGS, #{}).
 -define(DEFAULT_BATCH_SIZE, 10).
--define(DEFAULT_SAMPLE_TIME_INTERVAL, 10).
+-define(DEFAULT_SAMPLE_TIME_INTERVAL, 10000).
--define(DEFAULT_FLUSH_TIME_INTERVAL, 10).
+-define(DEFAULT_FLUSH_TIME_INTERVAL, 10000).

apps/emqx_statsd/src/emqx_statsd.app.src

@@ -1,6 +1,6 @@
 {application, emqx_statsd,
  [{description, "An OTP application"},
-  {vsn, "0.1.0"},
+  {vsn, "5.0.0"},
   {registered, []},
   {mod, {emqx_statsd_app, []}},
   {applications,

apps/emqx_statsd/src/emqx_statsd.erl

@@ -1,100 +1,115 @@
 %%--------------------------------------------------------------------
- %% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%% Copyright (c) 2021 EMQ Technologies Co., Ltd. All Rights Reserved.
- %%
+%%
- %% Licensed under the Apache License, Version 2.0 (the "License");
+%% Licensed under the Apache License, Version 2.0 (the "License");
- %% you may not use this file except in compliance with the License.
+%% you may not use this file except in compliance with the License.
- %% You may obtain a copy of the License at
+%% You may obtain a copy of the License at
- %%
+%%
- %%     http://www.apache.org/licenses/LICENSE-2.0
+%%     http://www.apache.org/licenses/LICENSE-2.0
- %%
+%%
- %% Unless required by applicable law or agreed to in writing, software
+%% Unless required by applicable law or agreed to in writing, software
- %% distributed under the License is distributed on an "AS IS" BASIS,
+%% distributed under the License is distributed on an "AS IS" BASIS,
- %% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- %% See the License for the specific language governing permissions and
+%% See the License for the specific language governing permissions and
- %% limitations under the License.
+%% limitations under the License.
- %%--------------------------------------------------------------------
+%%--------------------------------------------------------------------
 
- -module(emqx_statsd).
+-module(emqx_statsd).
 
- -behaviour(gen_server).
+-behaviour(gen_server).
 
- -ifdef(TEST).
+-ifdef(TEST).
- -compile(export_all).
+-compile(export_all).
- -compile(nowarn_export_all).
+-compile(nowarn_export_all).
- -endif.
+-endif.
 
- -include_lib("emqx/include/logger.hrl").
+-include("emqx_statsd.hrl").
 
- %% Interface
+%% Interface
- -export([start_link/1]).
+-export([start_link/1]).
 
- %% Internal Exports
+%% Internal Exports
- -export([ init/1
+-export([ init/1
-         , handle_call/3
+        , handle_call/3
-         , handle_cast/2
+        , handle_cast/2
-         , handle_info/2
+        , handle_info/2
-         , code_change/3
+        , code_change/3
-         , terminate/2
+        , terminate/2
-         ]).
+        ]).
 
- -record(state, {
+-record(state, {
-           timer                :: reference(),
+    timer                :: reference() | undefined,
-           sample_time_interval :: pos_integer(),
+    sample_time_interval :: pos_integer(),
-           flush_time_interval  :: pos_integer(),
+    flush_time_interval  :: pos_integer(),
-           estatsd_pid          :: pid()
+    estatsd_pid          :: pid()
-         }).
+}).
 
- start_link(Opts) ->
+start_link(Opts) ->
-     gen_server:start_link({local, ?MODULE}, ?MODULE, [Opts], []).
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [Opts], []).
 
- init([Opts]) ->
+init([Opts]) ->
-     SampleTimeInterval = proplists:get_value(sample_time_interval, Opts),
+    process_flag(trap_exit, true),
-     FlushTimeInterval = proplists:get_value(flush_time_interval, Opts),
+    Tags = tags(maps:get(tags, Opts, ?DEFAULT_TAGS)),
-     Ref = erlang:start_timer(SampleTimeInterval, self(), sample_timeout),
+    Opts1 = maps:without([sample_time_interval,
-     Pid = proplists:get_value(estatsd_pid, Opts),
+                          flush_time_interval], Opts#{tags => Tags}),
-     {ok, #state{timer = Ref,
+    {ok, Pid} = estatsd:start_link(maps:to_list(Opts1)),
-                 sample_time_interval = SampleTimeInterval,
+    SampleTimeInterval = maps:get(sample_time_interval, Opts, ?DEFAULT_SAMPLE_TIME_INTERVAL),
-                 flush_time_interval = FlushTimeInterval,
+    FlushTimeInterval = maps:get(flush_time_interval, Opts, ?DEFAULT_FLUSH_TIME_INTERVAL),
-                 estatsd_pid = Pid}}.
+    {ok, ensure_timer(#state{sample_time_interval = SampleTimeInterval,
+                             flush_time_interval = FlushTimeInterval,
+                             estatsd_pid = Pid})}.
 
- handle_call(_Req, _From, State) ->
+handle_call(_Req, _From, State) ->
-     {noreply, State}.
+    {noreply, State}.
 
- handle_cast(_Msg, State) ->
+handle_cast(_Msg, State) ->
-     {noreply, State}.
+    {noreply, State}.
 
- handle_info({timeout, Ref, sample_timeout}, State = #state{sample_time_interval = SampleTimeInterval,
+handle_info({timeout, Ref, sample_timeout},
-                                                            flush_time_interval = FlushTimeInterval,
+            State = #state{sample_time_interval = SampleTimeInterval,
-                                                            estatsd_pid = Pid,
+                           flush_time_interval = FlushTimeInterval,
-                                                            timer = Ref}) ->
+                           estatsd_pid = Pid,
-     Metrics = emqx_metrics:all() ++ emqx_stats:getstats() ++ emqx_vm_data(),
+                           timer = Ref}) ->
-     SampleRate = SampleTimeInterval / FlushTimeInterval,
+    Metrics = emqx_metrics:all() ++ emqx_stats:getstats() ++ emqx_vm_data(),
-     StatsdMetrics = [{gauge, trans_metrics_name(Name), Value, SampleRate, []} || {Name, Value} <- Metrics],
+    SampleRate = SampleTimeInterval / FlushTimeInterval,
-     estatsd:submit(Pid, StatsdMetrics),
+    StatsdMetrics = [{gauge, trans_metrics_name(Name), Value, SampleRate, []} || {Name, Value} <- Metrics],
-     {noreply, State#state{timer = erlang:start_timer(SampleTimeInterval, self(), sample_timeout)}};
+    estatsd:submit(Pid, StatsdMetrics),
+    {noreply, ensure_timer(State)};
 
- handle_info(_Msg, State) ->
+handle_info({'EXIT', Pid, Error}, State = #state{estatsd_pid = Pid}) ->
-     {noreply, State}.
+    {stop, {shutdown, Error}, State};
 
- code_change(_OldVsn, State, _Extra) ->
+handle_info(_Msg, State) ->
-     {ok, State}.
+    {noreply, State}.
 
- terminate(_Reason, _State) ->
+code_change(_OldVsn, State, _Extra) ->
-     ok.
+    {ok, State}.
 
- %%------------------------------------------------------------------------------
+terminate(_Reason, #state{estatsd_pid = Pid}) ->
- %% Internale function
+    estatsd:stop(Pid),
- %%------------------------------------------------------------------------------
+    ok.
- trans_metrics_name(Name) ->
-     Name0 = atom_to_binary(Name, utf8),
-     binary_to_atom(<<"emqx.", Name0/binary>>, utf8).
 
- emqx_vm_data() ->
+%%------------------------------------------------------------------------------
-     Idle = case cpu_sup:util([detailed]) of
+%% Internale function
-                {_, 0, 0, _} -> 0; %% Not support for Windows
+%%------------------------------------------------------------------------------
-                {_Num, _Use, IdleList, _} -> proplists:get_value(idle, IdleList, 0)
+trans_metrics_name(Name) ->
-            end,
+    Name0 = atom_to_binary(Name, utf8),
-     RunQueue = erlang:statistics(run_queue),
+    binary_to_atom(<<"emqx.", Name0/binary>>, utf8).
-     [{run_queue, RunQueue},
+
-      {cpu_idle, Idle},
+emqx_vm_data() ->
-      {cpu_use, 100 - Idle}] ++ emqx_vm:mem_info().
+    Idle = case cpu_sup:util([detailed]) of
+               {_, 0, 0, _} -> 0; %% Not support for Windows
+               {_Num, _Use, IdleList, _} -> proplists:get_value(idle, IdleList, 0)
+           end,
+    RunQueue = erlang:statistics(run_queue),
+    [{run_queue, RunQueue},
+     {cpu_idle, Idle},
+     {cpu_use, 100 - Idle}] ++ emqx_vm:mem_info().
+
+tags(Map) ->
+    Tags = maps:to_list(Map),
+    [{atom_to_binary(Key, utf8), Value} || {Key, Value} <- Tags].
+
+
+ensure_timer(State =#state{sample_time_interval = SampleTimeInterval}) ->
+    State#state{timer = emqx_misc:start_timer(SampleTimeInterval, sample_timeout)}.

apps/emqx_statsd/src/emqx_statsd_app.erl

@@ -18,21 +18,12 @@
 
  -behaviour(application).
 
--include_lib("emqx/include/logger.hrl").
-
- -emqx_plugin(?MODULE).
-
  -export([ start/2
          , stop/1
          ]).
 
 start(_StartType, _StartArgs) ->
-    {ok, Sup} = emqx_statsd_sup:start_link(),
+    emqx_statsd_sup:start_link().
-    {ok, _} = emqx_statsd_sup:start_statsd(),
-    ?LOG(info, "emqx statsd start: successfully"),
-    {ok, Sup}.
 
 stop(_) ->
-    ok = emqx_statsd_sup:stop_statsd(),
-    ?LOG(info, "emqx statsd stop: successfully"),
     ok.

apps/emqx_statsd/src/emqx_statsd_sup.erl

@@ -11,59 +11,21 @@
 
 -export([start_link/0]).
 
--export([start_statsd/0, stop_statsd/0]).
-
 -export([init/1]).
 
--export([estatsd_options/0]).
 
  start_link() ->
      supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
- init([]) ->
+init([]) ->
-    {ok, {{one_for_one, 10, 100}, []}}.
+    Opts = emqx_config:get([?APP], #{}),
+    {ok, {{one_for_one, 10, 3600},
+          [#{id       => emqx_statsd,
+             start    => {emqx_statsd, start_link, [Opts]},
+             restart  => permanent,
+             shutdown => 5000,
+             type     => worker,
+             modules  => [emqx_statsd]}]}}.
 
-start_statsd() ->
-    {ok, Pid} = supervisor:start_child(?MODULE, estatsd_child_spec()),
-    {ok, _Pid1} = supervisor:start_child(?MODULE, emqx_statsd_child_spec(Pid)).
 
-stop_statsd() ->
-    ok = supervisor:terminate_child(?MODULE, emqx_statsd),
-    ok = supervisor:terminate_child(?MODULE, estatsd).
-%%==============================================================================================
-%% internal
-estatsd_child_spec() ->
-    #{id       => estatsd
-    , start    => {estatsd, start_link, [estatsd_options()]}
-    , restart  => permanent
-    , shutdown => 5000
-    , type     => worker
-    , modules  => [estatsd]}.
 
-estatsd_options() ->
-    Host =  get_conf(host, ?DEFAULT_HOST),
-    Port =  get_conf(port, ?DEFAULT_PORT),
-    Prefix = get_conf(prefix, ?DEFAULT_PREFIX),
-    Tags = tags(get_conf(tags, ?DEFAULT_TAGS)),
-    BatchSize = get_conf(batch_size, ?DEFAULT_BATCH_SIZE),
-    [{host, Host}, {port, Port}, {prefix, Prefix}, {tags, Tags}, {batch_size, BatchSize}].
-
-tags(Map) ->
-    Tags = maps:to_list(Map),
-    [{atom_to_binary(Key, utf8), Value} || {Key, Value} <- Tags].
-
-emqx_statsd_child_spec(Pid) ->
-    #{id       => emqx_statsd
-    , start    => {emqx_statsd, start_link, [[{estatsd_pid, Pid} | emqx_statsd_options()]]}
-    , restart  => permanent
-    , shutdown => 5000
-    , type     => worker
-    , modules  => [emqx_statsd]}.
-
-emqx_statsd_options() ->
-    SampleTimeInterval = get_conf(sample_time_interval, ?DEFAULT_SAMPLE_TIME_INTERVAL) * 1000,
-    FlushTimeInterval = get_conf(flush_time_interval, ?DEFAULT_FLUSH_TIME_INTERVAL) * 1000,
-    [{sample_time_interval, SampleTimeInterval}, {flush_time_interval, FlushTimeInterval}].
-
-get_conf(Key, Default) ->
-    emqx_config:get([?APP, Key], Default).

apps/emqx_statsd/test/emqx_statsd_SUITE.erl

@@ -13,7 +13,7 @@ init_per_suite(Config) ->
 end_per_suite(_Config) ->
     emqx_ct_helpers:stop_apps([emqx_statsd]).
 
-all() -> 
+all() ->
     emqx_ct:all(?MODULE).
 
 t_statsd(_) ->