From b45f83327aff79853708688c157e3fadb477a660 Mon Sep 17 00:00:00 2001 From: kacerr Date: Wed, 16 Sep 2026 17:43:27 +0000 Subject: [PATCH] kvm2: SSH transport, ctx-first libvirt seam, host table and span decorator 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 --- .../2026-09-16-1333-kvm2-provider.md | 48 +- internal/provider/kvm2/backend.go | 369 ++++++++++++ internal/provider/kvm2/backend_test.go | 252 +++++++++ internal/provider/kvm2/fake_test.go | 534 ++++++++++++++++++ internal/provider/kvm2/hosts.go | 183 ++++++ internal/provider/kvm2/hosts_test.go | 212 +++++++ internal/provider/kvm2/kvm2.go | 16 +- internal/provider/kvm2/kvm2_test.go | 6 +- internal/provider/kvm2/observe.go | 263 +++++++++ internal/provider/kvm2/observe_test.go | 334 +++++++++++ internal/provider/kvm2/ssh.go | 216 +++++++ internal/provider/kvm2/ssh_test.go | 343 +++++++++++ internal/provider/kvm2/sshserver_test.go | 289 ++++++++++ 13 files changed, 3055 insertions(+), 10 deletions(-) create mode 100644 internal/provider/kvm2/backend.go create mode 100644 internal/provider/kvm2/backend_test.go create mode 100644 internal/provider/kvm2/fake_test.go create mode 100644 internal/provider/kvm2/hosts.go create mode 100644 internal/provider/kvm2/hosts_test.go create mode 100644 internal/provider/kvm2/observe.go create mode 100644 internal/provider/kvm2/observe_test.go create mode 100644 internal/provider/kvm2/ssh.go create mode 100644 internal/provider/kvm2/ssh_test.go create mode 100644 internal/provider/kvm2/sshserver_test.go diff --git a/docs/plans-executions/2026-09-16-1333-kvm2-provider.md b/docs/plans-executions/2026-09-16-1333-kvm2-provider.md index a500777..e3e142c 100644 --- a/docs/plans-executions/2026-09-16-1333-kvm2-provider.md +++ b/docs/plans-executions/2026-09-16-1333-kvm2-provider.md @@ -6,7 +6,7 @@ Plan: [docs/plans/2026-09-16-1333-kvm2-provider.md](../plans/2026-09-16-1333-kvm - [x] Step 1 — The spec, `docs/kvm2-provider-spec.md` - [x] Step 2 — Config wiring, IDs, state mapping, classify - [x] Step 3 — Pure builders: domain and volume XML, metadata, seed -- [ ] Step 4 — Transport, seam, decorator, fakes +- [x] Step 4 — Transport, seam, decorator, fakes - [ ] Step 5 — Image cache and pool-space precheck - [ ] Step 6 — The four contract methods - [ ] Step 7 — Checker and the CRD comment @@ -125,3 +125,49 @@ go build ./... && go test -short ./... All `ok`. Review greps for a v1 import or v1 identifiers are empty; the only `type="volume"` in the package is the comment explaining why it is not used. The agent reported "files read outside the allowed set: none". + +## Step 4 — Transport, seam, decorator, fakes + +2026-09-16 17:42. Third subagent step, the largest: `backend.go` (the +ctx-first seam `backend`, the production adapter `libvirtBackend`, a +generic `call[T]` helper with per-call budgets and the abandon policy), +`ssh.go` (credentials read on every dial, stage-tagged `dialError`, the +forwarded socket, keepalive), `hosts.go` (`hostTable` with a per-host +channel semaphore so waiting for a dial ends with the caller's ctx, +`dialFresh` for dedicated connections, the production `libvirtDialer` +starting the `kvm2.dial` span and applying the decorator once), +`observe.go` (one span `kvm2.` and one `libvirt call` line per +RPC), plus an in-memory `fakeBackend`, an in-process SSH server and 41 +tests. The agent also drove the real adapter over go-libvirt's own +`libvirttest` mock through a `net.Pipe`, which the brief did not ask for. + +Abandon policy as built: a call that outlives its ctx or budget kills the +link (closes the socket and the SSH client), returns Transient wrapping the +ctx or deadline error, and other in-flight calls on that connection fail once +with go-libvirt's `ErrInterrupted`; the next `get` redials. Results travel +through a channel so a late-finishing abandoned call never writes into its +caller's frame; the agent reported that the first `-race` run caught exactly +that bug in an earlier version. + +Decisions the agent made, accepted: liveness is a separate unexported +interface so the reflection test over `backend` stays at the 20 RPCs plus +`Close`; `Close` sends a bounded `ConnectClose` before killing the link; +`observe` and the dialer take `...tracing.Option` so tests can inject a +recording tracer provider, as `provider.WithTracing` does. + +Clean-room note: three type names coincide with names v1 uses +(`dialError`, `dialStage`, `hostEntry`). The brief described each concept +("an error type that records which stage failed", "a per-host entry") so the +names follow from the description; the greps for a v1 import and for the +listed v1 identifiers are empty and the agent attested to reading nothing +under `internal/provider/kvm/`. + +Commands (agent and orchestrator): + +```sh +gofmt -l ./internal/provider/kvm2 ; go vet ./internal/provider/kvm2/ && go test -race -count=2 ./internal/provider/kvm2/ +go build ./... && go test -short ./... +grep -n 'context.Background()' internal/provider/kvm2/*.go | grep -v _test.go # empty +``` + +All `ok`; the agent also ran the timing-sensitive tests at `-count=8`. diff --git a/internal/provider/kvm2/backend.go b/internal/provider/kvm2/backend.go new file mode 100644 index 0000000..7a74306 --- /dev/null +++ b/internal/provider/kvm2/backend.go @@ -0,0 +1,369 @@ +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 +} diff --git a/internal/provider/kvm2/backend_test.go b/internal/provider/kvm2/backend_test.go new file mode 100644 index 0000000..0da1b9f --- /dev/null +++ b/internal/provider/kvm2/backend_test.go @@ -0,0 +1,252 @@ +package kvm2 + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + "github.com/digitalocean/go-libvirt/libvirttest" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// blockingFn returns a func that blocks until release is closed, plus a +// counter of how often it was started. +func blockingFn(t *testing.T) (fn func() error, started *atomic.Int32) { + t.Helper() + release := make(chan struct{}) + t.Cleanup(func() { close(release) }) + started = &atomic.Int32{} + return func() error { + started.Add(1) + <-release + return nil + }, started +} + +func TestCall_deadlineClosesConnectionAndReturnsTransient(t *testing.T) { + t.Parallel() + var killed atomic.Int32 + l := newLink("beelink", func() { killed.Add(1) }) + fn, started := blockingFn(t) + + start := time.Now() + err := l.run(context.Background(), "DomainCreate", 50*time.Millisecond, fn) + elapsed := time.Since(start) + if !errors.Is(err, provider.ErrTransient) { + t.Fatalf("run() error = %v, want ErrTransient", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("run() error = %v, want it to wrap context.DeadlineExceeded", err) + } + if !strings.Contains(err.Error(), "DomainCreate") || !strings.Contains(err.Error(), "beelink") { + t.Errorf("run() error = %q, want the call name and host in it", err) + } + if elapsed > time.Second { + t.Errorf("run() took %v, want about the 50ms budget", elapsed) + } + if killed.Load() != 1 { + t.Errorf("kill hook called %d times, want 1", killed.Load()) + } + if !l.isDead() { + t.Error("link not dead after an abandoned call") + } + if !strings.Contains(l.deadReason(), "budget") { + t.Errorf("deadReason() = %q, want the budget mentioned", l.deadReason()) + } + + start = time.Now() + err = l.run(context.Background(), "DomainGetState", time.Minute, fn) + if !errors.Is(err, provider.ErrTransient) { + t.Fatalf("second run() error = %v, want ErrTransient", err) + } + if time.Since(start) > 100*time.Millisecond { + t.Errorf("second run() took %v, want immediate", time.Since(start)) + } + if started.Load() != 1 { + t.Errorf("fn started %d times, want 1: a dead link must not touch the socket", started.Load()) + } + if killed.Load() != 1 { + t.Errorf("kill hook called %d times, want still 1", killed.Load()) + } +} + +func TestCall_cancelUnblocksRPC(t *testing.T) { + t.Parallel() + var killed atomic.Int32 + l := newLink("beelink", func() { killed.Add(1) }) + fn, _ := blockingFn(t) + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(10 * time.Millisecond) + cancel() + }() + + start := time.Now() + err := l.run(ctx, "ConnectListAllDomains", time.Minute, fn) + if time.Since(start) > 100*time.Millisecond { + t.Errorf("run() took %v after cancel, want under 100ms", time.Since(start)) + } + if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.Canceled) { + t.Errorf("run() error = %v, want ErrTransient wrapping context.Canceled", err) + } + if killed.Load() != 1 { + t.Errorf("kill hook called %d times, want 1", killed.Load()) + } +} + +func TestCall_alreadyCancelledDoesNotStart(t *testing.T) { + t.Parallel() + l := newLink("beelink", nil) + fn, started := blockingFn(t) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err := l.run(ctx, "DomainGetState", time.Minute, fn) + if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.Canceled) { + t.Errorf("run() error = %v, want ErrTransient wrapping context.Canceled", err) + } + if started.Load() != 0 { + t.Error("fn started although ctx was already cancelled") + } + if l.isDead() { + t.Error("link marked dead although nothing was abandoned") + } +} + +func TestCall_returnsFnResult(t *testing.T) { + t.Parallel() + l := newLink("beelink", nil) + want := libvirt.Error{Code: uint32(libvirt.ErrNoDomain), Message: "gone"} + err := l.run(context.Background(), "DomainLookupByName", time.Second, func() error { return want }) + if !hasCode(err, libvirt.ErrNoDomain) { + t.Errorf("run() error = %v, want the libvirt error through unchanged", err) + } + if l.isDead() { + t.Error("a backend error must not kill the link") + } +} + +func TestCall_killedByAnotherCallerFailsTransient(t *testing.T) { + t.Parallel() + l := newLink("beelink", nil) + fn, _ := blockingFn(t) + done := make(chan error, 1) + go func() { done <- l.run(context.Background(), "DomainGetState", time.Minute, fn) }() + time.Sleep(10 * time.Millisecond) + l.markDead("keepalive failed") + select { + case err := <-done: + if !errors.Is(err, provider.ErrTransient) || !strings.Contains(err.Error(), "keepalive failed") { + t.Errorf("run() error = %v, want ErrTransient with the death reason", err) + } + case <-time.After(time.Second): + t.Fatal("run() did not return after the link was killed") + } +} + +func TestBudgets_upload(t *testing.T) { + t.Parallel() + tests := []struct { + name string + length uint64 + want time.Duration + }{ + {name: "seed sized", length: 400 << 10, want: defaultBudgets.seedUpload}, + {name: "exactly the limit", length: smallUploadBytes, want: defaultBudgets.seedUpload}, + {name: "one over the limit", length: smallUploadBytes + 1, want: defaultBudgets.transfer}, + {name: "base image", length: 435 << 20, want: defaultBudgets.transfer}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := defaultBudgets.upload(tc.length); got != tc.want { + t.Errorf("upload(%d) = %v, want %v", tc.length, got, tc.want) + } + }) + } +} + +// mockAdapter wires a libvirtBackend to go-libvirt's own mock server over a +// net.Pipe, so the adapter is exercised with the real RPC client. +func mockAdapter(t *testing.T, b budgets) *libvirtBackend { + t.Helper() + m := libvirttest.New() + conn, err := m.Dial() + if err != nil { + t.Fatalf("mock dial: %v", err) + } + ad := newLibvirtBackend(libvirt.New(conn), "mock", conn, b) + if err := ad.connect(context.Background()); err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(func() { _ = ad.Close() }) + return ad +} + +func TestLibvirtBackend_roundTripThroughMock(t *testing.T) { + t.Parallel() + ad := mockAdapter(t, defaultBudgets) + v, err := ad.ConnectGetLibVersion(context.Background()) + if err != nil { + t.Fatalf("ConnectGetLibVersion() error = %v", err) + } + if v == 0 { + t.Error("ConnectGetLibVersion() = 0, want the mock's version") + } + pool, err := ad.StoragePoolLookupByName(context.Background(), "default") + if err != nil { + t.Fatalf("StoragePoolLookupByName() error = %v", err) + } + if pool.Name == "" { + t.Error("StoragePoolLookupByName() returned an empty pool") + } + if isDead(ad) { + t.Error("adapter dead after successful calls") + } +} + +func TestLibvirtBackend_unansweredCallIsAbandoned(t *testing.T) { + t.Parallel() + b := defaultBudgets + b.rpc = 100 * time.Millisecond + ad := mockAdapter(t, b) + // The mock does not implement ConnectGetHostname, so it never replies. + _, err := ad.ConnectGetHostname(context.Background()) + if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("ConnectGetHostname() error = %v, want ErrTransient wrapping DeadlineExceeded", err) + } + if !isDead(ad) { + t.Error("adapter not dead after an abandoned call") + } + select { + case <-ad.lv.Disconnected(): + case <-time.After(time.Second): + t.Error("go-libvirt did not notice the closed socket") + } + if _, err := ad.ConnectGetLibVersion(context.Background()); !errors.Is(err, provider.ErrTransient) { + t.Errorf("call on dead adapter error = %v, want ErrTransient", err) + } + if err := ad.Close(); err != nil { + t.Errorf("Close() on dead adapter error = %v, want nil", err) + } +} + +func TestLibvirtBackend_closeIsIdempotent(t *testing.T) { + t.Parallel() + ad := mockAdapter(t, defaultBudgets) + first := ad.Close() + second := ad.Close() + if first != nil || second != nil { + t.Errorf("Close() errors = %v, %v, want nil, nil", first, second) + } + if !isDead(ad) { + t.Error("adapter not dead after Close") + } + if _, err := ad.ConnectGetLibVersion(context.Background()); !errors.Is(err, provider.ErrTransient) { + t.Errorf("call after Close error = %v, want ErrTransient", err) + } +} diff --git a/internal/provider/kvm2/fake_test.go b/internal/provider/kvm2/fake_test.go new file mode 100644 index 0000000..3ea5aa3 --- /dev/null +++ b/internal/provider/kvm2/fake_test.go @@ -0,0 +1,534 @@ +package kvm2 + +import ( + "context" + "encoding/xml" + "fmt" + "io" + "slices" + "sync" + "sync/atomic" + + "github.com/digitalocean/go-libvirt" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// fakeBackend is the in-memory backend of spec entry 21: enough of a +// libvirt host for the provider's Create, Get, Delete and ListByTag, with +// per-method error hooks and blocking, call counters and an ordered log. +type fakeBackend struct { + mu sync.Mutex + domains map[string]*fakeDomain + pools map[string]*fakePool + hostname string + version uint64 + + // hooks returns the error a method fails with before touching state. + hooks map[string]func() error + // block makes a method wait on its channel (or ctx) before running. + block map[string]chan struct{} + + calls map[string]int + log []string + closed int + deadCh chan struct{} + deadMu sync.Once +} + +type fakeDomain struct { + state libvirt.DomainState + xml string + metadata string + addrs map[uint32][]libvirt.DomainInterface + uuid libvirt.UUID + id int32 +} + +type fakePool struct { + state libvirt.StoragePoolState + capacity uint64 + available uint64 + vols map[string]*fakeVol +} + +type fakeVol struct { + content []byte + capacity uint64 +} + +const fakePoolName = "default" + +func newFakeBackend() *fakeBackend { + return &fakeBackend{ + domains: make(map[string]*fakeDomain), + pools: map[string]*fakePool{fakePoolName: newFakePool()}, + hostname: "fake-host", + version: 11006000, + hooks: make(map[string]func() error), + block: make(map[string]chan struct{}), + calls: make(map[string]int), + deadCh: make(chan struct{}), + } +} + +func newFakePool() *fakePool { + return &fakePool{state: libvirt.StoragePoolRunning, capacity: 100 << 30, available: 50 << 30, vols: make(map[string]*fakeVol)} +} + +func (f *fakeBackend) dead() <-chan struct{} { return f.deadCh } + +func (f *fakeBackend) markDead() { f.deadMu.Do(func() { close(f.deadCh) }) } + +// addDomain defines a domain directly, bypassing DomainDefineXML. +func (f *fakeBackend) addDomain(name string, state libvirt.DomainState, metadata string) *fakeDomain { + f.mu.Lock() + defer f.mu.Unlock() + d := &fakeDomain{state: state, metadata: metadata, addrs: make(map[uint32][]libvirt.DomainInterface), uuid: fakeUUID(name)} + f.domains[name] = d + return d +} + +func (f *fakeBackend) failWith(method string, err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.hooks[method] = func() error { return err } +} + +func (f *fakeBackend) blockOn(method string) chan struct{} { + f.mu.Lock() + defer f.mu.Unlock() + ch := make(chan struct{}) + f.block[method] = ch + return ch +} + +func (f *fakeBackend) callCount(method string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls[method] +} + +func (f *fakeBackend) callLog() []string { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.log) +} + +func fakeUUID(name string) libvirt.UUID { + var u libvirt.UUID + copy(u[:], name) + return u +} + +func fakeErr(code libvirt.ErrorNumber, format string, args ...any) error { + return libvirt.Error{Code: uint32(code), Message: fmt.Sprintf(format, args...)} +} + +// enter records the call, honours a block and runs the hook. It is called +// without f.mu held; state access happens afterwards under the lock. +func (f *fakeBackend) enter(ctx context.Context, method string) error { + f.mu.Lock() + f.calls[method]++ + f.log = append(f.log, method) + block := f.block[method] + hook := f.hooks[method] + f.mu.Unlock() + if block != nil { + select { + case <-block: + case <-ctx.Done(): + return ctx.Err() + } + } + if hook != nil { + return hook() + } + return nil +} + +func (f *fakeBackend) domain(name string) (*fakeDomain, error) { + d, ok := f.domains[name] + if !ok { + return nil, fakeErr(libvirt.ErrNoDomain, "Domain not found: no domain with matching name '%s'", name) + } + return d, nil +} + +func (f *fakeBackend) pool(name string) (*fakePool, error) { + p, ok := f.pools[name] + if !ok { + return nil, fakeErr(libvirt.ErrNoStoragePool, "Storage pool not found: no storage pool with matching name '%s'", name) + } + return p, nil +} + +func (f *fakeBackend) volume(vol libvirt.StorageVol) (*fakeVol, error) { + p, err := f.pool(vol.Pool) + if err != nil { + return nil, err + } + v, ok := p.vols[vol.Name] + if !ok { + return nil, fakeErr(libvirt.ErrNoStorageVol, "Storage volume not found: no storage vol with matching name '%s'", vol.Name) + } + return v, nil +} + +func (f *fakeBackend) libvirtDomain(name string, d *fakeDomain) libvirt.Domain { + id := int32(-1) + if isActive(d.state) { + id = d.id + } + return libvirt.Domain{Name: name, UUID: d.uuid, ID: id} +} + +func (f *fakeBackend) DomainLookupByName(ctx context.Context, name string) (libvirt.Domain, error) { + if err := f.enter(ctx, "DomainLookupByName"); err != nil { + return libvirt.Domain{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + d, err := f.domain(name) + if err != nil { + return libvirt.Domain{}, err + } + return f.libvirtDomain(name, d), nil +} + +func (f *fakeBackend) DomainDefineXML(ctx context.Context, doc string) (libvirt.Domain, error) { + if err := f.enter(ctx, "DomainDefineXML"); err != nil { + return libvirt.Domain{}, err + } + var parsed struct { + Name string `xml:"name"` + Metadata struct { + Inner string `xml:",innerxml"` + } `xml:"metadata"` + } + if err := xml.Unmarshal([]byte(doc), &parsed); err != nil || parsed.Name == "" { + return libvirt.Domain{}, fakeErr(libvirt.ErrXMLError, "XML error: %v", err) + } + f.mu.Lock() + defer f.mu.Unlock() + d, ok := f.domains[parsed.Name] + if !ok { + d = &fakeDomain{state: libvirt.DomainShutoff, addrs: make(map[uint32][]libvirt.DomainInterface), uuid: fakeUUID(parsed.Name)} + f.domains[parsed.Name] = d + } + d.xml = doc + d.metadata = parsed.Metadata.Inner + return f.libvirtDomain(parsed.Name, d), nil +} + +func (f *fakeBackend) DomainCreate(ctx context.Context, dom libvirt.Domain) error { + if err := f.enter(ctx, "DomainCreate"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + d, err := f.domain(dom.Name) + if err != nil { + return err + } + if isActive(d.state) { + return fakeErr(libvirt.ErrOperationInvalid, "Requested operation is not valid: domain is already running") + } + d.state = libvirt.DomainRunning + d.id = int32(len(f.domains)) + return nil +} + +func (f *fakeBackend) DomainDestroy(ctx context.Context, dom libvirt.Domain) error { + if err := f.enter(ctx, "DomainDestroy"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + d, err := f.domain(dom.Name) + if err != nil { + return err + } + if !isActive(d.state) { + return fakeErr(libvirt.ErrOperationInvalid, "Requested operation is not valid: domain is not running") + } + d.state = libvirt.DomainShutoff + return nil +} + +func (f *fakeBackend) DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, _ libvirt.DomainUndefineFlagsValues) error { + if err := f.enter(ctx, "DomainUndefineFlags"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + if _, err := f.domain(dom.Name); err != nil { + return err + } + delete(f.domains, dom.Name) + return nil +} + +func (f *fakeBackend) DomainGetState(ctx context.Context, dom libvirt.Domain, _ uint32) (int32, int32, error) { + if err := f.enter(ctx, "DomainGetState"); err != nil { + return 0, 0, err + } + f.mu.Lock() + defer f.mu.Unlock() + d, err := f.domain(dom.Name) + if err != nil { + return 0, 0, err + } + return int32(d.state), 0, nil +} + +func (f *fakeBackend) DomainGetMetadata(ctx context.Context, dom libvirt.Domain, typ int32, uri libvirt.OptString, _ libvirt.DomainModificationImpact) (string, error) { + if err := f.enter(ctx, "DomainGetMetadata"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + d, err := f.domain(dom.Name) + if err != nil { + return "", err + } + if typ != int32(libvirt.DomainMetadataElement) || len(uri) != 1 || uri[0] != metadataURI || d.metadata == "" { + return "", fakeErr(libvirt.ErrNoDomainMetadata, "Requested metadata element is not present") + } + return d.metadata, nil +} + +func (f *fakeBackend) DomainInterfaceAddresses(ctx context.Context, dom libvirt.Domain, source uint32, _ uint32) ([]libvirt.DomainInterface, error) { + if err := f.enter(ctx, "DomainInterfaceAddresses"); err != nil { + return nil, err + } + f.mu.Lock() + defer f.mu.Unlock() + d, err := f.domain(dom.Name) + if err != nil { + return nil, err + } + if !isActive(d.state) { + return nil, fakeErr(libvirt.ErrOperationInvalid, "Requested operation is not valid: domain is not running") + } + return slices.Clone(d.addrs[source]), nil +} + +func (f *fakeBackend) ConnectListAllDomains(ctx context.Context, _ int32, _ libvirt.ConnectListAllDomainsFlags) ([]libvirt.Domain, uint32, error) { + if err := f.enter(ctx, "ConnectListAllDomains"); err != nil { + return nil, 0, err + } + f.mu.Lock() + defer f.mu.Unlock() + names := make([]string, 0, len(f.domains)) + for name := range f.domains { + names = append(names, name) + } + slices.Sort(names) + out := make([]libvirt.Domain, 0, len(names)) + for _, name := range names { + out = append(out, f.libvirtDomain(name, f.domains[name])) + } + return out, uint32(len(out)), nil +} + +func (f *fakeBackend) StoragePoolLookupByName(ctx context.Context, name string) (libvirt.StoragePool, error) { + if err := f.enter(ctx, "StoragePoolLookupByName"); err != nil { + return libvirt.StoragePool{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + if _, err := f.pool(name); err != nil { + return libvirt.StoragePool{}, err + } + return libvirt.StoragePool{Name: name, UUID: fakeUUID(name)}, nil +} + +func (f *fakeBackend) StoragePoolRefresh(ctx context.Context, pool libvirt.StoragePool, _ uint32) error { + if err := f.enter(ctx, "StoragePoolRefresh"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + _, err := f.pool(pool.Name) + return err +} + +func (f *fakeBackend) StoragePoolGetInfo(ctx context.Context, pool libvirt.StoragePool) (uint8, uint64, uint64, uint64, error) { + if err := f.enter(ctx, "StoragePoolGetInfo"); err != nil { + return 0, 0, 0, 0, err + } + f.mu.Lock() + defer f.mu.Unlock() + p, err := f.pool(pool.Name) + if err != nil { + return 0, 0, 0, 0, err + } + return uint8(p.state), p.capacity, p.capacity - p.available, p.available, nil +} + +func (f *fakeBackend) StorageVolLookupByName(ctx context.Context, pool libvirt.StoragePool, name string) (libvirt.StorageVol, error) { + if err := f.enter(ctx, "StorageVolLookupByName"); err != nil { + return libvirt.StorageVol{}, err + } + f.mu.Lock() + defer f.mu.Unlock() + vol := libvirt.StorageVol{Pool: pool.Name, Name: name, Key: fakeVolPath(pool.Name, name)} + if _, err := f.volume(vol); err != nil { + return libvirt.StorageVol{}, err + } + return vol, nil +} + +func fakeVolPath(pool, name string) string { return "/pool/" + pool + "/" + name } + +func (f *fakeBackend) StorageVolGetPath(ctx context.Context, vol libvirt.StorageVol) (string, error) { + if err := f.enter(ctx, "StorageVolGetPath"); err != nil { + return "", err + } + f.mu.Lock() + defer f.mu.Unlock() + if _, err := f.volume(vol); err != nil { + return "", err + } + return fakeVolPath(vol.Pool, vol.Name), nil +} + +func (f *fakeBackend) StorageVolCreateXML(ctx context.Context, pool libvirt.StoragePool, doc string, _ libvirt.StorageVolCreateFlags) (libvirt.StorageVol, error) { + if err := f.enter(ctx, "StorageVolCreateXML"); err != nil { + return libvirt.StorageVol{}, err + } + var parsed struct { + Name string `xml:"name"` + Capacity uint64 `xml:"capacity"` + } + if err := xml.Unmarshal([]byte(doc), &parsed); err != nil || parsed.Name == "" { + return libvirt.StorageVol{}, fakeErr(libvirt.ErrXMLError, "XML error: %v", err) + } + f.mu.Lock() + defer f.mu.Unlock() + p, err := f.pool(pool.Name) + if err != nil { + return libvirt.StorageVol{}, err + } + if _, exists := p.vols[parsed.Name]; exists { + return libvirt.StorageVol{}, fakeErr(libvirt.ErrStorageVolExist, "storage volume '%s' exists already", parsed.Name) + } + p.vols[parsed.Name] = &fakeVol{capacity: parsed.Capacity} + return libvirt.StorageVol{Pool: pool.Name, Name: parsed.Name, Key: fakeVolPath(pool.Name, parsed.Name)}, nil +} + +func (f *fakeBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, _, _ uint64, _ libvirt.StorageVolUploadFlags) error { + if err := f.enter(ctx, "StorageVolUpload"); err != nil { + return err + } + var content []byte + if r != nil { + var err error + if content, err = io.ReadAll(r); err != nil { + return err + } + } + f.mu.Lock() + defer f.mu.Unlock() + v, err := f.volume(vol) + if err != nil { + return err + } + v.content = content + return nil +} + +func (f *fakeBackend) StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, _, _ uint64, _ libvirt.StorageVolDownloadFlags) error { + if err := f.enter(ctx, "StorageVolDownload"); err != nil { + return err + } + f.mu.Lock() + v, err := f.volume(vol) + var content []byte + if v != nil { + content = slices.Clone(v.content) + } + f.mu.Unlock() + if err != nil { + return err + } + if w == nil { + return nil + } + _, err = w.Write(content) + return err +} + +func (f *fakeBackend) StorageVolDelete(ctx context.Context, vol libvirt.StorageVol, _ libvirt.StorageVolDeleteFlags) error { + if err := f.enter(ctx, "StorageVolDelete"); err != nil { + return err + } + f.mu.Lock() + defer f.mu.Unlock() + if _, err := f.volume(vol); err != nil { + return err + } + delete(f.pools[vol.Pool].vols, vol.Name) + return nil +} + +func (f *fakeBackend) ConnectGetLibVersion(ctx context.Context) (uint64, error) { + if err := f.enter(ctx, "ConnectGetLibVersion"); err != nil { + return 0, err + } + return f.version, nil +} + +func (f *fakeBackend) ConnectGetHostname(ctx context.Context) (string, error) { + if err := f.enter(ctx, "ConnectGetHostname"); err != nil { + return "", err + } + return f.hostname, nil +} + +func (f *fakeBackend) Close() error { + f.mu.Lock() + defer f.mu.Unlock() + f.closed++ + return nil +} + +// stubDialer covers every dialer a test needs: it counts dials, fails with +// err when set, waits on block (or ctx) when set, and hands out backend, or +// one from fresh per call, or a new fakeBackend when neither is given. +type stubDialer struct { + backend backend + fresh func() backend + err error + block chan struct{} + + dials atomic.Int32 + mu sync.Mutex + hosts []string +} + +func (d *stubDialer) dial(ctx context.Context, host provider.KVMHost) (backend, error) { + d.dials.Add(1) + d.mu.Lock() + d.hosts = append(d.hosts, host.Name) + d.mu.Unlock() + if d.block != nil { + select { + case <-d.block: + case <-ctx.Done(): + return nil, fmt.Errorf("%w: dialing %s: %w", provider.ErrTransient, host.Name, ctx.Err()) + } + } + if d.err != nil { + return nil, d.err + } + switch { + case d.fresh != nil: + return d.fresh(), nil + case d.backend != nil: + return d.backend, nil + default: + return newFakeBackend(), nil + } +} diff --git a/internal/provider/kvm2/hosts.go b/internal/provider/kvm2/hosts.go new file mode 100644 index 0000000..2bffc32 --- /dev/null +++ b/internal/provider/kvm2/hosts.go @@ -0,0 +1,183 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/digitalocean/go-libvirt" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +// dialer opens one connection to a host. The production dialer is +// libvirtDialer; tests inject one that hands out fakes. +type dialer interface { + dial(ctx context.Context, host provider.KVMHost) (backend, error) +} + +// hostTable holds the pooled connection per host (spec entry 19). One dial +// per host at a time: concurrent callers wait for it and share the result. +// A dead backend (abandoned call, keepalive failure, socket lost) is dropped +// and redialled on the next get. +// +// What a cancelled or overlong call does to the shared connection (spec +// entry 18): the adapter kills it, so every other call in flight on that +// host fails once with Transient and its caller retries on a fresh +// connection. Callers that must not be exposed to that, the image transfer +// and the Checker, use dialFresh and get a connection of their own. +type hostTable struct { + dialer dialer + + mu sync.Mutex + hosts map[string]*hostEntry +} + +// hostEntry serialises dials for one host. The semaphore, not a mutex, so a +// waiter can give up when its ctx ends. +type hostEntry struct { + sem chan struct{} + backend backend +} + +func newHostTable(d dialer) *hostTable { + return &hostTable{dialer: d, hosts: make(map[string]*hostEntry)} +} + +func (t *hostTable) entry(name string) *hostEntry { + t.mu.Lock() + defer t.mu.Unlock() + e, ok := t.hosts[name] + if !ok { + e = &hostEntry{sem: make(chan struct{}, 1)} + t.hosts[name] = e + } + return e +} + +// get returns the live pooled backend for host, dialling it when there is +// none. Waiting for another caller's dial ends with ctx. +func (t *hostTable) get(ctx context.Context, host provider.KVMHost) (backend, error) { + e := t.entry(host.Name) + select { + case e.sem <- struct{}{}: + case <-ctx.Done(): + return nil, fmt.Errorf("%w: waiting for the connection to host %s: %w", provider.ErrTransient, host.Name, ctx.Err()) + } + defer func() { <-e.sem }() + if e.backend != nil { + if !isDead(e.backend) { + return e.backend, nil + } + _ = e.backend.Close() + e.backend = nil + } + b, err := t.dialer.dial(ctx, host) + if err != nil { + return nil, err + } + e.backend = b + return b, nil +} + +// dialFresh returns a dedicated connection that is never pooled. The +// caller owns it and must Close it. +func (t *hostTable) dialFresh(ctx context.Context, host provider.KVMHost) (backend, error) { + return t.dialer.dial(ctx, host) +} + +// closeAll closes every pooled connection, waiting for in-flight dials. +func (t *hostTable) closeAll() error { + t.mu.Lock() + entries := make([]*hostEntry, 0, len(t.hosts)) + for _, e := range t.hosts { + entries = append(entries, e) + } + t.mu.Unlock() + var errs []error + for _, e := range entries { + e.sem <- struct{}{} + if e.backend != nil { + errs = append(errs, e.backend.Close()) + e.backend = nil + } + <-e.sem + } + return errors.Join(errs...) +} + +// libvirtDialer is the production dialer: SSH transport, go-libvirt over +// the forwarded socket, the connect handshake, then the observe decorator, +// so provider code never holds an undecorated backend. +type libvirtDialer struct { + providerName string + cfg provider.KVMConfig + budgets budgets + keepalive time.Duration + tracing []tracing.Option +} + +func newLibvirtDialer(providerName string, cfg provider.KVMConfig) *libvirtDialer { + return &libvirtDialer{ + providerName: providerName, + cfg: cfg, + budgets: defaultBudgets, + keepalive: defaultKeepaliveInterval, + } +} + +func (d *libvirtDialer) dial(ctx context.Context, host provider.KVMHost) (backend, error) { + ctx, span := tracing.StartSpan(ctx, tracing.Tracer(d.tracing...), "kvm2.dial", + trace.WithSpanKind(trace.SpanKindClient), + trace.WithAttributes( + attribute.String("kvm2.host", host.Name), + attribute.String("kvm2.address", host.Address), + attribute.String("kvm2.user", host.User), + )) + defer span.End() + log := logf.FromContext(ctx).WithName("kvm2").WithValues("provider", d.providerName, "host", host.Name) + start := time.Now() + b, err := d.connect(ctx, host) + if err != nil { + span.SetStatus(codes.Error, err.Error()) + kv := []any{"address", host.Address, "user", host.User, "duration", time.Since(start), "error", err.Error()} + if stage, ok := stageOf(err); ok { + kv = append(kv, "stage", string(stage)) + } + kv = append(kv, libvirtErrorFields(err)...) + log.V(1).Info("libvirt host connect failed", kv...) + return nil, err + } + log.V(1).Info("libvirt host connected", "address", host.Address, "user", host.User, "duration", time.Since(start)) + return observe(b, d.providerName, host.Name, d.tracing...), nil +} + +// connect is the dial sequence under one budget: the SSH stages, then the +// libvirt handshake. On success the keepalive and the disconnect watcher +// run until the link dies. +func (d *libvirtDialer) connect(ctx context.Context, host provider.KVMHost) (*libvirtBackend, error) { + ctx, cancel := context.WithTimeout(ctx, d.budgets.dial) + defer cancel() + sock, err := dialSSHSocket(ctx, host, d.cfg, 0) + if err != nil { + return nil, err + } + b := newLibvirtBackend(libvirt.New(sock), host.Name, sock, d.budgets) + if err := b.connect(ctx); err != nil { + b.link.markDead("connect failed") + return nil, fmt.Errorf("connecting to libvirt on host %s: %w", host.Name, err) + } + go func() { + if err := sock.keepalive(d.keepalive, b.link.dead()); err != nil { + b.link.markDead(err.Error()) + } + }() + return b, nil +} diff --git a/internal/provider/kvm2/hosts_test.go b/internal/provider/kvm2/hosts_test.go new file mode 100644 index 0000000..7e83857 --- /dev/null +++ b/internal/provider/kvm2/hosts_test.go @@ -0,0 +1,212 @@ +package kvm2 + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +var ( + hostA = provider.KVMHost{Name: "beelink", Address: "beelink:22", User: "virt"} + hostB = provider.KVMHost{Name: "lab2", Address: "lab2:22", User: "virt"} +) + +func TestHosts_oneDialPerHostUnderConcurrency(t *testing.T) { + t.Parallel() + shared := newFakeBackend() + d := &stubDialer{backend: shared} + table := newHostTable(d) + + const n = 50 + var wg sync.WaitGroup + got := make([]backend, n) + errs := make([]error, n) + for i := range n { + wg.Add(1) + go func() { + defer wg.Done() + got[i], errs[i] = table.get(context.Background(), hostA) + }() + } + wg.Wait() + for i := range n { + if errs[i] != nil { + t.Fatalf("get #%d error = %v", i, errs[i]) + } + if got[i] != backend(shared) { + t.Fatalf("get #%d returned a different backend", i) + } + } + if d.dials.Load() != 1 { + t.Errorf("dials = %d, want 1", d.dials.Load()) + } +} + +func TestHosts_redialsAfterDrop(t *testing.T) { + t.Parallel() + first := newFakeBackend() + second := newFakeBackend() + backends := []backend{first, second} + d := &stubDialer{fresh: func() backend { + b := backends[0] + backends = backends[1:] + return b + }} + table := newHostTable(d) + + b, err := table.get(context.Background(), hostA) + if err != nil || b != backend(first) { + t.Fatalf("first get = %v, %v, want the first backend", b, err) + } + if b, _ := table.get(context.Background(), hostA); b != backend(first) { + t.Fatal("second get did not reuse the live backend") + } + first.markDead() + b, err = table.get(context.Background(), hostA) + if err != nil || b != backend(second) { + t.Fatalf("get after drop = %v, %v, want the second backend", b, err) + } + if first.closed != 1 { + t.Errorf("dropped backend closed %d times, want 1", first.closed) + } + if d.dials.Load() != 2 { + t.Errorf("dials = %d, want 2", d.dials.Load()) + } +} + +func TestHosts_dialFreshIsNotCached(t *testing.T) { + t.Parallel() + d := &stubDialer{fresh: func() backend { return newFakeBackend() }} + table := newHostTable(d) + + pooled, err := table.get(context.Background(), hostA) + if err != nil { + t.Fatalf("get error = %v", err) + } + fresh, err := table.dialFresh(context.Background(), hostA) + if err != nil { + t.Fatalf("dialFresh error = %v", err) + } + if fresh == pooled { + t.Fatal("dialFresh returned the pooled backend") + } + again, _ := table.get(context.Background(), hostA) + if again != pooled { + t.Error("get after dialFresh did not return the pooled backend") + } + if d.dials.Load() != 2 { + t.Errorf("dials = %d, want 2", d.dials.Load()) + } + if fresh.(*fakeBackend).closed != 0 { + t.Error("dialFresh's backend was closed by the table; the caller owns it") + } +} + +func TestHosts_waitingForDialEndsWithContext(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + d := &stubDialer{block: release} + table := newHostTable(d) + + firstDone := make(chan error, 1) + go func() { + _, err := table.get(context.Background(), hostA) + firstDone <- err + }() + waitForDials(t, d, 1) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + start := time.Now() + _, err := table.get(ctx, hostA) + if time.Since(start) > time.Second { + t.Errorf("second get took %v, want to give up with its ctx", time.Since(start)) + } + if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("second get error = %v, want ErrTransient wrapping DeadlineExceeded", err) + } + close(release) + if err := <-firstDone; err != nil { + t.Errorf("first get error = %v, want nil once the dial completes", err) + } + if d.dials.Load() != 1 { + t.Errorf("dials = %d, want 1: the waiter must not dial itself", d.dials.Load()) + } +} + +func TestHosts_hostsDialIndependently(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + defer close(release) + dA := &stubDialer{block: release} + dB := &stubDialer{} + table := newHostTable(perHostDialer{hostA.Name: dA, hostB.Name: dB}) + + go func() { _, _ = table.get(context.Background(), hostA) }() + waitForDials(t, dA, 1) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + start := time.Now() + if _, err := table.get(ctx, hostB); err != nil { + t.Fatalf("get(hostB) error = %v while hostA's dial is blocked", err) + } + if time.Since(start) > 500*time.Millisecond { + t.Errorf("get(hostB) took %v, want not to wait for hostA", time.Since(start)) + } +} + +func TestHosts_dialErrorIsNotCached(t *testing.T) { + t.Parallel() + d := &stubDialer{err: transientf("boom")} + table := newHostTable(d) + if _, err := table.get(context.Background(), hostA); !errors.Is(err, provider.ErrTransient) { + t.Fatalf("get error = %v, want the dial error", err) + } + d.err = nil + if _, err := table.get(context.Background(), hostA); err != nil { + t.Fatalf("get after the dialer recovered error = %v", err) + } + if d.dials.Load() != 2 { + t.Errorf("dials = %d, want 2", d.dials.Load()) + } +} + +func TestHosts_closeAll(t *testing.T) { + t.Parallel() + table := newHostTable(&stubDialer{fresh: func() backend { return newFakeBackend() }}) + a, _ := table.get(context.Background(), hostA) + b, _ := table.get(context.Background(), hostB) + if err := table.closeAll(); err != nil { + t.Fatalf("closeAll error = %v", err) + } + if a.(*fakeBackend).closed != 1 || b.(*fakeBackend).closed != 1 { + t.Errorf("closed counts = %d, %d, want 1, 1", a.(*fakeBackend).closed, b.(*fakeBackend).closed) + } + again, _ := table.get(context.Background(), hostA) + if again == a { + t.Error("get after closeAll returned the closed backend") + } +} + +// perHostDialer routes each host to its own stub. +type perHostDialer map[string]*stubDialer + +func (p perHostDialer) dial(ctx context.Context, host provider.KVMHost) (backend, error) { + return p[host.Name].dial(ctx, host) +} + +func waitForDials(t *testing.T, d *stubDialer, n int32) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for d.dials.Load() < n { + if time.Now().After(deadline) { + t.Fatalf("dials = %d, want %d", d.dials.Load(), n) + } + time.Sleep(time.Millisecond) + } +} diff --git a/internal/provider/kvm2/kvm2.go b/internal/provider/kvm2/kvm2.go index 8560d32..b3e6e72 100644 --- a/internal/provider/kvm2/kvm2.go +++ b/internal/provider/kvm2/kvm2.go @@ -25,9 +25,10 @@ const ( // Provider is one configured kvm2 entry: a set of hosts sharing one SSH key // and one image and machine-type catalogue. type Provider struct { - name string - cfg provider.KVMConfig - now func() time.Time + name string + cfg provider.KVMConfig + now func() time.Time + hosts *hostTable } var _ provider.Provider = (*Provider)(nil) @@ -40,7 +41,8 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er if pc.KVM2 != nil { cfg = *pc.KVM2 } - p := newProvider(pc.Name, cfg) + cfg = withDefaults(cfg) + p := newProvider(pc.Name, cfg, newLibvirtDialer(pc.Name, cfg)) log := p.logger(ctx) if _, err := os.Stat(p.cfg.SSHKeyFile); err != nil { log.Info("WARNING: ssh key file is not readable; every call fails until it is", "path", p.cfg.SSHKeyFile, "reason", err.Error()) @@ -56,8 +58,10 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er return p, nil } -func newProvider(name string, cfg provider.KVMConfig) *Provider { - return &Provider{name: name, cfg: withDefaults(cfg), now: time.Now} +// newProvider is the constructor tests use: d hands out the backends, the +// production one being newLibvirtDialer. +func newProvider(name string, cfg provider.KVMConfig, d dialer) *Provider { + return &Provider{name: name, cfg: withDefaults(cfg), now: time.Now, hosts: newHostTable(d)} } // withDefaults returns cfg with every optional field filled in, so the rest diff --git a/internal/provider/kvm2/kvm2_test.go b/internal/provider/kvm2/kvm2_test.go index ad48b37..70145ce 100644 --- a/internal/provider/kvm2/kvm2_test.go +++ b/internal/provider/kvm2/kvm2_test.go @@ -146,7 +146,7 @@ func TestHost_table(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { t.Parallel() - p := newProvider("kvm2-home", provider.KVMConfig{Hosts: tc.hosts}) + p := newProvider("kvm2-home", provider.KVMConfig{Hosts: tc.hosts}, &stubDialer{}) h, err := p.host(tc.zone) if tc.wantErr != "" { if err == nil { @@ -250,7 +250,7 @@ func TestWithDefaults_doesNotAliasInput(t *testing.T) { func TestWrapErr_carriesProviderNameOpAndID(t *testing.T) { t.Parallel() - p := newProvider("kvm2-home", twoHosts()) + p := newProvider("kvm2-home", twoHosts(), &stubDialer{}) err := p.wrapErr("get", "hosts/beelink/domains/proxy-x", permanentf("boom")) var pe *provider.Error if !errors.As(err, &pe) { @@ -264,7 +264,7 @@ func TestWrapErr_carriesProviderNameOpAndID(t *testing.T) { func TestStubs_returnPermanentWithProviderName(t *testing.T) { t.Parallel() ctx := context.Background() - p := newProvider("kvm2-home", twoHosts()) + p := newProvider("kvm2-home", twoHosts(), &stubDialer{}) _, createErr := p.Create(ctx, provider.CreateRequest{Name: "proxy-x"}) inst, getErr := p.Get(ctx, "hosts/beelink/domains/proxy-x") deleteErr := p.Delete(ctx, "hosts/beelink/domains/proxy-x") diff --git a/internal/provider/kvm2/observe.go b/internal/provider/kvm2/observe.go new file mode 100644 index 0000000..05d0a8c --- /dev/null +++ b/internal/provider/kvm2/observe.go @@ -0,0 +1,263 @@ +package kvm2 + +import ( + "context" + "io" + "time" + + "github.com/digitalocean/go-libvirt" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +// observed is the span-decorator of spec entries 17 and 20: one child span +// kvm2. and one V(1) "libvirt call" line per call, with request and +// response detail at V(2). Applied once, in the dialer, over every backend +// that leaves it. +type observed struct { + inner backend + provider string + host string + tracer trace.Tracer +} + +var ( + _ backend = (*observed)(nil) + _ liveness = (*observed)(nil) +) + +func observe(inner backend, providerName, hostName string, opts ...tracing.Option) backend { + return &observed{inner: inner, provider: providerName, host: hostName, tracer: tracing.Tracer(opts...)} +} + +// dead forwards the inner backend's liveness; a backend without one is +// never dropped by the host table. +func (o *observed) dead() <-chan struct{} { + if l, ok := o.inner.(liveness); ok { + return l.dead() + } + return nil +} + +// id is one identifier a call names: a domain, pool or volume. It becomes +// both the span attribute kvm2. and the log key. +type id struct{ key, val string } + +func domainID(name string) id { return id{"domain", name} } +func poolID(name string) id { return id{"pool", name} } +func volumeID(name string) id { return id{"volume", name} } + +// detail returns V(2)-only key/value pairs. It runs only when V(2) is on, +// so building the list of a hundred domain names costs nothing otherwise. +type detail func() []any + +// call wraps one backend call: span, timing, status and the log line. fn +// returns the V(2) detail for the line, or nil. +func (o *observed) call(ctx context.Context, rpc string, ids []id, fn func(ctx context.Context) (detail, error)) error { + attrs := make([]attribute.KeyValue, 0, len(ids)+1) + attrs = append(attrs, attribute.String("kvm2.host", o.host)) + for _, i := range ids { + attrs = append(attrs, attribute.String("kvm2."+i.key, i.val)) + } + ctx, span := tracing.StartSpan(ctx, o.tracer, "kvm2."+rpc, trace.WithSpanKind(trace.SpanKindClient), trace.WithAttributes(attrs...)) + defer span.End() + log := logf.FromContext(ctx).WithName("kvm2").WithValues("provider", o.provider, "host", o.host) + start := time.Now() + more, err := fn(ctx) + kv := make([]any, 0, 8+2*len(ids)) + kv = append(kv, "rpc", rpc) + for _, i := range ids { + kv = append(kv, i.key, i.val) + } + kv = append(kv, "duration", time.Since(start)) + if err != nil { + span.SetStatus(codes.Error, err.Error()) + if code, ok := libvirtCode(err); ok { + span.SetAttributes(attribute.Int("kvm2.libvirt_code", int(code))) + } + kv = append(kv, "error", err.Error()) + kv = append(kv, libvirtErrorFields(err)...) + } + if more != nil && log.V(2).Enabled() { + kv = append(kv, more()...) + } + log.V(1).Info("libvirt call", kv...) + return err +} + +func (o *observed) DomainLookupByName(ctx context.Context, name string) (dom libvirt.Domain, err error) { + err = o.call(ctx, "DomainLookupByName", []id{domainID(name)}, func(ctx context.Context) (detail, error) { + dom, err = o.inner.DomainLookupByName(ctx, name) + return nil, err + }) + return dom, err +} + +func (o *observed) DomainDefineXML(ctx context.Context, xml string) (dom libvirt.Domain, err error) { + err = o.call(ctx, "DomainDefineXML", nil, func(ctx context.Context) (detail, error) { + dom, err = o.inner.DomainDefineXML(ctx, xml) + return func() []any { return []any{"domain", dom.Name, "xml", xml} }, err + }) + return dom, err +} + +func (o *observed) DomainCreate(ctx context.Context, dom libvirt.Domain) error { + return o.call(ctx, "DomainCreate", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) { + return nil, o.inner.DomainCreate(ctx, dom) + }) +} + +func (o *observed) DomainDestroy(ctx context.Context, dom libvirt.Domain) error { + return o.call(ctx, "DomainDestroy", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) { + return nil, o.inner.DomainDestroy(ctx, dom) + }) +} + +func (o *observed) DomainUndefineFlags(ctx context.Context, dom libvirt.Domain, flags libvirt.DomainUndefineFlagsValues) error { + return o.call(ctx, "DomainUndefineFlags", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) { + return nil, o.inner.DomainUndefineFlags(ctx, dom, flags) + }) +} + +func (o *observed) DomainGetState(ctx context.Context, dom libvirt.Domain, flags uint32) (state int32, reason int32, err error) { + err = o.call(ctx, "DomainGetState", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) { + state, reason, err = o.inner.DomainGetState(ctx, dom, flags) + return func() []any { return []any{"state", libvirt.DomainState(state).String(), "reason", reason} }, err + }) + return state, reason, err +} + +func (o *observed) DomainGetMetadata(ctx context.Context, dom libvirt.Domain, typ int32, uri libvirt.OptString, flags libvirt.DomainModificationImpact) (md string, err error) { + err = o.call(ctx, "DomainGetMetadata", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) { + md, err = o.inner.DomainGetMetadata(ctx, dom, typ, uri, flags) + return func() []any { return []any{"metadata", md} }, err + }) + return md, err +} + +func (o *observed) DomainInterfaceAddresses(ctx context.Context, dom libvirt.Domain, source uint32, flags uint32) (ifaces []libvirt.DomainInterface, err error) { + err = o.call(ctx, "DomainInterfaceAddresses", []id{domainID(dom.Name)}, func(ctx context.Context) (detail, error) { + ifaces, err = o.inner.DomainInterfaceAddresses(ctx, dom, source, flags) + return func() []any { return []any{"source", source, "addresses", interfaceAddresses(ifaces)} }, err + }) + return ifaces, err +} + +func (o *observed) ConnectListAllDomains(ctx context.Context, needResults int32, flags libvirt.ConnectListAllDomainsFlags) (doms []libvirt.Domain, n uint32, err error) { + err = o.call(ctx, "ConnectListAllDomains", nil, func(ctx context.Context) (detail, error) { + doms, n, err = o.inner.ConnectListAllDomains(ctx, needResults, flags) + return func() []any { return []any{"count", len(doms), "domains", domainNames(doms)} }, err + }) + return doms, n, err +} + +func (o *observed) StoragePoolLookupByName(ctx context.Context, name string) (pool libvirt.StoragePool, err error) { + err = o.call(ctx, "StoragePoolLookupByName", []id{poolID(name)}, func(ctx context.Context) (detail, error) { + pool, err = o.inner.StoragePoolLookupByName(ctx, name) + return nil, err + }) + return pool, err +} + +func (o *observed) StoragePoolRefresh(ctx context.Context, pool libvirt.StoragePool, flags uint32) error { + return o.call(ctx, "StoragePoolRefresh", []id{poolID(pool.Name)}, func(ctx context.Context) (detail, error) { + return nil, o.inner.StoragePoolRefresh(ctx, pool, flags) + }) +} + +func (o *observed) StoragePoolGetInfo(ctx context.Context, pool libvirt.StoragePool) (state uint8, capacity, allocation, available uint64, err error) { + err = o.call(ctx, "StoragePoolGetInfo", []id{poolID(pool.Name)}, func(ctx context.Context) (detail, error) { + state, capacity, allocation, available, err = o.inner.StoragePoolGetInfo(ctx, pool) + return func() []any { + return []any{"state", libvirt.StoragePoolState(state).String(), "capacity", capacity, "allocation", allocation, "available", available} + }, err + }) + return state, capacity, allocation, available, err +} + +func (o *observed) StorageVolLookupByName(ctx context.Context, pool libvirt.StoragePool, name string) (vol libvirt.StorageVol, err error) { + err = o.call(ctx, "StorageVolLookupByName", []id{poolID(pool.Name), volumeID(name)}, func(ctx context.Context) (detail, error) { + vol, err = o.inner.StorageVolLookupByName(ctx, pool, name) + return nil, err + }) + return vol, err +} + +func (o *observed) StorageVolGetPath(ctx context.Context, vol libvirt.StorageVol) (path string, err error) { + err = o.call(ctx, "StorageVolGetPath", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) { + path, err = o.inner.StorageVolGetPath(ctx, vol) + return func() []any { return []any{"path", path} }, err + }) + return path, err +} + +func (o *observed) StorageVolCreateXML(ctx context.Context, pool libvirt.StoragePool, xml string, flags libvirt.StorageVolCreateFlags) (vol libvirt.StorageVol, err error) { + err = o.call(ctx, "StorageVolCreateXML", []id{poolID(pool.Name)}, func(ctx context.Context) (detail, error) { + vol, err = o.inner.StorageVolCreateXML(ctx, pool, xml, flags) + return func() []any { return []any{"volume", vol.Name, "xml", xml} }, err + }) + return vol, err +} + +// StorageVolUpload and StorageVolDownload log the length only; the bodies +// are a seed ISO carrying CloudInit or a base image. +func (o *observed) StorageVolUpload(ctx context.Context, vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error { + return o.call(ctx, "StorageVolUpload", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) { + err := o.inner.StorageVolUpload(ctx, vol, r, offset, length, flags) + return func() []any { return []any{"offset", offset, "length", length} }, err + }) +} + +func (o *observed) StorageVolDownload(ctx context.Context, vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error { + return o.call(ctx, "StorageVolDownload", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) { + err := o.inner.StorageVolDownload(ctx, vol, w, offset, length, flags) + return func() []any { return []any{"offset", offset, "length", length} }, err + }) +} + +func (o *observed) StorageVolDelete(ctx context.Context, vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error { + return o.call(ctx, "StorageVolDelete", []id{poolID(vol.Pool), volumeID(vol.Name)}, func(ctx context.Context) (detail, error) { + return nil, o.inner.StorageVolDelete(ctx, vol, flags) + }) +} + +func (o *observed) ConnectGetLibVersion(ctx context.Context) (v uint64, err error) { + err = o.call(ctx, "ConnectGetLibVersion", nil, func(ctx context.Context) (detail, error) { + v, err = o.inner.ConnectGetLibVersion(ctx) + return func() []any { return []any{"libvirtVersion", v} }, err + }) + return v, err +} + +func (o *observed) ConnectGetHostname(ctx context.Context) (name string, err error) { + err = o.call(ctx, "ConnectGetHostname", nil, func(ctx context.Context) (detail, error) { + name, err = o.inner.ConnectGetHostname(ctx) + return func() []any { return []any{"hostname", name} }, err + }) + return name, err +} + +// Close is passed through: it is not a libvirt call and gets no span. +func (o *observed) Close() error { return o.inner.Close() } + +func domainNames(doms []libvirt.Domain) []string { + names := make([]string, len(doms)) + for i, d := range doms { + names[i] = d.Name + } + return names +} + +func interfaceAddresses(ifaces []libvirt.DomainInterface) []string { + var out []string + for _, iface := range ifaces { + for _, a := range iface.Addrs { + out = append(out, iface.Name+"="+a.Addr) + } + } + return out +} diff --git a/internal/provider/kvm2/observe_test.go b/internal/provider/kvm2/observe_test.go new file mode 100644 index 0000000..d787de2 --- /dev/null +++ b/internal/provider/kvm2/observe_test.go @@ -0,0 +1,334 @@ +package kvm2 + +import ( + "context" + "errors" + "reflect" + "strconv" + "strings" + "sync" + "testing" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr/funcr" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +// logCapture collects funcr output at a verbosity, safe for concurrent use. +type logCapture struct { + mu sync.Mutex + lines []string +} + +func (c *logCapture) all() []string { + c.mu.Lock() + defer c.mu.Unlock() + return append([]string(nil), c.lines...) +} + +func (c *logCapture) matching(substr string) []string { + var out []string + for _, l := range c.all() { + if strings.Contains(l, substr) { + out = append(out, l) + } + } + return out +} + +// observeFixture is a decorated fake with a span recorder and a log capture +// wired into ctx. +type observeFixture struct { + fake *fakeBackend + b backend + sr *tracetest.SpanRecorder + tp *sdktrace.TracerProvider + logs *logCapture +} + +func newObserveFixture(t *testing.T, verbosity int) (*observeFixture, context.Context) { + t.Helper() + sr := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + logs := &logCapture{} + log := funcr.New(func(prefix, args string) { + logs.mu.Lock() + defer logs.mu.Unlock() + logs.lines = append(logs.lines, prefix+" "+args) + }, funcr.Options{Verbosity: verbosity}) + fake := newFakeBackend() + fx := &observeFixture{ + fake: fake, + b: observe(fake, "kvm2-home", "beelink", tracing.WithTracerProvider(tp)), + sr: sr, + tp: tp, + logs: logs, + } + return fx, logf.IntoContext(context.Background(), log) +} + +func TestObserve_everySeamMethodInstrumented(t *testing.T) { + t.Parallel() + seam := reflect.TypeOf((*backend)(nil)).Elem() + for i := range seam.NumMethod() { + m := seam.Method(i) + if m.Name == "Close" { + continue + } + t.Run(m.Name, func(t *testing.T) { + t.Parallel() + fx, ctx := newObserveFixture(t, 1) + args := make([]reflect.Value, m.Type.NumIn()) + args[0] = reflect.ValueOf(ctx) + for j := 1; j < m.Type.NumIn(); j++ { + args[j] = reflect.Zero(m.Type.In(j)) + } + reflect.ValueOf(fx.b).MethodByName(m.Name).Call(args) + + ended := fx.sr.Ended() + if len(ended) != 1 { + t.Fatalf("got %d spans, want 1", len(ended)) + } + if got, want := ended[0].Name(), "kvm2."+m.Name; got != want { + t.Errorf("span name = %q, want %q", got, want) + } + if !hasAttr(ended[0].Attributes(), attribute.String("kvm2.host", "beelink")) { + t.Errorf("span attributes %v lack kvm2.host", ended[0].Attributes()) + } + lines := fx.logs.matching(`"msg"="libvirt call"`) + if len(lines) != 1 { + t.Fatalf("got %d 'libvirt call' lines, want 1:\n%s", len(lines), strings.Join(fx.logs.all(), "\n")) + } + for _, want := range []string{`"rpc"="` + m.Name + `"`, `"provider"="kvm2-home"`, `"host"="beelink"`, `"duration"=`, `"traceID"=`} { + if !strings.Contains(lines[0], want) { + t.Errorf("log line lacks %s: %s", want, lines[0]) + } + } + if fx.fake.callCount(m.Name) != 1 { + t.Errorf("inner %s called %d times, want 1", m.Name, fx.fake.callCount(m.Name)) + } + }) + } +} + +func TestObserve_childSpansUnderParent(t *testing.T) { + t.Parallel() + fx, ctx := newObserveFixture(t, 1) + ctx, parent := tracing.StartSpan(ctx, fx.tp.Tracer("test"), "provider.get") + _, _ = fx.b.DomainLookupByName(ctx, "proxy-x") + parent.End() + + var child sdktrace.ReadOnlySpan + for _, s := range fx.sr.Ended() { + if s.Name() == "kvm2.DomainLookupByName" { + child = s + } + } + if child == nil { + t.Fatal("no kvm2.DomainLookupByName span recorded") + } + if child.Parent().SpanID() != parent.SpanContext().SpanID() { + t.Errorf("child parent = %s, want %s", child.Parent().SpanID(), parent.SpanContext().SpanID()) + } + if child.SpanContext().TraceID() != parent.SpanContext().TraceID() { + t.Error("child is in a different trace than its parent") + } + if !hasAttr(child.Attributes(), attribute.String("kvm2.domain", "proxy-x")) { + t.Errorf("child attributes %v lack kvm2.domain", child.Attributes()) + } + lines := fx.logs.matching(`"msg"="libvirt call"`) + if len(lines) != 1 || !strings.Contains(lines[0], `"spanID"="`+child.SpanContext().SpanID().String()+`"`) { + t.Errorf("log line does not carry the child's spanID: %v", lines) + } +} + +func TestObserve_logsOneLinePerCallAtV1(t *testing.T) { + t.Parallel() + fx, ctx := newObserveFixture(t, 1) + fx.fake.addDomain("proxy-x", libvirt.DomainRunning, "") + pool, _ := fx.b.StoragePoolLookupByName(ctx, fakePoolName) + _, _ = fx.b.DomainLookupByName(ctx, "proxy-x") + _, _ = fx.b.StorageVolLookupByName(ctx, pool, "proxy-x.qcow2") + + lines := fx.logs.matching(`"msg"="libvirt call"`) + if len(lines) != 3 { + t.Fatalf("got %d lines, want 3:\n%s", len(lines), strings.Join(fx.logs.all(), "\n")) + } + if len(fx.logs.all()) != 3 { + t.Errorf("got %d lines in total, want only the 3 call lines:\n%s", len(fx.logs.all()), strings.Join(fx.logs.all(), "\n")) + } + wants := []string{ + `"rpc"="StoragePoolLookupByName" "pool"="default"`, + `"rpc"="DomainLookupByName" "domain"="proxy-x"`, + `"rpc"="StorageVolLookupByName" "pool"="default" "volume"="proxy-x.qcow2"`, + } + for i, want := range wants { + if !strings.Contains(lines[i], want) { + t.Errorf("line %d = %s, want %s in it", i, lines[i], want) + } + if !strings.Contains(lines[i], `"level"=1`) { + t.Errorf("line %d is not V(1): %s", i, lines[i]) + } + } +} + +func TestObserve_detailOnlyAtV2(t *testing.T) { + t.Parallel() + const xmlDoc = "proxy-xu1" + tests := []struct { + name string + verbosity int + wantXML bool + }{ + {name: "V1 has no xml", verbosity: 1, wantXML: false}, + {name: "V2 has xml", verbosity: 2, wantXML: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fx, ctx := newObserveFixture(t, tc.verbosity) + dom, err := fx.b.DomainDefineXML(ctx, xmlDoc) + if err != nil { + t.Fatalf("DomainDefineXML error = %v", err) + } + _, _ = fx.b.DomainGetMetadata(ctx, dom, int32(libvirt.DomainMetadataElement), libvirt.OptString{metadataURI}, 0) + _, _, _ = fx.b.ConnectListAllDomains(ctx, 1, 0) + all := strings.Join(fx.logs.all(), "\n") + if got := strings.Contains(all, ""); got != tc.wantXML { + t.Errorf("xml logged = %v, want %v:\n%s", got, tc.wantXML, all) + } + if got := strings.Contains(all, `"metadata"="`); got != tc.wantXML { + t.Errorf("metadata logged = %v, want %v:\n%s", got, tc.wantXML, all) + } + if got := strings.Contains(all, `"domains"=["proxy-x"]`); got != tc.wantXML { + t.Errorf("domain names logged = %v, want %v:\n%s", got, tc.wantXML, all) + } + if got := len(fx.logs.matching(`"msg"="libvirt call"`)); got != 3 { + t.Errorf("got %d call lines, want 3", got) + } + }) + } +} + +func TestObserve_uploadBodyNeverLogged(t *testing.T) { + t.Parallel() + const sentinel = "SENTINEL-CLOUDINIT-3c7a" + fx, ctx := newObserveFixture(t, 2) + pool, _ := fx.b.StoragePoolLookupByName(ctx, fakePoolName) + vol, err := fx.b.StorageVolCreateXML(ctx, pool, "proxy-x-seed.iso1024", 0) + if err != nil { + t.Fatalf("StorageVolCreateXML error = %v", err) + } + body := "#cloud-config\npassword: " + sentinel + "\n" + if err := fx.b.StorageVolUpload(ctx, vol, strings.NewReader(body), 0, uint64(len(body)), 0); err != nil { + t.Fatalf("StorageVolUpload error = %v", err) + } + var sink strings.Builder + if err := fx.b.StorageVolDownload(ctx, vol, &sink, 0, 0, 0); err != nil { + t.Fatalf("StorageVolDownload error = %v", err) + } + if sink.String() != body { + t.Errorf("downloaded %q, want the uploaded body", sink.String()) + } + all := strings.Join(fx.logs.all(), "\n") + if strings.Contains(all, sentinel) { + t.Errorf("upload body leaked into logs:\n%s", all) + } + uploads := fx.logs.matching(`"rpc"="StorageVolUpload"`) + if len(uploads) != 1 || !strings.Contains(uploads[0], `"length"=`+strconv.Itoa(len(body))) { + t.Errorf("upload line lacks length: %v", uploads) + } + for _, s := range fx.sr.Ended() { + for _, a := range s.Attributes() { + if strings.Contains(a.Value.Emit(), sentinel) { + t.Errorf("span %s attribute %s carries the body", s.Name(), a.Key) + } + } + } +} + +func TestObserve_errorFieldsOnFailure(t *testing.T) { + t.Parallel() + fx, ctx := newObserveFixture(t, 1) + _, err := fx.b.DomainLookupByName(ctx, "missing") + if !hasCode(err, libvirt.ErrNoDomain) { + t.Fatalf("error = %v, want ErrNoDomain through the decorator", err) + } + lines := fx.logs.matching(`"msg"="libvirt call"`) + if len(lines) != 1 { + t.Fatalf("got %d lines, want 1", len(lines)) + } + for _, want := range []string{`"error"="Domain not found`, `"libvirtCode"=42`, `"libvirtError"="ErrNoDomain"`} { + if !strings.Contains(lines[0], want) { + t.Errorf("line lacks %s: %s", want, lines[0]) + } + } + ended := fx.sr.Ended() + if len(ended) != 1 { + t.Fatalf("got %d spans, want 1", len(ended)) + } + if ended[0].Status().Code != codes.Error { + t.Errorf("span status = %v, want Error", ended[0].Status().Code) + } + if !hasAttr(ended[0].Attributes(), attribute.Int("kvm2.libvirt_code", 42)) { + t.Errorf("span attributes %v lack kvm2.libvirt_code=42", ended[0].Attributes()) + } + + fx.fake.failWith("ConnectGetHostname", errors.New("plain failure")) + _, _ = fx.b.ConnectGetHostname(ctx) + lines = fx.logs.matching(`"rpc"="ConnectGetHostname"`) + if len(lines) != 1 || strings.Contains(lines[0], "libvirtCode") || !strings.Contains(lines[0], `"error"="plain failure"`) { + t.Errorf("non-libvirt failure line = %v, want error without libvirt fields", lines) + } +} + +func TestObserve_v0Silent(t *testing.T) { + t.Parallel() + fx, ctx := newObserveFixture(t, 0) + fx.fake.addDomain("proxy-x", libvirt.DomainRunning, "") + _, _ = fx.b.DomainLookupByName(ctx, "proxy-x") + _, _ = fx.b.DomainLookupByName(ctx, "missing") + if lines := fx.logs.all(); len(lines) != 0 { + t.Errorf("got %d lines at V(0), want none:\n%s", len(lines), strings.Join(lines, "\n")) + } + if got := len(fx.sr.Ended()); got != 2 { + t.Errorf("got %d spans, want 2: spans do not depend on verbosity", got) + } +} + +func TestObserve_closeAndLivenessPassThrough(t *testing.T) { + t.Parallel() + fx, _ := newObserveFixture(t, 1) + if isDead(fx.b) { + t.Error("decorated live backend reported dead") + } + fx.fake.markDead() + if !isDead(fx.b) { + t.Error("decorated dead backend reported live") + } + if err := fx.b.Close(); err != nil || fx.fake.closed != 1 { + t.Errorf("Close() = %v, closed %d times, want nil and 1", err, fx.fake.closed) + } + if got := len(fx.sr.Ended()); got != 0 { + t.Errorf("Close produced %d spans, want none", got) + } + if lines := fx.logs.all(); len(lines) != 0 { + t.Errorf("Close produced log lines: %v", lines) + } +} + +func hasAttr(attrs []attribute.KeyValue, want attribute.KeyValue) bool { + for _, a := range attrs { + if a.Key == want.Key && a.Value == want.Value { + return true + } + } + return false +} diff --git a/internal/provider/kvm2/ssh.go b/internal/provider/kvm2/ssh.go new file mode 100644 index 0000000..b0faeca --- /dev/null +++ b/internal/provider/kvm2/ssh.go @@ -0,0 +1,216 @@ +package kvm2 + +import ( + "context" + "errors" + "fmt" + "io/fs" + "net" + "os" + "sync" + "time" + + "golang.org/x/crypto/ssh" + "golang.org/x/crypto/ssh/knownhosts" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// dialStage names the step of a dial that failed, so the Checker (spec +// entry 16) can pick a reason without parsing messages. +type dialStage string + +const ( + // stageCredentials: the key or known_hosts file is missing or unparsable. + stageCredentials dialStage = "credentials" + // stageTCP: the TCP connection to host.Address failed. + stageTCP dialStage = "tcp" + // stageSSH: the SSH handshake failed, including host key and auth. + stageSSH dialStage = "ssh" + // stageSocket: the forwarded channel to the libvirt socket was refused. + stageSocket dialStage = "socket" +) + +// dialError tags a dial failure with its stage and host. Unwrap keeps the +// cause visible, so classify still sees *knownhosts.KeyError, fs.ErrNotExist, +// the "unable to authenticate" text and *net.OpError. +type dialError struct { + stage dialStage + host string + err error +} + +func (e *dialError) Error() string { + return fmt.Sprintf("dialing host %s (%s): %v", e.host, e.stage, e.err) +} + +func (e *dialError) Unwrap() error { return e.err } + +// stageOf returns the failed stage of a dial error, if err is one. +func stageOf(err error) (dialStage, bool) { + var de *dialError + if !errors.As(err, &de) { + return "", false + } + return de.stage, true +} + +const ( + defaultKeepaliveInterval = 15 * time.Second + keepaliveRequest = "keepalive@openssh.com" +) + +// credentials is what one dial needs from the mounted files. They are read +// on every dial and never cached (spec entry 14), so rotating the Secret +// takes effect on the next connection. +type credentials struct { + signer ssh.Signer + hostKey ssh.HostKeyCallback +} + +// loadCredentials reads and parses the key and known_hosts files. A missing +// file keeps fs.ErrNotExist (Transient: the Secret has not landed); a file +// that does not parse is Permanent. Messages name the path and carry the +// parser's own error, never the file's content. +func loadCredentials(cfg provider.KVMConfig) (credentials, error) { + keyBytes, err := os.ReadFile(cfg.SSHKeyFile) + if err != nil { + return credentials{}, fmt.Errorf("reading ssh key file %s: %w", cfg.SSHKeyFile, err) + } + signer, err := ssh.ParsePrivateKey(keyBytes) + if err != nil { + return credentials{}, permanentf("parsing ssh key file %s: %v", cfg.SSHKeyFile, err) + } + if cfg.InsecureIgnoreHostKey { + return credentials{signer: signer, hostKey: ssh.InsecureIgnoreHostKey()}, nil + } + hostKey, err := knownhosts.New(cfg.KnownHostsFile) + if err != nil { + var pathErr *fs.PathError + if errors.As(err, &pathErr) { + return credentials{}, fmt.Errorf("reading known_hosts file %s: %w", cfg.KnownHostsFile, err) + } + return credentials{}, permanentf("parsing known_hosts file %s: %v", cfg.KnownHostsFile, err) + } + return credentials{signer: signer, hostKey: hostKey}, nil +} + +// sshSocket is the libvirt socket forwarded through one SSH connection. The +// embedded Conn is the channel; Close also closes the SSH client, which is +// what unblocks anything reading the channel. +type sshSocket struct { + net.Conn + client *ssh.Client + + closeOnce sync.Once + closeErr error +} + +func (s *sshSocket) Close() error { + s.closeOnce.Do(func() { + s.closeErr = errors.Join(s.Conn.Close(), s.client.Close()) + }) + return s.closeErr +} + +// keepalive sends keepalive@openssh.com every interval until stop is closed +// or a request fails or takes longer than the interval, and returns why it +// stopped (nil for stop). The caller marks its link dead on an error. +func (s *sshSocket) keepalive(interval time.Duration, stop <-chan struct{}) error { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-stop: + return nil + case <-ticker.C: + } + done := make(chan error, 1) + go func() { + _, _, err := s.client.SendRequest(keepaliveRequest, true, nil) + done <- err + }() + timer := time.NewTimer(interval) + select { + case err := <-done: + timer.Stop() + if err != nil { + return fmt.Errorf("keepalive: %w", err) + } + case <-timer.C: + return fmt.Errorf("keepalive: no reply within %s", interval) + case <-stop: + timer.Stop() + return nil + } + } +} + +// dialSSHSocket opens host's libvirt socket through SSH: credentials, TCP, +// handshake, channel. timeout bounds the whole sequence on top of ctx. Each +// failure is a *dialError naming its stage. Nothing here logs. +func dialSSHSocket(ctx context.Context, host provider.KVMHost, cfg provider.KVMConfig, timeout time.Duration) (*sshSocket, error) { + if timeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, timeout) + defer cancel() + } + creds, err := loadCredentials(cfg) + if err != nil { + return nil, &dialError{stage: stageCredentials, host: host.Name, err: err} + } + var nd net.Dialer + tcp, err := nd.DialContext(ctx, "tcp", host.Address) + if err != nil { + return nil, &dialError{stage: stageTCP, host: host.Name, err: err} + } + clientCfg := &ssh.ClientConfig{ + User: host.User, + Auth: []ssh.AuthMethod{ssh.PublicKeys(creds.signer)}, + HostKeyCallback: creds.hostKey, + Timeout: timeout, + } + client, err := await(ctx, tcp.Close, func() (*ssh.Client, error) { + c, chans, reqs, err := ssh.NewClientConn(tcp, host.Address, clientCfg) + if err != nil { + return nil, err + } + return ssh.NewClient(c, chans, reqs), nil + }) + if err != nil { + _ = tcp.Close() + return nil, &dialError{stage: stageSSH, host: host.Name, err: err} + } + conn, err := await(ctx, func() error { return client.Close() }, func() (net.Conn, error) { + return client.Dial("unix", host.Socket) + }) + if err != nil { + _ = client.Close() + return nil, &dialError{stage: stageSocket, host: host.Name, err: fmt.Errorf("opening %s: %w", host.Socket, err)} + } + return &sshSocket{Conn: conn, client: client}, nil +} + +// await runs a blocking step in a goroutine. When ctx ends first it calls +// abort, which closes the transport the step is blocked on, waits for the +// step to give up, and returns ctx's error instead. +func await[T any](ctx context.Context, abort func() error, step func() (T, error)) (T, error) { + type result struct { + v T + err error + } + done := make(chan result, 1) + go func() { + v, err := step() + done <- result{v, err} + }() + select { + case r := <-done: + return r.v, r.err + case <-ctx.Done(): + _ = abort() + <-done + var zero T + return zero, ctx.Err() + } +} diff --git a/internal/provider/kvm2/ssh_test.go b/internal/provider/kvm2/ssh_test.go new file mode 100644 index 0000000..35db011 --- /dev/null +++ b/internal/provider/kvm2/ssh_test.go @@ -0,0 +1,343 @@ +package kvm2 + +import ( + "context" + "errors" + "io/fs" + "net" + "os" + "strings" + "testing" + "time" + + "golang.org/x/crypto/ssh/knownhosts" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// sshFixture is one in-process server with a client key it accepts and a +// known_hosts file naming its host key. +type sshFixture struct { + server *testSSHServer + socket string + cfg provider.KVMConfig + host provider.KVMHost +} + +func newSSHFixture(t *testing.T) *sshFixture { + t.Helper() + dir := t.TempDir() + socket := startEchoSocket(t) + keyPath, pub := writeKeyFile(t, dir, "id_ed25519") + server := startSSHServer(t, pub, socket) + return &sshFixture{ + server: server, + socket: socket, + cfg: provider.KVMConfig{ + SSHKeyFile: keyPath, + KnownHostsFile: writeKnownHosts(t, dir, "known_hosts", server.addr, server.hostKey), + }, + host: provider.KVMHost{Name: "beelink", Address: server.addr, User: "virt", Socket: socket}, + } +} + +const dialTestTimeout = 5 * time.Second + +func TestDial_forwardsUnixSocket(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + for round := range 2 { + sock, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + if err != nil { + t.Fatalf("round %d: dial error = %v", round, err) + } + msg := []byte("ping over libvirt-sock") + if _, err := sock.Write(msg); err != nil { + t.Fatalf("round %d: write: %v", round, err) + } + buf := make([]byte, len(msg)) + _ = sock.SetReadDeadline(time.Now().Add(dialTestTimeout)) + if _, err := readFull(sock, buf); err != nil { + t.Fatalf("round %d: read: %v", round, err) + } + if string(buf) != string(msg) { + t.Errorf("round %d: echoed %q, want %q", round, buf, msg) + } + if err := sock.Close(); err != nil { + t.Errorf("round %d: close error = %v", round, err) + } + if err := sock.Close(); err != nil { + t.Errorf("round %d: second close error = %v, want nil", round, err) + } + } + for _, p := range fx.server.requestedPaths() { + if p != fx.socket { + t.Errorf("server saw channel for %q, want %q", p, fx.socket) + } + } +} + +func readFull(sock *sshSocket, buf []byte) (int, error) { + n := 0 + for n < len(buf) { + m, err := sock.Read(buf[n:]) + n += m + if err != nil { + return n, err + } + } + return n, nil +} + +func TestDial_unknownHostKeyRejected(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.KnownHostsFile = writeKnownHosts(t, t.TempDir(), "known_hosts", "other.example:22", fx.server.hostKey) + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageSSH, provider.ErrPermanent) + var keyErr *knownhosts.KeyError + if !errors.As(err, &keyErr) { + t.Fatalf("error = %v, want *knownhosts.KeyError", err) + } + if len(keyErr.Want) != 0 { + t.Errorf("KeyError.Want = %v, want empty (host unknown)", keyErr.Want) + } +} + +func TestDial_mismatchedHostKeyRejected(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.KnownHostsFile = writeKnownHosts(t, t.TempDir(), "known_hosts", fx.server.addr, otherPublicKey(t)) + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageSSH, provider.ErrPermanent) + var keyErr *knownhosts.KeyError + if !errors.As(err, &keyErr) { + t.Fatalf("error = %v, want *knownhosts.KeyError", err) + } + if len(keyErr.Want) == 0 { + t.Error("KeyError.Want empty, want the mismatching known key") + } +} + +func TestDial_insecureIgnoreHostKeyAccepts(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.KnownHostsFile = "" + fx.cfg.InsecureIgnoreHostKey = true + sock, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + if err != nil { + t.Fatalf("dial error = %v, want nil with host keys ignored", err) + } + _ = sock.Close() +} + +func TestDial_missingKeyFileIsTransient(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.SSHKeyFile = fx.cfg.SSHKeyFile + ".missing" + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageCredentials, provider.ErrTransient) + if !errors.Is(err, fs.ErrNotExist) { + t.Errorf("error = %v, want fs.ErrNotExist", err) + } + if !strings.Contains(err.Error(), fx.cfg.SSHKeyFile) { + t.Errorf("error = %q, want the path in it", err) + } +} + +func TestDial_missingKnownHostsIsTransient(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.KnownHostsFile = fx.cfg.KnownHostsFile + ".missing" + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageCredentials, provider.ErrTransient) + if !errors.Is(err, fs.ErrNotExist) { + t.Errorf("error = %v, want fs.ErrNotExist", err) + } +} + +func TestDial_unparsableKeyIsPermanent(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.SSHKeyFile = writeFile(t, t.TempDir(), "id_bad", "this is not a key\n") + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageCredentials, provider.ErrPermanent) + if !strings.Contains(err.Error(), "parsing ssh key file "+fx.cfg.SSHKeyFile) { + t.Errorf("error = %q, want 'parsing ssh key file '", err) + } +} + +func TestDial_unparsableKnownHostsIsPermanent(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.KnownHostsFile = writeFile(t, t.TempDir(), "known_hosts", "host.example ssh-ed25519 not-base64!\n") + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageCredentials, provider.ErrPermanent) + if !strings.Contains(err.Error(), "parsing known_hosts file "+fx.cfg.KnownHostsFile) { + t.Errorf("error = %q, want 'parsing known_hosts file '", err) + } +} + +func TestDial_wrongKeyIsUnauthorized(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.cfg.SSHKeyFile, _ = writeKeyFile(t, t.TempDir(), "id_other") + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageSSH, provider.ErrPermanent) + if !strings.Contains(err.Error(), sshAuthFailed) { + t.Errorf("error = %q, want %q in it", err, sshAuthFailed) + } +} + +func TestDial_handshakeTimeout(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.host.Address = silentListener(t) + const budget = 200 * time.Millisecond + start := time.Now() + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, budget) + elapsed := time.Since(start) + assertDialFailure(t, err, stageSSH, provider.ErrTransient) + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("error = %v, want context.DeadlineExceeded", err) + } + if elapsed > budget+time.Second { + t.Errorf("dial took %v, want about the %v budget", elapsed, budget) + } +} + +func TestDial_cancelledContext(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.host.Address = silentListener(t) + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + start := time.Now() + _, err := dialSSHSocket(ctx, fx.host, fx.cfg, dialTestTimeout) + if time.Since(start) > time.Second { + t.Errorf("dial took %v after cancel", time.Since(start)) + } + assertDialFailure(t, err, stageSSH, provider.ErrTransient) + if !errors.Is(err, context.Canceled) { + t.Errorf("error = %v, want context.Canceled", err) + } +} + +func TestDial_refusedIsTransient(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.host.Address = closedPort(t) + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageTCP, provider.ErrTransient) + var opErr *net.OpError + if !errors.As(err, &opErr) { + t.Errorf("error = %v, want *net.OpError", err) + } +} + +func TestDial_socketRefusedIsSocketStage(t *testing.T) { + t.Parallel() + fx := newSSHFixture(t) + fx.host.Socket = "/run/nonexistent/libvirt-sock" + _, err := dialSSHSocket(context.Background(), fx.host, fx.cfg, dialTestTimeout) + assertDialFailure(t, err, stageSocket, provider.ErrTransient) + if !strings.Contains(err.Error(), fx.host.Socket) { + t.Errorf("error = %q, want the socket path in it", err) + } +} + +func TestSSH_keyMaterialNeverInErrorsOrLogs(t *testing.T) { + t.Parallel() + const keySentinel = "SENTINEL-KEY-4f9c2b" + const hostsSentinel = "SENTINEL-HOSTS-8a1d7e" + fx := newSSHFixture(t) + dir := t.TempDir() + tests := []struct { + name string + cfg func(c *provider.KVMConfig) + }{ + { + name: "bad key file", + cfg: func(c *provider.KVMConfig) { + c.SSHKeyFile = writeFile(t, dir, "id_bad", "-----BEGIN NOT A KEY-----\n"+keySentinel+"\n-----END NOT A KEY-----\n") + }, + }, + { + name: "bad known_hosts file", + cfg: func(c *provider.KVMConfig) { + c.KnownHostsFile = writeFile(t, dir, "known_hosts_bad", "beelink ssh-ed25519 "+hostsSentinel+"\n") + }, + }, + { + name: "good key, host unknown", + cfg: func(c *provider.KVMConfig) { + c.KnownHostsFile = writeFile(t, dir, "known_hosts_other", knownhosts.Line([]string{"other.example:22"}, fx.server.hostKey)+"\n") + }, + }, + } + keyBytes := readFileT(t, fx.cfg.SSHKeyFile) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + cfg := fx.cfg + tc.cfg(&cfg) + _, err := dialSSHSocket(context.Background(), fx.host, cfg, dialTestTimeout) + if err == nil { + t.Fatal("dial succeeded, want an error") + } + msg := err.Error() + for _, s := range []string{keySentinel, hostsSentinel} { + if strings.Contains(msg, s) { + t.Errorf("error %q contains sentinel %q", msg, s) + } + } + if strings.Contains(msg, keyBytes) || strings.Contains(msg, "PRIVATE KEY") { + t.Errorf("error %q contains key material", msg) + } + }) + } +} + +func readFileT(t *testing.T, path string) string { + t.Helper() + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("reading %s: %v", path, err) + } + return strings.TrimSpace(string(b)) +} + +func assertDialFailure(t *testing.T, err error, wantStage dialStage, wantClass error) { + t.Helper() + if err == nil { + t.Fatal("dial succeeded, want an error") + } + stage, ok := stageOf(err) + if !ok { + t.Fatalf("error %v is not a *dialError", err) + } + if stage != wantStage { + t.Errorf("stage = %q, want %q (error: %v)", stage, wantStage, err) + } + if got := classify("get", err); !errors.Is(got, wantClass) { + t.Errorf("classify() = %v, want %v (error: %v)", got, wantClass, err) + } +} + +func TestDialError_unwrapsToCause(t *testing.T) { + t.Parallel() + cause := &knownhosts.KeyError{} + err := &dialError{stage: stageSSH, host: "beelink", err: cause} + if !errors.Is(err, cause) { + t.Error("errors.Is(dialError, cause) = false") + } + if !strings.Contains(err.Error(), "beelink") || !strings.Contains(err.Error(), "(ssh)") { + t.Errorf("Error() = %q, want the host and the stage", err) + } + if _, ok := stageOf(errors.New("plain")); ok { + t.Error("stageOf(plain error) reported a stage") + } +} diff --git a/internal/provider/kvm2/sshserver_test.go b/internal/provider/kvm2/sshserver_test.go new file mode 100644 index 0000000..9053335 --- /dev/null +++ b/internal/provider/kvm2/sshserver_test.go @@ -0,0 +1,289 @@ +package kvm2 + +import ( + "bytes" + "crypto/ed25519" + "crypto/rand" + "encoding/pem" + "errors" + "io" + "net" + "os" + "path/filepath" + "sync" + "testing" + + "golang.org/x/crypto/ssh" + "golang.org/x/crypto/ssh/knownhosts" +) + +// testSSHServer is an in-process x/crypto/ssh server that accepts one +// authorised public key and forwards direct-streamlocal@openssh.com channels +// for socketPath to that unix socket, rejecting every other channel. +type testSSHServer struct { + addr string + hostKey ssh.PublicKey + socketPath string + + listener net.Listener + mu sync.Mutex + conns []net.Conn + wg sync.WaitGroup + stopOnce sync.Once + + reqMu sync.Mutex + requested []string +} + +func startSSHServer(t *testing.T, authorized ssh.PublicKey, socketPath string) *testSSHServer { + t.Helper() + _, hostPriv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generating host key: %v", err) + } + hostSigner, err := ssh.NewSignerFromKey(hostPriv) + if err != nil { + t.Fatalf("host signer: %v", err) + } + cfg := &ssh.ServerConfig{ + PublicKeyCallback: func(_ ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) { + if bytes.Equal(key.Marshal(), authorized.Marshal()) { + return &ssh.Permissions{}, nil + } + return nil, errors.New("key not authorised") + }, + } + cfg.AddHostKey(hostSigner) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + s := &testSSHServer{addr: ln.Addr().String(), hostKey: hostSigner.PublicKey(), socketPath: socketPath, listener: ln} + s.wg.Add(1) + go s.serve(cfg) + t.Cleanup(s.stop) + return s +} + +func (s *testSSHServer) serve(cfg *ssh.ServerConfig) { + defer s.wg.Done() + for { + conn, err := s.listener.Accept() + if err != nil { + return + } + s.mu.Lock() + s.conns = append(s.conns, conn) + s.mu.Unlock() + s.wg.Add(1) + go s.handle(conn, cfg) + } +} + +func (s *testSSHServer) handle(conn net.Conn, cfg *ssh.ServerConfig) { + defer s.wg.Done() + sc, chans, reqs, err := ssh.NewServerConn(conn, cfg) + if err != nil { + _ = conn.Close() + return + } + defer sc.Close() + go ssh.DiscardRequests(reqs) + for nc := range chans { + if nc.ChannelType() != "direct-streamlocal@openssh.com" { + _ = nc.Reject(ssh.UnknownChannelType, "only direct-streamlocal@openssh.com is served") + continue + } + var open struct { + SocketPath string + Reserved0 string + Reserved1 uint32 + } + if err := ssh.Unmarshal(nc.ExtraData(), &open); err != nil { + _ = nc.Reject(ssh.ConnectionFailed, "bad direct-streamlocal payload") + continue + } + s.reqMu.Lock() + s.requested = append(s.requested, open.SocketPath) + s.reqMu.Unlock() + if open.SocketPath != s.socketPath { + _ = nc.Reject(ssh.ConnectionFailed, "no such socket") + continue + } + target, err := net.Dial("unix", s.socketPath) + if err != nil { + _ = nc.Reject(ssh.ConnectionFailed, "socket unavailable") + continue + } + ch, chReqs, err := nc.Accept() + if err != nil { + _ = target.Close() + continue + } + go ssh.DiscardRequests(chReqs) + s.wg.Add(1) + go func() { + defer s.wg.Done() + proxy(ch, target) + }() + } +} + +func proxy(ch ssh.Channel, target net.Conn) { + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _, _ = io.Copy(target, ch) + _ = target.Close() + }() + go func() { + defer wg.Done() + _, _ = io.Copy(ch, target) + _ = ch.Close() + }() + wg.Wait() +} + +func (s *testSSHServer) stop() { + s.stopOnce.Do(func() { + _ = s.listener.Close() + s.mu.Lock() + for _, c := range s.conns { + _ = c.Close() + } + s.mu.Unlock() + s.wg.Wait() + }) +} + +func (s *testSSHServer) requestedPaths() []string { + s.reqMu.Lock() + defer s.reqMu.Unlock() + return append([]string(nil), s.requested...) +} + +// startEchoSocket listens on a unix socket that echoes what it reads. The +// path is kept short: unix socket paths are limited to about 100 bytes and +// t.TempDir() names are long. +func startEchoSocket(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp("", "kvm2") + if err != nil { + t.Fatalf("temp dir: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + path := filepath.Join(dir, "libvirt-sock") + ln, err := net.Listen("unix", path) + if err != nil { + t.Fatalf("listen unix: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { + _, _ = io.Copy(c, c) + _ = c.Close() + }() + } + }() + return path +} + +// silentListener accepts TCP connections and never speaks, for handshake +// timeouts. It returns the address. +func silentListener(t *testing.T) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + var conns []net.Conn + var mu sync.Mutex + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + mu.Lock() + conns = append(conns, c) + mu.Unlock() + } + }() + t.Cleanup(func() { + _ = ln.Close() + mu.Lock() + defer mu.Unlock() + for _, c := range conns { + _ = c.Close() + } + }) + return ln.Addr().String() +} + +// closedPort returns an address nothing listens on. +func closedPort(t *testing.T) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := ln.Addr().String() + _ = ln.Close() + return addr +} + +// writeKeyFile generates an ed25519 key, writes it in OpenSSH format and +// returns the path and the public half. +func writeKeyFile(t *testing.T, dir, name string) (string, ssh.PublicKey) { + t.Helper() + pub, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generating key: %v", err) + } + block, err := ssh.MarshalPrivateKey(priv, "kvm2 test") + if err != nil { + t.Fatalf("marshalling key: %v", err) + } + path := writeFile(t, dir, name, string(pem.EncodeToMemory(block))) + sshPub, err := ssh.NewPublicKey(pub) + if err != nil { + t.Fatalf("public key: %v", err) + } + return path, sshPub +} + +// writeKnownHosts writes one known_hosts line for addr and key. +func writeKnownHosts(t *testing.T, dir, name, addr string, key ssh.PublicKey) string { + t.Helper() + return writeFile(t, dir, name, knownhosts.Line([]string{addr}, key)+"\n") +} + +func writeFile(t *testing.T, dir, name, content string) string { + t.Helper() + path := filepath.Join(dir, name) + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatalf("writing %s: %v", path, err) + } + return path +} + +// otherPublicKey returns a public key unrelated to any file, for mismatch +// cases. +func otherPublicKey(t *testing.T) ssh.PublicKey { + t.Helper() + pub, _, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatalf("generating key: %v", err) + } + sshPub, err := ssh.NewPublicKey(pub) + if err != nil { + t.Fatalf("public key: %v", err) + } + return sshPub +}