from ganeti import jstore
from ganeti import rpc
+
JOBQUEUE_THREADS = 25
+class CancelJob:
+ """Special exception to cancel a job.
+
+ """
+
+
def TimeStampNow():
"""Returns the current timestamp.
status will be the same
- otherwise, the last opcode with the status one of:
- waitlock
+ - canceling
- running
will determine the job status
status = constants.JOB_STATUS_WAITLOCK
elif op.status == constants.OP_STATUS_RUNNING:
status = constants.JOB_STATUS_RUNNING
+ elif op.status == constants.OP_STATUS_CANCELING:
+ status = constants.JOB_STATUS_CANCELING
+ break
elif op.status == constants.OP_STATUS_ERROR:
status = constants.JOB_STATUS_ERROR
# The whole job fails if one opcode failed
self.queue.acquire()
try:
+ assert self.opcode.status in (constants.OP_STATUS_WAITLOCK,
+ constants.OP_STATUS_CANCELING)
+
+ # Cancel here if we were asked to
+ if self.opcode.status == constants.OP_STATUS_CANCELING:
+ raise CancelJob()
+
self.opcode.status = constants.OP_STATUS_RUNNING
finally:
self.queue.release()
queue.acquire()
try:
+ assert op.status == constants.OP_STATUS_QUEUED
job.run_op_index = idx
op.status = constants.OP_STATUS_WAITLOCK
op.result = None
logging.debug("Op %s/%s: Successfully finished %s",
idx + 1, count, op)
+ except CancelJob:
+ # Will be handled further up
+ raise
except Exception, err:
queue.acquire()
try:
queue.release()
raise
+ except CancelJob:
+ queue.acquire()
+ try:
+ queue.CancelJobUnlocked(job)
+ finally:
+ queue.release()
except errors.GenericError, err:
logging.exception("Ganeti exception")
except:
# Get initial list of nodes
self._nodes = dict((n.name, n.primary_ip)
- for n in self.context.cfg.GetAllNodesInfo().values())
+ for n in self.context.cfg.GetAllNodesInfo().values()
+ if n.master_candidate)
# Remove master node
try:
del self._nodes[self._my_hostname]
- except ValueError:
+ except KeyError:
pass
# TODO: Check consistency across nodes
logging.info("Inspecting job queue")
all_job_ids = self._GetJobIDsUnlocked()
+ jobs_count = len(all_job_ids)
lastinfo = time.time()
for idx, job_id in enumerate(all_job_ids):
# Give an update every 1000 jobs or 10 seconds
- if idx % 1000 == 0 or time.time() >= (lastinfo + 10.0):
- jobs_count = len(all_job_ids)
+ if (idx % 1000 == 0 or time.time() >= (lastinfo + 10.0) or
+ idx == (jobs_count - 1)):
logging.info("Job queue inspection: %d/%d (%0.1f %%)",
- idx, jobs_count, 100.0 * (idx + 1) / jobs_count)
+ idx, jobs_count - 1, 100.0 * (idx + 1) / jobs_count)
lastinfo = time.time()
job = self._LoadJobUnlocked(job_id)
self._wpool.AddTask(job)
elif status in (constants.JOB_STATUS_RUNNING,
- constants.JOB_STATUS_WAITLOCK):
+ constants.JOB_STATUS_WAITLOCK,
+ constants.JOB_STATUS_CANCELING):
logging.warning("Unfinished job %s found: %s", job.id, job)
try:
for op in job.ops:
# Clean queue directory on added node
rpc.RpcRunner.call_jobqueue_purge(node_name)
+ if not node.master_candidate:
+ # remove if existing, ignoring errors
+ self._nodes.pop(node_name, None)
+ # and skip the replication of the job ids
+ return
+
# Upload the whole queue excluding archived jobs
files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
@param job_id: job ID of job to be cancelled.
"""
- logging.debug("Cancelling job %s", job_id)
+ logging.info("Cancelling job %s", job_id)
job = self._LoadJobUnlocked(job_id)
if not job:
logging.debug("Job %s not found", job_id)
- return
+ return (False, "Job %s not found" % job_id)
+
+ job_status = job.CalcStatus()
- if job.CalcStatus() not in (constants.JOB_STATUS_QUEUED,):
+ if job_status not in (constants.JOB_STATUS_QUEUED,
+ constants.JOB_STATUS_WAITLOCK):
logging.debug("Job %s is no longer in the queue", job.id)
- return
+ return (False, "Job %s is no longer in the queue" % job.id)
+
+ if job_status == constants.JOB_STATUS_QUEUED:
+ self.CancelJobUnlocked(job)
+ return (True, "Job %s canceled" % job.id)
+ elif job_status == constants.JOB_STATUS_WAITLOCK:
+ # The worker will notice the new status and cancel the job
+ try:
+ for op in job.ops:
+ op.status = constants.OP_STATUS_CANCELING
+ finally:
+ self.UpdateJobUnlocked(job)
+ return (True, "Job %s will be canceled" % job.id)
+
+ @_RequireOpenQueue
+ def CancelJobUnlocked(self, job):
+ """Marks a job as canceled.
+
+ """
try:
for op in job.ops:
op.status = constants.OP_STATUS_ERROR
- op.result = "Job cancelled by request"
+ op.result = "Job canceled by request"
finally:
self.UpdateJobUnlocked(job)