diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index c031b5653..1b217b4a5 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -9,13 +9,13 @@ import os import sys import threading -import time from collections.abc import Awaitable, Callable, MutableMapping, Sequence from dataclasses import dataclass -from datetime import timedelta, timezone +from datetime import timezone from types import TracebackType import temporalio.api.common.v1 +import temporalio.bridge.proto.common import temporalio.bridge.proto.workflow_activation import temporalio.bridge.proto.workflow_completion import temporalio.bridge.runtime @@ -64,6 +64,17 @@ _DEFAULT_WORKFLOW_TASK_EXTERNAL_STORAGE_CONCURRENCY: int = 3 +def _set_external_storage_metrics( + target: temporalio.bridge.proto.common.ExternalStorageMetrics, + metrics: temporalio.converter._extstore.StorageOperationMetrics, +) -> None: + """Populate a proto ``ExternalStorageMetrics`` from measured storage metrics.""" + target.payload_count = metrics.payload_count + target.total_size_bytes = metrics.total_size + target.total_duration.FromTimedelta(metrics.total_duration) + target.driver_names.extend(sorted(metrics.driver_names)) + + class _WorkflowWorker: # type:ignore[reportUnusedClass] def __init__( self, @@ -325,7 +336,6 @@ async def _handle_activation( completion.successful.SetInParent() workflow = None data_converter = self._data_converter - task_start_time = time.monotonic() download_metrics = temporalio.converter._extstore.StorageOperationMetrics() try: if LOG_PROTOS: @@ -500,6 +510,17 @@ async def _handle_activation( completion.failed.Clear() completion.failed.failure.message = f"Failed encoding completion: {err}" + # Reported on the completion so core can include them in its workflow-task duration + # log; core measures the duration itself. + if download_metrics.payload_count > 0: + _set_external_storage_metrics( + completion.payload_download_metrics, download_metrics + ) + if upload_metrics.payload_count > 0: + _set_external_storage_metrics( + completion.payload_upload_metrics, upload_metrics + ) + # Send off completion if LOG_PROTOS: logger.debug("Sending workflow completion:\n%s", completion) @@ -511,84 +532,6 @@ async def _handle_activation( "Failed completing activation on workflow with run ID %s", act.run_id ) - # Log workflow task duration with external storage metrics - self._log_workflow_task_duration( - act, workflow, task_start_time, download_metrics, upload_metrics - ) - - def _log_workflow_task_duration( - self, - act: temporalio.bridge.proto.workflow_activation.WorkflowActivation, - workflow: _RunningWorkflow | None, - task_start_time: float, - download_metrics: temporalio.converter._extstore.StorageOperationMetrics, - upload_metrics: temporalio.converter._extstore.StorageOperationMetrics, - ) -> None: - task_duration = timedelta(seconds=time.monotonic() - task_start_time) - - def _fmt_duration(td: timedelta) -> str: - secs = td.total_seconds() - if secs >= 1: - return f"{secs:.3f}s" - return f"{secs * 1000:.3f}ms" - - completed_event_id = act.history_length + 1 - _info = workflow.get_info() if workflow is not None else None - attempt = _info.attempt if _info is not None else "unknown" - log_id = f"{act.run_id}:{completed_event_id}:{attempt}" - msg_details, extra = temporalio.workflow._build_log_context( - _info._logger_details() if _info is not None else None, - full_workflow_info=_info, - ) - msg_details["event_id"] = completed_event_id - msg_details["workflow_task_duration"] = _fmt_duration(task_duration) - msg_details["workflow_history_size"] = act.history_size_bytes - extra["event_id"] = completed_event_id - extra["workflow_task_duration"] = task_duration - extra["workflow_history_size"] = act.history_size_bytes - if download_metrics.payload_count > 0: - msg_details["payload_download_count"] = download_metrics.payload_count - msg_details["payload_download_size"] = download_metrics.total_size - msg_details["payload_download_duration"] = _fmt_duration( - download_metrics.total_duration - ) - msg_details["payload_download_drivers"] = sorted( - download_metrics.driver_names - ) - extra["payload_download_count"] = download_metrics.payload_count - extra["payload_download_size"] = download_metrics.total_size - extra["payload_download_duration"] = download_metrics.total_duration - extra["payload_download_drivers"] = sorted(download_metrics.driver_names) - if upload_metrics.payload_count > 0: - msg_details["payload_upload_count"] = upload_metrics.payload_count - msg_details["payload_upload_size"] = upload_metrics.total_size - msg_details["payload_upload_duration"] = _fmt_duration( - upload_metrics.total_duration - ) - msg_details["payload_upload_drivers"] = sorted(upload_metrics.driver_names) - extra["payload_upload_count"] = upload_metrics.payload_count - extra["payload_upload_size"] = upload_metrics.total_size - extra["payload_upload_duration"] = upload_metrics.total_duration - extra["payload_upload_drivers"] = sorted(upload_metrics.driver_names) - if task_duration.total_seconds() > 10: - logger.warning( - f"[TMPRL1104] {log_id} Workflow task exceeded 10 seconds (%s)", - msg_details, - extra=extra, - ) - elif task_duration.total_seconds() > 5: - logger.info( - f"[TMPRL1104] {log_id} Workflow task exceeded 5 seconds (%s)", - msg_details, - extra=extra, - ) - else: - logger.debug( - f"[TMPRL1104] {log_id} Workflow task duration information (%s)", - msg_details, - extra=extra, - ) - async def _handle_cache_eviction( self, act: temporalio.bridge.proto.workflow_activation.WorkflowActivation, diff --git a/tests/worker/test_extstore.py b/tests/worker/test_extstore.py index 2f8fde5fe..e8ef8edb2 100644 --- a/tests/worker/test_extstore.py +++ b/tests/worker/test_extstore.py @@ -1,8 +1,7 @@ +import contextlib import dataclasses -import logging -import re import uuid -from collections.abc import Sequence +from collections.abc import Iterator, Sequence from dataclasses import dataclass from datetime import timedelta from unittest import mock @@ -11,10 +10,10 @@ import temporalio import temporalio.bridge.client +import temporalio.bridge.proto.workflow_completion import temporalio.bridge.worker import temporalio.client import temporalio.converter -import temporalio.worker._workflow from temporalio import activity, workflow from temporalio.api.common.v1 import Payload from temporalio.client import Client, WorkflowFailureError, WorkflowHandle @@ -31,7 +30,7 @@ from temporalio.exceptions import ActivityError, ApplicationError from temporalio.testing._workflow import WorkflowEnvironment from temporalio.worker import Replayer -from tests.helpers import LogCapturer, assert_task_fail_eventually, new_worker +from tests.helpers import assert_task_fail_eventually, new_worker from tests.test_extstore import InMemoryTestDriver @@ -599,19 +598,32 @@ async def test_worker_storage_drivers_empty_without_external_storage( # TMPRL1104 workflow task duration logging # --------------------------------------------------------------------------- -_workflow_logger = logging.getLogger(temporalio.worker._workflow.__name__) +# The duration log itself is emitted (and tested) in sdk-core. The Python worker's part is +# attaching the external-storage metrics to the completion, so these tests capture the +# completion and assert on its fields directly rather than on core's asynchronously +# forwarded log, which would be nondeterministic to observe here. -def _tmprl1104_records(capturer: LogCapturer) -> list[logging.LogRecord]: - """Return all TMPRL1104 log records from the capturer.""" - return capturer.find_all(lambda r: r.getMessage().startswith("[TMPRL1104]")) +@contextlib.contextmanager +def _capture_completions() -> Iterator[ + list[temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion] +]: + """Capture every WorkflowActivationCompletion the worker hands to core.""" + completions: list[ + temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion + ] = [] + original = temporalio.bridge.worker.Worker.complete_workflow_activation + async def capturing(self, completion): # type: ignore[no-untyped-def] + completions.append(completion) + return await original(self, completion) -# Accept any duration-bucket wording: a loaded host can push a trivial task past 5s. -_TMPRL1104_DURATION_MESSAGE = re.compile( - r"\[TMPRL1104\] [^:]+:\d+:\d+ Workflow task " - r"(?:duration information|exceeded \d+ seconds) \(" -) + with mock.patch.object( + temporalio.bridge.worker.Worker, + "complete_workflow_activation", + capturing, + ): + yield completions async def _expected_payload_size( @@ -622,44 +634,33 @@ async def _expected_payload_size( return payloads[0].ByteSize() -@workflow.defn -class SimpleWorkflow: - """Minimal workflow for testing logging without external storage.""" - - @workflow.run - async def run(self) -> str: - return "done" - - async def test_tmprl1104_no_extstore(env: WorkflowEnvironment) -> None: - """Without external storage, TMPRL1104 logs contain duration but no - download/upload metrics.""" - with LogCapturer().logs_captured(_workflow_logger, level=logging.DEBUG) as capturer: - async with new_worker(env.client, SimpleWorkflow) as worker: + """Without external storage configured, completions carry no storage metrics.""" + with _capture_completions() as completions: + async with new_worker( + env.client, ExtStoreWorkflow, activities=[ext_store_activity] + ) as worker: await env.client.execute_workflow( - SimpleWorkflow.run, + ExtStoreWorkflow.run, + ExtStoreWorkflowInput( + input_data="small", + activity_input_size=10, + activity_output_size=10, + output_size=10, + ), id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, ) - records = _tmprl1104_records(capturer) - assert len(records) == 1 - record = records[0] - assert _TMPRL1104_DURATION_MESSAGE.match(record.getMessage()) - assert hasattr(record, "workflow_task_duration") - assert hasattr(record, "event_id") - # No external storage — download/upload fields must be absent - assert not hasattr(record, "payload_download_count") - assert not hasattr(record, "payload_download_size") - assert not hasattr(record, "payload_download_duration") - assert not hasattr(record, "payload_upload_count") - assert not hasattr(record, "payload_upload_size") - assert not hasattr(record, "payload_upload_duration") + assert completions, "expected the worker to complete at least one activation" + for c in completions: + assert not c.HasField("payload_download_metrics") + assert not c.HasField("payload_upload_metrics") async def test_tmprl1104_with_extstore_download(env: WorkflowEnvironment) -> None: - """When external storage decodes payloads, TMPRL1104 logs include download - metrics on the activation that retrieves them.""" + """When external storage retrieves payloads, the completion for the WFT that + retrieved them carries download metrics.""" driver = InMemoryTestDriver() data_converter = dataclasses.replace( temporalio.converter.default(), @@ -680,7 +681,7 @@ async def test_tmprl1104_with_extstore_download(env: WorkflowEnvironment) -> Non ) expected_input_size = await _expected_payload_size(data_converter, wf_input) - with LogCapturer().logs_captured(_workflow_logger, level=logging.DEBUG) as capturer: + with _capture_completions() as completions: async with new_worker( client, ExtStoreWorkflow, activities=[ext_store_activity] ) as worker: @@ -691,25 +692,19 @@ async def test_tmprl1104_with_extstore_download(env: WorkflowEnvironment) -> Non task_queue=worker.task_queue, ) - records = _tmprl1104_records(capturer) - assert len(records) == 2 - - # WFT 1: retrieves the externalized workflow input - assert _TMPRL1104_DURATION_MESSAGE.match(records[0].getMessage()) - assert getattr(records[0], "payload_download_count") == 1 - assert getattr(records[0], "payload_download_size") == expected_input_size - assert getattr(records[0], "payload_download_duration") > timedelta(0) - assert not hasattr(records[0], "payload_upload_count") - - # WFT 2: activity result is small — no external storage - assert _TMPRL1104_DURATION_MESSAGE.match(records[1].getMessage()) - assert not hasattr(records[1], "payload_download_count") - assert not hasattr(records[1], "payload_upload_count") + downloads = [c for c in completions if c.HasField("payload_download_metrics")] + assert len(downloads) == 1 + m = downloads[0].payload_download_metrics + assert m.payload_count == 1 + assert m.total_size_bytes == expected_input_size + assert m.total_duration.ToTimedelta() > timedelta(0) + assert list(m.driver_names) == [driver.name()] + assert not any(c.HasField("payload_upload_metrics") for c in completions) async def test_tmprl1104_with_extstore_upload(env: WorkflowEnvironment) -> None: - """When external storage encodes payloads, TMPRL1104 logs include upload - metrics on the WFT that produces them.""" + """When external storage stores payloads, the completion for the WFT that + produced them carries upload metrics.""" driver = InMemoryTestDriver() data_converter = dataclasses.replace( temporalio.converter.default(), @@ -725,7 +720,7 @@ async def test_tmprl1104_with_extstore_upload(env: WorkflowEnvironment) -> None: wf_output = "wo" * 1024 # 2048 bytes → stored externally expected_output_size = await _expected_payload_size(data_converter, wf_output) - with LogCapturer().logs_captured(_workflow_logger, level=logging.DEBUG) as capturer: + with _capture_completions() as completions: async with new_worker( client, ExtStoreWorkflow, activities=[ext_store_activity] ) as worker: @@ -741,27 +736,21 @@ async def test_tmprl1104_with_extstore_upload(env: WorkflowEnvironment) -> None: task_queue=worker.task_queue, ) - records = _tmprl1104_records(capturer) - assert len(records) == 2 - - # WFT 1: small input — no external storage - assert _TMPRL1104_DURATION_MESSAGE.match(records[0].getMessage()) - assert not hasattr(records[0], "payload_download_count") - assert not hasattr(records[0], "payload_upload_count") - - # WFT 2: workflow returns large result → uploaded - assert _TMPRL1104_DURATION_MESSAGE.match(records[1].getMessage()) - assert not hasattr(records[1], "payload_download_count") - assert getattr(records[1], "payload_upload_count") == 1 - assert getattr(records[1], "payload_upload_size") == expected_output_size - assert getattr(records[1], "payload_upload_duration") > timedelta(0) + uploads = [c for c in completions if c.HasField("payload_upload_metrics")] + assert len(uploads) == 1 + m = uploads[0].payload_upload_metrics + assert m.payload_count == 1 + assert m.total_size_bytes == expected_output_size + assert m.total_duration.ToTimedelta() > timedelta(0) + assert list(m.driver_names) == [driver.name()] + assert not any(c.HasField("payload_download_metrics") for c in completions) async def test_tmprl1104_with_extstore_download_and_upload( env: WorkflowEnvironment, ) -> None: - """When both download and upload happen across WFTs, TMPRL1104 logs include - both sets of metrics.""" + """When both download and upload happen across WFTs, the respective completions + carry the matching metrics.""" driver = InMemoryTestDriver() data_converter = dataclasses.replace( temporalio.converter.default(), @@ -784,7 +773,7 @@ async def test_tmprl1104_with_extstore_download_and_upload( wf_output = "wo" * 1024 expected_output_size = await _expected_payload_size(data_converter, wf_output) - with LogCapturer().logs_captured(_workflow_logger, level=logging.DEBUG) as capturer: + with _capture_completions() as completions: async with new_worker( client, ExtStoreWorkflow, activities=[ext_store_activity] ) as worker: @@ -795,22 +784,19 @@ async def test_tmprl1104_with_extstore_download_and_upload( task_queue=worker.task_queue, ) - records = _tmprl1104_records(capturer) - assert len(records) == 2 - - # WFT 1: retrieves externalized workflow input - assert _TMPRL1104_DURATION_MESSAGE.match(records[0].getMessage()) - assert getattr(records[0], "payload_download_count") == 1 - assert getattr(records[0], "payload_download_size") == expected_input_size - assert getattr(records[0], "payload_download_duration") > timedelta(0) - assert not hasattr(records[0], "payload_upload_count") - - # WFT 2: uploads externalized workflow result - assert _TMPRL1104_DURATION_MESSAGE.match(records[1].getMessage()) - assert not hasattr(records[1], "payload_download_count") - assert getattr(records[1], "payload_upload_count") == 1 - assert getattr(records[1], "payload_upload_size") == expected_output_size - assert getattr(records[1], "payload_upload_duration") > timedelta(0) + downloads = [c for c in completions if c.HasField("payload_download_metrics")] + assert len(downloads) == 1 + dm = downloads[0].payload_download_metrics + assert dm.payload_count == 1 + assert dm.total_size_bytes == expected_input_size + assert dm.total_duration.ToTimedelta() > timedelta(0) + + uploads = [c for c in completions if c.HasField("payload_upload_metrics")] + assert len(uploads) == 1 + um = uploads[0].payload_upload_metrics + assert um.payload_count == 1 + assert um.total_size_bytes == expected_output_size + assert um.total_duration.ToTimedelta() > timedelta(0) # ---------------------------------------------------------------------------