Copy recent changes in periodic tasks from nova.

The nova periodic task code has moved on. Bring oslo inline with
the new code. The next step after this is to update Quantum (the
only user of this code I can find) to the new interface, and then
move nova across to using this code.

Change-Id: I3f041ca9a6ba9f0e5be4b29198d16454c9797fef

2 files changed, 276 insertions, 27 deletions

diff --git a/openstack/common/periodic_task.py b/openstack/common/periodic_task.py
index 28ab22e..c48d8b9 100644
--- a/openstack/common/periodic_task.py
+++ b/openstack/common/periodic_task.py
@@ -13,26 +13,72 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import datetime
+import time
+from oslo.config import cfg
+
 from openstack.common.gettextutils import _
 from openstack.common import log as logging
+from openstack.common import timeutils
+
+
+periodic_opts = [
+    cfg.BoolOpt('run_external_periodic_tasks',
+                default=True,
+                help=('Some periodic tasks can be run in a separate process. '
+                      'Should we run them here?')),
+]
+
+CONF = cfg.CONF
+CONF.register_opts(periodic_opts)
 
 LOG = logging.getLogger(__name__)
 
+DEFAULT_INTERVAL = 60.0
+
+
+class InvalidPeriodicTaskArg(Exception):
+    message = _("Unexpected argument for periodic task creation: %(arg)s.")
+
 
 def periodic_task(*args, **kwargs):
     """Decorator to indicate that a method is a periodic task.
 
     This decorator can be used in two ways:
 
-        1. Without arguments '@periodic_task', this will be run on every tick
+        1. Without arguments '@periodic_task', this will be run on every cycle
            of the periodic scheduler.
 
-        2. With arguments, @periodic_task(ticks_between_runs=N), this will be
-           run on every N ticks of the periodic scheduler.
+        2. With arguments:
+           @periodic_task(spacing=N [, run_immediately=[True|False]])
+           this will be run on approximately every N seconds. If this number is
+           negative the periodic task will be disabled. If the run_immediately
+           argument is provided and has a value of 'True', the first run of the
+           task will be shortly after task scheduler starts.  If
+           run_immediately is omitted or set to 'False', the first time the
+           task runs will be approximately N seconds after the task scheduler
+           starts.
     """
     def decorator(f):
+        # Test for old style invocation
+        if 'ticks_between_runs' in kwargs:
+            raise InvalidPeriodicTaskArg(arg='ticks_between_runs')
+
+        # Control if run at all
         f._periodic_task = True
-        f._ticks_between_runs = kwargs.pop('ticks_between_runs', 0)
+        f._periodic_external_ok = kwargs.pop('external_process_ok', False)
+        if f._periodic_external_ok and not CONF.run_external_periodic_tasks:
+            f._periodic_enabled = False
+        else:
+            f._periodic_enabled = kwargs.pop('enabled', True)
+
+        # Control frequency
+        f._periodic_spacing = kwargs.pop('spacing', 0)
+        f._periodic_immediate = kwargs.pop('run_immediately', False)
+        if f._periodic_immediate:
+            f._periodic_last_run = None
+        else:
+            f._periodic_last_run = timeutils.utcnow()
         return f
 
     # NOTE(sirp): The `if` is necessary to allow the decorator to be used with
@@ -59,7 +105,7 @@ class _PeriodicTasksMeta(type):
         super(_PeriodicTasksMeta, cls).__init__(names, bases, dict_)
 
         # NOTE(sirp): if the attribute is not present then we must be the base
-        # class, so, go ahead and initialize it. If the attribute is present,
+        # class, so, go ahead an initialize it. If the attribute is present,
         # then we're a subclass so make a copy of it so we don't step on our
         # parent's toes.
         try:
@@ -68,20 +114,39 @@ class _PeriodicTasksMeta(type):
             cls._periodic_tasks = []
 
         try:
-            cls._ticks_to_skip = cls._ticks_to_skip.copy()
+            cls._periodic_last_run = cls._periodic_last_run.copy()
         except AttributeError:
-            cls._ticks_to_skip = {}
+            cls._periodic_last_run = {}
+
+        try:
+            cls._periodic_spacing = cls._periodic_spacing.copy()
+        except AttributeError:
+            cls._periodic_spacing = {}
 
-        # This uses __dict__ instead of
-        # inspect.getmembers(cls, inspect.ismethod) so only the methods of the
-        # current class are added when this class is scanned, and base classes
-        # are not added redundantly.
         for value in cls.__dict__.values():
             if getattr(value, '_periodic_task', False):
                 task = value
                 name = task.__name__
