tests/robustness: improve Kubernetes traffic generator watch coverage

tests/robustness/scenarios/scenarios.go

@@ -58,23 +58,55 @@ var trafficProfiles = []TrafficProfile{
 		},
 	},
 	{
-		Name:    "KubernetesHighTraffic",
-		Traffic: traffic.Kubernetes,
+		// Non-recursive (single-key) watch coverage. Lives in the generic etcd
+		// profile because Kubernetes does not issue non-recursive watches to etcd.
+		Name:    "EtcdSingleKeyWatch",
+		Traffic: traffic.EtcdPutSingleKeyWatch,
 		Profile: traffic.Profile{
-			KeyValue:   &traffic.KeyValueHigh,
+			KeyValue:   &traffic.KeyValueMedium,
 			Watch:      &traffic.WatchDefault,
 			Compaction: &traffic.CompactionDefault,
 		},
 	},
+	{
+		Name:    "KubernetesHighTraffic",
+		Traffic: traffic.Kubernetes,
+		Profile: kubernetesProfile(&traffic.KeyValueHigh, nil),
+	},
 	{
 		Name:    "KubernetesLowTraffic",
 		Traffic: traffic.Kubernetes,
-		Profile: traffic.Profile{
-			KeyValue:   &traffic.KeyValueMedium,
-			Watch:      &traffic.WatchDefault,
-			Compaction: &traffic.CompactionDefault,
-		},
+		Profile: kubernetesProfile(&traffic.KeyValueMedium, nil),
+	},
+	{
+		// events are not served from the watch cache, so they exercise the
+		// direct watch path (ProgressNotify=false).
+		Name:    "KubernetesEventsTraffic",
+		Traffic: traffic.KubernetesEvents,
+		Profile: kubernetesProfile(&traffic.KeyValueMedium, nil),
 	},
+	{
+		// Shared gRPC stream models the events resource: events bypass the watch
+		// cache, so a single apiserver holds many direct watches at once, all
+		// multiplexed onto its one client's stream (WatchesPerInstance). Cached
+		// resources (pods) have a single cacher watch and are not modeled this way.
+		Name:    "KubernetesEventsSharedStream",
+		Traffic: traffic.KubernetesEvents,
+		Profile: kubernetesProfile(&traffic.KeyValueMedium, &traffic.SharedStream{InstanceCount: 3, WatchesPerInstance: 10}),
+	},
+}
+
+// kubernetesProfile builds a Kubernetes traffic profile with the standard watch and
+// compaction settings. Pass a non-nil shared to multiplex all loops onto a single
+// gRPC stream (one client per simulated apiserver instance); nil keeps the default
+// separate-client behavior.
+func kubernetesProfile(kv *traffic.KeyValue, shared *traffic.SharedStream) traffic.Profile {
+	return traffic.Profile{
+		KeyValue:     kv,
+		Watch:        &traffic.WatchDefault,
+		Compaction:   &traffic.CompactionDefault,
+		SharedStream: shared,
+	}
 }
 
 type TestScenario struct {

tests/robustness/traffic/etcd.go

@@ -69,6 +69,28 @@ var (
 			{Choice: Put, Weight: 40},
 		},
 	}
