Hadoop3.2.1 【 YARN 】源码分析 : 节点健康状况检测

向右看齐 2023-01-01 15:54 87阅读 0赞

# 一. 前言 #

节点健康状况检测是NodeManager自带的健康状况诊断机制， 通过该机制，NodeManager可时刻掌握自己的健康状况， 并及时汇报给ResourceManager。 而ResourceManager则根据每个NodeManager的健康状况适当调整分配的任务数目。 当NodeManager认为自己的健康状况“欠佳”时， 可通知ResourceManager不再为之分配新任务， 待健康状况好转时， 再分配任务。 该机制不仅可帮助及时发现存在问题的NodeManager， 避免不必要的任务分配， 也可以用于动态升级（通过脚本指示ResourceManager不再分配任务， 等到NodeManager上面的任务运行完成后， 对它进行升级） 。

# 二.NodeHealthCheckerService #

该类提供了检查节点的运行状况并向其要求运行状况检查程序报告的服务的功能。

NodeManager上有专门一个服务判断所在节点的健康状况， 该服务通过两种策略判断节点健康状况，

*  第一种是通过管理员自定义的Shell脚本（NodeManager上专门有一个周期性任务执行该脚本， 一旦该脚本输出以"ERROR"开头的字符串， 则认为节点处于不健康状态） ，
 *  另一种是判断磁盘好坏（NodeManager上专门有一个周期性任务检测磁盘的好坏， 如果坏磁盘数目达到一定的比例， 则认为节点处于不健康状态） 。

## 2.1. 属性 ##

一共就五个属性.

// 健康检查线程类 : 周期性执行节点健康状况检测脚本
      private NodeHealthScriptRunner nodeHealthScriptRunner;
    
      // 提供检查节点本地目录运行状况的功能的类。
      private LocalDirsHandlerService dirsHandler;
    
      // 异常信息
      private Exception nodeHealthException;
      // 异常事件.
      private long nodeHealthExceptionReportTime;
    
      // 分隔符
      static final String SEPARATOR = ";";

## 2.2. 构造方法 ##

在NodeManager里面进行实例化, 其中核心的scriptRunner和dirHandlerService都是入参. 提前构建好的,  
所以这个类并没有啥特殊的操作.

public NodeHealthCheckerService(NodeHealthScriptRunner scriptRunner,
          LocalDirsHandlerService dirHandlerService) { 
        super(NodeHealthCheckerService.class.getName());
        nodeHealthScriptRunner = scriptRunner;
        dirsHandler = dirHandlerService;
        nodeHealthException = null;
        nodeHealthExceptionReportTime = 0;
      }

## 2.3. serviceInit ##

将nodeHealthScriptRunner 和 dirsHandler 加入到service里面, 等待启动

@Override
      protected void serviceInit(Configuration conf) throws Exception { 
        if (nodeHealthScriptRunner != null) { 
          addService(nodeHealthScriptRunner);
        }
        addService(dirsHandler);
        super.serviceInit(conf);
      }

## 2.4. getHealthReport ##

这个类应该是唯一比较重要的了,就是获取 健康检查的报告,以分号" ; "拼接, 并返回.  
在NodeStatusUpdaterImpl类中的getNodeStatus方法中呗调用

/** * @return the reporting string of health of the node */
      String getHealthReport() { 
        // 脚本运行报告.
        String scriptReport = Strings.emptyToNull(
            nodeHealthScriptRunner == null ? null :
            nodeHealthScriptRunner.getHealthReport());
    
        // 文件目录报过.
        String discReport =
            Strings.emptyToNull(
                dirsHandler.getDisksHealthReport(false));
    
        // 异常报告.
        String exceptionReport = Strings.emptyToNull(
            nodeHealthException == null ? null :
            nodeHealthException.getMessage());
    
        // 输出报告结果... 以 分号 " ; " 分隔..
        return Joiner.on(SEPARATOR).skipNulls()
            .join(scriptReport, discReport, exceptionReport);
      }

## 2.5. isHealthy 和 getLastHealthReportTime ##

获取节点的健康状态和时间, 其实就是调用健康检查脚本和目录检查,查看是否有异常信息,并返回其异常时间…

