Added test case for lease controller in separate test file.

charts/etcd/templates/etcd-statefulset.yaml

@@ -217,9 +217,13 @@ spec:
         - name: STORAGE_CONTAINER
           value: {{ .Values.store.storageContainer }}
         - name: POD_NAME
-          value: {{ .Values.name }}
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.name
         - name: POD_NAMESPACE
-          value: {{ .Release.Namespace }}
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
 {{- if eq .Values.store.storageProvider "S3" }}
         - name: "AWS_REGION"
           valueFrom:

controllers/config/compaction.go

@@ -20,8 +20,6 @@ import "time"
 type CompactionConfig struct {
 	// ActiveDeadlineDuration is the duration after which a running compaction job will be killed (Ex: "300ms", "20s", "-1.5h" or "2h45m")
 	ActiveDeadlineDuration time.Duration
-	// EnableCompactionJob enables compaction job to run in parallel
-	EnableCompactionJob bool
 	// EventsThreshold is Total number of events that can be allowed before a compaction job is triggered
 	EventsThreshold int64
 }

controllers/controller_ref_manager.go

@@ -686,7 +686,6 @@ func getMapFromEtcd(im imagevector.ImageVector, etcd *druidv1alpha1.Etcd) (map[s
 		"labels":                  etcd.Spec.Labels,
 		"annotations":             etcd.Spec.Annotations,
 		"etcd":                    etcdValues,
-		"serviceAccountName":      getServiceAccountName(etcd),
 		"backup":                  backupValues,
 		"sharedConfig":            sharedConfigValues,
 		"replicas":                etcd.Spec.Replicas,

controllers/controllers_suite_test.go

@@ -50,6 +50,9 @@ var (
 	mgr        manager.Manager
 	mgrStopped *sync.WaitGroup
 
+	activeDeadlineDuration   time.Duration
+	backupCompactionSchedule = "15 */24 * * *"
+
 	revertFns []func()
 
 	testLog = ctrl.Log.WithName("test")
@@ -120,14 +123,14 @@ var _ = BeforeSuite(func(done Done) {
 	Expect(err).NotTo(HaveOccurred())
 
 	err = etcdCopyBackupsTaskReconciler.SetupWithManager(mgr, 1)
+	Expect(err).NotTo(HaveOccurred())
 
-	activeDeadlineDuration, err := time.ParseDuration("2m")
+	activeDeadlineDuration, err = time.ParseDuration("2m")
 	Expect(err).NotTo(HaveOccurred())
 
 	lc, err := NewLeaseControllerWithImageVector(mgr, controllersconfig.CompactionConfig{
 		EventsThreshold:        1000000,
 		ActiveDeadlineDuration: activeDeadlineDuration,
-		EnableCompactionJob:    true,
 	})
 	Expect(err).NotTo(HaveOccurred())
 

controllers/etcd_controller.go

@@ -27,6 +27,7 @@ import (
 	"github.com/gardener/etcd-druid/pkg/common"
 	druidpredicates "github.com/gardener/etcd-druid/pkg/predicate"
 	"github.com/gardener/etcd-druid/pkg/utils"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	coordinationv1 "k8s.io/api/coordination/v1"
 
 	extensionspredicate "github.com/gardener/gardener/extensions/pkg/predicate"
@@ -42,6 +43,7 @@ import (
 	eventsv1 "k8s.io/api/events/v1"
 	eventsv1beta1 "k8s.io/api/events/v1beta1"
 	rbac "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
@@ -276,6 +278,19 @@ func (r *EtcdReconciler) reconcile(ctx context.Context, etcd *druidv1alpha1.Etcd
 	}
 	logger.Info("Available Delta Snapshot Lease: " + dl.Name)
 
+	// Delete any existing cronjob if required.
+	// TODO(abdasgupta) : This is for backward compatibility towards ETCD-Druid 0.6.0. Remove it.
+	cronJob, err := r.cleanCronJobs(ctx, logger, etcd)
+	if err != nil {
+		return ctrl.Result{
+			Requeue: true,
+		}, fmt.Errorf("error while cleaning compaction cron job: %v", err)
+	}
+
+	if cronJob != nil {
+		logger.Info("The running cron job is: " + cronJob.Name)
+	}
+
 	op, svc, sts, err := r.reconcileEtcd(ctx, logger, etcd)
 	if err != nil {
 		if err := r.updateEtcdErrorStatus(ctx, op, etcd, sts, err); err != nil {
@@ -299,10 +314,58 @@ func (r *EtcdReconciler) reconcile(ctx context.Context, etcd *druidv1alpha1.Etcd
 	}, nil
 }
 
+func (r *EtcdReconciler) cleanCronJobs(ctx context.Context, logger logr.Logger, etcd *druidv1alpha1.Etcd) (*batchv1beta1.CronJob, error) {
+	cronJob := &batchv1beta1.CronJob{}
+	err := r.Get(ctx, types.NamespacedName{Name: getCronJobName(etcd), Namespace: etcd.Namespace}, cronJob)
+	if err != nil {
+		if !errors.IsNotFound(err) {
+			return nil, err
+		}
+
+		return nil, nil
+	}
+
+	// Delete cronJob if there is no active job
+	if len(cronJob.Status.Active) == 0 {
+		err = client.IgnoreNotFound(r.Delete(ctx, cronJob, client.PropagationPolicy(metav1.DeletePropagationForeground)))
+		if err != nil {
+			return nil, err
+		}
+		return nil, nil
+	}
+
+	// Calculate time elapsed since the cron job is scheduled
+	timeElapsed := time.Since(cronJob.Status.LastScheduleTime.Time).Seconds()
+	// Delete the cron job if it's running for more than 3 hours
+	if timeElapsed > time.Duration(3*time.Hour).Seconds() {
+		if err := client.IgnoreNotFound(r.Delete(ctx, cronJob, client.PropagationPolicy(metav1.DeletePropagationForeground))); err != nil {
+			return nil, err
+		}
+		logger.Info("last cron job was stuck and deleted")
+		return nil, nil
+	}
+	return cronJob, nil
+}
+
 func (r *EtcdReconciler) delete(ctx context.Context, etcd *druidv1alpha1.Etcd) (ctrl.Result, error) {
 	logger := r.logger.WithValues("etcd", kutil.Key(etcd.Namespace, etcd.Name).String(), "operation", "delete")
 	logger.Info("Starting operation")
 
+	// TODO(abdasgupta) : This is for backward compatibility towards ETCD-Druid 0.6.0. Remove it.
+	cronJob := &batchv1beta1.CronJob{}
+	if err := client.IgnoreNotFound(r.Get(ctx, types.NamespacedName{Name: getCronJobName(etcd), Namespace: etcd.Namespace}, cronJob)); err != nil {
+		return ctrl.Result{RequeueAfter: 10 * time.Second}, fmt.Errorf("error while fetching compaction cron job: %v", err)
+	}
+
+	if cronJob.Name == getCronJobName(etcd) && cronJob.DeletionTimestamp == nil {
+		logger.Info("Deleting cron job", "cronjob", kutil.ObjectName(cronJob))
+		if err := client.IgnoreNotFound(r.Delete(ctx, cronJob, client.PropagationPolicy(metav1.DeletePropagationForeground))); err != nil {
+			return ctrl.Result{
+				Requeue: true,
+			}, fmt.Errorf("error while deleting compaction cron job: %v", err)
+		}
+	}
+
 	if waitForStatefulSetCleanup, err := r.removeDependantStatefulset(ctx, logger, etcd); err != nil {
 		if err = r.updateEtcdErrorStatus(ctx, deleteOp, etcd, nil, err); err != nil {
 			return ctrl.Result{
@@ -882,7 +945,7 @@ func (r *EtcdReconciler) reconcileFullLease(ctx context.Context, logger logr.Log
 }
 
 func getFullLease(etcd *druidv1alpha1.Etcd) string {
-	return fmt.Sprintf("%s-full-snap", string(etcd.UID[:6]))
+	return fmt.Sprintf("%s-full-snap", string(etcd.Name))
 }
 
 func (r *EtcdReconciler) reconcileDeltaLease(ctx context.Context, logger logr.Logger, etcd *druidv1alpha1.Etcd) (*coordinationv1.Lease, error) {
@@ -957,7 +1020,7 @@ func (r *EtcdReconciler) reconcileDeltaLease(ctx context.Context, logger logr.Lo
 }
 
 func getDeltaLease(etcd *druidv1alpha1.Etcd) string {
-	return fmt.Sprintf("%s-delta-snap", string(etcd.UID[:6]))
+	return fmt.Sprintf("%s-delta-snap", string(etcd.Name))
 }
 
 func decodeObject(renderedChart *chartrenderer.RenderedChart, path string, object interface{}) error {
@@ -1379,10 +1442,6 @@ func (r *EtcdReconciler) fetchPVCEventsFor(ctx context.Context, ss *appsv1.State
 	return pvcMessages, nil
 }
 
-func getServiceAccountName(etcd *druidv1alpha1.Etcd) string {
-	return fmt.Sprintf("%s-br-serviceaccount", etcd.Name)
-}
-
 func (r *EtcdReconciler) removeOperationAnnotation(ctx context.Context, logger logr.Logger, etcd *druidv1alpha1.Etcd) error {
 	if _, ok := etcd.Annotations[v1beta1constants.GardenerOperation]; ok {
 		logger.Info("Removing operation annotation")