tessl/pypi-opentelemetry-api
OpenTelemetry Python API providing core abstractions for distributed tracing, metrics collection, and context propagation
—
Pending
metrics.mddocs/
Metrics Collection
Comprehensive metrics collection capabilities for recording measurements in distributed systems. OpenTelemetry metrics provide both synchronous and asynchronous instruments for counters, gauges, histograms, and observable metrics with rich dimensional attributes.
Capabilities
Meter Creation and Management
Get meters for creating measurement instruments with proper instrumentation scope identification.
def get_meter(
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
meter_provider: Optional[MeterProvider] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Meter:
"""
Returns a Meter for use by the given instrumentation library.
Parameters:
- instrumenting_module_name: The name of the instrumentation scope
- instrumenting_library_version: Optional version of the instrumentation library
- meter_provider: Optional specific MeterProvider to use
- schema_url: Optional Schema URL of the emitted telemetry
- attributes: Optional attributes of the emitted telemetry
Returns:
Meter instance for creating instruments
"""
def get_meter_provider() -> MeterProvider:
"""Gets the current global MeterProvider object."""
def set_meter_provider(meter_provider: MeterProvider) -> None:
"""
Sets the current global MeterProvider object.
This can only be done once, a warning will be logged if any further attempt is made.
"""Synchronous Instruments
Create instruments for immediate measurement recording during code execution.
class Meter(ABC):
def create_counter(
self,
name: str,
unit: str = "",
description: str = "",
) -> Counter:
"""
Creates a Counter instrument for recording monotonically increasing values.
Parameters:
- name: The name of the instrument
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
Counter instrument for recording increasing values
"""
def create_up_down_counter(
self,
name: str,
unit: str = "",
description: str = "",
) -> UpDownCounter:
"""
Creates an UpDownCounter instrument for values that can increase and decrease.
Parameters:
- name: The name of the instrument
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
UpDownCounter instrument for recording values that go up and down
"""
def create_histogram(
self,
name: str,
unit: str = "",
description: str = "",
) -> Histogram:
"""
Creates a Histogram instrument for recording distributions of values.
Parameters:
- name: The name of the instrument
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
Histogram instrument for recording value distributions
"""
def create_gauge(
self,
name: str,
unit: str = "",
description: str = "",
) -> Gauge:
"""
Creates a Gauge instrument for recording current values.
Parameters:
- name: The name of the instrument
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
Gauge instrument for recording current values
"""Counter Operations
Record monotonically increasing values like request counts, bytes processed, or errors.
class Counter(Instrument):
"""Synchronous counter instrument for monotonically increasing values."""
def add(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""
Records an increment to the counter.
Parameters:
- amount: The increment amount (must be non-negative)
- attributes: Optional attributes to associate with the measurement
"""
class NoOpCounter(Counter):
"""No-op implementation of Counter."""
def add(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""No-op add implementation."""UpDownCounter Operations
Record values that can both increase and decrease like queue sizes, active connections, or memory usage.
class UpDownCounter(Instrument):
"""Synchronous up-down counter for values that can increase and decrease."""
def add(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""
Records a change to the up-down counter.
Parameters:
- amount: The change amount (can be positive or negative)
- attributes: Optional attributes to associate with the measurement
"""
class NoOpUpDownCounter(UpDownCounter):
"""No-op implementation of UpDownCounter."""
def add(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""No-op add implementation."""Histogram Operations
Record distributions of values like request durations, response sizes, or temperature readings.
class Histogram(Instrument):
"""Synchronous histogram instrument for recording value distributions."""
def record(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""
Records a measurement to the histogram.
Parameters:
- amount: The value to record
- attributes: Optional attributes to associate with the measurement
"""
class NoOpHistogram(Histogram):
"""No-op implementation of Histogram."""
def record(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""No-op record implementation."""Gauge Operations
Record current values like CPU usage, memory consumption, or current temperature.
class Gauge(Instrument):
"""Synchronous gauge instrument for recording current values."""
def set(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""
Records a measurement to the gauge.
Parameters:
- amount: The current value to record
- attributes: Optional attributes to associate with the measurement
"""
class NoOpGauge(Gauge):
"""No-op implementation of Gauge."""
def set(
self,
amount: Union[int, float],
attributes: Attributes = None,
) -> None:
"""No-op set implementation."""Asynchronous Instruments
Create callback-based instruments for periodic measurement collection.
class Meter(ABC):
def create_observable_counter(
self,
name: str,
callbacks: Optional[Sequence[CallbackT]] = None,
unit: str = "",
description: str = "",
) -> ObservableCounter:
"""
Creates an ObservableCounter for callback-based monotonic measurements.
Parameters:
- name: The name of the instrument
- callbacks: Optional sequence of callback functions
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
ObservableCounter instrument for callback-based counting
"""
def create_observable_up_down_counter(
self,
name: str,
callbacks: Optional[Sequence[CallbackT]] = None,
unit: str = "",
description: str = "",
) -> ObservableUpDownCounter:
"""
Creates an ObservableUpDownCounter for callback-based bidirectional measurements.
Parameters:
- name: The name of the instrument
- callbacks: Optional sequence of callback functions
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
ObservableUpDownCounter instrument for callback-based up-down counting
"""
def create_observable_gauge(
self,
name: str,
callbacks: Optional[Sequence[CallbackT]] = None,
unit: str = "",
description: str = "",
) -> ObservableGauge:
"""
Creates an ObservableGauge for callback-based current value measurements.
Parameters:
- name: The name of the instrument
- callbacks: Optional sequence of callback functions
- unit: Optional unit of measurement
- description: Optional description of what the instrument measures
Returns:
ObservableGauge instrument for callback-based gauge measurements
"""Observable Instrument Types
Asynchronous instruments that use callbacks for measurement collection.
class ObservableCounter(Asynchronous):
"""Asynchronous counter instrument using callbacks."""
class NoOpObservableCounter(ObservableCounter):
"""No-op implementation of ObservableCounter."""
class ObservableUpDownCounter(Asynchronous):
"""Asynchronous up-down counter instrument using callbacks."""
class NoOpObservableUpDownCounter(ObservableUpDownCounter):
"""No-op implementation of ObservableUpDownCounter."""
class ObservableGauge(Asynchronous):
"""Asynchronous gauge instrument using callbacks."""
class NoOpObservableGauge(ObservableGauge):
"""No-op implementation of ObservableGauge."""Callback System
Configure callback functions for asynchronous measurement collection.
class CallbackOptions:
"""Options for instrument callbacks."""
def __init__(self, timeout_millis: float = 10000) -> None:
"""
Parameters:
- timeout_millis: Timeout for callback execution in milliseconds
"""
class Observation:
"""Single measurement observation from a callback."""
def __init__(
self,
value: Union[int, float],
attributes: Attributes = None,
) -> None:
"""
Parameters:
- value: The observed measurement value
- attributes: Optional attributes for the observation
"""
@property
def value(self) -> Union[int, float]:
"""Returns the observation value."""
@property
def attributes(self) -> Attributes:
"""Returns the observation attributes."""
CallbackT = Callable[[CallbackOptions], Iterable[Observation]]Base Instrument Classes
Abstract base classes defining the instrument hierarchy.
class Instrument(ABC):
"""Base class for all measurement instruments."""
@property
def name(self) -> str:
"""Returns the instrument name."""
@property
def description(self) -> str:
"""Returns the instrument description."""
@property
def unit(self) -> str:
"""Returns the instrument unit."""
class Synchronous(Instrument):
"""Base class for synchronous instruments that record measurements immediately."""
class Asynchronous(Instrument):
"""Base class for asynchronous instruments that use callbacks for measurement."""Provider Implementations
Provider classes for different deployment scenarios.
class MeterProvider(ABC):
"""Abstract base class for meter providers."""
def get_meter(
self,
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Meter:
"""Returns a Meter for use by the given instrumentation library."""
class NoOpMeterProvider(MeterProvider):
"""The default MeterProvider, used when no implementation is available."""
def get_meter(
self,
instrumenting_module_name: str,
instrumenting_library_version: Optional[str] = None,
schema_url: Optional[str] = None,
attributes: Optional[Attributes] = None,
) -> Meter:
"""Returns a no-op meter."""
class NoOpMeter(Meter):
"""No-op implementation of Meter."""
def create_counter(self, name: str, **kwargs) -> Counter:
"""Returns a no-op counter."""
def create_up_down_counter(self, name: str, **kwargs) -> UpDownCounter:
"""Returns a no-op up-down counter."""
def create_histogram(self, name: str, **kwargs) -> Histogram:
"""Returns a no-op histogram."""
def create_gauge(self, name: str, **kwargs) -> Gauge:
"""Returns a no-op gauge."""
def create_observable_counter(self, name: str, **kwargs) -> ObservableCounter:
"""Returns a no-op observable counter."""
def create_observable_up_down_counter(self, name: str, **kwargs) -> ObservableUpDownCounter:
"""Returns a no-op observable up-down counter."""
def create_observable_gauge(self, name: str, **kwargs) -> ObservableGauge:
"""Returns a no-op observable gauge."""Usage Examples
Basic Counter Usage
from opentelemetry import metrics
# Get a meter
meter = metrics.get_meter(__name__)
# Create a counter for tracking requests
request_counter = meter.create_counter(
name="http_requests_total",
description="Total number of HTTP requests",
unit="1"
)
# Record measurements
request_counter.add(1, {"method": "GET", "endpoint": "/api/users"})
request_counter.add(1, {"method": "POST", "endpoint": "/api/users"})Histogram for Latency Tracking
from opentelemetry import metrics
import time
meter = metrics.get_meter(__name__)
# Create a histogram for request duration
duration_histogram = meter.create_histogram(
name="http_request_duration_seconds",
description="HTTP request duration in seconds",
unit="s"
)
# Track request duration
start_time = time.time()
# ... handle request ...
duration = time.time() - start_time
duration_histogram.record(duration, {
"method": "GET",
"status_code": "200",
"endpoint": "/api/data"
})Observable Metrics with Callbacks
from opentelemetry import metrics
from opentelemetry.metrics import CallbackOptions, Observation
import psutil
meter = metrics.get_meter(__name__)
def get_cpu_usage(options: CallbackOptions):
"""Callback function to collect CPU usage."""
cpu_percent = psutil.cpu_percent()
return [Observation(cpu_percent, {"cpu": "all"})]
# Create observable gauge with callback
cpu_gauge = meter.create_observable_gauge(
name="system_cpu_usage_percent",
description="Current CPU usage percentage",
unit="%",
callbacks=[get_cpu_usage]
)Up-Down Counter for Active Connections
from opentelemetry import metrics
meter = metrics.get_meter(__name__)
# Create up-down counter for active connections
active_connections = meter.create_up_down_counter(
name="active_connections",
description="Number of active connections",
unit="1"
)
# Track connection lifecycle
def on_connection_open():
active_connections.add(1, {"protocol": "http"})
def on_connection_close():
active_connections.add(-1, {"protocol": "http"})Install with Tessl CLI
npx tessl i tessl/pypi-opentelemetry-api