22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

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

61.11

/localstack-core/localstack/services/internal.py

"""Module for localstack internal resources, such as health, graph, or _localstack/cloudformation/deploy."""

import logging
import os
import re
import time
from collections import defaultdict
from datetime import datetime

from plux import PluginManager

from localstack import config, constants
from localstack.deprecations import deprecated_endpoint
from localstack.http import Request, Resource, Response, Router
from localstack.http.dispatcher import handler_dispatcher
from localstack.runtime.legacy import signal_supervisor_restart
from localstack.utils.analytics.metadata import (
    get_client_metadata,
    get_localstack_edition,
    is_license_activated,
)
from localstack.utils.collections import merge_recursive
from localstack.utils.functions import call_safe
from localstack.utils.numbers import is_number
from localstack.utils.objects import singleton_factory

LOG = logging.getLogger(__name__)

HTTP_METHODS = ["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH"]


class DeprecatedResource:
    """
    Resource class which wraps a given resource in the deprecated_endpoint (i.e. logs deprecation warnings on every
    invocation).
    """

    def __init__(self, resource, previous_path: str, deprecation_version: str, new_path: str):
        for http_method in HTTP_METHODS:
            fn_name = f"on_{http_method.lower()}"
            fn = getattr(resource, fn_name, None)
            if fn:
                wrapped = deprecated_endpoint(
                    fn,
                    previous_path=previous_path,
                    deprecation_version=deprecation_version,
                    new_path=new_path,
                )
                setattr(self, fn_name, wrapped)


class HealthResource:
    """
    Resource for the LocalStack /health endpoint. It provides access to the service states and other components of
    localstack. We support arbitrary data to be put into the health state to support things like the
    run_startup_scripts function in docker-entrypoint.sh which sets the status of the init scripts feature.
    """

    def __init__(self, service_manager) -> None:
        super().__init__()
        self.service_manager = service_manager
        self.state = {}

    def on_post(self, request: Request):
        data = request.get_json(True, True)
        if not data:
            return Response("invalid request", 400)

        # backdoor API to support restarting the instance
        if data.get("action") == "restart":
            signal_supervisor_restart()
        elif data.get("action") == "kill":
            from localstack.runtime import get_current_runtime

            get_current_runtime().exit(0)

        return Response("ok", 200)

    def on_get(self, request: Request):
        path = request.path

        reload = "reload" in path

        # get service state
        if reload:
            self.service_manager.check_all()
        services = {
            service: state.value for service, state in self.service_manager.get_states().items()
        }

        # build state dict from internal state and merge into it the service states
        result = dict(self.state)
        result = merge_recursive({"services": services}, result)
        result["edition"] = get_localstack_edition()
        result["version"] = constants.VERSION
        return result

    def on_head(self, request: Request):
        return Response("ok", 200)

    def on_put(self, request: Request):
        data = request.get_json(True, True) or {}

        # keys like "features:initScripts" should be interpreted as ['features']['initScripts']
        state = defaultdict(dict)
        for k, v in data.items():
            if ":" in k:
                path = k.split(":")
            else:
                path = [k]

            d = state
            for p in path[:-1]:
                d = state[p]
            d[path[-1]] = v

        self.state = merge_recursive(state, self.state, overwrite=True)
        return {"status": "OK"}


class InfoResource:
    """
    Resource that is exposed to /_localstack/info and used to get generalized information about the current
    localstack instance.
    """

    def on_get(self, request):
        return self.get_info_data()

    @staticmethod
    def get_info_data() -> dict:
        client_metadata = get_client_metadata()
        uptime = int(time.time() - config.load_start_time)

        return {
            "version": client_metadata.version,
            "edition": get_localstack_edition(),
            "is_license_activated": is_license_activated(),
            "session_id": client_metadata.session_id,
            "machine_id": client_metadata.machine_id,
            "system": client_metadata.system,
            "is_docker": client_metadata.is_docker,
            "server_time_utc": datetime.utcnow().isoformat(timespec="seconds"),
            "uptime": uptime,
        }


class UsageResource:
    def on_get(self, request):
        from localstack.utils import diagnose

        return call_safe(diagnose.get_usage) or {}


