基于QJM/Qurom Journal Manager/Paxos的HDFS HA原理及代码分析

转载 http://yanbohappy.sinaapp.com/?p=205

HDFS HA的解决方案可谓百花齐放，Linux HA, VMware FT, shared NAS+NFS, BookKeeper, QJM/Quorum Journal Manager, BackupNode等等。目前普遍采用的是shard NAS+NFS，因为简单易用，但是需要提供一个HA的共享存储设备。而社区已经把基于QJM/Quorum Journal Manager的方案merge到trunk了，clouderea提供的发行版中也包含了这个feature，这种方案也是社区在未来发行版中默认的HA方案。本文从代码的角度分析这种方案的实现。

关于HDFS源代码中HA机制的整体框架实现，Active NN, Standby NN两种角色各自的代码执行流程，client如何做failover，为什么要fencing，每个DN要向Active和Standby NN都要report block，这样才能保证hot standby等等类似的问题，可以参考前面的文章: http://yanbohappy.sinaapp.com/?p=50
和 http://yanbohappy.sinaapp.com/?p=55 。

在HA具体实现方法不同的情况下，HA框架的流程是一致的。不一致的就是如何存储和管理日志。在Active NN和Standby NN之间要有个共享的存储日志的地方，Active NN把EditLog写到这个共享的存储日志的地方，Standby NN去读取日志然后执行，这样Active和Standby NN内存中的HDFS元数据保持着同步。一旦发生主从切换Standby NN可以尽快接管Active NN的工作（虽然要经历一小段时间让原来Standby追上原来的Active，但是时间很短）。

说到这个共享的存储日志的地方，目前采用最多的就是用共享存储NAS+NFS。缺点有：1）这个存储设备要求是HA的，不能挂掉；2）主从切换时需要fencing方法让原来的Active不再写EditLog，否则的话会发生brain-split，因为如果不阻止原来的Active停止向共享存储写EditLog，那么就有两个Active NN了，这样就会破坏HDFS的元数据了。对于防止brain-split问题，在QJM出现之前，常见的方法就是在发生主从切换的时候，把共享存储上存放EditLog的文件夹对原来的Active的写权限拿掉，那么就可以保证同时至多只有一个Active
NN，防止了破坏HDFS元数据。

Clouera为解决这个问题提出了QJM/Qurom Journal Manager，这是一个基于Paxos算法实现的HDFS HA方案。QJM的结构图如下所示：





QJM的基本原理就是用2N+1台JournalNode存储EditLog，每次写数据操作有大多数（>=N+1）返回成功时即认为该次写成功，数据不会丢失了。当然这个算法所能容忍的是最多有N台机器挂掉，如果多于N台挂掉，这个算法就失效了。这个原理是基于Paxos算法的，可以参考 http://en.wikipedia.org/wiki/Paxos_(computer_science) 。

用QJM的方式来实现HA的主要好处有：1）不需要配置额外的高共享存储，这样对于基于commodity hardware的云计算数据中心来说，降低了复杂度和维护成本；2）不在需要单独配置fencing实现，因为QJM本身内置了fencing的功能；3）不存在Single Point Of Failure；4）系统鲁棒性的程度是可配置的（QJM基于Paxos算法，所以如果配置2N+1台JournalNode组成的集群，能容忍最多N台机器挂掉）；5）QJM中存储日志的JournalNode不会因为其中一台的延迟而影响整体的延迟，而且也不会因为JournalNode的数量增多而影响性能（因为NN向JournalNode发送日志是并行的）。

1, NameNode格式化和启动

关于HDFS NN的元数据管理逻辑，FSImage和EditLog相关的源代码分析请参考：http://yanbohappy.sinaapp.com/?p=84 和http://yanbohappy.sinaapp.com/?p=101，NN的这部分代码在不同的HA解决方案中是一样的。先格式化HDFS，生成存放FSImage和EditLog的目录，目录初始化，把文件系统元数据持久化到文件。然后在启动的时候加载最新的FSImage和在那之后的EditLog。

NN存放FSImage和EditLog的目录用NNStorage这个类来管理。看FSImage的构造函数，传进去两个URI的集合，分别是存放FSImage和EditLog的地方。我们一直在说的NN HA解决的是EditLog的共享存储问题，不包括FSImage，除非我们配置把这两个东西存储到一个地方，但是一般是不会这么做的。

就像前面说的，EditLog的管理相对FSImage要复杂很多。所以接下来就是fsImage.getEditLog().initJournalsForWrite()来初始化存放日志的地方。这个在FSEditLog.initJournals()中完成，对于基于File的共享存储（NFS）来说，就是创建了一个用于管理这个设备和其中EditLog文件的FileJournaManger，然后加入EditLog.journalSet集合统一管理；而对于其他类型的共享存储(BookKeeper,QJM,BackupNode)则是创建对应的JournalManager对象。对于我们的QJM来说，就是创建了一个QuorumJournalManager对象。

