From c2aad92044e4aba2f85106dbd60b475a95c4c33b Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Sat, 5 Sep 2026 21:41:02 -0500 Subject: [PATCH] fix: stop the detach seal, EIP teardown and key-pair races failing the nightly (#1003) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: stop the detach seal, EIP teardown and key-pair races failing the nightly Four independent faults the full-suite nightly surfaced, each of which had been masked while the heavy cells were skipped. unmountVolume SIGKILLed nbdkit the moment the request arrived. Removing the guest device has already started the plugin's own Close, which drains to the backend and leaves a seal receipt, so the kill landed 61ms into that flush and threw it away — the receipt path never fired once across a whole run. What was left was 74 orphaned WAL files for the fallback seal to replay, which ran past the caller's 180s deadline and returned DetachVolume ServerInternal with the volume still attached. It now terminates gracefully first, bounded by the plugin's own drain budget, and only escalates if that does not take. DeleteEIP returned early when the dnat_and_snat row was already absent, before the neighbour flush and the host unbind. An absent row is not a reassignment — nobody owns the address — so the /32 route and proxy-ARP entry were orphaned and a disassociate left the address still answering from the host. The reassignment cases keep their early return. EnsureKeyPair was not atomic: a failure after CreateKeyPair left the pair on the cluster with no cleanup registered, so every later caller failed with InvalidKeyPair.Duplicate in a test that never ran. The describe that follows a create also had no tolerance for the propagation window on a multi-node cluster; it now retries, and still reports a duplicate the moment it sees one. An admission refusal could not name its cause. The static vCPU/memory budget and the live MemAvailable clamp both surface as InsufficientInstanceCapacity, and they have different fixes; the subscription transition now records which gate closed. * fix: identify the default-deny by priority, not by being a drop DenyIsNotEthertypeScoped bucketed every drop ACL by direction, so the last one seen won each direction. The platform WAN egress block added at priority 1049 is also a drop and is legitimately ip4-scoped, so it displaced the default-deny in the map and the assertion fired on its qualifier instead: from-lport deny on sg_c94870c0d68288dc6 is IPv4-scoped: "inport == @sg_c94870c0d68288dc6 && ip4 && ip4.dst != {10.0.0.0/8, ...}" Select the two default-denies by their own priorities. What the test asserts is unchanged — the default-deny still must carry no ethertype qualifier, for the reason denyIngressACL states: OVN allows on no-match, so an ip4-scoped deny lets every other ethertype through. * fix: drain every EIP on an ENI at teardown, not just the first findByENI returned the first matching record and stopped, so an ENI holding more than one EIP had all but one left marked associated to an interface being deleted — with the host route still installed and no record left naming the ENI to find them. Teardown now clears them all. * ci: serialise e2e-nightly runs so two do not share the hypervisors Only iso-single and iso-multi were guarded, by the iso-qemu-rack group. Every other job could overlap a second run of the workflow, with both sets of cells building clusters on the same five hosts. Queue rather than cancel: a dispatch is usually a verification run whose result someone is waiting on. --- .github/workflows/e2e-nightly.yml | 8 ++ spinifex/daemon/daemon.go | 29 ++++++- spinifex/handlers/ec2/eip/service_impl.go | 66 +++++++++------- .../handlers/ec2/eip/service_impl_test.go | 35 +++++++++ spinifex/network/policy/nat.go | 15 +++- spinifex/network/policy/nat_binder_test.go | 37 +++++++++ .../services/viperblockd/provider_handlers.go | 11 +++ spinifex/utils/process.go | 19 +++++ spinifex/utils/process_terminate_test.go | 76 +++++++++++++++++++ tests/e2e/harness/fixtures.go | 58 +++++++++++--- tests/e2e/single/datapath_test.go | 18 ++++- 11 files changed, 327 insertions(+), 45 deletions(-) create mode 100644 spinifex/utils/process_terminate_test.go 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