1import logging
2import threading
3from collections.abc import Mapping
4from typing import Any
5
6from opentelemetry import _logs, metrics, trace
7from opentelemetry._logs._internal import ProxyLoggerProvider
8from opentelemetry.exporter.otlp.proto.http import Compression
9from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
10from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
11from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
12from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
13from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
14from opentelemetry.sdk.metrics import Counter, Histogram, MeterProvider, UpDownCounter
15from opentelemetry.sdk.metrics.export import (
16 AggregationTemporality,
17 PeriodicExportingMetricReader,
18)
19from opentelemetry.sdk.resources import Resource
20from opentelemetry.sdk.trace import TracerProvider, sampling
21from opentelemetry.sdk.trace.export import BatchSpanProcessor
22from opentelemetry.semconv.attributes import exception_attributes, service_attributes
23from plain.packages import PackageConfig, register_config
24from plain.runtime import settings
25from plain.utils.otel import format_exception_type
26
27
28class _ExporterLoopFilter(logging.Filter):
29 """Block records that would feed back into OTLP export under failure.
30
31 Two sources to suppress:
32
33 1. The OpenTelemetry SDK's own namespace — its `failed to export`
34 warnings would re-queue indefinitely.
35 2. Anything emitted from inside an OTel SDK exporter thread (e.g.
36 urllib3 connection errors raised by the OTLP HTTP exporter). The
37 SDK names every exporter thread with an `Otel` prefix —
38 `OtelBatch{Span,Log}RecordProcessor` for traces/logs, and
39 `OtelPeriodicExportingMetricReader` for metrics — so matching on
40 that prefix catches all three. Scoping to the thread, not the
41 urllib3 namespace, lets user-code urllib3 logs flow through.
42
43 See `opentelemetry/sdk/_shared_internal/__init__.py` (BatchProcessor)
44 and `opentelemetry/sdk/metrics/export/__init__.py`
45 (PeriodicExportingMetricReader) for the thread names.
46 """
47
48 def filter(self, record: logging.LogRecord) -> bool:
49 name = record.name
50 if name == "opentelemetry" or name.startswith("opentelemetry."):
51 return False
52 return not threading.current_thread().name.startswith("Otel")
53
54
55class _QualifiedExceptionTypeLoggingHandler(LoggingHandler):
56 """LoggingHandler that writes module-qualified `exception.type` values.
57
58 The stock handler stamps log records with the bare exception class
59 `__name__`, while the span SDK's `record_exception` writes the
60 module-qualified name — giving the same exception class two different
61 identities across signals. Match the span convention via
62 `format_exception_type` (builtins stay bare) so log- and span-reported
63 exceptions agree.
64
65 `_get_attributes` is a private SDK staticmethod — the internal
66 `test_log_exception_type` tests exercise the full emit path, so an
67 SDK release that changes its shape fails CI instead of silently
68 killing log export.
69 """
70
71 @staticmethod
72 def _get_attributes(record: logging.LogRecord) -> Mapping[str, Any]:
73 attributes = LoggingHandler._get_attributes(record)
74 if record.exc_info:
75 exc = record.exc_info[1]
76 if isinstance(exc, BaseException):
77 # Copied only on this branch — the parent annotates its
78 # return as an immutable Mapping.
79 attributes = {
80 **attributes,
81 exception_attributes.EXCEPTION_TYPE: format_exception_type(exc),
82 }
83 return attributes
84
85
86@register_config
87class Config(PackageConfig):
88 package_label = "plainconnect"
89
90 def ready(self) -> None:
91 if not settings.CONNECT_EXPORT_ENABLED or not settings.CONNECT_EXPORT_TOKEN:
92 return
93
94 # Don't capture per-batch OTLP export failures as Sentry events. The OTel
95 # SDK's exporters log "Failed to export X batch" at ERROR after retries
96 # are exhausted, which Sentry's LoggingIntegration would otherwise turn
97 # into an issue per app per incident — noise the app owner can't act on
98 # (network/edge timeouts, ingest backend hiccups). The records still
99 # flow to console/file/etc. — only the Sentry capture is suppressed.
100 # Mirrors Sentry SDK's own self-protection for `sentry_sdk.errors` and
101 # `urllib3.connectionpool` in sentry_sdk/integrations/logging.py.
102 try:
103 import sentry_sdk.integrations.logging as _sentry_log # ty: ignore[unresolved-import]
104 except ImportError:
105 pass
106 else:
107 for name in (
108 "opentelemetry.exporter.otlp.proto.http.trace_exporter",
109 "opentelemetry.exporter.otlp.proto.http.metric_exporter",
110 "opentelemetry.exporter.otlp.proto.http._log_exporter",
111 "opentelemetry.sdk._shared_internal",
112 "opentelemetry.sdk._logs._internal.export",
113 "opentelemetry.sdk.metrics._internal.export",
114 ):
115 _sentry_log.ignore_logger(name)
116
117 resource = Resource.create(
118 {
119 service_attributes.SERVICE_NAME: settings.NAME,
120 service_attributes.SERVICE_VERSION: settings.VERSION,
121 }
122 )
123
124 export_url = str(settings.CONNECT_EXPORT_URL).rstrip("/")
125 headers = {"Authorization": f"Bearer {settings.CONNECT_EXPORT_TOKEN}"}
126
127 # Traces
128 current_provider = trace.get_tracer_provider()
129 if current_provider and not isinstance(
130 current_provider, trace.ProxyTracerProvider
131 ):
132 raise RuntimeError(
133 "A tracer provider already exists."
134 " plain.connect must install its own TracerProvider before any other package does."
135 )
136
137 span_exporter = OTLPSpanExporter(
138 endpoint=f"{export_url}/v1/traces",
139 headers=headers,
140 timeout=30,
141 compression=Compression.Gzip,
142 )
143 sampler = sampling.TraceIdRatioBased(settings.CONNECT_TRACE_SAMPLE_RATE)
144 tracer_provider = TracerProvider(sampler=sampler, resource=resource)
145 tracer_provider.add_span_processor(BatchSpanProcessor(span_exporter))
146 trace.set_tracer_provider(tracer_provider)
147
148 # Metrics — use delta temporality so each export contains only the
149 # increment since the last export, not a running total. This makes
150 # server-side aggregation (sum/avg in ClickHouse) straightforward.
151 metric_exporter = OTLPMetricExporter(
152 endpoint=f"{export_url}/v1/metrics",
153 headers=headers,
154 timeout=30,
155 compression=Compression.Gzip,
156 preferred_temporality={
157 Counter: AggregationTemporality.DELTA,
158 Histogram: AggregationTemporality.DELTA,
159 UpDownCounter: AggregationTemporality.DELTA,
160 },
161 )
162 reader = PeriodicExportingMetricReader(metric_exporter)
163 meter_provider = MeterProvider(metric_readers=[reader], resource=resource)
164 metrics.set_meter_provider(meter_provider)
165
166 # Logs
167 if settings.CONNECT_EXPORT_LOGS:
168 current_logger_provider = _logs.get_logger_provider()
169 if current_logger_provider and not isinstance(
170 current_logger_provider, ProxyLoggerProvider
171 ):
172 raise RuntimeError(
173 "A logger provider already exists."
174 " plain.connect must install its own LoggerProvider before any other package does."
175 )
176
177 # Accept either a level name ("INFO") or an int (20).
178 raw_level = settings.CONNECT_LOG_LEVEL
179 if isinstance(raw_level, str):
180 level_names = logging.getLevelNamesMapping()
181 if raw_level.upper() not in level_names:
182 raise ValueError(
183 f"CONNECT_LOG_LEVEL={raw_level!r} is not a valid logging level."
184 )
185 log_level = level_names[raw_level.upper()]
186 else:
187 log_level = int(raw_level)
188
189 log_exporter = OTLPLogExporter(
190 endpoint=f"{export_url}/v1/logs",
191 headers=headers,
192 timeout=30,
193 compression=Compression.Gzip,
194 )
195 logger_provider = LoggerProvider(resource=resource)
196 logger_provider.add_log_record_processor(
197 BatchLogRecordProcessor(log_exporter)
198 )
199 _logs.set_logger_provider(logger_provider)
200
201 handler = _QualifiedExceptionTypeLoggingHandler(
202 level=log_level, logger_provider=logger_provider
203 )
204 # Filter on the handler (not the loggers) so OTLP exporter and
205 # HTTP-client diagnostics still reach the app's console/file
206 # handlers — we only stop them from being re-exported, which
207 # would loop under failure.
208 handler.addFilter(_ExporterLoopFilter())
209
210 # Plain's `configure_logging` sets `plain` and `app` to
211 # propagate=False, so attaching only to root would miss them.
212 # Attach to root for everything else (user `getLogger(__name__)`,
213 # third-party libs).
214 for name in ("", "plain", "app"):
215 logging.getLogger(name).addHandler(handler)
216
217 # Root defaults to WARNING. A library that uses
218 # `logging.getLogger(__name__)` without setting its own level
219 # inherits root's effective level — so INFO/DEBUG records get
220 # dropped before the OTLP handler runs. Widen root just enough
221 # to let CONNECT_LOG_LEVEL through; never narrow it.
222 # NOTSET (0) on root already means "all messages processed",
223 # so leave it alone in that case.
224 root = logging.getLogger()
225 if root.level != logging.NOTSET and root.level > log_level:
226 root.setLevel(log_level)