Coverage for anaconda_opentelemetry/signals.py: 92%

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

2# SPDX-FileCopyrightText: 2025 Anaconda, Inc 

3# SPDX-License-Identifier: Apache-2.0 

4 

5# signals.py 

6""" 

7Anaconda Telemetry - Metrics Module 

8 

9This module provides functionality for logging, metrics, and tracing (together called 

10signals) using OpenTelemetry. It includes classes for handling logging, metrics, and 

11tracing, as well as functions for initializing the telemetry system and recording metrics. 

12""" 

13 

14import logging, socket, threading 

15from typing import Dict, Iterator, List, Optional 

16from contextlib import contextmanager 

17 

18from opentelemetry import trace, metrics, _logs 

19from opentelemetry.sdk.trace import TracerProvider 

20from opentelemetry.sdk.metrics import MeterProvider 

21from opentelemetry.sdk._logs import LoggingHandler, LoggerProvider 

22 

23from .config import Configuration as Config 

24from .attributes import ResourceAttributes as Attributes 

25from .formatting import AttrDict 

26 

27from .common import _AnacondaCommon, MetricsNotInitialized 

28from .logging import _AnacondaLogger 

29from .metrics import _AnacondaMetrics 

30from .tracing import _AnacondaTrace, ASpan, _ASpan 

31 

32 

33_SUPPRESSED_LOGGER_ROOTS = ('opentelemetry',) 

34 

35 

36def _suppress_otel_export_errors(): 

37 for root in _SUPPRESSED_LOGGER_ROOTS: 

38 logging.getLogger(root).setLevel(logging.CRITICAL) 

39 prefix = root + '.' 

40 for name, logger in list(logging.Logger.manager.loggerDict.items()): 

41 if name.startswith(prefix) and isinstance(logger, logging.Logger): 

42 logger.setLevel(logging.NOTSET) 

43 

44 

45# Internet and endpoint access check method 

46def __check_internet_status(config: Config, timeout: float = 5.0) -> tuple[bool,bool]: # seconds max to pause.... 

47 # Relies on Configuration to validate the endpoint... 

48 internet = True 

49 access = True 

50 if config._get_skip_internet_check(): 

51 return True, True 

52 endpoint = config._get_default_endpoint() 

53 try: 

54 # Access to a highly available DNS site... 

55 socket.create_connection(('8.8.8.8', 53), timeout=timeout / 2).close() 

56 except OSError: 

57 logging.getLogger(__package__).warning("Anaconda OpenTelemetry: No Internet was detected!") 

58 internet = False # No internet, but internet is not an absolute requirement for on-prem solutions. 

59 try: 

60 socket.create_connection((config._endpoints['default_endpoint'].host, config._endpoints['default_endpoint']._internet_check_port), timeout=timeout / 2).close() 

61 except OSError: 

62 logging.getLogger(__package__).fatal(f"Anaconda OpenTelemetry: No access to the endpoint '{endpoint}'!") 

63 access = False # This could be fatal, not endpoint for telemetry. 

64 if access == True: 

65 logging.getLogger(__package__).info(f"Anaconda OpenTelemetry: Successful access to the endpoint '{endpoint}'!") 

66 return internet, access 

67 

68__ANACONDA_TELEMETRY_INITIALIZED = False 

69__SIGNALS = None 

70__CONFIG = None 

71 

72 

73################################################################################ 

74# Exposed APIs 

75def initialize_telemetry(config: Config, 

76 attributes: Attributes = None, 

77 signal_types: List[str] = ['metrics']): 

78 """ 

79 Initializes the telemetry system. 

80 

81 Args: 

82 service_name (str): The name of the service. 

83 service_version (str): The version of the service. 

84 config (Configuration): The configuration for the telemetry. At a minimum, the Configuration must have a default endpoint 

85 for connection to the collector. 

86 attributes (ResourceAttributes, optional): A class containing common attributes. If provided, 

87 it will override any values shared with configuration file. 

88 signal_types (list, optional): List of metric types to initialize. Defaults to ['logging','metrics','tracing']. 

89 Supported values are 'logging', 'metrics', and 'tracing'. If an empty list is provided, no metrics will be initialized. 

90 

91 Raises: 

92 ValueError: If the config passed is None or the attributes passed are None. 

93 """ 

