diff --git a/src/system-health/health_checker/sysmonitor.py b/src/system-health/health_checker/sysmonitor.py index 2828564611f..0e73085ea89 100755 --- a/src/system-health/health_checker/sysmonitor.py +++ b/src/system-health/health_checker/sysmonitor.py @@ -24,6 +24,8 @@ 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'] @@ -31,15 +33,18 @@ #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 diff --git a/src/system-health/tests/test_system_health.py b/src/system-health/tests/test_system_health.py index 83673755c85..b9476d68380 100644 --- a/src/system-health/tests/test_system_health.py +++ b/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()