# 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 stevedore
from ironic_inspector.common.i18n import _
CONF = cfg.CONF
LOG = log.getLogger(__name__)
[docs]
class ProcessingHook(object, metaclass=abc.ABCMeta): # pragma: no cover
"""Abstract base class for introspection data processing hooks."""
dependencies = []
"""An ordered list of hooks that must be enabled before this one.
The items here should be entry point names, not classes.
"""
[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))
[docs]
class RuleConditionPlugin(WithValidation, metaclass=abc.ABCMeta): # pragma: no cover # noqa
"""Abstract base class for rule condition plugins."""
REQUIRED_PARAMS = {'value'}
ALLOW_NONE = False
"""Whether this condition accepts None when field is not found."""
[docs]
@abc.abstractmethod
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
"""
[docs]
class RuleActionPlugin(WithValidation, metaclass=abc.ABCMeta): # pragma: no cover # noqa
"""Abstract base class for rule action plugins."""
FORMATTED_PARAMS = []
"""List of params will be formatted with python format."""
[docs]
@abc.abstractmethod
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
"""
_HOOKS_MGR = None
_NOT_FOUND_HOOK_MGR = None
_CONDITIONS_MGR = None
_ACTIONS_MGR = None
_INTROSPECTION_DATA_MGR = None
[docs]
def reset():
"""Reset cached managers."""
global _HOOKS_MGR
global _NOT_FOUND_HOOK_MGR
global _CONDITIONS_MGR
global _ACTIONS_MGR
global _INTROSPECTION_DATA_MGR
_HOOKS_MGR = None
_NOT_FOUND_HOOK_MGR = None
_CONDITIONS_MGR = None
_ACTIONS_MGR = None
_INTROSPECTION_DATA_MGR = None
[docs]
def missing_entrypoints_callback(names):
"""Raise MissingHookError with comma-separated list of missing hooks"""
error = _('The following hook(s) are missing or failed to load: %s')
raise RuntimeError(error % ', '.join(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 validate_processing_hooks():
"""Validate the enabled processing hooks.
:raises: MissingHookError on missing or failed to load hooks
:raises: RuntimeError on validation failure
:returns: the list of hooks passed validation
"""
hooks = [ext for ext in processing_hooks_manager()]
enabled = set()
errors = []
for hook in hooks:
deps = getattr(hook.obj, 'dependencies', ())
missing = [d for d in deps if d not in enabled]
if missing:
errors.append('Hook %(hook)s requires the following hooks to be '
'enabled before it: %(deps)s. The following hooks '
'are missing: %(missing)s.' %
{'hook': hook.name,
'deps': ', '.join(deps),
'missing': ', '.join(missing)})
enabled.add(hook.name)
if errors:
raise RuntimeError("Some hooks failed to load due to dependency "
"problems:\n%s" % "\n".join(errors))
return hooks
[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)
return _ACTIONS_MGR
[docs]
def introspection_data_manager():
global _INTROSPECTION_DATA_MGR
if _INTROSPECTION_DATA_MGR is None:
_INTROSPECTION_DATA_MGR = stevedore.ExtensionManager(
'ironic_inspector.introspection_data.store',
invoke_on_load=True)
return _INTROSPECTION_DATA_MGR
