+ raw_data = utils.ReadFile(filepath)
+ except IOError, err:
+ if err.errno in (errno.ENOENT, ):
+ return None
+ raise
+
+ data = serializer.LoadJson(raw_data)
+
+ try:
+ job = _QueuedJob.Restore(self, data)
+ except Exception, err:
+ new_path = self._GetArchivedJobPath(job_id)
+ if filepath == new_path:
+ # job already archived (future case)
+ logging.exception("Can't parse job %s", job_id)
+ else:
+ # non-archived case
+ logging.exception("Can't parse job %s, will archive.", job_id)
+ self._RenameFilesUnlocked([(filepath, new_path)])
+ return None
+
+ self._memcache[job_id] = job
+ logging.debug("Added job %s to the cache", job_id)
+ return job
+
+ def _GetJobsUnlocked(self, job_ids):
+ """Return a list of jobs based on their IDs.
+
+ @type job_ids: list
+ @param job_ids: either an empty list (meaning all jobs),
+ or a list of job IDs
+ @rtype: list
+ @return: the list of job objects
+
+ """
+ if not job_ids:
+ job_ids = self._GetJobIDsUnlocked()
+
+ return [self._LoadJobUnlocked(job_id) for job_id in job_ids]
+
+ @staticmethod
+ def _IsQueueMarkedDrain():
+ """Check if the queue is marked from drain.
+
+ This currently uses the queue drain file, which makes it a
+ per-node flag. In the future this can be moved to the config file.
+
+ @rtype: boolean
+ @return: True of the job queue is marked for draining
+
+ """
+ return os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
+
+ @staticmethod
+ def SetDrainFlag(drain_flag):
+ """Sets the drain flag for the queue.
+
+ This is similar to the function L{backend.JobQueueSetDrainFlag},
+ and in the future we might merge them.
+
+ @type drain_flag: boolean
+ @param drain_flag: Whether to set or unset the drain flag
+
+ """
+ if drain_flag:
+ utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
+ else:
+ utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
+ return True
+
+ @_RequireOpenQueue
+ def _SubmitJobUnlocked(self, job_id, ops):
+ """Create and store a new job.
+
+ This enters the job into our job queue and also puts it on the new
+ 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
+ @type ops: list
+ @param ops: The list of OpCodes that will become the new job.
+ @rtype: job ID
+ @return: the job ID of the newly created job
+ @raise errors.JobQueueDrainError: if the job is marked for draining
+
+ """
+ if self._IsQueueMarkedDrain():
+ raise errors.JobQueueDrainError("Job queue is drained, refusing job")
+
+ # Check job queue size
+ size = len(self._ListJobFiles())
+ if size >= constants.JOB_QUEUE_SIZE_SOFT_LIMIT:
+ # TODO: Autoarchive jobs. Make sure it's not done on every job
+ # submission, though.
+ #size = ...
+ pass
+
+ if size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
+ raise errors.JobQueueFull()
+
+ job = _QueuedJob(self, job_id, ops)
+
+ # Write to disk
+ self.UpdateJobUnlocked(job)
+
+ logging.debug("Adding new job %s to the cache", job_id)
+ self._memcache[job_id] = job
+
+ # Add to worker pool
+ self._wpool.AddTask(job)
+
+ return job.id
+
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def SubmitJob(self, ops):
+ """Create and store a new job.
+
+ @see: L{_SubmitJobUnlocked}
+
+ """
+ job_id = self._NewSerialsUnlocked(1)[0]
+ return self._SubmitJobUnlocked(job_id, ops)
+
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def SubmitManyJobs(self, jobs):
+ """Create and store multiple jobs.
+
+ @see: L{_SubmitJobUnlocked}
+
+ """
+ results = []
+ all_job_ids = self._NewSerialsUnlocked(len(jobs))
+ for job_id, ops in zip(all_job_ids, jobs):
+ try:
+ data = self._SubmitJobUnlocked(job_id, 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.
+
+ After a job has been modified, this function needs to be called in
+ order to write the changes to disk and replicate them to the other
+ nodes.
+
+ @type job: L{_QueuedJob}
+ @param job: the changed job
+
+ """
+ filename = self._GetJobPath(job.id)
+ data = serializer.DumpJson(job.Serialize(), indent=False)
+ logging.debug("Writing job %s to %s", job.id, filename)
+ self._WriteAndReplicateFileUnlocked(filename, data)
+
+ # Notify waiters about potential changes
+ job.change.notifyAll()
+
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
+ timeout):
+ """Waits for changes in a job.
+
+ @type job_id: string
+ @param job_id: Job identifier
+ @type fields: list of strings
+ @param fields: Which fields to check for changes
+ @type prev_job_info: list or None
+ @param prev_job_info: Last job information returned
+ @type prev_log_serial: int
+ @param prev_log_serial: Last job message serial number
+ @type timeout: float
+ @param timeout: maximum time to wait
+ @rtype: tuple (job info, log entries)
+ @return: a tuple of the job information as required via
+ the fields parameter, and the log entries as a list
+
+ if the job has not changed and the timeout has expired,
+ we instead return a special value,
+ L{constants.JOB_NOTCHANGED}, which should be interpreted
+ as such by the clients
+
+ """
+ logging.debug("Waiting for changes in job %s", job_id)
+
+ job_info = None
+ log_entries = None
+
+ end_time = time.time() + timeout
+ while True:
+ delta_time = end_time - time.time()
+ if delta_time < 0:
+ return constants.JOB_NOTCHANGED
+
+ job = self._LoadJobUnlocked(job_id)
+ if not job:
+ logging.debug("Job %s not found", job_id)
+ break
+
+ status = job.CalcStatus()
+ job_info = self._GetJobInfoUnlocked(job, fields)
+ log_entries = job.GetLogEntries(prev_log_serial)
+
+ # Serializing and deserializing data can cause type changes (e.g. from
+ # tuple to list) or precision loss. We're doing it here so that we get
+ # the same modifications as the data received from the client. Without
+ # this, the comparison afterwards might fail without the data being
+ # significantly different.
+ job_info = serializer.LoadJson(serializer.DumpJson(job_info))
+ log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
+
+ if status not in (constants.JOB_STATUS_QUEUED,
+ constants.JOB_STATUS_RUNNING,
+ constants.JOB_STATUS_WAITLOCK):
+ # Don't even try to wait if the job is no longer running, there will be
+ # no changes.
+ break
+
+ if (prev_job_info != job_info or
+ (log_entries and prev_log_serial != log_entries[0][0])):
+ break
+
+ logging.debug("Waiting again")
+
+ # Release the queue lock while waiting
+ job.change.wait(delta_time)
+
+ logging.debug("Job %s changed", job_id)
+
+ if job_info is None and log_entries is None:
+ return None
+ else:
+ return (job_info, log_entries)
+
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def CancelJob(self, job_id):
+ """Cancels a job.
+
+ This will only succeed if the job has not started yet.
+
+ @type job_id: string
+ @param job_id: job ID of job to be cancelled.
+
+ """
+ logging.info("Cancelling job %s", job_id)
+
+ job = self._LoadJobUnlocked(job_id)
+ if not job:
+ logging.debug("Job %s not found", job_id)
+ return (False, "Job %s not found" % job_id)
+
+ job_status = job.CalcStatus()
+
+ if job_status not in (constants.JOB_STATUS_QUEUED,
+ constants.JOB_STATUS_WAITLOCK):
+ logging.debug("Job %s is no longer waiting in the queue", job.id)
+ return (False, "Job %s is no longer waiting 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:
+ job.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
+ finally:
+ self.UpdateJobUnlocked(job)
+ return (True, "Job %s will be canceled" % job.id)
+
+ @_RequireOpenQueue
+ def CancelJobUnlocked(self, job):
+ """Marks a job as canceled.
+
+ """
+ try:
+ job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
+ "Job canceled by request")