Synnefo » snf-ganeti

from ganeti import compat

class _JobChangesChecker(object):

  def __init__(self, fields, prev_job_info, prev_log_serial):

    """Initializes this class.

    @type fields: list of strings

    @param fields: Fields requested by LUXI client

    @type prev_job_info: string

    @param prev_job_info: previous job info, as passed by the LUXI client

    @type prev_log_serial: string

    @param prev_log_serial: previous job serial, as passed by the LUXI client

    """

    self._fields = fields

    self._prev_job_info = prev_job_info

    self._prev_log_serial = prev_log_serial

  def __call__(self, job):

    """Checks whether job has changed.

    @type job: L{_QueuedJob}

    @param job: Job object

    status = job.CalcStatus()

    job_info = job.GetInfo(self._fields)

    log_entries = job.GetLogEntries(self._prev_log_serial)

    # TODO: Doing something like in testutils.py:UnifyValueType might be more

    # efficient, though floats will be tricky

    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) or

        job_info != self._prev_job_info or

        (log_entries and self._prev_log_serial != log_entries[0][0])):

      logging.debug("Job %s changed", job.id)

      return (job_info, log_entries)

    return None

class _JobFileChangesWaiter(object):

  def __init__(self, filename):

    """Initializes this class.

    @type filename: string

    @param filename: Path to job file

    @raises errors.InotifyError: if the notifier cannot be setup

    """

    self._wm = pyinotify.WatchManager()

    self._inotify_handler = \

      asyncnotifier.SingleFileEventHandler(self._wm, self._OnInotify, filename)

    self._notifier = \

      pyinotify.Notifier(self._wm, default_proc_fun=self._inotify_handler)

    try:

      self._inotify_handler.enable()

    except Exception:

      # pyinotify doesn't close file descriptors automatically

      self._notifier.stop()

      raise

  def _OnInotify(self, notifier_enabled):

    """Callback for inotify.

    """

      self._inotify_handler.enable()

  def Wait(self, timeout):

    """Waits for the job file to change.

    @type timeout: float

    @param timeout: Timeout in seconds

    @return: Whether there have been events

    """

    assert timeout >= 0

    have_events = self._notifier.check_events(timeout * 1000)

    if have_events:

      self._notifier.read_events()

    self._notifier.process_events()

    return have_events

  def Close(self):

    """Closes underlying notifier and its file descriptor.

    """

    self._notifier.stop()

class _JobChangesWaiter(object):

  def __init__(self, filename):

    """Initializes this class.

    @type filename: string

    @param filename: Path to job file

    """

    self._filewaiter = None

    self._filename = filename

  def Wait(self, timeout):

    """Waits for a job to change.

    @type timeout: float

    @param timeout: Timeout in seconds

    @return: Whether there have been events

    """

    if self._filewaiter:

      return self._filewaiter.Wait(timeout)

    # Lazy setup: Avoid inotify setup cost when job file has already changed.

    # If this point is reached, return immediately and let caller check the job

    # file again in case there were changes since the last check. This avoids a

    # race condition.

    self._filewaiter = _JobFileChangesWaiter(self._filename)

    return True

  def Close(self):

    """Closes underlying waiter.

    """

    if self._filewaiter:

      self._filewaiter.Close()

class _WaitForJobChangesHelper(object):

  """Helper class using inotify to wait for changes in a job file.

  This class takes a previous job status and serial, and alerts the client when

  the current job status has changed.

  """

  @staticmethod

  def _CheckForChanges(job_load_fn, check_fn):

    job = job_load_fn()

    if not job:

      raise errors.JobLost()

    result = check_fn(job)

    if result is None:

      raise utils.RetryAgain()

    return result

  def __call__(self, filename, job_load_fn,

               fields, prev_job_info, prev_log_serial, timeout):

    """Waits for changes on a job.

    @type filename: string

    @param filename: File on which to wait for changes

    @type job_load_fn: callable

    @param job_load_fn: Function to load job

    @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 in seconds

    """

      check_fn = _JobChangesChecker(fields, prev_job_info, prev_log_serial)

      waiter = _JobChangesWaiter(filename)

      try:

        return utils.Retry(compat.partial(self._CheckForChanges,

                                          job_load_fn, check_fn),

                           utils.RETRY_REMAINING_TIME, timeout,

                           wait_fn=waiter.Wait)

      finally:

        waiter.Close()

    @param timeout: maximum time to wait in seconds

    load_fn = compat.partial(self.SafeLoadJobFromDisk, job_id)

    helper = _WaitForJobChangesHelper()

    return helper(self._GetJobPath(job_id), load_fn,

                  fields, prev_job_info, prev_log_serial, timeout)

      if job.CalcStatus() not in constants.JOBS_FINALIZED: