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`.

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

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=

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
+		}
+	}
+}

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()
+}

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...)
+}

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")
+	}
+}