From 6322184d50541358af2db34e64a6ea1bff8d2e30 Mon Sep 17 00:00:00 2001 From: kacerr Date: Sat, 5 Sep 2026 03:27:57 +0000 Subject: [PATCH] Add KVM provider transport: SSH-forwarded libvirt connections and error classification Per-host connection pool over SSH with the libvirt unix socket forwarded through the session, lazy dial and redial after disconnect, context cancellation for go-libvirt's context-less RPCs, handshake deadline and SSH keepalives. Error classification onto the provider taxonomy. Tested with a fake dialer and an in-process SSH server that forwards unix sockets. Step 3 of docs/plans/2026-09-04-1404-kvm-provider.md. Co-Authored-By: Claude Fable 5.1 --- .../2026-09-04-1404-kvm-provider.md | 54 +- go.mod | 3 +- go.sum | 2 + internal/provider/kvm/conn.go | 301 +++++++++ internal/provider/kvm/conn_test.go | 569 ++++++++++++++++++ internal/provider/kvm/errors.go | 65 ++ internal/provider/kvm/errors_test.go | 67 +++ 7 files changed, 1059 insertions(+), 2 deletions(-) create mode 100644 internal/provider/kvm/conn.go create mode 100644 internal/provider/kvm/conn_test.go create mode 100644 internal/provider/kvm/errors.go create mode 100644 internal/provider/kvm/errors_test.go diff --git a/docs/plans-executions/2026-09-04-1404-kvm-provider.md b/docs/plans-executions/2026-09-04-1404-kvm-provider.md index b630134..533ebf8 100644 --- a/docs/plans-executions/2026-09-04-1404-kvm-provider.md +++ b/docs/plans-executions/2026-09-04-1404-kvm-provider.md @@ -5,7 +5,7 @@ Plan: [2026-09-04-1404-kvm-provider.md](../plans/2026-09-04-1404-kvm-provider.md - [x] Step 0 — Save plan + branch `feat/kvm-provider` - [x] Step 1 — Config (`KVMConfig` & friends, `validate()` case, tests) - [x] Step 2 — Pure builders (`domainxml.go`, `volume.go`, `seed.go`, `id.go`, `metadata.go`) -- [ ] Step 3 — Transport + errors (`conn.go`, `errors.go`) +- [x] Step 3 — Transport + errors (`conn.go`, `errors.go`) - [ ] Step 3b — Image cache (`image.go`) - [ ] Step 4 — Provider methods (`kvm.go`) over a fake API - [ ] Step 5 — Wiring, manifests, samples, docs, CHANGELOG @@ -99,3 +99,55 @@ instead of mutating the config in place, so the registry's shared `qemu-guest-agent` via `packages:` + `runcmd`, which means the first boot needs network access to the distro mirrors; a Proxy on an isolated bridge would still work (ARP discovery), just without the agent. + +## Step 3 — Transport + errors + +`conn.go` holds the `libvirtAPI` seam (17 methods, signatures as in the +plan), a per-host connection pool, and the SSH dialer; `errors.go` holds +`classify`, `isLibvirtError`, `hasCode` and `logAPIError`. Both are +unit-tested; the whole package still runs in about a second under `-race`. + +The one thing the plan did not anticipate: **go-libvirt RPCs take no +context and never time out**. A hung hypervisor would pin a reconcile worker +forever. So the pool exposes `call(ctx, host, fn)` rather than handing out +raw clients: `fn` runs in a goroutine, and if `ctx` ends first the pool +closes the SSH connection out from under it (which makes the blocked RPC +return) and reports `ctx.Err()`. The same helper (`awaitOrCancel`) bounds +`lv.Connect()` during dial. Two more liveness measures for the same reason: +the SSH handshake gets a deadline (ctx's, or 30 s), and each connection runs +an OpenSSH `keepalive@openssh.com` loop every 15 s that closes the client +when a request errors or goes unanswered for one interval. + +Connection-drop policy in `call`: any error that is *not* a `libvirt.Error` +(so io.EOF, "connection closed", SSH failures) drops the cached connection +and the next call redials; a daemon-reported `libvirt.Error` keeps it. Dials +are serialised per host (concurrent Creates on one host share a dial) but +hosts don't block each other. + +As Step 1 noted, the dialer uses `libvirt.NewWithDialer` with a +`socket.Dialer` that wraps `ssh.Client.Dial("unix", socket)` rather than the +deprecated `New(net.Conn)`. + +Tests: the pool is exercised through a fake dialer (reuse, redial after +`Disconnected`, drop on transport error, keep on libvirt error, cancel +releases a blocked call, concurrent callers share one dial, hosts +independent). The SSH layer is tested against an in-process +`x/crypto/ssh` server that serves `direct-streamlocal@openssh.com` channels +by dialing the requested unix path (an echo socket standing in for +`libvirt-sock`), with keys generated per test and a real `known_hosts` file +written via `knownhosts.Line`. Covered: forward round-trip and the socket +path the server saw, unknown host key rejected with `knownhosts.KeyError`, +`insecureIgnoreHostKey`, wrong client key, cancelled ctx, missing key or +known_hosts file, and keepalive both closing a stalled client and leaving a +healthy one alone. The libvirt handshake (`Connect()`) itself is *not* +covered here — faking libvirt's XDR RPC just for that is not worth it; the +Step 6 integration test covers it against a real host. + +`go mod tidy` promoted `go-libvirt` and `golang.org/x/crypto` to direct +dependencies. + +Worth noting: `classify` maps `ErrOperationInvalid` (e.g. destroying a +domain that is already off) to Transient; Delete in Step 4 must check that +code itself via `hasCode` and treat it as success rather than relying on +the classifier. Nothing maps to `ErrQuotaExceeded` — libvirt has no quota +concept; an out-of-space pool surfaces as a Transient `ErrInternalError`. diff --git a/go.mod b/go.mod index de8c5e3..8e6cf30 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.0 require ( cloud.google.com/go/compute v1.65.0 + github.com/digitalocean/go-libvirt v0.0.0-20260814190004-1a83157e1858 github.com/go-logr/logr v1.4.4 github.com/kdomanski/iso9660 v0.4.0 github.com/onsi/ginkgo/v2 v2.27.4 @@ -16,6 +17,7 @@ require ( go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.45.0 go.opentelemetry.io/otel/sdk v1.45.0 go.opentelemetry.io/otel/trace v1.45.0 + golang.org/x/crypto v0.54.0 google.golang.org/api v0.292.0 google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af k8s.io/api v0.36.0 @@ -81,7 +83,6 @@ require ( go.uber.org/zap v1.27.1 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/crypto v0.54.0 // indirect golang.org/x/exp v0.0.0-20251219203646-944ab1f22d93 // indirect golang.org/x/mod v0.37.0 // indirect golang.org/x/net v0.57.0 // indirect diff --git a/go.sum b/go.sum index 807f975..b0c2886 100644 --- a/go.sum +++ b/go.sum @@ -28,6 +28,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/digitalocean/go-libvirt v0.0.0-20260814190004-1a83157e1858 h1:8xCFt73OddCD5WXxoYidJBWp3bJMC+CpE7Zwf5tcwk8= +github.com/digitalocean/go-libvirt v0.0.0-20260814190004-1a83157e1858/go.mod h1:qb0Ofa71d3oXARQf633h2tNaeBxLsVxuDp+jcsVO2+4= github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes= github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= github.com/evanphx/json-patch v0.5.2 h1:xVCHIVMUu1wtM/VkR9jVZ45N3FhZfYMMYGorLCR8P3k= diff --git a/internal/provider/kvm/conn.go b/internal/provider/kvm/conn.go new file mode 100644 index 0000000..b0034b0 --- /dev/null +++ b/internal/provider/kvm/conn.go @@ -0,0 +1,301 @@ +package kvm + +import ( + "context" + "fmt" + "io" + "net" + "os" + "sync" + "time" + + libvirt "github.com/digitalocean/go-libvirt" + "golang.org/x/crypto/ssh" + "golang.org/x/crypto/ssh/knownhosts" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +const ( + defaultHandshakeTimeout = 30 * time.Second + keepaliveInterval = 15 * time.Second +) + +// libvirtAPI is the slice of *libvirt.Libvirt the provider uses, and the +// test seam: the fake in kvm_test.go implements exactly this. +type libvirtAPI interface { + DomainLookupByName(name string) (libvirt.Domain, error) + DomainDefineXML(xml string) (libvirt.Domain, error) + DomainCreate(d libvirt.Domain) error + DomainDestroy(d libvirt.Domain) error + DomainUndefineFlags(d libvirt.Domain, flags libvirt.DomainUndefineFlagsValues) error + DomainGetState(d libvirt.Domain, flags uint32) (state int32, reason int32, err error) + DomainGetMetadata(d libvirt.Domain, typ int32, uri libvirt.OptString, flags libvirt.DomainModificationImpact) (string, error) + DomainInterfaceAddresses(d libvirt.Domain, source uint32, flags uint32) ([]libvirt.DomainInterface, error) + ConnectListAllDomains(needResults int32, flags libvirt.ConnectListAllDomainsFlags) ([]libvirt.Domain, uint32, error) + StoragePoolLookupByName(name string) (libvirt.StoragePool, error) + StoragePoolRefresh(pool libvirt.StoragePool, flags uint32) error + StorageVolLookupByName(pool libvirt.StoragePool, name string) (libvirt.StorageVol, error) + StorageVolGetPath(vol libvirt.StorageVol) (string, error) + StorageVolCreateXML(pool libvirt.StoragePool, xml string, flags libvirt.StorageVolCreateFlags) (libvirt.StorageVol, error) + StorageVolUpload(vol libvirt.StorageVol, r io.Reader, offset, length uint64, flags libvirt.StorageVolUploadFlags) error + StorageVolDownload(vol libvirt.StorageVol, w io.Writer, offset, length uint64, flags libvirt.StorageVolDownloadFlags) error + StorageVolDelete(vol libvirt.StorageVol, flags libvirt.StorageVolDeleteFlags) error +} + +// hostClient is one live connection to a hypervisor. +type hostClient struct { + api libvirtAPI + // done is closed once the connection is known to be lost. A nil channel + // never fires, which suits fakes. + done <-chan struct{} + close func() +} + +func (c *hostClient) alive() bool { + select { + case <-c.done: + return false + default: + return true + } +} + +// hostDialer opens a connection to one host. The SSH implementation is the +// real one; tests inject their own. +type hostDialer func(ctx context.Context, host provider.KVMHost) (*hostClient, error) + +// hostPool caches one connection per host, dialing lazily and redialing +// after the connection drops. Dials for different hosts don't block each +// other; concurrent callers for the same host share a single dial. +type hostPool struct { + dial hostDialer + + mu sync.Mutex + entries map[string]*hostEntry +} + +type hostEntry struct { + mu sync.Mutex + client *hostClient +} + +func newHostPool(dial hostDialer) *hostPool { + return &hostPool{dial: dial, entries: make(map[string]*hostEntry)} +} + +func (p *hostPool) entry(name string) *hostEntry { + p.mu.Lock() + defer p.mu.Unlock() + e, ok := p.entries[name] + if !ok { + e = &hostEntry{} + p.entries[name] = e + } + return e +} + +func (p *hostPool) get(ctx context.Context, host provider.KVMHost) (*hostClient, error) { + e := p.entry(host.Name) + e.mu.Lock() + defer e.mu.Unlock() + if e.client != nil { + if e.client.alive() { + return e.client, nil + } + e.client.close() + e.client = nil + } + c, err := p.dial(ctx, host) + if err != nil { + return nil, fmt.Errorf("connecting to host %s (%s): %w", host.Name, host.Address, err) + } + e.client = c + return c, nil +} + +// drop closes and forgets c if it is still the cached client for host, so a +// connection that was already replaced by a redial is left alone. +func (p *hostPool) drop(host string, c *hostClient) { + e := p.entry(host) + e.mu.Lock() + defer e.mu.Unlock() + if e.client == c { + e.client = nil + } + c.close() +} + +// call runs fn against host's connection. go-libvirt RPCs take no context +// and block until the daemon answers, so when ctx ends first the connection +// is closed — which unblocks fn with an error — and ctx.Err() is returned. +// A transport error (anything that is not a libvirt.Error) also drops the +// cached connection so the next call redials instead of reusing a dead +// link. +func (p *hostPool) call(ctx context.Context, host provider.KVMHost, fn func(api libvirtAPI) error) error { + c, err := p.get(ctx, host) + if err != nil { + return err + } + err = awaitOrCancel(ctx, func() error { return fn(c.api) }, func() { p.drop(host.Name, c) }) + if err != nil && !isLibvirtError(err) { + p.drop(host.Name, c) + } + return err +} + +// awaitOrCancel runs fn in the background and returns its result, unless ctx +// ends first: then abort is invoked (it must make fn return eventually, e.g. +// by closing the connection fn is blocked on) and ctx.Err() is returned. The +// abandoned fn's result is discarded. +func awaitOrCancel(ctx context.Context, fn func() error, abort func()) error { + errc := make(chan error, 1) + go func() { errc <- fn() }() + select { + case err := <-errc: + return err + case <-ctx.Done(): + abort() + return ctx.Err() + } +} + +// sshAuth holds the pieces of the SSH client config shared by every host: +// the private key and the host-key policy, both loaded once at startup so a +// missing or unreadable file fails the operator fast rather than on the +// first Create. +type sshAuth struct { + signer ssh.Signer + hostKey ssh.HostKeyCallback +} + +func newSSHAuth(cfg provider.KVMConfig) (*sshAuth, error) { + key, err := os.ReadFile(cfg.SSHKeyFile) + if err != nil { + return nil, fmt.Errorf("reading kvm.sshKeyFile: %w", err) + } + signer, err := ssh.ParsePrivateKey(key) + if err != nil { + return nil, fmt.Errorf("parsing kvm.sshKeyFile: %w", err) + } + hostKey := ssh.InsecureIgnoreHostKey() //nolint:gosec // opt-in via config, documented as insecure + if !cfg.InsecureIgnoreHostKey { + hostKey, err = knownhosts.New(cfg.KnownHostsFile) + if err != nil { + return nil, fmt.Errorf("loading kvm.knownHostsFile: %w", err) + } + } + return &sshAuth{signer: signer, hostKey: hostKey}, nil +} + +// dialer returns the production hostDialer: SSH to the host, forward its +// libvirt unix socket through the connection, and open a libvirt session +// over that. +func (a *sshAuth) dialer() hostDialer { + return func(ctx context.Context, host provider.KVMHost) (*hostClient, error) { + client, err := a.dialSSH(ctx, host) + if err != nil { + return nil, err + } + hc, err := connectLibvirt(ctx, client, host.Socket) + if err != nil { + client.Close() + return nil, err + } + return hc, nil + } +} + +// dialSSH establishes the SSH connection. The handshake is bounded by ctx's +// deadline, or defaultHandshakeTimeout when ctx has none, since a +// half-open hypervisor must not pin a reconcile worker indefinitely. +func (a *sshAuth) dialSSH(ctx context.Context, host provider.KVMHost) (*ssh.Client, error) { + deadline, ok := ctx.Deadline() + if !ok { + deadline = time.Now().Add(defaultHandshakeTimeout) + } + var d net.Dialer + nc, err := d.DialContext(ctx, "tcp", host.Address) + if err != nil { + return nil, fmt.Errorf("ssh dial: %w", err) + } + _ = nc.SetDeadline(deadline) + conn, chans, reqs, err := ssh.NewClientConn(nc, host.Address, &ssh.ClientConfig{ + User: host.User, + Auth: []ssh.AuthMethod{ssh.PublicKeys(a.signer)}, + HostKeyCallback: a.hostKey, + }) + if err != nil { + nc.Close() + return nil, fmt.Errorf("ssh handshake: %w", err) + } + _ = nc.SetDeadline(time.Time{}) + return ssh.NewClient(conn, chans, reqs), nil +} + +// unixForward adapts an SSH client to go-libvirt's socket.Dialer: each Dial +// opens a direct-streamlocal channel to the libvirt socket on the host. +type unixForward struct { + client *ssh.Client + path string +} + +func (u unixForward) Dial() (net.Conn, error) { + return u.client.Dial("unix", u.path) +} + +// connectLibvirt opens the libvirt session over the SSH client and starts a +// keepalive loop so a dead link is noticed within roughly two intervals +// rather than at the TCP stack's leisure. Closing the returned client tears +// down the SSH connection, which in turn ends the libvirt session. +func connectLibvirt(ctx context.Context, client *ssh.Client, socket string) (*hostClient, error) { + lv := libvirt.NewWithDialer(unixForward{client: client, path: socket}) + if err := awaitOrCancel(ctx, lv.Connect, func() { client.Close() }); err != nil { + return nil, fmt.Errorf("libvirt connect via %s: %w", socket, err) + } + stop := make(chan struct{}) + var once sync.Once + go keepalive(client, keepaliveInterval, stop) + return &hostClient{ + api: lv, + done: lv.Disconnected(), + close: func() { + once.Do(func() { + close(stop) + client.Close() + }) + }, + }, nil +} + +// keepalive sends OpenSSH keepalive requests until stop closes; a failed or +// unanswered request closes the client, which surfaces as Disconnected on +// the libvirt side and makes the pool redial. +func keepalive(client *ssh.Client, interval time.Duration, stop <-chan struct{}) { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + } + replied := make(chan error, 1) + go func() { + _, _, err := client.SendRequest("keepalive@openssh.com", true, nil) + replied <- err + }() + select { + case <-stop: + return + case err := <-replied: + if err != nil { + client.Close() + return + } + case <-time.After(interval): + client.Close() + return + } + } +} diff --git a/internal/provider/kvm/conn_test.go b/internal/provider/kvm/conn_test.go new file mode 100644 index 0000000..732413b --- /dev/null +++ b/internal/provider/kvm/conn_test.go @@ -0,0 +1,569 @@ +package kvm + +import ( + "bytes" + "context" + "crypto/ed25519" + "crypto/rand" + "encoding/pem" + "errors" + "io" + "net" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + libvirt "github.com/digitalocean/go-libvirt" + "golang.org/x/crypto/ssh" + "golang.org/x/crypto/ssh/knownhosts" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// --- hostPool over a fake dialer --- + +type fakeClient struct { + done chan struct{} + closed atomic.Bool +} + +func (c *fakeClient) close() { + if c.closed.CompareAndSwap(false, true) { + close(c.done) + } +} + +type fakeDialer struct { + mu sync.Mutex + dials int + err error + delay time.Duration + clients []*fakeClient +} + +func (d *fakeDialer) dial(ctx context.Context, _ provider.KVMHost) (*hostClient, error) { + if d.delay > 0 { + select { + case <-time.After(d.delay): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + d.mu.Lock() + defer d.mu.Unlock() + d.dials++ + if d.err != nil { + return nil, d.err + } + fc := &fakeClient{done: make(chan struct{})} + d.clients = append(d.clients, fc) + return &hostClient{api: nil, done: fc.done, close: fc.close}, nil +} + +func (d *fakeDialer) dialCount() int { + d.mu.Lock() + defer d.mu.Unlock() + return d.dials +} + +var testHost = provider.KVMHost{Name: "nas1", Address: "nas1:22", User: "virt", Socket: defaultSocket, Pool: "default", Bridge: "br0"} + +func noop(libvirtAPI) error { return nil } + +func TestHostPool_call_reusesConnection(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + for range 3 { + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Fatalf("call() error = %v", err) + } + } + if d.dialCount() != 1 { + t.Errorf("dials = %d, want 1", d.dialCount()) + } +} + +func TestHostPool_call_redialsAfterDisconnect(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Fatalf("call() error = %v", err) + } + d.clients[0].close() + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Fatalf("second call() error = %v", err) + } + if d.dialCount() != 2 { + t.Errorf("dials = %d, want 2 after the first connection dropped", d.dialCount()) + } +} + +func TestHostPool_call_transportErrorDropsConnection(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + err := p.call(context.Background(), testHost, func(libvirtAPI) error { return io.ErrUnexpectedEOF }) + if !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("call() error = %v, want the transport error passed through", err) + } + if !d.clients[0].closed.Load() { + t.Error("connection not closed after a transport error") + } + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Fatalf("call() after transport error = %v", err) + } + if d.dialCount() != 2 { + t.Errorf("dials = %d, want 2", d.dialCount()) + } +} + +func TestHostPool_call_libvirtErrorKeepsConnection(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + err := p.call(context.Background(), testHost, func(libvirtAPI) error { return lvErr(libvirt.ErrNoDomain) }) + if !hasCode(err, libvirt.ErrNoDomain) { + t.Fatalf("call() error = %v, want the libvirt error passed through", err) + } + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Fatalf("call() error = %v", err) + } + if d.dialCount() != 1 { + t.Errorf("dials = %d, want 1: a daemon-reported error must not cost the connection", d.dialCount()) + } + if d.clients[0].closed.Load() { + t.Error("connection closed after a libvirt error") + } +} + +func TestHostPool_call_dialErrorPropagates(t *testing.T) { + t.Parallel() + d := &fakeDialer{err: errors.New("ssh: handshake failed")} + p := newHostPool(d.dial) + err := p.call(context.Background(), testHost, noop) + if err == nil || !strings.Contains(err.Error(), "nas1") || !strings.Contains(err.Error(), "handshake failed") { + t.Errorf("call() error = %v, want the dial error naming the host", err) + } +} + +func TestHostPool_call_cancelAbandonsBlockedCall(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + ctx, cancel := context.WithCancel(context.Background()) + released := make(chan struct{}) + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + err := p.call(ctx, testHost, func(libvirtAPI) error { + // Simulates an RPC stuck on a hung hypervisor: it only returns once + // the pool closes the connection out from under it. + <-d.clients[0].done + close(released) + return io.EOF + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("call() error = %v, want context.Canceled", err) + } + select { + case <-released: + case <-time.After(time.Second): + t.Fatal("blocked call was not released by closing the connection") + } + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Fatalf("call() after cancel = %v", err) + } + if d.dialCount() != 2 { + t.Errorf("dials = %d, want 2", d.dialCount()) + } +} + +func TestHostPool_call_concurrentCallersShareOneDial(t *testing.T) { + t.Parallel() + d := &fakeDialer{delay: 30 * time.Millisecond} + p := newHostPool(d.dial) + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + if err := p.call(context.Background(), testHost, noop); err != nil { + t.Errorf("call() error = %v", err) + } + }) + } + wg.Wait() + if d.dialCount() != 1 { + t.Errorf("dials = %d, want 1", d.dialCount()) + } +} + +func TestHostPool_call_hostsAreIndependent(t *testing.T) { + t.Parallel() + d := &fakeDialer{} + p := newHostPool(d.dial) + other := testHost + other.Name = "lab2" + for _, h := range []provider.KVMHost{testHost, other, testHost, other} { + if err := p.call(context.Background(), h, noop); err != nil { + t.Fatalf("call(%s) error = %v", h.Name, err) + } + } + if d.dialCount() != 2 { + t.Errorf("dials = %d, want one per host", d.dialCount()) + } +} + +// --- SSH dial against an in-process server --- + +func newKeyPair(t *testing.T) (ssh.Signer, []byte) { + t.Helper() + _, priv, err := ed25519.GenerateKey(rand.Reader) + if err != nil { + t.Fatal(err) + } + signer, err := ssh.NewSignerFromKey(priv) + if err != nil { + t.Fatal(err) + } + block, err := ssh.MarshalPrivateKey(priv, "") + if err != nil { + t.Fatal(err) + } + return signer, pem.EncodeToMemory(block) +} + +// testSSHServer accepts one public key and serves +// direct-streamlocal@openssh.com channels by dialing the requested unix +// path, which is exactly what OpenSSH does for ssh.Client.Dial("unix"). +type testSSHServer struct { + addr string + hostKey ssh.PublicKey + + mu sync.Mutex + paths []string +} + +func startSSHServer(t *testing.T, clientKey ssh.PublicKey, answerGlobalRequests bool) *testSSHServer { + t.Helper() + hostSigner, _ := newKeyPair(t) + cfg := &ssh.ServerConfig{ + PublicKeyCallback: func(_ ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) { + if bytes.Equal(key.Marshal(), clientKey.Marshal()) { + return nil, nil + } + return nil, errors.New("unknown public key") + }, + } + cfg.AddHostKey(hostSigner) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + s := &testSSHServer{addr: ln.Addr().String(), hostKey: hostSigner.PublicKey()} + go func() { + for { + nc, err := ln.Accept() + if err != nil { + return + } + go s.serve(nc, cfg, answerGlobalRequests) + } + }() + return s +} + +func (s *testSSHServer) serve(nc net.Conn, cfg *ssh.ServerConfig, answerGlobalRequests bool) { + _, chans, reqs, err := ssh.NewServerConn(nc, cfg) + if err != nil { + return + } + if answerGlobalRequests { + go ssh.DiscardRequests(reqs) + } else { + go func() { + for range reqs { //nolint:revive // deliberately never replies, to stall keepalives + } + }() + } + for newCh := range chans { + if newCh.ChannelType() != "direct-streamlocal@openssh.com" { + _ = newCh.Reject(ssh.UnknownChannelType, "test server only forwards unix sockets") + continue + } + var payload struct { + Path string + Reserved0 string + Reserved1 uint32 + } + if err := ssh.Unmarshal(newCh.ExtraData(), &payload); err != nil { + _ = newCh.Reject(ssh.ConnectionFailed, err.Error()) + continue + } + s.mu.Lock() + s.paths = append(s.paths, payload.Path) + s.mu.Unlock() + target, err := net.Dial("unix", payload.Path) + if err != nil { + _ = newCh.Reject(ssh.ConnectionFailed, err.Error()) + continue + } + ch, chReqs, err := newCh.Accept() + if err != nil { + target.Close() + continue + } + go ssh.DiscardRequests(chReqs) + go func() { _, _ = io.Copy(ch, target); ch.Close() }() + go func() { _, _ = io.Copy(target, ch); target.Close() }() + } +} + +func (s *testSSHServer) requestedPaths() []string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]string(nil), s.paths...) +} + +// startEchoSocket serves a unix socket that echoes everything back, standing +// in for libvirt-sock. +func startEchoSocket(t *testing.T) string { + t.Helper() + path := filepath.Join(t.TempDir(), "libvirt-sock") + ln, err := net.Listen("unix", path) + if err != nil { + t.Fatal(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 +} + +type sshFixture struct { + server *testSSHServer + cfg provider.KVMConfig + host provider.KVMHost + keyFile string +} + +func newSSHFixture(t *testing.T, answerGlobalRequests bool) *sshFixture { + t.Helper() + clientSigner, clientPEM := newKeyPair(t) + server := startSSHServer(t, clientSigner.PublicKey(), answerGlobalRequests) + dir := t.TempDir() + keyFile := filepath.Join(dir, "id_ed25519") + if err := os.WriteFile(keyFile, clientPEM, 0o600); err != nil { + t.Fatal(err) + } + khFile := filepath.Join(dir, "known_hosts") + line := knownhosts.Line([]string{server.addr}, server.hostKey) + if err := os.WriteFile(khFile, []byte(line+"\n"), 0o600); err != nil { + t.Fatal(err) + } + return &sshFixture{ + server: server, + keyFile: keyFile, + cfg: provider.KVMConfig{SSHKeyFile: keyFile, KnownHostsFile: khFile}, + host: provider.KVMHost{Name: "nas1", Address: server.addr, User: "virt", Socket: startEchoSocket(t)}, + } +} + +func TestSSHAuth_dialSSH_forwardsUnixSocket(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, true) + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + client, err := auth.dialSSH(ctx, f.host) + if err != nil { + t.Fatalf("dialSSH() error = %v", err) + } + defer client.Close() + + conn, err := unixForward{client: client, path: f.host.Socket}.Dial() + if err != nil { + t.Fatalf("unixForward.Dial() error = %v", err) + } + defer conn.Close() + if _, err := conn.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + buf := make([]byte, 4) + if _, err := io.ReadFull(conn, buf); err != nil || string(buf) != "ping" { + t.Fatalf("echo through the forward = %q, %v; want ping", buf, err) + } + if paths := f.server.requestedPaths(); len(paths) != 1 || paths[0] != f.host.Socket { + t.Errorf("server saw forward requests %v, want [%s]", paths, f.host.Socket) + } +} + +func TestSSHAuth_dialSSH_rejectsUnknownHostKey(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, true) + impostor, _ := newKeyPair(t) + line := knownhosts.Line([]string{f.server.addr}, impostor.PublicKey()) + if err := os.WriteFile(f.cfg.KnownHostsFile, []byte(line+"\n"), 0o600); err != nil { + t.Fatal(err) + } + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + client, err := auth.dialSSH(context.Background(), f.host) + if err == nil { + client.Close() + t.Fatal("dialSSH() succeeded against a host whose key is not in known_hosts") + } + var keyErr *knownhosts.KeyError + if !errors.As(err, &keyErr) { + t.Errorf("dialSSH() error = %v, want a knownhosts.KeyError", err) + } +} + +func TestSSHAuth_dialSSH_insecureIgnoreHostKey(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, true) + f.cfg.KnownHostsFile = "" + f.cfg.InsecureIgnoreHostKey = true + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + client, err := auth.dialSSH(context.Background(), f.host) + if err != nil { + t.Fatalf("dialSSH() error = %v", err) + } + client.Close() +} + +func TestSSHAuth_dialSSH_rejectsWrongClientKey(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, true) + _, otherPEM := newKeyPair(t) + if err := os.WriteFile(f.keyFile, otherPEM, 0o600); err != nil { + t.Fatal(err) + } + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + client, err := auth.dialSSH(context.Background(), f.host) + if err == nil { + client.Close() + t.Fatal("dialSSH() succeeded with a key the server does not know") + } + if !strings.Contains(err.Error(), "ssh handshake") { + t.Errorf("dialSSH() error = %v, want an auth failure from the handshake", err) + } +} + +func TestSSHAuth_dialSSH_cancelledContext(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, true) + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if client, err := auth.dialSSH(ctx, f.host); err == nil { + client.Close() + t.Fatal("dialSSH() succeeded with an already-cancelled context") + } +} + +func TestNewSSHAuth_missingFiles(t *testing.T) { + t.Parallel() + _, keyPEM := newKeyPair(t) + keyFile := filepath.Join(t.TempDir(), "key") + if err := os.WriteFile(keyFile, keyPEM, 0o600); err != nil { + t.Fatal(err) + } + tests := []struct { + name string + cfg provider.KVMConfig + wantSub string + }{ + {"key file missing", provider.KVMConfig{SSHKeyFile: "/nonexistent/key", InsecureIgnoreHostKey: true}, "sshKeyFile"}, + {"key file garbage", provider.KVMConfig{SSHKeyFile: keyFile + ".bad", InsecureIgnoreHostKey: true}, "sshKeyFile"}, + {"known_hosts missing", provider.KVMConfig{SSHKeyFile: keyFile, KnownHostsFile: "/nonexistent/kh"}, "knownHostsFile"}, + } + if err := os.WriteFile(keyFile+".bad", []byte("not a key"), 0o600); err != nil { + t.Fatal(err) + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + _, err := newSSHAuth(tc.cfg) + if err == nil || !strings.Contains(err.Error(), tc.wantSub) { + t.Errorf("newSSHAuth() error = %v, want one mentioning %s", err, tc.wantSub) + } + }) + } +} + +func TestKeepalive_closesClientWhenServerStalls(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, false) + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + client, err := auth.dialSSH(context.Background(), f.host) + if err != nil { + t.Fatalf("dialSSH() error = %v", err) + } + defer client.Close() + stop := make(chan struct{}) + defer close(stop) + go keepalive(client, 20*time.Millisecond, stop) + + closed := make(chan struct{}) + go func() { _ = client.Wait(); close(closed) }() + select { + case <-closed: + case <-time.After(2 * time.Second): + t.Fatal("client still open after keepalives went unanswered") + } +} + +func TestKeepalive_keepsHealthyClientOpen(t *testing.T) { + t.Parallel() + f := newSSHFixture(t, true) + auth, err := newSSHAuth(f.cfg) + if err != nil { + t.Fatalf("newSSHAuth() error = %v", err) + } + client, err := auth.dialSSH(context.Background(), f.host) + if err != nil { + t.Fatalf("dialSSH() error = %v", err) + } + defer client.Close() + stop := make(chan struct{}) + defer close(stop) + go keepalive(client, 10*time.Millisecond, stop) + time.Sleep(60 * time.Millisecond) + fwd := unixForward{client: client, path: f.host.Socket} + conn, err := fwd.Dial() + if err != nil { + t.Fatalf("forward after several keepalives failed: %v", err) + } + conn.Close() +} diff --git a/internal/provider/kvm/errors.go b/internal/provider/kvm/errors.go new file mode 100644 index 0000000..cacf2f0 --- /dev/null +++ b/internal/provider/kvm/errors.go @@ -0,0 +1,65 @@ +package kvm + +import ( + "errors" + + libvirt "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// classify maps a libvirt error onto the provider taxonomy. Missing domain +// or volume → NotFound; anything the host will keep rejecting (bad XML, +// unsupported config, denied access, missing pool) → Permanent; every other +// libvirt code, plus SSH/transport failures and cancellation, → Transient, +// because retrying is safer than latching Failed on an error nobody taught +// this function to recognise. libvirt exposes no quota signal, so nothing +// maps to QuotaExceeded. +func classify(err error) error { + var lerr libvirt.Error + if !errors.As(err, &lerr) { + return provider.ErrTransient + } + switch libvirt.ErrorNumber(lerr.Code) { + case libvirt.ErrNoDomain, libvirt.ErrNoStorageVol: + return provider.ErrNotFound + case libvirt.ErrNoStoragePool, + libvirt.ErrXMLError, libvirt.ErrXMLDetail, libvirt.ErrXMLInvalidSchema, + libvirt.ErrConfigUnsupported, libvirt.ErrNoSupport, + libvirt.ErrOperationUnsupported, libvirt.ErrArgumentUnsupported, + libvirt.ErrInvalidArg, + libvirt.ErrAuthFailed, libvirt.ErrAccessDenied, libvirt.ErrOperationDenied: + return provider.ErrPermanent + default: + return provider.ErrTransient + } +} + +// isLibvirtError reports whether err came back from the libvirt daemon (as +// opposed to the SSH or RPC transport), which is what decides whether the +// cached connection is still trustworthy. +func isLibvirtError(err error) bool { + var lerr libvirt.Error + return errors.As(err, &lerr) +} + +func hasCode(err error, code libvirt.ErrorNumber) bool { + var lerr libvirt.Error + return errors.As(err, &lerr) && libvirt.ErrorNumber(lerr.Code) == code +} + +// logAPIError records the raw libvirt code and message at V(1); classify +// collapses them onto the coarser taxonomy, so this is the only place the +// original detail survives. +func logAPIError(log logr.Logger, op string, err error) { + if !log.V(1).Enabled() { + return + } + kv := []any{"op", op, "error", err.Error()} + var lerr libvirt.Error + if errors.As(err, &lerr) { + kv = append(kv, "libvirtCode", lerr.Code, "libvirtMessage", lerr.Message) + } + log.V(1).Info("libvirt call failed", kv...) +} diff --git a/internal/provider/kvm/errors_test.go b/internal/provider/kvm/errors_test.go new file mode 100644 index 0000000..9369c49 --- /dev/null +++ b/internal/provider/kvm/errors_test.go @@ -0,0 +1,67 @@ +package kvm + +import ( + "context" + "errors" + "fmt" + "io" + "testing" + + libvirt "github.com/digitalocean/go-libvirt" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +func lvErr(code libvirt.ErrorNumber) error { + return libvirt.Error{Code: uint32(code), Message: code.String()} +} + +func TestClassify(t *testing.T) { + t.Parallel() + tests := []struct { + name string + err error + want error + }{ + {"no domain", lvErr(libvirt.ErrNoDomain), provider.ErrNotFound}, + {"no storage vol", lvErr(libvirt.ErrNoStorageVol), provider.ErrNotFound}, + {"no storage pool", lvErr(libvirt.ErrNoStoragePool), provider.ErrPermanent}, + {"xml error", lvErr(libvirt.ErrXMLError), provider.ErrPermanent}, + {"xml detail", lvErr(libvirt.ErrXMLDetail), provider.ErrPermanent}, + {"config unsupported", lvErr(libvirt.ErrConfigUnsupported), provider.ErrPermanent}, + {"no support", lvErr(libvirt.ErrNoSupport), provider.ErrPermanent}, + {"invalid arg", lvErr(libvirt.ErrInvalidArg), provider.ErrPermanent}, + {"auth failed", lvErr(libvirt.ErrAuthFailed), provider.ErrPermanent}, + {"access denied", lvErr(libvirt.ErrAccessDenied), provider.ErrPermanent}, + {"operation invalid", lvErr(libvirt.ErrOperationInvalid), provider.ErrTransient}, + {"operation failed", lvErr(libvirt.ErrOperationFailed), provider.ErrTransient}, + {"internal error", lvErr(libvirt.ErrInternalError), provider.ErrTransient}, + {"storage vol exists", lvErr(libvirt.ErrStorageVolExist), provider.ErrTransient}, + {"wrapped libvirt error", fmt.Errorf("defining: %w", lvErr(libvirt.ErrNoDomain)), provider.ErrNotFound}, + {"transport eof", io.EOF, provider.ErrTransient}, + {"context cancelled", context.Canceled, provider.ErrTransient}, + {"plain error", errors.New("ssh: handshake failed"), provider.ErrTransient}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := classify(tc.err); !errors.Is(got, tc.want) { + t.Errorf("classify(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +func TestHasCode(t *testing.T) { + t.Parallel() + err := fmt.Errorf("lookup: %w", lvErr(libvirt.ErrNoDomain)) + if !hasCode(err, libvirt.ErrNoDomain) { + t.Error("hasCode(ErrNoDomain) = false through a wrap") + } + if hasCode(err, libvirt.ErrNoStorageVol) { + t.Error("hasCode(ErrNoStorageVol) = true for an ErrNoDomain error") + } + if hasCode(io.EOF, libvirt.ErrNoDomain) || isLibvirtError(io.EOF) { + t.Error("io.EOF classified as a libvirt error") + } +}