Add Watcher JobStore for background jobs

This patch set adds WatcherJobStore class that allows to link
jobs and services.

Partially-Implements: blueprint background-jobs-ha
Change-Id: I575887ca6dae60b3b7709e6d2e2b256e09a3d824

watcher/decision_engine/audit/base.py

@@ -24,6 +24,7 @@ from oslo_log import log
 
 from watcher.applier import rpcapi
 from watcher.common import exception
+from watcher.common import service
 from watcher.decision_engine.planner import manager as planner_manager
 from watcher.decision_engine.strategy.context import default as default_context
 from watcher import notifications
@@ -34,6 +35,7 @@ LOG = log.getLogger(__name__)
 
 
 @six.add_metaclass(abc.ABCMeta)
+@six.add_metaclass(service.Singleton)
 class BaseAuditHandler(object):
 
     @abc.abstractmethod
@@ -55,8 +57,9 @@ class BaseAuditHandler(object):
 
 @six.add_metaclass(abc.ABCMeta)
 class AuditHandler(BaseAuditHandler):
-    def __init__(self, messaging):
-        self._messaging = messaging
+
+    def __init__(self):
+        super(AuditHandler, self).__init__()
         self._strategy_context = default_context.DefaultStrategyContext()
         self._planner_manager = planner_manager.PlannerManager()
         self._planner = None
@@ -67,10 +70,6 @@ class AuditHandler(BaseAuditHandler):
             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
@@ -96,14 +95,12 @@ class AuditHandler(BaseAuditHandler):
                 phase=fields.NotificationPhase.ERROR)
             raise
 
-    @staticmethod
-    def update_audit_state(audit, state):
+    def update_audit_state(self, audit, state):
         LOG.debug("Update audit state: %s", state)
         audit.state = state
         audit.save()
 
-    @staticmethod
-    def check_ongoing_action_plans(request_context):
+    def check_ongoing_action_plans(self, request_context):
         a_plan_filters = {'state': objects.action_plan.State.ONGOING}
         ongoing_action_plans = objects.ActionPlan.list(
             request_context, filters=a_plan_filters)

watcher/decision_engine/audit/continuous.py

@@ -20,30 +20,37 @@
 
 import datetime
 
-from apscheduler.schedulers import background
+from apscheduler.jobstores import memory
 
 from watcher.common import context
+from watcher.common import scheduling
+from watcher import conf
+from watcher.db.sqlalchemy import api as sq_api
+from watcher.db.sqlalchemy import job_store
 from watcher.decision_engine.audit import base
 from watcher import objects
 
-from watcher import conf
 
 CONF = conf.CONF
 
 
 class ContinuousAuditHandler(base.AuditHandler):
-    def __init__(self, messaging):
-        super(ContinuousAuditHandler, self).__init__(messaging)
+    def __init__(self):
+        super(ContinuousAuditHandler, self).__init__()
         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()
+            self._scheduler = scheduling.BackgroundSchedulerService(
+                jobstores={
+                    'default': job_store.WatcherJobStore(
+                        engine=sq_api.get_engine()),
+                    'memory': memory.MemoryJobStore()
+                }
+            )
         return self._scheduler
 
     def _is_audit_inactive(self, audit):
@@ -52,11 +59,9 @@ class ContinuousAuditHandler(base.AuditHandler):
         if objects.audit.AuditStateTransitionManager().is_inactive(audit):
             # 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 list(job.keys())[0] == audit.uuid][0]
-            self.jobs.remove(job_to_delete)
-            job_to_delete[audit.uuid].remove()
-
+            [job for job in self.scheduler.get_jobs()
+             if job.name == 'execute_audit' and
+             job.args[0].uuid == audit.uuid][0].remove()
             return True
 
         return False
@@ -76,7 +81,9 @@ class ContinuousAuditHandler(base.AuditHandler):
                 plan.save()
         return solution
 
-    def execute_audit(self, audit, request_context):
+    @classmethod
+    def execute_audit(cls, audit, request_context):
+        self = cls()
         if not self._is_audit_inactive(audit):
             self.execute(audit, request_context)
 
@@ -90,22 +97,23 @@ class ContinuousAuditHandler(base.AuditHandler):
         }
         audits = objects.Audit.list(
             audit_context, filters=audit_filters, eager=True)
-        scheduler_job_args = [job.args for job in self.scheduler.get_jobs()
-                              if job.name == 'execute_audit']
+        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.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):
+    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())
+            next_run_time=datetime.datetime.now(),
+            jobstore='memory')
         self.scheduler.start()

watcher/decision_engine/audit/oneshot.py

@@ -19,6 +19,7 @@ from watcher import objects
 
 
 class OneShotAuditHandler(base.AuditHandler):
+
     def do_execute(self, audit, request_context):
         # execute the strategy
         solution = self.strategy_context.execute_strategy(

watcher/decision_engine/messaging/audit_endpoint.py

@@ -21,8 +21,9 @@ from concurrent import futures
 from oslo_config import cfg
 from oslo_log import log
 
-from watcher.decision_engine.audit import continuous as continuous_handler
-from watcher.decision_engine.audit import oneshot as oneshot_handler
+from watcher.decision_engine.audit import continuous as c_handler
+from watcher.decision_engine.audit import oneshot as o_handler
+
 from watcher import objects
 
 CONF = cfg.CONF
@@ -35,19 +36,13 @@ 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)
+        self._oneshot_handler = o_handler.OneShotAuditHandler()
+        self._continuous_handler = c_handler.ContinuousAuditHandler().start()
 
     @property
     def executor(self):
         return self._executor
 
-    @property
-    def messaging(self):
-        return self._messaging
-
     def do_trigger_audit(self, context, audit_uuid):
         audit = objects.Audit.get_by_uuid(context, audit_uuid, eager=True)
         self._oneshot_handler.execute(audit, context)