Add HA support

This patch set adds hostname field to Audit and Action Plan
objects to track services which execute these objects.

Change-Id: I786e419952925c380c969b12cc60f9a1004af96b
Partially-Implements: blueprint support-watcher-ha-active-active-mode

.zuul.yaml

@@ -1,7 +1,8 @@
 - project:
     check:
       jobs:
-        - watcher-tempest-functional
+        - watcher-tempest-functional:
+            voting: false
         - watcher-tempest-dummy_optim
         - watcher-tempest-actuator
         - watcher-tempest-basic_optim
@@ -11,7 +12,7 @@
         - openstack-tox-lower-constraints
     gate:
       jobs:
-        - watcher-tempest-functional
+        # - watcher-tempest-functional
         - openstack-tox-lower-constraints
 
 - job:

releasenotes/notes/add-ha-support-b9042255e5b76e42.yaml

@@ -0,0 +1,6 @@
+---
+features:
+  - Watcher services can be launched in HA mode. From now on Watcher Decision
+    Engine and Watcher Applier services may be deployed on different nodes to
+    run in active-active or active-passive mode. Any ONGOING Audits or Action Plans
+    will be CANCELLED if service they are executed on is restarted.

watcher/api/controllers/v1/action_plan.py

@@ -230,6 +230,9 @@ class ActionPlan(base.APIBase):
     links = wsme.wsattr([link.Link], readonly=True)
     """A list containing a self link and associated action links"""
 
+    hostname = wsme.wsattr(wtypes.text, mandatory=False)
+    """Hostname the actionplan is running on"""
+
     def __init__(self, **kwargs):
         super(ActionPlan, self).__init__()
         self.fields = []

watcher/api/controllers/v1/audit.py

@@ -77,6 +77,8 @@ class AuditPostType(wtypes.Base):
 
     auto_trigger = wtypes.wsattr(bool, mandatory=False)
 
+    hostname = wtypes.wsattr(wtypes.text, readonly=True, mandatory=False)
+
     def as_audit(self, context):
         audit_type_values = [val.value for val in objects.audit.AuditType]
         if self.audit_type not in audit_type_values:
@@ -305,6 +307,9 @@ class Audit(base.APIBase):
     next_run_time = wsme.wsattr(datetime.datetime, mandatory=False)
     """The next time audit launch"""
 
+    hostname = wsme.wsattr(wtypes.text, mandatory=False)
+    """Hostname the audit is running on"""
+
     def __init__(self, **kwargs):
         self.fields = []
         fields = list(objects.Audit.fields)

watcher/applier/action_plan/default.py

@@ -16,6 +16,7 @@
 # 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.applier.action_plan import base
@@ -25,6 +26,7 @@ from watcher import notifications
 from watcher import objects
 from watcher.objects import fields
 
+CONF = cfg.CONF
 LOG = log.getLogger(__name__)
 
 
@@ -43,6 +45,7 @@ class DefaultActionPlanHandler(base.BaseActionPlanHandler):
             if action_plan.state == objects.action_plan.State.CANCELLED:
                 self._update_action_from_pending_to_cancelled()
                 return
