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

from logging import getLogger
from threading import RLock
from time import time_ns

from opentelemetry.metrics import (
    Asynchronous,
    Counter,
    ObservableCounter,
)
from opentelemetry.sdk.metrics._internal._view_instrument_match import (
    _DEFAULT_CARDINALITY_LIMIT,
    _ViewInstrumentMatch,
)
from opentelemetry.sdk.metrics._internal.aggregation import (
    Aggregation,
    AggregationTemporality,
    ExplicitBucketHistogramAggregation,
    ExponentialBucketHistogramAggregation,
    _DropAggregation,
    _ExplicitBucketHistogramAggregation,
    _ExponentialBucketHistogramAggregation,
    _LastValueAggregation,
    _SumAggregation,
)
from opentelemetry.sdk.metrics._internal.instrument import (
    _Asynchronous,
    _Instrument,
)
from opentelemetry.sdk.metrics._internal.measurement import Measurement
from opentelemetry.sdk.metrics._internal.point import (
    ExponentialHistogram,
    Gauge,
    Histogram,
    Metric,
    MetricsData,
    ResourceMetrics,
    ScopeMetrics,
    Sum,
)
from opentelemetry.sdk.metrics._internal.sdk_configuration import (
    SdkConfiguration,
)
from opentelemetry.sdk.metrics._internal.view import View
from opentelemetry.sdk.util.instrumentation import InstrumentationScope

_logger = getLogger(__name__)

_DEFAULT_VIEW = View(instrument_name="")


