hadoop HA----Quorum Journal 设计…

原文参考这个链接中的附件：https://issues.apache.org/jira/browse/HDFS-3077

1 概述

1.1 背景

1.2 当前实现的一些局限

1. 自定义硬盘 - NAS设备和远程控制的PDU非常昂贵，也有别于标准部署。
2. 复杂的部署 - HDFS安装完成后，管理员必须用额外的步骤去配置NFS挂载，自己写互斥脚本，等。这样复杂的HA部署可能因为一点错误配置导致不可用。
3. 没有很好的NFS客户端 - 很多linux版本中，NFS客户端的实现都有bug，容易出现误配。例如，管理员很容易弄错挂载选项，导致Namenode被冻结并不可恢复。

1.3 一个替换方案的需求

1.3.1 不同点

1. 没有特殊硬件的需求 - 用普通的商品机器就能满足设计，和现有的hadoop集群机器一样就行。
2. 不需要额外自己弄一些互斥配置 - 所有需要的互斥都由软件完成，构建进系统了。
3. 没有单点故障 - 作为HA的解决方案，edit logs的存储应该是完全HA的。

1.3.2 正确性需求

​ 当然我们要保证HDFS中任何修改edit logs的正确性需求：

1. 任何同步了的edit操作必须不能被忘记 - 如果NameNode成功的调用FSEditLog.logSync()，所有的同步的修改必须被持久化的记录下来，即使有任何失败。
2. 任何还没同步的edit操作可能被忘记，也可能不会 - 如果一个NameNode写一个edit，并且在调用logSnyc()之前或者调用中挂掉，系统可能记住这个edit，也可能忘记。
3. 如果一个edit被读了，它不能被忘记 - 如果StandbyNode读取尾部的edits并且看到了某条edit，那这条edit就不能被忘记。
4. 对于任何给定的txid，这必须要恰好有一条可用的事件 - 任何节点用一个给定的ID读事件，然后其它节点用相同的ID读到的数据必须是一样的。

1.3.3 额外目标

1. 可配置任意允许失败的节点数 - 如果管理员想容忍多余一个节点的失败，他可以配置多一点节点来达到他期望的容忍节点数。具体说，配置2N+1个节点，就能够容忍N个节点失败。
2. 一个慢journal节点不会影响延时 - 如果一个存储edits的节点变慢或者挂掉，系统应该继续操作而没有延时的惩罚。当一个节点挂掉时，我们不应该由于超时而停止客户端的编辑操作。
3. 添加新的journal节点不会影响延时 - 为了容忍多余一个的失败，管理员会配置5个或者更多的journal节点。和journal节点间的通信应该是并行的，这样添加节点才不会引起延时的线性增长。

1.3.4 运维的需求

 接下来的需求不是和算法相关的，但是对于部署HDFS集群有着重要的意义：

1. Metrics/日志 - 任何后台进程的引入，应该集成HDFS现有的Metrics和日志系统。这对现存的监控架构是必要的。
2. 配置 - 任何必要的配置应该和现有的xml一致。
3. 安全 - 任何跨多节点的操作应该(a)相互授权 和(b)加密，采用现有的Hadoop机制。例如任何IPC/RPC交互应该用SASL-based传输附带Kerberos提供的相互授权。任何ZooKeeper的使用，应该支持ZooKeeper ACLs和授权。

1.4 Quorum-based 方法

2 设计-写logs

2.1 总共有两个组件：

1. QuorumJournalManager, 运行在每个Namenode上（现在的ha集群最多两个Namenode），它通过rpc联系JournalNodes，发送编辑、互斥、同步等命令。
2. JournalNode进程，运行在N个机器上，暴露hadoop ipc接口允许QuormJournalManager远程写edits到它的本地磁盘上

2.2 QuorumJournalManager工作流

