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

Merge pull request #5222 from DDDHuang/nodes_api 

refactor: nodes api ; add: api test util module
This commit is contained in:
DDDHuang 2021-07-13 16:56:43 +08:00 committed by GitHub
parent a8e4ea18ea 500047fa30
commit fe779925c5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 303 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

175
apps/emqx_management/src/emqx_mgmt_api_nodes.erl
View File

@ -13,49 +13,158 @@
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------
-module(emqx_mgmt_api_nodes).
-rest_api(#{name => list_nodes,
method => 'GET',
path => "/nodes/",
func => list,
descr => "A list of nodes in the cluster"}).
-behavior(minirest_api).
-rest_api(#{name => get_node,
method => 'GET',
path => "/nodes/:atom:node",
func => get,
descr => "Lookup a node in the cluster"}).
-export([api_spec/0]).
-export([ list/2
, get/2
]).
-export([ nodes/2
, node/2]).
list(_Bindings, _Params) ->
emqx_mgmt:return({ok, [format(Node, Info) || {Node, Info} <- emqx_mgmt:list_nodes()]}).
-include_lib("emqx/include/emqx.hrl").
get(#{node := Node}, _Params) ->
emqx_mgmt:return({ok, emqx_mgmt:lookup_node(Node)}).
api_spec() ->
{apis(), schemas()}.
format(Node, {error, Reason}) -> #{node => Node, error => Reason};
apis() ->
[ nodes_api()
, node_api()].
schemas() ->
[node_schema()].
node_schema() ->
DefinitionName = <<"node">>,
DefinitionProperties = #{
<<"node">> => #{
type => <<"string">>,
description => <<"Node name">>},
<<"connections">> => #{
type => <<"integer">>,
description => <<"Number of clients currently connected to this node">>},
<<"load1">> => #{
type => <<"string">>,
description => <<"CPU average load in 1 minute">>},
<<"load5">> => #{
type => <<"string">>,
description => <<"CPU average load in 5 minute">>},
<<"load15">> => #{
type => <<"string">>,
description => <<"CPU average load in 15 minute">>},
<<"max_fds">> => #{
type => <<"integer">>,
description => <<"Maximum file descriptor limit for the operating system">>},
<<"memory_total">> => #{
type => <<"string">>,
description => <<"VM allocated system memory">>},
<<"memory_used">> => #{
type => <<"string">>,
description => <<"VM occupied system memory">>},
<<"node_status">> => #{
type => <<"string">>,
description => <<"Node status">>},
<<"otp_release">> => #{
type => <<"string">>,
description => <<"Erlang/OTP version used by EMQ X Broker">>},
<<"process_available">> => #{
type => <<"integer">>,
description => <<"Number of available processes">>},
<<"process_used">> => #{
type => <<"integer">>,
description => <<"Number of used processes">>},
<<"uptime">> => #{
type => <<"string">>,
description => <<"EMQ X Broker runtime">>},
<<"version">> => #{
type => <<"string">>,
description => <<"EMQ X Broker version">>},
<<"sys_path">> => #{
type => <<"string">>,
description => <<"EMQ X system file location">>},
<<"log_path">> => #{
type => <<"string">>,
description => <<"EMQ X log file location">>},
<<"config_path">> => #{
type => <<"string">>,
description => <<"EMQ X config file location">>}
},
{DefinitionName, DefinitionProperties}.
nodes_api() ->
Metadata = #{
get => #{
description => "List EMQ X nodes",
responses => #{
<<"200">> => #{description => <<"List EMQ X Nodes">>,
schema => #{
type => array,
items => cowboy_swagger:schema(<<"node">>)}}}}},
{"/nodes", Metadata, nodes}.
node_api() ->
Metadata = #{
get => #{
description => "Get node info",
parameters => [#{
name => node_name,
in => path,
description => "node name",
type => string,
required => true,
default => node()}],
responses => #{
<<"400">> =>
emqx_mgmt_util:not_found_schema(<<"Node error">>, [<<"SOURCE_ERROR">>]),
<<"200">> => #{
description => <<"Get EMQ X Nodes info by name">>,
schema => cowboy_swagger:schema(<<"node">>)}}}},
{"/nodes/:node_name", Metadata, node}.
%%%==============================================================================================
%% parameters trans
nodes(get, _Request) ->
list(#{}).
node(get, Request) ->
NodeName = cowboy_req:binding(node_name, Request),
Node = binary_to_atom(NodeName, utf8),
get_node(#{node => Node}).
%%%==============================================================================================
%% api apply
list(#{}) ->
NodesInfo = [format(Node, NodeInfo) || {Node, NodeInfo} <- emqx_mgmt:list_nodes()],
Response = emqx_json:encode(NodesInfo),
{200, Response}.
get_node(#{node := Node}) ->
case emqx_mgmt:lookup_node(Node) of
#{node_status := 'ERROR'} ->
{400, emqx_json:encode(#{code => 'SOURCE_ERROR', reason => <<"rpc_failed">>})};
NodeInfo ->
Response = emqx_json:encode(format(Node, NodeInfo)),
{200, Response}
end.
%%============================================================================================================
%% internal function
format(_Node, Info = #{memory_total := Total, memory_used := Used}) ->
{ok, SysPathBinary} = file:get_cwd(),
SysPath = list_to_binary(SysPathBinary),
ConfigPath = <<SysPath/binary, "/etc/emqx.conf">>,
LogPath = case log_path() of
undefined ->
<<"not found">>;
Path0 ->
Path = list_to_binary(Path0),
<<SysPath/binary, Path/binary>>
end,
Info#{ memory_total := emqx_mgmt_util:kmg(Total)
, memory_used := emqx_mgmt_util:kmg(Used)
, sys_path => SysPath
, config_path => ConfigPath
, log_path => LogPath}.
SysPath = list_to_binary(SysPathBinary),
ConfigPath = <<SysPath/binary, "/etc/emqx.conf">>,
LogPath = case log_path() of
undefined ->
<<"not found">>;
Path0 ->
Path = list_to_binary(Path0),
<<SysPath/binary, Path/binary>>
end,
Info#{ memory_total := emqx_mgmt_util:kmg(Total)
, memory_used := emqx_mgmt_util:kmg(Used)
, sys_path => SysPath
, config_path => ConfigPath
, log_path => LogPath}.
log_path() ->
Configs = logger:get_handler_config(),

2
apps/emqx_management/src/emqx_mgmt_util.erl
View File

@ -22,6 +22,7 @@
, ntoa/1
, merge_maps/2
, not_found_schema/1
, not_found_schema/2
, batch_operation/3
]).
@ -94,6 +95,7 @@ not_found_schema(Description, Enum) ->
reason => #{
type => string}}}
}.
batch_operation(Module, Function, ArgsList) ->
Failed = batch_operation(Module, Function, ArgsList, []),
Len = erlang:length(Failed),

