Stop periodic watcher tasks before deleting stack

In the case where a stack that has a watch task associated with it is
being deleted and the engine successfully acquires a lock, no attempt to
stop the stack will be made since it has no action in-progress.  This
will result in orphaned watcher tasks being left behind, since the
watcher tasks are normally stopped as part of the call to
ThreadGroup.stop().

To fix this, explicitly stop the periodic watcher tasks before calling
stack.delete().

Change-Id: Idf5b7c1100fc85743ac6dc5831731f7ecc96bfd6
Closes-Bug: #1315044

heat/engine/service.py

@@ -150,6 +150,10 @@ class ThreadGroupManager(object):
         self.groups[stack_id].add_timer(cfg.CONF.periodic_interval,
                                         func, *args, **kwargs)
 
+    def stop_timers(self, stack_id):
+        if stack_id in self.groups:
+            self.groups[stack_id].stop_timers()
+
     def stop(self, stack_id, graceful=False):
         '''Stop any active threads on a stack.'''
         if stack_id in self.groups:
@@ -712,6 +716,7 @@ class EngineService(service.Service):
 
         # Successfully acquired lock
         if acquire_result is None:
+            self.thread_group_mgr.stop_timers(stack.id)
             self.thread_group_mgr.start_with_acquired_lock(stack, lock,
                                                            stack.delete)
             return

heat/tests/test_engine_service.py

@@ -680,6 +680,27 @@ class StackServiceCreateUpdateDeleteTest(HeatTestCase):
         self.man.thread_group_mgr.groups[sid].wait()
         self.m.VerifyAll()
 
+    def test_stack_delete_acquired_lock_stop_timers(self):
+        stack_name = 'service_delete_test_stack'
+        stack = get_wordpress_stack(stack_name, self.ctx)
+        sid = stack.store()
+
+        st = db_api.stack_get(self.ctx, sid)
+        self.m.StubOutWithMock(parser.Stack, 'load')
+        parser.Stack.load(self.ctx, stack=st).MultipleTimes().AndReturn(stack)
+        self.man.tg = DummyThreadGroup()
+
+        self.m.StubOutWithMock(stack_lock.StackLock, 'try_acquire')
+        stack_lock.StackLock.try_acquire().AndReturn(self.man.engine_id)
+        self.m.ReplayAll()
+
+        self.man.thread_group_mgr.add_timer(stack.id, 'test')
+        self.assertEqual(1, len(self.man.thread_group_mgr.groups[sid].timers))
+        self.assertIsNone(self.man.delete_stack(self.ctx, stack.identifier()))
+        self.assertEqual(0, len(self.man.thread_group_mgr.groups[sid].timers))
+        self.man.thread_group_mgr.groups[sid].wait()
+        self.m.VerifyAll()
+
     def test_stack_delete_current_engine_active_lock(self):
         stack_name = 'service_delete_test_stack'
         stack = get_wordpress_stack(stack_name, self.ctx)