分布式系統中Membership Change 源碼解讀

01

背景

在分布式系統的應用場景中，難免會出現增刪節點或者替換節點的需求，最簡單的解決方式就是臨時關閉集群，然后直接修改配置文件添加新的節點，完成后再將集群重新啟動，這樣的方式的確能達成我們的目的，但是其存在的問題也很明顯，變更期間集群是不可用的狀態，這對需要高可用性的系統來說是無法接受的，并且手動操作的過程還可能引發其他的錯誤，這也會降低系統的穩定性。因此，如何才能高效且安全的完成集群成員變更，也就成為了分布式系統開發過程中的關鍵問題。對于 Xline 來說，不僅要處理常規的變更流程，還要將其與 Curp 協議相結合，保證引入集群成員變更不會導致前端協議出錯。

02

動態成員變更的問題以及解決方案

由于 Xline 使用 Raft 作為后端協議，因此想要為 Xline 添加動態變更成員的能力，就需要解決 Raft 協議自身會遇到的問題。Raft 協議能夠正常工作的一個重要前提是，同一時間只能有一個 Leader，如果不加任何限制，直接向集群添加節點，那么就有可能破壞這個前提，具體情況如下圖所示：

由于網絡延遲等原因，無法保證各節點從切換到 的時間相同，就有可能出現圖中的情況，假設此時 Server 1 和 Server 5 同時開始選舉，Server 1 獲得了 Server 2 ?的投票，滿足 中的 quorum 需求，成為 Leader，Server 5 獲得了 Server 3 和 Server 4 的投票，滿足 中的 quorum 需求，Server 5 也成為 Leader，此時集群就會同時有兩個 Leader，產生一致性問題。 ? 為了解決這個問題，Raft 的作者提供了兩種解決思路。

Joint Consensus

單步成員變更

Joint Consensus Joint Consensus 本質上就是在成員變更過程中添加一個中間狀態。

Leader 收到成員變更請求時，會創建一條一個配置并通過 AppendEntries 同步到 Follower，收到 的節點會同時使用兩個配置做決策，也就是選舉等操作需要 和 都同意才視為成功 ，等到 ?commit 之后，Leader 會再創建 配置并同步給 Follower。 ? 在這種方案下，集群成員變更的中間狀態有以下幾種可能： ? 1.?? 創建之后提交之前，這個階段集群中可能同時存在 ， 兩種配置，在此階段內任意節點想要成為 Leader 都需要 配置同意，因此不會出現兩個 Leader。 ? 2.?提交之后，創建之前，這個階段集群中可能同時存在 ? 兩種配置，但只有使用 的節點才能夠成為 Leader，因為此階段 中的多數節點已經切換到了 配置，剩余還未切換的節點不足以選出新 Leader。 ? 3.?創建之后提交之前，這個階段集群中可能同時存在 三種配置，其中 配置無法選出 Leader，原因如上，和 中要選出 Leader，就需要 同意，此情況也不會出現兩個 Leader。 ? 4.?commit 之后，由 獨立決策，不會出現兩個 Leader。 ? ? 單步節點變更 除了 Joint Consensus 以外，還有一種方法可以安全的完成集群成員變更，那就是單步節點變更。該方法每次只會增加或減少一個節點，這種情況下，新舊配置的 majority 中必然有重疊節點，重疊的節點只能給一個節點投票，這樣就保證了不會同時存在兩個 Leader。復雜的變更行為就需要轉換成多次單步節點變更來完成。

這種方案沒有中間狀態，只需要一步操作就可以完成變更，邏輯上比 Joint Consensus 更加簡潔，沒有那么多復雜的中間狀態，實現起來也會簡單一點，當然它的功能也沒有 Joint Consensus 強大。 Xline 目前采用的方法就是單步成員變更，未來我們也會添加對 Joint Consensus 的支持。

03

Curp 協議的融合

