Merge pull request #12053 from zmstone/1129-prepare-for-otp26

1129 prepare for OTP 26
This commit is contained in:
Zaiming (Stone) Shi 2023-11-30 19:38:40 +01:00 committed by GitHub
commit dcb1c0680b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
36 changed files with 314 additions and 235 deletions

View File

@ -1,2 +1,2 @@
erlang 25.3.2-2 erlang 26.1.2-1
elixir 1.14.5-otp-25 elixir 1.15.7-otp-26

View File

@ -40,7 +40,7 @@ init_per_testcase(TestCase, Config) when
Cluster = cluster(#{n => 1}), Cluster = cluster(#{n => 1}),
ClusterOpts = #{work_dir => emqx_cth_suite:work_dir(TestCase, Config)}, ClusterOpts = #{work_dir => emqx_cth_suite:work_dir(TestCase, Config)},
NodeSpecs = emqx_cth_cluster:mk_nodespecs(Cluster, ClusterOpts), NodeSpecs = emqx_cth_cluster:mk_nodespecs(Cluster, ClusterOpts),
Nodes = emqx_cth_cluster:start(Cluster, ClusterOpts), Nodes = emqx_cth_cluster:start(NodeSpecs),
[ [
{cluster, Cluster}, {cluster, Cluster},
{node_specs, NodeSpecs}, {node_specs, NodeSpecs},
@ -116,27 +116,8 @@ start_client(Opts0 = #{}) ->
restart_node(Node, NodeSpec) -> restart_node(Node, NodeSpec) ->
?tp(will_restart_node, #{}), ?tp(will_restart_node, #{}),
?tp(notice, "restarting node", #{node => Node}), emqx_cth_cluster:restart(Node, NodeSpec),
true = monitor_node(Node, true),
ok = erpc:call(Node, init, restart, []),
receive
{nodedown, Node} ->
ok
after 10_000 ->
ct:fail("node ~p didn't stop", [Node])
end,
?tp(notice, "waiting for nodeup", #{node => Node}),
wait_nodeup(Node), wait_nodeup(Node),
wait_gen_rpc_down(NodeSpec),
?tp(notice, "restarting apps", #{node => Node}),
Apps = maps:get(apps, NodeSpec),
ok = erpc:call(Node, emqx_cth_suite, load_apps, [Apps]),
_ = erpc:call(Node, emqx_cth_suite, start_apps, [Apps, NodeSpec]),
%% have to re-inject this so that we may stop the node succesfully at the
%% end....
ok = emqx_cth_cluster:set_node_opts(Node, NodeSpec),
ok = snabbkaffe:forward_trace(Node),
?tp(notice, "node restarted", #{node => Node}),
?tp(restarted_node, #{}), ?tp(restarted_node, #{}),
ok. ok.

View File

@ -27,9 +27,9 @@
{lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.2"}}}, {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.2"}}},
{gproc, {git, "https://github.com/emqx/gproc", {tag, "0.9.0.1"}}}, {gproc, {git, "https://github.com/emqx/gproc", {tag, "0.9.0.1"}}},
{cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}}, {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}},
{esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.7"}}}, {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.8"}}},
{ekka, {git, "https://github.com/emqx/ekka", {tag, "0.15.16"}}}, {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.15.16"}}},
{gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.2.1"}}}, {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.2.2"}}},
{hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.40.0"}}}, {hocon, {git, "https://github.com/emqx/hocon.git", {tag, "0.40.0"}}},
{emqx_http_lib, {git, "https://github.com/emqx/emqx_http_lib.git", {tag, "0.5.3"}}}, {emqx_http_lib, {git, "https://github.com/emqx/emqx_http_lib.git", {tag, "0.5.3"}}},
{pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}}, {pbkdf2, {git, "https://github.com/emqx/erlang-pbkdf2.git", {tag, "2.0.4"}}},

View File

@ -418,6 +418,9 @@ get_otp_version() ->
end. end.
read_otp_version() -> read_otp_version() ->
string:trim(do_read_otp_version()).
do_read_otp_version() ->
ReleasesDir = filename:join([code:root_dir(), "releases"]), ReleasesDir = filename:join([code:root_dir(), "releases"]),
Filename = filename:join([ReleasesDir, emqx_app:get_release(), "BUILD_INFO"]), Filename = filename:join([ReleasesDir, emqx_app:get_release(), "BUILD_INFO"]),
case file:read_file(Filename) of case file:read_file(Filename) of

View File

@ -753,24 +753,15 @@ start_slave(Name, Opts) when is_map(Opts) ->
case SlaveMod of case SlaveMod of
ct_slave -> ct_slave ->
ct:pal("~p: node data dir: ~s", [Node, NodeDataDir]), ct:pal("~p: node data dir: ~s", [Node, NodeDataDir]),
ct_slave:start( Envs = [
Node, {"HOCON_ENV_OVERRIDE_PREFIX", "EMQX_"},
[ {"EMQX_NODE__COOKIE", Cookie},
{kill_if_fail, true}, {"EMQX_NODE__DATA_DIR", NodeDataDir}
{monitor_master, true}, ],
{init_timeout, 20_000}, emqx_cth_peer:start(Node, erl_flags(), Envs);
{startup_timeout, 20_000},
{erl_flags, erl_flags()},
{env, [
{"HOCON_ENV_OVERRIDE_PREFIX", "EMQX_"},
{"EMQX_NODE__COOKIE", Cookie},
{"EMQX_NODE__DATA_DIR", NodeDataDir}
]}
]
);
slave -> slave ->
Env = " -env HOCON_ENV_OVERRIDE_PREFIX EMQX_", Envs = [{"HOCON_ENV_OVERRIDE_PREFIX", "EMQX_"}],
slave:start_link(host(), Name, ebin_path() ++ Env) emqx_cth_peer:start(Node, ebin_path(), Envs)
end end
end, end,
case DoStart() of case DoStart() of
@ -789,13 +780,7 @@ start_slave(Name, Opts) when is_map(Opts) ->
%% Node stopping %% Node stopping
stop_slave(Node0) -> stop_slave(Node0) ->
Node = node_name(Node0), Node = node_name(Node0),
SlaveMod = get_peer_mod(Node), emqx_cth_peer:stop(Node).
erase_peer_mod(Node),
case SlaveMod:stop(Node) of
ok -> ok;
{ok, _} -> ok;
{error, not_started, _} -> ok
end.
%% EPMD starting %% EPMD starting
start_epmd() -> start_epmd() ->
@ -1022,11 +1007,11 @@ set_envs(Node, Env) ->
). ).
erl_flags() -> erl_flags() ->
%% One core and redirecting logs to master %% One core
"+S 1:1 -master " ++ atom_to_list(node()) ++ " " ++ ebin_path(). ["+S", "1:1"] ++ ebin_path().
ebin_path() -> ebin_path() ->
string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " "). ["-pa" | lists:filter(fun is_lib/1, code:get_path())].
is_lib(Path) -> is_lib(Path) ->
string:prefix(Path, code:lib_dir()) =:= nomatch andalso string:prefix(Path, code:lib_dir()) =:= nomatch andalso

View File