94 global __ANACONDA_TELEMETRY_INITIALIZED 

95 global __SIGNALS 

96 global __CONFIG 

97 

98 if __ANACONDA_TELEMETRY_INITIALIZED is True: 

99 return # Already initialized 

100 if config is None: 

101 raise ValueError(f"The config argument is required but was None") 

102 if attributes is None: 

103 raise ValueError(f"The attributes argument is required but was None") 

104 

105 __CONFIG = config 

106 __SIGNALS = signal_types 

107 

108 # Check ResourceAttributes object 

109 if attributes is None: 

110 raise ValueError(f"The attributes argument is required but was None") 

111 elif type(attributes.parameters) != dict: 

112 raise ValueError(f"The parameters attribute in ResourceAttributes must be a dictionary") 

113 

114 if not config._get_verbose_export_errors(): 

115 _suppress_otel_export_errors() 

116 

117 # Right now, no action is taken but it possible to disable telemetry with no access to the endpoint... 

118 _, _ = __check_internet_status(config, timeout=4) # Max wait 4 seconds... 

119 

120 # all params are the same currently so only write them once 

121 init_params = (config, attributes) 

122 

123 # Initialize logging here... 

124 signal_type_count = 0 

125 if 'logging' in signal_types: 

126 _AnacondaLogger._instance = _AnacondaLogger(*init_params) 

127 signal_type_count += 1 

128 

129 # Initialize the telemetry system here 

130 if 'metrics' in signal_types: 

131 _AnacondaMetrics._instance = _AnacondaMetrics(*init_params) 

132 signal_type_count += 1 

133 

134 # Initialize tracing here... 

135 if 'tracing' in signal_types: 

136 _AnacondaTrace._instance = _AnacondaTrace(*init_params) 

137 signal_type_count += 1 

138 

139 if signal_type_count == 0: 

140 logging.getLogger(__package__).warning( 

141 "No signal types were initialized. Was this intended? If not please check the " + 

142 "'metrics' section in the configuration file and/or the list of " + 

143 "metric types in the parameter 'signal_types'." 

144 ) 

145 __ANACONDA_TELEMETRY_INITIALIZED = True 

146 

147_SHUTDOWN_DONE = False 

148_shutdown_lock = threading.Lock() 

149 

150 

151def flush_telemetry() -> bool: 

152 """Force-flush all initialized telemetry providers. 

153 

154 Uses the standard OTel global getters to retrieve providers. 

155 Returns True if all providers flushed successfully. 

156 """ 

157 if not __ANACONDA_TELEMETRY_INITIALIZED: 

158 return False 

159 success = True 

160 try: 

161 tp = trace.get_tracer_provider() 

162 if isinstance(tp, TracerProvider): 

163 try: 

164 tp.force_flush() 

165 except Exception: 

166 logging.getLogger(__package__).debug("Tracer flush failed", exc_info=True) 

167 success = False 

168 

169 mp = metrics.get_meter_provider() 

170 if isinstance(mp, MeterProvider): 

171 try: 

172 mp.force_flush() 

173 except Exception: 

174 logging.getLogger(__package__).debug("Meter flush failed", exc_info=True) 

175 success = False 

176 

177 lp = _logs.get_logger_provider() 

178 if isinstance(lp, LoggerProvider): 

179 try: 

180 lp.force_flush() 

181 except Exception: 

182 logging.getLogger(__package__).debug("Logger flush failed", exc_info=True) 

183 success = False 

184 except Exception: 

185 logging.getLogger(__package__).debug("flush_telemetry failed", exc_info=True) 

186 success = False 

187 return success 

188 

189 

190def shutdown_telemetry(*, timeout_seconds: Optional[float] = None) -> bool: 

