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 package org.apache.oozie.command.wf; 019 020 import org.apache.oozie.client.WorkflowJob; 021 import org.apache.oozie.client.SLAEvent.SlaAppType; 022 import org.apache.oozie.client.SLAEvent.Status; 023 import org.apache.oozie.ErrorCode; 024 import org.apache.oozie.WorkflowActionBean; 025 import org.apache.oozie.WorkflowJobBean; 026 import org.apache.oozie.XException; 027 import org.apache.oozie.command.CommandException; 028 import org.apache.oozie.command.PreconditionException; 029 import org.apache.oozie.command.coord.CoordActionUpdateXCommand; 030 import org.apache.oozie.executor.jpa.JPAExecutorException; 031 import org.apache.oozie.executor.jpa.WorkflowActionUpdateJPAExecutor; 032 import org.apache.oozie.executor.jpa.WorkflowActionsGetForJobJPAExecutor; 033 import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor; 034 import org.apache.oozie.executor.jpa.WorkflowJobUpdateJPAExecutor; 035 import org.apache.oozie.service.JPAService; 036 import org.apache.oozie.service.Services; 037 import org.apache.oozie.workflow.WorkflowException; 038 import org.apache.oozie.workflow.WorkflowInstance; 039 import org.apache.oozie.workflow.lite.LiteWorkflowInstance; 040 import org.apache.oozie.util.InstrumentUtils; 041 import org.apache.oozie.util.LogUtils; 042 import org.apache.oozie.util.ParamChecker; 043 import org.apache.oozie.util.db.SLADbXOperations; 044 045 import java.util.Date; 046 import java.util.List; 047 048 /** 049 * Kill workflow job and its workflow instance and queue a {@link WorkflowActionKillXCommand} to kill the workflow 050 * actions. 051 */ 052 public class KillXCommand extends WorkflowXCommand<Void> { 053 054 private String wfId; 055 private WorkflowJobBean wfJob; 056 private List<WorkflowActionBean> actionList; 057 private JPAService jpaService = null; 058 059 public KillXCommand(String wfId) { 060 super("kill", "kill", 1); 061 this.wfId = ParamChecker.notEmpty(wfId, "wfId"); 062 } 063 064 @Override 065 protected boolean isLockRequired() { 066 return true; 067 } 068 069 @Override 070 public String getEntityKey() { 071 return this.wfId; 072 } 073 074 @Override 075 protected void loadState() throws CommandException { 076 try { 077 jpaService = Services.get().get(JPAService.class); 078 if (jpaService != null) { 079 this.wfJob = jpaService.execute(new WorkflowJobGetJPAExecutor(wfId)); 080 this.actionList = jpaService.execute(new WorkflowActionsGetForJobJPAExecutor(wfId)); 081 LogUtils.setLogInfo(wfJob, logInfo); 082 } 083 else { 084 throw new CommandException(ErrorCode.E0610); 085 } 086 } 087 catch (XException ex) { 088 throw new CommandException(ex); 089 } 090 } 091 092 @Override 093 protected void verifyPrecondition() throws CommandException, PreconditionException { 094 if (wfJob.getStatus() != WorkflowJob.Status.PREP && wfJob.getStatus() != WorkflowJob.Status.RUNNING 095 && wfJob.getStatus() != WorkflowJob.Status.SUSPENDED && wfJob.getStatus() != WorkflowJob.Status.FAILED) { 096 throw new PreconditionException(ErrorCode.E0725, wfJob.getId()); 097 } 098 } 099 100 @Override 101 protected Void execute() throws CommandException { 102 LOG.info("STARTED WorkflowKillXCommand for jobId=" + wfId); 103 104 wfJob.setEndTime(new Date()); 105 106 if (wfJob.getStatus() != WorkflowJob.Status.FAILED) { 107 InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation()); 108 wfJob.setStatus(WorkflowJob.Status.KILLED); 109 SLADbXOperations.writeStausEvent(wfJob.getSlaXml(), wfJob.getId(), Status.KILLED, SlaAppType.WORKFLOW_JOB); 110 try { 111 wfJob.getWorkflowInstance().kill(); 112 } 113 catch (WorkflowException e) { 114 throw new CommandException(ErrorCode.E0725, e.getMessage(), e); 115 } 116 WorkflowInstance wfInstance = wfJob.getWorkflowInstance(); 117 ((LiteWorkflowInstance) wfInstance).setStatus(WorkflowInstance.Status.KILLED); 118 wfJob.setWorkflowInstance(wfInstance); 119 } 120 try { 121 for (WorkflowActionBean action : actionList) { 122 if (action.getStatus() == WorkflowActionBean.Status.RUNNING 123 || action.getStatus() == WorkflowActionBean.Status.DONE) { 124 action.setPending(); 125 action.setStatus(WorkflowActionBean.Status.KILLED); 126 127 jpaService.execute(new WorkflowActionUpdateJPAExecutor(action)); 128 129 queue(new ActionKillXCommand(action.getId(), action.getType())); 130 } 131 if (action.getStatus() == WorkflowActionBean.Status.PREP 132 || action.getStatus() == WorkflowActionBean.Status.START_RETRY 133 || action.getStatus() == WorkflowActionBean.Status.START_MANUAL 134 || action.getStatus() == WorkflowActionBean.Status.END_RETRY 135 || action.getStatus() == WorkflowActionBean.Status.END_MANUAL) { 136 137 action.setStatus(WorkflowActionBean.Status.KILLED); 138 action.resetPending(); 139 SLADbXOperations.writeStausEvent(action.getSlaXml(), action.getId(), Status.KILLED, 140 SlaAppType.WORKFLOW_ACTION); 141 jpaService.execute(new WorkflowActionUpdateJPAExecutor(action)); 142 } 143 } 144 jpaService.execute(new WorkflowJobUpdateJPAExecutor(wfJob)); 145 queue(new NotificationXCommand(wfJob)); 146 } 147 catch (JPAExecutorException je) { 148 throw new CommandException(je); 149 } 150 finally { 151 if(wfJob.getStatus() == WorkflowJob.Status.KILLED) { 152 new WfEndXCommand(wfJob).call(); //To delete the WF temp dir 153 } 154 // update coordinator action 155 new CoordActionUpdateXCommand(wfJob).call(); 156 } 157 158 LOG.info("ENDED WorkflowKillXCommand for jobId=" + wfId); 159 return null; 160 } 161 162 }