16820655284

Committed 07 Aug 2025 05:03PM UTC coverage: 86.841% (-0.05%) from 86.892%

Build # 16820655284

Build Type

push

github

Committed by

web-flow

Commit Message

CFNV2: support CDK bootstrap and deployment (#12967)

Coverage Stats

32 of 38 new or added lines in 5 files covered. (84.21%)

2013 existing lines in 125 files now uncovered.

66606 of 76699 relevant lines covered (86.84%)

0.87 hits per line

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

33.15

/localstack-core/localstack/http/trace.py

import dataclasses
import inspect
import logging
import time
from collections.abc import Callable
from typing import Any

from rolo import Response
from rolo.gateway import ExceptionHandler, Handler, HandlerChain, RequestContext
from werkzeug.datastructures import Headers

from localstack.utils.patch import Patch, Patches

LOG = logging.getLogger(__name__)


class Action:
    """
    Encapsulates something that the handler performed on the request context, request, or response objects.
    """

    name: str

    def __init__(self, name: str):
        self.name = name

    def __repr__(self):
        return self.name


class SetAttributeAction(Action):
    """
    The handler set an attribute of the request context or something else.
    """

    key: str
    value: Any | None

    def __init__(self, key: str, value: Any | None = None):
        super().__init__("set")
        self.key = key
        self.value = value

    def __repr__(self):
        if self.value is None:
            return f"set {self.key}"
        return f"set {self.key} = {self.value!r}"


class ModifyHeadersAction(Action):
    """
    The handler modified headers in some way, either adding, updating, or removing headers.
    """

    def __init__(self, name: str, before: Headers, after: Headers):
        super().__init__(name)
        self.before = before
        self.after = after

    @property
    def header_actions(self) -> list[Action]:
        after = self.after
        before = self.before

        actions = []

        headers_set = dict(set(after.items()) - set(before.items()))
        headers_removed = {k: v for k, v in before.items() if k not in after}

        for k, v in headers_set.items():
            actions.append(Action(f"set '{k}: {v}'"))
        for k, v in headers_removed.items():
            actions.append(Action(f"del '{k}: {v}'"))

        return actions


@dataclasses.dataclass
class HandlerTrace:
    handler: Handler
    """The handler"""
    duration_ms: float
    """The runtime duration of the handler in milliseconds"""
    actions: list[Action]
    """The actions the handler chain performed"""

    @property
    def handler_module(self):
        return self.handler.__module__

    @property
    def handler_name(self):
        if inspect.isfunction(self.handler):
            return self.handler.__name__
        else:
            return self.handler.__class__.__name__


def _log_method_call(name: str, actions: list[Action]):
    """Creates a wrapper around the original method `_fn`. It appends an action to the `actions`
    list indicating that the function was called and then returns the original function."""

    def _proxy(self, _fn, *args, **kwargs):
        actions.append(Action(f"call {name}"))
        return _fn(*args, **kwargs)

    return _proxy


class TracingHandlerBase:
    """
    This class is a Handler that records a trace of the execution of another request handler. It has two
    attributes: `trace`, which stores the tracing information, and `delegate`, which is the handler or
    exception handler that will be traced.
    """

    trace: HandlerTrace | None
    delegate: Handler | ExceptionHandler

    def __init__(self, delegate: Handler | ExceptionHandler):
        self.trace = None
        self.delegate = delegate

    def do_trace_call(
        self, fn: Callable, chain: HandlerChain, context: RequestContext, response: Response
    ):
        """
        Wraps the function call with the tracing functionality and records a HandlerTrace.

        The method determines changes made by the request handler to specific aspects of the request.
        Changes made to the request context and the response headers/status by the request handler are then
        examined, and appropriate actions are added to the `actions` list of the trace.

        :param fn: which is the function to be traced, which is the request/response/exception handler
        :param chain: the handler chain
        :param context: the request context
        :param response: the response object
        """
        then = time.perf_counter()

        actions = []

        prev_context = dict(context.__dict__)
        prev_stopped = chain.stopped
        prev_request_identity = id(context.request)
        prev_terminated = chain.terminated
        prev_request_headers = context.request.headers.copy()
        prev_response_headers = response.headers.copy()
        prev_response_status = response.status_code

        # add patches to log invocations or certain functions
        patches = Patches(
            [
                Patch.function(
                    context.request.get_data,
                    _log_method_call("request.get_data", actions),
                ),
                Patch.function(
                    context.request._load_form_data,
                    _log_method_call("request._load_form_data", actions),
                ),
                Patch.function(
                    response.get_data,
                    _log_method_call("response.get_data", actions),
                ),
            ]
        )
        patches.apply()

        try:
            return fn()
        finally:
            now = time.perf_counter()
            # determine some basic things the handler changed in the context
            patches.undo()

            # chain
            if chain.stopped and not prev_stopped:
                actions.append(Action("stop chain"))
            if chain.terminated and not prev_terminated:
                actions.append(Action("terminate chain"))

            # detect when attributes are set in the request contex
            context_args = dict(context.__dict__)
            context_args.pop("request", None)  # request is handled separately

            for k, v in context_args.items():
                if not v:
                    continue
                if prev_context.get(k):
                    # TODO: we could introduce "ModifyAttributeAction(k,v)" with an additional check
                    #  ``if v != prev_context.get(k)``
                    continue
                actions.append(SetAttributeAction(k, v))

            # request
            if id(context.request) != prev_request_identity:
                actions.append(Action("replaced request object"))

            # response
            if response.status_code != prev_response_status:
                actions.append(SetAttributeAction("response stats_code", response.status_code))
            if context.request.headers != prev_request_headers:
                actions.append(
                    ModifyHeadersAction(
                        "modify request headers",
                        prev_request_headers,
                        context.request.headers.copy(),
                    )
                )
            if response.headers != prev_response_headers:
                actions.append(
                    ModifyHeadersAction(
                        "modify response headers", prev_response_headers, response.headers.copy()
                    )
                )

            self.trace = HandlerTrace(
                handler=self.delegate, duration_ms=(now - then) * 1000, actions=actions
            )


class TracingHandler(TracingHandlerBase):
    delegate: Handler

    def __init__(self, delegate: Handler):
        super().__init__(delegate)

    def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
        def _call():
            return self.delegate(chain, context, response)

        return self.do_trace_call(_call, chain, context, response)


class TracingExceptionHandler(TracingHandlerBase):
    delegate: ExceptionHandler

    def __init__(self, delegate: ExceptionHandler):
        super().__init__(delegate)

    def __call__(
        self, chain: HandlerChain, exception: Exception, context: RequestContext, response: Response
    ):
        def _call():
            return self.delegate(chain, exception, context, response)

        return self.do_trace_call(_call, chain, context, response)


class TracingHandlerChain(HandlerChain):
    """
    DebuggingHandlerChain - A subclass of HandlerChain for logging and tracing handlers.

    Attributes:
    - duration (float): Total time taken for handling request in milliseconds.
    - request_handler_traces (list[HandlerTrace]): List of request handler traces.
    - response_handler_traces (list[HandlerTrace]): List of response handler traces.
    - finalizer_traces (list[HandlerTrace]): List of finalizer traces.
    - exception_handler_traces (list[HandlerTrace]): List of exception handler traces.
    """

    duration: float
    request_handler_traces: list[HandlerTrace]
    response_handler_traces: list[HandlerTrace]
    finalizer_traces: list[HandlerTrace]
    exception_handler_traces: list[HandlerTrace]

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.request_handler_traces = []
        self.response_handler_traces = []
        self.finalizer_traces = []
        self.exception_handler_traces = []

    def handle(self, context: RequestContext, response: Response):
        """Overrides HandlerChain's handle method and adds tracing handler to request handlers. Logs the trace
        report with request and response details."""
        then = time.perf_counter()
        try:
            self.request_handlers = [TracingHandler(handler) for handler in self.request_handlers]
            return super().handle(context, response)
        finally:
            self.duration = (time.perf_counter() - then) * 1000
            self.request_handler_traces = [handler.trace for handler in self.request_handlers]
            self._log_report()

    def _call_response_handlers(self, response):
        self.response_handlers = [TracingHandler(handler) for handler in self.response_handlers]
        try:
            return super()._call_response_handlers(response)
        finally:
            self.response_handler_traces = [handler.trace for handler in self.response_handlers]

    def _call_finalizers(self, response):
        self.finalizers = [TracingHandler(handler) for handler in self.finalizers]
        try:
            return super()._call_response_handlers(response)
        finally:
            self.finalizer_traces = [handler.trace for handler in self.finalizers]

    def _call_exception_handlers(self, e, response):
        self.exception_handlers = [
            TracingExceptionHandler(handler) for handler in self.exception_handlers
        ]
        try:
            return super()._call_exception_handlers(e, response)
        finally:
            self.exception_handler_traces = [handler.trace for handler in self.exception_handlers]

    def _log_report(self):
        report = []
        request = self.context.request
        response = self.response

        def _append_traces(traces: list[HandlerTrace]):
            """Format and appends a list of traces to the report, and recursively append the trace's
            actions (if any)."""

            for trace in traces:
                if trace is None:
                    continue

                report.append(
                    f"{trace.handler_module:43s} {trace.handler_name:30s} {trace.duration_ms:8.2f}ms"
                )
                _append_actions(trace.actions, 46)

        def _append_actions(actions: list[Action], indent: int):
            for action in actions:
                report.append((" " * indent) + f"- {action!r}")

                if isinstance(action, ModifyHeadersAction):
                    _append_actions(action.header_actions, indent + 2)

        report.append(f"request:  {request.method} {request.url}")
        report.append(f"response: {response.status_code}")
        report.append("---- request handlers " + ("-" * 63))
        _append_traces(self.request_handler_traces)
        report.append("---- response handlers " + ("-" * 63))
        _append_traces(self.response_handler_traces)
        report.append("---- finalizers " + ("-" * 63))
        _append_traces(self.finalizer_traces)
        report.append("---- exception handlers " + ("-" * 63))
        _append_traces(self.exception_handler_traces)
        # Add a separator and total duration value to the end of the report
        report.append(f"{'=' * 68} total {self.duration:8.2f}ms")

        LOG.info("handler chain trace report:\n%s\n%s", "=" * 85, "\n".join(report))

1	import dataclasses	1✔
2	import inspect	1✔
3	import logging	1✔
4	import time	1✔
5	from collections.abc import Callable	1✔
6	from typing import Any	1✔
7
8	from rolo import Response	1✔
9	from rolo.gateway import ExceptionHandler, Handler, HandlerChain, RequestContext	1✔
10	from werkzeug.datastructures import Headers	1✔
11
12	from localstack.utils.patch import Patch, Patches	1✔
13
14	LOG = logging.getLogger(__name__)	1✔
15
16
17	class Action:	1✔
18	"""
19	Encapsulates something that the handler performed on the request context, request, or response objects.
20	"""
21
22	name: str	1✔
23
24	def __init__(self, name: str):	1✔
UNCOV 25	self.name = name	×
26
27	def __repr__(self):
28	return self.name
29
30
31	class SetAttributeAction(Action):	1✔
32	"""
33	The handler set an attribute of the request context or something else.
34	"""
35
36	key: str	1✔
37	value: Any \| None	1✔
38
39	def __init__(self, key: str, value: Any \| None = None):	1✔
40	super().__init__("set")	×
41	self.key = key	×
UNCOV 42	self.value = value	×
43
44	def __repr__(self):
45	if self.value is None:
46	return f"set {self.key}"
47	return f"set {self.key} = {self.value!r}"
48
49
50	class ModifyHeadersAction(Action):	1✔
51	"""
52	The handler modified headers in some way, either adding, updating, or removing headers.
53	"""
54
55	def __init__(self, name: str, before: Headers, after: Headers):	1✔
56	super().__init__(name)	×
57	self.before = before	×
UNCOV 58	self.after = after	×
59
60	@property	1✔
61	def header_actions(self) -> list[Action]:	1✔
62	after = self.after	×
UNCOV 63	before = self.before	×
64
UNCOV 65	actions = []	×
66
67	headers_set = dict(set(after.items()) - set(before.items()))	×
UNCOV 68	headers_removed = {k: v for k, v in before.items() if k not in after}	×
69
70	for k, v in headers_set.items():	×
71	actions.append(Action(f"set '{k}: {v}'"))	×
72	for k, v in headers_removed.items():	×
UNCOV 73	actions.append(Action(f"del '{k}: {v}'"))	×
74
UNCOV 75	return actions	×
76
77
78	@dataclasses.dataclass	1✔
79	class HandlerTrace:	1✔
80	handler: Handler	1✔
81	"""The handler"""	1✔
82	duration_ms: float	1✔
83	"""The runtime duration of the handler in milliseconds"""	1✔
84	actions: list[Action]	1✔
85	"""The actions the handler chain performed"""	1✔
86
87	@property	1✔
88	def handler_module(self):	1✔
UNCOV 89	return self.handler.__module__	×
90
91	@property	1✔
92	def handler_name(self):	1✔
93	if inspect.isfunction(self.handler):	×
UNCOV 94	return self.handler.__name__	×
95	else:
UNCOV 96	return self.handler.__class__.__name__	×
97
98
99	def _log_method_call(name: str, actions: list[Action]):	1✔
100	"""Creates a wrapper around the original method `_fn`. It appends an action to the `actions`
101	list indicating that the function was called and then returns the original function."""
102
103	def _proxy(self, _fn, args, *kwargs):	×
104	actions.append(Action(f"call {name}"))	×
UNCOV 105	return _fn(args, *kwargs)	×
106
UNCOV 107	return _proxy	×
108
109
110	class TracingHandlerBase:	1✔
111	"""
112	This class is a Handler that records a trace of the execution of another request handler. It has two
113	attributes: `trace`, which stores the tracing information, and `delegate`, which is the handler or
114	exception handler that will be traced.
115	"""
116
117	trace: HandlerTrace \| None	1✔
118	delegate: Handler \| ExceptionHandler	1✔
119
120	def __init__(self, delegate: Handler \| ExceptionHandler):	1✔
121	self.trace = None	×
UNCOV 122	self.delegate = delegate	×
123
124	def do_trace_call(	1✔
125	self, fn: Callable, chain: HandlerChain, context: RequestContext, response: Response
126	):
127	"""
128	Wraps the function call with the tracing functionality and records a HandlerTrace.
129
130	The method determines changes made by the request handler to specific aspects of the request.
131	Changes made to the request context and the response headers/status by the request handler are then
132	examined, and appropriate actions are added to the `actions` list of the trace.
133
134	:param fn: which is the function to be traced, which is the request/response/exception handler
135	:param chain: the handler chain
136	:param context: the request context
137	:param response: the response object
138	"""
UNCOV 139	then = time.perf_counter()	×
140
UNCOV 141	actions = []	×
142
143	prev_context = dict(context.__dict__)	×
144	prev_stopped = chain.stopped	×
145	prev_request_identity = id(context.request)	×
146	prev_terminated = chain.terminated	×
147	prev_request_headers = context.request.headers.copy()	×
148	prev_response_headers = response.headers.copy()	×
UNCOV 149	prev_response_status = response.status_code	×
150
151	# add patches to log invocations or certain functions
UNCOV 152	patches = Patches(	×
153	[
154	Patch.function(
155	context.request.get_data,
156	_log_method_call("request.get_data", actions),
157	),
158	Patch.function(
159	context.request._load_form_data,
160	_log_method_call("request._load_form_data", actions),
161	),
162	Patch.function(
163	response.get_data,
164	_log_method_call("response.get_data", actions),
165	),
166	]
167	)
UNCOV 168	patches.apply()	×
169
170	try:	×
UNCOV 171	return fn()	×
172	finally:
UNCOV 173	now = time.perf_counter()	×
174	# determine some basic things the handler changed in the context
UNCOV 175	patches.undo()	×
176
177	# chain
178	if chain.stopped and not prev_stopped:	×
179	actions.append(Action("stop chain"))	×
180	if chain.terminated and not prev_terminated:	×
UNCOV 181	actions.append(Action("terminate chain"))	×
182
183	# detect when attributes are set in the request contex
184	context_args = dict(context.__dict__)	×
UNCOV 185	context_args.pop("request", None) # request is handled separately	×
186
187	for k, v in context_args.items():	×
188	if not v:	×
189	continue	×
UNCOV 190	if prev_context.get(k):	×
191	# TODO: we could introduce "ModifyAttributeAction(k,v)" with an additional check
192	# ``if v != prev_context.get(k)``
193	continue	×
UNCOV 194	actions.append(SetAttributeAction(k, v))	×
195
196	# request
197	if id(context.request) != prev_request_identity:	×
UNCOV 198	actions.append(Action("replaced request object"))	×
199
200	# response
201	if response.status_code != prev_response_status:	×
202	actions.append(SetAttributeAction("response stats_code", response.status_code))	×
203	if context.request.headers != prev_request_headers:	×
UNCOV 204	actions.append(	×
205	ModifyHeadersAction(
206	"modify request headers",
207	prev_request_headers,
208	context.request.headers.copy(),
209	)
210	)
211	if response.headers != prev_response_headers:	×
UNCOV 212	actions.append(	×
213	ModifyHeadersAction(
214	"modify response headers", prev_response_headers, response.headers.copy()
215	)
216	)
217
UNCOV 218	self.trace = HandlerTrace(	×
219	handler=self.delegate, duration_ms=(now - then) * 1000, actions=actions
220	)
221
222
223	class TracingHandler(TracingHandlerBase):	1✔
224	delegate: Handler	1✔
225
226	def __init__(self, delegate: Handler):	1✔
UNCOV 227	super().__init__(delegate)	×
228
229	def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):	1✔
230	def _call():	×
UNCOV 231	return self.delegate(chain, context, response)	×
232
UNCOV 233	return self.do_trace_call(_call, chain, context, response)	×
234
235
236	class TracingExceptionHandler(TracingHandlerBase):	1✔
237	delegate: ExceptionHandler	1✔
238
239	def __init__(self, delegate: ExceptionHandler):	1✔
UNCOV 240	super().__init__(delegate)	×
241
242	def __call__(	1✔
243	self, chain: HandlerChain, exception: Exception, context: RequestContext, response: Response
244	):
245	def _call():	×
UNCOV 246	return self.delegate(chain, exception, context, response)	×
247
UNCOV 248	return self.do_trace_call(_call, chain, context, response)	×
249
250
251	class TracingHandlerChain(HandlerChain):	1✔
252	"""
253	DebuggingHandlerChain - A subclass of HandlerChain for logging and tracing handlers.
254
255	Attributes:
256	- duration (float): Total time taken for handling request in milliseconds.
257	- request_handler_traces (list[HandlerTrace]): List of request handler traces.
258	- response_handler_traces (list[HandlerTrace]): List of response handler traces.
259	- finalizer_traces (list[HandlerTrace]): List of finalizer traces.
260	- exception_handler_traces (list[HandlerTrace]): List of exception handler traces.
261	"""
262
263	duration: float	1✔
264	request_handler_traces: list[HandlerTrace]	1✔
265	response_handler_traces: list[HandlerTrace]	1✔
266	finalizer_traces: list[HandlerTrace]	1✔
267	exception_handler_traces: list[HandlerTrace]	1✔
268
269	def __init__(self, args, *kwargs):	1✔
270	super().__init__(args, *kwargs)	×
271	self.request_handler_traces = []	×
272	self.response_handler_traces = []	×
273	self.finalizer_traces = []	×
UNCOV 274	self.exception_handler_traces = []	×
275
276	def handle(self, context: RequestContext, response: Response):	1✔
277	"""Overrides HandlerChain's handle method and adds tracing handler to request handlers. Logs the trace
278	report with request and response details."""
279	then = time.perf_counter()	×
280	try:	×
281	self.request_handlers = [TracingHandler(handler) for handler in self.request_handlers]	×
UNCOV 282	return super().handle(context, response)	×
283	finally:
284	self.duration = (time.perf_counter() - then) * 1000	×
285	self.request_handler_traces = [handler.trace for handler in self.request_handlers]	×
UNCOV 286	self._log_report()	×
287
288	def _call_response_handlers(self, response):	1✔
289	self.response_handlers = [TracingHandler(handler) for handler in self.response_handlers]	×
290	try:	×
UNCOV 291	return super()._call_response_handlers(response)	×
292	finally:
UNCOV 293	self.response_handler_traces = [handler.trace for handler in self.response_handlers]	×
294
295	def _call_finalizers(self, response):	1✔
296	self.finalizers = [TracingHandler(handler) for handler in self.finalizers]	×
297	try:	×
UNCOV 298	return super()._call_response_handlers(response)	×
299	finally:
UNCOV 300	self.finalizer_traces = [handler.trace for handler in self.finalizers]	×
301
302	def _call_exception_handlers(self, e, response):	1✔
UNCOV 303	self.exception_handlers = [	×
304	TracingExceptionHandler(handler) for handler in self.exception_handlers
305	]
306	try:	×
UNCOV 307	return super()._call_exception_handlers(e, response)	×
308	finally:
UNCOV 309	self.exception_handler_traces = [handler.trace for handler in self.exception_handlers]	×
310
311	def _log_report(self):	1✔
312	report = []	×
313	request = self.context.request	×
UNCOV 314	response = self.response	×
315
UNCOV 316	def _append_traces(traces: list[HandlerTrace]):	×
317	"""Format and appends a list of traces to the report, and recursively append the trace's
318	actions (if any)."""
319
320	for trace in traces:	×
321	if trace is None:	×
UNCOV 322	continue	×
323
UNCOV 324	report.append(	×
325	f"{trace.handler_module:43s} {trace.handler_name:30s} {trace.duration_ms:8.2f}ms"
326	)
UNCOV 327	_append_actions(trace.actions, 46)	×
328
329	def _append_actions(actions: list[Action], indent: int):	×
330	for action in actions:	×
UNCOV 331	report.append((" " * indent) + f"- {action!r}")	×
332
333	if isinstance(action, ModifyHeadersAction):	×
UNCOV 334	_append_actions(action.header_actions, indent + 2)	×
335
336	report.append(f"request: {request.method} {request.url}")	×
337	report.append(f"response: {response.status_code}")	×
338	report.append("---- request handlers " + ("-" * 63))	×
339	_append_traces(self.request_handler_traces)	×
340	report.append("---- response handlers " + ("-" * 63))	×
341	_append_traces(self.response_handler_traces)	×
342	report.append("---- finalizers " + ("-" * 63))	×
343	_append_traces(self.finalizer_traces)	×
344	report.append("---- exception handlers " + ("-" * 63))	×
UNCOV 345	_append_traces(self.exception_handler_traces)	×
346	# Add a separator and total duration value to the end of the report
UNCOV 347	report.append(f"{'=' * 68} total {self.duration:8.2f}ms")	×
348
UNCOV 349	LOG.info("handler chain trace report:\n%s\n%s", "=" * 85, "\n".join(report))	×

localstack / localstack / 16820655284

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