Skip to content

Commit a88cf0d

Browse files
sureshanapartisureshanaparti
committed
refuse the cancel for a job completed, keep the result
1 parent 6d200b2 commit a88cf0d

4 files changed

Lines changed: 160 additions & 45 deletions

File tree

‎framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/dao/AsyncJobDao.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,4 +82,10 @@ public interface AsyncJobDao extends GenericDao<AsyncJobVO, Long> {
8282

8383
/** Flags the job's cancel as delivered to its worker; true only for the caller that flipped the flag. */
8484
boolean claimCancelDelivery(long jobId);
85+
86+
/** Records a cancel request with its result on a job that is still in progress and not yet being cancelled; false once the job completed or is already being cancelled. */
87+
boolean requestCancel(long jobId, int resultCode, String result);
88+
89+
/** Takes a cancel request back from a job that is still in progress, reopening its result; false if there was none to take back. */
90+
boolean withdrawCancel(long jobId);
8591
}

‎framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/dao/AsyncJobDaoImpl.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.cloudstack.framework.jobs.impl.AsyncJobVO;
2929
import org.apache.cloudstack.jobs.JobInfo;
3030

31+
import com.cloud.utils.DateUtil;
3132
import com.cloud.utils.db.DB;
3233
import com.cloud.utils.db.Filter;
3334
import com.cloud.utils.db.GenericDaoBase;
@@ -55,6 +56,7 @@ public class AsyncJobDaoImpl extends GenericDaoBase<AsyncJobVO, Long> implements
5556
private final SearchBuilder<AsyncJobVO> cancelledAsyncJobSearch;
5657
private final SearchBuilder<AsyncJobVO> runningWorkJobSearch;
5758
private final SearchBuilder<AsyncJobVO> undeliveredCancelSearch;
59+
private final SearchBuilder<AsyncJobVO> cancelRequestSearch;
5860

5961
public AsyncJobDaoImpl() {
6062
pendingAsyncJobSearch = createSearchBuilder();
@@ -146,6 +148,12 @@ public AsyncJobDaoImpl() {
146148
undeliveredCancelSearch.or("status", undeliveredCancelSearch.entity().getStatus(), SearchCriteria.Op.EQ);
147149
undeliveredCancelSearch.cp();
148150
undeliveredCancelSearch.done();
151+
152+
cancelRequestSearch = createSearchBuilder();
153+
cancelRequestSearch.and("id", cancelRequestSearch.entity().getId(), SearchCriteria.Op.EQ);
154+
cancelRequestSearch.and("status", cancelRequestSearch.entity().getStatus(), SearchCriteria.Op.EQ);
155+
cancelRequestSearch.and("cancelRequested", cancelRequestSearch.entity().isCancelRequested(), SearchCriteria.Op.EQ);
156+
cancelRequestSearch.done();
149157
}
150158

151159
@Override
@@ -383,4 +391,34 @@ public boolean claimCancelDelivery(long jobId) {
383391
sc.setParameters("status", JobInfo.Status.CANCELLED);
384392
return update(ub, sc, null) == 1;
385393
}
394+
395+
@Override
396+
public boolean requestCancel(long jobId, int resultCode, String result) {
397+
AsyncJobVO requested = createForUpdate();
398+
requested.setCancelRequested(true);
399+
requested.setCancelDelivered(false);
400+
requested.setResultCode(resultCode);
401+
requested.setResult(result);
402+
requested.setLastUpdated(DateUtil.currentGMTTime());
403+
return update(getUpdateBuilder(requested), inProgressWithCancelRequested(jobId, false), null) == 1;
404+
}
405+
406+
@Override
407+
public boolean withdrawCancel(long jobId) {
408+
AsyncJobVO withdrawn = createForUpdate();
409+
withdrawn.setCancelRequested(false);
410+
withdrawn.setCancelDelivered(false);
411+
withdrawn.setResultCode(0);
412+
withdrawn.setResult(null);
413+
withdrawn.setLastUpdated(DateUtil.currentGMTTime());
414+
return update(getUpdateBuilder(withdrawn), inProgressWithCancelRequested(jobId, true), null) == 1;
415+
}
416+
417+
private SearchCriteria<AsyncJobVO> inProgressWithCancelRequested(long jobId, boolean cancelRequested) {
418+
SearchCriteria<AsyncJobVO> sc = cancelRequestSearch.create();
419+
sc.setParameters("id", jobId);
420+
sc.setParameters("status", JobInfo.Status.IN_PROGRESS);
421+
sc.setParameters("cancelRequested", cancelRequested);
422+
return sc;
423+
}
386424
}

