path: root/ironic/conductor/base_manager.py

#    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 conductor manager functionality."""

import inspect
import threading

import eventlet
import futurist
from futurist import periodics
from futurist import rejection
from ironic_lib import mdns
from oslo_db import exception as db_exception
from oslo_log import log
from oslo_utils import excutils
from oslo_utils import versionutils

from ironic.common import context as ironic_context
from ironic.common import driver_factory
from ironic.common import exception
from ironic.common import hash_ring
from ironic.common.i18n import _
from ironic.common import release_mappings as versions
from ironic.common import rpc
from ironic.common import states
from ironic.conductor import allocations
from ironic.conductor import notification_utils as notify_utils
from ironic.conductor import task_manager
from ironic.conf import CONF
from ironic.db import api as dbapi
from ironic.drivers import base as driver_base
from ironic.drivers.modules import deploy_utils
from ironic import objects
from ironic.objects import fields as obj_fields


LOG = log.getLogger(__name__)


def _check_enabled_interfaces():
    """Sanity-check enabled_*_interfaces configs.

    We do this before we even bother to try to load up drivers. If we have any
    dynamic drivers enabled, then we need interfaces enabled as well.

    :raises: ConfigInvalid if an enabled interfaces config option is empty.
    """
    empty_confs = []
    iface_types = ['enabled_%s_interfaces' % i
                   for i in driver_base.ALL_INTERFACES]
    for iface_type in iface_types:
        conf_value = getattr(CONF, iface_type)
        if not conf_value:
            empty_confs.append(iface_type)
    if empty_confs:
        msg = (_('Configuration options %s cannot be an empty list.') %
               ', '.join(empty_confs))
        raise exception.ConfigInvalid(error_msg=msg)


