openstack/watcher
Code Issues Proposed changes

Code refactoring - StrategyContext and Auditendpoint

Browse Source 
This patchset aim to remove useless code in StrategyContext
and AuditEndPoint.
This patchset also add a parameter for strategy context to define the
numbers of thread of execute the strategies.

DocImpact
Change-Id: I83e87165b03b42fe6b863921502a300bd94d2982
This commit is contained in:
Jean-Emile DARTOIS
2015-12-17 14:21:18 +01:00
parent 1698eb31f3
commit 4c3073efb4
11 changed files with 116 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
etc/watcher/watcher.conf.sample
View File

@@ -788,6 +788,10 @@
# value)
#publisher_id = watcher.decision.api
# The maximum number of threads that can be used to execute strategies
# (integer value)
#max_workers = 2
[watcher_goals]

41
watcher/decision_engine/audit/default.py
View File

@@ -20,7 +20,8 @@ from watcher.decision_engine.audit.base import \
BaseAuditHandler
from watcher.decision_engine.messaging.events import Events
from watcher.decision_engine.planner.default import DefaultPlanner
from watcher.decision_engine.strategy.context.default import StrategyContext
from watcher.decision_engine.strategy.context.default import \
DefaultStrategyContext
from watcher.objects.audit import Audit
from watcher.objects.audit import AuditStatus
from watcher.objects.audit_template import AuditTemplate
@@ -29,11 +30,23 @@ LOG = log.getLogger(__name__)
class DefaultAuditHandler(BaseAuditHandler):
def __init__(self, messaging, model_collector):
def __init__(self, messaging, cluster_collector):
super(DefaultAuditHandler, self).__init__()
self.messaging = messaging
self.model_collector = model_collector
self.strategy_context = StrategyContext()
self._messaging = messaging
self._cluster_collector = cluster_collector
self._strategy_context = DefaultStrategyContext()
@property
def messaging(self):
return self._messaging
@property
def cluster_collector(self):
return self._cluster_collector
@property
def strategy_context(self):
return self._strategy_context
def notify(self, audit_uuid, event_type, status):
event = Event()
@@ -56,27 +69,25 @@ class DefaultAuditHandler(BaseAuditHandler):
try:
LOG.debug("Trigger audit %s" % audit_uuid)
# change state to ONGOING
# change state of the audit to ONGOING
audit = self.update_audit_state(request_context, audit_uuid,
AuditStatus.ONGOING)
# Retrieve cluster-data-model
cluster = self.model_collector.get_latest_cluster_data_model()
cluster = self.cluster_collector.get_latest_cluster_data_model()
# Select appropriate strategy
# Retrieve the Audit Template
audit_template = AuditTemplate.get_by_id(request_context,
audit.audit_template_id)
# execute the strategy
solution = self.strategy_context.execute_strategy(audit_template.
goal, cluster)
self.strategy_context.set_goal(audit_template.goal)
# compute change requests
solution = self.strategy_context.execute_strategy(cluster)
# create an action plan
# schedule the actions and create in the watcher db the ActionPlan
planner = DefaultPlanner()
planner.schedule(request_context, audit.id, solution)
# change state to SUCCEEDED and notify
# change state of the audit to SUCCEEDED
self.update_audit_state(request_context, audit_uuid,
AuditStatus.SUCCEEDED)
except Exception as e:

28
watcher/decision_engine/manager.py
View File

@@ -16,7 +16,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
from concurrent.futures import ThreadPoolExecutor
from oslo_config import cfg
from oslo_log import log
@@ -26,7 +25,6 @@ from watcher.common.messaging.notification_handler import NotificationHandler
from watcher.decision_engine.event.consumer_factory import EventConsumerFactory
from watcher.decision_engine.messaging.audit_endpoint import AuditEndpoint
from watcher.decision_engine.messaging.events import Events
from watcher.decision_engine.strategy.context.default import StrategyContext
LOG = log.getLogger(__name__)
CONF = cfg.CONF
@@ -47,31 +45,35 @@ WATCHER_DECISION_ENGINE_OPTS = [
cfg.StrOpt('publisher_id',
default='watcher.decision.api',
help='The identifier used by watcher '
'module on the message broker')
'module on the message broker'),
cfg.IntOpt('max_workers',
default=2,
required=True,
help='The maximum number of threads that can be used to '
'execute strategies',
),
]
decision_engine_opt_group = cfg.OptGroup(
name='watcher_decision_engine',
title='Defines the parameters of the module decision engine')
decision_engine_opt_group = cfg.OptGroup(name='watcher_decision_engine',
title='Defines the parameters of '
'the module decision engine')
CONF.register_group(decision_engine_opt_group)
CONF.register_opts(WATCHER_DECISION_ENGINE_OPTS, decision_engine_opt_group)
class DecisionEngineManager(MessagingCore):
def __init__(self):
super(DecisionEngineManager, self).__init__(
CONF.watcher_decision_engine.publisher_id,
CONF.watcher_decision_engine.topic_control,
CONF.watcher_decision_engine.topic_status,
api_version=self.API_VERSION,
)
api_version=self.API_VERSION)
self.handler = NotificationHandler(self.publisher_id)
self.handler.register_observer(self)
self.add_event_listener(Events.ALL, self.event_receive)
# todo(jed) oslo_conf
self.executor = ThreadPoolExecutor(max_workers=2)
self.topic_control.add_endpoint(AuditEndpoint(self))
self.context = StrategyContext(self)
endpoint = AuditEndpoint(self,
max_workers=CONF.watcher_decision_engine.
max_workers)
self.topic_control.add_endpoint(endpoint)
def join(self):
self.topic_control.join()

