Coverage for anaconda_opentelemetry/common.py: 100%

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

2# SPDX-FileCopyrightText: 2025 Anaconda, Inc 

3# SPDX-License-Identifier: Apache-2.0 

4 

5# common.py 

6""" 

7Anaconda Telemetry - Common base class and exceptions for signal classes. 

8""" 

9 

10import logging, hashlib, json 

11from typing import Dict 

12from dataclasses import fields 

13 

14from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION 

15 

16from .config import Configuration as Config 

17from .attributes import ResourceAttributes as Attributes 

18from .__version__ import __SDK_VERSION__, __TELEMETRY_SCHEMA_VERSION__ 

19from .formatting import AttrDict 

20 

21class MetricsNotInitialized(RuntimeError): 

22 pass 

23 

24 

25class _AnacondaCommon: 

26 # Base class for common attributes and methods (internal only) 

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

28 self._config = config 

29 # Init resource_attributes 

30 self._resource_attributes = {} 

31 self.resource = None 

32 # session id 

33 self._session_id = None 

34 # user id 

35 self._user_id = None 

36 

37 # Make self._resource_attributes and self.resource 

38 self.make_otel_resource(attributes) 

39 

40 self.logger = logging.getLogger(__package__) 

41 

42 # assemble config and attribute values 

43 # default endpoint 

44 self.default_endpoint = config._get_default_endpoint() 

45 # export options 

46 self.use_console_exporters = config._get_console_exporter() 

47 # shutdown on exit flag 

48 self._shutdown_on_exit = config._get_shutdown_on_exit() 

49 

50 def make_otel_resource(self, attributes: Attributes): 

51 # Read resource attributes 

52 resource_attrs = attributes._get_attributes() 

53 # Required parameters 

54 self.service_name = resource_attrs["service_name"] 

55 self.service_version = resource_attrs["service_version"] 

56 # prepare to use `_process_attributes` 

57 self._user_id = resource_attrs["user_id"] 

58 del resource_attrs["service_name"], resource_attrs["service_version"], resource_attrs["user_id"] 

59 

60 # convert parameters value to stringified JSON 

61 resource_attrs["parameters"] = json.dumps(resource_attrs["parameters"]) 

62 # Init resource_attributes 

63 self._resource_attributes = { 

64 SERVICE_NAME: self.service_name, 

65 SERVICE_VERSION: self.service_version 

66 } 

67 self._resource_attributes.update(resource_attrs) 

68 # convert to otel names 

69 for attr in fields(attributes): 

70 otel_name = attr.metadata.get('otel_name', None) 

71 if otel_name and attr.name in self._resource_attributes: 

72 self._resource_attributes[attr.metadata['otel_name']] = self._resource_attributes.pop(attr.name) 

73 if not self._config._get_disable_session_id(): 

74 self._session_id = self._hash_session_id(self._config._get_tracing_session_entropy()) 

75 self._resource_attributes['session.id'] = self._session_id 

76 self.resource = Resource.create(self._resource_attributes) 

77 

78 def _hash_session_id(self, entropy): 

79 # Hashes a session id for common attributes based on timestamp and user_id 

80 # entropy value ensures unique session_ids 

81 if entropy is None: 

82 raise KeyError("The entropy key has been removed.") 

83 

84 user_id = self._resource_attributes.get('user.id', '') 

85 combined = f"{entropy}|{user_id}|{self.service_name}" 

86 hashed = hashlib.sha256(combined.encode("utf-8")).hexdigest() 

87 

88 return hashed 

89 

90 def _build_http_exporter_kwargs(self, signal: str, endpoint: str, headers: Dict[str, str], **extra_kwargs) -> Dict: 

91 get_ca_cert = getattr(self._config, f"_get_ca_cert_{signal}") 

92 kwargs = dict( 

93 endpoint=endpoint, 

94 certificate_file=get_ca_cert(), 

95 headers=headers, 

96 **extra_kwargs 

97 ) 

98 session = self._config._create_proxy_session() 

99 if session is not None: 

100 kwargs['session'] = session 

101 return kwargs 

102 

103 def _process_attributes(self, attributes: AttrDict={}): 

104 # ensure attributes are of type AttrDict 

105 if not isinstance(attributes, Dict): 

106 self.logger.debug(f"Attributes `{attributes}` are not a dictionary, they are not valid. They will be converted to an empty one.") 

107 attributes = {} 

108 # check attributes for invalid keys and filter them out 

109 invalid_keys = [k for k in attributes if not isinstance(k, str) or not k] 

110 if invalid_keys: 

111 self.logger.debug(f"Dropping attributes with invalid keys: {invalid_keys}") 

112 attributes = {k: v for k, v in attributes.items() if isinstance(k, str) and k} 

113 

114 processed = {} 

115 for key, value in attributes.items(): 

116 if isinstance(value, (str, bool, int, float)): 

117 processed[key] = value 

118 elif isinstance(value, (list, tuple)): 

119 if all(isinstance(item, (str, bool, int, float)) for item in value): 

120 processed[key] = tuple(value) 

121 else: 

122 self.logger.debug(f"Skipping attribute '{key}' - sequence contains non-primitive types") 

123 else: 

124 self.logger.debug(f"Skipping attribute '{key}' with unsupported type {type(value).__name__}") 

125 

126 # pulls a user id initially passed to ResourceAttributes and adds it to event specific events 

127 # for backwards compatability if people have been setting user.id with ResourceAttributes 

128 if not self._user_id: 

129 return processed # no op 

130 elif 'user.id' in processed: 

131 return processed # key already exists 

132 else: 

133 processed['user.id'] = self._user_id 

134 return processed