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.io.IOException;
022
023import org.apache.hadoop.conf.Configuration;
024import org.apache.oozie.action.ActionExecutor;
025import org.apache.oozie.action.control.ForkActionExecutor;
026import org.apache.oozie.action.control.StartActionExecutor;
027import org.apache.oozie.action.oozie.SubWorkflowActionExecutor;
028import org.apache.oozie.client.WorkflowAction;
029import org.apache.oozie.client.WorkflowJob;
030import org.apache.oozie.client.SLAEvent.SlaAppType;
031import org.apache.oozie.client.SLAEvent.Status;
032import org.apache.oozie.client.rest.JsonBean;
033import org.apache.oozie.SLAEventBean;
034import org.apache.oozie.WorkflowActionBean;
035import org.apache.oozie.WorkflowJobBean;
036import org.apache.oozie.ErrorCode;
037import org.apache.oozie.XException;
038import org.apache.oozie.command.CommandException;
039import org.apache.oozie.command.PreconditionException;
040import org.apache.oozie.command.wf.ActionXCommand.ActionExecutorContext;
041import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
042import org.apache.oozie.executor.jpa.BatchQueryExecutor;
043import org.apache.oozie.executor.jpa.JPAExecutorException;
044import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
045import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
046import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
047import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
048import org.apache.oozie.service.ActionService;
049import org.apache.oozie.service.ELService;
050import org.apache.oozie.service.EventHandlerService;
051import org.apache.oozie.service.JPAService;
052import org.apache.oozie.service.Services;
053import org.apache.oozie.service.UUIDService;
054import org.apache.oozie.service.WorkflowStoreService;
055import org.apache.oozie.workflow.WorkflowException;
056import org.apache.oozie.workflow.WorkflowInstance;
057import org.apache.oozie.workflow.lite.KillNodeDef;
058import org.apache.oozie.workflow.lite.NodeDef;
059import org.apache.oozie.util.ELEvaluator;
060import org.apache.oozie.util.InstrumentUtils;
061import org.apache.oozie.util.LogUtils;
062import org.apache.oozie.util.XConfiguration;
063import org.apache.oozie.util.ParamChecker;
064import org.apache.oozie.util.XmlUtils;
065import org.apache.oozie.util.db.SLADbXOperations;
066import org.jdom.Element;
067
068import java.io.StringReader;
069import java.util.ArrayList;
070import java.util.Date;
071import java.util.List;
072import java.util.Map;
073
074import org.apache.oozie.client.OozieClient;
075
076@SuppressWarnings("deprecation")
077public class SignalXCommand extends WorkflowXCommand<Void> {
078
079    private JPAService jpaService = null;
080    private String jobId;
081    private String actionId;
082    private WorkflowJobBean wfJob;
083    private WorkflowActionBean wfAction;
084    private List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
085    private List<JsonBean> insertList = new ArrayList<JsonBean>();
086    private boolean generateEvent = false;
087    private String wfJobErrorCode;
088    private String wfJobErrorMsg;
089
090    public SignalXCommand(String name, int priority, String jobId) {
091        super(name, name, priority);
092        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
093    }
094
095    public SignalXCommand(String jobId, String actionId) {
096        this("signal", 1, jobId);
097        this.actionId = ParamChecker.notEmpty(actionId, "actionId");
098    }
099
100    @Override
101    protected void setLogInfo() {
102        if (jobId != null) {
103            LogUtils.setLogInfo(jobId);
104        }
105        else if (actionId !=null) {
106            LogUtils.setLogInfo(actionId);
107        }
108    }
109
110    @Override
111    protected boolean isLockRequired() {
112        return true;
113    }
114
115    @Override
116    public String getEntityKey() {
117        return this.jobId;
118    }
119
120    @Override
121    public String getKey() {
122        return getName() + "_" + jobId + "_" + actionId;
123    }
124
125    @Override
126    protected void loadState() throws CommandException {
127        try {
128            jpaService = Services.get().get(JPAService.class);
129            if (jpaService != null) {
130                this.wfJob = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW, jobId);
131                LogUtils.setLogInfo(wfJob);
132                if (actionId != null) {
133                    this.wfAction = WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION_SIGNAL, actionId);
134                    LogUtils.setLogInfo(wfAction);
135                }
136            }
137            else {
138                throw new CommandException(ErrorCode.E0610);
139            }
140        }
141        catch (XException ex) {
142            throw new CommandException(ex);
143        }
144    }
145
146    @Override
147    protected void verifyPrecondition() throws CommandException, PreconditionException {
148        if ((wfAction == null) || (wfAction.isComplete() && wfAction.isPending())) {
149            if (wfJob.getStatus() != WorkflowJob.Status.RUNNING && wfJob.getStatus() != WorkflowJob.Status.PREP) {
150                throw new PreconditionException(ErrorCode.E0813, wfJob.getStatusStr());
151            }
152        }
153        else {
154            throw new PreconditionException(ErrorCode.E0814, actionId, wfAction.getStatusStr(), wfAction.isPending());
155        }
156    }
157
158    @Override
159    protected Void execute() throws CommandException {
160
161        LOG.debug("STARTED SignalCommand for jobid=" + jobId + ", actionId=" + actionId);
162        WorkflowInstance workflowInstance = wfJob.getWorkflowInstance();
163        workflowInstance.setTransientVar(WorkflowStoreService.WORKFLOW_BEAN, wfJob);
164        WorkflowJob.Status prevStatus = wfJob.getStatus();
165        boolean completed = false, skipAction = false;
166        WorkflowActionBean syncAction = null;
167
168        if (wfAction == null) {
169            if (wfJob.getStatus() == WorkflowJob.Status.PREP) {
170                try {
171                    completed = workflowInstance.start();
172                }
173                catch (WorkflowException e) {
174                    throw new CommandException(e);
175                }
176                wfJob.setStatus(WorkflowJob.Status.RUNNING);
177                wfJob.setStartTime(new Date());
178                wfJob.setWorkflowInstance(workflowInstance);
179                generateEvent = true;
180                // 1. Add SLA status event for WF-JOB with status STARTED
181                SLAEventBean slaEvent = SLADbXOperations.createStatusEvent(wfJob.getSlaXml(), jobId, Status.STARTED,
182                        SlaAppType.WORKFLOW_JOB);
183                if (slaEvent != null) {
184                    insertList.add(slaEvent);
185                }
186                // 2. Add SLA registration events for all WF_ACTIONS
187                createSLARegistrationForAllActions(workflowInstance.getApp().getDefinition(), wfJob.getUser(),
188                        wfJob.getGroup(), wfJob.getConf());
189                queue(new WorkflowNotificationXCommand(wfJob));
190            }
191            else {
192                throw new CommandException(ErrorCode.E0801, wfJob.getId());
193            }
194        }
195        else {
196            WorkflowInstance.Status initialStatus = workflowInstance.getStatus();
197            String skipVar = workflowInstance.getVar(wfAction.getName() + WorkflowInstance.NODE_VAR_SEPARATOR
198                    + ReRunXCommand.TO_SKIP);
199            if (skipVar != null) {
200                skipAction = skipVar.equals("true");
201            }
202            try {
203                completed = workflowInstance.signal(wfAction.getExecutionPath(), wfAction.getSignalValue());
204            }
205            catch (WorkflowException e) {
206               LOG.error("Workflow action failed : " + e.getMessage(), e);
207                wfJob.setStatus(WorkflowJob.Status.valueOf(workflowInstance.getStatus().toString()));
208                completed = true;
209            }
210            wfJob.setWorkflowInstance(workflowInstance);
211            wfAction.resetPending();
212            if (!skipAction) {
213                wfAction.setTransition(workflowInstance.getTransition(wfAction.getName()));
214                queue(new WorkflowNotificationXCommand(wfJob, wfAction));
215            }
216            updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_PENDING_TRANS,
217                    wfAction));
218            WorkflowInstance.Status endStatus = workflowInstance.getStatus();
219            if (endStatus != initialStatus) {
220                generateEvent = true;
221            }
222        }
223
224        if (completed) {
225            try {
226                for (String actionToKillId : WorkflowStoreService.getActionsToKill(workflowInstance)) {
227                    WorkflowActionBean actionToKill;
228
229                    actionToKill = WorkflowActionQueryExecutor.getInstance().get(
230                            WorkflowActionQuery.GET_ACTION_ID_TYPE_LASTCHECK, actionToKillId);
231
232                    actionToKill.setPending();
233                    actionToKill.setStatus(WorkflowActionBean.Status.KILLED);
234                    updateList.add(new UpdateEntry<WorkflowActionQuery>(
235                            WorkflowActionQuery.UPDATE_ACTION_STATUS_PENDING, actionToKill));
236                    queue(new ActionKillXCommand(actionToKill.getId(), actionToKill.getType()));
237                }
238
239                for (String actionToFailId : WorkflowStoreService.getActionsToFail(workflowInstance)) {
240                    WorkflowActionBean actionToFail = WorkflowActionQueryExecutor.getInstance().get(
241                            WorkflowActionQuery.GET_ACTION_FAIL, actionToFailId);
242                    actionToFail.resetPending();
243                    actionToFail.setStatus(WorkflowActionBean.Status.FAILED);
244                    if (wfJobErrorCode != null) {
245                        wfJobErrorCode = actionToFail.getErrorCode();
246                        wfJobErrorMsg = actionToFail.getErrorMessage();
247                    }
248                    queue(new WorkflowNotificationXCommand(wfJob, actionToFail));
249                    SLAEventBean slaEvent = SLADbXOperations.createStatusEvent(wfAction.getSlaXml(), wfAction.getId(),
250                            Status.FAILED, SlaAppType.WORKFLOW_ACTION);
251                    if (slaEvent != null) {
252                        insertList.add(slaEvent);
253                    }
254                    updateList.add(new UpdateEntry<WorkflowActionQuery>(
255                            WorkflowActionQuery.UPDATE_ACTION_STATUS_PENDING, actionToFail));
256                }
257            }
258            catch (JPAExecutorException je) {
259                throw new CommandException(je);
260            }
261
262            wfJob.setStatus(WorkflowJob.Status.valueOf(workflowInstance.getStatus().toString()));
263            wfJob.setEndTime(new Date());
264            wfJob.setWorkflowInstance(workflowInstance);
265            Status slaStatus = Status.SUCCEEDED;
266            switch (wfJob.getStatus()) {
267                case SUCCEEDED:
268                    slaStatus = Status.SUCCEEDED;
269                    break;
270                case KILLED:
271                    slaStatus = Status.KILLED;
272                    break;
273                case FAILED:
274                    slaStatus = Status.FAILED;
275                    break;
276                default: // TODO SUSPENDED
277                    break;
278            }
279            SLAEventBean slaEvent = SLADbXOperations.createStatusEvent(wfJob.getSlaXml(), jobId, slaStatus,
280                    SlaAppType.WORKFLOW_JOB);
281            if (slaEvent != null) {
282                insertList.add(slaEvent);
283            }
284            queue(new WorkflowNotificationXCommand(wfJob));
285            if (wfJob.getStatus() == WorkflowJob.Status.SUCCEEDED) {
286                InstrumentUtils.incrJobCounter(INSTR_SUCCEEDED_JOBS_COUNTER_NAME, 1, getInstrumentation());
287            }
288
289            // output message for Kill node
290            if (wfAction != null) { // wfAction could be a no-op job
291                NodeDef nodeDef = workflowInstance.getNodeDef(wfAction.getExecutionPath());
292                if (nodeDef != null && nodeDef instanceof KillNodeDef) {
293                    boolean isRetry = false;
294                    boolean isUserRetry = false;
295                    ActionExecutorContext context = new ActionXCommand.ActionExecutorContext(wfJob, wfAction, isRetry,
296                            isUserRetry);
297                    InstrumentUtils.incrJobCounter(INSTR_KILLED_JOBS_COUNTER_NAME, 1, getInstrumentation());
298                    try {
299                        String tmpNodeConf = nodeDef.getConf();
300                        String message = context.getELEvaluator().evaluate(tmpNodeConf, String.class);
301                        LOG.debug(
302                                "Try to resolve KillNode message for jobid [{0}], actionId [{1}], before resolve [{2}], "
303                                        + "after resolve [{3}]", jobId, actionId, tmpNodeConf, message);
304                        if (wfAction.getErrorCode() != null) {
305                            wfAction.setErrorInfo(wfAction.getErrorCode(), message);
306                        }
307                        else {
308                            wfAction.setErrorInfo(ErrorCode.E0729.toString(), message);
309                        }
310                    }
311                    catch (Exception ex) {
312                        LOG.warn("Exception in SignalXCommand when processing Kill node message: {0}", ex.getMessage(), ex);
313                        wfAction.setErrorInfo(ErrorCode.E0756.toString(), ErrorCode.E0756.format(ex.getMessage()));
314                        wfAction.setStatus(WorkflowAction.Status.ERROR);
315                    }
316                    updateList.add(new UpdateEntry<WorkflowActionQuery>(
317                            WorkflowActionQuery.UPDATE_ACTION_PENDING_TRANS_ERROR, wfAction));
318                }
319            }
320
321        }
322        else {
323            for (WorkflowActionBean newAction : WorkflowStoreService.getActionsToStart(workflowInstance)) {
324                boolean isOldWFAction = false;
325
326                // In case of subworkflow rerun when failed option have been provided, rerun command do not delete
327                // old action. To avoid twice entry for same action, Checking in Db if the workflow action already exist.
328                if(SubWorkflowActionExecutor.ACTION_TYPE.equals(newAction.getType())) {
329                    try {
330                        WorkflowActionBean oldAction = WorkflowActionQueryExecutor.getInstance()
331                                .get(WorkflowActionQuery.GET_ACTION_CHECK,
332                                        newAction.getId());
333                        newAction.setExternalId(oldAction.getExternalId());
334                        newAction.setCreatedTime(oldAction.getCreatedTime());
335                        isOldWFAction = true;
336                    } catch (JPAExecutorException e) {
337                        if(e.getErrorCode() != ErrorCode.E0605) {
338                            throw new CommandException(e);
339                        }
340                    }
341                }
342
343                String skipVar = workflowInstance.getVar(newAction.getName() + WorkflowInstance.NODE_VAR_SEPARATOR
344                        + ReRunXCommand.TO_SKIP);
345                boolean skipNewAction = false, suspendNewAction = false;
346                if (skipVar != null) {
347                    skipNewAction = skipVar.equals("true");
348                }
349
350                if (skipNewAction) {
351                    WorkflowActionBean oldAction = new WorkflowActionBean();
352                    oldAction.setId(newAction.getId());
353                    oldAction.setPending();
354                    oldAction.setExecutionPath(newAction.getExecutionPath());
355                    updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_PENDING,
356                            oldAction));
357                    queue(new SignalXCommand(jobId, oldAction.getId()));
358                }
359                else {
360                    if(!skipAction) {
361                        try {
362                            // Make sure that transition node for a forked action
363                            // is inserted only once
364                            WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION_ID_TYPE_LASTCHECK,
365                                    newAction.getId());
366
367                            continue;
368                        } catch (JPAExecutorException jee) {
369                        }
370                    }
371                    suspendNewAction = checkForSuspendNode(newAction);
372                    newAction.setPending();
373                    String actionSlaXml = getActionSLAXml(newAction.getName(), workflowInstance.getApp()
374                            .getDefinition(), wfJob.getConf());
375                    newAction.setSlaXml(actionSlaXml);
376                    if(!isOldWFAction) {
377                        newAction.setCreatedTime(new Date());
378                        insertList.add(newAction);
379                    } else {
380                        updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_START,
381                                newAction));
382                    }
383                    LOG.debug("SignalXCommand: Name: " + newAction.getName() + ", Id: " + newAction.getId()
384                            + ", Authcode:" + newAction.getCred());
385                    if (wfAction != null) { // null during wf job submit
386                        ActionService as = Services.get().get(ActionService.class);
387                        ActionExecutor current = as.getExecutor(wfAction.getType());
388                        LOG.trace("Current Action Type:" + current.getClass());
389                        if (!suspendNewAction) {
390                            if (!(current instanceof ForkActionExecutor) && !(current instanceof StartActionExecutor)) {
391                                // Excluding :start: here from executing first action synchronously since it
392                                // blocks the consumer thread till the action is submitted to Hadoop,
393                                // in turn reducing the number of new submissions the threads can accept.
394                                // Would also be susceptible to longer delays in case Hadoop cluster is busy.
395                                syncAction = newAction;
396                            }
397                            else {
398                                queue(new ActionStartXCommand(newAction.getId(), newAction.getType()));
399                            }
400                        }
401                    }
402                    else {
403                        syncAction = newAction; // first action after wf submit should always be sync
404                    }
405                }
406            }
407        }
408
409        try {
410            wfJob.setLastModifiedTime(new Date());
411            updateList.add(new UpdateEntry<WorkflowJobQuery>(
412                    WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_INSTANCE_MOD_START_END, wfJob));
413            // call JPAExecutor to do the bulk writes
414            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
415            if (prevStatus != wfJob.getStatus()) {
416                LOG.debug("Updated the workflow status to " + wfJob.getId() + "  status =" + wfJob.getStatusStr());
417            }
418            if (generateEvent && EventHandlerService.isEnabled()) {
419                generateEvent(wfJob, wfJobErrorCode, wfJobErrorMsg);
420            }
421        }
422        catch (JPAExecutorException je) {
423            throw new CommandException(je);
424        }
425        // Changing to synchronous call from asynchronous queuing to prevent
426        // undue delay from between end of previous and start of next action
427        if (wfJob.getStatus() != WorkflowJob.Status.RUNNING
428                && wfJob.getStatus() != WorkflowJob.Status.SUSPENDED) {
429            // only for asynchronous actions, parent coord action's external id will
430            // persisted and following update will succeed.
431            updateParentIfNecessary(wfJob);
432            new WfEndXCommand(wfJob).call(); // To delete the WF temp dir
433        }
434        else if (syncAction != null) {
435            new ActionStartXCommand(wfJob, syncAction.getId(), syncAction.getType()).call(getEntityKey());
436        }
437        LOG.debug("ENDED SignalCommand for jobid=" + jobId + ", actionId=" + actionId);
438        return null;
439    }
440
441    public static ELEvaluator createELEvaluatorForGroup(Configuration conf, String group) {
442        ELEvaluator eval = Services.get().get(ELService.class).createEvaluator(group);
443        for (Map.Entry<String, String> entry : conf) {
444            eval.setVariable(entry.getKey(), entry.getValue());
445        }
446        return eval;
447    }
448
449    @SuppressWarnings("unchecked")
450    private String getActionSLAXml(String actionName, String wfXml, String wfConf) throws CommandException {
451        String slaXml = null;
452        try {
453            Element eWfJob = XmlUtils.parseXml(wfXml);
454            for (Element action : (List<Element>) eWfJob.getChildren("action", eWfJob.getNamespace())) {
455                if (action.getAttributeValue("name").equals(actionName) == false) {
456                    continue;
457                }
458                Element eSla = XmlUtils.getSLAElement(action);
459                if (eSla != null) {
460                    slaXml = XmlUtils.prettyPrint(eSla).toString();
461                    break;
462                }
463            }
464        }
465        catch (Exception e) {
466            throw new CommandException(ErrorCode.E1004, e.getMessage(), e);
467        }
468        return slaXml;
469    }
470
471    private String resolveSla(Element eSla, Configuration conf) throws CommandException {
472        String slaXml = null;
473        try {
474            ELEvaluator evalSla = SubmitXCommand.createELEvaluatorForGroup(conf, "wf-sla-submit");
475            slaXml = SubmitXCommand.resolveSla(eSla, evalSla);
476        }
477        catch (Exception e) {
478            throw new CommandException(ErrorCode.E1004, e.getMessage(), e);
479        }
480        return slaXml;
481    }
482
483    @SuppressWarnings("unchecked")
484    private void createSLARegistrationForAllActions(String wfXml, String user, String group, String strConf)
485            throws CommandException {
486        try {
487            Element eWfJob = XmlUtils.parseXml(wfXml);
488            Configuration conf = new XConfiguration(new StringReader(strConf));
489            for (Element action : (List<Element>) eWfJob.getChildren("action", eWfJob.getNamespace())) {
490                Element eSla = XmlUtils.getSLAElement(action);
491                if (eSla != null) {
492                    String slaXml = resolveSla(eSla, conf);
493                    eSla = XmlUtils.parseXml(slaXml);
494                    String actionId = Services.get().get(UUIDService.class)
495                            .generateChildId(jobId, action.getAttributeValue("name") + "");
496                    SLAEventBean slaEvent = SLADbXOperations.createSlaRegistrationEvent(eSla, actionId,
497                            SlaAppType.WORKFLOW_ACTION, user, group);
498                    if (slaEvent != null) {
499                        insertList.add(slaEvent);
500                    }
501                }
502            }
503        }
504        catch (Exception e) {
505            throw new CommandException(ErrorCode.E1007, "workflow:Actions " + jobId, e.getMessage(), e);
506        }
507
508    }
509
510    private boolean checkForSuspendNode(WorkflowActionBean newAction) {
511        boolean suspendNewAction = false;
512        try {
513            XConfiguration wfjobConf = new XConfiguration(new StringReader(wfJob.getConf()));
514            String[] values = wfjobConf.getTrimmedStrings(OozieClient.OOZIE_SUSPEND_ON_NODES);
515            if (values != null) {
516                if (values.length == 1 && values[0].equals("*")) {
517                    LOG.info("Reached suspend node at [{0}], suspending workflow [{1}]", newAction.getName(),
518                            wfJob.getId());
519                    queue(new SuspendXCommand(jobId));
520                    suspendNewAction = true;
521                }
522                else {
523                    for (String suspendPoint : values) {
524                        if (suspendPoint.equals(newAction.getName())) {
525                            LOG.info("Reached suspend node at [{0}], suspending workflow [{1}]", newAction.getName(),
526                                    wfJob.getId());
527                            queue(new SuspendXCommand(jobId));
528                            suspendNewAction = true;
529                            break;
530                        }
531                    }
532                }
533            }
534        }
535        catch (IOException ex) {
536            LOG.warn("Error reading " + OozieClient.OOZIE_SUSPEND_ON_NODES + ", ignoring [{0}]", ex.getMessage());
537        }
538        return suspendNewAction;
539    }
540
541}