egress-proxies-operator/internal/provider/kvm2/backend.go

package kvm2

import (
	"context"
	"errors"
	"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

// transferGrace keeps the image download's own deadline inside the transfer
// budget. A stalled mirror then ends by its body failing, which the upload
// seam turns into a clean end of stream, instead of by the budget killing
// the socket, which go-libvirt's stream sender does not survive.
const transferGrace = time.Minute

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 {
	src := &eofReader{r: r}
	err := b.link.run(ctx, "StorageVolUpload", b.budgets.upload(length), func() error {
		return b.lv.StorageVolUpload(vol, src, offset, length, flags)
	})
	if err == nil {
		if srcErr := src.sourceErr(); srcErr != nil {
			return fmt.Errorf("reading the upload source for %s: %w", vol.Name, srcErr)
		}
	}
	return err
}

// eofReader hides a source error from go-libvirt's stream sender and keeps
// it for the caller. SendStream answers a non-EOF error with a StatusError
// packet and then blocks forever handing its result to an unbuffered
// channel that requestStream no longer reads (go-libvirt rpc.go), which
// wedges the call and leaks the goroutine along with its 4 MiB buffer;
// closing the socket does not help, because both ends are parked on
// channel sends rather than on I/O. io.EOF is the one outcome that ends
// the stream the normal way. The field is guarded because an abandoned
// call returns while go-libvirt is still reading.
type eofReader struct {
	r   io.Reader
	mu  sync.Mutex
	err error
}

func (e *eofReader) Read(b []byte) (int, error) {
	n, err := e.r.Read(b)
	if err == nil || errors.Is(err, io.EOF) {
		return n, err
	}
	e.mu.Lock()
	if e.err == nil {
		e.err = err
	}
	e.mu.Unlock()
	return n, io.EOF
}

func (e *eofReader) sourceErr() error {
	e.mu.Lock()
	defer e.mu.Unlock()
	return e.err
}

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
}