YARN-5505. Create an agent-less docker provider in the native-services framework. Contributed by Billie Rinaldi

This commit is contained in:
Jian He 2016-09-01 22:38:42 +08:00
parent 3279baecb9
commit bce06ed1af
23 changed files with 1996 additions and 965 deletions

View File

@ -41,7 +41,20 @@ public interface OptionKeys extends InternalKeys {
* Prefix for site.xml options: {@value}
*/
String SITE_XML_PREFIX = "site.";
/**
* Prefix for config file options: {@value}
*/
String CONF_FILE_PREFIX = "conf.";
/**
* Prefix for package options: {@value}
*/
String PKG_FILE_PREFIX = "pkg.";
/**
* Prefix for export options: {@value}
*/
String EXPORT_PREFIX = "export.";
String TYPE_SUFFIX = ".type";
String NAME_SUFFIX = ".name";
/**
* Zookeeper quorum host list: {@value}

View File

@ -151,7 +151,6 @@ import org.apache.slider.core.registry.YarnAppListClient;
import org.apache.slider.core.registry.docstore.ConfigFormat;
import org.apache.slider.core.registry.docstore.PublishedConfigSet;
import org.apache.slider.core.registry.docstore.PublishedConfiguration;
import org.apache.slider.core.registry.docstore.PublishedConfigurationOutputter;
import org.apache.slider.core.registry.docstore.PublishedExports;
import org.apache.slider.core.registry.docstore.PublishedExportsOutputter;
import org.apache.slider.core.registry.docstore.PublishedExportsSet;
@ -162,6 +161,7 @@ import org.apache.slider.core.zk.ZKPathBuilder;
import org.apache.slider.providers.AbstractClientProvider;
import org.apache.slider.providers.SliderProviderFactory;
import org.apache.slider.providers.agent.AgentKeys;
import org.apache.slider.providers.docker.DockerClientProvider;
import org.apache.slider.providers.slideram.SliderAMClientProvider;
import org.apache.slider.server.appmaster.SliderAppMaster;
import org.apache.slider.server.appmaster.rpc.RpcBinder;
@ -2081,7 +2081,7 @@ public class SliderClient extends AbstractSliderLaunchedService implements RunSe
// add the tags if available
Set<String> applicationTags = provider.getApplicationTags(sliderFileSystem,
getApplicationDefinitionPath(appOperations));
appOperations);
Credentials credentials = null;
if (clusterSecure) {
@ -2242,11 +2242,12 @@ public class SliderClient extends AbstractSliderLaunchedService implements RunSe
);
// TODO: consider supporting apps that don't have an image path
Path imagePath =
extractImagePath(sliderFileSystem, internalOptions);
if (sliderFileSystem.maybeAddImagePath(localResources, imagePath)) {
log.debug("Registered image path {}", imagePath);
if (!(provider instanceof DockerClientProvider)) {
Path imagePath =
extractImagePath(sliderFileSystem, internalOptions);
if (sliderFileSystem.maybeAddImagePath(localResources, imagePath)) {
log.debug("Registered image path {}", imagePath);
}
}
// build the environment
@ -3814,7 +3815,7 @@ public class SliderClient extends AbstractSliderLaunchedService implements RunSe
Path subPath = new Path(path1, appReport.getApplicationId()
.toString() + "/agent");
imagePath = subPath.toString();
String pathStr = imagePath + "/" + AGENT_TAR;
String pathStr = imagePath + "/" + AgentKeys.AGENT_TAR;
try {
validateHDFSFile(sliderFileSystem, pathStr);
log.info("Slider agent package is properly installed at " + pathStr);

View File

@ -81,6 +81,10 @@ public interface SliderKeys extends SliderXmlConfKeys {
String COMPONENT_SEPARATOR = "-";
String[] COMPONENT_KEYS_TO_SKIP = {"zookeeper.", "env.MALLOC_ARENA_MAX",
"site.fs.", "site.dfs."};
/**
* A component type for a client component
*/
String COMPONENT_TYPE_CLIENT = "client";
/**
* Key for application version. This must be set in app_config/global {@value}
@ -222,7 +226,6 @@ public interface SliderKeys extends SliderXmlConfKeys {
String SLIDER_JAR = "slider.jar";
String JCOMMANDER_JAR = "jcommander.jar";
String GSON_JAR = "gson.jar";
String AGENT_TAR = "slider-agent.tar.gz";
String DEFAULT_APP_PKG = "appPkg.zip";
String DEFAULT_JVM_HEAP = "256M";
@ -288,4 +291,21 @@ public interface SliderKeys extends SliderXmlConfKeys {
String SLIDER_CLASSPATH_EXTRA = "SLIDER_CLASSPATH_EXTRA";
String YARN_CONTAINER_PATH = "/node/container/";
String GLOBAL_CONFIG_TAG = "global";
String SYSTEM_CONFIGS = "system_configs";
String JAVA_HOME = "java_home";
String TWO_WAY_SSL_ENABLED = "ssl.server.client.auth";
String INFRA_RUN_SECURITY_DIR = "infra/run/security/";
String CERT_FILE_LOCALIZATION_PATH = INFRA_RUN_SECURITY_DIR + "ca.crt";
String AM_CONFIG_GENERATION = "am.config.generation";
String APP_CONF_DIR = "app/conf";
String APP_RESOURCES = "application.resources";
String APP_RESOURCES_DIR = "app/resources";
String PER_COMPONENT = "per.component";
String PER_GROUP = "per.group";
String APP_PACKAGES_DIR = "app/packages";
}

View File

@ -183,6 +183,10 @@ public final class SliderUtils {
return !isUnset(s);
}
public static boolean isEmpty(List l) {
return l == null || l.isEmpty();
}
/**
* Probe for a list existing and not being empty
* @param l list

View File

@ -52,6 +52,8 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import static org.apache.slider.providers.docker.DockerKeys.DEFAULT_DOCKER_NETWORK;
/**
* Launcher of applications: base class
*/
@ -79,6 +81,7 @@ public abstract class AbstractLauncher extends Configured {
protected LogAggregationContext logAggregationContext;
protected boolean yarnDockerMode = false;
protected String dockerImage;
protected String dockerNetwork = DEFAULT_DOCKER_NETWORK;
protected String yarnContainerMountPoints;
protected String runPrivilegedContainer;
@ -232,7 +235,8 @@ public abstract class AbstractLauncher extends Configured {
if(yarnDockerMode){
Map<String, String> env = containerLaunchContext.getEnvironment();
env.put("YARN_CONTAINER_RUNTIME_TYPE", "docker");
env.put("YARN_CONTAINER_RUNTIME_DOCKER_IMAGE", dockerImage);//if yarnDockerMode, then dockerImage is set
env.put("YARN_CONTAINER_RUNTIME_DOCKER_IMAGE", dockerImage);
env.put("YARN_CONTAINER_RUNTIME_DOCKER_CONTAINER_NETWORK", dockerNetwork);
env.put("YARN_CONTAINER_RUNTIME_DOCKER_RUN_PRIVILEGED_CONTAINER", runPrivilegedContainer);
StringBuilder sb = new StringBuilder();
for (Entry<String,String> mount : mountPaths.entrySet()) {
@ -517,6 +521,10 @@ public abstract class AbstractLauncher extends Configured {
this.dockerImage = dockerImage;
}
public void setDockerNetwork(String dockerNetwork) {
this.dockerNetwork = dockerNetwork;
}
public void setYarnContainerMountPoints(String yarnContainerMountPoints) {
this.yarnContainerMountPoints = yarnContainerMountPoints;
}
@ -525,4 +533,12 @@ public abstract class AbstractLauncher extends Configured {
this.runPrivilegedContainer = runPrivilegedContainer;
}
public void setRunPrivilegedContainer(boolean runPrivilegedContainer) {
if (runPrivilegedContainer) {
this.runPrivilegedContainer = Boolean.toString(true);
} else {
this.runPrivilegedContainer = Boolean.toString(false);
}
}
}

View File

@ -39,6 +39,8 @@ import java.util.Properties;
*/
public abstract class PublishedConfigurationOutputter {
private static final String COMMENTS = "Generated by Apache Slider";
protected final PublishedConfiguration owner;
protected PublishedConfigurationOutputter(PublishedConfiguration owner) {
@ -143,13 +145,13 @@ public abstract class PublishedConfigurationOutputter {
@Override
public void save(OutputStream out) throws IOException {
properties.store(out, "");
properties.store(out, COMMENTS);
}
public String asString() throws IOException {
StringWriter sw = new StringWriter();
properties.store(sw, "");
properties.store(sw, COMMENTS);
return sw.toString();
}
}

View File

@ -216,8 +216,8 @@ public abstract class AbstractClientProvider extends Configured {
* Return a set of application specific string tags.
* @return the set of tags.
*/
public Set<String> getApplicationTags (SliderFileSystem fileSystem,
String appDef) throws SliderException {
public Set<String> getApplicationTags(SliderFileSystem fileSystem,
ConfTreeOperations appConf) throws SliderException {
return Collections.emptySet();
}

View File

@ -22,6 +22,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.registry.client.binding.RegistryTypeUtils;
@ -138,6 +139,19 @@ public abstract class AbstractProviderService
this.restOps = agentRestOperations;
}
/**
* Load default Configuration
* @param confDir configuration directory
* @return configuration
* @throws BadCommandArgumentsException
* @throws IOException
*/
@Override
public Configuration loadProviderConfigurationInformation(File confDir)
throws BadCommandArgumentsException, IOException {
return new Configuration(false);
}
/**
* Load a specific XML configuration file for the provider config
* @param confDir configuration directory
@ -369,8 +383,6 @@ public abstract class AbstractProviderService
@Override
public void applyInitialRegistryDefinitions(URL amWebURI,
URL agentOpsURI,
URL agentStatusURI,
ServiceRecord serviceRecord)
throws IOException {
this.amWebAPI = amWebURI;
@ -422,4 +434,10 @@ public abstract class AbstractProviderService
public void rebuildContainerDetails(List<Container> liveContainers,
String applicationId, Map<Integer, ProviderRole> providerRoles) {
}
@Override
public boolean processContainerStatus(ContainerId containerId,
ContainerStatus status) {
return false;
}
}

View File

@ -24,6 +24,7 @@ import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.registry.client.types.ServiceRecord;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.slider.api.ClusterDescription;
import org.apache.slider.common.tools.SliderFileSystem;
import org.apache.slider.core.conf.AggregateConf;
@ -189,13 +190,9 @@ public interface ProviderService extends ProviderCore,
/**
* Prior to going live -register the initial service registry data
* @param amWebURI URL to the AM. This may be proxied, so use relative paths
* @param agentOpsURI URI for agent operations. This will not be proxied
* @param agentStatusURI URI For agent status. Again: no proxy
* @param serviceRecord service record to build up
*/
void applyInitialRegistryDefinitions(URL amWebURI,
URL agentOpsURI,
URL agentStatusURI,
ServiceRecord serviceRecord)
throws IOException;
@ -216,4 +213,11 @@ public interface ProviderService extends ProviderCore,
*/
void rebuildContainerDetails(List<Container> liveContainers,
String applicationId, Map<Integer, ProviderRole> providerRoles);
/**
* Process container status
* @return true if status needs to be requested again, false otherwise
*/
boolean processContainerStatus(ContainerId containerId,
ContainerStatus status);
}

View File

@ -82,6 +82,8 @@ import java.util.Set;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;
import static org.apache.slider.common.tools.SliderUtils.getApplicationDefinitionPath;
/** This class implements the client-side aspects of the agent deployer */
public class AgentClientProvider extends AbstractClientProvider
implements AgentKeys, SliderKeys {
@ -132,13 +134,13 @@ public class AgentClientProvider extends AbstractClientProvider
sliderFileSystem.verifyFileExists(appDefPath);
String agentConf = instanceDefinition.getAppConfOperations().
getGlobalOptions().getOption(AgentKeys.AGENT_CONF, "");
getGlobalOptions().getOption(AGENT_CONF, "");
if (StringUtils.isNotEmpty(agentConf)) {
sliderFileSystem.verifyFileExists(new Path(agentConf));
}
String appHome = instanceDefinition.getAppConfOperations().
getGlobalOptions().get(AgentKeys.PACKAGE_PATH);
getGlobalOptions().get(PACKAGE_PATH);
if (SliderUtils.isUnset(appHome)) {
String agentImage = instanceDefinition.getInternalOperations().
get(InternalKeys.INTERNAL_APPLICATION_IMAGE_PATH);
@ -173,7 +175,7 @@ public class AgentClientProvider extends AbstractClientProvider
}
Set<String> names = resources.getComponentNames();
names.remove(SliderKeys.COMPONENT_AM);
names.remove(COMPONENT_AM);
Map<Integer, String> priorityMap = new HashMap<Integer, String>();
for (String name : names) {
@ -271,7 +273,7 @@ public class AgentClientProvider extends AbstractClientProvider
String agentImage = instanceDefinition.getInternalOperations().
get(InternalKeys.INTERNAL_APPLICATION_IMAGE_PATH);
if (SliderUtils.isUnset(agentImage)) {
Path agentPath = new Path(tempPath.getParent(), AgentKeys.PROVIDER_AGENT);
Path agentPath = new Path(tempPath.getParent(), PROVIDER_AGENT);
log.info("Automatically uploading the agent tarball at {}", agentPath);
fileSystem.getFileSystem().mkdirs(agentPath);
if (ProviderUtils.addAgentTar(this, AGENT_TAR, fileSystem, agentPath)) {
@ -283,6 +285,12 @@ public class AgentClientProvider extends AbstractClientProvider
}
@Override
public Set<String> getApplicationTags(SliderFileSystem fileSystem,
ConfTreeOperations appConf) throws SliderException {
return getApplicationTags(fileSystem,
getApplicationDefinitionPath(appConf));
}
public Set<String> getApplicationTags(SliderFileSystem fileSystem,
String appDef) throws SliderException {
Set<String> tags;
@ -437,19 +445,19 @@ public class AgentClientProvider extends AbstractClientProvider
if (config != null) {
try {
clientRoot = config.getJSONObject("global")
.getString(AgentKeys.APP_CLIENT_ROOT);
.getString(APP_CLIENT_ROOT);
} catch (JSONException e) {
log.info("Couldn't read {} from provided client config, falling " +
"back on default", AgentKeys.APP_CLIENT_ROOT);
"back on default", APP_CLIENT_ROOT);
}
}
if (clientRoot == null && defaultConfig != null) {
try {
clientRoot = defaultConfig.getJSONObject("global")
.getString(AgentKeys.APP_CLIENT_ROOT);
.getString(APP_CLIENT_ROOT);
} catch (JSONException e) {
log.info("Couldn't read {} from default client config, using {}",
AgentKeys.APP_CLIENT_ROOT, clientInstallPath);
APP_CLIENT_ROOT, clientInstallPath);
}
}
if (clientRoot == null) {
@ -500,7 +508,7 @@ public class AgentClientProvider extends AbstractClientProvider
try {
String clientScriptPath = appPkgDir.getAbsolutePath() + File.separator + "package" +
File.separator + clientScript;
List<String> command = Arrays.asList(AgentKeys.PYTHON_EXE,
List<String> command = Arrays.asList(PYTHON_EXE,
"-S",
clientScriptPath,
"INSTALL",
@ -510,12 +518,12 @@ public class AgentClientProvider extends AbstractClientProvider
"DEBUG");
ProcessBuilder pb = new ProcessBuilder(command);
log.info("Command: " + StringUtils.join(pb.command(), " "));
pb.environment().put(SliderKeys.PYTHONPATH,
pb.environment().put(PYTHONPATH,
agentPkgDir.getAbsolutePath()
+ File.separator + "slider-agent" + File.pathSeparator
+ agentPkgDir.getAbsolutePath()
+ File.separator + "slider-agent/jinja2");
log.info("{}={}", SliderKeys.PYTHONPATH, pb.environment().get(SliderKeys.PYTHONPATH));
log.info("{}={}", PYTHONPATH, pb.environment().get(PYTHONPATH));
Process proc = pb.start();
InputStream stderr = proc.getErrorStream();
@ -555,8 +563,8 @@ public class AgentClientProvider extends AbstractClientProvider
private void expandAgentTar(File agentPkgDir) throws IOException {
String libDirProp =
System.getProperty(SliderKeys.PROPERTY_LIB_DIR);
File tarFile = new File(libDirProp, SliderKeys.AGENT_TAR);
System.getProperty(PROPERTY_LIB_DIR);
File tarFile = new File(libDirProp, AGENT_TAR);
expandTar(tarFile, agentPkgDir);
}
@ -620,7 +628,7 @@ public class AgentClientProvider extends AbstractClientProvider
String name) throws SliderException {
try {
JSONObject pkgList = new JSONObject();
pkgList.put(AgentKeys.PACKAGE_LIST,
pkgList.put(PACKAGE_LIST,
AgentProviderService.getPackageListFromApplication(metainfo.getApplication()));
JSONObject obj = new JSONObject();
obj.put("hostLevelParams", pkgList);

View File

@ -23,6 +23,7 @@ package org.apache.slider.providers.agent;
*/
public interface AgentKeys {
String AGENT_TAR = "slider-agent.tar.gz";
String PROVIDER_AGENT = "agent";
String ROLE_NODE = "echo";
@ -76,23 +77,13 @@ public interface AgentKeys {
String AGENT_CONF = "agent.conf";
String ADDON_FOR_ALL_COMPONENTS = "ALL";
String APP_RESOURCES = "application.resources";
String APP_RESOURCES_DIR = "app/resources";
String APP_CONF_DIR = "app/conf";
String AGENT_INSTALL_DIR = "infra/agent";
String APP_DEFINITION_DIR = "app/definition";
String ADDON_DEFINITION_DIR = "addon/definition";
String AGENT_CONFIG_FILE = "infra/conf/agent.ini";
String AGENT_VERSION_FILE = "infra/version";
String APP_PACKAGES_DIR = "app/packages";
String PER_COMPONENT = "per.component";
String PER_GROUP = "per.group";
String JAVA_HOME = "java_home";
String PACKAGE_LIST = "package_list";
String SYSTEM_CONFIGS = "system_configs";
String WAIT_HEARTBEAT = "wait.heartbeat";
String PYTHON_EXE = "python";
String CREATE_DEF_ZK_NODE = "create.default.zookeeper.node";
@ -104,7 +95,6 @@ public interface AgentKeys {
String CERT_FILE_LOCALIZATION_PATH = INFRA_RUN_SECURITY_DIR + "ca.crt";
String KEY_CONTAINER_LAUNCH_DELAY = "container.launch.delay.sec";
String TEST_RELAX_VERIFICATION = "test.relax.validation";
String AM_CONFIG_GENERATION = "am.config.generation";
String DEFAULT_METAINFO_MAP_KEY = "DEFAULT_KEY";
}

View File

@ -0,0 +1,96 @@
/*
* 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.slider.providers.docker;
import org.apache.hadoop.conf.Configuration;
import org.apache.slider.common.SliderKeys;
import org.apache.slider.common.tools.SliderFileSystem;
import org.apache.slider.core.conf.AggregateConf;
import org.apache.slider.core.conf.ConfTreeOperations;
import org.apache.slider.core.exceptions.BadConfigException;
import org.apache.slider.core.exceptions.SliderException;
import org.apache.slider.providers.AbstractClientProvider;
import org.apache.slider.providers.ProviderRole;
import org.apache.slider.providers.ProviderUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import static org.apache.slider.providers.docker.DockerKeys.DOCKER_IMAGE;
public class DockerClientProvider extends AbstractClientProvider
implements SliderKeys {
protected static final Logger log =
LoggerFactory.getLogger(DockerClientProvider.class);
private static final ProviderUtils providerUtils = new ProviderUtils(log);
protected static final String NAME = "docker";
public DockerClientProvider(Configuration conf) {
super(conf);
}
@Override
public String getName() {
return NAME;
}
@Override
public List<ProviderRole> getRoles() {
return Collections.emptyList();
}
@Override
public void validateInstanceDefinition(AggregateConf instanceDefinition,
SliderFileSystem fs) throws SliderException {
super.validateInstanceDefinition(instanceDefinition, fs);
ConfTreeOperations appConf = instanceDefinition.getAppConfOperations();
ConfTreeOperations resources = instanceDefinition.getResourceOperations();
for (String roleGroup : resources.getComponentNames()) {
if (roleGroup.equals(COMPONENT_AM)) {
continue;
}
if (appConf.getComponentOpt(roleGroup, DOCKER_IMAGE, null) == null &&
appConf.getGlobalOptions().get(DOCKER_IMAGE) == null) {
throw new BadConfigException("Property " + DOCKER_IMAGE + " not " +
"specified for " + roleGroup);
}
providerUtils.getPackages(roleGroup, appConf);
if (appConf.getComponentOptBool(roleGroup, AM_CONFIG_GENERATION, false)) {
// build and localize configuration files
Map<String, Map<String, String>> configurations =
providerUtils.buildConfigurations(appConf, appConf, null, roleGroup,
roleGroup, null);
try {
providerUtils.localizeConfigFiles(null, roleGroup, roleGroup, appConf,
configurations, null, fs, null);
} catch (IOException e) {
throw new BadConfigException(e.toString());
}
}
}
}
}

View File

@ -0,0 +1,32 @@
/*
* 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.slider.providers.docker;
public interface DockerKeys {
String PROVIDER_DOCKER = "docker";
String DOCKER_PREFIX = "docker.";
String DOCKER_IMAGE = DOCKER_PREFIX + "image";
String DOCKER_NETWORK = DOCKER_PREFIX + "network";
String DOCKER_USE_PRIVILEGED = DOCKER_PREFIX + "usePrivileged";
String DOCKER_START_COMMAND = DOCKER_PREFIX + "startCommand";
String DEFAULT_DOCKER_NETWORK = "bridge";
String OUT_FILE = "stdout.txt";
String ERR_FILE = "stderr.txt";
}

View File

@ -0,0 +1,43 @@
/*
* 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.slider.providers.docker;
import org.apache.hadoop.conf.Configuration;
import org.apache.slider.providers.AbstractClientProvider;
import org.apache.slider.providers.ProviderService;
import org.apache.slider.providers.SliderProviderFactory;
public class DockerProviderFactory extends SliderProviderFactory {
public DockerProviderFactory() {
}
public DockerProviderFactory(Configuration conf) {
super(conf);
}
@Override
public AbstractClientProvider createClientProvider() {
return new DockerClientProvider(getConf());
}
@Override
public ProviderService createServerProvider() {
return new DockerProviderService();
}
}

View File

@ -0,0 +1,355 @@
/*
* 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.slider.providers.docker;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.registry.client.types.ServiceRecord;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.slider.api.ClusterDescription;
import org.apache.slider.api.ClusterNode;
import org.apache.slider.api.OptionKeys;
import org.apache.slider.common.SliderKeys;
import org.apache.slider.common.tools.SliderFileSystem;
import org.apache.slider.common.tools.SliderUtils;
import org.apache.slider.core.conf.AggregateConf;
import org.apache.slider.core.conf.ConfTreeOperations;
import org.apache.slider.core.conf.MapOperations;
import org.apache.slider.core.exceptions.SliderException;
import org.apache.slider.core.launch.CommandLineBuilder;
import org.apache.slider.core.launch.ContainerLauncher;
import org.apache.slider.core.registry.docstore.ConfigFormat;
import org.apache.slider.core.registry.docstore.ConfigUtils;
import org.apache.slider.core.registry.docstore.ExportEntry;
import org.apache.slider.providers.AbstractProviderService;
import org.apache.slider.providers.ProviderCore;
import org.apache.slider.providers.ProviderRole;
import org.apache.slider.providers.ProviderUtils;
import org.apache.slider.server.appmaster.state.RoleInstance;
import org.apache.slider.server.appmaster.state.StateAccessForProviders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Scanner;
public class DockerProviderService extends AbstractProviderService implements
ProviderCore,
DockerKeys,
SliderKeys {
protected static final Logger log =
LoggerFactory.getLogger(DockerProviderService.class);
private static final ProviderUtils providerUtils = new ProviderUtils(log);
private static final String EXPORT_GROUP = "quicklinks";
private static final String APPLICATION_TAG = "application";
private String clusterName = null;
private SliderFileSystem fileSystem = null;
protected DockerProviderService() {
super("DockerProviderService");
}
@Override
public List<ProviderRole> getRoles() {
return Collections.emptyList();
}
@Override
public boolean isSupportedRole(String role) {
return true;
}
@Override
public void validateInstanceDefinition(AggregateConf instanceDefinition)
throws SliderException {
}
private String getClusterName() {
if (SliderUtils.isUnset(clusterName)) {
clusterName = getAmState().getInternalsSnapshot().get(OptionKeys.APPLICATION_NAME);
}
return clusterName;
}
@Override
public void buildContainerLaunchContext(ContainerLauncher launcher,
AggregateConf instanceDefinition, Container container,
ProviderRole providerRole, SliderFileSystem fileSystem,
Path generatedConfPath, MapOperations resourceComponent,
MapOperations appComponent, Path containerTmpDirPath)
throws IOException, SliderException {
String roleName = providerRole.name;
String roleGroup = providerRole.group;
initializeApplicationConfiguration(instanceDefinition, fileSystem,
roleGroup);
log.info("Build launch context for Docker");
log.debug(instanceDefinition.toString());
ConfTreeOperations appConf = instanceDefinition.getAppConfOperations();
launcher.setYarnDockerMode(true);
launcher.setDockerImage(appConf.getComponentOpt(roleGroup, DOCKER_IMAGE,
null));
launcher.setDockerNetwork(appConf.getComponentOpt(roleGroup, DOCKER_NETWORK,
DEFAULT_DOCKER_NETWORK));
launcher.setRunPrivilegedContainer(appConf.getComponentOptBool(roleGroup,
DOCKER_USE_PRIVILEGED, false));
// Set the environment
launcher.putEnv(SliderUtils.buildEnvMap(appComponent,
providerUtils.getStandardTokenMap(getAmState().getAppConfSnapshot(),
getAmState().getInternalsSnapshot(), roleName, roleGroup,
getClusterName())));
String workDir = ApplicationConstants.Environment.PWD.$();
launcher.setEnv("WORK_DIR", workDir);
log.info("WORK_DIR set to {}", workDir);
String logDir = ApplicationConstants.LOG_DIR_EXPANSION_VAR;
launcher.setEnv("LOG_DIR", logDir);
log.info("LOG_DIR set to {}", logDir);
if (System.getenv(HADOOP_USER_NAME) != null) {
launcher.setEnv(HADOOP_USER_NAME, System.getenv(HADOOP_USER_NAME));
}
//add english env
launcher.setEnv("LANG", "en_US.UTF-8");
launcher.setEnv("LC_ALL", "en_US.UTF-8");
launcher.setEnv("LANGUAGE", "en_US.UTF-8");
//local resources
providerUtils.localizePackages(launcher, fileSystem, appConf, roleGroup,
getClusterName());
if (SliderUtils.isHadoopClusterSecure(getConfig())) {
providerUtils.localizeServiceKeytabs(launcher, instanceDefinition,
fileSystem, getClusterName());
}
if (providerUtils.areStoresRequested(appComponent)) {
providerUtils.localizeContainerSecurityStores(launcher, container,
roleName, fileSystem, instanceDefinition, appComponent, getClusterName());
}
if (appComponent.getOptionBool(AM_CONFIG_GENERATION, false)) {
// build and localize configuration files
Map<String, Map<String, String>> configurations =
providerUtils.buildConfigurations(
instanceDefinition.getAppConfOperations(),
instanceDefinition.getInternalOperations(),
container.getId().toString(), roleName, roleGroup,
getAmState());
providerUtils.localizeConfigFiles(launcher, roleName, roleGroup,
appConf, configurations, launcher.getEnv(), fileSystem,
getClusterName());
}
//add the configuration resources
launcher.addLocalResources(fileSystem.submitDirectory(
generatedConfPath,
PROPAGATED_CONF_DIR_NAME));
CommandLineBuilder operation = new CommandLineBuilder();
operation.add(appConf.getComponentOpt(roleGroup, DOCKER_START_COMMAND,
"/bin/bash"));
operation.add("> " + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/"
+ OUT_FILE + " 2>" + ERR_FILE);
launcher.addCommand(operation.build());
// Additional files to localize
String appResourcesString = instanceDefinition.getAppConfOperations()
.getGlobalOptions().getOption(APP_RESOURCES, null);
log.info("Configuration value for extra resources to localize: {}", appResourcesString);
if (null != appResourcesString) {
try (Scanner scanner = new Scanner(appResourcesString).useDelimiter(",")) {
while (scanner.hasNext()) {
String resource = scanner.next();
Path resourcePath = new Path(resource);
LocalResource extraResource = fileSystem.createAmResource(
fileSystem.getFileSystem().resolvePath(resourcePath),
LocalResourceType.FILE);
String destination = APP_RESOURCES_DIR + "/" + resourcePath.getName();
log.info("Localizing {} to {}", resourcePath, destination);
// TODO Can we try harder to avoid collisions?
launcher.addLocalResource(destination, extraResource);
}
}
}
}
@Override
public void initializeApplicationConfiguration(
AggregateConf instanceDefinition, SliderFileSystem fileSystem,
String roleGroup)
throws IOException, SliderException {
this.fileSystem = fileSystem;
}
@Override
public void applyInitialRegistryDefinitions(URL amWebURI,
ServiceRecord serviceRecord)
throws IOException {
super.applyInitialRegistryDefinitions(amWebURI, serviceRecord);
// identify client component
String clientName = null;
ConfTreeOperations appConf = getAmState().getAppConfSnapshot();
for (String component : appConf.getComponentNames()) {
if (COMPONENT_TYPE_CLIENT.equals(appConf.getComponentOpt(component,
COMPONENT_TYPE_KEY, null))) {
clientName = component;
break;
}
}
if (clientName == null) {
log.info("No client component specified, not publishing client configs");
return;
}
// register AM-generated client configs
// appConf should already be resolved!
MapOperations clientOperations = appConf.getComponent(clientName);
if (!clientOperations.getOptionBool(AM_CONFIG_GENERATION, false)) {
log.info("AM config generation is false, not publishing client configs");
return;
}
// build and localize configuration files
Map<String, Map<String, String>> configurations =
providerUtils.buildConfigurations(appConf, getAmState()
.getInternalsSnapshot(), null, clientName, clientName,
getAmState());
for (String configFileDN : configurations.keySet()) {
String configFileName = appConf.getComponentOpt(clientName,
OptionKeys.CONF_FILE_PREFIX + configFileDN + OptionKeys
.NAME_SUFFIX, null);
String configFileType = appConf.getComponentOpt(clientName,
OptionKeys.CONF_FILE_PREFIX + configFileDN + OptionKeys
.TYPE_SUFFIX, null);
if (configFileName == null && configFileType == null) {
continue;
}
ConfigFormat configFormat = ConfigFormat.resolve(configFileType);
Map<String, String> config = configurations.get(configFileDN);
ConfigUtils.prepConfigForTemplateOutputter(configFormat, config,
fileSystem, getClusterName(),
new File(configFileName).getName());
providerUtils.publishApplicationInstanceData(configFileDN, configFileDN,
config.entrySet(), getAmState());
}
}
@Override
public boolean processContainerStatus(ContainerId containerId,
ContainerStatus status) {
log.debug("Handling container status: {}", status);
if (SliderUtils.isEmpty(status.getIPs()) ||
SliderUtils.isUnset(status.getHost())) {
return true;
}
RoleInstance instance = getAmState().getOwnedContainer(containerId);
if (instance == null) {
// container is completed?
return false;
}
String roleName = instance.role;
String roleGroup = instance.group;
String containerIdStr = containerId.toString();
providerUtils.updateServiceRecord(getAmState(), yarnRegistry,
containerIdStr, roleName, status.getIPs(), status.getHost());
publishExportGroups(containerIdStr, roleName, roleGroup,
status.getHost());
return false;
}
/**
* This method looks for configuration properties of the form
* export.key,value and publishes the key,value pair. Standard tokens are
* substituted into the value, and COMPONENTNAME_HOST and THIS_HOST tokens
* are substituted with the actual hostnames of the containers.
*/
protected void publishExportGroups(String containerId,
String roleName, String roleGroup, String thisHost) {
ConfTreeOperations appConf = getAmState().getAppConfSnapshot();
ConfTreeOperations internalsConf = getAmState().getInternalsSnapshot();
Map<String, String> exports = providerUtils.getExports(
getAmState().getAppConfSnapshot(), roleGroup);
String hostKeyFormat = "${%s_HOST}";
// publish export groups if any
Map<String, String> replaceTokens =
providerUtils.filterSiteOptions(
appConf.getComponent(roleGroup).options,
providerUtils.getStandardTokenMap(appConf, internalsConf, roleName,
roleGroup, containerId, getClusterName()));
for (Map.Entry<String, Map<String, ClusterNode>> entry :
getAmState().getRoleClusterNodeMapping().entrySet()) {
String hostName = providerUtils.getHostsList(
entry.getValue().values(), true).iterator().next();
replaceTokens.put(String.format(hostKeyFormat,
entry.getKey().toUpperCase(Locale.ENGLISH)), hostName);
}
replaceTokens.put("${THIS_HOST}", thisHost);
Map<String, List<ExportEntry>> entries = new HashMap<>();
for (Entry<String, String> export : exports.entrySet()) {
String value = export.getValue();
// replace host names and site properties
for (String token : replaceTokens.keySet()) {
if (value.contains(token)) {
value = value.replace(token, replaceTokens.get(token));
}
}
ExportEntry entry = new ExportEntry();
entry.setLevel(APPLICATION_TAG);
entry.setValue(value);
entry.setUpdatedTime(new Date().toString());
// over-write, app exports are singletons
entries.put(export.getKey(), new ArrayList(Arrays.asList(entry)));
log.info("Preparing to publish. Key {} and Value {}",
export.getKey(), value);
}
providerUtils.publishExportGroup(entries, getAmState(), EXPORT_GROUP);
}
}

View File

@ -103,13 +103,9 @@ public class SliderAMProviderService extends AbstractProviderService implements
@Override
public void applyInitialRegistryDefinitions(URL amWebURI,
URL agentOpsURI,
URL agentStatusURI,
ServiceRecord serviceRecord)
throws IOException {
super.applyInitialRegistryDefinitions(amWebURI,
agentOpsURI,
agentStatusURI,
serviceRecord);
// now publish site.xml files
YarnConfiguration defaultYarnConfig = new YarnConfiguration();

View File

@ -784,8 +784,10 @@ public class SliderAppMaster extends AbstractSliderLaunchedService
contentCache);
initAMFilterOptions(serviceConf);
// start the agent web app
startAgentWebApp(appInformation, serviceConf, webAppApi);
if (providerService instanceof AgentProviderService) {
// start the agent web app
startAgentWebApp(appInformation, serviceConf, webAppApi);
}
int webAppPort = deployWebApplication(webAppApi);
String scheme = WebAppUtils.HTTP_PREFIX;
@ -1296,8 +1298,6 @@ public class SliderAppMaster extends AbstractSliderLaunchedService
// the registry is running, so register services
URL amWebURI = new URL(appMasterProxiedUrl);
URL agentOpsURI = new URL(agentOpsUrl);
URL agentStatusURI = new URL(agentStatusUrl);
//Give the provider restricted access to the state, registry
setupInitialRegistryPaths();
@ -1324,15 +1324,20 @@ public class SliderAppMaster extends AbstractSliderLaunchedService
// internal services
sliderAMProvider.applyInitialRegistryDefinitions(amWebURI,
agentOpsURI,
agentStatusURI,
serviceRecord);
// provider service dynamic definitions.
providerService.applyInitialRegistryDefinitions(amWebURI,
agentOpsURI,
agentStatusURI,
serviceRecord);
if (providerService instanceof AgentProviderService) {
URL agentOpsURI = new URL(agentOpsUrl);
URL agentStatusURI = new URL(agentStatusUrl);
((AgentProviderService)providerService).applyInitialRegistryDefinitions(
amWebURI,
agentOpsURI,
agentStatusURI,
serviceRecord);
} else {
providerService.applyInitialRegistryDefinitions(amWebURI, serviceRecord);
}
// set any provided attributes
setProvidedServiceRecordAttributes(
@ -2285,6 +2290,20 @@ public class SliderAppMaster extends AbstractSliderLaunchedService
ContainerStatus containerStatus) {
LOG_YARN.debug("Container Status: id={}, status={}", containerId,
containerStatus);
if (providerService.processContainerStatus(containerId, containerStatus)) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
RoleInstance cinfo = appState.getOwnedContainer(containerId);
if (cinfo != null) {
LOG_YARN.info("Re-requesting status for role {}, {}",
cinfo.role, containerId);
//trigger another async container status
nmClientAsync.getContainerStatusAsync(containerId,
cinfo.container.getNodeId());
}
}
}
@Override // NMClientAsync.CallbackHandler

View File

@ -27,4 +27,8 @@
<name>slider.provider.agent</name>
<value>org.apache.slider.providers.agent.AgentProviderFactory</value>
</property>
<property>
<name>slider.provider.docker</name>
<value>org.apache.slider.providers.docker.DockerProviderFactory</value>
</property>
</configuration>

View File

@ -0,0 +1,42 @@
{
"schema": "http://example.org/specification/v2.0.0",
"metadata": {},
"global": {
"am.config.generation": "true",
"component.unique.names": "true",
"export.app.monitor": "${COMPONENT1_HOST} : ${@//site/test-xml/xmlkey}",
"export.other.key": "exportvalue",
"docker.image": "docker.io/centos:centos6",
"docker.startCommand": "sleep 600",
"conf.test-json.type": "json",
"conf.test-json.name": "/tmp/test.json",
"conf.test-xml.type": "xml",
"conf.test-xml.name": "/tmp/test.xml",
"conf.test-properties.type": "properties",
"conf.test-properties.name": "/tmp/test.xml",
"conf.test-yaml.type": "yaml",
"conf.test-yaml.name": "/tmp/test.yaml",
"conf.test-env.type": "env",
"conf.test-env.name": "/tmp/testenv",
"conf.test-template.type": "template",
"conf.test-template.name": "/tmp/test.template",
"conf.test-hadoop-xml.type": "hadoop-xml",
"conf.test-hadoop-xml.name": "/tmp/test-hadoop.xml",
"site.test-json.jsonkey": "val1",
"site.test-xml.xmlkey": "val2",
"site.test-hadoop-xml.xmlkey": "val3",
"site.test-properties.propkey": "val4",
"site.test-yaml.yamlkey": "val5",
"site.test-env.content": "test ${envkey1} {{envkey2}} content",
"site.test-env.envkey1": "envval1",
"site.test-env.envkey2": "envval2",
"site.test-template.templatekey1": "templateval1",
"site.test-template.templatekey2": "templateval2"
},
"components": {
}
}

View File

@ -0,0 +1,16 @@
{
"schema": "http://example.org/specification/v2.0.0",
"metadata": {},
"global": {},
"components": {
"slider-appmaster": {
"yarn.memory": "384"
},
"COMPONENT": {
"yarn.role.priority": "1",
"yarn.component.instances": 2,
"yarn.memory": "512",
"yarn.vcores": "2"
}
}
}

View File

@ -0,0 +1,16 @@
# 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.
test ${templatekey1} {{templatekey2}} content