# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import logging
import os
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, overload
from opentelemetry.exporter.otlp.common import http as _http
from opentelemetry.exporter.otlp.common._exporter_metrics import (
create_exporter_metrics,
)
from opentelemetry.exporter.otlp.proto.common._log_encoder import encode_logs
from opentelemetry.exporter.otlp.proto.http import Compression
from opentelemetry.exporter.otlp.proto.http._common import (
_DEFAULT_MAX_REQUEST_SIZE,
RequestPayloadTooLargeError,
_build_transport,
_is_request_too_large,
_load_session_from_envvar,
_normalize_compression,
_resolve_compression,
_resolve_endpoint,
_resolve_headers,
_resolve_timeout,
)
from opentelemetry.metrics import MeterProvider
from opentelemetry.sdk._logs import ReadableLogRecord
from opentelemetry.sdk._logs.export import (
LogRecordExporter,
LogRecordExportResult,
)
from opentelemetry.sdk._shared_internal import DuplicateFilter
from opentelemetry.sdk.environment_variables import (
_OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER,
OTEL_EXPORTER_OTLP_LOGS_CERTIFICATE,
OTEL_EXPORTER_OTLP_LOGS_CLIENT_CERTIFICATE,
OTEL_EXPORTER_OTLP_LOGS_CLIENT_KEY,
OTEL_EXPORTER_OTLP_LOGS_COMPRESSION,
OTEL_EXPORTER_OTLP_LOGS_ENDPOINT,
OTEL_EXPORTER_OTLP_LOGS_HEADERS,
OTEL_EXPORTER_OTLP_LOGS_TIMEOUT,
OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED,
)
from opentelemetry.semconv._incubating.attributes.otel_attributes import (
OtelComponentTypeValues,
)
from opentelemetry.semconv.attributes.http_attributes import (
HTTP_RESPONSE_STATUS_CODE,
)
if TYPE_CHECKING:
import requests
from opentelemetry.exporter.http.transport._base import BaseHTTPTransport
_logger = logging.getLogger(__name__)
# This prevents logs generated when a log fails to be written to generate another log which fails to be written etc. etc.
_logger.addFilter(DuplicateFilter())
DEFAULT_COMPRESSION = Compression.NoCompression
DEFAULT_ENDPOINT = "http://localhost:4318/"
DEFAULT_LOGS_EXPORT_PATH = "v1/logs"
DEFAULT_TIMEOUT = 10 # in seconds
[docs]
class OTLPLogExporter(LogRecordExporter):
@overload
def __init__(
self,
endpoint: str | None = None,
certificate_file: str | None = None,
client_key_file: str | None = None,
client_certificate_file: str | None = None,
headers: Mapping[str, str] | None = None,
timeout: float | None = None,
compression: Compression | _http.Compression | None = None,
session: requests.Session | None = None,
*,
max_request_size: int | None = None,
meter_provider: MeterProvider | None = None,
) -> None: ...
@overload
def __init__(
self,
endpoint: str | None = None,
certificate_file: None = None,
client_key_file: None = None,
client_certificate_file: None = None,
headers: Mapping[str, str] | None = None,
timeout: float | None = None,
compression: Compression | _http.Compression | None = None,
session: requests.Session | None = None,
*,
max_request_size: int | None = None,
meter_provider: MeterProvider | None = None,
_transport: BaseHTTPTransport,
) -> None: ...
def __init__(
self,
endpoint: str | None = None,
certificate_file: str | None = None,
client_key_file: str | None = None,
client_certificate_file: str | None = None,
headers: Mapping[str, str] | None = None,
timeout: float | None = None,
compression: Compression | _http.Compression | None = None,
session: requests.Session | None = None,
*,
max_request_size: int | None = None,
meter_provider: MeterProvider | None = None,
_transport: BaseHTTPTransport | None = None,
) -> None:
"""OTLP HTTP log exporter.
Args:
endpoint: Full URL of the OTLP/HTTP signal endpoint, including the signal
path. Example: ``http://collector:4318/v1/logs``. For a base URL without a
signal path, set the ``OTEL_EXPORTER_OTLP_ENDPOINT`` environment variable.
certificate_file: Path to the CA certificate file for TLS.
client_key_file: Path to the client key file for mTLS.
client_certificate_file: Path to the client certificate file for mTLS.
headers: Headers to send with each export request.
timeout: Timeout in seconds for each export request.
compression: Compression to use; one of none, gzip, deflate.
session: Requests session to use at export.
max_request_size: Maximum size in bytes of a serialized request,
measured before compression. A request exceeding this size is
dropped before being sent. Defaults to 64 MiB; a value of 0 (or
any non-positive value) disables the limit. Batch processors
group log records by count rather than serialized size, so a
batch whose serialized request exceeds this limit is dropped as
a whole and recorded as a failed export; reduce the processor's
``max_export_batch_size`` (or raise/disable this limit) if
batches may approach it.
meter_provider: MeterProvider used for the exporter's own metrics.
"""
self._endpoint = endpoint or _resolve_endpoint(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, DEFAULT_LOGS_EXPORT_PATH)
self._compression = _normalize_compression(compression) or _resolve_compression(
OTEL_EXPORTER_OTLP_LOGS_COMPRESSION
)
transport = _transport or _build_transport(
certificate_file,
client_key_file,
client_certificate_file,
OTEL_EXPORTER_OTLP_LOGS_CERTIFICATE,
OTEL_EXPORTER_OTLP_LOGS_CLIENT_KEY,
OTEL_EXPORTER_OTLP_LOGS_CLIENT_CERTIFICATE,
session=session or _load_session_from_envvar(_OTEL_PYTHON_EXPORTER_OTLP_HTTP_LOGS_CREDENTIAL_PROVIDER),
)
self._max_request_size = _DEFAULT_MAX_REQUEST_SIZE if max_request_size is None else max_request_size
self._client = _http._OTLPHTTPClient(
transport=transport,
endpoint=self._endpoint,
kind="logs",
timeout=timeout if timeout is not None else _resolve_timeout(OTEL_EXPORTER_OTLP_LOGS_TIMEOUT),
compression=self._compression,
headers=_resolve_headers(headers, OTEL_EXPORTER_OTLP_LOGS_HEADERS),
logger=_logger,
)
self._shutdown = False
self._metrics = create_exporter_metrics(
OtelComponentTypeValues.OTLP_HTTP_LOG_EXPORTER,
"logs",
self._endpoint,
meter_provider,
os.environ.get(OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED, "").strip().lower() == "true",
)
[docs]
def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult:
if self._shutdown:
_logger.warning("Exporter already shutdown, ignoring batch")
return LogRecordExportResult.FAILURE
with self._metrics.export_operation(len(batch)) as result:
try:
serialized_data = encode_logs(batch).SerializeToString()
# pylint: disable-next=broad-exception-caught
except Exception as error:
_logger.error("Failed to encode logs batch: %s", error)
result.error = error
return LogRecordExportResult.FAILURE
if _is_request_too_large(serialized_data, self._max_request_size):
_logger.warning(
"Dropping logs batch: serialized size %d bytes exceeds max_request_size %d bytes.",
len(serialized_data),
self._max_request_size,
)
result.error = RequestPayloadTooLargeError(
f"Serialized logs request size {len(serialized_data)} "
f"bytes exceeds max_request_size "
f"{self._max_request_size} bytes."
)
return LogRecordExportResult.FAILURE
export_result = self._client.export(serialized_data)
if not export_result.success:
result.error = export_result.error
result.error_attrs = (
{HTTP_RESPONSE_STATUS_CODE: export_result.status_code}
if export_result.status_code is not None
else None
)
return LogRecordExportResult.FAILURE
return LogRecordExportResult.SUCCESS
[docs]
def force_flush(self, timeout_millis: int = 10_000) -> bool:
"""Nothing is buffered in this exporter, so this method does nothing."""
return True
[docs]
def shutdown(self):
if self._shutdown:
_logger.warning("Exporter already shutdown, ignoring call")
return
self._shutdown = True
self._client.shutdown()