yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(rebalance): fix start order of rebalance applications

This commit is contained in:
Ilya Averyanov 2024-04-12 14:37:10 +03:00
parent e7a4210943
commit 500d4fedda
13 changed files with 73 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
apps/emqx_eviction_agent/src/emqx_eviction_agent.app.src
View File

@ -9,7 +9,8 @@
{applications, [ {applications, [
kernel, kernel,
stdlib, stdlib,
emqx_ctl emqx_ctl,
emqx
]}, ]},
{mod, {emqx_eviction_agent_app, []}}, {mod, {emqx_eviction_agent_app, []}},
{env, []}, {env, []},

7
apps/emqx_node_rebalance/src/emqx_node_rebalance.app.src
View File

@ -1,6 +1,6 @@
{application, emqx_node_rebalance, [ {application, emqx_node_rebalance, [
{description, "EMQX Node Rebalance"}, {description, "EMQX Node Rebalance"},
{vsn, "5.0.7"}, {vsn, "5.0.8"},
{registered, [ {registered, [
emqx_node_rebalance_sup, emqx_node_rebalance_sup,
emqx_node_rebalance, emqx_node_rebalance,
@ -10,7 +10,10 @@
]}, ]},
{applications, [ {applications, [
kernel, kernel,
stdlib stdlib,
emqx,
emqx_ctl,
emqx_eviction_agent
]}, ]},
{mod, {emqx_node_rebalance_app, []}}, {mod, {emqx_node_rebalance_app, []}},
{env, []}, {env, []},

108
apps/emqx_node_rebalance/src/emqx_node_rebalance_api.erl
View File

@ -423,7 +423,7 @@ param_node() ->
fields(rebalance_start) -> fields(rebalance_start) ->
[ [
{"wait_health_check", {wait_health_check,
mk( mk(
emqx_schema:timeout_duration_s(), emqx_schema:timeout_duration_s(),
#{ #{
@ -431,7 +431,7 @@ fields(rebalance_start) ->
required => false required => false
} }
)}, )},
{"conn_evict_rate", {conn_evict_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -439,7 +439,7 @@ fields(rebalance_start) ->
required => false required => false
} }
)}, )},
{"sess_evict_rate", {sess_evict_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -447,7 +447,7 @@ fields(rebalance_start) ->
required => false required => false
} }
)}, )},
{"abs_conn_threshold", {abs_conn_threshold,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -455,7 +455,7 @@ fields(rebalance_start) ->
required => false required => false
} }
)}, )},
{"rel_conn_threshold", {rel_conn_threshold,
mk( mk(
number(), number(),
#{ #{
@ -464,7 +464,7 @@ fields(rebalance_start) ->
validator => [fun(Value) -> Value > 1.0 end] validator => [fun(Value) -> Value > 1.0 end]
} }
)}, )},
{"abs_sess_threshold", {abs_sess_threshold,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -472,7 +472,7 @@ fields(rebalance_start) ->
required => false required => false
} }
)}, )},
{"rel_sess_threshold", {rel_sess_threshold,
mk( mk(
number(), number(),
#{ #{
@ -481,7 +481,7 @@ fields(rebalance_start) ->
validator => [fun(Value) -> Value > 1.0 end] validator => [fun(Value) -> Value > 1.0 end]
} }
)}, )},
{"wait_takeover", {wait_takeover,
mk( mk(
emqx_schema:timeout_duration_s(), emqx_schema:timeout_duration_s(),
#{ #{
@ -489,7 +489,7 @@ fields(rebalance_start) ->
required => false required => false
} }
)}, )},
{"nodes", {nodes,
mk( mk(
list(binary()), list(binary()),
#{ #{
@ -501,7 +501,7 @@ fields(rebalance_start) ->
]; ];
fields(rebalance_evacuation_start) -> fields(rebalance_evacuation_start) ->
[ [
{"wait_health_check", {wait_health_check,
mk( mk(
emqx_schema:timeout_duration_s(), emqx_schema:timeout_duration_s(),
#{ #{
@ -509,7 +509,7 @@ fields(rebalance_evacuation_start) ->
required => false required => false
} }
)}, )},
{"conn_evict_rate", {conn_evict_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -517,7 +517,7 @@ fields(rebalance_evacuation_start) ->
required => false required => false
} }
)}, )},
{"sess_evict_rate", {sess_evict_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -525,7 +525,7 @@ fields(rebalance_evacuation_start) ->
required => false required => false
} }
)}, )},
{"redirect_to", {redirect_to,
mk( mk(
binary(), binary(),
#{ #{
@ -533,7 +533,7 @@ fields(rebalance_evacuation_start) ->
required => false required => false
} }
)}, )},
{"wait_takeover", {wait_takeover,
mk( mk(
emqx_schema:timeout_duration_s(), emqx_schema:timeout_duration_s(),
#{ #{
@ -541,7 +541,7 @@ fields(rebalance_evacuation_start) ->
required => false required => false
} }
)}, )},
{"migrate_to", {migrate_to,
mk( mk(
nonempty_list(binary()), nonempty_list(binary()),
#{ #{
@ -552,7 +552,7 @@ fields(rebalance_evacuation_start) ->
]; ];
fields(purge_start) -> fields(purge_start) ->
[ [
{"purge_rate", {purge_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -563,7 +563,7 @@ fields(purge_start) ->
]; ];
fields(local_status_disabled) -> fields(local_status_disabled) ->
[ [
{"status", {status,
mk( mk(
disabled, disabled,
#{ #{
@ -574,7 +574,7 @@ fields(local_status_disabled) ->
]; ];
fields(local_status_enabled) -> fields(local_status_enabled) ->
[ [
{"status", {status,
mk( mk(
enabled, enabled,
#{ #{
@ -582,7 +582,7 @@ fields(local_status_enabled) ->
required => true required => true
} }
)}, )},
{"process", {process,
mk( mk(
hoconsc:enum([rebalance, evacuation]), hoconsc:enum([rebalance, evacuation]),
#{ #{
@ -590,7 +590,7 @@ fields(local_status_enabled) ->
required => true required => true
} }
)}, )},
{"state", {state,
mk( mk(
atom(), atom(),
#{ #{
@ -598,7 +598,7 @@ fields(local_status_enabled) ->
required => true required => true
} }
)}, )},
{"coordinator_node", {coordinator_node,
mk( mk(
binary(), binary(),
#{ #{
@ -606,7 +606,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"connection_eviction_rate", {connection_eviction_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -614,7 +614,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"session_eviction_rate", {session_eviction_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -622,7 +622,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"connection_goal", {connection_goal,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -630,7 +630,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"session_goal", {session_goal,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -638,7 +638,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"disconnected_session_goal", {disconnected_session_goal,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -646,7 +646,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"session_recipients", {session_recipients,
mk( mk(
list(binary()), list(binary()),
#{ #{
@ -654,7 +654,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"recipients", {recipients,
mk( mk(
list(binary()), list(binary()),
#{ #{
@ -662,7 +662,7 @@ fields(local_status_enabled) ->
required => false required => false
} }
)}, )},
{"stats", {stats,
mk( mk(
ref(status_stats), ref(status_stats),
#{ #{
@ -673,7 +673,7 @@ fields(local_status_enabled) ->
]; ];
fields(status_stats) -> fields(status_stats) ->
[ [
{"initial_connected", {initial_connected,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -681,7 +681,7 @@ fields(status_stats) ->
required => true required => true
} }
)}, )},
{"current_connected", {current_connected,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -689,7 +689,7 @@ fields(status_stats) ->
required => true required => true
} }
)}, )},
{"initial_sessions", {initial_sessions,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -697,7 +697,7 @@ fields(status_stats) ->
required => true required => true
} }
)}, )},
{"current_sessions", {current_sessions,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -705,7 +705,7 @@ fields(status_stats) ->
required => true required => true
} }
)}, )},
{"current_disconnected_sessions", {current_disconnected_sessions,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -716,11 +716,11 @@ fields(status_stats) ->
]; ];
fields(global_coordinator_status) -> fields(global_coordinator_status) ->
without( without(
["status", "process", "session_goal", "session_recipients", "stats"], [status, process, session_goal, session_recipients, stats],
fields(local_status_enabled) fields(local_status_enabled)
) ++ ) ++
[ [
{"donors", {donors,
mk( mk(
list(binary()), list(binary()),
#{ #{
@ -728,7 +728,7 @@ fields(global_coordinator_status) ->
required => false required => false
} }
)}, )},
{"donor_conn_avg", {donor_conn_avg,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -736,7 +736,7 @@ fields(global_coordinator_status) ->
required => false required => false
} }
)}, )},
{"donor_sess_avg", {donor_sess_avg,
mk( mk(
non_neg_integer(), non_neg_integer(),
#{ #{
@ -744,7 +744,7 @@ fields(global_coordinator_status) ->
required => false required => false
} }
)}, )},
{"node", {node,
mk( mk(
binary(), binary(),
#{ #{
@ -754,9 +754,9 @@ fields(global_coordinator_status) ->
)} )}
]; ];
fields(global_evacuation_status) -> fields(global_evacuation_status) ->
without(["status", "process"], fields(local_status_enabled)) ++ without([status, process], fields(local_status_enabled)) ++
[ [
{"node", {node,
mk( mk(
binary(), binary(),
#{ #{
@ -768,19 +768,19 @@ fields(global_evacuation_status) ->
fields(global_purge_status) -> fields(global_purge_status) ->
without( without(
[ [
"status", status,
"process", process,
"connection_eviction_rate", connection_eviction_rate,
"session_eviction_rate", session_eviction_rate,
"connection_goal", connection_goal,
"disconnected_session_goal", disconnected_session_goal,
"session_recipients", session_recipients,
"recipients" recipients
], ],
fields(local_status_enabled) fields(local_status_enabled)
) ++ ) ++
[ [
{"purge_rate", {purge_rate,
mk( mk(
pos_integer(), pos_integer(),
#{ #{
@ -788,7 +788,7 @@ fields(global_purge_status) ->
required => false required => false
} }
)}, )},
{"node", {node,
mk( mk(
binary(), binary(),
#{ #{
@ -799,7 +799,7 @@ fields(global_purge_status) ->
]; ];
fields(global_status) -> fields(global_status) ->
[ [
{"evacuations", {evacuations,
mk( mk(
hoconsc:array(ref(global_evacuation_status)), hoconsc:array(ref(global_evacuation_status)),
#{ #{
@ -807,7 +807,7 @@ fields(global_status) ->
required => false required => false
} }
)}, )},
{"purges", {purges,
mk( mk(
hoconsc:array(ref(global_purge_status)), hoconsc:array(ref(global_purge_status)),
#{ #{
@ -815,7 +815,7 @@ fields(global_status) ->
required => false required => false
} }
)}, )},
{"rebalances", {rebalances,
mk( mk(
hoconsc:array(ref(global_coordinator_status)), hoconsc:array(ref(global_coordinator_status)),
#{ #{

2
apps/emqx_node_rebalance/test/emqx_node_rebalance_SUITE.erl
View File

@ -48,7 +48,7 @@ init_per_testcase(Case, Config) ->
ClusterNodes = start_cluster( ClusterNodes = start_cluster(
Config, Config,
NodeNames, NodeNames,
[emqx, emqx_eviction_agent, emqx_node_rebalance] [emqx, emqx_node_rebalance]
), ),
ok = snabbkaffe:start_trace(), ok = snabbkaffe:start_trace(),
[{cluster_nodes, ClusterNodes} | Config]. [{cluster_nodes, ClusterNodes} | Config].

4
apps/emqx_node_rebalance/test/emqx_node_rebalance_agent_SUITE.erl
View File

@ -38,7 +38,7 @@ groups() ->
]. ].
init_per_suite(Config) -> init_per_suite(Config) ->
Apps = emqx_cth_suite:start([emqx, emqx_eviction_agent, emqx_node_rebalance], #{ Apps = emqx_cth_suite:start([emqx, emqx_node_rebalance], #{
work_dir => ?config(priv_dir, Config) work_dir => ?config(priv_dir, Config)
}), }),
[{apps, Apps} | Config]. [{apps, Apps} | Config].
@ -60,7 +60,7 @@ init_per_testcase(Case, Config) ->
ClusterNodes = emqx_cth_cluster:start( ClusterNodes = emqx_cth_cluster:start(
[ [
{case_specific_node_name(?MODULE, Case), #{ {case_specific_node_name(?MODULE, Case), #{
apps => [emqx, emqx_eviction_agent, emqx_node_rebalance] apps => [emqx, emqx_node_rebalance]
}} }}
], ],
#{work_dir => emqx_cth_suite:work_dir(Case, Config)} #{work_dir => emqx_cth_suite:work_dir(Case, Config)}

3
apps/emqx_node_rebalance/test/emqx_node_rebalance_api_SUITE.erl
View File

@ -29,7 +29,7 @@ all() ->
emqx_common_test_helpers:all(?MODULE). emqx_common_test_helpers:all(?MODULE).
init_per_suite(Config) -> init_per_suite(Config) ->
Apps = emqx_cth_suite:start([emqx, emqx_eviction_agent, emqx_node_rebalance], #{ Apps = emqx_cth_suite:start([emqx, emqx_node_rebalance], #{
work_dir => ?config(priv_dir, Config) work_dir => ?config(priv_dir, Config)
}), }),
[{apps, Apps} | Config]. [{apps, Apps} | Config].
@ -548,7 +548,6 @@ app_specs() ->
#{enable => true} #{enable => true}
} }
}}, }},
emqx_eviction_agent,
emqx_node_rebalance emqx_node_rebalance
]. ].

