From 750469f419d7eb26c573b302bfe780bd21befc26 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 | 33 +++++++++--- tests/test_orchestrator_sql.py | 60 +++++++++++++++++++++ 2 files changed, 86 insertions(+), 7 deletions(-) diff --git a/cloud_pipelines_backend/orchestrator_sql.py b/cloud_pipelines_backend/orchestrator_sql.py index 178fb3b..6416b2c 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,6 +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 + self._heartbeat_running_container_execution_count += 1 execution_node_ids = list( session.scalars( @@ -241,9 +249,6 @@ def internal_process_running_executions_queue(self, session: orm.Session): time.monotonic_ns() - query_start_timestamp ) / 1_000_000 start_timestamp = time.monotonic_ns() - _logger.info( - f"Before processing running container execution. Queries duration {queries_duration_ms}ms." - ) 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)" + ) app_metrics.record_refresh_duration( queries_duration_ms=queries_duration_ms, processing_duration_ms=duration_ms, @@ -1007,7 +1013,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})." @@ -1240,6 +1245,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 bd75704..4234395 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 @@ -995,3 +996,62 @@ def test_a_succeeded_container_missing_outputs_records_the_committed_failed_stat app_metrics.record_completion_lag.assert_called_once_with( ended_at=self._ENDED_AT, status=bts.ContainerExecutionStatus.FAILED ) + + +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 + )