/** * @return <em>true</em> if the node is healthy */
      boolean isHealthy() { 
        boolean scriptHealthy = nodeHealthScriptRunner == null ||
            nodeHealthScriptRunner.isHealthy();
        return nodeHealthException == null &&
            scriptHealthy && dirsHandler.areDisksHealthy();
      }
    
      /** * @return when the last time the node health status is reported */
      long getLastHealthReportTime() { 
        return Collections.max(Arrays.asList(
            dirsHandler.getLastDisksCheckTime(),
            nodeHealthScriptRunner == null ? 0 :
                nodeHealthScriptRunner.getLastReportedTime(),
            nodeHealthExceptionReportTime));
      }

# 三.NodeHealthScriptRunner #

## 3.1. 概述 ##

NodeHealthScriptRunner服务主要工作是周期性执行节点健康状况检测脚本。  
该服务允许管理员配置一个“健康监测脚本”以检查节点健康状况， 且管理员可在该脚本中添加任何检查语句作为节点是否健康运行的依据。 如果脚本检测到该节点处于不健康状态， 它需要在标准输出中打印一条以字符串"ERROR"开头的输出语句。 NodeHealthScriptRunner服务周期性调用健康监测脚本并检查其输出， 一旦发现脚本输出是以"ERROR"开头的字符串， 则认为节点处于不健康状态， 进而将其标注为"unhealthy"并通过心跳告诉ResourceManager。 而ResourceManager得知节点状态变为"unhealthy"后， 会将其加入黑名单， 此后不再为它分配新任务。 需要注意的是， 只要NodeManager服务是活着的， 该线程就会一直运行该脚本， 一旦发现节点又变为"healthy"， ResourceManager会立刻将其从黑名单中移除， 从而又可以为它分配任务。 通过引入该机制， 带来了以下几点好处：

❑**可作为节点负载的反馈。** 当前YARN仅对CPU和内存两种资源进行了隔离， 其他资源， 比如网络和磁盘IO等， 尚未有任何隔离措施， 这使得不同任务之间仍会有干扰。 而健康脚本检测的方式可从一定程度上缓解该问题， 比如， 可让健康检测脚本检查网络、 磁  
盘、 文件系统等运行状况， 一旦发现特殊情况， 比如网络拥塞、 磁盘空间不足或者文件系统出现问题， 可将健康状况为"unhealthy"， 暂时不接收新的任务， 待它们恢复正常后再继续接收新任务。

❑**人为暂时维护NodeManager。** 如果发现NodeManager所在节点出现故障， 可通过控制脚本输出暂时让该NodeManager停止接收新任务以便进行维护， 待维护完成后， 修改脚本输出以让NodeManager继续接收新任务。

## 3.2. 构建 ##

NodeHealthScriptRunner 在NodeManager中通过getNodeHealthScriptRunner(conf)方法进行构建.

NodeHealthScriptRunner服务包含四个可配置参数， 管理员可根据需要在yarn-site.xml文件中配置这些参数。

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>yarn.nodemanager.health-checker.script.path</td> 
   <td>健康检查脚本所在的绝对路径</td> 
  </tr> 
  <tr> 
   <td>yarn.nodemanager.health-checker.script.opts</td> 
   <td>健康检查脚本参数</td> 
  </tr> 
  <tr> 
   <td>yarn.nodemanager.health-checker.interval-ms</td> 
   <td>健康检查脚本检测周期 : 10min</td> 
  </tr> 
  <tr> 
   <td>yarn.nodemanager.health-checker.script.timeout-ms</td> 
   <td>健康检查脚本超时时间 : 20min</td> 
  </tr> 
 </tbody> 
</table>

构建NodeHealthScriptRunner代码 :