class BaseConductorManager(object):

    def __init__(self, host, topic):
        super(BaseConductorManager, self).__init__()
        if not host:
            host = CONF.host
        self.host = host
        self.topic = topic
        self.sensors_notifier = rpc.get_sensors_notifier()
        self._started = False
        self._shutdown = None
        self._zeroconf = None

    def init_host(self, admin_context=None):
        """Initialize the conductor host.

        :param admin_context: the admin context to pass to periodic tasks.
        :raises: RuntimeError when conductor is already running.
        :raises: NoDriversLoaded when no drivers are enabled on the conductor.
        :raises: DriverNotFound if a driver is enabled that does not exist.
        :raises: DriverLoadError if an enabled driver cannot be loaded.
        :raises: DriverNameConflict if a classic driver and a dynamic driver
                 are both enabled and have the same name.
        """
        if self._started:
            raise RuntimeError(_('Attempt to start an already running '
                                 'conductor manager'))
        self._shutdown = False

        self.dbapi = dbapi.get_instance()

        self._keepalive_evt = threading.Event()
        """Event for the keepalive thread."""

        # TODO(dtantsur): make the threshold configurable?
        rejection_func = rejection.reject_when_reached(
            CONF.conductor.workers_pool_size)
        self._executor = futurist.GreenThreadPoolExecutor(
            max_workers=CONF.conductor.workers_pool_size,
            check_and_reject=rejection_func)
        """Executor for performing tasks async."""

        # TODO(jroll) delete the use_groups argument and use the default
        # in Stein.
        self.ring_manager = hash_ring.HashRingManager(
            use_groups=self._use_groups())
        """Consistent hash ring which maps drivers to conductors."""

        _check_enabled_interfaces()

        # NOTE(deva): these calls may raise DriverLoadError or DriverNotFound
        # NOTE(vdrok): Instantiate network and storage interface factory on
        # startup so that all the interfaces are loaded at the very
        # beginning, and failures prevent the conductor from starting.
        hardware_types = driver_factory.hardware_types()
        driver_factory.NetworkInterfaceFactory()
        driver_factory.StorageInterfaceFactory()

        # NOTE(jroll) this is passed to the dbapi, which requires a list, not
        # a generator (which keys() returns in py3)
        hardware_type_names = list(hardware_types)

        # check that at least one driver is loaded, whether classic or dynamic
        if not hardware_type_names:
            msg = ("Conductor %s cannot be started because no hardware types "
                   "were specified in the 'enabled_hardware_types' config "
                   "option.")
            LOG.error(msg, self.host)
            raise exception.NoDriversLoaded(conductor=self.host)

        self._collect_periodic_tasks(admin_context)

        # clear all target_power_state with locks by this conductor
        self.dbapi.clear_node_target_power_state(self.host)
        # clear all locks held by this conductor before registering
        self.dbapi.clear_node_reservations_for_conductor(self.host)
        try:
            # Register this conductor with the cluster
            self.conductor = objects.Conductor.register(
                admin_context, self.host, hardware_type_names,
                CONF.conductor.conductor_group)
        except exception.ConductorAlreadyRegistered:
            # This conductor was already registered and did not shut down
            # properly, so log a warning and update the record.
            LOG.warning("A conductor with hostname %(hostname)s was "
                        "previously registered. Updating registration",
                        {'hostname': self.host})
            self.conductor = objects.Conductor.register(
                admin_context, self.host, hardware_type_names,
                CONF.conductor.conductor_group, update_existing=True)

        # register hardware types and interfaces supported by this conductor
        # and validate them against other conductors
        try:
            self._register_and_validate_hardware_interfaces(hardware_types)
        except (exception.DriverLoadError, exception.DriverNotFound,
                exception.ConductorHardwareInterfacesAlreadyRegistered,
                exception.InterfaceNotFoundInEntrypoint,
                exception.NoValidDefaultForInterface) as e:
            with excutils.save_and_reraise_exception():
                LOG.error('Failed to register hardware types. %s', e)
                self.del_host()

        # Start periodic tasks
        self._periodic_tasks_worker = self._executor.submit(
            self._periodic_tasks.start, allow_empty=True)
        self._periodic_tasks_worker.add_done_callback(
            self._on_periodic_tasks_stop)

        for state in states.STUCK_STATES_TREATED_AS_FAIL:
            self._fail_transient_state(
                state,
                _("The %(state)s state can't be resumed by conductor "
                  "%(host)s. Moving to fail state.") %
                {'state': state, 'host': self.host})

        # Start consoles if it set enabled in a greenthread.
        try:
            self._spawn_worker(self._start_consoles,
                               ironic_context.get_admin_context())
        except exception.NoFreeConductorWorker:
            LOG.warning('Failed to start worker for restarting consoles.')

        # Spawn a dedicated greenthread for the keepalive
        try:
            self._spawn_worker(self._conductor_service_record_keepalive)
            LOG.info('Successfully started conductor with hostname '
                     '%(hostname)s.',
                     {'hostname': self.host})
        except exception.NoFreeConductorWorker:
            with excutils.save_and_reraise_exception():
                LOG.critical('Failed to start keepalive')
                self.del_host()

        # Resume allocations that started before the restart.
        try:
            self._spawn_worker(self._resume_allocations,
                               ironic_context.get_admin_context())
        except exception.NoFreeConductorWorker:
            LOG.warning('Failed to start worker for resuming allocations.')

        if CONF.conductor.enable_mdns:
            self._publish_endpoint()

        self._started = True

    def _use_groups(self):
        release_ver = versions.RELEASE_MAPPING.get(CONF.pin_release_version)
        # NOTE(jroll) self.RPC_API_VERSION is actually defined in a subclass,
        # but we only use this class from there.
        version_cap = (release_ver['rpc'] if release_ver
                       else self.RPC_API_VERSION)
        return versionutils.is_compatible('1.47', version_cap)

    def _fail_transient_state(self, state, last_error):
        """Apply "fail" transition to nodes in a transient state.

        If the conductor server dies abruptly mid deployment or cleaning
        (OMM Killer, power outage, etc...) we can not resume the process even
        if the conductor comes back online. Cleaning the reservation of
        the nodes (dbapi.clear_node_reservations_for_conductor) is not enough
        to unstick it, so let's gracefully fail the process.
        """
        filters = {'reserved': False, 'provision_state': state}
        self._fail_if_in_state(ironic_context.get_admin_context(), filters,
                               state, 'provision_updated_at',
                               last_error=last_error)

    def _collect_periodic_tasks(self, admin_context):
        """Collect driver-specific periodic tasks.

        Conductor periodic tasks accept context argument, driver periodic
        tasks accept this manager and context. We have to ensure that the
        same driver interface class is not traversed twice, otherwise
        we'll have several instances of the same task.

        :param admin_context: Administrator context to pass to tasks.
        """
        LOG.debug('Collecting periodic tasks')
        # collected callables
        periodic_task_callables = []
        # list of visited classes to avoid adding the same tasks twice
        periodic_task_classes = set()

        def _collect_from(obj, args):
            """Collect tasks from the given object.

            :param obj: the object to collect tasks from.
            :param args: a tuple of arguments to pass to tasks.
            """
            if obj and obj.__class__ not in periodic_task_classes:
                for name, member in inspect.getmembers(obj):
                    if periodics.is_periodic(member):
                        LOG.debug('Found periodic task %(owner)s.%(member)s',
                                  {'owner': obj.__class__.__name__,
                                   'member': name})
                        periodic_task_callables.append((member, args, {}))
                periodic_task_classes.add(obj.__class__)

        # First, collect tasks from the conductor itself
        _collect_from(self, (admin_context,))

        # Second, collect tasks from hardware interfaces
        for ifaces in driver_factory.all_interfaces().values():
            for iface in ifaces.values():
                _collect_from(iface, args=(self, admin_context))
        # TODO(dtantsur): allow periodics on hardware types themselves?

        if len(periodic_task_callables) > CONF.conductor.workers_pool_size:
            LOG.warning('This conductor has %(tasks)d periodic tasks '
                        'enabled, but only %(workers)d task workers '
                        'allowed by [conductor]workers_pool_size option',
                        {'tasks': len(periodic_task_callables),
                         'workers': CONF.conductor.workers_pool_size})

        self._periodic_tasks = periodics.PeriodicWorker(
            periodic_task_callables,
            executor_factory=periodics.ExistingExecutor(self._executor))
        # This is only used in tests currently. Delete it?
        self._periodic_task_callables = periodic_task_callables

    def del_host(self, deregister=True):
        # Conductor deregistration fails if called on non-initialized
        # conductor (e.g. when rpc server is unreachable).
        if not hasattr(self, 'conductor'):
            return
        self._shutdown = True
        self._keepalive_evt.set()
        if deregister:
            try:
                # Inform the cluster that this conductor is shutting down.
                # Note that rebalancing will not occur immediately, but when
                # the periodic sync takes place.
                self.conductor.unregister()
                LOG.info('Successfully stopped conductor with hostname '
                         '%(hostname)s.',
                         {'hostname': self.host})
            except exception.ConductorNotFound:
                pass
        else:
            LOG.info('Not deregistering conductor with hostname %(hostname)s.',
                     {'hostname': self.host})
        # Waiting here to give workers the chance to finish. This has the
        # benefit of releasing locks workers placed on nodes, as well as
        # having work complete normally.
        self._periodic_tasks.stop()
        self._periodic_tasks.wait()
        self._executor.shutdown(wait=True)

        if self._zeroconf is not None:
            self._zeroconf.close()
            self._zeroconf = None

        self._started = False

    def _register_and_validate_hardware_interfaces(self, hardware_types):
        """Register and validate hardware interfaces for this conductor.

        Registers a row in the database for each combination of
        (hardware type, interface type, interface) that is supported and
        enabled.

        TODO: Validates against other conductors to check if the
        set of registered hardware interfaces for a given hardware type is the
        same, and warns if not (we can't error out, otherwise all conductors
        must be restarted at once to change configuration).

        :param hardware_types: Dictionary mapping hardware type name to
                               hardware type object.
        :raises: ConductorHardwareInterfacesAlreadyRegistered
        :raises: InterfaceNotFoundInEntrypoint
        :raises: NoValidDefaultForInterface if the default value cannot be
                 calculated and is not provided in the configuration
        """
        # first unregister, in case we have cruft laying around
        self.conductor.unregister_all_hardware_interfaces()

        for ht_name, ht in hardware_types.items():
            interface_map = driver_factory.enabled_supported_interfaces(ht)
            for interface_type, interface_names in interface_map.items():
                default_interface = driver_factory.default_interface(
                    ht, interface_type, driver_name=ht_name)
                self.conductor.register_hardware_interfaces(ht_name,
                                                            interface_type,
                                                            interface_names,
                                                            default_interface)

        # TODO(jroll) validate against other conductor, warn if different
        # how do we do this performantly? :|

    def _on_periodic_tasks_stop(self, fut):
        try:
            fut.result()
        except Exception as exc:
            LOG.critical('Periodic tasks worker has failed: %s', exc)
        else:
            LOG.info('Successfully shut down periodic tasks')

    def iter_nodes(self, fields=None, **kwargs):
        """Iterate over nodes mapped to this conductor.

        Requests node set from and filters out nodes that are not
        mapped to this conductor.

        Yields tuples (node_uuid, driver, conductor_group, ...) where ... is
        derived from fields argument, e.g.: fields=None means yielding ('uuid',
        'driver', 'conductor_group'), fields=['foo'] means yielding ('uuid',
        'driver', 'conductor_group', 'foo').

        :param fields: list of fields to fetch in addition to uuid, driver,
                       and conductor_group
        :param kwargs: additional arguments to pass to dbapi when looking for
                       nodes
        :return: generator yielding tuples of requested fields
        """
        columns = ['uuid', 'driver', 'conductor_group'] + list(fields or ())
        node_list = self.dbapi.get_nodeinfo_list(columns=columns, **kwargs)
        for result in node_list:
            if self._shutdown:
                break
            if self._mapped_to_this_conductor(*result[:3]):
                yield result

    def _spawn_worker(self, func, *args, **kwargs):

        """Create a greenthread to run func(*args, **kwargs).

        Spawns a greenthread if there are free slots in pool, otherwise raises
        exception. Execution control returns immediately to the caller.

        :returns: Future object.
        :raises: NoFreeConductorWorker if worker pool is currently full.

        """
        try:
            return self._executor.submit(func, *args, **kwargs)
        except futurist.RejectedSubmission:
            raise exception.NoFreeConductorWorker()

    def _conductor_service_record_keepalive(self):
        while not self._keepalive_evt.is_set():
            try:
                self.conductor.touch()
            except db_exception.DBConnectionError:
                LOG.warning('Conductor could not connect to database '
                            'while heartbeating.')
            except Exception as e:
                LOG.exception('Error while heartbeating. Error: %(err)s',
                              {'err': e})
            self._keepalive_evt.wait(CONF.conductor.heartbeat_interval)

    def _mapped_to_this_conductor(self, node_uuid, driver, conductor_group):
        """Check that node is mapped to this conductor.

        Note that because mappings are eventually consistent, it is possible
        for two conductors to simultaneously believe that a node is mapped to
        them. Any operation that depends on exclusive control of a node should
        take out a lock.
        """
        try:
            ring = self.ring_manager.get_ring(driver, conductor_group)
        except exception.DriverNotFound:
            return False

        return self.host in ring.get_nodes(node_uuid.encode('utf-8'))

    def _fail_if_in_state(self, context, filters, provision_state,
                          sort_key, callback_method=None,
                          err_handler=None, last_error=None,
                          keep_target_state=False):
        """Fail nodes that are in specified state.

        Retrieves nodes that satisfy the criteria in 'filters'.
        If any of these nodes is in 'provision_state', it has failed
        in whatever provisioning activity it was currently doing.
        That failure is processed here.

        :param context: request context
        :param filters: criteria (as a dictionary) to get the desired
                        list of nodes that satisfy the filter constraints.
                        For example, if filters['provisioned_before'] = 60,
                        this would process nodes whose provision_updated_at
                        field value was 60 or more seconds before 'now'.
        :param provision_state: provision_state that the node is in,
                                for the provisioning activity to have failed,
                                either one string or a set.
        :param sort_key: the nodes are sorted based on this key.
        :param callback_method: the callback method to be invoked in a
                                spawned thread, for a failed node. This
                                method must take a :class:`TaskManager` as
                                the first (and only required) parameter.
        :param err_handler: for a failed node, the error handler to invoke
                            if an error occurs trying to spawn an thread
                            to do the callback_method.
        :param last_error: the error message to be updated in node.last_error
        :param keep_target_state: if True, a failed node will keep the same
                                  target provision state it had before the
                                  failure. Otherwise, the node's target
                                  provision state will be determined by the
                                  fsm.

        """
        if isinstance(provision_state, str):
            provision_state = {provision_state}

        node_iter = self.iter_nodes(filters=filters,
                                    sort_key=sort_key,
                                    sort_dir='asc')
        desired_maintenance = filters.get('maintenance')
        workers_count = 0
        for node_uuid, driver, conductor_group in node_iter:
            try:
                with task_manager.acquire(context, node_uuid,
                                          purpose='node state check') as task:
                    # Check maintenance value since it could have changed
                    # after the filtering was done.
                    if (desired_maintenance is not None
                            and desired_maintenance != task.node.maintenance):
                        continue

                    if task.node.provision_state not in provision_state:
                        continue

                    target_state = (None if not keep_target_state else
                                    task.node.target_provision_state)

                    # timeout has been reached - process the event 'fail'
                    if callback_method:
                        task.process_event('fail',
                                           callback=self._spawn_worker,
                                           call_args=(callback_method, task),
                                           err_handler=err_handler,
                                           target_state=target_state)
                    else:
                        task.node.last_error = last_error
                        task.process_event('fail', target_state=target_state)
            except exception.NoFreeConductorWorker:
                break
            except (exception.NodeLocked, exception.NodeNotFound):
                continue
            workers_count += 1
            if workers_count >= CONF.conductor.periodic_max_workers:
                break

    def _start_consoles(self, context):
        """Start consoles if set enabled.

        :param context: request context
        """
        filters = {'console_enabled': True}

        node_iter = self.iter_nodes(filters=filters)

        for node_uuid, driver, conductor_group in node_iter:
            try:
                with task_manager.acquire(context, node_uuid, shared=False,
                                          purpose='start console') as task:
                    notify_utils.emit_console_notification(
                        task, 'console_restore',
                        obj_fields.NotificationStatus.START)
                    try:
                        LOG.debug('Trying to start console of node %(node)s',
                                  {'node': node_uuid})
                        task.driver.console.start_console(task)
                        LOG.info('Successfully started console of node '
                                 '%(node)s', {'node': node_uuid})
                        notify_utils.emit_console_notification(
                            task, 'console_restore',
                            obj_fields.NotificationStatus.END)
                    except Exception as err:
                        msg = (_('Failed to start console of node %(node)s '
                                 'while starting the conductor, so changing '
                                 'the console_enabled status to False, error: '
                                 '%(err)s')
                               % {'node': node_uuid, 'err': err})
                        LOG.error(msg)
                        # If starting console failed, set node console_enabled
                        # back to False and set node's last error.
                        task.node.last_error = msg
                        task.node.console_enabled = False
                        task.node.save()
                        notify_utils.emit_console_notification(
                            task, 'console_restore',
                            obj_fields.NotificationStatus.ERROR)
            except exception.NodeLocked:
                LOG.warning('Node %(node)s is locked while trying to '
                            'start console on conductor startup',
                            {'node': node_uuid})
                continue
            except exception.NodeNotFound:
                LOG.warning("During starting console on conductor "
                            "startup, node %(node)s was not found",
                            {'node': node_uuid})
                continue
            finally:
                # Yield on every iteration
                eventlet.sleep(0)

    def _resume_allocations(self, context):
        """Resume unfinished allocations on restart."""
        filters = {'state': states.ALLOCATING,
                   'conductor_affinity': self.conductor.id}
        for allocation in objects.Allocation.list(context, filters=filters):
            LOG.debug('Resuming unfinished allocation %s', allocation.uuid)
            allocations.do_allocate(context, allocation)

    def _publish_endpoint(self):
        params = {}
        if CONF.debug:
            params['ipa_debug'] = True
        self._zeroconf = mdns.Zeroconf()
        self._zeroconf.register_service('baremetal',
                                        deploy_utils.get_ironic_api_url(),
                                        params=params)