20014894797

Committed 06 Dec 2025 01:32PM UTC coverage: 86.869% (-0.04%) from 86.904%

Build # 20014894797

Build Type

push

github

Committed by

web-flow

Commit Message

CFn: handle updates with empty resource properties (#13471)

Run Details

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

419 existing lines in 20 files now uncovered.

69891 of 80456 relevant lines covered (86.87%)

0.87 hits per line

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

80.34

/localstack-core/localstack/utils/docker_utils.py

import functools
import logging
import platform
import random

from localstack import config
from localstack.constants import DEFAULT_VOLUME_DIR, DOCKER_IMAGE_NAME
from localstack.utils.collections import ensure_list
from localstack.utils.container_utils.container_client import (
    ContainerClient,
    DockerNotAvailable,
    PortMappings,
    VolumeInfo,
)
from localstack.utils.net import IntOrPort, Port, PortNotAvailableException, PortRange
from localstack.utils.objects import singleton_factory
from localstack.utils.strings import to_str

LOG = logging.getLogger(__name__)

# port range instance used to reserve Docker container ports
PORT_START = 0
PORT_END = 65536
RANDOM_PORT_START = 1024
RANDOM_PORT_END = 65536


def is_docker_sdk_installed() -> bool:
    try:
        import docker  # noqa: F401

        return True
    except ModuleNotFoundError:
        return False


def create_docker_client() -> ContainerClient:
    # never use the sdk client if it is not installed or not in docker - too risky for wrong version
    if config.LEGACY_DOCKER_CLIENT or not is_docker_sdk_installed() or not config.is_in_docker:
        from localstack.utils.container_utils.docker_cmd_client import CmdDockerClient

        LOG.debug(
            "Using CmdDockerClient. LEGACY_DOCKER_CLIENT: %s, SDK installed: %s",
            config.LEGACY_DOCKER_CLIENT,
            is_docker_sdk_installed(),
        )

        return CmdDockerClient()
    else:
        from localstack.utils.container_utils.docker_sdk_client import SdkDockerClient

        LOG.debug(
            "Using SdkDockerClient. LEGACY_DOCKER_CLIENT: %s, SDK installed: %s",
            config.LEGACY_DOCKER_CLIENT,
            is_docker_sdk_installed(),
        )

        return SdkDockerClient()


def get_current_container_id() -> str:
    """
    Returns the ID of the current container, or raises a ValueError if we're not in docker.

    :return: the ID of the current container
    """
    if not config.is_in_docker:
        raise ValueError("not in docker")

    container_id = platform.node()
    if not container_id:
        raise OSError("no hostname returned to use as container id")

    return container_id


def inspect_current_container_mounts() -> list[VolumeInfo]:
    return DOCKER_CLIENT.inspect_container_volumes(get_current_container_id())


@functools.lru_cache
def get_default_volume_dir_mount() -> VolumeInfo | None:
    """
    Returns the volume information of LocalStack's DEFAULT_VOLUME_DIR (/var/lib/localstack), if mounted,
    else it returns None. If we're not currently in docker a VauleError is raised. in a container, a ValueError is
    raised.

    :return: the volume info of the default volume dir or None
    """
    for volume in inspect_current_container_mounts():
        if volume.destination.rstrip("/") == DEFAULT_VOLUME_DIR:
            return volume

    return None


def get_host_path_for_path_in_docker(path):
    """
    Returns the calculated host location for a given subpath of DEFAULT_VOLUME_DIR inside the localstack container.
    The path **has** to be a subdirectory of DEFAULT_VOLUME_DIR (the dir itself *will not* work).

    :param path: Path to be replaced (subpath of DEFAULT_VOLUME_DIR)
    :return: Path on the host
    """
    if config.is_in_docker and DOCKER_CLIENT.has_docker():
        volume = get_default_volume_dir_mount()

        if volume:
            if volume.type != "bind":
                raise ValueError(
                    f"Mount to {DEFAULT_VOLUME_DIR} needs to be a bind mount for mounting to work"
                )

            if not path.startswith(f"{DEFAULT_VOLUME_DIR}/") and path != DEFAULT_VOLUME_DIR:
                # We should be able to replace something here.
                # if this warning is printed, the usage of this function is probably wrong.
                # Please check if the target path is indeed prefixed by /var/lib/localstack
                # if this happens, mounts may fail
                LOG.warning(
                    "Error while performing automatic host path replacement for path '%s' to source '%s'",
                    path,
                    volume.source,
                )
            else:
                relative_path = path.removeprefix(DEFAULT_VOLUME_DIR)
                result = volume.source + relative_path
                return result
        else:
            raise ValueError(f"No volume mounted to {DEFAULT_VOLUME_DIR}")

    return path


def container_ports_can_be_bound(
    ports: IntOrPort | list[IntOrPort],
    address: str | None = None,
) -> bool:
    """Determine whether a given list of ports can be bound by Docker containers

    :param ports: single port or list of ports to check
    :return: True iff all ports can be bound
    """
    port_mappings = PortMappings(bind_host=address or "")
    ports = ensure_list(ports)
    for port in ports:
        port = Port.wrap(port)
        port_mappings.add(port.port, port.port, protocol=port.protocol)
    try:
        result = DOCKER_CLIENT.run_container(
            _get_ports_check_docker_image(),
            entrypoint="sh",
            command=["-c", "echo test123"],
            ports=port_mappings,
            remove=True,
        )
    except DockerNotAvailable as e:
        LOG.warning("Cannot perform port check because Docker is not available.")
        raise e
    except Exception as e:
        if "port is already allocated" not in str(e) and "address already in use" not in str(e):
            LOG.warning(
                "Unexpected error when attempting to determine container port status",
                exc_info=LOG.isEnabledFor(logging.DEBUG),
            )
        return False
    # TODO(srw): sometimes the command output from the docker container is "None", particularly when this function is
    #  invoked multiple times consecutively. Work out why.
    if to_str(result[0] or "").strip() != "test123":
        LOG.warning(
            "Unexpected output when attempting to determine container port status: %s", result
        )
    return True


class _DockerPortRange(PortRange):
    """
    PortRange which checks whether the port can be bound on the host instead of inside the container.
    """

    def _port_can_be_bound(self, port: IntOrPort) -> bool:
        return container_ports_can_be_bound(port)


reserved_docker_ports = _DockerPortRange(PORT_START, PORT_END)


def is_port_available_for_containers(port: IntOrPort) -> bool:
    """Check whether the given port can be bound by containers and is not currently reserved"""
    return not is_container_port_reserved(port) and container_ports_can_be_bound(port)


def reserve_container_port(port: IntOrPort, duration: int = None):
    """Reserve the given container port for a short period of time"""
    reserved_docker_ports.reserve_port(port, duration=duration)


def is_container_port_reserved(port: IntOrPort) -> bool:
    """Return whether the given container port is currently reserved"""
    port = Port.wrap(port)
    return reserved_docker_ports.is_port_reserved(port)


def reserve_available_container_port(
    duration: int = None,
    port_start: int = None,
    port_end: int = None,
    protocol: str = None,
) -> int:
    """
    Determine a free port within the given port range that can be bound by a Docker container, and reserve
    the port for the given number of seconds

    :param duration: the number of seconds to reserve the port (default: ~6 seconds)
    :param port_start: the start of the port range to check (default: 1024)
    :param port_end: the end of the port range to check (default: 65536)
    :param protocol: the network protocol (default: tcp)
    :return: a random port
    :raises PortNotAvailableException: if no port is available within the given range
    """

    protocol = protocol or "tcp"

    def _random_port():
        port = None
        while not port or reserved_docker_ports.is_port_reserved(port):
            port_number = random.randint(
                RANDOM_PORT_START if port_start is None else port_start,
                RANDOM_PORT_END if port_end is None else port_end,
            )
            port = Port(port=port_number, protocol=protocol)
        return port

    retries = 10
    for i in range(retries):
        port = _random_port()
        try:
            reserve_container_port(port, duration=duration)
            return port.port
        except PortNotAvailableException as e:
            LOG.debug("Could not bind port %s, trying the next one: %s", port, e)

    raise PortNotAvailableException(
        f"Unable to determine available Docker container port after {retries} retries"
    )


@singleton_factory
def _get_ports_check_docker_image() -> str:
    """
    Determine the Docker image to use for Docker port availability checks.
    Uses either PORTS_CHECK_DOCKER_IMAGE (if configured), or otherwise inspects the running container's image.
    """
    if config.PORTS_CHECK_DOCKER_IMAGE:
        # explicit configuration takes precedence
        return config.PORTS_CHECK_DOCKER_IMAGE
    if not config.is_in_docker:
        # local import to prevent circular imports
        from localstack.utils.bootstrap import get_docker_image_to_start

        # Use whatever image the user is trying to run LocalStack with, since they either have
        # it already, or need it by definition to start LocalStack.
        return get_docker_image_to_start()
    try:
        # inspect the running container to determine the image
        container = DOCKER_CLIENT.inspect_container(get_current_container_id())
        return container["Config"]["Image"]
    except Exception:
        # fall back to using the default Docker image
        return DOCKER_IMAGE_NAME


DOCKER_CLIENT: ContainerClient = create_docker_client()

1	import functools	1✔
2	import logging	1✔
3	import platform	1✔
4	import random	1✔
5
6	from localstack import config	1✔
7	from localstack.constants import DEFAULT_VOLUME_DIR, DOCKER_IMAGE_NAME	1✔
8	from localstack.utils.collections import ensure_list	1✔
9	from localstack.utils.container_utils.container_client import (	1✔
10	ContainerClient,
11	DockerNotAvailable,
12	PortMappings,
13	VolumeInfo,
14	)
15	from localstack.utils.net import IntOrPort, Port, PortNotAvailableException, PortRange	1✔
16	from localstack.utils.objects import singleton_factory	1✔
17	from localstack.utils.strings import to_str	1✔
18
19	LOG = logging.getLogger(__name__)	1✔
20
21	# port range instance used to reserve Docker container ports
22	PORT_START = 0	1✔
23	PORT_END = 65536	1✔
24	RANDOM_PORT_START = 1024	1✔
25	RANDOM_PORT_END = 65536	1✔
26
27
28	def is_docker_sdk_installed() -> bool:	1✔
29	try:	1✔
30	import docker # noqa: F401	1✔
31
32	return True	1✔
33	except ModuleNotFoundError:	×
34	return False	×
35
36
37	def create_docker_client() -> ContainerClient:	1✔
38	# never use the sdk client if it is not installed or not in docker - too risky for wrong version
39	if config.LEGACY_DOCKER_CLIENT or not is_docker_sdk_installed() or not config.is_in_docker:	1✔
40	from localstack.utils.container_utils.docker_cmd_client import CmdDockerClient	1✔
41
42	LOG.debug(	1✔
43	"Using CmdDockerClient. LEGACY_DOCKER_CLIENT: %s, SDK installed: %s",
44	config.LEGACY_DOCKER_CLIENT,
45	is_docker_sdk_installed(),
46	)
47
48	return CmdDockerClient()	1✔
49	else:
50	from localstack.utils.container_utils.docker_sdk_client import SdkDockerClient	1✔
51
52	LOG.debug(	1✔
53	"Using SdkDockerClient. LEGACY_DOCKER_CLIENT: %s, SDK installed: %s",
54	config.LEGACY_DOCKER_CLIENT,
55	is_docker_sdk_installed(),
56	)
57
58	return SdkDockerClient()	1✔
59
60
61	def get_current_container_id() -> str:	1✔
62	"""
63	Returns the ID of the current container, or raises a ValueError if we're not in docker.
64
65	:return: the ID of the current container
66	"""
67	if not config.is_in_docker:	1✔
68	raise ValueError("not in docker")	×
69
70	container_id = platform.node()	1✔
71	if not container_id:	1✔
72	raise OSError("no hostname returned to use as container id")	×
73
74	return container_id	1✔
75
76
77	def inspect_current_container_mounts() -> list[VolumeInfo]:	1✔
78	return DOCKER_CLIENT.inspect_container_volumes(get_current_container_id())	1✔
79
80
81	@functools.lru_cache	1✔
82	def get_default_volume_dir_mount() -> VolumeInfo \| None:	1✔
83	"""
84	Returns the volume information of LocalStack's DEFAULT_VOLUME_DIR (/var/lib/localstack), if mounted,
85	else it returns None. If we're not currently in docker a VauleError is raised. in a container, a ValueError is
86	raised.
87
88	:return: the volume info of the default volume dir or None
89	"""
90	for volume in inspect_current_container_mounts():	1✔
91	if volume.destination.rstrip("/") == DEFAULT_VOLUME_DIR:	1✔
92	return volume	1✔
93
94	return None	×
95
96
97	def get_host_path_for_path_in_docker(path):	1✔
98	"""
99	Returns the calculated host location for a given subpath of DEFAULT_VOLUME_DIR inside the localstack container.
100	The path has to be a subdirectory of DEFAULT_VOLUME_DIR (the dir itself will not work).
101
102	:param path: Path to be replaced (subpath of DEFAULT_VOLUME_DIR)
103	:return: Path on the host
104	"""
105	if config.is_in_docker and DOCKER_CLIENT.has_docker():	1✔
106	volume = get_default_volume_dir_mount()	1✔
107
108	if volume:	1✔
109	if volume.type != "bind":	1✔
110	raise ValueError(	×
111	f"Mount to {DEFAULT_VOLUME_DIR} needs to be a bind mount for mounting to work"
112	)
113
114	if not path.startswith(f"{DEFAULT_VOLUME_DIR}/") and path != DEFAULT_VOLUME_DIR:	1✔
115	# We should be able to replace something here.
116	# if this warning is printed, the usage of this function is probably wrong.
117	# Please check if the target path is indeed prefixed by /var/lib/localstack
118	# if this happens, mounts may fail
119	LOG.warning(	1✔
120	"Error while performing automatic host path replacement for path '%s' to source '%s'",
121	path,
122	volume.source,
123	)
124	else:
125	relative_path = path.removeprefix(DEFAULT_VOLUME_DIR)	1✔
126	result = volume.source + relative_path	1✔
127	return result	1✔
128	else:
129	raise ValueError(f"No volume mounted to {DEFAULT_VOLUME_DIR}")	×
130
131	return path	1✔
132
133
134	def container_ports_can_be_bound(	1✔
135	ports: IntOrPort \| list[IntOrPort],
136	address: str \| None = None,
137	) -> bool:
138	"""Determine whether a given list of ports can be bound by Docker containers
139
140	:param ports: single port or list of ports to check
141	:return: True iff all ports can be bound
142	"""
143	port_mappings = PortMappings(bind_host=address or "")	1✔
144	ports = ensure_list(ports)	1✔
145	for port in ports:	1✔
146	port = Port.wrap(port)	1✔
147	port_mappings.add(port.port, port.port, protocol=port.protocol)	1✔
148	try:	1✔
149	result = DOCKER_CLIENT.run_container(	1✔
150	_get_ports_check_docker_image(),
151	entrypoint="sh",
152	command=["-c", "echo test123"],
153	ports=port_mappings,
154	remove=True,
155	)
156	except DockerNotAvailable as e:	1✔
157	LOG.warning("Cannot perform port check because Docker is not available.")	×
158	raise e	×
159	except Exception as e:	1✔
160	if "port is already allocated" not in str(e) and "address already in use" not in str(e):	1✔
161	LOG.warning(	1✔
162	"Unexpected error when attempting to determine container port status",
163	exc_info=LOG.isEnabledFor(logging.DEBUG),
164	)
165	return False	1✔
166	# TODO(srw): sometimes the command output from the docker container is "None", particularly when this function is
167	# invoked multiple times consecutively. Work out why.
168	if to_str(result[0] or "").strip() != "test123":	1✔
169	LOG.warning(	×
170	"Unexpected output when attempting to determine container port status: %s", result
171	)
172	return True	1✔
173
174
175	class _DockerPortRange(PortRange):	1✔
176	"""
177	PortRange which checks whether the port can be bound on the host instead of inside the container.
178	"""
179
180	def _port_can_be_bound(self, port: IntOrPort) -> bool:	1✔
181	return container_ports_can_be_bound(port)	1✔
182
183
184	reserved_docker_ports = _DockerPortRange(PORT_START, PORT_END)	1✔
185
186
187	def is_port_available_for_containers(port: IntOrPort) -> bool:	1✔
188	"""Check whether the given port can be bound by containers and is not currently reserved"""
189	return not is_container_port_reserved(port) and container_ports_can_be_bound(port)	1✔
190
191
192	def reserve_container_port(port: IntOrPort, duration: int = None):	1✔
193	"""Reserve the given container port for a short period of time"""
194	reserved_docker_ports.reserve_port(port, duration=duration)	1✔
195
196
197	def is_container_port_reserved(port: IntOrPort) -> bool:	1✔
198	"""Return whether the given container port is currently reserved"""
199	port = Port.wrap(port)	1✔
200	return reserved_docker_ports.is_port_reserved(port)	1✔
201
202
203	def reserve_available_container_port(	1✔
204	duration: int = None,
205	port_start: int = None,
206	port_end: int = None,
207	protocol: str = None,
208	) -> int:
209	"""
210	Determine a free port within the given port range that can be bound by a Docker container, and reserve
211	the port for the given number of seconds
212
213	:param duration: the number of seconds to reserve the port (default: ~6 seconds)
214	:param port_start: the start of the port range to check (default: 1024)
215	:param port_end: the end of the port range to check (default: 65536)
216	:param protocol: the network protocol (default: tcp)
217	:return: a random port
218	:raises PortNotAvailableException: if no port is available within the given range
219	"""
220
221	protocol = protocol or "tcp"	1✔
222
223	def _random_port():	1✔
224	port = None	1✔
225	while not port or reserved_docker_ports.is_port_reserved(port):	1✔
226	port_number = random.randint(	1✔
227	RANDOM_PORT_START if port_start is None else port_start,
228	RANDOM_PORT_END if port_end is None else port_end,
229	)
230	port = Port(port=port_number, protocol=protocol)	1✔
231	return port	1✔
232
233	retries = 10	1✔
234	for i in range(retries):	1✔
235	port = _random_port()	1✔
236	try:	1✔
237	reserve_container_port(port, duration=duration)	1✔
238	return port.port	1✔
UNCOV 239	except PortNotAvailableException as e:	×
UNCOV 240	LOG.debug("Could not bind port %s, trying the next one: %s", port, e)	×
241
242	raise PortNotAvailableException(	×
243	f"Unable to determine available Docker container port after {retries} retries"
244	)
245
246
247	@singleton_factory	1✔
248	def _get_ports_check_docker_image() -> str:	1✔
249	"""
250	Determine the Docker image to use for Docker port availability checks.
251	Uses either PORTS_CHECK_DOCKER_IMAGE (if configured), or otherwise inspects the running container's image.
252	"""
253	if config.PORTS_CHECK_DOCKER_IMAGE:	×
254	# explicit configuration takes precedence
255	return config.PORTS_CHECK_DOCKER_IMAGE	×
256	if not config.is_in_docker:	×
257	# local import to prevent circular imports
258	from localstack.utils.bootstrap import get_docker_image_to_start	×
259
260	# Use whatever image the user is trying to run LocalStack with, since they either have
261	# it already, or need it by definition to start LocalStack.
262	return get_docker_image_to_start()	×
263	try:	×
264	# inspect the running container to determine the image
265	container = DOCKER_CLIENT.inspect_container(get_current_container_id())	×
266	return container["Config"]["Image"]	×
267	except Exception:	×
268	# fall back to using the default Docker image
269	return DOCKER_IMAGE_NAME	×
270
271
272	DOCKER_CLIENT: ContainerClient = create_docker_client()	1✔

localstack / localstack / 20014894797

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