testplan.common.utils.observability package

Submodules

testplan.common.utils.observability.tracing module

OpenTelemetry observability integration for Testplan.

This module provides tracing capabilities using OpenTelemetry to monitor and analyze test execution across distributed systems.

class testplan.common.utils.observability.tracing.RootTraceIdGenerator(tracing_instance: Tracing)

Bases: object

Custom ID generator that uses a root trace ID when available. Falls back to random ID generation when no root trace ID is set.

generate_span_id() → int

Generate a span ID using random generation.

Returns:

Span ID as integer

Return type:

int

generate_trace_id() → int

Generate a trace ID, using the root trace ID if available.

Returns:

Trace ID as integer

Return type:

int

class testplan.common.utils.observability.tracing.Tracing

Bases: Loggable

Global tracing object to handle OpenTelemetry tracing.

This class manages the lifecycle of OpenTelemetry spans for Testplan execution, providing both automatic and manual span creation capabilities.

The class is instantiated as a singleton (tracing) and should be accessed through that global instance.

Required Environment Variables:

• OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: OTLP traces endpoint URL

• OTEL_RESOURCE_ATTRIBUTES: Resource attributes as comma-separated key=value pairs

Additional Required Environment Variables (gRPC exporter only):

• OTEL_EXPORTER_OTLP_CERTIFICATE: Path to CA certificate

• OTEL_EXPORTER_OTLP_CLIENT_KEY: Path to client private key

• OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE: Path to client certificate

Optional Environment Variables:

• OTEL_BSP_SCHEDULE_DELAY: Batch span processor delay in milliseconds (default: 200)

attach_to_root_context() → Iterator[None]

Context manager to attach to root trace context. Context in opentelemetry is thread-local, so to attach the logs to the root trace, we need to attach to the root context when we start a new thread (in TestRunner and Test subclasses)

conditional_span(name: str, context: Dict[str, Any] | None = None, condition: bool = True, **kwargs: Any) → Iterator[Span | None]

Utility function to create a span conditionally.

Parameters:

• name (str) – Name of the span

• context (Optional[Dict[str, Any]]) – Optional trace context for span propagation, defaults to current parent span if not provided

• condition (bool) – Condition to determine whether to create the span

• kwargs (Any) – Additional span attributes as keyword arguments

Yield:

Span object if tracing is enabled and condition is met, None otherwise

Return type:

Iterator[Optional[Span]]

Example:

>>> with tracing.conditional_span("operation", condition=True, example_attr="value1") as span:
...     # Perform operation
...     pass

end_span(span: Span, end_time: int | None = None) → None

Manually end a span started with start_span().

Parameters:

• span (Span) – Span to end

• end_time (Optional[int]) – Optional end time in nanoseconds since epoch

Example:

>>> tracing.start_span("operation")
>>> # Perform operation
>>> tracing.end_span("operation")

force_flush() → None

Force the current span processor to export all ended spans.

set_span_as_failed(span: Span | None = None, description: str | None = None) → None

Mark a span as failed with an error status.

Parameters:

• span (Optional[Span]) – Span to mark as failed, if None, do nothing

• description (Optional[str]) – Optional error description

Example:

>>> with tracing.span("validation") as span:
...     if not is_valid:
...         tracing.set_span_as_failed(
...             span,
...             description="Validation failed"
...         )

set_span_attrs(span: Span | None = None, **kwargs: Any) → None

Set attributes on a span.

Parameters:

• span (Optional[Span]) – Span to set attributes on, if None, do nothing

• kwargs (Any) – Attributes as keyword arguments

Example:

>>> with tracing.span("api_call") as span:
...     response = call_api()
...     tracing.set_span_attrs(
...         span,
...         status_code=response.status_code,
...         response_time_ms=response.elapsed
...     )

span(name: str, context: Dict[str, Any] | None = None, **kwargs: Any) → Iterator[Span | None]

Create a span using context manager (recommended approach).

Parameters:

• name (str) – Name of the span

• context (Optional[Dict[str, Any]]) – Optional trace context for span propagation, defaults to current parent span if not provided

• kwargs (Any) – Additional span attributes as keyword arguments

Yield:

Span object if tracing is enabled, None otherwise

Return type:

Iterator[Optional[Span]]

Example:

>>> with tracing.span("operation", example_attr="value1") as span:
...     # Perform operation
...     pass

start_span(span_name: str, context: Dict[str, Any] | None = None, start_time: int | None = None, **kwargs: Any) → Span | None

Manually start a span (use when context manager is not suitable).

The span must be explicitly ended using end_span() else it will not be exported.

Parameters:

• span_name (str) – Name of the span

• context (Optional[Dict[str, Any]]) – Optional trace context for span propagation

• start_time (Optional[int]) – Optional start time in nanoseconds since epoch

• kwargs (Any) – Additional span attributes as keyword arguments

Returns:

Span object if tracing is enabled, None otherwise

Return type:

Optional[Span]

Example:

>>> span = tracing.start_span("operation", task_id="123")
>>> try:
...     # Perform operation
...     pass
... finally:
...     tracing.end_span("operation")

trace(func: Callable[[...], Any]) → Callable[[...], Any]

A decorator that wraps a function with a span.

Example:

>>> @tracing.trace
... def my_function():
...     # Function implementation
...     pass

testplan.common.utils.observability.tracing.tracing = <testplan.common.utils.observability.tracing.Tracing object>

Global tracing instance for Testplan observability

testplan.common.utils.observability.logging module

class testplan.common.utils.observability.logging.OtelLogging

Bases: Loggable

Global logging object to handle OpenTelemetry logging integration.

This class manages the lifecycle of OpenTelemetry logging for Testplan execution, enabling log collection and correlation with distributed traces. It attaches to the Testplan logger and exports logs to Grafana Loki.

The class is instantiated as a singleton (otel_logging) and should be accessed through that global instance.

Note

This class must be run alongside the tracing setup (tracing) to ensure logs are properly correlated with traces via trace_id and span_id.

Required Environment Variables:

• OTEL_EXPORTER_LOKI_ENDPOINT: Loki endpoint URL

• OTEL_EXPORTER_OTLP_HEADERS: OTLP headers

• OTEL_EXPORTER_OTLP_CERTIFICATE: Path to CA certificate

• OTEL_EXPORTER_OTLP_CLIENT_KEY: Path to client private key

• OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE: Path to client certificate

force_flush()

Force the current log processor to export all recorded logs.