Julia Kreger
5731f4fafc
One of the challeges with eventlet is it has helped make some of our tests run nice and quick because time.sleep ends up getting replaced deep under the hood when we monkey patch. In order for us to get to a point where we can begin to consider removing the patch, the tests need to begin to change. To that end, we have some test reliance upon that patching and also calling helpers and then testing the helper in the driver tests as well which increases the test time, and really is sort of "excess" complexity. So the goal here is to see if folks are good with such minor changes, since we can do them now and should try to avoid making changes to these tests when we begin to really remove eventlet. As a note, this is the first of two, or possibly three patches to do this overall cleanup. Change-Id: I28a36ed1facdf4fd6ced2efceacb39d12de72e8f
179 lines
7.3 KiB
Python
179 lines
7.3 KiB
Python
# 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.
|
|
|
|
"""Conductor periodics."""
|
|
|
|
import collections
|
|
import functools
|
|
import inspect
|
|
import time
|
|
|
|
from futurist import periodics
|
|
from oslo_log import log
|
|
|
|
from ironic.common import exception
|
|
from ironic.common import metrics_utils
|
|
from ironic.conductor import base_manager
|
|
from ironic.conductor import task_manager
|
|
from ironic.drivers import base as driver_base
|
|
|
|
|
|
LOG = log.getLogger(__name__)
|
|
|
|
|
|
METRICS = metrics_utils.get_metrics_logger(__name__)
|
|
|
|
|
|
def periodic(spacing, enabled=True, **kwargs):
|
|
"""A decorator to define a periodic task.
|
|
|
|
:param spacing: how often (in seconds) to run the periodic task.
|
|
:param enabled: whether the task is enabled; defaults to ``spacing > 0``.
|
|
"""
|
|
return periodics.periodic(spacing=spacing,
|
|
enabled=enabled and spacing > 0,
|
|
**kwargs)
|
|
|
|
|
|
class Stop(Exception):
|
|
"""A signal to stop the current iteration of a periodic task."""
|
|
|
|
|
|
def node_periodic(purpose, spacing, enabled=True, filters=None,
|
|
predicate=None, predicate_extra_fields=(), limit=None,
|
|
shared_task=True, node_count_metric_name=None):
|
|
"""A decorator to define a periodic task to act on nodes.
|
|
|
|
Defines a periodic task that fetches the list of nodes mapped to the
|
|
current conductor which satisfy the provided filters.
|
|
|
|
The decorated function must be a method on either the conductor manager
|
|
or a hardware interface. The signature is:
|
|
|
|
* for conductor manager: ``(self, task, context)``
|
|
* for hardware interfaces: ``(self, task, manager, context)``.
|
|
|
|
When the periodic is running on a hardware interface, only tasks
|
|
using this interface are considered.
|
|
|
|
``NodeNotFound`` and ``NodeLocked`` exceptions are ignored. Raise ``Stop``
|
|
to abort the current iteration of the task and reschedule it.
|
|
|
|
:param purpose: a human-readable description of the activity, e.g.
|
|
"verifying that the cat is purring".
|
|
:param spacing: how often (in seconds) to run the periodic task.
|
|
:param enabled: whether the task is enabled; defaults to ``spacing > 0``.
|
|
:param filters: database-level filters for the nodes.
|
|
:param predicate: a callable to run on the fetched nodes *before* creating
|
|
a task for them. The only parameter will be a named tuple with fields
|
|
``uuid``, ``driver``, ``conductor_group`` plus everything from
|
|
``predicate_extra_fields``. If the callable accepts a 2nd parameter,
|
|
it will be the conductor manager instance.
|
|
:param predicate_extra_fields: extra fields to fetch on the initial
|
|
request and pass into the ``predicate``. Must not contain ``uuid``,
|
|
``driver`` and ``conductor_group`` since they are always included.
|
|
:param limit: how many nodes to process before stopping the current
|
|
iteration. If ``predicate`` returns ``False``, the node is not counted.
|
|
If the decorated function returns ``False``, the node is not counted
|
|
either. Can be a callable, in which case it will be called on each
|
|
iteration to determine the limit.
|
|
:param shared_task: if ``True``, the task will have a shared lock. It is
|
|
recommended to start with a shared lock and upgrade it only if needed.
|
|
:param node_count_metric_name: A string value to identify a metric
|
|
representing the count of matching nodes to be recorded upon the
|
|
completion of the periodic.
|
|
"""
|
|
node_type = collections.namedtuple(
|
|
'Node',
|
|
['uuid', 'driver', 'conductor_group'] + list(predicate_extra_fields)
|
|
)
|
|
|
|
# Accepting a conductor manager is a bit of an edge case, doing a bit of
|
|
# a signature magic to avoid passing it everywhere.
|
|
accepts_manager = (predicate is not None
|
|
and len(inspect.signature(predicate).parameters) > 1)
|
|
|
|
def decorator(func):
|
|
@periodic(spacing=spacing, enabled=enabled)
|
|
@functools.wraps(func)
|
|
def wrapper(self, *args, **kwargs):
|
|
# Make it work with both drivers and the conductor manager
|
|
if isinstance(self, base_manager.BaseConductorManager):
|
|
manager = self
|
|
context = args[0]
|
|
else:
|
|
manager = args[0]
|
|
context = args[1]
|
|
|
|
interface_type = (getattr(self, 'interface_type', None)
|
|
if isinstance(self, driver_base.BaseInterface)
|
|
else None)
|
|
|
|
if callable(limit):
|
|
local_limit = limit()
|
|
else:
|
|
local_limit = limit
|
|
assert local_limit is None or local_limit > 0
|
|
node_count = 0
|
|
nodes = manager.iter_nodes(filters=filters,
|
|
fields=predicate_extra_fields)
|
|
for (node_uuid, *other) in nodes:
|
|
node_count += 1
|
|
if predicate is not None:
|
|
node = node_type(node_uuid, *other)
|
|
if accepts_manager:
|
|
result = predicate(node, manager)
|
|
else:
|
|
result = predicate(node)
|
|
if not result:
|
|
continue
|
|
|
|
try:
|
|
with task_manager.acquire(context, node_uuid,
|
|
purpose=purpose,
|
|
shared=shared_task) as task:
|
|
if interface_type is not None:
|
|
impl = getattr(task.driver, interface_type)
|
|
if not isinstance(impl, self.__class__):
|
|
continue
|
|
|
|
result = func(self, task, *args, **kwargs)
|
|
except exception.NodeNotFound:
|
|
LOG.info("During %(action)s, node %(node)s was not found "
|
|
"and presumed deleted by another process.",
|
|
{'node': node_uuid, 'action': purpose})
|
|
except exception.NodeLocked:
|
|
LOG.info("During %(action)s, node %(node)s was already "
|
|
"locked by another process. Skip.",
|
|
{'node': node_uuid, 'action': purpose})
|
|
except Stop:
|
|
break
|
|
finally:
|
|
# Yield on every iteration
|
|
time.sleep(0)
|
|
|
|
if (local_limit is not None
|
|
and (result is None or result)):
|
|
local_limit -= 1
|
|
if not local_limit:
|
|
return
|
|
if node_count_metric_name:
|
|
# Send post-run metrics.
|
|
METRICS.send_gauge(
|
|
node_count_metric_name,
|
|
node_count)
|
|
LOG.debug('Completed periodic task for purpose %s.', purpose)
|
|
|
|
return wrapper
|
|
|
|
return decorator
|