1. 互斥写者 - 它必须保证没有其它的QJM在写edit logs。这是一种互斥机制，即使两个Namenodes都认为它自己是活跃状态，并进行写入edit logs，互斥机制会保证只有一个Namenode会写入成功。后面会详细说明这个互斥机制。
2. 恢复正在写入的logs - 一个写者前一次写logs失败了，它可能造成不同备份中出现不同长度的log（例如：前一个写者只发送了edit给三个JNs中的一个，然后挂掉了）。我们必须先同步logs。
3. 开始一个新的log段 - 现有的实现中，这是写edit logs的正常流程。
4. 写edits - 对于每一批edits，写者发送这一批edits给所有的JNs。一旦它接收到超过半数JNs的返回成功，它就认为这次写入成功了。写者维持一个写入过程的pipeline，这样就算临时有一个节点变慢也不会影响整个系统的吞吐和延时。 如果一个JN失败，或者回复得太慢导致超时，这个JN就会被标记为outOfSync，在当前的log段就不再使用这个JN了。只有大于半数的JNs还活着，就不会出问题。之前失败的那个节点，会在下一轮的edit log中被重试。
5. 完成log段 - 现有的实现中，QJM发送一个完成log段RPC给所有的JNs，当接收到大于半数的JNs确认后，这个log段被认为完成，下一个log段可以开始。
6. Go to step 3

2.3 互斥写者

• 当一个写者变为活跃时，会分配给它一个epoch number
• 每一个epoch number都是唯一的，没有任意两个写者有相同的epoch number
• epoch numbers定义了写者顺序，对于任意的两个写者，epoch numbers定义了一种关系，一个写者被认为更后于另一个写者，当 且仅当它的epoch number更加大一些。

• 在对edit logs做任何改动的时候，QJM必须要被分配一个epoch number
• QJM发送它的epoch number给所有JNs，包含在消息newEpoch(N)中。它不会用epoch number进行处理，除非大于半数的JournalNodes返回一个成功指示。
• 当JN回复了这样的请求，它会记录这个epoch number在变量lastPromisedEpoch中，这个变量会被写入磁盘。
• 任何请求改变edit logs的RPC，必须包含请求者的epoch number。
• 在任何RPC动作（除了newEpoch()）之前，JournalNode拿请求者的epoche number和它自己的lastPromisedEpoch做比较。如果请求者的epoch更小，它就会拒绝这次请求。如果请求者的epoch更大，它会更新自己的lastPromisedEpoch。这会使JN更新自己的lastPromisedEpoch，即使它挂掉的那会有新的的写者变成活跃者。

2.4 写者epochs

这样做得重要性，在接下来讨论段恢复的边界条件时，变得清晰。

2.5 产生epoch numbers

1. QJM发送getJournalState()给所有JNs。每个JN回复它自己的lastPromisedEpoch。
2. 接收到大于一半的JNs回复，QJM拿出它接收到的最大值，然后把它加一，这就产生了一个proposedEpoch。
3. QJM发送newEpoch(proposedEpoch)给所有的JNs，每个JN原子的比较这个建议值和它当前的lastPromisedEpoch。如果这个新的建议值比它存储的值大，则更新它的lastPromisedEpoch为新值，并且返回成功。如果小，则返回失败。
4. 如果QJM接收到超过一半的JNs的返回成功，则设置它的epoch number为proposedEpoch。否则，它中止尝试成为一个活跃的写者，并抛出一个IOException。这个异常被Namenode同写NFS失败一样的方式处理。-- 如果QJM被用作共享的edits单元，它将会导致Namenode挂掉。

40,319 INFO QJM - Starting recovery process for unclosed journal segments...

40,320 TRACE Outgoing IPC) - 1: Call -> null@/127.0.0.1:39595:

        getJournalState {jid { identifier: "test-journal" }}

40,323 TRACE IPC Response) - 1: Response <- null@/127.0.0.1:39595:

        getJournalState {lastPromisedEpoch: 0 httpPort: 45029}

40,323 TRACE Outgoing IPC) - 1: Call -> null@/127.0.0.1:36212:

        getJournalState {jid { identifier: "test-journal" }}

40,325 TRACE IPC Response) - 1: Response <- null@/127.0.0.1:36212:

        getJournalState {lastPromisedEpoch: 0 httpPort: 49574}

40,325 TRACE Outgoing IPC) - 1: Call -> null@/127.0.0.1:33664:

        getJournalState {jid { identifier: "test-journal" }}

40,327 TRACE IPC Response) - 1: Response <- null@/127.0.0.1:33664:

        getJournalState {lastPromisedEpoch: 0 httpPort: 36092}

40,329 TRACE Outgoing IPC) - 1: Call -> null@/127.0.0.1:39595:

        newEpoch {jid { identifier: "test-journal" } nsInfo { .. .} epoch: 1}

