Merge pull request #3957 from Madhu-1/tol-csi

Add Toleration and NodeAffinity to CSI provisioners

cluster/charts/rook-ceph/templates/deployment.yaml

@@ -134,6 +134,22 @@ spec:
           value: {{ .Values.csi.attacher.image | quote }}
 {{- end }}
 {{- end }}
+{{- if .Values.csi.provisionerTolerations }}
+        - name: CSI_PROVISIONER_TOLERATIONS
+          value: {{ toYaml .Values.csi.provisionerTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.provisionerNodeAffinity }}
+        - name: CSI_PROVISIONER_NODE_AFFINITY
+          value: {{ .Values.csi.provisionerNodeAffinity }}
+{{- end }}
+{{- if .Values.csi.pluginTolerations }}
+        - name: CSI_PLUGIN_TOLERATIONS
+          value: {{ toYaml .Values.csi.pluginTolerations | quote }}
+{{- end }}
+{{- if .Values.csi.pluginNodeAffinity }}
+        - name: CSI_PLUGIN_NODE_AFFINITY
+          value: {{ .Values.csi.pluginNodeAffinity }}
+{{- end }}
 {{- end }}
         - name: ROOK_ENABLE_FLEX_DRIVER
           value: "{{ .Values.enableFlexDriver }}"

cluster/charts/rook-ceph/values.yaml

@@ -59,6 +59,20 @@ csi:
   enableRbdDriver: true
   enableCephfsDriver: true
   enableGrpcMetrics: true
+  # Set provisonerTolerations and provisionerNodeAffinity for provisioner pod.
+  # The CSI provisioner would be best to start on the same nodes as other ceph daemons.
+  # provisionerTolerations:
+  #    - key: key
+  #      operator: Exists
+  #      effect: NoSchedule
+  # provisionerNodeAffinity: key1=value1,value2; key2=value3
+  # Set pluginTolerations and pluginNodeAffinity for plugin daemonset pods.
+  # The CSI plugins need to be started on all the nodes where the clients need to mount the storage.
+  # pluginTolerations:
+  #    - key: key
+  #      operator: Exists
+  #      effect: NoSchedule
+  # pluginNodeAffinity: key1=value1,value2; key2=value3
   #kubeletDirPath: /var/lib/kubelet
   #cephcsi:
     #image: quay.io/cephcsi/cephcsi:v1.2.1

cluster/examples/kubernetes/ceph/operator-openshift.yaml

@@ -222,6 +222,32 @@ spec:
         # kubelet directory path, if kubelet configured to use other than /var/lib/kubelet path.
         #- name: ROOK_CSI_KUBELET_DIR_PATH
         #  value: "/var/lib/kubelet"
+        # (Optional) Ceph Provisioner NodeAffinity.
+        # - name: CSI_PROVISIONER_NODE_AFFINITY
+        #   value: "role=storage-node; storage=rook, ceph"
+        # (Optional) CEPH CSI provisioner tolerations list. Put here list of taints you want to tolerate in YAML format.
+        # CSI provisioner would be best to start on the same nodes as other ceph daemons.
+        # - name: CSI_PROVISIONER_TOLERATIONS
+        #   value: |
+        #     - effect: NoSchedule
+        #       key: node-role.kubernetes.io/controlplane
+        #       operator: Exists
+        #     - effect: NoExecute
+        #       key: node-role.kubernetes.io/etcd
+        #       operator: Exists
+        # (Optional) Ceph CSI plugin NodeAffinity.
+        # - name: CSI_PLUGIN_NODE_AFFINITY
+        #   value: "role=storage-node; storage=rook, ceph"
+        # (Optional) CEPH CSI plugin tolerations list. Put here list of taints you want to tolerate in YAML format.
+        # CSI plugins need to be started on all the nodes where the clients need to mount the storage.
+        # - name: CSI_PLUGIN_TOLERATIONS
+        #   value: |
+        #     - effect: NoSchedule
+        #       key: node-role.kubernetes.io/controlplane
+        #       operator: Exists
+        #     - effect: NoExecute
+        #       key: node-role.kubernetes.io/etcd
+        #       operator: Exists
         - name: ROOK_HOSTPATH_REQUIRES_PRIVILEGED
           value: "true"
         # The name of the node to pass with the downward API

cluster/examples/kubernetes/ceph/operator.yaml

