Merge pull request #20683 from henrybear327/robustness-test-introduce-cont-watch-creation-config

Implement a feature to periodically open watches in robustness test and make it configurable

Author: serathius

tests/antithesis/test-template/robustness/traffic/main.go

@@ -33,6 +33,7 @@ import (
 	"go.etcd.io/etcd/tests/v3/antithesis/test-template/robustness/common"
 	"go.etcd.io/etcd/tests/v3/robustness/client"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
+	"go.etcd.io/etcd/tests/v3/robustness/options"
 	robustnessrand "go.etcd.io/etcd/tests/v3/robustness/random"
 	"go.etcd.io/etcd/tests/v3/robustness/report"
 	"go.etcd.io/etcd/tests/v3/robustness/traffic"
@@ -46,6 +47,10 @@ var (
 		MemberClientCount:              3,
 		ClusterClientCount:             1,
 		MaxNonUniqueRequestConcurrency: 3,
+		BackgroundWatchConfig: options.BackgroundWatchConfig{
+			Interval:       0,
+			RevisionOffset: 0,
+		},
 	}
 	trafficNames = []string{
 		"etcd",
@@ -121,11 +126,12 @@ func runTraffic(ctx context.Context, lg *zap.Logger, tf traffic.Traffic, hosts [
 	defer watchSet.Close()
 	g.Go(func() error {
 		err := client.CollectClusterWatchEvents(ctx, client.CollectClusterWatchEventsParam{
-			Lg:              lg,
-			Endpoints:       hosts,
-			MaxRevisionChan: maxRevisionChan,
-			Cfg:             watchConfig,
-			ClientSet:       watchSet,
+			Lg:                    lg,
+			Endpoints:             hosts,
+			MaxRevisionChan:       maxRevisionChan,
+			Cfg:                   watchConfig,
+			ClientSet:             watchSet,
+			BackgroundWatchConfig: profile.BackgroundWatchConfig,
 		})
 		return err
 	})

tests/robustness/client/watch.go

@@ -18,10 +18,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"time"
 
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
+	"go.etcd.io/etcd/tests/v3/robustness/options"
 	"go.etcd.io/etcd/tests/v3/robustness/report"
 )
 
@@ -31,6 +33,7 @@ type CollectClusterWatchEventsParam struct {
 	MaxRevisionChan <-chan int64
 	Cfg             WatchConfig
 	ClientSet       *ClientSet
+	options.BackgroundWatchConfig
 }
 
 func CollectClusterWatchEvents(ctx context.Context, param CollectClusterWatchEventsParam) error {
@@ -51,14 +54,29 @@ func CollectClusterWatchEvents(ctx context.Context, param CollectClusterWatchEve
 			return err
 		})
 	}
-
+	finish := make(chan struct{})
 	g.Go(func() error {
 		maxRevision := <-param.MaxRevisionChan
 		for _, memberChan := range memberMaxRevisionChans {
 			memberChan <- maxRevision
 		}
+		close(finish)
 		return nil
 	})
+
+	if param.BackgroundWatchConfig.Interval > 0 {
+		for _, endpoint := range param.Endpoints {
+			g.Go(func() error {
+				c, err := param.ClientSet.NewClient([]string{endpoint})
+				if err != nil {
+					return err
+				}
+				defer c.Close()
+				return openWatchPeriodically(ctx, &g, c, param.BackgroundWatchConfig, finish)
+			})
+		}
+	}
+
 	return g.Wait()
 }
 
@@ -130,3 +148,38 @@ resetWatch:
 		}
 	}
 }
+
+func openWatchPeriodically(ctx context.Context, g *errgroup.Group, c *RecordingClient, backgroundWatchConfig options.BackgroundWatchConfig, finish <-chan struct{}) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-finish:
+			return nil
+		case <-time.After(backgroundWatchConfig.Interval):
+		}
+		g.Go(func() error {
+			resp, err := c.Get(ctx, "/key")
+			if err != nil {
+				return err
+			}
+			rev := resp.Header.Revision + backgroundWatchConfig.RevisionOffset
+
+			watchCtx, cancel := context.WithCancel(ctx)
+			defer cancel()
+			w := c.Watch(watchCtx, "", rev, true, true, true)
+			for {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-finish:
+					return nil
+				case _, ok := <-w:
+					if !ok {
+						return nil
+					}
+				}
+			}
+		})
+	}
+}

tests/robustness/main_test.go

@@ -168,11 +168,12 @@ func runScenario(ctx context.Context, t *testing.T, s scenarios.TestScenario, lg
 	g.Go(func() error {
 		endpoints := processEndpoints(clus)
 		err := client.CollectClusterWatchEvents(ctx, client.CollectClusterWatchEventsParam{
-			Lg:              lg,
-			Endpoints:       endpoints,
-			MaxRevisionChan: maxRevisionChan,
-			Cfg:             s.Watch,
-			ClientSet:       watchSet,
+			Lg:                    lg,
+			Endpoints:             endpoints,
+			MaxRevisionChan:       maxRevisionChan,
+			Cfg:                   s.Watch,
+			ClientSet:             watchSet,
+			BackgroundWatchConfig: s.Profile.BackgroundWatchConfig,
 		})
 		return err
 	})

tests/robustness/options/options.go

@@ -0,0 +1,22 @@
+// Copyright 2025 The etcd Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package options
+
+import "time"
+
+type BackgroundWatchConfig struct {
+	Interval       time.Duration
+	RevisionOffset int64
+}

tests/robustness/traffic/traffic.go

@@ -28,6 +28,7 @@ import (
 	"go.etcd.io/etcd/tests/v3/robustness/client"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
 	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.etcd.io/etcd/tests/v3/robustness/options"
 	"go.etcd.io/etcd/tests/v3/robustness/report"
 	"go.etcd.io/etcd/tests/v3/robustness/validate"
 )
@@ -295,6 +296,7 @@ type Profile struct {
 	ClusterClientCount             int
 	ForbidCompaction               bool
 	CompactPeriod                  time.Duration
+	options.BackgroundWatchConfig
 }
 
 func (p Profile) WithoutCompaction() Profile {
@@ -307,6 +309,16 @@ func (p Profile) WithCompactionPeriod(cp time.Duration) Profile {
 	return p
 }
 
+func (p Profile) WithBackgroundWatchConfigInterval(interval time.Duration) Profile {
+	p.BackgroundWatchConfig.Interval = interval
+	return p
+}
+
+func (p Profile) WithBackgroundWatchConfigRevisionOffset(offset int64) Profile {
+	p.BackgroundWatchConfig.RevisionOffset = offset
+	return p
+}
+
 type RunTrafficLoopParam struct {
 	Client                             *client.RecordingClient
 	QPSLimiter                         *rate.Limiter