// 构建检查脚本对象
      // |名称| 含义 |
      // |--|--|
      // | yarn.nodemanager.health-checker.script.path | 健康检查脚本所在的绝对路径 |
      // | yarn.nodemanager.health-checker.script.opts | 健康检查脚本参数 |
      // | yarn.nodemanager.health-checker.interval-ms | 健康检查脚本检测周期 : 10min |
      // | yarn.nodemanager.health-checker.script.timeout-ms | 健康检查脚本超时时间 : 20min |
      public static NodeHealthScriptRunner getNodeHealthScriptRunner(Configuration conf) { 
    
        // 健康检查脚本所在的绝对路径
       String nodeHealthScript =  conf.get(YarnConfiguration.NM_HEALTH_CHECK_SCRIPT_PATH);
    
       // 脚本是否有效 [是否存在/可执行]
        if(!NodeHealthScriptRunner.shouldRun(nodeHealthScript)) { 
          LOG.info("Node Manager health check script is not available "
              + "or doesn't have execute permission, so not "
              + "starting the node health script runner.");
          return null;
        }
    
        // 健康检查脚本检测周期
        // yarn.nodemanager.health-checker.interval-ms : 10min
        long nmCheckintervalTime = conf.getLong(
            YarnConfiguration.NM_HEALTH_CHECK_INTERVAL_MS,
            YarnConfiguration.DEFAULT_NM_HEALTH_CHECK_INTERVAL_MS);
        //健康检查脚本超时时间
        // yarn.nodemanager.health-checker.script.timeout-ms : 20 min
        long scriptTimeout = conf.getLong(
            YarnConfiguration.NM_HEALTH_CHECK_SCRIPT_TIMEOUT_MS,
            YarnConfiguration.DEFAULT_NM_HEALTH_CHECK_SCRIPT_TIMEOUT_MS);
    
        // 健康检查脚本参数
        // yarn.nodemanager.health-checker.script.opts
        String[] scriptArgs = conf.getStrings(
            YarnConfiguration.NM_HEALTH_CHECK_SCRIPT_OPTS, new String[] { });
        return new NodeHealthScriptRunner(nodeHealthScript,
            nmCheckintervalTime, scriptTimeout, scriptArgs);
      }

构造方法就不没啥可说的了, 基本就是根据参数传递.

public NodeHealthScriptRunner(String scriptName, long chkInterval, long timeout,
          String[] scriptArgs) { 
        super(NodeHealthScriptRunner.class.getName());
        this.lastReportedTime = System.currentTimeMillis();
        this.isHealthy = true;
        this.healthReport = "";
        this.nodeHealthScript = scriptName;
        this.intervalTime = chkInterval;
        this.scriptTimeout = timeout;
        this.timer = new NodeHealthMonitorExecutor(scriptArgs);
      }

## 3.3. NodeHealthMonitorExecutor ##

在构造方法里面需要注意的是NodeHealthMonitorExecutor. 看一下这里干了啥…  
NodeHealthMonitorExecutor 继承了TimerTask . 用于周期性的执行健康检查脚本.  
这个类只有一个exceptionStackTrace属性,用于存放报错的信息.

### 3.3.1. 构造方法 ###

这里主要是对健康检查的脚本进行设置,构造ShellCommandExecutor .

public NodeHealthMonitorExecutor(String[] args) { 
          ArrayList<String> execScript = new ArrayList<String>();
          execScript.add(nodeHealthScript);
          if (args != null) { 
            execScript.addAll(Arrays.asList(args));
          }
          shexec = new ShellCommandExecutor(execScript
              .toArray(new String[execScript.size()]), null, null, scriptTimeout);
        }

### 3.3.2. run方法 ###

核心方法,用于执行健康检查的脚本.

@Override
        public void run() { 
          HealthCheckerExitStatus status = HealthCheckerExitStatus.SUCCESS;
          try { 
            // 执行健康检查脚本..
            shexec.execute();
          } catch (ExitCodeException e) { 
            // ignore the exit code of the script
            status = HealthCheckerExitStatus.FAILED_WITH_EXIT_CODE;
            // On Windows, we will not hit the Stream closed IOException
            // thrown by stdout buffered reader for timeout event.
            if (Shell.WINDOWS && shexec.isTimedOut()) { 
              status = HealthCheckerExitStatus.TIMED_OUT;
            }
          } catch (Exception e) { 
            LOG.warn("Caught exception : " + e.getMessage());
            if (!shexec.isTimedOut()) { 
              status = HealthCheckerExitStatus.FAILED_WITH_EXCEPTION;
            } else { 
              status = HealthCheckerExitStatus.TIMED_OUT;
            }
            exceptionStackTrace = StringUtils.stringifyException(e);
          } finally { 
            if (status == HealthCheckerExitStatus.SUCCESS) { 
              // 处理输出 是否已"ERROR" 开头的错误信息
              if (hasErrors(shexec.getOutput())) { 
                status = HealthCheckerExitStatus.FAILED;
              }
            }
            // 报告健康状态
            reportHealthStatus(status);
          }
        }

