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

93.2

/localstack-core/localstack/runtime/init.py

"""Module for initialization hooks https://docs.localstack.cloud/references/init-hooks/"""

import dataclasses
import logging
import os.path
import subprocess
import time
from enum import Enum
from functools import cached_property

from plux import Plugin, PluginManager

from localstack.runtime import hooks
from localstack.utils.objects import singleton_factory

LOG = logging.getLogger(__name__)


class State(Enum):
    UNKNOWN = "UNKNOWN"
    RUNNING = "RUNNING"
    SUCCESSFUL = "SUCCESSFUL"
    ERROR = "ERROR"

    def __str__(self):
        return self.name

    def __repr__(self):
        return self.name


class Stage(Enum):
    BOOT = 0
    START = 1
    READY = 2
    SHUTDOWN = 3

    def __str__(self):
        return self.name

    def __repr__(self):
        return self.name


@dataclasses.dataclass
class Script:
    path: str
    stage: Stage
    state: State = State.UNKNOWN


class ScriptRunner(Plugin):
    """
    Interface for running scripts.
    """

    namespace = "localstack.init.runner"
    suffixes = []

    def run(self, path: str) -> None:
        """
        Run the given script with the appropriate runtime.

        :param path: the path to the script
        """
        raise NotImplementedError

    def should_run(self, script_file: str) -> bool:
        """
        Checks whether the given file should be run with this script runner. In case multiple runners
        evaluate this condition to true on the same file (ideally this doesn't happen), the first one
        loaded will be used, which is potentially indeterministic.

        :param script_file: the script file to run
        :return: True if this runner should be used, False otherwise
        """
        for suffix in self.suffixes:
            if script_file.endswith(suffix):
                return True
        return False


class ShellScriptRunner(ScriptRunner):
    """
    Runner that interprets scripts as shell scripts and calls them directly.
    """

    name = "sh"
    suffixes = [".sh"]

    def run(self, path: str) -> None:
        exit_code = subprocess.call(args=[], executable=path)
        if exit_code != 0:
            raise OSError("Script %s returned a non-zero exit code %s" % (path, exit_code))


class PythonScriptRunner(ScriptRunner):
    """
    Runner that uses ``exec`` to run a python script.
    """

    name = "py"
    suffixes = [".py"]

    def run(self, path: str) -> None:
        with open(path, "rb") as fd:
            exec(fd.read(), {})


class InitScriptManager:
    _stage_directories: dict[Stage, str] = {
        Stage.BOOT: "boot.d",
        Stage.START: "start.d",
        Stage.READY: "ready.d",
        Stage.SHUTDOWN: "shutdown.d",
    }

    script_root: str
    stage_completed: dict[Stage, bool]

    def __init__(self, script_root: str):
        self.script_root = script_root
        self.stage_completed = dict.fromkeys(Stage, False)
        self.runner_manager: PluginManager[ScriptRunner] = PluginManager(ScriptRunner.namespace)

    @cached_property
    def scripts(self) -> dict[Stage, list[Script]]:
        return self._find_scripts()

    def get_script_runner(self, script_file: str) -> ScriptRunner | None:
        runners = self.runner_manager.load_all()
        for runner in runners:
            if runner.should_run(script_file):
                return runner
        return None

    def has_script_runner(self, script_file: str) -> bool:
        return self.get_script_runner(script_file) is not None

    def run_stage(self, stage: Stage) -> list[Script]:
        """
        Runs all scripts in the given stage.

        :param stage: the stage to run
        :return: the scripts that were in the stage
        """
        scripts = self.scripts.get(stage, [])

        if self.stage_completed[stage]:
            LOG.debug("Stage %s already completed, skipping", stage)
            return scripts

        try:
            for script in scripts:
                LOG.debug("Running %s script %s", script.stage, script.path)

                env_original = os.environ.copy()

                try:
                    script.state = State.RUNNING
                    runner = self.get_script_runner(script.path)
                    runner.run(script.path)
                except Exception as e:
                    script.state = State.ERROR
                    if LOG.isEnabledFor(logging.DEBUG):
                        LOG.exception("Error while running script %s", script)
                    else:
                        LOG.error("Error while running script %s: %s", script, e)
                else:
                    script.state = State.SUCCESSFUL
                finally:
                    # Discard env variables overridden in startup script that may cause side-effects
                    for env_var in (
                        "AWS_ACCESS_KEY_ID",
                        "AWS_SECRET_ACCESS_KEY",
                        "AWS_SESSION_TOKEN",
                        "AWS_DEFAULT_REGION",
                        "AWS_PROFILE",
                        "AWS_REGION",
                    ):
                        if env_var in env_original:
                            os.environ[env_var] = env_original[env_var]
                        else:
                            os.environ.pop(env_var, None)
        finally:
            self.stage_completed[stage] = True

        return scripts

    def _find_scripts(self) -> dict[Stage, list[Script]]:
        scripts = {}

        if self.script_root is None:
            LOG.debug("Unable to discover init scripts as script_root is None")
            return {}

        for stage in Stage:
            scripts[stage] = []

            stage_dir = self._stage_directories[stage]
            if not stage_dir:
                continue

            stage_path = os.path.join(self.script_root, stage_dir)
            if not os.path.isdir(stage_path):
                continue

            for root, dirs, files in os.walk(stage_path, topdown=True):
                # from the docs: "When topdown is true, the caller can modify the dirnames list in-place"
                dirs.sort()
                files.sort()
                for file in files:
                    script_path = os.path.abspath(os.path.join(root, file))
                    if not os.path.isfile(script_path):
                        continue

                    # only add the script if there's a runner for it
                    if not self.has_script_runner(script_path):
                        LOG.debug("No runner available for script %s", script_path)
                        continue

                    scripts[stage].append(Script(path=script_path, stage=stage))
        LOG.debug("Init scripts discovered: %s", scripts)

        return scripts