+
+                if task._periodic_spacing < 0:
+                    LOG.info(_('Skipping periodic task %(task)s because '
+                               'its interval is negative'),
+                             {'task': name})
+                    continue
+                if not task._periodic_enabled:
+                    LOG.info(_('Skipping periodic task %(task)s because '
+                               'it is disabled'),
+                             {'task': name})
+                    continue
+
+                # A periodic spacing of zero indicates that this task should
+                # be run every pass
+                if task._periodic_spacing == 0:
+                    task._periodic_spacing = None
+
                 cls._periodic_tasks.append((name, task))
-                cls._ticks_to_skip[name] = task._ticks_between_runs
+                cls._periodic_spacing[name] = task._periodic_spacing
+                cls._periodic_last_run[name] = task._periodic_last_run
 
 
 class PeriodicTasks(object):
@@ -89,27 +154,34 @@ class PeriodicTasks(object):
 
     def run_periodic_tasks(self, context, raise_on_error=False):
         """Tasks to be run at a periodic interval."""
+        idle_for = DEFAULT_INTERVAL
         for task_name, task in self._periodic_tasks:
             full_task_name = '.'.join([self.__class__.__name__, task_name])
 
-            ticks_to_skip = self._ticks_to_skip[task_name]
-            if ticks_to_skip > 0:
-                LOG.debug(_("Skipping %(full_task_name)s, %(ticks_to_skip)s"
-                            " ticks left until next run"),
-                          dict(full_task_name=full_task_name,
-                               ticks_to_skip=ticks_to_skip))
-                self._ticks_to_skip[task_name] -= 1
-                continue
+            now = timeutils.utcnow()
+            spacing = self._periodic_spacing[task_name]
+            last_run = self._periodic_last_run[task_name]
+
+            # If a periodic task is _nearly_ due, then we'll run it early
+            if spacing is not None and last_run is not None:
+                due = last_run + datetime.timedelta(seconds=spacing)
+                if not timeutils.is_soon(due, 0.2):
+                    idle_for = min(idle_for, timeutils.delta_seconds(now, due))
+                    continue
 
-            self._ticks_to_skip[task_name] = task._ticks_between_runs
-            LOG.debug(_("Running periodic task %(full_task_name)s"),
-                      dict(full_task_name=full_task_name))
+            if spacing is not None:
+                idle_for = min(idle_for, spacing)
+
+            LOG.debug(_("Running periodic task %(full_task_name)s"), locals())
+            self._periodic_last_run[task_name] = timeutils.utcnow()
 
             try:
                 task(self, context)
             except Exception as e:
                 if raise_on_error:
                     raise
-                LOG.exception(_("Error during %(full_task_name)s:"
-                                " %(e)s"),
-                              dict(e=e, full_task_name=full_task_name))
+                LOG.exception(_("Error during %(full_task_name)s: %(e)s"),
+                              locals())
+            time.sleep(0)
+
+        return idle_for
diff --git a/tests/unit/test_periodic.py b/tests/unit/test_periodic.py
index c39d085..1fb1574 100644
--- a/tests/unit/test_periodic.py
+++ b/tests/unit/test_periodic.py
@@ -19,8 +19,13 @@
 """
 Unit Tests for periodic_task decorator and PeriodicTasks class.
 """
+
+import datetime
+
 from openstack.common import periodic_task
+from openstack.common import timeutils
 from tests import utils
+from testtools import matchers
 
 
 class AService(periodic_task.PeriodicTasks):
@@ -37,7 +42,7 @@ class AService(periodic_task.PeriodicTasks):
         self.called['urg'] = True
         raise Exception('urg')
 
-    @periodic_task.periodic_task(ticks_between_runs=1)
+    @periodic_task.periodic_task(spacing=10)
     def doit_with_kwargs_odd(self, context):
         self.called['ticks'] = True
 
@@ -64,3 +69,175 @@ class PeriodicTasksTestCase(utils.BaseTestCase):
         self.assertRaises(Exception,
                           serv.run_periodic_tasks,
                           None, raise_on_error=True)
