Added Actuator Strategy

This strategy now allow us to create action plans with an explicit
set of actions.

Co-Authored-By: Mikhail Kizilov <kizilov.mikhail@gmail.com>
Change-Id: I7b04b9936ce5f3b5b38f319da7f8737e0f3eea88
Closes-Bug: #1659243

doc/source/contributor/plugin/action-plugin.rst

@@ -217,3 +217,11 @@ which is only able to process the Watcher built-in actions. Therefore, you will
 either have to use an existing third-party planner or :ref:`implement another
 planner <implement_planner_plugin>` that will be able to take into account your
 new action plugin.
+
+
+Test your new action
+====================
+
+In order to test your new action via a manual test or a Tempest test, you can
+use the  :py:class:`~.Actuator` strategy and pass it one or more actions to
+execute. This way, you can isolate your action to see if it works as expected.

setup.cfg

@@ -66,6 +66,7 @@ watcher_strategies =
     dummy = watcher.decision_engine.strategy.strategies.dummy_strategy:DummyStrategy
     dummy_with_scorer = watcher.decision_engine.strategy.strategies.dummy_with_scorer:DummyWithScorer
     dummy_with_resize = watcher.decision_engine.strategy.strategies.dummy_with_resize:DummyWithResize
+    actuator = watcher.decision_engine.strategy.strategies.actuation:Actuator
     basic = watcher.decision_engine.strategy.strategies.basic_consolidation:BasicConsolidation
     outlet_temperature = watcher.decision_engine.strategy.strategies.outlet_temp_control:OutletTempControl
     saving_energy =  watcher.decision_engine.strategy.strategies.saving_energy:SavingEnergy

watcher/decision_engine/strategy/strategies/__init__.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from watcher.decision_engine.strategy.strategies import actuation
 from watcher.decision_engine.strategy.strategies import basic_consolidation
 from watcher.decision_engine.strategy.strategies import dummy_strategy
 from watcher.decision_engine.strategy.strategies import dummy_with_scorer
@@ -26,6 +27,7 @@ from watcher.decision_engine.strategy.strategies import \
 from watcher.decision_engine.strategy.strategies import workload_balance
 from watcher.decision_engine.strategy.strategies import workload_stabilization
 
+Actuator = actuation.Actuator
 BasicConsolidation = basic_consolidation.BasicConsolidation
 OutletTempControl = outlet_temp_control.OutletTempControl
 DummyStrategy = dummy_strategy.DummyStrategy
@@ -37,7 +39,7 @@ WorkloadStabilization = workload_stabilization.WorkloadStabilization
 UniformAirflow = uniform_airflow.UniformAirflow
 NoisyNeighbor = noisy_neighbor.NoisyNeighbor
 
-__all__ = ("BasicConsolidation", "OutletTempControl", "DummyStrategy",
-           "DummyWithScorer", "VMWorkloadConsolidation", "WorkloadBalance",
-           "WorkloadStabilization", "UniformAirflow", "NoisyNeighbor",
-           "SavingEnergy")
+__all__ = ("Actuator", "BasicConsolidation", "OutletTempControl",
+           "DummyStrategy", "DummyWithScorer", "VMWorkloadConsolidation",
+           "WorkloadBalance", "WorkloadStabilization", "UniformAirflow",
+           "NoisyNeighbor", "SavingEnergy")

watcher/decision_engine/strategy/strategies/actuation.py