# runtime integration


@singleton_factory
def init_script_manager() -> InitScriptManager:
    from localstack import config

    return InitScriptManager(script_root=config.dirs.init)


@hooks.on_infra_start()
def _run_init_scripts_on_start():
    # this is a hack since we currently cannot know whether boot scripts have been executed or not
    init_script_manager().stage_completed[Stage.BOOT] = True
    _run_and_log(Stage.START)


@hooks.on_infra_ready()
def _run_init_scripts_on_ready():
    _run_and_log(Stage.READY)


@hooks.on_infra_shutdown()
def _run_init_scripts_on_shutdown():
    _run_and_log(Stage.SHUTDOWN)


def _run_and_log(stage: Stage):
    from localstack.utils.analytics import log

    then = time.time()
    scripts = init_script_manager().run_stage(stage)
    took = (time.time() - then) * 1000

    if scripts:
        log.event("run_init", {"stage": stage.name, "scripts": len(scripts), "duration": took})


def main():
    """
    Run the init scripts for a particular stage. For example, to run all boot scripts run::

        python -m localstack.runtime.init BOOT

    The __main__ entrypoint is currently mainly used for the docker-entrypoint.sh. Other stages
    are executed from runtime hooks.
    """
    import sys

    stage = Stage[sys.argv[1]]
    init_script_manager().run_stage(stage)


if __name__ == "__main__":
    main()

