Coverage for anaconda_opentelemetry / attributes.py: 100%
56 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-17 15:45 +0000
1# -*- coding: utf-8 -*-
2# SPDX-FileCopyrightText: 2025 Anaconda, Inc
3# SPDX-License-Identifier: Apache-2.0
5# attributes.py
7import logging, platform, re
8from typing import Dict, Tuple, Literal
9from dataclasses import dataclass, field, fields
10from .__version__ import __SDK_VERSION__, __TELEMETRY_SCHEMA_VERSION__
12@dataclass
13class ResourceAttributes:
14 """
15 Class used to configure common attributes on initialization and dynamic attributes thereafter
17 Parameters:
18 service_name (str): name of client service. REQUIRED (enforced regex of ^[a-zA-Z0-9._-]{1,30}$), converted later to service.name
19 service_version (str): version of client service. REQUIRED (enforced regex of ^[a-zA-Z0-9._-]{1,30}$), converted later to service.version
20 os_type (str): operating system type of client machine
21 os_version (str): operating system version of client machine
22 python_version (str): python version of client the package
23 hostname (str): hostname of client machine
24 platform (str): infrastructure on which the software is provided
25 environment (Literal["", "test", "development", "staging", "production"]): envrionment the software is running in
26 user_id (str): some string denoting a user of a client application.
27 This will not be stored in Resource Attributes and will be moved to attributes.
28 parameters (Dict[str, str]): optional dictionary containing all other telemetry attributes a client would like to add
29 client_sdk_version (str): version of package. READONLY
30 schema_version (str): version of telemetry schema used by package. READONLY
31 """
32 # settable
33 service_name: str
34 service_version: str
35 os_type: str = field(
36 default="",
37 metadata={"otel_name": "os.type"}
38 )
39 os_version: str = field(
40 default="",
41 metadata={"otel_name": "os.version"}
42 )
43 python_version: str = field(
44 default="",
45 metadata={"otel_name": "python.version"}
46 )
47 hostname: str = field(
48 default="",
49 metadata={"otel_name": "hostname"}
50 )
51 platform: str = field(
52 default="",
53 metadata={"otel_name": "platform"}
54 )
55 environment: Literal["", "test", "development", "staging", "production"] = field(
56 default="",
57 metadata={"otel_name": "environment"}
58 )
59 user_id: str = field(
60 default=""
61 )
62 # Readonly
63 client_sdk_version: str = field(
64 default=__SDK_VERSION__,
65 init=False,
66 metadata={"readonly": True, "otel_name": "client.sdk.version"}
67 )
68 schema_version: str = field(
69 default=__TELEMETRY_SCHEMA_VERSION__,
70 init=False,
71 metadata={"readonly": True, "otel_name": "schema.version"}
72 )
73 parameters: dict = field(
74 default_factory=dict,
75 init=False,
76 metadata={"readonly": True, "otel_name": "parameters"}
77 )
79 def __setattr__(self, key, value):
80 if value is None or key is None:
81 logging.getLogger(__package__).warning(f"Either an attribute or key is None which is not allowed. Attribute: `{key}`. Value: `{value}`")
82 elif hasattr(self, '_readonly_fields') and key in self._readonly_fields:
83 logging.getLogger(__package__).warning(f"Attempted overwrite of readonly common attribute {key}")
84 elif (key == "service_name" or key == "service_version") and not self._check_valid_string(value):
85 raise ValueError(f"{key} not set. {value} is invalid regex for this key: `^[a-zA-Z0-9._-]{{1,30}}$`. This is a required parameter")
86 else:
87 super().__setattr__(
88 str(key),
89 value if key == "parameters" else str(value)
90 )
92 def __post_init__(self):
93 # set non-init readonly
94 self.client_sdk_version = __SDK_VERSION__
95 self.schema_version = __TELEMETRY_SCHEMA_VERSION__
96 self._readonly_fields = {
97 f.name for f in fields(self)
98 if f.metadata.get("readonly", False) is True
99 }
100 # default certain attribute values if needed
101 if not self.os_type or not self.os_version:
102 self.os_type, self.os_version = self._get_os_info()
103 if not self.python_version:
104 self.python_version = platform.python_version()
105 if not self.hostname:
106 self.hostname = self._get_host_name()
108 # check for valid environment
109 valid_environments = {"", "test", "development", "staging", "production"}
111 # enforce lowercase
112 self.environment = self.environment.strip().lower()
113 if self.environment not in valid_environments:
114 logging.getLogger(__package__).warning(f"Invalid environment value `{self.environment}`, setting to empty string. Envrionment must be in {valid_environments}")
115 self.environment = ""
117 def _get_os_info(self) -> Tuple[str, str]:
118 """Get system OS type and version"""
119 return platform.system(), platform.release()
121 def _get_host_name(self) -> str:
122 """Get the hostname of the machine"""
123 from socket import gethostname
124 return gethostname()
126 def _check_valid_string(self, value) -> bool:
127 """Check that service_name and service_version match valid regex"""
128 if re.match(r"^[a-zA-Z0-9._-]{1,30}$", str(value)):
129 return True
130 return False
132 def _get_attributes(self) -> Dict[str, str]:
133 """Convert all attributes to a dictionary"""
134 return {k: v for k, v in self.__dict__.items() if k != '_readonly_fields'}
136 def set_attributes(self, **kwargs) -> None:
137 """
138 Sets attributes according to key value pairs passed to this function. Will overwrite existing attributes, unless they are readonly.
140 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.
142 Parameters:
143 \\*\\*kwargs: any keyword arguments. This can set named class properties (common attributes), or any other wildcard name (stored in `parameters`)
144 The following are the common attributes that can be set:
145 service_name (str): name of client service\n
146 service_version (str): version of client service\n
147 os_type (str): operating system type of client machine\n
148 os_version (str): operating system version of client machine\n
149 python_version (str): python version of client the package\n
150 hostname (str): hostname of client machine\n
151 platform (str): infrastructure on which the software runs\n
152 environment (Literal["", "test", "development", "staging", "production"]): environment of the software\n
153 user_id (str): some string denoting a user of a client application\n
154 """
155 for kwarg in kwargs:
156 # if kwarg has already been initialized as a property
157 if kwarg in self.__dict__.keys():
158 self.__setattr__(kwarg, kwargs[kwarg])
159 else:
160 self.parameters[str(kwarg)] = str(kwargs[kwarg])
162 return self