Merge pull request #5292 from turtleDeng/statsd

feat(statsd): optimize default value & add statsd http API

apps/emqx/src/emqx.erl

@@ -250,5 +250,4 @@ emqx_feature() ->
     , emqx_bridge_mqtt
     , emqx_modules
     , emqx_management
-    , emqx_retainer
+    , emqx_retainer].
-    , emqx_statsd].

apps/emqx_statsd/etc/emqx_statsd.conf

@@ -3,11 +3,8 @@
  ##--------------------------------------------------------------------
 
 emqx_statsd:{
-    host: "127.0.0.1"
+    enable: true
-    port: 8125
+    server: "127.0.0.1:8125"
-    batch_size: 10
-    prefix: "emqx"
-    tags: {"from": "emqx"}
     sample_time_interval: "10s"
     flush_time_interval: "10s"
 }

apps/emqx_statsd/include/emqx_statsd.hrl

@@ -1,9 +1,5 @@
 -define(APP, emqx_statsd).
-
+-define(DEFAULT_SAMPLE_TIME_INTERVAL, 10000).
--define(DEFAULT_HOST, {127, 0, 0, 1}).
+-define(DEFAULT_FLUSH_TIME_INTERVAL, 10000).
+-define(DEFAULT_HOST, "127.0.0.1").
 -define(DEFAULT_PORT, 8125).
