Hadoop-Yarn-NodeManager是如何启动容器的

一、源码下载

下面是hadoop官方源码下载地址，我下载的是hadoop-3.2.4，那就一起来看下吧

Index of /dist/hadoop/core

二、上下文

在我的博客<Hadoop-Yarn-NodeManager都做了什么>中的NodeManager服务列表中有一项是ContainerManagerImpl，字面上也能看处理（容器管理实现），容器启动的逻辑就在这里。

三、开始捋源码

1、ContainerManagerImpl

public class ContainerManagerImpl extends CompositeService implements
    ContainerManager {
  //......省略......
  public ContainerManagerImpl(Context context, ContainerExecutor exec,
    DeletionService deletionContext, NodeStatusUpdater nodeStatusUpdater,
    NodeManagerMetrics metrics, LocalDirsHandlerService dirsHandler) {
    
    //ContainerManager级调度程序
    dispatcher = new AsyncDispatcher("NM ContainerManager dispatcher");
    //创建ContainerLauncher，用于处理ContainerLauncherEvent
    containersLauncher = createContainersLauncher(context, exec);
    addService(containersLauncher);
    //......省略......
    dispatcher.register(ContainersLauncherEventType.class, containersLauncher);

  }

  protected AbstractContainersLauncher createContainersLauncher(
      Context ctxt, ContainerExecutor exec) {
    //获取 yarn.nodemanager.containers-launcher.class 的值 默认值就是 ContainersLauncher
    //官方解释：容器启动器实现，用于确定如何在NodeManager中启动容器。
    Class<? extends AbstractContainersLauncher> containersLauncherClass =
        ctxt.getConf()
            .getClass(YarnConfiguration.NM_CONTAINERS_LAUNCHER_CLASS,
                ContainersLauncher.class, AbstractContainersLauncher.class);
    AbstractContainersLauncher launcher;
    try {
      launcher = ReflectionUtils.newInstance(containersLauncherClass,
          ctxt.getConf());
      launcher.init(ctxt, this.dispatcher, exec, dirsHandler, this);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
    return launcher;
  }

  //......省略......
}

2、ContainersLauncher

源码对该类的注释：容器的启动器。只有在 ResourceLocalizationService 启动后才能启动此服务，因为它取决于在本地文件系统上创建系统目录。

public class ContainersLauncher extends AbstractService
    implements AbstractContainersLauncher {
  //实际是一个线程池（线程池可容纳Integer.MAX_VALUE（2的23次方-1）个线程）
  public ExecutorService containerLauncher =
      HadoopExecutors.newCachedThreadPool(
        new ThreadFactoryBuilder()
          .setNameFormat("ContainersLauncher #%d")
          .build());

  public ContainersLauncher() {
    super("containers-launcher");
  }

  public ContainersLauncher(Context context, Dispatcher dispatcher,
      ContainerExecutor exec, LocalDirsHandlerService dirsHandler,
      ContainerManagerImpl containerManager) {
    this();
    init(context, dispatcher, exec, dirsHandler, containerManager);
  }

  public void init(Context nmContext, Dispatcher nmDispatcher,
      ContainerExecutor containerExec, LocalDirsHandlerService nmDirsHandler,
      ContainerManagerImpl nmContainerManager) {
    this.exec = containerExec;
    this.context = nmContext;
    this.dispatcher = nmDispatcher;
    this.dirsHandler = nmDirsHandler;
    this.containerManager = nmContainerManager;
  }

public void handle(ContainersLauncherEvent event) {
    // TODO: ContainersLauncher launches containers one by one!!
    Container container = event.getContainer();
    ContainerId containerId = container.getContainerId();
    switch (event.getType()) {
      //着重看容器启动事件
      case LAUNCH_CONTAINER:
        Application app =
          context.getApplications().get(
              containerId.getApplicationAttemptId().getApplicationId());
        //Launch 动词 发射
        //Launcher 名词 发射器
        //在英文中在动词后面er，在这里会让它有这样的意思： Launcher "一个来启动容器的对象"
        ContainerLaunch launch =
            new ContainerLaunch(context, getConfig(), dispatcher, exec, app,
              event.getContainer(), dirsHandler, containerManager);
        //将容器启动任务提交给这个线程池
        containerLauncher.submit(launch);
        running.put(containerId, launch);
        break;
      //其他事件只列举下，先不分析
      case RELAUNCH_CONTAINER:

        break;
      case RECOVER_CONTAINER:

        break;
      case RECOVER_PAUSED_CONTAINER:

        break;
      case CLEANUP_CONTAINER:

        break;
      case CLEANUP_CONTAINER_FOR_REINIT:

        break;
      case SIGNAL_CONTAINER:

        break;
      case PAUSE_CONTAINER:

        break;
      case RESUME_CONTAINER:

        break;
    }
  }

}

3、ContainerLaunch

ContainerLaunch 是一个实现了 Callable接口的线程，我们直接看它的构造方法和call方法

public class ContainerLaunch implements Callable<Integer> {
  public ContainerLaunch(Context context, Configuration configuration,
      Dispatcher dispatcher, ContainerExecutor exec, Application app,
      Container container, LocalDirsHandlerService dirsHandler,
      ContainerManagerImpl containerManager) {
    this.context = context;
    this.conf = configuration;
    this.app = app;
    //这个是用于在底层操作系统上启动容器的执行器，也是我们所关心的
    this.exec = exec;
    this.container = container;
    this.dispatcher = dispatcher;
    this.dirsHandler = dirsHandler;
    this.containerManager = containerManager;
    this.maxKillWaitTime =
        conf.getLong(YarnConfiguration.NM_PROCESS_KILL_WAIT_MS,
            YarnConfiguration.DEFAULT_NM_PROCESS_KILL_WAIT_MS);
  }

  public Integer call() {
    //如果容器状态时KILLING，那么直接返回0
    if (!validateContainerState()) {
      return 0;
    }

    //ContainerLaunchContext表示NodeManager启动容器所需的所有信息
    //它包括以下详细信息：
    //    1、容器的ContainerId
    //    2、分配给容器的资源
    //    3、容器分配给的用户
    //    4、安全令牌（如果启用了安全性）
    //    5、运行容器所需的LocalResource，如二进制文件、jar、共享对象、辅助文件等
    //    6、可选的、特定于应用程序的二进制服务数据
    //    7、已启动进程的环境变量
    //    8、启动容器的命令
    //    9、容器失败退出时的重试策略
    final ContainerLaunchContext launchContext = container.getLaunchContext();
    ContainerId containerID = container.getContainerId();
    String containerIdStr = containerID.toString();
    final List<String> command = launchContext.getCommands();
    int ret = -1;

    Path containerLogDir;
    try {
      Map<Path, List<String>> localResources = getLocalizedResources();

      final String user = container.getUser();
      //在写出容器脚本之前 进行 变量扩展
      List<String> newCmds = new ArrayList<String>(command.size());
      String appIdStr = app.getAppId().toString();
      String relativeContainerLogDir = ContainerLaunch
          .getRelativeContainerLogDir(appIdStr, containerIdStr);
      containerLogDir =
          dirsHandler.getLogPathForWrite(relativeContainerLogDir, false);
      recordContainerLogDir(containerID, containerLogDir.toString());
      for (String str : command) {
        // TODO: Should we instead work via symlinks without this grammar?
        newCmds.add(expandEnvironment(str, containerLogDir));
      }
      launchContext.setCommands(newCmds);

      //环境变量的实际扩展发生在调用消毒环境之后。
      //这允许NM_ADMIN_USER_ENV中指定的变量引用用户或容器定义的变量。
      Map<String, String> environment = launchContext.getEnvironment();
      // /// 结束变量扩展

      //使用此选项可以跟踪按nm添加到环境中的变量
      LinkedHashSet<String> nmEnvVars = new LinkedHashSet<String>();

      FileContext lfs = FileContext.getLocalFSFileContext();

      Path nmPrivateContainerScriptPath = dirsHandler.getLocalPathForWrite(
          getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR
              + CONTAINER_SCRIPT);
      Path nmPrivateTokensPath = dirsHandler.getLocalPathForWrite(
          getContainerPrivateDir(appIdStr, containerIdStr) + Path.SEPARATOR
              + String.format(TOKEN_FILE_NAME_FMT, containerIdStr));
      Path nmPrivateClasspathJarDir = dirsHandler.getLocalPathForWrite(
          getContainerPrivateDir(appIdStr, containerIdStr));

      //选择容器的工作目录
      Path containerWorkDir = deriveContainerWorkDir();
      recordContainerWorkDir(containerID, containerWorkDir.toString());

      String pidFileSubpath = getPidFileSubpath(appIdStr, containerIdStr);
      //pid文件应该在nm私有目录中，这样用户就无法访问它
      pidFilePath = dirsHandler.getLocalPathForWrite(pidFileSubpath);
      List<String> localDirs = dirsHandler.getLocalDirs();
      List<String> localDirsForRead = dirsHandler.getLocalDirsForRead();
      List<String> logDirs = dirsHandler.getLogDirs();
      List<String> filecacheDirs = getNMFilecacheDirs(localDirsForRead);
      List<String> userLocalDirs = getUserLocalDirs(localDirs);
      List<String> containerLocalDirs = getContainerLocalDirs(localDirs);
      List<String> containerLogDirs = getContainerLogDirs(logDirs);
      List<String> userFilecacheDirs = getUserFilecacheDirs(localDirsForRead);
      List<String> applicationLocalDirs = getApplicationLocalDirs(localDirs,
          appIdStr);

      if (!dirsHandler.areDisksHealthy()) {
        ret = ContainerExitStatus.DISKS_FAILED;
        throw new IOException("Most of the disks failed. "
            + dirsHandler.getDisksHealthReport(false));
      }
      List<Path> appDirs = new ArrayList<Path>(localDirs.size());
      for (String localDir : localDirs) {
        Path usersdir = new Path(localDir, ContainerLocalizer.USERCACHE);
        Path userdir = new Path(usersdir, user);
        Path appsdir = new Path(userdir, ContainerLocalizer.APPCACHE);
        appDirs.add(new Path(appsdir, appIdStr));
      }

      //设置令牌位置
      addToEnvMap(environment, nmEnvVars,
          ApplicationConstants.CONTAINER_TOKEN_FILE_ENV_NAME,
          new Path(containerWorkDir,
              FINAL_CONTAINER_TOKENS_FILE).toUri().getPath());

      // /// 在nmPrivate空间中写出容器脚本。
      try (DataOutputStream containerScriptOutStream =
               lfs.create(nmPrivateContainerScriptPath,
                   EnumSet.of(CREATE, OVERWRITE))) {
        //对容器环境进行消毒
        sanitizeEnv(environment, containerWorkDir, appDirs, userLocalDirs,
            containerLogDirs, localResources, nmPrivateClasspathJarDir,
            nmEnvVars);

        //扩展所有的环境变量
        //替换参数扩展标记。
        //例如，｛｛VAR｝｝在Windows上被替换为%VAR%，而在Linux上则被替换为“$VAR”
        expandAllEnvironmentVars(environment, containerLogDir);

        //准备容器
        prepareContainer(localResources, containerLocalDirs);

        //将容器的启动环境写入默认的容器启动脚本
        exec.writeLaunchEnv(containerScriptOutStream, environment,
            localResources, launchContext.getCommands(),
            containerLogDir, user, nmEnvVars);
      }
      // /// 容器脚本编写完毕

      // /// nmPrivate空间中写出container-tokens
      try (DataOutputStream tokensOutStream =
               lfs.create(nmPrivateTokensPath, EnumSet.of(CREATE, OVERWRITE))) {
        Credentials creds = container.getCredentials();
        creds.writeTokenStorageToStream(tokensOutStream);
      }
      // /// container-tokens编写完毕

      //ContainerStartContext : 封装启动/启动容器所需的信息
      //启动容器
      ret = launchContainer(new ContainerStartContext.Builder()
          .setContainer(container)
          .setLocalizedResources(localResources)
          .setNmPrivateContainerScriptPath(nmPrivateContainerScriptPath)
          .setNmPrivateTokensPath(nmPrivateTokensPath)
          .setUser(user)
          .setAppId(appIdStr)
          .setContainerWorkDir(containerWorkDir)
          .setLocalDirs(localDirs)
          .setLogDirs(logDirs)
          .setFilecacheDirs(filecacheDirs)
          .setUserLocalDirs(userLocalDirs)
          .setContainerLocalDirs(containerLocalDirs)
          .setContainerLogDirs(containerLogDirs)
          .setUserFilecacheDirs(userFilecacheDirs)
          .setApplicationLocalDirs(applicationLocalDirs).build());
    } catch (ConfigurationException e) {
      LOG.error("Failed to launch container due to configuration error.", e);
      dispatcher.getEventHandler().handle(new ContainerExitEvent(
          containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret,
          e.getMessage()));
      // Mark the node as unhealthy
      context.getNodeStatusUpdater().reportException(e);
      return ret;
    } catch (Throwable e) {
      LOG.warn("Failed to launch container.", e);
      dispatcher.getEventHandler().handle(new ContainerExitEvent(
          containerID, ContainerEventType.CONTAINER_EXITED_WITH_FAILURE, ret,
          e.getMessage()));
      return ret;
    } finally {
      setContainerCompletedStatus(ret);
    }

    handleContainerExitCode(ret, containerLogDir);
    return ret;
  }

  protected int launchContainer(ContainerStartContext ctx)
      throws IOException, ConfigurationException {
    //为启动容器做准备
    int launchPrep = prepareForLaunch(ctx);
    if (launchPrep == 0) {
      launchLock.lock();
      try {
        //在节点上启动容器。这是一个阻塞调用，仅在容器退出时返回
        //默认的实现是 DefaultContainerExecutor 去启动容器
        //如果Yarn设置了使用CGroups，就是LinuxContainerExecutor 去启动容器
        //这里先分析 DefaultContainerExecutor 是如何启动的
        return exec.launchContainer(ctx);
      } finally {
        launchLock.unlock();
      }
    }
    return launchPrep;
  }

  protected int prepareForLaunch(ContainerStartContext ctx) throws IOException {
    ContainerId containerId = container.getContainerId();
    if (container.isMarkedForKilling()) {
      LOG.info("Container " + containerId + " not launched as it has already "
          + "been marked for Killing");
      this.killedBeforeStart = true;
      return ExitCode.TERMINATED.getExitCode();
    }
    //LaunchContainer是一个阻塞调用。我们在这里几乎意味着容器已启动，所以请发送事件。
    //ContainerEventType.CONTAINER_LAUNCHED事件会被ContainerImpl处理并对容器进行监控
    dispatcher.getEventHandler().handle(new ContainerEvent(
        containerId,
        ContainerEventType.CONTAINER_LAUNCHED));
    context.getNMStateStore().storeContainerLaunched(containerId);

    //检查容器是否有被杀死的信号
    if (!containerAlreadyLaunched.compareAndSet(false, true)) {
      LOG.info("Container " + containerId + " not launched as "
          + "cleanup already called");
      return ExitCode.TERMINATED.getExitCode();
    } else {
      //将容器标记为活动
      exec.activateContainer(containerId, pidFilePath);
    }
    return ExitCode.SUCCESS.getExitCode();
  }

}

4、DefaultContainerExecutor

DefaultContainerExecuter 类提供通用的容器执行服务。通过ProcessBuilder以独立于平台的方式处理流程执行

该类提供通用的容器执行服务，流程执行是通过 ProcessBuilder 以独立于平台的方式处理的

1、ProcessBuilder

此类用于创建操作系统进程，这意味着每个容器都是操作系统中的一个进程

每个ProcessBuilder实例都管理一组流程属性。start()方法使用这些属性创建一个新的Process实例。可以从同一实例重复调用start()方法，以创建具有相同或相关属性的新子流程。

每个ProcessBuilder都管理这些流程属性：

a command 一个命令

一个字符串列表，表示要调用的外部程序文件及其参数（如果有的话）。哪些字符串列表表示有效的操作系统命令取决于系统。例如，每个概念参数都是该列表中的一个元素是很常见的，但在某些操作系统中，程序需要将命令行字符串本身标记化——在这样的系统中，Java实现可能需要命令恰好包含两个元素。

an environment 一个环境

它是从变量到值的依赖于系统的映射。初始值是当前进程的环境的副本，请参阅System.getenv()

a working directory 工作目录

默认值是当前进程的当前工作目录，通常是由系统属性user.dir命名的目录。

a source of standard input 标准输入的来源

默认情况下，子流程从管道中读取输入。Java代码可以通过Process.getOutputStream()返回的输出流访问此管道。但是，可以使用重定向输入将标准输入重定向到另一个源。在这种情况下，Process.getOutputStream()将返回一个空输出流，其中：

    write方法总是抛出IOException

    close方法没有任何作用

a destination for standard output and standard error 标准输出和标准错误的目的地

默认情况下，子流程将标准输出和标准错误写入管道。Java代码可以通过Process.getInputStream()和Process.getErrorStream()返回的输入流访问这些管道。但是，可以使用重定向输出和重定向错误将标准输出和标准错误重定向到其他目的地。在这种情况下，Process.getInputStream()或Process.getErrorStream()将返回一个空输入流，其中：

    read方法总是返回-1

    可用方法总是返回0

    close方法没有任何作用

a redirectErrorStream property redirectErrorStream属性

默认false，这意味着子流程的标准输出和错误输出被发送到两个独立的流，这两个流可以使用Process.getInputStream()和Process.getErrorStream()方法访问。如果该值设置为true，则：

    标准错误与标准输出合并，并始终发送到相同的目的地（这使得将错误消息与相应的输出关联起来更容易）

    可以使用重定向输出重定向标准错误和标准输出的公共目的地

    创建子进程时，将忽略redirectError方法设置的任何重定向

    从Process.getErrorStream()返回的流将始终是null输入流

修改流程生成器的属性将影响该对象的start()随后启动的流程，但永远不会影响之前启动的流程或Java流程本身。

大多数错误检查都是通过start()执行的。可以修改对象的状态，使start()失败。例如，将命令属性设置为空列表不会引发异常，除非调用了start()

请注意，此类不是同步的。如果多个线程同时访问一个ProcessBuilder实例，并且至少有一个线程在结构上修改了其中一个属性，则必须对其进行外部同步。

启动使用默认工作目录和环境的新进程很容易：

Process p = new ProcessBuilder("myCommand", "myArg").start();

以下是一个示例，该示例使用修改后的工作目录和环境启动进程，并重定向标准输出和错误以附加到日志文件：

 ProcessBuilder pb =
   new ProcessBuilder("myCommand", "myArg1", "myArg2");
 Map<String, String> env = pb.environment();
 env.put("VAR1", "myValue");
 env.remove("OTHERVAR");
 env.put("VAR2", env.get("VAR1") + "suffix");
 pb.directory(new File("myDir"));
 File log = new File("log");
 pb.redirectErrorStream(true);
 pb.redirectOutput(Redirect.appendTo(log));
 Process p = pb.start();
 assert pb.redirectInput() == Redirect.PIPE;
 assert pb.redirectOutput().file() == log;
 assert p.getInputStream().read() == -1;

要使用一组明确的环境变量启动进程，请在添加环境变量之前首先调用Map.clear()

以上是补充说明，下面来具体看下DefaultContainerExecutor 是如何启动容器的

2、启动容器

public class DefaultContainerExecutor extends ContainerExecutor {

  public int launchContainer(ContainerStartContext ctx)
      throws IOException, ConfigurationException {
    Container container = ctx.getContainer();
    //启动脚本目录
    Path nmPrivateContainerScriptPath = ctx.getNmPrivateContainerScriptPath();
    //Token目录
    Path nmPrivateTokensPath = ctx.getNmPrivateTokensPath();
    String user = ctx.getUser();
    //启动容器的工作目录
    Path containerWorkDir = ctx.getContainerWorkDir();
    //容器本地目录
    List<String> localDirs = ctx.getLocalDirs();
    //日志目录
    List<String> logDirs = ctx.getLogDirs();

    FsPermission dirPerm = new FsPermission(APPDIR_PERM);
    ContainerId containerId = container.getContainerId();

    //在所有磁盘上创建容器目录
    String containerIdStr = containerId.toString();
    String appIdStr =
            containerId.getApplicationAttemptId().
                getApplicationId().toString();
    for (String sLocalDir : localDirs) {
      Path usersdir = new Path(sLocalDir, ContainerLocalizer.USERCACHE);
      Path userdir = new Path(usersdir, user);
      Path appCacheDir = new Path(userdir, ContainerLocalizer.APPCACHE);
      Path appDir = new Path(appCacheDir, appIdStr);
      Path containerDir = new Path(appDir, containerIdStr);
      createDir(containerDir, dirPerm, true, user);
    }

    //在所有磁盘上创建容器日志目录
    createContainerLogDirs(appIdStr, containerIdStr, logDirs, user);

    Path tmpDir = new Path(containerWorkDir,
        YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR);
    createDir(tmpDir, dirPerm, false, user);

    //将Token复制到工作目录
    Path tokenDst =
      new Path(containerWorkDir, ContainerLaunch.FINAL_CONTAINER_TOKENS_FILE);
    copyFile(nmPrivateTokensPath, tokenDst, user);

    //将启动脚本复制到工作目录
    Path launchDst =
        new Path(containerWorkDir, ContainerLaunch.CONTAINER_SCRIPT);
    copyFile(nmPrivateContainerScriptPath, launchDst, user);

    //创建新的本地启动包装脚本
    LocalWrapperScriptBuilder sb = getLocalWrapperScriptBuilder(
        containerIdStr, containerWorkDir); 

    //如果尝试启动包装脚本由于Windows路径长度限制而失败，则会快速失败。
    if (Shell.WINDOWS &&
        sb.getWrapperScriptPath().toString().length() > WIN_MAX_PATH) {
      throw new IOException(String.format(
        "Cannot launch container using script at path %s, because it exceeds " +
        "the maximum supported path length of %d characters.  Consider " +
        "configuring shorter directories in %s.", sb.getWrapperScriptPath(),
        WIN_MAX_PATH, YarnConfiguration.NM_LOCAL_DIRS));
    }

    //获取容器的pid文件
    //pid文件是一个简单的文本文件，它包含一个运行中的进程的PID。该文件在进程启动时被创建，在进程结束后被删除。.pid文件通常位于/var/run或/var/run/ 目录下，并以其代表的进程命名
    Path pidFile = getPidFilePath(containerId);
    if (pidFile != null) {
      sb.writeLocalWrapperScript(launchDst, pidFile);
    } else {
      LOG.info("Container " + containerIdStr
          + " pid file not set. Returning terminated error");
      return ExitCode.TERMINATED.getExitCode();
    }
    
    //在应用程序下创建日志目录
    // fork script 脚本中调用脚本
    Shell.CommandExecutor shExec = null;
    try {
      setScriptExecutable(launchDst, user);
      setScriptExecutable(sb.getWrapperScriptPath(), user);

      //使用参数创建一个新的 ShellCommandExecutor （一个简单的shell命令执行器）
      shExec = buildCommandExecutor(sb.getWrapperScriptPath().toString(),
          containerIdStr, user, pidFile, container.getResource(),
          new File(containerWorkDir.toUri().getPath()),
          container.getLaunchContext().getEnvironment());
      
      if (isContainerActive(containerId)) {
        //执行shell命令
        shExec.execute();
      } else {
        LOG.info("Container " + containerIdStr +
            " was marked as inactive. Returning terminated error");
        return ExitCode.TERMINATED.getExitCode();
      }
    } catch (IOException e) {
     //......省略......
    } finally {
      if (shExec != null) shExec.close();
    }
    return 0;
  }

  protected CommandExecutor buildCommandExecutor(String wrapperScriptPath, 
      String containerIdStr, String user, Path pidFile, Resource resource,
      File workDir, Map<String, String> environment) {
    
    //返回一个命令行以在操作系统外壳中执行给定的命令
    String[] command = getRunCommand(wrapperScriptPath,
        containerIdStr, user, pidFile, this.getConf(), resource);

      LOG.info("launchContainer: " + Arrays.toString(command));
      return new ShellCommandExecutor(
          command,
          workDir,
          environment,
          0L,
          false);
  }
}