### 3.3.3.reportHealthStatus ###

根据健康检查的返回状态 , 处理 NodeManager的状态.

/** * Method which is used to parse output from the node health monitor and * send to the report address. * * The timed out script or script which causes IOException output is * ignored. * * The node is marked unhealthy if * <ol> * <li>The node health script times out</li> * <li>The node health scripts output has a line which begins with ERROR</li> * <li>An exception is thrown while executing the script</li> * </ol> * If the script throws {@link IOException} or {@link ExitCodeException} the * output is ignored and node is left remaining healthy, as script might * have syntax error. * * @param status */
        void reportHealthStatus(HealthCheckerExitStatus status) { 
          long now = System.currentTimeMillis();
          switch (status) { 
          case SUCCESS:
            setHealthStatus(true, "", now);
            break;
          case TIMED_OUT:
            setHealthStatus(false, NODE_HEALTH_SCRIPT_TIMED_OUT_MSG);
            break;
          case FAILED_WITH_EXCEPTION:
            setHealthStatus(false, exceptionStackTrace);
            break;
          case FAILED_WITH_EXIT_CODE:
            setHealthStatus(true, "", now);
            break;
          case FAILED:
            setHealthStatus(false, shexec.getOutput());
            break;
          }
        }

### 3.3.4. hasErrors ###

检查脚本输出是否"ERROR" 字符开头的错误信息,

> 注意: 代码写死了, 错误信息必须以: “ERROR” 开头

/** * Method to check if the output string has line which begins with ERROR. * * @param output * string * @return true if output string has error pattern in it. */
        private boolean hasErrors(String output) { 
          String[] splits = output.split("\n");
          for (String split : splits) { 
            if (split.startsWith(ERROR_PATTERN)) { 
              return true;
            }
          }
          return false;
        }

## 3.4. HealthCheckerExitStatus ##

不多说,一共就五种状态:

private enum HealthCheckerExitStatus { 
        SUCCESS,                // 成功
        TIMED_OUT,              // 超时
        FAILED_WITH_EXIT_CODE,  // 已知的错误, 有错误码
        FAILED_WITH_EXCEPTION,  // 未知错误
        FAILED                  // 失败
      }

## 3.5. 属性 ##

// 健康检查脚本绝对路径
      /** Absolute path to the health script. */
      private String nodeHealthScript;
      // 健康检查脚本执行间隔, 默认10min
      /** Delay after which node health script to be executed */
      private long intervalTime;
      
      // 健康检查脚本超时时间
      /** Time after which the script should be timedout */
      private long scriptTimeout;
      
      // 定时器
      /** Timer used to schedule node health monitoring script execution */
      private Timer nodeHealthScriptScheduler;
    
      // 执行健康检查脚本的工具类
      /** ShellCommandExecutor used to execute monitoring script */
      ShellCommandExecutor shexec = null;
    
      // 异常信息的开头定义: ERROR
      /** Pattern used for searching in the output of the node health script */
      static private final String ERROR_PATTERN = "ERROR";
    
      // 健康检查脚本超时提示信息
      /** Time out error message */
      public static final String NODE_HEALTH_SCRIPT_TIMED_OUT_MSG = "Node health script timed out";
    
      // 是否健康
      private boolean isHealthy;
    
      // 健康检查脚本报告
      private String healthReport;
    
      // 健康检查 最后一次时间戳
      private long lastReportedTime;
    
      // 定时器 task
      private TimerTask timer;

## 3.6. serviceStart 启动服务 ##

这里没啥可说的, 就是设置定时Timer以及具体的TimerTask .  
默认执行周期为10min, 超时时间为2\*10min

/** * Method used to start the Node health monitoring. * */
      @Override
      protected void serviceStart() throws Exception { 
        nodeHealthScriptScheduler = new Timer("NodeHealthMonitor-Timer", true);
        // Start the timer task immediately and
        // then periodically at interval time.
        nodeHealthScriptScheduler.scheduleAtFixedRate(timer, 0, intervalTime);
        super.serviceStart();
      }

## 3.7. 脚本示例 ##

