You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ka...@apache.org on 2017/08/30 01:41:17 UTC
[1/2] storm git commit: [STORM-2201] Add dynamic scheduler
configuration loading
Repository: storm
Updated Branches:
refs/heads/master 08d9d5a87 -> e1dd247ce
[STORM-2201] Add dynamic scheduler configuration loading
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/f1f95449
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/f1f95449
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/f1f95449
Branch: refs/heads/master
Commit: f1f954493408e5c62d4a0e061ae6180c8ad7972a
Parents: d06bb38
Author: Ethan Li <et...@gmail.com>
Authored: Mon Jul 10 12:04:50 2017 -0500
Committer: Ethan Li <et...@gmail.com>
Committed: Tue Aug 29 12:54:00 2017 -0500
----------------------------------------------------------------------
docs/IConfigLoader.md | 51 +++
pom.xml | 7 +
storm-server/pom.xml | 5 +
.../java/org/apache/storm/DaemonConfig.java | 94 +++--
.../multitenant/MultitenantScheduler.java | 62 ++-
.../resource/ResourceAwareScheduler.java | 89 +++--
.../utils/ArtifactoryConfigLoader.java | 395 +++++++++++++++++++
.../utils/ArtifactoryConfigLoaderFactory.java | 50 +++
.../utils/ConfigLoaderFactoryService.java | 68 ++++
.../storm/scheduler/utils/FileConfigLoader.java | 72 ++++
.../utils/FileConfigLoaderFactory.java | 51 +++
.../storm/scheduler/utils/IConfigLoader.java | 32 ++
.../scheduler/utils/IConfigLoaderFactory.java | 35 ++
.../utils/ArtifactoryConfigLoaderTest.java | 235 +++++++++++
.../scheduler/utils/FileConfigLoaderTest.java | 115 ++++++
15 files changed, 1275 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/docs/IConfigLoader.md
----------------------------------------------------------------------
diff --git a/docs/IConfigLoader.md b/docs/IConfigLoader.md
new file mode 100644
index 0000000..3c2f4de
--- /dev/null
+++ b/docs/IConfigLoader.md
@@ -0,0 +1,51 @@
+---
+title: IConfigLoader
+layout: documentation
+documentation: true
+---
+
+
+### Introduction
+IConfigLoader is an interface designed to allow dynamic loading of scheduler resource constraints. Currently, the MultiTenant scheduler uses this interface to dynamically load the number of isolated nodes a given user has been guaranteed, and the ResoureAwareScheduler uses the interface to dynamically load per user resource guarantees.
+
+The following interface is provided for users to create an IConfigLoader instance based on the scheme of the `scheduler.config.loader.uri`.
+```
+ConfigLoaderFactoryService.createConfigLoader(Map<String, Object> conf)
+```
+
+------
+
+### Interface
+```
+public interface IConfigLoader {
+ Map<?,?> load();
+};
+```
+#### Description
+ - load is called by the scheduler whenever it wishes to retrieve the most recent configuration map.
+
+#### Loader Configuration
+The loaders are dynamically selected and dynamically configured through configuration items in the scheduler implementations.
+
+##### Example
+```
+scheduler.config.loader.uri: "artifactory+http://artifactory.my.company.com:8000/artifactory/configurations/clusters/my_cluster/ras_pools"
+scheduler.config.loader.timeout.sec: 30
+```
+Or
+```
+scheduler.config.loader.uri: "file:///path/to/my/config.yaml"
+```
+### Implementations
+
+There are currently two implemenations of IConfigLoader
+ - org.apache.storm.scheduler.utils.ArtifactoryConfigLoader: Load configurations from an Artifactory server.
+ It will be used if users add `artifactory+` to the scheme of the real URI and set to `scheduler.config.loader.uri`.
+ - org.apache.storm.scheduler.utils.FileConfigLoader: Load configurations from a local file. It will be used if users use `file` scheme.
+
+#### Configurations
+ - scheduler.config.loader.uri: For `ArtifactoryConfigLoader`, this can either be a reference to an individual file in Artifactory or to a directory. If it is a directory, the file with the largest lexographic name will be returned.
+ For `FileConfigLoader`, this is the URI pointing to a file.
+ - scheduler.config.loader.timeout.secs: Currently only used in `ArtifactoryConfigLoader`. It is the amount of time an http connection to the artifactory server will wait before timing out. The default is 10.
+ - scheduler.config.loader.polltime.secs: Currently only used in `ArtifactoryConfigLoader`. It is the frequency at which the plugin will call out to artifactory instead of returning the most recently cached result. The default is 600 seconds.
+ - scheduler.config.loader.artifactory.base.directory: Only used in `ArtifactoryConfigLoader`. It is the part of the uri, configurable in Artifactory, which represents the top of the directory tree. It defaults to "/artifactory".
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index c3a04b9..9ea2501 100644
--- a/pom.xml
+++ b/pom.xml
@@ -261,6 +261,7 @@
<disruptor.version>3.3.2</disruptor.version>
<jgrapht.version>0.9.0</jgrapht.version>
<guava.version>16.0.1</guava.version>
+ <auto-service.version>1.0-rc3</auto-service.version>
<netty.version>3.9.0.Final</netty.version>
<sysout-over-slf4j.version>1.0.2</sysout-over-slf4j.version>
<log4j-over-slf4j.version>1.6.6</log4j-over-slf4j.version>
@@ -899,6 +900,12 @@
<artifactId>guava</artifactId>
<version>${guava.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.google.auto.service</groupId>
+ <artifactId>auto-service</artifactId>
+ <version>${auto-service.version}</version>
+ <optional>true</optional>
+ </dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/pom.xml
----------------------------------------------------------------------
diff --git a/storm-server/pom.xml b/storm-server/pom.xml
index 311b58f..b2a2c0e 100644
--- a/storm-server/pom.xml
+++ b/storm-server/pom.xml
@@ -59,6 +59,11 @@
<groupId>org.apache.commons</groupId>
<artifactId>commons-compress</artifactId>
</dependency>
+ <dependency>
+ <groupId>com.google.auto.service</groupId>
+ <artifactId>auto-service</artifactId>
+ <optional>true</optional>
+ </dependency>
<!-- test -->
<dependency>
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
index a1292e0..75a9cad 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -221,7 +221,7 @@ public class DaemonConfig implements Validated {
public static final String NIMBUS_TOPOLOGY_VALIDATOR = "nimbus.topology.validator";
/**
- * Class name for authorization plugin for Nimbus
+ * Class name for authorization plugin for Nimbus.
*/
@isString
public static final String NIMBUS_AUTHORIZER = "nimbus.authorizer";
@@ -271,7 +271,7 @@ public class DaemonConfig implements Validated {
public static final String UI_CENTRAL_LOGGING_URL = "ui.central.logging.url";
/**
- * HTTP UI port for log viewer
+ * HTTP UI port for log viewer.
*/
@isInteger
@isPositiveNumber
@@ -284,46 +284,46 @@ public class DaemonConfig implements Validated {
public static final String LOGVIEWER_CHILDOPTS = "logviewer.childopts";
/**
- * How often to clean up old log files
+ * How often to clean up old log files.
*/
@isInteger
@isPositiveNumber
public static final String LOGVIEWER_CLEANUP_INTERVAL_SECS = "logviewer.cleanup.interval.secs";
/**
- * How many minutes since a log was last modified for the log to be considered for clean-up
+ * How many minutes since a log was last modified for the log to be considered for clean-up.
*/
@isInteger
@isPositiveNumber
public static final String LOGVIEWER_CLEANUP_AGE_MINS = "logviewer.cleanup.age.mins";
/**
- * The maximum number of bytes all worker log files can take up in MB
+ * The maximum number of bytes all worker log files can take up in MB.
*/
@isPositiveNumber
public static final String LOGVIEWER_MAX_SUM_WORKER_LOGS_SIZE_MB = "logviewer.max.sum.worker.logs.size.mb";
/**
- * The maximum number of bytes per worker's files can take up in MB
+ * The maximum number of bytes per worker's files can take up in MB.
*/
@isPositiveNumber
public static final String LOGVIEWER_MAX_PER_WORKER_LOGS_SIZE_MB = "logviewer.max.per.worker.logs.size.mb";
/**
- * Storm Logviewer HTTPS port
+ * Storm Logviewer HTTPS port.
*/
@isInteger
@isPositiveNumber
public static final String LOGVIEWER_HTTPS_PORT = "logviewer.https.port";
/**
- * Path to the keystore containing the certs used by Storm Logviewer for HTTPS communications
+ * Path to the keystore containing the certs used by Storm Logviewer for HTTPS communications.
*/
@isString
public static final String LOGVIEWER_HTTPS_KEYSTORE_PATH = "logviewer.https.keystore.path";
/**
- * Password for the keystore for HTTPS for Storm Logviewer
+ * Password for the keystore for HTTPS for Storm Logviewer.
*/
@isString
public static final String LOGVIEWER_HTTPS_KEYSTORE_PASSWORD = "logviewer.https.keystore.password";
@@ -342,13 +342,13 @@ public class DaemonConfig implements Validated {
public static final String LOGVIEWER_HTTPS_KEY_PASSWORD = "logviewer.https.key.password";
/**
- * Path to the truststore containing the certs used by Storm Logviewer for HTTPS communications
+ * Path to the truststore containing the certs used by Storm Logviewer for HTTPS communications.
*/
@isString
public static final String LOGVIEWER_HTTPS_TRUSTSTORE_PATH = "logviewer.https.truststore.path";
/**
- * Password for the truststore for HTTPS for Storm Logviewer
+ * Password for the truststore for HTTPS for Storm Logviewer.
*/
@isString
public static final String LOGVIEWER_HTTPS_TRUSTSTORE_PASSWORD = "logviewer.https.truststore.password";
@@ -370,13 +370,13 @@ public class DaemonConfig implements Validated {
public static final String LOGVIEWER_HTTPS_NEED_CLIENT_AUTH = "logviewer.https.need.client.auth";
/**
- * A list of users allowed to view logs via the Log Viewer
+ * A list of users allowed to view logs via the Log Viewer.
*/
@isStringList
public static final String LOGS_USERS = "logs.users";
/**
- * A list of groups allowed to view logs via the Log Viewer
+ * A list of groups allowed to view logs via the Log Viewer.
*/
@isStringList
public static final String LOGS_GROUPS = "logs.groups";
@@ -394,19 +394,19 @@ public class DaemonConfig implements Validated {
public static final String UI_CHILDOPTS = "ui.childopts";
/**
- * A class implementing javax.servlet.Filter for authenticating/filtering UI requests
+ * A class implementing javax.servlet.Filter for authenticating/filtering UI requests.
*/
@isString
public static final String UI_FILTER = "ui.filter";
/**
- * Initialization parameters for the javax.servlet.Filter
+ * Initialization parameters for the javax.servlet.Filter.
*/
@isMapEntryType(keyType = String.class, valueType = String.class)
public static final String UI_FILTER_PARAMS = "ui.filter.params";
/**
- * The size of the header buffer for the UI in bytes
+ * The size of the header buffer for the UI in bytes.
*/
@isInteger
@isPositiveNumber
@@ -555,7 +555,7 @@ public class DaemonConfig implements Validated {
public static final String DRPC_HTTPS_NEED_CLIENT_AUTH = "drpc.https.need.client.auth";
/**
- * Class name for authorization plugin for DRPC client
+ * Class name for authorization plugin for DRPC client.
*/
@isString
public static final String DRPC_AUTHORIZER = "drpc.authorizer";
@@ -578,7 +578,7 @@ public class DaemonConfig implements Validated {
public static final String DRPC_CHILDOPTS = "drpc.childopts";
/**
- * the metadata configured on the supervisor
+ * the metadata configured on the supervisor.
*/
@isMapEntryType(keyType = String.class, valueType = String.class)
public static final String SUPERVISOR_SCHEDULER_META = "supervisor.scheduler.meta";
@@ -590,7 +590,7 @@ public class DaemonConfig implements Validated {
*/
@isNoDuplicateInList
@NotNull
- @isListEntryCustom(entryValidatorClasses={ConfigValidation.IntegerValidator.class,ConfigValidation.PositiveNumberValidator.class})
+ @isListEntryCustom(entryValidatorClasses = {ConfigValidation.IntegerValidator.class,ConfigValidation.PositiveNumberValidator.class})
public static final String SUPERVISOR_SLOTS_PORTS = "supervisor.slots.ports";
/**
@@ -652,14 +652,14 @@ public class DaemonConfig implements Validated {
public static final String NIMBUS_SLOTS_PER_TOPOLOGY = "nimbus.slots.perTopology";
/**
- * A class implementing javax.servlet.Filter for DRPC HTTP requests
+ * A class implementing javax.servlet.Filter for DRPC HTTP requests.
*/
@isString
public static final String DRPC_HTTP_FILTER = "drpc.http.filter";
/**
* Initialization parameters for the javax.servlet.Filter of the DRPC HTTP
- * service
+ * service.
*/
@isMapEntryType(keyType = String.class, valueType = String.class)
public static final String DRPC_HTTP_FILTER_PARAMS = "drpc.http.filter.params";
@@ -679,7 +679,7 @@ public class DaemonConfig implements Validated {
public static final String SUPERVISOR_CHILDOPTS = "supervisor.childopts";
/**
- * How many seconds to sleep for before shutting down threads on worker
+ * How many seconds to sleep for before shutting down threads on worker.
*/
@isInteger
@isPositiveNumber
@@ -739,7 +739,7 @@ public class DaemonConfig implements Validated {
/**
* The command launched supervisor with worker arguments
* pid, action and [target_directory]
- * Where action is - start profile, stop profile, jstack, heapdump and kill against pid
+ * Where action is - start profile, stop profile, jstack, heapdump and kill against pid.
*
*/
@isString
@@ -762,7 +762,7 @@ public class DaemonConfig implements Validated {
public static final String STORM_CLUSTER_METRICS_CONSUMER_PUBLISH_INTERVAL_SECS = "storm.cluster.metrics.consumer.publish.interval.secs";
/**
- * Enables user-first classpath. See topology.classpath.beginning
+ * Enables user-first classpath. See topology.classpath.beginning.
*/
@isBoolean
public static final String STORM_TOPOLOGY_CLASSPATH_BEGINNING_ENABLED="storm.topology.classpath.beginning.enabled";
@@ -792,6 +792,38 @@ public class DaemonConfig implements Validated {
public static final String ISOLATION_SCHEDULER_MACHINES = "isolation.scheduler.machines";
/**
+ * For ArtifactoryConfigLoader, this can either be a reference to an individual file in Artifactory or to a directory.
+ * If it is a directory, the file with the largest lexographic name will be returned. Users need to add "artifactory+" to the beginning of
+ * the real URI to use ArtifactoryConfigLoader.
+ * For FileConfigLoader, this is the URI pointing to a file.
+ */
+ @isString
+ public static final String SCHEDULER_CONFIG_LOADER_URI = "scheduler.config.loader.uri";
+
+ /**
+ * It is the frequency at which the plugin will call out to artifactory instead of returning the most recently cached result.
+ * Currently it's only used in ArtifactoryConfigLoader.
+ */
+ @isInteger
+ @isPositiveNumber
+ public static final String SCHEDULER_CONFIG_LOADER_POLLTIME_SECS = "scheduler.config.loader.polltime.secs";
+
+ /**
+ * It is the amount of time an http connection to the artifactory server will wait before timing out.
+ * Currently it's only used in ArtifactoryConfigLoader.
+ */
+ @isInteger
+ @isPositiveNumber
+ public static final String SCHEDULER_CONFIG_LOADER_TIMEOUT_SECS = "scheduler.config.loader.timeout.secs";
+
+ /**
+ * It is the part of the uri, configurable in Artifactory, which represents the top of the directory tree.
+ * It's only used in ArtifactoryConfigLoader.
+ */
+ @isString
+ public static final String SCHEDULER_CONFIG_LOADER_ARTIFACTORY_BASE_DIRECTORY = "scheduler.config.loader.artifactory.base.directory";
+
+ /**
* A map from the user name to the number of machines that should that user is allowed to use. Set storm.scheduler
* to org.apache.storm.scheduler.multitenant.MultitenantScheduler
*/
@@ -806,14 +838,14 @@ public class DaemonConfig implements Validated {
public static final String RESOURCE_AWARE_SCHEDULER_USER_POOLS = "resource.aware.scheduler.user.pools";
/**
- * The class that specifies the eviction strategy to use in ResourceAwareScheduler
+ * The class that specifies the eviction strategy to use in ResourceAwareScheduler.
*/
@NotNull
@isImplementationOfClass(implementsClass = IEvictionStrategy.class)
public static final String RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY = "resource.aware.scheduler.eviction.strategy";
/**
- * the class that specifies the scheduling priority strategy to use in ResourceAwareScheduler
+ * the class that specifies the scheduling priority strategy to use in ResourceAwareScheduler.
*/
@NotNull
@isImplementationOfClass(implementsClass = ISchedulingPriorityStrategy.class)
@@ -842,7 +874,7 @@ public class DaemonConfig implements Validated {
public static final String STORM_CGROUP_RESOURCES = "storm.cgroup.resources";
/**
- * name for the cgroup hierarchy
+ * name for the cgroup hierarchy.
*/
@isString
public static final String STORM_CGROUP_HIERARCHY_NAME = "storm.cgroup.hierarchy.name";
@@ -856,25 +888,25 @@ public class DaemonConfig implements Validated {
public static final String STORM_RESOURCE_ISOLATION_PLUGIN_ENABLE = "storm.resource.isolation.plugin.enable";
/**
- * root directory for cgoups
+ * root directory for cgoups.
*/
@isString
public static String STORM_SUPERVISOR_CGROUP_ROOTDIR = "storm.supervisor.cgroup.rootdir";
/**
- * the manually set memory limit (in MB) for each CGroup on supervisor node
+ * the manually set memory limit (in MB) for each CGroup on supervisor node.
*/
@isPositiveNumber
public static String STORM_WORKER_CGROUP_MEMORY_MB_LIMIT = "storm.worker.cgroup.memory.mb.limit";
/**
- * the manually set cpu share for each CGroup on supervisor node
+ * the manually set cpu share for each CGroup on supervisor node.
*/
@isPositiveNumber
public static String STORM_WORKER_CGROUP_CPU_LIMIT = "storm.worker.cgroup.cpu.limit";
/**
- * full path to cgexec command
+ * full path to cgexec command.
*/
@isString
public static String STORM_CGROUP_CGEXEC_CMD = "storm.cgroup.cgexec.cmd";
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/multitenant/MultitenantScheduler.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/multitenant/MultitenantScheduler.java b/storm-server/src/main/java/org/apache/storm/scheduler/multitenant/MultitenantScheduler.java
index 95cbf63..8bd9035 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/multitenant/MultitenantScheduler.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/multitenant/MultitenantScheduler.java
@@ -20,44 +20,66 @@ package org.apache.storm.scheduler.multitenant;
import java.util.HashMap;
import java.util.Map;
-
import org.apache.storm.DaemonConfig;
-import org.apache.storm.utils.Utils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.storm.Config;
import org.apache.storm.scheduler.Cluster;
import org.apache.storm.scheduler.IScheduler;
import org.apache.storm.scheduler.Topologies;
import org.apache.storm.scheduler.TopologyDetails;
+import org.apache.storm.scheduler.utils.ConfigLoaderFactoryService;
+import org.apache.storm.scheduler.utils.IConfigLoader;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class MultitenantScheduler implements IScheduler {
private static final Logger LOG = LoggerFactory.getLogger(MultitenantScheduler.class);
@SuppressWarnings("rawtypes")
- private Map _conf;
+ private Map conf;
+ protected IConfigLoader configLoader;
@Override
- public void prepare(@SuppressWarnings("rawtypes") Map<String, Object> conf) {
- _conf = conf;
+ public void prepare(Map<String, Object> conf) {
+ this.conf = conf;
+ configLoader = ConfigLoaderFactoryService.createConfigLoader(conf);
+
}
-
+
+ /**
+ * Load from configLoaders first; if no config available, read from multitenant-scheduler.yaml;
+ * if no config available from multitenant-scheduler.yaml, get configs from conf. Only one will be used.
+ * @return User pool configs.
+ */
private Map<String, Number> getUserConf() {
- Map<String, Number> ret = (Map<String, Number>)_conf.get(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
- if (ret == null) {
- ret = new HashMap<>();
- } else {
- ret = new HashMap<>(ret);
+ Map<String, Number> ret;
+
+ // Try the loader plugin, if configured
+ if (configLoader != null) {
+ ret = (Map<String, Number>) configLoader.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ if (ret != null) {
+ return ret;
+ } else {
+ LOG.warn("Config loader returned null. Will try to read from multitenant-scheduler.yaml");
+ }
}
+ // If that fails, fall back on the multitenant-scheduler.yaml file
Map fromFile = Utils.findAndReadConfigFile("multitenant-scheduler.yaml", false);
- Map<String, Number> tmp = (Map<String, Number>)fromFile.get(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
- if (tmp != null) {
- ret.putAll(tmp);
+ ret = (Map<String, Number>)fromFile.get(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ if (ret != null) {
+ return ret;
+ } else {
+ LOG.warn("Reading from multitenant-scheduler.yaml returned null. This could because the file is not available. "
+ + "Will load configs from storm configuration");
}
- return ret;
- }
+ // If that fails, use config
+ ret = (Map<String, Number>) conf.get(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ if (ret == null) {
+ return new HashMap<>();
+ } else {
+ return ret;
+ }
+ }
@Override
public Map<String, Object> config() {
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
index 19838f4..669079b 100644
--- a/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/resource/ResourceAwareScheduler.java
@@ -20,7 +20,6 @@ package org.apache.storm.scheduler.resource;
import java.util.HashMap;
import java.util.Map;
-
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.scheduler.Cluster;
@@ -31,8 +30,10 @@ import org.apache.storm.scheduler.TopologyDetails;
import org.apache.storm.scheduler.resource.strategies.eviction.IEvictionStrategy;
import org.apache.storm.scheduler.resource.strategies.priority.ISchedulingPriorityStrategy;
import org.apache.storm.scheduler.resource.strategies.scheduling.IStrategy;
+import org.apache.storm.scheduler.utils.ConfigLoaderFactoryService;
import org.apache.storm.utils.ReflectionUtils;
import org.apache.storm.utils.DisallowedStrategyException;
+import org.apache.storm.scheduler.utils.IConfigLoader;
import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,14 +43,16 @@ public class ResourceAwareScheduler implements IScheduler {
private Map<String, Object> conf;
private ISchedulingPriorityStrategy schedulingPrioritystrategy;
private IEvictionStrategy evictionStrategy;
+ private IConfigLoader configLoader;
@Override
public void prepare(Map<String, Object> conf) {
this.conf = conf;
schedulingPrioritystrategy = (ISchedulingPriorityStrategy) ReflectionUtils.newInstance(
- (String) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY));
+ (String) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_PRIORITY_STRATEGY));
evictionStrategy = (IEvictionStrategy) ReflectionUtils.newInstance(
- (String) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY));
+ (String) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_EVICTION_STRATEGY));
+ configLoader = ConfigLoaderFactoryService.createConfigLoader(conf);
}
@Override
@@ -75,7 +78,7 @@ public class ResourceAwareScheduler implements IScheduler {
td = schedulingPrioritystrategy.getNextTopologyToSchedule(cluster, userMap);
} catch (Exception ex) {
LOG.error("Exception thrown when running priority strategy {}. No topologies will be scheduled!",
- schedulingPrioritystrategy.getClass().getName(), ex);
+ schedulingPrioritystrategy.getClass().getName(), ex);
break;
}
if (td == null) {
@@ -104,8 +107,8 @@ public class ResourceAwareScheduler implements IScheduler {
} catch (DisallowedStrategyException e) {
topologySubmitter.markTopoUnsuccess(td);
cluster.setStatus(td.getId(), "Unsuccessful in scheduling - " + e.getAttemptedClass()
- + " is not an allowed strategy. Please make sure your " + Config.TOPOLOGY_SCHEDULER_STRATEGY
- + " config is one of the allowed strategies: " + e.getAllowedStrategies().toString());
+ + " is not an allowed strategy. Please make sure your " + Config.TOPOLOGY_SCHEDULER_STRATEGY
+ + " config is one of the allowed strategies: " + e.getAllowedStrategies().toString());
return;
} catch (RuntimeException e) {
LOG.error("failed to create instance of IStrategy: {} Topology {} will not be scheduled.",
@@ -115,7 +118,7 @@ public class ResourceAwareScheduler implements IScheduler {
+ strategyConf + ". Please check logs for details");
return;
}
-
+
while (true) {
// A copy of the cluster that restricts the strategy to only modify a single topology
SingleTopologyCluster toSchedule = new SingleTopologyCluster(workingState, td.getId());
@@ -139,8 +142,8 @@ public class ResourceAwareScheduler implements IScheduler {
LOG.error("Unsuccessful attempting to assign executors to nodes.", ex);
topologySubmitter.markTopoUnsuccess(td);
cluster.setStatus(td.getId(), "Unsuccessful in scheduling - "
- + "IllegalStateException thrown when attempting to assign executors to nodes. Please check"
- + " log for details.");
+ + "IllegalStateException thrown when attempting to assign executors to nodes. Please check"
+ + " log for details.");
}
return;
} else {
@@ -151,8 +154,8 @@ public class ResourceAwareScheduler implements IScheduler {
madeSpace = evictionStrategy.makeSpaceForTopo(td, workingState, userMap);
} catch (Exception ex) {
LOG.error("Exception thrown when running eviction strategy {} to schedule topology {}."
- + " No evictions will be done!", evictionStrategy.getClass().getName(),
- td.getName(), ex);
+ + " No evictions will be done!", evictionStrategy.getClass().getName(),
+ td.getName(), ex);
topologySubmitter.markTopoUnsuccess(td);
return;
}
@@ -160,7 +163,7 @@ public class ResourceAwareScheduler implements IScheduler {
LOG.debug("Could not make space for topo {} will move to attempted", td);
topologySubmitter.markTopoUnsuccess(td);
cluster.setStatus(td.getId(), "Not enough resources to schedule - "
- + result.getErrorMessage());
+ + result.getErrorMessage());
return;
}
continue;
@@ -171,7 +174,7 @@ public class ResourceAwareScheduler implements IScheduler {
}
} else {
LOG.warn("Scheduling results returned from topology {} is not vaild! Topology with be ignored.",
- td.getName());
+ td.getName());
topologySubmitter.markTopoUnsuccess(td, cluster);
return;
}
@@ -202,19 +205,11 @@ public class ResourceAwareScheduler implements IScheduler {
return userMap;
}
- /**
- * Get resource guarantee configs.
- *
- * @return a map that contains resource guarantees of every user of the following format
- * {userid->{resourceType->amountGuaranteed}}
- */
- private Map<String, Map<String, Double>> getUserResourcePools() {
- Map<String, Map<String, Number>> raw =
- (Map<String, Map<String, Number>>) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS);
- Map<String, Map<String, Double>> ret = new HashMap<>();
+ private Map<String, Map<String, Double>> convertToDouble(Map<String, Map<String, Number>> raw) {
+ Map<String, Map<String, Double>> ret = new HashMap<String, Map<String, Double>>();
if (raw != null) {
- for (Map.Entry<String, Map<String, Number>> userPoolEntry : raw.entrySet()) {
+ for (Map.Entry<String, Map<String, Number>> userPoolEntry : raw.entrySet()) {
String user = userPoolEntry.getKey();
ret.put(user, new HashMap<String, Double>());
for (Map.Entry<String, Number> resourceEntry : userPoolEntry.getValue().entrySet()) {
@@ -223,18 +218,42 @@ public class ResourceAwareScheduler implements IScheduler {
}
}
- Map fromFile = Utils.findAndReadConfigFile("user-resource-pools.yaml", false);
- Map<String, Map<String, Number>> tmp =
- (Map<String, Map<String, Number>>) fromFile.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS);
- if (tmp != null) {
- for (Map.Entry<String, Map<String, Number>> userPoolEntry : tmp.entrySet()) {
- String user = userPoolEntry.getKey();
- ret.put(user, new HashMap<String, Double>());
- for (Map.Entry<String, Number> resourceEntry : userPoolEntry.getValue().entrySet()) {
- ret.get(user).put(resourceEntry.getKey(), resourceEntry.getValue().doubleValue());
- }
+ return ret;
+ }
+
+ /**
+ * Get resource guarantee configs.
+ * Load from configLoaders first; if no config available, read from user-resource-pools.yaml;
+ * if no config available from user-resource-pools.yaml, get configs from conf. Only one will be used.
+ * @return a map that contains resource guarantees of every user of the following format
+ * {userid->{resourceType->amountGuaranteed}}
+ */
+ private Map<String, Map<String, Double>> getUserResourcePools() {
+ Map<String, Map<String, Number>> raw;
+
+ // Try the loader plugin, if configured
+ if (configLoader != null) {
+ raw = (Map<String, Map<String, Number>>) configLoader.load(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS);
+ if (raw != null) {
+ return convertToDouble(raw);
+ } else {
+ LOG.warn("Config loader returned null. Will try to read from user-resource-pools.yaml");
}
}
- return ret;
+
+ // if no configs from loader, try to read from user-resource-pools.yaml
+ Map fromFile = Utils.findAndReadConfigFile("user-resource-pools.yaml", false);
+ raw = (Map<String, Map<String, Number>>) fromFile.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS);
+ if (raw != null) {
+ return convertToDouble(raw);
+ } else {
+ LOG.warn("Reading from user-resource-pools.yaml returned null. This could because the file is not available. "
+ + "Will load configs from storm configuration");
+ }
+
+ // if no configs from user-resource-pools.yaml, get configs from conf
+ raw = (Map<String, Map<String, Number>>) conf.get(DaemonConfig.RESOURCE_AWARE_SCHEDULER_USER_POOLS);
+
+ return convertToDouble(raw);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoader.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoader.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoader.java
new file mode 100644
index 0000000..3afa903
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoader.java
@@ -0,0 +1,395 @@
+/**
+ * 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.storm.scheduler.utils;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Map;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.util.EntityUtils;
+import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
+import org.apache.storm.utils.Time;
+import org.apache.storm.utils.Utils;
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+/**
+ * A dynamic loader that can load scheduler configurations for user resource guarantees from Artifactory (an artifact repository manager).
+ */
+public class ArtifactoryConfigLoader implements IConfigLoader {
+ protected static final String LOCAL_ARTIFACT_DIR = "scheduler_artifacts";
+ static final String cacheFilename = "latest.yaml";
+ private static final String DEFAULT_ARTIFACTORY_BASE_DIRECTORY = "/artifactory";
+ private static final int DEFAULT_POLLTIME_SECS = 600;
+ private static final int DEFAULT_TIMEOUT_SECS = 10;
+ private static final String ARTIFACTORY_SCHEME_PREFIX = "artifactory+";
+
+ private static final Logger LOG = LoggerFactory.getLogger(ArtifactoryConfigLoader.class);
+
+ private Map<String, Object> conf;
+ private int artifactoryPollTimeSecs = DEFAULT_POLLTIME_SECS;
+ private boolean cacheInitialized = false;
+ // Location of the file in the artifactory archive. Also used to name file in cache.
+ private String localCacheDir;
+ private String baseDirectory = DEFAULT_ARTIFACTORY_BASE_DIRECTORY;
+ private int lastReturnedTime = 0;
+ private int timeoutSeconds = DEFAULT_TIMEOUT_SECS;
+ private Map lastReturnedValue;
+ private URI targetURI = null;
+ private JSONParser jsonParser;
+ private String scheme;
+
+ public ArtifactoryConfigLoader(Map<String, Object> conf) {
+ this.conf = conf;
+ Integer thisTimeout = (Integer) conf.get(DaemonConfig.SCHEDULER_CONFIG_LOADER_TIMEOUT_SECS);
+ if (thisTimeout != null) {
+ timeoutSeconds = thisTimeout;
+ }
+ Integer thisPollTime = (Integer) conf.get(DaemonConfig.SCHEDULER_CONFIG_LOADER_POLLTIME_SECS);
+ if (thisPollTime != null) {
+ artifactoryPollTimeSecs =thisPollTime;
+ }
+ String thisBase = (String) conf.get(DaemonConfig.SCHEDULER_CONFIG_LOADER_ARTIFACTORY_BASE_DIRECTORY);
+ if (thisBase != null) {
+ baseDirectory = thisBase;
+ }
+ String uriString = (String) conf.get(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI);
+ if (uriString == null) {
+ LOG.error("No URI defined in {} configuration.", DaemonConfig.SCHEDULER_CONFIG_LOADER_URI);
+ } else {
+ try {
+ targetURI = new URI(uriString);
+ scheme = targetURI.getScheme().substring(ARTIFACTORY_SCHEME_PREFIX.length());
+ } catch(URISyntaxException e) {
+ LOG.error("Failed to parse uri={}", uriString);
+ }
+ }
+ jsonParser = new JSONParser();
+ }
+
+ /**
+ * Load the configs associated with the configKey from the targetURI.
+ * @param configKey The key from which we want to get the scheduler config.
+ * @return The scheduler configuration if exists; null otherwise.
+ */
+ @Override
+ public Map load(String configKey) {
+ if (targetURI == null) {
+ return null;
+ }
+
+ // Check for new file every so often
+ int currentTimeSecs = Time.currentTimeSecs();
+ if (lastReturnedValue != null && ((currentTimeSecs - lastReturnedTime) < artifactoryPollTimeSecs)) {
+ LOG.debug("currentTimeSecs: {}; lastReturnedTime {}; artifactoryPollTimeSecs: {}. Returning our last map.",
+ currentTimeSecs, lastReturnedTime, artifactoryPollTimeSecs);
+ return (Map) lastReturnedValue.get(configKey);
+ }
+
+ try {
+ Map raw = loadFromURI(targetURI);
+ if (raw != null) {
+ return (Map) raw.get(configKey);
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to load from uri {}", targetURI);
+ }
+ return null;
+ }
+
+
+ /**
+ * A private class used to check the response coming back from httpclient.
+ */
+ private static class GETStringResponseHandler implements ResponseHandler<String> {
+ private static GETStringResponseHandler singleton = null;
+
+ /**
+ * @return a singleton httpclient GET response handler
+ */
+ public static GETStringResponseHandler getInstance() {
+ if (singleton == null) {
+ singleton = new GETStringResponseHandler();
+ }
+ return singleton;
+ }
+
+ /**
+ * @param response The http response to verify.
+ * @return null on failure or the response string if return code is in 200 range
+ */
+ @Override
+ public String handleResponse(final HttpResponse response) throws IOException {
+ int status = response.getStatusLine().getStatusCode();
+ HttpEntity entity = response.getEntity();
+ String entityString = (entity != null ? EntityUtils.toString(entity) : null);
+ if (status >= 200 && status < 300) {
+ return entityString;
+ } else {
+ LOG.error("Got unexpected response code {}; entity: {}", status, entityString);
+ return null;
+ }
+ }
+ }
+
+ /**
+ * @param api null if we are trying to download artifact, otherwise a string to call REST api,
+ * e.g. "/api/storage"
+ * @param artifact location of artifact
+ * @param host Artifactory hostname
+ * @param port Artifactory port
+ * @return null on failure or the response string if return code is in 200 range
+ *
+ * <p>Protected so we can override this in unit tests
+ */
+ protected String doGet(String api, String artifact, String host, Integer port) {
+ URIBuilder builder = new URIBuilder().setScheme(scheme).setHost(host).setPort(port);
+
+ String path = null;
+ if (api != null) {
+ path = baseDirectory + "/" + api + "/" + artifact;
+ } else {
+ path = baseDirectory + "/" + artifact;
+ }
+
+ // Get rid of multiple '/' in url
+ path = path.replaceAll("/[/]+", "/");
+ builder.setPath(path);
+
+ RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(timeoutSeconds * 1000).build();
+ HttpClient httpclient = HttpClientBuilder.create().setDefaultRequestConfig(requestConfig).build();
+
+ String returnValue;
+ try {
+ LOG.debug("About to issue a GET to {}", builder);
+ HttpGet httpget = new HttpGet(builder.build());
+ String responseBody;
+ responseBody = httpclient.execute(httpget, GETStringResponseHandler.getInstance());
+ returnValue = responseBody;
+ } catch (Exception e) {
+ LOG.error("Received exception while connecting to Artifactory", e);
+ returnValue = null;
+ }
+
+ LOG.debug("Returning {}", returnValue);
+ return returnValue;
+ }
+
+ private JSONObject getArtifactMetadata(String location, String host, Integer port) {
+ String metadataStr = null;
+
+ metadataStr = doGet("/api/storage", location, host, port);
+
+ if (metadataStr == null) {
+ return null;
+ }
+
+ JSONObject returnValue;
+ try {
+ returnValue = (JSONObject) jsonParser.parse(metadataStr);
+ } catch (ParseException e) {
+ LOG.error("Could not parse JSON string {}", metadataStr, e);
+ return null;
+ }
+
+ return returnValue;
+ }
+
+ private class DirEntryCompare implements Comparator<JSONObject> {
+ @Override
+ public int compare(JSONObject o1, JSONObject o2) {
+ return ((String)o1.get("uri")).compareTo((String)o2.get("uri"));
+ }
+ }
+
+ private String loadMostRecentArtifact(String location, String host, Integer port) {
+ // Is this a directory or is it a file?
+ JSONObject json = getArtifactMetadata(location, host, port);
+ if (json == null) {
+ LOG.error("got null metadata");
+ return null;
+ }
+ String downloadURI = (String) json.get("downloadUri");
+
+ // This means we are pointing at a file.
+ if (downloadURI != null) {
+ // Then get it and return the file as string.
+ String returnValue = doGet(null, location, host, port);
+ saveInArtifactoryCache(returnValue);
+ return returnValue;
+ }
+
+ // This should mean that we were pointed at a directory.
+ // Find the most recent child and load that.
+ JSONArray msg = (JSONArray) json.get("children");
+ if (msg == null || msg.size() == 0) {
+ LOG.error("Expected directory children not present");
+ return null;
+ }
+ JSONObject newest = (JSONObject) Collections.max(msg, new DirEntryCompare());
+ if (newest == null) {
+ LOG.error("Failed to find most recent artifact uri in {}", location);
+ return null;
+ }
+
+ String uri = (String) newest.get("uri");
+ if (uri == null) {
+ LOG.error("Expected directory uri not present");
+ return null;
+ }
+ String returnValue = doGet(null, location + uri, host, port);
+ saveInArtifactoryCache(returnValue);
+ return returnValue;
+ }
+
+ private void updateLastReturned(Map ret) {
+ lastReturnedTime = Time.currentTimeSecs();
+ lastReturnedValue = ret;
+ }
+
+ private Map loadFromFile(File file) {
+ Map ret = null;
+
+ try {
+ ret = (Map) Utils.readYamlFile(file.getCanonicalPath());
+ } catch (IOException e) {
+ LOG.error("Filed to load from file. Exception: {}", e.getMessage());
+ }
+
+ if (ret != null) {
+ try {
+ LOG.debug("returning a new map from file {}", file.getCanonicalPath());
+ } catch (java.io.IOException e) {
+ LOG.debug("Could not get PATH from file object in debug print. Ignoring");
+ }
+ return ret;
+ }
+
+ return null;
+ }
+
+ private Map getLatestFromCache() {
+ String localFileName = localCacheDir + File.separator + cacheFilename;
+ return loadFromFile(new File(localFileName));
+ }
+
+ private void saveInArtifactoryCache(String yamlData) {
+ if (yamlData == null) {
+ LOG.warn("Will not save null data into the artifactory cache");
+ return;
+ }
+
+ String localFileName = localCacheDir + File.separator + cacheFilename;
+
+ File cacheFile = new File(localFileName);
+ try (FileOutputStream fos = new FileOutputStream(cacheFile)) {
+ fos.write(yamlData.getBytes());
+ fos.flush();
+ } catch (IOException e) {
+ LOG.error("Received exception when writing file {}. Attempting delete", localFileName, e);
+ try {
+ cacheFile.delete();
+ } catch (Exception deleteException) {
+ LOG.error("Received exception when deleting file {}.", localFileName, deleteException);
+ }
+ }
+ }
+
+ private void makeArtifactoryCache(String location) {
+ // Just make sure appropriate directories exist
+ String localDirName = (String) conf.get(Config.STORM_LOCAL_DIR);
+ if (localDirName == null) {
+ return;
+ }
+
+ // First make the cache dir
+ localDirName = localDirName + File.separator + "nimbus" + File.separator + LOCAL_ARTIFACT_DIR;
+ File dir = new File(localDirName);
+ if (! dir.exists()) {
+ dir.mkdirs();
+ }
+
+ localCacheDir = localDirName + File.separator + location.replaceAll(File.separator, "_");
+ dir = new File(localCacheDir);
+ if (! dir.exists()) {
+ dir.mkdir();
+ }
+ cacheInitialized = true;
+ }
+
+ private Map loadFromURI(URI uri) {
+ String host = uri.getHost();
+ Integer port = uri.getPort();
+ String location = uri.getPath();
+ if (location.toLowerCase().startsWith(baseDirectory.toLowerCase())) {
+ location = location.substring(baseDirectory.length());
+ }
+
+ if (!cacheInitialized) {
+ makeArtifactoryCache(location);
+ }
+
+ // Get the most recent artifact as a String, and then parse the yaml
+ String yamlConfig = loadMostRecentArtifact(location, host, port);
+
+ // If we failed to get anything from Artifactory try to get it from our local cache
+ if (yamlConfig == null) {
+ Map ret = getLatestFromCache();
+ updateLastReturned(ret);
+ return ret;
+ }
+
+ // Now parse it and return the map.
+ Yaml yaml = new Yaml(new SafeConstructor());
+ Map ret = null;
+ try {
+ ret = (Map) yaml.load(yamlConfig);
+ } catch (Exception e) {
+ LOG.error("Could not parse yaml.");
+ return null;
+ }
+
+ if (ret != null) {
+ LOG.debug("returning a new map from Artifactory");
+ updateLastReturned(ret);
+ return ret;
+ }
+
+ return null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderFactory.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderFactory.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderFactory.java
new file mode 100644
index 0000000..6494ae2
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderFactory.java
@@ -0,0 +1,50 @@
+/**
+ * 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.storm.scheduler.utils;
+
+import com.google.auto.service.AutoService;
+import java.net.URI;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory class for ArtifactoryConfigLoader.
+ */
+@AutoService(IConfigLoaderFactory.class)
+public class ArtifactoryConfigLoaderFactory implements IConfigLoaderFactory {
+ private static final Logger LOG = LoggerFactory.getLogger(ArtifactoryConfigLoaderFactory.class);
+
+ /**
+ * Create a ArtifactoryConfigLoader if the scheme of the URI is "artifactory+http" or "artifactory+https"; otherwise return null.
+ * @param uri The URI which pointing to the config file/directory location on Artifactory server.
+ * @param conf The storm configuration.
+ * @return A ArtifactoryConfigLoader if the scheme is "artifactory+http" or "artifactory+https"; otherwise, null.
+ */
+ @Override
+ public IConfigLoader createIfSupported(URI uri, Map<String, Object> conf) {
+ String scheme = uri.getScheme();
+ if ("artifactory+http".equalsIgnoreCase(scheme) || "artifactory+https".equalsIgnoreCase(scheme)) {
+ return new ArtifactoryConfigLoader(conf);
+ } else {
+ LOG.debug("scheme {} not supported in this factory.", scheme);
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/ConfigLoaderFactoryService.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/ConfigLoaderFactoryService.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/ConfigLoaderFactoryService.java
new file mode 100644
index 0000000..53df78b
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/ConfigLoaderFactoryService.java
@@ -0,0 +1,68 @@
+/**
+ * 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.storm.scheduler.utils;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+import java.util.ServiceLoader;
+import org.apache.storm.DaemonConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The user interface to create a concrete IConfigLoader instance for use.
+ */
+public class ConfigLoaderFactoryService {
+
+ private static Logger LOG = LoggerFactory.getLogger(IConfigLoaderFactory.class);
+
+ /**
+ * SerivceLoader loads all the implementations of IConfigLoaderFactory for use.
+ */
+ private static ServiceLoader<IConfigLoaderFactory> serviceLoader = ServiceLoader.load(IConfigLoaderFactory.class);
+
+ /**
+ * The user interface to create an IConfigLoader instance.
+ * It iterates all the implementations of IConfigLoaderFactory and finds the one which supports the
+ * specific scheme of the URI and then uses it to create an IConfigLoader instance.
+ * @param conf The storm configuration.
+ * @return A concrete IConfigLoader implementation which supports the scheme of the URI.
+ * If multiple implementations are available, return the first one; otherwise, return null.
+ */
+ public static IConfigLoader createConfigLoader(Map<String, Object> conf) {
+ String uriString = (String) conf.get(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI);
+ if (null != uriString) {
+ try {
+ URI uri = new URI(uriString);
+ for (IConfigLoaderFactory factory : serviceLoader) {
+ IConfigLoader ret = factory.createIfSupported(uri, conf);
+ if (ret != null) {
+ return ret;
+ }
+ }
+ } catch (URISyntaxException e) {
+ LOG.error("Failed to parse uri={}", uriString);
+ }
+ } else {
+ LOG.debug("Config {} is not set.", DaemonConfig.SCHEDULER_CONFIG_LOADER_URI);
+ }
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoader.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoader.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoader.java
new file mode 100644
index 0000000..e785e5c
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoader.java
@@ -0,0 +1,72 @@
+/**
+ * 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.storm.scheduler.utils;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
+import org.apache.storm.DaemonConfig;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Scheduler configuration loader which loads configs from a file.
+ */
+public class FileConfigLoader implements IConfigLoader {
+ private static final Logger LOG = LoggerFactory.getLogger(FileConfigLoader.class);
+
+ private Map<String, Object> conf;
+ private String targetFilePath = null;
+
+ public FileConfigLoader(Map<String, Object> conf) {
+ this.conf = conf;
+ String uriString = (String) conf.get(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI);
+ if (uriString == null) {
+ LOG.error("No URI defined in {} configuration.", DaemonConfig.SCHEDULER_CONFIG_LOADER_URI);
+ } else {
+ try {
+ targetFilePath = new URI(uriString).getPath();
+ } catch (URISyntaxException e) {
+ LOG.error("Failed to parse uri={}", uriString);
+ }
+ }
+ }
+
+ /**
+ * Load the configs associated with the configKey from the targetFilePath.
+ * @param configKey The key from which we want to get the scheduler config.
+ * @return The scheduler configuration if exists; null otherwise.
+ */
+ @Override
+ public Map load(String configKey) {
+ if (targetFilePath != null) {
+ try {
+ Map raw = (Map) Utils.readYamlFile(targetFilePath);
+ if (raw != null) {
+ return (Map) raw.get(configKey);
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to load from file {}", targetFilePath);
+ }
+ }
+ return null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoaderFactory.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoaderFactory.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoaderFactory.java
new file mode 100644
index 0000000..9a2ac5c
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/FileConfigLoaderFactory.java
@@ -0,0 +1,51 @@
+/**
+ * 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.storm.scheduler.utils;
+
+import com.google.auto.service.AutoService;
+import java.net.URI;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Factory class for FileConfigLoader.
+ */
+@AutoService(IConfigLoaderFactory.class)
+public class FileConfigLoaderFactory implements IConfigLoaderFactory {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileConfigLoaderFactory.class);
+
+ /**
+ * Create a FileConfigLoader if the scheme of the URI is "file"; else return null.
+ * @param uri The URI which pointing to the config file location.
+ * @param conf The storm configuration.
+ * @return A FileConfigLoader if the scheme is "file"; otherwise, null.
+ */
+ @Override
+ public IConfigLoader createIfSupported(URI uri, Map<String, Object> conf) {
+ String scheme = uri.getScheme();
+ if ("file".equalsIgnoreCase(scheme)) {
+ return new FileConfigLoader(conf);
+ } else {
+ LOG.debug("scheme {} not supported in this factory.", scheme);
+ return null;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoader.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoader.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoader.java
new file mode 100644
index 0000000..5071133
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoader.java
@@ -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.storm.scheduler.utils;
+
+import java.util.Map;
+
+public interface IConfigLoader {
+
+ /**
+ * Load scheduler configs associated with the configKey.
+ * @param configKey The key from which we want to get the scheduler config.
+ * @return The scheduler configs
+ */
+ Map<?,?> load(String configKey);
+
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoaderFactory.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoaderFactory.java b/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoaderFactory.java
new file mode 100644
index 0000000..a039870
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/scheduler/utils/IConfigLoaderFactory.java
@@ -0,0 +1,35 @@
+/**
+ * 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.storm.scheduler.utils;
+
+import java.net.URI;
+import java.util.Map;
+
+public interface IConfigLoaderFactory {
+
+ /**
+ * Create an IConfigLoader implementation if the scheme of the URI is supported; otherwise returns null.
+ *
+ * @param uri The URI of the config location.
+ * @param conf The storm configuration.
+ * @return An concrete implementation if the scheme is supported, or null if not.
+ */
+ IConfigLoader createIfSupported(URI uri, Map<String, Object> conf);
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/test/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderTest.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderTest.java b/storm-server/src/test/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderTest.java
new file mode 100644
index 0000000..009f479
--- /dev/null
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/utils/ArtifactoryConfigLoaderTest.java
@@ -0,0 +1,235 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.storm.scheduler.utils;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
+import org.apache.storm.utils.Time;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.Before;
+import org.junit.After;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Files;
+import java.util.HashMap;
+import java.util.Map;
+
+public class ArtifactoryConfigLoaderTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ArtifactoryConfigLoaderTest.class);
+ private Path tmpDirPath;
+ private static final String ARTIFACTORY_HTTP_SCHEME_PREFIX = "artifactory+http://";
+
+ private class ArtifactoryConfigLoaderMock extends ArtifactoryConfigLoader {
+ String getData;
+ HashMap<String, String> getDataMap = new HashMap<String, String>();
+
+ public ArtifactoryConfigLoaderMock(Map conf) {
+ super(conf);
+ }
+
+ public void setData(String api, String artifact, String data) {
+ if (api == null) {
+ getData = data;
+ } else {
+ getDataMap.put(artifact, data);
+ }
+ }
+
+ @Override
+ protected String doGet(String api, String artifact, String host, Integer port) {
+ if (api == null) {
+ return getData;
+ }
+ return getDataMap.get(artifact);
+ }
+ };
+
+ @Before
+ public void createTempDir() throws Exception {
+ tmpDirPath = Files.createTempDirectory("TestArtifactoryConfigLoader");
+ File f = tmpDirPath.toFile();
+ f.mkdir();
+ File dir = new File(f, "nimbus");
+ dir.mkdir();
+ }
+
+ @After
+ public void removeTempDir() throws Exception {
+ FileUtils.deleteDirectory(tmpDirPath.toFile());
+ }
+
+ @Test
+ public void testInvalidConfig() {
+ Config conf = new Config();
+ ArtifactoryConfigLoaderMock loaderMock = new ArtifactoryConfigLoaderMock(conf);
+ Map ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNull("Unexpectedly returned not null", ret);
+ }
+
+ @Test
+ public void testPointingAtDirectory() {
+ // This is a test where we are configured to point right at an artifact dir
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, ARTIFACTORY_HTTP_SCHEME_PREFIX + "bogushost.yahoo.com:9999/location/of/this/dir");
+ conf.put(Config.STORM_LOCAL_DIR, tmpDirPath.toString());
+
+ ArtifactoryConfigLoaderMock loaderMock = new ArtifactoryConfigLoaderMock(conf);
+
+ loaderMock.setData("Anything", "/location/of/this/dir",
+ "{\"children\" : [ { \"uri\" : \"/20160621204337.yaml\", \"folder\" : false }]}" );
+ loaderMock.setData(null, null, "{ \"" + DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS + "\": {one: 1, two: 2, three: 3, four : 4}}");
+
+ Map ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret);
+ Assert.assertEquals(1, ret.get("one"));
+ Assert.assertEquals(2, ret.get("two"));
+ Assert.assertEquals(3, ret.get("three"));
+ Assert.assertEquals(4, ret.get("four"));
+
+ // Now let's load w/o setting up gets and we should still get valid map back
+ ArtifactoryConfigLoaderMock tc2 = new ArtifactoryConfigLoaderMock(conf);
+
+ Map ret2 = tc2.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret2);
+ Assert.assertEquals(1, ret2.get("one"));
+ Assert.assertEquals(2, ret2.get("two"));
+ Assert.assertEquals(3, ret2.get("three"));
+ Assert.assertEquals(4, ret2.get("four"));
+ }
+
+ @Test
+ public void testArtifactUpdate() {
+ // This is a test where we are configured to point right at an artifact dir
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, ARTIFACTORY_HTTP_SCHEME_PREFIX + "bogushost.yahoo.com:9999/location/of/test/dir");
+ conf.put(Config.STORM_LOCAL_DIR, tmpDirPath.toString());
+
+ Time.startSimulating();
+
+ try {
+ ArtifactoryConfigLoaderMock loaderMock = new ArtifactoryConfigLoaderMock(conf);
+
+ loaderMock.setData("Anything", "/location/of/test/dir",
+ "{\"children\" : [ { \"uri\" : \"/20160621204337.yaml\", \"folder\" : false }]}" );
+ loaderMock.setData(null, null, "{ \"" + DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS + "\": {one: 1, two: 2, three: 3}}");
+ Map ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+
+ Assert.assertNotNull("Unexpectedly returned null", ret);
+ Assert.assertEquals(1, ret.get("one"));
+ Assert.assertEquals(2, ret.get("two"));
+ Assert.assertEquals(3, ret.get("three"));
+ Assert.assertNull("Unexpectedly contained \"four\"", ret.get("four"));
+
+ // Now let's load w/o setting up gets and we should still get valid map back
+ ArtifactoryConfigLoaderMock tc2 = new ArtifactoryConfigLoaderMock(conf);
+ Map ret2 = tc2.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret2);
+ Assert.assertEquals(1, ret2.get("one"));
+ Assert.assertEquals(2, ret2.get("two"));
+ Assert.assertEquals(3, ret2.get("three"));
+ Assert.assertNull("Unexpectedly did not return null", ret2.get("four"));
+
+ // Now let's update it, but not advance time. Should get old map again.
+ loaderMock.setData("Anything", "/location/of/test/dir",
+ "{\"children\" : [ { \"uri\" : \"/20160621204999.yaml\", \"folder\" : false }]}");
+ loaderMock.setData(null, null, "{ \"" + DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS + "\": {one: 1, two: 2, three: 3, four : 4}}");
+ ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret);
+ Assert.assertEquals(1, ret.get("one"));
+ Assert.assertEquals(2, ret.get("two"));
+ Assert.assertEquals(3, ret.get("three"));
+ Assert.assertNull("Unexpectedly did not return null, not enough time passed!", ret.get("four"));
+
+ // Re-load from cached' file.
+ ret2 = tc2.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret2);
+ Assert.assertEquals(1, ret2.get("one"));
+ Assert.assertEquals(2, ret2.get("two"));
+ Assert.assertEquals(3, ret2.get("three"));
+ Assert.assertNull("Unexpectedly did not return null, last cached result should not have \"four\"", ret2.get("four"));
+
+ // Now, let's advance time.
+ Time.advanceTime(11*60*1000);
+
+ ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret);
+ Assert.assertEquals(1, ret.get("one"));
+ Assert.assertEquals(2, ret.get("two"));
+ Assert.assertEquals(3, ret.get("three"));
+ Assert.assertEquals(4, ret.get("four"));
+
+ // Re-load from cached' file.
+ ret2 = tc2.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret2);
+ Assert.assertEquals(1, ret2.get("one"));
+ Assert.assertEquals(2, ret2.get("two"));
+ Assert.assertEquals(3, ret2.get("three"));
+ Assert.assertEquals(4, ret2.get("four"));
+ } finally {
+ Time.stopSimulating();
+ }
+ }
+
+ @Test
+ public void testPointingAtSpecificArtifact() {
+ // This is a test where we are configured to point right at a single artifact
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, ARTIFACTORY_HTTP_SCHEME_PREFIX + "bogushost.yahoo.com:9999/location/of/this/artifact");
+ conf.put(Config.STORM_LOCAL_DIR, tmpDirPath.toString());
+
+ ArtifactoryConfigLoaderMock loaderMock = new ArtifactoryConfigLoaderMock(conf);
+
+ loaderMock.setData("Anything", "/location/of/this/artifact", "{ \"downloadUri\": \"anything\"}");
+ loaderMock.setData(null, null, "{ \"" + DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS + "\": {one: 1, two: 2, three: 3}}");
+ Map ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+
+ Assert.assertNotNull("Unexpectedly returned null", ret);
+ Assert.assertEquals(1, ret.get("one"));
+ Assert.assertEquals(2, ret.get("two"));
+ Assert.assertEquals(3, ret.get("three"));
+
+ // Now let's load w/o setting up gets and we should still get valid map back
+ ArtifactoryConfigLoaderMock tc2 = new ArtifactoryConfigLoaderMock(conf);
+ Map ret2 = tc2.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNotNull("Unexpectedly returned null", ret2);
+ Assert.assertEquals(1, ret2.get("one"));
+ Assert.assertEquals(2, ret2.get("two"));
+ Assert.assertEquals(3, ret2.get("three"));
+ }
+
+ @Test
+ public void testMalformedYaml() throws Exception {
+ // This is a test where we are configured to point right at a single artifact
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, ARTIFACTORY_HTTP_SCHEME_PREFIX + "bogushost.yahoo.com:9999/location/of/this/artifact");
+ conf.put(Config.STORM_LOCAL_DIR, tmpDirPath.toString());
+
+ ArtifactoryConfigLoaderMock loaderMock = new ArtifactoryConfigLoaderMock(conf);
+ loaderMock.setData("Anything", "/location/of/this/artifact", "{ \"downloadUri\": \"anything\"}");
+ loaderMock.setData(null, null, "ThisIsNotValidYaml");
+
+ Map ret = loaderMock.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNull("Unexpectedly returned a map", ret);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/storm/blob/f1f95449/storm-server/src/test/java/org/apache/storm/scheduler/utils/FileConfigLoaderTest.java
----------------------------------------------------------------------
diff --git a/storm-server/src/test/java/org/apache/storm/scheduler/utils/FileConfigLoaderTest.java b/storm-server/src/test/java/org/apache/storm/scheduler/utils/FileConfigLoaderTest.java
new file mode 100644
index 0000000..33df1a1
--- /dev/null
+++ b/storm-server/src/test/java/org/apache/storm/scheduler/utils/FileConfigLoaderTest.java
@@ -0,0 +1,115 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.storm.scheduler.utils;
+
+import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
+import org.apache.storm.utils.Time;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.File;
+import java.io.FileWriter;
+import java.util.HashMap;
+import java.util.Map;
+import org.yaml.snakeyaml.Yaml;
+
+public class FileConfigLoaderTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FileConfigLoaderTest.class);
+
+ private static final String FILE_SCHEME_PREFIX = "file://";
+
+ @Test
+ public void testFileNotThere() {
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, FILE_SCHEME_PREFIX + "/file/not/exist/");
+ FileConfigLoader testLoader = new FileConfigLoader(conf);
+ Map result = testLoader.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNull("Unexpectedly returned a map", result);
+ }
+
+ @Test
+ public void testInvalidConfig() throws Exception {
+ Config conf = new Config();
+ FileConfigLoader testLoader = new FileConfigLoader(conf);
+ Map result = testLoader.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNull("Unexpectedly returned a map", result);
+ }
+
+ @Test
+ public void testMalformedYaml() throws Exception {
+
+ File temp = File.createTempFile("FileLoader", ".yaml");
+ temp.deleteOnExit();
+
+ FileWriter fw = new FileWriter(temp);
+ String outputData = "ThisIsNotValidYaml";
+ fw.write(outputData, 0, outputData.length());
+ fw.flush();
+ fw.close();
+
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, FILE_SCHEME_PREFIX + temp.getCanonicalPath());
+
+ FileConfigLoader testLoader = new FileConfigLoader(conf);
+ Map result = testLoader.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+ Assert.assertNull("Unexpectedly returned a map", result);
+ }
+
+ @Test
+ public void testValidFile() throws Exception {
+
+ File temp = File.createTempFile("FileLoader", ".yaml");
+ temp.deleteOnExit();
+
+ Map<String, Integer> testMap = new HashMap<String, Integer>();
+ testMap.put("a", 1);
+ testMap.put("b", 2);
+ testMap.put("c", 3);
+ testMap.put("d", 4);
+ testMap.put("e", 5);
+
+ Map<String, Map<String, Integer>> confMap = new HashMap<>();
+ confMap.put(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS, testMap);
+
+ Yaml yaml = new Yaml();
+ FileWriter fw = new FileWriter(temp);
+ yaml.dump(confMap, fw);
+ fw.flush();
+ fw.close();
+
+ Config conf = new Config();
+ conf.put(DaemonConfig.SCHEDULER_CONFIG_LOADER_URI, FILE_SCHEME_PREFIX + temp.getCanonicalPath());
+ FileConfigLoader loader = new FileConfigLoader(conf);
+
+ Map result = loader.load(DaemonConfig.MULTITENANT_SCHEDULER_USER_POOLS);
+
+ Assert.assertNotNull("Unexpectedly returned null", result);
+
+ Assert.assertEquals("Maps are a different size", testMap.keySet().size(), result.keySet().size());
+
+ for (String key : testMap.keySet() ) {
+ Integer expectedValue = testMap.get(key);
+ Integer returnedValue = (Integer) result.get(key);
+ Assert.assertEquals("Bad value for key=" + key, expectedValue, returnedValue);
+ }
+ }
+}
\ No newline at end of file
[2/2] storm git commit: Merge branch 'STORM-2201' of
https://github.com/Ethanlm/storm into STORM-2201-merge
Posted by ka...@apache.org.
Merge branch 'STORM-2201' of https://github.com/Ethanlm/storm into STORM-2201-merge
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/e1dd247c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/e1dd247c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/e1dd247c
Branch: refs/heads/master
Commit: e1dd247ce97ad560c72554ed612e1ec8e69e9777
Parents: 08d9d5a f1f9544
Author: Jungtaek Lim <ka...@gmail.com>
Authored: Wed Aug 30 10:41:07 2017 +0900
Committer: Jungtaek Lim <ka...@gmail.com>
Committed: Wed Aug 30 10:41:07 2017 +0900
----------------------------------------------------------------------
docs/IConfigLoader.md | 51 +++
pom.xml | 7 +
storm-server/pom.xml | 5 +
.../java/org/apache/storm/DaemonConfig.java | 94 +++--
.../multitenant/MultitenantScheduler.java | 62 ++-
.../resource/ResourceAwareScheduler.java | 89 +++--
.../utils/ArtifactoryConfigLoader.java | 395 +++++++++++++++++++
.../utils/ArtifactoryConfigLoaderFactory.java | 50 +++
.../utils/ConfigLoaderFactoryService.java | 68 ++++
.../storm/scheduler/utils/FileConfigLoader.java | 72 ++++
.../utils/FileConfigLoaderFactory.java | 51 +++
.../storm/scheduler/utils/IConfigLoader.java | 32 ++
.../scheduler/utils/IConfigLoaderFactory.java | 35 ++
.../utils/ArtifactoryConfigLoaderTest.java | 235 +++++++++++
.../scheduler/utils/FileConfigLoaderTest.java | 115 ++++++
15 files changed, 1275 insertions(+), 86 deletions(-)
----------------------------------------------------------------------