+ Job identifiers are unique during the lifetime of a cluster.
+
+ @rtype: str
+ @return: a string representing the job identifier.
+
+ """
+ # New number
+ serial = self._last_serial + 1
+
+ # Write to file
+ self._WriteAndReplicateFileUnlocked(constants.JOB_QUEUE_SERIAL_FILE,
+ "%s\n" % serial)
+
+ # Keep it only if we were able to write the file
+ self._last_serial = serial
+
+ return self._FormatJobID(serial)
+
+ @staticmethod
+ def _GetJobPath(job_id):
+ """Returns the job file for a given job id.
+
+ @type job_id: str
+ @param job_id: the job identifier
+ @rtype: str
+ @return: the path to the job file
+
+ """
+ return os.path.join(constants.QUEUE_DIR, "job-%s" % job_id)
+
+ @staticmethod
+ def _GetArchivedJobPath(job_id):
+ """Returns the archived job file for a give job id.
+
+ @type job_id: str
+ @param job_id: the job identifier
+ @rtype: str
+ @return: the path to the archived job file
+
+ """
+ return os.path.join(constants.JOB_QUEUE_ARCHIVE_DIR, "job-%s" % job_id)
+
+ @classmethod
+ def _ExtractJobID(cls, name):
+ """Extract the job id from a filename.
+
+ @type name: str
+ @param name: the job filename
+ @rtype: job id or None
+ @return: the job id corresponding to the given filename,
+ or None if the filename does not represent a valid
+ job file
+
+ """
+ m = cls._RE_JOB_FILE.match(name)
+ if m:
+ return m.group(1)
+ else:
+ return None
+
+ def _GetJobIDsUnlocked(self, archived=False):
+ """Return all known job IDs.
+
+ If the parameter archived is True, archived jobs IDs will be
+ included. Currently this argument is unused.
+
+ The method only looks at disk because it's a requirement that all
+ jobs are present on disk (so in the _memcache we don't have any
+ extra IDs).
+
+ @rtype: list
+ @return: the list of job IDs
+
+ """
+ jlist = [self._ExtractJobID(name) for name in self._ListJobFiles()]
+ jlist = utils.NiceSort(jlist)
+ return jlist
+
+ def _ListJobFiles(self):
+ """Returns the list of current job files.
+
+ @rtype: list
+ @return: the list of job file names
+
+ """
+ return [name for name in utils.ListVisibleFiles(constants.QUEUE_DIR)
+ if self._RE_JOB_FILE.match(name)]
+
+ def _LoadJobUnlocked(self, job_id):
+ """Loads a job from the disk or memory.
+
+ Given a job id, this will return the cached job object if
+ existing, or try to load the job from the disk. If loading from
+ disk, it will also add the job to the cache.
+
+ @param job_id: the job id
+ @rtype: L{_QueuedJob} or None
+ @return: either None or the job object
+
+ """
+ job = self._memcache.get(job_id, None)
+ if job:
+ logging.debug("Found job %s in memcache", job_id)
+ return job
+
+ filepath = self._GetJobPath(job_id)
+ logging.debug("Loading job from %s", filepath)
+ try:
+ fd = open(filepath, "r")
+ except IOError, err:
+ if err.errno in (errno.ENOENT, ):
+ return None
+ raise
+ try:
+ data = serializer.LoadJson(fd.read())
+ finally:
+ fd.close()
+
+ 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._RenameFileUnlocked(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: wheter to set or unset the drain flag