class _QueuedOpCode(object):
- """Encasulates an opcode object.
+ """Encapsulates an opcode object.
@ivar log: holds the execution log and consists of tuples
of the form C{(log_serial, timestamp, level, message)}
@ivar stop_timestamp: timestamp for the end of the execution
"""
+ __slots__ = ["input", "status", "result", "log",
+ "start_timestamp", "end_timestamp",
+ "__weakref__"]
+
def __init__(self, op):
"""Constructor for the _QuededOpCode.
@ivar change: a Condition variable we use for waiting for job changes
"""
+ __slots__ = ["queue", "id", "ops", "run_op_index", "log_serial",
+ "received_timestamp", "start_timestamp", "end_timestamp",
+ "change",
+ "__weakref__"]
+
def __init__(self, queue, job_id, ops):
"""Constructor for the _QueuedJob.
"""Selectively returns the log entries.
@type newer_than: None or int
- @param newer_than: if this is None, return all log enties,
+ @param newer_than: if this is None, return all log entries,
otherwise return only the log entries with serial higher
than this value
@rtype: list
@param job: the job to be processed
"""
- logging.debug("Worker %s processing job %s",
+ logging.info("Worker %s processing job %s",
self.worker_id, job.id)
proc = mcpu.Processor(self.pool.queue.context)
self.queue = queue = job.queue
try:
count = len(job.ops)
for idx, op in enumerate(job.ops):
+ op_summary = op.input.Summary()
+ if op.status == constants.OP_STATUS_SUCCESS:
+ # this is a job that was partially completed before master
+ # daemon shutdown, so it can be expected that some opcodes
+ # are already completed successfully (if any did error
+ # out, then the whole job should have been aborted and not
+ # resubmitted for processing)
+ logging.info("Op %s/%s: opcode %s already processed, skipping",
+ idx + 1, count, op_summary)
+ continue
try:
- logging.debug("Op %s/%s: Starting %s", idx + 1, count, op)
+ logging.info("Op %s/%s: Starting opcode %s", idx + 1, count,
+ op_summary)
queue.acquire()
try:
+ if op.status == constants.OP_STATUS_CANCELED:
+ raise CancelJob()
assert op.status == constants.OP_STATUS_QUEUED
job.run_op_index = idx
op.status = constants.OP_STATUS_WAITLOCK
finally:
queue.release()
- logging.debug("Op %s/%s: Successfully finished %s",
- idx + 1, count, op)
+ logging.info("Op %s/%s: Successfully finished opcode %s",
+ idx + 1, count, op_summary)
except CancelJob:
# Will be handled further up
raise
op.status = constants.OP_STATUS_ERROR
op.result = str(err)
op.end_timestamp = TimeStampNow()
- logging.debug("Op %s/%s: Error in %s", idx + 1, count, op)
+ logging.info("Op %s/%s: Error in opcode %s: %s",
+ idx + 1, count, op_summary, err)
finally:
queue.UpdateJobUnlocked(job)
finally:
queue.acquire()
try:
try:
- job.run_op_idx = -1
+ job.run_op_index = -1
job.end_timestamp = TimeStampNow()
queue.UpdateJobUnlocked(job)
finally:
status = job.CalcStatus()
finally:
queue.release()
- logging.debug("Worker %s finished job %s, status = %s",
- self.worker_id, job_id, status)
+ logging.info("Worker %s finished job %s, status = %s",
+ self.worker_id, job_id, status)
class _JobQueueWorkerPool(workerpool.WorkerPool):
class JobQueue(object):
- """Quue used to manaage the jobs.
+ """Queue used to manage the jobs.
@cvar _RE_JOB_FILE: regex matching the valid job file names
Since we aim to keep consistency should this node (the current
master) fail, we will log errors if our rpc fail, and especially
- log the case when more than half of the nodes failes.
+ log the case when more than half of the nodes fails.
@param result: the data as returned from the rpc call
@type nodes: list
@param rename: List containing tuples mapping old to new names
"""
+ # Rename them locally
for old, new in rename:
utils.RenameFile(old, new, mkdir=True)
- names, addrs = self._GetNodeIp()
- result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, old, new)
- self._CheckRpcResult(result, self._nodes,
- "Moving %s to %s" % (old, new))
+ # ... and on all nodes
+ names, addrs = self._GetNodeIp()
+ result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
+ self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
def _FormatJobID(self, job_id):
"""Convert a job ID to string format.
and in the future we might merge them.
@type drain_flag: boolean
- @param drain_flag: wheter to set or unset the drain flag
+ @param drain_flag: Whether to set or unset the drain flag
"""
if drain_flag:
utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
return True
- @utils.LockedMethod
@_RequireOpenQueue
- def SubmitJob(self, ops):
+ def _SubmitJobUnlocked(self, ops):
"""Create and store a new job.
This enters the job into our job queue and also puts it on the new
"""
if self._IsQueueMarkedDrain():
- raise errors.JobQueueDrainError()
+ raise errors.JobQueueDrainError("Job queue is drained, refusing job")
# Check job queue size
size = len(self._ListJobFiles())
return job.id
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def SubmitJob(self, ops):
+ """Create and store a new job.
+
+ @see: L{_SubmitJobUnlocked}
+
+ """
+ return self._SubmitJobUnlocked(ops)
+
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def SubmitManyJobs(self, jobs):
+ """Create and store multiple jobs.
+
+ @see: L{_SubmitJobUnlocked}
+
+ """
+ results = []
+ for ops in jobs:
+ try:
+ data = self._SubmitJobUnlocked(ops)
+ status = True
+ except errors.GenericError, err:
+ data = str(err)
+ status = False
+ results.append((status, data))
+
+ return results
+
+
@_RequireOpenQueue
def UpdateJobUnlocked(self, job):
"""Update a job's on disk storage.
"""
try:
for op in job.ops:
- op.status = constants.OP_STATUS_ERROR
+ op.status = constants.OP_STATUS_CANCELED
op.result = "Job canceled by request"
finally:
self.UpdateJobUnlocked(job)
"""Archives jobs.
@type jobs: list of L{_QueuedJob}
- @param job: Job objects
+ @param jobs: Job objects
@rtype: int
@return: Number of archived jobs
def ArchiveJob(self, job_id):
"""Archives a job.
- This is just a wrapper over L{_ArchiveJobUnlocked}.
+ This is just a wrapper over L{_ArchiveJobsUnlocked}.
@type job_id: string
@param job_id: Job ID of job to be archived.
logging.debug("Job %s not found", job_id)
return False
- return self._ArchiveJobUnlocked([job]) == 1
+ return self._ArchiveJobsUnlocked([job]) == 1
@utils.LockedMethod
@_RequireOpenQueue
projects / ganeti-local / blobdiff