in perfkitbenchmarker/benchmark_spec.py [0:0]
def Provision(self):
"""Prepares the VMs and networks necessary for the benchmark to run."""
should_restore = (
hasattr(self, 'restore_spec') and self.restore_spec is not None
)
# Create capacity reservations if the cloud supports it. Note that the
# capacity reservation class may update the VMs themselves. This is true
# on AWS, because the VM needs to be aware of the capacity reservation id
# before its Create() method is called. Furthermore, if the user does not
# specify an AWS zone, but a region instead, the AwsCapacityReservation
# class will make a reservation in a zone that has sufficient capacity.
# In this case the VM's zone attribute, and the VMs network instance
# need to be updated as well.
if self.capacity_reservations:
background_tasks.RunThreaded(
lambda res: res.Create(), self.capacity_reservations
)
# Sort networks into a guaranteed order of creation based on dict key.
# There is a finite limit on the number of threads that are created to
# provision networks. Until support is added to provision resources in an
# order based on dependencies, this key ordering can be used to avoid
# deadlock by placing dependent networks later and their dependencies
# earlier.
networks = [self.networks[key] for key in sorted(self.networks.keys())]
background_tasks.RunThreaded(lambda net: net.Create(), networks)
# VPC peering is currently only supported for connecting 2 VPC networks
if self.vpc_peering:
if len(networks) > 2:
raise errors.Error(
'Networks of size %d are not currently supported.' % (len(networks))
)
# Ignore Peering for one network
elif len(networks) == 2:
networks[0].Peer(networks[1])
if self.container_registry:
self.container_registry.Create()
for container_spec in self.container_specs.values():
if container_spec.static_image:
continue
container_spec.image = self.container_registry.GetOrBuild(
container_spec.image
)
if self.container_cluster:
self.container_cluster.Create()
# do after network setup but before VM created
if self.nfs_service and self.nfs_service.CLOUD != nfs_service.UNMANAGED:
self.nfs_service.Create()
if self.smb_service:
self.smb_service.Create()
for placement_group_object in self.placement_groups.values():
placement_group_object.Create()
if self.vms:
# We separate out creating, booting, and preparing the VMs into two phases
# so that we don't slow down the creation of all the VMs by running
# commands on the VMs that booted.
background_tasks.RunThreaded(
lambda vm: vm.CreateAndBoot(),
self.vms,
post_task_delay=FLAGS.create_and_boot_post_task_delay,
)
if self.nfs_service and self.nfs_service.CLOUD == nfs_service.UNMANAGED:
self.nfs_service.Create()
if not FLAGS.skip_vm_preparation:
background_tasks.RunThreaded(lambda vm: vm.PrepareAfterBoot(), self.vms)
else:
logging.info('Skipping VM preparation.')
sshable_vms = [
vm for vm in self.vms if vm.OS_TYPE not in os_types.WINDOWS_OS_TYPES
]
sshable_vm_groups = {}
for group_name, group_vms in self.vm_groups.items():
sshable_vm_groups[group_name] = [
vm
for vm in group_vms
if vm.OS_TYPE not in os_types.WINDOWS_OS_TYPES
]
vm_util.GenerateSSHConfig(sshable_vms, sshable_vm_groups)
if self.cluster:
if self.cluster.unmanaged:
self.cluster.ImportVmGroups(
self.vm_groups['headnode'][0], self.vm_groups['workers']
)
self.cluster.Create()
if self.dpb_service:
self.dpb_service.Create()
if hasattr(self, 'relational_db') and self.relational_db:
self.relational_db.SetVms(self.vm_groups)
self.relational_db.Create(restore=should_restore)
if self.non_relational_db:
self.non_relational_db.Create(restore=should_restore)
if hasattr(self, 'key') and self.key:
self.key.Create()
if self.tpus:
background_tasks.RunThreaded(lambda tpu: tpu.Create(), self.tpus)
if self.ai_model:
self.ai_model.Create()
if self.pinecone:
self.pinecone.Create()
if self.edw_service:
if (
not self.edw_service.user_managed
and self.edw_service.SERVICE_TYPE == 'redshift'
):
# The benchmark creates the Redshift cluster's subnet group in the
# already provisioned virtual private cloud (vpc).
for network in networks:
if network.__class__.__name__ == 'AwsNetwork':
self.edw_service.cluster_subnet_group.subnet_id = network.subnet.id
self.edw_service.Create()
if self.edw_compute_resource:
self.edw_compute_resource.Create()
if hasattr(self, 'memory_store') and self.memory_store:
self.memory_store.Create()
if self.example_resource:
self.example_resource.Create()
if self.base_job:
self.base_job.Create()
if self.vpn_service:
self.vpn_service.Create()
if hasattr(self, 'messaging_service') and self.messaging_service:
self.messaging_service.Create()
if self.data_discovery_service:
self.data_discovery_service.Create()