Repository: kubernetes
Issue: Watch with resourceVersion="" can take down control plane
Labels: kind/bug,sig/api-machinery,triage/accepted,sig/etcd

### What happened?

I was investigating a customer case where the Kubernetes control plane was unavailable for prolonged time due to pod Watch being broken. I was suspecting an issue like https://github.com/etcd-io/etcd/issues/15402. The only clues I had was large sized of pods (>50KBs), watch cache being stale (`rage(apiserver_watch_cache_events_dispatched_total[1m])` equal zero) and a lot of errors `Fast watcher, slow processing` and occasional `prevKV=nil`.

Due to lack of proper watch instrumentation in kube-apiserver, I was only able to reproduce a similar situation on my local cluster. What I found is that even in large clusters with thousands of watches, misguided clients can still open a watch with `resourceVersion=""` that will directly read from etcd and if left inactive, can take down the whole control plane.

In my testing I have observed that slow/inactive watch on throughput heavy resource opened directly to etcd can cause:
* Huge memory leak on both kube-apiserver and etcd. This is caused by direct etcd watches having an infinite buffer and never kicking clients. 
* Overload of the apiserver-etcd watch stream, causing starvation of other watches, including a watch opened by watch cache. Starved watches experience a `prevKV=nil` error, causing them to break. If the watch opened by the watch cache is broken enough times it can cause termination of all watches and any new watch will be immediately closed. Short watches can lead to the whole APF being exhausted, as watches take the same amount of seats no matter how long they persist.

This goes against user expectation, where a large correctly running cluster with large pod objects can be taken down by single misguided client.

### What did you expect to happen?

* Inactive/slow watchers are kicked to prevent infinite memory increase. This already happens for watches opened to watch cache.
* It should not be possible for direct watches to etcd to cause termination of watch cache watches. Connection for watch cache should be separated to avoid cross pollution
* When watch cache is enabled, users should not be able to create direct watches to etcd. We need https://github.com/kubernetes/enhancements/blob/master/keps/sig-api-machinery/2340-Consistent-reads-from-cache/README.md for Watch.
* Window of available events should be limited by compaction window not by PrevKV availability
* Kube-apiserver should provide a metrics allowing to trace state of watch requests, also separate watches to etcd and watch cache.

Some of the issues listed here require improvements on etcd side. I will create separate issue in etcd repo after repeating the investigation for etcd alone.

### How can we reproduce it (as minimally and precisely as possible)?

My testing showed limit of 400MB/s of watch throughput. 

This was achieved by:
* 100 watches with resourceVersion="" to a single resource, each consuming 1 event per second
* 100 write QPS of 50KB objects to a single resource

To reproduce prevKV=nil, before my workstation run out of memory I also reduced the compaction window to 5s. 

Code used
```
package main

import (
	"context"
	"fmt"
	"math/rand"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

var objectSize = 50000
var objectCount = 10
var timeBetweenRequests = time.Millisecond * 100
var maxQPS float32 = 100

var initResourceVersion = ""
var watchers = 100
var watcherSleep = time.Second * 1

func main() {
	config, err := clientcmd.BuildConfigFromFlags("", filepath.Join(homedir.HomeDir(), ".kube", "config"))
	if err != nil {
		panic(err.Error())
	}
	config.QPS = maxQPS
	config.Burst = 100

	// create the clientset
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}

	var wg sync.WaitGroup
	var events atomic.Int64
	startWriters(clientset, &wg)
	startWatchers(clientset, &wg, &events)
	start := time.Now()
	events.Store(0)

	fmt.Printf("Start %s\n", start)
	for range time.Tick(time.Second) {
		fmt.Printf("%s events: %d\n", time.Since(start), events.Load())
	}
	wg.Wait()
}

func startWriters(clientset kubernetes.Interface, wg *sync.WaitGroup) {
	for i := 0; i < objectCount; i++ {
		name := fmt.Sprintf("%d", i)
		wg.Add(1)
		go func() {
			defer wg.Done()
			clientset.CoreV1().ConfigMaps("default").Delete(context.TODO(), name, metav1.DeleteOptions{})
			_, err := clientset.CoreV1().ConfigMaps("default").Create(context.TODO(), randomConfigmap(name), metav1.CreateOptions{})
			if err != nil {
				panic(err)
			}
			time.Sleep(timeBetweenRequests)
			for {
				_, err := clientset.CoreV1().ConfigMaps("default").Update(context.TODO(), randomConfigmap(name), metav1.UpdateOptions{})
				if err != nil {
					panic(err)
				}
				time.Sleep(timeBetweenRequests)
			}
		}()
	}
}
func startWatchers(clientset kubernetes.Interface, wg *sync.WaitGroup, events *atomic.Int64) {
	for i := 0; i < watchers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for {
				watch, err := clientset.CoreV1().ConfigMaps("default").Watch(context.TODO(), metav1.ListOptions{ResourceVersion: initResourceVersion})
				if err != nil {
					panic(err)
				}
				for event := range watch.ResultChan() {
					switch event.Object.(type) {
					case *v1.ConfigMap:
						events.Add(1)
					case *metav1.Status:
					default:
						fmt.Printf("Event, type: %s, obj: %+v\n", event.Type, event.Object)
					}
					time.Sleep(watcherSleep)
				}
			}
		}()
	}
}

func randomConfigmap(name string) *v1.ConfigMap {
	return &v1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name: name,
		},
		Immutable: nil,
		Data: map[string]string{
			"random": RandString(objectSize),
		},
		BinaryData: nil,
	}
}

const chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"

func RandString(l int) string {
	s := make([]byte, l)
	for i := 0; i < l; i++ {
		s[i] = chars[rand.Intn(len(chars))]
	}
	return string(s)
}
```

### Anything else we need to know?

_No response_

### Kubernetes version

1.29.2

### Cloud provider

KIND

### OS version

n/a


### Install tools

<details>

</details>


### Container runtime (CRI) and version (if applicable)

<details>

</details>


### Related plugins (CNI, CSI, ...) and versions (if applicable)

<details>

</details>