HDFS特性之DN选盘策略与自动摘盘(一)

(Updated: )
云存储
hdfs, 分布式, 文件系统, 磁盘

之前的写/追加写流程中, 已经讲过写失败会触发块级别的错误恢复, 那么今天接着说一下从HDFS2.X版本开始, 引入的异常磁盘自动摘除和热更换的功能, 以及这个功能使用后可能存在的一些问题, 系统的来学习一下Datanode是如何操作磁盘的

0x00. 主题

HDFS关于选DN的操作在Namenode端, 而对磁盘的操作主要是在Datanode端, 我们先来看看DN端.

大体有以下几个关键问题, 因为本文涉及的内容稍有点多, 会拆分为两篇, 先看一下后面的章节.

1.1 Datanode如何识别磁盘

HDFS而言, 它并不直接和任何块设备(磁盘卷)打交道, 不管是RAM/SSD/HDD, 都是通过手工在配置文件的dfs.datanode.data.dir 选项配置的, 如下所示:

1
2
3
4
5
<!--下面就配置让HDFS认为一个目录是SSD存储, 另两个是HDD-->
<property>
  <name>dfs.datanode.data.dir</name>
  <value>[SSD]/data01/block,[DISK]/data02/block,[DISK]/data03/block</value>
</property>

备注: 其中不同磁盘类型属于异构相关范畴, 在NN篇会详述. 可简单了解

1.2 Datanode如何选取磁盘

简言之, DN有两种磁盘选取策略, 默认的是轮询(1->2->3), 改进的是优选空盘的带权策略. (后面详述)

2.1 DN如何发现坏盘

DN会对配置的磁盘卷(volume)进行基本的检查, 包括目录存在且可以正常读写. 并且会定期检查 (待确定), 如果发现前面说的几个条件有任一不满足, 则认为此盘已坏, 自动把它”摘掉”, 但是不能读写是不是就一定是盘坏了呢? 这里之后再细说. 整个集群的坏盘信息也可以从HDFS的UI界面查看, 如下所示:

failedVolumes00

2.2 坏盘对Datanode的影响

HDFS配置里有一个dfs.datanode.failed.volumes.tolerated项, 意为当前DN能够容忍的最大坏盘数, 默认值是0也就是说挂一个盘DN就会停止工作.

2.3 更换坏盘

在HDFS2.X之前的版本中, 如果存在坏盘, 那么只能先让这台DN下线, 更换/修复磁盘之后, 再重新启动, 在2.6之后的版本支持”热替换“功能, 可直接修改#1 中的配置目录, 去掉坏盘对应的映射, 然后reconfig重新加载配置文件, 之后再换掉坏的磁盘, 把映射加回去刷新一次就行了. (方便许多)

0x01. Datanode加载磁盘

DataNode类中包括了对磁盘相关设置的调用, 不过我们得把来龙去脉先理一下, 看看这些方法是如何被调用的.

首先DN启动后, 它会进行一系列的初始化. 其中一个重要初始化是BlockPoolManager 对象, 它的功能就是管理对应的BPOfferService对象

而BPOfferService这个对象会管理DN的一个核心工作线程BPServiceActor ,它负责以下四个主要功能:

  1. 与Namenode注册前的握手通信
  2. 将当前DN汇报(注册)给Namenode (1和2是否应该合并 –> 将当前DN注册到NN上)
  3. 定时给Namenode发送心跳
  4. 处理Namenode发来的指令

其中步骤1, 2包含初始化和注册DN的过程, 是磁盘初始化相关的点, 大体流程如下所示:

sequenceDiagram
    participant A as BPServiceActor
    participant B as BPOfferService
    participant C as DataNode
    %%participant D as FSVolume
    participant E as NameNode


    C->>B: 0.初始化bpos对象
    activate B
    B->>A: 初始化Actor线程
    deactivate B

    activate A
    C->>+E: connectToNN()
    E-->>-A: 1.获得NN-Proxy

    A->>+E: .
    E-->>-A: 2.获取Namespace信息 
    B->>+C: 3.1设置Namespace信息
    C->>C: initBlockPool()
    C->>-C: 检查磁盘健康
    Note right of C: 重点: 初始化块池
    A->>+B: 5.register()
    B->>E: registerDatanode()
    B-->>-A: 注册成功
    deactivate A

