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; 020 021import org.apache.commons.lang.StringUtils; 022import org.apache.hadoop.fs.FileSystem; 023import org.apache.hadoop.fs.Path; 024import org.apache.hadoop.conf.Configuration; 025import org.apache.oozie.client.WorkflowAction; 026import org.apache.oozie.client.WorkflowJob; 027import org.apache.oozie.service.ConfigurationService; 028import org.apache.oozie.util.ELEvaluator; 029import org.apache.oozie.util.ParamChecker; 030import org.apache.oozie.util.XLog; 031import org.apache.oozie.service.HadoopAccessorException; 032import org.apache.oozie.service.Services; 033 034import java.io.ByteArrayOutputStream; 035import java.io.IOException; 036import java.io.PrintStream; 037import java.net.URISyntaxException; 038import java.util.HashMap; 039import java.util.Map; 040import java.util.Properties; 041import java.util.LinkedHashMap; 042 043/** 044 * Base action executor class. <p/> All the action executors should extend this class. 045 */ 046public abstract class ActionExecutor { 047 048 /** 049 * Configuration prefix for action executor (sub-classes) properties. 050 */ 051 public static final String CONF_PREFIX = "oozie.action."; 052 053 public static final String MAX_RETRIES = CONF_PREFIX + "retries.max"; 054 055 public static final String ACTION_RETRY_INTERVAL = CONF_PREFIX + "retry.interval"; 056 057 public static final String ACTION_RETRY_POLICY = CONF_PREFIX + "retry.policy"; 058 059 /** 060 * Error code used by {@link #convertException} when there is not register error information for an exception. 061 */ 062 public static final String ERROR_OTHER = "OTHER"; 063 064 public static enum RETRYPOLICY { 065 EXPONENTIAL, PERIODIC 066 } 067 068 private static class ErrorInfo { 069 ActionExecutorException.ErrorType errorType; 070 String errorCode; 071 Class<?> errorClass; 072 073 private ErrorInfo(ActionExecutorException.ErrorType errorType, String errorCode, Class<?> errorClass) { 074 this.errorType = errorType; 075 this.errorCode = errorCode; 076 this.errorClass = errorClass; 077 } 078 } 079 080 private static boolean initMode = false; 081 private static Map<String, Map<String, ErrorInfo>> ERROR_INFOS = new HashMap<String, Map<String, ErrorInfo>>(); 082 083 /** 084 * Context information passed to the ActionExecutor methods. 085 */ 086 public interface Context { 087 088 /** 089 * Create the callback URL for the action. 090 * 091 * @param externalStatusVar variable for the caller to inject the external status. 092 * @return the callback URL. 093 */ 094 public String getCallbackUrl(String externalStatusVar); 095 096 /** 097 * Return a proto configuration for actions with auth properties already set. 098 * 099 * @return a proto configuration for actions with auth properties already set. 100 */ 101 public Configuration getProtoActionConf(); 102 103 /** 104 * Return the workflow job. 105 * 106 * @return the workflow job. 107 */ 108 public WorkflowJob getWorkflow(); 109 110 /** 111 * Return an ELEvaluator with the context injected. 112 * 113 * @return configured ELEvaluator. 114 */ 115 public ELEvaluator getELEvaluator(); 116 117 /** 118 * Set a workflow action variable. <p/> Convenience method that prefixes the variable name with the action name 119 * plus a '.'. 120 * 121 * @param name variable name. 122 * @param value variable value, <code>null</code> removes the variable. 123 */ 124 public void setVar(String name, String value); 125 126 /** 127 * Get a workflow action variable. <p/> Convenience method that prefixes the variable name with the action name 128 * plus a '.'. 129 * 130 * @param name variable name. 131 * @return the variable value, <code>null</code> if not set. 132 */ 133 public String getVar(String name); 134 135 /** 136 * Set the action tracking information for an successfully started action. 137 * 138 * @param externalId the action external ID. 139 * @param trackerUri the action tracker URI. 140 * @param consoleUrl the action console URL. 141 */ 142 void setStartData(String externalId, String trackerUri, String consoleUrl); 143 144 /** 145 * Set the action execution completion information for an action. The action status is set to {@link 146 * org.apache.oozie.client.WorkflowAction.Status#DONE} 147 * 148 * @param externalStatus the action external end status. 149 * @param actionData the action data on completion, <code>null</code> if none. 150 */ 151 void setExecutionData(String externalStatus, Properties actionData); 152 153 /** 154 * Set execution statistics information for a particular action. The action status is set to {@link 155 * org.apache.oozie.client.WorkflowAction.Status#DONE} 156 * 157 * @param jsonStats the JSON string representation of the stats. 158 */ 159 void setExecutionStats(String jsonStats); 160 161 /** 162 * Set external child IDs for a particular action (Eg: pig). The action status is set to {@link 163 * org.apache.oozie.client.WorkflowAction.Status#DONE} 164 * 165 * @param externalChildIDs the external child IDs as a comma-delimited string. 166 */ 167 void setExternalChildIDs(String externalChildIDs); 168 169 /** 170 * Set the action end completion information for a completed action. 171 * 172 * @param status the action end status, it can be {@link org.apache.oozie.client.WorkflowAction.Status#OK} or 173 * {@link org.apache.oozie.client.WorkflowAction.Status#ERROR}. 174 * @param signalValue the action external end status. 175 */ 176 void setEndData(WorkflowAction.Status status, String signalValue); 177 178 /** 179 * Return if the executor invocation is a retry or not. 180 * 181 * @return if the executor invocation is a retry or not. 182 */ 183 boolean isRetry(); 184 185 /** 186 * Sets the external status for the action in context. 187 * 188 * @param externalStatus the external status. 189 */ 190 void setExternalStatus(String externalStatus); 191 192 /** 193 * Get the Action Recovery ID. 194 * 195 * @return recovery ID. 196 */ 197 String getRecoveryId(); 198 199 /* 200 * @return the path that will be used to store action specific data 201 * @throws IOException @throws URISyntaxException @throws HadoopAccessorException 202 */ 203 public Path getActionDir() throws HadoopAccessorException, IOException, URISyntaxException; 204 205 /** 206 * @return filesystem handle for the application deployment fs. 207 * @throws IOException 208 * @throws URISyntaxException 209 * @throws HadoopAccessorException 210 */ 211 public FileSystem getAppFileSystem() throws HadoopAccessorException, IOException, URISyntaxException; 212 213 public void setErrorInfo(String str, String exMsg); 214 } 215 216 217 /** 218 * Define the default inteval in seconds between retries. 219 */ 220 public static final long RETRY_INTERVAL = 60; 221 222 private String type; 223 private int maxRetries; 224 private long retryInterval; 225 private RETRYPOLICY retryPolicy; 226 227 /** 228 * Create an action executor with default retry parameters. 229 * 230 * @param type action executor type. 231 */ 232 protected ActionExecutor(String type) { 233 this(type, RETRY_INTERVAL); 234 } 235 236 /** 237 * Create an action executor. 238 * 239 * @param type action executor type. 240 * @param defaultRetryInterval retry interval, in seconds. 241 */ 242 protected ActionExecutor(String type, long defaultRetryInterval) { 243 this.type = ParamChecker.notEmpty(type, "type"); 244 this.maxRetries = ConfigurationService.getInt(MAX_RETRIES); 245 int retryInterval = ConfigurationService.getInt(ACTION_RETRY_INTERVAL); 246 this.retryInterval = retryInterval > 0 ? retryInterval : defaultRetryInterval; 247 this.retryPolicy = getRetryPolicyFromConf(); 248 } 249 250 private RETRYPOLICY getRetryPolicyFromConf() { 251 String retryPolicy = ConfigurationService.get(ACTION_RETRY_POLICY); 252 if (StringUtils.isBlank(retryPolicy)) { 253 return RETRYPOLICY.PERIODIC; 254 } else { 255 try { 256 return RETRYPOLICY.valueOf(retryPolicy.toUpperCase().trim()); 257 } catch (IllegalArgumentException e) { 258 return RETRYPOLICY.PERIODIC; 259 } 260 } 261 } 262 263 /** 264 * Clear all init settings for all action types. 265 */ 266 public static void resetInitInfo() { 267 if (!initMode) { 268 throw new IllegalStateException("Error, action type info locked"); 269 } 270 ERROR_INFOS.clear(); 271 } 272 273 /** 274 * Enable action type initialization. 275 */ 276 public static void enableInit() { 277 initMode = true; 278 } 279 280 /** 281 * Disable action type initialization. 282 */ 283 public static void disableInit() { 284 initMode = false; 285 } 286 287 /** 288 * Invoked once at system initialization time. <p/> It can be used to register error information for the expected 289 * exceptions. Exceptions should be register from subclasses to superclasses to ensure proper detection, same thing 290 * that it is done in a normal catch. <p/> This method should invoke the {@link #registerError} method to register 291 * all its possible errors. <p/> Subclasses overriding must invoke super. 292 */ 293 public void initActionType() { 294 XLog.getLog(getClass()).trace(" Init Action Type : [{0}]", getType()); 295 ERROR_INFOS.put(getType(), new LinkedHashMap<String, ErrorInfo>()); 296 } 297 298 /** 299 * Return the system ID, this ID is defined in Oozie configuration. 300 * 301 * @return the system ID. 302 */ 303 public String getOozieSystemId() { 304 return Services.get().getSystemId(); 305 } 306 307 /** 308 * Return the runtime directory of the Oozie instance. <p/> The directory is created under TMP and it is always a 309 * new directory per system initialization. 310 * 311 * @return the runtime directory of the Oozie instance. 312 */ 313 public String getOozieRuntimeDir() { 314 return Services.get().getRuntimeDir(); 315 } 316 317 /** 318 * Return Oozie configuration. <p/> This is useful for actions that need access to configuration properties. 319 * 320 * @return Oozie configuration. 321 */ 322 public Configuration getOozieConf() { 323 return Services.get().getConf(); 324 } 325 326 /** 327 * Register error handling information for an exception. 328 * 329 * @param exClass excpetion class name (to work in case of a particular exception not being in the classpath, needed 330 * to be able to handle multiple version of Hadoop or other JARs used by executors with the same codebase). 331 * @param errorType error type for the exception. 332 * @param errorCode error code for the exception. 333 */ 334 protected void registerError(String exClass, ActionExecutorException.ErrorType errorType, String errorCode) { 335 if (!initMode) { 336 throw new IllegalStateException("Error, action type info locked"); 337 } 338 try { 339 Class errorClass = Thread.currentThread().getContextClassLoader().loadClass(exClass); 340 Map<String, ErrorInfo> executorErrorInfo = ERROR_INFOS.get(getType()); 341 executorErrorInfo.put(exClass, new ErrorInfo(errorType, errorCode, errorClass)); 342 } 343 catch (ClassNotFoundException cnfe) { 344 XLog.getLog(getClass()).warn( 345 "Exception [{0}] not in classpath, ActionExecutor [{1}] will handle it as ERROR", exClass, 346 getType()); 347 } 348 catch (java.lang.NoClassDefFoundError err) { 349 ByteArrayOutputStream baos = new ByteArrayOutputStream(); 350 err.printStackTrace(new PrintStream(baos)); 351 XLog.getLog(getClass()).warn(baos.toString()); 352 } 353 } 354 355 /** 356 * Return the action executor type. 357 * 358 * @return the action executor type. 359 */ 360 public String getType() { 361 return type; 362 } 363 364 /** 365 * Return the maximum number of retries for the action executor. 366 * 367 * @return the maximum number of retries for the action executor. 368 */ 369 public int getMaxRetries() { 370 return maxRetries; 371 } 372 373 /** 374 * Set the maximum number of retries for the action executor. 375 * 376 * @param maxRetries the maximum number of retries. 377 */ 378 public void setMaxRetries(int maxRetries) { 379 this.maxRetries = maxRetries; 380 } 381 382 /** 383 * Return the retry policy for the action executor. 384 * 385 * @return the retry policy for the action executor. 386 */ 387 public RETRYPOLICY getRetryPolicy() { 388 return retryPolicy; 389 } 390 391 /** 392 * Sets the retry policy for the action executor. 393 * 394 * @param retryPolicy retry policy for the action executor. 395 */ 396 public void setRetryPolicy(RETRYPOLICY retryPolicy) { 397 this.retryPolicy = retryPolicy; 398 } 399 400 /** 401 * Return the retry interval for the action executor in seconds. 402 * 403 * @return the retry interval for the action executor in seconds. 404 */ 405 public long getRetryInterval() { 406 return retryInterval; 407 } 408 409 /** 410 * Sets the retry interval for the action executor. 411 * 412 * @param retryInterval retry interval in seconds. 413 */ 414 public void setRetryInterval(long retryInterval) { 415 this.retryInterval = retryInterval; 416 } 417 418 /** 419 * Utility method to handle exceptions in the {@link #start}, {@link #end}, {@link #kill} and {@link #check} methods 420 * <p/> It uses the error registry to convert exceptions to {@link ActionExecutorException}s. 421 * 422 * @param ex exception to convert. 423 * @return ActionExecutorException converted exception. 424 */ 425 @SuppressWarnings({"ThrowableInstanceNeverThrown"}) 426 protected ActionExecutorException convertException(Exception ex) { 427 if (ex instanceof ActionExecutorException) { 428 return (ActionExecutorException) ex; 429 } 430 431 ActionExecutorException aee = null; 432 // Check the cause of the exception first 433 if (ex.getCause() != null) { 434 aee = convertExceptionHelper(ex.getCause()); 435 } 436 // If the cause isn't registered or doesn't exist, check the exception itself 437 if (aee == null) { 438 aee = convertExceptionHelper(ex); 439 // If the cause isn't registered either, then just create a new ActionExecutorException 440 if (aee == null) { 441 String exClass = ex.getClass().getName(); 442 String errorCode = exClass.substring(exClass.lastIndexOf(".") + 1); 443 aee = new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, errorCode, "{0}", ex.getMessage(), ex); 444 } 445 } 446 return aee; 447 } 448 449 private ActionExecutorException convertExceptionHelper(Throwable ex) { 450 Map<String, ErrorInfo> executorErrorInfo = ERROR_INFOS.get(getType()); 451 // Check if we have registered ex 452 ErrorInfo classErrorInfo = executorErrorInfo.get(ex.getClass().getName()); 453 if (classErrorInfo != null) { 454 return new ActionExecutorException(classErrorInfo.errorType, classErrorInfo.errorCode, "{0}", ex.getMessage(), ex); 455 } 456 // Else, check if a parent class of ex is registered 457 else { 458 for (ErrorInfo errorInfo : executorErrorInfo.values()) { 459 if (errorInfo.errorClass.isInstance(ex)) { 460 return new ActionExecutorException(errorInfo.errorType, errorInfo.errorCode, "{0}", ex.getMessage(), ex); 461 } 462 } 463 } 464 return null; 465 } 466 467 /** 468 * Convenience method that return the signal for an Action based on the action status. 469 * 470 * @param status action status. 471 * @return the action signal. 472 */ 473 protected String getActionSignal(WorkflowAction.Status status) { 474 switch (status) { 475 case OK: 476 return "OK"; 477 case ERROR: 478 case KILLED: 479 return "ERROR"; 480 default: 481 throw new IllegalArgumentException("Action status for signal can only be OK or ERROR"); 482 } 483 } 484 485 /** 486 * Return the path that will be used to store action specific data 487 * 488 * @param jobId Worfklow ID 489 * @param action Action 490 * @param key An Identifier 491 * @param temp temp directory flag 492 * @return A string that has the path 493 */ 494 protected String getActionDirPath(String jobId, WorkflowAction action, String key, boolean temp) { 495 String name = jobId + "/" + action.getName() + "--" + key; 496 if (temp) { 497 name += ".temp"; 498 } 499 return getOozieSystemId() + "/" + name; 500 } 501 502 /** 503 * Return the path that will be used to store action specific data. 504 * 505 * @param jobId Workflow ID 506 * @param action Action 507 * @param key An identifier 508 * @param temp Temp directory flag 509 * @return Path to the directory 510 */ 511 public Path getActionDir(String jobId, WorkflowAction action, String key, boolean temp) { 512 return new Path(getActionDirPath(jobId, action, key, temp)); 513 } 514 515 /** 516 * Start an action. <p/> The {@link Context#setStartData} method must be called within this method. <p/> If the 517 * action has completed, the {@link Context#setExecutionData} method must be called within this method. 518 * 519 * @param context executor context. 520 * @param action the action to start. 521 * @throws ActionExecutorException thrown if the action could not start. 522 */ 523 public abstract void start(Context context, WorkflowAction action) throws ActionExecutorException; 524 525 /** 526 * End an action after it has executed. <p/> The {@link Context#setEndData} method must be called within this 527 * method. 528 * 529 * @param context executor context. 530 * @param action the action to end. 531 * @throws ActionExecutorException thrown if the action could not end. 532 */ 533 public abstract void end(Context context, WorkflowAction action) throws ActionExecutorException; 534 535 /** 536 * Check if an action has completed. This method must be implemented by Async Action Executors. <p/> If the action 537 * has completed, the {@link Context#setExecutionData} method must be called within this method. <p/> If the action 538 * has not completed, the {@link Context#setExternalStatus} method must be called within this method. 539 * 540 * @param context executor context. 541 * @param action the action to end. 542 * @throws ActionExecutorException thrown if the action could not be checked. 543 */ 544 public abstract void check(Context context, WorkflowAction action) throws ActionExecutorException; 545 546 /** 547 * Kill an action. <p/> The {@link Context#setEndData} method must be called within this method. 548 * 549 * @param context executor context. 550 * @param action the action to kill. 551 * @throws ActionExecutorException thrown if the action could not be killed. 552 */ 553 public abstract void kill(Context context, WorkflowAction action) throws ActionExecutorException; 554 555 /** 556 * Return if the external status indicates that the action has completed. 557 * 558 * @param externalStatus external status to check. 559 * @return if the external status indicates that the action has completed. 560 */ 561 public abstract boolean isCompleted(String externalStatus); 562 563 /** 564 * Returns true if this action type requires a NameNode and JobTracker. These can either be specified directly in the action 565 * via <name-node> and <job-tracker>, from the fields in the global section, or from their default values. If 566 * false, Oozie won't ensure (i.e. won't throw an Exception if non-existant) that this action type has these values. 567 * 568 * @return true if a NameNode and JobTracker are required; false if not 569 */ 570 public boolean requiresNameNodeJobTracker() { 571 return false; 572 } 573 574 /** 575 * Returns true if this action type supports a Configuration and JobXML. In this case, Oozie will include the 576 * <configuration> and <job-xml> elements from the global section (if provided) with the action. If false, Oozie 577 * won't add these. 578 * 579 * @return true if the global section's Configuration and JobXML should be given; false if not 580 */ 581 public boolean supportsConfigurationJobXML() { 582 return false; 583 } 584}