diff --git a/lib/galaxy/app/__init__.py b/lib/galaxy/app/__init__.py index 28ed51e4c3a..c61bb46eabd 100644 --- a/lib/galaxy/app/__init__.py +++ b/lib/galaxy/app/__init__.py @@ -1011,12 +1011,17 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication, InstallationT self.database_heartbeat.add_change_callback(self.watchers.change_state) self.application_stack.register_postfork_function(self.database_heartbeat.start) - # History audit monitor for SSE-based history updates + # History audit monitor for SSE-based history updates. The monitor only + # runs on the single process elected via DatabaseHeartbeat's + # is_history_audit_monitor role — a standalone ``galaxy-sse-monitor`` + # daemon wins that election when present, otherwise one webapp picks it + # up. start/stop are driven by heartbeat role transitions rather than + # postfork, so the monitor cleanly migrates when the leader dies. if self.config.enable_sse_history_updates: from galaxy.managers.history_audit_monitor import HistoryAuditMonitor self._history_audit_monitor = self._register_singleton(HistoryAuditMonitor) - self.application_stack.register_postfork_function(self._history_audit_monitor.start) + self.database_heartbeat.add_audit_monitor_change_callback(self._history_audit_monitor.on_role_change) # Start web stack message handling self.application_stack.register_postfork_function(self.application_stack.start) diff --git a/lib/galaxy/managers/history_audit_monitor.py b/lib/galaxy/managers/history_audit_monitor.py index c3291dd2e84..eca33f4aa1f 100644 --- a/lib/galaxy/managers/history_audit_monitor.py +++ b/lib/galaxy/managers/history_audit_monitor.py @@ -130,6 +130,7 @@ class HistoryAuditMonitor: if self._active: return self._active = True + self._exit.clear() # allow restart after a previous shutdown target = self._listen_postgres if self._is_postgres else self._poll_audit_table self._thread = threading.Thread( target=target, @@ -144,10 +145,21 @@ class HistoryAuditMonitor: ) def shutdown(self) -> None: + if not self._active: + return self._active = False self._exit.set() if self._thread: self._thread.join(timeout=5) + self._thread = None + log.info("HistoryAuditMonitor stopped") + + def on_role_change(self, is_leader: bool) -> None: + """Heartbeat callback: start/stop the monitor as this process's election state changes.""" + if is_leader: + self.start() + else: + self.shutdown() # --- PostgreSQL LISTEN/NOTIFY mode --- diff --git a/lib/galaxy/model/database_heartbeat.py b/lib/galaxy/model/database_heartbeat.py index c16a8c64de0..143a94f1f11 100644 --- a/lib/galaxy/model/database_heartbeat.py +++ b/lib/galaxy/model/database_heartbeat.py @@ -17,6 +17,8 @@ from galaxy.model.orm.now import now log = logging.getLogger(__name__) WEBAPP = "webapp" # WorkerProcess.app_type for web apps. +SSE_MONITOR = "sse_monitor" # WorkerProcess.app_type for the standalone SSE monitor process. +SSE_MONITOR_SERVER_PREFIX = "sse_monitor." # server_name prefix used by the standalone SSE monitor process. class DatabaseHeartbeat: @@ -27,7 +29,9 @@ class DatabaseHeartbeat: self.hostname = socket.gethostname() self._engine = application_stack.app.model.engine self._is_config_watcher = False + self._is_history_audit_monitor = False self._observers = [] + self._audit_monitor_observers = [] self.exit = threading.Event() self.thread = None self.active = False @@ -72,6 +76,9 @@ class DatabaseHeartbeat: def add_change_callback(self, callback): self._observers.append(callback) + def add_audit_monitor_change_callback(self, callback): + self._audit_monitor_observers.append(callback) + @property def is_config_watcher(self): return self._is_config_watcher @@ -83,6 +90,28 @@ class DatabaseHeartbeat: for callback in self._observers: callback(self._is_config_watcher) + @property + def is_history_audit_monitor(self): + return self._is_history_audit_monitor + + @is_history_audit_monitor.setter + def is_history_audit_monitor(self, value): + self._is_history_audit_monitor = value + log.debug( + "%s %s history audit monitor", + self.server_name, + "is" if self._is_history_audit_monitor else "is not", + ) + for callback in self._audit_monitor_observers: + callback(self._is_history_audit_monitor) + + def _app_type(self): + if self.application_stack.app.is_webapp: + return WEBAPP + if self.server_name.startswith(SSE_MONITOR_SERVER_PREFIX): + return SSE_MONITOR + return None + def update_watcher_designation(self): expression = self._worker_process_identifying_clause() stmt = select(WorkerProcess).with_for_update(of=WorkerProcess).where(expression) @@ -90,19 +119,37 @@ class DatabaseHeartbeat: worker_process = session.scalars(stmt).first() if not worker_process: worker_process = WorkerProcess(server_name=self.server_name, hostname=self.hostname) - if self.application_stack.app.is_webapp: - worker_process.app_type = WEBAPP + app_type = self._app_type() + if app_type is not None: + worker_process.app_type = app_type worker_process.update_time = now() worker_process.pid = self.pid session.add(worker_process) + active = list(self.get_active_processes(self.heartbeat_interval + 1)) # We only want a single process watching the various config files on the file system. # We just pick the max server name for simplicity - webapp_servers = [ - p.server_name for p in self.get_active_processes(self.heartbeat_interval + 1) if p.app_type == WEBAPP - ] + webapp_servers = [p.server_name for p in active if p.app_type == WEBAPP] is_config_watcher = bool(webapp_servers) and self.server_name == max(webapp_servers) if is_config_watcher != self.is_config_watcher: self.is_config_watcher = is_config_watcher + # The history-audit monitor is a single elected process too, but preference + # goes to a standalone sse_monitor daemon when one is running so the + # monitor's postgres LISTEN isn't blocked by webapp GIL pauses. If no + # dedicated process is registered we fall back to a webapp (same + # max-server_name tiebreaker as config_watcher). + audit_leader = self._elect_audit_leader(active, webapp_servers) + is_history_audit_monitor = audit_leader is not None and self.server_name == audit_leader + if is_history_audit_monitor != self.is_history_audit_monitor: + self.is_history_audit_monitor = is_history_audit_monitor + + @staticmethod + def _elect_audit_leader(active, webapp_servers): + monitor_servers = [p.server_name for p in active if p.app_type == SSE_MONITOR] + if monitor_servers: + return min(monitor_servers) + if webapp_servers: + return max(webapp_servers) + return None def send_database_heartbeat(self): if self.active: diff --git a/lib/galaxy/sse_monitor/__init__.py b/lib/galaxy/sse_monitor/__init__.py new file mode 100644 index 00000000000..740f5574324 --- /dev/null +++ b/lib/galaxy/sse_monitor/__init__.py @@ -0,0 +1,30 @@ +"""Standalone SSE monitor process. + +Runs the :class:`~galaxy.managers.history_audit_monitor.HistoryAuditMonitor` +(and, in principle, any future SSE event producer) in its own OS process so a +webapp's GIL pauses or fork-lifecycle events can never block history-update +dispatch. Webapps still consume SSE events normally; only the *producer* moves. + +The process registers itself via :class:`~galaxy.model.database_heartbeat.DatabaseHeartbeat` +with a server_name beginning with ``sse_monitor.``. The heartbeat's +``is_history_audit_monitor`` election prefers any such process, so the monitor +automatically migrates here when this daemon is running. If it stops, a webapp +is re-elected within one heartbeat interval (~60s). + +Starting the daemon +------------------- + +Installed:: + + GALAXY_CONFIG_FILE=/etc/galaxy/galaxy.yml galaxy-sse-monitor + +From a source checkout:: + + GALAXY_CONFIG_FILE=config/galaxy.yml python -m galaxy.sse_monitor + +For production deployments, run under systemd / supervisord alongside the +webapp. A native ``gravity`` (``galaxyctl``) service type for this daemon is +tracked as a follow-up against the ``gravity`` package — until that lands, +either start the process via your process manager of choice, or omit it and +let a webapp fall back to the monitor role. +""" diff --git a/lib/galaxy/sse_monitor/__main__.py b/lib/galaxy/sse_monitor/__main__.py new file mode 100644 index 00000000000..e8ce22d62a2 --- /dev/null +++ b/lib/galaxy/sse_monitor/__main__.py @@ -0,0 +1,101 @@ +"""Entry point for the standalone ``galaxy-sse-monitor`` daemon. + +Loads a minimal :class:`GalaxyManagerApplication` (same shape the Celery +workers use), forces the server_name to ``sse_monitor..`` so +DatabaseHeartbeat's election picks this process as the history-audit monitor, +and blocks on SIGINT/SIGTERM. +""" + +import logging +import os +import signal +import socket +import sys +import threading + +from galaxy.celery import get_app_properties +from galaxy.model.database_heartbeat import DatabaseHeartbeat + +log = logging.getLogger("galaxy.sse_monitor") + + +def _build_server_name() -> str: + return f"sse_monitor.{socket.gethostname()}.{os.getpid()}" + + +def _build_app(server_name: str): + kwargs = get_app_properties() or {} + if not kwargs: + raise RuntimeError( + "GALAXY_CONFIG_FILE (or GALAXY_ROOT_DIR with an on-disk Galaxy config) is required " + "to start galaxy-sse-monitor" + ) + kwargs = dict(kwargs) + kwargs["check_migrate_databases"] = False + kwargs["use_display_applications"] = False + kwargs["use_converters"] = False + kwargs["server_name"] = server_name + + import galaxy.app + + return galaxy.app.GalaxyManagerApplication(configure_logging=True, **kwargs) + + +def main() -> int: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") + + server_name = _build_server_name() + log.info("Starting galaxy-sse-monitor as %s", server_name) + + app = _build_app(server_name) + + # GalaxyManagerApplication doesn't wire a DatabaseHeartbeat (that lives on + # UniverseApplication, which pulls in the webapp stack we don't need). We + # spin up our own and register the audit-monitor callback so election + # transitions start/stop the producer cleanly. + heartbeat = DatabaseHeartbeat(application_stack=app.application_stack) + + monitor = None + if app.config.enable_sse_history_updates: + from galaxy.managers.history_audit_monitor import HistoryAuditMonitor + + monitor = app[HistoryAuditMonitor] + heartbeat.add_audit_monitor_change_callback(monitor.on_role_change) + else: + log.warning( + "enable_sse_history_updates is False — galaxy-sse-monitor will idle with no producers" + ) + + heartbeat.start() + + shutdown = threading.Event() + + def _handle_signal(signum, _frame): + log.info("Received signal %s, shutting down galaxy-sse-monitor", signum) + shutdown.set() + + signal.signal(signal.SIGINT, _handle_signal) + signal.signal(signal.SIGTERM, _handle_signal) + + try: + shutdown.wait() + finally: + if monitor is not None: + try: + monitor.shutdown() + except Exception: + log.exception("Error shutting down HistoryAuditMonitor") + try: + heartbeat.shutdown() + except Exception: + log.exception("Error shutting down database heartbeat") + try: + app.shutdown() + except Exception: + log.exception("Error shutting down GalaxyManagerApplication") + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/packages/app/setup.cfg b/packages/app/setup.cfg index e1001c06c6a..c455dc2cec4 100644 --- a/packages/app/setup.cfg +++ b/packages/app/setup.cfg @@ -101,6 +101,7 @@ test = console_scripts = galaxy-main = galaxy.main:main galaxy-dependencies = galaxy.dependencies.script:main + galaxy-sse-monitor = galaxy.sse_monitor.__main__:main [options.packages.find] where = src diff --git a/packages/app/src/galaxy/sse_monitor b/packages/app/src/galaxy/sse_monitor new file mode 120000 index 00000000000..e714ba834b2 --- /dev/null +++ b/packages/app/src/galaxy/sse_monitor @@ -0,0 +1 @@ +../../../../lib/galaxy/sse_monitor \ No newline at end of file