From f0daa9217e1498efb3d41eefc0bce2215fa85301 Mon Sep 17 00:00:00 2001 From: Morgan Wowk Date: Fri, 30 Oct 2026 16:18:09 -0700 Subject: [PATCH] chore(orchestrator): Replace per-visit sweep logs with a heartbeat The running queue visits every pending or running container execution round-robin, and each visit logged three INFO lines even when nothing changed. With hundreds of active executions that is tens of lines per second carrying no information. Drop the "Before processing", "After processing" and "remains in state" lines. Log one line only when a visit is slow: its queries or its processing take over 1s, matching the existing slow-query threshold for the idle queue log. The queries and processing times stay separate so a slow database is distinguishable from a slow launcher. Add a heartbeat line about once a minute from process_each_queue_once with the number of queued and running executions processed since the last one. The per-visit lines were the only sign that the sweep was still going; the heartbeat keeps that signal at one line a minute, and its absence means the sweep has stopped. State changes, terminations, retries and errors keep their existing logs. --- cloud_pipelines_backend/orchestrator_sql.py | 35 +++++++++--- tests/test_orchestrator_sql.py | 60 +++++++++++++++++++++ 2 files changed, 87 insertions(+), 8 deletions(-) diff --git a/cloud_pipelines_backend/orchestrator_sql.py b/cloud_pipelines_backend/orchestrator_sql.py index 00a0113..72e3f74 100644 --- a/cloud_pipelines_backend/orchestrator_sql.py +++ b/cloud_pipelines_backend/orchestrator_sql.py @@ -33,6 +33,8 @@ _T = typing.TypeVar("_T") +_HEARTBEAT_INTERVAL_SECONDS = 60 + DYNAMIC_DATA_SECRET_KEY = "secret" DYNAMIC_DATA_SECRET_NAME_KEY = "name" @@ -90,6 +92,9 @@ def __init__( _max_container_execution_refresh_error_retries ) self._container_execution_refresh_error_counts: dict[Any, int] = {} + self._heartbeat_started_at = time.monotonic() + self._heartbeat_queued_execution_count = 0 + self._heartbeat_running_container_execution_count = 0 self._max_queue_batch_size = _max_queue_batch_size self._max_queue_batch_duration = _max_queue_batch_duration @@ -136,6 +141,7 @@ def process_each_queue_once(self): > self._max_queue_batch_duration ): break + self._maybe_log_heartbeat() def internal_process_queued_executions_queue(self, session: orm.Session): _logger.debug("Selecting queued execution to process.)") @@ -155,6 +161,7 @@ def internal_process_queued_executions_queue(self, session: orm.Session): query_duration_ms = (time.monotonic_ns() - query_start_timestamp) / 1_000_000 if queued_execution: self._queued_executions_queue_idle = False + self._heartbeat_queued_execution_count += 1 start_timestamp = time.monotonic_ns() with contextual_logging.execution_logging_context(queued_execution): @@ -223,7 +230,7 @@ def internal_process_running_executions_queue(self, session: orm.Session): query_duration_ms = (time.monotonic_ns() - query_start_timestamp) / 1_000_000 if running_container_execution: self._running_executions_queue_idle = False - start_timestamp = time.monotonic_ns() + self._heartbeat_running_container_execution_count += 1 execution_node_ids = list( session.scalars( @@ -241,9 +248,7 @@ def internal_process_running_executions_queue(self, session: orm.Session): queries_duration_ms = ( time.monotonic_ns() - query_start_timestamp ) / 1_000_000 - _logger.info( - f"Before processing running container execution. Queries duration {queries_duration_ms}ms." - ) + start_timestamp = time.monotonic_ns() try: with orchestrator_tracing.operation_span( "orchestrator.process_running_execution", @@ -331,9 +336,10 @@ def internal_process_running_executions_queue(self, session: orm.Session): ) finally: duration_ms = (time.monotonic_ns() - start_timestamp) / 1_000_000 - _logger.info( - f"After processing running container execution (duration: {duration_ms}ms)" - ) + if queries_duration_ms > 1000 or duration_ms > 1000: + _logger.info( + f"Slow running container execution visit (queries: {queries_duration_ms}ms, processing: {duration_ms}ms)" + ) return True else: if not self._running_executions_queue_idle or query_duration_ms > 1000: @@ -1003,7 +1009,6 @@ def internal_process_one_running_execution( reloaded_launched_container.status ) if new_status == previous_status: - _logger.info(f"Container execution remains in {new_status} state.") return _logger.info( f"Container execution is now in state {new_status} (was {previous_status})." @@ -1232,6 +1237,20 @@ def _maybe_preload_value( f"Processing running container execution {container_execution.id}: Error doing launched container cleanup." ) + def _maybe_log_heartbeat(self): + """Logs, about once a minute, how many queued and running executions the sweeps processed since the last one.""" + elapsed_seconds = time.monotonic() - self._heartbeat_started_at + if elapsed_seconds < _HEARTBEAT_INTERVAL_SECONDS: + return + _logger.info( + f"Orchestrator heartbeat: processed {self._heartbeat_queued_execution_count} queued executions and " + f"{self._heartbeat_running_container_execution_count} running container executions in the last " + f"{elapsed_seconds:.0f}s." + ) + self._heartbeat_started_at = time.monotonic() + self._heartbeat_queued_execution_count = 0 + self._heartbeat_running_container_execution_count = 0 + def _get_direct_downstream_executions( session: orm.Session, execution: bts.ExecutionNode diff --git a/tests/test_orchestrator_sql.py b/tests/test_orchestrator_sql.py index 77b4988..252e586 100644 --- a/tests/test_orchestrator_sql.py +++ b/tests/test_orchestrator_sql.py @@ -1,6 +1,7 @@ """Tests for ``orchestrator_sql``.""" import datetime +import logging from typing import Callable from unittest import mock @@ -921,3 +922,62 @@ def test_non_retriable_termination_error_is_a_system_error(self) -> None: sql.select(bts.ContainerExecution) ) assert container_execution.status == bts.ContainerExecutionStatus.SYSTEM_ERROR + + +def _heartbeat_messages(caplog: pytest.LogCaptureFixture) -> list[str]: + return [ + record.getMessage() + for record in caplog.records + if record.getMessage().startswith("Orchestrator heartbeat:") + ] + + +class TestHeartbeat: + def test_idle_sweeps_log_a_heartbeat_once_the_interval_passes( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.INFO, logger=orchestrator_sql.__name__) + orchestrator = _make_orchestrator( + session_factory=_create_session_factory(), + launched_container_mock=_make_launched_container_mock(), + ) + + orchestrator.process_each_queue_once() + assert _heartbeat_messages(caplog) == [] + + orchestrator._heartbeat_started_at -= 61 + orchestrator.process_each_queue_once() + assert len(_heartbeat_messages(caplog)) == 1 + assert ( + "processed 0 queued executions and 0 running container executions" + in _heartbeat_messages(caplog)[0] + ) + + def test_heartbeat_counts_processed_executions_and_then_resets( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.INFO, logger=orchestrator_sql.__name__) + session_factory = _create_session_factory() + _create_pipeline_run(session_factory, _single_task_pipeline()) + orchestrator = _make_running_orchestrator( + session_factory, _make_launched_container_mock() + ) + + assert orchestrator.internal_process_queued_executions_queue( + session=session_factory() + ) + assert orchestrator.internal_process_running_executions_queue( + session=session_factory() + ) + orchestrator._heartbeat_started_at -= 61 + orchestrator._maybe_log_heartbeat() + orchestrator._heartbeat_started_at -= 61 + orchestrator._maybe_log_heartbeat() + + first, second = _heartbeat_messages(caplog) + assert ( + "processed 1 queued executions and 1 running container executions" in first + ) + assert ( + "processed 0 queued executions and 0 running container executions" in second + )