在这个Shell脚本中， 当一个节点上的空闲内存量低于总内存量的10%时， 将打印以"ERROR"开头的字符串， 这样该节点将不再向ResourceManager请求新任务

#!/bin/bash
    MEMORY_RATIO=0.1
    freeMem=`grep MemFree /proc/meminfo | awk '{print $2}'`
    totalMem=`grep MemTotal /proc/meminfo | awk '{print $2}'`
    limitMem=`echo | awk '{print int("'$totalMem'"*"'$MEMORY_RATIO'")}'`
    if [ $freeMem -lt $limitMem ];then
    echo "ERROR, totalMem=$totalMem, freeMem=$freeMem, limitMem=$limitMem"
    else
    echo "OK, totalMem=$totalMem, freeMem=$freeMem, limitMem=$limitMem"
    fi

# 四. LocalDirsHandlerService #

除了健康状况检测机制外， YARN还提供了另外一种判断NodeManager是否健康的机制： 检测磁盘损坏数目。 管理员可通过参数yarn.nodemanager.disk-health-checker.enable设置是否启用该功能， 默认情况下是启用的。 该机制是由LocalDirsHandlerService服务实现的， 它周期性任务检测NodeManager本地磁盘的好坏， 一旦发现正常磁盘的比例低于一定的比例， 则认为节点处于不健康状态， 便通过心跳告诉ResourceManager， 从而不再接收到新的任务。

管理员配置YARN时， 会设置NodeManager的本地可用目录列表（由参数yarn.nodemanager.local-dirs设置， 通常用于存储应用程序中间结果， 比如MapReduce作业中Map Task的中间输出结果） 和日志存放目录列表（由参数yarn.nodemanager.log-dirs设置） ， 这些目录的可用性直接决定着NodeManager的可用性。 因此， NodeManager作为节点的代理和管理者， 应该负责检测这两类目录列表的可用性， 并及时将不可用目录剔除掉。 LocalDirsHandlerService服务中专门有一个定时任务周期性检测这些目录的好坏， 一旦发现正常磁盘的比例低于yarn.nodemanager.disk-health-checker.min-healthy-disks（默认是0.25） ， 就认为该节点处于“不健康”的状态， 此后ResourceManager不再为它分配新任务。  
NodeManager判断一个目录（磁盘） 好坏的方法是， 如果一个目录具有读、 写和执行权限， 则认为它是正常的， 否则将被加入坏磁盘列表， 此后不再使用。

## 4.1. 属性 ##

// 磁盘容量不足异常提示信息
      private static final String diskCapacityExceededErrorMsg =  "usable space is below configured utilization percentage/no more usable space";
    
      /** * 可用的本地目录,内部使用, 和 NM_LOCAL_DIRS同值. * Good local directories, use internally, initial value is the same as NM_LOCAL_DIRS. * yarn.nodemanager.good-local-dirs */
      @Private
      static final String NM_GOOD_LOCAL_DIRS = YarnConfiguration.NM_PREFIX + "good-local-dirs";
    
      /** * * 可用的日志目录,内部使用和 NM_LOG_DIRS 同值 * Good log directories, use internally, initial value is the same as NM_LOG_DIRS. * * yarn.nodemanager.good-log-dirs */
      @Private
      static final String NM_GOOD_LOG_DIRS =  YarnConfiguration.NM_PREFIX + "good-log-dirs";
    
      // 用于计划磁盘运行状况监视代码执行的计时器
      /** Timer used to schedule disk health monitoring code execution */
      private Timer dirsHandlerScheduler;
      // 监控周期...
      private long diskHealthCheckInterval;
    
      // 磁盘健康检查是否启用...
      private boolean isDiskHealthCheckerEnabled;
    
      /** * 最小可用磁盘的比例. * Minimum fraction of disks to be healthy for the node to be healthy in terms of disks. * This applies to nm-local-dirs and nm-log-dirs. * */
      private float minNeededHealthyDisksFactor;
    
      // TimerTask 实例
      private MonitoringTimerTask monitoringTimerTask;
    
      // 存储 本地化文件的目录
      /** Local dirs to store localized files in */
      private DirectoryCollection localDirs = null;
    
      // 存储 container 日志的目录.
      /** storage for container logs*/
      private DirectoryCollection logDirs = null;
    
      /** * 每个人都应该通过共用的 LocalDirAllocator 去 read/write YarnConfiguration#NM_LOCAL_DIRS 文件 * * Everybody should go through this LocalDirAllocator object for read/write * of any local path corresponding to {@link YarnConfiguration#NM_LOCAL_DIRS} * instead of creating his/her own LocalDirAllocator objects */ 
      private LocalDirAllocator localDirsAllocator;
    
      /** * 每个人都应该通过共用的 LocalDirAllocator 去 read/write YarnConfiguration#NM_LOG_DIRS 文件 * * Everybody should go through this LocalDirAllocator object for read/write * of any local path corresponding to {@link YarnConfiguration#NM_LOG_DIRS} * instead of creating his/her own LocalDirAllocator objects */ 
      private LocalDirAllocator logDirsAllocator;
    
      // 磁盘健康检查最后一次运行的时间
      /** when disk health checking code was last run */
      private long lastDisksCheckTime;
    
      // 文件类型
      private static String FILE_SCHEME = "file";
    
      // 度量信息
      private NodeManagerMetrics nodeManagerMetrics = null;