Membership change 的主要流程，只通過后端的 Raft 就能夠完成，但是這個過程可能會擾亂前端 Curp 協議的流程。正常處理時，Curp client 會向集群中所有節點廣播 Propose 請求，并根據成功相應的數量是否大于當前集群成員數量的 superquorum 來判斷本次 propose 是否在 curp 中 commit，實現 membership change 以前，所有成員都在創建 client 時確定，但是引入了 membership change 之后，就需要有一種機制能夠保證 Client 在使用舊的配置時，也能夠探測到服務端使用的新配置，并且使用新配置重試當前的請求，否則可能導致 Curp 協議無法正常工作。

如圖所示，假設 Client 向一個三節點集群廣播 Propose，那么 Client 收到 3（3 節點的 superquorum） 個成功響應后就會認為這一次 Propose 已經在 Curp 中 commit，在此次 Propose 過程中，集群成員發生了變更，Server4 加入了集群，然而 4 節點的 superquorum 是 4，也就是說剛剛在 3 節點集群中被 curp commit 的請求，在成員變更之后就不再滿足 Curp 的 commit 條件了，這可能會導致已經返回給 Client 的請求丟失。 為了解決這個問題，我們為外部 client 發送的請求引入了一個新的字段 cluster_version ，這個字段表示集群當前使用配置的版本，每次執行成員變更，都會增加這個值，這樣 Server 端就能夠通過這個字段來判斷發送請求的 Client 是否在使用最新配置，并且直接拒絕使用舊配置的請求，Client 探測到 cluster_version 不一致后，就會主動拉取 Server 端的最新配置，并以最新配置發起新一輪的請求。在上述實例中，Propose 和成員變更同時發生時，Server1、2、3 中一定有節點已經在使用新配置了，那么該節點就會使用更大的 cluster_version 拒絕本次 Propose，Client 檢測到更大的 cluster_version 后，會重新向集群拉取當前的成員配置，然后以新配置重試整個請求。

04

源碼解讀

Leader 發起成員變更 開始變更成員的第一步，就是向 Leader 發送 ProposeConfChangeRequest，這個請求包含了本次 propose 要變更的節點信息和一些其他的輔助字段。 Server 端收到該請求后，首先會檢查請求攜帶的 cluster_version 是否和集群當前 cluster_version 匹配，不匹配的請求直接拒絕，然后才會進入 Server 端的處理邏輯:

///Handle`propose_conf_change`request
pub(super)fnhandle_propose_conf_change(
&self,
propose_id:ProposeId,
conf_changes:Vec,
)->Result<(),?CurpError>{
//...
self.check_new_config(&conf_changes)?;
letentry=log_w.push(st_r.term,propose_id,conf_changes.clone())?;
debug!("{}getsnewlog[{}]",self.id(),entry.index);
let(addrs,name,is_learner)=self.apply_conf_change(conf_changes);
self.ctx
.last_conf_change_idx
.store(entry.index,Ordering::Release);
let_ig=log_w.fallback_contexts.insert(
entry.index,
FallbackContext::clone(&entry),addrs,name,is_learner),
);
//...
}

Leader 節點在處理時，會通過 check_new_config 方法檢查本次 conf change 的有效性，提前拒絕一些無法處理的變更，比如插入一個已經存在的節點或者移除一個不存在的節點。檢查通過之后，就會進入和常規請求相同的流程，通過共識將其同步到所有 Follower 上，除了這部分相同流程以外， conf change 還需要做一些特殊處理，在其插入 log 之后，就會立刻應用新配置，并且記錄用于回退配置的上下文。這里和 Raft 論文中提到的方式相同，在節點擁有這條日志之后，不需要等待它 commit，就讓它立刻生效，在 Raft 中沒有 commit 的日志是有被覆蓋的可能的，因此才需要記錄上下文，如果日志被覆蓋，就能夠通過這個上下文來回退本次修改。 Follower 處理成員變更 對于 Follower 節點，成員變革的主要邏輯發生在 handle_append_entries 中，這個方法被用來處理 Leader 發送的日志，其中就包括 conf change

