利用QJM实现HDFS自动主从切换(HA Automatic Failover)源码详析

最近研究了下NameNode HA Automatic Failover方面的东西，当Active NN因为异常或其他原因不能正常提供服务时，处于Standby状态的NN就可以自动切换为Active状态，从而到达真正的高可用

NN HA Automatic Failover架构图

为了实现自动切换，需要依赖ZooKeeper和ZKFC组件，ZooKeeper主要用来记录NN的相关状态信息，zkfc组件以单独的JVM进程的形式运行在NN所在的节点上。下面首先分析下NN的启动流程，NN对象在实例化过程中，如果在hdfs-site.xml中配置的dfs.ha.namenodes.${dfs.nameservices}个数多于1个，则属性haEnabled为true,state的初始状态即为STANDBY

HAUtil.isHAEnabled(Configuration conf, String nsId) {
    //根据${dfs.nameservices}获取dfs.ha.namenodes.${dfs.nameservices}
    Map<String, Map<String, InetSocketAddress>> addresses =
      DFSUtil.getHaNnRpcAddresses(conf);
    if (addresses == null) return false;
    Map<String, InetSocketAddress> nnMap = addresses.get(nsId);
    //如果ndId对应的ha.namenodes个数大于1并且nnMap不为空,返回true
    return nnMap != null && nnMap.size() > 1;
  }
  
  //NN启动过程中首先判断haEnabled属性值，为true则初始状态为STANDBY
  NameNode.createHAState() {
    return !haEnabled ? ACTIVE_STATE : STANDBY_STATE;
  }

假设集群中有两个NN,这两个NN在初始启动时，都为STANDBY状态，如果不配置Automatic Failover，则需要手动将其中的一个NN切换为ACTIVE模式，命令如下

hdfs haadmin-transitionToActive nn1

如果配置并启动zkfc组件，则会自动将本节点的一个NameNode切换为ACTIVE状态，这个取决于先在哪台NN上启动ZKFC,ZKFC和NN的启动顺序并没有强制的要求。下面来主要分析下，当配置的两个NN节点都启动之后，ZKFC组件的启动主要做了哪些事。ZKFC的启动类是DFSZKFailoverController，继承自ZKFailoverController，首先从main方法入手

DFSZKFailoverController.main(String args[]){
    if (DFSUtil.parseHelpArgument(args, 
        ZKFailoverController.USAGE, System.out, true)) {
      System.exit(0);
    }
    //加载hdfs的配置信息
    GenericOptionsParser parser = new GenericOptionsParser(
        new HdfsConfiguration(), args);
    //根据配置信息构造DFSZKFailoverController
    //根据nsId,nnId实例化NNHAServiceTarget对象，并设置到zkfc对象中
    //HAServiceTarget对象用来建立到指定NN的各种网络参数
    DFSZKFailoverController zkfc = DFSZKFailoverController.create(
        parser.getConfiguration());
    //建立与zk的连接并格式化zk的目录
    //实例化ActiveStandbyElector和HealthMonitor
    //启动RPCServer
    System.exit(zkfc.run(parser.getRemainingArgs()));
  }

