Wait for dbus and FEATURE subscriptions

src/system-health/health_checker/sysmonitor.py

@@ -24,22 +24,27 @@
 SELECT_TIMEOUT_MSECS = 1000
 QUEUE_TIMEOUT = 15
 TASK_STOP_TIMEOUT = 10
+# Combined wall-clock budget for both monitor threads to signal readiness before initial service scan
+SUBSCRIPTION_READY_TIMEOUT_SEC = 60
 logger = Logger(log_identifier=SYSLOG_IDENTIFIER)
 exclude_srv_list = ['ztp.service']
 
 #Thread which subscribes to STATE_DB FEATURE table for any update
 #and push service events to main thread via queue
 class MonitorStateDbTask(ThreadTaskBase):
 
-    def __init__(self,myQ):
+    def __init__(self, myQ, subscription_ready=None):
         ThreadTaskBase.__init__(self)
         self.task_queue = myQ
+        self._subscription_ready = subscription_ready
 
     def subscribe_statedb(self):
         state_db = swsscommon.DBConnector("STATE_DB", REDIS_TIMEOUT_MS, False)
         sel = swsscommon.Select()
         cst = swsscommon.SubscriberStateTable(state_db, "FEATURE")
         sel.addSelectable(cst)
+        if self._subscription_ready is not None:
+            self._subscription_ready.set()
 
         while not self.task_stopping_event.is_set():
             (state, c) = sel.select(SELECT_TIMEOUT_MSECS)
@@ -73,9 +78,10 @@ def task_notify(self, msg):
 #and push service events to main thread via queue
 class MonitorSystemBusTask(ThreadTaskBase):
 
-    def __init__(self,myQ):
+    def __init__(self, myQ, subscription_ready=None):
         ThreadTaskBase.__init__(self)
         self.task_queue = myQ
+        self._subscription_ready = subscription_ready
 
     def on_job_removed(self, id, job, unit, result):
         if result == "done" or result == "failed":
@@ -96,6 +102,8 @@ def subscribe_sysbus(self):
         manager = dbus.Interface(systemd, 'org.freedesktop.systemd1.Manager')
         manager.Subscribe()
         manager.connect_to_signal('JobRemoved', self.on_job_removed)
+        if self._subscription_ready is not None:
+            self._subscription_ready.set()
 
         loop = GLib.MainLoop()
         loop.run()
@@ -471,23 +479,53 @@ def check_unit_status(self, event):
 
         return 0
 
+    def _wait_for_monitor_subscriptions(self, dbus_ready, statedb_ready,
+                                        monitor_system_bus=None, monitor_statedb_table=None):
+        """Block until both monitor threads signal listener registration.
+
+        Single monotonic deadline: total wall time is at most SUBSCRIPTION_READY_TIMEOUT_SEC
+        (both threads already run in parallel).
+
+        On timeout, stops monitor tasks before exiting so non-daemon threads do not stall
+        interpreter shutdown.
+        """
+        deadline = time.monotonic() + SUBSCRIPTION_READY_TIMEOUT_SEC
+        monitors = (
+            (dbus_ready, "systemd dbus monitor did not finish Subscribe"),
+            (statedb_ready, "STATE_DB FEATURE monitor did not finish subscribing"),
+        )
+        for ev, detail in monitors:
+            remaining = deadline - time.monotonic()
+            if not ev.wait(timeout=max(0.0, remaining)):
+                logger.log_error(
+                    "Timeout: {} within {}s (combined budget)".format(detail, SUBSCRIPTION_READY_TIMEOUT_SEC))
+                if monitor_system_bus is not None:
+                    monitor_system_bus.task_stop()
+                if monitor_statedb_table is not None:
+                    monitor_statedb_table.task_stop()
+                sys.exit(1)
+
     def system_service(self):
         if not self.state_db:
             self.state_db = swsscommon.SonicV2Connector(use_unix_socket_path=True)
             self.state_db.connect(self.state_db.STATE_DB)
 
+        dbus_ready = threading.Event()
+        statedb_ready = threading.Event()
+
         try:
-            monitor_system_bus = MonitorSystemBusTask(self.myQ)
+            monitor_system_bus = MonitorSystemBusTask(self.myQ, subscription_ready=dbus_ready)
             monitor_system_bus.task_run()
 
-            monitor_statedb_table = MonitorStateDbTask(self.myQ)
+            monitor_statedb_table = MonitorStateDbTask(self.myQ, subscription_ready=statedb_ready)
             monitor_statedb_table.task_run()
 
         except Exception as e:
-            logger.log_error("SubProcess-{}".format(str(e)))
+            logger.log_error("monitor threads-{}".format(str(e)))
             sys.exit(1)
 
