diff --git a/src/classy.erl b/src/classy.erl index e485c37..bb3d5bb 100644 --- a/src/classy.erl +++ b/src/classy.erl @@ -18,6 +18,7 @@ This MFA can contain calls to various @code{classy:on_...} functions. , info/2 , n_restarts/0 , n_restarts/1 + , node_to_site/0 , node_of_site/2 , join_node/2 , kick_site/2 @@ -48,7 +49,8 @@ This MFA can contain calls to various @code{classy:on_...} functions. , pre_join/2 , post_join/2 , pre_kick/2 - , post_kick/2 + , on_kick_decided/2 + , on_leave/2 , pre_autoclean/2 , pre_autocluster/2 , run_level/2 @@ -127,9 +129,9 @@ When join is triggered by autocluster. -type join_intent() :: term(). -doc """ -Kick intent is an arbitrary term passed to @ref{classy:pre_kick/2} and @ref{classy:post_kick/2} hooks. -@code{pre_kick} may match on intent to prevent node from leaving the cluster in certain cases, -while to @code{post_kick} this value is merely informational. +Kick intent is an arbitrary term passed to @link{classy:pre_kick/2}, @link{classy:on_kick_decided/2} and @link{classy:on_leave/2} hooks. +@code{pre_kick} may use intent to make a decision leaving the cluster in certain cases, +while to @code{on_leave} this value is merely informational. Classy itself uses the following intents: @itemize @@ -270,6 +272,20 @@ as values returned by this function may be out-of-date. n_restarts(Site) -> classy_node:n_restarts(Site). +-doc """ +Return a mapping from known node names to site IDs. +""". +-spec node_to_site() -> {ok, #{node() => site()}} | {error, not_in_cluster}. +node_to_site() -> + maybe + {ok, _TheSite} ?= the_site(), + {ok, _TheCluster} ?= the_cluster(), + {ok, classy_node:node_to_site()} + else + undefined -> + {error, not_in_cluster} + end. + -doc """ Locate a node that is currently hosting a site. @@ -346,13 +362,13 @@ with @code{Intent} equal to the value of the argument: @item @ref{classy:pre_kick/2}. It can decide that removing a site is unsafe and abort the command. -@item @ref{classy:post_kick/2}. +@item @ref{classy:on_leave/2}. This hook is executed after the target is successfully kicked. @end enumerate +NOTE: the intent is not propagated across different sites. If the target site is not the same as the local site, -then it also runs @ref{classy:post_kick/2} with pre-defined intent @code{kicked}. - +then the target runs @ref{classy:on_leave/2} with pre-defined intent @code{kicked}. """. -spec kick_site(site(), kick_intent()) -> ok | {error, _}. kick_site(Site, Intent) -> @@ -659,25 +675,49 @@ This hook runs on the node that initiates the kick. WARNING: this hook cannot have side effects. """. -spec pre_kick( - fun((cluster_id(), Remote, kick_intent()) -> ok | {error, _}), + fun((cluster_id(), Target, kick_intent()) -> ok | {error, _}), classy_hook:prio() ) -> classy_hook:hook() - when Remote :: site(). + when Target :: site(). pre_kick(Hook, Prio) -> classy_hook:insert(?on_pre_kick, Hook, Prio). +-doc """ +Register a hook that is executed after @ref{classy:pre_kick/2} hooks allow the kick to proceed, +but before the membership change is applied and before the site left the cluster. + +This hook is executed only on the node that initiates the kick procedure. +It can be used to trigger some side effects while the site is still part of the cluster. + +WARNING: theoretically, kick procedure can be aborted after this hook fires, +or in the middle of its execution. +In this case kick procedure @emph{won't} be retried. +Then side effects of this hook will be observed, +but the site will stay in the cluster. +As such, it's not recommended to perform any destructive actions here. + +Normally, such actions should be performed in @link{classy:on_membership_change/2}. +""". +-spec on_kick_decided( + fun((cluster_id(), Target, kick_intent()) -> _), + classy_hook:prio() + ) -> classy_hook:hook() + when Target :: site(). +on_kick_decided(Hook, Prio) -> + classy_hook:insert(?on_kick_decided, Hook, Prio). + -doc """ Register a hook that is executed after the local site leaves a cluster. This hook can perform destructive actions associated with cleanup. """. --spec post_kick( +-spec on_leave( fun((OldCluster, Local, kick_intent()) -> _), classy_hook:prio() ) -> classy_hook:hook() when OldCluster :: cluster_id(), Local :: site(). -post_kick(Hook, Prio) -> - classy_hook:insert(?on_post_kick, Hook, Prio). +on_leave(Hook, Prio) -> + classy_hook:insert(?on_leave, Hook, Prio). -doc """ Register a hook that runs before autoclean finalizes the decision to kick a down site. diff --git a/src/classy_builtin_hooks.erl b/src/classy_builtin_hooks.erl index 35a30dc..67cf054 100644 --- a/src/classy_builtin_hooks.erl +++ b/src/classy_builtin_hooks.erl @@ -5,7 +5,7 @@ %% API: -export([ gen_random_site_id/0 - , maybe_reinitialize_after_kick/3 + , maybe_reinitialize_after_leave/3 , log_create_site/1 , log_create_cluster/2 , log_pre_join/4 @@ -34,7 +34,7 @@ gen_random_site_id() -> classy_node:maybe_init_the_site(undefined). %% @doc Create a new cluster after getting kicked. -maybe_reinitialize_after_kick(OldCluster, Local, Intent) -> +maybe_reinitialize_after_leave(OldCluster, Local, Intent) -> ?tp(info, classy_kicked_from_cluster, #{ old_cluster => OldCluster , local => Local diff --git a/src/classy_hook.erl b/src/classy_hook.erl index e0d9ca3..ce00916 100644 --- a/src/classy_hook.erl +++ b/src/classy_hook.erl @@ -63,7 +63,7 @@ init() -> ets:new(?tab, [named_table, ordered_set, public, {keypos, 1}]), %% Default initialization: classy:on_node_init(fun classy_builtin_hooks:gen_random_site_id/0, ?min_hook_prio), - classy:post_kick(fun classy_builtin_hooks:maybe_reinitialize_after_kick/3, ?min_hook_prio), + classy:on_leave(fun classy_builtin_hooks:maybe_reinitialize_after_leave/3, ?min_hook_prio), %% Liveness tracking: classy:run_level(fun classy_liveness:on_run_level/2, ?max_hook_prio), classy:on_peer_connection_change(fun classy_liveness:on_peer_connection_change/3, ?max_hook_prio), diff --git a/src/classy_internal.hrl b/src/classy_internal.hrl index 3844583..22248ba 100644 --- a/src/classy_internal.hrl +++ b/src/classy_internal.hrl @@ -33,7 +33,8 @@ -define(on_pre_join, on_pre_join). -define(on_post_join, on_post_join). -define(on_pre_kick, on_pre_kick). --define(on_post_kick, on_post_kick). +-define(on_kick_decided, on_kick_decided). +-define(on_leave, on_leave). -define(on_change_run_level, on_change_run_level). -define(on_pre_autoclean, on_pre_autoclean). -define(on_pre_autocluster, on_pre_autocluster). diff --git a/src/classy_membership.erl b/src/classy_membership.erl index c7d8ffc..1c816e8 100644 --- a/src/classy_membership.erl +++ b/src/classy_membership.erl @@ -258,7 +258,7 @@ members(Cluster, Local) -> , [] , ['$1'] }, - ets:select(?ptab, [MS]). + classy_table:select(?ptab, [MS]). -doc "List local sites and clusters.". -spec list_local_sites(running | all) -> [{classy:cluster_id(), classy:site()}]. @@ -273,7 +273,7 @@ list_local_sites(all) -> , [] , [{{'$1', '$2'}}] }, - ets:select(?ptab, [MS]). + classy_table:select(?ptab, [MS]). -doc """ Return mapping of nodes to sites. @@ -833,7 +833,7 @@ memtab_since(Since, #s{cluster = Cluster, site = Local}) -> , [{'>=', '$2', Since}] , ['$1'] }, - ets:select(?ptab, [MS]). + classy_table:select(?ptab, [MS]). -spec peers(#s{}) -> [classy:site()]. peers(#s{cluster = Cluster, site = Local}) -> @@ -843,7 +843,7 @@ peers(#s{cluster = Cluster, site = Local}) -> , [] , ['$1'] }, - ets:select(?ptab, [MS]). + classy_table:select(?ptab, [MS]). -spec nodes_of_cluster(#s{}) -> #{classy:site() => node()}. nodes_of_cluster(#s{cluster = Cluster, site = Local}) -> @@ -857,7 +857,7 @@ select_nodes(Cluster, Local, Action) -> , [] , [Action] }, - ets:select(?ptab, [MS]). + classy_table:select(?ptab, [MS]). -doc """ Find minimal Lamport clock, @@ -892,7 +892,7 @@ max_toi(Site, #s{cluster = Cluster, site = Local}) -> , [] , ['$1'] } || Key <- [#mem{s = Site}, #host{s = Site}, #info{s = Site}, #live{s = Site}]], - L = ets:select(?ptab, MS), + L = classy_table:select(?ptab, MS), case L of [] -> undefined; _ -> lists:max(L) diff --git a/src/classy_node.erl b/src/classy_node.erl index 2e60465..b14decd 100644 --- a/src/classy_node.erl +++ b/src/classy_node.erl @@ -19,6 +19,7 @@ Management of the local site and node. , nodes/1 , peer_info/0 , node_of_site/2 + , node_to_site/0 , n_restarts/1 , prep_stop/1 ]). @@ -196,6 +197,18 @@ node_of_site(Site, OnlyConnected) -> undefined end. +-doc false. +-spec node_to_site() -> #{node() => classy:site()}. +node_to_site() -> + MS = { #classy_kv{ k = '$2' + , v = #site_info{node = '$1', _ = '_'} + , _ = '_' + } + , [{'=/=', '$1', undefined}] + , [{{'$1', '$2'}}] + }, + maps:from_list(classy_table:select(?site_info, [MS])). + -doc false. -spec n_restarts(classy:site()) -> {ok, non_neg_integer()} | undefined. n_restarts(Site) -> @@ -383,6 +396,7 @@ update_runtime(S) -> handle_kick(Cluster, Local, Target, Intent) -> case classy_hook:all(?on_pre_kick, [Cluster, Target, Intent]) of ok -> + classy_hook:foreach(?on_kick_decided, [Cluster, Target, Intent]), Ret = classy_membership:set_member(Cluster, Local, Target, false), classy_membership:flush(Cluster, Local), Ret; @@ -468,7 +482,7 @@ on_leave(S = #s{cluster = Cluster, site = Local}, Intent) -> prep_stop(leave, infinity), %% Sync with the business apps: classy_table:delete(?globals, ?the_cluster), - classy_hook:foreach(?on_post_kick, [Cluster, Local, Intent]), + classy_hook:foreach(?on_leave, [Cluster, Local, Intent]), classy_table:clear(?site_info), case Intent of join -> diff --git a/src/classy_table.erl b/src/classy_table.erl index 02d28c2..ba73748 100644 --- a/src/classy_table.erl +++ b/src/classy_table.erl @@ -88,6 +88,7 @@ They must not contain any sort of heavy or long-running tasks. , flush/1 , force_compaction/1 , lookup/2 + , select/2 %% For debugging: , dump_wal/1 , dump_wal/2 @@ -408,6 +409,24 @@ lookup(Tab, Key) -> [V || #classy_kv{v = V} <- ets:lookup(Tab, Key)] end. +-doc """ +Run @code{ets:select} on the table after making sure it's open and restored. + +WARNING: this function can block the caller until the table is fully restored. +""". +-spec select(tab(), ets:match_spec()) -> list(). +select(Tab, MS) -> + case ets:whereis(Tab) of + undefined -> + %% Protection against typos and deadlocks. If this happens, the + %% user must fix application startup order. + error({badtable, Tab}); + _ -> + %% Avoid reads while table is not fully restored: + optvar:read(?optvar(Tab)), + ets:select(Tab, MS) + end. + -doc """ Delete all data in the table. This is a durable operation. diff --git a/test/classy_SUITE.erl b/test/classy_SUITE.erl index 24d80e4..0ca93a7 100644 --- a/test/classy_SUITE.erl +++ b/test/classy_SUITE.erl @@ -661,6 +661,13 @@ t_091_node_of_site(_) -> {error, {_, J}}, ?ON(I, classy:node_of_site(J, OnlyLive))) || I <- Sites, J <- Sites, I =/= J, OnlyLive <- [true, false]], + %% Verify node_to_site: + ?assertEqual( + {ok, #{N1 => S1}}, + ?ON(S1, classy:node_to_site())), + ?assertEqual( + {ok, #{N2 => S2}}, + ?ON(S2, classy:node_to_site())), %% Form cluster: ?assertMatch(ok, ?ON(S2, classy:join_node(N1, join))), wait_site_joined(Sites, Cluster1, S2), @@ -669,6 +676,11 @@ t_091_node_of_site(_) -> {ok, maps:get(J, NodeMap)}, ?ON(I, classy:node_of_site(J, OnlyLive))) || I <- Sites, J <- Sites, OnlyLive <- [true, false]], + %% Verify node_to_site: + [?assertEqual( + {ok, #{N1 => S1, N2 => S2}}, + ?ON(I, classy:node_to_site())) + || I <- [S1, S2]], %% Shut down S2 and verify that S1 reacted on changes: stop_site(S2), ?block_until(#{?snk_kind := classy_peer_disconnected, site := S2}), @@ -678,7 +690,11 @@ t_091_node_of_site(_) -> ?ON(S1, classy:node_of_site(S2, true))), ?assertEqual( {ok, N2}, - ?ON(S1, classy:node_of_site(S2, false))) + ?ON(S1, classy:node_of_site(S2, false))), + %% Verify node_to_site doesn't change (returns last known node name): + ?assertEqual( + {ok, #{N1 => S1, N2 => S2}}, + ?ON(S1, classy:node_to_site())) end, [ fun no_unexpected_events/1 , fun events_on_all_sites/1 @@ -944,7 +960,7 @@ t_310_rpc_to_failing_node(_) -> begin %% Prepare N1 = create_start_site(S1, #{}), - N2 = create_start_site(S2, #{}), + _N2 = create_start_site(S2, #{}), {ok, Cluster} = ?ON(S1, classy:the_cluster()), ?assertMatch(ok, ?ON(S2, classy:join_node(N1, join))), wait_site_joined([S1, S2], Cluster, S2), @@ -1475,22 +1491,22 @@ t_413_fold_votes(_) -> ?check_trace( #{timetrap => 30_000}, begin - N1 = create_start_site(S1, #{peer => #{shutdown => halt}}), + _N1 = create_start_site(S1, #{peer => #{shutdown => halt}}), %% Make sure votes hang long enough for us to inspect them: ?force_ordering( #{?snk_kind := test_go}, #{?snk_kind := K} when K =:= classy_test_vote_commit; K =:= classy_test_post_vote), - {ok, ID1} = ?ON(S1, + {ok, _ID1} = ?ON(S1, classy_vote:create(#{ tag => Ref1 - , actions => #{S1 => make_vote(true, true, Ref1, 1)} - , post_vote => make_post_vote(Ref1) - })), - {ok, ID2} = ?ON(S1, - classy_vote:create(#{ tag => Ref2 - , actions => #{S1 => make_vote(true, true, Ref2, 1)} - , post_vote => make_post_vote(Ref2) - })), + , actions => #{S1 => make_vote(true, true, Ref1, 1)} + , post_vote => make_post_vote(Ref1) + })), + {ok, _ID2} = ?ON(S1, + classy_vote:create(#{ tag => Ref2 + , actions => #{S1 => make_vote(true, true, Ref2, 1)} + , post_vote => make_post_vote(Ref2) + })), ct:sleep(100), ?assertMatch( [ #{ id := _