Add support for setting OTLP transport

dnstapir/opentelemetry.py

@@ -1,33 +1,95 @@
 import logging
+import re
+from typing import Annotated, Any
 
 from fastapi import FastAPI
 from opentelemetry import metrics, trace
-from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
-from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as OTLPMetricExporterGrpc
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as OTLPSpanExporterGrpc
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPMetricExporterHttp
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as OTLPSpanExporterHttp
 from opentelemetry.instrumentation.botocore import BotocoreInstrumentor
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.instrumentation.httpx import HTTPX2ClientInstrumentor
 from opentelemetry.instrumentation.pymongo import PymongoInstrumentor
 from opentelemetry.instrumentation.redis import RedisInstrumentor
 from opentelemetry.sdk.metrics import MeterProvider
-from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader
+from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, MetricExporter, PeriodicExportingMetricReader
 from opentelemetry.sdk.resources import SERVICE_NAME, Resource
 from opentelemetry.sdk.trace import TracerProvider
-from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
-from pydantic import AnyHttpUrl, BaseModel
+from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter, SpanExporter
+from pydantic import AliasChoices, BaseModel, Field, UrlConstraints, field_validator
+from pydantic_core import Url
 
 logger = logging.getLogger(__name__)
 
+OtlpUrl = Annotated[Url, UrlConstraints(allowed_schemes=["http", "https", "grpc"])]
+
 DEFAULT_OTLP_SERVICE_NAME = "dnstapir"
+DEFAULT_OTLP_GRPC_PORT = 4317
 
 _configured = False
 
 
 class OtlpSettings(BaseModel):
     service_name: str | None = None
