Install plugin for datamover pipeline

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

test/e2e/Makefile

@@ -73,6 +73,7 @@ BSL_PREFIX ?=
 BSL_CONFIG ?=
 VSL_CONFIG ?=
 CLOUD_PROVIDER ?=
+STANDBY_CLUSTER_CLOUD_PROVIDER ?=
 OBJECT_STORE_PROVIDER ?=
 INSTALL_VELERO ?= true
 REGISTRY_CREDENTIAL_FILE ?=
@@ -99,6 +100,9 @@ STANDBY_CLUSTER ?=
 
 UPLOADER_TYPE ?=
 
+SNAPSHOT_MOVE_DATA ?= false
+DATA_MOVER_PLUGIN ?=
+
 
 .PHONY:ginkgo
 ginkgo: # Make sure ginkgo is in $GOPATH/bin
@@ -143,7 +147,10 @@ run: ginkgo
 		-velero-server-debug-mode=$(VELERO_SERVER_DEBUG_MODE) \
 		-default-cluster=$(DEFAULT_CLUSTER) \
 		-standby-cluster=$(STANDBY_CLUSTER) \
-		-uploader-type=$(UPLOADER_TYPE)
+		-uploader-type=$(UPLOADER_TYPE) \
+		-snapshot-move-data=$(SNAPSHOT_MOVE_DATA) \
+		-data-mover-plugin=$(DATA_MOVER_plugin) \
+		-standby-cluster-cloud-provider=$(STANDBY_CLUSTER_CLOUD_PROVIDER)
 
 build: ginkgo
 	mkdir -p $(OUTPUT_DIR)

test/e2e/backup/backup.go

@@ -87,7 +87,7 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 				} else {
 					veleroCfg.DefaultVolumesToFsBackup = !useVolumeSnapshots
 				}
-				Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
+				Expect(VeleroInstall(context.Background(), &veleroCfg, false)).To(Succeed())
 			}
 			backupName = "backup-" + UUIDgen.String()
 			restoreName = "restore-" + UUIDgen.String()
@@ -125,12 +125,9 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 					veleroCfg.DefaultVolumesToFsBackup = useVolumeSnapshots
 				}
 
-				Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
+				Expect(VeleroInstall(context.Background(), &veleroCfg, false)).To(Succeed())
 			}
-
-			Expect(VeleroAddPluginsForProvider(context.TODO(), veleroCfg.VeleroCLI,
-				veleroCfg.VeleroNamespace, veleroCfg.AdditionalBSLProvider,
-				veleroCfg.AddBSLPlugins, veleroCfg.Features)).To(Succeed())
+			Expect(VeleroAddPluginsForProvider(context.TODO(), veleroCfg)).To(Succeed())
 
 			// Create Secret for additional BSL
 			secretName := fmt.Sprintf("bsl-credentials-%s", UUIDgen)

test/e2e/backups/sync_backups.go

@@ -59,7 +59,7 @@ func BackupsSyncTest() {
 		if VeleroCfg.InstallVelero {
 			veleroCfg := VeleroCfg
 			veleroCfg.UseVolumeSnapshots = false
-			Expect(VeleroInstall(context.Background(), &VeleroCfg)).To(Succeed())
+			Expect(VeleroInstall(context.Background(), &VeleroCfg, false)).To(Succeed())
 		}
 	})
 
