前言

Alluxio在HA的实现上，早期实现的方式是基于ZK(用来做领导选举)+shared journal storage(状态同步)的方式来达到其服务高可用性的，这种方式和HDFS的HA实现十分类似。不过后来Alluxio社区实现了基于Raft协议的新的HA实现方式，这里的Raft实现依赖了开源Raft Java实现库Apache Ratis。作为全新的HA实现，本文笔者结合Alluxio相关代码来简单聊聊里面的一些实现细节。

基于Raft实现的要点

Raft一致性协议算法目前逐渐被越来越多的大型系统所使用，比如对象存储系统Apache Ozone。Ozone内部的数据一致性控制依赖的实现也是Apache Ratis。

对于同样要依赖Apache Ratis做Raft HA实现的Alluxio系统来说，它需要特别关注哪几个要点的实现呢？这里笔者结合之前对于Ozone以及Apache Ratis的了解，列出以下几点：

StateMachine，状态机的定义，不同的系统它所谓的状态机的概念是不同的。比如以存储系统而言，大部分情况可理解为为master元数据的控制更新
Leader/Follower节点的选举，重新选举时的回调执行操作

以上两点是笔者认为做Raft实现需要尤其考虑实现的点，其它的部分我们再结合实际的系统实现做对应逻辑的适配修改。比如本文今天所讲述的这样的一个系统就是Alluxio。

Alluxio Raft HA实现的相关角色类

下面我们结合Alluxio的代码做Alluxio Raft HA实现的介绍。

首先一个主要的中心控制类RaftJournalSystem，此类里包括了状态机，raft journal writer等等与Raft journal HA实现的相关角色类。

RaftPrimarySelector， Primary选举监听类，当有新的leader选举时，此类会监听回调对应的执行执行。

RaftJournalWriter，Raft Journal信息的写出类。此类会调用Raft Client向其它master server组进行journal信息的写出。

JournalStateMachine，状态机的定义实现类，此类负责master状态的更新以及snapshot的定期take操作。

BufferedJournalApplier，负责apply journal信息到master的类。此类内部额外维护了一个suspend buffer队列，用类临时存放暂停时间段待apply的raft journal信息。

SnapshotReplicationManager，snapshot管理类，在Raft server中，Follower会进行snapshot的take并upload snapshot到Leader的Raft Server里。

Alluxio Raft HA部分场景分析

Leader重新选举监听处理

当发生了新的Leader选举时，Alluxio的master目前是怎么样的一个action操作？

首先我们来看与此相关的角色类，RaftPrimarySelector，代码如下：

/*** A primary selector backed by a Raft consensus cluster.*/
@ThreadSafe
public class RaftPrimarySelector extends AbstractPrimarySelector {/*** Notifies leadership state changed.* @param state the leadership state*/public void notifyStateChanged(State state) {setState(state);}@Overridepublic void start(InetSocketAddress address) throws IOException {// The Ratis cluster is owned by the outer {@link RaftJournalSystem}.}@Overridepublic void stop() throws IOException {// The Ratis cluster is owned by the outer {@link RaftJournalSystem}.}
}

此类基本继承父类的实现，只是对外方法里额外重置了一个状态。我们在基于ZK做Leader选举的时候，ZK是有提供对应接口监听得到新的Leader信息的。同理基于Raft实现的Apache Ratis同样有这么一个接口方法。

在JournalStateMachine的notifyLeaderChanged方法里，能监听到这个动作，随之会调用到RaftPrimarySelector#notifyStateChanged方法的执行，相关代码如下：

