# Source code for opentelemetry.exporter.prometheus

    # 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.

    """
    This library allows export of metrics data to `Prometheus <https://prometheus.io/>`_.

    Usage
    -----

    The **OpenTelemetry Prometheus Exporter** allows export of `OpenTelemetry`_
    metrics to `Prometheus`_.

    .. _Prometheus: https://prometheus.io/
    .. _OpenTelemetry: https://github.com/open-telemetry/opentelemetry-python/

    .. code:: python

        from prometheus_client import start_http_server

        from opentelemetry.exporter.prometheus import PrometheusMetricReader
        from opentelemetry.metrics import get_meter_provider, set_meter_provider
        from opentelemetry.sdk.metrics import MeterProvider

        # Start Prometheus client
        start_http_server(port=8000, addr="localhost")

        # Exporter to export metrics to Prometheus
        prefix = "MyAppPrefix"
        reader = PrometheusMetricReader(prefix)

        # Meter is responsible for creating and recording metrics
        set_meter_provider(MeterProvider(metric_readers=[reader]))
        meter = get_meter_provider().get_meter("myapp", "0.1.2")

        counter = meter.create_counter(
            "requests",
            "requests",
            "number of requests",
        )

        # Labels are used to identify key-values that are associated with a specific
        # metric that you want to record. These are useful for pre-aggregation and can
        # be used to store custom dimensions pertaining to a metric
        labels = {"environment": "staging"}

        counter.add(25, labels)
        input("Press any key to exit...")

    API
    ---
    """

    from collections import deque
    from itertools import chain
    from json import dumps
    from logging import getLogger
    from os import environ
    from typing import Deque, Dict, Iterable, Sequence, Tuple, Union

    from prometheus_client import start_http_server
    from prometheus_client.core import (
        REGISTRY,
        CounterMetricFamily,
        GaugeMetricFamily,
        HistogramMetricFamily,
        InfoMetricFamily,
    )
    from prometheus_client.core import Metric as PrometheusMetric

    from opentelemetry.exporter.prometheus._mapping import (
        map_unit,
        sanitize_attribute,
        sanitize_full_name,
    )
    from opentelemetry.sdk.environment_variables import (
        OTEL_EXPORTER_PROMETHEUS_HOST,
        OTEL_EXPORTER_PROMETHEUS_PORT,
    )
    from opentelemetry.sdk.metrics import (
        Counter,
        ObservableCounter,
        ObservableGauge,
        ObservableUpDownCounter,
        UpDownCounter,
    )
    from opentelemetry.sdk.metrics import Histogram as HistogramInstrument
    from opentelemetry.sdk.metrics.export import (
        AggregationTemporality,
        Gauge,
        Histogram,
        HistogramDataPoint,
        MetricReader,
        MetricsData,
        Sum,
    )
    from opentelemetry.util.types import Attributes

    _logger = getLogger(__name__)

    _TARGET_INFO_NAME = "target"
    _TARGET_INFO_DESCRIPTION = "Target metadata"

    def _convert_buckets(
        bucket_counts: Sequence[int], explicit_bounds: Sequence[float]
    ) -> Sequence[Tuple[str, int]]:
        buckets = []
        total_count = 0
        for upper_bound, count in zip(
            chain(explicit_bounds, ["+Inf"]),
            bucket_counts,
        ):
            total_count += count
            buckets.append((f"{upper_bound}", total_count))

        return buckets

    class PrometheusMetricReader(MetricReader):
        """Prometheus metric exporter for OpenTelemetry."""

        def __init__(self, disable_target_info: bool = False) -> None:
            super().__init__(
                preferred_temporality={
                    Counter: AggregationTemporality.CUMULATIVE,
                    UpDownCounter: AggregationTemporality.CUMULATIVE,
                    HistogramInstrument: AggregationTemporality.CUMULATIVE,
                    ObservableCounter: AggregationTemporality.CUMULATIVE,
                    ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
                    ObservableGauge: AggregationTemporality.CUMULATIVE,
                }
            )
            self._collector = _CustomCollector(disable_target_info)
            REGISTRY.register(self._collector)
            self._collector._callback = self.collect

        def _receive_metrics(
            self,
            metrics_data: MetricsData,
            timeout_millis: float = 10_000,
            **kwargs,
        ) -> None:
            if metrics_data is None:
                return
            self._collector.add_metrics_data(metrics_data)

        def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
            REGISTRY.unregister(self._collector)

    class _CustomCollector:
        """_CustomCollector represents the Prometheus Collector object

        See more:
        https://github.com/prometheus/client_python#custom-collectors
        """

        def __init__(self, disable_target_info: bool = False):
            self._callback = None
            self._metrics_datas: Deque[MetricsData] = deque()
            self._disable_target_info = disable_target_info
            self._target_info = None

        def add_metrics_data(self, metrics_data: MetricsData) -> None:
            """Add metrics to Prometheus data"""
            self._metrics_datas.append(metrics_data)

        def collect(self) -> Iterable[PrometheusMetric]:
            """Collect fetches the metrics from OpenTelemetry
            and delivers them as Prometheus Metrics.
            Collect is invoked every time a ``prometheus.Gatherer`` is run
            for example when the HTTP endpoint is invoked by Prometheus.
            """
            if self._callback is not None:
                self._callback()

            metric_family_id_metric_family = {}

            if len(self._metrics_datas):
                if not self._disable_target_info:
                    if self._target_info is None:
                        attributes: Attributes = {}
                        for res in self._metrics_datas[0].resource_metrics:
                            attributes = {**attributes, **res.resource.attributes}

                        self._target_info = self._create_info_metric(
                            _TARGET_INFO_NAME, _TARGET_INFO_DESCRIPTION, attributes
                        )
                    metric_family_id_metric_family[_TARGET_INFO_NAME] = (
                        self._target_info
                    )

            while self._metrics_datas:
                self._translate_to_prometheus(
                    self._metrics_datas.popleft(), metric_family_id_metric_family
                )

                if metric_family_id_metric_family:
                    yield from metric_family_id_metric_family.values()

        # pylint: disable=too-many-locals,too-many-branches
        def _translate_to_prometheus(
            self,
            metrics_data: MetricsData,
            metric_family_id_metric_family: Dict[str, PrometheusMetric],
        ):
            metrics = []

            for resource_metrics in metrics_data.resource_metrics:
                for scope_metrics in resource_metrics.scope_metrics:
                    for metric in scope_metrics.metrics:
                        metrics.append(metric)

            for metric in metrics:
                label_values_data_points = []
                label_keys_data_points = []
                values = []

                per_metric_family_ids = []

                metric_name = sanitize_full_name(metric.name)
                metric_description = metric.description or ""
                metric_unit = map_unit(metric.unit)

                for number_data_point in metric.data.data_points:
                    label_keys = []
                    label_values = []

                    for key, value in sorted(number_data_point.attributes.items()):
                        label_keys.append(sanitize_attribute(key))
                        label_values.append(self._check_value(value))

                    per_metric_family_ids.append(
                        "|".join(
                            [
                                metric_name,
                                metric_description,
                                "%".join(label_keys),
                                metric_unit,
                            ]
                        )
                    )

                    label_values_data_points.append(label_values)
                    label_keys_data_points.append(label_keys)
                    if isinstance(number_data_point, HistogramDataPoint):
                        values.append(
                            {
                                "bucket_counts": number_data_point.bucket_counts,
                                "explicit_bounds": (
                                    number_data_point.explicit_bounds
                                ),
                                "sum": number_data_point.sum,
                            }
                        )
                    else:
                        values.append(number_data_point.value)

                for per_metric_family_id, label_keys, label_values, value in zip(
                    per_metric_family_ids,
                    label_keys_data_points,
                    label_values_data_points,
                    values,
                ):
                    is_non_monotonic_sum = (
                        isinstance(metric.data, Sum)
                        and metric.data.is_monotonic is False
                    )
                    is_cumulative = (
                        isinstance(metric.data, Sum)
                        and metric.data.aggregation_temporality
                        == AggregationTemporality.CUMULATIVE
                    )

                    # The prometheus compatibility spec for sums says: If the aggregation temporality is cumulative and the sum is non-monotonic, it MUST be converted to a Prometheus Gauge.
                    should_convert_sum_to_gauge = (
                        is_non_monotonic_sum and is_cumulative
                    )

                    if (
                        isinstance(metric.data, Sum)
                        and not should_convert_sum_to_gauge
                    ):
                        metric_family_id = "|".join(
                            [per_metric_family_id, CounterMetricFamily.__name__]
                        )

                        if metric_family_id not in metric_family_id_metric_family:
                            metric_family_id_metric_family[metric_family_id] = (
                                CounterMetricFamily(
                                    name=metric_name,
                                    documentation=metric_description,
                                    labels=label_keys,
                                    unit=metric_unit,
                                )
                            )
                        metric_family_id_metric_family[
                            metric_family_id
                        ].add_metric(labels=label_values, value=value)
                    elif (
                        isinstance(metric.data, Gauge)
                        or should_convert_sum_to_gauge
                    ):
                        metric_family_id = "|".join(
                            [per_metric_family_id, GaugeMetricFamily.__name__]
                        )

                        if (
                            metric_family_id
                            not in metric_family_id_metric_family.keys()
                        ):
                            metric_family_id_metric_family[metric_family_id] = (
                                GaugeMetricFamily(
                                    name=metric_name,
                                    documentation=metric_description,
                                    labels=label_keys,
                                    unit=metric_unit,
                                )
                            )
                        metric_family_id_metric_family[
                            metric_family_id
                        ].add_metric(labels=label_values, value=value)
                    elif isinstance(metric.data, Histogram):
                        metric_family_id = "|".join(
                            [per_metric_family_id, HistogramMetricFamily.__name__]
                        )

                        if (
                            metric_family_id
                            not in metric_family_id_metric_family.keys()
                        ):
                            metric_family_id_metric_family[metric_family_id] = (
                                HistogramMetricFamily(
                                    name=metric_name,
                                    documentation=metric_description,
                                    labels=label_keys,
                                    unit=metric_unit,
                                )
                            )
                        metric_family_id_metric_family[
                            metric_family_id
                        ].add_metric(
                            labels=label_values,
                            buckets=_convert_buckets(
                                value["bucket_counts"], value["explicit_bounds"]
                            ),
                            sum_value=value["sum"],
                        )
                    else:
                        _logger.warning(
                            "Unsupported metric data. %s", type(metric.data)
                        )

        # pylint: disable=no-self-use
        def _check_value(self, value: Union[int, float, str, Sequence]) -> str:
            """Check the label value and return is appropriate representation"""
            if not isinstance(value, str):
                return dumps(value, default=str)
            return str(value)

        def _create_info_metric(
            self, name: str, description: str, attributes: Dict[str, str]
        ) -> InfoMetricFamily:
            """Create an Info Metric Family with list of attributes"""
            # sanitize the attribute names according to Prometheus rule
            attributes = {
                sanitize_attribute(key): self._check_value(value)
                for key, value in attributes.items()
            }
            info = InfoMetricFamily(name, description, labels=attributes)
            info.add_metric(labels=list(attributes.keys()), value=attributes)
            return info

    class _AutoPrometheusMetricReader(PrometheusMetricReader):
        """Thin wrapper around PrometheusMetricReader used for the opentelemetry_metrics_exporter entry point.

        This allows users to use the prometheus exporter with opentelemetry-instrument. It handles
        starting the Prometheus http server on the the correct port and host.
        """

        def __init__(self) -> None:
            super().__init__()

            # Default values are specified in
            # https://github.com/open-telemetry/opentelemetry-specification/blob/v1.24.0/specification/configuration/sdk-environment-variables.md#prometheus-exporter
            start_http_server(
                port=int(environ.get(OTEL_EXPORTER_PROMETHEUS_PORT, "9464")),
                addr=environ.get(OTEL_EXPORTER_PROMETHEUS_HOST, "localhost"),
            )