You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oozie.apache.org by ry...@apache.org on 2013/11/15 21:12:06 UTC
svn commit: r1542383 - in /oozie/trunk: ./
core/src/main/java/org/apache/oozie/action/hadoop/
core/src/main/java/org/apache/oozie/service/ core/src/main/resources/
core/src/test/java/org/apache/oozie/action/hadoop/
core/src/test/java/org/apache/oozie/s...
Author: ryota
Date: Fri Nov 15 20:12:05 2013
New Revision: 1542383
URL: http://svn.apache.org/r1542383
Log:
OOZIE-1584 Setup sharelib using script and pickup latest(honor ship.launcher) and remove DFS dependency at startup (puru via ryota)
Modified:
oozie/trunk/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java
oozie/trunk/core/src/main/java/org/apache/oozie/service/ShareLibService.java
oozie/trunk/core/src/main/resources/oozie-default.xml
oozie/trunk/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java
oozie/trunk/core/src/test/java/org/apache/oozie/service/TestShareLibService.java
oozie/trunk/docs/src/site/twiki/AG_Install.twiki
oozie/trunk/docs/src/site/twiki/DG_QuickStart.twiki
oozie/trunk/release-log.txt
oozie/trunk/tools/src/main/java/org/apache/oozie/tools/OozieSharelibCLI.java
oozie/trunk/tools/src/test/java/org/apache/oozie/tools/TestOozieSharelibCLI.java
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/action/hadoop/JavaActionExecutor.java Fri Nov 15 20:12:05 2013
@@ -17,8 +17,11 @@
*/
package org.apache.oozie.action.hadoop;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
+import java.io.PrintStream;
import java.io.StringReader;
import java.net.ConnectException;
import java.net.URI;
@@ -62,6 +65,7 @@ import org.apache.oozie.service.URIHandl
import org.apache.oozie.service.WorkflowAppService;
import org.apache.oozie.servlet.CallbackServlet;
import org.apache.oozie.util.ELEvaluator;
+import org.apache.oozie.util.IOUtils;
import org.apache.oozie.util.PropertiesUtils;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XLog;
@@ -479,45 +483,41 @@ public class JavaActionExecutor extends
protected void addShareLib(Path appPath, Configuration conf, String[] actionShareLibNames)
throws ActionExecutorException {
if (actionShareLibNames != null) {
- for (String actionShareLibName : actionShareLibNames) {
+ String user = conf.get("user.name");
+ FileSystem fs;
+ try {
+
Path systemLibPath = Services.get().get(WorkflowAppService.class).getSystemLibPath();
- if (systemLibPath != null) {
- Path actionLibPath = new Path(systemLibPath, actionShareLibName.trim());
- String user = conf.get("user.name");
- FileSystem fs;
- // If the actionLibPath has a valid scheme and authority,
- // then use them to determine the filesystem that the sharelib resides on;
- // otherwise, assume it resides on the same filesystem as the appPath and use
- // the appPath to determine the filesystem
- try {
- if (actionLibPath.toUri().getScheme() != null && actionLibPath.toUri().getAuthority() != null) {
- fs = Services.get().get(HadoopAccessorService.class)
- .createFileSystem(user, actionLibPath.toUri(), conf);
- }
- else {
- fs = Services.get().get(HadoopAccessorService.class)
- .createFileSystem(user, appPath.toUri(), conf);
- }
- if (fs.exists(actionLibPath)) {
- FileStatus[] files = fs.listStatus(actionLibPath);
- for (FileStatus file : files) {
- if (!file.isDir()) {
- addToCache(conf, actionLibPath, file.getPath().toUri().getPath(), false);
- }
+ if (systemLibPath.toUri().getScheme() != null && systemLibPath.toUri().getAuthority() != null) {
+ fs = Services.get().get(HadoopAccessorService.class)
+ .createFileSystem(user, systemLibPath.toUri(), conf);
+ }
+ else {
+ fs = Services.get().get(HadoopAccessorService.class).createFileSystem(user, appPath.toUri(), conf);
+ }
+ for (String actionShareLibName : actionShareLibNames) {
+
+ if (systemLibPath != null) {
+ ShareLibService shareLibService = Services.get().get(ShareLibService.class);
+ List<Path> listOfPaths = shareLibService.getShareLibJars(actionShareLibName);
+ if (listOfPaths != null && !listOfPaths.isEmpty()) {
+
+ for (Path actionLibPath : listOfPaths) {
+ DistributedCache.addFileToClassPath(actionLibPath, conf, fs);
+ DistributedCache.createSymlink(conf);
}
}
}
- catch (HadoopAccessorException ex) {
- throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, ex.getErrorCode()
- .toString(), ex.getMessage());
- }
- catch (IOException ex) {
- throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "It should never happen",
- ex.getMessage());
- }
-
}
}
+ catch (HadoopAccessorException ex) {
+ throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, ex.getErrorCode()
+ .toString(), ex.getMessage());
+ }
+ catch (IOException ex) {
+ throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "It should never happen",
+ ex.getMessage());
+ }
}
}
@@ -526,14 +526,13 @@ public class JavaActionExecutor extends
// ShareLibService is null for test cases
if (shareLibService != null) {
try {
- List<Path> listOfPaths = shareLibService
- .getActionSystemLibCommonJars(JavaActionExecutor.OOZIE_COMMON_LIBDIR);
+ List<Path> listOfPaths = shareLibService.getSystemLibJars(JavaActionExecutor.OOZIE_COMMON_LIBDIR);
FileSystem fs = listOfPaths.get(0).getFileSystem(conf);
for (Path actionLibPath : listOfPaths) {
DistributedCache.addFileToClassPath(actionLibPath, conf, fs);
DistributedCache.createSymlink(conf);
}
- listOfPaths = shareLibService.getActionSystemLibCommonJars(getType());
+ listOfPaths = shareLibService.getSystemLibJars(getType());
if (listOfPaths != null) {
for (Path actionLibPath : listOfPaths) {
DistributedCache.addFileToClassPath(actionLibPath, conf, fs);
Modified: oozie/trunk/core/src/main/java/org/apache/oozie/service/ShareLibService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/java/org/apache/oozie/service/ShareLibService.java?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/core/src/main/java/org/apache/oozie/service/ShareLibService.java (original)
+++ oozie/trunk/core/src/main/java/org/apache/oozie/service/ShareLibService.java Fri Nov 15 20:12:05 2013
@@ -17,20 +17,6 @@
*/
package org.apache.oozie.service;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathFilter;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.oozie.action.ActionExecutor;
-import org.apache.oozie.action.hadoop.JavaActionExecutor;
-import org.apache.oozie.service.Service;
-import org.apache.oozie.service.ServiceException;
-import org.apache.oozie.service.Services;
-import org.apache.oozie.util.XLog;
-import org.apache.oozie.ErrorCode;
-import com.google.common.annotations.VisibleForTesting;
import java.io.File;
import java.io.IOException;
import java.net.URI;
@@ -46,58 +32,110 @@ import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TimeZone;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.oozie.action.ActionExecutor;
+import org.apache.oozie.action.hadoop.JavaActionExecutor;
+import org.apache.oozie.util.XConfiguration;
+import org.apache.oozie.util.XLog;
+
+import com.google.common.annotations.VisibleForTesting;
+
public class ShareLibService implements Service {
- public static final String LAUNCHERJAR_LIB_RETENTION = CONF_PREFIX
- + "ShareLibService.temp.sharelib.retention.days";
+ public static final String LAUNCHERJAR_LIB_RETENTION = CONF_PREFIX + "ShareLibService.temp.sharelib.retention.days";
+
+ public static final String SHARELIB_MAPPING_FILE = CONF_PREFIX + "ShareLibService.mapping.file";
+
+ public static final String SHIP_LAUNCHER_JAR = "oozie.action.ship.launcher.jar";
+
private static final String PERMISSION_STRING = "-rwxr-xr-x";
+
+ public static final String LAUNCHER_PREFIX = "launcher_";
+
+ public static final String SHARED_LIB_PREFIX = "lib_";
+
+ public static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
+
private Services services;
- private Map<String, List<Path>> shareLibLauncherMap = new HashMap<String, List<Path>>();
+
+ private Map<String, List<Path>> shareLibMap = new HashMap<String, List<Path>>();
+
+ private Map<String, List<Path>> launcherLibMap = new HashMap<String, List<Path>>();
+
private static XLog LOG = XLog.getLog(ShareLibService.class);
+ private String sharelibMappingFile;
+
+ private boolean isShipLauncherEnabled = false;
+
+ public static String SHARE_LIB_CONF_PREFIX = "oozie";
+
+ private boolean shareLibLoadAttempted = false;
+
+ FileSystem fs;
+
+ @Override
public void init(Services services) throws ServiceException {
this.services = services;
- String formattedDate = new SimpleDateFormat("yyyyMMddHHmmss").format(Calendar.getInstance(
- TimeZone.getTimeZone("GMT")).getTime());
- Path tmpShareLibPath = new Path(services.get(WorkflowAppService.class).getSystemLibPath(), "tmp-"
- + formattedDate);
- HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
- URI uri = tmpShareLibPath.toUri();
try {
- FileSystem fs = FileSystem.get(has.createJobConf(uri.getAuthority()));
- copyLauncherJarsToShareLib(fs, tmpShareLibPath);
- recursiveChangePermissions(fs, tmpShareLibPath, FsPermission.valueOf(PERMISSION_STRING));
- purgeShareLibs(fs, tmpShareLibPath);
+ sharelibMappingFile = services.getConf().get(SHARELIB_MAPPING_FILE, "");
+ isShipLauncherEnabled = services.getConf().getBoolean(SHIP_LAUNCHER_JAR, false);
+ Path launcherlibPath = getLauncherlibPath();
+ HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
+ URI uri = launcherlibPath.toUri();
+ fs = FileSystem.get(has.createJobConf(uri.getAuthority()));
+ updateLauncherLib();
+ updateShareLib();
+ purgeLibs(fs, LAUNCHER_PREFIX);
+ purgeLibs(fs, SHARED_LIB_PREFIX);
}
catch (Exception e) {
- throw new ServiceException(ErrorCode.E0100, getClass().getName(), e.getMessage(), e);
+ LOG.error("Not able to cache shareLib. Admin need to issue oozlie cli command to update sharelib.", e);
}
+
}
- private void recursiveChangePermissions(FileSystem fs, Path path, FsPermission fsPerm) throws IOException {
- fs.setPermission(path, fsPerm);
- FileStatus[] filesStatus = fs.listStatus(path);
- for (int i = 0; i < filesStatus.length; i++) {
- Path p = filesStatus[i].getPath();
- if (filesStatus[i].isDir()) {
- recursiveChangePermissions(fs, p, fsPerm);
- }
- else {
- fs.setPermission(p, fsPerm);
- }
+ /**
+ * Recursively change permissions.
+ *
+ * @throws IOException Signals that an I/O exception has occurred.
+ * @throws ClassNotFoundException the class not found exception
+ */
+
+ private void updateLauncherLib() throws IOException {
+ if (isShipLauncherEnabled) {
+ Path launcherlibPath = getLauncherlibPath();
+ setupLauncherLibPath(fs, launcherlibPath);
+ recursiveChangePermissions(fs, launcherlibPath, FsPermission.valueOf(PERMISSION_STRING));
}
+
}
- private void copyLauncherJarsToShareLib(FileSystem fs, Path tmpShareLibPath) throws IOException,
- ClassNotFoundException {
+ /**
+ * Copy launcher jars to Temp directory
+ *
+ * @param fs the FileSystem
+ * @param tmpShareLibPath destination path
+ * @throws IOException Signals that an I/O exception has occurred.
+ * @throws ClassNotFoundException the class not found exception
+ */
+ private void setupLauncherLibPath(FileSystem fs, Path tmpLauncherLibPath) throws IOException {
+
ActionService actionService = Services.get().get(ActionService.class);
List<Class> classes = JavaActionExecutor.getCommonLauncherClasses();
- Path baseDir = new Path(tmpShareLibPath, JavaActionExecutor.OOZIE_COMMON_LIBDIR);
+ Path baseDir = new Path(tmpLauncherLibPath, JavaActionExecutor.OOZIE_COMMON_LIBDIR);
copyJarContainingClasses(classes, fs, baseDir, JavaActionExecutor.OOZIE_COMMON_LIBDIR);
Set<String> actionTypes = actionService.getActionTypes();
for (String key : actionTypes) {
@@ -107,13 +145,44 @@ public class ShareLibService implements
classes = jexecutor.getLauncherClasses();
if (classes != null) {
String type = executor.getType();
- Path executorDir = new Path(baseDir, type);
+ Path executorDir = new Path(tmpLauncherLibPath, type);
copyJarContainingClasses(classes, fs, executorDir, type);
}
}
}
}
+ /**
+ * Recursive change permissions.
+ *
+ * @param fs the FileSystem
+ * @param path the Path
+ * @param perm is permission
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ private void recursiveChangePermissions(FileSystem fs, Path path, FsPermission fsPerm) throws IOException {
+ fs.setPermission(path, fsPerm);
+ FileStatus[] filesStatus = fs.listStatus(path);
+ for (int i = 0; i < filesStatus.length; i++) {
+ Path p = filesStatus[i].getPath();
+ if (filesStatus[i].isDir()) {
+ recursiveChangePermissions(fs, p, fsPerm);
+ }
+ else {
+ fs.setPermission(p, fsPerm);
+ }
+ }
+ }
+
+ /**
+ * Copy jar containing classes.
+ *
+ * @param classes the classes
+ * @param fs the FileSystem
+ * @param executorDir is Path
+ * @param type is actionKey
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
private void copyJarContainingClasses(List<Class> classes, FileSystem fs, Path executorDir, String type)
throws IOException {
fs.mkdirs(executorDir);
@@ -135,13 +204,98 @@ public class ShareLibService implements
listOfPaths.add(path);
LOG.info(localJar.getName() + " uploaded to " + executorDir.toString());
}
- shareLibLauncherMap.put(type, listOfPaths);
+ launcherLibMap.put(type, listOfPaths);
+
}
- public List<Path> getActionSystemLibCommonJars(String type) {
- return shareLibLauncherMap.get(type);
+ /**
+ * Gets the path recursively.
+ *
+ * @param fs the FileSystem
+ * @param rootDir the root directory
+ * @param listOfPaths the list of paths
+ * @return the path recursively
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ private void getPathRecursively(FileSystem fs, Path rootDir, List<Path> listOfPaths) throws IOException {
+ if (rootDir == null) {
+ return;
+ }
+
+ FileStatus[] status = fs.listStatus(rootDir);
+ if (status == null) {
+ LOG.info("Shared lib " + rootDir + " doesn't exist, not adding to cache");
+ return;
+ }
+
+ for (FileStatus file : status) {
+ if (file.isDir()) {
+ getPathRecursively(fs, file.getPath(), listOfPaths);
+ }
+ else {
+ listOfPaths.add(file.getPath());
+ }
+ }
+ }
+
+ /**
+ * Gets the action system lib common jars.
+ *
+ * @param actionKey the action key
+ * @return List of paths
+ * @throws IOException
+ */
+ public List<Path> getShareLibJars(String actionKey) throws IOException {
+ // Sharelib map is empty means that on previous or startup attempt of
+ // caching sharelib has failed.Trying to reload
+ if (shareLibMap.isEmpty() && !shareLibLoadAttempted) {
+ synchronized (ShareLibService.class) {
+ if (shareLibMap.isEmpty()) {
+ updateShareLib();
+ shareLibLoadAttempted = true;
+ }
+ }
+ }
+ return shareLibMap.get(actionKey);
}
+ /**
+ * Gets the launcher jars.
+ *
+ * @param actionKey the action key
+ * @return launcher jars paths
+ * @throws ClassNotFoundException
+ * @throws IOException
+ */
+ public List<Path> getSystemLibJars(String actionKey) throws IOException {
+ List<Path> returnList = new ArrayList<Path>();
+ // Sharelib map is empty means that on previous or startup attempt of
+ // caching launcher jars has failed.Trying to reload
+ if (isShipLauncherEnabled) {
+ if (launcherLibMap.isEmpty()) {
+ synchronized (ShareLibService.class) {
+ if (launcherLibMap.isEmpty()) {
+ updateLauncherLib();
+ }
+ }
+ }
+ returnList = launcherLibMap.get(actionKey);
+ }
+ if (actionKey.equals(JavaActionExecutor.OOZIE_COMMON_LIBDIR)) {
+ List<Path> sharelibList = getShareLibJars(actionKey);
+ if (sharelibList != null) {
+ returnList.addAll(sharelibList);
+ }
+ }
+ return returnList;
+ }
+
+ /**
+ * Find containing jar containing.
+ *
+ * @param clazz the clazz
+ * @return the string
+ */
@VisibleForTesting
protected String findContainingJar(Class clazz) {
ClassLoader loader = clazz.getClassLoader();
@@ -178,12 +332,22 @@ public class ShareLibService implements
return null;
}
- private void purgeShareLibs(FileSystem fs, Path launcherJarLibPath) throws IOException, ParseException {
+ /**
+ * Purge libs.
+ *
+ * @param fs the fs
+ * @param prefix the prefix
+ * @throws IOException Signals that an I/O exception has occurred.
+ * @throws ParseException the parse exception
+ */
+ private void purgeLibs(FileSystem fs, final String prefix) throws IOException, ParseException {
Configuration conf = services.getConf();
Path executorLibBasePath = services.get(WorkflowAppService.class).getSystemLibPath();
+
PathFilter directoryFilter = new PathFilter() {
+ @Override
public boolean accept(Path path) {
- return path.getName().startsWith("tmp-");
+ return path.getName().startsWith(prefix);
}
};
FileStatus[] dirList = fs.listStatus(executorLibBasePath, directoryFilter);
@@ -200,20 +364,167 @@ public class ShareLibService implements
long retentionTime = 1000 * 60 * 60 * 24 * conf.getInt(LAUNCHERJAR_LIB_RETENTION, 7);
for (int i = 2; i < dirList.length; i++) {
Path dirPath = dirList[i].getPath();
- Date ts = new SimpleDateFormat("yyyyMMddHHmmss").parse(dirPath.getName().split("tmp-", -1)[1]);
- if ((current.getTime() - ts.getTime()) > retentionTime) {
+ String name = dirPath.getName().toString();
+ String time = name.substring(prefix.length());
+ Date d = dateFormat.parse(time);
+ if ((current.getTime() - d.getTime()) > retentionTime) {
fs.delete(dirPath, true);
LOG.info("Deleted old launcher jar lib directory {0}", dirPath.getName());
}
}
}
+ @Override
public void destroy() {
- shareLibLauncherMap.clear();
+ shareLibMap.clear();
+ launcherLibMap.clear();
+
}
+ @Override
public Class<? extends Service> getInterface() {
return ShareLibService.class;
}
+ /**
+ * Update share lib cache.
+ *
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ public void updateShareLib() throws IOException {
+ Map<String, List<Path>> tempShareLibMap = new HashMap<String, List<Path>>();
+
+ if (!StringUtils.isEmpty(sharelibMappingFile)) {
+ loadShareLibMetaFile(tempShareLibMap, sharelibMappingFile);
+ }
+ else {
+ loadShareLibfromDFS(tempShareLibMap);
+ }
+ shareLibMap = tempShareLibMap;
+
+ }
+
+ /**
+ * Update share lib cache. Parse the share lib directory and each sub
+ * directory is a action key
+ *
+ * @param shareLibMap the share lib jar map
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ private void loadShareLibfromDFS(Map<String, List<Path>> shareLibMap) throws IOException {
+ Path shareLibpath = getLatestLibPath(services.get(WorkflowAppService.class).getSystemLibPath(),
+ SHARED_LIB_PREFIX);
+
+ if (shareLibpath == null) {
+ LOG.info("No share lib directory found");
+ return;
+
+ }
+
+ FileStatus[] dirList = fs.listStatus(shareLibpath);
+
+ if (dirList == null) {
+ return;
+ }
+
+ for (FileStatus dir : dirList) {
+ List<Path> listOfPaths = new ArrayList<Path>();
+ getPathRecursively(fs, dir.getPath(), listOfPaths);
+ shareLibMap.put(dir.getPath().getName(), listOfPaths);
+ LOG.info("Share lib for " + dir.getPath().getName() + ":" + listOfPaths);
+
+ }
+
+ }
+
+ /**
+ * Load share lib text file. Sharelib mapping files contains list of
+ * key=value. where key is the action key and value is the DFS location of
+ * sharelib files.
+ *
+ * @param shareLibMap the share lib jar map
+ * @param sharelipFileMapping the sharelip file mapping
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ @SuppressWarnings("unchecked")
+ private void loadShareLibMetaFile(Map<String, List<Path>> shareLibMap, String sharelibFileMapping)
+ throws IOException {
+
+
+ Path shareFileMappingPath= new Path(sharelibFileMapping);
+ HadoopAccessorService has = Services.get().get(HadoopAccessorService.class);
+ FileSystem filesystem = FileSystem.get(has.createJobConf(shareFileMappingPath.toUri().getAuthority()));
+ XConfiguration conf = new XConfiguration(filesystem.open(shareFileMappingPath));
+ Iterator<Map.Entry<String, String>> it = conf.iterator();
+
+
+ while (it.hasNext()) {
+ Map.Entry<String, String> en = it.next();
+ if (en.getKey().toLowerCase().startsWith(SHARE_LIB_CONF_PREFIX)) {
+ String key = en.getKey().substring(SHARE_LIB_CONF_PREFIX.length() + 1);
+ String pathList[] = en.getValue().split(",");
+ List<Path> listOfPaths = new ArrayList<Path>();
+ for (String dfsPath : pathList) {
+ getPathRecursively(fs, new Path(dfsPath), listOfPaths);
+ }
+ shareLibMap.put(key, listOfPaths);
+ LOG.info("Share lib for " + en.getKey() + ":" + listOfPaths);
+
+
+ }
+ else {
+ LOG.info(" Not adding " + en.getKey() + " to sharelib, not prefix with " + SHARE_LIB_CONF_PREFIX);
+ }
+
+ }
+ }
+
+ /**
+ * Gets the launcherlib path.
+ *
+ * @return the launcherlib path
+ */
+ private Path getLauncherlibPath() {
+ String formattedDate = dateFormat.format(Calendar.getInstance(TimeZone.getTimeZone("GMT")).getTime());
+ Path tmpLauncherLibPath = new Path(services.get(WorkflowAppService.class).getSystemLibPath(), LAUNCHER_PREFIX
+ + formattedDate);
+ return tmpLauncherLibPath;
+ }
+
+ /**
+ * Gets the Latest lib path.
+ *
+ * @param rootDir the root dir
+ * @param prefix the prefix
+ * @return latest lib path
+ * @throws IOException Signals that an I/O exception has occurred.
+ */
+ public Path getLatestLibPath(Path rootDir, final String prefix) throws IOException {
+ Date max = new Date(0L);
+ Path path = null;
+ PathFilter directoryFilter = new PathFilter() {
+ @Override
+ public boolean accept(Path path) {
+ return path.getName().startsWith(prefix);
+ }
+ };
+
+ FileStatus[] files = fs.listStatus(rootDir, directoryFilter);
+ for (FileStatus file : files) {
+ String name = file.getPath().getName().toString();
+ String time = name.substring(prefix.length());
+ Date d = null;
+ try {
+ d = dateFormat.parse(time);
+ }
+ catch (ParseException e) {
+ continue;
+ }
+ if (d.compareTo(max) > 0) {
+ path = file.getPath();
+ max = d;
+ }
+ }
+ return path;
+ }
}
Modified: oozie/trunk/core/src/main/resources/oozie-default.xml
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/main/resources/oozie-default.xml?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/core/src/main/resources/oozie-default.xml (original)
+++ oozie/trunk/core/src/main/resources/oozie-default.xml Fri Nov 15 20:12:05 2013
@@ -1958,4 +1958,23 @@
</description>
</property>
+
+
+
+ <!-- Sharelib Configuration -->
+ <property>
+ <name>oozie.service.ShareLibService.mapping.file</name>
+ <value></value>
+ <description>
+ Sharelib mapping files contains list of key=value,
+ where key will be the sharelib name for the action and value is a comma separated list of
+ DFS directories or jar files.
+ Example.
+ oozie.pig_10=hdfs:///share/lib/pig/pig-0.10.1/lib/
+ oozie.pig=hdfs:///share/lib/pig/pig-0.11.1/lib/
+ oozie.distcp=hdfs:///share/lib/hadoop-2.2.0/share/hadoop/tools/lib/hadoop-distcp-2.2.0.jar
+ </description>
+ </property>
+
+
</configuration>
Modified: oozie/trunk/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java (original)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/action/hadoop/TestJavaActionExecutor.java Fri Nov 15 20:12:05 2013
@@ -24,6 +24,7 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.io.StringReader;
import java.net.URI;
+import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
@@ -52,6 +53,7 @@ import org.apache.oozie.client.WorkflowJ
import org.apache.oozie.service.HadoopAccessorService;
import org.apache.oozie.service.LiteWorkflowStoreService;
import org.apache.oozie.service.Services;
+import org.apache.oozie.service.ShareLibService;
import org.apache.oozie.service.UUIDService;
import org.apache.oozie.service.WorkflowAppService;
import org.apache.oozie.service.WorkflowStoreService;
@@ -74,6 +76,7 @@ public class TestJavaActionExecutor exte
@Override
protected void setSystemProps() throws Exception {
super.setSystemProps();
+
setSystemProperty("oozie.service.ActionService.executor.classes", JavaActionExecutor.class.getName());
setSystemProperty("oozie.service.HadoopAccessorService.action.configurations",
"*=hadoop-conf," + getJobTrackerUri() + "=action-conf");
@@ -82,6 +85,7 @@ public class TestJavaActionExecutor exte
InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream("test-action-config.xml");
OutputStream os = new FileOutputStream(new File(getTestCaseConfDir() + "/action-conf", "java.xml"));
IOUtils.copyStream(is, os);
+
}
public void testSetupMethods() throws Exception {
@@ -1137,7 +1141,12 @@ public class TestJavaActionExecutor exte
@Test
public void testAddActionShareLib() throws Exception {
- Path systemLibPath = new Path(getFsTestCaseDir(), "systemlib");
+
+ WorkflowAppService wps = Services.get().get(WorkflowAppService.class);
+
+ Path systemLibPath = new Path(wps.getSystemLibPath(), ShareLibService.SHARED_LIB_PREFIX
+ + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date()).toString());
+
Path javaShareLibPath = new Path(systemLibPath, "java");
getFileSystem().mkdirs(javaShareLibPath);
Path jar1Path = new Path(javaShareLibPath, "jar1.jar");
@@ -1157,6 +1166,12 @@ public class TestJavaActionExecutor exte
Path jar5Path = new Path(otherShareLibPath, "jar5.jar");
getFileSystem().create(jar5Path).close();
+ Path launcherPath = new Path(systemLibPath, "oozie");
+ getFileSystem().mkdirs(launcherPath);
+ Path jar6Path = new Path(launcherPath, "jar5.jar");
+ getFileSystem().create(jar6Path).close();
+
+
String actionXml = "<java>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" +
"<name-node>" + getNameNodeUri() + "</name-node>" +
"<job-xml>job.xml</job-xml>" + "<job-xml>job2.xml</job-xml>" +
@@ -1165,6 +1180,8 @@ public class TestJavaActionExecutor exte
Element eActionXml = XmlUtils.parseXml(actionXml);
Context context = createContext(actionXml, null);
+ Services.get().setService(ShareLibService.class);
+
// Test oozie server action sharelib setting
WorkflowJobBean workflow = (WorkflowJobBean) context.getWorkflow();
XConfiguration wfConf = new XConfiguration();
@@ -1231,6 +1248,8 @@ public class TestJavaActionExecutor exte
Services.get().destroy();
setSystemProperty(WorkflowAppService.SYSTEM_LIB_PATH, "/user/" + getOozieUser() + "/share/");
new Services().init();
+ Services.get().setService(ShareLibService.class);
+
Path appPath = getAppPath();
JobConf conf = ae.createBaseHadoopConf(context, eActionXml);
// The next line should not throw an Exception because it will get the scheme and authority from the appPath, and not the
@@ -1253,6 +1272,7 @@ public class TestJavaActionExecutor exte
Services.get().destroy();
setSystemProperty(WorkflowAppService.SYSTEM_LIB_PATH, getNameNodeUri() + "/user/" + getOozieUser() + "/share/");
new Services().init();
+ Services.get().setService(ShareLibService.class);
appPath = new Path("foo://bar:1234/blah");
conf = ae.createBaseHadoopConf(context, eActionXml);
// The next line should not throw an Exception because it will get the scheme and authority from the sharelib path (and not
Modified: oozie/trunk/core/src/test/java/org/apache/oozie/service/TestShareLibService.java
URL: http://svn.apache.org/viewvc/oozie/trunk/core/src/test/java/org/apache/oozie/service/TestShareLibService.java?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/core/src/test/java/org/apache/oozie/service/TestShareLibService.java (original)
+++ oozie/trunk/core/src/test/java/org/apache/oozie/service/TestShareLibService.java Fri Nov 15 20:12:05 2013
@@ -27,16 +27,16 @@ import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
+import org.apache.oozie.action.hadoop.ActionExecutorTestCase.Context;
import org.apache.oozie.action.hadoop.JavaActionExecutor;
import org.apache.oozie.action.hadoop.PigActionExecutor;
-import org.apache.oozie.action.hadoop.ActionExecutorTestCase.Context;
import org.apache.oozie.action.hadoop.TestJavaActionExecutor;
-import org.apache.oozie.service.Services;
-import org.apache.oozie.service.ActionService;
+import org.apache.oozie.client.OozieClient;
import org.apache.oozie.test.XFsTestCase;
import org.apache.oozie.util.IOUtils;
import org.apache.oozie.util.XConfiguration;
@@ -48,6 +48,7 @@ public class TestShareLibService extends
Services services;
private static String testCaseDirPath;
+ String shareLibPath = "shareLibPath";
@Override
protected void setUp() throws Exception {
@@ -73,7 +74,7 @@ public class TestShareLibService extends
public static class DummyShareLibService extends ShareLibService {
@Override
public String findContainingJar(Class clazz) {
- if (JavaActionExecutor.getCommonLauncherClasses().contains(clazz) ){
+ if (JavaActionExecutor.getCommonLauncherClasses().contains(clazz)) {
return testCaseDirPath + "/" + MyOozie.class.getName() + ".jar";
}
return testCaseDirPath + "/" + clazz.getName() + ".jar";
@@ -100,62 +101,138 @@ public class TestShareLibService extends
public void testCreateLauncherLibPath() throws Exception {
services = new Services();
setSystemProps();
- try {
- services.init();
- ShareLibService shareLibService = Services.get().get(ShareLibService.class);
- List<Path> launcherPath = shareLibService.getActionSystemLibCommonJars(JavaActionExecutor.OOZIE_COMMON_LIBDIR);
- assertNotNull(launcherPath);
- assertTrue(getFileSystem().exists(launcherPath.get(0)));
- List<Path> pigLauncherPath = shareLibService.getActionSystemLibCommonJars("pig");
- assertTrue(getFileSystem().exists(pigLauncherPath.get(0)));
- }
- finally {
- services.destroy();
- }
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+ services.init();
+ ShareLibService shareLibService = Services.get().get(ShareLibService.class);
+ List<Path> launcherPath = shareLibService.getSystemLibJars(JavaActionExecutor.OOZIE_COMMON_LIBDIR);
+ assertNotNull(launcherPath);
+ assertTrue(getFileSystem().exists(launcherPath.get(0)));
+ List<Path> pigLauncherPath = shareLibService.getSystemLibJars("pig");
+ assertTrue(getFileSystem().exists(pigLauncherPath.get(0)));
+
+ services.destroy();
+
}
@Test
public void testAddShareLibDistributedCache() throws Exception {
services = new Services();
setSystemProps();
- try {
- services.init();
- String actionXml = "<java>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" +
- "<name-node>" + getNameNodeUri() + "</name-node>" +
- "</java>";
- Element eActionXml = XmlUtils.parseXml(actionXml);
- XConfiguration protoConf = new XConfiguration();
- protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser());
- WorkflowJobBean wfj = new WorkflowJobBean();
- wfj.setProtoActionConf(XmlUtils.prettyPrint(protoConf).toString());
- wfj.setConf(XmlUtils.prettyPrint(new XConfiguration()).toString());
- Context context = new TestJavaActionExecutor().new Context(wfj, new WorkflowActionBean());
- PigActionExecutor ae = new PigActionExecutor();
- Configuration jobConf = ae.createBaseHadoopConf(context, eActionXml);
- ae.setLibFilesArchives(context, eActionXml, new Path("hdfs://dummyAppPath"), jobConf);
-
- URI[] cacheFiles = DistributedCache.getCacheFiles(jobConf);
- String cacheFilesStr = Arrays.toString(cacheFiles);
- assertTrue(cacheFilesStr.contains(MyPig.class.getName() + ".jar"));
- assertTrue(cacheFilesStr.contains(MyOozie.class.getName() + ".jar"));
- // Hadoop 2 has two extra jars
- if (cacheFiles.length == 4) {
- assertTrue(cacheFilesStr.contains("MRAppJar.jar"));
- assertTrue(cacheFilesStr.contains("hadoop-mapreduce-client-jobclient-"));
- }
- else {
- assertEquals(2, cacheFiles.length);
- }
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+
+ services.init();
+ String actionXml = "<java>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>"
+ + getNameNodeUri() + "</name-node>" + "</java>";
+ Element eActionXml = XmlUtils.parseXml(actionXml);
+ XConfiguration protoConf = new XConfiguration();
+ protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser());
+ WorkflowJobBean wfj = new WorkflowJobBean();
+ wfj.setProtoActionConf(XmlUtils.prettyPrint(protoConf).toString());
+ wfj.setConf(XmlUtils.prettyPrint(new XConfiguration()).toString());
+ Context context = new TestJavaActionExecutor().new Context(wfj, new WorkflowActionBean());
+ PigActionExecutor ae = new PigActionExecutor();
+ Configuration jobConf = ae.createBaseHadoopConf(context, eActionXml);
+ ae.setLibFilesArchives(context, eActionXml, new Path("hdfs://dummyAppPath"), jobConf);
+
+ URI[] cacheFiles = DistributedCache.getCacheFiles(jobConf);
+ String cacheFilesStr = Arrays.toString(cacheFiles);
+ assertTrue(cacheFilesStr.contains(MyPig.class.getName() + ".jar"));
+ assertTrue(cacheFilesStr.contains(MyOozie.class.getName() + ".jar"));
+ // Hadoop 2 has two extra jars
+ if (cacheFiles.length == 4) {
+ assertTrue(cacheFilesStr.contains("MRAppJar.jar"));
+ assertTrue(cacheFilesStr.contains("hadoop-mapreduce-client-jobclient-"));
}
- finally {
- services.destroy();
+ else {
+ assertEquals(2, cacheFiles.length);
}
+ services.destroy();
+ }
+
+ @Test
+ public void testAddShareLib_pig() throws Exception {
+ services = new Services();
+ setSystemProps();
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+ services.init();
+ String actionXml = "<pig>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>"
+ + getNameNodeUri() + "</name-node>" + "</pig>";
+ Element eActionXml = XmlUtils.parseXml(actionXml);
+ XConfiguration protoConf = new XConfiguration();
+ protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser());
+ WorkflowJobBean wfj = new WorkflowJobBean();
+ wfj.setProtoActionConf(XmlUtils.prettyPrint(protoConf).toString());
+ wfj.setConf(XmlUtils.prettyPrint(new XConfiguration()).toString());
+ Context context = new TestJavaActionExecutor().new Context(wfj, new WorkflowActionBean());
+ PigActionExecutor ae = new PigActionExecutor();
+ Configuration jobConf = ae.createBaseHadoopConf(context, eActionXml);
+ ae.setLibFilesArchives(context, eActionXml, new Path("hdfs://dummyAppPath"), jobConf);
+
+ URI[] cacheFiles = DistributedCache.getCacheFiles(jobConf);
+ assertEquals(2, cacheFiles.length);
+ assertTrue(cacheFiles[0].getPath().endsWith("MyPig.jar") || cacheFiles[1].getPath().endsWith("MyPig.jar"));
+ services.destroy();
+
+ }
+
+ @Test
+ public void testAddShareLib_pig_withVersion() throws Exception {
+ services = new Services();
+ setSystemProps();
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+
+ FileSystem fs = getFileSystem();
+ Date time = new Date(System.currentTimeMillis());
+
+ Path basePath = new Path(services.getConf().get(WorkflowAppService.SYSTEM_LIB_PATH));
+ Path libpath = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + ShareLibService.dateFormat.format(time));
+ fs.mkdirs(libpath);
+
+ Path pigPath = new Path(libpath.toString() + Path.SEPARATOR + "pig");
+ Path pigPath1 = new Path(libpath.toString() + Path.SEPARATOR + "pig_9");
+ Path pigPath2 = new Path(libpath.toString() + Path.SEPARATOR + "pig_10");
+ fs.mkdirs(pigPath);
+ fs.mkdirs(pigPath1);
+ fs.mkdirs(pigPath2);
+
+ createFile(libpath.toString() + Path.SEPARATOR + "pig_10" + Path.SEPARATOR + "pig-10.jar");
+
+ services.init();
+ String actionXml = "<pig>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>"
+ + getNameNodeUri() + "</name-node>"
+ + "<property><name>oozie.action.sharelib.for.pig</name><value>pig_10</value></property>" + "</pig>";
+ Element eActionXml = XmlUtils.parseXml(actionXml);
+ XConfiguration protoConf = new XConfiguration();
+ protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser());
+ WorkflowJobBean wfj = new WorkflowJobBean();
+ protoConf.setBoolean(OozieClient.USE_SYSTEM_LIBPATH, true);
+ wfj.setProtoActionConf(XmlUtils.prettyPrint(protoConf).toString());
+ wfj.setConf(XmlUtils.prettyPrint(protoConf).toString());
+
+ Context context = new TestJavaActionExecutor().new Context(wfj, new WorkflowActionBean());
+ PigActionExecutor ae = new PigActionExecutor();
+ Configuration jobConf = ae.createBaseHadoopConf(context, eActionXml);
+ jobConf.set("oozie.action.sharelib.for.pig", "pig_10");
+ ae.setLibFilesArchives(context, eActionXml, new Path("hdfs://dummyAppPath"), jobConf);
+
+ URI[] cacheFiles = DistributedCache.getCacheFiles(jobConf);
+ assertEquals(3, cacheFiles.length);
+ assertTrue(cacheFiles[0].getPath().endsWith("pig-10.jar") || cacheFiles[1].getPath().endsWith("pig-10.jar")
+ || cacheFiles[2].getPath().endsWith("pig-10.jar"));
+ services.destroy();
+
}
@Test
public void testPurgeShareLib() throws Exception {
services = new Services();
setSystemProps();
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
FileSystem fs = getFileSystem();
// for directory created 8 days back to be deleted
int expire1 = services.getConf().getInt(ShareLibService.LAUNCHERJAR_LIB_RETENTION, 7) + 1;
@@ -171,9 +248,9 @@ public class TestShareLibService extends
String noexpireTs = new SimpleDateFormat("yyyyMMddHHmmss").format(noexpireDate);
String noexpireTs1 = new SimpleDateFormat("yyyyMMddHHmmss").format(noexpireDate1);
Path basePath = new Path(services.getConf().get(WorkflowAppService.SYSTEM_LIB_PATH));
- Path expirePath = new Path(basePath, "tmp-" + expireTs);
- Path noexpirePath = new Path(basePath, "tmp-" + noexpireTs);
- Path noexpirePath1 = new Path(basePath, "tmp-" + noexpireTs1);
+ Path expirePath = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + expireTs);
+ Path noexpirePath = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + noexpireTs);
+ Path noexpirePath1 = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + noexpireTs1);
fs.mkdirs(expirePath);
fs.mkdirs(noexpirePath);
fs.mkdirs(noexpirePath1);
@@ -189,4 +266,244 @@ public class TestShareLibService extends
}
}
+ @Test
+ public void testPurgeLauncherJar() throws Exception {
+ services = new Services();
+ setSystemProps();
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+ FileSystem fs = getFileSystem();
+ // for directory created 8 days back to be deleted
+ int expire1 = services.getConf().getInt(ShareLibService.LAUNCHERJAR_LIB_RETENTION, 7) + 1;
+ // for directory created 6 days back NOT to be deleted
+ int noexpire = services.getConf().getInt(ShareLibService.LAUNCHERJAR_LIB_RETENTION, 7) - 1;
+ // for directory created 5 days back NOT to be deleted
+ int noexpire1 = services.getConf().getInt(ShareLibService.LAUNCHERJAR_LIB_RETENTION, 7) - 2;
+
+ Date expireDate = new Date(System.currentTimeMillis() - (1000 * 60 * 60 * 24 * expire1));
+ Date noexpireDate = new Date(System.currentTimeMillis() - (1000 * 60 * 60 * 24 * noexpire));
+ Date noexpireDate1 = new Date(System.currentTimeMillis() - (1000 * 60 * 60 * 24 * noexpire1));
+ String expireTs = new SimpleDateFormat("yyyyMMddHHmmss").format(expireDate);
+ String noexpireTs = new SimpleDateFormat("yyyyMMddHHmmss").format(noexpireDate);
+ String noexpireTs1 = new SimpleDateFormat("yyyyMMddHHmmss").format(noexpireDate1);
+ Path basePath = new Path(services.getConf().get(WorkflowAppService.SYSTEM_LIB_PATH));
+ Path expirePath = new Path(basePath, ShareLibService.LAUNCHER_PREFIX + expireTs);
+ Path noexpirePath = new Path(basePath, ShareLibService.LAUNCHER_PREFIX + noexpireTs);
+
+ Path noexpirePath1 = new Path(basePath, ShareLibService.LAUNCHER_PREFIX + noexpireTs1);
+
+ fs.mkdirs(expirePath);
+ fs.mkdirs(noexpirePath);
+ fs.mkdirs(noexpirePath1);
+ services.init();
+ assertEquals(3, fs.listStatus(basePath).length);
+ assertTrue(fs.exists(noexpirePath));
+ assertTrue(fs.exists(noexpirePath1));
+ assertTrue(!fs.exists(expirePath));
+ services.destroy();
+ }
+
+ @Test
+ public void testGetShareLibPath() throws Exception {
+ services = new Services();
+ setSystemProps();
+ FileSystem fs = getFileSystem();
+ Date time = new Date(System.currentTimeMillis());
+
+ Path basePath = new Path(services.getConf().get(WorkflowAppService.SYSTEM_LIB_PATH));
+ Path libpath = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + ShareLibService.dateFormat.format(time));
+ fs.mkdirs(libpath);
+
+ Path pigPath = new Path(libpath.toString() + Path.SEPARATOR + "pig");
+ Path pigPath1 = new Path(libpath.toString() + Path.SEPARATOR + "pig_9");
+ Path pigPath2 = new Path(libpath.toString() + Path.SEPARATOR + "pig_10");
+ fs.mkdirs(pigPath);
+ fs.mkdirs(pigPath1);
+ fs.mkdirs(pigPath2);
+ services.init();
+ ShareLibService shareLibService = Services.get().get(ShareLibService.class);
+ assertNotNull(shareLibService.getShareLibJars("pig"));
+ assertNotNull(shareLibService.getShareLibJars("pig_9"));
+ assertNotNull(shareLibService.getShareLibJars("pig_10"));
+ assertNull(shareLibService.getShareLibJars("pig_11"));
+ services.destroy();
+ }
+
+ @Test
+ public void testShareLib() throws Exception {
+ services = new Services();
+ setSystemProps();
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+ FileSystem fs = getFileSystem();
+
+ Date day1 = new Date(System.currentTimeMillis() - (1000 * 60 * 60 * 24 * 1));
+ Date day2 = new Date(System.currentTimeMillis() - (1000 * 60 * 60 * 24 * 2));
+ Date day3 = new Date(System.currentTimeMillis() - (1000 * 60 * 60 * 24 * 3));
+ String dir1 = new SimpleDateFormat("yyyyMMddHHmmss").format(day1);
+ String dir2 = new SimpleDateFormat("yyyyMMddHHmmss").format(day2);
+ String dir3 = new SimpleDateFormat("yyyyMMddHHmmss").format(day3);
+ Path basePath = new Path(services.getConf().get(WorkflowAppService.SYSTEM_LIB_PATH));
+ Path path1 = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + dir1);
+ Path path2 = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + dir2);
+
+ Path path3 = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + dir3);
+ fs.mkdirs(path1);
+ fs.mkdirs(path2);
+ fs.mkdirs(path3);
+ createFile(path1.toString() + Path.SEPARATOR + "pig" + Path.SEPARATOR + "pig.jar");
+ services.init();
+ ShareLibService shareLibService = Services.get().get(ShareLibService.class);
+ assertTrue(shareLibService.getShareLibJars("pig").get(0).getName().endsWith("pig.jar"));
+ services.destroy();
+ }
+
+ @Test
+ public void testShareLibLoadFile() throws Exception {
+
+ services = new Services();
+ FileSystem fs = getFileSystem();
+ setSystemProps();
+ creatTempshareLibKeymap(fs);
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHARELIB_MAPPING_FILE, fs.getUri() + "/user/test/config.properties");
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+ services.init();
+ ShareLibService shareLibService = Services.get().get(ShareLibService.class);
+ assertTrue(shareLibService.getShareLibJars("something_new").get(0).getName().endsWith("somethingNew.jar"));
+ assertTrue(shareLibService.getShareLibJars("pig").get(0).getName().endsWith("pig.jar"));
+ fs.delete(new Path("shareLibPath/"), true);
+ services.destroy();
+ }
+
+ @Test
+ public void testLoadfromDFS() throws Exception {
+ services = new Services();
+ setSystemProps();
+ services.init();
+ FileSystem fs = getFileSystem();
+ Date time = new Date(System.currentTimeMillis());
+
+ Path basePath = new Path(services.getConf().get(WorkflowAppService.SYSTEM_LIB_PATH));
+ Path libpath = new Path(basePath, ShareLibService.SHARED_LIB_PREFIX + ShareLibService.dateFormat.format(time));
+ fs.mkdirs(libpath);
+
+ Path pigPath = new Path(libpath.toString() + Path.SEPARATOR + "pig");
+ Path ooziePath = new Path(libpath.toString() + Path.SEPARATOR + "oozie");
+ Path pigPath1 = new Path(libpath.toString() + Path.SEPARATOR + "pig_9");
+ Path pigPath2 = new Path(libpath.toString() + Path.SEPARATOR + "pig_10");
+ fs.mkdirs(pigPath);
+ fs.mkdirs(ooziePath);
+ fs.mkdirs(pigPath1);
+ fs.mkdirs(pigPath2);
+
+ createFile(libpath.toString() + Path.SEPARATOR + "pig_10" + Path.SEPARATOR + "pig-10.jar");
+ createFile(libpath.toString() + Path.SEPARATOR + "oozie" + Path.SEPARATOR + "oozie_luncher.jar");
+
+ String actionXml = "<pig>" + "<job-tracker>" + getJobTrackerUri() + "</job-tracker>" + "<name-node>"
+ + getNameNodeUri() + "</name-node>"
+ + "<property><name>oozie.action.sharelib.for.pig</name><value>pig_10</value></property>" + "</pig>";
+ Element eActionXml = XmlUtils.parseXml(actionXml);
+ XConfiguration protoConf = new XConfiguration();
+ protoConf.set(WorkflowAppService.HADOOP_USER, getTestUser());
+ WorkflowJobBean wfj = new WorkflowJobBean();
+ protoConf.setBoolean(OozieClient.USE_SYSTEM_LIBPATH, true);
+ wfj.setProtoActionConf(XmlUtils.prettyPrint(protoConf).toString());
+ wfj.setConf(XmlUtils.prettyPrint(protoConf).toString());
+
+ Context context = new TestJavaActionExecutor().new Context(wfj, new WorkflowActionBean());
+ PigActionExecutor ae = new PigActionExecutor();
+ Configuration jobConf = ae.createBaseHadoopConf(context, eActionXml);
+ jobConf.set("oozie.action.sharelib.for.pig", "pig_10");
+ ae.setLibFilesArchives(context, eActionXml, new Path("hdfs://dummyAppPath"), jobConf);
+ URI[] cacheFiles = DistributedCache.getCacheFiles(jobConf);
+ assertEquals(2, cacheFiles.length);
+ assertTrue(cacheFiles[0].getPath().endsWith("pig-10.jar") || cacheFiles[1].getPath().endsWith("pig-10.jar"));
+ services.destroy();
+
+ }
+
+ @Test
+ public void testShareLibLoadFileMultipleFile() throws Exception {
+ FileSystem fs = getFileSystem();
+ services = new Services();
+ creatTempshareLibKeymap_multipleFile(fs);
+ setSystemProps();
+ Configuration conf = services.getConf();
+ conf.set(ShareLibService.SHARELIB_MAPPING_FILE, fs.getUri() + "/user/test/config.properties");
+ conf.set(ShareLibService.SHIP_LAUNCHER_JAR, "true");
+ services.init();
+ ShareLibService shareLibService = Services.get().get(ShareLibService.class);
+ assertNull(shareLibService.getShareLibJars("something_new"));
+ assertEquals(shareLibService.getShareLibJars("pig").size(), 2);
+ fs.delete(new Path("shareLibPath/"), true);
+ services.destroy();
+ }
+
+ public void createFile(String filename) throws IOException {
+ Path path = new Path(filename);
+ FSDataOutputStream out = getFileSystem().create(path);
+ out.close();
+ }
+
+ public void creatTempshareLibKeymap(FileSystem fs) {
+ XConfiguration prop = new XConfiguration();
+
+ try {
+
+ String testPath = "shareLibPath/";
+
+ Path basePath = new Path(testPath + Path.SEPARATOR + "testPath");
+
+ Path somethingNew = new Path(testPath + Path.SEPARATOR + "something_new");
+ fs.mkdirs(basePath);
+ fs.mkdirs(somethingNew);
+
+ createFile(basePath.toString() + Path.SEPARATOR + "pig" + Path.SEPARATOR + "pig.jar");
+ createFile(somethingNew.toString() + Path.SEPARATOR + "somethingNew" + Path.SEPARATOR + "somethingNew.jar");
+
+ prop.set(ShareLibService.SHARE_LIB_CONF_PREFIX + ".pig", "/user/test/" + basePath.toString());
+ prop.set(ShareLibService.SHARE_LIB_CONF_PREFIX + ".something_new", "/user/test/" + somethingNew.toString());
+
+ FSDataOutputStream out = fs.create(new Path("/user/test/config.properties"));
+
+ prop.writeXml(out);
+ out.close();
+
+ }
+ catch (IOException ex) {
+ ex.printStackTrace();
+ }
+ }
+
+ public void creatTempshareLibKeymap_multipleFile(FileSystem fs) {
+
+ XConfiguration prop = new XConfiguration();
+ try {
+
+ String testPath = "shareLibPath/";
+
+ Path basePath = new Path(testPath + Path.SEPARATOR + "testPath");
+ Path somethingNew = new Path(testPath + Path.SEPARATOR + "something_new");
+ fs.mkdirs(basePath);
+ fs.mkdirs(somethingNew);
+
+ createFile(basePath.toString() + Path.SEPARATOR + "pig" + Path.SEPARATOR + "pig.jar");
+ createFile(somethingNew.toString() + Path.SEPARATOR + "somethingNew" + Path.SEPARATOR + "somethingNew.jar");
+
+ prop.set(ShareLibService.SHARE_LIB_CONF_PREFIX + ".pig", "/user/test/" + basePath.toString()
+ + Path.SEPARATOR + "pig" + Path.SEPARATOR + "pig.jar," + "/user/test/" + somethingNew.toString()
+ + Path.SEPARATOR + "somethingNew" + Path.SEPARATOR + "somethingNew.jar");
+
+ FSDataOutputStream out = fs.create(new Path("/user/test/config.properties"));
+
+ prop.writeXml(out);
+ out.close();
+
+ }
+ catch (IOException ex) {
+ ex.printStackTrace();
+ }
+ }
+
}
Modified: oozie/trunk/docs/src/site/twiki/AG_Install.twiki
URL: http://svn.apache.org/viewvc/oozie/trunk/docs/src/site/twiki/AG_Install.twiki?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/docs/src/site/twiki/AG_Install.twiki (original)
+++ oozie/trunk/docs/src/site/twiki/AG_Install.twiki Fri Nov 15 20:12:05 2013
@@ -77,7 +77,8 @@ Usage : oozie-setup.sh <OPTIONS>"
installation directory will be used)"
(action failes if sharelib is already installed"
in HDFS)"
- sharelib upgrade -fs FS_URI [-locallib SHARED_LIBRARY] (upgrade existing sharelib, fails if there"
+ sharelib upgrade -fs FS_URI [-locallib SHARED_LIBRARY] ([deprecated][use create command to create new version]
+ upgrade existing sharelib, fails if there"
is no existing sharelib installed in HDFS)"
db create|upgrade|postupgrade -run [-sqlfile <FILE>] (create, upgrade or postupgrade oozie db with an"
optional sql file)"
@@ -810,11 +811,10 @@ The Oozie sharelib TAR.GZ file bundled w
hive, sqooop, and distcp actions. There is also a sharelib for HCatalog. The sharelib is required for these actions to work; any
other actions (mapreduce, shell, ssh, and java) do not require the sharelib to be installed.
-As of Oozie 4.0, the following property is included. If true, Oozie will create and ship a "launcher jar" that contains classes
-necessary for the launcher job. If false, Oozie will not do this, and it is assumed that the necessary classes are in their
+As of Oozie 4.0, the following property is included. If true, Oozie will create and ship a "launcher jar" to hdfs that contains
+classes necessary for the launcher job. If false, Oozie will not do this, and it is assumed that the necessary classes are in their
respective sharelib jars or the "oozie" sharelib instead. When false, the sharelib is required for ALL actions; when true, the
-sharelib is only required for actions that need additional jars (the original list from above). The main advantage of setting this
-to false is that launching jobs should be slightly faster.
+sharelib is only required for actions that need additional jars (the original list from above).
<verbatim>
<property>
@@ -823,6 +823,31 @@ to false is that launching jobs should b
</property>
</verbatim>
+Using sharelib CLI, sharelib files are copied to new lib_<timestamped> directory. At start, server picks the sharelib from latest
+time-stamp directory. While starting, server also purges sharelib directory which are older than sharelib retention days
+(defined as oozie.service.ShareLibService.temp.sharelib.retention.days and 7 days is default).
+
+Sharelib mapping file can be also configured. Configured file is a key value mapping, where key will be the sharelib name for the
+action and value is a comma separated list of DFS directories or jar files.
+ This can be configured in oozie-site.xml as :
+ <verbatim>
+ <!-- OOZIE -->
+ <property>
+ <name>oozie.service.ShareLibService.mapping.file</name>
+ <value></value>
+ <description>
+ Sharelib mapping files contains list of key=value,
+ where key will be the sharelib name for the action and value is a comma separated list of
+ DFS directories or jar files.
+ Example.
+ oozie.pig_10=hdfs:///share/lib/pig/pig-0.10.1/lib/
+ oozie.pig=hdfs:///share/lib/pig/pig-0.11.1/lib/
+ oozie.distcp=hdfs:///share/lib/hadoop-2.2.0/share/hadoop/tools/lib/hadoop-distcp-2.2.0.jar
+ </description>
+ </property>
+ </verbatim>
+
+
---++ Oozie Coordinators/Bundles Processing Timezone
By default Oozie runs coordinator and bundle jobs using =UTC= timezone for datetime values specified in the application
Modified: oozie/trunk/docs/src/site/twiki/DG_QuickStart.twiki
URL: http://svn.apache.org/viewvc/oozie/trunk/docs/src/site/twiki/DG_QuickStart.twiki?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/docs/src/site/twiki/DG_QuickStart.twiki (original)
+++ oozie/trunk/docs/src/site/twiki/DG_QuickStart.twiki Fri Nov 15 20:12:05 2013
@@ -118,10 +118,14 @@ required Hadoop JARs from such version i
If using the ExtJS library copy the ZIP file to the *libext/* directory.
-A "sharelib create|upgrade -fs fs_default_name [-locallib sharelib]" command is available when running oozie-setup.sh
-for uploading new or upgrading existing sharelib into hdfs where the first argument is the default fs name
+A "sharelib create -fs fs_default_name [-locallib sharelib]" command is available when running oozie-setup.sh
+for uploading new sharelib into hdfs where the first argument is the default fs name
and the second argument is the Oozie sharelib to install, it can be a tarball or the expanded version of it.
If the second argument is omitted, the Oozie sharelib tarball from the Oozie installation directory will be used.
+Upgrade command is deprecated, one should use create command to create new version of sharelib.
+Sharelib files are copied to new lib_<timestamped> directory. At start, server picks the sharelib from latest time-stamp directory.
+While starting server also purge sharelib directory which is older than sharelib retention days
+(defined as oozie.service.ShareLibService.temp.sharelib.retention.days and 7 days is default).
"prepare-war [-d directory]" command is for creating war files for oozie with an optional alternative directory other
than libext.
Modified: oozie/trunk/release-log.txt
URL: http://svn.apache.org/viewvc/oozie/trunk/release-log.txt?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/release-log.txt (original)
+++ oozie/trunk/release-log.txt Fri Nov 15 20:12:05 2013
@@ -1,5 +1,6 @@
-- Oozie 4.1.0 release (trunk - unreleased)
+OOZIE-1584 Setup sharelib using script and pickup latest(honor ship.launcher) and remove DFS dependency at startup (puru via ryota)
OOZIE-1550 Create a safeguard to kill errant recursive workflows before they bring down oozie (rkanter)
OOZIE-1314 IllegalArgumentException: wfId cannot be empty (shwethags via virag)
OOZIE-1606 Update Curator to 2.3.0 and fix some misc minor ZK related things (rkanter)
Modified: oozie/trunk/tools/src/main/java/org/apache/oozie/tools/OozieSharelibCLI.java
URL: http://svn.apache.org/viewvc/oozie/trunk/tools/src/main/java/org/apache/oozie/tools/OozieSharelibCLI.java?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/tools/src/main/java/org/apache/oozie/tools/OozieSharelibCLI.java (original)
+++ oozie/trunk/tools/src/main/java/org/apache/oozie/tools/OozieSharelibCLI.java Fri Nov 15 20:12:05 2013
@@ -16,7 +16,15 @@
* limitations under the License.
*/
package org.apache.oozie.tools;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.text.SimpleDateFormat;
+import java.util.Collection;
+import java.util.Date;
+
import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.filefilter.WildcardFileFilter;
@@ -25,16 +33,10 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.oozie.cli.CLIParser;
-import org.apache.commons.cli.Options;
import org.apache.oozie.service.HadoopAccessorService;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.WorkflowAppService;
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.util.Collection;
-
public class OozieSharelibCLI {
public static final String[] HELP_INFO = {
"",
@@ -46,6 +48,8 @@ public class OozieSharelibCLI {
public static final String LIB_OPT = "locallib";
public static final String FS_OPT = "fs";
public static final String OOZIE_HOME = "oozie.home.dir";
+ public static final String SHARE_LIB_PREFIX = "lib_";
+
private boolean used;
public static void main(String[] args) throws Exception{
@@ -75,8 +79,11 @@ public class OozieSharelibCLI {
CLIParser parser = new CLIParser("oozie-setup.sh", HELP_INFO);
String oozieHome = System.getProperty(OOZIE_HOME);
parser.addCommand(HELP_CMD, "", "display usage for all commands or specified command", new Options(), false);
- parser.addCommand(CREATE_CMD, "", "create oozie sharelib", createUpgradeOptions(CREATE_CMD), false);
- parser.addCommand(UPGRADE_CMD, "", "upgrade oozie sharelib", createUpgradeOptions(UPGRADE_CMD), false);
+ parser.addCommand(CREATE_CMD, "", "create a new timestamped version of oozie sharelib",
+ createUpgradeOptions(CREATE_CMD), false);
+ parser.addCommand(UPGRADE_CMD, "",
+ "[deprecated][use command \"create\" to create new version] upgrade oozie sharelib \n",
+ createUpgradeOptions(UPGRADE_CMD), false);
try {
final CLIParser.Command command = parser.parse(args);
@@ -137,6 +144,11 @@ public class OozieSharelibCLI {
WorkflowAppService lwas = services.get(WorkflowAppService.class);
HadoopAccessorService has = services.get(HadoopAccessorService.class);
Path dstPath = lwas.getSystemLibPath();
+
+ if (sharelibAction.equals(CREATE_CMD) || sharelibAction.equals(UPGRADE_CMD)){
+ dstPath= new Path(dstPath.toString() + Path.SEPARATOR + SHARE_LIB_PREFIX + getTimestampDirectory() );
+ }
+
System.out.println("the destination path for sharelib is: " + dstPath);
URI uri = new Path(hdfsUri).toUri();
@@ -148,19 +160,6 @@ public class OozieSharelibCLI {
throw new IOException(srcPath + " cannot be found");
}
- if (fs.exists(dstPath)){
- if (sharelibAction.equals(CREATE_CMD)){
- throw new IllegalStateException("share lib already exists");
- }
-
- fs.delete(dstPath, true);
- }
- else {
- if (sharelibAction.equals(UPGRADE_CMD)){
- throw new IllegalStateException("original share lib doesn't exist");
- }
- }
-
fs.copyFromLocalFile(false, srcPath, dstPath);
services.destroy();
FileUtils.deleteDirectory(temp);
@@ -185,4 +184,11 @@ public class OozieSharelibCLI {
return 1;
}
}
+
+ public String getTimestampDirectory() {
+ SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmmss");
+ Date date = new Date();
+ return dateFormat.format(date).toString();
+ }
+
}
Modified: oozie/trunk/tools/src/test/java/org/apache/oozie/tools/TestOozieSharelibCLI.java
URL: http://svn.apache.org/viewvc/oozie/trunk/tools/src/test/java/org/apache/oozie/tools/TestOozieSharelibCLI.java?rev=1542383&r1=1542382&r2=1542383&view=diff
==============================================================================
--- oozie/trunk/tools/src/test/java/org/apache/oozie/tools/TestOozieSharelibCLI.java (original)
+++ oozie/trunk/tools/src/test/java/org/apache/oozie/tools/TestOozieSharelibCLI.java Fri Nov 15 20:12:05 2013
@@ -21,17 +21,24 @@ package org.apache.oozie.tools;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileWriter;
+import java.io.IOException;
import java.io.PrintStream;
import java.io.Writer;
import java.net.URI;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
import org.apache.oozie.service.HadoopAccessorService;
import org.apache.oozie.service.ServiceException;
import org.apache.oozie.service.Services;
+import org.apache.oozie.service.ShareLibService;
import org.apache.oozie.service.WorkflowAppService;
import org.apache.oozie.test.XTestCase;
import org.junit.AfterClass;
@@ -72,8 +79,11 @@ public class TestOozieSharelibCLI extend
try {
String[] argsHelp = { "help" };
assertEquals(0, execOozieSharelibCLICommands(argsHelp));
- assertTrue(data.toString().contains("oozie-setup.sh create <OPTIONS> : create oozie sharelib"));
- assertTrue(data.toString().contains("oozie-setup.sh upgrade <OPTIONS> : upgrade oozie sharelib"));
+ assertTrue(data.toString().contains(
+ "oozie-setup.sh create <OPTIONS> : create a new timestamped version of oozie sharelib"));
+ assertTrue(data.toString().contains(
+ "oozie-setup.sh upgrade <OPTIONS> : [deprecated][use command \"create\" "
+ + "to create new version] upgrade oozie sharelib "));
assertTrue(data.toString().contains(" oozie-setup.sh help"));
}
finally {
@@ -104,9 +114,13 @@ public class TestOozieSharelibCLI extend
assertEquals(0, execOozieSharelibCLICommands(argsCreate));
FileSystem fs = getTargetFileSysyem();
+ ShareLibService sharelibService = getServices().get(ShareLibService.class);
+
// test files in new folder
- assertEquals(9, fs.getFileStatus(new Path(getDistPath(), "file1")).getLen());
- assertEquals(10, fs.getFileStatus(new Path(getDistPath(), "file2")).getLen());
+ assertEquals(9, fs.getFileStatus(new Path(sharelibService.getLatestLibPath(getDistPath(),
+ ShareLibService.SHARED_LIB_PREFIX), "file1")).getLen());
+ assertEquals(10, fs.getFileStatus(new Path(sharelibService.getLatestLibPath(getDistPath(),
+ ShareLibService.SHARED_LIB_PREFIX), "file2")).getLen());
}
@@ -144,8 +158,9 @@ public class TestOozieSharelibCLI extend
private Services getServices() throws ServiceException {
if (services == null) {
services = new Services();
- services.getConf().set(Services.CONF_SERVICE_CLASSES,
- "org.apache.oozie.service.LiteWorkflowAppService, org.apache.oozie.service.HadoopAccessorService");
+ services.getConf()
+ .set(Services.CONF_SERVICE_CLASSES,"org.apache.oozie.service.LiteWorkflowAppService,"
+ + "org.apache.oozie.service.HadoopAccessorService,org.apache.oozie.service.ShareLibService");
services.init();
}
return services;