Coverage for anaconda_opentelemetry/_compat.py: 84%

1# -*- coding: utf-8 -*- 

2# SPDX-FileCopyrightText: 2025 Anaconda, Inc 

3# SPDX-License-Identifier: Apache-2.0 

4 

5# _compat.py 

6""" 

7Anaconda Telemetry - single point of contact with private OpenTelemetry APIs. 

8 

9Every OpenTelemetry symbol upstream does not guarantee (underscore-prefixed module 

10or name) is imported here and nowhere else, so a ceiling bump is a one-file audit. 

11``docs/source/otel-private-api.md`` holds the inventory and the supported range 

12(opentelemetry-sdk >=1.40.0,<1.44). 

13 

14A private import that disappears degrades to a no-op with one warning, so it cannot 

15raise out of a log handler and take down the host application. 

16 

17Re-exports are annotated ``Any``: each name is bound to unrelated types in the 

18working and degraded branches. 

19""" 

20 

21from __future__ import annotations 

22 

23import logging 

24from typing import Any 

25 

26__all__ = [ 

27 "OTEL_PRIVATE_LOGS_AVAILABLE", 

28 "BatchLogRecordProcessor", 

29 "ConsoleLogRecordExporter", 

30 "GaugeInstrument", 

31 "LoggerProvider", 

32 "LoggingHandler", 

33 "LogRecordExporter", 

34 "get_logger_provider", 

35 "set_logger_provider", 

36] 

37 

38 

39class _UnavailablePrivateAPI: 

40 """Inert stand-in: constructible and subclassable, every operation a no-op.""" 

41 

42 def __init__(self, *args: Any, **kwargs: Any) -> None: 

43 pass 

44 

45 def __getattr__(self, name: str) -> Any: 

46 return _UnavailablePrivateAPI() 

47 

48 def __call__(self, *args: Any, **kwargs: Any) -> Any: 

49 return None 

50 

51 def __bool__(self) -> bool: 

52 return False 

53 

54 

55def _unavailable(*args: Any, **kwargs: Any) -> None: 

56 return None 

57 

58 

59set_logger_provider: Any 

60get_logger_provider: Any 

61LoggerProvider: Any 

62LoggingHandler: Any 

63BatchLogRecordProcessor: Any 

64ConsoleLogRecordExporter: Any 

65LogRecordExporter: Any 

66 

67try: 

68 import opentelemetry._logs as _api_logs 

69 import opentelemetry.sdk._logs as _sdk_logs 

70 import opentelemetry.sdk._logs.export as _sdk_logs_export 

71 

72 set_logger_provider = _api_logs.set_logger_provider 

73 get_logger_provider = _api_logs.get_logger_provider 

74 

75 LoggerProvider = _sdk_logs.LoggerProvider 

76 LoggingHandler = _sdk_logs.LoggingHandler 

77 

78 # `LogExporter` is a deprecated alias that warns on subclassing; both names 

79 # exist across 1.40 - 1.43. 

80 LogRecordExporter = _sdk_logs_export.LogRecordExporter 

81 BatchLogRecordProcessor = _sdk_logs_export.BatchLogRecordProcessor 

82 ConsoleLogRecordExporter = _sdk_logs_export.ConsoleLogRecordExporter 

83 

84 OTEL_PRIVATE_LOGS_AVAILABLE = True 

85except Exception as _exc: # pragma: no cover - only on an unsupported SDK 

86 logging.getLogger(__package__).warning( 

87 "Anaconda OpenTelemetry: the installed opentelemetry-sdk does not expose the " 

88 "private logs API this package requires (%s: %s). Log telemetry is disabled; " 

89 "metrics and tracing are unaffected. Supported range: >=1.40.0,<1.44.", 

90 type(_exc).__name__, 

91 _exc, 

92 ) 

93 set_logger_provider = _unavailable 

94 get_logger_provider = _unavailable 

95 LoggerProvider = _UnavailablePrivateAPI 

96 LoggingHandler = _UnavailablePrivateAPI 

97 LogRecordExporter = _UnavailablePrivateAPI 

98 BatchLogRecordProcessor = _UnavailablePrivateAPI 

99 ConsoleLogRecordExporter = _UnavailablePrivateAPI 

100 OTEL_PRIVATE_LOGS_AVAILABLE = False 

101 

102 

103def _resolve_gauge_instrument() -> Any: 

104 """Return the synchronous gauge instrument class. 

105 

106 `opentelemetry.sdk.metrics` still exposes only `_Gauge` as of 1.43. The public 

107 name is tried first so a promotion needs no code change. Used as a key in the 

108 aggregation-temporality maps, where a miss silently falls back to the reader 

109 default -- hence the warning. 

110 """ 

111 import opentelemetry.sdk.metrics as _sdk_metrics 

112 

113 for name in ("Gauge", "_Gauge"): 

114 instrument = getattr(_sdk_metrics, name, None) 

115 if instrument is not None: 

116 return instrument 

117 logging.getLogger(__package__).warning( 

118 "Anaconda OpenTelemetry: no Gauge instrument found in opentelemetry.sdk.metrics; " 

119 "gauge aggregation temporality will fall back to the metric reader default." 

120 ) 

121 return None 

122 

123 

124GaugeInstrument: Any = _resolve_gauge_instrument()