Coverage for anaconda_opentelemetry/logging.py: 92%

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

2# SPDX-FileCopyrightText: 2025 Anaconda, Inc 

3# SPDX-License-Identifier: Apache-2.0 

4 

5# logging.py 

6""" 

7Anaconda Telemetry - Logging signal class and EventLogger. 

8""" 

9 

10import json 

11import logging 

12from typing import Dict 

13 

14from ._compat import ( 

15 BatchLogRecordProcessor, 

16 ConsoleLogRecordExporter, 

17 LoggerProvider, 

18 LoggingHandler, 

19 get_logger_provider, 

20 set_logger_provider, 

21) 

22from .common import _AnacondaCommon 

23from .config import Configuration as Config 

24from .attributes import ResourceAttributes as Attributes 

25from .exporter_shim import OTLPLogExporterShim 

26from .formatting import AttrDict, EventPayload, log_event_name_key 

27 

28 

29class EventLogger: 

30 """ 

31 Emits log records purely as OTel log telemetry, bypassing Python's 

32 logging hierarchy so they never appear in console/file handlers or 

33 interfere with developer log levels. Should be used if the logs 

34 being collected are not developer logs, and correspond to events. 

35 """ 

36 

37 def __init__( 

38 self, 

39 provider: LoggerProvider, 

40 logger_name: str = "event_logger", 

41 ): 

42 self._logger = provider.get_logger(logger_name) 

43 

44 def _send_event( 

45 self, 

46 body: EventPayload, 

47 event_name: str, 

48 attributes: AttrDict={}, 

49 ): 

50 if not isinstance(body, str): 

51 body = json.dumps(body) 

52 # update attributes with event name - mandatory for event logs 

53 # copy first: `attributes` may be a caller-owned dict (or the default {}) 

54 attributes = {**attributes, log_event_name_key: event_name} 

55 self._logger.emit(body=body, attributes=attributes) 

56 

57 

58class _AnacondaLogger(_AnacondaCommon): 

59 # Singleton instance (internal only); provide a logger handler for OpenTelemetry log instrumentation 

60 _instance = None 

61 

62 _default_log_attributes = {log_event_name_key: "__LOG__"} 

63 

64 def __init__(self, config: Config, attributes: Attributes): 

65 super().__init__(config, attributes) 

66 self.log_level = self._get_log_level(config._get_logging_level()) 

67 self.logger_endpoint = config._get_logging_endpoint() 

68 

69 # send_event and get_telemetry_logger_handler emit through self._provider, not the 

70 # global one. set_logger_provider ignores a second caller instead of raising, so 

71 # check identity to detect that another library got there first. 

72 self._provider = LoggerProvider(resource=self.resource, shutdown_on_exit=self._shutdown_on_exit) 

73 set_logger_provider(self._provider) 

74 if get_logger_provider() is not self._provider: 

75 logging.getLogger(__package__).debug( 

76 "A global OTel LoggerProvider was already set by another library; " 

77 "anaconda_opentelemetry keeps using its own provider for log telemetry." 

78 ) 

79 self._console_exporter: ConsoleLogRecordExporter | None = None 

80 # Add OTLP exporter 

81 if self.use_console_exporters: 

82 exporter = ConsoleLogRecordExporter() 

83 self._console_exporter = exporter 

84 else: 

85 auth_token = config._get_auth_token_logging() 

86 headers: Dict[str, str] = {} 

87 if auth_token is not None: 

88 headers['authorization'] = f'Bearer {auth_token}' 

89 if config._get_request_protocol_logging() in ['grpc', 'grpcs']: # gRPC 

90 from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter as OTLPLogExportergRPC 

91 insecure = not config._get_TLS_logging() 

92 exporter = OTLPLogExporterShim( 

93 OTLPLogExportergRPC, 

94 endpoint=self.logger_endpoint, 

95 insecure=insecure, 

96 credentials=config._get_ca_cert_logging() if not insecure else None, 

97 headers=headers 

98 ) 

99 else: # HTTP 

100 from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter as OTLPLogExporterHTTP 

101 http_kwargs = self._build_http_exporter_kwargs('logging', self.logger_endpoint, headers) 

102 exporter = OTLPLogExporterShim( 

103 OTLPLogExporterHTTP, 

104 **http_kwargs 

105 ) 

106 

107 self.exporter = exporter 

108 self._processor = BatchLogRecordProcessor(self.exporter) 

109 self._provider.add_log_record_processor(self._processor) 

110 

111 def _get_log_handler(self) -> LoggingHandler: 

112 handler = LoggingHandler(level=self.log_level, logger_provider=self._provider) 

113 handler.addFilter(self._set_default_attribute_filter()) 

114 return handler 

115 

116 def _set_default_attribute_filter(self) -> logging.Filter: 

117 attrs = self._default_log_attributes 

118 f = logging.Filter() 

119 def _filter(record): 

120 for k, v in attrs.items(): 

121 if not hasattr(record, k): 

122 setattr(record, k, v) 

123 return True 

124 f.filter = _filter 

125 return f 

126 

127 def _get_event_logger(self, logger_name: str = None) -> EventLogger: 

128 if logger_name is None: 

129 logger_name = f'{self.service_name}_event_logger' 

130 return EventLogger(self._provider, logger_name=logger_name) 

131 

132 def _get_log_level(self, str_level: str)-> int: 

133 # Convert string from config file to logging level. 

134 levels = { 

135 "debug": logging.DEBUG, 

136 "info": logging.INFO, 

137 "warning": logging.WARNING, 

138 "warn": logging.WARNING, 

139 "error": logging.ERROR, 

140 "critical": logging.CRITICAL, 

141 "fatal": logging.CRITICAL 

142 } 

143 return levels.get(str_level.lower(), logging.DEBUG) 

144 

145 def _test_set_console_mock(self, new_out): # For testing only... 

146 if self._console_exporter is not None and new_out is not None: 

147 saved = self._console_exporter.out 

148 self._console_exporter.out = new_out 

149 return saved 

150 return None