Arnike/watcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

New default planner

Browse Source 
Co-Authored-By: Vincent Francoise <Vincent.FRANCOISE@b-com.com>
Change-Id: Ide2c8fc521488e486eac8f9f89d3f808ccf4b4d7
Implements: blueprint planner-storage-action-plan
This commit is contained in:
Alexander Chadin
2016-12-05 17:32:15 +03:00
parent 7039a9d247
commit 0e440d37ee
30 changed files with 2358 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files

169
watcher/decision_engine/planner/default.py
View File

@@ -1,169 +0,0 @@
# -*- encoding: utf-8 -*-
# Copyright (c) 2015 b<>com
#
# Authors: Jean-Emile DARTOIS <jean-emile.dartois@b-com.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 oslo_config import cfg
from oslo_log import log
from watcher._i18n import _LW
from watcher.common import utils
from watcher.decision_engine.planner import base
from watcher import objects
LOG = log.getLogger(__name__)
class DefaultPlanner(base.BasePlanner):
"""Default planner implementation
This implementation comes with basic rules with a set of action types that
are weighted. An action having a lower weight will be scheduled before the
other ones. The set of action types can be specified by 'weights' in the
``watcher.conf``. You need to associate a different weight to all available
actions into the configuration file, otherwise you will get an error when
the new action will be referenced in the solution produced by a strategy.
"""
weights_dict = {
'nop': 0,
'sleep': 1,
'change_nova_service_state': 2,
'migrate': 3,
}
@classmethod
def get_config_opts(cls):
return [
cfg.DictOpt(
'weights',
help="These weights are used to schedule the actions",
default=cls.weights_dict),
]
def create_action(self,
action_plan_id,
action_type,
input_parameters=None):
uuid = utils.generate_uuid()
action = {
'uuid': uuid,
'action_plan_id': int(action_plan_id),
'action_type': action_type,
'input_parameters': input_parameters,
'state': objects.action.State.PENDING,
'next': None,
}
return action
def schedule(self, context, audit_id, solution):
LOG.debug('Creating an action plan for the audit uuid: %s', audit_id)
priorities = self.config.weights
action_plan = self._create_action_plan(context, audit_id, solution)
actions = list(solution.actions)
to_schedule = []
for action in actions:
json_action = self.create_action(
action_plan_id=action_plan.id,
action_type=action.get('action_type'),
input_parameters=action.get('input_parameters'))
to_schedule.append((priorities[action.get('action_type')],
json_action))
self._create_efficacy_indicators(
context, action_plan.id, solution.efficacy_indicators)
# scheduling
scheduled = sorted(to_schedule, key=lambda x: (x[0]))
if len(scheduled) == 0:
LOG.warning(_LW("The action plan is empty"))
action_plan.first_action_id = None
action_plan.state = objects.action_plan.State.SUCCEEDED
action_plan.save()
else:
# create the first action
parent_action = self._create_action(context,
scheduled[0][1],
None)
# remove first
scheduled.pop(0)
action_plan.first_action_id = parent_action.id
action_plan.save()
for s_action in scheduled:
current_action = self._create_action(context, s_action[1],
parent_action)
parent_action = current_action
return action_plan
def _create_action_plan(self, context, audit_id, solution):
strategy = objects.Strategy.get_by_name(
context, solution.strategy.name)
action_plan_dict = {
'uuid': utils.generate_uuid(),
'audit_id': audit_id,
'strategy_id': strategy.id,
'first_action_id': None,
'state': objects.action_plan.State.RECOMMENDED,
'global_efficacy': solution.global_efficacy,
}
new_action_plan = objects.ActionPlan(context, **action_plan_dict)
new_action_plan.create()
return new_action_plan
def _create_efficacy_indicators(self, context, action_plan_id, indicators):
efficacy_indicators = []
for indicator in indicators:
efficacy_indicator_dict = {
'uuid': utils.generate_uuid(),
'name': indicator.name,
'description': indicator.description,
'unit': indicator.unit,
'value': indicator.value,
'action_plan_id': action_plan_id,
}
new_efficacy_indicator = objects.EfficacyIndicator(
context, **efficacy_indicator_dict)
new_efficacy_indicator.create()
efficacy_indicators.append(new_efficacy_indicator)
return efficacy_indicators
def _create_action(self, context, _action, parent_action):
try:
LOG.debug("Creating the %s in the Watcher database",
_action.get("action_type"))
new_action = objects.Action(context, **_action)
new_action.create()
new_action.save()
if parent_action:
parent_action.next = new_action.id
parent_action.save()
return new_action
except Exception as exc:
LOG.exception(exc)
raise

