Install plugin for datamover pipeline

Signed-off-by: danfengl <[email protected]>

test/e2e/Makefile

@@ -99,6 +99,9 @@ STANDBY_CLUSTER ?=
 
 UPLOADER_TYPE ?=
 
+SNAPSHOT_MOVE_DATA ?= false
+DATA_MOVER_PLUGIN ?=
+
 
 .PHONY:ginkgo
 ginkgo: # Make sure ginkgo is in $GOPATH/bin
@@ -143,7 +146,9 @@ 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)
 
 build: ginkgo
 	mkdir -p $(OUTPUT_DIR)

test/e2e/backup/backup.go

@@ -127,10 +127,7 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 
 				Expect(VeleroInstall(context.Background(), &veleroCfg)).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/deletion.go

@@ -108,7 +108,7 @@ func runBackupDeletionTests(client TestClient, veleroCfg VeleroConfig, backupNam
 	}
 
 	if err := KibishiiPrepareBeforeBackup(oneHourTimeout, client, providerName, deletionTest,
-		registryCredentialFile, veleroFeatures, kibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData); err != nil {
+		registryCredentialFile, veleroFeatures, kibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData, DefaultKibishiiWorkerCounts); err != nil {
 		return errors.Wrapf(err, "Failed to install and prepare data for kibishii %s", deletionTest)
 	}
 	err := ObjectsShouldNotBeInBucket(veleroCfg.CloudProvider, veleroCfg.CloudCredentialsFile, veleroCfg.BSLBucket, veleroCfg.BSLPrefix, veleroCfg.BSLConfig, backupName, BackupObjectsPrefix, 1)

test/e2e/backups/ttl.go

@@ -101,7 +101,7 @@ func TTLTest() {
 		By("Deploy sample workload of Kibishii", func() {
 			Expect(KibishiiPrepareBeforeBackup(ctx, client, veleroCfg.CloudProvider,
 				test.testNS, veleroCfg.RegistryCredentialFile, veleroCfg.Features,
-				veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData)).To(Succeed())
+				veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData, DefaultKibishiiWorkerCounts)).To(Succeed())
 		})
 
 		var BackupCfg BackupConfig

test/e2e/basic/namespace-mapping.go

@@ -91,7 +91,7 @@ func (n *NamespaceMapping) CreateResources() error {
 		By("Deploy sample workload of Kibishii", func() {
 			Expect(KibishiiPrepareBeforeBackup(n.Ctx, n.Client, VeleroCfg.CloudProvider,
 				ns, VeleroCfg.RegistryCredentialFile, VeleroCfg.Features,
-				VeleroCfg.KibishiiDirectory, false, n.kibishiiData)).To(Succeed())
+				VeleroCfg.KibishiiDirectory, false, n.kibishiiData, DefaultKibishiiWorkerCounts)).To(Succeed())
 		})
 	}
 	return nil

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)
@@ -152,7 +150,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 			By("Deploy sample workload of Kibishii", func() {
 				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *veleroCfg.ClientToInstallVelero, veleroCfg.CloudProvider,
 					bslDeletionTestNs, veleroCfg.RegistryCredentialFile, veleroCfg.Features,
-					veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData)).To(Succeed())
+					veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData, DefaultKibishiiWorkerCounts)).To(Succeed())
 			})
 
 			// Restic can not backup PV only, so pod need to be labeled also

test/e2e/e2e_suite_test.go

@@ -84,6 +84,8 @@ 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.")
 
 }
 

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")
 		}
@@ -104,6 +105,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,24 +143,9 @@ 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())
 				if veleroCLI2Version.VeleroVersion != "self" {
 					Expect(CheckVeleroVersion(context.Background(), OriginVeleroCfg.VeleroCLI,
@@ -168,16 +164,21 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 			})
 
 			By("Deploy sample workload of Kibishii", func() {
+				var replicas int32 = DefaultKibishiiWorkerCounts
+				if OriginVeleroCfg.SnapshotMoveData {
+					replicas = 6
+				}
 				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *veleroCfg.DefaultClient, veleroCfg.CloudProvider,
 					migrationNamespace, veleroCfg.RegistryCredentialFile, veleroCfg.Features,
-					veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData)).To(Succeed())
+					veleroCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData, replicas)).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 +196,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 +213,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,6 +261,9 @@ func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version)
 				veleroCfg.ClientToInstallVelero = veleroCfg.StandbyClient
 				veleroCfg.UseNodeAgent = !useVolumeSnapshots
 				veleroCfg.UseRestic = false
+				if veleroCfg.SnapshotMoveData {
+					veleroCfg.UseNodeAgent = true
+				}
 				Expect(VeleroInstall(context.Background(), &veleroCfg)).To(Succeed())
 			})
 

test/e2e/types.go

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

test/e2e/upgrade/upgrade.go