然后我们梳理了整体的流程之后, 再来看看具体的几个磁盘检查点, 由于检查方式有同步和异步两个选择, 二者的实际内容完全一样, 放一起说

0x01. 磁盘检查

1. 启动DN时的检查

DN第一次初始化块池时, 以及之后每隔一段时间(存疑?), 都会使用异步的磁盘检查方法checkDiskError(), 步骤也挺清晰:

1
2
3
4
5
6
7
8
9
10
11
12
public void checkDiskError() {
  Set<FsVolumeSpi> unhealthyVolumes;
  try {
    // 1. 检查所有配置的volumes(可定时检查)
    unhealthyVolumes = volumeChecker.checkAllVolumes(data);
    lastDiskErrorCheck = Time.monotonicNow();
  } catch (InterruptedException e) {
    throw new IOException("Interrupted while running disk check", e);
  }
  // 2. 如果有volumes异常, 则进入处理步骤
  if (unhealthyVolumes.size() > 0) handleVolumeFailures(unhealthyVolumes);
}

这里说的就是几个基本的检查项是否能通过, 如果不通过则会把当前目录(磁盘)加入异常volumes处理列表, 并会将坏盘信息上报NN, 并检查坏盘数是否超过配置, 超过则停止当前DN服务, 磁盘检查核心内容如下图所示:

2. 检查内容

hdfsDiskCheck00

然后我们看一下磁盘检查实现的细节, 整个框架是个异步体系, 不关心的同学可以直接跳到最后的check()调用, 这里还是完整看看:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
  // 看看整体判断流程的
  public Set<FsVolumeSpi> checkAllVolumes(final FsDatasetSpi<? extends FsVolumeSpi> dataset){
    // 1. 默认超过15min检查一次, 少于间隔则跳过
    final long gap = timer.monotonicNow() - lastAllVolumesCheck;
    if (gap < minDiskCheckGapMs) {
      numSkippedChecks.incrementAndGet();
      return Collections.emptySet();
    }

    final FsDatasetSpi.FsVolumeReferences references = dataset.getFsVolumeReferences();
    if (references.size() == 0) return Collections.emptySet();

    lastAllVolumesCheck = timer.monotonicNow();
    final Set<FsVolumeSpi> healthyVolumes = new HashSet<>();
    final Set<FsVolumeSpi> failedVolumes = new HashSet<>();
    final Set<FsVolumeSpi> allVolumes = new HashSet<>();

    final AtomicLong numVolumes = new AtomicLong(references.size());
    final CountDownLatch latch = new CountDownLatch(1);

    // 2. 依次遍历配置的每一个volumes元素, 异步去检查
    for (int i = 0; i < references.size(); ++i) {
      final FsVolumeReference reference = references.getReference(i);
      Optional<ListenableFuture<VolumeCheckResult>> olf;
      olf = delegateChecker.schedule(reference.getVolume(), IGNORED_CONTEXT);
      //
      if (olf.isPresent()) {
        allVolumes.add(reference.getVolume());
        Futures.addCallback(olf.get(), new ResultHandler(reference, healthyVolumes,
                                                         failedVolumes, numVolumes,
                                                         (ignored1, ignored2) -> latch.countDown()));
      } else {
        IOUtils.cleanup(null, reference);
        if (numVolumes.decrementAndGet() == 0) latch.countDown();
      }
    }

    // 3. 磁盘卷检查默认最多10min, 超时则放弃后续的卷检查
    if (!latch.await(maxAllowedTimeForCheckMs, TimeUnit.MILLISECONDS)) {
      LOG.warn("checkAllVolumes timed out after xx ms");
    }
    numSyncDatasetChecks.incrementAndGet();
    
    // 4. 所有没有被标记为正常的磁盘卷, 都会在这直接被标记为异常
    // 算出异常卷的做法是取差集返回 {1, 2, 3} - {2}, 则{1,3}为异常
    synchronized (this) {
      return new HashSet<>(Sets.difference(allVolumes, healthyVolumes));
    }
  }

