emqx / ekka / 823

Committed 28 Mar 2024 09:46AM UTC coverage: 61.618% (-1.2%) from 62.78%

Build # 823

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #233 from SergeTupchiy/EMQX-11826-prevent-left-node-from-rejoining

EMQX-11826 prevent left node from rejoining

Run Details

5 of 15 new or added lines in 2 files covered. (33.33%)

2 existing lines in 1 file now uncovered.

419 of 680 relevant lines covered (61.62%)

50.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.47

/src/ekka.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2019-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.
%%--------------------------------------------------------------------

-module(ekka).

-include_lib("mria/include/mria.hrl").
-include_lib("snabbkaffe/include/snabbkaffe.hrl").

%% Start/Stop
-export([start/0, stop/0]).

%% Env
-export([env/1, env/2]).

%% Info
-export([info/0, info/1]).

%% Cluster API
-export([cluster_name/0]).
-export([ join/1
        , leave/0
        , force_leave/1
        ]).

%% Autocluster API
-export([ autocluster/0
        , autocluster/1
        , autocluster_enabled/0
        , enable_autocluster/0
        , disable_autocluster/0
        ]).

%% Callbacks
-export([ exec_callback/1
        , callback/2
        ]).

%% Node API
-export([ is_aliving/1
        , is_running/2
        ]).

%% Membership API
-export([ members/0
        , is_member/1
        , local_member/0
        , nodelist/0
        , nodelist/1
        ]).

%% Membership Monitor API
-export([ monitor/1
        , monitor/2
        , unmonitor/1
        , unmonitor/2
        ]).

%% Locker API
-export([ lock/1
        , lock/2
        , lock/3
        , unlock/1
        , unlock/2
        ]).

-define(IS_MON_TYPE(T), T == membership orelse T == partition).

-type(info_key() :: members | running_nodes | stopped_nodes | partitions).

-type(infos() :: #{members       := list(member()),
                   running_nodes := list(node()),
                   stopped_nodes := list(node()),
                   partitions    := list(node())
                  }).

-export_type([info_key/0, infos/0]).

%%--------------------------------------------------------------------
%% Start/Stop
%%--------------------------------------------------------------------