233
watcher/decision_engine/planner/weight.py Normal file
View File

@@ -0,0 +1,233 @@
# -*- encoding: utf-8 -*-
#
# Authors: Vincent Francoise <Vincent.FRANCOISE@b-com.com>
# Alexander Chadin <a.chadin@servionica.ru>
# 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 collections
import networkx as nx
from oslo_config import cfg
from oslo_config import types
from oslo_log import log
from watcher._i18n import _LW
from watcher.common import utils
from watcher.decision_engine.planner import base
from watcher import objects
LOG = log.getLogger(__name__)
class WeightPlanner(base.BasePlanner):
"""Weight planner implementation
This implementation builds actions with parents in accordance with weights.
Set of actions having a lower weight will be scheduled before
the other ones. There are two config options to configure:
action_weights and parallelization.
*Limitations*
- This planner requires to have action_weights and parallelization configs
tuned well.
"""
def __init__(self, config):
super(WeightPlanner, self).__init__(config)
action_weights = {
'turn_host_to_acpi_s3_state': 10,
'resize': 20,
'migrate': 30,
'sleep': 40,
'change_nova_service_state': 50,
'nop': 60,
}
parallelization = {
'turn_host_to_acpi_s3_state': 2,
'resize': 2,
'migrate': 2,
'sleep': 1,
'change_nova_service_state': 1,
'nop': 1,
}
@classmethod
def get_config_opts(cls):
return [
cfg.Opt(
'weights',
type=types.Dict(value_type=types.Integer()),
help="These weights are used to schedule the actions. "
"Action Plan will be build in accordance with sets of "
"actions ordered by descending weights."
"Two action types cannot have the same weight. ",
default=cls.action_weights),
cfg.Opt(
'parallelization',
type=types.Dict(value_type=types.Integer()),
help="Number of actions to be run in parallel on a per "
"action type basis.",
default=cls.parallelization),
]
@staticmethod
def format_action(action_plan_id, action_type,
input_parameters=None, parents=()):
return {
'uuid': utils.generate_uuid(),
'action_plan_id': int(action_plan_id),
'action_type': action_type,
'input_parameters': input_parameters,
'state': objects.action.State.PENDING,
'parents': parents or None,
}
@staticmethod
def chunkify(lst, n):
"""Yield successive n-sized chunks from lst."""
if n < 1:
# Just to make sure the number is valid
n = 1
# Split a flat list in a list of chunks of size n.
# e.g. chunkify([0, 1, 2, 3, 4], 2) -> [[0, 1], [2, 3], [4]]
for i in range(0, len(lst), n):
yield lst[i:i + n]
def compute_action_graph(self, sorted_weighted_actions):
reverse_weights = {v: k for k, v in self.config.weights.items()}
# leaf_groups contains a list of list of nodes called groups
# each group is a set of nodes from which a future node will
# branch off (parent nodes).
# START --> migrate-1 --> migrate-3
# \ \--> resize-1 --> FINISH
# \--> migrate-2 -------------/
# In the above case migrate-1 will the only memeber of the leaf
# group that migrate-3 will use as parent group, whereas
# resize-1 will have both migrate-2 and migrate-3 in its
# parent/leaf group
leaf_groups = []
action_graph = nx.DiGraph()
# We iterate through each action type category (sorted by weight) to
# insert them in a Directed Acyclic Graph
for idx, (weight, actions) in enumerate(sorted_weighted_actions):
action_chunks = self.chunkify(
actions, self.config.parallelization[reverse_weights[weight]])
# We split the actions into chunks/layers that will have to be
# spread across all the available branches of the graph
for chunk_idx, actions_chunk in enumerate(action_chunks):
for action in actions_chunk:
action_graph.add_node(action)
# all other actions
parent_nodes = []
if not idx and not chunk_idx:
parent_nodes = []
elif leaf_groups:
parent_nodes = leaf_groups
for parent_node in parent_nodes:
action_graph.add_edge(parent_node, action)
action.parents.append(parent_node.uuid)
if leaf_groups:
leaf_groups = []
leaf_groups.extend([a for a in actions_chunk])
return action_graph
def schedule(self, context, audit_id, solution):
LOG.debug('Creating an action plan for the audit uuid: %s', audit_id)
action_plan = self.create_action_plan(context, audit_id, solution)
sorted_weighted_actions = self.get_sorted_actions_by_weight(
context, action_plan, solution)
action_graph = self.compute_action_graph(sorted_weighted_actions)
self._create_efficacy_indicators(
context, action_plan.id, solution.efficacy_indicators)
if len(action_graph.nodes()) == 0:
LOG.warning(_LW("The action plan is empty"))
action_plan.state = objects.action_plan.State.SUCCEEDED
action_plan.save()
self.create_scheduled_actions(action_plan, action_graph)
return action_plan
def get_sorted_actions_by_weight(self, context, action_plan, solution):
# We need to make them immutable to add them to the graph
action_objects = list([
objects.Action(
context, uuid=utils.generate_uuid(), parents=[],
action_plan_id=action_plan.id, **a)
for a in solution.actions])
# This is a dict of list with each being a weight and the list being
# all the actions associated to this weight
weighted_actions = collections.defaultdict(list)
for action in action_objects:
action_weight = self.config.weights[action.action_type]
weighted_actions[action_weight].append(action)
return reversed(sorted(weighted_actions.items(), key=lambda x: x[0]))
def create_scheduled_actions(self, action_plan, graph):
for action in graph.nodes():
LOG.debug("Creating the %s in the Watcher database",
action.action_type)
try:
action.create()
except Exception as exc:
LOG.exception(exc)
raise
def create_action_plan(self, context, audit_id, solution):
strategy = objects.Strategy.get_by_name(
context, solution.strategy.name)
action_plan_dict = {
'uuid': utils.generate_uuid(),
'audit_id': audit_id,
'strategy_id': strategy.id,
'state': objects.action_plan.State.RECOMMENDED,
'global_efficacy': solution.global_efficacy,
}
new_action_plan = objects.ActionPlan(context, **action_plan_dict)
new_action_plan.create()
return new_action_plan
def _create_efficacy_indicators(self, context, action_plan_id, indicators):
efficacy_indicators = []
for indicator in indicators:
efficacy_indicator_dict = {
'uuid': utils.generate_uuid(),
'name': indicator.name,
'description': indicator.description,
'unit': indicator.unit,
'value': indicator.value,
'action_plan_id': action_plan_id,
}
new_efficacy_indicator = objects.EfficacyIndicator(
context, **efficacy_indicator_dict)
new_efficacy_indicator.create()
efficacy_indicators.append(new_efficacy_indicator)
return efficacy_indicators