1	"""Module for initialization hooks https://docs.localstack.cloud/references/init-hooks/"""
2
3	import dataclasses	1✔
4	import logging	1✔
5	import os.path	1✔
6	import subprocess	1✔
7	import time	1✔
8	from enum import Enum	1✔
9	from functools import cached_property	1✔
10
11	from plux import Plugin, PluginManager	1✔
12
13	from localstack.runtime import hooks	1✔
14	from localstack.utils.objects import singleton_factory	1✔
15
16	LOG = logging.getLogger(__name__)	1✔
17
18
19	class State(Enum):	1✔
20	UNKNOWN = "UNKNOWN"	1✔
21	RUNNING = "RUNNING"	1✔
22	SUCCESSFUL = "SUCCESSFUL"	1✔
23	ERROR = "ERROR"	1✔
24
25	def __str__(self):	1✔
UNCOV 26	return self.name	×
27
28	def __repr__(self):
29	return self.name
30
31
32	class Stage(Enum):	1✔
33	BOOT = 0	1✔
34	START = 1	1✔
35	READY = 2	1✔
36	SHUTDOWN = 3	1✔
37
38	def __str__(self):	1✔
39	return self.name	1✔
40
41	def __repr__(self):
42	return self.name
43
44
45	@dataclasses.dataclass	1✔
46	class Script:	1✔
47	path: str	1✔
48	stage: Stage	1✔
49	state: State = State.UNKNOWN	1✔
50
51
52	class ScriptRunner(Plugin):	1✔
53	"""
54	Interface for running scripts.
55	"""
56
57	namespace = "localstack.init.runner"	1✔
58	suffixes = []	1✔
59
60	def run(self, path: str) -> None:	1✔
61	"""
62	Run the given script with the appropriate runtime.
63
64	:param path: the path to the script
65	"""
66	raise NotImplementedError
67
68	def should_run(self, script_file: str) -> bool:	1✔
69	"""
70	Checks whether the given file should be run with this script runner. In case multiple runners
71	evaluate this condition to true on the same file (ideally this doesn't happen), the first one
72	loaded will be used, which is potentially indeterministic.
73
74	:param script_file: the script file to run
75	:return: True if this runner should be used, False otherwise
76	"""
77	for suffix in self.suffixes:	1✔
78	if script_file.endswith(suffix):	1✔
79	return True	1✔
80	return False	1✔
81
82
83	class ShellScriptRunner(ScriptRunner):	1✔
84	"""
85	Runner that interprets scripts as shell scripts and calls them directly.
86	"""
87
88	name = "sh"	1✔
89	suffixes = [".sh"]	1✔
90
91	def run(self, path: str) -> None:	1✔
92	exit_code = subprocess.call(args=[], executable=path)	1✔
93	if exit_code != 0:	1✔
94	raise OSError("Script %s returned a non-zero exit code %s" % (path, exit_code))	1✔
95
96
97	class PythonScriptRunner(ScriptRunner):	1✔
98	"""
99	Runner that uses ``exec`` to run a python script.
100	"""
101
102	name = "py"	1✔
103	suffixes = [".py"]	1✔
104
105	def run(self, path: str) -> None:	1✔
106	with open(path, "rb") as fd:	1✔
107	exec(fd.read(), {})	1✔
108
109
110	class InitScriptManager:	1✔
111	_stage_directories: dict[Stage, str] = {	1✔
112	Stage.BOOT: "boot.d",
113	Stage.START: "start.d",
114	Stage.READY: "ready.d",
115	Stage.SHUTDOWN: "shutdown.d",
116	}
117
118	script_root: str	1✔
119	stage_completed: dict[Stage, bool]	1✔
120
121	def __init__(self, script_root: str):	1✔
122	self.script_root = script_root	1✔
123	self.stage_completed = dict.fromkeys(Stage, False)	1✔
124	self.runner_manager: PluginManager[ScriptRunner] = PluginManager(ScriptRunner.namespace)	1✔
125
126	@cached_property	1✔
127	def scripts(self) -> dict[Stage, list[Script]]:	1✔
128	return self._find_scripts()	1✔
129
130	def get_script_runner(self, script_file: str) -> ScriptRunner \| None:	1✔
131	runners = self.runner_manager.load_all()	1✔
132	for runner in runners:	1✔
133	if runner.should_run(script_file):	1✔
134	return runner	1✔
135	return None	1✔
136
137	def has_script_runner(self, script_file: str) -> bool:	1✔
138	return self.get_script_runner(script_file) is not None	1✔
139
140	def run_stage(self, stage: Stage) -> list[Script]:	1✔
141	"""
142	Runs all scripts in the given stage.
143
144	:param stage: the stage to run
145	:return: the scripts that were in the stage
146	"""
147	scripts = self.scripts.get(stage, [])	1✔
148
149	if self.stage_completed[stage]:	1✔
UNCOV 150	LOG.debug("Stage %s already completed, skipping", stage)	×
151	return scripts	×
152
153	try:	1✔
154	for script in scripts:	1✔
155	LOG.debug("Running %s script %s", script.stage, script.path)	1✔
156
157	env_original = os.environ.copy()	1✔
158
159	try:	1✔
160	script.state = State.RUNNING	1✔
161	runner = self.get_script_runner(script.path)	1✔
162	runner.run(script.path)	1✔
163	except Exception as e:	1✔
164	script.state = State.ERROR	1✔
165	if LOG.isEnabledFor(logging.DEBUG):	1✔
166	LOG.exception("Error while running script %s", script)	1✔
167	else:
UNCOV 168	LOG.error("Error while running script %s: %s", script, e)	×
169	else:
170	script.state = State.SUCCESSFUL	1✔
171	finally:
172	# Discard env variables overridden in startup script that may cause side-effects
173	for env_var in (	1✔
174	"AWS_ACCESS_KEY_ID",
175	"AWS_SECRET_ACCESS_KEY",
176	"AWS_SESSION_TOKEN",
177	"AWS_DEFAULT_REGION",
178	"AWS_PROFILE",
179	"AWS_REGION",
180	):
181	if env_var in env_original:	1✔
182	os.environ[env_var] = env_original[env_var]	1✔
183	else:
184	os.environ.pop(env_var, None)	1✔
185	finally:
186	self.stage_completed[stage] = True	1✔
187
188	return scripts	1✔
189
190	def _find_scripts(self) -> dict[Stage, list[Script]]:	1✔
191	scripts = {}	1✔
192
193	if self.script_root is None:	1✔
194	LOG.debug("Unable to discover init scripts as script_root is None")	1✔
195	return {}	1✔
196
197	for stage in Stage:	1✔
198	scripts[stage] = []	1✔
199
200	stage_dir = self._stage_directories[stage]	1✔
201	if not stage_dir:	1✔
UNCOV 202	continue	×
203
204	stage_path = os.path.join(self.script_root, stage_dir)	1✔
205	if not os.path.isdir(stage_path):	1✔
206	continue	1✔
207
208	for root, dirs, files in os.walk(stage_path, topdown=True):	1✔
209	# from the docs: "When topdown is true, the caller can modify the dirnames list in-place"
210	dirs.sort()	1✔
211	files.sort()	1✔
212	for file in files:	1✔
213	script_path = os.path.abspath(os.path.join(root, file))	1✔
214	if not os.path.isfile(script_path):	1✔
UNCOV 215	continue	×
216
217	# only add the script if there's a runner for it
218	if not self.has_script_runner(script_path):	1✔
219	LOG.debug("No runner available for script %s", script_path)	1✔
220	continue	1✔
221
222	scripts[stage].append(Script(path=script_path, stage=stage))	1✔
223	LOG.debug("Init scripts discovered: %s", scripts)	1✔
224
225	return scripts	1✔
226
227
228	# runtime integration
229
230
231	@singleton_factory	1✔
232	def init_script_manager() -> InitScriptManager:	1✔
233	from localstack import config	1✔
234
235	return InitScriptManager(script_root=config.dirs.init)	1✔
236
237
238	@hooks.on_infra_start()	1✔
239	def _run_init_scripts_on_start():	1✔
240	# this is a hack since we currently cannot know whether boot scripts have been executed or not
241	init_script_manager().stage_completed[Stage.BOOT] = True	1✔
242	_run_and_log(Stage.START)	1✔
243
244
245	@hooks.on_infra_ready()	1✔
246	def _run_init_scripts_on_ready():	1✔
247	_run_and_log(Stage.READY)	1✔
248
249
250	@hooks.on_infra_shutdown()	1✔
251	def _run_init_scripts_on_shutdown():	1✔
252	_run_and_log(Stage.SHUTDOWN)	1✔
253
254
255	def _run_and_log(stage: Stage):	1✔
256	from localstack.utils.analytics import log	1✔
257
258	then = time.time()	1✔
259	scripts = init_script_manager().run_stage(stage)	1✔
260	took = (time.time() - then) * 1000	1✔
261
262	if scripts:	1✔
UNCOV 263	log.event("run_init", {"stage": stage.name, "scripts": len(scripts), "duration": took})	×
264
265
266	def main():	1✔
267	"""
268	Run the init scripts for a particular stage. For example, to run all boot scripts run::
269
270	python -m localstack.runtime.init BOOT
271
272	The __main__ entrypoint is currently mainly used for the docker-entrypoint.sh. Other stages
273	are executed from runtime hooks.
274	"""
UNCOV 275	import sys	×
276
UNCOV 277	stage = Stage[sys.argv[1]]	×
278	init_script_manager().run_stage(stage)	×
279
280
281	if __name__ == "__main__":
282	main()

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