1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import atexit
17import enum
18import inspect
19import json
20import logging
21import multiprocessing
22import multiprocessing.connection
23import multiprocessing.managers
24import multiprocessing.process
25import multiprocessing.util
26import os
27import threading
28from collections.abc import Callable, Iterable
29from typing import Any, ClassVar, Literal
30
31import opentelemetry.metrics as api_metrics
32import opentelemetry.trace as api_trace
33import psutil
34
35from .. import utils
36
37_MANAGER_START_TIMEOUT_SECONDS = 60.0
38_MANAGER_CLEANUP_TIMEOUT_SECONDS = 1.0
39
40# MSC telemetry prefers publishing raw samples when possible to support arbitrary post-hoc aggregations.
41#
42# Some setups, however, may need resampling to reduce sample volume. The resampling methods we use
43# sacrifice temporal resolution to preserve other information. Which method is used depends on if
44# the expected post-hoc aggregate function is decomposable:
45#
46# * Decomposable aggregate functions (e.g. count, sum, min, max).
47# * Use client-side aggregation.
48# * E.g. preserve request + response counts.
49# * Non-decomposable aggregate functions (e.g. average, percentile).
50# * Use decimation by an integer factor or last value.
51# * E.g. preserve the shape of the latency distribution (unlike tail sampling).
52
53_METRICS_EXPORTER_MAPPING = {
54 "console": "opentelemetry.sdk.metrics.export.ConsoleMetricExporter",
55 "otlp": "opentelemetry.exporter.otlp.proto.http.metric_exporter.OTLPMetricExporter",
56 # "Private" until it's decided whether this will be official.
57 "_otlp_msal": "multistorageclient.telemetry.metrics.exporters.otlp_msal._OTLPMSALMetricExporter",
58 "_otlp_mtls_vault": "multistorageclient.telemetry.metrics.exporters.otlp_mtls_vault._OTLPmTLSVaultMetricExporter",
59}
60
61_TRACE_EXPORTER_MAPPING = {
62 "console": "opentelemetry.sdk.trace.export.ConsoleSpanExporter",
63 "otlp": "opentelemetry.exporter.otlp.proto.http.trace_exporter.OTLPSpanExporter",
64 # "Private" until it's decided whether this will be official.
65 "_otlp_msal": "multistorageclient.telemetry.traces.exporters.otlp_msal._OTLPMSALSpanExporter",
66}
67
68logger = logging.getLogger(__name__)
69
70
[docs]
71class Telemetry:
72 """
73 Provides telemetry resources.
74
75 Instances shouldn't be copied between processes. Not fork-safe or pickleable.
76
77 Instances can be shared between processes by registering with a :py:class:`multiprocessing.managers.BaseManager` and using proxy objects.
78 """
79
80 # Metrics are named `multistorageclient.{property}(.{aggregation})?`.
81 #
82 # For example:
83 #
84 # - multistorageclient.data_size
85 # - Gauge for data size per individual operation.
86 # - For distributions (e.g. post-hoc histograms + heatmaps).
87 # - multistorageclient.data_size.sum
88 # - Counter (sum) for data size across all operations.
89 # - For aggregates (e.g. post-hoc data rate calculations).
90
91 # https://opentelemetry.io/docs/specs/semconv/general/naming#metrics
92 class GaugeName(enum.Enum):
93 LATENCY = "multistorageclient.latency"
94 DATA_SIZE = "multistorageclient.data_size"
95 DATA_RATE = "multistorageclient.data_rate"
96 FILE_DESCRIPTOR_DURATION = "multistorageclient.file_descriptor.duration"
97
98 # https://opentelemetry.io/docs/specs/semconv/general/metrics#units
99 _GAUGE_UNIT_MAPPING: ClassVar[dict[GaugeName, str]] = {
100 # Seconds.
101 GaugeName.LATENCY: "s",
102 # Bytes.
103 GaugeName.DATA_SIZE: "By",
104 # Bytes/second.
105 GaugeName.DATA_RATE: "By/s",
106 # Seconds.
107 GaugeName.FILE_DESCRIPTOR_DURATION: "s",
108 }
109
110 # https://opentelemetry.io/docs/specs/semconv/general/naming#metrics
111 class CounterName(enum.Enum):
112 REQUEST_SUM = "multistorageclient.request.sum"
113 RESPONSE_SUM = "multistorageclient.response.sum"
114 DATA_SIZE_SUM = "multistorageclient.data_size.sum"
115
116 # https://opentelemetry.io/docs/specs/semconv/general/metrics#units
117 _COUNTER_UNIT_MAPPING: ClassVar[dict[CounterName, str]] = {
118 # Unitless.
119 CounterName.REQUEST_SUM: "{request}",
120 # Unitless.
121 CounterName.RESPONSE_SUM: "{response}",
122 # Bytes.
123 CounterName.DATA_SIZE_SUM: "By",
124 }
125
126 # https://opentelemetry.io/docs/specs/semconv/general/naming#metrics
127 class UpDownCounterName(enum.Enum):
128 FILE_DESCRIPTOR_OPEN = "multistorageclient.file_descriptor.open"
129
130 # https://opentelemetry.io/docs/specs/semconv/general/metrics#units
131 _UP_DOWN_COUNTER_UNIT_MAPPING: ClassVar[dict[UpDownCounterName, str]] = {
132 UpDownCounterName.FILE_DESCRIPTOR_OPEN: "{file_descriptor}",
133 }
134
135 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to meter provider.
136 _meter_provider_cache: dict[str, api_metrics.MeterProvider]
137 _meter_provider_cache_lock: threading.Lock
138 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to meter.
139 _meter_cache: dict[str, api_metrics.Meter]
140 _meter_cache_lock: threading.Lock
141 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to gauge name to gauge.
142 _gauge_cache: dict[str, dict[GaugeName, api_metrics._Gauge]]
143 _gauge_cache_lock: threading.Lock
144 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to counter name to counter.
145 _counter_cache: dict[str, dict[CounterName, api_metrics.Counter]]
146 _counter_cache_lock: threading.Lock
147 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to up-down counter name to counter.
148 _up_down_counter_cache: dict[str, dict[UpDownCounterName, api_metrics.UpDownCounter]]
149 _up_down_counter_cache_lock: threading.Lock
150 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to tracer provider.
151 _tracer_provider_cache: dict[str, api_trace.TracerProvider]
152 _tracer_provider_cache_lock: threading.Lock
153 # Map of config as a sorted JSON string (since dictionaries can't be hashed) to tracer.
154 _tracer_cache: dict[str, api_trace.Tracer]
155 _tracer_cache_lock: threading.Lock
156
157 def __init__(self):
158 self._meter_provider_cache = {}
159 self._meter_provider_cache_lock = threading.Lock()
160 self._meter_cache = {}
161 self._meter_cache_lock = threading.Lock()
162 self._gauge_cache = {}
163 self._gauge_cache_lock = threading.Lock()
164 self._counter_cache = {}
165 self._counter_cache_lock = threading.Lock()
166 self._up_down_counter_cache = {}
167 self._up_down_counter_cache_lock = threading.Lock()
168 self._tracer_provider_cache = {}
169 self._tracer_provider_cache_lock = threading.Lock()
170 self._tracer_cache = {}
171 self._tracer_cache_lock = threading.Lock()
172
173 def _reinitialize_instance_locks_after_fork(self) -> None:
174 """
175 Reinitialize internal locks after fork to prevent deadlocks.
176
177 This method reinitializes all internal locks. Caches are kept as they
178 may still be valid, but locks must be fresh to avoid deadlocks.
179 """
180 self._meter_provider_cache_lock = threading.Lock()
181 self._meter_cache_lock = threading.Lock()
182 self._gauge_cache_lock = threading.Lock()
183 self._counter_cache_lock = threading.Lock()
184 self._up_down_counter_cache_lock = threading.Lock()
185 self._tracer_provider_cache_lock = threading.Lock()
186 self._tracer_cache_lock = threading.Lock()
187
188 def meter_provider(self, config: dict[str, Any]) -> api_metrics.MeterProvider | None:
189 """
190 Create or return an existing :py:class:`api_metrics.MeterProvider` for a config.
191
192 :param config: ``.opentelemetry.metrics`` config dict.
193 :return: A :py:class:`api_metrics.MeterProvider` or ``None`` if no valid exporter is configured.
194 """
195 config_json = json.dumps(config, sort_keys=True)
196 with self._meter_provider_cache_lock:
197 if config_json in self._meter_provider_cache:
198 return self._meter_provider_cache[config_json]
199 else:
200 if "exporter" in config:
201 try:
202 import opentelemetry.sdk.metrics as sdk_metrics
203 import opentelemetry.sdk.metrics.export as sdk_metrics_export
204
205 from .metrics.readers.diperiodic_exporting import DiperiodicExportingMetricReader
206
207 exporter_type: str = config["exporter"]["type"]
208 exporter_fully_qualified_name = _METRICS_EXPORTER_MAPPING.get(exporter_type, exporter_type)
209 exporter_module_name, exporter_class_name = exporter_fully_qualified_name.rsplit(".", 1)
210 cls = utils.import_class(exporter_class_name, exporter_module_name)
211 exporter_options = config["exporter"].get("options", {}).copy()
212 exporter: sdk_metrics_export.MetricExporter = cls(**exporter_options)
213
214 reader_options = config.get("reader", {}).get("options", {})
215 reader: sdk_metrics_export.MetricReader = DiperiodicExportingMetricReader(
216 **reader_options, exporter=exporter
217 )
218
219 return self._meter_provider_cache.setdefault(
220 config_json, sdk_metrics.MeterProvider(metric_readers=[reader])
221 )
222 except (AttributeError, ImportError):
223 logger.exception("Failed to import OpenTelemetry Python SDK or exporter! Disabling metrics.")
224 return None
225 else:
226 # Don't return a no-op meter provider to avoid unnecessary overhead.
227 logger.error("No exporter configured! Disabling metrics.")
228 return None
229
230 def meter(self, config: dict[str, Any]) -> api_metrics.Meter | None:
231 """
232 Create or return an existing :py:class:`api_metrics.Meter` for a config.
233
234 :param config: ``.opentelemetry.metrics`` config dict.
235 :return: A :py:class:`api_metrics.Meter` or ``None`` if no valid exporter is configured.
236 """
237 config_json = json.dumps(config, sort_keys=True)
238 with self._meter_cache_lock:
239 if config_json in self._meter_cache:
240 return self._meter_cache[config_json]
241 else:
242 meter_provider = self.meter_provider(config=config)
243 if meter_provider is None:
244 return None
245 else:
246 return self._meter_cache.setdefault(
247 config_json, meter_provider.get_meter(name="multistorageclient")
248 )
249
250 def gauge(self, config: dict[str, Any], name: GaugeName) -> api_metrics._Gauge | None:
251 """
252 Create or return an existing :py:class:`api_metrics.Gauge` for a config and gauge name.
253
254 :param config: ``.opentelemetry.metrics`` config dict.
255 :return: A :py:class:`api_metrics.Gauge` or ``None`` if no valid exporter is configured.
256 """
257 config_json = json.dumps(config, sort_keys=True)
258 with self._gauge_cache_lock:
259 if config_json in self._gauge_cache and name in self._gauge_cache[config_json]:
260 return self._gauge_cache[config_json][name]
261 else:
262 meter = self.meter(config=config)
263 if meter is None:
264 return None
265 else:
266 return self._gauge_cache.setdefault(config_json, {}).setdefault(
267 name,
268 meter.create_gauge(name=name.value, unit=Telemetry._GAUGE_UNIT_MAPPING.get(name, "")),
269 )
270
271 def counter(self, config: dict[str, Any], name: CounterName) -> api_metrics.Counter | None:
272 """
273 Create or return an existing :py:class:`api_metrics.Counter` for a config and counter name.
274
275 :param config: ``.opentelemetry.metrics`` config dict.
276 :return: A :py:class:`api_metrics.Counter` or ``None`` if no valid exporter is configured.
277 """
278 config_json = json.dumps(config, sort_keys=True)
279 with self._counter_cache_lock:
280 if config_json in self._counter_cache and name in self._counter_cache[config_json]:
281 return self._counter_cache[config_json][name]
282 else:
283 meter = self.meter(config=config)
284 if meter is None:
285 return None
286 else:
287 return self._counter_cache.setdefault(config_json, {}).setdefault(
288 name,
289 meter.create_counter(name=name.value, unit=Telemetry._COUNTER_UNIT_MAPPING.get(name, "")),
290 )
291
292 def up_down_counter(self, config: dict[str, Any], name: UpDownCounterName) -> api_metrics.UpDownCounter | None:
293 """
294 Create or return an existing :py:class:`api_metrics.UpDownCounter` for a config and counter name.
295
296 :param config: ``.opentelemetry.metrics`` config dict.
297 :param name: Up-down counter name.
298 :return: A :py:class:`api_metrics.UpDownCounter` or ``None`` if no valid exporter is configured.
299 """
300 config_json = json.dumps(config, sort_keys=True)
301 with self._up_down_counter_cache_lock:
302 if config_json in self._up_down_counter_cache and name in self._up_down_counter_cache[config_json]:
303 return self._up_down_counter_cache[config_json][name]
304
305 meter = self.meter(config=config)
306 if meter is None:
307 return None
308
309 return self._up_down_counter_cache.setdefault(config_json, {}).setdefault(
310 name,
311 meter.create_up_down_counter(
312 name=name.value,
313 unit=Telemetry._UP_DOWN_COUNTER_UNIT_MAPPING.get(name, ""),
314 ),
315 )
316
317 def tracer_provider(self, config: dict[str, Any]) -> api_trace.TracerProvider | None:
318 """
319 Create or return an existing :py:class:`api_trace.TracerProvider` for a config.
320
321 :param config: ``.opentelemetry.traces`` config dict.
322 :return: A :py:class:`api_trace.TracerProvider` or ``None`` if no valid exporter is configured.
323 """
324 config_json = json.dumps(config, sort_keys=True)
325 with self._tracer_provider_cache_lock:
326 if config_json in self._tracer_provider_cache:
327 return self._tracer_provider_cache[config_json]
328 else:
329 if "exporter" in config:
330 try:
331 import opentelemetry.sdk.trace as sdk_trace
332 import opentelemetry.sdk.trace.export as sdk_trace_export
333 import opentelemetry.sdk.trace.sampling as sdk_trace_sampling
334
335 exporter_type: str = config["exporter"]["type"]
336 exporter_fully_qualified_name = _TRACE_EXPORTER_MAPPING.get(exporter_type, exporter_type)
337 exporter_module_name, exporter_class_name = exporter_fully_qualified_name.rsplit(".", 1)
338 cls = utils.import_class(exporter_class_name, exporter_module_name)
339 exporter_options = config["exporter"].get("options", {})
340 exporter: sdk_trace_export.SpanExporter = cls(**exporter_options)
341
342 processor: sdk_trace.SpanProcessor = sdk_trace.SynchronousMultiSpanProcessor()
343 processor.add_span_processor(sdk_trace_export.BatchSpanProcessor(span_exporter=exporter))
344
345 # TODO: Add sampler to configuration schema.
346 sampler: sdk_trace_sampling.Sampler = sdk_trace_sampling.ALWAYS_ON
347
348 return self._tracer_provider_cache.setdefault(
349 config_json,
350 sdk_trace.TracerProvider(active_span_processor=processor, sampler=sampler),
351 )
352 except (AttributeError, ImportError):
353 logger.exception("Failed to import OpenTelemetry Python SDK or exporter! Disabling traces.")
354 return None
355 else:
356 logger.error("No exporter configured! Disabling traces.")
357 return None
358
359 def tracer(self, config: dict[str, Any]) -> api_trace.Tracer | None:
360 """
361 Create or return an existing :py:class:`api_trace.Tracer` for a config.
362
363 :param config: ``.opentelemetry.traces`` config dict.
364 :return: A :py:class:`api_trace.Tracer` or ``None`` if no valid exporter is configured.
365 """
366 config_json = json.dumps(config, sort_keys=True)
367 with self._tracer_cache_lock:
368 if config_json in self._tracer_cache:
369 return self._tracer_cache[config_json]
370 else:
371 tracer_provider = self.tracer_provider(config=config)
372 if tracer_provider is None:
373 return None
374 else:
375 return self._tracer_cache.setdefault(
376 config_json, tracer_provider.get_tracer(instrumenting_module_name="multistorageclient")
377 )
378
379
380# To share a single :py:class:`Telemetry` within a process (e.g. local, manager).
381#
382# A manager's server processes shouldn't be forked, so this should be safe.
383_TELEMETRY: Telemetry | None = None
384_TELEMETRY_LOCK = threading.Lock()
385
386
387def _init() -> Telemetry:
388 """
389 Create or return an existing :py:class:`Telemetry`.
390
391 :return: A telemetry instance.
392 """
393 global _TELEMETRY
394 global _TELEMETRY_LOCK # noqa: PLW0602
395
396 with _TELEMETRY_LOCK:
397 if _TELEMETRY is None:
398 _TELEMETRY = Telemetry()
399 return _TELEMETRY
400
401
[docs]
402class TelemetryManager(multiprocessing.managers.BaseManager):
403 """
404 A :py:class:`multiprocessing.managers.BaseManager` for telemetry resources.
405
406 The OpenTelemetry Python SDK isn't fork-safe since telemetry sample buffers can be duplicated.
407
408 In addition, Python ≤3.12 doesn't call exit handlers for forked processes.
409 This causes the OpenTelemetry Python SDK to not flush telemetry before exiting.
410
411 * https://github.com/open-telemetry/opentelemetry-python/issues/4215
412 * https://github.com/open-telemetry/opentelemetry-python/issues/3307
413
414 Forking is multiprocessing's default start method for non-macOS POSIX systems until Python 3.14.
415
416 * https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods
417
418 To fully support multiprocessing, resampling + publishing is handled by
419 a single process that's (ideally) a child of (i.e. directly under) the main process. This:
420
421 * Relieves other processes of this work.
422
423 * Avoids issues with duplicate samples when forking and unpublished samples when exiting forks.
424
425 * Allows cross-process resampling.
426 * Reuses a single connection pool to telemetry backends.
427
428 The downside is it essentially re-introduces global interpreter lock (GIL) with
429 additional IPC overhead. Telemetry operations, however, should be lightweight so
430 this isn't expected to be a problem. Remote data store latency should still be
431 the primary throughput limiter for storage clients.
432
433 :py:class:`multiprocessing.managers.BaseManager` is used for this since it creates
434 a separate server process for shared objects.
435
436 Telemetry resources are provided as
437 `proxy objects <https://docs.python.org/3/library/multiprocessing.html#proxy-objects>`_
438 for location transparency.
439
440 The documentation isn't particularly detailed, but others have written comprehensively on this:
441
442 * https://zpz.github.io/blog/python-mp-manager-1
443 * https://zpz.github.io/blog/python-mp-manager-2
444 * https://zpz.github.io/blog/python-mp-manager-3
445
446 By specification, metric and tracer providers must call shutdown on any
447 underlying metric readers + span processors + exporters.
448
449 * https://opentelemetry.io/docs/specs/otel/metrics/sdk#shutdown
450 * https://opentelemetry.io/docs/specs/otel/trace/sdk#shutdown
451
452 In the OpenTelemetry Python SDK, provider shutdown is called automatically
453 by exit handlers (when they work at least). Consequently, clients should:
454
455 * Only receive proxy objects.
456
457 * Enables metric reader + span processor + exporter re-use across processes.
458
459 * Never call shutdown on the proxy objects.
460
461 * The shutdown exit handler is registered on the manager's server process.
462 * ⚠️ We expect a finite number of providers (i.e. no dynamic configs) so we don't leak them.
463 """
464
465 _process: multiprocessing.process.BaseProcess
466
467 # BaseManager's private startup protocol is absent from its type stubs.
468 def start(self: Any, initializer: Callable[..., Any] | None = None, initargs: Iterable[Any] = ()) -> None:
469 """Start the telemetry child with a bounded wait for its listening address."""
470 if self._state.value != self._state.INITIAL:
471 raise multiprocessing.ProcessError("Telemetry manager has already been started")
472 if initializer is not None and not callable(initializer):
473 raise TypeError("initializer must be a callable")
474
475 # BaseManager.start() has no timeout for its readiness pipe. Keep its
476 # process and finalizer protocol, but bound the receive and clean up failures.
477 reader, writer = multiprocessing.connection.Pipe(duplex=False)
478 self._process = self._ctx.Process(
479 target=type(self)._run_server,
480 args=(self._registry, self._address, self._authkey, self._serializer, writer, initializer, initargs),
481 )
482 self._process.name = f"{type(self).__name__}-" + ":".join(str(i) for i in self._process._identity)
483 try:
484 self._process.start()
485 writer.close()
486 if not reader.poll(_MANAGER_START_TIMEOUT_SECONDS):
487 raise TimeoutError(f"Telemetry manager did not start within {_MANAGER_START_TIMEOUT_SECONDS:g} seconds")
488 self._address = reader.recv()
489 except BaseException:
490 if self._process.pid is not None:
491 if self._process.is_alive():
492 self._process.terminate()
493 self._process.join(_MANAGER_CLEANUP_TIMEOUT_SECONDS)
494 if self._process.is_alive():
495 self._process.kill()
496 self._process.join(_MANAGER_CLEANUP_TIMEOUT_SECONDS)
497 raise
498 finally:
499 reader.close()
500 writer.close()
501
502 self._state.value = self._state.STARTED
503 finalizer_args = (
504 self._process,
505 self._address,
506 self._authkey,
507 self._state,
508 self._Client,
509 self._shutdown_timeout,
510 )
511 self.shutdown = multiprocessing.util.Finalize(
512 self, type(self)._finalize_manager, args=finalizer_args, exitpriority=0
513 )
514
515
516def _fully_qualified_name(c: type[Any]) -> str:
517 """
518 Return the fully qualified name for a class (e.g. ``module.Class``).
519
520 For :py:class:`multiprocessing.Manager` type IDs.
521 """
522 return f"{c.__module__}.{c.__qualname__}"
523
524
525# Metrics proxy object setup.
526#
527# ``exposed`` is derived from the instrument class's public functions rather than relying on
528# ``multiprocessing``'s default discovery (``public_methods`` via ``dir()``): an OpenTelemetry build
529# that delegates an instrument's method isn't surfaced by ``dir()``, so the ``AutoProxy`` would drop
530# it and raise ``AttributeError: 'AutoProxy[...]' object has no attribute 'add'`` on record.
531#
532# Unlike ``Span`` below, the metric instrument classes define their own ``__init__``; it must be
533# excluded (public-only), otherwise it would override ``BaseProxy.__init__`` and break proxy setup.
534TelemetryManager.register(
535 typeid=_fully_qualified_name(api_metrics._Gauge),
536 exposed=[
537 name
538 for name, _ in inspect.getmembers(api_metrics._Gauge, predicate=inspect.isfunction)
539 if not name.startswith("_")
540 ],
541)
542TelemetryManager.register(
543 typeid=_fully_qualified_name(api_metrics.Counter),
544 exposed=[
545 name
546 for name, _ in inspect.getmembers(api_metrics.Counter, predicate=inspect.isfunction)
547 if not name.startswith("_")
548 ],
549)
550TelemetryManager.register(
551 typeid=_fully_qualified_name(api_metrics.UpDownCounter),
552 exposed=[
553 name
554 for name, _ in inspect.getmembers(api_metrics.UpDownCounter, predicate=inspect.isfunction)
555 if not name.startswith("_")
556 ],
557)
558TelemetryManager.register(
559 typeid=_fully_qualified_name(api_metrics.Meter),
560 method_to_typeid={
561 api_metrics.Meter.create_gauge.__name__: _fully_qualified_name(api_metrics._Gauge),
562 api_metrics.Meter.create_counter.__name__: _fully_qualified_name(api_metrics.Counter),
563 api_metrics.Meter.create_up_down_counter.__name__: _fully_qualified_name(api_metrics.UpDownCounter),
564 },
565)
566TelemetryManager.register(
567 typeid=_fully_qualified_name(api_metrics.MeterProvider),
568 method_to_typeid={api_metrics.MeterProvider.get_meter.__name__: _fully_qualified_name(api_metrics.Meter)},
569)
570
571# Traces proxy object setup.
572TelemetryManager.register(
573 typeid=_fully_qualified_name(api_trace.Span),
574 # Non-public methods (i.e. ones starting with a ``_``) are omitted by default.
575 #
576 # We need ``__enter__`` and ``__exit__`` so the ``Span`` can be used as a ``ContextManager``.
577 exposed=[name for name, _ in inspect.getmembers(api_trace.Span, predicate=inspect.isfunction)],
578 method_to_typeid={api_trace.Span.__enter__.__name__: _fully_qualified_name(api_trace.Span)},
579)
580TelemetryManager.register(
581 typeid=_fully_qualified_name(api_trace.Tracer),
582 # Can't proxy ``Tracer.start_as_current_span`` since it returns a generator (not pickleable)
583 # and tries to use the process-local global context (in this case, the manager's server process).
584 #
585 # Instead, spans should be constructed by:
586 #
587 # 1. Calling ``opentelemetry.context.get_current()`` to get the process-local global context (pickleable).
588 # 2. Creating a new span with the process-local global context.
589 # 3. Calling ``opentelemetry.trace.use_span()`` with the span and ``end_on_exit=True``.
590 method_to_typeid={api_trace.Tracer.start_span.__name__: _fully_qualified_name(api_trace.Span)},
591)
592TelemetryManager.register(
593 typeid=_fully_qualified_name(api_trace.TracerProvider),
594 method_to_typeid={api_trace.TracerProvider.get_tracer.__name__: _fully_qualified_name(api_trace.Tracer)},
595)
596
597# Telemetry proxy object setup.
598#
599# This should be the only registered type with a ``callable``.
600# It's the only type we create directly with a ``TelemetryManager``.
601TelemetryManager.register(
602 typeid=Telemetry.__name__,
603 callable=_init,
604 method_to_typeid={
605 Telemetry.meter_provider.__name__: _fully_qualified_name(api_metrics.MeterProvider),
606 Telemetry.meter.__name__: _fully_qualified_name(api_metrics.Meter),
607 Telemetry.gauge.__name__: _fully_qualified_name(api_metrics._Gauge),
608 Telemetry.counter.__name__: _fully_qualified_name(api_metrics.Counter),
609 Telemetry.up_down_counter.__name__: _fully_qualified_name(api_metrics.UpDownCounter),
610 Telemetry.tracer_provider.__name__: _fully_qualified_name(api_trace.TracerProvider),
611 Telemetry.tracer.__name__: _fully_qualified_name(api_trace.Tracer),
612 },
613)
614
615
616# Map of init options as a sorted JSON string (since dictionaries can't be hashed) to telemetry proxy.
617_TELEMETRY_PROXIES: dict[str, Telemetry] = {}
618# To share :py:class:`Telemetry` proxy objects within a process (e.g. client, server).
619#
620# Forking isn't expected to happen while this is held (may lead to a deadlock).
621_TELEMETRY_PROXIES_LOCK = threading.Lock()
622
623
624def _reinitialize_locks_after_fork() -> None:
625 """
626 Reinitialize telemetry locks after fork to prevent deadlocks.
627
628 This function is called automatically after a fork to reinitialize all locks.
629 Caches and instances are kept as they may still be valid, but locks must be
630 fresh to avoid deadlocks.
631 """
632 global _TELEMETRY, _TELEMETRY_LOCK, _TELEMETRY_PROXIES_LOCK # noqa: PLW0602
633
634 _TELEMETRY_LOCK = threading.Lock()
635 _TELEMETRY_PROXIES_LOCK = threading.Lock()
636
637 # Reinitialize LOCAL mode telemetry instance locks if it exists
638 if _TELEMETRY is not None:
639 _TELEMETRY._reinitialize_instance_locks_after_fork()
640
641
642if hasattr(os, "register_at_fork"):
643 os.register_at_fork(after_in_child=_reinitialize_locks_after_fork)
644
645
[docs]
646class TelemetryMode(enum.Enum):
647 """
648 How to create a :py:class:`Telemetry` object.
649 """
650
651 #: Keep everything local to the process (not fork-safe).
652 LOCAL = "local"
653 #: Start + connect to a telemetry IPC server.
654 SERVER = "server"
655 #: Connect to a telemetry IPC server.
656 CLIENT = "client"
657
658
659def _telemetry_proxies_key(
660 mode: Literal[TelemetryMode.SERVER, TelemetryMode.CLIENT], address: str | tuple[str, int]
661) -> str:
662 """
663 Get the key for the _TELEMETRY_PROXIES dictionary.
664 """
665 init_options = {"mode": mode.value, "address": address}
666 return json.dumps(init_options, sort_keys=True)
667
668
669def _telemetry_manager_server_port(process_id: int) -> int:
670 """
671 Get the default telemetry manager server port.
672
673 This is PID-based to:
674
675 * Avoid collisions between multiple independent Python interpreters running on the same machine.
676 * Let child processes deterministically find their parent's telemetry manager server.
677
678 :param process_id: Process ID used to calculate the port.
679 """
680
681 # Use the dynamic/private/ephemeral port range.
682 #
683 # https://www.rfc-editor.org/rfc/rfc6335.html#section-6
684 # https://en.wikipedia.org/wiki/List_of_TCP_and_UDP_port_numbers#Dynamic,_private_or_ephemeral_ports
685 #
686 # Modulo the parent/child process PID by the port range length, then add the initial offset.
687 return (2**15 + 2**14) + (process_id % ((2**16) - (2**15 + 2**14)))
688
689
690def _init_server(address: str | tuple[str, int] | None = None) -> Telemetry:
691 """
692 Start + connect to a telemetry IPC server.
693 """
694 global _TELEMETRY_PROXIES # noqa: PLW0602
695 global _TELEMETRY_PROXIES_LOCK # noqa: PLW0602
696
697 address = address or ("127.0.0.1", _telemetry_manager_server_port(process_id=psutil.Process().pid))
698 telemetry_proxies_key = _telemetry_proxies_key(mode=TelemetryMode.SERVER, address=address)
699
700 with _TELEMETRY_PROXIES_LOCK:
701 if telemetry_proxies_key in _TELEMETRY_PROXIES:
702 return _TELEMETRY_PROXIES[telemetry_proxies_key]
703 else:
704 telemetry_manager = TelemetryManager(
705 address=address,
706 # Use spawn instead of the platform-specific default (may be fork) to avoid aforementioned issues with fork.
707 ctx=multiprocessing.get_context(method="spawn"),
708 )
709
710 logger.debug(f"Creating telemetry manager server at {telemetry_manager.address}.")
711 try:
712 telemetry_manager.start()
713 atexit.register(telemetry_manager.shutdown)
714 logger.debug(f"Started telemetry manager server at {telemetry_manager.address}.")
715 except Exception:
716 logger.debug(
717 f"Failed to create telemetry manager server at {telemetry_manager.address}!", exc_info=True
718 )
719 raise
720
721 logger.debug(f"Connecting to telemetry manager server at {telemetry_manager.address}.")
722 try:
723 telemetry_manager.connect()
724 logger.debug(f"Connected to telemetry manager server at {telemetry_manager.address}.")
725 return _TELEMETRY_PROXIES.setdefault(telemetry_proxies_key, telemetry_manager.Telemetry()) # pyright: ignore [reportAttributeAccessIssue]
726 except Exception:
727 logger.debug(
728 f"Failed to connect to telemetry manager server at {telemetry_manager.address}!", exc_info=True
729 )
730 raise
731
732
733def _init_client(address: str | tuple[str, int] | None = None) -> Telemetry:
734 """
735 Connect to a telemetry IPC server.
736 """
737 global _TELEMETRY_PROXIES # noqa: PLW0602
738 global _TELEMETRY_PROXIES_LOCK # noqa: PLW0602
739
740 candidate_addresses: list[str | tuple[str, int]] = []
741
742 if address is not None:
743 candidate_addresses = [address]
744 else:
745 current_process = psutil.Process()
746 # Python processes from leaf to root.
747 python_process_ancestry: list[psutil.Process] = [
748 current_process,
749 # Try the default telemetry manager server port for all ancestor process IDs.
750 #
751 # psutil is used since multiprocessing only exposes the parent process.
752 #
753 # We can't use `itertools.takewhile(lambda ancestor_process: ancestor_process.name() == current_process.name(), ...)`
754 # to limit ourselves to ancestor Python processes by process name since some may not be named
755 # `python{optional version}` in some cases (e.g. may be named `pytest`).
756 *current_process.parents(),
757 ]
758 # Try to connect from leaf to root.
759 candidate_addresses = [
760 ("127.0.0.1", _telemetry_manager_server_port(process_id=process.pid)) for process in python_process_ancestry
761 ]
762
763 for candidate_address in candidate_addresses:
764 telemetry_proxies_key = _telemetry_proxies_key(mode=TelemetryMode.CLIENT, address=candidate_address)
765
766 with _TELEMETRY_PROXIES_LOCK:
767 if telemetry_proxies_key in _TELEMETRY_PROXIES:
768 return _TELEMETRY_PROXIES[telemetry_proxies_key]
769 else:
770 telemetry_manager = TelemetryManager(address=candidate_address)
771
772 logger.debug(f"Connecting to telemetry manager server at {telemetry_manager.address}.")
773 try:
774 telemetry_manager.connect()
775 logger.debug(f"Connected to telemetry manager server at {telemetry_manager.address}.")
776 return _TELEMETRY_PROXIES.setdefault(telemetry_proxies_key, telemetry_manager.Telemetry()) # pyright: ignore [reportAttributeAccessIssue]
777 except Exception:
778 logger.debug(
779 f"Failed to connect to telemetry manager server at {telemetry_manager.address}!",
780 exc_info=True,
781 )
782
783 raise ConnectionError(f"Failed to connect to telemetry manager server at any of {candidate_addresses}!")
784
785
[docs]
786def init(
787 mode: TelemetryMode | None = None,
788 address: str | tuple[str, int] | None = None,
789) -> Telemetry:
790 """
791 Create or return an existing :py:class:`Telemetry` instance or :py:class:`Telemetry` proxy object.
792
793 :param mode: How to create a :py:class:`Telemetry` object. If ``None``, the default heuristic chooses a mode based on the presence of telemetry IPC servers in the process tree.
794 :param address: Telemetry IPC server address. Passed directly to a :py:class:`multiprocessing.managers.BaseManager`. Ignored if the mode is :py:const:`TelemetryMode.LOCAL`.
795 :return: A telemetry instance.
796 """
797
798 if mode is None:
799 # Main process.
800 if multiprocessing.parent_process() is None:
801 # Daemons can't have child processes.
802 #
803 # Try to create a telemetry instance in local mode.
804 #
805 # ⚠️ This may cause CPU contention if the current process is compute-intensive
806 # and a high collect and/or export frequency is used due to global interpreter lock (GIL).
807 if multiprocessing.current_process().daemon:
808 return _init()
809 # Start + connect to a telemetry IPC server.
810 else:
811 return _init_server(address=address)
812 # Child process.
813 else:
814 # Connect to a telemetry IPC server.
815 try:
816 return _init_client(address=address)
817 # No existing telemetry IPC server.
818 except ConnectionError:
819 # Daemons can't have child processes.
820 #
821 # Try to create a telemetry instance in local mode.
822 #
823 # ⚠️ This may cause CPU contention if the current process is compute-intensive
824 # and a high collect and/or export frequency is used due to global interpreter lock (GIL).
825 if multiprocessing.current_process().daemon:
826 return _init()
827 # Start + connect to a telemetry IPC server.
828 else:
829 return _init_server(address=address)
830 elif mode == TelemetryMode.LOCAL:
831 return _init()
832 elif mode == TelemetryMode.SERVER:
833 return _init_server(address=address)
834 elif mode == TelemetryMode.CLIENT:
835 return _init_client(address=address)
836 else:
837 raise ValueError(f"Unsupported telemetry mode: {mode}")