17144436094

Committed 21 Aug 2025 11:28PM UTC coverage: 86.843% (-0.03%) from 86.876%

Build # 17144436094

Build Type

push

github

Committed by

web-flow

Commit Message

APIGW: internalize DeleteIntegrationResponse (#13046)

Coverage Stats

40 of 45 new or added lines in 1 file covered. (88.89%)

235 existing lines in 11 files now uncovered.

67068 of 77229 relevant lines covered (86.84%)

0.87 hits per line

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

80.51

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

import functools
import logging
import platform
import random
from typing import Optional, Union

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

localstack / localstack / 17144436094

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