diff --git a/.github/workflows/e2e-nightly.yml b/.github/workflows/e2e-nightly.yml index 4210a09a7..543c8b0a1 100644 --- a/.github/workflows/e2e-nightly.yml +++ b/.github/workflows/e2e-nightly.yml @@ -16,6 +16,14 @@ permissions: # context exposes the id itself. actions: read +# Every cell builds its cluster on the same five hypervisors, so two runs in +# flight at once compete for their disks and their libvirt guests. Queue the +# second rather than cancelling it: a dispatch is usually a verification run +# and discarding it loses the result someone is waiting on. +concurrency: + group: e2e-nightly + cancel-in-progress: false + # Scheduled runs are loaded from the default branch (main), so the schedule # trigger must stay in sync on main for cron to fire. env: diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 5091228df..2c340180f 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -2832,6 +2832,22 @@ func (rm *ResourceManager) canAllocate(instanceType *ec2.InstanceTypeInfo, count // admission never advertises more instances than there are GPU slots to // back them. func (rm *ResourceManager) canAllocateLocked(instanceType *ec2.InstanceTypeInfo, count int) int { + n, _ := rm.admitLocked(instanceType, count) + return n +} + +// admit is admitLocked under the read lock. +func (rm *ResourceManager) admit(instanceType *ec2.InstanceTypeInfo, count int) (int, string) { + rm.mu.RLock() + defer rm.mu.RUnlock() + return rm.admitLocked(instanceType, count) +} + +// admitLocked returns the admission count and which gate bound it. A refusal +// that names its gate is the difference between "the node is full" and "the +// host is short of free memory right now", which have different fixes and are +// otherwise indistinguishable from InsufficientInstanceCapacity. +func (rm *ResourceManager) admitLocked(instanceType *ec2.InstanceTypeInfo, count int) (int, string) { instanceTypeName := "" if instanceType.InstanceType != nil { instanceTypeName = *instanceType.InstanceType @@ -2846,7 +2862,7 @@ func (rm *ResourceManager) canAllocateLocked(instanceType *ec2.InstanceTypeInfo, } } - n := canAllocateCount( + budget := canAllocateCount( rm.hostVCPU-rm.reservedVCPU-rm.reservedCRVCPU, rm.allocatedVCPU, rm.hostMemGB-rm.reservedMem-rm.reservedCRMem, rm.allocatedMem, instanceTypeVCPUs(instanceType), @@ -2854,7 +2870,11 @@ func (rm *ResourceManager) canAllocateLocked(instanceType *ec2.InstanceTypeInfo, count, availGPU, requiresGPU, ) - return rm.liveMemGate(n, instanceType) + n := rm.liveMemGate(budget, instanceType) + if n < budget { + return n, "live-memory" + } + return n, "budget" } // liveMemGate clamps n by current MemAvailable, catching overcommit that the @@ -2970,7 +2990,8 @@ func (rm *ResourceManager) updateInstanceSubscriptions() { defer rm.subsMu.Unlock() for typeName, typeInfo := range rm.instanceTypes { - canFit := rm.canAllocate(typeInfo, 1) >= 1 + fits, gate := rm.admit(typeInfo, 1) + canFit := fits >= 1 subjectRoot := "ec2.RunInstances" handler := rm.handler @@ -3002,7 +3023,7 @@ func (rm *ResourceManager) updateInstanceSubscriptions() { slog.Error("Failed to unsubscribe from instance type topic", "topic", queueTopic, "err", err) } delete(rm.instanceSubs, queueTopic) - slog.Info("Unsubscribed from instance type (capacity full)", "topic", queueTopic) + slog.Info("Unsubscribed from instance type (capacity full)", "topic", queueTopic, "gate", gate) } if rm.nodeID != "" { diff --git a/spinifex/handlers/ec2/eip/service_impl.go b/spinifex/handlers/ec2/eip/service_impl.go index db976bcfd..4a236539d 100644 --- a/spinifex/handlers/ec2/eip/service_impl.go +++ b/spinifex/handlers/ec2/eip/service_impl.go @@ -337,50 +337,64 @@ func (s *EIPServiceImpl) DisassociateByENI(ctx context.Context, accountID, eniID if eniID == "" { return false, nil } - record, key, revision, err := s.findByENI(ctx, accountID, eniID) + matches, err := s.findByENI(ctx, accountID, eniID) if err != nil { return false, err } - if record == nil { + if len(matches) == 0 { return false, nil } - // The MAC comes off the record rather than a fresh ENI lookup: this runs as - // part of tearing that ENI down, so the interface may already be gone. - s.publishNATEvent("vpc.delete-nat", record.VpcId, record.PublicIp, record.PrivateIp, eniID, record.MacAddress) + for _, m := range matches { + record := m.record - publicIP, allocationID := record.PublicIp, record.AllocationId - clearAssociation(record) + // The MAC comes off the record rather than a fresh ENI lookup: this runs + // as part of tearing that ENI down, so the interface may already be gone. + s.publishNATEvent("vpc.delete-nat", record.VpcId, record.PublicIp, record.PrivateIp, eniID, record.MacAddress) - data, err := json.Marshal(record) - if err != nil { - return false, fmt.Errorf("marshal EIP record: %w", err) + publicIP, allocationID := record.PublicIp, record.AllocationId + clearAssociation(record) + + data, err := json.Marshal(record) + if err != nil { + return false, fmt.Errorf("marshal EIP record: %w", err) + } + if _, err := s.eipKV.Update(ctx, m.key, data, m.revision); err != nil { + return false, errors.New(awserrors.ErrorServerInternal) + } + + slog.InfoContext(ctx, "DisassociateByENI completed", + "eniId", eniID, "publicIp", publicIP, "allocationId", allocationID, "accountID", accountID) } - if _, err := s.eipKV.Update(ctx, key, data, revision); err != nil { - return false, errors.New(awserrors.ErrorServerInternal) - } - - slog.InfoContext(ctx, "DisassociateByENI completed", - "eniId", eniID, "publicIp", publicIP, "allocationId", allocationID, "accountID", accountID) return true, nil } -// findByENI returns the associated EIP record holding eniID, or a nil record -// when the account has none. Unlike findByAssociationID an absent match is not -// an error: teardown asks this of every ENI, and most carry no EIP. -func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string) (*EIPRecord, string, uint64, error) { +// eipMatch is a decoded EIP record with the CAS handle needed to write it back. +type eipMatch struct { + record *EIPRecord + key string + revision uint64 +} + +// findByENI returns every associated EIP record holding eniID, empty when the +// account has none. Unlike findByAssociationID an absent match is not an error: +// teardown asks this of every ENI, and most carry no EIP. All matches are +// returned, not the first: an ENI can hold more than one EIP, and stopping at +// one strands the rest associated to an interface that is being deleted. +func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string) ([]eipMatch, error) { prefix := accountID + "." keys, err := s.eipKV.Keys(ctx) if err != nil { if errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, "", 0, nil + return nil, nil } - return nil, "", 0, errors.New(awserrors.ErrorServerInternal) + return nil, errors.New(awserrors.ErrorServerInternal) } + var matches []eipMatch for _, k := range keys { if err := ctx.Err(); err != nil { - return nil, "", 0, err + return nil, err } if k == utils.VersionKey || !strings.HasPrefix(k, prefix) { continue @@ -393,7 +407,7 @@ func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string) // An unreadable record must not read as "this ENI has no EIP": // the caller deletes the interface on that answer, and the // association would be stranded with nothing left to find it. - return nil, "", 0, fmt.Errorf("eipKV.Get(%s): %w", k, err) + return nil, fmt.Errorf("eipKV.Get(%s): %w", k, err) } var record EIPRecord if err := json.Unmarshal(entry.Value(), &record); err != nil { @@ -401,10 +415,10 @@ func (s *EIPServiceImpl) findByENI(ctx context.Context, accountID, eniID string) continue } if record.ENIId == eniID { - return &record, k, entry.Revision(), nil + matches = append(matches, eipMatch{record: &record, key: k, revision: entry.Revision()}) } } - return nil, "", 0, nil + return matches, nil } // describeAddressesValidFilters defines the set of filter names accepted by DescribeAddresses. diff --git a/spinifex/handlers/ec2/eip/service_impl_test.go b/spinifex/handlers/ec2/eip/service_impl_test.go index b2a2ec91e..a7aba8e39 100644 --- a/spinifex/handlers/ec2/eip/service_impl_test.go +++ b/spinifex/handlers/ec2/eip/service_impl_test.go @@ -776,3 +776,38 @@ func TestEIP_PublishNATEvent_PortNameHasPortPrefix(t *testing.T) { assert.Equal(t, "10.0.0.5", got.LogicalIP) assert.Equal(t, "198.51.100.10", got.ExternalIP) } + +// TestEIP_DisassociateByENI_ClearsEveryEIPOnTheInterface pins that teardown +// drains all of an ENI's EIPs. Stopping at the first left the rest marked +// associated to an interface being deleted, with their host routes installed +// and nothing remaining that could find them. +func TestEIP_DisassociateByENI_ClearsEveryEIPOnTheInterface(t *testing.T) { + svc, _, _ := setupTestEIP(t) + + var allocIDs []string + for range 3 { + out, err := svc.AllocateAddress(context.Background(), &ec2.AllocateAddressInput{}, testAccountID) + require.NoError(t, err) + allocIDs = append(allocIDs, *out.AllocationId) + associateToENI(t, svc, *out.AllocationId, "eni-shared") + } + + found, err := svc.DisassociateByENI(t.Context(), testAccountID, "eni-shared") + require.NoError(t, err) + assert.True(t, found) + + for _, allocID := range allocIDs { + entry, err := svc.eipKV.Get(t.Context(), testAccountID+"."+allocID) + require.NoError(t, err) + var record EIPRecord + require.NoError(t, json.Unmarshal(entry.Value(), &record)) + assert.Equal(t, "allocated", record.State, "%s still associated", allocID) + assert.Empty(t, record.ENIId, "%s still names the ENI", allocID) + assert.Empty(t, record.AssociationId, "%s kept its association", allocID) + } + + // One pass drains them all, so the retry teardown always makes finds none. + found, err = svc.DisassociateByENI(t.Context(), testAccountID, "eni-shared") + require.NoError(t, err) + assert.False(t, found) +} diff --git a/spinifex/network/policy/nat.go b/spinifex/network/policy/nat.go index 68ac1dc82..4cf8f8d42 100644 --- a/spinifex/network/policy/nat.go +++ b/spinifex/network/policy/nat.go @@ -391,6 +391,11 @@ func (m *natManager) DeleteEIP(ctx context.Context, vpcID, externalIP, logicalIP // Deleting on a pair that recycled identically would tear down the new owner's // rule and ARP entry; the stamped logical port is the discriminator that // survives identical-pair reuse. + // + // A row that is simply absent is not a reassignment. Nobody owns the IP, so + // the OVN delete has nothing to do but the host plumbing below is orphaned + // and is exactly what needs removing. + rowGone := false if logicalIP != "" || portName != "" { existing, err := m.ovn.FindNATByExternalIP(ctx, "dnat_and_snat", externalIP) switch { @@ -398,7 +403,7 @@ func (m *natManager) DeleteEIP(ctx context.Context, vpcID, externalIP, logicalIP slog.Warn("policy: DeleteEIP ownership lookup failed, proceeding with delete", "external_ip", externalIP, "logical_ip", logicalIP, "err", err) case existing == nil: - return nil + rowGone = true case logicalIP != "" && existing.LogicalIP != logicalIP: slog.Info("policy: DeleteEIP skip — external IP reassigned to a different logical IP (stale delete)", "external_ip", externalIP, "stale_logical_ip", logicalIP, "current_logical_ip", existing.LogicalIP) @@ -411,9 +416,11 @@ func (m *natManager) DeleteEIP(ctx context.Context, vpcID, externalIP, logicalIP return nil } } - if err := m.ovn.DeleteNATByExternalIP(ctx, router, "dnat_and_snat", externalIP); err != nil { - if !errors.Is(err, ovn.ErrNATNotFound) { - return fmt.Errorf("delete dnat_and_snat %s on %s: %w", externalIP, router, err) + if !rowGone { + if err := m.ovn.DeleteNATByExternalIP(ctx, router, "dnat_and_snat", externalIP); err != nil { + if !errors.Is(err, ovn.ErrNATNotFound) { + return fmt.Errorf("delete dnat_and_snat %s on %s: %w", externalIP, router, err) + } } } // Flush host ARP for the released IP so the next owner isn't shadowed. Best-effort. diff --git a/spinifex/network/policy/nat_binder_test.go b/spinifex/network/policy/nat_binder_test.go index 7f5d1e49e..7af063909 100644 --- a/spinifex/network/policy/nat_binder_test.go +++ b/spinifex/network/policy/nat_binder_test.go @@ -136,6 +136,43 @@ func TestNATManager_DeleteEIP_Routed_UnbindsHostEIP(t *testing.T) { assert.Equal(t, []string{"192.168.1.200"}, b.unbinds) } +// An absent dnat_and_snat row is not a reassignment: nobody owns the address, +// so the orphaned host plumbing is precisely what the delete has to remove. +// Returning early on it left the /32 route and proxy-ARP entry behind, which +// is what a disassociate looked like from the host. +func TestNATManager_DeleteEIP_Routed_UnbindsWhenRowAlreadyGone(t *testing.T) { + m := mock.New() + seedRouter(t, m, "vpc-1") + seedGatewayPortIP(t, m, "vpc-1", "100.127.0.10", nil) + b := &recordedBinder{} + mgr, err := NewNATManager(m, NATModeRouted, WithHostEIPBinder(b.hooks())) + require.NoError(t, err) + + // No AddEIP: the row never existed, standing in for one already removed. + require.NoError(t, mgr.DeleteEIP(context.Background(), "vpc-1", "192.168.1.200", "10.0.1.5", "eni-1")) + assert.Equal(t, []string{"192.168.1.200"}, b.unbinds) +} + +// The reassignment cases keep their early return: a different owner holds the +// address, and its host plumbing must survive a late delete for the old one. +func TestNATManager_DeleteEIP_Routed_ReassignedRowKeepsHostPlumbing(t *testing.T) { + m := mock.New() + seedRouter(t, m, "vpc-1") + seedGatewayPortIP(t, m, "vpc-1", "100.127.0.10", nil) + b := &recordedBinder{} + mgr, err := NewNATManager(m, NATModeRouted, WithHostEIPBinder(b.hooks())) + require.NoError(t, err) + + require.NoError(t, mgr.AddEIP(context.Background(), EIPSpec{ + VPCID: "vpc-1", ExternalIP: "192.168.1.200", LogicalIP: "10.0.1.9", + })) + b.unbinds = nil + + // Stale delete naming the previous logical IP. + require.NoError(t, mgr.DeleteEIP(context.Background(), "vpc-1", "192.168.1.200", "10.0.1.5", "")) + assert.Empty(t, b.unbinds, "a reassigned EIP must keep its host plumbing") +} + func TestNATManager_AddEIP_NonRoutedNeverBinds(t *testing.T) { for _, mode := range []NATMode{NATModeDistributed, NATModeCentralized} { m := mock.New() diff --git a/spinifex/services/viperblockd/provider_handlers.go b/spinifex/services/viperblockd/provider_handlers.go index cde31280e..e29067537 100644 --- a/spinifex/services/viperblockd/provider_handlers.go +++ b/spinifex/services/viperblockd/provider_handlers.go @@ -34,6 +34,11 @@ import ( // persists and the byte units the ebsprovider wire contract uses. const bytesPerGiB = 1024 * 1024 * 1024 +// pluginSealGrace bounds the wait for an nbdkit that is already sealing to +// finish and exit. The plugin bounds its own drain at 20s, so this only has to +// cover that plus writing the receipt. +const pluginSealGrace = 25 * time.Second + // providerObjectStoreFactory builds the objectstore.ObjectStore DeleteVolume // and DeleteSnapshot use to remove S3 object prefixes. Tests override this to // inject objectstore.NewMemoryObjectStore(), keeping the unit tests free of @@ -1613,6 +1618,12 @@ func unmountVolume(ctx context.Context, cfg *Config, volumeName string) (types.E matched.VB.Detach() } + // Removing the guest device starts the plugin's own Close, which + // drains to the backend and leaves a receipt. SIGKILLing into that + // discards the drain and leaves the WAL for the fallback seal below to + // replay, which costs far more than the caller's deadline allows. + utils.TerminateProcess(matched.PID, pluginSealGrace) + // The seal below rewrites the directory nbdkit writes, so a kill that // did not take makes it a concurrent write to that directory. Fail the // unmount instead: the entry stays mounted and a retry re-attempts diff --git a/spinifex/utils/process.go b/spinifex/utils/process.go index b87765298..192d1f2dc 100644 --- a/spinifex/utils/process.go +++ b/spinifex/utils/process.go @@ -79,6 +79,25 @@ func ForceKillProcess(pid int, timeout time.Duration) error { return WaitForProcessExit(pid, timeout) } +// TerminateProcess SIGTERMs a process and waits up to grace for it to exit, +// reporting whether it did. It never escalates: callers that must establish +// the process is gone follow it with ForceKillProcess, which is where the +// SIGKILL and the exit confirmation belong. +func TerminateProcess(pid int, grace time.Duration) bool { + if pid <= 0 { + return false + } + process, err := os.FindProcess(pid) + if err != nil { + return false + } + // Already gone is the outcome asked for, so report it as one. + if err := process.Signal(syscall.SIGTERM); err != nil { + return errors.Is(err, os.ErrProcessDone) || errors.Is(err, syscall.ESRCH) + } + return WaitForProcessExit(pid, grace) == nil +} + // killProcessPollInterval and killProcessGracePeriod drive KillProcess's // liveness poll: check every pollInterval instead of blocking for the full // gracePeriod, so a process that dies quickly is detected almost immediately. diff --git a/spinifex/utils/process_terminate_test.go b/spinifex/utils/process_terminate_test.go new file mode 100644 index 000000000..57c7a75dd --- /dev/null +++ b/spinifex/utils/process_terminate_test.go @@ -0,0 +1,76 @@ +package utils_test + +import ( + "bufio" + "os/exec" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/mulgadc/spinifex/spinifex/utils" +) + +// The unmount path calls this on an nbdkit that is already sealing, so the +// grace has to be spent only when the process needs it: one that exits on +// SIGTERM must be reaped in milliseconds, not after the full budget. +func TestTerminateProcessReturnsAsSoonAsTheProcessExits(t *testing.T) { + cmd := exec.Command("sleep", "60") // dies immediately on default SIGTERM + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + done := make(chan struct{}) + go func() { _ = cmd.Wait(); close(done) }() + + start := time.Now() + exited := utils.TerminateProcess(pid, 10*time.Second) + elapsed := time.Since(start) + + assert.True(t, exited, "a process that dies on SIGTERM must be reported as exited") + assert.Less(t, elapsed, time.Second, "must not wait out the grace period") + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("process did not exit after TerminateProcess returned") + } +} + +// A plugin wedged mid-drain must not hold the unmount open past the grace: +// the caller escalates to ForceKillProcess, and can only do that if this +// returns. +func TestTerminateProcessGivesUpAtTheGraceAndDoesNotKill(t *testing.T) { + cmd := exec.Command("sh", "-c", "trap '' TERM; echo ready; while true; do sleep 1; done") + stdout, err := cmd.StdoutPipe() + require.NoError(t, err) + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + }) + + scanner := bufio.NewScanner(stdout) + require.True(t, scanner.Scan(), "expected ready line from test process") + require.Equal(t, "ready", scanner.Text()) + + const grace = 200 * time.Millisecond + start := time.Now() + exited := utils.TerminateProcess(pid, grace) + elapsed := time.Since(start) + + assert.False(t, exited) + assert.GreaterOrEqual(t, elapsed, grace, "must wait out the grace before giving up") + assert.Less(t, elapsed, grace+2*time.Second, "must be bounded by the grace, not hang") + assert.True(t, utils.ProcessAlive(pid), "escalation belongs to the caller, not here") +} + +func TestTerminateProcessAlreadyExitedIsSuccess(t *testing.T) { + cmd := exec.Command("true") + require.NoError(t, cmd.Start()) + pid := cmd.Process.Pid + require.NoError(t, cmd.Wait()) + + assert.True(t, utils.TerminateProcess(pid, time.Second)) + assert.False(t, utils.TerminateProcess(0, time.Second), "an invalid pid is not an exit") +} diff --git a/tests/e2e/harness/fixtures.go b/tests/e2e/harness/fixtures.go index e86f39fae..7704b9c0e 100644 --- a/tests/e2e/harness/fixtures.go +++ b/tests/e2e/harness/fixtures.go @@ -364,11 +364,22 @@ func EnsureKeyPair(t *testing.T, fx *Fixture) (string, string) { if err != nil { return "", nil, fmt.Errorf("CreateKeyPair %s: %w", name, err) } + // Past this point the pair exists on the cluster but no cleanup is + // registered yet, so anything that fails has to remove it. A pair left + // behind fails every later caller with InvalidKeyPair.Duplicate, which + // reads as a different bug in a test that never ran. + abandon := func(cause error) (string, func() error, error) { + _, derr := fx.EC2.DeleteKeyPair(&ec2.DeleteKeyPairInput{KeyName: aws.String(name)}) + if derr != nil { + return "", nil, fmt.Errorf("%w (and could not remove the pair it created: %v)", cause, derr) + } + return "", nil, cause + } if err := os.WriteFile(pemPath, []byte(aws.StringValue(out.KeyMaterial)), 0o600); err != nil { - return "", nil, fmt.Errorf("write pem %s: %w", pemPath, err) + return abandon(fmt.Errorf("write pem %s: %w", pemPath, err)) } if err := assertOneKeyPair(fx, name); err != nil { - return "", nil, err + return abandon(err) } fx.tagRunResources(aws.StringValue(out.KeyPairId)) return aws.StringValue(out.KeyName), func() error { @@ -385,6 +396,13 @@ func EnsureKeyPair(t *testing.T, fx *Fixture) (string, string) { return id, pemPath } +// How long a just-created key pair is given to become visible to a describe, +// and how often that is retried. +const ( + keyPairVisibleWithin = 30 * time.Second + keyPairVisiblePoll = 500 * time.Millisecond +) + // assertOneKeyPair fails when the cluster holds more than one key pair under // name, which it should not be able to. // @@ -394,14 +412,34 @@ func EnsureKeyPair(t *testing.T, fx *Fixture) (string, string) { // rejecting the key at the SSH gate five minutes later, once per guest. Cheap // to check here, and it names the cause outright. func assertOneKeyPair(fx *Fixture, name string) error { - out, err := fx.EC2.DescribeKeyPairs(&ec2.DescribeKeyPairsInput{ - KeyNames: []*string{aws.String(name)}, - }) - if err != nil { - return fmt.Errorf("DescribeKeyPairs %s: %w", name, err) - } - if len(out.KeyPairs) == 1 { - return nil + // The create this follows has already succeeded, so nothing here can + // legitimately see zero. On a multi-node cluster the describe can land on a + // node that has not caught up yet, which is a propagation window rather + // than a result. Two is the answer this exists to catch, and two is + // reported the moment it is seen. + var out *ec2.DescribeKeyPairsOutput + deadline := time.Now().Add(keyPairVisibleWithin) + for { + var err error + out, err = fx.EC2.DescribeKeyPairs(&ec2.DescribeKeyPairsInput{ + KeyNames: []*string{aws.String(name)}, + }) + switch { + case err == nil && len(out.KeyPairs) == 1: + return nil + case err == nil && len(out.KeyPairs) > 1: + // The defect. Report it without waiting. + case err != nil && !ErrorCodeIs(err, "InvalidKeyPair.NotFound"): + return fmt.Errorf("DescribeKeyPairs %s: %w", name, err) + default: + // Not visible yet, or reported as absent. + if time.Now().Before(deadline) { + time.Sleep(keyPairVisiblePoll) + continue + } + return fmt.Errorf("key pair %s was created but is still not visible after %s", name, keyPairVisibleWithin) + } + break } got := make([]string, 0, len(out.KeyPairs)) diff --git a/tests/e2e/single/datapath_test.go b/tests/e2e/single/datapath_test.go index 2f0217ab8..63d049c92 100644 --- a/tests/e2e/single/datapath_test.go +++ b/tests/e2e/single/datapath_test.go @@ -363,11 +363,19 @@ func runSGPolicyDatapath(t *testing.T, fix *Fixture) { harness.Step(t, "8e-5b %s denies carry no ethertype qualifier", targetPG) rows := portGroupACLs(t, targetPG) + // Match the default-denies by their priorities, not by "any drop". + // The platform egress block is also a drop and is legitimately + // ip4-scoped, so bucketing every drop by direction lets it stand in + // for the rule this is about and fail on its qualifier. + denyPriority := map[string]int{ + "to-lport": aclPriorityDefaultDenyIngress, + "from-lport": aclPriorityDefaultDenyEgress, + } denies := map[string]aclRow{} arps := map[string]aclRow{} for _, r := range rows { switch { - case r.Action == "drop": + case r.Action == "drop" && r.Priority == denyPriority[r.Direction]: denies[r.Direction] = r case strings.Contains(r.Match, "arp"): arps[r.Direction] = r @@ -625,6 +633,14 @@ func runSGEInstance(t *testing.T, fix *Fixture, subnetID, sgID, userData string) } // aclRow is one NB ACL as OVN holds it, read back rather than rebuilt. +// The two default-deny priorities, mirroring policy.ACLPriorityDefaultDeny*. +// Duplicated rather than imported so this suite reads what OVN holds without +// depending on the package that put it there. +const ( + aclPriorityDefaultDenyIngress = 900 + aclPriorityDefaultDenyEgress = 800 +) + type aclRow struct { Priority int Direction string