# Copyright (c) Microsoft. All rights reserved. """Utilities shared for OpenTelemetry span (attributes) support.""" import json import logging import traceback from typing import Any, Dict, List, Sequence, Type, TypeVar, Union, cast from warnings import filterwarnings import opentelemetry.trace as trace_api from agentops.sdk.exporters import OTLPSpanExporter from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, SpanProcessor, SynchronousMultiSpanProcessor, Tracer from opentelemetry.sdk.trace import TracerProvider as TracerProviderImpl from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor from opentelemetry.sdk.util.instrumentation import InstrumentationInfo, InstrumentationScope from opentelemetry.semconv.attributes import exception_attributes from opentelemetry.trace import get_tracer_provider as otel_get_tracer_provider from pydantic import TypeAdapter from agentlightning.env_var import LightningEnvVar, resolve_bool_env_var from agentlightning.semconv import LightningSpanAttributes, LinkAttributes, LinkPydanticModel from agentlightning.types import Attributes, AttributeValue, SpanLike from agentlightning.utils.otlp import LightningStoreOTLPExporter logger = logging.getLogger(__name__) __all__ = [ "full_qualified_name", "get_tracer_provider", "get_tracer", "make_tag_attributes", "extract_tags_from_attributes", "make_link_attributes", "query_linked_spans", "extract_links_from_attributes", "filter_attributes", "filter_and_unflatten_attributes", "flatten_attributes", "unflatten_attributes", "sanitize_attribute_value", "sanitize_attributes", "sanitize_list_attribute_sanity", "check_attributes_sanity", "format_exception_attributes", ] T_SpanLike = TypeVar("T_SpanLike", bound=SpanLike) T_SpanProcessor = TypeVar("T_SpanProcessor", bound=SpanProcessor) def full_qualified_name(obj: type) -> str: if str(obj.__module__) == "builtins": return obj.__qualname__ return f"{obj.__module__}.{obj.__qualname__}" def get_tracer_provider(inspect: bool = True) -> TracerProviderImpl: """Get the OpenTelemetry tracer provider configured for Agent Lightning. Args: inspect: Whether to inspect the tracer provider and log its configuration. When it's on, make sure you also set the logger level to DEBUG to see the logs. """ from agentlightning.tracer.otel import LightningSpanProcessor if hasattr(trace_api, "_TRACER_PROVIDER") and trace_api._TRACER_PROVIDER is None: # type: ignore[attr-defined] raise RuntimeError("Tracer is not initialized. Cannot emit a meaningful span.") tracer_provider = otel_get_tracer_provider() if not isinstance(tracer_provider, TracerProviderImpl): logger.error( "Tracer provider is expected to be an instance of opentelemetry.sdk.trace.TracerProvider, found: %s", full_qualified_name(type(tracer_provider)), ) return cast(TracerProviderImpl, tracer_provider) if not inspect: return tracer_provider emitter_debug = resolve_bool_env_var(LightningEnvVar.AGL_EMITTER_DEBUG, fallback=None) logger_effective_level = logger.getEffectiveLevel() if emitter_debug is True and logger_effective_level > logging.DEBUG: logger.warning( "Emitter debug logging is enabled but logging level is not set to DEBUG. Nothing will be logged." ) if emitter_debug is None: # Set to true by default if the logging level is lower than DEBUG emitter_debug = logging.DEBUG >= logger_effective_level if emitter_debug: active_span_processor = tracer_provider._active_span_processor # pyright: ignore[reportPrivateUsage] processors: List[str] = [] active_span_processor_cls = active_span_processor.__class__.__name__ for processor in active_span_processor._span_processors: # pyright: ignore[reportPrivateUsage] if isinstance(processor, LightningSpanProcessor): # The legacy case for tracers without OTLP support. processors.append(f"{active_span_processor_cls} - {processor!r}") elif isinstance(processor, (SimpleSpanProcessor, BatchSpanProcessor)): processor_cls = processor.__class__.__name__ if isinstance(processor.span_exporter, LightningStoreOTLPExporter): # This should be the main path now. processors.append(f"{active_span_processor_cls} - {processor_cls} - {processor.span_exporter!r}") elif isinstance(processor.span_exporter, OTLPSpanExporter): # You need to be careful if the code goes into this path. endpoint = processor.span_exporter._endpoint # pyright: ignore[reportPrivateUsage] processors.append( f"{active_span_processor_cls} - {processor_cls} - " f"{processor.span_exporter.__class__.__name__}(endpoint={endpoint!r})" ) else: # Other cases like Console Span Exporter. processors.append( f"{active_span_processor_cls} - {processor_cls} - {processor.span_exporter.__class__.__name__}" ) else: processors.append(f"{active_span_processor_cls} - {processor.__class__.__name__}") logger.debug(f"Tracer provider: {tracer_provider!r}. Active span processors:") for processor in processors: logger.debug(" * " + processor) return tracer_provider def get_span_processors( tracer_provider: TracerProviderImpl, expected_type: Type[T_SpanProcessor] ) -> List[T_SpanProcessor]: """Get the span processors from the tracer provider. Args: tracer_provider: The tracer provider to get the span processors from. expected_type: The type of the span processors to get. Returns: A list of span processors of the expected type. """ processors: List[T_SpanProcessor] = [] for processor in tracer_provider._active_span_processor._span_processors: # pyright: ignore[reportPrivateUsage] if isinstance(processor, expected_type): processors.append(processor) return processors def get_tracer(use_active_span_processor: bool = True) -> trace_api.Tracer: """Resolve the OpenTelemetry tracer configured for Agent Lightning. Args: use_active_span_processor: Whether to use the active span processor. Returns: OpenTelemetry tracer tagged with the `agentlightning` instrumentation name. Raises: RuntimeError: If OpenTelemetry was not initialized before calling this helper. """ if hasattr(trace_api, "_TRACER_PROVIDER") and trace_api._TRACER_PROVIDER is None: # type: ignore[attr-defined] raise RuntimeError("Tracer is not initialized. Cannot emit a meaningful span.") tracer_provider = get_tracer_provider(inspect=True) # inspection is on by default if use_active_span_processor: return tracer_provider.get_tracer("agentlightning") else: filterwarnings( "ignore", message=r"You should use InstrumentationScope. Deprecated since version 1.11.1.", category=DeprecationWarning, module="opentelemetry.sdk.trace", ) return Tracer( tracer_provider.sampler, tracer_provider.resource, # We use an empty span processor to avoid emitting spans to the tracer SynchronousMultiSpanProcessor(), tracer_provider.id_generator, InstrumentationInfo("agentlightning", "", ""), # type: ignore SpanLimits(), InstrumentationScope( "agentlightning", "", "", {}, ), ) def make_tag_attributes(tags: List[str]) -> Dict[str, Any]: """Convert a list of tags into flattened attributes for span tagging. There is no syntax enforced for tags, they are just strings. For example: ```python ["gen_ai.model:gpt-4", "reward.extrinsic"] ``` """ return flatten_attributes({LightningSpanAttributes.TAG.value: tags}, expand_leaf_lists=True) def extract_tags_from_attributes(attributes: Dict[str, Any]) -> List[str]: """Extract tag attributes from flattened span attributes. Args: attributes: A dictionary of flattened span attributes. """ maybe_tag_list = filter_and_unflatten_attributes(attributes, LightningSpanAttributes.TAG.value) return TypeAdapter(List[str]).validate_python(maybe_tag_list) def make_link_attributes(links: Dict[str, str]) -> Dict[str, Any]: """Convert a dictionary of links into flattened attributes for span linking. Links example: ```python { "gen_ai.response.id": "response-123", "span_id": "abcd-efgh-ijkl", } ``` """ link_list: List[Dict[str, str]] = [] for key, value in links.items(): if not isinstance(value, str): # pyright: ignore[reportUnnecessaryIsInstance] raise ValueError(f"Link value must be a string, got {type(value)} for key '{key}'") link_list.append({LinkAttributes.KEY_MATCH.value: key, LinkAttributes.VALUE_MATCH.value: value}) return flatten_attributes({LightningSpanAttributes.LINK.value: link_list}, expand_leaf_lists=True) def query_linked_spans(spans: Sequence[T_SpanLike], links: List[LinkPydanticModel]) -> List[T_SpanLike]: """Query spans that are linked by the given link attributes. Args: spans: A sequence of spans to search. links: A list of link attributes to match. Returns: A list of spans that match the given link attributes. """ matched_spans: List[T_SpanLike] = [] for span in spans: span_attributes = span.attributes or {} is_match = True for link in links: # trace_id and span_id must be full match. if link.key_match == "trace_id": if isinstance(span, ReadableSpan): trace_id = trace_api.format_trace_id(span.context.trace_id) if span.context else None else: trace_id = span.trace_id if trace_id != link.value_match: is_match = False break elif link.key_match == "span_id": if isinstance(span, ReadableSpan): span_id = trace_api.format_span_id(span.context.span_id) if span.context else None else: span_id = span.span_id if span_id != link.value_match: is_match = False break else: attribute = span_attributes.get(link.key_match) # attributes must also be a full match currently. if attribute != link.value_match: is_match = False break if is_match: matched_spans.append(span) return matched_spans def extract_links_from_attributes(attributes: Dict[str, Any]) -> List[LinkPydanticModel]: """Extract link attributes from flattened span attributes. Args: attributes: A dictionary of flattened span attributes. """ maybe_link_list = filter_and_unflatten_attributes(attributes, LightningSpanAttributes.LINK.value) return TypeAdapter(List[LinkPydanticModel]).validate_python(maybe_link_list) def filter_attributes(attributes: Dict[str, Any], prefix: str) -> Dict[str, Any]: """Filter attributes that start with the given prefix. The attribute must start with `prefix.` or be exactly `prefix` to be included. Args: attributes: A dictionary of span attributes. prefix: The prefix to filter by. Returns: A dictionary of attributes that start with the given prefix. """ return {k: v for k, v in attributes.items() if k.startswith(prefix + ".") or k == prefix} def filter_and_unflatten_attributes(attributes: Dict[str, Any], prefix: str) -> Union[Dict[str, Any], List[Any]]: """Filter attributes that start with the given prefix and unflatten them. The prefix will be removed during unflattening. Args: attributes: A dictionary of span attributes. prefix: The prefix to filter by. Returns: A nested dictionary or list of attributes that start with the given prefix. """ filtered_attributes = filter_attributes(attributes, prefix) stripped_attributes: Dict[str, Any] = {} for k, v in filtered_attributes.items(): if k == prefix: raise ValueError(f"Cannot unflatten attribute with key exactly equal to prefix: {prefix}") else: stripped_key = k[len(prefix) + 1 :] # +1 to remove the dot stripped_attributes[stripped_key] = v return unflatten_attributes(stripped_attributes) def flatten_attributes( nested_data: Union[Dict[str, Any], List[Any]], *, expand_leaf_lists: bool = False ) -> Dict[str, Any]: """Flatten a nested dictionary or list into a flat dictionary with dotted keys. This function recursively traverses dictionaries and lists, producing a flat key-value mapping where nested paths are represented via dot-separated keys. Lists are indexed numerically. Example: >>> flatten_attributes({"a": {"b": 1, "c": [2, 3]}}, expand_leaf_lists=True) {"a.b": 1, "a.c.0": 2, "a.c.1": 3} Args: nested_data: A nested structure composed of dictionaries, lists, or primitive values. expand_leaf_lists: Whether to expand lists composed only of primitive values. When `False` (the default), lists of str/int/float/bool are treated as leaf values and stored without enumerating their indices. Returns: A flat dictionary mapping dotted-string paths to primitive values. """ flat: Dict[str, Any] = {} def _primitive_type(value: Any) -> Union[type[str], type[int], type[float], type[bool]]: if isinstance(value, bool): return bool if isinstance(value, int): return int if isinstance(value, float): return float return str def _walk(value: Any, prefix: str = "") -> None: if isinstance(value, dict): for k, v in cast(Dict[Any, Any], value).items(): if not isinstance(k, str): raise ValueError( f"Only string keys are supported in dictionaries, got '{k}' of type {type(k)} in {prefix}" ) new_prefix = f"{prefix}.{k}" if prefix else k _walk(v, new_prefix) elif isinstance(value, list): maybe_list = cast(List[Any], value) is_leaf_candidate = bool(maybe_list) and all( isinstance(item, (str, int, float, bool)) for item in maybe_list ) if not expand_leaf_lists and is_leaf_candidate and prefix: primitive_types = {_primitive_type(item) for item in maybe_list} if len(primitive_types) == 1: flat[prefix] = maybe_list return logger.warning( "List attribute '%s' contains mixed primitive types %s; expanding indexed keys instead.", prefix, primitive_types, ) for idx, item in enumerate(maybe_list): new_prefix = f"{prefix}.{idx}" if prefix else str(idx) _walk(item, new_prefix) else: flat[prefix] = value _walk(nested_data) return flat def unflatten_attributes(flat_data: Dict[str, Any]) -> Union[Dict[str, Any], List[Any]]: """Reconstruct a nested dictionary/list structure from a flat dictionary. Keys are dot-separated paths. Segments that are digit strings will only become list indices if *all* keys in that dict form a consecutive 0..n-1 range. Otherwise they remain dict keys. Example: >>> unflatten_attributes({"a.b": 1, "a.c.0": 2, "a.c.1": 3}) {"a": {"b": 1, "c": [2, 3]}} Args: flat_data: A dictionary whose keys are dot-separated paths and whose values are primitive data elements. Returns: A nested dictionary (and lists where appropriate) corresponding to the flattened structure. """ # 1) Build a pure dict tree first (no lists yet) root: Dict[str, Any] = {} for flat_key, value in flat_data.items(): parts = flat_key.split(".") curr: Dict[str, Any] = root for part in parts[:-1]: # Ensure intermediate node is a dict if part not in curr or not isinstance(curr[part], dict): curr[part] = {} curr = curr[part] # type: ignore[assignment] curr[parts[-1]] = value # 2) Recursively convert dicts-with-consecutive-numeric-keys into lists def convert(node: Union[Dict[str, Any], List[Any]]) -> Union[Dict[str, Any], List[Any]]: if isinstance(node, dict): # First convert children for k, v in list(node.items()): node[k] = convert(v) if not node: # empty dict stays dict return node # Check if keys are all numeric strings keys = list(node.keys()) if all(isinstance(k, str) and k.isdigit() for k in keys): # pyright: ignore[reportUnnecessaryIsInstance] indices = sorted(int(k) for k in keys) # Must be exactly 0..n-1 if indices == list(range(len(indices))): return [node[str(i)] for i in range(len(indices))] return node if isinstance(node, list): # pyright: ignore[reportUnnecessaryIsInstance] return [convert(v) for v in node] # Keep as is return node return convert(root) def sanitize_attribute_value(object: Any, force: bool = True) -> AttributeValue: """Sanitize an attribute value to be a valid OpenTelemetry attribute value.""" if isinstance(object, (str, int, float, bool)): return object if isinstance(object, list): try: return sanitize_list_attribute_sanity(cast(List[Any], object)) except ValueError as exc: logger.warning(f"Failed to sanitize list attribute. Fallback to JSON serialization: {exc}") try: # This include null, dict, etc. serialized = json.dumps(object, default=str if force else None) except (TypeError, ValueError) as exc: raise ValueError(f"Object must be JSON serializable, got: {type(cast(Any, object))}.") from exc return serialized def sanitize_attributes(attributes: Dict[str, Any], force: bool = True) -> Attributes: """Sanitize a dictionary of attributes to be a valid OpenTelemetry attributes. Args: attributes: A dictionary of attributes to sanitize. force: Whether to force sanitization even when the value is not JSON serializable. """ result: Attributes = {} for k, v in attributes.items(): try: result[k] = sanitize_attribute_value(v, force=force) except ValueError as exc: raise ValueError(f"Failed to sanitize attribute '{k}': {exc}") from exc return result def sanitize_list_attribute_sanity(maybe_list: List[Any]) -> AttributeValue: """Try to sanitize a list of attributes to be a valid OpenTelemetry attribute value. Raise error if the list contains multiple types of primitive values. """ if all(isinstance(item, str) for item in maybe_list): return list[str](maybe_list) if all(isinstance(item, bool) for item in maybe_list): return list[bool](maybe_list) if all(isinstance(item, (int, bool)) for item in maybe_list): return [int(item) for item in maybe_list] if all(isinstance(item, (float, int, bool)) for item in maybe_list): return [float(item) for item in maybe_list] list_types: List[Any] = [type(item) for item in maybe_list] raise ValueError(f"List must contain only one type of primitive values, got: {set(list_types)}.") def check_attributes_sanity(attributes: Dict[Any, Any]) -> None: """Check if a dictionary of attributes is a valid OpenTelemetry attributes.""" for k, v in attributes.items(): if not isinstance(k, str): raise ValueError(f"Attribute key must be a string, got {type(k)} for key '{k}'") if isinstance(v, list): try: sanitize_list_attribute_sanity(cast(List[Any], v)) except ValueError as exc: raise ValueError(f"Failed to sanitize list attribute '{k}': {exc}") from exc elif not isinstance(v, (str, int, float, bool)): raise ValueError( f"Attribute value must be a string, int, float, bool, or list of these, got {type(v)} for value '{v}'" ) def format_exception_attributes(exception: BaseException) -> Attributes: """Format an exception into a dictionary of attributes.""" stacktrace = "".join(traceback.format_exception(type(exception), exception, exception.__traceback__)) span_attributes: Attributes = { exception_attributes.EXCEPTION_TYPE: type(exception).__name__, exception_attributes.EXCEPTION_MESSAGE: str(exception), exception_attributes.EXCEPTION_ESCAPED: True, } if stacktrace.strip(): span_attributes[exception_attributes.EXCEPTION_STACKTRACE] = stacktrace return span_attributes