Replace NATS with dbus

It's way simpler to do the signaling using dbus instead of NATS. It's an open, local only solution with authorization built-in.

.github/workflows/test.yaml

@@ -17,6 +17,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
       - name: Install dependencies
         run: |
+          apt install libgirepository1.0-dev -y
           pip install -r requirements.txt
           pip install tox-gh-actions
       - name: Test with tox

debian/control

@@ -8,6 +8,6 @@ X-Python3-Version: >= 3.5
 
 Package: nginx-config-reloader
 Architecture: all
-Depends: ${python3:Depends}, ${misc:Depends}, python3-pyinotify
+Depends: ${python3:Depends}, ${misc:Depends}, python3-pyinotify, python3-dasbus
 Description: nginx config files reloader
  Daemon that installs and reloads nginx configuration

debian/install

@@ -0,0 +1 @@
+debian/usr/share/dbus-1/system.d/nginx-config-reloader-bus.conf /usr/share/dbus-1/system.d

debian/usr/share/dbus-1/system.d/nginx-config-reloader-bus.conf

@@ -0,0 +1,20 @@
+<!DOCTYPE busconfig PUBLIC
+          "-//freedesktop//DTD D-BUS Bus Configuration 1.0//EN"
+          "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
+<busconfig>
+    <policy user="root">
+        <allow own="com.hypernode.NginxConfigReloader"/>
+    </policy>
+    <policy user="root">
+        <allow send_destination="com.hypernode.NginxConfigReloader"/>
+        <allow receive_sender="com.hypernode.NginxConfigReloader"/>
+    </policy>
+    <policy group="adm">
+        <allow send_destination="com.hypernode.NginxConfigReloader"/>
+        <allow receive_sender="com.hypernode.NginxConfigReloader"/>
+    </policy>
+    <policy user="app">
+        <allow send_destination="com.hypernode.NginxConfigReloader"/>
+        <allow receive_sender="com.hypernode.NginxConfigReloader"/>
+    </policy>
+</busconfig>

nginx_config_reloader/__init__.py

@@ -12,13 +12,15 @@
 import sys
 import threading
 import time
-from typing import Callable, Optional
+from typing import Optional
 
 import pyinotify
-from pynats import NATSClient, NATSMessage
+from dasbus.loop import EventLoop
+from dasbus.signal import Signal
 
 from nginx_config_reloader.copy_files import safe_copy_files
-from nginx_config_reloader.nats import initialize_nats, publish_nats_message
+from nginx_config_reloader.dbus.common import NGINX_CONFIG_RELOADER, SYSTEM_BUS
+from nginx_config_reloader.dbus.server import NginxConfigReloaderInterface
 from nginx_config_reloader.settings import (
     BACKUP_CONFIG_DIR,
     CUSTOM_CONFIG_DIR,
@@ -28,8 +30,6 @@
     MAGENTO1_CONF,
     MAGENTO2_CONF,
     MAGENTO_CONF,
-    NATS_RELOAD_BODY,
-    NATS_SUBJECT,
     NGINX,
     NGINX_PID_FILE,
     UNPRIVILEGED_GID,
@@ -38,6 +38,7 @@
 )
 
 logger = logging.getLogger(__name__)
+dbus_loop: Optional[EventLoop] = None
 
 
 class NginxConfigReloader(pyinotify.ProcessEvent):
@@ -50,11 +51,10 @@ def my_init(
         magento2_flag=None,
         notifier=None,
         use_systemd=False,
-        nats_client: Optional[NATSClient] = None,
     ):
         """Constructor called by ProcessEvent
 
-        :param obj logger: The logger object
+        :param logging.Logger logger: The logger object
         :param bool no_magento_config: True if we should not install Magento configuration
         :param bool no_custom_config: True if we should not copy custom configuration
         :param str dir_to_watch: The directory to watch
@@ -76,7 +76,7 @@ def my_init(
         self.use_systemd = use_systemd
         self.dirty = False
         self.applying = False
-        self.nats_client = nats_client
+        self._on_config_reload = Signal()
 
     def process_IN_DELETE(self, event):
         """Triggered by inotify on removal of file or removal of dir
@@ -107,11 +107,7 @@ def process_IN_MOVE_SELF(self, event):
     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))
-            if self.nats_client:
-                if not self.dirty:
-                    self.nats_client = publish_nats_message(self.nats_client)
-            else:
-                self.dirty = True
+            self.dirty = True
 
     def install_magento_config(self):
         # Check if configs are present