@@ -155,7 +155,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, veleroCLI2Version VeleroC
 			By("Deploy sample workload of Kibishii", func() {
 				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *veleroCfg.ClientToInstallVelero, tmpCfg.CloudProvider,
 					upgradeNamespace, tmpCfg.RegistryCredentialFile, tmpCfg.Features,
-					tmpCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData)).To(Succeed())
+					tmpCfg.KibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData, DefaultKibishiiWorkerCounts)).To(Succeed())
 			})
 
 			By(fmt.Sprintf("Backup namespace %s", upgradeNamespace), func() {

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 int32) 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
+}

test/e2e/util/kibishii/kibishii_utils.go

@@ -49,6 +49,7 @@ type KibishiiData struct {
 }
 
 var DefaultKibishiiData = &KibishiiData{2, 10, 10, 1024, 1024, 0, 2}
+var DefaultKibishiiWorkerCounts int32 = 2
 var KibishiiPVCNameList = []string{"kibishii-data-kibishii-deployment-0", "kibishii-data-kibishii-deployment-1"}
 
 // RunKibishiiTests runs kibishii tests on the provider.
@@ -83,7 +84,7 @@ func RunKibishiiTests(veleroCfg VeleroConfig, backupName, restoreName, backupLoc
 	fmt.Printf("KibishiiPrepareBeforeBackup %s\n", time.Now().Format("2006-01-02 15:04:05"))
 	if err := KibishiiPrepareBeforeBackup(oneHourTimeout, client, providerName,
 		kibishiiNamespace, registryCredentialFile, veleroFeatures,
-		kibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData); err != nil {
+		kibishiiDirectory, useVolumeSnapshots, DefaultKibishiiData, DefaultKibishiiWorkerCounts); err != nil {
 		return errors.Wrapf(err, "Failed to install and prepare data for kibishii %s", kibishiiNamespace)
 	}
 	fmt.Printf("KibishiiPrepareBeforeBackup done %s\n", time.Now().Format("2006-01-02 15:04:05"))
@@ -196,11 +197,15 @@ func RunKibishiiTests(veleroCfg VeleroConfig, backupName, restoreName, backupLoc
 }
 
 func installKibishii(ctx context.Context, namespace string, cloudPlatform, veleroFeatures,
-	kibishiiDirectory string, useVolumeSnapshots bool) error {
+	kibishiiDirectory string, useVolumeSnapshots bool, workerReplicas int32) error {
 	if strings.EqualFold(cloudPlatform, "azure") &&
 		strings.EqualFold(veleroFeatures, "EnableCSI") {
 		cloudPlatform = "azure-csi"
 	}
+	if strings.EqualFold(cloudPlatform, "aws") &&
+		strings.EqualFold(veleroFeatures, "EnableCSI") {
+		cloudPlatform = "aws-csi"
+	}
 	// We use kustomize to generate YAML for Kibishii from the checked-in yaml directories
 	kibishiiInstallCmd := exec.CommandContext(ctx, "kubectl", "apply", "-n", namespace, "-k",
 		kibishiiDirectory+cloudPlatform, "--timeout=90s")
@@ -217,6 +222,7 @@ func installKibishii(ctx context.Context, namespace string, cloudPlatform, veler
 		return errors.Wrapf(err, "failed to label namespace with PSA policy, stderr=%s", stderr)
 	}
 
+	ScaleStatefulSet(ctx, "kibishii-deployment", workerReplicas)
 	kibishiiSetWaitCmd := exec.CommandContext(ctx, "kubectl", "rollout", "status", "statefulset.apps/kibishii-deployment",
 		"-n", namespace, "-w", "--timeout=30m")
 	_, stderr, err = veleroexec.RunCommand(kibishiiSetWaitCmd)
@@ -296,7 +302,8 @@ func waitForKibishiiPods(ctx context.Context, client TestClient, kibishiiNamespa
 
 func KibishiiPrepareBeforeBackup(oneHourTimeout context.Context, client TestClient,
 	providerName, kibishiiNamespace, registryCredentialFile, veleroFeatures,
-	kibishiiDirectory string, useVolumeSnapshots bool, kibishiiData *KibishiiData) error {
+	kibishiiDirectory string, useVolumeSnapshots bool, kibishiiData *KibishiiData,
+	workerReplicas int32) error {
 	fmt.Printf("installKibishii %s\n", time.Now().Format("2006-01-02 15:04:05"))
 	serviceAccountName := "default"
 
@@ -311,7 +318,7 @@ func KibishiiPrepareBeforeBackup(oneHourTimeout context.Context, client TestClie
 	}
 
 	if err := installKibishii(oneHourTimeout, kibishiiNamespace, providerName, veleroFeatures,
-		kibishiiDirectory, useVolumeSnapshots); err != nil {
+		kibishiiDirectory, useVolumeSnapshots, workerReplicas); err != nil {
 		return errors.Wrap(err, "Failed to install Kibishii workload")
 	}
 	// wait for kibishii pod startup