Add Audit Scope Handler

This patch set adds audit scope mechanism.
It also removes host_aggregate field.

Change-Id: Ia98ed180a93fc8c19599735e2b41471d322bae9a
Partially-Implements: blueprint define-the-audit-scope

watcher/decision_engine/audit/base.py

@@ -80,7 +80,7 @@ class AuditHandler(BaseAuditHandler):
                    'audit_status': status}
         self.messaging.publish_status_event(event.type.name, payload)
 
-    def update_audit_state(self, request_context, audit, state):
+    def update_audit_state(self, audit, state):
         LOG.debug("Update audit state: %s", state)
         audit.state = state
         audit.save()
@@ -89,15 +89,13 @@ class AuditHandler(BaseAuditHandler):
     def pre_execute(self, audit, request_context):
         LOG.debug("Trigger audit %s", audit.uuid)
         # change state of the audit to ONGOING
-        self.update_audit_state(request_context, audit,
-                                audit_objects.State.ONGOING)
+        self.update_audit_state(audit, audit_objects.State.ONGOING)
 
     def post_execute(self, audit, solution, request_context):
         self.planner.schedule(request_context, audit.id, solution)
 
         # change state of the audit to SUCCEEDED
-        self.update_audit_state(request_context, audit,
-                                audit_objects.State.SUCCEEDED)
+        self.update_audit_state(audit, audit_objects.State.SUCCEEDED)
 
     def execute(self, audit, request_context):
         try:
@@ -106,5 +104,4 @@ class AuditHandler(BaseAuditHandler):
             self.post_execute(audit, solution, request_context)
         except Exception as e:
             LOG.exception(e)
-            self.update_audit_state(request_context, audit,
-                                    audit_objects.State.FAILED)
+            self.update_audit_state(audit, audit_objects.State.FAILED)

watcher/decision_engine/model/collector/base.py