/////////////////////////////////分隔线//////////////////////////////////

  /* 上述的异步检查实际在ThrottledAsyncChecker的schedule()中调用check()实现
   * 基本都是异步编程的体系内容, 这里不再详述, 只说一下调用的地方
   * check主要有两个实现, 本质都是调用的DiskChecker的checkDir方法
   * 最后落到checkAccessByFileMethods()
   */
  @Override  // StorageLocation的check实现
  public VolumeCheckResult check(CheckContext context) throws IOException {
    if (storageType != StorageType.PROVIDED) { // 默认类型DISK
      // 检查配置的地址是否通过验证, 有异常则直接抛出
      // 检查点主要两个1.目录存在 2.拥有读写执行权限
      DiskChecker.checkDir(context.localFileSystem,
                           new Path(baseURI), // 比如[DISK]/data01/xx
                           context.expectedPermission);
    }
    return VolumeCheckResult.HEALTHY;
  }

上面说的磁盘检查, 基本只有DN自触发, 更多的情况, 发现磁盘问题, 其实是通过读写副本块时手动被触发的, 那么也就是接下来说的异步的磁盘检查. 先来看看有哪些情况会直接触发:

  • FileIoProvider.onFailure(volume, xx)方法: DN读写磁盘的工作类, 当读写操作失败(抛各种异常)时, 会触发此方法进行磁盘检查.
  • FsDataSetImpl.validateBlockFile(bpid, blockId)方法: 找副本块对应的文件, 如果副本块存在, 但文件不存在, 则认为磁盘有故障, 进行检查

简单说, 也就当DN端的读写磁盘发生异常时, 会触发异步的磁盘检查, 检查内容和同步检查一样, 处理方式也一样, 不再重复.

接下来我们来关注, 如果一块磁盘是健康的, 那么写数据时, DN会如何去选择.

0x02. 选盘策略

HDFS3中自带有两种检查策略:

  1. RoundRobinVolumeChoosingPolicy (默认): 轮询策略
  2. AvailableSpaceVolumeChoosingPolicy(推荐): 空盘优先策略

参考下官方blog的图, 它们总体的对比如下: (蓝色是磁盘已有数据, 黄色是一个block)

dnDiskPolicy00

大概理解意思之后, 我们再来逐一看看他们具体的实现细节, 尤其是第二个优选空盘策略

1. 轮询选择策略

这是HDFS一直以来默认的磁盘卷选择方式, 它的选法很简单, 当前盘是否还能满足写入要求, 满足就挨个盘去写: (假设仅一种存储类型, 异构单独说)

  1. 从第一个磁盘卷开始, 获取它的实际可用空间 (怎么算的, 后续单说)

  2. 当前磁盘卷可用空间 > 待写入的副本块大小 (默认也就 >128M)

    • 满足, 则选取当前磁盘卷, 结束
    • 不满足, 记录当前磁盘可用值, 然后遍历下一个磁盘卷
      1. 遍历到任一磁盘卷满足, 则结束
      2. 如果遍历完所有都不满足, 则抛出异常

  3. 之后的操作从第二个磁盘卷开始遍历, 达到轮询写入的效果 (一个10个块的文件, 两块盘, 那么每块盘就正好写5个块)

