kacerr/egress-proxies-operator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0239fb7e692e7ad0041a7db0a578b3495a43e507
egress-proxies-operator/internal/provider/kvm2/backend.go
T
kacerrandClaude Opus 5 0239fb7e69 kvm2: do not let a stalled write delay the connection teardown (H1) 
sshSocket.Close was errors.Join(s.Conn.Close(), s.client.Close()). Go
evaluates those left to right, so the blocking close ran before the
unblocking one. s.Conn is the forwarded libvirt socket, and closing an SSH
channel writes a close message under ch.writeMu — which go-libvirt already
holds while streaming a base image through that same channel. Against a
host that vanished without RST or FIN, that write blocks until the kernel
stops retransmitting (tcp_retries2, ~15 min), and s.client.Close(), a bare
net.Conn.Close() that takes no lock and unblocks the stalled writer through
netpoll, was never reached. Close the client only; nothing wants the polite
channel-close message, and reordering instead would leave an error to
filter in a memoised closeErr.

link.markDead compounded it by running kill() before close(l.closed), both
inside a sync.Once. Reordering alone was not enough: sync.Once.Do blocks
concurrent callers until the first body returns, so every other markDead —
including the one in call is own ctx-cancel branch — queued behind the
stalled kill, and cancelling the reconcile did not help. The once field is
gone; the mutex the struct already had guards the close-exactly-once
invariant, reason is set before the signal, and kill runs outside the lock.

Blast radius was the whole operator, not one host: MaxConcurrentReconciles
is 3 and ensureImage holds a per-image lock whose acquire is ctx-aware but
reconcile contexts carry no deadline, so three Proxies on one host with an
uncached image parked every worker.

The same edits close a second route to the same pin: go-libvirt is
Disconnect closes the net.Conn it was handed (libvirt.go:331 ->
socket.go:164), and that conn is our sshSocket, so libvirtBackend.Close is
Disconnect goroutine entered closeOnce and hung; the 5 s closeBudget
released Close, which called markDead -> kill -> sshSocket.Close and
blocked in the other goroutine is sync.Once.

Verified against the pinned x/crypto v0.54.0 rather than assumed: the
flow-control park (remoteWin.reserve, channel.go:268) is outside writeMu
and only writePacket holds it across the TCP write, so a writer parked in
SSH flow control never blocks the close and a writer inside the TCP write
always does. To reproduce by hand, shrink SO_SNDBUF well under the 2 MiB
channel window (channelWindowSize = 64 * channelMaxPacket) so the socket
buffer fills before the window does.

The kvm provider does not have this bug: it closes client only and never
touches the forwarded channel.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-17 15:30:41 +00:00

374 lines
15 KiB
Go
Raw Blame History

