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"
@@ -108,6 +110,14 @@ func SimulateTraffic(ctx context.Context, t *testing.T, lg *zap.Logger, clus *e2
 			defer wg.Done()
 			defer c.Close()
 
+			if profile.EndpointSwitchPeriod > 0 {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					runEndpointSwitchLoop(ctx, c, endpoints, profile.EndpointSwitchPeriod, finish)
+				}()
+			}
+
 			traffic.RunKeyValueLoop(ctx, RunTrafficLoopParam{
 				Client:                             c,
 				QPSLimiter:                         limiter,
@@ -145,6 +155,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.EndpointSwitchPeriod > 0 {
+					wg.Add(1)
+					go func() {
+						defer wg.Done()
+						runEndpointSwitchLoop(ctx, c, endpoints, profile.EndpointSwitchPeriod, finish)
+					}()
+				}
+
 				traffic.RunWatchLoop(ctx, RunWatchLoopParam{
 					Client:     c,
 					QPSLimiter: limiter,
@@ -337,6 +356,7 @@ type Profile struct {
 	ForbidCompaction               bool
 	CompactPeriod                  time.Duration
 	ForbidRunWatchLoop             bool
+	EndpointSwitchPeriod           time.Duration
 }
 
 func (p Profile) WithoutCompaction() Profile {
@@ -354,6 +374,11 @@ func (p Profile) WithoutWatchLoop() Profile {
 	return p
 }
 
+func (p Profile) WithEndpointSwitchPeriod(esp time.Duration) Profile {
+	p.EndpointSwitchPeriod = esp
+	return p
+}
+
 type RunTrafficLoopParam struct {
 	Client                             *client.RecordingClient
 	QPSLimiter                         *rate.Limiter
@@ -464,3 +489,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...)
+		}
+	}
+}