ceph: added support for multus for csi

CSI pods now utilize multus networking and connect to public
network specified in the CephCluster CR.

Signed-off-by: rohan47 <rohgupta@redhat.com>

pkg/operator/ceph/csi/csi.go

@@ -20,20 +20,21 @@ import (
 	"strconv"
 
 	"github.com/pkg/errors"
+	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
 	controllerutil "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/version"
 	"k8s.io/client-go/kubernetes"
 )
 
-func ValidateAndStartDrivers(clientset kubernetes.Interface, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) {
+func ValidateAndStartDrivers(clientset kubernetes.Interface, rookclientset rookclient.Interface, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) {
 	if err := validateCSIVersion(clientset, namespace, rookImage, securityAccount, ownerRef); err != nil {
 		logger.Errorf("invalid csi version. %+v", err)
 		return
 	}
 
-	if err := startDrivers(namespace, clientset, serverVersion, ownerRef); err != nil {
+	if err := startDrivers(namespace, clientset, rookclientset, serverVersion, ownerRef); err != nil {
 		logger.Errorf("failed to start Ceph csi drivers. %v", err)
 		return
 	}

pkg/operator/ceph/csi/spec.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 	"time"
 
+	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
 	controllerutil "github.com/rook/rook/pkg/operator/ceph/controller"
 
 	"github.com/pkg/errors"
@@ -196,7 +197,7 @@ func ValidateCSIParam() error {
 	return nil
 }
 
-func startDrivers(namespace string, clientset kubernetes.Interface, ver *version.Info, ownerRef *metav1.OwnerReference) error {
+func startDrivers(namespace string, clientset kubernetes.Interface, rookclientset rookclient.Interface, ver *version.Info, ownerRef *metav1.OwnerReference) error {
 	var (
 		err                                                   error
 		rbdPlugin, cephfsPlugin                               *apps.DaemonSet
@@ -378,6 +379,10 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version
 		// apply resource request and limit to rbdplugin containers
 		applyResourcesToContainers(clientset, rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&rbdPlugin.ObjectMeta, ownerRef)
+		err = applyCephClusterNetworkConfig(&rbdPlugin.Spec.Template.ObjectMeta, rookclientset)
+		if err != nil {
+			return errors.Wrapf(err, "failed to apply network config to rbd plugin daemonset: %+v", rbdPlugin)
+		}
 		err = k8sutil.CreateDaemonSet("csi-rbdplugin", namespace, clientset, rbdPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbdplugin daemonset: %+v", rbdPlugin)
@@ -406,6 +411,10 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version
 			Type: apps.RecreateDeploymentStrategyType,
 		}
 
+		err = applyCephClusterNetworkConfig(&rbdProvisionerDeployment.Spec.Template.ObjectMeta, rookclientset)
+		if err != nil {
+			return errors.Wrapf(err, "failed to apply network config to rbd plugin provisioner deployment: %+v", rbdPlugin)
+		}
 		err = k8sutil.CreateDeployment(clientset, "csi-rbdplugin-provisioner", namespace, rbdProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbd provisioner deployment: %+v", rbdProvisionerDeployment)
@@ -426,6 +435,10 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version
 		// apply resource request and limit to cephfs plugin containers
 		applyResourcesToContainers(clientset, cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&cephfsPlugin.ObjectMeta, ownerRef)
+		err = applyCephClusterNetworkConfig(&cephfsPlugin.Spec.Template.ObjectMeta, rookclientset)
+		if err != nil {
+			return errors.Wrapf(err, "failed to apply network config to cephfs plugin daemonset: %+v", rbdPlugin)
+		}
 		err = k8sutil.CreateDaemonSet("csi-cephfsplugin", namespace, clientset, cephfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs plugin daemonset: %+v", cephfsPlugin)
@@ -455,6 +468,11 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version
 		cephfsProvisionerDeployment.Spec.Strategy = apps.DeploymentStrategy{
 			Type: apps.RecreateDeploymentStrategyType,
 		}
+
+		err = applyCephClusterNetworkConfig(&cephfsProvisionerDeployment.Spec.Template.ObjectMeta, rookclientset)
+		if err != nil {
+			return errors.Wrapf(err, "failed to apply network config to cephfs plugin provisioner deployment: %+v", rbdPlugin)
+		}
 		err = k8sutil.CreateDeployment(clientset, "csi-cephfsplugin-provisioner", namespace, cephfsProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs provisioner deployment: %+v", cephfsProvisionerDeployment)
@@ -482,6 +500,20 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version
 	return nil
 }
 
+func applyCephClusterNetworkConfig(objectMeta *metav1.ObjectMeta, rookclientset rookclient.Interface) error {
+	cephClusters, err := rookclientset.CephV1().CephClusters(objectMeta.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		return err
+	}
+	for _, cephCluster := range cephClusters.Items {
+		if cephCluster.Spec.Network.IsMultus() {
+			k8sutil.ApplyMultus(cephCluster.Spec.Network.NetworkSpec, objectMeta)
+		}
+	}
+
+	return nil
+}
+
 // createCSIDriverInfo Registers CSI driver by creating a CSIDriver object
 func createCSIDriverInfo(clientset kubernetes.Interface, name string, ownerRef *metav1.OwnerReference) error {
 	attach := true

pkg/operator/ceph/csi/spec_test.go

@@ -19,6 +19,7 @@ package csi
 import (
 	"testing"
 
+	rookclient "github.com/rook/rook/pkg/client/clientset/versioned"
 	"github.com/rook/rook/pkg/operator/test"
 
 	"github.com/stretchr/testify/assert"
@@ -40,10 +41,11 @@ func TestStartCSI(t *testing.T) {
 		SnapshotterImage: "image",
 	}
 	clientset := test.New(t, 3)
+	var rookclientset rookclient.Interface
 	serverVersion, err := clientset.Discovery().ServerVersion()
 	if err != nil {
 		assert.Nil(t, err)
 	}
-	err = startDrivers("ns", clientset, serverVersion, nil)
+	err = startDrivers("ns", clientset, rookclientset, serverVersion, nil)
 	assert.Nil(t, err)
 }

pkg/operator/ceph/operator.go

@@ -256,7 +256,7 @@ func (o *Operator) updateDrivers() error {
 		return errors.Wrap(err, "invalid csi params")
 	}
 
-	go csi.ValidateAndStartDrivers(o.context.Clientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
+	go csi.ValidateAndStartDrivers(o.context.Clientset, o.context.RookClientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
 	return nil
 }
 

pkg/operator/k8sutil/network.go

@@ -109,7 +109,7 @@ func ApplyMultus(net rookv1.NetworkSpec, objectMeta *metav1.ObjectMeta) error {
 	shortSyntax := false
 	jsonSyntax := false
 
-	for _, ns := range net.Selectors {
+	for k, ns := range net.Selectors {
 		var multusMap map[string]string
 		err := json.Unmarshal([]byte(ns), &multusMap)
 
@@ -119,7 +119,14 @@ func ApplyMultus(net rookv1.NetworkSpec, objectMeta *metav1.ObjectMeta) error {
 			shortSyntax = true
 		}
 
-		v = append(v, string(ns))
+		isCsi := strings.Contains(objectMeta.Labels["app"], "csi-")
+		if isCsi {
+			if k == "public" {
+				v = append(v, string(ns))
+			}
+		} else {
+			v = append(v, string(ns))
+		}
 	}
 
 	if shortSyntax && jsonSyntax {