Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Autoscaling - pass BlockDeviceMapping from launch_config/launch_template (#5044)

Browse Source
This commit is contained in:
Bert Blommers 2022-04-21 14:19:36 +00:00 committed by GitHub
parent f8c2b621db
commit 8da9666a90
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 154 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
moto/autoscaling/models.py
View File

@ -163,7 +163,8 @@ class FakeLaunchConfiguration(CloudFormationModel):
spot_price=None,
ebs_optimized=instance.ebs_optimized,
associate_public_ip_address=instance.associate_public_ip,
block_device_mappings=instance.block_device_mapping,
# We expect a dictionary in the same format as when the user calls it
block_device_mappings=instance.block_device_mapping.to_source_dict(),
)
return config
@ -249,17 +250,16 @@ class FakeLaunchConfiguration(CloudFormationModel):
block_device_map = BlockDeviceMapping()
for mapping in self.block_device_mapping_dict:
block_type = BlockDeviceType()
mount_point = mapping.get("device_name")
if "ephemeral" in mapping.get("virtual_name", ""):
block_type.ephemeral_name = mapping.get("virtual_name")
mount_point = mapping.get("DeviceName")
if mapping.get("VirtualName") and "ephemeral" in mapping.get("VirtualName"):
block_type.ephemeral_name = mapping.get("VirtualName")
else:
block_type.volume_type = mapping.get("ebs._volume_type")
block_type.snapshot_id = mapping.get("ebs._snapshot_id")
block_type.delete_on_termination = mapping.get(
"ebs._delete_on_termination"
)
block_type.size = mapping.get("ebs._volume_size")
block_type.iops = mapping.get("ebs._iops")
ebs = mapping.get("Ebs", {})
block_type.volume_type = ebs.get("VolumeType")
block_type.snapshot_id = ebs.get("SnapshotId")
block_type.delete_on_termination = ebs.get("DeleteOnTermination")
block_type.size = ebs.get("VolumeSize")
block_type.iops = ebs.get("Iops")
block_device_map[mount_point] = block_type
return block_device_map
@ -620,6 +620,7 @@ class FakeAutoScalingGroup(CloudFormationModel):
instance_type=self.instance_type,
tags={"instance": propagated_tags},
placement=random.choice(self.availability_zones),
launch_config=self.launch_config,
)
for instance in reservation.instances:
instance.autoscaling_group = self

27
moto/autoscaling/responses.py
View File

@ -20,22 +20,23 @@ class AutoScalingResponse(BaseResponse):
instance_monitoring = True
else:
instance_monitoring = False
params = self._get_params()
self.autoscaling_backend.create_launch_configuration(
name=self._get_param("LaunchConfigurationName"),
image_id=self._get_param("ImageId"),
key_name=self._get_param("KeyName"),
ramdisk_id=self._get_param("RamdiskId"),
kernel_id=self._get_param("KernelId"),
name=params.get("LaunchConfigurationName"),
image_id=params.get("ImageId"),
key_name=params.get("KeyName"),
ramdisk_id=params.get("RamdiskId"),
kernel_id=params.get("KernelId"),
security_groups=self._get_multi_param("SecurityGroups.member"),
user_data=self._get_param("UserData"),
instance_type=self._get_param("InstanceType"),
user_data=params.get("UserData"),
instance_type=params.get("InstanceType"),
instance_monitoring=instance_monitoring,
instance_profile_name=self._get_param("IamInstanceProfile"),
spot_price=self._get_param("SpotPrice"),
ebs_optimized=self._get_param("EbsOptimized"),
associate_public_ip_address=self._get_param("AssociatePublicIpAddress"),
block_device_mappings=self._get_list_prefix("BlockDeviceMappings.member"),
instance_id=self._get_param("InstanceId"),
instance_profile_name=params.get("IamInstanceProfile"),
spot_price=params.get("SpotPrice"),
ebs_optimized=params.get("EbsOptimized"),
associate_public_ip_address=params.get("AssociatePublicIpAddress"),
block_device_mappings=params.get("BlockDeviceMappings"),
instance_id=params.get("InstanceId"),
)
template = self.response_template(CREATE_LAUNCH_CONFIGURATION_TEMPLATE)
return template.render()

