Synnefo » snf-ganeti

	test/ganeti.jqueue_unittest.py \

JOBS_FINALIZED = frozenset([

  JOB_STATUS_CANCELED,

  JOB_STATUS_SUCCESS,

  JOB_STATUS_ERROR,

  ])

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:

#!/usr/bin/python

#

# Copyright (C) 2010 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

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing ganeti.jqueue"""

import os

import sys

import unittest

import tempfile

import shutil

import errno

from ganeti import constants

from ganeti import utils

from ganeti import errors

from ganeti import jqueue

import testutils

class _FakeJob:

  def __init__(self, job_id, status):

    self.id = job_id

    self._status = status

    self._log = []

  def SetStatus(self, status):

    self._status = status

  def AddLogEntry(self, msg):

    self._log.append((len(self._log), msg))

  def CalcStatus(self):

    return self._status

  def GetInfo(self, fields):

    result = []

    for name in fields:

      if name == "status":

        result.append(self._status)

      else:

        raise Exception("Unknown field")

    return result

  def GetLogEntries(self, newer_than):

    assert newer_than is None or newer_than >= 0

    if newer_than is None:

      return self._log

    return self._log[newer_than:]

class TestJobChangesChecker(unittest.TestCase):

  def testStatus(self):

    job = _FakeJob(9094, constants.JOB_STATUS_QUEUED)

    checker = jqueue._JobChangesChecker(["status"], None, None)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_QUEUED], []))

    job.SetStatus(constants.JOB_STATUS_RUNNING)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))

    job.SetStatus(constants.JOB_STATUS_SUCCESS)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_SUCCESS], []))

    # job.id is used by checker

    self.assertEqual(job.id, 9094)

  def testStatusWithPrev(self):

    job = _FakeJob(12807, constants.JOB_STATUS_QUEUED)

    checker = jqueue._JobChangesChecker(["status"],

                                        [constants.JOB_STATUS_QUEUED], None)

    self.assert_(checker(job) is None)

    job.SetStatus(constants.JOB_STATUS_RUNNING)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))

  def testFinalStatus(self):

    for status in constants.JOBS_FINALIZED:

      job = _FakeJob(2178711, status)

      checker = jqueue._JobChangesChecker(["status"], [status], None)

      # There won't be any changes in this status, hence it should signal

      # a change immediately

      self.assertEqual(checker(job), ([status], []))

  def testLog(self):

    job = _FakeJob(9094, constants.JOB_STATUS_RUNNING)

    checker = jqueue._JobChangesChecker(["status"], None, None)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))

    job.AddLogEntry("Hello World")

    (job_info, log_entries) = checker(job)

    self.assertEqual(job_info, [constants.JOB_STATUS_RUNNING])

    self.assertEqual(log_entries, [[0, "Hello World"]])

    checker2 = jqueue._JobChangesChecker(["status"], job_info, len(log_entries))

    self.assert_(checker2(job) is None)

    job.AddLogEntry("Foo Bar")

    job.SetStatus(constants.JOB_STATUS_ERROR)

    (job_info, log_entries) = checker2(job)

    self.assertEqual(job_info, [constants.JOB_STATUS_ERROR])

    self.assertEqual(log_entries, [[1, "Foo Bar"]])

    checker3 = jqueue._JobChangesChecker(["status"], None, None)

    (job_info, log_entries) = checker3(job)

    self.assertEqual(job_info, [constants.JOB_STATUS_ERROR])

    self.assertEqual(log_entries, [[0, "Hello World"], [1, "Foo Bar"]])

class TestJobChangesWaiter(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    self.filename = utils.PathJoin(self.tmpdir, "job-1")

    utils.WriteFile(self.filename, data="")

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def _EnsureNotifierClosed(self, notifier):

    try:

      os.fstat(notifier._fd)

    except EnvironmentError, err:

      self.assertEqual(err.errno, errno.EBADF)

    else:

      self.fail("File descriptor wasn't closed")

  def testClose(self):

    for wait in [False, True]:

      waiter = jqueue._JobFileChangesWaiter(self.filename)

      try:

        if wait:

          waiter.Wait(0.001)

      finally:

        waiter.Close()

      # Ensure file descriptor was closed

      self._EnsureNotifierClosed(waiter._notifier)

  def testChangingFile(self):

    waiter = jqueue._JobFileChangesWaiter(self.filename)

    try:

      self.assertFalse(waiter.Wait(0.1))

      utils.WriteFile(self.filename, data="changed")

      self.assert_(waiter.Wait(60))

    finally:

      waiter.Close()

    self._EnsureNotifierClosed(waiter._notifier)

  def testChangingFile2(self):

    waiter = jqueue._JobChangesWaiter(self.filename)

    try:

      self.assertFalse(waiter._filewaiter)

      self.assert_(waiter.Wait(0.1))

      self.assert_(waiter._filewaiter)

      # File waiter is now used, but there have been no changes

      self.assertFalse(waiter.Wait(0.1))

      utils.WriteFile(self.filename, data="changed")

      self.assert_(waiter.Wait(60))

    finally:

      waiter.Close()

    self._EnsureNotifierClosed(waiter._filewaiter._notifier)

class TestWaitForJobChangesHelper(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    self.filename = utils.PathJoin(self.tmpdir, "job-2614226563")

    utils.WriteFile(self.filename, data="")

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def _LoadWaitingJob(self):

    return _FakeJob(2614226563, constants.JOB_STATUS_WAITLOCK)

  def _LoadLostJob(self):

    return None

  def testNoChanges(self):

    wfjc = jqueue._WaitForJobChangesHelper()

    # No change

    self.assertEqual(wfjc(self.filename, self._LoadWaitingJob, ["status"],

                          [constants.JOB_STATUS_WAITLOCK], None, 0.1),

                     constants.JOB_NOTCHANGED)

    # No previous information

    self.assertEqual(wfjc(self.filename, self._LoadWaitingJob,

                          ["status"], None, None, 1.0),

                     ([constants.JOB_STATUS_WAITLOCK], []))

  def testLostJob(self):

    wfjc = jqueue._WaitForJobChangesHelper()

    self.assert_(wfjc(self.filename, self._LoadLostJob,

                      ["status"], None, None, 1.0) is None)

if __name__ == "__main__":

  testutils.GanetiTestProgram()