Merge pull request #20149 from nwnt/nwnt/add-k8s-traffic-type

Add Kubernetes Traffic Type

Author: serathius

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

@@ -28,7 +28,6 @@ import (
 
 	"go.etcd.io/etcd/tests/v3/antithesis/test-template/robustness/common"
 	"go.etcd.io/etcd/tests/v3/robustness/report"
-	"go.etcd.io/etcd/tests/v3/robustness/traffic"
 	"go.etcd.io/etcd/tests/v3/robustness/validate"
 )
 
@@ -61,7 +60,7 @@ func validateReports(lg *zap.Logger, serversDataPath map[string]string, reports
 	persistedRequests, err := report.PersistedRequests(lg, slices.Collect(maps.Values(serversDataPath)))
 	assertResult(validate.ResultFromError(err), "Loaded persisted requests")
 
-	validateConfig := validate.Config{ExpectRevisionUnique: traffic.EtcdPutDeleteLease.ExpectUniqueRevision()}
+	validateConfig := validate.Config{ExpectRevisionUnique: false}
 	result := validate.ValidateAndReturnVisualize(lg, validateConfig, reports, persistedRequests, 5*time.Minute)
 	assertResult(result.Assumptions, "Validation assumptions fulfilled")
 	if result.Linearization.Timeout {

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

@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 	"flag"
+	"math/rand/v2"
 	"os"
 	"slices"
 	"sync"
@@ -37,13 +38,23 @@ import (
 	"go.etcd.io/etcd/tests/v3/robustness/traffic"
 )
 
-var profile = traffic.Profile{
-	MinimalQPS:                     100,
-	MaximalQPS:                     1000,
-	BurstableQPS:                   1000,
-	ClientCount:                    3,
-	MaxNonUniqueRequestConcurrency: 3,
-}
+var (
+	profile = traffic.Profile{
+		MinimalQPS:                     100,
+		MaximalQPS:                     1000,
+		BurstableQPS:                   1000,
+		ClientCount:                    3,
+		MaxNonUniqueRequestConcurrency: 3,
+	}
+	trafficNames = []string{
+		"etcd",
+		"kubernetes",
+	}
+	traffics = []traffic.Traffic{
+		traffic.EtcdPutDeleteLease,
+		traffic.Kubernetes,
+	}
+)
 
 func main() {
 	local := flag.Bool("local", false, "run tests locally and connect to etcd instances via localhost")
@@ -62,6 +73,9 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
+	choice := rand.IntN(len(traffics))
+	tf := traffics[choice]
+	lg.Info("Traffic", zap.String("Type", trafficNames[choice]))
 	r := report.TestReport{Logger: lg, ServersDataPath: etcdetcdDataPaths}
 	defer func() {
 		if err = r.Report(reportPath); err != nil {
@@ -70,14 +84,14 @@ func main() {
 	}()
 
 	lg.Info("Start traffic generation", zap.Duration("duration", duration))
-	r.Client, err = runTraffic(ctx, lg, hosts, baseTime, duration)
+	r.Client, err = runTraffic(ctx, lg, tf, hosts, baseTime, duration)
 	if err != nil {
 		lg.Error("Failed to generate traffic")
 		panic(err)
 	}
 }
 
-func runTraffic(ctx context.Context, lg *zap.Logger, hosts []string, baseTime time.Time, duration time.Duration) ([]report.ClientReport, error) {
+func runTraffic(ctx context.Context, lg *zap.Logger, tf traffic.Traffic, hosts []string, baseTime time.Time, duration time.Duration) ([]report.ClientReport, error) {
 	ids := identity.NewIDProvider()
 	r, err := traffic.CheckEmptyDatabaseAtStart(ctx, lg, hosts, ids, baseTime)
 	if err != nil {
@@ -93,7 +107,7 @@ func runTraffic(ctx context.Context, lg *zap.Logger, hosts []string, baseTime ti
 	startTime := time.Since(baseTime)
 	g.Go(func() error {
 		defer close(maxRevisionChan)
-		trafficReports = slices.Concat(trafficReports, simulateTraffic(ctx, hosts, ids, baseTime, duration))
+		trafficReports = slices.Concat(trafficReports, simulateTraffic(ctx, tf, hosts, ids, baseTime, duration))
 		maxRevision := report.OperationsMaxRevision(trafficReports)
 		maxRevisionChan <- maxRevision
 		lg.Info("Finished simulating Traffic", zap.Int64("max-revision", maxRevision))
@@ -120,7 +134,7 @@ func runTraffic(ctx context.Context, lg *zap.Logger, hosts []string, baseTime ti
 	return reports, nil
 }
 
-func simulateTraffic(ctx context.Context, hosts []string, ids identity.Provider, baseTime time.Time, duration time.Duration) []report.ClientReport {
+func simulateTraffic(ctx context.Context, tf traffic.Traffic, hosts []string, ids identity.Provider, baseTime time.Time, duration time.Duration) []report.ClientReport {
 	var mux sync.Mutex
 	var wg sync.WaitGroup
 	storage := identity.NewLeaseIDStorage()
@@ -136,7 +150,7 @@ func simulateTraffic(ctx context.Context, hosts []string, ids identity.Provider,
 			defer wg.Done()
 			defer c.Close()
 
-			traffic.EtcdPutDeleteLease.RunTrafficLoop(ctx, c, limiter,
+			tf.RunTrafficLoop(ctx, c, limiter,
 				ids,
 				storage,
 				concurrencyLimiter,
@@ -153,7 +167,7 @@ func simulateTraffic(ctx context.Context, hosts []string, ids identity.Provider,
 	go func(c *client.RecordingClient) {
 		defer wg.Done()
 		defer c.Close()
-		traffic.EtcdPutDeleteLease.RunCompactLoop(ctx, c, traffic.DefaultCompactionPeriod, finish)
+		tf.RunCompactLoop(ctx, c, traffic.DefaultCompactionPeriod, finish)
 		mux.Lock()
 		reports = append(reports, c.Report())
 		mux.Unlock()