18505123992

Committed 14 Oct 2025 05:30PM UTC coverage: 86.888% (-0.01%) from 86.899%

Build # 18505123992

Build Type

push

github

Committed by

web-flow

Commit Message

S3: fix `aws-global` validation in CreateBucket (#13250)

Coverage Stats

10 of 10 new or added lines in 4 files covered. (100.0%)

831 existing lines in 40 files now uncovered.

68028 of 78294 relevant lines covered (86.89%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.0

/localstack-core/localstack/utils/analytics/metrics/counter.py

import threading
from collections import defaultdict
from dataclasses import dataclass
from typing import Any

from localstack import config

from .api import Metric
from .registry import MetricRegistry


@dataclass(frozen=True)
class CounterPayload:
    """A data object storing the value of a Counter metric."""

    namespace: str
    name: str
    value: int
    type: str
    schema_version: int

    def as_dict(self) -> dict[str, Any]:
        return {
            "namespace": self.namespace,
            "name": self.name,
            "value": self.value,
            "type": self.type,
            "schema_version": self.schema_version,
        }


@dataclass(frozen=True)
class LabeledCounterPayload:
    """A data object storing the value of a LabeledCounter metric."""

    namespace: str
    name: str
    value: int
    type: str
    schema_version: int
    labels: dict[str, str | float]

    def as_dict(self) -> dict[str, Any]:
        payload_dict = {
            "namespace": self.namespace,
            "name": self.name,
            "value": self.value,
            "type": self.type,
            "schema_version": self.schema_version,
        }

        for i, (label_name, label_value) in enumerate(self.labels.items(), 1):
            payload_dict[f"label_{i}"] = label_name
            payload_dict[f"label_{i}_value"] = label_value

        return payload_dict


class ThreadSafeCounter:
    """
    A thread-safe counter for any kind of tracking.
    This class should not be instantiated directly, use Counter or LabeledCounter  instead.
    """

    _mutex: threading.Lock
    _count: int

    def __init__(self):
        super().__init__()
        self._mutex = threading.Lock()
        self._count = 0

    @property
    def count(self) -> int:
        return self._count

    def increment(self, value: int = 1) -> None:
        """Increments the counter unless events are disabled."""
        if config.DISABLE_EVENTS:
            return

        if value <= 0:
            raise ValueError("Increment value must be positive.")

        with self._mutex:
            self._count += value

    def reset(self) -> None:
        """Resets the counter to zero unless events are disabled."""
        if config.DISABLE_EVENTS:
            return

        with self._mutex:
            self._count = 0


class Counter(Metric, ThreadSafeCounter):
    """
    A thread-safe, unlabeled counter for tracking the total number of occurrences of a specific event.
    This class is intended for metrics that do not require differentiation across dimensions.
    For use cases where metrics need to be grouped or segmented by labels, use `LabeledCounter` instead.
    """

    _type: str

    def __init__(self, namespace: str, name: str, schema_version: int = 1):
        Metric.__init__(self, namespace=namespace, name=name, schema_version=schema_version)
        ThreadSafeCounter.__init__(self)

        self._type = "counter"
        MetricRegistry().register(self)

    def collect(self) -> list[CounterPayload]:
        """Collects the metric unless events are disabled."""
        if config.DISABLE_EVENTS:
            return []

        if self._count == 0:
            # Return an empty list if the count is 0, as there are no metrics to send to the analytics backend.
            return []

        return [
            CounterPayload(
                namespace=self._namespace,
                name=self.name,
                value=self._count,
                type=self._type,
                schema_version=self._schema_version,
            )
        ]


class LabeledCounter(Metric):
    """
    A thread-safe counter for tracking occurrences of an event across multiple combinations of label values.
    It enables fine-grained metric collection and analysis, with each unique label set stored and counted independently.
    Use this class when you need dimensional insights into event occurrences.
    For simpler, unlabeled use cases, see the `Counter` class.
    """

    _type: str
    _labels: list[str]
    _label_values: tuple[str | float | None, ...]
    _counters_by_label_values: defaultdict[tuple[str | float | None, ...], ThreadSafeCounter]

    def __init__(self, namespace: str, name: str, labels: list[str], schema_version: int = 1):
        super().__init__(namespace=namespace, name=name, schema_version=schema_version)

        if not labels:
            raise ValueError("At least one label is required; the labels list cannot be empty.")

        if any(not label for label in labels):
            raise ValueError("Labels must be non-empty strings.")

        if len(labels) > 6:
            raise ValueError("Too many labels: counters allow a maximum of 6.")

        self._type = "counter"
        self._labels = labels
        self._counters_by_label_values = defaultdict(ThreadSafeCounter)
        MetricRegistry().register(self)

    def labels(self, **kwargs: str | float | None) -> ThreadSafeCounter:
        """
        Create a scoped counter instance with specific label values.

        This method assigns values to the predefined labels of a labeled counter and returns
        a ThreadSafeCounter object that allows tracking metrics for that specific
        combination of label values.

        :raises ValueError:
            - If the set of keys provided labels does not match the expected set of labels.
        """
        if set(self._labels) != set(kwargs.keys()):
            raise ValueError(f"Expected labels {self._labels}, got {list(kwargs.keys())}")

        _label_values = tuple(kwargs[label] for label in self._labels)

        return self._counters_by_label_values[_label_values]

    def collect(self) -> list[LabeledCounterPayload]:
        if config.DISABLE_EVENTS:
            return []

        payload = []
        num_labels = len(self._labels)

        for label_values, counter in self._counters_by_label_values.items():
            if counter.count == 0:
                continue  # Skip items with a count of 0, as they should not be sent to the analytics backend.

            if len(label_values) != num_labels:
                raise ValueError(
                    f"Label count mismatch: expected {num_labels} labels {self._labels}, "
                    f"but got {len(label_values)} values {label_values}."
                )

            # Create labels dictionary
            labels_dict = dict(zip(self._labels, label_values, strict=False))

            payload.append(
                LabeledCounterPayload(
                    namespace=self._namespace,
                    name=self.name,
                    value=counter.count,
                    type=self._type,
                    schema_version=self._schema_version,
                    labels=labels_dict,
                )
            )

        return payload

1	import threading	1✔
2	from collections import defaultdict	1✔
3	from dataclasses import dataclass	1✔
4	from typing import Any	1✔
5
6	from localstack import config	1✔
7
8	from .api import Metric	1✔
9	from .registry import MetricRegistry	1✔
10
11
12	@dataclass(frozen=True)	1✔
13	class CounterPayload:	1✔
14	"""A data object storing the value of a Counter metric."""
15
16	namespace: str	1✔
17	name: str	1✔
18	value: int	1✔
19	type: str	1✔
20	schema_version: int	1✔
21
22	def as_dict(self) -> dict[str, Any]:	1✔
23	return {	×
24	"namespace": self.namespace,
25	"name": self.name,
26	"value": self.value,
27	"type": self.type,
28	"schema_version": self.schema_version,
29	}
30
31
32	@dataclass(frozen=True)	1✔
33	class LabeledCounterPayload:	1✔
34	"""A data object storing the value of a LabeledCounter metric."""
35
36	namespace: str	1✔
37	name: str	1✔
38	value: int	1✔
39	type: str	1✔
40	schema_version: int	1✔
41	labels: dict[str, str \| float]	1✔
42
43	def as_dict(self) -> dict[str, Any]:	1✔
44	payload_dict = {	1✔
45	"namespace": self.namespace,
46	"name": self.name,
47	"value": self.value,
48	"type": self.type,
49	"schema_version": self.schema_version,
50	}
51
52	for i, (label_name, label_value) in enumerate(self.labels.items(), 1):	1✔
53	payload_dict[f"label_{i}"] = label_name	1✔
54	payload_dict[f"label_{i}_value"] = label_value	1✔
55
56	return payload_dict	1✔
57
58
59	class ThreadSafeCounter:	1✔
60	"""
61	A thread-safe counter for any kind of tracking.
62	This class should not be instantiated directly, use Counter or LabeledCounter instead.
63	"""
64
65	_mutex: threading.Lock	1✔
66	_count: int	1✔
67
68	def __init__(self):	1✔
69	super().__init__()	1✔
70	self._mutex = threading.Lock()	1✔
71	self._count = 0	1✔
72
73	@property	1✔
74	def count(self) -> int:	1✔
75	return self._count	1✔
76
77	def increment(self, value: int = 1) -> None:	1✔
78	"""Increments the counter unless events are disabled."""
79	if config.DISABLE_EVENTS:	1✔
80	return	1✔
81
82	if value <= 0:	1✔
83	raise ValueError("Increment value must be positive.")	×
84
85	with self._mutex:	1✔
86	self._count += value	1✔
87
88	def reset(self) -> None:	1✔
89	"""Resets the counter to zero unless events are disabled."""
90	if config.DISABLE_EVENTS:	1✔
91	return	×
92
93	with self._mutex:	1✔
94	self._count = 0	1✔
95
96
97	class Counter(Metric, ThreadSafeCounter):	1✔
98	"""
99	A thread-safe, unlabeled counter for tracking the total number of occurrences of a specific event.
100	This class is intended for metrics that do not require differentiation across dimensions.
101	For use cases where metrics need to be grouped or segmented by labels, use `LabeledCounter` instead.
102	"""
103
104	_type: str	1✔
105
106	def __init__(self, namespace: str, name: str, schema_version: int = 1):	1✔
107	Metric.__init__(self, namespace=namespace, name=name, schema_version=schema_version)	1✔
108	ThreadSafeCounter.__init__(self)	1✔
109
110	self._type = "counter"	1✔
111	MetricRegistry().register(self)	1✔
112
113	def collect(self) -> list[CounterPayload]:	1✔
114	"""Collects the metric unless events are disabled."""
115	if config.DISABLE_EVENTS:	1✔
116	return []	1✔
117
118	if self._count == 0:	1✔
119	# Return an empty list if the count is 0, as there are no metrics to send to the analytics backend.
120	return []	1✔
121
122	return [	1✔
123	CounterPayload(
124	namespace=self._namespace,
125	name=self.name,
126	value=self._count,
127	type=self._type,
128	schema_version=self._schema_version,
129	)
130	]
131
132
133	class LabeledCounter(Metric):	1✔
134	"""
135	A thread-safe counter for tracking occurrences of an event across multiple combinations of label values.
136	It enables fine-grained metric collection and analysis, with each unique label set stored and counted independently.
137	Use this class when you need dimensional insights into event occurrences.
138	For simpler, unlabeled use cases, see the `Counter` class.
139	"""
140
141	_type: str	1✔
142	_labels: list[str]	1✔
143	_label_values: tuple[str \| float \| None, ...]	1✔
144	_counters_by_label_values: defaultdict[tuple[str \| float \| None, ...], ThreadSafeCounter]	1✔
145
146	def __init__(self, namespace: str, name: str, labels: list[str], schema_version: int = 1):	1✔
147	super().__init__(namespace=namespace, name=name, schema_version=schema_version)	1✔
148
149	if not labels:	1✔
150	raise ValueError("At least one label is required; the labels list cannot be empty.")	1✔
151
152	if any(not label for label in labels):	1✔
UNCOV 153	raise ValueError("Labels must be non-empty strings.")	×
154
155	if len(labels) > 6:	1✔
156	raise ValueError("Too many labels: counters allow a maximum of 6.")	1✔
157
158	self._type = "counter"	1✔
159	self._labels = labels	1✔
160	self._counters_by_label_values = defaultdict(ThreadSafeCounter)	1✔
161	MetricRegistry().register(self)	1✔
162
163	def labels(self, **kwargs: str \| float \| None) -> ThreadSafeCounter:	1✔
164	"""
165	Create a scoped counter instance with specific label values.
166
167	This method assigns values to the predefined labels of a labeled counter and returns
168	a ThreadSafeCounter object that allows tracking metrics for that specific
169	combination of label values.
170
171	:raises ValueError:
172	- If the set of keys provided labels does not match the expected set of labels.
173	"""
174	if set(self._labels) != set(kwargs.keys()):	1✔
175	raise ValueError(f"Expected labels {self._labels}, got {list(kwargs.keys())}")	1✔
176
177	_label_values = tuple(kwargs[label] for label in self._labels)	1✔
178
179	return self._counters_by_label_values[_label_values]	1✔
180
181	def collect(self) -> list[LabeledCounterPayload]:	1✔
182	if config.DISABLE_EVENTS:	1✔
183	return []	1✔
184
185	payload = []	1✔
186	num_labels = len(self._labels)	1✔
187
188	for label_values, counter in self._counters_by_label_values.items():	1✔
189	if counter.count == 0:	1✔
190	continue # Skip items with a count of 0, as they should not be sent to the analytics backend.	1✔
191
192	if len(label_values) != num_labels:	1✔
UNCOV 193	raise ValueError(	×
194	f"Label count mismatch: expected {num_labels} labels {self._labels}, "
195	f"but got {len(label_values)} values {label_values}."
196	)
197
198	# Create labels dictionary
199	labels_dict = dict(zip(self._labels, label_values, strict=False))	1✔
200
201	payload.append(	1✔
202	LabeledCounterPayload(
203	namespace=self._namespace,
204	name=self.name,
205	value=counter.count,
206	type=self._type,
207	schema_version=self._schema_version,
208	labels=labels_dict,
209	)
210	)
211
212	return payload	1✔

localstack / localstack / 18505123992

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous