Merge pull request csi-addons#171 from red-hat-storage/sync_ds--main

Syncing latest changes from main for kubernetes-csi-addons

Makefile

@@ -119,7 +119,7 @@ help: ## Display this help.
 
 .PHONY: manifests
 manifests: controller-gen kustomize ## Generate WebhookConfiguration, ClusterRole and CustomResourceDefinition objects.
-	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="{./apis/...,./cmd/...,./controllers/...,./internal/...,./sidecar/...}" output:crd:artifacts:config=config/crd/bases
+	$(CONTROLLER_GEN) rbac:roleName=manager-role crd webhook paths="{./api/...,./cmd/...,./internal/controller/...,./internal/...,./sidecar/...}" output:crd:artifacts:config=config/crd/bases
 	cd config/manager && $(KUSTOMIZE) edit set image controller=${CONTROLLER_IMG} rbac-proxy=${RBAC_PROXY_IMG}
 	$(KUSTOMIZE) build config/crd > deploy/controller/crds.yaml
 	$(KUSTOMIZE) build config/rbac > deploy/controller/rbac.yaml
@@ -136,7 +136,7 @@ bundle: gen-csv-base manifests operator-sdk
 
 .PHONY: generate
 generate: controller-gen ## Generate code containing DeepCopy, DeepCopyInto, and DeepCopyObject method implementations.
-	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./apis/..."
+	$(CONTROLLER_GEN) object:headerFile="hack/boilerplate.go.txt" paths="./api/..."
 
 .PHONY: generate-protobuf
 generate-protobuf: protoc-gen-go protoc-gen-go-grpc
@@ -219,7 +219,7 @@ uninstall: manifests ## Uninstall CRDs from the K8s cluster specified in ~/.kube
 
 .PHONY: deploy
 deploy: manifests ## Deploy controller to the K8s cluster specified in ~/.kube/config.
-	cd deploy/controller && kubectl apply -f crds.yaml -f rbac.yaml -f setup-controller.yaml
+	cd deploy/controller && kubectl apply -f setup-controller.yaml -f crds.yaml -f rbac.yaml
 
 .PHONY: undeploy
 undeploy: ## Undeploy controller from the K8s cluster specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource not found errors during deletion.

PROJECT

@@ -4,7 +4,7 @@
 # More info: https://book.kubebuilder.io/reference/project-config.html
 domain: openshift.io
 layout:
-- go.kubebuilder.io/v3
+- go.kubebuilder.io/v4
 multigroup: true
 projectName: kubernetes-csi-addons
 repo: github.com/csi-addons/kubernetes-csi-addons
@@ -16,7 +16,7 @@ resources:
   domain: openshift.io
   group: csiaddons
   kind: CSIAddonsNode
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/csiaddons/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1
   version: v1alpha1
 - api:
     crdVersion: v1
@@ -25,21 +25,21 @@ resources:
   domain: openshift.io
   group: csiaddons
   kind: ReclaimSpaceJob
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/csiaddons/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1
   version: v1alpha1
 - api:
     crdVersion: v1
   controller: true
   domain: openshift.io
   group: csiaddons
   kind: NetworkFence
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/csiaddons/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1
   version: v1alpha1
 - controller: true
   domain: openshift.io
   group: csiaddons
   kind: ReclaimSpaceCronJob
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/csiaddons/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/csiaddons/v1alpha1
   version: v1alpha1
 - controller: true
   group: core
@@ -52,23 +52,23 @@ resources:
   domain: openshift.io
   group: replication.storage
   kind: VolumeReplication
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1
   version: v1alpha1
 - api:
     crdVersion: v1
     namespaced: true
   domain: openshift.io
   group: replication.storage
   kind: VolumeReplicationClass
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1
   version: v1alpha1
 - api:
     crdVersion: v1
   controller: true
   domain: openshift.io
   group: replication.storage
   kind: VolumeGroupReplicationClass
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1
   version: v1alpha1
 - api:
     crdVersion: v1
@@ -77,14 +77,14 @@ resources:
   domain: openshift.io
   group: replication.storage
   kind: VolumeGroupReplication
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1
   version: v1alpha1
 - api:
     crdVersion: v1
   controller: true
   domain: openshift.io
   group: replication.storage
   kind: VolumeGroupReplicationContent
-  path: github.com/csi-addons/kubernetes-csi-addons/apis/replication.storage/v1alpha1
+  path: github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1
   version: v1alpha1
 version: "3"

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")

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)

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/api/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
+}