Repository: rook/rook
Pull request #6143 (2020-08-24 08:48:51)

ceph: added support for multus for csi (bp #5740)
This is an automated backport of pull request #5740 done by Mergify
---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://doc.mergify.io/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.io/) you can:

- look at your merge queues
- generate the Mergify configuration with the simulator.

Finally, you can contact us on https://mergify.io/
</details>



Patch:
diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md
index 7981d0d63..2f8f5204d 100755
--- a/Documentation/ceph-cluster-crd.md
+++ b/Documentation/ceph-cluster-crd.md
@@ -219,6 +219,7 @@ To use host networking, set `provider: host`.
 #### Multus (EXPERIMENTAL)
 
 Rook has experimental support for Multus.
+Currently there is an [open issue](https://github.com/ceph/ceph-csi/issues/1323) in ceph-csi which explains the csi-rbdPlugin issue while using multus network.
 
 The selector keys are required to be `public` and `cluster` where each represent:
 
diff --git a/pkg/operator/ceph/csi/csi.go b/pkg/operator/ceph/csi/csi.go
index b0a58cd3d..b65e9fe5a 100644
--- a/pkg/operator/ceph/csi/csi.go
+++ b/pkg/operator/ceph/csi/csi.go
@@ -20,6 +20,7 @@ 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"
@@ -27,7 +28,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-func ValidateAndConfigureDrivers(clientset kubernetes.Interface, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) {
+func ValidateAndConfigureDrivers(clientset kubernetes.Interface, rookclientset rookclient.Interface, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) {
 	if !AllowUnsupported {
 		if err := validateCSIVersion(clientset, namespace, rookImage, securityAccount, ownerRef); err != nil {
 			logger.Errorf("invalid csi version. %+v", err)
@@ -37,7 +38,7 @@ func ValidateAndConfigureDrivers(clientset kubernetes.Interface, namespace, rook
 		logger.Info("Skipping csi version check, since unsupported versions are allowed")
 	}
 
-	if err := startDrivers(clientset, namespace, serverVersion, ownerRef); err != nil {
+	if err := startDrivers(clientset, rookclientset, namespace, serverVersion, ownerRef); err != nil {
 		logger.Errorf("failed to start Ceph csi drivers. %v", err)
 		return
 	}
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 58c34b75c..9c7d7bc70 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/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/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/operator/k8sutil/cmdreporter"
@@ -201,7 +202,7 @@ func ValidateCSIParam() error {
 	return nil
 }
 
-func startDrivers(clientset kubernetes.Interface, namespace string, ver *version.Info, ownerRef *metav1.OwnerReference) error {
+func startDrivers(clientset kubernetes.Interface, rookclientset rookclient.Interface, namespace string, ver *version.Info, ownerRef *metav1.OwnerReference) error {
 	var (
 		err                                                   error
 		rbdPlugin, cephfsPlugin                               *apps.DaemonSet
@@ -380,6 +381,13 @@ func startDrivers(clientset kubernetes.Interface, namespace string, ver *version
 		// apply resource request and limit to rbdplugin containers
 		applyResourcesToContainers(clientset, rbdPluginResource, &rbdPlugin.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&rbdPlugin.ObjectMeta, ownerRef)
+		multusApplied, 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)
+		}
+		if multusApplied {
+			rbdPlugin.Spec.Template.Spec.HostNetwork = false
+		}
 		err = k8sutil.CreateDaemonSet(csiRBDPlugin, namespace, clientset, rbdPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbdplugin daemonset: %+v", rbdPlugin)
@@ -408,6 +416,10 @@ func startDrivers(clientset kubernetes.Interface, namespace string, 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", rbdProvisionerDeployment)
+		}
 		err = k8sutil.CreateDeployment(clientset, csiRBDProvisioner, namespace, rbdProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start rbd provisioner deployment: %+v", rbdProvisionerDeployment)
@@ -428,6 +440,13 @@ func startDrivers(clientset kubernetes.Interface, namespace string, ver *version
 		// apply resource request and limit to cephfs plugin containers
 		applyResourcesToContainers(clientset, cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec)
 		k8sutil.SetOwnerRef(&cephfsPlugin.ObjectMeta, ownerRef)
+		multusApplied, err := applyCephClusterNetworkConfig(&cephfsPlugin.Spec.Template.ObjectMeta, rookclientset)
+		if err != nil {
+			return errors.Wrapf(err, "failed to apply network config to cephfs plugin daemonset: %+v", cephfsPlugin)
+		}
+		if multusApplied {
+			cephfsPlugin.Spec.Template.Spec.HostNetwork = false
+		}
 		err = k8sutil.CreateDaemonSet(csiCephFSPlugin, namespace, clientset, cephfsPlugin)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs plugin daemonset: %+v", cephfsPlugin)
@@ -457,6 +476,11 @@ func startDrivers(clientset kubernetes.Interface, namespace string, 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", cephfsProvisionerDeployment)
+		}
 		err = k8sutil.CreateDeployment(clientset, csiCephFSProvisioner, namespace, cephfsProvisionerDeployment)
 		if err != nil {
 			return errors.Wrapf(err, "failed to start cephfs provisioner deployment: %+v", cephfsProvisionerDeployment)
@@ -547,6 +571,25 @@ func deleteCSIDriverResources(
 	return succeeded
 }
 
+func applyCephClusterNetworkConfig(objectMeta *metav1.ObjectMeta, rookclientset rookclient.Interface) (bool, error) {
+	var isMultusApplied bool
+	cephClusters, err := rookclientset.CephV1().CephClusters(objectMeta.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		return false, errors.Errorf("failed to find CephClusters in namespace %q", objectMeta.Namespace)
+	}
+	for _, cephCluster := range cephClusters.Items {
+		if cephCluster.Spec.Network.IsMultus() {
+			err = k8sutil.ApplyMultus(cephCluster.Spec.Network.NetworkSpec, objectMeta)
+			if err != nil {
+				return false, errors.Wrapf(err, "failed to apply multus configuration to CephCluster %q", cephCluster.Name)
+			}
+			isMultusApplied = true
+		}
+	}
+
+	return isMultusApplied, nil
+}
+
 // createCSIDriverInfo Registers CSI driver by creating a CSIDriver object
 func createCSIDriverInfo(clientset kubernetes.Interface, name string, ownerRef *metav1.OwnerReference) error {
 	attach := true
diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go
index aa2ffc75a..b191d137c 100644
--- a/pkg/operator/ceph/operator.go
+++ b/pkg/operator/ceph/operator.go
@@ -251,7 +251,7 @@ func (o *Operator) updateDrivers() error {
 		return errors.Wrap(err, "invalid csi params")
 	}
 
-	go csi.ValidateAndConfigureDrivers(o.context.Clientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
+	go csi.ValidateAndConfigureDrivers(o.context.Clientset, o.context.RookClientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
 	return nil
 }
 
diff --git a/pkg/operator/k8sutil/network.go b/pkg/operator/k8sutil/network.go
index 0eb62860b..ba6584419 100644
--- a/pkg/operator/k8sutil/network.go
+++ b/pkg/operator/k8sutil/network.go
@@ -26,6 +26,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	// publicNetworkSelectorKeyName is the network selector key for the ceph public network
+	publicNetworkSelectorKeyName = "public"
+)
+
 // NetworkAttachmentConfig represents the configuration of the NetworkAttachmentDefinitions object
 type NetworkAttachmentConfig struct {
 	CniVersion string `json:"cniVersion"`
@@ -110,7 +115,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)
 
@@ -120,7 +125,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 == publicNetworkSelectorKeyName {
+				v = append(v, string(ns))
+			}
+		} else {
+			v = append(v, string(ns))
+		}
 	}
 
 	if shortSyntax && jsonSyntax {


Test patch:
diff --git a/pkg/operator/ceph/csi/spec_test.go b/pkg/operator/ceph/csi/spec_test.go
index 70108d751..4293ce6b7 100644
--- a/pkg/operator/ceph/csi/spec_test.go
+++ b/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(clientset, "ns", serverVersion, nil)
+	err = startDrivers(clientset, rookclientset, "ns", serverVersion, nil)
 	assert.Nil(t, err)
 }