#
#
-# Copyright (C) 2006, 2007 Google Inc.
+# Copyright (C) 2006, 2007, 2008 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# 02110-1301, USA.
-"""Module implementing the job queue handling."""
+"""Module implementing the job queue handling.
+
+Locking:
+There's a single, large lock in the JobQueue class. It's used by all other
+classes in this module.
+
+"""
import os
import logging
import errno
import re
import time
+import weakref
from ganeti import constants
from ganeti import serializer
from ganeti import errors
from ganeti import mcpu
from ganeti import utils
+from ganeti import jstore
from ganeti import rpc
JOBQUEUE_THREADS = 5
+def TimeStampNow():
+ return utils.SplitTime(time.time())
+
+
class _QueuedOpCode(object):
"""Encasulates an opcode object.
- Access is synchronized by the '_lock' attribute.
-
The 'log' attribute holds the execution log and consists of tuples
- of the form (timestamp, level, message).
+ of the form (log_serial, timestamp, level, message).
"""
- def __new__(cls, *args, **kwargs):
- obj = object.__new__(cls, *args, **kwargs)
- # Create a special lock for logging
- obj._log_lock = threading.Lock()
- return obj
-
def __init__(self, op):
self.input = op
self.status = constants.OP_STATUS_QUEUED
self.result = None
self.log = []
+ self.start_timestamp = None
+ self.end_timestamp = None
@classmethod
def Restore(cls, state):
obj.status = state["status"]
obj.result = state["result"]
obj.log = state["log"]
+ obj.start_timestamp = state.get("start_timestamp", None)
+ obj.end_timestamp = state.get("end_timestamp", None)
return obj
def Serialize(self):
- self._log_lock.acquire()
- try:
- return {
- "input": self.input.__getstate__(),
- "status": self.status,
- "result": self.result,
- "log": self.log,
- }
- finally:
- self._log_lock.release()
-
- def Log(self, *args):
- """Append a log entry.
-
- """
- assert len(args) < 3
-
- if len(args) == 1:
- log_type = constants.ELOG_MESSAGE
- log_msg = args[0]
- else:
- log_type, log_msg = args
-
- self._log_lock.acquire()
- try:
- self.log.append((time.time(), log_type, log_msg))
- finally:
- self._log_lock.release()
-
- def RetrieveLog(self, start_at=0):
- """Retrieve (a part of) the execution log.
-
- """
- self._log_lock.acquire()
- try:
- return self.log[start_at:]
- finally:
- self._log_lock.release()
+ return {
+ "input": self.input.__getstate__(),
+ "status": self.status,
+ "result": self.result,
+ "log": self.log,
+ "start_timestamp": self.start_timestamp,
+ "end_timestamp": self.end_timestamp,
+ }
class _QueuedJob(object):
"""In-memory job representation.
- This is what we use to track the user-submitted jobs.
+ This is what we use to track the user-submitted jobs. Locking must be taken
+ care of by users of this class.
"""
def __init__(self, queue, job_id, ops):
self.id = job_id
self.ops = [_QueuedOpCode(op) for op in ops]
self.run_op_index = -1
+ self.log_serial = 0
+
+ # Condition to wait for changes
+ self.change = threading.Condition(self.queue._lock)
@classmethod
def Restore(cls, queue, state):
obj = _QueuedJob.__new__(cls)
obj.queue = queue
obj.id = state["id"]
- obj.ops = [_QueuedOpCode.Restore(op_state) for op_state in state["ops"]]
obj.run_op_index = state["run_op_index"]
+
+ obj.ops = []
+ obj.log_serial = 0
+ for op_state in state["ops"]:
+ op = _QueuedOpCode.Restore(op_state)
+ for log_entry in op.log:
+ obj.log_serial = max(obj.log_serial, log_entry[0])
+ obj.ops.append(op)
+
+ # Condition to wait for changes
+ obj.change = threading.Condition(obj.queue._lock)
+
return obj
def Serialize(self):
return status
+ def GetLogEntries(self, newer_than):
+ if newer_than is None:
+ serial = -1
+ else:
+ serial = newer_than
+
+ entries = []
+ for op in self.ops:
+ entries.extend(filter(lambda entry: entry[0] > newer_than, op.log))
+
+ return entries
+
class _JobQueueWorker(workerpool.BaseWorker):
def RunTask(self, job):
"""Job executor.
- This functions processes a job.
+ This functions processes a job. It is closely tied to the _QueuedJob and
+ _QueuedOpCode classes.
"""
logging.debug("Worker %s processing job %s",
job.run_op_index = idx
op.status = constants.OP_STATUS_RUNNING
op.result = None
+ op.start_timestamp = TimeStampNow()
queue.UpdateJobUnlocked(job)
input_opcode = op.input
finally:
queue.release()
- result = proc.ExecOpCode(input_opcode, op.Log)
+ def _Log(*args):
+ """Append a log entry.
+
+ """
+ assert len(args) < 3
+
+ if len(args) == 1:
+ log_type = constants.ELOG_MESSAGE
+ log_msg = args[0]
+ else:
+ log_type, log_msg = args
+
+ # The time is split to make serialization easier and not lose
+ # precision.
+ timestamp = utils.SplitTime(time.time())
+
+ queue.acquire()
+ try:
+ job.log_serial += 1
+ op.log.append((job.log_serial, timestamp, log_type, log_msg))
+
+ job.change.notifyAll()
+ finally:
+ queue.release()
+
+ # Make sure not to hold lock while _Log is called
+ result = proc.ExecOpCode(input_opcode, _Log)
queue.acquire()
try:
op.status = constants.OP_STATUS_SUCCESS
op.result = result
+ op.end_timestamp = TimeStampNow()
queue.UpdateJobUnlocked(job)
finally:
queue.release()
try:
op.status = constants.OP_STATUS_ERROR
op.result = str(err)
+ op.end_timestamp = TimeStampNow()
logging.debug("Op %s/%s: Error in %s", idx + 1, count, op)
finally:
queue.UpdateJobUnlocked(job)
finally:
queue.acquire()
try:
- job_id = job.id
- status = job.CalcStatus()
+ try:
+ job.run_op_idx = -1
+ queue.UpdateJobUnlocked(job)
+ finally:
+ job_id = job.id
+ status = job.CalcStatus()
finally:
queue.release()
logging.debug("Worker %s finished job %s, status = %s",
class JobQueue(object):
_RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)
+ def _RequireOpenQueue(fn):
+ """Decorator for "public" functions.
+
+ This function should be used for all "public" functions. That is, functions
+ usually called from other classes.
+
+ Important: Use this decorator only after utils.LockedMethod!
+
+ Example:
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def Example(self):
+ pass
+
+ """
+ def wrapper(self, *args, **kwargs):
+ assert self._queue_lock is not None, "Queue should be open"
+ return fn(self, *args, **kwargs)
+ return wrapper
+
def __init__(self, context):
self.context = context
- self._memcache = {}
+ self._memcache = weakref.WeakValueDictionary()
self._my_hostname = utils.HostInfo().name
# Locking
self.acquire = self._lock.acquire
self.release = self._lock.release
- # Make sure our directories exists
- for path in (constants.QUEUE_DIR, constants.JOB_QUEUE_ARCHIVE_DIR):
- try:
- os.mkdir(path, 0700)
- except OSError, err:
- if err.errno not in (errno.EEXIST, ):
- raise
+ # Initialize
+ self._queue_lock = jstore.InitAndVerifyQueue(must_lock=True)
- # Get queue lock
- self.lock_fd = open(constants.JOB_QUEUE_LOCK_FILE, "w")
- try:
- utils.LockFile(self.lock_fd)
- except:
- self.lock_fd.close()
- raise
+ # Read serial file
+ self._last_serial = jstore.ReadSerial()
+ assert self._last_serial is not None, ("Serial file was modified between"
+ " check in jstore and here")
- # Read version
- try:
- version_fd = open(constants.JOB_QUEUE_VERSION_FILE, "r")
- except IOError, err:
- if err.errno not in (errno.ENOENT, ):
- raise
-
- # Setup a new queue
- self._InitQueueUnlocked()
-
- # Try to open again
- version_fd = open(constants.JOB_QUEUE_VERSION_FILE, "r")
+ # Get initial list of nodes
+ self._nodes = set(self.context.cfg.GetNodeList())
+ # Remove master node
try:
- # Try to read version
- version = int(version_fd.read(128))
-
- # Verify version
- if version != constants.JOB_QUEUE_VERSION:
- raise errors.JobQueueError("Found version %s, expected %s",
- version, constants.JOB_QUEUE_VERSION)
- finally:
- version_fd.close()
+ self._nodes.remove(self._my_hostname)
+ except ValueError:
+ pass
- self._last_serial = self._ReadSerial()
- if self._last_serial is None:
- raise errors.ConfigurationError("Can't read/parse the job queue serial"
- " file")
+ # TODO: Check consistency across nodes
# Setup worker pool
self._wpool = _JobQueueWorkerPool(self)
finally:
self.release()
- @staticmethod
- def _ReadSerial():
- """Try to read the job serial file.
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def AddNode(self, node_name):
+ assert node_name != self._my_hostname
- @rtype: None or int
- @return: If the serial can be read, then it is returned. Otherwise None
- is returned.
+ # Clean queue directory on added node
+ rpc.call_jobqueue_purge(node_name)
- """
- try:
- serial_fd = open(constants.JOB_QUEUE_SERIAL_FILE, "r")
+ # Upload the whole queue excluding archived jobs
+ files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
+
+ # Upload current serial file
+ files.append(constants.JOB_QUEUE_SERIAL_FILE)
+
+ for file_name in files:
+ # Read file content
+ fd = open(file_name, "r")
try:
- # Read last serial
- serial = int(serial_fd.read(1024).strip())
+ content = fd.read()
finally:
- serial_fd.close()
- except (ValueError, EnvironmentError):
- serial = None
+ fd.close()
+
+ result = rpc.call_jobqueue_update([node_name], file_name, content)
+ if not result[node_name]:
+ logging.error("Failed to upload %s to %s", file_name, node_name)
+
+ self._nodes.add(node_name)
+
+ @utils.LockedMethod
+ @_RequireOpenQueue
+ def RemoveNode(self, node_name):
+ try:
+ # The queue is removed by the "leave node" RPC call.
+ self._nodes.remove(node_name)
+ except KeyError:
+ pass
+
+ def _CheckRpcResult(self, result, nodes, failmsg):
+ failed = []
+ success = []
+
+ for node in nodes:
+ if result[node]:
+ success.append(node)
+ else:
+ failed.append(node)
+
+ if failed:
+ logging.error("%s failed on %s", failmsg, ", ".join(failed))
+
+ # +1 for the master node
+ if (len(success) + 1) < len(failed):
+ # TODO: Handle failing nodes
+ logging.error("More than half of the nodes failed")
+
+ def _WriteAndReplicateFileUnlocked(self, file_name, data):
+ """Writes a file locally and then replicates it to all nodes.
+
+ """
+ utils.WriteFile(file_name, data=data)
- return serial
+ result = rpc.call_jobqueue_update(self._nodes, file_name, data)
+ self._CheckRpcResult(result, self._nodes,
+ "Updating %s" % file_name)
- def _InitQueueUnlocked(self):
- assert self.lock_fd, "Queue should be open"
+ def _RenameFileUnlocked(self, old, new):
+ os.rename(old, new)
- utils.WriteFile(constants.JOB_QUEUE_VERSION_FILE,
- data="%s\n" % constants.JOB_QUEUE_VERSION)
- if self._ReadSerial() is None:
- utils.WriteFile(constants.JOB_QUEUE_SERIAL_FILE,
- data="%s\n" % 0)
+ result = rpc.call_jobqueue_rename(self._nodes, old, new)
+ self._CheckRpcResult(result, self._nodes,
+ "Moving %s to %s" % (old, new))
def _FormatJobID(self, job_id):
if not isinstance(job_id, (int, long)):
return str(job_id)
- def _NewSerialUnlocked(self, nodes):
+ def _NewSerialUnlocked(self):
"""Generates a new job identifier.
Job identifiers are unique during the lifetime of a cluster.
Returns: A string representing the job identifier.
"""
- assert self.lock_fd, "Queue should be open"
-
# New number
serial = self._last_serial + 1
# Write to file
- utils.WriteFile(constants.JOB_QUEUE_SERIAL_FILE,
- data="%s\n" % serial)
+ 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
- # Distribute the serial to the other nodes
- try:
- nodes.remove(self._my_hostname)
- except ValueError:
- pass
-
- result = rpc.call_upload_file(nodes, constants.JOB_QUEUE_SERIAL_FILE)
- for node in nodes:
- if not result[node]:
- logging.error("copy of job queue file to node %s failed", node)
-
return self._FormatJobID(serial)
@staticmethod
return jlist
def _ListJobFiles(self):
- assert self.lock_fd, "Queue should be open"
-
return [name for name in utils.ListVisibleFiles(constants.QUEUE_DIR)
if self._RE_JOB_FILE.match(name)]
def _LoadJobUnlocked(self, job_id):
- assert self.lock_fd, "Queue should be open"
-
- if job_id in self._memcache:
+ job = self._memcache.get(job_id, None)
+ if job:
logging.debug("Found job %s in memcache", job_id)
- return self._memcache[job_id]
+ return job
filepath = self._GetJobPath(job_id)
logging.debug("Loading job from %s", filepath)
return [self._LoadJobUnlocked(job_id) for job_id in job_ids]
@utils.LockedMethod
- def SubmitJob(self, ops, nodes):
+ @_RequireOpenQueue
+ def SubmitJob(self, ops):
"""Create and store a new job.
This enters the job into our job queue and also puts it on the new
@type ops: list
@param ops: The list of OpCodes that will become the new job.
- @type nodes: list
- @param nodes: The list of nodes to which the new job serial will be
- distributed.
"""
- assert self.lock_fd, "Queue should be open"
-
# Get job identifier
- job_id = self._NewSerialUnlocked(nodes)
+ job_id = self._NewSerialUnlocked()
job = _QueuedJob(self, job_id, ops)
# Write to disk
self.UpdateJobUnlocked(job)
- logging.debug("Added new job %s to the cache", job_id)
+ logging.debug("Adding new job %s to the cache", job_id)
self._memcache[job_id] = job
# Add to worker pool
return job.id
+ @_RequireOpenQueue
def UpdateJobUnlocked(self, job):
- assert self.lock_fd, "Queue should be open"
-
filename = self._GetJobPath(job.id)
+ data = serializer.DumpJson(job.Serialize(), indent=False)
logging.debug("Writing job %s to %s", job.id, filename)
- utils.WriteFile(filename,
- data=serializer.DumpJson(job.Serialize(), indent=False))
- self._CleanCacheUnlocked([job.id])
+ self._WriteAndReplicateFileUnlocked(filename, data)
- def _CleanCacheUnlocked(self, exclude):
- """Clean the memory cache.
+ # Notify waiters about potential changes
+ job.change.notifyAll()
- The exceptions argument contains job IDs that should not be
- cleaned.
+ @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
"""
- assert isinstance(exclude, list)
+ logging.debug("Waiting for changes in job %s", job_id)
+ 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
- for job in self._memcache.values():
- if job.id in exclude:
- continue
- if job.CalcStatus() not in (constants.JOB_STATUS_QUEUED,
- constants.JOB_STATUS_RUNNING):
- logging.debug("Cleaning job %s from the cache", job.id)
- try:
- del self._memcache[job.id]
- except KeyError:
- pass
+ 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):
+ # 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)
+
+ return (job_info, log_entries)
@utils.LockedMethod
+ @_RequireOpenQueue
def CancelJob(self, job_id):
"""Cancels a job.
self.UpdateJobUnlocked(job)
@utils.LockedMethod
+ @_RequireOpenQueue
def ArchiveJob(self, job_id):
"""Archives a job.
logging.debug("Job %s is not yet done", job.id)
return
- try:
- old = self._GetJobPath(job.id)
- new = self._GetArchivedJobPath(job.id)
+ old = self._GetJobPath(job.id)
+ new = self._GetArchivedJobPath(job.id)
- os.rename(old, new)
+ self._RenameFileUnlocked(old, new)
- logging.debug("Successfully archived job %s", job.id)
- finally:
- # Cleaning the cache because we don't know what os.rename actually did
- # and to be on the safe side.
- self._CleanCacheUnlocked([])
+ logging.debug("Successfully archived job %s", job.id)
def _GetJobInfoUnlocked(self, job, fields):
row = []
row.append([op.result for op in job.ops])
elif fname == "opstatus":
row.append([op.status for op in job.ops])
- elif fname == "ticker":
- ji = job.run_op_index
- if ji < 0:
- lmsg = None
- else:
- lmsg = job.ops[ji].RetrieveLog(-1)
- # message might be empty here
- if lmsg:
- lmsg = lmsg[0]
- else:
- lmsg = None
- row.append(lmsg)
else:
raise errors.OpExecError("Invalid job query field '%s'" % fname)
return row
@utils.LockedMethod
+ @_RequireOpenQueue
def QueryJobs(self, job_ids, fields):
"""Returns a list of jobs in queue.
return jobs
@utils.LockedMethod
+ @_RequireOpenQueue
def Shutdown(self):
"""Stops the job queue.
"""
- assert self.lock_fd, "Queue should be open"
-
self._wpool.TerminateWorkers()
- self.lock_fd.close()
- self.lock_fd = None
+ self._queue_lock.Close()
+ self._queue_lock = None
projects / ganeti-local / blobdiff