opentelemetry._metrics package

Warning

OpenTelemetry Python metrics are in an experimental state. The APIs within opentelemetry._metrics are subject to change in minor/patch releases and make no backward compatability guarantees at this time.

Once metrics become stable, this package will be be renamed to opentelemetry.metrics.

Module contents

This module provides abstract and concrete (but noop) classes that can be used to generate metrics.

class opentelemetry._metrics.MeterProvider[source]

Bases: abc.ABC

abstract get_meter(name, version=None, schema_url=None)[source]
Return type

Meter

class opentelemetry._metrics.NoOpMeterProvider[source]

Bases: opentelemetry._metrics.MeterProvider

get_meter(name, version=None, schema_url=None)[source]
Return type

Meter

class opentelemetry._metrics.Meter(name, version=None, schema_url=None)[source]

Bases: abc.ABC

property name
property version
property schema_url
abstract create_counter(name, unit='', description='')[source]
Return type

Counter

abstract create_up_down_counter(name, unit='', description='')[source]
Return type

UpDownCounter

abstract create_observable_counter(name, callback, unit='', description='')[source]

Creates an observable counter instrument

An observable counter observes a monotonically increasing count by calling a provided callback which returns multiple Measurement.

For example, an observable counter could be used to report system CPU time periodically. Here is a basic implementation:

def cpu_time_callback() -> Iterable[Measurement]:
    measurements = []
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if not line.startswith("cpu"): break
            cpu, *states = line.split()
            measurements.append(Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
            measurements.append(Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
            measurements.append(Measurement(int(states[2]) // 100, {"cpu": cpu, "state": "system"}))
            # ... other states
    return measurements

meter.create_observable_counter(
    "system.cpu.time",
    callback=cpu_time_callback,
    unit="s",
    description="CPU time"
)

To reduce memory usage, you can use generator callbacks instead of building the full list:

def cpu_time_callback() -> Iterable[Measurement]:
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if not line.startswith("cpu"): break
            cpu, *states = line.split()
            yield Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"})
            yield Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"})
            # ... other states

Alternatively, you can pass a generator directly instead of a callback, which should return iterables of Measurement:

def cpu_time_callback(states_to_include: set[str]) -> Iterable[Iterable[Measurement]]:
    while True:
        measurements = []
        with open("/proc/stat") as procstat:
            procstat.readline()  # skip the first line
            for line in procstat:
                if not line.startswith("cpu"): break
                cpu, *states = line.split()
                if "user" in states_to_include:
                    measurements.append(Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
                if "nice" in states_to_include:
                    measurements.append(Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
                # ... other states
        yield measurements

meter.create_observable_counter(
    "system.cpu.time",
    callback=cpu_time_callback({"user", "system"}),
    unit="s",
    description="CPU time"
)
Parameters
  • name – The name of the instrument to be created

  • callback – A callback that returns an iterable of Measurement. Alternatively, can be a generator that yields iterables of Measurement.

  • unit – The unit for measurements this instrument reports. For example, By for bytes. UCUM units are recommended.

  • description – A description for this instrument and what it measures.

Return type

ObservableCounter

abstract create_histogram(name, unit='', description='')[source]
Return type

Histogram

abstract create_observable_gauge(name, callback, unit='', description='')[source]
Return type

ObservableGauge

abstract create_observable_up_down_counter(name, callback, unit='', description='')[source]
Return type

ObservableUpDownCounter

class opentelemetry._metrics.NoOpMeter(name, version=None, schema_url=None)[source]

Bases: opentelemetry._metrics.Meter

create_counter(name, unit='', description='')[source]
Return type

Counter

create_up_down_counter(name, unit='', description='')[source]
Return type

UpDownCounter

create_observable_counter(name, callback, unit='', description='')[source]

Creates an observable counter instrument

An observable counter observes a monotonically increasing count by calling a provided callback which returns multiple Measurement.

For example, an observable counter could be used to report system CPU time periodically. Here is a basic implementation:

def cpu_time_callback() -> Iterable[Measurement]:
    measurements = []
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if not line.startswith("cpu"): break
            cpu, *states = line.split()
            measurements.append(Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
            measurements.append(Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
            measurements.append(Measurement(int(states[2]) // 100, {"cpu": cpu, "state": "system"}))
            # ... other states
    return measurements

meter.create_observable_counter(
    "system.cpu.time",
    callback=cpu_time_callback,
    unit="s",
    description="CPU time"
)

To reduce memory usage, you can use generator callbacks instead of building the full list:

def cpu_time_callback() -> Iterable[Measurement]:
    with open("/proc/stat") as procstat:
        procstat.readline()  # skip the first line
        for line in procstat:
            if not line.startswith("cpu"): break
            cpu, *states = line.split()
            yield Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"})
            yield Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"})
            # ... other states

Alternatively, you can pass a generator directly instead of a callback, which should return iterables of Measurement:

def cpu_time_callback(states_to_include: set[str]) -> Iterable[Iterable[Measurement]]:
    while True:
        measurements = []
        with open("/proc/stat") as procstat:
            procstat.readline()  # skip the first line
            for line in procstat:
                if not line.startswith("cpu"): break
                cpu, *states = line.split()
                if "user" in states_to_include:
                    measurements.append(Measurement(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
                if "nice" in states_to_include:
                    measurements.append(Measurement(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
                # ... other states
        yield measurements

meter.create_observable_counter(
    "system.cpu.time",
    callback=cpu_time_callback({"user", "system"}),
    unit="s",
    description="CPU time"
)
Parameters
  • name – The name of the instrument to be created

  • callback – A callback that returns an iterable of Measurement. Alternatively, can be a generator that yields iterables of Measurement.

  • unit – The unit for measurements this instrument reports. For example, By for bytes. UCUM units are recommended.

  • description – A description for this instrument and what it measures.

Return type

ObservableCounter

create_histogram(name, unit='', description='')[source]
Return type

Histogram

create_observable_gauge(name, callback, unit='', description='')[source]
Return type

ObservableGauge

create_observable_up_down_counter(name, callback, unit='', description='')[source]
Return type

ObservableUpDownCounter

opentelemetry._metrics.get_meter(name, version='', meter_provider=None)[source]

Returns a Meter for use by the given instrumentation library.

This function is a convenience wrapper for opentelemetry.trace.MeterProvider.get_meter.

If meter_provider is omitted the current configured one is used.

Return type

Meter

opentelemetry._metrics.set_meter_provider(meter_provider)[source]

Sets the current global MeterProvider object.

This can only be done once, a warning will be logged if any furter attempt is made.

Return type

None

opentelemetry._metrics.get_meter_provider()[source]

Gets the current global MeterProvider object.

Return type

MeterProvider