robustness: implement cluster client endpoint switching

Signed-off-by: Chun-Hung Tseng <henrytseng@google.com>

Author: henrybear327

tests/robustness/client/client.go

@@ -279,6 +279,10 @@ func (c *RecordingClient) Endpoints() []string {
 	return c.client.Endpoints()
 }
 
+func (c *RecordingClient) SetEndpoints(endpoints ...string) {
+	c.client.SetEndpoints(endpoints...)
+}
+
 func (c *RecordingClient) Watch(ctx context.Context, key string, rev int64, withPrefix bool, withProgressNotify bool, withPrevKV bool) clientv3.WatchChan {
 	request := model.WatchRequest{
 		Key:                key,

tests/robustness/traffic/traffic.go

@@ -16,6 +16,8 @@ package traffic
 
 import (
 	"context"
+	"math/rand"
+	"slices"
 	"sync"
 	"testing"
 	"time"
@@ -148,6 +150,14 @@ func SimulateTraffic(ctx context.Context, t *testing.T, lg *zap.Logger, clus *e2
 			defer wg.Done()
 			defer c.Close()
 
+			if profile.Watch != nil && profile.Watch.EndpointSwitchPeriod != nil {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					runEndpointSwitchLoop(ctx, c, endpoints, *profile.Watch.EndpointSwitchPeriod, finish)
+				}()
+			}
+
 			traffic.RunKeyValueLoop(ctx, RunTrafficLoopParam{
 				Client:                             c,
 				QPSLimiter:                         limiter,
@@ -186,6 +196,15 @@ func SimulateTraffic(ctx context.Context, t *testing.T, lg *zap.Logger, clus *e2
 			go func(c *client.RecordingClient) {
 				defer wg.Done()
 				defer c.Close()
+
+				if profile.Watch != nil && profile.Watch.EndpointSwitchPeriod != nil {
+					wg.Add(1)
+					go func() {
+						defer wg.Done()
+						runEndpointSwitchLoop(ctx, c, endpoints, *profile.Watch.EndpointSwitchPeriod, finish)
+					}()
+				}
+
 				traffic.RunWatchLoop(ctx, RunWatchLoopParam{
 					Config:     profile.Watch,
 					Client:     c,
@@ -384,9 +403,10 @@ type KeyValue struct {
 }
 
 type Watch struct {
-	MemberClientCount   int
-	ClusterClientCount  int
-	RivisionOffsetRange *Range
+	MemberClientCount    int
+	ClusterClientCount   int
+	RivisionOffsetRange  *Range
+	EndpointSwitchPeriod *time.Duration
 }
 
 type Compaction struct {
@@ -506,3 +526,20 @@ func CheckEmptyDatabaseAtStart(ctx context.Context, lg *zap.Logger, endpoints []
 	}
 	return nil
 }
+
+func runEndpointSwitchLoop(ctx context.Context, c *client.RecordingClient, endpoints []string, period time.Duration, finish <-chan struct{}) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-finish:
+			return
+		case <-time.After(period + time.Duration(rand.Int63n(int64(period)))):
+			shuffledEndpoints := slices.Clone(endpoints)
+			rand.Shuffle(len(shuffledEndpoints), func(i, j int) {
+				shuffledEndpoints[i], shuffledEndpoints[j] = shuffledEndpoints[j], shuffledEndpoints[i]
+			})
+			c.SetEndpoints(shuffledEndpoints...)
+		}
+	}
+}