Add snapshotvolumes default behavior E2E test

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

test/e2e/util/csi/common.go

@@ -129,12 +129,13 @@ func GetVolumeSnapshotContentNameByPod(client TestClient, podName, namespace, ba
 	return "", errors.New(fmt.Sprintf("Fail to get VolumeSnapshotContentName for pod %s under namespace %s", podName, namespace))
 }
 
-func CheckVolumeSnapshotCR(client TestClient, backupName string, expectedCount int) error {
+func CheckVolumeSnapshotCR(client TestClient, backupName string, expectedCount int) ([]string, error) {
 	var err error
 	var snapshotContentNameList []string
-	if snapshotContentNameList, err = GetCsiSnapshotHandle(client, backupName); err != nil || len(snapshotContentNameList) != expectedCount {
-		return errors.Wrap(err, "Fail to get Azure CSI snapshot content")
+	if snapshotContentNameList, err = GetCsiSnapshotHandle(client, backupName); err != nil ||
+		len(snapshotContentNameList) != expectedCount {
+		return nil, errors.Wrap(err, "Fail to get Azure CSI snapshot content")
 	}
 	fmt.Println(snapshotContentNameList)
-	return nil
+	return snapshotContentNameList, nil
 }

test/e2e/util/k8s/common.go

@@ -341,3 +341,17 @@ func WaitForCRDEstablished(crdName string) error {
 	}
 	return nil
 }
+
+func GetAllService(ctx context.Context) (string, error) {
+	args := []string{"get", "service", "-A"}
+	cmd := exec.CommandContext(context.Background(), "kubectl", args...)
+	fmt.Printf("Kubectl exec cmd =%v\n", cmd)
+	stdout, stderr, err := veleroexec.RunCommand(cmd)
+	fmt.Println(stdout)
+	if err != nil {
+		fmt.Println(stderr)
+		fmt.Println(err)
+		return "", err
+	}
+	return stdout, nil
+}

test/e2e/util/k8s/deployment.go

@@ -36,7 +36,16 @@ const (
 )
 
 // newDeployment returns a RollingUpdate Deployment with a fake container image
-func NewDeployment(name, ns string, replicas int32, labels map[string]string) *apps.Deployment {
+func NewDeployment(name, ns string, replicas int32, labels map[string]string, containers []v1.Container) *apps.Deployment {
+	if containers == nil {
+		containers = []v1.Container{
+			{
+				Name:    fmt.Sprintf("container-%s", "busybox"),
+				Image:   "gcr.io/velero-gcp/busybox:latest",
+				Command: []string{"sleep", "1000000"},
+			},
+		}
+	}
 	return &apps.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Deployment",
@@ -59,19 +68,17 @@ func NewDeployment(name, ns string, replicas int32, labels map[string]string) *a
 					Labels: labels,
 				},
 				Spec: v1.PodSpec{
-					Containers: []v1.Container{
-						{
-							Name:    name,
-							Image:   "gcr.io/velero-gcp/busybox:latest",
-							Command: []string{"sleep", "1000000"},
-						},
-					},
+					Containers: containers,
 				},
 			},
 		},
 	}
 }
 
+func CreateDeploy(c clientset.Interface, ns string, deployment *apps.Deployment) error {
+	_, err := c.AppsV1().Deployments(ns).Create(context.TODO(), deployment, metav1.CreateOptions{})
+	return err
+}
 func CreateDeployment(c clientset.Interface, ns string, deployment *apps.Deployment) (*apps.Deployment, error) {
 	return c.AppsV1().Deployments(ns).Create(context.TODO(), deployment, metav1.CreateOptions{})
 }

test/e2e/util/k8s/namespace.go

@@ -100,7 +100,7 @@ func CleanupNamespacesWithPoll(ctx context.Context, client TestClient, nsBaseNam
 			if err != nil {
 				return errors.Wrapf(err, "Could not delete namespace %s", checkNamespace.Name)
 			}
-			fmt.Printf("Delete namespace %s\n", checkNamespace.Name)
+			fmt.Printf("Namespace %s was deleted\n", checkNamespace.Name)
 		}
 	}
 	return nil

test/e2e/util/k8s/service.go

