Ozone OM服务HA原理分析

2024-02-26 11:36:40

文章目录

前言

在分布式系统中，为了避免中心服务节点存在单点问题，我们往往会有HA(High Availability)的防御措施。比如一个简单的HA部署方案，额外在另一个机器上部署一个完全一样的服务。当当前的这个服务出现问题时，能够快速切换到这个用来做HA的服务。HA服务切换本身并不是重点，这里的重点是如何保持HA服务间的状态同步。这点在分布式系统的HA设计实现中尤其之关键。本文笔者来聊聊Ozone对象存储系统中核心服务OzoneManager的HA的设计实现原理。

OzoneManager HA的目标

在之前Ozone的alpha版本中，OM服务都是以单点服务部署的方式来处理外部请求的。显然这种缺乏HA的运行模式在运用到实际生产环境中将会存在风险。于是Ozone社区在JIRA[HDDS-505: OzoneManager HA]下设计实现了OM HA的特性。同样是作为分布式存储系统的HA实现，OzoneManager的HA实现和HDFS NameNode HA实现有许多异同之处。

在官方社区的OM HA的设计文档中，列出了以下几点目标实现要点：

  • 底层使用Raft协议来实现OM服务的HA特性。换句话说，这里将不是只有1个Active，另1个Standby服务这样的模式。而是2N+1的模式，1个OM Leader服务，2N个OM Follower服务。OM Follwer间通过投票选举出Leader服务。中间投票选举的过程遵从的就是Raft协议。这点即OM HA服务实现基于的一大核心要点。
  • 有了OM Leader，OM Follower服务之后，另外一个要点是Leader/Follower间的状态同步控制，以此保证服务在角色切换后，还能保证完全一致的服务状态，就是我们所说的Strong Consistency。
  • OM HA服务对于用户的透明性。用户理应对于OM HA完全透明，它在配置了给定的多OM服务地址后，就可以进行服务的请求了了。不管背后的服务处于什么角色状态，客户端用户都无须做任何调整。换言之，客户端始终能找到正确的OM服务，进行请求的发送，而不是单一定向的定点服务发送。

OM HA的Raft方式实现

OM HA中使用了Raft做HA的底层实现，那么它是如何工作的呢？以及它是如何做服务状态的同步的呢？

OM HA在这里使用了Raft的Java实现库Apache Ratis，里面实现了基于Raft协议的Leader选举以及Leader/Follower的状态一致性的控制。不过今天笔者并不准备展开篇幅阐述Raft中一致性同步控制的具体机理，我们还是着重谈谈OM HA本身。

因为有了Raft库Ratis的底层一致性的实现保证后，在上层OM中，它需要做的事情就变得清晰，明了了。在Apache Ratis中，是通过定义StateMachine状态机的方式来表示一个服务的当前状态。外界的每次请求就是此StateMachine即将要apply的一个新的State，每次apply完这个State后，整个状态机就会变为下一个State。因此，多个服务初始拥有相同初始State的StateMachine，在经历了若干次顺序一致的状态改变后，StateMachine的最终状态必然是一致的。这就是基于StateMachine的保证状态同步的核心原理。

这里我们将StateMachine的抽象概念定义转化为更为具体化的概念定义：

  • 每次外界而来的待apply的State即为外界的request请求。
  • 服务的StateMachine为OM服务的metadata信息，包括volume, bucket,key信息等各种表信息。

在Leader/Follower的HA模式下，请求状态的处理流程如下：
1）客户端发送请求给Leader处理
2）Leader接收到客户端请求，然后将请求复制分发到其它Follower上
3）Leader服务获取到超过半数以上Follower收到请求信息的ack回复之后，apply用户请求到内部状态机。
4）Leader会通知其它Follower去apply它们接收到的请求。

上述请求我们可以理解为是Transaction，从StateMachine的角度来看，概念叫做log entry。Leader复制请求到其它Follower的过程为log的write过程, 如果log被其它Follower成功接收到了，则意为此log为committed的log。

因此在这里StateMachine的实现异常关键，Ozone在这里实现了其内部使用的状态机类OzoneManagerStateMachine，对应的2个核心方法，log entry的写入，和log entry的状态apply方法：

  /**
   * Validate/pre-process the incoming update request in the state machine.
   * @return the content to be written to the log entry. Null means the request
   * should be rejected.
   * @throws IOException thrown by the state machine while validating
   */
  @Override
  public TransactionContext startTransaction(
      RaftClientRequest raftClientRequest) throws IOException {
    ByteString messageContent = raftClientRequest.getMessage().getContent();
    OMRequest omRequest = OMRatisHelper.convertByteStringToOMRequest(
        messageContent);

    Preconditions.checkArgument(raftClientRequest.getRaftGroupId().equals(
        raftGroupId));
    try {
      handler.validateRequest(omRequest);
    } catch (IOException ioe) {
      TransactionContext ctxt = TransactionContext.newBuilder()
          .setClientRequest(raftClientRequest)
          .setStateMachine(this)
          .setServerRole(RaftProtos.RaftPeerRole.LEADER)
          .build();
      ctxt.setException(ioe);
      return ctxt;
    }
    return handleStartTransactionRequests(raftClientRequest, omRequest);
  }

  /*
   * Apply a committed log entry to the state machine.
   */
  @Override
  public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
    try {
      OMRequest request = OMRatisHelper.convertByteStringToOMRequest(
          trx.getStateMachineLogEntry().getLogData());
      long trxLogIndex = trx.getLogEntry().getIndex();
      CompletableFuture<Message> ratisFuture =
          new CompletableFuture<>();

      applyTransactionMap.put(trxLogIndex, trx.getLogEntry().getTerm());
      CompletableFuture<OMResponse> future = CompletableFuture.supplyAsync(
          () -> runCommand(request, trxLogIndex), executorService);
      future.thenApply(omResponse -> {
        if(!omResponse.getSuccess()) {
          if (omResponse.getStatus() == INTERNAL_ERROR) {
            terminate(omResponse, OMException.ResultCodes.INTERNAL_ERROR);
          } else if (omResponse.getStatus() == METADATA_ERROR) {
            terminate(omResponse, OMException.ResultCodes.METADATA_ERROR);
          }
        }

        // For successful response and for all other errors which are not
        // critical, we can complete future normally.
        ratisFuture.complete(OMRatisHelper.convertResponseToMessage(
            omResponse));
        return ratisFuture;
      });
      return ratisFuture;
    } catch (Exception e) {
      return completeExceptionally(e);
    }
  }