@@ -184,6 +184,32 @@ spec:
         # kubelet directory path, if kubelet configured to use other than /var/lib/kubelet path.
         #- name: ROOK_CSI_KUBELET_DIR_PATH
         #  value: "/var/lib/kubelet"
+        # (Optional) Ceph Provisioner NodeAffinity.
+        # - name: CSI_PROVISIONER_NODE_AFFINITY
+        #   value: "role=storage-node; storage=rook, ceph"
+        # (Optional) CEPH CSI provisioner tolerations list. Put here list of taints you want to tolerate in YAML format.
+        #  CSI provisioner would be best to start on the same nodes as other ceph daemons.
+        # - name: CSI_PROVISIONER_TOLERATIONS
+        #   value: |
+        #     - effect: NoSchedule
+        #       key: node-role.kubernetes.io/controlplane
+        #       operator: Exists
+        #     - effect: NoExecute
+        #       key: node-role.kubernetes.io/etcd
+        #       operator: Exists
+        # (Optional) Ceph CSI plugin NodeAffinity.
+        # - name: CSI_PLUGIN_NODE_AFFINITY
+        #   value: "role=storage-node; storage=rook, ceph"
+        # (Optional) CEPH CSI plugin tolerations list. Put here list of taints you want to tolerate in YAML format.
+        # CSI plugins need to be started on all the nodes where the clients need to mount the storage.
+        # - name: CSI_PLUGIN_TOLERATIONS
+        #   value: |
+        #     - effect: NoSchedule
+        #       key: node-role.kubernetes.io/controlplane
+        #       operator: Exists
+        #     - effect: NoExecute
+        #       key: node-role.kubernetes.io/etcd
+        #       operator: Exists
         # The name of the node to pass with the downward API
         - name: NODE_NAME
           valueFrom:

pkg/apis/rook.io/v1alpha2/placement_test.go