40,334 TRACE IPC Response) - 1: Response <- null@/127.0.0.1:39595: newEpoch {}

40,335 TRACE Outgoing IPC) - 1: Call -> null@/127.0.0.1:36212:

        newEpoch {jid { identifier: "test-journal" } nsInfo { .. .} epoch: 1}

40,339 TRACE IPC Response) - 1: Response <- null@/127.0.0.1:36212: newEpoch {}

40,339 TRACE Outgoing IPC) - 1: Call -> null@/127.0.0.1:33664:

        newEpoch {jid { identifier: "test-journal" } nsInfo { .. .} epoch: 1}

40,342 TRACE IPC Response) - 1: Response <- null@/127.0.0.1:33664: newEpoch {}

40,344 INFO QJM - Successfully started new epoch 1

2.6 同步logs

1. 确定最新log段的事件ID。
2. 确定最新一个事件ID N已经成功提交在大于一半的节点上。这就是等于确定了哪些JournalNode包含最新写入事件的log。
3. 确保大于一半的节点同步这个log段，从包括最新段的节点上拷贝。
4. 命令这些节点标记已完成log段。

1. 任何之前提交的事件必须存储在大于一半的节点中。
2. 大于一半的节点同意这些内容，并且以txid作为上一个log段的结束，并且标记为已完成段。

2.7 常量

 常量 1 一旦log段被标记完成，就不再会被标记非完成。

 常量 2 在任意节点上如果有一个log段开始于txid N，则有大于一半的节点包含一个已完成的log段结束于rxid N-1

常量 3 在任意节点上如果有一个已完成的log段结束于txid N，则有大于半数的节点有一个结束于txid N的log段。

2.8 恢复算法

• 已完成的log段必须包含所有之前提交的事件。一个事件被认为已提交只有当超过半数的JNs对前一个写者确认过这个事件。
• 这个log段必须标记为已完成在超过半数的journal节点上。
• 所有的日志记录者必须完成log段成相同的长度和内容。也就是说，如果有两个日志记录者包含一个已完成的log段开始于相同的事件ID，则这些log文件必须是语义一致的。

1. 确定哪个段需要恢复：依据newEpoch()的返回值，每一个JournalNode发送它的最大的log段事件ID。如果任何log段已经成功的开始在大于一半的节点上，NN就会找出这个段需要恢复（因为返回newEpoch()的这一大半节点和在新段中提交事件的一大半节点是肯定会有重叠的）
2. PrepareRecovery RPC：写者发送一个RPC给每一个JN要求准备恢复给定的段。没一个JN返回它自己本地磁盘中当前段的状态，包括长度和完成状态（已完成或者正处理）。
   这个RPC请求和回复分别对应Paxos中的Prepare(Phase 1a)和Promise(Phase 1b)。
3. AcceptRecovery RPC：基于PrepareRecovery的回复，写者指定一个段（这里是指一个JN）作为恢复的源。选出这个段是根据它必须包含所有之前提交的事件。这个决定的细节会在下面的2.9节描述。
   选择源后，写者发送AcceptRecovery RPC给每一个JournalNode，包含两个东西，段状态和可以获得这个log段拷贝的URL。
   AcceptRecovery RPC对应Paxos的Phase 2a，常叫做Accept。
   接收到AcceptRecovery RPC，JournalNode将会做以下动作：
   （a）Log同步：如果当前的磁盘log丢失，或者是长度不同，JN会从URL中下载log，替换掉当前的log。
   （b）持久化恢复元数据：JN写入磁盘一个数据结构，这个数据结构包含段状态和写者的epoch number。在未    
        来，对这个段的PrepareRecovery调用，这个状态和epoch number将被返回。
   如果这些动作成功的完成，JournalNode返回成功给写者。如果写者接收到大多数JournalNode的返回成功，它就会往后继续。
   JournalNode回复AcceptRecovery的行为，对应Paxos的Phase 2b。
4. 完成段：这个阶段，写者知道超过半数的JournalNodes已经有了一致的log段，并且持久化了恢复信息。因此，将来任何写者发起PrepareRecovery将会看到这次的决定并作出相同的结果。我们现在可以安全的完成这个log段（类似于Paxos的Commit phase）。写者简单的发起一个FinalizeLogSegment调用给每个JournalNodes。
   当接到这个时，JournalNodes设置log段标志为已完成。他们都可以删除掉已经持久化的段信息，因为已完成状态本身信息量已经足够了，对于做恢复决定。