在ZKFC启动的过程中，启动了两个非常重要的进程内组件:HealthMonitor和ActiveStandbyElector。ZKFC主要从HealthMonitor和ActiveStandbyElector中订阅事件并管理NN的状态并负责fencing。HealthMonitor定期检查NN的健康状况，如果出现问题，以捕获异常的方式通过回调方法将变化通知给ZKFailoverController。ActiveStandbyElector主要用于管理NN在zk上的状态，包括创建节点，节点监控等。HealthMonitor在初始启动时，如果本地节点处于健康状态，则会触发一系列的事件使当前NN节点参加选举并切换为ACTIVE状态，具体代码如下

 HealthMonitor.doHealthChecks() {
    while (shouldRun) {
      HAServiceStatus status = null;
      boolean healthy = false;
      try {
    	//通过proxy获取NN的状态
        status = proxy.getServiceStatus();
        proxy.monitorHealth();
        healthy = true;
      } catch (HealthCheckFailedException e) {
        LOG.warn("Service health check failed for " + targetToMonitor
            + ": " + e.getMessage());
        //调用callbacks集合中的回调方法，以事件的形式通知ZKFC
        enterState(State.SERVICE_UNHEALTHY);
      } catch (Throwable t) {
        RPC.stopProxy(proxy);
        proxy = null;
        enterState(State.SERVICE_NOT_RESPONDING);
        Thread.sleep(sleepAfterDisconnectMillis);
        return;
      }
      
      if (status != null) {
        setLastServiceStatus(status);
      }
      if (healthy) {
	//初始启动时State=INITIALIZING，当前状态与初始状态不一致时，触发状态转移
        enterState(State.SERVICE_HEALTHY);
      }

      Thread.sleep(checkIntervalMillis);
    }
  }
  
  
  HealthMonitor.enterState(State newState) {
    //本次状态和上一次状态不一致的时候，触发回调
    if (newState != state) {
      LOG.info("Entering state " + newState);
      state = newState;
      synchronized (callbacks) {
        for (Callback cb : callbacks) {
          cb.enteredState(newState);
        }
      }
    }
  }
  
  
  HealthCallbacks.enteredState(HealthMonitor.State newState) {
      setLastHealthState(newState);
      //根据当前状态判断是否可以参加选举
      recheckElectability();
   }
   
  HealthMonitor.recheckElectability() {
    synchronized (elector) {
      synchronized (this) {
        boolean healthy = lastHealthState == State.SERVICE_HEALTHY;
    
        long remainingDelay = delayJoiningUntilNanotime - System.nanoTime(); 
        if (remainingDelay > 0) {
          if (healthy) {
            LOG.info("Would have joined master election, but this node is " +
                "prohibited from doing so for " +
                TimeUnit.NANOSECONDS.toMillis(remainingDelay) + " more ms");
          }
          scheduleRecheck(remainingDelay);
          return;
        }
    
        switch (lastHealthState) {
	//状态为SERVICE_HEALTHY,可参加选举
        case SERVICE_HEALTHY:
          elector.joinElection(targetToData(localTarget));
          break;
        //状态为INITIALIZING，失去选举资格
        case INITIALIZING:
          elector.quitElection(false);
          break;
    
	//状态为SERVICE_UNHEALTHY或者SERVICE_NOT_RESPONDING，失去选举资格
        case SERVICE_UNHEALTHY:
        case SERVICE_NOT_RESPONDING:
          elector.quitElection(true);
          break;
          
        case HEALTH_MONITOR_FAILED:
          fatalError("Health monitor failed!");
          break;
          
        default:
          throw new IllegalArgumentException("Unhandled state:" + lastHealthState);
        }
      }
    }
  }
  
  
  ActiveStandbyElector.joinElectionInternal() {
    Preconditions.checkState(appData != null,
        "trying to join election without any app data");
    if (zkClient == null) {
      if (!reEstablishSession()) {
        fatalError("Failed to reEstablish connection with ZooKeeper");
        return;
      }
    }

    createRetryCount = 0;
    wantToBeInElection = true;
    //向zk写入ephemeral类型的znode，当NN挂掉后，会被自动删除
    createLockNodeAsync();
  }
  
  
  ActiveStandbyElector.createLockNodeAsync() {
    //异步调用，当方法返回时，触发回调方法
    //processResult(int rc, String path, Object ctx,String name)
    zkClient.create(zkLockFilePath, appData, zkAcl, CreateMode.EPHEMERAL,
        this, zkClient);
  }
  
  
  ActiveStandbyElector.processResult(int rc, String path, Object ctx,
      String name) {
    Code code = Code.get(rc);
    if (isSuccess(code)) {
      //创建成功,试图使节点变为ACTIVE状态
      //becomeActive()通过与本地NN进行RPC通信，将NN的state设置为ACTIVE
      //从而使得当前节点的NN更新为主节点
      if (becomeActive()) {
	//监控节点
        monitorActiveStatus();
      } else {
        reJoinElectionAfterFailureToBecomeActive();
      }
      return;
    }
    //节点已经存在,说明已经有NN成为了主节点，此时本节点只能作为热备节点存在
    //故状态设置为STANDBY
    if (isNodeExists(code)) {
      if (createRetryCount == 0) {
        becomeStandby();
      }
      //监控ACTIVE NN写入的znode，当节点状态改变时，通过watch机制触发回调事件
      monitorActiveStatus();
      return;
    }
    
    
    ActiveStandbyElector.becomeActive() {
      if (state == State.ACTIVE) {
        // already active
        return true;
      }
      try {
    	//获取上一个ACTIVE NN的面包屑节点数据,并对上一个ACTIVE NN执行fence操作
        Stat oldBreadcrumbStat = fenceOldActive();
        //更新面包屑节点的数据
        writeBreadCrumbNode(oldBreadcrumbStat);
      
        //rpc方式与当前NN交互，使得当前节点的NN变为ACTIVE状态
        appClient.becomeActive();
        state = State.ACTIVE;
        return true;
      } catch (Exception e) {
        return false;
      }
    }

处于STANDBY状态的NN会监控ACTIVE NN写入zk的znode节点，当节点状态改变时，触发zk的watch回调，使得STANDBY NN重新参与到选举中，从而完成状态的自动切换，代码如下

 ActiveStandbyElector.processWatchEvent(ZooKeeper zk, WatchedEvent event) {
   String path = event.getPath();
   if (path != null) {
     switch (eventType) {
     case NodeDeleted:
       if (state == State.ACTIVE) {
         enterNeutralMode();
       }
       //重新参加选举
       joinElectionInternal();
       break;
     case NodeDataChanged:
       monitorActiveStatus();
       break;
     default:
       monitorActiveStatus();
     }
 }

上述主要是从代码的角度去理解和分析了NN自动切换的大致流程，相比手动切换的方式，可用性大大提升，同时减轻了运维的负担。

利用QJM实现HDFS自动主从切换(HA Automatic Failover)源码详析