Clean up watch resources after watcher.stop()

We were assuming that watcher threads will be cleaning up after
themselves - i.e. will remove paths from Watcher._watching dict on
Watcher._stop_watch(). Turns out _stop_watch() is killing the threads in
a hard way using thread.stop(). This means that paths are never removed
from Watcher._watching dict and on restart (i.e. Watcher.start()), the
method considers that there is no path that we're not already
processing and does nothing.

This commit fixes that by cleaning up Watcher._watching dict in
Watcher._stop_watch() method.

Closes-Bug: 1790912
Change-Id: I17baaab1769ca5882f0b8edf496f92ac39507969
(cherry picked from commit 346f76292a)

kuryr_kubernetes/tests/unit/test_watcher.py

@@ -332,3 +332,14 @@ class TestWatcher(test_base.TestCase):
 
         m_handler.assert_has_calls([mock.call(e) for e in events])
         m_sys_exit.assert_called_once_with(1)
+
+    def test_watch_restart(self):
+        tg = mock.Mock()
+        w = watcher.Watcher(lambda e: None, tg)
+        w.add('/test')
+        w.start()
+        tg.add_thread.assert_called_once_with(mock.ANY, '/test')
+        w.stop()
+        tg.add_thread = mock.Mock()  # Reset mock.
+        w.start()
+        tg.add_thread.assert_called_once_with(mock.ANY, '/test')

kuryr_kubernetes/watcher.py

@@ -144,14 +144,17 @@ class Watcher(health.HealthHandler):
 
     def _stop_watch(self, path):
         if self._idle.get(path):
-            if self._thread_group:
+            if self._thread_group and path in self._watching:
                 self._watching[path].stop()
+                # NOTE(dulek): Thread gets killed immediately, so we need to
+                # take care of this ourselves.
+                self._watching.pop(path, None)
+                self._idle.pop(path, None)
 
     def _graceful_watch_exit(self, path):
         try:
-            self.remove(path)
-            self._watching.pop(path)
-            self._idle.pop(path)
+            self._watching.pop(path, None)
+            self._idle.pop(path, None)
             LOG.info("Stopped watching '%s'", path)
         except KeyError:
             LOG.error("Failed to exit watch gracefully")