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 org.apache.oozie.ErrorCode;
026import org.apache.oozie.SLAEventBean;
027import org.apache.oozie.WorkflowActionBean;
028import org.apache.oozie.WorkflowJobBean;
029import org.apache.oozie.XException;
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.command.CommandException;
034import org.apache.oozie.command.PreconditionException;
035import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
036import org.apache.oozie.executor.jpa.BatchQueryExecutor;
037import org.apache.oozie.executor.jpa.JPAExecutorException;
038import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor;
039import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor;
040import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor.WorkflowActionQuery;
041import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor.WorkflowJobQuery;
042import org.apache.oozie.action.ActionExecutor;
043import org.apache.oozie.action.ActionExecutorException;
044import org.apache.oozie.action.control.ControlNodeActionExecutor;
045import org.apache.oozie.service.ActionService;
046import org.apache.oozie.service.EventHandlerService;
047import org.apache.oozie.service.JPAService;
048import org.apache.oozie.service.UUIDService;
049import org.apache.oozie.service.Services;
050import org.apache.oozie.util.LogUtils;
051import org.apache.oozie.util.Instrumentation;
052import org.apache.oozie.util.db.SLADbXOperations;
053
054/**
055 * Kill workflow action and invoke action executor to kill the underlying context.
056 *
057 */
058@SuppressWarnings("deprecation")
059public class ActionKillXCommand extends ActionXCommand<Void> {
060    private String actionId;
061    private String jobId;
062    private WorkflowJobBean wfJob;
063    private WorkflowActionBean wfAction;
064    private JPAService jpaService = null;
065    private List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
066    private List<JsonBean> insertList = new ArrayList<JsonBean>();
067
068    public ActionKillXCommand(String actionId, String type) {
069        super("action.kill", type, 0);
070        this.actionId = actionId;
071        this.jobId = Services.get().get(UUIDService.class).getId(actionId);
072    }
073
074    public ActionKillXCommand(String actionId) {
075        this(actionId, "action.kill");
076    }
077
078    @Override
079    protected void setLogInfo() {
080        LogUtils.setLogInfo(actionId);
081    }
082
083    @Override
084    protected boolean isLockRequired() {
085        return true;
086    }
087
088    @Override
089    public String getEntityKey() {
090        return this.jobId;
091    }
092
093    @Override
094    public String getKey() {
095        return getName() + "_" + this.actionId;
096    }
097
098    @Override
099    protected void loadState() throws CommandException {
100        try {
101            jpaService = Services.get().get(JPAService.class);
102
103            if (jpaService != null) {
104                this.wfJob = WorkflowJobQueryExecutor.getInstance().get(WorkflowJobQuery.GET_WORKFLOW_ACTION_OP, jobId);
105                this.wfAction = WorkflowActionQueryExecutor.getInstance().get(WorkflowActionQuery.GET_ACTION, actionId);
106                LogUtils.setLogInfo(wfJob);
107                LogUtils.setLogInfo(wfAction);
108            }
109            else {
110                throw new CommandException(ErrorCode.E0610);
111            }
112        }
113        catch (XException ex) {
114            throw new CommandException(ex);
115        }
116    }
117
118    @Override
119    protected void verifyPrecondition() throws CommandException, PreconditionException {
120        if (wfAction.getStatus() != WorkflowActionBean.Status.KILLED) {
121            throw new PreconditionException(ErrorCode.E0726, wfAction.getId());
122        }
123    }
124
125    @Override
126    protected Void execute() throws CommandException {
127        LOG.debug("STARTED WorkflowActionKillXCommand for action " + actionId);
128
129        if (wfAction.isPending()) {
130            ActionExecutor executor = Services.get().get(ActionService.class).getExecutor(wfAction.getType());
131            if (executor != null) {
132                try {
133                    boolean isRetry = false;
134                    boolean isUserRetry = false;
135                    ActionExecutorContext context = new ActionXCommand.ActionExecutorContext(wfJob, wfAction,
136                            isRetry, isUserRetry);
137                    incrActionCounter(wfAction.getType(), 1);
138
139                    Instrumentation.Cron cron = new Instrumentation.Cron();
140                    cron.start();
141                    executor.kill(context, wfAction);
142                    cron.stop();
143                    addActionCron(wfAction.getType(), cron);
144
145                    wfAction.resetPending();
146                    wfAction.setStatus(WorkflowActionBean.Status.KILLED);
147                    wfAction.setEndTime(new Date());
148
149                    updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_END, wfAction));
150                    wfJob.setLastModifiedTime(new Date());
151                    updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_MODTIME, wfJob));
152                    // Add SLA status event (KILLED) for WF_ACTION
153                    SLAEventBean slaEvent = SLADbXOperations.createStatusEvent(wfAction.getSlaXml(), wfAction.getId(), Status.KILLED,
154                            SlaAppType.WORKFLOW_ACTION);
155                    if(slaEvent != null) {
156                        insertList.add(slaEvent);
157                    }
158                    queue(new WorkflowNotificationXCommand(wfJob, wfAction));
159                }
160                catch (ActionExecutorException ex) {
161                    wfAction.resetPending();
162                    wfAction.setStatus(WorkflowActionBean.Status.FAILED);
163                    wfAction.setErrorInfo(ex.getErrorCode().toString(),
164                            "KILL COMMAND FAILED - exception while executing job kill");
165                    wfAction.setEndTime(new Date());
166
167                    wfJob.setStatus(WorkflowJobBean.Status.KILLED);
168                    updateList.add(new UpdateEntry<WorkflowActionQuery>(WorkflowActionQuery.UPDATE_ACTION_END, wfAction));
169                    wfJob.setLastModifiedTime(new Date());
170                    updateList.add(new UpdateEntry<WorkflowJobQuery>(WorkflowJobQuery.UPDATE_WORKFLOW_STATUS_MODTIME, wfJob));
171                    // What will happen to WF and COORD_ACTION, NOTIFICATION?
172                    SLAEventBean slaEvent = SLADbXOperations.createStatusEvent(wfAction.getSlaXml(), wfAction.getId(), Status.FAILED,
173                            SlaAppType.WORKFLOW_ACTION);
174                    if(slaEvent != null) {
175                        insertList.add(slaEvent);
176                    }
177                    LOG.warn("Exception while executing kill(). Error Code [{0}], Message[{1}]",
178                            ex.getErrorCode(), ex.getMessage(), ex);
179                }
180                finally {
181                    try {
182                        BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
183                        if (!(executor instanceof ControlNodeActionExecutor) && EventHandlerService.isEnabled()) {
184                            generateEvent(wfAction, wfJob.getUser());
185                        }
186                    }
187                    catch (JPAExecutorException e) {
188                        throw new CommandException(e);
189                    }
190                }
191            }
192        }
193        LOG.debug("ENDED WorkflowActionKillXCommand for action " + actionId);
194        return null;
195    }
196
197}