You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by ch...@apache.org on 2017/02/07 18:54:41 UTC
[02/13] twill git commit: (TWILL-63) Speed up application launch time
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
index 562d1da..5f599d6 100644
--- a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
+++ b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillPreparer.java
@@ -17,10 +17,9 @@
*/
package org.apache.twill.yarn;
-import com.google.common.base.Charsets;
+import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
-import com.google.common.base.Predicates;
import com.google.common.base.Throwables;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.HashMultimap;
@@ -33,6 +32,8 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
+import com.google.common.hash.Hasher;
+import com.google.common.hash.Hashing;
import com.google.common.io.ByteStreams;
import com.google.common.io.OutputSupplier;
import com.google.common.reflect.TypeToken;
@@ -43,6 +44,7 @@ import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.twill.api.ClassAcceptor;
+import org.apache.twill.api.Configs;
import org.apache.twill.api.EventHandlerSpecification;
import org.apache.twill.api.LocalFile;
import org.apache.twill.api.RunId;
@@ -54,10 +56,8 @@ import org.apache.twill.api.TwillSpecification;
import org.apache.twill.api.logging.LogEntry;
import org.apache.twill.api.logging.LogHandler;
import org.apache.twill.filesystem.Location;
-import org.apache.twill.filesystem.LocationFactory;
import org.apache.twill.internal.ApplicationBundler;
import org.apache.twill.internal.Arguments;
-import org.apache.twill.internal.Configs;
import org.apache.twill.internal.Constants;
import org.apache.twill.internal.DefaultLocalFile;
import org.apache.twill.internal.DefaultRuntimeSpecification;
@@ -67,11 +67,11 @@ import org.apache.twill.internal.JvmOptions;
import org.apache.twill.internal.LogOnlyEventHandler;
import org.apache.twill.internal.ProcessController;
import org.apache.twill.internal.ProcessLauncher;
-import org.apache.twill.internal.RunIds;
import org.apache.twill.internal.TwillRuntimeSpecification;
import org.apache.twill.internal.appmaster.ApplicationMasterInfo;
import org.apache.twill.internal.appmaster.ApplicationMasterMain;
import org.apache.twill.internal.container.TwillContainerMain;
+import org.apache.twill.internal.io.LocationCache;
import org.apache.twill.internal.json.ArgumentsCodec;
import org.apache.twill.internal.json.JvmOptionsCodec;
import org.apache.twill.internal.json.LocalFileCodec;
@@ -88,6 +88,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedOutputStream;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -96,6 +97,11 @@ import java.io.Writer;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -112,12 +118,18 @@ import java.util.jar.JarOutputStream;
final class YarnTwillPreparer implements TwillPreparer {
private static final Logger LOG = LoggerFactory.getLogger(YarnTwillPreparer.class);
+ private static final Function<Class<?>, String> CLASS_TO_NAME = new Function<Class<?>, String>() {
+ @Override
+ public String apply(Class<?> cls) {
+ return cls.getName();
+ }
+ };
private final YarnConfiguration yarnConfig;
private final TwillSpecification twillSpec;
private final YarnAppClient yarnAppClient;
private final String zkConnectString;
- private final LocationFactory locationFactory;
+ private final Location appLocation;
private final YarnTwillControllerFactory controllerFactory;
private final RunId runId;
@@ -131,29 +143,37 @@ final class YarnTwillPreparer implements TwillPreparer {
private final List<String> applicationClassPaths = Lists.newArrayList();
private final Credentials credentials;
private final int reservedMemory;
+ private final File localStagingDir;
private final Map<String, Map<String, String>> logLevels = new HashMap<>();
+ private final LocationCache locationCache;
+ private final Set<URL> twillClassPaths;
private String schedulerQueue;
private String extraOptions;
private JvmOptions.DebugOptions debugOptions = JvmOptions.DebugOptions.NO_DEBUG;
private ClassAcceptor classAcceptor;
- YarnTwillPreparer(YarnConfiguration yarnConfig, TwillSpecification twillSpec,
- YarnAppClient yarnAppClient, String zkConnectString,
- LocationFactory locationFactory, String extraOptions, LogEntry.Level logLevel,
- YarnTwillControllerFactory controllerFactory) {
+ YarnTwillPreparer(YarnConfiguration yarnConfig, TwillSpecification twillSpec, RunId runId,
+ YarnAppClient yarnAppClient, String zkConnectString, Location appLocation, Set<URL> twillClassPaths,
+ String extraOptions, LocationCache locationCache, YarnTwillControllerFactory controllerFactory) {
this.yarnConfig = yarnConfig;
this.twillSpec = twillSpec;
+ this.runId = runId;
this.yarnAppClient = yarnAppClient;
this.zkConnectString = zkConnectString;
- this.locationFactory = locationFactory;
+ this.appLocation = appLocation;
this.controllerFactory = controllerFactory;
- this.runId = RunIds.generate();
this.credentials = createCredentials();
this.reservedMemory = yarnConfig.getInt(Configs.Keys.JAVA_RESERVED_MEMORY_MB,
Configs.Defaults.JAVA_RESERVED_MEMORY_MB);
+ this.localStagingDir = new File(yarnConfig.get(Configs.Keys.LOCAL_STAGING_DIRECTORY,
+ Configs.Defaults.LOCAL_STAGING_DIRECTORY));
this.extraOptions = extraOptions;
this.classAcceptor = new ClassAcceptor();
- saveLogLevels(logLevel);
+ this.locationCache = locationCache;
+ this.twillClassPaths = twillClassPaths;
+
+ // By default, the root logger is having INFO log level
+ setLogLevel(LogEntry.Level.INFO);
}
@Override
@@ -338,23 +358,27 @@ final class YarnTwillPreparer implements TwillPreparer {
// Local files needed by AM
Map<String, LocalFile> localFiles = Maps.newHashMap();
- // Local files declared by runnables
- Multimap<String, LocalFile> runnableLocalFiles = HashMultimap.create();
-
- createAppMasterJar(createBundler(), localFiles);
- createContainerJar(createBundler(), localFiles);
- populateRunnableLocalFiles(twillSpec, runnableLocalFiles);
- saveSpecification(twillSpec, runnableLocalFiles, localFiles);
- saveLogback(localFiles);
- saveLauncher(localFiles);
- saveJvmOptions(localFiles);
- saveArguments(new Arguments(arguments, runnableArgs), localFiles);
- saveEnvironments(localFiles);
- saveLocalFiles(localFiles, ImmutableSet.of(Constants.Files.TWILL_SPEC,
- Constants.Files.LOGBACK_TEMPLATE,
- Constants.Files.CONTAINER_JAR,
- Constants.Files.LAUNCHER_JAR,
- Constants.Files.ARGUMENTS));
+
+ createLauncherJar(localFiles);
+ createTwillJar(createBundler(classAcceptor), localFiles);
+ createApplicationJar(createApplicationJarBundler(classAcceptor), localFiles);
+ createResourcesJar(createBundler(classAcceptor), localFiles);
+
+ Path runtimeConfigDir = Files.createTempDirectory(localStagingDir.toPath(),
+ Constants.Files.RUNTIME_CONFIG_JAR);
+ try {
+ saveSpecification(twillSpec, runtimeConfigDir.resolve(Constants.Files.TWILL_SPEC));
+ saveLogback(runtimeConfigDir.resolve(Constants.Files.LOGBACK_TEMPLATE));
+ saveClassPaths(runtimeConfigDir);
+ saveJvmOptions(runtimeConfigDir.resolve(Constants.Files.JVM_OPTIONS));
+ saveArguments(new Arguments(arguments, runnableArgs), runtimeConfigDir.resolve(Constants.Files.ARGUMENTS));
+ saveEnvironments(runtimeConfigDir.resolve(Constants.Files.ENVIRONMENTS));
+ createRuntimeConfigJar(runtimeConfigDir, localFiles);
+ } finally {
+ Paths.deleteRecursively(runtimeConfigDir);
+ }
+
+ createLocalizeFilesJson(localFiles);
LOG.debug("Submit AM container spec: {}", appMasterInfo);
// java -Djava.io.tmpdir=tmp -cp launcher.jar:$HADOOP_CONF_DIR -XmxMemory
@@ -375,7 +399,6 @@ final class YarnTwillPreparer implements TwillPreparer {
"-Xmx" + memory + "m",
extraOptions == null ? "" : extraOptions,
TwillLauncher.class.getName(),
- Constants.Files.APP_MASTER_JAR,
ApplicationMasterMain.class.getName(),
Boolean.FALSE.toString())
.launch();
@@ -406,15 +429,6 @@ final class YarnTwillPreparer implements TwillPreparer {
}
}
- private void saveLogLevels(LogEntry.Level level) {
- Preconditions.checkNotNull(level);
- Map<String, String> appLogLevels = new HashMap<>();
- appLogLevels.put(Logger.ROOT_LOGGER_NAME, level.name());
- for (String runnableName : twillSpec.getRunnables().keySet()) {
- this.logLevels.put(runnableName, appLogLevels);
- }
- }
-
private void saveLogLevels(String runnableName, Map<String, LogEntry.Level> logLevels) {
Map<String, String> newLevels = new HashMap<>();
for (Map.Entry<String, LogEntry.Level> entry : logLevels.entrySet()) {
@@ -430,9 +444,11 @@ final class YarnTwillPreparer implements TwillPreparer {
try {
credentials.addAll(UserGroupInformation.getCurrentUser().getCredentials());
- List<Token<?>> tokens = YarnUtils.addDelegationTokens(yarnConfig, locationFactory, credentials);
- for (Token<?> token : tokens) {
- LOG.debug("Delegation token acquired for {}, {}", locationFactory.getHomeLocation(), token);
+ List<Token<?>> tokens = YarnUtils.addDelegationTokens(yarnConfig, appLocation.getLocationFactory(), credentials);
+ if (LOG.isDebugEnabled()) {
+ for (Token<?> token : tokens) {
+ LOG.debug("Delegation token acquired for {}, {}", appLocation, token);
+ }
}
} catch (IOException e) {
LOG.warn("Failed to check for secure login type. Not gathering any delegation token.", e);
@@ -440,10 +456,6 @@ final class YarnTwillPreparer implements TwillPreparer {
return credentials;
}
- private ApplicationBundler createBundler() {
- return new ApplicationBundler(classAcceptor);
- }
-
private LocalFile createLocalFile(String name, Location location) throws IOException {
return createLocalFile(name, location, false);
}
@@ -452,62 +464,104 @@ final class YarnTwillPreparer implements TwillPreparer {
return new DefaultLocalFile(name, location.toURI(), location.lastModified(), location.length(), archive, null);
}
- private void createAppMasterJar(ApplicationBundler bundler, Map<String, LocalFile> localFiles) throws IOException {
- try {
- LOG.debug("Create and copy {}", Constants.Files.APP_MASTER_JAR);
- Location location = createTempLocation(Constants.Files.APP_MASTER_JAR);
+ private void createTwillJar(final ApplicationBundler bundler, Map<String, LocalFile> localFiles) throws IOException {
+ LOG.debug("Create and copy {}", Constants.Files.TWILL_JAR);
+ Location location = locationCache.get(Constants.Files.TWILL_JAR, new LocationCache.Loader() {
+ @Override
+ public void load(String name, Location targetLocation) throws IOException {
+ // Stuck in the yarnAppClient class to make bundler being able to pickup the right yarn-client version
+ bundler.createBundle(targetLocation, ApplicationMasterMain.class,
+ yarnAppClient.getClass(), TwillContainerMain.class);
+ }
+ });
+
+ LOG.debug("Done {}", Constants.Files.TWILL_JAR);
+ localFiles.put(Constants.Files.TWILL_JAR, createLocalFile(Constants.Files.TWILL_JAR, location, true));
+ }
- List<Class<?>> classes = Lists.newArrayList();
- classes.add(ApplicationMasterMain.class);
+ private void createApplicationJar(final ApplicationBundler bundler,
+ Map<String, LocalFile> localFiles) throws IOException {
+ try {
+ final Set<Class<?>> classes = Sets.newIdentityHashSet();
+ classes.addAll(dependencies);
- // Stuck in the yarnAppClient class to make bundler being able to pickup the right yarn-client version
- classes.add(yarnAppClient.getClass());
+ ClassLoader classLoader = getClassLoader();
+ for (RuntimeSpecification spec : twillSpec.getRunnables().values()) {
+ classes.add(classLoader.loadClass(spec.getRunnableSpecification().getClassName()));
+ }
// Add the TwillRunnableEventHandler class
if (twillSpec.getEventHandler() != null) {
classes.add(getClassLoader().loadClass(twillSpec.getEventHandler().getClassName()));
}
- bundler.createBundle(location, classes);
- LOG.debug("Done {}", Constants.Files.APP_MASTER_JAR);
+ // The location name is computed from the MD5 of all the classes names
+ // The localized name is always APPLICATION_JAR
+ List<String> classList = Lists.newArrayList(Iterables.transform(classes, CLASS_TO_NAME));
+ Collections.sort(classList);
+ Hasher hasher = Hashing.md5().newHasher();
+ for (String name : classList) {
+ hasher.putString(name);
+ }
+ String name = twillSpec.getName() + "-" + hasher.hash().toString() + "-" + Constants.Files.APPLICATION_JAR;
+
+ LOG.debug("Create and copy {}", Constants.Files.APPLICATION_JAR);
+ Location location = locationCache.get(name, new LocationCache.Loader() {
+ @Override
+ public void load(String name, Location targetLocation) throws IOException {
+ bundler.createBundle(targetLocation, classes);
+ }
+ });
+
+ LOG.debug("Done {}", Constants.Files.APPLICATION_JAR);
+
+ localFiles.put(Constants.Files.APPLICATION_JAR, createLocalFile(Constants.Files.APPLICATION_JAR, location, true));
- localFiles.put(Constants.Files.APP_MASTER_JAR, createLocalFile(Constants.Files.APP_MASTER_JAR, location));
} catch (ClassNotFoundException e) {
throw Throwables.propagate(e);
}
}
- private void createContainerJar(ApplicationBundler bundler, Map<String, LocalFile> localFiles) throws IOException {
- try {
- Set<Class<?>> classes = Sets.newIdentityHashSet();
- classes.add(TwillContainerMain.class);
- classes.addAll(dependencies);
-
- ClassLoader classLoader = getClassLoader();
- for (RuntimeSpecification spec : twillSpec.getRunnables().values()) {
- classes.add(classLoader.loadClass(spec.getRunnableSpecification().getClassName()));
- }
+ private void createResourcesJar(ApplicationBundler bundler, Map<String, LocalFile> localFiles) throws IOException {
+ // If there is no resources, no need to create the jar file.
+ if (resources.isEmpty()) {
+ return;
+ }
- LOG.debug("Create and copy {}", Constants.Files.CONTAINER_JAR);
- Location location = createTempLocation(Constants.Files.CONTAINER_JAR);
- bundler.createBundle(location, classes, resources);
- LOG.debug("Done {}", Constants.Files.CONTAINER_JAR);
+ LOG.debug("Create and copy {}", Constants.Files.RESOURCES_JAR);
+ Location location = createTempLocation(Constants.Files.RESOURCES_JAR);
+ bundler.createBundle(location, Collections.<Class<?>>emptyList(), resources);
+ LOG.debug("Done {}", Constants.Files.RESOURCES_JAR);
+ localFiles.put(Constants.Files.RESOURCES_JAR, createLocalFile(Constants.Files.RESOURCES_JAR, location, true));
+ }
- localFiles.put(Constants.Files.CONTAINER_JAR, createLocalFile(Constants.Files.CONTAINER_JAR, location));
+ private void createRuntimeConfigJar(Path dir, Map<String, LocalFile> localFiles) throws IOException {
+ LOG.debug("Create and copy {}", Constants.Files.RUNTIME_CONFIG_JAR);
- } catch (ClassNotFoundException e) {
- throw Throwables.propagate(e);
+ // Jar everything under the given directory, which contains different files needed by AM/runnable containers
+ Location location = createTempLocation(Constants.Files.RUNTIME_CONFIG_JAR);
+ try (
+ JarOutputStream jarOutput = new JarOutputStream(location.getOutputStream());
+ DirectoryStream<Path> stream = Files.newDirectoryStream(dir)
+ ) {
+ for (Path path : stream) {
+ jarOutput.putNextEntry(new JarEntry(path.getFileName().toString()));
+ Files.copy(path, jarOutput);
+ jarOutput.closeEntry();
+ }
}
+
+ LOG.debug("Done {}", Constants.Files.RUNTIME_CONFIG_JAR);
+ localFiles.put(Constants.Files.RUNTIME_CONFIG_JAR,
+ createLocalFile(Constants.Files.RUNTIME_CONFIG_JAR, location, true));
}
/**
* Based on the given {@link TwillSpecification}, upload LocalFiles to Yarn Cluster.
* @param twillSpec The {@link TwillSpecification} for populating resource.
- * @param localFiles A Multimap to store runnable name to transformed LocalFiles.
- * @throws IOException
*/
- private void populateRunnableLocalFiles(TwillSpecification twillSpec,
- Multimap<String, LocalFile> localFiles) throws IOException {
+ private Multimap<String, LocalFile> populateRunnableLocalFiles(TwillSpecification twillSpec) throws IOException {
+ Multimap<String, LocalFile> localFiles = HashMultimap.create();
LOG.debug("Populating Runnable LocalFiles");
for (Map.Entry<String, RuntimeSpecification> entry: twillSpec.getRunnables().entrySet()) {
@@ -516,14 +570,14 @@ final class YarnTwillPreparer implements TwillPreparer {
Location location;
URI uri = localFile.getURI();
- if (locationFactory.getHomeLocation().toURI().getScheme().equals(uri.getScheme())) {
+ if (appLocation.toURI().getScheme().equals(uri.getScheme())) {
// If the source file location is having the same scheme as the target location, no need to copy
- location = locationFactory.create(uri);
+ location = appLocation.getLocationFactory().create(uri);
} else {
URL url = uri.toURL();
LOG.debug("Create and copy {} : {}", runnableName, url);
// Preserves original suffix for expansion.
- location = copyFromURL(url, createTempLocation(Paths.appendSuffix(url.getFile(), localFile.getName())));
+ location = copyFromURL(url, createTempLocation(Paths.addExtension(url.getFile(), localFile.getName())));
LOG.debug("Done {} : {}", runnableName, url);
}
@@ -533,10 +587,12 @@ final class YarnTwillPreparer implements TwillPreparer {
}
}
LOG.debug("Done Runnable LocalFiles");
+ return localFiles;
}
- private void saveSpecification(TwillSpecification spec, final Multimap<String, LocalFile> runnableLocalFiles,
- Map<String, LocalFile> localFiles) throws IOException {
+ private void saveSpecification(TwillSpecification spec, Path targetFile) throws IOException {
+ final Multimap<String, LocalFile> runnableLocalFiles = populateRunnableLocalFiles(spec);
+
// Rewrite LocalFiles inside twillSpec
Map<String, RuntimeSpecification> runtimeSpec = Maps.transformEntries(
spec.getRunnables(), new Maps.EntryTransformer<String, RuntimeSpecification, RuntimeSpecification>() {
@@ -548,9 +604,8 @@ final class YarnTwillPreparer implements TwillPreparer {
});
// Serialize into a local temp file.
- LOG.debug("Create and copy {}", Constants.Files.TWILL_SPEC);
- Location location = createTempLocation(Constants.Files.TWILL_SPEC);
- try (Writer writer = new OutputStreamWriter(location.getOutputStream(), Charsets.UTF_8)) {
+ LOG.debug("Creating {}", targetFile);
+ try (Writer writer = Files.newBufferedWriter(targetFile, StandardCharsets.UTF_8)) {
EventHandlerSpecification eventHandler = spec.getEventHandler();
if (eventHandler == null) {
eventHandler = new LogOnlyEventHandler().configure();
@@ -558,136 +613,124 @@ final class YarnTwillPreparer implements TwillPreparer {
TwillSpecification newTwillSpec = new DefaultTwillSpecification(spec.getName(), runtimeSpec, spec.getOrders(),
spec.getPlacementPolicies(), eventHandler);
TwillRuntimeSpecificationAdapter.create().toJson(
- new TwillRuntimeSpecification(newTwillSpec, locationFactory.getHomeLocation().getName(),
- getAppLocation().toURI(), zkConnectString, runId, twillSpec.getName(),
+ new TwillRuntimeSpecification(newTwillSpec, appLocation.getLocationFactory().getHomeLocation().getName(),
+ appLocation.toURI(), zkConnectString, runId, twillSpec.getName(),
reservedMemory, yarnConfig.get(YarnConfiguration.RM_SCHEDULER_ADDRESS),
logLevels), writer);
}
- LOG.debug("Done {}", Constants.Files.TWILL_SPEC);
-
- localFiles.put(Constants.Files.TWILL_SPEC, createLocalFile(Constants.Files.TWILL_SPEC, location));
+ LOG.debug("Done {}", targetFile);
}
- private void saveLogback(Map<String, LocalFile> localFiles) throws IOException {
- LOG.debug("Create and copy {}", Constants.Files.LOGBACK_TEMPLATE);
- Location location = copyFromURL(getClass().getClassLoader().getResource(Constants.Files.LOGBACK_TEMPLATE),
- createTempLocation(Constants.Files.LOGBACK_TEMPLATE));
- LOG.debug("Done {}", Constants.Files.LOGBACK_TEMPLATE);
+ private void saveLogback(Path targetFile) throws IOException {
+ URL url = getClass().getClassLoader().getResource(Constants.Files.LOGBACK_TEMPLATE);
+ if (url == null) {
+ return;
+ }
- localFiles.put(Constants.Files.LOGBACK_TEMPLATE, createLocalFile(Constants.Files.LOGBACK_TEMPLATE, location));
+ LOG.debug("Creating {}", targetFile);
+ try (InputStream is = url.openStream()) {
+ Files.copy(is, targetFile);
+ }
+ LOG.debug("Done {}", targetFile);
}
/**
* Creates the launcher.jar for launch the main application.
*/
- private void saveLauncher(Map<String, LocalFile> localFiles) throws URISyntaxException, IOException {
+ private void createLauncherJar(Map<String, LocalFile> localFiles) throws URISyntaxException, IOException {
LOG.debug("Create and copy {}", Constants.Files.LAUNCHER_JAR);
- Location location = createTempLocation(Constants.Files.LAUNCHER_JAR);
- final String launcherName = TwillLauncher.class.getName();
- final String portFinderName = FindFreePort.class.getName();
-
- // Create a jar file with the TwillLauncher optionally a json serialized classpath.json in it.
- // Also a little utility to find a free port, used for debugging.
- final JarOutputStream jarOut = new JarOutputStream(location.getOutputStream());
- ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
- if (classLoader == null) {
- classLoader = getClass().getClassLoader();
- }
- Dependencies.findClassDependencies(classLoader, new ClassAcceptor() {
+ Location location = locationCache.get(Constants.Files.LAUNCHER_JAR, new LocationCache.Loader() {
@Override
- public boolean accept(String className, URL classUrl, URL classPathUrl) {
- Preconditions.checkArgument(className.startsWith(launcherName) || className.equals(portFinderName),
- "Launcher jar should not have dependencies: %s", className);
- try {
- jarOut.putNextEntry(new JarEntry(className.replace('.', '/') + ".class"));
- try (InputStream is = classUrl.openStream()) {
- ByteStreams.copy(is, jarOut);
+ public void load(String name, Location targetLocation) throws IOException {
+ // Create a jar file with the TwillLauncher and FindFreePort and dependent classes inside.
+ try (final JarOutputStream jarOut = new JarOutputStream(targetLocation.getOutputStream())) {
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ if (classLoader == null) {
+ classLoader = getClass().getClassLoader();
}
- } catch (IOException e) {
- throw Throwables.propagate(e);
+ Dependencies.findClassDependencies(classLoader, new ClassAcceptor() {
+ @Override
+ public boolean accept(String className, URL classUrl, URL classPathUrl) {
+ try {
+ jarOut.putNextEntry(new JarEntry(className.replace('.', '/') + ".class"));
+ try (InputStream is = classUrl.openStream()) {
+ ByteStreams.copy(is, jarOut);
+ }
+ } catch (IOException e) {
+ throw Throwables.propagate(e);
+ }
+ return true;
+ }
+ }, TwillLauncher.class.getName(), FindFreePort.class.getName());
}
- return true;
}
- }, launcherName, portFinderName);
+ });
- try {
- addClassPaths(Constants.CLASSPATH, classPaths, jarOut);
- addClassPaths(Constants.APPLICATION_CLASSPATH, applicationClassPaths, jarOut);
- } finally {
- jarOut.close();
- }
LOG.debug("Done {}", Constants.Files.LAUNCHER_JAR);
localFiles.put(Constants.Files.LAUNCHER_JAR, createLocalFile(Constants.Files.LAUNCHER_JAR, location));
}
- private void addClassPaths(String classpathId, List<String> classPaths, JarOutputStream jarOut) throws IOException {
- if (!classPaths.isEmpty()) {
- jarOut.putNextEntry(new JarEntry(classpathId));
- jarOut.write(Joiner.on(':').join(classPaths).getBytes(Charsets.UTF_8));
- }
+ private void saveClassPaths(Path targetDir) throws IOException {
+ Files.write(targetDir.resolve(Constants.Files.APPLICATION_CLASSPATH),
+ Joiner.on(':').join(applicationClassPaths).getBytes(StandardCharsets.UTF_8));
+ Files.write(targetDir.resolve(Constants.Files.CLASSPATH),
+ Joiner.on(':').join(classPaths).getBytes(StandardCharsets.UTF_8));
}
- private void saveJvmOptions(Map<String, LocalFile> localFiles) throws IOException {
+ private void saveJvmOptions(final Path targetPath) throws IOException {
if ((extraOptions == null || extraOptions.isEmpty()) &&
JvmOptions.DebugOptions.NO_DEBUG.equals(debugOptions)) {
// If no vm options, no need to localize the file.
return;
}
- LOG.debug("Create and copy {}", Constants.Files.JVM_OPTIONS);
- final Location location = createTempLocation(Constants.Files.JVM_OPTIONS);
+ LOG.debug("Creating {}", targetPath);
JvmOptionsCodec.encode(new JvmOptions(extraOptions, debugOptions), new OutputSupplier<Writer>() {
@Override
public Writer getOutput() throws IOException {
- return new OutputStreamWriter(location.getOutputStream(), Charsets.UTF_8);
+ return Files.newBufferedWriter(targetPath, StandardCharsets.UTF_8);
}
});
- LOG.debug("Done {}", Constants.Files.JVM_OPTIONS);
-
- localFiles.put(Constants.Files.JVM_OPTIONS, createLocalFile(Constants.Files.JVM_OPTIONS, location));
+ LOG.debug("Done {}", targetPath);
}
- private void saveArguments(Arguments arguments, Map<String, LocalFile> localFiles) throws IOException {
- LOG.debug("Create and copy {}", Constants.Files.ARGUMENTS);
- final Location location = createTempLocation(Constants.Files.ARGUMENTS);
+ private void saveArguments(Arguments arguments, final Path targetPath) throws IOException {
+ LOG.debug("Creating {}", targetPath);
ArgumentsCodec.encode(arguments, new OutputSupplier<Writer>() {
@Override
public Writer getOutput() throws IOException {
- return new OutputStreamWriter(location.getOutputStream(), Charsets.UTF_8);
+ return Files.newBufferedWriter(targetPath, StandardCharsets.UTF_8);
}
});
- LOG.debug("Done {}", Constants.Files.ARGUMENTS);
-
- localFiles.put(Constants.Files.ARGUMENTS, createLocalFile(Constants.Files.ARGUMENTS, location));
+ LOG.debug("Done {}", targetPath);
}
- private void saveEnvironments(Map<String, LocalFile> localFiles) throws IOException {
+ private void saveEnvironments(Path targetPath) throws IOException {
if (environments.isEmpty()) {
return;
}
- LOG.debug("Create and copy {}", Constants.Files.ENVIRONMENTS);
- final Location location = createTempLocation(Constants.Files.ENVIRONMENTS);
- try (Writer writer = new OutputStreamWriter(location.getOutputStream(), Charsets.UTF_8)) {
+ LOG.debug("Creating {}", targetPath);
+ try (Writer writer = Files.newBufferedWriter(targetPath, StandardCharsets.UTF_8)) {
new Gson().toJson(environments, writer);
}
- LOG.debug("Done {}", Constants.Files.ENVIRONMENTS);
-
- localFiles.put(Constants.Files.ENVIRONMENTS, createLocalFile(Constants.Files.ENVIRONMENTS, location));
+ LOG.debug("Done {}", targetPath);
}
/**
- * Serializes the list of files that needs to localize from AM to Container.
+ * Serializes the information for files that are localized to all YARN containers.
*/
- private void saveLocalFiles(Map<String, LocalFile> localFiles, Set<String> includes) throws IOException {
- Map<String, LocalFile> localize = ImmutableMap.copyOf(Maps.filterKeys(localFiles, Predicates.in(includes)));
+ private void createLocalizeFilesJson(Map<String, LocalFile> localFiles) throws IOException {
LOG.debug("Create and copy {}", Constants.Files.LOCALIZE_FILES);
Location location = createTempLocation(Constants.Files.LOCALIZE_FILES);
- try (Writer writer = new OutputStreamWriter(location.getOutputStream(), Charsets.UTF_8)) {
+
+ // Serialize the list of LocalFiles, except the one we are generating here, as this file is used by AM only.
+ // This file should never use LocationCache.
+ try (Writer writer = new OutputStreamWriter(location.getOutputStream(), StandardCharsets.UTF_8)) {
new GsonBuilder().registerTypeAdapter(LocalFile.class, new LocalFileCodec())
- .create().toJson(localize.values(), new TypeToken<List<LocalFile>>() {
+ .create().toJson(localFiles.values(), new TypeToken<List<LocalFile>>() {
}.getType(), writer);
}
LOG.debug("Done {}", Constants.Files.LOCALIZE_FILES);
@@ -711,16 +754,12 @@ final class YarnTwillPreparer implements TwillPreparer {
name = fileName.substring(0, fileName.length() - suffix.length() - 1);
try {
- return getAppLocation().append(name).getTempFile('.' + suffix);
+ return appLocation.append(name).getTempFile('.' + suffix);
} catch (IOException e) {
throw Throwables.propagate(e);
}
}
- private Location getAppLocation() {
- return locationFactory.create(String.format("/%s/%s", twillSpec.getName(), runId.getId()));
- }
-
/**
* Returns the context ClassLoader if there is any, otherwise, returns ClassLoader of this class.
*/
@@ -728,4 +767,23 @@ final class YarnTwillPreparer implements TwillPreparer {
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
return classLoader == null ? getClass().getClassLoader() : classLoader;
}
+
+ private ApplicationBundler createBundler(ClassAcceptor classAcceptor) {
+ return new ApplicationBundler(classAcceptor).setTempDir(localStagingDir);
+ }
+
+ /**
+ * Creates a {@link ApplicationBundler} for building application jar. The bundler will include classes
+ * accepted by the given {@link ClassAcceptor}, as long as it is not a twill class.
+ */
+ private ApplicationBundler createApplicationJarBundler(final ClassAcceptor classAcceptor) {
+ // Accept classes based on the classAcceptor and also excluding all twill classes as they are already
+ // in the twill.jar
+ return createBundler(new ClassAcceptor() {
+ @Override
+ public boolean accept(String className, URL classUrl, URL classPathUrl) {
+ return !twillClassPaths.contains(classPathUrl) && classAcceptor.accept(className, classUrl, classPathUrl);
+ }
+ });
+ }
}
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
index 1b7c4b9..405eb24 100644
--- a/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
+++ b/twill-yarn/src/main/java/org/apache/twill/yarn/YarnTwillRunnerService.java
@@ -17,7 +17,7 @@
*/
package org.apache.twill.yarn;
-import com.google.common.base.Charsets;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
@@ -34,20 +34,18 @@ import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.collect.Table;
import com.google.common.util.concurrent.AbstractIdleService;
-import com.google.common.util.concurrent.Callables;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Service;
-import com.google.gson.Gson;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.security.Credentials;
import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.twill.api.ClassAcceptor;
+import org.apache.twill.api.Configs;
+import org.apache.twill.api.LocalFile;
import org.apache.twill.api.ResourceSpecification;
import org.apache.twill.api.RunId;
import org.apache.twill.api.SecureStore;
@@ -58,23 +56,24 @@ import org.apache.twill.api.TwillPreparer;
import org.apache.twill.api.TwillRunnable;
import org.apache.twill.api.TwillRunnerService;
import org.apache.twill.api.TwillSpecification;
-import org.apache.twill.api.logging.LogEntry;
import org.apache.twill.api.logging.LogHandler;
import org.apache.twill.common.Cancellable;
import org.apache.twill.common.Threads;
import org.apache.twill.filesystem.FileContextLocationFactory;
import org.apache.twill.filesystem.Location;
import org.apache.twill.filesystem.LocationFactory;
-import org.apache.twill.internal.Configs;
import org.apache.twill.internal.Constants;
import org.apache.twill.internal.ProcessController;
import org.apache.twill.internal.RunIds;
import org.apache.twill.internal.SingleRunnableApplication;
import org.apache.twill.internal.appmaster.ApplicationMasterLiveNodeData;
+import org.apache.twill.internal.io.BasicLocationCache;
+import org.apache.twill.internal.io.LocationCache;
+import org.apache.twill.internal.io.NoCachingLocationCache;
+import org.apache.twill.internal.utils.Dependencies;
import org.apache.twill.internal.yarn.VersionDetectYarnAppClientFactory;
import org.apache.twill.internal.yarn.YarnAppClient;
import org.apache.twill.internal.yarn.YarnApplicationReport;
-import org.apache.twill.internal.yarn.YarnUtils;
import org.apache.twill.zookeeper.NodeChildren;
import org.apache.twill.zookeeper.NodeData;
import org.apache.twill.zookeeper.RetryStrategies;
@@ -93,6 +92,8 @@ import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.net.URL;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
@@ -109,7 +110,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
public final class YarnTwillRunnerService implements TwillRunnerService {
private static final Logger LOG = LoggerFactory.getLogger(YarnTwillRunnerService.class);
-
private static final int ZK_TIMEOUT = 10000;
private static final Function<String, RunId> STRING_TO_RUN_ID = new Function<String, RunId>() {
@Override
@@ -130,8 +130,11 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
private final ZKClientService zkClientService;
private final LocationFactory locationFactory;
private final Table<String, RunId, YarnTwillController> controllers;
+ private final Set<URL> twillClassPaths;
// A Guava service to help the state transition.
private final Service serviceDelegate;
+ private LocationCache locationCache;
+ private LocationCacheCleaner locationCacheCleaner;
private ScheduledExecutorService secureStoreScheduler;
private Iterable<LiveInfo> liveInfos;
@@ -163,6 +166,7 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
this.locationFactory = locationFactory;
this.zkClientService = getZKClientService(zkConnect);
this.controllers = HashBasedTable.create();
+ this.twillClassPaths = new HashSet<>();
this.serviceDelegate = new AbstractIdleService() {
@Override
protected void startUp() throws Exception {
@@ -277,9 +281,16 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
Preconditions.checkState(serviceDelegate.isRunning(), "Service not start. Please call start() first.");
final TwillSpecification twillSpec = application.configure();
final String appName = twillSpec.getName();
+ RunId runId = RunIds.generate();
+ Location appLocation = locationFactory.create(String.format("/%s/%s", twillSpec.getName(), runId.getId()));
+ LocationCache locationCache = this.locationCache;
+ if (locationCache == null) {
+ locationCache = new NoCachingLocationCache(appLocation);
+ }
- return new YarnTwillPreparer(yarnConfig, twillSpec, yarnAppClient, zkClientService.getConnectString(),
- locationFactory, jvmOptions, LogEntry.Level.INFO, new YarnTwillControllerFactory() {
+ return new YarnTwillPreparer(yarnConfig, twillSpec, runId, yarnAppClient,
+ zkClientService.getConnectString(), appLocation, twillClassPaths, jvmOptions,
+ locationCache, new YarnTwillControllerFactory() {
@Override
public YarnTwillController create(RunId runId, Iterable<LogHandler> logHandlers,
Callable<ProcessController<YarnApplicationReport>> startUp,
@@ -324,6 +335,19 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
private void startUp() throws Exception {
zkClientService.startAndWait();
+ // Find all the classpaths for Twill classes. It is used for class filtering when building application jar
+ // in the YarnTwillPreparer
+ Dependencies.findClassDependencies(getClass().getClassLoader(), new ClassAcceptor() {
+ @Override
+ public boolean accept(String className, URL classUrl, URL classPathUrl) {
+ if (!className.startsWith("org.apache.twill.")) {
+ return false;
+ }
+ twillClassPaths.add(classPathUrl);
+ return true;
+ }
+ }, getClass().getName());
+
// Create the root node, so that the namespace root would get created if it is missing
// If the exception is caused by node exists, then it's ok. Otherwise propagate the exception.
ZKOperations.ignoreError(zkClientService.create("/", null, CreateMode.PERSISTENT),
@@ -346,6 +370,70 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
scheduleSecureStoreUpdate(new LocationSecureStoreUpdater(yarnConfig, locationFactory),
delay, delay, TimeUnit.MILLISECONDS);
}
+
+ // Optionally create a LocationCache
+ String cacheDir = yarnConfig.get(Configs.Keys.LOCATION_CACHE_DIR);
+ if (cacheDir != null) {
+ String sessionId = Long.toString(System.currentTimeMillis());
+ try {
+ Location cacheBase = locationFactory.create(cacheDir);
+ cacheBase.mkdirs();
+ cacheBase.setPermissions("775");
+
+ // Use a unique cache directory for each instance of this class
+ Location cacheLocation = cacheBase.append(sessionId);
+ cacheLocation.mkdirs();
+ cacheLocation.setPermissions("775");
+
+ locationCache = new BasicLocationCache(cacheLocation);
+ locationCacheCleaner = startLocationCacheCleaner(cacheBase, sessionId);
+ } catch (IOException e) {
+ LOG.warn("Failed to create location cache directory. Location cache cannot be enabled.", e);
+ }
+ }
+ }
+
+ /**
+ * Forces a cleanup of location cache based on the given time.
+ */
+ @VisibleForTesting
+ void forceLocationCacheCleanup(long currentTime) {
+ locationCacheCleaner.forceCleanup(currentTime);
+ }
+
+ private LocationCacheCleaner startLocationCacheCleaner(final Location cacheBase, final String sessionId) {
+ LocationCacheCleaner cleaner = new LocationCacheCleaner(
+ yarnConfig, cacheBase, sessionId, new Predicate<Location>() {
+ @Override
+ public boolean apply(Location location) {
+ // Collects all the locations that is being used by any live applications
+ Set<Location> activeLocations = new HashSet<>();
+ synchronized (YarnTwillRunnerService.this) {
+ for (YarnTwillController controller : controllers.values()) {
+ ApplicationMasterLiveNodeData amLiveNodeData = controller.getApplicationMasterLiveNodeData();
+ if (amLiveNodeData != null) {
+ for (LocalFile localFile : amLiveNodeData.getLocalFiles()) {
+ activeLocations.add(locationFactory.create(localFile.getURI()));
+ }
+ }
+ }
+ }
+
+ try {
+ // Always keep the launcher.jar and twill.jar from the current session as they should never change,
+ // hence never expires
+ activeLocations.add(cacheBase.append(sessionId).append(Constants.Files.LAUNCHER_JAR));
+ activeLocations.add(cacheBase.append(sessionId).append(Constants.Files.TWILL_JAR));
+ } catch (IOException e) {
+ // This should not happen
+ LOG.warn("Failed to construct cache location", e);
+ }
+
+ return !activeLocations.contains(location);
+ }
+ });
+ cleaner.startAndWait();
+ return cleaner;
}
private void shutDown() throws Exception {
@@ -354,6 +442,9 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
// when the JVM process is about to exit. Hence it is important that threads created in the controllers are
// daemon threads.
synchronized (this) {
+ if (locationCacheCleaner != null) {
+ locationCacheCleaner.stopAndWait();
+ }
if (secureStoreScheduler != null) {
secureStoreScheduler.shutdownNow();
}
@@ -496,8 +587,9 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
if (cancelled.get()) {
return;
}
- ApplicationId appId = getApplicationId(result);
- if (appId == null) {
+
+ ApplicationMasterLiveNodeData amLiveNodeData = ApplicationMasterLiveNodeDecoder.decode(result);
+ if (amLiveNodeData == null) {
return;
}
@@ -505,8 +597,7 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
if (!controllers.contains(appName, runId)) {
ZKClient zkClient = ZKClients.namespace(zkClientService, "/" + appName);
YarnTwillController controller = listenController(
- new YarnTwillController(appName, runId, zkClient,
- Callables.returning(yarnAppClient.createProcessController(appId))));
+ new YarnTwillController(appName, runId, zkClient, amLiveNodeData, yarnAppClient));
controllers.put(appName, runId, controller);
controller.start();
}
@@ -521,40 +612,6 @@ public final class YarnTwillRunnerService implements TwillRunnerService {
}
- /**
- * Decodes application ID stored inside the node data.
- * @param nodeData The node data to decode from. If it is {@code null}, this method would return {@code null}.
- * @return The ApplicationId or {@code null} if failed to decode.
- */
- private ApplicationId getApplicationId(NodeData nodeData) {
- byte[] data = nodeData == null ? null : nodeData.getData();
- if (data == null) {
- return null;
- }
-
- Gson gson = new Gson();
- JsonElement json = gson.fromJson(new String(data, Charsets.UTF_8), JsonElement.class);
- if (!json.isJsonObject()) {
- LOG.warn("Unable to decode live data node.");
- return null;
- }
-
- JsonObject jsonObj = json.getAsJsonObject();
- json = jsonObj.get("data");
- if (!json.isJsonObject()) {
- LOG.warn("Property data not found in live data node.");
- return null;
- }
-
- try {
- ApplicationMasterLiveNodeData amLiveNode = gson.fromJson(json, ApplicationMasterLiveNodeData.class);
- return YarnUtils.createApplicationId(amLiveNode.getAppIdClusterTime(), amLiveNode.getAppId());
- } catch (Exception e) {
- LOG.warn("Failed to decode application live node data.", e);
- return null;
- }
- }
-
private void updateSecureStores(Table<String, RunId, SecureStore> secureStores) {
for (Table.Cell<String, RunId, SecureStore> cell : secureStores.cellSet()) {
Object store = cell.getValue().getStore();
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/test/java/org/apache/twill/yarn/BaseYarnTest.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/BaseYarnTest.java b/twill-yarn/src/test/java/org/apache/twill/yarn/BaseYarnTest.java
index 4f95391..4c7d84b 100644
--- a/twill-yarn/src/test/java/org/apache/twill/yarn/BaseYarnTest.java
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/BaseYarnTest.java
@@ -21,6 +21,7 @@ import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
import org.apache.hadoop.yarn.api.records.NodeReport;
+import org.apache.twill.api.Configs;
import org.apache.twill.api.TwillController;
import org.apache.twill.api.TwillRunner;
import org.junit.After;
@@ -51,7 +52,7 @@ public abstract class BaseYarnTest {
* A singleton wrapper so that yarn cluster only bring up once across all tests in the YarnTestSuite.
*/
@ClassRule
- public static final TwillTester TWILL_TESTER = new TwillTester() {
+ public static final TwillTester TWILL_TESTER = new TwillTester(Configs.Keys.LOCATION_CACHE_DIR, ".cache") {
private final AtomicInteger instances = new AtomicInteger();
@Override
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/test/java/org/apache/twill/yarn/LocalFileTestRun.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/LocalFileTestRun.java b/twill-yarn/src/test/java/org/apache/twill/yarn/LocalFileTestRun.java
index e20dd64..b646bbc 100644
--- a/twill-yarn/src/test/java/org/apache/twill/yarn/LocalFileTestRun.java
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/LocalFileTestRun.java
@@ -19,7 +19,6 @@ package org.apache.twill.yarn;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
-import com.google.common.io.ByteStreams;
import com.google.common.io.Files;
import com.google.common.io.LineReader;
import org.apache.twill.api.TwillApplication;
@@ -36,12 +35,13 @@ import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
-import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.jar.JarEntry;
import java.util.jar.JarOutputStream;
@@ -53,15 +53,23 @@ public final class LocalFileTestRun extends BaseYarnTest {
@Test
public void testLocalFile() throws Exception {
- String header = Files.readFirstLine(new File(getClass().getClassLoader().getResource("header.txt").toURI()),
- Charsets.UTF_8);
+ // Generate a header and a footer files.
+ File headerFile = tmpFolder.newFile("header.txt");
+ File footerFile = tmpFolder.newFile("footer.txt");
+
+ String headerMsg = "Header Message";
+ String footerMsg = "Footer Message";
+
+ Files.write(headerMsg, headerFile, StandardCharsets.UTF_8);
+ Files.write(footerMsg, footerFile, StandardCharsets.UTF_8);
TwillRunner runner = getTwillRunner();
- TwillController controller = runner.prepare(new LocalFileApplication())
+ TwillController controller = runner.prepare(new LocalFileApplication(headerFile))
.addJVMOptions(" -verbose:gc -Xloggc:gc.log -XX:+PrintGCDetails")
.withApplicationArguments("local")
.withArguments("LocalFileSocketServer", "local2")
+ .withResources(footerFile.toURI())
.addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
.start();
@@ -75,8 +83,9 @@ public final class LocalFileTestRun extends BaseYarnTest {
String msg = "Local file test";
writer.println(msg);
- Assert.assertEquals(header, reader.readLine());
+ Assert.assertEquals(headerMsg, reader.readLine());
Assert.assertEquals(msg, reader.readLine());
+ Assert.assertEquals(footerMsg, reader.readLine());
}
controller.terminate().get(120, TimeUnit.SECONDS);
@@ -91,14 +100,14 @@ public final class LocalFileTestRun extends BaseYarnTest {
*/
public static final class LocalFileApplication implements TwillApplication {
- private final File headerFile;
+ private final File headerJar;
- public LocalFileApplication() throws Exception {
+ public LocalFileApplication(File headerFile) throws Exception {
// Create a jar file that contains the header.txt file inside.
- headerFile = tmpFolder.newFile("header.jar");
- try (JarOutputStream os = new JarOutputStream(new FileOutputStream(headerFile))) {
- os.putNextEntry(new JarEntry("header.txt"));
- ByteStreams.copy(getClass().getClassLoader().getResourceAsStream("header.txt"), os);
+ headerJar = tmpFolder.newFile("header.jar");
+ try (JarOutputStream os = new JarOutputStream(new FileOutputStream(headerJar))) {
+ os.putNextEntry(new JarEntry(headerFile.getName()));
+ Files.copy(headerFile, os);
}
}
@@ -109,7 +118,7 @@ public final class LocalFileTestRun extends BaseYarnTest {
.withRunnable()
.add(new LocalFileSocketServer())
.withLocalFiles()
- .add("header", headerFile, true).apply()
+ .add("header", headerJar, true).apply()
.anyOrder()
.build();
}
@@ -123,14 +132,21 @@ public final class LocalFileTestRun extends BaseYarnTest {
private static final Logger LOG = LoggerFactory.getLogger(LocalFileSocketServer.class);
@Override
- public void handleRequest(BufferedReader reader, PrintWriter writer) throws IOException {
+ public void handleRequest(BufferedReader reader, PrintWriter writer) throws Exception {
// Verify there is a gc.log file locally
Preconditions.checkState(new File("gc.log").exists());
+ // Get the footer file from classloader. Since it was added as resources as well, it should be loadable from CL.
+ URL footerURL = getClass().getClassLoader().getResource("footer.txt");
+ Preconditions.checkState(footerURL != null, "Missing footer.txt file from classloader");
+
LOG.info("handleRequest");
- String header = Files.toString(new File("header/header.txt"), Charsets.UTF_8);
- writer.write(header);
+ // Read from the localized file
+ writer.println(Files.readFirstLine(new File("header/header.txt"), Charsets.UTF_8));
+ // Read from the request
writer.println(reader.readLine());
+ // Read from resource
+ writer.println(Files.readFirstLine(new File(footerURL.toURI()), Charsets.UTF_8));
LOG.info("Flushed response");
}
}
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/test/java/org/apache/twill/yarn/LocationCacheTest.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/LocationCacheTest.java b/twill-yarn/src/test/java/org/apache/twill/yarn/LocationCacheTest.java
new file mode 100644
index 0000000..8ed9ae4
--- /dev/null
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/LocationCacheTest.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.yarn;
+
+import com.google.common.util.concurrent.Uninterruptibles;
+import org.apache.twill.api.AbstractTwillRunnable;
+import org.apache.twill.api.Configs;
+import org.apache.twill.api.ResourceReport;
+import org.apache.twill.api.TwillController;
+import org.apache.twill.api.TwillRunner;
+import org.apache.twill.api.TwillRunnerService;
+import org.apache.twill.api.logging.PrinterLogHandler;
+import org.apache.twill.filesystem.Location;
+import org.apache.twill.filesystem.LocationFactory;
+import org.apache.twill.internal.Constants;
+import org.apache.twill.internal.io.LocationCache;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.io.PrintWriter;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Unit test for {@link LocationCache} usage in {@link YarnTwillRunnerService}.
+ */
+public class LocationCacheTest {
+
+ // Create a TwillTester with LocationCache enabled
+ @ClassRule
+ public static final TwillTester TWILL_TESTER = new TwillTester(Configs.Keys.LOCATION_CACHE_DIR, ".cache");
+
+ @Test(timeout = 120000L)
+ public void testLocationCache() throws Exception {
+ TwillRunner twillRunner = TWILL_TESTER.getTwillRunner();
+
+ // Start the runnable
+ TwillController controller = twillRunner.prepare(new BlockingTwillRunnable())
+ .addLogHandler(new PrinterLogHandler(new PrintWriter(System.out, true)))
+ .start();
+
+ // Wait until the runnable is runnable
+ String runnableName = BlockingTwillRunnable.class.getSimpleName();
+ ResourceReport resourceReport = controller.getResourceReport();
+ while (resourceReport == null || resourceReport.getRunnableResources(runnableName).isEmpty()) {
+ TimeUnit.SECONDS.sleep(1);
+ resourceReport = controller.getResourceReport();
+ }
+
+ long startTime = System.currentTimeMillis();
+
+ // Inspect the cache directory, there should be a directory, which is the current session
+ // inside that directory, there should be three files, launcher.jar, twill.jar and an application jar
+ LocationFactory locationFactory = TWILL_TESTER.createLocationFactory();
+ Location cacheBase = locationFactory.create(".cache");
+
+ List<Location> cacheDirs = cacheBase.list();
+ Assert.assertEquals(1, cacheDirs.size());
+
+ Location currentSessionCache = cacheDirs.get(0);
+ Assert.assertEquals(3, currentSessionCache.list().size());
+
+ // Force a cleanup of cache. The first call is to collect the locations to be cleanup.
+ // The second call is the actual cleanup.
+ ((YarnTwillRunnerService) twillRunner).forceLocationCacheCleanup(startTime);
+ ((YarnTwillRunnerService) twillRunner).forceLocationCacheCleanup(startTime +
+ Configs.Defaults.LOCATION_CACHE_EXPIRY_MS);
+
+ // Since the app is still runnable, no files in the cache should get removed.
+ Assert.assertEquals(3, currentSessionCache.list().size());
+
+ // Stop the app
+ controller.terminate().get();
+
+ // Force a cleanup of cache. The first call is to collect the locations to be cleanup.
+ // The second call is the actual cleanup.
+ ((YarnTwillRunnerService) twillRunner).forceLocationCacheCleanup(startTime);
+ ((YarnTwillRunnerService) twillRunner).forceLocationCacheCleanup(startTime +
+ Configs.Defaults.LOCATION_CACHE_EXPIRY_MS);
+
+ // Since the app is stopped, there should only be two files, the launcher.jar and twill.jar, as they
+ // will never get removed for the current session.
+ Set<Location> cachedLocations = new HashSet<>(currentSessionCache.list());
+ Assert.assertEquals(2, cachedLocations.size());
+ Assert.assertTrue(cachedLocations.contains(currentSessionCache.append(Constants.Files.LAUNCHER_JAR)));
+ Assert.assertTrue(cachedLocations.contains(currentSessionCache.append(Constants.Files.TWILL_JAR)));
+
+ // Start another YarnTwillRunnerService
+ TwillRunnerService newTwillRunner = TWILL_TESTER.createTwillRunnerService();
+ newTwillRunner.start();
+
+ // Force a cleanup using the antique expiry. The list of locations that need to be cleanup was already
+ // collected when the new twill runner was started
+ ((YarnTwillRunnerService) newTwillRunner)
+ .forceLocationCacheCleanup(System.currentTimeMillis() + Configs.Defaults.LOCATION_CACHE_ANTIQUE_EXPIRY_MS);
+
+ // Now there shouldn't be any file under the current session cache directory
+ Assert.assertTrue(currentSessionCache.list().isEmpty());
+ }
+
+ /**
+ * A runnable that blocks until stopped explicitly.
+ */
+ public static final class BlockingTwillRunnable extends AbstractTwillRunnable {
+
+ private final CountDownLatch stopLatch = new CountDownLatch(1);
+
+ @Override
+ public void run() {
+ Uninterruptibles.awaitUninterruptibly(stopLatch);
+ }
+
+ @Override
+ public void stop() {
+ stopLatch.countDown();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/test/java/org/apache/twill/yarn/SocketServer.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/SocketServer.java b/twill-yarn/src/test/java/org/apache/twill/yarn/SocketServer.java
index e9e6a99..d59a15b 100644
--- a/twill-yarn/src/test/java/org/apache/twill/yarn/SocketServer.java
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/SocketServer.java
@@ -120,5 +120,5 @@ public abstract class SocketServer extends AbstractTwillRunnable {
}
}
- public abstract void handleRequest(BufferedReader reader, PrintWriter writer) throws IOException;
+ public abstract void handleRequest(BufferedReader reader, PrintWriter writer) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/test/java/org/apache/twill/yarn/TwillTester.java
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/java/org/apache/twill/yarn/TwillTester.java b/twill-yarn/src/test/java/org/apache/twill/yarn/TwillTester.java
index 4a2146e..9daf06c 100644
--- a/twill-yarn/src/test/java/org/apache/twill/yarn/TwillTester.java
+++ b/twill-yarn/src/test/java/org/apache/twill/yarn/TwillTester.java
@@ -19,9 +19,11 @@ package org.apache.twill.yarn;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
+import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Service;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
@@ -32,8 +34,12 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.server.MiniYARNCluster;
import org.apache.hadoop.yarn.server.resourcemanager.ClientRMService;
import org.apache.hadoop.yarn.util.Records;
+import org.apache.twill.api.Configs;
+import org.apache.twill.api.TwillController;
import org.apache.twill.api.TwillRunner;
import org.apache.twill.api.TwillRunnerService;
+import org.apache.twill.filesystem.FileContextLocationFactory;
+import org.apache.twill.filesystem.LocationFactory;
import org.apache.twill.internal.yarn.VersionDetectYarnAppClientFactory;
import org.apache.twill.internal.yarn.YarnAppClient;
import org.apache.twill.internal.yarn.YarnUtils;
@@ -45,7 +51,10 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
/**
* A TwillTester rule allows creation of mini Yarn cluster and {@link TwillRunner} used for testing that is
@@ -69,6 +78,8 @@ public class TwillTester extends ExternalResource {
private static final Logger LOG = LoggerFactory.getLogger(TwillTester.class);
private final TemporaryFolder tmpFolder = new TemporaryFolder();
+ private final Map<String, String> extraConfig;
+
private InMemoryZKServer zkServer;
private MiniDFSCluster dfsCluster;
private MiniYARNCluster cluster;
@@ -76,6 +87,25 @@ public class TwillTester extends ExternalResource {
private TwillRunnerService twillRunner;
private YarnAppClient yarnAppClient;
+ /**
+ * Creates a new instance with the give list of configurations.
+ *
+ * @param configs list of configuration pairs.
+ * The list must be in the form of {@code (key1, value1, key2, value2, ...)},
+ * hence the length of configs must be even.
+ * The {@link Object#toString()} method will be called to obtain the keys and values that go into
+ * the configuration.
+ */
+ public TwillTester(Object...configs) {
+ Preconditions.checkArgument(configs.length % 2 == 0,
+ "Arguments must be in pair form like (k1, v1, k2, v2): %s", Arrays.toString(configs));
+
+ this.extraConfig = new HashMap<>();
+ for (int i = 0; i < configs.length; i += 2) {
+ this.extraConfig.put(configs[i].toString(), configs[i + 1].toString());
+ }
+ }
+
@Override
protected void before() throws Throwable {
tmpFolder.create();
@@ -89,6 +119,11 @@ public class TwillTester extends ExternalResource {
LOG.info("Starting Mini DFS on path {}", miniDFSDir);
Configuration fsConf = new HdfsConfiguration(new Configuration());
fsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, miniDFSDir.getAbsolutePath());
+
+ for (Map.Entry<String, String> entry : extraConfig.entrySet()) {
+ fsConf.set(entry.getKey(), entry.getValue());
+ }
+
dfsCluster = new MiniDFSCluster.Builder(fsConf).numDataNodes(1).build();
Configuration conf = new YarnConfiguration(dfsCluster.getFileSystem().getConf());
@@ -108,11 +143,13 @@ public class TwillTester extends ExternalResource {
conf.set("yarn.scheduler.minimum-allocation-mb", "128");
conf.set("yarn.nodemanager.delete.debug-delay-sec", "3600");
+ conf.set(Configs.Keys.LOCAL_STAGING_DIRECTORY, tmpFolder.newFolder().getAbsolutePath());
+
cluster = new MiniYARNCluster("test-cluster", 3, 1, 1);
cluster.init(conf);
cluster.start();
- this.config = new YarnConfiguration(cluster.getConfig());
+ config = new YarnConfiguration(cluster.getConfig());
twillRunner = createTwillRunnerService();
twillRunner.start();
@@ -122,6 +159,17 @@ public class TwillTester extends ExternalResource {
@Override
protected void after() {
+ // Stop all runnable applications
+ for (TwillRunner.LiveInfo info : twillRunner.lookupLive()) {
+ for (TwillController controller : info.getControllers()) {
+ try {
+ controller.terminate().get();
+ } catch (Exception e) {
+ LOG.warn("Exception raised when awaiting termination of {}", info.getApplicationName());
+ }
+ }
+ }
+
try {
twillRunner.stop();
} catch (Exception e) {
@@ -146,13 +194,26 @@ public class TwillTester extends ExternalResource {
* Creates an unstarted instance of {@link org.apache.twill.api.TwillRunnerService}.
*/
public TwillRunnerService createTwillRunnerService() throws IOException {
- YarnTwillRunnerService runner = new YarnTwillRunnerService(config, zkServer.getConnectionStr() + "/twill");
+ YarnTwillRunnerService runner = new YarnTwillRunnerService(config, zkServer.getConnectionStr() + "/twill",
+ createLocationFactory());
// disable tests stealing focus
runner.setJVMOptions("-Djava.awt.headless=true");
return runner;
}
/**
+ * Creates a {@link LocationFactory} for the mini YARN cluster.
+ */
+ public LocationFactory createLocationFactory() {
+ try {
+ FileContext fc = FileContext.getFileContext(config);
+ return new FileContextLocationFactory(config, fc, fc.getHomeDirectory().toUri().getPath());
+ } catch (IOException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+
+ /**
* Returns a {@link TwillRunner} that interact with the mini Yarn cluster.
*/
public TwillRunner getTwillRunner() {
http://git-wip-us.apache.org/repos/asf/twill/blob/5986553b/twill-yarn/src/test/resources/header.txt
----------------------------------------------------------------------
diff --git a/twill-yarn/src/test/resources/header.txt b/twill-yarn/src/test/resources/header.txt
deleted file mode 100644
index b6e25e6..0000000
--- a/twill-yarn/src/test/resources/header.txt
+++ /dev/null
@@ -1 +0,0 @@
-Local file header