class MetricReaderStorage:
    """The SDK's storage for a given reader"""

    def __init__(
        self,
        sdk_config: SdkConfiguration,
        instrument_class_temporality: dict[type, AggregationTemporality],
        instrument_class_aggregation: dict[type, Aggregation],
        instrument_class_cardinality_limit: dict[type, int] | None = None,
    ) -> None:
        self._lock = RLock()
        self._sdk_config = sdk_config
        self._instrument_view_instrument_matches: dict[_Instrument, list[_ViewInstrumentMatch]] = {}
        self._instrument_class_temporality = instrument_class_temporality
        self._instrument_class_aggregation = instrument_class_aggregation
        self._instrument_class_cardinality_limit = instrument_class_cardinality_limit or {}

    def _new_view_instrument_match(self, view: View, instrument: _Instrument) -> _ViewInstrumentMatch:
        # pylint: disable=protected-access
        cardinality_limit = view._aggregation_cardinality_limit
        if cardinality_limit is None:
            cardinality_limit = self._instrument_class_cardinality_limit.get(
                instrument.__class__, _DEFAULT_CARDINALITY_LIMIT
            )

        # Idle attribute sets of synchronous instruments are only released on
        # delta collection, so that the limit applies per collection cycle.
        # Asynchronous instruments keep their state, which delta conversion
        # relies on.
        reset_on_collect = self._instrument_class_temporality[
            instrument.__class__
        ] is AggregationTemporality.DELTA and not isinstance(instrument, _Asynchronous)

        return _ViewInstrumentMatch(
            view=view,
            instrument=instrument,
            instrument_class_aggregation=self._instrument_class_aggregation,
            cardinality_limit=cardinality_limit,
            reset_on_collect=reset_on_collect,
        )

    def _get_or_init_view_instrument_match(self, instrument: _Instrument) -> list[_ViewInstrumentMatch]:
        # Optimistically get the relevant views for the given instrument. Once set for a given
        # instrument, the mapping will never change

        if instrument in self._instrument_view_instrument_matches:
            return self._instrument_view_instrument_matches[instrument]

        with self._lock:
            # double check if it was set before we held the lock
            if instrument in self._instrument_view_instrument_matches:
                return self._instrument_view_instrument_matches[instrument]

            # not present, hold the lock and add a new mapping
            view_instrument_matches = []

            self._handle_view_instrument_match(instrument, view_instrument_matches)

            # if no view targeted the instrument, use the default
            if not view_instrument_matches:
                view_instrument_matches.append(self._new_view_instrument_match(_DEFAULT_VIEW, instrument))
            self._instrument_view_instrument_matches[instrument] = view_instrument_matches

            return view_instrument_matches

    def consume_measurement(self, measurement: Measurement, should_sample_exemplar: bool = True) -> None:
        for view_instrument_match in self._get_or_init_view_instrument_match(measurement.instrument):
            view_instrument_match.consume_measurement(measurement, should_sample_exemplar)

    def collect(self) -> MetricsData | None:
        # Use a list instead of yielding to prevent a slow reader from holding
        # SDK locks

        # While holding the lock, new _ViewInstrumentMatch can't be added from
        # another thread (so we are sure we collect all existing view).
        # However, instruments can still send measurements that will make it
        # into the individual aggregations; collection will acquire those locks
        # iteratively to keep locking as fine-grained as possible. One side
        # effect is that end times can be slightly skewed among the metric
        # streams produced by the SDK, but we still align the output timestamps
        # for a single instrument.

        collection_start_nanos = time_ns()

        with self._lock:
            instrumentation_scope_scope_metrics: dict[InstrumentationScope, ScopeMetrics] = {}

            instrument_matches_snapshot = list(self._instrument_view_instrument_matches.items())

            for (
                instrument,
                view_instrument_matches,
            ) in instrument_matches_snapshot:
                aggregation_temporality = self._instrument_class_temporality[instrument.__class__]

                metrics: list[Metric] = []

                for view_instrument_match in view_instrument_matches:
                    data_points = view_instrument_match.collect(aggregation_temporality, collection_start_nanos)

                    if data_points is None:
                        continue

                    if isinstance(
                        # pylint: disable=protected-access
                        view_instrument_match._aggregation,
                        _SumAggregation,
                    ):
                        data = Sum(
                            aggregation_temporality=aggregation_temporality,
                            data_points=data_points,
                            is_monotonic=isinstance(instrument, (Counter, ObservableCounter)),
                        )
                    elif isinstance(
                        # pylint: disable=protected-access
                        view_instrument_match._aggregation,
                        _LastValueAggregation,
                    ):
                        data = Gauge(data_points=data_points)
                    elif isinstance(
                        # pylint: disable=protected-access
                        view_instrument_match._aggregation,
                        _ExplicitBucketHistogramAggregation,
                    ):
                        data = Histogram(
                            data_points=data_points,
                            aggregation_temporality=aggregation_temporality,
                        )
                    elif isinstance(
                        # pylint: disable=protected-access
                        view_instrument_match._aggregation,
                        _DropAggregation,
                    ):
                        continue

                    elif isinstance(
                        # pylint: disable=protected-access
                        view_instrument_match._aggregation,
                        _ExponentialBucketHistogramAggregation,
                    ):
                        data = ExponentialHistogram(
                            data_points=data_points,
                            aggregation_temporality=aggregation_temporality,
                        )

                    metrics.append(
                        Metric(
                            # pylint: disable=protected-access
                            # pylint: disable=possibly-used-before-assignment
                            name=view_instrument_match._name,
                            description=view_instrument_match._description,
                            unit=view_instrument_match._instrument.unit,
                            data=data,
                        )
                    )

                if metrics:
                    if instrument.instrumentation_scope not in (instrumentation_scope_scope_metrics):
                        instrumentation_scope_scope_metrics[instrument.instrumentation_scope] = ScopeMetrics(
                            scope=instrument.instrumentation_scope,
                            metrics=metrics,
                            schema_url=instrument.instrumentation_scope.schema_url,
                        )
                    else:
                        instrumentation_scope_scope_metrics[instrument.instrumentation_scope].metrics.extend(metrics)

            if instrumentation_scope_scope_metrics:
                return MetricsData(
                    resource_metrics=[
                        ResourceMetrics(
                            resource=self._sdk_config.resource,
                            scope_metrics=list(instrumentation_scope_scope_metrics.values()),
                            schema_url=self._sdk_config.resource.schema_url,
                        )
                    ]
                )

            return None

    def _handle_view_instrument_match(
        self,
        instrument: _Instrument,
        view_instrument_matches: list["_ViewInstrumentMatch"],
    ) -> None:
        for view in self._sdk_config.views:
            # pylint: disable=protected-access
            if not view._match(instrument):
                continue

            if not self._check_view_instrument_compatibility(view, instrument):
                continue

            new_view_instrument_match = self._new_view_instrument_match(view, instrument)

            for existing_view_instrument_matches in self._instrument_view_instrument_matches.values():
                for existing_view_instrument_match in existing_view_instrument_matches:
                    if existing_view_instrument_match.conflicts(new_view_instrument_match):
                        _logger.warning(
                            "Views %s and %s will cause conflicting metrics identities",
                            existing_view_instrument_match._view,
                            new_view_instrument_match._view,
                        )

            view_instrument_matches.append(new_view_instrument_match)

    @staticmethod
    def _check_view_instrument_compatibility(view: View, instrument: _Instrument) -> bool:
        """
        Checks if a view and an instrument are compatible.

        Returns `true` if they are compatible and a `_ViewInstrumentMatch`
        object should be created, `false` otherwise.
        """

        result = True

        # pylint: disable=protected-access
        if isinstance(instrument, Asynchronous) and isinstance(
            view._aggregation,
            (
                ExplicitBucketHistogramAggregation,
                ExponentialBucketHistogramAggregation,
            ),
        ):
            _logger.warning(
                "View %s and instrument %s will produce semantic errors when matched, the view has not been applied.",
                view,
                instrument,
            )
            result = False

        return result