DataNode生命线消息

Posted 2020-08-24 Android路上的人

前言

---

在HDFS中，我们都知道DataNode是通过定期发送心跳信息到NameNode，以此证明自己还“活着”。当然心跳信息发送的另一项作用是发送自身的块报告信息给NameNode，以此保证集群数据的更新。然后NameNode会反馈给各DataNode一个回复命令。从这里看出，心跳在这里的所执行的操作还是比较“重”的。所以这里会引发出一个问题，一方面DataNode需要及时将自身的块信息报告给NameNode，另一方面，DataNode又要等待NameNode的心跳返回命令，换句话说，如果NameNode当前处理忙碌状态，处理心跳的速度异常地缓慢，那么就有可能造成DataNode下次心跳发送的延时，最糟糕的结果就是被认定为了dead node。所以在这里，我们应该要将心跳的“存活”检查功能从当前逻辑中剥离开来，从而达到一个轻量级的DataNode的生命检查。在本文的讲述中，我们将此称之为生命线（Lifeline）。

DataNode生命线介绍

---

DataNode生命线，英文全称为DataNode Lifeline。随着HDFS代码功能的叠加，每次的心跳处理过程变得也越来越“重”。其实心跳过程中的块上报过程和等待回复命令这2个过程是可以分开的。而DataNode Lifeline消息就是基于前者实现的。用下面一句话来概括DataNode Lifeline的概念。

DataNoded Lifeline本质上是一次轻量级的块信息的汇报，它不需要等待NameNode的回复结果，同时它能达到DataNode状态检查的目的。

DataNode生命线的适用场景

---

那么DataNode生命线到底能帮我们解决哪些异常情况下的问题呢？这里举2个例子，这样大家就会明白它的重要性了。

场景一：NameNode持续忙碌状态

---

在原有的逻辑中，DataNode的心跳报告方法是阻塞式的，它必须等待NameNode完毕，并给出回复命令，然后DataNode接着执行这些命令操作。所以这里会有一个严重的问题，当NameNode正在处理一个很大的块报告信息时，为了保持数据的一致性，处理过程是需要加锁的，其它的心跳信息就会被迫处于阻塞状态，等待当前处理过程的结束。这就会造成DataNode下次心跳的延时，如果DataNode超过了心跳检测的最长容忍时间（默认630s），就会被认为是dead node了，最终会导致一次完全没有必要的大量的replication操作。

场景二：DataNode持续忙碌状态

---

不仅仅是NameNode会处于忙碌状态，DataNode自身也会存在忙碌状态。比如说，DataNode用于发送心跳信息的BPServiceActor线程服务突然卡在了其中某一步操作上了，导致其没有及时的执行blockReport操作，时间长了也会被认为是dead node了。

其实从这里我们可以看出，将心跳的存活检查功能和等待命令处理过程放在一起是一件不太靠谱的事情。一个好的办法就是构造一个轻量级的消息发送接口，放在一个不同于BPServiceActor的心跳发送进程内，这样能很好地做到DataNode存活状态的更新了。而DataNode Lifeline就是这么做的。

DataNode生命线设计

---

DataNode生命线消息功能目前暂未发布，实现于社区issue：HDFS-9239（DataNode Lifeline Protocol: an alternative protocol for reporting DataNode liveness）。笔者在阅读完它的设计文档之后，总结了以下几个关键设计点：

• 构造了一个轻量级的块报告信息协议，汇报的块信息的格式与当前心跳形式完全一致，区别在于它不需要带返回结果，而且不需要NameNode进行加锁操作。
• DataNode生命线消息发送需要放在一个独立的线程中，进行定期的执行，避免心跳发送主线程的影响，同时做到一个备用的功能。

主要是以上2点，更多细节，大家可以阅读文章末尾的DataNode Lifeline设计文档链接。

DataNode生命线协议的核心实现

---

下面从源代码的层面来分析DataNode Lifeline功能，希望能帮助大家更好地理解此功能特性。

首先需要在pb中新增接口方法（这里的返回内容为空）：

 // The lifeline protocol does not use a new request message type. Instead, it
 // reuses the existing heartbeat request message.

 // Unlike heartbeats, the response is empty. There is no command dispatch.
 message LifelineResponseProto {
 }

 service DatanodeLifelineProtocolService {
   rpc sendLifeline(hadoop.hdfs.datanode.HeartbeatRequestProto)
       returns(LifelineResponseProto);
 }

定义完接口之后，然后实现对应的server端和client端的pb实现方法。这里就不展开具体介绍了。下面我们重点来看看，DataNode Lifeline生命线消息是如何发送给NameNode的。

