fix(clients v2 mgmt api): support exact username filtering

Fixes https://emqx.atlassian.net/browse/EMQX-12873

apps/emqx_management/src/emqx_mgmt_api_clients.erl

@@ -1043,7 +1043,8 @@ do_ets_select(Nodes, QString0, #{node := Node, node_idx := NodeIdx, cont := Cont
     {Rows, next_ets_cursor(Nodes, NewNodeIdx, NewCont)}.
 
 maybe_run_fuzzy_filter(Rows, QString0) ->
-    {_, {_, FuzzyQString}} = emqx_mgmt_api:parse_qstring(QString0, ?CLIENT_QSCHEMA),
+    {_NClauses, {QString1, FuzzyQString1}} = emqx_mgmt_api:parse_qstring(QString0, ?CLIENT_QSCHEMA),
+    {_QString, FuzzyQString} = adapt_custom_filters(QString1, FuzzyQString1),
     FuzzyFilterFn = fuzzy_filter_fun(FuzzyQString),
     case FuzzyFilterFn of
         undefined ->
@@ -1055,6 +1056,27 @@ maybe_run_fuzzy_filter(Rows, QString0) ->
             )
     end.
 
+%% These filters, while they are able to be adapted to efficient ETS match specs, must be
+%% used as fuzzy filters when iterating over offlient persistent clients, which live
+%% outside ETS.
+adapt_custom_filters(Qs, Fuzzy) ->
+    lists:foldl(
+        fun
+            ({Field, '=:=', X}, {QsAcc, FuzzyAcc}) when
+                Field =:= username
+            ->
+                Xs = wrap(X),
+                {QsAcc, [{Field, in, Xs} | FuzzyAcc]};
+            (Clause, {QsAcc, FuzzyAcc}) ->
+                {[Clause | QsAcc], FuzzyAcc}
+        end,
+        {[], Fuzzy},
+        Qs
+    ).
+
+wrap(Xs) when is_list(Xs) -> Xs;
+wrap(X) -> [X].
+
 initial_ets_cursor([Node | _Rest] = _Nodes) ->
     #{
         type => ?CURSOR_TYPE_ETS,
@@ -1611,8 +1633,8 @@ qs2ms(_Tab, {QString, FuzzyQString}) ->
 
 -spec qs2ms(list()) -> ets:match_spec().
 qs2ms(Qs) ->
