openstack/nova
Code Issues Proposed changes

Merge "Update to using oslo periodic tasks implementation."

Browse Source
This commit is contained in:
Jenkins
2013-05-01 16:23:57 +00:00
committed by Gerrit Code Review
parent 2caf27936e 01b0f6554b
commit 3266bff670
7 changed files with 30 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
nova/cells/manager.py
View File

@@ -28,6 +28,7 @@ from nova import context
from nova import exception
from nova import manager
from nova.openstack.common import importutils
from nova.openstack.common import periodic_task
from nova.openstack.common import timeutils
cell_manager_opts = [
@@ -97,7 +98,7 @@ class CellsManager(manager.Manager):
else:
self._update_our_parents(ctxt)
@manager.periodic_task
@periodic_task.periodic_task
def _update_our_parents(self, ctxt):
"""Update our parent cells with our capabilities and capacity
if we're at the bottom of the tree.
@@ -105,7 +106,7 @@ class CellsManager(manager.Manager):
self.msg_runner.tell_parents_our_capabilities(ctxt)
self.msg_runner.tell_parents_our_capacities(ctxt)
@manager.periodic_task
@periodic_task.periodic_task
def _heal_instances(self, ctxt):
"""Periodic task to send updates for a number of instances to
parent cells.

34
nova/compute/manager.py
View File

@@ -65,6 +65,7 @@ from nova.openstack.common import jsonutils
from nova.openstack.common import lockutils
from nova.openstack.common import log as logging
from nova.openstack.common.notifier import api as notifier
from nova.openstack.common import periodic_task
from nova.openstack.common import rpc
from nova.openstack.common.rpc import common as rpc_common
from nova.openstack.common import timeutils
@@ -1059,7 +1060,7 @@ class ComputeManager(manager.SchedulerDependentManager):
scheduler_method(context, *method_args)
return True
@manager.periodic_task
@periodic_task.periodic_task
def _check_instance_build_time(self, context):
"""Ensure that instances are not stuck in build."""
timeout = CONF.instance_build_timeout
@@ -3458,7 +3459,7 @@ class ComputeManager(manager.SchedulerDependentManager):
self.driver.destroy(instance, self._legacy_nw_info(network_info),
block_device_info)
@manager.periodic_task
@periodic_task.periodic_task
def _heal_instance_info_cache(self, context):
"""Called periodically. On every call, try to update the
info_cache's network information for another instance by
@@ -3511,7 +3512,7 @@ class ComputeManager(manager.SchedulerDependentManager):
# We don't care about any failures
pass
@manager.periodic_task
@periodic_task.periodic_task
def _poll_rebooting_instances(self, context):
if CONF.reboot_timeout > 0:
instances = self.conductor_api.instance_get_all_hung_in_rebooting(
@@ -3519,7 +3520,7 @@ class ComputeManager(manager.SchedulerDependentManager):
self.driver.poll_rebooting_instances(CONF.reboot_timeout,
instances)
@manager.periodic_task
@periodic_task.periodic_task
def _poll_rescued_instances(self, context):
if CONF.rescue_timeout > 0:
instances = self.conductor_api.instance_get_all_by_host(
@@ -3539,7 +3540,7 @@ class ComputeManager(manager.SchedulerDependentManager):
for instance in to_unrescue:
self.conductor_api.compute_unrescue(context, instance)
@manager.periodic_task
@periodic_task.periodic_task
def _poll_unconfirmed_resizes(self, context):
if CONF.resize_confirm_window > 0:
capi = self.conductor_api
@@ -3596,7 +3597,7 @@ class ComputeManager(manager.SchedulerDependentManager):
"Will retry later.")
LOG.error(msg % locals(), instance=instance)
@manager.periodic_task
@periodic_task.periodic_task
def _instance_usage_audit(self, context):
if CONF.instance_usage_audit:
if not compute_utils.has_audit_been_run(context,
@@ -3644,7 +3645,7 @@ class ComputeManager(manager.SchedulerDependentManager):
num_instances,
time.time() - start_time))
@manager.periodic_task
@periodic_task.periodic_task
def _poll_bandwidth_usage(self, context):
prev_time, start_time = utils.last_completed_audit_period()
@@ -3751,7 +3752,7 @@ class ComputeManager(manager.SchedulerDependentManager):
notifier.INFO,
compute_utils.usage_volume_info(vol_usage))
@manager.periodic_task
@periodic_task.periodic_task
def _poll_volume_usage(self, context, start_time=None):
if CONF.volume_usage_poll_interval == 0:
return
@@ -3781,7 +3782,7 @@ class ComputeManager(manager.SchedulerDependentManager):
self._send_volume_usage_notifications(context, start_time)
@manager.periodic_task
@periodic_task.periodic_task
def _report_driver_status(self, context):
curr_time = time.time()
if curr_time - self._last_host_check > CONF.host_state_interval:
@@ -3795,8 +3796,8 @@ class ComputeManager(manager.SchedulerDependentManager):
capability['host_ip'] = CONF.my_ip
self.update_service_capabilities(capabilities)
@manager.periodic_task(spacing=CONF.sync_power_state_interval,
run_immediately=True)
@periodic_task.periodic_task(spacing=CONF.sync_power_state_interval,
run_immediately=True)
def _sync_power_states(self, context):
"""Align power states between the database and the hypervisor.
@@ -3956,7 +3957,7 @@ class ComputeManager(manager.SchedulerDependentManager):
LOG.warn(_("Instance is not (soft-)deleted."),
instance=db_instance)
@manager.periodic_task
@periodic_task.periodic_task
def _reclaim_queued_deletes(self, context):
"""Reclaim instances that are queued for deletion."""
interval = CONF.reclaim_instance_interval
@@ -3987,7 +3988,7 @@ class ComputeManager(manager.SchedulerDependentManager):
# pass reservations here.
self._delete_instance(context, instance, bdms)
@manager.periodic_task
@periodic_task.periodic_task
def update_available_resource(self, context):
"""See driver.get_available_resource()
@@ -4023,7 +4024,8 @@ class ComputeManager(manager.SchedulerDependentManager):
return service_ref['compute_node']
@manager.periodic_task(spacing=CONF.running_deleted_instance_poll_interval)
@periodic_task.periodic_task(
spacing=CONF.running_deleted_instance_poll_interval)
def _cleanup_running_deleted_instances(self, context):
"""Cleanup any instances which are erroneously still running after
having been deleted.
@@ -4143,8 +4145,8 @@ class ComputeManager(manager.SchedulerDependentManager):
aggregate, host,
isinstance(e, exception.AggregateError))
@manager.periodic_task(spacing=CONF.image_cache_manager_interval,
external_process_ok=True)
@periodic_task.periodic_task(spacing=CONF.image_cache_manager_interval,
external_process_ok=True)
def _run_image_cache_manager_pass(self, context):
"""Run a single pass of the image cache manager."""