@@ -0,0 +1,99 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2017 b<>com
+#
+# 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.
+#
+"""
+*Actuator*
+
+This strategy allows anyone to create an action plan with a predefined set of
+actions. This strategy can be used for 2 different purposes:
+
+- Test actions
+- Use this strategy based on an event trigger to perform some explicit task
+
+"""
+
+from oslo_log import log
+
+from watcher._i18n import _
+from watcher.decision_engine.strategy.strategies import base
+
+LOG = log.getLogger(__name__)
+
+
+class Actuator(base.UnclassifiedStrategy):
+    """Actuator that simply executes the actions given as parameter"""
+
+    @classmethod
+    def get_name(cls):
+        return "actuator"
+
+    @classmethod
+    def get_display_name(cls):
+        return _("Actuator")
+
+    @classmethod
+    def get_translatable_display_name(cls):
+        return "Actuator"
+
+    @classmethod
+    def get_schema(cls):
+        # Mandatory default setting for each element
+        return {
+            "$schema": "http://json-schema.org/draft-04/schema#",
+            "type": "object",
+            "properties": {
+                "actions": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "action_type": {
+                                "type": "string"
+                            },
+                            "resource_id": {
+                                "type": "string"
+                            },
+                            "input_parameters": {
+                                "type": "object",
+                                "properties": {},
+                                "additionalProperties": True
+                            }
+                        },
+                        "required": [
+                            "action_type", "input_parameters"
+                        ],
+                        "additionalProperties": True,
+                    }
+                }
+            },
+            "required": [
+                "actions"
+            ]
+        }
+
+    @property
+    def actions(self):
+        return self.input_parameters.get('actions', [])
+
+    def pre_execute(self):
+        LOG.info("Preparing Actuator strategy...")
+
+    def do_execute(self):
+        for action in self.actions:
+            self.solution.add_action(**action)
+
+    def post_execute(self):
+        pass

watcher/tests/decision_engine/strategy/strategies/test_actuator.py

@@ -0,0 +1,37 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2017 b<>com
+#
+# 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 mock
+
+from watcher.common import utils
+from watcher.decision_engine.strategy import strategies
+from watcher.tests import base
+
+
+class TestActuator(base.TestCase):
+
+    def setUp(self):
+        super(TestActuator, self).setUp()
+        # fake cluster
+        self.strategy = strategies.Actuator(config=mock.Mock())
+
+    def test_actuator_strategy(self):
+        fake_action = {"action_type": "TEST", "input_parameters": {"a": "b"}}
+        self.strategy.input_parameters = utils.Struct(
+            {"actions": [fake_action]})
+        solution = self.strategy.execute()
+        self.assertEqual(1, len(solution.actions))
+        self.assertEqual([fake_action], solution.actions)

watcher_tempest_plugin/tests/api/admin/base.py

@@ -165,18 +165,19 @@ class BaseInfraOptimTest(test.BaseTestCase):
 
     @classmethod
     def create_audit(cls, audit_template_uuid, audit_type='ONESHOT',
-                     state=None, interval=None):
+                     state=None, interval=None, parameters=None):
         """Wrapper utility for creating a test audit
 
         :param audit_template_uuid: Audit Template UUID this audit will use
         :param audit_type: Audit type (either ONESHOT or CONTINUOUS)
         :param state: Audit state (str)
         :param interval: Audit interval in seconds or cron syntax (str)
+        :param parameters: list of execution parameters
         :return: A tuple with The HTTP response and its body
         """
         resp, body = cls.client.create_audit(
             audit_template_uuid=audit_template_uuid, audit_type=audit_type,
-            state=state, interval=interval)
+            state=state, interval=interval, parameters=parameters)
 
         cls.created_audits.add(body['uuid'])
         cls.created_action_plans_audit_uuids.add(body['uuid'])
@@ -251,11 +252,6 @@ class BaseInfraOptimTest(test.BaseTestCase):
 
         return resp
 
-    @classmethod
-    def has_action_plan_finished(cls, action_plan_uuid):
-        _, action_plan = cls.client.show_action_plan(action_plan_uuid)
-        return action_plan.get('state') in cls.FINISHED_STATES
-
     @classmethod
     def is_action_plan_idle(cls, action_plan_uuid):
         """This guard makes sure your action plan is not running"""

watcher_tempest_plugin/tests/api/admin/test_action_plan.py

