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:50 UTC
[6/6] git commit: SLIDER-219. Multiple dynamic ports may be asked as
part of the same config value
SLIDER-219. Multiple dynamic ports may be asked as part of the same config value
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/dca4818c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/dca4818c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/dca4818c
Branch: refs/heads/develop
Commit: dca4818c551c8fff655c8898c0786aae2fb0fb6a
Parents: ad2c798
Author: Sumit Mohanty <sm...@hortonworks.com>
Authored: Fri Aug 1 11:52:35 2014 -0700
Committer: Sumit Mohanty <sm...@hortonworks.com>
Committed: Tue Aug 5 18:51:58 2014 -0700
----------------------------------------------------------------------
.../slider/providers/ProviderService.java | 7 +
.../providers/agent/AgentProviderService.java | 329 ++++++++++---------
.../server/appmaster/SliderAppMaster.java | 2 +
.../agent/TestAgentProviderService.java | 49 ++-
4 files changed, 224 insertions(+), 163 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/dca4818c/slider-core/src/main/java/org/apache/slider/providers/ProviderService.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/providers/ProviderService.java b/slider-core/src/main/java/org/apache/slider/providers/ProviderService.java
index 56e24e9..b9fa34c 100644
--- a/slider-core/src/main/java/org/apache/slider/providers/ProviderService.java
+++ b/slider-core/src/main/java/org/apache/slider/providers/ProviderService.java
@@ -22,6 +22,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.slider.api.ClusterDescription;
import org.apache.slider.common.tools.SliderFileSystem;
import org.apache.slider.core.conf.AggregateConf;
@@ -69,6 +70,12 @@ public interface ProviderService extends ProviderCore, Service,
SliderException;
/**
+ * Notify the providers of container completion
+ * @param containerId
+ */
+ void notifyContainerCompleted(ContainerId containerId);
+
+ /**
* Execute a process in the AM
* @param instanceDefinition cluster description
* @param confDir configuration directory
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/dca4818c/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 b39601a..bbe2001 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
@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.slider.api.ClusterDescription;
@@ -110,6 +111,7 @@ 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 int MAX_LOG_ENTRIES = 20;
private static final int DEFAULT_HEARTBEAT_MONITOR_INTERVAL = 60 * 1000;
private final Object syncLock = new Object();
@@ -290,116 +292,6 @@ public class AgentProviderService extends AbstractProviderService implements
}
/**
- * Reads and sets the heartbeat monitoring interval. If bad value is provided then log it and set to default.
- * @param instanceDefinition
- */
- private void readAndSetHeartbeatMonitoringInterval(AggregateConf instanceDefinition) {
- String hbMonitorInterval = instanceDefinition.getAppConfOperations().
- getGlobalOptions().getOption(AgentKeys.HEARTBEAT_MONITOR_INTERVAL,
- Integer.toString(DEFAULT_HEARTBEAT_MONITOR_INTERVAL));
- try {
- setHeartbeatMonitorInterval(Integer.parseInt(hbMonitorInterval));
- }catch (NumberFormatException e) {
- log.warn(
- "Bad value {} for {}. Defaulting to ",
- hbMonitorInterval,
- HEARTBEAT_MONITOR_INTERVAL,
- DEFAULT_HEARTBEAT_MONITOR_INTERVAL);
- }
- }
-
- /**
- * Reads and sets the heartbeat monitoring interval. If bad value is provided then log it and set to default.
- * @param instanceDefinition
- */
- private void initializeAgentDebugCommands(AggregateConf instanceDefinition) {
- String launchParameterStr = instanceDefinition.getAppConfOperations().
- getGlobalOptions().getOption(AgentKeys.AGENT_INSTANCE_DEBUG_DATA, "");
- agentLaunchParameter = new AgentLaunchParameter(launchParameterStr);
- }
-
- @VisibleForTesting
- protected Metainfo getMetainfo() {
- return this.metainfo;
- }
-
- @VisibleForTesting
- protected Map<String, ComponentInstanceState> getComponentStatuses() {
- return componentStatuses;
- }
-
- @VisibleForTesting
- protected Metainfo getApplicationMetainfo(SliderFileSystem fileSystem,
- String appDef) throws IOException {
- return AgentUtils.getApplicationMetainfo(fileSystem, appDef);
- }
-
- @VisibleForTesting
- protected void setHeartbeatMonitorInterval(int heartbeatMonitorInterval) {
- this.heartbeatMonitorInterval = heartbeatMonitorInterval;
- }
-
- private int getHeartbeatMonitorInterval() {
- return this.heartbeatMonitorInterval;
- }
-
- /**
- * Publish a named config bag that may contain name-value pairs for app configurations such as hbase-site
- * @param name
- * @param description
- * @param entries
- */
- protected void publishComponentConfiguration(String name, String description,
- Iterable<Map.Entry<String, String>> entries) {
- PublishedConfiguration pubconf = new PublishedConfiguration();
- pubconf.description = description;
- pubconf.putValues(entries);
- log.info("publishing {}", pubconf);
- getAmState().getPublishedSliderConfigurations().put(name, pubconf);
- }
-
- /**
- * Get a list of all hosts for all role/container per role
- * @return
- */
- protected Map<String, Map<String, ClusterNode>> getRoleClusterNodeMapping() {
- amState.refreshClusterStatus();
- return (Map<String, Map<String, ClusterNode>>)
- amState.getClusterStatus().status.get(
- ClusterDescriptionKeys.KEY_CLUSTER_LIVE);
- }
-
- private String getContainerLabel(Container container, String role) {
- return container.getId().toString() + LABEL_MAKER + role;
- }
-
- protected String getClusterInfoPropertyValue(String name) {
- StateAccessForProviders accessor = getAmState();
- assert accessor.isApplicationLive();
- ClusterDescription description = accessor.getClusterStatus();
- return description.getInfo(name);
- }
-
- /**
- * Lost heartbeat from the container - release it and ask for a replacement
- *
- * @param label
- *
- * @return if release is requested successfully
- */
- protected boolean releaseContainer(String label) {
- componentStatuses.remove(label);
- try {
- getAppMaster().refreshContainer(getContainerId(label), true);
- } catch (SliderException e) {
- log.info("Error while requesting container release for {}. Message: {}", label, e.getMessage());
- return false;
- }
-
- return true;
- }
-
- /**
* Run this service
*
* @param instanceDefinition component description
@@ -421,16 +313,6 @@ public class AgentProviderService extends AbstractProviderService implements
return false;
}
- /**
- * Build the provider status, can be empty
- *
- * @return the provider status - map of entries to add to the info section
- */
- public Map<String, String> buildProviderStatus() {
- Map<String, String> stats = new HashMap<>();
- return stats;
- }
-
@Override
public boolean isSupportedRole(String role) {
return true;
@@ -489,7 +371,7 @@ public class AgentProviderService extends AbstractProviderService implements
// Otherwise, wait till the master that can publish is ready
if (isMaster &&
(canAnyMasterPublishConfig() == false || canPublishConfig(roleName))) {
- processReturnedStatus(heartBeat, componentStatus);
+ publishConfigAndExportGroups(heartBeat, componentStatus);
}
List<CommandReport> reports = heartBeat.getReports();
@@ -511,7 +393,7 @@ public class AgentProviderService extends AbstractProviderService implements
log.info("Component operation. Status: {}", result);
if (command == Command.INSTALL && report.getFolders() != null && report.getFolders().size() > 0) {
- processFolderPaths(report.getFolders(), containerId, heartBeat.getFqdn());
+ publishLogFolderPaths(report.getFolders(), containerId, heartBeat.getFqdn());
}
}
@@ -558,13 +440,168 @@ public class AgentProviderService extends AbstractProviderService implements
return response;
}
+ @Override
+ public Map<String, String> buildMonitorDetails(ClusterDescription clusterDesc) {
+ Map<String, String> details = super.buildMonitorDetails(clusterDesc);
+ buildRoleHostDetails(details);
+ return details;
+ }
+
+ @Override
+ public void applyInitialRegistryDefinitions(URL unsecureWebAPI,
+ URL secureWebAPI,
+ ServiceInstanceData instanceData) throws IOException {
+ super.applyInitialRegistryDefinitions(unsecureWebAPI,
+ secureWebAPI,
+ instanceData
+ );
+
+ try {
+ instanceData.internalView.endpoints.put(
+ CustomRegistryConstants.AGENT_REST_API,
+ new RegisteredEndpoint(
+ new URL(secureWebAPI, SLIDER_PATH_AGENTS),
+ "Agent REST API"));
+ } catch (URISyntaxException e) {
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public void notifyContainerCompleted(ContainerId containerId) {
+ if(containerId != null) {
+ String containerIdStr = containerId.toString();
+ }
+ }
+
/**
- * Format the folder locations before publishing in the registry service
+ * Reads and sets the heartbeat monitoring interval. If bad value is provided then log it and set to default.
+ * @param instanceDefinition
+ */
+ private void readAndSetHeartbeatMonitoringInterval(AggregateConf instanceDefinition) {
+ String hbMonitorInterval = instanceDefinition.getAppConfOperations().
+ getGlobalOptions().getOption(AgentKeys.HEARTBEAT_MONITOR_INTERVAL,
+ Integer.toString(DEFAULT_HEARTBEAT_MONITOR_INTERVAL));
+ try {
+ setHeartbeatMonitorInterval(Integer.parseInt(hbMonitorInterval));
+ }catch (NumberFormatException e) {
+ log.warn(
+ "Bad value {} for {}. Defaulting to ",
+ hbMonitorInterval,
+ HEARTBEAT_MONITOR_INTERVAL,
+ DEFAULT_HEARTBEAT_MONITOR_INTERVAL);
+ }
+ }
+
+ /**
+ * Reads and sets the heartbeat monitoring interval. If bad value is provided then log it and set to default.
+ * @param instanceDefinition
+ */
+ private void initializeAgentDebugCommands(AggregateConf instanceDefinition) {
+ String launchParameterStr = instanceDefinition.getAppConfOperations().
+ getGlobalOptions().getOption(AgentKeys.AGENT_INSTANCE_DEBUG_DATA, "");
+ agentLaunchParameter = new AgentLaunchParameter(launchParameterStr);
+ }
+
+ @VisibleForTesting
+ protected Metainfo getMetainfo() {
+ return this.metainfo;
+ }
+
+ @VisibleForTesting
+ protected Map<String, ComponentInstanceState> getComponentStatuses() {
+ return componentStatuses;
+ }
+
+ @VisibleForTesting
+ protected Metainfo getApplicationMetainfo(SliderFileSystem fileSystem,
+ String appDef) throws IOException {
+ return AgentUtils.getApplicationMetainfo(fileSystem, appDef);
+ }
+
+ @VisibleForTesting
+ protected void setHeartbeatMonitorInterval(int heartbeatMonitorInterval) {
+ this.heartbeatMonitorInterval = heartbeatMonitorInterval;
+ }
+
+ private int getHeartbeatMonitorInterval() {
+ return this.heartbeatMonitorInterval;
+ }
+
+ /**
+ * Publish a named property bag that may contain name-value pairs for app configurations such as hbase-site
+ * @param name
+ * @param description
+ * @param entries
+ */
+ protected void publishComponentConfiguration(String name, String description,
+ Iterable<Map.Entry<String, String>> entries) {
+ PublishedConfiguration pubconf = new PublishedConfiguration();
+ pubconf.description = description;
+ pubconf.putValues(entries);
+ log.info("publishing {}", pubconf);
+ getAmState().getPublishedSliderConfigurations().put(name, pubconf);
+ }
+
+ /**
+ * Get a list of all hosts for all role/container per role
+ * @return
+ */
+ protected Map<String, Map<String, ClusterNode>> getRoleClusterNodeMapping() {
+ amState.refreshClusterStatus();
+ return (Map<String, Map<String, ClusterNode>>)
+ amState.getClusterStatus().status.get(
+ ClusterDescriptionKeys.KEY_CLUSTER_LIVE);
+ }
+
+ private String getContainerLabel(Container container, String role) {
+ return container.getId().toString() + LABEL_MAKER + role;
+ }
+
+ protected String getClusterInfoPropertyValue(String name) {
+ StateAccessForProviders accessor = getAmState();
+ assert accessor.isApplicationLive();
+ ClusterDescription description = accessor.getClusterStatus();
+ return description.getInfo(name);
+ }
+
+ /**
+ * Lost heartbeat from the container - release it and ask for a replacement
+ *
+ * @param label
+ *
+ * @return if release is requested successfully
+ */
+ protected boolean releaseContainer(String label) {
+ componentStatuses.remove(label);
+ try {
+ getAppMaster().refreshContainer(getContainerId(label), true);
+ } catch (SliderException e) {
+ log.info("Error while requesting container release for {}. Message: {}", label, e.getMessage());
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Build the provider status, can be empty
+ *
+ * @return the provider status - map of entries to add to the info section
+ */
+ public Map<String, String> buildProviderStatus() {
+ Map<String, String> stats = new HashMap<>();
+ return stats;
+ }
+
+
+ /**
+ * Format the folder locations and publish in the registry service
* @param folders
* @param containerId
* @param hostFqdn
*/
- private void processFolderPaths(Map<String, String> folders, String containerId, String hostFqdn) {
+ private void publishLogFolderPaths(Map<String, String> folders, String containerId, String hostFqdn) {
for (String key : folders.keySet()) {
workFolders.put(String.format("%s-%s-%s", hostFqdn, containerId, key), folders.get(key));
}
@@ -578,7 +615,7 @@ public class AgentProviderService extends AbstractProviderService implements
* @param heartBeat
* @param componentStatus
*/
- protected void processReturnedStatus(HeartBeat heartBeat, ComponentInstanceState componentStatus) {
+ protected void publishConfigAndExportGroups(HeartBeat heartBeat, ComponentInstanceState componentStatus) {
List<ComponentStatus> statuses = heartBeat.getComponentStatus();
if (statuses != null && !statuses.isEmpty()) {
log.info("Processing {} status reports.", statuses.size());
@@ -658,11 +695,11 @@ public class AgentProviderService extends AbstractProviderService implements
boolean publishData = false;
String portValPattern = String.format(portVarFormat, portName);
if(templateToExport.contains(portValPattern)) {
- templateToExport.replace(portValPattern, ports.get(portName));
+ templateToExport = templateToExport.replace(portValPattern, ports.get(portName));
publishData = true;
}
if(templateToExport.contains(hostNamePattern)) {
- templateToExport.replace(hostNamePattern, hostFqdn);
+ templateToExport = templateToExport.replace(hostNamePattern, hostFqdn);
publishData = true;
}
if(publishData) {
@@ -1032,9 +1069,13 @@ public class AgentProviderService extends AbstractProviderService implements
//apply any port updates
if (!this.getAllocatedPorts().isEmpty()) {
for (String key : config.keySet()) {
- String lookupKey = configName + "." + key;
- if (this.getAllocatedPorts().containsKey(lookupKey)) {
- config.put(key, getAllocatedPorts().get(lookupKey));
+ 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;
+ if (this.getAllocatedPorts().containsKey(lookupKey)) {
+ config.put(key, getAllocatedPorts().get(lookupKey));
+ }
}
}
}
@@ -1065,13 +1106,6 @@ public class AgentProviderService extends AbstractProviderService implements
config.put("app_install_dir", "${AGENT_WORK_ROOT}/app/install");
}
- @Override
- public Map<String, String> buildMonitorDetails(ClusterDescription clusterDesc) {
- Map<String, String> details = super.buildMonitorDetails(clusterDesc);
- buildRoleHostDetails(details);
- return details;
- }
-
private void buildRoleHostDetails(Map<String, String> details) {
for (Map.Entry<String, Map<String, ClusterNode>> entry :
getRoleClusterNodeMapping().entrySet()) {
@@ -1080,25 +1114,4 @@ public class AgentProviderService extends AbstractProviderService implements
"");
}
}
-
- @Override
- public void applyInitialRegistryDefinitions(URL unsecureWebAPI,
- URL secureWebAPI,
- ServiceInstanceData instanceData) throws IOException {
- super.applyInitialRegistryDefinitions(unsecureWebAPI,
- secureWebAPI,
- instanceData
- );
-
- try {
- instanceData.internalView.endpoints.put(
- CustomRegistryConstants.AGENT_REST_API,
- new RegisteredEndpoint(
- new URL(secureWebAPI, SLIDER_PATH_AGENTS),
- "Agent REST API"));
- } catch (URISyntaxException e) {
- throw new IOException(e);
- }
-
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/dca4818c/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java
----------------------------------------------------------------------
diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java
index 7e0ae5e..a2dc36f 100644
--- a/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java
+++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java
@@ -1061,6 +1061,8 @@ public class SliderAppMaster extends AbstractSliderLaunchedService
RoleInstance ri = result.roleInstance;
log.error("Role instance {} failed ", ri);
}
+
+ getProviderService().notifyContainerCompleted(containerId);
}
// ask for more containers if any failed
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/dca4818c/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 4d63263..1f9cb41 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
@@ -355,6 +355,45 @@ public class TestAgentProviderService {
}
@Test
+ public void testComponentSpecificPublishes() throws Exception {
+ InputStream metainfo_1 = new ByteArrayInputStream(metainfo_1_str.getBytes());
+ Metainfo metainfo = new MetainfoParser().parse(metainfo_1);
+ AgentProviderService aps = new AgentProviderService();
+ AgentProviderService mockAps = Mockito.spy(aps);
+ doNothing().when(mockAps).publishComponentConfiguration(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");
+ ArgumentCaptor<Collection> entriesCaptor = ArgumentCaptor.
+ forClass(Collection.class);
+ ArgumentCaptor<String> publishNameCaptor = ArgumentCaptor.
+ forClass(String.class);
+ Mockito.verify(mockAps, Mockito.times(1)).publishComponentConfiguration(
+ anyString(),
+ publishNameCaptor.capture(),
+ entriesCaptor.capture());
+ assert entriesCaptor.getAllValues().size() == 1;
+ for (Collection coll : entriesCaptor.getAllValues()) {
+ Set<Map.Entry<String, String>> entrySet = (Set<Map.Entry<String, String>>) coll;
+ for (Map.Entry entry : entrySet) {
+ log.info("{}:{}", entry.getKey(), entry.getValue().toString());
+ if (entry.getKey().equals("PropertyA")) {
+ assert entry.getValue().toString().equals("host1:10010");
+ }
+ }
+ }
+ assert publishNameCaptor.getAllValues().size() == 1;
+ for (String coll : publishNameCaptor.getAllValues()) {
+ assert coll.equals("ComponentInstanceData");
+ }
+ }
+
+ @Test
public void testProcessConfig() throws Exception {
InputStream metainfo_1 = new ByteArrayInputStream(metainfo_1_str.getBytes());
Metainfo metainfo = new MetainfoParser().parse(metainfo_1);
@@ -387,16 +426,16 @@ public class TestAgentProviderService {
doReturn(metainfo).when(mockAps).getMetainfo();
doReturn(roleClusterNodeMap).when(mockAps).getRoleClusterNodeMapping();
- mockAps.processReturnedStatus(hb, componentStatus);
+ mockAps.publishConfigAndExportGroups(hb, componentStatus);
assert componentStatus.getConfigReported() == true;
- ArgumentCaptor<Collection> commandCaptor = ArgumentCaptor.
+ ArgumentCaptor<Collection> entriesCaptor = ArgumentCaptor.
forClass(Collection.class);
Mockito.verify(mockAps, Mockito.times(3)).publishComponentConfiguration(
anyString(),
anyString(),
- commandCaptor.capture());
- assert commandCaptor.getAllValues().size() == 3;
- for (Collection coll : commandCaptor.getAllValues()) {
+ entriesCaptor.capture());
+ assert entriesCaptor.getAllValues().size() == 3;
+ for (Collection coll : entriesCaptor.getAllValues()) {
Set<Map.Entry<String, String>> entrySet = (Set<Map.Entry<String, String>>) coll;
for (Map.Entry entry : entrySet) {
log.info("{}:{}", entry.getKey(), entry.getValue().toString());