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); 
}