From d2a1286606ae345ee85dbd08c4fbfeb8fe601c31 Mon Sep 17 00:00:00 2001 From: subhamkrai Date: Tue, 9 Sep 2025 16:52:33 +0530 Subject: [PATCH] csi: add multus network annotation to csi-operator this commit adds multus network annotation to the csi-operator node plugin and controller plugin pods when multus is enabled in the ceph cluster. Signed-off-by: subhamkrai --- pkg/operator/ceph/csi/operator_driver.go | 38 +++- pkg/operator/ceph/csi/operator_driver_test.go | 168 ++++++++++++------ 2 files changed, 143 insertions(+), 63 deletions(-) diff --git a/pkg/operator/ceph/csi/operator_driver.go b/pkg/operator/ceph/csi/operator_driver.go index a4ddc80ef..3f065ef54 100644 --- a/pkg/operator/ceph/csi/operator_driver.go +++ b/pkg/operator/ceph/csi/operator_driver.go @@ -27,6 +27,7 @@ import ( "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" + "github.com/rook/rook/pkg/operator/k8sutil" v1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" k8scsiv1 "k8s.io/api/storage/v1" @@ -76,7 +77,7 @@ func (r *ReconcileCSI) createOrUpdateDriverResources(cluster cephv1.CephCluster, func (r *ReconcileCSI) createOrUpdateRBDDriverResource(cluster cephv1.CephCluster, clusterInfo *cephclient.ClusterInfo) error { resourceName := fmt.Sprintf("%s.rbd.csi.ceph.com", r.opConfig.OperatorNamespace) - spec, err := r.generateDriverSpec(cluster.Name) + spec, err := r.generateDriverSpec(cluster) if err != nil { return err } @@ -122,7 +123,7 @@ func (r *ReconcileCSI) createOrUpdateRBDDriverResource(cluster cephv1.CephCluste func (r *ReconcileCSI) createOrUpdateCephFSDriverResource(cluster cephv1.CephCluster, clusterInfo *cephclient.ClusterInfo) error { resourceName := fmt.Sprintf("%s.cephfs.csi.ceph.com", r.opConfig.OperatorNamespace) - spec, err := r.generateDriverSpec(cluster.Name) + spec, err := r.generateDriverSpec(cluster) if err != nil { return err } @@ -173,7 +174,7 @@ func (r *ReconcileCSI) createOrUpdateCephFSDriverResource(cluster cephv1.CephClu func (r *ReconcileCSI) createOrUpdateNFSDriverResource(cluster cephv1.CephCluster, clusterInfo *cephclient.ClusterInfo) error { resourceName := fmt.Sprintf("%s.nfs.csi.ceph.com", r.opConfig.OperatorNamespace) - spec, err := r.generateDriverSpec(cluster.Name) + spec, err := r.generateDriverSpec(cluster) if err != nil { return err } @@ -242,7 +243,7 @@ func (r ReconcileCSI) createOrUpdateDriverResource(clusterInfo *cephclient.Clust return nil } -func (r *ReconcileCSI) generateDriverSpec(clusterName string) (csiopv1.DriverSpec, error) { +func (r *ReconcileCSI) generateDriverSpec(cluster cephv1.CephCluster) (csiopv1.DriverSpec, error) { cephfsClientType := csiopv1.KernelCephFsClient if CSIParam.ForceCephFSKernelClient == "false" { cephfsClientType = csiopv1.AutoDetectCephFsClient @@ -252,14 +253,14 @@ func (r *ReconcileCSI) generateDriverSpec(clusterName string) (csiopv1.DriverSpe return csiopv1.DriverSpec{}, errors.Wrapf(err, "failed to create ceph-CSI operator config ImageSetConfigmap for CR %s", opConfigCRName) } - return csiopv1.DriverSpec{ + driverSpec := csiopv1.DriverSpec{ Log: &csiopv1.LogSpec{ Verbosity: int(CSIParam.LogLevel), }, ImageSet: &corev1.LocalObjectReference{ Name: imageSetCmName, }, - ClusterName: &clusterName, + ClusterName: &cluster.Name, EnableMetadata: &CSIParam.CSIEnableMetadata, GenerateOMapInfo: &CSIParam.EnableOMAPGenerator, FsGroupPolicy: k8scsiv1.FileFSGroupPolicy, @@ -288,7 +289,30 @@ func (r *ReconcileCSI) generateDriverSpec(clusterName string) (csiopv1.DriverSpe }, DeployCsiAddons: &CSIParam.EnableCSIAddonsSideCar, CephFsClientType: cephfsClientType, - }, nil + } + + // Apply Multus annotations if the cluster uses Multus networking + if cluster.Spec.Network.IsMultus() { + // Apply Multus to ControllerPlugin only + if driverSpec.ControllerPlugin.PodCommonSpec.Annotations == nil { + driverSpec.ControllerPlugin.PodCommonSpec.Annotations = make(map[string]string) + } + controllerObjectMeta := metav1.ObjectMeta{ + Annotations: driverSpec.ControllerPlugin.PodCommonSpec.Annotations, + } + err = r.applyMultusToObjectMeta(cluster.Namespace, &cluster.Spec.Network, &controllerObjectMeta) + if err != nil { + return csiopv1.DriverSpec{}, errors.Wrap(err, "failed to apply multus configuration to ControllerPlugin") + } + driverSpec.ControllerPlugin.PodCommonSpec.Annotations = controllerObjectMeta.Annotations + } + + return driverSpec, nil +} + +// applyMultusToObjectMeta applies Multus network annotations to the given ObjectMeta +func (r *ReconcileCSI) applyMultusToObjectMeta(clusterNamespace string, netSpec *cephv1.NetworkSpec, objectMeta *metav1.ObjectMeta) error { + return k8sutil.ApplyMultus(clusterNamespace, netSpec, objectMeta) } func createDriverControllerPluginResources(key string) csiopv1.ControllerPluginResourcesSpec { diff --git a/pkg/operator/ceph/csi/operator_driver_test.go b/pkg/operator/ceph/csi/operator_driver_test.go index e101f906e..9218a12f0 100644 --- a/pkg/operator/ceph/csi/operator_driver_test.go +++ b/pkg/operator/ceph/csi/operator_driver_test.go @@ -38,70 +38,126 @@ import ( ) func TestReconcileCSI_createOrUpdateDriverResources(t *testing.T) { - ns := "test" - r := &ReconcileCSI{ - context: &clusterd.Context{ - Clientset: testop.New(t, 1), - RookClientset: rookclient.NewSimpleClientset(), - }, - opManagerContext: context.TODO(), - opConfig: opcontroller.OperatorConfig{ - OperatorNamespace: "test", - }, - } - cluster := &cephv1.CephCluster{ - ObjectMeta: metav1.ObjectMeta{ - Name: "testCluster", - Namespace: ns, - }, - Spec: cephv1.ClusterSpec{ - CSI: cephv1.CSIDriverSpec{ - ReadAffinity: cephv1.ReadAffinitySpec{ - Enabled: true, - CrushLocationLabels: []string{"kubernetes.io/hostname"}, - }, - CephFS: cephv1.CSICephFSSpec{ - KernelMountOptions: "ms_mode=crc", - }, - }, - Network: cephv1.NetworkSpec{ + tests := []struct { + name string + networkSpec cephv1.NetworkSpec + expectMultusAnnots bool + }{ + { + name: "without multus", + networkSpec: cephv1.NetworkSpec{ Connections: &cephv1.ConnectionsSpec{ Encryption: &cephv1.EncryptionSpec{ Enabled: true, }, }, }, + expectMultusAnnots: false, + }, + { + name: "with multus", + networkSpec: cephv1.NetworkSpec{ + Provider: "multus", + Selectors: map[cephv1.CephNetworkType]string{ + "public": "rook-ceph/rook-ceph-public-network", + "cluster": "rook-ceph/rook-ceph-cluster-network", + }, + }, + expectMultusAnnots: true, }, } - driver := &csiopv1.Driver{} - // Register operator types with the runtime scheme. - s := scheme.Scheme - s.AddKnownTypes(cephv1.SchemeGroupVersion, driver, &cephv1.CephCluster{}, &v1.ConfigMap{}) - object := []runtime.Object{ - cluster, + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ns := "test" + r := &ReconcileCSI{ + context: &clusterd.Context{ + Clientset: testop.New(t, 1), + RookClientset: rookclient.NewSimpleClientset(), + }, + opManagerContext: context.TODO(), + opConfig: opcontroller.OperatorConfig{ + OperatorNamespace: "test", + }, + } + cluster := &cephv1.CephCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: "testCluster", + Namespace: ns, + }, + Spec: cephv1.ClusterSpec{ + CSI: cephv1.CSIDriverSpec{ + ReadAffinity: cephv1.ReadAffinitySpec{ + Enabled: true, + CrushLocationLabels: []string{"kubernetes.io/hostname"}, + }, + CephFS: cephv1.CSICephFSSpec{ + KernelMountOptions: "ms_mode=crc", + }, + }, + Network: tt.networkSpec, + }, + } + driver := &csiopv1.Driver{} + + // Register operator types with the runtime scheme. + s := scheme.Scheme + s.AddKnownTypes(cephv1.SchemeGroupVersion, driver, &cephv1.CephCluster{}, &v1.ConfigMap{}) + object := []runtime.Object{ + cluster, + } + cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() + r.client = cl + + EnableRBD = true + EnableCephFS = true + EnableNFS = true + + c := clienttest.CreateTestClusterInfo(3) + c.Namespace = ns + c.SetName("testcluster") + c.NamespacedName() + + err := r.createOrUpdateDriverResources(*cluster, c) + assert.NoError(t, err) + + // Test RBD driver + err = cl.Get(context.TODO(), types.NamespacedName{Name: fmt.Sprintf("%s.rbd.csi.ceph.com", c.Namespace), Namespace: ns}, driver) + assert.NoError(t, err) + + if tt.expectMultusAnnots { + // Verify that both NodePlugin and ControllerPlugin have Multus annotations + assert.Nil(t, driver.Spec.NodePlugin.PodCommonSpec.Annotations) + assert.NotContains(t, driver.Spec.NodePlugin.PodCommonSpec.Annotations, "k8s.v1.cni.cncf.io/networks") + + assert.NotNil(t, driver.Spec.ControllerPlugin.PodCommonSpec.Annotations) + assert.Contains(t, driver.Spec.ControllerPlugin.PodCommonSpec.Annotations, "k8s.v1.cni.cncf.io/networks") + + // Verify the annotation value contains the public network only not the cluster network + nodeMultusAnnotation := driver.Spec.NodePlugin.PodCommonSpec.Annotations["k8s.v1.cni.cncf.io/networks"] + assert.NotContains(t, nodeMultusAnnotation, "rook-ceph-public-network") + assert.NotContains(t, nodeMultusAnnotation, "rook-ceph-cluster-network") + + controllerMultusAnnotation := driver.Spec.ControllerPlugin.PodCommonSpec.Annotations["k8s.v1.cni.cncf.io/networks"] + assert.Contains(t, controllerMultusAnnotation, "rook-ceph-public-network") + assert.NotContains(t, controllerMultusAnnotation, "rook-ceph-cluster-network") + } else { + // Verify no Multus annotations when not using Multus + if driver.Spec.NodePlugin.PodCommonSpec.Annotations != nil { + assert.NotContains(t, driver.Spec.NodePlugin.PodCommonSpec.Annotations, "k8s.v1.cni.cncf.io/networks") + } + if driver.Spec.ControllerPlugin.PodCommonSpec.Annotations != nil { + assert.NotContains(t, driver.Spec.ControllerPlugin.PodCommonSpec.Annotations, "k8s.v1.cni.cncf.io/networks") + } + } + + // Test CephFS driver + err = cl.Get(context.TODO(), types.NamespacedName{Name: fmt.Sprintf("%s.cephfs.csi.ceph.com", c.Namespace), Namespace: ns}, driver) + assert.NoError(t, err) + + // Test NFS driver + err = cl.Get(context.TODO(), types.NamespacedName{Name: fmt.Sprintf("%s.nfs.csi.ceph.com", c.Namespace), Namespace: ns}, driver) + assert.NoError(t, err) + }) } - cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build() - r.client = cl - - EnableRBD = true - EnableCephFS = true - EnableNFS = true - - c := clienttest.CreateTestClusterInfo(3) - c.Namespace = ns - c.SetName("testcluster") - c.NamespacedName() - - err := r.createOrUpdateDriverResources(*cluster, c) - assert.NoError(t, err) - - err = cl.Get(context.TODO(), types.NamespacedName{Name: fmt.Sprintf("%s.rbd.csi.ceph.com", c.Namespace), Namespace: ns}, driver) - assert.NoError(t, err) - - err = cl.Get(context.TODO(), types.NamespacedName{Name: fmt.Sprintf("%s.cephfs.csi.ceph.com", c.Namespace), Namespace: ns}, driver) - assert.NoError(t, err) - - err = cl.Get(context.TODO(), types.NamespacedName{Name: fmt.Sprintf("%s.nfs.csi.ceph.com", c.Namespace), Namespace: ns}, driver) - assert.NoError(t, err) }