DefaultContainerExecutor.java
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.yarn.server.nodemanager;
import static org.apache.hadoop.fs.CreateFlag.CREATE;
import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
import static org.apache.hadoop.yarn.conf.YarnConfiguration.numaAwarenessEnabled;
import org.apache.hadoop.classification.VisibleForTesting;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.PrintStream;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnsupportedFileSystemException;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.protocol.datatransfer.IOStreamPair;
import org.apache.hadoop.service.ServiceStateException;
import org.apache.hadoop.util.Shell;
import org.apache.hadoop.util.Shell.CommandExecutor;
import org.apache.hadoop.util.Shell.ExitCodeException;
import org.apache.hadoop.util.Shell.ShellCommandExecutor;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.ConfigurationException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerDiagnosticsUpdateEvent;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.ResourceHandlerException;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocation;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocator;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerExecutionException;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerExecContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerLivenessContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReacquisitionContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerReapContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerSignalContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerStartContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.DeletionAsUserContext;
import org.apache.hadoop.yarn.server.nodemanager.executor.LocalizerStartContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* The {@code DefaultContainerExecuter} class offers generic container
* execution services. Process execution is handled in a platform-independent
* way via {@link ProcessBuilder}.
*/
public class DefaultContainerExecutor extends ContainerExecutor {
private static final Logger LOG =
LoggerFactory.getLogger(DefaultContainerExecutor.class);
private static final int WIN_MAX_PATH = 260;
/**
* A {@link FileContext} for the local file system.
*/
protected final FileContext lfs;
private String logDirPermissions = null;
private NumaResourceAllocator numaResourceAllocator;
private String numactl;
/**
* Default constructor for use in testing.
*/
@VisibleForTesting
public DefaultContainerExecutor() {
try {
this.lfs = FileContext.getLocalFSFileContext();
} catch (UnsupportedFileSystemException e) {
throw new RuntimeException(e);
}
}
/**
* Create an instance with a given {@link FileContext}.
*
* @param lfs the given {@link FileContext}
*/
DefaultContainerExecutor(FileContext lfs) {
this.lfs = lfs;
}
/**
* Copy a file using the {@link #lfs} {@link FileContext}.
*
* @param src the file to copy
* @param dst where to copy the file
* @param owner the owner of the new copy. Used only in secure Windows
* clusters
* @throws IOException when the copy fails
* @see WindowsSecureContainerExecutor
*/
protected void copyFile(Path src, Path dst, String owner) throws IOException {
lfs.util().copy(src, dst, false, true);
}
/**
* Make a file executable using the {@link #lfs} {@link FileContext}.
*
* @param script the path to make executable
* @param owner the new owner for the file. Used only in secure Windows
* clusters
* @throws IOException when the change mode operation fails
* @see WindowsSecureContainerExecutor
*/
protected void setScriptExecutable(Path script, String owner)
throws IOException {
lfs.setPermission(script, ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION);
}
@Override
public void init(Context nmContext) throws IOException {
if(numaAwarenessEnabled(getConf())) {
numaResourceAllocator = new NumaResourceAllocator(nmContext);
numactl = this.getConf().get(YarnConfiguration.NM_NUMA_AWARENESS_NUMACTL_CMD,
YarnConfiguration.DEFAULT_NM_NUMA_AWARENESS_NUMACTL_CMD);
try {
numaResourceAllocator.init(this.getConf());
LOG.info("NUMA resources allocation is enabled in DefaultContainer Executor," +
" Successfully initialized NUMA resources allocator.");
} catch (YarnException e) {
LOG.warn("Improper NUMA configuration provided.", e);
throw new IOException("Failed to initialize configured numa subsystem!");
}
}
}
@Override
public void startLocalizer(LocalizerStartContext ctx)
throws IOException, InterruptedException {
Path nmPrivateContainerTokensPath = ctx.getNmPrivateContainerTokens();
InetSocketAddress nmAddr = ctx.getNmAddr();
String user = ctx.getUser();
String appId = ctx.getAppId();
String locId = ctx.getLocId();
LocalDirsHandlerService dirsHandler = ctx.getDirsHandler();
List<String> localDirs = dirsHandler.getLocalDirs();
List<String> logDirs = dirsHandler.getLogDirs();
createUserLocalDirs(localDirs, user);
createUserCacheDirs(localDirs, user);
createAppDirs(localDirs, user, appId);
createAppLogDirs(appId, logDirs, user);
// randomly choose the local directory
Path appStorageDir = getWorkingDir(localDirs, user, appId);
String tokenFn = String.format(TOKEN_FILE_NAME_FMT, locId);
Path tokenDst = new Path(appStorageDir, tokenFn);
copyFile(nmPrivateContainerTokensPath, tokenDst, user);
LOG.info("Copying from {} to {}", nmPrivateContainerTokensPath, tokenDst);
FileContext localizerFc =
FileContext.getFileContext(lfs.getDefaultFileSystem(), getConf());
localizerFc.setUMask(lfs.getUMask());
localizerFc.setWorkingDirectory(appStorageDir);
LOG.info("Localizer CWD set to {} = {}", appStorageDir,
localizerFc.getWorkingDirectory());
ContainerLocalizer localizer =
createContainerLocalizer(user, appId, locId, tokenFn, localDirs,
localizerFc);
// TODO: DO it over RPC for maintaining similarity?
localizer.runLocalization(nmAddr);
}
/**
* Create a new {@link ContainerLocalizer} instance.
*
* @param user the user who owns the job for which the localization is being
* run
* @param appId the ID of the application for which the localization is being
* run
* @param locId the ID of the container for which the localization is being
* run
* @param localDirs a list of directories to use as destinations for the
* localization
* @param localizerFc the {@link FileContext} to use when localizing files
* @return the new {@link ContainerLocalizer} instance
* @throws IOException if {@code user} or {@code locId} is {@code null} or if
* the container localizer has an initialization failure
*/
@Private
@VisibleForTesting
protected ContainerLocalizer createContainerLocalizer(String user,
String appId, String locId, String tokenFileName, List<String> localDirs,
FileContext localizerFc) throws IOException {
ContainerLocalizer localizer =
new ContainerLocalizer(localizerFc, user, appId, locId, tokenFileName,
getPaths(localDirs),
RecordFactoryProvider.getRecordFactory(getConf()));
return localizer;
}
@Override
public int launchContainer(ContainerStartContext ctx)
throws IOException, ConfigurationException {
Container container = ctx.getContainer();
Path nmPrivateContainerScriptPath = ctx.getNmPrivateContainerScriptPath();
Path nmPrivateTokensPath = ctx.getNmPrivateTokensPath();
Path nmPrivateKeystorePath = ctx.getNmPrivateKeystorePath();
Path nmPrivateTruststorePath = ctx.getNmPrivateTruststorePath();
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();
// create container dirs on all disks
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);
}
// Create the container log-dirs on all disks
createContainerLogDirs(appIdStr, containerIdStr, logDirs, user);
Path tmpDir = new Path(containerWorkDir,
YarnConfiguration.DEFAULT_CONTAINER_TEMP_DIR);
createDir(tmpDir, dirPerm, false, user);
// copy container tokens to work dir
Path tokenDst =
new Path(containerWorkDir, ContainerLaunch.FINAL_CONTAINER_TOKENS_FILE);
copyFile(nmPrivateTokensPath, tokenDst, user);
if (nmPrivateKeystorePath != null) {
Path keystoreDst =
new Path(containerWorkDir, ContainerLaunch.KEYSTORE_FILE);
copyFile(nmPrivateKeystorePath, keystoreDst, user);
}
if (nmPrivateTruststorePath != null) {
Path truststoreDst =
new Path(containerWorkDir, ContainerLaunch.TRUSTSTORE_FILE);
copyFile(nmPrivateTruststorePath, truststoreDst, user);
}
// copy launch script to work dir
Path launchDst =
new Path(containerWorkDir, ContainerLaunch.CONTAINER_SCRIPT);
copyFile(nmPrivateContainerScriptPath, launchDst, user);
// Create new local launch wrapper script
LocalWrapperScriptBuilder sb = getLocalWrapperScriptBuilder(
containerIdStr, containerWorkDir);
// Fail fast if attempting to launch the wrapper script would fail due to
// Windows path length limitation.
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));
}
Path pidFile = getPidFilePath(containerId);
if (pidFile != null) {
sb.writeLocalWrapperScript(launchDst, pidFile);
} else {
LOG.info("Container {} pid file not set. Returning terminated error",
containerIdStr);
return ExitCode.TERMINATED.getExitCode();
}
// create log dir under app
// fork script
Shell.CommandExecutor shExec = null;
try {
setScriptExecutable(launchDst, user);
setScriptExecutable(sb.getWrapperScriptPath(), user);
// adding numa commands based on configuration
String[] numaCommands = new String[]{};
if (numaResourceAllocator != null) {
try {
NumaResourceAllocation numaResourceAllocation =
numaResourceAllocator.allocateNumaNodes(container);
if (numaResourceAllocation != null) {
numaCommands = getNumaCommands(numaResourceAllocation);
}
} catch (ResourceHandlerException e) {
LOG.error("NumaResource Allocation failed!", e);
throw new IOException("NumaResource Allocation Error!", e);
}
}
shExec = buildCommandExecutor(sb.getWrapperScriptPath().toString(),
containerIdStr, user, pidFile, container.getResource(),
new File(containerWorkDir.toUri().getPath()),
container.getLaunchContext().getEnvironment(),
numaCommands);
if (isContainerActive(containerId)) {
shExec.execute();
} else {
LOG.info("Container {} was marked as inactive. "
+ "Returning terminated error", containerIdStr);
return ExitCode.TERMINATED.getExitCode();
}
} catch (IOException e) {
if (null == shExec) {
return -1;
}
int exitCode = shExec.getExitCode();
LOG.warn("Exit code from container {} is : {}", containerId, exitCode);
// 143 (SIGTERM) and 137 (SIGKILL) exit codes means the container was
// terminated/killed forcefully. In all other cases, log the
// container-executor's output
if (exitCode != ExitCode.FORCE_KILLED.getExitCode()
&& exitCode != ExitCode.TERMINATED.getExitCode()) {
LOG.warn("Exception from container-launch with container ID: {}"
+ " and exit code: {}", containerId, exitCode, e);
StringBuilder builder = new StringBuilder();
builder.append("Exception from container-launch.\n")
.append("Container id: ").append(containerId).append("\n")
.append("Exit code: ").append(exitCode).append("\n");
if (!Optional.ofNullable(e.getMessage()).orElse("").isEmpty()) {
builder.append("Exception message: ")
.append(e.getMessage()).append("\n");
}
if (!shExec.getOutput().isEmpty()) {
builder.append("Shell output: ")
.append(shExec.getOutput()).append("\n");
}
String diagnostics = builder.toString();
logOutput(diagnostics);
container.handle(new ContainerDiagnosticsUpdateEvent(containerId,
diagnostics));
} else {
container.handle(new ContainerDiagnosticsUpdateEvent(containerId,
"Container killed on request. Exit code is " + exitCode));
}
return exitCode;
} finally {
if (shExec != null) shExec.close();
postComplete(containerId);
}
return 0;
}
@Override
public int relaunchContainer(ContainerStartContext ctx)
throws IOException, ConfigurationException {
return launchContainer(ctx);
}
/**
* Create a new {@link ShellCommandExecutor} using the parameters.
*
* @param wrapperScriptPath the path to the script to execute
* @param containerIdStr the container ID
* @param user the application owner's username
* @param pidFile the path to the container's PID file
* @param resource this parameter controls memory and CPU limits.
* @param workDir If not-null, specifies the directory which should be set
* as the current working directory for the command. If null,
* the current working directory is not modified.
* @param environment the container environment
* @param numaCommands list of prefix numa commands
* @return the new {@link ShellCommandExecutor}
* @see ShellCommandExecutor
*/
protected CommandExecutor buildCommandExecutor(String wrapperScriptPath,
String containerIdStr, String user, Path pidFile, Resource resource,
File workDir, Map<String, String> environment, String[] numaCommands) {
String[] command = getRunCommand(wrapperScriptPath,
containerIdStr, user, pidFile, this.getConf(), resource);
// check if numa commands are passed and append it as prefix commands
if(numaCommands != null && numaCommands.length!=0) {
command = concatStringCommands(numaCommands, command);
}
LOG.info("launchContainer: {}", Arrays.toString(command));
return new ShellCommandExecutor(
command,
workDir,
environment,
0L,
false);
}
/**
* Create a {@link LocalWrapperScriptBuilder} for the given container ID
* and path that is appropriate to the current platform.
*
* @param containerIdStr the container ID
* @param containerWorkDir the container's working directory
* @return a new {@link LocalWrapperScriptBuilder}
*/
protected LocalWrapperScriptBuilder getLocalWrapperScriptBuilder(
String containerIdStr, Path containerWorkDir) {
return Shell.WINDOWS ?
new WindowsLocalWrapperScriptBuilder(containerIdStr, containerWorkDir) :
new UnixLocalWrapperScriptBuilder(containerWorkDir);
}
/**
* This class is a utility to create a wrapper script that is platform
* appropriate.
*/
protected abstract class LocalWrapperScriptBuilder {
private final Path wrapperScriptPath;
/**
* Return the path for the wrapper script.
*
* @return the path for the wrapper script
*/
public Path getWrapperScriptPath() {
return wrapperScriptPath;
}
/**
* Write out the wrapper script for the container launch script. This method
* will create the script at the configured wrapper script path.
*
* @param launchDst the script to launch
* @param pidFile the file that will hold the PID
* @throws IOException if the wrapper script cannot be created
* @see #getWrapperScriptPath
*/
public void writeLocalWrapperScript(Path launchDst, Path pidFile)
throws IOException {
try (DataOutputStream out =
lfs.create(wrapperScriptPath, EnumSet.of(CREATE, OVERWRITE));
PrintStream pout =
new PrintStream(out, false, "UTF-8")) {
writeLocalWrapperScript(launchDst, pidFile, pout);
}
}
/**
* Write out the wrapper script for the container launch script.
*
* @param launchDst the script to launch
* @param pidFile the file that will hold the PID
* @param pout the stream to use to write out the wrapper script
*/
protected abstract void writeLocalWrapperScript(Path launchDst,
Path pidFile, PrintStream pout);
/**
* Create an instance for the given container working directory.
*
* @param containerWorkDir the working directory for the container
*/
protected LocalWrapperScriptBuilder(Path containerWorkDir) {
this.wrapperScriptPath = new Path(containerWorkDir,
Shell.appendScriptExtension("default_container_executor"));
}
}
/**
* This class is an instance of {@link LocalWrapperScriptBuilder} for
* non-Windows hosts.
*/
private final class UnixLocalWrapperScriptBuilder
extends LocalWrapperScriptBuilder {
private final Path sessionScriptPath;
/**
* Create an instance for the given container path.
*
* @param containerWorkDir the container's working directory
*/
public UnixLocalWrapperScriptBuilder(Path containerWorkDir) {
super(containerWorkDir);
this.sessionScriptPath = new Path(containerWorkDir,
Shell.appendScriptExtension("default_container_executor_session"));
}
@Override
public void writeLocalWrapperScript(Path launchDst, Path pidFile)
throws IOException {
writeSessionScript(launchDst, pidFile);
super.writeLocalWrapperScript(launchDst, pidFile);
}
@Override
public void writeLocalWrapperScript(Path launchDst, Path pidFile,
PrintStream pout) {
String exitCodeFile = ContainerLaunch.getExitCodeFile(
pidFile.toString());
String tmpFile = exitCodeFile + ".tmp";
pout.println("#!/bin/bash");
pout.println("/bin/bash \"" + sessionScriptPath.toString() + "\"");
pout.println("rc=$?");
pout.println("echo $rc > \"" + tmpFile + "\"");
pout.println("/bin/mv -f \"" + tmpFile + "\" \"" + exitCodeFile + "\"");
pout.println("exit $rc");
}
private void writeSessionScript(Path launchDst, Path pidFile)
throws IOException {
try (DataOutputStream out =
lfs.create(sessionScriptPath, EnumSet.of(CREATE, OVERWRITE));
PrintStream pout =
new PrintStream(out, false, "UTF-8")) {
// We need to do a move as writing to a file is not atomic
// Process reading a file being written to may get garbled data
// hence write pid to tmp file first followed by a mv
pout.println("#!/bin/bash");
pout.println();
pout.println("echo $$ > " + pidFile.toString() + ".tmp");
pout.println("/bin/mv -f " + pidFile.toString() + ".tmp " + pidFile);
String exec = Shell.isSetsidAvailable? "exec setsid" : "exec";
pout.printf("%s /bin/bash \"%s\"", exec, launchDst.toUri().getPath());
}
lfs.setPermission(sessionScriptPath,
ContainerExecutor.TASK_LAUNCH_SCRIPT_PERMISSION);
}
}
/**
* This class is an instance of {@link LocalWrapperScriptBuilder} for
* Windows hosts.
*/
private final class WindowsLocalWrapperScriptBuilder
extends LocalWrapperScriptBuilder {
private final String containerIdStr;
/**
* Create an instance for the given container and working directory.
*
* @param containerIdStr the container ID
* @param containerWorkDir the container's working directory
*/
public WindowsLocalWrapperScriptBuilder(String containerIdStr,
Path containerWorkDir) {
super(containerWorkDir);
this.containerIdStr = containerIdStr;
}
@Override
public void writeLocalWrapperScript(Path launchDst, Path pidFile,
PrintStream pout) {
// TODO: exit code script for Windows
// On Windows, the pid is the container ID, so that it can also serve as
// the name of the job object created by winutils for task management.
// Write to temp file followed by atomic move.
String normalizedPidFile = new File(pidFile.toString()).getPath();
pout.println("@echo " + containerIdStr + " > " + normalizedPidFile +
".tmp");
pout.println("@move /Y " + normalizedPidFile + ".tmp " +
normalizedPidFile);
pout.println("@call " + launchDst.toString());
}
}
@Override
public boolean signalContainer(ContainerSignalContext ctx)
throws IOException {
String user = ctx.getUser();
String pid = ctx.getPid();
Signal signal = ctx.getSignal();
LOG.debug("Sending signal {} to pid {} as user {}",
signal.getValue(), pid, user);
if (!containerIsAlive(pid)) {
return false;
}
try {
killContainer(pid, signal);
} catch (IOException e) {
if (!containerIsAlive(pid)) {
return false;
}
throw e;
}
return true;
}
/**
* No-op for reaping containers within the DefaultContainerExecutor.
*
* @param ctx Encapsulates information necessary for reaping containers.
* @return true given no operations are needed.
*/
@Override
public boolean reapContainer(ContainerReapContext ctx) {
return true;
}
@Override
public boolean isContainerAlive(ContainerLivenessContext ctx)
throws IOException {
String pid = ctx.getPid();
return containerIsAlive(pid);
}
/**
* Returns true if the process with the specified pid is alive.
*
* @param pid String pid
* @return boolean true if the process is alive
* @throws IOException if the command to test process liveliness fails
*/
@VisibleForTesting
public static boolean containerIsAlive(String pid) throws IOException {
try {
new ShellCommandExecutor(Shell.getCheckProcessIsAliveCommand(pid))
.execute();
// successful execution means process is alive
return true;
}
catch (ExitCodeException e) {
// failure (non-zero exit code) means process is not alive
return false;
}
}
/**
* Send a specified signal to the specified pid
*
* @param pid the pid of the process [group] to signal.
* @param signal signal to send
* @throws IOException if the command to kill the process fails
*/
protected void killContainer(String pid, Signal signal) throws IOException {
new ShellCommandExecutor(Shell.getSignalKillCommand(signal.getValue(), pid))
.execute();
}
@Override
public void deleteAsUser(DeletionAsUserContext ctx)
throws IOException, InterruptedException {
Path subDir = ctx.getSubDir();
List<Path> baseDirs = ctx.getBasedirs();
if (baseDirs == null || baseDirs.size() == 0) {
LOG.info("Deleting absolute path : {}", subDir);
if (!lfs.delete(subDir, true)) {
//Maybe retry
LOG.warn("delete returned false for path: [{}]", subDir);
}
return;
}
for (Path baseDir : baseDirs) {
Path del = subDir == null ? baseDir : new Path(baseDir, subDir);
LOG.info("Deleting path : {}", del);
try {
if (!lfs.delete(del, true)) {
LOG.warn("delete returned false for path: [{}]", del);
}
} catch (FileNotFoundException e) {
continue;
}
}
}
@Override
public void symLink(String target, String symlink) throws IOException {
FileUtil.symLink(target, symlink);
}
/**
* Permissions for user dir.
* $local.dir/usercache/$user
*/
static final short USER_PERM = (short)0750;
/**
* Permissions for user appcache dir.
* $local.dir/usercache/$user/appcache
*/
static final short APPCACHE_PERM = (short)0710;
/**
* Permissions for user filecache dir.
* $local.dir/usercache/$user/filecache
*/
static final short FILECACHE_PERM = (short)0710;
/**
* Permissions for user app dir.
* $local.dir/usercache/$user/appcache/$appId
*/
static final short APPDIR_PERM = (short)0710;
private long getDiskFreeSpace(Path base) throws IOException {
return lfs.getFsStatus(base).getRemaining();
}
private Path getApplicationDir(Path base, String user, String appId) {
return new Path(getAppcacheDir(base, user), appId);
}
private Path getUserCacheDir(Path base, String user) {
return new Path(new Path(base, ContainerLocalizer.USERCACHE), user);
}
private Path getAppcacheDir(Path base, String user) {
return new Path(getUserCacheDir(base, user),
ContainerLocalizer.APPCACHE);
}
private Path getFileCacheDir(Path base, String user) {
return new Path(getUserCacheDir(base, user),
ContainerLocalizer.FILECACHE);
}
/**
* Return a randomly chosen application directory from a list of local storage
* directories. The probability of selecting a directory is proportional to
* its size.
*
* @param localDirs the target directories from which to select
* @param user the user who owns the application
* @param appId the application ID
* @return the selected directory
* @throws IOException if no application directories for the user can be
* found
*/
protected Path getWorkingDir(List<String> localDirs, String user,
String appId) throws IOException {
long totalAvailable = 0L;
long[] availableOnDisk = new long[localDirs.size()];
int i = 0;
// randomly choose the app directory
// the chance of picking a directory is proportional to
// the available space on the directory.
// firstly calculate the sum of all available space on these directories
for (String localDir : localDirs) {
Path curBase = getApplicationDir(new Path(localDir), user, appId);
long space = 0L;
try {
space = getDiskFreeSpace(curBase);
} catch (IOException e) {
LOG.warn("Unable to get Free Space for {}", curBase, e);
}
availableOnDisk[i++] = space;
totalAvailable += space;
}
// throw an IOException if totalAvailable is 0.
if (totalAvailable <= 0L) {
throw new IOException("Not able to find a working directory for " + user);
}
// make probability to pick a directory proportional to
// the available space on the directory.
long randomPosition = RandomUtils.nextLong() % totalAvailable;
int dir = pickDirectory(randomPosition, availableOnDisk);
return getApplicationDir(new Path(localDirs.get(dir)), user, appId);
}
/**
* Picks a directory based on the input random number and
* available size at each dir.
*/
@Private
@VisibleForTesting
int pickDirectory(long randomPosition, final long[] availableOnDisk) {
int dir = 0;
// skip zero available space directory,
// because totalAvailable is greater than 0 and randomPosition
// is less than totalAvailable, we can find a valid directory
// with nonzero available space.
while (availableOnDisk[dir] == 0L) {
dir++;
}
while (randomPosition >= availableOnDisk[dir]) {
randomPosition -= availableOnDisk[dir++];
}
return dir;
}
/**
* Use the {@link #lfs} {@link FileContext} to create the target directory.
*
* @param dirPath the target directory
* @param perms the target permissions for the target directory
* @param createParent whether the parent directories should also be created
* @param user the user as whom the target directory should be created.
* Used only on secure Windows hosts.
* @throws IOException if there's a failure performing a file operation
* @see WindowsSecureContainerExecutor
*/
protected void createDir(Path dirPath, FsPermission perms,
boolean createParent, String user) throws IOException {
lfs.mkdir(dirPath, perms, createParent);
if (!perms.equals(perms.applyUMask(lfs.getUMask()))) {
lfs.setPermission(dirPath, perms);
}
}
/**
* Initialize the local directories for a particular user.
* <ul>.mkdir
* <li>$local.dir/usercache/$user</li>
* </ul>
*
* @param localDirs the target directories to create
* @param user the user whose local cache directories should be initialized
* @throws IOException if there's an issue initializing the user local
* directories
*/
void createUserLocalDirs(List<String> localDirs, String user)
throws IOException {
boolean userDirStatus = false;
FsPermission userperms = new FsPermission(USER_PERM);
for (String localDir : localDirs) {
// create $local.dir/usercache/$user and its immediate parent
try {
createDir(getUserCacheDir(new Path(localDir), user), userperms, true,
user);
} catch (IOException e) {
LOG.warn("Unable to create the user directory : {}", localDir, e);
continue;
}
userDirStatus = true;
}
if (!userDirStatus) {
throw new IOException("Not able to initialize user directories "
+ "in any of the configured local directories for user " + user);
}
}
/**
* Initialize the local cache directories for a particular user.
* <ul>
* <li>$local.dir/usercache/$user</li>
* <li>$local.dir/usercache/$user/appcache</li>
* <li>$local.dir/usercache/$user/filecache</li>
* </ul>
*
* @param localDirs the target directories to create
* @param user the user whose local cache directories should be initialized
* @throws IOException if there's an issue initializing the cache
* directories
*/
void createUserCacheDirs(List<String> localDirs, String user)
throws IOException {
LOG.info("Initializing user {}", user);
boolean appcacheDirStatus = false;
boolean distributedCacheDirStatus = false;
FsPermission appCachePerms = new FsPermission(APPCACHE_PERM);
FsPermission fileperms = new FsPermission(FILECACHE_PERM);
for (String localDir : localDirs) {
// create $local.dir/usercache/$user/appcache
Path localDirPath = new Path(localDir);
final Path appDir = getAppcacheDir(localDirPath, user);
try {
createDir(appDir, appCachePerms, true, user);
appcacheDirStatus = true;
} catch (IOException e) {
LOG.warn("Unable to create app cache directory : {}", appDir, e);
}
// create $local.dir/usercache/$user/filecache
final Path distDir = getFileCacheDir(localDirPath, user);
try {
createDir(distDir, fileperms, true, user);
distributedCacheDirStatus = true;
} catch (IOException e) {
LOG.warn("Unable to create file cache directory : {}", distDir, e);
}
}
if (!appcacheDirStatus) {
throw new IOException("Not able to initialize app-cache directories "
+ "in any of the configured local directories for user " + user);
}
if (!distributedCacheDirStatus) {
throw new IOException(
"Not able to initialize distributed-cache directories "
+ "in any of the configured local directories for user "
+ user);
}
}
/**
* Initialize the local directories for a particular user.
* <ul>
* <li>$local.dir/usercache/$user/appcache/$appid</li>
* </ul>
*
* @param localDirs the target directories to create
* @param user the user whose local cache directories should be initialized
* @param appId the application ID
* @throws IOException if there's an issue initializing the application
* directories
*/
void createAppDirs(List<String> localDirs, String user, String appId)
throws IOException {
boolean initAppDirStatus = false;
FsPermission appperms = new FsPermission(APPDIR_PERM);
for (String localDir : localDirs) {
Path fullAppDir = getApplicationDir(new Path(localDir), user, appId);
// create $local.dir/usercache/$user/appcache/$appId
try {
createDir(fullAppDir, appperms, true, user);
initAppDirStatus = true;
} catch (IOException e) {
LOG.warn("Unable to create app directory {}",
fullAppDir, e);
}
}
if (!initAppDirStatus) {
throw new IOException("Not able to initialize app directories "
+ "in any of the configured local directories for app "
+ appId.toString());
}
}
/**
* Create application log directories on all disks.
*
* @param appId the application ID
* @param logDirs the target directories to create
* @param user the user whose local cache directories should be initialized
* @throws IOException if there's an issue initializing the application log
* directories
*/
void createAppLogDirs(String appId, List<String> logDirs, String user)
throws IOException {
boolean appLogDirStatus = false;
FsPermission appLogDirPerms = new
FsPermission(getLogDirPermissions());
for (String rootLogDir : logDirs) {
// create $log.dir/$appid
Path appLogDir = new Path(rootLogDir, appId);
try {
createDir(appLogDir, appLogDirPerms, true, user);
} catch (IOException e) {
LOG.warn("Unable to create the app-log directory : {}", appLogDir, e);
continue;
}
appLogDirStatus = true;
}
if (!appLogDirStatus) {
throw new IOException("Not able to initialize app-log directories "
+ "in any of the configured local directories for app " + appId);
}
}
/**
* Create application log directories on all disks.
*
* @param appId the application ID
* @param containerId the container ID
* @param logDirs the target directories to create
* @param user the user as whom the directories should be created.
* Used only on secure Windows hosts.
* @throws IOException if there's an issue initializing the container log
* directories
*/
void createContainerLogDirs(String appId, String containerId,
List<String> logDirs, String user) throws IOException {
boolean containerLogDirStatus = false;
FsPermission containerLogDirPerms = new
FsPermission(getLogDirPermissions());
for (String rootLogDir : logDirs) {
// create $log.dir/$appid/$containerid
Path appLogDir = new Path(rootLogDir, appId);
Path containerLogDir = new Path(appLogDir, containerId);
try {
createDir(containerLogDir, containerLogDirPerms, true, user);
} catch (IOException e) {
LOG.warn("Unable to create the container-log directory : {}",
appLogDir, e);
continue;
}
containerLogDirStatus = true;
}
if (!containerLogDirStatus) {
throw new IOException(
"Not able to initialize container-log directories "
+ "in any of the configured local directories for container "
+ containerId);
}
}
/**
* Return the default container log directory permissions.
*
* @return the default container log directory permissions
*/
@VisibleForTesting
public String getLogDirPermissions() {
if (this.logDirPermissions==null) {
this.logDirPermissions = getConf().get(
YarnConfiguration.NM_DEFAULT_CONTAINER_EXECUTOR_LOG_DIRS_PERMISSIONS,
YarnConfiguration.NM_DEFAULT_CONTAINER_EXECUTOR_LOG_DIRS_PERMISSIONS_DEFAULT);
}
return this.logDirPermissions;
}
/**
* Clear the internal variable for repeatable testing.
*/
@VisibleForTesting
public void clearLogDirPermissions() {
this.logDirPermissions = null;
}
/**
*
* @param ctx Encapsulates information necessary for exec containers.
* @return the input/output stream of interactive docker shell.
* @throws ContainerExecutionException
*/
@Override
public IOStreamPair execContainer(ContainerExecContext ctx)
throws ContainerExecutionException {
return null;
}
/**
* Return the list of paths of given local directories.
*
* @return the list of paths of given local directories
*/
private static List<Path> getPaths(List<String> dirs) {
List<Path> paths = new ArrayList<>(dirs.size());
for (int i = 0; i < dirs.size(); i++) {
paths.add(new Path(dirs.get(i)));
}
return paths;
}
@Override
public void updateYarnSysFS(Context ctx, String user,
String appId, String spec) throws IOException {
throw new ServiceStateException("Implementation unavailable");
}
@Override
public int reacquireContainer(ContainerReacquisitionContext ctx)
throws IOException, InterruptedException {
try {
if (numaResourceAllocator != null) {
numaResourceAllocator.recoverNumaResource(ctx.getContainerId());
}
return super.reacquireContainer(ctx);
} finally {
postComplete(ctx.getContainerId());
}
}
/**
* clean up and release of resources.
*
* @param containerId containerId of running container
*/
public void postComplete(final ContainerId containerId) {
if (numaResourceAllocator != null) {
try {
numaResourceAllocator.releaseNumaResource(containerId);
} catch (ResourceHandlerException e) {
LOG.warn("NumaResource release failed for " +
"containerId: {}. Exception: ", containerId, e);
}
}
}
/**
* @param resourceAllocation NonNull NumaResourceAllocation object reference
* @return Array of numa specific commands
*/
String[] getNumaCommands(NumaResourceAllocation resourceAllocation) {
String[] numaCommand = new String[3];
numaCommand[0] = numactl;
numaCommand[1] = "--interleave=" + String.join(",", resourceAllocation.getMemNodes());
numaCommand[2] = "--cpunodebind=" + String.join(",", resourceAllocation.getCpuNodes());
return numaCommand;
}
/**
* @param firstStringArray Array of String
* @param secondStringArray Array of String
* @return combined array of string where first elements are from firstStringArray
* and later are the elements from secondStringArray
*/
String[] concatStringCommands(String[] firstStringArray, String[] secondStringArray) {
if(firstStringArray == null && secondStringArray == null) {
return secondStringArray;
}
else if(firstStringArray == null || firstStringArray.length == 0) {
return secondStringArray;
}
else if(secondStringArray == null || secondStringArray.length == 0){
return firstStringArray;
}
int len = firstStringArray.length + secondStringArray.length;
String[] ret = new String[len];
int idx = 0;
for (String s : firstStringArray) {
ret[idx] = s;
idx++;
}
for (String s : secondStringArray) {
ret[idx] = s;
idx++;
}
return ret;
}
@VisibleForTesting
public void setNumaResourceAllocator(NumaResourceAllocator numaResourceAllocator) {
this.numaResourceAllocator = numaResourceAllocator;
}
@VisibleForTesting
public void setNumactl(String numactl) {
this.numactl = numactl;
}
}