use sync map for the cache

Author: kmala

pkg/controller/job/job_controller.go

@@ -124,9 +124,9 @@ type Controller struct {
 	// recreation in case of pod failures.
 	podBackoffStore *backoffStore
 
-	// completedJobStore contains the job ids for which the job status is updated to completed
+	// finishedJobStore contains the job ids for which the job status is finished
 	// but the corresponding event is not yet received.
-	completedJobStore *jobUIDCache
+	finishedJobStore sync.Map
 }
 
 type syncJobCtx struct {
@@ -188,9 +188,7 @@ func newControllerWithClock(ctx context.Context, podInformer coreinformers.PodIn
 		recorder:              eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "job-controller"}),
 		clock:                 clock,
 		podBackoffStore:       newBackoffStore(),
-		completedJobStore: &jobUIDCache{
-			set: sets.New[types.UID](),
-		},
+		finishedJobStore:      sync.Map{},
 	}
 
 	if _, err := jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -543,7 +541,7 @@ func (jm *Controller) deleteJob(logger klog.Logger, obj interface{}) {
 			return
 		}
 	}
-	jm.completedJobStore.remove(jobObj.UID)
+	jm.finishedJobStore.Delete(jobObj.UID)
 	jm.enqueueLabelSelector(jobObj)
 }
 
@@ -828,6 +826,7 @@ func (jm *Controller) syncJob(ctx context.Context, key string) (rErr error) {
 		}
 		return err
 	}
+
 	// Skip syncing of the job it is managed by another controller.
 	// We cannot rely solely on skipping of queueing such jobs for synchronization,
 	// because it is possible a synchronization task is queued for a job, without
@@ -848,10 +847,10 @@ func (jm *Controller) syncJob(ctx context.Context, key string) (rErr error) {
 			// re-syncing here as the record has to be removed for finished/deleted jobs
 			return fmt.Errorf("error removing backoff record %w", err)
 		}
-		jm.completedJobStore.remove(job.UID)
+		jm.finishedJobStore.Delete(job.UID)
 		return nil
 	}
-	if jm.completedJobStore.exists(job.UID) {
+	if _, ok := jm.finishedJobStore.Load(job.UID); ok {
 		logger.V(2).Info("Skip syncing the job as its marked completed but the completed update event is not yet received", "uid", job.UID, "key", key)
 		return nil
 	}
@@ -1316,7 +1315,7 @@ func (jm *Controller) trackJobStatusAndRemoveFinalizers(ctx context.Context, job
 		}
 		if jobFinished {
 			jm.recordJobFinished(jobCtx.job, jobCtx.finishedCondition)
-			jm.completedJobStore.add(jobCtx.job.UID)
+			jm.finishedJobStore.Store(jobCtx.job.UID, struct{}{})
 		}
 		recordJobPodFinished(logger, jobCtx.job, oldCounters)
 	}

pkg/controller/job/tracking_utils.go

@@ -151,26 +151,3 @@ func isFinishedPodWithTrackingFinalizer(pod *v1.Pod) bool {
 	}
 	return (pod.Status.Phase == v1.PodFailed || pod.Status.Phase == v1.PodSucceeded) && hasJobTrackingFinalizer(pod)
 }
-
-type jobUIDCache struct {
-	sync.RWMutex
-	set sets.Set[types.UID]
-}
-
-func (j *jobUIDCache) add(uid types.UID) {
-	j.Lock()
-	defer j.Unlock()
-	j.set.Insert(uid)
-}
-
-func (j *jobUIDCache) remove(uid types.UID) {
-	j.Lock()
-	defer j.Unlock()
-	j.set.Delete(uid)
-}
-
-func (j *jobUIDCache) exists(uid types.UID) bool {
-	j.RLock()
-	defer j.RUnlock()
-	return j.set.Has(uid)
-}