Fix silent volume restoration failures for encrypted snapshots

Resolves velero-io/velero#3145 and velero-io/velero#9128

Previously, when restoring EBS snapshots with KMS encryption, the plugin
would report success even when volume creation failed due to missing KMS
permissions (kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant). This created
a silent failure scenario where Velero logs showed successful restoration
but the volume was never actually created.

Changes:
- Add volume creation verification with polling in CreateVolumeFromSnapshot
- Wait for volume to reach 'available' state before returning success
- Enhanced error handling for KMS permission failures with actionable messages
- Add configurable timeout (volumeCreationTimeout) and poll interval (volumeCreationPollInterval)
- Comprehensive test coverage for new error handling and configuration

The fix transforms silent failures into clear error messages, helping users
quickly identify and resolve KMS permission issues during volume restoration.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: kaovilai

velero-plugin-for-aws/volume_snapshotter.go

@@ -25,6 +25,7 @@ import (
 	"os"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -37,8 +38,14 @@ import (
 )
 
 const (
-	regionKey    = "region"
-	ebsCSIDriver = "ebs.csi.aws.com"
+	regionKey                      = "region"
+	ebsCSIDriver                   = "ebs.csi.aws.com"
+	volumeCreationTimeoutKey       = "volumeCreationTimeout"
+	volumeCreationPollIntervalKey  = "volumeCreationPollInterval"
+	
+	// Volume creation verification settings
+	defaultVolumeCreationTimeout = 10 * time.Minute
+	volumeStatusPollInterval     = 15 * time.Second
 )
 
 // iopsVolumeTypes is a set of AWS EBS volume types for which IOPS should
@@ -47,16 +54,18 @@ const (
 var iopsVolumeTypes = sets.NewString("io1", "io2")
 
 type VolumeSnapshotter struct {
-	log logrus.FieldLogger
-	ec2 *ec2.Client
+	log                    logrus.FieldLogger
+	ec2                    *ec2.Client
+	volumeCreationTimeout  time.Duration
+	volumePollInterval     time.Duration
 }
 
 func newVolumeSnapshotter(logger logrus.FieldLogger) *VolumeSnapshotter {
 	return &VolumeSnapshotter{log: logger}
 }
 
 func (b *VolumeSnapshotter) Init(config map[string]string) error {
-	if err := veleroplugin.ValidateVolumeSnapshotterConfigKeys(config, regionKey, credentialProfileKey, credentialsFileKey, enableSharedConfigKey); err != nil {
+	if err := veleroplugin.ValidateVolumeSnapshotterConfigKeys(config, regionKey, credentialProfileKey, credentialsFileKey, enableSharedConfigKey, volumeCreationTimeoutKey, volumeCreationPollIntervalKey); err != nil {
 		return err
 	}
 
@@ -67,6 +76,27 @@ func (b *VolumeSnapshotter) Init(config map[string]string) error {
 	if region == "" {
 		return errors.Errorf("missing %s in aws configuration", regionKey)
 	}
+
+	// Parse volume creation timeout
+	b.volumeCreationTimeout = defaultVolumeCreationTimeout
+	if timeoutStr := config[volumeCreationTimeoutKey]; timeoutStr != "" {
+		if timeout, err := time.ParseDuration(timeoutStr); err != nil {
+			return errors.Wrapf(err, "invalid %s duration format", volumeCreationTimeoutKey)
+		} else {
+			b.volumeCreationTimeout = timeout
+		}
+	}
+
+	// Parse volume poll interval  
+	b.volumePollInterval = volumeStatusPollInterval
+	if intervalStr := config[volumeCreationPollIntervalKey]; intervalStr != "" {
+		if interval, err := time.ParseDuration(intervalStr); err != nil {
+			return errors.Wrapf(err, "invalid %s duration format", volumeCreationPollIntervalKey)
+		} else {
+			b.volumePollInterval = interval
+		}
+	}
+
 	cfg, err := newConfigBuilder(b.log).
 		WithRegion(region).
 		WithProfile(credentialProfile).
@@ -119,7 +149,15 @@ func (b *VolumeSnapshotter) CreateVolumeFromSnapshot(snapshotID, volumeType, vol
 		return "", errors.WithStack(err)
 	}
 
-	return *output.VolumeId, nil
+	volumeID = *output.VolumeId
+	
+	// Verify that the volume is actually created and available
+	// This is critical for detecting KMS permission failures and other async errors
+	if err := b.waitForVolumeAvailable(volumeID); err != nil {
+		return "", errors.Wrapf(err, "volume creation failed for snapshot %s", snapshotID)
+	}
+
+	return volumeID, nil
 }
 
 func (b *VolumeSnapshotter) GetVolumeInfo(volumeID, volumeAZ string) (string, *int64, error) {
@@ -159,6 +197,83 @@ func (b *VolumeSnapshotter) describeVolume(volumeID string) (types.Volume, error
 	return output.Volumes[0], nil
 }
 
+// waitForVolumeAvailable polls the volume status until it becomes available or times out.
+// This is essential for detecting KMS permission failures and other async volume creation errors.
+func (b *VolumeSnapshotter) waitForVolumeAvailable(volumeID string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), b.volumeCreationTimeout)
+	defer cancel()
+
+	b.log.WithFields(logrus.Fields{
+		"volumeID": volumeID,
+		"timeout":  b.volumeCreationTimeout,
+		"interval": b.volumePollInterval,
+	}).Info("Waiting for volume to become available")
+
+	for {
+		select {
+		case <-ctx.Done():
+			return errors.Errorf("timeout waiting for volume %s to become available after %v. Check AWS CloudTrail for detailed error information", volumeID, b.volumeCreationTimeout)
+		default:
+		}
+
+		volume, err := b.describeVolume(volumeID)
+		if err != nil {
+			// Check if volume doesn't exist yet (still being created)
+			var apiErr smithy.APIError
+			if errors.As(err, &apiErr) {
+				if apiErr.ErrorCode() == "InvalidVolume.NotFound" {
+					b.log.WithField("volumeID", volumeID).Debug("Volume not found yet, continuing to wait")
+					time.Sleep(b.volumePollInterval)
+					continue
+				}
+			}
+			
+			// For other errors, return immediately with enhanced context
+			return b.enhanceVolumeCreationError(err, volumeID)
+		}
+
+		state := volume.State
+		b.log.WithFields(logrus.Fields{
+			"volumeID": volumeID,
+			"state":    state,
+		}).Debug("Volume status check")
+
+		switch state {
+		case types.VolumeStateAvailable:
+			b.log.WithField("volumeID", volumeID).Info("Volume successfully created and available")
+			return nil
+		case types.VolumeStateError:
+			return errors.Errorf("volume %s creation failed with state 'error'. This often indicates KMS permission issues for encrypted snapshots. Required KMS permissions: kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant", volumeID)
+		case types.VolumeStateCreating:
+			// Volume is still being created, continue waiting
+			b.log.WithField("volumeID", volumeID).Debug("Volume is still being created")
+		default:
+			b.log.WithFields(logrus.Fields{
+				"volumeID": volumeID,
+				"state":    state,
+			}).Debug("Volume in intermediate state, continuing to wait")
+		}
+
+		time.Sleep(b.volumePollInterval)
+	}
+}
+
+// enhanceVolumeCreationError provides more detailed error messages for common volume creation failures
+func (b *VolumeSnapshotter) enhanceVolumeCreationError(err error, volumeID string) error {
+	var apiErr smithy.APIError
+	if errors.As(err, &apiErr) {
+		switch apiErr.ErrorCode() {
+		case "UnauthorizedOperation":
+			return errors.Errorf("insufficient permissions to access volume %s or related KMS key. Required KMS permissions: kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant. Original error: %v", volumeID, err)
+		case "InvalidKey.Malformed", "KMSKeyNotAccessibleFault":
+			return errors.Errorf("KMS key access failed for volume %s. Ensure the KMS key exists and grants necessary permissions: kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant. Original error: %v", volumeID, err)
+		default:
+			return errors.Wrapf(err, "failed to verify volume %s creation status", volumeID)
+		}
+	}
+	return errors.Wrapf(err, "failed to verify volume %s creation status", volumeID)
+}
+
 func (b *VolumeSnapshotter) CreateSnapshot(volumeID, volumeAZ string, tags map[string]string) (string, error) {
 	// describe the volume so we can copy its tags to the snapshot
 	volumeInfo, err := b.describeVolume(volumeID)

velero-plugin-for-aws/volume_snapshotter_test.go

@@ -18,10 +18,13 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"github.com/aws/aws-sdk-go-v2/service/ec2/types"
+	"github.com/aws/smithy-go"
 	"os"
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
@@ -109,7 +112,7 @@ func TestGetVolumeIDForCSI(t *testing.T) {
 				Object: map[string]interface{}{},
 			}
 			csi := map[string]interface{}{}
-			json.Unmarshal([]byte(tt.csiJSON), &csi)
+			_ = json.Unmarshal([]byte(tt.csiJSON), &csi)
 			res.Object["spec"] = map[string]interface{}{
 				"csi": csi,
 			}
@@ -223,7 +226,7 @@ func TestSetVolumeIDForCSI(t *testing.T) {
 				Object: map[string]interface{}{},
 			}
 			csi := map[string]interface{}{}
-			json.Unmarshal([]byte(tt.csiJSON), &csi)
+			_ = json.Unmarshal([]byte(tt.csiJSON), &csi)
 			res.Object["spec"] = map[string]interface{}{
 				"csi": csi,
 			}
@@ -319,7 +322,7 @@ func TestGetTagsForCluster(t *testing.T) {
 			assert.Equal(t, test.expected, res)
 
 			if test.isNameSet {
-				os.Unsetenv("AWS_CLUSTER_NAME")
+				_ = os.Unsetenv("AWS_CLUSTER_NAME")
 			}
 		})
 	}
@@ -405,3 +408,132 @@ func TestGetTags(t *testing.T) {
 		})
 	}
 }