2，构造QuroumJournalManager

QuromJournalManager的构造函数初始化了一些变量，比如这个QJM的uri,nsinfo等。注意我们使用URI来区分不同的JournalNode集群，JournalNode集群的URI表示类似于Zookeeper，这里有个例子，我们在hdfs-site.xml中会通过以下的形式配置使用QJM：

然后构造一个AsyncLoggerSet对象用于管理该QuromJournalManager向多个JournalNode的连接。这个AsyncLoggerSet对象里面有个存放AsyncLogger接口（真正的实现类是IPCLoggerChannel，每个IPCLoggerChannel管理QJM与一个JournalNode的连接和异步通信）的List。

this.loggers = new AsyncLoggerSet(createLoggers(loggerFactory));

进一步调用

static List<AsyncLogger> createLoggers (Configuration conf,

URI uri, NamespaceInfo nsInfo, AsyncLogger.Factory factory)

创建一系列的AsyncLogger用于写Log

3，格式化qjournal

New出这个FSImage对象之后就该初始化了，FSImage.format()->editLog.formatNonFileJournals (ns) 负责format非File的EditLog存放URI。进一步会调用

QuorumJournalManager.format(NamespaceInfo nsInfo)会格式化qjournal。

这里就涉及到QJM的核心原理Paxos算法了。关于这个算法的原理大家可参考wikipedia，简单说就是Active NN把日志写到2N+1个JournalNode上，每次写日志的操作只要其中a quorumof JNs(即大多数，大于等于N+1台JN)返回成功即认为这次操作是成功的。但是这个format操作是比较特殊的，要求所有的JN返回都是成功的才行，因为它相当于是做了个初始化的工作。在后面的写数据的过程中，只要大多数success
response就认为这次写成功了。

QuorumCall这个类包装了整个异步调用的过程：每次QuorumJournalManager对象向2N+1台JN发送写日志请求都是异步的，发出之后不是同步等待每个JN的返回值，而是注册一个callback函数，每当有一个返回，就把response计数加1（如果返回是success，把success计数加1；如果返回是failure，把failure计数加1）。这样QuorumJournalManager这端只需要发出去请求，然后循环检测时候有足够的success response或者足够的exception或者是time
out。

上面在代码中提到了RPC，QJM的RPC主要就一个协议类：QuorumJournalManager与多个JournalNode通信的协议QJournalProtocol。那么RPC的通信双方的实体类分别是哪个呢？客户端（QuorumJournalManager）是QJournalProtocolTranslatorPB；服务器端(JournalNode)是JournalNodeRpcServer。

看看这个format命令到了JN端做了哪些事情？

先是根据journalId创建了Journal对象，然后调用Journal.format()。接下来就是创建本地存储目录，创建Journal元数据，写元数据到目录等。由于JournalNode管理本地的数据采用的是FileJournalManager对象，所以后面的逻辑跟使用FileJournalManager的NN很像了。

4，NN发生主从切换

接下来就该看看一个Standby NN由Standby变成Active时，需要执行哪些操作：

1) fencing原来Active NN的写。

2) recover in-progress logs。原来Active NN写EditLog过程中发生了主从切换，那么处在不同JournalNode上的EditLog的数据可能不一致，需要把不同JournalNode上的EditLog同步一致，并且finalized。（这个过程类似于HDFS append中的recover lease的过程）

3) startLogSegment。不一致的EditLog都同步一致且finalized，那么原来的Standby NN正式行驶正常的Active NN的写日志功能。

4) write edits

5) finalizeLogSegment

4.1，fencing原来Active NN的写

前面说过，基于QJM的HA不需要处理fencing问题。这是怎么做到的呢？解决这个问题靠的是epoch number，这个和Paxos算法中选主（master election）所做的工作类似。

当Active 和Standby NN 发生主从切换时，原来的Standby NN需要执行：

NameNode.startActiveServices()->FSNamesystem.startActiveServices()->FSEditLog.recoverUnclosedStreams()->JournalSet.recoverUnfinalizedSegments()->QourumJournalManager.recoverUnfinalizedSegment()。这个过程说白了就是给原来的Active NN擦屁股，也可以算作是Standby要接管qjournal写权利的开始。这里面就出现了我们所说的brain-split的问题，Standby
NN怎么保证原来的Active NN已经不再往qjournal上写数据了。看看QourumJournalManager.recoverUnfinalizedSegment()是怎么实现的：