31
watcher/decision_engine/messaging/audit_endpoint.py
View File

@@ -16,6 +16,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
from concurrent.futures import ThreadPoolExecutor
from oslo_log import log
from watcher.decision_engine.audit.default import DefaultAuditHandler
@@ -26,19 +28,32 @@ LOG = log.getLogger(__name__)
class AuditEndpoint(object):
def __init__(self, de):
self.de = de
self.manager = CollectorManager()
def __init__(self, messaging, max_workers):
self._messaging = messaging
self._collector_manager = CollectorManager()
self._executor = ThreadPoolExecutor(max_workers=max_workers)
@property
def collector_manager(self):
return self._collector_manager
@property
def executor(self):
return self._executor
@property
def messaging(self):
return self._messaging
def do_trigger_audit(self, context, audit_uuid):
model_collector = self.manager.get_cluster_model_collector()
model_collector = self.collector_manager.get_cluster_model_collector()
audit = DefaultAuditHandler(self.de, model_collector)
audit = DefaultAuditHandler(self.messaging, model_collector)
audit.execute(audit_uuid, context)
def trigger_audit(self, context, audit_uuid):
LOG.debug("Trigger audit %s" % audit_uuid)
self.de.executor.submit(self.do_trigger_audit,
context,
audit_uuid)
self.executor.submit(self.do_trigger_audit,
context,
audit_uuid)
return audit_uuid

2
watcher/decision_engine/strategy/context/base.py
View File

@@ -23,5 +23,5 @@ import six
@six.add_metaclass(abc.ABCMeta)
class BaseStrategyContext(object):
@abc.abstractmethod
def execute_strategy(self, model):
def execute_strategy(self, goal, cluster_data_model):
raise NotImplementedError()

37
watcher/decision_engine/strategy/context/default.py
View File

@@ -15,34 +15,23 @@
# limitations under the License.
from oslo_log import log
from watcher.decision_engine.planner.default import DefaultPlanner
from watcher.decision_engine.strategy.context.base import BaseStrategyContext
from watcher.decision_engine.strategy.selection.default import StrategySelector
from watcher.decision_engine.strategy.selection.default import \
DefaultStrategySelector
LOG = log.getLogger(__name__)
class StrategyContext(BaseStrategyContext):
def __init__(self, broker=None):
LOG.debug("Initializing decision_engine Engine API ")
self.strategies = {}
self.selected_strategies = []
self.broker = broker
self.planner = DefaultPlanner()
self.strategy_selector = StrategySelector()
self.goal = None
class DefaultStrategyContext(BaseStrategyContext):
def __init__(self):
super(DefaultStrategyContext, self).__init__()
LOG.debug("Initializing Strategy Context")
self._strategy_selector = DefaultStrategySelector()
def add_strategy(self, strategy):
self.strategies[strategy.name] = strategy
self.selected_strategy = strategy.name
@property
def strategy_selector(self):
return self._strategy_selector
def remove_strategy(self, strategy):
pass
def set_goal(self, goal):
self.goal = goal
def execute_strategy(self, model):
# todo(jed) create thread + refactoring
selected_strategy = self.strategy_selector.define_from_goal(self.goal)
return selected_strategy.execute(model)
def execute_strategy(self, goal, cluster_data_model):
selected_strategy = self.strategy_selector.define_from_goal(goal)
return selected_strategy.execute(cluster_data_model)

3
watcher/decision_engine/strategy/selection/default.py
View File

@@ -41,9 +41,10 @@ CONF.register_group(goals_opt_group)
CONF.register_opts(WATCHER_GOALS_OPTS, goals_opt_group)
class StrategySelector(BaseSelector):
class DefaultStrategySelector(BaseSelector):
def __init__(self):
super(DefaultStrategySelector, self).__init__()
self.strategy_loader = DefaultStrategyLoader()
def define_from_goal(self, goal_name):

3
watcher/tests/decision_engine/audit/test_default_audit_handler.py
View File