Lifeline消息发送线程服务被定义在了类BPServiceActor中，下面是此类的定义：

  private final class LifelineSender implements Runnable, Closeable {
    // NameNode通信地址
    private final InetSocketAddress lifelineNnAddr;
    // Lifeline消息发送线程
    private Thread lifelineThread;
    // Lifeline消息RPC接口调用类
    private DatanodeLifelineProtocolClientSideTranslatorPB lifelineNamenode;

    public LifelineSender(InetSocketAddress lifelineNnAddr) {
      this.lifelineNnAddr = lifelineNnAddr;
    }
    ...

然后我们来看发送Lifeline消息的主逻辑：

    @Override
    public void run() {
      // The lifeline RPC depends on registration with the NameNode, so wait for
      // initial registration to complete.
      ...
      while (shouldRun()) {
        try {
          if (lifelineNamenode == null) {
            lifelineNamenode = dn.connectToLifelineNN(lifelineNnAddr);
          }
          // 如果当前时间在发送Lifeline消息的周期时间内，则发送Lifeline消息
          sendLifelineIfDue();
          Thread.sleep(scheduler.getLifelineWaitTime());
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        } catch (IOException e) {
          LOG.warn("IOException in LifelineSender for " + BPServiceActor.this,
              e);
        }
      }

      LOG.info("LifelineSender for " + BPServiceActor.this + " exiting.");
    }

这里我们进入sendLifelineIfDue方法，

    private void sendLifelineIfDue() throws IOException {
      // 获取当前发送时间
      long startTime = scheduler.monotonicNow();
      // 如果当前发送时间还没到目标下一次发送的时间，则跳过此次发送动作
      if (!scheduler.isLifelineDue(startTime)) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Skipping sending lifeline for " + BPServiceActor.this
              + ", because it is not due.");
        }
        return;
      }
      if (dn.areHeartbeatsDisabledForTests()) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Skipping sending lifeline for " + BPServiceActor.this
              + ", because heartbeats are disabled for tests.");
        }
        return;
      }
      // 否则发送Lifeline消息，即块报告信息
      sendLifeline();
      // 进行Lifeline消息的metric统计
      dn.getMetrics().addLifeline(scheduler.monotonicNow() - startTime);
      // 设置下一次发送Lifeline消息的时间
      scheduler.scheduleNextLifeline(scheduler.monotonicNow());
    }

设置下一次Lifeline消息发送时间方法如下，

    long scheduleNextLifeline(long baseTime) {
      // Numerical overflow is possible here and is okay.
      nextLifelineTime = baseTime + lifelineIntervalMs;
      return nextLifelineTime;
    }

这里Lifeline消息的发送间隔为3倍的心跳发送间隔时间，也就是默认9秒。

看到这里，如果大家比较仔细想的话，会有这么一个特殊的问题：当心跳发送线程被阻塞的时候，Lifeline发送线程确实能够代替进行块报告信息 的发送，但是当2个线程都在正常跑的情况下，那不是造成块报告信息的重复发送了吗？这是很特殊的一点，笔者在之前也没考虑到这点，设计者在这里做了一个巧妙的设计。

在每次心跳成功发送后，也进行下次Lifeline消息发送时间的设置，表明此次周期内不需要进行Lifeline消息的发送了，因为心跳已经将块信息报告给NameNode了。

设置下次心跳发送时间的逻辑中新增了此处理逻辑：

    long scheduleNextHeartbeat() {
      // Numerical overflow is possible here and is okay.
      nextHeartbeatTime = monotonicNow() + heartbeatIntervalMs;
      // 以下次心跳时间为起始时间点，重新设置下次Lifeline时间
      scheduleNextLifeline(nextHeartbeatTime);
      return nextHeartbeatTime;
    }

对于上面的处理过程，也就是说，在心跳发送正常的情况下，Lifeline消息线程其实是不会发送消息出去的。

下面我们继续来看NameNode端的Lifeline消息处理过程，在类FSNamesystem中，因为Lifeline消息只是用来发送块报告信息，是一个轻量级的处理方法，这里并不需要获取锁的操作,代码如下：

  void handleLifeline(DatanodeRegistration nodeReg, StorageReport[] reports,
      long cacheCapacity, long cacheUsed, int xceiverCount, int xmitsInProgress,
      int failedVolumes, VolumeFailureSummary volumeFailureSummary)
      throws IOException {
    int maxTransfer = blockManager.getMaxReplicationStreams() - xmitsInProgress;
    blockManager.getDatanodeManager().handleLifeline(nodeReg, reports,
        getBlockPoolId(), cacheCapacity, cacheUsed, xceiverCount, maxTransfer,
        failedVolumes, volumeFailureSummary);
  }

而正常情况下的心跳处理方法，是需要加锁的，代码如下：

  HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
      StorageReport[] reports, long cacheCapacity, long cacheUsed,
      int xceiverCount, int xmitsInProgress, int failedVolumes,
      VolumeFailureSummary volumeFailureSummary,
      boolean requestFullBlockReportLease) throws IOException {
    // 需要进行获取锁操作
    readLock();
    try {
      //get datanode commands
      final int maxTransfer = blockManager.getMaxReplicationStreams()
          - xmitsInProgress;
      DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat(
          nodeReg, reports, getBlockPoolId(), cacheCapacity, cacheUsed,
          xceiverCount, maxTransfer, failedVolumes, volumeFailureSummary);
      ...

      return new HeartbeatResponse(cmds, haState, rollingUpgradeInfo,
          blockReportLeaseId);
    } finally {
      // 操作结束释放锁操作
      readUnlock("handleHeartbeat");
    }
  }

所以传统的心跳处理方法会存在一定的锁的竞争。随后的块报告的更新逻辑部分，两种方式基本一致，详细代码大家可以再HDFS-9239上进行阅读学习。

最后简单说两句，笔者个人觉得这个功能在集群规模比较大的情况下时，作用会比较明显，因为NameNode、DataNode会比较容易出现忙碌状态。还有一点注意，此功能默认是不开启的，使用的时候需要在配置项dfs.namenode.lifeline.rpc-address中配置RPC地址来启动此功能。

参考资料

---

[1].DataNode Lifeline Protocol: an alternative protocol for reporting DataNode liveness
[2].DataNode-Lifeline-Protocol.pdf