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.util.ArrayList;
022import java.util.Date;
023import java.util.HashSet;
024import java.util.List;
025import java.util.Map;
026import java.util.Map.Entry;
027import java.util.Set;
028
029import org.apache.commons.lang.StringUtils;
030import org.apache.oozie.CoordinatorActionBean;
031import org.apache.oozie.CoordinatorJobBean;
032import org.apache.oozie.ErrorCode;
033import org.apache.oozie.XException;
034import org.apache.oozie.client.CoordinatorAction;
035import org.apache.oozie.client.CoordinatorJob;
036import org.apache.oozie.client.Job;
037import org.apache.oozie.client.OozieClient;
038import org.apache.oozie.client.rest.JsonBean;
039import org.apache.oozie.command.CommandException;
040import org.apache.oozie.command.PreconditionException;
041import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand;
042import org.apache.oozie.executor.jpa.BatchQueryExecutor;
043import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
044import org.apache.oozie.executor.jpa.CoordActionGetJPAExecutor;
045import org.apache.oozie.executor.jpa.CoordJobGetActionByActionNumberJPAExecutor;
046import org.apache.oozie.executor.jpa.CoordJobGetJPAExecutor;
047import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
048import org.apache.oozie.executor.jpa.JPAExecutorException;
049import org.apache.oozie.executor.jpa.SLARegistrationQueryExecutor;
050import org.apache.oozie.executor.jpa.SLARegistrationQueryExecutor.SLARegQuery;
051import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor;
052import org.apache.oozie.executor.jpa.SLASummaryQueryExecutor.SLASummaryQuery;
053import org.apache.oozie.service.JPAService;
054import org.apache.oozie.service.Services;
055import org.apache.oozie.sla.SLARegistrationBean;
056import org.apache.oozie.sla.SLASummaryBean;
057import org.apache.oozie.sla.service.SLAService;
058import org.apache.oozie.util.DateUtils;
059import org.apache.oozie.util.JobUtils;
060import org.apache.oozie.util.LogUtils;
061import org.apache.oozie.util.ParamChecker;
062import org.apache.oozie.util.StatusUtils;
063
064public class CoordChangeXCommand extends CoordinatorXCommand<Void> {
065    private final String jobId;
066    private Date newEndTime = null;
067    private Integer oldConcurrency = null;
068    private Integer newConcurrency = null;
069    private Date newPauseTime = null;
070    private Date oldPauseTime = null;
071    private boolean resetPauseTime = false;
072    private CoordinatorJob.Status jobStatus = null;
073    private CoordinatorJobBean coordJob;
074    private JPAService jpaService = null;
075    private Job.Status prevStatus;
076    private List<UpdateEntry> updateList = new ArrayList<UpdateEntry>();
077    private List<JsonBean> deleteList = new ArrayList<JsonBean>();
078
079    private static final Set<String> ALLOWED_CHANGE_OPTIONS = new HashSet<String>();
080    static {
081        ALLOWED_CHANGE_OPTIONS.add("endtime");
082        ALLOWED_CHANGE_OPTIONS.add("concurrency");
083        ALLOWED_CHANGE_OPTIONS.add("pausetime");
084        ALLOWED_CHANGE_OPTIONS.add(OozieClient.CHANGE_VALUE_STATUS);
085
086    }
087
088    /**
089     * This command is used to update the Coordinator job with the new values Update the coordinator job bean and update
090     * that to database.
091     *
092     * @param id Coordinator job id.
093     * @param changeValue This the changed value in the form key=value.
094     * @throws CommandException thrown if changeValue cannot be parsed properly.
095     */
096    public CoordChangeXCommand(String id, String changeValue) throws CommandException {
097        super("coord_change", "coord_change", 0);
098        this.jobId = ParamChecker.notEmpty(id, "id");
099        ParamChecker.notEmpty(changeValue, "value");
100
101        validateChangeValue(changeValue);
102    }
103
104    @Override
105    protected void setLogInfo() {
106        LogUtils.setLogInfo(jobId);
107    }
108
109    /**
110     * @param changeValue change value.
111     * @throws CommandException thrown if changeValue cannot be parsed properly.
112     */
113    private void validateChangeValue(String changeValue) throws CommandException {
114        Map<String, String> map = JobUtils.parseChangeValue(changeValue);
115
116        if (map.size() > ALLOWED_CHANGE_OPTIONS.size()) {
117            throw new CommandException(ErrorCode.E1015, changeValue, "must change endtime|concurrency|pausetime|status");
118        }
119
120        java.util.Iterator<Entry<String, String>> iter = map.entrySet().iterator();
121        while (iter.hasNext()) {
122            Entry<String, String> entry = iter.next();
123            String key = entry.getKey();
124            String value = entry.getValue();
125
126            if (!ALLOWED_CHANGE_OPTIONS.contains(key)) {
127                throw new CommandException(ErrorCode.E1015, changeValue, "must change endtime|concurrency|pausetime|status");
128            }
129
130            if (!key.equals(OozieClient.CHANGE_VALUE_PAUSETIME) && value.equalsIgnoreCase("")) {
131                throw new CommandException(ErrorCode.E1015, changeValue, "value on " + key + " can not be empty");
132            }
133        }
134
135        if (map.containsKey(OozieClient.CHANGE_VALUE_ENDTIME)) {
136            String value = map.get(OozieClient.CHANGE_VALUE_ENDTIME);
137            try {
138                newEndTime = DateUtils.parseDateOozieTZ(value);
139            }
140            catch (Exception ex) {
141                throw new CommandException(ErrorCode.E1015, value, "must be a valid date");
142            }
143        }
144
145        if (map.containsKey(OozieClient.CHANGE_VALUE_CONCURRENCY)) {
146            String value = map.get(OozieClient.CHANGE_VALUE_CONCURRENCY);
147            try {
148                newConcurrency = Integer.parseInt(value);
149            }
150            catch (NumberFormatException ex) {
151                throw new CommandException(ErrorCode.E1015, value, "must be a valid integer");
152            }
153        }
154
155        if (map.containsKey(OozieClient.CHANGE_VALUE_PAUSETIME)) {
156            String value = map.get(OozieClient.CHANGE_VALUE_PAUSETIME);
157            if (value.equals("")) { // this is to reset pause time to null;
158                resetPauseTime = true;
159            }
160            else {
161                try {
162                    newPauseTime = DateUtils.parseDateOozieTZ(value);
163                }
164                catch (Exception ex) {
165                    throw new CommandException(ErrorCode.E1015, value, "must be a valid date");
166                }
167            }
168        }
169
170        if (map.containsKey(OozieClient.CHANGE_VALUE_STATUS)) {
171            String value = map.get(OozieClient.CHANGE_VALUE_STATUS);
172            if (!StringUtils.isEmpty(value)) {
173                jobStatus = CoordinatorJob.Status.valueOf(value);
174            }
175        }
176    }
177
178    /**
179     * Check if new end time is valid.
180     *
181     * @param coordJob coordinator job id.
182     * @param newEndTime new end time.
183     * @throws CommandException thrown if new end time is not valid.
184     */
185    private void checkEndTime(CoordinatorJobBean coordJob, Date newEndTime) throws CommandException {
186        //It's ok to set end date before start date.
187    }
188
189    /**
190     * Check if new pause time is valid.
191     *
192     * @param coordJob coordinator job id.
193     * @param newPauseTime new pause time.
194     * @param newEndTime new end time, can be null meaning no change on end time.
195     * @throws CommandException thrown if new pause time is not valid.
196     */
197    private void checkPauseTime(CoordinatorJobBean coordJob, Date newPauseTime)
198            throws CommandException {
199        //no check.
200    }
201
202    /**
203     * Check if status change is valid.
204     *
205     * @param coordJob the coord job
206     * @param jobStatus the job status
207     * @throws CommandException the command exception
208     */
209    private void checkStatusChange(CoordinatorJobBean coordJob, CoordinatorJob.Status jobStatus)
210            throws CommandException {
211        if (!jobStatus.equals(CoordinatorJob.Status.RUNNING) && !jobStatus.equals(CoordinatorJob.Status.IGNORED)) {
212            throw new CommandException(ErrorCode.E1015, jobStatus, " must be RUNNING or IGNORED");
213        }
214
215        if (jobStatus.equals(CoordinatorJob.Status.RUNNING)) {
216            if (!(coordJob.getStatus().equals(CoordinatorJob.Status.FAILED) || coordJob.getStatus().equals(
217                    CoordinatorJob.Status.KILLED) || coordJob.getStatus().equals(CoordinatorJob.Status.IGNORED))) {
218                throw new CommandException(ErrorCode.E1015, jobStatus,
219                        " Only FAILED, KILLED, IGNORED job can be changed to RUNNING. Current job status is "
220                                + coordJob.getStatus());
221            }
222        }
223        else {
224            if (!(coordJob.getStatus().equals(CoordinatorJob.Status.FAILED) || coordJob.getStatus().equals(
225                    CoordinatorJob.Status.KILLED))
226                    || coordJob.isPending()) {
227                throw new CommandException(ErrorCode.E1015, jobStatus,
228                        " Only FAILED or KILLED non-pending job can be changed to IGNORED. Current job status is "
229                                + coordJob.getStatus() + " and pending status is " + coordJob.isPending());
230            }
231        }
232    }
233
234    /**
235     * Process lookahead created actions that become invalid because of the new pause time,
236     * These actions will be deleted from DB, also the coordinator job will be updated accordingly
237     *
238     * @param coordJob coordinator job
239     * @param newPauseTime new pause time
240     * @throws JPAExecutorException, CommandException
241     */
242    private void processLookaheadActions(CoordinatorJobBean coordJob, Date newTime) throws CommandException,
243            JPAExecutorException {
244        int lastActionNumber = coordJob.getLastActionNumber();
245        Date  lastActionTime = null;
246        Date  tempDate = null;
247
248        while ((tempDate = deleteAction(lastActionNumber, newTime)) != null) {
249            lastActionNumber--;
250            lastActionTime = tempDate;
251        }
252        if (lastActionTime != null) {
253            LOG.debug("New pause/end date is : " + newTime + " and last action number is : " + lastActionNumber);
254            coordJob.setLastActionNumber(lastActionNumber);
255            coordJob.setLastActionTime(lastActionTime);
256            coordJob.setNextMaterializedTime(lastActionTime);
257            coordJob.resetDoneMaterialization();
258        }
259    }
260
261    /**
262     * Delete coordinator action
263     *
264     * @param actionNum coordinator action number
265     */
266    private Date deleteAction(int actionNum, Date afterDate) throws CommandException {
267        try {
268            if (actionNum <= 0) {
269                return null;
270            }
271
272            String actionId = jpaService.execute(new CoordJobGetActionByActionNumberJPAExecutor(jobId, actionNum));
273            CoordinatorActionBean bean = jpaService.execute(new CoordActionGetJPAExecutor(actionId));
274            if (afterDate.compareTo(bean.getNominalTime()) <= 0) {
275                // delete SLA registration entry (if any) for action
276                if (SLAService.isEnabled()) {
277                    Services.get().get(SLAService.class).removeRegistration(actionId);
278                }
279                SLARegistrationBean slaReg = SLARegistrationQueryExecutor.getInstance().get(SLARegQuery.GET_SLA_REG_ALL, actionId);
280                if (slaReg != null) {
281                    LOG.debug("Deleting registration bean corresponding to action " + slaReg.getId());
282                    deleteList.add(slaReg);
283                }
284                SLASummaryBean slaSummaryBean = SLASummaryQueryExecutor.getInstance().get(
285                        SLASummaryQuery.GET_SLA_SUMMARY, actionId);
286                if (slaSummaryBean != null) {
287                    LOG.debug("Deleting summary bean corresponding to action " + slaSummaryBean.getId());
288                    deleteList.add(slaSummaryBean);
289                }
290                if (bean.getStatus() == CoordinatorAction.Status.WAITING
291                        || bean.getStatus() == CoordinatorAction.Status.READY) {
292                    deleteList.add(bean);
293                }
294                else {
295                    throw new CommandException(ErrorCode.E1022, bean.getId());
296                }
297                return bean.getNominalTime();
298            }
299            else {
300                return null;
301            }
302
303        }
304        catch (JPAExecutorException e) {
305            throw new CommandException(e);
306        }
307    }
308
309    /**
310     * Check if new end time, new concurrency, new pause time are valid.
311     *
312     * @param coordJob coordinator job id.
313     * @param newEndTime new end time.
314     * @param newConcurrency new concurrency.
315     * @param newPauseTime new pause time.
316     * @throws CommandException thrown if new values are not valid.
317     */
318    private void check(CoordinatorJobBean coordJob, Date newEndTime, Integer newConcurrency, Date newPauseTime,
319            CoordinatorJob.Status jobStatus) throws CommandException {
320
321        if (coordJob.getStatus() == CoordinatorJob.Status.KILLED
322                || coordJob.getStatus() == CoordinatorJob.Status.IGNORED) {
323            if (jobStatus == null || (newEndTime != null || newConcurrency != null || newPauseTime != null)) {
324                throw new CommandException(ErrorCode.E1016);
325            }
326        }
327
328        if (newEndTime != null) {
329            checkEndTime(coordJob, newEndTime);
330        }
331
332        if (newPauseTime != null) {
333            checkPauseTime(coordJob, newPauseTime);
334        }
335        if (jobStatus != null) {
336            checkStatusChange(coordJob, jobStatus);
337        }
338    }
339
340    /* (non-Javadoc)
341     * @see org.apache.oozie.command.XCommand#execute()
342     */
343    @Override
344    protected Void execute() throws CommandException {
345        LOG.info("STARTED CoordChangeXCommand for jobId=" + jobId);
346
347        try {
348            oldConcurrency = this.coordJob.getConcurrency();
349            if (newEndTime != null) {
350                // during coord materialization, nextMaterializedTime is set to
351                // startTime + n(actions materialized) * frequency and this can be AFTER endTime,
352                // while doneMaterialization is true. Hence the following checks
353                // for newEndTime being in the middle of endTime and nextMatdTime.
354                // Since job is already done materialization so no need to change
355                boolean dontChange = coordJob.getEndTime().before(newEndTime)
356                        && coordJob.getNextMaterializedTime() != null
357                        && coordJob.getNextMaterializedTime().after(newEndTime);
358                if (!dontChange) {
359                    coordJob.setEndTime(newEndTime);
360                    // OOZIE-1703, we should SUCCEEDED the coord, if it's in PREP and new endtime is before start time
361                    if (coordJob.getStartTime().compareTo(newEndTime) >= 0) {
362                        if (coordJob.getStatus() != CoordinatorJob.Status.PREP) {
363                            processLookaheadActions(coordJob, newEndTime);
364                        }
365                        if (coordJob.getStatus() == CoordinatorJob.Status.PREP
366                                || coordJob.getStatus() == CoordinatorJob.Status.RUNNING) {
367                            LOG.info("Changing coord status to SUCCEEDED, because it's in " + coordJob.getStatus()
368                                    + " and new end time is before start time. Startime is " + coordJob.getStartTime()
369                                    + " and new end time is " + newEndTime);
370
371                            coordJob.setStatus(CoordinatorJob.Status.SUCCEEDED);
372                            coordJob.resetPending();
373                        }
374                        coordJob.setDoneMaterialization();
375                    }
376                    else {
377                        // move it to running iff new end time is after starttime.
378                        if (coordJob.getStatus() == CoordinatorJob.Status.SUCCEEDED) {
379                            coordJob.setStatus(CoordinatorJob.Status.RUNNING);
380                        }
381                        if (coordJob.getStatus() == CoordinatorJob.Status.DONEWITHERROR
382                                || coordJob.getStatus() == CoordinatorJob.Status.FAILED) {
383                            // Check for backward compatibility for Oozie versions (3.2 and before)
384                            // when RUNNINGWITHERROR, SUSPENDEDWITHERROR and
385                            // PAUSEDWITHERROR is not supported
386                            coordJob.setStatus(StatusUtils
387                                    .getStatusIfBackwardSupportTrue(CoordinatorJob.Status.RUNNINGWITHERROR));
388                        }
389                        coordJob.setPending();
390                        coordJob.resetDoneMaterialization();
391                        processLookaheadActions(coordJob, newEndTime);
392                    }
393                }
394
395                else {
396                    LOG.info("Didn't change endtime. Endtime is in between coord end time and next materialization time."
397                            + "Coord endTime = " + DateUtils.formatDateOozieTZ(newEndTime)
398                            + " next materialization time ="
399                            + DateUtils.formatDateOozieTZ(coordJob.getNextMaterializedTime()));
400                }
401            }
402
403            if (newConcurrency != null) {
404                this.coordJob.setConcurrency(newConcurrency);
405            }
406
407            if (newPauseTime != null || resetPauseTime == true) {
408                this.coordJob.setPauseTime(newPauseTime);
409                if (oldPauseTime != null && newPauseTime != null) {
410                    if (oldPauseTime.before(newPauseTime)) {
411                        if (this.coordJob.getStatus() == Job.Status.PAUSED) {
412                            this.coordJob.setStatus(Job.Status.RUNNING);
413                        }
414                        else if (this.coordJob.getStatus() == Job.Status.PAUSEDWITHERROR) {
415                            this.coordJob.setStatus(Job.Status.RUNNINGWITHERROR);
416                        }
417                    }
418                }
419                else if (oldPauseTime != null && newPauseTime == null) {
420                    if (this.coordJob.getStatus() == Job.Status.PAUSED) {
421                        this.coordJob.setStatus(Job.Status.RUNNING);
422                    }
423                    else if (this.coordJob.getStatus() == Job.Status.PAUSEDWITHERROR) {
424                        this.coordJob.setStatus(Job.Status.RUNNINGWITHERROR);
425                    }
426                }
427                if (!resetPauseTime) {
428                    processLookaheadActions(coordJob, newPauseTime);
429                }
430            }
431            if (jobStatus != null) {
432                coordJob.setStatus(jobStatus);
433                LOG.info("Coord status is changed to " + jobStatus + " from " + prevStatus);
434                if (jobStatus.equals(CoordinatorJob.Status.RUNNING)) {
435                    coordJob.setPending();
436                    if (coordJob.getNextMaterializedTime() != null
437                            && coordJob.getEndTime().after(coordJob.getNextMaterializedTime())) {
438                        coordJob.resetDoneMaterialization();
439                    }
440                } else if (jobStatus.equals(CoordinatorJob.Status.IGNORED)) {
441                    coordJob.resetPending();
442                    coordJob.setDoneMaterialization();
443                }
444            }
445
446            if (coordJob.getNextMaterializedTime() != null && coordJob.getEndTime().compareTo(coordJob.getNextMaterializedTime()) <= 0) {
447                LOG.info("[" + coordJob.getId() + "]: all actions have been materialized, job status = " + coordJob.getStatus()
448                        + ", set pending to true");
449                // set doneMaterialization to true when materialization is done
450                coordJob.setDoneMaterialization();
451            }
452
453            coordJob.setLastModifiedTime(new Date());
454            updateList.add(new UpdateEntry<CoordJobQuery>(CoordJobQuery.UPDATE_COORD_JOB_CHANGE, coordJob));
455            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(null, updateList, deleteList);
456
457            if (newConcurrency != null && newConcurrency > oldConcurrency) {
458                queue(new CoordActionReadyXCommand(jobId));
459            }
460
461            return null;
462        }
463        catch (XException ex) {
464            throw new CommandException(ex);
465        }
466        finally {
467            LOG.info("ENDED CoordChangeXCommand for jobId=" + jobId);
468            // update bundle action
469            if (coordJob.getBundleId() != null) {
470                //ignore pending as it'sync command
471                BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, prevStatus, true);
472                bundleStatusUpdate.call();
473            }
474        }
475    }
476
477    /* (non-Javadoc)
478     * @see org.apache.oozie.command.XCommand#getEntityKey()
479     */
480    @Override
481    public String getEntityKey() {
482        return this.jobId;
483    }
484
485    /* (non-Javadoc)
486     * @see org.apache.oozie.command.XCommand#loadState()
487     */
488    @Override
489    protected void loadState() throws CommandException{
490        jpaService = Services.get().get(JPAService.class);
491
492        if (jpaService == null) {
493            throw new CommandException(ErrorCode.E0610);
494        }
495
496        try {
497            this.coordJob = jpaService.execute(new CoordJobGetJPAExecutor(jobId));
498            oldPauseTime = coordJob.getPauseTime();
499            prevStatus = coordJob.getStatus();
500        }
501        catch (JPAExecutorException e) {
502            throw new CommandException(e);
503        }
504
505        LogUtils.setLogInfo(this.coordJob);
506    }
507
508    /* (non-Javadoc)
509     * @see org.apache.oozie.command.XCommand#verifyPrecondition()
510     */
511    @Override
512    protected void verifyPrecondition() throws CommandException,PreconditionException {
513        check(this.coordJob, newEndTime, newConcurrency, newPauseTime, jobStatus);
514    }
515
516    /* (non-Javadoc)
517     * @see org.apache.oozie.command.XCommand#isLockRequired()
518     */
519    @Override
520    protected boolean isLockRequired() {
521        return true;
522    }
523}