2
apps/emqx_node_rebalance/test/emqx_node_rebalance_cli_SUITE.erl
View File

@ -15,7 +15,7 @@
[emqtt_connect_many/2, stop_many/1, case_specific_node_name/3] [emqtt_connect_many/2, stop_many/1, case_specific_node_name/3]
). ).
-define(START_APPS, [emqx, emqx_eviction_agent, emqx_node_rebalance]). -define(START_APPS, [emqx, emqx_node_rebalance]).
all() -> all() ->
emqx_common_test_helpers:all(?MODULE). emqx_common_test_helpers:all(?MODULE).

2
apps/emqx_node_rebalance/test/emqx_node_rebalance_evacuation_SUITE.erl
View File

@ -70,7 +70,7 @@ init_per_testcase(Case, Config) ->
case_specific_node_name(?MODULE, Case, '_recipient') case_specific_node_name(?MODULE, Case, '_recipient')
] ]
end, end,
ClusterNodes = start_cluster(Config, NodeNames, [emqx, emqx_eviction_agent, emqx_node_rebalance]), ClusterNodes = start_cluster(Config, NodeNames, [emqx, emqx_node_rebalance]),
ok = snabbkaffe:start_trace(), ok = snabbkaffe:start_trace(),
[{cluster_nodes, ClusterNodes} | Config]. [{cluster_nodes, ClusterNodes} | Config].

