Set Stricter Predicates For Compaction Controller (#1158)

* reconcile only compaction jobs, not jobs of any type

* compaction controller predicate to allow reconciles only if necessary job.Status params change

Author: anveshreddy18

internal/controller/compaction/reconciler.go

@@ -297,7 +297,7 @@ func (r *Reconciler) createCompactionJob(ctx context.Context, logger logr.Logger
 					APIVersion:         druidv1alpha1.SchemeGroupVersion.String(),
 					BlockOwnerDeletion: ptr.To(true),
 					Controller:         ptr.To(true),
-					Kind:               "Etcd",
+					Kind:               druidv1alpha1.SchemeGroupVersion.WithKind("Etcd").Kind,
 					Name:               etcd.Name,
 					UID:                etcd.UID,
 				},

internal/controller/compaction/register.go

@@ -12,7 +12,7 @@ import (
 
 	batchv1 "k8s.io/api/batch/v1"
 	coordinationv1 "k8s.io/api/coordination/v1"
-	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -32,7 +32,7 @@ func (r *Reconciler) RegisterWithManager(mgr ctrl.Manager) error {
 		}).
 		For(&druidv1alpha1.Etcd{}).
 		WithEventFilter(predicate.
-			Or(snapshotRevisionChanged(), jobStatusChanged())).
+			Or(snapshotRevisionChanged(), compactionJobStatusChanged())).
 		Owns(&coordinationv1.Lease{}).
 		Owns(&batchv1.Job{}).
 		Complete(r)
@@ -79,8 +79,42 @@ func snapshotRevisionChanged() predicate.Predicate {
 	}
 }
 
