Source code for ironic_inspector.plugins.base

# 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.

"""Base code for plugins support."""

import abc

from oslo_config import cfg
from oslo_log import log
import six
import stevedore

from ironic_inspector.common.i18n import _, _LW


CONF = cfg.CONF
LOG = log.getLogger(__name__)


@six.add_metaclass(abc.ABCMeta)
[docs]class ProcessingHook(object): # pragma: no cover """Abstract base class for introspection data processing hooks."""
[docs] def before_processing(self, introspection_data, **kwargs): """Hook to run before any other data processing. This hook is run even before sanity checks. :param introspection_data: raw information sent by the ramdisk, may be modified by the hook. :param kwargs: used for extensibility without breaking existing hooks :returns: nothing. """
[docs] def before_update(self, introspection_data, node_info, **kwargs): """Hook to run before Ironic node update. This hook is run after node is found and ports are created, just before the node is updated with the data. :param introspection_data: processed data from the ramdisk. :param node_info: NodeInfo instance. :param kwargs: used for extensibility without breaking existing hooks. :returns: nothing. [RFC 6902] - http://tools.ietf.org/html/rfc6902 """
[docs]class WithValidation(object): REQUIRED_PARAMS = set() """Set with names of required parameters.""" OPTIONAL_PARAMS = set() """Set with names of optional parameters."""
[docs] def validate(self, params, **kwargs): """Validate params passed during creation. Default implementation checks for presence of fields from REQUIRED_PARAMS and fails for unexpected fields (not from REQUIRED_PARAMS + OPTIONAL_PARAMS). :param params: params as a dictionary :param kwargs: used for extensibility without breaking existing plugins :raises: ValueError on validation failure """ passed = {k for k, v in params.items() if v is not None} missing = self.REQUIRED_PARAMS - passed unexpected = passed - self.REQUIRED_PARAMS - self.OPTIONAL_PARAMS msg = [] if missing: msg.append(_('missing required parameter(s): %s') % ', '.join(missing)) if unexpected: msg.append(_('unexpected parameter(s): %s') % ', '.join(unexpected)) if msg: raise ValueError('; '.join(msg))
@six.add_metaclass(abc.ABCMeta)
[docs]class RuleConditionPlugin(WithValidation): # pragma: no cover """Abstract base class for rule condition plugins.""" REQUIRED_PARAMS = {'value'} ALLOW_NONE = False """Whether this condition accepts None when field is not found.""" @abc.abstractmethod
[docs] def check(self, node_info, field, params, **kwargs): """Check if condition holds for a given field. :param node_info: NodeInfo object :param field: field value :param params: parameters as a dictionary, changing it here will change what will be stored in database :param kwargs: used for extensibility without breaking existing plugins :raises ValueError: on unacceptable field value :returns: True if check succeeded, otherwise False """
@six.add_metaclass(abc.ABCMeta)
[docs]class RuleActionPlugin(WithValidation): # pragma: no cover """Abstract base class for rule action plugins.""" FORMATTED_PARAMS = [] """List of params will be formatted with python format.""" @abc.abstractmethod
[docs] def apply(self, node_info, params, **kwargs): """Run action on successful rule match. :param node_info: NodeInfo object :param params: parameters as a dictionary :param kwargs: used for extensibility without breaking existing plugins :raises: utils.Error on failure """
[docs] def rollback(self, node_info, params, **kwargs): """Rollback action effects from previous run on a failed match. Default implementation does nothing. :param node_info: NodeInfo object :param params: parameters as a dictionary :param kwargs: used for extensibility without breaking existing plugins :raises: utils.Error on failure """
_HOOKS_MGR = None _NOT_FOUND_HOOK_MGR = None _CONDITIONS_MGR = None _ACTIONS_MGR = None
[docs]def missing_entrypoints_callback(names): """Raise MissingHookError with comma-separated list of missing hooks""" missing_names = ', '.join(names) raise MissingHookError(missing_names)
[docs]def processing_hooks_manager(*args): """Create a Stevedore extension manager for processing hooks. :param args: arguments to pass to the hooks constructor. """ global _HOOKS_MGR if _HOOKS_MGR is None: names = [x.strip() for x in CONF.processing.processing_hooks.split(',') if x.strip()] _HOOKS_MGR = stevedore.NamedExtensionManager( 'ironic_inspector.hooks.processing', names=names, invoke_on_load=True, invoke_args=args, on_missing_entrypoints_callback=missing_entrypoints_callback, name_order=True) return _HOOKS_MGR
[docs]def node_not_found_hook_manager(*args): global _NOT_FOUND_HOOK_MGR if _NOT_FOUND_HOOK_MGR is None: name = CONF.processing.node_not_found_hook if name: _NOT_FOUND_HOOK_MGR = stevedore.DriverManager( 'ironic_inspector.hooks.node_not_found', name=name) return _NOT_FOUND_HOOK_MGR
[docs]def rule_conditions_manager(): """Create a Stevedore extension manager for conditions in rules.""" global _CONDITIONS_MGR if _CONDITIONS_MGR is None: _CONDITIONS_MGR = stevedore.ExtensionManager( 'ironic_inspector.rules.conditions', invoke_on_load=True) return _CONDITIONS_MGR
[docs]def rule_actions_manager(): """Create a Stevedore extension manager for actions in rules.""" global _ACTIONS_MGR if _ACTIONS_MGR is None: _ACTIONS_MGR = stevedore.ExtensionManager( 'ironic_inspector.rules.actions', invoke_on_load=True) for act in _ACTIONS_MGR: # a trick to detect if function was overridden if "rollback" in act.obj.__class__.__dict__: LOG.warning(_LW('Defining "rollback" for introspection rules ' 'actions is deprecated (action "%s")'), act.name) return _ACTIONS_MGR
[docs]class MissingHookError(KeyError): """Exception when hook is not found when processing it."""