@@ -70,42 +70,6 @@ class TestCreateDeleteExecuteActionPlan(base.BaseInfraOptimTest):
         self.assertRaises(exceptions.NotFound, self.client.show_action_plan,
                           action_plan['uuid'])
 
-    @decorators.attr(type='smoke')
-    def test_execute_dummy_action_plan(self):
-        _, goal = self.client.show_goal("dummy")
-        _, audit_template = self.create_audit_template(goal['uuid'])
-        _, audit = self.create_audit(audit_template['uuid'])
-
-        self.assertTrue(test_utils.call_until_true(
-            func=functools.partial(self.has_audit_finished, audit['uuid']),
-            duration=30,
-            sleep_for=.5
-        ))
-        _, action_plans = self.client.list_action_plans(
-            audit_uuid=audit['uuid'])
-        action_plan = action_plans['action_plans'][0]
-
-        _, action_plan = self.client.show_action_plan(action_plan['uuid'])
-
-        if action_plan['state'] in ['SUPERSEDED', 'SUCCEEDED']:
-            # This means the action plan is superseded so we cannot trigger it,
-            # or it is empty.
-            return
-
-        # Execute the action by changing its state to PENDING
-        _, updated_ap = self.client.start_action_plan(action_plan['uuid'])
-
-        self.assertTrue(test_utils.call_until_true(
-            func=functools.partial(
-                self.has_action_plan_finished, action_plan['uuid']),
-            duration=30,
-            sleep_for=.5
-        ))
-        _, finished_ap = self.client.show_action_plan(action_plan['uuid'])
-
-        self.assertIn(updated_ap['state'], ('PENDING', 'ONGOING'))
-        self.assertIn(finished_ap['state'], ('SUCCEEDED', 'SUPERSEDED'))
-
 
 class TestShowListActionPlan(base.BaseInfraOptimTest):
     """Tests for action_plan."""

watcher_tempest_plugin/tests/scenario/base.py

@@ -24,6 +24,7 @@ from oslo_log import log
 from tempest import config
 from tempest import exceptions
 from tempest.lib.common.utils import data_utils
+from tempest.lib.common.utils import test_utils
 
 from watcher_tempest_plugin import infra_optim_clients as clients
 from watcher_tempest_plugin.tests.scenario import manager
@@ -75,6 +76,19 @@ class BaseInfraOptimScenarioTest(manager.ScenarioTest):
             LOG.error(msg)
             raise exceptions.InvalidConfiguration(msg)
 
+    @classmethod
+    def _are_all_action_plans_finished(cls):
+        _, action_plans = cls.client.list_action_plans()
+        return all([ap['state'] in cls.FINISHED_STATES
+                    for ap in action_plans['action_plans']])
+
+    def wait_for_all_action_plans_to_finish(self):
+        assert test_utils.call_until_true(
+            func=self._are_all_action_plans_finished,
+            duration=300,
+            sleep_for=5
+        )
+
     # ### AUDIT TEMPLATES ### #
 
     def create_audit_template(self, goal, name=None, description=None,
@@ -111,18 +125,19 @@ class BaseInfraOptimScenarioTest(manager.ScenarioTest):
     # ### AUDITS ### #
 
     def create_audit(self, audit_template_uuid, audit_type='ONESHOT',
-                     state=None, parameters=None):
+                     state=None, interval=None, parameters=None):
         """Wrapper utility for creating a test audit
 
         :param audit_template_uuid: Audit Template UUID this audit will use
-        :param audit_type: Audit type (either ONESHOT or CONTINUOUS)
-        :param state: Audit state
-        :param parameters: Input parameters of the audit
+        :param type: Audit type (either ONESHOT or CONTINUOUS)
+        :param state: Audit state (str)
+        :param interval: Audit interval in seconds (int)
+        :param parameters: list of execution parameters
         :return: A tuple with The HTTP response and its body
         """
         resp, body = self.client.create_audit(
             audit_template_uuid=audit_template_uuid, audit_type=audit_type,
-            state=state, parameters=parameters)
+            state=state, interval=interval, parameters=parameters)
 
         self.addCleanup(self.delete_audit, audit_uuid=body["uuid"])
         return resp, body