+
+
+class ManagerMetaTestCase(utils.BaseTestCase):
+    """Tests for the meta class which manages the creation of periodic tasks.
+    """
+
+    def test_meta(self):
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task
+            def foo(self):
+                return 'foo'
+
+            @periodic_task.periodic_task(spacing=4)
+            def bar(self):
+                return 'bar'
+
+            @periodic_task.periodic_task(enabled=False)
+            def baz(self):
+                return 'baz'
+
+        m = Manager()
+        self.assertThat(m._periodic_tasks, matchers.HasLength(2))
+        self.assertEqual(None, m._periodic_spacing['foo'])
+        self.assertEqual(4, m._periodic_spacing['bar'])
+        self.assertThat(
+            m._periodic_spacing, matchers.Not(matchers.Contains('baz')))
+
+
+class ManagerTestCase(utils.BaseTestCase):
+    """Tests the periodic tasks portion of the manager class."""
+
+    def test_periodic_tasks_with_idle(self):
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=200)
+            def bar(self):
+                return 'bar'
+
+        m = Manager()
+        self.assertThat(m._periodic_tasks, matchers.HasLength(1))
+        self.assertEqual(200, m._periodic_spacing['bar'])
+
+        # Now a single pass of the periodic tasks
+        idle = m.run_periodic_tasks(None)
+        self.assertAlmostEqual(60, idle, 1)
+
+    def test_periodic_tasks_constant(self):
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=0)
+            def bar(self):
+                return 'bar'
+
+        m = Manager()
+        idle = m.run_periodic_tasks(None)
+        self.assertAlmostEqual(60, idle, 1)
+
+    def test_periodic_tasks_idle_calculation(self):
+        fake_time = datetime.datetime(3000, 1, 1)
+        timeutils.set_time_override(fake_time)
+
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=10)
+            def bar(self, context):
+                return 'bar'
+
+        m = Manager()
+
+        # Ensure initial values are correct
+        self.assertEqual(1, len(m._periodic_tasks))
+        task_name, task = m._periodic_tasks[0]
+
+        # Test task values
+        self.assertEqual('bar', task_name)
+        self.assertEqual(10, task._periodic_spacing)
+        self.assertEqual(True, task._periodic_enabled)
+        self.assertEqual(False, task._periodic_external_ok)
+        self.assertEqual(False, task._periodic_immediate)
+        self.assertNotEqual(None, task._periodic_last_run)
+
+        # Test the manager's representation of those values
+        self.assertEqual(10, m._periodic_spacing[task_name])
+        self.assertNotEqual(None, m._periodic_last_run[task_name])
+
+        timeutils.advance_time_delta(datetime.timedelta(seconds=5))
+        m.run_periodic_tasks(None)
+
+        timeutils.advance_time_delta(datetime.timedelta(seconds=5))
+        idle = m.run_periodic_tasks(None)
+        self.assertAlmostEqual(10, idle, 1)
+
+    def test_periodic_tasks_immediate_runs_now(self):
+        fake_time = datetime.datetime(3000, 1, 1)
+        timeutils.set_time_override(fake_time)
+
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=10, run_immediately=True)
+            def bar(self, context):
+                return 'bar'
+
+        m = Manager()
+
+        # Ensure initial values are correct
+        self.assertEqual(1, len(m._periodic_tasks))
+        task_name, task = m._periodic_tasks[0]
+
+        # Test task values
+        self.assertEqual('bar', task_name)
+        self.assertEqual(10, task._periodic_spacing)
+        self.assertEqual(True, task._periodic_enabled)
+        self.assertEqual(False, task._periodic_external_ok)
+        self.assertEqual(True, task._periodic_immediate)
+        self.assertEqual(None, task._periodic_last_run)
+
+        # Test the manager's representation of those values
+        self.assertEqual(10, m._periodic_spacing[task_name])
+        self.assertEqual(None, m._periodic_last_run[task_name])
+
+        idle = m.run_periodic_tasks(None)
+        self.assertEqual(datetime.datetime(3000, 1, 1, 0, 0),
+                         m._periodic_last_run[task_name])
+        self.assertAlmostEqual(10, idle, 1)
+
+        timeutils.advance_time_delta(datetime.timedelta(seconds=5))
+        idle = m.run_periodic_tasks(None)
+        self.assertAlmostEqual(5, idle, 1)
+
+    def test_periodic_tasks_disabled(self):
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=-1)
+            def bar(self):
+                return 'bar'
+
+        m = Manager()
+        idle = m.run_periodic_tasks(None)
+        self.assertAlmostEqual(60, idle, 1)
+
+    def test_external_running_here(self):
+        self.config(run_external_periodic_tasks=True)
+
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=200, external_process_ok=True)
+            def bar(self):
+                return 'bar'
+
+        m = Manager()
+        self.assertThat(m._periodic_tasks, matchers.HasLength(1))
+
+    def test_external_running_elsewhere(self):
+        self.config(run_external_periodic_tasks=False)
+
+        class Manager(periodic_task.PeriodicTasks):
+            __metaclass__ = periodic_task._PeriodicTasksMeta
+
+            @periodic_task.periodic_task(spacing=200, external_process_ok=True)
+            def bar(self):
+                return 'bar'
+
+        m = Manager()
+        self.assertEqual([], m._periodic_tasks)