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 beebc58..02da6f9 100644 --- a/docs/plans-executions/2026-09-04-1404-kvm-provider.md +++ b/docs/plans-executions/2026-09-04-1404-kvm-provider.md @@ -7,7 +7,7 @@ Plan: [2026-09-04-1404-kvm-provider.md](../plans/2026-09-04-1404-kvm-provider.md - [x] Step 2 — Pure builders (`domainxml.go`, `volume.go`, `seed.go`, `id.go`, `metadata.go`) - [x] Step 3 — Transport + errors (`conn.go`, `errors.go`) - [x] Step 3b — Image cache (`image.go`) -- [ ] Step 4 — Provider methods (`kvm.go`) over a fake API +- [x] Step 4 — Provider methods (`kvm.go`) over a fake API - [ ] Step 5 — Wiring, manifests, samples, docs, CHANGELOG - [ ] Step 6 — Integration test against a real hypervisor - [ ] Step 7 — Manual validation + MR @@ -196,3 +196,90 @@ confirms this on a real host. Also, the whole download runs inside one `pool.call`, so cancellation aborts the HTTP body (ctx-bound) and the upload stream fails; the half-written base volume is deleted on the spot and, failing that, on the next attempt via the missing marker. + +## Step 4 — Provider methods + +`kvm.go` now holds `Provider`, `New` (loads the SSH key and known_hosts +up front via `newSSHAuth`, so a bad path fails at startup), the +`newWithDialer` seam, and `Create`/`Get`/`Delete`/`ListByTag`. Each +method resolves its host from config, then runs the whole libvirt +conversation inside one `hostPool.call`, so a cancelled reconcile drops +the connection and the blocked RPC returns. Errors pass through `classify` +in one `wrapErr`, as the Step 3b note intended. + +Behaviour that goes beyond the plan text, or that a future reader should +know before touching this file: + +- **Create looks up the domain first.** If it exists, only the running + check runs and the volumes are never touched. If not, nothing can be + using the volumes yet: the overlay is reused when it exists (its + creation is atomic; `ErrStorageVolExist` → reuse) but the **seed is + always deleted and rebuilt**. The plan said "exists → skip upload"; the + reviewer pointed out that a reconcile cancelled mid-upload leaves a + truncated seed that the compensating delete cannot remove, because + `hostPool.call` has already closed the connection the delete would go + over. Skipping such a seed would boot the VM without cloud-init and no + retry would ever repair it. Rebuilding costs one ~40 KiB upload per + attempt before the domain exists, nothing afterwards. The table test + breaks Create after each step and re-runs it, and a separate test + starts from a leftover truncated seed. +- **Get swallows address-source errors.** `DomainInterfaceAddresses` from + ARP returns nothing until the guest has sent traffic and the agent + source errors with `ErrAgentUnresponsive` until `qemu-guest-agent` is + up, so both are logged at V(2) and mean "no IP yet" → `Provisioning`. + Loopback, link-local and unspecified addresses are skipped when + choosing the first IPv4 (the agent reports `lo` too). +- **State mapping:** `running`/`blocked` are active; `nostate` → + Provisioning (the reconciler answers Stopped with delete-and-recreate, + too drastic for a state libvirt could not determine); everything else + (paused, pmsuspended, shutdown, shutoff, crashed) → Stopped. +- **Delete treats a missing pool as "nothing to clean"** rather than + Permanent: failing there would pin the Proxy's finalizer on a config + edit that renamed the pool. Domain removal is destroy (`ErrOperationInvalid` + on a stopped domain is fine) → undefine with managed-save, snapshot + metadata and nvram flags → overlay and seed deleted by name. +- **ListByTag** reads each domain's metadata once (passed into the same + `inspect` that Get uses, so no second `DomainGetMetadata` round-trip) + and skips domains without the operator's element. A domain that cannot + be read at all — vanished mid-sweep, corrupt metadata, state call + failing — is *skipped*, not fatal: GC never deletes what it does not + see, so leaving one out is safe, whereas one bad domain blocking GC for + every other one (the first version's behaviour, caught in review) is + not. An unreachable host still fails the whole sweep. + `ConnectListAllDomains` is called with flags 0 (active and inactive). +- A host named in a providerID but no longer in `kvm.hosts` is Permanent + for Get and Delete: the operator cannot reach it, and retrying forever + would hide the config problem. + +Tests run the provider over a `hostDialer` that maps host names to +`fakeAPI` instances (an unmapped host is "unreachable"). The image is +pre-seeded in the pool for most tests; one test runs the full first-use +download through `httptest` and confirms a second Proxy on the same host +does not download again. + +Found while running the package: `TestBuildSeedISO_deterministic` from +Step 2 was wrong. `iso9660.ImageWriter` stamps `time.Now()` into the +volume descriptor, so two builds in different seconds differ by six +timestamp bytes; the test had only passed because both builds landed in +the same second. Byte identity is nothing the provider depends on (an +existing seed is never re-uploaded), so the test now compares label, +files and size, and the comment on `buildSeedISO` no longer claims +identical output. + +The `go-operator-reviewer` agent ran on the first version of this step +and found the two bugs above (truncated seed, wedged GC sweep) plus the +double metadata read and the `nostate` mapping; all were fixed before +committing. One finding it raised is deliberately **not** fixed here +because it lives in Step 3b's `image.go` and changes the volume naming +the plan chose: re-fetching a base image (checksum edited in config, or +upstream content changed) deletes `base-.` while existing +overlays still back onto it, breaking those VMs' disks. Content-addressed +base names would avoid it. Flagged for the user; until decided, Step 5's +docs must warn that an image's `sha256`/`url` must not change while +Proxies using it exist. + +Worth noting: `go test -cover` reports 88.2 % for the package; the +uncovered lines are the real SSH/libvirt dial path and log-only +branches. `go vet` and `-race` (three runs) are clean. Wiring into +`cmd/main.go` is Step 5, so `New` is not yet reachable from a running +operator. diff --git a/internal/provider/kvm/kvm.go b/internal/provider/kvm/kvm.go index e5653e3..e510d08 100644 --- a/internal/provider/kvm/kvm.go +++ b/internal/provider/kvm/kvm.go @@ -7,7 +7,20 @@ // NoCloud seed ISO built in-process. package kvm -import "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +import ( + "bytes" + "context" + "fmt" + "net" + "net/http" + "time" + + libvirt "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) const ( defaultDiskSizeGB = 10 @@ -15,6 +28,12 @@ const ( defaultPool = "default" formatQcow2 = "qcow2" formatRaw = "raw" + + // undefineFlags removes everything libvirt may keep alongside a domain + // definition, so an undefine never fails on leftover state. + undefineFlags = libvirt.DomainUndefineManagedSave | + libvirt.DomainUndefineSnapshotsMetadata | + libvirt.DomainUndefineNvram ) func withDefaults(cfg provider.KVMConfig) provider.KVMConfig { @@ -45,3 +64,497 @@ func withDefaults(cfg provider.KVMConfig) provider.KVMConfig { cfg.Hosts = hosts return cfg } + +// Provider implements provider.Provider on one or more libvirt hosts. +type Provider struct { + name string + cfg provider.KVMConfig + hosts *hostPool + images *imageCache + now func() time.Time +} + +// New builds a Provider that reaches its hosts over SSH. The key and +// known_hosts files are read here so a bad path fails at startup rather +// than on the first Create. Satisfies registry.Constructor. +func New(_ context.Context, pc provider.ProviderConfig) (provider.Provider, error) { + cfg := provider.KVMConfig{} + if pc.KVM != nil { + cfg = *pc.KVM + } + auth, err := newSSHAuth(cfg) + if err != nil { + return nil, fmt.Errorf("kvm provider %q: %w", pc.Name, err) + } + return newWithDialer(pc, auth.dialer(), nil), nil +} + +// newWithDialer is the test seam: the dialer decides what libvirtAPI each +// host resolves to, and the HTTP client is what image downloads go through. +func newWithDialer(pc provider.ProviderConfig, dial hostDialer, httpClient *http.Client) *Provider { + cfg := provider.KVMConfig{} + if pc.KVM != nil { + cfg = *pc.KVM + } + return &Provider{ + name: pc.Name, + cfg: withDefaults(cfg), + hosts: newHostPool(dial), + images: newImageCache(httpClient), + now: time.Now, + } +} + +func (p *Provider) logger(ctx context.Context) logr.Logger { + return logf.FromContext(ctx).WithName("kvm").WithValues("provider", p.name) +} + +func (p *Provider) wrapErr(op, id string, err error) error { + return provider.Wrap(classify(err), op, p.name, id, err) +} + +// host resolves placement.zone (or a providerID's host segment) to a +// configured host. An empty zone is allowed only when exactly one host is +// configured; anything else is a spec error. +func (p *Provider) host(zone string) (provider.KVMHost, error) { + if zone == "" { + switch len(p.cfg.Hosts) { + case 0: + return provider.KVMHost{}, fmt.Errorf("%w: kvm.hosts is empty", provider.ErrPermanent) + case 1: + return p.cfg.Hosts[0], nil + default: + return provider.KVMHost{}, fmt.Errorf("%w: placement.zone is required when kvm.hosts lists more than one host", + provider.ErrPermanent) + } + } + for _, h := range p.cfg.Hosts { + if h.Name == zone { + return h, nil + } + } + return provider.KVMHost{}, fmt.Errorf("%w: host %q is not in kvm.hosts", provider.ErrPermanent, zone) +} + +// Create makes the VM exist and running on the placed host, and is +// idempotent: a defined domain is only started if needed, its volumes +// untouched. Before the domain exists nothing can be using the volumes, +// so an existing overlay is reused (its creation is atomic) but the seed +// is rebuilt from scratch — a previous attempt may have been cut off mid +// upload, and a truncated seed would boot a VM without cloud-init. The +// first Create for an image on a host also downloads that image into the +// host's pool, synchronously. +func (p *Provider) Create(ctx context.Context, req provider.CreateRequest) (string, error) { + pl := req.Placement + host, err := p.host(pl.Zone) + if err != nil { + return "", p.wrapErr("create", "", err) + } + mt, ok := p.cfg.MachineTypes[pl.MachineType] + if !ok { + return "", p.wrapErr("create", "", fmt.Errorf("%w: placement.machineType %q is not in kvm.machineTypes", + provider.ErrPermanent, pl.MachineType)) + } + src, err := resolveImage(p.cfg, pl.Image) + if err != nil { + return "", p.wrapErr("create", "", err) + } + id := formatProviderID(host.Name, req.Name) + log := p.logger(ctx).WithValues("host", host.Name, "name", req.Name) + log.V(2).Info("KVM create resolved", + "pool", host.Pool, "bridge", host.Bridge, "machineType", pl.MachineType, + "vcpus", mt.VCPUs, "memoryMiB", mt.MemoryMiB, "diskSizeGb", p.cfg.DiskSizeGB, + "image", src.volumeName(), "cloudInitBytes", len(req.CloudInit)) + + err = p.hosts.call(ctx, host, func(api libvirtAPI) error { + dom, err := api.DomainLookupByName(req.Name) + if err == nil { + log.V(1).Info("domain already defined") + return p.ensureRunning(log, api, dom) + } + if !hasCode(err, libvirt.ErrNoDomain) { + return fmt.Errorf("looking up domain: %w", err) + } + pool, err := api.StoragePoolLookupByName(host.Pool) + if err != nil { + return fmt.Errorf("looking up pool %s: %w", host.Pool, err) + } + basePath, err := p.images.ensure(ctx, log, api, host.Name, pool, src) + if err != nil { + return err + } + if err := p.ensureOverlay(log, api, pool, req.Name, basePath, src.Format); err != nil { + return err + } + if err := p.writeSeed(log, api, pool, req.Name, req.CloudInit); err != nil { + return err + } + dom, err = p.defineDomain(log, api, host, req, mt) + if err != nil { + return err + } + return p.ensureRunning(log, api, dom) + }) + if err != nil { + logAPIError(log, "create", err) + return "", p.wrapErr("create", id, err) + } + log.V(1).Info("KVM domain ensured running", "id", id) + return id, nil +} + +func (p *Provider) ensureOverlay(log logr.Logger, api libvirtAPI, pool libvirt.StoragePool, name, basePath, baseFormat string) error { + xml, err := buildVolumeXML(overlayVolume(name, p.cfg.DiskSizeGB, basePath, baseFormat)) + if err != nil { + return err + } + _, err = api.StorageVolCreateXML(pool, xml, 0) + switch { + case err == nil: + log.V(1).Info("overlay volume created", "volume", overlayVolumeName(name), "base", basePath) + case hasCode(err, libvirt.ErrStorageVolExist): + log.V(1).Info("overlay volume already exists", "volume", overlayVolumeName(name)) + default: + return fmt.Errorf("creating overlay volume: %w", err) + } + return nil +} + +// writeSeed replaces the seed volume with a freshly built NoCloud ISO. +// Only called while no domain references it. +func (p *Provider) writeSeed(log logr.Logger, api libvirtAPI, pool libvirt.StoragePool, name, userData string) error { + seedName := seedVolumeName(name) + if err := deleteIfExists(api, pool, seedName); err != nil { + return err + } + iso, err := buildSeedISO(name, userData, p.cfg.VendorData) + if err != nil { + return fmt.Errorf("%w: %v", provider.ErrPermanent, err) + } + xml, err := buildVolumeXML(rawVolume(seedName, int64(len(iso)))) + if err != nil { + return err + } + vol, err := api.StorageVolCreateXML(pool, xml, 0) + if err != nil { + return fmt.Errorf("creating seed volume: %w", err) + } + if err := api.StorageVolUpload(vol, bytes.NewReader(iso), 0, 0, 0); err != nil { + return fmt.Errorf("uploading seed volume: %w", err) + } + log.V(1).Info("seed volume uploaded", "volume", seedName, "bytes", len(iso)) + return nil +} + +func (p *Provider) defineDomain(log logr.Logger, api libvirtAPI, host provider.KVMHost, req provider.CreateRequest, mt provider.KVMMachineType) (libvirt.Domain, error) { + xml, err := buildDomainXML(domainParams{ + Name: req.Name, + UID: req.UID, + CreatedAt: p.now(), + VCPUs: mt.VCPUs, + MemoryMiB: mt.MemoryMiB, + Pool: host.Pool, + Bridge: host.Bridge, + DiskVol: overlayVolumeName(req.Name), + SeedVol: seedVolumeName(req.Name), + }) + if err != nil { + return libvirt.Domain{}, err + } + dom, err := api.DomainDefineXML(xml) + if err != nil { + return libvirt.Domain{}, fmt.Errorf("defining domain: %w", err) + } + log.V(1).Info("domain defined", "uid", req.UID) + return dom, nil +} + +func (p *Provider) ensureRunning(log logr.Logger, api libvirtAPI, dom libvirt.Domain) error { + state, _, err := api.DomainGetState(dom, 0) + if err != nil { + return fmt.Errorf("reading domain state: %w", err) + } + if isActive(libvirt.DomainState(state)) { + return nil + } + err = api.DomainCreate(dom) + switch { + case err == nil: + log.V(1).Info("domain started") + case hasCode(err, libvirt.ErrOperationInvalid): + log.V(1).Info("domain already running, start treated as success") + default: + return fmt.Errorf("starting domain: %w", err) + } + return nil +} + +// Get reads the domain's state, metadata and addresses. The providerID +// carries the host, so this looks at the right hypervisor even after a +// placement.zone edit. +func (p *Provider) Get(ctx context.Context, providerID string) (*provider.Instance, error) { + hostName, name, err := parseProviderID(providerID) + if err != nil { + return nil, provider.Wrap(provider.ErrPermanent, "get", p.name, providerID, err) + } + host, err := p.host(hostName) + if err != nil { + return nil, p.wrapErr("get", providerID, err) + } + log := p.logger(ctx).WithValues("host", host.Name, "name", name) + var inst *provider.Instance + err = p.hosts.call(ctx, host, func(api libvirtAPI) error { + dom, err := api.DomainLookupByName(name) + if err != nil { + return fmt.Errorf("looking up domain: %w", err) + } + meta, err := p.metadata(api, dom) + if err != nil { + return err + } + inst, err = p.inspect(log, api, host.Name, dom, meta) + return err + }) + if err != nil { + logAPIError(log, "get", err) + return nil, p.wrapErr("get", providerID, err) + } + log.V(1).Info("KVM domain fetched", "state", inst.State, "ip", inst.IP, "uid", inst.UID) + return inst, nil +} + +// inspect assembles an Instance from a domain that is known to exist and +// whose metadata has already been read (the zero value for a domain +// without the operator's element, which reports an empty UID as the +// contract requires). +func (p *Provider) inspect(log logr.Logger, api libvirtAPI, hostName string, dom libvirt.Domain, meta instanceMetadata) (*provider.Instance, error) { + state, _, err := api.DomainGetState(dom, 0) + if err != nil { + return nil, fmt.Errorf("reading domain state: %w", err) + } + var ip string + if isActive(libvirt.DomainState(state)) { + ip = p.address(log, api, dom) + } + return &provider.Instance{ + ID: formatProviderID(hostName, dom.Name), + IP: ip, + State: mapState(libvirt.DomainState(state), ip), + UID: meta.UID, + CreatedAt: meta.createdAt(), + }, nil +} + +// metadata returns the operator's element, or the zero value when the +// domain has none (it was not created by this operator). +func (p *Provider) metadata(api libvirtAPI, dom libvirt.Domain) (instanceMetadata, error) { + raw, err := api.DomainGetMetadata(dom, int32(libvirt.DomainMetadataElement), libvirt.OptString{metadataURI}, libvirt.DomainAffectCurrent) + if hasCode(err, libvirt.ErrNoDomainMetadata) { + return instanceMetadata{}, nil + } + if err != nil { + return instanceMetadata{}, fmt.Errorf("reading domain metadata: %w", err) + } + meta, err := parseInstanceMetadata(raw) + if err != nil { + return instanceMetadata{}, fmt.Errorf("%w: %v", provider.ErrPermanent, err) + } + return meta, nil +} + +// address returns the domain's first routable IPv4, asking the host's ARP +// table first and the guest agent second. Both sources fail routinely while +// the guest is still booting (no traffic seen yet, agent not installed +// yet), so their errors mean "no address yet", not "Get failed". +func (p *Provider) address(log logr.Logger, api libvirtAPI, dom libvirt.Domain) string { + for _, source := range []libvirt.DomainInterfaceAddressesSource{ + libvirt.DomainInterfaceAddressesSrcArp, + libvirt.DomainInterfaceAddressesSrcAgent, + } { + ifaces, err := api.DomainInterfaceAddresses(dom, uint32(source), 0) + if err != nil { + log.V(2).Info("no addresses from source", "source", source.String(), "error", err.Error()) + continue + } + if ip := firstIPv4(ifaces); ip != "" { + return ip + } + } + return "" +} + +func firstIPv4(ifaces []libvirt.DomainInterface) string { + for _, iface := range ifaces { + for _, a := range iface.Addrs { + if a.Type != int32(libvirt.IPAddrTypeIpv4) { + continue + } + ip := net.ParseIP(a.Addr) + if ip == nil || ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.IsUnspecified() { + continue + } + return a.Addr + } + } + return "" +} + +// isActive reports whether the guest is executing (or blocked on the +// hypervisor, which is still "up" from the outside). +func isActive(s libvirt.DomainState) bool { + return s == libvirt.DomainRunning || s == libvirt.DomainBlocked +} + +// mapState collapses libvirt's states onto the provider's. An active +// domain without an address is still Provisioning: an empty IP must never +// be published as Running. "No state" is also Provisioning — the +// reconciler answers Stopped with delete-and-recreate, too drastic for a +// state libvirt could not determine; the next poll decides. Everything +// else — paused, suspended, shut off, crashed, shutting down — is Stopped. +func mapState(s libvirt.DomainState, ip string) provider.InstanceState { + if s == libvirt.DomainNostate { + return provider.StateProvisioning + } + if !isActive(s) { + return provider.StateStopped + } + if ip == "" { + return provider.StateProvisioning + } + return provider.StateRunning +} + +// Delete tears the VM down and is idempotent at every step: a stopped +// domain is not destroyed again, a missing domain is not undefined, and +// the per-VM volumes are removed by name even when the domain is already +// gone, so a crash between volume creation and define is cleaned up on the +// next reconcile. +func (p *Provider) Delete(ctx context.Context, providerID string) error { + hostName, name, err := parseProviderID(providerID) + if err != nil { + return provider.Wrap(provider.ErrPermanent, "delete", p.name, providerID, err) + } + host, err := p.host(hostName) + if err != nil { + return p.wrapErr("delete", providerID, err) + } + log := p.logger(ctx).WithValues("host", host.Name, "name", name) + err = p.hosts.call(ctx, host, func(api libvirtAPI) error { + if err := p.removeDomain(log, api, name); err != nil { + return err + } + return p.removeVolumes(log, api, host.Pool, name) + }) + if err != nil { + logAPIError(log, "delete", err) + return p.wrapErr("delete", providerID, err) + } + log.V(1).Info("KVM domain deleted") + return nil +} + +func (p *Provider) removeDomain(log logr.Logger, api libvirtAPI, name string) error { + dom, err := api.DomainLookupByName(name) + if hasCode(err, libvirt.ErrNoDomain) { + log.V(1).Info("domain already gone") + return nil + } + if err != nil { + return fmt.Errorf("looking up domain: %w", err) + } + err = api.DomainDestroy(dom) + switch { + case err == nil: + log.V(1).Info("domain destroyed") + case hasCode(err, libvirt.ErrOperationInvalid): + log.V(1).Info("domain not running, destroy skipped") + case hasCode(err, libvirt.ErrNoDomain): + return nil + default: + return fmt.Errorf("destroying domain: %w", err) + } + if err := api.DomainUndefineFlags(dom, undefineFlags); err != nil && !hasCode(err, libvirt.ErrNoDomain) { + return fmt.Errorf("undefining domain: %w", err) + } + log.V(1).Info("domain undefined") + return nil +} + +// removeVolumes deletes the overlay and seed. A missing pool means there +// is nothing left to clean, not a failure: refusing to finish a Delete +// over a pool that no longer exists would pin the Proxy's finalizer. +func (p *Provider) removeVolumes(log logr.Logger, api libvirtAPI, poolName, name string) error { + pool, err := api.StoragePoolLookupByName(poolName) + if hasCode(err, libvirt.ErrNoStoragePool) { + log.Info("pool not found, no volumes to delete", "pool", poolName) + return nil + } + if err != nil { + return fmt.Errorf("looking up pool %s: %w", poolName, err) + } + for _, vol := range []string{overlayVolumeName(name), seedVolumeName(name)} { + if err := deleteIfExists(api, pool, vol); err != nil { + return err + } + } + log.V(1).Info("per-VM volumes deleted") + return nil +} + +// ListByTag inspects every domain on every host and keeps those carrying +// the operator's managed metadata. One unreachable host fails the whole +// sweep, so orphan GC never acts on a partial view of the fleet; a single +// domain that cannot be read is skipped instead, which is safe in the +// same direction — a domain missing from the list is never collected. +func (p *Provider) ListByTag(ctx context.Context) ([]provider.Instance, error) { + log := p.logger(ctx) + var out []provider.Instance + for _, host := range p.cfg.Hosts { + hlog := log.WithValues("host", host.Name) + err := p.hosts.call(ctx, host, func(api libvirtAPI) error { + doms, _, err := api.ConnectListAllDomains(1, 0) + if err != nil { + return fmt.Errorf("listing domains: %w", err) + } + for _, dom := range doms { + if inst := p.managedInstance(hlog, api, host.Name, dom); inst != nil { + out = append(out, *inst) + } + } + return nil + }) + if err != nil { + logAPIError(hlog, "list", err) + return nil, p.wrapErr("list", "", err) + } + } + log.V(1).Info("KVM domains listed", "hosts", len(p.cfg.Hosts), "managed", len(out)) + for _, inst := range out { + log.V(2).Info("KVM listed instance", "id", inst.ID, "state", inst.State, "uid", inst.UID, "createdAt", inst.CreatedAt) + } + return out, nil +} + +// managedInstance returns nil for domains the operator does not own, for +// domains that vanished between the listing and the inspection, and for +// domains whose metadata or state cannot be read (corrupt element, mid +// teardown): one bad domain must not block GC for every other one. A +// transport failure would also land here, but it fails the next RPC in +// the same sweep too, so the host still counts as unreachable. +func (p *Provider) managedInstance(log logr.Logger, api libvirtAPI, hostName string, dom libvirt.Domain) *provider.Instance { + log = log.WithValues("name", dom.Name) + meta, err := p.metadata(api, dom) + if err != nil { + log.V(1).Info("skipping domain with unreadable metadata", "error", err.Error()) + return nil + } + if !meta.managed() { + return nil + } + inst, err := p.inspect(log, api, hostName, dom, meta) + if err != nil { + log.V(1).Info("skipping domain that could not be inspected", "error", err.Error()) + return nil + } + return inst +} diff --git a/internal/provider/kvm/kvm_test.go b/internal/provider/kvm/kvm_test.go index 0f04970..c627c34 100644 --- a/internal/provider/kvm/kvm_test.go +++ b/internal/provider/kvm/kvm_test.go @@ -1,7 +1,16 @@ package kvm import ( + "context" + "encoding/xml" + "errors" + "fmt" + "io" + "strings" "testing" + "time" + + libvirt "github.com/digitalocean/go-libvirt" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" ) @@ -42,3 +51,830 @@ func TestWithDefaults_keepsExplicitValues(t *testing.T) { t.Errorf("cfg = %+v, want explicit values kept", cfg) } } + +// --- provider over fake hosts --- + +const ( + testName = "proxy-abc123def456ghij" + testUID = "11111111-2222-3333-4444-555555555555" + testImageURL = "https://images.example/debian-13.qcow2" + testCloudInit = "#cloud-config\npackages: [squid]\n" +) + +var ( + testNow = time.Date(2026, 9, 8, 10, 0, 0, 0, time.UTC) + errHostOff = errors.New("ssh dial: connection refused") +) + +// hostFakes routes each configured host name to its own fakeAPI; a host +// with no entry is unreachable. +type hostFakes map[string]*fakeAPI + +func (h hostFakes) dial(_ context.Context, host provider.KVMHost) (*hostClient, error) { + api, ok := h[host.Name] + if !ok { + return nil, errHostOff + } + return &hostClient{api: api, close: func() {}}, nil +} + +func testConfig(hosts ...string) provider.ProviderConfig { + cfg := &provider.KVMConfig{ + SSHKeyFile: "/k", + KnownHostsFile: "/kh", + Images: map[string]provider.KVMImage{"debian-13": {URL: testImageURL}}, + MachineTypes: map[string]provider.KVMMachineType{"small": {VCPUs: 2, MemoryMiB: 2048}}, + } + for _, h := range hosts { + cfg.Hosts = append(cfg.Hosts, provider.KVMHost{Name: h, Address: h + ":22", User: "virt", Bridge: "br0"}) + } + return provider.ProviderConfig{Name: "kvm-home", Type: "kvm", KVM: cfg} +} + +func newTestProvider(fakes hostFakes, hosts ...string) *Provider { + p := newWithDialer(testConfig(hosts...), fakes.dial, nil) + p.now = func() time.Time { return testNow } + return p +} + +// cacheImage pre-populates the pool with the configured image and its +// marker so Create never touches the network. +func cacheImage(api *fakeAPI) { + src := imageSource{Key: "debian-13", URL: testImageURL, Format: formatQcow2} + api.putVol("default", src.volumeName(), formatQcow2, []byte("qcow2")) + api.putVol("default", src.markerName(), formatRaw, []byte(testImageSHA())) +} + +func readyHost() *fakeAPI { + api := newFakeAPI("default") + cacheImage(api) + return api +} + +func testRequest() provider.CreateRequest { + return provider.CreateRequest{ + Name: testName, + UID: testUID, + Namespace: "default", + ProxyName: "home-proxy-1", + Placement: provider.Placement{Zone: "nas1", MachineType: "small", Image: "debian-13"}, + CloudInit: testCloudInit, + Port: 3128, + } +} + +func metadataXML(t *testing.T, uid string, created time.Time) string { + t.Helper() + out, err := xml.Marshal(newInstanceMetadata(uid, created)) + if err != nil { + t.Fatal(err) + } + return string(out) +} + +func ipv4(addr string) libvirt.DomainIPAddr { + return libvirt.DomainIPAddr{Type: int32(libvirt.IPAddrTypeIpv4), Addr: addr, Prefix: 24} +} + +func ipv6(addr string) libvirt.DomainIPAddr { + return libvirt.DomainIPAddr{Type: int32(libvirt.IPAddrTypeIpv6), Addr: addr, Prefix: 64} +} + +// managedDomain seeds a domain as Create would have left it. +func managedDomain(t *testing.T, api *fakeAPI, name, uid string, running bool, addrs map[uint32][]libvirt.DomainInterface) { + t.Helper() + api.putDomain(&fakeDomain{name: name, metadata: metadataXML(t, uid, testNow), running: running, addrs: addrs}) + api.putVol("default", overlayVolumeName(name), formatQcow2, nil) + api.putVol("default", seedVolumeName(name), formatRaw, []byte("iso")) +} + +func wantClass(t *testing.T, err error, class error) { + t.Helper() + if err == nil { + t.Fatalf("got nil error, want %v", class) + } + if !errors.Is(err, class) { + t.Fatalf("error class = %v, want %v (err: %v)", provider.Class(err), class, err) + } +} + +func assertCreated(t *testing.T, api *fakeAPI) { + t.Helper() + dom := api.domain(testName) + if dom == nil { + t.Fatal("domain not defined") + } + if !dom.running { + t.Error("domain not running") + } + overlay := api.vol("default", overlayVolumeName(testName)) + if overlay == nil { + t.Fatal("overlay volume missing") + } + if overlay.backing != "/var/lib/libvirt/images/base-debian-13.qcow2" || overlay.format != formatQcow2 { + t.Errorf("overlay = %+v, want qcow2 backed by the cached base", overlay) + } + if overlay.capacity != defaultDiskSizeGB*bytesPerGiB { + t.Errorf("overlay capacity = %d, want %d", overlay.capacity, defaultDiskSizeGB*bytesPerGiB) + } + seed := api.vol("default", seedVolumeName(testName)) + if seed == nil { + t.Fatal("seed volume missing") + } + if seed.format != formatRaw || len(seed.data) == 0 || int64(len(seed.data)) != seed.capacity { + t.Errorf("seed = format %s, %d bytes, capacity %d; want raw with capacity == uploaded size", seed.format, len(seed.data), seed.capacity) + } +} + +func TestCreate_freshHost(t *testing.T) { + t.Parallel() + api := readyHost() + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + + id, err := p.Create(context.Background(), testRequest()) + if err != nil { + t.Fatalf("Create: %v", err) + } + if want := "hosts/nas1/domains/" + testName; id != want { + t.Errorf("providerID = %s, want %s", id, want) + } + assertCreated(t, api) + + var d domainXML + if err := xml.Unmarshal([]byte(api.domain(testName).xml), &d); err != nil { + t.Fatal(err) + } + if d.VCPU != 2 || d.Memory.Value != 2048 || d.Memory.Unit != "MiB" { + t.Errorf("domain size = %d vcpu / %d %s, want 2 / 2048 MiB", d.VCPU, d.Memory.Value, d.Memory.Unit) + } + if d.Devices.Interfaces[0].Source.Bridge != "br0" { + t.Errorf("bridge = %q, want br0", d.Devices.Interfaces[0].Source.Bridge) + } + if m := d.Metadata.Instance; m.UID != testUID || !m.managed() || !m.createdAt().Equal(testNow) { + t.Errorf("metadata = %+v, want managed, uid %s, createdAt %s", m, testUID, testNow) + } + if api.hasCall("StorageVolDownload") || api.callCount("StorageVolUpload") != 1 { + t.Errorf("calls = %v, want no image download and exactly one upload (the seed)", api.calls) + } +} + +func TestCreate_fetchesImageOnFirstUse(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + cfg := testConfig("nas1") + cfg.KVM.Images["debian-13"] = provider.KVMImage{URL: srv.URL + "/debian.qcow2", SHA256: testImageSHA()} + p := newWithDialer(cfg, hostFakes{"nas1": api}.dial, nil) + + if _, err := p.Create(context.Background(), testRequest()); err != nil { + t.Fatalf("Create: %v", err) + } + if srv.hits.Load() != 1 { + t.Errorf("image server hits = %d, want 1", srv.hits.Load()) + } + base := api.vol("default", "base-debian-13.qcow2") + if base == nil || string(base.data) != string(testImage) { + t.Fatal("base image not cached in the pool") + } + if api.domain(testName) == nil || !api.domain(testName).running { + t.Error("domain not running after image fetch") + } + + // A second Proxy with the same image on the same host reuses the cache. + req := testRequest() + req.Name, req.UID = "proxy-second", "uid-2" + if _, err := p.Create(context.Background(), req); err != nil { + t.Fatalf("second Create: %v", err) + } + if srv.hits.Load() != 1 { + t.Errorf("image server hits after second Create = %d, want still 1", srv.hits.Load()) + } +} + +func TestCreate_isIdempotentWhenComplete(t *testing.T) { + t.Parallel() + api := readyHost() + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + if _, err := p.Create(context.Background(), testRequest()); err != nil { + t.Fatal(err) + } + before := len(api.calls) + + id, err := p.Create(context.Background(), testRequest()) + if err != nil { + t.Fatalf("second Create: %v", err) + } + if id != "hosts/nas1/domains/"+testName { + t.Errorf("providerID = %s", id) + } + for _, m := range []string{"DomainDefineXML", "DomainCreate", "StorageVolUpload"} { + if n := api.callCount(m); n != 1 { + t.Errorf("%s called %d times across two Creates, want 1", m, n) + } + } + if len(api.calls) == before { + t.Error("second Create made no calls at all; expected lookups") + } +} + +// Each case breaks Create after a different step, then re-runs it and +// expects the second attempt to pick up where the first stopped without +// duplicating what already exists. +func TestCreate_resumesAfterPartialFailure(t *testing.T) { + t.Parallel() + cases := []struct { + name string + sabotage func(api *fakeAPI) + class error + check func(t *testing.T, api *fakeAPI) + }{ + { + name: "seed upload fails after overlay exists", + sabotage: func(api *fakeAPI) { + api.uploadHook = func(vol string, _ io.Reader) error { + api.uploadHook = nil + return io.ErrUnexpectedEOF + } + }, + class: provider.ErrTransient, + check: func(t *testing.T, api *fakeAPI) { + if n := api.callCount("StorageVolUpload"); n != 2 { + t.Errorf("seed uploads = %d, want 2 (failed, then redone)", n) + } + if n := api.callCount("StorageVolCreateXML"); n != 4 { + t.Errorf("volume creates = %d, want 4 (overlay, seed; overlay rejected as existing, seed again)", n) + } + if names := api.volNames("default"); len(names) != 4 { + t.Errorf("pool holds %v, want base, marker, one overlay, one seed", names) + } + }, + }, + { + name: "define fails after volumes exist", + sabotage: func(api *fakeAPI) { api.failOnce["DomainDefineXML"] = lvError(libvirt.ErrInternalError, "boom") }, + class: provider.ErrTransient, + check: func(t *testing.T, api *fakeAPI) { + if n := api.callCount("StorageVolUpload"); n != 2 { + t.Errorf("seed uploads = %d, want 2 (seed rebuilt while no domain exists)", n) + } + if n := api.callCount("DomainDefineXML"); n != 2 { + t.Errorf("defines = %d, want 2", n) + } + }, + }, + { + name: "start fails after define", + sabotage: func(api *fakeAPI) { api.failOnce["DomainCreate"] = lvError(libvirt.ErrInternalError, "boom") }, + class: provider.ErrTransient, + check: func(t *testing.T, api *fakeAPI) { + if n := api.callCount("DomainDefineXML"); n != 1 { + t.Errorf("defines = %d, want 1 (existing domain reused)", n) + } + if n := api.callCount("DomainCreate"); n != 2 { + t.Errorf("starts = %d, want 2", n) + } + }, + }, + { + name: "transport drops mid-create", + sabotage: func(api *fakeAPI) { api.failOnce["DomainLookupByName"] = io.EOF }, + class: provider.ErrTransient, + check: func(t *testing.T, api *fakeAPI) { + if n := api.callCount("DomainDefineXML"); n != 1 { + t.Errorf("defines = %d, want 1", n) + } + }, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + api := readyHost() + tc.sabotage(api) + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + + _, err := p.Create(context.Background(), testRequest()) + wantClass(t, err, tc.class) + if _, err := p.Create(context.Background(), testRequest()); err != nil { + t.Fatalf("retry Create: %v", err) + } + assertCreated(t, api) + tc.check(t, api) + }) + } +} + +// A reconcile cancelled mid-upload leaves a seed volume whose content is +// short; the compensating delete cannot run because the pool has already +// closed the connection. The next Create must not trust that volume. +func TestCreate_replacesLeftoverSeed(t *testing.T) { + t.Parallel() + api := readyHost() + api.putVol("default", seedVolumeName(testName), formatRaw, []byte("trunc")) + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + + if _, err := p.Create(context.Background(), testRequest()); err != nil { + t.Fatal(err) + } + assertCreated(t, api) + if seed := api.vol("default", seedVolumeName(testName)); string(seed.data) == "trunc" { + t.Error("leftover seed was kept") + } + if n := api.callCount("StorageVolUpload"); n != 1 { + t.Errorf("uploads = %d, want 1 (the rebuilt seed)", n) + } +} + +func TestCreate_startsStoppedDomain(t *testing.T) { + t.Parallel() + api := readyHost() + managedDomain(t, api, testName, testUID, false, nil) + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + + if _, err := p.Create(context.Background(), testRequest()); err != nil { + t.Fatal(err) + } + if !api.domain(testName).running { + t.Error("domain still stopped") + } + if api.hasCall("DomainDefineXML") || api.hasCall("StorageVolUpload") || api.hasCall("StorageVolDelete") { + t.Errorf("calls = %v, want no define, upload or volume delete for an existing domain", api.calls) + } +} + +func TestCreate_singleHostNeedsNoZone(t *testing.T) { + t.Parallel() + api := readyHost() + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + req := testRequest() + req.Placement.Zone = "" + + id, err := p.Create(context.Background(), req) + if err != nil { + t.Fatal(err) + } + if id != "hosts/nas1/domains/"+testName { + t.Errorf("providerID = %s, want the single host filled in", id) + } +} + +func TestCreate_placementErrors(t *testing.T) { + t.Parallel() + cases := []struct { + name string + hosts []string + edit func(*provider.CreateRequest) + want string + }{ + {"unknown host", []string{"nas1"}, func(r *provider.CreateRequest) { r.Placement.Zone = "nas9" }, "not in kvm.hosts"}, + {"empty zone with two hosts", []string{"nas1", "nas2"}, func(r *provider.CreateRequest) { r.Placement.Zone = "" }, "placement.zone is required"}, + {"empty zone with no hosts", nil, func(r *provider.CreateRequest) { r.Placement.Zone = "" }, "kvm.hosts is empty"}, + {"unknown machine type", []string{"nas1"}, func(r *provider.CreateRequest) { r.Placement.MachineType = "huge" }, "not in kvm.machineTypes"}, + {"empty machine type", []string{"nas1"}, func(r *provider.CreateRequest) { r.Placement.MachineType = "" }, "not in kvm.machineTypes"}, + {"unknown image", []string{"nas1"}, func(r *provider.CreateRequest) { r.Placement.Image = "windows" }, "neither a key"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fakes := hostFakes{} + for _, h := range tc.hosts { + fakes[h] = readyHost() + } + p := newTestProvider(fakes, tc.hosts...) + req := testRequest() + tc.edit(&req) + + _, err := p.Create(context.Background(), req) + wantClass(t, err, provider.ErrPermanent) + if !strings.Contains(err.Error(), tc.want) { + t.Errorf("error = %q, want it to mention %q", err, tc.want) + } + for _, api := range fakes { + if len(api.calls) != 0 { + t.Errorf("host was called: %v", api.calls) + } + } + }) + } +} + +func TestCreate_hostFailures(t *testing.T) { + t.Parallel() + cases := []struct { + name string + setup func() (hostFakes, string) + class error + }{ + { + name: "host unreachable", + setup: func() (hostFakes, string) { return hostFakes{}, "nas1" }, + class: provider.ErrTransient, + }, + { + name: "pool missing", + setup: func() (hostFakes, string) { + api := newFakeAPI("other") + return hostFakes{"nas1": api}, "nas1" + }, + class: provider.ErrPermanent, + }, + { + name: "domain xml rejected", + setup: func() (hostFakes, string) { + api := readyHost() + api.fail["DomainDefineXML"] = lvError(libvirt.ErrConfigUnsupported, "no q35 here") + return hostFakes{"nas1": api}, "nas1" + }, + class: provider.ErrPermanent, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + fakes, host := tc.setup() + p := newTestProvider(fakes, host) + _, err := p.Create(context.Background(), testRequest()) + wantClass(t, err, tc.class) + }) + } +} + +func TestCreate_cancelledContext(t *testing.T) { + t.Parallel() + api := readyHost() + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := p.Create(ctx, testRequest()) + wantClass(t, err, provider.ErrTransient) + if !errors.Is(err, context.Canceled) { + t.Errorf("error = %v, want it to unwrap to context.Canceled", err) + } +} + +func TestGet(t *testing.T) { + t.Parallel() + arp := map[uint32][]libvirt.DomainInterface{ + uint32(libvirt.DomainInterfaceAddressesSrcArp): {{Name: "vnet0", Addrs: []libvirt.DomainIPAddr{ipv4("192.168.1.50")}}}, + } + agentOnly := map[uint32][]libvirt.DomainInterface{ + uint32(libvirt.DomainInterfaceAddressesSrcAgent): { + {Name: "lo", Addrs: []libvirt.DomainIPAddr{ipv4("127.0.0.1"), ipv6("::1")}}, + {Name: "enp1s0", Addrs: []libvirt.DomainIPAddr{ipv6("fe80::1"), ipv4("192.168.1.51")}}, + }, + } + linkLocalOnly := map[uint32][]libvirt.DomainInterface{ + uint32(libvirt.DomainInterfaceAddressesSrcAgent): { + {Name: "enp1s0", Addrs: []libvirt.DomainIPAddr{ipv4("169.254.10.10")}}, + }, + } + cases := []struct { + name string + running bool + addrs map[uint32][]libvirt.DomainInterface + state provider.InstanceState + ip string + }{ + {"running with arp address", true, arp, provider.StateRunning, "192.168.1.50"}, + {"running, agent only, skips loopback and v6", true, agentOnly, provider.StateRunning, "192.168.1.51"}, + {"running, no address yet", true, nil, provider.StateProvisioning, ""}, + {"running, only link-local", true, linkLocalOnly, provider.StateProvisioning, ""}, + {"shut off", false, arp, provider.StateStopped, ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + api := readyHost() + managedDomain(t, api, testName, testUID, tc.running, tc.addrs) + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + + inst, err := p.Get(context.Background(), "hosts/nas1/domains/"+testName) + if err != nil { + t.Fatalf("Get: %v", err) + } + want := provider.Instance{ID: "hosts/nas1/domains/" + testName, IP: tc.ip, State: tc.state, UID: testUID, CreatedAt: testNow} + if *inst != want { + t.Errorf("instance = %+v, want %+v", *inst, want) + } + }) + } +} + +func TestGet_unmanagedDomainHasNoUID(t *testing.T) { + t.Parallel() + api := readyHost() + api.putDomain(&fakeDomain{name: "hand-made", running: true}) + p := newTestProvider(hostFakes{"nas1": api}, "nas1") + + inst, err := p.Get(context.Background(), "hosts/nas1/domains/hand-made") + if err != nil { + t.Fatalf("Get: %v", err) + } + if inst.UID != "" || !inst.CreatedAt.IsZero() { + t.Errorf("instance = %+v, want empty UID and zero CreatedAt", inst) + } +} + +func TestGet_errors(t *testing.T) { + t.Parallel() + cases := []struct { + name string + id string + setup func(t *testing.T, api *fakeAPI) + class error + }{ + {"missing domain", "hosts/nas1/domains/" + testName, func(*testing.T, *fakeAPI) {}, provider.ErrNotFound}, + {"malformed id", "zones/x/instances/y", func(*testing.T, *fakeAPI) {}, provider.ErrPermanent}, + {"host not configured", "hosts/nas9/domains/" + testName, func(*testing.T, *fakeAPI) {}, provider.ErrPermanent}, + { + "state read fails", "hosts/nas1/domains/" + testName, + func(t *testing.T, api *fakeAPI) { + managedDomain(t, api, testName, testUID, true, nil) + api.fail["DomainGetState"] = lvError(libvirt.ErrInternalError, "boom") + }, + provider.ErrTransient, + }, + { + "corrupt metadata", "hosts/nas1/domains/" + testName, + func(_ *testing.T, api *fakeAPI) { + api.putDomain(&fakeDomain{name: testName, metadata: "false`}) + p := newTestProvider(hostFakes{"nas1": nas1, "nas2": nas2}, "nas1", "nas2") + + got, err := p.ListByTag(context.Background()) + if err != nil { + t.Fatalf("ListByTag: %v", err) + } + want := []provider.Instance{ + {ID: "hosts/nas1/domains/proxy-one", IP: "192.168.1.10", State: provider.StateRunning, UID: "uid-1", CreatedAt: testNow}, + {ID: "hosts/nas2/domains/proxy-two", State: provider.StateStopped, UID: "uid-2", CreatedAt: testNow}, + } + if len(got) != len(want) { + t.Fatalf("got %d instances %+v, want %d", len(got), got, len(want)) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("instance[%d] = %+v, want %+v", i, got[i], want[i]) + } + } +} + +func TestListByTag_oneHostDownFailsSweep(t *testing.T) { + t.Parallel() + nas1 := readyHost() + managedDomain(t, nas1, "proxy-one", "uid-1", true, nil) + p := newTestProvider(hostFakes{"nas1": nas1}, "nas1", "nas2") + + got, err := p.ListByTag(context.Background()) + wantClass(t, err, provider.ErrTransient) + if got != nil { + t.Errorf("got %+v alongside the error, want nil (no partial view)", got) + } +} + +// A domain that cannot be read is left out of the sweep rather than +// failing it: GC never deletes what it does not see, so skipping is the +// safe direction, while one bad domain blocking GC for all others is not. +func TestListByTag_skipsUnreadableDomain(t *testing.T) { + t.Parallel() + cases := []struct { + name string + setup func(t *testing.T, api *fakeAPI) + }{ + { + "deleted mid-sweep", + func(t *testing.T, api *fakeAPI) { + managedDomain(t, api, "proxy-one", "uid-1", true, nil) + api.failOnce["DomainGetMetadata"] = lvError(libvirt.ErrNoDomain, "gone") + }, + }, + { + "corrupt metadata", + func(_ *testing.T, api *fakeAPI) { + api.putDomain(&fakeDomain{name: "proxy-one", metadata: "