Create watch tasks in its own thread

This change moves the creation of stack watch tasks into its own thread
which retries until it has been done without error.

The intent is to prevent create_periodic_tasks from raising an error
when the database is not available, and retry stack watch creation until
the database becomes available.

Once execution gets past create_periodic_tasks() then systemd is
notified that the service is up, and heat will recover when the database
comes back.

Change-Id: I4d61022db505581f0c1b2feed009c56df18a9ee5
Closes-Bug: #1545885

heat/engine/service.py

@@ -332,18 +332,31 @@ class EngineService(service.Service):
         self.stack_watch = service_stack_watch.StackWatch(
             self.thread_group_mgr)
 
-        # Create a periodic_watcher_task per-stack
+        def create_watch_tasks():
-        admin_context = context.get_admin_context()
+            while True:
-        stacks = stack_object.Stack.get_all(
+                try:
-            admin_context,
+                    # Create a periodic_watcher_task per-stack
-            tenant_safe=False,
+                    admin_context = context.get_admin_context()
-            show_hidden=True)
+                    stacks = stack_object.Stack.get_all(
-        for s in stacks:
+                        admin_context,
-            self.stack_watch.start_watch_task(s.id, admin_context)
+                        tenant_safe=False,
+                        show_hidden=True)
+                    for s in stacks:
+                        self.stack_watch.start_watch_task(s.id, admin_context)
+                    LOG.info(_LI("Watch tasks created"))
+                    return
+                except Exception as e:
+                    LOG.error(_LE("Watch task creation attempt failed, %s"), e)
+                    eventlet.sleep(5)
+
+        if self.manage_thread_grp is None:
+            self.manage_thread_grp = threadgroup.ThreadGroup()
+        self.manage_thread_grp.add_thread(create_watch_tasks)
 
     def start(self):
         self.engine_id = stack_lock.StackLock.generate_engine_id()
-        self.thread_group_mgr = ThreadGroupManager()
+        if self.thread_group_mgr is None:
+            self.thread_group_mgr = ThreadGroupManager()
         self.listener = EngineListener(self.host, self.engine_id,
                                        self.thread_group_mgr)
         LOG.debug("Starting listener for engine %s" % self.engine_id)
@@ -370,7 +383,8 @@ class EngineService(service.Service):
 
         self._configure_db_conn_pool_size()
         self.service_manage_cleanup()
-        self.manage_thread_grp = threadgroup.ThreadGroup()
+        if self.manage_thread_grp is None:
+            self.manage_thread_grp = threadgroup.ThreadGroup()
         self.manage_thread_grp.add_timer(cfg.CONF.periodic_interval,
                                          self.service_manage_report)
         self.manage_thread_grp.add_thread(self.reset_stack_status)

heat/tests/engine/service/test_stack_watch.py

@@ -35,9 +35,12 @@ class StackWatchTest(common.HeatTestCase):
 
         self.ctx = utils.dummy_context(tenant_id='stack_watch_test_tenant')
         self.eng = service.EngineService('a-host', 'a-topic')
-        self.eng.create_periodic_tasks()
         # self.eng.engine_id = 'engine-fake-uuid'
 
+    def _create_periodic_tasks(self):
+        self.eng.create_periodic_tasks()
+        self.eng.manage_thread_grp.wait()
+
     @mock.patch.object(service_stack_watch.StackWatch, 'start_watch_task')
     @mock.patch.object(stack_object.Stack, 'get_all')
     @mock.patch.object(service.service.Service, 'start')
@@ -49,7 +52,7 @@ class StackWatchTest(common.HeatTestCase):
         start_watch_task.return_value = None
 
         self.eng.thread_group_mgr = None
-        self.eng.create_periodic_tasks()
+        self._create_periodic_tasks()
 
         mock_get_all.assert_called_once_with(mock.ANY, tenant_safe=False,
                                              show_hidden=True)
@@ -162,6 +165,7 @@ class StackWatchTest(common.HeatTestCase):
     @tools.stack_context('service_show_watch_state_test_stack')
     @mock.patch.object(stack.Stack, 'resource_by_refid')
     def test_set_watch_state(self, mock_ref):
+        self._create_periodic_tasks()
         # Insert dummy watch rule into the DB
         rule = {u'EvaluationPeriods': u'1',
                 u'AlarmActions': [u'WebServerRestartPolicy'],