hermes-chatto-plugin/vendor/common/opentelemetry/metrics/_internal/__init__.py

# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0

# pylint: disable=too-many-ancestors

"""
The OpenTelemetry metrics API  describes the classes used to generate
metrics.

The :class:`.MeterProvider` provides users access to the :class:`.Meter` which in
turn is used to create :class:`.Instrument` objects. The :class:`.Instrument` objects are
used to record measurements.

This module provides abstract (i.e. unimplemented) classes required for
metrics, and a concrete no-op implementation :class:`.NoOpMeter` that allows applications
to use the API package alone without a supporting implementation.

To get a meter, you need to provide the package name from which you are
calling the meter APIs to OpenTelemetry by calling `MeterProvider.get_meter`
with the calling instrumentation name and the version of your package.

The following code shows how to obtain a meter using the global :class:`.MeterProvider`::

    from opentelemetry.metrics import get_meter

    meter = get_meter("example-meter")
    counter = meter.create_counter("example-counter")

.. versionadded:: 1.10.0
"""

import warnings
from abc import ABC, abstractmethod
from collections.abc import Sequence
from dataclasses import dataclass
from logging import getLogger
from os import environ
from threading import Lock
from typing import cast

from opentelemetry.environment_variables import OTEL_PYTHON_METER_PROVIDER
from opentelemetry.metrics._internal.instrument import (
    CallbackT,
    Counter,
    Gauge,
    Histogram,
    NoOpCounter,
    NoOpGauge,
    NoOpHistogram,
    NoOpObservableCounter,
    NoOpObservableGauge,
    NoOpObservableUpDownCounter,
    NoOpUpDownCounter,
    ObservableCounter,
    ObservableGauge,
    ObservableUpDownCounter,
    UpDownCounter,
    _MetricsHistogramAdvisory,
    _ProxyCounter,
    _ProxyGauge,
    _ProxyHistogram,
    _ProxyObservableCounter,
    _ProxyObservableGauge,
    _ProxyObservableUpDownCounter,
    _ProxyUpDownCounter,
)
from opentelemetry.util._once import Once
from opentelemetry.util._providers import _load_provider
from opentelemetry.util.types import (
    Attributes,
)

_logger = getLogger(__name__)


# pylint: disable=invalid-name
_ProxyInstrumentT = (
    _ProxyCounter
    | _ProxyHistogram
    | _ProxyGauge
    | _ProxyObservableCounter
    | _ProxyObservableGauge
    | _ProxyObservableUpDownCounter
    | _ProxyUpDownCounter
)


class MeterProvider(ABC):
    """
    MeterProvider is the entry point of the API. It provides access to `Meter` instances.
    """

    @abstractmethod
    def get_meter(
        self,
        name: str,
        version: str | None = None,
        schema_url: str | None = None,
        attributes: Attributes | None = None,
    ) -> "Meter":
        """Returns a `Meter` for use by the given instrumentation library.

        For any two calls it is undefined whether the same or different
        `Meter` instances are returned, even for different library names.

        This function may return different `Meter` types (e.g. a no-op meter
        vs. a functional meter).

        Args:
            name: The name of the instrumenting module.
                ``__name__`` should be avoided as this can result in
                different meter names if the meters are in different files.
                It is better to use a fixed string that can be imported where
                needed and used consistently as the name of the meter.

                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"``.

            version: Optional. The version string of the
                instrumenting library.  Usually this should be the same as
                ``importlib.metadata.version(instrumenting_library_name)``.

            schema_url: Optional. Specifies the Schema URL of the emitted telemetry.
            attributes: Optional. Attributes that are associated with the emitted telemetry.
        """


class NoOpMeterProvider(MeterProvider):
    """The default MeterProvider used when no MeterProvider implementation is available."""

    def get_meter(
        self,
        name: str,
        version: str | None = None,
        schema_url: str | None = None,
        attributes: Attributes | None = None,
    ) -> "Meter":
        """Returns a NoOpMeter."""
        return NoOpMeter(name, version=version, schema_url=schema_url)


