func()

in pkg/providers/v1/aws.go [2780:2876]

• 74 lines of code
• 15 McCabe index (conditional complexity)

func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (KubernetesVolumeID, error) {
	var createType string
	var iops int64
	switch volumeOptions.VolumeType {
	case VolumeTypeGP2, VolumeTypeSC1, VolumeTypeST1:
		createType = volumeOptions.VolumeType

	case VolumeTypeIO1:
		// See http://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateVolume.html
		// for IOPS constraints. AWS will throw an error if IOPS per GB gets out
		// of supported bounds, no need to check it here.
		createType = volumeOptions.VolumeType
		iops = int64(volumeOptions.CapacityGB * volumeOptions.IOPSPerGB)

		// Cap at min/max total IOPS, AWS would throw an error if it gets too
		// low/high.
		if iops < MinTotalIOPS {
			iops = MinTotalIOPS
		}
		if iops > MaxTotalIOPS {
			iops = MaxTotalIOPS
		}

	case "":
		createType = DefaultVolumeType

	default:
		return KubernetesVolumeID(""), fmt.Errorf("invalid AWS VolumeType %q", volumeOptions.VolumeType)
	}

	request := &ec2.CreateVolumeInput{}
	request.AvailabilityZone = aws.String(volumeOptions.AvailabilityZone)
	request.Size = aws.Int64(int64(volumeOptions.CapacityGB))
	request.VolumeType = aws.String(createType)
	request.Encrypted = aws.Bool(volumeOptions.Encrypted)
	if len(volumeOptions.KmsKeyID) > 0 {
		request.KmsKeyId = aws.String(volumeOptions.KmsKeyID)
		request.Encrypted = aws.Bool(true)
	}
	if iops > 0 {
		request.Iops = aws.Int64(iops)
	}

	tags := volumeOptions.Tags
	tags = c.tagging.buildTags(ResourceLifecycleOwned, tags)

	var tagList []*ec2.Tag
	for k, v := range tags {
		tagList = append(tagList, &ec2.Tag{
			Key: aws.String(k), Value: aws.String(v),
		})
	}
	request.TagSpecifications = append(request.TagSpecifications, &ec2.TagSpecification{
		Tags:         tagList,
		ResourceType: aws.String(ec2.ResourceTypeVolume),
	})

	response, err := c.ec2.CreateVolume(request)
	if err != nil {
		return KubernetesVolumeID(""), err
	}

	awsID := EBSVolumeID(aws.StringValue(response.VolumeId))
	if awsID == "" {
		return KubernetesVolumeID(""), fmt.Errorf("VolumeID was not returned by CreateVolume")
	}
	volumeName := KubernetesVolumeID("aws://" + aws.StringValue(response.AvailabilityZone) + "/" + string(awsID))

	err = c.waitUntilVolumeAvailable(volumeName)
	if err != nil {
		// AWS has a bad habbit of reporting success when creating a volume with
		// encryption keys that either don't exists or have wrong permissions.
		// Such volume lives for couple of seconds and then it's silently deleted
		// by AWS. There is no other check to ensure that given KMS key is correct,
		// because Kubernetes may have limited permissions to the key.
		if isAWSErrorVolumeNotFound(err) {
			err = fmt.Errorf("failed to create encrypted volume: the volume disappeared after creation, most likely due to inaccessible KMS encryption key")
		} else {
			// When DescribeVolumes api failed, plugin will lose track on the volumes' state
			// driver should be able to clean up these kind of volumes to make sure they are not leaked on customers' account
			klog.V(5).Infof("Failed to create the volume %v due to %v. Will try to delete it.", volumeName, err)
			awsDisk, newDiskError := newAWSDisk(c, volumeName)
			if newDiskError != nil {
				klog.Errorf("Failed to delete the volume %v due to error: %v", volumeName, newDiskError)
			} else {
				if _, deleteVolumeError := awsDisk.deleteVolume(); deleteVolumeError != nil {
					klog.Errorf("Failed to delete the volume %v due to error: %v", volumeName, deleteVolumeError)
				} else {
					klog.V(5).Infof("%v is deleted because it is not in desired state after waiting", volumeName)
				}
			}
		}
		return KubernetesVolumeID(""), err
	}

	return volumeName, nil
}