diff --git a/pkg/hhfab/matrix.go b/pkg/hhfab/matrix.go index 13f0b9b47..7abb6c4c1 100644 --- a/pkg/hhfab/matrix.go +++ b/pkg/hhfab/matrix.go @@ -568,9 +568,12 @@ func IsSameEndpointNode(a, b *Endpoint) bool { } type matrixTestDeps struct { - sshByServer map[string]*sshutil.Config - pings *semaphore.Weighted - iperfs *semaphore.Weighted + sshByServer map[string]*sshutil.Config + pings *semaphore.Weighted + iperfs *semaphore.Weighted + // probes is fixed at 1: overlapping proto-port probes made one probe's + // congestion look like another's ACL drop (#1937). + probes *semaphore.Weighted curls *semaphore.Weighted inSources func(string) bool inDestinations func(string) bool @@ -835,57 +838,69 @@ func startMatrixProtoPortListeners(ctx context.Context, matrix *ConnectivityMatr // runMatrixProtoPortPhase exercises every non-zero ProtoPort matrix entry with a // protocol-specific probe: "icmp" → ping, "tcp" → nc connect, "udp" → iperf3 -u -// loss check. +// loss check. Probes run one at a time phase-wide, in the (icmp, tcp, udp) order +// ProtoPortEntries sorts a pair's entries into. func runMatrixProtoPortPhase(ctx context.Context, opts TestConnectivityOpts, matrix *ConnectivityMatrix, deps *matrixTestDeps) { for _, src := range matrix.AllEndpoints { for _, dst := range matrix.AllEndpoints { if !deps.probesServerPair(src, dst) { continue } + var entries []ConnectivityExpectation for _, entry := range matrix.ProtoPortEntries(src, dst) { if matrix.entryOwner(entry) != matrixPhaseProtoPort { continue } - expected := reachabilityFromExpectation(entry) - pp := entry.ProtoPort - fromName := src.Server.Name - toName := dst.Server.Name - fromSSH := deps.sshByServer[fromName] - toIP := dst.Server.IP - if entry.NAT != nil && entry.NAT.DestinationIP.IsValid() { - toIP = entry.NAT.DestinationIP - } - if !toIP.IsValid() { - deps.errChan <- fmt.Errorf("matrix proto entry %s→%s (%s/%d) has no valid target IP", fromName, toName, pp.Protocol, pp.Port) //nolint:goerr113 + entries = append(entries, entry) + } + if len(entries) == 0 { + continue + } - continue - } + fromName := src.Server.Name + toName := dst.Server.Name + fromSSH := deps.sshByServer[fromName] + dstIP := dst.Server.IP + + deps.wg.Go(func() { + for _, entry := range entries { + expected := reachabilityFromExpectation(entry) + pp := entry.ProtoPort + toIP := dstIP + if entry.NAT != nil && entry.NAT.DestinationIP.IsValid() { + toIP = entry.NAT.DestinationIP + } + if !toIP.IsValid() { + deps.errChan <- fmt.Errorf("matrix proto entry %s→%s (%s/%d) has no valid target IP", fromName, toName, pp.Protocol, pp.Port) //nolint:goerr113 + + continue + } + if err := deps.probes.Acquire(ctx, 1); err != nil { + deps.errChan <- fmt.Errorf("acquiring proto-port probe semaphore for %s→%s: %w", fromName, toName, err) + + return + } - switch pp.Protocol { - case "icmp": - deps.wg.Go(func() { + switch pp.Protocol { + case "icmp": if pe := checkPing(ctx, opts.PingsCount, deps.pings, fromName, toName, fromSSH, toIP, nil, expected); pe != nil { deps.errChan <- pe } - }) - case "tcp": - port := pp.Port - deps.wg.Go(func() { - if ie := checkTCPPort(ctx, deps.iperfs, fromName, fromSSH, toIP, port, expected); ie != nil { + case "tcp": + if ie := checkTCPPort(ctx, deps.iperfs, fromName, fromSSH, toIP, pp.Port, expected); ie != nil { deps.errChan <- ie } - }) - case "udp": - port := pp.Port - deps.wg.Go(func() { - if ie := checkUDPPort(ctx, opts, deps.iperfs, fromName, fromSSH, toIP, port, expected); ie != nil { + case "udp": + if ie := checkUDPPort(ctx, opts, deps.iperfs, fromName, fromSSH, toIP, pp.Port, expected); ie != nil { deps.errChan <- ie } - }) - default: - deps.errChan <- fmt.Errorf("matrix proto entry %s→%s has unsupported protocol %q", fromName, toName, pp.Protocol) //nolint:goerr113 + default: + deps.errChan <- fmt.Errorf("matrix proto entry %s→%s has unsupported protocol %q", fromName, toName, pp.Protocol) //nolint:goerr113 + } + + deps.probes.Release(1) } - } + }) } } } @@ -954,6 +969,7 @@ func (c *Config) TestConnectivityWithMatrix(ctx context.Context, vlab *VLAB, opt sshByServer: sshByServer, pings: semaphore.NewWeighted(opts.PingsParallel), iperfs: semaphore.NewWeighted(opts.IPerfsParallel), + probes: semaphore.NewWeighted(1), curls: semaphore.NewWeighted(opts.CurlsParallel), inSources: func(name string) bool { return len(opts.Sources) == 0 || slices.Contains(opts.Sources, name) diff --git a/pkg/hhfab/testing.go b/pkg/hhfab/testing.go index b2d107730..cf1f3ec4d 100644 --- a/pkg/hhfab/testing.go +++ b/pkg/hhfab/testing.go @@ -3083,9 +3083,10 @@ func checkPing(ctx context.Context, pingCount int, semaphore *semaphore.Weighted } // -D timestamps each reply line ([unixtime]) so a lost seq can be placed on - // the wall clock; it prefixes reply lines only, not the summary line the - // sent/received parser and parsePingLostSeqs read. - cmd = fmt.Sprintf("ping -i 0.5 -c %d -W 1 -D", pingCount) + // the wall clock, and -O places the loss itself there by reporting each + // unanswered seq as it times out. Neither line shape is read as a reply + // (they carry no "bytes from") nor as the sent/received summary. + cmd = fmt.Sprintf("ping -i 0.5 -c %d -W 1 -D -O", pingCount) if sourceIP != nil { cmd += " -I " + sourceIP.String() } @@ -3541,8 +3542,11 @@ func udpProbeTimingFor(secs int, expectReachable bool) udpProbeTiming { return udpProbeTiming{connect: connect, inner: inner, outer: inner + udpProbeSSHHeadroom} } +// 1Mbps clears the loss thresholds the verdict uses by the same margin any +// higher rate would, and a higher one congested the software dataplane enough to +// drop ICMP replies on the paths probed next to it (#1937). func udpProbeCmd(toIP netip.Addr, port uint16, secs int, timing udpProbeTiming) string { - return fmt.Sprintf("sudo docker exec iperf3 timeout -k 5 %d iperf3 -u -J --connect-timeout %d -c %s -p %d -t %d -b 10M -l 1000", + return fmt.Sprintf("sudo docker exec iperf3 timeout -k 5 %d iperf3 -u -J --connect-timeout %d -c %s -p %d -t %d -b 1M -l 1000", int(timing.inner.Seconds()), timing.connect.Milliseconds(), toIP.String(), port, secs) } diff --git a/pkg/hhfab/testing_test.go b/pkg/hhfab/testing_test.go index 9b606182c..6a120323f 100644 --- a/pkg/hhfab/testing_test.go +++ b/pkg/hhfab/testing_test.go @@ -354,6 +354,18 @@ From 10.20.4.1 icmp_seq=2 Destination Host Unreachable --- 10.20.4.2 ping statistics --- 5 packets transmitted, 4 received, 20% packet loss, time 2010ms rtt min/avg/max/mdev = 0.611/0.912/1.308/0.251 ms +` + + // Same loss with the -D -O pair the probe now runs: -O adds a line per + // unanswered seq, which is neither a reply nor the summary. + const middleLostReported = `PING 10.20.1.4 (10.20.1.4) 56(84) bytes of data. +[1782458023.423317] 64 bytes from 10.20.1.4: icmp_seq=1 ttl=62 time=0.253 ms +[1782458023.927269] 64 bytes from 10.20.1.4: icmp_seq=2 ttl=62 time=0.400 ms +[1782458024.935112] no answer yet for icmp_seq=3 +[1782458024.431201] 64 bytes from 10.20.1.4: icmp_seq=4 ttl=62 time=0.425 ms +[1782458024.935112] 64 bytes from 10.20.1.4: icmp_seq=5 ttl=62 time=0.478 ms +--- 10.20.1.4 ping statistics --- +5 packets transmitted, 4 received, 20% packet loss, time 2016ms ` for _, test := range []struct { @@ -365,6 +377,7 @@ rtt min/avg/max/mdev = 0.611/0.912/1.308/0.251 ms {name: "all received", stdout: allReceived, sent: 5}, {name: "first lost (real flake)", stdout: firstLost, sent: 5, expected: []int{1}}, {name: "first lost with -D timestamps", stdout: firstLostTimestamped, sent: 5, expected: []int{1}}, + {name: "middle lost with -D -O", stdout: middleLostReported, sent: 5, expected: []int{3}}, {name: "middle lost", stdout: middleLost, sent: 5, expected: []int{3}}, {name: "last lost", stdout: lastLost, sent: 5, expected: []int{5}}, {name: "all lost", stdout: allLost, sent: 5, expected: []int{1, 2, 3, 4, 5}}, @@ -484,18 +497,18 @@ func TestUDPProbeCmd(t *testing.T) { { name: "deny probe bounds the control connect", secs: 3, - expected: "sudo docker exec iperf3 timeout -k 5 18 iperf3 -u -J --connect-timeout 5000 -c 10.0.1.2 -p 5201 -t 3 -b 10M -l 1000", + expected: "sudo docker exec iperf3 timeout -k 5 18 iperf3 -u -J --connect-timeout 5000 -c 10.0.1.2 -p 5201 -t 3 -b 1M -l 1000", }, { name: "allow probe gets the longer connect budget", secs: 3, reachable: true, - expected: "sudo docker exec iperf3 timeout -k 5 28 iperf3 -u -J --connect-timeout 15000 -c 10.0.1.2 -p 5201 -t 3 -b 10M -l 1000", + expected: "sudo docker exec iperf3 timeout -k 5 28 iperf3 -u -J --connect-timeout 15000 -c 10.0.1.2 -p 5201 -t 3 -b 1M -l 1000", }, { name: "extended run stretches the backstop", secs: 10, - expected: "sudo docker exec iperf3 timeout -k 5 25 iperf3 -u -J --connect-timeout 5000 -c 10.0.1.2 -p 5201 -t 10 -b 10M -l 1000", + expected: "sudo docker exec iperf3 timeout -k 5 25 iperf3 -u -J --connect-timeout 5000 -c 10.0.1.2 -p 5201 -t 10 -b 1M -l 1000", }, } { t.Run(test.name, func(t *testing.T) {