add cg controllers for cephfs

cmd/main.go

@@ -179,13 +179,9 @@ func setupReconcilersCluster(mgr ctrl.Manager, ramenConfig *ramendrv1alpha1.Rame
 		os.Exit(1)
 	}
 
-	fsCGSupport, err := rmnutil.IsFSCGSupport(mgr.GetConfig(), mgr.GetScheme())
-	if err != nil {
-		setupLog.Error(err, "failed to check if ceph fs consistency group is supported")
-		os.Exit(1)
-	}
+	if !ramenConfig.VolSync.Disabled {
+		setupLog.Info("VolSync enabled, setup ReplicationGroupSource and ReplicationGroupDestination controllers")
 
-	if fsCGSupport {
 		if err := (&controllers.ReplicationGroupDestinationReconciler{
 			Client: mgr.GetClient(),
 			Scheme: mgr.GetScheme(),

internal/controller/replicationgroupsource_controller.go

@@ -5,9 +5,11 @@ package controllers
 
 import (
 	"context"
+	"fmt"
 
 	vgsv1alphfa1 "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumegroupsnapshot/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -53,24 +55,32 @@ ReplicationGroupDestination will create application PVC which is the same with c
 // ReplicationGroupSourceReconciler reconciles a ReplicationGroupSource object
 type ReplicationGroupSourceReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
+	Scheme                           *runtime.Scheme
+	volumeGroupSnapshotCRsAreWatched bool
 }
 
-//+kubebuilder:rbac:groups=ramendr.openshift.io,resources=replicationgroupsources,verbs=get;list;watch;create;update;patch;delete
-//+kubebuilder:rbac:groups=ramendr.openshift.io,resources=replicationgroupsources/status,verbs=get;update;patch
-//+kubebuilder:rbac:groups=ramendr.openshift.io,resources=replicationgroupsources/finalizers,verbs=update
-//+kubebuilder:rbac:groups=groupsnapshot.storage.k8s.io,resources=volumegroupsnapshots,verbs=get;list;watch;create;update;patch;delete
-//+kubebuilder:rbac:groups=groupsnapshot.storage.k8s.io,resources=volumegroupsnapshotclasses,verbs=get;list;watch
-//+kubebuilder:rbac:groups=groupsnapshot.storage.k8s.io,resources=volumegroupsnapshotcontents,verbs=get;list;watch
-//+kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch;create;update;patch;delete
-//+kubebuilder:rbac:groups=volsync.backube,resources=replicationsources,verbs=get;list;watch;create;update;patch;delete
-//+kubebuilder:rbac:groups=snapshot.storage.k8s.io,resources=volumesnapshots,verbs=get;list;watch
-//+kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get
-
+// +kubebuilder:rbac:groups=ramendr.openshift.io,resources=replicationgroupsources,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=ramendr.openshift.io,resources=replicationgroupsources/status,verbs=get;update;patch
+// +kubebuilder:rbac:groups=ramendr.openshift.io,resources=replicationgroupsources/finalizers,verbs=update
+// +kubebuilder:rbac:groups=groupsnapshot.storage.k8s.io,resources=volumegroupsnapshots,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=groupsnapshot.storage.k8s.io,resources=volumegroupsnapshotclasses,verbs=get;list;watch
+// +kubebuilder:rbac:groups=groupsnapshot.storage.k8s.io,resources=volumegroupsnapshotcontents,verbs=get;list;watch
+// +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=volsync.backube,resources=replicationsources,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=snapshot.storage.k8s.io,resources=volumesnapshots,verbs=get;list;watch
+// +kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get
+//
+//nolint:funlen
 func (r *ReplicationGroupSourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	logger := log.FromContext(ctx)
 	logger.Info("Get ReplicationGroupSource")
 
+	if !r.volumeGroupSnapshotCRsAreWatched {
+		return ctrl.Result{},
+			fmt.Errorf("ReplicationGroupSource {%s/%s} doesn't work if VolumeGroupSnapshot CRD is not installed. "+
+				"Please install VolumeGroupSnapshot CRD and restart the operator", req.Namespace, req.Name)
+	}
+
 	rgs := &ramendrv1alpha1.ReplicationGroupSource{}
 	if err := r.Client.Get(ctx, req.NamespacedName, rgs); err != nil {
 		if !errors.IsNotFound(err) {
@@ -129,13 +139,24 @@ func (r *ReplicationGroupSourceReconciler) Reconcile(ctx context.Context, req ct
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *ReplicationGroupSourceReconciler) SetupWithManager(mgr ctrl.Manager) error {
-	return ctrl.NewControllerManagedBy(mgr).
+	vgsCRD := &apiextensionsv1.CustomResourceDefinition{}
+	if err := r.Client.Get(context.TODO(),
+		types.NamespacedName{Name: "volumegroupsnapshots.groupsnapshot.storage.k8s.io"}, vgsCRD,
+	); err == nil {
+		r.volumeGroupSnapshotCRsAreWatched = true
+	}
+
+	builder := ctrl.NewControllerManagedBy(mgr).
 		WithOptions(ctrlcontroller.Options{
 			MaxConcurrentReconciles: getMaxConcurrentReconciles(ctrl.Log),
 		}).
-		Owns(&vgsv1alphfa1.VolumeGroupSnapshot{}).
 		Owns(&corev1.PersistentVolumeClaim{}).
 		Owns(&volsyncv1alpha1.ReplicationSource{}).
-		For(&ramendrv1alpha1.ReplicationGroupSource{}).
-		Complete(r)
+		For(&ramendrv1alpha1.ReplicationGroupSource{})
+
+	if r.volumeGroupSnapshotCRsAreWatched {
+		builder.Owns(&vgsv1alphfa1.VolumeGroupSnapshot{})
+	}
+
+	return builder.Complete(r)
 }

internal/controller/util/cephfs_cg.go

@@ -12,12 +12,9 @@ import (
 	ramendrv1alpha1 "github.com/ramendr/ramen/api/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
-	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/rest"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -30,27 +27,6 @@ const (
 	RGSOwnerLabel string = "ramendr.openshift.io/rgs"
 )
 
-func IsFSCGSupport(restConfig *rest.Config, scheme *runtime.Scheme) (bool, error) {
-	k8sClient, err := client.New(restConfig, client.Options{Scheme: scheme})
-	if err != nil {
-		return false, err
-	}
-
-	vgsCRD := &apiextensionsv1.CustomResourceDefinition{}
-	if err := k8sClient.Get(context.Background(),
-		types.NamespacedName{Name: "volumegroupsnapshots.groupsnapshot.storage.k8s.io"},
-		vgsCRD,
-	); err != nil {
-		if errors.IsNotFound(err) {
-			return false, nil
-		}
-
-		return false, err
-	}
-
-	return true, nil
-}
-
 func IsReplicationGroupDestinationReady(
 	ctx context.Context, k8sClient client.Client,
 	rgd *ramendrv1alpha1.ReplicationGroupDestination,

internal/controller/util/cephfs_cg_test.go

@@ -17,21 +17,11 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 var _ = Describe("CephfsCg", func() {
-	Describe("IsFSCGSupport", func() {
-		Context("There is volume group snapshot class", func() {
-			It("Should be true", func() {
-				isFSCGSupport, err := util.IsFSCGSupport(cfg, scheme.Scheme)
-				Expect(err).To(BeNil())
-				Expect(isFSCGSupport).To(BeTrue())
-			})
-		})
-	})
 	Describe("IsReplicationGroupDestinationReady", func() {
 		Describe("ReplicationGroupDestination is empty", func() {
 			It("Should be false", func() {

internal/controller/vrg_volsync.go

@@ -299,9 +299,9 @@ func (v *VRGInstance) reconcileRDSpecForDeletionOrReplication() bool {
 
 					requeue = true
 				}
-			}
 
-			rdinCGs = append(rdinCGs, rdinCG...)
+				rdinCGs = append(rdinCGs, rdinCG...)
+			}
 		}
 	}