85
apps/emqx_management/test/emqx_mgmt_api_test_util.erl Normal file
View File

@ -0,0 +1,85 @@
%%--------------------------------------------------------------------
%% 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_mgmt_api_test_util).
-compile(export_all).
-compile(nowarn_export_all).
-define(SERVER, "http://127.0.0.1:8081").
-define(BASE_PATH, "/api/v5").
default_init() ->
ekka_mnesia:start(),
emqx_mgmt_auth:mnesia(boot),
emqx_ct_helpers:start_apps([emqx_management], fun set_special_configs/1),
ok.
default_end() ->
emqx_ct_helpers:stop_apps([emqx_management]).
set_special_configs(emqx_management) ->
emqx_config:put([emqx_management], #{listeners => [#{protocol => http, port => 8081}],
applications =>[#{id => "admin", secret => "public"}]}),
ok;
set_special_configs(_App) ->
ok.
request_api(Method, Url) ->
request_api(Method, Url, [], auth_header_(), []).
request_api(Method, Url, Auth) ->
request_api(Method, Url, [], Auth, []).
request_api(Method, Url, QueryParams, Auth) ->
request_api(Method, Url, QueryParams, Auth, []).
request_api(Method, Url, QueryParams, Auth, []) ->
NewUrl = case QueryParams of
"" -> Url;
_ -> Url ++ "?" ++ QueryParams
end,
do_request_api(Method, {NewUrl, [Auth]});
request_api(Method, Url, QueryParams, Auth, Body) ->
NewUrl = case QueryParams of
"" -> Url;
_ -> Url ++ "?" ++ QueryParams
end,
do_request_api(Method, {NewUrl, [Auth], "application/json", emqx_json:encode(Body)}).
do_request_api(Method, Request)->
ct:pal("Method: ~p, Request: ~p", [Method, Request]),
case httpc:request(Method, Request, [], []) of
{error, socket_closed_remotely} ->
{error, socket_closed_remotely};
{ok, {{"HTTP/1.1", Code, _}, _, Return} }
when Code =:= 200 orelse Code =:= 201 ->
{ok, Return};
{ok, {Reason, _, _}} ->
{error, Reason}
end.
auth_header_() ->
AppId = <<"admin">>,
AppSecret = <<"public">>,
auth_header_(binary_to_list(AppId), binary_to_list(AppSecret)).
auth_header_(User, Pass) ->
Encoded = base64:encode_to_string(lists:append([User,":",Pass])),
{"Authorization","Basic " ++ Encoded}.
api_path(Parts)->
?SERVER ++ filename:join([?BASE_PATH | Parts]).

92
apps/emqx_management/test/emqx_mgmt_clients_api_SUITE.erl
View File

@ -21,27 +21,15 @@
-define(APP, emqx_management).
-define(SERVER, "http://127.0.0.1:8081").
-define(BASE_PATH, "/api/v5").
all() ->
emqx_ct:all(?MODULE).
init_per_suite(Config) ->
ekka_mnesia:start(),
emqx_mgmt_auth:mnesia(boot),
emqx_ct_helpers:start_apps([emqx_management], fun set_special_configs/1),
emqx_mgmt_api_test_util:default_init(),
Config.
end_per_suite(_) ->
emqx_ct_helpers:stop_apps([emqx_management]).
set_special_configs(emqx_management) ->
emqx_config:put([emqx_management], #{listeners => [#{protocol => http, port => 8081}],
applications =>[#{id => "admin", secret => "public"}]}),
ok;
set_special_configs(_App) ->
ok.
emqx_mgmt_api_test_util:default_end().
t_clients(_) ->
process_flag(trap_exit, true),
@ -55,6 +43,8 @@ t_clients(_) ->
Topic = <<"topic_1">>,
Qos = 0,
AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
{ok, C1} = emqtt:start_link(#{username => Username1, clientid => ClientId1}),
{ok, _} = emqtt:connect(C1),
{ok, C2} = emqtt:start_link(#{username => Username2, clientid => ClientId2}),
@ -63,7 +53,8 @@ t_clients(_) ->
timer:sleep(300),
%% get /clients
{ok, Clients} = request_api(get, api_path(["clients"])),
ClientsPath = emqx_mgmt_api_test_util:api_path(["clients"]),
{ok, Clients} = emqx_mgmt_api_test_util:request_api(get, ClientsPath),
ClientsResponse = emqx_json:decode(Clients, [return_maps]),
ClientsMeta = maps:get(<<"meta">>, ClientsResponse),
ClientsPage = maps:get(<<"page">>, ClientsMeta),
@ -74,83 +65,32 @@ t_clients(_) ->
?assertEqual(ClientsCount, 2),
%% get /clients/:clientid
{ok, Client1} = request_api(get, api_path(["clients", binary_to_list(ClientId1)])),
Client1Path = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId1)]),
{ok, Client1} = emqx_mgmt_api_test_util:request_api(get, Client1Path),
Client1Response = emqx_json:decode(Client1, [return_maps]),
?assertEqual(Username1, maps:get(<<"username">>, Client1Response)),
?assertEqual(ClientId1, maps:get(<<"clientid">>, Client1Response)),
%% delete /clients/:clientid kickout
{ok, _} = request_api(delete, api_path(["clients", binary_to_list(ClientId2)])),
AfterKickoutResponse = request_api(get, api_path(["clients", binary_to_list(ClientId2)])),
Client2Path = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId2)]),
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, Client2Path),
AfterKickoutResponse = emqx_mgmt_api_test_util:request_api(get, Client2Path),
?assertEqual({error, {"HTTP/1.1", 404, "Not Found"}}, AfterKickoutResponse),
%% get /clients/:clientid/acl_cache should has no acl cache
{ok, Client1AclCache} = request_api(get,
api_path(["clients", binary_to_list(ClientId1), "acl_cache"])),
?assertEqual("[]", Client1AclCache),
%% get /clients/:clientid/acl_cache should has no acl cache
{ok, Client1AclCache} = request_api(get,
api_path(["clients", binary_to_list(ClientId1), "acl_cache"])),
Client1AclCachePath = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId1), "acl_cache"]),
{ok, Client1AclCache} = emqx_mgmt_api_test_util:request_api(get, Client1AclCachePath),
?assertEqual("[]", Client1AclCache),
%% post /clients/:clientid/subscribe
SubscribeBody = #{topic => Topic, qos => Qos},
SubscribePath = api_path(["clients", binary_to_list(ClientId1), "subscribe"]),
{ok, _} = request_api(post, SubscribePath, "", auth_header_(), SubscribeBody),
SubscribePath = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId1), "subscribe"]),
{ok, _} = emqx_mgmt_api_test_util:request_api(post, SubscribePath, "", AuthHeader, SubscribeBody),
[{{_, AfterSubTopic}, #{qos := AfterSubQos}}] = emqx_mgmt:lookup_subscriptions(ClientId1),
?assertEqual(AfterSubTopic, Topic),
?assertEqual(AfterSubQos, Qos),
%% delete /clients/:clientid/subscribe
UnSubscribeQuery = "topic=" ++ binary_to_list(Topic),
{ok, _} = request_api(delete, SubscribePath, UnSubscribeQuery, auth_header_()),
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, SubscribePath, UnSubscribeQuery, AuthHeader),
?assertEqual([], emqx_mgmt:lookup_subscriptions(Client1)).
%%%==============================================================================================
%% test util function
request_api(Method, Url) ->
request_api(Method, Url, [], auth_header_(), []).
request_api(Method, Url, Auth) ->
request_api(Method, Url, [], Auth, []).
request_api(Method, Url, QueryParams, Auth) ->
request_api(Method, Url, QueryParams, Auth, []).
request_api(Method, Url, QueryParams, Auth, []) ->
NewUrl = case QueryParams of
"" -> Url;
_ -> Url ++ "?" ++ QueryParams
end,
do_request_api(Method, {NewUrl, [Auth]});
request_api(Method, Url, QueryParams, Auth, Body) ->
NewUrl = case QueryParams of
"" -> Url;
_ -> Url ++ "?" ++ QueryParams
end,
do_request_api(Method, {NewUrl, [Auth], "application/json", emqx_json:encode(Body)}).
do_request_api(Method, Request)->
ct:pal("Method: ~p, Request: ~p", [Method, Request]),
case httpc:request(Method, Request, [], []) of
{error, socket_closed_remotely} ->
{error, socket_closed_remotely};
{ok, {{"HTTP/1.1", Code, _}, _, Return} }
when Code =:= 200 orelse Code =:= 201 ->
{ok, Return};
{ok, {Reason, _, _}} ->
{error, Reason}
end.
auth_header_() ->
AppId = <<"admin">>,
AppSecret = <<"public">>,
auth_header_(binary_to_list(AppId), binary_to_list(AppSecret)).
auth_header_(User, Pass) ->
Encoded = base64:encode_to_string(lists:append([User,":",Pass])),
{"Authorization","Basic " ++ Encoded}.
api_path(Parts)->
?SERVER ++ filename:join([?BASE_PATH | Parts]).

58
apps/emqx_management/test/emqx_mgmt_nodes_api_SUITE.erl Normal file
View File

@ -0,0 +1,58 @@
%%--------------------------------------------------------------------
%% 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_mgmt_nodes_api_SUITE).
-compile(export_all).
-compile(nowarn_export_all).
-include_lib("eunit/include/eunit.hrl").
-define(APP, emqx_management).
-define(SERVER, "http://127.0.0.1:8081").
-define(BASE_PATH, "/api/v5").
all() ->
emqx_ct:all(?MODULE).
init_per_suite(Config) ->
ekka_mnesia:start(),
emqx_mgmt_auth:mnesia(boot),
emqx_ct_helpers:start_apps([emqx_management], fun set_special_configs/1),
Config.
end_per_suite(_) ->
emqx_ct_helpers:stop_apps([emqx_management]).
set_special_configs(emqx_management) ->
emqx_config:put([emqx_management], #{listeners => [#{protocol => http, port => 8081}],
applications =>[#{id => "admin", secret => "public"}]}),
ok;
set_special_configs(_App) ->
ok.
t_nodes_api(_) ->
NodesPath = emqx_mgmt_api_test_util:api_path(["nodes"]),
{ok, Nodes} = emqx_mgmt_api_test_util:request_api(get, NodesPath),
NodesResponse = emqx_json:decode(Nodes, [return_maps]),
LocalNodeInfo = hd(NodesResponse),
Node = binary_to_atom(maps:get(<<"node">>, LocalNodeInfo), utf8),
?assertEqual(Node, node()),
NodePath = emqx_mgmt_api_test_util:api_path(["nodes", atom_to_list(node())]),
{ok, NodeInfo} = emqx_mgmt_api_test_util:request_api(get, NodePath),
NodeNameResponse = binary_to_atom(maps:get(<<"node">>, emqx_json:decode(NodeInfo, [return_maps])), utf8),
?assertEqual(node(), NodeNameResponse).