@@ -21,7 +21,7 @@ import (
 
 	"github.com/ghodss/yaml"
 	"github.com/stretchr/testify/assert"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 )
 
 func TestPlacement_spec(t *testing.T) {
@@ -76,7 +76,7 @@ tolerations:
 
 func TestPlacement_ApplyToPodSpec(t *testing.T) {
 	to := placementTestGetTolerations("foo", "bar")
-	na := placementTestGetNodeAffinity()
+	na := placementTestGenerateNodeAffinity()
 	expected := &v1.PodSpec{Affinity: &v1.Affinity{NodeAffinity: na}, Tolerations: to}
 
 	var p Placement
@@ -100,7 +100,7 @@ func TestPlacement_ApplyToPodSpec(t *testing.T) {
 	assert.Equal(t, expected, ps)
 
 	p = Placement{NodeAffinity: na}
-	nap := placementTestGetNodeAffinity()
+	nap := placementTestGenerateNodeAffinity()
 	nap.PreferredDuringSchedulingIgnoredDuringExecution[0].Weight = 5
 	ps = &v1.PodSpec{Affinity: &v1.Affinity{NodeAffinity: nap}, Tolerations: to}
 	p.ApplyToPodSpec(ps)
@@ -109,7 +109,7 @@ func TestPlacement_ApplyToPodSpec(t *testing.T) {
 
 func TestPlacement_Merge(t *testing.T) {
 	to := placementTestGetTolerations("foo", "bar")
-	na := placementTestGetNodeAffinity()
+	na := placementTestGenerateNodeAffinity()
 
 	var original, with, expected, merged Placement
 
@@ -145,7 +145,7 @@ func placementTestGetTolerations(key, value string) []v1.Toleration {
 	}
 }
 
-func placementTestGetNodeAffinity() *v1.NodeAffinity {
+func placementTestGenerateNodeAffinity() *v1.NodeAffinity {
 	return &v1.NodeAffinity{
 		RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
 			NodeSelectorTerms: []v1.NodeSelectorTerm{

pkg/operator/ceph/agent/agent.go

@@ -28,7 +28,7 @@ import (
 	k8sutil "github.com/rook/rook/pkg/operator/k8sutil"
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
-	kserrors "k8s.io/apimachinery/pkg/api/errors"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -252,7 +252,7 @@ func (a *Agent) createAgentDaemonSet(namespace, agentImage, serviceAccount strin
 	// Add NodeAffinity if any
 	nodeAffinity := os.Getenv(agentDaemonsetNodeAffinityEnv)
 	if nodeAffinity != "" {
-		v1NodeAffinity, err := k8sutil.AddNodeAffinity(nodeAffinity)
+		v1NodeAffinity, err := k8sutil.GenerateNodeAffinity(nodeAffinity)
 		if err != nil {
 			logger.Errorf("failed to create NodeAffinity. %+v", err)
 		} else {
@@ -264,7 +264,7 @@ func (a *Agent) createAgentDaemonSet(namespace, agentImage, serviceAccount strin
 
 	_, err = a.clientset.AppsV1().DaemonSets(namespace).Create(ds)
 	if err != nil {
-		if !kserrors.IsAlreadyExists(err) {
+		if !k8serrors.IsAlreadyExists(err) {
 			return fmt.Errorf("failed to create rook-ceph-agent daemon set. %+v", err)
 		}
 		logger.Infof("rook-ceph-agent daemonset already exists, updating ...")

pkg/operator/ceph/csi/spec.go

@@ -82,6 +82,12 @@ const (
 	DefaultAttacherImage    = "quay.io/k8scsi/csi-attacher:v1.2.0"
 	DefaultSnapshotterImage = "quay.io/k8scsi/csi-snapshotter:v1.2.0"
 
+	// toleration and node affinity
+	provisionerTolerationsEnv  = "CSI_PROVISIONER_TOLERATIONS"
+	provisionerNodeAffinityEnv = "CSI_PROVISIONER_NODE_AFFINITY"
+	pluginTolerationsEnv       = "CSI_PLUGIN_TOLERATIONS"
+	pluginNodeAffinityEnv      = "CSI_PLUGIN_NODE_AFFINITY"
+
 	// kubelet directory path
 	DefaultKubeletDirPath = "/var/lib/kubelet"
 
@@ -213,19 +219,28 @@ func StartCSIDrivers(namespace string, clientset kubernetes.Interface, ver *vers
 			return fmt.Errorf("failed to load cephfs plugin service template: %+v", err)
 		}
 	}
-
+	// get provisioner toleration and node affinity
+	provisionerTolerations := getToleration(true)
+	provisionerNodeAffinity := getNodeAffinity(true)
+	// get plugin toleration and node affinity
+	pluginTolerations := getToleration(false)
+	pluginNodeAffinity := getNodeAffinity(false)
 	if rbdPlugin != nil {
+		applyToPodSpec(&rbdPlugin.Spec.Template.Spec, pluginNodeAffinity, pluginTolerations)
 		err = k8sutil.CreateDaemonSet("csi-rbdplugin", namespace, clientset, rbdPlugin)
 		if err != nil {
 			return fmt.Errorf("failed to start rbdplugin daemonset: %+v\n%+v", err, rbdPlugin)
 		}
 	}
+
 	if rbdProvisionerSTS != nil {
+		applyToPodSpec(&rbdProvisionerSTS.Spec.Template.Spec, provisionerNodeAffinity, provisionerTolerations)
 		err = k8sutil.CreateStatefulSet("csi-rbdplugin-provisioner", namespace, clientset, rbdProvisionerSTS)
 		if err != nil {
 			return fmt.Errorf("failed to start rbd provisioner statefulset: %+v\n%+v", err, rbdProvisionerSTS)
 		}
 	} else if rbdProvisionerDeployment != nil {
+		applyToPodSpec(&rbdProvisionerDeployment.Spec.Template.Spec, provisionerNodeAffinity, provisionerTolerations)
 		err = k8sutil.CreateDeployment("csi-rbdplugin-provisioner", namespace, clientset, rbdProvisionerDeployment)
 		if err != nil {
 			return fmt.Errorf("failed to start rbd provisioner deployment: %+v\n%+v", err, rbdProvisionerDeployment)
@@ -240,6 +255,7 @@ func StartCSIDrivers(namespace string, clientset kubernetes.Interface, ver *vers
 	}
 
 	if cephfsPlugin != nil {
+		applyToPodSpec(&cephfsPlugin.Spec.Template.Spec, pluginNodeAffinity, pluginTolerations)
 		err = k8sutil.CreateDaemonSet("csi-cephfsplugin", namespace, clientset, cephfsPlugin)
 		if err != nil {
 			return fmt.Errorf("failed to start cephfs plugin daemonset: %+v\n%+v", err, cephfsPlugin)
@@ -247,12 +263,14 @@ func StartCSIDrivers(namespace string, clientset kubernetes.Interface, ver *vers
 	}
 
 	if cephfsProvisionerSTS != nil {
+		applyToPodSpec(&cephfsProvisionerSTS.Spec.Template.Spec, provisionerNodeAffinity, provisionerTolerations)
 		err = k8sutil.CreateStatefulSet("csi-cephfsplugin-provisioner", namespace, clientset, cephfsProvisionerSTS)
 		if err != nil {
 			return fmt.Errorf("failed to start cephfs provisioner statefulset: %+v\n%+v", err, cephfsProvisionerSTS)
 		}
 
 	} else if cephfsProvisionerDeployment != nil {
+		applyToPodSpec(&cephfsProvisionerDeployment.Spec.Template.Spec, provisionerNodeAffinity, provisionerTolerations)
 		err = k8sutil.CreateDeployment("csi-cephfsplugin-provisioner", namespace, clientset, cephfsProvisionerDeployment)
 		if err != nil {
 			return fmt.Errorf("failed to start cephfs provisioner deployment: %+v\n%+v", err, cephfsProvisionerDeployment)

pkg/operator/ceph/csi/util.go

@@ -20,10 +20,12 @@ import (
 	"bytes"
 	"fmt"
 	"io/ioutil"
+	"os"
 	"text/template"
 
-	"github.com/ghodss/yaml"
+	k8sutil "github.com/rook/rook/pkg/operator/k8sutil"
 
+	"github.com/ghodss/yaml"
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -99,3 +101,56 @@ func templateToDeployment(name, templatePath string, p templateParam) (*apps.Dep
 	}
 	return &ds, nil
 }
+
+func getToleration(provisioner bool) []corev1.Toleration {
+	// Add toleration if any
+	tolerations := []corev1.Toleration{}
+	var err error
+	tolerationsRaw := ""
+	if provisioner {
+		tolerationsRaw = os.Getenv(provisionerTolerationsEnv)
+	} else {
+		tolerationsRaw = os.Getenv(pluginTolerationsEnv)
+	}
+	if tolerationsRaw != "" {
+		tolerations, err = k8sutil.YamlToTolerations(tolerationsRaw)
+		if err != nil {
+			logger.Warningf("failed to parse %s. %+v", tolerationsRaw, err)
+		}
+	}
+	for i := range tolerations {
+		if tolerations[i].Key == "" {
+			tolerations[i].Operator = corev1.TolerationOpExists
+		}
+
+		if tolerations[i].Operator == corev1.TolerationOpExists {
+			tolerations[i].Value = ""
+		}
+	}
+	return tolerations
+}
+
+func getNodeAffinity(provisioner bool) *corev1.NodeAffinity {
+	// Add NodeAffinity if any
+	nodeAffinity := ""
+	if provisioner {
+		nodeAffinity = os.Getenv(provisionerNodeAffinityEnv)
+	} else {
+		nodeAffinity = os.Getenv(pluginNodeAffinityEnv)
+	}
+	if nodeAffinity != "" {
+		v1NodeAffinity, err := k8sutil.GenerateNodeAffinity(nodeAffinity)
+		if err != nil {
+			logger.Warningf("failed to parse %s. %+v", nodeAffinity, err)
+		}
+		return v1NodeAffinity
+	}
+	return nil
+}
+
+func applyToPodSpec(pod *corev1.PodSpec, n *corev1.NodeAffinity, t []corev1.Toleration) {
+	pod.Tolerations = t
+	pod.Affinity = &corev1.Affinity{
+		NodeAffinity: n,
+	}
+}

pkg/operator/discover/discover.go

@@ -31,9 +31,10 @@ import (
 	discoverDaemon "github.com/rook/rook/pkg/daemon/discover"
 	k8sutil "github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/pkg/util/sys"
+
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
-	kserrors "k8s.io/apimachinery/pkg/api/errors"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
@@ -183,7 +184,7 @@ func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAcc
 	// Add NodeAffinity if any
 	nodeAffinity := os.Getenv(discoverDaemonSetNodeAffinityEnv)
 	if nodeAffinity != "" {
-		v1NodeAffinity, err := k8sutil.AddNodeAffinity(nodeAffinity)
+		v1NodeAffinity, err := k8sutil.GenerateNodeAffinity(nodeAffinity)
 		if err != nil {
 			logger.Errorf("failed to create NodeAffinity. %+v", err)
 		} else {
@@ -195,7 +196,7 @@ func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAcc
 
 	_, err = d.clientset.AppsV1().DaemonSets(namespace).Create(ds)
 	if err != nil {
-		if !kserrors.IsAlreadyExists(err) {
+		if !k8serrors.IsAlreadyExists(err) {
 			return fmt.Errorf("failed to create rook-discover daemon set. %+v", err)
 		}
 		logger.Infof("rook-discover daemonset already exists, updating ...")
@@ -427,7 +428,7 @@ func GetAvailableDevices(context *clusterd.Context, nodeName, clusterName string
 		}
 		_, err = context.Clientset.CoreV1().ConfigMaps(namespace).Create(cm)
 		if err != nil {
-			if !kserrors.IsAlreadyExists(err) {
+			if !k8serrors.IsAlreadyExists(err) {
 				return results, fmt.Errorf("failed to update device in use for cluster %s node %s: %v", clusterName, nodeName, err)
 			}
 			if _, err := context.Clientset.CoreV1().ConfigMaps(namespace).Update(cm); err != nil {

pkg/operator/k8sutil/node.go

@@ -306,8 +306,8 @@ func NodeIsInRookNodeList(targetNodeName string, rookNodes []rookalpha.Node) boo
 	return false
 }
 
-// AddNodeAffinity will return v1.NodeAffinity or error
-func AddNodeAffinity(nodeAffinity string) (*v1.NodeAffinity, error) {
+// GenerateNodeAffinity will return v1.NodeAffinity or error
+func GenerateNodeAffinity(nodeAffinity string) (*v1.NodeAffinity, error) {
 	newNodeAffinity := &v1.NodeAffinity{
 		RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
 			NodeSelectorTerms: []v1.NodeSelectorTerm{

pkg/operator/k8sutil/node_test.go

@@ -342,7 +342,7 @@ func TestNodeIsInRookList(t *testing.T) {
 	assert.True(t, NodeIsInRookNodeList("node0-hostname", rookNodes))
 }
 
-func TestAddNodeAffinity(t *testing.T) {
+func TestGenerateNodeAffinity(t *testing.T) {
 	type args struct {
 		nodeAffinity string
 	}
@@ -353,7 +353,7 @@ func TestAddNodeAffinity(t *testing.T) {
 		wantErr bool
 	}{
 		{
-			name: "AddNodeAffinity",
+			name: "GenerateNodeAffinity",
 			args: args{
 				nodeAffinity: "rook.io/ceph=true",
 			},
@@ -375,7 +375,7 @@ func TestAddNodeAffinity(t *testing.T) {
 			wantErr: false,
 		},
 		{
-			name: "FailAddNodeAffinity",
+			name: "FailGenerateNodeAffinity",
 			args: args{
 				nodeAffinity: "rook.io/ceph,minio=true",
 			},
@@ -383,7 +383,7 @@ func TestAddNodeAffinity(t *testing.T) {
 			wantErr: true,
 		},
 		{
-			name: "AddNodeAffinityWithKeyOnly",
+			name: "GenerateNodeAffinityWithKeyOnly",
 			args: args{
 				nodeAffinity: "rook.io/ceph",
 			},
@@ -406,13 +406,13 @@ func TestAddNodeAffinity(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := AddNodeAffinity(tt.args.nodeAffinity)
+			got, err := GenerateNodeAffinity(tt.args.nodeAffinity)
 			if (err != nil) != tt.wantErr {
-				t.Errorf("AddNodeAffinity() error = %v, wantErr %v", err, tt.wantErr)
+				t.Errorf("GenerateNodeAffinity() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
 			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("AddNodeAffinity() = %v, want %v", got, tt.want)
+				t.Errorf("GenerateNodeAffinity() = %v, want %v", got, tt.want)
 			}
 		})
 	}

pkg/operator/k8sutil/tolerations.go

@@ -37,7 +37,7 @@ func YamlToTolerations(raw string) ([]v1.Toleration, error) {
 
 	err = json.Unmarshal(rawJSON, &tolerations)
 	if err != nil {
-		return tolerations, err
+		return []v1.Toleration{}, err
 	}
 
 	return tolerations, nil