replication: support for volumegroup replication

update the RPC calls to set the replication source for group replication and add support for volumegroup replication Signed-off-by: Madhu Rajanna <[email protected]>
csi-addons · Jul 10, 2024 · 0f8a18e · 0f8a18e
1 parent fa5f8df
commit 0f8a18e
Show file tree

Hide file tree

Showing 13 changed files with 831 additions and 162 deletions.
diff --git a/cmd/csi-addons/main.go b/cmd/csi-addons/main.go
@@ -37,15 +37,16 @@ const (
 // command contains the parsed arguments that were passed while running the
 // executable.
 type command struct {
-	endpoint         string
-	stagingPath      string
-	operation        string
-	persistentVolume string
-	drivername       string
-	secret           string
-	cidrs            string
-	clusterid        string
-	legacy           bool
+	endpoint                      string
+	stagingPath                   string
+	operation                     string
+	persistentVolume              string
+	volumeGroupReplicationContent string
+	drivername                    string
+	secret                        string
+	cidrs                         string
+	clusterid                     string
+	legacy                        bool
 }
 
 // cmd is the single instance of the command struct, used inside main().
@@ -58,6 +59,7 @@ func init() {
 	flag.StringVar(&cmd.stagingPath, "stagingpath", stagingPath, "staging path")
 	flag.StringVar(&cmd.operation, "operation", "", "csi-addons operation")
 	flag.StringVar(&cmd.persistentVolume, "persistentvolume", "", "name of the PersistentVolume")
+	flag.StringVar(&cmd.volumeGroupReplicationContent, "volumegroupreplicationcontent", "", "name of the VolumeGroupReplicationContent")
 	flag.StringVar(&cmd.drivername, "drivername", "", "name of the CSI driver")
 	flag.StringVar(&cmd.secret, "secret", "", "kubernetes secret in the format `namespace/name`")
 	flag.StringVar(&cmd.cidrs, "cidrs", "", "comma separated list of cidrs")

diff --git a/cmd/csi-addons/replication.go b/cmd/csi-addons/replication.go
@@ -36,6 +36,7 @@ type VolumeReplicationBase struct {
 	secretName      string
 	secretNamespace string
 	volumeID        string
+	groupID         string
 }
 
 func (rep *VolumeReplicationBase) Init(c *command) error {
@@ -59,28 +60,69 @@ func (rep *VolumeReplicationBase) Init(c *command) error {
 		return errors.New("secret name is not set")
 	}
 
-	pv, err := getKubernetesClient().CoreV1().PersistentVolumes().Get(context.Background(), c.persistentVolume, metav1.GetOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to get pv %q", c.persistentVolume)
-	}
-
-	if pv.Spec.CSI == nil {
-		return fmt.Errorf("pv %q is not a CSI volume", c.persistentVolume)
+	if c.persistentVolume != "" && c.volumeGroupReplicationContent != "" {
+		return errors.New("only one of persistentVolume or volumeGroupReplicationContent should be set")
 	}
 
-	if pv.Spec.CSI.VolumeHandle == "" {
-		return errors.New("volume ID is not set")
+	if c.persistentVolume != "" {
+		pv, err := getKubernetesClient().CoreV1().PersistentVolumes().Get(context.Background(), c.persistentVolume, metav1.GetOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to get pv %q", c.persistentVolume)
+		}
+
+		if pv.Spec.CSI == nil {
+			return fmt.Errorf("pv %q is not a CSI volume", c.persistentVolume)
+		}
+
+		if pv.Spec.CSI.VolumeHandle == "" {
+			return errors.New("volume ID is not set")
+		}
+		rep.volumeID = pv.Spec.CSI.VolumeHandle
+		return nil
+	} else if c.volumeGroupReplicationContent != "" {
+		vgrc, err := getVolumeReplicationClient().getVolumeGroupReplicationContent(context.Background(), c.volumeGroupReplicationContent, metav1.GetOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to get VolumeGroupReplicationContent %q", c.volumeGroupReplicationContent)
+		}
+		if vgrc.Spec.VolumeGroupReplicationHandle == "" {
+			return errors.New("volume group ID is not set")
+		}
+		rep.groupID = vgrc.Spec.VolumeGroupReplicationHandle
+		return nil
 	}
-	rep.volumeID = pv.Spec.CSI.VolumeHandle
 
-	return nil
+	return errors.New("either persistentVolume or volumeGroupReplicationContent should be set")
 }
 
 // EnableVolumeReplication executes the EnableVolumeReplication operation.
 type EnableVolumeReplication struct {
 	VolumeReplicationBase
 }
 
+func (v VolumeReplicationBase) setReplicationSource(req *proto.ReplicationSource) error {
+	switch {
+	case req == nil:
+		return errors.New("replication source is not set")
+	case v.volumeID != "" && v.groupID != "":
+		return errors.New("only one of volumeID or groupID should be set")
+	case v.volumeID != "":
+		req.Type = &proto.ReplicationSource_Volume{
+			Volume: &proto.ReplicationSource_VolumeSource{
+				VolumeId: v.volumeID,
+			},
+		}
+		return nil
+	case v.groupID != "":
+		req.Type = &proto.ReplicationSource_VolumeGroup{
+			VolumeGroup: &proto.ReplicationSource_VolumeGroupSource{
+				VolumeGroupId: v.groupID,
+			},
+		}
+		return nil
+	}
+	return errors.New("both volumeID and groupID is not set")
+}
+
 var _ = registerOperation("EnableVolumeReplication", &EnableVolumeReplication{})
 
 func (rep *EnableVolumeReplication) Execute() error {
@@ -91,10 +133,12 @@ func (rep *EnableVolumeReplication) Execute() error {
 	req := &proto.EnableVolumeReplicationRequest{
 		SecretName:      rep.secretName,
 		SecretNamespace: rep.secretNamespace,
-		VolumeId:        rep.volumeID,
 	}
-
-	_, err := rs.EnableVolumeReplication(context.TODO(), req)
+	err := rep.setReplicationSource(req.ReplicationSource)
+	if err != nil {
+		return err
+	}
+	_, err = rs.EnableVolumeReplication(context.TODO(), req)
 	if err != nil {
 		return err
 	}
@@ -119,10 +163,13 @@ func (rep *DisableVolumeReplication) Execute() error {
 	req := &proto.DisableVolumeReplicationRequest{
 		SecretName:      rep.secretName,
 		SecretNamespace: rep.secretNamespace,
-		VolumeId:        rep.volumeID,
+	}
+	err := rep.setReplicationSource(req.ReplicationSource)
+	if err != nil {
+		return err
 	}
 
-	_, err := rs.DisableVolumeReplication(context.TODO(), req)
+	_, err = rs.DisableVolumeReplication(context.TODO(), req)
 	if err != nil {
 		return err
 	}
@@ -147,10 +194,12 @@ func (rep *PromoteVolume) Execute() error {
 	req := &proto.PromoteVolumeRequest{
 		SecretName:      rep.secretName,
 		SecretNamespace: rep.secretNamespace,
-		VolumeId:        rep.volumeID,
 	}
-
-	_, err := rs.PromoteVolume(context.TODO(), req)
+	err := rep.setReplicationSource(req.ReplicationSource)
+	if err != nil {
+		return err
+	}
+	_, err = rs.PromoteVolume(context.TODO(), req)
 	if err != nil {
 		return err
 	}
@@ -175,10 +224,12 @@ func (rep *DemoteVolume) Execute() error {
 	req := &proto.DemoteVolumeRequest{
 		SecretName:      rep.secretName,
 		SecretNamespace: rep.secretNamespace,
-		VolumeId:        rep.volumeID,
 	}
-
-	_, err := rs.DemoteVolume(context.TODO(), req)
+	err := rep.setReplicationSource(req.ReplicationSource)
+	if err != nil {
+		return err
+	}
+	_, err = rs.DemoteVolume(context.TODO(), req)
 	if err != nil {
 		return err
 	}
@@ -203,10 +254,12 @@ func (rep *ResyncVolume) Execute() error {
 	req := &proto.ResyncVolumeRequest{
 		SecretName:      rep.secretName,
 		SecretNamespace: rep.secretNamespace,
-		VolumeId:        rep.volumeID,
 	}
-
-	_, err := rs.ResyncVolume(context.TODO(), req)
+	err := rep.setReplicationSource(req.ReplicationSource)
+	if err != nil {
+		return err
+	}
+	_, err = rs.ResyncVolume(context.TODO(), req)
 	if err != nil {
 		return err
 	}
@@ -231,7 +284,10 @@ func (rep *GetVolumeReplicationInfo) Execute() error {
 	req := &proto.GetVolumeReplicationInfoRequest{
 		SecretName:      rep.secretName,
 		SecretNamespace: rep.secretNamespace,
-		VolumeId:        rep.volumeID,
+	}
+	err := rep.setReplicationSource(req.ReplicationSource)
+	if err != nil {
+		return err
 	}
 
 	res, err := rs.GetVolumeReplicationInfo(context.TODO(), req)

diff --git a/cmd/csi-addons/replicationClient.go b/cmd/csi-addons/replicationClient.go
@@ -0,0 +1,70 @@
+/*
+Copyright 2024 The Ceph-CSI Authors.
+
+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 main
+
+import (
+	"context"
+
+	replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+)
+
+type replicationClient struct {
+	restClient *rest.RESTClient
+}
+
+func getVolumeReplicationClient() *replicationClient {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		panic(err.Error())
+	}
+	scheme, err := replicationv1alpha1.SchemeBuilder.Build()
+	if err != nil {
+		panic(err.Error())
+	}
+
+	crdConfig := *config
+	crdConfig.ContentConfig.GroupVersion = &replicationv1alpha1.GroupVersion
+	crdConfig.APIPath = "/apis"
+	crdConfig.NegotiatedSerializer = serializer.NewCodecFactory(scheme)
+	crdConfig.UserAgent = rest.DefaultKubernetesUserAgent()
+
+	restClient, err := rest.UnversionedRESTClientFor(&crdConfig)
+	if err != nil {
+		panic(err)
+	}
+
+	return &replicationClient{restClient: restClient}
+}
+
+func (r *replicationClient) getVolumeGroupReplicationContent(ctx context.Context, name string, opts metav1.GetOptions) (*replicationv1alpha1.VolumeGroupReplicationContent, error) {
+	result := replicationv1alpha1.VolumeGroupReplicationContent{}
+	err := r.restClient.
+		Get().
+		Namespace("").
+		Resource("volumegroupreplicationcontents").
+		Name(name).
+		VersionedParams(&opts, scheme.ParameterCodec).
+		Do(ctx).
+		Into(&result)
+
+	return &result, err
+}
diff --git a/controllers/replication.storage/finalizers.go b/controllers/replication.storage/finalizers.go
@@ -31,6 +31,7 @@ import (
 const (
 	volumeReplicationFinalizer = "replication.storage.openshift.io"
 	pvcReplicationFinalizer    = "replication.storage.openshift.io/pvc-protection"
+	vgrReplicationFinalizer    = "replication.storage.openshift.io/vgr-protection"
 )
 
 // addFinalizerToVR adds the VR finalizer on the VolumeReplication instance.
@@ -94,3 +95,34 @@ func (r *VolumeReplicationReconciler) removeFinalizerFromPVC(logger logr.Logger,
 
 	return nil
 }
+
+// addFinalizerToVGR adds the VR finalizer on the VolumeGroupReplication.
+func (r *VolumeReplicationReconciler) addFinalizerToVGR(logger logr.Logger, vgr *replicationv1alpha1.VolumeGroupReplication) error {
+	if !slices.Contains(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer) {
+		logger.Info("adding finalizer to VolumeGroupReplication object", "Finalizer", vgrReplicationFinalizer)
+		vgr.ObjectMeta.Finalizers = append(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer)
+		if err := r.Client.Update(context.TODO(), vgr); err != nil {
+			return fmt.Errorf("failed to add finalizer (%s) to VolumeGroupReplication resource"+
+				" (%s/%s) %w",
+				vgrReplicationFinalizer, vgr.Namespace, vgr.Name, err)
+		}
+	}
+
+	return nil
+}
+
+// removeFinalizerFromVGR removes the VR finalizer on VolumeGroupReplication.
+func (r *VolumeReplicationReconciler) removeFinalizerFromVGR(logger logr.Logger, vgr *replicationv1alpha1.VolumeGroupReplication,
+) error {
+	if slices.Contains(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer) {
+		logger.Info("removing finalizer from VolumeGroupReplication object", "Finalizer", vgrReplicationFinalizer)
+		vgr.ObjectMeta.Finalizers = util.RemoveFromSlice(vgr.ObjectMeta.Finalizers, vgrReplicationFinalizer)
+		if err := r.Client.Update(context.TODO(), vgr); err != nil {
+			return fmt.Errorf("failed to remove finalizer (%s) from VolumeGroupReplication resource"+
+				" (%s/%s), %w",
+				vgrReplicationFinalizer, vgr.Namespace, vgr.Name, err)
+		}
+	}
+
+	return nil
+}