Add live status for partitions

Summary:
Add a new "live" status for partitions that can be queried from `wa_raft_info` that represents whether a partition
believes it is an active and good participant in its current RAFT cluster. This status differs from "stale" as it is
not intended to indicate anything about whether the underlying FSM is in a good state.

Reviewed By: SarahDesouky

Differential Revision: D80120378

fbshipit-source-id: aee12c87d161a7d874eee72306ed1dde088b1f87

Author: hsun324

include/wa_raft.hrl

@@ -202,14 +202,15 @@
 -define(RAFT_ELECTION_TIMEOUT_MAX, raft_election_timeout_ms_max).
 -define(RAFT_ELECTION_TIMEOUT_MAX(App), ?RAFT_APP_CONFIG(App, ?RAFT_ELECTION_TIMEOUT_MAX, 7500)).
 
-%% Maximum time in milliseconds for which no valid heartbeat was received from a leader
-%% before considering a follower stale
--define(RAFT_FOLLOWER_STALE_INTERVAL, raft_follower_heartbeat_stale_ms).
--define(RAFT_FOLLOWER_STALE_INTERVAL(App), ?RAFT_APP_CONFIG(App, ?RAFT_FOLLOWER_STALE_INTERVAL, 5000)).
-%% Maximum number of unapplied log entries compared to the commited log entry before
-%% considering a follower stale
--define(RAFT_FOLLOWER_STALE_ENTRIES, raft_follower_max_lagging).
--define(RAFT_FOLLOWER_STALE_ENTRIES(App), ?RAFT_APP_CONFIG(App, ?RAFT_FOLLOWER_STALE_ENTRIES, 5000)).
+%% The maximum time in milliseconds during which a leader can continue to be considered live without
+%% receiving an updated heartbeat response quorum from replicas or during which a follower or witness
+%% can be considered live without receiving a heartbeat from a valid leader of the current term.
+-define(RAFT_LIVENESS_GRACE_PERIOD_MS, raft_liveness_grace_period_ms).
+-define(RAFT_LIVENESS_GRACE_PERIOD_MS(App), ?RAFT_APP_CONFIG(App, ?RAFT_LIVENESS_GRACE_PERIOD_MS, 30_000)).
+%% The maximum number of log entries that can be not yet applied to a follower or witnesse's log
+%% compared to the leader's commit index before the replica is considered stale.
+-define(RAFT_STALE_GRACE_PERIOD_ENTRIES, raft_stale_grace_period_entries).
+-define(RAFT_STALE_GRACE_PERIOD_ENTRIES(App), ?RAFT_APP_CONFIG(App, ?RAFT_STALE_GRACE_PERIOD_ENTRIES, 5_000)).
 
 %% Minium amount of time in seconds since the last successfully received
 %% heartbeat from a leader of a term for non-forced promotion to be allowed.

src/wa_raft_info.erl

@@ -16,6 +16,7 @@
     get_leader/2,
     get_current_term_and_leader/2,
     get_membership/2,
+    get_live/2,
     get_stale/2,
     get_state/2,
     get_message_queue_length/1
@@ -27,6 +28,7 @@
     delete_state/2,
     set_current_term_and_leader/4,
     set_membership/3,
+    set_live/3,
     set_stale/3,
     set_state/3
 ]).
@@ -35,6 +37,8 @@
 -define(RAFT_SERVER_STATE_KEY(Table, Partition), {state, Table, Partition}).
 %% Local RAFT server's most recently known term and leader
 -define(RAFT_CURRENT_TERM_AND_LEADER_KEY(Table, Partition), {term, Table, Partition}).
+%% Local RAFT server's current live flag - indicates if the server thinks it is part of a live cluster
+-define(RAFT_LIVE_KEY(Table, Partition), {live, Table, Partition}).
 %% Local RAFT server's current stale flag - indicates if the server thinks its data is stale
 -define(RAFT_STALE_KEY(Table, Partition), {stale, Table, Partition}).
 %% Local RAFT server's message queue length
@@ -77,6 +81,10 @@ get_current_term_and_leader(Table, Partition) ->
 get_state(Table, Partition) ->
     get(?RAFT_SERVER_STATE_KEY(Table, Partition), undefined).
 
+-spec get_live(wa_raft:table(), wa_raft:partition()) -> boolean().
+get_live(Table, Partition) ->
+    get(?RAFT_LIVE_KEY(Table, Partition), false).
+
 -spec get_stale(wa_raft:table(), wa_raft:partition()) -> boolean().
 get_stale(Table, Partition) ->
     get(?RAFT_STALE_KEY(Table, Partition), true).
@@ -118,6 +126,10 @@ set_state(Table, Partition, State) ->
 delete_state(Table, Partition) ->
     delete(?RAFT_SERVER_STATE_KEY(Table, Partition)).
 
