# Source code for opentelemetry.sdk.trace.export

    # 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.
    from __future__ import annotations

    import logging
    import sys
    import typing
    from enum import Enum
    from os import environ, linesep

    from opentelemetry.context import (
        _SUPPRESS_INSTRUMENTATION_KEY,
        Context,
        attach,
        detach,
        set_value,
    )
    from opentelemetry.sdk._shared_internal import BatchProcessor
    from opentelemetry.sdk.environment_variables import (
        OTEL_BSP_EXPORT_TIMEOUT,
        OTEL_BSP_MAX_EXPORT_BATCH_SIZE,
        OTEL_BSP_MAX_QUEUE_SIZE,
        OTEL_BSP_SCHEDULE_DELAY,
    )
    from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor

    _DEFAULT_SCHEDULE_DELAY_MILLIS = 5000
    _DEFAULT_MAX_EXPORT_BATCH_SIZE = 512
    _DEFAULT_EXPORT_TIMEOUT_MILLIS = 30000
    _DEFAULT_MAX_QUEUE_SIZE = 2048
    _ENV_VAR_INT_VALUE_ERROR_MESSAGE = (
        "Unable to parse value for %s as integer. Defaulting to %s."
    )

    logger = logging.getLogger(__name__)

    class SpanExportResult(Enum):
        SUCCESS = 0
        FAILURE = 1

    class SpanExporter:
        """Interface for exporting spans.

        Interface to be implemented by services that want to export spans recorded
        in their own format.

        To export data this MUST be registered to the :class`opentelemetry.sdk.trace.Tracer` using a
        `SimpleSpanProcessor` or a `BatchSpanProcessor`.
        """

        def export(
            self, spans: typing.Sequence[ReadableSpan]
        ) -> "SpanExportResult":  # pyright: ignore[reportReturnType]
            """Exports a batch of telemetry data.

            Args:
                spans: The list of `opentelemetry.trace.Span` objects to be exported

            Returns:
                The result of the export
            """

        def shutdown(self) -> None:
            """Shuts down the exporter.

            Called when the SDK is shut down.
            """

        def force_flush(self, timeout_millis: int = 30000) -> bool:  # pyright: ignore[reportReturnType]
            """Hint to ensure that the export of any spans the exporter has received
            prior to the call to ForceFlush SHOULD be completed as soon as possible, preferably
            before returning from this method.
            """

    class SimpleSpanProcessor(SpanProcessor):
        """Simple SpanProcessor implementation.

        SimpleSpanProcessor is an implementation of `SpanProcessor` that
        passes ended spans directly to the configured `SpanExporter`.
        """

        def __init__(self, span_exporter: SpanExporter):
            self.span_exporter = span_exporter

        def on_start(
            self, span: Span, parent_context: typing.Optional[Context] = None
        ) -> None:
            pass

        def on_end(self, span: ReadableSpan) -> None:
            if not (span.context and span.context.trace_flags.sampled):
                return
            token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
            try:
                self.span_exporter.export((span,))
            # pylint: disable=broad-exception-caught
            except Exception:
                logger.exception("Exception while exporting Span.")
            detach(token)

        def shutdown(self) -> None:
            self.span_exporter.shutdown()

        def force_flush(self, timeout_millis: int = 30000) -> bool:
            # pylint: disable=unused-argument
            return True

    class BatchSpanProcessor(SpanProcessor):
        """Batch span processor implementation.

        `BatchSpanProcessor` is an implementation of `SpanProcessor` that
        batches ended spans and pushes them to the configured `SpanExporter`.

        `BatchSpanProcessor` is configurable with the following environment
        variables which correspond to constructor parameters:

        - :envvar:`OTEL_BSP_SCHEDULE_DELAY`
        - :envvar:`OTEL_BSP_MAX_QUEUE_SIZE`
        - :envvar:`OTEL_BSP_MAX_EXPORT_BATCH_SIZE`
        - :envvar:`OTEL_BSP_EXPORT_TIMEOUT`

        All the logic for emitting spans, shutting down etc. resides in the `BatchProcessor` class.
        """

        def __init__(
            self,
            span_exporter: SpanExporter,
            max_queue_size: int | None = None,
            schedule_delay_millis: float | None = None,
            max_export_batch_size: int | None = None,
            export_timeout_millis: float | None = None,
        ):
            if max_queue_size is None:
                max_queue_size = BatchSpanProcessor._default_max_queue_size()

            if schedule_delay_millis is None:
                schedule_delay_millis = (
                    BatchSpanProcessor._default_schedule_delay_millis()
                )

            if max_export_batch_size is None:
                max_export_batch_size = (
                    BatchSpanProcessor._default_max_export_batch_size()
                )

            # Not used. No way currently to pass timeout to export.
            if export_timeout_millis is None:
                export_timeout_millis = (
                    BatchSpanProcessor._default_export_timeout_millis()
                )

            BatchSpanProcessor._validate_arguments(
                max_queue_size, schedule_delay_millis, max_export_batch_size
            )

            self._batch_processor = BatchProcessor(
                span_exporter,
                schedule_delay_millis,
                max_export_batch_size,
                export_timeout_millis,
                max_queue_size,
                "Span",
            )

        # Added for backward compatibility. Not recommended to directly access/use underlying exporter.
        @property
        def span_exporter(self):
            return self._batch_processor._exporter  # pylint: disable=protected-access

        def on_start(
            self, span: Span, parent_context: Context | None = None
        ) -> None:
            pass

        def on_end(self, span: ReadableSpan) -> None:
            if not (span.context and span.context.trace_flags.sampled):
                return
            self._batch_processor.emit(span)

        def shutdown(self):
            return self._batch_processor.shutdown()

        def force_flush(self, timeout_millis: typing.Optional[int] = None) -> bool:
            return self._batch_processor.force_flush(timeout_millis)

        @staticmethod
        def _default_max_queue_size():
            try:
                return int(
                    environ.get(OTEL_BSP_MAX_QUEUE_SIZE, _DEFAULT_MAX_QUEUE_SIZE)
                )
            except ValueError:
                logger.exception(
                    _ENV_VAR_INT_VALUE_ERROR_MESSAGE,
                    OTEL_BSP_MAX_QUEUE_SIZE,
                    _DEFAULT_MAX_QUEUE_SIZE,
                )
                return _DEFAULT_MAX_QUEUE_SIZE

        @staticmethod
        def _default_schedule_delay_millis():
            try:
                return int(
                    environ.get(
                        OTEL_BSP_SCHEDULE_DELAY, _DEFAULT_SCHEDULE_DELAY_MILLIS
                    )
                )
            except ValueError:
                logger.exception(
                    _ENV_VAR_INT_VALUE_ERROR_MESSAGE,
                    OTEL_BSP_SCHEDULE_DELAY,
                    _DEFAULT_SCHEDULE_DELAY_MILLIS,
                )
                return _DEFAULT_SCHEDULE_DELAY_MILLIS

        @staticmethod
        def _default_max_export_batch_size():
            try:
                return int(
                    environ.get(
                        OTEL_BSP_MAX_EXPORT_BATCH_SIZE,
                        _DEFAULT_MAX_EXPORT_BATCH_SIZE,
                    )
                )
            except ValueError:
                logger.exception(
                    _ENV_VAR_INT_VALUE_ERROR_MESSAGE,
                    OTEL_BSP_MAX_EXPORT_BATCH_SIZE,
                    _DEFAULT_MAX_EXPORT_BATCH_SIZE,
                )
                return _DEFAULT_MAX_EXPORT_BATCH_SIZE

        @staticmethod
        def _default_export_timeout_millis():
            try:
                return int(
                    environ.get(
                        OTEL_BSP_EXPORT_TIMEOUT, _DEFAULT_EXPORT_TIMEOUT_MILLIS
                    )
                )
            except ValueError:
                logger.exception(
                    _ENV_VAR_INT_VALUE_ERROR_MESSAGE,
                    OTEL_BSP_EXPORT_TIMEOUT,
                    _DEFAULT_EXPORT_TIMEOUT_MILLIS,
                )
                return _DEFAULT_EXPORT_TIMEOUT_MILLIS

        @staticmethod
        def _validate_arguments(
            max_queue_size, schedule_delay_millis, max_export_batch_size
        ):
            if max_queue_size <= 0:
                raise ValueError("max_queue_size must be a positive integer.")

            if schedule_delay_millis <= 0:
                raise ValueError("schedule_delay_millis must be positive.")

            if max_export_batch_size <= 0:
                raise ValueError(
                    "max_export_batch_size must be a positive integer."
                )

            if max_export_batch_size > max_queue_size:
                raise ValueError(
                    "max_export_batch_size must be less than or equal to max_queue_size."
                )

    class ConsoleSpanExporter(SpanExporter):
        """Implementation of :class:`SpanExporter` that prints spans to the
        console.

        This class can be used for diagnostic purposes. It prints the exported
        spans to the console STDOUT.
        """

        def __init__(
            self,
            service_name: str | None = None,
            out: typing.IO = sys.stdout,
            formatter: typing.Callable[
                [ReadableSpan], str
            ] = lambda span: span.to_json() + linesep,
        ):
            self.out = out
            self.formatter = formatter
            self.service_name = service_name

        def export(self, spans: typing.Sequence[ReadableSpan]) -> SpanExportResult:
            for span in spans:
                self.out.write(self.formatter(span))
            self.out.flush()
            return SpanExportResult.SUCCESS

        def force_flush(self, timeout_millis: int = 30000) -> bool:
            return True