@@ -296,6 +292,16 @@ def write_error_file(self, error):
         with open(os.path.join(self.dir_to_watch, ERROR_FILE), "w") as f:
             f.write(error)
 
+    @property
+    def reloaded(self):
+        """Signal for the reload event."""
+        return self._on_config_reload
+
+    def reload(self, send_signal=True):
+        self.apply_new_config()
+        if send_signal:
+            self._on_config_reload.emit()
+
 
 class ListenTargetTerminated(BaseException):
     pass
@@ -304,56 +310,16 @@ class ListenTargetTerminated(BaseException):
 def after_loop(nginx_config_reloader: NginxConfigReloader) -> None:
     if nginx_config_reloader.dirty:
         try:
-            nginx_config_reloader.apply_new_config()
+            nginx_config_reloader.reload()
         except:
             pass
         nginx_config_reloader.dirty = False
         nginx_config_reloader.applying = False
 
 
-def construct_message_handler(
-    nginx_config_reloader: NginxConfigReloader,
-) -> Callable[[NATSMessage], None]:
-    def message_handler(msg: NATSMessage) -> None:
-        if msg.subject == NATS_SUBJECT and msg.payload == NATS_RELOAD_BODY:
-            logger.debug("NATS message received, reloading config")
-            nginx_config_reloader.dirty = True
-            # Trigger manually to ensure it's running. The `.applying` flag will prevent
-            # concurrent runs.
-            after_loop(nginx_config_reloader)
-
-    return message_handler
-
-
-def start_message_subscribe_loop(
-    nginx_config_reloader: NginxConfigReloader, nats_server: str
-) -> None:
-    def listen_nats() -> None:
-        while True:
-            # Create new connection to throw away any old subscriptions.
-            # This is useful when there are many writes queued up, and we
-            # only want to reload once.
-            try:
-                nc = initialize_nats(nats_server)
-                nginx_config_reloader.nats_client = nc
-                sub = nc.subscribe(
-                    NATS_SUBJECT,
-                    callback=construct_message_handler(nginx_config_reloader),
-                    max_messages=1,
-                )
-                nc.auto_unsubscribe(sub)
-            except Exception as e:
-                logger.debug(f"Couldn't make NATS client: {e}")
-                continue
-
-            logger.debug(f"Waiting for message on {NATS_SUBJECT}")
-            try:
-                nc.wait(count=1)
-            except Exception as e:
-                logger.debug(f"NATS error: {e}")
-
-    t = threading.Thread(target=listen_nats)
-    t.start()
+def dbus_event_loop():
+    dbus_loop = EventLoop()
+    dbus_loop.run()
 
 
 def wait_loop(
@@ -363,7 +329,7 @@ def wait_loop(
     dir_to_watch=DIR_TO_WATCH,
     recursive_watch=False,
     use_systemd=False,
-    nats_server=None,
+    no_dbus=False,
 ):
     """Main event loop
 
@@ -373,11 +339,13 @@ def wait_loop(
     renamed or removed, the inotify-handler raises an exception to break out of the
     inner loop and we're back here in the outer loop.
 
-    :param obj logger: The logger object
+    :param logging.Logger logger: The logger object
     :param bool no_magento_config: True if we should not install Magento configuration
     :param bool no_custom_config: True if we should not copy custom configuration
     :param str dir_to_watch: The directory to watch
     :param bool recursive_watch: True if we should watch the dir recursively
+    :param use_systemd: True if we should reload nginx using systemd instead of process signal
+    :param bool no_dbus: True if we should not use DBus
     :return None:
     """
     dir_to_watch = os.path.abspath(dir_to_watch)
@@ -399,8 +367,14 @@ def process_IN_DELETE(self, event):
         use_systemd=use_systemd,
     )
 
-    if nats_server:
-        start_message_subscribe_loop(nginx_config_changed_handler, nats_server)
+    if not no_dbus:
+        SYSTEM_BUS.publish_object(
+            NGINX_CONFIG_RELOADER.object_path,
+            NginxConfigReloaderInterface(nginx_config_changed_handler),
+        )
+        SYSTEM_BUS.register_service(NGINX_CONFIG_RELOADER.service_name)
+        dbus_thread = threading.Thread(target=dbus_event_loop)
+        dbus_thread.start()
 
     while True:
         while not os.path.exists(dir_to_watch):
