# Copyright The OpenTelemetry Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
The OpenTelemetry metrics API describes the classes used to report raw
measurements, as well as metrics with known aggregation and labels.
The `Meter` class is used to construct `Metric` s to record raw statistics
as well as metrics with predefined aggregation.
`Meter` s can be obtained via the `MeterProvider`, corresponding to the name
of the instrumenting library and (optionally) a version.
See the `metrics api`_ spec for terminology and context clarification.
.. _metrics api:
https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/metrics/api.md
"""
import abc
from logging import getLogger
from typing import (
Callable,
Dict,
Optional,
Sequence,
Tuple,
Type,
TypeVar,
Union,
)
from opentelemetry.util import _load_meter_provider
logger = getLogger(__name__)
ValueT = TypeVar("ValueT", int, float)
[docs]class BoundCounter(abc.ABC):
[docs] @abc.abstractmethod
def add(self, value: ValueT) -> None:
"""Increases the value of the bound counter by ``value``.
Args:
value: The value to add to the bound counter. Must be positive.
"""
[docs]class DefaultBoundCounter(BoundCounter):
"""The default bound counter instrument.
Used when no bound counter implementation is available.
"""
[docs] def add(self, value: ValueT) -> None:
pass
[docs]class BoundUpDownCounter(abc.ABC):
[docs] @abc.abstractmethod
def add(self, value: ValueT) -> None:
"""Increases the value of the bound updowncounter by ``value``.
Args:
value: The value to add to the bound updowncounter. Can be positive
or negative.
"""
[docs]class DefaultBoundUpDownCounter(BoundUpDownCounter):
"""The default bound updowncounter instrument.
Used when no bound updowncounter implementation is available.
"""
[docs] def add(self, value: ValueT) -> None:
pass
[docs]class BoundValueRecorder(abc.ABC):
[docs] @abc.abstractmethod
def record(self, value: ValueT) -> None:
"""Records the given ``value`` to this bound valuerecorder.
Args:
value: The value to record to the bound valuerecorder.
"""
[docs]class DefaultBoundValueRecorder(BoundValueRecorder):
"""The default bound valuerecorder instrument.
Used when no bound valuerecorder implementation is available.
"""
[docs] def record(self, value: ValueT) -> None:
pass
[docs]class Metric(abc.ABC):
"""Base class for various types of metrics.
Metric class that inherit from this class are specialized with the type of
bound metric instrument that the metric holds.
"""
[docs] @abc.abstractmethod
def bind(self, labels: Dict[str, str]) -> "object":
"""Gets a bound metric instrument.
Bound metric instruments are useful to reduce the cost of repeatedly
recording a metric with a pre-defined set of label values.
Args:
labels: Labels to associate with the bound instrument.
"""
[docs]class Counter(Metric):
"""A counter type metric that expresses the computation of a sum."""
[docs] @abc.abstractmethod
def add(self, value: ValueT, labels: Dict[str, str]) -> None:
"""Increases the value of the counter by ``value``.
Args:
value: The value to add to the counter metric. Should be positive
or zero. For a Counter that can decrease, use
`UpDownCounter`.
labels: Labels to associate with the bound instrument.
"""
[docs]class DefaultCounter(Counter):
"""The default counter instrument.
Used when no `Counter` implementation is available.
"""
[docs] def bind(self, labels: Dict[str, str]) -> "DefaultBoundCounter":
return DefaultBoundCounter()
[docs] def add(self, value: ValueT, labels: Dict[str, str]) -> None:
pass
[docs]class UpDownCounter(Metric):
"""A counter type metric that expresses the computation of a sum,
allowing negative increments."""
[docs] @abc.abstractmethod
def add(self, value: ValueT, labels: Dict[str, str]) -> None:
"""Increases the value of the counter by ``value``.
Args:
value: The value to add to the counter metric. Can be positive or
negative. For a Counter that is never decreasing, use
`Counter`.
labels: Labels to associate with the bound instrument.
"""
[docs]class DefaultUpDownCounter(UpDownCounter):
"""The default updowncounter instrument.
Used when no `UpDownCounter` implementation is available.
"""
[docs] def bind(self, labels: Dict[str, str]) -> "DefaultBoundUpDownCounter":
return DefaultBoundUpDownCounter()
[docs] def add(self, value: ValueT, labels: Dict[str, str]) -> None:
pass
[docs]class ValueRecorder(Metric):
"""A valuerecorder type metric that represent raw stats."""
[docs] @abc.abstractmethod
def record(self, value: ValueT, labels: Dict[str, str]) -> None:
"""Records the ``value`` to the valuerecorder.
Args:
value: The value to record to this valuerecorder metric.
labels: Labels to associate with the bound instrument.
"""
[docs]class DefaultValueRecorder(ValueRecorder):
"""The default valuerecorder instrument.
Used when no `ValueRecorder` implementation is available.
"""
[docs] def bind(self, labels: Dict[str, str]) -> "DefaultBoundValueRecorder":
return DefaultBoundValueRecorder()
[docs] def record(self, value: ValueT, labels: Dict[str, str]) -> None:
pass
[docs]class Observer(abc.ABC):
"""An observer type metric instrument used to capture a current set of
values.
Observer instruments are asynchronous, a callback is invoked with the
observer instrument as argument allowing the user to capture multiple
values per collection interval.
"""
[docs] @abc.abstractmethod
def observe(self, value: ValueT, labels: Dict[str, str]) -> None:
"""Captures ``value`` to the observer.
Args:
value: The value to capture to this observer metric.
labels: Labels associated to ``value``.
"""
# pylint: disable=W0223
[docs]class SumObserver(Observer):
"""Asynchronous instrument used to capture a monotonic sum."""
[docs]class DefaultSumObserver(SumObserver):
"""No-op implementation of ``SumObserver``."""
[docs] def observe(self, value: ValueT, labels: Dict[str, str]) -> None:
pass
# pylint: disable=W0223
[docs]class UpDownSumObserver(Observer):
"""Asynchronous instrument used to capture a non-monotonic count."""
[docs]class DefaultUpDownSumObserver(UpDownSumObserver):
"""No-op implementation of ``UpDownSumObserver``."""
[docs] def observe(self, value: ValueT, labels: Dict[str, str]) -> None:
pass
# pylint: disable=W0223
[docs]class ValueObserver(Observer):
"""Asynchronous instrument used to capture grouping measurements."""
[docs]class DefaultValueObserver(ValueObserver):
"""No-op implementation of ``ValueObserver``."""
[docs] def observe(self, value: ValueT, labels: Dict[str, str]) -> None:
pass
[docs]class MeterProvider(abc.ABC):
[docs] @abc.abstractmethod
def get_meter(
self,
instrumenting_module_name: str,
instrumenting_library_version: str = "",
) -> "Meter":
"""Returns a `Meter` for use by the given instrumentation library.
This function may return different `Meter` types (e.g. a no-op meter
vs. a functional meter).
Args:
instrumenting_module_name: The name of the instrumenting module
(usually just ``__name__``).
This should *not* be the name of the module that is
instrumented but the name of the module doing the instrumentation.
E.g., instead of ``"requests"``, use
``"opentelemetry.instrumentation.requests"``.
instrumenting_library_version: Optional. The version string of the
instrumenting library. Usually this should be the same as
``pkg_resources.get_distribution(instrumenting_library_name).version``.
"""
[docs]class DefaultMeterProvider(MeterProvider):
"""The default MeterProvider, used when no implementation is available.
All operations are no-op.
"""
[docs] def get_meter(
self,
instrumenting_module_name: str,
instrumenting_library_version: str = "",
) -> "Meter":
# pylint:disable=no-self-use,unused-argument
return DefaultMeter()
InstrumentT = TypeVar("InstrumentT", bound=Union[Metric, Observer])
ObserverCallbackT = Callable[[Observer], None]
# pylint: disable=unused-argument
[docs]class Meter(abc.ABC):
"""An interface to allow the recording of metrics.
`Metric` s or metric instruments, are devices used for capturing raw
measurements. Each metric instrument supports a single method, each with
fixed interpretation to capture measurements.
"""
[docs] @abc.abstractmethod
def record_batch(
self,
labels: Dict[str, str],
record_tuples: Sequence[Tuple["Metric", ValueT]],
) -> None:
"""Atomically records a batch of `Metric` and value pairs.
Allows the functionality of acting upon multiple metrics with a single
API call. Implementations should find bound metric instruments that
match the key-value pairs in the labels.
Args:
labels: Labels associated with all measurements in the
batch.
record_tuples: A sequence of pairs of `Metric` s and the
corresponding value to record for that metric.
"""
[docs] @abc.abstractmethod
def create_counter(
self,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
enabled: bool = True,
) -> "Counter":
"""Creates a `Counter` metric with type ``value_type``.
Args:
name: The name of the metric.
description: Human-readable description of the metric.
unit: Unit of the metric values following the UCUM convention
(https://unitsofmeasure.org/ucum.html).
value_type: The type of values being recorded by the metric.
enabled: Whether to report the metric by default.
"""
[docs] @abc.abstractmethod
def create_updowncounter(
self,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
enabled: bool = True,
) -> "UpDownCounter":
"""Creates a `UpDownCounter` metric with type ``value_type``.
Args:
name: The name of the metric.
description: Human-readable description of the metric.
unit: Unit of the metric values following the UCUM convention
(https://unitsofmeasure.org/ucum.html).
value_type: The type of values being recorded by the metric.
enabled: Whether to report the metric by default.
"""
[docs] @abc.abstractmethod
def create_valuerecorder(
self,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
enabled: bool = True,
) -> "ValueRecorder":
"""Creates a `ValueRecorder` metric with type ``value_type``.
Args:
name: The name of the metric.
description: Human-readable description of the metric.
unit: Unit of the metric values following the UCUM convention
(https://unitsofmeasure.org/ucum.html).
value_type: The type of values being recorded by the metric.
enabled: Whether to report the metric by default.
"""
[docs] @abc.abstractmethod
def register_sumobserver(
self,
callback: ObserverCallbackT,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
label_keys: Sequence[str] = (),
enabled: bool = True,
) -> "SumObserver":
"""Registers an ``SumObserver`` metric instrument.
Args:
callback: Callback invoked each collection interval with the
observer as argument.
name: The name of the metric.
description: Human-readable description of the metric.
unit: Unit of the metric values following the UCUM convention
(https://unitsofmeasure.org/ucum.html).
value_type: The type of values being recorded by the metric.
label_keys: The keys for the labels with dynamic values.
enabled: Whether to report the metric by default.
Returns: A new ``SumObserver`` metric instrument.
"""
[docs] @abc.abstractmethod
def register_updownsumobserver(
self,
callback: ObserverCallbackT,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
label_keys: Sequence[str] = (),
enabled: bool = True,
) -> "UpDownSumObserver":
"""Registers an ``UpDownSumObserver`` metric instrument.
Args:
callback: Callback invoked each collection interval with the
observer as argument.
name: The name of the metric.
description: Human-readable description of the metric.
unit: Unit of the metric values following the UCUM convention
(https://unitsofmeasure.org/ucum.html).
value_type: The type of values being recorded by the metric.
label_keys: The keys for the labels with dynamic values.
enabled: Whether to report the metric by default.
Returns: A new ``UpDownSumObserver`` metric instrument.
"""
[docs] @abc.abstractmethod
def register_valueobserver(
self,
callback: ObserverCallbackT,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
label_keys: Sequence[str] = (),
enabled: bool = True,
) -> "ValueObserver":
"""Registers an ``ValueObserver`` metric instrument.
Args:
callback: Callback invoked each collection interval with the
observer as argument.
name: The name of the metric.
description: Human-readable description of the metric.
unit: Unit of the metric values following the UCUM convention
(https://unitsofmeasure.org/ucum.html).
value_type: The type of values being recorded by the metric.
label_keys: The keys for the labels with dynamic values.
enabled: Whether to report the metric by default.
Returns: A new ``ValueObserver`` metric instrument.
"""
[docs] @abc.abstractmethod
def unregister_observer(self, observer: "Observer") -> None:
"""Unregisters an ``Observer`` metric instrument.
Args:
observer: The observer to unregister.
"""
[docs]class DefaultMeter(Meter):
"""The default Meter used when no Meter implementation is available."""
[docs] def record_batch(
self,
labels: Dict[str, str],
record_tuples: Sequence[Tuple["Metric", ValueT]],
) -> None:
pass
[docs] def create_counter(
self,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
enabled: bool = True,
) -> "Counter":
# pylint: disable=no-self-use
return DefaultCounter()
[docs] def create_updowncounter(
self,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
enabled: bool = True,
) -> "UpDownCounter":
# pylint: disable=no-self-use
return DefaultUpDownCounter()
[docs] def create_valuerecorder(
self,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
enabled: bool = True,
) -> "ValueRecorder":
# pylint: disable=no-self-use
return DefaultValueRecorder()
[docs] def register_sumobserver(
self,
callback: ObserverCallbackT,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
label_keys: Sequence[str] = (),
enabled: bool = True,
) -> "DefaultSumObserver":
return DefaultSumObserver()
[docs] def register_updownsumobserver(
self,
callback: ObserverCallbackT,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
label_keys: Sequence[str] = (),
enabled: bool = True,
) -> "DefaultUpDownSumObserver":
return DefaultUpDownSumObserver()
[docs] def register_valueobserver(
self,
callback: ObserverCallbackT,
name: str,
description: str,
unit: str,
value_type: Type[ValueT],
label_keys: Sequence[str] = (),
enabled: bool = True,
) -> "DefaultValueObserver":
return DefaultValueObserver()
[docs] def unregister_observer(self, observer: "Observer") -> None:
pass
_METER_PROVIDER = None
[docs]def get_meter(
instrumenting_module_name: str,
instrumenting_library_version: str = "",
meter_provider: Optional[MeterProvider] = None,
) -> "Meter":
"""Returns a `Meter` for use by the given instrumentation library.
This function is a convenience wrapper for
opentelemetry.metrics.get_meter_provider().get_meter
If meter_provider is omitted the current configured one is used.
"""
if meter_provider is None:
meter_provider = get_meter_provider()
return meter_provider.get_meter(
instrumenting_module_name, instrumenting_library_version
)
[docs]def set_meter_provider(meter_provider: MeterProvider) -> None:
"""Sets the current global :class:`~.MeterProvider` object."""
global _METER_PROVIDER # pylint: disable=global-statement
if _METER_PROVIDER is not None:
logger.warning("Overriding of current MeterProvider is not allowed")
return
_METER_PROVIDER = meter_provider
[docs]def get_meter_provider() -> MeterProvider:
"""Gets the current global :class:`~.MeterProvider` object."""
global _METER_PROVIDER # pylint: disable=global-statement
if _METER_PROVIDER is None:
_METER_PROVIDER = _load_meter_provider("meter_provider")
return _METER_PROVIDER