Unverified Commit e2deade5 authored by Ozan Tezcan's avatar Ozan Tezcan Committed by GitHub
Browse files

Unknown node handling (#51)

Unknown node handling
parent 64747dee
...@@ -211,14 +211,14 @@ When we receive an entry log from the client it's possible we might not be a lea ...@@ -211,14 +211,14 @@ When we receive an entry log from the client it's possible we might not be a lea
If we aren't currently the leader of the raft cluster, we MUST send a redirect error message to the client. This is so that the client can connect directly to the leader in future connections. This enables future requests to be faster (ie. no redirects are required after the first redirect until the leader changes). If we aren't currently the leader of the raft cluster, we MUST send a redirect error message to the client. This is so that the client can connect directly to the leader in future connections. This enables future requests to be faster (ie. no redirects are required after the first redirect until the leader changes).
We use the ``raft_get_current_leader`` function to check who is the current leader. We use the ``raft_get_leader_id`` function to check who is the current leader.
*Example of ticketd sending a 301 HTTP redirect response:* *Example of ticketd sending a 301 HTTP redirect response:*
.. code-block:: c .. code-block:: c
/* redirect to leader if needed */ /* redirect to leader if needed */
raft_node_t* leader = raft_get_current_leader_node(sv->raft); raft_node_t* leader = raft_get_leader_node(sv->raft);
if (!leader) if (!leader)
{ {
return h2oh_respond_with_error(req, 503, "Leader unavailable"); return h2oh_respond_with_error(req, 503, "Leader unavailable");
......
...@@ -28,9 +28,7 @@ typedef enum { ...@@ -28,9 +28,7 @@ typedef enum {
RAFT_MEMBERSHIP_REMOVE, RAFT_MEMBERSHIP_REMOVE,
} raft_membership_e; } raft_membership_e;
#define RAFT_REQUESTVOTE_ERR_GRANTED 1 #define RAFT_UNKNOWN_NODE_ID (-1)
#define RAFT_REQUESTVOTE_ERR_NOT_GRANTED 0
#define RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE (-1)
typedef enum { typedef enum {
RAFT_STATE_NONE, RAFT_STATE_NONE,
...@@ -163,6 +161,10 @@ typedef struct ...@@ -163,6 +161,10 @@ typedef struct
* This message could force a leader/candidate to become a follower. */ * This message could force a leader/candidate to become a follower. */
typedef struct typedef struct
{ {
/** used to identify the sender node. Useful when this message is received
* from the nodes that are not part of the configuration yet. **/
raft_node_id_t leader_id;
/** id, to make it possible to associate responses with requests. */ /** id, to make it possible to associate responses with requests. */
raft_msg_id_t msg_id; raft_msg_id_t msg_id;
...@@ -282,14 +284,15 @@ typedef int ( ...@@ -282,14 +284,15 @@ typedef int (
/** Callback for providing debug logging information. /** Callback for providing debug logging information.
* This callback is optional * This callback is optional
* @param[in] raft The Raft server making this callback * @param[in] raft The Raft server making this callback
* @param[in] node The node that is the subject of this log. Could be NULL. * @param[in] node_id The node id that is the subject of this log. If log is not
* related to a node, this could be 'RAFT_UNKNOWN_NODE_ID'.
* @param[in] user_data User data that is passed from Raft server * @param[in] user_data User data that is passed from Raft server
* @param[in] buf The buffer that was logged */ * @param[in] buf The buffer that was logged */
typedef void ( typedef void (
*func_log_f *func_log_f
) ( ) (
raft_server_t* raft, raft_server_t* raft,
raft_node_t* node, raft_node_id_t node_id,
void *user_data, void *user_data,
const char *buf const char *buf
); );
...@@ -880,13 +883,14 @@ int raft_get_voted_for(raft_server_t* me); ...@@ -880,13 +883,14 @@ int raft_get_voted_for(raft_server_t* me);
/** Get what this node thinks the node ID of the leader is. /** Get what this node thinks the node ID of the leader is.
* @return node of what this node thinks is the valid leader; * @return node of what this node thinks is the valid leader;
* -1 if the leader is unknown */ * RAFT_UNKNOWN_NODE_ID if there is no leader */
raft_node_id_t raft_get_current_leader(raft_server_t* me); raft_node_id_t raft_get_leader_id(raft_server_t* me_);
/** Get what this node thinks the node of the leader is. /** Get what this node thinks the node of the leader is.
* @return node of what this node thinks is the valid leader; * @return node of what this node thinks is the valid leader;
* NULL if the leader is unknown */ * NULL if there is no leader or
raft_node_t* raft_get_current_leader_node(raft_server_t* me); * if the leader is not part of the local configuration yet */
raft_node_t* raft_get_leader_node(raft_server_t* me_);
/** /**
* @return callback user data */ * @return callback user data */
......
...@@ -74,9 +74,9 @@ typedef struct { ...@@ -74,9 +74,9 @@ typedef struct {
/* latest quorum id for the previous quorum_timeout round */ /* latest quorum id for the previous quorum_timeout round */
raft_msg_id_t last_acked_msg_id; raft_msg_id_t last_acked_msg_id;
/* what this node thinks is the node ID of the current leader, or NULL if /* what this node thinks is the node ID of the current leader or
* there isn't a known current leader. */ * RAFT_UNKNOWN_NODE_ID if there isn't a known current leader. */
raft_node_t* current_leader; raft_node_id_t leader_id;
/* callbacks */ /* callbacks */
raft_cbs_t cb; raft_cbs_t cb;
......
...@@ -46,8 +46,13 @@ void raft_set_heap_functions(void *(*_malloc)(size_t), ...@@ -46,8 +46,13 @@ void raft_set_heap_functions(void *(*_malloc)(size_t),
raft_free = _free; raft_free = _free;
} }
static void raft_log(raft_server_t *me_, raft_node_t* node, const char *fmt, ...) __attribute__ ((format (printf, 3, 4))); static void raft_log_node(raft_server_t *me_,
static void raft_log(raft_server_t *me_, raft_node_t* node, const char *fmt, ...) raft_node_id_t id,
const char *fmt, ...) __attribute__ ((format (printf, 3, 4)));
static void raft_log_node(raft_server_t *me_,
raft_node_id_t id,
const char *fmt, ...)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
...@@ -64,16 +69,18 @@ static void raft_log(raft_server_t *me_, raft_node_t* node, const char *fmt, ... ...@@ -64,16 +69,18 @@ static void raft_log(raft_server_t *me_, raft_node_t* node, const char *fmt, ...
} }
va_end(args); va_end(args);
me->cb.log(me_, node, me->udata, buf); me->cb.log(me_, id, me->udata, buf);
} }
#define raft_log(me, ...) (raft_log_node(me, RAFT_UNKNOWN_NODE_ID, __VA_ARGS__))
void raft_randomize_election_timeout(raft_server_t* me_) void raft_randomize_election_timeout(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
/* [election_timeout, 2 * election_timeout) */ /* [election_timeout, 2 * election_timeout) */
me->election_timeout_rand = me->election_timeout + rand() % me->election_timeout; me->election_timeout_rand = me->election_timeout + rand() % me->election_timeout;
raft_log(me_, NULL, "randomize election timeout to %d", me->election_timeout_rand); raft_log(me_, "randomize election timeout to %d", me->election_timeout_rand);
} }
void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id) void raft_update_quorum_meta(raft_server_t* me_, raft_msg_id_t id)
...@@ -109,7 +116,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg) ...@@ -109,7 +116,7 @@ raft_server_t* raft_new_with_log(const raft_log_impl_t *log_impl, void *log_arg)
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
me->snapshot_in_progress = 0; me->snapshot_in_progress = 0;
raft_set_snapshot_metadata((raft_server_t*)me, 0, 0); raft_set_snapshot_metadata((raft_server_t*)me, 0, 0);
...@@ -148,7 +155,7 @@ void raft_clear(raft_server_t* me_) ...@@ -148,7 +155,7 @@ void raft_clear(raft_server_t* me_)
raft_randomize_election_timeout(me_); raft_randomize_election_timeout(me_);
me->voting_cfg_change_log_idx = -1; me->voting_cfg_change_log_idx = -1;
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
me->commit_idx = 0; me->commit_idx = 0;
me->last_applied_idx = 0; me->last_applied_idx = 0;
me->num_nodes = 0; me->num_nodes = 0;
...@@ -160,6 +167,9 @@ raft_node_t* raft_add_node_internal(raft_server_t* me_, raft_entry_t *ety, void* ...@@ -160,6 +167,9 @@ raft_node_t* raft_add_node_internal(raft_server_t* me_, raft_entry_t *ety, void*
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
if (id == RAFT_UNKNOWN_NODE_ID)
return NULL;
/* set to voting if node already exists */ /* set to voting if node already exists */
raft_node_t* node = raft_get_node(me_, id); raft_node_t* node = raft_get_node(me_, id);
if (node) if (node)
...@@ -242,10 +252,8 @@ void raft_remove_node(raft_server_t* me_, raft_node_t* node) ...@@ -242,10 +252,8 @@ void raft_remove_node(raft_server_t* me_, raft_node_t* node)
memmove(&me->nodes[i], &me->nodes[i + 1], sizeof(*me->nodes) * (me->num_nodes - i - 1)); memmove(&me->nodes[i], &me->nodes[i + 1], sizeof(*me->nodes) * (me->num_nodes - i - 1));
me->num_nodes--; me->num_nodes--;
/* if we are removing the current leader, have to reset it, as this data is now stale */ if (me->leader_id == raft_node_get_id(node)) {
raft_node_t *current_leader_node = raft_get_current_leader_node(me_); me->leader_id = RAFT_UNKNOWN_NODE_ID;
if (node == current_leader_node) {
me->current_leader = NULL;
} }
raft_node_free(node); raft_node_free(node);
...@@ -352,9 +360,10 @@ int raft_election_start(raft_server_t* me_) ...@@ -352,9 +360,10 @@ int raft_election_start(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
raft_log(me_, NULL, "election starting: %d %d, term: %ld ci: %ld", raft_log(me_,
me->election_timeout_rand, me->timeout_elapsed, me->current_term, "election starting: %d %d, term: %ld ci: %ld",
raft_get_current_idx(me_)); me->election_timeout_rand, me->timeout_elapsed, me->current_term,
raft_get_current_idx(me_));
return raft_become_candidate(me_); return raft_become_candidate(me_);
} }
...@@ -364,7 +373,7 @@ int raft_become_leader(raft_server_t* me_) ...@@ -364,7 +373,7 @@ int raft_become_leader(raft_server_t* me_)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i; int i;
raft_log(me_, NULL, "becoming leader term:%ld", raft_get_current_term(me_)); raft_log(me_, "becoming leader term:%ld", raft_get_current_term(me_));
if (me->cb.notify_state_event) if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_LEADER); me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_LEADER);
...@@ -410,7 +419,7 @@ int raft_become_candidate(raft_server_t* me_) ...@@ -410,7 +419,7 @@ int raft_become_candidate(raft_server_t* me_)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int i; int i;
raft_log(me_, NULL, "becoming candidate"); raft_log(me_, "becoming candidate");
if (me->cb.notify_state_event) if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_CANDIDATE); me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_CANDIDATE);
...@@ -419,8 +428,11 @@ int raft_become_candidate(raft_server_t* me_) ...@@ -419,8 +428,11 @@ int raft_become_candidate(raft_server_t* me_)
return e; return e;
for (i = 0; i < me->num_nodes; i++) for (i = 0; i < me->num_nodes; i++)
raft_node_vote_for_me(me->nodes[i], 0); raft_node_vote_for_me(me->nodes[i], 0);
raft_vote(me_, me->node);
me->current_leader = NULL; if (raft_node_is_voting(me->node))
raft_vote(me_, me->node);
me->leader_id = RAFT_UNKNOWN_NODE_ID;
raft_set_state(me_, RAFT_STATE_CANDIDATE); raft_set_state(me_, RAFT_STATE_CANDIDATE);
raft_randomize_election_timeout(me_); raft_randomize_election_timeout(me_);
...@@ -442,7 +454,7 @@ void raft_become_follower(raft_server_t* me_) ...@@ -442,7 +454,7 @@ void raft_become_follower(raft_server_t* me_)
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
raft_log(me_, NULL, "becoming follower"); raft_log(me_, "becoming follower");
if (me->cb.notify_state_event) if (me->cb.notify_state_event)
me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_FOLLOWER); me->cb.notify_state_event(me_, raft_get_udata(me_), RAFT_STATE_FOLLOWER);
...@@ -537,7 +549,7 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -537,7 +549,7 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
if (me->last_acked_msg_id == quorum_id) if (me->last_acked_msg_id == quorum_id)
{ {
raft_log(me_, NULL, "quorum does not exist, stepping down"); raft_log(me_, "quorum does not exist, stepping down");
raft_become_follower(me_); raft_become_follower(me_);
} }
...@@ -549,13 +561,9 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period) ...@@ -549,13 +561,9 @@ int raft_periodic(raft_server_t* me_, int msec_since_last_period)
* happen when we get a client request */ * happen when we get a client request */
!raft_snapshot_is_in_progress(me_)) !raft_snapshot_is_in_progress(me_))
{ {
if (1 < raft_get_num_voting_nodes(me_) && int e = raft_election_start(me_);
raft_node_is_voting(raft_get_my_node(me_))) if (0 != e)
{ return e;
int e = raft_election_start(me_);
if (0 != e)
return e;
}
} }
if (me->last_applied_idx < raft_get_commit_idx(me_) && if (me->last_applied_idx < raft_get_commit_idx(me_) &&
...@@ -588,7 +596,7 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -588,7 +596,7 @@ int raft_recv_appendentries_response(raft_server_t* me_,
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
raft_log(me_, node, raft_log_node(me_, raft_node_get_id(node),
"received appendentries response %s ci:%ld rci:%ld msgid:%lu", "received appendentries response %s ci:%ld rci:%ld msgid:%lu",
r->success == 1 ? "SUCCESS" : "fail", r->success == 1 ? "SUCCESS" : "fail",
raft_get_current_idx(me_), raft_get_current_idx(me_),
...@@ -613,7 +621,7 @@ int raft_recv_appendentries_response(raft_server_t* me_, ...@@ -613,7 +621,7 @@ int raft_recv_appendentries_response(raft_server_t* me_,
if (0 != e) if (0 != e)
return e; return e;
raft_become_follower(me_); raft_become_follower(me_);
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
return 0; return 0;
} }
else if (me->current_term != r->term) else if (me->current_term != r->term)
...@@ -712,13 +720,14 @@ int raft_recv_appendentries( ...@@ -712,13 +720,14 @@ int raft_recv_appendentries(
int e = 0; int e = 0;
if (0 < ae->n_entries) if (0 < ae->n_entries)
raft_log(me_, node, "recvd appendentries t:%ld ci:%ld lc:%ld pli:%ld plt:%ld #%d", {
ae->term, raft_log_node(
raft_get_current_idx(me_), me_, ae->leader_id,
ae->leader_commit, "recvd appendentries li:%d t:%ld ci:%ld lc:%ld pli:%ld plt:%ld #%d",
ae->prev_log_idx, ae->leader_id, ae->term, raft_get_current_idx(me_),
ae->prev_log_term, ae->leader_commit, ae->prev_log_idx, ae->prev_log_term,
ae->n_entries); ae->n_entries);
}
r->msg_id = ae->msg_id; r->msg_id = ae->msg_id;
r->prev_log_idx = ae->prev_log_idx; r->prev_log_idx = ae->prev_log_idx;
...@@ -738,14 +747,14 @@ int raft_recv_appendentries( ...@@ -738,14 +747,14 @@ int raft_recv_appendentries(
else if (ae->term < me->current_term) else if (ae->term < me->current_term)
{ {
/* 1. Reply false if term < currentTerm (§5.1) */ /* 1. Reply false if term < currentTerm (§5.1) */
raft_log(me_, node, "AE term %ld is less than current term %ld", raft_log_node(me_, ae->leader_id,
ae->term, me->current_term); "AE term %ld is less than current term %ld",
ae->term, me->current_term);
goto out; goto out;
} }
/* update current leader because ae->term is up to date */ /* update current leader because ae->term is up to date */
me->current_leader = node; me->leader_id = ae->leader_id;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
/* Not the first appendentries we've received */ /* Not the first appendentries we've received */
...@@ -760,7 +769,8 @@ int raft_recv_appendentries( ...@@ -760,7 +769,8 @@ int raft_recv_appendentries(
if (me->snapshot_last_term != ae->prev_log_term) if (me->snapshot_last_term != ae->prev_log_term)
{ {
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log(me_, node, "Snapshot AE prev conflicts with committed entry"); raft_log_node(me_, ae->leader_id,
"Snapshot AE prev conflicts with committed entry");
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
if (ety) if (ety)
raft_entry_release(ety); raft_entry_release(ety);
...@@ -771,18 +781,20 @@ int raft_recv_appendentries( ...@@ -771,18 +781,20 @@ int raft_recv_appendentries(
whose term matches prevLogTerm (§5.3) */ whose term matches prevLogTerm (§5.3) */
else if (!ety) else if (!ety)
{ {
raft_log(me_, node, "AE no log at prev_idx %ld", ae->prev_log_idx); raft_log_node(me_, ae->leader_id,
"AE no log at prev_idx %ld", ae->prev_log_idx);
goto out; goto out;
} }
else if (ety->term != ae->prev_log_term) else if (ety->term != ae->prev_log_term)
{ {
raft_log(me_, node, "AE term doesn't match prev_term (ie. %ld vs %ld) ci:%ld comi:%ld lcomi:%ld pli:%ld", raft_log_node(me_, ae->leader_id, "AE term doesn't match prev_term (ie. %ld vs %ld) ci:%ld comi:%ld lcomi:%ld pli:%ld",
ety->term, ae->prev_log_term, raft_get_current_idx(me_), ety->term, ae->prev_log_term, raft_get_current_idx(me_),
raft_get_commit_idx(me_), ae->leader_commit, ae->prev_log_idx); raft_get_commit_idx(me_), ae->leader_commit, ae->prev_log_idx);
if (ae->prev_log_idx <= raft_get_commit_idx(me_)) if (ae->prev_log_idx <= raft_get_commit_idx(me_))
{ {
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log(me_, node, "AE prev conflicts with committed entry"); raft_log_node(me_, ae->leader_id,
"AE prev conflicts with committed entry");
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
raft_entry_release(ety); raft_entry_release(ety);
goto out; goto out;
...@@ -818,7 +830,7 @@ int raft_recv_appendentries( ...@@ -818,7 +830,7 @@ int raft_recv_appendentries(
if (ety_index <= raft_get_commit_idx(me_)) if (ety_index <= raft_get_commit_idx(me_))
{ {
/* Should never happen; something is seriously wrong! */ /* Should never happen; something is seriously wrong! */
raft_log(me_, node, "AE entry conflicts with committed entry ci:%ld comi:%ld lcomi:%ld pli:%ld", raft_log_node(me_, ae->leader_id, "AE entry conflicts with committed entry ci:%ld comi:%ld lcomi:%ld pli:%ld",
raft_get_current_idx(me_), raft_get_commit_idx(me_), raft_get_current_idx(me_), raft_get_commit_idx(me_),
ae->leader_commit, ae->prev_log_idx); ae->leader_commit, ae->prev_log_idx);
e = RAFT_ERR_SHUTDOWN; e = RAFT_ERR_SHUTDOWN;
...@@ -865,9 +877,6 @@ int raft_already_voted(raft_server_t* me_) ...@@ -865,9 +877,6 @@ int raft_already_voted(raft_server_t* me_)
static int raft_should_grant_vote(raft_server_t* me_, msg_requestvote_t* vr) static int raft_should_grant_vote(raft_server_t* me_, msg_requestvote_t* vr)
{ {
if (!raft_node_is_voting(raft_get_my_node(me_)))
return 0;
if (vr->term < raft_get_current_term(me_)) if (vr->term < raft_get_current_term(me_))
return 0; return 0;
...@@ -902,13 +911,15 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -902,13 +911,15 @@ int raft_recv_requestvote(raft_server_t* me_,
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
int e = 0; int e = 0;
r->vote_granted = 0;
if (!node) if (!node)
node = raft_get_node(me_, vr->candidate_id); node = raft_get_node(me_, vr->candidate_id);
/* Reject request if we have a leader */ /* Reject request if we have a leader */
if (me->current_leader && me->current_leader != node && if (me->leader_id != RAFT_UNKNOWN_NODE_ID &&
(me->timeout_elapsed < me->election_timeout)) { me->leader_id != vr->candidate_id &&
r->vote_granted = 0; me->timeout_elapsed < me->election_timeout) {
goto done; goto done;
} }
...@@ -916,11 +927,10 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -916,11 +927,10 @@ int raft_recv_requestvote(raft_server_t* me_,
{ {
e = raft_set_current_term(me_, vr->term); e = raft_set_current_term(me_, vr->term);
if (0 != e) { if (0 != e) {
r->vote_granted = 0;
goto done; goto done;
} }
raft_become_follower(me_); raft_become_follower(me_);
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
} }
if (raft_should_grant_vote(me_, vr)) if (raft_should_grant_vote(me_, vr))
...@@ -932,32 +942,15 @@ int raft_recv_requestvote(raft_server_t* me_, ...@@ -932,32 +942,15 @@ int raft_recv_requestvote(raft_server_t* me_,
e = raft_vote_for_nodeid(me_, vr->candidate_id); e = raft_vote_for_nodeid(me_, vr->candidate_id);
if (0 == e) if (0 == e)
r->vote_granted = 1; r->vote_granted = 1;
else
r->vote_granted = 0;
/* must be in an election. */ /* must be in an election. */
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
me->timeout_elapsed = 0; me->timeout_elapsed = 0;
} }
else
{
/* It's possible the candidate node has been removed from the cluster but
* hasn't received the appendentries that confirms the removal. Therefore
* the node is partitioned and still thinks its part of the cluster. It
* will eventually send a requestvote. This is error response tells the
* node that it might be removed. */
if (!node)
{
r->vote_granted = RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE;
goto done;
}
else
r->vote_granted = 0;
}
done: done:
raft_log(me_, node, "node requested vote: %d replying: %s", raft_log_node(me_, vr->candidate_id, "node requested vote: %d replying: %s",
raft_node_get_id(node), vr->candidate_id,
r->vote_granted == 1 ? "granted" : r->vote_granted == 1 ? "granted" :
r->vote_granted == 0 ? "not granted" : "unknown"); r->vote_granted == 0 ? "not granted" : "unknown");
...@@ -979,9 +972,12 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -979,9 +972,12 @@ int raft_recv_requestvote_response(raft_server_t* me_,
{ {
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
raft_log(me_, node, "node responded to requestvote status: %s", raft_log_node(me_, raft_node_get_id(node),
r->vote_granted == 1 ? "granted" : "node responded to requestvote status:%s ct:%ld rt:%ld",
r->vote_granted == 0 ? "not granted" : "unknown"); r->vote_granted == 1 ? "granted" :
r->vote_granted == 0 ? "not granted" : "unknown",
me->current_term,
r->term);
if (!raft_is_candidate(me_) || raft_get_current_term(me_) > r->term) if (!raft_is_candidate(me_) || raft_get_current_term(me_) > r->term)
{ {
...@@ -994,40 +990,20 @@ int raft_recv_requestvote_response(raft_server_t* me_, ...@@ -994,40 +990,20 @@ int raft_recv_requestvote_response(raft_server_t* me_,
if (0 != e) if (0 != e)
return e; return e;
raft_become_follower(me_); raft_become_follower(me_);
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
return 0; return 0;
} }
raft_log(me_, node, "node responded to requestvote status:%s ct:%ld rt:%ld", if (r->vote_granted)
r->vote_granted == 1 ? "granted" :
r->vote_granted == 0 ? "not granted" : "unknown",
me->current_term,
r->term);
switch (r->vote_granted)
{ {
case RAFT_REQUESTVOTE_ERR_GRANTED: if (node)
if (node) raft_node_vote_for_me(node, 1);
raft_node_vote_for_me(node, 1); int votes = raft_get_nvotes_for_me(me_);
int votes = raft_get_nvotes_for_me(me_); if (raft_votes_is_majority(raft_get_num_voting_nodes(me_), votes)) {
if (raft_votes_is_majority(raft_get_num_voting_nodes(me_), votes)) { int e = raft_become_leader(me_);
int e = raft_become_leader(me_); if (0 != e)
if (0 != e) return e;
return e; }
}
break;
case RAFT_REQUESTVOTE_ERR_NOT_GRANTED:
break;
case RAFT_REQUESTVOTE_ERR_UNKNOWN_NODE:
if (raft_node_is_voting(raft_get_my_node(me_)) &&
me->connected == RAFT_NODE_STATUS_DISCONNECTING)
return RAFT_ERR_SHUTDOWN;
break;
default:
assert(0);
} }
return 0; return 0;
...@@ -1055,7 +1031,7 @@ int raft_recv_entry(raft_server_t* me_, ...@@ -1055,7 +1031,7 @@ int raft_recv_entry(raft_server_t* me_,
if (!raft_is_leader(me_)) if (!raft_is_leader(me_))
return RAFT_ERR_NOT_LEADER; return RAFT_ERR_NOT_LEADER;
raft_log(me_, NULL, "received entry t:%ld id: %d idx: %ld", raft_log(me_, "received entry t:%ld id: %d idx: %ld",
me->current_term, ety->id, raft_get_current_idx(me_) + 1); me->current_term, ety->id, raft_get_current_idx(me_) + 1);
ety->term = me->current_term; ety->term = me->current_term;
...@@ -1103,7 +1079,8 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node) ...@@ -1103,7 +1079,8 @@ int raft_send_requestvote(raft_server_t* me_, raft_node_t* node)
assert(node); assert(node);
assert(node != me->node); assert(node != me->node);
raft_log(me_, node, "sending requestvote to: %d", raft_node_get_id(node)); raft_node_id_t id = raft_node_get_id(node);
raft_log_node(me_, id, "sending requestvote to: %d", id);
rv.term = me->current_term; rv.term = me->current_term;
rv.last_log_idx = raft_get_current_idx(me_); rv.last_log_idx = raft_get_current_idx(me_);
...@@ -1149,7 +1126,7 @@ int raft_apply_entry(raft_server_t* me_) ...@@ -1149,7 +1126,7 @@ int raft_apply_entry(raft_server_t* me_)
if (!ety) if (!ety)
return -1; return -1;
raft_log(me_, NULL, "applying log: %ld, id: %d size: %d", raft_log(me_, "applying log: %ld, id: %d size: %d",
log_idx, ety->id, ety->data_len); log_idx, ety->id, ety->data_len);
me->last_applied_idx++; me->last_applied_idx++;
...@@ -1276,15 +1253,17 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node) ...@@ -1276,15 +1253,17 @@ int raft_send_appendentries(raft_server_t* me_, raft_node_t* node)
} }
} }
raft_log(me_, node, "sending appendentries: ci:%lu comi:%lu t:%lu lc:%lu pli:%lu plt:%lu msgid:%lu #%d", raft_log_node(me_,
raft_get_current_idx(me_), raft_node_get_id(node),
raft_get_commit_idx(me_), "sending appendentries: ci:%lu comi:%lu t:%lu lc:%lu pli:%lu plt:%lu msgid:%lu #%d",
ae.term, raft_get_current_idx(me_),
ae.leader_commit, raft_get_commit_idx(me_),
ae.prev_log_idx, ae.term,
ae.prev_log_term, ae.leader_commit,
ae.msg_id, ae.prev_log_idx,
ae.n_entries); ae.prev_log_term,
ae.msg_id,
ae.n_entries);
int res = me->cb.send_appendentries(me_, me->udata, node, &ae); int res = me->cb.send_appendentries(me_, me->udata, node, &ae);
raft_entry_release_list(ae.entries, ae.n_entries); raft_entry_release_list(ae.entries, ae.n_entries);
...@@ -1468,7 +1447,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags) ...@@ -1468,7 +1447,7 @@ int raft_begin_snapshot(raft_server_t *me_, int flags)
me->snapshot_in_progress = 1; me->snapshot_in_progress = 1;
me->snapshot_flags = flags; me->snapshot_flags = flags;
raft_log(me_, NULL, raft_log(me_,
"begin snapshot sli:%ld slt:%ld slogs:%ld", "begin snapshot sli:%ld slt:%ld slogs:%ld",
me->snapshot_last_idx, me->snapshot_last_idx,
me->snapshot_last_term, me->snapshot_last_term,
...@@ -1509,7 +1488,7 @@ int raft_end_snapshot(raft_server_t *me_) ...@@ -1509,7 +1488,7 @@ int raft_end_snapshot(raft_server_t *me_)
me->snapshot_in_progress = 0; me->snapshot_in_progress = 0;
raft_log(me_, NULL, raft_log(me_,
"end snapshot base:%ld commit-index:%ld current-index:%ld", "end snapshot base:%ld commit-index:%ld current-index:%ld",
me->log_impl->first_idx(me->log) - 1, me->log_impl->first_idx(me->log) - 1,
raft_get_commit_idx(me_), raft_get_commit_idx(me_),
...@@ -1568,7 +1547,7 @@ int raft_begin_load_snapshot( ...@@ -1568,7 +1547,7 @@ int raft_begin_load_snapshot(
} }
raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER); raft_set_state((raft_server_t*)me, RAFT_STATE_FOLLOWER);
me->current_leader = NULL; me->leader_id = RAFT_UNKNOWN_NODE_ID;
me->log_impl->reset(me->log, last_included_index + 1, last_included_term); me->log_impl->reset(me->log, last_included_index + 1, last_included_term);
...@@ -1594,7 +1573,7 @@ int raft_begin_load_snapshot( ...@@ -1594,7 +1573,7 @@ int raft_begin_load_snapshot(
me->nodes[0] = me->nodes[my_node_by_idx]; me->nodes[0] = me->nodes[my_node_by_idx];
me->num_nodes = 1; me->num_nodes = 1;
raft_log(me_, NULL, raft_log(me_,
"loaded snapshot sli:%ld slt:%ld slogs:%ld", "loaded snapshot sli:%ld slt:%ld slogs:%ld",
me->snapshot_last_idx, me->snapshot_last_idx,
me->snapshot_last_term, me->snapshot_last_term,
......
...@@ -137,7 +137,7 @@ void raft_set_state(raft_server_t* me_, int state) ...@@ -137,7 +137,7 @@ void raft_set_state(raft_server_t* me_, int state)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
/* if became the leader, then update the current leader entry */ /* if became the leader, then update the current leader entry */
if (state == RAFT_STATE_LEADER) if (state == RAFT_STATE_LEADER)
me->current_leader = me->node; me->leader_id = raft_node_get_id(me->node);
me->state = state; me->state = state;
} }
...@@ -176,18 +176,16 @@ raft_node_t* raft_get_node_from_idx(raft_server_t* me_, const raft_index_t idx) ...@@ -176,18 +176,16 @@ raft_node_t* raft_get_node_from_idx(raft_server_t* me_, const raft_index_t idx)
return me->nodes[idx]; return me->nodes[idx];
} }
raft_node_id_t raft_get_current_leader(raft_server_t* me_) raft_node_id_t raft_get_leader_id(raft_server_t* me_)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
if (me->current_leader) return me->leader_id;
return raft_node_get_id(me->current_leader);
return -1;
} }
raft_node_t* raft_get_current_leader_node(raft_server_t* me_) raft_node_t* raft_get_leader_node(raft_server_t* me_)
{ {
raft_server_private_t* me = (void*)me_; raft_server_private_t* me = (void*)me_;
return me->current_leader; return raft_get_node(me_, me->leader_id);
} }
void* raft_get_udata(raft_server_t* me_) void* raft_get_udata(raft_server_t* me_)
...@@ -277,4 +275,4 @@ int raft_is_single_node_voting_cluster(raft_server_t *me_) ...@@ -277,4 +275,4 @@ int raft_is_single_node_voting_cluster(raft_server_t *me_)
raft_server_private_t* me = (raft_server_private_t*)me_; raft_server_private_t* me = (raft_server_private_t*)me_;
return (1 == raft_get_num_voting_nodes(me_) && raft_node_is_voting(me->node)); return (1 == raft_get_num_voting_nodes(me_) && raft_node_is_voting(me->node));
} }
\ No newline at end of file
...@@ -203,7 +203,7 @@ void TestRaft_server_add_non_voting_node_with_already_existing_voting_id_is_not_ ...@@ -203,7 +203,7 @@ void TestRaft_server_add_non_voting_node_with_already_existing_voting_id_is_not_
void TestRaft_server_remove_node(CuTest * tc) void TestRaft_server_remove_node(CuTest * tc)
{ {
void *r = raft_new(); void *r = raft_new();
void* n1 = raft_add_node(r, NULL, 1, 0); void* n1 = raft_add_node(r, NULL, 1, 1);
void* n2 = raft_add_node(r, NULL, 9, 0); void* n2 = raft_add_node(r, NULL, 9, 0);
raft_remove_node(r, n1); raft_remove_node(r, n1);
...@@ -216,6 +216,7 @@ void TestRaft_server_remove_node(CuTest * tc) ...@@ -216,6 +216,7 @@ void TestRaft_server_remove_node(CuTest * tc)
void TestRaft_election_start_increments_term(CuTest * tc) void TestRaft_election_start_increments_term(CuTest * tc)
{ {
void *r = raft_new(); void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_callbacks(r, &generic_funcs, NULL); raft_set_callbacks(r, &generic_funcs, NULL);
raft_set_current_term(r, 1); raft_set_current_term(r, 1);
raft_election_start(r); raft_election_start(r);
...@@ -783,7 +784,7 @@ void TestRaft_leader_recv_requestvote_does_not_step_down( ...@@ -783,7 +784,7 @@ void TestRaft_leader_recv_requestvote_does_not_step_down(
memset(&rv, 0, sizeof(msg_requestvote_t)); memset(&rv, 0, sizeof(msg_requestvote_t));
rv.term = 1; rv.term = 1;
raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr); raft_recv_requestvote(r, raft_get_node(r, 2), &rv, &rvr);
CuAssertIntEquals(tc, 1, raft_get_current_leader(r)); CuAssertIntEquals(tc, 1, raft_get_leader_id(r));
} }
/* Reply true if term >= currentTerm (§5.1) */ /* Reply true if term >= currentTerm (§5.1) */
...@@ -989,6 +990,7 @@ void TestRaft_server_recv_requestvote_ignore_if_master_is_fresh(CuTest * tc) ...@@ -989,6 +990,7 @@ void TestRaft_server_recv_requestvote_ignore_if_master_is_fresh(CuTest * tc)
void TestRaft_follower_becomes_follower_is_follower(CuTest * tc) void TestRaft_follower_becomes_follower_is_follower(CuTest * tc)
{ {
void *r = raft_new(); void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_become_follower(r); raft_become_follower(r);
CuAssertTrue(tc, raft_is_follower(r)); CuAssertTrue(tc, raft_is_follower(r));
} }
...@@ -1025,7 +1027,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt ...@@ -1025,7 +1027,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt
raft_add_node(r, NULL, 1, 1); raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0); raft_add_node(r, NULL, 2, 0);
/* no leader known at this point */ /* no leader known at this point */
CuAssertTrue(tc, -1 == raft_get_current_leader(r)); CuAssertTrue(tc, -1 == raft_get_leader_id(r));
/* term is low */ /* term is low */
msg_appendentries_t ae; msg_appendentries_t ae;
...@@ -1039,7 +1041,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt ...@@ -1039,7 +1041,7 @@ void TestRaft_follower_recv_appendentries_reply_false_if_term_less_than_currentt
CuAssertTrue(tc, 0 == aer.success); CuAssertTrue(tc, 0 == aer.success);
/* rejected appendentries doesn't change the current leader. */ /* rejected appendentries doesn't change the current leader. */
CuAssertTrue(tc, -1 == raft_get_current_leader(r)); CuAssertTrue(tc, -1 == raft_get_leader_id(r));
} }
void TestRaft_follower_recv_appendentries_does_not_need_node(CuTest * tc) void TestRaft_follower_recv_appendentries_does_not_need_node(CuTest * tc)
...@@ -1079,13 +1081,14 @@ void TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_current ...@@ -1079,13 +1081,14 @@ void TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_current
/* older currentterm */ /* older currentterm */
raft_set_current_term(r, 1); raft_set_current_term(r, 1);
CuAssertTrue(tc, -1 == raft_get_current_leader(r)); CuAssertTrue(tc, -1 == raft_get_leader_id(r));
/* newer term for appendentry */ /* newer term for appendentry */
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
/* no prev log idx */ /* no prev log idx */
ae.prev_log_idx = 0; ae.prev_log_idx = 0;
ae.term = 2; ae.term = 2;
ae.leader_id = 2;
/* appendentry has newer term, so we change our currentterm */ /* appendentry has newer term, so we change our currentterm */
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer); raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
...@@ -1094,7 +1097,7 @@ void TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_current ...@@ -1094,7 +1097,7 @@ void TestRaft_follower_recv_appendentries_updates_currentterm_if_term_gt_current
/* term has been updated */ /* term has been updated */
CuAssertTrue(tc, 2 == raft_get_current_term(r)); CuAssertTrue(tc, 2 == raft_get_current_term(r));
/* and leader has been updated */ /* and leader has been updated */
CuAssertIntEquals(tc, 2, raft_get_current_leader(r)); CuAssertIntEquals(tc, 2, raft_get_leader_id(r));
} }
void TestRaft_follower_recv_appendentries_does_not_log_if_no_entries_are_specified( void TestRaft_follower_recv_appendentries_does_not_log_if_no_entries_are_specified(
...@@ -1883,7 +1886,7 @@ void TestRaft_candidate_becomes_candidate_is_candidate(CuTest * tc) ...@@ -1883,7 +1886,7 @@ void TestRaft_candidate_becomes_candidate_is_candidate(CuTest * tc)
void *r = raft_new(); void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_become_candidate(r); raft_become_candidate(r);
CuAssertTrue(tc, raft_is_candidate(r)); CuAssertTrue(tc, raft_is_candidate(r));
} }
...@@ -1897,6 +1900,7 @@ void TestRaft_follower_becoming_candidate_increments_current_term(CuTest * tc) ...@@ -1897,6 +1900,7 @@ void TestRaft_follower_becoming_candidate_increments_current_term(CuTest * tc)
}; };
void *r = raft_new(); void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
CuAssertTrue(tc, 0 == raft_get_current_term(r)); CuAssertTrue(tc, 0 == raft_get_current_term(r));
...@@ -1931,6 +1935,8 @@ void TestRaft_follower_becoming_candidate_resets_election_timeout(CuTest * tc) ...@@ -1931,6 +1935,8 @@ void TestRaft_follower_becoming_candidate_resets_election_timeout(CuTest * tc)
}; };
void *r = raft_new(); void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_callbacks(r, &funcs, NULL); raft_set_callbacks(r, &funcs, NULL);
raft_set_election_timeout(r, 1000); raft_set_election_timeout(r, 1000);
...@@ -2159,7 +2165,7 @@ void TestRaft_candidate_recv_requestvote_response_becomes_follower_if_current_te ...@@ -2159,7 +2165,7 @@ void TestRaft_candidate_recv_requestvote_response_becomes_follower_if_current_te
raft_set_state(r, RAFT_STATE_CANDIDATE); raft_set_state(r, RAFT_STATE_CANDIDATE);
raft_vote(r, 0); raft_vote(r, 0);
CuAssertTrue(tc, 0 == raft_is_follower(r)); CuAssertTrue(tc, 0 == raft_is_follower(r));
CuAssertTrue(tc, -1 == raft_get_current_leader(r)); CuAssertTrue(tc, -1 == raft_get_leader_id(r));
CuAssertTrue(tc, 1 == raft_get_current_term(r)); CuAssertTrue(tc, 1 == raft_get_current_term(r));
msg_requestvote_response_t rvr; msg_requestvote_response_t rvr;
...@@ -2190,7 +2196,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower( ...@@ -2190,7 +2196,7 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(
raft_set_state(r, RAFT_STATE_CANDIDATE); raft_set_state(r, RAFT_STATE_CANDIDATE);
raft_vote(r, 0); raft_vote(r, 0);
CuAssertTrue(tc, 0 == raft_is_follower(r)); CuAssertTrue(tc, 0 == raft_is_follower(r));
CuAssertTrue(tc, -1 == raft_get_current_leader(r)); CuAssertTrue(tc, -1 == raft_get_leader_id(r));
CuAssertTrue(tc, 0 == raft_get_current_term(r)); CuAssertTrue(tc, 0 == raft_get_current_term(r));
/* receive recent appendentries */ /* receive recent appendentries */
...@@ -2199,10 +2205,11 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower( ...@@ -2199,10 +2205,11 @@ void TestRaft_candidate_recv_appendentries_frm_leader_results_in_follower(
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 1; ae.term = 1;
ae.leader_id = 2;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer); raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
/* after accepting a leader, it's available as the last known leader */ /* after accepting a leader, it's available as the last known leader */
CuAssertTrue(tc, 2 == raft_get_current_leader(r)); CuAssertTrue(tc, 2 == raft_get_leader_id(r));
CuAssertTrue(tc, 1 == raft_get_current_term(r)); CuAssertTrue(tc, 1 == raft_get_current_term(r));
CuAssertTrue(tc, -1 == raft_get_voted_for(r)); CuAssertTrue(tc, -1 == raft_get_voted_for(r));
} }
...@@ -2256,8 +2263,11 @@ void TestRaft_candidate_recv_appendentries_from_same_term_results_in_step_down( ...@@ -2256,8 +2263,11 @@ void TestRaft_candidate_recv_appendentries_from_same_term_results_in_step_down(
void TestRaft_leader_becomes_leader_is_leader(CuTest * tc) void TestRaft_leader_becomes_leader_is_leader(CuTest * tc)
{ {
void *r = raft_new(); void *r = raft_new();
raft_node_t *n = raft_add_node(r, NULL, 1, 1);
raft_become_leader(r); raft_become_leader(r);
CuAssertTrue(tc, raft_is_leader(r)); CuAssertTrue(tc, raft_is_leader(r));
CuAssertTrue(tc, raft_get_leader_node(r) == n);
} }
void TestRaft_leader_becomes_leader_does_not_clear_voted_for(CuTest * tc) void TestRaft_leader_becomes_leader_does_not_clear_voted_for(CuTest * tc)
...@@ -3431,7 +3441,7 @@ void TestRaft_leader_recv_appendentries_response_steps_down_if_term_is_newer( ...@@ -3431,7 +3441,7 @@ void TestRaft_leader_recv_appendentries_response_steps_down_if_term_is_newer(
aer.current_idx = 2; aer.current_idx = 2;
raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer); raft_recv_appendentries_response(r, raft_get_node(r, 2), &aer);
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
CuAssertTrue(tc, -1 == raft_get_current_leader(r)); CuAssertTrue(tc, -1 == raft_get_leader_id(r));
} }
void TestRaft_leader_recv_appendentries_steps_down_if_newer( void TestRaft_leader_recv_appendentries_steps_down_if_newer(
...@@ -3454,10 +3464,11 @@ void TestRaft_leader_recv_appendentries_steps_down_if_newer( ...@@ -3454,10 +3464,11 @@ void TestRaft_leader_recv_appendentries_steps_down_if_newer(
raft_set_current_term(r, 5); raft_set_current_term(r, 5);
/* check that node 1 considers itself the leader */ /* check that node 1 considers itself the leader */
CuAssertTrue(tc, 1 == raft_is_leader(r)); CuAssertTrue(tc, 1 == raft_is_leader(r));
CuAssertTrue(tc, 1 == raft_get_current_leader(r)); CuAssertTrue(tc, 1 == raft_get_leader_id(r));
memset(&ae, 0, sizeof(msg_appendentries_t)); memset(&ae, 0, sizeof(msg_appendentries_t));
ae.term = 6; ae.term = 6;
ae.leader_id = 2;
ae.prev_log_idx = 6; ae.prev_log_idx = 6;
ae.prev_log_term = 5; ae.prev_log_term = 5;
raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer); raft_recv_appendentries(r, raft_get_node(r, 2), &ae, &aer);
...@@ -3465,7 +3476,7 @@ void TestRaft_leader_recv_appendentries_steps_down_if_newer( ...@@ -3465,7 +3476,7 @@ void TestRaft_leader_recv_appendentries_steps_down_if_newer(
/* after more recent appendentries from node 2, node 1 should /* after more recent appendentries from node 2, node 1 should
* consider node 2 the leader. */ * consider node 2 the leader. */
CuAssertTrue(tc, 1 == raft_is_follower(r)); CuAssertTrue(tc, 1 == raft_is_follower(r));
CuAssertTrue(tc, 2 == raft_get_current_leader(r)); CuAssertTrue(tc, 2 == raft_get_leader_id(r));
} }
void TestRaft_leader_recv_appendentries_steps_down_if_newer_term( void TestRaft_leader_recv_appendentries_steps_down_if_newer_term(
...@@ -3942,6 +3953,200 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc) ...@@ -3942,6 +3953,200 @@ void TestRaft_leader_steps_down_if_there_is_no_quorum(CuTest * tc)
CuAssertTrue(tc, !raft_is_leader(r)); CuAssertTrue(tc, !raft_is_leader(r));
} }
void TestRaft_vote_for_unknown_node(CuTest * tc)
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
msg_requestvote_response_t resp;
msg_requestvote_t req = {
.term = 2,
.last_log_idx = 1,
.last_log_term = 1,
// not part of the configuration
.candidate_id = 500,
};
raft_recv_requestvote(r, NULL, &req, &resp);
CuAssertTrue(tc, resp.vote_granted == 1);
raft_destroy(r);
}
void TestRaft_recv_appendreq_from_unknown_node(CuTest * tc)
{
void *r = raft_new();
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_become_leader(r);
msg_appendentries_response_t resp;
msg_appendentries_t req = {
.term = 1,
// not part of the configuration
.leader_id = 10000,
};
// Receive the request and set node as leader
raft_recv_appendentries(r, NULL, &req, &resp);
CuAssertTrue(tc, resp.success == 1);
// Verify leader node and leader id returns correct value
CuAssertIntEquals(tc, 10000, raft_get_leader_id(r));
CuAssertPtrEquals(tc, NULL, raft_get_leader_node(r));
// Receive it again to verify everything is ok
resp = (msg_appendentries_response_t){0};
raft_recv_appendentries(r, NULL, &req, &resp);
CuAssertTrue(tc, resp.success == 1);
CuAssertIntEquals(tc, 10000, raft_get_leader_id(r));
CuAssertPtrEquals(tc, NULL, raft_get_leader_node(r));
raft_destroy(r);
}
void TestRaft_unknown_node_can_become_leader(CuTest * tc)
{
raft_cbs_t funcs = {
.log_get_node_id = __raft_log_get_node_id
};
void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_add_node(r, NULL, 1, 1);
raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
msg_appendentries_response_t resp;
raft_entry_t **entries = __MAKE_ENTRY_ARRAY(1, 1, "100");
entries[0]->type = RAFT_LOGTYPE_ADD_NONVOTING_NODE;
msg_appendentries_t req_append = {
.term = 1,
.prev_log_idx = 0,
.prev_log_term = 0,
.leader_id = 100,
.msg_id = 1,
.entries = entries,
.n_entries = 1
};
// Append the entry, this will set the node as leader
raft_recv_appendentries(r, NULL, &req_append, &resp);
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
CuAssertTrue(tc, resp.success == 1);
// Send another req with incremented leader_commit to commit the addition
msg_appendentries_t req_commit = {
.term = 1,
.prev_log_idx = 1,
.prev_log_term = 1,
.leader_id = 100,
.leader_commit = 1
};
raft_recv_appendentries(r, NULL, &req_commit, &resp);
// Validate added node is the leader
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
CuAssertTrue(tc, resp.success == 1);
// Validate added node is still the leader
raft_periodic(r, 1000);
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
// Promote node from non-voting to voting
raft_node_t *added = raft_get_node(r, req_commit.leader_id);
CuAssertIntEquals(tc, raft_node_get_id(added), raft_get_leader_id(r));
CuAssertTrue(tc, raft_node_is_active(added) == 1);
// Validate node is non-voter
CuAssertTrue(tc, raft_node_is_voting(added) == 0);
entries = __MAKE_ENTRY_ARRAY(1, 1, "100");
entries[0]->type = RAFT_LOGTYPE_ADD_NODE;
msg_appendentries_t req_add_node = {
.term = 1,
.leader_id = 100,
.msg_id = 1,
.entries = entries,
.n_entries = 1,
.prev_log_idx = 1,
.prev_log_term = 1,
};
raft_recv_appendentries(r, NULL, &req_add_node, &resp);
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
CuAssertTrue(tc, resp.success == 1);
raft_periodic(r, 1000);
CuAssertIntEquals(tc, raft_get_leader_id(r), req_append.leader_id);
CuAssertTrue(tc, resp.success == 1);
added = raft_get_node(r, req_commit.leader_id);
CuAssertIntEquals(tc, raft_node_get_id(added), raft_get_leader_id(r));
CuAssertTrue(tc, raft_node_is_active(added) == 1);
// Validate node is voter
CuAssertTrue(tc, raft_node_is_voting(added) == 1);
raft_destroy(r);
}
void TestRaft_removed_node_starts_election(CuTest * tc)
{
raft_cbs_t funcs = {
.log_get_node_id = __raft_log_get_node_id
};
void *r = raft_new();
raft_set_callbacks(r, &funcs, NULL);
raft_node_t *n1 = raft_add_node(r, NULL, 1, 1);
raft_node_t *n2 = raft_add_node(r, NULL, 2, 0);
raft_set_current_term(r, 1);
raft_become_leader(r);
msg_entry_response_t entry_resp;
raft_entry_t *entry = __MAKE_ENTRY(1, 1, "1");
entry->type = RAFT_LOGTYPE_REMOVE_NODE;
raft_recv_entry(r, entry, &entry_resp);
CuAssertTrue(r, raft_is_leader(r));
CuAssertTrue(r, !raft_node_is_active(n1));
CuAssertTrue(r, raft_node_is_active(n2));
CuAssertIntEquals(r, raft_get_num_nodes(r), 2);
raft_become_follower(r);
raft_set_election_timeout(r, 1000);
raft_periodic(r, 2000);
msg_requestvote_response_t reqresp = {
.vote_granted = 1,
.term = 2
};
raft_recv_requestvote_response(r, raft_get_node(r, 2), &reqresp);
CuAssertTrue(r, raft_is_leader(r));
CuAssertTrue(r, !raft_node_is_active(n1));
CuAssertTrue(r, raft_node_is_active(n2));
CuAssertIntEquals(r, raft_get_num_nodes(r), 2);
raft_destroy(r);
}
int main(void) int main(void)
{ {
CuString *output = CuStringNew(); CuString *output = CuStringNew();
...@@ -4068,6 +4273,10 @@ int main(void) ...@@ -4068,6 +4273,10 @@ int main(void)
SUITE_ADD_TEST(suite, TestRaft_single_node_commits_noop); SUITE_ADD_TEST(suite, TestRaft_single_node_commits_noop);
SUITE_ADD_TEST(suite, TestRaft_quorum_msg_id_correctness); SUITE_ADD_TEST(suite, TestRaft_quorum_msg_id_correctness);
SUITE_ADD_TEST(suite, TestRaft_leader_steps_down_if_there_is_no_quorum); SUITE_ADD_TEST(suite, TestRaft_leader_steps_down_if_there_is_no_quorum);
SUITE_ADD_TEST(suite, TestRaft_vote_for_unknown_node);
SUITE_ADD_TEST(suite, TestRaft_recv_appendreq_from_unknown_node);
SUITE_ADD_TEST(suite, TestRaft_unknown_node_can_become_leader);
SUITE_ADD_TEST(suite, TestRaft_removed_node_starts_election);
CuSuiteRun(suite); CuSuiteRun(suite);
CuSuiteDetails(suite, output); CuSuiteDetails(suite, output);
......
...@@ -224,14 +224,11 @@ def raft_notify_membership_event(raft, udata, node, ety, event_type): ...@@ -224,14 +224,11 @@ def raft_notify_membership_event(raft, udata, node, ety, event_type):
def raft_log(raft, node, udata, buf): def raft_log(raft, node, udata, buf):
server = ffi.from_handle(lib.raft_get_udata(raft)) server = ffi.from_handle(lib.raft_get_udata(raft))
# print(server.id, ffi.string(buf).decode('utf8'))
if node != ffi.NULL:
node = ffi.from_handle(lib.raft_node_get_udata(node))
# if server.id in [1] or (node and node.id in [1]): # if server.id in [1] or (node and node.id in [1]):
logger.info('{0}> {1}:{2}: {3}'.format( logger.info('{0}> {1}:{2}: {3}'.format(
server.network.iteration, server.network.iteration,
server.id, server.id,
node.id if node else '', node,
ffi.string(buf).decode('utf8'), ffi.string(buf).decode('utf8'),
)) ))
...@@ -251,7 +248,7 @@ class Network(object): ...@@ -251,7 +248,7 @@ class Network(object):
self.dupe_rate = 0 self.dupe_rate = 0
self.partition_rate = 0 self.partition_rate = 0
self.iteration = 0 self.iteration = 0
self.leader = None self.leader = -1
self.ety_id = 0 self.ety_id = 0
self.entries = [] self.entries = []
self.random = random.Random(seed) self.random = random.Random(seed)
...@@ -355,17 +352,17 @@ class Network(object): ...@@ -355,17 +352,17 @@ class Network(object):
# Count leadership changes # Count leadership changes
assert len(self.active_servers) > 0, self.diagnostic_info() assert len(self.active_servers) > 0, self.diagnostic_info()
leader_node = lib.raft_get_current_leader_node(self.active_servers[0].raft) leader = lib.raft_get_leader_id(self.active_servers[0].raft)
if leader_node: if leader != -1:
leader = ffi.from_handle(lib.raft_node_get_udata(leader_node)) if self.leader != leader:
if self.leader is not leader:
self.leadership_changes += 1 self.leadership_changes += 1
logger.info(f"old leader: {self.leader.id}") logger.info(f"old leader: {self.leader}")
logger.info(f"leader change: {lib.raft_get_nodeid(leader.raft)}") logger.info(f"leader change: {leader}")
self.diagnostic_info() self.diagnostic_info()
self.leader = leader self.leader = leader
def enqueue_msg(self, msg, sendor, sendee): def enqueue_msg(self, msg, sendor, sendee):
logger.debug(f"enqueue_msg: {sendor.id} to {sendee.id}") logger.debug(f"enqueue_msg: {sendor.id} to {sendee.id}")
# Drop message if this edge is partitioned # Drop message if this edge is partitioned
...@@ -509,7 +506,7 @@ class Network(object): ...@@ -509,7 +506,7 @@ class Network(object):
Add configuration change for leader's node Add configuration change for leader's node
""" """
server = self.active_servers[0] server = self.active_servers[0]
self.leader = server self.leader = server.id
server.set_connection_status(NODE_CONNECTING) server.set_connection_status(NODE_CONNECTING)
lib.raft_add_non_voting_node(server.raft, server.udata, server.id, 1) lib.raft_add_non_voting_node(server.raft, server.udata, server.id, 1)
...@@ -543,17 +540,25 @@ class Network(object): ...@@ -543,17 +540,25 @@ class Network(object):
if net.num_of_servers <= len(self.active_servers): if net.num_of_servers <= len(self.active_servers):
return return
if not self.leader: if self.leader == -1:
logger.info('add_member: no leader') logger.info('add_member: no leader')
return return
leader = self.leader leader = self.leader
leader_node = None
for server in self.active_servers:
if server.id == leader:
leader_node = server
if leader_node is None:
return
if not lib.raft_is_leader(leader.raft): if not lib.raft_is_leader(leader_node.raft):
return return
if lib.raft_voting_change_is_in_progress(leader.raft): if lib.raft_voting_change_is_in_progress(leader_node.raft):
logger.info('{} voting change in progress'.format(leader.id)) logger.info('{} voting change in progress'.format(leader_node.id))
return return
server = RaftServer(self) server = RaftServer(self)
...@@ -564,7 +569,7 @@ class Network(object): ...@@ -564,7 +569,7 @@ class Network(object):
ChangeRaftEntry(server.id)) ChangeRaftEntry(server.id))
assert(lib.raft_entry_is_cfg_change(ety)) assert(lib.raft_entry_is_cfg_change(ety))
e = leader.recv_entry(ety) e = leader_node.recv_entry(ety)
if 0 != e: if 0 != e:
logger.error(err2str(e)) logger.error(err2str(e))
return return
...@@ -578,21 +583,29 @@ class Network(object): ...@@ -578,21 +583,29 @@ class Network(object):
assert added_node assert added_node
def remove_member(self): def remove_member(self):
if not self.leader: if self.leader == -1:
logger.error('no leader') logger.error('no leader')
return return
leader = self.leader leader = self.leader
server = self.random.choice(self.active_servers) server = self.random.choice(self.active_servers)
leader_node = None
for sv in self.active_servers:
if sv.id == leader:
leader_node = sv
if leader_node is None:
return
if not lib.raft_is_leader(leader.raft): if not lib.raft_is_leader(leader_node.raft):
return return
if lib.raft_voting_change_is_in_progress(leader.raft): if lib.raft_voting_change_is_in_progress(leader_node.raft):
logger.info('{} voting change in progress'.format(server)) logger.info('{} voting change in progress'.format(server))
return return
if leader == server: if leader_node == server:
logger.info('can not remove leader') logger.info('can not remove leader')
return return
...@@ -610,7 +623,7 @@ class Network(object): ...@@ -610,7 +623,7 @@ class Network(object):
assert server.connection_status == NODE_CONNECTED assert server.connection_status == NODE_CONNECTED
assert(lib.raft_entry_is_cfg_change(ety)) assert(lib.raft_entry_is_cfg_change(ety))
e = leader.recv_entry(ety) e = leader_node.recv_entry(ety)
if 0 != e: if 0 != e:
logger.error(err2str(e)) logger.error(err2str(e))
return return
...@@ -775,7 +788,7 @@ class RaftServer(object): ...@@ -775,7 +788,7 @@ class RaftServer(object):
self.raft_logentry_get_node_id = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_get_node_id) self.raft_logentry_get_node_id = ffi.callback("int(raft_server_t*, void*, raft_entry_t*, raft_index_t)", raft_logentry_get_node_id)
self.raft_node_has_sufficient_logs = ffi.callback("int(raft_server_t* raft, void *user_data, raft_node_t* node)", raft_node_has_sufficient_logs) self.raft_node_has_sufficient_logs = ffi.callback("int(raft_server_t* raft, void *user_data, raft_node_t* node)", raft_node_has_sufficient_logs)
self.raft_notify_membership_event = ffi.callback("void(raft_server_t* raft, void *user_data, raft_node_t* node, raft_entry_t* ety, raft_membership_e)", raft_notify_membership_event) self.raft_notify_membership_event = ffi.callback("void(raft_server_t* raft, void *user_data, raft_node_t* node, raft_entry_t* ety, raft_membership_e)", raft_notify_membership_event)
self.raft_log = ffi.callback("void(raft_server_t*, raft_node_t*, void*, const char* buf)", raft_log) self.raft_log = ffi.callback("void(raft_server_t*, raft_node_id_t, void*, const char* buf)", raft_log)
def recv_entry(self, ety): def recv_entry(self, ety):
# FIXME: leak # FIXME: leak
...@@ -1140,7 +1153,7 @@ class RaftServer(object): ...@@ -1140,7 +1153,7 @@ class RaftServer(object):
"snapshot": lib.raft_get_snapshot_last_idx(self.raft), "snapshot": lib.raft_get_snapshot_last_idx(self.raft),
"removed": getattr(self, 'removed', False), "removed": getattr(self, 'removed', False),
"partitioned": partitioned_from, "partitioned": partitioned_from,
"leader": lib.raft_get_current_leader(self.raft), "leader": lib.raft_get_leader_id(self.raft),
} }
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment