senlin/senlin/engine/cluster.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.

from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import timeutils
import six

from senlin.common import consts
from senlin.common import exception
from senlin.engine import cluster_policy as cpm
from senlin.engine import health_manager
from senlin.engine import node as node_mod
from senlin.objects import cluster as co
from senlin.objects import cluster_policy as cpo
from senlin.objects import node as no
from senlin.policies import base as pcb
from senlin.profiles import base as pfb

LOG = logging.getLogger(__name__)

CONF = cfg.CONF


class Cluster(object):
    """A cluster is a collection of objects of the same profile type.

    All operations are performed without further checking because the
    checkings are supposed to be done before/after/during an action is
    executed.
    """

    def __init__(self, name, desired_capacity, profile_id,
                 context=None, **kwargs):
        '''Initialize a cluster object.

        The cluster defaults to have 0 node with no profile assigned.
        '''

        self.id = kwargs.get('id', None)
        self.name = name
        self.profile_id = profile_id

        # Initialize the fields using kwargs passed in
        self.user = kwargs.get('user', '')
        self.project = kwargs.get('project', '')
        self.domain = kwargs.get('domain', '')

        self.init_at = kwargs.get('init_at', None)
        self.created_at = kwargs.get('created_at', None)
        self.updated_at = kwargs.get('updated_at', None)

        self.min_size = (kwargs.get('min_size') or
                         consts.CLUSTER_DEFAULT_MIN_SIZE)
        self.max_size = (kwargs.get('max_size') or
                         consts.CLUSTER_DEFAULT_MAX_SIZE)
        self.desired_capacity = desired_capacity
        self.next_index = kwargs.get('next_index', 1)
        self.timeout = (kwargs.get('timeout') or
                        cfg.CONF.default_action_timeout)

        self.status = kwargs.get('status', consts.CS_INIT)
        self.status_reason = kwargs.get('status_reason', 'Initializing')
        self.data = kwargs.get('data', {})
        self.metadata = kwargs.get('metadata') or {}
        self.dependents = kwargs.get('dependents') or {}
        self.config = kwargs.get('config') or {}

        # rt is a dict for runtime data
        self.rt = {
            'profile': None,
            'nodes': [],
            'policies': []
        }

        if context is not None:
            self._load_runtime_data(context)

    def _load_runtime_data(self, context):
        if self.id is None:
            return

        policies = []
        bindings = cpo.ClusterPolicy.get_all(context, self.id)
        for b in bindings:
            # Detect policy type conflicts
            policy = pcb.Policy.load(context, b.policy_id, project_safe=False)
            policies.append(policy)

        self.rt = {
            'profile': pfb.Profile.load(context,
                                        profile_id=self.profile_id,
                                        project_safe=False),
            'nodes': no.Node.get_all_by_cluster(context, self.id),
            'policies': policies
        }

    def store(self, context):
        '''Store the cluster in database and return its ID.

        If the ID already exists, we do an update.
        '''

        values = {
            'name': self.name,
            'profile_id': self.profile_id,
            'user': self.user,
            'project': self.project,
            'domain': self.domain,
            'init_at': self.init_at,
            'created_at': self.created_at,
            'updated_at': self.updated_at,
            'min_size': self.min_size,
            'max_size': self.max_size,
            'desired_capacity': self.desired_capacity,
            'next_index': self.next_index,
            'timeout': self.timeout,
            'status': self.status,
            'status_reason': self.status_reason,
            'meta_data': self.metadata,
            'data': self.data,
            'dependents': self.dependents,
            'config': self.config,
        }

        timestamp = timeutils.utcnow(True)
        if self.id:
            values['updated_at'] = timestamp
            co.Cluster.update(context, self.id, values)
        else:
            self.init_at = timestamp
            values['init_at'] = timestamp
            cluster = co.Cluster.create(context, values)
            self.id = cluster.id

        self._load_runtime_data(context)
        return self.id

    @classmethod
    def _from_object(cls, context, obj):
        """Construct a cluster from database object.

        :param context: the context used for DB operations;
        :param obj: a DB cluster object that will receive all fields;
        """
        kwargs = {
            'id': obj.id,
            'user': obj.user,
            'project': obj.project,
            'domain': obj.domain,
            'init_at': obj.init_at,
            'created_at': obj.created_at,
            'updated_at': obj.updated_at,
            'min_size': obj.min_size,
            'max_size': obj.max_size,
            'next_index': obj.next_index,
            'timeout': obj.timeout,
            'status': obj.status,
            'status_reason': obj.status_reason,
            'data': obj.data,
            'metadata': obj.metadata,
            'dependents': obj.dependents,
            'config': obj.config,
        }

        return cls(obj.name, obj.desired_capacity, obj.profile_id,
                   context=context, **kwargs)

    @classmethod
    def load(cls, context, cluster_id=None, dbcluster=None, project_safe=True):
        '''Retrieve a cluster from database.'''
        if dbcluster is None:
            dbcluster = co.Cluster.get(context, cluster_id,
                                       project_safe=project_safe)
            if dbcluster is None:
                raise exception.ResourceNotFound(type='cluster', id=cluster_id)

        return cls._from_object(context, dbcluster)

    @classmethod
    def load_all(cls, context, limit=None, marker=None, sort=None,
                 filters=None, project_safe=True):
        """Retrieve all clusters from database."""

        objs = co.Cluster.get_all(context, limit=limit, marker=marker,
                                  sort=sort, filters=filters,
                                  project_safe=project_safe)

        for obj in objs:
            cluster = cls._from_object(context, obj)
            yield cluster

    def set_status(self, context, status, reason=None, **kwargs):
        """Set status of the cluster.

        :param context: A DB session for accessing the backend database.
        :param status: A string providing the new status of the cluster.
        :param reason: A string containing the reason for the status change.
                       It can be omitted when invoking this method.
        :param dict kwargs: Other optional attributes to be updated.
        :returns: Nothing.
        """

        values = {}
        now = timeutils.utcnow(True)
        if status == consts.CS_ACTIVE and self.status == consts.CS_CREATING:
            self.created_at = now
            values['created_at'] = now
        elif (status == consts.CS_ACTIVE and
              self.status in (consts.CS_UPDATING, consts.CS_RESIZING)):
            self.updated_at = now
            values['updated_at'] = now

        self.status = status
        values['status'] = status
        if reason:
            self.status_reason = reason
            values['status_reason'] = reason

        for k, v in kwargs.items():
            if hasattr(self, k):
                setattr(self, k, v)
                values[k] = v

        # There is a possibility that the profile id is changed
        if 'profile_id' in values:
            profile = pfb.Profile.load(context, profile_id=self.profile_id)
            self.rt['profile'] = profile
        co.Cluster.update(context, self.id, values)
        return

    def do_create(self, context, **kwargs):
        '''Additional logic at the beginning of cluster creation process.

        Set cluster status to CREATING.
        '''
        if self.status != consts.CS_INIT:
            LOG.error('Cluster is in status "%s"', self.status)
            return False

        self.set_status(context, consts.CS_CREATING, 'Creation in progress')
        try:
            pfb.Profile.create_cluster_object(context, self)
        except exception.EResourceCreation as ex:
            self.set_status(context, consts.CS_ERROR, six.text_type(ex))
            return False

        return True

    def do_delete(self, context, **kwargs):
        """Additional logic at the end of cluster deletion process."""
        self.set_status(context, consts.CS_DELETING, 'Deletion in progress')

        try:
            pfb.Profile.delete_cluster_object(context, self)
        except exception.EResourceDeletion as ex:
            self.set_status(context, consts.CS_ERROR, six.text_type(ex))
            return False

        co.Cluster.delete(context, self.id)
        return True

    def do_update(self, context, **kwargs):
        '''Additional logic at the beginning of cluster updating progress.

        This method is intended to be called only from an action.
        '''
        self.set_status(context, consts.CS_UPDATING, 'Update in progress')
        return True

    def do_check(self, context, **kwargs):
        """Additional logic at the beginning of cluster checking process.

        Set cluster status to CHECKING.
        """
        self.set_status(context, consts.CS_CHECKING, 'Check in progress')
        return True

    def do_recover(self, context, **kwargs):
        '''Additional logic at the beginning of cluster recovering process.

        Set cluster status to RECOVERING.
        '''
        self.set_status(context, consts.CS_RECOVERING, 'Recovery in progress')
        return True

    def do_operation(self, context, **kwargs):
        """Additional logic at the beginning of cluster recovering process.

        Set cluster status to OPERATING.
        """
        operation = kwargs.get("operation", "unknown")
        self.set_status(context, consts.CS_OPERATING,
                        "Operation %s in progress" % operation)
        return True

    def attach_policy(self, ctx, policy_id, values):
        """Attach policy object to the cluster.

        Note this method MUST be called with the cluster locked.

        :param ctx: A context for DB operation.
        :param policy_id: ID of the policy object.
        :param values: Optional dictionary containing binding properties.

        :returns: A tuple containing a boolean result and a reason string.
        """

        policy = pcb.Policy.load(ctx, policy_id)
        # Check if policy has already been attached
        for existing in self.rt['policies']:
            # Policy already attached
            if existing.id == policy_id:
                return True, 'Policy already attached.'

            # Detect policy type conflicts
            if (existing.type == policy.type) and policy.singleton:
                reason = ("Only one instance of policy type (%(ptype)s) can "
                          "be attached to a cluster, but another instance "
                          "(%(existing)s) is found attached to the cluster "
                          "(%(cluster)s) already."
                          ) % {'ptype': policy.type,
                               'existing': existing.id,
                               'cluster': self.id}
                return False, reason

        # invoke policy callback
        enabled = bool(values.get('enabled', True))
        res, data = policy.attach(self, enabled=enabled)
        if not res:
            return False, data

        kwargs = {
            'enabled': enabled,
            'data': data,
            'priority': policy.PRIORITY
        }

        cp = cpm.ClusterPolicy(self.id, policy_id, **kwargs)
        cp.store(ctx)

        # refresh cached runtime
        self.rt['policies'].append(policy)

        return True, 'Policy attached.'

    def update_policy(self, ctx, policy_id, **values):
        """Update a policy that is already attached to a cluster.

        Note this method must be called with the cluster locked.
        :param ctx: A context for DB operation.
        :param policy_id: ID of the policy object.
        :param values: Optional dictionary containing new binding properties.

        :returns: A tuple containing a boolean result and a string reason.
        """
        # Check if policy has already been attached
        found = False
        for existing in self.policies:
            if existing.id == policy_id:
                found = True
                break
        if not found:
            return False, 'Policy not attached.'

        enabled = values.get('enabled', None)
        if enabled is None:
            return True, 'No update is needed.'

        params = {'enabled': bool(enabled)}
        # disable health check if necessary
        policy_type = existing.type.split('-')[0]
        if policy_type == 'senlin.policy.health':
            if enabled is True:
                health_manager.enable(self.id)
            else:
                health_manager.disable(self.id)

        cpo.ClusterPolicy.update(ctx, self.id, policy_id, params)
        return True, 'Policy updated.'

    def detach_policy(self, ctx, policy_id):
        """Detach policy object from the cluster.

        Note this method MUST be called with the cluster locked.

        :param ctx: A context for DB operation.
        :param policy_id: ID of the policy object.

        :returns: A tuple containing a boolean result and a reason string.
        """
        # Check if policy has already been attached
        found = None
        for existing in self.policies:
            if existing.id == policy_id:
                found = existing
                break
        if found is None:
            return False, 'Policy not attached.'

        policy = pcb.Policy.load(ctx, policy_id)
        res, reason = policy.detach(self)
        if not res:
            return res, reason

        cpo.ClusterPolicy.delete(ctx, self.id, policy_id)
        self.rt['policies'].remove(found)

        return True, 'Policy detached.'

    @property
    def nodes(self):
        return self.rt['nodes']

    def add_node(self, node):
        """Append specified node to the cluster cache.

        :param node: The node to become a new member of the cluster.
        """
        self.rt['nodes'].append(node)

    def remove_node(self, node_id):
        """Remove node with specified ID from cache.

        :param node_id: ID of the node to be removed from cache.
        """
        for node in self.rt['nodes']:
            if node.id == node_id:
                self.rt['nodes'].remove(node)

    def update_node(self, nodes):
        """Update cluster runtime data

        :param nodes: List of node objects
        """
        self.rt['nodes'] = nodes

    @property
    def policies(self):
        return self.rt['policies']

    def get_region_distribution(self, regions):
        """Get node distribution regarding given regions.

        :param regions: list of region names to check.
        :return: a dict containing region and number as key value pairs.
        """
        dist = dict.fromkeys(regions, 0)

        for node in self.nodes:
            placement = node.data.get('placement', {})
            if placement:
                region = placement.get('region_name', None)
                if region and region in regions:
                    dist[region] += 1

        return dist

    def get_zone_distribution(self, ctx, zones):
        """Get node distribution regarding the given the availability zones.

        The availability zone information is only available for some profiles.

        :param ctx: context used to access node details.
        :param zones: list of zone names to check.
        :returns: a dict containing zone and number as key-value pairs.
        """
        dist = dict.fromkeys(zones, 0)

        for node in self.nodes:
            placement = node.data.get('placement', {})
            if placement and 'zone' in placement:
                zone = placement['zone']
                dist[zone] += 1
            else:
                details = node.get_details(ctx)
                zname = details.get('OS-EXT-AZ:availability_zone', None)
                if zname and zname in dist:
                    dist[zname] += 1

        return dist

    def nodes_by_region(self, region):
        """Get list of nodes that belong to the specified region.

        :param region: Name of region for filtering.
        :return: A list of nodes that are from the specified region.
        """
        result = []
        for node in self.nodes:
            placement = node.data.get('placement', {})
            if placement and 'region_name' in placement:
                if region == placement['region_name']:
                    result.append(node)
        return result

    def nodes_by_zone(self, zone):
        """Get list of nodes that reside in the specified availability zone.

        :param zone: Name of availability zone for filtering.
        :return: A list of nodes that reside in the specified AZ.
        """
        result = []
        for node in self.nodes:
            placement = node.data.get('placement', {})
            if placement and 'zone' in placement:
                if zone == placement['zone']:
                    result.append(node)
        return result

    def health_check(self, ctx):
        """Check physical resources status

        :param ctx: The context to operate node object
        """
        # Note this procedure is a pure sequential operation,
        # its not suitable for large scale clusters.

        old_nodes = self.nodes
        for node in old_nodes:
            node.do_check(ctx)

        nodes = node_mod.Node.load_all(ctx, cluster_id=self.id)
        self.update_node([n for n in nodes])

    def eval_status(self, ctx, operation, **params):
        """Re-evaluate cluster's health status.

        :param ctx: The requesting context.
        :param operation: The operation that triggers this status evaluation.
        :returns: ``None``.
        """
        nodes = node_mod.Node.load_all(ctx, cluster_id=self.id)
        self.rt['nodes'] = [n for n in nodes]

        active_count = 0
        for node in self.nodes:
            if node.status == consts.NS_ACTIVE:
                active_count += 1

        # get provided desired_capacity/min_size/max_size
        desired = params.get('desired_capacity', self.desired_capacity)
        min_size = params.get('min_size', self.min_size)
        max_size = params.get('max_size', self.max_size)

        values = params or {}
        if active_count < min_size:
            status = consts.CS_ERROR
            reason = ("%(o)s: number of active nodes is below min_size "
                      "(%(n)d).") % {'o': operation, 'n': min_size}
        elif active_count < desired:
            status = consts.CS_WARNING
            reason = ("%(o)s: number of active nodes is below "
                      "desired_capacity "
                      "(%(n)d).") % {'o': operation, 'n': desired}
        elif max_size < 0 or active_count <= max_size:
            status = consts.CS_ACTIVE
            reason = ("%(o)s: number of active nodes is equal or above "
                      "desired_capacity "
                      "(%(n)d).") % {'o': operation, 'n': desired}
        else:
            status = consts.CS_WARNING
            reason = ("%(o)s: number of active nodes is above max_size "
                      "(%(n)d).") % {'o': operation, 'n': max_size}

        values.update({'status': status, 'status_reason': reason})
        co.Cluster.update(ctx, self.id, values)