a73x

ef2f020d

fix(agent): traffic on a published port cannot starve fleet management

a73x   2026-08-09 16:42

Commit message
fix(agent): traffic on a published port cannot starve fleet management

The agent that proxies a guest's published ports is the same process that
runs the guests. Every proxied connection costs it two descriptors, the
caller's and the dial into the guest, and nothing bounded how many a
single exposure could hold: a few hundred callers holding one published
port open spent the process's whole descriptor budget, and what failed
next was not the proxy. Spawning cloud-hypervisor, opening a console,
carrying the sync tunnel — the fleet work sharing the process — began
failing for want of a descriptor the exposure had taken.

So each exposure serves at most 256 connections at once, counted on the
way in and released when the pipe ends. Past that the arriving connection
is closed immediately: a caller that is refused knows it, where a caller
parked in a backlog nobody is draining learns nothing, and neither does
whoever is trying to work out why the host stopped answering. There are
no idle deadlines to go with it — a connection held open to a real
service is exactly what a published port is for, and only the aggregate
was ever the problem.

An accept loop no longer dies of a descriptor shortage either. Handing
the listener back asks the next converge to find a descriptor for a fresh
one, which is precisely what the process has none of, so EMFILE and
ENFILE pause the loop instead and the port stays bound with its backlog
intact. Every other accept error is still a broken listener, still ends
the loop, and still heals by being rebound.

