Repository: kubernetes/kubernetes (language: go)

Issue:
The event handlers of job controller are slow
### What happened?

The event handlers in job controller are much slower than other controllers. It can cause the ring buffer in the processorListener to grow unbounded when the job churn is high enough.
The processing logic in the job controller events handler appears to be more complicated than other controllers. e.g. in [updateJob](https://github.com/kubernetes/kubernetes/blob/dbc2b0a5c7acc349ea71a14e49913661eaf708d2/pkg/controller/job/job_controller.go#L464-L504), it invokes function `cleanupPodFinalizers` that lists the pods matching some selectors and the enqueue them to another queue. Similarly in [deleteJob](https://github.com/kubernetes/kubernetes/blob/dbc2b0a5c7acc349ea71a14e49913661eaf708d2/pkg/controller/job/job_controller.go#L508-L524).

I added some instrumenting code to understand the latency of each event handler.
For job controller P99:
job add event handler: 0.080 ms
job update event handler: 3.075 ms
job delete event handler: 3.076 ms
pod add event handler: 0.136 ms
pod update event handler: 0.709 ms
pod delete event handler: 0.405 ms

For ttl-after-completion controller P99:
job add event handler: 0.0065 ms
job update event handler: 0.024 ms

The above 2 controllers processes same amount of job objects.

### What did you expect to happen?

Complicated logic should not be in the job controller event handler. They should be in the reconciler instead.
So that the event handler can be faster and pendingNotification can be drained faster.

### How can we reproduce it (as minimally and precisely as possible)?

I use a few thousands of cronjobs with TTL after completion.
I will add more details.

### Anything else we need to know?

_No response_

### Kubernetes version

1.30
<details>

Current HEAD of master
</details>


### Cloud provider

EKS



### OS version

<details>

```console
# On Linux:
$ cat /etc/os-release
# paste output here
$ uname -a
# paste output here

# On Windows:
C:\> wmic os get Caption, Version, BuildNumber, OSArchitecture
# paste output here
```

</details>


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>



Discussion / hints:
This issue is currently awaiting triage.

If a SIG or subproject determines this is a relevant issue, they will accept it by applying the `triage/accepted` label and provide further guidance.

The `triage/accepted` label can be added by org members by writing `/triage accepted` in a comment.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/job/job_controller.go#L485-L487

is most probably the offender - without looking deeper into the code, this should most probably go to the worker function

/cc @mimowo @alculquicondor 
cc: @soltysh 
@mengqiy thanks for reporting the issue. 

To help triage the issue I traced down the introduction of `cleanupPodFinalizers` code:
- for the `deleteJob` it was introduced in 1.24: https://github.com/kubernetes/kubernetes/pull/108752
- for the `updateJob` it was introduced in 1.29: https://github.com/kubernetes/kubernetes/pull/119944

The `updateJob` seems to be expensive only when finished Jobs are updated. 
Given we only run one single goroutine for `run()` which is responsible for invoking the event handler https://github.com/kubernetes/kubernetes/blob/dbc2b0a5c7acc349ea71a14e49913661eaf708d2/staging/src/k8s.io/client-go/tools/cache/shared_informer.go#L802
this can easily be the bottleneck in a controller whenever the event handler is slow.
The event handler that used by the informer is expected to be very thin and fast. It seems `cleanupPodFinalizers` has some expansive logic and slow down updateJob and deleteJob. 
IMO we should move `cleanupPodFinalizers` to the reconciler. User can easily increase the numbers of goroutines in the job controller by using higher `--concurrent-cron-job-syncs` to increase the parallelism.

/sig apps

I was wondering @mengqiy  given you already have the metrics instrumentation code in place, can we quickly remove `cleanupPodFinalizers` from `event handler` and crunch the numbers to see how much it will benefit ?? Leaving the correctness of job controller aside for testing purposes ?? (as discussed offline )
I think this should work - I don't see anything that would block achieving correctness, let us know if some complications are exposed by tests.
/subscribe
/cc @atiratree 
@mengqiy how can we reproduce this? What jobs did you use?

>  It can cause the ring buffer in the processorListener to grow unbounded when the job churn is high enough.

Have you observed this or do you have any data on the limitations? 
repro and also a pprof may be useful
Will find some time to add the detailed steps to repro this week.
A simple way to reproduce this is creating about 10k copy of the following cronjobs. In case 10k doesn't reproduce it for your setup, adding 1 or 2k more should be able to do it.
Each cronjob creates a job every minute.
```
apiVersion: batch/v1
kind: CronJob
metadata:
  generateName: simple-
  namespace: kube-stress
spec:
  schedule: "* * * * *"
  successfulJobsHistoryLimit: 2
  failedJobsHistoryLimit: 2
  jobTemplate:
    spec:
      ttlSecondsAfterFinished: 30
      template:
        spec:
          automountServiceAccountToken: false
          containers:
            - name: simple
              image: busybox
              command: ["sleep", "10"]
              resources:
                requests:
                  cpu: 5m
                  memory: 10Mi
                limits:
                  cpu: 5m
                  memory: 10Mi
          restartPolicy: OnFailure
```

Control plane configuration:
kube-controller-manager:
- qps: 500
- concurrent-job-syncs: 50

kube-scheduler:
- qps: 400

These configuration is large enough to ensure QPS doesn't become a bottleneck.

I added [this commit](https://github.com/mengqiy/kubernetes/commit/8114fc925f16deaf88ee2e21ee18bc030babfbec) to show some metrics about event processing latency and pending notification count. It is a bit hacky.




![Screenshot 2024-08-28 at 12 36 16 AM](https://github.com/user-attachments/assets/7c38cdd7-ac4d-4480-937e-d2c5e0294c6a)

Would you like to propose a fix?
We may need to track job namespacedName to selectors mapping.
For deletion events, we can see the object in the event handler. But we won't be able to get it from the cache after that.
Some logic needs the selector info to get its pods. So we may need to maintain a mapping from job namespacedName to selectors.
@alculquicondor I'm a bit new to the job controller codebase, if there are something existing to leverage or you have better ideas, please LMK.
Maybe we don't need to over optimize. Just removing the logic for the update events into the reconciler could be enough to prevent the processorListener buffer from growing too much.

Or... we could change the orphanQueue reconciler to accept one of two things: a pod name or a pod selector.

Fix patch:
diff --git a/pkg/controller/job/job_controller.go b/pkg/controller/job/job_controller.go
index df8d9367f2825..c00ffb6da3fc7 100644
--- a/pkg/controller/job/job_controller.go
+++ b/pkg/controller/job/job_controller.go
@@ -479,12 +479,6 @@ func (jm *Controller) updateJob(logger klog.Logger, old, cur interface{}) {
 		jm.enqueueSyncJobImmediately(logger, curJob)
 	}
 
-	// The job shouldn't be marked as finished until all pod finalizers are removed.
-	// This is a backup operation in this case.
-	if util.IsJobFinished(curJob) {
-		jm.cleanupPodFinalizers(curJob)
-	}
-
 	// check if need to add a new rsync for ActiveDeadlineSeconds
 	if curJob.Status.StartTime != nil {
 		curADS := curJob.Spec.ActiveDeadlineSeconds
@@ -772,6 +766,9 @@ func (jm *Controller) syncJob(ctx context.Context, key string) (rErr error) {
 
 	// if job was finished previously, we don't want to redo the termination
 	if util.IsJobFinished(&job) {
+		// The job shouldn't be marked as finished until all pod finalizers are removed.
+		// Cleaning pod finalizers one more time just in case.
+		jm.cleanupPodFinalizers(&job)
 		err := jm.podBackoffStore.removeBackoffRecord(key)
 		if err != nil {
 			// re-syncing here as the record has to be removed for finished/deleted jobs


Test patch:
diff --git a/pkg/controller/job/job_controller_test.go b/pkg/controller/job/job_controller_test.go
index 9bcee76ef981a..1e5fbdc7bab88 100644
--- a/pkg/controller/job/job_controller_test.go
+++ b/pkg/controller/job/job_controller_test.go
@@ -7624,9 +7624,9 @@ func TestFinalizerCleanup(t *testing.T) {
 	// Start the Pod and Job informers.
 	sharedInformers.Start(ctx.Done())
 	sharedInformers.WaitForCacheSync(ctx.Done())
-	// Initialize the controller with 0 workers to make sure the
-	// pod finalizers are not removed by the "syncJob" function.
-	go manager.Run(ctx, 0)
+	// Initialize the controller with 1 worker to make sure the
+	// pod finalizers are removed by the "syncJob" function.
+	go manager.Run(ctx, 1)
 	// Make sure the pod finalizers are removed by the "orphanWorker" function.
 	go wait.UntilWithContext(ctx, manager.orphanWorker, time.Second)
 
@@ -7645,22 +7645,24 @@ func TestFinalizerCleanup(t *testing.T) {
 		t.Fatalf("Waiting for Job object to appear in jobLister: %v", err)
 	}
 
-	// Create a Pod with the job tracking finalizer
-	pod := newPod("test-pod", job)
-	pod.Finalizers = append(pod.Finalizers, batch.JobTrackingFinalizer)
-	pod, err = clientset.CoreV1().Pods(pod.GetNamespace()).Create(ctx, pod, metav1.CreateOptions{})
+	selector, err := metav1.LabelSelectorAsSelector(job.Spec.Selector)
 	if err != nil {
-		t.Fatalf("Creating pod: %v", err)
+		t.Fatalf("Error parsing job selector: %v", err)
 	}
 
-	// Await for the Pod to appear in the podStore to ensure that the pod exists when cleaning up the Job.
-	// In a production environment, there wouldn't be these guarantees, but the Pod would be cleaned up
-	// by the orphan pod worker, when the Pod finishes.
+	var pods []*v1.Pod
+	// Wait for the Pod to be created by the Job controller
 	if err := wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, wait.ForeverTestTimeout, false, func(ctx context.Context) (bool, error) {
-		pod, _ := manager.podStore.Pods(pod.GetNamespace()).Get(pod.Name)
-		return pod != nil, nil
+		pods, err = manager.podStore.Pods(job.Namespace).List(selector)
+		if err != nil {
+			return false, err
+		}
+		if len(pods) > 0 {
+			return true, nil
+		}
+		return false, nil
 	}); err != nil {
-		t.Fatalf("Waiting for Pod to appear in podLister: %v", err)
+		t.Fatalf("Waiting for Pod to be created by Job: %v", err)
 	}
 
 	// Mark Job as complete.
@@ -7673,10 +7675,9 @@ func TestFinalizerCleanup(t *testing.T) {
 		t.Fatalf("Updating job status: %v", err)
 	}
 
-	// Verify the pod finalizer is removed for a finished Job,
-	// even if the jobs pods are not tracked by the main reconciliation loop.
+	// Verify the pod finalizer is removed for a finished Job
 	if err := wait.PollUntilContextTimeout(ctx, 100*time.Millisecond, wait.ForeverTestTimeout, true, func(ctx context.Context) (bool, error) {
-		p, err := clientset.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{})
+		p, err := clientset.CoreV1().Pods(pods[0].Namespace).Get(ctx, pods[0].Name, metav1.GetOptions{})
 		if err != nil {
 			return false, err
 		}