从上述方法我们可以看到，只有在apply State操作的时候，才是真正请求处理执行的阶段。

Ozone的读写请求的区别处理

Ozone为了保证请求状态处理的一致，引入了StateMachine的方式来做一致性的处理。但并不是所有的请求都需要以这样的方式被处理，比如那些不会影响服务内部状态的请求，即READ类型的请求，其实可以直接被服务处理返回。

简单来说，Client甚至可以直接请求到其中任何一个服务，然后被服务处理得到返回结果。但是为了保证服务内部持有最新的状态数据，Ozone目前只允许Leader服务接受READ读请求，客户端发送到Follower的请求会被failover到当前的Leader服务中去。然后直接被Leader服务的RPC Server处理逻辑所处理。

WRITE类型的请求同样会被failover到Leader服务，但是它会有上述提到的replicate request的额外过程，会有上述startTransaction(log entry written)，applyTransaction(app transaction log)的过程来保证状态一致性的处理。

对于请求客户端而言，它在发起请求前并不知道哪些服务是Leader服务，因此在Client层面需要实现一个FailOverProxy类，做自动的重定向处理，Ozone这边专门实现了这样的类，叫做OMFailoverProxyProvider。

这个FailoverProxy在处理返回得到Follower服务返回的OMNotLeaderException后，会提取里面的Leader ID信息，进行请求的重试，并failover当前指向的OM为建议的Leader OM服务。

相关代码逻辑如下，

if (exception instanceof ServiceException) {
          OMNotLeaderException notLeaderException =
              getNotLeaderException(exception);
          if (notLeaderException != null &&
              notLeaderException.getSuggestedLeaderNodeId() != null) {
            // We need to failover manually to the suggested Leader OM Node.
            // OMFailoverProxyProvider#performFailover() is a dummy call and
            // does not perform any failover.
            omFailoverProxyProvider.performFailoverIfRequired(
                notLeaderException.getSuggestedLeaderNodeId());
            return getRetryAction(RetryAction.FAILOVER_AND_RETRY, failovers);
          }

OM服务的定期Snapshot、Checkpoint行为

和HDFS的FSImage Checkpoint行为类似，Ozone为了OM服务避免长时间做log entry的状态重建，也做了定期生成Snapshot的行为处理。由于Follower定期做Checkpoint，然后Leader定期向Follower上download最新的Snapshot到其本地。新加入的OM服务也可以快速从Leader 服务中bootstrap相关Snapshot文件和log数据，进行状态的快速同步。

综合上述几方面的模块描述，OM HA的实现整体架构如下图所示：

Ozone OM服务HA原理分析

附：DoubleBuffer+Table Cache的请求处理

Ozone社区在实现OM HA模式下的请求处理时，额外设计实现了基于双缓冲+Table Cache的高效请求处理方式，大致原理如下：

  • OM的StateMachine在处理transaction请求时，先将请求执行到Table Cache中，然后返回一个OMClientResponse对象。OMClientResponse定义了执行写入db具体操作逻辑。
  • OM将OMClientResponse加入到OM DoubleBuffer中。
  • OM DoubleBuffer内部线程定期同步OMClientResponse的transaction到db中，然后清理掉对应过期的Table Cache的entry项。

核心处理方法如下，这个处理逻辑同样适用于非HA模式，

  @Override
  public OMClientResponse handleWriteRequest(OMRequest omRequest,
      long transactionLogIndex) {
    OMClientRequest omClientRequest =
        OzoneManagerRatisUtils.createClientRequest(omRequest);
    OMClientResponse omClientResponse =
        omClientRequest.validateAndUpdateCache(getOzoneManager(),
            transactionLogIndex, ozoneManagerDoubleBuffer::add);
    return omClientResponse;
  }

引用

[1]. https://issues.apache.org/jira/browse/HDDS-505 . OzoneManager HA

Ozone OM服务HA原理分析Ozone OM服务HA原理分析 Android路上的人 博客专家 发布了373 篇原创文章 · 获赞 403 · 访问量 203万+ 他的留言板 关注
上一篇：模拟散列表


下一篇：python的matplotlib的热门可视化动图