-    {MtchHead, Conds} = qs2ms(Qs, 2, {#{}, []}),
-    [{{{'$1', '_'}, MtchHead, '_'}, Conds, ['$_']}].
+    {MatchHead, Conds} = qs2ms(Qs, 2, {#{}, []}),
+    [{{{'$1', '_'}, MatchHead, '_'}, Conds, ['$_']}].
 
 qs2ms([], _, {MtchHead, Conds}) ->
     {MtchHead, lists:reverse(Conds)};
@@ -1685,6 +1707,20 @@ run_fuzzy_filter(
     %% Row from DS
     run_fuzzy_filter1(ClientInfo, Key, SubStr) andalso
         run_fuzzy_filter(Row, RestArgs);
+run_fuzzy_filter(
+    Row = {_, #{metadata := #{clientinfo := ClientInfo}}},
+    [{Key, in, Xs} | RestArgs]
+) ->
+    %% Row from DS
+    IsMatch =
+        case maps:find(Key, ClientInfo) of
+            {ok, X} ->
+                lists:member(X, Xs);
+            error ->
+                false
+        end,
+    IsMatch andalso
+        run_fuzzy_filter(Row, RestArgs);
 run_fuzzy_filter(Row = {_, #{clientinfo := ClientInfo}, _}, [{Key, like, SubStr} | RestArgs]) ->
     %% Row from ETS
     run_fuzzy_filter1(ClientInfo, Key, SubStr) andalso

apps/emqx_management/test/emqx_mgmt_api_clients_SUITE.erl

@@ -31,6 +31,9 @@
 -define(HTTP400, {"HTTP/1.1", 400, "Bad Request"}).
 -define(HTTP404, {"HTTP/1.1", 404, "Not Found"}).
 
+-define(ON(NODE, BODY), erpc:call(NODE, fun() -> BODY end)).
+-define(assertContainsClientids(RES, EXPECTED), assert_contains_clientids(RES, EXPECTED, ?LINE)).
+
 all() ->
     [
         {group, general},
@@ -65,7 +68,8 @@ persistent_session_testcases() ->
         t_persistent_sessions5,
         t_persistent_sessions6,
         t_persistent_sessions_subscriptions1,
-        t_list_clients_v2
+        t_list_clients_v2,
+        t_list_clients_v2_exact_filters
     ].
 non_persistent_cluster_testcases() ->
     [
@@ -1683,7 +1687,7 @@ t_list_clients_v2(Config) ->
                 ],
                 Res1
             ),
-            assert_contains_clientids(Res1, AllClientIds),
+            ?assertContainsClientids(Res1, AllClientIds),
 
             %% Reusing the same cursors yield the same pages
             traverse_in_reverse_v2(QueryParams1, Res1, Config),
@@ -1713,7 +1717,7 @@ t_list_clients_v2(Config) ->
                 ],
                 Res2
             ),
-            assert_contains_clientids(Res2, AllClientIds),
+            ?assertContainsClientids(Res2, AllClientIds),
             traverse_in_reverse_v2(QueryParams2, Res2, Config),
 
             QueryParams3 = #{limit => "2"},
@@ -1749,7 +1753,7 @@ t_list_clients_v2(Config) ->
                 ],
                 Res3
             ),
-            assert_contains_clientids(Res3, AllClientIds),
+            ?assertContainsClientids(Res3, AllClientIds),
             traverse_in_reverse_v2(QueryParams3, Res3, Config),
 
             %% fuzzy filters
@@ -1768,7 +1772,7 @@ t_list_clients_v2(Config) ->
                 ],
                 Res4
             ),
-            assert_contains_clientids(Res4, [ClientId1, ClientId4, ClientId5]),
+            ?assertContainsClientids(Res4, [ClientId1, ClientId4, ClientId5]),
             traverse_in_reverse_v2(QueryParams4, Res4, Config),
             QueryParams5 = #{limit => "1", like_clientid => "ca"},
             Res5 = list_all_v2(QueryParams5, Config),
@@ -1803,7 +1807,7 @@ t_list_clients_v2(Config) ->
                 ],
                 Res5
             ),
-            assert_contains_clientids(Res5, [ClientId1, ClientId4, ClientId5]),
+            ?assertContainsClientids(Res5, [ClientId1, ClientId4, ClientId5]),
             traverse_in_reverse_v2(QueryParams5, Res5, Config),
 
             lists:foreach(
@@ -1845,6 +1849,82 @@ t_list_clients_v2(Config) ->
     ),
     ok.
 
