# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0
# pyright: reportUnnecessaryIsInstance=false
import copy
import logging
import threading
from collections.abc import Mapping, MutableMapping, Sequence
from types import NoneType
from typing import TYPE_CHECKING, Any, overload
from typing_extensions import assert_never, deprecated
from opentelemetry.util import types
_logger = logging.getLogger(__name__)
# Calling str(x) will use an object's `__str__` method if it exists, otherwise it will use it's `__repr__` method.
# If neither is defined it uses the base class's `object.__repr__` method, which returns a string that is hard to understand.
# So in that case we drop the key/value pair.
def _is_non_custom_str(key: Any) -> bool:
return type(key).__str__ is not object.__str__ or type(key).__repr__ is not object.__repr__
@overload
def _clean_attribute_value(
value: Mapping[str, types.AnyValue],
max_string_value_length: int | None,
) -> Mapping[str, types.AnyValue]: ...
@overload
def _clean_attribute_value(
value: types.AnyValue,
max_string_value_length: int | None,
) -> types.AnyValue: ...
def _clean_attribute_value(
value: types.AnyValue,
max_string_value_length: int | None,
) -> types.AnyValue:
"""Recursively checks if an attribute value is valid and cleans it if required.
String values are truncated to max_string_value_length if provided.
Anything that isn't of `types.AnyValue`, we attempt to cast to `str`.
If this fails, the value is replaced with None. Sequence's are converted to tuples and mappings
are copied into new dicts.
Returns:
The recursively cleaned AnyValue.
"""
if isinstance(value, (NoneType, bool, int, float, bytes)):
return value
if isinstance(value, str):
if max_string_value_length is not None and len(value) > max_string_value_length:
_logger.warning(
"String attribute value exceeds max length of %d, truncating.",
max_string_value_length,
)
value = value[:max_string_value_length]
return value
if isinstance(value, Sequence):
return tuple(_clean_attribute_value(v, max_string_value_length) for v in value)
if isinstance(value, Mapping):
cleaned_mapping: dict[str, types.AnyValue] = {}
for key, val in value.items():
if not key:
_logger.warning(
"invalid attribute key `%s`. must be non-empty string. Dropping key from attributes.",
key,
)
continue
# Spec says to convert unknown types to strings if possible (here and below too).
if not isinstance(key, str):
_logger.warning(
"Invalid type `%s` for attribute key `%s`, must be a str. Key's `__str__/__repr__` method will be called if it exists, otherwise the key/value pair will be dropped.",
type(key),
key,
)
if _is_non_custom_str(key):
key = str(key)
else:
continue
cleaned_mapping[key] = _clean_attribute_value(val, max_string_value_length)
return cleaned_mapping
if TYPE_CHECKING:
assert_never(value)
_logger.warning(
"Invalid type `%s` for attribute value. Expected one of bool, str, None, bytes, int, float or a "
"Mapping or Sequence of those types. Value's __str__ method will be called if it exists, otherwise the value will be replaced with None.",
type(value),
)
if _is_non_custom_str(value):
return str(value)
return None
[docs]
class BoundedAttributes(MutableMapping[str, types.AnyValue]):
"""A dict with a fixed max capacity which cleans and potentially drops values to ensure they are valid attribute values.
Args:
maxlen: The maximum number of attributes to store, use None for no limit.
attributes: The initial attributes to store.
immutable: Defaults to true. Whether to allow adding/removing of attributes after the initialisation of the instance.
max_value_len: The maximum length of string values, use None for no limit.
extended_attributes: Deprecated. Kept for backwards compatibility. Extended attributes are now always used for attributes everywhere.
When the dict is full and a new element is added, the oldest element is dropped. Attributes are made to be immutable when set in this container.
So passing a mutable list as an attribute value, and then mutating it after will not change it's value in this container.
"""
@overload
def __init__(
self,
maxlen: int | None = None,
attributes: types.Attributes = None,
immutable: bool = True,
max_value_len: int | None = None,
) -> None: ...
@overload
@deprecated(
"Creating BoundedAttributes with `extended_attributes` set is deprecated. "
"The `extended_attributes` param is no longer used and will be removed "
"in a future release. Extended attributes are now always used for attributes everywhere."
)
def __init__(
self,
maxlen: int | None = None,
attributes: types.Attributes = None,
immutable: bool = True,
max_value_len: int | None = None,
extended_attributes: bool = False,
) -> None: ...
def __init__(
self,
maxlen: int | None = None,
attributes: types.Attributes = None,
immutable: bool = True,
max_value_len: int | None = None,
extended_attributes: bool = False,
) -> None:
if maxlen is not None and maxlen < 0:
raise ValueError("maxlen must be valid int greater or equal to 0")
if max_value_len is not None and max_value_len < 0:
raise ValueError("max_value_len must be valid int greater or equal to 0")
self._dict: dict[str, types.AnyValue] = {}
self.maxlen = maxlen
self.dropped = 0
self.max_value_len = max_value_len
self._lock = threading.Lock()
# setting False before adding items
self._immutable = False
if attributes:
self._set_items(attributes)
self._immutable = immutable
def __repr__(self) -> str:
return f"{dict(self._dict)}"
def __getitem__(self, key: str) -> types.AnyValue:
return self._dict[key]
def _raise_if_immutable(self) -> None:
if self._immutable:
raise TypeError("Cannot mutate immutable BoundedAttributes")
def __setitem__(self, key: str, value: types.AnyValue) -> None:
self._raise_if_immutable()
if self.maxlen is not None and self.maxlen == 0:
with self._lock:
self.dropped += 1
return
if not key or not isinstance(key, str):
_logger.warning(
"invalid key `%s`. must be non-empty string. Dropping key from attributes.",
key,
)
with self._lock:
self.dropped += 1
return
cleaned = _clean_attribute_value(value, self.max_value_len)
with self._lock:
self._setitem_locked(key, cleaned)
def _set_items(self, attributes: Mapping[str, types.AnyValue]) -> None:
self._raise_if_immutable()
if self.maxlen is not None and self.maxlen == 0:
with self._lock:
self.dropped += len(attributes)
return
cleaned_attributes: Mapping[str, types.AnyValue] = _clean_attribute_value(attributes, self.max_value_len)
with self._lock:
self.dropped += len(attributes) - len(cleaned_attributes)
for key, value in cleaned_attributes.items():
self._setitem_locked(key, value)
def _setitem_locked(self, key: str, value: types.AnyValue) -> None:
if key in self._dict:
del self._dict[key]
if self.maxlen is not None and len(self._dict) >= self.maxlen:
_logger.warning(
"Attributes dict is full. Dropping the oldest key-value pair from attributes to make space for the new key-value pair.",
)
# Dictionaries are insertion ordered in Python, this is the recommended way to get the oldest value.
del self._dict[next(iter(self._dict.keys()))]
self.dropped += 1
self._dict[key] = value
def __delitem__(self, key: str) -> None:
self._raise_if_immutable()
del self._dict[key]
def __iter__(self):
if self._immutable:
return iter(self._dict)
with self._lock:
return iter(list(self._dict))
def __len__(self) -> int:
return len(self._dict)
def __deepcopy__(self, memo: dict[int, Any]) -> "BoundedAttributes":
copy_ = BoundedAttributes(
maxlen=self.maxlen,
immutable=self._immutable,
max_value_len=self.max_value_len,
)
memo[id(self)] = copy_
with self._lock:
# Assign _dict directly to avoid re-cleaning already clean values
# and to bypass the immutability guard in __setitem__
copy_._dict = copy.deepcopy(self._dict, memo)
copy_.dropped = self.dropped
return copy_
[docs]
def copy(self) -> MutableMapping[str, types.AnyValue]:
return self._dict.copy()