Ozone Datanod Container Replication复制过程

前言

---

在分布式存储系统中，为了保证文件数据的冗余性，系统往往会对数据进行多副本的设置保存，比如典型的HDFS三副本文件设置。在Ozone中，同样也有副本的概念，不过它的副本单位粒度比HDFS的Block要粗一些。Ozone的是以Datanode上的Container为粒度，进行Container级别的Replication。在一个Container下 ，才是实际具体的block chunk文件。因此在Replication的数据规模上来说，Ozone的Container Replication要比HDFS的Block Replication要略微复杂一些。本文笔者来聊聊Ozone的Container Replication原理内容，在Ozone Datanode基于Container提供数据存储服务以及分布式元数据管理的模式下，它是如何来做Container副本复制的。

Ozone Datanode Container Replication请求处理

---

说到副本的复制，首先始发点比较容易能够联系到无非是主服务发现某个副本损坏了达不到期望副本数，然后发送新的副本复制请求命令到别的机器上进行Replication的复制。在Ozone中，同样也是如此。

因此第一步，类似地，Datanode收到了SCM服务的Replication请求命令，然后此请求被dispatch到对应的CommandHandler中被处理，

以下为类ReplicateContainerCommandHandler的handle方法处理逻辑：

  @Override
  public void handle(SCMCommand command, OzoneContainer container,
      StateContext context, SCMConnectionManager connectionManager) {

    final ReplicateContainerCommand replicateCommand =
        (ReplicateContainerCommand) command;
    final List<DatanodeDetails> sourceDatanodes =
        replicateCommand.getSourceDatanodes();
    final long containerID = replicateCommand.getContainerID();

    Preconditions.checkArgument(sourceDatanodes.size() > 0,
        "Replication command is received for container %s "
            + "without source datanodes.", containerID);

    // 1) 添加Replication task，并带上source node，目标远程拷贝数据的节点
    supervisor.addTask(new ReplicationTask(containerID, sourceDatanodes));
  }

上面的supervisor变量(ReplicationSupervisor实例类)为实际执行Replication task的executor service，

  /**
   * Queue an asynchronous download of the given container.
   */
  public void addTask(ReplicationTask task) {
    if (containersInFlight.add(task.getContainerId())) {
      executor.execute(new TaskRunner(task));
    }
  }

Ozone Datanode Container的远程拷贝以及Container导入

---

SCM的Replication请求被转化为Container Replication的task后，后面就是Container远程拷贝复制的过程了。鉴于Ozone是是基于Container粒度单位做数据存储的，它所需要复制的数据文件比HDFS单一block文件将会多不少。另外地，Ozone Datanode还需要复制Container文件的metadata文件。

Container在Datanode中是以目录的形式存储的，为了提升数据replication的速度，在拷贝过程中，Datanode这边的做法是把目录打出tar包，然后在解压这个Container tar包文件。

下面为实际数据拷贝的操作方法：

  public void replicate(ReplicationTask task) {
    long containerID = task.getContainerId();

    List<DatanodeDetails> sourceDatanodes = task.getSources();

    LOG.info("Starting replication of container {} from {}", containerID,
        sourceDatanodes);

    // 1) 从持有此副本Container中下载文件，tar包文件的方式下载到本地
    CompletableFuture<Path> tempTarFile = downloader
        .getContainerDataFromReplicas(containerID,
            sourceDatanodes);

    try {
      //wait for the download. This thread pool is limiting the paralell
      //downloads, so it's ok to block here and wait for the full download.
      Path path = tempTarFile.get();
      LOG.info("Container {} is downloaded, starting to import.",
          containerID);
      // 2) 解压tar包文件并导入Container数据到Datanode的元数据内
      importContainer(containerID, path);
      LOG.info("Container {} is replicated successfully", containerID);
      task.setStatus(Status.DONE);
    } catch (Exception e) {
      LOG.error("Container replication was unsuccessful .", e);
      task.setStatus(Status.FAILED);
    }
  }

可能有人会比较好奇，为什么这里会有导入元数据的过程，因为Ozone采用的是分布式元数据管理的方式，Container内部的Block数据由其实际存储节点所维护，并且保存于本地。SCM中心管理服务只管理Container级别的信息(相关细节文章可阅读此文章：Ozone Datanode的分布式元数据管理)。

Datanode随后会执行一次心跳汇报，将新增的Container信息报告给SCM，然后SCM就能知道最新的Container信息了。

相关Handler类KeyValueHandler的import Container处理逻辑，

  public Container importContainer(final long containerID,
      final long maxSize, final String originPipelineId,
      final String originNodeId, final InputStream rawContainerStream,
      final TarContainerPacker packer)
      throws IOException {

    // TODO: Add layout version!
    KeyValueContainerData containerData =
        new KeyValueContainerData(containerID,
            maxSize, originPipelineId, originNodeId);

    KeyValueContainer container = new KeyValueContainer(containerData,
        conf);

    populateContainerPathFields(container, maxSize);
    // 导入Container信息到Datanode metadata内，包括memory，block db文件中
    container.importContainerData(rawContainerStream, packer);
    // 发送增量Container报告信息给SCM 
    sendICR(container);
    return container;

  }

综上整个过程的过程图如下所示：

简单来概括Ozone Datanode Container Replication和HDFS的Block Replication的区别，不同点主要在以下两点：

• 前者的Replication粒度更粗，一次需要Replication的总量数据可能会非常大，因此它需要进行tar压缩的方式，提高拷贝效率。
• Ozone Datanode上保存于Container级别的元数据信息，除了正常的Container心跳汇报会，还需要额外进行本地元数据的信息导入。

转载：https://blog.csdn.net/Androidlushangderen/article/details/104086284