openstack
/
watcher
Code Issues Proposed changes

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
This commit is contained in:
Vincent Françoise 2017-01-25 11:21:23 +01:00 committed by Mikhail Kizilov
parent 529b0d34ee
commit 0b31828a01
9 changed files with 514 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
doc/source/contributor/plugin/action-plugin.rst
View File

@ -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.

1
setup.cfg
View File

@ -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

10
watcher/decision_engine/strategy/strategies/__init__.py
View File

@ -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")

99
watcher/decision_engine/strategy/strategies/actuation.py Normal file
View File

@ -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

37
watcher/tests/decision_engine/strategy/strategies/test_actuator.py Normal file
View File

@ -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)

10
watcher_tempest_plugin/tests/api/admin/base.py
View File

@ -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"""

36
watcher_tempest_plugin/tests/api/admin/test_action_plan.py
View File

@ -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."""

25
watcher_tempest_plugin/tests/scenario/base.py
View File

@ -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

340
watcher_tempest_plugin/tests/scenario/test_execute_actuator.py Normal file
View File

@ -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)