Fix EC2 throttling during computenode launch is not retried and is reported as ICE when using multi InstanceTypes or SubnetIds

CHANGELOG.md

@@ -3,6 +3,13 @@ aws-parallelcluster-node CHANGELOG
 
 This file is used to list changes made in each version of the aws-parallelcluster-node package.
 
+3.17.0
+------
+
+**BUG FIXES**
+- Fix an issue where EC2 throttling during compute node launch is not retried and is reported as insufficient capacity
+  when using Multiple Instance Types or multiple subnets.
+
 3.16.0
 ------
 

src/slurm_plugin/fleet_manager.py

@@ -30,6 +30,17 @@
 INSTANCE_INFO_RETRIEVAL_TIMEOUT_DEFAULT = 90
 INSTANCE_INFO_RETRIEVAL_MAX_BACKOFF = 30
 
+# The only error code EC2 returns when a launch request is throttled, see
+# https://docs.aws.amazon.com/ec2/latest/devguide/ec2-api-throttling.html
+LAUNCH_THROTTLING_ERROR_CODE = "RequestLimitExceeded"
+
+# An override that CreateFleet did not fulfill because another override failed is reported with this exact
+# code and message pair, which points at the real error rather than being one.
+UNFULFILLED_OVERRIDE_ERROR = (
+    "UnfulfillableCapacity",
+    "Failed to fulfill capacity. Please review errors in the response.",
+)
+
 
 class EC2Instance:
     def __init__(self, id, private_ip, hostname, all_private_ips, launch_time):
@@ -430,7 +441,27 @@ def _launch_instances(self, launch_params):
             if partial_instance_ids:
                 logger.error("Unable to retrieve instance info for instances: %s", partial_instance_ids)
 
-            if not instances and len(err_list) == 1:
+            if not instances:
+                # Drop the entries that only point at the real error, unless the response carries nothing else.
+                real_errors = [
+                    err
+                    for err in err_list
+                    if (err.get("ErrorCode"), err.get("ErrorMessage")) != UNFULFILLED_OVERRIDE_ERROR
+                ]
+                if real_errors:
+                    err_list = real_errors
+                # A single cause is normally left. Should there be several, prefer throttling as a safety net: it is
+                # the only cause that resolves on its own, and reporting it as insufficient capacity would instead
+                # fail the compute resource over.
+                throttling = next(
+                    (err for err in err_list if err.get("ErrorCode") == LAUNCH_THROTTLING_ERROR_CODE), None
+                )
+                if throttling:
+                    raise LaunchInstancesError(throttling.get("ErrorCode"), throttling.get("ErrorMessage"))
+            # Normally a single cause is left. Reporting the first one of several is a second safety net: the
+            # caller otherwise records a hardcoded InsufficientInstanceCapacity, and any code EC2 actually
+            # returned is more useful than an invented one, whichever of them the response happens to list first.
+            if not instances and err_list:
                 raise LaunchInstancesError(err_list[0].get("ErrorCode"), err_list[0].get("ErrorMessage"))
             return {"Instances": instances}
         except ClientError as e:

tests/slurm_plugin/test_fleet_manager.py

@@ -28,6 +28,21 @@
 
 from tests.common import FLEET_CONFIG, MockedBoto3Request
 
+UNFULFILLED_OVERRIDE = {
+    "ErrorCode": "UnfulfillableCapacity",
+    "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+}
+UNSUPPORTED_ERROR = {"ErrorCode": "Unsupported", "ErrorMessage": "Not supported in this AZ."}
+MIN_TARGET_CAPACITY_ERROR = {
+    "ErrorCode": "UnfulfillableCapacity",
+    "ErrorMessage": "Unable to fulfill request due to MinTargetCapacity constraints. Please adjust your request.",
+}
+
+
+def _raises_launch_error(err_list):
+    """Mirror when _launch_instances turns a CreateFleet response with no instances into an exception."""
+    return bool(err_list)
+
 
 def _expected_describe_attempts(timeout):
     """Compute DescribeInstances attempts for a never-converging instance, mirroring _get_instances_info."""
@@ -735,6 +750,31 @@ def test_evaluate_launch_params(
                 ],
                 [],
             ),
