import copy
import sys
import uuid
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import TYPE_CHECKING
import sentry_sdk
from sentry_sdk.consts import SPANDATA
from sentry_sdk.profiler.continuous_profiler import (
get_profiler_id,
try_autostart_continuous_profiler,
try_profile_lifecycle_trace_start,
)
from sentry_sdk.utils import (
capture_internal_exceptions,
deprecation_warning,
format_attribute,
get_current_thread_meta,
logger,
nanosecond_time,
should_be_treated_as_error,
)
if TYPE_CHECKING:
from typing import (
Any,
Callable,
Iterator,
Optional,
ParamSpec,
TypeVar,
Union,
overload,
)
from sentry_sdk._types import Attributes, AttributeValue, SpanJSON, SpanLink
from sentry_sdk.profiler.continuous_profiler import ContinuousProfile
P = ParamSpec("P")
R = TypeVar("R")
BAGGAGE_HEADER_NAME = "baggage"
SENTRY_TRACE_HEADER_NAME = "sentry-trace"
class SpanStatus(str, Enum):
OK = "ok"
ERROR = "error"
def __str__(self) -> str:
return self.value
_VALID_SPAN_STATUSES = frozenset(e.value for e in SpanStatus)
# The source of a segment name, see
# https://getsentry.github.io/sentry-conventions/attributes/sentry/#sentry-segment-name-source
class SegmentNameSource(str, Enum):
COMPONENT = "component"
CUSTOM = "custom"
ROUTE = "route"
TASK = "task"
URL = "url"
VIEW = "view"
def __str__(self) -> str:
return self.value
# These are typically high cardinality and the server hates them
LOW_QUALITY_SEGMENT_SOURCES = [
SegmentNameSource.URL,
]
# Sentinel value for an unset parent_span to be able to distinguish it from
# a None set by the user
_DEFAULT_PARENT_SPAN = object()
class _AgentFrameworkChatGenerationContext:
"""
Starts a span for a chat generation and marks the current scope as being
inside a chat generation as viewed by an agent framework.
Agent framework integrations should use this class to create chat client spans. By using this context
manager, chat client spans in "lower-level" client libraries can be suppressed by checking the
`_agent_framework_chat_generation_entered` scope flag.
"""
def __init__(
self,
name: str,
attributes: "Optional[Attributes]" = None,
parent_span: "Optional[Span]" = _DEFAULT_PARENT_SPAN, # type: ignore[assignment]
active: bool = True,
):
self._span = start_span(
name=name,
attributes=attributes,
parent_span=parent_span,
active=active,
)
if type(self._span) is not Span:
return
self.span._scope._agent_framework_chat_generation_entered = True
@property
def span(self) -> "Span":
return self._span
def __enter__(self) -> "_AgentFrameworkChatGenerationContext":
return self
def __exit__(
self, ty: "Optional[Any]", value: "Optional[Any]", tb: "Optional[Any]"
) -> None:
if type(self._span) is not Span:
self._span.__exit__(ty, value, tb)
try:
self._span.__exit__(ty, value, tb)
finally:
self._span._scope._agent_framework_chat_generation_entered = False
def start_span(
name: str,
attributes: "Optional[Attributes]" = None,
parent_span: "Optional[Span]" = _DEFAULT_PARENT_SPAN, # type: ignore[assignment]
active: bool = True,
links: "Optional[list[SpanLink]]" = None,
) -> "Span":
return sentry_sdk.get_current_scope().start_span(
name, attributes, parent_span, active, links
)
def continue_trace(incoming: "dict[str, Any]") -> None:
# This is set both on the isolation and the current scope for compatibility
# reasons. Conceptually, it belongs on the isolation scope, and it also
# used to be set there in non-span-first mode. But in span first mode, we
# start spans on the current scope, regardless of type, like JS does, so we
# need to set the propagation context there.
sentry_sdk.get_isolation_scope().generate_propagation_context(
incoming,
)
sentry_sdk.get_current_scope().generate_propagation_context(
incoming,
)
def new_trace() -> None:
sentry_sdk.get_isolation_scope().set_new_propagation_context()
sentry_sdk.get_current_scope().set_new_propagation_context()
class Span:
"""
A span holds timing information of a block of code.
Spans can have multiple child spans, thus forming a span tree.
This is the Span First span implementation that streams spans. The original
transaction-based span implementation lives in tracing.Span.
"""
__slots__ = (
"_name",
"_attributes",
"_links",
"_active",
"_span_id",
"_trace_id",
"_parent_span_id",
"_segment",
"_parent_sampled",
"_start_timestamp",
"_start_timestamp_monotonic_ns",
"_end_timestamp",
"_status",
"_scope",
"_previous_span_on_scope",
"_baggage",
"_sample_rand",
"_sample_rate",
"_continuous_profile",
)
def __init__(
self,
*,
name: str,
attributes: "Optional[Attributes]" = None,
active: bool = True,
scope: "sentry_sdk.Scope",
segment: "Optional[Span]" = None,
trace_id: "Optional[str]" = None,
parent_span_id: "Optional[str]" = None,
parent_sampled: "Optional[bool]" = None,
baggage: "Optional[Baggage]" = None,
sample_rate: "Optional[float]" = None,
sample_rand: "Optional[float]" = None,
links: "Optional[list[SpanLink]]" = None,
):
self._name: str = name
self._active: bool = active
self._attributes: "Attributes" = {
"sentry.origin": "manual",
"sentry.trace_lifecycle": "stream",
}
if attributes:
for attribute, value in attributes.items():
self.set_attribute(attribute, value)
self._links: "list[SpanLink]" = []
if links:
self.add_links(links)
self._scope = scope
self._segment = segment or self
self._trace_id: "Optional[str]" = trace_id
self._parent_span_id = parent_span_id
self._parent_sampled = parent_sampled
self._baggage = baggage
self._sample_rand = sample_rand
self._sample_rate = sample_rate
self._start_timestamp = datetime.now(timezone.utc)
self._end_timestamp: "Optional[datetime]" = None
# profiling depends on this value and requires that
# it is measured in nanoseconds
self._start_timestamp_monotonic_ns = nanosecond_time()
self._span_id: "Optional[str]" = None
self._status = SpanStatus.OK.value
self._update_active_thread()
self._continuous_profile: "Optional[ContinuousProfile]" = None
self._start_profile()
self._set_profile_id(get_profiler_id())
self._set_segment_attributes()
self._start()
def __repr__(self) -> str:
return (
f"<{self.__class__.__name__}("
f"name={self._name}, "
f"trace_id={self.trace_id}, "
f"span_id={self.span_id}, "
f"parent_span_id={self._parent_span_id}, "
f"active={self._active})>"
)
def __enter__(self) -> "Span":
return self
def __exit__(
self, ty: "Optional[Any]", value: "Optional[Any]", tb: "Optional[Any]"
) -> None:
if self._end_timestamp is not None:
# This span is already finished, ignore
return
if value is not None and should_be_treated_as_error(ty, value):
self.status = SpanStatus.ERROR.value
self._end()
def end(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None:
"""
Finish this span and queue it for sending.
:param end_timestamp: End timestamp to use instead of current time.
:type end_timestamp: "Optional[Union[float, datetime]]"
"""
self._end(end_timestamp)
def finish(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None:
deprecation_warning(
"span.finish() is deprecated. Use span.end() instead.",
)
self.end(end_timestamp)
def _start(self) -> None:
if self._active:
old_span = self._scope.span
self._scope.span = self
self._previous_span_on_scope = old_span
def _end(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None:
if self._end_timestamp is not None:
# This span is already finished, ignore.
return
# Stop the profiler
if self._is_segment() and self._continuous_profile is not None:
with capture_internal_exceptions():
self._continuous_profile.stop()
# Detach from scope
if self._active:
with capture_internal_exceptions():
old_span = self._previous_span_on_scope
del self._previous_span_on_scope
self._scope.span = old_span
# Set attributes from the segment. These are set on span end on purpose
# so that we have the best chance to capture the segment's final name
# (since it might change during its lifetime)
self.set_attribute("sentry.segment.id", self._segment.span_id)
self.set_attribute("sentry.segment.name", self._segment.name)
# Set conversation ID if one is set and this is a correct kind of span
conversation_id = self._scope.get_conversation_id()
if conversation_id:
has_ai_operation_name = SPANDATA.GEN_AI_OPERATION_NAME in self._attributes
op = self._attributes.get("sentry.op")
has_gen_ai_op = isinstance(op, str) and (
op.startswith("ai.") or op.startswith("gen_ai.")
)
if has_ai_operation_name or has_gen_ai_op:
self.set_attribute("gen_ai.conversation.id", conversation_id)
# Set the end timestamp
if end_timestamp is not None:
if isinstance(end_timestamp, (float, int)):
try:
end_timestamp = datetime.fromtimestamp(end_timestamp, timezone.utc)
except Exception:
pass
if isinstance(end_timestamp, datetime):
self._end_timestamp = end_timestamp
else:
logger.debug(
"[Tracing] Failed to set end_timestamp. Using current time instead."
)
if self._end_timestamp is None:
elapsed = nanosecond_time() - self._start_timestamp_monotonic_ns
self._end_timestamp = self._start_timestamp + timedelta(
microseconds=elapsed / 1000
)
client = sentry_sdk.get_client()
if not client.is_active():
return
# Finally, queue the span for sending to Sentry
self._scope._capture_span(self)
def get_attributes(self) -> "Attributes":
return self._attributes
def set_attribute(self, key: str, value: "AttributeValue") -> None:
self._attributes[key] = format_attribute(value)
def set_attributes(self, attributes: "Attributes") -> None:
for key, value in attributes.items():
self.set_attribute(key, value)
def remove_attribute(self, key: str) -> None:
try:
del self._attributes[key]
except KeyError:
pass
def add_link(self, link: "SpanLink") -> None:
stored: "SpanLink" = {"trace_id": link["trace_id"], "span_id": link["span_id"]}
if "sampled" in link:
stored["sampled"] = link["sampled"]
if link.get("attributes"):
stored["attributes"] = {
key: format_attribute(value)
for key, value in link["attributes"].items()
}
self._links.append(stored)
def add_links(self, links: "list[SpanLink]") -> None:
for link in links:
self.add_link(link)
@property
def status(self) -> "str":
return self._status
@status.setter
def status(self, status: "Union[SpanStatus, str]") -> None:
if isinstance(status, Enum):
status = status.value
if status not in _VALID_SPAN_STATUSES:
logger.debug(
f'[Tracing] Unsupported span status {status}. Expected one of: "ok", "error"'
)
return
self._status = status
@property
def name(self) -> str:
return self._name
@name.setter
def name(self, name: str) -> None:
self._name = name
@property
def active(self) -> bool:
return self._active
@property
def span_id(self) -> str:
if not self._span_id:
self._span_id = uuid.uuid4().hex[16:]
return self._span_id
@property
def trace_id(self) -> str:
if not self._trace_id:
self._trace_id = uuid.uuid4().hex
return self._trace_id
@property
def sampled(self) -> "Optional[bool]":
return True
@property
def start_timestamp(self) -> "Optional[datetime]":
return self._start_timestamp
@property
def end_timestamp(self) -> "Optional[datetime]":
return self._end_timestamp
def _is_segment(self) -> bool:
return self._segment is self
def _update_active_thread(self) -> None:
thread_id, thread_name = get_current_thread_meta()
if thread_id is not None:
self.set_attribute(SPANDATA.THREAD_ID, str(thread_id))
if thread_name is not None:
self.set_attribute(SPANDATA.THREAD_NAME, thread_name)
def _dynamic_sampling_context(self) -> "dict[str, str]":
return self._segment._get_baggage().dynamic_sampling_context()
def _to_traceparent(self) -> str:
if self._segment.sampled is True:
sampled = "1"
elif self._segment.sampled is False:
sampled = "0"
else:
sampled = None
traceparent = f"{self.trace_id}-{self.span_id}"
if sampled is not None:
traceparent += f"-{sampled}"
return traceparent
def _to_baggage(self) -> "Optional[Baggage]":
if self._segment:
return self._segment._get_baggage()
return None
def _get_baggage(self) -> "Baggage":
"""
Return the :py:class:`~sentry_sdk.tracing_utils.Baggage` associated with
the segment.
The first time a new baggage with Sentry items is made, it will be frozen.
"""
if not self._baggage or self._baggage.mutable:
self._baggage = Baggage.populate_from_segment(self)
return self._baggage
def _iter_headers(self) -> "Iterator[tuple[str, str]]":
if not self._segment:
return
yield SENTRY_TRACE_HEADER_NAME, self._to_traceparent()
baggage = self._segment._get_baggage().serialize()
if baggage:
yield BAGGAGE_HEADER_NAME, baggage
def _get_trace_context(self) -> "dict[str, Any]":
# Even if spans themselves are not event-based anymore, we need this
# to populate trace context on events
context: "dict[str, Any]" = {
"trace_id": self.trace_id,
"span_id": self.span_id,
"parent_span_id": self._parent_span_id,
"dynamic_sampling_context": self._dynamic_sampling_context(),
}
if "sentry.op" in self._attributes:
context["op"] = self._attributes["sentry.op"]
if "sentry.origin" in self._attributes:
context["origin"] = self._attributes["sentry.origin"]
return context
def _set_profile_id(self, profiler_id: "Optional[str]") -> None:
if profiler_id is not None:
self.set_attribute("sentry.profiler_id", profiler_id)
def _start_profile(self) -> None:
if not self._is_segment():
return
try_autostart_continuous_profiler()
self._continuous_profile = try_profile_lifecycle_trace_start()
def _set_segment_attributes(self) -> None:
if not self._is_segment():
return
client = sentry_sdk.get_client()
self.set_attribute(SPANDATA.SENTRY_PLATFORM, "python")
self.set_attribute(SPANDATA.PROCESS_COMMAND_ARGS, sys.argv)
self.set_attribute(
SPANDATA.SENTRY_SDK_INTEGRATIONS, sorted(client.integrations.keys())
)
if not self.get_attributes().get("sentry.segment.name.source"):
self.set_attribute(
"sentry.segment.name.source", SegmentNameSource.CUSTOM.value
)
if client.options.get("dist") and SPANDATA.SENTRY_DIST not in self._attributes:
self.set_attribute(
SPANDATA.SENTRY_DIST, str(client.options["dist"]).strip()
)
def _to_json(self) -> "SpanJSON":
res: "SpanJSON" = {
"trace_id": self.trace_id,
"span_id": self.span_id,
"name": self._name if self._name is not None else "<unlabeled span>",
"status": self._status,
"is_segment": self._is_segment(),
"start_timestamp": self._start_timestamp.timestamp(),
}
if self._end_timestamp:
res["end_timestamp"] = self._end_timestamp.timestamp()
if self._parent_span_id:
res["parent_span_id"] = self._parent_span_id
res["attributes"] = {k: v for k, v in self._attributes.items()}
if self._links:
res["links"] = copy.deepcopy(self._links)
return res
class NoOpSpan(Span):
__slots__ = (
"_sampled",
"_finished",
"_unsampled_reason",
)
def __init__(
self,
name: "Optional[str]" = None,
attributes: "Optional[Attributes]" = None,
segment: "Optional[Span]" = None,
trace_id: "Optional[str]" = None,
parent_span_id: "Optional[str]" = None,
parent_sampled: "Optional[bool]" = None,
baggage: "Optional[Baggage]" = None,
sampled: "Optional[bool]" = False,
unsampled_reason: "Optional[str]" = None,
scope: "Optional[sentry_sdk.Scope]" = None,
sample_rand: "Optional[float]" = None,
sample_rate: "Optional[float]" = None,
) -> None:
self._name = name # type: ignore[assignment]
self._attributes = {}
if attributes is not None and "sentry.segment.name.source" in attributes:
self.set_attribute(
"sentry.segment.name.source", attributes["sentry.segment.name.source"]
)
self._span_id: "Optional[str]" = None
self._sampled = sampled
self._segment = segment or self
self._trace_id: "Optional[str]" = trace_id
self._parent_span_id = parent_span_id
self._parent_sampled = parent_sampled
self._baggage = baggage
self._sample_rand = sample_rand
self._sample_rate = sample_rate
self._scope = scope # type: ignore[assignment]
self._unsampled_reason = unsampled_reason
self._finished = False
self._start()
def __repr__(self) -> str:
return f"<{self.__class__.__name__}(name={self.name}, sampled={self.sampled})>"
def __enter__(self) -> "NoOpSpan":
return self
def __exit__(
self, ty: "Optional[Any]", value: "Optional[Any]", tb: "Optional[Any]"
) -> None:
self._end()
def _start(self) -> None:
if self._scope is None:
return
old_span = self._scope.span
self._scope.span = self
self._previous_span_on_scope = old_span
def _end(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None:
if self._finished:
return
if self._unsampled_reason is not None:
client = sentry_sdk.get_client()
if client.is_active() and client.transport:
logger.debug(
f"[Tracing] Discarding span because sampled=False (reason: {self._unsampled_reason})"
)
client.transport.record_lost_event(
reason=self._unsampled_reason,
data_category="span",
quantity=1,
)
if self._scope and hasattr(self, "_previous_span_on_scope"):
with capture_internal_exceptions():
old_span = self._previous_span_on_scope
del self._previous_span_on_scope
self._scope.span = old_span
self._finished = True
def end(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None:
self._end()
def finish(self, end_timestamp: "Optional[Union[float, datetime]]" = None) -> None:
deprecation_warning(
"span.finish() is deprecated. Use span.end() instead.",
)
self._end()
def get_attributes(self) -> "Attributes":
return {}
def set_attribute(self, key: str, value: "AttributeValue") -> None:
if key != "sentry.segment.name.source":
return
super().set_attribute("sentry.segment.name.source", value)
def set_attributes(self, attributes: "Attributes") -> None:
for key, value in attributes.items():
if key != "sentry.segment.name.source":
continue
self.set_attribute("sentry.segment.name.source", value)
def remove_attribute(self, key: str) -> None:
pass
def add_link(self, link: "SpanLink") -> None:
pass
def add_links(self, links: "list[SpanLink]") -> None:
pass
@property
def status(self) -> "str":
return SpanStatus.OK.value
@status.setter
def status(self, status: "Union[SpanStatus, str]") -> None:
pass
@property
def name(self) -> str:
return self._name or ""
@name.setter
def name(self, name: str) -> None:
self._name = name
@property
def active(self) -> bool:
return True
@property
def sampled(self) -> "Optional[bool]":
return self._sampled
@property
def start_timestamp(self) -> "Optional[datetime]":
return None
@property
def end_timestamp(self) -> "Optional[datetime]":
return None
def _get_trace_context(self) -> "dict[str, Any]":
return {
"trace_id": self.trace_id,
"span_id": self.span_id,
"parent_span_id": self._parent_span_id,
"dynamic_sampling_context": self._dynamic_sampling_context(),
}
# backwards compat
StreamedSpan = Span
NoOpStreamedSpan = NoOpSpan
if TYPE_CHECKING:
@overload
def trace(
func: "Callable[P, R]",
) -> "Callable[P, R]": ...
@overload
def trace(
func: None = None,
*,
name: "Optional[str]" = None,
attributes: "Optional[dict[str, Any]]" = None,
active: bool = True,
) -> "Callable[[Callable[P, R]], Callable[P, R]]": ...
[docs]
def trace(
func: "Optional[Callable[P, R]]" = None,
*,
name: "Optional[str]" = None,
attributes: "Optional[dict[str, Any]]" = None,
active: bool = True,
) -> "Union[Callable[P, R], Callable[[Callable[P, R]], Callable[P, R]]]":
"""
Decorator to start a span around a function call in streaming mode.
This decorator automatically creates a new span when the decorated function
is called, and finishes the span when the function returns or raises an exception.
:param func: The function to trace. When used as a decorator without parentheses,
this is the function being decorated. When used with parameters (e.g.,
``@trace(op="custom")``, this should be None.
:type func: Callable or None
:param name: The human-readable name/description for the span. If not provided,
defaults to the function name. This provides more specific details about
what the span represents (e.g., "GET /api/users", "process_user_data").
:type name: str or None
:param attributes: A dictionary of key-value pairs to add as attributes to the span.
Attribute values must be strings, integers, floats, or booleans. These
attributes provide additional context about the span's execution.
:type attributes: dict[str, Any] or None
:param active: Controls whether spans started while this span is running
will automatically become its children. That's the default behavior. If
you want to create a span that shouldn't have any children (unless
provided explicitly via the `parent_span` argument), set this to False.
:type active: bool
:returns: When used as ``@trace``, returns the decorated function. When used as
``@trace(...)`` with parameters, returns a decorator function.
:rtype: Callable or decorator function
Example::
import sentry_sdk
# Simple usage with default values
@sentry_sdk.trace
def process_data():
# Function implementation
pass
# With custom parameters
@sentry_sdk.trace(
name="Get user data",
attributes={"postgres": True}
)
def make_db_query(sql):
# Function implementation
pass
"""
from sentry_sdk.tracing_utils import (
create_span_decorator,
)
decorator = create_span_decorator(
name=name,
attributes=attributes,
active=active,
)
if func:
return decorator(func)
else:
return decorator
def get_current_span(
scope: "Optional[sentry_sdk.Scope]" = None,
) -> "Optional[Span]":
scope = scope or sentry_sdk.get_current_scope()
current_span = scope.span
return current_span
# Circular import
from sentry_sdk.tracing_utils import Baggage # noqa: E402, F401, I001
sentry-python