191 """Flush all telemetry providers at process shutdown, optionally time-bounded. 

192 

193 Performs a bounded *force-flush* (via :func:`flush_telemetry`). It intentionally does 

194 not call ``provider.shutdown()``: at process exit that only adds worker-thread joins 

195 (more blocking) with no benefit. Pair with 

196 ``config.set_shutdown_on_exit(False)`` to control flush timing from a 

197 signal handler or atexit path. 

198 

199 With ``timeout_seconds=None`` the flush runs synchronously (unbounded). When set, the 

200 flush runs on a daemon thread joined for at most ``timeout_seconds``; only the 

201 caller's wait is bounded (a still-running flush thread is reaped at interpreter exit). 

202 

203 Idempotent and thread-safe: once a flush completes it is not repeated; concurrent or 

204 re-entrant (signal-handler) calls never double-flush or block. A call that times out 

205 does not mark completion, so a later call may retry. 

206 

207 The ``join`` blocks the calling thread, so do not call this with a ``timeout_seconds`` 

208 from inside an async event loop; use it from a signal handler, an atexit path, or a 

209 dedicated thread. 

210 

211 Returns True if the flush completed (now or previously); False if telemetry was never 

212 initialized, the flush timed out, or another call is already in progress. 

213 """ 

214 global _SHUTDOWN_DONE 

215 if not __ANACONDA_TELEMETRY_INITIALIZED: 

216 return False 

217 if _SHUTDOWN_DONE: 

218 return True 

219 # Non-blocking so a concurrent or re-entrant caller returns immediately instead of 

220 # double-flushing or deadlocking (this may run inside a signal handler). 

221 if not _shutdown_lock.acquire(blocking=False): 

222 return _SHUTDOWN_DONE 

223 try: 

224 if _SHUTDOWN_DONE: 

225 return True 

226 if timeout_seconds is None: 

227 completed = flush_telemetry() 

228 else: 

229 flush_thread = threading.Thread(target=flush_telemetry, daemon=True) 

230 flush_thread.start() 

231 flush_thread.join(timeout=timeout_seconds) 

232 completed = not flush_thread.is_alive() 

233 if completed: 

234 _SHUTDOWN_DONE = True 

235 return completed 

236 finally: 

237 _shutdown_lock.release() 

238 

239 

240def change_signal_endpoint(signal_type: str, 

241 new_endpoint: str, 

242 auth_token: str = None): 

243 """ 

244 Updates the endpoint for the passed signal 

245 

246 Args: 

247 signal_type (str): signal type to update endpoint for. Supported values are 'logging', 'metrics', and 'tracing' 

248 

249 Returns: 

250 boolean: value indicating whether the update was successful or not 

251 """ 

252 if signal_type.lower() == 'metrics': 

253 _AnacondaTelInstance = _AnacondaMetrics 

254 batch_access = _AnacondaTelInstance._instance.metric_reader 

255 elif signal_type.lower() == 'tracing': 

256 _AnacondaTelInstance = _AnacondaTrace 

257 batch_access = _AnacondaTelInstance._instance._processor 

258 elif signal_type.lower() == 'logging': 

259 _AnacondaTelInstance = _AnacondaLogger 

260 batch_access = _AnacondaTelInstance._instance._processor 

261 else: 

262 logging.getLogger(__package__).warning(f"{signal_type} not a valid signal type.") 

263 return False 

264 

265 # execute OpenTelemetry changes 

266 updated_endpoint = _AnacondaTelInstance._instance.exporter.change_signal_endpoint( 

267 batch_access, 

268 _AnacondaTelInstance._instance._config, 

269 new_endpoint, 

270 auth_token=auth_token 

271 ) 

272 

273 if not updated_endpoint: 

274 logging.getLogger(__package__).warning(f"Endpoint for {signal_type} failed to update.") 

275 return False 

276 else: 

277 logging.getLogger(__package__).info(f"Endpoint for {signal_type} was successfully updated.") 

278 return True 

279 

280def record_histogram(metric_name, value, attributes: AttrDict={}) -> bool: 

281 """ 

282 Records a increasing only metric with the given name and value. The value will 

283 always appear in the attributes section in the raw OTLP output and the timestamp 

284 will be the histogram value. 

285 

286 Will catch any exceptions generated by metric usage. 

287 

288 Args: 

289 metric_name (str): The name of the metric. 

290 value (float): The value of the metric. Can be any float since the timestamp is the ever increasing value of the histogram. 

291 attributes (dict, optional): Additional attributes for the metric. Defaults to {}. 

292 

293 Returns: 

294 bool: True if the metric was recorded successfully, False otherwise (logging the error). 

295 """ 

296 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

297 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") # Since init didn't happen this is not exported in OTel!!! 

298 return False 

299 try: 

