Support learner change

Previously, learner is a static srv config, provide a cmd for use to flip the learner flag.

Author: yuwmao

include/libnuraft/raft_server.hxx

@@ -260,6 +260,18 @@ public:
     ptr< cmd_result< ptr<buffer> > >
         append_entries(const std::vector< ptr<buffer> >& logs);
 
+    /**
+     * Flip learner flag of given server.
+     * Learner will be excluded from the quorum.
+     * Only leader will accept this operation.
+     * This is also an asynchronous task.
+     *
+     * @param srv_id ID of the server to set as a learner.
+     * @param to If `true`, set the server as a learner, otherwise, clear learner flag.
+     * @return `ret->get_result_code()` will be OK on success.
+     */
+    ptr<cmd_result<ptr<buffer>>> flip_learner_flag(int32 srv_id, bool to);
+
     /**
      * Parameters for `req_ext_cb` callback function.
      */

include/libnuraft/srv_config.hxx

@@ -77,6 +77,8 @@ public:
 
     bool is_learner() const { return learner_; }
 
+    void set_learner(bool to) { learner_ = to; }
+
     bool is_new_joiner() const { return new_joiner_; }
 
     void set_new_joiner(bool to) { new_joiner_ = to; }

src/handle_user_cmd.cxx

@@ -109,6 +109,64 @@ ptr< cmd_result< ptr<buffer> > > raft_server::append_entries_ext
     return send_msg_to_leader(req, ext_params);
 }
 
+
+ptr< cmd_result< ptr<buffer> > > raft_server::flip_learner_flag(int32 srv_id, bool to)
+{
+    ptr<cmd_result<ptr<buffer>>> ret = cs_new<cmd_result<ptr<buffer>>>(nullptr);
+    if (role_ != srv_role::leader || write_paused_) {
+        p_er("this is not a leader, cannot handle flip_learner_flag");
+        ret->set_result_code(cmd_result_code::NOT_LEADER);
+        return ret;
+    }
+    if (config_changing_) {
+        p_wn("previous config has not committed yet");
+        ret->set_result_code(cmd_result_code::CONFIG_CHANGING);
+        return ret;
+    }
+
+    ptr<cluster_config> cur_conf = get_config();
+    ptr<buffer> enc_conf_buf = cur_conf->serialize();
+    ptr<cluster_config> new_conf = cluster_config::deserialize(*enc_conf_buf);
+    new_conf->set_log_idx(log_store_->next_slot());
+
+    bool found = false;
+    for (auto& ss: new_conf->get_servers()) {
+        if (ss->get_id() != srv_id) {
+            continue;
+        }
+        found = true;
+        if (ss->is_learner() == to) {
+            p_in("server %d already has learner flag set to %s\n",
+                 srv_id,
+                 to ? "true" : "false");
+            ret->set_result_code(cmd_result_code::OK);
+            return ret;
+        }
+        p_in("set learner flag to %s for server %d",
+                 to ? "true" : "false", srv_id);
+        ss->set_learner(to);
+        break;
+    }
+    if (!found) {
+        p_er("server %d not found", srv_id);
+        ret->set_result_code(SERVER_NOT_FOUND);
+        return ret;
+    }
+    p_in("copy new conf buf");
+    ptr<buffer> new_conf_buf(new_conf->serialize());
+    ptr<log_entry> entry(cs_new<log_entry>(state_->get_term(),
+                                           new_conf_buf,
+                                           log_val_type::conf,
+                                           timer_helper::get_timeofday_us()));
+    store_log_entry(entry);
+    config_changing_ = true;
+    uncommitted_config_ = new_conf;
+    p_in("request append entries");
+    request_append_entries();
+    ret->set_result_code(OK);
+    return ret;
+}
+
 ptr< cmd_result< ptr<buffer> > > raft_server::send_msg_to_leader
                                  ( ptr<req_msg>& req,
                                    const req_ext_params& ext_params )

src/raft_server.cxx

@@ -759,7 +759,6 @@ ptr<resp_msg> raft_server::process_req(req_msg& req,
 
     } else if (req.get_type() == msg_type::priority_change_request) {
         resp = handle_priority_change_req(req);
-
     } else {
         // extended requests
         resp = handle_ext_msg(req, guard);

tests/unit/asio_service_test.cxx

@@ -2409,6 +2409,58 @@ int full_consensus_test() {
     return 0;
 }
 
+
+int flip_learner_flag_test() {
+    reset_log_files();
+
+    std::string s1_addr = "tcp://127.0.0.1:20010";
+    std::string s2_addr = "tcp://127.0.0.1:20020";
+    std::string s3_addr = "tcp://127.0.0.1:20030";
+
+    RaftAsioPkg s1(1, s1_addr);
+    RaftAsioPkg s2(2, s2_addr);
+    RaftAsioPkg s3(3, s3_addr);
+    std::vector<RaftAsioPkg*> pkgs = {&s1, &s2, &s3};
+
+    _msg("launching asio-raft servers\n");
+    CHK_Z( launch_servers(pkgs, false) );
+    _msg("organizing raft group\n");
+    CHK_Z( make_group(pkgs) );
+    // Set async.
+    for (auto& entry: pkgs) {
+        RaftAsioPkg* pp = entry;
+        raft_params param = pp->raftServer->get_current_params();
+        param.return_method_ = raft_params::async_handler;
+        pp->raftServer->update_params(param);
+    }
+    // Set to learner.
+    ptr<cmd_result<ptr<buffer>>> result = s1.raftServer->flip_learner_flag(s3.myId, true);
+    CHK_EQ(cmd_result_code::OK, result->get_result_code());
+    TestSuite::sleep_ms(RaftAsioPkg::HEARTBEAT_MS * 5, "wait for replication");
+    for (auto& entry: pkgs) {
+        RaftAsioPkg* pp = entry;
+        ptr<cluster_config> conf = pp->raftServer->get_config();
+        CHK_TRUE(conf->get_server(s3.myId)->is_learner());
+    }
+    // Clear leaner.
+    result = s1.raftServer->flip_learner_flag(s3.myId, false);
+    CHK_EQ(cmd_result_code::OK, result->get_result_code());
+    TestSuite::sleep_ms(RaftAsioPkg::HEARTBEAT_MS * 5, "wait for replication");
+    for (auto& entry: pkgs) {
+        RaftAsioPkg* pp = entry;
+        ptr<cluster_config> conf = pp->raftServer->get_config();
+        CHK_FALSE(conf->get_server(s3.myId)->is_learner());
+    }
+
+    s1.raftServer->shutdown();
+    s2.raftServer->shutdown();
+    s3.raftServer->shutdown();
+    TestSuite::sleep_sec(1, "shutting down");
+
+    SimpleLogger::shutdown();
+    return 0;
+}
+
 int custom_commit_condition_test() {
     reset_log_files();
 
@@ -2916,6 +2968,9 @@ int main(int argc, char** argv) {
     ts.doTest( "full consensus test",
                full_consensus_test );
 
+    ts.doTest("flip learner flag test",
+           flip_learner_flag_test);
+
     ts.doTest( "custom commit condition test",
                custom_commit_condition_test );