Repository: etcd-io/etcd
File: tests/antithesis/test-template/robustness/traffic/main.go (Go)

Code before:
				concurrencyLimiter,
				keyStore,
				finish,
			)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c)
	}
	wg.Add(1)
	compactClient := connect(hosts, ids, baseTime)
	go func(c *client.RecordingClient) {
		defer wg.Done()
		defer c.Close()
		traffic.EtcdPutDeleteLease.RunCompactLoop(ctx, c, traffic.DefaultCompactionPeriod, finish)
		mux.Lock()
		reports = append(reports, c.Report())
		mux.Unlock()
	}(compactClient)
	for _, h := range hosts {
		c := connect([]string{h}, ids, baseTime)
		wg.Add(1)
		go func(c *client.RecordingClient, lg *zap.Logger) {
			defer wg.Done()
			defer c.Close()
			runDefragLoop(ctx, lg, c, traffic.DefaultCompactionPeriod, finish)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c, lg)
	}
	wg.Wait()
	return reports
}

func runDefragLoop(ctx context.Context, lg *zap.Logger, c *client.RecordingClient, period time.Duration, finish <-chan struct{}) {
	for {
		jittered := time.Duration(robustnessrand.RandRange(int64(period-period/2), int64(period+period/2)))
		lg.Info("next defragment run", zap.String("host", c.Endpoints()[0]), zap.Time("at", time.Now().Add(jittered)))
		ticker := time.NewTicker(jittered)
		defer ticker.Stop()

		select {
		case <-ctx.Done():
			return
		case <-finish:
			return
		case <-ticker.C:
		}
		dctx, cancel := context.WithTimeout(ctx, traffic.RequestTimeout)
		_, err := c.Defragment(dctx)

Review comment (bug):
```suggestion
	defragPeriod := traffic.DefaultCompactionPeriod * len(hosts)
	for _, h := range hosts {
		c := connect([]string{h}, ids, baseTime)
		wg.Add(1)
		go func(c *client.RecordingClient, lg *zap.Logger) {
			defer wg.Done()
			defer c.Close()
			runDefragLoop(ctx, lg, c, defragPeriod, finish)
```

With client per host we should avoid making it too frequent.

Code after:
				concurrencyLimiter,
				keyStore,
				finish,
			)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c)
	}
	wg.Add(1)
	compactClient := connect(hosts, ids, baseTime)
	go func(c *client.RecordingClient) {
		defer wg.Done()
		defer c.Close()
		traffic.EtcdPutDeleteLease.RunCompactLoop(ctx, c, traffic.DefaultCompactionPeriod, finish)
		mux.Lock()
		reports = append(reports, c.Report())
		mux.Unlock()
	}(compactClient)
	defragPeriod := traffic.DefaultCompactionPeriod * time.Duration(len(hosts))
	for _, h := range hosts {
		c := connect([]string{h}, ids, baseTime)
		wg.Add(1)
		go func(c *client.RecordingClient) {
			defer wg.Done()
			defer c.Close()
			runDefragLoop(ctx, c, defragPeriod, finish)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c)
	}
	wg.Wait()
	return reports
}

func runDefragLoop(ctx context.Context, c *client.RecordingClient, period time.Duration, finish <-chan struct{}) {
	jittered := time.Duration(robustnessrand.RandRange(int64(period-period/2), int64(period+period/2)))
	ticker := time.NewTicker(jittered)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-finish:
			return
		case <-ticker.C:
		}
		dctx, cancel := context.WithTimeout(ctx, traffic.RequestTimeout)
		_, err := c.Defragment(dctx)
		cancel()