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

from __future__ import annotations

import collections
import enum
import inspect
import logging
import os
import threading
import time
import weakref
from abc import abstractmethod
from typing import (
    Generic,
    Protocol,
    TypeVar,
)

from opentelemetry.context import (
    _SUPPRESS_INSTRUMENTATION_KEY,
    attach,
    detach,
    set_value,
)
from opentelemetry.sdk._shared_internal._processor_metrics import (
    ProcessorMetricsT,
)
from opentelemetry.util._once import Once


class DuplicateFilter(logging.Filter):
    """Filter that can be applied to internal `logger`'s.

    Currently applied to `logger`s on the export logs path to prevent endlessly logging the same log
    in cases where logging itself is failing."""

    def filter(self, record):
        current_log = (
            record.module,
            record.levelno,
            record.msg,
            # We need to pick a time longer than the OTLP LogExporter timeout
            # which defaults to 10 seconds, but not pick something so long that
            # it filters out useful logs.
            time.time() // 20,
        )
        if current_log != getattr(self, "last_log", None):
            self.last_log = current_log  # pylint: disable=attribute-defined-outside-init
            return True
        # False means python's `logging` module will no longer process this log.
        return False


class BatchExportStrategy(enum.Enum):
    EXPORT_ALL = 0
    EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD = 1
    EXPORT_AT_LEAST_ONE_BATCH = 2


Telemetry = TypeVar("Telemetry")


class Exporter(Protocol[Telemetry]):
    @abstractmethod
    def export(self, batch: list[Telemetry], /):
        raise NotImplementedError

    @abstractmethod
    def shutdown(self):
        raise NotImplementedError


_logger = logging.getLogger(__name__)
_logger.addFilter(DuplicateFilter())


