yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(telemetry): telemetry API support swagger

This commit is contained in:
Turtle 2021-07-22 14:59:13 +08:00 committed by turtleDeng
parent 318beef503
commit 709f6a535b
20 changed files with 420 additions and 443 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
apps/emqx/src/emqx_schema.erl
View File

@ -84,7 +84,6 @@ includes() ->[].
-else.
includes() ->
[ "emqx_data_bridge"
, "emqx_telemetry"
, "emqx_retainer"
, "emqx_statsd"
, "emqx_authn"

4
apps/emqx_modules/etc/emqx_modules.conf
View File

@ -28,6 +28,10 @@ emqx_modules: {
type: topic_metrics
enable: false
topics: ["topic/#"]
},
{
type: telemetry
enable: true
}
]
}

0
apps/emqx_telemetry/include/emqx_telemetry.hrl → apps/emqx_modules/include/emqx_modules.hrl
View File

10
apps/emqx_modules/src/emqx_mod_sup.erl
View File

@ -36,7 +36,6 @@
type => Type,
modules => [Mod]}).
-spec(start_link() -> startlink_ret()).
start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []).
@ -60,7 +59,14 @@ stop_child(ChildId) ->
%%--------------------------------------------------------------------
init([]) ->
{ok, {{one_for_one, 10, 100}, []}}.
Env = [],
{ok, {{one_for_one, 10, 3600},
[#{id => telemetry,
start => {emqx_mod_telemetry, start_link, [Env]},
restart => permanent,
shutdown => 5000,
type => worker,
modules => [emqx_mod_telemetry]}]}}.
%%--------------------------------------------------------------------
%% Internal functions

125
apps/emqx_telemetry/src/emqx_telemetry.erl → apps/emqx_modules/src/emqx_mod_telemetry.erl
View File

