From 9bb08686280ef42ec146f68956d33f41b47407de Mon Sep 17 00:00:00 2001 From: Andrew Mayorov Date: Thu, 8 Feb 2024 13:23:58 +0100 Subject: [PATCH 1/3] test(api-topics): add testcase verifying paged queries work --- apps/emqx_management/src/emqx_mgmt_api.erl | 4 +- .../test/emqx_mgmt_api_topics_SUITE.erl | 143 +++++++++++++----- 2 files changed, 102 insertions(+), 45 deletions(-) diff --git a/apps/emqx_management/src/emqx_mgmt_api.erl b/apps/emqx_management/src/emqx_mgmt_api.erl index 6bf9d5da1..be8f24bc3 100644 --- a/apps/emqx_management/src/emqx_mgmt_api.erl +++ b/apps/emqx_management/src/emqx_mgmt_api.erl @@ -166,9 +166,7 @@ node_query(Node, Tab, QString, QSchema, MsFun, FmtFun, Options) -> {_CodCnt, NQString} = parse_qstring(QString, QSchema), ResultAcc = init_query_result(), QueryState = init_query_state(Tab, NQString, MsFun, Meta, Options), - NResultAcc = do_node_query( - Node, QueryState, ResultAcc - ), + NResultAcc = do_node_query(Node, QueryState, ResultAcc), format_query_result(FmtFun, Meta, NResultAcc) end. diff --git a/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl index e2ab34010..bbddde749 100644 --- a/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl +++ b/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl @@ -18,7 +18,6 @@ -compile(export_all). -compile(nowarn_export_all). --include_lib("emqx/include/emqx_router.hrl"). -include_lib("eunit/include/eunit.hrl"). -include_lib("common_test/include/ct.hrl"). @@ -26,48 +25,44 @@ all() -> emqx_common_test_helpers:all(?MODULE). init_per_suite(Config) -> - emqx_mgmt_api_test_util:init_suite(), + Apps = emqx_cth_suite:start( + [ + emqx, + emqx_management, + emqx_mgmt_api_test_util:emqx_dashboard() + ], + #{work_dir => emqx_cth_suite:work_dir(Config)} + ), Peer = emqx_common_test_helpers:start_peer(node1, []), - [{peer, Peer} | Config]. + [{apps, Apps}, {peer, Peer} | Config]. end_per_suite(Config) -> - Peer = ?config(peer, Config), - emqx_common_test_helpers:stop_peer(Peer), - mria:clear_table(?ROUTE_TAB), - emqx_mgmt_api_test_util:end_suite(). + _ = emqx_common_test_helpers:stop_peer(?config(peer, Config)), + ok = emqx_cth_suite:stop(?config(apps, Config)). t_nodes_api(Config) -> Node = atom_to_binary(node(), utf8), Topic = <<"test_topic">>, - {ok, Client} = emqtt:start_link(#{ - username => <<"routes_username">>, clientid => <<"routes_cid">> - }), - {ok, _} = emqtt:connect(Client), + Client = client(?FUNCTION_NAME), {ok, _, _} = emqtt:subscribe(Client, Topic), %% list all - Path = emqx_mgmt_api_test_util:api_path(["topics"]), - {ok, Response} = emqx_mgmt_api_test_util:request_api(get, Path), - RoutesData = emqx_utils_json:decode(Response, [return_maps]), + RoutesData = request_json(get, ["topics"]), Meta = maps:get(<<"meta">>, RoutesData), ?assertEqual(1, maps:get(<<"page">>, Meta)), ?assertEqual(emqx_mgmt:default_row_limit(), maps:get(<<"limit">>, Meta)), ?assertEqual(1, maps:get(<<"count">>, Meta)), - Data = maps:get(<<"data">>, RoutesData), - Route = erlang:hd(Data), + [Route | _] = maps:get(<<"data">>, RoutesData), ?assertEqual(Topic, maps:get(<<"topic">>, Route)), ?assertEqual(Node, maps:get(<<"node">>, Route)), %% exact match Topic2 = <<"test_topic_2">>, {ok, _, _} = emqtt:subscribe(Client, Topic2), - QS = uri_string:compose_query([ + MatchData = request_json(get, ["topics"], [ {"topic", Topic2}, {"node", atom_to_list(node())} ]), - Headers = emqx_mgmt_api_test_util:auth_header_(), - {ok, MatchResponse} = emqx_mgmt_api_test_util:request_api(get, Path, QS, Headers), - MatchData = emqx_utils_json:decode(MatchResponse, [return_maps]), ?assertMatch( #{<<"count">> := 1, <<"page">> := 1, <<"limit">> := 100}, maps:get(<<"meta">>, MatchData) @@ -82,37 +77,78 @@ t_nodes_api(Config) -> %% multiple routes for a single topic Peer = ?config(peer, Config), ok = emqx_router:add_route(Topic, Peer), - RoutePath = emqx_mgmt_api_test_util:api_path(["topics", Topic]), - {ok, RouteResponse} = emqx_mgmt_api_test_util:request_api(get, RoutePath), + RouteResponse = request_json(get, ["topics", Topic]), ok = emqx_router:delete_route(Topic, Peer), [ #{<<"topic">> := Topic, <<"node">> := Node1}, #{<<"topic">> := Topic, <<"node">> := Node2} - ] = emqx_utils_json:decode(RouteResponse, [return_maps]), + ] = RouteResponse, - ?assertEqual(lists:usort([Node, atom_to_binary(Peer)]), lists:usort([Node1, Node2])), + ?assertEqual(lists:sort([Node, atom_to_binary(Peer)]), lists:sort([Node1, Node2])), ok = emqtt:stop(Client). +t_paging(_Config) -> + Node = atom_to_list(node()), + Client1 = client(c_paging_1), + Client2 = client(c_paging_2), + Topics1 = [ + <<"t/+">>, + <<"test/client/#">>, + <<"test/1">>, + <<"test/2">>, + <<"test/3">> + ], + Topics2 = [ + <<"t/+">>, + <<"test/client/#">>, + <<"test/4">>, + <<"test/5">>, + <<"test/6">> + ], + ok = lists:foreach(fun(T) -> {ok, _, _} = emqtt:subscribe(Client1, T) end, Topics1), + ok = lists:foreach(fun(T) -> {ok, _, _} = emqtt:subscribe(Client2, T) end, Topics2), + Matched = request_json(get, ["topics"]), + ?assertEqual( + Matched, + request_json(get, ["topics"], [{"node", Node}]) + ), + R1 = #{<<"data">> := Data1} = request_json(get, ["topics"], [{"page", "1"}, {"limit", "5"}]), + R2 = #{<<"data">> := Data2} = request_json(get, ["topics"], [{"page", "2"}, {"limit", "5"}]), + ?assertMatch( + #{ + <<"meta">> := #{<<"hasnext">> := true, <<"page">> := 1, <<"count">> := 8}, + <<"data">> := [_1, _2, _3, _4, _5] + }, + R1 + ), + ?assertMatch( + #{ + <<"meta">> := #{<<"hasnext">> := false, <<"page">> := 2, <<"count">> := 8}, + <<"data">> := [_6, _7, _8] + }, + R2 + ), + ?assertEqual( + lists:usort(Topics1 ++ Topics2), + lists:sort([T || #{<<"topic">> := T} <- Data1 ++ Data2]) + ), + + ok = emqtt:stop(Client1), + ok = emqtt:stop(Client2). + t_percent_topics(_Config) -> Node = atom_to_binary(node(), utf8), Topic = <<"test_%%1">>, - {ok, Client} = emqtt:start_link(#{ - username => <<"routes_username">>, clientid => <<"routes_cid">> - }), - {ok, _} = emqtt:connect(Client), + Client = client(?FUNCTION_NAME), {ok, _, _} = emqtt:subscribe(Client, Topic), %% exact match with percent encoded topic - Path = emqx_mgmt_api_test_util:api_path(["topics"]), - QS = uri_string:compose_query([ + MatchData = request_json(get, ["topics"], [ {"topic", Topic}, {"node", atom_to_list(node())} ]), - Headers = emqx_mgmt_api_test_util:auth_header_(), - {ok, MatchResponse} = emqx_mgmt_api_test_util:request_api(get, Path, QS, Headers), - MatchData = emqx_utils_json:decode(MatchResponse, [return_maps]), ?assertMatch( #{<<"count">> := 1, <<"page">> := 1, <<"limit">> := 100}, maps:get(<<"meta">>, MatchData) @@ -129,22 +165,15 @@ t_shared_topics(_Configs) -> RealTopic = <<"t/+">>, Topic = <<"$share/g1/", RealTopic/binary>>, - {ok, Client} = emqtt:start_link(#{ - username => <<"routes_username">>, clientid => <<"routes_cid">> - }), - {ok, _} = emqtt:connect(Client), + Client = client(?FUNCTION_NAME), {ok, _, _} = emqtt:subscribe(Client, Topic), {ok, _, _} = emqtt:subscribe(Client, RealTopic), %% exact match with shared topic - Path = emqx_mgmt_api_test_util:api_path(["topics"]), - QS = uri_string:compose_query([ + MatchData = request_json(get, ["topics"], [ {"topic", Topic}, {"node", atom_to_list(node())} ]), - Headers = emqx_mgmt_api_test_util:auth_header_(), - {ok, MatchResponse1} = emqx_mgmt_api_test_util:request_api(get, Path, QS, Headers), - MatchData = emqx_utils_json:decode(MatchResponse1, [return_maps]), ?assertMatch( #{<<"count">> := 1, <<"page">> := 1, <<"limit">> := 100}, maps:get(<<"meta">>, MatchData) @@ -172,3 +201,33 @@ t_shared_topics_invalid(_Config) -> #{<<"code">> := <<"INVALID_PARAMTER">>, <<"message">> := <<"topic_filter_invalid">>}, emqx_utils_json:decode(Body, [return_maps]) ). + +%% Utilities + +client(Name) -> + {ok, Client} = emqtt:start_link(#{ + username => emqx_utils_conv:bin(Name), + clientid => emqx_utils_conv:bin(Name) + }), + {ok, _} = emqtt:connect(Client), + Client. + +request_json(Method, Path) -> + decode_response(request_api(Method, Path)). + +request_json(Method, Path, QS) -> + decode_response(request_api(Method, Path, QS)). + +decode_response({ok, Response}) -> + emqx_utils_json:decode(Response, [return_maps]); +decode_response({error, Reason}) -> + error({request_api_error, Reason}). + +request_api(Method, API) -> + Path = emqx_mgmt_api_test_util:api_path(API), + emqx_mgmt_api_test_util:request_api(Method, Path). + +request_api(Method, API, QS) -> + Path = emqx_mgmt_api_test_util:api_path(API), + Auth = emqx_mgmt_api_test_util:auth_header_(), + emqx_mgmt_api_test_util:request_api(Method, Path, uri_string:compose_query(QS), Auth). From bd578a799efc2edb6096ede684403cf525dede04 Mon Sep 17 00:00:00 2001 From: Andrew Mayorov Date: Thu, 8 Feb 2024 13:28:15 +0100 Subject: [PATCH 2/3] fix(api-topics): avoid doing full scans over router tables --- apps/emqx_management/src/emqx_mgmt_api_topics.erl | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/apps/emqx_management/src/emqx_mgmt_api_topics.erl b/apps/emqx_management/src/emqx_mgmt_api_topics.erl index 75fa789bd..ed2de351b 100644 --- a/apps/emqx_management/src/emqx_mgmt_api_topics.erl +++ b/apps/emqx_management/src/emqx_mgmt_api_topics.erl @@ -164,14 +164,17 @@ eval_topic_query(MS, QState) -> finalize_query(eval_topic_query(MS, QState, emqx_mgmt_api:init_query_result())). eval_topic_query(MS, QState, QResult) -> - QPage = eval_topic_query_page(MS, QState), - case QPage of + case eval_topic_query_page(MS, QState) of {Rows, '$end_of_table'} -> {_, NQResult} = emqx_mgmt_api:accumulate_query_rows(node(), Rows, QState, QResult), NQResult#{complete => true}; {Rows, NCont} -> - {_, NQResult} = emqx_mgmt_api:accumulate_query_rows(node(), Rows, QState, QResult), - eval_topic_query(MS, QState#{continuation := NCont}, NQResult); + case emqx_mgmt_api:accumulate_query_rows(node(), Rows, QState, QResult) of + {more, NQResult} -> + eval_topic_query(MS, QState#{continuation := NCont}, NQResult); + {enough, NQResult} -> + NQResult#{complete => false} + end; '$end_of_table' -> QResult#{complete => true} end. From c3b044a37b1b177423d822af259f145af3daf71a Mon Sep 17 00:00:00 2001 From: Andrew Mayorov Date: Wed, 28 Feb 2024 11:28:33 +0100 Subject: [PATCH 3/3] fix(api-topics): respond with `count` when it's cheap to compute --- .../src/emqx_mgmt_api_topics.erl | 13 +++++-- .../test/emqx_mgmt_api_topics_SUITE.erl | 38 ++++++++++--------- 2 files changed, 30 insertions(+), 21 deletions(-) diff --git a/apps/emqx_management/src/emqx_mgmt_api_topics.erl b/apps/emqx_management/src/emqx_mgmt_api_topics.erl index ed2de351b..9ab111fed 100644 --- a/apps/emqx_management/src/emqx_mgmt_api_topics.erl +++ b/apps/emqx_management/src/emqx_mgmt_api_topics.erl @@ -115,7 +115,7 @@ do_list(Params) -> {_, Query} = emqx_mgmt_api:parse_qstring(Params, ?TOPICS_QUERY_SCHEMA), QState = Pager#{continuation => undefined}, QResult = eval_topic_query(qs2ms(Query), QState), - {200, format_list_response(Pager, QResult)} + {200, format_list_response(Pager, Query, QResult)} catch throw:{error, page_limit_invalid} -> {400, #{code => <<"INVALID_PARAMETER">>, message => <<"page_limit_invalid">>}}; @@ -186,15 +186,22 @@ finalize_query(QResult = #{overflow := Overflow, complete := Complete}) -> HasNext = Overflow orelse not Complete, QResult#{hasnext => HasNext}. -format_list_response(Meta, _QResult = #{hasnext := HasNext, rows := RowsAcc, cursor := Cursor}) -> +format_list_response(Meta, Query, QResult = #{rows := RowsAcc}) -> #{ - meta => Meta#{hasnext => HasNext, count => Cursor}, + meta => format_response_meta(Meta, Query, QResult), data => lists:flatmap( fun({_Node, Rows}) -> [format(R) || R <- Rows] end, RowsAcc ) }. +format_response_meta(Meta, _Query, #{hasnext := HasNext, complete := true, cursor := Cursor}) -> + Meta#{hasnext => HasNext, count => Cursor}; +format_response_meta(Meta, _Query = {[], []}, #{hasnext := HasNext}) -> + Meta#{hasnext => HasNext, count => emqx_router:stats(n_routes)}; +format_response_meta(Meta, _Query, #{hasnext := HasNext}) -> + Meta#{hasnext => HasNext}. + format(#route{topic = Topic, dest = {Group, Node}}) -> #{topic => ?SHARE(Group, Topic), node => Node}; format(#route{topic = Topic, dest = Node}) when is_atom(Node) -> diff --git a/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl index bbddde749..8b362af1b 100644 --- a/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl +++ b/apps/emqx_management/test/emqx_mgmt_api_topics_SUITE.erl @@ -64,12 +64,11 @@ t_nodes_api(Config) -> {"node", atom_to_list(node())} ]), ?assertMatch( - #{<<"count">> := 1, <<"page">> := 1, <<"limit">> := 100}, - maps:get(<<"meta">>, MatchData) - ), - ?assertMatch( - [#{<<"topic">> := Topic2, <<"node">> := Node}], - maps:get(<<"data">>, MatchData) + #{ + <<"data">> := [#{<<"topic">> := Topic2, <<"node">> := Node}], + <<"meta">> := #{<<"page">> := 1, <<"limit">> := 100, <<"count">> := 1} + }, + MatchData ), %% get topics/:topic @@ -114,6 +113,11 @@ t_paging(_Config) -> Matched, request_json(get, ["topics"], [{"node", Node}]) ), + ?assertEqual( + %% NOTE: No `count` in this case. + #{<<"hasnext">> => true, <<"page">> => 1, <<"limit">> => 3}, + maps:get(<<"meta">>, request_json(get, ["topics"], [{"node", Node}, {"limit", "3"}])) + ), R1 = #{<<"data">> := Data1} = request_json(get, ["topics"], [{"page", "1"}, {"limit", "5"}]), R2 = #{<<"data">> := Data2} = request_json(get, ["topics"], [{"page", "2"}, {"limit", "5"}]), ?assertMatch( @@ -150,12 +154,11 @@ t_percent_topics(_Config) -> {"node", atom_to_list(node())} ]), ?assertMatch( - #{<<"count">> := 1, <<"page">> := 1, <<"limit">> := 100}, - maps:get(<<"meta">>, MatchData) - ), - ?assertMatch( - [#{<<"topic">> := Topic, <<"node">> := Node}], - maps:get(<<"data">>, MatchData) + #{ + <<"data">> := [#{<<"topic">> := Topic, <<"node">> := Node}], + <<"meta">> := #{<<"page">> := 1, <<"limit">> := 100, <<"count">> := 1} + }, + MatchData ), ok = emqtt:stop(Client). @@ -175,12 +178,11 @@ t_shared_topics(_Configs) -> {"node", atom_to_list(node())} ]), ?assertMatch( - #{<<"count">> := 1, <<"page">> := 1, <<"limit">> := 100}, - maps:get(<<"meta">>, MatchData) - ), - ?assertMatch( - [#{<<"topic">> := Topic, <<"node">> := Node}], - maps:get(<<"data">>, MatchData) + #{ + <<"data">> := [#{<<"topic">> := Topic, <<"node">> := Node}], + <<"meta">> := #{<<"page">> := 1, <<"limit">> := 100, <<"count">> := 1} + }, + MatchData ), ok = emqtt:stop(Client).