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