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 # Hash any attributes with the hash property 

52 attributes._hash_attributes() 

53 # Read resource attributes 

54 resource_attrs = attributes._get_attributes() 

55 # Required parameters 

56 self.service_name = resource_attrs["service_name"] 

57 self.service_version = resource_attrs["service_version"] 

58 # prepare to use `_process_attributes` 

59 self._user_id = resource_attrs["user_id"] 

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

61 

62 # convert parameters value to stringified JSON 

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

64 # Init resource_attributes 

65 self._resource_attributes = { 

66 SERVICE_NAME: self.service_name, 

67 SERVICE_VERSION: self.service_version 

68 } 

69 self._resource_attributes.update(resource_attrs) 

70 # convert to otel names 

71 for attr in fields(attributes): 

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

73 if otel_name: 

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

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

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

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

78 

79 def _hash_session_id(self, entropy): 

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

81 # entropy value ensures unique session_ids 

82 if entropy is None: 

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

84 

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

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

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

88 

89 return hashed 

90 

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

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

93 kwargs = dict( 

94 endpoint=endpoint, 

95 certificate_file=get_ca_cert(), 

96 headers=headers, 

97 **extra_kwargs 

98 ) 

99 session = self._config._create_proxy_session() 

100 if session is not None: 

101 kwargs['session'] = session 

102 return kwargs 

103 

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

105 # ensure attributes are of type AttrDict 

106 if not isinstance(attributes, Dict): 

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

108 attributes = {} 

109 # check attributes for invalid keys and filter them out 

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

111 if invalid_keys: 

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

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

114 

115 processed = {} 

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

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

118 processed[key] = value 

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

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

121 processed[key] = tuple(value) 

122 else: 

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

124 else: 

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

126 

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

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

129 if not self._user_id: 

130 return processed # no op 

131 elif 'user.id' in processed: 

132 return processed # key already exists 

133 else: 

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

135 return processed