feat: support multiple clientid / username Qs params in "/clients" API
Additionally, add an option to specify which Client info fields to return in the response.
This commit is contained in:
parent
f4e20f8fc9
commit
b3d44125f6
|
@ -444,20 +444,79 @@ check_parameter(
|
||||||
check_parameter([], _Bindings, _QueryStr, _Module, NewBindings, NewQueryStr) ->
|
check_parameter([], _Bindings, _QueryStr, _Module, NewBindings, NewQueryStr) ->
|
||||||
{NewBindings, NewQueryStr};
|
{NewBindings, NewQueryStr};
|
||||||
check_parameter([{Name, Type} | Spec], Bindings, QueryStr, Module, BindingsAcc, QueryStrAcc) ->
|
check_parameter([{Name, Type} | Spec], Bindings, QueryStr, Module, BindingsAcc, QueryStrAcc) ->
|
||||||
Schema = ?INIT_SCHEMA#{roots => [{Name, Type}]},
|
|
||||||
case hocon_schema:field_schema(Type, in) of
|
case hocon_schema:field_schema(Type, in) of
|
||||||
path ->
|
path ->
|
||||||
|
Schema = ?INIT_SCHEMA#{roots => [{Name, Type}]},
|
||||||
Option = #{atom_key => true},
|
Option = #{atom_key => true},
|
||||||
NewBindings = hocon_tconf:check_plain(Schema, Bindings, Option),
|
NewBindings = hocon_tconf:check_plain(Schema, Bindings, Option),
|
||||||
NewBindingsAcc = maps:merge(BindingsAcc, NewBindings),
|
NewBindingsAcc = maps:merge(BindingsAcc, NewBindings),
|
||||||
check_parameter(Spec, Bindings, QueryStr, Module, NewBindingsAcc, QueryStrAcc);
|
check_parameter(Spec, Bindings, QueryStr, Module, NewBindingsAcc, QueryStrAcc);
|
||||||
query ->
|
query ->
|
||||||
|
Type1 = maybe_wrap_array_qs_param(Type),
|
||||||
|
Schema = ?INIT_SCHEMA#{roots => [{Name, Type1}]},
|
||||||
Option = #{},
|
Option = #{},
|
||||||
NewQueryStr = hocon_tconf:check_plain(Schema, QueryStr, Option),
|
NewQueryStr = hocon_tconf:check_plain(Schema, QueryStr, Option),
|
||||||
NewQueryStrAcc = maps:merge(QueryStrAcc, NewQueryStr),
|
NewQueryStrAcc = maps:merge(QueryStrAcc, NewQueryStr),
|
||||||
check_parameter(Spec, Bindings, QueryStr, Module, BindingsAcc, NewQueryStrAcc)
|
check_parameter(Spec, Bindings, QueryStr, Module, BindingsAcc, NewQueryStrAcc)
|
||||||
end.
|
end.
|
||||||
|
|
||||||
|
%% Compatibility layer for minirest 1.4.0 that parses repetitive QS params into lists.
|
||||||
|
%% Previous minirest releases dropped all but the last repetitive params.
|
||||||
|
|
||||||
|
maybe_wrap_array_qs_param(FieldSchema) ->
|
||||||
|
Conv = hocon_schema:field_schema(FieldSchema, converter),
|
||||||
|
Type = hocon_schema:field_schema(FieldSchema, type),
|
||||||
|
case array_or_single_qs_param(Type, Conv) of
|
||||||
|
any ->
|
||||||
|
FieldSchema;
|
||||||
|
array ->
|
||||||
|
override_conv(FieldSchema, fun wrap_array_conv/2, Conv);
|
||||||
|
single ->
|
||||||
|
override_conv(FieldSchema, fun unwrap_array_conv/2, Conv)
|
||||||
|
end.
|
||||||
|
|
||||||
|
array_or_single_qs_param(?ARRAY(_Type), undefined) ->
|
||||||
|
array;
|
||||||
|
%% Qs field schema is an array and defines a converter:
|
||||||
|
%% don't change (wrap/unwrap) the original value, and let the converter handle it.
|
||||||
|
%% For example, it can be a CSV list.
|
||||||
|
array_or_single_qs_param(?ARRAY(_Type), _Conv) ->
|
||||||
|
any;
|
||||||
|
array_or_single_qs_param(?UNION(Types), _Conv) ->
|
||||||
|
HasArray = lists:any(
|
||||||
|
fun
|
||||||
|
(?ARRAY(_)) -> true;
|
||||||
|
(_) -> false
|
||||||
|
end,
|
||||||
|
Types
|
||||||
|
),
|
||||||
|
case HasArray of
|
||||||
|
true -> any;
|
||||||
|
false -> single
|
||||||
|
end;
|
||||||
|
array_or_single_qs_param(_, _Conv) ->
|
||||||
|
single.
|
||||||
|
|
||||||
|
override_conv(FieldSchema, NewConv, OldConv) ->
|
||||||
|
Conv = compose_converters(NewConv, OldConv),
|
||||||
|
hocon_schema:override(FieldSchema, FieldSchema#{converter => Conv}).
|
||||||
|
|
||||||
|
compose_converters(NewFun, undefined = _OldFun) ->
|
||||||
|
NewFun;
|
||||||
|
compose_converters(NewFun, OldFun) ->
|
||||||
|
case erlang:fun_info(OldFun, arity) of
|
||||||
|
{_, 2} ->
|
||||||
|
fun(V, Opts) -> OldFun(NewFun(V, Opts), Opts) end;
|
||||||
|
{_, 1} ->
|
||||||
|
fun(V, Opts) -> OldFun(NewFun(V, Opts)) end
|
||||||
|
end.
|
||||||
|
|
||||||
|
wrap_array_conv(Val, _Opts) when is_list(Val); Val =:= undefined -> Val;
|
||||||
|
wrap_array_conv(SingleVal, _Opts) -> [SingleVal].
|
||||||
|
|
||||||
|
unwrap_array_conv([HVal | _], _Opts) -> HVal;
|
||||||
|
unwrap_array_conv(SingleVal, _Opts) -> SingleVal.
|
||||||
|
|
||||||
check_request_body(#{body := Body}, Schema, Module, CheckFun, true) ->
|
check_request_body(#{body := Body}, Schema, Module, CheckFun, true) ->
|
||||||
Type0 = hocon_schema:field_schema(Schema, type),
|
Type0 = hocon_schema:field_schema(Schema, type),
|
||||||
Type =
|
Type =
|
||||||
|
|
|
@ -48,6 +48,7 @@
|
||||||
finalize_query/2,
|
finalize_query/2,
|
||||||
mark_complete/2,
|
mark_complete/2,
|
||||||
format_query_result/3,
|
format_query_result/3,
|
||||||
|
format_query_result/4,
|
||||||
maybe_collect_total_from_tail_nodes/2
|
maybe_collect_total_from_tail_nodes/2
|
||||||
]).
|
]).
|
||||||
|
|
||||||
|
@ -619,10 +620,13 @@ is_fuzzy_key(<<"match_", _/binary>>) ->
|
||||||
is_fuzzy_key(_) ->
|
is_fuzzy_key(_) ->
|
||||||
false.
|
false.
|
||||||
|
|
||||||
format_query_result(_FmtFun, _MetaIn, Error = {error, _Node, _Reason}) ->
|
format_query_result(FmtFun, MetaIn, ResultAcc) ->
|
||||||
|
format_query_result(FmtFun, MetaIn, ResultAcc, #{}).
|
||||||
|
|
||||||
|
format_query_result(_FmtFun, _MetaIn, Error = {error, _Node, _Reason}, _Opts) ->
|
||||||
Error;
|
Error;
|
||||||
format_query_result(
|
format_query_result(
|
||||||
FmtFun, MetaIn, ResultAcc = #{hasnext := HasNext, rows := RowsAcc}
|
FmtFun, MetaIn, ResultAcc = #{hasnext := HasNext, rows := RowsAcc}, Opts
|
||||||
) ->
|
) ->
|
||||||
Meta =
|
Meta =
|
||||||
case ResultAcc of
|
case ResultAcc of
|
||||||
|
@ -638,7 +642,10 @@ format_query_result(
|
||||||
data => lists:flatten(
|
data => lists:flatten(
|
||||||
lists:foldl(
|
lists:foldl(
|
||||||
fun({Node, Rows}, Acc) ->
|
fun({Node, Rows}, Acc) ->
|
||||||
[lists:map(fun(Row) -> exec_format_fun(FmtFun, Node, Row) end, Rows) | Acc]
|
[
|
||||||
|
lists:map(fun(Row) -> exec_format_fun(FmtFun, Node, Row, Opts) end, Rows)
|
||||||
|
| Acc
|
||||||
|
]
|
||||||
end,
|
end,
|
||||||
[],
|
[],
|
||||||
RowsAcc
|
RowsAcc
|
||||||
|
@ -646,10 +653,11 @@ format_query_result(
|
||||||
)
|
)
|
||||||
}.
|
}.
|
||||||
|
|
||||||
exec_format_fun(FmtFun, Node, Row) ->
|
exec_format_fun(FmtFun, Node, Row, Opts) ->
|
||||||
case erlang:fun_info(FmtFun, arity) of
|
case erlang:fun_info(FmtFun, arity) of
|
||||||
{arity, 1} -> FmtFun(Row);
|
{arity, 1} -> FmtFun(Row);
|
||||||
{arity, 2} -> FmtFun(Node, Row)
|
{arity, 2} -> FmtFun(Node, Row);
|
||||||
|
{arity, 3} -> FmtFun(Node, Row, Opts)
|
||||||
end.
|
end.
|
||||||
|
|
||||||
parse_pager_params(Params) ->
|
parse_pager_params(Params) ->
|
||||||
|
|
|
@ -56,7 +56,8 @@
|
||||||
qs2ms/2,
|
qs2ms/2,
|
||||||
run_fuzzy_filter/2,
|
run_fuzzy_filter/2,
|
||||||
format_channel_info/1,
|
format_channel_info/1,
|
||||||
format_channel_info/2
|
format_channel_info/2,
|
||||||
|
format_channel_info/3
|
||||||
]).
|
]).
|
||||||
|
|
||||||
%% for batch operation
|
%% for batch operation
|
||||||
|
@ -66,7 +67,10 @@
|
||||||
|
|
||||||
-define(CLIENT_QSCHEMA, [
|
-define(CLIENT_QSCHEMA, [
|
||||||
{<<"node">>, atom},
|
{<<"node">>, atom},
|
||||||
|
%% list
|
||||||
{<<"username">>, binary},
|
{<<"username">>, binary},
|
||||||
|
%% list
|
||||||
|
{<<"clientid">>, binary},
|
||||||
{<<"ip_address">>, ip},
|
{<<"ip_address">>, ip},
|
||||||
{<<"conn_state">>, atom},
|
{<<"conn_state">>, atom},
|
||||||
{<<"clean_start">>, atom},
|
{<<"clean_start">>, atom},
|
||||||
|
@ -125,10 +129,13 @@ schema("/clients") ->
|
||||||
example => <<"emqx@127.0.0.1">>
|
example => <<"emqx@127.0.0.1">>
|
||||||
})},
|
})},
|
||||||
{username,
|
{username,
|
||||||
hoconsc:mk(binary(), #{
|
hoconsc:mk(hoconsc:array(binary()), #{
|
||||||
in => query,
|
in => query,
|
||||||
required => false,
|
required => false,
|
||||||
desc => <<"User name">>
|
desc => <<
|
||||||
|
"User name, multiple values can be specified by"
|
||||||
|
" repeating the parameter: username=u1&username=u2"
|
||||||
|
>>
|
||||||
})},
|
})},
|
||||||
{ip_address,
|
{ip_address,
|
||||||
hoconsc:mk(binary(), #{
|
hoconsc:mk(binary(), #{
|
||||||
|
@ -202,7 +209,17 @@ schema("/clients") ->
|
||||||
"Search client connection creation time by less"
|
"Search client connection creation time by less"
|
||||||
" than or equal method, rfc3339 or timestamp(millisecond)"
|
" than or equal method, rfc3339 or timestamp(millisecond)"
|
||||||
>>
|
>>
|
||||||
})}
|
})},
|
||||||
|
{clientid,
|
||||||
|
hoconsc:mk(hoconsc:array(binary()), #{
|
||||||
|
in => query,
|
||||||
|
required => false,
|
||||||
|
desc => <<
|
||||||
|
"Client ID, multiple values can be specified by"
|
||||||
|
" repeating the parameter: clientid=c1&clientid=c2"
|
||||||
|
>>
|
||||||
|
})},
|
||||||
|
?R_REF(requested_client_fields)
|
||||||
],
|
],
|
||||||
responses => #{
|
responses => #{
|
||||||
200 =>
|
200 =>
|
||||||
|
@ -656,6 +673,30 @@ fields(message) ->
|
||||||
{from_clientid, hoconsc:mk(binary(), #{desc => ?DESC(msg_from_clientid)})},
|
{from_clientid, hoconsc:mk(binary(), #{desc => ?DESC(msg_from_clientid)})},
|
||||||
{from_username, hoconsc:mk(binary(), #{desc => ?DESC(msg_from_username)})},
|
{from_username, hoconsc:mk(binary(), #{desc => ?DESC(msg_from_username)})},
|
||||||
{payload, hoconsc:mk(binary(), #{desc => ?DESC(msg_payload)})}
|
{payload, hoconsc:mk(binary(), #{desc => ?DESC(msg_payload)})}
|
||||||
|
];
|
||||||
|
fields(requested_client_fields) ->
|
||||||
|
%% NOTE: some Client fields actually returned in response are missing in schema:
|
||||||
|
%% enable_authn, is_persistent, listener, peerport
|
||||||
|
ClientFields = [element(1, F) || F <- fields(client)],
|
||||||
|
[
|
||||||
|
{fields,
|
||||||
|
hoconsc:mk(
|
||||||
|
hoconsc:union([all, hoconsc:array(hoconsc:enum(ClientFields))]),
|
||||||
|
#{
|
||||||
|
in => query,
|
||||||
|
required => false,
|
||||||
|
default => all,
|
||||||
|
desc => <<"Comma separated list of client fields to return in the response">>,
|
||||||
|
converter => fun
|
||||||
|
(all, _Opts) ->
|
||||||
|
all;
|
||||||
|
(<<"all">>, _Opts) ->
|
||||||
|
all;
|
||||||
|
(CsvFields, _Opts) when is_binary(CsvFields) ->
|
||||||
|
binary:split(CsvFields, <<",">>, [global, trim_all])
|
||||||
|
end
|
||||||
|
}
|
||||||
|
)}
|
||||||
].
|
].
|
||||||
|
|
||||||
%%%==============================================================================================
|
%%%==============================================================================================
|
||||||
|
@ -971,7 +1012,10 @@ list_clients_cluster_query(QString, Options) ->
|
||||||
?CHAN_INFO_TAB, NQString, fun ?MODULE:qs2ms/2, Meta, Options
|
?CHAN_INFO_TAB, NQString, fun ?MODULE:qs2ms/2, Meta, Options
|
||||||
),
|
),
|
||||||
Res = do_list_clients_cluster_query(Nodes, QueryState, ResultAcc),
|
Res = do_list_clients_cluster_query(Nodes, QueryState, ResultAcc),
|
||||||
emqx_mgmt_api:format_query_result(fun ?MODULE:format_channel_info/2, Meta, Res)
|
Opts = #{fields => maps:get(<<"fields">>, QString, all)},
|
||||||
|
emqx_mgmt_api:format_query_result(
|
||||||
|
fun ?MODULE:format_channel_info/3, Meta, Res, Opts
|
||||||
|
)
|
||||||
catch
|
catch
|
||||||
throw:{bad_value_type, {Key, ExpectedType, AcutalValue}} ->
|
throw:{bad_value_type, {Key, ExpectedType, AcutalValue}} ->
|
||||||
{error, invalid_query_string_param, {Key, ExpectedType, AcutalValue}}
|
{error, invalid_query_string_param, {Key, ExpectedType, AcutalValue}}
|
||||||
|
@ -1023,7 +1067,8 @@ list_clients_node_query(Node, QString, Options) ->
|
||||||
?CHAN_INFO_TAB, NQString, fun ?MODULE:qs2ms/2, Meta, Options
|
?CHAN_INFO_TAB, NQString, fun ?MODULE:qs2ms/2, Meta, Options
|
||||||
),
|
),
|
||||||
Res = do_list_clients_node_query(Node, QueryState, ResultAcc),
|
Res = do_list_clients_node_query(Node, QueryState, ResultAcc),
|
||||||
emqx_mgmt_api:format_query_result(fun ?MODULE:format_channel_info/2, Meta, Res)
|
Opts = #{fields => maps:get(<<"fields">>, QString, all)},
|
||||||
|
emqx_mgmt_api:format_query_result(fun ?MODULE:format_channel_info/3, Meta, Res, Opts)
|
||||||
end.
|
end.
|
||||||
|
|
||||||
add_persistent_session_count(QueryState0 = #{total := Totals0}) ->
|
add_persistent_session_count(QueryState0 = #{total := Totals0}) ->
|
||||||
|
@ -1190,19 +1235,36 @@ qs2ms(_Tab, {QString, FuzzyQString}) ->
|
||||||
-spec qs2ms(list()) -> ets:match_spec().
|
-spec qs2ms(list()) -> ets:match_spec().
|
||||||
qs2ms(Qs) ->
|
qs2ms(Qs) ->
|
||||||
{MtchHead, Conds} = qs2ms(Qs, 2, {#{}, []}),
|
{MtchHead, Conds} = qs2ms(Qs, 2, {#{}, []}),
|
||||||
[{{'$1', MtchHead, '_'}, Conds, ['$_']}].
|
[{{{'$1', '_'}, MtchHead, '_'}, Conds, ['$_']}].
|
||||||
|
|
||||||
qs2ms([], _, {MtchHead, Conds}) ->
|
qs2ms([], _, {MtchHead, Conds}) ->
|
||||||
{MtchHead, lists:reverse(Conds)};
|
{MtchHead, lists:reverse(Conds)};
|
||||||
|
qs2ms([{Key, '=:=', Value} | Rest], N, {MtchHead, Conds}) when is_list(Value) ->
|
||||||
|
{Holder, NxtN} = holder_and_nxt(Key, N),
|
||||||
|
NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(Key, Holder)),
|
||||||
|
qs2ms(Rest, NxtN, {NMtchHead, [orelse_cond(Holder, Value) | Conds]});
|
||||||
qs2ms([{Key, '=:=', Value} | Rest], N, {MtchHead, Conds}) ->
|
qs2ms([{Key, '=:=', Value} | Rest], N, {MtchHead, Conds}) ->
|
||||||
NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(Key, Value)),
|
NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(Key, Value)),
|
||||||
qs2ms(Rest, N, {NMtchHead, Conds});
|
qs2ms(Rest, N, {NMtchHead, Conds});
|
||||||
qs2ms([Qs | Rest], N, {MtchHead, Conds}) ->
|
qs2ms([Qs | Rest], N, {MtchHead, Conds}) ->
|
||||||
Holder = binary_to_atom(iolist_to_binary(["$", integer_to_list(N)]), utf8),
|
Holder = holder(N),
|
||||||
NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(element(1, Qs), Holder)),
|
NMtchHead = emqx_mgmt_util:merge_maps(MtchHead, ms(element(1, Qs), Holder)),
|
||||||
NConds = put_conds(Qs, Holder, Conds),
|
NConds = put_conds(Qs, Holder, Conds),
|
||||||
qs2ms(Rest, N + 1, {NMtchHead, NConds}).
|
qs2ms(Rest, N + 1, {NMtchHead, NConds}).
|
||||||
|
|
||||||
|
%% This is a special case: clientid is a part of the key (ClientId, Pid}, as the table is ordered_set,
|
||||||
|
%% using partially bound key optimizes traversal.
|
||||||
|
holder_and_nxt(clientid, N) ->
|
||||||
|
{'$1', N};
|
||||||
|
holder_and_nxt(_, N) ->
|
||||||
|
{holder(N), N + 1}.
|
||||||
|
|
||||||
|
holder(N) -> list_to_atom([$$ | integer_to_list(N)]).
|
||||||
|
|
||||||
|
orelse_cond(Holder, ValuesList) ->
|
||||||
|
Conds = [{'=:=', Holder, V} || V <- ValuesList],
|
||||||
|
erlang:list_to_tuple(['orelse' | Conds]).
|
||||||
|
|
||||||
put_conds({_, Op, V}, Holder, Conds) ->
|
put_conds({_, Op, V}, Holder, Conds) ->
|
||||||
[{Op, Holder, V} | Conds];
|
[{Op, Holder, V} | Conds];
|
||||||
put_conds({_, Op1, V1, Op2, V2}, Holder, Conds) ->
|
put_conds({_, Op1, V1, Op2, V2}, Holder, Conds) ->
|
||||||
|
@ -1212,8 +1274,8 @@ put_conds({_, Op1, V1, Op2, V2}, Holder, Conds) ->
|
||||||
| Conds
|
| Conds
|
||||||
].
|
].
|
||||||
|
|
||||||
ms(clientid, X) ->
|
ms(clientid, _X) ->
|
||||||
#{clientinfo => #{clientid => X}};
|
#{};
|
||||||
ms(username, X) ->
|
ms(username, X) ->
|
||||||
#{clientinfo => #{username => X}};
|
#{clientinfo => #{username => X}};
|
||||||
ms(conn_state, X) ->
|
ms(conn_state, X) ->
|
||||||
|
@ -1257,7 +1319,11 @@ format_channel_info({ClientId, PSInfo}) ->
|
||||||
%% offline persistent session
|
%% offline persistent session
|
||||||
format_persistent_session_info(ClientId, PSInfo).
|
format_persistent_session_info(ClientId, PSInfo).
|
||||||
|
|
||||||
format_channel_info(WhichNode, {_, ClientInfo0, ClientStats}) ->
|
format_channel_info(WhichNode, ChanInfo) ->
|
||||||
|
DefaultOpts = #{fields => all},
|
||||||
|
format_channel_info(WhichNode, ChanInfo, DefaultOpts).
|
||||||
|
|
||||||
|
format_channel_info(WhichNode, {_, ClientInfo0, ClientStats}, Opts) ->
|
||||||
Node = maps:get(node, ClientInfo0, WhichNode),
|
Node = maps:get(node, ClientInfo0, WhichNode),
|
||||||
ClientInfo1 = emqx_utils_maps:deep_remove([conninfo, clientid], ClientInfo0),
|
ClientInfo1 = emqx_utils_maps:deep_remove([conninfo, clientid], ClientInfo0),
|
||||||
ClientInfo2 = emqx_utils_maps:deep_remove([conninfo, username], ClientInfo1),
|
ClientInfo2 = emqx_utils_maps:deep_remove([conninfo, username], ClientInfo1),
|
||||||
|
@ -1276,12 +1342,47 @@ format_channel_info(WhichNode, {_, ClientInfo0, ClientStats}) ->
|
||||||
ClientInfoMap5 = convert_expiry_interval_unit(ClientInfoMap4),
|
ClientInfoMap5 = convert_expiry_interval_unit(ClientInfoMap4),
|
||||||
ClientInfoMap = maps:put(connected, Connected, ClientInfoMap5),
|
ClientInfoMap = maps:put(connected, Connected, ClientInfoMap5),
|
||||||
|
|
||||||
|
#{fields := RequestedFields} = Opts,
|
||||||
|
TimesKeys = [created_at, connected_at, disconnected_at],
|
||||||
|
%% format timestamp to rfc3339
|
||||||
|
result_format_undefined_to_null(
|
||||||
|
lists:foldl(
|
||||||
|
fun result_format_time_fun/2,
|
||||||
|
with_client_info_fields(ClientInfoMap, RequestedFields),
|
||||||
|
TimesKeys
|
||||||
|
)
|
||||||
|
);
|
||||||
|
format_channel_info(undefined, {ClientId, PSInfo0 = #{}}, _Opts) ->
|
||||||
|
format_persistent_session_info(ClientId, PSInfo0).
|
||||||
|
|
||||||
|
format_persistent_session_info(ClientId, PSInfo0) ->
|
||||||
|
Metadata = maps:get(metadata, PSInfo0, #{}),
|
||||||
|
PSInfo1 = maps:with([created_at, expiry_interval], Metadata),
|
||||||
|
CreatedAt = maps:get(created_at, PSInfo1),
|
||||||
|
PSInfo2 = convert_expiry_interval_unit(PSInfo1),
|
||||||
|
PSInfo3 = PSInfo2#{
|
||||||
|
clientid => ClientId,
|
||||||
|
connected => false,
|
||||||
|
connected_at => CreatedAt,
|
||||||
|
ip_address => undefined,
|
||||||
|
is_persistent => true,
|
||||||
|
port => undefined
|
||||||
|
},
|
||||||
|
PSInfo = lists:foldl(
|
||||||
|
fun result_format_time_fun/2,
|
||||||
|
PSInfo3,
|
||||||
|
[created_at, connected_at]
|
||||||
|
),
|
||||||
|
result_format_undefined_to_null(PSInfo).
|
||||||
|
|
||||||
|
with_client_info_fields(ClientInfoMap, all) ->
|
||||||
RemoveList =
|
RemoveList =
|
||||||
[
|
[
|
||||||
auth_result,
|
auth_result,
|
||||||
peername,
|
peername,
|
||||||
sockname,
|
sockname,
|
||||||
peerhost,
|
peerhost,
|
||||||
|
peerport,
|
||||||
conn_state,
|
conn_state,
|
||||||
send_pend,
|
send_pend,
|
||||||
conn_props,
|
conn_props,
|
||||||
|
@ -1305,37 +1406,9 @@ format_channel_info(WhichNode, {_, ClientInfo0, ClientStats}) ->
|
||||||
id,
|
id,
|
||||||
acl
|
acl
|
||||||
],
|
],
|
||||||
TimesKeys = [created_at, connected_at, disconnected_at],
|
maps:without(RemoveList, ClientInfoMap);
|
||||||
%% format timestamp to rfc3339
|
with_client_info_fields(ClientInfoMap, RequestedFields) when is_list(RequestedFields) ->
|
||||||
result_format_undefined_to_null(
|
maps:with(RequestedFields, ClientInfoMap).
|
||||||
lists:foldl(
|
|
||||||
fun result_format_time_fun/2,
|
|
||||||
maps:without(RemoveList, ClientInfoMap),
|
|
||||||
TimesKeys
|
|
||||||
)
|
|
||||||
);
|
|
||||||
format_channel_info(undefined, {ClientId, PSInfo0 = #{}}) ->
|
|
||||||
format_persistent_session_info(ClientId, PSInfo0).
|
|
||||||
|
|
||||||
format_persistent_session_info(ClientId, PSInfo0) ->
|
|
||||||
Metadata = maps:get(metadata, PSInfo0, #{}),
|
|
||||||
PSInfo1 = maps:with([created_at, expiry_interval], Metadata),
|
|
||||||
CreatedAt = maps:get(created_at, PSInfo1),
|
|
||||||
PSInfo2 = convert_expiry_interval_unit(PSInfo1),
|
|
||||||
PSInfo3 = PSInfo2#{
|
|
||||||
clientid => ClientId,
|
|
||||||
connected => false,
|
|
||||||
connected_at => CreatedAt,
|
|
||||||
ip_address => undefined,
|
|
||||||
is_persistent => true,
|
|
||||||
port => undefined
|
|
||||||
},
|
|
||||||
PSInfo = lists:foldl(
|
|
||||||
fun result_format_time_fun/2,
|
|
||||||
PSInfo3,
|
|
||||||
[created_at, connected_at]
|
|
||||||
),
|
|
||||||
result_format_undefined_to_null(PSInfo).
|
|
||||||
|
|
||||||
format_msgs_resp(MsgType, Msgs, Meta, QString) ->
|
format_msgs_resp(MsgType, Msgs, Meta, QString) ->
|
||||||
#{
|
#{
|
||||||
|
|
|
@ -715,6 +715,238 @@ t_query_clients_with_time(_) ->
|
||||||
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, Client1Path),
|
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, Client1Path),
|
||||||
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, Client2Path).
|
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, Client2Path).
|
||||||
|
|
||||||
|
t_query_multiple_clients(_) ->
|
||||||
|
process_flag(trap_exit, true),
|
||||||
|
ClientIdsUsers = [
|
||||||
|
{<<"multi_client1">>, <<"multi_user1">>},
|
||||||
|
{<<"multi_client1-1">>, <<"multi_user1">>},
|
||||||
|
{<<"multi_client2">>, <<"multi_user2">>},
|
||||||
|
{<<"multi_client2-1">>, <<"multi_user2">>},
|
||||||
|
{<<"multi_client3">>, <<"multi_user3">>},
|
||||||
|
{<<"multi_client3-1">>, <<"multi_user3">>},
|
||||||
|
{<<"multi_client4">>, <<"multi_user4">>},
|
||||||
|
{<<"multi_client4-1">>, <<"multi_user4">>}
|
||||||
|
],
|
||||||
|
_Clients = lists:map(
|
||||||
|
fun({ClientId, Username}) ->
|
||||||
|
{ok, C} = emqtt:start_link(#{clientid => ClientId, username => Username}),
|
||||||
|
{ok, _} = emqtt:connect(C),
|
||||||
|
C
|
||||||
|
end,
|
||||||
|
ClientIdsUsers
|
||||||
|
),
|
||||||
|
timer:sleep(100),
|
||||||
|
|
||||||
|
Auth = emqx_mgmt_api_test_util:auth_header_(),
|
||||||
|
|
||||||
|
%% Not found clients/users
|
||||||
|
?assertEqual([], get_clients(Auth, "clientid=no_such_client")),
|
||||||
|
?assertEqual([], get_clients(Auth, "clientid=no_such_client&clientid=no_such_client1")),
|
||||||
|
%% Duplicates must cause no issues
|
||||||
|
?assertEqual([], get_clients(Auth, "clientid=no_such_client&clientid=no_such_client")),
|
||||||
|
?assertEqual([], get_clients(Auth, "username=no_such_user&clientid=no_such_user1")),
|
||||||
|
?assertEqual([], get_clients(Auth, "username=no_such_user&clientid=no_such_user")),
|
||||||
|
?assertEqual(
|
||||||
|
[],
|
||||||
|
get_clients(
|
||||||
|
Auth,
|
||||||
|
"clientid=no_such_client&clientid=no_such_client"
|
||||||
|
"username=no_such_user&clientid=no_such_user1"
|
||||||
|
)
|
||||||
|
),
|
||||||
|
|
||||||
|
%% Requested ClientId / username values relate to different clients
|
||||||
|
?assertEqual([], get_clients(Auth, "clientid=multi_client1&username=multi_user2")),
|
||||||
|
?assertEqual(
|
||||||
|
[],
|
||||||
|
get_clients(
|
||||||
|
Auth,
|
||||||
|
"clientid=multi_client1&clientid=multi_client1-1"
|
||||||
|
"&username=multi_user2&username=multi_user3"
|
||||||
|
)
|
||||||
|
),
|
||||||
|
?assertEqual([<<"multi_client1">>], get_clients(Auth, "clientid=multi_client1")),
|
||||||
|
%% Duplicates must cause no issues
|
||||||
|
?assertEqual(
|
||||||
|
[<<"multi_client1">>], get_clients(Auth, "clientid=multi_client1&clientid=multi_client1")
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
[<<"multi_client1">>], get_clients(Auth, "clientid=multi_client1&username=multi_user1")
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(get_clients(Auth, "username=multi_user1"))
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(get_clients(Auth, "clientid=multi_client1&clientid=multi_client1-1"))
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(
|
||||||
|
get_clients(
|
||||||
|
Auth,
|
||||||
|
"clientid=multi_client1&clientid=multi_client1-1"
|
||||||
|
"&username=multi_user1"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(
|
||||||
|
get_clients(
|
||||||
|
Auth,
|
||||||
|
"clientid=no-such-client&clientid=multi_client1&clientid=multi_client1-1"
|
||||||
|
"&username=multi_user1"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(
|
||||||
|
get_clients(
|
||||||
|
Auth,
|
||||||
|
"clientid=no-such-client&clientid=multi_client1&clientid=multi_client1-1"
|
||||||
|
"&username=multi_user1&username=no-such-user"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
),
|
||||||
|
|
||||||
|
AllQsFun = fun(QsKey, Pos) ->
|
||||||
|
QsParts = [
|
||||||
|
QsKey ++ "=" ++ binary_to_list(element(Pos, ClientUser))
|
||||||
|
|| ClientUser <- ClientIdsUsers
|
||||||
|
],
|
||||||
|
lists:flatten(lists:join("&", QsParts))
|
||||||
|
end,
|
||||||
|
AllClientsQs = AllQsFun("clientid", 1),
|
||||||
|
AllUsersQs = AllQsFun("username", 2),
|
||||||
|
AllClientIds = lists:sort([C || {C, _U} <- ClientIdsUsers]),
|
||||||
|
|
||||||
|
?assertEqual(AllClientIds, lists:sort(get_clients(Auth, AllClientsQs))),
|
||||||
|
?assertEqual(AllClientIds, lists:sort(get_clients(Auth, AllUsersQs))),
|
||||||
|
?assertEqual(AllClientIds, lists:sort(get_clients(Auth, AllClientsQs ++ "&" ++ AllUsersQs))),
|
||||||
|
|
||||||
|
%% Test with other filter params
|
||||||
|
NodeQs = "&node=" ++ atom_to_list(node()),
|
||||||
|
NoNodeQs = "&node=nonode@nohost",
|
||||||
|
?assertEqual(
|
||||||
|
AllClientIds, lists:sort(get_clients(Auth, AllClientsQs ++ "&" ++ AllUsersQs ++ NodeQs))
|
||||||
|
),
|
||||||
|
?assertMatch(
|
||||||
|
{error, _}, get_clients_expect_error(Auth, AllClientsQs ++ "&" ++ AllUsersQs ++ NoNodeQs)
|
||||||
|
),
|
||||||
|
|
||||||
|
%% fuzzy search (like_{key}) must be ignored if accurate filter ({key}) is present
|
||||||
|
?assertEqual(
|
||||||
|
AllClientIds,
|
||||||
|
lists:sort(get_clients(Auth, AllClientsQs ++ "&" ++ AllUsersQs ++ "&like_clientid=multi"))
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
AllClientIds,
|
||||||
|
lists:sort(get_clients(Auth, AllClientsQs ++ "&" ++ AllUsersQs ++ "&like_username=multi"))
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
AllClientIds,
|
||||||
|
lists:sort(
|
||||||
|
get_clients(Auth, AllClientsQs ++ "&" ++ AllUsersQs ++ "&like_clientid=does-not-matter")
|
||||||
|
)
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
AllClientIds,
|
||||||
|
lists:sort(
|
||||||
|
get_clients(Auth, AllClientsQs ++ "&" ++ AllUsersQs ++ "&like_username=does-not-matter")
|
||||||
|
)
|
||||||
|
),
|
||||||
|
|
||||||
|
%% Combining multiple clientids with like_username and vice versa must narrow down search results
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(get_clients(Auth, AllClientsQs ++ "&like_username=user1"))
|
||||||
|
),
|
||||||
|
?assertEqual(
|
||||||
|
lists:sort([<<"multi_client1">>, <<"multi_client1-1">>]),
|
||||||
|
lists:sort(get_clients(Auth, AllUsersQs ++ "&like_clientid=client1"))
|
||||||
|
),
|
||||||
|
?assertEqual([], get_clients(Auth, AllClientsQs ++ "&like_username=nouser")),
|
||||||
|
?assertEqual([], get_clients(Auth, AllUsersQs ++ "&like_clientid=nouser")).
|
||||||
|
|
||||||
|
t_query_multiple_clients_urlencode(_) ->
|
||||||
|
process_flag(trap_exit, true),
|
||||||
|
ClientIdsUsers = [
|
||||||
|
{<<"multi_client=a?">>, <<"multi_user=a?">>},
|
||||||
|
{<<"mutli_client=b?">>, <<"multi_user=b?">>}
|
||||||
|
],
|
||||||
|
_Clients = lists:map(
|
||||||
|
fun({ClientId, Username}) ->
|
||||||
|
{ok, C} = emqtt:start_link(#{clientid => ClientId, username => Username}),
|
||||||
|
{ok, _} = emqtt:connect(C),
|
||||||
|
C
|
||||||
|
end,
|
||||||
|
ClientIdsUsers
|
||||||
|
),
|
||||||
|
timer:sleep(100),
|
||||||
|
|
||||||
|
Auth = emqx_mgmt_api_test_util:auth_header_(),
|
||||||
|
ClientsQs = uri_string:compose_query([{<<"clientid">>, C} || {C, _} <- ClientIdsUsers]),
|
||||||
|
UsersQs = uri_string:compose_query([{<<"username">>, U} || {_, U} <- ClientIdsUsers]),
|
||||||
|
ExpectedClients = lists:sort([C || {C, _} <- ClientIdsUsers]),
|
||||||
|
?assertEqual(ExpectedClients, lists:sort(get_clients(Auth, ClientsQs))),
|
||||||
|
?assertEqual(ExpectedClients, lists:sort(get_clients(Auth, UsersQs))).
|
||||||
|
|
||||||
|
t_query_clients_with_fields(_) ->
|
||||||
|
process_flag(trap_exit, true),
|
||||||
|
TCBin = atom_to_binary(?FUNCTION_NAME),
|
||||||
|
ClientId = <<TCBin/binary, "_client">>,
|
||||||
|
Username = <<TCBin/binary, "_user">>,
|
||||||
|
{ok, C} = emqtt:start_link(#{clientid => ClientId, username => Username}),
|
||||||
|
{ok, _} = emqtt:connect(C),
|
||||||
|
timer:sleep(100),
|
||||||
|
|
||||||
|
Auth = emqx_mgmt_api_test_util:auth_header_(),
|
||||||
|
?assertEqual([#{<<"clientid">> => ClientId}], get_clients_all_fields(Auth, "fields=clientid")),
|
||||||
|
?assertEqual(
|
||||||
|
[#{<<"clientid">> => ClientId, <<"username">> => Username}],
|
||||||
|
get_clients_all_fields(Auth, "fields=clientid,username")
|
||||||
|
),
|
||||||
|
|
||||||
|
AllFields = get_clients_all_fields(Auth, "fields=all"),
|
||||||
|
DefaultFields = get_clients_all_fields(Auth, ""),
|
||||||
|
|
||||||
|
?assertEqual(AllFields, DefaultFields),
|
||||||
|
?assertMatch(
|
||||||
|
[#{<<"clientid">> := ClientId, <<"username">> := Username}],
|
||||||
|
AllFields
|
||||||
|
),
|
||||||
|
?assert(map_size(hd(AllFields)) > 2),
|
||||||
|
?assertMatch({error, _}, get_clients_expect_error(Auth, "fields=bad_field_name")),
|
||||||
|
?assertMatch({error, _}, get_clients_expect_error(Auth, "fields=all,bad_field_name")),
|
||||||
|
?assertMatch({error, _}, get_clients_expect_error(Auth, "fields=all,username,clientid")).
|
||||||
|
|
||||||
|
get_clients_all_fields(Auth, Qs) ->
|
||||||
|
get_clients(Auth, Qs, false, false).
|
||||||
|
|
||||||
|
get_clients_expect_error(Auth, Qs) ->
|
||||||
|
get_clients(Auth, Qs, true, true).
|
||||||
|
|
||||||
|
get_clients(Auth, Qs) ->
|
||||||
|
get_clients(Auth, Qs, false, true).
|
||||||
|
|
||||||
|
get_clients(Auth, Qs, ExpectError, ClientIdOnly) ->
|
||||||
|
ClientsPath = emqx_mgmt_api_test_util:api_path(["clients"]),
|
||||||
|
Resp = emqx_mgmt_api_test_util:request_api(get, ClientsPath, Qs, Auth),
|
||||||
|
case ExpectError of
|
||||||
|
false ->
|
||||||
|
{ok, Body} = Resp,
|
||||||
|
#{<<"data">> := Clients} = emqx_utils_json:decode(Body),
|
||||||
|
case ClientIdOnly of
|
||||||
|
true -> [ClientId || #{<<"clientid">> := ClientId} <- Clients];
|
||||||
|
false -> Clients
|
||||||
|
end;
|
||||||
|
true ->
|
||||||
|
Resp
|
||||||
|
end.
|
||||||
|
|
||||||
t_keepalive(_Config) ->
|
t_keepalive(_Config) ->
|
||||||
Username = "user_keepalive",
|
Username = "user_keepalive",
|
||||||
ClientId = "client_keepalive",
|
ClientId = "client_keepalive",
|
||||||
|
|
|
@ -0,0 +1,12 @@
|
||||||
|
## Support multiple clientid and username Query string parameters in "/clients" API
|
||||||
|
|
||||||
|
Multi clientid/username queries examples:
|
||||||
|
- "/clients?clientid=client1&clientid=client2
|
||||||
|
- "/clients?username=user11&username=user2"
|
||||||
|
- "/clients?clientid=client1&clientid=client2&username=user1&username=user2"
|
||||||
|
|
||||||
|
## Add an option to specify which client info fields must be included in the response
|
||||||
|
|
||||||
|
Request response fields examples:
|
||||||
|
- "/clients?fields=all" (omitting "fields" Qs parameter defaults to returning all fields)
|
||||||
|
- "/clients?fields=clientid,username"
|
2
mix.exs
2
mix.exs
|
@ -58,7 +58,7 @@ defmodule EMQXUmbrella.MixProject do
|
||||||
{:ekka, github: "emqx/ekka", tag: "0.19.0", override: true},
|
{:ekka, github: "emqx/ekka", tag: "0.19.0", override: true},
|
||||||
{:gen_rpc, github: "emqx/gen_rpc", tag: "3.3.1", override: true},
|
{:gen_rpc, github: "emqx/gen_rpc", tag: "3.3.1", override: true},
|
||||||
{:grpc, github: "emqx/grpc-erl", tag: "0.6.12", override: true},
|
{:grpc, github: "emqx/grpc-erl", tag: "0.6.12", override: true},
|
||||||
{:minirest, github: "emqx/minirest", tag: "1.3.15", override: true},
|
{:minirest, github: "emqx/minirest", tag: "1.4.0", override: true},
|
||||||
{:ecpool, github: "emqx/ecpool", tag: "0.5.7", override: true},
|
{:ecpool, github: "emqx/ecpool", tag: "0.5.7", override: true},
|
||||||
{:replayq, github: "emqx/replayq", tag: "0.3.7", override: true},
|
{:replayq, github: "emqx/replayq", tag: "0.3.7", override: true},
|
||||||
{:pbkdf2, github: "emqx/erlang-pbkdf2", tag: "2.0.4", override: true},
|
{:pbkdf2, github: "emqx/erlang-pbkdf2", tag: "2.0.4", override: true},
|
||||||
|
|
|
@ -86,7 +86,7 @@
|
||||||
{ekka, {git, "https://github.com/emqx/ekka", {tag, "0.19.0"}}},
|
{ekka, {git, "https://github.com/emqx/ekka", {tag, "0.19.0"}}},
|
||||||
{gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.1"}}},
|
{gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.3.1"}}},
|
||||||
{grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.12"}}},
|
{grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.12"}}},
|
||||||
{minirest, {git, "https://github.com/emqx/minirest", {tag, "1.3.15"}}},
|
{minirest, {git, "https://github.com/emqx/minirest", {tag, "1.4.0"}}},
|
||||||
{ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.7"}}},
|
{ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.7"}}},
|
||||||
{replayq, {git, "https://github.com/emqx/replayq.git", {tag, "0.3.7"}}},
|
{replayq, {git, "https://github.com/emqx/replayq.git", {tag, "0.3.7"}}},
|
||||||
{pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}},
|
{pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}},
|
||||||
|
|
Loading…
Reference in New Issue