Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog/fixed/traceable-query-quail.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix OpenTelemetryPlugin dropping the worker-side span for a rejected update or a failed query. An exception from an update validator or query handler does not fail the workflow task, so its span is now ended and exported with an error status.
3 changes: 3 additions & 0 deletions temporalio/contrib/opentelemetry/_tracer_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,13 @@ def end(self, end_time: int | None = None) -> None:
if (
workflow.in_workflow()
and self._exception is not None
and not workflow.unsafe.is_read_only()
and not workflow.is_failure_exception(self._exception)
):
# Skip ending spans with workflow task failures. Otherwise, each failure will create its own span
# This may still occur for spans which were completed during failed workflow tasks.
# Queries and update validators run read-only and their exceptions
# never fail the workflow task, so their spans are always ended.
return

self._span.end(end_time=end_time)
Expand Down
84 changes: 83 additions & 1 deletion tests/contrib/opentelemetry/test_opentelemetry_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,12 @@

import temporalio.contrib.opentelemetry.workflow
from temporalio import activity, nexus, workflow
from temporalio.client import Client, WorkflowFailureError
from temporalio.client import (
Client,
WorkflowFailureError,
WorkflowQueryFailedError,
WorkflowUpdateFailedError,
)
from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider
from temporalio.exceptions import ApplicationError
from temporalio.testing import WorkflowEnvironment
Expand Down Expand Up @@ -711,3 +716,80 @@ def test_replay_safe_span_delegates_extra_attributes():
_ = wrapper.nonexistent_attribute_xyz

inner_span.end()


@workflow.defn
class RejectingHandlersWorkflow:
def __init__(self) -> None:
self._value: int | None = None

@workflow.run
async def run(self) -> int | None:
await workflow.wait_condition(lambda: self._value is not None)
return self._value

@workflow.update
def set_value(self, value: int) -> None:
self._value = value

@set_value.validator
def validate_set_value(self, value: int) -> None:
if value < 0:
raise ValueError("value must not be negative")

@workflow.query
def failing_query(self) -> str:
raise ValueError("query failed")


async def test_otel_tracing_rejected_update_and_failed_query(
client: Client,
reset_otel_tracer_provider: Any, # type: ignore[reportUnusedParameter]
):
exporter = InMemorySpanExporter()
provider = create_tracer_provider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
opentelemetry.trace.set_tracer_provider(provider)

new_config = client.config()
new_config["plugins"] = [OpenTelemetryPlugin(add_temporal_spans=True)]
new_client = Client(**new_config)

async with new_worker(new_client, RejectingHandlersWorkflow) as worker:
handle = await new_client.start_workflow(
RejectingHandlersWorkflow.run,
id=f"rejecting-handlers-{uuid.uuid4()}",
task_queue=worker.task_queue,
)
with pytest.raises(WorkflowUpdateFailedError):
await handle.execute_update(RejectingHandlersWorkflow.set_value, -1)
with pytest.raises(WorkflowQueryFailedError):
await handle.query(RejectingHandlersWorkflow.failing_query)
await handle.execute_update(RejectingHandlersWorkflow.set_value, 1)
assert await handle.result() == 1

spans = exporter.get_finished_spans()

# The rejected update and the failed query are still traced
expected_hierarchy = [
"StartWorkflow:RejectingHandlersWorkflow",
" RunWorkflow:RejectingHandlersWorkflow",
"StartWorkflowUpdate:set_value",
" ValidateUpdate:set_value",
"QueryWorkflow:failing_query",
" HandleQuery:failing_query",
"StartWorkflowUpdate:set_value",
" ValidateUpdate:set_value",
" HandleUpdate:set_value",
]
actual_hierarchy = dump_spans(spans, with_attributes=False)
assert actual_hierarchy == expected_hierarchy, (
f"Span hierarchy mismatch.\nExpected:\n{expected_hierarchy}\nActual:\n{actual_hierarchy}"
)
failed_spans = [
span
for span in spans
if span.name in ("ValidateUpdate:set_value", "HandleQuery:failing_query")
and span.status.status_code == opentelemetry.trace.StatusCode.ERROR
]
assert len(failed_spans) == 2
Loading