monitor velero logs and fix E2E issues

1. Capture Velero pod log and K8S cluster event; 2. Fix wrong path of storageclass yaml file issue caused by pert test; 3. Fix change storageclass test issue that no sc named 'default' in EKS cluster; 4. Support AWS credential as config format; 5. Support more E2E script input parameters like standy cluster plugins and provider. Signed-off-by: danfengl <danfengl@vmware.com>
2023-08-14 03:48:34 +00:00 · 2023-08-14 03:48:34 +00:00 · 15be42f47b
parent 164431b2b3
commit 15be42f47b
23 changed files with 466 additions and 139 deletions
--- a/test/e2e/Makefile
+++ b/test/e2e/Makefile
@ -51,7 +51,6 @@ SKIP_STR := $(foreach var, $(subst ., ,$(GINKGO_SKIP)),-skip "$(var)")
 FOCUS_STR := $(foreach var, $(subst ., ,$(GINKGO_FOCUS)),-focus "$(var)")
 VELERO_CLI ?=$$(pwd)/../../_output/bin/$(GOOS)/$(GOARCH)/velero
 VELERO_IMAGE ?= velero/velero:main
-VELERO_VERSION ?= $(VERSION)
 PLUGINS ?=
 RESTORE_HELPER_IMAGE ?=
 #Released version only
@ -74,6 +73,8 @@ BSL_CONFIG ?=
 VSL_CONFIG ?=
 CLOUD_PROVIDER ?=
 STANDBY_CLUSTER_CLOUD_PROVIDER ?=
+STANDBY_CLUSTER_PLUGINS ?=
+STANDBY_CLUSTER_OBJECT_STORE_PROVIDER ?=
 OBJECT_STORE_PROVIDER ?=
 INSTALL_VELERO ?= true
 REGISTRY_CREDENTIAL_FILE ?=
@ -90,6 +91,7 @@ ADDITIONAL_BSL_CONFIG ?=

 FEATURES ?=
 DEBUG_E2E_TEST ?= false
+DEBUG_VELERO_POD_RESTART ?= false
 VELERO_SERVER_DEBUG_MODE ?= false

 # Parameters to run migration tests along with all other E2E tests, and both of them should
@ -119,7 +121,7 @@ run: ginkgo
 	@$(GINKGO) -v $(FOCUS_STR) $(SKIP_STR) . -- -velerocli=$(VELERO_CLI) \
 		-velero-image=$(VELERO_IMAGE) \
 		-plugins=$(PLUGINS) \
-		-velero-version=$(VELERO_VERSION) \
+		-velero-version=$(VERSION) \
 		-restore-helper-image=$(RESTORE_HELPER_IMAGE) \
 		-upgrade-from-velero-cli=$(UPGRADE_FROM_VELERO_CLI) \
 		-upgrade-from-velero-version=$(UPGRADE_FROM_VELERO_VERSION) \
@ -150,7 +152,10 @@ run: ginkgo
 		-uploader-type=$(UPLOADER_TYPE) \
 		-snapshot-move-data=$(SNAPSHOT_MOVE_DATA) \
 		-data-mover-plugin=$(DATA_MOVER_plugin) \
-		-standby-cluster-cloud-provider=$(STANDBY_CLUSTER_CLOUD_PROVIDER)
+		-standby-cluster-cloud-provider=$(STANDBY_CLUSTER_CLOUD_PROVIDER) \
+		-standby-cluster-plugins=$(STANDBY_CLUSTER_PLUGINS) \
+		-standby-cluster-object-store-provider=$(STANDBY_CLUSTER_OBJECT_STORE_PROVIDER) \
+		-debug-velero-pod-restart=$(DEBUG_VELERO_POD_RESTART)

 build: ginkgo
 	mkdir -p $(OUTPUT_DIR)
--- a/test/e2e/backup/backup.go
+++ b/test/e2e/backup/backup.go
@ -66,7 +66,7 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 	AfterEach(func() {
 		if !veleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
+				DeleteAllBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
 			})
 			if veleroCfg.InstallVelero {
 				ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
--- a/test/e2e/backups/deletion.go
+++ b/test/e2e/backups/deletion.go
@ -69,7 +69,7 @@ func backup_deletion_test(useVolumeSnapshots bool) {
 	AfterEach(func() {
 		if !veleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
+				DeleteAllBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
 			})
 		}
 	})
--- a/test/e2e/backups/sync_backups.go
+++ b/test/e2e/backups/sync_backups.go
@ -66,7 +66,7 @@ func BackupsSyncTest() {
 	AfterEach(func() {
 		if !VeleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *VeleroCfg.ClientToInstallVelero)
+				DeleteAllBackups(context.Background(), *VeleroCfg.ClientToInstallVelero)
 			})
 			if VeleroCfg.InstallVelero {
 				ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
--- a/test/e2e/backups/ttl.go
+++ b/test/e2e/backups/ttl.go
@ -78,7 +78,7 @@ func TTLTest() {
 		veleroCfg.GCFrequency = ""
 		if !veleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
+				DeleteAllBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
 			})
 			ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
 			defer ctxCancel()
