Coverage for anaconda_opentelemetry/attributes.py: 99%
74 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-11 20:51 +0000
1# -*- coding: utf-8 -*-
2# SPDX-FileCopyrightText: 2025 Anaconda, Inc
3# SPDX-License-Identifier: Apache-2.0
5# attributes.py
7import hashlib, json, logging, platform, re
8from typing import Dict, Tuple, Literal
9from dataclasses import dataclass, field, fields, InitVar
10from .__version__ import __SDK_VERSION__, __TELEMETRY_SCHEMA_VERSION__
12try:
13 from anaconda_anon_usage import tokens
14 # map token funcs to otel resource attribute names
15 TOKEN_FUNCS = [
16 ("aau.version", tokens.version_token),
17 ("aau.client.token", tokens.client_token),
18 ("aau.session.token", tokens.session_token),
19 ("aau.environment.token", tokens.environment_token),
20 ("aau.organization.tokens", tokens.organization_tokens),
21 ("aau.installer.tokens", tokens.installer_tokens),
22 ("aau.machine.tokens", tokens.machine_tokens),
23 ("aau.anaconda_auth.token", tokens.anaconda_auth_token),
24 ]
25except ImportError:
26 TOKEN_FUNCS = []
29@dataclass
30class ResourceAttributes:
31 """
32 Class used to configure common attributes on initialization and dynamic attributes thereafter
34 Parameters:
35 service_name (str): name of client service. REQUIRED (enforced regex of ^[a-zA-Z0-9._-]{1,30}$), converted later to service.name
36 service_version (str): version of client service. REQUIRED (enforced regex of ^[a-zA-Z0-9._-]{1,30}$), converted later to service.version
37 os_type (str): operating system type of client machine
38 os_version (str): operating system version of client machine
39 python_version (str): python version of client the package
40 hostname (str): hostname of client machine
41 platform (str): infrastructure on which the software is provided
42 environment (Literal["", "test", "development", "staging", "production"]): envrionment the software is running in
43 user_id (str): some string denoting a user of a client application.
44 This will not be stored in Resource Attributes and will be moved to attributes.
45 parameters (Dict[str, str]): optional dictionary containing all other telemetry attributes a client would like to add
46 client_sdk_version (str): version of package. READONLY
47 schema_version (str): version of telemetry schema used by package. READONLY
48 """
49 # settable
50 service_name: str
51 service_version: str
52 os_type: str = field(
53 default="",
54 metadata={"otel_name": "os.type"}
55 )
56 os_version: str = field(
57 default="",
58 metadata={"otel_name": "os.version"}
59 )
60 python_version: str = field(
61 default="",
62 metadata={"otel_name": "python.version"}
63 )
64 hostname: str = field(
65 default="",
66 metadata={"otel_name": "hostname", "hash": True}
67 )
68 platform: str = field(
69 default="",
70 metadata={"otel_name": "platform"}
71 )
72 environment: Literal["", "test", "development", "staging", "production"] = field(
73 default="",
74 metadata={"otel_name": "environment"}
75 )
76 user_id: str = field(
77 default=""
78 )
79 anon_usage: InitVar[bool] = False
80 # Readonly
81 client_sdk_version: str = field(
82 default=__SDK_VERSION__,
83 init=False,
84 metadata={"readonly": True, "otel_name": "client.sdk.version"}
85 )
86 schema_version: str = field(
87 default=__TELEMETRY_SCHEMA_VERSION__,
88 init=False,
89 metadata={"readonly": True, "otel_name": "schema.version"}
90 )
91 parameters: dict = field(
92 default_factory=dict,
93 init=False,
94 metadata={"readonly": True, "otel_name": "parameters"}
95 )
97 def __setattr__(self, key, value):
98 is_aau_key = isinstance(key, str) and key.startswith("aau.")
99 if value is None or key is None:
100 if not is_aau_key:
101 logging.getLogger(__package__).warning(f"Either an attribute or key is None which is not allowed. Attribute: `{key}`. Value: `{value}`")
102 elif hasattr(self, '_readonly_fields') and key in self._readonly_fields:
103 logging.getLogger(__package__).warning(f"Attempted overwrite of readonly common attribute {key}")
104 elif (key == "service_name" or key == "service_version") and not self._check_valid_string(value):
105 raise ValueError(f"{key} not set. {value} is invalid regex for this key: `^[a-zA-Z0-9._-]{{1,30}}$`. This is a required parameter")
106 else:
107 super().__setattr__(
108 str(key),
109 value if key == "parameters" else (json.dumps(value) if isinstance(value, (list, dict)) else str(value))
110 )
112 def __post_init__(self, anon_usage: bool):
113 # set non-init readonly
114 self.client_sdk_version = __SDK_VERSION__
115 self.schema_version = __TELEMETRY_SCHEMA_VERSION__
116 self._readonly_fields = {
117 f.name for f in fields(self)
118 if f.metadata.get("readonly", False) is True
119 }
120 # default certain attribute values if needed
121 if not self.os_type or not self.os_version:
122 self.os_type, self.os_version = self._get_os_info()
123 if not self.python_version:
124 self.python_version = platform.python_version()
125 if not self.hostname:
126 self.hostname = self._get_host_name()
128 # if anon-usage is specified
129 if anon_usage:
130 for name, func in TOKEN_FUNCS:
131 self.__setattr__(name, func())
133 # check for valid environment
134 valid_environments = {"", "test", "development", "staging", "production"}
136 # enforce lowercase
137 self.environment = self.environment.strip().lower()
138 if self.environment not in valid_environments:
139 logging.getLogger(__package__).warning(f"Invalid environment value `{self.environment}`, setting to empty string. Envrionment must be in {valid_environments}")
140 self.environment = ""
142 def _get_os_info(self) -> Tuple[str, str]:
143 """Get system OS type and version"""
144 return platform.system(), platform.release()
146 def _get_host_name(self) -> str:
147 """Get the hostname of the machine"""
148 from socket import gethostname
149 return gethostname()
151 def _check_valid_string(self, value) -> bool:
152 """Check that service_name and service_version match valid regex"""
153 if re.match(r"^[a-zA-Z0-9._-]{1,30}$", str(value)):
154 return True
155 return False
157 def _get_attributes(self) -> Dict[str, str]:
158 """Convert all attributes to a dictionary"""
159 return {k: v for k, v in self.__dict__.items() if k != '_readonly_fields'}
161 def _hash_attributes(self) -> None:
162 """Hash any attributes that have the hash metadata property set to True"""
163 for f in fields(self):
164 if f.metadata.get("hash", False) is True:
165 attr_value = getattr(self, f.name, "")
166 if attr_value:
167 hashed = hashlib.sha256(str(attr_value).encode("utf-8")).hexdigest()
168 setattr(self, f.name, hashed)
170 def set_attributes(self, **kwargs) -> None:
171 """
172 Sets attributes according to key value pairs passed to this function. Will overwrite existing attributes, unless they are readonly.
174 Note: Setting user_id via this method is maintained for backwards compatability. Doing so will override any user_ids set later in event specific attributes.
176 Parameters:
177 \\*\\*kwargs: any keyword arguments. This can set named class properties (common attributes), or any other wildcard name (stored in `parameters`)
178 The following are the common attributes that can be set:
179 service_name (str): name of client service\n
180 service_version (str): version of client service\n
181 os_type (str): operating system type of client machine\n
182 os_version (str): operating system version of client machine\n
183 python_version (str): python version of client the package\n
184 hostname (str): hostname of client machine\n
185 platform (str): infrastructure on which the software runs\n
186 environment (Literal["", "test", "development", "staging", "production"]): environment of the software\n
187 user_id (str): some string denoting a user of a client application\n
188 """
189 for kwarg in kwargs:
190 # if kwarg has already been initialized as a property
191 if kwarg in self.__dict__.keys():
192 self.__setattr__(kwarg, kwargs[kwarg])
193 else:
194 self.parameters[str(kwarg)] = str(kwargs[kwarg])
196 return self