--define(DEFAULT_PREFIX, undefined).
--define(DEFAULT_TAGS, #{}).
--define(DEFAULT_BATCH_SIZE, 10).
--define(DEFAULT_SAMPLE_TIME_INTERVAL, 10).
--define(DEFAULT_FLUSH_TIME_INTERVAL, 10).

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,119 @@
 %%--------------------------------------------------------------------
- %% 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, #{})),
-     Ref = erlang:start_timer(SampleTimeInterval, self(), sample_timeout),
+    {Host, Port} = maps:get(server, Opts, {?DEFAULT_HOST, ?DEFAULT_PORT}),
-     Pid = proplists:get_value(estatsd_pid, Opts),
+    Opts1 = maps:without([sample_time_interval,
-     {ok, #state{timer = Ref,
+                          flush_time_interval], Opts#{tags => Tags,
-                 sample_time_interval = SampleTimeInterval,
+                                                      host => Host,
-                 flush_time_interval = FlushTimeInterval,
+                                                      port => Port,
-                 estatsd_pid = Pid}}.
+                                                      prefix => <<"emqx">>}),
+    {ok, Pid} = estatsd:start_link(maps:to_list(Opts1)),
+    SampleTimeInterval = maps:get(sample_time_interval, Opts, ?DEFAULT_FLUSH_TIME_INTERVAL),
+    FlushTimeInterval = maps:get(flush_time_interval, Opts, ?DEFAULT_FLUSH_TIME_INTERVAL),
+    {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_api.erl

@@ -0,0 +1,108 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_statsd_api).
+
+-behaviour(minirest_api).
+
+-include("emqx_statsd.hrl").
+
+-import(emqx_mgmt_util, [ response_schema/1
+                        , response_schema/2
+                        , request_body_schema/1
+                        , request_body_schema/2
+                        ]).
+
+-export([api_spec/0]).
+
+-export([ statsd/2
+        ]).
+
+api_spec() ->
+    {statsd_api(), schemas()}.
+
+schemas() ->
+    [#{statsd => #{
+        type => object,
+        properties => #{
+            server => #{
+                type => string,
+                description => <<"Statsd Server">>,
+                example => get_raw(<<"server">>, <<"127.0.0.1:8125">>)},
+            enable => #{
+                type => boolean,
+                description => <<"Statsd status">>,
+                example => get_raw(<<"enable">>, false)},
+            sample_time_interval => #{
+                type => string,
+                description => <<"Sample Time Interval">>,
+                example => get_raw(<<"sample_time_interval">>, <<"10s">>)},
+            flush_time_interval => #{
+                type => string,
+                description => <<"Flush Time Interval">>,
+                example => get_raw(<<"flush_time_interval">>, <<"10s">>)}
+        }
+    }}].
+
+statsd_api() ->
+    Metadata = #{
+        get => #{
+            description => <<"Get statsd info">>,
+            responses => #{
+                <<"200">> => response_schema(<<"statsd">>)
+            }
+        },
+        put => #{
+            description => <<"Update Statsd">>,
+            'requestBody' => request_body_schema(<<"statsd">>),
+            responses => #{
+                <<"200">> =>
+                    response_schema(<<"Update Statsd successfully">>),
+                <<"400">> =>
+                    response_schema(<<"Bad Request">>, #{
+                        type => object,
+                        properties => #{
+                            message => #{type => string},
+                            code => #{type => string}
+                        }
+                    })
+            }
+        }
+    },
+    [{"/statsd", Metadata, statsd}].
+
+statsd(get, _Request) ->
+    Response = emqx_config:get_raw([<<"emqx_statsd">>], #{}),
+    {200, Response};
+
+statsd(put, Request) ->
+    {ok, Body, _} = cowboy_req:read_body(Request),
+    Params = emqx_json:decode(Body, [return_maps]),
+    Enable = maps:get(<<"enable">>, Params),
+    ok = emqx_config:update_config([emqx_statsd], Params),
+    enable_statsd(Enable).
+
+enable_statsd(true) ->
+    ok = emqx_statsd_sup:stop_child(?APP),
+    emqx_statsd_sup:start_child(?APP, emqx_config:get([?APP], #{})),
+    {200};
+enable_statsd(false) ->
+    ok = emqx_statsd_sup:stop_child(?APP),
+    {200}.
+
+
+get_raw(Key, Def) ->
+    emqx_config:get_raw([<<"emqx_statsd">>]++ [Key], Def).

apps/emqx_statsd/src/emqx_statsd_app.erl

@@ -18,9 +18,7 @@
 
  -behaviour(application).
 
--include_lib("emqx/include/logger.hrl").
+-include("emqx_statsd.hrl").
-
- -emqx_plugin(?MODULE).
 
  -export([ start/2
          , stop/1
@@ -28,11 +26,15 @@
 
 start(_StartType, _StartArgs) ->
     {ok, Sup} = emqx_statsd_sup:start_link(),
-    {ok, _} = emqx_statsd_sup:start_statsd(),
+    maybe_enable_statsd(),
-    ?LOG(info, "emqx statsd start: successfully"),
     {ok, Sup}.
-
 stop(_) ->
-    ok = emqx_statsd_sup:stop_statsd(),
-    ?LOG(info, "emqx statsd stop: successfully"),
     ok.
+
+maybe_enable_statsd() ->
+    case emqx_config:get([?APP, enable], false) of
+        true ->
+            emqx_statsd_sup:start_child(emqx_statsd, emqx_config:get([?APP], #{}));
+        false ->
+            ok
+    end.

apps/emqx_statsd/src/emqx_statsd_schema.erl

@@ -4,41 +4,38 @@
 
 -behaviour(hocon_schema).
 
+-export([to_ip_port/1]).
+
 -export([ structs/0
         , fields/1]).
 
+-typerefl_from_string({ip_port/0, emqx_statsd_schema, to_ip_port}).
+
 structs() -> ["emqx_statsd"].
 
 fields("emqx_statsd") ->
-    [ {host, fun host/1}
+    [ {enable, emqx_schema:t(boolean(), undefined, false)}
-    , {port, fun port/1}
+    , {server, fun server/1}
-    , {prefix, fun prefix/1}
+    , {sample_time_interval, fun duration_ms/1}
-    , {tags, map()}
+    , {flush_time_interval,  fun duration_ms/1}
-    , {batch_size, fun batch_size/1}
+    ].
-    , {sample_time_interval, fun duration_s/1}
-    , {flush_time_interval,  fun duration_s/1}].
 
-host(type) -> string();
+server(type) -> emqx_schema:ip_port();
-host(default) -> "127.0.0.1";
+server(default) -> "127.0.0.1:8125";
-host(nullable) -> false;
+server(nullable) -> false;
-host(_) -> undefined.
+server(_) -> undefined.
 
-port(type) -> integer();
+duration_ms(type) -> emqx_schema:duration_ms();
-port(default) -> 8125;
+duration_ms(nullable) -> false;
-port(nullable) -> true;
+duration_ms(default) -> "10s";
-port(_) -> undefined.
+duration_ms(_) -> undefined.
 
-prefix(type) -> string();
+to_ip_port(Str) ->
-prefix(default) -> "emqx";
+     case string:tokens(Str, ":") of
-prefix(nullable) -> true;
+         [Ip, Port] ->
-prefix(_) -> undefined.
+             case inet:parse_address(Ip) of
-
+                 {ok, R} -> {ok, {R, list_to_integer(Port)}};
-batch_size(type) -> integer();
+                 _ -> {error, Str}
-batch_size(nullable) -> false;
+             end;
-batch_size(default) -> 10;
+         _ -> {error, Str}
-batch_size(_) -> undefined.
+     end.
-
-duration_s(type) -> emqx_schema:duration_s();
-duration_s(nullable) -> false;
-duration_s(default) -> "10s";
-duration_s(_) -> undefined.

apps/emqx_statsd/src/emqx_statsd_sup.erl

@@ -7,63 +7,48 @@
 
 -behaviour(supervisor).
 
--include("emqx_statsd.hrl").
+-export([ start_link/0
-
+        , start_child/1
--export([start_link/0]).
+        , start_child/2
-
+        , stop_child/1
--export([start_statsd/0, stop_statsd/0]).
+        ]).
 
 -export([init/1]).
 
--export([estatsd_options/0]).
+%% Helper macro for declaring children of supervisor
+-define(CHILD(Mod, Opts), #{id => Mod,
+                            start => {Mod, start_link, [Opts]},
+                            restart => permanent,
+                            shutdown => 5000,
+                            type => worker,
+                            modules => [Mod]}).
 
- start_link() ->
+start_link() ->
-     supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
 
- init([]) ->
+-spec start_child(supervisor:child_spec()) -> ok.
-    {ok, {{one_for_one, 10, 100}, []}}.
+start_child(ChildSpec) when is_map(ChildSpec) ->
+    assert_started(supervisor:start_child(?MODULE, ChildSpec)).
 
-start_statsd() ->
+-spec start_child(atom(), map()) -> ok.
-    {ok, Pid} = supervisor:start_child(?MODULE, estatsd_child_spec()),
+start_child(Mod, Opts) when is_atom(Mod) andalso is_map(Opts) ->
-    {ok, _Pid1} = supervisor:start_child(?MODULE, emqx_statsd_child_spec(Pid)).
+    assert_started(supervisor:start_child(?MODULE, ?CHILD(Mod, Opts))).
 
-stop_statsd() ->
+-spec(stop_child(any()) -> ok | {error, term()}).
-    ok = supervisor:terminate_child(?MODULE, emqx_statsd),
+stop_child(ChildId) ->
-    ok = supervisor:terminate_child(?MODULE, estatsd).
+    case supervisor:terminate_child(?MODULE, ChildId) of
-%%==============================================================================================
+        ok -> supervisor:delete_child(?MODULE, ChildId);
-%% internal
+        Error -> Error
-estatsd_child_spec() ->
+    end.
-    #{id       => estatsd
-    , start    => {estatsd, start_link, [estatsd_options()]}
-    , restart  => permanent
-    , shutdown => 5000
-    , type     => worker
-    , modules  => [estatsd]}.
 
-estatsd_options() ->
+init([]) ->
-    Host =  get_conf(host, ?DEFAULT_HOST),
+    {ok, {{one_for_one, 10, 3600}, []}}.
-    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),
+%% Internal functions
-    [{atom_to_binary(Key, utf8), Value} || {Key, Value} <- Tags].
+%%--------------------------------------------------------------------
 
-emqx_statsd_child_spec(Pid) ->
+assert_started({ok, _Pid}) -> ok;
-    #{id       => emqx_statsd
+assert_started({ok, _Pid, _Info}) -> ok;
-    , start    => {emqx_statsd, start_link, [[{estatsd_pid, Pid} | emqx_statsd_options()]]}
+assert_started({error, {already_tarted, _Pid}}) -> ok;
-    , restart  => permanent
+assert_started({error, Reason}) -> erlang:error(Reason).
-    , 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(_) ->