301
watcher/decision_engine/planner/workload_stabilization.py Normal file
View File

@@ -0,0 +1,301 @@
# -*- encoding: utf-8 -*-
#
# 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
from oslo_config import cfg
from oslo_config import types
from oslo_log import log
from watcher._i18n import _LW
from watcher.common import clients
from watcher.common import exception
from watcher.common import nova_helper
from watcher.common import utils
from watcher.decision_engine.planner import base
from watcher import objects
LOG = log.getLogger(__name__)
class WorkloadStabilizationPlanner(base.BasePlanner):
"""Workload Stabilization planner implementation
This implementation comes with basic rules with a set of action types that
are weighted. An action having a lower weight will be scheduled before the
other ones. The set of action types can be specified by 'weights' in the
``watcher.conf``. You need to associate a different weight to all available
actions into the configuration file, otherwise you will get an error when
the new action will be referenced in the solution produced by a strategy.
*Limitations*
- This is a proof of concept that is not meant to be used in production
"""
def __init__(self, config):
super(WorkloadStabilizationPlanner, self).__init__(config)
self._osc = clients.OpenStackClients()
@property
def osc(self):
return self._osc
weights_dict = {
'turn_host_to_acpi_s3_state': 0,
'resize': 1,
'migrate': 2,
'sleep': 3,
'change_nova_service_state': 4,
'nop': 5,
}
@classmethod
def get_config_opts(cls):
return [
cfg.Opt(
'weights',
type=types.Dict(value_type=types.Integer()),
help="These weights are used to schedule the actions",
default=cls.weights_dict),
]
def create_action(self,
action_plan_id,
action_type,
input_parameters=None):
uuid = utils.generate_uuid()
action = {
'uuid': uuid,
'action_plan_id': int(action_plan_id),
'action_type': action_type,
'input_parameters': input_parameters,
'state': objects.action.State.PENDING,
'parents': None
}
return action
def load_child_class(self, child_name):
for c in BaseActionValidator.__subclasses__():
if child_name == c.action_name:
return c()
return None
def schedule(self, context, audit_id, solution):
LOG.debug('Creating an action plan for the audit uuid: %s', audit_id)
weights = self.config.weights
action_plan = self._create_action_plan(context, audit_id, solution)
actions = list(solution.actions)
to_schedule = []
for action in actions:
json_action = self.create_action(
action_plan_id=action_plan.id,
action_type=action.get('action_type'),
input_parameters=action.get('input_parameters'))
to_schedule.append((weights[action.get('action_type')],
json_action))
self._create_efficacy_indicators(
context, action_plan.id, solution.efficacy_indicators)
# scheduling
scheduled = sorted(to_schedule, key=lambda weight: (weight[0]),
reverse=True)
if len(scheduled) == 0:
LOG.warning(_LW("The action plan is empty"))
action_plan.state = objects.action_plan.State.SUCCEEDED
action_plan.save()
else:
resource_action_map = {}
scheduled_actions = [x[1] for x in scheduled]
for action in scheduled_actions:
a_type = action['action_type']
if a_type != 'turn_host_to_acpi_s3_state':
plugin_action = self.load_child_class(
action.get("action_type"))
if not plugin_action:
raise exception.UnsupportedActionType(
action_type=action.get("action_type"))
db_action = self._create_action(context, action)
parents = plugin_action.validate_parents(
resource_action_map, action)
if parents:
db_action.parents = parents
db_action.save()
# if we have an action that will make host unreachable, we need
# to complete all actions (resize and migration type)
# related to the host.
# Note(alexchadin): turn_host_to_acpi_s3_state doesn't
# actually exist. Placed code shows relations between
# action types.
# TODO(alexchadin): add turn_host_to_acpi_s3_state action type.
else:
host_to_acpi_s3 = action['input_parameters']['resource_id']
host_actions = resource_action_map.get(host_to_acpi_s3)
action_parents = []
if host_actions:
resize_actions = [x[0] for x in host_actions
if x[1] == 'resize']
migrate_actions = [x[0] for x in host_actions
if x[1] == 'migrate']
resize_migration_parents = [
x.parents for x in
[objects.Action.get_by_uuid(context, resize_action)
for resize_action in resize_actions]]
# resize_migration_parents should be one level list
resize_migration_parents = [
parent for sublist in resize_migration_parents
for parent in sublist]
action_parents.extend([uuid for uuid in
resize_actions])
action_parents.extend([uuid for uuid in
migrate_actions if uuid not in
resize_migration_parents])
db_action = self._create_action(context, action)
db_action.parents = action_parents
db_action.save()
return action_plan
def _create_action_plan(self, context, audit_id, solution):
strategy = objects.Strategy.get_by_name(
context, solution.strategy.name)
action_plan_dict = {
'uuid': utils.generate_uuid(),
'audit_id': audit_id,
'strategy_id': strategy.id,
'state': objects.action_plan.State.RECOMMENDED,
'global_efficacy': solution.global_efficacy,
}
new_action_plan = objects.ActionPlan(context, **action_plan_dict)
new_action_plan.create()
return new_action_plan
def _create_efficacy_indicators(self, context, action_plan_id, indicators):
efficacy_indicators = []
for indicator in indicators:
efficacy_indicator_dict = {
'uuid': utils.generate_uuid(),
'name': indicator.name,
'description': indicator.description,
'unit': indicator.unit,
'value': indicator.value,
'action_plan_id': action_plan_id,
}
new_efficacy_indicator = objects.EfficacyIndicator(
context, **efficacy_indicator_dict)
new_efficacy_indicator.create()
efficacy_indicators.append(new_efficacy_indicator)
return efficacy_indicators
def _create_action(self, context, _action):
try:
LOG.debug("Creating the %s in the Watcher database",
_action.get("action_type"))
new_action = objects.Action(context, **_action)
new_action.create()
return new_action
except Exception as exc:
LOG.exception(exc)
raise
class BaseActionValidator(object):
action_name = None
def __init__(self):
super(BaseActionValidator, self).__init__()
self._osc = None
@property
def osc(self):
if not self._osc:
self._osc = clients.OpenStackClients()
return self._osc
@abc.abstractmethod
def validate_parents(self, resource_action_map, action):
raise NotImplementedError()
def _mapping(self, resource_action_map, resource_id, action_uuid,
action_type):
if resource_id not in resource_action_map:
resource_action_map[resource_id] = [(action_uuid,
action_type,)]
else:
resource_action_map[resource_id].append((action_uuid,
action_type,))
class MigrationActionValidator(BaseActionValidator):
action_name = "migrate"
def validate_parents(self, resource_action_map, action):
instance_uuid = action['input_parameters']['resource_id']
host_name = action['input_parameters']['source_node']
self._mapping(resource_action_map, instance_uuid, action['uuid'],
'migrate')
self._mapping(resource_action_map, host_name, action['uuid'],
'migrate')
class ResizeActionValidator(BaseActionValidator):
action_name = "resize"
def validate_parents(self, resource_action_map, action):
nova = nova_helper.NovaHelper(osc=self.osc)
instance_uuid = action['input_parameters']['resource_id']
parent_actions = resource_action_map.get(instance_uuid)
host_of_instance = nova.get_hostname(
nova.get_instance_by_uuid(instance_uuid)[0])
self._mapping(resource_action_map, host_of_instance, action['uuid'],
'resize')
if parent_actions:
return [x[0] for x in parent_actions]
else:
return []
class ChangeNovaServiceStateActionValidator(BaseActionValidator):
action_name = "change_nova_service_state"
def validate_parents(self, resource_action_map, action):
host_name = action['input_parameters']['resource_id']
self._mapping(resource_action_map, host_name, action.uuid,
'change_nova_service_state')
return []
class SleepActionValidator(BaseActionValidator):
action_name = "sleep"
def validate_parents(self, resource_action_map, action):
return []
class NOPActionValidator(BaseActionValidator):
action_name = "nop"
def validate_parents(self, resource_action_map, action):
return []