YARN-9860. Enable service mode for Docker containers on YARN
Contributed by Prabhu Joseph and Shane Kumpf
This commit is contained in:
Eric Yang
parent
7a4b3d42c4
commit
31e0122f4d
@ -24,6 +24,7 @@
|
||||
import io.swagger.annotations.ApiModelProperty;
|
||||
import org.apache.hadoop.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.classification.InterfaceStability;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
|
||||
|
||||
import javax.xml.bind.annotation.XmlElement;
|
||||
import javax.xml.bind.annotation.XmlEnum;
|
||||
@ -73,6 +74,7 @@ public String toString() {
|
||||
private TypeEnum type = null;
|
||||
private String destFile = null;
|
||||
private String srcFile = null;
|
||||
private LocalResourceVisibility visibility = null;
|
||||
private Map<String, String> properties = new HashMap<>();
|
||||
|
||||
public ConfigFile copy() {
|
||||
@ -80,6 +82,7 @@ public ConfigFile copy() {
|
||||
copy.setType(this.getType());
|
||||
copy.setSrcFile(this.getSrcFile());
|
||||
copy.setDestFile(this.getDestFile());
|
||||
copy.setVisibility(this.visibility);
|
||||
if (this.getProperties() != null && !this.getProperties().isEmpty()) {
|
||||
copy.getProperties().putAll(this.getProperties());
|
||||
}
|
||||
@ -150,6 +153,26 @@ public void setSrcFile(String srcFile) {
|
||||
this.srcFile = srcFile;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Visibility of the Config file.
|
||||
**/
|
||||
public ConfigFile visibility(LocalResourceVisibility localrsrcVisibility) {
|
||||
this.visibility = localrsrcVisibility;
|
||||
return this;
|
||||
}
|
||||
|
||||
@ApiModelProperty(example = "null", value = "Visibility of the Config file")
|
||||
@JsonProperty("visibility")
|
||||
public LocalResourceVisibility getVisibility() {
|
||||
return visibility;
|
||||
}
|
||||
|
||||
@XmlElement(name = "visibility", defaultValue="APPLICATION")
|
||||
public void setVisibility(LocalResourceVisibility localrsrcVisibility) {
|
||||
this.visibility = localrsrcVisibility;
|
||||
}
|
||||
|
||||
/**
|
||||
A blob of key value pairs that will be dumped in the dest_file in the format
|
||||
as specified in type. If src_file is specified, src_file content are dumped
|
||||
@ -200,12 +223,13 @@ public boolean equals(java.lang.Object o) {
|
||||
return Objects.equals(this.type, configFile.type)
|
||||
&& Objects.equals(this.destFile, configFile.destFile)
|
||||
&& Objects.equals(this.srcFile, configFile.srcFile)
|
||||
&& Objects.equals(this.visibility, configFile.visibility)
|
||||
&& Objects.equals(this.properties, configFile.properties);
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(type, destFile, srcFile, properties);
|
||||
return Objects.hash(type, destFile, srcFile, visibility, properties);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -217,6 +241,8 @@ public String toString() {
|
||||
.append(" destFile: ").append(toIndentedString(destFile))
|
||||
.append("\n")
|
||||
.append(" srcFile: ").append(toIndentedString(srcFile)).append("\n")
|
||||
.append(" visibility: ").append(toIndentedString(visibility))
|
||||
.append("\n")
|
||||
.append(" properties: ").append(toIndentedString(properties))
|
||||
.append("\n")
|
||||
.append("}");
|
||||
|
||||
@ -817,6 +817,21 @@ public int actionDestroy(String serviceName) throws YarnException,
|
||||
+ appDir);
|
||||
ret = EXIT_NOT_FOUND;
|
||||
}
|
||||
|
||||
// Delete Public Resource Dir
|
||||
Path publicResourceDir = new Path(fs.getBasePath(), serviceName);
|
||||
if (fileSystem.exists(publicResourceDir)) {
|
||||
if (fileSystem.delete(publicResourceDir, true)) {
|
||||
LOG.info("Successfully deleted public resource dir for "
|
||||
+ serviceName + ": " + publicResourceDir);
|
||||
} else {
|
||||
String message = "Failed to delete public resource dir for service "
|
||||
+ serviceName + " at: " + publicResourceDir;
|
||||
LOG.info(message);
|
||||
throw new YarnException(message);
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
deleteZKNode(serviceName);
|
||||
// don't set destroySucceed to false if no ZK node exists because not
|
||||
@ -1315,7 +1330,8 @@ private boolean addAMLog4jResource(String serviceName, Configuration conf,
|
||||
new Path(remoteConfPath, YarnServiceConstants.YARN_SERVICE_LOG4J_FILENAME);
|
||||
copy(conf, localFilePath, remoteFilePath);
|
||||
LocalResource localResource =
|
||||
fs.createAmResource(remoteConfPath, LocalResourceType.FILE);
|
||||
fs.createAmResource(remoteConfPath, LocalResourceType.FILE,
|
||||
LocalResourceVisibility.APPLICATION);
|
||||
localResources.put(localFilePath.getName(), localResource);
|
||||
hasAMLog4j = true;
|
||||
} else {
|
||||
@ -1465,7 +1481,7 @@ private void addKeytabResourceIfSecure(SliderFileSystem fileSystem,
|
||||
return;
|
||||
}
|
||||
LocalResource keytabRes = fileSystem.createAmResource(keytabOnhdfs,
|
||||
LocalResourceType.FILE);
|
||||
LocalResourceType.FILE, LocalResourceVisibility.PRIVATE);
|
||||
localResource.put(String.format(YarnServiceConstants.KEYTAB_LOCATION,
|
||||
service.getName()), keytabRes);
|
||||
LOG.info("Adding " + service.getName() + "'s keytab for "
|
||||
|
||||
@ -47,6 +47,8 @@ public interface YarnServiceConstants {
|
||||
|
||||
String SERVICES_DIRECTORY = "services";
|
||||
|
||||
String SERVICES_PUBLIC_DIRECTORY = "/tmp/hadoop-yarn/staging/";
|
||||
|
||||
/**
|
||||
* JVM property to define the service lib directory;
|
||||
* this is set by the yarn.sh script
|
||||
|
||||
@ -27,6 +27,7 @@
|
||||
import org.apache.hadoop.yarn.api.records.Container;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResource;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResourceType;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
|
||||
import org.apache.hadoop.yarn.service.ServiceContext;
|
||||
import org.apache.hadoop.yarn.service.api.records.ConfigFile;
|
||||
import org.apache.hadoop.yarn.service.api.records.ConfigFormat;
|
||||
@ -191,6 +192,17 @@ public static Path initCompInstanceDir(SliderFileSystem fs,
|
||||
return compInstanceDir;
|
||||
}
|
||||
|
||||
public static Path initCompPublicResourceDir(SliderFileSystem fs,
|
||||
ContainerLaunchService.ComponentLaunchContext compLaunchContext,
|
||||
ComponentInstance instance) {
|
||||
Path compDir = fs.getComponentPublicResourceDir(
|
||||
compLaunchContext.getServiceVersion(), compLaunchContext.getName());
|
||||
Path compPublicResourceDir = new Path(compDir,
|
||||
instance.getCompInstanceName());
|
||||
return compPublicResourceDir;
|
||||
}
|
||||
|
||||
|
||||
// 1. Create all config files for a component on hdfs for localization
|
||||
// 2. Add the config file to localResource
|
||||
public static synchronized void createConfigFileAndAddLocalResource(
|
||||
@ -212,6 +224,20 @@ public static synchronized void createConfigFileAndAddLocalResource(
|
||||
log.info("Component instance conf dir already exists: " + compInstanceDir);
|
||||
}
|
||||
|
||||
Path compPublicResourceDir = initCompPublicResourceDir(fs,
|
||||
compLaunchContext, instance);
|
||||
if (!fs.getFileSystem().exists(compPublicResourceDir)) {
|
||||
log.info("{} version {} : Creating Public Resource dir on hdfs: {}",
|
||||
instance.getCompInstanceId(), compLaunchContext.getServiceVersion(),
|
||||
compPublicResourceDir);
|
||||
fs.getFileSystem().mkdirs(compPublicResourceDir,
|
||||
new FsPermission(FsAction.ALL, FsAction.READ_EXECUTE,
|
||||
FsAction.EXECUTE));
|
||||
} else {
|
||||
log.info("Component instance public resource dir already exists: "
|
||||
+ compPublicResourceDir);
|
||||
}
|
||||
|
||||
log.debug("Tokens substitution for component instance: {}{}{}" + instance
|
||||
.getCompInstanceName(), System.lineSeparator(), tokensForSubstitution);
|
||||
|
||||
@ -236,7 +262,14 @@ public static synchronized void createConfigFileAndAddLocalResource(
|
||||
* substitution and merges in new configs, and writes a new file to
|
||||
* compInstanceDir/fileName.
|
||||
*/
|
||||
Path remoteFile = new Path(compInstanceDir, fileName);
|
||||
Path remoteFile = null;
|
||||
LocalResourceVisibility visibility = configFile.getVisibility();
|
||||
if (visibility != null &&
|
||||
visibility.equals(LocalResourceVisibility.PUBLIC)) {
|
||||
remoteFile = new Path(compPublicResourceDir, fileName);
|
||||
} else {
|
||||
remoteFile = new Path(compInstanceDir, fileName);
|
||||
}
|
||||
|
||||
if (!fs.getFileSystem().exists(remoteFile)) {
|
||||
log.info("Saving config file on hdfs for component " + instance
|
||||
@ -268,7 +301,8 @@ public static synchronized void createConfigFileAndAddLocalResource(
|
||||
|
||||
// Add resource for localization
|
||||
LocalResource configResource =
|
||||
fs.createAmResource(remoteFile, LocalResourceType.FILE);
|
||||
fs.createAmResource(remoteFile, LocalResourceType.FILE,
|
||||
configFile.getVisibility());
|
||||
Path destFile = new Path(configFile.getDestFile());
|
||||
String symlink = APP_CONF_DIR + "/" + fileName;
|
||||
addLocalResource(launcher, symlink, configResource, destFile,
|
||||
@ -311,7 +345,8 @@ public static synchronized void handleStaticFilesForLocalization(
|
||||
LocalResource localResource = fs.createAmResource(sourceFile,
|
||||
(staticFile.getType() == ConfigFile.TypeEnum.ARCHIVE ?
|
||||
LocalResourceType.ARCHIVE :
|
||||
LocalResourceType.FILE));
|
||||
LocalResourceType.FILE), staticFile.getVisibility());
|
||||
|
||||
Path destFile = new Path(sourceFile.getName());
|
||||
if (staticFile.getDestFile() != null && !staticFile.getDestFile()
|
||||
.isEmpty()) {
|
||||
|
||||
@ -20,6 +20,7 @@
|
||||
import org.apache.hadoop.fs.Path;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResource;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResourceType;
|
||||
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
|
||||
import org.apache.hadoop.yarn.service.api.records.Service;
|
||||
import org.apache.hadoop.yarn.service.component.instance.ComponentInstance;
|
||||
import org.apache.hadoop.yarn.service.containerlaunch.ContainerLaunchService;
|
||||
@ -43,7 +44,8 @@ public void processArtifact(AbstractLauncher launcher,
|
||||
}
|
||||
log.info("Adding resource {}", artifact);
|
||||
LocalResourceType type = LocalResourceType.ARCHIVE;
|
||||
LocalResource packageResource = fileSystem.createAmResource(artifact, type);
|
||||
LocalResource packageResource = fileSystem.createAmResource(artifact, type,
|
||||
LocalResourceVisibility.APPLICATION);
|
||||
launcher.addLocalResource(APP_LIB_DIR, packageResource);
|
||||
}
|
||||
}
|
||||
|
||||
@ -384,13 +384,19 @@ public Path getHomeDirectory() {
|
||||
* @param resourceType resource type
|
||||
* @return the local resource for AM
|
||||
*/
|
||||
public LocalResource createAmResource(Path destPath, LocalResourceType resourceType) throws IOException {
|
||||
public LocalResource createAmResource(Path destPath,
|
||||
LocalResourceType resourceType,
|
||||
LocalResourceVisibility visibility) throws IOException {
|
||||
|
||||
FileStatus destStatus = fileSystem.getFileStatus(destPath);
|
||||
LocalResource amResource = Records.newRecord(LocalResource.class);
|
||||
amResource.setType(resourceType);
|
||||
// Set visibility of the resource
|
||||
// Setting to most private option
|
||||
amResource.setVisibility(LocalResourceVisibility.APPLICATION);
|
||||
if (visibility == null) {
|
||||
visibility = LocalResourceVisibility.APPLICATION;
|
||||
}
|
||||
amResource.setVisibility(visibility);
|
||||
// Set the resource to be copied over
|
||||
amResource.setResource(
|
||||
URL.fromPath(fileSystem.resolvePath(destStatus.getPath())));
|
||||
@ -419,7 +425,7 @@ public Map<String, LocalResource> submitDirectory(Path srcDir, String destRelati
|
||||
for (FileStatus entry : fileset) {
|
||||
|
||||
LocalResource resource = createAmResource(entry.getPath(),
|
||||
LocalResourceType.FILE);
|
||||
LocalResourceType.FILE, LocalResourceVisibility.APPLICATION);
|
||||
String relativePath = destRelativeDir + "/" + entry.getPath().getName();
|
||||
localResources.put(relativePath, resource);
|
||||
}
|
||||
@ -465,7 +471,8 @@ public LocalResource submitFile(File localFile, Path tempPath, String subdir, St
|
||||
// Set the type of resource - file or archive
|
||||
// archives are untarred at destination
|
||||
// we don't need the jar file to be untarred for now
|
||||
return createAmResource(destPath, LocalResourceType.FILE);
|
||||
return createAmResource(destPath, LocalResourceType.FILE,
|
||||
LocalResourceVisibility.APPLICATION);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -483,7 +490,7 @@ public void submitTarGzipAndUpdate(
|
||||
BadClusterStateException {
|
||||
Path dependencyLibTarGzip = getDependencyTarGzip();
|
||||
LocalResource lc = createAmResource(dependencyLibTarGzip,
|
||||
LocalResourceType.ARCHIVE);
|
||||
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION);
|
||||
providerResources.put(YarnServiceConstants.DEPENDENCY_LOCALIZED_DIR_LINK, lc);
|
||||
}
|
||||
|
||||
|
||||
@ -21,6 +21,7 @@
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.fs.FileSystem;
|
||||
import org.apache.hadoop.fs.Path;
|
||||
import org.apache.hadoop.yarn.service.conf.YarnServiceConstants;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
@ -63,6 +64,26 @@ public Path getComponentDir(String serviceVersion, String compName) {
|
||||
serviceVersion + "/" + compName);
|
||||
}
|
||||
|
||||
public Path getBasePath() {
|
||||
String tmpDir = configuration.get("hadoop.tmp.dir");
|
||||
String basePath = YarnServiceConstants.SERVICE_BASE_DIRECTORY
|
||||
+ "/" + YarnServiceConstants.SERVICES_DIRECTORY;
|
||||
return new Path(tmpDir, basePath);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the component public resource directory path.
|
||||
*
|
||||
* @param serviceVersion service version
|
||||
* @param compName component name
|
||||
* @return component public resource directory
|
||||
*/
|
||||
public Path getComponentPublicResourceDir(String serviceVersion,
|
||||
String compName) {
|
||||
return new Path(new Path(getBasePath(), getAppDir().getName() + "/"
|
||||
+ "components"), serviceVersion + "/" + compName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Deletes the component directory.
|
||||
*
|
||||
@ -77,6 +98,12 @@ public void deleteComponentDir(String serviceVersion, String compName)
|
||||
fileSystem.delete(path, true);
|
||||
LOG.debug("deleted dir {}", path);
|
||||
}
|
||||
Path publicResourceDir = getComponentPublicResourceDir(serviceVersion,
|
||||
compName);
|
||||
if (fileSystem.exists(publicResourceDir)) {
|
||||
fileSystem.delete(publicResourceDir, true);
|
||||
LOG.debug("deleted public resource dir {}", publicResourceDir);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@ -92,6 +119,13 @@ public void deleteComponentsVersionDirIfEmpty(String serviceVersion)
|
||||
fileSystem.delete(path, true);
|
||||
LOG.info("deleted dir {}", path);
|
||||
}
|
||||
Path publicResourceDir = new Path(new Path(getBasePath(),
|
||||
getAppDir().getName() + "/" + "components"), serviceVersion);
|
||||
if (fileSystem.exists(publicResourceDir)
|
||||
&& fileSystem.listStatus(publicResourceDir).length == 0) {
|
||||
fileSystem.delete(publicResourceDir, true);
|
||||
LOG.info("deleted public resource dir {}", publicResourceDir);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
@ -63,95 +63,100 @@ public void testStaticFileLocalization() throws IOException {
|
||||
List<ConfigFile> configFileList = new ArrayList<>();
|
||||
when(conf.getFiles()).thenReturn(configFileList);
|
||||
when(compLaunchCtx.getConfiguration()).thenReturn(conf);
|
||||
when(sfs.createAmResource(any(Path.class), any(LocalResourceType.class)))
|
||||
.thenAnswer(invocationOnMock -> new LocalResource() {
|
||||
@Override
|
||||
public URL getResource() {
|
||||
return URL.fromPath(((Path) invocationOnMock.getArguments()[0]));
|
||||
}
|
||||
when(sfs.createAmResource(any(Path.class), any(LocalResourceType.class),
|
||||
any(LocalResourceVisibility.class))).thenAnswer(
|
||||
invocationOnMock -> new LocalResource() {
|
||||
@Override
|
||||
public URL getResource() {
|
||||
return URL.fromPath(((Path) invocationOnMock.getArguments()[0]));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setResource(URL resource) {
|
||||
@Override
|
||||
public void setResource(URL resource) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public long getSize() {
|
||||
return 0;
|
||||
}
|
||||
@Override
|
||||
public long getSize() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setSize(long size) {
|
||||
@Override
|
||||
public void setSize(long size) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public long getTimestamp() {
|
||||
return 0;
|
||||
}
|
||||
@Override
|
||||
public long getTimestamp() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setTimestamp(long timestamp) {
|
||||
@Override
|
||||
public void setTimestamp(long timestamp) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public LocalResourceType getType() {
|
||||
return (LocalResourceType) invocationOnMock.getArguments()[1];
|
||||
}
|
||||
@Override
|
||||
public LocalResourceType getType() {
|
||||
return (LocalResourceType) invocationOnMock.getArguments()[1];
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setType(LocalResourceType type) {
|
||||
@Override
|
||||
public void setType(LocalResourceType type) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public LocalResourceVisibility getVisibility() {
|
||||
return null;
|
||||
}
|
||||
@Override
|
||||
public LocalResourceVisibility getVisibility() {
|
||||
return LocalResourceVisibility.APPLICATION;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setVisibility(LocalResourceVisibility visibility) {
|
||||
@Override
|
||||
public void setVisibility(LocalResourceVisibility visibility) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getPattern() {
|
||||
return null;
|
||||
}
|
||||
@Override
|
||||
public String getPattern() {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setPattern(String pattern) {
|
||||
@Override
|
||||
public void setPattern(String pattern) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean getShouldBeUploadedToSharedCache() {
|
||||
return false;
|
||||
}
|
||||
@Override
|
||||
public boolean getShouldBeUploadedToSharedCache() {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setShouldBeUploadedToSharedCache(
|
||||
boolean shouldBeUploadedToSharedCache) {
|
||||
@Override
|
||||
public void setShouldBeUploadedToSharedCache(
|
||||
boolean shouldBeUploadedToSharedCache) {
|
||||
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
// Initialize list of files.
|
||||
//archive
|
||||
configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile1")
|
||||
.destFile("destFile1").type(ConfigFile.TypeEnum.ARCHIVE));
|
||||
.destFile("destFile1").type(ConfigFile.TypeEnum.ARCHIVE)
|
||||
.visibility(LocalResourceVisibility.APPLICATION));
|
||||
|
||||
//static file
|
||||
configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile2")
|
||||
.destFile("folder/destFile_2").type(ConfigFile.TypeEnum.STATIC));
|
||||
.destFile("folder/destFile_2").type(ConfigFile.TypeEnum.STATIC)
|
||||
.visibility(LocalResourceVisibility.APPLICATION));
|
||||
|
||||
//This will be ignored since type is JSON
|
||||
configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile3")
|
||||
.destFile("destFile3").type(ConfigFile.TypeEnum.JSON));
|
||||
.destFile("destFile3").type(ConfigFile.TypeEnum.JSON)
|
||||
.visibility(LocalResourceVisibility.APPLICATION));
|
||||
//No destination file specified
|
||||
configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile4")
|
||||
.type(ConfigFile.TypeEnum.STATIC));
|
||||
.type(ConfigFile.TypeEnum.STATIC)
|
||||
.visibility(LocalResourceVisibility.APPLICATION));
|
||||
|
||||
ProviderService.ResolvedLaunchParams resolved =
|
||||
new ProviderService.ResolvedLaunchParams();
|
||||
|
||||
@ -235,6 +235,9 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
|
||||
@InterfaceAudience.Private
|
||||
public static final String ENV_DOCKER_CONTAINER_DOCKER_RUNTIME =
|
||||
"YARN_CONTAINER_RUNTIME_DOCKER_RUNTIME";
|
||||
@InterfaceAudience.Private
|
||||
public static final String ENV_DOCKER_CONTAINER_DOCKER_SERVICE_MODE =
|
||||
"YARN_CONTAINER_RUNTIME_DOCKER_SERVICE_MODE";
|
||||
|
||||
@InterfaceAudience.Private
|
||||
private static final String RUNTIME_TYPE = "DOCKER";
|
||||
@ -588,7 +591,9 @@ public void launchContainer(ContainerRuntimeContext ctx)
|
||||
String network = environment.get(ENV_DOCKER_CONTAINER_NETWORK);
|
||||
String hostname = environment.get(ENV_DOCKER_CONTAINER_HOSTNAME);
|
||||
String runtime = environment.get(ENV_DOCKER_CONTAINER_DOCKER_RUNTIME);
|
||||
boolean useEntryPoint = checkUseEntryPoint(environment);
|
||||
boolean serviceMode = Boolean.parseBoolean(environment.get(
|
||||
ENV_DOCKER_CONTAINER_DOCKER_SERVICE_MODE));
|
||||
boolean useEntryPoint = serviceMode || checkUseEntryPoint(environment);
|
||||
|
||||
if (imageName == null || imageName.isEmpty()) {
|
||||
imageName = defaultImageName;
|
||||
@ -679,10 +684,12 @@ public void launchContainer(ContainerRuntimeContext ctx)
|
||||
runCommand.addRuntime(runtime);
|
||||
}
|
||||
|
||||
runCommand.addAllReadWriteMountLocations(containerLogDirs);
|
||||
runCommand.addAllReadWriteMountLocations(applicationLocalDirs);
|
||||
runCommand.addAllReadOnlyMountLocations(filecacheDirs);
|
||||
runCommand.addAllReadOnlyMountLocations(userFilecacheDirs);
|
||||
if (!serviceMode) {
|
||||
runCommand.addAllReadWriteMountLocations(containerLogDirs);
|
||||
runCommand.addAllReadWriteMountLocations(applicationLocalDirs);
|
||||
runCommand.addAllReadOnlyMountLocations(filecacheDirs);
|
||||
runCommand.addAllReadOnlyMountLocations(userFilecacheDirs);
|
||||
}
|
||||
|
||||
if (environment.containsKey(ENV_DOCKER_CONTAINER_MOUNTS)) {
|
||||
Matcher parsedMounts = USER_MOUNT_PATTERN.matcher(
|
||||
@ -800,11 +807,20 @@ public void launchContainer(ContainerRuntimeContext ctx)
|
||||
runCommand.setYarnSysFS(true);
|
||||
}
|
||||
|
||||
// In service mode, the YARN log dirs are not mounted into the container.
|
||||
// As a result, the container fails to start due to stdout and stderr output
|
||||
// being sent to a file in a directory that does not exist. In service mode,
|
||||
// only supply the command with no stdout or stderr redirection.
|
||||
List<String> commands = container.getLaunchContext().getCommands();
|
||||
if (serviceMode) {
|
||||
commands = Arrays.asList(
|
||||
String.join(" ", commands).split("1>")[0].split(" "));
|
||||
}
|
||||
|
||||
if (useEntryPoint) {
|
||||
runCommand.setOverrideDisabled(true);
|
||||
runCommand.addEnv(environment);
|
||||
runCommand.setOverrideCommandWithArgs(container.getLaunchContext()
|
||||
.getCommands());
|
||||
runCommand.setOverrideCommandWithArgs(commands);
|
||||
runCommand.disableDetach();
|
||||
runCommand.setLogDir(container.getLogDir());
|
||||
} else {
|
||||
@ -818,6 +834,10 @@ public void launchContainer(ContainerRuntimeContext ctx)
|
||||
runCommand.detachOnRun();
|
||||
}
|
||||
|
||||
if (serviceMode) {
|
||||
runCommand.setServiceMode(serviceMode);
|
||||
}
|
||||
|
||||
if(enableUserReMapping) {
|
||||
if (!allowPrivilegedContainerExecution(container)) {
|
||||
runCommand.groupAdd(groups);
|
||||
@ -1279,11 +1299,14 @@ private void handleContainerKill(ContainerRuntimeContext ctx,
|
||||
throw new ContainerExecutionException(e);
|
||||
}
|
||||
|
||||
boolean serviceMode = Boolean.parseBoolean(env.get(
|
||||
ENV_DOCKER_CONTAINER_DOCKER_SERVICE_MODE));
|
||||
|
||||
// Only need to check whether the container was asked to be privileged.
|
||||
// If the container had failed the permissions checks upon launch, it
|
||||
// would have never been launched and thus we wouldn't be here
|
||||
// attempting to signal it.
|
||||
if (isContainerRequestedAsPrivileged(container)) {
|
||||
if (isContainerRequestedAsPrivileged(container) || serviceMode) {
|
||||
String containerId = container.getContainerId().toString();
|
||||
DockerCommandExecutor.DockerContainerStatus containerStatus =
|
||||
DockerCommandExecutor.getContainerStatus(containerId,
|
||||
|
||||
@ -199,6 +199,12 @@ public DockerRunCommand setLogDir(String logDir) {
|
||||
return this;
|
||||
}
|
||||
|
||||
public DockerRunCommand setServiceMode(boolean serviceMode) {
|
||||
String value = Boolean.toString(serviceMode);
|
||||
super.addCommandArguments("service-mode", value);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if user defined environment variables are empty.
|
||||
*
|
||||
|
||||
@ -325,12 +325,6 @@ int sync_yarn_sysfs(char* const* local_dirs, const char *running_user,
|
||||
*/
|
||||
int execute_regex_match(const char *regex_str, const char *input);
|
||||
|
||||
/**
|
||||
* Validate the docker image name matches the expected input.
|
||||
* Return 0 on success.
|
||||
*/
|
||||
int validate_docker_image_name(const char *image_name);
|
||||
|
||||
struct configuration* get_cfg();
|
||||
|
||||
/**
|
||||
|
||||
@ -28,6 +28,7 @@
|
||||
#include "docker-util.h"
|
||||
#include "string-utils.h"
|
||||
#include "util.h"
|
||||
#include "container-executor.h"
|
||||
#include <grp.h>
|
||||
#include <pwd.h>
|
||||
#include <errno.h>
|
||||
@ -374,6 +375,8 @@ const char *get_docker_error_message(const int error_code) {
|
||||
return "Invalid docker tmpfs mount";
|
||||
case INVALID_DOCKER_RUNTIME:
|
||||
return "Invalid docker runtime";
|
||||
case SERVICE_MODE_DISABLED:
|
||||
return "Service mode disabled";
|
||||
default:
|
||||
return "Unknown error";
|
||||
}
|
||||
@ -987,6 +990,22 @@ static int set_runtime(const struct configuration *command_config,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int is_service_mode_enabled(const struct configuration *command_config,
|
||||
const struct configuration *executor_cfg, args *args) {
|
||||
int ret = 0;
|
||||
struct section *section = get_configuration_section(CONTAINER_EXECUTOR_CFG_DOCKER_SECTION, executor_cfg);
|
||||
char *value = get_configuration_value("service-mode", DOCKER_COMMAND_FILE_SECTION, command_config);
|
||||
if (value != NULL && strcasecmp(value, "true") == 0) {
|
||||
if (is_feature_enabled(DOCKER_SERVICE_MODE_ENABLED_KEY, ret, section)) {
|
||||
ret = 1;
|
||||
} else {
|
||||
ret = SERVICE_MODE_DISABLED;
|
||||
}
|
||||
}
|
||||
free(value);
|
||||
return ret;
|
||||
}
|
||||
|
||||
static int add_ports_mapping_to_command(const struct configuration *command_config, args *args) {
|
||||
int i = 0, ret = 0;
|
||||
char *network_type = (char*) malloc(128);
|
||||
@ -1595,12 +1614,19 @@ int get_docker_run_command(const char *command_file, const struct configuration
|
||||
char *privileged = NULL;
|
||||
char *no_new_privileges_enabled = NULL;
|
||||
char *use_entry_point = NULL;
|
||||
int service_mode_enabled = 0;
|
||||
struct configuration command_config = {0, NULL};
|
||||
ret = read_and_verify_command_file(command_file, DOCKER_RUN_COMMAND, &command_config);
|
||||
if (ret != 0) {
|
||||
goto free_and_exit;
|
||||
}
|
||||
|
||||
service_mode_enabled = is_service_mode_enabled(&command_config, conf, args);
|
||||
if (service_mode_enabled == SERVICE_MODE_DISABLED) {
|
||||
ret = SERVICE_MODE_DISABLED;
|
||||
goto free_and_exit;
|
||||
}
|
||||
|
||||
use_entry_point = get_configuration_value("use-entry-point", DOCKER_COMMAND_FILE_SECTION, &command_config);
|
||||
if (use_entry_point != NULL && strcasecmp(use_entry_point, "true") == 0) {
|
||||
entry_point = 1;
|
||||
@ -1612,10 +1638,13 @@ int get_docker_run_command(const char *command_file, const struct configuration
|
||||
ret = INVALID_DOCKER_CONTAINER_NAME;
|
||||
goto free_and_exit;
|
||||
}
|
||||
user = get_configuration_value("user", DOCKER_COMMAND_FILE_SECTION, &command_config);
|
||||
if (user == NULL) {
|
||||
ret = INVALID_DOCKER_USER_NAME;
|
||||
goto free_and_exit;
|
||||
|
||||
if (!service_mode_enabled) {
|
||||
user = get_configuration_value("user", DOCKER_COMMAND_FILE_SECTION, &command_config);
|
||||
if (user == NULL) {
|
||||
ret = INVALID_DOCKER_USER_NAME;
|
||||
goto free_and_exit;
|
||||
}
|
||||
}
|
||||
image = get_configuration_value("image", DOCKER_COMMAND_FILE_SECTION, &command_config);
|
||||
if (image == NULL || validate_docker_image_name(image) != 0) {
|
||||
@ -1640,12 +1669,14 @@ int get_docker_run_command(const char *command_file, const struct configuration
|
||||
privileged = get_configuration_value("privileged", DOCKER_COMMAND_FILE_SECTION, &command_config);
|
||||
|
||||
if (privileged == NULL || strcmp(privileged, "false") == 0) {
|
||||
char *user_buffer = make_string("--user=%s", user);
|
||||
ret = add_to_args(args, user_buffer);
|
||||
free(user_buffer);
|
||||
if (ret != 0) {
|
||||
ret = BUFFER_TOO_SMALL;
|
||||
goto free_and_exit;
|
||||
if (!service_mode_enabled) {
|
||||
char *user_buffer = make_string("--user=%s", user);
|
||||
ret = add_to_args(args, user_buffer);
|
||||
free(user_buffer);
|
||||
if (ret != 0) {
|
||||
ret = BUFFER_TOO_SMALL;
|
||||
goto free_and_exit;
|
||||
}
|
||||
}
|
||||
no_new_privileges_enabled =
|
||||
get_configuration_value("docker.no-new-privileges.enabled",
|
||||
@ -1725,9 +1756,11 @@ int get_docker_run_command(const char *command_file, const struct configuration
|
||||
goto free_and_exit;
|
||||
}
|
||||
|
||||
ret = set_group_add(&command_config, args);
|
||||
if (ret != 0) {
|
||||
goto free_and_exit;
|
||||
if (!service_mode_enabled) {
|
||||
ret = set_group_add(&command_config, args);
|
||||
if (ret != 0) {
|
||||
goto free_and_exit;
|
||||
}
|
||||
}
|
||||
|
||||
ret = set_devices(&command_config, conf, args);
|
||||
|
||||
@ -36,6 +36,7 @@
|
||||
#define DOCKER_START_COMMAND "start"
|
||||
#define DOCKER_EXEC_COMMAND "exec"
|
||||
#define DOCKER_IMAGES_COMMAND "images"
|
||||
#define DOCKER_SERVICE_MODE_ENABLED_KEY "docker.service-mode.enabled"
|
||||
#define DOCKER_ARG_MAX 1024
|
||||
#define ARGS_INITIAL_VALUE { 0 };
|
||||
|
||||
@ -71,7 +72,8 @@ enum docker_error_codes {
|
||||
INVALID_PID_NAMESPACE,
|
||||
INVALID_DOCKER_IMAGE_TRUST,
|
||||
INVALID_DOCKER_TMPFS_MOUNT,
|
||||
INVALID_DOCKER_RUNTIME
|
||||
INVALID_DOCKER_RUNTIME,
|
||||
SERVICE_MODE_DISABLED
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@ -285,6 +285,7 @@ are allowed. It contains the following properties:
|
||||
| `docker.inspect.max.retries` | Integer value to check docker container readiness. Each inspection is set with 3 seconds delay. Default value of 10 will wait 30 seconds for docker container to become ready before marked as container failed. |
|
||||
| `docker.no-new-privileges.enabled` | Enable/disable the no-new-privileges flag for docker run. Set to "true" to enable, disabled by default. |
|
||||
| `docker.allowed.runtimes` | Comma seperated runtimes that containers are allowed to use. By default no runtimes are allowed to be added.|
|
||||
| `docker.service-mode.enabled` | Set to "true" or "false" to enable or disable docker container service mode. Default value is "false". |
|
||||
|
||||
Please note that if you wish to run Docker containers that require access to the YARN local directories, you must add them to the docker.allowed.rw-mounts list.
|
||||
|
||||
@ -436,6 +437,7 @@ environment variables in the application's environment:
|
||||
| `YARN_CONTAINER_RUNTIME_DOCKER_TMPFS_MOUNTS` | Adds additional tmpfs mounts to the Docker container. The value of the environment variable should be a comma-separated list of absolute mount points within the container. |
|
||||
| `YARN_CONTAINER_RUNTIME_DOCKER_DELAYED_REMOVAL` | Allows a user to request delayed deletion of the Docker container on a per container basis. If true, Docker containers will not be removed until the duration defined by yarn.nodemanager.delete.debug-delay-sec has elapsed. Administrators can disable this feature through the yarn-site property yarn.nodemanager.runtime.linux.docker.delayed-removal.allowed. This feature is disabled by default. When this feature is disabled or set to false, the container will be removed as soon as it exits. |
|
||||
| `YARN_CONTAINER_RUNTIME_YARN_SYSFS_ENABLE` | Enable mounting of container working directory sysfs sub-directory into Docker container /hadoop/yarn/sysfs. This is useful for populating cluster information into container. |
|
||||
| `YARN_CONTAINER_RUNTIME_DOCKER_SERVICE_MODE` | Enable Service Mode which runs the docker container as defined by the image but does not set the user (--user and --group-add). |
|
||||
|
||||
The first two are required. The remainder can be set as needed. While
|
||||
controlling the container type through environment variables is somewhat less
|
||||
@ -1080,3 +1082,24 @@ YARN service framework automatically populates cluster information
|
||||
to /hadoop/yarn/sysfs/app.json. For more information about
|
||||
YARN service, see: [YARN Service](./yarn-service/Overview.html).
|
||||
|
||||
Docker Container Service Mode
|
||||
-----------------------------
|
||||
|
||||
Docker Container Service Mode runs the container as defined by the image
|
||||
but does not set the user (--user and --group-add). This mode is disabled
|
||||
by default. The administrator sets docker.service-mode.enabled to true
|
||||
in container-executor.cfg under docker section to enable.
|
||||
|
||||
Part of a container-executor.cfg which allows docker service mode is below:
|
||||
|
||||
```
|
||||
yarn.nodemanager.linux-container-executor.group=yarn
|
||||
[docker]
|
||||
module.enabled=true
|
||||
docker.privileged-containers.enabled=true
|
||||
docker.service-mode.enabled=true
|
||||
```
|
||||
|
||||
Application User can enable or disable service mode at job level by exporting
|
||||
environment variable YARN_CONTAINER_RUNTIME_DOCKER_SERVICE_MODE in the application's
|
||||
environment with value true or false respectively.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user