@@ -109,7 +109,7 @@ func BackupsSyncTest() {
 		By("Install velero", func() {
 			veleroCfg := VeleroCfg
 			veleroCfg.UseVolumeSnapshots = false
-			Expect(VeleroInstall(ctx, &VeleroCfg)).To(Succeed())
+			Expect(VeleroInstall(ctx, &VeleroCfg, false)).To(Succeed())
 		})
 
 		By("Check all backups in object storage are synced to Velero", func() {

test/e2e/backups/ttl.go

@@ -70,7 +70,7 @@ func TTLTest() {
 			// Make sure GCFrequency is shorter than backup TTL
 			veleroCfg.GCFrequency = "4m0s"
 			veleroCfg.UseVolumeSnapshots = useVolumeSnapshots
-			Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
+			Expect(VeleroInstall(context.Background(), &veleroCfg, false)).To(Succeed())
 		}
 	})
 

test/e2e/basic/api-group/enable_api_group_extentions.go

@@ -100,7 +100,7 @@ func APIExtensionsVersionsTest() {
 				Expect(KubectlConfigUseContext(context.Background(), veleroCfg.DefaultCluster)).To(Succeed())
 				veleroCfg.Features = "EnableAPIGroupVersions"
 				veleroCfg.UseVolumeSnapshots = false
-				Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
+				Expect(VeleroInstall(context.Background(), &veleroCfg, false)).To(Succeed())
 			})
 
 			By(fmt.Sprintf("Install CRD of apiextenstions v1beta1 in cluster-A (%s)", veleroCfg.DefaultCluster), func() {
@@ -129,7 +129,7 @@ func APIExtensionsVersionsTest() {
 			By(fmt.Sprintf("Install Velero in cluster-B (%s) to restore workload", veleroCfg.StandbyCluster), func() {
 				Expect(KubectlConfigUseContext(context.Background(), veleroCfg.StandbyCluster)).To(Succeed())
 				veleroCfg.ClientToInstallVelero = veleroCfg.StandbyClient
-				Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
+				Expect(VeleroInstall(context.Background(), &veleroCfg, false)).To(Succeed())
 			})
 
 			By(fmt.Sprintf("Waiting for backups sync to Velero in cluster-B (%s)", veleroCfg.StandbyCluster), func() {

test/e2e/basic/api-group/enable_api_group_versions.go

@@ -75,7 +75,7 @@ func APIGropuVersionsTest() {
 		if veleroCfg.InstallVelero {
 			veleroCfg.Features = "EnableAPIGroupVersions"
 			veleroCfg.UseVolumeSnapshots = false
-			err = VeleroInstall(context.Background(), &veleroCfg)
+			err = VeleroInstall(context.Background(), &veleroCfg, false)
 			Expect(err).NotTo(HaveOccurred())
 		}
 		testCaseNum = 4

test/e2e/bsl-mgmt/deletion.go

@@ -104,9 +104,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 			}
 
 			By(fmt.Sprintf("Add an additional plugin for provider %s", veleroCfg.AdditionalBSLProvider), func() {
-				Expect(VeleroAddPluginsForProvider(context.TODO(), veleroCfg.VeleroCLI,
-					veleroCfg.VeleroNamespace, veleroCfg.AdditionalBSLProvider,
-					veleroCfg.AddBSLPlugins, veleroCfg.Features)).To(Succeed())
+				Expect(VeleroAddPluginsForProvider(context.TODO(), veleroCfg)).To(Succeed())
 			})
 
 			additionalBsl := fmt.Sprintf("bsl-%s", UUIDgen)

test/e2e/e2e_suite_test.go

@@ -84,7 +84,9 @@ func init() {
 	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.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.")
 }
 
 var _ = Describe("[APIGroup][APIVersion] Velero tests with various CRD API group versions", APIGropuVersionsTest)

test/e2e/migration/migration.go

@@ -59,13 +59,14 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 	BeforeEach(func() {
 		veleroCfg = VeleroCfg
 		UUIDgen, err = uuid.NewRandom()
-		migrationNamespace = "migration-workload-" + UUIDgen.String()
+		migrationNamespace = "migr" + UUIDgen.String()
 		if useVolumeSnapshots && veleroCfg.CloudProvider == "kind" {
 			Skip("Volume snapshots not supported on kind")
 		}
-		if useVolumeSnapshots && veleroCfg.CloudProvider == "aws" {
+		if useVolumeSnapshots && veleroCfg.CloudProvider == "aws" && !veleroCfg.SnapshotMoveData {
 			Skip("Volume snapshots migration not supported on AWS provisioned by Sheperd public pool")
 		}
+
 		if veleroCfg.DefaultCluster == "" && veleroCfg.StandbyCluster == "" {
 			Skip("Migration test needs 2 clusters")
 		}
@@ -79,9 +80,10 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 	})
 	AfterEach(func() {
 		if !veleroCfg.Debug {
-			By("Clean backups after test", func() {
-				DeleteBackups(context.Background(), *veleroCfg.DefaultClient)
-			})
+			// 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() {
 					Expect(KubectlConfigUseContext(context.Background(), veleroCfg.DefaultCluster)).To(Succeed())
@@ -104,6 +106,16 @@ 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() {
+
+			if veleroCfg.SnapshotMoveData {
+				if !useVolumeSnapshots {
+					Skip("FSB migration test is not needed in data mover scenario")
+				}
+				// TODO: remove this block once Velero version in cluster A is great than V1.11 for all migration path.
+				if veleroCLI2Version.VeleroVersion != "self" {
+					Skip(fmt.Sprintf("Only V1.12 support data mover scenario instead of %s", veleroCLI2Version.VeleroVersion))
+				}
+			}
 			oneHourTimeout, ctxCancel := context.WithTimeout(context.Background(), time.Minute*60)
 			defer ctxCancel()
 			flag.Parse()
@@ -132,25 +144,10 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 				OriginVeleroCfg.ClientToInstallVelero = OriginVeleroCfg.DefaultClient
 				OriginVeleroCfg.UseVolumeSnapshots = useVolumeSnapshots
 				OriginVeleroCfg.UseNodeAgent = !useVolumeSnapshots
-				// TODO: self means 1.10 and upper version
-				if veleroCLI2Version.VeleroVersion != "self" {
-					Expect(err).To(Succeed())
-					fmt.Printf("Using default images address of Velero CLI %s\n", veleroCLI2Version.VeleroVersion)
-					OriginVeleroCfg.VeleroImage = ""
-					OriginVeleroCfg.RestoreHelperImage = ""
-					OriginVeleroCfg.Plugins = ""
-					//TODO: Remove this once origin Velero version is 1.10 and upper
-					OriginVeleroCfg.UploaderType = ""
-					if supportUploaderType {
-						OriginVeleroCfg.UseRestic = false
-						OriginVeleroCfg.UseNodeAgent = !useVolumeSnapshots
-					} else {
-						OriginVeleroCfg.UseRestic = !useVolumeSnapshots
-						OriginVeleroCfg.UseNodeAgent = false
-					}
+				if OriginVeleroCfg.SnapshotMoveData {
+					OriginVeleroCfg.UseNodeAgent = true
 				}
-
-				Expect(VeleroInstall(context.Background(), &OriginVeleroCfg)).To(Succeed())
+				Expect(VeleroInstall(context.Background(), &OriginVeleroCfg, false)).To(Succeed())
 				if veleroCLI2Version.VeleroVersion != "self" {
 					Expect(CheckVeleroVersion(context.Background(), OriginVeleroCfg.VeleroCLI,
 						OriginVeleroCfg.MigrateFromVeleroVersion)).To(Succeed())
@@ -168,16 +165,23 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 			})
 
 			By("Deploy sample workload of Kibishii", func() {
+				var replicas int = DefaultKibishiiWorkerCounts
+				if OriginVeleroCfg.SnapshotMoveData {
+					replicas = 6
+				}
+				KibishiiData := *DefaultKibishiiData
+				KibishiiData.ExpectedNodes = int(replicas)
 				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *veleroCfg.DefaultClient, veleroCfg.CloudProvider,
 					migrationNamespace, veleroCfg.RegistryCredentialFile, veleroCfg.Features,
-					veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData)).To(Succeed())
+					veleroCfg.KibishiiDirectory, useVolumeSnapshots, &KibishiiData)).To(Succeed())
 			})
 
 			By(fmt.Sprintf("Backup namespace %s", migrationNamespace), func() {
 				var BackupStorageClassCfg BackupConfig
 				BackupStorageClassCfg.BackupName = backupScName
 				BackupStorageClassCfg.IncludeResources = "StorageClass"
 				BackupStorageClassCfg.IncludeClusterResources = true
+
 				//TODO Remove UseRestic parameter once minor version is 1.10 or upper
 				BackupStorageClassCfg.UseResticIfFSBackup = !supportUploaderType
 				Expect(VeleroBackupNamespace(context.Background(), OriginVeleroCfg.VeleroCLI,
@@ -195,6 +199,7 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 				BackupCfg.DefaultVolumesToFsBackup = !useVolumeSnapshots
 				//TODO Remove UseRestic parameter once minor version is 1.10 or upper
 				BackupCfg.UseResticIfFSBackup = !supportUploaderType
+				BackupCfg.SnapshotMoveData = OriginVeleroCfg.SnapshotMoveData
 
 				Expect(VeleroBackupNamespace(context.Background(), OriginVeleroCfg.VeleroCLI,
 					OriginVeleroCfg.VeleroNamespace, BackupCfg)).To(Succeed(), func() string {
@@ -211,16 +216,22 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 							migrationNamespace, 2)).To(Succeed())
 					})
 				}
+
 				var snapshotCheckPoint SnapshotCheckPoint
 				snapshotCheckPoint.NamespaceBackedUp = migrationNamespace
-				By("Snapshot should be created in cloud object store", func() {
-					snapshotCheckPoint, err := GetSnapshotCheckPoint(*veleroCfg.DefaultClient, veleroCfg, 2,
-						migrationNamespace, backupName, KibishiiPVCNameList)
-					Expect(err).NotTo(HaveOccurred(), "Fail to get snapshot checkpoint")
-					Expect(SnapshotsShouldBeCreatedInCloud(veleroCfg.CloudProvider,
-						veleroCfg.CloudCredentialsFile, veleroCfg.BSLBucket,
-						veleroCfg.BSLConfig, backupName, snapshotCheckPoint)).To(Succeed())
-				})
+
+				if !OriginVeleroCfg.SnapshotMoveData {
+					By("Snapshot should be created in cloud object store", func() {
+						snapshotCheckPoint, err := GetSnapshotCheckPoint(*veleroCfg.DefaultClient, veleroCfg, 2,
+							migrationNamespace, backupName, KibishiiPVCNameList)
+						Expect(err).NotTo(HaveOccurred(), "Fail to get snapshot checkpoint")
+						Expect(SnapshotsShouldBeCreatedInCloud(veleroCfg.CloudProvider,
+							veleroCfg.CloudCredentialsFile, veleroCfg.BSLBucket,
+							veleroCfg.BSLConfig, backupName, snapshotCheckPoint)).To(Succeed())
+					})
+				} else {
+					//TODO: checkpoint for datamover
+				}
 			}
 
 			if useVolumeSnapshots && veleroCfg.CloudProvider == "azure" && strings.EqualFold(veleroCfg.Features, "EnableCSI") {
@@ -253,7 +264,10 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 				veleroCfg.ClientToInstallVelero = veleroCfg.StandbyClient
 				veleroCfg.UseNodeAgent = !useVolumeSnapshots
 				veleroCfg.UseRestic = false
-				Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
+				if veleroCfg.SnapshotMoveData {
+					veleroCfg.UseNodeAgent = true
+				}
+				Expect(VeleroInstall(context.Background(), &veleroCfg, true)).To(Succeed())
 			})
 
 			By(fmt.Sprintf("Waiting for backups sync to Velero in cluster-B (%s)", veleroCfg.StandbyCluster), func() {

test/e2e/types.go

@@ -72,6 +72,9 @@ type VeleroConfig struct {
 	DefaultVolumesToFsBackup    bool
 	UseVolumeSnapshots          bool
 	VeleroServerDebugMode       bool
+	SnapshotMoveData            bool
+	DataMoverPlugin             string
+	StandbyClusterCloudProvider string
 }
 
 type SnapshotCheckPoint struct {
@@ -98,6 +101,7 @@ type BackupConfig struct {
 	OrderedResources            string
 	UseResticIfFSBackup         bool
 	DefaultVolumesToFsBackup    bool
+	SnapshotMoveData            bool
 }
 
 type VeleroCLI2Version struct {

test/e2e/upgrade/upgrade.go

@@ -136,7 +136,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, veleroCLI2Version VeleroC
 					tmpCfgForOldVeleroInstall.UseNodeAgent = false
 				}
 
-				Expect(VeleroInstall(context.Background(), &tmpCfgForOldVeleroInstall)).To(Succeed())
+				Expect(VeleroInstall(context.Background(), &tmpCfgForOldVeleroInstall, false)).To(Succeed())
 				Expect(CheckVeleroVersion(context.Background(), tmpCfgForOldVeleroInstall.VeleroCLI,
 					tmpCfgForOldVeleroInstall.UpgradeFromVeleroVersion)).To(Succeed())
 			})
@@ -223,7 +223,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, veleroCLI2Version VeleroC
 				tmpCfg.UseNodeAgent = !useVolumeSnapshots
 				Expect(err).To(Succeed())
 				if supportUploaderType {
-					Expect(VeleroInstall(context.Background(), &tmpCfg)).To(Succeed())
+					Expect(VeleroInstall(context.Background(), &tmpCfg, false)).To(Succeed())
 					Expect(CheckVeleroVersion(context.Background(), tmpCfg.VeleroCLI,
 						tmpCfg.VeleroVersion)).To(Succeed())
 				} else {

test/e2e/util/k8s/common.go

@@ -200,6 +200,7 @@ func AddLabelToCRD(ctx context.Context, crd, label string) error {
 
 func KubectlApplyByFile(ctx context.Context, file string) error {
 	args := []string{"apply", "-f", file, "--force=true"}
+	fmt.Println(args)
 	return exec.CommandContext(ctx, "kubectl", args...).Run()
 }
 

test/e2e/util/k8s/statefulset.go

@@ -0,0 +1,39 @@
+/*
+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 (
+	"fmt"
+	"os/exec"
+
+	"github.com/pkg/errors"
+	"golang.org/x/net/context"
+
+	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
+)
+
+func ScaleStatefulSet(ctx context.Context, namespace, name string, replicas int) error {
+	cmd := exec.CommandContext(ctx, "kubectl", "scale", "statefulsets", name, fmt.Sprintf("--replicas=%d", replicas), "-n", namespace)
+	fmt.Printf("Scale kibishii stateful set in namespace %s with CMD: %s", name, cmd.Args)
+
+	_, stderr, err := veleroexec.RunCommand(cmd)
+	if err != nil {
+		return errors.Wrap(err, stderr)
+	}
+
+	return nil
+}