‎framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java‎

Lines changed: 52 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@
5555
import org.apache.cloudstack.framework.config.ConfigKey;
5656
import org.apache.cloudstack.framework.config.Configurable;
5757
import org.apache.cloudstack.framework.jobs.AsyncJob;
58+
import org.apache.cloudstack.framework.jobs.AsyncJobDataHelper;
5859
import org.apache.cloudstack.framework.jobs.AsyncJobDispatcher;
5960
import org.apache.cloudstack.framework.jobs.AsyncJobExecutionContext;
6061
import org.apache.cloudstack.framework.jobs.AsyncJobManager;
@@ -412,21 +413,7 @@ public List<Long> doInTransaction(final TransactionStatus status) {
412413
if (logger.isDebugEnabled()) {
413414
logger.debug("Update db status for job-" + jobId);
414415
}
415-
job.setStatus(jobStatus);
416-
job.setResultCode(resultCode);
417-
418-
if (resultObject != null) {
419-
job.updateResultWithEncryptionIfNeeded(resultObject);
420-
} else {
421-
job.updateResultWithEncryptionIfNeeded(null);
422-
}
423-
424-
final Date currentGMTTime = DateUtil.currentGMTTime();
425-
job.setLastUpdated(currentGMTTime);
426-
job.setCompleteMsid(getMsid());
427-
job.setRemoved(currentGMTTime);
428-
job.setExecutingMsid(null);
429-
_jobDao.update(jobId, job);
416+
recordCompletion(job, jobStatus, resultCode, resultObject);
430417

431418
if (logger.isDebugEnabled()) {
432419
logger.debug("Wake up jobs joined with job-" + jobId + " and disjoin all subjobs created from job- " + jobId);
@@ -457,6 +444,24 @@ public List<Long> doInTransaction(final TransactionStatus status) {
457444
_messageBus.publish(null, AsyncJob.Topics.JOB_STATE, PublishScope.GLOBAL, jobId);
458445
}
459446

447+
void recordCompletion(final AsyncJobVO job, final Status jobStatus, final int resultCode, final String resultObject) {
448+
job.setStatus(jobStatus);
449+
job.setResultCode(resultCode);
450+
if (jobStatus != JobInfo.Status.CANCELLED) {
451+
// a cancel requested after this job's last read did not take: it ended on its own
452+
job.setCancelRequested(false);
453+
job.setCancelDelivered(false);
454+
}
455+
job.updateResultWithEncryptionIfNeeded(resultObject);
456+
457+
final Date currentGMTTime = DateUtil.currentGMTTime();
458+
job.setLastUpdated(currentGMTTime);
459+
job.setCompleteMsid(getMsid());
460+
job.setRemoved(currentGMTTime);
461+
job.setExecutingMsid(null);
462+
_jobDao.update(job.getId(), job);
463+
}
464+
460465
private String convertHumanReadableJson(String resultObj) {
461466

462467
if (resultObj != null && resultObj.contains("/") && resultObj.contains("{")){
@@ -1607,31 +1612,35 @@ static String getJobCancelReasonFromResult(final AsyncJobVO job) {
16071612
}
16081613
}
16091614

1610-
private void requestJobCancel(final AsyncJobVO job, final String cancelledBy) {
1611-
job.setCancelRequested(true);
1612-
job.setCancelDelivered(false);
1613-
job.setResultCode(ApiErrorCode.INTERNAL_ERROR.getHttpCode());
1614-
job.updateResultWithEncryptionIfNeeded(cancelledResult(cancelledBy));
1615-
job.setLastUpdated(DateUtil.currentGMTTime());
1616-
_jobDao.update(job.getId(), job);
1615+
private boolean requestJobCancel(final AsyncJobVO job, final String cancelledBy) {
1616+
final String result = AsyncJobDataHelper.encryptResultIfNeeded(job.getCmd(), cancelledResult(cancelledBy));
1617+
return _jobDao.requestCancel(job.getId(), ApiErrorCode.INTERNAL_ERROR.getHttpCode(), result);
16171618
}
16181619

16191620
private void withdrawJobCancel(final long jobId) {
1620-
final AsyncJobVO job = _jobDao.findById(jobId);
1621-
if (job == null || !job.isCancelRequested() || job.getStatus() != JobInfo.Status.IN_PROGRESS) {
1621+
if (!_jobDao.withdrawCancel(jobId)) {
16221622
return;
16231623
}
1624-
job.setCancelRequested(false);
1625-
job.setCancelDelivered(false);
1626-
job.setResultCode(0);
1627-
job.updateResultWithEncryptionIfNeeded(null);
1628-
job.setLastUpdated(DateUtil.currentGMTTime());
1629-
_jobDao.update(jobId, job);
16301624
for (final JobCancellationHandler handler : getJobCancellationHandlers()) {
16311625
handler.jobCancelWithdrawn(jobId);
16321626
}
16331627
}
16341628

1629+
private CloudRuntimeException refusedByState(final long jobId, final AsyncJobVO job) {
1630+
if (job == null) {
1631+
return refused("Cannot cancel, job-" + jobId + " no longer exists.");
1632+
}
1633+
if (job.isCancelRequested() && job.getStatus() == JobInfo.Status.IN_PROGRESS) {
1634+
final String earlier = getJobCancelReasonFromResult(job);
1635+
return refused("Cannot cancel job-" + jobId + ", it is already being cancelled" + (earlier != null ? " (" + earlier + ")" : "") + ".");
1636+
}
1637+
if (job.getStatus() != null && job.getStatus().done()) {
1638+
_queueMgr.purgeAsyncJobQueueItemId(jobId);
1639+
return refused("Cannot cancel, job-" + jobId + " is not running. Current job status is " + job.getStatus() + ".");
1640+
}
1641+
return null;
1642+
}
1643+
16351644
@Override
16361645
public boolean hasOnlyQueuedWork(final long jobId) {
16371646
final List<AsyncJobVO> children = _jobDao.listChildJobs(jobId);
@@ -1731,9 +1740,9 @@ public void cancelAsyncJob(final long jobId, final String reason) {
17311740
CallContext.current().setEventDetails("Job id: " + job.getUuid() + ". " + cancelledBy);
17321741
}
17331742

1734-
if (job.isCancelRequested() && job.getStatus() == JobInfo.Status.IN_PROGRESS) {
1735-
final String earlier = getJobCancelReasonFromResult(job);
1736-
throw refused("Cannot cancel job-" + jobId + ", it is already being cancelled" + (earlier != null ? " (" + earlier + ")" : "") + ".");
1743+
final CloudRuntimeException stateRefusal = refusedByState(jobId, job);
1744+
if (stateRefusal != null) {
1745+
throw stateRefusal;
17371746
}
17381747

17391748
boolean onlyQueuedWork = false;
@@ -1759,11 +1768,6 @@ public void cancelAsyncJob(final long jobId, final String reason) {
17591768
}
17601769
}
17611770

1762-
if (job.getStatus() != null && job.getStatus().done()) {
1763-
_queueMgr.purgeAsyncJobQueueItemId(jobId);
1764-
throw refused("Cannot cancel, job-" + jobId + " is not running. Current job status is " + job.getStatus() + ".");
1765-
}
1766-
17671771
// ask before changing state: a job whose work cannot be stopped is refused, not recorded as cancelled
17681772
for (final JobCancellationHandler handler : getJobCancellationHandlers()) {
17691773
final String refusal = handler.checkJobCannotBeCancelled(jobId);
@@ -1773,7 +1777,11 @@ public void cancelAsyncJob(final long jobId, final String reason) {
17731777
}
17741778

17751779
logger.info("Cancellation of job-{} requested: {}", jobId, cancelledBy);
1776-
requestJobCancel(job, cancelledBy);
1780+
if (!requestJobCancel(job, cancelledBy)) {
1781+
// the job completed, or another cancel got in, since it was read: its result is left as it is
1782+
final CloudRuntimeException lateRefusal = refusedByState(jobId, _jobDao.findByIdIncludingRemoved(jobId));
1783+
throw lateRefusal != null ? lateRefusal : refused("Cannot cancel job-" + jobId + ", its state changed meanwhile.");
1784+
}
17771785
// the flag stops a queued work job from starting; one taken by a server before the flag is work under way that cannot be stopped
17781786
if (onlyQueuedWork && !hasOnlyQueuedWork(jobId)) {
17791787
withdrawJobCancel(jobId);
@@ -1795,6 +1803,11 @@ public void cancelAsyncJob(final long jobId, final String reason) {
17951803
throw new CloudRuntimeException(errMessage, t);
17961804
}
17971805

1806+
final AsyncJobVO current = _jobDao.findByIdIncludingRemoved(jobId);
1807+
if (current != null && current.getStatus() != null && current.getStatus().done() && current.getStatus() != JobInfo.Status.CANCELLED) {
1808+
throw refused("Cannot cancel job-" + jobId + ", it completed meanwhile. Current job status is " + current.getStatus() + ".");
1809+
}
1810+
17981811
final Long executingMsid = job.getExecutingMsid();
17991812
if (executingMsid == null || executingMsid.equals(getMsid())) {
18001813
finalizeCancelledJob(jobId);

‎framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java‎

Lines changed: 64 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -576,12 +576,13 @@ public void testNotCancellableJobIsRefusedUnlessAllItsWorkIsStillQueued() {
576576
// still waiting in the queue: nothing has started
577577
when(queueItemDao.findById(5L)).thenReturn(queueItem(5L, null));
578578
Mockito.lenient().when(asyncJobDao.findById(10L)).thenReturn(job);
579+
when(asyncJobDao.requestCancel(Mockito.eq(10L), Mockito.anyInt(), Mockito.anyString())).thenReturn(true);
579580
try {
580581
asyncJobManager.cancelAsyncJob(10L, "Cancelled by admin");
581582
} catch (CloudRuntimeException e) {
582583
Assert.assertFalse("Refused although nothing had started: " + e.getMessage(), e.getMessage().contains("is not cancellable"));
583584
}
584-
Assert.assertTrue(job.isCancelRequested());
585+
Mockito.verify(asyncJobDao).requestCancel(Mockito.eq(10L), Mockito.eq(ApiErrorCode.INTERNAL_ERROR.getHttpCode()), Mockito.contains("Cancelled by admin"));
585586
}
586587

587588
@Test
@@ -592,7 +593,7 @@ public void testWorkTakenBetweenTheCheckAndTheFlagWithdrawsTheCancel() {
592593
job.setCmd("org.apache.cloudstack.api.command.user.vm.DeployVMCmd");
593594
job.setExecutingMsid(1L);
594595
when(asyncJobDao.findByIdIncludingRemoved(10L)).thenReturn(job);
595-
when(asyncJobDao.findById(10L)).thenReturn(job);
596+
when(asyncJobDao.requestCancel(Mockito.eq(10L), Mockito.anyInt(), Mockito.anyString())).thenReturn(true);
596597
AsyncJobVO child = childJob(11L, JobInfo.Status.IN_PROGRESS);
597598
when(asyncJobDao.listChildJobs(10L)).thenReturn(Collections.singletonList(child));
598599
when(queueItemDao.getQueueItemIdByContentIdAndType(11L, SyncQueueItem.AsyncJobContentType)).thenReturn(5L);
@@ -605,7 +606,7 @@ public void testWorkTakenBetweenTheCheckAndTheFlagWithdrawsTheCancel() {
605606
} catch (CloudRuntimeException e) {
606607
Assert.assertTrue(e.getMessage(), e.getMessage().contains("started meanwhile"));
607608
}
608-
Assert.assertFalse(job.isCancelRequested());
609+
Mockito.verify(asyncJobDao).withdrawCancel(10L);
609610
}
610611

611612
@Test
@@ -616,7 +617,8 @@ public void testWithdrawnCancelIsAnnouncedToTheHandlers() throws Exception {
616617
job.setCmd("org.apache.cloudstack.api.command.user.vm.DeployVMCmd");
617618
job.setExecutingMsid(1L);
618619
when(asyncJobDao.findByIdIncludingRemoved(10L)).thenReturn(job);
619-
when(asyncJobDao.findById(10L)).thenReturn(job);
620+
when(asyncJobDao.requestCancel(Mockito.eq(10L), Mockito.anyInt(), Mockito.anyString())).thenReturn(true);
621+
when(asyncJobDao.withdrawCancel(10L)).thenReturn(true);
620622
AsyncJobVO child = childJob(11L, JobInfo.Status.IN_PROGRESS);
621623
when(asyncJobDao.listChildJobs(10L)).thenReturn(Collections.singletonList(child));
622624
when(queueItemDao.getQueueItemIdByContentIdAndType(11L, SyncQueueItem.AsyncJobContentType)).thenReturn(5L);
@@ -630,9 +632,65 @@ public void testWithdrawnCancelIsAnnouncedToTheHandlers() throws Exception {
630632
} catch (CloudRuntimeException expected) {
631633
// the cancel was withdrawn
632634
}
635+
Mockito.verify(handler).jobCancelWithdrawn(10L);
636+
}
637+
638+
@Test
639+
public void testCancelOfAJobThatCompletedBeforeTheRequestIsRefusedAndKeepsItsResult() {
640+
AsyncJobVO job = childJob(10L, JobInfo.Status.IN_PROGRESS);
641+
job.setRelated(null);
642+
job.setUuid("job-uuid");
643+
AsyncJobVO completed = childJob(10L, JobInfo.Status.SUCCEEDED);
644+
completed.setRelated(null);
645+
// in progress when read, completed by the time the flag is written
646+
when(asyncJobDao.findByIdIncludingRemoved(10L)).thenReturn(job, completed);
647+
648+
try {
649+
asyncJobManager.cancelAsyncJob(10L, "Cancelled by admin");
650+
Assert.fail("A completed job was cancelled");
651+
} catch (CloudRuntimeException e) {
652+
Assert.assertTrue(e.getMessage(), e.getMessage().contains("is not running. Current job status is SUCCEEDED"));
653+
}
654+
Mockito.verify(asyncJobDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
655+
Mockito.verify(asyncJobDao, Mockito.never()).withdrawCancel(Mockito.anyLong());
656+
}
657+
658+
@Test
659+
public void testCancelOfAJobThatCompletedWhileItsWorkWasBeingStoppedIsRefused() throws Exception {
660+
AsyncJobVO job = childJob(10L, JobInfo.Status.IN_PROGRESS);
661+
job.setRelated(null);
662+
job.setUuid("job-uuid");
663+
AsyncJobVO completed = childJob(10L, JobInfo.Status.SUCCEEDED);
664+
completed.setRelated(null);
665+
when(asyncJobDao.findByIdIncludingRemoved(10L)).thenReturn(job, completed);
666+
when(asyncJobDao.requestCancel(Mockito.eq(10L), Mockito.anyInt(), Mockito.anyString())).thenReturn(true);
667+
JobCancellationHandler handler = Mockito.mock(JobCancellationHandler.class);
668+
when(handler.cancelJobExecution(10L, "Cancelled by admin")).thenReturn(true);
669+
setField("jobCancellationHandlers", Collections.singletonList(handler));
670+
671+
try {
672+
asyncJobManager.cancelAsyncJob(10L, "Cancelled by admin");
673+
Assert.fail("A job that completed meanwhile was cancelled");
674+
} catch (CloudRuntimeException e) {
675+
Assert.assertTrue(e.getMessage(), e.getMessage().contains("completed meanwhile"));
676+
}
677+
Mockito.verify(asyncJobManager, Mockito.never()).finalizeCancelledJob(10L);
678+
}
679+
680+
@Test
681+
public void testCompletionOtherThanCancelledClearsACancelThatDidNotTake() {
682+
AsyncJobVO job = childJob(10L, JobInfo.Status.IN_PROGRESS);
683+
job.setRelated(null);
684+
job.setCancelRequested(true);
685+
job.setCancelDelivered(true);
686+
687+
asyncJobManager.recordCompletion(job, JobInfo.Status.SUCCEEDED, 0, "the volume");
688+
689+
Assert.assertEquals(JobInfo.Status.SUCCEEDED, job.getStatus());
633690
Assert.assertFalse(job.isCancelRequested());
634691
Assert.assertFalse(job.isCancelDelivered());
635-
Mockito.verify(handler).jobCancelWithdrawn(10L);
692+
Assert.assertEquals("the volume", job.getResult());
693+
Mockito.verify(asyncJobDao).update(10L, job);
636694
}
637695

638696
@Test
@@ -671,7 +729,7 @@ public void testSecondCancelOfAJobBeingCancelledIsRefused() {
671729
} catch (CloudRuntimeException e) {
672730
Assert.assertTrue(e.getMessage(), e.getMessage().contains("already being cancelled (Cancelled by admin via mgmt2)"));
673731
}
674-
Mockito.verify(asyncJobDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
732+
Mockito.verify(asyncJobDao, Mockito.never()).requestCancel(Mockito.anyLong(), Mockito.anyInt(), Mockito.any());
675733
}
676734

677735
@Test

0 commit comments

Comments
 (0)