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

Merge "Add continuously optimization"

Browse Source
This commit is contained in:
Jenkins
2016-07-08 14:07:51 +00:00
committed by Gerrit Code Review
parent e6b4befbf4 1de00086f5
commit b25ecde0db
18 changed files with 516 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
watcher/decision_engine/audit/base.py
View File

@@ -2,6 +2,7 @@
# Copyright (c) 2015 b<>com
#
# Authors: Jean-Emile DARTOIS <jean-emile.dartois@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.
@@ -19,9 +20,92 @@
import abc
import six
from oslo_log import log
from watcher.common.messaging.events import event as watcher_event
from watcher.decision_engine.messaging import events as de_events
from watcher.decision_engine.planner import manager as planner_manager
from watcher.decision_engine.strategy.context import default as default_context
from watcher.objects import audit as audit_objects
LOG = log.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class BaseAuditHandler(object):
@abc.abstractmethod
def execute(self, audit_uuid, request_context):
raise NotImplementedError()
@abc.abstractmethod
def pre_execute(self, audit_uuid, request_context):
raise NotImplementedError()
@abc.abstractmethod
def do_execute(self, audit, request_context):
raise NotImplementedError()
@abc.abstractmethod
def post_execute(self, audit, solution, request_context):
raise NotImplementedError()
@six.add_metaclass(abc.ABCMeta)
class AuditHandler(BaseAuditHandler):
def __init__(self, messaging):
self._messaging = messaging
self._strategy_context = default_context.DefaultStrategyContext()
self._planner_manager = planner_manager.PlannerManager()
self._planner = None
@property
def planner(self):
if self._planner is None:
self._planner = self._planner_manager.load()
return self._planner
@property
def messaging(self):
return self._messaging
@property
def strategy_context(self):
return self._strategy_context
def notify(self, audit_uuid, event_type, status):
event = watcher_event.Event()
event.type = event_type
event.data = {}
payload = {'audit_uuid': audit_uuid,
'audit_status': status}
self.messaging.status_topic_handler.publish_event(
event.type.name, payload)
def update_audit_state(self, request_context, audit, state):
LOG.debug("Update audit state: %s", state)
audit.state = state
audit.save()
self.notify(audit.uuid, de_events.Events.TRIGGER_AUDIT, state)
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)
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)
def execute(self, audit, request_context):
try:
self.pre_execute(audit, request_context)
solution = self.do_execute(audit, request_context)
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)

126
watcher/decision_engine/audit/continuous.py Normal file
View File

@@ -0,0 +1,126 @@
# -*- encoding: utf-8 -*-
# Copyright (c) 2016 Servionica LTD
#
# Authors: 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 datetime
from apscheduler.schedulers import background
from oslo_config import cfg
from watcher.common import context
from watcher.decision_engine.audit import base
from watcher.objects import action_plan as action_objects
from watcher.objects import audit as audit_objects
CONF = cfg.CONF
WATCHER_CONTINUOUS_OPTS = [
cfg.IntOpt('continuous_audit_interval',
default=10,
help='Interval, in seconds, for checking new created'
'continuous audit.')
]
CONF.register_opts(WATCHER_CONTINUOUS_OPTS, 'watcher_decision_engine')
class ContinuousAuditHandler(base.AuditHandler):
def __init__(self, messaging):
super(ContinuousAuditHandler, self).__init__(messaging)
self._scheduler = None
self.jobs = []
self._start()
self.context_show_deleted = context.RequestContext(is_admin=True,
show_deleted=True)
@property
def scheduler(self):
if self._scheduler is None:
self._scheduler = background.BackgroundScheduler()
return self._scheduler
def _is_audit_inactive(self, audit):
audit = audit_objects.Audit.get_by_uuid(self.context_show_deleted,
audit.uuid)
if audit.state in (audit_objects.State.CANCELLED,
audit_objects.State.DELETED,
audit_objects.State.FAILED):
# if audit isn't in active states, audit's job must be removed to
# prevent using of inactive audit in future.
job_to_delete = [job for job in self.jobs
if job.keys()[0] == audit.uuid][0]
self.jobs.remove(job_to_delete)
job_to_delete[audit.uuid].remove()
return True
return False
def do_execute(self, audit, request_context):
# execute the strategy
solution = self.strategy_context.execute_strategy(audit.uuid,
request_context)
if audit.audit_type == audit_objects.AuditType.CONTINUOUS.value:
a_plan_filters = {'audit_uuid': audit.uuid,
'state': action_objects.State.RECOMMENDED}
action_plans = action_objects.ActionPlan.list(
request_context,
filters=a_plan_filters)
for plan in action_plans:
plan.state = action_objects.State.CANCELLED
plan.save()
return solution
def execute_audit(self, audit, request_context):
if not self._is_audit_inactive(audit):
self.execute(audit, request_context)
def post_execute(self, audit, solution, request_context):
self.planner.schedule(request_context, audit.id, solution)
def launch_audits_periodically(self):
audit_context = context.RequestContext(is_admin=True)
audit_filters = {
'audit_type': audit_objects.AuditType.CONTINUOUS.value,
'state__in': (audit_objects.State.PENDING,
audit_objects.State.ONGOING,
audit_objects.State.SUCCEEDED)
}
audits = audit_objects.Audit.list(audit_context,
filters=audit_filters)
scheduler_job_args = [job.args for job in self.scheduler.get_jobs()
if job.name == 'execute_audit']
for audit in audits:
if audit.uuid not in [arg[0].uuid for arg in scheduler_job_args]:
job = self.scheduler.add_job(
self.execute_audit, 'interval',
args=[audit, audit_context],
seconds=audit.interval,
name='execute_audit',
next_run_time=datetime.datetime.now())
self.jobs.append({audit.uuid: job})
def _start(self):
self.scheduler.add_job(
self.launch_audits_periodically,
'interval',
seconds=CONF.watcher_decision_engine.continuous_audit_interval,
next_run_time=datetime.datetime.now())
self.scheduler.start()

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

