Unverified Commit 2b7f9c7f authored by Shaya Potter's avatar Shaya Potter Committed by GitHub
Browse files

RR-238 - Recording and Retrieving of RAFT RPC stats



Record counts for raft RPC stats, sending / retrieving / success / failure and the like.

cover append entry, requestvote (including prevote) and snapshots

---------
Co-authored-by: default avatarOzan Tezcan <ozantezcan@gmail.com>
parent 3973759e
...@@ -85,6 +85,37 @@ typedef enum { ...@@ -85,6 +85,37 @@ typedef enum {
RAFT_LOGTYPE_NUM = 100, RAFT_LOGTYPE_NUM = 100,
} raft_logtype_e; } raft_logtype_e;
typedef struct raft_server_stats {
/** Miscellaneous */
unsigned long long appendentries_req_with_entry;
unsigned long long snapshots_created;
unsigned long long snapshots_received;
unsigned long long exec_throttled;
/** Message types */
unsigned long long appendentries_req_sent;
unsigned long long appendentries_req_received;
unsigned long long appendentries_req_failed;
unsigned long long appendentries_resp_received;
unsigned long long snapshot_req_sent;
unsigned long long snapshot_req_received;
unsigned long long snapshot_req_failed;
unsigned long long snapshot_resp_received;
unsigned long long requestvote_prevote_req_sent;
unsigned long long requestvote_prevote_req_received;
unsigned long long requestvote_prevote_req_failed;
unsigned long long requestvote_prevote_req_granted;
unsigned long long requestvote_prevote_resp_received;
unsigned long long requestvote_req_sent;
unsigned long long requestvote_req_received;
unsigned long long requestvote_req_failed;
unsigned long long requestvote_req_granted;
unsigned long long requestvote_resp_received;
} raft_server_stats_t;
/** Entry that is stored in the server's entry log. */ /** Entry that is stored in the server's entry log. */
typedef struct raft_entry typedef struct raft_entry
{ {
...@@ -1459,6 +1490,13 @@ int raft_transfer_leader(raft_server_t* me, raft_node_id_t node_id, long timeout ...@@ -1459,6 +1490,13 @@ int raft_transfer_leader(raft_server_t* me, raft_node_id_t node_id, long timeout
*/ */
raft_node_id_t raft_get_transfer_leader(raft_server_t *me); raft_node_id_t raft_get_transfer_leader(raft_server_t *me);
/** Retrieves collected raft stats
*
* @param[in] me The Raft raft_server_t
* @param[out] stats a pointer to a client allocated buffer to fill collected stats
*/
void raft_get_server_stats(raft_server_t *me, raft_server_stats_t *stats);
/** Force this server to start an election /** Force this server to start an election
* *
* As the last step of leader transfer, current leader calls * As the last step of leader transfer, current leader calls
......
...@@ -131,6 +131,8 @@ struct raft_server { ...@@ -131,6 +131,8 @@ struct raft_server {
int auto_flush; /* Automatically call raft_flush() */ int auto_flush; /* Automatically call raft_flush() */
int log_enabled; /* Enable library logs */ int log_enabled; /* Enable library logs */
int disable_apply; /* Do not apply entries, useful for testing */ int disable_apply; /* Do not apply entries, useful for testing */
raft_server_stats_t stats;
}; };
int raft_election_start(raft_server_t *me, int skip_precandidate); int raft_election_start(raft_server_t *me, int skip_precandidate);
......
...@@ -731,6 +731,12 @@ int raft_recv_appendentries_response(raft_server_t *me, ...@@ -731,6 +731,12 @@ int raft_recv_appendentries_response(raft_server_t *me,
raft_get_nodeid(me), raft_node_get_id(node), raft_get_nodeid(me), raft_node_get_id(node),
resp->msg_id, resp->term, resp->success, resp->current_idx); resp->msg_id, resp->term, resp->success, resp->current_idx);
me->stats.appendentries_resp_received++;
if (resp->success == 0) {
me->stats.appendentries_req_failed++;
}
if (!node || !raft_is_leader(me)) { if (!node || !raft_is_leader(me)) {
return 0; return 0;
} }
...@@ -827,6 +833,11 @@ int raft_recv_appendentries(raft_server_t *me, ...@@ -827,6 +833,11 @@ int raft_recv_appendentries(raft_server_t *me,
req->leader_id, req->msg_id, req->term, req->prev_log_idx, req->leader_id, req->msg_id, req->term, req->prev_log_idx,
req->prev_log_term, req->leader_commit, req->n_entries); req->prev_log_term, req->leader_commit, req->n_entries);
me->stats.appendentries_req_received++;
if (req->n_entries) {
me->stats.appendentries_req_with_entry++;
}
resp->msg_id = req->msg_id; resp->msg_id = req->msg_id;
resp->success = 0; resp->success = 0;
...@@ -990,6 +1001,9 @@ int raft_recv_requestvote(raft_server_t *me, ...@@ -990,6 +1001,9 @@ int raft_recv_requestvote(raft_server_t *me,
req->prevote, req->term, req->candidate_id, req->last_log_idx, req->prevote, req->term, req->candidate_id, req->last_log_idx,
req->last_log_term); req->last_log_term);
req->prevote ? me->stats.requestvote_prevote_req_received++ :
me->stats.requestvote_req_received++;
resp->prevote = req->prevote; resp->prevote = req->prevote;
resp->request_term = req->term; resp->request_term = req->term;
resp->vote_granted = 0; resp->vote_granted = 0;
...@@ -1051,6 +1065,11 @@ int raft_recv_requestvote(raft_server_t *me, ...@@ -1051,6 +1065,11 @@ int raft_recv_requestvote(raft_server_t *me,
} }
done: done:
if (resp->vote_granted) {
resp->prevote ? me->stats.requestvote_prevote_req_granted++ :
me->stats.requestvote_req_granted++;
}
resp->term = me->current_term; resp->term = me->current_term;
raft_log(me, "%d --> %d, sent requestvote_resp " raft_log(me, "%d --> %d, sent requestvote_resp "
...@@ -1078,6 +1097,14 @@ int raft_recv_requestvote_response(raft_server_t *me, ...@@ -1078,6 +1097,14 @@ int raft_recv_requestvote_response(raft_server_t *me,
raft_get_nodeid(me), raft_node_get_id(node), raft_get_nodeid(me), raft_node_get_id(node),
resp->prevote, resp->request_term, resp->term, resp->vote_granted); resp->prevote, resp->request_term, resp->term, resp->vote_granted);
resp->prevote ? me->stats.requestvote_prevote_resp_received++ :
me->stats.requestvote_resp_received++;
if (resp->vote_granted == 0) {
resp->prevote ? me->stats.requestvote_prevote_req_failed++ :
me->stats.requestvote_req_failed++;
}
if (resp->term > me->current_term) { if (resp->term > me->current_term) {
int e = raft_set_current_term(me, resp->term); int e = raft_set_current_term(me, resp->term);
if (e != 0) { if (e != 0) {
...@@ -1201,6 +1228,9 @@ int raft_send_requestvote(raft_server_t *me, raft_node_t *node) ...@@ -1201,6 +1228,9 @@ int raft_send_requestvote(raft_server_t *me, raft_node_t *node)
req.last_log_term); req.last_log_term);
if (me->cb.send_requestvote) { if (me->cb.send_requestvote) {
req.prevote ? me->stats.requestvote_prevote_req_sent++ :
me->stats.requestvote_req_sent++;
e = me->cb.send_requestvote(me, me->udata, node, &req); e = me->cb.send_requestvote(me, me->udata, node, &req);
} }
...@@ -1359,6 +1389,7 @@ int raft_send_snapshot(raft_server_t *me, raft_node_t *node) ...@@ -1359,6 +1389,7 @@ int raft_send_snapshot(raft_server_t *me, raft_node_t *node)
req.snapshot_term, req.chunk.offset, req.chunk.len, req.snapshot_term, req.chunk.offset, req.chunk.len,
req.chunk.last_chunk); req.chunk.last_chunk);
me->stats.snapshot_req_sent++;
e = me->cb.send_snapshot(me, me->udata, node, &req); e = me->cb.send_snapshot(me, me->udata, node, &req);
if (e != 0) { if (e != 0) {
return e; return e;
...@@ -1388,6 +1419,8 @@ int raft_recv_snapshot(raft_server_t *me, ...@@ -1388,6 +1419,8 @@ int raft_recv_snapshot(raft_server_t *me,
req->snapshot_term, req->chunk.offset, req->chunk.len, req->snapshot_term, req->chunk.offset, req->chunk.len,
req->chunk.last_chunk); req->chunk.last_chunk);
me->stats.snapshot_req_received++;
resp->msg_id = req->msg_id; resp->msg_id = req->msg_id;
resp->last_chunk = req->chunk.last_chunk; resp->last_chunk = req->chunk.last_chunk;
resp->offset = 0; resp->offset = 0;
...@@ -1476,6 +1509,12 @@ int raft_recv_snapshot_response(raft_server_t *me, ...@@ -1476,6 +1509,12 @@ int raft_recv_snapshot_response(raft_server_t *me,
raft_get_nodeid(me), raft_node_get_id(node), resp->msg_id, raft_get_nodeid(me), raft_node_get_id(node), resp->msg_id,
resp->term, resp->success, resp->offset, resp->last_chunk); resp->term, resp->success, resp->offset, resp->last_chunk);
me->stats.snapshot_resp_received++;
if (resp->success == 0) {
me->stats.snapshot_req_failed++;
}
if (!node || !raft_is_leader(me)) { if (!node || !raft_is_leader(me)) {
return 0; return 0;
} }
...@@ -1573,6 +1612,7 @@ int raft_send_appendentries(raft_server_t *me, raft_node_t *node) ...@@ -1573,6 +1612,7 @@ int raft_send_appendentries(raft_server_t *me, raft_node_t *node)
req.leader_id, req.msg_id, req.term, req.prev_log_idx, req.leader_id, req.msg_id, req.term, req.prev_log_idx,
req.prev_log_term, req.leader_commit, req.n_entries); req.prev_log_term, req.leader_commit, req.n_entries);
me->stats.appendentries_req_sent++;
e = me->cb.send_appendentries(me, me->udata, node, &req); e = me->cb.send_appendentries(me, me->udata, node, &req);
raft_node_set_next_idx(node, next_idx + req.n_entries); raft_node_set_next_idx(node, next_idx + req.n_entries);
...@@ -1662,6 +1702,22 @@ int raft_msg_entry_response_committed(raft_server_t* me, ...@@ -1662,6 +1702,22 @@ int raft_msg_entry_response_committed(raft_server_t* me,
return r->idx <= me->commit_idx; return r->idx <= me->commit_idx;
} }
int raft_pending_operations(raft_server_t *me)
{
return me->pending_operations;
}
static void raft_set_pending(raft_server_t *me)
{
me->stats.exec_throttled++;
me->pending_operations = 1;
}
static void raft_unset_pending(raft_server_t *me)
{
me->pending_operations = 0;
}
int raft_apply_all(raft_server_t *me) int raft_apply_all(raft_server_t *me)
{ {
if (!raft_is_apply_allowed(me)) { if (!raft_is_apply_allowed(me)) {
...@@ -1670,7 +1726,7 @@ int raft_apply_all(raft_server_t *me) ...@@ -1670,7 +1726,7 @@ int raft_apply_all(raft_server_t *me)
while (me->commit_idx > me->last_applied_idx) { while (me->commit_idx > me->last_applied_idx) {
if (raft_time_millis(me) > me->exec_deadline) { if (raft_time_millis(me) > me->exec_deadline) {
me->pending_operations = 1; raft_set_pending(me);
return 0; return 0;
} }
...@@ -1798,6 +1854,8 @@ int raft_end_snapshot(raft_server_t *me) ...@@ -1798,6 +1854,8 @@ int raft_end_snapshot(raft_server_t *me)
me->log_impl->first_idx(me->log) - 1, me->commit_idx, me->log_impl->first_idx(me->log) - 1, me->commit_idx,
raft_get_current_idx(me)); raft_get_current_idx(me));
me->stats.snapshots_created++;
if (!raft_is_leader(me)) if (!raft_is_leader(me))
return 0; return 0;
...@@ -1889,6 +1947,8 @@ int raft_end_load_snapshot(raft_server_t *me) ...@@ -1889,6 +1947,8 @@ int raft_end_load_snapshot(raft_server_t *me)
raft_node_set_addition_committed(me->nodes[i], 1); raft_node_set_addition_committed(me->nodes[i], 1);
} }
me->stats.snapshots_received++;
raft_log(me, "loaded snapshot sli:%ld slt:%ld", raft_log(me, "loaded snapshot sli:%ld slt:%ld",
me->snapshot_last_idx, me->snapshot_last_term); me->snapshot_last_idx, me->snapshot_last_term);
...@@ -2018,7 +2078,7 @@ void raft_process_read_queue(raft_server_t *me) ...@@ -2018,7 +2078,7 @@ void raft_process_read_queue(raft_server_t *me)
me->read_queue_head->read_idx <= me->last_applied_idx) { me->read_queue_head->read_idx <= me->last_applied_idx) {
if (raft_time_millis(me) > me->exec_deadline) { if (raft_time_millis(me) > me->exec_deadline) {
me->pending_operations = 1; raft_set_pending(me);
return; return;
} }
...@@ -2259,18 +2319,13 @@ int raft_config(raft_server_t *me, int set, raft_config_e config, ...) ...@@ -2259,18 +2319,13 @@ int raft_config(raft_server_t *me, int set, raft_config_e config, ...)
return ret; return ret;
} }
int raft_pending_operations(raft_server_t *me)
{
return me->pending_operations;
}
int raft_exec_operations(raft_server_t *me) int raft_exec_operations(raft_server_t *me)
{ {
/* Operations should take less than `request-timeout`. If we block here more /* Operations should take less than `request-timeout`. If we block here more
* than request-timeout, it means this server won't generate responses on * than request-timeout, it means this server won't generate responses on
* time for the existing requests. */ * time for the existing requests. */
me->exec_deadline = raft_time_millis(me) + (me->request_timeout / 2); me->exec_deadline = raft_time_millis(me) + (me->request_timeout / 2);
me->pending_operations = 0; raft_unset_pending(me);
int e = raft_apply_all(me); int e = raft_apply_all(me);
if (e != 0) { if (e != 0) {
......
...@@ -288,3 +288,7 @@ raft_node_id_t raft_get_transfer_leader(raft_server_t *me) ...@@ -288,3 +288,7 @@ raft_node_id_t raft_get_transfer_leader(raft_server_t *me)
return me->node_transferring_leader_to; return me->node_transferring_leader_to;
} }
void raft_get_server_stats(raft_server_t *me, raft_server_stats_t *stats)
{
*stats = me->stats;
}
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