refactor: use string type for server and servers

apps/emqx/src/emqx_config.erl

@@ -413,6 +413,31 @@ check_config(SchemaMod, RawConf) ->
     check_config(SchemaMod, RawConf, #{}).
 
 check_config(SchemaMod, RawConf, Opts0) ->
+    try
+        do_check_config(SchemaMod, RawConf, Opts0)
+    catch
+        throw:{Schema, Errors} ->
+            compact_errors(Schema, Errors)
+    end.
+
+%% HOCON tries to be very informative about all the detailed errors
+%% it's maybe too much when reporting to the user
+-spec compact_errors(any(), any()) -> no_return().
+compact_errors(Schema, [Error0 | More]) when is_map(Error0) ->
+    Error1 = Error0#{discarded_errors_count => length(More)},
+    Error =
+        case is_atom(Schema) of
+            true ->
+                Error1#{schema_module => Schema};
+            false ->
+                Error1
+        end,
+    throw(Error);
+compact_errors(Schema, Errors) ->
+    %% unexpected, we need the stacktrace reported, hence error
+    error({Schema, Errors}).
+
+do_check_config(SchemaMod, RawConf, Opts0) ->
     Opts1 = #{
         return_plain => true,
         format => map,

apps/emqx/src/emqx_config_handler.erl

@@ -245,7 +245,7 @@ process_update_request(ConfKeyPath, Handlers, {{update, UpdateReq}, Opts}) ->
             BinKeyPath = bin_path(ConfKeyPath),
             case check_permissions(update, BinKeyPath, NewRawConf, Opts) of
                 allow ->
-                    OverrideConf = update_override_config(NewRawConf, Opts),
+                    OverrideConf = merge_to_override_config(NewRawConf, Opts),
                     {ok, NewRawConf, OverrideConf, Opts};
                 {deny, Reason} ->
                     {error, {permission_denied, Reason}}
@@ -447,9 +447,10 @@ remove_from_override_config(BinKeyPath, Opts) ->
     OldConf = emqx_config:read_override_conf(Opts),
     emqx_map_lib:deep_remove(BinKeyPath, OldConf).
 
-update_override_config(_RawConf, #{persistent := false}) ->
+%% apply new config on top of override config
+merge_to_override_config(_RawConf, #{persistent := false}) ->
     undefined;
-update_override_config(RawConf, Opts) ->
+merge_to_override_config(RawConf, Opts) ->
     OldConf = emqx_config:read_override_conf(Opts),
     maps:merge(OldConf, RawConf).
 

apps/emqx/src/emqx_schema.erl

@@ -40,8 +40,9 @@
 -type comma_separated_atoms() :: [atom()].
 -type bar_separated_list() :: list().
 -type ip_port() :: tuple() | integer().
--type host_port() :: tuple().
 -type cipher() :: map().
+-type port_number() :: 1..65536.
+-type server_parse_option() :: #{default_port => port_number(), no_port => boolean()}.
 
 -typerefl_from_string({duration/0, emqx_schema, to_duration}).
 -typerefl_from_string({duration_s/0, emqx_schema, to_duration_s}).
@@ -53,7 +54,6 @@
 -typerefl_from_string({comma_separated_binary/0, emqx_schema, to_comma_separated_binary}).
 -typerefl_from_string({bar_separated_list/0, emqx_schema, to_bar_separated_list}).
 -typerefl_from_string({ip_port/0, emqx_schema, to_ip_port}).
--typerefl_from_string({host_port/0, emqx_schema, to_host_port}).
 -typerefl_from_string({cipher/0, emqx_schema, to_erl_cipher_suite}).
 -typerefl_from_string({comma_separated_atoms/0, emqx_schema, to_comma_separated_atoms}).
 
@@ -80,11 +80,19 @@
     to_comma_separated_binary/1,
     to_bar_separated_list/1,
     to_ip_port/1,
-    to_host_port/1,
     to_erl_cipher_suite/1,
     to_comma_separated_atoms/1
 ]).
 
+-export([
+    parse_server/2,
+    parse_servers/2,
+    servers_validator/2,
+    servers_sc/2,
+    convert_servers/1,
+    convert_servers/2
+]).
+
 -behaviour(hocon_schema).
 
 -reflect_type([
@@ -99,7 +107,6 @@
     comma_separated_binary/0,
     bar_separated_list/0,
     ip_port/0,
-    host_port/0,
     cipher/0,
     comma_separated_atoms/0
 ]).
@@ -2172,40 +2179,15 @@ to_bar_separated_list(Str) ->
 %%  - :1883
 %%  - :::1883
 to_ip_port(Str) ->
-    to_host_port(Str, ip_addr).
+    case split_ip_port(Str) of
-
+        {"", Port} ->
-%% @doc support the following format:
-%%  - 127.0.0.1:1883
-%%  - ::1:1883
-%%  - [::1]:1883
-%%  - :1883
-%%  - :::1883
-%%  - example.com:80
-to_host_port(Str) ->
-    to_host_port(Str, hostname).
-
-%%  - example.com:80
-to_host_port(Str, IpOrHost) ->
-    case split_host_port(Str) of
-        {"", Port} when IpOrHost =:= ip_addr ->
             %% this is a local address
             {ok, list_to_integer(Port)};
-        {"", _Port} ->
-            %% must specify host part when it's a remote endpoint
-            {error, bad_host_port};
         {MaybeIp, Port} ->
             PortVal = list_to_integer(Port),
             case inet:parse_address(MaybeIp) of
                 {ok, IpTuple} ->
                     {ok, {IpTuple, PortVal}};
-                _ when IpOrHost =:= hostname ->
-                    %% check is a rfc1035's hostname
-                    case inet_parse:domain(MaybeIp) of
-                        true ->
-                            {ok, {MaybeIp, PortVal}};
-                        _ ->
-                            {error, bad_hostname}
-                    end;
                 _ ->
                     {error, bad_ip_port}
             end;
@@ -2213,7 +2195,7 @@ to_host_port(Str, IpOrHost) ->
             {error, bad_ip_port}
     end.
 
-split_host_port(Str0) ->
+split_ip_port(Str0) ->
     Str = re:replace(Str0, " ", "", [{return, list}, global]),
     case lists:split(string:rchr(Str, $:), Str) of
         %% no colon
@@ -2376,3 +2358,229 @@ non_empty_string(<<>>) -> {error, empty_string_not_allowed};
 non_empty_string("") -> {error, empty_string_not_allowed};
 non_empty_string(S) when is_binary(S); is_list(S) -> ok;
 non_empty_string(_) -> {error, invalid_string}.
+
+%% @doc Make schema for 'server' or 'servers' field.
+%% for each field, there are three passes:
+%% 1. converter: Normalize the value.
+%%               This normalized value is stored in EMQX's raw config.
+%% 2. validator: Validate the normalized value.
+%%               Besides checkin if the value can be empty or undefined
+%%               it also calls the 3rd pass to see if the provided
+%%               hosts can be successfully parsed.
+%% 3. parsing: Done at runtime in each module which uses this config
+servers_sc(Meta0, ParseOpts) ->
+    Required = maps:get(required, Meta0, true),
+    Meta = #{
+        required => Required,
+        converter => fun convert_servers/2,
+        validator => servers_validator(ParseOpts, Required)
+    },
+    sc(string(), maps:merge(Meta, Meta0)).
+
+%% @hidden Convert a deep map to host:port pairs.
+%% This is due to the fact that a host:port string
+%% often can be parsed as a HOCON struct.
+%% e.g. when a string from environment variable is `host.domain.name:80'
+%% without escaped quotes, it's parsed as
+%% `#{<<"host">> => #{<<"domain">> => #{<<"name">> => 80}}}'
+%% and when it is a comma-separated list of host:port pairs
+%% like `h1.foo:80, h2.bar:81' then it is parsed as
+%% `#{<<"h1">> => #{<<"foo">> => 80}, <<"h2">> => #{<<"bar">> => 81}}'
+%% This function is to format the map back to host:port (pairs)
+%% This function also tries to remove spaces around commas in comma-separated,
+%% `host:port' list, and format string array to comma-separated.
+convert_servers(HoconValue, _HoconOpts) ->
+    convert_servers(HoconValue).
+
+convert_servers(undefined) ->
+    %% should not format 'undefined' as string
+    %% not to throw exception either
+    %% (leave it to the 'required => true | false' check)
+    undefined;
+convert_servers(Map) when is_map(Map) ->
+    try
+        List = convert_hocon_map_host_port(Map),
+        iolist_to_binary(string:join(List, ","))
+    catch
+        _:_ ->
+            throw("bad_host_port")
+    end;
+convert_servers([H | _] = Array) when is_binary(H) orelse is_list(H) ->
+    %% if the old config was a string array
+    %% we want to make sure it's converted to a comma-separated
+    iolist_to_binary([[I, ","] || I <- Array]);
+convert_servers(Str) ->
+    normalize_host_port_str(Str).
+
+%% remove spaces around comma (,)
+normalize_host_port_str(Str) ->
+    iolist_to_binary(re:replace(Str, "(\s)*,(\s)*", ",")).
+
+%% @doc Shared validation function for both 'server' and 'servers' string.
+%% NOTE: Validator is called after converter.
+servers_validator(Opts, Required) ->
+    fun(Str0) ->
+        Str = str(Str0),
+        case Str =:= "" orelse Str =:= "undefined" of
+            true when Required ->
+                %% it's a required field
+                %% but value is set to an empty string (from environment override)
+                %% or when the filed is not set in config file
+                %% NOTE: assuming nobody is going to name their server "undefined"
+                throw("cannot_be_empty");
+            true ->
+                ok;
+            _ ->
+                %% it's valid as long as it can be parsed
+                _ = parse_servers(Str, Opts),
+                ok
+        end
+    end.
+
+%% @doc Parse `host[:port]' endpoint to a `{Host, Port}' tuple or just `Host' string.
+%% `Opt' is a `map()' with below options supported:
+%%
+%% `default_port': a port number, so users are not forced to configure
+%%                 port number.
+%% `no_port': by default it's `false', when set to `true',
+%%            a `throw' exception is raised if the port is found.
+-spec parse_server(undefined | string() | binary(), server_parse_option()) ->
+    {string(), port_number()}.
+parse_server(Str, Opts) ->
+    case parse_servers(Str, Opts) of
+        undefined ->
+            undefined;
+        [L] ->
+            L;
+        [_ | _] = L ->
+            throw("expecting_one_host_but_got: " ++ integer_to_list(length(L)))
+    end.
+
+%% @doc Parse comma separated `host[:port][,host[:port]]' endpoints
+%% into a list of `{Host, Port}' tuples or just `Host' string.
+-spec parse_servers(undefined | string() | binary(), server_parse_option()) ->
+    [{string(), port_number()}].
+parse_servers(undefined, _Opts) ->
+    %% should not parse 'undefined' as string,
+    %% not to throw exception either,
+    %% leave it to the 'required => true | false' check
+    undefined;
+parse_servers(Str, Opts) ->
+    case do_parse_servers(Str, Opts) of
+        [] ->
+            %% treat empty as 'undefined'
+            undefined;
+        [_ | _] = L ->
+            L
+    end.
+
+do_parse_servers([H | _] = Array, Opts) when is_binary(H) orelse is_list(H) ->
+    %% the old schema allowed providing a list of strings
+    %% e.g. ["server1:80", "server2:80"]
+    lists:map(
+        fun(HostPort) ->
+            do_parse_server(str(HostPort), Opts)
+        end,
+        Array
+    );
+do_parse_servers(Str, Opts) when is_binary(Str) orelse is_list(Str) ->
+    lists:map(
+        fun(HostPort) ->
+            do_parse_server(HostPort, Opts)
+        end,
+        split_host_port(Str)
+    ).
+
+split_host_port(Str) ->
+    lists:filtermap(
+        fun(S) ->
+            case string:strip(S) of
+                "" -> false;
+                X -> {true, X}
+            end
+        end,
+        string:tokens(str(Str), ",")
+    ).
+
+do_parse_server(Str, Opts) ->
+    DefaultPort = maps:get(default_port, Opts, undefined),
+    NotExpectingPort = maps:get(no_port, Opts, false),
+    case is_integer(DefaultPort) andalso NotExpectingPort of
+        true ->
+            %% either provide a default port from schema,
+            %% or do not allow user to set port number
+            error("bad_schema");
+        false ->
+            ok
+    end,
+    %% do not split with space, there should be no space allowed between host and port
+    case string:tokens(Str, ":") of
+        [Hostname, Port] ->
+            NotExpectingPort andalso throw("not_expecting_port_number"),
+            {check_hostname(Hostname), parse_port(Port)};
+        [Hostname] ->
+            case is_integer(DefaultPort) of
+                true ->
+                    {check_hostname(Hostname), DefaultPort};
+                false when NotExpectingPort ->
+                    check_hostname(Hostname);
+                false ->
+                    throw("missing_port_number")
+            end;
+        _ ->
+            throw("bad_host_port")
+    end.
+
+check_hostname(Str) ->
+    %% not intended to use inet_parse:domain here
+    %% only checking space because it interferes the parsing
+    case string:tokens(Str, " ") of
+        [H] ->
+            case is_port_number(H) of
+                true ->
+                    throw("expecting_hostname_but_got_a_number");
+                false ->
+                    H
+            end;
+        _ ->
+            throw("hostname_has_space")
+    end.
+
+convert_hocon_map_host_port(Map) ->
+    lists:map(
+        fun({Host, Port}) ->
+            %% Only when Host:Port string is a valid HOCON object
+            %% is it possible for the converter to reach here.
+            %%
+            %% For example EMQX_FOO__SERVER='1.2.3.4:1234' is parsed as
+            %% a HOCON string value "1.2.3.4:1234" but not a map because
+            %% 1 is not a valid HOCON field.
+            %%
+            %% EMQX_FOO__SERVER="local.domain.host" (without ':port')
+            %% is also not a valid HOCON object (because it has no value),
+            %% hence parsed as string.
+            true = (Port > 0),
+            str(Host) ++ ":" ++ integer_to_list(Port)
+        end,
+        hocon_maps:flatten(Map, #{})
+    ).
+
+is_port_number(Port) ->
+    try
+        _ = parse_port(Port),
+        true
+    catch
+        _:_ ->
+            false
+    end.
+
+parse_port(Port) ->
+    try
+        P = list_to_integer(string:strip(Port)),
+        true = (P > 0),
+        true = (P =< 65535),
+        P
+    catch
+        _:_ ->
+            throw("bad_port_number")
+    end.

apps/emqx/test/emqx_schema_tests.erl

@@ -178,27 +178,290 @@ ssl_opts_gc_after_handshake_test_not_rancher_listener_test() ->
 
 to_ip_port_test_() ->
     Ip = fun emqx_schema:to_ip_port/1,
-    Host = fun(Str) ->
-        case Ip(Str) of
-            {ok, {_, _} = Res} ->
-                %% assert
-                {ok, Res} = emqx_schema:to_host_port(Str);
-            _ ->
-                emqx_schema:to_host_port(Str)
-        end
-    end,
     [
         ?_assertEqual({ok, 80}, Ip("80")),
-        ?_assertEqual({error, bad_host_port}, Host("80")),
         ?_assertEqual({ok, 80}, Ip(":80")),
-        ?_assertEqual({error, bad_host_port}, Host(":80")),
         ?_assertEqual({error, bad_ip_port}, Ip("localhost:80")),
-        ?_assertEqual({ok, {"localhost", 80}}, Host("localhost:80")),
-        ?_assertEqual({ok, {"example.com", 80}}, Host("example.com:80")),
         ?_assertEqual({ok, {{127, 0, 0, 1}, 80}}, Ip("127.0.0.1:80")),
         ?_assertEqual({error, bad_ip_port}, Ip("$:1900")),
-        ?_assertEqual({error, bad_hostname}, Host("$:1900")),
         ?_assertMatch({ok, {_, 1883}}, Ip("[::1]:1883")),
         ?_assertMatch({ok, {_, 1883}}, Ip("::1:1883")),
         ?_assertMatch({ok, {_, 1883}}, Ip(":::1883"))
     ].
+
+-define(T(CASE, EXPR), {CASE, fun() -> EXPR end}).
+
+parse_server_test_() ->
+    DefaultPort = ?LINE,
+    DefaultOpts = #{default_port => DefaultPort},
+    Parse2 = fun(Value0, Opts) ->
+        Value = emqx_schema:convert_servers(Value0),
+        Validator = emqx_schema:servers_validator(Opts, _Required = true),
+        try
+            Result = emqx_schema:parse_servers(Value, Opts),
+            ?assertEqual(ok, Validator(Value)),
+            Result
+        catch
+            throw:Throw ->
+                %% assert validator throws the same exception
+                ?assertThrow(Throw, Validator(Value)),
+                %% and then let the test code validate the exception
+                throw(Throw)
+        end
+    end,
+    Parse = fun(Value) -> Parse2(Value, DefaultOpts) end,
+    HoconParse = fun(Str0) ->
+        {ok, Map} = hocon:binary(Str0),
+        Str = emqx_schema:convert_servers(Map),
+        Parse(Str)
+    end,
+    [
+        ?T(
+            "single server, binary, no port",
+            ?assertEqual(
+                [{"localhost", DefaultPort}],
+                Parse(<<"localhost">>)
+            )
+        ),
+        ?T(
+            "single server, string, no port",
+            ?assertEqual(
+                [{"localhost", DefaultPort}],
+                Parse("localhost")
+            )
+        ),
+        ?T(
+            "single server, list(string), no port",
+            ?assertEqual(
+                [{"localhost", DefaultPort}],
+                Parse(["localhost"])
+            )
+        ),
+        ?T(
+            "single server, list(binary), no port",
+            ?assertEqual(
+                [{"localhost", DefaultPort}],
+                Parse([<<"localhost">>])
+            )
+        ),
+        ?T(
+            "single server, binary, with port",
+            ?assertEqual(
+                [{"localhost", 9999}],
+                Parse(<<"localhost:9999">>)
+            )
+        ),
+        ?T(
+            "single server, list(string), with port",
+            ?assertEqual(
+                [{"localhost", 9999}],
+                Parse(["localhost:9999"])
+            )
+        ),
+        ?T(
+            "single server, string, with port",
+            ?assertEqual(
+                [{"localhost", 9999}],
+                Parse("localhost:9999")
+            )
+        ),
+        ?T(
+            "single server, list(binary), with port",
+            ?assertEqual(
+                [{"localhost", 9999}],
+                Parse([<<"localhost:9999">>])
+            )
+        ),
+        ?T(
+            "multiple servers, string, no port",
+            ?assertEqual(
+                [{"host1", DefaultPort}, {"host2", DefaultPort}],
+                Parse("host1, host2")
+            )
+        ),
+        ?T(
+            "multiple servers, binary, no port",
+            ?assertEqual(
+                [{"host1", DefaultPort}, {"host2", DefaultPort}],
+                Parse(<<"host1, host2,,,">>)
+            )
+        ),
+        ?T(
+            "multiple servers, list(string), no port",
+            ?assertEqual(
+                [{"host1", DefaultPort}, {"host2", DefaultPort}],
+                Parse(["host1", "host2"])
+            )
+        ),
+        ?T(
+            "multiple servers, list(binary), no port",
+            ?assertEqual(
+                [{"host1", DefaultPort}, {"host2", DefaultPort}],
+                Parse([<<"host1">>, <<"host2">>])
+            )
+        ),
+        ?T(
+            "multiple servers, string, with port",
+            ?assertEqual(
+                [{"host1", 1234}, {"host2", 2345}],
+                Parse("host1:1234, host2:2345")
+            )
+        ),
+        ?T(
+            "multiple servers, binary, with port",
+            ?assertEqual(
+                [{"host1", 1234}, {"host2", 2345}],
+                Parse(<<"host1:1234, host2:2345, ">>)
+            )
+        ),
+        ?T(
+            "multiple servers, list(string), with port",
+            ?assertEqual(
+                [{"host1", 1234}, {"host2", 2345}],
+                Parse([" host1:1234 ", "host2:2345"])
+            )
+        ),
+        ?T(
+            "multiple servers, list(binary), with port",
+            ?assertEqual(
+                [{"host1", 1234}, {"host2", 2345}],
+                Parse([<<"host1:1234">>, <<"host2:2345">>])
+            )
+        ),
+        ?T(
+            "unexpected multiple servers",
+            ?assertThrow(
+                "expecting_one_host_but_got: 2",
+                emqx_schema:parse_server(<<"host1:1234, host2:1234">>, #{default_port => 1})
+            )
+        ),
+        ?T(
+            "multiple servers without ports invalid string list",
+            ?assertThrow(
+                "hostname_has_space",
+                Parse2(["host1 host2"], #{no_port => true})
+            )
+        ),
+        ?T(
+            "multiple servers without ports invalid binary list",
+            ?assertThrow(
+                "hostname_has_space",
+                Parse2([<<"host1 host2">>], #{no_port => true})
+            )
+        ),
+        ?T(
+            "multiple servers wihtout port, mixed list(binary|string)",
+            ?assertEqual(
+                ["host1", "host2"],
+                Parse2([<<"host1">>, "host2"], #{no_port => true})
+            )
+        ),
+        ?T(
+            "no default port, missing port number in config",
+            ?assertThrow(
+                "missing_port_number",
+                emqx_schema:parse_server(<<"a">>, #{})
+            )
+        ),
+        ?T(
+            "empty binary string",
+            ?assertEqual(
+                undefined,
+                emqx_schema:parse_server(<<>>, #{no_port => true})
+            )
+        ),
+        ?T(
+            "empty array",
+            ?assertEqual(
+                undefined,
+                emqx_schema:parse_servers([], #{no_port => true})
+            )
+        ),
+        ?T(
+            "empty binary array",
+            ?assertThrow(
+                "bad_host_port",
+                emqx_schema:parse_servers([<<>>], #{no_port => true})
+            )
+        ),
+        ?T(
+            "HOCON value undefined",
+            ?assertEqual(
+                undefined,
+                emqx_schema:parse_server(undefined, #{no_port => true})
+            )
+        ),
+        ?T(
+            "single server map",
+            ?assertEqual(
+                [{"host1.domain", 1234}],
+                HoconParse("host1.domain:1234")
+            )
+        ),
+        ?T(
+            "multiple servers map",
+            ?assertEqual(
+                [{"host1.domain", 1234}, {"host2.domain", 2345}, {"host3.domain", 3456}],
+                HoconParse("host1.domain:1234,host2.domain:2345,host3.domain:3456")
+            )
+        ),
+        ?T(
+            "no port expected valid port",
+            ?assertThrow(
+                "not_expecting_port_number",
+                emqx_schema:parse_server("localhost:80", #{no_port => true})
+            )
+        ),
+        ?T(
+            "no port expected invalid port",
+            ?assertThrow(
+                "not_expecting_port_number",
+                emqx_schema:parse_server("localhost:notaport", #{no_port => true})
+            )
+        ),
+
+        ?T(
+            "bad hostname",
+            ?assertThrow(
+                "expecting_hostname_but_got_a_number",
+                emqx_schema:parse_server(":80", #{default_port => 80})
+            )
+        ),
+        ?T(
+            "bad port",
+            ?assertThrow(
+                "bad_port_number",
+                emqx_schema:parse_server("host:33x", #{default_port => 33})
+            )
+        ),
+        ?T(
+            "bad host with port",
+            ?assertThrow(
+                "bad_host_port",
+                emqx_schema:parse_server("host:name:80", #{default_port => 80})
+            )
+        ),
+        ?T(
+            "bad schema",
+            ?assertError(
+                "bad_schema",
+                emqx_schema:parse_server("whatever", #{default_port => 10, no_port => true})
+            )
+        )
+    ].
+
+servers_validator_test() ->
+    Required = emqx_schema:servers_validator(#{}, true),
+    NotRequired = emqx_schema:servers_validator(#{}, false),
+    ?assertThrow("cannot_be_empty", Required("")),
+    ?assertThrow("cannot_be_empty", Required(<<>>)),
+    ?assertThrow("cannot_be_empty", Required(undefined)),
+    ?assertEqual(ok, NotRequired("")),
+    ?assertEqual(ok, NotRequired(<<>>)),
+    ?assertEqual(ok, NotRequired(undefined)),
+    ok.
+
+converter_invalid_input_test() ->
+    ?assertEqual(undefined, emqx_schema:convert_servers(undefined)),
+    %% 'foo: bar' is a valid HOCON value, but 'bar' is not a port number
+    ?assertThrow("bad_host_port", emqx_schema:convert_servers(#{foo => bar})).

apps/emqx_authn/test/emqx_authn_mysql_SUITE.erl

@@ -100,7 +100,6 @@ t_create_invalid(_Config) ->
 
     InvalidConfigs =
         [
-            maps:without([<<"server">>], AuthConfig),
             AuthConfig#{<<"server">> => <<"unknownhost:3333">>},
             AuthConfig#{<<"password">> => <<"wrongpass">>},
             AuthConfig#{<<"database">> => <<"wrongdatabase">>}
@@ -541,7 +540,7 @@ mysql_config() ->
         username => <<"root">>,
         password => <<"public">>,
         pool_size => 8,
-        server => {?MYSQL_HOST, ?MYSQL_DEFAULT_PORT},
+        server => <<?MYSQL_HOST>>,
         ssl => #{enable => false}
     }.
 

apps/emqx_authn/test/emqx_authn_pgsql_SUITE.erl

@@ -32,7 +32,11 @@
 -define(PATH, [authentication]).
 
 all() ->
-    [{group, require_seeds}, t_create_invalid].
+    [
+        {group, require_seeds},
+        t_update_with_invalid_config,
+        t_update_with_bad_config_value
+    ].
 
 groups() ->
     [{require_seeds, [], [t_create, t_authenticate, t_update, t_destroy, t_is_superuser]}].
@@ -96,12 +100,36 @@ t_create(_Config) ->
     {ok, [#{provider := emqx_authn_pgsql}]} = emqx_authentication:list_authenticators(?GLOBAL),
     emqx_authn_test_lib:delete_config(?ResourceID).
 
-t_create_invalid(_Config) ->
+%% invalid config which does not pass the schema check should result in an error
+t_update_with_invalid_config(_Config) ->
+    AuthConfig = raw_pgsql_auth_config(),
+    BadConfig = maps:without([<<"server">>], AuthConfig),
+    ?assertMatch(
+        {error,
+            {bad_authenticator_config, #{
+                reason :=
+                    {emqx_authn_pgsql, [
+                        #{
+                            kind := validation_error,
+                            path := "authentication.server",
+                            reason := required_field,
+                            value := undefined
+                        }
+                    ]}
+            }}},
+        emqx:update_config(
+            ?PATH,
+            {create_authenticator, ?GLOBAL, BadConfig}
+        )
+    ),
+    ok.
+
+%% bad config values may cause connection failure, but should still be able to update
+t_update_with_bad_config_value(_Config) ->
     AuthConfig = raw_pgsql_auth_config(),
 
     InvalidConfigs =
         [
-            maps:without([<<"server">>], AuthConfig),
             AuthConfig#{<<"server">> => <<"unknownhost:3333">>},
             AuthConfig#{<<"password">> => <<"wrongpass">>},
             AuthConfig#{<<"database">> => <<"wrongdatabase">>}
@@ -591,7 +619,7 @@ pgsql_config() ->
         username => <<"root">>,
         password => <<"public">>,
         pool_size => 8,
-        server => {?PGSQL_HOST, ?PGSQL_DEFAULT_PORT},
+        server => pgsql_server(),
         ssl => #{enable => false}
     }.
 

apps/emqx_authn/test/emqx_authn_redis_SUITE.erl

@@ -31,7 +31,12 @@
 -define(ResourceID, <<"password_based:redis">>).
 
 all() ->
-    [{group, require_seeds}, t_create, t_create_invalid].
+    [
+        {group, require_seeds},
+        t_create,
+        t_create_with_config_values_wont_work,
+        t_create_invalid_config
+    ].
 
 groups() ->
     [{require_seeds, [], [t_authenticate, t_update, t_destroy]}].
@@ -97,7 +102,7 @@ t_create(_Config) ->
 
     {ok, [#{provider := emqx_authn_redis}]} = emqx_authentication:list_authenticators(?GLOBAL).
 
-t_create_invalid(_Config) ->
+t_create_with_config_values_wont_work(_Config) ->
     AuthConfig = raw_redis_auth_config(),
     InvalidConfigs =
         [
@@ -131,7 +136,6 @@ t_create_invalid(_Config) ->
 
     InvalidConfigs1 =
         [
-            maps:without([<<"server">>], AuthConfig),
             AuthConfig#{<<"server">> => <<"unknownhost:3333">>},
             AuthConfig#{<<"password">> => <<"wrongpass">>},
             AuthConfig#{<<"database">> => <<"5678">>}
@@ -152,6 +156,22 @@ t_create_invalid(_Config) ->
         InvalidConfigs1
     ).
 
+t_create_invalid_config(_Config) ->
+    Config0 = raw_redis_auth_config(),
+    Config = maps:without([<<"server">>], Config0),
+    ?assertMatch(
+        {error,
+            {bad_authenticator_config, #{
+                reason := {emqx_authn_redis, [#{kind := validation_error}]}
+            }}},
+        emqx:update_config(?PATH, {create_authenticator, ?GLOBAL, Config})
+    ),
+    ?assertMatch([], emqx_config:get_raw([authentication])),
+    ?assertEqual(
+        {error, {not_found, {chain, ?GLOBAL}}},
+        emqx_authentication:list_authenticators(?GLOBAL)
+    ).
+
 t_authenticate(_Config) ->
     ok = lists:foreach(
         fun(Sample) ->
@@ -591,7 +611,7 @@ redis_config() ->
         pool_size => 8,
         redis_type => single,
         password => "public",
-        server => {?REDIS_HOST, ?REDIS_DEFAULT_PORT},
+        server => <<?REDIS_HOST>>,
         ssl => #{enable => false}
     }.
 

apps/emqx_authz/test/emqx_authz_mysql_SUITE.erl

@@ -317,8 +317,7 @@ raw_mysql_authz_config() ->
             "SELECT permission, action, topic "
             "FROM acl WHERE username = ${username}"
         >>,
-
+        <<"server">> => <<?MYSQL_HOST>>
-        <<"server">> => mysql_server()
     }.
 
 q(Sql) ->
@@ -385,9 +384,6 @@ setup_config(SpecialParams) ->
         SpecialParams
     ).
 
-mysql_server() ->
-    iolist_to_binary(io_lib:format("~s", [?MYSQL_HOST])).
-
 mysql_config() ->
     #{
         auto_reconnect => true,
@@ -395,7 +391,7 @@ mysql_config() ->
         username => <<"root">>,
         password => <<"public">>,
         pool_size => 8,
-        server => {?MYSQL_HOST, ?MYSQL_DEFAULT_PORT},
+        server => <<?MYSQL_HOST>>,
         ssl => #{enable => false}
     }.
 

apps/emqx_authz/test/emqx_authz_postgresql_SUITE.erl

@@ -322,8 +322,7 @@ raw_pgsql_authz_config() ->
             "SELECT permission, action, topic "
             "FROM acl WHERE username = ${username}"
         >>,
-
+        <<"server">> => <<?PGSQL_HOST>>
-        <<"server">> => pgsql_server()
     }.
 
 q(Sql) ->
@@ -393,9 +392,6 @@ setup_config(SpecialParams) ->
         SpecialParams
     ).
 
-pgsql_server() ->
-    iolist_to_binary(io_lib:format("~s", [?PGSQL_HOST])).
-
 pgsql_config() ->
     #{
         auto_reconnect => true,
@@ -403,7 +399,7 @@ pgsql_config() ->
         username => <<"root">>,
         password => <<"public">>,
         pool_size => 8,
-        server => {?PGSQL_HOST, ?PGSQL_DEFAULT_PORT},
+        server => <<?PGSQL_HOST>>,
         ssl => #{enable => false}
     }.
 

apps/emqx_authz/test/emqx_authz_redis_SUITE.erl

@@ -163,7 +163,7 @@ t_lookups(_Config) ->
 
 %% should still succeed to create even if the config will not work,
 %% because it's not a part of the schema check
-t_create_with_config_values_wont_works(_Config) ->
+t_create_with_config_values_wont_work(_Config) ->
     AuthzConfig = raw_redis_authz_config(),
 
     InvalidConfigs =
@@ -182,11 +182,15 @@ t_create_with_config_values_wont_works(_Config) ->
     ).
 
 %% creating without a require filed should return error
-t_create_invalid_schema(_Config) ->
+t_create_invalid_config(_Config) ->
     AuthzConfig = raw_redis_authz_config(),
     C = maps:without([<<"server">>], AuthzConfig),
     ?assertMatch(
-        {error, {emqx_conf_schema, _}},
+        {error, #{
+            kind := validation_error,
+            path := "authorization.sources.1",
+            discarded_errors_count := 0
+        }},
         emqx_authz:update(?CMD_REPLACE, [C])
     ).
 
@@ -255,12 +259,9 @@ raw_redis_authz_config() ->
         <<"cmd">> => <<"HGETALL mqtt_user:${username}">>,
         <<"database">> => <<"1">>,
         <<"password">> => <<"public">>,
-        <<"server">> => redis_server()
+        <<"server">> => <<?REDIS_HOST>>
     }.
 
-redis_server() ->
-    iolist_to_binary(io_lib:format("~s", [?REDIS_HOST])).
-
 q(Command) ->
     emqx_resource:query(
         ?REDIS_RESOURCE,
@@ -274,7 +275,7 @@ redis_config() ->
         pool_size => 8,
         redis_type => single,
         password => "public",
-        server => {?REDIS_HOST, ?REDIS_DEFAULT_PORT},
+        server => <<?REDIS_HOST>>,
         ssl => #{enable => false}
     }.
 

apps/emqx_connector/src/emqx_connector.app.src

@@ -1,6 +1,6 @@
 %% -*- mode: erlang -*-
 {application, emqx_connector, [
-    {description, "An OTP application"},
+    {description, "EMQX Data Integration Connectors"},
     {vsn, "0.1.11"},
     {registered, []},
     {mod, {emqx_connector_app, []}},

apps/emqx_connector/src/emqx_connector_ldap.erl

@@ -35,6 +35,11 @@
 -export([connect/1]).
 
 -export([search/4]).
+
+%% port is not expected from configuration because
+%% all servers expected to use the same port number
+-define(LDAP_HOST_OPTIONS, #{no_port => true}).
+
 %%=====================================================================
 roots() ->
     ldap_fields() ++ emqx_connector_schema_lib:ssl_fields().
@@ -63,12 +68,7 @@ on_start(
         connector => InstId,
         config => Config
     }),
-    Servers = [
+    Servers = emqx_schema:parse_servers(Servers0, ?LDAP_HOST_OPTIONS),
-        begin
-            proplists:get_value(host, S)
-        end
-     || S <- Servers0
-    ],
     SslOpts =
         case maps:get(enable, SSL) of
             true ->
@@ -86,8 +86,7 @@ on_start(
         {bind_password, BindPassword},
         {timeout, Timeout},
         {pool_size, PoolSize},
-        {auto_reconnect, reconn_interval(AutoReconn)},
+        {auto_reconnect, reconn_interval(AutoReconn)}
-        {servers, Servers}
     ],
     PoolName = emqx_plugin_libs_pool:pool_name(InstId),
     case emqx_plugin_libs_pool:start_pool(PoolName, ?MODULE, Opts ++ SslOpts) of
@@ -166,7 +165,7 @@ connect(Opts) ->
 
 ldap_fields() ->
     [
-        {servers, fun servers/1},
+        {servers, servers()},
         {port, fun port/1},
         {pool_size, fun emqx_connector_schema_lib:pool_size/1},
         {bind_dn, fun bind_dn/1},
@@ -175,11 +174,8 @@ ldap_fields() ->
         {auto_reconnect, fun emqx_connector_schema_lib:auto_reconnect/1}
     ].
 
-servers(type) -> list();
+servers() ->
-servers(validator) -> [?NOT_EMPTY("the value of the field 'servers' cannot be empty")];
+    emqx_schema:servers_sc(#{}, ?LDAP_HOST_OPTIONS).
-servers(converter) -> fun to_servers_raw/1;
-servers(required) -> true;
-servers(_) -> undefined.
 
 bind_dn(type) -> binary();
 bind_dn(default) -> 0;
@@ -191,24 +187,3 @@ port(_) -> undefined.
 
 duration(type) -> emqx_schema:duration_ms();
 duration(_) -> undefined.
-
-to_servers_raw(Servers) ->
-    {ok,
-        lists:map(
-            fun(Server) ->
-                case string:tokens(Server, ": ") of
-                    [Ip] ->
-                        [{host, Ip}];
-                    [Ip, Port] ->
-                        [{host, Ip}, {port, list_to_integer(Port)}]
-                end
-            end,
-            string:tokens(str(Servers), ", ")
-        )}.
-
-str(A) when is_atom(A) ->
-    atom_to_list(A);
-str(B) when is_binary(B) ->
-    binary_to_list(B);
-str(S) when is_list(S) ->
-    S.

apps/emqx_connector/src/emqx_connector_lib.erl

@@ -0,0 +1,22 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(emqx_connector_lib).
+
+-export([resolve_dns/2]).
+
+%% @doc Mostly for meck.
+resolve_dns(DNS, Type) ->
+    inet_res:lookup(DNS, in, Type).

apps/emqx_connector/src/emqx_connector_mongo.erl

@@ -39,18 +39,16 @@
 
 -export([mongo_query/5, mongo_insert/3, check_worker_health/1]).
 
+%% for testing
+-export([maybe_resolve_srv_and_txt_records/1]).
+
 -define(HEALTH_CHECK_TIMEOUT, 30000).
 
 %% mongo servers don't need parse
 -define(MONGO_HOST_OPTIONS, #{
-    host_type => hostname,
     default_port => ?MONGO_DEFAULT_PORT
 }).
 
--ifdef(TEST).
--export([to_servers_raw/1]).
--endif.
-
 %%=====================================================================
 roots() ->
     [
@@ -73,7 +71,7 @@ fields(single) ->
             required => true,
             desc => ?DESC("single_mongo_type")
         }},
-        {server, fun server/1},
+        {server, server()},
         {w_mode, fun w_mode/1}
     ] ++ mongo_fields();
 fields(rs) ->
@@ -84,7 +82,7 @@ fields(rs) ->
             required => true,
             desc => ?DESC("rs_mongo_type")
         }},
-        {servers, fun servers/1},
+        {servers, servers()},
         {w_mode, fun w_mode/1},
         {r_mode, fun r_mode/1},
         {replica_set_name, fun replica_set_name/1}
@@ -97,7 +95,7 @@ fields(sharded) ->
             required => true,
             desc => ?DESC("sharded_mongo_type")
         }},
-        {servers, fun servers/1},
+        {servers, servers()},
         {w_mode, fun w_mode/1}
     ] ++ mongo_fields();
 fields(topology) ->
@@ -161,7 +159,7 @@ on_start(
             sharded -> "starting_mongodb_sharded_connector"
         end,
     ?SLOG(info, #{msg => Msg, connector => InstId, config => Config}),
-    NConfig = #{hosts := Hosts} = may_parse_srv_and_txt_records(Config),
+    NConfig = #{hosts := Hosts} = maybe_resolve_srv_and_txt_records(Config),
     SslOpts =
         case maps:get(enable, SSL) of
             true ->
@@ -387,19 +385,13 @@ init_worker_options([], Acc) ->
 %% ===================================================================
 %% Schema funcs
 
-server(type) -> emqx_schema:host_port();
+server() ->
-server(required) -> true;
+    Meta = #{desc => ?DESC("server")},
-server(validator) -> [?NOT_EMPTY("the value of the field 'server' cannot be empty")];
+    emqx_schema:servers_sc(Meta, ?MONGO_HOST_OPTIONS).
-server(converter) -> fun to_server_raw/1;
-server(desc) -> ?DESC("server");
-server(_) -> undefined.
 
-servers(type) -> list();
+servers() ->
-servers(required) -> true;
+    Meta = #{desc => ?DESC("servers")},
-servers(validator) -> [?NOT_EMPTY("the value of the field 'servers' cannot be empty")];
+    emqx_schema:servers_sc(Meta, ?MONGO_HOST_OPTIONS).
-servers(converter) -> fun to_servers_raw/1;
-servers(desc) -> ?DESC("servers");
-servers(_) -> undefined.
 
 w_mode(type) -> hoconsc:enum([unsafe, safe]);
 w_mode(desc) -> ?DESC("w_mode");
@@ -434,163 +426,109 @@ srv_record(_) -> undefined.
 %% ===================================================================
 %% Internal funcs
 
-may_parse_srv_and_txt_records(#{server := Server} = Config) ->
+maybe_resolve_srv_and_txt_records(#{server := Server} = Config) ->
     NConfig = maps:remove(server, Config),
-    may_parse_srv_and_txt_records_(NConfig#{servers => [Server]});
+    maybe_resolve_srv_and_txt_records1(Server, NConfig);
-may_parse_srv_and_txt_records(Config) ->
+maybe_resolve_srv_and_txt_records(#{servers := Servers} = Config) ->
-    may_parse_srv_and_txt_records_(Config).
+    NConfig = maps:remove(servers, Config),
+    maybe_resolve_srv_and_txt_records1(Servers, NConfig).
 
-may_parse_srv_and_txt_records_(
+maybe_resolve_srv_and_txt_records1(
+    Servers0,
     #{
         mongo_type := Type,
-        srv_record := false,
+        srv_record := false
-        servers := Servers
     } = Config
 ) ->
     case Type =:= rs andalso maps:is_key(replica_set_name, Config) =:= false of
         true ->
-            error({missing_parameter, replica_set_name});
+            throw(#{
+                reason => "missing_parameter",
+                param => replica_set_name
+            });
         false ->
-            Config#{hosts => servers_to_bin(lists:flatten(Servers))}
+            Servers = parse_servers(Servers0),
+            Config#{hosts => format_hosts(Servers)}
     end;
-may_parse_srv_and_txt_records_(
+maybe_resolve_srv_and_txt_records1(
+    Servers,
     #{
         mongo_type := Type,
-        srv_record := true,
+        srv_record := true
-        servers := Servers
     } = Config
 ) ->
-    Hosts = parse_srv_records(Type, Servers),
+    %% when srv is in use, it's typically only one DNS resolution needed,
-    ExtraOpts = parse_txt_records(Type, Servers),
+    %% however, by the schema definition, it's allowed to configure more than one.
+    %% here we keep only the fist
+    [{DNS, _IgnorePort} | _] = parse_servers(Servers),
+    DnsRecords = resolve_srv_records(DNS),
+    Hosts = format_hosts(DnsRecords),
+    ?tp(info, resolved_srv_records, #{dns => DNS, resolved_hosts => Hosts}),
+    ExtraOpts = resolve_txt_records(Type, DNS),
+    ?tp(info, resolved_txt_records, #{dns => DNS, resolved_options => ExtraOpts}),
     maps:merge(Config#{hosts => Hosts}, ExtraOpts).
 
-parse_srv_records(Type, Servers) ->
+resolve_srv_records(DNS0) ->
-    Fun = fun(AccIn, {IpOrHost, _Port}) ->
+    DNS = "_mongodb._tcp." ++ DNS0,
-        case
+    DnsData = emqx_connector_lib:resolve_dns(DNS, srv),
-            inet_res:lookup(
+    case [{Host, Port} || {_, _, Port, Host} <- DnsData] of
-                "_mongodb._tcp." ++
+        [] ->
-                    ip_or_host_to_string(IpOrHost),
+            throw(#{
-                in,
+                reason => "failed_to_resolve_srv_record",
-                srv
+                dns => DNS
-            )
+            });
-        of
+        L ->
-            [] ->
+            L
-                error(service_not_found);
-            Services ->
-                [
-                    [server_to_bin({Host, Port}) || {_, _, Port, Host} <- Services]
-                    | AccIn
-                ]
-        end
-    end,
-    Res = lists:foldl(Fun, [], Servers),
-    case Type of
-        single -> lists:nth(1, Res);
-        _ -> Res
     end.
 
-parse_txt_records(Type, Servers) ->
+resolve_txt_records(Type, DNS) ->
-    Fields =
+    case emqx_connector_lib:resolve_dns(DNS, txt) of
-        case Type of
+        [] ->
-            rs -> ["authSource", "replicaSet"];
+            #{};
-            _ -> ["authSource"]
+        [[QueryString]] = L ->
-        end,
+            %% e.g. "authSource=admin&replicaSet=atlas-wrnled-shard-0"
-    Fun = fun(AccIn, {IpOrHost, _Port}) ->
+            case uri_string:dissect_query(QueryString) of
-        case inet_res:lookup(IpOrHost, in, txt) of
+                {error, _, _} ->
-            [] ->
+                    throw(#{
-                #{};
+                        reason => "bad_txt_record_resolution",
-            [[QueryString]] ->
+                        resolved => L
-                case uri_string:dissect_query(QueryString) of
+                    });
-                    {error, _, _} ->
+                Options ->
-                        error({invalid_txt_record, invalid_query_string});
+                    convert_options(Type, normalize_options(Options))
-                    Options ->
+            end;
-                        maps:merge(AccIn, take_and_convert(Fields, Options))
+        L ->
-                end;
+            throw(#{
-            _ ->
+                reason => "multiple_txt_records",
-                error({invalid_txt_record, multiple_records})
+                resolved => L
-        end
+            })
-    end,
+    end.
-    lists:foldl(Fun, #{}, Servers).
 
-take_and_convert(Fields, Options) ->
+normalize_options([]) ->
-    take_and_convert(Fields, Options, #{}).
+    [];
+normalize_options([{Name, Value} | Options]) ->
+    [{string:lowercase(Name), Value} | normalize_options(Options)].
 
-take_and_convert([], [_ | _], _Acc) ->
+convert_options(rs, Options) ->
-    error({invalid_txt_record, invalid_option});
+    M1 = maybe_add_option(auth_source, "authSource", Options),
-take_and_convert([], [], Acc) ->
+    M2 = maybe_add_option(replica_set_name, "replicaSet", Options),
-    Acc;
+    maps:merge(M1, M2);
-take_and_convert([Field | More], Options, Acc) ->
+convert_options(_, Options) ->
-    case lists:keytake(Field, 1, Options) of
+    maybe_add_option(auth_source, "authSource", Options).
-        {value, {"authSource", V}, NOptions} ->
+
-            take_and_convert(More, NOptions, Acc#{auth_source => list_to_binary(V)});
+maybe_add_option(ConfigKey, OptName0, Options) ->
-        {value, {"replicaSet", V}, NOptions} ->
+    OptName = string:lowercase(OptName0),
-            take_and_convert(More, NOptions, Acc#{replica_set_name => list_to_binary(V)});
+    case lists:keyfind(OptName, 1, Options) of
-        {value, _, _} ->
+        {_, OptValue} ->
-            error({invalid_txt_record, invalid_option});
+            #{ConfigKey => iolist_to_binary(OptValue)};
         false ->
-            take_and_convert(More, Options, Acc)
+            #{}
     end.
 
--spec ip_or_host_to_string(binary() | string() | tuple()) ->
+format_host({Host, Port}) ->
-    string().
+    iolist_to_binary([Host, ":", integer_to_list(Port)]).
-ip_or_host_to_string(Ip) when is_tuple(Ip) ->
-    inet:ntoa(Ip);
-ip_or_host_to_string(Host) ->
-    str(Host).
 
-servers_to_bin([Server | Rest]) ->
+format_hosts(Hosts) ->
-    [server_to_bin(Server) | servers_to_bin(Rest)];
+    lists:map(fun format_host/1, Hosts).
-servers_to_bin([]) ->
-    [].
 
-server_to_bin({IpOrHost, Port}) ->
+parse_servers(HoconValue) ->
-    iolist_to_binary(ip_or_host_to_string(IpOrHost) ++ ":" ++ integer_to_list(Port)).
+    emqx_schema:parse_servers(HoconValue, ?MONGO_HOST_OPTIONS).
-
-%% ===================================================================
-%% typereflt funcs
-
--spec to_server_raw(string()) ->
-    {string(), pos_integer()}.
-to_server_raw(Server) ->
-    emqx_connector_schema_lib:parse_server(Server, ?MONGO_HOST_OPTIONS).
-
--spec to_servers_raw(string()) ->
-    [{string(), pos_integer()}].
-to_servers_raw(Servers) ->
-    lists:map(
-        fun(Server) ->
-            emqx_connector_schema_lib:parse_server(Server, ?MONGO_HOST_OPTIONS)
-        end,
-        split_servers(Servers)
-    ).
-
-split_servers(L) when is_list(L) ->
-    PossibleTypes = [
-        list(binary()),
-        list(string()),
-        string()
-    ],
-    TypeChecks = lists:map(fun(T) -> typerefl:typecheck(T, L) end, PossibleTypes),
-    case TypeChecks of
-        [ok, _, _] ->
-            %% list(binary())
-            lists:map(fun binary_to_list/1, L);
-        [_, ok, _] ->
-            %% list(string())
-            L;
-        [_, _, ok] ->
-            %% string()
-            string:tokens(L, ", ");
-        [_, _, _] ->
-            %% invalid input
-            throw("List of servers must contain only strings")
-    end;
-split_servers(B) when is_binary(B) ->
-    string:tokens(str(B), ", ").
-
-str(A) when is_atom(A) ->
-    atom_to_list(A);
-str(B) when is_binary(B) ->
-    binary_to_list(B);
-str(S) when is_list(S) ->
-    S.

apps/emqx_connector/src/emqx_connector_mysql.erl

@@ -43,7 +43,6 @@
 -export([do_get_status/1]).
 
 -define(MYSQL_HOST_OPTIONS, #{
-    host_type => inet_addr,
     default_port => ?MYSQL_DEFAULT_PORT
 }).
 
@@ -66,17 +65,14 @@ roots() ->
     [{config, #{type => hoconsc:ref(?MODULE, config)}}].
 
 fields(config) ->
-    [{server, fun server/1}] ++
+    [{server, server()}] ++
         emqx_connector_schema_lib:relational_db_fields() ++
         emqx_connector_schema_lib:ssl_fields() ++
         emqx_connector_schema_lib:prepare_statement_fields().
 
-server(type) -> emqx_schema:host_port();
+server() ->
-server(required) -> true;
+    Meta = #{desc => ?DESC("server")},
-server(validator) -> [?NOT_EMPTY("the value of the field 'server' cannot be empty")];
+    emqx_schema:servers_sc(Meta, ?MYSQL_HOST_OPTIONS).
-server(converter) -> fun to_server/1;
-server(desc) -> ?DESC("server");
-server(_) -> undefined.
 
 %% ===================================================================
 callback_mode() -> always_sync.
@@ -85,7 +81,7 @@ callback_mode() -> always_sync.
 on_start(
     InstId,
     #{
-        server := {Host, Port},
+        server := Server,
         database := DB,
         username := User,
         password := Password,
@@ -94,6 +90,7 @@ on_start(
         ssl := SSL
     } = Config
 ) ->
+    {Host, Port} = emqx_schema:parse_server(Server, ?MYSQL_HOST_OPTIONS),
     ?SLOG(info, #{
         msg => "starting_mysql_connector",
         connector => InstId,
@@ -239,11 +236,6 @@ reconn_interval(false) -> false.
 connect(Options) ->
     mysql:start_link(Options).
 
--spec to_server(string()) ->
-    {inet:ip_address() | inet:hostname(), pos_integer()}.
-to_server(Str) ->
-    emqx_connector_schema_lib:parse_server(Str, ?MYSQL_HOST_OPTIONS).
-
 init_prepare(State = #{prepare_statement := Prepares, poolname := PoolName}) ->
     case maps:size(Prepares) of
         0 ->

apps/emqx_connector/src/emqx_connector_pgsql.erl

@@ -44,7 +44,6 @@
 -export([do_get_status/1]).
 
 -define(PGSQL_HOST_OPTIONS, #{
-    host_type => inet_addr,
     default_port => ?PGSQL_DEFAULT_PORT
 }).
 
@@ -54,17 +53,14 @@ roots() ->
     [{config, #{type => hoconsc:ref(?MODULE, config)}}].
 
 fields(config) ->
-    [{server, fun server/1}] ++
+    [{server, server()}] ++
         emqx_connector_schema_lib:relational_db_fields() ++
         emqx_connector_schema_lib:ssl_fields() ++
         emqx_connector_schema_lib:prepare_statement_fields().
 
-server(type) -> emqx_schema:host_port();
+server() ->
-server(required) -> true;
+    Meta = #{desc => ?DESC("server")},
-server(validator) -> [?NOT_EMPTY("the value of the field 'server' cannot be empty")];
+    emqx_schema:servers_sc(Meta, ?PGSQL_HOST_OPTIONS).
-server(converter) -> fun to_server/1;
-server(desc) -> ?DESC("server");
-server(_) -> undefined.
 
 %% ===================================================================
 callback_mode() -> always_sync.
@@ -72,7 +68,7 @@ callback_mode() -> always_sync.
 on_start(
     InstId,
     #{
-        server := {Host, Port},
+        server := Server,
         database := DB,
         username := User,
         password := Password,
@@ -81,6 +77,7 @@ on_start(
         ssl := SSL
     } = Config
 ) ->
+    {Host, Port} = emqx_schema:parse_server(Server, ?PGSQL_HOST_OPTIONS),
     ?SLOG(info, #{
         msg => "starting_postgresql_connector",
         connector => InstId,
@@ -209,11 +206,3 @@ conn_opts([Opt = {ssl_opts, _} | Opts], Acc) ->
     conn_opts(Opts, [Opt | Acc]);
 conn_opts([_Opt | Opts], Acc) ->
     conn_opts(Opts, Acc).
-
-%% ===================================================================
-%% typereflt funcs
-
--spec to_server(string()) ->
-    {inet:ip_address() | inet:hostname(), pos_integer()}.
-to_server(Str) ->
-    emqx_connector_schema_lib:parse_server(Str, ?PGSQL_HOST_OPTIONS).

apps/emqx_connector/src/emqx_connector_redis.erl

@@ -41,7 +41,6 @@
 
 %% redis host don't need parse
 -define(REDIS_HOST_OPTIONS, #{
-    host_type => hostname,
     default_port => ?REDIS_DEFAULT_PORT
 }).
 
@@ -61,7 +60,7 @@ roots() ->
 
 fields(single) ->
     [
-        {server, fun server/1},
+        {server, server()},
         {redis_type, #{
             type => single,
             default => single,
@@ -73,7 +72,7 @@ fields(single) ->
         emqx_connector_schema_lib:ssl_fields();
 fields(cluster) ->
     [
-        {servers, fun servers/1},
+        {servers, servers()},
         {redis_type, #{
             type => cluster,
             default => cluster,
@@ -85,7 +84,7 @@ fields(cluster) ->
         emqx_connector_schema_lib:ssl_fields();
 fields(sentinel) ->
     [
-        {servers, fun servers/1},
+        {servers, servers()},
         {redis_type, #{
             type => sentinel,
             default => sentinel,
@@ -101,21 +100,16 @@ fields(sentinel) ->
         redis_fields() ++
         emqx_connector_schema_lib:ssl_fields().
 
-server(type) -> emqx_schema:host_port();
+server() ->
-server(required) -> true;
+    Meta = #{desc => ?DESC("server")},
-server(validator) -> [?NOT_EMPTY("the value of the field 'server' cannot be empty")];
+    emqx_schema:servers_sc(Meta, ?REDIS_HOST_OPTIONS).
-server(converter) -> fun to_server_raw/1;
-server(desc) -> ?DESC("server");
-server(_) -> undefined.
 
-servers(type) -> list();
+servers() ->
-servers(required) -> true;
+    Meta = #{desc => ?DESC("servers")},
-servers(validator) -> [?NOT_EMPTY("the value of the field 'servers' cannot be empty")];
+    emqx_schema:servers_sc(Meta, ?REDIS_HOST_OPTIONS).
-servers(converter) -> fun to_servers_raw/1;
-servers(desc) -> ?DESC("servers");
-servers(_) -> undefined.
 
 %% ===================================================================
+
 callback_mode() -> always_sync.
 
 on_start(
@@ -132,11 +126,13 @@ on_start(
         connector => InstId,
         config => Config
     }),
-    Servers =
+    ConfKey =
         case Type of
-            single -> [{servers, [maps:get(server, Config)]}];
+            single -> server;
-            _ -> [{servers, maps:get(servers, Config)}]
+            _ -> servers
         end,
+    Servers0 = maps:get(ConfKey, Config),
+    Servers = [{servers, emqx_schema:parse_servers(Servers0, ?REDIS_HOST_OPTIONS)}],
     Database =
         case Type of
             cluster -> [];
@@ -299,25 +295,3 @@ redis_fields() ->
         }},
         {auto_reconnect, fun emqx_connector_schema_lib:auto_reconnect/1}
     ].
-
--spec to_server_raw(string()) ->
-    {string(), pos_integer()}.
-to_server_raw(Server) ->
-    emqx_connector_schema_lib:parse_server(Server, ?REDIS_HOST_OPTIONS).
-
--spec to_servers_raw(string()) ->
-    [{string(), pos_integer()}].
-to_servers_raw(Servers) ->
-    lists:map(
-        fun(Server) ->
-            emqx_connector_schema_lib:parse_server(Server, ?REDIS_HOST_OPTIONS)
-        end,
-        string:tokens(str(Servers), ", ")
-    ).
-
-str(A) when is_atom(A) ->
-    atom_to_list(A);
-str(B) when is_binary(B) ->
-    binary_to_list(B);
-str(S) when is_list(S) ->
-    S.

apps/emqx_connector/src/emqx_connector_schema_lib.erl

@@ -25,10 +25,6 @@
     prepare_statement_fields/0
 ]).
 
--export([
-    parse_server/2
-]).
-
 -export([
     pool_size/1,
     database/1,
@@ -111,53 +107,3 @@ auto_reconnect(type) -> boolean();
 auto_reconnect(desc) -> ?DESC("auto_reconnect");
 auto_reconnect(default) -> true;
 auto_reconnect(_) -> undefined.
-
-parse_server(Str, #{host_type := inet_addr, default_port := DefaultPort}) ->
-    case string:tokens(str(Str), ": ") of
-        [Ip, Port] ->
-            {parse_ip(Ip), parse_port(Port)};
-        [Ip] ->
-            {parse_ip(Ip), DefaultPort};
-        _ ->
-            throw("Bad server schema")
-    end;
-parse_server(Str, #{host_type := hostname, default_port := DefaultPort}) ->
-    case string:tokens(str(Str), ": ") of
-        [Hostname, Port] ->
-            {Hostname, parse_port(Port)};
-        [Hostname] ->
-            {Hostname, DefaultPort};
-        _ ->
-            throw("Bad server schema")
-    end;
-parse_server(_, _) ->
-    throw("Invalid Host").
-
-parse_ip(Str) ->
-    case inet:parse_address(Str) of
-        {ok, R} ->
-            R;
-        _ ->
-            %% check is a rfc1035's hostname
-            case inet_parse:domain(Str) of
-                true ->
-                    Str;
-                _ ->
-                    throw("Bad IP or Host")
-            end
-    end.
-
-parse_port(Port) ->
-    try
-        list_to_integer(Port)
-    catch
-        _:_ ->
-            throw("Bad port number")
-    end.
-
-str(A) when is_atom(A) ->
-    atom_to_list(A);
-str(B) when is_binary(B) ->
-    binary_to_list(B);
-str(S) when is_list(S) ->
-    S.

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_mod.erl

@@ -51,15 +51,15 @@
 
 start(Config) ->
     Parent = self(),
-    {Host, Port} = maps:get(server, Config),
+    ServerStr = iolist_to_binary(maps:get(server, Config)),
+    {Server, Port} = emqx_connector_mqtt_schema:parse_server(ServerStr),
     Mountpoint = maps:get(receive_mountpoint, Config, undefined),
     Subscriptions = maps:get(subscriptions, Config, undefined),
     Vars = emqx_connector_mqtt_msg:make_pub_vars(Mountpoint, Subscriptions),
-    ServerStr = ip_port_to_server_str(Host, Port),
     Handlers = make_hdlr(Parent, Vars, #{server => ServerStr}),
     Config1 = Config#{
         msg_handler => Handlers,
-        host => Host,
+        host => Server,
         port => Port,
         force_ping => true,
         proto_ver => maps:get(proto_ver, Config, v4)
@@ -234,11 +234,3 @@ printable_maps(Headers) ->
         #{},
         Headers
     ).
-
-ip_port_to_server_str(Host, Port) ->
-    HostStr =
-        case inet:ntoa(Host) of
-            {error, einval} -> Host;
-            IPStr -> IPStr
-        end,
-    list_to_binary(io_lib:format("~s:~w", [HostStr, Port])).

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_schema.erl

@@ -25,13 +25,16 @@
     namespace/0,
     roots/0,
     fields/1,
-    desc/1
+    desc/1,
+    parse_server/1
 ]).
 
 -import(emqx_schema, [mk_duration/2]).
 
 -import(hoconsc, [mk/2, ref/2]).
 
+-define(MQTT_HOST_OPTS, #{default_port => 1883}).
+
 namespace() -> "connector-mqtt".
 
 roots() ->
@@ -67,14 +70,7 @@ fields("server_configs") ->
                     desc => ?DESC("mode")
                 }
             )},
-        {server,
+        {server, emqx_schema:servers_sc(#{desc => ?DESC("server")}, ?MQTT_HOST_OPTS)},
-            mk(
-                emqx_schema:host_port(),
-                #{
-                    required => true,
-                    desc => ?DESC("server")
-                }
-            )},
         {clientid_prefix, mk(binary(), #{required => false, desc => ?DESC("clientid_prefix")})},
         {reconnect_interval,
             mk_duration(
@@ -299,3 +295,6 @@ desc(_) ->
 
 qos() ->
     hoconsc:union([emqx_schema:qos(), binary()]).
+
+parse_server(Str) ->
+    emqx_schema:parse_server(Str, ?MQTT_HOST_OPTS).

apps/emqx_connector/test/emqx_connector_mongo_tests.erl

@@ -18,151 +18,135 @@
 
 -include_lib("eunit/include/eunit.hrl").
 
--define(DEFAULT_MONGO_PORT, 27017).
+srv_record_test() ->
+    with_dns_mock(
+        fun normal_dns_resolution_mock/2,
+        fun() ->
+            Single = single_config(),
+            Rs = simple_rs_config(),
+            Hosts = [
+                <<"cluster0-shard-00-02.zkemc.mongodb.net:27017">>,
+                <<"cluster0-shard-00-01.zkemc.mongodb.net:27017">>,
+                <<"cluster0-shard-00-00.zkemc.mongodb.net:27017">>
+            ],
+            ?assertMatch(
+                #{
+                    hosts := Hosts,
+                    auth_source := <<"admin">>
+                },
+                resolve(Single)
+            ),
+            ?assertMatch(
+                #{
+                    hosts := Hosts,
+                    auth_source := <<"admin">>,
+                    replica_set_name := <<"atlas-wrnled-shard-0">>
+                },
+                resolve(Rs)
+            ),
+            ok
+        end
+    ).
 
-%%------------------------------------------------------------------------------
+empty_srv_record_test() ->
-%% Helper fns
+    with_dns_mock(
-%%------------------------------------------------------------------------------
+        bad_srv_record_mock(_DnsResolution = []),
+        fun() ->
+            ?assertThrow(#{reason := "failed_to_resolve_srv_record"}, resolve(simple_rs_config()))
+        end
+    ).
 
-%%------------------------------------------------------------------------------
+empty_txt_record_test() ->
-%% Test cases
+    with_dns_mock(
-%%------------------------------------------------------------------------------
+        bad_txt_record_mock(_DnsResolution = []),
+        fun() ->
+            Config = resolve(single_config()),
+            ?assertNot(maps:is_key(auth_source, Config)),
+            ?assertNot(maps:is_key(replica_set_name, Config)),
+            ok
+        end
+    ).
 
-to_servers_raw_test_() ->
+multiple_txt_records_test() ->
+    with_dns_mock(
+        bad_txt_record_mock(_DnsResolution = [1, 2]),
+        fun() ->
+            ?assertThrow(#{reason := "multiple_txt_records"}, resolve(simple_rs_config()))
+        end
+    ).
+
+bad_query_string_test() ->
+    with_dns_mock(
+        bad_txt_record_mock(_DnsResolution = [["%-111"]]),
+        fun() ->
+            ?assertThrow(#{reason := "bad_txt_record_resolution"}, resolve(simple_rs_config()))
+        end
+    ).
+
+resolve(Config) ->
+    emqx_connector_mongo:maybe_resolve_srv_and_txt_records(Config).
+
+checked_config(Hocon) ->
+    {ok, Config} = hocon:binary(Hocon),
+    hocon_tconf:check_plain(
+        emqx_connector_mongo,
+        #{<<"config">> => Config},
+        #{atom_key => true}
+    ).
+
+simple_rs_config() ->
+    #{config := Rs} = checked_config(
+        "mongo_type = rs\n"
+        "servers = \"cluster0.zkemc.mongodb.net:27017\"\n"
+        "srv_record = true\n"
+        "database = foobar\n"
+        "replica_set_name = configured_replicaset_name\n"
+    ),
+    Rs.
+
+single_config() ->
+    #{config := Single} = checked_config(
+        "mongo_type = single\n"
+        "server = \"cluster0.zkemc.mongodb.net:27017,cluster0.zkemc.mongodb.net:27017\"\n"
+        "srv_record = true\n"
+        "database = foobar\n"
+    ),
+    Single.
+
+normal_srv_resolution() ->
     [
-        {"single server, binary, no port",
+        {0, 0, 27017, "cluster0-shard-00-02.zkemc.mongodb.net"},
-            ?_test(
+        {0, 0, 27017, "cluster0-shard-00-01.zkemc.mongodb.net"},
-                ?assertEqual(
+        {0, 0, 27017, "cluster0-shard-00-00.zkemc.mongodb.net"}
-                    [{"localhost", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw(<<"localhost">>)
-                )
-            )},
-        {"single server, string, no port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw("localhost")
-                )
-            )},
-        {"single server, list(binary), no port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw([<<"localhost">>])
-                )
-            )},
-        {"single server, list(string), no port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw(["localhost"])
-                )
-            )},
-        %%%%%%%%%
-        {"single server, binary, with port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", 9999}], emqx_connector_mongo:to_servers_raw(<<"localhost:9999">>)
-                )
-            )},
-        {"single server, string, with port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", 9999}], emqx_connector_mongo:to_servers_raw("localhost:9999")
-                )
-            )},
-        {"single server, list(binary), with port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", 9999}],
-                    emqx_connector_mongo:to_servers_raw([<<"localhost:9999">>])
-                )
-            )},
-        {"single server, list(string), with port",
-            ?_test(
-                ?assertEqual(
-                    [{"localhost", 9999}], emqx_connector_mongo:to_servers_raw(["localhost:9999"])
-                )
-            )},
-        %%%%%%%%%
-        {"multiple servers, string, no port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", ?DEFAULT_MONGO_PORT}, {"host2", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw("host1, host2")
-                )
-            )},
-        {"multiple servers, binary, no port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", ?DEFAULT_MONGO_PORT}, {"host2", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw(<<"host1, host2">>)
-                )
-            )},
-        {"multiple servers, list(string), no port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", ?DEFAULT_MONGO_PORT}, {"host2", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw(["host1", "host2"])
-                )
-            )},
-        {"multiple servers, list(binary), no port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", ?DEFAULT_MONGO_PORT}, {"host2", ?DEFAULT_MONGO_PORT}],
-                    emqx_connector_mongo:to_servers_raw([<<"host1">>, <<"host2">>])
-                )
-            )},
-        %%%%%%%%%
-        {"multiple servers, string, with port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", 1234}, {"host2", 2345}],
-                    emqx_connector_mongo:to_servers_raw("host1:1234, host2:2345")
-                )
-            )},
-        {"multiple servers, binary, with port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", 1234}, {"host2", 2345}],
-                    emqx_connector_mongo:to_servers_raw(<<"host1:1234, host2:2345">>)
-                )
-            )},
-        {"multiple servers, list(string), with port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", 1234}, {"host2", 2345}],
-                    emqx_connector_mongo:to_servers_raw(["host1:1234", "host2:2345"])
-                )
-            )},
-        {"multiple servers, list(binary), with port",
-            ?_test(
-                ?assertEqual(
-                    [{"host1", 1234}, {"host2", 2345}],
-                    emqx_connector_mongo:to_servers_raw([<<"host1:1234">>, <<"host2:2345">>])
-                )
-            )},
-        %%%%%%%%
-        {"multiple servers, invalid list(string)",
-            ?_test(
-                ?assertThrow(
-                    _,
-                    emqx_connector_mongo:to_servers_raw(["host1, host2"])
-                )
-            )},
-        {"multiple servers, invalid list(binary)",
-            ?_test(
-                ?assertThrow(
-                    _,
-                    emqx_connector_mongo:to_servers_raw([<<"host1, host2">>])
-                )
-            )},
-        %% TODO: handle this case??
-        {"multiple servers, mixed list(binary|string)",
-            ?_test(
-                ?assertThrow(
-                    _,
-                    emqx_connector_mongo:to_servers_raw([<<"host1">>, "host2"])
-                )
-            )}
     ].