+            # create-fleet - throttling reported together with one generic error per override
+            (
+                test_on_demand_params,
+                [
+                    MockedBoto3Request(
+                        method="create_fleet",
+                        response={
+                            "Instances": [],
+                            "Errors": [
+                                {"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."},
+                            ]
+                            + [
+                                {
+                                    "ErrorCode": "UnfulfillableCapacity",
+                                    "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                                }
+                            ]
+                            * 35,
+                            "ResponseMetadata": {"RequestId": "37633199-bcc6-4a88-89e3-89d859d76096"},
+                        },
+                        expected_params=test_on_demand_params,
+                    ),
+                ],
+                [],
+            ),
         ],
         ids=[
             "fleet_spot",
@@ -743,6 +783,7 @@ def test_evaluate_launch_params(
             "fleet_capacity_block",
             "fleet_throttling",
             "fleet_multiple_errors",
+            "fleet_throttling_with_generic_errors",
         ],
     )
     def test_launch_instances(
@@ -765,10 +806,10 @@ def test_launch_instances(
             with pytest.raises(Exception) as e:
                 fleet_manager._launch_instances(launch_params)
                 assert isinstance(e, ClientError)
-        elif len(expected_assigned_nodes) == 0 and len(mocked_boto3_request[0].response.get("Errors")) == 1:
+        elif not expected_assigned_nodes and _raises_launch_error(mocked_boto3_request[0].response.get("Errors", [])):
             with pytest.raises(LaunchInstancesError) as e:
                 fleet_manager._launch_instances(launch_params)
-                assert isinstance(e, LaunchInstancesError)
+            assert_that(e.value.code).is_equal_to(mocked_boto3_request[0].response.get("Errors")[0].get("ErrorCode"))
         else:
             assigned_nodes = fleet_manager._launch_instances(launch_params)
             assert_that(assigned_nodes.get("Instances", [])).is_equal_to(expected_assigned_nodes)
@@ -1326,3 +1367,96 @@ def test_launch_ec2_instances(self, mocker, count, job_id):
 
         fleet_manager._evaluate_launch_params.assert_called_once_with(count)
         fleet_manager._launch_instances.assert_called_once()
+
+    def test_launch_ec2_instances_retries_on_throttling(self, mocker):
+        """Verify CreateFleet throttling is retried, also when the response carries one error per override."""
+        mocker.patch("time.sleep")
+        fleet_manager = FleetManagerFactory.get_manager(
+            "hit", "region", "boto3_config", FLEET_CONFIG, "queue2", "fleet-ondemand", True, {}, {}
+        )
+        mocker.patch.object(fleet_manager, "_evaluate_launch_params", return_value={})
+        launched_instance_info = {
+            "InstanceId": "i-12345",
+            "PrivateIpAddress": "ip-1",
+            "PrivateDnsName": "hostname",
+            "LaunchTime": datetime(2020, 1, 1, tzinfo=timezone.utc),
+            "NetworkInterfaces": [
+                {"Attachment": {"DeviceIndex": 0, "NetworkCardIndex": 0}, "PrivateIpAddress": "ip-1"},
+            ],
+        }
+        mocker.patch.object(
+            fleet_manager,
+            "_get_instances_info",
+            side_effect=lambda instance_ids: ([launched_instance_info] if instance_ids else [], []),
+        )
+        throttled_response = {
+            "Instances": [],
+            "Errors": [{"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."}]
+            + [
+                {
+                    "ErrorCode": "UnfulfillableCapacity",
+                    "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                }
+            ]
+            * 35,
+            "ResponseMetadata": {"RequestId": "1234-abcde"},
+        }
+        create_fleet = mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            side_effect=[throttled_response, {"Instances": [{"InstanceIds": ["i-12345"]}]}],
+        )
+
+        launched = fleet_manager.launch_ec2_instances(1)
+
+        assert_that(create_fleet.call_count).is_equal_to(2)
+        assert_that(launched).is_length(1)
+
+    @pytest.mark.parametrize(
+        ("err_list", "expected_error_code"),
+        [
+            # The real cause is reported even though every other override adds an entry pointing back at it.
+            ([UNSUPPORTED_ERROR] + [UNFULFILLED_OVERRIDE] * 35, "Unsupported"),
+            # The entries pointing at the real cause carry no order guarantee.
+            ([UNFULFILLED_OVERRIDE] * 35 + [UNSUPPORTED_ERROR], "Unsupported"),
+            # UnfulfillableCapacity messages other than that one do describe a cause and are kept.
+            ([MIN_TARGET_CAPACITY_ERROR] + [UNFULFILLED_OVERRIDE] * 35, "UnfulfillableCapacity"),
+            # A single override is unaffected, whatever the entry says.
+            ([UNFULFILLED_OVERRIDE], "UnfulfillableCapacity"),
+            ([MIN_TARGET_CAPACITY_ERROR], "UnfulfillableCapacity"),
+            # Nothing to prefer: report the first entry rather than let a hardcoded code be recorded.
+            ([UNFULFILLED_OVERRIDE] * 36, "UnfulfillableCapacity"),
+            ([UNSUPPORTED_ERROR, {"ErrorCode": "VcpuLimitExceeded", "ErrorMessage": "vCPU limit"}], "Unsupported"),
+            # An empty error list is the only case left to the caller, which records insufficient capacity.
+            ([], None),
+        ],
+        ids=[
+            "real_cause_first",
+            "real_cause_last",
+            "min_target_capacity_kept",
+            "single_override_unfulfilled",
+            "single_override_min_target_capacity",
+            "only_unfulfilled_overrides",
+            "two_real_causes",
+            "no_errors_reported",
+        ],
+    )
+    def test_launch_instances_reports_the_cause_among_unfulfilled_overrides(
+        self, mocker, err_list, expected_error_code
+    ):
+        """An entry is added per override that was not fulfilled; only a real cause is worth reporting."""
+        fleet_manager = FleetManagerFactory.get_manager(
+            "hit", "region", "boto3_config", FLEET_CONFIG, "queue2", "fleet-ondemand", False, {}, {}
+        )
+        mocker.patch.object(fleet_manager, "_evaluate_launch_params", return_value={})
+        mocker.patch.object(fleet_manager, "_get_instances_info", return_value=([], []))
+        mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            return_value={"Instances": [], "Errors": err_list, "ResponseMetadata": {"RequestId": "1234-abcde"}},
+        )
+
+        if expected_error_code:
+            with pytest.raises(LaunchInstancesError) as e:
+                fleet_manager._launch_instances({})
+            assert_that(e.value.code).is_equal_to(expected_error_code)
+        else:
+            assert_that(fleet_manager._launch_instances({})).is_equal_to({"Instances": []})

tests/slurm_plugin/test_instance_manager.py

@@ -4033,6 +4033,38 @@ def test_launch_instances(
         assert_that(instances_launched).is_equal_to(expected_instances_launched)
         assert_that(instance_manager.failed_nodes).is_equal_to(expected_failed_nodes)
 
+    def test_launch_instances_reports_throttling_not_insufficient_capacity(self, mocker, instance_manager):
+        """A throttled CreateFleet must not be recorded as insufficient capacity, which would fail over.
+
+        CreateFleet reports one error per launch template override, so the throttling entry comes with a
+        generic entry for every other override.
+        """
+        mocker.patch("time.sleep")
+        mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            return_value={
+                "Instances": [],
+                "Errors": [{"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."}]
+                + [
+                    {
+                        "ErrorCode": "UnfulfillableCapacity",
+                        "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                    }
+                ]
+                * 29,
+                "ResponseMetadata": {"RequestId": "1234-abcde"},
+            },
+        )
+
+        instance_manager._launch_instances(
+            job=None,
+            nodes_to_launch={"queue2": {"fleet-ondemand": ["queue2-dy-fleet-ondemand-1"]}},
+            launch_batch_size=1,
+            scaling_strategy=ScalingStrategy.BEST_EFFORT,
+        )
+
+        assert_that(instance_manager.failed_nodes).is_equal_to({"RequestLimitExceeded": {"queue2-dy-fleet-ondemand-1"}})
+
     @pytest.mark.parametrize(
         "job_list, launch_batch_size, assign_node_batch_size, update_node_address, "
         "expected_single_nodes_no_oversubscribe, scaling_strategy",