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 &lt;name-node&gt; and &lt;job-tracker&gt;, 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     * &lt;configuration&gt; and &lt;job-xml&gt; 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}