From 8d38008a142c76d092d80c60c8d0fa83241f11a4 Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Mon, 14 Sep 2015 12:55:03 +0100 Subject: [PATCH 01/13] Fixes for starting mulitple, and making ApplicationMasterServiceImpl easier to override. --- .../service/ApplicationMasterServiceImpl.java | 28 ++++++++++++------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index 71425b8..86c3cd5 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -132,12 +132,18 @@ protected void startUp() throws IOException { ContainerLaunchContextFactory factory = new ContainerLaunchContextFactory( registration.getMaximumResourceCapability(), tokens); for (ContainerLaunchParameters clp : parameters.getContainerLaunchParameters()) { - ContainerTracker tracker = new ContainerTracker(clp); + ContainerTracker tracker = getTracker(clp); tracker.init(factory); - trackers.add(tracker); + synchronized(trackers) { //trackers can be added while containers are being allocated (c.f. onContainersAllocated()) + trackers.add(tracker); + } } this.hasRunningContainers = true; } + + protected ContainerTracker getTracker(ContainerLaunchParameters clp) { + return new ContainerTracker(clp); + } @Override protected void shutDown() { @@ -210,14 +216,16 @@ public void onContainersCompleted(List containerStatuses) { public void onContainersAllocated(List allocatedContainers) { LOG.info("Allocating " + allocatedContainers.size() + " container(s)"); Set assigned = Sets.newHashSet(); - for (ContainerTracker tracker : trackers) { - if (tracker.needsContainers()) { - for (Container allocated : allocatedContainers) { - if (!assigned.contains(allocated) && tracker.matches(allocated)) { - tracker.launchContainer(allocated); - assigned.add(allocated); + synchronized (trackers) { + for (ContainerTracker tracker : trackers) { + //we need to check tracker.needsContainers() for each container, as the + //previously allocated container may have addressed the containers' need + for (Container allocated : allocatedContainers) { + if (tracker.needsContainers() && ! assigned.contains(allocated) && tracker.matches(allocated)) { + tracker.launchContainer(allocated); + assigned.add(allocated); + } } - } } } if (assigned.size() < allocatedContainers.size()) { @@ -256,7 +264,7 @@ public void onError(Throwable throwable) { stop(); } - private class ContainerTracker implements NMClientAsync.CallbackHandler { + protected class ContainerTracker implements NMClientAsync.CallbackHandler { private final ContainerLaunchParameters parameters; private final ConcurrentMap containers = Maps.newConcurrentMap(); From 6da3ae94f870a375ce0c3779d409bc7a7bd61ac0 Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Mon, 14 Sep 2015 13:13:41 +0100 Subject: [PATCH 02/13] bump to Hadoop 2.6.0 for node label expression support --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 237ccaa..30ef1ff 100644 --- a/pom.xml +++ b/pom.xml @@ -19,7 +19,7 @@ UTF-8 - 2.2.0 + 2.6.0 11.0.2 2.0.2 4.8.2 From a2743f7038ed56222ac275f61ed922821fa84a8d Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Mon, 14 Sep 2015 13:14:04 +0100 Subject: [PATCH 03/13] we need junit for testing both common and master --- java/common/pom.xml | 7 +++++++ java/master/pom.xml | 7 +++++++ 2 files changed, 14 insertions(+) diff --git a/java/common/pom.xml b/java/common/pom.xml index 260d4f0..99db2a8 100644 --- a/java/common/pom.xml +++ b/java/common/pom.xml @@ -41,6 +41,13 @@ luaj-jse ${luaj.version} + + + junit + junit + ${junit.version} + test + diff --git a/java/master/pom.xml b/java/master/pom.xml index 74df1f9..9659c0e 100644 --- a/java/master/pom.xml +++ b/java/master/pom.xml @@ -54,6 +54,13 @@ ${hadoop.version} test + + + junit + junit + ${junit.version} + test + From 30fb2067419bebdd62083b93e3fe9b9017a707fe Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Mon, 14 Sep 2015 13:15:02 +0100 Subject: [PATCH 04/13] Added support for Hadoop 2.6.0 node label expressions --- .../kitten/ContainerLaunchContextFactory.java | 9 ++++++--- .../kitten/ContainerLaunchParameters.java | 6 ++++++ .../lua/LuaContainerLaunchParameters.java | 7 +++++++ .../java/com/cloudera/kitten/lua/LuaFields.java | 1 + .../service/ApplicationMasterServiceImpl.java | 17 +++++++++++++---- 5 files changed, 33 insertions(+), 7 deletions(-) diff --git a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java index f59fcb4..2a2baec 100644 --- a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java +++ b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java @@ -14,14 +14,13 @@ */ package com.cloudera.kitten; +import java.nio.ByteBuffer; + import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; import org.apache.hadoop.yarn.api.records.Priority; import org.apache.hadoop.yarn.api.records.Resource; -import org.apache.hadoop.yarn.api.records.ResourceRequest; import org.apache.hadoop.yarn.util.Records; -import java.nio.ByteBuffer; - /** * Functions for constructing YARN objects from the parameter values. */ @@ -45,6 +44,10 @@ public ContainerLaunchContext create(ContainerLaunchParameters parameters) { } return clc; } + + public String getNodeLabelExpression(ContainerLaunchParameters parameters) { + return parameters.getNodeLabelsExpression(); + } public Resource createResource(ContainerLaunchParameters parameters) { return parameters.getContainerResource(clusterMax); diff --git a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java index d4d700c..2202bef 100644 --- a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java +++ b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java @@ -54,4 +54,10 @@ public interface ContainerLaunchParameters { * The commands to execute that start the application within the container. */ List getCommands(); + + /** + * The nodeLabelsExpression that defines the types of nodes that can be allocated to the container. + * Defaults to null. + */ + String getNodeLabelsExpression(); } diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java index 289d5eb..2c54222 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java @@ -74,6 +74,11 @@ public int getMemory() { return lv.getInteger(LuaFields.MEMORY); } + @Override + public String getNodeLabelsExpression() { + return lv.isNil(LuaFields.NODE_LABELS) ? null : lv.getString(LuaFields.NODE_LABELS); + } + @Override public Resource getContainerResource(Resource clusterMax) { Resource rsrc = Records.newRecord(Resource.class); @@ -251,4 +256,6 @@ public String toCommand(LuaWrapper table) { } return sb.toString(); } + + } diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java index 05970f0..eff8665 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java @@ -37,6 +37,7 @@ public class LuaFields { public static final String CORES = "cores"; public static final String MEMORY = "memory"; public static final String PRIORITY = "priority"; + public static final String NODE_LABELS = "nodeLabels"; // For constructing commands from a LuaTable. public static final String COMMAND_BASE = "base"; diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index 86c3cd5..350c7da 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -275,13 +275,15 @@ protected class ContainerTracker implements NMClientAsync.CallbackHandler { private NMClientAsync nodeManager; private Resource resource; private Priority priority; + private String nodeLabelsExpression; private ContainerLaunchContext ctxt; public ContainerTracker(ContainerLaunchParameters parameters) { this.parameters = parameters; } - public void init(ContainerLaunchContextFactory factory) { + @SuppressWarnings("unchecked") + public void init(ContainerLaunchContextFactory factory) { this.nodeManager = NMClientAsync.createNMClientAsync(this); nodeManager.init(conf); nodeManager.start(); @@ -289,12 +291,19 @@ public void init(ContainerLaunchContextFactory factory) { this.ctxt = factory.create(parameters); this.resource = factory.createResource(parameters); this.priority = factory.createPriority(parameters.getPriority()); + this.nodeLabelsExpression = factory.getNodeLabelExpression(parameters); + String[] nodes = null; + String[] racks = null; AMRMClient.ContainerRequest containerRequest = new AMRMClient.ContainerRequest( resource, - null, // nodes - null, // racks - priority); + nodes, // nodes + racks, // racks + priority, + true, //default for "relaxLocality" + nodeLabelsExpression //usually null + ); int numInstances = parameters.getNumInstances(); + LOG.debug(this.toString() + " needs " + numInstances + " instances of this container type"); for (int j = 0; j < numInstances; j++) { resourceManager.addContainerRequest(containerRequest); } From 93d5d151145de7854332f80abf09377c10c017d5 Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Thu, 1 Sep 2016 10:29:46 +0100 Subject: [PATCH 05/13] make a lua compiler error more verbose --- .../java/com/cloudera/kitten/lua/LuaWrapper.java | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaWrapper.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaWrapper.java index 7ba32b8..538b55d 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaWrapper.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaWrapper.java @@ -20,8 +20,10 @@ import java.util.List; import java.util.Map; +import org.apache.commons.io.IOUtils; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.util.StringUtils; import org.luaj.vm2.LoadState; import org.luaj.vm2.LuaTable; import org.luaj.vm2.LuaValue; @@ -58,11 +60,19 @@ public LuaWrapper(String script, Map extras) { env.set(e.getKey(), CoerceJavaToLua.coerce(e.getValue())); } InputStream luaCode = LocalDataHelper.getFileOrResource(script); + assert luaCode != null : "Lua script " + script + " not found"; LoadState.load(luaCode, script, env).call(); } catch (IOException e) { LOG.error("Lua initialization error", e); throw new RuntimeException(e); - } + } catch (NullPointerException e) { + LOG.error("Lua compiler error", e); + try{ + List lines = IOUtils.readLines( LocalDataHelper.getFileOrResource(script)); + LOG.error(StringUtils.join("\n", lines)); + } catch (IOException ioe) {} + throw new RuntimeException(e); + } } public LuaWrapper(LuaTable table) { From 8bb28ee77a20616dfe35b801e50c075df2ab282a Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Thu, 1 Sep 2016 10:38:52 +0100 Subject: [PATCH 06/13] * Add support for varying the application type * credential fixes * node label expressions * fix for resources * support for requesting a named node * fixes for the way that allocation is made --- .../kitten/client/YarnClientParameters.java | 5 + .../params/lua/LuaYarnClientParameters.java | 9 + .../client/service/YarnClientServiceImpl.java | 17 +- java/common/pom.xml | 8 +- .../kitten/ContainerLaunchContextFactory.java | 9 +- .../kitten/ContainerLaunchParameters.java | 10 + .../lua/LuaContainerLaunchParameters.java | 26 ++- .../com/cloudera/kitten/lua/LuaFields.java | 4 + .../cloudera/kitten/util/LocalDataHelper.java | 3 + .../appmaster/ApplicationMasterService.java | 4 + .../service/ApplicationMasterServiceImpl.java | 219 +++++++++++++----- pom.xml | 2 +- 12 files changed, 249 insertions(+), 67 deletions(-) diff --git a/java/client/src/main/java/com/cloudera/kitten/client/YarnClientParameters.java b/java/client/src/main/java/com/cloudera/kitten/client/YarnClientParameters.java index 3259944..39c9dd3 100644 --- a/java/client/src/main/java/com/cloudera/kitten/client/YarnClientParameters.java +++ b/java/client/src/main/java/com/cloudera/kitten/client/YarnClientParameters.java @@ -30,6 +30,11 @@ public interface YarnClientParameters { */ String getApplicationName(); + /** + * The type of this YARN application. Defaults to "Kitten" + */ + String getApplicationType(); + /** * The queue the application master is assigned to run in. */ diff --git a/java/client/src/main/java/com/cloudera/kitten/client/params/lua/LuaYarnClientParameters.java b/java/client/src/main/java/com/cloudera/kitten/client/params/lua/LuaYarnClientParameters.java index 6ce03af..a0525e9 100644 --- a/java/client/src/main/java/com/cloudera/kitten/client/params/lua/LuaYarnClientParameters.java +++ b/java/client/src/main/java/com/cloudera/kitten/client/params/lua/LuaYarnClientParameters.java @@ -20,7 +20,9 @@ import java.util.Map; import org.apache.commons.logging.Log; + import org.apache.commons.logging.LogFactory; + import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.yarn.api.records.ApplicationId; @@ -97,6 +99,11 @@ public Configuration getConfiguration() { public String getApplicationName() { return env.getString(LuaFields.APP_NAME); } + + @Override + public String getApplicationType() { + return env.isNil(LuaFields.APP_TYPE) ? "kitten" : env.getString(LuaFields.APP_TYPE); + } @Override public String getQueue() { @@ -166,4 +173,6 @@ public long getClientTimeoutMillis() { return env.isNil(LuaFields.TIMEOUT) ? -1 : env.getLong(LuaFields.TIMEOUT); } + + } diff --git a/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java b/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java index 906b77d..dc1d11b 100644 --- a/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java +++ b/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java @@ -64,7 +64,7 @@ public class YarnClientServiceImpl extends AbstractScheduledService private final YarnClientParameters parameters; private final MasterConnectionFactory yarnClientFactory; private final Stopwatch stopwatch; - + private Credentials credentials; private YarnClient yarnClient; private ApplicationId applicationId; private ApplicationReport finalReport; @@ -72,15 +72,23 @@ public class YarnClientServiceImpl extends AbstractScheduledService public YarnClientServiceImpl(YarnClientParameters params) { this(params, new YarnClientFactory(params.getConfiguration()), - new Stopwatch()); + new Stopwatch(), new Credentials()); } + public YarnClientServiceImpl(YarnClientParameters params, Credentials credentials) { + this(params, new YarnClientFactory(params.getConfiguration()), + new Stopwatch(), credentials); + } + public YarnClientServiceImpl(YarnClientParameters parameters, MasterConnectionFactory yarnClientFactory, - Stopwatch stopwatch) { + Stopwatch stopwatch, Credentials credentials) { this.parameters = Preconditions.checkNotNull(parameters); this.yarnClientFactory = yarnClientFactory; this.stopwatch = stopwatch; + this.credentials = credentials; + if (this.credentials == null) + this.credentials = new Credentials(); } @Override @@ -89,7 +97,6 @@ protected void startUp() throws IOException { if (UserGroupInformation.isSecurityEnabled()) { Configuration conf = this.yarnClientFactory.getConfig(); FileSystem fs = FileSystem.get(conf); - Credentials credentials = new Credentials(); String tokenRenewer = this.yarnClientFactory.getConfig().get(YarnConfiguration.RM_PRINCIPAL); if (tokenRenewer == null || tokenRenewer.length() == 0) { throw new IOException("Can't get Master Kerberos principal for the RM to use as renewer"); @@ -116,6 +123,8 @@ protected void startUp() throws IOException { ApplicationSubmissionContext appContext = clientApp.getApplicationSubmissionContext(); this.applicationId = appContext.getApplicationId(); appContext.setApplicationName(parameters.getApplicationName()); + if (parameters.getApplicationType() != null) + appContext.setApplicationType(parameters.getApplicationType()); // Setup the container for the application master. ContainerLaunchParameters appMasterParams = parameters.getApplicationMasterParameters(applicationId); diff --git a/java/common/pom.xml b/java/common/pom.xml index 260d4f0..af02b80 100644 --- a/java/common/pom.xml +++ b/java/common/pom.xml @@ -41,7 +41,13 @@ luaj-jse ${luaj.version} - + + + junit + junit + 4.11 + + diff --git a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java index f59fcb4..309c09a 100644 --- a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java +++ b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java @@ -14,14 +14,13 @@ */ package com.cloudera.kitten; +import java.nio.ByteBuffer; + import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; import org.apache.hadoop.yarn.api.records.Priority; import org.apache.hadoop.yarn.api.records.Resource; -import org.apache.hadoop.yarn.api.records.ResourceRequest; import org.apache.hadoop.yarn.util.Records; -import java.nio.ByteBuffer; - /** * Functions for constructing YARN objects from the parameter values. */ @@ -46,6 +45,10 @@ public ContainerLaunchContext create(ContainerLaunchParameters parameters) { return clc; } + public String getNodeLabelExpression(ContainerLaunchParameters parameters) { + return parameters.getNodeLabelsExpression(); + } + public Resource createResource(ContainerLaunchParameters parameters) { return parameters.getContainerResource(clusterMax); } diff --git a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java index d4d700c..29341eb 100644 --- a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java +++ b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchParameters.java @@ -54,4 +54,14 @@ public interface ContainerLaunchParameters { * The commands to execute that start the application within the container. */ List getCommands(); + + /** + * The nodeLabelsExpression that defines the types of nodes that can be allocated to the container. + * Defaults to null. + */ + String getNodeLabelsExpression(); + + /** The node name that is /required/ for this container. Defaults to null, which signifies no + * requirement */ + String getNode(); } diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java index 289d5eb..3a271b7 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java @@ -27,7 +27,6 @@ import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.yarn.api.records.LocalResource; import org.apache.hadoop.yarn.api.records.LocalResourceType; import org.apache.hadoop.yarn.api.records.LocalResourceVisibility; @@ -66,6 +65,7 @@ public LuaContainerLaunchParameters(LuaWrapper lv, Configuration conf, this.extras = extras; } + public int getCores() { return lv.getInteger(LuaFields.CORES); } @@ -73,6 +73,11 @@ public int getCores() { public int getMemory() { return lv.getInteger(LuaFields.MEMORY); } + + @Override + public String getNodeLabelsExpression() { + return lv.isNil(LuaFields.NODE_LABELS) ? null : lv.getString(LuaFields.NODE_LABELS); + } @Override public Resource getContainerResource(Resource clusterMax) { @@ -92,6 +97,12 @@ public int getNumInstances() { return lv.isNil(LuaFields.INSTANCES) ? 1 : lv.getInteger(LuaFields.INSTANCES); } + + @Override + public String getNode() { + return lv.isNil(LuaFields.NODE) ? null : lv.getString(LuaFields.NODE); + } + @Override public Map getLocalResources() { Map localResources = Maps.newHashMap(); @@ -123,12 +134,17 @@ public Map getLocalResources() { } private LocalResource constructExtraResource(String key) { - LocalResource rsrc = Records.newRecord(LocalResource.class); + if (key.equals("job.xml") && ! localFileUris.containsKey(key)) + return null; + LocalResource rsrc = Records.newRecord(LocalResource.class); rsrc.setType(LocalResourceType.FILE); rsrc.setVisibility(LocalResourceVisibility.APPLICATION); try { Path path = new Path(localFileUris.get(key)); configureLocalResourceForPath(rsrc, path); + } catch (NullPointerException npe) { + LOG.warn("No local URI found for "+ key, npe); + return null; } catch (IOException e) { LOG.error("Error constructing extra local resource: " + key, e); return null; @@ -191,9 +207,10 @@ private NamedLocalResource constructResource(LuaPair lp) throws IOException { private void configureLocalResourceForPath(LocalResource rsrc, Path path) throws IOException { FileSystem fs = FileSystem.get(conf); FileStatus stat = fs.getFileStatus(path); + Path p = path.makeQualified(fs.getUri(), fs.getWorkingDirectory()); rsrc.setSize(stat.getLen()); rsrc.setTimestamp(stat.getModificationTime()); - rsrc.setResource(ConverterUtils.getYarnUrlFromPath(path)); + rsrc.setResource(ConverterUtils.getYarnUrlFromPath(p)); } @Override @@ -251,4 +268,7 @@ public String toCommand(LuaWrapper table) { } return sb.toString(); } + + + } diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java index 05970f0..c4c3c03 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaFields.java @@ -24,6 +24,7 @@ public class LuaFields { public static final String CONTAINERS = "containers"; public static final String CONTAINER = "container"; public static final String APP_NAME = "name"; + public static final String APP_TYPE = "app_type"; public static final String TIMEOUT = "timeout"; public static final String USER = "user"; public static final String QUEUE = "queue"; @@ -37,6 +38,9 @@ public class LuaFields { public static final String CORES = "cores"; public static final String MEMORY = "memory"; public static final String PRIORITY = "priority"; + public static final String NODE_LABELS = "node_labels"; + public static final String NODE = "node"; + // For constructing commands from a LuaTable. public static final String COMMAND_BASE = "base"; diff --git a/java/common/src/main/java/com/cloudera/kitten/util/LocalDataHelper.java b/java/common/src/main/java/com/cloudera/kitten/util/LocalDataHelper.java index 80ab666..39ac84d 100644 --- a/java/common/src/main/java/com/cloudera/kitten/util/LocalDataHelper.java +++ b/java/common/src/main/java/com/cloudera/kitten/util/LocalDataHelper.java @@ -31,7 +31,9 @@ import org.apache.commons.codec.binary.Base64; import org.apache.commons.logging.Log; + import org.apache.commons.logging.LogFactory; + import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; @@ -125,6 +127,7 @@ private void copyToHdfs(String key, String localDataName) throws IOException { Path src = new Path(localDataName); Path dst = getPath(fs, src.getName()); InputStream data = getFileOrResource(localDataName); + assert data != null : "Could not access file/resource " + localDataName; FSDataOutputStream os = fs.create(dst, true); ByteStreams.copy(data, os); os.close(); diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/ApplicationMasterService.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/ApplicationMasterService.java index 1ca5038..5e30f2b 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/ApplicationMasterService.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/ApplicationMasterService.java @@ -32,4 +32,8 @@ public interface ApplicationMasterService extends Service { * monitoring. */ boolean hasRunningContainers(); + + int getTotalRequested(); + int getTotalCompleted(); + int getTotalFailures(); } diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index 71425b8..a4093ab 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -16,6 +16,7 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -24,16 +25,16 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import com.cloudera.kitten.ContainerLaunchContextFactory; -import com.google.common.collect.Lists; -import com.google.common.collect.Sets; import org.apache.commons.logging.Log; + import org.apache.commons.logging.LogFactory; + import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.io.DataOutputBuffer; import org.apache.hadoop.security.Credentials; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hadoop.security.token.Token; +import org.apache.hadoop.util.StringUtils; import org.apache.hadoop.yarn.api.ApplicationConstants; import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse; import org.apache.hadoop.yarn.api.records.Container; @@ -46,17 +47,21 @@ import org.apache.hadoop.yarn.api.records.Priority; import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.client.api.AMRMClient; +import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest; import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync; import org.apache.hadoop.yarn.client.api.async.NMClientAsync; import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.security.AMRMTokenIdentifier; +import com.cloudera.kitten.ContainerLaunchContextFactory; import com.cloudera.kitten.ContainerLaunchParameters; import com.cloudera.kitten.appmaster.ApplicationMasterParameters; import com.cloudera.kitten.appmaster.ApplicationMasterService; import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; import com.google.common.collect.Maps; +import com.google.common.collect.Sets; import com.google.common.util.concurrent.AbstractScheduledService; -import org.apache.hadoop.yarn.security.AMRMTokenIdentifier; public class ApplicationMasterServiceImpl extends AbstractScheduledService implements ApplicationMasterService, @@ -69,9 +74,9 @@ public class ApplicationMasterServiceImpl extends private AtomicInteger totalRequested = new AtomicInteger(); private AtomicInteger totalCompleted = new AtomicInteger(); private final AtomicInteger totalFailures = new AtomicInteger(); - private final List trackers = Lists.newArrayList(); + protected final List trackers = Lists.newArrayList(); - private AMRMClientAsync resourceManager; + private AMRMClientAsync resourceManager; private UserGroupInformation appSubmitterUgi; private boolean hasRunningContainers = false; private Throwable throwable; @@ -91,27 +96,33 @@ public boolean hasRunningContainers() { return hasRunningContainers; } + @Override protected void startUp() throws IOException { - Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials(); - DataOutputBuffer dob = new DataOutputBuffer(); - credentials.writeTokenStorageToStream(dob); - // Now remove the AM->RM token so that containers cannot access it. - Iterator> iter = credentials.getAllTokens().iterator(); - LOG.info("Executing with tokens:"); - while (iter.hasNext()) { - Token token = iter.next(); - LOG.info(token); - if (token.getKind().equals(AMRMTokenIdentifier.KIND_NAME)) { - iter.remove(); - } - } - ByteBuffer tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength()); + + ByteBuffer tokens = null; + if (UserGroupInformation.isSecurityEnabled()) { + Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials(); + DataOutputBuffer dob = new DataOutputBuffer(); + credentials.writeTokenStorageToStream(dob); + // Now remove the AM->RM token so that containers cannot access it. + Iterator> iter = credentials.getAllTokens().iterator(); + LOG.info("Executing with tokens:"); + while (iter.hasNext()) { + Token token = iter.next(); + LOG.info(token); + if (token.getKind().equals(AMRMTokenIdentifier.KIND_NAME)) { + iter.remove(); + } + } + tokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength()); + - // Create appSubmitterUgi and add original tokens to it - String userName = System.getenv(ApplicationConstants.Environment.USER.name()); - appSubmitterUgi = UserGroupInformation.createRemoteUser(userName); - appSubmitterUgi.addCredentials(credentials); + // Create appSubmitterUgi and add original tokens to it + String userName = System.getenv(ApplicationConstants.Environment.USER.name()); + appSubmitterUgi = UserGroupInformation.createRemoteUser(userName); + appSubmitterUgi.addCredentials(credentials); + } this.resourceManager = AMRMClientAsync.createAMRMClientAsync(1000, this); this.resourceManager.init(conf); @@ -131,18 +142,31 @@ protected void startUp() throws IOException { ContainerLaunchContextFactory factory = new ContainerLaunchContextFactory( registration.getMaximumResourceCapability(), tokens); + int totalRequested = 0; for (ContainerLaunchParameters clp : parameters.getContainerLaunchParameters()) { - ContainerTracker tracker = new ContainerTracker(clp); + ContainerTracker tracker = getTracker(clp); tracker.init(factory); - trackers.add(tracker); + totalRequested += tracker.getTotalNumInstances(); + synchronized (trackers) { //trackers can be added while containers are being allocated (c.f. onContainersAllocated()) + trackers.add(tracker); + } + } + LOG.info("Created " + trackers.size() + " ContainerTrackers, requested " + totalRequested + " containers"); this.hasRunningContainers = true; } + + protected ContainerTracker getTracker(ContainerLaunchParameters clp) { + return new ContainerTracker(clp); + } @Override protected void shutDown() { // Stop the containers in the case that we're finishing because of a timeout. - LOG.info("Stopping trackers"); + LOG.info("Stopping trackers: status : " + + totalRequested.get() + " requested, " + + totalCompleted.get() + " completed, " + + totalFailures.get() + " failures"); this.hasRunningContainers = false; for (ContainerTracker tracker : trackers) { @@ -152,11 +176,17 @@ protected void shutDown() { } FinalApplicationStatus status; String message = null; - if (state() == State.FAILED || totalFailures.get() > parameters.getAllowedFailures()) { - //TODO: diagnostics + if (state() == State.FAILED + || totalFailures.get() > parameters.getAllowedFailures() + || totalFailures.get() >= totalCompleted.get() + ) { status = FinalApplicationStatus.FAILED; if (throwable != null) { message = throwable.getLocalizedMessage(); + } else if (totalFailures.get() > parameters.getAllowedFailures() + || totalFailures.get() >= totalCompleted.get()) { + message = "Too many failures ("+totalFailures.get() + +" failures, out of "+totalCompleted.get()+" total)"; } } else { status = FinalApplicationStatus.SUCCEEDED; @@ -176,8 +206,12 @@ protected Scheduler scheduler() { @Override protected void runOneIteration() throws Exception { - if (totalFailures.get() > parameters.getAllowedFailures() || - totalCompleted.get() == totalRequested.get()) { + if (totalFailures.get() > parameters.getAllowedFailures()) { + LOG.error("Failures exceeded allowed max of "+parameters.getAllowedFailures() + ", terminating"); + stop(); + } + if (totalCompleted.get() == totalRequested.get()) { + LOG.info("Got all requested ("+totalCompleted.get()+"), finishing"); stop(); } } @@ -190,7 +224,8 @@ public void onContainersCompleted(List containerStatuses) { int exitStatus = status.getExitStatus(); if (0 != exitStatus) { // container failed - if (ContainerExitStatus.ABORTED != exitStatus) { + LOG.warn("Container " + status.getContainerId() + " exited with code "+ exitStatus); + if (ContainerExitStatus.ABORTED != exitStatus) { totalCompleted.incrementAndGet(); totalFailures.incrementAndGet(); } else { @@ -210,20 +245,39 @@ public void onContainersCompleted(List containerStatuses) { public void onContainersAllocated(List allocatedContainers) { LOG.info("Allocating " + allocatedContainers.size() + " container(s)"); Set assigned = Sets.newHashSet(); - for (ContainerTracker tracker : trackers) { - if (tracker.needsContainers()) { - for (Container allocated : allocatedContainers) { - if (!assigned.contains(allocated) && tracker.matches(allocated)) { - tracker.launchContainer(allocated); - assigned.add(allocated); - } - } - } + synchronized (trackers) { + ALLOCATED: for (Container allocated : allocatedContainers) { + LOG.info("Looking for home for "+ allocated.getId().toString()); + for (ContainerTracker tracker : trackers) { + //we need to check tracker.needsContainers() for each container, as the + //previously allocated container may have addressed the containers' need + if (tracker.needsContainers() && ! assigned.contains(allocated) && tracker.matches(allocated)) { + LOG.info("Allocated to " + tracker.toString()); + tracker.launchContainer(allocated); + assigned.add(allocated); + resourceManager.removeContainerRequest(tracker.containerRequest); + continue ALLOCATED; + } + } + } + + +// for (ContainerTracker tracker : trackers) { +// //we need to check tracker.needsContainers() for each container, as the +// //previously allocated container may have addressed the containers' need +// for (Container allocated : allocatedContainers) { +// if (tracker.needsContainers() && ! assigned.contains(allocated) && tracker.matches(allocated)) { +// tracker.launchContainer(allocated); +// assigned.add(allocated); +// resourceManager.removeContainerRequest(tracker.containerRequest); +// } +// } +// } } if (assigned.size() < allocatedContainers.size()) { - LOG.error(String.format("Not all containers were allocated (%d out of %d)", assigned.size(), + LOG.warn(String.format("Not all containers were allocated (%d out of %d)", assigned.size(), allocatedContainers.size())); - stop(); + //stop(); } } @@ -247,7 +301,22 @@ public float getProgress() { if (den == 0) { return 0.0f; } - return ((float) num) / den; + return ((float) num) / (float)den; + } + + @Override + public int getTotalRequested() { + return totalRequested.get(); + } + + @Override + public int getTotalCompleted() { + return totalCompleted.get(); + } + + @Override + public int getTotalFailures() { + return totalFailures.get(); } @Override @@ -256,10 +325,11 @@ public void onError(Throwable throwable) { stop(); } - private class ContainerTracker implements NMClientAsync.CallbackHandler { - private final ContainerLaunchParameters parameters; + protected class ContainerTracker implements NMClientAsync.CallbackHandler { + protected final ContainerLaunchParameters parameters; private final ConcurrentMap containers = Maps.newConcurrentMap(); + private int total; private AtomicInteger needed = new AtomicInteger(); private AtomicInteger started = new AtomicInteger(); private AtomicInteger completed = new AtomicInteger(); @@ -267,26 +337,42 @@ private class ContainerTracker implements NMClientAsync.CallbackHandler { private NMClientAsync nodeManager; private Resource resource; private Priority priority; - private ContainerLaunchContext ctxt; + private String nodeLabelsExpression; + protected ContainerLaunchContext ctxt; + AMRMClient.ContainerRequest containerRequest; + String[] nodes; public ContainerTracker(ContainerLaunchParameters parameters) { this.parameters = parameters; } - public void init(ContainerLaunchContextFactory factory) { + public void init(ContainerLaunchContextFactory factory) { this.nodeManager = NMClientAsync.createNMClientAsync(this); nodeManager.init(conf); nodeManager.start(); this.ctxt = factory.create(parameters); + this.resource = factory.createResource(parameters); this.priority = factory.createPriority(parameters.getPriority()); - AMRMClient.ContainerRequest containerRequest = new AMRMClient.ContainerRequest( + this.nodeLabelsExpression = factory.getNodeLabelExpression(parameters); + nodes = null; + if (parameters.getNode() != null) + { + nodes = new String[]{parameters.getNode()}; + } + + String[] racks = null; + containerRequest = new AMRMClient.ContainerRequest( resource, - null, // nodes - null, // racks - priority); - int numInstances = parameters.getNumInstances(); + nodes, // nodes + racks, // racks + priority, + nodes == null, //we can relax locality only when no node names are specified + nodeLabelsExpression //usually null + ); + int numInstances = total = parameters.getNumInstances(); + LOG.info(this.toString() + " needs " + numInstances + " instances of this container type"); for (int j = 0; j < numInstances; j++) { resourceManager.addContainerRequest(containerRequest); } @@ -321,6 +407,7 @@ public void onContainerStopped(ContainerId containerId) { @Override public void onStartContainerError(ContainerId containerId, Throwable throwable) { LOG.warn("Start container error for container id = " + containerId, throwable); + LOG.warn("LaunchContext was " + this.ctxt.toString()); containers.remove(containerId); completed.incrementAndGet(); failed.incrementAndGet(); @@ -338,16 +425,32 @@ public void onStopContainerError(ContainerId containerId, Throwable throwable) { } public boolean needsContainers() { + LOG.debug(this.toString() + " still needs " + needed.get()); return needed.get() > 0; } public boolean matches(Container c) { - return true; // TODO + LOG.info("Trying to match container " + c.toString() + " with "+ c.getResource().toString()); + LOG.info("... to " + this.resource.toString()); + //if (! c.getResource().equals(this.resource)) + // return false; + if (nodes != null && nodes.length > 0 && ! Arrays.asList(nodes).contains(c.getNodeId().getHost())) + return false; + return true; + } + + public boolean owns(ContainerId cid) { + return containers.containsKey(cid); } public void launchContainer(Container c) { - LOG.info("Launching container id = " + c.getId() + " on node = " + c.getNodeId()); + LOG.info(this.toString() + " is launching container id = " + c.getId() + " on node = " + c.getNodeId() + + " cmd = " + StringUtils.join(" ", ctxt.getCommands()) + + " env= " + ctxt.getEnvironment().toString() + ); + LOG.info("Requested resource is " + ctxt.getLocalResources().toString()); needed.decrementAndGet(); + //LOG.debug(this.toString() + " still needs " + needed.get()); containers.put(c.getId(), c); nodeManager.startContainerAsync(c, ctxt); } @@ -355,6 +458,10 @@ public void launchContainer(Container c) { public boolean hasRunningContainers() { return !containers.isEmpty(); } + + public int getTotalNumInstances() { + return total; + } public void kill() { for (Container c : containers.values()) { @@ -366,4 +473,6 @@ public boolean hasMoreContainers() { return needsContainers() || hasRunningContainers(); } } + + } diff --git a/pom.xml b/pom.xml index 237ccaa..30ef1ff 100644 --- a/pom.xml +++ b/pom.xml @@ -19,7 +19,7 @@ UTF-8 - 2.2.0 + 2.6.0 11.0.2 2.0.2 4.8.2 From 14891a14b560c14982ce5ff0bec2fc1a0861bfb8 Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Sat, 3 Sep 2016 19:26:28 +0100 Subject: [PATCH 07/13] Fixed poor merging --- .../kitten/ContainerLaunchContextFactory.java | 4 -- .../lua/LuaContainerLaunchParameters.java | 5 --- .../service/ApplicationMasterServiceImpl.java | 42 +------------------ 3 files changed, 2 insertions(+), 49 deletions(-) diff --git a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java index 0464dbf..309c09a 100644 --- a/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java +++ b/java/common/src/main/java/com/cloudera/kitten/ContainerLaunchContextFactory.java @@ -44,10 +44,6 @@ public ContainerLaunchContext create(ContainerLaunchParameters parameters) { } return clc; } - - public String getNodeLabelExpression(ContainerLaunchParameters parameters) { - return parameters.getNodeLabelsExpression(); - } public String getNodeLabelExpression(ContainerLaunchParameters parameters) { return parameters.getNodeLabelsExpression(); diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java index 1604f8b..8f0e1ae 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java @@ -73,11 +73,6 @@ public int getCores() { public int getMemory() { return lv.getInteger(LuaFields.MEMORY); } - - @Override - public String getNodeLabelsExpression() { - return lv.isNil(LuaFields.NODE_LABELS) ? null : lv.getString(LuaFields.NODE_LABELS); - } @Override public String getNodeLabelsExpression() { diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index f9aab40..61a33e6 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -246,7 +246,6 @@ public void onContainersAllocated(List allocatedContainers) { LOG.info("Allocating " + allocatedContainers.size() + " container(s)"); Set assigned = Sets.newHashSet(); synchronized (trackers) { -<<<<<<< HEAD ALLOCATED: for (Container allocated : allocatedContainers) { LOG.info("Looking for home for "+ allocated.getId().toString()); for (ContainerTracker tracker : trackers) { @@ -274,18 +273,6 @@ public void onContainersAllocated(List allocatedContainers) { // } // } // } -======= - for (ContainerTracker tracker : trackers) { - //we need to check tracker.needsContainers() for each container, as the - //previously allocated container may have addressed the containers' need - for (Container allocated : allocatedContainers) { - if (tracker.needsContainers() && ! assigned.contains(allocated) && tracker.matches(allocated)) { - tracker.launchContainer(allocated); - assigned.add(allocated); - } - } - } ->>>>>>> 30fb2067419bebdd62083b93e3fe9b9017a707fe } if (assigned.size() < allocatedContainers.size()) { LOG.warn(String.format("Not all containers were allocated (%d out of %d)", assigned.size(), @@ -339,11 +326,7 @@ public void onError(Throwable throwable) { } protected class ContainerTracker implements NMClientAsync.CallbackHandler { -<<<<<<< HEAD protected final ContainerLaunchParameters parameters; -======= - private final ContainerLaunchParameters parameters; ->>>>>>> 30fb2067419bebdd62083b93e3fe9b9017a707fe private final ConcurrentMap containers = Maps.newConcurrentMap(); private int total; @@ -355,23 +338,15 @@ protected class ContainerTracker implements NMClientAsync.CallbackHandler { private Resource resource; private Priority priority; private String nodeLabelsExpression; -<<<<<<< HEAD protected ContainerLaunchContext ctxt; AMRMClient.ContainerRequest containerRequest; String[] nodes; -======= - private ContainerLaunchContext ctxt; ->>>>>>> 30fb2067419bebdd62083b93e3fe9b9017a707fe public ContainerTracker(ContainerLaunchParameters parameters) { this.parameters = parameters; } -<<<<<<< HEAD -======= - @SuppressWarnings("unchecked") ->>>>>>> 30fb2067419bebdd62083b93e3fe9b9017a707fe - public void init(ContainerLaunchContextFactory factory) { + public void init(ContainerLaunchContextFactory factory) { this.nodeManager = NMClientAsync.createNMClientAsync(this); nodeManager.init(conf); nodeManager.start(); @@ -381,7 +356,6 @@ public void init(ContainerLaunchContextFactory factory) { this.resource = factory.createResource(parameters); this.priority = factory.createPriority(parameters.getPriority()); this.nodeLabelsExpression = factory.getNodeLabelExpression(parameters); -<<<<<<< HEAD nodes = null; if (parameters.getNode() != null) { @@ -390,28 +364,16 @@ public void init(ContainerLaunchContextFactory factory) { String[] racks = null; containerRequest = new AMRMClient.ContainerRequest( -======= - String[] nodes = null; - String[] racks = null; - AMRMClient.ContainerRequest containerRequest = new AMRMClient.ContainerRequest( ->>>>>>> 30fb2067419bebdd62083b93e3fe9b9017a707fe resource, nodes, // nodes racks, // racks priority, -<<<<<<< HEAD + nodes == null, //we can relax locality only when no node names are specified nodeLabelsExpression //usually null ); int numInstances = total = parameters.getNumInstances(); LOG.info(this.toString() + " needs " + numInstances + " instances of this container type"); -======= - true, //default for "relaxLocality" - nodeLabelsExpression //usually null - ); - int numInstances = parameters.getNumInstances(); - LOG.debug(this.toString() + " needs " + numInstances + " instances of this container type"); ->>>>>>> 30fb2067419bebdd62083b93e3fe9b9017a707fe for (int j = 0; j < numInstances; j++) { resourceManager.addContainerRequest(containerRequest); } From facac92c19b1b0ca247c51a3415c79eeacf391ca Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Sun, 4 Sep 2016 11:08:05 +0100 Subject: [PATCH 08/13] Hadoop 2.6 has more files in the default directory --- .../java/com/cloudera/kitten/TestKittenDistributedShell.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java/client/src/test/java/com/cloudera/kitten/TestKittenDistributedShell.java b/java/client/src/test/java/com/cloudera/kitten/TestKittenDistributedShell.java index 50f851c..cb8ef57 100644 --- a/java/client/src/test/java/com/cloudera/kitten/TestKittenDistributedShell.java +++ b/java/client/src/test/java/com/cloudera/kitten/TestKittenDistributedShell.java @@ -86,6 +86,6 @@ public void testKittenShell() throws Exception { client.setConf(conf); System.out.println("Running..."); assertEquals(0, client.run(new String[]{config, "distshell"})); - assertEquals(12, Files.readLines(tmpFile, Charsets.UTF_8).size()); + assertEquals(14, Files.readLines(tmpFile, Charsets.UTF_8).size()); } } From d74bc2a6eb8bd9c99b1740e1e695b98edc8adfa5 Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Sun, 5 Feb 2017 10:52:37 +0000 Subject: [PATCH 09/13] - Prevent ConcurrentModificationException while determining progress - Log diagnostic message when a container is fail-stopped --- .../service/ApplicationMasterServiceImpl.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index 61a33e6..cb25898 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -224,7 +224,7 @@ public void onContainersCompleted(List containerStatuses) { int exitStatus = status.getExitStatus(); if (0 != exitStatus) { // container failed - LOG.warn("Container " + status.getContainerId() + " exited with code "+ exitStatus); + LOG.warn("Container " + status.getContainerId() + " exited with code "+ exitStatus + " diagnostic " + status.getDiagnostics()); if (ContainerExitStatus.ABORTED != exitStatus) { totalCompleted.incrementAndGet(); totalFailures.incrementAndGet(); @@ -294,9 +294,11 @@ public void onNodesUpdated(List nodeReports) { @Override public float getProgress() { int num = 0, den = 0; - for (ContainerTracker tracker : trackers) { - num += tracker.completed.get(); - den += tracker.parameters.getNumInstances(); + synchronized (trackers) { + for (ContainerTracker tracker : trackers) { + num += tracker.completed.get(); + den += tracker.parameters.getNumInstances(); + } } if (den == 0) { return 0.0f; From 69ad00082e0712e5ef11a5f61d87a2ce66aeca7b Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Thu, 2 Mar 2017 07:45:13 +0000 Subject: [PATCH 10/13] Master should emit a warning when a resource is reduced --- .../com/cloudera/kitten/lua/LuaContainerLaunchParameters.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java index 8f0e1ae..3aa9040 100644 --- a/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java +++ b/java/common/src/main/java/com/cloudera/kitten/lua/LuaContainerLaunchParameters.java @@ -84,6 +84,10 @@ public Resource getContainerResource(Resource clusterMax) { Resource rsrc = Records.newRecord(Resource.class); rsrc.setMemory(Math.min(clusterMax.getMemory(), getMemory())); rsrc.setVirtualCores(Math.min(clusterMax.getVirtualCores(), getCores())); + if (rsrc.getMemory() < getMemory()) + LOG.warn("Memory reduced from "+getMemory()+" to cluster maximum " + rsrc.getMemory()); + if (rsrc.getMemory() < getMemory()) + LOG.warn("VCores reduced from "+getCores()+" to cluster maximum " + rsrc.getVirtualCores()); return rsrc; } From ec227ebb71bd617473b5e29d1b859491d850ba1f Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Thu, 2 Mar 2017 07:45:43 +0000 Subject: [PATCH 11/13] more verbose logging about maximum, requested and obtained resources --- .../appmaster/service/ApplicationMasterServiceImpl.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index cb25898..73a6818 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -142,6 +142,8 @@ protected void startUp() throws IOException { ContainerLaunchContextFactory factory = new ContainerLaunchContextFactory( registration.getMaximumResourceCapability(), tokens); + LOG.info("Maximum resources in this cluster is " + registration.getMaximumResourceCapability().toString() ); + int totalRequested = 0; for (ContainerLaunchParameters clp : parameters.getContainerLaunchParameters()) { ContainerTracker tracker = getTracker(clp); @@ -376,6 +378,8 @@ public void init(ContainerLaunchContextFactory factory) { ); int numInstances = total = parameters.getNumInstances(); LOG.info(this.toString() + " needs " + numInstances + " instances of this container type"); + LOG.info(this.toString() + " container request is resource=" + + resource.toString() + " nodes="+ nodes + " racks="+racks + " priority="+priority); for (int j = 0; j < numInstances; j++) { resourceManager.addContainerRequest(containerRequest); } @@ -433,7 +437,7 @@ public boolean needsContainers() { } public boolean matches(Container c) { - LOG.info("Trying to match container " + c.toString() + " with "+ c.getResource().toString()); + LOG.info("Trying to match container " + c.toString() + " @ " +c.getNodeId() + " with "+ c.getResource().toString()); LOG.info("... to " + this.resource.toString()); //if (! c.getResource().equals(this.resource)) // return false; From f8239530738b04bdad8ea01b23bd417faf8bd553 Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Wed, 28 Feb 2018 18:40:12 +0000 Subject: [PATCH 12/13] allow a client application to change something in the appContext --- .../cloudera/kitten/client/service/YarnClientServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java b/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java index dc1d11b..ac85319 100644 --- a/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java +++ b/java/client/src/main/java/com/cloudera/kitten/client/service/YarnClientServiceImpl.java @@ -149,7 +149,7 @@ public void run() { stopwatch.start(); } - private void submitApplication(ApplicationSubmissionContext appContext) { + protected void submitApplication(ApplicationSubmissionContext appContext) { LOG.info("Submitting application to the applications manager"); try { yarnClient.submitApplication(appContext); From 9f9904ba1a6c2f6735517710786c531f121659ad Mon Sep 17 00:00:00 2001 From: Craig Macdonald Date: Wed, 14 Nov 2018 11:08:42 +0000 Subject: [PATCH 13/13] better logging --- .../kitten/appmaster/service/ApplicationMasterServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java index 73a6818..1f3f2fc 100644 --- a/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java +++ b/java/master/src/main/java/com/cloudera/kitten/appmaster/service/ApplicationMasterServiceImpl.java @@ -379,7 +379,7 @@ public void init(ContainerLaunchContextFactory factory) { int numInstances = total = parameters.getNumInstances(); LOG.info(this.toString() + " needs " + numInstances + " instances of this container type"); LOG.info(this.toString() + " container request is resource=" - + resource.toString() + " nodes="+ nodes + " racks="+racks + " priority="+priority); + + resource.toString() + " nodes="+ Arrays.toString(nodes) + " racks="+Arrays.toString(racks) + " priority="+priority); for (int j = 0; j < numInstances; j++) { resourceManager.addContainerRequest(containerRequest); }