ceph: ceph-csi version detection #3824

Checks the version of the configured ceph-csi image while starting the operator. The operator will fail if the image is not supported.
Added an additional parameter to operator to disable the check e.g. to test not yet supported csi images.

Signed-off-by: Stefan Haas <shaas@suse.com>

Documentation/helm-operator.md

@@ -130,6 +130,7 @@ The following tables lists the configurable parameters of the rook-operator char
 | `csi.rbdLivenessMetricsPort`     | Ceph CSI RBD driver metrics port.                                                                                           | `8080`                                                 |
 | `csi.enableSnapshotter`          | Enable deployment of snapshotter container in ceph-csi provisioner.                                                         | `true`                                                 |
 | `csi.forceCephFSKernelClient`    | Enable Ceph Kernel clients on kernel < 4.17 which support quotas for Cephfs.                                                | `true`                                                 |
+| `csi.allowUnsupportedVersion`    | Allow to run unsupported and untested CSI image.                                                                           | `false`                                                |
 | `csi.kubeletDirPath`             | Kubelet root directory path (if the Kubelet uses a different path for the `--root-dir` flag)                                | `/var/lib/kubelet`                                     |
 | `csi.cephcsi.image`              | Ceph CSI image.                                                                                                             | `quay.io/cephcsi/cephcsi:v2.0.0`                       |
 | `csi.rbdPluginUpdateStrategy`    | CSI Rbd plugin daemonset update strategy, supported values are OnDelete and RollingUpdate.                                  | `OnDelete`                                             |

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

@@ -166,6 +166,10 @@ spec:
         - name: CSI_PROVISIONER_NODE_AFFINITY
           value: {{ .Values.csi.provisionerNodeAffinity }}
 {{- end }}
+{{- if .Values.csi.allowUnsupportedVersion }}
+        - name: ROOK_CSI_ALLOW_UNSUPPORTED_VERSION
+          value: {{ .Values.csi.allowUnsupportedVersion | quote }}
+{{- end }}     
 {{- if .Values.csi.pluginTolerations }}
         - name: CSI_PLUGIN_TOLERATIONS
           value: {{ toYaml .Values.csi.pluginTolerations | quote }}

cluster/charts/rook-ceph/values.yaml

@@ -67,6 +67,8 @@ csi:
   # CSI Rbd plugin daemonset update strategy, supported values are OnDelete and RollingUpdate.
   # Default value is RollingUpdate.
   #cephFSPluginUpdateStrategy: OnDelete
+  # Allow starting unsupported ceph-csi image
+  allowUnsupportedVersion: false
 
   # Set provisonerTolerations and provisionerNodeAffinity for provisioner pod.
   # The CSI provisioner would be best to start on the same nodes as other ceph daemons.

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

@@ -308,6 +308,9 @@ spec:
         # (Optional) Ceph Provisioner NodeAffinity.
         # - name: CSI_PROVISIONER_NODE_AFFINITY
         #   value: "role=storage-node; storage=rook, ceph"
+        # (Optional) Allow starting unsupported ceph-csi image
+        - name: ROOK_CSI_ALLOW_UNSUPPORTED_VERSION
+          value: "false"
         # (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

cluster/examples/kubernetes/ceph/operator.yaml

@@ -258,6 +258,9 @@ spec:
         # (Optional) Ceph Provisioner NodeAffinity.
         # - name: CSI_PROVISIONER_NODE_AFFINITY
         #   value: "role=storage-node; storage=rook, ceph"
+        # (Optional) Allow starting unsupported ceph-csi image
+        - name: ROOK_CSI_ALLOW_UNSUPPORTED_VERSION
+          value: "false"
         # (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

cmd/rook/ceph/operator.go

