- return data
-
- def SubmitJob(self, job):
- """Submit a job"""
- result = self.SendRequest(REQ_SUBMIT, SerializeJob(job))
- if not result['success']:
- raise RequestError(result['result'])
- return result['result']
-
- def Query(self, data):
- """Make a query"""
- result = self.SendRequest(REQ_QUERY, data)
- if not result['success']:
- raise RequestError(result[result])
- result = result['result']
- if data["object"] == "jobs":
- # custom job processing of query values
- for row in result:
- for idx, field in enumerate(data["fields"]):
- if field == "op_list":
- row[idx] = [opcodes.OpCode.LoadOpCode(i) for i in row[idx]]
+
+ result = data[KEY_RESULT]
+
+ if not data[KEY_SUCCESS]:
+ # TODO: decide on a standard exception
+ if (isinstance(result, (tuple, list)) and len(result) == 2 and
+ isinstance(result[1], (tuple, list))):
+ # custom ganeti errors
+ err_class = errors.GetErrorClass(result[0])
+ if err_class is not None:
+ raise err_class, tuple(result[1])
+
+ raise RequestError(result)
+
+ return result
+
+ def SetQueueDrainFlag(self, drain_flag):
+ return self.CallMethod(REQ_QUEUE_SET_DRAIN_FLAG, drain_flag)
+
+ def SubmitJob(self, ops):
+ ops_state = map(lambda op: op.__getstate__(), ops)
+ return self.CallMethod(REQ_SUBMIT_JOB, ops_state)
+
+ def SubmitManyJobs(self, jobs):
+ jobs_state = []
+ for ops in jobs:
+ jobs_state.append([op.__getstate__() for op in ops])
+ return self.CallMethod(REQ_SUBMIT_MANY_JOBS, jobs_state)
+
+ def CancelJob(self, job_id):
+ return self.CallMethod(REQ_CANCEL_JOB, job_id)
+
+ def ArchiveJob(self, job_id):
+ return self.CallMethod(REQ_ARCHIVE_JOB, job_id)
+
+ def AutoArchiveJobs(self, age):
+ timeout = (DEF_RWTO - 1) / 2
+ return self.CallMethod(REQ_AUTOARCHIVE_JOBS, (age, timeout))
+
+ def WaitForJobChange(self, job_id, fields, prev_job_info, prev_log_serial):
+ timeout = (DEF_RWTO - 1) / 2
+ while True:
+ result = self.CallMethod(REQ_WAIT_FOR_JOB_CHANGE,
+ (job_id, fields, prev_job_info,
+ prev_log_serial, timeout))
+ if result != constants.JOB_NOTCHANGED:
+ break