From 009a81eb75944c78ef031fea842582e46f747bf7 Mon Sep 17 00:00:00 2001 From: Jeremy Leach <44558776+jbml@users.noreply.github.com> Date: Sun, 21 Jun 2026 12:53:42 +0000 Subject: [PATCH] tests/robustness: improve Kubernetes traffic generator watch coverage Address #20622 by making the robustness Kubernetes traffic generator's watches more faithful to apiserver behavior: - Derive watch ProgressNotify from the resource: cached resources (pods) are served by the watch cache (internal etcd watch sets ProgressNotify=true), while the uncached events resource is watched directly (ProgressNotify=false). Add a KubernetesEvents traffic profile to model the direct path. PrevKV stays set, matching the apiserver, which passes WithPrevKV() unconditionally. - Add non-recursive (single-key) watch coverage to the generic etcd traffic profile (EtcdPutSingleKeyWatch / EtcdSingleKeyWatch). Kubernetes does not meaningfully issue non-recursive watches to etcd, so this belongs in the generic profile rather than the Kubernetes one. - Add SharedStream mode modeling the events watch fan-in: one cluster-scoped client per simulated apiserver instance opens WatchesPerInstance concurrent watch substreams, all multiplexed onto a single gRPC stream. Used by the KubernetesEventsSharedStream scenario. Signed-off-by: Jeremy Leach <44558776+jbml@users.noreply.github.com> --- tests/robustness/scenarios/scenarios.go | 48 +++++++-- tests/robustness/traffic/etcd.go | 28 ++++- tests/robustness/traffic/kubernetes.go | 39 ++++++- tests/robustness/traffic/traffic.go | 131 ++++++++++++++++++++---- 4 files changed, 215 insertions(+), 31 deletions(-) diff --git a/tests/robustness/scenarios/scenarios.go b/tests/robustness/scenarios/scenarios.go index 64f4ab3e2497..05363968ccb9 100644 --- a/tests/robustness/scenarios/scenarios.go +++ b/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 { diff --git a/tests/robustness/traffic/etcd.go b/tests/robustness/traffic/etcd.go index a34c072f2ac8..ae4a2e221ae9 100644 --- a/tests/robustness/traffic/etcd.go +++ b/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, }) } diff --git a/tests/robustness/traffic/kubernetes.go b/tests/robustness/traffic/kubernetes.go index e45171c03fa1..8dcfc55e889f 100644 --- a/tests/robustness/traffic/kubernetes.go +++ b/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) diff --git a/tests/robustness/traffic/traffic.go b/tests/robustness/traffic/traffic.go index 04159c6ecc99..2efeb259e9dc 100644 --- a/tests/robustness/traffic/traffic.go +++ b/tests/robustness/traffic/traffic.go @@ -109,7 +109,7 @@ 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, @@ -117,28 +117,40 @@ func SimulateTraffic(ctx context.Context, t *testing.T, lg *zap.Logger, clus *e2 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 {