请选择 进入手机版 | 继续访问电脑版

Alluxio的Raft HA实现

[复制链接]
丁翼 发表于 2021-1-1 18:31:17 | 显示全部楼层 |阅读模式 打印 上一主题 下一主题
文章目次



前言

  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,代码如下:
  1. /** * A primary selector backed by a Raft consensus cluster. */@ThreadSafepublic class RaftPrimarySelector extends AbstractPrimarySelector {  /**   * Notifies leadership state changed.   * @param state the leadership state   */  public void notifyStateChanged(State state) {    setState(state);  }  @Override  public void start(InetSocketAddress address) throws IOException {    // The Ratis cluster is owned by the outer {@link RaftJournalSystem}.  }  @Override  public 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方法的执行,相关代码如下:
  1.   @Override  public 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):
  1.   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))) {      // 判断当前脚色是否是Primary      if (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脚色之后执行的。
  1.   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:
  1.   @Override  public CompletableFuture 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 overhead      return EMPTY_FUTURE;    } catch (Exception e) {      return RaftJournalUtils.completeExceptionally(e);    }  }
复制代码
在上面的过程中,TransactionContext会被解析成详细的journal entry,然后apply到master state里去。
  1.   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

来源:https://blog.csdn.net/Androidlushangderen/article/details/111939044
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发布主题

专注素材教程免费分享
全国免费热线电话

18768367769

周一至周日9:00-23:00

反馈建议

27428564@qq.com 在线QQ咨询

扫描二维码关注我们

Powered by Discuz! X3.4© 2001-2013 Comsenz Inc.( 蜀ICP备2021001884号-1 )