Coverage for src/lumen/observability/otel.py: 45%

1"""OpenTelemetry SDK setup and instrumentation (Phase 13b, Stages 2-4). 

2 

3``setup_telemetry`` wires the trace, metric and (optionally) log providers plus 

4the asyncpg instrumentation once per process. ``instrument_fastapi`` attaches the 

5ASGI request instrumentation to the app, redacting query strings so no search 

6parameter *values* or patient identifiers leak into span attributes. 

7 

8Both functions are no-ops when ``settings.enabled`` is false. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TYPE_CHECKING, Any, cast 

14 

15import structlog 

16 

17if TYPE_CHECKING: 

18 from fastapi import FastAPI 

19 from opentelemetry.sdk.resources import Resource 

20 from opentelemetry.trace import Span 

21 

22 from lumen.config import ObservabilitySettings 

23 

24logger = structlog.get_logger(__name__) 

25 

26# Guards against configuring the global providers more than once per process. 

27_configured = False 

28_fastapi_instrumented = False 

29 

30 

31def setup_telemetry(settings: ObservabilitySettings) -> None: 

32 """Configure the global trace/metric/log providers and DB instrumentation. 

33 

34 Idempotent and safe: returns immediately when telemetry is disabled or has 

35 already been configured. Reusable from any entry point (API server or a 

36 future Procrastinate worker) so background jobs are not traced blind. 

37 """ 

38 global _configured 

39 if not settings.enabled or _configured: 

40 return 

41 

42 resource = _build_resource(settings) 

43 _setup_tracing(settings, resource) 

44 _setup_metrics(settings, resource) 

45 if settings.log_export: 

46 _setup_log_bridge(settings, resource) 

47 _instrument_asyncpg() 

48 

49 _configured = True 

50 logger.info( 

51 "otel.configured", 

52 endpoint=settings.exporter_endpoint, 

53 service=settings.service_name, 

54 log_export=settings.log_export, 

55 ) 

56 

57 

58def instrument_fastapi(app: FastAPI, settings: ObservabilitySettings) -> None: 

59 """Attach FastAPI/ASGI request instrumentation with URL redaction.""" 

60 global _fastapi_instrumented 

61 if not settings.enabled or _fastapi_instrumented: 

62 return 

63 

64 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor 

65 

66 FastAPIInstrumentor.instrument_app(app, server_request_hook=_redact_url_hook) 

67 _fastapi_instrumented = True 

68 

69 

70def _build_resource(settings: ObservabilitySettings) -> Resource: 

71 from opentelemetry.sdk.resources import SERVICE_NAME, Resource 

72 

73 return Resource.create({SERVICE_NAME: settings.service_name}) 

74 

75 

76def _setup_tracing(settings: ObservabilitySettings, resource: Resource) -> None: 

77 from opentelemetry import trace 

78 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter 

79 from opentelemetry.sdk.trace import TracerProvider 

80 from opentelemetry.sdk.trace.export import BatchSpanProcessor 

81 from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased 

82 

83 provider = TracerProvider( 

84 resource=resource, 

85 sampler=ParentBased(TraceIdRatioBased(settings.traces_sample_rate)), 

86 ) 

87 exporter = OTLPSpanExporter(endpoint=f"{settings.exporter_endpoint}/v1/traces") 

88 provider.add_span_processor(BatchSpanProcessor(exporter)) 

89 trace.set_tracer_provider(provider) 

90 

91 

92def _setup_metrics(settings: ObservabilitySettings, resource: Resource) -> None: 

93 from opentelemetry import metrics 

94 from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter 

95 from opentelemetry.instrumentation.system_metrics import SystemMetricsInstrumentor 

96 from opentelemetry.sdk.metrics import MeterProvider 

97 from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader 

98 

99 exporter = OTLPMetricExporter(endpoint=f"{settings.exporter_endpoint}/v1/metrics") 

100 reader = PeriodicExportingMetricReader(exporter) 

101 provider = MeterProvider(resource=resource, metric_readers=[reader]) 

102 metrics.set_meter_provider(provider) 

103 

104 # Process-level metrics: CPU, RSS memory, GC, open file descriptors, uptime. 

105 SystemMetricsInstrumentor().instrument() 

106 

107 

108def _setup_log_bridge(settings: ObservabilitySettings, resource: Resource) -> None: 

109 import logging 

110 

111 from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter 

112 from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler 

113 from opentelemetry.sdk._logs.export import BatchLogRecordProcessor 

114 

115 provider = LoggerProvider(resource=resource) 

116 exporter = OTLPLogExporter(endpoint=f"{settings.exporter_endpoint}/v1/logs") 

117 provider.add_log_record_processor(BatchLogRecordProcessor(exporter)) 

118 

119 # structlog renders to the stdlib root logger; bridging that root logger sends 

120 # the same records to OTLP while keeping stdout JSON (Weg A) untouched. The 

121 # handler attaches the active trace/span context automatically. 

122 handler = LoggingHandler(level=logging.NOTSET, logger_provider=provider) 

123 logging.getLogger().addHandler(handler) 

124 

125 

126def _instrument_asyncpg() -> None: 

127 from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor 

128 

129 AsyncPGInstrumentor().instrument() 

130 

131 

132def _redact_url_hook(span: Span, scope: dict[str, Any]) -> None: 

133 """Strip query strings from request spans (PII rule for span attributes). 

134 

135 Search parameter *values* and patient identifiers must never reach a trace, 

136 so only the path is kept and the query component is blanked on every URL 

137 attribute the FastAPI instrumentation may populate (old and new semconv). 

138 """ 

139 if not span.is_recording(): 

140 return 

141 

142 path: str = scope.get("path", "") or "" 

143 clean_url = _clean_url(scope, path) 

144 # Overwrite (last-set wins) so no query string survives on any variant. 

145 span.set_attribute("http.target", path) 

146 span.set_attribute("http.url", clean_url) 

147 span.set_attribute("url.path", path) 

148 span.set_attribute("url.query", "") 

149 span.set_attribute("url.full", clean_url) 

150 

151 

152def _clean_url(scope: dict[str, Any], path: str) -> str: 

153 """Reconstruct scheme://host[:port]/path from the ASGI scope, without query.""" 

154 scheme = cast(str, scope.get("scheme", "http")) 

155 server = cast("tuple[str, int | None] | None", scope.get("server")) 

156 if server is not None: 

157 host, port = server[0], server[1] 

158 netloc = f"{host}:{port}" if port else host 

159 return f"{scheme}://{netloc}{path}" 

160 return path