Source code

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.coord;
020
021import java.io.IOException;
022import java.io.StringReader;
023import java.util.Date;
024import java.util.List;
025import org.apache.hadoop.conf.Configuration;
026import org.apache.hadoop.fs.Path;
027import org.apache.oozie.CoordinatorActionBean;
028import org.apache.oozie.CoordinatorActionInfo;
029import org.apache.oozie.CoordinatorJobBean;
030import org.apache.oozie.ErrorCode;
031import org.apache.oozie.SLAEventBean;
032import org.apache.oozie.XException;
033import org.apache.oozie.action.ActionExecutorException;
034import org.apache.oozie.action.hadoop.FsActionExecutor;
035import org.apache.oozie.client.CoordinatorAction;
036import org.apache.oozie.client.CoordinatorJob;
037import org.apache.oozie.client.Job;
038import org.apache.oozie.client.SLAEvent.SlaAppType;
039import org.apache.oozie.client.rest.RestConstants;
040import org.apache.oozie.command.CommandException;
041import org.apache.oozie.command.PreconditionException;
042import org.apache.oozie.command.RerunTransitionXCommand;
043import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand;
044import org.apache.oozie.coord.CoordELFunctions;
045import org.apache.oozie.coord.CoordUtils;
046import org.apache.oozie.executor.jpa.BatchQueryExecutor;
047import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
048import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
049import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
050import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
051import org.apache.oozie.executor.jpa.JPAExecutorException;
052import org.apache.oozie.service.EventHandlerService;
053import org.apache.oozie.sla.SLAOperations;
054import org.apache.oozie.sla.service.SLAService;
055import org.apache.oozie.util.InstrumentUtils;
056import org.apache.oozie.util.LogUtils;
057import org.apache.oozie.util.ParamChecker;
058import org.apache.oozie.util.StatusUtils;
059import org.apache.oozie.util.XConfiguration;
060import org.apache.oozie.util.XLog;
061import org.apache.oozie.util.XmlUtils;
062import org.apache.oozie.util.db.SLADbOperations;
063import org.jdom.Element;
064import org.jdom.JDOMException;
065
066/**
067 * Rerun coordinator actions by a list of dates or ids. User can specify if refresh or noCleanup.
068 * <p/>
069 * The "rerunType" can be set as {@link RestConstants.JOB_COORD_RERUN_DATE} or
070 * {@link RestConstants.JOB_COORD_RERUN_ACTION}.
071 * <p/>
072 * The "refresh" is used to indicate if user wants to refresh an action's input and output events.
073 * <p/>
074 * The "noCleanup" is used to indicate if user wants to cleanup output events for given rerun actions
075 */
076@SuppressWarnings("deprecation")
077public class CoordRerunXCommand extends RerunTransitionXCommand<CoordinatorActionInfo> {
078
079    private String rerunType;
080    private String scope;
081    private boolean refresh;
082    private boolean noCleanup;
083    private CoordinatorJobBean coordJob = null;
084    protected boolean prevPending;
085    private boolean failed;
086
087    /**
088     * The constructor for class {@link CoordRerunXCommand}
089     *
090     * @param jobId the job id
091     * @param rerunType rerun type {@link RestConstants.JOB_COORD_RERUN_DATE} or {@link RestConstants.JOB_COORD_RERUN_ACTION}
092     * @param scope the rerun scope for given rerunType separated by ","
093     * @param refresh true if user wants to refresh input/output dataset urls
094     * @param noCleanup false if user wants to cleanup output events for given rerun actions
095     */
096    public CoordRerunXCommand(String jobId, String rerunType, String scope, boolean refresh, boolean noCleanup,
097                              boolean failed) {
098        super("coord_rerun", "coord_rerun", 1);
099        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
100        this.rerunType = ParamChecker.notEmpty(rerunType, "rerunType");
101        this.scope = ParamChecker.notEmpty(scope, "scope");
102        this.refresh = refresh;
103        this.noCleanup = noCleanup;
104        this.failed = failed;
105    }
106
107    /**
108     * Check if all given actions are eligible to rerun.
109     *
110     * @param actions list of CoordinatorActionBean
111     * @return true if all actions are eligible to rerun
112     */
113    private static boolean checkAllActionsRunnable(List<CoordinatorActionBean> coordActions) {
114        ParamChecker.notNull(coordActions, "Coord actions to be rerun");
115        boolean ret = false;
116        for (CoordinatorActionBean coordAction : coordActions) {
117            ret = true;
118            if (!coordAction.isTerminalStatus()) {
119                ret = false;
120                break;
121            }
122        }
123        return ret;
124    }
125
126    /**
127     * Cleanup output-events directories
128     *
129     * @param eAction coordinator action xml
130     * @param user user name
131     * @param group group name
132     */
133    @SuppressWarnings("unchecked")
134    private void cleanupOutputEvents(Element eAction, String user, String group) {
135        Element outputList = eAction.getChild("output-events", eAction.getNamespace());
136        if (outputList != null) {
137            for (Element data : (List<Element>) outputList.getChildren("data-out", eAction.getNamespace())) {
138                String nocleanup = data.getAttributeValue("nocleanup");
139                if (data.getChild("uris", data.getNamespace()) != null
140                        && (nocleanup == null || !nocleanup.equals("true"))) {
141                    String uris = data.getChild("uris", data.getNamespace()).getTextTrim();
142                    if (uris != null) {
143                        String[] uriArr = uris.split(CoordELFunctions.INSTANCE_SEPARATOR);
144                        FsActionExecutor fsAe = new FsActionExecutor();
145                        for (String uri : uriArr) {
146                            Path path = new Path(uri);
147                            try {
148                                fsAe.delete(user, group, path);
149                                LOG.debug("Cleanup the output dir " + path);
150                            }
151                            catch (ActionExecutorException ae) {
152                                LOG.warn("Failed to cleanup the output dir " + uri, ae);
153                            }
154                        }
155                    }
156
157                }
158            }
159        }
160        else {
161            LOG.info("No output-events defined in coordinator xml. Therefore nothing to cleanup");
162        }
163    }
164
165    /**
166     * Refresh an action's input and ouput events.
167     *
168     * @param coordJob coordinator job bean
169     * @param coordAction coordinator action bean
170     * @throws Exception thrown if failed to materialize coordinator action
171     */
172    private void refreshAction(CoordinatorJobBean coordJob, CoordinatorActionBean coordAction) throws Exception {
173        Configuration jobConf = null;
174        try {
175            jobConf = new XConfiguration(new StringReader(coordJob.getConf()));
176        }
177        catch (IOException ioe) {
178            LOG.warn("Configuration parse error. read from DB :" + coordJob.getConf(), ioe);
179            throw new CommandException(ErrorCode.E1005, ioe.getMessage(), ioe);
180        }
181        String jobXml = coordJob.getJobXml();
182        Element eJob = XmlUtils.parseXml(jobXml);
183        Date actualTime = new Date();
184        String actionXml = CoordCommandUtils.materializeOneInstance(jobId, dryrun, (Element) eJob.clone(), coordAction
185                .getNominalTime(), actualTime, coordAction.getActionNumber(), jobConf, coordAction);
186        LOG.debug("Refresh Action actionId=" + coordAction.getId() + ", actionXml="
187                + XmlUtils.prettyPrint(actionXml).toString());
188        coordAction.setActionXml(actionXml);
189    }
190
191    /**
192     * Update an action into database table
193     *
194     * @param coordJob coordinator job bean
195     * @param coordAction coordinator action bean
196     * @throws Exception thrown failed to update coordinator action bean or unable to write sla registration event
197     */
198    private void updateAction(CoordinatorJobBean coordJob, CoordinatorActionBean coordAction)
199            throws Exception {
200        LOG.debug("updateAction for actionId=" + coordAction.getId());
201        if (coordAction.getStatus() == CoordinatorAction.Status.TIMEDOUT) {
202            LOG.debug("Updating created time for TIMEDOUT action id =" + coordAction.getId());
203            coordAction.setCreatedTime(new Date());
204        }
205        coordAction.setStatus(CoordinatorAction.Status.WAITING);
206        if(!failed) {
207            coordAction.setExternalId(null);
208        }
209        coordAction.setExternalStatus(null);
210        coordAction.setRerunTime(new Date());
211        coordAction.setLastModifiedTime(new Date());
212        coordAction.setErrorCode("");
213        coordAction.setErrorMessage("");
214        updateList.add(new UpdateEntry<CoordActionQuery>(CoordActionQuery.UPDATE_COORD_ACTION_RERUN, coordAction));
215        writeActionRegistration(coordAction.getActionXml(), coordAction, coordJob.getUser(), coordJob.getGroup());
216    }
217
218    /**
219     * Create SLA RegistrationEvent
220     *
221     * @param actionXml action xml
222     * @param actionBean coordinator action bean
223     * @param user user name
224     * @param group group name
225     * @throws Exception thrown if unable to write sla registration event
226     */
227    private void writeActionRegistration(String actionXml, CoordinatorActionBean actionBean, String user, String group)
228            throws Exception {
229        Element eAction = XmlUtils.parseXml(actionXml);
230        Element eSla = eAction.getChild("action", eAction.getNamespace()).getChild("info", eAction.getNamespace("sla"));
231        SLAEventBean slaEvent = SLADbOperations.createSlaRegistrationEvent(eSla, actionBean.getId(),
232                SlaAppType.COORDINATOR_ACTION, user, group, LOG);
233        if(slaEvent != null) {
234            insertList.add(slaEvent);
235        }
236    }
237
238    /* (non-Javadoc)
239     * @see org.apache.oozie.command.XCommand#getEntityKey()
240     */
241    @Override
242    public String getEntityKey() {
243        return jobId;
244    }
245
246    /* (non-Javadoc)
247     * @see org.apache.oozie.command.XCommand#isLockRequired()
248     */
249    @Override
250    protected boolean isLockRequired() {
251        return true;
252    }
253
254    /* (non-Javadoc)
255     * @see org.apache.oozie.command.XCommand#loadState()
256     */
257    @Override
258    protected void loadState() throws CommandException {
259        try {
260            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, jobId);
261            prevPending = coordJob.isPending();
262        }
263        catch (JPAExecutorException je) {
264            throw new CommandException(je);
265        }
266        LogUtils.setLogInfo(coordJob);
267    }
268
269    /* (non-Javadoc)
270     * @see org.apache.oozie.command.XCommand#verifyPrecondition()
271     */
272    @Override
273    protected void verifyPrecondition() throws CommandException, PreconditionException {
274        BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, coordJob.getStatus());
275
276        // no actions have been created for PREP job
277        if (coordJob.getStatus() == CoordinatorJob.Status.PREP || coordJob.getStatus() == CoordinatorJob.Status.IGNORED) {
278            LOG.info("CoordRerunXCommand is not able to run, job status=" + coordJob.getStatus() + ", jobid=" + jobId);
279            // Call the parent so the pending flag is reset and state transition
280            // of bundle can happen
281            if (coordJob.getBundleId() != null) {
282                bundleStatusUpdate.call();
283            }
284            if (coordJob.getStatus() == CoordinatorJob.Status.PREP) {
285                throw new CommandException(ErrorCode.E1018,
286                        "coordinator job is PREP so no actions are materialized to rerun!");
287            }
288            else {
289                throw new CommandException(ErrorCode.E1018,
290                        "coordinator job is IGNORED, please change it to RUNNING before rerunning actions");
291            }
292        }
293    }
294
295    @Override
296    protected void eagerLoadState() throws CommandException {
297        try {
298            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, jobId);
299        }
300        catch (JPAExecutorException e) {
301            throw new CommandException(e);
302        }
303    }
304
305    @Override
306    protected void eagerVerifyPrecondition() throws CommandException, PreconditionException {
307        verifyPrecondition();
308    }
309
310    @Override
311    public void rerunChildren() throws CommandException {
312        boolean isError = false;
313        try {
314            CoordinatorActionInfo coordInfo = null;
315            InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
316            List<CoordinatorActionBean> coordActions = CoordUtils.getCoordActions(rerunType, jobId, scope, false);
317            if (checkAllActionsRunnable(coordActions)) {
318                for (CoordinatorActionBean coordAction : coordActions) {
319                    String actionXml = coordAction.getActionXml();
320                    if (!noCleanup) {
321                        Element eAction = XmlUtils.parseXml(actionXml);
322                        cleanupOutputEvents(eAction, coordJob.getUser(), coordJob.getGroup());
323                    }
324                    if (refresh) {
325                        refreshAction(coordJob, coordAction);
326                    }
327                    updateAction(coordJob, coordAction);
328                    if (SLAService.isEnabled()) {
329                        SLAOperations.updateRegistrationEvent(coordAction.getId());
330                    }
331                    queue(new CoordActionNotificationXCommand(coordAction), 100);
332                    queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()), 100);
333                }
334            }
335            else {
336                isError = true;
337                throw new CommandException(ErrorCode.E1018, "part or all actions are not eligible to rerun!");
338            }
339            coordInfo = new CoordinatorActionInfo(coordActions);
340
341            ret = coordInfo;
342        }
343        catch (XException xex) {
344            isError = true;
345            throw new CommandException(xex);
346        }
347        catch (JDOMException jex) {
348            isError = true;
349            throw new CommandException(ErrorCode.E0700, jex.getMessage(), jex);
350        }
351        catch (Exception ex) {
352            isError = true;
353            throw new CommandException(ErrorCode.E1018, ex.getMessage(), ex);
354        }
355        finally{
356            if(isError){
357                transitToPrevious();
358            }
359        }
360    }
361
362    /*
363     * (non-Javadoc)
364     * @see org.apache.oozie.command.TransitionXCommand#getJob()
365     */
366    @Override
367    public Job getJob() {
368        return coordJob;
369    }
370
371    @Override
372    public void notifyParent() throws CommandException {
373        //update bundle action
374        if (getPrevStatus() != null && coordJob.getBundleId() != null) {
375            BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, getPrevStatus());
376            bundleStatusUpdate.call();
377        }
378    }
379
380    @Override
381    public void updateJob() {
382        if (getPrevStatus()!= null){
383            Job.Status coordJobStatus = getPrevStatus();
384            if(coordJobStatus.equals(Job.Status.PAUSED) || coordJobStatus.equals(Job.Status.PAUSEDWITHERROR)) {
385                coordJob.setStatus(coordJobStatus);
386            }
387            if (prevPending) {
388                coordJob.setPending();
389            } else {
390                coordJob.resetPending();
391            }
392        }
393        updateList.add(new UpdateEntry<CoordJobQuery>(CoordJobQuery.UPDATE_COORD_JOB_STATUS_PENDING, coordJob));
394    }
395
396    /* (non-Javadoc)
397     * @see org.apache.oozie.command.RerunTransitionXCommand#performWrites()
398     */
399    @Override
400    public void performWrites() throws CommandException {
401        try {
402            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
403            if (EventHandlerService.isEnabled()) {
404                generateEvents(coordJob, null);
405            }
406        }
407        catch (JPAExecutorException e) {
408            throw new CommandException(e);
409        }
410    }
411
412    /* (non-Javadoc)
413     * @see org.apache.oozie.command.RerunTransitionXCommand#getLog()
414     */
415    @Override
416    public XLog getLog() {
417        return LOG;
418    }
419
420    @Override
421    public final void transitToNext() {
422        prevStatus = coordJob.getStatus();
423        if (prevStatus == CoordinatorJob.Status.SUCCEEDED || prevStatus == CoordinatorJob.Status.PAUSED
424                || prevStatus == CoordinatorJob.Status.SUSPENDED || prevStatus == CoordinatorJob.Status.RUNNING) {
425            coordJob.setStatus(Job.Status.RUNNING);
426        }
427        else {
428            // Check for backward compatibility for Oozie versions (3.2 and before)
429            // when RUNNINGWITHERROR, SUSPENDEDWITHERROR and
430            // PAUSEDWITHERROR is not supported
431            coordJob.setStatus(StatusUtils.getStatusIfBackwardSupportTrue(CoordinatorJob.Status.RUNNINGWITHERROR));
432        }
433        // used for backward support of coordinator 0.1 schema
434        coordJob.setStatus(StatusUtils.getStatusForCoordRerun(coordJob, prevStatus));
435        coordJob.setPending();
436    }
437
438    private final void transitToPrevious() throws CommandException {
439        coordJob.setStatus(getPrevStatus());
440        if (!prevPending) {
441            coordJob.resetPending();
442        }
443        else {
444            coordJob.setPending();
445        }
446    }
447}