46
moto/ec2/_models/instances.py
View File

@ -66,17 +66,8 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
launch_template_arg = kwargs.get("launch_template", {})
if launch_template_arg and not image_id:
# the image id from the template should be used
template = (
ec2_backend.describe_launch_templates(
template_ids=[launch_template_arg["LaunchTemplateId"]]
)[0]
if "LaunchTemplateId" in launch_template_arg
else ec2_backend.describe_launch_templates(
template_names=[launch_template_arg["LaunchTemplateName"]]
)[0]
)
version = launch_template_arg.get("Version", template.latest_version_number)
self.image_id = template.get_version(int(version)).image_id
template_version = ec2_backend._get_template_from_args(launch_template_arg)
self.image_id = template_version.image_id
else:
self.image_id = image_id
@ -183,6 +174,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
encrypted=False,
delete_on_termination=False,
kms_key_id=None,
volume_type=None,
):
volume = self.ec2_backend.create_volume(
size=size,
@ -190,6 +182,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel):
snapshot_id=snapshot_id,
encrypted=encrypted,
kms_key_id=kms_key_id,
volume_type=volume_type,
)
self.ec2_backend.attach_volume(
volume.id, self.id, device_path, delete_on_termination
@ -563,16 +556,30 @@ class InstanceBackend(object):
)
new_reservation.instances.append(new_instance)
new_instance.add_tags(instance_tags)
block_device_mappings = None
if "block_device_mappings" in kwargs:
for block_device in kwargs["block_device_mappings"]:
block_device_mappings = kwargs["block_device_mappings"]
elif kwargs.get("launch_template"):
template = self._get_template_from_args(kwargs["launch_template"])
block_device_mappings = template.data.get("BlockDeviceMapping")
elif kwargs.get("launch_config"):
block_device_mappings = kwargs[
"launch_config"
].block_device_mapping_dict
if block_device_mappings:
for block_device in block_device_mappings:
device_name = block_device["DeviceName"]
volume_size = block_device["Ebs"].get("VolumeSize")
volume_type = block_device["Ebs"].get("VolumeType")
snapshot_id = block_device["Ebs"].get("SnapshotId")
encrypted = block_device["Ebs"].get("Encrypted", False)
if isinstance(encrypted, str):
encrypted = encrypted.lower() == "true"
delete_on_termination = block_device["Ebs"].get(
"DeleteOnTermination", False
)
kms_key_id = block_device["Ebs"].get("KmsKeyId")
if block_device.get("NoDevice") != "":
new_instance.add_block_device(
volume_size,
@ -581,6 +588,7 @@ class InstanceBackend(object):
encrypted,
delete_on_termination,
kms_key_id,
volume_type=volume_type,
)
else:
new_instance.setup_defaults()
@ -759,3 +767,17 @@ class InstanceBackend(object):
if filters is not None:
reservations = filter_reservations(reservations, filters)
return reservations
def _get_template_from_args(self, launch_template_arg):
template = (
self.describe_launch_templates(
template_ids=[launch_template_arg["LaunchTemplateId"]]
)[0]
if "LaunchTemplateId" in launch_template_arg
else self.describe_launch_templates(
template_names=[launch_template_arg["LaunchTemplateName"]]
)[0]
)
version = launch_template_arg.get("Version", template.latest_version_number)
template_version = template.get_version(int(version))
return template_version

16
moto/packages/boto/ec2/blockdevicemapping.py
View File

