def PopulateNVMEDevicePath()

in perfkitbenchmarker/providers/aws/aws_disk_strategies.py [0:0]
33 lines of code
16 McCabe index (conditional complexity)

  def PopulateNVMEDevicePath(self, scratch_disk, nvme_devices):
    local_devices = []
    ebs_devices = {}
    for device in nvme_devices:
      device_path = device['DevicePath']
      model_number = device['ModelNumber']
      volume_id = device['SerialNumber'].replace('vol', 'vol-')
      if model_number == 'Amazon Elastic Block Store':
        ebs_devices[volume_id] = device_path
      elif model_number == 'Amazon EC2 NVMe Instance Storage':
        local_devices.append(device_path)
      else:
        raise errors.Benchmarks.UnsupportedConfigError(
            f'{model_number} NVME devices is not supported.'
        )

    # because local devices are assigned in a round robin manner,
    # some local devices might already have been assigned (previous spec_index).
    # remove local devices that have been already been assigned.
    for _, aws_id in self.vm.disk_identifiers_by_device.items():
      if aws_id.path in local_devices:
        local_devices.remove(aws_id.path)

    if not local_devices and not ebs_devices:
      return
    # pytype: disable=attribute-error
    disks = scratch_disk.disks if scratch_disk.is_striped else [scratch_disk]
    for d in disks:
      if d.disk_type == disk.NFS:
        continue
      elif d.disk_type == disk.LOCAL:
        if aws_disk.LocalDriveIsNvme(self.vm.machine_type):
          # assign in a round robin manner. Since NVME local disks
          # are created ignoring all pkb naming conventions and assigned
          # random names on the fly.
          disk_name = self.GetDeviceByDiskSpecId(d.disk_spec_id)
          self.vm.LogDeviceByName(disk_name, None, local_devices.pop())
      elif aws_disk.EbsDriveIsNvme(self.vm.machine_type):
        # EBS NVME volumes have disk_name assigned
        # looks up disk_name by disk_spec_id
        # populate the aws identifier information
        disk_name = self.GetDeviceByDiskSpecId(d.disk_spec_id)
        volume_id = self.GetVolumeIdByDevice(disk_name)
        if volume_id in ebs_devices:
          self.UpdateDeviceByName(disk_name, volume_id, ebs_devices[volume_id])