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.command.wf;
020
021import java.util.ArrayList;
022import java.util.Date;
023import java.util.List;
024
025import javax.servlet.jsp.el.ELException;
026
027import org.apache.hadoop.conf.Configuration;
028import org.apache.oozie.ErrorCode;
029import org.apache.oozie.FaultInjection;
030import org.apache.oozie.SLAEventBean;
031import org.apache.oozie.WorkflowActionBean;
032import org.apache.oozie.WorkflowJobBean;
033import org.apache.oozie.XException;
034import org.apache.oozie.action.ActionExecutor;
035import org.apache.oozie.action.ActionExecutorException;
036import org.apache.oozie.action.control.ControlNodeActionExecutor;
037import org.apache.oozie.client.OozieClient;
038import org.apache.oozie.client.WorkflowAction;
039import org.apache.oozie.client.WorkflowJob;
040import org.apache.oozie.client.SLAEvent.SlaAppType;
041import org.apache.oozie.client.SLAEvent.Status;
042import org.apache.oozie.client.rest.JsonBean;
043import org.apache.oozie.command.CommandException;
044import org.apache.oozie.command.PreconditionException;
045import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
046import org.apache.oozie.executor.jpa.BatchQueryExecutor;
047import org.apache.oozie.executor.jpa.JPAExecutorException;
048import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
049import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
050import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
051import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
052import org.apache.oozie.service.ActionService;
053import org.apache.oozie.service.EventHandlerService;
054import org.apache.oozie.service.JPAService;
055import org.apache.oozie.service.Services;
056import org.apache.oozie.service.UUIDService;
057import org.apache.oozie.util.ELEvaluationException;
058import org.apache.oozie.util.Instrumentation;
059import org.apache.oozie.util.LogUtils;
060import org.apache.oozie.util.XLog;
061import org.apache.oozie.util.XmlUtils;
062import org.apache.oozie.util.db.SLADbXOperations;
063
064@SuppressWarnings("deprecation")
065public class ActionStartXCommand extends ActionXCommand<Void> {
066    public static final String EL_ERROR = "EL_ERROR";
067    public static final String EL_EVAL_ERROR = "EL_EVAL_ERROR";
068    public static final String COULD_NOT_START = "COULD_NOT_START";
069    public static final String START_DATA_MISSING = "START_DATA_MISSING";
070    public static final String EXEC_DATA_MISSING = "EXEC_DATA_MISSING";
071    public static final String OOZIE_ACTION_YARN_TAG = "oozie.action.yarn.tag";
072
073    private String jobId = null;
074    private String actionId = null;
075    private WorkflowJobBean wfJob = null;
076    private WorkflowActionBean wfAction = null;
077    private JPAService jpaService = null;
078    private ActionExecutor executor = null;
079    private List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
080    private List<JsonBean> insertList = new ArrayList<JsonBean>();
081
082    public ActionStartXCommand(String actionId, String type) {
083        super("action.start", type, 0);
084        this.actionId = actionId;
085        this.jobId = Services.get().get(UUIDService.class).getId(actionId);
086    }
087
088    public ActionStartXCommand(WorkflowJobBean job, String actionId, String type) {
089        super("action.start", type, 0);
090        this.actionId = actionId;
091        this.wfJob = job;
092        this.jobId = wfJob.getId();
093    }
094
095    @Override
096    protected void setLogInfo() {
097        LogUtils.setLogInfo(actionId);
098    }
099
100    @Override
101    protected boolean isLockRequired() {
102        return true;
103    }
104
105    @Override
106    public String getEntityKey() {
107        return this.jobId;
108    }
109
110    @Override
111    protected void loadState() throws CommandException {
112        try {
113            jpaService = Services.get().get(JPAService.class);
114            if (jpaService != null) {
115                if (wfJob == null) {
116                    this.wfJob = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW, jobId);
117                }
118                this.wfAction = WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION, actionId);
119                LogUtils.setLogInfo( wfJob);
120                LogUtils.setLogInfo(wfAction);
121            }
122            else {
123                throw new CommandException(ErrorCode.E0610);
124            }
125        }
126        catch (XException ex) {
127            throw new CommandException(ex);
128        }
129    }
130
131    @Override
132    protected void verifyPrecondition() throws CommandException, PreconditionException {
133        if (wfJob == null) {
134            throw new PreconditionException(ErrorCode.E0604, jobId);
135        }
136        if (wfAction == null) {
137            throw new PreconditionException(ErrorCode.E0605, actionId);
138        }
139        if (wfAction.isPending()
140                && (wfAction.getStatus() == WorkflowActionBean.Status.PREP
141                        || wfAction.getStatus() == WorkflowActionBean.Status.START_RETRY
142                        || wfAction.getStatus() == WorkflowActionBean.Status.START_MANUAL
143                        || wfAction.getStatus() == WorkflowActionBean.Status.USER_RETRY
144                        )) {
145            if (wfJob.getStatus() != WorkflowJob.Status.RUNNING) {
146                throw new PreconditionException(ErrorCode.E0810, WorkflowJob.Status.RUNNING.toString());
147            }
148        }
149        else {
150            throw new PreconditionException(ErrorCode.E0816, wfAction.isPending(), wfAction.getStatusStr());
151        }
152
153        executor = Services.get().get(ActionService.class).getExecutor(wfAction.getType());
154        if (executor == null) {
155            throw new CommandException(ErrorCode.E0802, wfAction.getType());
156        }
157    }
158
159    @Override
160    protected Void execute() throws CommandException {
161
162        LOG.debug("STARTED ActionStartXCommand for wf actionId=" + actionId);
163        Configuration conf = wfJob.getWorkflowInstance().getConf();
164
165        int maxRetries = 0;
166        long retryInterval = 0;
167        boolean execSynchronous = false;
168
169        if (!(executor instanceof ControlNodeActionExecutor)) {
170            maxRetries = conf.getInt(OozieClient.ACTION_MAX_RETRIES, executor.getMaxRetries());
171            retryInterval = conf.getLong(OozieClient.ACTION_RETRY_INTERVAL, executor.getRetryInterval());
172        }
173
174        executor.setMaxRetries(maxRetries);
175        executor.setRetryInterval(retryInterval);
176
177        ActionExecutorContext context = null;
178        try {
179            boolean isRetry = false;
180            if (wfAction.getStatus() == WorkflowActionBean.Status.START_RETRY
181                    || wfAction.getStatus() == WorkflowActionBean.Status.START_MANUAL) {
182                isRetry = true;
183                prepareForRetry(wfAction);
184            }
185            boolean isUserRetry = false;
186            if (wfAction.getStatus() == WorkflowActionBean.Status.USER_RETRY) {
187                isUserRetry = true;
188                prepareForRetry(wfAction);
189            }
190            context = new ActionXCommand.ActionExecutorContext(wfJob, wfAction, isRetry, isUserRetry);
191            boolean caught = false;
192            try {
193                if (!(executor instanceof ControlNodeActionExecutor)) {
194                    String tmpActionConf = XmlUtils.removeComments(wfAction.getConf());
195                    String actionConf = context.getELEvaluator().evaluate(tmpActionConf, String.class);
196                    wfAction.setConf(actionConf);
197                    LOG.debug("Start, name [{0}] type [{1}] configuration{E}{E}{2}{E}", wfAction.getName(), wfAction
198                            .getType(), actionConf);
199                }
200            }
201            catch (ELEvaluationException ex) {
202                caught = true;
203                throw new ActionExecutorException(ActionExecutorException.ErrorType.TRANSIENT, EL_EVAL_ERROR, ex
204                        .getMessage(), ex);
205            }
206            catch (ELException ex) {
207                caught = true;
208                context.setErrorInfo(EL_ERROR, ex.getMessage());
209                LOG.warn("ELException in ActionStartXCommand ", ex.getMessage(), ex);
210                handleError(context, wfJob, wfAction);
211            }
212            catch (org.jdom.JDOMException je) {
213                caught = true;
214                context.setErrorInfo("ParsingError", je.getMessage());
215                LOG.warn("JDOMException in ActionStartXCommand ", je.getMessage(), je);
216                handleError(context, wfJob, wfAction);
217            }
218            catch (Exception ex) {
219                caught = true;
220                context.setErrorInfo(EL_ERROR, ex.getMessage());
221                LOG.warn("Exception in ActionStartXCommand ", ex.getMessage(), ex);
222                handleError(context, wfJob, wfAction);
223            }
224            if(!caught) {
225                wfAction.setErrorInfo(null, null);
226                incrActionCounter(wfAction.getType(), 1);
227
228                LOG.info("Start action [{0}] with user-retry state : userRetryCount [{1}], userRetryMax [{2}], userRetryInterval [{3}]",
229                                wfAction.getId(), wfAction.getUserRetryCount(), wfAction.getUserRetryMax(), wfAction
230                                        .getUserRetryInterval());
231
232                Instrumentation.Cron cron = new Instrumentation.Cron();
233                cron.start();
234                context.setStartTime();
235                /*
236                Creating and forwarding the tag, It will be useful during repeat attempts of Launcher, to ensure only
237                one child job is running. Tag is formed as follows:
238                For workflow job, tag = action-id
239                For Coord job, tag = coord-action-id@action-name (if not part of sub flow), else
240                coord-action-id@subflow-action-name@action-name.
241                 */
242                if (conf.get(OOZIE_ACTION_YARN_TAG) != null) {
243                    context.setVar(OOZIE_ACTION_YARN_TAG, conf.get(OOZIE_ACTION_YARN_TAG) + "@" + wfAction.getName());
244                } else if (wfJob.getParentId() != null) {
245                    context.setVar(OOZIE_ACTION_YARN_TAG, wfJob.getParentId() + "@" + wfAction.getName());
246                } else {
247                    context.setVar(OOZIE_ACTION_YARN_TAG, wfAction.getId());
248                }
249
250                executor.start(context, wfAction);
251                cron.stop();
252                FaultInjection.activate("org.apache.oozie.command.SkipCommitFaultInjection");
253                addActionCron(wfAction.getType(), cron);
254
255                wfAction.setRetries(0);
256                if (wfAction.isExecutionComplete()) {
257                    if (!context.isExecuted()) {
258                        LOG.warn(XLog.OPS, "Action Completed, ActionExecutor [{0}] must call setExecutionData()", executor
259                                .getType());
260                        wfAction.setErrorInfo(EXEC_DATA_MISSING,
261                                "Execution Complete, but Execution Data Missing from Action");
262                        failJob(context);
263                    } else {
264                        wfAction.setPending();
265                        if (!(executor instanceof ControlNodeActionExecutor)) {
266                            queue(new ActionEndXCommand(wfAction.getId(), wfAction.getType()));
267                        }
268                        else {
269                            execSynchronous = true;
270                        }
271                    }
272                }
273                else {
274                    if (!context.isStarted()) {
275                        LOG.warn(XLog.OPS, "Action Started, ActionExecutor [{0}] must call setStartData()", executor
276                                .getType());
277                        wfAction.setErrorInfo(START_DATA_MISSING, "Execution Started, but Start Data Missing from Action");
278                        failJob(context);
279                    } else {
280                        queue(new WorkflowNotificationXCommand(wfJob, wfAction));
281                    }
282                }
283
284                LOG.info(XLog.STD, "[***" + wfAction.getId() + "***]" + "Action status=" + wfAction.getStatusStr());
285
286                updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_START, wfAction));
287                wfJob.setLastModifiedTime(new Date());
288                updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MODIFIED, wfJob));
289                // Add SLA status event (STARTED) for WF_ACTION
290                SLAEventBean slaEvent = SLADbXOperations.createStatusEvent(wfAction.getSlaXml(), wfAction.getId(), Status.STARTED,
291                        SlaAppType.WORKFLOW_ACTION);
292                if(slaEvent != null) {
293                    insertList.add(slaEvent);
294                }
295                LOG.info(XLog.STD, "[***" + wfAction.getId() + "***]" + "Action updated in DB!");
296            }
297        }
298        catch (ActionExecutorException ex) {
299            LOG.warn("Error starting action [{0}]. ErrorType [{1}], ErrorCode [{2}], Message [{3}]",
300                    wfAction.getName(), ex.getErrorType(), ex.getErrorCode(), ex.getMessage(), ex);
301            wfAction.setErrorInfo(ex.getErrorCode(), ex.getMessage());
302            switch (ex.getErrorType()) {
303                case TRANSIENT:
304                    if (!handleTransient(context, executor, WorkflowAction.Status.START_RETRY)) {
305                        handleNonTransient(context, executor, WorkflowAction.Status.START_MANUAL);
306                        wfAction.setPendingAge(new Date());
307                        wfAction.setRetries(0);
308                        wfAction.setStartTime(null);
309                    }
310                    break;
311                case NON_TRANSIENT:
312                    handleNonTransient(context, executor, WorkflowAction.Status.START_MANUAL);
313                    break;
314                case ERROR:
315                    handleError(context, executor, WorkflowAction.Status.ERROR.toString(), true,
316                            WorkflowAction.Status.DONE);
317                    break;
318                case FAILED:
319                    try {
320                        failJob(context);
321                        updateParentIfNecessary(wfJob, 3);
322                        new WfEndXCommand(wfJob).call(); // To delete the WF temp dir
323                        SLAEventBean slaEvent1 = SLADbXOperations.createStatusEvent(wfAction.getSlaXml(), wfAction.getId(), Status.FAILED,
324                                SlaAppType.WORKFLOW_ACTION);
325                        if(slaEvent1 != null) {
326                            insertList.add(slaEvent1);
327                        }
328                        SLAEventBean slaEvent2 = SLADbXOperations.createStatusEvent(wfJob.getSlaXml(), wfJob.getId(), Status.FAILED,
329                                SlaAppType.WORKFLOW_JOB);
330                        if(slaEvent2 != null) {
331                            insertList.add(slaEvent2);
332                        }
333                    }
334                    catch (XException x) {
335                        LOG.warn("ActionStartXCommand - case:FAILED ", x.getMessage());
336                    }
337                    break;
338            }
339            updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_START, wfAction));
340            wfJob.setLastModifiedTime(new Date());
341            updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MODIFIED, wfJob));
342        }
343        finally {
344            try {
345                BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
346                if (!(executor instanceof ControlNodeActionExecutor) && EventHandlerService.isEnabled()) {
347                    generateEvent(wfAction, wfJob.getUser());
348                }
349                if (execSynchronous) {
350                    // Changing to synchronous call from asynchronous queuing to prevent
351                    // undue delay from ::start:: to action due to queuing
352                    new ActionEndXCommand(wfAction.getId(), wfAction.getType()).call(getEntityKey());
353                }
354            }
355            catch (JPAExecutorException e) {
356                throw new CommandException(e);
357            }
358        }
359
360        LOG.debug("ENDED ActionStartXCommand for wf actionId=" + actionId + ", jobId=" + jobId);
361
362        return null;
363    }
364
365    private void handleError(ActionExecutorContext context, WorkflowJobBean workflow, WorkflowActionBean action)
366            throws CommandException {
367        failJob(context);
368        updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_START, wfAction));
369        wfJob.setLastModifiedTime(new Date());
370        updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MODIFIED, wfJob));
371        SLAEventBean slaEvent1 = SLADbXOperations.createStatusEvent(action.getSlaXml(), action.getId(),
372                Status.FAILED, SlaAppType.WORKFLOW_ACTION);
373        if(slaEvent1 != null) {
374            insertList.add(slaEvent1);
375        }
376        SLAEventBean slaEvent2 = SLADbXOperations.createStatusEvent(workflow.getSlaXml(), workflow.getId(),
377                Status.FAILED, SlaAppType.WORKFLOW_JOB);
378        if(slaEvent2 != null) {
379            insertList.add(slaEvent2);
380        }
381
382        new WfEndXCommand(wfJob).call(); //To delete the WF temp dir
383        return;
384    }
385
386    /* (non-Javadoc)
387     * @see org.apache.oozie.command.XCommand#getKey()
388     */
389    @Override
390    public String getKey(){
391        return getName() + "_" + actionId;
392    }
393
394    private void prepareForRetry(WorkflowActionBean wfAction) {
395        if (wfAction.getType().equals("map-reduce")) {
396            // need to delete child job id of original run
397            wfAction.setExternalChildIDs("");
398        }
399    }
400
401}