## 4.2. 构建 ##

提供两种构建方法,一个是包含度量信息的,一个是不包含的…

public LocalDirsHandlerService() { 
        this(null);
      }
    
      public LocalDirsHandlerService(NodeManagerMetrics nodeManagerMetrics) { 
        super(LocalDirsHandlerService.class.getName());
        this.nodeManagerMetrics = nodeManagerMetrics;
      }

## 4.3.MonitoringTimerTask ##

定时执行磁盘健康检查的TimerTask , 这里只有两个方法. 一个构造方法一个run方法

### 4.3.1. 构造方法 ###

public MonitoringTimerTask(Configuration conf) throws YarnRuntimeException { 
    
          // 磁盘被标记为脱机之后可以使用的最大磁盘空间百分比。 值的范围可以从0.0到100.0。
          // 如果该值大于或等于100，则NM将检查是否有完整的磁盘。 这适用于nm-local-dirs和nm-log-dirs。
    
          // yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage : 90.0F
          float highUsableSpacePercentagePerDisk =
              conf.getFloat(
                YarnConfiguration.NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE,
                YarnConfiguration.DEFAULT_NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE);
    
          //将脱机磁盘标记为联机时使用的磁盘空间的低阈值百分比。
          //值的范围可以从0.0到100.0。 该值不应超过NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE。
          // 如果其值大于NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE，
          // 则将其设置为与NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE相同的值。
          //这适用于nm-local-dirs和nm-log-dirs。
          // yarn.nodemanager.disk-health-checker.disk-utilization-watermark-low-per-disk-percentage : 90.0F
          float lowUsableSpacePercentagePerDisk =
              conf.getFloat(
                  YarnConfiguration.NM_WM_LOW_PER_DISK_UTILIZATION_PERCENTAGE,
                  highUsableSpacePercentagePerDisk);
    
          if (lowUsableSpacePercentagePerDisk > highUsableSpacePercentagePerDisk) { 
            LOG.warn("Using " + YarnConfiguration.
                NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE + " as " +
                YarnConfiguration.NM_WM_LOW_PER_DISK_UTILIZATION_PERCENTAGE +
                ", because " + YarnConfiguration.
                NM_WM_LOW_PER_DISK_UTILIZATION_PERCENTAGE +
                " is not configured properly.");
            lowUsableSpacePercentagePerDisk = highUsableSpacePercentagePerDisk;
          }
    
    
          // 最小空闲磁盘大小
          // yarn.nodemanager.disk-health-checker.min-free-space-per-disk-mb : 0
          long minFreeSpacePerDiskMB =
              conf.getLong(YarnConfiguration.NM_MIN_PER_DISK_FREE_SPACE_MB,
                YarnConfiguration.DEFAULT_NM_MIN_PER_DISK_FREE_SPACE_MB);
    
    
          // 文件存储目录
          localDirs =
              new DirectoryCollection(
                  // yarn.nodemanager.local-dirs : 默认路径: /tmp/nm-local-dir
                  validatePaths(conf.getTrimmedStrings(YarnConfiguration.NM_LOCAL_DIRS)),
                  highUsableSpacePercentagePerDisk,
                  lowUsableSpacePercentagePerDisk,
                  minFreeSpacePerDiskMB);
    
          // 日志存储目录
    
          logDirs = new DirectoryCollection(
                  // yarn.nodemanager.log-dirs 默认: /tmp/logs
                  validatePaths(conf .getTrimmedStrings(YarnConfiguration.NM_LOG_DIRS)),
                  highUsableSpacePercentagePerDisk,
                  lowUsableSpacePercentagePerDisk,
                  minFreeSpacePerDiskMB);
    
          // yarn.nodemanager.local-dirs
          String local = conf.get(YarnConfiguration.NM_LOCAL_DIRS);
          
          // yarn.nodemanager.good-local-dirs
          conf.set(NM_GOOD_LOCAL_DIRS, (local != null) ? local : "");
          // yarn.nodemanager.disk-validator : basic 
          String diskValidatorName = conf.get(YarnConfiguration.DISK_VALIDATOR,
                  YarnConfiguration.DEFAULT_DISK_VALIDATOR);
          try { 
            //构建 DiskValidator
            DiskValidator diskValidator = DiskValidatorFactory.getInstance(diskValidatorName);
            
            // 构建文件存储 LocalDirAllocator
            localDirsAllocator = new LocalDirAllocator( NM_GOOD_LOCAL_DIRS, diskValidator);
    
            // yarn.nodemanager.log-dirs 默认: /tmp/logs
            String log = conf.get(YarnConfiguration.NM_LOG_DIRS);
            conf.set(NM_GOOD_LOG_DIRS,  (log != null) ? log : "");
            
            // 构建日志存储 LocalDirAllocator
            logDirsAllocator = new LocalDirAllocator( NM_GOOD_LOG_DIRS, diskValidator);
          } catch (DiskErrorException e) { 
            throw new YarnRuntimeException(
                "Failed to create DiskValidator of type " + diskValidatorName + "!",
                e);
          }
        }

