rbd: VolumeGroupReplicationContent controller to regenerate the OMAP data

cmd/cephcsi.go

@@ -26,6 +26,7 @@ import (
 	"github.com/ceph/ceph-csi/internal/cephfs"
 	"github.com/ceph/ceph-csi/internal/controller"
 	"github.com/ceph/ceph-csi/internal/controller/persistentvolume"
+	"github.com/ceph/ceph-csi/internal/controller/volumegroup"
 	"github.com/ceph/ceph-csi/internal/liveness"
 	nfsdriver "github.com/ceph/ceph-csi/internal/nfs/driver"
 	rbddriver "github.com/ceph/ceph-csi/internal/rbd/driver"
@@ -294,6 +295,7 @@ func setPIDLimit(conf *util.Config) {
 func initControllers() {
 	// Add list of controller here.
 	persistentvolume.Init()
+	volumegroup.Init()
 }
 
 func validateCloneDepthFlag(conf *util.Config) {

go.mod

@@ -48,8 +48,11 @@ require (
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.0
 	github.com/Azure/azure-sdk-for-go/sdk/security/keyvault/azsecrets v1.2.0
+	github.com/csi-addons/kubernetes-csi-addons v0.9.0
 )
 
+require github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
+
 require (
 	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.16.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0 // indirect
@@ -87,7 +90,6 @@ require (
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.4 // indirect
-	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v5 v5.2.1 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -156,7 +158,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.28.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.26.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect
 	golang.org/x/oauth2 v0.22.0 // indirect
 	golang.org/x/sync v0.8.0 // indirect

go.sum

@@ -1504,6 +1504,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46t
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/creack/pty v1.1.18/go.mod h1:MOBLtS5ELjhRRrroQr9kyvTxUAFNvYEK993ew/Vr4O4=
+github.com/csi-addons/kubernetes-csi-addons v0.9.0 h1:Hhb44WcrxtbzmpLY+uqX+DBWCI6HgA/rwQMPyvsyCc8=
+github.com/csi-addons/kubernetes-csi-addons v0.9.0/go.mod h1:/YROZDdEi1N/1Ls9rdU5W2VNjm8MK7HHApl8W4Sqt9s=
 github.com/csi-addons/spec v0.2.1-0.20241104111131-27825f744db5 h1:j9NaWj5KmzEVarmsjxS/NDAhes6Uzq1qhkUGHvDlVBk=
 github.com/csi-addons/spec v0.2.1-0.20241104111131-27825f744db5/go.mod h1:Mwq4iLiUV4s+K1bszcWU6aMsR5KPsbIYzzszJ6+56vI=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -2413,8 +2415,9 @@ go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
 go.uber.org/zap v1.18.1/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
-go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=
 go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 golang.org/x/crypto v0.0.0-20180820150726-614d502a4dac/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

internal/cephfs/store/volumegroup.go

@@ -249,7 +249,7 @@ func ReserveVolumeGroup(
 	defer j.Destroy()
 
 	groupUUID, vgsi.FsVolumeGroupSnapshotName, err = j.ReserveName(
-		ctx, volOptions.MetadataPool, volOptions.RequestName, volOptions.NamePrefix)
+		ctx, volOptions.MetadataPool, volOptions.RequestName, volOptions.ReservedID, volOptions.NamePrefix)
 	if err != nil {
 		return nil, err
 	}

internal/controller/controller.go

@@ -20,13 +20,17 @@ import (
 
 	"github.com/ceph/ceph-csi/internal/util/log"
 
-	"k8s.io/apimachinery/pkg/runtime"
+	apiruntime "k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	clientConfig "sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1"
 )
 
 // Manager is the interface that will wrap Add function.
@@ -62,6 +66,9 @@ func addToManager(mgr manager.Manager, config Config) error {
 
 // Start will start all the registered managers.
 func Start(config Config) error {
+	scheme := apiruntime.NewScheme()
+	utilruntime.Must(replicationv1alpha1.AddToScheme(scheme))
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 	electionID := config.DriverName + "-" + config.Namespace
 	opts := manager.Options{
 		LeaderElection: true,
@@ -70,11 +77,12 @@ func Start(config Config) error {
 		LeaderElectionNamespace:    config.Namespace,
 		LeaderElectionResourceLock: resourcelock.LeasesResourceLock,
 		LeaderElectionID:           electionID,
+		Scheme:                     scheme,
 	}
 
 	kubeConfig := clientConfig.GetConfigOrDie()
 	coreKubeConfig := rest.CopyConfig(kubeConfig)
-	coreKubeConfig.ContentType = runtime.ContentTypeProtobuf
+	coreKubeConfig.ContentType = apiruntime.ContentTypeProtobuf
 	mgr, err := manager.New(coreKubeConfig, opts)
 	if err != nil {
 		log.ErrorLogMsg("failed to create manager %s", err)

internal/controller/volumegroup/volumegroupreplicationcontent.go

@@ -0,0 +1,200 @@
+/*
+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 volumegroup
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	replicationv1alpha1 "github.com/csi-addons/kubernetes-csi-addons/api/replication.storage/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
+
+	ctrl "github.com/ceph/ceph-csi/internal/controller"
+	"github.com/ceph/ceph-csi/internal/rbd"
+	"github.com/ceph/ceph-csi/internal/util"
+	"github.com/ceph/ceph-csi/internal/util/log"
+)
+
+type ReconcileVGRContent struct {
+	client client.Client
+	config ctrl.Config
+	Locks  *util.VolumeLocks
+}
+
+var (
+	_ reconcile.Reconciler = &ReconcileVGRContent{}
+	_ ctrl.Manager         = &ReconcileVGRContent{}
+)
+
+const (
+	secretNameParameterName      = "replication.storage.openshift.io/group-replication-secret-name"
+	secretNamespaceParameterName = "replication.storage.openshift.io/group-replication-secret-namespace"
+)
+
+// Init will add the ReconcileVGRContent to the list.
+func Init() {
+	// add ReconcileVGRContent to the list
+	ctrl.ControllerList = append(ctrl.ControllerList, &ReconcileVGRContent{})
+}
+
+// Add adds the newVGRContentReconciler.
+func (r *ReconcileVGRContent) Add(mgr manager.Manager, config ctrl.Config) error {
+	return add(mgr, newVGRContentReconciler(mgr, config))
+}
+
+// newVGRContentReconciler returns a ReconcileVGRContent.
+func newVGRContentReconciler(mgr manager.Manager, config ctrl.Config) reconcile.Reconciler {
+	r := &ReconcileVGRContent{
+		client: mgr.GetClient(),
+		config: config,
+		Locks:  util.NewVolumeLocks(),
+	}
+
+	return r
+}
+
+func add(mgr manager.Manager, r reconcile.Reconciler) error {
+	// Create a new controller
+	c, err := controller.New(
+		"vgrcontent-controller",
+		mgr,
+		controller.Options{MaxConcurrentReconciles: 1, Reconciler: r})
+	if err != nil {
+		return err
+	}
+
+	// Watch for changes to VolumeGroupReplicationContent
+	err = c.Watch(source.Kind(
+		mgr.GetCache(),
+		&replicationv1alpha1.VolumeGroupReplicationContent{},
+		&handler.TypedEnqueueRequestForObject[*replicationv1alpha1.VolumeGroupReplicationContent]{}),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to watch the changes: %w", err)
+	}
+
+	return nil
+}
+
+func (r *ReconcileVGRContent) getSecrets(
+	ctx context.Context,
+	name,
+	namespace string,
+) (map[string]string, error) {
+	if name == "" || namespace == "" {
+		return nil, errors.New("secret name or secret namespace is empty")
+	}
+	secret := &corev1.Secret{}
+	err := r.client.Get(ctx, types.NamespacedName{Name: name, Namespace: namespace}, secret)
+	if err != nil {
+		return nil, fmt.Errorf("error getting secret %s in namespace %s: %w", name, namespace, err)
+	}
+
+	secrets := map[string]string{}
+	for key, value := range secret.Data {
+		secrets[key] = string(value)
+	}
+
+	return secrets, nil
+}
+
+func (r *ReconcileVGRContent) reconcileVGRContent(ctx context.Context, obj runtime.Object) error {
+	vgrc, ok := obj.(*replicationv1alpha1.VolumeGroupReplicationContent)
+	if !ok {
+		return nil
+	}
+	if vgrc.Spec.Provisioner != r.config.DriverName {
+		return nil
+	}
+
+	reqName := vgrc.Name
+	groupHandle := vgrc.Spec.VolumeGroupReplicationHandle
+	volumeIds := vgrc.Spec.Source.VolumeHandles
+
+	if groupHandle == "" {
+		return errors.New("volume group replication handle is empty")
+	}
+
+	vgrClass := &replicationv1alpha1.VolumeGroupReplicationClass{}
+	err := r.client.Get(ctx, types.NamespacedName{Name: vgrc.Spec.VolumeGroupReplicationClassName}, vgrClass)
+	if err != nil {
+		return err
+	}
+
+	if ok = r.Locks.TryAcquire(groupHandle); !ok {
+		return fmt.Errorf("failed to acquire lock for group handle %s", groupHandle)
+	}
+	defer r.Locks.Release(groupHandle)
+
+	parameters := vgrClass.Spec.Parameters
+	secretName := vgrClass.Spec.Parameters[secretNameParameterName]
+	secretNamespace := vgrClass.Spec.Parameters[secretNamespaceParameterName]
+
+	secrets, err := r.getSecrets(ctx, secretName, secretNamespace)
+	if err != nil {
+		return err
+	}
+
+	mgr := rbd.NewManager(r.config.InstanceID, parameters, secrets)
+	defer mgr.Destroy(ctx)
+
+	groupID, err := mgr.RegenerateVolumeGroupJournal(ctx, groupHandle, reqName, volumeIds)
+	if err != nil {
+		return err
+	}
+	if groupID != groupHandle {
+		log.DebugLog(ctx, "groupHandle changed from %s to %s", groupHandle, groupID)
+	}
+
+	return nil
+}
+
+// Reconcile reconciles the VolumeGroupReplicationContent object and creates a new omap entries
+// for the volume group.
+func (r *ReconcileVGRContent) Reconcile(ctx context.Context,
+	request reconcile.Request,
+) (reconcile.Result, error) {
+	vgrc := &replicationv1alpha1.VolumeGroupReplicationContent{}
+	err := r.client.Get(ctx, request.NamespacedName, vgrc)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return reconcile.Result{}, nil
+		}
+
+		return reconcile.Result{}, err
+	}
+	// Check if the object is under deletion
+	if !vgrc.GetDeletionTimestamp().IsZero() {
+		return reconcile.Result{}, nil
+	}
+
+	err = r.reconcileVGRContent(ctx, vgrc)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+
+	return reconcile.Result{}, nil
+}

internal/journal/volumegroupjournal.go

@@ -54,6 +54,7 @@ type VolumeGroupJournal interface {
 		ctx context.Context,
 		journalPool,
 		reqName,
+		groupUUID,
 		namePrefix string) (string, string, error)
 	// AddVolumesMapping adds a volumeMap map which contains volumeID's and its
 	// corresponding values mapping which need to be added to the UUID
@@ -312,6 +313,7 @@ held, to prevent parallel operations from modifying the state of the omaps for t
 Input arguments:
   - journalPool: Pool where the CSI journal is stored
   - reqName: Name of the volumeGroupSnapshot request received
+  - groupUUID: UUID need to be reserved instead of auto-generating one (this is useful for RBD mirroring)
   - namePrefix: Prefix to use when generating the volumeGroupName name (suffix is an auto-generated UUID)
 
 Return values:
@@ -320,7 +322,7 @@ Return values:
   - error: non-nil in case of any errors
 */
 func (vgjc *volumeGroupJournalConnection) ReserveName(ctx context.Context,
-	journalPool, reqName, namePrefix string,
+	journalPool, reqName, groupUUID, namePrefix string,
 ) (string, string, error) {
 	cj := vgjc.config
 
@@ -335,7 +337,7 @@ func (vgjc *volumeGroupJournalConnection) ReserveName(ctx context.Context,
 		journalPool,
 		cj.namespace,
 		cj.cephUUIDDirectoryPrefix,
-		"")
+		groupUUID)
 	if err != nil {
 		return "", "", err
 	}