4 # Copyright (C) 2006, 2007, 2008 Google Inc.
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
22 """Module implementing the job queue handling.
24 Locking: there's a single, large lock in the L{JobQueue} class. It's
25 used by all other classes in this module.
27 @var JOBQUEUE_THREADS: the number of worker threads we start for
40 from ganeti import constants
41 from ganeti import serializer
42 from ganeti import workerpool
43 from ganeti import opcodes
44 from ganeti import errors
45 from ganeti import mcpu
46 from ganeti import utils
47 from ganeti import jstore
48 from ganeti import rpc
52 JOBS_PER_ARCHIVE_DIRECTORY = 10000
55 class CancelJob(Exception):
56 """Special exception to cancel a job.
62 """Returns the current timestamp.
65 @return: the current time in the (seconds, microseconds) format
68 return utils.SplitTime(time.time())
71 class _QueuedOpCode(object):
72 """Encapsulates an opcode object.
74 @ivar log: holds the execution log and consists of tuples
75 of the form C{(log_serial, timestamp, level, message)}
76 @ivar input: the OpCode we encapsulate
77 @ivar status: the current status
78 @ivar result: the result of the LU execution
79 @ivar start_timestamp: timestamp for the start of the execution
80 @ivar exec_timestamp: timestamp for the actual LU Exec() function invocation
81 @ivar stop_timestamp: timestamp for the end of the execution
84 __slots__ = ["input", "status", "result", "log",
85 "start_timestamp", "exec_timestamp", "end_timestamp",
88 def __init__(self, op):
89 """Constructor for the _QuededOpCode.
91 @type op: L{opcodes.OpCode}
92 @param op: the opcode we encapsulate
96 self.status = constants.OP_STATUS_QUEUED
99 self.start_timestamp = None
100 self.exec_timestamp = None
101 self.end_timestamp = None
104 def Restore(cls, state):
105 """Restore the _QueuedOpCode from the serialized form.
108 @param state: the serialized state
109 @rtype: _QueuedOpCode
110 @return: a new _QueuedOpCode instance
113 obj = _QueuedOpCode.__new__(cls)
114 obj.input = opcodes.OpCode.LoadOpCode(state["input"])
115 obj.status = state["status"]
116 obj.result = state["result"]
117 obj.log = state["log"]
118 obj.start_timestamp = state.get("start_timestamp", None)
119 obj.exec_timestamp = state.get("exec_timestamp", None)
120 obj.end_timestamp = state.get("end_timestamp", None)
124 """Serializes this _QueuedOpCode.
127 @return: the dictionary holding the serialized state
131 "input": self.input.__getstate__(),
132 "status": self.status,
133 "result": self.result,
135 "start_timestamp": self.start_timestamp,
136 "exec_timestamp": self.exec_timestamp,
137 "end_timestamp": self.end_timestamp,
141 class _QueuedJob(object):
142 """In-memory job representation.
144 This is what we use to track the user-submitted jobs. Locking must
145 be taken care of by users of this class.
147 @type queue: L{JobQueue}
148 @ivar queue: the parent queue
151 @ivar ops: the list of _QueuedOpCode that constitute the job
152 @type log_serial: int
153 @ivar log_serial: holds the index for the next log entry
154 @ivar received_timestamp: the timestamp for when the job was received
155 @ivar start_timestmap: the timestamp for start of execution
156 @ivar end_timestamp: the timestamp for end of execution
157 @ivar lock_status: In-memory locking information for debugging
158 @ivar change: a Condition variable we use for waiting for job changes
161 # pylint: disable-msg=W0212
162 __slots__ = ["queue", "id", "ops", "log_serial",
163 "received_timestamp", "start_timestamp", "end_timestamp",
164 "lock_status", "change",
167 def __init__(self, queue, job_id, ops):
168 """Constructor for the _QueuedJob.
170 @type queue: L{JobQueue}
171 @param queue: our parent queue
173 @param job_id: our job id
175 @param ops: the list of opcodes we hold, which will be encapsulated
180 raise errors.GenericError("A job needs at least one opcode")
184 self.ops = [_QueuedOpCode(op) for op in ops]
186 self.received_timestamp = TimeStampNow()
187 self.start_timestamp = None
188 self.end_timestamp = None
190 # In-memory attributes
191 self.lock_status = None
193 # Condition to wait for changes
194 self.change = threading.Condition(self.queue._lock)
197 status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
199 "ops=%s" % ",".join([op.input.Summary() for op in self.ops])]
201 return "<%s at %#x>" % (" ".join(status), id(self))
204 def Restore(cls, queue, state):
205 """Restore a _QueuedJob from serialized state:
207 @type queue: L{JobQueue}
208 @param queue: to which queue the restored job belongs
210 @param state: the serialized state
212 @return: the restored _JobQueue instance
215 obj = _QueuedJob.__new__(cls)
218 obj.received_timestamp = state.get("received_timestamp", None)
219 obj.start_timestamp = state.get("start_timestamp", None)
220 obj.end_timestamp = state.get("end_timestamp", None)
222 # In-memory attributes
223 obj.lock_status = None
227 for op_state in state["ops"]:
228 op = _QueuedOpCode.Restore(op_state)
229 for log_entry in op.log:
230 obj.log_serial = max(obj.log_serial, log_entry[0])
233 # Condition to wait for changes
234 obj.change = threading.Condition(obj.queue._lock)
239 """Serialize the _JobQueue instance.
242 @return: the serialized state
247 "ops": [op.Serialize() for op in self.ops],
248 "start_timestamp": self.start_timestamp,
249 "end_timestamp": self.end_timestamp,
250 "received_timestamp": self.received_timestamp,
253 def CalcStatus(self):
254 """Compute the status of this job.
256 This function iterates over all the _QueuedOpCodes in the job and
257 based on their status, computes the job status.
260 - if we find a cancelled, or finished with error, the job
261 status will be the same
262 - otherwise, the last opcode with the status one of:
267 will determine the job status
269 - otherwise, it means either all opcodes are queued, or success,
270 and the job status will be the same
272 @return: the job status
275 status = constants.JOB_STATUS_QUEUED
279 if op.status == constants.OP_STATUS_SUCCESS:
284 if op.status == constants.OP_STATUS_QUEUED:
286 elif op.status == constants.OP_STATUS_WAITLOCK:
287 status = constants.JOB_STATUS_WAITLOCK
288 elif op.status == constants.OP_STATUS_RUNNING:
289 status = constants.JOB_STATUS_RUNNING
290 elif op.status == constants.OP_STATUS_CANCELING:
291 status = constants.JOB_STATUS_CANCELING
293 elif op.status == constants.OP_STATUS_ERROR:
294 status = constants.JOB_STATUS_ERROR
295 # The whole job fails if one opcode failed
297 elif op.status == constants.OP_STATUS_CANCELED:
298 status = constants.OP_STATUS_CANCELED
302 status = constants.JOB_STATUS_SUCCESS
306 def GetLogEntries(self, newer_than):
307 """Selectively returns the log entries.
309 @type newer_than: None or int
310 @param newer_than: if this is None, return all log entries,
311 otherwise return only the log entries with serial higher
314 @return: the list of the log entries selected
317 if newer_than is None:
324 entries.extend(filter(lambda entry: entry[0] > serial, op.log))
328 def MarkUnfinishedOps(self, status, result):
329 """Mark unfinished opcodes with a given status and result.
331 This is an utility function for marking all running or waiting to
332 be run opcodes with a given status. Opcodes which are already
333 finalised are not changed.
335 @param status: a given opcode status
336 @param result: the opcode result
341 if op.status in constants.OPS_FINALIZED:
342 assert not_marked, "Finalized opcodes found after non-finalized ones"
349 class _OpExecCallbacks(mcpu.OpExecCbBase):
350 def __init__(self, queue, job, op):
351 """Initializes this class.
353 @type queue: L{JobQueue}
354 @param queue: Job queue
355 @type job: L{_QueuedJob}
356 @param job: Job object
357 @type op: L{_QueuedOpCode}
361 assert queue, "Queue is missing"
362 assert job, "Job is missing"
363 assert op, "Opcode is missing"
369 def NotifyStart(self):
370 """Mark the opcode as running, not lock-waiting.
372 This is called from the mcpu code as a notifier function, when the LU is
373 finally about to start the Exec() method. Of course, to have end-user
374 visible results, the opcode must be initially (before calling into
375 Processor.ExecOpCode) set to OP_STATUS_WAITLOCK.
378 self._queue.acquire()
380 assert self._op.status in (constants.OP_STATUS_WAITLOCK,
381 constants.OP_STATUS_CANCELING)
383 # All locks are acquired by now
384 self._job.lock_status = None
386 # Cancel here if we were asked to
387 if self._op.status == constants.OP_STATUS_CANCELING:
390 self._op.status = constants.OP_STATUS_RUNNING
391 self._op.exec_timestamp = TimeStampNow()
393 self._queue.release()
395 def Feedback(self, *args):
396 """Append a log entry.
402 log_type = constants.ELOG_MESSAGE
405 (log_type, log_msg) = args
407 # The time is split to make serialization easier and not lose
409 timestamp = utils.SplitTime(time.time())
411 self._queue.acquire()
413 self._job.log_serial += 1
414 self._op.log.append((self._job.log_serial, timestamp, log_type, log_msg))
416 self._job.change.notifyAll()
418 self._queue.release()
420 def ReportLocks(self, msg):
421 """Write locking information to the job.
423 Called whenever the LU processor is waiting for a lock or has acquired one.
426 # Not getting the queue lock because this is a single assignment
427 self._job.lock_status = msg
430 class _JobQueueWorker(workerpool.BaseWorker):
431 """The actual job workers.
434 def RunTask(self, job): # pylint: disable-msg=W0221
437 This functions processes a job. It is closely tied to the _QueuedJob and
438 _QueuedOpCode classes.
440 @type job: L{_QueuedJob}
441 @param job: the job to be processed
444 logging.info("Processing job %s", job.id)
445 proc = mcpu.Processor(self.pool.queue.context, job.id)
450 for idx, op in enumerate(job.ops):
451 op_summary = op.input.Summary()
452 if op.status == constants.OP_STATUS_SUCCESS:
453 # this is a job that was partially completed before master
454 # daemon shutdown, so it can be expected that some opcodes
455 # are already completed successfully (if any did error
456 # out, then the whole job should have been aborted and not
457 # resubmitted for processing)
458 logging.info("Op %s/%s: opcode %s already processed, skipping",
459 idx + 1, count, op_summary)
462 logging.info("Op %s/%s: Starting opcode %s", idx + 1, count,
467 if op.status == constants.OP_STATUS_CANCELED:
469 assert op.status == constants.OP_STATUS_QUEUED
470 op.status = constants.OP_STATUS_WAITLOCK
472 op.start_timestamp = TimeStampNow()
473 if idx == 0: # first opcode
474 job.start_timestamp = op.start_timestamp
475 queue.UpdateJobUnlocked(job)
477 input_opcode = op.input
481 # Make sure not to hold queue lock while calling ExecOpCode
482 result = proc.ExecOpCode(input_opcode,
483 _OpExecCallbacks(queue, job, op))
487 op.status = constants.OP_STATUS_SUCCESS
489 op.end_timestamp = TimeStampNow()
490 queue.UpdateJobUnlocked(job)
494 logging.info("Op %s/%s: Successfully finished opcode %s",
495 idx + 1, count, op_summary)
497 # Will be handled further up
499 except Exception, err:
503 op.status = constants.OP_STATUS_ERROR
504 if isinstance(err, errors.GenericError):
505 op.result = errors.EncodeException(err)
508 op.end_timestamp = TimeStampNow()
509 logging.info("Op %s/%s: Error in opcode %s: %s",
510 idx + 1, count, op_summary, err)
512 queue.UpdateJobUnlocked(job)
520 queue.CancelJobUnlocked(job)
523 except errors.GenericError, err:
524 logging.exception("Ganeti exception")
526 logging.exception("Unhandled exception")
531 job.lock_status = None
532 job.end_timestamp = TimeStampNow()
533 queue.UpdateJobUnlocked(job)
536 status = job.CalcStatus()
540 logging.info("Finished job %s, status = %s", job_id, status)
543 class _JobQueueWorkerPool(workerpool.WorkerPool):
544 """Simple class implementing a job-processing workerpool.
547 def __init__(self, queue):
548 super(_JobQueueWorkerPool, self).__init__("JobQueue",
554 def _RequireOpenQueue(fn):
555 """Decorator for "public" functions.
557 This function should be used for all 'public' functions. That is,
558 functions usually called from other classes. Note that this should
559 be applied only to methods (not plain functions), since it expects
560 that the decorated function is called with a first argument that has
561 a '_queue_filelock' argument.
563 @warning: Use this decorator only after utils.LockedMethod!
572 def wrapper(self, *args, **kwargs):
573 # pylint: disable-msg=W0212
574 assert self._queue_filelock is not None, "Queue should be open"
575 return fn(self, *args, **kwargs)
579 class JobQueue(object):
580 """Queue used to manage the jobs.
582 @cvar _RE_JOB_FILE: regex matching the valid job file names
585 _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
587 def __init__(self, context):
588 """Constructor for JobQueue.
590 The constructor will initialize the job queue object and then
591 start loading the current jobs from disk, either for starting them
592 (if they were queue) or for aborting them (if they were already
595 @type context: GanetiContext
596 @param context: the context object for access to the configuration
597 data and other ganeti objects
600 self.context = context
601 self._memcache = weakref.WeakValueDictionary()
602 self._my_hostname = utils.HostInfo().name
605 self._lock = threading.Lock()
606 self.acquire = self._lock.acquire
607 self.release = self._lock.release
609 # Initialize the queue, and acquire the filelock.
610 # This ensures no other process is working on the job queue.
611 self._queue_filelock = jstore.InitAndVerifyQueue(must_lock=True)
614 self._last_serial = jstore.ReadSerial()
615 assert self._last_serial is not None, ("Serial file was modified between"
616 " check in jstore and here")
618 # Get initial list of nodes
619 self._nodes = dict((n.name, n.primary_ip)
620 for n in self.context.cfg.GetAllNodesInfo().values()
621 if n.master_candidate)
625 del self._nodes[self._my_hostname]
629 # TODO: Check consistency across nodes
632 self._UpdateQueueSizeUnlocked()
633 self._drained = self._IsQueueMarkedDrain()
636 self._wpool = _JobQueueWorkerPool(self)
638 # We need to lock here because WorkerPool.AddTask() may start a job while
639 # we're still doing our work.
642 logging.info("Inspecting job queue")
644 all_job_ids = self._GetJobIDsUnlocked()
645 jobs_count = len(all_job_ids)
646 lastinfo = time.time()
647 for idx, job_id in enumerate(all_job_ids):
648 # Give an update every 1000 jobs or 10 seconds
649 if (idx % 1000 == 0 or time.time() >= (lastinfo + 10.0) or
650 idx == (jobs_count - 1)):
651 logging.info("Job queue inspection: %d/%d (%0.1f %%)",
652 idx, jobs_count - 1, 100.0 * (idx + 1) / jobs_count)
653 lastinfo = time.time()
655 job = self._LoadJobUnlocked(job_id)
657 # a failure in loading the job can cause 'None' to be returned
661 status = job.CalcStatus()
663 if status in (constants.JOB_STATUS_QUEUED, ):
664 self._wpool.AddTask(job)
666 elif status in (constants.JOB_STATUS_RUNNING,
667 constants.JOB_STATUS_WAITLOCK,
668 constants.JOB_STATUS_CANCELING):
669 logging.warning("Unfinished job %s found: %s", job.id, job)
671 job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
672 "Unclean master daemon shutdown")
674 self.UpdateJobUnlocked(job)
676 logging.info("Job queue inspection finished")
680 self._wpool.TerminateWorkers()
685 def AddNode(self, node):
686 """Register a new node with the queue.
688 @type node: L{objects.Node}
689 @param node: the node object to be added
692 node_name = node.name
693 assert node_name != self._my_hostname
695 # Clean queue directory on added node
696 result = rpc.RpcRunner.call_jobqueue_purge(node_name)
697 msg = result.fail_msg
699 logging.warning("Cannot cleanup queue directory on node %s: %s",
702 if not node.master_candidate:
703 # remove if existing, ignoring errors
704 self._nodes.pop(node_name, None)
705 # and skip the replication of the job ids
708 # Upload the whole queue excluding archived jobs
709 files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
711 # Upload current serial file
712 files.append(constants.JOB_QUEUE_SERIAL_FILE)
714 for file_name in files:
716 content = utils.ReadFile(file_name)
718 result = rpc.RpcRunner.call_jobqueue_update([node_name],
721 msg = result[node_name].fail_msg
723 logging.error("Failed to upload file %s to node %s: %s",
724 file_name, node_name, msg)
726 self._nodes[node_name] = node.primary_ip
730 def RemoveNode(self, node_name):
731 """Callback called when removing nodes from the cluster.
734 @param node_name: the name of the node to remove
738 # The queue is removed by the "leave node" RPC call.
739 del self._nodes[node_name]
744 def _CheckRpcResult(result, nodes, failmsg):
745 """Verifies the status of an RPC call.
747 Since we aim to keep consistency should this node (the current
748 master) fail, we will log errors if our rpc fail, and especially
749 log the case when more than half of the nodes fails.
751 @param result: the data as returned from the rpc call
753 @param nodes: the list of nodes we made the call to
755 @param failmsg: the identifier to be used for logging
762 msg = result[node].fail_msg
765 logging.error("RPC call %s (%s) failed on node %s: %s",
766 result[node].call, failmsg, node, msg)
770 # +1 for the master node
771 if (len(success) + 1) < len(failed):
772 # TODO: Handle failing nodes
773 logging.error("More than half of the nodes failed")
775 def _GetNodeIp(self):
776 """Helper for returning the node name/ip list.
779 @return: a tuple of two lists, the first one with the node
780 names and the second one with the node addresses
783 name_list = self._nodes.keys()
784 addr_list = [self._nodes[name] for name in name_list]
785 return name_list, addr_list
787 def _WriteAndReplicateFileUnlocked(self, file_name, data):
788 """Writes a file locally and then replicates it to all nodes.
790 This function will replace the contents of a file on the local
791 node and then replicate it to all the other nodes we have.
794 @param file_name: the path of the file to be replicated
796 @param data: the new contents of the file
799 utils.WriteFile(file_name, data=data)
801 names, addrs = self._GetNodeIp()
802 result = rpc.RpcRunner.call_jobqueue_update(names, addrs, file_name, data)
803 self._CheckRpcResult(result, self._nodes,
804 "Updating %s" % file_name)
806 def _RenameFilesUnlocked(self, rename):
807 """Renames a file locally and then replicate the change.
809 This function will rename a file in the local queue directory
810 and then replicate this rename to all the other nodes we have.
812 @type rename: list of (old, new)
813 @param rename: List containing tuples mapping old to new names
816 # Rename them locally
817 for old, new in rename:
818 utils.RenameFile(old, new, mkdir=True)
820 # ... and on all nodes
821 names, addrs = self._GetNodeIp()
822 result = rpc.RpcRunner.call_jobqueue_rename(names, addrs, rename)
823 self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
826 def _FormatJobID(job_id):
827 """Convert a job ID to string format.
829 Currently this just does C{str(job_id)} after performing some
830 checks, but if we want to change the job id format this will
831 abstract this change.
833 @type job_id: int or long
834 @param job_id: the numeric job id
836 @return: the formatted job id
839 if not isinstance(job_id, (int, long)):
840 raise errors.ProgrammerError("Job ID '%s' not numeric" % job_id)
842 raise errors.ProgrammerError("Job ID %s is negative" % job_id)
847 def _GetArchiveDirectory(cls, job_id):
848 """Returns the archive directory for a job.
851 @param job_id: Job identifier
853 @return: Directory name
856 return str(int(job_id) / JOBS_PER_ARCHIVE_DIRECTORY)
858 def _NewSerialsUnlocked(self, count):
859 """Generates a new job identifier.
861 Job identifiers are unique during the lifetime of a cluster.
864 @param count: how many serials to return
866 @return: a string representing the job identifier.
871 serial = self._last_serial + count
874 self._WriteAndReplicateFileUnlocked(constants.JOB_QUEUE_SERIAL_FILE,
877 result = [self._FormatJobID(v)
878 for v in range(self._last_serial, serial + 1)]
879 # Keep it only if we were able to write the file
880 self._last_serial = serial
885 def _GetJobPath(job_id):
886 """Returns the job file for a given job id.
889 @param job_id: the job identifier
891 @return: the path to the job file
894 return utils.PathJoin(constants.QUEUE_DIR, "job-%s" % job_id)
897 def _GetArchivedJobPath(cls, job_id):
898 """Returns the archived job file for a give job id.
901 @param job_id: the job identifier
903 @return: the path to the archived job file
906 return utils.PathJoin(constants.JOB_QUEUE_ARCHIVE_DIR,
907 cls._GetArchiveDirectory(job_id), "job-%s" % job_id)
909 def _GetJobIDsUnlocked(self, sort=True):
910 """Return all known job IDs.
912 The method only looks at disk because it's a requirement that all
913 jobs are present on disk (so in the _memcache we don't have any
917 @param sort: perform sorting on the returned job ids
919 @return: the list of job IDs
923 for filename in utils.ListVisibleFiles(constants.QUEUE_DIR, sort=False):
924 m = self._RE_JOB_FILE.match(filename)
926 jlist.append(m.group(1))
928 jlist = utils.NiceSort(jlist)
931 def _LoadJobUnlocked(self, job_id):
932 """Loads a job from the disk or memory.
934 Given a job id, this will return the cached job object if
935 existing, or try to load the job from the disk. If loading from
936 disk, it will also add the job to the cache.
938 @param job_id: the job id
939 @rtype: L{_QueuedJob} or None
940 @return: either None or the job object
943 job = self._memcache.get(job_id, None)
945 logging.debug("Found job %s in memcache", job_id)
948 filepath = self._GetJobPath(job_id)
949 logging.debug("Loading job from %s", filepath)
951 raw_data = utils.ReadFile(filepath)
953 if err.errno in (errno.ENOENT, ):
957 data = serializer.LoadJson(raw_data)
960 job = _QueuedJob.Restore(self, data)
961 except Exception, err: # pylint: disable-msg=W0703
962 new_path = self._GetArchivedJobPath(job_id)
963 if filepath == new_path:
964 # job already archived (future case)
965 logging.exception("Can't parse job %s", job_id)
968 logging.exception("Can't parse job %s, will archive.", job_id)
969 self._RenameFilesUnlocked([(filepath, new_path)])
972 self._memcache[job_id] = job
973 logging.debug("Added job %s to the cache", job_id)
976 def _GetJobsUnlocked(self, job_ids):
977 """Return a list of jobs based on their IDs.
980 @param job_ids: either an empty list (meaning all jobs),
983 @return: the list of job objects
987 job_ids = self._GetJobIDsUnlocked()
989 return [self._LoadJobUnlocked(job_id) for job_id in job_ids]
992 def _IsQueueMarkedDrain():
993 """Check if the queue is marked from drain.
995 This currently uses the queue drain file, which makes it a
996 per-node flag. In the future this can be moved to the config file.
999 @return: True of the job queue is marked for draining
1002 return os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
1004 def _UpdateQueueSizeUnlocked(self):
1005 """Update the queue size.
1008 self._queue_size = len(self._GetJobIDsUnlocked(sort=False))
1012 def SetDrainFlag(self, drain_flag):
1013 """Sets the drain flag for the queue.
1015 @type drain_flag: boolean
1016 @param drain_flag: Whether to set or unset the drain flag
1020 utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
1022 utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
1024 self._drained = drain_flag
1029 def _SubmitJobUnlocked(self, job_id, ops):
1030 """Create and store a new job.
1032 This enters the job into our job queue and also puts it on the new
1033 queue, in order for it to be picked up by the queue processors.
1035 @type job_id: job ID
1036 @param job_id: the job ID for the new job
1038 @param ops: The list of OpCodes that will become the new job.
1040 @return: the job ID of the newly created job
1041 @raise errors.JobQueueDrainError: if the job is marked for draining
1044 # Ok when sharing the big job queue lock, as the drain file is created when
1045 # the lock is exclusive.
1047 raise errors.JobQueueDrainError("Job queue is drained, refusing job")
1049 if self._queue_size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
1050 raise errors.JobQueueFull()
1052 job = _QueuedJob(self, job_id, ops)
1055 self.UpdateJobUnlocked(job)
1057 self._queue_size += 1
1059 logging.debug("Adding new job %s to the cache", job_id)
1060 self._memcache[job_id] = job
1062 # Add to worker pool
1063 self._wpool.AddTask(job)
1069 def SubmitJob(self, ops):
1070 """Create and store a new job.
1072 @see: L{_SubmitJobUnlocked}
1075 job_id = self._NewSerialsUnlocked(1)[0]
1076 return self._SubmitJobUnlocked(job_id, ops)
1080 def SubmitManyJobs(self, jobs):
1081 """Create and store multiple jobs.
1083 @see: L{_SubmitJobUnlocked}
1087 all_job_ids = self._NewSerialsUnlocked(len(jobs))
1088 for job_id, ops in zip(all_job_ids, jobs):
1090 data = self._SubmitJobUnlocked(job_id, ops)
1092 except errors.GenericError, err:
1095 results.append((status, data))
1100 def UpdateJobUnlocked(self, job):
1101 """Update a job's on disk storage.
1103 After a job has been modified, this function needs to be called in
1104 order to write the changes to disk and replicate them to the other
1107 @type job: L{_QueuedJob}
1108 @param job: the changed job
1111 filename = self._GetJobPath(job.id)
1112 data = serializer.DumpJson(job.Serialize(), indent=False)
1113 logging.debug("Writing job %s to %s", job.id, filename)
1114 self._WriteAndReplicateFileUnlocked(filename, data)
1116 # Notify waiters about potential changes
1117 job.change.notifyAll()
1121 def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
1123 """Waits for changes in a job.
1125 @type job_id: string
1126 @param job_id: Job identifier
1127 @type fields: list of strings
1128 @param fields: Which fields to check for changes
1129 @type prev_job_info: list or None
1130 @param prev_job_info: Last job information returned
1131 @type prev_log_serial: int
1132 @param prev_log_serial: Last job message serial number
1133 @type timeout: float
1134 @param timeout: maximum time to wait
1135 @rtype: tuple (job info, log entries)
1136 @return: a tuple of the job information as required via
1137 the fields parameter, and the log entries as a list
1139 if the job has not changed and the timeout has expired,
1140 we instead return a special value,
1141 L{constants.JOB_NOTCHANGED}, which should be interpreted
1142 as such by the clients
1145 job = self._LoadJobUnlocked(job_id)
1147 logging.debug("Job %s not found", job_id)
1150 def _CheckForChanges():
1151 logging.debug("Waiting for changes in job %s", job_id)
1153 status = job.CalcStatus()
1154 job_info = self._GetJobInfoUnlocked(job, fields)
1155 log_entries = job.GetLogEntries(prev_log_serial)
1157 # Serializing and deserializing data can cause type changes (e.g. from
1158 # tuple to list) or precision loss. We're doing it here so that we get
1159 # the same modifications as the data received from the client. Without
1160 # this, the comparison afterwards might fail without the data being
1161 # significantly different.
1162 job_info = serializer.LoadJson(serializer.DumpJson(job_info))
1163 log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
1165 # Don't even try to wait if the job is no longer running, there will be
1167 if (status not in (constants.JOB_STATUS_QUEUED,
1168 constants.JOB_STATUS_RUNNING,
1169 constants.JOB_STATUS_WAITLOCK) or
1170 prev_job_info != job_info or
1171 (log_entries and prev_log_serial != log_entries[0][0])):
1172 logging.debug("Job %s changed", job_id)
1173 return (job_info, log_entries)
1175 raise utils.RetryAgain()
1178 # Setting wait function to release the queue lock while waiting
1179 return utils.Retry(_CheckForChanges, utils.RETRY_REMAINING_TIME, timeout,
1180 wait_fn=job.change.wait)
1181 except utils.RetryTimeout:
1182 return constants.JOB_NOTCHANGED
1186 def CancelJob(self, job_id):
1189 This will only succeed if the job has not started yet.
1191 @type job_id: string
1192 @param job_id: job ID of job to be cancelled.
1195 logging.info("Cancelling job %s", job_id)
1197 job = self._LoadJobUnlocked(job_id)
1199 logging.debug("Job %s not found", job_id)
1200 return (False, "Job %s not found" % job_id)
1202 job_status = job.CalcStatus()
1204 if job_status not in (constants.JOB_STATUS_QUEUED,
1205 constants.JOB_STATUS_WAITLOCK):
1206 logging.debug("Job %s is no longer waiting in the queue", job.id)
1207 return (False, "Job %s is no longer waiting in the queue" % job.id)
1209 if job_status == constants.JOB_STATUS_QUEUED:
1210 self.CancelJobUnlocked(job)
1211 return (True, "Job %s canceled" % job.id)
1213 elif job_status == constants.JOB_STATUS_WAITLOCK:
1214 # The worker will notice the new status and cancel the job
1216 job.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
1218 self.UpdateJobUnlocked(job)
1219 return (True, "Job %s will be canceled" % job.id)
1222 def CancelJobUnlocked(self, job):
1223 """Marks a job as canceled.
1227 job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
1228 "Job canceled by request")
1230 self.UpdateJobUnlocked(job)
1233 def _ArchiveJobsUnlocked(self, jobs):
1236 @type jobs: list of L{_QueuedJob}
1237 @param jobs: Job objects
1239 @return: Number of archived jobs
1245 if job.CalcStatus() not in (constants.JOB_STATUS_CANCELED,
1246 constants.JOB_STATUS_SUCCESS,
1247 constants.JOB_STATUS_ERROR):
1248 logging.debug("Job %s is not yet done", job.id)
1251 archive_jobs.append(job)
1253 old = self._GetJobPath(job.id)
1254 new = self._GetArchivedJobPath(job.id)
1255 rename_files.append((old, new))
1257 # TODO: What if 1..n files fail to rename?
1258 self._RenameFilesUnlocked(rename_files)
1260 logging.debug("Successfully archived job(s) %s",
1261 utils.CommaJoin(job.id for job in archive_jobs))
1263 # Since we haven't quite checked, above, if we succeeded or failed renaming
1264 # the files, we update the cached queue size from the filesystem. When we
1265 # get around to fix the TODO: above, we can use the number of actually
1266 # archived jobs to fix this.
1267 self._UpdateQueueSizeUnlocked()
1268 return len(archive_jobs)
1272 def ArchiveJob(self, job_id):
1275 This is just a wrapper over L{_ArchiveJobsUnlocked}.
1277 @type job_id: string
1278 @param job_id: Job ID of job to be archived.
1280 @return: Whether job was archived
1283 logging.info("Archiving job %s", job_id)
1285 job = self._LoadJobUnlocked(job_id)
1287 logging.debug("Job %s not found", job_id)
1290 return self._ArchiveJobsUnlocked([job]) == 1
1294 def AutoArchiveJobs(self, age, timeout):
1295 """Archives all jobs based on age.
1297 The method will archive all jobs which are older than the age
1298 parameter. For jobs that don't have an end timestamp, the start
1299 timestamp will be considered. The special '-1' age will cause
1300 archival of all jobs (that are not running or queued).
1303 @param age: the minimum age in seconds
1306 logging.info("Archiving jobs with age more than %s seconds", age)
1309 end_time = now + timeout
1313 all_job_ids = self._GetJobIDsUnlocked()
1315 for idx, job_id in enumerate(all_job_ids):
1316 last_touched = idx + 1
1318 # Not optimal because jobs could be pending
1319 # TODO: Measure average duration for job archival and take number of
1320 # pending jobs into account.
1321 if time.time() > end_time:
1324 # Returns None if the job failed to load
1325 job = self._LoadJobUnlocked(job_id)
1327 if job.end_timestamp is None:
1328 if job.start_timestamp is None:
1329 job_age = job.received_timestamp
1331 job_age = job.start_timestamp
1333 job_age = job.end_timestamp
1335 if age == -1 or now - job_age[0] > age:
1338 # Archive 10 jobs at a time
1339 if len(pending) >= 10:
1340 archived_count += self._ArchiveJobsUnlocked(pending)
1344 archived_count += self._ArchiveJobsUnlocked(pending)
1346 return (archived_count, len(all_job_ids) - last_touched)
1349 def _GetJobInfoUnlocked(job, fields):
1350 """Returns information about a job.
1352 @type job: L{_QueuedJob}
1353 @param job: the job which we query
1355 @param fields: names of fields to return
1357 @return: list with one element for each field
1358 @raise errors.OpExecError: when an invalid field
1363 for fname in fields:
1366 elif fname == "status":
1367 row.append(job.CalcStatus())
1368 elif fname == "ops":
1369 row.append([op.input.__getstate__() for op in job.ops])
1370 elif fname == "opresult":
1371 row.append([op.result for op in job.ops])
1372 elif fname == "opstatus":
1373 row.append([op.status for op in job.ops])
1374 elif fname == "oplog":
1375 row.append([op.log for op in job.ops])
1376 elif fname == "opstart":
1377 row.append([op.start_timestamp for op in job.ops])
1378 elif fname == "opexec":
1379 row.append([op.exec_timestamp for op in job.ops])
1380 elif fname == "opend":
1381 row.append([op.end_timestamp for op in job.ops])
1382 elif fname == "received_ts":
1383 row.append(job.received_timestamp)
1384 elif fname == "start_ts":
1385 row.append(job.start_timestamp)
1386 elif fname == "end_ts":
1387 row.append(job.end_timestamp)
1388 elif fname == "lock_status":
1389 row.append(job.lock_status)
1390 elif fname == "summary":
1391 row.append([op.input.Summary() for op in job.ops])
1393 raise errors.OpExecError("Invalid job query field '%s'" % fname)
1398 def QueryJobs(self, job_ids, fields):
1399 """Returns a list of jobs in queue.
1401 This is a wrapper of L{_GetJobsUnlocked}, which actually does the
1402 processing for each job.
1405 @param job_ids: sequence of job identifiers or None for all
1407 @param fields: names of fields to return
1409 @return: list one element per job, each element being list with
1410 the requested fields
1415 for job in self._GetJobsUnlocked(job_ids):
1419 jobs.append(self._GetJobInfoUnlocked(job, fields))
1426 """Stops the job queue.
1428 This shutdowns all the worker threads an closes the queue.
1431 self._wpool.TerminateWorkers()
1433 self._queue_filelock.Close()
1434 self._queue_filelock = None