package kvm2
import (
"context"
"fmt"
"io"
"sync"
"time"
"github.com/digitalocean/go-libvirt"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
)
// backend is the seam over go-libvirt (spec entry 17): exactly the calls the
// provider uses, each with ctx first, since go-libvirt's own methods take
// none and block. The production adapter is libvirtBackend; tests use an
// in-memory fake. Nothing outside this package sees it.
type backend interface {
DomainLookupByName(ctx context.Context, name string) (libvirt.Domain, error)
DomainDefineXML(ctx context.Context, xml string) (libvirt.Domain, error)
DomainCreate(ctx context.Context, dom libvirt.Domain) error
DomainDestroy(ctx context.Context, dom libvirt.Domain) error
DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, flags libvirt.DomainUndefineFlagsValues) error
DomainGetState(ctx context.Context, dom libvirt.Domain, flags uint32) (state int32, reason int32, err error)
DomainGetMetadata(ctx context.Context, dom libvirt.Domain, typ int32, uri libvirt.OptString, flags libvirt.DomainModificationImpact) (string, error)
DomainInterfaceAddresses(ctx context.Context, dom libvirt.Domain, source uint32, flags uint32) ([]libvirt.DomainInterface, error)
ConnectListAllDomains(ctx context.Context, needResults int32, flags libvirt.ConnectListAllDomainsFlags) ([]libvirt.Domain, uint32, error)
StoragePoolLookupByName(ctx context.Context, name string) (libvirt.StoragePool, error)
StoragePoolRefresh(ctx context.Context, pool libvirt.StoragePool, flags uint32) error
StoragePoolGetInfo(ctx context.Context, pool libvirt.StoragePool) (state uint8, capacity, allocation, available uint64, err error)
StorageVolLookupByName(ctx context.Context, pool libvirt.StoragePool, name string) (libvirt.StorageVol, error)
StorageVolGetPath(ctx context.Context, vol libvirt.StorageVol) (string, error)
StorageVolCreateXML(ctx context.Context, pool libvirt.StoragePool, xml string, flags libvirt.StorageVolCreateFlags) (libvirt.StorageVol, error)
StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error
StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error
StorageVolDelete(ctx context.Context, vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error
ConnectGetLibVersion(ctx context.Context) (uint64, error)
ConnectGetHostname(ctx context.Context) (string, error)
Close() error
}
// liveness is implemented by backends whose connection can die underneath
// them. The host table drops a backend whose dead channel is closed. It is
// deliberately not part of backend: the fake needs it, the seam does not.
type liveness interface {
dead() <-chan struct{}
}
func isDead(b backend) bool {
l, ok := b.(liveness)
if !ok {
return false
}
select {
case <-l.dead():
return true
default:
return false
}
}
// budgets are the per-call deadlines of spec entry 18, independent of the
// caller's ctx, which by contract carries none.
type budgets struct {
// dial bounds TCP connect, SSH handshake and libvirt connect together.
dial time.Duration
// rpc bounds every ordinary call.
rpc time.Duration
// create bounds DomainCreate, which waits for QEMU to start.
create time.Duration
// seedUpload bounds a StorageVolUpload of at most smallUploadBytes.
seedUpload time.Duration
// transfer caps a large upload or a download; ctx may end it earlier.
transfer time.Duration
}
var defaultBudgets = budgets{
dial: 10 * time.Second,
rpc: 30 * time.Second,
create: 90 * time.Second,
seedUpload: 60 * time.Second,
transfer: 20 * time.Minute,
}
// smallUploadBytes separates a seed ISO (tens of KiB) from a base image
// (hundreds of MiB) for the upload budget.
const smallUploadBytes = 16 << 20
// closeBudget bounds the courtesy ConnectClose RPC in Close.
const closeBudget = 5 * time.Second
func (b budgets) upload(length uint64) time.Duration {
if length <= smallUploadBytes {
return b.seedUpload
}
return b.transfer
}
// link is the liveness of one connection. dead is closed exactly once, by
// the first of: an abandoned call, a keepalive failure, go-libvirt noticing
// the socket went away, or Close. kill closes the transport, which is the
// only way to unblock a go-libvirt call; go-libvirt then fails every other
// call in flight on the connection with ErrInterrupted.
type link struct {
label string
kill func()
closed chan struct{}
mu sync.Mutex
reason string
}
func newLink(label string, kill func()) *link {
return &link{label: label, kill: kill, closed: make(chan struct{})}
}
func (l *link) dead() <-chan struct{} { return l.closed }
func (l *link) isDead() bool {
select {
case <-l.closed:
return true
default:
return false
}
}
func (l *link) deadReason() string {
l.mu.Lock()
defer l.mu.Unlock()
return l.reason
}
// markDead records why the connection is unusable, signals dead and then
// closes the transport. Later calls are no-ops. Signalling before killing,
// and killing outside the lock, is what keeps a kill that takes its time
// from holding up the callers waiting on dead or the next markDead: a
// transport close can block for as long as the kernel retransmits.
func (l *link) markDead(reason string) {
l.mu.Lock()
if l.isDead() {
l.mu.Unlock()
return
}
l.reason = reason
close(l.closed)
l.mu.Unlock()
if l.kill != nil {
l.kill()
}
}
// call executes a blocking go-libvirt call with a budget and returns its
// result, or Transient when the link is already dead, ctx ends, the budget
// runs out, or another caller kills the link meanwhile. Abandoning a call
// kills the link (spec entry 18); the goroutine running fn then returns on
// its own once go-libvirt sees the closed socket, and its result is dropped.
// Results travel through the channel so an abandoned call never writes
// into its caller's frame.
func call[T any](ctx context.Context, l *link, name string, budget time.Duration, fn func() (T, error)) (T, error) {
var zero T
if l.isDead() {
return zero, transientf("libvirt call %s on %s: connection is closed: %s", name, l.label, l.deadReason())
}
if err := ctx.Err(); err != nil {
return zero, fmt.Errorf("%w: libvirt call %s on %s not started: %w", provider.ErrTransient, name, l.label, err)
}
type result struct {
v T
err error
}
done := make(chan result, 1)
go func() {
v, err := fn()
done <- result{v, err}
}()
timer := time.NewTimer(budget)
defer timer.Stop()
select {
case r := <-done:
return r.v, r.err
case <-ctx.Done():
l.markDead(fmt.Sprintf("call %s abandoned: %v", name, ctx.Err()))
return zero, fmt.Errorf("%w: libvirt call %s on %s abandoned: %w", provider.ErrTransient, name, l.label, ctx.Err())
case <-timer.C:
l.markDead(fmt.Sprintf("call %s exceeded its %s budget", name, budget))
return zero, fmt.Errorf("%w: libvirt call %s on %s exceeded its %s budget: %w", provider.ErrTransient, name, l.label, budget, context.DeadlineExceeded)
case <-l.closed:
return zero, transientf("libvirt call %s on %s: connection closed while waiting: %s", name, l.label, l.deadReason())
}
}
// run is call for go-libvirt methods that return only an error.
func (l *link) run(ctx context.Context, name string, budget time.Duration, fn func() error) error {
_, err := call(ctx, l, name, budget, func() (struct{}, error) { return struct{}{}, fn() })
return err
}
// libvirtBackend is the production adapter: a go-libvirt client over one
// forwarded socket, with the budgets applied around every call. It never
// logs; the observe decorator does.
type libvirtBackend struct {
lv *libvirt.Libvirt
link *link
budgets budgets
closeOnce sync.Once
closeErr error
}
var _ backend = (*libvirtBackend)(nil)
// newLibvirtBackend wraps lv, whose socket runs over transport. Closing the
// transport is how a hung call is abandoned, so it must close the socket
// conn given to libvirt.New as well as whatever carries it.
func newLibvirtBackend(lv *libvirt.Libvirt, label string, transport io.Closer, b budgets) *libvirtBackend {
return &libvirtBackend{
lv: lv,
link: newLink(label, func() { _ = transport.Close() }),
budgets: b,
}
}
func (b *libvirtBackend) dead() <-chan struct{} { return b.link.dead() }
// connect runs the libvirt handshake under the dial budget and, once it
// holds, drops the link as soon as go-libvirt reports the socket gone.
func (b *libvirtBackend) connect(ctx context.Context) error {
if err := b.link.run(ctx, "Connect", b.budgets.dial, b.lv.Connect); err != nil {
return err
}
disconnected := b.lv.Disconnected()
go func() {
select {
case <-disconnected:
b.link.markDead("connection lost")
case <-b.link.closed:
}
}()
return nil
}
func (b *libvirtBackend) DomainLookupByName(ctx context.Context, name string) (libvirt.Domain, error) {
return call(ctx, b.link, "DomainLookupByName", b.budgets.rpc, func() (libvirt.Domain, error) { return b.lv.DomainLookupByName(name) })
}
func (b *libvirtBackend) DomainDefineXML(ctx context.Context, xml string) (libvirt.Domain, error) {
return call(ctx, b.link, "DomainDefineXML", b.budgets.rpc, func() (libvirt.Domain, error) { return b.lv.DomainDefineXML(xml) })
}
func (b *libvirtBackend) DomainCreate(ctx context.Context, dom libvirt.Domain) error {
return b.link.run(ctx, "DomainCreate", b.budgets.create, func() error { return b.lv.DomainCreate(dom) })
}
func (b *libvirtBackend) DomainDestroy(ctx context.Context, dom libvirt.Domain) error {
return b.link.run(ctx, "DomainDestroy", b.budgets.rpc, func() error { return b.lv.DomainDestroy(dom) })
}
func (b *libvirtBackend) DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, flags libvirt.DomainUndefineFlagsValues) error {
return b.link.run(ctx, "DomainUndefineFlags", b.budgets.rpc, func() error { return b.lv.DomainUndefineFlags(dom, flags) })
}
func (b *libvirtBackend) DomainGetState(ctx context.Context, dom libvirt.Domain, flags uint32) (int32, int32, error) {
type stateReason struct{ state, reason int32 }
r, err := call(ctx, b.link, "DomainGetState", b.budgets.rpc, func() (stateReason, error) {
state, reason, err := b.lv.DomainGetState(dom, flags)
return stateReason{state, reason}, err
})
return r.state, r.reason, err
}
func (b *libvirtBackend) DomainGetMetadata(ctx context.Context, dom libvirt.Domain, typ int32, uri libvirt.OptString, flags libvirt.DomainModificationImpact) (string, error) {
return call(ctx, b.link, "DomainGetMetadata", b.budgets.rpc, func() (string, error) { return b.lv.DomainGetMetadata(dom, typ, uri, flags) })
}
func (b *libvirtBackend) DomainInterfaceAddresses(ctx context.Context, dom libvirt.Domain, source uint32, flags uint32) ([]libvirt.DomainInterface, error) {
return call(ctx, b.link, "DomainInterfaceAddresses", b.budgets.rpc, func() ([]libvirt.DomainInterface, error) {
return b.lv.DomainInterfaceAddresses(dom, source, flags)
})
}
func (b *libvirtBackend) ConnectListAllDomains(ctx context.Context, needResults int32, flags libvirt.ConnectListAllDomainsFlags) ([]libvirt.Domain, uint32, error) {
type listed struct {
doms []libvirt.Domain
n uint32
}
r, err := call(ctx, b.link, "ConnectListAllDomains", b.budgets.rpc, func() (listed, error) {
doms, n, err := b.lv.ConnectListAllDomains(needResults, flags)
return listed{doms, n}, err
})
return r.doms, r.n, err
}
func (b *libvirtBackend) StoragePoolLookupByName(ctx context.Context, name string) (libvirt.StoragePool, error) {
return call(ctx, b.link, "StoragePoolLookupByName", b.budgets.rpc, func() (libvirt.StoragePool, error) { return b.lv.StoragePoolLookupByName(name) })
}
func (b *libvirtBackend) StoragePoolRefresh(ctx context.Context, pool libvirt.StoragePool, flags uint32) error {
return b.link.run(ctx, "StoragePoolRefresh", b.budgets.rpc, func() error { return b.lv.StoragePoolRefresh(pool, flags) })
}
func (b *libvirtBackend) StoragePoolGetInfo(ctx context.Context, pool libvirt.StoragePool) (uint8, uint64, uint64, uint64, error) {
type info struct {
state uint8
capacity, allocation, available uint64
}
r, err := call(ctx, b.link, "StoragePoolGetInfo", b.budgets.rpc, func() (info, error) {
state, capacity, allocation, available, err := b.lv.StoragePoolGetInfo(pool)
return info{state, capacity, allocation, available}, err
})
return r.state, r.capacity, r.allocation, r.available, err
}
func (b *libvirtBackend) StorageVolLookupByName(ctx context.Context, pool libvirt.StoragePool, name string) (libvirt.StorageVol, error) {
return call(ctx, b.link, "StorageVolLookupByName", b.budgets.rpc, func() (libvirt.StorageVol, error) { return b.lv.StorageVolLookupByName(pool, name) })
}
func (b *libvirtBackend) StorageVolGetPath(ctx context.Context, vol libvirt.StorageVol) (string, error) {
return call(ctx, b.link, "StorageVolGetPath", b.budgets.rpc, func() (string, error) { return b.lv.StorageVolGetPath(vol) })
}
func (b *libvirtBackend) StorageVolCreateXML(ctx context.Context, pool libvirt.StoragePool, xml string, flags libvirt.StorageVolCreateFlags) (libvirt.StorageVol, error) {
return call(ctx, b.link, "StorageVolCreateXML", b.budgets.rpc, func() (libvirt.StorageVol, error) { return b.lv.StorageVolCreateXML(pool, xml, flags) })
}
func (b *libvirtBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error {
return b.link.run(ctx, "StorageVolUpload", b.budgets.upload(length), func() error {
return b.lv.StorageVolUpload(vol, r, offset, length, flags)
})
}
func (b *libvirtBackend) StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error {
return b.link.run(ctx, "StorageVolDownload", b.budgets.transfer, func() error {
return b.lv.StorageVolDownload(vol, w, offset, length, flags)
})
}
func (b *libvirtBackend) StorageVolDelete(ctx context.Context, vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error {
return b.link.run(ctx, "StorageVolDelete", b.budgets.rpc, func() error { return b.lv.StorageVolDelete(vol, flags) })
}
func (b *libvirtBackend) ConnectGetLibVersion(ctx context.Context) (uint64, error) {
return call(ctx, b.link, "ConnectGetLibVersion", b.budgets.rpc, b.lv.ConnectGetLibVersion)
}
func (b *libvirtBackend) ConnectGetHostname(ctx context.Context) (string, error) {
return call(ctx, b.link, "ConnectGetHostname", b.budgets.rpc, b.lv.ConnectGetHostname)
}
// Close sends libvirt's ConnectClose when the link is still live, bounded
// because a hung host would otherwise pin the caller, then kills the link,
// which closes the socket and the SSH client. Repeat calls return the first
// result.
func (b *libvirtBackend) Close() error {
b.closeOnce.Do(func() {
if b.link.isDead() {
return
}
done := make(chan error, 1)
go func() { done <- b.lv.Disconnect() }()
timer := time.NewTimer(closeBudget)
defer timer.Stop()
select {
case b.closeErr = <-done:
case <-timer.C:
b.closeErr = transientf("libvirt disconnect on %s exceeded its %s budget", b.link.label, closeBudget)
case <-b.link.closed:
}
b.link.markDead("closed")
})
return b.closeErr
}
Reference in New Issue View Git Blame Copy Permalink