001/** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018 019package org.apache.oozie.action.hadoop; 020 021import java.io.ByteArrayOutputStream; 022import java.io.File; 023import java.io.FileNotFoundException; 024import java.io.IOException; 025import java.io.PrintStream; 026import java.io.StringReader; 027import java.net.ConnectException; 028import java.net.URI; 029import java.net.URISyntaxException; 030import java.net.UnknownHostException; 031import java.util.ArrayList; 032import java.util.HashMap; 033import java.util.HashSet; 034import java.util.Iterator; 035import java.util.List; 036import java.util.Map; 037import java.util.Properties; 038import java.util.Set; 039import java.util.Map.Entry; 040import java.util.regex.Matcher; 041import java.util.regex.Pattern; 042 043import org.apache.hadoop.conf.Configuration; 044import org.apache.hadoop.filecache.DistributedCache; 045import org.apache.hadoop.fs.FileStatus; 046import org.apache.hadoop.fs.FileSystem; 047import org.apache.hadoop.fs.Path; 048import org.apache.hadoop.fs.permission.AccessControlException; 049import org.apache.hadoop.io.Text; 050import org.apache.hadoop.mapred.JobClient; 051import org.apache.hadoop.mapred.JobConf; 052import org.apache.hadoop.mapred.JobID; 053import org.apache.hadoop.mapred.RunningJob; 054import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier; 055import org.apache.hadoop.util.DiskChecker; 056import org.apache.oozie.WorkflowActionBean; 057import org.apache.oozie.WorkflowJobBean; 058import org.apache.oozie.action.ActionExecutor; 059import org.apache.oozie.action.ActionExecutorException; 060import org.apache.oozie.client.OozieClient; 061import org.apache.oozie.client.WorkflowAction; 062import org.apache.oozie.command.wf.ActionStartXCommand; 063import org.apache.oozie.service.ConfigurationService; 064import org.apache.oozie.service.HadoopAccessorException; 065import org.apache.oozie.service.HadoopAccessorService; 066import org.apache.oozie.service.Services; 067import org.apache.oozie.service.ShareLibService; 068import org.apache.oozie.service.URIHandlerService; 069import org.apache.oozie.service.WorkflowAppService; 070import org.apache.oozie.util.ELEvaluator; 071import org.apache.oozie.util.JobUtils; 072import org.apache.oozie.util.LogUtils; 073import org.apache.oozie.util.PropertiesUtils; 074import org.apache.oozie.util.XConfiguration; 075import org.apache.oozie.util.XLog; 076import org.apache.oozie.util.XmlUtils; 077import org.apache.oozie.util.ELEvaluationException; 078import org.jdom.Element; 079import org.jdom.JDOMException; 080import org.jdom.Namespace; 081import org.apache.hadoop.security.token.Token; 082import org.apache.hadoop.security.token.TokenIdentifier; 083import org.apache.oozie.hadoop.utils.HadoopShims; 084 085public class JavaActionExecutor extends ActionExecutor { 086 087 private static final String HADOOP_USER = "user.name"; 088 public static final String HADOOP_JOB_TRACKER = "mapred.job.tracker"; 089 public static final String HADOOP_JOB_TRACKER_2 = "mapreduce.jobtracker.address"; 090 public static final String HADOOP_YARN_RM = "yarn.resourcemanager.address"; 091 public static final String HADOOP_NAME_NODE = "fs.default.name"; 092 private static final String HADOOP_JOB_NAME = "mapred.job.name"; 093 public static final String OOZIE_COMMON_LIBDIR = "oozie"; 094 private static final Set<String> DISALLOWED_PROPERTIES = new HashSet<String>(); 095 public final static String MAX_EXTERNAL_STATS_SIZE = "oozie.external.stats.max.size"; 096 public static final String ACL_VIEW_JOB = "mapreduce.job.acl-view-job"; 097 public static final String ACL_MODIFY_JOB = "mapreduce.job.acl-modify-job"; 098 public static final String HADOOP_YARN_UBER_MODE = "mapreduce.job.ubertask.enable"; 099 public static final String HADOOP_MAP_MEMORY_MB = "mapreduce.map.memory.mb"; 100 public static final String HADOOP_CHILD_JAVA_OPTS = "mapred.child.java.opts"; 101 public static final String HADOOP_MAP_JAVA_OPTS = "mapreduce.map.java.opts"; 102 public static final String HADOOP_REDUCE_JAVA_OPTS = "mapreduce.reduce.java.opts"; 103 public static final String HADOOP_CHILD_JAVA_ENV = "mapred.child.env"; 104 public static final String HADOOP_MAP_JAVA_ENV = "mapreduce.map.env"; 105 public static final String YARN_AM_RESOURCE_MB = "yarn.app.mapreduce.am.resource.mb"; 106 public static final String YARN_AM_COMMAND_OPTS = "yarn.app.mapreduce.am.command-opts"; 107 public static final String YARN_AM_ENV = "yarn.app.mapreduce.am.env"; 108 private static final String JAVA_MAIN_CLASS_NAME = "org.apache.oozie.action.hadoop.JavaMain"; 109 public static final int YARN_MEMORY_MB_MIN = 512; 110 private static int maxActionOutputLen; 111 private static int maxExternalStatsSize; 112 private static int maxFSGlobMax; 113 private static final String SUCCEEDED = "SUCCEEDED"; 114 private static final String KILLED = "KILLED"; 115 private static final String FAILED = "FAILED"; 116 private static final String FAILED_KILLED = "FAILED/KILLED"; 117 protected XLog LOG = XLog.getLog(getClass()); 118 private static final Pattern heapPattern = Pattern.compile("-Xmx(([0-9]+)[mMgG])"); 119 private static final String JAVA_TMP_DIR_SETTINGS = "-Djava.io.tmpdir="; 120 public static final String CONF_HADOOP_YARN_UBER_MODE = "oozie.action.launcher." + HADOOP_YARN_UBER_MODE; 121 public static final String HADOOP_JOB_CLASSLOADER = "mapreduce.job.classloader"; 122 public static final String HADOOP_USER_CLASSPATH_FIRST = "mapreduce.user.classpath.first"; 123 public static final String OOZIE_CREDENTIALS_SKIP = "oozie.credentials.skip"; 124 125 static { 126 DISALLOWED_PROPERTIES.add(HADOOP_USER); 127 DISALLOWED_PROPERTIES.add(HADOOP_JOB_TRACKER); 128 DISALLOWED_PROPERTIES.add(HADOOP_NAME_NODE); 129 DISALLOWED_PROPERTIES.add(HADOOP_JOB_TRACKER_2); 130 DISALLOWED_PROPERTIES.add(HADOOP_YARN_RM); 131 } 132 133 public JavaActionExecutor() { 134 this("java"); 135 } 136 137 protected JavaActionExecutor(String type) { 138 super(type); 139 } 140 141 public static List<Class> getCommonLauncherClasses() { 142 List<Class> classes = new ArrayList<Class>(); 143 classes.add(LauncherMapper.class); 144 classes.add(OozieLauncherInputFormat.class); 145 classes.add(LauncherMainHadoopUtils.class); 146 classes.add(HadoopShims.class); 147 classes.addAll(Services.get().get(URIHandlerService.class).getClassesForLauncher()); 148 return classes; 149 } 150 151 public List<Class> getLauncherClasses() { 152 List<Class> classes = new ArrayList<Class>(); 153 try { 154 classes.add(Class.forName(JAVA_MAIN_CLASS_NAME)); 155 } 156 catch (ClassNotFoundException e) { 157 throw new RuntimeException("Class not found", e); 158 } 159 return classes; 160 } 161 162 @Override 163 public void initActionType() { 164 super.initActionType(); 165 maxActionOutputLen = ConfigurationService.getInt(LauncherMapper.CONF_OOZIE_ACTION_MAX_OUTPUT_DATA); 166 //Get the limit for the maximum allowed size of action stats 167 maxExternalStatsSize = ConfigurationService.getInt(JavaActionExecutor.MAX_EXTERNAL_STATS_SIZE); 168 maxExternalStatsSize = (maxExternalStatsSize == -1) ? Integer.MAX_VALUE : maxExternalStatsSize; 169 //Get the limit for the maximum number of globbed files/dirs for FS operation 170 maxFSGlobMax = ConfigurationService.getInt(LauncherMapper.CONF_OOZIE_ACTION_FS_GLOB_MAX); 171 172 registerError(UnknownHostException.class.getName(), ActionExecutorException.ErrorType.TRANSIENT, "JA001"); 173 registerError(AccessControlException.class.getName(), ActionExecutorException.ErrorType.NON_TRANSIENT, 174 "JA002"); 175 registerError(DiskChecker.DiskOutOfSpaceException.class.getName(), 176 ActionExecutorException.ErrorType.NON_TRANSIENT, "JA003"); 177 registerError(org.apache.hadoop.hdfs.protocol.QuotaExceededException.class.getName(), 178 ActionExecutorException.ErrorType.NON_TRANSIENT, "JA004"); 179 registerError(org.apache.hadoop.hdfs.server.namenode.SafeModeException.class.getName(), 180 ActionExecutorException.ErrorType.NON_TRANSIENT, "JA005"); 181 registerError(ConnectException.class.getName(), ActionExecutorException.ErrorType.TRANSIENT, " JA006"); 182 registerError(JDOMException.class.getName(), ActionExecutorException.ErrorType.ERROR, "JA007"); 183 registerError(FileNotFoundException.class.getName(), ActionExecutorException.ErrorType.ERROR, "JA008"); 184 registerError(IOException.class.getName(), ActionExecutorException.ErrorType.TRANSIENT, "JA009"); 185 } 186 187 188 /** 189 * Get the maximum allowed size of stats 190 * 191 * @return maximum size of stats 192 */ 193 public static int getMaxExternalStatsSize() { 194 return maxExternalStatsSize; 195 } 196 197 static void checkForDisallowedProps(Configuration conf, String confName) throws ActionExecutorException { 198 for (String prop : DISALLOWED_PROPERTIES) { 199 if (conf.get(prop) != null) { 200 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "JA010", 201 "Property [{0}] not allowed in action [{1}] configuration", prop, confName); 202 } 203 } 204 } 205 206 public JobConf createBaseHadoopConf(Context context, Element actionXml) { 207 Namespace ns = actionXml.getNamespace(); 208 String jobTracker = actionXml.getChild("job-tracker", ns).getTextTrim(); 209 String nameNode = actionXml.getChild("name-node", ns).getTextTrim(); 210 JobConf conf = Services.get().get(HadoopAccessorService.class).createJobConf(jobTracker); 211 conf.set(HADOOP_USER, context.getProtoActionConf().get(WorkflowAppService.HADOOP_USER)); 212 conf.set(HADOOP_JOB_TRACKER, jobTracker); 213 conf.set(HADOOP_JOB_TRACKER_2, jobTracker); 214 conf.set(HADOOP_YARN_RM, jobTracker); 215 conf.set(HADOOP_NAME_NODE, nameNode); 216 conf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "true"); 217 return conf; 218 } 219 220 private void injectLauncherProperties(Configuration srcConf, Configuration launcherConf) { 221 for (Map.Entry<String, String> entry : srcConf) { 222 if (entry.getKey().startsWith("oozie.launcher.")) { 223 String name = entry.getKey().substring("oozie.launcher.".length()); 224 String value = entry.getValue(); 225 // setting original KEY 226 launcherConf.set(entry.getKey(), value); 227 // setting un-prefixed key (to allow Hadoop job config 228 // for the launcher job 229 launcherConf.set(name, value); 230 } 231 } 232 } 233 234 Configuration setupLauncherConf(Configuration conf, Element actionXml, Path appPath, Context context) 235 throws ActionExecutorException { 236 try { 237 Namespace ns = actionXml.getNamespace(); 238 XConfiguration launcherConf = new XConfiguration(); 239 // Inject action defaults for launcher 240 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 241 XConfiguration actionDefaultConf = has.createActionDefaultConf(conf.get(HADOOP_JOB_TRACKER), getType()); 242 injectLauncherProperties(actionDefaultConf, launcherConf); 243 // Inject <configuration> for launcher 244 Element e = actionXml.getChild("configuration", ns); 245 if (e != null) { 246 String strConf = XmlUtils.prettyPrint(e).toString(); 247 XConfiguration inlineConf = new XConfiguration(new StringReader(strConf)); 248 injectLauncherProperties(inlineConf, launcherConf); 249 } 250 // Inject use uber mode for launcher 251 injectLauncherUseUberMode(launcherConf); 252 XConfiguration.copy(launcherConf, conf); 253 checkForDisallowedProps(launcherConf, "launcher configuration"); 254 e = actionXml.getChild("config-class", actionXml.getNamespace()); 255 if (e != null) { 256 conf.set(LauncherMapper.OOZIE_ACTION_CONFIG_CLASS, e.getTextTrim()); 257 } 258 return conf; 259 } 260 catch (IOException ex) { 261 throw convertException(ex); 262 } 263 } 264 265 void injectLauncherUseUberMode(Configuration launcherConf) { 266 // Set Uber Mode for the launcher (YARN only, ignored by MR1) 267 // Priority: 268 // 1. action's <configuration> 269 // 2. oozie.action.#action-type#.launcher.mapreduce.job.ubertask.enable 270 // 3. oozie.action.launcher.mapreduce.job.ubertask.enable 271 if (launcherConf.get(HADOOP_YARN_UBER_MODE) == null) { 272 if (ConfigurationService.get("oozie.action." + getType() + ".launcher." + HADOOP_YARN_UBER_MODE).length() > 0) { 273 if (ConfigurationService.getBoolean("oozie.action." + getType() + ".launcher." + HADOOP_YARN_UBER_MODE)) { 274 launcherConf.setBoolean(HADOOP_YARN_UBER_MODE, true); 275 } 276 } else { 277 if (ConfigurationService.getBoolean("oozie.action.launcher." + HADOOP_YARN_UBER_MODE)) { 278 launcherConf.setBoolean(HADOOP_YARN_UBER_MODE, true); 279 } 280 } 281 } 282 } 283 284 void updateConfForUberMode(Configuration launcherConf) { 285 286 // child.env 287 boolean hasConflictEnv = false; 288 String launcherMapEnv = launcherConf.get(HADOOP_MAP_JAVA_ENV); 289 if (launcherMapEnv == null) { 290 launcherMapEnv = launcherConf.get(HADOOP_CHILD_JAVA_ENV); 291 } 292 String amEnv = launcherConf.get(YARN_AM_ENV); 293 StringBuffer envStr = new StringBuffer(); 294 HashMap<String, List<String>> amEnvMap = null; 295 HashMap<String, List<String>> launcherMapEnvMap = null; 296 if (amEnv != null) { 297 envStr.append(amEnv); 298 amEnvMap = populateEnvMap(amEnv); 299 } 300 if (launcherMapEnv != null) { 301 launcherMapEnvMap = populateEnvMap(launcherMapEnv); 302 if (amEnvMap != null) { 303 Iterator<String> envKeyItr = launcherMapEnvMap.keySet().iterator(); 304 while (envKeyItr.hasNext()) { 305 String envKey = envKeyItr.next(); 306 if (amEnvMap.containsKey(envKey)) { 307 List<String> amValList = amEnvMap.get(envKey); 308 List<String> launcherValList = launcherMapEnvMap.get(envKey); 309 Iterator<String> valItr = launcherValList.iterator(); 310 while (valItr.hasNext()) { 311 String val = valItr.next(); 312 if (!amValList.contains(val)) { 313 hasConflictEnv = true; 314 break; 315 } 316 else { 317 valItr.remove(); 318 } 319 } 320 if (launcherValList.isEmpty()) { 321 envKeyItr.remove(); 322 } 323 } 324 } 325 } 326 } 327 if (hasConflictEnv) { 328 launcherConf.setBoolean(HADOOP_YARN_UBER_MODE, false); 329 } 330 else { 331 if (launcherMapEnvMap != null) { 332 for (String key : launcherMapEnvMap.keySet()) { 333 List<String> launcherValList = launcherMapEnvMap.get(key); 334 for (String val : launcherValList) { 335 if (envStr.length() > 0) { 336 envStr.append(","); 337 } 338 envStr.append(key).append("=").append(val); 339 } 340 } 341 } 342 343 launcherConf.set(YARN_AM_ENV, envStr.toString()); 344 345 // memory.mb 346 int launcherMapMemoryMB = launcherConf.getInt(HADOOP_MAP_MEMORY_MB, 1536); 347 int amMemoryMB = launcherConf.getInt(YARN_AM_RESOURCE_MB, 1536); 348 // YARN_MEMORY_MB_MIN to provide buffer. 349 // suppose launcher map aggressively use high memory, need some 350 // headroom for AM 351 int memoryMB = Math.max(launcherMapMemoryMB, amMemoryMB) + YARN_MEMORY_MB_MIN; 352 // limit to 4096 in case of 32 bit 353 if (launcherMapMemoryMB < 4096 && amMemoryMB < 4096 && memoryMB > 4096) { 354 memoryMB = 4096; 355 } 356 launcherConf.setInt(YARN_AM_RESOURCE_MB, memoryMB); 357 358 // We already made mapred.child.java.opts and 359 // mapreduce.map.java.opts equal, so just start with one of them 360 String launcherMapOpts = launcherConf.get(HADOOP_MAP_JAVA_OPTS, ""); 361 String amChildOpts = launcherConf.get(YARN_AM_COMMAND_OPTS); 362 StringBuilder optsStr = new StringBuilder(); 363 int heapSizeForMap = extractHeapSizeMB(launcherMapOpts); 364 int heapSizeForAm = extractHeapSizeMB(amChildOpts); 365 int heapSize = Math.max(heapSizeForMap, heapSizeForAm) + YARN_MEMORY_MB_MIN; 366 // limit to 3584 in case of 32 bit 367 if (heapSizeForMap < 4096 && heapSizeForAm < 4096 && heapSize > 3584) { 368 heapSize = 3584; 369 } 370 if (amChildOpts != null) { 371 optsStr.append(amChildOpts); 372 } 373 optsStr.append(" ").append(launcherMapOpts.trim()); 374 if (heapSize > 0) { 375 // append calculated total heap size to the end 376 optsStr.append(" ").append("-Xmx").append(heapSize).append("m"); 377 } 378 launcherConf.set(YARN_AM_COMMAND_OPTS, optsStr.toString().trim()); 379 } 380 } 381 382 void updateConfForJavaTmpDir(Configuration conf) { 383 String amChildOpts = conf.get(YARN_AM_COMMAND_OPTS); 384 String oozieJavaTmpDirSetting = "-Djava.io.tmpdir=./tmp"; 385 if (amChildOpts != null && !amChildOpts.contains(JAVA_TMP_DIR_SETTINGS)) { 386 conf.set(YARN_AM_COMMAND_OPTS, amChildOpts + " " + oozieJavaTmpDirSetting); 387 } 388 } 389 390 private HashMap<String, List<String>> populateEnvMap(String input) { 391 HashMap<String, List<String>> envMaps = new HashMap<String, List<String>>(); 392 String[] envEntries = input.split(","); 393 for (String envEntry : envEntries) { 394 String[] envKeyVal = envEntry.split("="); 395 String envKey = envKeyVal[0].trim(); 396 List<String> valList = envMaps.get(envKey); 397 if (valList == null) { 398 valList = new ArrayList<String>(); 399 } 400 valList.add(envKeyVal[1].trim()); 401 envMaps.put(envKey, valList); 402 } 403 return envMaps; 404 } 405 406 public int extractHeapSizeMB(String input) { 407 int ret = 0; 408 if(input == null || input.equals("")) 409 return ret; 410 Matcher m = heapPattern.matcher(input); 411 String heapStr = null; 412 String heapNum = null; 413 // Grabs the last match which takes effect (in case that multiple Xmx options specified) 414 while (m.find()) { 415 heapStr = m.group(1); 416 heapNum = m.group(2); 417 } 418 if (heapStr != null) { 419 // when Xmx specified in Gigabyte 420 if(heapStr.endsWith("g") || heapStr.endsWith("G")) { 421 ret = Integer.parseInt(heapNum) * 1024; 422 } else { 423 ret = Integer.parseInt(heapNum); 424 } 425 } 426 return ret; 427 } 428 429 public static void parseJobXmlAndConfiguration(Context context, Element element, Path appPath, Configuration conf) 430 throws IOException, ActionExecutorException, HadoopAccessorException, URISyntaxException { 431 Namespace ns = element.getNamespace(); 432 Iterator<Element> it = element.getChildren("job-xml", ns).iterator(); 433 HashMap<String, FileSystem> filesystemsMap = new HashMap<String, FileSystem>(); 434 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 435 while (it.hasNext()) { 436 Element e = it.next(); 437 String jobXml = e.getTextTrim(); 438 Path pathSpecified = new Path(jobXml); 439 Path path = pathSpecified.isAbsolute() ? pathSpecified : new Path(appPath, jobXml); 440 FileSystem fs; 441 if (filesystemsMap.containsKey(path.toUri().getAuthority())) { 442 fs = filesystemsMap.get(path.toUri().getAuthority()); 443 } 444 else { 445 if (path.toUri().getAuthority() != null) { 446 fs = has.createFileSystem(context.getWorkflow().getUser(), path.toUri(), 447 has.createJobConf(path.toUri().getAuthority())); 448 } 449 else { 450 fs = context.getAppFileSystem(); 451 } 452 filesystemsMap.put(path.toUri().getAuthority(), fs); 453 } 454 Configuration jobXmlConf = new XConfiguration(fs.open(path)); 455 try { 456 String jobXmlConfString = XmlUtils.prettyPrint(jobXmlConf).toString(); 457 jobXmlConfString = XmlUtils.removeComments(jobXmlConfString); 458 jobXmlConfString = context.getELEvaluator().evaluate(jobXmlConfString, String.class); 459 jobXmlConf = new XConfiguration(new StringReader(jobXmlConfString)); 460 } 461 catch (ELEvaluationException ex) { 462 throw new ActionExecutorException(ActionExecutorException.ErrorType.TRANSIENT, "EL_EVAL_ERROR", ex 463 .getMessage(), ex); 464 } 465 catch (Exception ex) { 466 context.setErrorInfo("EL_ERROR", ex.getMessage()); 467 } 468 checkForDisallowedProps(jobXmlConf, "job-xml"); 469 XConfiguration.copy(jobXmlConf, conf); 470 } 471 Element e = element.getChild("configuration", ns); 472 if (e != null) { 473 String strConf = XmlUtils.prettyPrint(e).toString(); 474 XConfiguration inlineConf = new XConfiguration(new StringReader(strConf)); 475 checkForDisallowedProps(inlineConf, "inline configuration"); 476 XConfiguration.copy(inlineConf, conf); 477 } 478 } 479 480 Configuration setupActionConf(Configuration actionConf, Context context, Element actionXml, Path appPath) 481 throws ActionExecutorException { 482 try { 483 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 484 XConfiguration actionDefaults = has.createActionDefaultConf(actionConf.get(HADOOP_JOB_TRACKER), getType()); 485 XConfiguration.injectDefaults(actionDefaults, actionConf); 486 487 has.checkSupportedFilesystem(appPath.toUri()); 488 489 // Set the Java Main Class for the Java action to give to the Java launcher 490 setJavaMain(actionConf, actionXml); 491 492 parseJobXmlAndConfiguration(context, actionXml, appPath, actionConf); 493 494 // set cancel.delegation.token in actionConf that child job doesn't cancel delegation token 495 actionConf.setBoolean("mapreduce.job.complete.cancel.delegation.tokens", false); 496 updateConfForJavaTmpDir(actionConf); 497 return actionConf; 498 } 499 catch (IOException ex) { 500 throw convertException(ex); 501 } 502 catch (HadoopAccessorException ex) { 503 throw convertException(ex); 504 } 505 catch (URISyntaxException ex) { 506 throw convertException(ex); 507 } 508 } 509 510 Configuration addToCache(Configuration conf, Path appPath, String filePath, boolean archive) 511 throws ActionExecutorException { 512 513 URI uri = null; 514 try { 515 uri = new URI(filePath); 516 URI baseUri = appPath.toUri(); 517 if (uri.getScheme() == null) { 518 String resolvedPath = uri.getPath(); 519 if (!resolvedPath.startsWith("/")) { 520 resolvedPath = baseUri.getPath() + "/" + resolvedPath; 521 } 522 uri = new URI(baseUri.getScheme(), baseUri.getAuthority(), resolvedPath, uri.getQuery(), uri.getFragment()); 523 } 524 if (archive) { 525 DistributedCache.addCacheArchive(uri.normalize(), conf); 526 } 527 else { 528 String fileName = filePath.substring(filePath.lastIndexOf("/") + 1); 529 if (fileName.endsWith(".so") || fileName.contains(".so.")) { // .so files 530 uri = new URI(uri.getScheme(), uri.getAuthority(), uri.getPath(), uri.getQuery(), fileName); 531 DistributedCache.addCacheFile(uri.normalize(), conf); 532 } 533 else if (fileName.endsWith(".jar")) { // .jar files 534 if (!fileName.contains("#")) { 535 String user = conf.get("user.name"); 536 Path pathToAdd = new Path(uri.normalize()); 537 Services.get().get(HadoopAccessorService.class).addFileToClassPath(user, pathToAdd, conf); 538 } 539 else { 540 DistributedCache.addCacheFile(uri.normalize(), conf); 541 } 542 } 543 else { // regular files 544 if (!fileName.contains("#")) { 545 uri = new URI(uri.getScheme(), uri.getAuthority(), uri.getPath(), uri.getQuery(), fileName); 546 } 547 DistributedCache.addCacheFile(uri.normalize(), conf); 548 } 549 } 550 DistributedCache.createSymlink(conf); 551 return conf; 552 } 553 catch (Exception ex) { 554 LOG.debug( 555 "Errors when add to DistributedCache. Path=" + uri.toString() + ", archive=" + archive + ", conf=" 556 + XmlUtils.prettyPrint(conf).toString()); 557 throw convertException(ex); 558 } 559 } 560 561 public void prepareActionDir(FileSystem actionFs, Context context) throws ActionExecutorException { 562 try { 563 Path actionDir = context.getActionDir(); 564 Path tempActionDir = new Path(actionDir.getParent(), actionDir.getName() + ".tmp"); 565 if (!actionFs.exists(actionDir)) { 566 try { 567 actionFs.mkdirs(tempActionDir); 568 actionFs.rename(tempActionDir, actionDir); 569 } 570 catch (IOException ex) { 571 actionFs.delete(tempActionDir, true); 572 actionFs.delete(actionDir, true); 573 throw ex; 574 } 575 } 576 } 577 catch (Exception ex) { 578 throw convertException(ex); 579 } 580 } 581 582 void cleanUpActionDir(FileSystem actionFs, Context context) throws ActionExecutorException { 583 try { 584 Path actionDir = context.getActionDir(); 585 if (!context.getProtoActionConf().getBoolean("oozie.action.keep.action.dir", false) 586 && actionFs.exists(actionDir)) { 587 actionFs.delete(actionDir, true); 588 } 589 } 590 catch (Exception ex) { 591 throw convertException(ex); 592 } 593 } 594 595 protected void addShareLib(Configuration conf, String[] actionShareLibNames) 596 throws ActionExecutorException { 597 if (actionShareLibNames != null) { 598 try { 599 ShareLibService shareLibService = Services.get().get(ShareLibService.class); 600 FileSystem fs = shareLibService.getFileSystem(); 601 if (fs != null) { 602 for (String actionShareLibName : actionShareLibNames) { 603 List<Path> listOfPaths = shareLibService.getShareLibJars(actionShareLibName); 604 if (listOfPaths != null && !listOfPaths.isEmpty()) { 605 606 for (Path actionLibPath : listOfPaths) { 607 JobUtils.addFileToClassPath(actionLibPath, conf, fs); 608 DistributedCache.createSymlink(conf); 609 } 610 } 611 } 612 } 613 } 614 catch (IOException ex) { 615 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "It should never happen", 616 ex.getMessage()); 617 } 618 } 619 } 620 621 protected void addSystemShareLibForAction(Configuration conf) throws ActionExecutorException { 622 ShareLibService shareLibService = Services.get().get(ShareLibService.class); 623 // ShareLibService is null for test cases 624 if (shareLibService != null) { 625 try { 626 List<Path> listOfPaths = shareLibService.getSystemLibJars(JavaActionExecutor.OOZIE_COMMON_LIBDIR); 627 if (listOfPaths == null || listOfPaths.isEmpty()) { 628 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "EJ001", 629 "Could not locate Oozie sharelib"); 630 } 631 FileSystem fs = listOfPaths.get(0).getFileSystem(conf); 632 for (Path actionLibPath : listOfPaths) { 633 JobUtils.addFileToClassPath(actionLibPath, conf, fs); 634 DistributedCache.createSymlink(conf); 635 } 636 listOfPaths = shareLibService.getSystemLibJars(getType()); 637 if (listOfPaths != null) { 638 for (Path actionLibPath : listOfPaths) { 639 JobUtils.addFileToClassPath(actionLibPath, conf, fs); 640 DistributedCache.createSymlink(conf); 641 } 642 } 643 } 644 catch (IOException ex) { 645 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "It should never happen", 646 ex.getMessage()); 647 } 648 } 649 } 650 651 protected void addActionLibs(Path appPath, Configuration conf) throws ActionExecutorException { 652 String[] actionLibsStrArr = conf.getStrings("oozie.launcher.oozie.libpath"); 653 if (actionLibsStrArr != null) { 654 try { 655 for (String actionLibsStr : actionLibsStrArr) { 656 actionLibsStr = actionLibsStr.trim(); 657 if (actionLibsStr.length() > 0) 658 { 659 Path actionLibsPath = new Path(actionLibsStr); 660 String user = conf.get("user.name"); 661 FileSystem fs = Services.get().get(HadoopAccessorService.class).createFileSystem(user, appPath.toUri(), conf); 662 if (fs.exists(actionLibsPath)) { 663 FileStatus[] files = fs.listStatus(actionLibsPath); 664 for (FileStatus file : files) { 665 addToCache(conf, appPath, file.getPath().toUri().getPath(), false); 666 } 667 } 668 } 669 } 670 } 671 catch (HadoopAccessorException ex){ 672 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, 673 ex.getErrorCode().toString(), ex.getMessage()); 674 } 675 catch (IOException ex){ 676 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, 677 "It should never happen", ex.getMessage()); 678 } 679 } 680 } 681 682 @SuppressWarnings("unchecked") 683 public void setLibFilesArchives(Context context, Element actionXml, Path appPath, Configuration conf) 684 throws ActionExecutorException { 685 Configuration proto = context.getProtoActionConf(); 686 687 // Workflow lib/ 688 String[] paths = proto.getStrings(WorkflowAppService.APP_LIB_PATH_LIST); 689 if (paths != null) { 690 for (String path : paths) { 691 addToCache(conf, appPath, path, false); 692 } 693 } 694 695 // Action libs 696 addActionLibs(appPath, conf); 697 698 // files and archives defined in the action 699 for (Element eProp : (List<Element>) actionXml.getChildren()) { 700 if (eProp.getName().equals("file")) { 701 String[] filePaths = eProp.getTextTrim().split(","); 702 for (String path : filePaths) { 703 addToCache(conf, appPath, path.trim(), false); 704 } 705 } 706 else if (eProp.getName().equals("archive")) { 707 String[] archivePaths = eProp.getTextTrim().split(","); 708 for (String path : archivePaths){ 709 addToCache(conf, appPath, path.trim(), true); 710 } 711 } 712 } 713 714 addAllShareLibs(appPath, conf, context, actionXml); 715 } 716 717 // Adds action specific share libs and common share libs 718 private void addAllShareLibs(Path appPath, Configuration conf, Context context, Element actionXml) 719 throws ActionExecutorException { 720 // Add action specific share libs 721 addActionShareLib(appPath, conf, context, actionXml); 722 // Add common sharelibs for Oozie and launcher jars 723 addSystemShareLibForAction(conf); 724 } 725 726 private void addActionShareLib(Path appPath, Configuration conf, Context context, Element actionXml) 727 throws ActionExecutorException { 728 XConfiguration wfJobConf = null; 729 try { 730 wfJobConf = new XConfiguration(new StringReader(context.getWorkflow().getConf())); 731 } 732 catch (IOException ioe) { 733 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "It should never happen", 734 ioe.getMessage()); 735 } 736 // Action sharelibs are only added if user has specified to use system libpath 737 if (wfJobConf.getBoolean(OozieClient.USE_SYSTEM_LIBPATH, false)) { 738 // add action specific sharelibs 739 addShareLib(conf, getShareLibNames(context, actionXml, conf)); 740 } 741 } 742 743 744 protected String getLauncherMain(Configuration launcherConf, Element actionXml) { 745 return launcherConf.get(LauncherMapper.CONF_OOZIE_ACTION_MAIN_CLASS, JavaMain.class.getName()); 746 } 747 748 private void setJavaMain(Configuration actionConf, Element actionXml) { 749 Namespace ns = actionXml.getNamespace(); 750 Element e = actionXml.getChild("main-class", ns); 751 if (e != null) { 752 actionConf.set(JavaMain.JAVA_MAIN_CLASS, e.getTextTrim()); 753 } 754 } 755 756 private static final String QUEUE_NAME = "mapred.job.queue.name"; 757 758 private static final Set<String> SPECIAL_PROPERTIES = new HashSet<String>(); 759 760 static { 761 SPECIAL_PROPERTIES.add(QUEUE_NAME); 762 SPECIAL_PROPERTIES.add(ACL_VIEW_JOB); 763 SPECIAL_PROPERTIES.add(ACL_MODIFY_JOB); 764 } 765 766 @SuppressWarnings("unchecked") 767 JobConf createLauncherConf(FileSystem actionFs, Context context, WorkflowAction action, Element actionXml, Configuration actionConf) 768 throws ActionExecutorException { 769 try { 770 771 // app path could be a file 772 Path appPathRoot = new Path(context.getWorkflow().getAppPath()); 773 if (actionFs.isFile(appPathRoot)) { 774 appPathRoot = appPathRoot.getParent(); 775 } 776 777 // launcher job configuration 778 JobConf launcherJobConf = createBaseHadoopConf(context, actionXml); 779 // cancel delegation token on a launcher job which stays alive till child job(s) finishes 780 // otherwise (in mapred action), doesn't cancel not to disturb running child job 781 launcherJobConf.setBoolean("mapreduce.job.complete.cancel.delegation.tokens", true); 782 setupLauncherConf(launcherJobConf, actionXml, appPathRoot, context); 783 784 String launcherTag = null; 785 // Extracting tag and appending action name to maintain the uniqueness. 786 if (context.getVar(ActionStartXCommand.OOZIE_ACTION_YARN_TAG) != null) { 787 launcherTag = context.getVar(ActionStartXCommand.OOZIE_ACTION_YARN_TAG); 788 } else { //Keeping it to maintain backward compatibly with test cases. 789 launcherTag = action.getId(); 790 } 791 792 // Properties for when a launcher job's AM gets restarted 793 LauncherMapperHelper.setupYarnRestartHandling(launcherJobConf, actionConf, launcherTag); 794 795 String actionShareLibProperty = actionConf.get(ACTION_SHARELIB_FOR + getType()); 796 if (actionShareLibProperty != null) { 797 launcherJobConf.set(ACTION_SHARELIB_FOR + getType(), actionShareLibProperty); 798 } 799 setLibFilesArchives(context, actionXml, appPathRoot, launcherJobConf); 800 801 String jobName = launcherJobConf.get(HADOOP_JOB_NAME); 802 if (jobName == null || jobName.isEmpty()) { 803 jobName = XLog.format( 804 "oozie:launcher:T={0}:W={1}:A={2}:ID={3}", getType(), 805 context.getWorkflow().getAppName(), action.getName(), 806 context.getWorkflow().getId()); 807 launcherJobConf.setJobName(jobName); 808 } 809 810 // Inject Oozie job information if enabled. 811 injectJobInfo(launcherJobConf, actionConf, context, action); 812 813 injectLauncherCallback(context, launcherJobConf); 814 815 String jobId = context.getWorkflow().getId(); 816 String actionId = action.getId(); 817 Path actionDir = context.getActionDir(); 818 String recoveryId = context.getRecoveryId(); 819 820 // Getting the prepare XML from the action XML 821 Namespace ns = actionXml.getNamespace(); 822 Element prepareElement = actionXml.getChild("prepare", ns); 823 String prepareXML = ""; 824 if (prepareElement != null) { 825 if (prepareElement.getChildren().size() > 0) { 826 prepareXML = XmlUtils.prettyPrint(prepareElement).toString().trim(); 827 } 828 } 829 LauncherMapperHelper.setupLauncherInfo(launcherJobConf, jobId, actionId, actionDir, recoveryId, actionConf, 830 prepareXML); 831 832 // Set the launcher Main Class 833 LauncherMapperHelper.setupMainClass(launcherJobConf, getLauncherMain(launcherJobConf, actionXml)); 834 LauncherMapperHelper.setupLauncherURIHandlerConf(launcherJobConf); 835 LauncherMapperHelper.setupMaxOutputData(launcherJobConf, maxActionOutputLen); 836 LauncherMapperHelper.setupMaxExternalStatsSize(launcherJobConf, maxExternalStatsSize); 837 LauncherMapperHelper.setupMaxFSGlob(launcherJobConf, maxFSGlobMax); 838 839 List<Element> list = actionXml.getChildren("arg", ns); 840 String[] args = new String[list.size()]; 841 for (int i = 0; i < list.size(); i++) { 842 args[i] = list.get(i).getTextTrim(); 843 } 844 LauncherMapperHelper.setupMainArguments(launcherJobConf, args); 845 846 // Make mapred.child.java.opts and mapreduce.map.java.opts equal, but give values from the latter priority; also append 847 // <java-opt> and <java-opts> and give those highest priority 848 StringBuilder opts = new StringBuilder(launcherJobConf.get(HADOOP_CHILD_JAVA_OPTS, "")); 849 if (launcherJobConf.get(HADOOP_MAP_JAVA_OPTS) != null) { 850 opts.append(" ").append(launcherJobConf.get(HADOOP_MAP_JAVA_OPTS)); 851 } 852 List<Element> javaopts = actionXml.getChildren("java-opt", ns); 853 for (Element opt: javaopts) { 854 opts.append(" ").append(opt.getTextTrim()); 855 } 856 Element opt = actionXml.getChild("java-opts", ns); 857 if (opt != null) { 858 opts.append(" ").append(opt.getTextTrim()); 859 } 860 launcherJobConf.set(HADOOP_CHILD_JAVA_OPTS, opts.toString().trim()); 861 launcherJobConf.set(HADOOP_MAP_JAVA_OPTS, opts.toString().trim()); 862 863 // setting for uber mode 864 if (launcherJobConf.getBoolean(HADOOP_YARN_UBER_MODE, false)) { 865 if (checkPropertiesToDisableUber(launcherJobConf)) { 866 launcherJobConf.setBoolean(HADOOP_YARN_UBER_MODE, false); 867 } 868 else { 869 updateConfForUberMode(launcherJobConf); 870 } 871 } 872 updateConfForJavaTmpDir(launcherJobConf); 873 874 // properties from action that are needed by the launcher (e.g. QUEUE NAME, ACLs) 875 // maybe we should add queue to the WF schema, below job-tracker 876 actionConfToLauncherConf(actionConf, launcherJobConf); 877 878 return launcherJobConf; 879 } 880 catch (Exception ex) { 881 throw convertException(ex); 882 } 883 } 884 885 private boolean checkPropertiesToDisableUber(Configuration launcherConf) { 886 boolean disable = false; 887 if (launcherConf.getBoolean(HADOOP_JOB_CLASSLOADER, false)) { 888 disable = true; 889 } 890 else if (launcherConf.getBoolean(HADOOP_USER_CLASSPATH_FIRST, false)) { 891 disable = true; 892 } 893 return disable; 894 } 895 896 private void injectCallback(Context context, Configuration conf) { 897 String callback = context.getCallbackUrl("$jobStatus"); 898 if (conf.get("job.end.notification.url") != null) { 899 LOG.warn("Overriding the action job end notification URI"); 900 } 901 conf.set("job.end.notification.url", callback); 902 } 903 904 void injectActionCallback(Context context, Configuration actionConf) { 905 injectCallback(context, actionConf); 906 } 907 908 void injectLauncherCallback(Context context, Configuration launcherConf) { 909 injectCallback(context, launcherConf); 910 } 911 912 private void actionConfToLauncherConf(Configuration actionConf, JobConf launcherConf) { 913 for (String name : SPECIAL_PROPERTIES) { 914 if (actionConf.get(name) != null && launcherConf.get("oozie.launcher." + name) == null) { 915 launcherConf.set(name, actionConf.get(name)); 916 } 917 } 918 } 919 920 public void submitLauncher(FileSystem actionFs, Context context, WorkflowAction action) throws ActionExecutorException { 921 JobClient jobClient = null; 922 boolean exception = false; 923 try { 924 Path appPathRoot = new Path(context.getWorkflow().getAppPath()); 925 926 // app path could be a file 927 if (actionFs.isFile(appPathRoot)) { 928 appPathRoot = appPathRoot.getParent(); 929 } 930 931 Element actionXml = XmlUtils.parseXml(action.getConf()); 932 933 // action job configuration 934 Configuration actionConf = createBaseHadoopConf(context, actionXml); 935 setupActionConf(actionConf, context, actionXml, appPathRoot); 936 LOG.debug("Setting LibFilesArchives "); 937 setLibFilesArchives(context, actionXml, appPathRoot, actionConf); 938 939 String jobName = actionConf.get(HADOOP_JOB_NAME); 940 if (jobName == null || jobName.isEmpty()) { 941 jobName = XLog.format("oozie:action:T={0}:W={1}:A={2}:ID={3}", 942 getType(), context.getWorkflow().getAppName(), 943 action.getName(), context.getWorkflow().getId()); 944 actionConf.set(HADOOP_JOB_NAME, jobName); 945 } 946 947 injectActionCallback(context, actionConf); 948 949 if(actionConf.get(ACL_MODIFY_JOB) == null || actionConf.get(ACL_MODIFY_JOB).trim().equals("")) { 950 // ONLY in the case where user has not given the 951 // modify-job ACL specifically 952 if (context.getWorkflow().getAcl() != null) { 953 // setting the group owning the Oozie job to allow anybody in that 954 // group to modify the jobs. 955 actionConf.set(ACL_MODIFY_JOB, context.getWorkflow().getAcl()); 956 } 957 } 958 959 // Setting the credential properties in launcher conf 960 JobConf credentialsConf = null; 961 HashMap<String, CredentialsProperties> credentialsProperties = setCredentialPropertyToActionConf(context, 962 action, actionConf); 963 if (credentialsProperties != null) { 964 965 // Adding if action need to set more credential tokens 966 credentialsConf = new JobConf(false); 967 XConfiguration.copy(actionConf, credentialsConf); 968 setCredentialTokens(credentialsConf, context, action, credentialsProperties); 969 970 // insert conf to action conf from credentialsConf 971 for (Entry<String, String> entry : credentialsConf) { 972 if (actionConf.get(entry.getKey()) == null) { 973 actionConf.set(entry.getKey(), entry.getValue()); 974 } 975 } 976 } 977 978 JobConf launcherJobConf = createLauncherConf(actionFs, context, action, actionXml, actionConf); 979 980 LOG.debug("Creating Job Client for action " + action.getId()); 981 jobClient = createJobClient(context, launcherJobConf); 982 String launcherId = LauncherMapperHelper.getRecoveryId(launcherJobConf, context.getActionDir(), context 983 .getRecoveryId()); 984 boolean alreadyRunning = launcherId != null; 985 RunningJob runningJob; 986 987 // if user-retry is on, always submit new launcher 988 boolean isUserRetry = ((WorkflowActionBean)action).isUserRetry(); 989 990 if (alreadyRunning && !isUserRetry) { 991 runningJob = jobClient.getJob(JobID.forName(launcherId)); 992 if (runningJob == null) { 993 String jobTracker = launcherJobConf.get(HADOOP_JOB_TRACKER); 994 throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "JA017", 995 "unknown job [{0}@{1}], cannot recover", launcherId, jobTracker); 996 } 997 } 998 else { 999 LOG.debug("Submitting the job through Job Client for action " + action.getId()); 1000 1001 // setting up propagation of the delegation token. 1002 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 1003 Token<DelegationTokenIdentifier> mrdt = jobClient.getDelegationToken(has 1004 .getMRDelegationTokenRenewer(launcherJobConf)); 1005 launcherJobConf.getCredentials().addToken(HadoopAccessorService.MR_TOKEN_ALIAS, mrdt); 1006 1007 // insert credentials tokens to launcher job conf if needed 1008 if (needInjectCredentials() && credentialsConf != null) { 1009 for (Token<? extends TokenIdentifier> tk : credentialsConf.getCredentials().getAllTokens()) { 1010 Text fauxAlias = new Text(tk.getKind() + "_" + tk.getService()); 1011 LOG.debug("ADDING TOKEN: " + fauxAlias); 1012 launcherJobConf.getCredentials().addToken(fauxAlias, tk); 1013 } 1014 } 1015 else { 1016 LOG.info("No need to inject credentials."); 1017 } 1018 runningJob = jobClient.submitJob(launcherJobConf); 1019 if (runningJob == null) { 1020 throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "JA017", 1021 "Error submitting launcher for action [{0}]", action.getId()); 1022 } 1023 launcherId = runningJob.getID().toString(); 1024 LOG.debug("After submission get the launcherId " + launcherId); 1025 } 1026 1027 String jobTracker = launcherJobConf.get(HADOOP_JOB_TRACKER); 1028 String consoleUrl = runningJob.getTrackingURL(); 1029 context.setStartData(launcherId, jobTracker, consoleUrl); 1030 } 1031 catch (Exception ex) { 1032 exception = true; 1033 throw convertException(ex); 1034 } 1035 finally { 1036 if (jobClient != null) { 1037 try { 1038 jobClient.close(); 1039 } 1040 catch (Exception e) { 1041 if (exception) { 1042 LOG.error("JobClient error: ", e); 1043 } 1044 else { 1045 throw convertException(e); 1046 } 1047 } 1048 } 1049 } 1050 } 1051 private boolean needInjectCredentials() { 1052 boolean methodExists = true; 1053 1054 Class klass; 1055 try { 1056 klass = Class.forName("org.apache.hadoop.mapred.JobConf"); 1057 klass.getMethod("getCredentials"); 1058 } 1059 catch (ClassNotFoundException ex) { 1060 methodExists = false; 1061 } 1062 catch (NoSuchMethodException ex) { 1063 methodExists = false; 1064 } 1065 1066 return methodExists; 1067 } 1068 1069 protected HashMap<String, CredentialsProperties> setCredentialPropertyToActionConf(Context context, 1070 WorkflowAction action, Configuration actionConf) throws Exception { 1071 HashMap<String, CredentialsProperties> credPropertiesMap = null; 1072 if (context != null && action != null) { 1073 if (!"true".equals(actionConf.get(OOZIE_CREDENTIALS_SKIP))) { 1074 XConfiguration wfJobConf = null; 1075 try { 1076 wfJobConf = new XConfiguration(new StringReader(context.getWorkflow().getConf())); 1077 } catch (IOException ioe) { 1078 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "It should never happen", 1079 ioe.getMessage()); 1080 } 1081 if ("false".equals(actionConf.get(OOZIE_CREDENTIALS_SKIP)) || 1082 !wfJobConf.getBoolean(OOZIE_CREDENTIALS_SKIP, ConfigurationService.getBoolean(OOZIE_CREDENTIALS_SKIP))) { 1083 credPropertiesMap = getActionCredentialsProperties(context, action); 1084 if (credPropertiesMap != null) { 1085 for (String key : credPropertiesMap.keySet()) { 1086 CredentialsProperties prop = credPropertiesMap.get(key); 1087 if (prop != null) { 1088 LOG.debug("Credential Properties set for action : " + action.getId()); 1089 for (String property : prop.getProperties().keySet()) { 1090 actionConf.set(property, prop.getProperties().get(property)); 1091 LOG.debug("property : '" + property + "', value : '" + prop.getProperties().get(property) 1092 + "'"); 1093 } 1094 } 1095 } 1096 } else { 1097 LOG.warn("No credential properties found for action : " + action.getId() + ", cred : " + action.getCred()); 1098 } 1099 } else { 1100 LOG.info("Skipping credentials (" + OOZIE_CREDENTIALS_SKIP + "=true)"); 1101 } 1102 } else { 1103 LOG.info("Skipping credentials (" + OOZIE_CREDENTIALS_SKIP + "=true)"); 1104 } 1105 } else { 1106 LOG.warn("context or action is null"); 1107 } 1108 return credPropertiesMap; 1109 } 1110 1111 protected void setCredentialTokens(JobConf jobconf, Context context, WorkflowAction action, 1112 HashMap<String, CredentialsProperties> credPropertiesMap) throws Exception { 1113 1114 if (context != null && action != null && credPropertiesMap != null) { 1115 // Make sure we're logged into Kerberos; if not, or near expiration, it will relogin 1116 CredentialsProvider.ensureKerberosLogin(); 1117 for (Entry<String, CredentialsProperties> entry : credPropertiesMap.entrySet()) { 1118 String credName = entry.getKey(); 1119 CredentialsProperties credProps = entry.getValue(); 1120 if (credProps != null) { 1121 CredentialsProvider credProvider = new CredentialsProvider(credProps.getType()); 1122 Credentials credentialObject = credProvider.createCredentialObject(); 1123 if (credentialObject != null) { 1124 credentialObject.addtoJobConf(jobconf, credProps, context); 1125 LOG.debug("Retrieved Credential '" + credName + "' for action " + action.getId()); 1126 } 1127 else { 1128 LOG.debug("Credentials object is null for name= " + credName + ", type=" + credProps.getType()); 1129 throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "JA020", 1130 "Could not load credentials of type [{0}] with name [{1}]]; perhaps it was not defined" 1131 + " in oozie-site.xml?", credProps.getType(), credName); 1132 } 1133 } 1134 } 1135 } 1136 1137 } 1138 1139 protected HashMap<String, CredentialsProperties> getActionCredentialsProperties(Context context, 1140 WorkflowAction action) throws Exception { 1141 HashMap<String, CredentialsProperties> props = new HashMap<String, CredentialsProperties>(); 1142 if (context != null && action != null) { 1143 String credsInAction = action.getCred(); 1144 LOG.debug("Get credential '" + credsInAction + "' properties for action : " + action.getId()); 1145 String[] credNames = credsInAction.split(","); 1146 for (String credName : credNames) { 1147 CredentialsProperties credProps = getCredProperties(context, credName); 1148 props.put(credName, credProps); 1149 } 1150 } 1151 else { 1152 LOG.warn("context or action is null"); 1153 } 1154 return props; 1155 } 1156 1157 @SuppressWarnings("unchecked") 1158 protected CredentialsProperties getCredProperties(Context context, String credName) 1159 throws Exception { 1160 CredentialsProperties credProp = null; 1161 String workflowXml = ((WorkflowJobBean) context.getWorkflow()).getWorkflowInstance().getApp().getDefinition(); 1162 XConfiguration wfjobConf = new XConfiguration(new StringReader(context.getWorkflow().getConf())); 1163 Element elementJob = XmlUtils.parseXml(workflowXml); 1164 Element credentials = elementJob.getChild("credentials", elementJob.getNamespace()); 1165 if (credentials != null) { 1166 for (Element credential : (List<Element>) credentials.getChildren("credential", credentials.getNamespace())) { 1167 String name = credential.getAttributeValue("name"); 1168 String type = credential.getAttributeValue("type"); 1169 LOG.debug("getCredProperties: Name: " + name + ", Type: " + type); 1170 if (name.equalsIgnoreCase(credName)) { 1171 credProp = new CredentialsProperties(name, type); 1172 for (Element property : (List<Element>) credential.getChildren("property", 1173 credential.getNamespace())) { 1174 String propertyName = property.getChildText("name", property.getNamespace()); 1175 String propertyValue = property.getChildText("value", property.getNamespace()); 1176 ELEvaluator eval = new ELEvaluator(); 1177 for (Map.Entry<String, String> entry : wfjobConf) { 1178 eval.setVariable(entry.getKey(), entry.getValue().trim()); 1179 } 1180 propertyName = eval.evaluate(propertyName, String.class); 1181 propertyValue = eval.evaluate(propertyValue, String.class); 1182 1183 credProp.getProperties().put(propertyName, propertyValue); 1184 LOG.debug("getCredProperties: Properties name :'" + propertyName + "', Value : '" 1185 + propertyValue + "'"); 1186 } 1187 } 1188 } 1189 } else { 1190 LOG.debug("credentials is null for the action"); 1191 } 1192 return credProp; 1193 } 1194 1195 @Override 1196 public void start(Context context, WorkflowAction action) throws ActionExecutorException { 1197 LogUtils.setLogInfo(action); 1198 try { 1199 LOG.debug("Starting action " + action.getId() + " getting Action File System"); 1200 FileSystem actionFs = context.getAppFileSystem(); 1201 LOG.debug("Preparing action Dir through copying " + context.getActionDir()); 1202 prepareActionDir(actionFs, context); 1203 LOG.debug("Action Dir is ready. Submitting the action "); 1204 submitLauncher(actionFs, context, action); 1205 LOG.debug("Action submit completed. Performing check "); 1206 check(context, action); 1207 LOG.debug("Action check is done after submission"); 1208 } 1209 catch (Exception ex) { 1210 throw convertException(ex); 1211 } 1212 } 1213 1214 @Override 1215 public void end(Context context, WorkflowAction action) throws ActionExecutorException { 1216 try { 1217 String externalStatus = action.getExternalStatus(); 1218 WorkflowAction.Status status = externalStatus.equals(SUCCEEDED) ? WorkflowAction.Status.OK 1219 : WorkflowAction.Status.ERROR; 1220 context.setEndData(status, getActionSignal(status)); 1221 } 1222 catch (Exception ex) { 1223 throw convertException(ex); 1224 } 1225 finally { 1226 try { 1227 FileSystem actionFs = context.getAppFileSystem(); 1228 cleanUpActionDir(actionFs, context); 1229 } 1230 catch (Exception ex) { 1231 throw convertException(ex); 1232 } 1233 } 1234 } 1235 1236 /** 1237 * Create job client object 1238 * 1239 * @param context 1240 * @param jobConf 1241 * @return JobClient 1242 * @throws HadoopAccessorException 1243 */ 1244 protected JobClient createJobClient(Context context, JobConf jobConf) throws HadoopAccessorException { 1245 String user = context.getWorkflow().getUser(); 1246 String group = context.getWorkflow().getGroup(); 1247 return Services.get().get(HadoopAccessorService.class).createJobClient(user, jobConf); 1248 } 1249 1250 protected RunningJob getRunningJob(Context context, WorkflowAction action, JobClient jobClient) throws Exception{ 1251 RunningJob runningJob = jobClient.getJob(JobID.forName(action.getExternalId())); 1252 return runningJob; 1253 } 1254 1255 /** 1256 * Useful for overriding in actions that do subsequent job runs 1257 * such as the MapReduce Action, where the launcher job is not the 1258 * actual job that then gets monitored. 1259 */ 1260 protected String getActualExternalId(WorkflowAction action) { 1261 return action.getExternalId(); 1262 } 1263 1264 @Override 1265 public void check(Context context, WorkflowAction action) throws ActionExecutorException { 1266 JobClient jobClient = null; 1267 boolean exception = false; 1268 LogUtils.setLogInfo(action); 1269 try { 1270 Element actionXml = XmlUtils.parseXml(action.getConf()); 1271 FileSystem actionFs = context.getAppFileSystem(); 1272 JobConf jobConf = createBaseHadoopConf(context, actionXml); 1273 jobClient = createJobClient(context, jobConf); 1274 RunningJob runningJob = getRunningJob(context, action, jobClient); 1275 if (runningJob == null) { 1276 context.setExecutionData(FAILED, null); 1277 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "JA017", 1278 "Could not lookup launched hadoop Job ID [{0}] which was associated with " + 1279 " action [{1}]. Failing this action!", getActualExternalId(action), action.getId()); 1280 } 1281 if (runningJob.isComplete()) { 1282 Path actionDir = context.getActionDir(); 1283 String newId = null; 1284 // load sequence file into object 1285 Map<String, String> actionData = LauncherMapperHelper.getActionData(actionFs, actionDir, jobConf); 1286 if (actionData.containsKey(LauncherMapper.ACTION_DATA_NEW_ID)) { 1287 newId = actionData.get(LauncherMapper.ACTION_DATA_NEW_ID); 1288 String launcherId = action.getExternalId(); 1289 runningJob = jobClient.getJob(JobID.forName(newId)); 1290 if (runningJob == null) { 1291 context.setExternalStatus(FAILED); 1292 throw new ActionExecutorException(ActionExecutorException.ErrorType.FAILED, "JA017", 1293 "Unknown hadoop job [{0}] associated with action [{1}]. Failing this action!", newId, 1294 action.getId()); 1295 } 1296 context.setExternalChildIDs(newId); 1297 LOG.info(XLog.STD, "External ID swap, old ID [{0}] new ID [{1}]", launcherId, 1298 newId); 1299 } 1300 else { 1301 String externalIDs = actionData.get(LauncherMapper.ACTION_DATA_EXTERNAL_CHILD_IDS); 1302 if (externalIDs != null) { 1303 context.setExternalChildIDs(externalIDs); 1304 LOG.info(XLog.STD, "Hadoop Jobs launched : [{0}]", externalIDs); 1305 } 1306 } 1307 if (runningJob.isComplete()) { 1308 // fetching action output and stats for the Map-Reduce action. 1309 if (newId != null) { 1310 actionData = LauncherMapperHelper.getActionData(actionFs, context.getActionDir(), jobConf); 1311 } 1312 LOG.info(XLog.STD, "action completed, external ID [{0}]", 1313 action.getExternalId()); 1314 if (LauncherMapperHelper.isMainSuccessful(runningJob)) { 1315 if (getCaptureOutput(action) && LauncherMapperHelper.hasOutputData(actionData)) { 1316 context.setExecutionData(SUCCEEDED, PropertiesUtils.stringToProperties(actionData 1317 .get(LauncherMapper.ACTION_DATA_OUTPUT_PROPS))); 1318 LOG.info(XLog.STD, "action produced output"); 1319 } 1320 else { 1321 context.setExecutionData(SUCCEEDED, null); 1322 } 1323 if (LauncherMapperHelper.hasStatsData(actionData)) { 1324 context.setExecutionStats(actionData.get(LauncherMapper.ACTION_DATA_STATS)); 1325 LOG.info(XLog.STD, "action produced stats"); 1326 } 1327 getActionData(actionFs, runningJob, action, context); 1328 } 1329 else { 1330 String errorReason; 1331 if (actionData.containsKey(LauncherMapper.ACTION_DATA_ERROR_PROPS)) { 1332 Properties props = PropertiesUtils.stringToProperties(actionData 1333 .get(LauncherMapper.ACTION_DATA_ERROR_PROPS)); 1334 String errorCode = props.getProperty("error.code"); 1335 if ("0".equals(errorCode)) { 1336 errorCode = "JA018"; 1337 } 1338 if ("-1".equals(errorCode)) { 1339 errorCode = "JA019"; 1340 } 1341 errorReason = props.getProperty("error.reason"); 1342 LOG.warn("Launcher ERROR, reason: {0}", errorReason); 1343 String exMsg = props.getProperty("exception.message"); 1344 String errorInfo = (exMsg != null) ? exMsg : errorReason; 1345 context.setErrorInfo(errorCode, errorInfo); 1346 String exStackTrace = props.getProperty("exception.stacktrace"); 1347 if (exMsg != null) { 1348 LOG.warn("Launcher exception: {0}{E}{1}", exMsg, exStackTrace); 1349 } 1350 } 1351 else { 1352 errorReason = XLog.format("LauncherMapper died, check Hadoop LOG for job [{0}:{1}]", action 1353 .getTrackerUri(), action.getExternalId()); 1354 LOG.warn(errorReason); 1355 } 1356 context.setExecutionData(FAILED_KILLED, null); 1357 } 1358 } 1359 else { 1360 context.setExternalStatus("RUNNING"); 1361 LOG.info(XLog.STD, "checking action, hadoop job ID [{0}] status [RUNNING]", 1362 runningJob.getID()); 1363 } 1364 } 1365 else { 1366 context.setExternalStatus("RUNNING"); 1367 LOG.info(XLog.STD, "checking action, hadoop job ID [{0}] status [RUNNING]", 1368 runningJob.getID()); 1369 } 1370 } 1371 catch (Exception ex) { 1372 LOG.warn("Exception in check(). Message[{0}]", ex.getMessage(), ex); 1373 exception = true; 1374 throw convertException(ex); 1375 } 1376 finally { 1377 if (jobClient != null) { 1378 try { 1379 jobClient.close(); 1380 } 1381 catch (Exception e) { 1382 if (exception) { 1383 LOG.error("JobClient error: ", e); 1384 } 1385 else { 1386 throw convertException(e); 1387 } 1388 } 1389 } 1390 } 1391 } 1392 1393 /** 1394 * Get the output data of an action. Subclasses should override this method 1395 * to get action specific output data. 1396 * 1397 * @param actionFs the FileSystem object 1398 * @param runningJob the runningJob 1399 * @param action the Workflow action 1400 * @param context executor context 1401 * 1402 */ 1403 protected void getActionData(FileSystem actionFs, RunningJob runningJob, WorkflowAction action, Context context) 1404 throws HadoopAccessorException, JDOMException, IOException, URISyntaxException { 1405 } 1406 1407 protected final void readExternalChildIDs(WorkflowAction action, Context context) throws IOException { 1408 if (action.getData() != null) { 1409 // Load stored Hadoop jobs ids and promote them as external child ids 1410 // See LauncherMain#writeExternalChildIDs for how they are written 1411 Properties props = new Properties(); 1412 props.load(new StringReader(action.getData())); 1413 context.setExternalChildIDs((String) props.get(LauncherMain.HADOOP_JOBS)); 1414 } 1415 } 1416 1417 protected boolean getCaptureOutput(WorkflowAction action) throws JDOMException { 1418 Element eConf = XmlUtils.parseXml(action.getConf()); 1419 Namespace ns = eConf.getNamespace(); 1420 Element captureOutput = eConf.getChild("capture-output", ns); 1421 return captureOutput != null; 1422 } 1423 1424 @Override 1425 public void kill(Context context, WorkflowAction action) throws ActionExecutorException { 1426 JobClient jobClient = null; 1427 boolean exception = false; 1428 try { 1429 Element actionXml = XmlUtils.parseXml(action.getConf()); 1430 JobConf jobConf = createBaseHadoopConf(context, actionXml); 1431 jobClient = createJobClient(context, jobConf); 1432 RunningJob runningJob = getRunningJob(context, action, jobClient); 1433 if (runningJob != null) { 1434 runningJob.killJob(); 1435 } 1436 context.setExternalStatus(KILLED); 1437 context.setExecutionData(KILLED, null); 1438 } 1439 catch (Exception ex) { 1440 exception = true; 1441 throw convertException(ex); 1442 } 1443 finally { 1444 try { 1445 FileSystem actionFs = context.getAppFileSystem(); 1446 cleanUpActionDir(actionFs, context); 1447 if (jobClient != null) { 1448 jobClient.close(); 1449 } 1450 } 1451 catch (Exception ex) { 1452 if (exception) { 1453 LOG.error("Error: ", ex); 1454 } 1455 else { 1456 throw convertException(ex); 1457 } 1458 } 1459 } 1460 } 1461 1462 private static Set<String> FINAL_STATUS = new HashSet<String>(); 1463 1464 static { 1465 FINAL_STATUS.add(SUCCEEDED); 1466 FINAL_STATUS.add(KILLED); 1467 FINAL_STATUS.add(FAILED); 1468 FINAL_STATUS.add(FAILED_KILLED); 1469 } 1470 1471 @Override 1472 public boolean isCompleted(String externalStatus) { 1473 return FINAL_STATUS.contains(externalStatus); 1474 } 1475 1476 1477 /** 1478 * Return the sharelib names for the action. 1479 * <p/> 1480 * If <code>NULL</code> or empty, it means that the action does not use the action 1481 * sharelib. 1482 * <p/> 1483 * If a non-empty string, i.e. <code>foo</code>, it means the action uses the 1484 * action sharelib sub-directory <code>foo</code> and all JARs in the sharelib 1485 * <code>foo</code> directory will be in the action classpath. Multiple sharelib 1486 * sub-directories can be specified as a comma separated list. 1487 * <p/> 1488 * The resolution is done using the following precedence order: 1489 * <ul> 1490 * <li><b>action.sharelib.for.#ACTIONTYPE#</b> in the action configuration</li> 1491 * <li><b>action.sharelib.for.#ACTIONTYPE#</b> in the job configuration</li> 1492 * <li><b>action.sharelib.for.#ACTIONTYPE#</b> in the oozie configuration</li> 1493 * <li>Action Executor <code>getDefaultShareLibName()</code> method</li> 1494 * </ul> 1495 * 1496 * 1497 * @param context executor context. 1498 * @param actionXml 1499 * @param conf action configuration. 1500 * @return the action sharelib names. 1501 */ 1502 protected String[] getShareLibNames(Context context, Element actionXml, Configuration conf) { 1503 String[] names = conf.getStrings(ACTION_SHARELIB_FOR + getType()); 1504 if (names == null || names.length == 0) { 1505 try { 1506 XConfiguration jobConf = new XConfiguration(new StringReader(context.getWorkflow().getConf())); 1507 names = jobConf.getStrings(ACTION_SHARELIB_FOR + getType()); 1508 if (names == null || names.length == 0) { 1509 names = Services.get().getConf().getStrings(ACTION_SHARELIB_FOR + getType()); 1510 if (names == null || names.length == 0) { 1511 String name = getDefaultShareLibName(actionXml); 1512 if (name != null) { 1513 names = new String[] { name }; 1514 } 1515 } 1516 } 1517 } 1518 catch (IOException ex) { 1519 throw new RuntimeException("It cannot happen, " + ex.toString(), ex); 1520 } 1521 } 1522 return names; 1523 } 1524 1525 private final static String ACTION_SHARELIB_FOR = "oozie.action.sharelib.for."; 1526 1527 1528 /** 1529 * Returns the default sharelib name for the action if any. 1530 * 1531 * @param actionXml the action XML fragment. 1532 * @return the sharelib name for the action, <code>NULL</code> if none. 1533 */ 1534 protected String getDefaultShareLibName(Element actionXml) { 1535 return null; 1536 } 1537 1538 /** 1539 * Sets some data for the action on completion 1540 * 1541 * @param context executor context 1542 * @param actionFs the FileSystem object 1543 */ 1544 protected void setActionCompletionData(Context context, FileSystem actionFs) throws IOException, 1545 HadoopAccessorException, URISyntaxException { 1546 } 1547 1548 private void injectJobInfo(JobConf launcherJobConf, Configuration actionConf, Context context, WorkflowAction action) { 1549 if (OozieJobInfo.isJobInfoEnabled()) { 1550 try { 1551 OozieJobInfo jobInfo = new OozieJobInfo(actionConf, context, action); 1552 String jobInfoStr = jobInfo.getJobInfo(); 1553 launcherJobConf.set(OozieJobInfo.JOB_INFO_KEY, jobInfoStr + "launcher=true"); 1554 actionConf.set(OozieJobInfo.JOB_INFO_KEY, jobInfoStr + "launcher=false"); 1555 } 1556 catch (Exception e) { 1557 // Just job info, should not impact the execution. 1558 LOG.error("Error while populating job info", e); 1559 } 1560 } 1561 } 1562 1563 public boolean requiresNameNodeJobTracker() { 1564 return true; 1565 } 1566 1567 public boolean supportsConfigurationJobXML() { 1568 return true; 1569 } 1570}