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..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 @@ -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); @@ -140,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); 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()); } } 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/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..5f65ef8 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..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 @@ -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); } @@ -74,11 +74,20 @@ 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); 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; } @@ -92,6 +101,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 +138,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 +211,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 +272,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..1d54745 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,7 +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"; public static final String ARGS = "args"; 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) { 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/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 + 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..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 @@ -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,33 @@ 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 = 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 +178,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 +208,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 +226,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 + " diagnostic " + status.getDiagnostics()); + if (ContainerExitStatus.ABORTED != exitStatus) { totalCompleted.incrementAndGet(); totalFailures.incrementAndGet(); } else { @@ -210,20 +247,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(); } } @@ -240,14 +296,31 @@ 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; } - 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 +329,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,7 +341,10 @@ 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; @@ -279,14 +356,30 @@ public void init(ContainerLaunchContextFactory factory) { 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"); + LOG.info(this.toString() + " container request is resource=" + + resource.toString() + " nodes="+ Arrays.toString(nodes) + " racks="+Arrays.toString(racks) + " priority="+priority); for (int j = 0; j < numInstances; j++) { resourceManager.addContainerRequest(containerRequest); } @@ -321,6 +414,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 +432,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() + " @ " +c.getNodeId() + " 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 +465,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 +480,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