feat: impl bridge reset_metrics api

apps/emqx_bridge/src/emqx_bridge.erl

@@ -52,6 +52,7 @@
         , update/3
         , stop/2
         , restart/2
+        , reset_metrics/1
         ]).
 
 -export([ send_message/2
@@ -210,6 +211,9 @@ lookup(Type, Name, RawConf) ->
                    raw_config => RawConf}}
     end.
 
+reset_metrics(ResourceId) ->
+    emqx_resource:reset_metrics(ResourceId).
+
 stop(Type, Name) ->
     emqx_resource:stop(resource_id(Type, Name)).
 

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -34,6 +34,7 @@
         , '/bridges/:id'/2
         , '/bridges/:id/operation/:operation'/2
         , '/nodes/:node/bridges/:id/operation/:operation'/2
+        , '/bridges/:id/reset_metrics'/2
         ]).
 
 -export([ lookup_from_local_node/2
@@ -76,7 +77,8 @@ api_spec() ->
     emqx_dashboard_swagger:spec(?MODULE, #{check_schema => false}).
 
 paths() -> ["/bridges", "/bridges/:id", "/bridges/:id/operation/:operation",
-            "/nodes/:node/bridges/:id/operation/:operation"].
+            "/nodes/:node/bridges/:id/operation/:operation",
+            "/bridges/:id/reset_metrics"].
 
 error_schema(Code, Message) when is_atom(Code) ->
     error_schema([Code], Message);
@@ -282,6 +284,20 @@ schema("/bridges/:id") ->
         }
     };
 
+schema("/bridges/:id/reset_metrics") ->
+    #{
+        'operationId' => '/bridges/:id/reset_metrics',
+        put => #{
+            tags => [<<"bridges">>],
+            summary => <<"Reset Bridge Metrics">>,
+            description => <<"Reset a bridge metrics by Id">>,
+            parameters => [param_path_id()],
+            responses => #{
+                200 => <<"Reset success">>,
+                400 => error_schema(['BAD_REQUEST'], "RPC Call Failed")
+            }
+        }
+    };
 schema("/bridges/:id/operation/:operation") ->
     #{
         'operationId' => '/bridges/:id/operation/:operation',
@@ -363,6 +379,12 @@ schema("/nodes/:node/bridges/:id/operation/:operation") ->
                 {500, error_msg('INTERNAL_ERROR', Reason)}
         end).
 
+'/bridges/:id/reset_metrics'(put, #{bindings := #{id := Id}}) ->
+    case emqx_bridge:reset_metrics(Id) of
+        ok -> {200, <<"Reset success">>};
+        Reason -> {400, error_msg('BAD_REQUEST', Reason)}
+    end.
+
 lookup_from_all_nodes(BridgeType, BridgeName, SuccCode) ->
     Nodes = mria_mnesia:running_nodes(),
     case is_ok(emqx_bridge_proto_v1:lookup_from_all_nodes(Nodes, BridgeType, BridgeName)) of

apps/emqx_bridge/test/emqx_bridge_api_SUITE.erl

@@ -333,6 +333,30 @@ t_enable_disable_bridges(_) ->
     {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
     {ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []).
 
+t_reset_bridges(_) ->
+    %% assert we there's no bridges at first
+    {ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
+
+    Port = start_http_server(fun handle_fun_200_ok/2),
+    URL1 = ?URL(Port, "abc"),
+    {ok, 201, Bridge} = request(post, uri(["bridges"]),
+        ?HTTP_BRIDGE(URL1, ?BRIDGE_TYPE, ?BRIDGE_NAME)),
+    %ct:pal("the bridge ==== ~p", [Bridge]),
+    #{ <<"type">> := ?BRIDGE_TYPE
+     , <<"name">> := ?BRIDGE_NAME
+     , <<"status">> := <<"connected">>
+     , <<"node_status">> := [_|_]
+     , <<"metrics">> := _
+     , <<"node_metrics">> := [_|_]
+     , <<"url">> := URL1
+     } = jsx:decode(Bridge),
+    BridgeID = emqx_bridge:bridge_id(?BRIDGE_TYPE, ?BRIDGE_NAME),
+    {ok, 200, <<"Reset success">>} = request(put, uri(["bridges", BridgeID, "reset_metrics"]), []),
+
+    %% delete the bridge
+    {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
+    {ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []).
+
 request(Method, Url, Body) ->
     request(<<"bridge_admin">>, Method, Url, Body).