20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

82.81

/localstack-core/localstack/aws/handlers/metric_handler.py

import csv
import logging
import os
from datetime import datetime
from pathlib import Path

from localstack import config
from localstack.aws.api import RequestContext
from localstack.aws.chain import HandlerChain
from localstack.constants import ENV_INTERNAL_TEST_STORE_METRICS_PATH
from localstack.http import Response
from localstack.utils.strings import short_uid

LOG = logging.getLogger(__name__)


class MetricHandlerItem:
    """
    MetricHandlerItem to reference and update requests by the MetricHandler
    """

    request_id: str
    request_context: RequestContext
    parameters_after_parse: list[str] | None

    def __init__(self, request_contex: RequestContext) -> None:
        super().__init__()
        self.request_id = str(hash(request_contex))
        self.request_context = request_contex
        self.parameters_after_parse = None


class Metric:
    """
    Data object to store relevant information for a metric entry in the raw-data collection (csv)
    """

    service: str
    operation: str
    headers: str
    parameters: str
    status_code: int
    response_code: str | None
    exception: str
    origin: str
    xfail: bool
    aws_validated: bool
    snapshot: bool
    node_id: str

    RAW_DATA_HEADER = [
        "service",
        "operation",
        "request_headers",
        "parameters",
        "response_code",
        "response_data",
        "exception",
        "origin",
        "test_node_id",
        "xfail",
        "aws_validated",
        "snapshot",
        "snapshot_skipped_paths",
    ]

    def __init__(
        self,
        service: str,
        operation: str,
        headers: str,
        parameters: str,
        response_code: int,
        response_data: str,
        exception: str,
        origin: str,
        node_id: str = "",
        xfail: bool = False,
        aws_validated: bool = False,
        snapshot: bool = False,
        snapshot_skipped_paths: str = "",
    ) -> None:
        self.service = service
        self.operation = operation
        self.headers = headers
        self.parameters = parameters
        self.response_code = response_code
        self.response_data = response_data
        self.exception = exception
        self.origin = origin
        self.node_id = node_id
        self.xfail = xfail
        self.aws_validated = aws_validated
        self.snapshot = snapshot
        self.snapshot_skipped_paths = snapshot_skipped_paths

    def __iter__(self):
        return iter(
            [
                self.service,
                self.operation,
                self.headers,
                self.parameters,
                self.response_code,
                self.response_data,
                self.exception,
                self.origin,
                self.node_id,
                self.xfail,
                self.aws_validated,
                self.snapshot,
                self.snapshot_skipped_paths,
            ]
        )

    def __eq__(self, other):
        # ignore header in comparison, because timestamp will be different
        if self.service != other.service:
            return False
        if self.operation != other.operation:
            return False
        if self.parameters != other.parameters:
            return False
        if self.response_code != other.response_code:
            return False
        if self.response_data != other.response_data:
            return False
        if self.exception != other.exception:
            return False
        if self.origin != other.origin:
            return False
        if self.xfail != other.xfail:
            return False
        if self.aws_validated != other.aws_validated:
            return False
        if self.node_id != other.node_id:
            return False
        return True


class MetricHandler:
    metric_data: list[Metric] = []

    def __init__(self) -> None:
        self.metrics_handler_items = {}
        self.local_filename = None

        if self.should_store_metric_locally():
            self.local_filename = self.create_local_file()

    @staticmethod
    def should_store_metric_locally() -> bool:
        return config.is_collect_metrics_mode() and config.store_test_metrics_in_local_filesystem()

    @staticmethod
    def create_local_file():
        folder = Path(
            os.environ.get(ENV_INTERNAL_TEST_STORE_METRICS_PATH, "/tmp/localstack-metrics")
        )
        if not folder.exists():
            folder.mkdir(parents=True, exist_ok=True)
        LOG.debug("Metric reports will be stored in %s", folder)
        filename = (
            folder
            / f"metric-report-raw-data-{datetime.utcnow().strftime('%Y-%m-%d__%H_%M_%S')}-{short_uid()}.csv"
        )
        with open(filename, "w") as fd:
            LOG.debug("Creating new metric data file %s", filename)
            writer = csv.writer(fd)
            writer.writerow(Metric.RAW_DATA_HEADER)
        return filename

    def create_metric_handler_item(
        self, chain: HandlerChain, context: RequestContext, response: Response
    ):
        if not config.is_collect_metrics_mode():
            return
        item = MetricHandlerItem(context)
        self.metrics_handler_items[context] = item

    def _get_metric_handler_item_for_context(self, context: RequestContext) -> MetricHandlerItem:
        return self.metrics_handler_items[context]

    def record_parsed_request(
        self, chain: HandlerChain, context: RequestContext, response: Response
    ):
        if not config.is_collect_metrics_mode():
            return
        item = self._get_metric_handler_item_for_context(context)
        item.parameters_after_parse = (
            list(context.service_request.keys()) if context.service_request else []
        )

    def record_exception(
        self, chain: HandlerChain, exception: Exception, context: RequestContext, response: Response
    ):
        if not config.is_collect_metrics_mode():
            return
        item = self._get_metric_handler_item_for_context(context)
        item.caught_exception_name = exception.__class__.__name__

    def update_metric_collection(
        self, chain: HandlerChain, context: RequestContext, response: Response
    ):
        if not config.is_collect_metrics_mode() or not context.service_operation:
            return

        item = self._get_metric_handler_item_for_context(context)

        # parameters might get changed when dispatched to the service - we use the params stored in
        # parameters_after_parse
        parameters = ",".join(item.parameters_after_parse or [])

        response_data = response.data.decode("utf-8") if response.status_code >= 300 else ""
        metric = Metric(
            service=context.service_operation.service,
            operation=context.service_operation.operation,
            headers=context.request.headers,
            parameters=parameters,
            response_code=response.status_code,
            response_data=response_data,
            exception=context.service_exception.__class__.__name__
            if context.service_exception
            else "",
            origin="internal" if context.is_internal_call else "external",
        )
        # refrain from adding duplicates
        if metric not in MetricHandler.metric_data:
            self.append_metric(metric)

        # cleanup
        del self.metrics_handler_items[context]

    def append_metric(self, metric: Metric):
        if self.should_store_metric_locally():
            with open(self.local_filename, "a") as fd:
                writer = csv.writer(fd)
                writer.writerow(metric)
        else:
            MetricHandler.metric_data.append(metric)