class BatchProcessor(Generic[Telemetry]):
    """This class can be used with exporter's that implement the above
    Exporter interface to buffer and send telemetry in batch through
     the exporter."""

    def __init__(
        self,
        exporter: Exporter[Telemetry],
        schedule_delay_millis: float,
        max_export_batch_size: int,
        export_timeout_millis: float,
        max_queue_size: int,
        exporting: str,
        metrics: ProcessorMetricsT,
    ):
        self._bsp_reset_once = Once()
        self._exporter = exporter
        self._max_queue_size = max_queue_size
        self._schedule_delay_millis = schedule_delay_millis
        self._schedule_delay = schedule_delay_millis / 1e3
        self._max_export_batch_size = max_export_batch_size
        # Not used. No way currently to pass timeout to export.
        # TODO(https://github.com/open-telemetry/opentelemetry-python/issues/4555): figure out what this should do.
        self._export_timeout_millis = export_timeout_millis
        # Deque is thread safe.
        self._queue = collections.deque([], max_queue_size)
        self._worker_thread = threading.Thread(
            name=f"OtelBatch{exporting}RecordProcessor",
            target=self.worker,
            daemon=True,
        )
        self._exporting = exporting

        self._shutdown = False
        self._shutdown_timeout_exceeded = False
        self._export_lock = threading.Lock()
        self._worker_awaken = threading.Event()
        self._worker_thread.start()
        if hasattr(os, "register_at_fork"):
            weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
            os.register_at_fork(after_in_child=lambda: weak_reinit()())  # pyright: ignore[reportOptionalCall] pylint: disable=unnecessary-lambda
        self._pid = os.getpid()

        metrics.register_queue_size(lambda: len(self._queue))
        self._metrics = metrics

    def _should_export_batch(self, batch_strategy: BatchExportStrategy, num_iterations: int) -> bool:
        if not self._queue or self._shutdown_timeout_exceeded:
            return False
        # Always continue to export while queue length exceeds max batch size.
        if len(self._queue) >= self._max_export_batch_size:
            return True
        if batch_strategy is BatchExportStrategy.EXPORT_ALL:
            return True
        if batch_strategy is BatchExportStrategy.EXPORT_AT_LEAST_ONE_BATCH:
            return num_iterations == 0
        return False

    def _at_fork_reinit(self):
        self._export_lock = threading.Lock()
        self._worker_awaken = threading.Event()
        self._queue.clear()
        self._worker_thread = threading.Thread(
            name=f"OtelBatch{self._exporting}RecordProcessor",
            target=self.worker,
            daemon=True,
        )
        self._worker_thread.start()
        self._pid = os.getpid()

    def worker(self):
        while not self._shutdown:
            # Lots of strategies in the spec for setting next timeout.
            # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/sdk.md#batching-processor.
            # Shutdown will interrupt this sleep. Emit will interrupt this sleep only if the queue is bigger then threshold.
            sleep_interrupted = self._worker_awaken.wait(self._schedule_delay)
            if self._shutdown:
                break
            self._export(
                BatchExportStrategy.EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD
                if sleep_interrupted
                else BatchExportStrategy.EXPORT_AT_LEAST_ONE_BATCH
            )
            self._worker_awaken.clear()
        self._export(BatchExportStrategy.EXPORT_ALL)

    def _export(self, batch_strategy: BatchExportStrategy) -> None:
        with self._export_lock:
            iteration = 0
            # We could see concurrent export calls from worker and force_flush. We call _should_export_batch
            # once the lock is obtained to see if we still need to make the requested export.
            while self._should_export_batch(batch_strategy, iteration):
                iteration += 1
                token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
                count = min(
                    self._max_export_batch_size,
                    len(self._queue),
                )
                # Oldest records are at the back, so pop from there.
                batch = [self._queue.pop() for _ in range(count)]
                # Record on submission to the exporter.
                self._metrics.finish_items(count)
                try:
                    self._exporter.export(batch)
                except Exception:  # pylint: disable=broad-exception-caught
                    _logger.exception("Exception while exporting %s.", self._exporting)
                detach(token)

    def emit(self, data: Telemetry) -> None:
        if self._shutdown:
            _logger.info("Shutdown called, ignoring %s.", self._exporting)
            self._metrics.drop_items(1, "already_shutdown")
            return
        if self._pid != os.getpid():
            self._bsp_reset_once.do_once(self._at_fork_reinit)
        if len(self._queue) == self._max_queue_size:
            _logger.warning("Queue full, dropping %s.", self._exporting)
            self._metrics.drop_items(1)
        # This will drop a log from the right side if the queue is at _max_queue_size.
        self._queue.appendleft(data)
        if len(self._queue) >= self._max_export_batch_size and not self._worker_awaken.is_set():
            self._worker_awaken.set()

    def shutdown(self, timeout_millis: int = 30000):
        if self._shutdown:
            return
        shutdown_should_end = time.time() + (timeout_millis / 1000)
        # Causes emit to reject telemetry and makes force_flush a no-op.
        self._shutdown = True
        # Interrupts sleep in the worker if it's sleeping.
        self._worker_awaken.set()
        self._worker_thread.join(timeout_millis / 1000)
        # Stops worker thread from calling export again if queue is still not empty.
        self._shutdown_timeout_exceeded = True
        # We want to shutdown immediately only if we already waited `timeout_secs`.
        # Otherwise we pass the remaining timeout to the exporter.
        # Some exporter's shutdown support a timeout param.
        if "timeout_millis" in inspect.getfullargspec(self._exporter.shutdown).args:
            remaining_millis = (shutdown_should_end - time.time()) * 1000
            self._exporter.shutdown(timeout_millis=max(0, remaining_millis))  # type: ignore
        else:
            self._exporter.shutdown()
        # Worker thread **should** be finished at this point, because we called shutdown on the exporter,
        # and set shutdown_is_occuring to prevent further export calls. It's possible that a single export
        # call is ongoing and the thread isn't finished. In this case we will return instead of waiting on
        # the thread to finish.

    # TODO: Fix force flush so the timeout is used https://github.com/open-telemetry/opentelemetry-python/issues/4568.
    def force_flush(self, timeout_millis: int | None = None) -> bool:
        if self._shutdown:
            return False
        # Blocking call to export.
        self._export(BatchExportStrategy.EXPORT_ALL)
        return True