来看看代码的细节, 顺便想一下这样的选盘策略弊端是什么:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
// 空盘优先策略最后也会调用到轮询
private V chooseVolume(int curVolumeIndex, List<V> volumes,
                       long blockSize) throws IOException {
    // 因为磁盘卷可能会因异常被摘去, 这里需要确保下标不越界
    int curVolume = curVolumes[curVolumeIndex] < volumes.size() ?
                    curVolumes[curVolumeIndex] : 0;
    int startVolume = curVolume;
    long maxAvailable = 0;

    while (true) {
      // 取出对应的磁盘卷
      final V volume = volumes.get(curVolume);
      // 通过取余轮+1询选择磁盘卷
      curVolume = (curVolume + 1) % volumes.size();
      long availableVolumeSize = volume.getAvailable();
      // 可用空间 > 待写入磁盘数据大小, 则直接轮询 (不去管保留空间)
      if (availableVolumeSize > blockSize) {
        curVolumes[curVolumeIndex] = curVolume;
        return volume;
      }

      if (availableVolumeSize > maxAvailable) {
        maxAvailable = availableVolumeSize;
      }

      // 所有盘都不足待写入数据大小, 抛出异常
      if (curVolume == startVolume) {
       String err = "The volume with the most available space < the block size xx"
       throw new DiskOutOfSpaceException(err);
      }
    }
  }

2. 空盘优选策略

轮询策略的弊端很好设想, 它可能会造成短时间严重的磁盘使用比不均匀, 那么如果选择AvailableSpace策略呢? 看看下面的流程图:

  • 默认都充裕的定义是每个盘都多10G以上
  • 空盘权重占比75% (3倍于将满盘)
graph TD

a(选择考量剩余空间策略) --> b --不是--> c(划分为充裕和紧张两类) -->d(根据充裕盘优选的原则) --> e(根据实际比例算出权重) --> f(使用轮询策略从中选盘)
b -.是.-> f
subgraph Part A 
b
c
end
subgraph Part B 
d
e
end

然后再看看具体的代码细节, 以及具体的计算方式, 还是比较好理解的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
private V doChooseVolume(final List<V> volumes, long replicaSize,
                         String storageId) throws IOException {
    // 获取配置的所有磁盘卷和算出的可用空间
    var volumesWithSpaces = new AvailableSpaceVolumeList(volumes);
    
    // 判断是否所有磁盘卷的可用空间都 < 预留空间(比如配置了50G)
    if (volumesWithSpaces.areAllVolumesWithinFreeSpaceThreshold()) {
      // 所有磁盘卷都还充裕, 说明选择哪个盘差别不大, 则直接退回轮询策略
      V volume = roundRobinPolicyBalanced.chooseVolume(volumes, replicaSize, storageId);
      return volume;
    } else { // 只有磁盘剩余空间不均匀时, 才会执行当前策略
      V volume;
      // 如果所有的磁盘卷所剩空间都不足写入, 则先选其中剩余最多的一个
      long mostAvailableAmongLowVolumes = volumesWithSpaces
          .getMostAvailableSpaceAmongVolumesWithLowAvailableSpace();
      // 获取充裕 & 紧缺空间磁盘卷列表
      List<V> highAvailableVolumes = extractVolumesFromPairs(
          volumesWithSpaces.getVolumesWithHighAvailableSpace());
      List<V> lowAvailableVolumes = extractVolumesFromPairs(
          volumesWithSpaces.getVolumesWithLowAvailableSpace());
      
      /* (充裕磁盘数 * 0.75) / (充裕磁盘卷数 * 0.75 + 紧缺磁盘卷数 * 0.25)
       * 算出一个权重值, 之后根据权重来选择
       * 举几个例子:
       * A. 1块盘充裕, 1块盘不足 --> 权重: 0.75 / 1 = 75%选充裕盘
       * B. 1块盘充裕, 4块盘不足 --> 权重: 0.75 / 1.75 ≈ 43%选充裕盘
       * C. 4块盘充裕, 1块盘不足 --> 权重: 3 / 3.25 ≈ 92%选充裕盘
       */
      float preferencePercentScaler = (highAvailableVolumes.size() * balancedPreferencePercent) +
                                      (lowAvailableVolumes.size() * (1 - balancedPreferencePercent));
      float scaledPreferencePercent = (highAvailableVolumes.size() * 
                                       balancedPreferencePercent) / preferencePercentScaler;
      // 如果紧缺磁盘中剩余最多的盘都比待写入数据小 or 概率胜出
      // 则用轮询策略从充裕磁盘列表中选择一个(优先选择充裕)
      if (mostAvailableAmongLowVolumes < replicaSize ||
          random.nextFloat() < scaledPreferencePercent) {
            volume = roundRobinPolicyHighAvailable.chooseVolume(
            highAvailableVolumes, replicaSize, storageId);
         } else { // 按轮询策略从紧缺磁盘中选择一个
            volume = roundRobinPolicyLowAvailable.chooseVolume(
            lowAvailableVolumes, replicaSize, storageId);
      }
      return volume;
    }
  }

