feat(cluster link api): add metrics reset endpoint

Part of https://emqx.atlassian.net/browse/EMQX-12942

apps/emqx_cluster_link/src/emqx_cluster_link_api.erl

@@ -23,7 +23,8 @@
 -export([
     '/cluster/links'/2,
     '/cluster/links/link/:name'/2,
-    '/cluster/links/link/:name/metrics'/2
+    '/cluster/links/link/:name/metrics'/2,
+    '/cluster/links/link/:name/metrics/reset'/2
 ]).
 
 -define(CONF_PATH, [cluster, links]).
@@ -40,7 +41,8 @@ paths() ->
     [
         "/cluster/links",
         "/cluster/links/link/:name",
-        "/cluster/links/link/:name/metrics"
+        "/cluster/links/link/:name/metrics",
+        "/cluster/links/link/:name/metrics/reset"
     ].
 
 schema("/cluster/links") ->
@@ -133,6 +135,23 @@ schema("/cluster/links/link/:name/metrics") ->
                         )
                     }
             }
+    };
+schema("/cluster/links/link/:name/metrics/reset") ->
+    #{
+        'operationId' => '/cluster/links/link/:name/metrics/reset',
+        put =>
+            #{
+                description => "Reset a cluster link's metrics",
+                tags => ?TAGS,
+                parameters => [param_path_name()],
+                responses =>
+                    #{
+                        204 => <<"Reset">>,
+                        404 => emqx_dashboard_swagger:error_codes(
+                            [?NOT_FOUND], <<"Cluster link not found">>
+                        )
+                    }
+            }
     }.
 
 fields(link_config_response) ->
@@ -195,6 +214,9 @@ fields(node_metrics) ->
 '/cluster/links/link/:name/metrics'(get, #{bindings := #{name := Name}}) ->
     with_link(Name, fun() -> handle_metrics(Name) end, not_found()).
 
+'/cluster/links/link/:name/metrics/reset'(put, #{bindings := #{name := Name}}) ->
+    with_link(Name, fun() -> handle_reset_metrics(Name) end, not_found()).
+
 %%--------------------------------------------------------------------
 %% Internal funcs
 %%--------------------------------------------------------------------
@@ -351,6 +373,28 @@ format_metrics(Node, RouterMetrics, ResourceMetrics) ->
         }
     }.
 
+handle_reset_metrics(Name) ->
+    Res = emqx_cluster_link_metrics:reset_metrics(Name),
+    ErrorNodes =
+        lists:filtermap(
+            fun
+                ({_Node, {ok, ok}, {ok, ok}}) ->
+                    false;
+                ({Node, _, _}) ->
+                    {true, Node}
+            end,
+            Res
+        ),
+    case ErrorNodes of
+        [] ->
+            ?NO_CONTENT;
+        [_ | _] ->
+            Msg0 = <<"Metrics reset failed on one or more nodes. Please try again.">>,
+            Msg1 = ?ERROR_MSG('INTERNAL_ERROR', Msg0),
+            Msg = Msg1#{nodes => ErrorNodes},
+            {500, Msg}
+    end.
+
 add_status(Name, Link) ->
     NodeRPCResults = emqx_cluster_link_mqtt:get_resource_cluster(Name),
     Status = collect_single_status(NodeRPCResults),

apps/emqx_cluster_link/src/emqx_cluster_link_metrics.erl

@@ -11,6 +11,7 @@
     drop_metrics/1,
 
     get_metrics/1,
+    reset_metrics/1,
     routes_set/2
 ]).
 
@@ -47,6 +48,16 @@ maybe_create_metrics(ClusterName) ->
             )
     end.
 
+reset_metrics(ClusterName) ->
+    Nodes = emqx:running_nodes(),
+    Timeout = 15_000,
+    RouterResults = emqx_metrics_proto_v2:reset_metrics(Nodes, ?METRIC_NAME, ClusterName, Timeout),
+    ResourceId = emqx_cluster_link_mqtt:resource_id(ClusterName),
+    ResourceResults = emqx_metrics_proto_v2:reset_metrics(
+        Nodes, resource_metrics, ResourceId, Timeout
+    ),
+    lists:zip3(Nodes, RouterResults, ResourceResults).
+
 drop_metrics(ClusterName) ->
     ok = emqx_metrics_worker:clear_metrics(?METRIC_NAME, ClusterName).
 

apps/emqx_cluster_link/test/emqx_cluster_link_api_SUITE.erl

@@ -181,6 +181,11 @@ get_metrics(SourceOrTargetCluster, Name) ->
     Path = emqx_mgmt_api_test_util:api_path(Host, [api_root(), "link", Name, "metrics"]),
     emqx_mgmt_api_test_util:simple_request(get, Path, _Params = []).
 
+reset_metrics(SourceOrTargetCluster, Name) ->
+    Host = host(SourceOrTargetCluster),
+    Path = emqx_mgmt_api_test_util:api_path(Host, [api_root(), "link", Name, "metrics", "reset"]),
+    emqx_mgmt_api_test_util:simple_request(put, Path, _Params = []).
+
 host(source) -> "http://127.0.0.1:18083";
 host(target) -> "http://127.0.0.1:28083".
 
@@ -755,6 +760,34 @@ t_metrics(Config) ->
         )
     ),
 
+    %% Reset metrics
+    ?assertMatch({204, _}, reset_metrics(source, SourceName)),
+    ?assertMatch(
+        {200, #{
+            <<"metrics">> := #{
+                <<"router">> := #{<<"routes">> := 0},
+                <<"forwarding">> := #{<<"matched">> := 0}
+            },
+            <<"node_metrics">> := [
+                #{
+                    <<"metrics">> :=
+                        #{
+                            <<"router">> := #{<<"routes">> := 0},
+                            <<"forwarding">> := #{<<"matched">> := 0}
+                        }
+                },
+                #{
+                    <<"metrics">> :=
+                        #{
+                            <<"router">> := #{<<"routes">> := 0},
+                            <<"forwarding">> := #{<<"matched">> := 0}
+                        }
+                }
+            ]
+        }},
+        get_metrics(source, SourceName)
+    ),
+
     ok.
 
 %% Checks that we can update a link via the API in the same fashion as the frontend does,

changes/ee/feat-13835.en.md

@@ -0,0 +1 @@
+Added a `PUT /cluster/links/link/:name/metrics/reset` HTTP API endpoint that resets metrics for a given cluster link.