300 return _AnacondaMetrics._instance.record_histogram(metric_name, value, _AnacondaMetrics._instance._process_attributes(attributes)) 

301 except MetricsNotInitialized as me: 

302 logging.getLogger(__package__).warning(f"An attempt was made to record a histogram metric when metrics were not configured.") 

303 return False 

304 except Exception as e: 

305 logging.getLogger(__package__).error(f"UNCAUGHT EXCEPTION:\n{e}") 

306 return False 

307 

308def set_gauge(metric_name, value, attributes: AttrDict={}) -> bool: 

309 """ 

310 Sets a gauge metric with the given name to the given value. A gauge records the last 

311 value set rather than a sum, so it is the right choice for values that go up and down 

312 and are sampled rather than accumulated (queue depth, memory in use, temperature). 

313 

314 Will catch any exceptions generated by metric usage. 

315 

316 Args: 

317 metric_name (str): The name of the metric. 

318 value (int | float): The current value of the metric. Must be an int or a float; negative 

319 values are allowed. A value of any other type (including None and bool) is rejected. 

320 attributes (dict, optional): Additional attributes for the metric. Defaults to {}. 

321 

322 Returns: 

323 bool: True if the metric was recorded successfully, False otherwise (logging the error). 

324 """ 

325 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

326 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") # Since init didn't happen this is not exported in OTel!!! 

327 return False 

328 try: 

329 return _AnacondaMetrics._instance.set_gauge(metric_name, value, _AnacondaMetrics._instance._process_attributes(attributes)) 

330 except MetricsNotInitialized: 

331 logging.getLogger(__package__).warning("An attempt was made to set a gauge metric when metrics were not configured.") 

332 return False 

333 except Exception as e: 

334 logging.getLogger(__package__).error(f"UNCAUGHT EXCEPTION:\n{e}") 

335 return False 

336 

337def increment_counter(counter_name, by=1, attributes: AttrDict={}) -> bool: 

338 """ 

339 Increments a counter or up down counter by the given parameter 'by'. 

340 

341 Will catch any exceptions generated by metric usage. 

342 

343 Args: 

344 counter_name (str): The name of the counter. 

345 by (int, optional): The value to increment by. Defaults to 1. The abs(by) is used to protect from negative numbers. 

346 attributes (dict, optional): Additional attributes for the counter. Defaults to {}. 

347 

348 Returns: 

349 bool: True if the counter was incremented successfully, False otherwise (logging the error). 

350 """ 

351 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

352 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") # Since init didn't happen this is not exported in OTel!!! 

353 return False 

354 try: 

355 return _AnacondaMetrics._instance.increment_counter(counter_name, by, _AnacondaMetrics._instance._process_attributes(attributes)) 

356 except MetricsNotInitialized: 

357 logging.getLogger(__package__).warning(f"An attempt was made to change/create a counter metric when metrics were not configured.") 

358 return False 

359 except Exception as e: 

360 logging.getLogger(__package__).error(f"UNCAUGHT EXCEPTION:\n{e}") 

361 return False 

362 

363def decrement_counter(counter_name, by=1, attributes: AttrDict={}) -> bool: 

364 """ 

365 Decrements a up down counter with the given name and value. If applied to a regular counter it will log a warning and silently fail. 

366 

367 Will catch any exceptions generated by metric usage. 

368 

369 Args: 

370 counter_name (str): The name of the counter. 

371 by (int, optional): The value to decrement by. Defaults to 1. abs(by) is used to protect from negative numbers. 

372 attributes (dict, optional): Additional attributes for the counter. Defaults to {}. 

373 

374 Returns: 

375 bool: True if the counter was decremented successfully, False otherwise (logging the error). 

376 """ 

377 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

378 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") # Since init didn't happen this is not exported in OTel!!! 

379 return False 

380 try: 

381 return _AnacondaMetrics._instance.decrement_counter(counter_name, by, _AnacondaMetrics._instance._process_attributes(attributes)) 

382 except MetricsNotInitialized: 

383 logging.getLogger(__package__).warning(f"An attempt was made to change/create a counter metric when metrics were not configured.") 

384 return False 

385 except Exception as e: 

386 logging.getLogger(__package__).error(f"UNCAUGHT EXCEPTION:\n{e}") 

387 return False 