@@ -31,9 +31,8 @@ to know the current relationships between the different :ref:`resources
 during an :ref:`Audit <audit_definition>` and enables the :ref:`Strategy
 <strategy_definition>` to request information such as:
 
-- What compute nodes are in a given :ref:`Availability Zone
-  <availability_zone_definition>` or a given :ref:`Host Aggregate
-  <host_aggregates_definition>`?
+- What compute nodes are in a given :ref:`Audit Scope
+  <audit_scope_definition>`?
 - What :ref:`Instances <instance_definition>` are hosted on a given compute
   node?
 - What is the current load of a compute node?

watcher/decision_engine/scope/base.py

@@ -0,0 +1,38 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2016 Servionica
+#
+# 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.
+#
+
+import abc
+import six
+
+from watcher.common import context
+
+
+@six.add_metaclass(abc.ABCMeta)
+class BaseScope(object):
+    """A base class for Scope mechanism
+
+    Child of this class is called when audit launches strategy. This strategy
+    requires Cluster Data Model which can be segregated to achieve audit scope.
+    """
+
+    def __init__(self, scope):
+        self.ctx = context.make_context()
+        self.scope = scope
+
+    @abc.abstractmethod
+    def get_scoped_model(self, cluster_model):
+        """Leave only nodes and instances proposed in the audit scope"""

watcher/decision_engine/scope/default.py

@@ -0,0 +1,219 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2016 Servionica
+#
+# 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.
+#
+
+import copy
+
+from oslo_log import log
+
+from watcher._i18n import _LW
+from watcher.common import exception
+from watcher.common import nova_helper
+from watcher.decision_engine.scope import base
+
+
+LOG = log.getLogger(__name__)
+
+
+class DefaultScope(base.BaseScope):
+    """Default Audit Scope Handler"""
+
+    DEFAULT_SCHEMA = {
+        "$schema": "http://json-schema.org/draft-04/schema#",
+        "type": "array",
+        "items": {
+            "type": "object",
+            "properties": {
+                "host_aggregates": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "anyOf": [
+                                {"type": ["string", "number"]}
+                            ]
+                        },
+                    }
+                },
+                "availability_zones": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "name": {
+                                "type": "string"
+                            }
+                        },
+                        "additionalProperties": False
+                    }
+                },
+                "exclude": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "instances": {
+                                "type": "array",
+                                "items": {
+                                    "type": "object",
+                                    "properties": {
+                                        "uuid": {
+                                            "type": "string"
+                                        }
+                                    }
+                                }
+                            },
+                            "compute_nodes": {
+                                "type": "array",
+                                "items": {
+                                    "type": "object",
+                                    "properties": {
+                                        "name": {
+                                            "type": "string"
+                                        }
+                                    }
+                                }
+                            }
+                        },
+                        "additionalProperties": False
+                    }
+                }
+            },
+            "additionalProperties": False
+        }
+    }
+
+    def __init__(self, scope, osc=None):
+        super(DefaultScope, self).__init__(scope)
+        self._osc = osc
+        self.wrapper = nova_helper.NovaHelper(osc=self._osc)
+
+    def _remove_instance(self, cluster_model, instance_uuid, node_name):
+        node = cluster_model.get_node_by_uuid(node_name)
+        instance = cluster_model.get_instance_by_uuid(instance_uuid)
+        cluster_model.delete_instance(instance, node)
+
+    def _check_wildcard(self, aggregate_list):
+        if '*' in aggregate_list:
+            if len(aggregate_list) == 1:
+                return True
+            else:
+                raise exception.WildcardCharacterIsUsed(
+                    resource="host aggregates")
+        return False
+
+    def _collect_aggregates(self, host_aggregates, allowed_nodes):
+        aggregate_list = self.wrapper.get_aggregate_list()
+        aggregate_ids = [aggregate['id'] for aggregate
+                         in host_aggregates if 'id' in aggregate]
+        aggregate_names = [aggregate['name'] for aggregate
+                           in host_aggregates if 'name' in aggregate]
+        include_all_nodes = any(self._check_wildcard(field)
+                                for field in (aggregate_ids, aggregate_names))
+
+        for aggregate in aggregate_list:
+            detailed_aggregate = self.wrapper.get_aggregate_detail(
+                aggregate.id)
+            if (detailed_aggregate.id in aggregate_ids or
+                detailed_aggregate.name in aggregate_names or
+                    include_all_nodes):
+                allowed_nodes.extend(detailed_aggregate.hosts)
+
+    def _collect_zones(self, availability_zones, allowed_nodes):
+        zone_list = self.wrapper.get_availability_zone_list()
+        zone_names = [zone['name'] for zone
+                      in availability_zones]
+        include_all_nodes = False
+        if '*' in zone_names:
+            if len(zone_names) == 1:
+                include_all_nodes = True
+            else:
+                raise exception.WildcardCharacterIsUsed(
+                    resource="availability zones")
+        for zone in zone_list:
+            if zone.zoneName in zone_names or include_all_nodes:
+                allowed_nodes.extend(zone.hosts.keys())
+
+    def _exclude_resources(self, resources, **kwargs):
+        instances_to_exclude = kwargs.get('instances')
+        nodes_to_exclude = kwargs.get('nodes')
+        for resource in resources:
+            if 'instances' in resource:
+                instances_to_exclude.extend(
+                    [instance['uuid'] for instance
+                     in resource['instances']])
+            elif 'compute_nodes' in resource:
+                nodes_to_exclude.extend(
+                    [host['name'] for host
+                     in resource['compute_nodes']])
+
+    def _remove_node_from_model(self, nodes_to_remove, cluster_model):
+        for node_name in nodes_to_remove:
+            instances = copy.copy(
+                cluster_model.get_mapping().get_node_instances_by_uuid(
+                    node_name))
+            for instance_uuid in instances:
+                self._remove_instance(cluster_model, instance_uuid, node_name)
+            node = cluster_model.get_node_by_uuid(node_name)
+            cluster_model.remove_node(node)
+
+    def _remove_instances_from_model(self, instances_to_remove, cluster_model):
+        for instance_uuid in instances_to_remove:
+            try:
+                node_name = (cluster_model.get_mapping()
+                             .get_node_by_instance_uuid(instance_uuid).uuid)
+            except KeyError:
+                LOG.warning(_LW("The following instance %s cannot be found. "
+                                "It might be deleted from CDM along with node"
+                                " instance was hosted on."),
+                            instance_uuid)
+                continue
+            self._remove_instance(cluster_model, instance_uuid, node_name)
+
+    def get_scoped_model(self, cluster_model):
+        """Leave only nodes and instances proposed in the audit scope"""
+
+        if not cluster_model:
+            return None
+
+        allowed_nodes = []
+        nodes_to_exclude = []
+        instances_to_exclude = []
+        model_hosts = list(cluster_model.get_all_compute_nodes().keys())
+
+        if not self.scope:
+            return cluster_model
+
+        for rule in self.scope:
+            if 'host_aggregates' in rule:
+                self._collect_aggregates(rule['host_aggregates'],
+                                         allowed_nodes)
+            elif 'availability_zones' in rule:
+                self._collect_zones(rule['availability_zones'],
+                                    allowed_nodes)
+            elif 'exclude' in rule:
+                self._exclude_resources(
+                    rule['exclude'], instances=instances_to_exclude,
+                    nodes=nodes_to_exclude)
+
+        instances_to_remove = set(instances_to_exclude)
+        nodes_to_remove = set(model_hosts) - set(allowed_nodes)
+        nodes_to_remove.update(nodes_to_exclude)
+
+        self._remove_node_from_model(nodes_to_remove, cluster_model)
+        self._remove_instances_from_model(instances_to_remove, cluster_model)
+
+        return cluster_model

watcher/decision_engine/strategy/context/default.py

@@ -53,6 +53,8 @@ class DefaultStrategyContext(base.BaseStrategyContext):
 
         selected_strategy = strategy_selector.select()
 
+        selected_strategy.audit_scope = audit.scope
+
         schema = selected_strategy.get_schema()
         if not audit.parameters and schema:
             # Default value feedback if no predefined strategy

watcher/decision_engine/strategy/strategies/base.py

@@ -40,11 +40,13 @@ import abc
 import six
 
 from watcher.common import clients
+from watcher.common import context
 from watcher.common import exception
 from watcher.common.loader import loadable
 from watcher.common import utils
 from watcher.decision_engine.loading import default as loading
 from watcher.decision_engine.model.collector import manager
+from watcher.decision_engine.scope import default as default_scope
 from watcher.decision_engine.solution import default
 from watcher.decision_engine.strategy.common import level
 
@@ -66,6 +68,7 @@ class BaseStrategy(loadable.Loadable):
         :type osc: :py:class:`~.OpenStackClients` instance
         """
         super(BaseStrategy, self).__init__(config)
+        self.ctx = context.make_context()
         self._name = self.get_name()
         self._display_name = self.get_display_name()
         self._goal = self.get_goal()
@@ -78,6 +81,8 @@ class BaseStrategy(loadable.Loadable):
         self._collector_manager = None
         self._compute_model = None
         self._input_parameters = utils.Struct()
+        self._audit_scope = None
+        self._audit_scope_handler = None
 
     @classmethod
     @abc.abstractmethod
@@ -174,7 +179,8 @@ class BaseStrategy(loadable.Loadable):
         if self._compute_model is None:
             collector = self.collector_manager.get_cluster_model_collector(
                 'compute', osc=self.osc)
-            self._compute_model = collector.get_latest_cluster_data_model()
+            self._compute_model = self.audit_scope_handler.get_scoped_model(
+                collector.get_latest_cluster_data_model())
 
         if not self._compute_model:
             raise exception.ClusterStateNotDefined()
@@ -212,6 +218,21 @@ class BaseStrategy(loadable.Loadable):
     def solution(self, s):
         self._solution = s
 
+    @property
+    def audit_scope(self):
+        return self._audit_scope
+
+    @audit_scope.setter
+    def audit_scope(self, s):
+        self._audit_scope = s
+
+    @property
+    def audit_scope_handler(self):
+        if not self._audit_scope_handler:
+            self._audit_scope_handler = default_scope.DefaultScope(
+                self.audit_scope)
+        return self._audit_scope_handler
+
     @property
     def name(self):
         return self._name