You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@slider.apache.org by sm...@apache.org on 2014/08/06 05:12:46 UTC
[2/6] git commit: SLIDER-273. Add ability to specify a port as shared
across component instances or not
SLIDER-273. Add ability to specify a port as shared across component instances or not
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/50dc76c9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/50dc76c9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/50dc76c9
Branch: refs/heads/develop
Commit: 50dc76c978ef4ff684263b93e8b56619cc947284
Parents: dca4818
Author: Sumit Mohanty <sm...@hortonworks.com>
Authored: Sun Aug 3 19:30:13 2014 -0700
Committer: Sumit Mohanty <sm...@hortonworks.com>
Committed: Tue Aug 5 18:51:58 2014 -0700
----------------------------------------------------------------------
app-packages/memcached/appConfig.json | 2 +-
.../python/agent/CustomServiceOrchestrator.py | 27 +++-
.../agent/TestCustomServiceOrchestrator.py | 18 ++-
.../providers/AbstractProviderService.java | 5 +
.../providers/agent/AgentProviderService.java | 126 ++++++++++++-------
.../model/mock/MockProviderService.groovy | 5 +
.../agent/TestAgentProviderService.java | 58 +++++++--
.../publisher/TestAgentProviderService.java | 4 +-
8 files changed, 185 insertions(+), 60 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/app-packages/memcached/appConfig.json
----------------------------------------------------------------------
diff --git a/app-packages/memcached/appConfig.json b/app-packages/memcached/appConfig.json
index 11fb863..78cb486 100644
--- a/app-packages/memcached/appConfig.json
+++ b/app-packages/memcached/appConfig.json
@@ -11,7 +11,7 @@
"site.global.xmx_val": "256m",
"site.global.xms_val": "128m",
"site.global.memory_val": "200M",
- "site.global.listen_port": "${MEMCACHED.ALLOCATED_PORT}{DEFAULT_11211}"
+ "site.global.listen_port": "${MEMCACHED.ALLOCATED_PORT}{DEFAULT_11211}{DO_NOT_PROPAGATE}"
},
"components": {
"slider-appmaster": {
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py b/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
index f6cdfb2..0f4acc6 100644
--- a/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
+++ b/slider-agent/src/main/python/agent/CustomServiceOrchestrator.py
@@ -233,12 +233,17 @@ class CustomServiceOrchestrator():
patch content
${AGENT_WORK_ROOT} -> AgentConfig.getWorkRootPath()
${AGENT_LOG_ROOT} -> AgentConfig.getLogPath()
+ ALLOCATED_PORT is a hint to allocate port. It works as follows:
+ Its of the form {component_name.ALLOCATED_PORT}[{DEFAULT_default_port}][{DO_NOT_PROPAGATE}]
+ Either a port gets allocated or if not then just set the value to "0"
"""
def finalize_command(self, command, store_config, allocated_ports):
component = command['componentName']
- allocated_port_format = "${{{0}.ALLOCATED_PORT}}"
- port_allocation_req = allocated_port_format.format(component)
+ allocated_for_this_component_format = "${{{0}.ALLOCATED_PORT}}"
+ allocated_for_any = ".ALLOCATED_PORT}"
+
+ port_allocation_req = allocated_for_this_component_format.format(component)
if 'configurations' in command:
for key in command['configurations']:
if len(command['configurations'][key]) > 0:
@@ -251,6 +256,10 @@ class CustomServiceOrchestrator():
if port_allocation_req in value:
value = self.allocate_ports(value, port_allocation_req)
allocated_ports[key + "." + k] = value
+ elif allocated_for_any in value:
+ ## All unallocated ports should be set to 0
+ logger.info("Assigning port 0 " + "for " + value)
+ value = "0"
command['configurations'][key][k] = value
pass
pass
@@ -264,8 +273,15 @@ class CustomServiceOrchestrator():
pass
+ """
+ Port allocation can asks for multiple dynamic ports
+ port_req_pattern is of type ${component_name.ALLOCATED_PORT}
+ append {DEFAULT_ and find the default value
+ append {DO_NOT_PROPAGATE} if it exists
+ """
def allocate_ports(self, value, port_req_pattern):
default_port_pattern = "{DEFAULT_"
+ do_not_propagate_pattern = "{DO_NOT_PROPAGATE}"
index = value.find(port_req_pattern)
while index != -1:
replaced_pattern = port_req_pattern
@@ -276,8 +292,15 @@ class CustomServiceOrchestrator():
end_index = value.find("}", start_index)
def_port_str = value[start_index:end_index]
def_port = int(def_port_str)
+ # default value of 0 means allocate any dynamic port
+ if def_port == 0:
+ def_port = None
+
replaced_pattern = replaced_pattern + def_port_str + "}"
pass
+ if index == value.find(replaced_pattern + do_not_propagate_pattern):
+ replaced_pattern = replaced_pattern + do_not_propagate_pattern
+ pass
port = self.allocate_port(def_port)
value = value.replace(replaced_pattern, str(port), 1)
logger.info("Allocated port " + str(port) + " for " + replaced_pattern)
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
----------------------------------------------------------------------
diff --git a/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py b/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
index 3343a8b..c07f05f 100644
--- a/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
+++ b/slider-agent/src/test/python/agent/TestCustomServiceOrchestrator.py
@@ -284,8 +284,9 @@ class TestCustomServiceOrchestrator(TestCase):
self.assertEqual(ret, 102)
+ @patch.object(CustomServiceOrchestrator, "is_port_available")
@patch.object(CustomServiceOrchestrator, "allocate_port")
- def test_allocate_port_combinations(self, allocate_port_mock):
+ def test_allocate_port_combinations(self, allocate_port_mock, is_port_available_mock):
tempdir = tempfile.gettempdir()
config = MagicMock()
config.get.return_value = "something"
@@ -296,13 +297,20 @@ class TestCustomServiceOrchestrator(TestCase):
dummy_controller = MagicMock()
orchestrator = CustomServiceOrchestrator(config, dummy_controller)
- allocate_port_mock.side_effect = [101, 102, 103]
+ is_port_available_mock.return_value = False
+ allocate_port_mock.side_effect = [101, 102, 103, 104, 105, 106]
ret = orchestrator.allocate_ports("1000", "${A.ALLOCATED_PORT}")
self.assertEqual(ret, "1000")
ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}", "${A.ALLOCATED_PORT}")
self.assertEqual(ret, "101")
ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT},${A.ALLOCATED_PORT}", "${A.ALLOCATED_PORT}")
self.assertEqual(ret, "102,103")
+ ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}{DEFAULT_0}", "${A.ALLOCATED_PORT}")
+ self.assertEqual(ret, "104")
+ ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}{DEFAULT_0}{DO_NOT_PROPAGATE}", "${A.ALLOCATED_PORT}")
+ self.assertEqual(ret, "105")
+ ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}{DO_NOT_PROPAGATE}", "${A.ALLOCATED_PORT}")
+ self.assertEqual(ret, "106")
@patch.object(CustomServiceOrchestrator, "is_port_available")
@@ -320,10 +328,14 @@ class TestCustomServiceOrchestrator(TestCase):
is_port_available_mock.return_value = True
ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}{DEFAULT_1005}", "${A.ALLOCATED_PORT}")
self.assertEqual(ret, "1005")
+
ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}{DEFAULT_1005}-${A.ALLOCATED_PORT}{DEFAULT_1006}",
"${A.ALLOCATED_PORT}")
self.assertEqual(ret, "1005-1006")
+ ret = orchestrator.allocate_ports("${A.ALLOCATED_PORT}{DEFAULT_1006}{DO_NOT_PROPAGATE}",
+ "${A.ALLOCATED_PORT}")
+ self.assertEqual(ret, "1006")
@patch("hostname.public_hostname")
@patch("os.path.isfile")
@@ -492,7 +504,7 @@ class TestCustomServiceOrchestrator(TestCase):
orchestrator.finalize_command(command, True, {})
self.assertEqual(command['configurations']['hbase-site']['log_root'], tempdir + "/log")
self.assertEqual(command['configurations']['hbase-site']['blog_root'], "/b/" + tempdir + "/log")
- self.assertEqual(command['configurations']['oozie-site']['b_port'], "${HBASE_REGIONSERVER.ALLOCATED_PORT}")
+ self.assertEqual(command['configurations']['oozie-site']['b_port'], "0")
self.assertEqual(orchestrator.applied_configs, command['configurations'])
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-core/src/main/java/org/apache/slider/providers/AbstractProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/providers/AbstractProviderService.java b/slider-core/src/main/java/org/apache/slider/providers/AbstractProviderService.java
index e35227c..7639625 100644
--- a/slider-core/src/main/java/org/apache/slider/providers/AbstractProviderService.java
+++ b/slider-core/src/main/java/org/apache/slider/providers/AbstractProviderService.java
@@ -20,6 +20,7 @@ package org.apache.slider.providers;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.Service;
+import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.slider.api.ClusterDescription;
import org.apache.slider.common.SliderKeys;
import org.apache.slider.common.tools.ConfigHelper;
@@ -104,6 +105,10 @@ public abstract class AbstractProviderService
return restOps;
}
+ @Override
+ public void notifyContainerCompleted(ContainerId containerId) {
+ }
+
public void setAgentRestOperations(AgentRestOperations agentRestOperations) {
this.restOps = agentRestOperations;
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java b/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
index bbe2001..28c874b 100644
--- a/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
+++ b/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java
@@ -111,11 +111,12 @@ public class AgentProviderService extends AbstractProviderService implements
private static final String GLOBAL_CONFIG_TAG = "global";
private static final String LOG_FOLDERS_TAG = "LogFolders";
private static final String COMPONENT_DATA_TAG = "ComponentInstanceData";
- private static final String SHARED_PORT_TAG = "{SHARED}";
+ private static final String SHARED_PORT_TAG = "SHARED";
+ private static final String DO_NOT_PROPAGATE_TAG = "{DO_NOT_PROPAGATE}";
private static final int MAX_LOG_ENTRIES = 20;
private static final int DEFAULT_HEARTBEAT_MONITOR_INTERVAL = 60 * 1000;
private final Object syncLock = new Object();
- private final Map<String, String> allocatedPorts = new ConcurrentHashMap<>();
+ private final Map<String, Map<String, String>> allocatedPorts = new ConcurrentHashMap<>();
private int heartbeatMonitorInterval = 0;
private AgentClientProvider clientProvider;
private Map<String, ComponentInstanceState> componentStatuses = new ConcurrentHashMap<>();
@@ -381,11 +382,12 @@ public class AgentProviderService extends AbstractProviderService implements
if (ports != null && !ports.isEmpty()) {
for (Map.Entry<String, String> port : ports.entrySet()) {
log.info("Recording allocated port for {} as {}", port.getKey(), port.getValue());
- this.allocatedPorts.put(port.getKey(), port.getValue());
+ this.getAllocatedPorts().put(port.getKey(), port.getValue());
+ this.getAllocatedPorts(containerId).put(port.getKey(), port.getValue());
}
// component specific publishes
- processComponentSpecificPublishes(ports, containerId, heartBeat.getFqdn(), roleName);
+ processAndPublishComponentSpecificData(ports, containerId, heartBeat.getFqdn(), roleName);
}
CommandResult result = CommandResult.getCommandResult(report.getStatus());
Command command = Command.getCommand(report.getRoleCommand());
@@ -469,8 +471,21 @@ public class AgentProviderService extends AbstractProviderService implements
@Override
public void notifyContainerCompleted(ContainerId containerId) {
- if(containerId != null) {
+ if (containerId != null) {
String containerIdStr = containerId.toString();
+ if (this.componentInstanceData.containsKey(containerIdStr)) {
+ synchronized (this.componentInstanceData) {
+ this.componentInstanceData.remove(containerIdStr);
+ }
+ log.info("Removing container specific data for {}", containerIdStr);
+ publishComponentInstanceData();
+ }
+
+ if (this.allocatedPorts.containsKey(containerIdStr)) {
+ synchronized (this.allocatedPorts) {
+ this.allocatedPorts.remove(containerIdStr);
+ }
+ }
}
}
@@ -534,8 +549,8 @@ public class AgentProviderService extends AbstractProviderService implements
* @param description
* @param entries
*/
- protected void publishComponentConfiguration(String name, String description,
- Iterable<Map.Entry<String, String>> entries) {
+ protected void publishApplicationInstanceData(String name, String description,
+ Iterable<Map.Entry<String, String>> entries) {
PublishedConfiguration pubconf = new PublishedConfiguration();
pubconf.description = description;
pubconf.putValues(entries);
@@ -606,7 +621,7 @@ public class AgentProviderService extends AbstractProviderService implements
workFolders.put(String.format("%s-%s-%s", hostFqdn, containerId, key), folders.get(key));
}
- publishComponentConfiguration(LOG_FOLDERS_TAG, LOG_FOLDERS_TAG, (new HashMap<>(this.workFolders)).entrySet());
+ publishApplicationInstanceData(LOG_FOLDERS_TAG, LOG_FOLDERS_TAG, (new HashMap<>(this.workFolders)).entrySet());
}
@@ -627,7 +642,7 @@ public class AgentProviderService extends AbstractProviderService implements
if (status.getConfigs() != null) {
for (String key : status.getConfigs().keySet()) {
Map<String, String> configs = status.getConfigs().get(key);
- publishComponentConfiguration(key, key, configs.entrySet());
+ publishApplicationInstanceData(key, key, configs.entrySet());
}
if (hasExportGroups) {
@@ -665,7 +680,7 @@ public class AgentProviderService extends AbstractProviderService implements
map.put(export.getName(), value);
log.info("Preparing to publish. Key {} and Value {}", export.getName(), value);
}
- publishComponentConfiguration(exportGroupName, exportGroupName, map.entrySet());
+ publishApplicationInstanceData(exportGroupName, exportGroupName, map.entrySet());
}
}
}
@@ -676,10 +691,10 @@ public class AgentProviderService extends AbstractProviderService implements
}
/** Publish component instance specific data if the component demands it */
- protected void processComponentSpecificPublishes(Map<String, String> ports,
- String containerId,
- String hostFqdn,
- String roleName) {
+ protected void processAndPublishComponentSpecificData(Map<String, String> ports,
+ String containerId,
+ String hostFqdn,
+ String roleName) {
String portVarFormat = "${site.%s}";
String hostNamePattern = "${THIS_HOST}";
Map<String, String> toPublish = new HashMap<>();
@@ -687,23 +702,25 @@ public class AgentProviderService extends AbstractProviderService implements
Application application = getMetainfo().getApplication();
for (Component component : application.getComponents()) {
if (component.getName().equals(roleName)) {
- if(component.getExports().size() > 0) {
+ if (component.getExports().size() > 0) {
- for(Export export : component.getExports()) {
+ for (Export export : component.getExports()) {
String templateToExport = export.getValue();
- for(String portName : ports.keySet()) {
+ for (String portName : ports.keySet()) {
boolean publishData = false;
String portValPattern = String.format(portVarFormat, portName);
- if(templateToExport.contains(portValPattern)) {
+ if (templateToExport.contains(portValPattern)) {
templateToExport = templateToExport.replace(portValPattern, ports.get(portName));
publishData = true;
}
- if(templateToExport.contains(hostNamePattern)) {
+ if (templateToExport.contains(hostNamePattern)) {
templateToExport = templateToExport.replace(hostNamePattern, hostFqdn);
publishData = true;
}
- if(publishData) {
+ if (publishData) {
toPublish.put(export.getName(), templateToExport);
+ log.info("Publishing {} for name {} and container {}",
+ templateToExport, export.getName(), containerId);
}
}
}
@@ -711,26 +728,30 @@ public class AgentProviderService extends AbstractProviderService implements
}
}
- if(toPublish.size() > 0) {
+ if (toPublish.size() > 0) {
Map<String, String> perContainerData = null;
- if(!this.componentInstanceData.containsKey(containerId)) {
+ if (!this.componentInstanceData.containsKey(containerId)) {
perContainerData = new ConcurrentHashMap<>();
} else {
perContainerData = this.componentInstanceData.get(containerId);
}
perContainerData.putAll(toPublish);
this.componentInstanceData.put(containerId, perContainerData);
- Map<String, String> dataToPublish = new HashMap<>();
- synchronized (this.componentInstanceData) {
- for(String container : this.componentInstanceData.keySet()) {
- for(String prop : this.componentInstanceData.get(container).keySet()) {
- dataToPublish.put(
- container + "." + prop, this.componentInstanceData.get(container).get(prop));
- }
+ publishComponentInstanceData();
+ }
+ }
+
+ private void publishComponentInstanceData() {
+ Map<String, String> dataToPublish = new HashMap<>();
+ synchronized (this.componentInstanceData) {
+ for (String container : this.componentInstanceData.keySet()) {
+ for (String prop : this.componentInstanceData.get(container).keySet()) {
+ dataToPublish.put(
+ container + "." + prop, this.componentInstanceData.get(container).get(prop));
}
}
- publishComponentConfiguration(COMPONENT_DATA_TAG, COMPONENT_DATA_TAG, dataToPublish.entrySet());
}
+ publishApplicationInstanceData(COMPONENT_DATA_TAG, COMPONENT_DATA_TAG, dataToPublish.entrySet());
}
/**
@@ -864,7 +885,7 @@ public class AgentProviderService extends AbstractProviderService implements
hostLevelParams.put(CONTAINER_ID, containerId);
cmd.setHostLevelParams(hostLevelParams);
- setInstallCommandConfigurations(cmd);
+ setInstallCommandConfigurations(cmd, containerId);
cmd.setCommandParams(setCommandParameters(scriptPath, false));
@@ -914,9 +935,9 @@ public class AgentProviderService extends AbstractProviderService implements
return cmdParams;
}
- private void setInstallCommandConfigurations(ExecutionCommand cmd) throws SliderException {
+ private void setInstallCommandConfigurations(ExecutionCommand cmd, String containerId) throws SliderException {
ConfTreeOperations appConf = getAmState().getAppConfSnapshot();
- Map<String, Map<String, String>> configurations = buildCommandConfigurations(appConf);
+ Map<String, Map<String, String>> configurations = buildCommandConfigurations(appConf, containerId);
cmd.setConfigurations(configurations);
}
@@ -943,7 +964,7 @@ public class AgentProviderService extends AbstractProviderService implements
cmd.setCommandParams(setCommandParameters(scriptPath, false));
- Map<String, Map<String, String>> configurations = buildCommandConfigurations(appConf);
+ Map<String, Map<String, String>> configurations = buildCommandConfigurations(appConf, containerId);
cmd.setConfigurations(configurations);
@@ -998,17 +1019,33 @@ public class AgentProviderService extends AbstractProviderService implements
cmd.setCommandParams(setCommandParameters(scriptPath, true));
- Map<String, Map<String, String>> configurations = buildCommandConfigurations(appConf);
+ Map<String, Map<String, String>> configurations = buildCommandConfigurations(appConf, containerId);
cmd.setConfigurations(configurations);
response.addExecutionCommand(cmd);
}
protected Map<String, String> getAllocatedPorts() {
- return this.allocatedPorts;
+ return getAllocatedPorts(SHARED_PORT_TAG);
}
- private Map<String, Map<String, String>> buildCommandConfigurations(ConfTreeOperations appConf)
+ protected Map<String, Map<String, String>> getComponentInstanceData() {
+ return this.componentInstanceData;
+ }
+
+ protected Map<String, String> getAllocatedPorts(String containerId) {
+ if (!this.allocatedPorts.containsKey(containerId)) {
+ synchronized (this.allocatedPorts) {
+ if (!this.allocatedPorts.containsKey(containerId)) {
+ this.allocatedPorts.put(containerId, new ConcurrentHashMap());
+ }
+ }
+ }
+ return this.allocatedPorts.get(containerId);
+ }
+
+ private Map<String, Map<String, String>> buildCommandConfigurations(
+ ConfTreeOperations appConf, String containerId)
throws SliderException {
Map<String, Map<String, String>> configurations = new TreeMap<>();
@@ -1019,7 +1056,7 @@ public class AgentProviderService extends AbstractProviderService implements
//Add global
for (String configType : configs) {
addNamedConfiguration(configType, appConf.getGlobalOptions().options,
- configurations, tokens);
+ configurations, tokens, containerId);
}
return configurations;
@@ -1057,7 +1094,7 @@ public class AgentProviderService extends AbstractProviderService implements
private void addNamedConfiguration(String configName, Map<String, String> sourceConfig,
Map<String, Map<String, String>> configurations,
- Map<String, String> tokens) {
+ Map<String, String> tokens, String containerId) {
Map<String, String> config = new HashMap<>();
if (configName.equals(GLOBAL_CONFIG_TAG)) {
addDefaultGlobalConfig(config);
@@ -1070,12 +1107,17 @@ public class AgentProviderService extends AbstractProviderService implements
if (!this.getAllocatedPorts().isEmpty()) {
for (String key : config.keySet()) {
String value = config.get(key);
- if(!value.contains(SHARED_PORT_TAG)) {
- // If the config property is not shared then do not pass on the already allocated value
- String lookupKey = configName + "." + key;
+ String lookupKey = configName + "." + key;
+ if(!value.contains(DO_NOT_PROPAGATE_TAG)) {
+ // If the config property is shared then pass on the already allocated value
+ // from any container
if (this.getAllocatedPorts().containsKey(lookupKey)) {
config.put(key, getAllocatedPorts().get(lookupKey));
}
+ } else {
+ if (this.getAllocatedPorts(containerId).containsKey(lookupKey)) {
+ config.put(key, getAllocatedPorts(containerId).get(lookupKey));
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockProviderService.groovy
----------------------------------------------------------------------
diff --git a/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockProviderService.groovy b/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockProviderService.groovy
index 7b73451..db28a22 100644
--- a/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockProviderService.groovy
+++ b/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockProviderService.groovy
@@ -23,6 +23,7 @@ import org.apache.hadoop.fs.Path
import org.apache.hadoop.service.LifecycleEvent
import org.apache.hadoop.service.ServiceStateChangeListener
import org.apache.hadoop.yarn.api.records.Container
+import org.apache.hadoop.yarn.api.records.ContainerId
import org.apache.slider.api.ClusterDescription
import org.apache.slider.common.tools.SliderFileSystem
import org.apache.slider.core.conf.AggregateConf
@@ -229,4 +230,8 @@ class MockProviderService implements ProviderService {
throws MalformedURLException, IOException {
}
+
+ @Override
+ public void notifyContainerCompleted(ContainerId containerId) {
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java b/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
index 1f9cb41..10b1eb7 100644
--- a/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
+++ b/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java
@@ -360,20 +360,20 @@ public class TestAgentProviderService {
Metainfo metainfo = new MetainfoParser().parse(metainfo_1);
AgentProviderService aps = new AgentProviderService();
AgentProviderService mockAps = Mockito.spy(aps);
- doNothing().when(mockAps).publishComponentConfiguration(anyString(), anyString(), anyCollection());
+ doNothing().when(mockAps).publishApplicationInstanceData(anyString(), anyString(), anyCollection());
doReturn(metainfo).when(mockAps).getMetainfo();
Map<String, String> ports = new HashMap<>();
ports.put("global.listen_port", "10010");
- mockAps.processComponentSpecificPublishes(ports,
- "cid1",
- "host1",
- "HBASE_REGIONSERVER");
+ mockAps.processAndPublishComponentSpecificData(ports,
+ "cid1",
+ "host1",
+ "HBASE_REGIONSERVER");
ArgumentCaptor<Collection> entriesCaptor = ArgumentCaptor.
forClass(Collection.class);
ArgumentCaptor<String> publishNameCaptor = ArgumentCaptor.
forClass(String.class);
- Mockito.verify(mockAps, Mockito.times(1)).publishComponentConfiguration(
+ Mockito.verify(mockAps, Mockito.times(1)).publishApplicationInstanceData(
anyString(),
publishNameCaptor.capture(),
entriesCaptor.capture());
@@ -422,7 +422,7 @@ public class TestAgentProviderService {
ComponentInstanceState componentStatus = new ComponentInstanceState("HBASE_MASTER", "aid", "cid");
AgentProviderService mockAps = Mockito.spy(aps);
- doNothing().when(mockAps).publishComponentConfiguration(anyString(), anyString(), anyCollection());
+ doNothing().when(mockAps).publishApplicationInstanceData(anyString(), anyString(), anyCollection());
doReturn(metainfo).when(mockAps).getMetainfo();
doReturn(roleClusterNodeMap).when(mockAps).getRoleClusterNodeMapping();
@@ -430,7 +430,7 @@ public class TestAgentProviderService {
assert componentStatus.getConfigReported() == true;
ArgumentCaptor<Collection> entriesCaptor = ArgumentCaptor.
forClass(Collection.class);
- Mockito.verify(mockAps, Mockito.times(3)).publishComponentConfiguration(
+ Mockito.verify(mockAps, Mockito.times(3)).publishApplicationInstanceData(
anyString(),
anyString(),
entriesCaptor.capture());
@@ -627,7 +627,7 @@ public class TestAgentProviderService {
anyString(),
anyString(),
any(HeartBeatResponse.class));
- doNothing().when(mockAps).publishComponentConfiguration(
+ doNothing().when(mockAps).publishApplicationInstanceData(
anyString(),
anyString(),
anyCollection());
@@ -797,13 +797,44 @@ public class TestAgentProviderService {
log.warn(he.getMessage());
}
- Mockito.verify(mockAps, Mockito.times(1)).publishComponentConfiguration(
+ Mockito.verify(mockAps, Mockito.times(1)).publishApplicationInstanceData(
anyString(),
anyString(),
anyCollection());
}
@Test
+ public void testNotifyContainerCompleted() {
+ AgentProviderService aps = new AgentProviderService();
+ AgentProviderService mockAps = Mockito.spy(aps);
+ doNothing().when(mockAps).publishApplicationInstanceData(anyString(), anyString(), anyCollection());
+
+ ContainerId cid = new MyContainerId(1);
+ String id = cid.toString();
+ mockAps.getAllocatedPorts().put("a", "100");
+ mockAps.getAllocatedPorts(id).put("b", "101");
+ mockAps.getAllocatedPorts("cid2").put("c", "102");
+
+ mockAps.getComponentInstanceData().put("cid2", new HashMap<String, String>());
+ mockAps.getComponentInstanceData().put(id, new HashMap<String, String>());
+
+ Assert.assertNotNull(mockAps.getComponentInstanceData().get(id));
+ Assert.assertNotNull(mockAps.getComponentInstanceData().get("cid2"));
+
+ Assert.assertEquals(mockAps.getAllocatedPorts().size(), 1);
+ Assert.assertEquals(mockAps.getAllocatedPorts(id).size(), 1);
+ Assert.assertEquals(mockAps.getAllocatedPorts("cid2").size(), 1);
+ mockAps.notifyContainerCompleted(new MyContainerId(1));
+
+ Assert.assertNull(mockAps.getComponentInstanceData().get(id));
+ Assert.assertNotNull(mockAps.getComponentInstanceData().get("cid2"));
+
+ Assert.assertEquals(mockAps.getAllocatedPorts().size(), 1);
+ Assert.assertEquals(mockAps.getAllocatedPorts(id).size(), 0);
+ Assert.assertEquals(mockAps.getAllocatedPorts("cid2").size(), 1);
+ }
+
+ @Test
public void testAddInstallCommand() throws Exception {
InputStream metainfo_1 = new ByteArrayInputStream(metainfo_1_str.getBytes());
Metainfo metainfo = new MetainfoParser().parse(metainfo_1);
@@ -867,6 +898,8 @@ public class TestAgentProviderService {
treeOps.set("config_types", "hbase-site");
treeOps.getGlobalOptions().put("site.hbase-site.a.port", "${HBASE_MASTER.ALLOCATED_PORT}");
treeOps.getGlobalOptions().put("site.hbase-site.b.port", "${HBASE_MASTER.ALLOCATED_PORT}");
+ treeOps.getGlobalOptions().put("site.hbase-site.random.port", "${HBASE_MASTER.ALLOCATED_PORT}{DO_NOT_PROPAGATE}");
+ treeOps.getGlobalOptions().put("site.hbase-site.random2.port", "${HBASE_MASTER.ALLOCATED_PORT}");
expect(access.getAppConfSnapshot()).andReturn(treeOps).anyTimes();
expect(access.getInternalsSnapshot()).andReturn(treeOps).anyTimes();
@@ -885,6 +918,9 @@ public class TestAgentProviderService {
allocatedPorts.put("hbase-site.a.port", "10023");
allocatedPorts.put("hbase-site.b.port", "10024");
doReturn(allocatedPorts).when(mockAps).getAllocatedPorts();
+ Map<String, String> allocatedPorts2 = new HashMap<>();
+ allocatedPorts2.put("hbase-site.random.port", "10025");
+ doReturn(allocatedPorts2).when(mockAps).getAllocatedPorts(anyString());
replay(access);
@@ -894,6 +930,8 @@ public class TestAgentProviderService {
Assert.assertTrue(hbaseSiteConf.containsKey("a.port"));
Assert.assertTrue(hbaseSiteConf.get("a.port").equals("10023"));
Assert.assertTrue(hbaseSiteConf.get("b.port").equals("10024"));
+ Assert.assertTrue(hbaseSiteConf.get("random.port").equals("10025"));
+ Assert.assertTrue(hbaseSiteConf.get("random2.port").equals("${HBASE_MASTER.ALLOCATED_PORT}"));
}
private static class MyContainer extends Container {
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/50dc76c9/slider-core/src/test/java/org/apache/slider/server/appmaster/web/rest/publisher/TestAgentProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/test/java/org/apache/slider/server/appmaster/web/rest/publisher/TestAgentProviderService.java b/slider-core/src/test/java/org/apache/slider/server/appmaster/web/rest/publisher/TestAgentProviderService.java
index 2427009..92e6491 100644
--- a/slider-core/src/test/java/org/apache/slider/server/appmaster/web/rest/publisher/TestAgentProviderService.java
+++ b/slider-core/src/test/java/org/apache/slider/server/appmaster/web/rest/publisher/TestAgentProviderService.java
@@ -46,8 +46,8 @@ public class TestAgentProviderService extends AgentProviderService{
dummyProps.put("prop1", "val1");
dummyProps.put("prop2", "val2");
log.info("publishing dummy-site.xml with values {}", dummyProps);
- publishComponentConfiguration("dummy-site", "dummy configuration",
- dummyProps.entrySet());
+ publishApplicationInstanceData("dummy-site", "dummy configuration",
+ dummyProps.entrySet());
}