+	// EtcdPutSingleKeyWatch mirrors EtcdPut but watches a single key non-recursively,
+	// providing coverage for etcd's non-recursive watch path. This lives in the generic
+	// etcd profile because Kubernetes does not meaningfully issue non-recursive watches
+	// to etcd.
+	EtcdPutSingleKeyWatch Traffic = etcdTraffic{
+		keyCount:       10,
+		leaseTTL:       DefaultLeaseTTL,
+		singleKeyWatch: true,
+		// Please keep the sum of weights equal 100.
+		requests: []random.ChoiceWeight[etcdRequestType]{
+			{Choice: Get, Weight: 15},
+			{Choice: List, Weight: 4},
+			{Choice: ListKeyOnly, Weight: 4},
+			{Choice: ListStream, Weight: 4},
+			{Choice: ListStreamKeyOnly, Weight: 3},
+			{Choice: StaleGet, Weight: 10},
+			{Choice: StaleList, Weight: 5},
+			{Choice: StaleListStream, Weight: 5},
+			{Choice: MultiOpTxn, Weight: 10},
+			{Choice: Put, Weight: 40},
+		},
+	}
 	EtcdDelete Traffic = etcdTraffic{
 		keyCount: 10,
 		leaseTTL: DefaultLeaseTTL,
@@ -84,6 +106,9 @@ type etcdTraffic struct {
 	keyCount int
 	requests []random.ChoiceWeight[etcdRequestType]
 	leaseTTL int64
+	// singleKeyWatch makes RunWatchLoop issue a non-recursive (single-key) watch
+	// instead of a recursive prefix watch, to exercise etcd's non-recursive path.
+	singleKeyWatch bool
 }
 
 func (t etcdTraffic) ExpectUniqueRevision() bool {
@@ -163,7 +188,8 @@ func (t etcdTraffic) RunKeyValueLoop(ctx context.Context, c *client.RecordingCli
 
 func (t etcdTraffic) RunWatchLoop(ctx context.Context, c *client.RecordingClient, p RunWatchLoopParam) {
 	runWatchLoop(ctx, c, p, watchLoopConfig{
-		key: p.KeyStore.GetPrefix(),
+		key:       p.KeyStore.GetPrefix(),
+		recursive: !t.singleKeyWatch,
 	})
 }
 

tests/robustness/traffic/kubernetes.go

@@ -74,6 +74,29 @@ var (
 			{Choice: KubernetesCreate, Weight: 60},
 		},
 	}
+
+	// KubernetesEvents models the events resource, which is NOT served from the
+	// apiserver watch cache. Request handlers watch etcd directly, so watches use
+	// ProgressNotify=false (the direct/uncached path).
+	KubernetesEvents Traffic = kubernetesTraffic{
+		averageKeyCount: 10,
+		resource:        "events",
+		namespace:       "default",
+		// Please keep the sum of weights equal 100.
+		readChoices: []random.ChoiceWeight[KubernetesRequestType]{
+			{Choice: KubernetesGet, Weight: 5},
+			{Choice: KubernetesGetStale, Weight: 2},
+			{Choice: KubernetesGetRev, Weight: 8},
+			{Choice: KubernetesListStale, Weight: 5},
+			{Choice: KubernetesListAndWatch, Weight: 80},
+		},
+		// Please keep the sum of weights equal 100.
+		writeChoices: []random.ChoiceWeight[KubernetesRequestType]{
+			{Choice: KubernetesUpdate, Weight: 90},
+			{Choice: KubernetesDelete, Weight: 5},
+			{Choice: KubernetesCreate, Weight: 5},
+		},
+	}
 )
 
 type kubernetesTraffic struct {
@@ -84,6 +107,16 @@ type kubernetesTraffic struct {
 	writeChoices    []random.ChoiceWeight[KubernetesRequestType]
 }
 
+// progressNotify reports whether this resource's List+Watch requests ProgressNotify.
+// It is derived from the resource: cached resources are watched via the apiserver
+// watch cache, whose internal etcd watch sets ProgressNotify=true, while uncached
+// resources (events) are watched directly by request handlers, which cannot set it.
+// https://github.com/kubernetes/kubernetes/blob/2016fab3085562b4132e6d3774b6ded5ba9939fd/staging/src/k8s.io/apiserver/pkg/storage/etcd3/watcher.go
+func (t kubernetesTraffic) progressNotify() bool {
+	// The events resource disables the watch cache in the apiserver.
+	return t.resource != "events"
+}
+
 func (t kubernetesTraffic) ExpectUniqueRevision() bool {
 	return true
 }
@@ -140,6 +173,7 @@ func (t kubernetesTraffic) RunWatchLoop(ctx context.Context, c *client.Recording
 	runWatchLoop(ctx, c, p, watchLoopConfig{
 		key:           "/registry/" + t.resource + "/",
 		requireLeader: true,
+		recursive:     true,
 	})
 }
 
@@ -267,8 +301,11 @@ func (t kubernetesTraffic) Watch(ctx context.Context, c *client.RecordingClient,
 	// Kubernetes issues Watch requests by requiring a leader to exist
 	// in the cluster:
 	// https://github.com/kubernetes/kubernetes/blob/2016fab3085562b4132e6d3774b6ded5ba9939fd/staging/src/k8s.io/apiserver/pkg/storage/etcd3/store.go#L872
+	// ProgressNotify depends on whether the resource is served from the watch
+	// cache (see kubernetesTraffic.progressNotify). PrevKV is always set, matching
+	// the apiserver, which passes clientv3.WithPrevKV() unconditionally.
 	watchCtx = clientv3.WithRequireLeader(watchCtx)
-	for e := range c.Watch(watchCtx, keyPrefix, revision, true, true, true) {
+	for e := range c.Watch(watchCtx, keyPrefix, revision, true, t.progressNotify(), true) {
 		s.Update(e)
 	}
 	limiter.Wait(ctx)

tests/robustness/traffic/traffic.go

@@ -109,36 +109,48 @@ func SimulateTraffic(ctx context.Context, t *testing.T, lg *zap.Logger, clus *e2
 
 	lg.Info("Start traffic")
 	startTime := time.Since(clientSet.BaseTime())
-	err = SimulateKeyValueTraffic(ctx, &wg, profile.KeyValue, endpoints, clientSet, traffic, RunTrafficLoopParam{
+	keyValueParam := RunTrafficLoopParam{
 		QPSLimiter:                         limiter,
 		IDs:                                clientSet.IdentityProvider(),
 		LeaseIDStorage:                     lm,
 		NonUniqueRequestConcurrencyLimiter: nonUniqueWriteLimiter,
 		KeyStore:                           keyStore,
 		Storage:                            kubernetesStorage,
 		Finish:                             finish,
-	})
-	require.NoError(t, err)
+	}
+	watchParam := RunWatchLoopParam{
+		QPSLimiter: limiter,
+		KeyStore:   keyStore,
+		Storage:    kubernetesStorage,
+		Finish:     finish,
+		Logger:     lg,
+	}
 	if profile.Watch != nil {
-		err = SimulateWatchTraffic(ctx, &wg, profile.Watch, endpoints, clientSet, traffic, RunWatchLoopParam{
-			Config:     *profile.Watch,
-			QPSLimiter: limiter,
-			KeyStore:   keyStore,
-			Storage:    kubernetesStorage,
-			Finish:     finish,
-			Logger:     lg,
-		})
-		require.NoError(t, err)
+		watchParam.Config = *profile.Watch
 	}
+	compactParam := RunCompactLoopParam{Finish: finish}
 	if profile.Compaction != nil {
 		if profile.Compaction.Period < MinimalCompactionPeriod {
 			t.Fatalf("Compaction period %v below minimal %v", profile.Compaction.Period, MinimalCompactionPeriod)
 		}
-		err = SimulateCompactionTraffic(ctx, &wg, profile.Compaction, endpoints, clientSet, traffic, RunCompactLoopParam{
-			Period: profile.Compaction.Period,
-			Finish: finish,
-		})
+		compactParam.Period = profile.Compaction.Period
+	}
+
+	if profile.SharedStream != nil {
+		require.NotNilf(t, profile.Watch, "SharedStream traffic requires a Watch profile")
+		err = SimulateSharedStreamTraffic(ctx, &wg, profile, endpoints, clientSet, traffic, keyValueParam, watchParam, compactParam)
 		require.NoError(t, err)
+	} else {
+		err = SimulateKeyValueTraffic(ctx, &wg, profile.KeyValue, endpoints, clientSet, traffic, keyValueParam)
+		require.NoError(t, err)
+		if profile.Watch != nil {
+			err = SimulateWatchTraffic(ctx, &wg, profile.Watch, endpoints, clientSet, traffic, watchParam)
+			require.NoError(t, err)
+		}
+		if profile.Compaction != nil {
+			err = SimulateCompactionTraffic(ctx, &wg, profile.Compaction, endpoints, clientSet, traffic, compactParam)
+			require.NoError(t, err)
+		}
 	}
 	var fr *report.FailpointInjection
 	select {
@@ -254,6 +266,56 @@ func SimulateCompactionTraffic(ctx context.Context, wg *sync.WaitGroup, profile
 	return nil
 }
 
+// SimulateSharedStreamTraffic models a set of Kubernetes apiserver instances. Each instance
+// is one cluster-scoped client running all of its loops (key/value, watch, compaction), so
+// its watches multiplex onto a single gRPC stream, as a real apiserver does. Each instance
+// opens WatchesPerInstance concurrent watch substreams, modeling the events fan-in where a
+// single apiserver holds many direct watches on its one stream.
+func SimulateSharedStreamTraffic(ctx context.Context, wg *sync.WaitGroup, profile Profile, endpoints []string, clientSet *client.ClientSet, tf Traffic, keyValueParam RunTrafficLoopParam, watchParam RunWatchLoopParam, compactParam RunCompactLoopParam) error {
+	apiservers := profile.SharedStream
+	watchesPerInstance := max(apiservers.WatchesPerInstance, 1)
+	for range apiservers.InstanceCount {
+		// One cluster-scoped client per apiserver instance (all endpoints, like a real
+		// apiserver configured via --etcd-servers).
+		c, err := clientSet.NewClient(endpoints)
+		if err != nil {
+			return err
+		}
+		wg.Add(1)
+		go func(c *client.RecordingClient) {
+			defer wg.Done()
+			defer c.Close()
+			// All of this instance's loops run on the one client, so its watches
+			// multiplex onto a single gRPC stream. loops tracks them so the client is
+			// closed only after they all finish.
+			var loops sync.WaitGroup
+			loops.Add(1)
+			go func() {
+				defer loops.Done()
+				tf.RunKeyValueLoop(ctx, c, keyValueParam)
+			}()
+			if profile.Watch != nil {
+				for range watchesPerInstance {
+					loops.Add(1)
+					go func() {
+						defer loops.Done()
+						tf.RunWatchLoop(ctx, c, watchParam)
+					}()
+				}
+			}
+			if profile.Compaction != nil {
+				loops.Add(1)
+				go func() {
+					defer loops.Done()
+					tf.RunCompactLoop(ctx, c, compactParam)
+				}()
+			}
+			loops.Wait()
+		}(c)
+	}
+	return nil
+}
+
 func CalculateWatchStats(reports []report.ClientReport, start, end time.Duration) (ws watchStats) {
 	ws.Period = end - start
 	if ws.Period <= 0 {
@@ -362,9 +424,25 @@ func (ts *trafficStats) QPS() float64 {
 }
 
 type Profile struct {
-	KeyValue   *KeyValue
-	Watch      *Watch
-	Compaction *Compaction
+	KeyValue     *KeyValue
+	Watch        *Watch
+	Compaction   *Compaction
+	SharedStream *SharedStream
+}
+
+// SharedStream runs all loops (key/value, watch, compaction) on a single client per
+// simulated apiserver instance, so their watches multiplex onto one gRPC stream, as a
+// real apiserver does. It models the events resource in particular: events are not served
+// from the watch cache, so a single apiserver holds many direct watches at once, all on
+// its one stream. nil keeps the default behavior of a separate client per loop.
+type SharedStream struct {
+	// InstanceCount is the number of simulated apiserver instances. Each instance is a
+	// single cluster-scoped client (configured with all endpoints, like a real apiserver
+	// is via --etcd-servers) running all of its loops.
+	InstanceCount int
+	// WatchesPerInstance is the number of concurrent watch substreams each instance opens,
+	// all multiplexed onto its single gRPC stream. Values < 1 are treated as 1.
+	WatchesPerInstance int
 }
 
 type KeyValue struct {
@@ -438,10 +516,17 @@ func runWatchLoop(ctx context.Context, c *client.RecordingClient, p RunWatchLoop
 }
 
 func runWatch(ctx context.Context, c *client.RecordingClient, p RunWatchLoopParam, cfg watchLoopConfig) error {
+	key := cfg.key
+	if !cfg.recursive {
+		// Non-recursive watch: target a single concrete key from the pool so the
+		// watch exercises etcd's single-key path rather than a prefix watch.
+		key = p.KeyStore.GetKey()
+	}
+
 	getCtx, getCancel := context.WithTimeout(ctx, RequestTimeout)
 	defer getCancel()
 
-	resp, err := c.Get(getCtx, cfg.key)
+	resp, err := c.Get(getCtx, key)
 	if err != nil {
 		return err
 	}
@@ -453,7 +538,7 @@ func runWatch(ctx context.Context, c *client.RecordingClient, p RunWatchLoopPara
 	if cfg.requireLeader {
 		watchCtx = clientv3.WithRequireLeader(watchCtx)
 	}
-	w := c.Watch(watchCtx, cfg.key, rev, true, true, true)
+	w := c.Watch(watchCtx, key, rev, cfg.recursive, true, true)
 	for {
 		select {
 		case <-ctx.Done():
@@ -471,6 +556,10 @@ func runWatch(ctx context.Context, c *client.RecordingClient, p RunWatchLoopPara
 type watchLoopConfig struct {
 	key           string
 	requireLeader bool
+	// recursive controls whether the watch is a prefix (recursive) watch on key.
+	// When false, runWatch issues a single-key (non-recursive) watch on a concrete
+	// key from the key pool, exercising etcd's non-recursive watch path.
+	recursive bool
 }
 
 func CheckEmptyDatabaseAtStart(ctx context.Context, lg *zap.Logger, endpoints []string, cs *client.ClientSet) error {