+ 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.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
+ op.status = constants.OP_STATUS_WAITLOCK
+ op.result = None
+ op.start_timestamp = TimeStampNow()
+ if idx == 0: # first opcode
+ job.start_timestamp = op.start_timestamp
+ queue.UpdateJobUnlocked(job)
+
+ input_opcode = op.input
+ finally:
+ queue.release()
+
+ # Make sure not to hold queue lock while calling ExecOpCode
+ result = proc.ExecOpCode(input_opcode,
+ _OpExecCallbacks(queue, job, op))
+
+ queue.acquire()
+ try:
+ op.status = constants.OP_STATUS_SUCCESS
+ op.result = result
+ op.end_timestamp = TimeStampNow()
+ queue.UpdateJobUnlocked(job)
+ finally:
+ queue.release()
+
+ logging.info("Op %s/%s: Successfully finished opcode %s",
+ idx + 1, count, op_summary)
+ except CancelJob:
+ # Will be handled further up
+ raise
+ except Exception, err:
+ queue.acquire()
+ try:
+ try:
+ op.status = constants.OP_STATUS_ERROR
+ if isinstance(err, errors.GenericError):
+ op.result = errors.EncodeException(err)
+ else:
+ op.result = str(err)
+ op.end_timestamp = TimeStampNow()
+ logging.info("Op %s/%s: Error in opcode %s: %s",
+ idx + 1, count, op_summary, err)
+ finally:
+ queue.UpdateJobUnlocked(job)
+ finally:
+ queue.release()
+ raise
+
+ except CancelJob:
+ queue.acquire()
+ try:
+ queue.CancelJobUnlocked(job)
+ finally:
+ queue.release()
+ except errors.GenericError, err:
+ logging.exception("Ganeti exception")
+ except:
+ logging.exception("Unhandled exception")