@@ -421,7 +395,7 @@ def process_IN_DELETE(self, event):
         )
 
         # Install initial configuration
-        nginx_config_changed_handler.apply_new_config()
+        nginx_config_changed_handler.reload(send_signal=False)
 
         try:
             logger.info("Listening for changes to {}".format(dir_to_watch))
@@ -431,8 +405,6 @@ def process_IN_DELETE(self, event):
             logger.critical(err)
         except ListenTargetTerminated:
             logger.warning("Configuration dir lost, waiting for it to reappear")
-    if nc:
-        nc.close()
 
 
 def as_unprivileged_user():
@@ -476,9 +448,10 @@ def parse_nginx_config_reloader_arguments():
         default=False,
     )
     parser.add_argument(
-        "-s",
-        "--nats-server",
-        help=f"NATS server to connect to. Will publish/subscribe to the topic '{NATS_SUBJECT}'. Will not use this if not set.",
+        "--no-dbus",
+        action="store_true",
+        help="Disable DBus interface",
+        default=False,
     )
     return parser.parse_args()
 
@@ -506,7 +479,7 @@ def main():
             dir_to_watch=args.watchdir,
             recursive_watch=args.recursivewatch,
             use_systemd=args.use_systemd,
-            nats_server=args.nats_server,
+            no_dbus=args.no_dbus,
         )
         # should never return
         return 1

nginx_config_reloader/dbus/common.py

@@ -0,0 +1,9 @@
+from dasbus.connection import SystemMessageBus
+from dasbus.identifier import DBusServiceIdentifier
+
+SYSTEM_BUS = SystemMessageBus()
+
+NGINX_CONFIG_RELOADER = DBusServiceIdentifier(
+    namespace=("com", "hypernode", "NginxConfigReloader"),
+    message_bus=SYSTEM_BUS,
+)

nginx_config_reloader/dbus/server.py

@@ -0,0 +1,21 @@
+from dasbus.server.interface import dbus_interface, dbus_signal
+from dasbus.server.property import emits_properties_changed
+from dasbus.server.template import InterfaceTemplate
+
+from nginx_config_reloader.dbus.common import NGINX_CONFIG_RELOADER
+
+
+@dbus_interface(NGINX_CONFIG_RELOADER.interface_name)
+class NginxConfigReloaderInterface(InterfaceTemplate):
+    def connect_signals(self):
+        self.implementation.reloaded.connect(self.ConfigReloaded)
+
+    @dbus_signal
+    def ConfigReloaded(self):
+        """Signal that the config was reloaded"""
+
+    @emits_properties_changed
+    def Reload(self):
+        """Mark the last reload at current time."""
+        # send_signal=False because we don't want to emit the signal
+        self.implementation.apply_new_config(send_signal=False)

nginx_config_reloader/settings.py

@@ -13,9 +13,6 @@
 NGINX_PID_FILE = "/var/run/nginx.pid"
 ERROR_FILE = "nginx_error_output"
 
-NATS_SUBJECT = "nginx-config-reloader"
-NATS_RELOAD_BODY = b"reload"
-
 WATCH_IGNORE_FILES = (
     # glob patterns
     ".*",

requirements.txt

@@ -6,4 +6,6 @@ pytest-xdist==3.2.0
 tox==4.4.5
 black==23.1.0
 pre-commit==2.21.0
--e git+https://github.com/ByteInternet/nats-python.git@755ce98487ad15bec2889365d8b7caa4b2455e84#egg=nats-python
+pygobject
+pygobject-stubs
+dasbus==1.7

setup.py

@@ -12,6 +12,6 @@
     entry_points={
         "console_scripts": ["nginx_config_reloader = nginx_config_reloader:main"]
     },
-    install_requires=["pyinotify>=0.9.2", "nats-python"],
+    install_requires=["pyinotify>=0.9.2", "dasbus>=1.7"],
     test_suite="tests",
 )

tests/test_dbus_event_loop.py

@@ -0,0 +1,13 @@
+from nginx_config_reloader import dbus_event_loop
+from tests.testcase import TestCase
+
+
+class TestDbusEventLoop(TestCase):
+    def setUp(self):
+        self.event_loop = self.set_up_patch("nginx_config_reloader.EventLoop")
+
+    def test_it_runs_dbus_event_loop(self):
+        dbus_event_loop()
+
+        self.event_loop.assert_called_once_with()
+        self.event_loop.return_value.run.assert_called_once_with()