-// jobStatusChanged is a predicate that is `true` if the status of a job changes.
-func jobStatusChanged() predicate.Predicate {
+// compactionJobStatusChanged is a predicate that is `true` if the status of a compaction job changes.
+func compactionJobStatusChanged() predicate.Predicate {
+	equalInt32Ptr := func(a, b *int32) bool {
+		if a == nil && b == nil {
+			return true
+		}
+		if a == nil || b == nil {
+			return false
+		}
+		return *a == *b
+	}
+
+	isCompactionJob := func(obj client.Object) bool {
+		job, ok := obj.(*batchv1.Job)
+		if !ok {
+			return false
+		}
+
+		etcdKind := druidv1alpha1.SchemeGroupVersion.WithKind("Etcd").Kind
+		// Extract etcd name from the job's owner reference
+		for _, ownerRef := range job.OwnerReferences {
+			if ownerRef.Kind == etcdKind && ownerRef.APIVersion == druidv1alpha1.SchemeGroupVersion.String() {
+				etcdObjMeta := metav1.ObjectMeta{
+					Name:      ownerRef.Name,
+					Namespace: job.Namespace,
+				}
+				expectedCompactionJobName := druidv1alpha1.GetCompactionJobName(etcdObjMeta)
+				return job.Name == expectedCompactionJobName
+			}
+		}
+		return false
+	}
+
+	// statusChange compares only the critical JobStatus fields that should trigger reconciliation.
+	// It checks Active, Succeeded, Failed, Terminating, and Ready fields.
+	// Returns false if the new job has active pods (Status.Active > 0) to prevent reconciliation during active compaction job execution.
 	statusChange := func(objOld, objNew client.Object) bool {
 		jobOld, ok := objOld.(*batchv1.Job)
 		if !ok {
@@ -90,15 +124,29 @@ func jobStatusChanged() predicate.Predicate {
 		if !ok {
 			return false
 		}
-		return !apiequality.Semantic.DeepEqual(jobOld.Status, jobNew.Status)
+
+		oldStatus := jobOld.Status
+		newStatus := jobNew.Status
+
+		// Prevent reconciliation when the job has active pods
+		if newStatus.Active > 0 {
+			return false
+		}
+
+		// Compare only the critical status fields
+		return oldStatus.Active != newStatus.Active ||
+			oldStatus.Succeeded != newStatus.Succeeded ||
+			oldStatus.Failed != newStatus.Failed ||
+			!equalInt32Ptr(oldStatus.Terminating, newStatus.Terminating) ||
+			!equalInt32Ptr(oldStatus.Ready, newStatus.Ready)
 	}
 
 	return predicate.Funcs{
 		CreateFunc: func(_ event.CreateEvent) bool {
 			return false
 		},
 		UpdateFunc: func(event event.UpdateEvent) bool {
-			return statusChange(event.ObjectOld, event.ObjectNew)
+			return isCompactionJob(event.ObjectNew) && statusChange(event.ObjectOld, event.ObjectNew)
 		},
 		GenericFunc: func(_ event.GenericEvent) bool {
 			return false

internal/controller/compaction/register_test.go

@@ -11,6 +11,9 @@ import (
 	"testing"
 	"time"
 
+	druidv1alpha1 "github.com/gardener/etcd-druid/api/core/v1alpha1"
+	"github.com/gardener/etcd-druid/test/utils"
+
 	batchv1 "k8s.io/api/batch/v1"
 	coordinationv1 "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -166,50 +169,94 @@ func TestJobStatusChangedForUpdateEvents(t *testing.T) {
 	tests := []struct {
 		name                   string
 		isObjectJob            bool
+		isObjectCompactionJob  bool
 		isStatusChanged        bool
 		shouldAllowUpdateEvent bool
 	}{
 		{
 			name:                   "object is not a job",
 			isObjectJob:            false,
+			isObjectCompactionJob:  false,
+			shouldAllowUpdateEvent: false,
+		},
+		{
+			name:                   "object is a non-compaction job, and status is not changed",
+			isObjectJob:            true,
+			isObjectCompactionJob:  false,
+			isStatusChanged:        false,
+			shouldAllowUpdateEvent: false,
+		},
+		{
+			name:                   "object is a non-compaction job, and status is changed",
+			isObjectJob:            true,
+			isObjectCompactionJob:  false,
+			isStatusChanged:        true,
 			shouldAllowUpdateEvent: false,
 		},
 		{
-			name:                   "object is a job, but status is not changed",
+			name:                   "object is a compaction job, but status is not changed",
 			isObjectJob:            true,
+			isObjectCompactionJob:  true,
 			isStatusChanged:        false,
 			shouldAllowUpdateEvent: false,
 		},
 		{
-			name:                   "object is a job, and status is changed",
+			name:                   "object is a compaction job, and status is changed",
 			isObjectJob:            true,
+			isObjectCompactionJob:  true,
 			isStatusChanged:        true,
 			shouldAllowUpdateEvent: true,
 		},
 	}
 
 	g := NewWithT(t)
 	t.Parallel()
-	predicate := jobStatusChanged()
+	predicate := compactionJobStatusChanged()
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
-			obj, oldObj := createObjectsForJobStatusChangedPredicate(g, "etcd-test-compaction-job", test.isObjectJob, test.isStatusChanged)
+			obj, oldObj := createObjectsForJobStatusChangedPredicate(g, druidv1alpha1.GetCompactionJobName(metav1.ObjectMeta{Name: utils.TestEtcdName}), test.isObjectJob, test.isObjectCompactionJob, test.isStatusChanged)
 			g.Expect(predicate.Update(event.UpdateEvent{ObjectOld: oldObj, ObjectNew: obj})).To(Equal(test.shouldAllowUpdateEvent))
 		})
 	}
 }
 
-func createObjectsForJobStatusChangedPredicate(g *WithT, name string, isJobObj, isStatusChanged bool) (obj client.Object, oldObj client.Object) {
+func createObjectsForJobStatusChangedPredicate(g *WithT, name string, isJobObj, isCompactionJob, isStatusChanged bool) (obj client.Object, oldObj client.Object) {
 	// if the object is not a job object, create a config map (random type chosen, could have been anything else as well).
 	if !isJobObj {
 		obj = createConfigMap(g, name)
 		oldObj = createConfigMap(g, name)
 		return
 	}
+	// If the object is a job but not a compaction job, create a regular job
+	if !isCompactionJob {
+		obj = createNonCompactionJob(g, name)
+		oldObj = createNonCompactionJob(g, name)
+		return
+	}
+
 	now := time.Now()
+
+	etcdName := utils.TestEtcdName
+	etcdKind := druidv1alpha1.SchemeGroupVersion.WithKind("Etcd").Kind
+
+	// Create proper owner reference for compaction job
+	ownerRef := metav1.OwnerReference{
+		APIVersion:         druidv1alpha1.SchemeGroupVersion.String(),
+		Kind:               etcdKind,
+		Name:               etcdName,
+		UID:                "test-etcd-uid-12345",
+		Controller:         ptr.To(true),
+		BlockOwnerDeletion: ptr.To(true),
+	}
+
 	// create job objects
 	oldObj = &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            name,
+			Namespace:       utils.TestNamespace,
+			OwnerReferences: []metav1.OwnerReference{ownerRef},
+		},
 		Status: batchv1.JobStatus{
 			Active: 1,
 			StartTime: &metav1.Time{
@@ -219,6 +266,11 @@ func createObjectsForJobStatusChangedPredicate(g *WithT, name string, isJobObj,
 	}
 	if isStatusChanged {
 		obj = &batchv1.Job{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            name,
+				Namespace:       utils.TestNamespace,
+				OwnerReferences: []metav1.OwnerReference{ownerRef},
+			},
 			Status: batchv1.JobStatus{
 				Succeeded: 1,
 				StartTime: &metav1.Time{
@@ -290,6 +342,22 @@ func createConfigMap(g *WithT, name string) *corev1.ConfigMap {
 	}
 }
 
+func createNonCompactionJob(g *WithT, name string) *batchv1.Job {
+	randInt := generateRandomInt(g)
+	return &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: utils.TestNamespace,
+		},
+		Spec: batchv1.JobSpec{
+			ActiveDeadlineSeconds: ptr.To(int64(randInt)),
+		},
+		Status: batchv1.JobStatus{
+			Active: 1,
+		},
+	}
+}
+
 func generateRandomInt(g *WithT) int {
 	randInt, err := rand.Int(rand.Reader, big.NewInt(1000))
 	g.Expect(err).NotTo(HaveOccurred())

test/integration/controllers/compaction/reconciler_test.go

@@ -128,10 +128,13 @@ var _ = Describe("Compaction Controller", func() {
 			}, timeout, pollingInterval).Should(BeNil())
 
 			// Update job status as failed
-			j.Status.Conditions = []batchv1.JobCondition{
-				{
-					Type:   batchv1.JobFailed,
-					Status: corev1.ConditionTrue,
+			j.Status = batchv1.JobStatus{
+				Failed: 1,
+				Conditions: []batchv1.JobCondition{
+					{
+						Type:   batchv1.JobFailed,
+						Status: corev1.ConditionTrue,
+					},
 				},
 			}
 			j.Status.StartTime = &metav1.Time{Time: time.Now()}
@@ -182,10 +185,13 @@ var _ = Describe("Compaction Controller", func() {
 			}, timeout, pollingInterval).Should(BeNil())
 
 			// Update job status as succeeded
-			j.Status.Conditions = []batchv1.JobCondition{
-				{
-					Type:   batchv1.JobComplete,
-					Status: corev1.ConditionTrue,
+			j.Status = batchv1.JobStatus{
+				Succeeded: 1,
+				Conditions: []batchv1.JobCondition{
+					{
+						Type:   batchv1.JobComplete,
+						Status: corev1.ConditionTrue,
+					},
 				},
 			}
 			Expect(k8sClient.Status().Update(context.TODO(), j)).To(Succeed())