Repository: kubernetes
Issue: Feature requests: Support adding label based indexes to apiserver cache
Labels: sig/api-machinery,kind/feature,triage/accepted

### What would you like to be added?

Support adding label based indexes to apiserver cacher to speed up list and watch requests by labels. 

### Why is this needed?

Background: Each job of the big data computing engine spark/flink will start a master pod to create/list/watch/delete worker pods, namely the Driver pod of spark and the JobManager pod of flink.
spark:
list: https://github.com/apache/spark/blob/branch-3.5/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala#L77-L92
watch: https://github.com/apache/spark/blob/branch-3.5/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsWatchSnapshotSource.scala#L63-L67

flink:
list: https://github.com/apache/flink/blob/master/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java#L207-L215
watch: https://github.com/apache/flink/blob/master/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java#L251-L257

In our production k8s cluster, each cluster may run thousands of spark or flink jobs in parallel (the number of jobs may be roughly the same as the number of nodes, or even higher). We found that this will cause great pressure on the apiserver. The number of pod add events sent by the apiserver to the scheduler per second will not keep up with the speed of pod creation per second, resulting in the scheduling speed not matching the speed of pod creation, resulting in resource waste.

Although https://github.com/kubernetes/kubernetes/pull/115918 has added namespace indexes to the cacher's watch function, this function is not so useful in scenarios where a large number of jobs are running in a small number of namespaces. Most events are still sent to each cacheWatcher through the cacher first and then filtered, causing high CPU usage on the apiserver and delays in event delivery.

In fact, each spark and flink job only lists and watches the worker pods of each job through the applicationId label of the job, and does not list and watch the worker pods of other jobs (as we can see from the list/watch code of spark and flink above). This is very similar to the way each pod of daemonsets watches the pod of each node according to spec.nodeName. The number of watchers for these jobs is also comparable to the number of watchers for daemonsets.

Since the apiserver has a watch index of spec.nodeName by default, we also want to add an applicationId index, considering that different scenarios may use different label names. Therefore, we optimized the apiserver to build an index based on certain configurable labels, such as `metadata.labels.app` or `metadata.labels.spark-app-selector`, so that the pod event of each job can query the matching watchers in the cacher's dispatchEvent, without having to broadcast to a large number of unmatched cacheWatchers first and then filter them out by cacheWatcher.

Before implementing this feature, we could only isolate an apiserver for the scheduler to use, and big data jobs or daemonsets would not access this apiserver, so that the number of events received by the scheduler per second could keep up with the number of pods created per second. With this feature, we no longer need to isolate the apiserver separately, and the CPU of the apiserver is also reduced a lot.