arena.utils.arena_telemetry

The ArenaTelemetry generates traces, metrics, and logs using OpenTelemetry (OTEL). It can export using OTEL's protocol (OTLP), send JSON OTEL spans to MQTT, or to the console.

The :envvar:ARENA_TELEMETRY environment variable enables the telemetry. The :envvar:OTLP_ENDPOINT environment variable defines the OTLP endpoint when OTLP is used.

TRACE_TOPIC_DFT = 'realm/ns/scene/t/traces'

class MQTTSpanExporter(opentelemetry.sdk.trace.export.SpanExporter):

Implementation of SpanExporter that sends spans to MQTT

MQTTSpanExporter( topic: Optional[str] = 'realm/ns/scene/t/traces', service_name: Optional[str] = None, formatter: Callable[[opentelemetry.sdk.trace.ReadableSpan], str] = <function MQTTSpanExporter.<lambda>>)

publish

publish_flush

topic

formatter

service_name

def export( self, spans: Sequence[opentelemetry.sdk.trace.ReadableSpan]) -> opentelemetry.sdk.trace.export.SpanExportResult:

Exports a batch of telemetry data.

Args: spans: The list of opentelemetry.trace.Span objects to be exported

Returns: The result of the export

def force_flush(self, timeout_millis: int = 30000) -> bool:

Hint to ensure that the export of any spans the exporter has received prior to the call to ForceFlush SHOULD be completed as soon as possible, preferably before returning from this method.

def shutdown(self) -> None:

Shuts down the exporter.

Called when the SDK is shut down.

class ArenaTelemetry:

Implementation of ARENA telemetry. According to :envvar:ARENA_TELEMETRY, exports using OTLP, send JSON OTEL spans to MQTT, or to the console.

ArenaTelemetry(name='scripts/pdoc/make-pdoc.py', id=None)

Return an ArenaTelemetry using given service name and id Provides utility calls that wrap open telemetry functionality to start spans, log events, and other.

Creates a parent span for all the spans related to the program.

Args: name: name of the service used with the telemetry backend id: additional id used with the telemetry backend

parent_span: opentelemetry.sdk.trace.Span = None

tracer

parent_span_ctx

def exit(self, error_msg=None, print_msg=True):

Record exit status on error

def start_span(self, name, **kwargs):

Wrapper to otel start_as_current_span; force context to be parent span

def start_process_msg_span(self, obj_id, action, **kwargs):

Wrapper to otel start_as_current_span to start a process message span; force context to be parent span

def start_publish_span(self, obj_id, action, type, **kwargs):

Wrapper to otel start_as_current_span to start a process message span; force context to be parent span

def add_event(self, name, span=None, print_msg=True, **kwargs):

Add event to given or current span

def set_error(self, error_msg, span=None, print_msg=True):

Set error on given or current span