ContainerExecutor implements Configurable { protected String[] getRunCommand(String command, String groupId, String userName, Path pidFile, Configuration config) { return getRunCommand(command, groupId, userName, pidFile, config, null); }  @Override void setConf(Configuration conf); @Override Configuration getConf(); abstract void init(Context nmContext); void start(); void stop(); Path localizeClasspathJar(Path jarPath, Path target, String owner); abstract void startLocalizer(LocalizerStartContext ctx); void prepareContainer(ContainerPrepareContext ctx); abstract int launchContainer(ContainerStartContext ctx); abstract int relaunchContainer(ContainerStartContext ctx); abstract boolean signalContainer(ContainerSignalContext ctx); abstract boolean reapContainer(ContainerReapContext ctx); abstract IOStreamPair execContainer(ContainerExecContext ctx); abstract void deleteAsUser(DeletionAsUserContext ctx); abstract void symLink(String target, String symlink); abstract boolean isContainerAlive(ContainerLivenessContext ctx); Map<String, LocalResource> getLocalResources(Container container); abstract void updateYarnSysFS(Context ctx, String user,
      String appId, String spec); int reacquireContainer(ContainerReacquisitionContext ctx); void writeLaunchEnv(OutputStream out, Map<String, String> environment,
      Map<Path, List<String>> resources, List<String> command, Path logDir,
      String user, LinkedHashSet<String> nmVars); @VisibleForTesting void writeLaunchEnv(OutputStream out, Map<String, String> environment,
      Map<Path, List<String>> resources, List<String> command, Path logDir,
      String user, String outFilename, LinkedHashSet<String> nmVars); void activateContainer(ContainerId containerId, Path pidFilePath); String[] getIpAndHost(Container container); static String[] getLocalIpAndHost(Container container); void deactivateContainer(ContainerId containerId); void pauseContainer(Container container); void resumeContainer(Container container); void cleanupBeforeRelaunch(Container container); String getProcessId(ContainerId containerID); String getExposedPorts(Container container); static final String TOKEN_FILE_NAME_FMT; static final FsPermission TASK_LAUNCH_SCRIPT_PERMISSION; static final String DIRECTORY_CONTENTS; }
@Test (timeout = 5000) public void testRunCommandNoPriority() throws Exception { Configuration conf = new Configuration(); String[] command = containerExecutor.getRunCommand("echo", "group1", "user", null, conf); assertTrue("first command should be the run command for the platform", command[0].equals(Shell.WINUTILS) || command[0].equals("bash")); }
@Test (timeout = 5000) public void testRunCommandwithPriority() throws Exception { Configuration conf = new Configuration(); conf.setInt(YarnConfiguration.NM_CONTAINER_EXECUTOR_SCHED_PRIORITY, 2); String[] command = containerExecutor.getRunCommand("echo", "group1", "user", null, conf); if (Shell.WINDOWS) { assertEquals("first command should be the run command for the platform", Shell.WINUTILS, command[0]); } else { assertEquals("first command should be nice", "nice", command[0]); assertEquals("second command should be -n", "-n", command[1]); assertEquals("third command should be the priority", Integer.toString(2), command[2]); } conf.setInt(YarnConfiguration.NM_CONTAINER_EXECUTOR_SCHED_PRIORITY, -5); command = containerExecutor.getRunCommand("echo", "group1", "user", null, conf); if (Shell.WINDOWS) { assertEquals("first command should be the run command for the platform", Shell.WINUTILS, command[0]); } else { assertEquals("first command should be nice", "nice", command[0]); assertEquals("second command should be -n", "-n", command[1]); assertEquals("third command should be the priority", Integer.toString(-5), command[2]); } }
@Test (timeout = 5000) public void testRunCommandWithNoResources() { assumeWindows(); Configuration conf = new Configuration(); String[] command = containerExecutor.getRunCommand("echo", "group1", null, null, conf, Resource.newInstance(1024, 1)); String[] expected = { Shell.WINUTILS, "task", "create", "-m", "-1", "-c", "-1", "group1", "cmd /c " + "echo" }; Assert.assertTrue(Arrays.equals(expected, command)); }
@Test (timeout = 5000) public void testRunCommandWithMemoryOnlyResources() { assumeWindows(); Configuration conf = new Configuration(); conf.set(YarnConfiguration.NM_WINDOWS_CONTAINER_MEMORY_LIMIT_ENABLED, "true"); String[] command = containerExecutor.getRunCommand("echo", "group1", null, null, conf, Resource.newInstance(1024, 1)); String[] expected = { Shell.WINUTILS, "task", "create", "-m", "1024", "-c", "-1", "group1", "cmd /c " + "echo" }; Assert.assertTrue(Arrays.equals(expected, command)); }
@Test (timeout = 5000) public void testRunCommandWithCpuAndMemoryResources() { assumeWindows(); int containerCores = 1; Configuration conf = new Configuration(); conf.set(YarnConfiguration.NM_WINDOWS_CONTAINER_CPU_LIMIT_ENABLED, "true"); conf.set(YarnConfiguration.NM_WINDOWS_CONTAINER_MEMORY_LIMIT_ENABLED, "true"); String[] command = containerExecutor.getRunCommand("echo", "group1", null, null, conf, Resource.newInstance(1024, 1)); int nodeVCores = NodeManagerHardwareUtils.getVCores(conf); Assert.assertEquals(YarnConfiguration.DEFAULT_NM_VCORES, nodeVCores); int cpuRate = Math.min(10000, (containerCores * 10000) / nodeVCores); String[] expected = {Shell.WINUTILS, "task", "create", "-m", "1024", "-c", String.valueOf(cpuRate), "group1", "cmd /c " + "echo" }; Assert.assertEquals(Arrays.toString(expected), Arrays.toString(command)); conf.setBoolean(YarnConfiguration.NM_ENABLE_HARDWARE_CAPABILITY_DETECTION, true); int nodeCPUs = NodeManagerHardwareUtils.getNodeCPUs(conf); float yarnCPUs = NodeManagerHardwareUtils.getContainersCPUs(conf); nodeVCores = NodeManagerHardwareUtils.getVCores(conf); Assert.assertEquals(nodeCPUs, (int) yarnCPUs); Assert.assertEquals(nodeCPUs, nodeVCores); command = containerExecutor.getRunCommand("echo", "group1", null, null, conf, Resource.newInstance(1024, 1)); cpuRate = Math.min(10000, (containerCores * 10000) / nodeVCores); expected[6] = String.valueOf(cpuRate); Assert.assertEquals(Arrays.toString(expected), Arrays.toString(command)); int yarnCpuLimit = 80; conf.setInt(YarnConfiguration.NM_RESOURCE_PERCENTAGE_PHYSICAL_CPU_LIMIT, yarnCpuLimit); yarnCPUs = NodeManagerHardwareUtils.getContainersCPUs(conf); nodeVCores = NodeManagerHardwareUtils.getVCores(conf); Assert.assertEquals(nodeCPUs * 0.8, yarnCPUs, 0.01); if (nodeCPUs == 1) { Assert.assertEquals(1, nodeVCores); } else { Assert.assertEquals((int) (nodeCPUs * 0.8), nodeVCores); } command = containerExecutor.getRunCommand("echo", "group1", null, null, conf, Resource.newInstance(1024, 1)); int containerPerc = (yarnCpuLimit * containerCores) / nodeVCores; cpuRate = Math.min(10000, 100 * containerPerc); expected[6] = String.valueOf(cpuRate); Assert.assertEquals(Arrays.toString(expected), Arrays.toString(command)); }