# Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 from __future__ import annotations import enum import gzip import logging import math import random import threading import time import zlib from collections.abc import Mapping from dataclasses import dataclass from datetime import timezone from email.utils import parsedate_to_datetime from http import HTTPStatus from io import BytesIO from typing import TYPE_CHECKING, Final, Literal if TYPE_CHECKING: from opentelemetry.exporter.http.transport._base import ( BaseHTTPResult, BaseHTTPTransport, ) _logger = logging.getLogger(__name__) _MAX_RETRIES: Final[int] = 6 _DEFAULT_TIMEOUT: Final[float] = 10.0 _DEFAULT_JITTER: Final[float] = 0.2 _RETRYABLE_STATUS_CODES: Final[frozenset[int]] = frozenset( { HTTPStatus.TOO_MANY_REQUESTS.value, HTTPStatus.BAD_GATEWAY.value, HTTPStatus.SERVICE_UNAVAILABLE.value, HTTPStatus.GATEWAY_TIMEOUT.value, } ) def _is_retryable(status_code: int | None) -> bool: return status_code in _RETRYABLE_STATUS_CODES def _extract_retry_after(result: BaseHTTPResult) -> float | None: """Parse the ``Retry-After`` header (RFC 7231).""" try: value = result.headers().get("retry-after") # pylint: disable-next=broad-exception-caught except Exception: return None if value is None: return None value = value.strip() # delay-seconds: a non-negative decimal integer. try: seconds = float(value) except ValueError: pass else: return max(seconds, 0.0) if math.isfinite(seconds) else None # HTTP-date: wait until the indicated absolute time. try: retry_at = parsedate_to_datetime(value) except (TypeError, ValueError): return None if retry_at.tzinfo is None: retry_at = retry_at.replace(tzinfo=timezone.utc) return max(retry_at.timestamp() - time.time(), 0.0) class Compression(enum.Enum): NONE = "none" DEFLATE = "deflate" GZIP = "gzip" @staticmethod def from_str(value: str) -> Compression: match value.strip().lower(): case "none": return Compression.NONE case "deflate": return Compression.DEFLATE case "gzip": return Compression.GZIP case _: raise ValueError(f"Invalid compression type: {value!r}. Expected one of: 'none', 'deflate', 'gzip'.") @dataclass(slots=True, frozen=True) class _ExportResult: """Outcome of an OTLP export attempt, including retry exhaustion.""" success: bool status_code: int | None reason: str | None error: Exception | None class _OTLPHTTPClient: """Sends serialized OTLP payloads over HTTP with retry logic. Compression, backoff, and connection-error recovery are handled internally. Callers interact through the :meth:`export` and :meth:`close` methods. """ def __init__( self, transport: BaseHTTPTransport, endpoint: str, kind: Literal["spans", "logs", "metrics"], timeout: float = _DEFAULT_TIMEOUT, compression: Compression = Compression.NONE, headers: Mapping[str, str] | None = None, jitter: float = _DEFAULT_JITTER, logger: logging.Logger | None = None, ) -> None: self._transport = transport self._endpoint = endpoint self._timeout = timeout self._compression = compression self._headers = dict(headers) if headers is not None else {} if self._compression is not Compression.NONE and not any( key.lower() == "content-encoding" for key in self._headers ): self._headers["Content-Encoding"] = self._compression.value self._kind = kind self._jitter = min(max(jitter, 0.0), 1.0) self._logger = logger if logger is not None else _logger self._shutdown = False self._shutdown_event = threading.Event() def _compute_backoff(self, retry: int) -> float: return 2**retry * random.uniform(1 - self._jitter, 1 + self._jitter) def _compress(self, serialized_data: bytes) -> bytes: if self._compression is Compression.GZIP: buf = BytesIO() with gzip.GzipFile(fileobj=buf, mode="w") as gz: gz.write(serialized_data) return buf.getvalue() if self._compression is Compression.DEFLATE: return zlib.compress(serialized_data) return serialized_data def _submit(self, data: bytes, timeout: float) -> BaseHTTPResult: deadline = time.time() + timeout result = self._transport.request( "POST", self._endpoint, headers=self._headers, data=data, timeout=timeout, ) if ( result.error is not None and self._transport.is_connection_error(result.error) and (remaining := deadline - time.time()) > 0 ): # Immediately retry connection errors once without backoff. These # usually indicate a stale pooled connection that the transport will # reestablish on the next attempt. result = self._transport.request( "POST", self._endpoint, headers=self._headers, data=data, timeout=remaining, ) return result def export(self, data: bytes) -> _ExportResult: """Export a serialized payload, retrying on transient failures. :param data: Serialized bytes to send. :returns: An :class:`ExportResult` indicating success or the reason for failure. """ data = self._compress(data) deadline = time.time() + self._timeout for retry in range(_MAX_RETRIES): backoff = self._compute_backoff(retry) status_code: int | None = None reason: str | None = None export_error: Exception | None retryable: bool try: result = self._submit(data, max(deadline - time.time(), 0.0)) # pylint: disable-next=broad-exception-caught except Exception as error: export_error = error retryable = False else: status_code = result.status_code reason = result.reason if status_code is not None and 200 <= status_code < 400: return _ExportResult(True, status_code, reason, None) export_error = result.error retryable = ( _is_retryable(status_code) if status_code else self._transport.is_connection_error(result.error) ) if retryable and status_code is not None and (retry_after := _extract_retry_after(result)) is not None: backoff = retry_after if not retryable: self._logger.error( "Failed to export %s batch code: %s, reason: %s", self._kind, status_code, reason or export_error or "unknown", ) return _ExportResult(False, status_code, reason, export_error) if retry + 1 == _MAX_RETRIES or backoff > (deadline - time.time()) or self._shutdown_event.is_set(): self._logger.error( "Failed to export %s batch due to timeout, max retries or shutdown.", self._kind, ) return _ExportResult(False, status_code, reason, export_error) self._logger.warning( "Transient error %s encountered while exporting %s batch, retrying in %.2fs.", reason or export_error, self._kind, backoff, ) shutdown = self._shutdown_event.wait(backoff) if shutdown: self._logger.warning("Shutdown in progress, aborting retry.") break return _ExportResult(False, None, None, None) def shutdown(self) -> None: """Shutdown the client.""" if self._shutdown: self._logger.warning("OTLP client already shutdown, ignoring call") return self._shutdown = True self._shutdown_event.set() self._transport.close()