ea65363c-1e20-4f77-ac7a-bf15452fc5ae

Committed 04 Feb 2025 08:12PM UTC coverage: 86.897% (-0.02%) from 86.912%

Build # ea65363c-1e20-4f77-ac7a-bf15452fc5ae

Build Type

push

circleci

Committed by

web-flow

Commit Message

Step Functions: Extended Support for Escape Sequences in String Literals (#12222)

Coverage Stats

9 of 9 new or added lines in 1 file covered. (100.0%)

21 existing lines in 6 files now uncovered.

61437 of 70701 relevant lines covered (86.9%)

0.87 hits per line

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

80.65

/localstack-core/localstack/services/dynamodb/server.py

import logging
import os
import threading

from localstack import config
from localstack.aws.connect import connect_externally_to
from localstack.aws.forwarder import AwsRequestProxy
from localstack.config import is_env_true
from localstack.constants import DEFAULT_AWS_ACCOUNT_ID
from localstack.services.dynamodb.packages import dynamodblocal_package
from localstack.utils.common import TMP_THREADS, ShellCommandThread, get_free_tcp_port, mkdir
from localstack.utils.functions import run_safe
from localstack.utils.net import wait_for_port_closed
from localstack.utils.objects import singleton_factory
from localstack.utils.platform import Arch, get_arch
from localstack.utils.run import FuncThread, run
from localstack.utils.serving import Server
from localstack.utils.sync import retry, synchronized

LOG = logging.getLogger(__name__)
RESTART_LOCK = threading.RLock()


def _log_listener(line, **_kwargs):
    LOG.debug(line.rstrip())


class DynamodbServer(Server):
    db_path: str | None
    heap_size: str

    delay_transient_statuses: bool
    optimize_db_before_startup: bool
    share_db: bool
    cors: str | None

    proxy: AwsRequestProxy

    def __init__(
        self,
        port: int | None = None,
        host: str = "localhost",
        db_path: str | None = None,
    ) -> None:
        """
        Creates a DynamoDB server from the local configuration.

        :param port: optional, the port to start the server on (defaults to a random port)
        :param host: localhost by default
        :param db_path: path to the persistence state files used by the DynamoDB Local process
        """

        port = port or get_free_tcp_port()
        super().__init__(port, host)

        self.db_path = (
            f"{config.dirs.data}/dynamodb" if not db_path and config.dirs.data else db_path
        )

        # the DYNAMODB_IN_MEMORY variable takes precedence and will set the DB path to None which forces inMemory=true
        if is_env_true("DYNAMODB_IN_MEMORY"):
            # note: with DYNAMODB_IN_MEMORY we do not support persistence
            self.db_path = None

        if self.db_path:
            self.db_path = os.path.abspath(self.db_path)

        self.heap_size = config.DYNAMODB_HEAP_SIZE
        self.delay_transient_statuses = is_env_true("DYNAMODB_DELAY_TRANSIENT_STATUSES")
        self.optimize_db_before_startup = is_env_true("DYNAMODB_OPTIMIZE_DB_BEFORE_STARTUP")
        self.share_db = is_env_true("DYNAMODB_SHARE_DB")
        self.cors = os.getenv("DYNAMODB_CORS", None)
        self.proxy = AwsRequestProxy(self.url)

    @staticmethod
    @singleton_factory
    def get() -> "DynamodbServer":
        return DynamodbServer(config.DYNAMODB_LOCAL_PORT)

    @synchronized(lock=RESTART_LOCK)
    def start_dynamodb(self) -> bool:
        """Start the DynamoDB server."""

        # We want this method to be idempotent.
        if self.is_running() and self.is_up():
            return True

        # For the v2 provider, the DynamodbServer has been made a singleton. Yet, the Server abstraction is modelled
        # after threading.Thread, where Start -> Stop -> Start is not allowed. This flow happens during state resets.
        # The following is a workaround that permits this flow
        self._started.clear()
        self._stopped.clear()

        # Note: when starting the server, we had a flag for wiping the assets directory before the actual start.
        # This behavior was needed in some particular cases:
        # - pod load with some assets already lying in the asset folder
        # - ...
        # The cleaning is now done via the reset endpoint
        if self.db_path:
            mkdir(self.db_path)

        started = self.start()
        self.wait_for_dynamodb()
        return started

    @synchronized(lock=RESTART_LOCK)
    def stop_dynamodb(self) -> None:
        """Stop the DynamoDB server."""
        import psutil

        if self._thread is None:
            return
        self._thread.auto_restart = False
        self.shutdown()
        self.join(timeout=10)
        try:
            wait_for_port_closed(self.port, sleep_time=0.8, retries=10)
        except Exception:
            LOG.warning(
                "DynamoDB server port %s (%s) unexpectedly still open; running processes: %s",
                self.port,
                self._thread,
                run(["ps", "aux"]),
            )

            # attempt to terminate/kill the process manually
            server_pid = self._thread.process.pid  # noqa
            LOG.info("Attempting to kill DynamoDB process %s", server_pid)
            process = psutil.Process(server_pid)
            run_safe(process.terminate)
            run_safe(process.kill)
            wait_for_port_closed(self.port, sleep_time=0.5, retries=8)

    @property
    def in_memory(self) -> bool:
        return self.db_path is None

    @property
    def jar_path(self) -> str:
        return f"{dynamodblocal_package.get_installed_dir()}/DynamoDBLocal.jar"

    @property
    def library_path(self) -> str:
        return f"{dynamodblocal_package.get_installed_dir()}/DynamoDBLocal_lib"

    def _get_java_vm_options(self) -> list[str]:
        # Workaround for JVM SIGILL crash on Apple Silicon M4
        # See https://bugs.openjdk.org/browse/JDK-8345296
        # To be removed after Java is bumped to 17.0.15+ and 21.0.7+
        return ["-XX:UseSVE=0"] if Arch.arm64 == get_arch() else []

    def _create_shell_command(self) -> list[str]:
        cmd = [
            "java",
            *self._get_java_vm_options(),
            "-Xmx%s" % self.heap_size,
            f"-javaagent:{dynamodblocal_package.get_installer().get_ddb_agent_jar_path()}",
            f"-Djava.library.path={self.library_path}",
            "-jar",
            self.jar_path,
        ]
        parameters = []

        parameters.extend(["-port", str(self.port)])
        if self.in_memory:
            parameters.append("-inMemory")
        if self.db_path:
            parameters.extend(["-dbPath", self.db_path])
        if self.delay_transient_statuses:
            parameters.extend(["-delayTransientStatuses"])
        if self.optimize_db_before_startup:
            parameters.extend(["-optimizeDbBeforeStartup"])
        if self.share_db:
            parameters.extend(["-sharedDb"])

        return cmd + parameters

    def do_start_thread(self) -> FuncThread:
        dynamodblocal_installer = dynamodblocal_package.get_installer()
        dynamodblocal_installer.install()

        cmd = self._create_shell_command()
        env_vars = {
            **dynamodblocal_installer.get_java_env_vars(),
            "DDB_LOCAL_TELEMETRY": "0",
        }

        LOG.debug("Starting DynamoDB Local: %s", cmd)
        t = ShellCommandThread(
            cmd,
            strip_color=True,
            log_listener=_log_listener,
            auto_restart=True,
            name="dynamodb-local",
            env_vars=env_vars,
        )
        TMP_THREADS.append(t)
        t.start()
        return t

    def check_dynamodb(self, expect_shutdown: bool = False) -> None:
        """Checks if DynamoDB server is up"""
        out = None

        try:
            self.wait_is_up()
            out = connect_externally_to(
                endpoint_url=self.url,
                aws_access_key_id=DEFAULT_AWS_ACCOUNT_ID,
                aws_secret_access_key=DEFAULT_AWS_ACCOUNT_ID,
            ).dynamodb.list_tables()
        except Exception:
            LOG.exception("DynamoDB health check failed")
        if expect_shutdown:
            assert out is None
        else:
            assert isinstance(out["TableNames"], list)

    def wait_for_dynamodb(self) -> None:
        retry(self.check_dynamodb, sleep=0.4, retries=10)

1	import logging	1✔
2	import os	1✔
3	import threading	1✔
4
5	from localstack import config	1✔
6	from localstack.aws.connect import connect_externally_to	1✔
7	from localstack.aws.forwarder import AwsRequestProxy	1✔
8	from localstack.config import is_env_true	1✔
9	from localstack.constants import DEFAULT_AWS_ACCOUNT_ID	1✔
10	from localstack.services.dynamodb.packages import dynamodblocal_package	1✔
11	from localstack.utils.common import TMP_THREADS, ShellCommandThread, get_free_tcp_port, mkdir	1✔
12	from localstack.utils.functions import run_safe	1✔
13	from localstack.utils.net import wait_for_port_closed	1✔
14	from localstack.utils.objects import singleton_factory	1✔
15	from localstack.utils.platform import Arch, get_arch	1✔
16	from localstack.utils.run import FuncThread, run	1✔
17	from localstack.utils.serving import Server	1✔
18	from localstack.utils.sync import retry, synchronized	1✔
19
20	LOG = logging.getLogger(__name__)	1✔
21	RESTART_LOCK = threading.RLock()	1✔
22
23
24	def _log_listener(line, **_kwargs):	1✔
25	LOG.debug(line.rstrip())	1✔
26
27
28	class DynamodbServer(Server):	1✔
29	db_path: str \| None	1✔
30	heap_size: str	1✔
31
32	delay_transient_statuses: bool	1✔
33	optimize_db_before_startup: bool	1✔
34	share_db: bool	1✔
35	cors: str \| None	1✔
36
37	proxy: AwsRequestProxy	1✔
38
39	def __init__(	1✔
40	self,
41	port: int \| None = None,
42	host: str = "localhost",
43	db_path: str \| None = None,
44	) -> None:
45	"""
46	Creates a DynamoDB server from the local configuration.
47
48	:param port: optional, the port to start the server on (defaults to a random port)
49	:param host: localhost by default
50	:param db_path: path to the persistence state files used by the DynamoDB Local process
51	"""
52
53	port = port or get_free_tcp_port()	1✔
54	super().__init__(port, host)	1✔
55
56	self.db_path = (	1✔
57	f"{config.dirs.data}/dynamodb" if not db_path and config.dirs.data else db_path
58	)
59
60	# the DYNAMODB_IN_MEMORY variable takes precedence and will set the DB path to None which forces inMemory=true
61	if is_env_true("DYNAMODB_IN_MEMORY"):	1✔
62	# note: with DYNAMODB_IN_MEMORY we do not support persistence
UNCOV 63	self.db_path = None	×
64
65	if self.db_path:	1✔
66	self.db_path = os.path.abspath(self.db_path)	1✔
67
68	self.heap_size = config.DYNAMODB_HEAP_SIZE	1✔
69	self.delay_transient_statuses = is_env_true("DYNAMODB_DELAY_TRANSIENT_STATUSES")	1✔
70	self.optimize_db_before_startup = is_env_true("DYNAMODB_OPTIMIZE_DB_BEFORE_STARTUP")	1✔
71	self.share_db = is_env_true("DYNAMODB_SHARE_DB")	1✔
72	self.cors = os.getenv("DYNAMODB_CORS", None)	1✔
73	self.proxy = AwsRequestProxy(self.url)	1✔
74
75	@staticmethod	1✔
76	@singleton_factory	1✔
77	def get() -> "DynamodbServer":	1✔
78	return DynamodbServer(config.DYNAMODB_LOCAL_PORT)	1✔
79
80	@synchronized(lock=RESTART_LOCK)	1✔
81	def start_dynamodb(self) -> bool:	1✔
82	"""Start the DynamoDB server."""
83
84	# We want this method to be idempotent.
85	if self.is_running() and self.is_up():	1✔
86	return True	1✔
87
88	# For the v2 provider, the DynamodbServer has been made a singleton. Yet, the Server abstraction is modelled
89	# after threading.Thread, where Start -> Stop -> Start is not allowed. This flow happens during state resets.
90	# The following is a workaround that permits this flow
91	self._started.clear()	1✔
92	self._stopped.clear()	1✔
93
94	# Note: when starting the server, we had a flag for wiping the assets directory before the actual start.
95	# This behavior was needed in some particular cases:
96	# - pod load with some assets already lying in the asset folder
97	# - ...
98	# The cleaning is now done via the reset endpoint
99	if self.db_path:	1✔
100	mkdir(self.db_path)	1✔
101
102	started = self.start()	1✔
103	self.wait_for_dynamodb()	1✔
104	return started	1✔
105
106	@synchronized(lock=RESTART_LOCK)	1✔
107	def stop_dynamodb(self) -> None:	1✔
108	"""Stop the DynamoDB server."""
UNCOV 109	import psutil	×
110
UNCOV 111	if self._thread is None:	×
112	return	×
113	self._thread.auto_restart = False	×
114	self.shutdown()	×
115	self.join(timeout=10)	×
116	try:	×
117	wait_for_port_closed(self.port, sleep_time=0.8, retries=10)	×
118	except Exception:	×
119	LOG.warning(	×
120	"DynamoDB server port %s (%s) unexpectedly still open; running processes: %s",
121	self.port,
122	self._thread,
123	run(["ps", "aux"]),
124	)
125
126	# attempt to terminate/kill the process manually
UNCOV 127	server_pid = self._thread.process.pid # noqa	×
128	LOG.info("Attempting to kill DynamoDB process %s", server_pid)	×
129	process = psutil.Process(server_pid)	×
130	run_safe(process.terminate)	×
131	run_safe(process.kill)	×
132	wait_for_port_closed(self.port, sleep_time=0.5, retries=8)	×
133
134	@property	1✔
135	def in_memory(self) -> bool:	1✔
136	return self.db_path is None	1✔
137
138	@property	1✔
139	def jar_path(self) -> str:	1✔
140	return f"{dynamodblocal_package.get_installed_dir()}/DynamoDBLocal.jar"	1✔
141
142	@property	1✔
143	def library_path(self) -> str:	1✔
144	return f"{dynamodblocal_package.get_installed_dir()}/DynamoDBLocal_lib"	1✔
145
146	def _get_java_vm_options(self) -> list[str]:	1✔
147	# Workaround for JVM SIGILL crash on Apple Silicon M4
148	# See https://bugs.openjdk.org/browse/JDK-8345296
149	# To be removed after Java is bumped to 17.0.15+ and 21.0.7+
150	return ["-XX:UseSVE=0"] if Arch.arm64 == get_arch() else []	1✔
151
152	def _create_shell_command(self) -> list[str]:	1✔
153	cmd = [	1✔
154	"java",
155	*self._get_java_vm_options(),
156	"-Xmx%s" % self.heap_size,
157	f"-javaagent:{dynamodblocal_package.get_installer().get_ddb_agent_jar_path()}",
158	f"-Djava.library.path={self.library_path}",
159	"-jar",
160	self.jar_path,
161	]
162	parameters = []	1✔
163
164	parameters.extend(["-port", str(self.port)])	1✔
165	if self.in_memory:	1✔
UNCOV 166	parameters.append("-inMemory")	×
167	if self.db_path:	1✔
168	parameters.extend(["-dbPath", self.db_path])	1✔
169	if self.delay_transient_statuses:	1✔
UNCOV 170	parameters.extend(["-delayTransientStatuses"])	×
171	if self.optimize_db_before_startup:	1✔
UNCOV 172	parameters.extend(["-optimizeDbBeforeStartup"])	×
173	if self.share_db:	1✔
UNCOV 174	parameters.extend(["-sharedDb"])	×
175
176	return cmd + parameters	1✔
177
178	def do_start_thread(self) -> FuncThread:	1✔
179	dynamodblocal_installer = dynamodblocal_package.get_installer()	1✔
180	dynamodblocal_installer.install()	1✔
181
182	cmd = self._create_shell_command()	1✔
183	env_vars = {	1✔
184	**dynamodblocal_installer.get_java_env_vars(),
185	"DDB_LOCAL_TELEMETRY": "0",
186	}
187
188	LOG.debug("Starting DynamoDB Local: %s", cmd)	1✔
189	t = ShellCommandThread(	1✔
190	cmd,
191	strip_color=True,
192	log_listener=_log_listener,
193	auto_restart=True,
194	name="dynamodb-local",
195	env_vars=env_vars,
196	)
197	TMP_THREADS.append(t)	1✔
198	t.start()	1✔
199	return t	1✔
200
201	def check_dynamodb(self, expect_shutdown: bool = False) -> None:	1✔
202	"""Checks if DynamoDB server is up"""
203	out = None	1✔
204
205	try:	1✔
206	self.wait_is_up()	1✔
207	out = connect_externally_to(	1✔
208	endpoint_url=self.url,
209	aws_access_key_id=DEFAULT_AWS_ACCOUNT_ID,
210	aws_secret_access_key=DEFAULT_AWS_ACCOUNT_ID,
211	).dynamodb.list_tables()
UNCOV 212	except Exception:	×
UNCOV 213	LOG.exception("DynamoDB health check failed")	×
214	if expect_shutdown:	1✔
UNCOV 215	assert out is None	×
216	else:
217	assert isinstance(out["TableNames"], list)	1✔
218
219	def wait_for_dynamodb(self) -> None:	1✔
220	retry(self.check_dynamodb, sleep=0.4, retries=10)	1✔

localstack / localstack / ea65363c-1e20-4f77-ac7a-bf15452fc5ae

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