1	import csv	1✔
2	import logging	1✔
3	import os	1✔
4	from datetime import datetime	1✔
5	from pathlib import Path	1✔
6
7	from localstack import config	1✔
8	from localstack.aws.api import RequestContext	1✔
9	from localstack.aws.chain import HandlerChain	1✔
10	from localstack.constants import ENV_INTERNAL_TEST_STORE_METRICS_PATH	1✔
11	from localstack.http import Response	1✔
12	from localstack.utils.strings import short_uid	1✔
13
14	LOG = logging.getLogger(__name__)	1✔
15
16
17	class MetricHandlerItem:	1✔
18	"""
19	MetricHandlerItem to reference and update requests by the MetricHandler
20	"""
21
22	request_id: str	1✔
23	request_context: RequestContext	1✔
24	parameters_after_parse: list[str] \| None	1✔
25
26	def __init__(self, request_contex: RequestContext) -> None:	1✔
27	super().__init__()	1✔
28	self.request_id = str(hash(request_contex))	1✔
29	self.request_context = request_contex	1✔
30	self.parameters_after_parse = None	1✔
31
32
33	class Metric:	1✔
34	"""
35	Data object to store relevant information for a metric entry in the raw-data collection (csv)
36	"""
37
38	service: str	1✔
39	operation: str	1✔
40	headers: str	1✔
41	parameters: str	1✔
42	status_code: int	1✔
43	response_code: str \| None	1✔
44	exception: str	1✔
45	origin: str	1✔
46	xfail: bool	1✔
47	aws_validated: bool	1✔
48	snapshot: bool	1✔
49	node_id: str	1✔
50
51	RAW_DATA_HEADER = [	1✔
52	"service",
53	"operation",
54	"request_headers",
55	"parameters",
56	"response_code",
57	"response_data",
58	"exception",
59	"origin",
60	"test_node_id",
61	"xfail",
62	"aws_validated",
63	"snapshot",
64	"snapshot_skipped_paths",
65	]
66
67	def __init__(	1✔
68	self,
69	service: str,
70	operation: str,
71	headers: str,
72	parameters: str,
73	response_code: int,
74	response_data: str,
75	exception: str,
76	origin: str,
77	node_id: str = "",
78	xfail: bool = False,
79	aws_validated: bool = False,
80	snapshot: bool = False,
81	snapshot_skipped_paths: str = "",
82	) -> None:
83	self.service = service	1✔
84	self.operation = operation	1✔
85	self.headers = headers	1✔
86	self.parameters = parameters	1✔
87	self.response_code = response_code	1✔
88	self.response_data = response_data	1✔
89	self.exception = exception	1✔
90	self.origin = origin	1✔
91	self.node_id = node_id	1✔
92	self.xfail = xfail	1✔
93	self.aws_validated = aws_validated	1✔
94	self.snapshot = snapshot	1✔
95	self.snapshot_skipped_paths = snapshot_skipped_paths	1✔
96
97	def __iter__(self):	1✔
98	return iter(	1✔
99	[
100	self.service,
101	self.operation,
102	self.headers,
103	self.parameters,
104	self.response_code,
105	self.response_data,
106	self.exception,
107	self.origin,
108	self.node_id,
109	self.xfail,
110	self.aws_validated,
111	self.snapshot,
112	self.snapshot_skipped_paths,
113	]
114	)
115
116	def __eq__(self, other):	1✔
117	# ignore header in comparison, because timestamp will be different
118	if self.service != other.service:	1✔
119	return False	1✔
120	if self.operation != other.operation:	1✔
121	return False	1✔
122	if self.parameters != other.parameters:	1✔
123	return False	1✔
124	if self.response_code != other.response_code:	1✔
125	return False	1✔
126	if self.response_data != other.response_data:	1✔
127	return False	1✔
128	if self.exception != other.exception:	1✔
129	return False	1✔
130	if self.origin != other.origin:	1✔
131	return False	1✔
132	if self.xfail != other.xfail:	1✔
133	return False	×
134	if self.aws_validated != other.aws_validated:	1✔
135	return False	1✔
136	if self.node_id != other.node_id:	1✔
137	return False	×
138	return True	1✔
139
140
141	class MetricHandler:	1✔
142	metric_data: list[Metric] = []	1✔
143
144	def __init__(self) -> None:	1✔
145	self.metrics_handler_items = {}	1✔
146	self.local_filename = None	1✔
147
148	if self.should_store_metric_locally():	1✔
149	self.local_filename = self.create_local_file()	×
150
151	@staticmethod	1✔
152	def should_store_metric_locally() -> bool:	1✔
153	return config.is_collect_metrics_mode() and config.store_test_metrics_in_local_filesystem()	1✔
154
155	@staticmethod	1✔
156	def create_local_file():	1✔
157	folder = Path(	×
158	os.environ.get(ENV_INTERNAL_TEST_STORE_METRICS_PATH, "/tmp/localstack-metrics")
159	)
160	if not folder.exists():	×
161	folder.mkdir(parents=True, exist_ok=True)	×
162	LOG.debug("Metric reports will be stored in %s", folder)	×
163	filename = (	×
164	folder
165	/ f"metric-report-raw-data-{datetime.utcnow().strftime('%Y-%m-%d__%H_%M_%S')}-{short_uid()}.csv"
166	)
167	with open(filename, "w") as fd:	×
168	LOG.debug("Creating new metric data file %s", filename)	×
169	writer = csv.writer(fd)	×
170	writer.writerow(Metric.RAW_DATA_HEADER)	×
171	return filename	×
172
173	def create_metric_handler_item(	1✔
174	self, chain: HandlerChain, context: RequestContext, response: Response
175	):
176	if not config.is_collect_metrics_mode():	1✔
177	return	×
178	item = MetricHandlerItem(context)	1✔
179	self.metrics_handler_items[context] = item	1✔
180
181	def _get_metric_handler_item_for_context(self, context: RequestContext) -> MetricHandlerItem:	1✔
182	return self.metrics_handler_items[context]	1✔
183
184	def record_parsed_request(	1✔
185	self, chain: HandlerChain, context: RequestContext, response: Response
186	):
187	if not config.is_collect_metrics_mode():	1✔
188	return	×
189	item = self._get_metric_handler_item_for_context(context)	1✔
190	item.parameters_after_parse = (	1✔
191	list(context.service_request.keys()) if context.service_request else []
192	)
193
194	def record_exception(	1✔
195	self, chain: HandlerChain, exception: Exception, context: RequestContext, response: Response
196	):
197	if not config.is_collect_metrics_mode():	×
198	return	×
199	item = self._get_metric_handler_item_for_context(context)	×
200	item.caught_exception_name = exception.__class__.__name__	×
201
202	def update_metric_collection(	1✔
203	self, chain: HandlerChain, context: RequestContext, response: Response
204	):
205	if not config.is_collect_metrics_mode() or not context.service_operation:	1✔
206	return	1✔
207
208	item = self._get_metric_handler_item_for_context(context)	1✔
209
210	# parameters might get changed when dispatched to the service - we use the params stored in
211	# parameters_after_parse
212	parameters = ",".join(item.parameters_after_parse or [])	1✔
213
214	response_data = response.data.decode("utf-8") if response.status_code >= 300 else ""	1✔
215	metric = Metric(	1✔
216	service=context.service_operation.service,
217	operation=context.service_operation.operation,
218	headers=context.request.headers,
219	parameters=parameters,
220	response_code=response.status_code,
221	response_data=response_data,
222	exception=context.service_exception.__class__.__name__
223	if context.service_exception
224	else "",
225	origin="internal" if context.is_internal_call else "external",
226	)
227	# refrain from adding duplicates
228	if metric not in MetricHandler.metric_data:	1✔
229	self.append_metric(metric)	1✔
230
231	# cleanup
232	del self.metrics_handler_items[context]	1✔
233
234	def append_metric(self, metric: Metric):	1✔
235	if self.should_store_metric_locally():	1✔
236	with open(self.local_filename, "a") as fd:	×
237	writer = csv.writer(fd)	×
238	writer.writerow(metric)	×
239	else:
240	MetricHandler.metric_data.append(metric)	1✔

localstack / localstack / 20565403496

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