388 

389@contextmanager 

390def get_trace(name: str, attributes: AttrDict = {}, carrier: Dict[str,str] = None) -> Iterator[_ASpan]: 

391 """ 

392 Create or continue a named trace (based on the 'carrier' parameter). 

393 

394 Use the function like a Python I/O object (keyword 'with') to ensure the span is closed properly. 

395 

396 Will catch any exceptions generated by tracing usage. 

397 

398 Args: 

399 name (str): The name of the trace. 

400 attributes (dict, optional): Additional attributes for the trace. Defaults to {}. 

401 carrier (dict, optional): The carrier used to continue a trace context in the output data. Defaults to None. 

402 

403 Example: 

404 with get_trace("my_trace_name", {"key": "value"}) as span: 

405 # Do some work here 

406 pass 

407 # The span will be closed automatically when exiting the 'with' block. 

408 

409 Returns: 

410 Iterator[Tracer]: An iterator for the tracer. 

411 """ 

412 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

413 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") # Since init didn't happen this is not exported in OTel!!! 

414 return None 

415 

416 try: 

417 aspan = _AnacondaTrace._instance.get_span(name, _AnacondaTrace._instance._process_attributes(attributes), carrier) 

418 except: # Trace is different than the other signals, there is no easy way to log and continue. 

419 logging.getLogger(__package__).warning(f"Attempt to trace a with-block when tracing was not configured.") 

420 aspan = _ASpan("UNKNOWN", span=None, noop=True) 

421 try: 

422 yield aspan 

423 except Exception as e: 

424 aspan.add_exception(e) 

425 aspan.set_error_status() 

426 _AnacondaTrace._instance.logger.error(f"Error in trace span {name}: {e}") 

427 finally: 

428 aspan._close() 

429 

430def get_telemetry_logger_handler() -> LoggingHandler: 

431 """ 

432 Returns the telemetry logger handler. This lets the package user control how the application uses the telemetry logger. 

433 Insert this handler into your named logger. 

434 

435 log = logging.getLogger("my_logger") 

436 log.addHandler(get_telemetry_logger_handler()) 

437 

438 Previously, this was injected into the root logger, but this turned out to be problematic for some applications that 

439 wanted to control the logging configuration more precisely. This injection behavior is now disabled. If you wish to 

440 inject the handler into the root logger, you can do so manually. See the Python logging documentation for more information. 

441 

442 Returns: 

443 logging.Logger: The telemetry logger handler if logging was enabled via signal_types in initialize_telemetry, 

444 otherwise this function returns None. 

445 Raises: 

446 RuntimeError: if `initialize_telemetry` has not been called 

447 """ 

448 global __ANACONDA_TELEMETRY_INITIALIZED 

449 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

450 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") # Since init didn't happen this is not exported in OTel!!! 

451 raise RuntimeError("Anaconda telemetry system not initialized.") 

452 if _AnacondaLogger._instance is not None: 

453 return _AnacondaLogger._instance._get_log_handler() 

454 return None # No logger handler available, logging not initialized or not configured. 

455 

456def send_event(body: str, event_name: str, attributes: AttrDict={}) -> bool: 

457 """ 

458 Sends a log event directly to the OpenTelemetry pipeline without using Python's logging module. 

459 This is useful when you want to export log telemetry but don't want the output mixing with 

460 your application's output or developer logs. 

461 

462 Params: 

463 body (str): the log message body 

464 event_name (str): mandatory event name added to attributes 

465 attributes (AttrDict): optional attributes dict 

466 Returns: 

467 bool: True if the event was sent, False if logging was not initialized 

468 Raises: 

469 RuntimeError: if `initialize_telemetry` has not been called 

470 """ 

471 global __ANACONDA_TELEMETRY_INITIALIZED 

472 if __ANACONDA_TELEMETRY_INITIALIZED is False: 

473 logging.getLogger(__package__).error("Anaconda telemetry system not initialized.") 

474 raise RuntimeError("Anaconda telemetry system not initialized.") 

475 if _AnacondaLogger._instance is not None: 

476 event_logger = _AnacondaLogger._instance._get_event_logger() 

477 event_logger._send_event(body, event_name, _AnacondaLogger._instance._process_attributes(attributes)) 

478 return True 

479 return False