opentelemetry

OpenTelemetry Task Metrics Setup.

Module

Functions

get_installation_uuid

def get_installation_uuid(storage_path: Optional[Path] = None) ‑> str:

Return the installation UUID.

Uses BITFOUNT_STORAGE_PATH by default.

get_task_meter

def get_task_meter() ‑> TaskMeter:

Fetch global TaskMeter.

setup_opentelemetry_metrics

def setup_opentelemetry_metrics(*, session: BitfountSession) ‑> None:

Setup Open Telemetry metrics exporter and TaskMeter.

Global variables

• task_meter : Optional[TaskMeter] - The OpenTelemetry Python SDK doesn't seem to have a way to unregister callbacks, (https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#asynchronous-gauge-operations) so to avoid creating a new gauge and registering a new callback that will have a reference kept my OpenTelemetry (memory leak) every time we instantiate a Protocol we have this global Meter that can be used by all Protocol instances.

Classes

BitfountOLTPMetricsExporter

class BitfountOLTPMetricsExporter(    *, session: BitfountSession, endpoint: str, insecure: bool,):

OTLPMetricExporter that sends Bitfount session headers.

The BitfountSession message_service_metadata property may change over time so we need to ensure we have it fresh when we export metrics.

Ancestors

• opentelemetry.exporter.otlp.proto.grpc.metric_exporter.OTLPMetricExporter
• opentelemetry.sdk.metrics._internal.export.MetricExporter
• opentelemetry.exporter.otlp.proto.grpc.exporter.OTLPExporterMixin
• abc.ABC
• typing.Generic
• opentelemetry.exporter.otlp.proto.common._internal.metrics_encoder.OTLPMetricExporterMixin

TaskMeter

class TaskMeter():

Meter that can be used to report task metrics.

Methods

---

submit_algorithm_records_per_class_returned

def submit_algorithm_records_per_class_returned(    self,    *,    records_count_per_class: dict[str, int],    task_id: str,    algorithm: Union[_BaseAlgorithm, str],    protocol_batch_num: Optional[int],    project_id: Optional[str] = None,) ‑> None:

Submit a algorithm_records_returned observation per class.

submit_algorithm_records_returned

def submit_algorithm_records_returned(    self,    *,    records_count: int,    task_id: str,    algorithm: Union[_BaseAlgorithm, str],    protocol_batch_num: Optional[int] = None,    project_id: Optional[str] = None,) ‑> None:

Submit algorithm_records_returned observation.

---