diff --git a/sandboxd/engine/capacity_test.go b/sandboxd/engine/capacity_test.go new file mode 100644 index 0000000..5c1be76 --- /dev/null +++ b/sandboxd/engine/capacity_test.go @@ -0,0 +1,28 @@ +package engine + +import ( + "errors" + "testing" +) + +// TestOnlyNodeWideFailuresCarryACapacitySignature guards the classifier. +// Parking on an ordinary clone failure would stall a healthy node over one bad VM. +func TestOnlyNodeWideFailuresCarryACapacitySignature(t *testing.T) { + for _, tc := range []struct { + name string + err error + want bool + }{ + {"bridge port exhaustion", errors.New(`failed to connect "veth1" to bridge cni0: exchange full`), true}, + {"disk exhaustion", errors.New("write overlay: no space left on device"), true}, + {"one VM failed to boot", errors.New("cocoon vm clone: exit status 1: Error: guest did not respond"), false}, + {"missing golden", errors.New("cocoon vm clone: exit status 1: Error: snapshot not found"), false}, + {"no error", nil, false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := CapacitySignature(tc.err) != ""; got != tc.want { + t.Errorf("CapacitySignature(%v) carries a signature: %v, want %v", tc.err, got, tc.want) + } + }) + } +} diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index 05c7e67..a3b1da5 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -47,6 +47,15 @@ const ( portForwardMax = 4096 ) +// capacitySignatures mean "this node cannot attach another VM", not "this VM +// failed"; matched as text because cocoon is a subprocess and the distinction +// survives only in its stderr. "exchange full" is EXFULL from a bridge at +// BR_MAX_PORTS — permanent until VMs are removed. +var capacitySignatures = []string{ + "exchange full", + "no space left on device", +} + // Engine runs cocoon commands on the local node. type Engine struct { bin string @@ -463,6 +472,20 @@ func readLine(conn net.Conn, max int) (string, error) { return "", fmt.Errorf("reply exceeds %d bytes", max) } +// CapacitySignature returns the node-capacity signature err carries, or "". +func CapacitySignature(err error) string { + if err == nil { + return "" + } + msg := strings.ToLower(err.Error()) + for _, sig := range capacitySignatures { + if strings.Contains(msg, sig) { + return sig + } + } + return "" +} + func tail(s string) string { s = strings.TrimSpace(s) if len(s) <= outputTail { diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index e1cc1e7..c34d919 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -141,9 +141,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { m.disarmEgress(sb.ID, true) sb.Transition.Unlock() // Committed: the store ck is authoritative now; reclaim the local footprint. - if rmErr := m.eng.Remove(ctx, vmName); rmErr != nil { - log.WithFunc("pool.archive").Warnf(ctx, "remove archived VM %s: %v", vmName, rmErr) - } + m.destroy(ctx, vmName) m.dropSnap(ctx, snap) m.counters.archives.Add(1) m.recordUsage(ctx, usageEvent{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant}) diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 5a88052..b628eb5 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -129,8 +129,8 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand m.purgeArchiveCk(ctx, id, ck, sb.Tenant) // archived: no local VM } var err error - if vmName != "" { - err = m.eng.Remove(ctx, vmName) + if vmName != "" && !m.removeOrRetry(ctx, vmName, id, "") { + err = fmt.Errorf("vm %s survived removal", vmName) } m.disarmEgress(id, err == nil) m.dropSnap(ctx, snap) @@ -291,7 +291,7 @@ func (m *Manager) rollbackClaim(ctx context.Context, sbs []*types.Sandbox) { m.mu.Unlock() m.recommit(ctx, rb) for _, sb := range sbs { - m.disarmEgress(sb.ID, m.removeVM(ctx, sb.VMName)) + m.disarmEgress(sb.ID, m.removeOrRetry(ctx, sb.VMName, sb.ID, "")) } } @@ -370,7 +370,7 @@ func (m *Manager) reapOnce(ctx context.Context) { logger.Errorf(ctx, err, "archive expired %s", v.id) } default: - m.disarmEgress(v.id, m.removeVM(ctx, v.vmName)) + m.disarmEgress(v.id, m.removeOrRetry(ctx, v.vmName, v.id, "")) m.dropSnap(ctx, v.snap) m.counters.reaps.Add(1) m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id, VMName: v.vmName}) diff --git a/sandboxd/pool/egress.go b/sandboxd/pool/egress.go index 5753750..290f141 100644 --- a/sandboxd/pool/egress.go +++ b/sandboxd/pool/egress.go @@ -120,15 +120,14 @@ func (m *Manager) lockEgressNIC(ctx context.Context, sb *types.Sandbox) error { } func (m *Manager) tapOf(ctx context.Context, vmName string) (string, error) { - vms, err := m.eng.List(ctx, vmName) + vm, ok, err := m.findVM(ctx, vmName) if err != nil { return "", fmt.Errorf("list %s: %w", vmName, err) } - i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == vmName }) - if i < 0 { + if !ok { return "", fmt.Errorf("no NIC config for %s", vmName) } - if tap := vms[i].TapDevice(); tap != "" { + if tap := vm.TapDevice(); tap != "" { return tap, nil } return "", fmt.Errorf("no tap for %s", vmName) diff --git a/sandboxd/pool/egress_test.go b/sandboxd/pool/egress_test.go index ec80d64..eada552 100644 --- a/sandboxd/pool/egress_test.go +++ b/sandboxd/pool/egress_test.go @@ -254,6 +254,9 @@ func TestBatchArmFailureRecordsNoUsage(t *testing.T) { func TestRollbackKeepsLockWhenRemoveFails(t *testing.T) { eng := newFakeEngine() eng.removeErrFor = "sbx-r1" + // Both VMs are running; sbx-r1's remove fails and the engine keeps listing it. + eng.vms["sbx-r1"] = "/run/sbx-r1.sock" + eng.vms["sbx-r2"] = "/run/sbx-r2.sock" m := egressManager(t, eng, config.PoolSpec{PoolKey: egKey, Egress: egPolicy}) // Both members armed before a later batch member failed; sbx-r1's remove fails. sbs := []*types.Sandbox{ @@ -287,6 +290,8 @@ func TestRollbackKeepsLockWhenRemoveFails(t *testing.T) { func TestQuarantineFailedRemoveStaysUnswept(t *testing.T) { eng := newFakeEngine() eng.removeErrFor = "sbx-q1" + // Running: removeVM reports "still here" only while the engine lists it. + eng.vms["sbx-q1"] = "/run/sbx-q1.sock" m := egressManager(t, eng, config.PoolSpec{PoolKey: egKey, Egress: egPolicy}) sb := &types.Sandbox{ID: "sb_q1", VMName: "sbx-q1", Key: egKey, TAP: "tap-q1"} m.mu.Lock() diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 8f62356..5a9f259 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -35,11 +35,19 @@ import ( ) const ( - refillInterval = 2 * time.Second - reapInterval = 5 * time.Second - buildRetryDelay = 30 * time.Second - claimProbeTimeout = 15 * time.Second - coldProbeTimeout = 90 * time.Second + refillInterval = 2 * time.Second + reapInterval = 5 * time.Second + buildRetryDelay = 30 * time.Second + claimProbeTimeout = 15 * time.Second + coldProbeTimeout = 90 * time.Second + // A refill has no caller waiting and holds a target-accounting slot for + // its whole probe; a clone answers in ~1s even saturated, so a silent one + // is replaced, not waited on. Claims keep the generous deadline. + warmProbeTimeout = 5 * time.Second + // One list; a wrong answer only costs an extra sweep. + removeVerifyTimeout = 15 * time.Second + // A full node clears only when VMs go away; retrying sooner buys nothing. + capacityBackoff = buildRetryDelay vsockPollInterval = 100 * time.Millisecond defaultTTL = 5 * time.Minute maxTTL = 24 * time.Hour @@ -134,6 +142,15 @@ type Gauges struct { Hibernated int Archived int Draining bool + // AtCapacity marks refill parked because the node refused another VM, so a + // placer reads a short warm count as "full", not "still filling". + AtCapacity bool + AtCapacityReason string +} + +type pendingRemoval struct { + sandboxID string + tap string } type pool struct { @@ -319,9 +336,15 @@ type Manager struct { dial egress.DialFunc sweep func(map[string]bool) error - mu sync.Mutex - pools map[types.PoolKey]*pool - claimed map[string]*types.Sandbox + mu sync.Mutex + pools map[types.PoolKey]*pool + claimed map[string]*types.Sandbox + pendingRemovals map[string]pendingRemoval + + // Node-wide, unlike the per-pool streak backoff: the resource that ran + // out — bridge ports, disk — is shared by every pool. + atCapacityUntil time.Time + atCapacityReason string refillSem chan struct{} probeSem chan struct{} @@ -350,6 +373,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg poolStore: newPoolStore(cfg.DataDir), pools: make(map[types.PoolKey]*pool, len(cfg.Pools)), claimed: map[string]*types.Sandbox{}, + pendingRemovals: map[string]pendingRemoval{}, tenantLive: map[string]int{}, archiving: map[string]struct{}{}, pendingCks: map[string]struct{}{}, @@ -485,6 +509,7 @@ func (m *Manager) Run(ctx context.Context) { case <-m.refillKick: m.refillOnce(ctx) case <-reap.C: + m.retryRemovals(ctx) m.reapOnce(ctx) m.idleOnce(ctx) m.archiveOnce(ctx) @@ -538,6 +563,9 @@ func (m *Manager) Info() ([]PoolInfo, Gauges) { } slices.SortFunc(pools, func(a, b PoolInfo) int { return strings.Compare(a.Key.Hash(), b.Key.Hash()) }) g := Gauges{Claimed: len(m.claimed), Draining: m.draining} + if now.Before(m.atCapacityUntil) { + g.AtCapacity, g.AtCapacityReason = true, m.atCapacityReason + } for _, sb := range m.claimed { if sb.HibernateSnap != "" { g.Hibernated++ diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index a029e6e..4edc817 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -131,7 +131,7 @@ func (m *Manager) sweepStaleVMs(ctx context.Context, live map[string]types.VMRec } gone := make([]bool, len(stale)) // distinct indices: no lock under the Wait barrier m.runBounded(ctx, len(stale), func(ctx context.Context, i int) { - if m.removeVM(ctx, stale[i]) { + if m.removeOrRetry(ctx, stale[i], "", live[stale[i]].TapDevice()) { gone[i] = true logger.Infof(ctx, "removed stale VM %s", stale[i]) } @@ -190,12 +190,9 @@ func (m *Manager) resyncEgress(ctx context.Context, live map[string]types.VMReco } } -// quarantineClaim removes a claim whose egress NIC could not be locked and drops -// it from service; it returns whether the VM is confirmed gone (a failed remove -// leaves it for the next restart's stale sweep). A never-applied lock plus a -// failed remove leaves the VM unguarded until a later remove succeeds. +// A failed remove stays out of service and queued until teardown succeeds. func (m *Manager) quarantineClaim(ctx context.Context, sb *types.Sandbox) bool { - gone := m.removeVM(ctx, sb.VMName) + gone := m.removeOrRetry(ctx, sb.VMName, sb.ID, "") m.mu.Lock() delete(m.claimed, sb.ID) m.tenantDelta(sb.Tenant, -1) diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index e6e356e..191d10f 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -13,6 +13,7 @@ import ( "github.com/projecteru2/core/log" + "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" ) @@ -32,6 +33,10 @@ func (m *Manager) refillOnce(ctx context.Context) { return } now := time.Now() + // Spawning VMs into a full node is what turned a capacity limit into an + // outage: every doomed attempt still takes the global rtnl lock. The park + // gates spawns only — sweeping removed pools is pure bookkeeping. + parked := now.Before(m.atCapacityUntil) inFlight := 0 for key, p := range m.pools { inFlight += p.refilling @@ -43,11 +48,14 @@ func (m *Manager) refillOnce(ctx context.Context) { } continue } - if p.goldenDir == "" && !p.building && now.After(p.nextBuild) { + if !parked && p.goldenDir == "" && !p.building && now.After(p.nextBuild) { p.building = true go m.buildGolden(ctx, p) } } + if parked { + return + } limit := cap(m.refillSem) + cap(m.probeSem) for spawned := true; spawned && inFlight < limit; { spawned = false @@ -82,11 +90,12 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string) { if ctx.Err() == nil { m.refillOnce(ctx) } - sb, err = m.readyBounded(ctx, sb, time.Now().Add(claimProbeTimeout)) + sb, err = m.readyBounded(ctx, sb, time.Now().Add(warmProbeTimeout)) } keep := false var fails int var wait time.Duration + capReason := engine.CapacitySignature(err) m.mu.Lock() now := time.Now() p.refilling-- @@ -95,14 +104,27 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string) { p.noteLead(time.Since(start)) keep = true } - // A canceled context is the daemon going down, not the pool failing. - backedOff := ctx.Err() == nil && p.noteRefillResult(now, err != nil) + // A canceled context is the daemon going down, and a capacity failure is + // the node's fault — neither says anything about this pool's health. + backedOff := false + if ctx.Err() == nil && capReason == "" { + backedOff = p.noteRefillResult(now, err != nil) + } if backedOff { fails, wait = p.refillFails, time.Until(p.nextRefill) } + entered := capReason != "" && m.noteCapacityLocked(now, capReason) + parked := now.Before(m.atCapacityUntil) m.mu.Unlock() if err != nil { - if ctx.Err() == nil { + switch { + case ctx.Err() != nil: + case entered: + // One line per episode; the storm this replaces wrote 20k an hour. + log.WithFunc("pool.refillOne").Errorf(ctx, err, + "node is at capacity; parking refill for %s", capacityBackoff) + case parked: + default: hash := p.key.Hash() log.WithFunc("pool.refillOne").Errorf(ctx, err, "refill %s", hash) if backedOff { @@ -121,6 +143,15 @@ func (m *Manager) refillOne(ctx context.Context, p *pool, golden string) { } } +// noteCapacityLocked parks refill node-wide for a capacity failure, reporting +// whether this call parked it so the episode is logged once. m.mu held. +func (m *Manager) noteCapacityLocked(now time.Time, reason string) bool { + first := !now.Before(m.atCapacityUntil) + m.atCapacityUntil = now.Add(capacityBackoff) + m.atCapacityReason = reason + return first +} + func (m *Manager) buildGolden(ctx context.Context, p *pool) { logger := log.WithFunc("pool.buildGolden") hash := p.key.Hash() @@ -140,6 +171,11 @@ func (m *Manager) buildGolden(ctx context.Context, p *pool) { p.goldenDir = final } else { p.nextBuild = time.Now().Add(buildRetryDelay) + // A cold boot hits the same bridge and disk as a refill; feed the park + // so a node failing only its builds still publishes at-capacity. + if reason := engine.CapacitySignature(err); reason != "" { + m.noteCapacityLocked(time.Now(), reason) + } } m.mu.Unlock() if err != nil { @@ -375,18 +411,30 @@ func (m *Manager) probeReady(ctx context.Context, name, sock string, timeout tim } func (m *Manager) vsockOf(ctx context.Context, name string) (string, error) { - vms, err := m.eng.List(ctx, name) + vm, ok, err := m.findVM(ctx, name) if err != nil { return "", err } - i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == name }) - if i < 0 { + if !ok { return "", fmt.Errorf("vm %s not found after create", name) } - if vms[i].VsockSocket == "" { + if vm.VsockSocket == "" { return "", fmt.Errorf("vm %s has no vsock socket", name) } - return vms[i].VsockSocket, nil + return vm.VsockSocket, nil +} + +// findVM returns the engine's record for name, if it still lists one. +func (m *Manager) findVM(ctx context.Context, name string) (types.VMRecord, bool, error) { + vms, err := m.eng.List(ctx, name) + if err != nil { + return types.VMRecord{}, false, err + } + i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == name }) + if i < 0 { + return types.VMRecord{}, false, nil + } + return vms[i], true, nil } // runBounded fans f over n items on the refill semaphore, so engine @@ -411,19 +459,6 @@ func (m *Manager) runBounded(ctx context.Context, n int, f func(context.Context, return &wg } -// removeVM deletes a VM (on a cancellation-immune ctx, else `cocoon vm rm` -// no-ops and orphans it) and reports whether it is confirmed gone; a failed -// remove leaves it running, so the caller must keep its lock. -func (m *Manager) removeVM(ctx context.Context, name string) bool { - if err := m.eng.Remove(context.WithoutCancel(ctx), name); err != nil { - log.WithFunc("pool.removeVM").Errorf(ctx, err, "remove vm %s", name) - return false - } - return true -} - -func (m *Manager) destroy(ctx context.Context, name string) { _ = m.removeVM(ctx, name) } - func (m *Manager) dropSnap(ctx context.Context, snap string) { if snap == "" { return diff --git a/sandboxd/pool/refillbackoff_test.go b/sandboxd/pool/refillbackoff_test.go index a13b9c3..c7ecad2 100644 --- a/sandboxd/pool/refillbackoff_test.go +++ b/sandboxd/pool/refillbackoff_test.go @@ -64,7 +64,9 @@ func TestBackoffGrowsAndIsCapped(t *testing.T) { // while a pool is backed off, the ticker spawns nothing. func TestRefillOnceHonorsTheBackoff(t *testing.T) { eng := newFakeEngine() - eng.cloneErr = errors.New("configure network: failed to connect veth to bridge cni0: exchange full") + // Deliberately not a capacitySignature: that parks the node after one + // failure; the streak backoff is the fallback for unclassifiable errors. + eng.cloneErr = errors.New("clone: guest kernel panicked before vsock came up") m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 50}) m.pools[testKey].goldenDir = "/goldens/x" diff --git a/sandboxd/pool/refillkick_test.go b/sandboxd/pool/refillkick_test.go index 2642748..3634362 100644 --- a/sandboxd/pool/refillkick_test.go +++ b/sandboxd/pool/refillkick_test.go @@ -1,6 +1,12 @@ package pool -import "testing" +import ( + "errors" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/config" +) func TestKickRefillCoalescesAndNeverBlocks(t *testing.T) { m := &Manager{refillKick: make(chan struct{}, 1)} @@ -16,3 +22,47 @@ func TestKickRefillCoalescesAndNeverBlocks(t *testing.T) { t.Fatalf("kick after drain not delivered") } } + +// TestNodeAtCapacityParksRefillInsteadOfRetrying pins the difference between +// "this VM failed to start" and "this node cannot hold another VM". +func TestNodeAtCapacityParksRefillInsteadOfRetrying(t *testing.T) { + eng := newFakeEngine() + eng.cloneErr = errors.New(`cocoon vm clone: exit status 1: Error: configure network: ` + + `cni add A/eth0: plugin type="bridge" failed (add): ` + + `failed to connect "veth3f2" to bridge cni0: exchange full`) + m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 4}) + m.pools[testKey].goldenDir = "/goldens/x" + + m.refillOnce(t.Context()) + waitFor(t, func() bool { + infos, _ := m.Info() + return infos[0].Refilling == 0 + }) + attempts := eng.cloneCount() + if attempts == 0 { + t.Fatal("no clone attempted, so the test proves nothing") + } + + if _, g := m.Info(); !g.AtCapacity || g.AtCapacityReason == "" { + t.Fatalf("at-capacity not published: %+v", g) + } + + // Every later tick is a no-op until the backoff expires. + for range 5 { + m.refillOnce(t.Context()) + } + if n := eng.cloneCount(); n != attempts { + t.Errorf("clones=%d after parking, want %d: a parked node must not attempt again", n, attempts) + } + + // A park, not a latch: once the backoff expires refill resumes. + m.mu.Lock() + m.atCapacityUntil = time.Now().Add(-time.Second) + m.mu.Unlock() + eng.cloneErr = nil + m.refillOnce(t.Context()) + waitFor(t, func() bool { + infos, _ := m.Info() + return infos[0].Warm == 4 + }) +} diff --git a/sandboxd/pool/remove.go b/sandboxd/pool/remove.go new file mode 100644 index 0000000..122ebff --- /dev/null +++ b/sandboxd/pool/remove.go @@ -0,0 +1,93 @@ +package pool + +import ( + "context" + "fmt" + "maps" + "slices" + "sync" + + "github.com/projecteru2/core/log" + + "github.com/cocoonstack/sandbox/sandboxd/netfilter" +) + +func (m *Manager) removeVM(ctx context.Context, name string) bool { + // Cancellation-immune: on a canceled ctx `cocoon vm rm` no-ops and orphans. + ctx = context.WithoutCancel(ctx) + err := m.eng.Remove(ctx, name) + if err == nil { + return true + } + log.WithFunc("pool.removeVM").Warnf(ctx, "remove vm %s: %v; verifying", name, err) + return m.confirmGone(ctx, name) +} + +func (m *Manager) confirmGone(ctx context.Context, name string) bool { + ctx, cancel := context.WithTimeout(ctx, removeVerifyTimeout) + defer cancel() + _, present, err := m.findVM(ctx, name) + if err != nil { + // Cannot tell: a false "gone" leaks a running VM, a retry only costs a sweep. + log.WithFunc("pool.confirmGone").Warnf(ctx, "verify remove of %s: %v", name, err) + return false + } + if present { + log.WithFunc("pool.confirmGone").Errorf(ctx, + fmt.Errorf("vm %s survived removal", name), + "remove did not take effect; leaving it accounted for retry") + return false + } + return true +} + +// removeOrRetry reports whether the VM is confirmed gone; a survivor is queued +// for the reap tick to retry, carrying what its cleanup needs (the sandbox ID +// when a claim owns the tap via egressTaps, the tap itself when none does). +func (m *Manager) removeOrRetry(ctx context.Context, name, sandboxID, tap string) bool { + if m.removeVM(ctx, name) { + return true + } + m.queueRemoval(name, sandboxID, tap) + return false +} + +func (m *Manager) queueRemoval(name, sandboxID, tap string) { + m.mu.Lock() + m.pendingRemovals[name] = pendingRemoval{sandboxID: sandboxID, tap: tap} + m.mu.Unlock() +} + +// retryRemovals dispatches by emptying the queue, so absence also means "in +// flight" and the next tick cannot double-dispatch; a failed retry re-queues +// itself on completion. Callers that need completion Wait. +func (m *Manager) retryRemovals(ctx context.Context) *sync.WaitGroup { + m.mu.Lock() + if len(m.pendingRemovals) == 0 { + m.mu.Unlock() + return new(sync.WaitGroup) + } + batch := m.pendingRemovals + m.pendingRemovals = map[string]pendingRemoval{} + m.mu.Unlock() + names := slices.Collect(maps.Keys(batch)) + return m.runBounded(ctx, len(names), func(ctx context.Context, i int) { + m.retryRemoval(ctx, names[i], batch[names[i]]) + }) +} + +func (m *Manager) retryRemoval(ctx context.Context, name string, pending pendingRemoval) { + if !m.removeVM(ctx, name) { + m.queueRemoval(name, pending.sandboxID, pending.tap) + return + } + if pending.sandboxID != "" { + m.disarmEgress(pending.sandboxID, true) + } else if pending.tap != "" { + _ = netfilter.Unlock(pending.tap) + } +} + +func (m *Manager) destroy(ctx context.Context, name string) { + m.removeOrRetry(ctx, name, "", "") +} diff --git a/sandboxd/pool/remove_test.go b/sandboxd/pool/remove_test.go new file mode 100644 index 0000000..7331896 --- /dev/null +++ b/sandboxd/pool/remove_test.go @@ -0,0 +1,130 @@ +package pool + +import ( + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestRemoveVMClassifiesOutcome(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + + eng.vms["survivor"] = "/run/survivor.sock" + eng.removeErrFor = "survivor" + if m.removeVM(t.Context(), "survivor") { + t.Fatal("running VM reported gone") + } + + delete(eng.vms, "survivor") + if !m.removeOrRetry(t.Context(), "survivor", "", "") { + t.Fatal("absent VM reported present") + } + m.mu.Lock() + pending := len(m.pendingRemovals) + m.mu.Unlock() + if pending != 0 { + t.Fatalf("confirmed-gone VM queued for retry: %d", pending) + } + + eng.vms["ordinary"] = "/run/ordinary.sock" + eng.removeErrFor = "" + if !m.removeVM(t.Context(), "ordinary") || !eng.removed("ordinary") { + t.Fatal("ordinary removal failed") + } +} + +func TestRemovalRetryConvergesConfirmedSurvivor(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + eng.vms["survivor"] = "/run/survivor.sock" + eng.removeErrFor = "survivor" + + m.destroy(t.Context(), "survivor") + m.mu.Lock() + _, pending := m.pendingRemovals["survivor"] + m.mu.Unlock() + if !pending { + t.Fatal("surviving VM was not retained for retry") + } + + // A failed retry re-queues itself, so the next tick tries again. + m.retryRemovals(t.Context()).Wait() + m.mu.Lock() + _, pending = m.pendingRemovals["survivor"] + m.mu.Unlock() + if !pending { + t.Fatal("failed retry dropped the survivor from the queue") + } + + eng.mu.Lock() + eng.removeErrFor = "" + eng.mu.Unlock() + m.retryRemovals(t.Context()).Wait() + m.mu.Lock() + _, pending = m.pendingRemovals["survivor"] + m.mu.Unlock() + if pending || !eng.removed("survivor") { + t.Fatalf("retry did not converge: pending=%v removed=%v", pending, eng.removed("survivor")) + } +} + +func TestRemovalRetryFinishesEgressCleanup(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + m.lockEgress = true + eng.vms["survivor"] = "/run/survivor.sock" + eng.removeErrFor = "survivor" + m.egressTaps["sb_survivor"] = "tap-survivor" + + removed := m.removeOrRetry(t.Context(), "survivor", "sb_survivor", "") + m.disarmEgress("sb_survivor", removed) + if removed { + t.Fatal("surviving VM reported gone") + } + + eng.mu.Lock() + eng.removeErrFor = "" + eng.mu.Unlock() + m.retryRemovals(t.Context()).Wait() + m.mu.Lock() + _, pending := m.pendingRemovals["survivor"] + _, locked := m.egressTaps["sb_survivor"] + m.mu.Unlock() + if pending || locked || !eng.removed("survivor") { + t.Fatalf("egress cleanup incomplete: pending=%v locked=%v removed=%v", + pending, locked, eng.removed("survivor")) + } +} + +func TestStaleRemovalRetryKeepsTapForCleanup(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + eng.vms["sbx-stale"] = "/run/stale.sock" + eng.removeErrFor = "sbx-stale" + live := map[string]types.VMRecord{ + "sbx-stale": { + Config: types.VMConfig{Name: "sbx-stale"}, + NetworkConfigs: []types.VMNetConfig{{TAP: "tap-stale"}}, + }, + } + + m.sweepStaleVMs(t.Context(), live, nil) + m.mu.Lock() + tap := m.pendingRemovals["sbx-stale"].tap + m.mu.Unlock() + if tap != "tap-stale" { + t.Fatalf("pending tap %q, want tap-stale", tap) + } + + eng.mu.Lock() + eng.removeErrFor = "" + eng.mu.Unlock() + m.retryRemovals(t.Context()).Wait() + m.mu.Lock() + _, pending := m.pendingRemovals["sbx-stale"] + m.mu.Unlock() + if pending || !eng.removed("sbx-stale") { + t.Fatalf("stale retry did not converge: pending=%v removed=%v", pending, eng.removed("sbx-stale")) + } +} diff --git a/sandboxd/pool/stats.go b/sandboxd/pool/stats.go index d10a883..e5c67ff 100644 --- a/sandboxd/pool/stats.go +++ b/sandboxd/pool/stats.go @@ -4,12 +4,9 @@ import ( "context" "os" "path/filepath" - "slices" "strconv" "strings" "time" - - "github.com/cocoonstack/sandbox/sandboxd/types" ) // SandboxStats is one sandbox's resource usage. CPUCount and MemTotalBytes come @@ -63,15 +60,11 @@ func (m *Manager) vmResidentBytes(ctx context.Context, vmName string) (int64, bo if vmName == "" { return 0, false } - vms, err := m.eng.List(ctx, vmName) - if err != nil { - return 0, false - } - i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == vmName }) - if i < 0 || vms[i].PID <= 0 { + vm, ok, err := m.findVM(ctx, vmName) + if err != nil || !ok || vm.PID <= 0 { return 0, false } - return residentBytes(vms[i].PID) + return residentBytes(vm.PID) } // residentBytes reads a process's resident page count from statm's second field. diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 27a71f5..bd81e45 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -120,6 +120,10 @@ type InfoResponse struct { Archived int `json:"archived"` Draining bool `json:"draining,omitempty"` Peers []string `json:"peers,omitempty"` + // AtCapacity marks refill parked because the node refused another VM, so a + // placer reads a short warm count as "full", not "still filling". + AtCapacity bool `json:"at_capacity,omitempty"` + AtCapacityReason string `json:"at_capacity_reason,omitempty"` } // PoolUpdateRequest is the wire body of PUT /v1/pools. It replaces the node's @@ -543,7 +547,10 @@ func (s *Server) handlePeers(w http.ResponseWriter, _ *http.Request) { func (s *Server) handleInfo(w http.ResponseWriter, _ *http.Request) { pools, g := s.mgr.Info() - resp := InfoResponse{Pools: pools, Claimed: g.Claimed, Hibernated: g.Hibernated, Archived: g.Archived, Draining: g.Draining} + resp := InfoResponse{ + Pools: pools, Claimed: g.Claimed, Hibernated: g.Hibernated, Archived: g.Archived, + Draining: g.Draining, AtCapacity: g.AtCapacity, AtCapacityReason: g.AtCapacityReason, + } if s.placer != nil { resp.Peers = s.placer.PeerAddrs() }