ec2: Enhance fleet creation with LaunchTemplateAndOverrides in responses

moto/ec2/models/fleets.py

@@ -60,7 +60,6 @@ def __init__(
 
         self.launch_specs: list[SpotFleetLaunchSpec] = []
 
-        launch_specs_from_config: list[dict[str, Any]] = []
         for config in launch_template_configs or []:
             launch_spec = config["LaunchTemplateSpecification"]
             if "LaunchTemplateId" in launch_spec:
@@ -73,33 +72,46 @@ def __init__(
                 )
             else:
                 continue
+
+            # Resolve $Latest to actual version number and always include LaunchTemplateId
+            # to match AWS behavior (AWS returns ID even when name was used in request)
+            resolved_launch_spec = launch_spec.copy()
+            if resolved_launch_spec.get("Version") == "$Latest":
+                resolved_launch_spec["Version"] = str(
+                    launch_template.latest_version_number
+                )
+            # Always include the template ID in response (AWS does this even when name is used)
+            resolved_launch_spec["LaunchTemplateId"] = launch_template.id
+
             launch_template_data = launch_template.latest_version().data
-            new_launch_template = launch_template_data.copy()
-            if config.get("Overrides"):
-                for override in config["Overrides"]:
-                    new_launch_template.update(override)
-            launch_specs_from_config.append(new_launch_template)
-
-        for spec in launch_specs_from_config:
-            tag_spec_set = spec.get("TagSpecifications", [])
-            tags = convert_tag_spec(tag_spec_set)
-            tags["instance"] = tags.get("instance", {}) | instance_tags
-            self.launch_specs.append(
-                SpotFleetLaunchSpec(
-                    ebs_optimized=spec.get("EbsOptimized"),
-                    group_set=spec.get("GroupSet", []),
-                    iam_instance_profile=spec.get("IamInstanceProfile"),
-                    image_id=spec["ImageId"],
-                    instance_type=spec["InstanceType"],
-                    key_name=spec.get("KeyName"),
-                    monitoring=spec.get("Monitoring"),
-                    spot_price=spec.get("SpotPrice"),
-                    subnet_id=spec.get("SubnetId"),
-                    tag_specifications=tags,
-                    user_data=spec.get("UserData"),
-                    weighted_capacity=spec.get("WeightedCapacity", 1),
+            overrides_list = config.get("Overrides") or [{}]
+            for override in overrides_list:
+                # Merge launch template data with override
+                spec = launch_template_data.copy()
+                spec.update(override)
+
+                tag_spec_set = spec.get("TagSpecifications", [])
+                tags = convert_tag_spec(tag_spec_set)
+                tags["instance"] = tags.get("instance", {}) | instance_tags
+
+                self.launch_specs.append(
+                    SpotFleetLaunchSpec(
+                        ebs_optimized=spec.get("EbsOptimized"),
+                        group_set=spec.get("GroupSet", []),
+                        iam_instance_profile=spec.get("IamInstanceProfile"),
+                        image_id=spec["ImageId"],
+                        instance_type=spec["InstanceType"],
+                        key_name=spec.get("KeyName"),
+                        monitoring=spec.get("Monitoring"),
+                        spot_price=spec.get("SpotPrice"),
+                        subnet_id=spec.get("SubnetId"),
+                        tag_specifications=tags,
+                        user_data=spec.get("UserData"),
+                        weighted_capacity=spec.get("WeightedCapacity", 1),
+                        launch_template_spec=resolved_launch_spec,
+                        overrides=override,
+                    )
                 )
-            )
 
         self.spot_requests: list[SpotInstanceRequest] = []
         self.on_demand_instances: list[dict[str, Any]] = []
@@ -150,6 +162,59 @@ def valid_until_as_string(self) -> Optional[str]:
     def physical_resource_id(self) -> str:
         return self.id
 
+    @property
+    def instances(self) -> Optional[list[dict[str, Any]]]:
+        """
+        Return instances for instant fleets, None for other fleet types.
+        This is part of the CreateFleet response for instant fleets only.
+        """
+        if self.fleet_type != "instant":
+            return None
+
+        instances = []
+
+        # Process on-demand instances
+        for item in self.on_demand_instances:
+            instance_data = self._build_instance_data(
+                instance=item["instance"],
+                lifecycle="on-demand",
+                launch_spec=item.get("launch_spec"),
+            )
+            instances.append(instance_data)
+
+        # Process spot instances
+        for spot_request in self.spot_requests:
+            instance_data = self._build_instance_data(
+                instance=spot_request.instance,
+                lifecycle="spot",
+                launch_spec=spot_request.launch_spec,
+            )
+            instances.append(instance_data)
+
+        return instances
+
+    @staticmethod
+    def _build_instance_data(
+        instance: Any, lifecycle: str, launch_spec: Optional[SpotFleetLaunchSpec]
+    ) -> dict[str, Any]:
+        instance_data = {
+            "Lifecycle": lifecycle,
+            "InstanceIds": [instance.id],
+            "InstanceType": instance.instance_type,
+        }
+
+        # Add launch template and overrides if available
+        if launch_spec and launch_spec.launch_template_spec:
+            launch_template_and_overrides = {
+                "LaunchTemplateSpecification": launch_spec.launch_template_spec
+            }
+            if launch_spec.overrides:
+                launch_template_and_overrides["Overrides"] = launch_spec.overrides
+
+            instance_data["LaunchTemplateAndOverrides"] = launch_template_and_overrides
+
+        return instance_data
+
     def create_spot_requests(self, weight_to_add: float) -> list[SpotInstanceRequest]:
         weight_map, added_weight = self.get_launch_spec_counts(weight_to_add)
         for launch_spec, count in weight_map.items():
@@ -173,6 +238,7 @@ def create_spot_requests(self, weight_to_add: float) -> list[SpotInstanceRequest
                 subnet_id=launch_spec.subnet_id,
                 spot_fleet_id=self.id,
                 tags=launch_spec.tag_specifications,
+                launch_spec=launch_spec,
             )
             self.spot_requests.extend(requests)
         self.fulfilled_capacity += added_weight
@@ -204,6 +270,7 @@ def create_on_demand_requests(self, weight_to_add: float) -> None:
                 {
                     "id": reservation.id,
                     "instance": instance,
+                    "launch_spec": launch_spec,
                 }
             )
         self.fulfilled_capacity += added_weight

moto/ec2/models/spot_requests.py

@@ -67,6 +67,7 @@ def __init__(
         tags: dict[str, dict[str, str]],
         spot_fleet_id: Optional[str],
         instance_interruption_behaviour: Optional[str],
+        launch_spec: Optional["SpotFleetLaunchSpec"] = None,
     ):
         super().__init__()
         self.ec2_backend = ec2_backend
@@ -97,6 +98,7 @@ def __init__(
         )
         self.user_data = user_data  # NOT
         self.spot_fleet_id = spot_fleet_id
+        self.launch_spec = launch_spec  # Track which launch spec created this request
         tag_map = tags.get("spot-instances-request", {})
         self.add_tags(tag_map)
         self.all_tags = tags
@@ -174,6 +176,8 @@ def __init__(
         tag_specifications: dict[str, dict[str, str]],
         user_data: Any,
         weighted_capacity: float,
+        launch_template_spec: Optional[dict[str, Any]] = None,
+        overrides: Optional[dict[str, Any]] = None,
     ):
         self.ebs_optimized = ebs_optimized
         self.group_set = group_set
@@ -187,6 +191,8 @@ def __init__(
         self.tag_specifications = tag_specifications
         self.user_data = user_data
         self.weighted_capacity = float(weighted_capacity)
+        self.launch_template_spec = launch_template_spec
+        self.overrides = overrides
 
 
 class SpotFleetRequest(TaggedEC2Resource, CloudFormationModel):
@@ -422,6 +428,7 @@ def request_spot_instances(
         tags: Optional[dict[str, dict[str, str]]] = None,
         spot_fleet_id: Optional[str] = None,
         instance_interruption_behaviour: Optional[str] = None,
+        launch_spec: Optional[SpotFleetLaunchSpec] = None,
     ) -> list[SpotInstanceRequest]:
         requests = []
         tags = tags or {}
@@ -449,6 +456,7 @@ def request_spot_instances(
                 tags,
                 spot_fleet_id,
                 instance_interruption_behaviour,
+                launch_spec,
             )
             self.spot_instance_requests[spot_request_id] = request
             requests.append(request)

moto/ec2/responses/fleets.py

@@ -70,7 +70,7 @@ def create_fleet(self) -> ActionResult:
 
         self.error_on_dryrun()
 
-        request = self.ec2_backend.create_fleet(
+        fleet = self.ec2_backend.create_fleet(
             on_demand_options=on_demand_options,
             spot_options=spot_options,
             target_capacity_specification=target_capacity_specification,
@@ -83,26 +83,7 @@ def create_fleet(self) -> ActionResult:
             valid_until=valid_until,
             tag_specifications=tag_specifications,
         )
-        result = {"FleetId": request.id}
-        if request.fleet_type == "instant":
-            # On Demand and Spot Instances are stored as incompatible types on the backend,
-            # so we have to do some extra work here.
-            # TODO: This should be addressed on the backend.
-            on_demand_instances = [
-                {
-                    "Lifecycle": "on-demand",
-                    "InstanceIds": [instance["instance"].id],
-                    "InstanceType": instance["instance"].instance_type,
-                }
-                for instance in request.on_demand_instances
-            ]
-            spot_requests = [
-                {
-                    "Lifecycle": "spot",
-                    "InstanceIds": [instance.instance.id],
-                    "InstanceType": instance.instance.instance_type,
-                }
-                for instance in request.spot_requests
-            ]
-            result["Instances"] = on_demand_instances + spot_requests
+        result = {"FleetId": fleet.id}
+        if fleet.instances is not None:
+            result["Instances"] = fleet.instances
         return ActionResult(result)

tests/test_ec2/test_fleets.py

@@ -27,8 +27,6 @@ def get_launch_template(conn, instance_type="t2.micro", ami_id=EXAMPLE_AMI_ID):
         LaunchTemplateData={
             "ImageId": ami_id,
             "InstanceType": instance_type,
-            "KeyName": "test",
-            "SecurityGroups": ["sg-123456"],
             "DisableApiTermination": False,
             "TagSpecifications": [
                 {
@@ -909,3 +907,132 @@ def test_user_data():
         Attribute="userData",
     )
     assert attrs["UserData"]["Value"] == str(user_data)
+
+
+@ec2_aws_verified()
+@pytest.mark.parametrize(
+    "use_template_name", [False, True], ids=["template-id", "template-name"]
+)
+@pytest.mark.parametrize(
+    [
+        "version",
+        "overrides",
+        "on_demand_count",
+        "spot_count",
+        "expected_instance_types",
+    ],
+    [
+        pytest.param(
+            "$Latest",
+            [{"InstanceType": "t3.micro"}, {"InstanceType": "t3.nano"}],
+            1,
+            0,
+            ["t3.micro", "t3.nano"],
+            id="on-demand-multi-override",
+        ),
+        pytest.param(
+            "$Latest",
+            [{"InstanceType": "t3.micro"}],
+            0,
+            2,
+            ["t3.micro"],
+            id="spot-only",
+        ),
+        pytest.param(
+            "1",
+            [{"InstanceType": "t3.micro"}],
+            1,
+            2,
+            ["t3.micro"],
+            id="mixed-on-demand-spot",
+        ),
+    ],
+)
+def test_create_instant_fleet_with_launch_template_overrides(
+    use_template_name,
+    version,
+    overrides,
+    on_demand_count,
+    spot_count,
+    expected_instance_types,
+    ec2_client=None,
+):
+    """Test that LaunchTemplateAndOverrides is included in instant fleet responses"""
+    with launch_template_context() as ctxt:
+        fleet_id = None
+        instance_ids = []
+
+        try:
+            if use_template_name:
+                lt_spec = {"LaunchTemplateName": ctxt.lt_name, "Version": version}
+            else:
+                lt_spec = {"LaunchTemplateId": ctxt.lt_id, "Version": version}
+
+            total_capacity = on_demand_count + spot_count
+            fleet_request = {
+                "LaunchTemplateConfigs": [
+                    {
+                        "LaunchTemplateSpecification": lt_spec,
+                        "Overrides": overrides,
+                    }
+                ],
+                "TargetCapacitySpecification": {
+                    "TotalTargetCapacity": total_capacity,
+                    "OnDemandTargetCapacity": on_demand_count,
+                    "SpotTargetCapacity": spot_count,
+                    "DefaultTargetCapacityType": "on-demand"
+                    if on_demand_count > 0
+                    else "spot",
+                },
+                "Type": "instant",
+            }
+
+            if on_demand_count > 0:
+                fleet_request["OnDemandOptions"] = {
+                    "AllocationStrategy": "lowest-price"
+                }
+            if spot_count > 0:
+                fleet_request["SpotOptions"] = {"AllocationStrategy": "lowest-price"}
+
+            fleet_response = ctxt.ec2.create_fleet(**fleet_request)
+
+            # Verify response structure
+            assert "FleetId" in fleet_response
+            assert "Instances" in fleet_response
+            fleet_id = fleet_response["FleetId"]
+
+            instances = fleet_response["Instances"]
+            # AWS groups instances with same config, moto returns them separately
+            # Count total instance IDs across all items
+            total_instance_ids = []
+            for inst in instances:
+                total_instance_ids.extend(inst["InstanceIds"])
+            assert len(total_instance_ids) == total_capacity
+
+            instance_ids.extend(total_instance_ids)
+
+            # Verify lifecycles if mixed
+            if on_demand_count > 0 and spot_count > 0:
+                lifecycles = {instance["Lifecycle"] for instance in instances}
+                assert "on-demand" in lifecycles
+                assert "spot" in lifecycles
+
+            for instance in instances:
+                assert "LaunchTemplateAndOverrides" in instance
+
+                lt_and_overrides = instance["LaunchTemplateAndOverrides"]
+                assert "LaunchTemplateSpecification" in lt_and_overrides
+
+                lt_spec_response = lt_and_overrides["LaunchTemplateSpecification"]
+                assert lt_spec_response["LaunchTemplateId"] == ctxt.lt_id
+                assert lt_spec_response["Version"] == "1"
+
+                assert "Overrides" in lt_and_overrides
+                overrides_response = lt_and_overrides["Overrides"]
+                assert overrides_response["InstanceType"] in expected_instance_types
+
+        finally:
+            if fleet_id:
+                ctxt.ec2.delete_fleets(FleetIds=[fleet_id], TerminateInstances=False)
+            if instance_ids:
+                ctxt.ec2.terminate_instances(InstanceIds=instance_ids)

tests/test_ec2/test_instance_types.py

@@ -59,8 +59,8 @@ def test_describe_instance_types_gpu_instance_types():
                     "MemoryInfo": {"SizeInMiB": 8192},
                     "Name": "Radeon Pro V520",
                     "Workloads": [
-                        "ml-ai",
                         "graphics",
+                        "ml-ai",
                     ],
                 }
             ],