Introduce ContinuousWatchCreation configuration for the robustness test

Author: henrybear327

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

@@ -121,11 +121,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,
+			ContinuousWatchCreation: 0,
 		})
 		return err
 	})

tests/framework/e2e/cluster.go

@@ -140,12 +140,13 @@ type EtcdProcessClusterConfig struct {
 
 	// Test config
 
-	KeepDataDir         bool
-	Logger              *zap.Logger
-	GoFailEnabled       bool
-	GoFailClientTimeout time.Duration
-	LazyFSEnabled       bool
-	PeerProxy           bool
+	KeepDataDir             bool
+	Logger                  *zap.Logger
+	GoFailEnabled           bool
+	GoFailClientTimeout     time.Duration
+	LazyFSEnabled           bool
+	PeerProxy               bool
+	ContinuousWatchCreation time.Duration
 
 	// Process config
 
@@ -370,6 +371,12 @@ func WithPeerProxy(enabled bool) EPClusterOption {
 	return func(c *EtcdProcessClusterConfig) { c.PeerProxy = enabled }
 }
 
+func WithContinuousWatchCreation(interval time.Duration) EPClusterOption {
+	return func(c *EtcdProcessClusterConfig) {
+		c.ContinuousWatchCreation = interval
+	}
+}
+
 func WithClientHTTPSeparate(enabled bool) EPClusterOption {
 	return func(c *EtcdProcessClusterConfig) { c.ClientHTTPSeparate = enabled }
 }
@@ -652,24 +659,25 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfig(tb testing.TB, i in
 	}
 
 	return &EtcdServerProcessConfig{
-		lg:                  cfg.Logger,
-		ExecPath:            execPath,
-		Args:                args,
-		EnvVars:             envVars,
-		TLSArgs:             cfg.TLSArgs(),
-		Client:              cfg.Client,
-		DataDirPath:         dataDirPath,
-		KeepDataDir:         cfg.KeepDataDir,
-		Name:                name,
-		PeerURL:             peerAdvertiseURL,
-		ClientURL:           curl,
-		ClientHTTPURL:       clientHTTPURL,
-		MetricsURL:          murl,
-		InitialToken:        cfg.ServerConfig.InitialClusterToken,
-		GoFailPort:          gofailPort,
-		GoFailClientTimeout: cfg.GoFailClientTimeout,
-		Proxy:               proxyCfg,
-		LazyFSEnabled:       cfg.LazyFSEnabled,
+		lg:                      cfg.Logger,
+		ExecPath:                execPath,
+		Args:                    args,
+		EnvVars:                 envVars,
+		TLSArgs:                 cfg.TLSArgs(),
+		Client:                  cfg.Client,
+		DataDirPath:             dataDirPath,
+		KeepDataDir:             cfg.KeepDataDir,
+		Name:                    name,
+		PeerURL:                 peerAdvertiseURL,
+		ClientURL:               curl,
+		ClientHTTPURL:           clientHTTPURL,
+		MetricsURL:              murl,
+		InitialToken:            cfg.ServerConfig.InitialClusterToken,
+		GoFailPort:              gofailPort,
+		GoFailClientTimeout:     cfg.GoFailClientTimeout,
+		Proxy:                   proxyCfg,
+		LazyFSEnabled:           cfg.LazyFSEnabled,
+		ContinuousWatchCreation: cfg.ContinuousWatchCreation,
 	}
 }
 

tests/framework/e2e/etcd_process.go

@@ -101,8 +101,9 @@ type EtcdServerProcessConfig struct {
 	GoFailPort          int
 	GoFailClientTimeout time.Duration
 
-	LazyFSEnabled bool
-	Proxy         *proxy.ServerConfig
+	LazyFSEnabled           bool
+	Proxy                   *proxy.ServerConfig
+	ContinuousWatchCreation time.Duration
 }
 
 func NewEtcdServerProcess(tb testing.TB, cfg *EtcdServerProcessConfig) (*EtcdServerProcess, error) {

tests/robustness/client/watch.go

@@ -27,11 +27,12 @@ import (
 )
 
 type CollectClusterWatchEventsParam struct {
-	Lg              *zap.Logger
-	Endpoints       []string
-	MaxRevisionChan <-chan int64
-	Cfg             WatchConfig
-	ClientSet       *ClientSet
+	Lg                      *zap.Logger
+	Endpoints               []string
+	MaxRevisionChan         <-chan int64
+	Cfg                     WatchConfig
+	ClientSet               *ClientSet
+	ContinuousWatchCreation time.Duration
 }
 
 func CollectClusterWatchEvents(ctx context.Context, param CollectClusterWatchEventsParam) error {
@@ -62,16 +63,18 @@ func CollectClusterWatchEvents(ctx context.Context, param CollectClusterWatchEve
 		return nil
 	})
 
-	for _, endpoint := range endpoints {
-		g.Go(func() error {
-			c, err := clientSet.NewClient([]string{endpoint})
-			if err != nil {
-				return err
-			}
-			defer c.Close()
-			period := 10 * time.Millisecond
-			return openWatchPeriodically(ctx, &g, c, period, finish)
-		})
+	if param.ContinuousWatchCreation > 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()
+				period := 10 * time.Millisecond
+				return openWatchPeriodically(ctx, &g, c, period, finish)
+			})
+		}
 	}
 
 	return g.Wait()
@@ -146,6 +149,7 @@ resetWatch:
 	}
 }
 
+// TODO: report created watches
 func openWatchPeriodically(ctx context.Context, g *errgroup.Group, c *RecordingClient, period time.Duration, finish <-chan struct{}) error {
 	for {
 		select {

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,
+			ContinuousWatchCreation: clus.Cfg.ContinuousWatchCreation,
 		})
 		return err
 	})

tests/robustness/scenarios/scenarios.go

@@ -308,6 +308,7 @@ func Regression(t *testing.T) []TestScenario {
 			e2e.WithGoFailEnabled(true),
 			e2e.WithPeerProxy(true),
 			e2e.WithIsPeerTLS(true),
+			e2e.WithContinuousWatchCreation(1*time.Millisecond),
 		),
 	})