Merge pull request #13605 from id/20240813-fix-cluster-link-issues

fix cluster link issues

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -601,7 +601,7 @@ schema("/bridges_probe") ->
                     ?NO_CONTENT;
                 {error, #{kind := validation_error} = Reason0} ->
                     Reason = redact(Reason0),
-                    ?BAD_REQUEST('TEST_FAILED', map_to_json(Reason));
+                    ?BAD_REQUEST('TEST_FAILED', emqx_utils_maps:to_json(Reason));
                 {error, Reason0} when not is_tuple(Reason0); element(1, Reason0) =/= 'exit' ->
                     Reason1 =
                         case Reason0 of
@@ -681,9 +681,9 @@ create_or_update_bridge(BridgeType0, BridgeName, Conf, HttpStatusCode) ->
         {ok, _} ->
             lookup_from_all_nodes(BridgeType, BridgeName, HttpStatusCode);
         {error, {pre_config_update, _HandlerMod, Reason}} when is_map(Reason) ->
-            ?BAD_REQUEST(map_to_json(redact(Reason)));
+            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)));
         {error, Reason} when is_map(Reason) ->
-            ?BAD_REQUEST(map_to_json(redact(Reason)))
+            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)))
     end.
 
 get_metrics_from_local_node(BridgeType0, BridgeName) ->
@@ -1159,19 +1159,6 @@ bpapi_version_range(From, To) ->
 redact(Term) ->
     emqx_utils:redact(Term).
 
-map_to_json(M0) ->
-    %% When dealing with Hocon validation errors, `value' might contain non-serializable
-    %% values (e.g.: user_lookup_fun), so we try again without that key if serialization
-    %% fails as a best effort.
-    M1 = emqx_utils_maps:jsonable_map(M0, fun(K, V) -> {K, emqx_utils_maps:binary_string(V)} end),
-    try
-        emqx_utils_json:encode(M1)
-    catch
-        error:_ ->
-            M2 = maps:without([value, <<"value">>], M1),
-            emqx_utils_json:encode(M2)
-    end.
-
 non_compat_bridge_msg() ->
     <<"bridge already exists as non Bridge V1 compatible action">>.
 

apps/emqx_bridge/src/emqx_bridge_v2_api.erl

@@ -927,7 +927,7 @@ handle_probe(ConfRootKey, Request) ->
                     ?NO_CONTENT;
                 {error, #{kind := validation_error} = Reason0} ->
                     Reason = redact(Reason0),
-                    ?BAD_REQUEST('TEST_FAILED', map_to_json(Reason));
+                    ?BAD_REQUEST('TEST_FAILED', emqx_utils_maps:to_json(Reason));
                 {error, Reason0} when not is_tuple(Reason0); element(1, Reason0) =/= 'exit' ->
                     Reason1 =
                         case Reason0 of
@@ -1426,7 +1426,7 @@ create_or_update_bridge(ConfRootKey, BridgeType, BridgeName, Conf, HttpStatusCod
             ok = emqx_resource:validate_name(BridgeName)
         catch
             throw:Error ->
-                ?BAD_REQUEST(map_to_json(Error))
+                ?BAD_REQUEST(emqx_utils_maps:to_json(Error))
         end,
     case Check of
         ok ->
@@ -1443,9 +1443,9 @@ do_create_or_update_bridge(ConfRootKey, BridgeType, BridgeName, Conf, HttpStatus
             PreOrPostConfigUpdate =:= pre_config_update;
             PreOrPostConfigUpdate =:= post_config_update
         ->
-            ?BAD_REQUEST(map_to_json(redact(Reason)));
+            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)));
         {error, Reason} when is_map(Reason) ->
-            ?BAD_REQUEST(map_to_json(redact(Reason)))
+            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)))
     end.
 
 enable_func(true) -> enable;
@@ -1471,19 +1471,6 @@ bin(S) when is_atom(S) ->
 bin(S) when is_binary(S) ->
     S.
 
-map_to_json(M0) ->
-    %% When dealing with Hocon validation errors, `value' might contain non-serializable
-    %% values (e.g.: user_lookup_fun), so we try again without that key if serialization
-    %% fails as a best effort.
-    M1 = emqx_utils_maps:jsonable_map(M0, fun(K, V) -> {K, emqx_utils_maps:binary_string(V)} end),
-    try
-        emqx_utils_json:encode(M1)
-    catch
-        error:_ ->
-            M2 = maps:without([value, <<"value">>], M1),
-            emqx_utils_json:encode(M2)
-    end.
-
 to_existing_atom(X) ->
     case emqx_utils:safe_to_existing_atom(X, utf8) of
         {ok, A} -> A;

apps/emqx_cluster_link/src/emqx_cluster_link_api.erl

@@ -60,7 +60,7 @@ schema("/cluster/links") ->
                 'requestBody' => link_config_schema(),
                 responses =>
                     #{
