16665047018

Committed 31 Jul 2025 06:34PM UTC coverage: 86.897% (+0.1%) from 86.781%

Build # 16665047018

Build Type

push

github

Committed by

web-flow

Commit Message

Apigw/enable vpce routing (#12937)

Coverage Stats

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

314 existing lines in 13 files now uncovered.

66469 of 76492 relevant lines covered (86.9%)

0.87 hits per line

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

71.31

/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 pathlib import Path
from typing import Callable, Dict, Literal, Optional

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"

InitializationType = Literal["on-demand", "provisioned-concurrency"]

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

PULLED_IMAGES: set[(str, DockerPlatform)] = set()
PULL_LOCKS: dict[(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 = dict()
        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: Optional[str]
    executor_endpoint: Optional[ExecutorEndpoint]
    container_name: str

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

localstack / localstack / 16665047018

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