-
+        self._wait_for_monitor_subscriptions(
+            dbus_ready, statedb_ready, monitor_system_bus, monitor_statedb_table)
         self.update_system_status()
 
         from queue import Empty

src/system-health/tests/test_system_health.py

@@ -41,6 +41,7 @@
 from health_checker.sysmonitor import Sysmonitor
 from health_checker.sysmonitor import MonitorStateDbTask
 from health_checker.sysmonitor import MonitorSystemBusTask
+from health_checker import sysmonitor as sysmonitor_module
 
 def load_source(modname, filename):
     loader = importlib.machinery.SourceFileLoader(modname, filename)
@@ -1148,6 +1149,7 @@ def test_monitor_sysbus_task():
     assert sysmon._task_thread is not None
     sysmon.task_stop()
 
+@patch('health_checker.sysmonitor.Sysmonitor._wait_for_monitor_subscriptions', MagicMock())
 @patch('health_checker.sysmonitor.MonitorSystemBusTask.subscribe_sysbus', MagicMock())
 @patch('health_checker.sysmonitor.MonitorStateDbTask.subscribe_statedb', MagicMock())
 def test_system_service():
@@ -1157,6 +1159,73 @@ def test_system_service():
     sysmon.task_stop()
 
 
+def test_wait_for_monitor_subscriptions_completes_when_both_events_signaled():
+    """_wait_for_monitor_subscriptions returns once dbus and STATE_DB listeners have signaled ready."""
+    sysmon = Sysmonitor()
+    dbus_ready = threading.Event()
+    statedb_ready = threading.Event()
+    dbus_ready.set()
+    statedb_ready.set()
+    sysmon._wait_for_monitor_subscriptions(dbus_ready, statedb_ready)
+
+
+@patch.object(sysmonitor_module, 'SUBSCRIPTION_READY_TIMEOUT_SEC', 0.05)
+def test_wait_for_monitor_subscriptions_system_exit_when_dbus_not_ready():
+    sysmon = Sysmonitor()
+    dbus_ready = threading.Event()
+    statedb_ready = threading.Event()
+    statedb_ready.set()
+    try:
+        sysmon._wait_for_monitor_subscriptions(dbus_ready, statedb_ready)
+    except SystemExit as exc:
+        assert exc.code == 1
+    else:
+        assert False, 'expected SystemExit when dbus ready event is never set'
+
+
+@patch.object(sysmonitor_module, 'SUBSCRIPTION_READY_TIMEOUT_SEC', 0.05)
+def test_wait_for_monitor_subscriptions_system_exit_when_statedb_not_ready():
+    sysmon = Sysmonitor()
+    dbus_ready = threading.Event()
+    statedb_ready = threading.Event()
+    dbus_ready.set()
+    try:
+        sysmon._wait_for_monitor_subscriptions(dbus_ready, statedb_ready)
+    except SystemExit as exc:
+        assert exc.code == 1
+    else:
+        assert False, 'expected SystemExit when STATE_DB ready event is never set'
+
+
+def test_monitor_statedb_subscribe_sets_subscription_ready_event():
+    """FEATURE subscriber sets subscription_ready after SubscriberStateTable is registered."""
+    ready = threading.Event()
+    task = MonitorStateDbTask(myQ, subscription_ready=ready)
+    task.task_stopping_event.set()
+    with patch('health_checker.sysmonitor.swsscommon.DBConnector', MagicMock()), \
+            patch('health_checker.sysmonitor.swsscommon.SubscriberStateTable', MagicMock(return_value=MagicMock())), \
+            patch('health_checker.sysmonitor.swsscommon.Select', MagicMock(return_value=MagicMock())):
+        task.subscribe_statedb()
+    assert ready.is_set()
+
+
+def test_monitor_system_bus_subscribe_sets_subscription_ready_event():
+    """systemd Manager Subscribe + JobRemoved hook registers before MainLoop.run(); ready is set."""
+    ready = threading.Event()
+    task = MonitorSystemBusTask(myQ, subscription_ready=ready)
+    with patch('dbus.mainloop.glib.DBusGMainLoop', MagicMock()), \
+            patch('dbus.SystemBus', MagicMock()), \
+            patch('dbus.Interface') as mock_interface, \
+            patch('gi.repository.GLib.MainLoop') as mock_main_loop:
+        manager = MagicMock()
+        mock_interface.return_value = manager
+        mock_main_loop.return_value.run = MagicMock()
+        task.subscribe_sysbus()
+    manager.Subscribe.assert_called_once()
+    manager.connect_to_signal.assert_called_once_with('JobRemoved', task.on_job_removed)
+    assert ready.is_set()
+
+
 @patch('sonic_py_common.device_info.get_device_runtime_metadata', MagicMock(return_value=device_runtime_metadata))
 def test_get_service_from_feature_table():
     sysmon = Sysmonitor()