+%% Checks that exact match filters (username) works in clients_v2 API.
+t_list_clients_v2_exact_filters(Config) ->
+    [N1, N2] = ?config(nodes, Config),
+    Port1 = get_mqtt_port(N1, tcp),
+    Port2 = get_mqtt_port(N2, tcp),
+    Id = fun(Bin) -> iolist_to_binary([atom_to_binary(?FUNCTION_NAME), <<"-">>, Bin]) end,
+    ?check_trace(
+        begin
+            ClientId1 = Id(<<"ps1">>),
+            ClientId2 = Id(<<"ps2">>),
+            ClientId3 = Id(<<"ps3-offline">>),
+            ClientId4 = Id(<<"ps4-offline">>),
+            ClientId5 = Id(<<"mem2">>),
+            ClientId6 = Id(<<"mem3">>),
+            Username1 = Id(<<"u1">>),
+            Username2 = Id(<<"u2">>),
+            C1 = connect_client(#{
+                port => Port1,
+                clientid => ClientId1,
+                clean_start => true,
+                username => Username1
+            }),
+            C2 = connect_client(#{
+                port => Port2,
+                clientid => ClientId2,
+                clean_start => true,
+                username => Username2
+            }),
+            C3 = connect_client(#{port => Port1, clientid => ClientId3, clean_start => true}),
+            C4 = connect_client(#{port => Port2, clientid => ClientId4, clean_start => true}),
+            %% in-memory clients
+            C5 = connect_client(#{
+                port => Port1,
+                clientid => ClientId5,
+                expiry => 0,
+                clean_start => true,
+                username => Username1
+            }),
+            C6 = connect_client(#{
+                port => Port2,
+                clientid => ClientId6,
+                expiry => 0,
+                clean_start => true,
+                username => Username2
+            }),
+            %% offline persistent clients
+            lists:foreach(fun stop_and_commit/1, [C3, C4]),
+
+            %% Username query
+            QueryParams1 = [
+                {"limit", "100"},
+                {"username", Username1}
+            ],
+            Res1 = list_all_v2(QueryParams1, Config),
+            ?assertContainsClientids(Res1, [ClientId1, ClientId5]),
+
+            QueryParams2 = [
+                {"limit", "100"},
+                {"username", Username1},
+                {"username", Username2}
+            ],
+            Res2 = list_all_v2(QueryParams2, Config),
+            ?assertContainsClientids(Res2, [ClientId1, ClientId2, ClientId5, ClientId6]),
+
+            C3B = connect_client(#{port => Port1, clientid => ClientId3}),
+            C4B = connect_client(#{port => Port2, clientid => ClientId4}),
+
+            lists:foreach(fun stop_and_commit/1, [C1, C2, C3B, C4B]),
+            lists:foreach(fun emqtt:stop/1, [C5, C6]),
+
+            ok
+        end,
+        []
+    ),
+    ok.
+
 t_cursor_serde_prop(_Config) ->
     ?assert(proper:quickcheck(cursor_serde_prop(), [{numtests, 100}, {to_file, user}])).
 
@@ -1989,11 +2069,11 @@ simplify_result(Res) ->
             {Status, Body}
     end.
 
-list_v2_request(QueryParams = #{}, Config) ->
+list_v2_request(QueryParams, Config) ->
     Path = emqx_mgmt_api_test_util:api_path(["clients_v2"]),
     request(get, Path, [], compose_query_string(QueryParams), Config).
 
-list_all_v2(QueryParams = #{}, Config) ->
+list_all_v2(QueryParams, Config) ->
     do_list_all_v2(QueryParams, Config, _Acc = []).
 
 do_list_all_v2(QueryParams, Config, Acc) ->
@@ -2018,8 +2098,10 @@ lookup_request(ClientId, Config) ->
 
 compose_query_string(QueryParams = #{}) ->
     QPList = maps:to_list(QueryParams),
+    compose_query_string(QPList);
+compose_query_string([{_, _} | _] = QueryParams) ->
     uri_string:compose_query(
-        [{emqx_utils_conv:bin(K), emqx_utils_conv:bin(V)} || {K, V} <- QPList]
+        [{emqx_utils_conv:bin(K), emqx_utils_conv:bin(V)} || {K, V} <- QueryParams]
     );
 compose_query_string(QueryString) when is_list(QueryString) ->
     QueryString.
@@ -2081,22 +2163,23 @@ connect_client(Opts) ->
         clean_start => false
     },
     #{
-        port := Port,
-        clientid := ClientId,
-        clean_start := CleanStart,
+        port := _Port,
+        clientid := _ClientId,
         expiry := EI
-    } = maps:merge(Defaults, Opts),
-    {ok, C} = emqtt:start_link([
-        {port, Port},
-        {proto_ver, v5},
-        {clientid, ClientId},
-        {clean_start, CleanStart},
-        {properties, #{'Session-Expiry-Interval' => EI}}
-    ]),
+    } = ConnOpts0 = maps:merge(Defaults, Opts),
+    ConnOpts1 = maps:without([expiry], ConnOpts0),
+    ConnOpts = emqx_utils_maps:deep_merge(
+        #{
+            proto_ver => v5,
+            properties => #{'Session-Expiry-Interval' => EI}
+        },
+        ConnOpts1
+    ),
+    {ok, C} = emqtt:start_link(ConnOpts),
     {ok, _} = emqtt:connect(C),
     C.
 
-assert_contains_clientids(Results, ExpectedClientIds) ->
+assert_contains_clientids(Results, ExpectedClientIds, Line) ->
     ContainedClientIds = [
         ClientId
      || #{<<"data">> := Rows} <- Results,
@@ -2105,7 +2188,7 @@ assert_contains_clientids(Results, ExpectedClientIds) ->
     ?assertEqual(
         lists:sort(ExpectedClientIds),
         lists:sort(ContainedClientIds),
-        #{results => Results}
+        #{results => Results, line => Line}
     ).
 
 traverse_in_reverse_v2(QueryParams0, Results, Config) ->