diff --git a/pkg/operator/ceph/csi/csi.go b/pkg/operator/ceph/csi/csi.go index c657403ae..a207619ec 100644 --- a/pkg/operator/ceph/csi/csi.go +++ b/pkg/operator/ceph/csi/csi.go @@ -27,17 +27,17 @@ import ( "k8s.io/client-go/kubernetes" ) -func ValidateAndStartDrivers(clientset kubernetes.Interface, namespace, rookImage, securityAccount string, serverVersion *version.Info, ownerRef *metav1.OwnerReference) { +func ValidateAndConfigureDrivers(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 { + if err := startDrivers(clientset, namespace, serverVersion, ownerRef); err != nil { logger.Errorf("failed to start Ceph csi drivers. %v", err) return } - logger.Infof("successfully started Ceph CSI driver(s)") + stopDrivers(clientset, namespace, serverVersion) } func SetParams(clientset kubernetes.Interface) error { diff --git a/pkg/operator/ceph/csi/spec.go b/pkg/operator/ceph/csi/spec.go index 2e9b9f48e..d2f71f2a8 100644 --- a/pkg/operator/ceph/csi/spec.go +++ b/pkg/operator/ceph/csi/spec.go @@ -156,6 +156,14 @@ const ( // update strategy rollingUpdate = "RollingUpdate" onDelete = "OnDelete" + + // driver daemonset names + csiRBDPlugin = "csi-rbdplugin" + csiCephFSPlugin = "csi-cephfsplugin" + + // driver deployment names + csiRBDProvisioner = "csi-rbdplugin-provisioner" + csiCephFSProvisioner = "csi-cephfsplugin-provisioner" ) func CSIEnabled() bool { @@ -197,7 +205,7 @@ func ValidateCSIParam() error { return nil } -func startDrivers(namespace string, clientset kubernetes.Interface, ver *version.Info, ownerRef *metav1.OwnerReference) error { +func startDrivers(clientset kubernetes.Interface, namespace string, ver *version.Info, ownerRef *metav1.OwnerReference) error { var ( err error rbdPlugin, cephfsPlugin *apps.DaemonSet @@ -343,6 +351,7 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version if err != nil { return errors.Wrap(err, "failed to load rbd plugin service template") } + logger.Info("successfully started CSI Ceph RBD") } if EnableCephFS { cephfsPlugin, err = templateToDaemonSet("cephfsplugin", CephFSPluginTemplatePath, tp) @@ -364,7 +373,9 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version if err != nil { return errors.Wrap(err, "failed to load cephfs plugin service template") } + logger.Info("successfully started CSI CephFS driver") } + // get provisioner toleration and node affinity provisionerTolerations := getToleration(clientset, true) provisionerNodeAffinity := getNodeAffinity(clientset, true) @@ -376,7 +387,7 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version // apply resource request and limit to rbdplugin containers applyResourcesToContainers(clientset, rbdPluginResource, &rbdPlugin.Spec.Template.Spec) k8sutil.SetOwnerRef(&rbdPlugin.ObjectMeta, ownerRef) - err = k8sutil.CreateDaemonSet("csi-rbdplugin", namespace, clientset, rbdPlugin) + err = k8sutil.CreateDaemonSet(csiRBDPlugin, namespace, clientset, rbdPlugin) if err != nil { return errors.Wrapf(err, "failed to start rbdplugin daemonset: %+v", rbdPlugin) } @@ -388,7 +399,7 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version // apply resource request and limit to rbd provisioner containers applyResourcesToContainers(clientset, rbdProvisionerResource, &rbdProvisionerSTS.Spec.Template.Spec) k8sutil.SetOwnerRef(&rbdProvisionerSTS.ObjectMeta, ownerRef) - err = k8sutil.CreateStatefulSet(clientset, "csi-rbdplugin-provisioner", namespace, rbdProvisionerSTS) + err = k8sutil.CreateStatefulSet(clientset, csiRBDProvisioner, namespace, rbdProvisionerSTS) if err != nil { return errors.Wrapf(err, "failed to start rbd provisioner statefulset: %+v", rbdProvisionerSTS) } @@ -398,13 +409,13 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version // apply resource request and limit to rbd provisioner containers applyResourcesToContainers(clientset, rbdProvisionerResource, &rbdProvisionerDeployment.Spec.Template.Spec) k8sutil.SetOwnerRef(&rbdProvisionerDeployment.ObjectMeta, ownerRef) - antiAffinity := GetPodAntiAffinity("app", "csi-rbdplugin-provisioner") + antiAffinity := GetPodAntiAffinity("app", csiRBDProvisioner) rbdProvisionerDeployment.Spec.Template.Spec.Affinity.PodAntiAffinity = &antiAffinity rbdProvisionerDeployment.Spec.Strategy = apps.DeploymentStrategy{ Type: apps.RecreateDeploymentStrategyType, } - err = k8sutil.CreateDeployment(clientset, "csi-rbdplugin-provisioner", namespace, rbdProvisionerDeployment) + err = k8sutil.CreateDeployment(clientset, csiRBDProvisioner, namespace, rbdProvisionerDeployment) if err != nil { return errors.Wrapf(err, "failed to start rbd provisioner deployment: %+v", rbdProvisionerDeployment) } @@ -424,7 +435,7 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version // apply resource request and limit to cephfs plugin containers applyResourcesToContainers(clientset, cephFSPluginResource, &cephfsPlugin.Spec.Template.Spec) k8sutil.SetOwnerRef(&cephfsPlugin.ObjectMeta, ownerRef) - err = k8sutil.CreateDaemonSet("csi-cephfsplugin", namespace, clientset, cephfsPlugin) + err = k8sutil.CreateDaemonSet(csiCephFSPlugin, namespace, clientset, cephfsPlugin) if err != nil { return errors.Wrapf(err, "failed to start cephfs plugin daemonset: %+v", cephfsPlugin) } @@ -436,7 +447,7 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version // apply resource request and limit to cephfs provisioner containers applyResourcesToContainers(clientset, cephFSProvisionerResource, &cephfsProvisionerSTS.Spec.Template.Spec) k8sutil.SetOwnerRef(&cephfsProvisionerSTS.ObjectMeta, ownerRef) - err = k8sutil.CreateStatefulSet(clientset, "csi-cephfsplugin-provisioner", namespace, cephfsProvisionerSTS) + err = k8sutil.CreateStatefulSet(clientset, csiCephFSProvisioner, namespace, cephfsProvisionerSTS) if err != nil { return errors.Wrapf(err, "failed to start cephfs provisioner statefulset: %+v", cephfsProvisionerSTS) } @@ -448,12 +459,12 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version // apply resource request and limit to cephfs provisioner containers applyResourcesToContainers(clientset, cephFSProvisionerResource, &cephfsProvisionerDeployment.Spec.Template.Spec) k8sutil.SetOwnerRef(&cephfsProvisionerDeployment.ObjectMeta, ownerRef) - antiAffinity := GetPodAntiAffinity("app", "csi-cephfsplugin-provisioner") + antiAffinity := GetPodAntiAffinity("app", csiCephFSProvisioner) cephfsProvisionerDeployment.Spec.Template.Spec.Affinity.PodAntiAffinity = &antiAffinity cephfsProvisionerDeployment.Spec.Strategy = apps.DeploymentStrategy{ Type: apps.RecreateDeploymentStrategyType, } - err = k8sutil.CreateDeployment(clientset, "csi-cephfsplugin-provisioner", namespace, cephfsProvisionerDeployment) + err = k8sutil.CreateDeployment(clientset, csiCephFSProvisioner, namespace, cephfsProvisionerDeployment) if err != nil { return errors.Wrapf(err, "failed to start cephfs provisioner deployment: %+v", cephfsProvisionerDeployment) } @@ -468,18 +479,81 @@ func startDrivers(namespace string, clientset kubernetes.Interface, ver *version } if ver.Major > KubeMinMajor || (ver.Major == KubeMinMajor && ver.Minor >= provDeploymentSuppVersion) { - err = createCSIDriverInfo(clientset, RBDDriverName, ownerRef) - if err != nil { - return errors.Wrapf(err, "failed to create CSI driver object for %q", RBDDriverName) + if EnableRBD { + err = createCSIDriverInfo(clientset, RBDDriverName, ownerRef) + if err != nil { + return errors.Wrapf(err, "failed to create CSI driver object for %q", RBDDriverName) + } } - err = createCSIDriverInfo(clientset, CephFSDriverName, ownerRef) - if err != nil { - return errors.Wrapf(err, "failed to create CSI driver object for %q", CephFSDriverName) + if EnableCephFS { + err = createCSIDriverInfo(clientset, CephFSDriverName, ownerRef) + if err != nil { + return errors.Wrapf(err, "failed to create CSI driver object for %q", CephFSDriverName) + } } } return nil } +func stopDrivers(clientset kubernetes.Interface, namespace string, ver *version.Info) { + if !EnableRBD { + logger.Info("CSI Ceph RBD driver disabled") + succeeded := deleteCSIDriverResources(clientset, ver, namespace, csiRBDPlugin, csiRBDProvisioner, "csi-rbdplugin-metrics", RBDDriverName) + if succeeded { + logger.Info("successfully removed CSI Ceph RBD driver") + } else { + logger.Error("failed to remove CSI Ceph RBD driver") + } + } + + if !EnableCephFS { + logger.Info("CSI CephFS driver disabled") + succeeded := deleteCSIDriverResources(clientset, ver, namespace, csiCephFSPlugin, csiCephFSProvisioner, "csi-cephfsplugin-metrics", CephFSDriverName) + if succeeded { + logger.Info("successfully removed CSI CephFS driver") + } else { + logger.Error("failed to remove CSI CephFS driver") + } + } +} + +func deleteCSIDriverResources( + clientset kubernetes.Interface, ver *version.Info, namespace, daemonset, deployment, service, driverName string) bool { + succeeded := true + err := k8sutil.DeleteDaemonset(clientset, namespace, daemonset) + if err != nil { + logger.Errorf("failed to delete the %q. %v", daemonset, err) + succeeded = false + } + + if ver.Major == KubeMinMajor && ver.Minor < provDeploymentSuppVersion { + err = k8sutil.DeleteStatefulset(clientset, namespace, deployment) + if err != nil { + logger.Errorf("failed to delete the %q. %v", deployment, err) + succeeded = false + } + } else { + err = k8sutil.DeleteDeployment(clientset, namespace, deployment) + if err != nil { + logger.Errorf("failed to delete the %q. %v", deployment, err) + succeeded = false + } + } + + err = k8sutil.DeleteService(clientset, namespace, service) + if err != nil { + logger.Errorf("failed to delete the %q. %v", service, err) + succeeded = false + } + + err = deleteCSIDriverInfo(clientset, driverName) + if err != nil { + logger.Errorf("failed to delete %q Driver Info. %v", driverName, err) + succeeded = false + } + return succeeded +} + // createCSIDriverInfo Registers CSI driver by creating a CSIDriver object func createCSIDriverInfo(clientset kubernetes.Interface, name string, ownerRef *metav1.OwnerReference) error { attach := true @@ -509,6 +583,17 @@ func createCSIDriverInfo(clientset kubernetes.Interface, name string, ownerRef * return err } +// deleteCSIDriverInfo deletes CSIDriverInfo and returns the error if any +func deleteCSIDriverInfo(clientset kubernetes.Interface, name string) error { + err := clientset.StorageV1beta1().CSIDrivers().Delete(name, &metav1.DeleteOptions{}) + if err != nil { + if apierrors.IsNotFound(err) { + return nil + } + } + return err +} + // ValidateCSIVersion checks if the configured ceph-csi image is supported func validateCSIVersion(clientset kubernetes.Interface, namespace, rookImage, serviceAccountName string, ownerRef *metav1.OwnerReference) error { timeout := 15 * time.Minute diff --git a/pkg/operator/ceph/csi/spec_test.go b/pkg/operator/ceph/csi/spec_test.go index a2b26f57e..70108d751 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 = startDrivers("ns", clientset, serverVersion, nil) + err = startDrivers(clientset, "ns", serverVersion, nil) assert.Nil(t, err) } diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go index 43193573c..5d69217cd 100644 --- a/pkg/operator/ceph/operator.go +++ b/pkg/operator/ceph/operator.go @@ -223,11 +223,6 @@ func (o *Operator) updateDrivers() error { return errors.Wrap(err, "failed to configure CSI parameters") } - if !csi.CSIEnabled() { - logger.Infof("CSI driver is not enabled") - return nil - } - if serverVersion.Major < csi.KubeMinMajor || serverVersion.Major == csi.KubeMinMajor && serverVersion.Minor < csi.KubeMinMinor { 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 @@ -256,7 +251,7 @@ func (o *Operator) updateDrivers() error { return errors.Wrap(err, "invalid csi params") } - go csi.ValidateAndStartDrivers(o.context.Clientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef) + go csi.ValidateAndConfigureDrivers(o.context.Clientset, o.operatorNamespace, o.rookImage, o.securityAccount, serverVersion, ownerRef) return nil } diff --git a/pkg/operator/k8sutil/service.go b/pkg/operator/k8sutil/service.go index e15e1cf12..894ef210b 100644 --- a/pkg/operator/k8sutil/service.go +++ b/pkg/operator/k8sutil/service.go @@ -62,6 +62,17 @@ func UpdateService( return clientset.CoreV1().Services(namespace).Update(serviceDefinition) } +// DeleteService deletes a Service and returns the error if any +func DeleteService(clientset kubernetes.Interface, namespace, name string) error { + err := clientset.CoreV1().Services(namespace).Delete(name, &metav1.DeleteOptions{}) + if err != nil { + if errors.IsNotFound(err) { + return nil + } + } + return err +} + // ParseServiceType parses a string and returns a*v1.ServiceType. If the ServiceType is invalid, // this should be considered an error. func ParseServiceType(serviceString string) v1.ServiceType { diff --git a/pkg/operator/k8sutil/statefulset.go b/pkg/operator/k8sutil/statefulset.go index d0d875829..34f960983 100644 --- a/pkg/operator/k8sutil/statefulset.go +++ b/pkg/operator/k8sutil/statefulset.go @@ -21,6 +21,7 @@ import ( apps "k8s.io/api/apps/v1" k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" ) @@ -38,6 +39,19 @@ func CreateStatefulSet(clientset kubernetes.Interface, name, namespace string, s return nil } +// DeleteStatefulset makes a best effort at deleting a statefulset and its pods, then waits for them to be deleted +func DeleteStatefulset(clientset kubernetes.Interface, namespace, name string) error { + logger.Debugf("removing %s statefulset if it exists", name) + deleteAction := func(options *metav1.DeleteOptions) error { + return clientset.AppsV1().StatefulSets(namespace).Delete(name, options) + } + getAction := func() error { + _, err := clientset.AppsV1().StatefulSets(namespace).Get(name, metav1.GetOptions{}) + return err + } + return deleteResourceAndWait(namespace, name, "statefulset", deleteAction, getAction) +} + // AddRookVersionLabelToStatefulSet adds or updates a label reporting the Rook version which last // modified a apps.statefulset. func AddRookVersionLabelToStatefulSet(ss *apps.StatefulSet) {