Epoch解决了我们所说的问题，Standby NN向每个JournalNode发送getJournalState RPC请求，JN返回自己的lastPromisedEpoch。QuorumJournalManager收到大多数JN返回的lastPromisedEpoch，在其中选择最大的一个，然后加1作为当前QJM的epoch，同时通过发送newEpoch RPC把这个新的epoch写到qjournal上。因为在这之后每次QuorumJournalManager在向qjournal执行写相关操作（startLogSegment(),logEdits(),finalizedLogSegment()等）的时候，都要把自己的epoch作为参数传递过去，写相关操作到达每个JournalNode端会比较如果传过来的epoch如果小于JournalNode端存储的lastPromisedEpoch，那么这次写相关操作会被拒绝。如果大多数JournalNode都拒绝了这次写相关操作，这次操作就失败了。回到我们目前的逻辑中，在主从切换时，原来的Standby
NN把epoch+1了之后，原来的Active NN的epoch就肯定比这个小了，那么如果它再向qjournal写日志就会被拒绝。因为qjournal不接收比lastPromisedEpoch小的QJM写日志。

看看JN收到newEpoch RPC之后怎么办：JN检查来自QJM的这个epoch和自己存储的lastPromisedEpoch：如果来自writer的epoch小于lastPromisedEpoch，那么说明不允许这个writer向JNs写数据了，抛出异常，writer端收到异常response，那么达不到大多数的success response，就不会有写qjournal的权限了。（其实这个过程就是Paxos算法里面选主的过程）

4.2 recover in-progress logs

接着上面的代码，Standby已经通过createNewUniqueEpoch()来fencing原来的Active，这个RPC请求除了会返回epoch，还会返回最后一个log segment的txid。因为只有最后一个log segment可能需要恢复。这个recover算法就是Paxos算法的一个实例(instance)，目的是使得分布在不同JN上的log segment的数据达成一致。

接下来就开始recoverUnclosedSegment()恢复算法

代码中留给我们一个问题，就是什么样的log segment是更好的，在recover的过程中被选为同步源呢。详细的设计可以参考Todd写的<<Quorum-Journal Design>>https://issues.apache.org/jira/secure/attachment/12547598/qjournal-design.pdf 的2.9和2.10。在代码中的实现是SegmentRecoveryComparator类。

简单描述下原理就是：有finalized的不用in-progress的；如果有多个finalized必须length一致；没有finalized的看谁的epoch更大；如果前面的都一样就看谁的最后一个txid更大。

在<<Quorum-Journal Design>>中有具体的例子。我看完这块之后感觉和HDFS append的block recover过程中选择同步源的思路有异曲同工之妙。

经历了上面的QourumJournalManager.recoverUnfinalizedSegment()过程，不完整的log segment都是完整的了，接下来就是调用NameNode.startActiveServices()->FSNamesystem.startActiveServices()->EditLogTailer.catchupDuringFailover()->EditLogTailer.doTailEdits()，原来Standby NN先去和原来Active NN同步EditLog，然后把EditLog执行，这时两台NN内存数据才真正一致。这里会调用QuorumJournalManager.selectInputStreams()从JNs中读取EditLog。而且目前HDFS中只有finalized
edit log才能被Standby NN读取并执行。在Standby NN从JNs读取EditLog时，首先向所有的JN节点发送getEditLogManifest() RPC去读取大于某一txid并且已经finalized edit log segment，收到大多数返回success，则把这些log segment整合成一个RedundantEditLogInputStream，然后Standby NN只要向其中的一台JN读取数据就行了。

至此原来的Standby NN所做的擦屁股的工作就结束了，那么它就正式变成了Active NN，接下来就是正常的记录日志的工作了。

4.3 startLogSegment

NameNode.startActiveServices()->FSNamesystem.startActiveServices()->FSEditLog.openForWrite()->FSEditLog.startLogSegmentAndWriteHeaderTxn()->FSEditLog.startLogSegment()->JournalSet.startLogSegment()->JournalSet.startLogSegment()->QuorumJournalManager.startLogSegment()。QJM向JNs发送startLogSegment
RPC调用，如果收到多数success response则成功，用这个AsynaLogSet构造QuorumOutputStream用于写log。

4.4 write edits

写EditLog的过程：FSEditLog.logEdit()->QuorumOutputStream.write()把Log写到QuorumOutputStream的double buffer里面。

Log持久化的过程：FSEditLog.logSync()->EditLogOutputStream.flush()->QuorumOutputStream.flushAndSync()，在这个函数里通过AsyncLoggerSet.sendEdits()调用Journal RPC把对应的日志写到JNs，同样是大多数success response即认为成功。如果大多数返回失败的话，这次logSync操作失败，那么NameNode会abort，因为没法正常写日志了。

4.5 finalizedLogSegment

流程和startLogSegment基本一样。