Merge pull request #127 from alicelyu-px/volumeStateCheck

Add modification state check functionality

Author: alicelyy

aws/aws.go

@@ -34,6 +34,7 @@ const (
 	awsDevicePrefixWithH = "/dev/hd"
 	awsDevicePrefixNvme  = "/dev/nvme"
 	contextTimeout       = 30 * time.Second
+	awsErrorModificationNotFound = "InvalidVolumeModification.NotFound"
 )
 
 type awsOps struct {
@@ -942,6 +943,42 @@ func (s *awsOps) detachInternal(volumeID, instanceName string, options map[strin
 	return err
 }
 
+func isErrorModificationNotFound(err error) bool {
+	return strings.HasPrefix(err.Error(), awsErrorModificationNotFound)
+}
+
+func (s *awsOps) AreVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	modificationStateRequest := &ec2.DescribeVolumesModificationsInput{
+		VolumeIds: volumeIDs,
+	}
+	describeOutput, err := s.ec2.DescribeVolumesModifications(modificationStateRequest)
+	if err != nil {
+		// modification state not found, this indicates no change has occurred before.
+		if isErrorModificationNotFound(err) {
+			return true, nil
+		}
+		return false, fmt.Errorf("unable to get modification states for aws volumes: %v", err)
+	}
+	states := describeOutput.VolumesModifications
+
+	var state string
+	for i := 0; i < len(states); i++ {
+		if states[i] == nil || states[i].ModificationState == nil {
+			logrus.Debugf("volume modification state is nil for volume id: %s", *volumeIDs[i])
+			continue
+		}
+
+		state = *states[i].ModificationState
+		logrus.Infof("retrived volume modification state: %s for volume id: %s", state, *volumeIDs[i])
+		if state == ec2.VolumeModificationStateModifying ||
+			state == ec2.VolumeModificationStateOptimizing {
+			return false, fmt.Errorf("aws has not fully completed the last modification: " +
+				"volume %s is in %s state. please retry later", *volumeIDs[i], state)
+		}
+	}
+	return true, nil
+}
+
 func (s *awsOps) Expand(
 	volumeID string,
 	newSizeInGiB uint64,

azure/azure.go

@@ -580,6 +580,12 @@ func (a *azureOps) DeleteFrom(diskName, _ string) error {
 	return a.Delete(diskName, nil)
 }
 
+func (a *azureOps) AreVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	return true, &cloudops.ErrNotSupported{
+		Operation: "azureOps.IsVolumesReadyToExpand",
+	}
+}
+
 func (a *azureOps) Expand(
 	diskName string,
 	newSizeInGiB uint64,

backoff/exponential.go

@@ -436,6 +436,10 @@ func (e *exponentialBackoff) DevicePath(volumeID string) (string, error) {
 	return devicePath, origErr
 }
 
+func (e *exponentialBackoff) AreVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	return e.cloudOps.AreVolumesReadyToExpand(volumeIDs)
+}
+
 func (e *exponentialBackoff) Expand(volumeID string, targetSize uint64, options map[string]string) (uint64, error) {
 	var (
 		actualSize uint64

cloudops.go

@@ -129,6 +129,9 @@ type Storage interface {
 	// Attach volumeID, accepts attachoOptions as opaque data
 	// Return attach path.
 	Attach(volumeID string, options map[string]string) (string, error)
+	// IsVolumeReadyToExpand pre-checks if a pool of volumes are in a state that can
+	// be modified. Should be called before sending an expand request to the cloud provider.
+	AreVolumesReadyToExpand(volumeIDs []*string) (bool, error)
 	// Expand expands the provided device from the existing size to the new size
 	// It returns the new size of the device. It is a blocking API where it will
 	// only return once the requested size is validated with the cloud provider or

gce/gce.go

@@ -664,6 +664,12 @@ func (s *gceOps) GetDeviceID(disk interface{}) (string, error) {
 	}
 }
 
+func (s *gceOps) AreVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	return true, &cloudops.ErrNotSupported{
+		Operation: "gceOps:IsVolumesReadyToExpand",
+	}
+}
+
 func (s *gceOps) Expand(
 	volumeID string,
 	newSizeInGiB uint64,

mock/cloudops.mock.go

@@ -293,7 +293,7 @@ func (o *oracleOps) GetInstance(displayName string) (interface{}, error) {
 	if len(listInstanceResp.Items) == 0 {
 		return nil, fmt.Errorf("no oracle instance found with display name: %s", displayName)
 	}
-	// Currently, torpedo uses this API to fetch details of the instance created 
+	// Currently, torpedo uses this API to fetch details of the instance created
 	// by OKE. OKE ensures that all the worker nodes created, have unique display
 	// names. In future, if we require to use this API to get details of vanilla
 	// compute instances, then we can modify below array indexing.
@@ -829,6 +829,12 @@ func nodePoolContainsNode(s []containerengine.Node, e string) bool {
 	return false
 }
 
+func (o *oracleOps) IsVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	return true, &cloudops.ErrNotSupported{
+		Operation: "oracleOps:IsVolumesReadyToExpand",
+	}
+}
+
 func (o *oracleOps) Expand(volumeID string, newSizeInGiB uint64, options map[string]string) (uint64, error) {
 	logrus.Debug("Expand volume to size ", newSizeInGiB, " GiB")
 

oracle/oracle.go

@@ -99,6 +99,13 @@ func (u *unsupportedStorage) Attach(volumeID string, options map[string]string)
 		Operation: "Attach",
 	}
 }
+
+func (u *unsupportedStorage) IsVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	return true, &cloudops.ErrNotSupported{
+		Operation: "unsupportedStorage:IsVolumesReadyToExpand",
+	}
+}
+
 func (u *unsupportedStorage) Expand(volumeID string, newSizeInGiB uint64, options map[string]string) (uint64, error) {
 	return 0, &cloudops.ErrNotSupported{
 		Operation: "Expand",

unsupported/unsupported.go

@@ -476,7 +476,7 @@ func (ops *vsphereOps) DeviceMappings() (map[string]string, error) {
 			if ok {
 				diskUUID, err := vmObj.Datacenter.GetVirtualDiskPage83Data(ctx, backing.FileName)
 				if err != nil {
-					vmName,_ := vmObj.ObjectName(ctx)
+					vmName, _ := vmObj.ObjectName(ctx)
 					return nil, fmt.Errorf("failed to get device path for disk: %s on vm: %s err: %s", backing.FileName, vmName, err)
 				}
 
@@ -533,6 +533,12 @@ func (ops *vsphereOps) Enumerate(volumeIds []*string,
 	}
 }
 
+func (ops *vsphereOps) IsVolumesReadyToExpand(volumeIDs []*string) (bool, error) {
+	return true, &cloudops.ErrNotSupported{
+		Operation: "vsphereOps:IsVolumesReadyToExpand",
+	}
+}
+
 func (ops *vsphereOps) Expand(
 	vmdkPath string,
 	newSizeInGiB uint64,