@@ -1,88 +0,0 @@
# -*- encoding: utf-8 -*-
# Copyright (c) 2015 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 oslo_log import log
from watcher.common.messaging.events import event as watcher_event
from watcher.decision_engine.audit import base
from watcher.decision_engine.messaging import events as de_events
from watcher.decision_engine.planner import manager as planner_manager
from watcher.decision_engine.strategy.context import default as default_context
from watcher.objects import audit as audit_objects
LOG = log.getLogger(__name__)
class DefaultAuditHandler(base.BaseAuditHandler):
def __init__(self, messaging):
super(DefaultAuditHandler, self).__init__()
self._messaging = messaging
self._strategy_context = default_context.DefaultStrategyContext()
self._planner_manager = planner_manager.PlannerManager()
self._planner = None
@property
def planner(self):
if self._planner is None:
self._planner = self._planner_manager.load()
return self._planner
@property
def messaging(self):
return self._messaging
@property
def strategy_context(self):
return self._strategy_context
def notify(self, audit_uuid, event_type, status):
event = watcher_event.Event()
event.type = event_type
event.data = {}
payload = {'audit_uuid': audit_uuid,
'audit_status': status}
self.messaging.status_topic_handler.publish_event(
event.type.name, payload)
def update_audit_state(self, request_context, audit_uuid, state):
LOG.debug("Update audit state: %s", state)
audit = audit_objects.Audit.get_by_uuid(request_context, audit_uuid)
audit.state = state
audit.save()
self.notify(audit_uuid, de_events.Events.TRIGGER_AUDIT, state)
return audit
def execute(self, audit_uuid, request_context):
try:
LOG.debug("Trigger audit %s", audit_uuid)
# change state of the audit to ONGOING
audit = self.update_audit_state(request_context, audit_uuid,
audit_objects.State.ONGOING)
# execute the strategy
solution = self.strategy_context.execute_strategy(audit_uuid,
request_context)
self.planner.schedule(request_context, audit.id, solution)
# change state of the audit to SUCCEEDED
self.update_audit_state(request_context, audit_uuid,
audit_objects.State.SUCCEEDED)
except Exception as e:
LOG.exception(e)
self.update_audit_state(request_context, audit_uuid,
audit_objects.State.FAILED)

26
watcher/decision_engine/audit/oneshot.py Normal file
View File

@@ -0,0 +1,26 @@
# -*- encoding: utf-8 -*-
# Copyright (c) 2015 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 watcher.decision_engine.audit import base
class OneShotAuditHandler(base.AuditHandler):
def do_execute(self, audit, request_context):
# execute the strategy
solution = self.strategy_context.execute_strategy(audit.uuid,
request_context)
return solution

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

@@ -21,7 +21,9 @@ from concurrent import futures
from oslo_config import cfg
from oslo_log import log
from watcher.decision_engine.audit import default
from watcher.decision_engine.audit import continuous as continuous_handler
from watcher.decision_engine.audit import oneshot as oneshot_handler
from watcher.objects import audit as audit_objects
CONF = cfg.CONF
LOG = log.getLogger(__name__)
@@ -32,6 +34,10 @@ class AuditEndpoint(object):
self._messaging = messaging
self._executor = futures.ThreadPoolExecutor(
max_workers=CONF.watcher_decision_engine.max_workers)
self._oneshot_handler = oneshot_handler.OneShotAuditHandler(
self.messaging)
self._continuous_handler = continuous_handler.ContinuousAuditHandler(
self.messaging)
@property
def executor(self):
@@ -42,8 +48,8 @@ class AuditEndpoint(object):
return self._messaging
def do_trigger_audit(self, context, audit_uuid):
audit = default.DefaultAuditHandler(self.messaging)
audit.execute(audit_uuid, context)
audit = audit_objects.Audit.get_by_uuid(context, audit_uuid)
self._oneshot_handler.execute(audit, context)
def trigger_audit(self, context, audit_uuid):
LOG.debug("Trigger audit %s" % audit_uuid)