-    spans_endpoint: AnyHttpUrl | None = None
-    metrics_endpoint: AnyHttpUrl | None = None
-    insecure: bool = False
+    spans_endpoint: OtlpUrl | None = Field(
+        default=None,
+        description="The endpoint for OTLP spans, e.g., 'grpc://localhost:4317' or 'http://localhost:4318/v1/traces'",
+    )
+    metrics_endpoint: OtlpUrl | None = Field(
+        default=None,
+        description="The endpoint for OTLP metrics, e.g., 'grpc://localhost:4317' or 'http://localhost:4318/v1/metrics'",
+    )
+    insecure_grpc: bool | None = Field(
+        default=None,
+        validation_alias=AliasChoices("insecure_grpc", "insecure"),
+        description="Whether to use an insecure connection for OTLP gRPC endpoints (default is a TLS connection)",
+    )
+
+    @field_validator("spans_endpoint", "metrics_endpoint", mode="before")
+    @classmethod
+    def grpc_host_port(cls, value: Any) -> Any:
+        # If the value is a bare host:port string, prepend "grpc://" to make it a valid URL
+        if isinstance(value, str) and (match := re.match(r"^([\w\.\-]+:\d+)$", value)):
+            return f"grpc://{match.group(1)}"
+        return value
+
+    @field_validator("spans_endpoint", "metrics_endpoint", mode="after")
+    @classmethod
+    def check_grpc_path(cls, value: OtlpUrl | None) -> OtlpUrl | None:
+        if isinstance(value, Url) and value.scheme == "grpc":
+            if value.path and value.path != "/":
+                raise ValueError("gRPC path not supported")
+            if value.query:
+                raise ValueError("gRPC query parameter not supported")
+        return value
+
+    def get_spans_exporter(self) -> SpanExporter:
+        """Get the appropriate OTLP span exporter based on the spans endpoint scheme."""
+        if self.spans_endpoint:
+            match self.spans_endpoint.scheme:
+                case "grpc":
+                    return OTLPSpanExporterGrpc(
+                        endpoint=f"{self.spans_endpoint.host}:{self.spans_endpoint.port or DEFAULT_OTLP_GRPC_PORT}",
+                        insecure=self.insecure_grpc,
+                    )
+                case "http" | "https":
+                    return OTLPSpanExporterHttp(endpoint=str(self.spans_endpoint))
+        return ConsoleSpanExporter()
+
+    def get_metrics_exporter(self) -> MetricExporter:
+        """Get the appropriate OTLP metric exporter based on the metrics endpoint scheme."""
+        if self.metrics_endpoint:
+            match self.metrics_endpoint.scheme:
+                case "grpc":
+                    return OTLPMetricExporterGrpc(
+                        endpoint=f"{self.metrics_endpoint.host}:{self.metrics_endpoint.port or DEFAULT_OTLP_GRPC_PORT}",
+                        insecure=self.insecure_grpc,
+                    )
+                case "http" | "https":
+                    return OTLPMetricExporterHttp(endpoint=str(self.metrics_endpoint))
+        return ConsoleMetricExporter()
 
 
 def configure_opentelemetry(
@@ -42,29 +104,25 @@ def configure_opentelemetry(
         service_name = settings.service_name or service_name or DEFAULT_OTLP_SERVICE_NAME
         resource = Resource(attributes={SERVICE_NAME: service_name})
 
+        # Configure tracing provider
         trace_provider = TracerProvider(resource=resource)
-        processor = BatchSpanProcessor(
-            OTLPSpanExporter(endpoint=str(settings.spans_endpoint), insecure=settings.insecure)
-            if settings.spans_endpoint
-            else ConsoleSpanExporter()
-        )
+        processor = BatchSpanProcessor(settings.get_spans_exporter())
         trace_provider.add_span_processor(processor)
         trace.set_tracer_provider(trace_provider)
         logger.debug("OTLP spans via %s", settings.spans_endpoint or "console")
 
-        reader = PeriodicExportingMetricReader(
-            OTLPMetricExporter(endpoint=str(settings.metrics_endpoint), insecure=settings.insecure)
-            if settings.metrics_endpoint
-            else ConsoleMetricExporter()
-        )
+        # Configure metrics provider
+        reader = PeriodicExportingMetricReader(settings.get_metrics_exporter())
         meter_provider = MeterProvider(resource=resource, metric_readers=[reader])
         metrics.set_meter_provider(meter_provider)
         logger.debug("OTLP metrics via %s", settings.metrics_endpoint or "console")
 
+        # Configure instrumentation for various libraries
         PymongoInstrumentor().instrument()
         BotocoreInstrumentor().instrument()
         RedisInstrumentor().instrument()
         HTTPX2ClientInstrumentor().instrument()
+
         _configured = True
         logger.info("OpenTelemetry configured")
     else:

pyproject.toml

@@ -17,7 +17,7 @@ keymanager = [
     "cryptography>=50",
     "httpx2[http2]>=2.13.1",
     "jwcrypto>=1.6.1",
-    "opentelemetry-api>=1.31.0",
+    "opentelemetry-api>=1.45.1",
     "pydantic>=2.10.6",
     "pymongo>=4.11.3",
     "redis>=5.2.1",
@@ -27,8 +27,8 @@ opentelemetry = [
     "botocore>=1.36.1",
     "fastapi>=0.115.11",
     "starlette>=1.7.0",
-    "opentelemetry-api>=1.31.0",
-    "opentelemetry-exporter-otlp>=1.31.0",
+    "opentelemetry-api>=1.45.1",
+    "opentelemetry-exporter-otlp>=1.45.1",
     "opentelemetry-instrumentation-botocore>=0.52b0",
     "opentelemetry-instrumentation-fastapi>=0.52b0",
     "opentelemetry-instrumentation-pymongo>=0.52b0",

tests/test_opentelemetry.py

@@ -1,25 +1,137 @@
 import pytest
 from fastapi import FastAPI
 from opentelemetry import metrics, trace
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as OTLPMetricExporterGrpc
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as OTLPSpanExporterGrpc
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as OTLPMetricExporterHttp
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as OTLPSpanExporterHttp
+from opentelemetry.sdk.metrics.export import ConsoleMetricExporter
+from opentelemetry.sdk.trace.export import ConsoleSpanExporter
 
 from dnstapir.opentelemetry import OtlpSettings, configure_opentelemetry
 
 
 @pytest.fixture(scope="module", autouse=True)
 def shutdown_providers():
     """Flush console exporters while pytest's output capture is still open"""
+
     yield
     trace.get_tracer_provider().shutdown()
     metrics.get_meter_provider().shutdown()
 
 
-def test_telemetry():
+def test_telemetry_settings():
+    """Test basic OTLP settings."""
+
+    res = OtlpSettings()
+    assert res.spans_endpoint is None
+    assert res.metrics_endpoint is None
+    assert isinstance(res.get_spans_exporter(), ConsoleSpanExporter)
+    assert isinstance(res.get_metrics_exporter(), ConsoleMetricExporter)
+
+    res = OtlpSettings(
+        spans_endpoint="grpc://localhost:4317",
+        metrics_endpoint="http://localhost:4318/v1/metrics",
+    )
+    assert res.spans_endpoint.scheme == "grpc"
+    assert res.spans_endpoint.host == "localhost"
+    assert res.spans_endpoint.port == 4317
+    assert res.metrics_endpoint.scheme == "http"
+    assert res.metrics_endpoint.host == "localhost"
+    assert res.metrics_endpoint.port == 4318
+
+
+def test_telemetry_settings_invalid():
+    """Test that invalid OTLP settings raise ValueError."""
+
+    with pytest.raises(ValueError):
+        _ = OtlpSettings(
+            metrics_endpoint="grpc://localhost:4317/xyzzy",
+        )
+
+    with pytest.raises(ValueError):
+        _ = OtlpSettings(
+            metrics_endpoint="grpc://localhost:4317?foo=bar",
+        )
+
+
+def test_telemetry_settings_compat():
+    """Test compatibility with various OTLP settings formats."""
+
+    res = OtlpSettings(spans_endpoint="localhost:4317")
+    assert res.spans_endpoint.scheme == "grpc"
+    assert res.spans_endpoint.host == "localhost"
+    assert res.spans_endpoint.port == 4317
+
+    res = OtlpSettings(spans_endpoint="127.0.0.1:4317")
+    assert res.spans_endpoint.scheme == "grpc"
+    assert res.spans_endpoint.host == "127.0.0.1"
+    assert res.spans_endpoint.port == 4317
+
+    res = OtlpSettings()
+    assert res.insecure_grpc is None
+
+    res = OtlpSettings(insecure=True)
+    assert res.insecure_grpc is True
+
+    res = OtlpSettings(insecure=False)
+    assert res.insecure_grpc is False
+
+    res = OtlpSettings(insecure_grpc=True)
+    assert res.insecure_grpc is True
+
+    res = OtlpSettings(insecure_grpc=False)
+    assert res.insecure_grpc is False
+
+
+@pytest.mark.parametrize(
+    "spans_endpoint,metrics_endpoint,spans_exporter_class,metrics_exporter_class",
+    [
+        (
+            "grpc://localhost:4317",
+            "grpc://localhost:4317",
+            OTLPSpanExporterGrpc,
+            OTLPMetricExporterGrpc,
+        ),
+        (
+            "http://localhost:4318/v1/traces",
+            "http://localhost:4318/v1/metrics",
+            OTLPSpanExporterHttp,
+            OTLPMetricExporterHttp,
+        ),
+    ],
+)
+def test_exporters_otlp(
+    spans_endpoint: str,
+    metrics_endpoint: str,
+    spans_exporter_class: type,
+    metrics_exporter_class: type,
+):
+    """Test that the correct OTLP exporters are returned based on the settings."""
+
+    # configure_opentelemetry() only builds exporters on its first call, so test the getters directly
+    settings = OtlpSettings(spans_endpoint=spans_endpoint, metrics_endpoint=metrics_endpoint)
+
+    spans_exporter = settings.get_spans_exporter()
+    assert isinstance(spans_exporter, spans_exporter_class)
+    spans_exporter.shutdown()
+
+    metrics_exporter = settings.get_metrics_exporter()
+    assert isinstance(metrics_exporter, metrics_exporter_class)
+    metrics_exporter.shutdown()
+
+
+def test_telemetry_console():
+    """Test configuring OpenTelemetry with a FastAPI app."""
+
     app = FastAPI()
     settings = OtlpSettings()
     configure_opentelemetry(service_name="test", settings=settings, fastapi_app=app)
 
 
 def test_telemetry_twice():
+    """Test that configuring OpenTelemetry twice does not reconfigure it."""
+
     settings = OtlpSettings()
     configure_opentelemetry(service_name="test", settings=settings)
     tracer_provider = trace.get_tracer_provider()