4
nova/exception.py
View File

@@ -452,10 +452,6 @@ class InvalidID(Invalid):
message = _("Invalid ID received %(id)s.")
class InvalidPeriodicTaskArg(Invalid):
message = _("Unexpected argument for periodic task creation: %(arg)s.")
class ConstraintNotMet(NovaException):
message = _("Constraint not met.")
code = 412

161
nova/manager.py
View File

@@ -53,146 +53,23 @@ This module provides Manager, a base class for managers.
"""
import datetime
import eventlet
from oslo.config import cfg
from nova import baserpc
from nova.db import base
from nova import exception
from nova.openstack.common import log as logging
from nova.openstack.common import periodic_task
from nova.openstack.common.plugin import pluginmanager
from nova.openstack.common.rpc import dispatcher as rpc_dispatcher
from nova.openstack.common import timeutils
from nova.scheduler import rpcapi as scheduler_rpcapi
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)
CONF.import_opt('host', 'nova.netconf')
LOG = logging.getLogger(__name__)
DEFAULT_INTERVAL = 60.0
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 cycle
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 exception.InvalidPeriodicTaskArg(arg='ticks_between_runs')
# Control if run at all
f._periodic_task = True
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
# and without parens.
#
# In the 'with-parens' case (with kwargs present), this function needs to
# return a decorator function since the interpreter will invoke it like:
#
# periodic_task(*args, **kwargs)(f)
#
# In the 'without-parens' case, the original function will be passed
# in as the first argument, like:
#
# periodic_task(f)
if kwargs:
return decorator
else:
return decorator(args[0])
class ManagerMeta(type):
def __init__(cls, names, bases, dict_):
"""Metaclass that allows us to collect decorated periodic tasks."""
super(ManagerMeta, cls).__init__(names, bases, dict_)
# NOTE(sirp): if the attribute is not present then we must be the base
# 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:
cls._periodic_tasks = cls._periodic_tasks[:]
except AttributeError:
cls._periodic_tasks = []
try:
cls._periodic_last_run = cls._periodic_last_run.copy()
except AttributeError:
cls._periodic_last_run = {}
try:
cls._periodic_spacing = cls._periodic_spacing.copy()
except AttributeError:
cls._periodic_spacing = {}
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._periodic_spacing[name] = task._periodic_spacing
cls._periodic_last_run[name] = task._periodic_last_run
class Manager(base.Base):
__metaclass__ = ManagerMeta
class Manager(base.Base, periodic_task.PeriodicTasks):
# Set RPC API version to 1.0 by default.
RPC_API_VERSION = '1.0'
@@ -220,37 +97,7 @@ class Manager(base.Base):
def 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])
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
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"),
locals())
eventlet.sleep(0)
return idle_for
return self.run_periodic_tasks(context, raise_on_error=raise_on_error)
def init_host(self):
"""Hook to do additional manager initialization when one requests
@@ -308,7 +155,7 @@ class SchedulerDependentManager(Manager):
capabilities = [capabilities]
self.last_capabilities = capabilities
@periodic_task
@periodic_task.periodic_task
def publish_service_capabilities(self, context):
"""Pass data back to the scheduler.

5
nova/network/manager.py
View File

@@ -68,6 +68,7 @@ from nova.openstack.common import importutils
from nova.openstack.common import jsonutils
from nova.openstack.common import lockutils
from nova.openstack.common import log as logging
from nova.openstack.common import periodic_task
from nova.openstack.common.rpc import common as rpc_common
from nova.openstack.common import timeutils
from nova.openstack.common import uuidutils
@@ -350,7 +351,7 @@ class NetworkManager(manager.Manager):
dev = self.driver.get_dev(network)
self.driver.update_dns(ctxt, dev, network)
@manager.periodic_task
@periodic_task.periodic_task
def _disassociate_stale_fixed_ips(self, context):
if self.timeout_fixed_ips:
now = timeutils.utcnow()
@@ -1408,7 +1409,7 @@ class NetworkManager(manager.Manager):
vif['net_uuid'] = network['uuid']
return vif
@manager.periodic_task(
@periodic_task.periodic_task(
spacing=CONF.dns_update_periodic_interval)
def _periodic_update_dns(self, context):
"""Update local DNS entries of all networks on this host."""

3
nova/scheduler/manager.py
View File

@@ -39,6 +39,7 @@ from nova.openstack.common import importutils
from nova.openstack.common import jsonutils
from nova.openstack.common import log as logging
from nova.openstack.common.notifier import api as notifier
from nova.openstack.common import periodic_task
from nova import quota
@@ -287,7 +288,7 @@ class SchedulerManager(manager.Manager):
return {'resource': resource, 'usage': usage}
@manager.periodic_task
@periodic_task.periodic_task
def _expire_reservations(self, context):
QUOTAS.expire(context)

182
nova/tests/test_periodic_tasks.py
View File

@@ -1,182 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import datetime
from testtools import matchers
from nova import manager
from nova.openstack.common import timeutils
from nova import test
class ManagerMetaTestCase(test.TestCase):
"""Tests for the meta class which manages the creation of periodic tasks.
"""
def test_meta(self):
class Manager(object):
__metaclass__ = manager.ManagerMeta
@manager.periodic_task
def foo(self):
return 'foo'
@manager.periodic_task(spacing=4)
def bar(self):
return 'bar'
@manager.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 Manager(test.TestCase):
"""Tests the periodic tasks portion of the manager class."""
def test_periodic_tasks_with_idle(self):
class Manager(manager.Manager):
@manager.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.periodic_tasks(None)
self.assertAlmostEqual(60, idle, 1)
def test_periodic_tasks_constant(self):
class Manager(manager.Manager):
@manager.periodic_task(spacing=0)
def bar(self):
return 'bar'
m = Manager()
idle = m.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(manager.Manager):
@manager.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.periodic_tasks(None)
timeutils.advance_time_delta(datetime.timedelta(seconds=5))
idle = m.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(manager.Manager):
@manager.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.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.periodic_tasks(None)
self.assertAlmostEqual(5, idle, 1)
def test_periodic_tasks_disabled(self):
class Manager(manager.Manager):
@manager.periodic_task(spacing=-1)
def bar(self):
return 'bar'
m = Manager()
idle = m.periodic_tasks(None)
self.assertAlmostEqual(60, idle, 1)
def test_external_running_here(self):
self.flags(run_external_periodic_tasks=True)
class Manager(manager.Manager):
@manager.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.flags(run_external_periodic_tasks=False)
class Manager(manager.Manager):
@manager.periodic_task(spacing=200, external_process_ok=True)
def bar(self):
return 'bar'
m = Manager()
self.assertEqual([], m._periodic_tasks)