otel-metrics

Reusable OpenTelemetry library: simplified wrappers around OTel metric instruments, plus an OTLP tracing pipeline.

Install

pip install otel-metrics

Or from a built wheel:

pip install dist/otel_metrics-0.4.0-py3-none-any.whl

Usage

from enum import auto

from otel_metrics import MetricName, OtelConfig, setup_metrics

# Define your metric names
class MyMetrics(MetricName):
    REQUEST_COUNT = auto()
    REQUEST_DURATION_MS = auto()
    ACTIVE_CONNECTIONS = auto()

# Configure and initialize
config = OtelConfig(endpoint="http://localhost:4317", service_name="my-service")

# setup_metrics returns a context manager that flushes and shuts down on exit
with setup_metrics(config) as metrics:
    counter = metrics.counter(MyMetrics.REQUEST_COUNT, description="Total requests")
    timer = metrics.timer(MyMetrics.REQUEST_DURATION_MS, description="Request latency")

    counter.inc()

Calling setup_metrics again in the same process reuses the provider it already created. Pass set_global=False for isolated setups, or inject a provider with setup_metrics(config, meter_provider=provider) and the library will not shut it down.

Timing operations

with timer.time(attributes={"route": "/orders"}):
    do_work()

For explicit control, record a TimestampNS yourself:

from otel_metrics import TimestampNS

start = TimestampNS.now()
# ... do work ...
timer.record(start)

Up/down counters

Use an up/down counter for values that rise and fall, such as active connections:

active = metrics.up_down_counter(
    MyMetrics.ACTIVE_CONNECTIONS, description="Open connections"
)

active.inc()
active.dec()

inc adds and dec subtracts, both taking an optional amount and attributes. Negative amounts raise ValueError for counters and histograms.

Gauges

A gauge keeps one value per attribute set:

gauge = metrics.gauge(MyMetrics.QUEUE_DEPTH, description="Queued jobs")
gauge.set(12, attributes={"queue": "default"})
gauge.set(3, attributes={"queue": "billing"})
gauge.get(attributes={"queue": "default"})  # 12.0
gauge.remove(attributes={"queue": "billing"})

A gauge emits nothing until its first set.

Configuration

OtelConfig is a pydantic-settings model. Values resolve from constructor args or environment variables:

Supported protocols: grpc, http/protobuf.

metric_prefix is optional. When set, instrument names are prefixed with it (myapp.request_count). It is not derived from service_name automatically, because the OTel resource already carries service.name.

headers accepts either a mapping or the OTLP key=value,key2=value2 string form, and is forwarded to both the metrics and traces exporters. Values may be URL-encoded per the OTLP spec, but plain values are accepted too.

Header values are held as pydantic.SecretStr, because they usually carry an API key. Printing or dumping the config shows the header names but masks the values:

config = OtelConfig(
    endpoint="https://api.braintrust.dev/otel",
    service_name="my-service",
    headers="Authorization=Bearer sk-live-123",
)

print(config.model_dump_json())
# {"endpoint":"...","headers":{"authorization":"**********"}}

config.resolved_headers()
# {'authorization': 'Bearer sk-live-123'}

Call resolved_headers() when you need the real values. Reading config.headers gives SecretStr objects, not plain strings.

tags are optional OTel resource attributes attached to every metric and span. They accept a mapping or the OTLP key=value,key2=value2 string form (via OTEL_RESOURCE_ATTRIBUTES or the tags field), parsed like headers:

config = OtelConfig(
    endpoint="https://api.braintrust.dev/otel",
    service_name="my-service",
    tags="env=prod,region=us-east-1",
)

The explicit service_name always wins over a service.name tag.

Tracing

from otel_metrics import OtelConfig, setup_telemetry

config = OtelConfig(
    endpoint="https://api.braintrust.dev/otel",
    service_name="my-service",
    protocol="http/protobuf",
    headers={"Authorization": "Bearer <key>", "x-bt-parent": "project_id:<id>"},
)
tracer = setup_telemetry(config)

with tracer.start_as_current_span("my-operation"):
    ...

setup_telemetry sets the process-global tracer provider and reuses it on later calls. Pass set_global=False for isolated setups, or inject a provider with tracer_provider=. Use shutdown_telemetry() (or force_flush_telemetry()) for explicit lifecycle control. Use build_test_tracer() for a non-exporting tracer in unit tests.

Instrumentors

Pass any BaseInstrumentor as a trailing argument and it is attached to the provider built here -- no need to reach for the global one:

from opentelemetry.instrumentation.langchain import LangchainInstrumentor

setup_telemetry(config, LangchainInstrumentor())

Several are fine, and an existing list can be splatted:

setup_telemetry(config, LangchainInstrumentor(), RequestsInstrumentor())
setup_telemetry(config, *instrumentors)

Testing

Use build_test_metrics() to get a no-op Metrics instance for unit tests:

from otel_metrics import build_test_metrics

metrics = build_test_metrics()

Development

uv sync
uv run pytest
uv run ruff check .

Build

uv build

Produces a wheel and an sdist in dist/.