Merge pull request #13612 from id/20240814-cluster-link-followup

cluster link followup

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', emqx_utils_maps:to_json(Reason));
+                    ?BAD_REQUEST('TEST_FAILED', emqx_utils_api: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(emqx_utils_maps:to_json(redact(Reason)));
+            ?BAD_REQUEST(emqx_utils_api:to_json(redact(Reason)));
         {error, Reason} when is_map(Reason) ->
-            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)))
+            ?BAD_REQUEST(emqx_utils_api:to_json(redact(Reason)))
     end.
 
 get_metrics_from_local_node(BridgeType0, BridgeName) ->

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', emqx_utils_maps:to_json(Reason));
+                    ?BAD_REQUEST('TEST_FAILED', emqx_utils_api: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(emqx_utils_maps:to_json(Error))
+                ?BAD_REQUEST(emqx_utils_api: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(emqx_utils_maps:to_json(redact(Reason)));
+            ?BAD_REQUEST(emqx_utils_api:to_json(redact(Reason)));
         {error, Reason} when is_map(Reason) ->
-            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)))
+            ?BAD_REQUEST(emqx_utils_api:to_json(redact(Reason)))
     end.
 
 enable_func(true) -> enable;

apps/emqx_cluster_link/src/emqx_cluster_link_api.erl

@@ -226,7 +226,7 @@ handle_create(Name, Params) ->
             ok = emqx_resource:validate_name(Name)
         catch
             throw:Error ->
-                ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Error)))
+                ?BAD_REQUEST(emqx_utils_api:to_json(redact(Error)))
         end,
     case Check of
         ok ->

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', emqx_utils_maps:to_json(Reason));
+                    ?BAD_REQUEST('TEST_FAILED', emqx_utils_api: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(emqx_utils_maps:to_json(Error))
+                ?BAD_REQUEST(emqx_utils_api: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(emqx_utils_maps:to_json(redact(Reason)));
+            ?BAD_REQUEST(emqx_utils_api:to_json(redact(Reason)));
         {error, Reason} when is_map(Reason) ->
-            ?BAD_REQUEST(emqx_utils_maps:to_json(redact(Reason)))
+            ?BAD_REQUEST(emqx_utils_api:to_json(redact(Reason)))
     end.
 
 '/connectors/:id/enable/:enable'(put, #{bindings := #{id := Id, enable := Enable}}) ->

apps/emqx_utils/src/emqx_utils_api.erl

@@ -17,6 +17,7 @@
 -module(emqx_utils_api).
 
 -export([
+    to_json/1,
     with_node/2,
     with_node_or_cluster/2
 ]).
@@ -45,6 +46,20 @@ with_node_or_cluster(<<"all">>, Fun) ->
 with_node_or_cluster(Node, Fun) ->
     with_node(Node, Fun).
 
+-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 = 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.
+
 %%--------------------------------------------------------------------
 %% Internal
 %%--------------------------------------------------------------------

apps/emqx_utils/src/emqx_utils_maps.erl

@@ -36,7 +36,6 @@
     put_if/4,
     rename/3,
     safe_atom_key_map/1,
-    to_json/1,
     unindent/2,
     unsafe_atom_key_map/1,
     update_if_present/3
@@ -176,20 +175,6 @@ 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).

scripts/dev-cluster-host.sh

@@ -15,20 +15,29 @@ cd -P -- "$(dirname -- "$0")/../"
 
 help() {
     echo
-    echo "start | stop"
+    echo "$0 start|stop [-h|--help] [-n|--nodes <number>] [-c|--cores <number>] [-b|--boot <script>]"
+    echo
     echo "-h|--help: To display this usage info"
     echo "-n|--nodes: total number of nodes to start (default: 2)"
-    echo "-c|--core_nodes: number of core nodes to start (default: 1)"
+    echo "-c|--cores: number of core nodes to start (default: 2)"
     echo "-b|--boot: boot script (default: ./_build/emqx/rel/emqx/bin/emqx)"
+    echo "--no-cluster: don't create a cluster, just start separate nodes"
 }
 
+if [ "$#" -lt 1 ]; then
+    help
+    exit 1
+fi
+
 CMD="$1"
 shift || true
 
 export EMQX_NODE__COOKIE=test
-BOOT_SCRIPT='./_build/emqx/rel/emqx/bin/emqx'
+PROFILE=${PROFILE:-emqx}
+BOOT_SCRIPT="./_build/${PROFILE}/rel/emqx/bin/emqx"
 NODES=2
-CORE_NODES=1
+CORE_NODES=2
+NO_CLUSTER=false
 
 while [ "$#" -gt 0 ]; do
     case $1 in
@@ -40,7 +49,7 @@ while [ "$#" -gt 0 ]; do
             NODES="$2"
             shift 2
             ;;
-        -c|--core-nodes)
+        -c|--cores)
             CORE_NODES="$2"
             shift 2
             ;;
@@ -48,6 +57,10 @@ while [ "$#" -gt 0 ]; do
             BOOT_SCRIPT="$2"
             shift 2
             ;;
+        --no-cluster)
+            NO_CLUSTER=true
+            shift
+            ;;
         *)
             echo "unknown option $1"
             exit 1
@@ -65,10 +78,15 @@ for i in $(seq 1 "$CORE_NODES"); do
     SEEDS_ARRAY+=("core${i}@127.0.0.$i")
     CORE_IDS+=("$i")
 done
-for i in $(seq "$((CORE_NODES+1))" "$((CORE_NODES+REPLICANT_NODES))"); do
-    REPLICANT_IDS+=("$i")
-done
+if [ "$REPLICANT_NODES" -gt 0 ]; then
+    for i in $(seq "$((CORE_NODES+1))" "$((CORE_NODES+REPLICANT_NODES))"); do
+        REPLICANT_IDS+=("$i")
+    done
+fi
 
+if [ "${NO_CLUSTER}" = true ]; then
+    SEEDS_ARRAY=()
+fi
 SEEDS="$(IFS=,; echo "${SEEDS_ARRAY[*]}")"
 
 if [ "$CMD" = "stop" ]; then
@@ -116,11 +134,23 @@ start_node() {
 }
 
 for id in "${CORE_IDS[@]}"; do
-    sudo ifconfig lo0 alias "127.0.0.$id" up
+    set +e
+    if ! ifconfig lo0 alias "127.0.0.$id" up; then
+        echo "Failed to create an alias for loopback address 127.0.0.$id. Retrying with sudo."
+        set -e
+        sudo ifconfig lo0 alias "127.0.0.$id" up
+    fi
+    set -e
     start_node core "$id" &
 done
 
 for id in "${REPLICANT_IDS[@]}"; do
-    sudo ifconfig lo0 alias "127.0.0.$id" up
+    set +e
+    if ! ifconfig lo0 alias "127.0.0.$id" up; then
+        echo "Failed to create an alias for loopback address 127.0.0.$id. Retrying with sudo."
+        set -e
+        sudo ifconfig lo0 alias "127.0.0.$id" up
+    fi
+    set -e
     start_node replicant "$id" &
 done