@@ -42,7 +42,6 @@ class TestDefaultAuditHandler(DbTestCase):
def test_trigger_audit_state_success(self):
model_collector = FakerModelCollector()
audit_handler = DefaultAuditHandler(MagicMock(), model_collector)
audit_handler.strategy_context.execute_strategy = MagicMock()
audit_handler.execute(self.audit.uuid, self.context)
audit = Audit.get_by_uuid(self.context, self.audit.uuid)
self.assertEqual(AuditStatus.SUCCEEDED, audit.state)
@@ -51,8 +50,6 @@ class TestDefaultAuditHandler(DbTestCase):
messaging = MagicMock()
model_collector = FakerModelCollector()
audit_handler = DefaultAuditHandler(messaging, model_collector)
audit_handler.strategy_context.execute_strategy = MagicMock()
audit_handler.execute(self.audit.uuid, self.context)
call_on_going = call(Events.TRIGGER_AUDIT.name, {

10
watcher/tests/decision_engine/messaging/test_audit_endpoint.py
View File

@@ -15,14 +15,15 @@
# limitations under the License.
import mock
from mock import MagicMock
from watcher.common import utils
from watcher.decision_engine.audit.default import DefaultAuditHandler
from watcher.decision_engine.messaging.audit_endpoint import AuditEndpoint
from watcher.metrics_engine.cluster_model_collector.manager import \
CollectorManager
from watcher.tests import base
from watcher.tests.decision_engine.strategy.strategies.faker_cluster_state import \
FakerModelCollector
from watcher.tests.decision_engine.strategy.strategies.faker_cluster_state \
import FakerModelCollector
class DefaultAuditHandlerMock(DefaultAuditHandler):
@@ -36,13 +37,12 @@ class DefaultAuditHandlerMock(DefaultAuditHandler):
class TestAuditEndpoint(base.TestCase):
def setUp(self):
super(TestAuditEndpoint, self).setUp()
self.endpoint = AuditEndpoint(MagicMock())
def test_do_trigger_audit(self):
audit_uuid = utils.generate_uuid()
model_collector = FakerModelCollector()
audit_handler = DefaultAuditHandler(MagicMock(), model_collector)
endpoint = AuditEndpoint(audit_handler)
endpoint = AuditEndpoint(audit_handler, max_workers=2)
with mock.patch.object(CollectorManager, 'get_cluster_model_collector') \
as mock_call2:
@@ -60,7 +60,7 @@ class TestAuditEndpoint(base.TestCase):
model_collector = FakerModelCollector()
audit_handler = DefaultAuditHandlerMock(MagicMock(),
model_collector)
endpoint = AuditEndpoint(audit_handler)
endpoint = AuditEndpoint(audit_handler, max_workers=2)
with mock.patch.object(DefaultAuditHandlerMock, 'executor') \
as mock_call:

29
watcher/tests/decision_engine/strategy/context/test_strategy_context.py
View File

@@ -13,19 +13,26 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from mock import MagicMock
from mock import patch
from watcher.decision_engine.strategy.context.default import StrategyContext
from watcher.decision_engine.solution.default import DefaultSolution
from watcher.decision_engine.strategy.context.default import \
DefaultStrategyContext
from watcher.decision_engine.strategy.selection.default import \
DefaultStrategySelector
from watcher.decision_engine.strategy.strategies.dummy_strategy import \
DummyStrategy
from watcher.tests import base
class FakeStrategy(object):
def __init__(self):
self.name = "BALANCE_LOAD"
class TestStrategyContext(base.BaseTestCase):
def test_add_remove_strategy(self):
strategy = FakeStrategy()
strategy_context = StrategyContext()
strategy_context.add_strategy(strategy)
strategy_context.remove_strategy(strategy)
strategy_context = DefaultStrategyContext()
@patch.object(DefaultStrategySelector, 'define_from_goal')
def test_execute_strategy(self, mock_call):
mock_call.return_value = DummyStrategy()
cluster_data_model = MagicMock()
solution = self.strategy_context.execute_strategy("dummy",
cluster_data_model)
self.assertIsInstance(solution, DefaultSolution)

23
watcher/tests/decision_engine/strategy/selector/test_strategy_selector.py
View File

@@ -19,21 +19,20 @@ from oslo_config import cfg
from watcher.common.exception import WatcherException
from watcher.decision_engine.strategy.loading.default import \
DefaultStrategyLoader
from watcher.decision_engine.strategy.selection.default import StrategySelector
from watcher.decision_engine.strategy.selection.default import \
DefaultStrategySelector
from watcher.tests.base import TestCase
CONF = cfg.CONF
class TestStrategySelector(TestCase):
strategy_selector = StrategySelector()
strategy_selector = DefaultStrategySelector()
@patch.object(DefaultStrategyLoader, 'load')
def test_define_from_goal(self, mock_call):
cfg.CONF.set_override(
'goals', {"DUMMY": "fake"}, group='watcher_goals'
)
cfg.CONF.set_override('goals',
{"DUMMY": "fake"}, group='watcher_goals')
expected_goal = 'DUMMY'
expected_strategy = CONF.watcher_goals.goals[expected_goal]
self.strategy_selector.define_from_goal(expected_goal)
@@ -41,12 +40,8 @@ class TestStrategySelector(TestCase):
@patch.object(DefaultStrategyLoader, 'load')
def test_define_from_goal_with_incorrect_mapping(self, mock_call):
cfg.CONF.set_override(
'goals', {}, group='watcher_goals'
)
self.assertRaises(
WatcherException,
self.strategy_selector.define_from_goal,
"DUMMY"
)
cfg.CONF.set_override('goals', {}, group='watcher_goals')
self.assertRaises(WatcherException,
self.strategy_selector.define_from_goal,
"DUMMY")
self.assertEqual(mock_call.call_count, 0)