所以可以看出, 空盘优先的策略考虑更为全面, 如果磁盘都比较空闲, 它会自动退化为轮询, 如果出现倾斜(比如高负载集群我加了几块新盘), 它会优先写入空闲的新盘, 从而使磁盘使用更加均匀, 线上推荐使用此策略. 也可以减少盘满的概率, 并可与NN的优选空DN策略搭配.

0x03. 坏盘处理

1. 处理异常磁盘

紧接着上面的逻辑, 如果检查volumes后发现异常, 那么进入的处理逻辑的大流程如下3个:

  1. FsDatasetImpl先处理坏盘 (预处理?)
  2. 从DN维护的所有关系中移除异常磁盘卷
  3. 汇报坏盘信息给NN, 并判断是否需要停止此DN
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
// Datanode类的方法
private void handleVolumeFailures(Set<FsVolumeSpi> unhealthyVolumes) {
    if (unhealthyVolumes.isEmpty()) return;

    // 1. FsDatasetImpl移除坏盘对应的副本块
    data.handleVolumeFailures(unhealthyVolumes);
    Set<StorageLocation> unhealthyLocations = new HashSet();

    // 2. 记录异常的卷信息, 之后发nn
    StringBuilder sb = new StringBuilder("DataNode failed volumes:");
    for (FsVolumeSpi vol : unhealthyVolumes) {
      unhealthyLocations.add(vol.getStorageLocation());
      sb.append(vol.getStorageLocation()).append(";");
    }

    try {
      // 3. 从DN所有关系中删除异常volumes, 但不清除错误信息
      removeVolumes(unhealthyLocations, false);
    } catch (IOException e) {
     LOG.warn("Error occurred when removing unhealthy storage dirs");
    }
    // 4. 最后给NN汇报坏盘信息, 并判断是否需要停止DN
    handleDiskError(sb.toString());
  }

先来看第一步在FSVolumeList 中:

  1. 更新Map<StorageLocation, VolumeFailureInfo>对象, 记录坏盘的基本信息(位置, 时间, 预估丢失的数据大小)
  2. 调用removeVolume 从volumeList中移除此目录
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
void handleVolumeFailures(Set<FsVolumeSpi> failedVolumes) {
  try (AutoCloseableLock lock = checkDirsLock.acquire()) {
      
    // 遍历异常的磁盘卷, 记录到map中, 然后
    for(FsVolumeSpi vol : failedVolumes) {
      FsVolumeImpl fsv = (FsVolumeImpl) vol;
      // 下面两个方法调用时的异常处理略, 都只是log一下
      addVolumeFailureInfo(fsv);
      removeVolume(fsv);
    }
    waitVolumeRemoved(5000, checkDirsLockCondition);
  }
}

//
private void removeVolume(FsVolumeImpl target) {
  if (volumes.remove(target)) { // 1.从当前DN的volumesList中删除异常卷
    // 2.从block扫描器中移除当前卷
    if (blockScanner != null) blockScanner.removeVolumeScanner(target);
    try {
      // 3. 中断对应卷的所有DataXceiver线程
      target.setClosed();
    } catch (IOException e) {
      LOG.warn("Error occurs when waiting volume xx to close: ");
    }
    // 4. 调用卷对应的块池slice的shutdown() (还不是很清楚)
    target.shutdown();
    volumesBeingRemoved.add(target); // 最后移除完成添加到被删除的卷队列中
  } // 移除失败则只LOG一下, 认为可能是重复操作.
}