+
+normal_txt_resolution() ->
+    [["authSource=admin&replicaSet=atlas-wrnled-shard-0"]].
+
+normal_dns_resolution_mock("_mongodb._tcp.cluster0.zkemc.mongodb.net", srv) ->
+    normal_srv_resolution();
+normal_dns_resolution_mock("cluster0.zkemc.mongodb.net", txt) ->
+    normal_txt_resolution().
+
+bad_srv_record_mock(DnsResolution) ->
+    fun("_mongodb._tcp.cluster0.zkemc.mongodb.net", srv) ->
+        DnsResolution
+    end.
+
+bad_txt_record_mock(DnsResolution) ->
+    fun
+        ("_mongodb._tcp.cluster0.zkemc.mongodb.net", srv) ->
+            normal_srv_resolution();
+        ("cluster0.zkemc.mongodb.net", txt) ->
+            DnsResolution
+    end.
+
+with_dns_mock(MockFn, TestFn) ->
+    meck:new(emqx_connector_lib, [non_strict, passthrough, no_history, no_link]),
+    meck:expect(emqx_connector_lib, resolve_dns, MockFn),
+    try
+        TestFn()
+    after
+        meck:unload(emqx_connector_lib)
+    end,
+    ok.

apps/emqx_connector/test/emqx_connector_mqtt_tests.erl

