Coverage for anaconda_opentelemetry/attributes.py: 94%
83 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-01 20:06 +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, List
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 auto_collect (bool): if True (default), automatically collects os_type, os_version, python_version, and hostname when not provided
46 exclude_auto_collect (List[str]): list of attribute names to exclude from auto-collection. Valid values: "os_type", "os_version", "python_version", "hostname"
47 anon_usage (bool): if True, collects anonymous usage tokens from anaconda-anon-usage package
48 parameters (Dict[str, str]): optional dictionary containing all other telemetry attributes a client would like to add
49 client_sdk_version (str): version of package. READONLY
50 schema_version (str): version of telemetry schema used by package. READONLY
51 """
52 # settable
53 service_name: str
54 service_version: str
55 os_type: str = field(
56 default="",
57 metadata={"otel_name": "os.type", "auto_collect": True}
58 )
59 os_version: str = field(
60 default="",
61 metadata={"otel_name": "os.version", "auto_collect": True}
62 )
63 python_version: str = field(
64 default="",
65 metadata={"otel_name": "python.version", "auto_collect": True}
66 )
67 hostname: str = field(
68 default="",
69 metadata={"otel_name": "hostname", "hash": True, "auto_collect": True}
70 )
71 platform: str = field(
72 default="",
73 metadata={"otel_name": "platform"}
74 )
75 environment: Literal["", "test", "development", "staging", "production"] = field(
76 default="",
77 metadata={"otel_name": "environment"}
78 )
79 user_id: str = field(
80 default=""
81 )
82 auto_collect: InitVar[bool] = True
83 exclude_auto_collect: InitVar[List[str]] = None
84 anon_usage: InitVar[bool] = False
85 # Readonly
86 client_sdk_version: str = field(
87 default=__SDK_VERSION__,
88 init=False,
89 metadata={"readonly": True, "otel_name": "client.sdk.version"}
90 )
91 schema_version: str = field(
92 default=__TELEMETRY_SCHEMA_VERSION__,
93 init=False,
94 metadata={"readonly": True, "otel_name": "schema.version"}
95 )
96 parameters: dict = field(
97 default_factory=dict,
98 init=False,
99 metadata={"readonly": True, "otel_name": "parameters"}
100 )
102 def __setattr__(self, key, value):
103 is_aau_key = isinstance(key, str) and key.startswith("aau.")
104 if value is None or key is None:
105 if not is_aau_key:
106 logging.getLogger(__package__).warning(f"Either an attribute or key is None which is not allowed. Attribute: `{key}`. Value: `{value}`")
107 elif hasattr(self, '_readonly_fields') and key in self._readonly_fields:
108 logging.getLogger(__package__).warning(f"Attempted overwrite of readonly common attribute {key}")
109 elif (key == "service_name" or key == "service_version") and not self._check_valid_string(value):
110 raise ValueError(f"{key} not set. {value} is invalid regex for this key: `^[a-zA-Z0-9._-]{{1,30}}$`. This is a required parameter")
111 else:
112 processed_value = value if key == "parameters" else (json.dumps(value) if isinstance(value, (list, dict)) else str(value))
113 # check if this attribute should be hashed
114 if processed_value:
115 for f in fields(self):
116 if f.name == key and f.metadata.get("hash", False) is True:
117 processed_value = hashlib.sha256(processed_value.encode("utf-8")).hexdigest()
118 break
119 super().__setattr__(str(key), processed_value)
121 def __post_init__(self, auto_collect: bool, exclude_auto_collect: List[str], anon_usage: bool):
122 # set non-init readonly
123 self.client_sdk_version = __SDK_VERSION__
124 self.schema_version = __TELEMETRY_SCHEMA_VERSION__
125 self._readonly_fields = {
126 f.name for f in fields(self)
127 if f.metadata.get("readonly", False) is True
128 }
129 self._auto_collect_fields = {
130 f.name for f in fields(self)
131 if f.metadata.get("auto_collect", False) is True
132 }
133 # default certain attribute values if needed (only if auto_collect is enabled)
134 if auto_collect:
135 exclude_set = set(exclude_auto_collect or [])
136 if not self.os_type or not self.os_version:
137 if "os_type" not in exclude_set and "os_version" not in exclude_set:
138 self.os_type, self.os_version = self._get_os_info()
139 elif "os_type" not in exclude_set:
140 self.os_type = self._get_os_info()[0]
141 elif "os_version" not in exclude_set:
142 self.os_version = self._get_os_info()[1]
143 if not self.python_version and "python_version" not in exclude_set:
144 self.python_version = platform.python_version()
145 if not self.hostname and "hostname" not in exclude_set:
146 self.hostname = self._get_host_name()
148 # if anon-usage is specified
149 if anon_usage:
150 for name, func in TOKEN_FUNCS:
151 self.__setattr__(name, func())
153 # check for valid environment
154 valid_environments = {"", "test", "development", "staging", "production"}
156 # enforce lowercase
157 self.environment = self.environment.strip().lower()
158 if self.environment not in valid_environments:
159 logging.getLogger(__package__).warning(f"Invalid environment value `{self.environment}`, setting to empty string. Envrionment must be in {valid_environments}")
160 self.environment = ""
162 def _get_os_info(self) -> Tuple[str, str]:
163 """Get system OS type and version"""
164 return platform.system(), platform.release()
166 def _get_host_name(self) -> str:
167 """Get the hostname of the machine"""
168 from socket import gethostname
169 return gethostname()
171 def _check_valid_string(self, value) -> bool:
172 """Check that service_name and service_version match valid regex"""
173 if re.match(r"^[a-zA-Z0-9._-]{1,30}$", str(value)):
174 return True
175 return False
177 def _get_attributes(self) -> Dict[str, str]:
178 """Convert all attributes to a dictionary, omitting auto-collected attributes when empty"""
179 return {
180 k: v for k, v in self.__dict__.items()
181 if k not in ('_readonly_fields', '_auto_collect_fields') and not (k in self._auto_collect_fields and v == '')
182 }
184 def set_attributes(self, **kwargs) -> None:
185 """
186 Sets attributes according to key value pairs passed to this function. Will overwrite existing attributes, unless they are readonly.
188 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.
190 Parameters:
191 \\*\\*kwargs: any keyword arguments. This can set named class properties (common attributes), or any other wildcard name (stored in `parameters`)
192 The following are the common attributes that can be set:
193 service_name (str): name of client service\n
194 service_version (str): version of client service\n
195 os_type (str): operating system type of client machine\n
196 os_version (str): operating system version of client machine\n
197 python_version (str): python version of client the package\n
198 hostname (str): hostname of client machine\n
199 platform (str): infrastructure on which the software runs\n
200 environment (Literal["", "test", "development", "staging", "production"]): environment of the software\n
201 user_id (str): some string denoting a user of a client application\n
202 """
203 for kwarg in kwargs:
204 # if kwarg has already been initialized as a property
205 if kwarg in self.__dict__.keys():
206 self.__setattr__(kwarg, kwargs[kwarg])
207 else:
208 self.parameters[str(kwarg)] = str(kwargs[kwarg])
210 return self