聊聊 Redis 哨兵選舉與故障轉(zhuǎn)移的實現(xiàn)
上一篇文章我們將哨兵主觀下線的核心流程都分析完成，這一篇我們將接著之前的思路，將哨兵獲取客觀下線結(jié)果并結(jié)合raft協(xié)議完成哨兵leader選舉完成故障轉(zhuǎn)移的流程分析完成，希望對你有幫助。
詳解哨兵選舉與故障轉(zhuǎn)移流程
1. 獲取客觀下線結(jié)果判斷
當(dāng)前哨兵主觀認(rèn)定master下線之后，為了明確知曉master節(jié)點是否真的下線，哨兵節(jié)點還會通過cc即異步命令指針?biāo)S護的socket連接發(fā)起is-master-down-by-addr的sentinel指令進(jìn)行詢問，其他哨兵所回復(fù)的結(jié)果都會通過回調(diào)函數(shù)sentinelReceiveIsMasterDownReply函數(shù)處理。
這段請求最終會被其他哨兵sentinel命令所對應(yīng)的函數(shù)sentinelCommand執(zhí)行，他們各自會在內(nèi)部查看自己對于master判斷是否是主觀下線，如果是則返回1。
最后我們的哨兵收到這個結(jié)果1，則通過位運算加master節(jié)點狀態(tài)flags類加上客觀下線的判斷標(biāo)識64，這里redis為了提升運算效率，采用的二進(jìn)制|=運算，這一點我們在閱讀大量的redis中源碼都會看到二進(jìn)制運算這一點優(yōu)化：
對此我們也給出哨兵處理每一個master實例的函數(shù)入口，可以看到在調(diào)用sentinelCheckSubjectivelyDown完成主觀下線的檢查之后，又會調(diào)用sentinelAskMasterStateToOtherSentinels并傳入SENTINEL_NO_FLAGS即僅僅檢查其他哨兵對于當(dāng)前master的主觀判斷結(jié)果:
//這個入?yún)∩诒鴮嵗彤?dāng)前主節(jié)點的從節(jié)點信息
void sentinelHandleRedisInstance(sentinelRedisInstance *ri) {
//......
//3. 主觀判斷是否下線
sentinelCheckSubjectivelyDown(ri);
//......
/* Only masters */
if (ri->flags & SRI_MASTER) {
//......
//傳入master信息ri以及標(biāo)識SENTINEL_NO_FLAGS意味僅了解其他哨兵對于master節(jié)點狀態(tài)的判斷
sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_NO_FLAGS);
}
}步入sentinelAskMasterStateToOtherSentinels即可看到哨兵詢問其他哨兵對于master判斷的邏輯，可以看到它遍歷出每一個哨兵實例，通過異步連接cc指針?biāo)赶虻倪B接發(fā)起SENTINEL is-master-down-by-addr指令獲取其他哨兵節(jié)點對于master下線的看法，并注冊sentinelReceiveIsMasterDownReply函數(shù)處理返回結(jié)果：
#define SENTINEL_ASK_FORCED (1<<0)
void sentinelAskMasterStateToOtherSentinels(sentinelRedisInstance *master, int flags) {
dictIterator *di;
dictEntry *de;
di = dictGetIterator(master->sentinels);
//遍歷哨兵實例
while((de = dictNext(di)) != NULL) {
sentinelRedisInstance *ri = dictGetVal(de);
//......
/* Ask */
ll2string(port,sizeof(port),master->addr->port);
//發(fā)送is-master-down-by-addr命令獲取其他哨兵客觀下線的結(jié)果，并通過sentinelReceiveIsMasterDownReply作為回調(diào)處理接收結(jié)果
retval = redisAsyncCommand(ri->cc,
sentinelReceiveIsMasterDownReply, NULL,
"SENTINEL is-master-down-by-addr %s %s %llu %s",
master->addr->ip, port,
sentinel.current_epoch,
//若大于SENTINEL_FAILOVER_STATE_NONE則說明執(zhí)行故障切換，傳入server.runid
(master->failover_state > SENTINEL_FAILOVER_STATE_NONE) ?
server.runid : "*");
if (retval == REDIS_OK) ri->pending_commands++;
}
dictReleaseIterator(di);
}其他哨兵收到sentinel指令后就會調(diào)用sentinelCommand處理這條指令，其內(nèi)部會判斷自己所維護的master的flags二進(jìn)制位是否包含SRI_S_DOWN，如果是則說明被請求的哨兵節(jié)點同樣認(rèn)為master已下線，則直接回復(fù)master的leaderid以及shared.cone即1(代表確認(rèn)當(dāng)前master確實下線)：
void sentinelCommand(redisClient *c) {
//......
else if (!strcasecmp(c->argv[1]->ptr,"is-master-down-by-addr")) {//處理客觀下線請求
//......
//如果master主觀判定下線即flags包含SRI_S_DOWN這個主觀下線標(biāo)識，則isdown設(shè)置為1
if (!sentinel.tilt && ri && (ri->flags & SRI_S_DOWN) &&
(ri->flags & SRI_MASTER))
isdown = 1;
//上文isdown 設(shè)置為1，返回 shared.cone告知對應(yīng)leaderid的master被我方認(rèn)定為下線
//響應(yīng)3部分內(nèi)容，下線狀態(tài)、leader id以及當(dāng)前l(fā)eader的紀(jì)元
addReplyMultiBulkLen(c,3);
addReply(c, isdown ? shared.cone : shared.czero);
addReplyBulkCString(c, leader ? leader : "*");
addReplyLongLong(c, (long long)leader_epoch);
if (leader) sdsfree(leader);
} //......
return;
//......
}最終我們的sentinel的回調(diào)函數(shù)sentinelReceiveIsMasterDownReply處理對端的結(jié)果，發(fā)現(xiàn)返回值為1，說明該節(jié)點對于我們的來說客觀認(rèn)為master下線了。
所以我們的哨兵就需要記錄這個消息，因為我們維護master->sentinels的字典記錄其他哨兵信息，所以定位到其他哨兵客觀下線的回復(fù)后，我們就會從這個字典中找到這個哨兵的結(jié)構(gòu)體將其flags累加一個SRI_MASTER_DOWN的常數(shù)值64，意味這個哨兵客觀認(rèn)定這個master下線了：
void sentinelReceiveIsMasterDownReply(redisAsyncContext *c, void *reply, void *privdata) {
//......
if ( //......)
{
//更新上次響應(yīng)時間
ri->last_master_down_reply_time = mstime();
if (r->element[0]->integer == 1) {//如果返回（cone默認(rèn)設(shè)置為1）1則說明其他哨兵認(rèn)為master下線，累加將當(dāng)前維護的哨兵字段的flags累加SRI_MASTER_DOWN
ri->flags |= SRI_MASTER_DOWN;
} else {
//......
}
//......
}
}2. 啟動故障轉(zhuǎn)移
上一步收集其他哨兵的判斷并更新到各自的flags位后，當(dāng)前哨兵的定時任務(wù)再次遍歷master調(diào)用sentinelHandleRedisInstance處理當(dāng)前master，其內(nèi)部會遍歷當(dāng)前哨兵維護的哨兵數(shù)組獲取這些哨兵對于master下線的看法，如果累加到的哨兵對于下線的看法大于或者等于我們配置quorum之后，則會判定會客觀下線：
我們還是從sentinelHandleRedisInstance方法查看方法入口，可以看到哨兵定時執(zhí)行該方法時會調(diào)用sentinelCheckObjectivelyDown檢查客觀下線狀態(tài)：
void sentinelHandleRedisInstance(sentinelRedisInstance *ri) {
//......
if (ri->flags & SRI_MASTER) {
//......
//檢查其當(dāng)前是否客觀下線
sentinelCheckObjectivelyDown(ri);
//......
}
}步入其內(nèi)部即可看到筆者所說的，遍歷哨兵查看下線結(jié)果并更新master下線狀態(tài)的邏輯：
void sentinelCheckObjectivelyDown(sentinelRedisInstance *master) {
//......
//如果是主觀下線，步入該邏輯
if (master->flags & SRI_S_DOWN) {
//自己的票數(shù)設(shè)置進(jìn)去，quorum為1
quorum = 1; /* the current sentinel. */
//遍歷其他哨兵，如果為客觀下線則累加quorum
di = dictGetIterator(master->sentinels);
while((de = dictNext(di)) != NULL) {
sentinelRedisInstance *ri = dictGetVal(de);
if (ri->flags & SRI_MASTER_DOWN) quorum++;
}
//如果投票數(shù)大于配置的quorum，則odown 為1，即說明客觀認(rèn)定下線了
if (quorum >= master->quorum) odown = 1;
}
//如果明確客觀下線，則廣播+odown事件
if (odown) {
if ((master->flags & SRI_O_DOWN) == 0) {
sentinelEvent(REDIS_WARNING,"+odown",master,"%@ #quorum %d/%d",
quorum, master->quorum);
//累加標(biāo)識，并更新master下線時間
master->flags |= SRI_O_DOWN;
master->o_down_since_time = mstime();
}
} else {
//......
}
}3. 發(fā)起新紀(jì)元leader選舉
基于上述結(jié)果redis會判斷是否發(fā)起故障轉(zhuǎn)移，若需要則通知其他哨兵進(jìn)行l(wèi)eader選舉，收到通知的哨兵會檢查當(dāng)前紀(jì)元是否小于發(fā)起選舉的哨兵紀(jì)元，若符合要求且在此期間沒有別的哨兵發(fā)起選舉，則向其投票。
后續(xù)我們的哨兵收到并收集這些響應(yīng)之后，更新自己所維護的哨兵數(shù)組中的leader_epoch，通過遍歷這個哨兵數(shù)組中的leader_epoch是否和自己所生成的leader_epoch一致，如果統(tǒng)計結(jié)果超過半數(shù)，則說明自己當(dāng)選leader，由此開始進(jìn)行故障轉(zhuǎn)移：
(1) 選舉源碼入口
我們還是以sentinelHandleRedisInstance作為程序入口，可以看到其內(nèi)部調(diào)用sentinelStartFailoverIfNeeded判斷是否需要進(jìn)行故障轉(zhuǎn)移，然后調(diào)用sentinelAskMasterStateToOtherSentinels并傳入SENTINEL_ASK_FORCED發(fā)起leader選舉請求：
//這個入?yún)∩诒鴮嵗彤?dāng)前主節(jié)點的從節(jié)點信息
void sentinelHandleRedisInstance(sentinelRedisInstance *ri) {
//......
if (ri->flags & SRI_MASTER) {
//......
// 判斷是否要進(jìn)行故障切換，若需要則調(diào)用sentinelAskMasterStateToOtherSentinels傳入SENTINEL_ASK_FORCED進(jìn)行l(wèi)eader選舉
if (sentinelStartFailoverIfNeeded(ri))
sentinelAskMasterStateToOtherSentinels(ri,SENTINEL_ASK_FORCED);
// 執(zhí)行故障切換
sentinelFailoverStateMachine(ri);
//......
}
}(2) 確認(rèn)故障轉(zhuǎn)移
我們步入sentinelStartFailoverIfNeeded即可看到其對于是否進(jìn)行故障轉(zhuǎn)移的判斷，邏輯比較簡單：
- 明確是否客觀認(rèn)定下線。
- 明確是否處于故障轉(zhuǎn)移。
- 近期是否有進(jìn)行故障轉(zhuǎn)移。
如果傷處條件都排除則：
- failover_state 即故障轉(zhuǎn)移狀態(tài)設(shè)置為等待故障轉(zhuǎn)移，后續(xù)的函數(shù)狀態(tài)機會根據(jù)這個標(biāo)識進(jìn)行故障轉(zhuǎn)移處理。
- flags標(biāo)識累加處于故障轉(zhuǎn)移中。
- 更新master紀(jì)元為哨兵紀(jì)元+1，用于后續(xù)哨兵leader選舉后更新紀(jì)元使用。
對此我們給出sentinelStartFailoverIfNeeded的判斷，可以看到它會按照上文所說的流程進(jìn)行判斷，明確排除三種情況后調(diào)用sentinelStartFailover設(shè)置故障轉(zhuǎn)移狀態(tài)：
int sentinelStartFailoverIfNeeded(sentinelRedisInstance *master) {
//是否客觀下線，若不是則返回0
if (!(master->flags & SRI_O_DOWN)) return 0;
//是否處于故障轉(zhuǎn)移中，如果是則直接返回0
if (master->flags & SRI_FAILOVER_IN_PROGRESS) return 0;
//距離上次故障轉(zhuǎn)移時間是否小于2倍的超時時間，如果是則返回0
if (mstime() - master->failover_start_time <
master->failover_timeout*2)
{
if (master->failover_delay_logged != master->failover_start_time) {
time_t clock = (master->failover_start_time +
master->failover_timeout*2) / 1000;
char ctimebuf[26];
ctime_r(&clock,ctimebuf);
ctimebuf[24] = '\0'; /* Remove newline. */
master->failover_delay_logged = master->failover_start_time;
redisLog(REDIS_WARNING,
"Next failover delay: I will not start a failover before %s",
ctimebuf);
}
return 0;
}
//啟動故障轉(zhuǎn)移 并返回1
sentinelStartFailover(master);
return 1;
}步入sentinelStartFailover即可看到我們上文所說故障轉(zhuǎn)移狀態(tài)更新：
void sentinelStartFailover(sentinelRedisInstance *master) {
redisAssert(master->flags & SRI_MASTER);
//故障轉(zhuǎn)移等待啟動
master->failover_state = SENTINEL_FAILOVER_STATE_WAIT_START;
//設(shè)置為正在故障轉(zhuǎn)移
master->flags |= SRI_FAILOVER_IN_PROGRESS;
//更新紀(jì)元
master->failover_epoch = ++sentinel.current_epoch;
//......
}結(jié)果上述步驟明確知曉redis需要進(jìn)行故障轉(zhuǎn)移之后，哨兵會再次調(diào)用sentinelAskMasterStateToOtherSentinels方法傳入當(dāng)前哨兵的server.runid向其他哨兵發(fā)起投票請求，并通過sentinelReceiveIsMasterDownReply處理響應(yīng)結(jié)果：
void sentinelAskMasterStateToOtherSentinels(sentinelRedisInstance *master, int flags) {
//遍歷其他哨兵
di = dictGetIterator(master->sentinels);
while((de = dictNext(di)) != NULL) {
//......
//因為failover_state 在上一步已經(jīng)改為傳入自己的SENTINEL_FAILOVER_STATE_WAIT_START即等待故障轉(zhuǎn)移，故大于SENTINEL_FAILOVER_STATE_NONE，于是傳入哨兵的server.runid發(fā)起投票選舉
retval = redisAsyncCommand(ri->cc,
sentinelReceiveIsMasterDownReply, NULL,
"SENTINEL is-master-down-by-addr %s %s %llu %s",
master->addr->ip, port,
sentinel.current_epoch,
//若大于SENTINEL_FAILOVER_STATE_NONE則說明執(zhí)行故障切換，傳入server.runid
(master->failover_state > SENTINEL_FAILOVER_STATE_NONE) ?
server.runid : "*");
if (retval == REDIS_OK) ri->pending_commands++;
}
dictReleaseIterator(di);
}(4) 對端哨兵處理發(fā)起選舉的投票結(jié)果
上述步驟發(fā)起投票的哨兵節(jié)點發(fā)起投票后，收到投票請求的哨兵實例就會進(jìn)行如下檢查：
- master紀(jì)元小于發(fā)起投票請求的哨兵紀(jì)元req_epoch。
- 當(dāng)前哨兵紀(jì)元小于req_epoch。
如果符合要求則說明發(fā)起投票請求的哨兵可以作為leader，當(dāng)前實例將leader 設(shè)置為該節(jié)點，然后回復(fù)結(jié)果給發(fā)送結(jié)果的實例：
char *sentinelVoteLeader(sentinelRedisInstance *master, uint64_t req_epoch, char *req_runid, uint64_t *leader_epoch) {
//發(fā)起選舉的哨兵紀(jì)元大于當(dāng)前紀(jì)元，則修改當(dāng)前紀(jì)元
if (req_epoch > sentinel.current_epoch) {
sentinel.current_epoch = req_epoch;
sentinelFlushConfig();
sentinelEvent(REDIS_WARNING,"+new-epoch",master,"%llu",
(unsigned long long) sentinel.current_epoch);
}
//如果master紀(jì)元小于發(fā)起選舉的紀(jì)元且當(dāng)前哨兵紀(jì)元小于等于發(fā)起選舉的紀(jì)元
if (master->leader_epoch < req_epoch && sentinel.current_epoch <= req_epoch)
{
sdsfree(master->leader);
//設(shè)置當(dāng)前的master為candidate的runid
master->leader = sdsnew(req_runid);
//更新紀(jì)元
master->leader_epoch = sentinel.current_epoch;
sentinelFlushConfig();
//投票給發(fā)起選舉的人
sentinelEvent(REDIS_WARNING,"+vote-for-leader",master,"%s %llu",
master->leader, (unsigned long long) master->leader_epoch);
/* If we did not voted for ourselves, set the master failover start
* time to now, in order to force a delay before we can start a
* failover for the same master. */
if (strcasecmp(master->leader,server.runid))
master->failover_start_time = mstime()+rand()%SENTINEL_MAX_DESYNC;
}
*leader_epoch = master->leader_epoch;
return master->leader ? sdsnew(master->leader) : NULL;
}(5) 處理投票結(jié)果
收到響應(yīng)后sentinelReceiveIsMasterDownReply回調(diào)函數(shù)就會解析出其他哨兵的leader_epoch 信息，作為后續(xù)選舉leader的依據(jù)，如果半數(shù)以上的leader_epoch 為當(dāng)前哨兵所設(shè)置的run_id，則說明當(dāng)前哨兵作為leader進(jìn)行故障轉(zhuǎn)移：
void sentinelReceiveIsMasterDownReply(redisAsyncContext *c, void *reply, void *privdata) {
sentinelRedisInstance *ri = c->data;
redisReply *r;
REDIS_NOTUSED(privdata);
//......
if (strcmp(r->element[1]->str,"*")) {//不為*則采集投票結(jié)果
//......
//基于返回結(jié)果更新當(dāng)前哨兵維護的哨兵數(shù)組中l(wèi)eader的leader_epoch 信息(記錄的是作為leader的哨兵的run_id)，作為后續(xù)選舉leader使用
ri->leader = sdsnew(r->element[1]->str);
ri->leader_epoch = r->element[2]->integer;
}
}
}最后基于狀態(tài)機模式，根據(jù)當(dāng)前master狀態(tài)為SENTINEL_FAILOVER_STATE_WAIT_START于是調(diào)用sentinelFailoverWaitStart選舉leader
void sentinelFailoverStateMachine(sentinelRedisInstance *ri) {
redisAssert(ri->flags & SRI_MASTER);
if (!(ri->flags & SRI_FAILOVER_IN_PROGRESS)) return;
switch(ri->failover_state) {
//如果狀態(tài)為SENTINEL_FAILOVER_STATE_WAIT_START，則調(diào)用sentinelFailoverWaitStart選舉出leader
case SENTINEL_FAILOVER_STATE_WAIT_START:
sentinelFailoverWaitStart(ri);
break;
//......
}
}步入sentinelFailoverWaitStart即可看到該方法調(diào)用sentinelGetLeader，如果發(fā)現(xiàn)是自己則發(fā)送廣播告知自己為leader進(jìn)行故障轉(zhuǎn)移：
void sentinelFailoverWaitStart(sentinelRedisInstance *ri) {
//......
//獲取leader
leader = sentinelGetLeader(ri, ri->failover_epoch);
isleader = leader && strcasecmp(leader,server.runid) == 0;
sdsfree(leader);
//......
//告知當(dāng)選的leader是自己
sentinelEvent(REDIS_WARNING,"+elected-leader",ri,"%@");
ri->failover_state = SENTINEL_FAILOVER_STATE_SELECT_SLAVE;
ri->failover_state_change_time = mstime();
sentinelEvent(REDIS_WARNING,"+failover-state-select-slave",ri,"%@");
}對此我們也給出選舉哨兵leader的核心方法sentinelGetLeader，核心步驟為：
- 如果投票結(jié)果給出的leader值不為空(這個leader記錄的是其他哨兵投票的實例的run_id)且紀(jì)元和當(dāng)前選舉紀(jì)元一致，則給對應(yīng)的leader票數(shù)+1。
- 將這個投票結(jié)果存入counter這個字典中。
- 遍歷counter如果這個值大于配置的quorum或哨兵的半數(shù)以上，則將其設(shè)置為winner，即最后的leader，由此讓這個leader哨兵進(jìn)行故障轉(zhuǎn)移：
對應(yīng)的我們也給出這段代碼的實現(xiàn)：
char *sentinelGetLeader(sentinelRedisInstance *master, uint64_t epoch) {
//......
//設(shè)置voters 為哨兵數(shù)+1
voters = dictSize(master->sentinels)+1; /* All the other sentinels and me. */
/* Count other sentinels votes */
//根據(jù)紀(jì)元遍歷其他哨兵的選票結(jié)果
di = dictGetIterator(master->sentinels);
while((de = dictNext(di)) != NULL) {
sentinelRedisInstance *ri = dictGetVal(de);
//如果其他哨兵投票的leader值不為空，且紀(jì)元和當(dāng)前投票紀(jì)元一致，則給這個leader的對應(yīng)的run_id對應(yīng)的投票數(shù)做個自增
if (ri->leader != NULL && ri->leader_epoch == sentinel.current_epoch)
sentinelLeaderIncr(counters,ri->leader);
}
//......
//找到得票最多的
di = dictGetIterator(counters);
while((de = dictNext(di)) != NULL) {
uint64_t votes = dictGetUnsignedIntegerVal(de);
if (votes > max_votes) {
max_votes = votes;
winner = dictGetKey(de);
}
}
dictReleaseIterator(di);
//......
//如果票數(shù)大于一半+1或大于配置的quorum則設(shè)置為leader
voters_quorum = voters/2+1;
if (winner && (max_votes < voters_quorum || max_votes < master->quorum))
winner = NULL;
winner = winner ? sdsnew(winner) : NULL;
//......
return winner;
}小結(jié)
自此我們來小結(jié)一下哨兵選舉與故障轉(zhuǎn)移的大體過程：
- 當(dāng)前哨兵主觀認(rèn)定下線之后，通過異步連接詢問其它哨兵是否客觀認(rèn)定master下線。
- 超過半數(shù)的哨兵認(rèn)為下線則當(dāng)前哨兵就認(rèn)為master下線于是開啟發(fā)起投票選舉。
- 更新自己的紀(jì)元并攜帶runid到其它哨兵節(jié)點上拉票。
- 基于回調(diào)函數(shù)獲取其它哨兵選票結(jié)果進(jìn)行遍歷匯總，用以一個字典以哨兵runid為key，投票值為value進(jìn)行維護。
- 匯總后通知全局哨兵leader。
- leader進(jìn)行故障轉(zhuǎn)移。