@@ -50,8 +50,8 @@ send_and_ack_test() ->
     try
         Max = 1,
         Batch = lists:seq(1, Max),
-        {ok, Conn} = emqx_connector_mqtt_mod:start(#{server => {{127, 0, 0, 1}, 1883}}),
+        {ok, Conn} = emqx_connector_mqtt_mod:start(#{server => "127.0.0.1:1883"}),
-        %     %% return last packet id as batch reference
+        %% return last packet id as batch reference
         {ok, _AckRef} = emqx_connector_mqtt_mod:send(Conn, Batch),
 
         ok = emqx_connector_mqtt_mod:stop(Conn)

apps/emqx_dashboard/src/emqx_dashboard.app.src

@@ -2,7 +2,7 @@
 {application, emqx_dashboard, [
     {description, "EMQX Web Dashboard"},
     % strict semver, bump manually!
-    {vsn, "5.0.10"},
+    {vsn, "5.0.11"},
     {modules, []},
     {registered, [emqx_dashboard_sup]},
     {applications, [kernel, stdlib, mnesia, minirest, emqx]},

apps/emqx_dashboard/src/emqx_dashboard_swagger.erl

@@ -675,8 +675,6 @@ typename_to_spec("file()", _Mod) ->
     #{type => string, example => <<"/path/to/file">>};
 typename_to_spec("ip_port()", _Mod) ->
     #{type => string, example => <<"127.0.0.1:80">>};
-typename_to_spec("host_port()", _Mod) ->
-    #{type => string, example => <<"example.host.domain:80">>};
 typename_to_spec("write_syntax()", _Mod) ->
     #{
         type => string,

apps/emqx_gateway/src/emqx_gateway.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_gateway, [
     {description, "The Gateway management application"},
-    {vsn, "0.1.9"},
+    {vsn, "0.1.10"},
     {registered, []},
     {mod, {emqx_gateway_app, []}},
     {applications, [kernel, stdlib, grpc, emqx, emqx_authn]},

apps/emqx_gateway/src/mqttsn/emqx_sn_broadcast.erl

@@ -18,7 +18,13 @@
 
 -behaviour(gen_server).
 
+-ifdef(TEST).
+%% make rebar3 ct happy when testing with --suite path/to/module_SUITE.erl
+-include_lib("emqx_gateway/src/mqttsn/include/emqx_sn.hrl").
+-else.
+%% make mix happy
 -include("src/mqttsn/include/emqx_sn.hrl").
+-endif.
 -include_lib("emqx/include/logger.hrl").
 
 -export([

apps/emqx_machine/src/emqx_machine.app.src

@@ -3,7 +3,7 @@
     {id, "emqx_machine"},
     {description, "The EMQX Machine"},
     % strict semver, bump manually!
-    {vsn, "0.1.1"},
+    {vsn, "0.1.2"},
     {modules, []},
     {registered, []},
     {applications, [kernel, stdlib]},

apps/emqx_statsd/include/emqx_statsd.hrl

@@ -1,2 +1,19 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2021-2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
 -define(APP, emqx_statsd).
 -define(STATSD, [statsd]).
+-define(SERVER_PARSE_OPTS, #{default_port => 8125}).

apps/emqx_statsd/src/emqx_statsd.app.src

@@ -1,7 +1,7 @@
 %% -*- mode: erlang -*-
 {application, emqx_statsd, [
     {description, "EMQX Statsd"},
-    {vsn, "5.0.3"},
+    {vsn, "5.0.4"},
     {registered, []},
     {mod, {emqx_statsd_app, []}},
     {applications, [

apps/emqx_statsd/src/emqx_statsd.erl

@@ -75,10 +75,11 @@ init([]) ->
     process_flag(trap_exit, true),
     #{
         tags := TagsRaw,
-        server := {Host, Port},
+        server := Server,
         sample_time_interval := SampleTimeInterval,
         flush_time_interval := FlushTimeInterval
     } = emqx_conf:get([statsd]),
+    {Host, Port} = emqx_schema:parse_server(Server, ?SERVER_PARSE_OPTS),
     Tags = maps:fold(fun(K, V, Acc) -> [{to_bin(K), to_bin(V)} | Acc] end, [], TagsRaw),
     Opts = [{tags, Tags}, {host, Host}, {port, Port}, {prefix, <<"emqx">>}],
     {ok, Pid} = estatsd:start_link(Opts),

apps/emqx_statsd/src/emqx_statsd_schema.erl

@@ -18,6 +18,7 @@
 
 -include_lib("hocon/include/hoconsc.hrl").
 -include_lib("typerefl/include/types.hrl").
+-include("emqx_statsd.hrl").
 
 -behaviour(hocon_schema).
 
@@ -44,7 +45,7 @@ fields("statsd") ->
                     desc => ?DESC(enable)
                 }
             )},
-        {server, fun server/1},
+        {server, server()},
         {sample_time_interval, fun sample_interval/1},
         {flush_time_interval, fun flush_interval/1},
         {tags, fun tags/1}
@@ -53,11 +54,12 @@ fields("statsd") ->
 desc("statsd") -> ?DESC(statsd);
 desc(_) -> undefined.
 
-server(type) -> emqx_schema:host_port();
+server() ->
-server(required) -> true;
+    Meta = #{
-server(default) -> "127.0.0.1:8125";
+        required => true,
-server(desc) -> ?DESC(?FUNCTION_NAME);
+        desc => ?DESC(?FUNCTION_NAME)
-server(_) -> undefined.
+    },
+    emqx_schema:servers_sc(Meta, ?SERVER_PARSE_OPTS).
 
 sample_interval(type) -> emqx_schema:duration_ms();
 sample_interval(required) -> true;

changes/v5.0.14-en.md

@@ -2,6 +2,10 @@
 
 ## Enhancements
 
+- Make possible to configure `host:port` from environment variables without quotes [#9614](https://github.com/emqx/emqx/pull/9614).
+  Prior to this change, when overriding a `host:port` config value from environment variable, one has to quote it as:
+  `env EMQX_BRIDGES__MQTT__XYZ__SERVER='"localhost:1883"'`.
+  Now it's possible to set it without quote as `env EMQX_BRIDGES__MQTT__XYZ__SERVER='localhost:1883'`.
 
 ## Bug Fixes
 

changes/v5.0.14-zh.md

@@ -2,6 +2,10 @@
 
 ## 增强
 
+- 允许环境变量重载 `host:port` 值时不使用引号 [#9614](https://github.com/emqx/emqx/pull/9614)。
+  在此修复前，环境变量中使用 `host:port` 这种配置时，用户必须使用引号，例如：
+  `env EMQX_BRIDGES__MQTT__XYZ__SERVER='"localhost:1883"'`。
+  此修复后，可以不使用引号，例如 `env EMQX_BRIDGES__MQTT__XYZ__SERVER='localhost:1883'`。
 
 ## 修复
 

lib-ee/emqx_ee_bridge/i18n/emqx_ee_bridge_kafka.conf

@@ -81,8 +81,8 @@ emqx_ee_bridge_kafka {
     }
     bootstrap_hosts {
         desc {
-            en: "A comma separated list of Kafka <code>host:port</code> endpoints to bootstrap the client."
+            en: "A comma separated list of Kafka <code>host[:port]</code> endpoints to bootstrap the client. Default port number is 9092."
-            zh: "用逗号分隔的 <code>host:port</code> 主机列表。"
+            zh: "用逗号分隔的 <code>host[:port]</code> 主机列表。默认端口号为 9092。"
         }
         label {
             en: "Bootstrap Hosts"

lib-ee/emqx_ee_bridge/src/emqx_ee_bridge_kafka.erl

@@ -26,7 +26,8 @@
     namespace/0,
     roots/0,
     fields/1,
-    desc/1
+    desc/1,
+    host_opts/0
 ]).
 
 %% -------------------------------------------------------------------------------------------------
@@ -54,6 +55,9 @@ values(put) ->
 %% -------------------------------------------------------------------------------------------------
 %% Hocon Schema Definitions
 
+host_opts() ->
+    #{default_port => 9092}.
+
 namespace() -> "bridge_kafka".
 
 roots() -> ["config"].
@@ -67,7 +71,17 @@ fields("get") ->
 fields("config") ->
     [
         {enable, mk(boolean(), #{desc => ?DESC("config_enable"), default => true})},
-        {bootstrap_hosts, mk(binary(), #{required => true, desc => ?DESC(bootstrap_hosts)})},
+        {bootstrap_hosts,
+            mk(
+                binary(),
+                #{
+                    required => true,
+                    desc => ?DESC(bootstrap_hosts),
+                    validator => emqx_schema:servers_validator(
+                        host_opts(), _Required = true
+                    )
+                }
+            )},
         {connect_timeout,
             mk(emqx_schema:duration_ms(), #{
                 default => "5s",

lib-ee/emqx_ee_bridge/src/kafka/emqx_bridge_impl_kafka_producer.erl

@@ -177,10 +177,8 @@ on_get_status(_InstId, _State) ->
     connected.
 
 %% Parse comma separated host:port list into a [{Host,Port}] list
-hosts(Hosts) when is_binary(Hosts) ->
+hosts(Hosts) ->
-    hosts(binary_to_list(Hosts));
+    emqx_schema:parse_servers(Hosts, emqx_ee_bridge_kafka:host_opts()).
-hosts(Hosts) when is_list(Hosts) ->
-    kpro:parse_endpoints(Hosts).
 
 %% Extra socket options, such as sndbuf size etc.
 socket_opts(Opts) when is_map(Opts) ->

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_gcp_pubsub_SUITE.erl

@@ -777,15 +777,13 @@ t_publish_success_batch(Config) ->
 
 t_not_a_json(Config) ->
     ?assertMatch(
-        {error,
+        {error, #{
-            {_, [
+            discarded_errors_count := 0,
-                #{
+            kind := validation_error,
-                    kind := validation_error,
+            reason := #{exception := {error, {badmap, "not a json"}}},
-                    reason := #{exception := {error, {badmap, "not a json"}}},
+            %% should be censored as it contains secrets
-                    %% should be censored as it contains secrets
+            value := <<"******">>
-                    value := <<"******">>
+        }},
-                }
-            ]}},
         create_bridge(
             Config,
             #{
@@ -797,15 +795,13 @@ t_not_a_json(Config) ->
 
 t_not_of_service_account_type(Config) ->
     ?assertMatch(
-        {error,
+        {error, #{
-            {_, [
+            discarded_errors_count := 0,
-                #{
+            kind := validation_error,
-                    kind := validation_error,
+            reason := {wrong_type, <<"not a service account">>},
-                    reason := {wrong_type, <<"not a service account">>},
+            %% should be censored as it contains secrets
-                    %% should be censored as it contains secrets
+            value := <<"******">>
-                    value := <<"******">>
+        }},
-                }
-            ]}},
         create_bridge(
             Config,
             #{
@@ -818,22 +814,20 @@ t_not_of_service_account_type(Config) ->
 t_json_missing_fields(Config) ->
     GCPPubSubConfig0 = ?config(gcp_pubsub_config, Config),
     ?assertMatch(
-        {error,
+        {error, #{
-            {_, [
+            discarded_errors_count := 0,
-                #{
+            kind := validation_error,
-                    kind := validation_error,
+            reason :=
-                    reason :=
+                {missing_keys, [
-                        {missing_keys, [
+                    <<"client_email">>,
-                            <<"client_email">>,
+                    <<"private_key">>,
-                            <<"private_key">>,
+                    <<"private_key_id">>,
-                            <<"private_key_id">>,
+                    <<"project_id">>,
-                            <<"project_id">>,
+                    <<"type">>
-                            <<"type">>
+                ]},
-                        ]},
+            %% should be censored as it contains secrets
-                    %% should be censored as it contains secrets
+            value := <<"******">>
-                    value := <<"******">>
+        }},
-                }
-            ]}},
         create_bridge([
             {gcp_pubsub_config, GCPPubSubConfig0#{<<"service_account_json">> := #{}}}
             | Config

lib-ee/emqx_ee_bridge/test/emqx_ee_bridge_redis_SUITE.erl

@@ -374,21 +374,15 @@ all_test_hosts() ->
     lists:flatmap(
         fun
             (#{<<"servers">> := ServersRaw}) ->
-                lists:map(
+                parse_servers(ServersRaw);
-                    fun(Server) ->
-                        parse_server(Server)
-                    end,
-                    string:tokens(binary_to_list(ServersRaw), ", ")
-                );
             (#{<<"server">> := ServerRaw}) ->
-                [parse_server(ServerRaw)]
+                parse_servers(ServerRaw)
         end,
         Confs
     ).
 
-parse_server(Server) ->
+parse_servers(Servers) ->
-    emqx_connector_schema_lib:parse_server(Server, #{
+    emqx_schema:parse_servers(Servers, #{
-        host_type => hostname,
         default_port => 6379
     }).
 

lib-ee/emqx_ee_connector/src/emqx_ee_connector_gcp_pubsub.erl

@@ -83,9 +83,7 @@ on_start(
     %% emulating the emulator behavior
     %% https://cloud.google.com/pubsub/docs/emulator
     HostPort = os:getenv("PUBSUB_EMULATOR_HOST", "pubsub.googleapis.com:443"),
-    {Host, Port} = emqx_connector_schema_lib:parse_server(
+    {Host, Port} = emqx_schema:parse_server(HostPort, #{default_port => 443}),
-        HostPort, #{host_type => hostname, default_port => 443}
-    ),
     PoolType = random,
     Transport = tls,
     TransportOpts = emqx_tls_lib:to_client_opts(#{enable => true, verify => verify_none}),

lib-ee/emqx_ee_connector/src/emqx_ee_connector_influxdb.erl

@@ -35,7 +35,6 @@
 
 %% influxdb servers don't need parse
 -define(INFLUXDB_HOST_OPTIONS, #{
-    host_type => hostname,
     default_port => ?INFLUXDB_DEFAULT_PORT
 }).
 
@@ -141,7 +140,7 @@ namespace() -> connector_influxdb.
 
 fields(common) ->
     [
-        {server, fun server/1},
+        {server, server()},
         {precision,
             mk(enum([ns, us, ms, s, m, h]), #{
                 required => false, default => ms, desc => ?DESC("precision")
@@ -164,13 +163,14 @@ fields(influxdb_api_v2) ->
             {token, mk(binary(), #{required => true, desc => ?DESC("token")})}
         ] ++ emqx_connector_schema_lib:ssl_fields().
 
-server(type) -> emqx_schema:ip_port();
+server() ->
-server(required) -> true;
+    Meta = #{
-server(validator) -> [?NOT_EMPTY("the value of the field 'server' cannot be empty")];
+        required => false,
-server(converter) -> fun to_server_raw/1;
+        default => <<"127.0.0.1:8086">>,
-server(default) -> <<"127.0.0.1:8086">>;
+        desc => ?DESC("server"),
-server(desc) -> ?DESC("server");
+        converter => fun convert_server/2
-server(_) -> undefined.
+    },
+    emqx_schema:servers_sc(Meta, ?INFLUXDB_HOST_OPTIONS).
 
 desc(common) ->
     ?DESC("common");
@@ -261,9 +261,10 @@ do_start_client(
 client_config(
     InstId,
     Config = #{
-        server := {Host, Port}
+        server := Server
     }
 ) ->
+    {Host, Port} = emqx_schema:parse_server(Server, ?INFLUXDB_HOST_OPTIONS),
     [
         {host, str(Host)},
         {port, Port},
@@ -542,17 +543,12 @@ log_error_points(InstId, Errs) ->
         Errs
     ).
 
-%% ===================================================================
+convert_server(<<"http://", Server/binary>>, HoconOpts) ->
-%% typereflt funcs
+    convert_server(Server, HoconOpts);
-
+convert_server(<<"https://", Server/binary>>, HoconOpts) ->
--spec to_server_raw(string() | binary()) ->
+    convert_server(Server, HoconOpts);
-    {string(), pos_integer()}.
+convert_server(Server, HoconOpts) ->
-to_server_raw(<<"http://", Server/binary>>) ->
+    emqx_schema:convert_servers(Server, HoconOpts).
-    emqx_connector_schema_lib:parse_server(Server, ?INFLUXDB_HOST_OPTIONS);
-to_server_raw(<<"https://", Server/binary>>) ->
-    emqx_connector_schema_lib:parse_server(Server, ?INFLUXDB_HOST_OPTIONS);
-to_server_raw(Server) ->
-    emqx_connector_schema_lib:parse_server(Server, ?INFLUXDB_HOST_OPTIONS).
 
 str(A) when is_atom(A) ->
     atom_to_list(A);
@@ -568,22 +564,6 @@ str(S) when is_list(S) ->
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 
-to_server_raw_test_() ->
-    [
-        ?_assertEqual(
-            {"foobar", 1234},
-            to_server_raw(<<"http://foobar:1234">>)
-        ),
-        ?_assertEqual(
-            {"foobar", 1234},
-            to_server_raw(<<"https://foobar:1234">>)
-        ),
-        ?_assertEqual(
-            {"foobar", 1234},
-            to_server_raw(<<"foobar:1234">>)
-        )
-    ].
-
 %% for coverage
 desc_test_() ->
     [
@@ -605,7 +585,7 @@ desc_test_() ->
         ),
         ?_assertMatch(
             {desc, _, _},
-            server(desc)
+            hocon_schema:field_schema(server(), desc)
         ),
         ?_assertMatch(
             connector_influxdb,

scripts/apps-version-check.sh

@@ -1,7 +1,8 @@
 #!/usr/bin/env bash
 set -euo pipefail
 
-latest_release=$(git describe --abbrev=0 --tags --exclude '*rc*' --exclude '*alpha*' --exclude '*beta*')
+latest_release=$(git describe --abbrev=0 --tags --exclude '*rc*' --exclude '*alpha*' --exclude '*beta*' --exclude '*docker*')
+echo "Compare base: $latest_release"
 
 bad_app_count=0
 

scripts/git-hook-pre-commit.sh

@@ -15,3 +15,4 @@ if ! (./scripts/erlfmt $OPT $files); then
     echo "EXECUTE 'make fmt' to fix" >&2
     exit 1
 fi
+./scripts/apps-version-check.sh