The shipped units state a descriptor budget rather than inheriting
whatever the distro chose. 65536 holds dozens of saturated published
ports with room left over for the agent's own work; the server takes the
same line, since what it holds open follows how many people are using the
fleet — a gate session is a pair of connections, an open console is a
stream that outlives the request that made it. An agent that upgrades
itself swaps only its binary, so an existing host keeps the unit it was
installed with until an operator installs the new one.
internal/agent/exposeproxy/exposeproxy.go
Old New
@@ -18,19 +18,45 @@
18 // Nothing here is persisted. Listeners are rebuilt from the first snapshot 18 // Nothing here is persisted. Listeners are rebuilt from the first snapshot
19 // after the agent starts, the same way the consoles are; connections in flight 19 // after the agent starts, the same way the consoles are; connections in flight
20 // across an agent restart drop, and reconnecting works. 20 // across an agent restart drop, and reconnecting works.
21 //
22 // Every published port serves under a cap, because the agent that carries this
23 // proxy is also the thing that runs the guests: traffic on a published port
24 // must never be able to starve fleet management of file descriptors.
21 package exposeproxy 25 package exposeproxy
22 26
23 import ( 27 import (
28 "errors"
24 "fmt" 29 "fmt"
25 "io" 30 "io"
26 "log/slog" 31 "log/slog"
27 "net" 32 "net"
28 "strconv" 33 "strconv"
29 "sync" 34 "sync"
35 "sync/atomic"
36 "syscall"
37 "time"
30 38
31 "github.com/a73x/eitri/internal/pb" 39 "github.com/a73x/eitri/internal/pb"
32 ) 40 )
33 41
42 // maxConnsPerExposure is how many connections one published port may hold open
43 // at once. Each costs two descriptors — the caller's and the dial into the
44 // guest — so an exposure sitting at its cap spends 512 of the 65536 the shipped
45 // unit grants the agent: a host could serve dozens of published ports, all of
46 // them saturated, and still leave the agent's own descriptors (hypervisor
47 // children, disks, consoles, the sync tunnel) untouched. 256 at once is far
48 // more than a service behind a published port meets in real use, and a caller
49 // past it is closed immediately — a refusal the caller can see beats a proxy
50 // wedged in a way nobody can diagnose.
51 const maxConnsPerExposure = 256
52
53 // acceptRetryFloor and acceptRetryCeiling bound the pause an accept loop takes
54 // when the process is momentarily out of descriptors.
55 const (
56 acceptRetryFloor = 5 * time.Millisecond
57 acceptRetryCeiling = time.Second
58 )
59
34 // Manager runs one listener per active exposure. 60 // Manager runs one listener per active exposure.
35 type Manager struct { 61 type Manager struct {
36 // addr answers a VM's current guest address, or "" when this host does not 62 // addr answers a VM's current guest address, or "" when this host does not
@@ -39,6 +65,11 @@ type Manager struct {
39 // goroutines; it must be safe for concurrent use. 65 // goroutines; it must be safe for concurrent use.
40 addr func(vmID string) string 66 addr func(vmID string) string
41 67
68 // maxConns is maxConnsPerExposure, held per-Manager so a test can serve the
69 // same behaviour with a handful of connections instead of hundreds. Written
70 // once at construction and only read after, so it needs no lock.
71 maxConns int64
72
42 mu sync.Mutex 73 mu sync.Mutex
43 live map[string]*exposure 74 live map[string]*exposure
44 } 75 }
@@ -46,15 +77,21 @@ type Manager struct {
46 // NewManager returns a Manager that resolves each connection's destination 77 // NewManager returns a Manager that resolves each connection's destination
47 // through addr. 78 // through addr.
48 func NewManager(addr func(vmID string) string) *Manager { 79 func NewManager(addr func(vmID string) string) *Manager {
49 return &Manager{addr: addr, live: map[string]*exposure{}} 80 return &Manager{addr: addr, maxConns: maxConnsPerExposure, live: map[string]*exposure{}}
50 } 81 }
51 82
52 // exposure is one published port's running state: the spec key it was bound 83 // exposure is one published port's running state: the spec key it was bound
53 // for, its listener (nil when the bind failed), and what the OS said if it did. 84 // for, its listener (nil when the bind failed), what the OS said if it did, and
85 // how many connections it is holding open right now.
86 //
87 // The count lives here rather than with the listener so that an exposure whose
88 // accept loop died and was rebound keeps counting what is still draining
89 // through it: the descriptors those connections hold are still spent.
54 type exposure struct { 90 type exposure struct {
55 key string 91 key string
56 ln net.Listener 92 ln net.Listener
57 reason string 93 reason string
94 conns atomic.Int64
58 } 95 }
59 96
60 // close tears down the listener, which is what ends its accept goroutine. 97 // close tears down the listener, which is what ends its accept goroutine.
@@ -112,7 +149,7 @@ func (m *Manager) Converge(desired []*pb.ExposureDesired) []*pb.ExposureActual {
112 continue 149 continue
113 } 150 }
114 ex.ln, ex.reason = ln, "" 151 ex.ln, ex.reason = ln, ""
115 go m.accept(d.GetId(), ln, d.GetVmId(), d.GetGuestPort()) 152 go m.accept(d.GetId(), ex, ln, d.GetVmId(), d.GetGuestPort())
116 } 153 }
117 out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "active"}) 154 out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "active"})
118 } 155 }
@@ -139,26 +176,71 @@ func (m *Manager) StopAll() {
139 // The pointer-identity guard is what keeps the normal exits quiet: on a 176 // The pointer-identity guard is what keeps the normal exits quiet: on a
140 // close-driven exit the entry is already gone, already nil, or already holds a 177 // close-driven exit the entry is already gone, already nil, or already holds a
141 // newer listener, so only the loop that owns the current listener writes here. 178 // newer listener, so only the loop that owns the current listener writes here.
142 func (m *Manager) accept(id string, ln net.Listener, vmID string, guestPort uint32) { 179 //
180 // A descriptor shortage is the one accept error this does not die on. Handing
181 // the listener back would unbind a working port and ask the next converge to
182 // find a descriptor for a fresh one — exactly what the process has none of —
183 // so the loop pauses and keeps the port, and the backlog keeps callers waiting
184 // rather than refusing them. Everything else is a broken listener and heals the
185 // only way a broken listener can, by being rebound.
186 func (m *Manager) accept(id string, ex *exposure, ln net.Listener, vmID string, guestPort uint32) {
187 var pause time.Duration
188 capped := false
143 for { 189 for {
144 conn, err := ln.Accept() 190 conn, err := ln.Accept()
145 if err != nil { 191 if err != nil {
192 if outOfDescriptors(err) {
193 if pause == 0 {
194 pause = acceptRetryFloor
195 slog.Warn("exposure accept out of descriptors, retrying", "exposure", id, "err", err)
196 } else {
197 pause = min(pause*2, acceptRetryCeiling)
198 }
199 time.Sleep(pause)
200 continue
201 }
146 m.mu.Lock() 202 m.mu.Lock()
147 if ex, ok := m.live[id]; ok && ex.ln == ln { 203 if cur, ok := m.live[id]; ok && cur.ln == ln {
148 ex.close() 204 cur.close()
149 ex.reason = "accept: " + err.Error() 205 cur.reason = "accept: " + err.Error()
150 // The reason travels in the next report, which nobody watching 206 // The reason travels in the next report, which nobody watching
151 // the host sees; fd exhaustion and a broken listener belong in 207 // the host sees; a broken listener belongs in the agent's log
152 // the agent's log too. 208 // too.
153 slog.Warn("exposure accept loop died", "exposure", id, "err", err) 209 slog.Warn("exposure accept loop died", "exposure", id, "err", err)
154 } 210 }
155 m.mu.Unlock() 211 m.mu.Unlock()
156 return 212 return
157 } 213 }
158 go m.pipe(conn, vmID, guestPort) 214 pause = 0
215
216 // Increment first and give it back if the cap says no: counting only
217 // after the decision would let a burst of simultaneous accepts each read
218 // a stale count and all pass.
219 if ex.conns.Add(1) > m.maxConns {
220 ex.conns.Add(-1)
221 conn.Close()
222 if !capped {
223 capped = true
224 slog.Warn("exposure at its connection cap, refusing", "exposure", id, "cap", m.maxConns)
225 }
226 continue
227 }
228 capped = false
229 go func() {
230 defer ex.conns.Add(-1)
231 m.pipe(conn, vmID, guestPort)
232 }()
159 } 233 }
160 } 234 }
161 235
236 // outOfDescriptors reports whether an accept failed because the process (EMFILE)
237 // or the machine (ENFILE) has no descriptor to give it. The runtime already
238 // retries the other transient accept errors — an interrupted call, a caller that
239 // went away between SYN and accept — so those never reach here.
240 func outOfDescriptors(err error) bool {
241 return errors.Is(err, syscall.EMFILE) || errors.Is(err, syscall.ENFILE)
242 }
243
162 // pipe connects one accepted connection to the guest. No address (a guest still 244 // pipe connects one accepted connection to the guest. No address (a guest still
163 // leasing) or a guest that will not answer closes immediately: the host half of 245 // leasing) or a guest that will not answer closes immediately: the host half of
164 // the pipe exists, the guest half does not, and waiting would only hold the 246 // the pipe exists, the guest half does not, and waiting would only hold the
internal/agent/exposeproxy/exposeproxy_test.go
Old New
@@ -4,6 +4,7 @@ import (
4 "io" 4 "io"
5 "net" 5 "net"
6 "strconv" 6 "strconv"
7 "syscall"
7 "testing" 8 "testing"
8 "time" 9 "time"
9 10
@@ -317,6 +318,174 @@ func TestRevokeDrainsRatherThanCuts(t *testing.T) {
317 assert.Equal(t, "two", exchange("two"), "closing a listener drains: only new dials die") 318 assert.Equal(t, "two", exchange("two"), "closing a listener drains: only new dials die")
318 } 319 }
319 320
321 // newHoldingGuest is a guest that echoes for as long as a caller keeps
322 // speaking, so a test can hold connections open across the proxy.
323 func newHoldingGuest(t *testing.T) *fakeGuest {
324 t.Helper()
325 ln, err := net.Listen("tcp", "127.0.0.1:0")
326 require.NoError(t, err)
327 t.Cleanup(func() { ln.Close() })
328 go func() {
329 for {
330 c, err := ln.Accept()
331 if err != nil {
332 return
333 }
334 go func() {
335 defer c.Close()
336 _, _ = io.Copy(c, c)
337 }()
338 }
339 }()
340 return &fakeGuest{ln: ln, addr: "127.0.0.1", port: portOf(t, ln.Addr())}
341 }
342
343 // hold opens a connection through the proxy and proves it is spliced through to
344 // the guest, which is also what makes the exposure's count of it deterministic:
345 // a byte came back, so the connection was counted before its pipe started.
346 func hold(t *testing.T, addr string) net.Conn {
347 t.Helper()
348 c, err := net.Dial("tcp", addr)
349 require.NoError(t, err)
350 t.Cleanup(func() { c.Close() })
351 require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
352 require.Equal(t, "ping", exchange(t, c, "ping"))
353 return c
354 }
355
356 // exchange writes msg on an open connection and reads the echo back.
357 func exchange(t *testing.T, c net.Conn, msg string) string {
358 t.Helper()
359 _, err := c.Write([]byte(msg))
360 require.NoError(t, err)
361 buf := make([]byte, len(msg))
362 _, err = io.ReadFull(c, buf)
363 require.NoError(t, err)
364 return string(buf)
365 }
366
367 // waitConns waits for an exposure's held-connection count to reach want. The
368 // increment is synchronous with the accept, but the decrement happens when the
369 // pipe ends, which is a goroutine finishing on its own time.
370 func waitConns(t *testing.T, m *Manager, id string, want int64) {
371 t.Helper()
372 var got int64
373 for i := 0; i < 200; i++ {
374 m.mu.Lock()
375 ex, ok := m.live[id]
376 m.mu.Unlock()
377 require.True(t, ok, "no exposure %q", id)
378 if got = ex.conns.Load(); got == want {
379 return
380 }
381 time.Sleep(10 * time.Millisecond)
382 }
383 t.Fatalf("exposure %q holds %d connections, want %d", id, got, want)
384 }
385
386 func TestExposureRefusesConnectionsPastItsCap(t *testing.T) {
387 g := newHoldingGuest(t)
388 m := newTestManager(t, map[string]string{"vm1": g.addr})
389 m.maxConns = 2
390 m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
391 addr := boundPort(t, m, "e1")
392
393 first, second := hold(t, addr), hold(t, addr)
394
395 // The listener still accepts — the refusal is a close, not a bind that went
396 // away — and the caller learns immediately rather than waiting on a promise.
397 over, err := net.Dial("tcp", addr)
398 require.NoError(t, err)
399 defer over.Close()
400 require.NoError(t, over.SetDeadline(time.Now().Add(5*time.Second)))
401 out, err := io.ReadAll(over)
402 require.NoError(t, err)
403 assert.Empty(t, out, "a connection past the cap is closed at once")
404
405 assert.Equal(t, "still", exchange(t, first, "still"), "a refusal must not disturb the connections under the cap")
406 assert.Equal(t, "serving", exchange(t, second, "serving"))
407 }
408
409 func TestTheCapReleasesWhenAConnectionEnds(t *testing.T) {
410 g := newHoldingGuest(t)
411 m := newTestManager(t, map[string]string{"vm1": g.addr})
412 m.maxConns = 1
413 m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
414 addr := boundPort(t, m, "e1")
415
416 first := hold(t, addr)
417 require.NoError(t, first.Close())
418 waitConns(t, m, "e1", 0)
419
420 // The cap is a level, not a ratchet: the slot the closed connection held is
421 // the slot this one takes.
422 assert.Equal(t, "after", exchange(t, hold(t, addr), "after"))
423 }
424
425 // scriptedListener hands an accept loop exactly the sequence a test writes to
426 // it — the errors a real listener only produces under a load a test cannot
427 // stage.
428 type scriptedListener struct {
429 next chan acceptResult
430 addr net.Addr
431 }
432
433 type acceptResult struct {
434 conn net.Conn
435 err error
436 }
437
438 func (l *scriptedListener) Accept() (net.Conn, error) {
439 r := <-l.next
440 return r.conn, r.err
441 }
442 func (l *scriptedListener) Close() error { return nil }
443 func (l *scriptedListener) Addr() net.Addr { return l.addr }
444
445 func TestAcceptKeepsThePortThroughADescriptorShortage(t *testing.T) {
446 g := newFakeGuest(t)
447 m := newTestManager(t, map[string]string{"vm1": g.addr})
448
449 ln := &scriptedListener{next: make(chan acceptResult), addr: g.ln.Addr()}
450 ex := &exposure{key: "k", ln: ln}
451 m.mu.Lock()
452 m.live["e1"] = ex
453 m.mu.Unlock()
454 go m.accept("e1", ex, ln, "vm1", g.port)
455
456 // Out of descriptors is not a broken listener. Handing the port back would
457 // only ask the next converge to find a descriptor the process does not have.
458 ln.next <- acceptResult{err: syscall.EMFILE}
459 ln.next <- acceptResult{err: syscall.EMFILE}
460
461 client, accepted := net.Pipe()
462 defer client.Close()
463 ln.next <- acceptResult{conn: accepted}
464 require.NoError(t, client.SetDeadline(time.Now().Add(5*time.Second)))
465 _, err := client.Write([]byte("hello"))
466 require.NoError(t, err)
467 buf := make([]byte, len("echo:hello"))
468 _, err = io.ReadFull(client, buf)
469 require.NoError(t, err)
470 assert.Equal(t, "echo:hello", string(buf), "the loop kept its listener and went on serving")
471
472 // A listener that is genuinely broken still ends the loop and gives the
473 // exposure back its listener-less state for the next converge to heal.
474 ln.next <- acceptResult{err: net.ErrClosed}
475 var reason string
476 for i := 0; i < 200; i++ {
477 m.mu.Lock()
478 gone, r := ex.ln == nil, ex.reason
479 m.mu.Unlock()
480 if gone {
481 reason = r
482 break
483 }
484 time.Sleep(10 * time.Millisecond)
485 }
486 assert.Contains(t, reason, "accept:")
487 }
488
320 func TestStopAllClosesEveryListener(t *testing.T) { 489 func TestStopAllClosesEveryListener(t *testing.T) {
321 g := newFakeGuest(t) 490 g := newFakeGuest(t)
322 m := NewManager(func(string) string { return g.addr }) 491 m := NewManager(func(string) string { return g.addr })
scripts/eitri-agent.service
Old New
@@ -33,6 +33,14 @@ RestartSec=5
33 # running VM on a mere agent stop. Only the agent itself may be signalled — 33 # running VM on a mere agent stop. Only the agent itself may be signalled —
34 # VMs survive agent restarts by design. 34 # VMs survive agent restarts by design.
35 KillMode=process 35 KillMode=process
36 # The agent's descriptor budget, stated rather than inherited from whatever the
37 # distro's default happens to be. Published guest ports are proxied inside this
38 # process, two descriptors per connection, and each exposure serves at most 256
39 # at once: dozens of saturated published ports fit here with room to spare, and
40 # what is left over is what the agent runs the fleet with — hypervisor children,
41 # disk images, consoles, the sync tunnel. Traffic on a published port must never
42 # be able to starve fleet management.
43 LimitNOFILE=65536
36 # Root is deliberate and scoped to THIS unit (eitri-server and eitri-oidc run 44 # Root is deliberate and scoped to THIS unit (eitri-server and eitri-oidc run
37 # as dedicated users): the agent opens /dev/kvm and /dev/net/tun, creates the 45 # as dedicated users): the agent opens /dev/kvm and /dev/net/tun, creates the
38 # bridge and taps, sets net.ipv4.ip_forward, installs the nftables masquerade, 46 # bridge and taps, sets net.ipv4.ip_forward, installs the nftables masquerade,
scripts/eitri-server.service
Old New
@@ -23,6 +23,12 @@ ExecStart=/usr/local/bin/eitri-server --config /etc/eitri/server.json
23 StateDirectory=eitri 23 StateDirectory=eitri
24 Restart=on-failure 24 Restart=on-failure
25 RestartSec=5 25 RestartSec=5
26 # Same stated budget as the agent, for the same reason: what this process holds
27 # open is decided by how many people are using the fleet, not by the box. Every
28 # gate session is a pair of connections and every open console is a stream that
29 # lives as long as the browser tab does, so the descriptor count follows the
30 # fleet's use and should not sit on a distro default nobody chose.
31 LimitNOFILE=65536
26 NoNewPrivileges=yes 32 NoNewPrivileges=yes
27 ProtectSystem=strict 33 ProtectSystem=strict
28 ProtectHome=yes 34 ProtectHome=yes