diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/api/ServiceApiConstants.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/api/ServiceApiConstants.java index 5f76f19..3533da5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/api/ServiceApiConstants.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/api/ServiceApiConstants.java @@ -36,6 +36,10 @@ String SERVICE_NAME_LC = $("SERVICE_NAME.lc"); + String USER = $("USER"); + + String DOMAIN = $("DOMAIN"); + // Constants for component String COMPONENT_NAME = $("COMPONENT_NAME"); @@ -47,4 +51,11 @@ String COMPONENT_ID = $("COMPONENT_ID"); String CONTAINER_ID = $("CONTAINER_ID"); + + // Constants for default cluster ZK + String CLUSTER_ZK_QUORUM = $("CLUSTER_ZK_QUORUM"); + String SERVICE_ZK_PATH = $("SERVICE_ZK_PATH"); + + // Constants for cluster fs + String SERVICE_HDFS_DIR = $("SERVICE_HDFS_DIR"); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/core/registry/docstore/ConfigFormat.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/core/registry/docstore/ConfigFormat.java index 723b975..081688b 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/core/registry/docstore/ConfigFormat.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/core/registry/docstore/ConfigFormat.java @@ -25,7 +25,7 @@ JSON("json"), PROPERTIES("properties"), XML("xml"), - HADOOP_XML("hadoop-xml"), + HADOOP_XML("hadoop_xml"), ENV("env"), TEMPLATE("template"), YAML("yaml"), diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderRole.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderRole.java index 9cc48e1..20cd81c 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderRole.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderRole.java @@ -20,10 +20,7 @@ import org.apache.slider.api.ResourceKeys; import org.apache.slider.api.resource.Component; -import org.apache.slider.server.appmaster.state.AppState; -import java.util.LinkedList; -import java.util.List; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicLong; @@ -44,7 +41,6 @@ public final String labelExpression; public final Component component; public AtomicLong componentIdCounter = null; - public AppState appState; public Queue failedInstanceName = new ConcurrentLinkedQueue(); public ProviderRole(String name, int id) { this(name, @@ -78,7 +74,7 @@ public ProviderRole(String name, nodeFailureThreshold, placementTimeoutSeconds, labelExpression, - new Component().name(name).numberOfContainers(0L), null); + new Component().name(name).numberOfContainers(0L)); } /** @@ -88,13 +84,13 @@ public ProviderRole(String name, * @param id ID. This becomes the YARN priority * @param policy placement policy * @param nodeFailureThreshold threshold for node failures (within a reset interval) -* after which a node failure is considered an app failure + * after which a node failure is considered an app failure * @param placementTimeoutSeconds for lax placement, timeout in seconds before * @param labelExpression label expression for requests; may be null */ public ProviderRole(String name, String group, int id, int policy, int nodeFailureThreshold, long placementTimeoutSeconds, - String labelExpression, Component component, AppState state) { + String labelExpression, Component component) { this.name = name; if (group == null) { this.group = name; @@ -110,7 +106,6 @@ public ProviderRole(String name, String group, int id, int policy, if(component.getUniqueComponentSupport()) { componentIdCounter = new AtomicLong(0); } - this.appState = state; } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderUtils.java index d384585..47d6bef 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderUtils.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/ProviderUtils.java @@ -32,7 +32,6 @@ import org.apache.hadoop.yarn.api.records.LocalResource; import org.apache.hadoop.yarn.api.records.LocalResourceType; import org.apache.slider.api.ClusterNode; -import org.apache.slider.api.OptionKeys; import org.apache.slider.api.ResourceKeys; import org.apache.slider.api.RoleKeys; import org.apache.slider.api.resource.Application; @@ -50,6 +49,7 @@ import org.apache.slider.core.registry.docstore.ConfigFormat; import org.apache.slider.core.registry.docstore.PublishedConfiguration; import org.apache.slider.core.registry.docstore.PublishedConfigurationOutputter; +import org.apache.slider.core.zk.ZKIntegration; import org.apache.slider.server.appmaster.state.RoleInstance; import org.apache.slider.server.appmaster.state.StateAccessForProviders; import org.apache.slider.server.services.yarnregistry.YarnRegistryViewForProviders; @@ -59,7 +59,6 @@ import java.io.FileNotFoundException; import java.io.IOException; import java.io.OutputStream; -import java.net.URI; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -271,8 +270,8 @@ public static void addEnvForSubstitution(Map env, // 2. Add the config file to localResource public synchronized void createConfigFileAndAddLocalResource( ContainerLauncher launcher, SliderFileSystem fs, Component component, - Map tokensForSubstitution, RoleInstance roleInstance) - throws IOException { + Map tokensForSubstitution, RoleInstance roleInstance, + StateAccessForProviders appState) throws IOException { Path compDir = new Path(new Path(fs.getAppDir(), "components"), component.getName()); Path compInstanceDir = @@ -315,12 +314,12 @@ public synchronized void createConfigFileAndAddLocalResource( case HADOOP_XML: // Hadoop_xml_template resolveHadoopXmlTemplateAndSaveOnHdfs(fs.getFileSystem(), - tokensForSubstitution, configFile, remoteFile, roleInstance); + tokensForSubstitution, configFile, remoteFile, appState); break; case TEMPLATE: // plain-template resolvePlainTemplateAndSaveOnHdfs(fs.getFileSystem(), - tokensForSubstitution, configFile, remoteFile, roleInstance); + tokensForSubstitution, configFile, remoteFile, appState); break; default: log.info("Not supporting loading src_file for " + configFile); @@ -383,11 +382,11 @@ private void resolveNonTemplateConfigsAndSaveOnHdfs(SliderFileSystem fs, @SuppressWarnings("unchecked") private void resolveHadoopXmlTemplateAndSaveOnHdfs(FileSystem fs, Map tokensForSubstitution, ConfigFile configFile, - Path remoteFile, RoleInstance roleInstance) throws IOException { + Path remoteFile, StateAccessForProviders appState) throws IOException { Map conf; try { - conf = (Map) roleInstance.providerRole. - appState.configFileCache.get(configFile); + conf = (Map) appState.getConfigFileCache() + .get(configFile); } catch (ExecutionException e) { log.info("Failed to load config file: " + configFile, e); return; @@ -426,17 +425,16 @@ private void resolveHadoopXmlTemplateAndSaveOnHdfs(FileSystem fs, // 3) save on hdfs private void resolvePlainTemplateAndSaveOnHdfs(FileSystem fs, Map tokensForSubstitution, ConfigFile configFile, - Path remoteFile, RoleInstance roleInstance) { + Path remoteFile, StateAccessForProviders appState) { String content; try { - content = (String) roleInstance.providerRole.appState.configFileCache - .get(configFile); + content = (String) appState.getConfigFileCache().get(configFile); } catch (ExecutionException e) { log.info("Failed to load config file: " + configFile, e); return; } // substitute tokens - substituteStrWithTokens(content, tokensForSubstitution); + content = substituteStrWithTokens(content, tokensForSubstitution); try (OutputStream output = fs.create(remoteFile)) { org.apache.commons.io.IOUtils.write(content, output); @@ -447,22 +445,20 @@ private void resolvePlainTemplateAndSaveOnHdfs(FileSystem fs, /** * Get initial token map to be substituted into config values. - * @param appConf app configurations + * @param roleInstance role instance + * @param fs file system * @param clusterName app name + * @param userName user name * @return tokens to replace */ - public Map getStandardTokenMap(Configuration appConf, - RoleInstance roleInstance, String clusterName) { + public Map getStandardTokenMap(RoleInstance roleInstance, + SliderFileSystem fs, String clusterName, String userName) { Map tokens = new HashMap<>(); - - String nnuri = appConf.getProperty("fs.defaultFS"); - if (nnuri != null && !nnuri.isEmpty()) { - tokens.put("${NN_URI}", nnuri); - tokens.put("${NN_HOST}", URI.create(nnuri).getHost()); - } - tokens.put("${ZK_HOST}", appConf.getProperty(OptionKeys.ZOOKEEPER_HOSTS)); - tokens.put("${DEFAULT_ZK_PATH}", appConf.getProperty(OptionKeys.ZOOKEEPER_PATH)); + tokens.put(SERVICE_ZK_PATH, ZKIntegration.mkClusterPath(userName, + clusterName)); + tokens.put(SERVICE_HDFS_DIR, fs.buildClusterDirPath(clusterName) + .toString()); tokens.put(SERVICE_NAME_LC, clusterName.toLowerCase()); tokens.put(SERVICE_NAME, clusterName); tokens.put(COMPONENT_NAME, roleInstance.role); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/docker/DockerProviderService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/docker/DockerProviderService.java index 93a481c..8b88371 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/docker/DockerProviderService.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/providers/docker/DockerProviderService.java @@ -23,6 +23,7 @@ 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.slider.api.ServiceApiConstants; import org.apache.slider.api.resource.Application; import org.apache.slider.api.resource.Component; import org.apache.slider.api.resource.ContainerState; @@ -86,10 +87,11 @@ public void buildContainerLaunchContext(ContainerLauncher launcher, // Generate tokens (key-value pair) for config substitution. // Get pre-defined tokens + Map globalTokens = amState.getGlobalSubstitutionTokens(); Map tokensForSubstitution = providerUtils - .getStandardTokenMap(application.getConfiguration(), roleInstance, - application.getName()); - + .getStandardTokenMap(roleInstance, fileSystem, application.getName(), + globalTokens.get(ServiceApiConstants.USER)); + tokensForSubstitution.putAll(globalTokens); // Set the environment variables in launcher launcher.putEnv(SliderUtils .buildEnvMap(component.getConfiguration(), tokensForSubstitution)); @@ -109,7 +111,7 @@ public void buildContainerLaunchContext(ContainerLauncher launcher, // create config file on hdfs and add local resource providerUtils.createConfigFileAndAddLocalResource(launcher, fileSystem, - component, tokensForSubstitution, roleInstance); + component, tokensForSubstitution, roleInstance, amState); // substitute launch command String launchCommand = ProviderUtils diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java index 9f7b4a8..2b64c05 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java @@ -24,11 +24,11 @@ import com.google.common.cache.CacheLoader; import com.google.common.cache.LoadingCache; import org.apache.commons.io.IOUtils; -import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.metrics2.lib.MutableGaugeInt; +import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.yarn.api.records.Container; import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.api.records.ContainerStatus; @@ -42,6 +42,7 @@ import org.apache.hadoop.yarn.util.resource.Resources; import org.apache.slider.api.ClusterNode; import org.apache.slider.api.InternalKeys; +import org.apache.slider.api.ServiceApiConstants; import org.apache.slider.api.StatusKeys; import org.apache.slider.api.proto.Messages; import org.apache.slider.api.proto.Messages.ComponentCountProto; @@ -75,6 +76,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.net.URI; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -89,6 +91,10 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY; +import static org.apache.hadoop.registry.client.api.RegistryConstants.DEFAULT_REGISTRY_ZK_QUORUM; +import static org.apache.hadoop.registry.client.api.RegistryConstants.KEY_DNS_DOMAIN; +import static org.apache.hadoop.registry.client.api.RegistryConstants.KEY_REGISTRY_ZK_QUORUM; import static org.apache.slider.api.ResourceKeys.*; import static org.apache.slider.api.StateValues.*; import static org.apache.slider.api.resource.ApplicationState.STARTED; @@ -193,14 +199,13 @@ private int containerMinMemory; private RoleHistory roleHistory; - private Configuration publishedProviderConf; private long startTimeThreshold; private int failureThreshold = 10; private int nodeFailureThreshold = 3; private String logServerURL = ""; - + public Map globalTokens = new HashMap<>(); /** * Selector of containers to release; application wide. */ @@ -335,6 +340,18 @@ public synchronized void buildInstance(AppStateBindingInfo binding) DEFAULT_CONTAINER_FAILURE_THRESHOLD); nodeFailureThreshold = conf.getPropertyInt(NODE_FAILURE_THRESHOLD, DEFAULT_NODE_FAILURE_THRESHOLD); + globalTokens.put(ServiceApiConstants.CLUSTER_ZK_QUORUM, + binding.serviceConfig + .getTrimmed(KEY_REGISTRY_ZK_QUORUM, DEFAULT_REGISTRY_ZK_QUORUM)); + globalTokens.put(ServiceApiConstants.USER, + UserGroupInformation.getCurrentUser().getShortUserName()); + globalTokens.put(ServiceApiConstants.DOMAIN, binding.serviceConfig + .getTrimmed(KEY_DNS_DOMAIN)); + String nnuri = binding.serviceConfig.getTrimmed(FS_DEFAULT_NAME_KEY); + if (nnuri != null && !nnuri.isEmpty()) { + globalTokens.put("${NN_URI}", nnuri); + globalTokens.put("${NN_HOST}", URI.create(nnuri).getHost()); + } //build the initial component list int priority = 1; @@ -411,7 +428,7 @@ public ProviderRole createComponent(String name, String group, DEF_YARN_LABEL_EXPRESSION); ProviderRole newRole = new ProviderRole(name, group, priority, (int)placementPolicy, threshold, - placementTimeout, label, component, this); + placementTimeout, label, component); buildRole(newRole, component); log.info("Created a new role " + newRole); return newRole; diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java index 8046472..8fc08b7 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java @@ -18,11 +18,13 @@ package org.apache.slider.server.appmaster.state; +import com.google.common.cache.LoadingCache; import org.apache.hadoop.yarn.api.records.Container; import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.slider.api.ClusterNode; import org.apache.slider.api.resource.Application; +import org.apache.slider.api.resource.ConfigFile; import org.apache.slider.api.types.ApplicationLivenessInformation; import org.apache.slider.api.types.ComponentInformation; import org.apache.slider.api.types.NodeInformation; @@ -262,4 +264,14 @@ public NodeInformation getNodeInformation(String hostname) { public RoleStatistics getRoleStatistics() { return appState.getRoleStatistics(); } + + @Override + public Map getGlobalSubstitutionTokens() { + return appState.globalTokens; + } + + @Override + public LoadingCache getConfigFileCache() { + return appState.configFileCache; + } } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/StateAccessForProviders.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/StateAccessForProviders.java index 5bc6dce..90221cb 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/StateAccessForProviders.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-slider/hadoop-yarn-slider-core/src/main/java/org/apache/slider/server/appmaster/state/StateAccessForProviders.java @@ -18,12 +18,14 @@ package org.apache.slider.server.appmaster.state; +import com.google.common.cache.LoadingCache; import org.apache.hadoop.yarn.api.records.Container; import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; import org.apache.slider.api.ClusterNode; import org.apache.slider.api.StatusKeys; import org.apache.slider.api.resource.Application; +import org.apache.slider.api.resource.ConfigFile; import org.apache.slider.api.types.ApplicationLivenessInformation; import org.apache.slider.api.types.ComponentInformation; import org.apache.slider.api.types.NodeInformation; @@ -260,4 +262,14 @@ RoleInstance getLiveInstanceByContainerID(String containerId) * @return role statistics */ RoleStatistics getRoleStatistics(); + + /** + * Get global substitution tokens. + */ + Map getGlobalSubstitutionTokens(); + + /** + * Get config file cache. + */ + LoadingCache getConfigFileCache(); }