class _ProxyMeterProvider(MeterProvider):
    def __init__(self) -> None:
        self._lock = Lock()
        self._meters: list[_ProxyMeter] = []
        self._real_meter_provider: MeterProvider | None = None

    def get_meter(
        self,
        name: str,
        version: str | None = None,
        schema_url: str | None = None,
        attributes: Attributes | None = None,
    ) -> "Meter":
        with self._lock:
            if self._real_meter_provider is not None:
                return self._real_meter_provider.get_meter(
                    name, version, schema_url
                )

            meter = _ProxyMeter(name, version=version, schema_url=schema_url)
            self._meters.append(meter)
            return meter

    def on_set_meter_provider(self, meter_provider: MeterProvider) -> None:
        with self._lock:
            self._real_meter_provider = meter_provider
            for meter in self._meters:
                meter.on_set_meter_provider(meter_provider)


@dataclass
class _InstrumentRegistrationStatus:
    instrument_id: str
    already_registered: bool
    conflict: bool
    current_advisory: _MetricsHistogramAdvisory | None


class Meter(ABC):
    """Handles instrument creation.

    This class provides methods for creating instruments which are then
    used to produce measurements.
    """

    def __init__(
        self,
        name: str,
        version: str | None = None,
        schema_url: str | None = None,
    ) -> None:
        super().__init__()
        self._name = name
        self._version = version
        self._schema_url = schema_url
        self._instrument_ids: dict[str, _MetricsHistogramAdvisory | None] = {}
        self._instrument_ids_lock = Lock()

    @property
    def name(self) -> str:
        """
        The name of the instrumenting module.
        """
        return self._name

    @property
    def version(self) -> str | None:
        """
        The version string of the instrumenting library.
        """
        return self._version

    @property
    def schema_url(self) -> str | None:
        """
        Specifies the Schema URL of the emitted telemetry
        """
        return self._schema_url

    def _register_instrument(
        self,
        name: str,
        type_: type,
        unit: str,
        description: str,
        advisory: _MetricsHistogramAdvisory | None = None,
    ) -> _InstrumentRegistrationStatus:
        """
        Register an instrument with the name, type, unit and description as
        identifying keys and the advisory as value.

        Returns a tuple. The first value is the instrument id.
        The second value is an `_InstrumentRegistrationStatus` where
        `already_registered` is `True` if the instrument has been registered
        already.
        If `conflict` is set to True the `current_advisory` attribute contains
        the registered instrument advisory.
        """

        instrument_id = ",".join(
            [name.strip().lower(), type_.__name__, unit, description]
        )

        already_registered = False
        conflict = False
        current_advisory = None

        with self._instrument_ids_lock:
            # we are not using get because None is a valid value
            already_registered = instrument_id in self._instrument_ids
            if already_registered:
                current_advisory = self._instrument_ids[instrument_id]
                conflict = current_advisory != advisory
            else:
                self._instrument_ids[instrument_id] = advisory

        return _InstrumentRegistrationStatus(
            instrument_id=instrument_id,
            already_registered=already_registered,
            conflict=conflict,
            current_advisory=current_advisory,
        )

    @staticmethod
    def _log_instrument_registration_conflict(
        name: str,
        instrumentation_type: str,
        unit: str,
        description: str,
        status: _InstrumentRegistrationStatus,
    ) -> None:
        _logger.warning(
            "An instrument with name %s, type %s, unit %s and "
            "description %s has been created already with a "
            "different advisory value %s and will be used instead.",
            name,
            instrumentation_type,
            unit,
            description,
            status.current_advisory,
        )

    @abstractmethod
    def create_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> Counter:
        """Creates a `Counter` instrument

        Args:
            name: The name of the instrument to be created
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """

    @abstractmethod
    def create_up_down_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> UpDownCounter:
        """Creates an `UpDownCounter` instrument

        Args:
            name: The name of the instrument to be created
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """

    @abstractmethod
    def create_observable_counter(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableCounter:
        """Creates an `ObservableCounter` instrument

        An observable counter observes a monotonically increasing count by calling provided
        callbacks which accept a :class:`~opentelemetry.metrics.CallbackOptions` and return
        multiple :class:`~opentelemetry.metrics.Observation`.

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

            def cpu_time_callback(options: CallbackOptions) -> Iterable[Observation]:
                observations = []
                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()
                        observations.append(Observation(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
                        observations.append(Observation(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
                        observations.append(Observation(int(states[2]) // 100, {"cpu": cpu, "state": "system"}))
                        # ... other states
                return observations

            meter.create_observable_counter(
                "system.cpu.time",
                callbacks=[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(options: CallbackOptions) -> Iterable[Observation]:
                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 Observation(int(states[0]) // 100, {"cpu": cpu, "state": "user"})
                        yield Observation(int(states[1]) // 100, {"cpu": cpu, "state": "nice"})
                        # ... other states

        Alternatively, you can pass a sequence of generators directly instead of a sequence of
        callbacks, which each should return iterables of :class:`~opentelemetry.metrics.Observation`::

            def cpu_time_callback(states_to_include: set[str]) -> Iterable[Iterable[Observation]]:
                # accept options sent in from OpenTelemetry
                options = yield
                while True:
                    observations = []
                    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:
                                observations.append(Observation(int(states[0]) // 100, {"cpu": cpu, "state": "user"}))
                            if "nice" in states_to_include:
                                observations.append(Observation(int(states[1]) // 100, {"cpu": cpu, "state": "nice"}))
                            # ... other states
                    # yield the observations and receive the options for next iteration
                    options = yield observations

            meter.create_observable_counter(
                "system.cpu.time",
                callbacks=[cpu_time_callback({"user", "system"})],
                unit="s",
                description="CPU time"
            )

        The :class:`~opentelemetry.metrics.CallbackOptions` contain a timeout which the
        callback should respect. For example if the callback does asynchronous work, like
        making HTTP requests, it should respect the timeout::

            def scrape_http_callback(options: CallbackOptions) -> Iterable[Observation]:
                r = requests.get('http://scrapethis.com', timeout=options.timeout_millis / 10**3)
                for value in r.json():
                    yield Observation(value)

        Args:
            name: The name of the instrument to be created
            callbacks: A sequence of callbacks that return an iterable of
                :class:`~opentelemetry.metrics.Observation`. Alternatively, can be a sequence of generators that each
                yields iterables of :class:`~opentelemetry.metrics.Observation`.
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """

    @abstractmethod
    def create_histogram(
        self,
        name: str,
        unit: str = "",
        description: str = "",
        *,
        explicit_bucket_boundaries_advisory: Sequence[float] | None = None,
    ) -> Histogram:
        """Creates a :class:`~opentelemetry.metrics.Histogram` instrument

        Args:
            name: The name of the instrument to be created
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """

    def create_gauge(  # type: ignore # pylint: disable=no-self-use
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> Gauge:  # pyright: ignore[reportReturnType]
        """Creates a ``Gauge`` instrument

        Args:
            name: The name of the instrument to be created
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """
        warnings.warn("create_gauge() is not implemented and will be a no-op")

    @abstractmethod
    def create_observable_gauge(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableGauge:
        """Creates an `ObservableGauge` instrument

        Args:
            name: The name of the instrument to be created
            callbacks: A sequence of callbacks that return an iterable of
                :class:`~opentelemetry.metrics.Observation`. Alternatively, can be a generator that yields iterables
                of :class:`~opentelemetry.metrics.Observation`.
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """

    @abstractmethod
    def create_observable_up_down_counter(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableUpDownCounter:
        """Creates an `ObservableUpDownCounter` instrument

        Args:
            name: The name of the instrument to be created
            callbacks: A sequence of callbacks that return an iterable of
                :class:`~opentelemetry.metrics.Observation`. Alternatively, can be a generator that yields iterables
                of :class:`~opentelemetry.metrics.Observation`.
            unit: The unit for observations this instrument reports. For
                example, ``By`` for bytes. UCUM units are recommended.
            description: A description for this instrument and what it measures.
        """


class _ProxyMeter(Meter):
    def __init__(
        self,
        name: str,
        version: str | None = None,
        schema_url: str | None = None,
    ) -> None:
        super().__init__(name, version=version, schema_url=schema_url)
        self._lock = Lock()
        self._instruments: list[_ProxyInstrumentT] = []
        self._real_meter: Meter | None = None

    def on_set_meter_provider(self, meter_provider: MeterProvider) -> None:
        """Called when a real meter provider is set on the creating _ProxyMeterProvider

        Creates a real backing meter for this instance and notifies all created
        instruments so they can create real backing instruments.
        """
        real_meter = meter_provider.get_meter(
            self._name, self._version, self._schema_url
        )

        with self._lock:
            self._real_meter = real_meter
            # notify all proxy instruments of the new meter so they can create
            # real instruments to back themselves
            for instrument in self._instruments:
                instrument.on_meter_set(real_meter)

    def create_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> Counter:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_counter(name, unit, description)
            proxy = _ProxyCounter(name, unit, description)
            self._instruments.append(proxy)
            return proxy

    def create_up_down_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> UpDownCounter:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_up_down_counter(
                    name, unit, description
                )
            proxy = _ProxyUpDownCounter(name, unit, description)
            self._instruments.append(proxy)
            return proxy

    def create_observable_counter(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableCounter:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_observable_counter(
                    name, callbacks, unit, description
                )
            proxy = _ProxyObservableCounter(
                name, callbacks, unit=unit, description=description
            )
            self._instruments.append(proxy)
            return proxy

    def create_histogram(
        self,
        name: str,
        unit: str = "",
        description: str = "",
        *,
        explicit_bucket_boundaries_advisory: Sequence[float] | None = None,
    ) -> Histogram:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_histogram(
                    name,
                    unit,
                    description,
                    explicit_bucket_boundaries_advisory=explicit_bucket_boundaries_advisory,
                )
            proxy = _ProxyHistogram(
                name, unit, description, explicit_bucket_boundaries_advisory
            )
            self._instruments.append(proxy)
            return proxy

    def create_gauge(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> Gauge:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_gauge(name, unit, description)
            proxy = _ProxyGauge(name, unit, description)
            self._instruments.append(proxy)
            return proxy

    def create_observable_gauge(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableGauge:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_observable_gauge(
                    name, callbacks, unit, description
                )
            proxy = _ProxyObservableGauge(
                name, callbacks, unit=unit, description=description
            )
            self._instruments.append(proxy)
            return proxy

    def create_observable_up_down_counter(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableUpDownCounter:
        with self._lock:
            if self._real_meter:
                return self._real_meter.create_observable_up_down_counter(
                    name,
                    callbacks,
                    unit,
                    description,
                )
            proxy = _ProxyObservableUpDownCounter(
                name, callbacks, unit=unit, description=description
            )
            self._instruments.append(proxy)
            return proxy


class NoOpMeter(Meter):
    """The default Meter used when no Meter implementation is available.

    All operations are no-op.
    """

    def create_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> Counter:
        """Returns a no-op Counter."""
        status = self._register_instrument(
            name, NoOpCounter, unit, description
        )
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                Counter.__name__,
                unit,
                description,
                status,
            )

        return NoOpCounter(name, unit=unit, description=description)

    def create_gauge(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> Gauge:
        """Returns a no-op Gauge."""
        status = self._register_instrument(name, NoOpGauge, unit, description)
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                Gauge.__name__,
                unit,
                description,
                status,
            )
        return NoOpGauge(name, unit=unit, description=description)

    def create_up_down_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> UpDownCounter:
        """Returns a no-op UpDownCounter."""
        status = self._register_instrument(
            name, NoOpUpDownCounter, unit, description
        )
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                UpDownCounter.__name__,
                unit,
                description,
                status,
            )
        return NoOpUpDownCounter(name, unit=unit, description=description)

    def create_observable_counter(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableCounter:
        """Returns a no-op ObservableCounter."""
        status = self._register_instrument(
            name, NoOpObservableCounter, unit, description
        )
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                ObservableCounter.__name__,
                unit,
                description,
                status,
            )
        return NoOpObservableCounter(
            name,
            callbacks,
            unit=unit,
            description=description,
        )

    def create_histogram(
        self,
        name: str,
        unit: str = "",
        description: str = "",
        *,
        explicit_bucket_boundaries_advisory: Sequence[float] | None = None,
    ) -> Histogram:
        """Returns a no-op Histogram."""
        status = self._register_instrument(
            name,
            NoOpHistogram,
            unit,
            description,
            _MetricsHistogramAdvisory(
                explicit_bucket_boundaries=explicit_bucket_boundaries_advisory
            ),
        )
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                Histogram.__name__,
                unit,
                description,
                status,
            )
        return NoOpHistogram(
            name,
            unit=unit,
            description=description,
            explicit_bucket_boundaries_advisory=explicit_bucket_boundaries_advisory,
        )

    def create_observable_gauge(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableGauge:
        """Returns a no-op ObservableGauge."""
        status = self._register_instrument(
            name, NoOpObservableGauge, unit, description
        )
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                ObservableGauge.__name__,
                unit,
                description,
                status,
            )
        return NoOpObservableGauge(
            name,
            callbacks,
            unit=unit,
            description=description,
        )

    def create_observable_up_down_counter(
        self,
        name: str,
        callbacks: Sequence[CallbackT] | None = None,
        unit: str = "",
        description: str = "",
    ) -> ObservableUpDownCounter:
        """Returns a no-op ObservableUpDownCounter."""
        status = self._register_instrument(
            name, NoOpObservableUpDownCounter, unit, description
        )
        if status.conflict:
            self._log_instrument_registration_conflict(
                name,
                ObservableUpDownCounter.__name__,
                unit,
                description,
                status,
            )
        return NoOpObservableUpDownCounter(
            name,
            callbacks,
            unit=unit,
            description=description,
        )


_METER_PROVIDER_SET_ONCE = Once()
_METER_PROVIDER: MeterProvider | None = None
_PROXY_METER_PROVIDER = _ProxyMeterProvider()


def get_meter(
    name: str,
    version: str = "",
    meter_provider: MeterProvider | None = None,
    schema_url: str | None = None,
    attributes: Attributes | None = None,
) -> "Meter":
    """Returns a `Meter` for use by the given instrumentation library.

    This function is a convenience wrapper for
    `opentelemetry.metrics.MeterProvider.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(name, version, schema_url, attributes)


def _set_meter_provider(meter_provider: MeterProvider, log: bool) -> None:
    def set_mp() -> None:
        global _METER_PROVIDER  # pylint: disable=global-statement
        _METER_PROVIDER = meter_provider

        # gives all proxies real instruments off the newly set meter provider
        _PROXY_METER_PROVIDER.on_set_meter_provider(meter_provider)

    did_set = _METER_PROVIDER_SET_ONCE.do_once(set_mp)

    if log and not did_set:
        _logger.warning("Overriding of current MeterProvider is not allowed")


def set_meter_provider(meter_provider: MeterProvider) -> None:
    """Sets the current global :class:`~.MeterProvider` object.

    This can only be done once, a warning will be logged if any further attempt
    is made.
    """
    _set_meter_provider(meter_provider, log=True)


def get_meter_provider() -> MeterProvider:
    """Gets the current global :class:`~.MeterProvider` object."""

    if _METER_PROVIDER is None:
        if OTEL_PYTHON_METER_PROVIDER not in environ:
            return _PROXY_METER_PROVIDER

        meter_provider: MeterProvider = _load_provider(  # type: ignore
            OTEL_PYTHON_METER_PROVIDER, "meter_provider"
        )
        _set_meter_provider(meter_provider, log=False)

    # _METER_PROVIDER will have been set by one thread
    return cast("MeterProvider", _METER_PROVIDER)