def __init__()

in perfkitbenchmarker/providers/azure/azure_virtual_machine.py [0:0]

• 89 lines of code
• 17 McCabe index (conditional complexity)

  def __init__(self, vm_spec):
    """Initialize an Azure virtual machine.

    Args:
      vm_spec: virtual_machine.BaseVmSpec object of the vm.
    """
    super().__init__(vm_spec)

    # PKB zone can be either a region or a region with an availability zone.
    # Format for Azure availability zone support is "region-availability_zone"
    # Example: eastus2-1 is Azure region eastus2 with availability zone 1.

    assert isinstance(self.zone, str)
    assert isinstance(self.machine_type, str)
    self.region = util.GetRegionFromZone(self.zone)
    self.availability_zone = util.GetAvailabilityZoneFromZone(self.zone)
    self.use_dedicated_host = vm_spec.use_dedicated_host
    self.num_vms_per_host = vm_spec.num_vms_per_host
    self.network = azure_network.AzureNetwork.GetNetwork(self)
    self.firewall = azure_network.AzureFirewall.GetFirewall()
    if azure_disk.HasTempDrive(self.machine_type):
      self.max_local_disks = NUM_LOCAL_VOLUMES.get(self.machine_type, 1)
    else:
      self.max_local_disks = 0
    self.lun_counter = itertools.count()
    self._deleted = False
    self.resource_group = azure_network.GetResourceGroup()

    # Configure NIC
    if self.assign_external_ip:
      public_ip_name = self.name + '-public-ip'
      self.public_ip = AzurePublicIPAddress(
          self.region, self.availability_zone, public_ip_name
      )
    else:
      public_ip_name = None
      self.public_ip = None
    self.nic: AzureNIC = AzureNIC(
        self.network,
        self.name + '-nic',
        public_ip_name,
        vm_spec.accelerated_networking,
    )

    self.storage_account = self.network.storage_account
    self.machine_type_is_confidential = any(
        re.search(machine_series, self.machine_type)
        for machine_series in CONFIDENTIAL_MILAN_TYPES
    )
    self.trusted_launch_unsupported_type = (
        any(
            re.search(machine_series, self.machine_type)
            for machine_series in TRUSTED_LAUNCH_UNSUPPORTED_TYPES
        )
        or self.OS_TYPE in TRUSTED_LAUNCH_UNSUPPORTED_OS_TYPES
    )
    arm_arch = _GetArmArch(self.machine_type)
    if arm_arch:
      self.host_arch = arm_arch
      self.is_aarch64 = True
    self.hypervisor_generation = 2
    if vm_spec.image:
      self.image = vm_spec.image
    elif self.machine_type in _MACHINE_TYPES_ONLY_SUPPORT_GEN1_IMAGES:
      self.hypervisor_generation = 1
      if hasattr(type(self), 'GEN1_IMAGE_URN'):
        self.image = type(self).GEN1_IMAGE_URN
      else:
        raise errors.Benchmarks.UnsupportedConfigError('No Azure gen1 image.')
    elif arm_arch:
      if hasattr(type(self), 'ARM_IMAGE_URN'):
        self.image = type(self).ARM_IMAGE_URN  # pytype: disable=attribute-error
      else:
        raise errors.Benchmarks.UnsupportedConfigError('No Azure ARM image.')
    elif self.machine_type_is_confidential:
      self.image = type(self).CONFIDENTIAL_IMAGE_URN  # pytype: disable=attribute-error
    else:
      if hasattr(type(self), 'GEN2_IMAGE_URN'):
        self.image = type(self).GEN2_IMAGE_URN
      else:
        raise errors.Benchmarks.UnsupportedConfigError(
            'No Azure gen2 image.  Set GEN2_IMAGE_URN or update'
            ' _MACHINE_TYPES_ONLY_SUPPORT_GEN1_IMAGES.'
        )

    self.host = None
    if self.use_dedicated_host:
      self.host_series_sku = _GetSkuType(self.machine_type)
      self.host_list = None
    self.low_priority = vm_spec.low_priority
    self.low_priority_status_code = None
    self.spot_early_termination = False
    self.ultra_ssd_enabled = False
    self.boot_disk_size = vm_spec.boot_disk_size
    self.boot_disk_type = vm_spec.boot_disk_type
    self.create_os_disk_strategy = (
        azure_disk_strategies.AzureCreateOSDiskStrategy(
            self, disk.BaseDiskSpec('azure_os_disk'), 1
        )
    )
    self.create_disk_strategy = azure_disk_strategies.GetCreateDiskStrategy(
        self, None, 0
    )