watcher_tempest_plugin/tests/scenario/test_execute_actuator.py

@@ -0,0 +1,340 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2016 b<>com
+#
+#
+# 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 __future__ import unicode_literals
+
+import collections
+import functools
+
+from tempest import config
+from tempest.lib.common.utils import test_utils
+
+from watcher_tempest_plugin.tests.scenario import base
+
+CONF = config.CONF
+
+
+class TestExecuteActionsViaActuator(base.BaseInfraOptimScenarioTest):
+
+    scenarios = [
+        ("nop", {"actions": [
+            {"action_type": "nop",
+             "input_parameters": {
+                 "message": "hello World"}}]}),
+        ("sleep", {"actions": [
+            {"action_type": "sleep",
+             "input_parameters": {
+                 "duration": 1.0}}]}),
+        ("change_nova_service_state", {"actions": [
+            {"action_type": "change_nova_service_state",
+             "input_parameters": {
+                 "state": "enabled"},
+             "filling_function":
+                 "_prerequisite_param_for_"
+                 "change_nova_service_state_action"}]}),
+        ("resize", {"actions": [
+            {"action_type": "resize",
+             "filling_function": "_prerequisite_param_for_resize_action"}]}),
+        ("migrate", {"actions": [
+            {"action_type": "migrate",
+             "input_parameters": {
+                 "migration_type": "live"},
+             "filling_function": "_prerequisite_param_for_migrate_action"},
+            {"action_type": "migrate",
+             "filling_function": "_prerequisite_param_for_migrate_action"}]})
+    ]
+
+    @classmethod
+    def resource_setup(cls):
+        super(TestExecuteActionsViaActuator, cls).resource_setup()
+        if CONF.compute.min_compute_nodes < 2:
+            raise cls.skipException(
+                "Less than 2 compute nodes, skipping multinode tests.")
+        if not CONF.compute_feature_enabled.live_migration:
+            raise cls.skipException("Live migration is not enabled")
+
+        cls.initial_compute_nodes_setup = cls.get_compute_nodes_setup()
+        enabled_compute_nodes = [cn for cn in cls.initial_compute_nodes_setup
+                                 if cn.get('status') == 'enabled']
+
+        cls.wait_for_compute_node_setup()
+
+        if len(enabled_compute_nodes) < 2:
+            raise cls.skipException(
+                "Less than 2 compute nodes are enabled, "
+                "skipping multinode tests.")
+
+    @classmethod
+    def get_compute_nodes_setup(cls):
+        services_client = cls.mgr.services_client
+        available_services = services_client.list_services()['services']
+
+        return [srv for srv in available_services
+                if srv.get('binary') == 'nova-compute']
+
+    @classmethod
+    def wait_for_compute_node_setup(cls):
+
+        def _are_compute_nodes_setup():
+            try:
+                hypervisors_client = cls.mgr.hypervisor_client
+                hypervisors = hypervisors_client.list_hypervisors(
+                    detail=True)['hypervisors']
+                available_hypervisors = set(
+                    hyp['hypervisor_hostname'] for hyp in hypervisors)
+                available_services = set(
+                    service['host']
+                    for service in cls.get_compute_nodes_setup())
+
+                return (
+                    available_hypervisors == available_services and
+                    len(hypervisors) >= 2)
+            except Exception:
+                return False
+
+        assert test_utils.call_until_true(
+            func=_are_compute_nodes_setup,
+            duration=600,
+            sleep_for=2
+        )
+
+    @classmethod
+    def rollback_compute_nodes_status(cls):
+        current_compute_nodes_setup = cls.get_compute_nodes_setup()
+        for cn_setup in current_compute_nodes_setup:
+            cn_hostname = cn_setup.get('host')
+            matching_cns = [
+                cns for cns in cls.initial_compute_nodes_setup
+                if cns.get('host') == cn_hostname
+            ]
+            initial_cn_setup = matching_cns[0]  # Should return a single result
+            if cn_setup.get('status') != initial_cn_setup.get('status'):
+                if initial_cn_setup.get('status') == 'enabled':
+                    rollback_func = cls.mgr.services_client.enable_service
+                else:
+                    rollback_func = cls.mgr.services_client.disable_service
+                rollback_func(binary='nova-compute', host=cn_hostname)
+
+    def _create_one_instance_per_host(self):
+        """Create 1 instance per compute node
+
+        This goes up to the min_compute_nodes threshold so that things don't
+        get crazy if you have 1000 compute nodes but set min to 3.
+        """
+        host_client = self.mgr.hosts_client
+        all_hosts = host_client.list_hosts()['hosts']
+        compute_nodes = [x for x in all_hosts if x['service'] == 'compute']
+
+        created_servers = []
+        for _ in compute_nodes[:CONF.compute.min_compute_nodes]:
+            # by getting to active state here, this means this has
+            # landed on the host in question.
+            created_servers.append(
+                self.create_server(image_id=CONF.compute.image_ref,
+                                   wait_until='ACTIVE',
+                                   clients=self.mgr))
+
+        return created_servers
+
+    def _get_flavors(self):
+        return self.mgr.flavors_client.list_flavors()['flavors']
+
+    def _prerequisite_param_for_migrate_action(self):
+        created_instances = self._create_one_instance_per_host()
+        instance = created_instances[0]
+        source_node = created_instances[0]["OS-EXT-SRV-ATTR:host"]
+        destination_node = created_instances[-1]["OS-EXT-SRV-ATTR:host"]
+
+        parameters = {
+            "resource_id": instance['id'],
+            "migration_type": "live",
+            "source_node": source_node,
+            "destination_node": destination_node
+        }
+
+        return parameters
+
+    def _prerequisite_param_for_resize_action(self):
+        created_instances = self._create_one_instance_per_host()
+        instance = created_instances[0]
+        current_flavor_id = instance['flavor']['id']
+
+        flavors = self._get_flavors()
+        new_flavors = [f for f in flavors if f['id'] != current_flavor_id]
+        new_flavor = new_flavors[0]
+
+        parameters = {
+            "resource_id": instance['id'],
+            "flavor": new_flavor['name']
+        }
+
+        return parameters
+
+    def _prerequisite_param_for_change_nova_service_state_action(self):
+        enabled_compute_nodes = [cn for cn in
+                                 self.initial_compute_nodes_setup
+                                 if cn.get('status') == 'enabled']
+        enabled_compute_node = enabled_compute_nodes[0]
+
+        parameters = {
+            "resource_id": enabled_compute_node['host'],
+            "state": "enabled"
+        }
+
+        return parameters
+
+    def _fill_actions(self, actions):
+        for action in actions:
+            filling_function_name = action.pop('filling_function', None)
+
+            if filling_function_name is not None:
+                filling_function = getattr(self, filling_function_name, None)
+
+                if filling_function is not None:
+                    parameters = filling_function()
+
+                    resource_id = parameters.pop('resource_id', None)
+
+                    if resource_id is not None:
+                        action['resource_id'] = resource_id
+
+                    input_parameters = action.get('input_parameters', None)
+
+                    if input_parameters is not None:
+                        parameters.update(input_parameters)
+                        input_parameters.update(parameters)
+                    else:
+                        action['input_parameters'] = parameters
+
+    def _execute_actions(self, actions):
+        self.wait_for_all_action_plans_to_finish()
+
+        _, goal = self.client.show_goal("unclassified")
+        _, strategy = self.client.show_strategy("actuator")
+        _, audit_template = self.create_audit_template(
+            goal['uuid'], strategy=strategy['uuid'])
+        _, audit = self.create_audit(
+            audit_template['uuid'], parameters={"actions": actions})
+
+        self.assertTrue(test_utils.call_until_true(
+            func=functools.partial(self.has_audit_succeeded, audit['uuid']),
+            duration=30,
+            sleep_for=.5
+        ))
+        _, action_plans = self.client.list_action_plans(
+            audit_uuid=audit['uuid'])
+        action_plan = action_plans['action_plans'][0]
+
+        _, action_plan = self.client.show_action_plan(action_plan['uuid'])
+
+        # Execute the action plan
+        _, updated_ap = self.client.start_action_plan(action_plan['uuid'])
+
+        self.assertTrue(test_utils.call_until_true(
+            func=functools.partial(
+                self.has_action_plan_finished, action_plan['uuid']),
+            duration=300,
+            sleep_for=1
+        ))
+        _, finished_ap = self.client.show_action_plan(action_plan['uuid'])
+        _, action_list = self.client.list_actions(
+            action_plan_uuid=finished_ap["uuid"])
+
+        self.assertIn(updated_ap['state'], ('PENDING', 'ONGOING'))
+        self.assertIn(finished_ap['state'], ('SUCCEEDED', 'SUPERSEDED'))
+
+        expected_action_counter = collections.Counter(
+            act['action_type'] for act in actions)
+        action_counter = collections.Counter(
+            act['action_type'] for act in action_list['actions'])
+
+        self.assertEqual(expected_action_counter, action_counter)
+
+    def test_execute_nop(self):
+        self.addCleanup(self.rollback_compute_nodes_status)
+
+        actions = [{
+            "action_type": "nop",
+            "input_parameters": {"message": "hello World"}}]
+        self._execute_actions(actions)
+
+    def test_execute_sleep(self):
+        self.addCleanup(self.rollback_compute_nodes_status)
+
+        actions = [
+            {"action_type": "sleep",
+             "input_parameters": {"duration": 1.0}}
+        ]
+        self._execute_actions(actions)
+
+    def test_execute_change_nova_service_state(self):
+        self.addCleanup(self.rollback_compute_nodes_status)
+
+        enabled_compute_nodes = [
+            cn for cn in self.initial_compute_nodes_setup
+            if cn.get('status') == 'enabled']
+
+        enabled_compute_node = enabled_compute_nodes[0]
+        actions = [
+            {"action_type": "change_nova_service_state",
+             "resource_id": enabled_compute_node['host'],
+             "input_parameters": {"state": "enabled"}}
+        ]
+        self._execute_actions(actions)
+
+    def test_execute_resize(self):
+        self.addCleanup(self.rollback_compute_nodes_status)
+
+        created_instances = self._create_one_instance_per_host()
+        instance = created_instances[0]
+        current_flavor_id = instance['flavor']['id']
+
+        flavors = self._get_flavors()
+        new_flavors = [f for f in flavors if f['id'] != current_flavor_id]
+        new_flavor = new_flavors[0]
+
+        actions = [
+            {"action_type": "resize",
+             "resource_id": instance['id'],
+             "input_parameters": {"flavor": new_flavor['name']}}
+        ]
+        self._execute_actions(actions)
+
+    def test_execute_migrate(self):
+        self.addCleanup(self.rollback_compute_nodes_status)
+
+        created_instances = self._create_one_instance_per_host()
+        instance = created_instances[0]
+        source_node = created_instances[0]["OS-EXT-SRV-ATTR:host"]
+        destination_node = created_instances[-1]["OS-EXT-SRV-ATTR:host"]
+        actions = [
+            {"action_type": "migrate",
+             "resource_id": instance['id'],
+             "input_parameters": {
+                 "migration_type": "live",
+                 "source_node": source_node,
+                 "destination_node": destination_node}}
+        ]
+        self._execute_actions(actions)
+
+    def test_execute_scenarios(self):
+        self.addCleanup(self.rollback_compute_nodes_status)
+
+        for _, scenario in self.scenarios:
+            actions = scenario['actions']
+            self._fill_actions(actions)
+            self._execute_actions(actions)