Skip to content

fix ACL flakiness #1968

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Frostman merged 1 commit into from
Aug 14, 2026
Merged

fix ACL flakiness #1968

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 50 additions & 34 deletions pkg/hhfab/matrix.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Comment thread
Frostman marked this conversation as resolved.
}
}
})
}
}
}
Expand Down Expand Up @@ -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),
Comment thread
Frostman marked this conversation as resolved.
curls: semaphore.NewWeighted(opts.CurlsParallel),
inSources: func(name string) bool {
return len(opts.Sources) == 0 || slices.Contains(opts.Sources, name)
Expand Down
12 changes: 8 additions & 4 deletions pkg/hhfab/testing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}
Expand Down Expand Up @@ -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)
}

Expand Down
19 changes: 16 additions & 3 deletions pkg/hhfab/testing_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
Frostman marked this conversation as resolved.
[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 {
Expand All @@ -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}},
Expand Down Expand Up @@ -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) {
Expand Down
Loading