Redis源码解析:集群手动故障转移、从节点迁移详解

else if (!strcasecmp(c->argv[1]->ptr,"failover") &&       (c->argc == 2 || c->argc == 3)) {   /* CLUSTER FAILOVER [FORCE|TAKEOVER] */   int force = 0, takeover = 0;    if (c->argc == 3) {     if (!strcasecmp(c->argv[2]->ptr,"force")) {       force = 1;     } else if (!strcasecmp(c->argv[2]->ptr,"takeover")) {       takeover = 1;       force = 1; /* Takeover also implies force. */     } else {       addReply(c,shared.syntaxerr);       return;     }   }    /* Check preconditions. */   if (nodeIsMaster(myself)) {     addReplyError(c,"You should send CLUSTER FAILOVER to a slave");     return;   } else if (myself->slaveof == NULL) {     addReplyError(c,"I'm a slave but my master is unknown to me");     return;   } else if (!force &&         (nodeFailed(myself->slaveof) ||         myself->slaveof->link == NULL))   {     addReplyError(c,"Master is down or failed, "             "please use CLUSTER FAILOVER FORCE");     return;   }   resetManualFailover();   server.cluster->mf_end = mstime() + REDIS_CLUSTER_MF_TIMEOUT;    if (takeover) {     /* A takeover does not perform any initial check. It just      * generates a new configuration epoch for this node without      * consensus, claims the master's slots, and broadcast the new      * configuration. */     redisLog(REDIS_WARNING,"Taking over the master (user request).");     clusterBumpConfigEpochWithoutConsensus();     clusterFailoverReplaceYourMaster();   } else if (force) {     /* If this is a forced failover, we don't need to talk with our      * master to agree about the offset. We just failover taking over      * it without coordination. */     redisLog(REDIS_WARNING,"Forced failover user request accepted.");     server.cluster->mf_can_start = 1;   } else {     redisLog(REDIS_WARNING,"Manual failover user request accepted.");     clusterSendMFStart(myself->slaveof);   }   addReply(c,shared.ok); } 

else if (type == CLUSTERMSG_TYPE_MFSTART) {   /* This message is acceptable only if I'm a master and the sender    * is one of my slaves. */   if (!sender || sender->slaveof != myself) return 1;   /* Manual failover requested from slaves. Initialize the state    * accordingly. */   resetManualFailover();   server.cluster->mf_end = mstime() + REDIS_CLUSTER_MF_TIMEOUT;   server.cluster->mf_slave = sender;   pauseClients(mstime()+(REDIS_CLUSTER_MF_TIMEOUT*2));   redisLog(REDIS_WARNING,"Manual failover requested by slave %.40s.",     sender->name); } 

void clusterBuildMessageHdr(clusterMsg *hdr, int type) {   ...   /* Set the message flags. */   if (nodeIsMaster(myself) && server.cluster->mf_end)     hdr->mflags[0] |= CLUSTERMSG_FLAG0_PAUSED;   ... }   

int clusterProcessPacket(clusterLink *link) {   ...   /* Check if the sender is a known node. */   sender = clusterLookupNode(hdr->sender);   if (sender && !nodeInHandshake(sender)) {     ...     /* Update the replication offset info for this node. */     sender->repl_offset = ntohu64(hdr->offset);     sender->repl_offset_time = mstime();     /* If we are a slave performing a manual failover and our master      * sent its offset while already paused, populate the MF state. */     if (server.cluster->mf_end &&       nodeIsSlave(myself) &&       myself->slaveof == sender &&       hdr->mflags[0] & CLUSTERMSG_FLAG0_PAUSED &&       server.cluster->mf_master_offset == 0)     {       server.cluster->mf_master_offset = sender->repl_offset;       redisLog(REDIS_WARNING,         "Received replication offset for paused "         "master manual failover: %lld",         server.cluster->mf_master_offset);     }   } }   

void clusterHandleManualFailover(void) {   /* Return ASAP if no manual failover is in progress. */   if (server.cluster->mf_end == 0) return;   /* If mf_can_start is non-zero, the failover was already triggered so the    * next steps are performed by clusterHandleSlaveFailover(). */   if (server.cluster->mf_can_start) return;   if (server.cluster->mf_master_offset == 0) return; /* Wait for offset... */   if (server.cluster->mf_master_offset == replicationGetSlaveOffset()) {     /* Our replication offset matches the master replication offset      * announced after clients were paused. We can start the failover. */     server.cluster->mf_can_start = 1;     redisLog(REDIS_WARNING,       "All master replication stream processed, "       "manual failover can start.");   } } 

/* If a manual failover timed out, abort it. */ void manualFailoverCheckTimeout(void) {   if (server.cluster->mf_end && server.cluster->mf_end < mstime()) {     redisLog(REDIS_WARNING,"Manual failover timed out.");     resetManualFailover();   } } 

