diff --git a/cluster/charts/rook-ceph/templates/deployment.yaml b/cluster/charts/rook-ceph/templates/deployment.yaml index 6a893a19f..21e0ddd53 100644 --- a/cluster/charts/rook-ceph/templates/deployment.yaml +++ b/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 }}" diff --git a/cluster/charts/rook-ceph/values.yaml b/cluster/charts/rook-ceph/values.yaml index b193161be..74c36146e 100644 --- a/cluster/charts/rook-ceph/values.yaml +++ b/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 diff --git a/cluster/examples/kubernetes/ceph/operator-openshift.yaml b/cluster/examples/kubernetes/ceph/operator-openshift.yaml index d395bdc48..634aed2ba 100644 --- a/cluster/examples/kubernetes/ceph/operator-openshift.yaml +++ b/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 diff --git a/cluster/examples/kubernetes/ceph/operator.yaml b/cluster/examples/kubernetes/ceph/operator.yaml index abb67ce97..5280e0269 100644 --- a/cluster/examples/kubernetes/ceph/operator.yaml +++ b/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: diff --git a/pkg/apis/rook.io/v1alpha2/placement_test.go b/pkg/apis/rook.io/v1alpha2/placement_test.go index 8126a2344..bc597f46f 100644 --- a/pkg/apis/rook.io/v1alpha2/placement_test.go +++ b/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{ diff --git a/pkg/operator/ceph/agent/agent.go b/pkg/operator/ceph/agent/agent.go index 7f7c9e149..96944ac7c 100644 --- a/pkg/operator/ceph/agent/agent.go +++ b/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 ...") diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go index cd08893e3..df598ea76 100644 --- a/pkg/operator/ceph/csi/spec.go +++ b/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) diff --git a/pkg/operator/ceph/csi/util.go b/pkg/operator/ceph/csi/util.go index f127849ce..269dbabb6 100644 --- a/pkg/operator/ceph/csi/util.go +++ b/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, + } +} diff --git a/pkg/operator/discover/discover.go b/pkg/operator/discover/discover.go index 1ac3278a1..23839f85d 100644 --- a/pkg/operator/discover/discover.go +++ b/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 { diff --git a/pkg/operator/k8sutil/node.go b/pkg/operator/k8sutil/node.go index 243b0099b..5d27bf20e 100644 --- a/pkg/operator/k8sutil/node.go +++ b/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{ diff --git a/pkg/operator/k8sutil/node_test.go b/pkg/operator/k8sutil/node_test.go index 20e57d200..2d235147e 100644 --- a/pkg/operator/k8sutil/node_test.go +++ b/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) } }) } diff --git a/pkg/operator/k8sutil/tolerations.go b/pkg/operator/k8sutil/tolerations.go index 0dac565bc..bbc650df0 100644 --- a/pkg/operator/k8sutil/tolerations.go +++ b/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