ed350f2e-b6c3-4bef-a7fa-04255aec4056

Committed 03 Jun 2025 05:34PM UTC coverage: 86.768% (+0.04%) from 86.729%

Build # ed350f2e-b6c3-4bef-a7fa-04255aec4056

Build Type

push

circleci

Committed by

web-flow

Commit Message

CloudFormation v2 Engine: Base Support for Fn::Base64 (#12700)

Coverage Stats

20 of 22 new or added lines in 3 files covered. (90.91%)

185 existing lines in 14 files now uncovered.

65077 of 75001 relevant lines covered (86.77%)

0.87 hits per line

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

93.92

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

from __future__ import annotations

import datetime
import logging
import threading
from abc import ABC, abstractmethod
from collections import defaultdict
from dataclasses import dataclass
from typing import Any, Optional, Union, overload

from localstack import config
from localstack.runtime import hooks
from localstack.utils.analytics import get_session_id
from localstack.utils.analytics.events import Event, EventMetadata
from localstack.utils.analytics.publisher import AnalyticsClientPublisher

LOG = logging.getLogger(__name__)


@dataclass(frozen=True)
class MetricRegistryKey:
    namespace: str
    name: str


@dataclass(frozen=True)
class CounterPayload:
    """An immutable snapshot of a counter metric at the time of collection."""

    namespace: str
    name: str
    value: int
    type: str
    labels: Optional[dict[str, Union[str, float]]] = None

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

        if self.labels:
            # Convert labels to the expected format (label_1, label_1_value, etc.)
            for i, (label_name, label_value) in enumerate(self.labels.items(), 1):
                result[f"label_{i}"] = label_name
                result[f"label_{i}_value"] = label_value

        return result


@dataclass
class MetricPayload:
    """
    Stores all metric payloads collected during the execution of the LocalStack emulator.
    Currently, supports only counter-type metrics, but designed to accommodate other types in the future.
    """

    _payload: list[CounterPayload]  # support for other metric types may be added in the future.

    @property
    def payload(self) -> list[CounterPayload]:
        return self._payload

    def __init__(self, payload: list[CounterPayload]):
        self._payload = payload

    def as_dict(self) -> dict[str, list[dict[str, Any]]]:
        return {"metrics": [payload.as_dict() for payload in self._payload]}


class MetricRegistry:
    """
    A Singleton class responsible for managing all registered metrics.
    Provides methods for retrieving and collecting metrics.
    """

    _instance: "MetricRegistry" = None
    _mutex: threading.Lock = threading.Lock()

    def __new__(cls):
        # avoid locking if the instance already exist
        if cls._instance is None:
            with cls._mutex:
                # Prevents race conditions when multiple threads enter the first check simultaneously
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if not hasattr(self, "_registry"):
            self._registry = dict()

    @property
    def registry(self) -> dict[MetricRegistryKey, "Metric"]:
        return self._registry

    def register(self, metric: Metric) -> None:
        """
        Registers a new metric.

        :param metric: The metric instance to register.
        :type metric: Metric
        :raises TypeError: If the provided metric is not an instance of `Metric`.
        :raises ValueError: If a metric with the same name already exists.
        """
        if not isinstance(metric, Metric):
            raise TypeError("Only subclasses of `Metric` can be registered.")

        if not metric.namespace:
            raise ValueError("Metric 'namespace' must be defined and non-empty.")

        registry_unique_key = MetricRegistryKey(namespace=metric.namespace, name=metric.name)
        if registry_unique_key in self._registry:
            raise ValueError(
                f"A metric named '{metric.name}' already exists in the '{metric.namespace}' namespace"
            )

        self._registry[registry_unique_key] = metric

    def collect(self) -> MetricPayload:
        """
        Collects all registered metrics.
        """
        payload = [
            metric
            for metric_instance in self._registry.values()
            for metric in metric_instance.collect()
        ]

        return MetricPayload(payload=payload)


class Metric(ABC):
    """
    Base class for all metrics (e.g., Counter, Gauge).

    Each subclass must implement the `collect()` method.
    """

    _namespace: str
    _name: str

    def __init__(self, namespace: str, name: str):
        if not namespace or namespace.strip() == "":
            raise ValueError("Namespace must be non-empty string.")
        self._namespace = namespace

        if not name or name.strip() == "":
            raise ValueError("Metric name must be non-empty string.")
        self._name = name

    @property
    def namespace(self) -> str:
        return self._namespace

    @property
    def name(self) -> str:
        return self._name

    @abstractmethod
    def collect(
        self,
    ) -> list[CounterPayload]:  # support for other metric types may be added in the future.
        """
        Collects and returns metric data. Subclasses must implement this to return collected metric data.
        """
        pass


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

    _mutex: threading.Lock
    _count: int

    def __init__(self):
        super(BaseCounter, self).__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 CounterMetric(Metric, BaseCounter):
    """
    A thread-safe counter for tracking occurrences of an event without labels.
    This class should not be instantiated directly, use the Counter class instead.
    """

    _type: str

    def __init__(self, namespace: str, name: str):
        Metric.__init__(self, namespace=namespace, name=name)
        BaseCounter.__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 list()

        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 list()

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


class LabeledCounterMetric(Metric):
    """
    A labeled counter that tracks occurrences of an event across different label combinations.
    This class should not be instantiated directly, use the Counter class instead.
    """

    _type: str
    _unit: str
    _labels: list[str]
    _label_values: tuple[Optional[Union[str, float]], ...]
    _counters_by_label_values: defaultdict[tuple[Optional[Union[str, float]], ...], BaseCounter]

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

        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(BaseCounter)
        MetricRegistry().register(self)

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

        This method assigns values to the predefined labels of a labeled counter and returns
        a BaseCounter 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[CounterPayload]:
        if config.DISABLE_EVENTS:
            return list()

        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 = {
                label_name: label_value
                for label_name, label_value in zip(self._labels, label_values)
            }

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

        return payload


class Counter:
    """
    A factory class for creating counter instances.

    This class provides a flexible way to create either a simple counter
    (`CounterMetric`) or a labeled counter (`LabeledCounterMetric`) based on
    whether labels are provided.
    """

    @overload
    def __new__(cls, namespace: str, name: str) -> CounterMetric:
        return CounterMetric(namespace=namespace, name=name)

    @overload
    def __new__(cls, namespace: str, name: str, labels: list[str]) -> LabeledCounterMetric:
        return LabeledCounterMetric(namespace=namespace, name=name, labels=labels)

    def __new__(
        cls, namespace: str, name: str, labels: Optional[list[str]] = None
    ) -> Union[CounterMetric, LabeledCounterMetric]:
        if labels is not None:
            return LabeledCounterMetric(namespace=namespace, name=name, labels=labels)
        return CounterMetric(namespace=namespace, name=name)


@hooks.on_infra_shutdown()
def publish_metrics() -> None:
    """
    Collects all the registered metrics and immediately sends them to the analytics service.
    Skips execution if event tracking is disabled (`config.DISABLE_EVENTS`).

    This function is automatically triggered on infrastructure shutdown.
    """
    if config.DISABLE_EVENTS:
        return

    collected_metrics = MetricRegistry().collect()
    if not collected_metrics.payload:  # Skip publishing if no metrics remain after filtering
        return

    metadata = EventMetadata(
        session_id=get_session_id(),
        client_time=str(datetime.datetime.now()),
    )

    if collected_metrics:
        publisher = AnalyticsClientPublisher()
        publisher.publish(
            [Event(name="ls_metrics", metadata=metadata, payload=collected_metrics.as_dict())]
        )

1	from __future__ import annotations	1✔
2
3	import datetime	1✔
4	import logging	1✔
5	import threading	1✔
6	from abc import ABC, abstractmethod	1✔
7	from collections import defaultdict	1✔
8	from dataclasses import dataclass	1✔
9	from typing import Any, Optional, Union, overload	1✔
10
11	from localstack import config	1✔
12	from localstack.runtime import hooks	1✔
13	from localstack.utils.analytics import get_session_id	1✔
14	from localstack.utils.analytics.events import Event, EventMetadata	1✔
15	from localstack.utils.analytics.publisher import AnalyticsClientPublisher	1✔
16
17	LOG = logging.getLogger(__name__)	1✔
18
19
20	@dataclass(frozen=True)	1✔
21	class MetricRegistryKey:	1✔
22	namespace: str	1✔
23	name: str	1✔
24
25
26	@dataclass(frozen=True)	1✔
27	class CounterPayload:	1✔
28	"""An immutable snapshot of a counter metric at the time of collection."""
29
30	namespace: str	1✔
31	name: str	1✔
32	value: int	1✔
33	type: str	1✔
34	labels: Optional[dict[str, Union[str, float]]] = None	1✔
35
36	def as_dict(self) -> dict[str, Any]:	1✔
37	result = {	1✔
38	"namespace": self.namespace,
39	"name": self.name,
40	"value": self.value,
41	"type": self.type,
42	}
43
44	if self.labels:	1✔
45	# Convert labels to the expected format (label_1, label_1_value, etc.)
46	for i, (label_name, label_value) in enumerate(self.labels.items(), 1):	1✔
47	result[f"label_{i}"] = label_name	1✔
48	result[f"label_{i}_value"] = label_value	1✔
49
50	return result	1✔
51
52
53	@dataclass	1✔
54	class MetricPayload:	1✔
55	"""
56	Stores all metric payloads collected during the execution of the LocalStack emulator.
57	Currently, supports only counter-type metrics, but designed to accommodate other types in the future.
58	"""
59
60	_payload: list[CounterPayload] # support for other metric types may be added in the future.	1✔
61
62	@property	1✔
63	def payload(self) -> list[CounterPayload]:	1✔
64	return self._payload	1✔
65
66	def __init__(self, payload: list[CounterPayload]):	1✔
67	self._payload = payload	1✔
68
69	def as_dict(self) -> dict[str, list[dict[str, Any]]]:	1✔
70	return {"metrics": [payload.as_dict() for payload in self._payload]}	1✔
71
72
73	class MetricRegistry:	1✔
74	"""
75	A Singleton class responsible for managing all registered metrics.
76	Provides methods for retrieving and collecting metrics.
77	"""
78
79	_instance: "MetricRegistry" = None	1✔
80	_mutex: threading.Lock = threading.Lock()	1✔
81
82	def __new__(cls):	1✔
83	# avoid locking if the instance already exist
84	if cls._instance is None:	1✔
85	with cls._mutex:	1✔
86	# Prevents race conditions when multiple threads enter the first check simultaneously
87	if cls._instance is None:	1✔
88	cls._instance = super().__new__(cls)	1✔
89	return cls._instance	1✔
90
91	def __init__(self):	1✔
92	if not hasattr(self, "_registry"):	1✔
93	self._registry = dict()	1✔
94
95	@property	1✔
96	def registry(self) -> dict[MetricRegistryKey, "Metric"]:	1✔
97	return self._registry	1✔
98
99	def register(self, metric: Metric) -> None:	1✔
100	"""
101	Registers a new metric.
102
103	:param metric: The metric instance to register.
104	:type metric: Metric
105	:raises TypeError: If the provided metric is not an instance of `Metric`.
106	:raises ValueError: If a metric with the same name already exists.
107	"""
108	if not isinstance(metric, Metric):	1✔
UNCOV 109	raise TypeError("Only subclasses of `Metric` can be registered.")	×
110
111	if not metric.namespace:	1✔
UNCOV 112	raise ValueError("Metric 'namespace' must be defined and non-empty.")	×
113
114	registry_unique_key = MetricRegistryKey(namespace=metric.namespace, name=metric.name)	1✔
115	if registry_unique_key in self._registry:	1✔
116	raise ValueError(	1✔
117	f"A metric named '{metric.name}' already exists in the '{metric.namespace}' namespace"
118	)
119
120	self._registry[registry_unique_key] = metric	1✔
121
122	def collect(self) -> MetricPayload:	1✔
123	"""
124	Collects all registered metrics.
125	"""
126	payload = [	1✔
127	metric
128	for metric_instance in self._registry.values()
129	for metric in metric_instance.collect()
130	]
131
132	return MetricPayload(payload=payload)	1✔
133
134
135	class Metric(ABC):	1✔
136	"""
137	Base class for all metrics (e.g., Counter, Gauge).
138
139	Each subclass must implement the `collect()` method.
140	"""
141
142	_namespace: str	1✔
143	_name: str	1✔
144
145	def __init__(self, namespace: str, name: str):	1✔
146	if not namespace or namespace.strip() == "":	1✔
147	raise ValueError("Namespace must be non-empty string.")	1✔
148	self._namespace = namespace	1✔
149
150	if not name or name.strip() == "":	1✔
151	raise ValueError("Metric name must be non-empty string.")	1✔
152	self._name = name	1✔
153
154	@property	1✔
155	def namespace(self) -> str:	1✔
156	return self._namespace	1✔
157
158	@property	1✔
159	def name(self) -> str:	1✔
160	return self._name	1✔
161
162	@abstractmethod	1✔
163	def collect(	1✔
164	self,
165	) -> list[CounterPayload]: # support for other metric types may be added in the future.
166	"""
167	Collects and returns metric data. Subclasses must implement this to return collected metric data.
168	"""
UNCOV 169	pass	×
170
171
172	class BaseCounter:	1✔
173	"""
174	A thread-safe counter for any kind of tracking.
175	This class should not be instantiated directly, use the Counter class instead.
176	"""
177
178	_mutex: threading.Lock	1✔
179	_count: int	1✔
180
181	def __init__(self):	1✔
182	super(BaseCounter, self).__init__()	1✔
183	self._mutex = threading.Lock()	1✔
184	self._count = 0	1✔
185
186	@property	1✔
187	def count(self) -> int:	1✔
188	return self._count	1✔
189
190	def increment(self, value: int = 1) -> None:	1✔
191	"""Increments the counter unless events are disabled."""
192	if config.DISABLE_EVENTS:	1✔
193	return	1✔
194
195	if value <= 0:	1✔
UNCOV 196	raise ValueError("Increment value must be positive.")	×
197
198	with self._mutex:	1✔
199	self._count += value	1✔
200
201	def reset(self) -> None:	1✔
202	"""Resets the counter to zero unless events are disabled."""
203	if config.DISABLE_EVENTS:	1✔
UNCOV 204	return	×
205
206	with self._mutex:	1✔
207	self._count = 0	1✔
208
209
210	class CounterMetric(Metric, BaseCounter):	1✔
211	"""
212	A thread-safe counter for tracking occurrences of an event without labels.
213	This class should not be instantiated directly, use the Counter class instead.
214	"""
215
216	_type: str	1✔
217
218	def __init__(self, namespace: str, name: str):	1✔
219	Metric.__init__(self, namespace=namespace, name=name)	1✔
220	BaseCounter.__init__(self)	1✔
221
222	self._type = "counter"	1✔
223	MetricRegistry().register(self)	1✔
224
225	def collect(self) -> list[CounterPayload]:	1✔
226	"""Collects the metric unless events are disabled."""
227	if config.DISABLE_EVENTS:	1✔
228	return list()	1✔
229
230	if self._count == 0:	1✔
231	# Return an empty list if the count is 0, as there are no metrics to send to the analytics backend.
232	return list()	1✔
233
234	return [	1✔
235	CounterPayload(
236	namespace=self._namespace, name=self.name, value=self._count, type=self._type
237	)
238	]
239
240
241	class LabeledCounterMetric(Metric):	1✔
242	"""
243	A labeled counter that tracks occurrences of an event across different label combinations.
244	This class should not be instantiated directly, use the Counter class instead.
245	"""
246
247	_type: str	1✔
248	_unit: str	1✔
249	_labels: list[str]	1✔
250	_label_values: tuple[Optional[Union[str, float]], ...]	1✔
251	_counters_by_label_values: defaultdict[tuple[Optional[Union[str, float]], ...], BaseCounter]	1✔
252
253	def __init__(self, namespace: str, name: str, labels: list[str]):	1✔
254	super(LabeledCounterMetric, self).__init__(namespace=namespace, name=name)	1✔
255
256	if not labels:	1✔
257	raise ValueError("At least one label is required; the labels list cannot be empty.")	1✔
258
259	if any(not label for label in labels):	1✔
UNCOV 260	raise ValueError("Labels must be non-empty strings.")	×
261
262	if len(labels) > 6:	1✔
263	raise ValueError("Too many labels: counters allow a maximum of 6.")	1✔
264
265	self._type = "counter"	1✔
266	self._labels = labels	1✔
267	self._counters_by_label_values = defaultdict(BaseCounter)	1✔
268	MetricRegistry().register(self)	1✔
269
270	def labels(self, **kwargs: Union[str, float, None]) -> BaseCounter:	1✔
271	"""
272	Create a scoped counter instance with specific label values.
273
274	This method assigns values to the predefined labels of a labeled counter and returns
275	a BaseCounter object that allows tracking metrics for that specific
276	combination of label values.
277
278	:raises ValueError:
279	- If the set of keys provided labels does not match the expected set of labels.
280	"""
281	if set(self._labels) != set(kwargs.keys()):	1✔
282	raise ValueError(f"Expected labels {self._labels}, got {list(kwargs.keys())}")	1✔
283
284	_label_values = tuple(kwargs[label] for label in self._labels)	1✔
285
286	return self._counters_by_label_values[_label_values]	1✔
287
288	def collect(self) -> list[CounterPayload]:	1✔
289	if config.DISABLE_EVENTS:	1✔
290	return list()	1✔
291
292	payload = []	1✔
293	num_labels = len(self._labels)	1✔
294
295	for label_values, counter in self._counters_by_label_values.items():	1✔
296	if counter.count == 0:	1✔
297	continue # Skip items with a count of 0, as they should not be sent to the analytics backend.	1✔
298
299	if len(label_values) != num_labels:	1✔
UNCOV 300	raise ValueError(	×
301	f"Label count mismatch: expected {num_labels} labels {self._labels}, "
302	f"but got {len(label_values)} values {label_values}."
303	)
304
305	# Create labels dictionary
306	labels_dict = {	1✔
307	label_name: label_value
308	for label_name, label_value in zip(self._labels, label_values)
309	}
310
311	payload.append(	1✔
312	CounterPayload(
313	namespace=self._namespace,
314	name=self.name,
315	value=counter.count,
316	type=self._type,
317	labels=labels_dict,
318	)
319	)
320
321	return payload	1✔
322
323
324	class Counter:	1✔
325	"""
326	A factory class for creating counter instances.
327
328	This class provides a flexible way to create either a simple counter
329	(`CounterMetric`) or a labeled counter (`LabeledCounterMetric`) based on
330	whether labels are provided.
331	"""
332
333	@overload	1✔
334	def __new__(cls, namespace: str, name: str) -> CounterMetric:	1✔
UNCOV 335	return CounterMetric(namespace=namespace, name=name)	×
336
337	@overload	1✔
338	def __new__(cls, namespace: str, name: str, labels: list[str]) -> LabeledCounterMetric:	1✔
UNCOV 339	return LabeledCounterMetric(namespace=namespace, name=name, labels=labels)	×
340
341	def __new__(	1✔
342	cls, namespace: str, name: str, labels: Optional[list[str]] = None
343	) -> Union[CounterMetric, LabeledCounterMetric]:
344	if labels is not None:	1✔
345	return LabeledCounterMetric(namespace=namespace, name=name, labels=labels)	1✔
346	return CounterMetric(namespace=namespace, name=name)	1✔
347
348
349	@hooks.on_infra_shutdown()	1✔
350	def publish_metrics() -> None:	1✔
351	"""
352	Collects all the registered metrics and immediately sends them to the analytics service.
353	Skips execution if event tracking is disabled (`config.DISABLE_EVENTS`).
354
355	This function is automatically triggered on infrastructure shutdown.
356	"""
357	if config.DISABLE_EVENTS:	1✔
UNCOV 358	return	×
359
360	collected_metrics = MetricRegistry().collect()	1✔
361	if not collected_metrics.payload: # Skip publishing if no metrics remain after filtering	1✔
UNCOV 362	return	×
363
364	metadata = EventMetadata(	1✔
365	session_id=get_session_id(),
366	client_time=str(datetime.datetime.now()),
367	)
368
369	if collected_metrics:	1✔
370	publisher = AnalyticsClientPublisher()	1✔
371	publisher.publish(	1✔
372	[Event(name="ls_metrics", metadata=metadata, payload=collected_metrics.as_dict())]
373	)

localstack / localstack / ed350f2e-b6c3-4bef-a7fa-04255aec4056

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