a73x

eba17d0a

feat(agent): admit VMs against host resources through one serialized gate

a73x   2026-07-25 20:17

Commit message
feat(agent): admit VMs against host resources through one serialized gate
cmd/eitri-agent/main.go
Old New
@@ -219,6 +219,13 @@ func runAgent(st *state.Store, cfg agentConfig) {
219 MaxDiskGB: cfg.MaxDiskGB, 219 MaxDiskGB: cfg.MaxDiskGB,
220 } 220 }
221 221
222 // Seed the admission ledger from persisted records so the first reconcile
223 // accounts for VMs that survived the agent restart (their compute must
224 // count against the caps before any new create is admitted).
225 if recs, err := st.LoadVMs(); err == nil {
226 engine.SeedLedger(recs)
227 }
228
222 client := &syncclient.Client{ 229 client := &syncclient.Client{
223 Engine: engine, 230 Engine: engine,
224 St: st, 231 St: st,
docs/shape.html
Old New
@@ -126,7 +126,9 @@
126 "importPath": "internal/agent/dhcp", 126 "importPath": "internal/agent/dhcp",
127 "plane": "data", 127 "plane": "data",
128 "synopsis": "Package dhcp is an in-process, reservation-only DHCPv4 responder for the eitri bridge.", 128 "synopsis": "Package dhcp is an in-process, reservation-only DHCPv4 responder for the eitri bridge.",
129 "imports": [] 129 "imports": [
130 "internal/agent/ipalloc"
131 ]
130 }, 132 },
131 { 133 {
132 "importPath": "internal/agent/enrollclient", 134 "importPath": "internal/agent/enrollclient",
@@ -170,7 +172,6 @@
170 "imports": [ 172 "imports": [
171 "internal/agent/dhcp", 173 "internal/agent/dhcp",
172 "internal/agent/exec", 174 "internal/agent/exec",
173 "internal/agent/ipalloc",
174 "internal/agent/state" 175 "internal/agent/state"
175 ] 176 ]
176 }, 177 },
docs/shape.json
Old New
@@ -75,7 +75,9 @@
75 "importPath": "internal/agent/dhcp", 75 "importPath": "internal/agent/dhcp",
76 "plane": "data", 76 "plane": "data",
77 "synopsis": "Package dhcp is an in-process, reservation-only DHCPv4 responder for the eitri bridge.", 77 "synopsis": "Package dhcp is an in-process, reservation-only DHCPv4 responder for the eitri bridge.",
78 "imports": [] 78 "imports": [
79 "internal/agent/ipalloc"
80 ]
79 }, 81 },
80 { 82 {
81 "importPath": "internal/agent/enrollclient", 83 "importPath": "internal/agent/enrollclient",
@@ -119,7 +121,6 @@
119 "imports": [ 121 "imports": [
120 "internal/agent/dhcp", 122 "internal/agent/dhcp",
121 "internal/agent/exec", 123 "internal/agent/exec",
122 "internal/agent/ipalloc",
123 "internal/agent/state" 124 "internal/agent/state"
124 ] 125 ]
125 }, 126 },
internal/agent/dhcp/dhcp.go
Old New
@@ -14,11 +14,14 @@ import (
14 14
15 "github.com/insomniacslk/dhcp/dhcpv4" 15 "github.com/insomniacslk/dhcp/dhcpv4"
16 "github.com/insomniacslk/dhcp/dhcpv4/server4" 16 "github.com/insomniacslk/dhcp/dhcpv4/server4"
17
18 "github.com/a73x/eitri/internal/agent/ipalloc"
17 ) 19 )
18 20
19 // Server serves reserved DHCPv4 leases on a single interface. 21 // Server serves reserved DHCPv4 leases on a single interface.
20 type Server struct { 22 type Server struct {
21 iface string 23 iface string
24 cidr string // host bridge CIDR, e.g. "10.77.1.0/24"; the pool Reserve draws from
22 gateway net.IP 25 gateway net.IP
23 mask net.IPMask 26 mask net.IPMask
24 dns []net.IP 27 dns []net.IP
@@ -31,9 +34,10 @@ type Server struct {
31 // NewServer builds a Server. gateway is the host's address on the bridge (also 34 // NewServer builds a Server. gateway is the host's address on the bridge (also
32 // the DHCP router and server-identifier); mask is the bridge subnet mask; dns 35 // the DHCP router and server-identifier); mask is the bridge subnet mask; dns
33 // is handed to guests; lease is the offered lease time. 36 // is handed to guests; lease is the offered lease time.
34 func NewServer(iface string, gateway net.IP, mask net.IPMask, dns []net.IP, lease time.Duration) *Server { 37 func NewServer(iface, cidr string, gateway net.IP, mask net.IPMask, dns []net.IP, lease time.Duration) *Server {
35 return &Server{ 38 return &Server{
36 iface: iface, 39 iface: iface,
40 cidr: cidr,
37 gateway: gateway, 41 gateway: gateway,
38 mask: mask, 42 mask: mask,
39 dns: dns, 43 dns: dns,
@@ -59,6 +63,31 @@ func (s *Server) RemoveReservation(mac net.HardwareAddr) {
59 delete(s.res, key(mac)) 63 delete(s.res, key(mac))
60 } 64 }
61 65
66 // Reserve returns a sticky IP for mac, allocating one from the host CIDR on
67 // first call and recording it in the reservation table (which doubles as the
68 // used-address set). A MAC that already holds a reservation gets it back
69 // unchanged, so a VM keeps its address across reboots and agent restarts.
70 // Serialized by the same lock as the table, so concurrent callers never collide
71 // on an address. Errors when the CIDR is exhausted.
72 func (s *Server) Reserve(mac net.HardwareAddr) (net.IP, error) {
73 s.mu.Lock()
74 defer s.mu.Unlock()
75 if ip, ok := s.res[key(mac)]; ok {
76 return ip, nil // sticky: already allocated
77 }
78 used := make([]string, 0, len(s.res))
79 for _, ip := range s.res {
80 used = append(used, ip.String())
81 }
82 ipStr, err := ipalloc.Alloc(s.cidr, used)
83 if err != nil {
84 return nil, err
85 }
86 ip := net.ParseIP(ipStr)
87 s.res[key(mac)] = ip
88 return ip, nil
89 }
90
62 // lookup returns the reserved IP for mac. 91 // lookup returns the reserved IP for mac.
63 func (s *Server) lookup(mac net.HardwareAddr) (net.IP, bool) { 92 func (s *Server) lookup(mac net.HardwareAddr) (net.IP, bool) {
64 s.mu.RLock() 93 s.mu.RLock()
internal/agent/dhcp/dhcp_test.go
Old New
@@ -6,16 +6,52 @@ import (
6 "time" 6 "time"
7 7
8 "github.com/insomniacslk/dhcp/dhcpv4" 8 "github.com/insomniacslk/dhcp/dhcpv4"
9 "github.com/stretchr/testify/assert"
10 "github.com/stretchr/testify/require"
9 ) 11 )
10 12
11 func testServer() *Server { 13 func testServer() *Server {
12 return NewServer("eitri0", 14 return NewServer("eitri0", "10.77.1.0/24",
13 net.IPv4(10, 77, 1, 1), 15 net.IPv4(10, 77, 1, 1),
14 net.CIDRMask(24, 32), 16 net.CIDRMask(24, 32),
15 []net.IP{net.IPv4(1, 1, 1, 1)}, 17 []net.IP{net.IPv4(1, 1, 1, 1)},
16 12*time.Hour) 18 12*time.Hour)
17 } 19 }
18 20
21 func newTestServer() *Server {
22 return NewServer("eitri0", "10.77.1.0/24",
23 net.IPv4(10, 77, 1, 1), net.CIDRMask(24, 32), nil, time.Hour)
24 }
25
26 func TestReserveAllocatesDistinctStickyAddresses(t *testing.T) {
27 s := newTestServer()
28 macA, _ := net.ParseMAC("52:54:00:aa:aa:aa")
29 macB, _ := net.ParseMAC("52:54:00:bb:bb:bb")
30
31 ipA, err := s.Reserve(macA)
32 require.NoError(t, err)
33 assert.Equal(t, "10.77.1.2", ipA.String(), "first free skips network(.0) + gateway(.1)")
34
35 ipB, err := s.Reserve(macB)
36 require.NoError(t, err)
37 assert.Equal(t, "10.77.1.3", ipB.String(), "second VM gets a distinct address")
38
39 ipA2, err := s.Reserve(macA)
40 require.NoError(t, err)
41 assert.Equal(t, ipA.String(), ipA2.String(), "re-reserving a MAC is sticky")
42 }
43
44 func TestReserveSkipsPreexistingReservation(t *testing.T) {
45 s := newTestServer()
46 pinned, _ := net.ParseMAC("52:54:00:cc:cc:cc")
47 s.SetReservation(pinned, net.ParseIP("10.77.1.2"))
48
49 fresh, _ := net.ParseMAC("52:54:00:dd:dd:dd")
50 ip, err := s.Reserve(fresh)
51 require.NoError(t, err)
52 assert.Equal(t, "10.77.1.3", ip.String(), "an externally-set reservation is treated as used")
53 }
54
19 func mustMAC(t *testing.T, s string) net.HardwareAddr { 55 func mustMAC(t *testing.T, s string) net.HardwareAddr {
20 t.Helper() 56 t.Helper()
21 m, err := net.ParseMAC(s) 57 m, err := net.ParseMAC(s)
internal/agent/netenv/netenv.go
Old New
@@ -17,7 +17,6 @@ import (
17 17
18 "github.com/a73x/eitri/internal/agent/dhcp" 18 "github.com/a73x/eitri/internal/agent/dhcp"
19 "github.com/a73x/eitri/internal/agent/exec" 19 "github.com/a73x/eitri/internal/agent/exec"
20 "github.com/a73x/eitri/internal/agent/ipalloc"
21 "github.com/a73x/eitri/internal/agent/state" 20 "github.com/a73x/eitri/internal/agent/state"
22 ) 21 )
23 22
@@ -52,7 +51,7 @@ func New(run exec.Runner, cidr string) (*Net, error) {
52 n := &Net{run: run, cidr: p, isTap: sysfsIsTap} 51 n := &Net{run: run, cidr: p, isTap: sysfsIsTap}
53 gw := net.ParseIP(n.Gateway()) 52 gw := net.ParseIP(n.Gateway())
54 mask := net.CIDRMask(p.Bits(), 32) 53 mask := net.CIDRMask(p.Bits(), 32)
55 n.dhcp = dhcp.NewServer(Bridge, gw, mask, guestDNS, 12*time.Hour) 54 n.dhcp = dhcp.NewServer(Bridge, n.cidr.String(), gw, mask, guestDNS, 12*time.Hour)
56 return n, nil 55 return n, nil
57 } 56 }
58 57
@@ -88,14 +87,23 @@ func (e tapConflictError) Permanent() bool { return true }
88 // Gateway returns the host-side IP (.1) on the bridge, as a bare address string. 87 // Gateway returns the host-side IP (.1) on the bridge, as a bare address string.
89 func (n *Net) Gateway() string { return n.cidr.Masked().Addr().Next().String() } 88 func (n *Net) Gateway() string { return n.cidr.Masked().Addr().Next().String() }
90 89
91 // AllocateIP returns an unused VM IP within the bridge network. It implements 90 // ReserveIP returns a sticky IP for vmID, allocated from the bridge CIDR and
92 // the reconcile NetEnv addressing seam with host-local allocation; a future 91 // recorded as the guest's DHCP reservation (keyed by the VM's deterministic
93 // central/per-network allocator replaces this method, not the reconcile loop. 92 // MAC). Idempotent: the same VM gets the same address across reboots and agent
94 func (n *Net) AllocateIP(_ context.Context, used []string) (string, error) { 93 // restarts. This is the reconcile addressing seam — the DHCP server owns the
95 return ipalloc.Alloc(n.cidr.String(), used) 94 // used-address set, so no caller-supplied used-set is needed.
95 func (n *Net) ReserveIP(vmID string) (string, error) {
96 mac, err := net.ParseMAC(state.MAC(vmID))
97 if err != nil {
98 return "", err
99 }
100 ip, err := n.dhcp.Reserve(mac)
101 if err != nil {
102 return "", err
103 }
104 return ip.String(), nil
96 } 105 }
97 106
98
99 // tolerated reports whether err carries one of the given substrings in either 107 // tolerated reports whether err carries one of the given substrings in either
100 // the command stdout or the error message. iproute2 puts the same condition in 108 // the command stdout or the error message. iproute2 puts the same condition in
101 // different streams across versions, so both are checked. 109 // different streams across versions, so both are checked.
internal/agent/netenv/netenv_test.go
Old New
@@ -364,6 +364,25 @@ func TestCreateTapAddsReservationAndTapCommands(t *testing.T) {
364 assert.Equal(t, "10.77.1.7", ip.String()) 364 assert.Equal(t, "10.77.1.7", ip.String())
365 } 365 }
366 366
367 func TestReserveIPAllocatesAndRecordsReservation(t *testing.T) {
368 noop := func(_ context.Context, _ string, _ ...string) (string, error) { return "", nil }
369 n, err := New(noop, "10.77.1.0/24")
370 require.NoError(t, err)
371
372 ip, err := n.ReserveIP("vm-alpha")
373 require.NoError(t, err)
374 assert.Equal(t, "10.77.1.2", ip)
375
376 mac, _ := net.ParseMAC(stateMAC("vm-alpha"))
377 got, ok := n.dhcp.LookupForTest(mac)
378 require.True(t, ok, "ReserveIP must record the DHCP reservation")
379 assert.Equal(t, "10.77.1.2", got.String())
380
381 again, err := n.ReserveIP("vm-alpha")
382 require.NoError(t, err)
383 assert.Equal(t, ip, again, "ReserveIP is sticky per VM")
384 }
385
367 func TestDeleteTapRemovesReservationAndTap(t *testing.T) { 386 func TestDeleteTapRemovesReservationAndTap(t *testing.T) {
368 errs := map[string]error{ 387 errs := map[string]error{
369 "ip link show dev eit-vm-abc12": errors.New("does not exist"), 388 "ip link show dev eit-vm-abc12": errors.New("does not exist"),
internal/agent/reconcile/reconcile.go
Old New
@@ -26,6 +26,7 @@ import (
26 "errors" 26 "errors"
27 "fmt" 27 "fmt"
28 "strings" 28 "strings"
29 "sync"
29 "time" 30 "time"
30 31
31 "github.com/a73x/eitri/internal/agent/seed" 32 "github.com/a73x/eitri/internal/agent/seed"
@@ -46,11 +47,10 @@ type Provisioner interface {
46 type NetEnv interface { 47 type NetEnv interface {
47 CreateTap(ctx context.Context, vmID, ip string) error 48 CreateTap(ctx context.Context, vmID, ip string) error
48 DeleteTap(ctx context.Context, vmID string) error 49 DeleteTap(ctx context.Context, vmID string) error
49 // AllocateIP returns an unused VM IP for this host's network. `used` lists 50 // ReserveIP returns a sticky IP for vmID from this host's network,
50 // addresses already taken. Kept behind the seam so a future central or 51 // recording its DHCP reservation. The host networking layer owns the
51 // per-network allocator can replace host-local allocation without touching 52 // used-address set, so allocation needs no caller-supplied used-set.
52 // the reconcile loop. 53 ReserveIP(vmID string) (string, error)
53 AllocateIP(ctx context.Context, used []string) (string, error)
54 } 54 }
55 55
56 // Engine is the reconcile loop. All fields must be set before calling Step. 56 // Engine is the reconcile loop. All fields must be set before calling Step.
@@ -73,6 +73,16 @@ type Engine struct {
73 // Now returns the current time. Injectable for deterministic tests. 73 // Now returns the current time. Injectable for deterministic tests.
74 Now func() time.Time 74 Now func() time.Time
75 75
76 // mu guards committed. It serializes admission so concurrent per-VM callers
77 // cannot oversubscribe a cap or double-count. Under the current
78 // single-threaded step it is uncontended.
79 mu sync.Mutex
80 // committed is the in-memory admission ledger: vm_id -> the spec whose
81 // compute counts against the host caps. A VM is committed at create,
82 // released at quarantine (compute frees while the guest is stopped), and
83 // re-noted while it exists. Rebuilt from persisted records via SeedLedger.
84 committed map[string]state.VMSpec
85
76 // TombstoneGrace is the quarantine period for tombstoned VMs before destroy. 86 // TombstoneGrace is the quarantine period for tombstoned VMs before destroy.
77 TombstoneGrace time.Duration 87 TombstoneGrace time.Duration
78 88
@@ -84,7 +94,7 @@ type Engine struct {
84 94
85 // MaxVCPUs, MaxMemMB, MaxDiskGB cap the host resources this agent will 95 // MaxVCPUs, MaxMemMB, MaxDiskGB cap the host resources this agent will
86 // commit to live VMs (0 = unlimited). A create whose resources would push 96 // commit to live VMs (0 = unlimited). A create whose resources would push
87 // the running total past a cap is refused — see quotaBlock. This is the 97 // the running total past a cap is refused — see admit. This is the
88 // enforced half of agent-side quotas; syncclient advertises the same caps. 98 // enforced half of agent-side quotas; syncclient advertises the same caps.
89 MaxVCPUs int64 99 MaxVCPUs int64
90 MaxMemMB int64 100 MaxMemMB int64
@@ -184,6 +194,11 @@ func (e *Engine) Step(ctx context.Context, snap *pb.DesiredStateSnapshot) *pb.Ac
184 continue // active desired VM; handled in converge pass 194 continue // active desired VM; handled in converge pass
185 } 195 }
186 196
197 // Being reaped (absent from desired or tombstoned): free its compute
198 // from the admission ledger so a new VM can use it. Idempotent; the IP
199 // reservation is released on destroy by DeleteTap.
200 e.releaseCompute(id)
201
187 now := e.Now() 202 now := e.Now()
188 203
189 if rec.QuarantinedAt == nil { 204 if rec.QuarantinedAt == nil {
@@ -284,29 +299,79 @@ func (e *Engine) graceFor(rec state.Record) time.Duration {
284 return e.VanishGrace 299 return e.VanishGrace
285 } 300 }
286 301
287 // quotaBlock returns a non-empty reason when booting d would exceed a configured 302 // admit is the single serialized admission gate: it checks compute quota and,
288 // host resource cap, else "". It sums the resources of the currently-committed 303 // if admitted, reserves the VM's IP. Returns quotaMsg non-empty for a
289 // local VMs — excluding quarantined records (being torn down; their guests are 304 // NON-TERMINAL quota refusal (nothing is committed, no IP reserved). Otherwise
290 // already stopped) and d's own record (so a retry or spec-edit of an existing 305 // it commits the VM's compute to the ledger BEFORE reserving the IP, so an IP
291 // VM does not double-count itself) — and adds d's request. The binding 306 // reservation failure still counts this VM against a same-step sibling's cap
292 // dimension is named in the message so the operator sees which cap was hit. 307 // (parity with the old failCreate publish) — err carries the reservation error.
293 func (e *Engine) quotaBlock(d *pb.VMDesired, recs map[string]state.Record) string { 308 func (e *Engine) admit(vmID string, spec state.VMSpec) (ip, quotaMsg string, err error) {
309 e.mu.Lock()
310 defer e.mu.Unlock()
311 if e.committed == nil {
312 e.committed = map[string]state.VMSpec{}
313 }
314 if msg := e.quotaCheckLocked(vmID, spec); msg != "" {
315 return "", msg, nil
316 }
317 e.committed[vmID] = spec
318 ip, err = e.Net.ReserveIP(vmID)
319 return ip, "", err
320 }
321
322 // note commits spec's compute for an existing VM without a quota check. Used for
323 // VMs already created (converge, un-delete re-adoption, startup rebuild) so the
324 // ledger always reflects every live VM regardless of how it got there.
325 func (e *Engine) note(vmID string, spec state.VMSpec) {
326 e.mu.Lock()
327 defer e.mu.Unlock()
328 if e.committed == nil {
329 e.committed = map[string]state.VMSpec{}
330 }
331 e.committed[vmID] = spec
332 }
333
334 // releaseCompute frees a VM's compute from the ledger (at quarantine). The IP
335 // reservation is released separately, on destroy, by DeleteTap.
336 func (e *Engine) releaseCompute(vmID string) {
337 e.mu.Lock()
338 defer e.mu.Unlock()
339 delete(e.committed, vmID)
340 }
341
342 // SeedLedger rebuilds the compute ledger from persisted records at startup so
343 // the first step accounts for every surviving VM. Quarantined records are
344 // excluded (their guests are stopped; their compute is free).
345 func (e *Engine) SeedLedger(recs map[string]state.Record) {
346 e.mu.Lock()
347 defer e.mu.Unlock()
348 if e.committed == nil {
349 e.committed = map[string]state.VMSpec{}
350 }
351 for id, rec := range recs {
352 if rec.QuarantinedAt == nil {
353 e.committed[id] = rec.Spec
354 }
355 }
356 }
357
358 // quotaCheckLocked returns a non-empty reason when booting spec would exceed a
359 // configured host cap, else "". Caller holds e.mu. It sums the committed specs
360 // (excluding vmID itself, so a retry does not double-count) and adds spec's
361 // request. Quarantined VMs are absent from committed, so they are excluded for
362 // free. The binding dimension is named so the operator sees which cap was hit.
363 func (e *Engine) quotaCheckLocked(vmID string, spec state.VMSpec) string {
294 if e.MaxVCPUs == 0 && e.MaxMemMB == 0 && e.MaxDiskGB == 0 { 364 if e.MaxVCPUs == 0 && e.MaxMemMB == 0 && e.MaxDiskGB == 0 {
295 return "" // no caps configured — unlimited 365 return "" // no caps configured — unlimited
296 } 366 }
297 spec := specFromDesired(d)
298 vcpus, mem, disk := spec.VCPUs, spec.MemMB, spec.DiskGB 367 vcpus, mem, disk := spec.VCPUs, spec.MemMB, spec.DiskGB
299 // recs is the tick's live record map (loaded once post-reap in Step, kept 368 for id, s := range e.committed {
300 // intra-tick-consistent as each create commits its record). Reading it — 369 if id == vmID {
301 // rather than re-LoadVMs here — is what lets a second create in the same tick
302 // see the first's committed resources and trip the cap.
303 for _, r := range recs {
304 if r.QuarantinedAt != nil || r.Spec.VMID == d.VmId {
305 continue 370 continue
306 } 371 }
307 vcpus += r.Spec.VCPUs 372 vcpus += s.VCPUs
308 mem += r.Spec.MemMB 373 mem += s.MemMB
309 disk += r.Spec.DiskGB 374 disk += s.DiskGB
310 } 375 }
311 switch { 376 switch {
312 case e.MaxVCPUs > 0 && vcpus > e.MaxVCPUs: 377 case e.MaxVCPUs > 0 && vcpus > e.MaxVCPUs:
@@ -319,18 +384,10 @@ func (e *Engine) quotaBlock(d *pb.VMDesired, recs map[string]state.Record) strin
319 return "" 384 return ""
320 } 385 }
321 386
322 // create attempts to create a new VM from desired state d. 387 // create attempts to create a new VM from desired state d. recs is the tick's
323 // recs is the tick's live record map (loaded once in Step); create looks up 388 // record map; create reads recs[d.VmId] for this VM's own prior record (retry
324 // d.VmId in it to get the existing (potentially stale) record, if any. create 389 // budget, existing IP). Quota and IP now come from the serialized admission
325 // also reads recs for the quota sum and the IP used-set INSTEAD of 390 // ledger (see admit), not from recs.
326 // re-LoadVMs-ing, and — crucially —
327 // writes the new record back into it (recs[d.VmId]=rec) the moment that record's
328 // Spec+IP are durably committed, so a later create in the same Step's converge
329 // loop observes this VM's committed IP and resources. This reproduces exactly
330 // what the old per-create LoadVMs did: pre-change it re-read disk and saw the
331 // earlier create's just-saved record; now it reads the same fact from the shared
332 // map. Step passes the same map reference to every create, so the write is
333 // visible to subsequent calls.
334 func (e *Engine) create(ctx context.Context, d *pb.VMDesired, rep *pb.ActualStateReport, recs map[string]state.Record) { 391 func (e *Engine) create(ctx context.Context, d *pb.VMDesired, rep *pb.ActualStateReport, recs map[string]state.Record) {
335 rec, ok := recs[d.VmId] 392 rec, ok := recs[d.VmId]
336 393
@@ -360,63 +417,31 @@ func (e *Engine) create(ctx context.Context, d *pb.VMDesired, rep *pb.ActualStat
360 return 417 return
361 } 418 }
362 419
363 // Quota gate: refuse to boot a VM that would push this host's committed 420 // Serialized admission: quota then a sticky IP, under one lock. A quota
364 // resources past a configured cap. NON-TERMINAL by design — it returns 421 // refusal is NON-TERMINAL — it returns before touching CreateAttempts, so
365 // before touching CreateAttempts or any state, so once a running VM is 422 // once room frees the next tick retries and boots. An IP reservation
366 // removed and room frees, the next level-triggered tick retries and boots. 423 // failure DOES spend an attempt (via failCreate); admit has already
367 // A newly-lowered cap never kills a running guest; only new boots are gated. 424 // committed this VM's compute, so a same-tick sibling counts it.
368 if msg := e.quotaBlock(d, recs); msg != "" { 425 rec.Spec = spec
369 addReport(rep, d.VmId, rec.IP, "stopped", "failed", msg) 426 ip, quotaMsg, err := e.admit(d.VmId, spec)
427 if quotaMsg != "" {
428 addReport(rep, d.VmId, rec.IP, "stopped", "failed", quotaMsg)
370 return 429 return
371 } 430 }
372
373 // Build the spec from desired.
374 rec.Spec = spec
375 rec.CreateAttempts++ 431 rec.CreateAttempts++
376 rec.CreatedAt = e.Now() 432 rec.CreatedAt = e.Now()
377 rec.LastError = "" // clear for this attempt 433 rec.LastError = "" // clear for this attempt
378 434 if err != nil {
379 // Allocate an IP if we don't have one yet. The used-set is built from the 435 e.failCreate(ctx, rec, err, rep)
380 // tick's live record map (kept intra-tick-consistent below), so two VMs 436 return
381 // created in one Step never collide on an address.
382 if rec.IP == "" {
383 used := make([]string, 0, len(recs))
384 for _, r := range recs {
385 if r.IP != "" {
386 used = append(used, r.IP)
387 }
388 }
389 ip, err := e.Net.AllocateIP(ctx, used)
390 if err != nil {
391 // failCreate durably saves this record (Spec set, IP still empty).
392 // Publish it so a same-tick sibling's quota check counts its Spec —
393 // matching the old per-create LoadVMs, which re-read the failCreate
394 // save. Tick-scoped: conservative inclusion only over-counts for one
395 // tick and self-heals on the next fresh load.
396 recs[d.VmId] = rec
397 e.failCreate(ctx, rec, err, rep)
398 return
399 }
400 rec.IP = ip
401 } 437 }
438 rec.IP = ip
402 439
403 // Record BEFORE side effects so a crash is recoverable. 440 // Record BEFORE side effects so a crash is recoverable.
404 if err := e.St.SaveVM(rec); err != nil { 441 if err := e.St.SaveVM(rec); err != nil {
405 // This save failed but failCreate's save may succeed, durably committing
406 // the allocated IP. Publish so a same-tick sibling excludes that IP from
407 // its used-set and counts its Spec — preventing a double-allocation the
408 // old per-create LoadVMs could not produce.
409 recs[d.VmId] = rec
410 e.failCreate(ctx, rec, err, rep) 442 e.failCreate(ctx, rec, err, rep)
411 return 443 return
412 } 444 }
413 // Intra-tick consistency: this record's Spec+IP are now durably committed.
414 // Publish it into the shared tick map so a later create in this same Step
415 // sees it — its quotaBlock sums this VM's resources and its IP used-set
416 // excludes this VM's address. This mirrors the old per-create LoadVMs, which
417 // re-read exactly this just-saved record. Only Spec/IP matter to quota/IP;
418 // the later BootID/StopRequested save does not change them, so we update here.
419 recs[d.VmId] = rec
420 445
421 // Resolve base image. 446 // Resolve base image.
422 basePath, err := e.Images(ctx, d.ImageUrl, d.ImageSha256) 447 basePath, err := e.Images(ctx, d.ImageUrl, d.ImageSha256)
@@ -521,6 +546,10 @@ func (e *Engine) failConverge(rec state.Record, err error, rep *pb.ActualStateRe
521 // converge drives an existing VM toward its desired power state, 546 // converge drives an existing VM toward its desired power state,
522 // handling lost detection and restart logic. 547 // handling lost detection and restart logic.
523 func (e *Engine) converge(ctx context.Context, d *pb.VMDesired, rec state.Record, rep *pb.ActualStateReport) { 548 func (e *Engine) converge(ctx context.Context, d *pb.VMDesired, rec state.Record, rep *pb.ActualStateReport) {
549 // Keep the ledger reflecting this live VM (covers the un-delete case, where a
550 // quarantined VM returns to desired after its compute was released).
551 e.note(d.VmId, rec.Spec)
552
524 running := e.Prov.Running(d.VmId) 553 running := e.Prov.Running(d.VmId)
525 bootID := e.BootID() 554 bootID := e.BootID()
526 555
internal/agent/reconcile/reconcile_test.go
Old New
@@ -62,10 +62,11 @@ func (f *fakeProv) Kill(_ context.Context, id string) error {
62 func (f *fakeProv) Running(id string) bool { return f.running[id] } 62 func (f *fakeProv) Running(id string) bool { return f.running[id] }
63 63
64 type fakeNet struct { 64 type fakeNet struct {
65 taps []string 65 taps []string
66 deleted []string 66 deleted []string
67 cidr string 67 cidr string
68 allocErr error // one-shot: consumed and cleared on the first AllocateIP call 68 reserved map[string]string // vmID -> ip (sticky, mirrors dhcp)
69 reserveErr error // one-shot: consumed and cleared on first ReserveIP
69 } 70 }
70 71
71 func (f *fakeNet) CreateTap(_ context.Context, vmID, ip string) error { 72 func (f *fakeNet) CreateTap(_ context.Context, vmID, ip string) error {
@@ -77,15 +78,31 @@ func (f *fakeNet) DeleteTap(_ context.Context, vmID string) error {
77 return nil 78 return nil
78 } 79 }
79 80
80 // AllocateIP mirrors netenv's host-local behavior so reconcile tests exercise 81 // ReserveIP mirrors netenv/dhcp: sticky per VM, allocating distinct addresses
81 // identical addressing through the seam. 82 // from the CIDR over the set of already-reserved ones, so reconcile tests
82 func (f *fakeNet) AllocateIP(_ context.Context, used []string) (string, error) { 83 // exercise identical addressing through the seam.
83 if f.allocErr != nil { 84 func (f *fakeNet) ReserveIP(vmID string) (string, error) {
84 err := f.allocErr 85 if f.reserveErr != nil {
85 f.allocErr = nil // one-shot 86 err := f.reserveErr
87 f.reserveErr = nil // one-shot
86 return "", err 88 return "", err
87 } 89 }
88 return ipalloc.Alloc(f.cidr, used) 90 if f.reserved == nil {
91 f.reserved = map[string]string{}
92 }
93 if ip, ok := f.reserved[vmID]; ok {
94 return ip, nil // sticky
95 }
96 used := make([]string, 0, len(f.reserved))
97 for _, ip := range f.reserved {
98 used = append(used, ip)
99 }
100 ip, err := ipalloc.Alloc(f.cidr, used)
101 if err != nil {
102 return "", err
103 }
104 f.reserved[vmID] = ip
105 return ip, nil
89 } 106 }
90 107
91 type fixture struct { 108 type fixture struct {
@@ -220,7 +237,7 @@ func TestCreateSurvivesRecordLoadFailure(t *testing.T) {
220 func TestSameTickSiblingCountsFailedCreateAgainstQuota(t *testing.T) { 237 func TestSameTickSiblingCountsFailedCreateAgainstQuota(t *testing.T) {
221 f := setup(t) 238 f := setup(t)
222 f.eng.MaxVCPUs = 3 239 f.eng.MaxVCPUs = 3
223 f.net.allocErr = assert.AnError 240 f.net.reserveErr = assert.AnError
224 twoVCPU := func(v *pb.VMDesired) { v.Vcpus = 2 } 241 twoVCPU := func(v *pb.VMDesired) { v.Vcpus = 2 }
225 242
226 rep := f.eng.Step(context.Background(), snap(1, vm("vm1", twoVCPU), vm("vm2", twoVCPU))) 243 rep := f.eng.Step(context.Background(), snap(1, vm("vm1", twoVCPU), vm("vm2", twoVCPU)))
@@ -526,12 +543,10 @@ func TestPermanentCreateErrorFailsTerminallyInOneAttempt(t *testing.T) {
526 assert.Equal(t, 1, f.prov.prepCalls, "no further provisioner attempts after a permanent failure") 543 assert.Equal(t, 1, f.prov.prepCalls, "no further provisioner attempts after a permanent failure")
527 } 544 }
528 545
529 // TestTwoVMsCreatedInOneStepGetDistinctIPs pins intra-tick IP consistency: 546 // TestTwoVMsCreatedInOneStepGetDistinctIPs pins that when a single Step creates
530 // when a single Step creates two VMs, the second create must SEE the first's 547 // two VMs, each gets a distinct address. Admission is serialized, so the second
531 // just-allocated IP in its used-set and pick a different one. A naive 548 // create's ReserveIP allocates over the reservation table the first just wrote
532 // implementation that threads the pre-create records map (loaded before any 549 // and picks a different address. The reconcile loop's converge order is
533 // create) into the IP used-set would hand both creates the same empty used-set
534 // and double-allocate the same address. The reconcile loop's converge order is
535 // randomized, so this must hold regardless of which VM is created first. 550 // randomized, so this must hold regardless of which VM is created first.
536 func TestTwoVMsCreatedInOneStepGetDistinctIPs(t *testing.T) { 551 func TestTwoVMsCreatedInOneStepGetDistinctIPs(t *testing.T) {
537 f := setup(t) 552 f := setup(t)
@@ -543,16 +558,15 @@ func TestTwoVMsCreatedInOneStepGetDistinctIPs(t *testing.T) {
543 assert.NotEmpty(t, recs["vm1"].IP) 558 assert.NotEmpty(t, recs["vm1"].IP)
544 assert.NotEmpty(t, recs["vm2"].IP) 559 assert.NotEmpty(t, recs["vm2"].IP)
545 assert.NotEqual(t, recs["vm1"].IP, recs["vm2"].IP, 560 assert.NotEqual(t, recs["vm1"].IP, recs["vm2"].IP,
546 "two VMs created in one Step must not share an IP (intra-tick used-set)") 561 "two VMs created in one Step must not share an IP (serialized admission)")
547 } 562 }
548 563
549 // TestSecondVMInOneStepBustingCapIsBlocked pins intra-tick quota consistency: 564 // TestSecondVMInOneStepBustingCapIsBlocked pins intra-tick quota consistency:
550 // two 2-vcpu VMs sum to 4 > the 3-vcpu cap, so exactly one may boot per tick. 565 // two 2-vcpu VMs sum to 4 > the 3-vcpu cap, so exactly one may boot per tick.
551 // The second create's quotaBlock must SEE the first's just-committed record so 566 // Admission is serialized, so the second create's quota check sees the first's
552 // the running total (2 + 2 = 4) trips the cap. A naive implementation feeding 567 // just-committed ledger entry and the running total (2 + 2 = 4) trips the cap.
553 // the pre-create records map to quotaBlock would count 0 committed for both and 568 // Converge order is randomized, so we assert on the count and identify the
554 // let both boot (over-commit). Converge order is randomized, so we assert on the 569 // blocked VM dynamically.
555 // count and identify the blocked VM dynamically.
556 func TestSecondVMInOneStepBustingCapIsBlocked(t *testing.T) { 570 func TestSecondVMInOneStepBustingCapIsBlocked(t *testing.T) {
557 f := setup(t) 571 f := setup(t)
558 f.eng.MaxVCPUs = 3 572 f.eng.MaxVCPUs = 3
internal/agent/syncclient/client_test.go
Old New
@@ -38,7 +38,7 @@ type noopNet struct{}
38 38
39 func (noopNet) CreateTap(context.Context, string, string) error { return nil } 39 func (noopNet) CreateTap(context.Context, string, string) error { return nil }
40 func (noopNet) DeleteTap(context.Context, string) error { return nil } 40 func (noopNet) DeleteTap(context.Context, string) error { return nil }
41 func (noopNet) AllocateIP(context.Context, []string) (string, error) { 41 func (noopNet) ReserveIP(string) (string, error) {
42 return "10.77.1.2", nil 42 return "10.77.1.2", nil
43 } 43 }
44 44