@ -54,6 +54,7 @@ class BlockDeviceType(object):
self.volume_type = volume_type
self.iops = iops
self.encrypted = encrypted
self.kms_key_id = None
# for backwards compatibility
@ -81,3 +82,18 @@ class BlockDeviceMapping(dict):
self.connection = connection
self.current_name = None
self.current_value = None
def to_source_dict(self):
return [
{
"DeviceName": device_name,
"Ebs": {
"DeleteOnTermination": block.delete_on_termination,
"Encrypted": block.encrypted,
"VolumeType": block.volume_type,
"VolumeSize": block.size,
},
"VirtualName": block.ephemeral_name,
}
for device_name, block in self.items()
]

31
tests/test_autoscaling/test_autoscaling.py
View File

@ -2781,3 +2781,34 @@ def test_set_desired_capacity_without_protection(original, new):
group["DesiredCapacity"].should.equal(new)
instances = client.describe_auto_scaling_instances()["AutoScalingInstances"]
instances.should.have.length_of(new)
@mock_autoscaling
@mock_ec2
def test_create_template_with_block_device():
ec2_client = boto3.client("ec2", region_name="ap-southeast-2")
ec2_client.create_launch_template(
LaunchTemplateName="launchie",
LaunchTemplateData={
"ImageId": EXAMPLE_AMI_ID,
"BlockDeviceMappings": [
{
"DeviceName": "/dev/sda1",
"Ebs": {
"VolumeSize": 20,
"DeleteOnTermination": True,
"VolumeType": "gp3",
"Encrypted": True,
},
}
],
},
)
ec2_client.run_instances(
MaxCount=1, MinCount=1, LaunchTemplate={"LaunchTemplateName": "launchie"}
)
ec2_client = boto3.client("ec2", region_name="ap-southeast-2")
volumes = ec2_client.describe_volumes()["Volumes"]
volumes[0]["VolumeType"].should.equal("gp3")
volumes[0]["Size"].should.equal(20)

48
tests/test_autoscaling/test_launch_configurations.py
View File

@ -5,7 +5,7 @@ from botocore.exceptions import ClientError
import pytest
import sure # noqa # pylint: disable=unused-import
from moto import mock_autoscaling
from moto import mock_autoscaling, mock_ec2
from moto.core import ACCOUNT_ID
from tests import EXAMPLE_AMI_ID
@ -245,3 +245,49 @@ def test_invalid_launch_configuration_request_raises_error(request_params):
ex.value.response["Error"]["Message"].should.match(
r"^Valid requests must contain.*"
)
@mock_autoscaling
@mock_ec2
def test_launch_config_with_block_device_mappings__volumes_are_created():
as_client = boto3.client("autoscaling", "us-east-2")
ec2_client = boto3.client("ec2", "us-east-2")
random_image_id = ec2_client.describe_images()["Images"][0]["ImageId"]
as_client.create_launch_configuration(
LaunchConfigurationName=f"lc-{random_image_id}",
ImageId=random_image_id,
InstanceType="t2.nano",
BlockDeviceMappings=[
{
"DeviceName": "/dev/sdf",
"Ebs": {
"VolumeSize": 10,
"VolumeType": "standard",
"Encrypted": False,
"DeleteOnTermination": True,
},
}
],
)
asg_name = f"asg-{random_image_id}"
as_client.create_auto_scaling_group(
AutoScalingGroupName=asg_name,
LaunchConfigurationName=f"lc-{random_image_id}",
MinSize=1,
MaxSize=1,
DesiredCapacity=1,
AvailabilityZones=["us-east-2b"],
)
instances = as_client.describe_auto_scaling_instances()["AutoScalingInstances"]
instance_id = instances[0]["InstanceId"]
volumes = ec2_client.describe_volumes(
Filters=[{"Name": "attachment.instance-id", "Values": [instance_id]}]
)["Volumes"]
volumes.should.have.length_of(1)
volumes[0].should.have.key("Size").equals(10)
volumes[0].should.have.key("Encrypted").equals(False)
volumes[0].should.have.key("VolumeType").equals("standard")