pub(super)fnhandle_append_entries(
&self,
term:u64,
leader_id:ServerId,
prev_log_index:LogIndex,
prev_log_term:u64,
entries:Vec>,
leader_commit:LogIndex,
)->Result{
//...
//appendlogentries
letmutlog_w=self.log.write();
let(cc_entries,fallback_indexes)=log_w
.try_append_entries(entries,prev_log_index,prev_log_term)
.map_err(|_ig|(term,log_w.commit_index+1))?;
//fallbackoverwrittenconfchangeentries
foridxinfallback_indexes.iter().sorted().rev(){
letinfo=log_w.fallback_contexts.remove(idx).unwrap_or_else(||{
unreachable!("fall_back_infosshouldcontaintheentryneedtofallback")
});
letEntryData::ConfChange(refconf_change)=info.origin_entry.entry_dataelse{
unreachable!("theentryinthefallback_infoshouldbeconfchangeentry");
};
letchanges=conf_change.clone();
self.fallback_conf_change(changes,info.addrs,info.name,info.is_learner);
}
//applyconfchangeentries
foreincc_entries{
letEntryData::ConfChange(refcc)=e.entry_dataelse{
unreachable!("cc_entryshouldbeconfchangeentry");
};
let(addrs,name,is_learner)=self.apply_conf_change(cc.clone());
let_ig=log_w.fallback_contexts.insert(
e.index,
FallbackContext::clone(&e),addrs,name,is_learner),
);
}
//...
}

對于常規日志的處理此處直接省略，不再贅述。Follower 在嘗試追加 Leader 發來的日志時，會判斷當前節點上有哪些新的 conf change 日志，以及哪些沒有被 commit 的 conf change 會被覆蓋，然后通過預先記錄的上下文，將被覆蓋的變更逆序回退，并且應用新的變更，在應用新變更時，也需要在此處記錄新變更的上下文。 成員變更日志的 commit

asyncfnworker_as,RC:RoleChange>(
entry:Arc>,
prepare:Option,
ce:&CE,
curp:&RawCurp,
)->bool{
//...
letsuccess=matchentry.entry_data{
EntryData::ConfChange(refconf_change)=>{
//...
letshutdown_self=
change.change_type()==ConfChangeType::Remove&&change.node_id==id;
//...
ifshutdown_self{
curp.shutdown_trigger().self_shutdown();
}
true
}
_=>//...
};
ce.trigger(entry.inflight_id(),entry.index);
success
}

在 Conf change 被 commit 之后的 after sync 階段，除了一些常規操作以外，還需要判斷被 commit 的 conf change是否將當前節點 remove，如果這個節點被 remove 的話，就需要在此處開始 shutdown 當前節點，一般只有 leader 節點會執行到此處并將 remove 自身的日志 commit，在其 shutdown 自身后，剩余節點會選出一個擁有最新日志的 leader。 New Node 加入集群 為了區分創建新集群運行的節點，和新啟動的需要加入現有集群的節點，需要在啟動時傳入一個新的參數 `InitialClusterState`，這是一個枚舉類型，只有兩個成員， `InitialClusterState::New` 表示本次啟動的節點是新啟動集群的成員之一；`InitialClusterState::Existing` 表示本次啟動的節點是要加入已有集群的新節點。

letcluster_info=match*cluster_config.initial_cluster_state(){
InitialClusterState::New=>init_cluster_info,
InitialClusterState::Existing=>get_cluster_info_from_remote(
&init_cluster_info,
server_addr_str,
&name,
Duration::from_secs(3),
)
.await
.ok_or_else(||anyhow!("Failedtogetclusterinfofromremote"))?,
_=>unreachable!("xlineonlysupportstwoinitialclusterstates:new,existing"),
};

這兩種方式的本質區別在于，新建集群是各節點初始的集群成員是相同的，可以直接通過這部分初始信息各自計算出一個全局統一的節點 ID，保證每個節點都有一個唯一 ID，而加入現有集群時，新節點不能自己計算節點的 ID，需要通過 get_cluster_info_from_remote 方法去拉取現有集群的信息，直接繼承現有集群正在使用的 ID 和其他信息，以保證集群內 ID 和節點的對應關系，避免出現 ID 重復或一個節點有多個 ID 的情況。 為保證與 etcd 接口的兼容，新節點開始運行前時沒有 name 的，etcdctl 中會根據 name 是否為空來判斷相應節點是否已啟動，在新節點啟動并加入集群后，會向 Leader 發送一個 Publish Rpc，用來在集群內發布自己的 name。 Node remove 假設我們在 remove 一個節點之后，不將其關閉，那么它將會選舉超時并向其余節點發送 Vote 請求，浪費其他節點的網絡和 CPU 資源，想要解決這個問題，首先能想到的有兩個辦法：

