Add e2e for verify snapshto in VSL

Signed-off-by: danfengl <danfengl@vmware.com>

changelogs/unreleased/4401-danfengliu

@@ -1 +0,0 @@
-Add backup deletion e2e test 

go.mod

@@ -5,7 +5,7 @@ go 1.17
 require (
 	cloud.google.com/go/storage v1.10.0
 	github.com/Azure/azure-pipeline-go v0.2.3
-	github.com/Azure/azure-sdk-for-go v42.0.0+incompatible
+	github.com/Azure/azure-sdk-for-go v61.4.0+incompatible
 	github.com/Azure/azure-storage-blob-go v0.14.0
 	github.com/Azure/go-autorest/autorest v0.11.21
 	github.com/Azure/go-autorest/autorest/azure/auth v0.5.8
@@ -35,6 +35,7 @@ require (
 	github.com/vmware-tanzu/crash-diagnostics v0.3.7
 	golang.org/x/mod v0.4.2
 	golang.org/x/net v0.0.0-20210520170846-37e1c6afe023
+	golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f
 	google.golang.org/api v0.56.0
 	google.golang.org/grpc v1.40.0
 	k8s.io/api v0.22.2
@@ -102,7 +103,6 @@ require (
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.19.0 // indirect
 	golang.org/x/crypto v0.0.0-20210817164053-32db794688a5 // indirect
-	golang.org/x/oauth2 v0.0.0-20210819190943-2bc19b11175f // indirect
 	golang.org/x/sys v0.0.0-20211019181941-9d821ace8654 // indirect
 	golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d // indirect
 	golang.org/x/text v0.3.7 // indirect

go.sum

@@ -48,8 +48,8 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-pipeline-go v0.2.3 h1:7U9HBg1JFK3jHl5qmo4CTZKFTVgMwdFHMVtCdfBE21U=
 github.com/Azure/azure-pipeline-go v0.2.3/go.mod h1:x841ezTBIMG6O3lAcl8ATHnsOPVl2bqk7S3ta6S6u4k=
-github.com/Azure/azure-sdk-for-go v42.0.0+incompatible h1:yz6sFf5bHZ+gEOQVuK5JhPqTTAmv+OvSLSaqgzqaCwY=
-github.com/Azure/azure-sdk-for-go v42.0.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
+github.com/Azure/azure-sdk-for-go v61.4.0+incompatible h1:BF2Pm3aQWIa6q9KmxyF1JYKYXtVw67vtvu2Wd54NGuY=
+github.com/Azure/azure-sdk-for-go v61.4.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-storage-blob-go v0.14.0 h1:1BCg74AmVdYwO3dlKwtFU1V0wU2PZdREkXvAmZJRUlM=
 github.com/Azure/azure-storage-blob-go v0.14.0/go.mod h1:SMqIBi+SuiQH32bvyjngEewEeXoPfKMgWlBDaYf6fck=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=

test/e2e/backups/deletion.go

@@ -33,17 +33,15 @@ import (
 	. "github.com/vmware-tanzu/velero/test/e2e/util/velero"
 )
 
-const (
-	deletionTest = "deletion-workload"
-)
+const deletionTest = "deletion-workload"
 
 // Test backup and restore of Kibishi using restic
 
-func Backup_deletion_with_snapshots() {
+func BackupDeletionWithSnapshots() {
 	backup_deletion_test(true)
 }
 
-func Backup_deletion_with_restic() {
+func BackupDeletionWithRestic() {
 	backup_deletion_test(false)
 }
 func backup_deletion_test(useVolumeSnapshots bool) {
@@ -86,7 +84,6 @@ func backup_deletion_test(useVolumeSnapshots bool) {
 // runUpgradeTests runs upgrade test on the provider by kibishii.
 func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNamespace, backupName, backupLocation string,
 	useVolumeSnapshots bool, registryCredentialFile, bslPrefix, bslConfig string) error {
-
 	oneHourTimeout, _ := context.WithTimeout(context.Background(), time.Minute*60)
 
 	if err := CreateNamespace(oneHourTimeout, client, deletionTest); err != nil {
@@ -124,15 +121,31 @@ func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNa
 	if err != nil {
 		return err
 	}
+	if useVolumeSnapshots {
+		var snapshotCheckPoint SnapshotCheckPoint
+		snapshotCheckPoint.ExpectCount = 2
+		snapshotCheckPoint.NamespaceBackedUp = deletionTest
+		err = SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix, snapshotCheckPoint)
+		if err != nil {
+			return err
+		}
+	}
 	err = DeleteBackupResource(context.Background(), veleroCLI, backupName)
 	if err != nil {
 		return err
 	}
+
 	err = ObjectsShouldNotBeInBucket(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix, 5)
 	if err != nil {
-		fmt.Println(errors.Wrapf(err, "Failed to get object from bucket %q", backupName))
 		return err
 	}
+	if useVolumeSnapshots {
+		err = SnapshotsShouldNotExistInCloud(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix)
+		if err != nil {
+			return err
+		}
+	}
+
 	backupName = "backup-1-" + UUIDgen.String()
 	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, deletionTest, backupLocation, useVolumeSnapshots); err != nil {
 		// TODO currently, the upgrade case covers the upgrade path from 1.6 to main and the velero v1.6 doesn't support "debug" command
@@ -142,7 +155,6 @@ func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNa
 	}
 	err = DeleteObjectsInBucket(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix)
 	if err != nil {
-		fmt.Println(errors.Wrapf(err, "Failed to delete object in bucket %q", backupName))
 		return err
 	}
 	err = ObjectsShouldNotBeInBucket(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix, 1)
@@ -151,8 +163,9 @@ func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNa
 	}
 	err = DeleteBackupResource(context.Background(), veleroCLI, backupName)
 	if err != nil {
-		fmt.Println(errors.Wrapf(err, "|| UNEXPECTED || - Failed to delete backup %q", backupName))
-		return err
+		return errors.Wrapf(err, "|| UNEXPECTED || - Failed to delete backup %q", backupName)
+	} else {
+		fmt.Printf("|| EXPECTED || - Success to delete backup %s locally\n", backupName)
 	}
 	fmt.Printf("|| EXPECTED || - Backup deletion test completed successfully\n")
 	return nil

test/e2e/e2e_suite_test.go

@@ -89,7 +89,8 @@ var _ = Describe("[ResourceFiltering][IncludeNamespaces][Restore] Velero test on
 var _ = Describe("[ResourceFiltering][IncludeResources][Backup] Velero test on include resources from the cluster backup", BackupWithIncludeResources)
 var _ = Describe("[ResourceFiltering][IncludeResources][Restore] Velero test on include resources from the cluster restore", RestoreWithIncludeResources)
 var _ = Describe("[ResourceFiltering][LabelSelector] Velero test on backup include resources matching the label selector", BackupWithLabelSelector)
-var _ = Describe("[Backups][Deletion] Velero tests on cluster using the plugin provider for object storage and Restic for volume backups", Backup_deletion_with_restic)
+var _ = Describe("[Backups][Deletion][Restic] Velero tests of Restic backup deletion", BackupDeletionWithRestic)
+var _ = Describe("[Backups][Deletion][Snapshot] Velero tests of snapshot backup deletion", BackupDeletionWithSnapshots)
 var _ = Describe("[PrivilegesMgmt][SSR] Velero test on ssr object when controller namespace mix-ups", SSRTest)
 var _ = Describe("[Backups][BackupsSync] Backups in object storage are synced to a new Velero and deleted backups in object storage are synced to be deleted in Velero", BackupsSyncTest)
 

test/e2e/types.go

@@ -49,3 +49,9 @@ type VerleroConfig struct {
 	AddBSLPlugins            string
 	InstallVelero            bool
 }
+
+type SnapshotCheckPoint struct {
+	NamespaceBackedUp string
+	SnapshotIDList    []string
+	ExpectCount       int
+}

test/e2e/util/providers/aws_utils.go

@@ -23,11 +23,13 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
 	"github.com/pkg/errors"
 
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
+	. "github.com/vmware-tanzu/velero/test/e2e"
 )
 
 type AWSStorage string
@@ -92,8 +94,10 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 	}
 	var backupNameInStorage string
 	for _, item := range bucketObjects.CommonPrefixes {
+		fmt.Println("item:")
+		fmt.Println(item)
 		backupNameInStorage = strings.TrimPrefix(*item.Prefix, strings.Trim(bslPrefix, "/")+"/")
-		fmt.Println(backupNameInStorage)
+		fmt.Println("backupNameInStorage:" + backupNameInStorage)
 		if strings.Contains(backupNameInStorage, backupObject) {
 			fmt.Printf("Backup %s was found under prefix %s \n", backupObject, bslPrefix)
 			return true, nil
@@ -139,3 +143,43 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 	fmt.Printf("Deleted object(s) from bucket: %s %s \n", bslBucket, fullPrefix)
 	return nil
 }
+
+func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
+
+	config := flag.NewMap()
+	config.Set(bslConfig)
+	region := config.Data()["region"]
+	s3Config := &aws.Config{
+		Region:      aws.String(region),
+		Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
+	}
+	if region == "minio" {
+		return errors.New("No snapshot for Minio provider")
+	}
+	sess, err := session.NewSession(s3Config)
+	if err != nil {
+		return errors.Wrapf(err, "Error waiting for uploads to complete")
+	}
+	svc := ec2.New(sess)
+	params := &ec2.DescribeSnapshotsInput{
+		OwnerIds: []*string{aws.String("self")},
+		Filters: []*ec2.Filter{
+			{
+				Name: aws.String("tag:velero.io/backup"),
+				Values: []*string{
+					aws.String(backupObject),
+				},
+			},
+		},
+	}
+	result, err := svc.DescribeSnapshots(params)
+	if err != nil {
+		fmt.Println(err)
+	}
+	if len(result.Snapshots) != snapshotCheck.ExpectCount {
+		return errors.New(fmt.Sprintf("Snapshot count is not as expected %d", snapshotCheck.ExpectCount))
+	} else {
+		fmt.Printf("Snapshot count %d is as expected %d\n", len(result.Snapshots), snapshotCheck.ExpectCount)
+		return nil
+	}
+}

test/e2e/util/providers/azure_utils.go

@@ -22,29 +22,47 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/Azure/azure-pipeline-go/pipeline"
+
+	disk "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2021-08-01/compute"
 	storagemgmt "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2019-06-01/storage"
 	"github.com/Azure/azure-storage-blob-go/azblob"
+	"github.com/Azure/go-autorest/autorest"
 	"github.com/Azure/go-autorest/autorest/azure"
 	"github.com/Azure/go-autorest/autorest/azure/auth"
 	"github.com/joho/godotenv"
 	"github.com/pkg/errors"
 	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
+	. "github.com/vmware-tanzu/velero/test/e2e"
 )
 
 type AzureStorage string
 
+const fqdn = "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2019-07-01/compute"
+
 const (
-	subscriptionIDEnvVar = "AZURE_SUBSCRIPTION_ID"
-	cloudNameEnvVar      = "AZURE_CLOUD_NAME"
-	resourceGroupEnvVar  = "AZURE_RESOURCE_GROUP"
-	storageAccountKey    = "AZURE_STORAGE_ACCOUNT_ACCESS_KEY"
-	storageAccount       = "storageAccount"
-	subscriptionID       = "subscriptionId"
-	resourceGroup        = "resourceGroup"
+	subscriptionIDConfigKey = "subscriptionId"
+	subscriptionIDEnvVar    = "AZURE_SUBSCRIPTION_ID"
+	cloudNameEnvVar         = "AZURE_CLOUD_NAME"
+	resourceGroupEnvVar     = "AZURE_RESOURCE_GROUP"
+	storageAccountKey       = "AZURE_STORAGE_ACCOUNT_ACCESS_KEY"
+	storageAccount          = "storageAccount"
+	subscriptionID          = "subscriptionId"
+	resourceGroup           = "resourceGroup"
+
+	apiTimeoutConfigKey       = "apiTimeout"
+	snapsIncrementalConfigKey = "incremental"
+
+	snapshotsResource = "snapshots"
+	disksResource     = "disks"
+
+	resourceGroupConfigKey   = "resourceGroup"
+	credentialsFileConfigKey = "credentialsFile"
 )
 
 func getStorageCredential(cloudCredentialsFile, bslConfig string) (string, string, error) {
@@ -63,6 +81,7 @@ func getStorageCredential(cloudCredentialsFile, bslConfig string) (string, strin
 	}
 	return accountName, accountKey, nil
 }
+
 func loadCredentialsIntoEnv(credentialsFile string) error {
 	if credentialsFile == "" {
 		return nil
@@ -160,6 +179,42 @@ func handleErrors(err error) {
 	}
 }
 
+func getRequiredValues(getValue func(string) string, keys ...string) (map[string]string, error) {
+	missing := []string{}
+	results := map[string]string{}
+
+	for _, key := range keys {
+		if val := getValue(key); val == "" {
+			missing = append(missing, key)
+		} else {
+			results[key] = val
+		}
+	}
+
+	if len(missing) > 0 {
+		return nil, errors.Errorf("the following keys do not have values: %s", strings.Join(missing, ", "))
+	}
+
+	return results, nil
+}
+
+func validateConfigKeys(config map[string]string, validKeys ...string) error {
+	validKeysSet := sets.NewString(validKeys...)
+
+	var invalidKeys []string
+	for k := range config {
+		if !validKeysSet.Has(k) {
+			invalidKeys = append(invalidKeys, k)
+		}
+	}
+
+	if len(invalidKeys) > 0 {
+		return errors.Errorf("config has invalid keys %v; valid keys are %v", invalidKeys, validKeys)
+	}
+
+	return nil
+}
+
 func deleteBlob(p pipeline.Pipeline, accountName, containerName, blobName string) error {
 	ctx := context.Background()
 
@@ -239,7 +294,7 @@ func (s AzureStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bsl
 	for marker := (azblob.Marker{}); marker.NotDone(); {
 		listBlob, err := containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{})
 		if err != nil {
-			return errors.Wrapf(err, "Fail to create gcloud client")
+			return errors.Wrapf(err, "Fail to list blobs client")
 		}
 
 		marker = listBlob.NextMarker
@@ -256,3 +311,88 @@ func (s AzureStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bsl
 	}
 	return nil
 }
+
+func mapLookup(data map[string]string) func(string) string {
+	return func(key string) string {
+		return data[key]
+	}
+}
+func (s AzureStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
+
+	ctx := context.Background()
+	config := flag.NewMap()
+	config.Set(bslConfig)
+	if err := validateConfigKeys(config.Data(),
+		resourceGroupConfigKey,
+		subscriptionIDConfigKey,
+		storageAccount,
+	); err != nil {
+		return err
+	}
+
+	if err := loadCredentialsIntoEnv(cloudCredentialsFile); err != nil {
+		return err
+	}
+	// we need AZURE_SUBSCRIPTION_ID, AZURE_RESOURCE_GROUP
+	envVars, err := getRequiredValues(os.Getenv, subscriptionIDEnvVar, resourceGroupEnvVar)
+	if err != nil {
+		return errors.Wrap(err, "unable to get all required environment variables")
+	}
+
+	// Get Azure cloud from AZURE_CLOUD_NAME, if it exists. If the env var does not
+	// exist, parseAzureEnvironment will return azure.PublicCloud.
+	env, err := parseAzureEnvironment(os.Getenv(cloudNameEnvVar))
+	if err != nil {
+		return errors.Wrap(err, "unable to parse azure cloud name environment variable")
+	}
+
+	// set a different subscriptionId for snapshots if specified
+	snapshotsSubscriptionID := envVars[subscriptionIDEnvVar]
+	if val := config.Data()[subscriptionIDConfigKey]; val != "" {
+		// if subscription was set in config, it is required to also set the resource group
+		if _, err := getRequiredValues(mapLookup(config.Data()), resourceGroupConfigKey); err != nil {
+			return errors.Wrap(err, "resourceGroup not specified, but is a requirement when backing up to a different subscription")
+		}
+		snapshotsSubscriptionID = val
+	}
+	// set up clients
+	snapsClient := disk.NewSnapshotsClientWithBaseURI(env.ResourceManagerEndpoint, snapshotsSubscriptionID)
+	snapsClient.PollingDelay = 5 * time.Second
+
+	authorizer, err := auth.NewAuthorizerFromEnvironment()
+	if err != nil {
+		return errors.Wrap(err, "error getting authorizer from environment")
+	}
+	// // if config["snapsIncrementalConfigKey"] is empty, default to nil; otherwise, parse i
+	snapsClient.Authorizer = authorizer
+	snaps := &snapsClient
+	//return ListByResourceGroup(ctx, snaps, envVars[resourceGroupEnvVar], backupObject, snapshotCount)
+	req, err := snaps.ListByResourceGroupPreparer(ctx, envVars[resourceGroupEnvVar])
+	if err != nil {
+		return autorest.NewErrorWithError(err, "compute.SnapshotsClient", "ListByResourceGroup", nil, "Failure preparing request")
+	}
+
+	resp, err := snaps.ListByResourceGroupSender(req)
+	if err != nil {
+		return autorest.NewErrorWithError(err, "compute.SnapshotsClient", "ListByResourceGroup", resp, "Failure sending request")
+	}
+	result, err := snaps.ListByResourceGroupResponder(resp)
+	snapshotCountFound := 0
+	backupNameInSnapshot := ""
+	for _, v := range *result.Value {
+		backupNameInSnapshot = *v.Tags["velero.io-backup"]
+		fmt.Println(backupNameInSnapshot)
+		if backupObject == backupNameInSnapshot {
+			snapshotCountFound++
+		}
+	}
+	if err != nil {
+		return autorest.NewErrorWithError(err, "compute.SnapshotsClient", "ListByResourceGroup", resp, "Failure responding to request")
+	}
+	if snapshotCountFound != snapshotCheck.ExpectCount {
+		return errors.New(fmt.Sprintf("Snapshot count %d is not as expected %d\n", snapshotCountFound, snapshotCheck.ExpectCount))
+	} else {
+		fmt.Printf("Snapshot count %d is as expected %d\n", snapshotCountFound, snapshotCheck.ExpectCount)
+		return nil
+	}
+}

test/e2e/util/providers/common.go

@@ -17,16 +17,21 @@ limitations under the License.
 package providers
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
 
 	"github.com/pkg/errors"
+
+	. "github.com/vmware-tanzu/velero/test/e2e"
+	velero "github.com/vmware-tanzu/velero/test/e2e/util/velero"
 )
 
 type ObjectsInStorage interface {
 	IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) (bool, error)
 	DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) error
+	IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error
 }
 
 func ObjectsShouldBeInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string) error {
@@ -92,7 +97,7 @@ func IsObjectsInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix
 
 func DeleteObjectsInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string) error {
 	bslPrefix = getFullPrefix(bslPrefix, subPrefix)
-	fmt.Printf("|| VERIFICATION || - Delete backup %s in storage %s", backupName, bslPrefix)
+	fmt.Printf("|| VERIFICATION || - Delete backup %s in storage %s\n", backupName, bslPrefix)
 	s, err := getProvider(cloudProvider)
 	if err != nil {
 		return errors.Wrapf(err, fmt.Sprintf("Cloud provider %s is not valid", cloudProvider))
@@ -103,3 +108,67 @@ func DeleteObjectsInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPr
 	}
 	return nil
 }
+
+func SnapshotsShouldNotExistInCloud(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string) error {
+	fmt.Printf("|| VERIFICATION || - Snapshots should not exist in cloud, backup %s\n", backupName)
+	var snapshotCheckPoint SnapshotCheckPoint
+	snapshotCheckPoint.ExpectCount = 0
+	err := IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix, snapshotCheckPoint)
+	if err != nil {
+		return errors.Wrapf(err, fmt.Sprintf("|| UNEXPECTED ||Snapshots %s is existed in cloud after backup as expected", backupName))
+	}
+	fmt.Printf("|| EXPECTED || - Snapshots are not existed in cloud, backup %s\n", backupName)
+	return nil
+}
+
+func SnapshotsShouldBeCreatedInCloud(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string, snapshotCheckPoint SnapshotCheckPoint) error {
+	fmt.Printf("|| VERIFICATION || - Snapshots should exist in cloud, backup %s\n", backupName)
+	err := IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix, snapshotCheckPoint)
+	if err != nil {
+		return errors.Wrapf(err, fmt.Sprintf("|| UNEXPECTED ||Snapshots %s are not existed in cloud after backup as expected", backupName))
+	}
+	fmt.Printf("|| EXPECTED || - Snapshots are existed in cloud, backup %s\n", backupName)
+	return nil
+}
+
+func IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string, snapshotCheck SnapshotCheckPoint) error {
+	bslPrefix = getFullPrefix(bslPrefix, subPrefix)
+	s, err := getProvider(cloudProvider)
+	if err != nil {
+		return errors.Wrapf(err, fmt.Sprintf("Cloud provider %s is not valid", cloudProvider))
+	}
+	if cloudProvider == "vsphere" {
+		var retSnapshotIDs []string
+		ctx, _ := context.WithTimeout(context.Background(), time.Minute*2)
+		retSnapshotIDs, err = velero.GetVsphereSnapshotIDs(ctx, time.Hour, snapshotCheck.NamespaceBackedUp)
+		if err != nil {
+			return errors.Wrapf(err, fmt.Sprintf("Fail to get snapshot CRs of backup%s", backupName))
+		}
+		bslPrefix = "plugins"
+		subPrefix = "vsphere-astrolabe-repo/ivd/data"
+		if snapshotCheck.ExpectCount == 0 {
+			for _, snapshotID := range retSnapshotIDs {
+				err := ObjectsShouldNotBeInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, snapshotID, subPrefix, 5)
+				if err != nil {
+					return errors.Wrapf(err, "|| UNEXPECTED || - Snapshot %s of backup %s exist in object store", snapshotID, backupName)
+				}
+			}
+		} else {
+			if snapshotCheck.ExpectCount != len(retSnapshotIDs) {
+				return errors.New(fmt.Sprintf("Snapshot CRs count %d is not equal to expected %d", len(retSnapshotIDs), snapshotCheck.ExpectCount))
+			}
+			for _, snapshotID := range retSnapshotIDs {
+				err := ObjectsShouldBeInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, snapshotID, subPrefix)
+				if err != nil {
+					return errors.Wrapf(err, "|| UNEXPECTED || - Snapshot %s of backup %s does not exist in object store", snapshotID, backupName)
+				}
+			}
+		}
+	} else {
+		err = s.IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, snapshotCheck)
+		if err != nil {
+			return errors.Wrapf(err, fmt.Sprintf("Fail to get snapshot of backup%s", backupName))
+		}
+	}
+	return nil
+}

