diff --git a/pkg/operator/ceph/cluster/crash/add.go b/pkg/operator/ceph/cluster/crash/add.go index 3816b244b..40a755c65 100644 --- a/pkg/operator/ceph/cluster/crash/add.go +++ b/pkg/operator/ceph/cluster/crash/add.go @@ -54,8 +54,9 @@ func Add(mgr manager.Manager, context *clusterd.Context) error { // newReconciler returns a new reconcile.Reconciler func newReconciler(mgr manager.Manager, context *clusterd.Context) reconcile.Reconciler { return &ReconcileNode{ - client: mgr.GetClient(), - scheme: mgr.GetScheme(), + client: mgr.GetClient(), + scheme: mgr.GetScheme(), + context: context, } } diff --git a/pkg/operator/ceph/cluster/crash/crash.go b/pkg/operator/ceph/cluster/crash/crash.go index 814a980f8..8ff255ac4 100644 --- a/pkg/operator/ceph/cluster/crash/crash.go +++ b/pkg/operator/ceph/cluster/crash/crash.go @@ -21,16 +21,20 @@ import ( "fmt" "path" + "k8s.io/api/batch/v1beta1" + "github.com/pkg/errors" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/operator/ceph/config" "github.com/rook/rook/pkg/operator/ceph/config/keyring" "github.com/rook/rook/pkg/operator/ceph/controller" - "github.com/rook/rook/pkg/operator/ceph/version" + + cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" appsv1 "k8s.io/api/apps/v1" - "k8s.io/api/batch/v1beta1" + v1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" ) @@ -43,7 +47,7 @@ const ( ) // createOrUpdateCephCrash is a wrapper around controllerutil.CreateOrUpdate -func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []corev1.Toleration, cephCluster cephv1.CephCluster, cephVersion *version.CephVersion) (controllerutil.OperationResult, error) { +func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []corev1.Toleration, cephCluster cephv1.CephCluster, cephVersion *cephver.CephVersion) (controllerutil.OperationResult, error) { // Create or Update the deployment default/foo nodeHostnameLabel, ok := node.ObjectMeta.Labels[corev1.LabelHostname] if !ok { @@ -123,50 +127,74 @@ func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations [] } // createOrUpdateCephCron is a wrapper around controllerutil.CreateOrUpdate -func (r *ReconcileNode) createOrUpdateCephCron(cephCluster cephv1.CephCluster, cephVersion *version.CephVersion) (controllerutil.OperationResult, error) { - cronJob := &v1beta1.CronJob{ +func (r *ReconcileNode) createOrUpdateCephCron(cephCluster cephv1.CephCluster, cephVersion *cephver.CephVersion, useCronJobV1 bool) (controllerutil.OperationResult, error) { + objectMeta := metav1.ObjectMeta{ + Name: prunerName, + Namespace: cephCluster.GetNamespace(), + } + // Adding volumes to pods containing data needed to connect to the ceph cluster. + volumes := controller.DaemonVolumesBase(config.NewDatalessDaemonDataPathMap(cephCluster.GetNamespace(), cephCluster.Spec.DataDirHostPath), "") + volumes = append(volumes, keyring.Volume().CrashCollector()) + + // labels for the pod, the deployment, and the deploymentSelector + cronJobLabels := map[string]string{ + k8sutil.AppAttr: prunerName, + } + cronJobLabels[k8sutil.ClusterAttr] = cephCluster.GetNamespace() + + podTemplateSpec := corev1.PodTemplateSpec{ ObjectMeta: metav1.ObjectMeta{ - Name: prunerName, - Namespace: cephCluster.GetNamespace(), + Labels: cronJobLabels, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{ + getCrashPruneContainer(cephCluster, *cephVersion), + }, + RestartPolicy: corev1.RestartPolicyNever, + HostNetwork: cephCluster.Spec.Network.IsHost(), + Volumes: volumes, }, } + + // After 100 failures, the cron job will no longer run. + // To avoid this, the cronjob is configured to only count the failures + // that occurred in the last hour. + deadline := int64(60) + + // minimum k8s version required for v1 cronJob is 'v1.21.0'. Apply v1 if k8s version is at least 'v1.21.0', else apply v1beta1 cronJob. + if useCronJobV1 { + // delete v1beta1 cronJob if it already exists + err := r.client.Delete(context.TODO(), &v1beta1.CronJob{ObjectMeta: objectMeta}) + if err != nil && !apierrors.IsNotFound(err) { + return controllerutil.OperationResultNone, errors.Wrapf(err, "failed to delete CronJob v1Beta1 %q", prunerName) + } + + cronJob := &v1.CronJob{ObjectMeta: objectMeta} + err = controllerutil.SetControllerReference(&cephCluster, cronJob, r.scheme) + if err != nil { + return controllerutil.OperationResultNone, errors.Errorf("failed to set owner reference of deployment %q", cronJob.Name) + } + mutateFunc := func() error { + cronJob.ObjectMeta.Labels = cronJobLabels + cronJob.Spec.JobTemplate.Spec.Template = podTemplateSpec + cronJob.Spec.Schedule = pruneSchedule + cronJob.Spec.StartingDeadlineSeconds = &deadline + + return nil + } + + return controllerutil.CreateOrUpdate(context.TODO(), r.client, cronJob, mutateFunc) + } + cronJob := &v1beta1.CronJob{ObjectMeta: objectMeta} err := controllerutil.SetControllerReference(&cephCluster, cronJob, r.scheme) if err != nil { return controllerutil.OperationResultNone, errors.Errorf("failed to set owner reference of deployment %q", cronJob.Name) } - // Adding volumes to pods containing data needed to connect to the ceph cluster. - volumes := controller.DaemonVolumesBase(config.NewDatalessDaemonDataPathMap(cephCluster.GetNamespace(), cephCluster.Spec.DataDirHostPath), "") - volumes = append(volumes, keyring.Volume().CrashCollector()) - mutateFunc := func() error { - - // labels for the pod, the deployment, and the deploymentSelector - cronJobLabels := map[string]string{ - k8sutil.AppAttr: prunerName, - } - cronJobLabels[k8sutil.ClusterAttr] = cephCluster.GetNamespace() - cronJob.ObjectMeta.Labels = cronJobLabels - cronJob.Spec.JobTemplate.Spec.Template = corev1.PodTemplateSpec{ - ObjectMeta: metav1.ObjectMeta{ - Labels: cronJobLabels, - }, - Spec: corev1.PodSpec{ - Containers: []corev1.Container{ - getCrashPruneContainer(cephCluster, *cephVersion), - }, - RestartPolicy: corev1.RestartPolicyNever, - HostNetwork: cephCluster.Spec.Network.IsHost(), - Volumes: volumes, - }, - } - + cronJob.Spec.JobTemplate.Spec.Template = podTemplateSpec cronJob.Spec.Schedule = pruneSchedule - // After 100 failures, the cron job will no longer run. - // To avoid this, the cronjob is configured to only count the failures - // that occurred in the last hour. - deadline := int64(60) cronJob.Spec.StartingDeadlineSeconds = &deadline return nil @@ -208,7 +236,7 @@ func getCrashChownInitContainer(cephCluster cephv1.CephCluster) corev1.Container ) } -func getCrashDaemonContainer(cephCluster cephv1.CephCluster, cephVersion version.CephVersion) corev1.Container { +func getCrashDaemonContainer(cephCluster cephv1.CephCluster, cephVersion cephver.CephVersion) corev1.Container { cephImage := cephCluster.Spec.CephVersion.Image dataPathMap := config.NewDatalessDaemonDataPathMap(cephCluster.GetNamespace(), cephCluster.Spec.DataDirHostPath) crashEnvVar := generateCrashEnvVar() @@ -231,7 +259,7 @@ func getCrashDaemonContainer(cephCluster cephv1.CephCluster, cephVersion version return container } -func getCrashPruneContainer(cephCluster cephv1.CephCluster, cephVersion version.CephVersion) corev1.Container { +func getCrashPruneContainer(cephCluster cephv1.CephCluster, cephVersion cephver.CephVersion) corev1.Container { cephImage := cephCluster.Spec.CephVersion.Image envVars := append(controller.DaemonEnvVars(cephImage), generateCrashEnvVar()) dataPathMap := config.NewDatalessDaemonDataPathMap(cephCluster.GetNamespace(), cephCluster.Spec.DataDirHostPath) diff --git a/pkg/operator/ceph/cluster/crash/crash_test.go b/pkg/operator/ceph/cluster/crash/crash_test.go index 7ef16d499..fa4d9786e 100644 --- a/pkg/operator/ceph/cluster/crash/crash_test.go +++ b/pkg/operator/ceph/cluster/crash/crash_test.go @@ -17,9 +17,24 @@ limitations under the License. package crash import ( + "context" "testing" + cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" + rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake" + "github.com/rook/rook/pkg/client/clientset/versioned/scheme" + "github.com/rook/rook/pkg/clusterd" + cephver "github.com/rook/rook/pkg/operator/ceph/version" + "github.com/rook/rook/pkg/operator/test" "github.com/stretchr/testify/assert" + "github.com/tevino/abool" + v1 "k8s.io/api/batch/v1" + "k8s.io/api/batch/v1beta1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + cntrlutil "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" ) func TestGenerateCrashEnvVar(t *testing.T) { @@ -27,3 +42,68 @@ func TestGenerateCrashEnvVar(t *testing.T) { assert.Equal(t, "CEPH_ARGS", env.Name) assert.Equal(t, "-m $(ROOK_CEPH_MON_HOST) -k /etc/ceph/crash-collector-keyring-store/keyring", env.Value) } + +func TestCreateOrUpdateCephCron(t *testing.T) { + cephCluster := cephv1.CephCluster{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph"}} + cephVersion := &cephver.CephVersion{Major: 16, Minor: 2, Extra: 0} + ctx := context.TODO() + context := &clusterd.Context{ + Clientset: test.New(t, 1), + RookClientset: rookclient.NewSimpleClientset(), + RequestCancelOrchestration: abool.New(), + } + + s := scheme.Scheme + err := v1.AddToScheme(s) + if err != nil { + assert.Fail(t, "failed to build scheme") + } + err = v1beta1.AddToScheme(s) + if err != nil { + assert.Fail(t, "failed to build scheme") + } + + r := &ReconcileNode{ + scheme: s, + client: fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects().Build(), + context: context, + } + + cronV1 := &v1.CronJob{ + ObjectMeta: metav1.ObjectMeta{ + Name: prunerName, + Namespace: "rook-ceph", + }, + } + + cronV1Beta1 := &v1beta1.CronJob{ + ObjectMeta: metav1.ObjectMeta{ + Name: prunerName, + Namespace: "rook-ceph", + }, + } + + // check if v1beta1 cronJob is present and v1 cronJob is not + controllerutil, err := r.createOrUpdateCephCron(cephCluster, cephVersion, false) + assert.NoError(t, err) + assert.Equal(t, controllerutil, cntrlutil.OperationResult("created")) + + err = r.client.Get(ctx, types.NamespacedName{Namespace: "rook-ceph", Name: prunerName}, cronV1Beta1) + assert.NoError(t, err) + + err = r.client.Get(ctx, types.NamespacedName{Namespace: "rook-ceph", Name: prunerName}, cronV1) + assert.Error(t, err) + assert.True(t, kerrors.IsNotFound(err)) + + // check if v1 cronJob is present and v1beta1 cronJob is not + controllerutil, err = r.createOrUpdateCephCron(cephCluster, cephVersion, true) + assert.NoError(t, err) + assert.Equal(t, controllerutil, cntrlutil.OperationResult("created")) + + err = r.client.Get(ctx, types.NamespacedName{Namespace: "rook-ceph", Name: prunerName}, cronV1) + assert.NoError(t, err) + + err = r.client.Get(ctx, types.NamespacedName{Namespace: "rook-ceph", Name: prunerName}, cronV1Beta1) + assert.Error(t, err) + assert.True(t, kerrors.IsNotFound(err)) +} diff --git a/pkg/operator/ceph/cluster/crash/reconcile.go b/pkg/operator/ceph/cluster/crash/reconcile.go index 26e07b616..67ee3f2b4 100644 --- a/pkg/operator/ceph/cluster/crash/reconcile.go +++ b/pkg/operator/ceph/cluster/crash/reconcile.go @@ -21,12 +21,16 @@ import ( "time" "github.com/pkg/errors" + "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/ceph/cluster/mgr" "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/cluster/rbd" appsv1 "k8s.io/api/apps/v1" + v1 "k8s.io/api/batch/v1" + "k8s.io/api/batch/v1beta1" kerrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/version" "github.com/rook/rook/pkg/operator/ceph/file/mds" "github.com/rook/rook/pkg/operator/ceph/file/mirror" @@ -45,7 +49,6 @@ import ( opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" "github.com/rook/rook/pkg/operator/ceph/disruption/controllerconfig" cephver "github.com/rook/rook/pkg/operator/ceph/version" - "k8s.io/api/batch/v1beta1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -57,13 +60,15 @@ var ( // wait for secret "rook-ceph-crash-collector-keyring" to be created waitForRequeueIfSecretNotCreated = reconcile.Result{Requeue: true, RequeueAfter: 30 * time.Second} + minVersionForCronV1 = "1.21.0" ) // ReconcileNode reconciles ReplicaSets type ReconcileNode struct { // client can be used to retrieve objects from the APIServer. - scheme *runtime.Scheme - client client.Client + scheme *runtime.Scheme + client client.Client + context *clusterd.Context } // Reconcile reconciles a node and ensures that it has a crashcollector deployment @@ -272,13 +277,31 @@ func (r *ReconcileNode) deleteCrashCollector(deployment appsv1.Deployment) error } func (r *ReconcileNode) reconcileCrashRetention(namespace string, cephCluster cephv1.CephCluster, cephVersion *cephver.CephVersion) error { + k8sVersion, err := k8sutil.GetK8SVersion(r.context.Clientset) + if err != nil { + return errors.Wrap(err, "failed to fetch get k8s version") + } + useCronJobV1 := k8sVersion.AtLeast(version.MustParseSemantic(minVersionForCronV1)) + + objectMeta := metav1.ObjectMeta{ + Name: prunerName, + Namespace: namespace, + } + if cephCluster.Spec.CrashCollector.DaysToRetain == 0 { logger.Debug("deleting cronjob if it exists...") - cronJob := &v1beta1.CronJob{ - ObjectMeta: metav1.ObjectMeta{ - Name: prunerName, - Namespace: namespace, - }, + + var cronJob client.Object + // minimum k8s version required for v1 cronJob is 'v1.21.0'. Apply v1 if k8s version is at least 'v1.21.0', else apply v1beta1 cronJob. + if useCronJobV1 { + // delete v1beta1 cronJob if it already exists + err = r.client.Delete(context.TODO(), &v1beta1.CronJob{ObjectMeta: objectMeta}) + if err != nil && !kerrors.IsNotFound(err) { + return errors.Wrapf(err, "failed to delete CronJob v1beta1 %q", prunerName) + } + cronJob = &v1.CronJob{ObjectMeta: objectMeta} + } else { + cronJob = &v1beta1.CronJob{ObjectMeta: objectMeta} } err := r.client.Delete(context.TODO(), cronJob) @@ -293,7 +316,7 @@ func (r *ReconcileNode) reconcileCrashRetention(namespace string, cephCluster ce } } else { logger.Debugf("daysToRetain set to: %d", cephCluster.Spec.CrashCollector.DaysToRetain) - op, err := r.createOrUpdateCephCron(cephCluster, cephVersion) + op, err := r.createOrUpdateCephCron(cephCluster, cephVersion, useCronJobV1) if err != nil { return errors.Wrapf(err, "node reconcile failed on op %q", op) }