在節點應用會 remove 自身的新配置后，立刻關閉自身節點。很明顯，這種方案一定是不可行的，因為在應用新配置時，這一條日志還沒有被 commit，還有被回退的可能，如果在此處關閉自身，那假如配置變更發生了回退，被 remove 的這個節點就已經被關閉無法直接回復了，這不是我們想看到的結果。

在節點 commit 會 remove 自身的日志后，立刻關閉自身節點。因為已經被 commit，所以這種方法時沒有上述問題的，但是據此實現后就會發現，被 remove 的節點有時還是不能自動關閉。因為被 remove 的節點可能根本不會 commit 新配置，假設我們要 remove 一個 Follower 節點，Leader 講這一條 remove 記錄添加到自己的日志之后，立刻開始使用新日志，此時 Leader 已經不會向這個 Follower 發送任何請求了，Follower 自然也不可能commit 這條日志并關閉自身。這個問題在 Leader 上是不存在的，Leader 將會臨時管理不包含自己的集群，直到日志被 commit。

最直接的方法都不能使用，那被 remove 的節點應該如何關閉自身呢？假設我們不添加這里的關閉邏輯，會發生什么事？Leader 向集群同步 conf change 日志，新集群的所有成員都會正常處理并 commit 這條日志，被 remove 的節點會在自己不知情的請款下離開原集群，收不到 Leader 的心跳，這個節點就會超時并開始選舉，這里也就是我們最終我們決定修改的位置。

pub(super)fnhandle_pre_vote(
&self,
term:u64,
candidate_id:ServerId,
last_log_index:LogIndex,
last_log_term:u64,
)->Result<(u64,?Vec>),Option>{
//...
letcontains_candidate=self.cluster().contains(candidate_id);
letremove_candidate_is_not_committed=
log_r
.fallback_contexts
.iter()
.any(|(_,ctx)|matchctx.origin_entry.entry_data{
EntryData::ConfChange(refcc)=>cc.iter().any(|c|{
matches!(c.change_type(),ConfChangeType::Remove)
&&c.node_id==candidate_id
}),
_=>false,
});
//extrachecktoshutdownremovednode
if!contains_candidate&&!remove_candidate_is_not_committed{
returnErr(None);
}
//...
}

我們在 ProVote 階段加入了額外的檢查邏輯，收到 pre-vote 的節點會檢查 candidate 是否已經被 Remove，假設 candidate 不在當前節點的配置中，并且可能會進行的回退操作，也不會讓這個節點重新加入集群，那就說明這是一個已經被 Remove 的 candidate，此時處理請求的節點將會給 Follower 回復一個帶有 shutdown_candidate 字段的特殊 VoteResponse。Candidate 在收到該響應后，會判斷 shutdown_candidate 是否為 true，為 true 則開始關閉自身，不為 true則繼續選舉流程。

05

總結

本篇文章我們深入探討了在分布式系統中如何進行集群成員變更，簡單介紹了兩種主要的解決方案：Joint Consensus 和單步成員變更，Joint Consensus 通過引入中間狀態來保證變更期間不會出現兩個 Leader，單步集群變更則是犧牲了一定的功能，通過逐個變更節點來簡化實現邏輯。并且對 Xline 目前使用的單步成員變更方案進行了源碼級的分析，展示了 Leader 和 Follower 都是如何處理變更的，以及引入集群變更之后，會有哪些新邏輯需要處理。 目前 Xline 對集群成員變更的處理僅使用了單步集群變更這一種方法，提供了基本的變更能力，未來我們還會嘗試支持 Joint Consensus，增強 Xline 的功能。

審核編輯：黃飛