20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

71.02

/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 typing import Literal

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

localstack / localstack / 20565403496

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