emqx / ekka / 850

Committed 18 Dec 2024 02:02PM UTC coverage: 62.397% (-0.5%) from 62.861%

Build # 850

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #240 from keynslug/fix/EEC-112/autoheal-asymm

fix(autoheal): attempt healing complex asymmetric partitions

Run Details

2 of 49 new or added lines in 2 files covered. (4.08%)

94 existing lines in 8 files now uncovered.

682 of 1093 relevant lines covered (62.4%)

49.22 hits per line

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

77.5

/src/ekka.erl

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

%% Register callback
-export([ 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() ->
    case ekka_mnesia:start() of
        ok -> ok;
        {error, {timeout, Tables}} ->
            logger:error("Mnesia wait_for_tables timeout: ~p", [Tables]),
            ok;
        {error, Reason} ->
            error(Reason)
    end,
    {ok, _Apps} = application:ensure_all_started(ekka),
    ok.

-spec(stop() -> ok).
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 = ekka_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.

%%--------------------------------------------------------------------
%% Register callback
%%--------------------------------------------------------------------

-spec callback(atom()) -> undefined | {ok, function()}.
callback(Name) ->
    env({callback, Name}).

-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() -> ekka_membership:members().

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

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

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

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

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

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

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

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

unmonitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
    ekka_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 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("ekka.hrl").
20
21	%% Start/Stop
22	-export([start/0, stop/0]).
23
24	%% Env
25	-export([env/1, env/2]).
26
27	%% Info
28	-export([info/0, info/1]).
29
30	%% Cluster API
31	-export([cluster_name/0]).
32	-export([ join/1
33	, leave/0
34	, force_leave/1
35	]).
36
37	%% Autocluster API
38	-export([ autocluster/0
39	, autocluster/1
40	]).
41
42	%% Register callback
43	-export([ callback/1
44	, callback/2
45	]).
46
47	%% Node API
48	-export([ is_aliving/1
49	, is_running/2
50	]).
51
52	%% Membership API
53	-export([ members/0
54	, is_member/1
55	, local_member/0
56	, nodelist/0
57	, nodelist/1
58	]).
59
60	%% Membership Monitor API
61	-export([ monitor/1
62	, monitor/2
63	, unmonitor/1
64	, unmonitor/2
65	]).
66
67	%% Locker API
68	-export([ lock/1
69	, lock/2
70	, lock/3
71	, unlock/1
72	, unlock/2
73	]).
74
75	-define(IS_MON_TYPE(T), T == membership orelse T == partition).
76
77	-type(info_key() :: members \| running_nodes \| stopped_nodes \| partitions).
78
79	-type(infos() :: #{members := list(member()),
80	running_nodes := list(node()),
81	stopped_nodes := list(node()),
82	partitions := list(node())
83	}).
84
85	-export_type([info_key/0, infos/0]).
86
87	%%--------------------------------------------------------------------
88	%% Start/Stop
89	%%--------------------------------------------------------------------
90
91	-spec(start() -> ok).
92	start() ->
93	case ekka_mnesia:start() of	16✔
94	ok -> ok;	16✔
95	{error, {timeout, Tables}} ->
UNCOV 96	logger:error("Mnesia wait_for_tables timeout: ~p", [Tables]),	×
UNCOV 97	ok;	×
98	{error, Reason} ->
UNCOV 99	error(Reason)	×
100	end,
101	{ok, _Apps} = application:ensure_all_started(ekka),	16✔
102	ok.	16✔
103
104	-spec(stop() -> ok).
105	stop() ->
106	application:stop(ekka).	4✔
107
108	%%--------------------------------------------------------------------
109	%% Env
110	%%--------------------------------------------------------------------
111
112	-spec(env(atom() \| {callback, atom()}) -> undefined \| {ok, term()}).
113	env(Key) ->
114	%% TODO: hack, using apply to trick dialyzer.
115	apply(application, get_env, [ekka, Key]).	36✔
116
117	-spec(env(atom() \| {callback, atom()}, term()) -> term()).
118	env(Key, Default) ->
119	application:get_env(ekka, Key, Default).	480✔
120
121	%%--------------------------------------------------------------------
122	%% Info
123	%%--------------------------------------------------------------------
124
125	-spec(info(info_key()) -> term()).
126	info(Key) ->
127	maps:get(Key, info()).	38✔
128
129	-spec(info() -> infos()).
130	info() ->
131	ClusterInfo = ekka_cluster:info(),	38✔
132	Partitions = ekka_node_monitor:partitions(),	38✔
133	maps:merge(ClusterInfo, #{members => members(),	38✔
134	partitions => Partitions
135	}).
136
137	%%--------------------------------------------------------------------
138	%% Cluster API
139	%%--------------------------------------------------------------------
140
141	-spec(cluster_name() -> cluster()).
142	cluster_name() -> env(cluster_name, undefined).	1✔
143
144	%% @doc Join the cluster
145	-spec(join(node()) -> ok \| ignore \| {error, term()}).
146	join(Node) -> ekka_cluster:join(Node).	2✔
147
148	%% @doc Leave from Cluster.
149	-spec(leave() -> ok \| {error, term()}).
150	leave() -> ekka_cluster:leave().	2✔
151
152	%% @doc Force a node leave from cluster.
153	-spec(force_leave(node()) -> ok \| ignore \| {error, term()}).
154	force_leave(Node) -> ekka_cluster:force_leave(Node).	13✔
155
156	%%--------------------------------------------------------------------
157	%% Autocluster
158	%%--------------------------------------------------------------------
159
160	autocluster() -> autocluster(ekka).	2✔
161
162	autocluster(App) ->
163	case env(cluster_enable, true) andalso ekka_autocluster:enabled() of	2✔
164	true -> ekka_autocluster:run(App);	1✔
165	false -> ok	1✔
166	end.
167
168	%%--------------------------------------------------------------------
169	%% Register callback
170	%%--------------------------------------------------------------------
171
172	-spec callback(atom()) -> undefined \| {ok, function()}.
173	callback(Name) ->
174	env({callback, Name}).	5✔
175
176	-spec(callback(atom(), function()) -> ok).
177	callback(Name, Fun) ->
178	%% TODO: hack, using apply to trick dialyzer.
179	%% Using a tuple as a key of the application environment "works", but it violates the spec
180	apply(application, set_env, [ekka, {callback, Name}, Fun]).	×
181
182	%%--------------------------------------------------------------------
183	%% Node API
184	%%--------------------------------------------------------------------
185
186	%% @doc Is node aliving?
187	-spec(is_aliving(node()) -> boolean()).
188	is_aliving(Node) ->
189	ekka_node:is_aliving(Node).	2✔
190
191	%% @doc Is the application running?
192	-spec(is_running(node(), atom()) -> boolean()).
193	is_running(Node, App) ->
194	ekka_node:is_running(Node, App).	21✔
195
196	%%--------------------------------------------------------------------
197	%% Membership API
198	%%--------------------------------------------------------------------
199
200	%% Cluster members
201	-spec(members() -> list(member())).
202	members() -> ekka_membership:members().	38✔
203
204	%% Local member
205	-spec(local_member() -> member()).
206	local_member() -> ekka_membership:local_member().	1✔
207
208	%% Is node a member?
209	-spec(is_member(node()) -> boolean()).
210	is_member(Node) -> ekka_membership:is_member(Node).	2✔
211
212	%% Node list
213	-spec(nodelist() -> list(node())).
214	nodelist() -> ekka_membership:nodelist().	1✔
215
216	-spec(nodelist(up\|down) -> list(node())).
217	nodelist(Status) -> ekka_membership:nodelist(Status).	1✔
218
219	%%--------------------------------------------------------------------
220	%% Membership Monitor API
221	%%--------------------------------------------------------------------
222
223	monitor(Type) when ?IS_MON_TYPE(Type) ->
UNCOV 224	ekka_membership:monitor(Type, self(), true).	×
225
226	monitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
UNCOV 227	ekka_membership:monitor(Type, Fun, true).	×
228
229	unmonitor(Type) when ?IS_MON_TYPE(Type) ->
UNCOV 230	ekka_membership:monitor(Type, self(), false).	×
231
232	unmonitor(Type, Fun) when is_function(Fun), ?IS_MON_TYPE(Type) ->
UNCOV 233	ekka_membership:monitor(Type, Fun, false).	×
234
235	%%--------------------------------------------------------------------
236	%% Locker API
237	%%--------------------------------------------------------------------
238
239	-spec(lock(ekka_locker:resource()) -> ekka_locker:lock_result()).
240	lock(Resource) ->
241	ekka_locker:acquire(Resource).	1✔
242
243	-spec(lock(ekka_locker:resource(), ekka_locker:lock_type())
244	-> ekka_locker:lock_result()).
245	lock(Resource, Type) ->
246	ekka_locker:acquire(ekka_locker, Resource, Type).	2✔
247
248	-spec(lock(ekka_locker:resource(), ekka_locker:lock_type(), ekka_locker:piggyback())
249	-> ekka_locker:lock_result()).
250	lock(Resource, Type, Piggyback) ->
251	ekka_locker:acquire(ekka_locker, Resource, Type, Piggyback).	×
252
253	-spec(unlock(ekka_locker:resource()) -> ekka_locker:lock_result()).
254	unlock(Resource) ->
255	ekka_locker:release(Resource).	1✔
256
257	-spec(unlock(ekka_locker:resource(), ekka_locker:lock_type()) -> ekka_locker:lock_result()).
258	unlock(Resource, Type) ->
259	ekka_locker:release(ekka_locker, Resource, Type).	2✔

emqx / ekka / 850

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