--- a/test/e2e/basic/api-group/enable_api_group_extentions.go
+++ b/test/e2e/basic/api-group/enable_api_group_extentions.go
@ -69,7 +69,7 @@ func APIExtensionsVersionsTest() {
 	AfterEach(func() {
 		if !veleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.DefaultClient)
+				DeleteAllBackups(context.Background(), *veleroCfg.DefaultClient)
 			})
 			if veleroCfg.InstallVelero {
 				By("Uninstall Velero and delete CRD ", func() {
--- a/test/e2e/basic/api-group/enable_api_group_versions.go
+++ b/test/e2e/basic/api-group/enable_api_group_versions.go
@ -94,7 +94,7 @@ func APIGropuVersionsTest() {
 			}

 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
+				DeleteAllBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
 			})
 			if veleroCfg.InstallVelero {
 				By("Uninstall Velero in api group version case", func() {
--- a/test/e2e/basic/pvc-selected-node-changing.go
+++ b/test/e2e/basic/pvc-selected-node-changing.go
@ -73,6 +73,14 @@ func (p *PVCSelectedNodeChanging) CreateResources() error {
 			fmt.Sprintf("Failed to create namespace %s", p.namespace))
 	})

+	By(fmt.Sprintf("Create a storage class %s.", StorageClassName), func() {
+		Expect(InstallStorageClass(context.Background(), fmt.Sprintf("../testdata/storage-class/%s.yaml", p.VeleroCfg.CloudProvider))).To(Succeed())
+	})
+
+	By(fmt.Sprintf("Create a storage class %s.", StorageClassName), func() {
+		Expect(InstallTestStorageClasses(fmt.Sprintf("../testdata/storage-class/%s.yaml", VeleroCfg.CloudProvider))).To(Succeed(), "Failed to install storage class")
+	})
+
 	By(fmt.Sprintf("Create pod %s in namespace %s", p.podName, p.namespace), func() {
 		nodeNameList, err := GetWorkerNodes(p.Ctx)
 		Expect(err).To(Succeed())
@ -80,7 +88,7 @@ func (p *PVCSelectedNodeChanging) CreateResources() error {
 			p.oldNodeName = nodeName
 			fmt.Printf("Create PVC on node %s\n", p.oldNodeName)
 			pvcAnn := map[string]string{p.ann: nodeName}
-			_, err := CreatePod(p.Client, p.namespace, p.podName, "default", p.pvcName, []string{p.volume}, pvcAnn, nil)
+			_, err := CreatePod(p.Client, p.namespace, p.podName, StorageClassName, p.pvcName, []string{p.volume}, pvcAnn, nil)
 			Expect(err).To(Succeed())
 			err = WaitForPods(p.Ctx, p.Client, p.namespace, []string{p.podName})
 			Expect(err).To(Succeed())
--- a/test/e2e/basic/storage-class-changing.go
+++ b/test/e2e/basic/storage-class-changing.go
@ -50,8 +50,8 @@ func (s *StorageClasssChanging) Init() error {
 		Text: "Change the storage class of persistent volumes and persistent" +
 			" volume claims during restores",
 	}
-	s.srcStorageClass = "default"
-	s.desStorageClass = StorageClassName
+	s.srcStorageClass = StorageClassName
+	s.desStorageClass = StorageClassName2
 	s.labels = map[string]string{"velero.io/change-storage-class": "RestoreItemAction",
 		"velero.io/plugin-config": ""}
 	s.data = map[string]string{s.srcStorageClass: s.desStorageClass}
@ -75,10 +75,11 @@ func (s *StorageClasssChanging) CreateResources() error {
 		"app": "test",
 	}
 	s.Ctx, s.CtxCancel = context.WithTimeout(context.Background(), 10*time.Minute)
-	By(fmt.Sprintf("Create a storage class %s", s.desStorageClass), func() {
-		Expect(InstallStorageClass(s.Ctx, fmt.Sprintf("../testdata/storage-class/%s.yaml",
-			s.VeleroCfg.CloudProvider))).To(Succeed())
+
+	By(("Installing storage class..."), func() {
+		Expect(InstallTestStorageClasses(fmt.Sprintf("../testdata/storage-class/%s.yaml", s.VeleroCfg.CloudProvider))).To(Succeed(), "Failed to install storage class")
 	})
+
 	By(fmt.Sprintf("Create namespace %s", s.namespace), func() {
 		Expect(CreateNamespace(s.Ctx, s.Client, s.namespace)).To(Succeed(),
 			fmt.Sprintf("Failed to create namespace %s", s.namespace))
--- a/test/e2e/bsl-mgmt/deletion.go
+++ b/test/e2e/bsl-mgmt/deletion.go
@ -77,7 +77,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 	AfterEach(func() {
 		if !veleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.DefaultClient)
+				DeleteAllBackups(context.Background(), *veleroCfg.DefaultClient)
 			})
 			By(fmt.Sprintf("Delete sample workload namespace %s", bslDeletionTestNs), func() {
 				Expect(DeleteNamespace(context.Background(), *veleroCfg.ClientToInstallVelero, bslDeletionTestNs,
--- a/test/e2e/e2e_suite_test.go
+++ b/test/e2e/e2e_suite_test.go
@ -85,10 +85,13 @@ func init() {
 	flag.StringVar(&VeleroCfg.DefaultCluster, "default-cluster", "", "Default cluster context for migration test.")
 	flag.StringVar(&VeleroCfg.StandbyCluster, "standby-cluster", "", "Standby cluster context for migration test.")
 	flag.StringVar(&VeleroCfg.UploaderType, "uploader-type", "", "Identify persistent volume backup uploader.")
-	flag.BoolVar(&VeleroCfg.VeleroServerDebugMode, "velero-server-debug-mode", false, "Identify persistent volume backup uploader.")
+	flag.BoolVar(&VeleroCfg.VeleroServerDebugMode, "velero-server-debug-mode", false, "Switch for enabling Velero server log debug level.")
 	flag.BoolVar(&VeleroCfg.SnapshotMoveData, "snapshot-move-data", false, "Install default plugin for data mover.")
 	flag.StringVar(&VeleroCfg.DataMoverPlugin, "data-mover-plugin", "", "Install customized plugin for data mover.")
-	flag.StringVar(&VeleroCfg.StandbyClusterCloudProvider, "standby-cluster-cloud-provider", "", "Install customized plugin for data mover.")
+	flag.StringVar(&VeleroCfg.StandbyClusterCloudProvider, "standby-cluster-cloud-provider", "", "Cloud provider for standby cluster.")
+	flag.StringVar(&VeleroCfg.StandbyClusterPlugins, "standby-cluster-plugins", "", "Plugins provider for standby cluster.")
+	flag.StringVar(&VeleroCfg.StandbyClusterOjbectStoreProvider, "standby-cluster-object-store-provider", "", "Object store provider for standby cluster.")
+	flag.BoolVar(&VeleroCfg.DebugVeleroPodRestart, "debug-velero-pod-restart", false, "Switch for debugging velero pod restart.")
 }

 var _ = Describe("[APIGroup][APIVersion] Velero tests with various CRD API group versions", APIGropuVersionsTest)
--- a/test/e2e/migration/migration.go
+++ b/test/e2e/migration/migration.go
@ -82,22 +82,26 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 	})
 	AfterEach(func() {
 		if !veleroCfg.Debug {
-			// TODO: delete backup created by case self, not all
-			// By("Clean backups after test", func() {
-			// 	DeleteBackups(context.Background(), *veleroCfg.DefaultClient)
-			// })
-			if veleroCfg.InstallVelero {
-				By(fmt.Sprintf("Uninstall Velero and delete sample workload namespace %s", migrationNamespace), func() {
-					ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
-					defer ctxCancel()
-					Expect(KubectlConfigUseContext(context.Background(), veleroCfg.DefaultCluster)).To(Succeed())
-					Expect(VeleroUninstall(ctx, veleroCfg.VeleroCLI,
-						veleroCfg.VeleroNamespace)).To(Succeed())
-					DeleteNamespace(context.Background(), *veleroCfg.DefaultClient, migrationNamespace, true)
+			By(fmt.Sprintf("Uninstall Velero on cluster %s", veleroCfg.DefaultCluster), func() {
+				ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
+				defer ctxCancel()
+				Expect(KubectlConfigUseContext(context.Background(), veleroCfg.DefaultCluster)).To(Succeed())
+				Expect(VeleroUninstall(ctx, veleroCfg.VeleroCLI,
+					veleroCfg.VeleroNamespace)).To(Succeed())
+				DeleteNamespace(context.Background(), *veleroCfg.DefaultClient, migrationNamespace, true)
+			})

-					Expect(KubectlConfigUseContext(context.Background(), veleroCfg.StandbyCluster)).To(Succeed())
-					Expect(VeleroUninstall(ctx, veleroCfg.VeleroCLI,
-						veleroCfg.VeleroNamespace)).To(Succeed())
+			By(fmt.Sprintf("Uninstall Velero on cluster %s", veleroCfg.StandbyCluster), func() {
+				ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
+				defer ctxCancel()
+				Expect(KubectlConfigUseContext(context.Background(), veleroCfg.StandbyCluster)).To(Succeed())
+				Expect(VeleroUninstall(ctx, veleroCfg.VeleroCLI,
+					veleroCfg.VeleroNamespace)).To(Succeed())
+				DeleteNamespace(context.Background(), *veleroCfg.StandbyClient, migrationNamespace, true)
+			})
+
+			if veleroCfg.InstallVelero {
+				By(fmt.Sprintf("Delete sample workload namespace %s", migrationNamespace), func() {
 					DeleteNamespace(context.Background(), *veleroCfg.StandbyClient, migrationNamespace, true)
 				})
 			}
@ -110,7 +114,7 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 	})
 	When("kibishii is the sample workload", func() {
 		It("should be successfully backed up and restored to the default BackupStorageLocation", func() {
-
+			var backupNames []string
 			if veleroCfg.SnapshotMoveData {
 				if !useVolumeSnapshots {
 					Skip("FSB migration test is not needed in data mover scenario")
@ -185,7 +189,7 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 			KibishiiData := *DefaultKibishiiData
 			By("Deploy sample workload of Kibishii", func() {
 				if OriginVeleroCfg.SnapshotMoveData {
-					KibishiiData.ExpectedNodes = 6
+					KibishiiData.ExpectedNodes = 3
 				}

 				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *veleroCfg.DefaultClient, veleroCfg.CloudProvider,
@ -206,6 +210,7 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 					RunDebug(context.Background(), veleroCfg.VeleroCLI, veleroCfg.VeleroNamespace, BackupStorageClassCfg.BackupName, "")
 					return "Fail to backup workload"
 				})
+				backupNames = append(backupNames, BackupStorageClassCfg.BackupName)

 				var BackupCfg BackupConfig
 				BackupCfg.BackupName = backupName
@ -223,6 +228,7 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 					RunDebug(context.Background(), OriginVeleroCfg.VeleroCLI, OriginVeleroCfg.VeleroNamespace, BackupCfg.BackupName, "")
 					return "Fail to backup workload"
 				})
+				backupNames = append(backupNames, BackupCfg.BackupName)
 			})

 			if useVolumeSnapshots {
@ -283,7 +289,12 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 				veleroCfg.UseRestic = false
 				if veleroCfg.SnapshotMoveData {
 					veleroCfg.UseNodeAgent = true
+					// For SnapshotMoveData pipelines, we should use standby clustr setting for Velero installation
+					// In nightly CI, StandbyClusterPlugins is set properly if pipeline is for SnapshotMoveData.
+					veleroCfg.Plugins = veleroCfg.StandbyClusterPlugins
+					veleroCfg.ObjectStoreProvider = veleroCfg.StandbyClusterOjbectStoreProvider
 				}
+
 				Expect(VeleroInstall(context.Background(), &veleroCfg, true)).To(Succeed())
 			})

@ -326,6 +337,11 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 				Expect(KibishiiVerifyAfterRestore(*veleroCfg.StandbyClient, migrationNamespace,
 					oneHourTimeout, &KibishiiData)).To(Succeed(), "Fail to verify workload after restore")
 			})
+
+			// TODO: delete backup created by case self, not all
+			By("Clean backups after test", func() {
+				DeleteBackups(context.Background(), *veleroCfg.DefaultClient, backupNames)
+			})
 		})
 	})
 }
--- a/test/e2e/resourcepolicies/resource_policies.go
+++ b/test/e2e/resourcepolicies/resource_policies.go
@ -19,7 +19,6 @@ package filtering
 import (
 	"context"
 	"fmt"
-	"os"
 	"strings"
 	"time"

@ -33,6 +32,7 @@ import (
 	. "github.com/vmware-tanzu/velero/test"
 	. "github.com/vmware-tanzu/velero/test/e2e/test"
 	. "github.com/vmware-tanzu/velero/test/util/k8s"
+	. "github.com/vmware-tanzu/velero/test/util/velero"
 )

 const FileName = "test-data.txt"
@ -110,7 +110,7 @@ func (r *ResourcePoliciesCase) CreateResources() error {
 	r.Ctx, r.CtxCancel = context.WithTimeout(context.Background(), 10*time.Minute)

 	By(("Installing storage class..."), func() {
-		Expect(r.installTestStorageClasses(fmt.Sprintf("../testdata/storage-class/%s.yaml", VeleroCfg.CloudProvider))).To(Succeed(), "Failed to install storage class")
+		Expect(InstallTestStorageClasses(fmt.Sprintf("../testdata/storage-class/%s.yaml", VeleroCfg.CloudProvider))).To(Succeed(), "Failed to install storage class")
 	})

 	By(fmt.Sprintf("Create configmap %s in namespaces %s for workload\n", r.cmName, r.VeleroCfg.VeleroNamespace), func() {
@ -188,7 +188,7 @@ func (r *ResourcePoliciesCase) Verify() error {
 func (r *ResourcePoliciesCase) Clean() error {
 	// If created some resources which is not in current test namespace, we NEED to override the base Clean function
 	if !r.VeleroCfg.Debug {
-		if err := r.deleteTestStorageClassList([]string{"e2e-storage-class", "e2e-storage-class-2"}); err != nil {
+		if err := r.deleteTestStorageClassList([]string{StorageClassName, StorageClassName2}); err != nil {
 			return err
 		}

@ -207,13 +207,13 @@ func (r *ResourcePoliciesCase) createPVC(index int, namespace string, volList []
 		pvcName := fmt.Sprintf("pvc-%d", i)
 		By(fmt.Sprintf("Creating PVC %s in namespaces ...%s\n", pvcName, namespace))
 		if index%3 == 0 {
-			pvcBuilder := NewPVC(namespace, pvcName).WithStorageClass("e2e-storage-class") // Testing sc should not backup
+			pvcBuilder := NewPVC(namespace, pvcName).WithStorageClass(StorageClassName) // Testing sc should not backup
 			err = CreatePvc(r.Client, pvcBuilder)
 		} else if index%3 == 1 {
-			pvcBuilder := NewPVC(namespace, pvcName).WithStorageClass("e2e-storage-class-2") // Testing sc should backup
+			pvcBuilder := NewPVC(namespace, pvcName).WithStorageClass(StorageClassName2) // Testing sc should backup
 			err = CreatePvc(r.Client, pvcBuilder)
 		} else if index%3 == 2 {
-			pvcBuilder := NewPVC(namespace, pvcName).WithStorageClass("e2e-storage-class-2").WithResourceStorage(resource.MustParse("2Gi")) // Testing capacity should not backup
+			pvcBuilder := NewPVC(namespace, pvcName).WithStorageClass(StorageClassName2).WithResourceStorage(resource.MustParse("2Gi")) // Testing capacity should not backup
 			err = CreatePvc(r.Client, pvcBuilder)
 		}
 		if err != nil {
@ -263,28 +263,3 @@ func (r *ResourcePoliciesCase) deleteTestStorageClassList(scList []string) error
 	}
 	return nil
 }
-
-func (r *ResourcePoliciesCase) installTestStorageClasses(path string) error {
-	err := InstallStorageClass(r.Ctx, path)
-	if err != nil {
-		return err
-	}
-	content, err := os.ReadFile(path)
-	if err != nil {
-		return errors.Wrapf(err, "failed to get %s when install storage class", path)
-	}
-
-	// replace sc to new value
-	newContent := strings.ReplaceAll(string(content), "name: e2e-storage-class", "name: e2e-storage-class-2")
-
-	tmpFile, err := os.CreateTemp("", "sc-file")
-	if err != nil {
-		return errors.Wrapf(err, "failed to create temp file  when install storage class")
-	}
-
-	defer os.Remove(tmpFile.Name())
-	if _, err := tmpFile.WriteString(newContent); err != nil {
-		return errors.Wrapf(err, "failed to write content into temp file %s when install storage class", tmpFile.Name())
-	}
-	return InstallStorageClass(r.Ctx, tmpFile.Name())
-}
--- a/test/e2e/schedule/ordered_resources.go
+++ b/test/e2e/schedule/ordered_resources.go
@ -166,7 +166,7 @@ func (o *OrderedResources) Clean() error {
 	return nil
 }

-func (o *OrderedResources) DeleteBackups() error {
+func (o *OrderedResources) DeleteAllBackups() error {
 	backupList := new(velerov1api.BackupList)
 	if err := o.Client.Kubebuilder.List(o.Ctx, backupList, &kbclient.ListOptions{Namespace: o.VeleroCfg.VeleroNamespace}); err != nil {
 		return fmt.Errorf("failed to list backup object in %s namespace with err %v", o.VeleroCfg.VeleroNamespace, err)
--- a/test/e2e/test/test.go
+++ b/test/e2e/test/test.go
@ -192,7 +192,7 @@ func (t *TestCase) Clean() error {
 			CleanupNamespaces(t.Ctx, t.Client, t.CaseBaseName)
 		})
 		By("Clean backups after test", func() {
-			DeleteBackups(t.Ctx, t.Client)
+			DeleteAllBackups(t.Ctx, t.Client)
 		})
 	}
 	return nil
--- a/test/e2e/upgrade/upgrade.go
+++ b/test/e2e/upgrade/upgrade.go
@ -87,7 +87,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, veleroCLI2Version VeleroC
 	AfterEach(func() {
 		if !veleroCfg.Debug {
 			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
+				DeleteAllBackups(context.Background(), *veleroCfg.ClientToInstallVelero)
 			})
 			By(fmt.Sprintf("Delete sample workload namespace %s", upgradeNamespace), func() {
 				DeleteNamespace(context.Background(), *veleroCfg.ClientToInstallVelero, upgradeNamespace, true)
@ -136,6 +136,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, veleroCLI2Version VeleroC
 				version, err := GetVeleroVersion(oneHourTimeout, tmpCfgForOldVeleroInstall.VeleroCLI, true)
 				Expect(err).To(Succeed(), "Fail to get Velero version")
 				tmpCfgForOldVeleroInstall.VeleroVersion = version
+				tmpCfgForOldVeleroInstall.UseVolumeSnapshots = useVolumeSnapshots

 				if supportUploaderType {
 					tmpCfgForOldVeleroInstall.UseRestic = false
--- a/test/types.go
+++ b/test/types.go
@ -25,6 +25,7 @@ import (
 )

 const StorageClassName = "e2e-storage-class"
+const StorageClassName2 = "e2e-storage-class-2"

 var UUIDgen uuid.UUID

@ -39,50 +40,53 @@ var ReportData *Report

 type VeleroConfig struct {
 	VeleroCfgInPerf
-	VeleroCLI                   string
-	VeleroImage                 string
-	VeleroVersion               string
-	CloudCredentialsFile        string
-	BSLConfig                   string
-	BSLBucket                   string
-	BSLPrefix                   string
-	VSLConfig                   string
-	CloudProvider               string
-	ObjectStoreProvider         string
-	VeleroNamespace             string
-	AdditionalBSLProvider       string
-	AdditionalBSLBucket         string
-	AdditionalBSLPrefix         string
-	AdditionalBSLConfig         string
-	AdditionalBSLCredentials    string
-	RegistryCredentialFile      string
-	RestoreHelperImage          string
-	UpgradeFromVeleroVersion    string
-	UpgradeFromVeleroCLI        string
-	MigrateFromVeleroVersion    string
-	MigrateFromVeleroCLI        string
-	Plugins                     string
-	AddBSLPlugins               string
-	InstallVelero               bool
-	KibishiiDirectory           string
-	Features                    string
-	Debug                       bool
-	GCFrequency                 string
-	DefaultCluster              string
-	StandbyCluster              string
-	ClientToInstallVelero       *TestClient
-	DefaultClient               *TestClient
-	StandbyClient               *TestClient
-	UploaderType                string
-	UseNodeAgent                bool
-	UseRestic                   bool
-	ProvideSnapshotsVolumeParam bool
-	DefaultVolumesToFsBackup    bool
-	UseVolumeSnapshots          bool
-	VeleroServerDebugMode       bool
-	SnapshotMoveData            bool
-	DataMoverPlugin             string
-	StandbyClusterCloudProvider string
+	VeleroCLI                         string
+	VeleroImage                       string
+	VeleroVersion                     string
+	CloudCredentialsFile              string
+	BSLConfig                         string
+	BSLBucket                         string
+	BSLPrefix                         string
+	VSLConfig                         string
+	CloudProvider                     string
+	ObjectStoreProvider               string
+	VeleroNamespace                   string
+	AdditionalBSLProvider             string
+	AdditionalBSLBucket               string
+	AdditionalBSLPrefix               string
+	AdditionalBSLConfig               string
+	AdditionalBSLCredentials          string
+	RegistryCredentialFile            string
+	RestoreHelperImage                string
+	UpgradeFromVeleroVersion          string
+	UpgradeFromVeleroCLI              string
+	MigrateFromVeleroVersion          string
+	MigrateFromVeleroCLI              string
+	Plugins                           string
+	AddBSLPlugins                     string
+	InstallVelero                     bool
+	KibishiiDirectory                 string
+	Features                          string
+	Debug                             bool
+	GCFrequency                       string
+	DefaultCluster                    string
+	StandbyCluster                    string
+	ClientToInstallVelero             *TestClient
+	DefaultClient                     *TestClient
+	StandbyClient                     *TestClient
+	UploaderType                      string
+	UseNodeAgent                      bool
+	UseRestic                         bool
+	ProvideSnapshotsVolumeParam       bool
+	DefaultVolumesToFsBackup          bool
+	UseVolumeSnapshots                bool
+	VeleroServerDebugMode             bool
+	SnapshotMoveData                  bool
+	DataMoverPlugin                   string
+	StandbyClusterCloudProvider       string
+	StandbyClusterPlugins             string
+	StandbyClusterOjbectStoreProvider string
+	DebugVeleroPodRestart             bool
 }

 type VeleroCfgInPerf struct {
--- a/test/util/common/common.go
+++ b/test/util/common/common.go
@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"os"
 	"os/exec"
 )

@ -87,3 +88,25 @@ func CMDExecWithOutput(checkCMD *exec.Cmd) (*[]byte, error) {
 	}
 	return &jsonBuf, err
 }
+
+func WriteToFile(content, fileName string) error {
+	file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE, 0666)
+	if err != nil {
+		fmt.Println("fail to open file", err)
+		return err
+	}
+	defer file.Close()
+
+	write := bufio.NewWriter(file)
+	_, err = write.WriteString(content)
+	if err != nil {
+		fmt.Println("fail to WriteString file", err)
+		return err
+	}
+	err = write.Flush()
+	if err != nil {
+		fmt.Println("fail to Flush file", err)
+		return err
+	}
+	return nil
+}
--- a/test/util/k8s/common.go
+++ b/test/util/k8s/common.go
@ -20,6 +20,7 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"strings"
 	"time"

 	"github.com/pkg/errors"
@ -30,7 +31,7 @@ import (

 	"github.com/vmware-tanzu/velero/pkg/builder"
 	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
-	"github.com/vmware-tanzu/velero/test/util/common"
+	common "github.com/vmware-tanzu/velero/test/util/common"
 )

 // ensureClusterExists returns whether or not a Kubernetes cluster exists for tests to be run on.
@ -289,15 +290,15 @@ func ReadFileFromPodVolume(ctx context.Context, namespace, podName, containerNam
 	return stdout, err
 }

-func KubectlGetInfo(cmdName string, arg []string) {
+func RunCommand(cmdName string, arg []string) string {
 	cmd := exec.CommandContext(context.Background(), cmdName, arg...)
-	fmt.Printf("Kubectl exec cmd =%v\n", cmd)
+	fmt.Printf("Run cmd =%v\n", cmd)
 	stdout, stderr, err := veleroexec.RunCommand(cmd)
-	fmt.Println(stdout)
 	if err != nil {
 		fmt.Println(stderr)
 		fmt.Println(err)
 	}
+	return stdout
 }

 func KubectlGetDsJson(veleroNamespace string) (string, error) {
@ -363,3 +364,26 @@ func CreateVolumes(pvcName string, volumeNameList []string) (vols []*corev1.Volu
 	}
 	return
 }
+
+func CollectClusterEvents(key string, pods []string) {
+	prefix := "pod"
+	date := RunCommand("date", []string{"-u"})
+	logs := []string{}
+	logs = append(logs, date)
+	logs = append(logs, RunCommand("kubectl", []string{"get", "events", "-o", "custom-columns=FirstSeen:.firstTimestamp,Count:.count,From:.source.component,Type:.type,Reason:.reason,Message:.message", "--all-namespaces"}))
+	logs = append(logs, RunCommand("kubectl", []string{"get", "events", "-o", "yaml", "--all-namespaces"}))
+	for _, pod := range pods {
+		logs = append(logs, RunCommand("kubectl", []string{"logs", "-n", "velero", pod, "--previous"}))
+		prefix = fmt.Sprintf("%s-%s", prefix, pod)
+	}
+	logs = append(logs, RunCommand("date", []string{"-u"}))
+	log := strings.Join(logs, "\n")
+
+	fileName := fmt.Sprintf("%s-%s", prefix, key)
+	fmt.Printf("Cluster event log file %s: %s", fileName, log)
+
+	err := common.WriteToFile(log, fileName)
+	if err != nil {
+		fmt.Printf("Fail to log cluster event: %v", err)
+	}
+}
--- a/test/util/kibishii/kibishii_utils.go
+++ b/test/util/kibishii/kibishii_utils.go
@ -105,6 +105,7 @@ func RunKibishiiTests(veleroCfg VeleroConfig, backupName, restoreName, backupLoc
 		RunDebug(context.Background(), veleroCLI, veleroNamespace, backupName, "")
 		return errors.Wrapf(err, "Failed to backup kibishii namespace %s", kibishiiNamespace)
 	}
+
 	fmt.Printf("VeleroBackupNamespace done %s\n", time.Now().Format("2006-01-02 15:04:05"))
 	if useVolumeSnapshots {
 		if providerName == "vsphere" {
@ -291,15 +292,16 @@ func verifyData(ctx context.Context, namespace string, kibishiiData *KibishiiDat
 			return false, nil
 		}
 		if err != nil {
-			fmt.Printf("Kibishi verify stdout Timeout occurred: %s stderr: %s err: %s", stdout, stderr, err)
+			fmt.Printf("Kibishi verify stdout Timeout occurred: %s stderr: %s err: %s\n", stdout, stderr, err)
 			return false, nil
 		}
 		return true, nil
 	})

 	if err != nil {
-		return errors.Wrapf(err, fmt.Sprintf("Failed to wait generate data in namespace %s", namespace))
+		return errors.Wrapf(err, fmt.Sprintf("Failed to verify kibishii data in namespace %s\n", namespace))
 	}
+	fmt.Printf("Success to verify kibishii data in namespace %s\n", namespace)
 	return nil
 }

--- a/test/util/providers/aws_utils.go
+++ b/test/util/providers/aws_utils.go
@ -18,15 +18,20 @@ package providers

 import (
 	"fmt"
+	"net/url"
+	"os"
+	"strconv"
 	"strings"

 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/endpoints"
 	"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"
+	"golang.org/x/net/context"

 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	"github.com/vmware-tanzu/velero/test"
@ -34,6 +39,173 @@ import (

 type AWSStorage string

+const (
+	s3URLKey                     = "s3Url"
+	publicURLKey                 = "publicUrl"
+	kmsKeyIDKey                  = "kmsKeyId"
+	customerKeyEncryptionFileKey = "customerKeyEncryptionFile"
+	s3ForcePathStyleKey          = "s3ForcePathStyle"
+	bucketKey                    = "bucket"
+	signatureVersionKey          = "signatureVersion"
+	credentialsFileKey           = "credentialsFile"
+	credentialProfileKey         = "profile"
+	serverSideEncryptionKey      = "serverSideEncryption"
+	insecureSkipTLSVerifyKey     = "insecureSkipTLSVerify"
+	caCertKey                    = "caCert"
+	enableSharedConfigKey        = "enableSharedConfig"
+)
+
+func newSessionOptions(config aws.Config, profile, caCert, credentialsFile, enableSharedConfig string) (session.Options, error) {
+	sessionOptions := session.Options{Config: config, Profile: profile}
+	if caCert != "" {
+		sessionOptions.CustomCABundle = strings.NewReader(caCert)
+	}
+
+	if credentialsFile == "" && os.Getenv("AWS_SHARED_CREDENTIALS_FILE") != "" {
+		credentialsFile = os.Getenv("AWS_SHARED_CREDENTIALS_FILE")
+	}
+
+	if credentialsFile != "" {
+		if _, err := os.Stat(credentialsFile); err != nil {
+			if os.IsNotExist(err) {
+				return session.Options{}, errors.Wrapf(err, "provided credentialsFile does not exist")
+			}
+			return session.Options{}, errors.Wrapf(err, "could not get credentialsFile info")
+		}
+		sessionOptions.SharedConfigFiles = append(sessionOptions.SharedConfigFiles, credentialsFile)
+		sessionOptions.SharedConfigState = session.SharedConfigEnable
+	}
+
+	return sessionOptions, nil
+}
+
+// takes AWS session options to create a new session
+func getSession(options session.Options) (*session.Session, error) {
+	sess, err := session.NewSessionWithOptions(options)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	if _, err := sess.Config.Credentials.Get(); err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return sess, nil
+}
+
+// GetBucketRegion returns the AWS region that a bucket is in, or an error
+// if the region cannot be determined.
+func GetBucketRegion(bucket string) (string, error) {
+	var region string
+
+	session, err := session.NewSession()
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+
+	for _, partition := range endpoints.DefaultPartitions() {
+		for regionHint := range partition.Regions() {
+			region, _ = s3manager.GetBucketRegion(context.Background(), session, bucket, regionHint)
+
+			// we only need to try a single region hint per partition, so break after the first
+			break
+		}
+
+		if region != "" {
+			return region, nil
+		}
+	}
+
+	return "", errors.New("unable to determine bucket's region")
+}
+
+func (s AWSStorage) CreateSession(credentialProfile, credentialsFile, enableSharedConfig, caCert, bucket, bslPrefix, bslConfig string) (*session.Session, error) {
+	var err error
+	config := flag.NewMap()
+	config.Set(bslConfig)
+	region := config.Data()["region"]
+	objectsInput := s3.ListObjectsV2Input{}
+	objectsInput.Bucket = aws.String(bucket)
+	objectsInput.Delimiter = aws.String("/")
+	s3URL := ""
+	s3ForcePathStyleVal := ""
+	s3ForcePathStyle := false
+
+	if s3ForcePathStyleVal != "" {
+		if s3ForcePathStyle, err = strconv.ParseBool(s3ForcePathStyleVal); err != nil {
+			return nil, errors.Wrapf(err, "could not parse %s (expected bool)", s3ForcePathStyleKey)
+		}
+	}
+
+	// AWS (not an alternate S3-compatible API) and region not
+	// explicitly specified: determine the bucket's region
+	if s3URL == "" && region == "" {
+		var err error
+
+		region, err = GetBucketRegion(bucket)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	serverConfig, err := newAWSConfig(s3URL, region, s3ForcePathStyle)
+	if err != nil {
+		return nil, err
+	}
+
+	sessionOptions, err := newSessionOptions(*serverConfig, credentialProfile, caCert, credentialsFile, enableSharedConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	serverSession, err := getSession(sessionOptions)
+
+	if err != nil {
+		fmt.Println(err)
+		return nil, err
+	}
+
+	return serverSession, nil
+}
+
+// IsValidS3URLScheme returns true if the scheme is http:// or https://
+// and the url parses correctly, otherwise, return false
+func IsValidS3URLScheme(s3URL string) bool {
+	u, err := url.Parse(s3URL)
+	if err != nil {
+		return false
+	}
+	if u.Scheme != "http" && u.Scheme != "https" {
+		return false
+	}
+	return true
+}
+func newAWSConfig(url, region string, forcePathStyle bool) (*aws.Config, error) {
+	awsConfig := aws.NewConfig().
+		WithRegion(region).
+		WithS3ForcePathStyle(forcePathStyle)
+
+	if url != "" {
+		if !IsValidS3URLScheme(url) {
+			return nil, errors.Errorf("Invalid s3 url %s, URL must be valid according to https://golang.org/pkg/net/url/#Parse and start with http:// or https://", url)
+		}
+
+		awsConfig = awsConfig.WithEndpointResolver(
+			endpoints.ResolverFunc(func(service, region string, optFns ...func(*endpoints.Options)) (endpoints.ResolvedEndpoint, error) {
+				if service == s3.EndpointsID {
+					return endpoints.ResolvedEndpoint{
+						URL: url,
+					}, nil
+				}
+
+				return endpoints.DefaultResolver().EndpointFor(service, region, optFns...)
+			}),
+		)
+	}
+
+	return awsConfig, nil
+}
+
 func (s AWSStorage) ListItems(client *s3.S3, objectsV2Input *s3.ListObjectsV2Input) (*s3.ListObjectsV2Output, error) {
 	res, err := client.ListObjectsV2(objectsV2Input)
 	if err != nil {
@ -54,15 +226,20 @@ func (s AWSStorage) DeleteItem(client *s3.S3, deleteObjectV2Input *s3.DeleteObje
 func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) (bool, error) {
 	config := flag.NewMap()
 	config.Set(bslConfig)
-	region := config.Data()["region"]
 	objectsInput := s3.ListObjectsV2Input{}
 	objectsInput.Bucket = aws.String(bslBucket)
 	objectsInput.Delimiter = aws.String("/")
-	s3url := ""
+
 	if bslPrefix != "" {
 		objectsInput.Prefix = aws.String(bslPrefix)
 	}
+
+	var err error
 	var s3Config *aws.Config
+	var sess *session.Session
+	region := config.Data()["region"]
+	s3url := ""
+
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
 		s3Config = &aws.Config{
@ -72,21 +249,19 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 			DisableSSL:       aws.Bool(true),
 			S3ForcePathStyle: aws.Bool(true),
 		}
+		sess, err = session.NewSession(s3Config)
 	} else {
-		s3Config = &aws.Config{
-			Region:      aws.String(region),
-			Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-		}
+		sess, err = s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
 	}
-	sess, err := session.NewSession(s3Config)

 	if err != nil {
-		return false, errors.Wrapf(err, "Failed to create AWS session")
+		return false, errors.Wrapf(err, fmt.Sprintf("Failed to create AWS session of region %s", region))
 	}
 	svc := s3.New(sess)

 	bucketObjects, err := s.ListItems(svc, &objectsInput)
 	if err != nil {
+		fmt.Println("Couldn't retrieve bucket items!")
 		return false, errors.Wrapf(err, "Couldn't retrieve bucket items")
 	}

@ -111,12 +286,17 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) error {
 	config := flag.NewMap()
 	config.Set(bslConfig)
+
+	var err error
+	var sess *session.Session
 	region := config.Data()["region"]
 	s3url := ""
+
 	s3Config := &aws.Config{
 		Region:      aws.String(region),
 		Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
 	}
+
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
 		s3Config = &aws.Config{
@ -126,12 +306,17 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 			DisableSSL:       aws.Bool(true),
 			S3ForcePathStyle: aws.Bool(true),
 		}
+		sess, err = session.NewSession(s3Config)
+	} else {
+		sess, err = s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
 	}
-	sess, err := session.NewSession(s3Config)
+
 	if err != nil {
-		return errors.Wrapf(err, "Error waiting for uploads to complete")
+		return errors.Wrapf(err, fmt.Sprintf("Failed to create AWS session of region %s", region))
 	}
+
 	svc := s3.New(sess)
+
 	fullPrefix := strings.Trim(bslPrefix, "/") + "/" + strings.Trim(backupObject, "/") + "/"
 	iter := s3manager.NewDeleteListIterator(svc, &s3.ListObjectsInput{
 		Bucket: aws.String(bslBucket),
@ -139,7 +324,7 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 	})

 	if err := s3manager.NewBatchDeleteWithClient(svc).Delete(aws.BackgroundContext(), iter); err != nil {
-		return errors.Wrapf(err, "Error waiting for uploads to complete")
+		return errors.Wrapf(err, "Fail to delete object")
 	}
 	fmt.Printf("Deleted object(s) from bucket: %s %s \n", bslBucket, fullPrefix)
 	return nil
@ -150,16 +335,15 @@ func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObj
 	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)
+	sess, err := s.CreateSession("", cloudCredentialsFile, "false", "", "", "", bslConfig)
+
 	if err != nil {
-		return errors.Wrapf(err, "Error waiting for uploads to complete")
+		fmt.Printf("Fail to create session with profile %s and config %s", cloudCredentialsFile, bslConfig)
+		return errors.Wrapf(err, "Fail to create session with profile %s and config %s", cloudCredentialsFile, bslConfig)
 	}
 	svc := ec2.New(sess)
 	params := &ec2.DescribeSnapshotsInput{
--- a/test/util/velero/install.go
+++ b/test/util/velero/install.go
@ -240,7 +240,7 @@ func installVeleroServer(ctx context.Context, cli, cloudProvider string, options
 		args = append(args, "--features", options.Features)
 		if strings.EqualFold(options.Features, "EnableCSI") && options.UseVolumeSnapshots {
 			if strings.EqualFold(cloudProvider, "azure") {
-				if err := KubectlApplyByFile(ctx, "util/csi/AzureVolumeSnapshotClass.yaml"); err != nil {
+				if err := KubectlApplyByFile(ctx, "../util/csi/AzureVolumeSnapshotClass.yaml"); err != nil {
 					return err
 				}
 			}
--- a/test/util/velero/velero_utils.go
+++ b/test/util/velero/velero_utils.go
@ -217,8 +217,14 @@ func checkBackupPhase(ctx context.Context, veleroCLI string, veleroNamespace str
 		return err
 	}
 	if backup.Status.Phase != expectedPhase {
+		if VeleroCfg.DebugVeleroPodRestart {
+			pods, err := GetVeleroPodName(ctx)
+			fmt.Println(err)
+			CollectClusterEvents(backupName, pods)
+		}
 		return errors.Errorf("Unexpected backup phase got %s, expecting %s", backup.Status.Phase, expectedPhase)
 	}
+
 	return nil
 }

@ -239,6 +245,11 @@ func checkRestorePhase(ctx context.Context, veleroCLI string, veleroNamespace st
 		return err
 	}
 	if restore.Status.Phase != expectedPhase {
+		if VeleroCfg.DebugVeleroPodRestart {
+			pods, err := GetVeleroPodName(ctx)
+			fmt.Println(err)
+			CollectClusterEvents(restoreName, pods)
+		}
 		return errors.Errorf("Unexpected restore phase got %s, expecting %s", restore.Status.Phase, expectedPhase)
 	}
 	return nil
@ -442,7 +453,6 @@ func VeleroRestoreExec(ctx context.Context, veleroCLI, veleroNamespace, restoreN
 	if err := VeleroCmdExec(ctx, veleroCLI, args); err != nil {
 		return err
 	}
-
 	return checkRestorePhase(ctx, veleroCLI, veleroNamespace, restoreName, phaseExpect)
 }

@ -1241,7 +1251,7 @@ func GetVersionList(veleroCli, veleroVersion string) []VeleroCLI2Version {
 	}
 	return veleroCLI2VersionList
 }
-func DeleteBackups(ctx context.Context, client TestClient) error {
+func DeleteAllBackups(ctx context.Context, client TestClient) error {
 	backupList := new(velerov1api.BackupList)
 	if err := client.Kubebuilder.List(ctx, backupList, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
 		return fmt.Errorf("failed to list backup object in %s namespace with err %v", VeleroCfg.VeleroNamespace, err)
@ -1255,6 +1265,24 @@ func DeleteBackups(ctx context.Context, client TestClient) error {
 	return nil
 }

+func DeleteBackups(ctx context.Context, client TestClient, backupNames []string) error {
+	backupList := new(velerov1api.BackupList)
+	if err := client.Kubebuilder.List(ctx, backupList, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
+		return fmt.Errorf("failed to list backup object in %s namespace with err %v", VeleroCfg.VeleroNamespace, err)
+	}
+	for _, backup := range backupList.Items {
+		for _, bn := range backupNames {
+			if backup.Name == bn {
+				fmt.Printf("Backup %s is going to be deleted...\n", backup.Name)
+				if err := VeleroBackupDelete(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, backup.Name); err != nil {
+					return err
+				}
+			}
+		}
+	}
+	return nil
+}
+
 func DeleteRestores(ctx context.Context, client TestClient) error {
 	restoreList := new(velerov1api.RestoreList)
 	if err := client.Kubebuilder.List(ctx, restoreList, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
@ -1482,3 +1510,56 @@ func IsSupportUploaderType(version string) (bool, error) {
 		return false, nil
 	}
 }
+
+func GetVeleroPodName(ctx context.Context) ([]string, error) {
+	// Example:
+	//    NAME                                  STATUS   VOLUME                                     CAPACITY   ACCESS MODES   STORAGECLASS             AGE
+	//    kibishii-data-kibishii-deployment-0   Bound    pvc-94b9fdf2-c30f-4a7b-87bf-06eadca0d5b6   1Gi        RWO            kibishii-storage-class   115s
+	cmds := []*common.OsCommandLine{}
+	cmd := &common.OsCommandLine{
+		Cmd:  "kubectl",
+		Args: []string{"get", "pod", "-n", "velero"},
+	}
+	cmds = append(cmds, cmd)
+
+	cmd = &common.OsCommandLine{
+		Cmd:  "grep",
+		Args: []string{"velero"},
+	}
+	cmds = append(cmds, cmd)
+
+	cmd = &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+	cmds = append(cmds, cmd)
+
+	return common.GetListByCmdPipes(ctx, cmds)
+}
+
+func InstallTestStorageClasses(path string) error {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), time.Minute*5)
+	defer ctxCancel()
+	err := InstallStorageClass(ctx, path)
+	if err != nil {
+		return err
+	}
+	content, err := os.ReadFile(path)
+	if err != nil {
+		return errors.Wrapf(err, "failed to get %s when install storage class", path)
+	}
+
+	// replace sc to new value
+	newContent := strings.ReplaceAll(string(content), fmt.Sprintf("name: %s", StorageClassName), fmt.Sprintf("name: %s", StorageClassName2))
+
+	tmpFile, err := os.CreateTemp("", "sc-file")
+	if err != nil {
+		return errors.Wrapf(err, "failed to create temp file  when install storage class")
+	}
+
+	defer os.Remove(tmpFile.Name())
+	if _, err := tmpFile.WriteString(newContent); err != nil {
+		return errors.Wrapf(err, "failed to write content into temp file %s when install storage class", tmpFile.Name())
+	}
+	return InstallStorageClass(ctx, tmpFile.Name())
+}