+-spec set_live(wa_raft:table(), wa_raft:partition(), boolean()) -> true.
+set_live(Table, Partition, Live) ->
+    set(?RAFT_LIVE_KEY(Table, Partition), Live).
+
 -spec set_stale(wa_raft:table(), wa_raft:partition(), boolean()) -> true.
 set_stale(Table, Partition, Stale) ->
     set(?RAFT_STALE_KEY(Table, Partition), Stale),

src/wa_raft_server.erl

@@ -689,6 +689,7 @@ terminate(Reason, State, #raft_state{table = Table, partition = Partition} = Dat
     ?SERVER_LOG_NOTICE(State, Data, "terminating due to ~0P", [Reason, 20]),
     wa_raft_durable_state:sync(Data),
     wa_raft_info:delete_state(Table, Partition),
+    wa_raft_info:set_live(Table, Partition, false),
     wa_raft_info:set_stale(Table, Partition, true),
     ok.
 
@@ -934,7 +935,7 @@ stalled(
                 catch file:del_dir_r(Path)
             end;
         false ->
-            ?SERVER_LOG_NOTICE(State0, "at ~0p rejecting request to bootstrap with data.", [LastApplied, Index, Term]),
+            ?SERVER_LOG_NOTICE(State0, "at ~0p rejecting request to bootstrap with data.", [LastApplied]),
             {keep_state_and_data, {reply, From, {error, rejected}}}
     end;
 
@@ -1177,6 +1178,7 @@ leader(state_timeout = Type, Event, #raft_state{handover = {Peer, _, Timeout}} =
             ?SERVER_LOG_NOTICE(State, "handover to ~0p times out.", [Peer]),
             {keep_state, State#raft_state{handover = undefined}, {next_event, Type, Event}};
         false ->
+            check_leader_liveness(State),
             {keep_state_and_data, ?HEARTBEAT_TIMEOUT(State)}
     end;
 
@@ -1186,7 +1188,7 @@ leader(state_timeout, _, #raft_state{application = App} = State0) ->
         true ->
             State1 = append_entries_to_followers(State0),
             State2 = apply_single_node_cluster(State1),
-            check_leader_lagging(State2),
+            check_leader_liveness(State2),
             {keep_state, State1, ?HEARTBEAT_TIMEOUT(State2)};
         false ->
             ?SERVER_LOG_NOTICE(State0, "resigns from leadership because this node is ineligible.", []),
@@ -1679,7 +1681,7 @@ candidate(Type, Event, #raft_state{} = State) when is_tuple(Event), element(1, E
 
 %% [AppendEntries RPC] Switch to follower because current term now has a leader (5.2, 5.3)
 candidate(Type, ?REMOTE(Sender, ?APPEND_ENTRIES(_, _, _, _, _)) = Event, #raft_state{} = State) ->
-    ?SERVER_LOG_NOTICE(State, "switching to follower after receiving heartbeat from ~0P.", [Sender]),
+    ?SERVER_LOG_NOTICE(State, "switching to follower after receiving heartbeat from ~0p.", [Sender]),
     {next_state, follower_or_witness_state(State), State, {next_event, Type, Event}};
 
 %% [RequestVote RPC] Candidates should ignore incoming vote requests as they always vote for themselves (5.2)
@@ -1850,7 +1852,7 @@ disabled(Type, Event, #raft_state{} = State) ->
 witness(enter, PreviousStateName, #raft_state{} = State) ->
     ?RAFT_COUNT('raft.witness.enter'),
     ?SERVER_LOG_NOTICE(State, "becomes witness from state ~0p.", [PreviousStateName]),
-    {keep_state, enter_state(?FUNCTION_NAME, State)};
+    {keep_state, enter_state(?FUNCTION_NAME, State), ?ELECTION_TIMEOUT(State)};
 
 %% [Internal] Advance to newer term when requested
 witness(
@@ -1903,6 +1905,11 @@ witness(_, ?REMOTE(Candidate, ?REQUEST_VOTE(_, CandidateIndex, CandidateTerm)),
 witness(_, ?REMOTE(_, ?VOTE(_)), #raft_state{}) ->
     keep_state_and_data;
 
+%% [State Timeout] Check liveness, but do not restart state.
+witness(state_timeout, _, #raft_state{} = State) ->
+    check_follower_liveness(?FUNCTION_NAME, State),
+    {keep_state_and_data, ?ELECTION_TIMEOUT(State)};
+
 %% [Command] Defer to common handling for generic RAFT server commands
 witness(Type, ?RAFT_COMMAND(_, _) = Event, #raft_state{} = State) ->
     command(?FUNCTION_NAME, Type, Event, State);
@@ -2620,44 +2627,28 @@ enter_state(State, #raft_state{table = Table, partition = Partition} = Data0) ->
     Now = erlang:monotonic_time(millisecond),
     Data1 = Data0#raft_state{state_start_ts = Now},
     true = wa_raft_info:set_state(Table, Partition, State),
-    ok = check_stale_upon_entry(State, Now, Data1),
+    ok = check_stale_upon_entry(State, Data1),
     Data1.
 
--spec check_stale_upon_entry(State :: state(), Now :: integer(), Data :: #raft_state{}) -> ok.
-%% Followers and candidates may be stale upon entry due to not receiving a timely heartbeat from an active leader.
-check_stale_upon_entry(
-    State,
-    Now,
-    #raft_state{
-        application = Application,
-        table = Table,
-        partition = Partition,
-        leader_heartbeat_ts = LeaderHeartbeatTs
-    } = Data
-) when State =:= follower; State =:= candidate ->
-    Stale = case LeaderHeartbeatTs of
-        undefined ->
-            ?SERVER_LOG_NOTICE(State, Data, "is stale upon entry due to having no prior leader heartbeat.", []),
-            true;
-        _ ->
-            Delay = Now - LeaderHeartbeatTs,
-            case Delay > ?RAFT_FOLLOWER_STALE_INTERVAL(Application) of
-                true ->
-                    ?SERVER_LOG_NOTICE(State, Data, "is stale upon entry because the last leader heartbeat was received ~0p ms ago.", [Delay]),
-                    true;
-                false ->
-                    false
-            end
-    end,
-    true = wa_raft_info:set_stale(Table, Partition, Stale),
+-spec check_stale_upon_entry(State :: state(), Data :: #raft_state{}) -> ok.
+%% Followers and candidates are live upon entry if they've received a timely heartbeat
+%% and inherit their staleness from the previous state.
+check_stale_upon_entry(State, Data) when State =:= follower; State =:= candidate ->
+    check_follower_liveness(State, Data);
+%% Witnesses are live upon entry if they've received a timely heartbeat but are always stale.
+check_stale_upon_entry(witness, #raft_state{table = Table, partition = Partition} = Data) ->
+    check_follower_liveness(witness, Data),
+    wa_raft_info:set_stale(Table, Partition, true),
     ok;
-%% Leaders are never stale upon entry.
-check_stale_upon_entry(leader, _, #raft_state{table = Table, partition = Partition}) ->
-    true = wa_raft_info:set_stale(Table, Partition, false),
+%% Leaders are always live and never stale upon entry.
+check_stale_upon_entry(leader, #raft_state{table = Table, partition = Partition}) ->
+    wa_raft_info:set_live(Table, Partition, true),
+    wa_raft_info:set_stale(Table, Partition, false),
     ok;
-%% Witness, stalled and disabled servers are always stale.
-check_stale_upon_entry(_, _, #raft_state{table = Table, partition = Partition}) ->
-    true = wa_raft_info:set_stale(Table, Partition, true),
+%% Stalled and disabled servers are never live and always stale.
+check_stale_upon_entry(_, #raft_state{table = Table, partition = Partition}) ->
+    wa_raft_info:set_live(Table, Partition, false),
+    wa_raft_info:set_stale(Table, Partition, true),
     ok.
 
 %% Set a new current term and voted-for peer and clear any state that is associated with the previous term.
@@ -3069,7 +3060,8 @@ handle_heartbeat(
                 _ ->
                     Data2
             end,
-            check_follower_lagging(CommitIndex, Data3),
+            refresh_follower_liveness(State, Data3),
+            check_follower_lagging(State, CommitIndex, Data3),
             case follower_or_witness_state(Data3) of
                 State ->
                     {keep_state, Data3, ?ELECTION_TIMEOUT(Data3)};
@@ -3345,9 +3337,55 @@ should_heartbeat(#raft_state{application = App, last_heartbeat_ts = LastHeartbea
     Current = erlang:monotonic_time(millisecond),
     Current - Latest > ?RAFT_HEARTBEAT_INTERVAL(App).
 
-%% Check follower/witness state due to log entry lag and change stale flag if needed
--spec check_follower_lagging(LeaderCommit :: pos_integer(), State :: #raft_state{}) -> ok.
+-spec refresh_follower_liveness(State :: state(), Data :: #raft_state{}) -> ok.
+refresh_follower_liveness(State, #raft_state{table = Table, partition = Partition} = Data) ->
+    case wa_raft_info:get_live(Table, Partition) of
+        true ->
+            ok;
+        false ->
+            ?SERVER_LOG_NOTICE(State, Data, "is live", []),
+            wa_raft_info:set_live(Table, Partition, true)
+    end,
+    ok.
+
+%% Check the timestamp of a follower or witness's last received heartbeat to determine if
+%% the replica is live.
+-spec check_follower_liveness(State :: state(), Data :: #raft_state{}) -> ok.
+check_follower_liveness(
+    State,
+    #raft_state{
+        application = App,
+        table = Table,
+        partition = Partition,
+        leader_heartbeat_ts = LeaderHeartbeatTs
+    } = Data
+) ->
+    NowTs = erlang:monotonic_time(millisecond),
+    GracePeriod = ?RAFT_LIVENESS_GRACE_PERIOD_MS(App),
+    Liveness = wa_raft_info:get_live(Table, Partition),
+    Live = LeaderHeartbeatTs =/= undefined andalso LeaderHeartbeatTs + GracePeriod >= NowTs,
+    case Live of
+        Liveness ->
+            ok;
+        true ->
+            ?SERVER_LOG_NOTICE(State, Data, "is live", []),
+            wa_raft_info:set_live(Table, Partition, true);
+        false ->
+            ?SERVER_LOG_NOTICE(State, Data, "is no longer live after last leader heartbeat at ~0p", [LeaderHeartbeatTs]),
+            wa_raft_info:set_live(Table, Partition, false),
+            wa_raft_info:get_stale(Table, Partition) =:= true andalso begin
+                ?SERVER_LOG_NOTICE(State, Data, "is now stale due to liveness", []),
+                wa_raft_info:set_stale(Table, Partition, false)
+            end
+    end,
+    ok.
+
+%% Check the state of a follower or witness's last applied log entry versus the leader's
+%% commit index to determine if the replica is lagging behind and adjust the partition's
+%% staleness if needed.
+-spec check_follower_lagging(State :: state(), LeaderCommit :: pos_integer(), State :: #raft_state{}) -> ok.
 check_follower_lagging(
+    State,
     LeaderCommit,
     #raft_state{
         application = App,
@@ -3358,24 +3396,30 @@ check_follower_lagging(
 ) ->
     Lagging = LeaderCommit - LastApplied,
     ?RAFT_GATHER('raft.follower.lagging', Lagging),
-    case Lagging < ?RAFT_FOLLOWER_STALE_ENTRIES(App) of
-        true ->
-            wa_raft_info:get_stale(Table, Partition) =/= false andalso begin
-                ?SERVER_LOG_NOTICE(follower, Data, "catches up.", []),
+
+    % Witnesses are always considered stale and so do not re-check their staleness.
+    State =/= witness andalso begin
+        Stale = wa_raft_info:get_stale(Table, Partition),
+        case Lagging >= ?RAFT_STALE_GRACE_PERIOD_ENTRIES(App) of
+            Stale ->
+                ok;
+            true ->
+                ?SERVER_LOG_NOTICE(State, Data, "last applied at ~0p is ~0p behind leader's commit at ~0p.",
+                    [LastApplied, Lagging, LeaderCommit]),
+                wa_raft_info:set_stale(Table, Partition, true);
+            false ->
+                ?SERVER_LOG_NOTICE(State, Data, "catches up.", []),
                 wa_raft_info:set_stale(Table, Partition, false)
-            end;
-        false ->
-            wa_raft_info:get_stale(Table, Partition) =/= true andalso begin
-                ?SERVER_LOG_NOTICE(follower, Data, "is far behind ~p (leader ~p, follower ~p)",
-                    [Lagging, LeaderCommit, LastApplied]),
-                wa_raft_info:set_stale(Table, Partition, true)
-            end
+        end
     end,
+
     ok.
 
-%% Check leader state and set stale if needed
--spec check_leader_lagging(#raft_state{}) -> term().
-check_leader_lagging(
+%% As leader, compute the quorum of the most recent timestamps of follower's
+%% acknowledgement of heartbeats and update the partition's staleness and
+%% liveness when necessary.
+-spec check_leader_liveness(#raft_state{}) -> term().
+check_leader_liveness(
     #raft_state{
         application = App,
         table = Table,
@@ -3395,9 +3439,11 @@ check_leader_lagging(
             ok;
         true ->
             ?SERVER_LOG_NOTICE(leader, State, "is now stale due to last heartbeat quorum age being ~0p ms >= ~0p ms max", [QuorumAge, MaxAge]),
+            wa_raft_info:set_live(Table, Partition, true),
             wa_raft_info:set_stale(Table, Partition, true);
         false ->
             ?SERVER_LOG_NOTICE(leader, State, "is no longer stale after heartbeat quorum age drops to ~0p ms < ~0p ms max", [QuorumAge, MaxAge]),
+            wa_raft_info:set_live(Table, Partition, false),
             wa_raft_info:set_stale(Table, Partition, false)
     end.