通过第一步的移除操作, 已经把volumeList和一些准备关闭工作完成了, 下面的操作就是进一步移除其他所有关联

  • 从FsDataset中移除volumes和副本块信息
  • 从DataStorage中移除volumes信息
  • 在内存中把dfs.datanode.data.dir配置的磁盘卷更新
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
/* 这是Datanode的移除卷方法, 他有两个参数
 * 1. storageLocations是待移除的卷URI集合(不包括其他信息)
 * 2. clearFailure参数, 标识是否从存储磁盘卷错误信息的map中剔除
 */
private synchronized void removeVolumes(params..)throws IOException {
    if (storageLocations.isEmpty()) return;

    IOException ioe = null;
    // 1. 主要是把当前磁盘卷对应的所有副本块信息删除, 并告知NN.
    data.removeVolumes(storageLocations, clearFailure);

    try {
      // 2. 主要是从DataStorage中移除相关的目录(包括块池的目录)
      storage.removeVolumes(storageLocations);
    } catch (IOException e) {
      ioe = e;
    }

    // 3. 修改DN内存中的磁盘卷配置, 移除掉异常卷, 重新赋值
    for (Iterator<StorageLocation> it = dataDirs.iterator(); it.hasNext(); ) {
      StorageLocation loc = it.next();
      if (storageLocations.contains(loc)) it.remove();
    }
    getConf().set("dfs.datanode.data.dir", Joiner.on(",").join(dataDirs));

    if (ioe != null) throw ioe;
  }

那么分两个方法来单独看看这里的两个移除具体做了什么, 先看看FsDataSetImpl.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 @Override
 public void removeVolumes(storageLocsToRemove, boolean clearFailure) {
   var storageLocationsToRemove = new ArrayList<>(storageLocsToRemove);
   Map<String, List<ReplicaInfo>> blkToInvalidate = new HashMap<>();
   List<String> storageToRemove = new ArrayList<>();
   try (AutoCloseableLock lock = datasetLock.acquire()) {
     for (int idx = 0; idx < dataStorage.getNumStorageDirs(); idx++) {
       Storage.StorageDirectory sd = dataStorage.getStorageDir(idx);
       final StorageLocation sdLocation = sd.getStorageLocation();
       // 从现有的StorageDirs找出匹配的待移除目录
       if (storageLocationsToRemove.contains(sdLocation)) {
         // 1.从DN异步操作磁盘的多线程类中停止
         asyncDiskService.removeVolume(sd.getStorageUuid());
         volumes.removeVolume(sdLocation, clearFailure);
         volumes.waitVolumeRemoved(5000, datasetLockCondition);

         // 2. 遍历块池, 移除磁盘卷对应的所有副本块信息 (操作不会去重描磁盘)
         for (String bpid : volumeMap.getBlockPoolList()) {
           List<ReplicaInfo> blocks = new ArrayList<>();
           for (var it = volumeMap.replicas(bpid).iterator(); it.hasNext();) {
             ReplicaInfo block = it.next();
             final var blockStorageLocation = block.getVolume().getStorageLocation();
             if (blockStorageLocation.equals(sdLocation)) {
               blocks.add(block); // 添加到当前块池待失效block列表
               it.remove(); // 从块池中移除
             }
           }
           blkToInvalidate.put(bpid, blocks); // 汇总所有块池待失效的副本信息
         }

         storageToRemove.add(sd.getStorageUuid());
         storageLocationsToRemove.remove(sdLocation);
       }
     } // end lock

     // 这个是"热更换"磁盘功能专用清除的
     if (clearFailure) {
       for (StorageLocation storageLocToRemove : storageLocationsToRemove) {
         volumes.removeVolumeFailureInfo(storageLocToRemove);
       }
     }
     setupAsyncLazyPersistThreads();
   }

   // 遍历上面移除的副本块信息, 让他们失效, 但不从磁盘删除文件. (为啥要把此方法特意放到锁外?)
   //  主要做两个事: 1.从DN缓存中移除卷对应的块 2.通知NN删除块
   for (Map.Entry<String, List<ReplicaInfo>> entry : blkToInvalidate.entrySet()) {
     String bpid = entry.getKey();
     List<ReplicaInfo> blocks = entry.getValue();
     for (ReplicaInfo block : blocks) {
       invalidate(bpid, block);
     }
   }
     
   // 从map<storageUuid, DatanodeStorage>中移除
   try (AutoCloseableLock lock = datasetLock.acquire()) {
     for(String storageUuid : storageToRemove) {
       storageMap.remove(storageUuid);
     }
}}

