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/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,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,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.PassNum = 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 +198,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 +215,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 +263,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/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
+}

test/e2e/util/kibishii/kibishii_utils.go

@@ -48,7 +48,9 @@ type KibishiiData struct {
 	ExpectedNodes int
 }
 
-var DefaultKibishiiData = &KibishiiData{2, 10, 10, 1024, 1024, 0, 2}
+var DefaultKibishiiWorkerCounts = 2
+var DefaultKibishiiData = &KibishiiData{2, 10, 10, 1024, 1024, 0, DefaultKibishiiWorkerCounts}
+
 var KibishiiPVCNameList = []string{"kibishii-data-kibishii-deployment-0", "kibishii-data-kibishii-deployment-1"}
 
 // RunKibishiiTests runs kibishii tests on the provider.
@@ -196,11 +198,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 int) 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")
@@ -216,6 +222,12 @@ func installKibishii(ctx context.Context, namespace string, cloudPlatform, veler
 	if err != nil {
 		return errors.Wrapf(err, "failed to label namespace with PSA policy, stderr=%s", stderr)
 	}
+	if workerReplicas != DefaultKibishiiWorkerCounts {
+		err = ScaleStatefulSet(ctx, namespace, "kibishii-deployment", workerReplicas)
+		if err != nil {
+			return errors.Wrapf(err, "failed to scale statefulset, stderr=%s", err.Error())
+		}
+	}
 
 	kibishiiSetWaitCmd := exec.CommandContext(ctx, "kubectl", "rollout", "status", "statefulset.apps/kibishii-deployment",
 		"-n", namespace, "-w", "--timeout=30m")
@@ -311,7 +323,7 @@ func KibishiiPrepareBeforeBackup(oneHourTimeout context.Context, client TestClie
 	}
 
 	if err := installKibishii(oneHourTimeout, kibishiiNamespace, providerName, veleroFeatures,
-		kibishiiDirectory, useVolumeSnapshots); err != nil {
+		kibishiiDirectory, useVolumeSnapshots, kibishiiData.PassNum); err != nil {
 		return errors.Wrap(err, "Failed to install Kibishii workload")
 	}
 	// wait for kibishii pod startup

test/e2e/util/velero/install.go

@@ -68,8 +68,8 @@ func VeleroInstall(ctx context.Context, veleroCfg *VeleroConfig) error {
 			return errors.New("No object store provider specified - must be specified when using kind as the cloud provider") // Gotta have an object store provider
 		}
 	}
-
-	providerPluginsTmp, err := getProviderPlugins(ctx, veleroCfg.VeleroCLI, veleroCfg.ObjectStoreProvider, veleroCfg.CloudProvider, veleroCfg.Plugins, veleroCfg.Features)
+	//.VeleroCLI, veleroCfg.ObjectStoreProvider, veleroCfg.CloudProvider, veleroCfg.Plugins, veleroCfg.Features
+	providerPluginsTmp, err := getProviderPlugins(ctx, *veleroCfg)
 	if err != nil {
 		return errors.WithMessage(err, "Failed to get provider plugins")
 	}
@@ -96,17 +96,14 @@ func VeleroInstall(ctx context.Context, veleroCfg *VeleroConfig) error {
 		return errors.WithMessagef(err, "Failed to get Velero InstallOptions for plugin provider %s", veleroCfg.ObjectStoreProvider)
 	}
 	veleroInstallOptions.UseVolumeSnapshots = veleroCfg.UseVolumeSnapshots
-	if !veleroCfg.UseRestic {
-		veleroInstallOptions.UseNodeAgent = veleroCfg.UseNodeAgent
-	}
-	veleroInstallOptions.UseRestic = veleroCfg.UseRestic
+	veleroInstallOptions.UseNodeAgent = veleroCfg.UseNodeAgent
 	veleroInstallOptions.Image = veleroCfg.VeleroImage
 	veleroInstallOptions.Namespace = veleroCfg.VeleroNamespace
 	veleroInstallOptions.UploaderType = veleroCfg.UploaderType
 	GCFrequency, _ := time.ParseDuration(veleroCfg.GCFrequency)
 	veleroInstallOptions.GarbageCollectionFrequency = GCFrequency
 
-	err = installVeleroServer(ctx, veleroCfg.VeleroCLI, &installOptions{
+	err = installVeleroServer(ctx, veleroCfg.VeleroCLI, veleroCfg.CloudProvider, &installOptions{
 		Options:                veleroInstallOptions,
 		RegistryCredentialFile: veleroCfg.RegistryCredentialFile,
 		RestoreHelperImage:     veleroCfg.RestoreHelperImage,
@@ -176,7 +173,7 @@ func clearupvSpherePluginConfig(c clientset.Interface, ns, secretName, configMap
 	return nil
 }
 
-func installVeleroServer(ctx context.Context, cli string, options *installOptions) error {
+func installVeleroServer(ctx context.Context, cli, cloudProvider string, options *installOptions) error {
 	args := []string{"install"}
 	namespace := "velero"
 	if len(options.Namespace) > 0 {
@@ -192,9 +189,6 @@ func installVeleroServer(ctx context.Context, cli string, options *installOption
 	if options.DefaultVolumesToFsBackup {
 		args = append(args, "--default-volumes-to-fs-backup")
 	}
-	if options.UseRestic {
-		args = append(args, "--use-restic")
-	}
 	if options.UseVolumeSnapshots {
 		args = append(args, "--use-volume-snapshots")
 	}
@@ -219,10 +213,11 @@ func installVeleroServer(ctx context.Context, cli string, options *installOption
 	if len(options.Plugins) > 0 {
 		args = append(args, "--plugins", options.Plugins.String())
 	}
+	fmt.Println("Start to install Azure VolumeSnapshotClass ...1")
 	if len(options.Features) > 0 {
 		args = append(args, "--features", options.Features)
 		if strings.EqualFold(options.Features, "EnableCSI") && options.UseVolumeSnapshots {
-			if strings.EqualFold(options.ProviderName, "Azure") {
+			if strings.EqualFold(cloudProvider, "azure") {
 				if err := KubectlApplyByFile(ctx, "util/csi/AzureVolumeSnapshotClass.yaml"); err != nil {
 					return err
 				}