-                        200 => link_config_schema_response(),
+                        201 => link_config_schema_response(),
                         400 =>
                             emqx_dashboard_swagger:error_codes(
                                 [?BAD_REQUEST, ?ALREADY_EXISTS],
@@ -221,6 +221,21 @@ handle_list() ->
     ?OK(Response).
 
 handle_create(Name, Params) ->
+    Check =
+        try
+            ok = emqx_resource:validate_name(Name)
+        catch
+            throw:Error ->
+                ?BAD_REQUEST(emqx_utils_maps:to_json(Error))
+        end,
+    case Check of
+        ok ->
+            do_create(Name, Params);
+        BadRequest ->
+            BadRequest
+    end.
+
+do_create(Name, Params) ->
     case emqx_cluster_link_config:create_link(Params) of
         {ok, Link} ->
             ?CREATED(add_status(Name, Link));

apps/emqx_cluster_link/test/emqx_cluster_link_api_SUITE.erl

@@ -298,6 +298,39 @@ t_crud(_Config) ->
 
     ok.
 
+t_create_invalid(_Config) ->
+    Params = link_params(),
+    EmptyName = <<>>,
+    {400, #{<<"code">> := <<"BAD_REQUEST">>, <<"message">> := Message1}} = create_link(
+        EmptyName, Params
+    ),
+    ?assertMatch(
+        #{<<"kind">> := <<"validation_error">>, <<"reason">> := <<"Name cannot be empty string">>},
+        Message1
+    ),
+    LongName = binary:copy(<<$a>>, 256),
+    {400, #{<<"code">> := <<"BAD_REQUEST">>, <<"message">> := Message2}} = create_link(
+        LongName, Params
+    ),
+    ?assertMatch(
+        #{
+            <<"kind">> := <<"validation_error">>,
+            <<"reason">> := <<"Name length must be less than 255">>
+        },
+        Message2
+    ),
+    BadName = <<"~!@#$%^&*()_+{}:'<>?|">>,
+    {400, #{<<"code">> := <<"BAD_REQUEST">>, <<"message">> := Message3}} = create_link(
+        BadName, Params
+    ),
+    ?assertMatch(
+        #{
+            <<"kind">> := <<"validation_error">>,
+            <<"reason">> := <<"Invalid name format", _/binary>>
+        },
+        Message3
+    ).
+
 %% Verifies the behavior of reported status under different conditions when listing all
 %% links and when fetching a specific link.
 t_status(Config) ->

apps/emqx_connector/src/emqx_connector_api.erl

@@ -389,7 +389,7 @@ schema("/connectors_probe") ->
                     ?NO_CONTENT;
                 {error, #{kind := validation_error} = Reason0} ->
                     Reason = redact(Reason0),
-                    ?BAD_REQUEST('TEST_FAILED', map_to_json(Reason));
+                    ?BAD_REQUEST('TEST_FAILED', emqx_utils_maps:to_json(Reason));
                 {error, Reason0} when not is_tuple(Reason0); element(1, Reason0) =/= 'exit' ->
                     Reason1 =
                         case Reason0 of
@@ -449,7 +449,7 @@ create_or_update_connector(ConnectorType, ConnectorName, Conf, HttpStatusCode) -
             ok = emqx_resource:validate_name(ConnectorName)
         catch
             throw:Error ->
-                ?BAD_REQUEST(map_to_json(Error))
+                ?BAD_REQUEST(emqx_utils_maps:to_json(Error))
         end,
     case Check of
         ok ->
@@ -466,9 +466,9 @@ do_create_or_update_connector(ConnectorType, ConnectorName, Conf, HttpStatusCode
             PreOrPostConfigUpdate =:= pre_config_update;
             PreOrPostConfigUpdate =:= post_config_update
         ->
-            ?BAD_REQUEST(map_to_json(redact(Reason)));
+            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)));
         {error, Reason} when is_map(Reason) ->
-            ?BAD_REQUEST(map_to_json(redact(Reason)))
+            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)))
     end.
 
 '/connectors/:id/enable/:enable'(put, #{bindings := #{id := Id, enable := Enable}}) ->
@@ -803,16 +803,3 @@ maybe_unwrap(RpcMulticallResult) ->
 
 redact(Term) ->
     emqx_utils:redact(Term).
-
-map_to_json(M0) ->
-    %% When dealing with Hocon validation errors, `value' might contain non-serializable
-    %% values (e.g.: user_lookup_fun), so we try again without that key if serialization
-    %% fails as a best effort.
-    M1 = emqx_utils_maps:jsonable_map(M0, fun(K, V) -> {K, emqx_utils_maps:binary_string(V)} end),
-    try
-        emqx_utils_json:encode(M1)
-    catch
-        error:_ ->
-            M2 = maps:without([value, <<"value">>], M1),
-            emqx_utils_json:encode(M2)
-    end.

apps/emqx_utils/src/emqx_utils_maps.erl

@@ -36,6 +36,7 @@
     put_if/4,
     rename/3,
     safe_atom_key_map/1,
+    to_json/1,
     unindent/2,
     unsafe_atom_key_map/1,
     update_if_present/3
@@ -175,6 +176,20 @@ binary_key_map(Map) ->
 safe_atom_key_map(Map) ->
     convert_keys_to_atom(Map, fun(K) -> binary_to_existing_atom(K, utf8) end).
 
+-spec to_json(map()) -> emqx_utils_json:json_text().
+to_json(M0) ->
+    %% When dealing with Hocon validation errors, `value' might contain non-serializable
+    %% values (e.g.: user_lookup_fun), so we try again without that key if serialization
+    %% fails as a best effort.
+    M1 = jsonable_map(M0, fun(K, V) -> {K, binary_string(V)} end),
+    try
+        emqx_utils_json:encode(M1)
+    catch
+        error:_ ->
+            M2 = maps:without([value, <<"value">>], M1),
+            emqx_utils_json:encode(M2)
+    end.
+
 -spec jsonable_map(map() | list()) -> map() | list().
 jsonable_map(Map) ->
     jsonable_map(Map, fun(K, V) -> {K, V} end).