test/e2e/util/providers/gcloud_utils.go

@@ -17,14 +17,20 @@ limitations under the License.
 package providers
 
 import (
+	"encoding/json"
 	"fmt"
+	"io/ioutil"
 	"strings"
 
 	"cloud.google.com/go/storage"
 	"github.com/pkg/errors"
 	"golang.org/x/net/context"
+	"golang.org/x/oauth2/google"
+	"google.golang.org/api/compute/v1"
 	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
+
+	. "github.com/vmware-tanzu/velero/test/e2e"
 )
 
 type GCSStorage string
@@ -43,7 +49,8 @@ func (s GCSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 	for {
 		obj, err := iter.Next()
 		if err == iterator.Done {
-			return false, errors.Wrapf(err, fmt.Sprintf("Backup %s was not found under prefix %s \n", backupObject, bslPrefix))
+			//return false, errors.Wrapf(err, fmt.Sprintf("Backup %s was not found under prefix %s \n", backupObject, bslPrefix))
+			return false, nil
 		}
 		if err != nil {
 			return false, errors.WithStack(err)
@@ -97,3 +104,43 @@ func (s GCSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 		}
 	}
 }
+
+func (s GCSStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
+	ctx := context.Background()
+	data, err := ioutil.ReadFile(cloudCredentialsFile)
+	if err != nil {
+		return errors.Wrapf(err, fmt.Sprintf("Failed reading gcloud credential file %s", cloudCredentialsFile))
+	}
+
+	creds, err := google.CredentialsFromJSON(ctx, data)
+	if err != nil {
+		return errors.Wrapf(err, fmt.Sprintf("Failed getting credentials from JSON data %s", string(data)))
+	}
+
+	computeService, err := compute.NewService(context.Background(), option.WithCredentialsFile(cloudCredentialsFile))
+	if err != nil {
+		return errors.Wrapf(err, "Fail to create gcloud compute service")
+	}
+	// Project ID for this request.
+	project := creds.ProjectID
+	req := computeService.Snapshots.List(project)
+	snapshotCountFound := 0
+	if err := req.Pages(ctx, func(page *compute.SnapshotList) error {
+		for _, snapshot := range page.Items {
+			snapshotDesc := map[string]string{}
+			json.Unmarshal([]byte(snapshot.Description), &snapshotDesc)
+			if backupObject == snapshotDesc["velero.io/backup"] {
+				snapshotCountFound++
+			}
+		}
+		return nil
+	}); err != nil {
+		return errors.Wrapf(err, "Failed listing snapshot pages")
+	}
+	if snapshotCountFound != len(snapshotCheck.SnapshotIDList) {
+		return errors.New(fmt.Sprintf("Snapshot count %d is not as expected %d\n", snapshotCountFound, len(snapshotCheck.SnapshotIDList)))
+	} else {
+		fmt.Printf("Snapshot count %d is as expected %d\n", snapshotCountFound, len(snapshotCheck.SnapshotIDList))
+		return nil
+	}
+}

test/e2e/util/velero/velero_utils.go

@@ -19,6 +19,7 @@ package velero
 import (
 	"bytes"
 	"context"
+	b64 "encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -41,9 +42,8 @@ import (
 	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
 )
 
-const (
-	BackupObjectsPrefix = "backups"
-)
+const BackupObjectsPrefix = "backups"
+const PluginsObjectsPrefix = "plugins"
 
 var pluginsMatrix = map[string]map[string][]string{
 	"v1.4": {
@@ -478,6 +478,36 @@ func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration,
 	return err
 }
 
+func GetVsphereSnapshotIDs(ctx context.Context, timeout time.Duration, namespace string) ([]string, error) {
+	checkSnapshotCmd := exec.CommandContext(ctx, "kubectl",
+		"get", "-n", namespace, "snapshots.backupdriver.cnsdp.vmware.com", "-o=jsonpath='{range .items[*]}{.spec.resourceHandle.name}{\"=\"}{.status.snapshotID}{\"\\n\"}{end}'")
+	fmt.Printf("checkSnapshotCmd cmd =%v\n", checkSnapshotCmd)
+	stdout, stderr, err := veleroexec.RunCommand(checkSnapshotCmd)
+	if err != nil {
+		fmt.Print(stdout)
+		fmt.Print(stderr)
+		return nil, errors.Wrap(err, "failed to verify")
+	}
+	stdout = strings.Replace(stdout, "'", "", -1)
+	lines := strings.Split(stdout, "\n")
+	var result []string
+
+	for _, curLine := range lines {
+		fmt.Println("curLine:" + curLine)
+		curLine = strings.Replace(curLine, "\n", "", -1)
+		if len(curLine) == 0 {
+			continue
+		}
+		snapshotID := curLine[strings.LastIndex(curLine, ":")+1:]
+		fmt.Println("snapshotID:" + snapshotID)
+		snapshotIDDec, _ := b64.StdEncoding.DecodeString(snapshotID)
+		fmt.Println("snapshotIDDec:" + string(snapshotIDDec))
+		result = append(result, string(snapshotIDDec))
+	}
+	fmt.Println(result)
+	return result, nil
+}
+
 func getVeleroVersion(ctx context.Context, veleroCLI string, clientOnly bool) (string, error) {
 	args := []string{"version", "--timeout", "60s"}
 	if clientOnly {