kacerrandClaude Fable 5.1
b45f83327a
Every go-libvirt call runs under its own budget and a cancelled or overlong call kills the connection; credentials are read on every dial; one span and one log line per RPC come from a decorator applied where connections are made. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
370 lines
15 KiB
Go
370 lines
15 KiB
Go
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()
|
|
|
|
once sync.Once
|
|
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, closes the transport and
|
|
// signals dead. Later calls are no-ops.
|
|
func (l *link) markDead(reason string) {
|
|
l.once.Do(func() {
|
|
l.mu.Lock()
|
|
l.reason = reason
|
|
l.mu.Unlock()
|
|
if l.kill != nil {
|
|
l.kill()
|
|
}
|
|
close(l.closed)
|
|
})
|
|
}
|
|
|
|
// 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
|
|
}
|