Repository: rook/rook
Pull request #5572 (2020-06-01 17:18:59)

Start csi driver in parallel of cluster (bp #5568)

<!-- Please take a look at our [Contributing](https://rook.io/docs/rook/master/development-flow.html)
documentation before submitting a Pull Request!
Thank you for contributing to Rook! -->

**Description of your changes:**
The csi driver does not need to be started before the cluster is created. When the csi driver fails to load, neither should it prevent the cluster from being created. Therefore, we initialize the basic constructs in the csi driver that are required for cluster creation, then start the csi driver in a goroutine so the cluster creation can continue. If the csi driver fails or takes a long time to load, it will no longer affect cluster creation.

Signed-off-by: Travis Nielsen <tnielsen@redhat.com>
(cherry picked from commit 6c249e751b12425602351eae6b393da6973ccaef)

**Checklist:**

- [ ] **Commit Message Formatting**: Commit titles and messages follow guidelines in the [developer guide](https://rook.io/docs/rook/master/development-flow.html#commit-structure).
- [ ] **Skip Tests for Docs**: Add the flag for skipping the build if this is only a documentation change. See [here](https://github.com/rook/rook/blob/master/INSTALL.md#skip-ci) for the flag.
- [ ] **Skip Unrelated Tests**: Add a flag to run tests for a specific storage provider. See [test options](https://github.com/rook/rook/blob/master/INSTALL.md#test-storage-provider).
- [ ] Reviewed the developer guide on [Submitting a Pull Request](https://rook.io/docs/rook/master/development-flow.html#submitting-a-pull-request)
- [ ] Documentation has been updated, if necessary.
- [ ] Unit tests have been added, if necessary.
- [ ] Integration tests have been added, if necessary.
- [ ] Pending release notes updated with breaking and/or notable changes, if necessary.
- [ ] Upgrade from previous release is tested and upgrade user guide is updated, if necessary.
- [ ] Code generation (`make codegen`) has been run to update object specifications, if necessary.

[test ceph]


Patch:
diff --git a/pkg/operator/ceph/csi/csi.go b/pkg/operator/ceph/csi/csi.go
new file mode 100644
index 000000000..c657403ae
--- /dev/null
+++ b/pkg/operator/ceph/csi/csi.go
@@ -0,0 +1,103 @@
+/*
+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 (
+	"strconv"
+
+	"github.com/pkg/errors"
+	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) {
+	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 {
+		logger.Errorf("failed to start Ceph csi drivers. %v", err)
+		return
+	}
+	logger.Infof("successfully started Ceph CSI driver(s)")
+}
+
+func SetParams(clientset kubernetes.Interface) error {
+	var err error
+
+	csiEnableRBD, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_ENABLE_RBD", "true")
+	if err != nil {
+		return errors.Wrap(err, "unable to determine if CSI driver for RBD is enabled")
+	}
+	if EnableRBD, err = strconv.ParseBool(csiEnableRBD); err != nil {
+		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_RBD'")
+	}
+
+	csiEnableCephFS, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_ENABLE_CEPHFS", "true")
+	if err != nil {
+		return errors.Wrap(err, "unable to determine if CSI driver for CephFS is enabled")
+	}
+	if EnableCephFS, err = strconv.ParseBool(csiEnableCephFS); err != nil {
+		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_CEPHFS'")
+	}
+
+	csiAllowUnsupported, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_ALLOW_UNSUPPORTED_VERSION", "false")
+	if err != nil {
+		return errors.Wrap(err, "unable to determine if unsupported version is allowed")
+	}
+	if AllowUnsupported, err = strconv.ParseBool(csiAllowUnsupported); err != nil {
+		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ALLOW_UNSUPPORTED_VERSION'")
+	}
+
+	csiEnableCSIGRPCMetrics, err := k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_ENABLE_GRPC_METRICS", "true")
+	if err != nil {
+		return errors.Wrap(err, "unable to determine if CSI GRPC metrics is enabled")
+	}
+	if EnableCSIGRPCMetrics, err = strconv.ParseBool(csiEnableCSIGRPCMetrics); err != nil {
+		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_GRPC_METRICS'")
+	}
+
+	CSIParam.CSIPluginImage, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_CEPH_IMAGE", DefaultCSIPluginImage)
+	if err != nil {
+		return errors.Wrap(err, "unable to configure CSI plugin image")
+	}
+	CSIParam.RegistrarImage, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_REGISTRAR_IMAGE", DefaultRegistrarImage)
+	if err != nil {
+		return errors.Wrap(err, "unable to configure CSI registrar image")
+	}
+	CSIParam.ProvisionerImage, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_PROVISIONER_IMAGE", DefaultProvisionerImage)
+	if err != nil {
+		return errors.Wrap(err, "unable to configure CSI provisioner image")
+	}
+	CSIParam.AttacherImage, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_ATTACHER_IMAGE", DefaultAttacherImage)
+	if err != nil {
+		return errors.Wrap(err, "unable to configure CSI attacher image")
+	}
+	CSIParam.SnapshotterImage, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_SNAPSHOTTER_IMAGE", DefaultSnapshotterImage)
+	if err != nil {
+		return errors.Wrap(err, "unable to configure CSI snapshotter image")
+	}
+	CSIParam.KubeletDirPath, err = k8sutil.GetOperatorSetting(clientset, controllerutil.OperatorSettingConfigMapName, "ROOK_CSI_KUBELET_DIR_PATH", DefaultKubeletDirPath)
+	if err != nil {
+		return errors.Wrap(err, "unable to configure CSI kubelet directory path")
+	}
+	return nil
+}
diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go
index 2931dd8f2..5be29ab5e 100644
--- a/pkg/operator/ceph/csi/spec.go
+++ b/pkg/operator/ceph/csi/spec.go
@@ -18,7 +18,6 @@ package csi
 
 import (
 	"fmt"
-	"os"
 	"strconv"
 	"strings"
 	"time"
@@ -197,7 +196,7 @@ func ValidateCSIParam() error {
 	return nil
 }
 
-func StartCSIDrivers(namespace string, clientset kubernetes.Interface, ver *version.Info) error {
+func startDrivers(namespace string, clientset kubernetes.Interface, ver *version.Info, ownerRef *metav1.OwnerReference) error {
 	var (
 		err                                                   error
 		rbdPlugin, cephfsPlugin                               *apps.DaemonSet
@@ -207,22 +206,6 @@ func StartCSIDrivers(namespace string, clientset kubernetes.Interface, ver *vers
 		rbdService, cephfsService                             *corev1.Service
 	)
 
-	ownerRef, err := GetDeploymentOwnerReference(clientset, namespace)
-	if err != nil {
-		logger.Warningf("could not find deployment owner reference to assign to csi drivers. %v", err)
-	}
-	if ownerRef != nil {
-		blockOwnerDeletion := false
-		ownerRef.BlockOwnerDeletion = &blockOwnerDeletion
-	}
-
-	// create an empty config map. config map will be filled with data
-	// later when clusters have mons
-	err = CreateCsiConfigMap(namespace, clientset, ownerRef)
-	if err != nil {
-		return errors.Wrapf(err, "failed creating csi config map")
-	}
-
 	tp := templateParam{
 		Param:     CSIParam,
 		Namespace: namespace,
@@ -524,50 +507,12 @@ func createCSIDriverInfo(clientset kubernetes.Interface, name string, ownerRef *
 	return err
 }
 
-// GetDeploymentOwnerReference returns an OwnerReference to the rook-ceph-operator deployment
-func GetDeploymentOwnerReference(clientset kubernetes.Interface, namespace string) (*metav1.OwnerReference, error) {
-	var deploymentRef *metav1.OwnerReference
-	podName := os.Getenv(k8sutil.PodNameEnvVar)
-	pod, err := clientset.CoreV1().Pods(namespace).Get(podName, metav1.GetOptions{})
-	if err != nil {
-		return nil, errors.Wrapf(err, "could not find pod %q to find deployment owner reference", podName)
-	}
-	for _, podOwner := range pod.OwnerReferences {
-		if podOwner.Kind == "ReplicaSet" {
-			replicaset, err := clientset.AppsV1().ReplicaSets(namespace).Get(podOwner.Name, metav1.GetOptions{})
-			if err != nil {
-				return nil, errors.Wrapf(err, "could not find replicaset %q to find deployment owner reference", podOwner.Name)
-			}
-			for _, replicasetOwner := range replicaset.OwnerReferences {
-				if replicasetOwner.Kind == "Deployment" {
-					deploymentRef = &replicasetOwner
-				}
-			}
-		}
-	}
-	if deploymentRef == nil {
-		return nil, errors.New("could not find owner reference for rook-ceph deployment")
-	}
-	return deploymentRef, nil
-}
-
 // ValidateCSIVersion checks if the configured ceph-csi image is supported
-func ValidateCSIVersion(clientset kubernetes.Interface, namespace, rookImage, serviceAccountName string) error {
+func validateCSIVersion(clientset kubernetes.Interface, namespace, rookImage, serviceAccountName string, ownerRef *metav1.OwnerReference) error {
 	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].DeepCopy()
-
-	*ownerRef.BlockOwnerDeletion = false
-
 	versionReporter, err := cmdreporter.New(
 		clientset,
 		ownerRef,
diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go
index 7258d7f8c..ba6ca3170 100644
--- a/pkg/operator/ceph/operator.go
+++ b/pkg/operator/ceph/operator.go
@@ -20,7 +20,6 @@ package operator
 import (
 	"os"
 	"os/signal"
-	"strconv"
 	"syscall"
 	"time"
 
@@ -31,12 +30,13 @@ import (
 	"github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume/attachment"
 	"github.com/rook/rook/pkg/operator/ceph/agent"
 	"github.com/rook/rook/pkg/operator/ceph/cluster"
-	cephController "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/csi"
 	"github.com/rook/rook/pkg/operator/ceph/provisioner"
 	"github.com/rook/rook/pkg/operator/discover"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/sig-storage-lib-external-provisioner/controller"
 )
@@ -204,7 +204,7 @@ func (o *Operator) updateDrivers() error {
 		return errors.Wrapf(err, "error getting server version")
 	}
 
-	if err = o.setCSIParams(); err != nil {
+	if err = csi.SetParams(o.context.Clientset); err != nil {
 		return errors.Wrap(err, "failed to configure CSI parameters")
 	}
 
@@ -214,86 +214,60 @@ func (o *Operator) updateDrivers() error {
 	}
 
 	if serverVersion.Major < csi.KubeMinMajor || serverVersion.Major == csi.KubeMinMajor && serverVersion.Minor < csi.KubeMinMinor {
-		logger.Infof("CSI driver is only supported in K8s 1.13 or newer. version=%s", serverVersion.String())
+		logger.Infof("CSI drivers only supported in K8s 1.13 or newer. version=%s", serverVersion.String())
 		// disable csi control variables to disable other csi functions
 		csi.EnableRBD = false
 		csi.EnableCephFS = false
 		return nil
 	}
 
-	if err = csi.ValidateCSIParam(); err != nil {
-		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")
-	}
-	logger.Infof("successfully started Ceph CSI driver(s)")
-	return nil
-}
-
-func (o *Operator) setCSIParams() error {
-	var err error
-
-	csiEnableRBD, err := k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_ENABLE_RBD", "true")
+	ownerRef, err := getDeploymentOwnerReference(o.context.Clientset, o.operatorNamespace)
 	if err != nil {
-		return errors.Wrap(err, "unable to determine if CSI driver for RBD is enabled")
+		logger.Warningf("could not find deployment owner reference to assign to csi drivers. %v", err)
 	}
-	if csi.EnableRBD, err = strconv.ParseBool(csiEnableRBD); err != nil {
-		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_RBD'")
+	if ownerRef != nil {
+		blockOwnerDeletion := false
+		ownerRef.BlockOwnerDeletion = &blockOwnerDeletion
 	}
 
-	csiEnableCephFS, err := k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_ENABLE_CEPHFS", "true")
+	// create an empty config map. config map will be filled with data
+	// later when clusters have mons
+	err = csi.CreateCsiConfigMap(o.operatorNamespace, o.context.Clientset, ownerRef)
 	if err != nil {
-		return errors.Wrap(err, "unable to determine if CSI driver for CephFS is enabled")
-	}
-	if csi.EnableCephFS, err = strconv.ParseBool(csiEnableCephFS); err != nil {
-		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_CEPHFS'")
+		return errors.Wrap(err, "failed creating csi config map")
 	}
 
-	csiAllowUnsupported, err := k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_ALLOW_UNSUPPORTED_VERSION", "false")
-	if err != nil {
-		return errors.Wrap(err, "unable to determine if unsupported version is allowed")
-	}
-	if csi.AllowUnsupported, err = strconv.ParseBool(csiAllowUnsupported); err != nil {
-		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ALLOW_UNSUPPORTED_VERSION'")
+	if err = csi.ValidateCSIParam(); err != nil {
+		return errors.Wrap(err, "invalid csi params")
 	}
 
-	csiEnableCSIGRPCMetrics, err := k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_ENABLE_GRPC_METRICS", "true")
-	if err != nil {
-		return errors.Wrap(err, "unable to determine if CSI GRPC metrics is enabled")
-	}
-	if csi.EnableCSIGRPCMetrics, err = strconv.ParseBool(csiEnableCSIGRPCMetrics); err != nil {
-		return errors.Wrap(err, "unable to parse value for 'ROOK_CSI_ENABLE_GRPC_METRICS'")
-	}
+	go csi.ValidateAndStartDrivers(o.context.Clientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef)
+	return nil
+}
 
-	csi.CSIParam.CSIPluginImage, err = k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_CEPH_IMAGE", csi.DefaultCSIPluginImage)
-	if err != nil {
-		return errors.Wrap(err, "unable to configure CSI plugin image")
-	}
-	csi.CSIParam.RegistrarImage, err = k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_REGISTRAR_IMAGE", csi.DefaultRegistrarImage)
-	if err != nil {
-		return errors.Wrap(err, "unable to configure CSI registrar image")
-	}
-	csi.CSIParam.ProvisionerImage, err = k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_PROVISIONER_IMAGE", csi.DefaultProvisionerImage)
+// getDeploymentOwnerReference returns an OwnerReference to the rook-ceph-operator deployment
+func getDeploymentOwnerReference(clientset kubernetes.Interface, namespace string) (*metav1.OwnerReference, error) {
+	var deploymentRef *metav1.OwnerReference
+	podName := os.Getenv(k8sutil.PodNameEnvVar)
+	pod, err := clientset.CoreV1().Pods(namespace).Get(podName, metav1.GetOptions{})
 	if err != nil {
-		return errors.Wrap(err, "unable to configure CSI provisioner image")
-	}
-	csi.CSIParam.AttacherImage, err = k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_ATTACHER_IMAGE", csi.DefaultAttacherImage)
-	if err != nil {
-		return errors.Wrap(err, "unable to configure CSI attacher image")
-	}
-	csi.CSIParam.SnapshotterImage, err = k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_SNAPSHOTTER_IMAGE", csi.DefaultSnapshotterImage)
-	if err != nil {
-		return errors.Wrap(err, "unable to configure CSI snapshotter image")
+		return nil, errors.Wrapf(err, "could not find pod %q to find deployment owner reference", podName)
+	}
+	for _, podOwner := range pod.OwnerReferences {
+		if podOwner.Kind == "ReplicaSet" {
+			replicaset, err := clientset.AppsV1().ReplicaSets(namespace).Get(podOwner.Name, metav1.GetOptions{})
+			if err != nil {
+				return nil, errors.Wrapf(err, "could not find replicaset %q to find deployment owner reference", podOwner.Name)
+			}
+			for _, replicasetOwner := range replicaset.OwnerReferences {
+				if replicasetOwner.Kind == "Deployment" {
+					deploymentRef = &replicasetOwner
+				}
+			}
+		}
 	}
-	csi.CSIParam.KubeletDirPath, err = k8sutil.GetOperatorSetting(o.context.Clientset, cephController.OperatorSettingConfigMapName, "ROOK_CSI_KUBELET_DIR_PATH", csi.DefaultKubeletDirPath)
-	if err != nil {
-		return errors.Wrap(err, "unable to configure CSI kubelet directory path")
+	if deploymentRef == nil {
+		return nil, errors.New("could not find owner reference for rook-ceph deployment")
 	}
-	return nil
+	return deploymentRef, nil
 }


Test patch:
diff --git a/pkg/operator/ceph/csi/spec_test.go b/pkg/operator/ceph/csi/spec_test.go
index ed451824e..a2b26f57e 100644
--- a/pkg/operator/ceph/csi/spec_test.go
+++ b/pkg/operator/ceph/csi/spec_test.go
@@ -44,6 +44,6 @@ func TestStartCSI(t *testing.T) {
 	if err != nil {
 		assert.Nil(t, err)
 	}
-	err = StartCSIDrivers("ns", clientset, serverVersion)
+	err = startDrivers("ns", clientset, serverVersion, nil)
 	assert.Nil(t, err)
 }