class DiagnoseResource:
    def on_get(self, request):
        from localstack.utils import diagnose

        return {
            "version": {
                "image-version": call_safe(diagnose.get_docker_image_details),
                "localstack-version": call_safe(diagnose.get_localstack_version),
                "host": {
                    "kernel": call_safe(diagnose.get_host_kernel_version),
                },
            },
            "info": call_safe(InfoResource.get_info_data),
            "services": call_safe(diagnose.get_service_stats),
            "config": call_safe(diagnose.get_localstack_config),
            "docker-inspect": call_safe(diagnose.inspect_main_container),
            "docker-dependent-image-hashes": call_safe(diagnose.get_important_image_hashes),
            "file-tree": call_safe(diagnose.get_file_tree),
            "important-endpoints": call_safe(diagnose.resolve_endpoints),
            "logs": call_safe(diagnose.get_localstack_logs),
            "usage": call_safe(diagnose.get_usage),
        }


class PluginsResource:
    """
    Resource to list information about plux plugins.
    """

    plugin_managers: list[PluginManager] = []

    def __init__(self):
        # defer imports here to lazy-load code
        from localstack.runtime import hooks, init
        from localstack.services.plugins import SERVICE_PLUGINS

        # service providers
        PluginsResource.plugin_managers.append(SERVICE_PLUGINS.plugin_manager)
        # init script runners
        PluginsResource.plugin_managers.append(init.init_script_manager().runner_manager)
        # init hooks
        PluginsResource.plugin_managers.append(hooks.configure_localstack_container.manager)
        PluginsResource.plugin_managers.append(hooks.prepare_host.manager)
        PluginsResource.plugin_managers.append(hooks.on_infra_ready.manager)
        PluginsResource.plugin_managers.append(hooks.on_infra_start.manager)
        PluginsResource.plugin_managers.append(hooks.on_infra_shutdown.manager)

    def on_get(self, request):
        return {
            manager.namespace: [
                self._get_plugin_details(manager, name) for name in manager.list_names()
            ]
            for manager in self.plugin_managers
        }

    def _get_plugin_details(self, manager: PluginManager, plugin_name: str) -> dict:
        container = manager.get_container(plugin_name)

        details = {
            "name": plugin_name,
            "is_initialized": container.is_init,
            "is_loaded": container.is_loaded,
        }

        # optionally add requires_license information if the plugin provides it
        requires_license = None
        if container.plugin:
            try:
                requires_license = container.plugin.requires_license
            except AttributeError:
                pass
        if requires_license is not None:
            details["requires_license"] = requires_license

        return details


class InitScriptsResource:
    def on_get(self, request):
        from localstack.runtime.init import init_script_manager

        manager = init_script_manager()

        return {
            "completed": {
                stage.name: completed for stage, completed in manager.stage_completed.items()
            },
            "scripts": [
                {
                    "stage": script.stage.name,
                    "name": os.path.basename(script.path),
                    "state": script.state.name,
                }
                for scripts in manager.scripts.values()
                for script in scripts
            ],
        }


class InitScriptsStageResource:
    def on_get(self, request, stage: str):
        from localstack.runtime.init import Stage, init_script_manager

        manager = init_script_manager()

        try:
            stage = Stage[stage.upper()]
        except KeyError:
            return Response(f"no such stage {stage}", 404)

        return {
            "completed": manager.stage_completed.get(stage),
            "scripts": [
                {
                    "stage": script.stage.name,
                    "name": os.path.basename(script.path),
                    "state": script.state.name,
                }
                for script in manager.scripts.get(stage)
            ],
        }


class ConfigResource:
    def on_get(self, request):
        from localstack.utils import diagnose

        return call_safe(diagnose.get_localstack_config)

    def on_post(self, request: Request):
        from localstack.utils.config_listener import update_config_variable

        data = request.get_json(force=True)
        variable = data.get("variable", "")
        if not re.match(r"^[_a-zA-Z0-9]+$", variable):
            return Response("{}", mimetype="application/json", status=400)
        new_value = data.get("value")
        if is_number(new_value):
            new_value = float(new_value)
        update_config_variable(variable, new_value)
        value = getattr(config, variable, None)
        return {
            "variable": variable,
            "value": value,
        }


