[Antithesis+Robustness] Add a client that connects to all members

[serathius]

What would you like to be added?

Currently we configure clients to connect to a single etcd member. This ensures that member will always receive traffic after it goes down. If clients were created to all of etcd they would migrate out of it and not get back due to stickiness.

Still I think it would be good to have 1 client that loadbalances. We can split ClientCount available into MemberClientCount and ClusterClientCount, where member connects to single member and cluster to all members in cluster.

/cc @joshjms @henrybear327 @nwnt

Why is this needed?

Reason is finding bugs in retry logic in cases where request failed to first member and then loadbalancer moved it to other member for a retry.

[nwnt]

/assign

[nwnt]

@serathius looks like the current RecordingClient factory function can support connecting to all members. would it be better if we try picking one test scenario and then making it use both a client connecting to a member and client connecting to the whole cluster?

[serathius]

I don't understand. The goal is to have direct control over number of clients connecting directly to members and connected to all.

[nwnt]

I think I have an incorrect understanding about this one. Would you be able to give a bit more details about what the end result would look like?

[serathius]

Change

	for i := 0; i < profile.ClientCount; i++ {
		c := connect([]string{hosts[i%len(hosts)]}, ids, baseTime)
		defer c.Close()
		wg.Add(1)
		go func(c *client.RecordingClient) {
			defer wg.Done()
			defer c.Close()

			traffic.EtcdPutDeleteLease.RunTrafficLoop(ctx, c, limiter,
				ids,
				storage,
				concurrencyLimiter,
				keyStore,
				finish,
			)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c)
	}

To

	for i := 0; i < profile.MemberClientCount; i++ {
		c := connect([]string{hosts[i%len(hosts)]}, ids, baseTime)
		defer c.Close()
		wg.Add(1)
		go func(c *client.RecordingClient) {
			defer wg.Done()
			defer c.Close()

			traffic.EtcdPutDeleteLease.RunTrafficLoop(ctx, c, limiter,
				ids,
				storage,
				concurrencyLimiter,
				keyStore,
				finish,
			)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c)
	}	
	for i := 0; i < profile.ClusterClientCount; i++ {
		c := connect(hosts, ids, baseTime)
		defer c.Close()
		wg.Add(1)
		go func(c *client.RecordingClient) {
			defer wg.Done()
			defer c.Close()

			traffic.EtcdPutDeleteLease.RunTrafficLoop(ctx, c, limiter,
				ids,
				storage,
				concurrencyLimiter,
				keyStore,
				finish,
			)
			mux.Lock()
			reports = append(reports, c.Report())
			mux.Unlock()
		}(c)
	}

[serathius]

Possibly also refactor code to make it clearer. The code for setting up the traffic is becoming very convoluted.

[serathius]

#19893 would be nice

[nwnt]

Thank you. I'll get this and refactoring going once #20140 is merged (🤞) to avoid the potential conflict here.

[serathius]

Done