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