-spec(start() -> ok).
start() ->
    ok = mria:start(),
    ?tp(info, "Starting ekka", #{}),
    ekka_boot:register_mria_callbacks(),
    {ok, _Apps} = application:ensure_all_started(ekka),
    ?tp(info, "Ekka is running", #{}),
    %% NOTE
    %% Ekka doesn't run a separate "mnesia boot" phase anymore. Users are
    %% advised to manage tables explicitly by themselves as part of
    %% applications' startup routines.
    %% See also: `ekka_boot:create_tables/0`
    ekka:exec_callback(start),
    ok.

-spec(stop() -> ok).
stop() ->
    ekka:exec_callback(stop),
    application:stop(ekka).

%%--------------------------------------------------------------------
%% Env
%%--------------------------------------------------------------------

-spec(env(atom() | {callback, atom()}) -> undefined | {ok, term()}).
env(Key) ->
    %% TODO: hack, using apply to trick dialyzer.
    apply(application, get_env, [ekka, Key]).

-spec(env(atom() | {callback, atom()}, term()) -> term()).
env(Key, Default) ->
    application:get_env(ekka, Key, Default).

%%--------------------------------------------------------------------
%% Info
%%--------------------------------------------------------------------

-spec(info(info_key()) -> term()).
info(Key) ->
    maps:get(Key, info()).

-spec(info() -> infos()).
info() ->
    ClusterInfo = ekka_cluster:info(),
    Partitions = mria_node_monitor:partitions(),
    maps:merge(ClusterInfo, #{members    => members(),
                              partitions => Partitions
                             }).

%%--------------------------------------------------------------------
%% Cluster API
%%--------------------------------------------------------------------

-spec(cluster_name() -> cluster()).
cluster_name() -> env(cluster_name, undefined).

%% @doc Join the cluster
-spec(join(node()) -> ok | ignore | {error, term()}).
join(Node) -> ekka_cluster:join(Node).

%% @doc Leave from Cluster.
-spec(leave() -> ok | {error, term()}).
leave() -> ekka_cluster:leave().

%% @doc Force a node leave from cluster.
-spec(force_leave(node()) -> ok | ignore | {error, term()}).
force_leave(Node) -> ekka_cluster:force_leave(Node).

%%--------------------------------------------------------------------
%% Autocluster
%%--------------------------------------------------------------------

autocluster() -> autocluster(ekka).

autocluster(App) ->
    case env(cluster_enable, true) andalso ekka_autocluster:enabled() of
        true  -> ekka_autocluster:run(App);
        false -> ok
    end.

autocluster_enabled() ->
    env(cluster_enable, true) andalso ekka_autocluster:configured().

disable_autocluster() ->
    ok = application:set_env(ekka, cluster_enable, false),
    ok = mria:disable_core_node_discovery(),
    ok = ekka_autocluster:stop(disabled).

enable_autocluster() ->
    ok = application:set_env(ekka, cluster_enable, true),
    ok = mria:enable_core_node_discovery().

%%--------------------------------------------------------------------
%% Callbacks
%%--------------------------------------------------------------------
-spec exec_callback(atom()) -> term().
exec_callback(Name) ->
    case env({callback, Name}) of
      {ok, Fun} -> Fun();
      undefined -> ok
    end.

-spec(callback(atom(), function()) -> ok).
callback(Name, Fun) ->
    %% TODO: hack, using apply to trick dialyzer.
    %% Using a tuple as a key of the application environment "works", but it violates the spec
    apply(application, set_env, [ekka, {callback, Name}, Fun]).

%%--------------------------------------------------------------------
%% Node API
%%--------------------------------------------------------------------

%% @doc Is node aliving?
-spec(is_aliving(node()) -> boolean()).
is_aliving(Node) ->
    ekka_node:is_aliving(Node).

%% @doc Is the application running?
-spec(is_running(node(), atom()) -> boolean()).
is_running(Node, App) ->
    ekka_node:is_running(Node, App).

%%--------------------------------------------------------------------
%% Membership API
%%--------------------------------------------------------------------

%% Cluster members
-spec(members() -> list(member())).
members() -> mria_membership:members().

%% Local member
-spec(local_member() -> member()).
local_member() -> mria_membership:local_member().

%% Is node a member?
-spec(is_member(node()) -> boolean()).
is_member(Node) -> mria_membership:is_member(Node).

%% Node list
-spec(nodelist() -> list(node())).
nodelist() -> mria_membership:nodelist().

-spec(nodelist(up|down) -> list(node())).
nodelist(Status) -> mria_membership:nodelist(Status).

%%--------------------------------------------------------------------
%% Membership Monitor API
%%--------------------------------------------------------------------

monitor(Type) when ?IS_MON_TYPE(Type) ->
    mria_membership:monitor(Type, self(), true).

monitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
    mria_membership:monitor(Type, Fun, true).

unmonitor(Type) when ?IS_MON_TYPE(Type) ->
    mria_membership:monitor(Type, self(), false).

unmonitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
    mria_membership:monitor(Type, Fun, false).

%%--------------------------------------------------------------------
%% Locker API
%%--------------------------------------------------------------------

-spec(lock(ekka_locker:resource()) -> ekka_locker:lock_result()).
lock(Resource) ->
    ekka_locker:acquire(Resource).

-spec(lock(ekka_locker:resource(), ekka_locker:lock_type())
      -> ekka_locker:lock_result()).
lock(Resource, Type) ->
    ekka_locker:acquire(ekka_locker, Resource, Type).

-spec(lock(ekka_locker:resource(), ekka_locker:lock_type(), ekka_locker:piggyback())
      -> ekka_locker:lock_result()).
lock(Resource, Type, Piggyback) ->
    ekka_locker:acquire(ekka_locker, Resource, Type, Piggyback).

-spec(unlock(ekka_locker:resource()) -> ekka_locker:lock_result()).
unlock(Resource) ->
    ekka_locker:release(Resource).

-spec(unlock(ekka_locker:resource(), ekka_locker:lock_type()) -> ekka_locker:lock_result()).
unlock(Resource, Type) ->
    ekka_locker:release(ekka_locker, Resource, Type).

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2019-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16
17	-module(ekka).
18
19	-include_lib("mria/include/mria.hrl").
20	-include_lib("snabbkaffe/include/snabbkaffe.hrl").
21
22	%% Start/Stop
23	-export([start/0, stop/0]).
24
25	%% Env
26	-export([env/1, env/2]).
27
28	%% Info
29	-export([info/0, info/1]).
30
31	%% Cluster API
32	-export([cluster_name/0]).
33	-export([ join/1
34	, leave/0
35	, force_leave/1
36	]).
37
38	%% Autocluster API
39	-export([ autocluster/0
40	, autocluster/1
41	, autocluster_enabled/0
42	, enable_autocluster/0
43	, disable_autocluster/0
44	]).
45
46	%% Callbacks
47	-export([ exec_callback/1
48	, callback/2
49	]).
50
51	%% Node API
52	-export([ is_aliving/1
53	, is_running/2
54	]).
55
56	%% Membership API
57	-export([ members/0
58	, is_member/1
59	, local_member/0
60	, nodelist/0
61	, nodelist/1
62	]).
63
64	%% Membership Monitor API
65	-export([ monitor/1
66	, monitor/2
67	, unmonitor/1
68	, unmonitor/2
69	]).
70
71	%% Locker API
72	-export([ lock/1
73	, lock/2
74	, lock/3
75	, unlock/1
76	, unlock/2
77	]).
78
79	-define(IS_MON_TYPE(T), T == membership orelse T == partition).
80
81	-type(info_key() :: members \| running_nodes \| stopped_nodes \| partitions).
82
83	-type(infos() :: #{members := list(member()),
84	running_nodes := list(node()),
85	stopped_nodes := list(node()),
86	partitions := list(node())
87	}).
88
89	-export_type([info_key/0, infos/0]).
90
91	%%--------------------------------------------------------------------
92	%% Start/Stop
93	%%--------------------------------------------------------------------
94
95	-spec(start() -> ok).
96	start() ->
97	ok = mria:start(),	106✔
98	?tp(info, "Starting ekka", #{}),	106✔
99	ekka_boot:register_mria_callbacks(),	105✔
100	{ok, _Apps} = application:ensure_all_started(ekka),	105✔
101	?tp(info, "Ekka is running", #{}),	103✔
102	%% NOTE
103	%% Ekka doesn't run a separate "mnesia boot" phase anymore. Users are
104	%% advised to manage tables explicitly by themselves as part of
105	%% applications' startup routines.
106	%% See also: `ekka_boot:create_tables/0`
107	ekka:exec_callback(start),	102✔
108	ok.	102✔
109
110	-spec(stop() -> ok).
111	stop() ->
112	ekka:exec_callback(stop),	64✔
113	application:stop(ekka).	64✔
114
115	%%--------------------------------------------------------------------
116	%% Env
117	%%--------------------------------------------------------------------
118
119	-spec(env(atom() \| {callback, atom()}) -> undefined \| {ok, term()}).
120	env(Key) ->
121	%% TODO: hack, using apply to trick dialyzer.
122	apply(application, get_env, [ekka, Key]).	313✔
123
124	-spec(env(atom() \| {callback, atom()}, term()) -> term()).
125	env(Key, Default) ->
126	application:get_env(ekka, Key, Default).	981✔
127
128	%%--------------------------------------------------------------------
129	%% Info
130	%%--------------------------------------------------------------------
131
132	-spec(info(info_key()) -> term()).
133	info(Key) ->
134	maps:get(Key, info()).	37✔
135
136	-spec(info() -> infos()).
137	info() ->
138	ClusterInfo = ekka_cluster:info(),	37✔
139	Partitions = mria_node_monitor:partitions(),	37✔
140	maps:merge(ClusterInfo, #{members => members(),	37✔
141	partitions => Partitions
142	}).
143
144	%%--------------------------------------------------------------------
145	%% Cluster API
146	%%--------------------------------------------------------------------
147
148	-spec(cluster_name() -> cluster()).
149	cluster_name() -> env(cluster_name, undefined).	1✔
150
151	%% @doc Join the cluster
152	-spec(join(node()) -> ok \| ignore \| {error, term()}).
153	join(Node) -> ekka_cluster:join(Node).	8✔
154
155	%% @doc Leave from Cluster.
156	-spec(leave() -> ok \| {error, term()}).
157	leave() -> ekka_cluster:leave().	5✔
158
159	%% @doc Force a node leave from cluster.
160	-spec(force_leave(node()) -> ok \| ignore \| {error, term()}).
161	force_leave(Node) -> ekka_cluster:force_leave(Node).	15✔
162
163	%%--------------------------------------------------------------------
164	%% Autocluster
165	%%--------------------------------------------------------------------
166
167	autocluster() -> autocluster(ekka).	15✔
168
169	autocluster(App) ->
170	case env(cluster_enable, true) andalso ekka_autocluster:enabled() of	15✔
171	true -> ekka_autocluster:run(App);	13✔
172	false -> ok	2✔
173	end.
174
175	autocluster_enabled() ->
NEW 176	env(cluster_enable, true) andalso ekka_autocluster:configured().	×
177
178	disable_autocluster() ->
NEW 179	ok = application:set_env(ekka, cluster_enable, false),	×
NEW 180	ok = mria:disable_core_node_discovery(),	×
NEW 181	ok = ekka_autocluster:stop(disabled).	×
182
183	enable_autocluster() ->
NEW 184	ok = application:set_env(ekka, cluster_enable, true),	×
NEW 185	ok = mria:enable_core_node_discovery().	×
186
187	%%--------------------------------------------------------------------
188	%% Callbacks
189	%%--------------------------------------------------------------------
190	-spec exec_callback(atom()) -> term().
191	exec_callback(Name) ->
192	case env({callback, Name}) of	166✔
193	{ok, Fun} -> Fun();	×
194	undefined -> ok	166✔
195	end.
196
197	-spec(callback(atom(), function()) -> ok).
198	callback(Name, Fun) ->
199	%% TODO: hack, using apply to trick dialyzer.
200	%% Using a tuple as a key of the application environment "works", but it violates the spec
201	apply(application, set_env, [ekka, {callback, Name}, Fun]).	×
202
203	%%--------------------------------------------------------------------
204	%% Node API
205	%%--------------------------------------------------------------------
206
207	%% @doc Is node aliving?
208	-spec(is_aliving(node()) -> boolean()).
209	is_aliving(Node) ->
210	ekka_node:is_aliving(Node).	2✔
211
212	%% @doc Is the application running?
213	-spec(is_running(node(), atom()) -> boolean()).
214	is_running(Node, App) ->
215	ekka_node:is_running(Node, App).	26✔
216
217	%%--------------------------------------------------------------------
218	%% Membership API
219	%%--------------------------------------------------------------------
220
221	%% Cluster members
222	-spec(members() -> list(member())).
223	members() -> mria_membership:members().	37✔
224
225	%% Local member
226	-spec(local_member() -> member()).
227	local_member() -> mria_membership:local_member().	1✔
228
229	%% Is node a member?
230	-spec(is_member(node()) -> boolean()).
231	is_member(Node) -> mria_membership:is_member(Node).	2✔
232
233	%% Node list
234	-spec(nodelist() -> list(node())).
235	nodelist() -> mria_membership:nodelist().	1✔
236
237	-spec(nodelist(up\|down) -> list(node())).
238	nodelist(Status) -> mria_membership:nodelist(Status).	1✔
239
240	%%--------------------------------------------------------------------
241	%% Membership Monitor API
242	%%--------------------------------------------------------------------
243
244	monitor(Type) when ?IS_MON_TYPE(Type) ->
245	mria_membership:monitor(Type, self(), true).	×
246
247	monitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
248	mria_membership:monitor(Type, Fun, true).	×
249
250	unmonitor(Type) when ?IS_MON_TYPE(Type) ->
251	mria_membership:monitor(Type, self(), false).	×
252
253	unmonitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
254	mria_membership:monitor(Type, Fun, false).	×
255
256	%%--------------------------------------------------------------------
257	%% Locker API
258	%%--------------------------------------------------------------------
259
260	-spec(lock(ekka_locker:resource()) -> ekka_locker:lock_result()).
261	lock(Resource) ->
262	ekka_locker:acquire(Resource).	1✔
263
264	-spec(lock(ekka_locker:resource(), ekka_locker:lock_type())
265	-> ekka_locker:lock_result()).
266	lock(Resource, Type) ->
267	ekka_locker:acquire(ekka_locker, Resource, Type).	2✔
268
269	-spec(lock(ekka_locker:resource(), ekka_locker:lock_type(), ekka_locker:piggyback())
270	-> ekka_locker:lock_result()).
271	lock(Resource, Type, Piggyback) ->
272	ekka_locker:acquire(ekka_locker, Resource, Type, Piggyback).	×
273
274	-spec(unlock(ekka_locker:resource()) -> ekka_locker:lock_result()).
275	unlock(Resource) ->
276	ekka_locker:release(Resource).	1✔
277
278	-spec(unlock(ekka_locker:resource(), ekka_locker:lock_type()) -> ekka_locker:lock_result()).
279	unlock(Resource, Type) ->
280	ekka_locker:release(ekka_locker, Resource, Type).	2✔

emqx / ekka / 823

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous