Merge pull request #7871 from terry-xiaoyu/connector_restart_wrong_node

fix: restart bridges via HTTP APIs to wrong node

apps/emqx_bridge/i18n/emqx_bridge_api.conf

@@ -13,8 +13,8 @@ emqx_bridge_api {
 
     desc_param_path_operation_on_node {
                    desc {
-                         en: """Operations can be one of: start, stop, restart"""
-                         zh: """节点可用操作：启动、停止、重新启动"""
+                         en: """Operations can be one of: stop, restart"""
+                         zh: """节点可用操作：停止、重新启动"""
                         }
                    label: {
                            en: "Node Operation "

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -486,20 +486,21 @@ lookup_from_local_node(BridgeType, BridgeName) ->
 
 '/nodes/:node/bridges/:id/operation/:operation'(post, #{
     bindings :=
-        #{id := Id, operation := Op}
+        #{id := Id, operation := Op, node := Node}
 }) ->
     ?TRY_PARSE_ID(
         Id,
-        case operation_func(Op) of
+        case node_operation_func(Op) of
             invalid ->
                 {400, error_msg('BAD_REQUEST', <<"invalid operation">>)};
-            OperFunc when OperFunc == restart; OperFunc == stop ->
+            OperFunc ->
+                TargetNode = binary_to_atom(Node, utf8),
                 ConfMap = emqx:get_config([bridges, BridgeType, BridgeName]),
                 case maps:get(enable, ConfMap, false) of
                     false ->
                         {403, error_msg('FORBIDDEN_REQUEST', <<"forbidden operation">>)};
                     true ->
-                        case emqx_bridge:OperFunc(BridgeType, BridgeName) of
+                        case emqx_bridge_proto_v1:OperFunc(TargetNode, BridgeType, BridgeName) of
                             ok -> {200};
                             {error, Reason} -> {500, error_msg('INTERNAL_ERROR', Reason)}
                         end
@@ -507,6 +508,10 @@ lookup_from_local_node(BridgeType, BridgeName) ->
         end
     ).
 
+node_operation_func(<<"stop">>) -> stop_bridge_to_node;
+node_operation_func(<<"restart">>) -> restart_bridge_to_node;
+node_operation_func(_) -> invalid.
+
 operation_func(<<"stop">>) -> stop;
 operation_func(<<"restart">>) -> restart;
 operation_func(<<"enable">>) -> enable;

apps/emqx_bridge/src/proto/emqx_bridge_proto_v1.erl

@@ -22,6 +22,8 @@
     introduced_in/0,
 
     list_bridges/1,
+    restart_bridge_to_node/3,
+    stop_bridge_to_node/3,
     lookup_from_all_nodes/3,
     restart_bridges_to_all_nodes/3,
     stop_bridges_to_all_nodes/3
@@ -40,6 +42,28 @@ list_bridges(Node) ->
 
 -type key() :: atom() | binary() | [byte()].
 
+-spec restart_bridge_to_node(node(), key(), key()) ->
+    term().
+restart_bridge_to_node(Node, BridgeType, BridgeName) ->
+    rpc:call(
+        Node,
+        emqx_bridge,
+        restart,
+        [BridgeType, BridgeName],
+        ?TIMEOUT
+    ).
+
+-spec stop_bridge_to_node(node(), key(), key()) ->
+    term().
+stop_bridge_to_node(Node, BridgeType, BridgeName) ->
+    rpc:call(
+        Node,
+        emqx_bridge,
+        stop,
+        [BridgeType, BridgeName],
+        ?TIMEOUT
+    ).
+
 -spec restart_bridges_to_all_nodes([node()], key(), key()) ->
     emqx_rpc:erpc_multicall().
 restart_bridges_to_all_nodes(Nodes, BridgeType, BridgeName) ->