-
Notifications
You must be signed in to change notification settings - Fork 23
fix(otel): end recording spans on non-terminal #696
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
Open
Changes from all commits
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
143 changes: 143 additions & 0 deletions
143
...xecution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/durable_parent_span.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| 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 | ||
|
|
||
| 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 | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
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_fmn2a33o4jdr6kuqfzdrvwg7oknote_start_time/note_end_timemutate_earliest_start_time/_latest_end_timewith 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 viaConcurrentExecutor'sThreadPoolExecutor(seeoperation/parallel.py,operation/child.py), and each branch'son_user_function_start/on_user_function_endinexecution_plugin.pycalls_note_parent_start(info.parent_id, ...)/_note_parent_end(info.parent_id, ...)on the same containerDurableParentSpaninstance wheneverparent_idnames the container operation. Two branch threads racing onif self._earliest_start_time is None or timestamp < self._earliest_start_time: self._earliest_start_time = timestampcan lose one update, so the eventually-materialized operation span can fail to enclose one of its concurrent children — exactly the defectnormalized_start_time/normalized_end_timeare meant to prevent. Every other piece of shared plugin state (_operation_spans,_context_tokens) is guarded byself._lock/self._operation_spans_lock; this new mutable state onDurableParentSpanis not. Add athreading.LockinsideDurableParentSpanand acquire it innote_start_time,note_end_time,normalized_start_time, andnormalized_end_time.