diff --git a/lib/jqueue.py b/lib/jqueue.py index 654c68143a1e4cfa2c3e3615eb288928dd2598b2..1a0c20d2eabbf95b86fd41159c4e00f1fcdaa867 100644 --- a/lib/jqueue.py +++ b/lib/jqueue.py @@ -921,7 +921,14 @@ class _JobProcessor(object): except mcpu.LockAcquireTimeout: assert timeout is not None, "Received timeout for blocking acquire" logging.debug("Couldn't acquire locks in %0.6fs", timeout) - assert op.status == constants.OP_STATUS_WAITLOCK + + assert op.status in (constants.OP_STATUS_WAITLOCK, + constants.OP_STATUS_CANCELING) + + # Was job cancelled while we were waiting for the lock? + if op.status == constants.OP_STATUS_CANCELING: + return (constants.OP_STATUS_CANCELING, None) + return (constants.OP_STATUS_QUEUED, None) except CancelJob: logging.exception("%s: Canceling job", opctx.log_prefix) diff --git a/test/ganeti.jqueue_unittest.py b/test/ganeti.jqueue_unittest.py index dddf7c9f44ad306e9b0fe544b079546f6ae3026a..0939e08a08c7633c975a86f242a5e3b14a91cc38 100755 --- a/test/ganeti.jqueue_unittest.py +++ b/test/ganeti.jqueue_unittest.py @@ -690,6 +690,50 @@ class TestJobProcessor(unittest.TestCase, _JobProcessorTestUtils): [[constants.OP_STATUS_CANCELED for _ in job.ops], ["Job canceled by request" for _ in job.ops]]) + def testCancelWhileWaitlockWithTimeout(self): + queue = _FakeQueueForProc() + + ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) + for i in range(5)] + + # Create job + job_id = 24314 + job = self._CreateJob(queue, job_id, ops) + + self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) + + def _BeforeStart(timeout, priority): + self.assertFalse(queue.IsAcquired()) + self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK) + + # Mark as cancelled + (success, _) = job.Cancel() + self.assert_(success) + + self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING + for op in job.ops)) + + # Fake an acquire attempt timing out + raise mcpu.LockAcquireTimeout() + + def _AfterStart(op, cbs): + self.fail("Should not reach this") + + opexec = _FakeExecOpCodeForProc(_BeforeStart, _AfterStart) + + self.assert_(jqueue._JobProcessor(queue, opexec, job)()) + + # Check result + self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) + self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) + self.assert_(job.start_timestamp) + self.assert_(job.end_timestamp) + self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp + for op in job.ops)) + self.assertEqual(job.GetInfo(["opstatus", "opresult"]), + [[constants.OP_STATUS_CANCELED for _ in job.ops], + ["Job canceled by request" for _ in job.ops]]) + def testCancelWhileRunning(self): # Tests canceling a job with finished opcodes and more, unprocessed ones queue = _FakeQueueForProc()