Replace watchdog with pyinotify

Author: Jonathan Visser

nginx_config_reloader/__init__.py

@@ -13,8 +13,10 @@
 import threading
 import time
 from typing import Optional
+from pathlib import Path
 
-import pyinotify
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler
 from dasbus.loop import EventLoop
 from dasbus.signal import Signal
 
@@ -42,8 +44,8 @@
 dbus_loop: Optional[EventLoop] = None
 
 
-class NginxConfigReloader(pyinotify.ProcessEvent):
-    def my_init(
+class NginxConfigReloader(FileSystemEventHandler):
+    def __init__(
         self,
         logger=None,
         no_magento_config=False,
@@ -79,36 +81,51 @@ def my_init(
         self.applying = False
         self._on_config_reload = Signal()
 
-    def process_IN_DELETE(self, event):
+    def on_deleted(self, event):
         """Triggered by inotify on removal of file or removal of dir
 
         If the dir itself is removed, inotify will stop watching and also
         trigger IN_IGNORED.
         """
-        if not event.dir:  # Will also capture IN_DELETE_SELF
+        if not event.is_directory:
             self.handle_event(event)
 
-    def process_IN_MOVED(self, event):
+    def on_moved(self, event):
         """Triggered by inotify when a file is moved from or to the dir"""
         self.handle_event(event)
 
-    def process_IN_CREATE(self, event):
+    def on_created(self, event):
         """Triggered by inotify when a dir is created in the watch dir"""
-        if event.dir:
+        if event.is_directory:
             self.handle_event(event)
 
-    def process_IN_CLOSE_WRITE(self, event):
+    def on_modified(self, event):
         """Triggered by inotify when a file is written in the dir"""
         self.handle_event(event)
 
-    def process_IN_MOVE_SELF(self, event):
-        """Triggered by inotify when watched dir is moved"""
-        raise ListenTargetTerminated
+    def on_any_event(self, event):
+        """Triggered by inotify when watched dir is moved or deleted"""
+        if event.is_directory and event.event_type in ['moved', 'deleted']:
+            self.logger.warning(f"Directory {event.src_path} has been {event.event_type}.")
+            raise ListenTargetTerminated
 
     def handle_event(self, event):
-        if not any(fnmatch.fnmatch(event.name, pat) for pat in WATCH_IGNORE_FILES):
-            self.logger.info("{} detected on {}.".format(event.maskname, event.name))
+        file_path = Path(event.src_path)
+        if (
+            file_path.name.endswith(".swx")
+            or file_path.name.endswith(".swp")
+            or file_path.name.endswith("~")
+        ):
+            return
+        
+        if (event.is_directory):
+            return
+        
+        basename = os.path.basename(event.src_path)
+        if not any(fnmatch.fnmatch(basename, pat) for pat in WATCH_IGNORE_FILES):
+            self.logger.debug(f"{event.event_type.upper()} detected on {event.src_path}")
             self.dirty = True
+            # Additional handling if necessary
 
     def install_magento_config(self):
         # Check if configs are present
@@ -309,6 +326,20 @@ def reload(self, send_signal=True):
         if send_signal:
             self._on_config_reload.emit()
 
+    def start_observer(self):
+        self.observer = Observer()
+        self.observer.schedule(
+            self,
+            self.dir_to_watch,
+            recursive=True,
+            follow_symlink=True
+        )
+        self.observer.start()
+
+    def stop_observer(self):
+        self.observer.stop()
+        self.observer.join()
+        sys.exit()
 
 class ListenTargetTerminated(BaseException):
     pass
@@ -357,20 +388,11 @@ def wait_loop(
     """
     dir_to_watch = os.path.abspath(dir_to_watch)
 
-    wm = pyinotify.WatchManager()
-    notifier = pyinotify.Notifier(wm)
-
-    class SymlinkChangedHandler(pyinotify.ProcessEvent):
-        def process_IN_DELETE(self, event):
-            if event.pathname == dir_to_watch:
-                raise ListenTargetTerminated("watched directory was deleted")
-
     nginx_config_changed_handler = NginxConfigReloader(
         logger=logger,
         no_magento_config=no_magento_config,
         no_custom_config=no_custom_config,
         dir_to_watch=dir_to_watch,
-        notifier=notifier,
         use_systemd=use_systemd,
     )
 
@@ -383,35 +405,27 @@ def process_IN_DELETE(self, event):
         dbus_thread = threading.Thread(target=dbus_event_loop)
         dbus_thread.start()
 
-    while True:
-        while not os.path.exists(dir_to_watch):
-            logger.warning(
-                "Configuration dir {} not found, waiting...".format(dir_to_watch)
-            )
-            time.sleep(5)
-
-        wm.add_watch(
-            dir_to_watch,
-            pyinotify.ALL_EVENTS,
-            nginx_config_changed_handler,
-            rec=recursive_watch,
-            auto_add=True,
-        )
-        wm.watch_transient_file(
-            dir_to_watch, pyinotify.ALL_EVENTS, SymlinkChangedHandler
+    while not os.path.exists(dir_to_watch):
+        logger.warning(
+            "Configuration dir {} not found, waiting...".format(dir_to_watch)
         )
-
-        # Install initial configuration
-        nginx_config_changed_handler.reload(send_signal=False)
-
-        try:
-            logger.info("Listening for changes to {}".format(dir_to_watch))
-            notifier.coalesce_events()
-            notifier.loop(callback=lambda _: after_loop(nginx_config_changed_handler))
-        except pyinotify.NotifierError as err:
-            logger.critical(err)
-        except ListenTargetTerminated:
-            logger.warning("Configuration dir lost, waiting for it to reappear")
+        time.sleep(5)
+
+
+    try:
+        logger.info(f"Listening for changes to {dir_to_watch}")
+        nginx_config_changed_handler.start_observer()
+        while True:
+            time.sleep(1)
+            after_loop(nginx_config_changed_handler)
+    except ListenTargetTerminated:
+        logger.warning("Configuration dir lost, waiting for it to reappear")
+        nginx_config_changed_handler.stop_observer()
+        time.sleep(5)
+    except KeyboardInterrupt:
+        logger.info("Shutting down observer.")
+        nginx_config_changed_handler.stop_observer()
+        
 
 
 def as_unprivileged_user():

requirements.txt

@@ -1,4 +1,5 @@
-pyinotify==0.9.6
+# Follow_symlinks isnt released in 6.0.0 yet, so we use the latest current commit
+watchdog @ git+https://github.com/gorakhargosh/watchdog.git@f3e78cd4d9500d287bd11ec5d08a1f351601028d
 
 mock==5.0.1
 pytest==7.2.1
@@ -8,4 +9,4 @@ black==23.1.0
 pre-commit==2.21.0
 pygobject
 pygobject-stubs
-dasbus==1.7
+dasbus==1.7

tests/test_nginx_config_reloader.py

@@ -679,4 +679,6 @@ def _dest(self, name):
 class Event:
     def __init__(self, name):
         self.name = name
-        self.maskname = "IN_CLOSE_WRITE"
+        self.event_type = "modified"
+        self.src_path = name
+        self.is_directory = False

tests/test_inotify_callbacks.py tests/test_watchdog_callbacks.pytests/test_inotify_callbacks.py renamed to tests/test_watchdog_callbacks.py

@@ -4,12 +4,20 @@
 from tempfile import NamedTemporaryFile, mkdtemp
 
 import mock
-import pyinotify
+from watchdog.events import (
+    DirCreatedEvent,
+    DirDeletedEvent,
+    DirMovedEvent,
+    FileCreatedEvent,
+    FileDeletedEvent,
+    FileModifiedEvent,
+    FileMovedEvent,
+)
 
 import nginx_config_reloader
 
 
-class TestInotifyCallbacks(unittest.TestCase):
+class TestWatchdogCallbacks(unittest.TestCase):
     def setUp(self):
         patcher = mock.patch("nginx_config_reloader.NginxConfigReloader.handle_event")
         self.addCleanup(patcher.stop)
@@ -19,87 +27,58 @@ def setUp(self):
         with open(os.path.join(self.dir, "existing_file"), "w") as f:
             f.write("blablabla")
 
-        wm = pyinotify.WatchManager()
-        handler = nginx_config_reloader.NginxConfigReloader()
-        self.notifier = pyinotify.Notifier(wm, default_proc_fun=handler)
-        wm.add_watch(self.dir, pyinotify.ALL_EVENTS)
+        self.observer = mock.Mock()
+        self.handler = nginx_config_reloader.NginxConfigReloader(dir_to_watch=self.dir)
+        self.handler.observer = self.observer
 
     def tearDown(self):
-        self.notifier.stop()
         shutil.rmtree(self.dir, ignore_errors=True)
 
-    def _process_events(self):
-        while self.notifier.check_events(0):
-            self.notifier.read_events()
-            self.notifier.process_events()
-
-    def test_that_handle_event_is_called_when_new_file_is_created(self):
-        with open(os.path.join(self.dir, "testfile"), "w") as f:
-            f.write("blablabla")
-
-        self._process_events()
-
         self.assertEqual(len(self.handle_event.mock_calls), 1)
 
     def test_that_handle_event_is_called_when_new_dir_is_created(self):
-        mkdtemp(dir=self.dir)
-        self._process_events()
+        event = DirCreatedEvent(os.path.join(self.dir, "testdir"))
+        self.handler.on_created(event)
 
         self.assertEqual(len(self.handle_event.mock_calls), 1)
 
     def test_that_handle_event_is_called_when_a_file_is_removed(self):
-        os.remove(os.path.join(self.dir, "existing_file"))
-
-        self._process_events()
+        event = FileDeletedEvent(os.path.join(self.dir, "existing_file"))
+        self.handler.on_deleted(event)
 
         self.assertEqual(len(self.handle_event.mock_calls), 1)
 
     def test_that_handle_event_is_called_when_a_file_is_moved_in(self):
         with NamedTemporaryFile(delete=False) as f:
-            os.rename(f.name, os.path.join(self.dir, "newfile"))
-
-            self._process_events()
+            event = FileMovedEvent(
+                f.name, os.path.join(self.dir, "newfile")
+            )
+            self.handler.on_moved(event)
 
             self.assertEqual(len(self.handle_event.mock_calls), 1)
 
     def test_that_handle_event_is_called_when_a_file_is_moved_out(self):
         destdir = mkdtemp()
-        os.rename(
+        event = FileMovedEvent(
             os.path.join(self.dir, "existing_file"),
             os.path.join(destdir, "existing_file"),
         )
-
-        self._process_events()
+        self.handler.on_moved(event)
 
         self.assertEqual(len(self.handle_event.mock_calls), 1)
 
         shutil.rmtree(destdir)
 
     def test_that_handle_event_is_called_when_a_file_is_renamed(self):
-        os.rename(
-            os.path.join(self.dir, "existing_file"), os.path.join(self.dir, "new_name")
+        event = FileMovedEvent(
+            os.path.join(self.dir, "existing_file"),
+            os.path.join(self.dir, "new_name"),
         )
-
-        self._process_events()
+        self.handler.on_moved(event)
 
         self.assertGreaterEqual(len(self.handle_event.mock_calls), 1)
 
-    def test_that_listen_target_terminated_is_raised_if_dir_is_renamed(self):
-        destdir = mkdtemp()
-        os.rename(self.dir, destdir)
-
-        with self.assertRaises(nginx_config_reloader.ListenTargetTerminated):
-            self._process_events()
-
-        shutil.rmtree(destdir)
-
-    def test_that_listen_target_terminated_is_not_raised_if_dir_is_removed(self):
-        shutil.rmtree(self.dir)
-
-        self._process_events()
-
-
-class TestInotifyRecursiveCallbacks(TestInotifyCallbacks):
+class TestWatchdogRecursiveCallbacks(TestWatchdogCallbacks):
     # Run all callback tests on a subdir
     def setUp(self):
         patcher = mock.patch("nginx_config_reloader.NginxConfigReloader.handle_event")
@@ -111,11 +90,9 @@ def setUp(self):
         with open(os.path.join(self.dir, "existing_file"), "w") as f:
             f.write("blablabla")
 
-        wm = pyinotify.WatchManager()
-        handler = nginx_config_reloader.NginxConfigReloader()
-        self.notifier = pyinotify.Notifier(wm, default_proc_fun=handler)
-        wm.add_watch(self.rootdir, pyinotify.ALL_EVENTS, rec=True)
+        self.observer = mock.Mock()
+        self.handler = nginx_config_reloader.NginxConfigReloader(dir_to_watch=self.dir)
+        self.handler.observer = self.observer
 
     def tearDown(self):
-        self.notifier.stop()
-        shutil.rmtree(self.rootdir, ignore_errors=True)
+        shutil.rmtree(self.rootdir, ignore_errors=True)