EFS snapshot wait (#6167)

pkg/blockstorage/awsefs/awsefs.go

@@ -129,8 +129,8 @@ func (e *efs) SnapshotCreate(ctx context.Context, volume blockstorage.Volume, ta
 	return nil, errors.New("Not implemented")
 }
 
-func (e *efs) SnapshotCreateWaitForCompletion(context.Context, *blockstorage.Snapshot) error {
-	return errors.New("Not implemented")
+func (e *efs) SnapshotCreateWaitForCompletion(ctx context.Context, snapshot *blockstorage.Snapshot) error {
+	return e.waitUntilRecoveryPointCompleted(ctx, snapshot.ID)
 }
 
 func (e *efs) SnapshotDelete(ctx context.Context, snapshot *blockstorage.Snapshot) error {

pkg/blockstorage/awsefs/wait.go

@@ -3,6 +3,7 @@ package awsefs
 import (
 	"context"
 
+	"github.com/aws/aws-sdk-go/service/backup"
 	awsefs "github.com/aws/aws-sdk-go/service/efs"
 
 	"github.com/kanisterio/kanister/pkg/poll"
@@ -31,3 +32,26 @@ func (e *efs) waitUntilFileSystemAvailable(ctx context.Context, id string) error
 		return *state == awsefs.LifeCycleStateAvailable, nil
 	})
 }
+
+func (e *efs) waitUntilRecoveryPointCompleted(ctx context.Context, id string) error {
+	return poll.WaitWithRetries(ctx, maxNumErrorRetries, poll.IsAlwaysRetryable, func(ctx context.Context) (bool, error) {
+		req := &backup.DescribeRecoveryPointInput{}
+		req.SetBackupVaultName(k10BackupVaultName)
+		req.SetRecoveryPointArn(id)
+
+		desc, err := e.DescribeRecoveryPointWithContext(ctx, req)
+		if isRecoveryPointNotFound(err) {
+			// Recovery point doesn't appear when the backup jobs finishes.
+			// Since this case is special, it will be counted as non-error.
+			return false, nil
+		}
+		if err != nil {
+			return false, err
+		}
+		status := desc.Status
+		if status == nil {
+			return false, nil
+		}
+		return *status == backup.RecoveryPointStatusCompleted, nil
+	})
+}