@ivar id: the job ID
@type ops: list
@ivar ops: the list of _QueuedOpCode that constitute the job
- @type run_op_index: int
- @ivar run_op_index: the currently executing opcode, or -1 if
- we didn't yet start executing
@type log_serial: int
@ivar log_serial: holds the index for the next log entry
@ivar received_timestamp: the timestamp for when the job was received
@ivar change: a Condition variable we use for waiting for job changes
"""
- __slots__ = ["queue", "id", "ops", "run_op_index", "log_serial",
+ __slots__ = ["queue", "id", "ops", "log_serial",
"received_timestamp", "start_timestamp", "end_timestamp",
"lock_status", "change",
"__weakref__"]
self.queue = queue
self.id = job_id
self.ops = [_QueuedOpCode(op) for op in ops]
- self.run_op_index = -1
self.log_serial = 0
self.received_timestamp = TimeStampNow()
self.start_timestamp = None
obj = _QueuedJob.__new__(cls)
obj.queue = queue
obj.id = state["id"]
- obj.run_op_index = state["run_op_index"]
obj.received_timestamp = state.get("received_timestamp", None)
obj.start_timestamp = state.get("start_timestamp", None)
obj.end_timestamp = state.get("end_timestamp", None)
return {
"id": self.id,
"ops": [op.Serialize() for op in self.ops],
- "run_op_index": self.run_op_index,
"start_timestamp": self.start_timestamp,
"end_timestamp": self.end_timestamp,
"received_timestamp": self.received_timestamp,
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
op.result = None
op.start_timestamp = TimeStampNow()
try:
try:
job.lock_status = None
- job.run_op_index = -1
job.end_timestamp = TimeStampNow()
queue.UpdateJobUnlocked(job)
finally:
# Clean queue directory on added node
result = rpc.RpcRunner.call_jobqueue_purge(node_name)
- msg = result.RemoteFailMsg()
+ msg = result.fail_msg
if msg:
logging.warning("Cannot cleanup queue directory on node %s: %s",
node_name, msg)
result = rpc.RpcRunner.call_jobqueue_update([node_name],
[node.primary_ip],
file_name, content)
- msg = result[node_name].RemoteFailMsg()
+ msg = result[node_name].fail_msg
if msg:
logging.error("Failed to upload file %s to node %s: %s",
file_name, node_name, msg)
success = []
for node in nodes:
- msg = result[node].RemoteFailMsg()
+ msg = result[node].fail_msg
if msg:
failed.append(node)
logging.error("RPC call %s failed on node %s: %s",
queue, in order for it to be picked up by the queue processors.
@type job_id: job ID
- @param jod_id: the job ID for the new job
+ @param job_id: the job ID for the new job
@type ops: list
@param ops: The list of OpCodes that will become the new job.
@rtype: job ID