### 4.3.2. run方法 ###

直接调用checkDirs方法检查目录.

@Override
        public void run() { 
          try { 
            checkDirs();
          } catch (Throwable t) { 
            // Prevent uncaught exceptions from killing this thread
            LOG.warn("Error while checking local directories: ", t);
          }
        }

## 4.4. serviceInit 初始化 ##

/** * Method which initializes the timertask and its interval time. * */
      @Override
      protected void serviceInit(Configuration config) throws Exception { 
        // Clone the configuration as we may do modifications to dirs-list
        Configuration conf = new Configuration(config);
    
        // yarn.nodemanager.disk-health-checker.interval-ms : 2 * 60 * 1000 = 2min
        diskHealthCheckInterval = conf.getLong(YarnConfiguration.NM_DISK_HEALTH_CHECK_INTERVAL_MS, YarnConfiguration.DEFAULT_NM_DISK_HEALTH_CHECK_INTERVAL_MS);
    
    
        monitoringTimerTask = new MonitoringTimerTask(conf);
    
        // 启用磁盘健康检查 : 默认启用
        // yarn.nodemanager.disk-health-checker.enable : true
        isDiskHealthCheckerEnabled = conf.getBoolean(  YarnConfiguration.NM_DISK_HEALTH_CHECK_ENABLE, true);
    
        // 阈值
        // yarn.nodemanager.disk-health-checker.min-healthy-disks: 0.25f
        minNeededHealthyDisksFactor = conf.getFloat(  YarnConfiguration.NM_MIN_HEALTHY_DISKS_FRACTION, YarnConfiguration.DEFAULT_NM_MIN_HEALTHY_DISKS_FRACTION);
    
        // 更新最后一次检测时间.
        lastDisksCheckTime = System.currentTimeMillis();
        super.serviceInit(conf);
    
    
        FileContext localFs;
        try { 
          // 使用的是本地模式
          // workingDir : file:/opt/workspace/apache/hadoop-3.2.1-src/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager
          localFs = FileContext.getLocalFSFileContext(config);
    
        } catch (IOException e) { 
          throw new YarnRuntimeException("Unable to get the local filesystem", e);
        }
        // 权限
        FsPermission perm = new FsPermission((short)0755);
    
        // 设置目录
        boolean createSucceeded = localDirs.createNonExistentDirs(localFs, perm);
    
        createSucceeded &= logDirs.createNonExistentDirs(localFs, perm);
    
        if (!createSucceeded) { 
    
          updateDirsAfterTest();
        }
    
        // Check the disk health immediately to weed out bad directories
        // before other init code attempts to use them.
        checkDirs();
      }

