# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0
"""OTLP Span Exporter"""
import logging
from collections.abc import Iterable, Sequence
from collections.abc import Sequence as TypingSequence
from os import environ
from grpc import ChannelCredentials, Compression, StatusCode
from opentelemetry.exporter.otlp.proto.common.trace_encoder import (
encode_spans,
)
from opentelemetry.exporter.otlp.proto.grpc.exporter import ( # noqa: F401
OTLPExporterMixin,
_get_credentials,
environ_to_compression,
get_resource_data,
)
from opentelemetry.metrics import MeterProvider
from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import (
ExportTraceServiceRequest,
)
from opentelemetry.proto.collector.trace.v1.trace_service_pb2_grpc import (
TraceServiceStub,
)
from opentelemetry.proto.common.v1.common_pb2 import ( # noqa: F401
InstrumentationScope,
)
from opentelemetry.proto.trace.v1.trace_pb2 import ( # noqa: F401
ResourceSpans,
ScopeSpans,
Status,
)
from opentelemetry.proto.trace.v1.trace_pb2 import ( # noqa: F401
Span as CollectorSpan,
)
from opentelemetry.sdk.environment_variables import (
_OTEL_PYTHON_EXPORTER_OTLP_GRPC_TRACES_CREDENTIAL_PROVIDER,
OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE,
OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE,
OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY,
OTEL_EXPORTER_OTLP_TRACES_COMPRESSION,
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT,
OTEL_EXPORTER_OTLP_TRACES_HEADERS,
OTEL_EXPORTER_OTLP_TRACES_INSECURE,
OTEL_EXPORTER_OTLP_TRACES_TIMEOUT,
)
from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
from opentelemetry.semconv._incubating.attributes.otel_attributes import (
OtelComponentTypeValues,
)
logger = logging.getLogger(__name__)
# pylint: disable=no-member
[docs]
class OTLPSpanExporter(
SpanExporter,
OTLPExporterMixin[
Sequence[ReadableSpan],
ExportTraceServiceRequest,
SpanExportResult,
TraceServiceStub,
],
):
# pylint: disable=unsubscriptable-object
"""OTLP span exporter
Args:
endpoint: OpenTelemetry Collector receiver endpoint
insecure: Connection type
credentials: Credentials object for server authentication
headers: Headers to send when exporting
timeout: Backend request timeout in seconds
compression: gRPC compression method to use
"""
def __init__(
self,
endpoint: str | None = None,
insecure: bool | None = None,
credentials: ChannelCredentials | None = None,
headers: TypingSequence[tuple[str, str]]
| dict[str, str]
| str
| None = None,
timeout: float | None = None,
compression: Compression | None = None,
channel_options: tuple[tuple[str, str]] | None = None,
retryable_error_codes: Iterable[StatusCode] | None = None,
*,
meter_provider: MeterProvider | None = None,
):
insecure_spans = environ.get(OTEL_EXPORTER_OTLP_TRACES_INSECURE)
if insecure is None and insecure_spans is not None:
insecure = insecure_spans.lower() == "true"
if (
not insecure
and environ.get(OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE) is not None
):
credentials = _get_credentials(
credentials,
_OTEL_PYTHON_EXPORTER_OTLP_GRPC_TRACES_CREDENTIAL_PROVIDER,
OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE,
OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY,
OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE,
)
environ_timeout = environ.get(OTEL_EXPORTER_OTLP_TRACES_TIMEOUT)
environ_timeout = (
float(environ_timeout) if environ_timeout is not None else None
)
compression = (
environ_to_compression(OTEL_EXPORTER_OTLP_TRACES_COMPRESSION)
if compression is None
else compression
)
OTLPExporterMixin.__init__(
self,
stub=TraceServiceStub,
result=SpanExportResult,
endpoint=endpoint
or environ.get(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT),
insecure=insecure,
credentials=credentials,
headers=headers or environ.get(OTEL_EXPORTER_OTLP_TRACES_HEADERS),
timeout=timeout or environ_timeout,
compression=compression,
channel_options=channel_options,
retryable_error_codes=retryable_error_codes,
component_type=OtelComponentTypeValues.OTLP_GRPC_SPAN_EXPORTER,
signal="traces",
meter_provider=meter_provider,
)
def _translate_data(
self, data: Sequence[ReadableSpan]
) -> ExportTraceServiceRequest:
return encode_spans(data)
def _count_data(self, data: Sequence[ReadableSpan]):
return len(data)
[docs]
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
return self._export(spans)
[docs]
def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
OTLPExporterMixin.shutdown(self, timeout_millis=timeout_millis)
[docs]
def force_flush(self, timeout_millis: int = 30000) -> bool:
"""Nothing is buffered in this exporter, so this method does nothing."""
return True
@property
def _exporting(self):
return "traces"