feedc0de/my-rook-config
1
0
Fork 0
forked from rook/rook
Code Pull Requests Activity

ceph: update cronjob from v1beta1 to v1

Browse Source 
this commit update cronjob from v1beta1 to v1

Signed-off-by: subhamkrai <srai@redhat.com>
This commit is contained in:
subhamkrai
2021-06-18 08:35:02 +05:30
committed by srai
parent c961d9f7ac
commit f82fc46b97
4 changed files with 181 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/operator/ceph/cluster/crash/add.go
+3 -2
View File
@@ -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,
}
}
pkg/operator/ceph/cluster/crash/crash.go
+66 -38
View File
@@ -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)
pkg/operator/ceph/cluster/crash/crash_test.go
+80
View File
@@ -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))
}
pkg/operator/ceph/cluster/crash/reconcile.go
+32 -9
View File
@@ -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)
}