in perfkitbenchmarker/providers/aws/aws_virtual_machine.py [0:0]
def GetDefaultImage(cls, machine_type, region):
"""Returns the default image given the machine type and region.
If specified, the aws_image_name_filter and aws_image_name_regex flags will
override os_type defaults.
Args:
machine_type: The machine_type of the VM, used to determine virtualization
type.
region: The region of the VM, as images are region specific.
Raises:
AwsImageNotFoundError: If a default image cannot be found.
Returns:
The ID of the latest image, or None if no default image is configured or
none can be found.
"""
prefix = machine_type.split('.')[0]
virt_type = PV if prefix in NON_HVM_PREFIXES else HVM
processor_architecture = GetProcessorArchitecture(machine_type)
alternate_architecture = GetAlternateProcessorArchitecture(machine_type)
format_dict = {
'virt_type': virt_type,
'disk_type': cls.DEFAULT_ROOT_DISK_TYPE,
'architecture': processor_architecture,
'alternate_architecture': alternate_architecture,
}
if cls.IMAGE_SSM_PATTERN:
if FLAGS.aws_image_name_filter:
raise ValueError(
'--aws_image_name_filter is not supported for AWS OS Mixins that '
'use SSM to select AMIs. You can still pass --image.'
)
stdout, _, _ = vm_util.IssueCommand(
util.AWS_PREFIX
+ [
'ssm',
'get-parameters',
'--region',
region,
'--names',
cls.IMAGE_SSM_PATTERN.format(**format_dict),
]
)
response = json.loads(stdout)
if response['InvalidParameters']:
raise AwsImageNotFoundError('Invalid SSM parameters:\n' + stdout)
parameters = response['Parameters']
assert len(parameters) == 1
return parameters[0]['Value']
# These cannot be REQUIRED_ATTRS, because nesting REQUIRED_ATTRS breaks.
if not cls.IMAGE_OWNER:
raise NotImplementedError('AWS OSMixins require IMAGE_OWNER')
if not cls.IMAGE_NAME_FILTER_PATTERN:
raise NotImplementedError(
'AWS OSMixins require IMAGE_NAME_FILTER_PATTERN'
)
if FLAGS.aws_image_name_filter:
cls.IMAGE_NAME_FILTER_PATTERN = FLAGS.aws_image_name_filter
if FLAGS.aws_image_name_regex:
cls.IMAGE_NAME_REGEX = FLAGS.aws_image_name_regex
image_name_filter = cls.IMAGE_NAME_FILTER_PATTERN.format(**format_dict)
describe_cmd = util.AWS_PREFIX + [
'--region=%s' % region,
'ec2',
'describe-images',
'--query',
'Images[*].{Name:Name,ImageId:ImageId,CreationDate:CreationDate}',
'--filters',
'Name=name,Values=' + image_name_filter,
'Name=block-device-mapping.volume-type,Values=%s'
% cls.DEFAULT_ROOT_DISK_TYPE,
'Name=virtualization-type,Values=%s' % virt_type,
'Name=architecture,Values=%s' % processor_architecture,
]
if cls.IMAGE_PRODUCT_CODE_FILTER:
describe_cmd.extend(
['Name=product-code,Values=%s' % cls.IMAGE_PRODUCT_CODE_FILTER]
)
if cls.IMAGE_DESCRIPTION_FILTER:
describe_cmd.extend(
['Name=description,Values=%s' % cls.IMAGE_DESCRIPTION_FILTER]
)
if cls.ALLOW_DEPRECATED_IMAGE:
describe_cmd.append('--include-deprecated')
else:
# This is the default, but be explicit.
describe_cmd.append('--no-include-deprecated')
describe_cmd.extend(['--owners'] + cls.IMAGE_OWNER)
stdout, _ = util.IssueRetryableCommand(describe_cmd)
if not stdout:
raise AwsImageNotFoundError(
'aws describe-images did not produce valid output.'
)
if cls.IMAGE_NAME_REGEX:
# Further filter images by the IMAGE_NAME_REGEX filter.
image_name_regex = cls.IMAGE_NAME_REGEX.format(**format_dict)
images = []
excluded_images = []
for image in json.loads(stdout):
if re.search(image_name_regex, image['Name']):
images.append(image)
else:
excluded_images.append(image)
if excluded_images:
logging.debug(
'Excluded the following images with regex "%s": %s',
image_name_regex,
sorted(image['Name'] for image in excluded_images),
)
else:
images = json.loads(stdout)
if not images:
raise AwsImageNotFoundError('No AMIs with given filters found.')
return max(images, key=lambda image: image['CreationDate'])['ImageId']