2.9 Journal同步 - 选择恢复源

1. 如果一个节点回复它没有段开始于给定的事件ID，它不能成为恢复源。
2. 如果有节点已经完成了段，这表示前一次的恢复已经提交，没有再进行恢复的必要。这种情况下，包含已完成段的这个节点作为源。
3. 对于任何两个返回正在进行段的节点，他们按下面方式比较：
   （a）对每个JN，认为maxSeenEpoch比他的lastWriterEpoch更大，并且epoch number与之前任何接受了的恢复建议一样。
   （b）如果一个JN的maxSeenEpoch比另一个的更大，它就会是一个更好的恢复源。这个解释将在下节2.10.6中。
   （c）如果两个值相等，哪个JN的事件更多，就会被认为是更好的恢复源。

2.10 同步logs - 示例

2.10.1 正处理log不一致 - 大部分节点成功

JN segment last txidlastWriterEpoch

JN1 

JN2

JN3 

edits_inprogress_101

edits_inprogress_101

edits_inprogress_101

150
153
1531
1
1

2.10.2 正处理log不一致 - 还没有大部分节点成功

JN segment last txidlastWriterEpoch

JN1 

JN2

JN3 

edits_inprogress_101

edits_inprogress_101

edits_inprogress_101

150
153
1251
1
1

2.10.3 已完成log不一致 - 大部分节点成功

JNsegmentlast txidJN1
JN2
JN3edits_101-150
edits_101-150
edits_inprogress_101150
150
145

2.10.4 已完成log不一致 - 大部分节点未成功

JNsegmentlast txidJN1
JN2
JN3edits_101-150

edits_inprogress_101

edits_inprogress_101150
150
125

2.10.5 开始log不一致 - 大部分节点没开始

JNprev segmentcur segmentlast txidJN1
JN2
JN3edits_101-150
edits_101-150
edits_101-150edits_inprogress_151
-
-150
150
150

2.10.6 第一批log不一致 - 大部分节点没log

JNprev segmentcur segmentlast txidlast WriterEpochJN1
JN2
JN3edits_101-150
edits_101-150
edits_101-150edits_inprogress_151
-
-153
150
1501
1
1

JNprev segmentcur segmentlast txidlast WriterEpochJN1
JN2
JN3edits_101-150
edits_101-150
edits_101-150edits_inprogress_151
edits_inprogress_151
edits_inprogress_151153
151
1511
2
2

2.10.7 多次恢复 - 第一次失败

JN segment last txidacceptedInEpochlastWriterEpoch

JN1 

JN2

JN3 

edits_inprogress_101

edits_inprogress_101

edits_inprogress_101

150
153
125-
-
-1
1
1

JN segment last txidacceptedInEpochlastWriterEpoch

JN1 

JN2

JN3 

edits_inprogress_101

edits_inprogress_101

edits_101-150

150
153
1502
-
-1
1
1

2.11 写edits

1. 一旦logSync，拷贝队列中的数据到一个新的数组中。
2. 推送这些数据到每个远程的JournalNode的队列中。
3. 每个JN有个线程处理。这些线程发起logEdits RPCs给其它的JournalNodes。
4. 一旦接到，JournalNode (a)验证epoch number (b)验证每一批edits的事件ID，保证不会出现乱序和掉包 (c)写入并同步edits到当前的log段 (d)返回成功。
5. 最初的logSync线程等待一个超过半数的成功回复。如果超过半数的回复带有异常，或者超时，logSync()调用会抛出exception。这种情况下，QJM被用作共享的存储机制，它会引起NN的abort。

3 设计-读logs

当后备节点（StandbyNode）从JN上读的时候，它的第一件事是getEditLogManifest()RPC所有的节点。所有已完成的段返回并且合并在一起RedundantEditLogInputStreams，这样备份节点可以从任一节点上读取每一段。如果有一个JN在读取中失败，冗余的输入流可以自动的选择一个拥有相同段的不同节点恢复。

​​#linux​​ ​​#quorumJournal​​ ​​#hadoop​​