class LocalstackResources(Router):
    """
    Router for localstack-internal HTTP resources.
    """

    def __init__(self):
        super().__init__(dispatcher=handler_dispatcher())
        self.add_default_routes()
        # TODO: load routes as plugins

    def add_default_routes(self):
        from localstack.services.plugins import SERVICE_PLUGINS

        health_resource = HealthResource(SERVICE_PLUGINS)
        self.add(Resource("/_localstack/health", health_resource))
        self.add(Resource("/_localstack/info", InfoResource()))
        self.add(Resource("/_localstack/plugins", PluginsResource()))
        self.add(Resource("/_localstack/init", InitScriptsResource()))
        self.add(Resource("/_localstack/init/<stage>", InitScriptsStageResource()))

        if config.ENABLE_CONFIG_UPDATES:
            LOG.warning(
                "Enabling config endpoint, "
                "please be aware that this can expose sensitive information via your network."
            )
            self.add(Resource("/_localstack/config", ConfigResource()))

        if config.DEBUG:
            LOG.warning(
                "Enabling diagnose endpoint, "
                "please be aware that this can expose sensitive information via your network."
            )
            self.add(Resource("/_localstack/diagnose", DiagnoseResource()))
            self.add(Resource("/_localstack/usage", UsageResource()))


@singleton_factory
def get_internal_apis() -> LocalstackResources:
    """
    Get the LocalstackResources singleton.
    """
    return LocalstackResources()