## 4.5.checkDirs() ##

private void checkDirs() { 
        boolean disksStatusChange = false;
        Set<String> failedLocalDirsPreCheck = new HashSet<String>(localDirs.getFailedDirs());
        Set<String> failedLogDirsPreCheck =  new HashSet<String>(logDirs.getFailedDirs());
    
        if (localDirs.checkDirs()) { 
          disksStatusChange = true;
        }
        if (logDirs.checkDirs()) { 
          disksStatusChange = true;
        }
    
        Set<String> failedLocalDirsPostCheck =  new HashSet<String>(localDirs.getFailedDirs());
    
        Set<String> failedLogDirsPostCheck = new HashSet<String>(logDirs.getFailedDirs());
    
        boolean disksFailed = false;
        boolean disksTurnedGood = false;
    
        disksFailed = disksTurnedBad(failedLocalDirsPreCheck, failedLocalDirsPostCheck);
    
        disksTurnedGood = disksTurnedGood(failedLocalDirsPreCheck, failedLocalDirsPostCheck);
    
        // skip check if we have new failed or good local dirs since we're going to log anyway
    
        if (!disksFailed) { 
          disksFailed =  disksTurnedBad(failedLogDirsPreCheck, failedLogDirsPostCheck);
        }
    
        if (!disksTurnedGood) { 
          disksTurnedGood = disksTurnedGood(failedLogDirsPreCheck, failedLogDirsPostCheck);
        }
    
        logDiskStatus(disksFailed, disksTurnedGood);
    
        if (disksStatusChange) { 
          updateDirsAfterTest();
        }
    
        updateMetrics();
    
        lastDisksCheckTime = System.currentTimeMillis();
      }

## 4.6.serviceStart ##

启动定时监控程序.

/** * Method used to start the disk health monitoring, if enabled. */
      @Override
      protected void serviceStart() throws Exception { 
        if (isDiskHealthCheckerEnabled) { 
          dirsHandlerScheduler = new Timer("DiskHealthMonitor-Timer", true);
          dirsHandlerScheduler.scheduleAtFixedRate(monitoringTimerTask,
              diskHealthCheckInterval, diskHealthCheckInterval);
        }
        super.serviceStart();
      }

# 五.DiskValidator #

磁盘验证DiskValidator , 这只是一个接口,定义如下.

public interface DiskValidator { 
      /** * Check the status of a file/dir. * @param dir a file/dir * @throws DiskErrorException if any disk error */
      void checkStatus(File dir) throws DiskErrorException;
    }

我们看到这个只有一个方法checkStatus . 用于检查文件.  
我们找个实现类BasicDiskValidator看一下具体的实现方式:

public class BasicDiskValidator implements DiskValidator { 
      public static final String NAME = "basic";
    
      @Override
      public void checkStatus(File dir) throws DiskErrorException { 
        DiskChecker.checkDir(dir);
      }
    }

最终的目的: 检测一个目录具有读、 写和执行权限.

/** * Checks that the current running process can read, write, and execute the * given directory by using methods of the File object. * * @param dir File to check * @throws DiskErrorException if dir is not readable, not writable, or not * executable */
      private static void checkAccessByFileMethods(File dir)
          throws DiskErrorException { 
        if (!dir.isDirectory()) { 
          throw new DiskErrorException("Not a directory: "
              + dir.toString());
        }
    
        if (!FileUtil.canRead(dir)) { 
          throw new DiskErrorException("Directory is not readable: "
                                       + dir.toString());
        }
    
        if (!FileUtil.canWrite(dir)) { 
          throw new DiskErrorException("Directory is not writable: "
                                       + dir.toString());
        }
    
        if (!FileUtil.canExecute(dir)) { 
          throw new DiskErrorException("Directory is not executable: "
                                       + dir.toString());
        }
      }

算了,这玩意没啥可看的, 都三点了,明天还得搬砖, 就到这里了…