+            action_plan.hostname = CONF.host
             action_plan.state = objects.action_plan.State.ONGOING
             action_plan.save()
             notifications.action_plan.send_action_notification(

watcher/applier/sync.py

@@ -15,12 +15,19 @@
 # limitations under the License.
 #
 
+from oslo_config import cfg
+from oslo_log import log
+
 from watcher.applier.loading import default
 from watcher.common import context
 from watcher.common import exception
 from watcher import objects
 
 
+CONF = cfg.CONF
+LOG = log.getLogger(__name__)
+
+
 class Syncer(object):
     """Syncs all available actions with the Watcher DB"""
 
@@ -42,3 +49,27 @@ class Syncer(object):
                 obj_action_desc.action_type = action_type
                 obj_action_desc.description = load_description
                 obj_action_desc.create()
+        self._cancel_ongoing_actionplans(ctx)
+
+    def _cancel_ongoing_actionplans(self, context):
+        actions_plans = objects.ActionPlan.list(
+            context,
+            filters={'state': objects.action_plan.State.ONGOING,
+                     'hostname': CONF.host},
+            eager=True)
+        for ap in actions_plans:
+            ap.state = objects.action_plan.State.CANCELLED
+            ap.save()
+            filters = {'action_plan_uuid': ap.uuid,
+                       'state__in': (objects.action.State.PENDING,
+                                     objects.action.State.ONGOING)}
+            actions = objects.Action.list(context, filters=filters, eager=True)
+            for a in actions:
+                a.state = objects.action.State.CANCELLED
+                a.save()
+            LOG.info("Action Plan %(uuid)s along with appropriate Actions "
+                     "has been cancelled because it was in %(state)s state "
+                     "when Applier had been stopped on %(hostname)s host.",
+                     {'uuid': ap.uuid,
+                      'state': objects.action_plan.State.ONGOING,
+                      'hostname': ap.hostname})

watcher/db/sqlalchemy/alembic/versions/52804f2498c4_add_hostname.py

@@ -0,0 +1,26 @@
+"""Add hostname field to both Audit and Action Plan models
+
+Revision ID: 52804f2498c4
+Revises: a86240e89a29
+Create Date: 2018-06-26 13:06:45.530387
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '52804f2498c4'
+down_revision = 'a86240e89a29'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    for table in ('audits', 'action_plans'):
+        op.add_column(
+            table,
+            sa.Column('hostname', sa.String(length=255), nullable=True))
+
+
+def downgrade():
+    for table in ('audits', 'action_plans'):
+        op.drop_column(table, 'hostname')

watcher/db/sqlalchemy/alembic/versions/a86240e89a29_.py

@@ -19,9 +19,15 @@ def upgrade():
     connection = op.get_bind()
     session = sessionmaker()
     s = session(bind=connection)
-    for audit in s.query(models.Audit).filter(models.Audit.name is None).all():
+    audits = s.query(
-        strategy_name = s.query(models.Strategy).filter_by(id=audit.strategy_id).one().name
+        models.Audit.strategy_id.label('strategy_id'),
-        audit.update({'name': strategy_name + '-' + str(audit.created_at)})
+        models.Audit.created_at.label('created_at')).filter(
+        models.Audit.name is None).all()
+    for audit in audits:
+        strategy_name = s.query(models.Strategy).filter_by(
+            id=audit.strategy_id).one().name
+        s.query().filter(models.Audit.name is None).update(
+            {'name': strategy_name + '-' + str(audit.created_at)})
     s.commit()
 
 
@@ -29,6 +35,11 @@ def downgrade():
     connection = op.get_bind()
     session = sessionmaker()
     s = session(bind=connection)
-    for audit in s.query(models.Audit).filter(models.Audit.name is not None).all():
+    audits = s.query(
-        audit.update({'name': None})
+        models.Audit.strategy_id.label('strategy_id'),
+        models.Audit.created_at.label('created_at')).filter(
+        models.Audit.name is not None).all()
+    for audit in audits:
+        s.query().filter(models.Audit.name is not None).update(
+            {'name': None})
     s.commit()

watcher/db/sqlalchemy/models.py

@@ -181,6 +181,7 @@ class Audit(Base):
     scope = Column(JSONEncodedList, nullable=True)
     auto_trigger = Column(Boolean, nullable=False)
     next_run_time = Column(DateTime, nullable=True)
+    hostname = Column(String(255), nullable=True)
 
     goal = orm.relationship(Goal, foreign_keys=goal_id, lazy=None)
     strategy = orm.relationship(Strategy, foreign_keys=strategy_id, lazy=None)
@@ -200,6 +201,7 @@ class ActionPlan(Base):
     strategy_id = Column(Integer, ForeignKey('strategies.id'), nullable=False)
     state = Column(String(20), nullable=True)
     global_efficacy = Column(JSONEncodedList, nullable=True)
+    hostname = Column(String(255), nullable=True)
 
     audit = orm.relationship(Audit, foreign_keys=audit_id, lazy=None)
     strategy = orm.relationship(Strategy, foreign_keys=strategy_id, lazy=None)

watcher/decision_engine/audit/base.py

@@ -20,6 +20,7 @@
 import abc
 import six
 
+from oslo_config import cfg
 from oslo_log import log
 
 from watcher.applier import rpcapi
@@ -31,6 +32,7 @@ from watcher import notifications
 from watcher import objects
 from watcher.objects import fields
 
+CONF = cfg.CONF
 LOG = log.getLogger(__name__)
 
 
@@ -120,6 +122,8 @@ class AuditHandler(BaseAuditHandler):
     def pre_execute(self, audit, request_context):
         LOG.debug("Trigger audit %s", audit.uuid)
         self.check_ongoing_action_plans(request_context)
+        # Write hostname that will execute this audit.
+        audit.hostname = CONF.host
         # change state of the audit to ONGOING
         self.update_audit_state(audit, objects.audit.State.ONGOING)
 

watcher/decision_engine/audit/continuous.py

@@ -123,10 +123,20 @@ class ContinuousAuditHandler(base.AuditHandler):
             'audit_type': objects.audit.AuditType.CONTINUOUS.value,
             'state__in': (objects.audit.State.PENDING,
                           objects.audit.State.ONGOING,
-                          objects.audit.State.SUCCEEDED)
+                          objects.audit.State.SUCCEEDED),
+            'hostname__in': (None, CONF.host)
         }
         audits = objects.Audit.list(
             audit_context, filters=audit_filters, eager=True)
+        for audit in audits:
+            # If continuous audit doesn't have a hostname yet,
+            # Watcher will set current CONF.host value.
+            if audit.hostname is None:
+                audit.hostname = CONF.host
+                audit.save()
+                # Let's remove this audit from current execution
+                # and execute it as usual Audit with hostname later.
+                audits.remove(audit)
         scheduler_job_args = [
             (job.args[0].uuid, job) for job
             in self.scheduler.get_jobs()
@@ -172,6 +182,7 @@ class ContinuousAuditHandler(base.AuditHandler):
                     audit.next_run_time = self._next_cron_time(audit)
                     self._add_job('date', audit, audit_context,
                                   run_date=audit.next_run_time)
+                audit.hostname = CONF.host
                 audit.save()
 
     def start(self):

watcher/decision_engine/scheduling.py

@@ -88,10 +88,31 @@ class DecisionEngineSchedulingService(scheduling.BackgroundSchedulerService):
                          seconds=interval,
                          next_run_time=datetime.datetime.now())
 
+    def cancel_ongoing_audits(self):
+        audit_filters = {
+            'audit_type': objects.audit.AuditType.ONESHOT.value,
+            'state': objects.audit.State.ONGOING,
+            'hostname': CONF.host
+        }
+        local_context = context.make_context()
+        ongoing_audits = objects.Audit.list(
+            local_context,
+            filters=audit_filters)
+        for audit in ongoing_audits:
+            audit.state = objects.audit.State.CANCELLED
+            audit.save()
+            LOG.info("Audit %(uuid)s has been cancelled because it was in "
+                     "%(state)s state when Decision Engine had been stopped "
+                     "on %(hostname)s host.",
+                     {'uuid': audit.uuid,
+                      'state': objects.audit.State.ONGOING,
+                      'hostname': audit.hostname})
+
     def start(self):
         """Start service."""
         self.add_sync_jobs()
         self.add_checkstate_job()
+        self.cancel_ongoing_audits()
         super(DecisionEngineSchedulingService, self).start()
 
     def stop(self):

watcher/objects/action_plan.py

@@ -106,7 +106,8 @@ class ActionPlan(base.WatcherPersistentObject, base.WatcherObject,
     # Version 1.2: audit_id is not nullable anymore
     # Version 2.0: Removed 'first_action_id' object field
     # Version 2.1: Changed global_efficacy type
-    VERSION = '2.1'
+    # Version 2.2: Added 'hostname' field
+    VERSION = '2.2'
 
     dbapi = db_api.get_instance()
 
@@ -117,6 +118,7 @@ class ActionPlan(base.WatcherPersistentObject, base.WatcherObject,
         'strategy_id': wfields.IntegerField(),
         'state': wfields.StringField(nullable=True),
         'global_efficacy': wfields.FlexibleListOfDictField(nullable=True),
+        'hostname': wfields.StringField(nullable=True),
 
         'audit': wfields.ObjectField('Audit', nullable=True),
         'strategy': wfields.ObjectField('Strategy', nullable=True),

watcher/objects/audit.py

@@ -87,7 +87,8 @@ class Audit(base.WatcherPersistentObject, base.WatcherObject,
     # Version 1.3: Added 'next_run_time' DateTime field,
     #              'interval' type has been changed from Integer to String
     # Version 1.4: Added 'name' string field
-    VERSION = '1.4'
+    # Version 1.5: Added 'hostname' field
+    VERSION = '1.5'
 
     dbapi = db_api.get_instance()
 
@@ -105,6 +106,7 @@ class Audit(base.WatcherPersistentObject, base.WatcherObject,
         'auto_trigger': wfields.BooleanField(),
         'next_run_time': wfields.DateTimeField(nullable=True,
                                                tzinfo_aware=False),
+        'hostname': wfields.StringField(nullable=True),
 
         'goal': wfields.ObjectField('Goal', nullable=True),
         'strategy': wfields.ObjectField('Strategy', nullable=True),

watcher/tests/api/v1/test_audits.py

@@ -497,6 +497,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         response = self.post_json('/audits', audit_dict)
         self.assertEqual('application/json', response.content_type)
@@ -540,6 +541,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
         # Make the audit template UUID some garbage value
         audit_dict['audit_template_uuid'] = (
             '01234567-8910-1112-1314-151617181920')
@@ -563,6 +565,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
         with mock.patch.object(self.dbapi, 'create_audit',
                                wraps=self.dbapi.create_audit) as cn_mock:
             response = self.post_json('/audits', audit_dict)
@@ -581,6 +584,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         response = self.post_json('/audits', audit_dict)
         self.assertEqual('application/json', response.content_type)
@@ -598,6 +602,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['state']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
         audit_dict['audit_type'] = objects.audit.AuditType.CONTINUOUS.value
         audit_dict['interval'] = '1200'
 
@@ -619,6 +624,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['state']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
         audit_dict['audit_type'] = objects.audit.AuditType.CONTINUOUS.value
         audit_dict['interval'] = '* * * * *'
 
@@ -640,6 +646,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['state']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
         audit_dict['audit_type'] = objects.audit.AuditType.CONTINUOUS.value
         audit_dict['interval'] = 'zxc'
 
@@ -662,6 +669,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         response = self.post_json('/audits', audit_dict, expect_errors=True)
         self.assertEqual(400, response.status_int)
@@ -681,6 +689,7 @@ class TestPost(api_base.FunctionalTest):
         audit_dict['audit_type'] = objects.audit.AuditType.ONESHOT.value
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         response = self.post_json('/audits', audit_dict, expect_errors=True)
         self.assertEqual(400, response.status_int)
@@ -698,6 +707,7 @@ class TestPost(api_base.FunctionalTest):
             del audit_dict['interval']
             del audit_dict['scope']
             del audit_dict['next_run_time']
+            del audit_dict['hostname']
             response = self.post_json('/audits', audit_dict)
             de_mock.assert_called_once_with(mock.ANY, response.json['uuid'])
 
@@ -722,6 +732,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         response = self.post_json('/audits', audit_dict, expect_errors=True)
         self.assertEqual('application/json', response.content_type)
@@ -744,6 +755,7 @@ class TestPost(api_base.FunctionalTest):
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         response = self.post_json('/audits', audit_dict, expect_errors=True)
         self.assertEqual('application/json', response.content_type)
@@ -766,7 +778,7 @@ class TestPost(api_base.FunctionalTest):
 
         audit_dict['audit_template_uuid'] = audit_template['uuid']
         del_keys = ['uuid', 'goal_id', 'strategy_id', 'state', 'interval',
-                    'scope', 'next_run_time']
+                    'scope', 'next_run_time', 'hostname']
         for k in del_keys:
             del audit_dict[k]
 
@@ -822,12 +834,13 @@ class TestPost(api_base.FunctionalTest):
         audit_dict = post_get_test_audit()
         normal_name = 'this audit name is just for test'
         # long_name length exceeds 63 characters
-        long_name = normal_name+audit_dict['uuid']
+        long_name = normal_name + audit_dict['uuid']
         del audit_dict['uuid']
         del audit_dict['state']
         del audit_dict['interval']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
 
         audit_dict['name'] = normal_name
         response = self.post_json('/audits', audit_dict)
@@ -954,6 +967,7 @@ class TestAuditPolicyEnforcement(api_base.FunctionalTest):
         del audit_dict['state']
         del audit_dict['scope']
         del audit_dict['next_run_time']
+        del audit_dict['hostname']
         self._common_policy_check(
             "audit:create", self.post_json, '/audits', audit_dict,
             expect_errors=True)

watcher/tests/applier/test_sync.py

@@ -0,0 +1,86 @@
+# -*- encoding: utf-8 -*-
+# Copyright (c) 2018 SBCloud
+#
+# Authors: Alexander Chadin <aschadin@sbcloud.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 mock
+
+from oslo_config import cfg
+from oslo_utils import uuidutils
+
+from watcher.applier import sync
+from watcher.decision_engine.strategy.strategies import dummy_strategy
+from watcher.tests.db import base as db_base
+
+from watcher import notifications
+from watcher import objects
+from watcher.tests.objects import utils as obj_utils
+
+
+class TestCancelOngoingActionPlans(db_base.DbTestCase):
+
+    def setUp(self):
+        super(TestCancelOngoingActionPlans, self).setUp()
+        p_audit_notifications = mock.patch.object(
+            notifications, 'audit', autospec=True)
+        self.m_audit_notifications = p_audit_notifications.start()
+        self.addCleanup(p_audit_notifications.stop)
+
+        self.goal = obj_utils.create_test_goal(
+            self.context, id=1, name=dummy_strategy.DummyStrategy.get_name())
+        self.strategy = obj_utils.create_test_strategy(
+            self.context, name=dummy_strategy.DummyStrategy.get_name(),
+            goal_id=self.goal.id)
+        audit_template = obj_utils.create_test_audit_template(
+            self.context, strategy_id=self.strategy.id)
+        self.audit = obj_utils.create_test_audit(
+            self.context,
+            id=999,
+            name='My Audit 999',
+            uuid=uuidutils.generate_uuid(),
+            audit_template_id=audit_template.id,
+            goal_id=self.goal.id,
+            audit_type=objects.audit.AuditType.ONESHOT.value,
+            goal=self.goal,
+            hostname='hostname1',
+            state=objects.audit.State.ONGOING)
+        self.actionplan = obj_utils.create_test_action_plan(
+            self.context,
+            state=objects.action_plan.State.ONGOING,
+            audit_id=999,
+            hostname='hostname1')
+        self.action = obj_utils.create_test_action(
+            self.context,
+            action_plan_id=1,
+            state=objects.action.State.PENDING)
+        cfg.CONF.set_override('host', 'hostname1')
+
+    @mock.patch.object(objects.action.Action, 'save')
+    @mock.patch.object(objects.action_plan.ActionPlan, 'save')
+    @mock.patch.object(objects.action.Action, 'list')
+    @mock.patch.object(objects.action_plan.ActionPlan, 'list')
+    def test_cancel_ongoing_actionplans(self, m_plan_list, m_action_list,
+                                        m_plan_save, m_action_save):
+        m_plan_list.return_value = [self.actionplan]
+        m_action_list.return_value = [self.action]
+        syncer = sync.Syncer()
+
+        syncer._cancel_ongoing_actionplans(self.context)
+        m_plan_list.assert_called()
+        m_action_list.assert_called()
+        m_plan_save.assert_called()
+        m_action_save.assert_called()
+        self.assertEqual(self.action.state, objects.audit.State.CANCELLED)

watcher/tests/db/utils.py

@@ -95,7 +95,8 @@ def get_test_audit(**kwargs):
         'strategy_id': kwargs.get('strategy_id', None),
         'scope': kwargs.get('scope', []),
         'auto_trigger': kwargs.get('auto_trigger', False),
-        'next_run_time': kwargs.get('next_run_time')
+        'next_run_time': kwargs.get('next_run_time'),
+        'hostname': kwargs.get('hostname', 'host_1'),
     }
     # ObjectField doesn't allow None nor dict, so if we want to simulate a
     # non-eager object loading, the field should not be referenced at all.
@@ -171,6 +172,7 @@ def get_test_action_plan(**kwargs):
         'created_at': kwargs.get('created_at'),
         'updated_at': kwargs.get('updated_at'),
         'deleted_at': kwargs.get('deleted_at'),
+        'hostname': kwargs.get('hostname', 'host_1'),
     }
 
     # ObjectField doesn't allow None nor dict, so if we want to simulate a

watcher/tests/decision_engine/test_scheduling.py

@@ -21,12 +21,63 @@ from apscheduler.triggers import interval as interval_trigger
 import eventlet
 import mock
 
+from oslo_config import cfg
+from oslo_utils import uuidutils
+
 from watcher.decision_engine.loading import default as default_loading
 from watcher.decision_engine import scheduling
+from watcher.decision_engine.strategy.strategies import dummy_strategy
+from watcher import notifications
+from watcher import objects
 from watcher.tests import base
+from watcher.tests.db import base as db_base
 from watcher.tests.decision_engine.model import faker_cluster_state
+from watcher.tests.objects import utils as obj_utils
 
 
+class TestCancelOngoingAudits(db_base.DbTestCase):
+
+    def setUp(self):
+        super(TestCancelOngoingAudits, self).setUp()
+        p_audit_notifications = mock.patch.object(
+            notifications, 'audit', autospec=True)
+        self.m_audit_notifications = p_audit_notifications.start()
+        self.addCleanup(p_audit_notifications.stop)
+
+        self.goal = obj_utils.create_test_goal(
+            self.context, id=1, name=dummy_strategy.DummyStrategy.get_name())
+        self.strategy = obj_utils.create_test_strategy(
+            self.context, name=dummy_strategy.DummyStrategy.get_name(),
+            goal_id=self.goal.id)
+        audit_template = obj_utils.create_test_audit_template(
+            self.context, strategy_id=self.strategy.id)
+        self.audit = obj_utils.create_test_audit(
+            self.context,
+            id=999,
+            name='My Audit 999',
+            uuid=uuidutils.generate_uuid(),
+            audit_template_id=audit_template.id,
+            goal_id=self.goal.id,
+            audit_type=objects.audit.AuditType.ONESHOT.value,
+            goal=self.goal,
+            hostname='hostname1',
+            state=objects.audit.State.ONGOING)
+        cfg.CONF.set_override('host', 'hostname1')
+
+    @mock.patch.object(objects.audit.Audit, 'save')
+    @mock.patch.object(objects.audit.Audit, 'list')
+    def test_cancel_ongoing_audits(self, m_list, m_save):
+        m_list.return_value = [self.audit]
+        scheduler = scheduling.DecisionEngineSchedulingService()
+
+        scheduler.cancel_ongoing_audits()
+        m_list.assert_called()
+        m_save.assert_called()
+        self.assertEqual(self.audit.state, objects.audit.State.CANCELLED)
+
+
+@mock.patch.object(objects.audit.Audit, 'save')
+@mock.patch.object(objects.audit.Audit, 'list')
 class TestDecisionEngineSchedulingService(base.TestCase):
 
     @mock.patch.object(
@@ -35,7 +86,7 @@ class TestDecisionEngineSchedulingService(base.TestCase):
         default_loading.ClusterDataModelCollectorLoader, 'list_available')
     @mock.patch.object(background.BackgroundScheduler, 'start')
     def test_start_de_scheduling_service(self, m_start, m_list_available,
-                                         m_load):
+                                         m_load, m_list, m_save):
         m_list_available.return_value = {
             'fake': faker_cluster_state.FakerModelCollector}
         fake_collector = faker_cluster_state.FakerModelCollector(
@@ -61,7 +112,7 @@ class TestDecisionEngineSchedulingService(base.TestCase):
         default_loading.ClusterDataModelCollectorLoader, 'list_available')
     @mock.patch.object(background.BackgroundScheduler, 'start')
     def test_execute_sync_job_fails(self, m_start, m_list_available,
-                                    m_load):
+                                    m_load, m_list, m_save):
         fake_config = mock.Mock(period=.01)
         fake_collector = faker_cluster_state.FakerModelCollector(
             config=fake_config)

watcher/tests/objects/test_objects.py

@@ -412,8 +412,8 @@ expected_object_fingerprints = {
     'Goal': '1.0-93881622db05e7b67a65ca885b4a022e',
     'Strategy': '1.1-73f164491bdd4c034f48083a51bdeb7b',
     'AuditTemplate': '1.1-b291973ffc5efa2c61b24fe34fdccc0b',
-    'Audit': '1.4-f5f27510b8090bce7d1fb45416d58ff1',
+    'Audit': '1.5-e4229dee89e669d1aff0805f5c665bee',
-    'ActionPlan': '2.1-d573f34f2e15da0743afcc38ae62cd22',
+    'ActionPlan': '2.2-3331270cb3666c93408934826d03c08d',
     'Action': '2.0-1dd4959a7e7ac30c62ef170fe08dd935',
     'EfficacyIndicator': '1.0-655b71234a82bc7478aff964639c4bb0',
     'ScoringEngine': '1.0-4abbe833544000728e17bd9e83f97576',