1
apps/emqx_node_rebalance/test/emqx_node_rebalance_purge_SUITE.erl
View File

@ -117,7 +117,6 @@ app_specs() ->
config => config =>
#{delayed => #{enable => true}} #{delayed => #{enable => true}}
}}, }},
emqx_eviction_agent,
emqx_node_rebalance emqx_node_rebalance
]. ].

1
apps/emqx_node_rebalance/test/emqx_node_rebalance_status_SUITE.erl
View File

@ -32,7 +32,6 @@ init_per_suite(Config) ->
Apps = [ Apps = [
emqx_conf, emqx_conf,
emqx, emqx,
emqx_eviction_agent,
emqx_node_rebalance emqx_node_rebalance
], ],
Cluster = [ Cluster = [

1
changes/ee/fix-12871.en.md Normal file
View File

@ -0,0 +1 @@
Fix startup process of evacuated node. Previously, if a node was evacuated and stoped without stopping evacuation, it would not start back.

4
mix.exs
View File

@ -332,7 +332,9 @@ defmodule EMQXUmbrella.MixProject do
:emqx_s3, :emqx_s3,
:emqx_opentelemetry, :emqx_opentelemetry,
:emqx_durable_storage, :emqx_durable_storage,
:rabbit_common :rabbit_common,
:emqx_eviction_agent,
:emqx_node_rebalance
], ],
steps: steps, steps: steps,
strip_beams: false strip_beams: false

2
rebar.config.erl
View File

@ -116,6 +116,8 @@ is_community_umbrella_app("apps/emqx_gateway_ocpp") -> false;
is_community_umbrella_app("apps/emqx_gateway_jt808") -> false; is_community_umbrella_app("apps/emqx_gateway_jt808") -> false;
is_community_umbrella_app("apps/emqx_bridge_syskeeper") -> false; is_community_umbrella_app("apps/emqx_bridge_syskeeper") -> false;
is_community_umbrella_app("apps/emqx_message_validation") -> false; is_community_umbrella_app("apps/emqx_message_validation") -> false;
is_community_umbrella_app("apps/emqx_eviction_agent") -> false;
is_community_umbrella_app("apps/emqx_node_rebalance") -> false;
is_community_umbrella_app(_) -> true. is_community_umbrella_app(_) -> true.
%% BUILD_WITHOUT_JQ %% BUILD_WITHOUT_JQ