1	"""Module for localstack internal resources, such as health, graph, or _localstack/cloudformation/deploy."""
2
3	import logging	1✔
4	import os	1✔
5	import re	1✔
6	import time	1✔
7	from collections import defaultdict	1✔
8	from datetime import datetime	1✔
9
10	from plux import PluginManager	1✔
11
12	from localstack import config, constants	1✔
13	from localstack.deprecations import deprecated_endpoint	1✔
14	from localstack.http import Request, Resource, Response, Router	1✔
15	from localstack.http.dispatcher import handler_dispatcher	1✔
16	from localstack.runtime.legacy import signal_supervisor_restart	1✔
17	from localstack.utils.analytics.metadata import (	1✔
18	get_client_metadata,
19	get_localstack_edition,
20	is_license_activated,
21	)
22	from localstack.utils.collections import merge_recursive	1✔
23	from localstack.utils.functions import call_safe	1✔
24	from localstack.utils.numbers import is_number	1✔
25	from localstack.utils.objects import singleton_factory	1✔
26
27	LOG = logging.getLogger(__name__)	1✔
28
29	HTTP_METHODS = ["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH"]	1✔
30
31
32	class DeprecatedResource:	1✔
33	"""
34	Resource class which wraps a given resource in the deprecated_endpoint (i.e. logs deprecation warnings on every
35	invocation).
36	"""
37
38	def __init__(self, resource, previous_path: str, deprecation_version: str, new_path: str):	1✔
39	for http_method in HTTP_METHODS:	×
40	fn_name = f"on_{http_method.lower()}"	×
41	fn = getattr(resource, fn_name, None)	×
42	if fn:	×
43	wrapped = deprecated_endpoint(	×
44	fn,
45	previous_path=previous_path,
46	deprecation_version=deprecation_version,
47	new_path=new_path,
48	)
49	setattr(self, fn_name, wrapped)	×
50
51
52	class HealthResource:	1✔
53	"""
54	Resource for the LocalStack /health endpoint. It provides access to the service states and other components of
55	localstack. We support arbitrary data to be put into the health state to support things like the
56	run_startup_scripts function in docker-entrypoint.sh which sets the status of the init scripts feature.
57	"""
58
59	def __init__(self, service_manager) -> None:	1✔
60	super().__init__()	1✔
61	self.service_manager = service_manager	1✔
62	self.state = {}	1✔
63
64	def on_post(self, request: Request):	1✔
65	data = request.get_json(True, True)	×
66	if not data:	×
67	return Response("invalid request", 400)	×
68
69	# backdoor API to support restarting the instance
70	if data.get("action") == "restart":	×
71	signal_supervisor_restart()	×
72	elif data.get("action") == "kill":	×
73	from localstack.runtime import get_current_runtime	×
74
75	get_current_runtime().exit(0)	×
76
77	return Response("ok", 200)	×
78
79	def on_get(self, request: Request):	1✔
80	path = request.path	1✔
81
82	reload = "reload" in path	1✔
83
84	# get service state
85	if reload:	1✔
86	self.service_manager.check_all()	×
87	services = {	1✔
88	service: state.value for service, state in self.service_manager.get_states().items()
89	}
90
91	# build state dict from internal state and merge into it the service states
92	result = dict(self.state)	1✔
93	result = merge_recursive({"services": services}, result)	1✔
94	result["edition"] = get_localstack_edition()	1✔
95	result["version"] = constants.VERSION	1✔
96	return result	1✔
97
98	def on_head(self, request: Request):	1✔
UNCOV 99	return Response("ok", 200)	×
100
101	def on_put(self, request: Request):	1✔
102	data = request.get_json(True, True) or {}	1✔
103
104	# keys like "features:initScripts" should be interpreted as ['features']['initScripts']
105	state = defaultdict(dict)	1✔
106	for k, v in data.items():	1✔
107	if ":" in k:	1✔
108	path = k.split(":")	1✔
109	else:
110	path = [k]	×
111
112	d = state	1✔
113	for p in path[:-1]:	1✔
114	d = state[p]	1✔
115	d[path[-1]] = v	1✔
116
117	self.state = merge_recursive(state, self.state, overwrite=True)	1✔
118	return {"status": "OK"}	1✔
119
120
121	class InfoResource:	1✔
122	"""
123	Resource that is exposed to /_localstack/info and used to get generalized information about the current
124	localstack instance.
125	"""
126
127	def on_get(self, request):	1✔
UNCOV 128	return self.get_info_data()	×
129
130	@staticmethod	1✔
131	def get_info_data() -> dict:	1✔
UNCOV 132	client_metadata = get_client_metadata()	×
UNCOV 133	uptime = int(time.time() - config.load_start_time)	×
134
UNCOV 135	return {	×
136	"version": client_metadata.version,
137	"edition": get_localstack_edition(),
138	"is_license_activated": is_license_activated(),
139	"session_id": client_metadata.session_id,
140	"machine_id": client_metadata.machine_id,
141	"system": client_metadata.system,
142	"is_docker": client_metadata.is_docker,
143	"server_time_utc": datetime.utcnow().isoformat(timespec="seconds"),
144	"uptime": uptime,
145	}
146
147
148	class UsageResource:	1✔
149	def on_get(self, request):	1✔
150	from localstack.utils import diagnose	×
151
152	return call_safe(diagnose.get_usage) or {}	×
153
154
155	class DiagnoseResource:	1✔
156	def on_get(self, request):	1✔
UNCOV 157	from localstack.utils import diagnose	×
158
UNCOV 159	return {	×
160	"version": {
161	"image-version": call_safe(diagnose.get_docker_image_details),
162	"localstack-version": call_safe(diagnose.get_localstack_version),
163	"host": {
164	"kernel": call_safe(diagnose.get_host_kernel_version),
165	},
166	},
167	"info": call_safe(InfoResource.get_info_data),
168	"services": call_safe(diagnose.get_service_stats),
169	"config": call_safe(diagnose.get_localstack_config),
170	"docker-inspect": call_safe(diagnose.inspect_main_container),
171	"docker-dependent-image-hashes": call_safe(diagnose.get_important_image_hashes),
172	"file-tree": call_safe(diagnose.get_file_tree),
173	"important-endpoints": call_safe(diagnose.resolve_endpoints),
174	"logs": call_safe(diagnose.get_localstack_logs),
175	"usage": call_safe(diagnose.get_usage),
176	}
177
178
179	class PluginsResource:	1✔
180	"""
181	Resource to list information about plux plugins.
182	"""
183
184	plugin_managers: list[PluginManager] = []	1✔
185
186	def __init__(self):	1✔
187	# defer imports here to lazy-load code
188	from localstack.runtime import hooks, init	1✔
189	from localstack.services.plugins import SERVICE_PLUGINS	1✔
190
191	# service providers
192	PluginsResource.plugin_managers.append(SERVICE_PLUGINS.plugin_manager)	1✔
193	# init script runners
194	PluginsResource.plugin_managers.append(init.init_script_manager().runner_manager)	1✔
195	# init hooks
196	PluginsResource.plugin_managers.append(hooks.configure_localstack_container.manager)	1✔
197	PluginsResource.plugin_managers.append(hooks.prepare_host.manager)	1✔
198	PluginsResource.plugin_managers.append(hooks.on_infra_ready.manager)	1✔
199	PluginsResource.plugin_managers.append(hooks.on_infra_start.manager)	1✔
200	PluginsResource.plugin_managers.append(hooks.on_infra_shutdown.manager)	1✔
201
202	def on_get(self, request):	1✔
203	return {	×
204	manager.namespace: [
205	self._get_plugin_details(manager, name) for name in manager.list_names()
206	]
207	for manager in self.plugin_managers
208	}
209
210	def _get_plugin_details(self, manager: PluginManager, plugin_name: str) -> dict:	1✔
211	container = manager.get_container(plugin_name)	×
212
213	details = {	×
214	"name": plugin_name,
215	"is_initialized": container.is_init,
216	"is_loaded": container.is_loaded,
217	}
218
219	# optionally add requires_license information if the plugin provides it
220	requires_license = None	×
221	if container.plugin:	×
222	try:	×
223	requires_license = container.plugin.requires_license	×
224	except AttributeError:	×
225	pass	×
226	if requires_license is not None:	×
227	details["requires_license"] = requires_license	×
228
229	return details	×
230
231
232	class InitScriptsResource:	1✔
233	def on_get(self, request):	1✔
UNCOV 234	from localstack.runtime.init import init_script_manager	×
235
UNCOV 236	manager = init_script_manager()	×
237
UNCOV 238	return {	×
239	"completed": {
240	stage.name: completed for stage, completed in manager.stage_completed.items()
241	},
242	"scripts": [
243	{
244	"stage": script.stage.name,
245	"name": os.path.basename(script.path),
246	"state": script.state.name,
247	}
248	for scripts in manager.scripts.values()
249	for script in scripts
250	],
251	}
252
253
254	class InitScriptsStageResource:	1✔
255	def on_get(self, request, stage: str):	1✔
UNCOV 256	from localstack.runtime.init import Stage, init_script_manager	×
257
UNCOV 258	manager = init_script_manager()	×
259
UNCOV 260	try:	×
UNCOV 261	stage = Stage[stage.upper()]	×
UNCOV 262	except KeyError:	×
UNCOV 263	return Response(f"no such stage {stage}", 404)	×
264
UNCOV 265	return {	×
266	"completed": manager.stage_completed.get(stage),
267	"scripts": [
268	{
269	"stage": script.stage.name,
270	"name": os.path.basename(script.path),
271	"state": script.state.name,
272	}
273	for script in manager.scripts.get(stage)
274	],
275	}
276
277
278	class ConfigResource:	1✔
279	def on_get(self, request):	1✔
280	from localstack.utils import diagnose	×
281
282	return call_safe(diagnose.get_localstack_config)	×
283
284	def on_post(self, request: Request):	1✔
UNCOV 285	from localstack.utils.config_listener import update_config_variable	×
286
UNCOV 287	data = request.get_json(force=True)	×
UNCOV 288	variable = data.get("variable", "")	×
UNCOV 289	if not re.match(r"^[_a-zA-Z0-9]+$", variable):	×
290	return Response("{}", mimetype="application/json", status=400)	×
UNCOV 291	new_value = data.get("value")	×
UNCOV 292	if is_number(new_value):	×
UNCOV 293	new_value = float(new_value)	×
UNCOV 294	update_config_variable(variable, new_value)	×
UNCOV 295	value = getattr(config, variable, None)	×
UNCOV 296	return {	×
297	"variable": variable,
298	"value": value,
299	}
300
301
302	class LocalstackResources(Router):	1✔
303	"""
304	Router for localstack-internal HTTP resources.
305	"""
306
307	def __init__(self):	1✔
308	super().__init__(dispatcher=handler_dispatcher())	1✔
309	self.add_default_routes()	1✔
310	# TODO: load routes as plugins
311
312	def add_default_routes(self):	1✔
313	from localstack.services.plugins import SERVICE_PLUGINS	1✔
314
315	health_resource = HealthResource(SERVICE_PLUGINS)	1✔
316	self.add(Resource("/_localstack/health", health_resource))	1✔
317	self.add(Resource("/_localstack/info", InfoResource()))	1✔
318	self.add(Resource("/_localstack/plugins", PluginsResource()))	1✔
319	self.add(Resource("/_localstack/init", InitScriptsResource()))	1✔
320	self.add(Resource("/_localstack/init/<stage>", InitScriptsStageResource()))	1✔
321
322	if config.ENABLE_CONFIG_UPDATES:	1✔
323	LOG.warning(	×
324	"Enabling config endpoint, "
325	"please be aware that this can expose sensitive information via your network."
326	)
327	self.add(Resource("/_localstack/config", ConfigResource()))	×
328
329	if config.DEBUG:	1✔
330	LOG.warning(	1✔
331	"Enabling diagnose endpoint, "
332	"please be aware that this can expose sensitive information via your network."
333	)
334	self.add(Resource("/_localstack/diagnose", DiagnoseResource()))	1✔
335	self.add(Resource("/_localstack/usage", UsageResource()))	1✔
336
337
338	@singleton_factory	1✔
339	def get_internal_apis() -> LocalstackResources:	1✔
340	"""
341	Get the LocalstackResources singleton.
342	"""
343	return LocalstackResources()	1✔

localstack / localstack / 22709357475

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