diff --git a/samcli/local/docker/container.py b/samcli/local/docker/container.py index 45a05b9d983..24939f1a4b0 100644 --- a/samcli/local/docker/container.py +++ b/samcli/local/docker/container.py @@ -152,6 +152,10 @@ def __init__( # Runtime properties of the container. They won't have value until container is created or started self.id: Optional[str] = None + # Container capacity signaling for concurrent request management + self._is_at_capacity = False + self._busy_lock = threading.Lock() + # aws-lambda-rie defaults to 8080 as the port, however that's a common port. A port is chosen by # selecting the first free port in a range that's not ephemeral. self._start_port_range = 5000 @@ -483,37 +487,44 @@ def wait_for_result(self, full_path, event, stdout, stderr, start_timer=None): # NOTE(sriram-mv): All logging is re-directed to stderr, so that only the lambda function return # will be written to stdout. - # the log thread will not be closed until the container itself got deleted, - # so as long as the container is still there, no need to start a new log thread - if not self._logs_thread or not self._logs_thread.is_alive(): - self._logs_thread_event = self._create_threading_event() - self._logs_thread = threading.Thread( - target=self.wait_for_logs, args=(stderr, stderr, self._logs_thread_event), daemon=True - ) - self._logs_thread.start() - - # wait_for_http_response will attempt to establish a connection to the socket - # but it'll fail if the socket is not listening yet, so we wait for the socket - self._wait_for_socket_connection() - - # start the timer for function timeout right before executing the function, as waiting for the socket - # can take some time - timer = start_timer() if start_timer else None - response, is_image = self.wait_for_http_response(full_path, event, stdout) - if timer: - timer.cancel() - - self._logs_thread_event.wait(timeout=1) - if isinstance(response, str): - stdout.write_str(response) - elif isinstance(response, bytes) and is_image: - stdout.write_bytes(response) - elif isinstance(response, bytes): - stdout.write_str(response.decode("utf-8")) - stdout.flush() - stderr.write_str("\n") - stderr.flush() - self._logs_thread_event.clear() + # Mark container as busy at the start of request processing + self._mark_busy() + + try: + # the log thread will not be closed until the container itself got deleted, + # so as long as the container is still there, no need to start a new log thread + if not self._logs_thread or not self._logs_thread.is_alive(): + self._logs_thread_event = self._create_threading_event() + self._logs_thread = threading.Thread( + target=self.wait_for_logs, args=(stderr, stderr, self._logs_thread_event), daemon=True + ) + self._logs_thread.start() + + # wait_for_http_response will attempt to establish a connection to the socket + # but it'll fail if the socket is not listening yet, so we wait for the socket + self._wait_for_socket_connection() + + # start the timer for function timeout right before executing the function, as waiting for the socket + # can take some time + timer = start_timer() if start_timer else None + response, is_image = self.wait_for_http_response(full_path, event, stdout) + if timer: + timer.cancel() + + self._logs_thread_event.wait(timeout=1) + if isinstance(response, str): + stdout.write_str(response) + elif isinstance(response, bytes) and is_image: + stdout.write_bytes(response) + elif isinstance(response, bytes): + stdout.write_str(response.decode("utf-8")) + stdout.flush() + stderr.write_str("\n") + stderr.flush() + self._logs_thread_event.clear() + finally: + # Always mark container as idle when request completes, even if an exception occurred + self._mark_idle() def wait_for_logs( self, @@ -702,6 +713,37 @@ def is_running(self): except docker.errors.NotFound: return False + def has_capacity(self) -> bool: + """ + Checks if the container has capacity to handle a new request. + Thread-safe method that returns True if the container is idle and can accept a request. + + Returns + ------- + bool + True if the container is not busy and can handle a request, False otherwise + """ + with self._busy_lock: + return not self._is_at_capacity + + def _mark_busy(self) -> None: + """ + Marks the container as busy, indicating it is currently processing a request. + Thread-safe method that sets the busy state. + """ + with self._busy_lock: + self._is_at_capacity = True + LOG.debug("Container %s marked as busy", self.id[:12] if self.id else "unknown") + + def _mark_idle(self) -> None: + """ + Marks the container as idle, indicating it is available to handle new requests. + Thread-safe method that clears the busy state. + """ + with self._busy_lock: + self._is_at_capacity = False + LOG.debug("Container %s marked as idle", self.id[:12] if self.id else "unknown") + def _resolve_symlinks_in_context(self, context) -> bool: """ diff --git a/samcli/local/lambdafn/runtime.py b/samcli/local/lambdafn/runtime.py index 82c75c0e466..39419345b3e 100644 --- a/samcli/local/lambdafn/runtime.py +++ b/samcli/local/lambdafn/runtime.py @@ -9,7 +9,8 @@ import signal import tempfile import threading -from typing import Dict, Optional, Union +from collections import defaultdict +from typing import Dict, List, Optional, Union from samcli.lib.telemetry.metric import capture_parameter from samcli.lib.utils.file_observer import LambdaFunctionObserver @@ -421,12 +422,160 @@ def __init__(self, container_manager, image_builder, observer=None, mount_symlin Determines if the warm containers is enabled or not. """ self._function_configs = {} - self._containers = {} + # Changed from Dict[str, Container] to Dict[str, List[Container]] to support multiple containers per function + # Using defaultdict to automatically initialize empty lists for new function paths + self._containers: Dict[str, List[Container]] = defaultdict(list) + # Thread-safe access to container collections + self._container_lock = threading.Lock() self._observer = observer if observer else LambdaFunctionObserver(self._on_code_change) super().__init__(container_manager, image_builder, mount_symlinks=mount_symlinks, no_mem_limit=no_mem_limit) + def _find_available_container(self, function_path: str) -> Optional[Container]: + """ + Find the first available container for the given function using capacity-based selection. + Iterates through the containers list and checks each container's capacity using has_capacity(). + Returns the first container with capacity, or None if all are busy. + Uses simple first-available iteration (not round-robin). + This method should be called while holding self._container_lock. + + Parameters + ---------- + function_path : str + The full path of the function + + Returns + ------- + Optional[Container] + The first available container if one has capacity, None otherwise + """ + containers = self._containers.get(function_path, []) + for container in containers: + if container.is_created() and container.has_capacity(): + return container + return None + + def _acquire_container( + self, + function_config, + debug_context=None, + container_host=None, + container_host_interface=None, + extra_hosts=None, + ) -> Container: + """ + Acquire a container for the function using capacity-based selection. + Acquires runtime lock before container operations, calls _find_available_container() + to check for available containers, reuses container if one has capacity, + creates new container if all are at capacity, appends new containers to list (doesn't overwrite), + and handles function config changes (reloads all containers for function). + + Parameters + ---------- + function_config : FunctionConfig + Configuration of the function + debug_context : DebugContext, optional + Debugging context for the function + container_host : str, optional + Host of locally emulated Lambda container + container_host_interface : str, optional + Interface that Docker host binds ports to + extra_hosts : Dict, optional + Dict of hostname to IP resolutions + + Returns + ------- + Container + The acquired container (either reused or newly created) + """ + # Acquire runtime lock before container operations + with self._container_lock: + function_path = function_config.full_path + + # Check for configuration changes requiring container reload + # Handle function config changes (reload all containers for function) + exist_function_config = self._function_configs.get(function_path, None) + if exist_function_config and _require_container_reloading(exist_function_config, function_config): + LOG.info( + "Lambda Function '%s' definition has been changed in the stack template, " + "terminate all warm containers.", + function_path, + ) + self._stop_all_containers_for_function(function_path) + + # Call _find_available_container() to check for available containers + # Reuse container if one has capacity + available_container = self._find_available_container(function_path) + if available_container: + LOG.info("Reusing warm container with capacity for Lambda function '%s'", function_path) + # CRITICAL: Mark container as busy BEFORE releasing the lock to prevent race conditions + # This ensures concurrent requests don't see this container as available + available_container._mark_busy() + return available_container + + # Create new container if all are at capacity + LOG.info("Creating new warm container for Lambda function '%s'", function_path) + + # debug_context should be used only if the function name is the one defined + # in debug-function option + if debug_context and debug_context.debug_function != function_config.name: + LOG.debug( + "Disable the debugging for Lambda Function %s, as the passed debug function is %s", + function_config.name, + debug_context.debug_function, + ) + debug_context = None + + # Start observer if not already watching + self._observer.watch(function_config) + self._observer.start() + + # Create new container using parent class method + container: Container = super(WarmLambdaRuntime, self).create( + function_config, debug_context, container_host, container_host_interface, extra_hosts + ) + + # CRITICAL: Mark container as busy BEFORE releasing the lock to prevent race conditions + # This ensures concurrent requests don't see this container as available + container._mark_busy() + + # Append new containers to list (defaultdict automatically creates list if needed) + self._containers[function_path].append(container) + + # Update function config + self._function_configs[function_path] = function_config + + return container + + def _stop_all_containers_for_function(self, function_path: str) -> None: + """ + Stop all containers for a specific function. + This method should be called while holding self._container_lock. + Iterates container list and stops each one, handles errors gracefully, + and clears list after stopping all. + + Parameters + ---------- + function_path : str + The full path of the function + """ + containers = self._containers.get(function_path, []) + for container in containers: + try: + self._container_manager.stop(container) + except Exception as ex: + LOG.debug("Failed to stop container %s: %s", container.id, str(ex)) + + # Unwatch the function config before removing it + exist_function_config = self._function_configs.get(function_path, None) + if exist_function_config: + self._observer.unwatch(exist_function_config) + + # Clear containers list and function config + self._containers[function_path] = [] + self._function_configs.pop(function_path, None) + def create( self, function_config, @@ -436,9 +585,10 @@ def create( extra_hosts=None, ): """ - Create a new Container for the passed function, then store it in a dictionary using the function name, - so it can be retrieved later and used in the other functions. Make sure to use the debug_context only - if the function_config.name equals debug_context.debug-function or the warm_containers option is disabled + Create a new Container for the passed function, then store it in a list using the function name, + so it can be retrieved later and used in the other functions. Supports multiple containers per function + for concurrent requests. Make sure to use the debug_context only if the function_config.name equals + debug_context.debug-function or the warm_containers option is disabled. Parameters ---------- @@ -450,63 +600,80 @@ def create( Host of locally emulated Lambda container container_host_interface string Interface that Docker host binds ports to + extra_hosts Dict + Optional. Dict of hostname to IP resolutions Returns ------- Container - the created container + the created or reused container """ + # Use the new acquisition logic that handles container reuse and creation + return self._acquire_container( + function_config, debug_context, container_host, container_host_interface, extra_hosts + ) - # reuse the cached container if it is created, and if the function configuration is not changed - exist_function_config = self._function_configs.get(function_config.full_path, None) - container = self._containers.get(function_config.full_path, None) - if exist_function_config and _require_container_reloading(exist_function_config, function_config): - LOG.info( - "Lambda Function '%s' definition has been changed in the stack template, " - "terminate the created warm container.", - function_config.full_path, - ) - self._function_configs.pop(exist_function_config.full_path, None) - if container: - self._container_manager.stop(container) - self._containers.pop(exist_function_config.full_path, None) - self._observer.unwatch(exist_function_config) - elif container and container.is_created(): - LOG.info("Reuse the created warm container for Lambda function '%s'", function_config.full_path) - return container - - # debug_context should be used only if the function name is the one defined - # in debug-function option - if debug_context and debug_context.debug_function != function_config.name: - LOG.debug( - "Disable the debugging for Lambda Function %s, as the passed debug function is %s", - function_config.name, - debug_context.debug_function, - ) - debug_context = None + def run( + self, + container, + function_config, + debug_context, + container_host=None, + container_host_interface=None, + extra_hosts=None, + ): + """ + Override parent's run method to handle EAGER mode container initialization. + Containers are marked as busy in _acquire_container() to prevent race conditions. + For EAGER pre-warming, we mark them idle after starting so they're available for requests. - self._observer.watch(function_config) - self._observer.start() + Parameters + ---------- + container Container + the created container to be run + function_config FunctionConfig + Configuration of the function to run its created container. + debug_context DebugContext + Debugging context for the function (includes port, args, and path) + container_host string + Host of locally emulated Lambda container + container_host_interface string + Optional. Interface that Docker host binds ports to + extra_hosts Dict + Optional. Dict of hostname to IP resolutions - container = super().create( - function_config, debug_context, container_host, container_host_interface, extra_hosts + Returns + ------- + Container + the running container + """ + # Call parent's run method to start the container + container = super().run( + container, function_config, debug_context, container_host, container_host_interface, extra_hosts ) - self._function_configs[function_config.full_path] = function_config - self._containers[function_config.full_path] = container + + # For EAGER mode: containers are pre-warmed but not immediately invoked + # Mark them as idle so they're available for actual requests + # For normal invoke flow: wait_for_result() will call _mark_busy() again (idempotent) + if container and container._is_at_capacity: + container._mark_idle() return container def _on_invoke_done(self, container): """ - Cleanup the created resources, just before the invoke function ends. - In warm containers, the running containers will be closed just before the end of te command execution, - so no action is done here + Override parent's cleanup to preserve warm containers. + In warm containers mode, containers persist and are not stopped after each invocation. + The container automatically marks itself as idle via _mark_idle() in wait_for_result's finally block. Parameters ---------- container: Container The current running container """ + # Do NOT stop the container - warm containers persist between invocations + # Container state is managed internally by the Container class itself + pass def _configure_interrupt(self, function_full_path, timeout, container, is_debugging): """ @@ -555,40 +722,74 @@ def signal_handler(sig, frame): def clean_running_containers_and_related_resources(self): """ - Clean the running containers, the decompressed code dirs, and stop the created observer + Clean all running containers, the decompressed code dirs, and stop the created observer. + Handles multiple containers per function and errors gracefully. + Iterates over all containers in all lists, stops each with error handling, + tracks successful and failed cleanups, and logs a summary at debug level. + Continues cleanup even if individual containers fail. """ LOG.debug("Terminating all running warm containers") - for function_name, container in self._containers.items(): - LOG.debug("Terminate running warm container for Lambda Function '%s'", function_name) - self._container_manager.stop(container) - self._clean_decompressed_paths() - self._observer.stop() + + cleanup_errors = [] + successful_cleanups = 0 + total_containers = 0 + + with self._container_lock: + for function_name, containers in self._containers.items(): + total_containers += len(containers) + for container in containers: + try: + LOG.debug("Terminate warm container %s for function '%s'", container.id[:12], function_name) + self._container_manager.stop(container) + successful_cleanups += 1 + except Exception as ex: + LOG.debug("Failed to stop container %s: %s", container.id[:12], str(ex)) + cleanup_errors.append((function_name, container.id, ex)) + + # Log cleanup summary at debug level + LOG.debug( + "Container cleanup complete: %d/%d successful, %d failed", + successful_cleanups, + total_containers, + len(cleanup_errors), + ) + + # Clean up other resources - best effort, don't fail if these error + try: + self._clean_decompressed_paths() + except Exception as ex: + LOG.debug("Failed to clean decompressed paths: %s", str(ex)) + + try: + self._observer.stop() + except Exception as ex: + LOG.debug("Failed to stop observer: %s", str(ex)) def _on_code_change(self, functions): """ Handles the lambda function code change event. it determines if there is a real change in the code by comparing the checksum of the code path before and after the event. + Stops all containers for the changed function (not just one), clears entire list, + and unwatches function config. Parameters ---------- functions: list [FunctionConfig] the lambda functions that their source code or images got changed """ - for function_config in functions: - function_full_path = function_config.full_path - resource = "source code" if function_config.packagetype == ZIP else f"{function_config.imageuri} image" - LOG.info( - "Lambda Function '%s' %s has been changed, terminate its warm container. " - "The new container will be created in lazy mode", - function_full_path, - resource, - ) - self._observer.unwatch(function_config) - self._function_configs.pop(function_full_path, None) - container = self._containers.get(function_full_path, None) - if container: - self._container_manager.stop(container) - self._containers.pop(function_full_path, None) + with self._container_lock: + for function_config in functions: + function_full_path = function_config.full_path + resource = "source code" if function_config.packagetype == ZIP else f"{function_config.imageuri} image" + LOG.info( + "Lambda Function '%s' %s has been changed, terminate all warm containers. " + "New containers will be created in lazy mode", + function_full_path, + resource, + ) + + # Stop all containers for this function using the helper method + self._stop_all_containers_for_function(function_full_path) def _unzip_file(filepath): diff --git a/tests/integration/local/common_utils.py b/tests/integration/local/common_utils.py index 73277fac766..7d60650b392 100644 --- a/tests/integration/local/common_utils.py +++ b/tests/integration/local/common_utils.py @@ -2,6 +2,8 @@ import logging import random import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from typing import Callable, List, Optional LOG = logging.getLogger(__name__) @@ -36,3 +38,27 @@ def wait_for_local_process(process, port, collect_output=False) -> str: def random_port(): return random.randint(30000, 40000) + + +def send_concurrent_requests(request_func: Callable, count: int, timeout: int = 300) -> List: + """ + Send multiple concurrent requests using ThreadPoolExecutor. + + Args: + request_func: Callable that performs a single request (e.g., lambda: requests.post(url)) + count: Number of concurrent requests to send + timeout: Timeout for the entire operation in seconds + + Returns: + List of results from all requests + + Example: + results = send_concurrent_requests( + lambda: requests.post(url + "/endpoint", timeout=300), + count=3 + ) + """ + with ThreadPoolExecutor(max_workers=count) as thread_pool: + futures = [thread_pool.submit(request_func) for _ in range(count)] + results = [future.result() for future in as_completed(futures, timeout=timeout)] + return results diff --git a/tests/integration/local/start_api/test_start_api.py b/tests/integration/local/start_api/test_start_api.py index 4186b8890e8..442c8242cd0 100644 --- a/tests/integration/local/start_api/test_start_api.py +++ b/tests/integration/local/start_api/test_start_api.py @@ -23,6 +23,7 @@ from tests.testing_utils import IS_WINDOWS from .start_api_integ_base import StartApiIntegBaseClass, WritableStartApiIntegBaseClass from ..invoke.layer_utils import LayerUtils +from ..common_utils import send_concurrent_requests @parameterized_class( @@ -2153,15 +2154,23 @@ def setUp(self): self.mode_env_variable = getattr(self.__class__, "mode_env_variable", str(uuid.uuid4())) def count_running_containers(self): - """Count containers created by this test using Docker client directly.""" - # Use Docker client to find containers with SAM CLI labels + """ + Count containers created by this test using Docker client directly. + + Filters containers by: + 1. SAM CLI label (sam.cli.container.type=lambda) + 2. MODE environment variable (unique per test class) + + This ensures test isolation when running multiple test classes in parallel. + Note: Test methods within the same class share the same MODE and are NOT isolated. + """ try: # Get running containers with SAM CLI lambda container label sam_containers = self.docker_client.containers.list( all=False, filters={"label": "sam.cli.container.type=lambda"} ) - # Filter by our test's mode environment variable if possible + # Filter by our test's mode environment variable for isolation test_containers = [] for container in sam_containers: try: @@ -2174,15 +2183,11 @@ def count_running_containers(self): except Exception: continue - # If we found containers with our mode variable, return that count - if test_containers: - return len(test_containers) - - # Otherwise, return all SAM containers (fallback) - return len(sam_containers) + # Only return containers with our MODE - no fallback to avoid cross-contamination + return len(test_containers) except Exception as e: - # If we can't access Docker client, fall back to 0 + LOG.error(f"Failed to count containers: {e}") return 0 def _parse_container_ids_from_output(self): @@ -2380,6 +2385,280 @@ def test_only_one_new_created_containers_after_lambda_function_invoke(self): self.assertEqual(initiated_containers, initiated_containers_before_any_invoke + 1) +class TestLazyWarmContainersConcurrency(TestWarmContainersBaseClass): + """ + Test LAZY mode warm containers concurrency behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + + Verifies fixes for: + - Issue 1: Multiple containers tracked and cleaned up (not just one) + - Issue 2: All containers reused consistently (not just one) + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_concurrent_requests_all_succeed(self): + """Concurrent requests create multiple containers and all succeed.""" + initial_count = self.count_running_containers() + self.assertEqual(initial_count, 0) + + results = send_concurrent_requests(lambda: requests.post(self.url + "/id", timeout=300), count=3) + + self.assertEqual(len(results), 3) + for response in results: + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"hello": "world"}) + + +class TestLazyWarmContainersReuse(TestWarmContainersBaseClass): + """ + Test LAZY mode warm containers reuse behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_containers_reused_across_batches(self): + """Subsequent requests reuse all available containers.""" + first_batch = send_concurrent_requests(lambda: requests.post(self.url + "/id", timeout=300), count=3) + self.assertEqual(len(first_batch), 3) + for response in first_batch: + self.assertEqual(response.status_code, 200) + + sleep(2) + + second_batch = send_concurrent_requests(lambda: requests.post(self.url + "/id", timeout=300), count=3) + self.assertEqual(len(second_batch), 3) + for response in second_batch: + self.assertEqual(response.status_code, 200) + + +class TestLazyWarmContainersMultipleFunctions(TestWarmContainersBaseClass): + """ + Test LAZY mode warm containers with multiple functions. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_multiple_functions_work_correctly(self): + """Multiple functions each get their own containers.""" + function1_results = send_concurrent_requests(lambda: requests.post(self.url + "/id", timeout=300), count=2) + function2_results = send_concurrent_requests(lambda: requests.get(self.url + "/id/123", timeout=300), count=2) + + self.assertEqual(len(function1_results), 2) + self.assertEqual(len(function2_results), 2) + for response in function1_results + function2_results: + self.assertEqual(response.status_code, 200) + + +class TestLazyWarmContainersCleanup(TestWarmContainersBaseClass): + """ + Test LAZY mode SIGTERM cleanup in isolation. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. This is especially + important for SIGTERM tests since they terminate the server process. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @skipIf(IS_WINDOWS, "SIGTERM interrupt doesn't exist on Windows") + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_sigterm_cleans_up_all_containers(self): + """SIGTERM triggers complete cleanup of all containers.""" + response = requests.post(self.url + "/id", timeout=300) + self.assertEqual(response.status_code, 200) + + self.start_api_process.send_signal(signal.SIGTERM) + sleep(10) + + remaining_containers = self.count_running_containers() + self.assertEqual(remaining_containers, 0) + self.assertEqual(self.start_api_process.poll(), 0) + + +class TestEagerWarmContainersScaleUp(TestWarmContainersBaseClass): + """ + Test EAGER mode warm containers scale-up behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.EAGER.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_scales_up_under_load(self): + """Concurrent requests beyond initial containers trigger scale-up.""" + initial_count = self.count_running_containers() + # EAGER mode creates one container per function at startup + # Template has 3 functions: HelloWorldFunction, EchoEventFunction, SleepFunction + self.assertEqual(initial_count, 3) + + # Use sleep endpoint to ensure requests are truly concurrent + # Start requests in background using ThreadPoolExecutor + from concurrent.futures import ThreadPoolExecutor + import time + + def make_request(): + return requests.post(self.url + "/sleep", timeout=300) + + with ThreadPoolExecutor(max_workers=10) as executor: + # Submit all requests to SleepFunction (2-second sleep) + # Use 10 concurrent requests to force scale-up beyond 3 initial containers + # Submit ALL futures first before any start executing to ensure true concurrency + futures = [] + for i in range(10): + future = executor.submit(make_request) + futures.append(future) + + # Give containers time to start and scale up + # Wait long enough for multiple containers to be created + time.sleep(1.5) + + # Check container count while requests are still processing + container_count = self.count_running_containers() + self.assertGreater(container_count, 3, "Should scale up to handle concurrent requests") + + # Wait for all requests to complete + results = [f.result() for f in futures] + + for response in results: + self.assertEqual(response.status_code, 200) + + sleep(2) # Wait for containers to finish processing + + +class TestEagerWarmContainersReuse(TestWarmContainersBaseClass): + """ + Test EAGER mode warm containers reuse behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.EAGER.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_scaled_containers_reused(self): + """Scaled containers are reused in subsequent requests.""" + from concurrent.futures import ThreadPoolExecutor + import time + + def make_request(): + return requests.post(self.url + "/sleep", timeout=300) + + # First batch: scale up using sleep endpoint + with ThreadPoolExecutor(max_workers=10) as executor: + # Submit all futures first to ensure true concurrency + futures = [] + for i in range(10): + future = executor.submit(make_request) + futures.append(future) + + time.sleep(1.5) + container_count_after_scale = self.count_running_containers() + self.assertGreater(container_count_after_scale, 3, "Should scale up during first batch") + first_batch = [f.result() for f in futures] + + for response in first_batch: + self.assertEqual(response.status_code, 200) + + sleep(2) # Wait for containers to finish processing + + # Second batch: verify containers are reused (no additional scale-up) + with ThreadPoolExecutor(max_workers=10) as executor: + # Submit all futures first to ensure true concurrency + futures = [] + for i in range(10): + future = executor.submit(make_request) + futures.append(future) + + time.sleep(1.5) + container_count_after_reuse = self.count_running_containers() + # Should not create more containers since we already have enough + self.assertEqual( + container_count_after_reuse, container_count_after_scale, "Should reuse existing containers" + ) + second_batch = [f.result() for f in futures] + + for response in second_batch: + self.assertEqual(response.status_code, 200) + + sleep(2) # Wait for containers to finish processing + + +class TestEagerWarmContainersCleanup(TestWarmContainersBaseClass): + """ + Test EAGER mode warm containers cleanup behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.EAGER.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @skipIf(IS_WINDOWS, "SIGTERM interrupt doesn't exist on Windows") + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_sigterm_cleans_up_all_containers(self): + """SIGTERM cleans up all containers including scaled ones.""" + initial_count = self.count_running_containers() + self.assertEqual(initial_count, 2) + + results = send_concurrent_requests(lambda: requests.post(self.url + "/id", timeout=300), count=5) + for response in results: + self.assertEqual(response.status_code, 200) + + sleep(2) # Wait for containers to finish processing + scaled_count = self.count_running_containers() + self.assertGreater(scaled_count, initial_count) + + self.start_api_process.send_signal(signal.SIGTERM) + sleep(10) + + remaining_containers = self.count_running_containers() + self.assertEqual(remaining_containers, 0) + self.assertEqual(self.start_api_process.poll(), 0) + + class TestImagePackageType(StartApiIntegBaseClass): template_path = "/testdata/start_api/image_package_type/template.yaml" build_before_invoke = True diff --git a/tests/integration/local/start_lambda/test_start_lambda.py b/tests/integration/local/start_lambda/test_start_lambda.py index b25bd45a10b..a2e4edb3ad2 100644 --- a/tests/integration/local/start_lambda/test_start_lambda.py +++ b/tests/integration/local/start_lambda/test_start_lambda.py @@ -17,6 +17,7 @@ from samcli.commands.local.cli_common.invoke_context import ContainersInitializationMode from tests.testing_utils import IS_WINDOWS from .start_lambda_api_integ_base import StartLambdaIntegBaseClass, WatchWarmContainersIntegBaseClass +from ..common_utils import send_concurrent_requests class TestParallelRequests(StartLambdaIntegBaseClass): @@ -318,19 +319,32 @@ def setUp(self): region_name="us-east-1", use_ssl=False, verify=False, - config=Config(signature_version=UNSIGNED, read_timeout=120, retries={"max_attempts": 0}), + config=Config( + signature_version=UNSIGNED, + read_timeout=120, + retries={"max_attempts": 0}, + max_pool_connections=20, # Increase pool size for concurrent requests + ), ) def count_running_containers(self): - """Count containers created by this test using Docker client directly.""" - # Use Docker client to find containers with SAM CLI labels + """ + Count containers created by this test using Docker client directly. + + Filters containers by: + 1. SAM CLI label (sam.cli.container.type=lambda) + 2. MODE environment variable (unique per test class) + + This ensures test isolation when running multiple test classes in parallel. + Note: Test methods within the same class share the same MODE and are NOT isolated. + """ try: # Get running containers with SAM CLI lambda container label sam_containers = self.docker_client.containers.list( all=False, filters={"label": "sam.cli.container.type=lambda"} ) - # Filter by our test's mode environment variable if possible + # Filter by our test's mode environment variable for isolation test_containers = [] for container in sam_containers: try: @@ -540,6 +554,286 @@ def test_only_one_new_created_containers_after_lambda_function_invoke(self): self.assertEqual(initiated_containers, initiated_containers_before_any_invoke + 1) +class TestLazyWarmContainersConcurrency(TestWarmContainersBaseClass): + """ + Test LAZY mode warm containers concurrency behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + + Verifies fixes for: + - Issue 1: Multiple containers tracked and cleaned up (not just one) + - Issue 2: All containers reused consistently (not just one) + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_concurrent_invocations_all_succeed(self): + """Concurrent invocations create multiple containers and all succeed.""" + initial_count = self.count_running_containers() + self.assertEqual(initial_count, 0) + + results = send_concurrent_requests( + lambda: self.lambda_client.invoke(FunctionName="HelloWorldFunction"), count=3 + ) + + self.assertEqual(len(results), 3) + for result in results: + self.assertEqual(result.get("StatusCode"), 200) + response = json.loads(result.get("Payload").read().decode("utf-8")) + self.assertEqual(response.get("statusCode"), 200) + + +class TestLazyWarmContainersReuse(TestWarmContainersBaseClass): + """ + Test LAZY mode warm containers reuse behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_containers_reused_across_batches(self): + """Subsequent invocations reuse all available containers.""" + # First batch + first_batch = send_concurrent_requests( + lambda: self.lambda_client.invoke(FunctionName="HelloWorldFunction"), count=3 + ) + self.assertEqual(len(first_batch), 3) + for result in first_batch: + self.assertEqual(result.get("StatusCode"), 200) + + sleep(2) + + # Second batch should reuse containers + second_batch = send_concurrent_requests( + lambda: self.lambda_client.invoke(FunctionName="HelloWorldFunction"), count=3 + ) + self.assertEqual(len(second_batch), 3) + for result in second_batch: + self.assertEqual(result.get("StatusCode"), 200) + + +class TestLazyWarmContainersMultipleFunctions(TestWarmContainersBaseClass): + """ + Test LAZY mode warm containers with multiple functions. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_multiple_functions_work_correctly(self): + """Multiple functions each get their own containers.""" + function1_results = send_concurrent_requests( + lambda: self.lambda_client.invoke(FunctionName="HelloWorldFunction"), count=2 + ) + function2_results = send_concurrent_requests( + lambda: self.lambda_client.invoke(FunctionName="EchoEventFunction"), count=2 + ) + + self.assertEqual(len(function1_results), 2) + self.assertEqual(len(function2_results), 2) + for result in function1_results + function2_results: + self.assertEqual(result.get("StatusCode"), 200) + + +class TestLazyWarmContainersCleanup(TestWarmContainersBaseClass): + """ + Test LAZY mode SIGTERM cleanup in isolation. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. This is especially + important for SIGTERM tests since they terminate the server process. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.LAZY.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @skipIf(IS_WINDOWS, "SIGTERM interrupt doesn't exist on Windows") + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_sigterm_cleans_up_all_containers(self): + """SIGTERM triggers complete cleanup of all containers.""" + result = self.lambda_client.invoke(FunctionName="HelloWorldFunction") + self.assertEqual(result.get("StatusCode"), 200) + + self.start_lambda_process.send_signal(signal.SIGTERM) + sleep(10) + + remaining_containers = self.count_running_containers() + self.assertEqual(remaining_containers, 0) + self.assertEqual(self.start_lambda_process.poll(), 0) + + +class TestEagerWarmContainersScaleUp(TestWarmContainersBaseClass): + """ + Test EAGER mode warm containers scale-up behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.EAGER.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_scales_up_under_load(self): + """Concurrent invocations beyond initial containers trigger scale-up.""" + from concurrent.futures import ThreadPoolExecutor + import time + + initial_count = self.count_running_containers() + # EAGER mode creates one container per function at startup + # Template has 3 functions: HelloWorldFunction, EchoEventFunction, SleepFunction + self.assertEqual(initial_count, 3) + + def invoke_sleep(): + return self.lambda_client.invoke(FunctionName="SleepFunction") + + # Use SleepFunction to ensure invocations are truly concurrent + with ThreadPoolExecutor(max_workers=10) as executor: + # Submit all futures first to ensure true concurrency + futures = [] + for i in range(10): + future = executor.submit(invoke_sleep) + futures.append(future) + + time.sleep(1.5) + container_count = self.count_running_containers() + self.assertGreater(container_count, 3, "Should scale up to handle concurrent invocations") + results = [f.result() for f in futures] + + for result in results: + self.assertEqual(result.get("StatusCode"), 200) + + sleep(2) # Wait for containers to finish processing + + +class TestEagerWarmContainersReuse(TestWarmContainersBaseClass): + """ + Test EAGER mode warm containers reuse behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.EAGER.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_scaled_containers_reused(self): + """Scaled containers are reused in subsequent invocations.""" + from concurrent.futures import ThreadPoolExecutor + import time + + def invoke_sleep(): + return self.lambda_client.invoke(FunctionName="SleepFunction") + + # First batch: scale up using SleepFunction + with ThreadPoolExecutor(max_workers=10) as executor: + # Submit all futures first to ensure true concurrency + futures = [] + for i in range(10): + future = executor.submit(invoke_sleep) + futures.append(future) + + time.sleep(1.5) + container_count_after_scale = self.count_running_containers() + self.assertGreater(container_count_after_scale, 3, "Should scale up during first batch") + first_batch = [f.result() for f in futures] + + for result in first_batch: + self.assertEqual(result.get("StatusCode"), 200) + + sleep(2) # Wait for containers to finish processing + + # Second batch: verify containers are reused (no additional scale-up) + with ThreadPoolExecutor(max_workers=10) as executor: + # Submit all futures first to ensure true concurrency + futures = [] + for i in range(10): + future = executor.submit(invoke_sleep) + futures.append(future) + + time.sleep(1.5) + container_count_after_reuse = self.count_running_containers() + self.assertEqual( + container_count_after_reuse, container_count_after_scale, "Should reuse existing containers" + ) + second_batch = [f.result() for f in futures] + + for result in second_batch: + self.assertEqual(result.get("StatusCode"), 200) + + sleep(2) # Wait for containers to finish processing + + +class TestEagerWarmContainersCleanup(TestWarmContainersBaseClass): + """ + Test EAGER mode warm containers cleanup behavior. + + Note: Each test class creates isolated container environments to avoid noisy neighbor + issues where containers from one test affect another test's behavior. This is especially + important for SIGTERM tests since they terminate the server process. + """ + + template_path = "/testdata/start_api/template-warm-containers.yaml" + container_mode = ContainersInitializationMode.EAGER.value + mode_env_variable = str(uuid.uuid4()) + parameter_overrides = {"ModeEnvVariable": mode_env_variable} + + @skipIf(IS_WINDOWS, "SIGTERM interrupt doesn't exist on Windows") + @pytest.mark.flaky(reruns=3) + @pytest.mark.timeout(timeout=600, method="thread") + def test_sigterm_cleans_up_all_containers(self): + """SIGTERM cleans up all containers including scaled ones.""" + initial_count = self.count_running_containers() + self.assertEqual(initial_count, 2) + + results = send_concurrent_requests( + lambda: self.lambda_client.invoke(FunctionName="HelloWorldFunction"), count=5 + ) + for result in results: + self.assertEqual(result.get("StatusCode"), 200) + + sleep(2) # Wait for containers to finish processing + scaled_count = self.count_running_containers() + self.assertGreater(scaled_count, initial_count) + + self.start_lambda_process.send_signal(signal.SIGTERM) + sleep(10) + + remaining_containers = self.count_running_containers() + self.assertEqual(remaining_containers, 0) + self.assertEqual(self.start_lambda_process.poll(), 0) + + class TestImagePackageType(StartLambdaIntegBaseClass): template_path = "/testdata/start_api/image_package_type/template.yaml" build_before_invoke = True diff --git a/tests/integration/testdata/start_api/template-warm-containers.yaml b/tests/integration/testdata/start_api/template-warm-containers.yaml index 889d4c2b621..039157e6649 100644 --- a/tests/integration/testdata/start_api/template-warm-containers.yaml +++ b/tests/integration/testdata/start_api/template-warm-containers.yaml @@ -36,6 +36,23 @@ Resources: Method: GET Path: /proxypath/{proxy+} + SleepFunction: + Type: AWS::Serverless::Function + Properties: + Handler: main.sleep_10_sec_handler + Runtime: python3.13 + CodeUri: . + Timeout: 600 + Environment: + Variables: + MODE: !Ref ModeEnvVariable + Events: + SleepPath: + Type: Api + Properties: + Method: POST + Path: /sleep + EchoEventFunction: Type: AWS::Serverless::Function Properties: diff --git a/tests/unit/local/docker/test_container.py b/tests/unit/local/docker/test_container.py index 8a31996386f..09f5e2a8e77 100644 --- a/tests/unit/local/docker/test_container.py +++ b/tests/unit/local/docker/test_container.py @@ -1350,3 +1350,178 @@ def test_resolves_symlink(self, mock_path, mock_realpath, mock_basename, mock_sc volumes = self.container._create_mapped_symlink_files() self.assertEqual(volumes, {host_path: {"bind": container_path, "mode": ANY}}) + + +class TestContainer_capacity_signaling(TestCase): + """Tests for container capacity signaling interface""" + + def setUp(self): + self.image = IMAGE + self.cmd = "cmd" + self.working_dir = "working_dir" + self.host_dir = "host_dir" + self.mock_docker_client = Mock() + + self.container = Container( + self.image, + self.cmd, + self.working_dir, + self.host_dir, + docker_client=self.mock_docker_client, + ) + self.container.id = "test_container_id" + + @parameterized.expand( + [ + (True, False), + (False, True), + ] + ) + def test_has_capacity_returns_correct_result(self, is_at_capacity, expected_has_capacity): + """Test has_capacity() returns correct result""" + self.container._is_at_capacity = is_at_capacity + + result = self.container.has_capacity() + + self.assertEqual(expected_has_capacity, result) + + @patch("samcli.local.docker.container.LOG") + def test_mark_busy_sets_is_at_capacity_true(self, mock_log): + """Test _mark_busy() sets _is_at_capacity to True""" + self.container._is_at_capacity = False + + self.container._mark_busy() + + mock_log.debug.assert_called_once() + call_args = mock_log.debug.call_args[0] + self.assertIn("marked as busy", call_args[0]) + self.assertTrue(self.container._is_at_capacity) + + @patch("samcli.local.docker.container.LOG") + def test_mark_idle_sets_is_at_capacity_false(self, mock_log): + """Test _mark_idle() sets _is_at_capacity to False""" + self.container._is_at_capacity = True + + self.container._mark_idle() + + mock_log.debug.assert_called_once() + call_args = mock_log.debug.call_args[0] + self.assertIn("marked as idle", call_args[0]) + self.assertFalse(self.container._is_at_capacity) + + def test_thread_safe_concurrent_access_to_state(self): + """Test thread-safe concurrent access to container state with _busy_lock""" + import threading + + self.container._is_at_capacity = False + state_changes = [] + + def mark_busy_multiple_times(): + for _ in range(10): + self.container._mark_busy() + state_changes.append(("busy", self.container._is_at_capacity)) + self.container._mark_idle() + state_changes.append(("idle", self.container._is_at_capacity)) + + def check_capacity_multiple_times(): + for _ in range(10): + capacity = self.container.has_capacity() + state_changes.append(("check", capacity)) + + # Create threads that will access state concurrently + thread1 = threading.Thread(target=mark_busy_multiple_times) + thread2 = threading.Thread(target=check_capacity_multiple_times) + + thread1.start() + thread2.start() + + thread1.join() + thread2.join() + + # Verify all state changes are consistent (no corrupted state) + for change_type, value in state_changes: + if change_type == "busy": + self.assertTrue(value, "State should be True after _mark_busy()") + elif change_type == "idle": + self.assertFalse(value, "State should be False after _mark_idle()") + # check operations can return either True or False depending on timing + + def _setup_wait_for_result_mocks(self, patched_socket): + """Helper to setup common mocks for wait_for_result tests""" + self.container.is_created = Mock(return_value=True) + self.container._is_at_capacity = False + + real_container_mock = Mock() + self.mock_docker_client.containers.get.return_value = real_container_mock + real_container_mock.attach.return_value = Mock() + + socket_mock = Mock() + socket_mock.connect_ex.return_value = 0 + patched_socket.return_value = socket_mock + + @patch("socket.socket") + @patch("samcli.local.docker.container.requests") + def test_wait_for_result_marks_idle_after_success(self, mock_requests, patched_socket): + """Test wait_for_result() marks container idle after successful execution""" + self._setup_wait_for_result_mocks(patched_socket) + + response = Mock() + response.content = b'{"result": "success"}' + response.headers = {"Content-Type": "application/json"} + mock_requests.post.return_value = response + + self.container.wait_for_result(event="{}", full_path="test_function", stdout=Mock(), stderr=Mock()) + + self.assertFalse(self.container._is_at_capacity) + + @patch("socket.socket") + @patch("samcli.local.docker.container.requests") + def test_wait_for_result_marks_idle_after_exception(self, mock_requests, patched_socket): + """Test wait_for_result() marks container idle in finally block even on exception""" + self._setup_wait_for_result_mocks(patched_socket) + + mock_requests.post.side_effect = ContainerResponseException("Test exception") + + with self.assertRaises(ContainerResponseException): + self.container.wait_for_result(event="{}", full_path="test_function", stdout=Mock(), stderr=Mock()) + + self.assertFalse(self.container._is_at_capacity) + + def test_state_transitions_during_request_lifecycle(self): + """Test complete state transitions during request lifecycle""" + # Initial state should be idle + self.assertFalse(self.container._is_at_capacity) + self.assertTrue(self.container.has_capacity()) + + # Mark busy (simulating request start) + self.container._mark_busy() + self.assertTrue(self.container._is_at_capacity) + self.assertFalse(self.container.has_capacity()) + + # Mark idle (simulating request completion) + self.container._mark_idle() + self.assertFalse(self.container._is_at_capacity) + self.assertTrue(self.container.has_capacity()) + + def test_multiple_state_transitions(self): + """Test multiple state transitions work correctly""" + # Start idle + self.assertFalse(self.container._is_at_capacity) + + # First request + self.container._mark_busy() + self.assertTrue(self.container._is_at_capacity) + self.container._mark_idle() + self.assertFalse(self.container._is_at_capacity) + + # Second request + self.container._mark_busy() + self.assertTrue(self.container._is_at_capacity) + self.container._mark_idle() + self.assertFalse(self.container._is_at_capacity) + + # Third request + self.container._mark_busy() + self.assertTrue(self.container._is_at_capacity) + self.container._mark_idle() + self.assertFalse(self.container._is_at_capacity) diff --git a/tests/unit/local/lambdafn/test_runtime.py b/tests/unit/local/lambdafn/test_runtime.py index dffba424664..efa7789605d 100644 --- a/tests/unit/local/lambdafn/test_runtime.py +++ b/tests/unit/local/lambdafn/test_runtime.py @@ -872,7 +872,7 @@ def test_must_create_non_cached_container(self, LambdaContainerMock, LambdaFunct self.manager_mock.create.assert_called_with(container, ContainerContext.INVOKE) # validate that the created container got cached - self.assertEqual(self.runtime._containers[self.full_path], container) + self.assertIn(container, self.runtime._containers[self.full_path]) lambda_function_observer_mock.watch.assert_called_with(self.func_config) lambda_function_observer_mock.start.assert_called_with() @@ -944,7 +944,7 @@ def test_must_create_incase_function_config_changed(self, LambdaContainerMock, L ) self.manager_mock.stop.assert_called_with(container) # validate that the created container got cached - self.assertEqual(self.runtime._containers[self.full_path], container2) + self.assertIn(container2, self.runtime._containers[self.full_path]) self.assertEqual(result, container2) @patch("samcli.local.lambdafn.runtime.LambdaFunctionObserver") @@ -1013,7 +1013,7 @@ def test_must_ignore_debug_options_if_function_name_is_not_debug_function( ) self.manager_mock.create.assert_called_with(container, ContainerContext.INVOKE) # validate that the created container got cached - self.assertEqual(self.runtime._containers[self.full_path], container) + self.assertIn(container, self.runtime._containers[self.full_path]) class TestWarmLambdaRuntime_get_code_dir(TestCase): @@ -1057,10 +1057,12 @@ def setUp(self): self.runtime = WarmLambdaRuntime(self.manager_mock, lambda_image_mock, self.observer_mock) self.func1_container_mock = Mock() + self.func1_container_mock.id = "container1_id" self.func2_container_mock = Mock() + self.func2_container_mock.id = "container2_id" self.runtime._containers = { - "func_name1": self.func1_container_mock, - "func_name2": self.func2_container_mock, + "func_name1": [self.func1_container_mock], + "func_name2": [self.func2_container_mock], } self.runtime._temp_uncompressed_paths_to_be_cleaned = ["path1", "path2"] self.runtime._lock = MagicMock() @@ -1140,8 +1142,12 @@ def setUp(self): self.func1_container_mock = Mock() self.func2_container_mock = Mock() self.runtime._containers = { - self.func1_full_path: self.func1_container_mock, - self.func2_full_path: self.func2_container_mock, + self.func1_full_path: [self.func1_container_mock], + self.func2_full_path: [self.func2_container_mock], + } + self.runtime._function_configs = { + self.func1_full_path: self.func_config1, + self.func2_full_path: self.func_config2, } def test_only_one_container_get_stopped_when_its_code_dir_got_changed(self): @@ -1151,7 +1157,8 @@ def test_only_one_container_get_stopped_when_its_code_dir_got_changed(self): self.assertEqual( self.runtime._containers, { - self.func2_full_path: self.func2_container_mock, + self.func1_full_path: [], + self.func2_full_path: [self.func2_container_mock], }, ) @@ -1167,7 +1174,7 @@ def test_both_containers_get_stopped_when_both_functions_got_updated(self): call(self.func2_container_mock), ], ) - self.assertEqual(self.runtime._containers, {}) + self.assertEqual(self.runtime._containers, {self.func1_full_path: [], self.func2_full_path: []}) self.assertEqual( self.observer_mock.unwatch.call_args_list, @@ -1221,172 +1228,96 @@ def test_must_unzip_posix(self, os_mock, unzip_mock, tempfile_mock): class TestRequireContainerReloading(TestCase): - def test_function_should_reloaded_if_runtime_changed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [], - "x86_64", - ) - - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.8", - "app.handler", - None, - None, - ZIP, - "/code", - [], - "x86_64", - ) - self.assertTrue(_require_container_reloading(func, updated_func)) - - def test_function_should_reloaded_if_handler_changed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [], - "x86_64", - ) - - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler1", - None, - None, - ZIP, - "/code", - [], - "x86_64", - ) - self.assertTrue(_require_container_reloading(func, updated_func)) - - def test_function_should_reloaded_if_package_type_changed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - "imageUri", - None, - IMAGE, - None, - [], - "x86_64", - ) - - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [], - "x86_64", - ) - self.assertTrue(_require_container_reloading(func, updated_func)) - - def test_function_should_reloaded_if_image_uri_changed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - "imageUri", - None, - IMAGE, - None, - [], - "x86_64", - ) - - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - "imageUri1", - None, - IMAGE, - None, - [], - "x86_64", - ) - self.assertTrue(_require_container_reloading(func, updated_func)) + @parameterized.expand( + [ + ( + "runtime_changed", + {"runtime": "python3.12"}, + {"runtime": "python3.8"}, + ), + ( + "handler_changed", + {"handler": "app.handler"}, + {"handler": "app.handler1"}, + ), + ( + "package_type_changed", + {"packagetype": IMAGE, "imageuri": "imageUri", "code_path": None}, + {"packagetype": ZIP, "imageuri": None, "code_path": "/code"}, + ), + ( + "image_uri_changed", + {"packagetype": IMAGE, "imageuri": "imageUri", "code_path": None}, + {"packagetype": IMAGE, "imageuri": "imageUri1", "code_path": None}, + ), + ( + "image_config_changed", + { + "packagetype": IMAGE, + "imageuri": "imageUri", + "imageconfig": {"WorkingDirectory": "/opt"}, + "code_path": None, + }, + { + "packagetype": IMAGE, + "imageuri": "imageUri", + "imageconfig": {"WorkingDirectory": "/var"}, + "code_path": None, + }, + ), + ( + "code_path_changed", + {"code_path": "/code2"}, + {"code_path": "/code"}, + ), + ] + ) + def test_function_should_reloaded_when_config_changed(self, name, original_overrides, updated_overrides): + """Test that container reloading is required when function config changes""" + # Default config values + defaults = { + "name": "name", + "full_path": "stack/name", + "runtime": "python3.12", + "handler": "app.handler", + "imageuri": None, + "imageconfig": None, + "packagetype": ZIP, + "code_path": "/code", + "layers": [], + "architecture": "x86_64", + } - def test_function_should_reloaded_if_image_config_changed(self): + # Create original config + original_config = {**defaults, **original_overrides} func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - "imageUri", - {"WorkingDirectory": "/opt"}, - IMAGE, - None, - [], - "x86_64", + original_config["name"], + original_config["full_path"], + original_config["runtime"], + original_config["handler"], + original_config["imageuri"], + original_config["imageconfig"], + original_config["packagetype"], + original_config["code_path"], + original_config["layers"], + original_config["architecture"], ) + # Create updated config + updated_config = {**defaults, **updated_overrides} updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - "imageUri", - {"WorkingDirectory": "/var"}, - IMAGE, - None, - [], - "x86_64", - ) - self.assertTrue(_require_container_reloading(func, updated_func)) - - def test_function_should_reloaded_if_code_path_changed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code2", - [], - "x86_64", + updated_config["name"], + updated_config["full_path"], + updated_config["runtime"], + updated_config["handler"], + updated_config["imageuri"], + updated_config["imageconfig"], + updated_config["packagetype"], + updated_config["code_path"], + updated_config["layers"], + updated_config["architecture"], ) - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [], - "x86_64", - ) self.assertTrue(_require_container_reloading(func, updated_func)) def test_function_should_reloaded_if_env_vars_changed(self): @@ -1401,12 +1332,7 @@ def test_function_should_reloaded_if_env_vars_changed(self): "/code", [], "x86_64", - env_vars=EnvironmentVariables( - variables={ - "key1": "value1", - "key2": "value2", - } - ), + env_vars=EnvironmentVariables(variables={"key1": "value1", "key2": "value2"}), ) updated_func = FunctionConfig( @@ -1420,50 +1346,50 @@ def test_function_should_reloaded_if_env_vars_changed(self): "/code", [], "x86_64", - env_vars=EnvironmentVariables( - variables={ - "key1": "value1", - } - ), - ) - self.assertTrue(_require_container_reloading(func, updated_func)) - - def test_function_should_reloaded_if_one_layer_removed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [ - LayerVersion("Layer", "/somepath", stack_path=""), - LayerVersion("ServerlessLayer", "/somepath2", stack_path=""), - LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), - ], - "x86_64", - ) - - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [ - LayerVersion("Layer", "/somepath", stack_path=""), - LayerVersion("ServerlessLayer", "/somepath2", stack_path=""), - ], - "x86_64", + env_vars=EnvironmentVariables(variables={"key1": "value1"}), ) self.assertTrue(_require_container_reloading(func, updated_func)) - def test_function_should_reloaded_if_one_layer_added(self): + @parameterized.expand( + [ + ( + "one_layer_removed", + [ + LayerVersion("Layer", "/somepath", stack_path=""), + LayerVersion("ServerlessLayer", "/somepath2", stack_path=""), + LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), + ], + [ + LayerVersion("Layer", "/somepath", stack_path=""), + LayerVersion("ServerlessLayer", "/somepath2", stack_path=""), + ], + ), + ( + "one_layer_added", + [ + LayerVersion("Layer", "/somepath", stack_path=""), + LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), + ], + [ + LayerVersion("Layer", "/somepath", stack_path=""), + LayerVersion("ServerlessLayer", "/somepath2", stack_path=""), + ], + ), + ( + "layers_changed", + [ + LayerVersion("Layer", "/somepath", stack_path=""), + LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), + ], + [ + LayerVersion("Layer", "/somepath2", stack_path=""), + LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), + ], + ), + ] + ) + def test_function_should_reloaded_when_layers_changed(self, name, original_layers, updated_layers): + """Test that container reloading is required when layers change""" func = FunctionConfig( "name", "stack/name", @@ -1473,10 +1399,7 @@ def test_function_should_reloaded_if_one_layer_added(self): None, ZIP, "/code", - [ - LayerVersion("Layer", "/somepath", stack_path=""), - LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), - ], + original_layers, "x86_64", ) @@ -1489,46 +1412,10 @@ def test_function_should_reloaded_if_one_layer_added(self): None, ZIP, "/code", - [ - LayerVersion("Layer", "/somepath", stack_path=""), - LayerVersion("ServerlessLayer", "/somepath2", stack_path=""), - ], - "x86_64", - ) - self.assertTrue(_require_container_reloading(func, updated_func)) - - def test_function_should_reloaded_if_layers_changed(self): - func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [ - LayerVersion("Layer", "/somepath", stack_path=""), - LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), - ], + updated_layers, "x86_64", ) - updated_func = FunctionConfig( - "name", - "stack/name", - "python3.12", - "app.handler", - None, - None, - ZIP, - "/code", - [ - LayerVersion("Layer", "/somepath2", stack_path=""), - LayerVersion("arn:aws:lambda:region:account-id:layer:layer-name:1", None, stack_path=""), - ], - "x86_64", - ) self.assertTrue(_require_container_reloading(func, updated_func)) def test_function_should_not_reloaded_if_nothing_changed(self): @@ -1769,7 +1656,7 @@ def test_create_with_existing_config_and_no_container_stops_none_container(self, # Set up existing config but no container self.runtime._function_configs[self.full_path] = existing_config - self.runtime._containers[self.full_path] = None # No container + self.runtime._containers[self.full_path] = [] # No container container = Mock() LambdaContainerMock.return_value = container @@ -1784,7 +1671,7 @@ def test_create_with_existing_config_and_no_container_stops_none_container(self, self.observer_mock.unwatch.assert_called_once_with(existing_config) # Verify new container was created and stored self.assertEqual(result, container) - self.assertEqual(self.runtime._containers[self.full_path], container) + self.assertIn(container, self.runtime._containers[self.full_path]) # Verify new function config was stored (the old one was replaced) self.assertIn(self.full_path, self.runtime._function_configs) self.assertEqual(self.runtime._function_configs[self.full_path], self.func_config) @@ -1912,7 +1799,7 @@ def test_on_code_change_with_no_container_doesnt_stop_container(self): """Test _on_code_change when no container exists for the function - lines 589->577""" # Set up function config but no container self.runtime._function_configs[self.func_config.full_path] = self.func_config - self.runtime._containers[self.func_config.full_path] = None # No container + self.runtime._containers[self.func_config.full_path] = [] # No container with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: self.runtime._on_code_change([self.func_config]) @@ -1932,15 +1819,15 @@ def test_on_code_change_with_container_stops_container(self): # Set up function config and container self.runtime._function_configs[self.func_config.full_path] = self.func_config - self.runtime._containers[self.func_config.full_path] = container + self.runtime._containers[self.func_config.full_path] = [container] with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: self.runtime._on_code_change([self.func_config]) # Verify function config was removed self.assertNotIn(self.func_config.full_path, self.runtime._function_configs) - # Verify container was removed - self.assertNotIn(self.func_config.full_path, self.runtime._containers) + # Verify container list was cleared + self.assertEqual(self.runtime._containers[self.func_config.full_path], []) # Verify container stop was called self.manager_mock.stop.assert_called_once_with(container) # Verify observer unwatch was called @@ -1968,7 +1855,7 @@ def test_on_code_change_with_image_package_type_logs_image_resource(self): container = Mock() self.runtime._function_configs[image_func_config.full_path] = image_func_config - self.runtime._containers[image_func_config.full_path] = container + self.runtime._containers[image_func_config.full_path] = [container] with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: self.runtime._on_code_change([image_func_config]) @@ -1979,3 +1866,559 @@ def test_on_code_change_with_image_package_type_logs_image_resource(self): # The log format is: "Lambda Function '%s' %s has been changed..." # where %s is function_full_path and %s is resource (imageuri + " image") self.assertIn("my-image:latest image", log_call_args[2]) + + +class TestWarmLambdaRuntime_CleanupLogic(TestCase): + """Test cleanup logic for WarmLambdaRuntime""" + + def setUp(self): + self.manager_mock = Mock() + self.lambda_image_mock = Mock() + self.observer_mock = Mock() + self.runtime = WarmLambdaRuntime(self.manager_mock, self.lambda_image_mock, observer=self.observer_mock) + + def _create_containers(self, count, id_prefix="container"): + """Helper to create mock containers""" + from collections import defaultdict + + containers = [] + for i in range(count): + container = Mock() + container.id = f"{id_prefix}{i}_id" + containers.append(container) + return containers + + def _setup_containers(self, container_map): + """Helper to setup containers for multiple functions""" + from collections import defaultdict + + self.runtime._containers = defaultdict(list) + for func_path, containers in container_map.items(): + self.runtime._containers[func_path] = containers + + def test_cleanup_with_multiple_containers_per_function(self): + """Test cleanup stops all containers when multiple containers exist for a single function""" + containers = self._create_containers(3) + self._setup_containers({"stack/function1": containers}) + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + self.assertEqual(self.manager_mock.stop.call_count, 3) + for container in containers: + self.manager_mock.stop.assert_any_call(container) + + LogMock.debug.assert_any_call("Container cleanup complete: %d/%d successful, %d failed", 3, 3, 0) + + def test_cleanup_with_multiple_functions(self): + """Test cleanup stops all containers across multiple functions""" + func1_containers = self._create_containers(2, "func1_c") + func2_containers = self._create_containers(1, "func2_c") + func3_containers = self._create_containers(3, "func3_c") + + self._setup_containers( + { + "stack/function1": func1_containers, + "stack/function2": func2_containers, + "stack/function3": func3_containers, + } + ) + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + self.assertEqual(self.manager_mock.stop.call_count, 6) + LogMock.debug.assert_any_call("Container cleanup complete: %d/%d successful, %d failed", 6, 6, 0) + + def test_cleanup_continues_after_individual_failures(self): + """Test cleanup continues even when individual container stops fail""" + containers = self._create_containers(4) + self._setup_containers( + { + "stack/function1": containers[:2], + "stack/function2": containers[2:], + } + ) + + # Make containers[1] and containers[2] fail to stop + self.manager_mock.stop.side_effect = lambda c: ( + None if c in [containers[0], containers[3]] else (_ for _ in ()).throw(Exception(f"Failed to stop {c.id}")) + ) + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + self.assertEqual(self.manager_mock.stop.call_count, 4) + LogMock.debug.assert_any_call("Container cleanup complete: %d/%d successful, %d failed", 2, 4, 2) + + def test_cleanup_is_idempotent(self): + """Test cleanup can be called multiple times without errors""" + containers = self._create_containers(2) + self._setup_containers({"stack/function1": containers}) + + with patch("samcli.local.lambdafn.runtime.LOG"): + self.runtime.clean_running_containers_and_related_resources() + + self.assertEqual(self.manager_mock.stop.call_count, 2) + self.manager_mock.reset_mock() + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + self.assertEqual(self.manager_mock.stop.call_count, 2) + LogMock.debug.assert_any_call("Container cleanup complete: %d/%d successful, %d failed", 2, 2, 0) + + def test_cleanup_summary_logging(self): + """Test cleanup logs summary with success/failure counts""" + containers = self._create_containers(3) + self._setup_containers({"stack/function1": containers}) + + self.manager_mock.stop.side_effect = lambda c: ( + None if c != containers[1] else (_ for _ in ()).throw(Exception("Stop failed")) + ) + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + LogMock.debug.assert_any_call("Container cleanup complete: %d/%d successful, %d failed", 2, 3, 1) + debug_calls = [str(call) for call in LogMock.debug.call_args_list] + terminate_logs = [call for call in debug_calls if "Terminate warm container" in call] + self.assertEqual(len(terminate_logs), 3) + + @parameterized.expand([("with_container", Mock()), ("with_none", None)]) + def test_on_invoke_done_does_not_stop_containers_in_warm_mode(self, name, container): + """Test _on_invoke_done does NOT stop containers in warm mode""" + self.runtime._on_invoke_done(container) + self.manager_mock.stop.assert_not_called() + self.observer_mock.stop.assert_not_called() + + +class TestWarmLambdaRuntime_StopAllContainersForFunction(TestCase): + """Test _stop_all_containers_for_function helper method - Task 4.5""" + + def setUp(self): + self.manager_mock = Mock() + self.lambda_image_mock = Mock() + self.observer_mock = Mock() + self.runtime = WarmLambdaRuntime(self.manager_mock, self.lambda_image_mock, observer=self.observer_mock) + self.func_path = "stack/function1" + self.func_config = FunctionConfig( + "function1", self.func_path, "python3.9", "handler", None, None, ZIP, "code-path", [], "x86_64" + ) + + def _setup_function(self, container_count): + """Helper to setup function with containers""" + from collections import defaultdict + + self.runtime._containers = defaultdict(list) + self.runtime._function_configs[self.func_path] = self.func_config + containers = [Mock(id=f"container{i}_id") for i in range(container_count)] + self.runtime._containers[self.func_path] = containers + return containers + + def test_stop_all_containers_for_function_stops_all_containers(self): + """Test _stop_all_containers_for_function stops all containers for a specific function""" + containers = self._setup_function(3) + + with patch("samcli.local.lambdafn.runtime.LOG"): + self.runtime._stop_all_containers_for_function(self.func_path) + + self.assertEqual(self.manager_mock.stop.call_count, 3) + self.assertEqual(self.runtime._containers[self.func_path], []) + self.assertNotIn(self.func_path, self.runtime._function_configs) + self.observer_mock.unwatch.assert_called_once_with(self.func_config) + + def test_stop_all_containers_handles_errors_gracefully(self): + """Test _stop_all_containers_for_function handles errors gracefully""" + containers = self._setup_function(2) + self.manager_mock.stop.side_effect = lambda c: ( + None if c != containers[0] else (_ for _ in ()).throw(Exception("Stop failed")) + ) + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime._stop_all_containers_for_function(self.func_path) + + self.assertEqual(self.manager_mock.stop.call_count, 2) + LogMock.debug.assert_any_call("Failed to stop container %s: %s", "container0_id", "Stop failed") + self.assertEqual(self.runtime._containers[self.func_path], []) + self.assertNotIn(self.func_path, self.runtime._function_configs) + + def test_stop_all_containers_with_no_containers(self): + """Test _stop_all_containers_for_function handles function with no containers""" + self._setup_function(0) + + with patch("samcli.local.lambdafn.runtime.LOG"): + self.runtime._stop_all_containers_for_function(self.func_path) + + self.manager_mock.stop.assert_not_called() + self.assertEqual(self.runtime._containers[self.func_path], []) + self.assertNotIn(self.func_path, self.runtime._function_configs) + self.observer_mock.unwatch.assert_called_once_with(self.func_config) + + def test_stop_all_containers_with_nonexistent_function(self): + """Test _stop_all_containers_for_function handles nonexistent function gracefully""" + from collections import defaultdict + + self.runtime._containers = defaultdict(list) + self.runtime._function_configs = {} + + with patch("samcli.local.lambdafn.runtime.LOG"): + self.runtime._stop_all_containers_for_function("stack/nonexistent") + + self.manager_mock.stop.assert_not_called() + self.observer_mock.unwatch.assert_not_called() + + +class TestWarmLambdaRuntime_CleanupResourcesHandling(TestCase): + """Test cleanup handles decompressed paths and observer cleanup - Task 4.5""" + + def setUp(self): + from collections import defaultdict + + self.manager_mock = Mock() + self.lambda_image_mock = Mock() + self.observer_mock = Mock() + self.runtime = WarmLambdaRuntime(self.manager_mock, self.lambda_image_mock, observer=self.observer_mock) + self.runtime._containers = defaultdict(list) + + def test_cleanup_calls_clean_decompressed_paths(self): + """Test cleanup calls _clean_decompressed_paths""" + self.runtime._clean_decompressed_paths = Mock() + + with patch("samcli.local.lambdafn.runtime.LOG"): + self.runtime.clean_running_containers_and_related_resources() + + self.runtime._clean_decompressed_paths.assert_called_once() + + def test_cleanup_calls_observer_stop(self): + """Test cleanup calls observer.stop()""" + with patch("samcli.local.lambdafn.runtime.LOG"): + self.runtime.clean_running_containers_and_related_resources() + + self.observer_mock.stop.assert_called_once() + + def test_cleanup_handles_decompressed_paths_error(self): + """Test cleanup continues if _clean_decompressed_paths fails""" + container = Mock(id="container_id") + self.runtime._containers["stack/function1"] = [container] + self.runtime._clean_decompressed_paths = Mock(side_effect=Exception("Cleanup failed")) + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + self.manager_mock.stop.assert_called_once_with(container) + LogMock.debug.assert_any_call("Failed to clean decompressed paths: %s", "Cleanup failed") + self.observer_mock.stop.assert_called_once() + + def test_cleanup_handles_observer_stop_error(self): + """Test cleanup continues if observer.stop() fails""" + container = Mock(id="container_id") + self.runtime._containers["stack/function1"] = [container] + self.observer_mock.stop.side_effect = Exception("Observer stop failed") + + with patch("samcli.local.lambdafn.runtime.LOG") as LogMock: + self.runtime.clean_running_containers_and_related_resources() + + self.manager_mock.stop.assert_called_once_with(container) + LogMock.debug.assert_any_call("Failed to stop observer: %s", "Observer stop failed") + + +class TestWarmLambdaRuntime_ContainerSelection(TestCase): + """Tests for container selection logic in WarmLambdaRuntime""" + + def setUp(self): + self.manager_mock = Mock() + self.lambda_image_mock = Mock() + self.observer_mock = Mock() + self.runtime = WarmLambdaRuntime(self.manager_mock, self.lambda_image_mock, self.observer_mock) + self.full_path = "stack/test_function" + self.func_config = FunctionConfig( + "test_function", self.full_path, "python3.11", "app.handler", None, None, ZIP, "/code", [], "x86_64" + ) + + def _create_container(self, container_id, is_created=True, has_capacity=True): + """Helper to create a mock container with specified properties""" + container = Mock() + container.id = container_id + container.is_created.return_value = is_created + container.has_capacity.return_value = has_capacity + container._mark_busy = Mock() + return container + + def test_list_initialization_with_defaultdict(self): + """Test that _containers uses defaultdict(list) for automatic list initialization""" + from collections import defaultdict + + # Verify _containers is a defaultdict + self.assertIsInstance(self.runtime._containers, defaultdict) + + # Verify accessing a new key creates an empty list + function_path = "stack/new_function" + containers_list = self.runtime._containers[function_path] + self.assertIsInstance(containers_list, list) + self.assertEqual(len(containers_list), 0) + + def test_list_appending_with_multiple_containers(self): + """Test that containers are appended to list, not overwritten""" + from collections import defaultdict + + container1 = Mock() + container1.id = "container1" + container2 = Mock() + container2.id = "container2" + container3 = Mock() + container3.id = "container3" + + # Manually append containers to simulate concurrent creation + self.runtime._containers[self.full_path].append(container1) + self.runtime._containers[self.full_path].append(container2) + self.runtime._containers[self.full_path].append(container3) + + # Verify all containers are in the list + containers = self.runtime._containers[self.full_path] + self.assertEqual(len(containers), 3) + self.assertEqual(containers[0].id, "container1") + self.assertEqual(containers[1].id, "container2") + self.assertEqual(containers[2].id, "container3") + + def test_find_available_container_returns_first_with_capacity(self): + """Test _find_available_container returns first container with capacity""" + containers = [ + self._create_container("c1", has_capacity=False), + self._create_container("c2", has_capacity=True), + self._create_container("c3", has_capacity=True), + ] + self.runtime._containers[self.full_path] = containers + + result = self.runtime._find_available_container(self.full_path) + + self.assertEqual(result, containers[1]) + containers[0].has_capacity.assert_called_once() + containers[1].has_capacity.assert_called_once() + containers[2].has_capacity.assert_not_called() + + def test_find_available_container_returns_none_when_all_busy(self): + """Test _find_available_container returns None when all containers are busy""" + containers = [self._create_container(f"c{i}", has_capacity=False) for i in range(2)] + self.runtime._containers[self.full_path] = containers + + result = self.runtime._find_available_container(self.full_path) + + self.assertIsNone(result) + for container in containers: + container.has_capacity.assert_called_once() + + def test_find_available_container_returns_none_when_no_containers(self): + """Test _find_available_container returns None when no containers exist""" + result = self.runtime._find_available_container(self.full_path) + self.assertIsNone(result) + + @patch("samcli.local.lambdafn.runtime.LambdaContainer") + def test_acquire_container_creates_new_when_all_at_capacity(self, LambdaContainerMock): + """Test _acquire_container creates new container when all are at capacity""" + busy_container = self._create_container("c1", has_capacity=False) + self.runtime._containers[self.full_path] = [busy_container] + + new_container = self._create_container("new") + LambdaContainerMock.return_value = new_container + self.runtime._get_code_dir = MagicMock(return_value="/code/dir") + + result = self.runtime._acquire_container(self.func_config) + + self.assertEqual(result, new_container) + new_container._mark_busy.assert_called_once() + self.assertEqual(len(self.runtime._containers[self.full_path]), 2) + + @patch("samcli.local.lambdafn.runtime.LambdaContainer") + def test_acquire_container_reuses_available_container(self, LambdaContainerMock): + """Test _acquire_container reuses available container instead of creating new one""" + available_container = self._create_container("c1", has_capacity=True) + self.runtime._containers[self.full_path] = [available_container] + self.runtime._function_configs[self.full_path] = self.func_config + + result = self.runtime._acquire_container(self.func_config) + + self.assertEqual(result, available_container) + available_container._mark_busy.assert_called_once() + LambdaContainerMock.assert_not_called() + self.assertEqual(len(self.runtime._containers[self.full_path]), 1) + + def test_mark_busy_called_before_releasing_lock_for_reused_container(self): + """Test that _mark_busy is called before releasing lock when reusing container""" + # Setup available container + available_container = Mock() + available_container.is_created.return_value = True + available_container.has_capacity.return_value = True + available_container._mark_busy = Mock() + + self.runtime._containers[self.full_path] = [available_container] + self.runtime._function_configs[self.full_path] = self.func_config + + # Track lock state when _mark_busy is called + lock_held_during_mark_busy = [] + + def track_mark_busy(): + # Check if lock is held by trying to acquire it (should fail if held) + lock_held = self.runtime._container_lock.locked() + lock_held_during_mark_busy.append(lock_held) + + available_container._mark_busy.side_effect = track_mark_busy + + # Call _acquire_container + self.runtime._acquire_container(self.func_config) + + # Verify _mark_busy was called while lock was held + self.assertTrue(lock_held_during_mark_busy[0], "_mark_busy should be called while lock is held") + + @patch("samcli.local.lambdafn.runtime.LambdaContainer") + def test_mark_busy_called_before_releasing_lock_for_new_container(self, LambdaContainerMock): + """Test that _mark_busy is called before releasing lock when creating new container""" + # Setup new container + new_container = Mock() + new_container._mark_busy = Mock() + LambdaContainerMock.return_value = new_container + + self.runtime._get_code_dir = MagicMock(return_value="/code/dir") + + # Track lock state when _mark_busy is called + lock_held_during_mark_busy = [] + + def track_mark_busy(): + lock_held = self.runtime._container_lock.locked() + lock_held_during_mark_busy.append(lock_held) + + new_container._mark_busy.side_effect = track_mark_busy + + # Call _acquire_container + self.runtime._acquire_container(self.func_config) + + # Verify _mark_busy was called while lock was held + self.assertTrue(lock_held_during_mark_busy[0], "_mark_busy should be called while lock is held") + + def test_thread_safe_concurrent_access(self): + """Test thread-safe concurrent access to container list""" + import threading + + # Setup multiple containers that become busy after being acquired + containers = [] + for i in range(3): + container = Mock() + container.id = f"container{i}" + container.is_created.return_value = True + + # Initially has capacity, but becomes busy after _mark_busy is called + capacity_state = [True] # Use list to allow modification in closure + + def make_has_capacity(state): + def has_capacity(): + return state[0] + + return has_capacity + + def make_mark_busy(state): + def mark_busy(): + state[0] = False # Mark as busy + + return mark_busy + + container.has_capacity = make_has_capacity(capacity_state) + container._mark_busy = make_mark_busy(capacity_state) + containers.append(container) + + self.runtime._containers[self.full_path] = containers + self.runtime._function_configs[self.full_path] = self.func_config + + # Track which containers were acquired + acquired_containers = [] + errors = [] + + def acquire_container_thread(): + try: + container = self.runtime._acquire_container(self.func_config) + acquired_containers.append(container) + except Exception as e: + errors.append(e) + + # Create multiple threads to acquire containers concurrently + threads = [] + for _ in range(3): + thread = threading.Thread(target=acquire_container_thread) + threads.append(thread) + + # Start all threads + for thread in threads: + thread.start() + + # Wait for all threads to complete + for thread in threads: + thread.join(timeout=5) + + # Verify no errors occurred + self.assertEqual(len(errors), 0, f"Errors occurred during concurrent access: {errors}") + + # Verify all threads acquired containers + self.assertEqual(len(acquired_containers), 3) + + # Verify different containers were acquired (due to thread-safe locking) + acquired_ids = [c.id for c in acquired_containers] + self.assertEqual(len(set(acquired_ids)), 3, "Each thread should acquire a different container") + + def test_eager_mode_containers_marked_idle_after_run(self): + """Test that EAGER mode containers are marked idle after run()""" + container = Mock() + container._is_at_capacity = True # Container is marked busy + container._mark_idle = Mock() + + # Mock parent's run method to return the container + with patch.object(LambdaRuntime, "run", return_value=container): + result = self.runtime.run( + container, self.func_config, debug_context=None, container_host=None, container_host_interface=None + ) + + # Verify container was marked idle after starting + container._mark_idle.assert_called_once() + self.assertEqual(result, container) + + def test_eager_mode_containers_not_marked_idle_if_already_idle(self): + """Test that containers already idle are not marked idle again""" + container = Mock() + container._is_at_capacity = False # Container is already idle + container._mark_idle = Mock() + + # Mock parent's run method to return the container + with patch.object(LambdaRuntime, "run", return_value=container): + result = self.runtime.run( + container, self.func_config, debug_context=None, container_host=None, container_host_interface=None + ) + + # Verify _mark_idle was not called (container already idle) + container._mark_idle.assert_not_called() + self.assertEqual(result, container) + + def test_capacity_checking_integration(self): + """Test integration of capacity checking with container selection""" + # Create containers with different capacity states + busy_container = Mock() + busy_container.is_created.return_value = True + busy_container.has_capacity.return_value = False + + idle_container = Mock() + idle_container.is_created.return_value = True + idle_container.has_capacity.return_value = True + idle_container._mark_busy = Mock() + + self.runtime._containers[self.full_path] = [busy_container, idle_container] + self.runtime._function_configs[self.full_path] = self.func_config + + # Acquire container + result = self.runtime._acquire_container(self.func_config) + + # Verify idle container was selected + self.assertEqual(result, idle_container) + + # Verify capacity was checked + busy_container.has_capacity.assert_called_once() + idle_container.has_capacity.assert_called_once() + + # Verify selected container was marked busy + idle_container._mark_busy.assert_called_once()