@@ -0,0 +1,75 @@
+/*
+Copyright the Velero contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package k8s
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	waitutil "k8s.io/apimachinery/pkg/util/wait"
+)
+
+func CreateService(ctx context.Context, client TestClient, namespace string,
+	service string, labels map[string]string, serviceSpec *corev1.ServiceSpec) error {
+	se := &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Service",
+			APIVersion: "apps/v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      service,
+			Labels:    labels,
+			Namespace: namespace,
+		},
+		Spec: *serviceSpec,
+	}
+	_, err = client.ClientGo.CoreV1().Services(namespace).Create(ctx, se, metav1.CreateOptions{})
+
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		return err
+	}
+	return nil
+}
+
+func GetService(ctx context.Context, client TestClient, namespace string, service string) (*corev1.Service, error) {
+	return client.ClientGo.CoreV1().Services(namespace).Get(ctx, service, metav1.GetOptions{})
+}
+
+func WaitForServiceDelete(client TestClient, ns, name string, deleteFirst bool) error {
+	if deleteFirst {
+		if err := client.ClientGo.CoreV1().Services(ns).Delete(context.TODO(), name, metav1.DeleteOptions{}); err != nil {
+			return errors.Wrap(err, fmt.Sprintf("failed to delete  secret in namespace %q", ns))
+		}
+	}
+	return waitutil.PollImmediateInfinite(5*time.Second,
+		func() (bool, error) {
+			if _, err := client.ClientGo.CoreV1().Services(ns).Get(context.TODO(), ns, metav1.GetOptions{}); err != nil {
+				if apierrors.IsNotFound(err) {
+					return true, nil
+				}
+				return false, err
+			}
+			logrus.Debugf("service %q in namespace %q is still being deleted...", name, ns)
+			return false, nil
+		})
+}

test/e2e/util/kibishii/kibishii_utils.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/pkg/errors"
 	"golang.org/x/net/context"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
 	. "github.com/vmware-tanzu/velero/test/e2e"
@@ -51,15 +52,16 @@ var DefaultKibishiiData = &KibishiiData{2, 10, 10, 1024, 1024, 0, 2}
 var KibishiiPodNameList = []string{"kibishii-deployment-0", "kibishii-deployment-1"}
 
 // RunKibishiiTests runs kibishii tests on the provider.
-func RunKibishiiTests(client TestClient, veleroCfg VeleroConfig, backupName, restoreName, backupLocation, kibishiiNamespace string,
-	useVolumeSnapshots bool) error {
+func RunKibishiiTests(veleroCfg VeleroConfig, backupName, restoreName, backupLocation, kibishiiNamespace string,
+	useVolumeSnapshots, defaultVolumesToFsBackup bool) error {
+	client := *veleroCfg.ClientToInstallVelero
 	oneHourTimeout, _ := context.WithTimeout(context.Background(), time.Minute*60)
-	veleroCLI := VeleroCfg.VeleroCLI
-	providerName := VeleroCfg.CloudProvider
-	veleroNamespace := VeleroCfg.VeleroNamespace
-	registryCredentialFile := VeleroCfg.RegistryCredentialFile
-	veleroFeatures := VeleroCfg.Features
-	kibishiiDirectory := VeleroCfg.KibishiiDirectory
+	veleroCLI := veleroCfg.VeleroCLI
+	providerName := veleroCfg.CloudProvider
+	veleroNamespace := veleroCfg.VeleroNamespace
+	registryCredentialFile := veleroCfg.RegistryCredentialFile
+	veleroFeatures := veleroCfg.Features
+	kibishiiDirectory := veleroCfg.KibishiiDirectory
 	if _, err := GetNamespace(context.Background(), client, kibishiiNamespace); err == nil {
 		fmt.Printf("Workload namespace %s exists, delete it first.\n", kibishiiNamespace)
 		if err = DeleteNamespace(context.Background(), client, kibishiiNamespace, true); err != nil {
@@ -88,32 +90,66 @@ func RunKibishiiTests(client TestClient, veleroCfg VeleroConfig, backupName, res
 	BackupCfg.Namespace = kibishiiNamespace
 	BackupCfg.BackupLocation = backupLocation
 	BackupCfg.UseVolumeSnapshots = useVolumeSnapshots
+	BackupCfg.DefaultVolumesToFsBackup = defaultVolumesToFsBackup
 	BackupCfg.Selector = ""
+	BackupCfg.ProvideSnapshotsVolumeParam = veleroCfg.ProvideSnapshotsVolumeParam
 	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, BackupCfg); err != nil {
 		RunDebug(context.Background(), veleroCLI, veleroNamespace, backupName, "")
 		return errors.Wrapf(err, "Failed to backup kibishii namespace %s", kibishiiNamespace)
 	}
 	var snapshotCheckPoint SnapshotCheckPoint
 	var err error
+	pvbs, err := GetPVB(oneHourTimeout, veleroCfg.VeleroNamespace, kibishiiNamespace)
 	if useVolumeSnapshots {
+		if err != nil || len(pvbs) != 0 {
+			return errors.Wrapf(err, "failed to get PVB for namespace %s", kibishiiNamespace)
+		}
 		if providerName == "vsphere" {
 			// Wait for uploads started by the Velero Plug-in for vSphere to complete
 			// TODO - remove after upload progress monitoring is implemented
 			fmt.Println("Waiting for vSphere uploads to complete")
-			if err := WaitForVSphereUploadCompletion(oneHourTimeout, time.Hour, kibishiiNamespace); err != nil {
+			if err := WaitForVSphereUploadCompletion(oneHourTimeout, time.Hour, kibishiiNamespace, 2); err != nil {
 				return errors.Wrapf(err, "Error waiting for uploads to complete")
 			}
 		}
-		snapshotCheckPoint, err = GetSnapshotCheckPoint(client, VeleroCfg, 2, kibishiiNamespace, backupName, KibishiiPodNameList)
+		snapshotCheckPoint, err = GetSnapshotCheckPoint(client, veleroCfg, 2, kibishiiNamespace, backupName, KibishiiPodNameList)
 		if err != nil {
 			return errors.Wrap(err, "Fail to get snapshot checkpoint")
 		}
-		err = SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
-			VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, veleroCfg.BSLConfig,
+		err = SnapshotsShouldBeCreatedInCloud(veleroCfg.CloudProvider,
+			veleroCfg.CloudCredentialsFile, veleroCfg.BSLBucket, veleroCfg.BSLConfig,
 			backupName, snapshotCheckPoint)
 		if err != nil {
 			return errors.Wrap(err, "exceed waiting for snapshot created in cloud")
 		}
+	} else {
+		if err != nil || len(pvbs) != 2 {
+			return errors.Wrapf(err, "failed to get PVB for namespace %s", kibishiiNamespace)
+		}
+		if providerName == "vsphere" {
+			// Wait for uploads started by the Velero Plug-in for vSphere to complete
+			// TODO - remove after upload progress monitoring is implemented
+
+			// TODO[High] - uncomment code block below when vSphere plugin PR #500 is included in release version.
+			//   https://github.com/vmware-tanzu/velero-plugin-for-vsphere/pull/500/
+
+			// fmt.Println("Make sure no vSphere snapshot uploads created")
+			// if err := WaitForVSphereUploadCompletion(oneHourTimeout, time.Hour, kibishiiNamespace, 0); err != nil {
+			// 	return errors.Wrapf(err, "Error get vSphere snapshot uploads")
+			// }
+		} else {
+			if strings.EqualFold(veleroFeatures, "EnableCSI") {
+				_, err = GetSnapshotCheckPoint(*veleroCfg.ClientToInstallVelero, veleroCfg, 0,
+					kibishiiNamespace, backupName, KibishiiPodNameList)
+			} else {
+				err = SnapshotsShouldNotExistInCloud(veleroCfg.CloudProvider,
+					veleroCfg.CloudCredentialsFile, veleroCfg.BSLBucket, veleroCfg.BSLConfig,
+					backupName, snapshotCheckPoint)
+				if err != nil {
+					return errors.Wrap(err, "exceed waiting for snapshot created in cloud")
+				}
+			}
+		}
 	}
 
 	fmt.Printf("Simulating a disaster by removing namespace %s\n", kibishiiNamespace)
@@ -133,6 +169,12 @@ func RunKibishiiTests(client TestClient, veleroCfg VeleroConfig, backupName, res
 		RunDebug(context.Background(), veleroCLI, veleroNamespace, "", restoreName)
 		return errors.Wrapf(err, "Restore %s failed from backup %s", restoreName, backupName)
 	}
+	if !useVolumeSnapshots {
+		pvrs, err := GetPVR(oneHourTimeout, veleroCfg.VeleroNamespace, kibishiiNamespace)
+		if err != nil || len(pvrs) != 2 {
+			return errors.Wrapf(err, "failed to get PVB for namespace %s", kibishiiNamespace)
+		}
+	}
 
 	if err := KibishiiVerifyAfterRestore(client, kibishiiNamespace, oneHourTimeout, DefaultKibishiiData); err != nil {
 		return errors.Wrapf(err, "Error verifying kibishii after restore")
@@ -174,33 +216,55 @@ func installKibishii(ctx context.Context, namespace string, cloudPlatform, veler
 }
 
 func generateData(ctx context.Context, namespace string, kibishiiData *KibishiiData) error {
-	timeout, _ := context.WithTimeout(context.Background(), time.Minute*20)
-	kibishiiGenerateCmd := exec.CommandContext(timeout, "kubectl", "exec", "-n", namespace, "jump-pad", "--",
-		"/usr/local/bin/generate.sh", strconv.Itoa(kibishiiData.Levels), strconv.Itoa(kibishiiData.DirsPerLevel),
-		strconv.Itoa(kibishiiData.FilesPerLevel), strconv.Itoa(kibishiiData.FileLength),
-		strconv.Itoa(kibishiiData.BlockSize), strconv.Itoa(kibishiiData.PassNum), strconv.Itoa(kibishiiData.ExpectedNodes))
-	fmt.Printf("kibishiiGenerateCmd cmd =%v\n", kibishiiGenerateCmd)
+	timeout := 30 * time.Minute
+	interval := 1 * time.Second
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		timeout, _ := context.WithTimeout(context.Background(), time.Minute*20)
+		kibishiiGenerateCmd := exec.CommandContext(timeout, "kubectl", "exec", "-n", namespace, "jump-pad", "--",
+			"/usr/local/bin/generate.sh", strconv.Itoa(kibishiiData.Levels), strconv.Itoa(kibishiiData.DirsPerLevel),
+			strconv.Itoa(kibishiiData.FilesPerLevel), strconv.Itoa(kibishiiData.FileLength),
+			strconv.Itoa(kibishiiData.BlockSize), strconv.Itoa(kibishiiData.PassNum), strconv.Itoa(kibishiiData.ExpectedNodes))
+		fmt.Printf("kibishiiGenerateCmd cmd =%v\n", kibishiiGenerateCmd)
+
+		stdout, stderr, err := veleroexec.RunCommand(kibishiiGenerateCmd)
+		if err != nil || strings.Contains(stderr, "Timeout occurred") || strings.Contains(stderr, "dialing backend") {
+			fmt.Printf("Kibishi generate stdout Timeout occurred: %s stderr: %s err: %s", stdout, stderr, err)
+			return false, nil
+		}
+		return true, nil
+	})
 
-	_, stderr, err := veleroexec.RunCommand(kibishiiGenerateCmd)
 	if err != nil {
-		return errors.Wrapf(err, "failed to generate, stderr=%s", stderr)
+		return errors.Wrapf(err, fmt.Sprintf("Failed to wait generate data in namespace %s", namespace))
 	}
-
 	return nil
 }
 
 func verifyData(ctx context.Context, namespace string, kibishiiData *KibishiiData) error {
-	timeout, _ := context.WithTimeout(context.Background(), time.Minute*10)
-	kibishiiVerifyCmd := exec.CommandContext(timeout, "kubectl", "exec", "-n", namespace, "jump-pad", "--",
-		"/usr/local/bin/verify.sh", strconv.Itoa(kibishiiData.Levels), strconv.Itoa(kibishiiData.DirsPerLevel),
-		strconv.Itoa(kibishiiData.FilesPerLevel), strconv.Itoa(kibishiiData.FileLength),
-		strconv.Itoa(kibishiiData.BlockSize), strconv.Itoa(kibishiiData.PassNum),
-		strconv.Itoa(kibishiiData.ExpectedNodes))
-	fmt.Printf("kibishiiVerifyCmd cmd =%v\n", kibishiiVerifyCmd)
+	timeout := 10 * time.Minute
+	interval := 5 * time.Second
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		timeout, _ := context.WithTimeout(context.Background(), time.Minute*20)
+		kibishiiVerifyCmd := exec.CommandContext(timeout, "kubectl", "exec", "-n", namespace, "jump-pad", "--",
+			"/usr/local/bin/verify.sh", strconv.Itoa(kibishiiData.Levels), strconv.Itoa(kibishiiData.DirsPerLevel),
+			strconv.Itoa(kibishiiData.FilesPerLevel), strconv.Itoa(kibishiiData.FileLength),
+			strconv.Itoa(kibishiiData.BlockSize), strconv.Itoa(kibishiiData.PassNum),
+			strconv.Itoa(kibishiiData.ExpectedNodes))
+		fmt.Printf("kibishiiVerifyCmd cmd =%v\n", kibishiiVerifyCmd)
+
+		stdout, stderr, err := veleroexec.RunCommand(kibishiiVerifyCmd)
+		if strings.Contains(stderr, "Timeout occurred") {
+			return false, nil
+		}
+		if err != nil {
+			fmt.Printf("Kibishi verify stdout Timeout occurred: %s stderr: %s err: %s", stdout, stderr, err)
+			return false, nil
+		}
+		return true, nil
+	})
 
-	stdout, stderr, err := veleroexec.RunCommand(kibishiiVerifyCmd)
 	if err != nil {
-		return errors.Wrapf(err, "failed to verify, stderr=%s, stdout=%s", stderr, stdout)
+		return errors.Wrapf(err, fmt.Sprintf("Failed to wait generate data in namespace %s", namespace))
 	}
 	return nil
 }

test/e2e/util/velero/install.go

@@ -49,11 +49,9 @@ type installOptions struct {
 	RestoreHelperImage     string
 }
 
-func VeleroInstall(ctx context.Context, veleroCfg *VeleroConfig, useVolumeSnapshots bool) error {
+func VeleroInstall(ctx context.Context, veleroCfg *VeleroConfig) error {
 	if veleroCfg.CloudProvider != "kind" {
-		if veleroCfg.ObjectStoreProvider != "" {
-			return errors.New("For cloud platforms, object store plugin cannot be overridden") // Can't set an object store provider that is different than your cloud
-		}
+		fmt.Printf("For cloud platforms, object store plugin provider will be set as cloud provider")
 		veleroCfg.ObjectStoreProvider = veleroCfg.CloudProvider
 	} else {
 		if veleroCfg.ObjectStoreProvider == "" {
@@ -81,14 +79,13 @@ func VeleroInstall(ctx context.Context, veleroCfg *VeleroConfig, useVolumeSnapsh
 		}
 	}
 
-	veleroInstallOptions, err := getProviderVeleroInstallOptions(veleroCfg.ObjectStoreProvider, veleroCfg.CloudCredentialsFile, veleroCfg.BSLBucket,
-		veleroCfg.BSLPrefix, veleroCfg.BSLConfig, veleroCfg.VSLConfig, providerPluginsTmp, veleroCfg.Features)
+	veleroInstallOptions, err := getProviderVeleroInstallOptions(veleroCfg, providerPluginsTmp)
 	if err != nil {
 		return errors.WithMessagef(err, "Failed to get Velero InstallOptions for plugin provider %s", veleroCfg.ObjectStoreProvider)
 	}
-	veleroInstallOptions.UseVolumeSnapshots = useVolumeSnapshots
+	veleroInstallOptions.UseVolumeSnapshots = veleroCfg.UseVolumeSnapshots
 	if !veleroCfg.UseRestic {
-		veleroInstallOptions.UseNodeAgent = !useVolumeSnapshots
+		veleroInstallOptions.UseNodeAgent = veleroCfg.UseNodeAgent
 	}
 	veleroInstallOptions.UseRestic = veleroCfg.UseRestic
 	veleroInstallOptions.Image = veleroCfg.VeleroImage
@@ -179,6 +176,9 @@ func installVeleroServer(ctx context.Context, cli string, options *installOption
 	if options.UseNodeAgent {
 		args = append(args, "--use-node-agent")
 	}
+	if options.DefaultVolumesToFsBackup {
+		args = append(args, "--default-volumes-to-fs-backup")
+	}
 	if options.UseRestic {
 		args = append(args, "--use-restic")
 	}

test/e2e/util/velero/velero_utils.go

@@ -135,22 +135,14 @@ func GetProviderPluginsByVersion(version, providerName, feature string) ([]strin
 }
 
 // getProviderVeleroInstallOptions returns Velero InstallOptions for the provider.
-func getProviderVeleroInstallOptions(
-	pluginProvider,
-	credentialsFile,
-	objectStoreBucket,
-	objectStorePrefix,
-	bslConfig,
-	vslConfig string,
-	plugins []string,
-	features string,
-) (*cliinstall.InstallOptions, error) {
+func getProviderVeleroInstallOptions(veleroCfg *VeleroConfig,
+	plugins []string) (*cliinstall.InstallOptions, error) {
 
-	if credentialsFile == "" {
+	if veleroCfg.CloudCredentialsFile == "" {
 		return nil, errors.Errorf("No credentials were supplied to use for E2E tests")
 	}
 
-	realPath, err := filepath.Abs(credentialsFile)
+	realPath, err := filepath.Abs(veleroCfg.CloudCredentialsFile)
 	if err != nil {
 		return nil, err
 	}
@@ -158,20 +150,22 @@ func getProviderVeleroInstallOptions(
 	io := cliinstall.NewInstallOptions()
 	// always wait for velero and restic pods to be running.
 	io.Wait = true
-	io.ProviderName = pluginProvider
-	io.SecretFile = credentialsFile
+	io.ProviderName = veleroCfg.ObjectStoreProvider
+	io.SecretFile = veleroCfg.CloudCredentialsFile
 
-	io.BucketName = objectStoreBucket
-	io.Prefix = objectStorePrefix
+	io.BucketName = veleroCfg.BSLBucket
+	io.Prefix = veleroCfg.BSLPrefix
 	io.BackupStorageConfig = flag.NewMap()
-	io.BackupStorageConfig.Set(bslConfig)
+	io.BackupStorageConfig.Set(veleroCfg.BSLConfig)
 
 	io.VolumeSnapshotConfig = flag.NewMap()
-	io.VolumeSnapshotConfig.Set(vslConfig)
+	io.VolumeSnapshotConfig.Set(veleroCfg.VSLConfig)
 
 	io.SecretFile = realPath
 	io.Plugins = flag.NewStringArray(plugins...)
-	io.Features = features
+	io.Features = veleroCfg.Features
+	io.DefaultVolumesToFsBackup = veleroCfg.DefaultVolumesToFsBackup
+	io.UseVolumeSnapshots = veleroCfg.UseVolumeSnapshots
 	return io, nil
 }
 
@@ -346,8 +340,11 @@ func VeleroBackupNamespace(ctx context.Context, veleroCLI, veleroNamespace strin
 	}
 
 	if backupCfg.UseVolumeSnapshots {
-		args = append(args, "--snapshot-volumes")
-	} else {
+		if backupCfg.ProvideSnapshotsVolumeParam {
+			args = append(args, "--snapshot-volumes")
+		}
+	}
+	if backupCfg.DefaultVolumesToFsBackup {
 		if backupCfg.UseResticIfFSBackup {
 			args = append(args, "--default-volumes-to-restic")
 		} else {
@@ -358,7 +355,9 @@ func VeleroBackupNamespace(ctx context.Context, veleroCLI, veleroNamespace strin
 		// if the "--snapshot-volumes=false" isn't specified explicitly, the vSphere plugin will always take snapshots
 		// for the volumes even though the "--default-volumes-to-fs-backup" is specified
 		// TODO This can be removed if the logic of vSphere plugin bump up to 1.3
-		args = append(args, "--snapshot-volumes=false")
+		if backupCfg.ProvideSnapshotsVolumeParam && !backupCfg.UseVolumeSnapshots {
+			args = append(args, "--snapshot-volumes=false")
+		} // if "--snapshot-volumes" is not provide, snapshot should be taken as default behavior.
 	}
 	if backupCfg.BackupLocation != "" {
 		args = append(args, "--storage-location", backupCfg.BackupLocation)
@@ -417,15 +416,15 @@ func VeleroRestore(ctx context.Context, veleroCLI, veleroNamespace, restoreName,
 	if includeResources != "" {
 		args = append(args, "--include-resources", includeResources)
 	}
-	return VeleroRestoreExec(ctx, veleroCLI, veleroNamespace, restoreName, args)
+	return VeleroRestoreExec(ctx, veleroCLI, veleroNamespace, restoreName, args, velerov1api.RestorePhaseCompleted)
 }
 
-func VeleroRestoreExec(ctx context.Context, veleroCLI, veleroNamespace, restoreName string, args []string) error {
+func VeleroRestoreExec(ctx context.Context, veleroCLI, veleroNamespace, restoreName string, args []string, phaseExpect velerov1api.RestorePhase) error {
 	if err := VeleroCmdExec(ctx, veleroCLI, args); err != nil {
 		return err
 	}
 
-	return checkRestorePhase(ctx, veleroCLI, veleroNamespace, restoreName, velerov1api.RestorePhaseCompleted)
+	return checkRestorePhase(ctx, veleroCLI, veleroNamespace, restoreName, phaseExpect)
 }
 
 func VeleroBackupExec(ctx context.Context, veleroCLI string, veleroNamespace string, backupName string, args []string) error {
@@ -514,7 +513,7 @@ func RunDebug(ctx context.Context, veleroCLI, veleroNamespace, backup, restore s
 		args = append(args, "--backup", backup)
 	}
 	if len(restore) > 0 {
-		args = append(args, "--restore", restore)
+		//args = append(args, "--restore", restore)
 	}
 	fmt.Printf("Generating the debug tarball at %s\n", output)
 	if err := VeleroCmdExec(ctx, veleroCLI, args); err != nil {
@@ -620,7 +619,7 @@ func VeleroAddPluginsForProvider(ctx context.Context, veleroCLI string, veleroNa
 
 // WaitForVSphereUploadCompletion waits for uploads started by the Velero Plug-in for vSphere to complete
 // TODO - remove after upload progress monitoring is implemented
-func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration, namespace string) error {
+func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration, namespace string, expectCount int) error {
 	err := wait.PollImmediate(time.Second*5, timeout, func() (bool, error) {
 		checkSnapshotCmd := exec.CommandContext(ctx, "kubectl",
 			"get", "-n", namespace, "snapshots.backupdriver.cnsdp.vmware.com", "-o=jsonpath='{range .items[*]}{.spec.resourceHandle.name}{\"=\"}{.status.phase}{\"\\n\"}{end}'")
@@ -629,10 +628,12 @@ func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration,
 		if err != nil {
 			fmt.Print(stdout)
 			fmt.Print(stderr)
-			return false, errors.Wrap(err, "failed to verify")
+			return false, errors.Wrap(err, "failed to wait for vSphere upload completion")
 		}
 		lines := strings.Split(stdout, "\n")
 		complete := true
+		actualCount := 0
+
 		for _, curLine := range lines {
 			fmt.Println(curLine)
 			comps := strings.Split(curLine, "=")
@@ -651,6 +652,7 @@ func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration,
 			// Canceled - the operation was canceled, the snapshot ID is not valid
 			if len(comps) == 2 {
 				phase := comps[1]
+				actualCount++
 				switch phase {
 				case "Uploaded":
 				case "New", "InProgress", "Snapshotted", "Uploading":
@@ -660,6 +662,14 @@ func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration,
 				}
 			}
 		}
+
+		if expectCount != actualCount {
+			fmt.Printf("Snapshot expect count and actual count: %d-%d", expectCount, actualCount)
+			complete = false
+			if expectCount == 0 {
+				return true, nil
+			}
+		}
 		return complete, nil
 	})
 
@@ -1070,24 +1080,15 @@ func GetResticRepositories(ctx context.Context, veleroNamespace, targetNamespace
 
 func GetSnapshotCheckPoint(client TestClient, VeleroCfg VeleroConfig, expectCount int, namespaceBackedUp, backupName string, kibishiiPodNameList []string) (SnapshotCheckPoint, error) {
 	var snapshotCheckPoint SnapshotCheckPoint
-
+	var err error
 	snapshotCheckPoint.ExpectCount = expectCount
 	snapshotCheckPoint.NamespaceBackedUp = namespaceBackedUp
 	snapshotCheckPoint.PodName = kibishiiPodNameList
 	if VeleroCfg.CloudProvider == "azure" && strings.EqualFold(VeleroCfg.Features, "EnableCSI") {
 		snapshotCheckPoint.EnableCSI = true
-		if err := util.CheckVolumeSnapshotCR(client, backupName, expectCount); err != nil {
+		if snapshotCheckPoint.SnapshotIDList, err = util.CheckVolumeSnapshotCR(client, backupName, expectCount); err != nil {
 			return snapshotCheckPoint, errors.Wrapf(err, "Fail to get Azure CSI snapshot content")
 		}
-		var err error
-		snapshotCheckPoint.SnapshotIDList, err = util.GetCsiSnapshotHandle(client, backupName)
-		if err != nil {
-			return snapshotCheckPoint, errors.New(fmt.Sprintf("Fail to get CSI SnapshotHandle for backup %s", backupName))
-		}
-		fmt.Println(snapshotCheckPoint)
-		if len(snapshotCheckPoint.SnapshotIDList) != expectCount {
-			return snapshotCheckPoint, errors.New(fmt.Sprintf("Length of SnapshotIDList is not as expected %d", expectCount))
-		}
 	}
 	fmt.Println(snapshotCheckPoint)
 	return snapshotCheckPoint, nil
@@ -1208,7 +1209,7 @@ func UpdateVeleroDeployment(ctx context.Context, veleroCfg VeleroConfig) ([]stri
 	if veleroCfg.CloudProvider == "vsphere" {
 		args = fmt.Sprintf("s#\\\"image\\\"\\: \\\"velero\\/velero\\:v[0-9]*.[0-9]*.[0-9]\\\"#\\\"image\\\"\\: \\\"harbor-repo.vmware.com\\/velero_ci\\/velero\\:%s\\\"#g", veleroCfg.VeleroVersion)
 	} else {
-		args = fmt.Sprintf("s#\\\"image\\\"\\: \\\"velero\\/velero\\:v[0-9]*.[0-9]*.[0-9]\\\"#\\\"image\\\"\\: \\\"velero\\/velero\\:%s\\\"#g", veleroCfg.VeleroVersion)
+		args = fmt.Sprintf("s#\\\"image\\\"\\: \\\"velero\\/velero\\:v[0-9]*.[0-9]*.[0-9]\\\"#\\\"image\\\"\\: \\\"gcr.io\\/velero-gcp\\/nightly\\/velero\\:%s\\\"#g", veleroCfg.VeleroVersion)
 	}
 	cmd = &common.OsCommandLine{
 		Cmd:  "sed",
@@ -1261,7 +1262,7 @@ func UpdateNodeAgent(ctx context.Context, veleroCfg VeleroConfig, dsjson string)
 	if veleroCfg.CloudProvider == "vsphere" {
 		args = fmt.Sprintf("s#\\\"image\\\"\\: \\\"velero\\/velero\\:v[0-9]*.[0-9]*.[0-9]\\\"#\\\"image\\\"\\: \\\"harbor-repo.vmware.com\\/velero_ci\\/velero\\:%s\\\"#g", veleroCfg.VeleroVersion)
 	} else {
-		args = fmt.Sprintf("s#\\\"image\\\"\\: \\\"velero\\/velero\\:v[0-9]*.[0-9]*.[0-9]\\\"#\\\"image\\\"\\: \\\"velero\\/velero\\:%s\\\"#g", veleroCfg.VeleroVersion)
+		args = fmt.Sprintf("s#\\\"image\\\"\\: \\\"velero\\/velero\\:v[0-9]*.[0-9]*.[0-9]\\\"#\\\"image\\\"\\: \\\"gcr.io\\/velero-gcp\\/nightly\\/velero\\:%s\\\"#g", veleroCfg.VeleroVersion)
 	}
 	cmd = &common.OsCommandLine{
 		Cmd:  "sed",
@@ -1289,3 +1290,34 @@ func UpdateNodeAgent(ctx context.Context, veleroCfg VeleroConfig, dsjson string)
 
 	return common.GetListByCmdPipes(ctx, cmds)
 }
+
+func GetVeleroResource(ctx context.Context, veleroNamespace, namespace, resourceName string) ([]string, error) {
+	cmds := []*common.OsCommandLine{}
+	cmd := &common.OsCommandLine{
+		Cmd:  "kubectl",
+		Args: []string{"get", resourceName, "-n", veleroNamespace},
+	}
+	cmds = append(cmds, cmd)
+
+	cmd = &common.OsCommandLine{
+		Cmd:  "grep",
+		Args: []string{namespace},
+	}
+	cmds = append(cmds, cmd)
+
+	cmd = &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+	cmds = append(cmds, cmd)
+
+	return common.GetListByCmdPipes(ctx, cmds)
+}
+
+func GetPVB(ctx context.Context, veleroNamespace, namespace string) ([]string, error) {
+	return GetVeleroResource(ctx, veleroNamespace, namespace, "podvolumebackup")
+}
+
+func GetPVR(ctx context.Context, veleroNamespace, namespace string) ([]string, error) {
+	return GetVeleroResource(ctx, veleroNamespace, namespace, "podvolumerestore")
+}