@@ -56,6 +56,7 @@ func init() {
 	operatorCmd.Flags().StringVar(&csi.CSIParam.ProvisionerImage, "csi-provisioner-image", csi.DefaultProvisionerImage, "csi provisioner image")
 	operatorCmd.Flags().StringVar(&csi.CSIParam.AttacherImage, "csi-attacher-image", csi.DefaultAttacherImage, "csi attacher image")
 	operatorCmd.Flags().StringVar(&csi.CSIParam.SnapshotterImage, "csi-snapshotter-image", csi.DefaultSnapshotterImage, "csi snapshotter image")
+	operatorCmd.Flags().BoolVar(&csi.AllowUnsupported, "csi-allow-unsupported-version", false, "csi allow unsupported image version")
 
 	// csi deployment templates
 	operatorCmd.Flags().StringVar(&csi.RBDPluginTemplatePath, "csi-rbd-plugin-template-path", csi.DefaultRBDPluginTemplatePath, "path to ceph-csi rbd plugin template")

pkg/operator/ceph/csi/spec.go

@@ -21,10 +21,12 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/pkg/errors"
 
 	"github.com/rook/rook/pkg/operator/k8sutil"
+	"github.com/rook/rook/pkg/operator/k8sutil/cmdreporter"
 
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -68,6 +70,7 @@ var (
 	EnableRBD            = false
 	EnableCephFS         = false
 	EnableCSIGRPCMetrics = false
+	AllowUnsupported     = false
 
 	//driver names
 	CephFSDriverName string
@@ -134,6 +137,8 @@ const (
 	DefaultRBDGRPCMerticsPort        uint16 = 9090
 	DefaultRBDLivenessMerticsPort    uint16 = 9080
 
+	detectCSIVersionName   = "rook-ceph-csi-detect-version"
+	operatorDeploymentName = "rook-ceph-operator"
 	// default log level for csi containers
 	defaultLogLevel uint8 = 0
 )
@@ -469,3 +474,54 @@ func GetDeploymentOwnerReference(clientset kubernetes.Interface, namespace strin
 	}
 	return deploymentRef, nil
 }
+
+// ValidateCSIVersion checks if the configured ceph-csi image is supported
+func ValidateCSIVersion(clientset kubernetes.Interface, namespace, rookImage, serviceAccountName string) error {
+	var ownerRef metav1.OwnerReference
+	timeout := 15 * time.Minute
+
+	logger.Infof("detecting the ceph csi image version for image %q", CSIParam.CSIPluginImage)
+
+	pod, err := k8sutil.GetRunningPod(clientset)
+	if err != nil {
+		return errors.Wrap(err, "could not get the rook operator pod to obtain the owner reference")
+	}
+	if pod == nil || len(pod.GetOwnerReferences()) == 0 {
+		return errors.New("empty owner reference in rook operator pod")
+	}
+	ownerRef = pod.GetOwnerReferences()[0]
+
+	versionReporter, err := cmdreporter.New(
+		clientset,
+		&ownerRef,
+		detectCSIVersionName, detectCSIVersionName, namespace,
+		[]string{"cephcsi"}, []string{"--version"},
+		rookImage, CSIParam.CSIPluginImage)
+
+	if err != nil {
+		return errors.Wrap(err, "failed to set up ceph CSI version job")
+	}
+
+	job := versionReporter.Job()
+	job.Spec.Template.Spec.ServiceAccountName = serviceAccountName
+
+	stdout, _, retcode, err := versionReporter.Run(timeout)
+	if err != nil {
+		return errors.Wrap(err, "failed to complete ceph CSI version job")
+	}
+
+	if retcode != 0 {
+		return errors.Errorf("ceph CSI version job returned %d", retcode)
+	}
+
+	version, err := extractCephCSIVersion(stdout)
+	if err != nil {
+		return errors.Wrap(err, "failed to extract ceph CSI version")
+	}
+	logger.Infof("Detected ceph CSI image version: %q", version)
+
+	if !version.Supported() {
+		return errors.Errorf("ceph CSI image needs to be at least version %q", minimum.String())
+	}
+	return nil
+}

pkg/operator/ceph/csi/version.go

@@ -0,0 +1,110 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package csi
+
+import (
+	"fmt"
+	"regexp"
+	"strconv"
+
+	"github.com/pkg/errors"
+)
+
+var (
+	//minimum supported version is 2.0.0
+	minimum = CephCSIVersion{2, 0, 0}
+
+	//supportedCSIVersions are versions that rook supports
+	supportedCSIVersions = []CephCSIVersion{minimum}
+
+	// for parsing the output of `cephcsi`
+	versionCSIPattern = regexp.MustCompile(`v(\d+)\.(\d+)\.(\d+)`)
+)
+
+// CephCSIVersion represents the Ceph CSI version format
+type CephCSIVersion struct {
+	Major  int
+	Minor  int
+	Bugfix int
+}
+
+func (v *CephCSIVersion) String() string {
+	return fmt.Sprintf("v%d.%d.%d",
+		v.Major, v.Minor, v.Bugfix)
+}
+
+// Supported checks if the detected version is part of the known supported CSI versions
+func (v *CephCSIVersion) Supported() bool {
+	if v.isAtLeast(&minimum) == false {
+		return false
+	}
+
+	// if AllowUnsupported is set also a csi-image greater than the supported ones are allowed
+	if AllowUnsupported {
+		return true
+	}
+	for _, sv := range supportedCSIVersions {
+		if v.Major == sv.Major {
+			if v.Minor == sv.Minor {
+				if v.Bugfix >= sv.Bugfix {
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
+func (v *CephCSIVersion) isAtLeast(version *CephCSIVersion) bool {
+	if v.Major > version.Major {
+		return true
+	}
+	if v.Major == version.Major && v.Minor >= version.Minor {
+		if v.Minor > version.Minor {
+			return true
+		}
+		if v.Bugfix >= version.Bugfix {
+			return true
+		}
+	}
+	return false
+}
+
+// extractCephCSIVersion extracts the major, minor and extra digit of a Ceph CSI release
+func extractCephCSIVersion(src string) (*CephCSIVersion, error) {
+	m := versionCSIPattern.FindStringSubmatch(src)
+	if m == nil || len(m) < 3 {
+		return nil, errors.Errorf("failed to parse version from: %q", CSIParam.CSIPluginImage)
+	}
+
+	major, err := strconv.Atoi(m[1])
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to parse version major part: %q", m[0])
+	}
+
+	minor, err := strconv.Atoi(m[2])
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to parse version minor part: %q", m[1])
+	}
+
+	bugfix, err := strconv.Atoi(m[3])
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to parse version bugfix part: %q", m[2])
+	}
+
+	return &CephCSIVersion{major, minor, bugfix}, nil
+}

pkg/operator/ceph/csi/version_test.go

@@ -0,0 +1,82 @@
+/*
+Copyright 2020 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package csi
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	testVersion            = CephCSIVersion{2, 0, 0}
+	testVersionUnsupported = CephCSIVersion{2, 1, 0}
+)
+
+func TestIsAtLeast(t *testing.T) {
+	// Test version which is smaller
+	var version = CephCSIVersion{1, 40, 10}
+	ret := testVersion.isAtLeast(&version)
+	assert.Equal(t, true, ret)
+
+	// Test version which is equal
+	ret = testVersion.isAtLeast(&testVersion)
+	assert.Equal(t, true, ret)
+
+	// Test version which is greater (minor)
+	version = CephCSIVersion{2, 1, 0}
+	ret = testVersion.isAtLeast(&version)
+	assert.Equal(t, false, ret)
+
+	// Test version which is greater (bugfix)
+	version = CephCSIVersion{2, 0, 1}
+	ret = testVersion.isAtLeast(&version)
+	assert.Equal(t, false, ret)
+}
+
+func TestSupported(t *testing.T) {
+	ret := testVersion.Supported()
+	assert.Equal(t, true, ret)
+
+	ret = testVersionUnsupported.Supported()
+	assert.Equal(t, false, ret)
+}
+
+func Test_extractCephCSIVersion(t *testing.T) {
+	expectedVersion := CephCSIVersion{2, 0, 0}
+	csiString := []byte(`Cephcsi Version: v2.0.0
+		Git Commit: e58d537a07ca0184f67d33db85bf6b4911624b44
+		Go Version: go1.12.15
+		Compiler: gc
+		Platform: linux/amd64
+		`)
+	version, err := extractCephCSIVersion(string(csiString))
+
+	assert.Equal(t, &expectedVersion, version)
+	assert.Nil(t, err)
+
+	csiString = []byte(`Cephcsi Version: rubish
+	Git Commit: e58d537a07ca0184f67d33db85bf6b4911624b44
+	Go Version: go1.12.15
+	Compiler: gc
+	Platform: linux/amd64
+	`)
+	version, err = extractCephCSIVersion(string(csiString))
+
+	assert.Nil(t, version)
+	assert.Contains(t, err.Error(), "failed to parse version from")
+}

pkg/operator/ceph/operator.go

@@ -196,6 +196,10 @@ func (o *Operator) startSystemDaemons(clusterSpec *cephv1.ClusterSpec) error {
 		return errors.Wrapf(err, "invalid csi params")
 	}
 
+	if err = csi.ValidateCSIVersion(o.context.Clientset, o.operatorNamespace, o.rookImage, o.securityAccount); err != nil {
+		return errors.Wrap(err, "invalid csi version")
+	}
+
 	if err = csi.StartCSIDrivers(o.operatorNamespace, o.context.Clientset, serverVersion); err != nil {
 		return errors.Wrapf(err, "failed to start Ceph csi drivers")
 	}