  @Overridepublic void notifyLeaderChanged(RaftGroupMemberId groupMemberId, RaftPeerId raftPeerId) {if (mRaftGroupId == groupMemberId.getGroupId()) {mIsLeader = groupMemberId.getPeerId() == raftPeerId;mJournalSystem.notifyLeadershipStateChanged(mIsLeader);} else {LOG.warn("Received notification for unrecognized group {}, current group is {}",groupMemberId.getGroupId(), mRaftGroupId);}}.../*** Notifies the journal that the leadership state has changed.* @param isLeader whether the local server is teh current leader*/public void notifyLeadershipStateChanged(boolean isLeader) {mPrimarySelector.notifyStateChanged(isLeader ? PrimarySelector.State.PRIMARY : PrimarySelector.State.SECONDARY);}

PrimarySelector在setState操作里会调用之前注册过的listener方法，因此在每次master状态发生变化的时候，它会执行下面的操作方法(FaultTolerantAlluxioMasterProcess#gainPrimacy)：

  private boolean gainPrimacy() throws Exception {// Don't upgrade if this master's primacy is unstable.AtomicBoolean unstable = new AtomicBoolean(false);try (Scoped scoped = mLeaderSelector.onStateChange(state -> unstable.set(true))) {// 判断当前角色是否是Primaryif (mLeaderSelector.getState() != State.PRIMARY) {unstable.set(true);}stopMasters();LOG.info("Secondary stopped");try (Timer.Context ctx = MetricsSystem.timer(MetricKey.MASTER_JOURNAL_GAIN_PRIMACY_TIMER.getName()).time()) {// 先让journal system变为Primary的角色，此过程会有transaction的catch up操作mJournalSystem.gainPrimacy();}// 如果不是Primary服务，则再执行对应非Primary相关的执行操作，比如stop journal writer// 随后返回if (unstable.get()) {losePrimacy();return false;}}// 以Primary master身份启动master服务startMasters(true);mServingThread = new Thread(() -> {try {startServing(" (gained leadership)", " (lost leadership)");} catch (Throwable t) {Throwable root = Throwables.getRootCause(t);if ((root != null && (root instanceof InterruptedException)) || Thread.interrupted()) {return;}ProcessUtils.fatalError(LOG, t, "Exception thrown in main serving thread");}}, "MasterServingThread");mServingThread.start();if (!waitForReady(10 * Constants.MINUTE_MS)) {ThreadUtils.logAllThreads();throw new RuntimeException("Alluxio master failed to come up");}LOG.info("Primary started");return true;}

Journal system在变为Primary过程中，会进行关键的journal的catch up操作，保证其内部StateMachine apply了最新的journal transaction。

与此对应的(FaultTolerantAlluxioMasterProcess#)losePrimacy方法，发生在master监听发现自身已经不是Primary角色之后执行的。

  private void losePrimacy() throws Exception {if (mServingThread != null) {stopServing();}// Put the journal in secondary mode ASAP to avoid interfering with the new primary. This must// happen after stopServing because downgrading the journal system will reset master state,// which could cause NPEs for outstanding RPC threads. We need to first close all client// sockets in stopServing so that clients don't see NPEs.mJournalSystem.losePrimacy();if (mServingThread != null) {mServingThread.join(mServingThreadTimeoutMs);if (mServingThread.isAlive()) {ProcessUtils.fatalError(LOG,"Failed to stop serving thread after %dms. Serving thread stack trace:%n%s",mServingThreadTimeoutMs, ThreadUtils.formatStackTrace(mServingThread));}mServingThread = null;// 停止内部服务stopMasters();LOG.info("Primary stopped");}// 以非Primary角色重启内部服务startMasters(false);LOG.info("Secondary started");}

JournalStateMachine的状态apply处理

另外一块关键的处理是JournalStateMachine状态机的状态apply处理，关键操作方法如下：

JournalStateMachine#applyTransaction:

  @Overridepublic CompletableFuture<Message> applyTransaction(TransactionContext trx) {try {applyJournalEntryCommand(trx);RaftProtos.LogEntryProto entry = Objects.requireNonNull(trx.getLogEntry());updateLastAppliedTermIndex(entry.getTerm(), entry.getIndex());// explicitly return empty future since no response message is expected by the journal writer// avoid using super.applyTransaction() since it will echo the message and add overheadreturn EMPTY_FUTURE;} catch (Exception e) {return RaftJournalUtils.completeExceptionally(e);}}

在上面的过程中，TransactionContext会被解析成具体的journal entry，然后apply到master state里去。

  private void applySingleEntry(JournalEntry entry) {...mNextSequenceNumberToRead++;if (!mIgnoreApplys) {// journal applier(BufferedJournalApplier)类负责完成此步骤mJournalApplier.processJournalEntry(entry);}}

这里的master state有多种子类的实现，比如InodeTreePersistentState。

Raft HA过程调用

上面小节只展示了部分的Raft HA过程处理，一个全局的HA过程调用图如下所示，笔者列出了文中提到的几个关键角色服务在图内，并没有涵盖所有的细节。

以上就是本文所阐述的主要内容了，有兴趣的同学可以阅读学习Alluxio 其它方式的HA的实现细节。

参考资料

[1].https://docs.alluxio.io/os/user/stable/en/deploy/Running-Alluxio-On-a-HA-Cluster.html#zookeeper-and-shared-journal-storage

Alluxio的Raft HA实现相关推荐

云知声 Atlas 超算平台: 基于 Fluid + Alluxio 的计算加速实践
Fluid 是云原生基金会 CNCF 下的云原生数据编排和加速项目,由南京大学.阿里云及 Alluxio 社区联合发起并开源.本文主要介绍云知声 Atlas 超算平台基于 Fluid + Alluxi ...
【联通】数据编排技术在联通的应用
欢迎来到[微直播间],2min纵览大咖观点,本期分享的题目是数据编排技术在联通的应用. 本次分享内容将围绕四个方面讲述Alluxio数据编排技术在联通的应用,主要围绕缓存加速.存算分离.混合负载以及轻 ...
Alluxio HA 写入文件失败
Alluxio HA环境,今天发生,用户无法写入文件的情况. 创建文件夹,是正常的.但是最后copyFromLocal 文件的时候,就没有任何反应.最后可以看到这个新建的文件.但是文件size是0. ...
sqlite c++插入 timestamp_Dqlite,基于sqlite 高可用(HA)数据库
原文发表于我的博客, 特此版权声明 noosphere.site: Dqlite,基于sqlite 高可用(HA)数据库 csdn : Dqlite,基于sqlite 高可用(HA)数据库 k3s之前 ...
java 状态机_基于 RAFT 一致性算法的 Java 实现 SOFAJRaft
SOFAJRaft 是一个基于 RAFT 一致性算法的生产级高性能 Java 实现,支持 MULTI-RAFT-GROUP,适用于高负载低延迟的场景. 使用 SOFAJRaft 你可以专注于自己的业务 ...
二进制部署Kubernetes v1.13.4 HA可选
本次采用二进制文件方式部署,本文过程写成了更详细的ansible部署方案 https://github.com/zhangguanzhang/Kubernetes-ansible 和之前的步骤差不多都 ...
私有云中Kubernetes Cluster HA方案
2019独角兽企业重金招聘Python工程师标准>>> 更多关于Kubernetes的深度文章,请到我oschina/WaltonWang的博客主页. Kubernetes Mast ...
Alluxio在多级分布式缓存系统中的应用
1. 前言随着移动互联网的发展,越来越多的业务数据和日志数据需要用户处理.从而,用数据去驱动和迭代业务发展.数据处理主要包括:计算和查询.计算主要为离线计算.实时流计算.图计算.迭代计算等:查询主要 ...
Alluxio代码结构
Alluxio是一款基于云原生开源的数据编排技术,为数据计算与数据存储构建了桥梁,支持将数据从原始存储层移动到加速计算的虚拟分布式存储系统.Alluxio可为数据湖计算提供统一的数据湖存储访问入口,支 ...
分布式一致性算法 - raft 图解
前言应用服务架构的发展是一个循序渐进的过程,从最开始的单机架构逐步演化到如今的基于微服务的分布式架构.与此同时,应用服务最底层的数据支撑 – 数据存储服务也逐渐由单节点单副本逐步往多节点多副本的方向 ...

Alluxio的Raft HA实现

文章目录

前言