20942662173

Committed 12 Jan 2026 04:45PM UTC coverage: 86.905% (-0.03%) from 86.936%

Build # 20942662173

Build Type

push

github

Committed by

web-flow

Commit Message

Allow authenticated pull and push of docker images (#13569)

Run Details

34 of 51 new or added lines in 4 files covered. (66.67%)

247 existing lines in 15 files now uncovered.

70218 of 80799 relevant lines covered (86.9%)

0.87 hits per line

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

71.19

/localstack-core/localstack/services/lambda_/invocation/docker_runtime_executor.py

import dataclasses
import json
import logging
import shutil
import tempfile
import threading
from collections import defaultdict
from collections.abc import Callable
from pathlib import Path

from localstack import config
from localstack.aws.api.lambda_ import Architecture, PackageType, Runtime
from localstack.dns import server as dns_server
from localstack.services.lambda_ import hooks as lambda_hooks
from localstack.services.lambda_.invocation.executor_endpoint import (
    INVOCATION_PORT,
    ExecutorEndpoint,
)
from localstack.services.lambda_.invocation.lambda_models import FunctionVersion
from localstack.services.lambda_.invocation.runtime_executor import (
    ChmodPath,
    LambdaPrebuildContext,
    LambdaRuntimeException,
    RuntimeExecutor,
)
from localstack.services.lambda_.lambda_utils import HINT_LOG
from localstack.services.lambda_.networking import (
    get_all_container_networks_for_lambda,
    get_main_endpoint_from_container,
)
from localstack.services.lambda_.packages import get_runtime_client_path
from localstack.services.lambda_.runtimes import IMAGE_MAPPING
from localstack.utils.container_networking import get_main_container_name
from localstack.utils.container_utils.container_client import (
    BindMount,
    ContainerConfiguration,
    DockerNotAvailable,
    DockerPlatform,
    NoSuchContainer,
    NoSuchImage,
    PortMappings,
    VolumeMappings,
)
from localstack.utils.docker_utils import DOCKER_CLIENT as CONTAINER_CLIENT
from localstack.utils.files import chmod_r, rm_rf
from localstack.utils.net import get_free_tcp_port
from localstack.utils.strings import short_uid, truncate

LOG = logging.getLogger(__name__)

IMAGE_PREFIX = "public.ecr.aws/lambda/"
# IMAGE_PREFIX = "amazon/aws-lambda-"

RAPID_ENTRYPOINT = "/var/rapid/init"

LAMBDA_DOCKERFILE = """FROM {base_img}
COPY init {rapid_entrypoint}
COPY code/ /var/task
"""

PULLED_IMAGES: set[tuple[str, DockerPlatform]] = set()
PULL_LOCKS: dict[tuple[str, DockerPlatform], threading.RLock] = defaultdict(threading.RLock)

HOT_RELOADING_ENV_VARIABLE = "LOCALSTACK_HOT_RELOADING_PATHS"


"""Map AWS Lambda architecture to Docker platform flags. Example: arm64 => linux/arm64"""
ARCHITECTURE_PLATFORM_MAPPING: dict[Architecture, DockerPlatform] = {
    Architecture.x86_64: DockerPlatform.linux_amd64,
    Architecture.arm64: DockerPlatform.linux_arm64,
}


def docker_platform(lambda_architecture: Architecture) -> DockerPlatform | None:
    """
    Convert an AWS Lambda architecture into a Docker platform flag. Examples:
    * docker_platform("x86_64") == "linux/amd64"
    * docker_platform("arm64") == "linux/arm64"

    :param lambda_architecture: the instruction set that the function supports
    :return: Docker platform in the format ``os[/arch[/variant]]`` or None if configured to ignore the architecture
    """
    if config.LAMBDA_IGNORE_ARCHITECTURE:
        return None
    return ARCHITECTURE_PLATFORM_MAPPING[lambda_architecture]


def get_image_name_for_function(function_version: FunctionVersion) -> str:
    return f"localstack/prebuild-lambda-{function_version.id.qualified_arn().replace(':', '_').replace('$', '_').lower()}"


def get_default_image_for_runtime(runtime: Runtime) -> str:
    postfix = IMAGE_MAPPING.get(runtime)
    if not postfix:
        raise ValueError(f"Unsupported runtime {runtime}!")
    return f"{IMAGE_PREFIX}{postfix}"


def _ensure_runtime_image_present(image: str, platform: DockerPlatform) -> None:
    # Pull image for a given platform upon function creation such that invocations do not time out.
    if (image, platform) in PULLED_IMAGES:
        return
    # use a lock to avoid concurrent pulling of the same image
    with PULL_LOCKS[(image, platform)]:
        if (image, platform) in PULLED_IMAGES:
            return
        try:
            CONTAINER_CLIENT.pull_image(image, platform)
            PULLED_IMAGES.add((image, platform))
        except NoSuchImage as e:
            LOG.debug("Unable to pull image %s for runtime executor preparation.", image)
            raise e
        except DockerNotAvailable as e:
            HINT_LOG.error(
                "Failed to pull Docker image because Docker is not available in the LocalStack container "
                "but required to run Lambda functions. Please add the Docker volume mount "
                '"/var/run/docker.sock:/var/run/docker.sock" to your LocalStack startup. '
                "https://docs.localstack.cloud/user-guide/aws/lambda/#docker-not-available"
            )
            raise e


class RuntimeImageResolver:
    """
    Resolves Lambda runtimes to corresponding docker images
    The default behavior resolves based on a prefix (including the repository) and a suffix (per runtime).

    This can be customized via the LAMBDA_RUNTIME_IMAGE_MAPPING config in 2 distinct ways:

    Option A: use a pattern string for the config variable that includes the "<runtime>" string
        e.g. "myrepo/lambda:<runtime>-custom" would resolve the runtime "python3.9" to "myrepo/lambda:python3.9-custom"

    Option B: use a JSON dict string for the config variable, mapping the runtime to the full image name & tag
        e.g. {"python3.9": "myrepo/lambda:python3.9-custom", "python3.8": "myotherrepo/pylambda:3.8"}

        Note that with Option B this will only apply to the runtimes included in the dict.
        All other (non-included) runtimes will fall back to the default behavior.
    """

    _mapping: dict[Runtime, str]
    _default_resolve_fn: Callable[[Runtime], str]

    def __init__(
        self, default_resolve_fn: Callable[[Runtime], str] = get_default_image_for_runtime
    ):
        self._mapping = {}
        self._default_resolve_fn = default_resolve_fn

    def _resolve(self, runtime: Runtime, custom_image_mapping: str = "") -> str:
        if runtime not in IMAGE_MAPPING:
            raise ValueError(f"Unsupported runtime {runtime}")

        if not custom_image_mapping:
            return self._default_resolve_fn(runtime)

        # Option A (pattern string that includes <runtime> to replace)
        if "<runtime>" in custom_image_mapping:
            return custom_image_mapping.replace("<runtime>", runtime)

        # Option B (json dict mapping with fallback)
        try:
            mapping: dict = json.loads(custom_image_mapping)
            # at this point we're loading the whole dict to avoid parsing multiple times
            for k, v in mapping.items():
                if k not in IMAGE_MAPPING:
                    raise ValueError(
                        f"Unsupported runtime ({runtime}) provided in LAMBDA_RUNTIME_IMAGE_MAPPING"
                    )
                self._mapping[k] = v

            if runtime in self._mapping:
                return self._mapping[runtime]

            # fall back to default behavior if the runtime was not present in the custom config
            return self._default_resolve_fn(runtime)

        except Exception:
            LOG.error(
                "Failed to load config from LAMBDA_RUNTIME_IMAGE_MAPPING=%s",
                custom_image_mapping,
            )
            raise  # TODO: validate config at start and prevent startup

    def get_image_for_runtime(self, runtime: Runtime) -> str:
        if runtime not in self._mapping:
            resolved_image = self._resolve(runtime, config.LAMBDA_RUNTIME_IMAGE_MAPPING)
            self._mapping[runtime] = resolved_image

        return self._mapping[runtime]


resolver = RuntimeImageResolver()


def prepare_image(function_version: FunctionVersion, platform: DockerPlatform) -> None:
    if not function_version.config.runtime:
        raise NotImplementedError(
            "Custom images are currently not supported with image prebuilding"
        )

    # create dockerfile
    docker_file = LAMBDA_DOCKERFILE.format(
        base_img=resolver.get_image_for_runtime(function_version.config.runtime),
        rapid_entrypoint=RAPID_ENTRYPOINT,
    )

    code_path = function_version.config.code.get_unzipped_code_location()
    context_path = Path(
        f"{tempfile.gettempdir()}/lambda/prebuild_tmp/{function_version.id.function_name}-{short_uid()}"
    )
    context_path.mkdir(parents=True)
    prebuild_context = LambdaPrebuildContext(
        docker_file_content=docker_file,
        context_path=context_path,
        function_version=function_version,
    )
    lambda_hooks.prebuild_environment_image.run(prebuild_context)
    LOG.debug(
        "Prebuilding image for function %s from context %s and Dockerfile %s",
        function_version.qualified_arn,
        str(prebuild_context.context_path),
        prebuild_context.docker_file_content,
    )
    # save dockerfile
    docker_file_path = prebuild_context.context_path / "Dockerfile"
    with docker_file_path.open(mode="w") as f:
        f.write(prebuild_context.docker_file_content)

    # copy init file
    init_destination_path = prebuild_context.context_path / "init"
    src_init = f"{get_runtime_client_path()}/var/rapid/init"
    shutil.copy(src_init, init_destination_path)
    init_destination_path.chmod(0o755)

    # copy function code
    context_code_path = prebuild_context.context_path / "code"
    shutil.copytree(
        f"{str(code_path)}/",
        str(context_code_path),
        dirs_exist_ok=True,
    )
    # if layers are present, permissions should be 0755
    if prebuild_context.function_version.config.layers:
        chmod_r(str(context_code_path), 0o755)

    try:
        image_name = get_image_name_for_function(function_version)
        CONTAINER_CLIENT.build_image(
            dockerfile_path=str(docker_file_path),
            image_name=image_name,
            platform=platform,
        )
    except Exception as e:
        if LOG.isEnabledFor(logging.DEBUG):
            LOG.exception(
                "Error while building prebuilt lambda image for '%s'",
                function_version.qualified_arn,
            )
        else:
            LOG.error(
                "Error while building prebuilt lambda image for '%s', Error: %s",
                function_version.qualified_arn,
                e,
            )
    finally:
        rm_rf(str(prebuild_context.context_path))


@dataclasses.dataclass
class LambdaContainerConfiguration(ContainerConfiguration):
    copy_folders: list[tuple[str, str]] = dataclasses.field(default_factory=list)


class DockerRuntimeExecutor(RuntimeExecutor):
    ip: str | None
    executor_endpoint: ExecutorEndpoint | None
    container_name: str

    def __init__(self, id: str, function_version: FunctionVersion) -> None:
        super().__init__(id=id, function_version=function_version)
        self.ip = None
        self.executor_endpoint = ExecutorEndpoint(self.id)
        self.container_name = self._generate_container_name()
        LOG.debug("Assigning container name of %s to executor %s", self.container_name, self.id)

    def get_image(self) -> str:
        if not self.function_version.config.runtime:
            raise NotImplementedError("Container images are a Pro feature.")
        return (
            get_image_name_for_function(self.function_version)
            if config.LAMBDA_PREBUILD_IMAGES
            else resolver.get_image_for_runtime(self.function_version.config.runtime)
        )

    def _generate_container_name(self):
        """
        Format <main-container-name>-lambda-<function-name>-<executor-id>
        TODO: make the format configurable
        """
        container_name = "-".join(
            [
                get_main_container_name() or "localstack",
                "lambda",
                self.function_version.id.function_name.lower(),
            ]
        ).replace("_", "-")
        return f"{container_name}-{self.id}"

    def start(self, env_vars: dict[str, str]) -> None:
        self.executor_endpoint.start()
        main_network, *additional_networks = self._get_networks_for_executor()
        container_config = LambdaContainerConfiguration(
            image_name=None,
            name=self.container_name,
            env_vars=env_vars,
            network=main_network,
            entrypoint=RAPID_ENTRYPOINT,
            platform=docker_platform(self.function_version.config.architectures[0]),
            additional_flags=config.LAMBDA_DOCKER_FLAGS,
        )

        if self.function_version.config.package_type == PackageType.Zip:
            if self.function_version.config.code.is_hot_reloading():
                container_config.env_vars[HOT_RELOADING_ENV_VARIABLE] = "/var/task"
                if container_config.volumes is None:
                    container_config.volumes = VolumeMappings()
                container_config.volumes.add(
                    BindMount(
                        str(self.function_version.config.code.get_unzipped_code_location()),
                        "/var/task",
                        read_only=True,
                    )
                )
            else:
                container_config.copy_folders.append(
                    (
                        f"{str(self.function_version.config.code.get_unzipped_code_location())}/.",
                        "/var/task",
                    )
                )

        # always chmod /tmp to 700
        chmod_paths = [ChmodPath(path="/tmp", mode="0700")]

        # set the dns server of the lambda container to the LocalStack container IP
        # the dns server will automatically respond with the right target for transparent endpoint injection
        if config.LAMBDA_DOCKER_DNS:
            # Don't overwrite DNS container config if it is already set (e.g., using LAMBDA_DOCKER_DNS)
            LOG.warning(
                "Container DNS overridden to %s, connection to names pointing to LocalStack, like 'localhost.localstack.cloud' will need additional configuration.",
                config.LAMBDA_DOCKER_DNS,
            )
            container_config.dns = config.LAMBDA_DOCKER_DNS
        else:
            if dns_server.is_server_running():
                # Set the container DNS to LocalStack to resolve localhost.localstack.cloud and
                # enable transparent endpoint injection (Pro image only).
                container_config.dns = self.get_endpoint_from_executor()

        lambda_hooks.start_docker_executor.run(container_config, self.function_version)

        if not container_config.image_name:
            container_config.image_name = self.get_image()
        if config.LAMBDA_DEV_PORT_EXPOSE:
            self.executor_endpoint.container_port = get_free_tcp_port()
            if container_config.ports is None:
                container_config.ports = PortMappings()
            container_config.ports.add(self.executor_endpoint.container_port, INVOCATION_PORT)

        if config.LAMBDA_INIT_DEBUG:
            container_config.entrypoint = "/debug-bootstrap.sh"
            if not container_config.ports:
                container_config.ports = PortMappings()
            container_config.ports.add(config.LAMBDA_INIT_DELVE_PORT, config.LAMBDA_INIT_DELVE_PORT)

        if (
            self.function_version.config.layers
            and not config.LAMBDA_PREBUILD_IMAGES
            and self.function_version.config.package_type == PackageType.Zip
        ):
            # avoid chmod on mounted code paths
            hot_reloading_env = container_config.env_vars.get(HOT_RELOADING_ENV_VARIABLE, "")
            if "/opt" not in hot_reloading_env:
                chmod_paths.append(ChmodPath(path="/opt", mode="0755"))
            if "/var/task" not in hot_reloading_env:
                chmod_paths.append(ChmodPath(path="/var/task", mode="0755"))
        container_config.env_vars["LOCALSTACK_CHMOD_PATHS"] = json.dumps(chmod_paths)

        CONTAINER_CLIENT.create_container_from_config(container_config)
        if (
            not config.LAMBDA_PREBUILD_IMAGES
            or self.function_version.config.package_type != PackageType.Zip
        ):
            CONTAINER_CLIENT.copy_into_container(
                self.container_name, f"{str(get_runtime_client_path())}/.", "/"
            )
            # tiny bit inefficient since we actually overwrite the init, but otherwise the path might not exist
            if config.LAMBDA_INIT_BIN_PATH:
                CONTAINER_CLIENT.copy_into_container(
                    self.container_name, config.LAMBDA_INIT_BIN_PATH, "/var/rapid/init"
                )
            if config.LAMBDA_INIT_DEBUG:
                CONTAINER_CLIENT.copy_into_container(
                    self.container_name, config.LAMBDA_INIT_DELVE_PATH, "/var/rapid/dlv"
                )
                CONTAINER_CLIENT.copy_into_container(
                    self.container_name, config.LAMBDA_INIT_BOOTSTRAP_PATH, "/debug-bootstrap.sh"
                )

        if not config.LAMBDA_PREBUILD_IMAGES:
            # copy_folders should be empty here if package type is not zip
            for source, target in container_config.copy_folders:
                CONTAINER_CLIENT.copy_into_container(self.container_name, source, target)

        if additional_networks:
            for additional_network in additional_networks:
                CONTAINER_CLIENT.connect_container_to_network(
                    additional_network, self.container_name
                )

        CONTAINER_CLIENT.start_container(self.container_name)
        # still using main network as main entrypoint
        self.ip = CONTAINER_CLIENT.get_container_ipv4_for_network(
            container_name_or_id=self.container_name, container_network=main_network
        )
        if config.LAMBDA_DEV_PORT_EXPOSE:
            self.ip = "127.0.0.1"
        self.executor_endpoint.container_address = self.ip

        self.executor_endpoint.wait_for_startup()

    def stop(self) -> None:
        CONTAINER_CLIENT.stop_container(container_name=self.container_name, timeout=5)
        if config.LAMBDA_REMOVE_CONTAINERS:
            CONTAINER_CLIENT.remove_container(container_name=self.container_name)
        try:
            self.executor_endpoint.shutdown()
        except Exception as e:
            LOG.debug(
                "Error while stopping executor endpoint for lambda %s, error: %s",
                self.function_version.qualified_arn,
                e,
            )

    def get_address(self) -> str:
        if not self.ip:
            raise LambdaRuntimeException(f"IP address of executor '{self.id}' unknown")
        return self.ip

    def get_endpoint_from_executor(self) -> str:
        return get_main_endpoint_from_container()

    def _get_networks_for_executor(self) -> list[str]:
        return get_all_container_networks_for_lambda()

    def invoke(self, payload: dict[str, str]):
        LOG.debug(
            "Sending invoke-payload '%s' to executor '%s'",
            truncate(json.dumps(payload), config.LAMBDA_TRUNCATE_STDOUT),
            self.id,
        )
        return self.executor_endpoint.invoke(payload)

    def get_logs(self) -> str:
        try:
            return CONTAINER_CLIENT.get_container_logs(container_name_or_id=self.container_name)
        except NoSuchContainer:
            return "Container was not created"

    @classmethod
    def prepare_version(cls, function_version: FunctionVersion) -> None:
        lambda_hooks.prepare_docker_executor.run(function_version)
        # Trigger the installation of the Lambda runtime-init binary before invocation and
        # cache the result to save time upon every invocation.
        get_runtime_client_path()
        if function_version.config.code:
            function_version.config.code.prepare_for_execution()
            image_name = resolver.get_image_for_runtime(function_version.config.runtime)
            platform = docker_platform(function_version.config.architectures[0])
            _ensure_runtime_image_present(image_name, platform)
            if config.LAMBDA_PREBUILD_IMAGES:
                prepare_image(function_version, platform)

    @classmethod
    def cleanup_version(cls, function_version: FunctionVersion) -> None:
        if config.LAMBDA_PREBUILD_IMAGES:
            # TODO re-enable image cleanup.
            # Enabling it currently deletes image after updates as well
            # It also creates issues when cleanup is concurrently with build
            # probably due to intermediate layers being deleted
            # image_name = get_image_name_for_function(function_version)
            # LOG.debug("Removing image %s after version deletion", image_name)
            # CONTAINER_CLIENT.remove_image(image_name)
            pass

    def get_runtime_endpoint(self) -> str:
        return f"http://{self.get_endpoint_from_executor()}:{config.GATEWAY_LISTEN[0].port}{self.executor_endpoint.get_endpoint_prefix()}"

    @classmethod
    def validate_environment(cls) -> bool:
        if not CONTAINER_CLIENT.has_docker():
            LOG.warning(
                "WARNING: Docker not available in the LocalStack container but required to run Lambda "
                'functions. Please add the Docker volume mount "/var/run/docker.sock:/var/run/docker.sock" to your '
                "LocalStack startup. https://docs.localstack.cloud/user-guide/aws/lambda/#docker-not-available"
            )
            return False
        return True

1	import dataclasses	1✔
2	import json	1✔
3	import logging	1✔
4	import shutil	1✔
5	import tempfile	1✔
6	import threading	1✔
7	from collections import defaultdict	1✔
8	from collections.abc import Callable	1✔
9	from pathlib import Path	1✔
10
11	from localstack import config	1✔
12	from localstack.aws.api.lambda_ import Architecture, PackageType, Runtime	1✔
13	from localstack.dns import server as dns_server	1✔
14	from localstack.services.lambda_ import hooks as lambda_hooks	1✔
15	from localstack.services.lambda_.invocation.executor_endpoint import (	1✔
16	INVOCATION_PORT,
17	ExecutorEndpoint,
18	)
19	from localstack.services.lambda_.invocation.lambda_models import FunctionVersion	1✔
20	from localstack.services.lambda_.invocation.runtime_executor import (	1✔
21	ChmodPath,
22	LambdaPrebuildContext,
23	LambdaRuntimeException,
24	RuntimeExecutor,
25	)
26	from localstack.services.lambda_.lambda_utils import HINT_LOG	1✔
27	from localstack.services.lambda_.networking import (	1✔
28	get_all_container_networks_for_lambda,
29	get_main_endpoint_from_container,
30	)
31	from localstack.services.lambda_.packages import get_runtime_client_path	1✔
32	from localstack.services.lambda_.runtimes import IMAGE_MAPPING	1✔
33	from localstack.utils.container_networking import get_main_container_name	1✔
34	from localstack.utils.container_utils.container_client import (	1✔
35	BindMount,
36	ContainerConfiguration,
37	DockerNotAvailable,
38	DockerPlatform,
39	NoSuchContainer,
40	NoSuchImage,
41	PortMappings,
42	VolumeMappings,
43	)
44	from localstack.utils.docker_utils import DOCKER_CLIENT as CONTAINER_CLIENT	1✔
45	from localstack.utils.files import chmod_r, rm_rf	1✔
46	from localstack.utils.net import get_free_tcp_port	1✔
47	from localstack.utils.strings import short_uid, truncate	1✔
48
49	LOG = logging.getLogger(__name__)	1✔
50
51	IMAGE_PREFIX = "public.ecr.aws/lambda/"	1✔
52	# IMAGE_PREFIX = "amazon/aws-lambda-"
53
54	RAPID_ENTRYPOINT = "/var/rapid/init"	1✔
55
56	LAMBDA_DOCKERFILE = """FROM {base_img}	1✔
57	COPY init {rapid_entrypoint}
58	COPY code/ /var/task
59	"""
60
61	PULLED_IMAGES: set[tuple[str, DockerPlatform]] = set()	1✔
62	PULL_LOCKS: dict[tuple[str, DockerPlatform], threading.RLock] = defaultdict(threading.RLock)	1✔
63
64	HOT_RELOADING_ENV_VARIABLE = "LOCALSTACK_HOT_RELOADING_PATHS"	1✔
65
66
67	"""Map AWS Lambda architecture to Docker platform flags. Example: arm64 => linux/arm64"""	1✔
68	ARCHITECTURE_PLATFORM_MAPPING: dict[Architecture, DockerPlatform] = {	1✔
69	Architecture.x86_64: DockerPlatform.linux_amd64,
70	Architecture.arm64: DockerPlatform.linux_arm64,
71	}
72
73
74	def docker_platform(lambda_architecture: Architecture) -> DockerPlatform \| None:	1✔
75	"""
76	Convert an AWS Lambda architecture into a Docker platform flag. Examples:
77	* docker_platform("x86_64") == "linux/amd64"
78	* docker_platform("arm64") == "linux/arm64"
79
80	:param lambda_architecture: the instruction set that the function supports
81	:return: Docker platform in the format ``os[/arch[/variant]]`` or None if configured to ignore the architecture
82	"""
83	if config.LAMBDA_IGNORE_ARCHITECTURE:	1✔
84	return None	1✔
85	return ARCHITECTURE_PLATFORM_MAPPING[lambda_architecture]	1✔
86
87
88	def get_image_name_for_function(function_version: FunctionVersion) -> str:	1✔
UNCOV 89	return f"localstack/prebuild-lambda-{function_version.id.qualified_arn().replace(':', '_').replace('$', '_').lower()}"	×
90
91
92	def get_default_image_for_runtime(runtime: Runtime) -> str:	1✔
93	postfix = IMAGE_MAPPING.get(runtime)	1✔
94	if not postfix:	1✔
UNCOV 95	raise ValueError(f"Unsupported runtime {runtime}!")	×
96	return f"{IMAGE_PREFIX}{postfix}"	1✔
97
98
99	def _ensure_runtime_image_present(image: str, platform: DockerPlatform) -> None:	1✔
100	# Pull image for a given platform upon function creation such that invocations do not time out.
101	if (image, platform) in PULLED_IMAGES:	1✔
102	return	1✔
103	# use a lock to avoid concurrent pulling of the same image
104	with PULL_LOCKS[(image, platform)]:	1✔
105	if (image, platform) in PULLED_IMAGES:	1✔
UNCOV 106	return	×
107	try:	1✔
108	CONTAINER_CLIENT.pull_image(image, platform)	1✔
109	PULLED_IMAGES.add((image, platform))	1✔
UNCOV 110	except NoSuchImage as e:	×
UNCOV 111	LOG.debug("Unable to pull image %s for runtime executor preparation.", image)	×
UNCOV 112	raise e	×
113	except DockerNotAvailable as e:	×
114	HINT_LOG.error(	×
115	"Failed to pull Docker image because Docker is not available in the LocalStack container "
116	"but required to run Lambda functions. Please add the Docker volume mount "
117	'"/var/run/docker.sock:/var/run/docker.sock" to your LocalStack startup. '
118	"https://docs.localstack.cloud/user-guide/aws/lambda/#docker-not-available"
119	)
UNCOV 120	raise e	×
121
122
123	class RuntimeImageResolver:	1✔
124	"""
125	Resolves Lambda runtimes to corresponding docker images
126	The default behavior resolves based on a prefix (including the repository) and a suffix (per runtime).
127
128	This can be customized via the LAMBDA_RUNTIME_IMAGE_MAPPING config in 2 distinct ways:
129
130	Option A: use a pattern string for the config variable that includes the "<runtime>" string
131	e.g. "myrepo/lambda:<runtime>-custom" would resolve the runtime "python3.9" to "myrepo/lambda:python3.9-custom"
132
133	Option B: use a JSON dict string for the config variable, mapping the runtime to the full image name & tag
134	e.g. {"python3.9": "myrepo/lambda:python3.9-custom", "python3.8": "myotherrepo/pylambda:3.8"}
135
136	Note that with Option B this will only apply to the runtimes included in the dict.
137	All other (non-included) runtimes will fall back to the default behavior.
138	"""
139
140	_mapping: dict[Runtime, str]	1✔
141	_default_resolve_fn: Callable[[Runtime], str]	1✔
142
143	def __init__(	1✔
144	self, default_resolve_fn: Callable[[Runtime], str] = get_default_image_for_runtime
145	):
146	self._mapping = {}	1✔
147	self._default_resolve_fn = default_resolve_fn	1✔
148
149	def _resolve(self, runtime: Runtime, custom_image_mapping: str = "") -> str:	1✔
150	if runtime not in IMAGE_MAPPING:	1✔
UNCOV 151	raise ValueError(f"Unsupported runtime {runtime}")	×
152
153	if not custom_image_mapping:	1✔
154	return self._default_resolve_fn(runtime)	1✔
155
156	# Option A (pattern string that includes <runtime> to replace)
157	if "<runtime>" in custom_image_mapping:	1✔
158	return custom_image_mapping.replace("<runtime>", runtime)	1✔
159
160	# Option B (json dict mapping with fallback)
161	try:	1✔
162	mapping: dict = json.loads(custom_image_mapping)	1✔
163	# at this point we're loading the whole dict to avoid parsing multiple times
164	for k, v in mapping.items():	1✔
165	if k not in IMAGE_MAPPING:	1✔
UNCOV 166	raise ValueError(	×
167	f"Unsupported runtime ({runtime}) provided in LAMBDA_RUNTIME_IMAGE_MAPPING"
168	)
169	self._mapping[k] = v	1✔
170
171	if runtime in self._mapping:	1✔
172	return self._mapping[runtime]	1✔
173
174	# fall back to default behavior if the runtime was not present in the custom config
175	return self._default_resolve_fn(runtime)	1✔
176
UNCOV 177	except Exception:	×
UNCOV 178	LOG.error(	×
179	"Failed to load config from LAMBDA_RUNTIME_IMAGE_MAPPING=%s",
180	custom_image_mapping,
181	)
UNCOV 182	raise # TODO: validate config at start and prevent startup	×
183
184	def get_image_for_runtime(self, runtime: Runtime) -> str:	1✔
185	if runtime not in self._mapping:	1✔
186	resolved_image = self._resolve(runtime, config.LAMBDA_RUNTIME_IMAGE_MAPPING)	1✔
187	self._mapping[runtime] = resolved_image	1✔
188
189	return self._mapping[runtime]	1✔
190
191
192	resolver = RuntimeImageResolver()	1✔
193
194
195	def prepare_image(function_version: FunctionVersion, platform: DockerPlatform) -> None:	1✔
UNCOV 196	if not function_version.config.runtime:	×
197	raise NotImplementedError(
198	"Custom images are currently not supported with image prebuilding"
199	)
200
201	# create dockerfile
UNCOV 202	docker_file = LAMBDA_DOCKERFILE.format(	×
203	base_img=resolver.get_image_for_runtime(function_version.config.runtime),
204	rapid_entrypoint=RAPID_ENTRYPOINT,
205	)
206
UNCOV 207	code_path = function_version.config.code.get_unzipped_code_location()	×
UNCOV 208	context_path = Path(	×
209	f"{tempfile.gettempdir()}/lambda/prebuild_tmp/{function_version.id.function_name}-{short_uid()}"
210	)
211	context_path.mkdir(parents=True)	×
UNCOV 212	prebuild_context = LambdaPrebuildContext(	×
213	docker_file_content=docker_file,
214	context_path=context_path,
215	function_version=function_version,
216	)
UNCOV 217	lambda_hooks.prebuild_environment_image.run(prebuild_context)	×
UNCOV 218	LOG.debug(	×
219	"Prebuilding image for function %s from context %s and Dockerfile %s",
220	function_version.qualified_arn,
221	str(prebuild_context.context_path),
222	prebuild_context.docker_file_content,
223	)
224	# save dockerfile
UNCOV 225	docker_file_path = prebuild_context.context_path / "Dockerfile"	×
UNCOV 226	with docker_file_path.open(mode="w") as f:	×
UNCOV 227	f.write(prebuild_context.docker_file_content)	×
228
229	# copy init file
230	init_destination_path = prebuild_context.context_path / "init"	×
UNCOV 231	src_init = f"{get_runtime_client_path()}/var/rapid/init"	×
UNCOV 232	shutil.copy(src_init, init_destination_path)	×
233	init_destination_path.chmod(0o755)	×
234
235	# copy function code
236	context_code_path = prebuild_context.context_path / "code"	×
UNCOV 237	shutil.copytree(	×
238	f"{str(code_path)}/",
239	str(context_code_path),
240	dirs_exist_ok=True,
241	)
242	# if layers are present, permissions should be 0755
UNCOV 243	if prebuild_context.function_version.config.layers:	×
UNCOV 244	chmod_r(str(context_code_path), 0o755)	×
245
246	try:	×
247	image_name = get_image_name_for_function(function_version)	×
UNCOV 248	CONTAINER_CLIENT.build_image(	×
249	dockerfile_path=str(docker_file_path),
250	image_name=image_name,
251	platform=platform,
252	)
UNCOV 253	except Exception as e:	×
UNCOV 254	if LOG.isEnabledFor(logging.DEBUG):	×
UNCOV 255	LOG.exception(	×
256	"Error while building prebuilt lambda image for '%s'",
257	function_version.qualified_arn,
258	)
259	else:
UNCOV 260	LOG.error(	×
261	"Error while building prebuilt lambda image for '%s', Error: %s",
262	function_version.qualified_arn,
263	e,
264	)
265	finally:
UNCOV 266	rm_rf(str(prebuild_context.context_path))	×
267
268
269	@dataclasses.dataclass	1✔
270	class LambdaContainerConfiguration(ContainerConfiguration):	1✔
271	copy_folders: list[tuple[str, str]] = dataclasses.field(default_factory=list)	1✔
272
273
274	class DockerRuntimeExecutor(RuntimeExecutor):	1✔
275	ip: str \| None	1✔
276	executor_endpoint: ExecutorEndpoint \| None	1✔
277	container_name: str	1✔
278
279	def __init__(self, id: str, function_version: FunctionVersion) -> None:	1✔
280	super().__init__(id=id, function_version=function_version)	1✔
281	self.ip = None	1✔
282	self.executor_endpoint = ExecutorEndpoint(self.id)	1✔
283	self.container_name = self._generate_container_name()	1✔
284	LOG.debug("Assigning container name of %s to executor %s", self.container_name, self.id)	1✔
285
286	def get_image(self) -> str:	1✔
287	if not self.function_version.config.runtime:	1✔
288	raise NotImplementedError("Container images are a Pro feature.")
289	return (	1✔
290	get_image_name_for_function(self.function_version)
291	if config.LAMBDA_PREBUILD_IMAGES
292	else resolver.get_image_for_runtime(self.function_version.config.runtime)
293	)
294
295	def _generate_container_name(self):	1✔
296	"""
297	Format <main-container-name>-lambda-<function-name>-<executor-id>
298	TODO: make the format configurable
299	"""
300	container_name = "-".join(	1✔
301	[
302	get_main_container_name() or "localstack",
303	"lambda",
304	self.function_version.id.function_name.lower(),
305	]
306	).replace("_", "-")
307	return f"{container_name}-{self.id}"	1✔
308
309	def start(self, env_vars: dict[str, str]) -> None:	1✔
310	self.executor_endpoint.start()	1✔
311	main_network, *additional_networks = self._get_networks_for_executor()	1✔
312	container_config = LambdaContainerConfiguration(	1✔
313	image_name=None,
314	name=self.container_name,
315	env_vars=env_vars,
316	network=main_network,
317	entrypoint=RAPID_ENTRYPOINT,
318	platform=docker_platform(self.function_version.config.architectures[0]),
319	additional_flags=config.LAMBDA_DOCKER_FLAGS,
320	)
321
322	if self.function_version.config.package_type == PackageType.Zip:	1✔
323	if self.function_version.config.code.is_hot_reloading():	1✔
324	container_config.env_vars[HOT_RELOADING_ENV_VARIABLE] = "/var/task"	1✔
325	if container_config.volumes is None:	1✔
UNCOV 326	container_config.volumes = VolumeMappings()	×
327	container_config.volumes.add(	1✔
328	BindMount(
329	str(self.function_version.config.code.get_unzipped_code_location()),
330	"/var/task",
331	read_only=True,
332	)
333	)
334	else:
335	container_config.copy_folders.append(	1✔
336	(
337	f"{str(self.function_version.config.code.get_unzipped_code_location())}/.",
338	"/var/task",
339	)
340	)
341
342	# always chmod /tmp to 700
343	chmod_paths = [ChmodPath(path="/tmp", mode="0700")]	1✔
344
345	# set the dns server of the lambda container to the LocalStack container IP
346	# the dns server will automatically respond with the right target for transparent endpoint injection
347	if config.LAMBDA_DOCKER_DNS:	1✔
348	# Don't overwrite DNS container config if it is already set (e.g., using LAMBDA_DOCKER_DNS)
UNCOV 349	LOG.warning(	×
350	"Container DNS overridden to %s, connection to names pointing to LocalStack, like 'localhost.localstack.cloud' will need additional configuration.",
351	config.LAMBDA_DOCKER_DNS,
352	)
UNCOV 353	container_config.dns = config.LAMBDA_DOCKER_DNS	×
354	else:
355	if dns_server.is_server_running():	1✔
356	# Set the container DNS to LocalStack to resolve localhost.localstack.cloud and
357	# enable transparent endpoint injection (Pro image only).
358	container_config.dns = self.get_endpoint_from_executor()	1✔
359
360	lambda_hooks.start_docker_executor.run(container_config, self.function_version)	1✔
361
362	if not container_config.image_name:	1✔
363	container_config.image_name = self.get_image()	1✔
364	if config.LAMBDA_DEV_PORT_EXPOSE:	1✔
UNCOV 365	self.executor_endpoint.container_port = get_free_tcp_port()	×
UNCOV 366	if container_config.ports is None:	×
UNCOV 367	container_config.ports = PortMappings()	×
368	container_config.ports.add(self.executor_endpoint.container_port, INVOCATION_PORT)	×
369
370	if config.LAMBDA_INIT_DEBUG:	1✔
371	container_config.entrypoint = "/debug-bootstrap.sh"	×
UNCOV 372	if not container_config.ports:	×
UNCOV 373	container_config.ports = PortMappings()	×
374	container_config.ports.add(config.LAMBDA_INIT_DELVE_PORT, config.LAMBDA_INIT_DELVE_PORT)	×
375
376	if (	1✔
377	self.function_version.config.layers
378	and not config.LAMBDA_PREBUILD_IMAGES
379	and self.function_version.config.package_type == PackageType.Zip
380	):
381	# avoid chmod on mounted code paths
UNCOV 382	hot_reloading_env = container_config.env_vars.get(HOT_RELOADING_ENV_VARIABLE, "")	×
UNCOV 383	if "/opt" not in hot_reloading_env:	×
UNCOV 384	chmod_paths.append(ChmodPath(path="/opt", mode="0755"))	×
385	if "/var/task" not in hot_reloading_env:	×
386	chmod_paths.append(ChmodPath(path="/var/task", mode="0755"))	×
387	container_config.env_vars["LOCALSTACK_CHMOD_PATHS"] = json.dumps(chmod_paths)	1✔
388
389	CONTAINER_CLIENT.create_container_from_config(container_config)	1✔
390	if (	1✔
391	not config.LAMBDA_PREBUILD_IMAGES
392	or self.function_version.config.package_type != PackageType.Zip
393	):
394	CONTAINER_CLIENT.copy_into_container(	1✔
395	self.container_name, f"{str(get_runtime_client_path())}/.", "/"
396	)
397	# tiny bit inefficient since we actually overwrite the init, but otherwise the path might not exist
398	if config.LAMBDA_INIT_BIN_PATH:	1✔
UNCOV 399	CONTAINER_CLIENT.copy_into_container(	×
400	self.container_name, config.LAMBDA_INIT_BIN_PATH, "/var/rapid/init"
401	)
402	if config.LAMBDA_INIT_DEBUG:	1✔
UNCOV 403	CONTAINER_CLIENT.copy_into_container(	×
404	self.container_name, config.LAMBDA_INIT_DELVE_PATH, "/var/rapid/dlv"
405	)
406	CONTAINER_CLIENT.copy_into_container(	×
407	self.container_name, config.LAMBDA_INIT_BOOTSTRAP_PATH, "/debug-bootstrap.sh"
408	)
409
410	if not config.LAMBDA_PREBUILD_IMAGES:	1✔
411	# copy_folders should be empty here if package type is not zip
412	for source, target in container_config.copy_folders:	1✔
413	CONTAINER_CLIENT.copy_into_container(self.container_name, source, target)	1✔
414
415	if additional_networks:	1✔
416	for additional_network in additional_networks:	1✔
417	CONTAINER_CLIENT.connect_container_to_network(	1✔
418	additional_network, self.container_name
419	)
420
421	CONTAINER_CLIENT.start_container(self.container_name)	1✔
422	# still using main network as main entrypoint
423	self.ip = CONTAINER_CLIENT.get_container_ipv4_for_network(	1✔
424	container_name_or_id=self.container_name, container_network=main_network
425	)
426	if config.LAMBDA_DEV_PORT_EXPOSE:	1✔
UNCOV 427	self.ip = "127.0.0.1"	×
428	self.executor_endpoint.container_address = self.ip	1✔
429
430	self.executor_endpoint.wait_for_startup()	1✔
431
432	def stop(self) -> None:	1✔
433	CONTAINER_CLIENT.stop_container(container_name=self.container_name, timeout=5)	1✔
434	if config.LAMBDA_REMOVE_CONTAINERS:	1✔
435	CONTAINER_CLIENT.remove_container(container_name=self.container_name)	1✔
436	try:	1✔
437	self.executor_endpoint.shutdown()	1✔
UNCOV 438	except Exception as e:	×
UNCOV 439	LOG.debug(	×
440	"Error while stopping executor endpoint for lambda %s, error: %s",
441	self.function_version.qualified_arn,
442	e,
443	)
444
445	def get_address(self) -> str:	1✔
UNCOV 446	if not self.ip:	×
UNCOV 447	raise LambdaRuntimeException(f"IP address of executor '{self.id}' unknown")	×
UNCOV 448	return self.ip	×
449
450	def get_endpoint_from_executor(self) -> str:	1✔
451	return get_main_endpoint_from_container()	1✔
452
453	def _get_networks_for_executor(self) -> list[str]:	1✔
454	return get_all_container_networks_for_lambda()	1✔
455
456	def invoke(self, payload: dict[str, str]):	1✔
457	LOG.debug(	1✔
458	"Sending invoke-payload '%s' to executor '%s'",
459	truncate(json.dumps(payload), config.LAMBDA_TRUNCATE_STDOUT),
460	self.id,
461	)
462	return self.executor_endpoint.invoke(payload)	1✔
463
464	def get_logs(self) -> str:	1✔
465	try:	1✔
466	return CONTAINER_CLIENT.get_container_logs(container_name_or_id=self.container_name)	1✔
467	except NoSuchContainer:	1✔
468	return "Container was not created"	1✔
469
470	@classmethod	1✔
471	def prepare_version(cls, function_version: FunctionVersion) -> None:	1✔
472	lambda_hooks.prepare_docker_executor.run(function_version)	1✔
473	# Trigger the installation of the Lambda runtime-init binary before invocation and
474	# cache the result to save time upon every invocation.
475	get_runtime_client_path()	1✔
476	if function_version.config.code:	1✔
477	function_version.config.code.prepare_for_execution()	1✔
478	image_name = resolver.get_image_for_runtime(function_version.config.runtime)	1✔
479	platform = docker_platform(function_version.config.architectures[0])	1✔
480	_ensure_runtime_image_present(image_name, platform)	1✔
481	if config.LAMBDA_PREBUILD_IMAGES:	1✔
UNCOV 482	prepare_image(function_version, platform)	×
483
484	@classmethod	1✔
485	def cleanup_version(cls, function_version: FunctionVersion) -> None:	1✔
486	if config.LAMBDA_PREBUILD_IMAGES:	1✔
487	# TODO re-enable image cleanup.
488	# Enabling it currently deletes image after updates as well
489	# It also creates issues when cleanup is concurrently with build
490	# probably due to intermediate layers being deleted
491	# image_name = get_image_name_for_function(function_version)
492	# LOG.debug("Removing image %s after version deletion", image_name)
493	# CONTAINER_CLIENT.remove_image(image_name)
UNCOV 494	pass	×
495
496	def get_runtime_endpoint(self) -> str:	1✔
497	return f"http://{self.get_endpoint_from_executor()}:{config.GATEWAY_LISTEN[0].port}{self.executor_endpoint.get_endpoint_prefix()}"	1✔
498
499	@classmethod	1✔
500	def validate_environment(cls) -> bool:	1✔
501	if not CONTAINER_CLIENT.has_docker():	1✔
UNCOV 502	LOG.warning(	×
503	"WARNING: Docker not available in the LocalStack container but required to run Lambda "
504	'functions. Please add the Docker volume mount "/var/run/docker.sock:/var/run/docker.sock" to your '
505	"LocalStack startup. https://docs.localstack.cloud/user-guide/aws/lambda/#docker-not-available"
506	)
UNCOV 507	return False	×
508	return True	1✔

localstack / localstack / 20942662173

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