再来看看DataStorage的移除磁盘卷具体做了什么:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
synchronized void removeVolumes(storageLocations) throws IOException {
    if (storageLocations.isEmpty()) return;

    StringBuilder errorMsgBuilder = new StringBuilder();
    // 类似的遍历所有StorageDirectory对象, 然后找出包含待移除副本卷的目录?
    for (var it = getStorageDirs().iterator(); it.hasNext();) {
      StorageDirectory sd = it.next();
      StorageLocation sdLocation = sd.getStorageLocation();
      if (storageLocations.contains(sdLocation)) {
        // 先删除块池级别的存储目录 (块池这个概念待补, 还不是很理解)
        for (Map.Entry<String, BlockPoolSliceStorage> entry : bpStorageMap.entrySet()) {
          String bpid = entry.getKey();
          BlockPoolSliceStorage bpsStorage = entry.getValue();
          File bpRoot = BlockPoolSliceStorage.getBpRoot(bpid, sd.getCurrentDir());
          bpsStorage.remove(bpRoot.getAbsoluteFile());
        }
        // 再从storageDirs列表中删除对应的目录
        getStorageDirs().remove(sd);
        try {
          sd.unlock();
        } catch (IOException e) {
          errorMsgBuilder.append("Failed to remove %s", sd.getRoot());
        }
      }
    }
    if (errorMsgBuilder.length() > 0)
      throw new IOException(errorMsgBuilder.toString());
  }

至此, 整个DN启动后第一次, 和之后定时的同步磁盘卷检查流程就全部说完了, 然后在下一章再单独说说另外一种常见触发异步磁盘检查的可能.

3. 热刷新磁盘目录配置

主要的实现在DataNode类的reconfigurePropertyImpl方法中, 简单说就是不重启DN的前提下可以热加载磁盘配置项, 从而达到平滑更换磁盘的效果.

  1. refreshVolumes (热刷新)
  2. removeVolumes (摘盘, 并从map中剔除之前记录异常的磁盘卷信息, 是为了NN更新么?)
1
2
# start / status / properties三个选项
bin/hdfs dfsadmin -reconfig datanode dnIP:RPC-port start

这里待补充, 整体来说如果搞明白了前面的, 这里思路就比较清晰了 (注意这里可能有坑, 所以线上勿随意使用, 以后会补充这里的阅读测试)

0x04. 后续

1. 对磁盘的IO操作异常是否应普遍触发检查?

之前觉得问题主要在异步磁盘检查这里, 觉得是很多IO相关操作抛出的任何异常, 都会触发磁盘检查, 一旦检查读写权限不足, 就会认为这个盘坏了进入摘盘逻辑.

那么这里根本问题是不是说细化一下IO操作抛出的异常处理方式, 让某些异常不去做磁盘检查呢? 感觉这个方式也不能解决根本问题, 只能一定程度缓解, 所以目前的结论是, 磁盘IO异常, 应触发检查.

2. 磁盘卷容量已块满, 是否可以继续写入, 写入会导致失败被摘盘么?

满盘不会继续写入, 会换盘, 但是满盘不应被摘掉, 也应该正常可读, 详见下一篇的测试

小结:

这篇主要是分整体和细节介绍了HDFS_Datanode处理磁盘的逻辑, 下一篇来说一下这里面藏有的一个小坑, 以及关于触发摘盘的一系列测试, 一些遗留问题也放在下一篇解答.

参考资料:
  1. Datanode-disk-balancer(Cloudera-blog)
  2. HDFS3.1.2官方源码