+
+func TestVolumeSnapshotterInit_VolumeCreationTimeout(t *testing.T) {
+	tests := []struct {
+		name           string
+		config         map[string]string
+		expectedError  bool
+		expectedTimeout time.Duration
+		expectedInterval time.Duration
+	}{
+		{
+			name: "default timeout and interval",
+			config: map[string]string{
+				"region": "us-east-1",
+			},
+			expectedTimeout:  defaultVolumeCreationTimeout,
+			expectedInterval: volumeStatusPollInterval,
+		},
+		{
+			name: "custom timeout and interval",
+			config: map[string]string{
+				"region":                        "us-east-1",
+				"volumeCreationTimeout":         "5m",
+				"volumeCreationPollInterval":    "30s",
+			},
+			expectedTimeout:  5 * time.Minute,
+			expectedInterval: 30 * time.Second,
+		},
+		{
+			name: "invalid timeout format",
+			config: map[string]string{
+				"region":                 "us-east-1",
+				"volumeCreationTimeout":  "invalid",
+			},
+			expectedError: true,
+		},
+		{
+			name: "invalid interval format",
+			config: map[string]string{
+				"region":                       "us-east-1",
+				"volumeCreationPollInterval":   "invalid",
+			},
+			expectedError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			vs := newVolumeSnapshotter(logrus.New())
+			err := vs.Init(tt.config)
+
+			if tt.expectedError {
+				assert.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectedTimeout, vs.volumeCreationTimeout)
+			assert.Equal(t, tt.expectedInterval, vs.volumePollInterval)
+		})
+	}
+}
+
+// mockAPIError implements smithy.APIError for testing
+type mockAPIError struct {
+	errorCode string
+	message   string
+}
+
+func (m mockAPIError) Error() string {
+	return m.message
+}
+
+func (m mockAPIError) ErrorCode() string {
+	return m.errorCode
+}
+
+func (m mockAPIError) ErrorMessage() string {
+	return m.message
+}
+
+func (m mockAPIError) ErrorFault() smithy.ErrorFault {
+	return smithy.FaultClient
+}
+
+func TestEnhanceVolumeCreationError(t *testing.T) {
+	vs := &VolumeSnapshotter{
+		log: logrus.New(),
+	}
+	volumeID := "vol-123456"
+
+	tests := []struct {
+		name          string
+		inputError    error
+		expectedError string
+	}{
+		{
+			name:          "UnauthorizedOperation error",
+			inputError:    mockAPIError{errorCode: "UnauthorizedOperation", message: "access denied"},
+			expectedError: "insufficient permissions to access volume vol-123456 or related KMS key. Required KMS permissions: kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant. Original error: access denied",
+		},
+		{
+			name:          "InvalidKey.Malformed error",
+			inputError:    mockAPIError{errorCode: "InvalidKey.Malformed", message: "invalid key"},
+			expectedError: "KMS key access failed for volume vol-123456. Ensure the KMS key exists and grants necessary permissions: kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant. Original error: invalid key",
+		},
+		{
+			name:          "KMSKeyNotAccessibleFault error",
+			inputError:    mockAPIError{errorCode: "KMSKeyNotAccessibleFault", message: "kms key not accessible"},
+			expectedError: "KMS key access failed for volume vol-123456. Ensure the KMS key exists and grants necessary permissions: kms:Decrypt, kms:ReEncrypt*, kms:CreateGrant. Original error: kms key not accessible",
+		},
+		{
+			name:          "Other API error",
+			inputError:    mockAPIError{errorCode: "SomeOtherError", message: "some other error"},
+			expectedError: "failed to verify volume vol-123456 creation status: some other error",
+		},
+		{
+			name:          "Non-API error",
+			inputError:    errors.New("generic error"),
+			expectedError: "failed to verify volume vol-123456 creation status: generic error",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := vs.enhanceVolumeCreationError(tt.inputError, volumeID)
+			assert.Contains(t, result.Error(), tt.expectedError)
+		})
+	}
+}