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 */
018package org.apache.oozie.command.coord;
019
020import java.sql.Timestamp;
021import java.util.ArrayList;
022import java.util.Date;
023import java.util.List;
024
025import org.apache.oozie.CoordinatorActionBean;
026import org.apache.oozie.CoordinatorJobBean;
027import org.apache.oozie.ErrorCode;
028import org.apache.oozie.SLAEventBean;
029import org.apache.oozie.WorkflowJobBean;
030import org.apache.oozie.XException;
031import org.apache.oozie.service.EventHandlerService;
032import org.apache.oozie.service.JPAService;
033import org.apache.oozie.service.Services;
034import org.apache.oozie.util.InstrumentUtils;
035import org.apache.oozie.util.LogUtils;
036import org.apache.oozie.util.ParamChecker;
037import org.apache.oozie.util.db.SLADbOperations;
038import org.apache.oozie.client.CoordinatorAction;
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.BulkUpdateInsertForCoordActionStatusJPAExecutor;
046import org.apache.oozie.executor.jpa.CoordActionGetForCheckJPAExecutor;
047import org.apache.oozie.executor.jpa.CoordinatorJobGetForUserAppnameJPAExecutor;
048import org.apache.oozie.executor.jpa.WorkflowActionGetForSLAJPAExecutor;
049import org.apache.oozie.executor.jpa.WorkflowJobGetForSLAJPAExecutor;
050import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor;
051
052/**
053 * The command checks workflow status for coordinator action.
054 */
055@SuppressWarnings("deprecation")
056public class CoordActionCheckXCommand extends CoordinatorXCommand<Void> {
057    private String actionId;
058    private int actionCheckDelay;
059    private CoordinatorActionBean coordAction = null;
060    private CoordinatorJobBean coordJob;
061    private WorkflowJobBean workflowJob;
062    private JPAService jpaService = null;
063    private List<JsonBean> updateList = new ArrayList<JsonBean>();
064    private List<JsonBean> insertList = new ArrayList<JsonBean>();
065
066    public CoordActionCheckXCommand(String actionId, int actionCheckDelay) {
067        super("coord_action_check", "coord_action_check", 0);
068        this.actionId = ParamChecker.notEmpty(actionId, "actionId");
069        this.actionCheckDelay = actionCheckDelay;
070    }
071
072    /* (non-Javadoc)
073     * @see org.apache.oozie.command.XCommand#execute()
074     */
075    @Override
076    protected Void execute() throws CommandException {
077        try {
078            InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
079            WorkflowJobBean wf = jpaService.execute(new WorkflowJobGetJPAExecutor(coordAction.getExternalId()));
080            Status slaStatus = null;
081            CoordinatorAction.Status initialStatus = coordAction.getStatus();
082
083            if (wf.getStatus() == WorkflowJob.Status.SUCCEEDED) {
084                coordAction.setStatus(CoordinatorAction.Status.SUCCEEDED);
085                // set pending to false as the status is SUCCEEDED
086                coordAction.setPending(0);
087                slaStatus = Status.SUCCEEDED;
088            }
089            else {
090                if (wf.getStatus() == WorkflowJob.Status.FAILED) {
091                    coordAction.setStatus(CoordinatorAction.Status.FAILED);
092                    slaStatus = Status.FAILED;
093                    // set pending to false as the status is FAILED
094                    coordAction.setPending(0);
095                }
096                else {
097                    if (wf.getStatus() == WorkflowJob.Status.KILLED) {
098                        coordAction.setStatus(CoordinatorAction.Status.KILLED);
099                        slaStatus = Status.KILLED;
100                        // set pending to false as the status is KILLED
101                        coordAction.setPending(0);
102                    }
103                    else {
104                        LOG.warn("Unexpected workflow " + wf.getId() + " STATUS " + wf.getStatus());
105                        coordAction.setLastModifiedTime(new Date());
106                        updateList.add(coordAction);
107                        jpaService.execute(new BulkUpdateInsertForCoordActionStatusJPAExecutor(updateList, null));
108                        return null;
109                    }
110                }
111            }
112
113            LOG.debug("Updating Coordintaor actionId :" + coordAction.getId() + "status to ="
114                            + coordAction.getStatus());
115            coordAction.setLastModifiedTime(new Date());
116            updateList.add(coordAction);
117
118            if (slaStatus != null) {
119                SLAEventBean slaEvent = SLADbOperations.createStatusEvent(coordAction.getSlaXml(), coordAction.getId(), slaStatus,
120                        SlaAppType.COORDINATOR_ACTION, LOG);
121                if(slaEvent != null) {
122                    insertList.add(slaEvent);
123                }
124            }
125
126            jpaService.execute(new BulkUpdateInsertForCoordActionStatusJPAExecutor(updateList, insertList));
127            CoordinatorAction.Status endStatus = coordAction.getStatus();
128            if (endStatus != initialStatus && EventHandlerService.isEnabled()) {
129                generateEvent(coordAction, coordJob.getUser(), coordJob.getAppName(), workflowJob.getStartTime());
130            }
131        }
132        catch (XException ex) {
133            LOG.warn("CoordActionCheckCommand Failed ", ex);
134            throw new CommandException(ex);
135        }
136        return null;
137    }
138
139    /* (non-Javadoc)
140     * @see org.apache.oozie.command.XCommand#getEntityKey()
141     */
142    @Override
143    public String getEntityKey() {
144        return actionId;
145    }
146
147    @Override
148    public String getKey() {
149        return getName() + "_" + actionId;
150    }
151
152    /* (non-Javadoc)
153     * @see org.apache.oozie.command.XCommand#isLockRequired()
154     */
155    @Override
156    protected boolean isLockRequired() {
157        return true;
158    }
159
160    /* (non-Javadoc)
161     * @see org.apache.oozie.command.XCommand#loadState()
162     */
163    @Override
164    protected void loadState() throws CommandException {
165        try {
166            jpaService = Services.get().get(JPAService.class);
167
168            if (jpaService != null) {
169                coordAction = jpaService.execute(new CoordActionGetForCheckJPAExecutor(actionId));
170                coordJob = jpaService.execute(new CoordinatorJobGetForUserAppnameJPAExecutor(
171                        coordAction.getJobId()));
172                workflowJob = jpaService.execute (new WorkflowJobGetForSLAJPAExecutor(coordAction.getExternalId()));
173                LogUtils.setLogInfo(coordAction, logInfo);
174            }
175            else {
176                throw new CommandException(ErrorCode.E0610);
177            }
178        }
179        catch (XException ex) {
180            throw new CommandException(ex);
181        }
182    }
183
184    /* (non-Javadoc)
185     * @see org.apache.oozie.command.XCommand#verifyPrecondition()
186     */
187    @Override
188    protected void verifyPrecondition() throws CommandException, PreconditionException {
189        // if the action has been updated, quit this command
190        Timestamp actionCheckTs = new Timestamp(System.currentTimeMillis() - actionCheckDelay * 1000);
191        Timestamp cactionLmt = coordAction.getLastModifiedTimestamp();
192        if (cactionLmt.after(actionCheckTs)) {
193            throw new PreconditionException(ErrorCode.E1100, "The coord action :" + actionId
194                    + " has been udated. Ignore CoordActionCheckCommand!");
195        }
196        if (coordAction.getStatus().equals(CoordinatorAction.Status.SUCCEEDED)
197                || coordAction.getStatus().equals(CoordinatorAction.Status.FAILED)
198                || coordAction.getStatus().equals(CoordinatorAction.Status.KILLED)) {
199            throw new PreconditionException(ErrorCode.E1100, "The coord action [" + actionId + "] must not have status "
200                    + CoordinatorAction.Status.SUCCEEDED.name() + ", " + CoordinatorAction.Status.FAILED.name()
201                    + ", or " + CoordinatorAction.Status.KILLED.name() + " but has status [" + coordAction.getStatus().name()
202                    + "]");
203        }
204    }
205}