Merge pull request #96 from v-harihar/migration-rsync

CLI: Migration rsync

Author: openshift-merge-robot

kubectl-volsync/cmd/migration.go

@@ -17,10 +17,18 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
 package cmd
 
 import (
+	"context"
+	"fmt"
+	"time"
+
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
 	"k8s.io/kubectl/pkg/util/i18n"
 	"k8s.io/kubectl/pkg/util/templates"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	volsyncv1alpha1 "github.com/backube/volsync/api/v1alpha1"
 )
@@ -104,3 +112,62 @@ func init() {
 	cobra.CheckErr(migrationCmd.MarkPersistentFlagRequired("relationship"))
 	cobra.CheckErr(viper.BindPFlag("relationship", migrationCmd.PersistentFlags().Lookup("relationship")))
 }
+
+func loadMigrationRelationship(cmd *cobra.Command) (*migrationRelationship, error) {
+	r, err := LoadRelationshipFromCommand(cmd, MigrationRelationshipType)
+	if err != nil {
+		return nil, err
+	}
+
+	mr := &migrationRelationship{
+		Relationship: *r,
+	}
+
+	// Decode according to the file version
+	version := mr.GetInt("data.version")
+	switch version {
+	case 1:
+		if err := mr.GetData(&mr.data); err != nil {
+			return nil, err
+		}
+	default:
+		return nil, fmt.Errorf("unsupported config file version %d", version)
+	}
+	return mr, nil
+}
+
+func (mrd *migrationRelationshipDestination) waitForRDStatus(ctx context.Context, clientObject client.Client) (
+	*volsyncv1alpha1.ReplicationDestination, error) {
+	// wait for migrationdestination to become ready
+	nsName := types.NamespacedName{
+		Namespace: mrd.Namespace,
+		Name:      mrd.RDName,
+	}
+	rd := &volsyncv1alpha1.ReplicationDestination{}
+	err := wait.PollImmediate(5*time.Second, 2*time.Minute, func() (bool, error) {
+		err := clientObject.Get(ctx, nsName, rd)
+		if err != nil {
+			return false, err
+		}
+		if rd.Status == nil || rd.Status.Rsync == nil {
+			return false, nil
+		}
+		if rd.Status.Rsync.Address == nil {
+			klog.V(2).Infof("Waiting for MigrationDestination %s RSync address to populate", rd.Name)
+			return false, nil
+		}
+
+		if rd.Status.Rsync.SSHKeys == nil {
+			klog.V(2).Infof("Waiting for MigrationDestination %s RSync sshkeys to populate", rd.Name)
+			return false, nil
+		}
+
+		klog.V(2).Infof("Found MigrationDestination RSync Address: %s", *rd.Status.Rsync.Address)
+		return true, nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch rd status: %w,", err)
+	}
+
+	return rd, nil
+}

kubectl-volsync/cmd/migration_create.go

@@ -19,15 +19,13 @@ package cmd
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/spf13/cobra"
 	corev1 "k8s.io/api/core/v1"
 	kerrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	"k8s.io/kubectl/pkg/util/i18n"
 	"k8s.io/kubectl/pkg/util/templates"
@@ -63,7 +61,7 @@ type migrationCreate struct {
 	// SSH connections.
 	ServiceType *corev1.ServiceType
 	// client object to communicate with a cluster
-	clientObject client.Client
+	client client.Client
 	// PVC object associated with pvcName used to create destination object
 	PVC *corev1.PersistentVolumeClaim
 }
@@ -216,7 +214,7 @@ func (mc *migrationCreate) Run(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	mc.clientObject = k8sClient
+	mc.client = k8sClient
 
 	// Get the pvc from cluster
 	mc.PVC, err = mc.getDestinationPVC(ctx)
@@ -247,7 +245,7 @@ func (mc *migrationCreate) Run(ctx context.Context) error {
 	}
 
 	// Wait for ReplicationDestination to post address, sshkeys
-	err = mc.waitForRDStatus(ctx)
+	_, err = mc.mr.data.Destination.waitForRDStatus(ctx, mc.client)
 	if err != nil {
 		return err
 	}
@@ -264,7 +262,7 @@ func (mc *migrationCreate) ensureNamespace(ctx context.Context) (*corev1.Namespa
 			Name: mc.Namespace,
 		},
 	}
-	if err := mc.clientObject.Create(ctx, ns); err != nil {
+	if err := mc.client.Create(ctx, ns); err != nil {
 		if kerrs.IsAlreadyExists(err) {
 			klog.Infof("Namespace: \"%s\" is found, proceeding with the same",
 				mc.Namespace)
@@ -309,7 +307,7 @@ func (mc *migrationCreate) createDestinationPVC(ctx context.Context) (*corev1.Pe
 		},
 	}
 
-	if err := mc.clientObject.Create(ctx, destPVC); err != nil {
+	if err := mc.client.Create(ctx, destPVC); err != nil {
 		return nil, err
 	}
 
@@ -325,7 +323,7 @@ func (mc *migrationCreate) getDestinationPVC(ctx context.Context) (*corev1.Persi
 		Namespace: mc.Namespace,
 		Name:      mc.DestinationPVC,
 	}
-	err := mc.clientObject.Get(ctx, pvcInfo, destPVC)
+	err := mc.client.Get(ctx, pvcInfo, destPVC)
 	if err != nil {
 		if client.IgnoreNotFound(err) == nil {
 			klog.Infof("pvc: \"%s\" not found, creating the same", mc.DestinationPVC)
@@ -353,47 +351,11 @@ func (mc *migrationCreate) ensureReplicationDestination(ctx context.Context) (
 			},
 		},
 	}
-	if err := mc.clientObject.Create(ctx, rd); err != nil {
+	if err := mc.client.Create(ctx, rd); err != nil {
 		return nil, err
 	}
 	klog.Infof("Created ReplicationDestination: \"%s\" in Namespace: \"%s\" and Cluster: \"%s\"",
 		rd.Name, rd.Namespace, rd.ClusterName)
 
 	return rd, nil
 }
-
-func (mc *migrationCreate) waitForRDStatus(ctx context.Context) error {
-	mrd := mc.mr.data.Destination
-	// wait for migrationdestination to become ready
-	nsName := types.NamespacedName{
-		Namespace: mrd.Namespace,
-		Name:      mrd.RDName,
-	}
-	rd := &volsyncv1alpha1.ReplicationDestination{}
-	err := wait.PollImmediate(5*time.Second, 2*time.Minute, func() (bool, error) {
-		err := mc.clientObject.Get(ctx, nsName, rd)
-		if err != nil {
-			return false, err
-		}
-		if rd.Status == nil || rd.Status.Rsync == nil {
-			return false, nil
-		}
-		if rd.Status.Rsync.Address == nil {
-			klog.V(2).Infof("Waiting for MigrationDestination %s RSync address to populate", rd.Name)
-			return false, nil
-		}
-
-		if rd.Status.Rsync.SSHKeys == nil {
-			klog.V(2).Infof("Waiting for MigrationDestination %s RSync sshkeys to populate", rd.Name)
-			return false, nil
-		}
-
-		klog.V(2).Infof("Found MigrationDestination RSync Address: %s", *rd.Status.Rsync.Address)
-		return true, nil
-	})
-	if err != nil {
-		return fmt.Errorf("failed to fetch rd status: %w,", err)
-	}
-
-	return nil
-}

kubectl-volsync/cmd/migration_create_test.go

@@ -53,7 +53,7 @@ var _ = Describe("migration", func() {
 		err = mc.parseCLI(cmd)
 		Expect(err).NotTo(HaveOccurred())
 
-		mc.clientObject = k8sClient
+		mc.client = k8sClient
 		mc.Namespace = "foo-" + krand.String(5)
 		// create namespace
 		ns, err = mc.ensureNamespace(context.Background())
@@ -166,7 +166,7 @@ var _ = Describe("migration", func() {
 			}}
 		Expect(k8sClient.Status().Update(context.Background(), rd)).To(Succeed())
 		// Wait for mock address and sshKey to pop up
-		err = mc.waitForRDStatus(context.Background())
+		_, err = mc.mr.data.Destination.waitForRDStatus(context.Background(), mc.client)
 		Expect(err).ToNot(HaveOccurred())
 		// Retry creation of replicationdestination and it should fail as destination already exists
 		rd, err = mc.ensureReplicationDestination(context.Background())
@@ -176,7 +176,7 @@ var _ = Describe("migration", func() {
 		Expect(k8sClient.Get(context.Background(), types.NamespacedName{Namespace: mc.Namespace,
 			Name: mc.RDName}, rd)).To(Succeed())
 		// Should return existing address and sshkey
-		err = mc.waitForRDStatus(context.Background())
+		_, err = mc.mr.data.Destination.waitForRDStatus(context.Background(), mc.client)
 		Expect(err).ToNot(HaveOccurred())
 	})
 })