@ -14,17 +14,19 @@
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_telemetry).
-module(emqx_mod_telemetry).
-behaviour(gen_server).
-behaviour(emqx_gen_mod).
-include_lib("emqx/include/emqx.hrl").
-include_lib("emqx/include/logger.hrl").
-include_lib("kernel/include/file.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-include("emqx_telemetry.hrl").
-include("emqx_modules.hrl").
%% Mnesia bootstrap
-export([mnesia/1]).
@ -46,13 +48,19 @@
, code_change/3
]).
-export([ enable/0
, disable/0
, is_enabled/0
%% emqx_gen_mod callbacks
-export([ load/1
, unload/1
, description/0
]).
-export([ get_status/0
, get_uuid/0
, get_telemetry/0
]).
-export([official_version/1]).
-ifdef(TEST).
-compile(export_all).
-compile(nowarn_export_all).
@ -63,24 +71,17 @@
]).
-record(telemetry, {
id :: non_neg_integer(),
uuid :: binary(),
enabled :: boolean()
}).
id :: non_neg_integer(),
uuid :: binary()
}).
-record(state, {
uuid :: undefined | binary(),
enabled :: undefined | boolean(),
url :: string(),
report_interval :: undefined | non_neg_integer(),
timer = undefined :: undefined | reference()
}).
uuid :: undefined | binary(),
enabled :: undefined | boolean(),
url :: string(),
report_interval :: undefined | non_neg_integer(),
timer = undefined :: undefined | reference()
}).
%% The count of 100-nanosecond intervals between the UUID epoch
%% 1582-10-15 00:00:00 and the UNIX epoch 1970-01-01 00:00:00.
@ -116,14 +117,14 @@ start_link(Opts) ->
stop() ->
gen_server:stop(?MODULE).
enable() ->
load(_Env) ->
gen_server:call(?MODULE, enable).
disable() ->
unload(_Env) ->
gen_server:call(?MODULE, disable).
is_enabled() ->
gen_server:call(?MODULE, is_enabled).
get_status() ->
gen_server:call(?MODULE, get_status).
get_uuid() ->
gen_server:call(?MODULE, get_uuid).
@ -131,6 +132,9 @@ get_uuid() ->
get_telemetry() ->
gen_server:call(?MODULE, get_telemetry).
description() ->
"".
%%--------------------------------------------------------------------
%% gen_server callbacks
%%--------------------------------------------------------------------
@ -141,42 +145,40 @@ get_telemetry() ->
%% Given the chance of having two nodes bootstraping with the write
%% is very small, it should be safe to ignore.
-dialyzer([{nowarn_function, [init/1]}]).
init([Opts]) ->
State = #state{url = ?TELEMETRY_URL,
report_interval = timer:seconds(?REPORT_INTERVAR)},
NState = case mnesia:dirty_read(?TELEMETRY, ?UNIQUE_ID) of
[] ->
Enabled = proplists:get_value(enabled, Opts, true),
UUID = generate_uuid(),
ekka_mnesia:dirty_write(?TELEMETRY, #telemetry{id = ?UNIQUE_ID,
uuid = UUID,
enabled = Enabled}),
State#state{enabled = Enabled, uuid = UUID};
[#telemetry{uuid = UUID, enabled = Enabled} | _] ->
State#state{enabled = Enabled, uuid = UUID}
end,
case official_version(emqx_app:get_release()) of
init(_Opts) ->
UUID1 = case mnesia:dirty_read(?TELEMETRY, ?UNIQUE_ID) of
[] ->
UUID = generate_uuid(),
ekka_mnesia:dirty_write(?TELEMETRY, #telemetry{id = ?UNIQUE_ID,
uuid = UUID}),
UUID;
[#telemetry{uuid = UUID} | _] ->
UUID
end,
{ok, #state{url = ?TELEMETRY_URL,
report_interval = timer:seconds(?REPORT_INTERVAR),
enabled = false,
uuid = UUID1}}.
handle_call(enable, _From, State) ->
case ?MODULE:official_version(emqx_app:get_release()) of
true ->
_ = erlang:send(self(), first_report),
{ok, NState};
report_telemetry(State),
{reply, ok, ensure_report_timer(State#state{enabled = true})};
false ->
{ok, NState#state{enabled = false}}
end.
{reply, {error, not_official_version}, State}
end;
handle_call(enable, _From, State = #state{uuid = UUID}) ->
ekka_mnesia:dirty_write(?TELEMETRY, #telemetry{id = ?UNIQUE_ID,
uuid = UUID,
enabled = true}),
_ = erlang:send(self(), first_report),
{reply, ok, State#state{enabled = true}};
handle_call(disable, _From, State = #state{timer = Timer}) ->
case ?MODULE:official_version(emqx_app:get_release()) of
true ->
emqx_misc:cancel_timer(Timer),
{reply, ok, State#state{enabled = false, timer = undefined}};
false ->
{reply, {error, not_official_version}, State}
end;
handle_call(disable, _From, State = #state{uuid = UUID}) ->
ekka_mnesia:dirty_write(?TELEMETRY, #telemetry{id = ?UNIQUE_ID,
uuid = UUID,
enabled = false}),
{reply, ok, State#state{enabled = false}};
handle_call(is_enabled, _From, State = #state{enabled = Enabled}) ->
handle_call(get_status, _From, State = #state{enabled = Enabled}) ->
{reply, Enabled, State};
handle_call(get_uuid, _From, State = #state{uuid = UUID}) ->
@ -197,15 +199,6 @@ handle_continue(Continue, State) ->
?LOG(error, "Unexpected continue: ~p", [Continue]),
{noreply, State}.
handle_info(first_report, State) ->
case is_pid(erlang:whereis(emqx)) of
true ->
report_telemetry(State),
{noreply, ensure_report_timer(State)};
false ->
_ = erlang:send_after(1000, self(), first_report),
{noreply, State}
end;
handle_info({timeout, TRef, time_to_report_telemetry_data}, State = #state{timer = TRef,
enabled = false}) ->
{noreply, State};

236
apps/emqx_modules/src/emqx_mod_telemetry_api.erl Normal file
View File

@ -0,0 +1,236 @@
%%--------------------------------------------------------------------
%% 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_mod_telemetry_api).
-behavior(minirest_api).
-import(emqx_mgmt_util, [ response_schema/1
, response_schema/2
, request_body_schema/1
]).
% -export([cli/1]).
-export([ status/2
, data/2
]).
-export([enable_telemetry/2]).
-export([api_spec/0]).
api_spec() ->
{[status_api(), data_api()], schemas()}.
schemas() ->
[#{broker_info => #{
type => object,
properties => #{
emqx_version => #{
type => string,
description => <<"EMQ X Version">>},
license => #{
type => object,
properties => #{
edition => #{type => string}
},
description => <<"EMQ X License">>},
os_name => #{
type => string,
description => <<"OS Name">>},
os_version => #{
type => string,
description => <<"OS Version">>},
otp_version => #{
type => string,
description => <<"Erlang/OTP Version">>},
up_time => #{
type => integer,
description => <<"EMQ X Runtime">>},
uuid => #{
type => string,
description => <<"EMQ X UUID">>},
nodes_uuid => #{
type => array,
items => #{type => string},
description => <<"EMQ X Cluster Nodes UUID">>},
active_plugins => #{
type => array,
items => #{type => string},
description => <<"EMQ X Active Plugins">>},
active_modules => #{
type => array,
items => #{type => string},
description => <<"EMQ X Active Modules">>},
num_clients => #{
type => integer,
description => <<"EMQ X Current Connections">>},
messages_received => #{
type => integer,
description => <<"EMQ X Current Received Message">>},
messages_sent => #{
type => integer,
description => <<"EMQ X Current Sent Message">>}
}
}}].
status_api() ->
Metadata = #{
get => #{
description => "Get telemetry status",
responses => #{
<<"200">> => response_schema(<<"Bad Request">>,
#{
type => object,
properties => #{enable => #{type => boolean}}
}
)
}
},
put => #{
description => "Enable or disbale telemetry",
'requestBody' => request_body_schema(#{
type => object,
properties => #{
enable => #{
type => boolean
}
}
}),
responses => #{
<<"200">> =>
response_schema(<<"Enable or disbale telemetry successfully">>),
<<"400">> =>
response_schema(<<"Bad Request">>,
#{
type => object,
properties => #{
message => #{type => string},
code => #{type => string}
}
}
)
}
}
},
{"/telemetry/status", Metadata, status}.
data_api() ->
Metadata = #{
get => #{
responses => #{
<<"200">> => response_schema(<<"Get telemetry data">>, <<"broker_info">>)
}
}
},
{"/telemetry/data", Metadata, data}.
%%--------------------------------------------------------------------
%% HTTP API
%%--------------------------------------------------------------------
status(get, _Request) ->
{200, get_telemetry_status()};
status(put, Request) ->
{ok, Body, _} = cowboy_req:read_body(Request),
Params = emqx_json:decode(Body, [return_maps]),
Enable = maps:get(<<"enable">>, Params),
case Enable =:= emqx_mod_telemetry:get_status() of
true ->
Reason = case Enable of
true -> <<"Telemetry status is already enabled">>;
false -> <<"Telemetry status is already disable">>
end,
{400, #{code => "BAD_REQUEST", message => Reason}};
false ->
enable_telemetry(Enable),
{200}
end.
data(get, _Request) ->
{200, emqx_json:encode(get_telemetry_data())}.
%%--------------------------------------------------------------------
%% CLI
%%--------------------------------------------------------------------
% cli(["enable", Enable0]) ->
% Enable = list_to_atom(Enable0),
% case Enable =:= emqx_mod_telemetry:is_enabled() of
% true ->
% case Enable of
% true -> emqx_ctl:print("Telemetry status is already enabled~n");
% false -> emqx_ctl:print("Telemetry status is already disable~n")
% end;
% false ->
% enable_telemetry(Enable),
% case Enable of
% true -> emqx_ctl:print("Enable telemetry successfully~n");
% false -> emqx_ctl:print("Disable telemetry successfully~n")
% end
% end;
% cli(["get", "status"]) ->
% case get_telemetry_status() of
% [{enabled, true}] ->
% emqx_ctl:print("Telemetry is enabled~n");
% [{enabled, false}] ->
% emqx_ctl:print("Telemetry is disabled~n")
% end;
% cli(["get", "data"]) ->
% TelemetryData = get_telemetry_data(),
% case emqx_json:safe_encode(TelemetryData, [pretty]) of
% {ok, Bin} ->
% emqx_ctl:print("~s~n", [Bin]);
% {error, _Reason} ->
% emqx_ctl:print("Failed to get telemetry data")
% end;
% cli(_) ->
% emqx_ctl:usage([{"telemetry enable", "Enable telemetry"},
% {"telemetry disable", "Disable telemetry"},
% {"telemetry get data", "Get reported telemetry data"}]).
%%--------------------------------------------------------------------
%% internal function
%%--------------------------------------------------------------------
enable_telemetry(Enable) ->
lists:foreach(fun(Node) ->
enable_telemetry(Node, Enable)
end, ekka_mnesia:running_nodes()).
enable_telemetry(Node, Enable) when Node =:= node() ->
case Enable of
true ->
emqx_mod_telemetry:load(#{});
false ->
emqx_mod_telemetry:unload(#{})
end;
enable_telemetry(Node, Enable) ->
rpc_call(Node, ?MODULE, enable_telemetry, [Node, Enable]).
get_telemetry_status() ->
#{enabled => emqx_mod_telemetry:get_status()}.
get_telemetry_data() ->
{ok, TelemetryData} = emqx_mod_telemetry:get_telemetry(),
TelemetryData.
rpc_call(Node, Module, Fun, Args) ->
case rpc:call(Node, Module, Fun, Args) of
{badrpc, Reason} -> {error, Reason};
Result -> Result
end.

1
apps/emqx_modules/src/emqx_modules.erl
View File

@ -150,4 +150,5 @@ name(presence) -> emqx_mod_presence;
name(recon) -> emqx_mod_recon;
name(rewrite) -> emqx_mod_rewrite;
name(topic_metrics) -> emqx_mod_topic_metrics;
name(telemetry) -> emqx_mod_telemetry;
name(Name) -> Name.

6
apps/emqx_modules/src/emqx_modules_schema.erl
View File

@ -30,6 +30,7 @@ fields("emqx_modules") ->
, hoconsc:ref(?MODULE, "presence")
, hoconsc:ref(?MODULE, "rewrite")
, hoconsc:ref(?MODULE, "topic_metrics")
, hoconsc:ref(?MODULE, "telemetry")
]))}];
fields("common") ->
[ {type, hoconsc:enum([delayed, recon])}
@ -53,6 +54,11 @@ fields("topic_metrics") ->
, {topics, hoconsc:array(binary())}
];
fields("telemetry") ->
[ {type, hoconsc:enum([telemetry])}
, {enable, emqx_schema:t(boolean(), undefined, false)}
];
fields("rules") ->
[ {action, hoconsc:enum([publish, subscribe])}
, {source_topic, emqx_schema:t(binary())}

106
apps/emqx_modules/test/emqx_mod_telemetry_SUITE.erl Normal file
View File

@ -0,0 +1,106 @@
%%--------------------------------------------------------------------
%% 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_mod_telemetry_SUITE).
-compile(export_all).
-compile(nowarn_export_all).
-include_lib("common_test/include/ct.hrl").
-include_lib("eunit/include/eunit.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-import(proplists, [get_value/2]).
all() -> emqx_ct:all(?MODULE).
init_per_suite(Config) ->
ok = ekka_mnesia:start(),
ok = emqx_mod_telemetry:mnesia(boot),
emqx_ct_helpers:boot_modules(all),
emqx_ct_helpers:start_apps([emqx_modules]),
Config.
end_per_suite(_Config) ->
emqx_ct_helpers:stop_apps([emqx_modules]).
set_special_configs(emqx_modules) ->
application:set_env(emqx, plugins_etc_dir,
emqx_ct_helpers:deps_path(emqx_modules, "test")),
Conf = #{},
ok = file:write_file(filename:join(emqx:get_env(plugins_etc_dir), 'emqx_modules.conf'), jsx:encode(Conf)),
ok;
set_special_configs(_App) ->
ok.
t_uuid(_) ->
UUID = emqx_mod_telemetry:generate_uuid(),
Parts = binary:split(UUID, <<"-">>, [global, trim]),
?assertEqual(5, length(Parts)),
{ok, UUID2} = emqx_mod_telemetry:get_uuid(),
emqx_mod_telemetry:unload(#{}),
emqx_mod_telemetry:load(#{}),
{ok, UUID3} = emqx_mod_telemetry:get_uuid(),
?assertEqual(UUID2, UUID3).
t_official_version(_) ->
true = emqx_mod_telemetry:official_version("0.0.0"),
true = emqx_mod_telemetry:official_version("1.1.1"),
true = emqx_mod_telemetry:official_version("10.10.10"),
false = emqx_mod_telemetry:official_version("0.0.0.0"),
false = emqx_mod_telemetry:official_version("1.1.a"),
true = emqx_mod_telemetry:official_version("0.0-alpha.1"),
true = emqx_mod_telemetry:official_version("1.1-alpha.1"),
true = emqx_mod_telemetry:official_version("10.10-alpha.10"),
false = emqx_mod_telemetry:official_version("1.1-alpha.0"),
true = emqx_mod_telemetry:official_version("1.1-beta.1"),
true = emqx_mod_telemetry:official_version("1.1-rc.1"),
false = emqx_mod_telemetry:official_version("1.1-alpha.a").
t_get_telemetry(_) ->
{ok, TelemetryData} = emqx_mod_telemetry:get_telemetry(),
OTPVersion = bin(erlang:system_info(otp_release)),
?assertEqual(OTPVersion, get_value(otp_version, TelemetryData)),
{ok, UUID} = emqx_mod_telemetry:get_uuid(),
?assertEqual(UUID, get_value(uuid, TelemetryData)),
?assertEqual(0, get_value(num_clients, TelemetryData)).
t_enable(_) ->
ok = meck:new(emqx_mod_telemetry, [non_strict, passthrough, no_history, no_link]),
ok = meck:expect(emqx_mod_telemetry, official_version, fun(_) -> true end),
ok = emqx_mod_telemetry:load(#{}),
?assertEqual(true, emqx_mod_telemetry:get_status()),
ok = emqx_mod_telemetry:unload(#{}),
?assertEqual(false, emqx_mod_telemetry:get_status()),
meck:unload([emqx_mod_telemetry]).
t_send_after_enable(_) ->
ok = meck:new(emqx_mod_telemetry, [non_strict, passthrough, no_history, no_link]),
ok = meck:expect(emqx_mod_telemetry, official_version, fun(_) -> true end),
ok = emqx_mod_telemetry:unload(#{}),
ok = snabbkaffe:start_trace(),
try
ok = emqx_mod_telemetry:load(#{}),
?assertMatch({ok, _}, ?block_until(#{?snk_kind := telemetry_data_reported}, 2000, 100))
after
ok = snabbkaffe:stop(),
meck:unload([emqx_mod_telemetry])
end.
bin(L) when is_list(L) ->
list_to_binary(L);
bin(B) when is_binary(B) ->
B.

26
apps/emqx_telemetry/.gitignore vendored
View File

@ -1,26 +0,0 @@
.eunit
deps
*.o
*.beam
*.plt
erl_crash.dump
ebin
rel/example_project
.concrete/DEV_MODE
.rebar
data/
*.swp
*.swo
.erlang.mk/
emqx_retainer.d
erlang.mk
rebar3.crashdump
_build
cover/
ct.coverdata
eunit.coverdata
logs/
rebar.lock
test/ct.cover.spec
etc/emqx_telemetry.conf.rendered
.rebar3/

1
apps/emqx_telemetry/README.md
View File

@ -1 +0,0 @@
# emqx-telemetry

3
apps/emqx_telemetry/etc/emqx_telemetry.conf
View File

@ -1,3 +0,0 @@
emqx_telemetry:{
enabled: true
}

1
apps/emqx_telemetry/rebar.config
View File

@ -1 +0,0 @@
{deps, []}.

14
apps/emqx_telemetry/src/emqx_telemetry.app.src
View File

@ -1,14 +0,0 @@
{application, emqx_telemetry,
[{description, "EMQ X Telemetry"},
{vsn, "5.0.0"}, % strict semver, bump manually!
{modules, []},
{registered, [emqx_telemetry_sup]},
{applications, [kernel,stdlib]},
{mod, {emqx_telemetry_app,[]}},
{env, []},
{licenses, ["Apache-2.0"]},
{maintainers, ["EMQ X Team <contact@emqx.io>"]},
{links, [{"Homepage", "https://emqx.io/"},
{"Github", "https://github.com/emqx/emqx-telemetry"}
]}
]}.

132
apps/emqx_telemetry/src/emqx_telemetry_api.erl
View File

@ -1,132 +0,0 @@
%%--------------------------------------------------------------------
%% 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_telemetry_api).
-rest_api(#{name => enable_telemetry,
method => 'PUT',
path => "/telemetry/status",
func => enable,
descr => "Enable or disbale telemetry"}).
-rest_api(#{name => get_telemetry_status,
method => 'GET',
path => "/telemetry/status",
func => get_status,
descr => "Get telemetry status"}).
-rest_api(#{name => get_telemetry_data,
method => 'GET',
path => "/telemetry/data",
func => get_data,
descr => "Get reported telemetry data"}).
-export([ cli/1
, enable/2
, get_status/2
, get_data/2
, enable_telemetry/1
, disable_telemetry/1
, get_telemetry_status/0
, get_telemetry_data/0
]).
%%--------------------------------------------------------------------
%% CLI
%%--------------------------------------------------------------------
cli(["enable"]) ->
enable_telemetry(),
emqx_ctl:print("Enable telemetry successfully~n");
cli(["disable"]) ->
disable_telemetry(),
emqx_ctl:print("Disable telemetry successfully~n");
cli(["get", "status"]) ->
case get_telemetry_status() of
[{enabled, true}] ->
emqx_ctl:print("Telemetry is enabled~n");
[{enabled, false}] ->
emqx_ctl:print("Telemetry is disabled~n")
end;
cli(["get", "data"]) ->
{ok, TelemetryData} = get_telemetry_data(),
case emqx_json:safe_encode(TelemetryData, [pretty]) of
{ok, Bin} ->
emqx_ctl:print("~s~n", [Bin]);
{error, _Reason} ->
emqx_ctl:print("Failed to get telemetry data")
end;
cli(_) ->
emqx_ctl:usage([{"telemetry enable", "Enable telemetry"},
{"telemetry disable", "Disable telemetry"},
{"telemetry get data", "Get reported telemetry data"}]).
%%--------------------------------------------------------------------
%% HTTP API
%%--------------------------------------------------------------------
enable(_Bindings, Params) ->
case proplists:get_value(<<"enabled">>, Params) of
true ->
enable_telemetry(),
return(ok);
false ->
disable_telemetry(),
return(ok);
undefined ->
return({error, missing_required_params})
end.
get_status(_Bindings, _Params) ->
return({ok, get_telemetry_status()}).
get_data(_Bindings, _Params) ->
return(get_telemetry_data()).
enable_telemetry() ->
lists:foreach(fun enable_telemetry/1, ekka_mnesia:running_nodes()).
enable_telemetry(Node) when Node =:= node() ->
emqx_telemetry:enable();
enable_telemetry(Node) ->
rpc_call(Node, ?MODULE, enable_telemetry, [Node]).
disable_telemetry() ->
lists:foreach(fun disable_telemetry/1, ekka_mnesia:running_nodes()).
disable_telemetry(Node) when Node =:= node() ->
emqx_telemetry:disable();
disable_telemetry(Node) ->
rpc_call(Node, ?MODULE, disable_telemetry, [Node]).
get_telemetry_status() ->
[{enabled, emqx_telemetry:is_enabled()}].
get_telemetry_data() ->
emqx_telemetry:get_telemetry().
rpc_call(Node, Module, Fun, Args) ->
case rpc:call(Node, Module, Fun, Args) of
{badrpc, Reason} -> {error, Reason};
Result -> Result
end.
%% TODO: V5 API
return(_) -> ok.

34
apps/emqx_telemetry/src/emqx_telemetry_app.erl
View File

@ -1,34 +0,0 @@
%%--------------------------------------------------------------------
%% 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_telemetry_app).
-behaviour(application).
-define(APP, emqx_telemetry).
-export([ start/2
, stop/1
]).
start(_Type, _Args) ->
Enabled = emqx_config:get([?APP, enabled], true),
emqx_telemetry_sup:start_link([{enabled, Enabled}]).
stop(_State) ->
emqx_ctl:unregister_command(telemetry),
ok.

29
apps/emqx_telemetry/src/emqx_telemetry_schema.erl
View File

@ -1,29 +0,0 @@
%%--------------------------------------------------------------------
%% 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_telemetry_schema).
-include_lib("typerefl/include/types.hrl").
-behaviour(hocon_schema).
-export([ structs/0
, fields/1]).
structs() -> ["emqx_telemetry"].
fields("emqx_telemetry") ->
[{enabled, emqx_schema:t(boolean(), undefined, false)}].

35
apps/emqx_telemetry/src/emqx_telemetry_sup.erl
View File

@ -1,35 +0,0 @@
%%--------------------------------------------------------------------
%% 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_telemetry_sup).
-behaviour(supervisor).
-export([start_link/1]).
-export([init/1]).
start_link(Env) ->
supervisor:start_link({local, ?MODULE}, ?MODULE, [Env]).
init([Env]) ->
{ok, {{one_for_one, 10, 3600},
[#{id => telemetry,
start => {emqx_telemetry, start_link, [Env]},
restart => permanent,
shutdown => 5000,
type => worker,
modules => [emqx_telemetry]}]}}.

98
apps/emqx_telemetry/test/emqx_telemetry_SUITE.erl
View File

@ -1,98 +0,0 @@
%%--------------------------------------------------------------------
%% 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_telemetry_SUITE).
-compile(export_all).
-compile(nowarn_export_all).
-include_lib("common_test/include/ct.hrl").
-include_lib("eunit/include/eunit.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").
-import(proplists, [get_value/2]).
all() -> emqx_ct:all(?MODULE).
init_per_testcase(_, Config) ->
emqx_ct_helpers:boot_modules(all),
emqx_ct_helpers:start_apps([emqx_telemetry], fun set_special_configs/1),
Config.
end_per_testcase(_, _Config) ->
emqx_ct_helpers:stop_apps([emqx_telemetry]).
set_special_configs(emqx_telemetry) ->
application:set_env(emqx, plugins_etc_dir,
emqx_ct_helpers:deps_path(emqx_telemetry, "test")),
Conf = #{<<"emqx_telemetry">> => #{<<"enabled">> => true}},
ok = file:write_file(filename:join(emqx:get_env(plugins_etc_dir), 'emqx_telemetry.conf'), jsx:encode(Conf)),
ok;
set_special_configs(_App) ->
ok.
t_uuid(_) ->
UUID = emqx_telemetry:generate_uuid(),
Parts = binary:split(UUID, <<"-">>, [global, trim]),
?assertEqual(5, length(Parts)),
{ok, UUID2} = emqx_telemetry:get_uuid(),
emqx_telemetry:stop(),
emqx_telemetry:start_link([{enabled, true}]),
{ok, UUID3} = emqx_telemetry:get_uuid(),
?assertEqual(UUID2, UUID3).
t_official_version(_) ->
true = emqx_telemetry:official_version("0.0.0"),
true = emqx_telemetry:official_version("1.1.1"),
true = emqx_telemetry:official_version("10.10.10"),
false = emqx_telemetry:official_version("0.0.0.0"),
false = emqx_telemetry:official_version("1.1.a"),
true = emqx_telemetry:official_version("0.0-alpha.1"),
true = emqx_telemetry:official_version("1.1-alpha.1"),
true = emqx_telemetry:official_version("10.10-alpha.10"),
false = emqx_telemetry:official_version("1.1-alpha.0"),
true = emqx_telemetry:official_version("1.1-beta.1"),
true = emqx_telemetry:official_version("1.1-rc.1"),
false = emqx_telemetry:official_version("1.1-alpha.a").
t_get_telemetry(_) ->
{ok, TelemetryData} = emqx_telemetry:get_telemetry(),
OTPVersion = bin(erlang:system_info(otp_release)),
?assertEqual(OTPVersion, get_value(otp_version, TelemetryData)),
{ok, UUID} = emqx_telemetry:get_uuid(),
?assertEqual(UUID, get_value(uuid, TelemetryData)),
?assertEqual(0, get_value(num_clients, TelemetryData)).
t_enable(_) ->
ok = emqx_telemetry:enable(),
?assertEqual(true, emqx_telemetry:is_enabled()),
ok = emqx_telemetry:disable(),
?assertEqual(false, emqx_telemetry:is_enabled()).
t_send_after_enable(_) ->
ok = emqx_telemetry:disable(),
ok = snabbkaffe:start_trace(),
try
ok = emqx_telemetry:enable(),
?assertMatch({ok, _}, ?block_until(#{?snk_kind := telemetry_data_reported}, 2000, 100))
after
ok = snabbkaffe:stop()
end.
bin(L) when is_list(L) ->
list_to_binary(L);
bin(B) when is_binary(B) ->
B.

1
rebar.config.erl
View File

@ -281,7 +281,6 @@ relx_apps(ReleaseType) ->
, emqx_statsd
]
++ [quicer || is_quicer_supported()]
++ [emqx_telemetry || not is_enterprise()]
++ [emqx_license || is_enterprise()]
++ [bcrypt || provide_bcrypt_release(ReleaseType)]
++ relx_apps_per_rel(ReleaseType)