chore(orchestrator): Replace per-visit sweep logs with a heartbeat

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

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
+        )