Module madsci.common.otel.tracing - MADSci Documentation

OpenTelemetry tracing decorators and context managers for MADSci.

This module provides convenient decorators and context managers for creating OpenTelemetry spans in MADSci code, following the same patterns as the event client and ownership context decorators.

Functions¶

add_span_event(name: str, attributes: dict[str, typing.Any] | None = None, span: opentelemetry.trace.span.Span | None = None) ‑> None

Add an event to the current or provided span.

Args: name: The event name. attributes: Optional event attributes. span: The span to add the event to. Uses current span if not provided.

Example: add_span_event(“cache_hit”, {“key”: cache_key}) add_span_event(“validation_complete”, {“items_validated”: 100})

get_current_span() ‑> opentelemetry.trace.span.Span

Get the current active span.

Returns: The current Span, or a non-recording span if none is active.

get_tracer(name: str | None = None) ‑> opentelemetry.trace.Tracer

Get an OpenTelemetry tracer.

Args: name: The tracer name. Defaults to “madsci”.

Returns: An OpenTelemetry Tracer instance.

is_span_recording() ‑> bool

Check if the current span is recording.

Returns: True if there’s an active recording span, False otherwise.

safe_span_context(name: str, *, tracer_name: str | None = None, kind: opentelemetry.trace.SpanKind = SpanKind.INTERNAL, attributes: dict[str, typing.Any] | None = None) ‑> Generator[opentelemetry.trace.span.Span | None, None, None]

Context manager that creates a span if OTEL is configured, otherwise no-op.

This is useful for code that may run with or without OTEL configured. It never raises exceptions related to tracing.

Args: name: The name of the span. tracer_name: Optional tracer name. Defaults to “madsci”. kind: The span kind. attributes: Optional dictionary of span attributes.

Yields: The created Span object, or None if tracing is not available.

Example: with safe_span_context(“optional_trace”) as span: if span: span.set_attribute(“custom”, “value”) do_work()

set_span_attributes(attributes: dict[str, typing.Any], span: opentelemetry.trace.span.Span | None = None) ‑> None

Set multiple attributes on the current or provided span.

Args: attributes: Dictionary of attributes to set. span: The span to set attributes on. Uses current span if not provided.

Example: set_span_attributes({ “user.id”: user_id, “request.size”: len(data), “cache.enabled”: True, })

set_span_error(span: opentelemetry.trace.span.Span | None = None, exception: BaseException | None = None, message: str | None = None) ‑> None

Set the current or provided span to error status.

Args: span: The span to mark as error. Uses current span if not provided. exception: Optional exception to record. message: Optional error message.

Example: try: risky_operation() except Exception as e: set_span_error(exception=e, message=“Operation failed”) raise

span_context(name: str, *, tracer_name: str | None = None, kind: opentelemetry.trace.SpanKind = SpanKind.INTERNAL, attributes: dict[str, typing.Any] | None = None, record_exception: bool = True, set_status_on_exception: bool = True) ‑> Generator[opentelemetry.trace.span.Span, None, None]

Context manager that creates an OpenTelemetry span.

This is a convenience wrapper around tracer.start_as_current_span() that handles the common case of creating a span with attributes and automatic exception recording.

Args: name: The name of the span. tracer_name: Optional tracer name. Defaults to “madsci”. kind: The span kind (INTERNAL, SERVER, CLIENT, PRODUCER, CONSUMER). attributes: Optional dictionary of span attributes. record_exception: If True, record exceptions as span events. set_status_on_exception: If True, set span status to ERROR on exception.

Yields: The created Span object.

Example: with span_context(“process_data”, attributes={“data.size”: 100}) as span: result = process(data) span.set_attribute(“result.count”, len(result))

  with span_context("fetch_resource", kind=SpanKind.CLIENT) as span:
      response = requests.get(url)
      span.set_attribute("http.status_code", response.status_code)

traced_class(name: str | None = None, *, tracer_name: str | None = None, kind: opentelemetry.trace.SpanKind = SpanKind.INTERNAL, attributes: dict[str, typing.Any] | None = None, record_exception: bool = True, set_status_on_exception: bool = True) ‑> Callable[[type], type]

Class decorator that wraps all public methods in OpenTelemetry spans.

This decorator modifies a class so that all public methods (not starting with underscore) are automatically traced. The class gains a ‘current_span’ property that returns the active span.

Args: name: Base name for spans. Defaults to the class name. Method spans will be named “{class_name}.{method_name}”. tracer_name: Optional tracer name. Defaults to “madsci”. kind: The span kind for method spans. attributes: Optional dictionary of span attributes to add to all spans. record_exception: If True, record exceptions as span events. set_status_on_exception: If True, set span status to ERROR on exception.

Returns: A class decorator.

Example: traced_class(attributes={“component”: “data_processor”}) class DataProcessor: def process(self, data): # This method is automatically traced self.current_span.set_attribute(“data.size”, len(data)) return transform(data)

      def _helper(self, x):  # Private methods are NOT traced
          return x * 2

  @traced_class(name="CustomWorker")
  class Worker:
      def get_span_attributes(self) -> dict:
          # Return instance-specific attributes
          return {"worker.id": self.worker_id}

      def work(self):
          # Span includes both class-level and instance attributes
          self.current_span.add_event("started")
          do_work()

with_span(func: Callable[~P, typing.Any] | None = None, *, name: str | None = None, tracer_name: str | None = None, kind: opentelemetry.trace.SpanKind = SpanKind.INTERNAL, attributes: dict[str, typing.Any] | None = None, record_exception: bool = True, set_status_on_exception: bool = True) ‑> Callable[~P, typing.Any] | Callable[[Callable[~P, typing.Any]], Callable[~P, typing.Any]]

Decorator that wraps a function in an OpenTelemetry span.

This decorator creates a span for each function invocation, automatically recording the function name, arguments (optionally), and any exceptions. Works with both sync and async functions.

Can be used with or without arguments: with_span def my_function(): ...

  @with_span(name="custom_span", attributes={"component": "processor"})
  def my_function(): ...

Args: func: The function to wrap (when used without parentheses). name: Custom span name. Defaults to the function’s qualified name. tracer_name: Optional tracer name. Defaults to “madsci”. kind: The span kind (INTERNAL, SERVER, CLIENT, PRODUCER, CONSUMER). attributes: Optional dictionary of span attributes to add. record_exception: If True, record exceptions as span events. set_status_on_exception: If True, set span status to ERROR on exception.

Returns: The decorated function that runs within a span.

Example: with_span def process_data(data): return transform(data)

  @with_span(name="fetch_user", kind=SpanKind.CLIENT)
  def get_user(user_id: str):
      return api.fetch(user_id)

  @with_span(attributes={"workflow.type": "batch"})
  async def run_batch_workflow():
      await process_batch()