void clusterCron(void) {   ...   orphaned_masters = 0;   max_slaves = 0;   this_slaves = 0;   di = dictGetSafeIterator(server.cluster->nodes);   while((de = dictNext(di)) != NULL) {     clusterNode *node = dictGetVal(de);     now = mstime(); /* Use an updated time at every iteration. */     mstime_t delay;      if (node->flags &       (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE))         continue;      /* Orphaned master check, useful only if the current instance      * is a slave that may migrate to another master. */     if (nodeIsSlave(myself) && nodeIsMaster(node) && !nodeFailed(node)) {       int okslaves = clusterCountNonFailingSlaves(node);        /* A master is orphaned if it is serving a non-zero number of        * slots, have no working slaves, but used to have at least one        * slave. */       if (okslaves == 0 && node->numslots > 0 && node->numslaves)         orphaned_masters++;       if (okslaves > max_slaves) max_slaves = okslaves;       if (nodeIsSlave(myself) && myself->slaveof == node)         this_slaves = okslaves;     }     ...   }   ...   if (nodeIsSlave(myself)) {     ...     /* If there are orphaned slaves, and we are a slave among the masters      * with the max number of non-failing slaves, consider migrating to      * the orphaned masters. Note that it does not make sense to try      * a migration if there is no master with at least *two* working      * slaves. */     if (orphaned_masters && max_slaves >= 2 && this_slaves == max_slaves)       clusterHandleSlaveMigration(max_slaves);   }   ... }  

void clusterHandleSlaveMigration(int max_slaves) {   int j, okslaves = 0;   clusterNode *mymaster = myself->slaveof, *target = NULL, *candidate = NULL;   dictIterator *di;   dictEntry *de;   /* Step 1: Don't migrate if the cluster state is not ok. */   if (server.cluster->state != REDIS_CLUSTER_OK) return;   /* Step 2: Don't migrate if my master will not be left with at least    *     'migration-barrier' slaves after my migration. */   if (mymaster == NULL) return;   for (j = 0; j < mymaster->numslaves; j++)     if (!nodeFailed(mymaster->slaves[j]) &&       !nodeTimedOut(mymaster->slaves[j])) okslaves++;   if (okslaves <= server.cluster_migration_barrier) return;   /* Step 3: Idenitfy a candidate for migration, and check if among the    * masters with the greatest number of ok slaves, I'm the one with the    * smaller node ID.    *    * Note that this means that eventually a replica migration will occurr    * since slaves that are reachable again always have their FAIL flag    * cleared. At the same time this does not mean that there are no    * race conditions possible (two slaves migrating at the same time), but    * this is extremely unlikely to happen, and harmless. */   candidate = myself;   di = dictGetSafeIterator(server.cluster->nodes);   while((de = dictNext(di)) != NULL) {     clusterNode *node = dictGetVal(de);     int okslaves;     /* Only iterate over working masters. */     if (nodeIsSlave(node) || nodeFailed(node)) continue;     /* If this master never had slaves so far, don't migrate. We want      * to migrate to a master that remained orphaned, not masters that      * were never configured to have slaves. */     if (node->numslaves == 0) continue;     okslaves = clusterCountNonFailingSlaves(node);     if (okslaves == 0 && target == NULL && node->numslots > 0)       target = node;     if (okslaves == max_slaves) {       for (j = 0; j < node->numslaves; j++) {         if (memcmp(node->slaves[j]->name,               candidate->name,               REDIS_CLUSTER_NAMELEN) < 0)         {           candidate = node->slaves[j];         }       }     }   }   dictReleaseIterator(di);   /* Step 4: perform the migration if there is a target, and if I'm the    * candidate. */   if (target && candidate == myself) {     redisLog(REDIS_WARNING,"Migrating to orphaned master %.40s",       target->name);     clusterSetMaster(target);   } } 

void clusterHandleConfigEpochCollision(clusterNode *sender) {   /* Prerequisites: nodes have the same configEpoch and are both masters. */   if (sender->configEpoch != myself->configEpoch ||     !nodeIsMaster(sender) || !nodeIsMaster(myself)) return;   /* Don't act if the colliding node has a smaller Node ID. */   if (memcmp(sender->name,myself->name,REDIS_CLUSTER_NAMELEN) <= 0) return;   /* Get the next ID available at the best of this node knowledge. */   server.cluster->currentEpoch++;   myself->configEpoch = server.cluster->currentEpoch;   clusterSaveConfigOrDie(1);   redisLog(REDIS_VERBOSE,     "WARNING: configEpoch collision with node %.40s."     " configEpoch set to %llu",     sender->name,     (unsigned long long) myself->configEpoch); }