@ -38,14 +38,14 @@
%% in `end_per_suite/1` or `end_per_group/2`) with the result from step 2. %% in `end_per_suite/1` or `end_per_group/2`) with the result from step 2.
-module(emqx_cth_cluster). -module(emqx_cth_cluster).
-export([start/2]). -export([start/1, start/2, restart/2]).
-export([stop/1, stop_node/1]). -export([stop/1, stop_node/1]).
-export([start_bare_node/2]). -export([start_bare_nodes/1, start_bare_nodes/2]).
-export([share_load_module/2]). -export([share_load_module/2]).
-export([node_name/1, mk_nodespecs/2]). -export([node_name/1, mk_nodespecs/2]).
-export([start_apps/2, set_node_opts/2]). -export([start_apps/2]).
-define(APPS_CLUSTERING, [gen_rpc, mria, ekka]). -define(APPS_CLUSTERING, [gen_rpc, mria, ekka]).
@ -109,9 +109,12 @@ when
}. }.
start(Nodes, ClusterOpts) -> start(Nodes, ClusterOpts) ->
NodeSpecs = mk_nodespecs(Nodes, ClusterOpts), NodeSpecs = mk_nodespecs(Nodes, ClusterOpts),
ct:pal("Starting cluster:\n ~p", [NodeSpecs]), start(NodeSpecs).
start(NodeSpecs) ->
ct:pal("(Re)starting nodes:\n ~p", [NodeSpecs]),
% 1. Start bare nodes with only basic applications running % 1. Start bare nodes with only basic applications running
_ = emqx_utils:pmap(fun start_node_init/1, NodeSpecs, ?TIMEOUT_NODE_START_MS), ok = start_nodes_init(NodeSpecs, ?TIMEOUT_NODE_START_MS),
% 2. Start applications needed to enable clustering % 2. Start applications needed to enable clustering
% Generally, this causes some applications to restart, but we deliberately don't % Generally, this causes some applications to restart, but we deliberately don't
% start them yet. % start them yet.
@ -121,6 +124,11 @@ start(Nodes, ClusterOpts) ->
_ = emqx_utils:pmap(fun run_node_phase_apps/1, NodeSpecs, ?TIMEOUT_APPS_START_MS), _ = emqx_utils:pmap(fun run_node_phase_apps/1, NodeSpecs, ?TIMEOUT_APPS_START_MS),
[Node || #{name := Node} <- NodeSpecs]. [Node || #{name := Node} <- NodeSpecs].
restart(Node, Spec) ->
ct:pal("Stopping peer node ~p", [Node]),
ok = emqx_cth_peer:stop(Node),
start([Spec#{boot_type => restart}]).
mk_nodespecs(Nodes, ClusterOpts) -> mk_nodespecs(Nodes, ClusterOpts) ->
NodeSpecs = lists:zipwith( NodeSpecs = lists:zipwith(
fun(N, {Name, Opts}) -> mk_init_nodespec(N, Name, Opts, ClusterOpts) end, fun(N, {Name, Opts}) -> mk_init_nodespec(N, Name, Opts, ClusterOpts) end,
@ -282,8 +290,50 @@ allocate_listener_port(Type, #{base_port := BasePort}) ->
allocate_listener_ports(Types, Spec) -> allocate_listener_ports(Types, Spec) ->
lists:foldl(fun maps:merge/2, #{}, [allocate_listener_port(Type, Spec) || Type <- Types]). lists:foldl(fun maps:merge/2, #{}, [allocate_listener_port(Type, Spec) || Type <- Types]).
start_node_init(Spec = #{name := Node}) -> start_nodes_init(Specs, Timeout) ->
Node = start_bare_node(Node, Spec), Names = lists:map(fun(#{name := Name}) -> Name end, Specs),
Nodes = start_bare_nodes(Names, Timeout),
lists:foreach(fun node_init/1, Nodes).
start_bare_nodes(Names) ->
start_bare_nodes(Names, ?TIMEOUT_NODE_START_MS).
start_bare_nodes(Names, Timeout) ->
Args = erl_flags(),
Envs = [],
Waits = lists:map(
fun(Name) ->
WaitTag = {boot_complete, Name},
WaitBoot = {self(), WaitTag},
{ok, _} = emqx_cth_peer:start(Name, Args, Envs, WaitBoot),
WaitTag
end,
Names
),
Deadline = erlang:monotonic_time() + erlang:convert_time_unit(Timeout, millisecond, nanosecond),
Nodes = wait_boot_complete(Waits, Deadline),
lists:foreach(fun(Node) -> pong = net_adm:ping(Node) end, Nodes),
Nodes.
wait_boot_complete([], _) ->
[];
wait_boot_complete(Waits, Deadline) ->
case erlang:monotonic_time() > Deadline of
true ->
error({timeout, Waits});
false ->
ok
end,
receive
{{boot_complete, _Name} = Wait, {started, Node, _Pid}} ->
ct:pal("~p", [Wait]),
[Node | wait_boot_complete(Waits -- [Wait], Deadline)];
{{boot_complete, _Name}, Otherwise} ->
error({unexpected, Otherwise})
after 100 ->
wait_boot_complete(Waits, Deadline)
end.
node_init(Node) ->
% Make it possible to call `ct:pal` and friends (if running under rebar3) % Make it possible to call `ct:pal` and friends (if running under rebar3)
_ = share_load_module(Node, cthr), _ = share_load_module(Node, cthr),
% Enable snabbkaffe trace forwarding % Enable snabbkaffe trace forwarding
@ -300,12 +350,6 @@ run_node_phase_apps(Spec = #{name := Node}) ->
ok = start_apps(Node, Spec), ok = start_apps(Node, Spec),
ok. ok.
set_node_opts(Node, Spec) ->
erpc:call(Node, persistent_term, put, [{?MODULE, opts}, Spec]).
get_node_opts(Node) ->
erpc:call(Node, persistent_term, get, [{?MODULE, opts}]).
load_apps(Node, #{apps := Apps}) -> load_apps(Node, #{apps := Apps}) ->
erpc:call(Node, emqx_cth_suite, load_apps, [Apps]). erpc:call(Node, emqx_cth_suite, load_apps, [Apps]).
@ -322,8 +366,12 @@ start_apps(Node, #{apps := Apps} = Spec) ->
ok. ok.
suite_opts(Spec) -> suite_opts(Spec) ->
maps:with([work_dir], Spec). maps:with([work_dir, boot_type], Spec).
maybe_join_cluster(_Node, #{boot_type := restart}) ->
%% when restart, the node should already be in the cluster
%% hence no need to (re)join
ok;
maybe_join_cluster(_Node, #{role := replicant}) -> maybe_join_cluster(_Node, #{role := replicant}) ->
ok; ok;
maybe_join_cluster(Node, Spec) -> maybe_join_cluster(Node, Spec) ->
@ -352,23 +400,7 @@ stop(Nodes) ->
stop_node(Name) -> stop_node(Name) ->
Node = node_name(Name), Node = node_name(Name),
try get_node_opts(Node) of ok = emqx_cth_peer:stop(Node).
Opts ->
stop_node(Name, Opts)
catch
error:{erpc, _} ->
ok
end.
stop_node(Node, #{driver := ct_slave}) ->
case ct_slave:stop(Node, [{stop_timeout, ?TIMEOUT_NODE_STOP_S}]) of
{ok, _} ->
ok;
{error, Reason, _} when Reason == not_connected; Reason == not_started ->
ok
end;
stop_node(Node, #{driver := slave}) ->
slave:stop(Node).
%% Ports %% Ports
@ -391,36 +423,12 @@ listener_port(BasePort, wss) ->
%% %%
-spec start_bare_node(atom(), map()) -> node().
start_bare_node(Name, Spec = #{driver := ct_slave}) ->
{ok, Node} = ct_slave:start(
node_name(Name),
[
{kill_if_fail, true},
{monitor_master, true},
{init_timeout, 20_000},
{startup_timeout, 20_000},
{erl_flags, erl_flags()},
{env, []}
]
),
init_bare_node(Node, Spec);
start_bare_node(Name, Spec = #{driver := slave}) ->
{ok, Node} = slave:start_link(host(), Name, ebin_path()),
init_bare_node(Node, Spec).
init_bare_node(Node, Spec) ->
pong = net_adm:ping(Node),
% Preserve node spec right on the remote node
ok = set_node_opts(Node, Spec),
Node.
erl_flags() -> erl_flags() ->
%% One core and redirecting logs to master %% One core
"+S 1:1 -master " ++ atom_to_list(node()) ++ " " ++ ebin_path(). ["+S", "1:1"] ++ ebin_path().
ebin_path() -> ebin_path() ->
string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " "). ["-pa" | lists:filter(fun is_lib/1, code:get_path())].
is_lib(Path) -> is_lib(Path) ->
string:prefix(Path, code:lib_dir()) =:= nomatch andalso string:prefix(Path, code:lib_dir()) =:= nomatch andalso

View File

@ -0,0 +1,79 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2023 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.
%%--------------------------------------------------------------------
%% @doc Common Test Helper proxy module for slave -> peer migration.
%% OTP 26 has slave module deprecated, use peer instead.
-module(emqx_cth_peer).
-export([start/2, start/3, start/4]).
-export([start_link/2, start_link/3, start_link/4]).
-export([stop/1]).
start(Name, Args) ->
start(Name, Args, []).
start(Name, Args, Envs) ->
start(Name, Args, Envs, timer:seconds(20)).
start(Name, Args, Envs, Timeout) when is_atom(Name) ->
do_start(Name, Args, Envs, Timeout, start).
start_link(Name, Args) ->
start_link(Name, Args, []).
start_link(Name, Args, Envs) ->
start_link(Name, Args, Envs, timer:seconds(20)).
start_link(Name, Args, Envs, Timeout) when is_atom(Name) ->
do_start(Name, Args, Envs, Timeout, start_link).
do_start(Name0, Args, Envs, Timeout, Func) when is_atom(Name0) ->
{Name, Host} = parse_node_name(Name0),
{ok, Pid, Node} = peer:Func(#{
name => Name,
host => Host,
args => Args,
env => Envs,
wait_boot => Timeout,
longnames => true,
shutdown => {halt, 1000}
}),
true = register(Node, Pid),
{ok, Node}.
stop(Node) when is_atom(Node) ->
Pid = whereis(Node),
case is_pid(Pid) of
true ->
unlink(Pid),
ok = peer:stop(Pid);
false ->
ct:pal("The control process for node ~p is unexpetedly down", [Node]),
ok
end.
parse_node_name(NodeName) ->
case string:tokens(atom_to_list(NodeName), "@") of
[Name, Host] ->
{list_to_atom(Name), Host};
[_] ->
{NodeName, host()}
end.
host() ->
[_Name, Host] = string:tokens(atom_to_list(node()), "@"),
Host.

View File

@ -453,6 +453,9 @@ stop_apps(Apps) ->
%% %%
verify_clean_suite_state(#{boot_type := restart}) ->
%% when testing node restart, we do not need to verify clean state
ok;
verify_clean_suite_state(#{work_dir := WorkDir}) -> verify_clean_suite_state(#{work_dir := WorkDir}) ->
{ok, []} = file:list_dir(WorkDir), {ok, []} = file:list_dir(WorkDir),
false = emqx_schema_hooks:any_injections(), false = emqx_schema_hooks:any_injections(),

View File

@ -53,9 +53,9 @@ t_get_metrics(_) ->
?assertMatch( ?assertMatch(
#{ #{
rate := #{ rate := #{
a := #{current := 0.0, max := 0.0, last5m := 0.0}, a := #{current := +0.0, max := +0.0, last5m := +0.0},
b := #{current := 0.0, max := 0.0, last5m := 0.0}, b := #{current := +0.0, max := +0.0, last5m := +0.0},
c := #{current := 0.0, max := 0.0, last5m := 0.0} c := #{current := +0.0, max := +0.0, last5m := +0.0}
}, },
gauges := #{}, gauges := #{},
counters := #{ counters := #{
@ -118,9 +118,9 @@ t_clear_metrics(_Config) ->
?assertMatch( ?assertMatch(
#{ #{
rate := #{ rate := #{
a := #{current := 0.0, max := 0.0, last5m := 0.0}, a := #{current := +0.0, max := +0.0, last5m := +0.0},
b := #{current := 0.0, max := 0.0, last5m := 0.0}, b := #{current := +0.0, max := +0.0, last5m := +0.0},
c := #{current := 0.0, max := 0.0, last5m := 0.0} c := #{current := +0.0, max := +0.0, last5m := +0.0}
}, },
gauges := #{}, gauges := #{},
slides := #{}, slides := #{},
@ -145,7 +145,7 @@ t_clear_metrics(_Config) ->
#{ #{
counters => #{}, counters => #{},
gauges => #{}, gauges => #{},
rate => #{current => 0.0, last5m => 0.0, max => 0.0}, rate => #{current => +0.0, last5m => +0.0, max => +0.0},
slides => #{} slides => #{}
}, },
emqx_metrics_worker:get_metrics(?NAME, Id) emqx_metrics_worker:get_metrics(?NAME, Id)
@ -160,9 +160,9 @@ t_reset_metrics(_) ->
?assertMatch( ?assertMatch(
#{ #{
rate := #{ rate := #{
a := #{current := 0.0, max := 0.0, last5m := 0.0}, a := #{current := +0.0, max := +0.0, last5m := +0.0},
b := #{current := 0.0, max := 0.0, last5m := 0.0}, b := #{current := +0.0, max := +0.0, last5m := +0.0},
c := #{current := 0.0, max := 0.0, last5m := 0.0} c := #{current := +0.0, max := +0.0, last5m := +0.0}
}, },
gauges := #{}, gauges := #{},
counters := #{ counters := #{

View File

@ -58,9 +58,6 @@ t_mount_share(_) ->
TopicFilters = [T], TopicFilters = [T],
?assertEqual(TopicFilter, #share{group = <<"group">>, topic = <<"topic">>}), ?assertEqual(TopicFilter, #share{group = <<"group">>, topic = <<"topic">>}),
%% should not mount share topic when make message.
Msg = emqx_message:make(<<"clientid">>, TopicFilter, <<"payload">>),
?assertEqual( ?assertEqual(
TopicFilter, TopicFilter,
mount(undefined, TopicFilter) mount(undefined, TopicFilter)
@ -89,8 +86,6 @@ t_unmount_share(_) ->
?assertEqual(TopicFilter, #share{group = <<"group">>, topic = <<"topic">>}), ?assertEqual(TopicFilter, #share{group = <<"group">>, topic = <<"topic">>}),
%% should not unmount share topic when make message.
Msg = emqx_message:make(<<"clientid">>, TopicFilter, <<"payload">>),
?assertEqual( ?assertEqual(
TopicFilter, TopicFilter,
unmount(undefined, TopicFilter) unmount(undefined, TopicFilter)

View File

@ -80,7 +80,7 @@ t_mnesia(_) ->
ct:sleep(200). ct:sleep(200).
t_cleanup_membership_mnesia_down(_Config) -> t_cleanup_membership_mnesia_down(_Config) ->
Slave = emqx_cth_cluster:node_name(?FUNCTION_NAME), Slave = emqx_cth_cluster:node_name(node2),
emqx_router:add_route(<<"a/b/c">>, Slave), emqx_router:add_route(<<"a/b/c">>, Slave),
emqx_router:add_route(<<"d/e/f">>, node()), emqx_router:add_route(<<"d/e/f">>, node()),
?assertMatch([_, _], emqx_router:topics()), ?assertMatch([_, _], emqx_router:topics()),
@ -92,7 +92,7 @@ t_cleanup_membership_mnesia_down(_Config) ->
?assertEqual([<<"d/e/f">>], emqx_router:topics()). ?assertEqual([<<"d/e/f">>], emqx_router:topics()).
t_cleanup_membership_node_down(_Config) -> t_cleanup_membership_node_down(_Config) ->
Slave = emqx_cth_cluster:node_name(?FUNCTION_NAME), Slave = emqx_cth_cluster:node_name(node3),
emqx_router:add_route(<<"a/b/c">>, Slave), emqx_router:add_route(<<"a/b/c">>, Slave),
emqx_router:add_route(<<"d/e/f">>, node()), emqx_router:add_route(<<"d/e/f">>, node()),
?assertMatch([_, _], emqx_router:topics()), ?assertMatch([_, _], emqx_router:topics()),
@ -104,7 +104,7 @@ t_cleanup_membership_node_down(_Config) ->
?assertEqual([<<"d/e/f">>], emqx_router:topics()). ?assertEqual([<<"d/e/f">>], emqx_router:topics()).
t_cleanup_monitor_node_down(_Config) -> t_cleanup_monitor_node_down(_Config) ->
Slave = emqx_cth_cluster:start_bare_node(?FUNCTION_NAME, #{driver => ct_slave}), [Slave] = emqx_cth_cluster:start_bare_nodes([node4]),
emqx_router:add_route(<<"a/b/c">>, Slave), emqx_router:add_route(<<"a/b/c">>, Slave),
emqx_router:add_route(<<"d/e/f">>, node()), emqx_router:add_route(<<"d/e/f">>, node()),
?assertMatch([_, _], emqx_router:topics()), ?assertMatch([_, _], emqx_router:topics()),

View File

@ -218,38 +218,41 @@ t_routing_schema_switch(VFrom, VTo, Config) ->
], ],
#{work_dir => WorkDir} #{work_dir => WorkDir}
), ),
% Verify that new nodes switched to schema v1/v2 in presence of v1/v2 routes respectively
Nodes = [Node1, Node2, Node3], Nodes = [Node1, Node2, Node3],
?assertEqual( try
[{ok, VTo}, {ok, VTo}, {ok, VTo}], % Verify that new nodes switched to schema v1/v2 in presence of v1/v2 routes respectively
erpc:multicall(Nodes, emqx_router, get_schema_vsn, []) ?assertEqual(
), [{ok, VTo}, {ok, VTo}, {ok, VTo}],
% Wait for all nodes to agree on cluster state erpc:multicall(Nodes, emqx_router, get_schema_vsn, [])
?retry( ),
500, % Wait for all nodes to agree on cluster state
10, ?retry(
?assertMatch( 500,
[{ok, [Node1, Node2, Node3]}], 10,
lists:usort(erpc:multicall(Nodes, emqx, running_nodes, [])) ?assertMatch(
) [{ok, [Node1, Node2, Node3]}],
), lists:usort(erpc:multicall(Nodes, emqx, running_nodes, []))
% Verify that routing works as expected )
C2 = start_client(Node2), ),
ok = subscribe(C2, <<"a/+/d">>), % Verify that routing works as expected
C3 = start_client(Node3), C2 = start_client(Node2),
ok = subscribe(C3, <<"d/e/f/#">>), ok = subscribe(C2, <<"a/+/d">>),
{ok, _} = publish(C1, <<"a/b/d">>, <<"hey-newbies">>), C3 = start_client(Node3),
{ok, _} = publish(C2, <<"a/b/c">>, <<"hi">>), ok = subscribe(C3, <<"d/e/f/#">>),
{ok, _} = publish(C3, <<"d/e/f/42">>, <<"hello">>), {ok, _} = publish(C1, <<"a/b/d">>, <<"hey-newbies">>),
?assertReceive({pub, C2, #{topic := <<"a/b/d">>, payload := <<"hey-newbies">>}}), {ok, _} = publish(C2, <<"a/b/c">>, <<"hi">>),
?assertReceive({pub, C1, #{topic := <<"a/b/c">>, payload := <<"hi">>}}), {ok, _} = publish(C3, <<"d/e/f/42">>, <<"hello">>),
?assertReceive({pub, C1, #{topic := <<"d/e/f/42">>, payload := <<"hello">>}}), ?assertReceive({pub, C2, #{topic := <<"a/b/d">>, payload := <<"hey-newbies">>}}),
?assertReceive({pub, C3, #{topic := <<"d/e/f/42">>, payload := <<"hello">>}}), ?assertReceive({pub, C1, #{topic := <<"a/b/c">>, payload := <<"hi">>}}),
?assertNotReceive(_), ?assertReceive({pub, C1, #{topic := <<"d/e/f/42">>, payload := <<"hello">>}}),
ok = emqtt:stop(C1), ?assertReceive({pub, C3, #{topic := <<"d/e/f/42">>, payload := <<"hello">>}}),
ok = emqtt:stop(C2), ?assertNotReceive(_),
ok = emqtt:stop(C3), ok = emqtt:stop(C1),
ok = emqx_cth_cluster:stop(Nodes). ok = emqtt:stop(C2),
ok = emqtt:stop(C3)
after
ok = emqx_cth_cluster:stop(Nodes)
end.
%% %%

View File

@ -63,6 +63,7 @@ init_per_suite(Config) ->
end, end,
emqx_common_test_helpers:boot_modules(all), emqx_common_test_helpers:boot_modules(all),
emqx_common_test_helpers:start_apps([]), emqx_common_test_helpers:start_apps([]),
emqx_logger:set_log_level(debug),
[{dist_pid, DistPid} | Config]. [{dist_pid, DistPid} | Config].
end_per_suite(Config) -> end_per_suite(Config) ->
@ -574,7 +575,7 @@ t_local(Config) when is_list(Config) ->
<<"sticky_group">> => sticky <<"sticky_group">> => sticky
}, },
Node = start_slave('local_shared_sub_testtesttest', 21999), Node = start_slave('local_shared_sub_local_1', 21999),
ok = ensure_group_config(GroupConfig), ok = ensure_group_config(GroupConfig),
ok = ensure_group_config(Node, GroupConfig), ok = ensure_group_config(Node, GroupConfig),
@ -627,7 +628,7 @@ t_remote(Config) when is_list(Config) ->
<<"sticky_group">> => sticky <<"sticky_group">> => sticky
}, },
Node = start_slave('remote_shared_sub_testtesttest', 21999), Node = start_slave('remote_shared_sub_remote_1', 21999),
ok = ensure_group_config(GroupConfig), ok = ensure_group_config(GroupConfig),
ok = ensure_group_config(Node, GroupConfig), ok = ensure_group_config(Node, GroupConfig),
@ -676,7 +677,7 @@ t_local_fallback(Config) when is_list(Config) ->
Topic = <<"local_foo/bar">>, Topic = <<"local_foo/bar">>,
ClientId1 = <<"ClientId1">>, ClientId1 = <<"ClientId1">>,
ClientId2 = <<"ClientId2">>, ClientId2 = <<"ClientId2">>,
Node = start_slave('local_fallback_shared_sub_test', 11888), Node = start_slave('local_fallback_shared_sub_1', 11888),
{ok, ConnPid1} = emqtt:start_link([{clientid, ClientId1}]), {ok, ConnPid1} = emqtt:start_link([{clientid, ClientId1}]),
{ok, _} = emqtt:connect(ConnPid1), {ok, _} = emqtt:connect(ConnPid1),
@ -1253,34 +1254,24 @@ recv_msgs(Count, Msgs) ->
end. end.
start_slave(Name, Port) -> start_slave(Name, Port) ->
{ok, Node} = ct_slave:start( {ok, Node} = emqx_cth_peer:start_link(
list_to_atom(atom_to_list(Name) ++ "@" ++ host()), Name,
[ ebin_path()
{kill_if_fail, true},
{monitor_master, true},
{init_timeout, 10000},
{startup_timeout, 10000},
{erl_flags, ebin_path()}
]
), ),
pong = net_adm:ping(Node), pong = net_adm:ping(Node),
setup_node(Node, Port), setup_node(Node, Port),
Node. Node.
stop_slave(Node) -> stop_slave(Node) ->
rpc:call(Node, mria, leave, []), rpc:call(Node, mria, leave, []),
ct_slave:stop(Node). emqx_cth_peer:stop(Node).
host() -> host() ->
[_, Host] = string:tokens(atom_to_list(node()), "@"), [_, Host] = string:tokens(atom_to_list(node()), "@"),
Host. Host.
ebin_path() -> ebin_path() ->
string:join(["-pa" | lists:filter(fun is_lib/1, code:get_path())], " "). ["-pa" | code:get_path()].
is_lib(Path) ->
string:prefix(Path, code:lib_dir()) =:= nomatch.
setup_node(Node, Port) -> setup_node(Node, Port) ->
EnvHandler = EnvHandler =

View File

@ -126,7 +126,7 @@ check(Conf) when is_map(Conf) ->
%% erlfmt-ignore %% erlfmt-ignore
%% this is config generated from v5.0.11 %% this is config generated from v5.0.11
webhook_v5011_hocon() -> webhook_v5011_hocon() ->
""" "
bridges{ bridges{
webhook { webhook {
the_name{ the_name{
@ -143,7 +143,7 @@ bridges{
} }
} }
} }
""". ".
full_webhook_v5011_hocon() -> full_webhook_v5011_hocon() ->
"" ""
@ -215,7 +215,7 @@ full_webhook_v5019_hocon() ->
%% erlfmt-ignore %% erlfmt-ignore
%% this is a generated from v5.0.11 %% this is a generated from v5.0.11
mqtt_v5011_hocon() -> mqtt_v5011_hocon() ->
""" "
bridges { bridges {
mqtt { mqtt {
bridge_one { bridge_one {
@ -257,12 +257,12 @@ bridges {
} }
} }
} }
""". ".
%% erlfmt-ignore %% erlfmt-ignore
%% a more complete version %% a more complete version
mqtt_v5011_full_hocon() -> mqtt_v5011_full_hocon() ->
""" "
bridges { bridges {
mqtt { mqtt {
bridge_one { bridge_one {
@ -330,4 +330,4 @@ bridges {
} }
} }
} }
""". ".

View File

@ -2,7 +2,7 @@
{erl_opts, [debug_info]}. {erl_opts, [debug_info]}.
{deps, [ {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.8.0"}}} {deps, [ {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.8.0"}}}
, {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.3"}}} , {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.3"}}}
, {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.0"}}} , {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}}
, {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}} , {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}}
, {snappyer, "1.2.9"} , {snappyer, "1.2.9"}
, {emqx_connector, {path, "../../apps/emqx_connector"}} , {emqx_connector, {path, "../../apps/emqx_connector"}}

View File

@ -12,7 +12,7 @@
%% erlfmt-ignore %% erlfmt-ignore
aeh_producer_hocon() -> aeh_producer_hocon() ->
""" "
bridges.azure_event_hub_producer.my_producer { bridges.azure_event_hub_producer.my_producer {
enable = true enable = true
authentication { authentication {
@ -62,7 +62,7 @@ bridges.azure_event_hub_producer.my_producer {
server_name_indication = auto server_name_indication = auto
} }
} }
""". ".
%%=========================================================================== %%===========================================================================
%% Helper functions %% Helper functions

View File

@ -2,7 +2,7 @@
{erl_opts, [debug_info]}. {erl_opts, [debug_info]}.
{deps, [ {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.8.0"}}} {deps, [ {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.8.0"}}}
, {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.3"}}} , {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.3"}}}
, {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.0"}}} , {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}}
, {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}} , {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}}
, {snappyer, "1.2.9"} , {snappyer, "1.2.9"}
, {emqx_connector, {path, "../../apps/emqx_connector"}} , {emqx_connector, {path, "../../apps/emqx_connector"}}

View File

@ -12,7 +12,7 @@
%% erlfmt-ignore %% erlfmt-ignore
confluent_producer_action_hocon() -> confluent_producer_action_hocon() ->
""" "
actions.confluent_producer.my_producer { actions.confluent_producer.my_producer {
enable = true enable = true
connector = my_connector connector = my_connector
@ -40,7 +40,7 @@ actions.confluent_producer.my_producer {
} }
local_topic = \"t/confluent\" local_topic = \"t/confluent\"
} }
""". ".
confluent_producer_connector_hocon() -> confluent_producer_connector_hocon() ->
"" ""

View File

@ -12,7 +12,7 @@
%% erlfmt-ignore %% erlfmt-ignore
gcp_pubsub_producer_hocon() -> gcp_pubsub_producer_hocon() ->
""" "
bridges.gcp_pubsub.my_producer { bridges.gcp_pubsub.my_producer {
attributes_template = [ attributes_template = [
{key = \"${payload.key}\", value = fixed_value} {key = \"${payload.key}\", value = fixed_value}
@ -54,7 +54,7 @@ bridges.gcp_pubsub.my_producer {
type = service_account type = service_account
} }
} }
""". ".
%%=========================================================================== %%===========================================================================
%% Helper functions %% Helper functions

View File

@ -175,7 +175,7 @@ check_atom_key(Conf) when is_map(Conf) ->
%% erlfmt-ignore %% erlfmt-ignore
webhook_config_hocon() -> webhook_config_hocon() ->
""" "
bridges.webhook.a { bridges.webhook.a {
body = \"${.}\" body = \"${.}\"
connect_timeout = 15s connect_timeout = 15s
@ -209,4 +209,4 @@ bridges.webhook.a {
} }
url = \"http://some.host:4000/api/echo\" url = \"http://some.host:4000/api/echo\"
} }
""". ".

View File

@ -2,7 +2,7 @@
{erl_opts, [debug_info]}. {erl_opts, [debug_info]}.
{deps, [ {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.8.0"}}} {deps, [ {wolff, {git, "https://github.com/kafka4beam/wolff.git", {tag, "1.8.0"}}}
, {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.3"}}} , {kafka_protocol, {git, "https://github.com/kafka4beam/kafka_protocol.git", {tag, "4.1.3"}}}
, {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.0"}}} , {brod_gssapi, {git, "https://github.com/kafka4beam/brod_gssapi.git", {tag, "v0.1.1"}}}
, {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}} , {brod, {git, "https://github.com/kafka4beam/brod.git", {tag, "3.16.8"}}}
, {snappyer, "1.2.9"} , {snappyer, "1.2.9"}
, {emqx_connector, {path, "../../apps/emqx_connector"}} , {emqx_connector, {path, "../../apps/emqx_connector"}}

View File

@ -73,7 +73,7 @@ check_atom_key(Conf) when is_map(Conf) ->
%% erlfmt-ignore %% erlfmt-ignore
pulsar_producer_hocon() -> pulsar_producer_hocon() ->
""" "
bridges.pulsar_producer.my_producer { bridges.pulsar_producer.my_producer {
enable = true enable = true
servers = \"localhost:6650\" servers = \"localhost:6650\"
@ -90,4 +90,4 @@ bridges.pulsar_producer.my_producer {
server_name_indication = \"auto\" server_name_indication = \"auto\"
} }
} }
""". ".

View File

@ -24,7 +24,7 @@
%% erlfmt-ignore %% erlfmt-ignore
-define(BASE_CONF, -define(BASE_CONF,
""" "
log { log {
console { console {
enable = true enable = true
@ -36,7 +36,7 @@
path = \"log/emqx.log\" path = \"log/emqx.log\"
} }
} }
"""). ").
all() -> all() ->
emqx_common_test_helpers:all(?MODULE). emqx_common_test_helpers:all(?MODULE).

View File

@ -20,7 +20,7 @@
%% erlfmt-ignore %% erlfmt-ignore
-define(BASE_CONF, -define(BASE_CONF,
""" "
node { node {
name = \"emqx1@127.0.0.1\" name = \"emqx1@127.0.0.1\"
cookie = \"emqxsecretcookie\" cookie = \"emqxsecretcookie\"
@ -34,7 +34,7 @@
static.seeds = ~p static.seeds = ~p
core_nodes = ~p core_nodes = ~p
} }
"""). ").
array_nodes_test() -> array_nodes_test() ->
ensure_acl_conf(), ensure_acl_conf(),
@ -70,7 +70,7 @@ array_nodes_test() ->
%% erlfmt-ignore %% erlfmt-ignore
-define(OUTDATED_LOG_CONF, -define(OUTDATED_LOG_CONF,
""" "
log.console_handler { log.console_handler {
burst_limit { burst_limit {
enable = true enable = true
@ -124,7 +124,7 @@ log.file_handlers {
time_offset = \"+01:00\" time_offset = \"+01:00\"
} }
} }
""" "
). ).
-define(FORMATTER(TimeOffset), -define(FORMATTER(TimeOffset),
{emqx_logger_textfmt, #{ {emqx_logger_textfmt, #{
@ -196,7 +196,7 @@ validate_log(Conf) ->
%% erlfmt-ignore %% erlfmt-ignore
-define(FILE_LOG_BASE_CONF, -define(FILE_LOG_BASE_CONF,
""" "
log.file.default { log.file.default {
enable = true enable = true
file = \"log/xx-emqx.log\" file = \"log/xx-emqx.log\"
@ -206,7 +206,7 @@ validate_log(Conf) ->
rotation_size = ~s rotation_size = ~s
time_offset = \"+01:00\" time_offset = \"+01:00\"
} }
""" "
). ).
file_log_infinity_rotation_size_test_() -> file_log_infinity_rotation_size_test_() ->
@ -249,7 +249,7 @@ file_log_infinity_rotation_size_test_() ->
%% erlfmt-ignore %% erlfmt-ignore
-define(KERNEL_LOG_CONF, -define(KERNEL_LOG_CONF,
""" "
log.console { log.console {
enable = true enable = true
formatter = text formatter = text
@ -269,7 +269,7 @@ file_log_infinity_rotation_size_test_() ->
enable = true enable = true
file = \"log/my-emqx.log\" file = \"log/my-emqx.log\"
} }
""" "
). ).
log_test() -> log_test() ->
@ -279,7 +279,7 @@ log_test() ->
log_rotation_count_limit_test() -> log_rotation_count_limit_test() ->
ensure_acl_conf(), ensure_acl_conf(),
Format = Format =
""" "
log.file { log.file {
enable = true enable = true
path = \"log/emqx.log\" path = \"log/emqx.log\"
@ -288,7 +288,7 @@ log_rotation_count_limit_test() ->
rotation = {count = ~w} rotation = {count = ~w}
rotation_size = \"1024MB\" rotation_size = \"1024MB\"
} }
""", ",
BaseConf = to_bin(?BASE_CONF, ["emqx1@127.0.0.1", "emqx1@127.0.0.1"]), BaseConf = to_bin(?BASE_CONF, ["emqx1@127.0.0.1", "emqx1@127.0.0.1"]),
lists:foreach(fun({Conf, Count}) -> lists:foreach(fun({Conf, Count}) ->
Conf0 = <<BaseConf/binary, Conf/binary>>, Conf0 = <<BaseConf/binary, Conf/binary>>,
@ -320,7 +320,7 @@ log_rotation_count_limit_test() ->
%% erlfmt-ignore %% erlfmt-ignore
-define(BASE_AUTHN_ARRAY, -define(BASE_AUTHN_ARRAY,
""" "
authentication = [ authentication = [
{backend = \"http\" {backend = \"http\"
body {password = \"${password}\", username = \"${username}\"} body {password = \"${password}\", username = \"${username}\"}
@ -335,7 +335,7 @@ log_rotation_count_limit_test() ->
url = \"~ts\" url = \"~ts\"
} }
] ]
""" "
). ).
-define(ERROR(Error), -define(ERROR(Error),
@ -396,13 +396,13 @@ authn_validations_test() ->
%% erlfmt-ignore %% erlfmt-ignore
-define(LISTENERS, -define(LISTENERS,
""" "
listeners.ssl.default.bind = 9999 listeners.ssl.default.bind = 9999
listeners.wss.default.bind = 9998 listeners.wss.default.bind = 9998
listeners.wss.default.ssl_options.cacertfile = \"mytest/certs/cacert.pem\" listeners.wss.default.ssl_options.cacertfile = \"mytest/certs/cacert.pem\"
listeners.wss.new.bind = 9997 listeners.wss.new.bind = 9997
listeners.wss.new.websocket.mqtt_path = \"/my-mqtt\" listeners.wss.new.websocket.mqtt_path = \"/my-mqtt\"
""" "
). ).
listeners_test() -> listeners_test() ->

View File

@ -57,7 +57,7 @@
%% erlfmt-ignore %% erlfmt-ignore
-define(SYSMON_EXAMPLE, -define(SYSMON_EXAMPLE,
<<""" <<"
sysmon { sysmon {
os { os {
cpu_check_interval = 60s cpu_check_interval = 60s
@ -78,7 +78,7 @@
process_low_watermark = 60% process_low_watermark = 60%
} }
} }
""">> ">>
). ).
api_spec() -> api_spec() ->

View File

@ -399,7 +399,7 @@ do_install_package(FileName, Bin) ->
end, end,
{400, #{ {400, #{
code => 'BAD_PLUGIN_INFO', code => 'BAD_PLUGIN_INFO',
message => iolist_to_binary([Reason, ":", FileName]) message => iolist_to_binary([Reason, ": ", FileName])
}} }}
end. end.
@ -445,7 +445,8 @@ install_package(FileName, Bin) ->
case emqx_plugins:ensure_installed(PackageName) of case emqx_plugins:ensure_installed(PackageName) of
{error, #{return := not_found}} = NotFound -> {error, #{return := not_found}} = NotFound ->
NotFound; NotFound;
{error, _Reason} = Error -> {error, Reason} = Error ->
?SLOG(error, Reason#{msg => "failed_to_install_plugin"}),
_ = file:delete(File), _ = file:delete(File),
Error; Error;
Result -> Result ->

View File

@ -214,7 +214,22 @@ t_kickout_clients(_) ->
{ok, C3} = emqtt:start_link(#{clientid => ClientId3}), {ok, C3} = emqtt:start_link(#{clientid => ClientId3}),
{ok, _} = emqtt:connect(C3), {ok, _} = emqtt:connect(C3),
timer:sleep(300), emqx_common_test_helpers:wait_for(
?FUNCTION_NAME,
?LINE,
fun() ->
try
[_] = emqx_cm:lookup_channels(ClientId1),
[_] = emqx_cm:lookup_channels(ClientId2),
[_] = emqx_cm:lookup_channels(ClientId3),
true
catch
error:badmatch ->
false
end
end,
2000
),
%% get /clients %% get /clients
ClientsPath = emqx_mgmt_api_test_util:api_path(["clients"]), ClientsPath = emqx_mgmt_api_test_util:api_path(["clients"]),
@ -233,6 +248,15 @@ t_kickout_clients(_) ->
KickoutBody = [ClientId1, ClientId2, ClientId3], KickoutBody = [ClientId1, ClientId2, ClientId3],
{ok, 204, _} = emqx_mgmt_api_test_util:request_api_with_body(post, KickoutPath, KickoutBody), {ok, 204, _} = emqx_mgmt_api_test_util:request_api_with_body(post, KickoutPath, KickoutBody),
ReceiveExit = fun({ClientPid, ClientId}) ->
receive
{'EXIT', Pid, _} when Pid =:= ClientPid ->
ok
after 1000 ->
error({timeout, ClientId})
end
end,
lists:foreach(ReceiveExit, [{C1, ClientId1}, {C2, ClientId2}, {C3, ClientId3}]),
{ok, Clients2} = emqx_mgmt_api_test_util:request_api(get, ClientsPath), {ok, Clients2} = emqx_mgmt_api_test_util:request_api(get, ClientsPath),
ClientsResponse2 = emqx_utils_json:decode(Clients2, [return_maps]), ClientsResponse2 = emqx_utils_json:decode(Clients2, [return_maps]),
?assertMatch(#{<<"meta">> := #{<<"count">> := 0}}, ClientsResponse2). ?assertMatch(#{<<"meta">> := #{<<"count">> := 0}}, ClientsResponse2).

View File

@ -83,7 +83,7 @@
describe(NameVsn) -> read_plugin(NameVsn, #{fill_readme => true}). describe(NameVsn) -> read_plugin(NameVsn, #{fill_readme => true}).
%% @doc Install a .tar.gz package placed in install_dir. %% @doc Install a .tar.gz package placed in install_dir.
-spec ensure_installed(name_vsn()) -> ok | {error, any()}. -spec ensure_installed(name_vsn()) -> ok | {error, map()}.
ensure_installed(NameVsn) -> ensure_installed(NameVsn) ->
case read_plugin(NameVsn, #{}) of case read_plugin(NameVsn, #{}) of
{ok, _} -> {ok, _} ->

View File

@ -750,7 +750,7 @@ group_t_copy_plugin_to_a_new_node_single_node({init, Config}) ->
| Config | Config
]; ];
group_t_copy_plugin_to_a_new_node_single_node({'end', Config}) -> group_t_copy_plugin_to_a_new_node_single_node({'end', Config}) ->
CopyToNode = proplists:get_value(copy_to_node, Config), CopyToNode = proplists:get_value(copy_to_node_name, Config),
ok = emqx_common_test_helpers:stop_slave(CopyToNode), ok = emqx_common_test_helpers:stop_slave(CopyToNode),
ok = file:del_dir_r(proplists:get_value(to_install_dir, Config)), ok = file:del_dir_r(proplists:get_value(to_install_dir, Config)),
ok; ok;

View File

@ -134,7 +134,7 @@ check(Conf) when is_map(Conf) ->
%% erlfmt-ignore %% erlfmt-ignore
webhook_bridge_health_check_hocon(HealthCheckInterval) -> webhook_bridge_health_check_hocon(HealthCheckInterval) ->
io_lib:format( io_lib:format(
""" "
bridges.webhook.simple { bridges.webhook.simple {
url = \"http://localhost:4000\" url = \"http://localhost:4000\"
body = \"body\" body = \"body\"
@ -142,5 +142,5 @@ bridges.webhook.simple {
health_check_interval = \"~s\" health_check_interval = \"~s\"
} }
} }
""", ",
[HealthCheckInterval]). [HealthCheckInterval]).

View File

@ -24,7 +24,7 @@
%% erlfmt-ignore %% erlfmt-ignore
republish_hocon0() -> republish_hocon0() ->
""" "
rule_engine.rules.my_rule { rule_engine.rules.my_rule {
description = \"some desc\" description = \"some desc\"
metadata = {created_at = 1693918992079} metadata = {created_at = 1693918992079}
@ -55,7 +55,7 @@ rule_engine.rules.my_rule {
} }
] ]
} }
""". ".
%%=========================================================================== %%===========================================================================
%% Helper functions %% Helper functions

View File

@ -869,7 +869,7 @@ stop_slave(Node) ->
% This line don't work!! % This line don't work!!
%emqx_cluster_rpc:fast_forward_to_commit(Node, 100), %emqx_cluster_rpc:fast_forward_to_commit(Node, 100),
rpc:call(Node, ?MODULE, leave_cluster, []), rpc:call(Node, ?MODULE, leave_cluster, []),
ok = slave:stop(Node), ok = emqx_cth_peer:stop(Node),
?assertEqual([node()], mria:running_nodes()), ?assertEqual([node()], mria:running_nodes()),
?assertEqual([], nodes()), ?assertEqual([], nodes()),
_ = application:stop(mria), _ = application:stop(mria),

10
mix.exs
View File

@ -46,17 +46,17 @@ defmodule EMQXUmbrella.MixProject do
# other exact versions, and not ranges. # other exact versions, and not ranges.
[ [
{:lc, github: "emqx/lc", tag: "0.3.2", override: true}, {:lc, github: "emqx/lc", tag: "0.3.2", override: true},
{:redbug, "2.0.8"}, {:redbug, github: "emqx/redbug", tag: "2.0.10"},
{:covertool, github: "zmstone/covertool", tag: "2.0.4.1", override: true}, {:covertool, github: "zmstone/covertool", tag: "2.0.4.1", override: true},
{:typerefl, github: "ieQu1/typerefl", tag: "0.9.1", override: true}, {:typerefl, github: "ieQu1/typerefl", tag: "0.9.1", override: true},
{:ehttpc, github: "emqx/ehttpc", tag: "0.4.11", override: true}, {:ehttpc, github: "emqx/ehttpc", tag: "0.4.11", override: true},
{:gproc, github: "emqx/gproc", tag: "0.9.0.1", override: true}, {:gproc, github: "emqx/gproc", tag: "0.9.0.1", override: true},
{:jiffy, github: "emqx/jiffy", tag: "1.0.5", override: true}, {:jiffy, github: "emqx/jiffy", tag: "1.0.5", override: true},
{:cowboy, github: "emqx/cowboy", tag: "2.9.2", override: true}, {:cowboy, github: "emqx/cowboy", tag: "2.9.2", override: true},
{:esockd, github: "emqx/esockd", tag: "5.9.7", override: true}, {:esockd, github: "emqx/esockd", tag: "5.9.8", override: true},
{:rocksdb, github: "emqx/erlang-rocksdb", tag: "1.8.0-emqx-1", override: true}, {:rocksdb, github: "emqx/erlang-rocksdb", tag: "1.8.0-emqx-1", override: true},
{:ekka, github: "emqx/ekka", tag: "0.15.16", override: true}, {:ekka, github: "emqx/ekka", tag: "0.15.16", override: true},
{:gen_rpc, github: "emqx/gen_rpc", tag: "3.2.1", override: true}, {:gen_rpc, github: "emqx/gen_rpc", tag: "3.2.2", override: true},
{:grpc, github: "emqx/grpc-erl", tag: "0.6.8", override: true}, {:grpc, github: "emqx/grpc-erl", tag: "0.6.8", override: true},
{:minirest, github: "emqx/minirest", tag: "1.3.14", override: true}, {:minirest, github: "emqx/minirest", tag: "1.3.14", override: true},
{:ecpool, github: "emqx/ecpool", tag: "0.5.4", override: true}, {:ecpool, github: "emqx/ecpool", tag: "0.5.4", override: true},
@ -230,7 +230,7 @@ defmodule EMQXUmbrella.MixProject do
{:influxdb, github: "emqx/influxdb-client-erl", tag: "1.1.11", override: true}, {:influxdb, github: "emqx/influxdb-client-erl", tag: "1.1.11", override: true},
{:wolff, github: "kafka4beam/wolff", tag: "1.8.0"}, {:wolff, github: "kafka4beam/wolff", tag: "1.8.0"},
{:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.3", override: true}, {:kafka_protocol, github: "kafka4beam/kafka_protocol", tag: "4.1.3", override: true},
{:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.0"}, {:brod_gssapi, github: "kafka4beam/brod_gssapi", tag: "v0.1.1"},
{:brod, github: "kafka4beam/brod", tag: "3.16.8"}, {:brod, github: "kafka4beam/brod", tag: "3.16.8"},
{:snappyer, "1.2.9", override: true}, {:snappyer, "1.2.9", override: true},
{:crc32cer, "0.1.8", override: true}, {:crc32cer, "0.1.8", override: true},
@ -823,7 +823,7 @@ defmodule EMQXUmbrella.MixProject do
defp jq_dep() do defp jq_dep() do
if enable_jq?(), if enable_jq?(),
do: [{:jq, github: "emqx/jq", tag: "v0.3.11", override: true}], do: [{:jq, github: "emqx/jq", tag: "v0.3.12", override: true}],
else: [] else: []
end end

View File

@ -51,7 +51,7 @@
{deps, {deps,
[ {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.2"}}} [ {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.2"}}}
, {redbug, "2.0.8"} , {redbug, {git, "https://github.com/emqx/redbug", {tag, "2.0.10"}}}
, {covertool, {git, "https://github.com/zmstone/covertool", {tag, "2.0.4.1"}}} , {covertool, {git, "https://github.com/zmstone/covertool", {tag, "2.0.4.1"}}}
, {gpb, "4.19.9"} , {gpb, "4.19.9"}
, {typerefl, {git, "https://github.com/ieQu1/typerefl", {tag, "0.9.1"}}} , {typerefl, {git, "https://github.com/ieQu1/typerefl", {tag, "0.9.1"}}}
@ -60,10 +60,10 @@
, {gproc, {git, "https://github.com/emqx/gproc", {tag, "0.9.0.1"}}} , {gproc, {git, "https://github.com/emqx/gproc", {tag, "0.9.0.1"}}}
, {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}} , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}
, {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}} , {cowboy, {git, "https://github.com/emqx/cowboy", {tag, "2.9.2"}}}
, {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.7"}}} , {esockd, {git, "https://github.com/emqx/esockd", {tag, "5.9.8"}}}
, {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb", {tag, "1.8.0-emqx-1"}}} , {rocksdb, {git, "https://github.com/emqx/erlang-rocksdb", {tag, "1.8.0-emqx-1"}}}
, {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.15.16"}}} , {ekka, {git, "https://github.com/emqx/ekka", {tag, "0.15.16"}}}
, {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.2.1"}}} , {gen_rpc, {git, "https://github.com/emqx/gen_rpc", {tag, "3.2.2"}}}
, {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.8"}}} , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.8"}}}
, {minirest, {git, "https://github.com/emqx/minirest", {tag, "1.3.14"}}} , {minirest, {git, "https://github.com/emqx/minirest", {tag, "1.3.14"}}}
, {ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.4"}}} , {ecpool, {git, "https://github.com/emqx/ecpool", {tag, "0.5.4"}}}

View File

@ -16,7 +16,7 @@ do(Dir, CONFIG) ->
assert_otp() -> assert_otp() ->
Oldest = 24, Oldest = 24,
Latest = 25, Latest = 26,
OtpRelease = list_to_integer(erlang:system_info(otp_release)), OtpRelease = list_to_integer(erlang:system_info(otp_release)),
case OtpRelease < Oldest orelse OtpRelease > Latest of case OtpRelease < Oldest orelse OtpRelease > Latest of
true -> true ->
@ -42,7 +42,7 @@ quicer() ->
{quicer, {git, "https://github.com/emqx/quic.git", {tag, "0.0.202"}}}. {quicer, {git, "https://github.com/emqx/quic.git", {tag, "0.0.202"}}}.
jq() -> jq() ->
{jq, {git, "https://github.com/emqx/jq", {tag, "v0.3.11"}}}. {jq, {git, "https://github.com/emqx/jq", {tag, "v0.3.12"}}}.
deps(Config) -> deps(Config) ->
{deps, OldDeps} = lists:keyfind(deps, 1, Config), {deps, OldDeps} = lists:keyfind(deps, 1, Config),
@ -53,7 +53,10 @@ deps(Config) ->
lists:keystore(deps, 1, Config, {deps, OldDeps ++ MoreDeps}). lists:keystore(deps, 1, Config, {deps, OldDeps ++ MoreDeps}).
overrides() -> overrides() ->
[{add, [{extra_src_dirs, [{"etc", [{recursive, true}]}]}]}] ++ snabbkaffe_overrides(). [
{add, [{extra_src_dirs, [{"etc", [{recursive, true}]}]}]},
{add, jesse, [{erl_opts, [nowarn_match_float_zero]}]}
] ++ snabbkaffe_overrides().
%% Temporary workaround for a rebar3 erl_opts duplication %% Temporary workaround for a rebar3 erl_opts duplication
%% bug. Ideally, we want to set this define globally %% bug. Ideally, we want to set this define globally

View File

@ -18,6 +18,9 @@ case ${OTP_VSN} in
25*) 25*)
VERSION="3.19.0-emqx-9" VERSION="3.19.0-emqx-9"
;; ;;
26*)
VERSION="3.20.0-emqx-1"
;;
*) *)
echo "Unsupporetd Erlang/OTP version $OTP_VSN" echo "Unsupporetd Erlang/OTP version $OTP_VSN"
exit 1 exit 1