localstack · bentsku · May 23, 2025 · May 23, 2025 · dominikschubert · Jun 3, 2025
diff --git a/localstack-core/localstack/services/plugins.py b/localstack-core/localstack/services/plugins.py
@@ -13,6 +13,7 @@
 from localstack.aws.skeleton import DispatchTable, Skeleton
 from localstack.aws.spec import load_service
 from localstack.config import ServiceProviderConfig
+from localstack.runtime import hooks
 from localstack.state import StateLifecycleHook, StateVisitable, StateVisitor
 from localstack.utils.bootstrap import get_enabled_apis, is_api_enabled, log_duration
 from localstack.utils.functions import call_safe
@@ -691,3 +692,19 @@ def check_service_health(api, expect_shutdown=False):
         else:
             LOG.warning('Service "%s" still shutting down, retrying...', api)
         raise Exception("Service check failed for api: %s" % api)
+
+
+@hooks.on_infra_start(should_load=lambda: config.EAGER_SERVICE_LOADING)
+def eager_load_services():
+    from localstack.utils.bootstrap import get_preloaded_services
+
+    preloaded_apis = get_preloaded_services()
+    LOG.debug("Eager loading services: %s", sorted(preloaded_apis))
+
+    for api in preloaded_apis:
+        try:
+            SERVICE_PLUGINS.require(api)
+        except ServiceDisabled as e:
+            LOG.debug("%s", e)
+        except Exception:
+            LOG.exception("could not load service plugin %s", api)
diff --git a/localstack-core/localstack/utils/bootstrap.py b/localstack-core/localstack/utils/bootstrap.py
@@ -332,40 +332,27 @@ def get_preloaded_services() -> Set[str]:
 
     The result is cached, so it's safe to call. Clear the cache with get_preloaded_services.cache_clear().
     """
+    if not is_env_true("EAGER_SERVICE_LOADING"):
+        return set()
+
     services_env = os.environ.get("SERVICES", "").strip()
-    services = None
+    services = []
 
-    if services_env and is_env_true("EAGER_SERVICE_LOADING"):
+    if services_env:
         # SERVICES and EAGER_SERVICE_LOADING are set
         # SERVICES env var might contain ports, but we do not support these anymore
-        services = []
         for service_port in re.split(r"\s*,\s*", services_env):
             # Only extract the service name, discard the port
             parts = re.split(r"[:=]", service_port)
             service = parts[0]
             services.append(service)
 
     if not services:
-        from localstack.services.plugins import SERVICE_PLUGINS
-
-        services = SERVICE_PLUGINS.list_available()
+        LOG.warning("No services set in SERVICES environment variable, skipping eager loading.")
 
     return resolve_apis(services)
 
 
-def should_eager_load_api(api: str) -> bool:
-    apis = get_preloaded_services()
-
-    if api in apis:
-        return True
-
-    for enabled_api in apis:
-        if api.startswith(f"{enabled_api}:"):
-            return True
-
-    return False
-
-
 def start_infra_locally():
     from localstack.runtime.main import main
 

@@ -0,0 +1,141 @@
+import pytest
+import requests
+from botocore.exceptions import ClientError
+
+from localstack.config import in_docker
+from localstack.testing.pytest.container import ContainerFactory
+from localstack.utils.bootstrap import ContainerConfigurators, get_gateway_url
+
+pytestmarks = pytest.mark.skipif(
+    condition=in_docker(), reason="cannot run bootstrap tests in docker"
+)
+
+
+def test_strict_service_loading(
+    container_factory: ContainerFactory,
+    wait_for_localstack_ready,
+    aws_client_factory,
+):
+    ls_container = container_factory(
+        configurators=[
+            ContainerConfigurators.random_container_name,
+            ContainerConfigurators.random_gateway_port,
+            ContainerConfigurators.random_service_port_range(20),
+            ContainerConfigurators.env_vars(
+                {"STRICT_SERVICE_LOADING": "1", "SERVICES": "s3,sqs,sns"}
+            ),
+        ]
+    )
+    running_container = ls_container.start()
+    wait_for_localstack_ready(running_container)
+    url = get_gateway_url(ls_container)
+
+    # check service-status returned by health endpoint
+    response = requests.get(f"{url}/_localstack/health")
+    assert response.ok
+
+    services = response.json().get("services")
+
+    assert services.pop("sqs") == "available"
+    assert services.pop("s3") == "available"
+    assert services.pop("sns") == "available"
+
+    assert services
+    assert all(services.get(key) == "disabled" for key in services.keys())
+
+    # activate sqs service
+    client = aws_client_factory(endpoint_url=url)
+    result = client.sqs.list_queues()
+    assert result
+
+    # verify cloudwatch is not activated
+    with pytest.raises(ClientError) as e:
+        client.cloudwatch.list_metrics()
+
+    e.match(
+        "Service 'cloudwatch' is not enabled. Please check your 'SERVICES' configuration variable."
+    )
+    assert e.value.response["ResponseMetadata"]["HTTPStatusCode"] == 501
+
+    # check status again
+    response = requests.get(f"{url}/_localstack/health")
+    assert response.ok
+
+    services = response.json().get("services")
+
+    # sqs should be running now
+    assert services.get("sqs") == "running"
+    assert services.get("s3") == "available"
+    assert services.get("sns") == "available"
+    assert services.get("cloudwatch") == "disabled"
+
+
+def test_eager_service_loading(
+    container_factory: ContainerFactory,
+    wait_for_localstack_ready,
+    aws_client_factory,
+):
+    ls_container = container_factory(
+        configurators=[
+            ContainerConfigurators.random_container_name,
+            ContainerConfigurators.random_gateway_port,
+            ContainerConfigurators.random_service_port_range(20),
+            ContainerConfigurators.env_vars(
+                {"EAGER_SERVICE_LOADING": "1", "SERVICES": "s3,sqs,sns"}
+            ),
+        ]
+    )
+    running_container = ls_container.start()
+    wait_for_localstack_ready(running_container)
+    url = get_gateway_url(ls_container)
+
+    # check service-status returned by health endpoint
+    response = requests.get(f"{url}/_localstack/health")
+    assert response.ok
+
+    services = response.json().get("services")
+
+    assert services.pop("sqs") == "running"
+    assert services.pop("s3") == "running"
+    assert services.pop("sns") == "running"
+
+    assert services
+    assert all(services.get(key) == "disabled" for key in services.keys())
+
+
+def test_eager_and_strict_service_loading(
+    container_factory: ContainerFactory,
+    wait_for_localstack_ready,
+    aws_client_factory,
+):
+    # this is undocumented behavior, to allow eager loading of specific services while not restricting services loading
+    ls_container = container_factory(
+        configurators=[
+            ContainerConfigurators.random_container_name,
+            ContainerConfigurators.random_gateway_port,
+            ContainerConfigurators.random_service_port_range(20),
+            ContainerConfigurators.env_vars(
+                {
+                    "EAGER_SERVICE_LOADING": "1",
+                    "SERVICES": "s3,sqs,sns",
+                    "STRICT_SERVICE_LOADING": "0",
+                }
+            ),
+        ]
+    )
+    running_container = ls_container.start()
+    wait_for_localstack_ready(running_container)
+    url = get_gateway_url(ls_container)
+
+    # check service-status returned by health endpoint
+    response = requests.get(f"{url}/_localstack/health")
+    assert response.ok
+
+    services = response.json().get("services")
+
+    assert services.pop("sqs") == "running"
+    assert services.pop("s3") == "running"
+    assert services.pop("sns") == "running"
+
+    assert services
+    assert all(services.get(key) == "available" for key in services.keys())
diff --git a/tests/unit/utils/test_bootstrap.py b/tests/unit/utils/test_bootstrap.py
@@ -37,13 +37,11 @@ def reset_get_preloaded_services(self):
         yield
         get_preloaded_services.cache_clear()
 
-    def test_returns_default_service_ports(self):
-        from localstack.services.plugins import SERVICE_PLUGINS
-
+    def test_empty_services_returns_no_services(self):
         with temporary_env({"EAGER_SERVICE_LOADING": "1"}):
             result = get_preloaded_services()
 
-        assert result == set(SERVICE_PLUGINS.list_available())
+        assert result == set()
 
     def test_with_service_subset(self):
         with temporary_env({"SERVICES": "s3,sns", "EAGER_SERVICE_LOADING": "1"}):