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
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
"""OpenTelemetry parent span used while a durable span is not recording."""

from __future__ import annotations

import datetime
from collections.abc import Mapping
from typing import Any

from opentelemetry.trace import (
Span,
SpanContext,
SpanKind,
Status,
StatusCode,
)


class DurableParentSpan(Span):
"""A non-recording parent compatible with SDK span processors.

``NonRecordingSpan`` is the standard OpenTelemetry representation for a
span context without a recording span. Some vendor span processors,
including the ADOT processor used by Lambda, nevertheless inspect SDK-only
fields such as ``kind`` and ``attributes`` on every parent. This class
carries the same immutable context while exposing those read-only fields
with safe defaults.

The span also remembers the earliest and latest timestamps observed for
descendants. Durable operation spans are materialized only when the
operation completes, so this lets the eventual recording span enclose
attempts that started before that materialization point.
"""

def __init__(
self,
span_context: SpanContext,
*,
start_time: datetime.datetime | None = None,
) -> None:
self._span_context = span_context
self.kind = SpanKind.INTERNAL
self.attributes: Mapping[str, Any] = {}
self.name = ""
self.status = Status(StatusCode.UNSET)
self._earliest_start_time = start_time
self._latest_end_time: datetime.datetime | None = None

def get_span_context(self) -> SpanContext:
return self._span_context

def is_recording(self) -> bool:
return False

def end(self, end_time: int | None = None) -> None:
return

def set_attributes(self, attributes: Mapping[str, Any]) -> None:
return

def set_attribute(self, key: str, value: Any) -> None:
return

def add_event(
self,
name: str,
attributes: Mapping[str, Any] | None = None,
timestamp: int | None = None,
) -> None:
return

def update_name(self, name: str) -> None:
return

def set_status(
self,
status: Status | StatusCode,
description: str | None = None,
) -> None:
return

def record_exception(
self,
exception: BaseException,
attributes: Mapping[str, Any] | None = None,
timestamp: int | None = None,
escaped: bool = False,
) -> None:
return

def note_start_time(self, timestamp: datetime.datetime | None) -> None:
"""Include a descendant or operation start timestamp."""
if timestamp is None:
return
if self._earliest_start_time is None or timestamp < self._earliest_start_time:
self._earliest_start_time = timestamp

def note_end_time(self, timestamp: datetime.datetime | None) -> None:
"""Include a descendant or operation end timestamp."""
if timestamp is None:
return
if self._latest_end_time is None or timestamp > self._latest_end_time:
self._latest_end_time = timestamp
Comment on lines +90 to +102

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Claude AI review · Finding arf_v1_fmn2a33o4jdr6kuqfzdrvwg7ok

note_start_time/note_end_time mutate _earliest_start_time/_latest_end_time with an unsynchronized read-compare-write, but the object is shared as the parent placeholder for durable operations such as PARALLEL/MAP containers and open child contexts. context.parallel()/context.map() run branches concurrently on separate OS threads via ConcurrentExecutor's ThreadPoolExecutor (see operation/parallel.py, operation/child.py), and each branch's on_user_function_start/on_user_function_end in execution_plugin.py calls _note_parent_start(info.parent_id, ...)/_note_parent_end(info.parent_id, ...) on the same container DurableParentSpan instance whenever parent_id names the container operation. Two branch threads racing on if self._earliest_start_time is None or timestamp < self._earliest_start_time: self._earliest_start_time = timestamp can lose one update, so the eventually-materialized operation span can fail to enclose one of its concurrent children — exactly the defect normalized_start_time/normalized_end_time are meant to prevent. Every other piece of shared plugin state (_operation_spans, _context_tokens) is guarded by self._lock/self._operation_spans_lock; this new mutable state on DurableParentSpan is not. Add a threading.Lock inside DurableParentSpan and acquire it in note_start_time, note_end_time, normalized_start_time, and normalized_end_time.


def normalized_start_time(
self, timestamp: datetime.datetime | None
) -> datetime.datetime | None:
"""Return a start that encloses all observed descendants."""
if timestamp is None:
return self._earliest_start_time
if self._earliest_start_time is None:
return timestamp
return min(timestamp, self._earliest_start_time)

def normalized_end_time(
self,
timestamp: datetime.datetime | None,
*,
start_time: datetime.datetime | None = None,
) -> datetime.datetime | None:
"""Return an end that encloses all observed descendants."""
if timestamp is None:
normalized = self._latest_end_time
elif self._latest_end_time is None:
normalized = timestamp
else:
normalized = max(timestamp, self._latest_end_time)
if (
start_time is not None
and normalized is not None
and normalized <= start_time
):
return start_time + datetime.timedelta(microseconds=1)
return normalized


def ensure_end_after_start(
start_time: datetime.datetime | None,
end_time: datetime.datetime | None,
) -> datetime.datetime | None:
"""Prevent a completed span from ending at or before its start."""
if start_time is not None and end_time is not None and end_time <= start_time:
return start_time + datetime.timedelta(microseconds=1)
return end_time
Loading
Loading