def __init_

def init()

in perfkitbenchmarker/providers/gcp/gce_virtual_machine.py [0:0]
87 lines of code
28 McCabe index (conditional complexity)

  def __init__(self, vm_spec):
    """Initialize a GCE virtual machine.

    Args:
      vm_spec: virtual_machine.BaseVmSpec object of the vm.

    Raises:
      errors.Config.MissingOption: If the spec does not include a "machine_type"
          or both "cpus" and "memory".
      errors.Config.InvalidValue: If the spec contains both "machine_type" and
          at least one of "cpus" or "memory".
    """
    super().__init__(vm_spec)
    self.create_cmd: util.GcloudCommand = None
    self.boot_metadata = {}
    self.boot_metadata_from_file = {}
    if self.boot_startup_script:
      self.boot_metadata_from_file['startup-script'] = self.boot_startup_script
    self.ssd_interface = vm_spec.ssd_interface
    if (
        self.machine_type
        and self.machine_type in gce_disk.FIXED_SSD_MACHINE_TYPES
    ):
      self.ssd_interface = 'NVME'
    self.cpus = vm_spec.cpus
    self.image = self.image or self.DEFAULT_IMAGE
    self.memory_mib = vm_spec.memory
    self.preemptible = vm_spec.preemptible
    self.spot_early_termination = False
    self.preemptible_status_code = None
    self.project = vm_spec.project or util.GetDefaultProject()
    self.image_project = vm_spec.image_project or self.GetDefaultImageProject()
    self.mtu: int | None = FLAGS.mtu
    self.subnet_names = vm_spec.subnet_names
    self.network = self._GetNetwork()
    self.firewall = gce_network.GceFirewall.GetFirewall()
    self.boot_disk = gce_disk.GceBootDisk(self, vm_spec.boot_disk_spec)
    self.disks = [self.boot_disk]
    self.id = None
    self.node_type = vm_spec.node_type
    self.host = None
    self.use_dedicated_host = vm_spec.use_dedicated_host
    self.num_vms_per_host = vm_spec.num_vms_per_host
    self.min_cpu_platform = vm_spec.min_cpu_platform
    self.threads_per_core = vm_spec.threads_per_core
    self.visible_core_count = vm_spec.visible_core_count
    self.gce_remote_access_firewall_rule = FLAGS.gce_remote_access_firewall_rule
    self.gce_accelerator_type_override = FLAGS.gce_accelerator_type_override
    self.gce_tags = vm_spec.gce_tags
    self.gce_network_tier = FLAGS.gce_network_tier
    self.gce_nic_types = FLAGS.gce_nic_types
    self.max_local_disks = vm_spec.num_local_ssds
    if (
        self.machine_type
        and self.machine_type in gce_disk.FIXED_SSD_MACHINE_TYPES
    ):
      self.max_local_disks = gce_disk.FIXED_SSD_MACHINE_TYPES[self.machine_type]
    # For certain machine families, we need to explicitly set the GPU type
    # and counts. See the _FIXED_GPU_MACHINE_TYPES dictionary for more details.
    if self.machine_type and self.machine_type in _FIXED_GPU_MACHINE_TYPES:
      self.gpu_type = _FIXED_GPU_MACHINE_TYPES[self.machine_type][0]
      self.gpu_count = _FIXED_GPU_MACHINE_TYPES[self.machine_type][1]
    for idx, gce_nic_type in enumerate(self.gce_nic_types):
      if gce_nic_type == 'GVNIC' and not self.SupportGVNIC():
        logging.warning('Changing gce_nic_type to VIRTIO_NET')
        self.gce_nic_types[idx] = 'VIRTIO_NET'
    self.gce_egress_bandwidth_tier = gcp_flags.EGRESS_BANDWIDTH_TIER.value
    self.gce_shielded_secure_boot = FLAGS.gce_shielded_secure_boot
    # Default to GCE default (Live Migration)
    self.on_host_maintenance = None
    # https://cloud.google.com/compute/docs/instances/live-migration#gpusmaintenance
    # https://cloud.google.com/compute/docs/instances/define-instance-placement#restrictions
    # TODO(pclay): Update if this assertion ever changes
    if (
        FLAGS['gce_migrate_on_maintenance'].present
        and FLAGS.gce_migrate_on_maintenance
        and (self.gpu_count or self.network.placement_group)
    ):
      raise errors.Config.InvalidValue(
          'Cannot set flag gce_migrate_on_maintenance on instances with GPUs '
          'or network placement groups, as it is not supported by GCP.'
      )
    if (
        not FLAGS.gce_migrate_on_maintenance
        or self.gpu_count
        or self.network.placement_group
        or self.preemptible
    ):
      self.on_host_maintenance = 'TERMINATE'
    else:
      self.on_host_maintenance = 'MIGRATE'

    self.automatic_restart = FLAGS.gce_automatic_restart
    if self.preemptible:
      self.preempt_marker = f'gs://{FLAGS.gcp_preemptible_status_bucket}/{FLAGS.run_uri}/{self.name}'
    arm_arch = GetArmArchitecture(self.machine_type)
    if arm_arch:
      # Assign host_arch to avoid running detect_host on ARM
      self.host_arch = arm_arch
      self.is_aarch64 = True
    self.image_family = vm_spec.image_family or self.GetDefaultImageFamily(
        self.is_aarch64
    )
    self.create_disk_strategy = gce_disk_strategies.GetCreateDiskStrategy(
        self, None, 0
    )
def __init__()

def init()
