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 533ebf8..beebc58 100644 --- a/docs/plans-executions/2026-09-04-1404-kvm-provider.md +++ b/docs/plans-executions/2026-09-04-1404-kvm-provider.md @@ -6,7 +6,7 @@ Plan: [2026-09-04-1404-kvm-provider.md](../plans/2026-09-04-1404-kvm-provider.md - [x] Step 1 — Config (`KVMConfig` & friends, `validate()` case, tests) - [x] Step 2 — Pure builders (`domainxml.go`, `volume.go`, `seed.go`, `id.go`, `metadata.go`) - [x] Step 3 — Transport + errors (`conn.go`, `errors.go`) -- [ ] Step 3b — Image cache (`image.go`) +- [x] Step 3b — Image cache (`image.go`) - [ ] 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 @@ -151,3 +151,48 @@ 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`. + +## Step 3b — Image cache + +`image.go` follows the plan closely: `resolveImage` maps `placement.image` +to a configured key or a direct http(s) URL (else Permanent); +`imageCache.ensure` checks base volume + marker (+ marker content when a +checksum is configured), tears down and re-fetches when anything is off, +streams the HTTP body through `StorageVolUpload` with a `sha256` tee, +refreshes the pool, then writes the marker last. A per-`(host, volume)` +mutex serialises concurrent callers; the pool itself is the only cache +state. + +Errors from this file are pre-tagged with the taxonomy sentinel via +`fmt.Errorf("%w: …", provider.ErrPermanent, …)` and `classify` now honours +such tags before looking at libvirt codes. That keeps one `wrapErr` path in +the provider instead of special-casing image errors. + +`fake_test.go` landed with this step rather than Step 4: an in-memory +`libvirtAPI` with pools, volumes (content, format, capacity, backing path), +domains (state, metadata, per-source interface addresses), per-method error +injection (`fail` always / `failOnce`) and an `uploadHook` to break a stream +mid-way. It models the libvirt error codes the provider branches on +(`ErrStorageVolExist`, `ErrNoStorageVol`, `ErrNoDomain`, +`ErrOperationInvalid`, `ErrNoDomainMetadata`, `ErrAgentUnresponsive`). + +Tests cover every case in the plan plus: no checksum configured (marker is +still written with the computed digest; second call needs no marker read), +403/500 alongside 404/503, missing pool → Permanent, cancelled ctx → +Transient with nothing created, and different hosts fetching independently. + +One test-harness lesson worth keeping: `httptest`'s server drops +`Content-Length` and switches to chunked encoding for bodies larger than +its 2 KiB write buffer, so the fake mirror sets the header explicitly. +Real distro mirrors send it; a server that doesn't is reported as +Permanent with a message naming the header. + +Worth noting: the base volume is created with `capacity = Content-Length` +and the *image's* format (`qcow2`), and the raw bytes are uploaded over it. +That relies on libvirt's upload writing the file verbatim (which is how +`virsh vol-upload` populates volumes); the following `StoragePoolRefresh` +makes libvirt re-read the real virtual size from the qcow2 header. Step 6 +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. diff --git a/internal/provider/kvm/errors.go b/internal/provider/kvm/errors.go index cacf2f0..51bb20e 100644 --- a/internal/provider/kvm/errors.go +++ b/internal/provider/kvm/errors.go @@ -9,14 +9,21 @@ import ( "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. +// classify maps an error onto the provider taxonomy. Errors the package +// already tagged with a sentinel (image download failures, placement +// errors) keep their class. For libvirt errors: 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 { + for _, class := range []error{provider.ErrNotFound, provider.ErrQuotaExceeded, provider.ErrPermanent, provider.ErrTransient} { + if errors.Is(err, class) { + return class + } + } var lerr libvirt.Error if !errors.As(err, &lerr) { return provider.ErrTransient diff --git a/internal/provider/kvm/errors_test.go b/internal/provider/kvm/errors_test.go index 9369c49..1f00ad7 100644 --- a/internal/provider/kvm/errors_test.go +++ b/internal/provider/kvm/errors_test.go @@ -38,6 +38,8 @@ func TestClassify(t *testing.T) { {"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}, + {"pre-tagged permanent", fmt.Errorf("%w: image returned HTTP 404", provider.ErrPermanent), provider.ErrPermanent}, + {"pre-tagged transient wrapping libvirt not-found", fmt.Errorf("%w: %w", provider.ErrTransient, lvErr(libvirt.ErrNoDomain)), provider.ErrTransient}, {"transport eof", io.EOF, provider.ErrTransient}, {"context cancelled", context.Canceled, provider.ErrTransient}, {"plain error", errors.New("ssh: handshake failed"), provider.ErrTransient}, diff --git a/internal/provider/kvm/fake_test.go b/internal/provider/kvm/fake_test.go new file mode 100644 index 0000000..02c52e4 --- /dev/null +++ b/internal/provider/kvm/fake_test.go @@ -0,0 +1,408 @@ +package kvm + +import ( + "bytes" + "encoding/xml" + "fmt" + "io" + "slices" + "sync" + + libvirt "github.com/digitalocean/go-libvirt" +) + +// fakeAPI is an in-memory libvirtAPI: pools of volumes and a set of +// domains, with per-method error injection. It models just enough libvirt +// semantics for the provider's decisions to be observable — ErrStorageVolExist +// on duplicate create, ErrNoDomain/ErrNoStorageVol on missing lookups, +// ErrOperationInvalid on destroying a stopped domain, and so on. +type fakeAPI struct { + mu sync.Mutex + poolPath string + pools map[string]map[string]*fakeVol + domains map[string]*fakeDomain + calls []string + fail map[string]error + failOnce map[string]error + // uploadHook, when set, sees every StorageVolUpload before the fake + // reads the stream; it can return an error to simulate a broken upload. + uploadHook func(vol string, r io.Reader) error +} + +type fakeVol struct { + name string + format string + capacity int64 + backing string + data []byte +} + +type fakeDomain struct { + name string + xml string + metadata string + running bool + // addrs is what DomainInterfaceAddresses returns per source. + addrs map[uint32][]libvirt.DomainInterface +} + +func newFakeAPI(pools ...string) *fakeAPI { + f := &fakeAPI{ + poolPath: "/var/lib/libvirt/images", + pools: map[string]map[string]*fakeVol{}, + domains: map[string]*fakeDomain{}, + fail: map[string]error{}, + failOnce: map[string]error{}, + } + for _, p := range pools { + f.pools[p] = map[string]*fakeVol{} + } + return f +} + +// enter records the call and returns any injected error for method. +func (f *fakeAPI) enter(method string) error { + f.calls = append(f.calls, method) + if err, ok := f.failOnce[method]; ok { + delete(f.failOnce, method) + return err + } + return f.fail[method] +} + +func (f *fakeAPI) callCount(method string) int { + f.mu.Lock() + defer f.mu.Unlock() + n := 0 + for _, c := range f.calls { + if c == method { + n++ + } + } + return n +} + +func (f *fakeAPI) hasCall(method string) bool { return f.callCount(method) > 0 } + +func (f *fakeAPI) vol(pool, name string) *fakeVol { + f.mu.Lock() + defer f.mu.Unlock() + return f.pools[pool][name] +} + +func (f *fakeAPI) putVol(pool, name, format string, data []byte) { + f.mu.Lock() + defer f.mu.Unlock() + f.pools[pool][name] = &fakeVol{name: name, format: format, capacity: int64(len(data)), data: data} +} + +func (f *fakeAPI) volNames(pool string) []string { + f.mu.Lock() + defer f.mu.Unlock() + var out []string + for n := range f.pools[pool] { + out = append(out, n) + } + slices.Sort(out) + return out +} + +func (f *fakeAPI) domain(name string) *fakeDomain { + f.mu.Lock() + defer f.mu.Unlock() + return f.domains[name] +} + +func (f *fakeAPI) putDomain(d *fakeDomain) { + f.mu.Lock() + defer f.mu.Unlock() + f.domains[d.name] = d +} + +func lvError(code libvirt.ErrorNumber, format string, args ...any) error { + return libvirt.Error{Code: uint32(code), Message: fmt.Sprintf(format, args...)} +} + +// --- domains --- + +func (f *fakeAPI) DomainLookupByName(name string) (libvirt.Domain, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainLookupByName"); err != nil { + return libvirt.Domain{}, err + } + if _, ok := f.domains[name]; !ok { + return libvirt.Domain{}, lvError(libvirt.ErrNoDomain, "Domain not found: no domain with matching name '%s'", name) + } + return libvirt.Domain{Name: name}, nil +} + +func (f *fakeAPI) DomainDefineXML(raw string) (libvirt.Domain, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainDefineXML"); err != nil { + return libvirt.Domain{}, err + } + var d domainXML + if err := xml.Unmarshal([]byte(raw), &d); err != nil { + return libvirt.Domain{}, lvError(libvirt.ErrXMLError, "%v", err) + } + meta, err := xml.Marshal(d.Metadata.Instance) + if err != nil { + return libvirt.Domain{}, lvError(libvirt.ErrXMLError, "%v", err) + } + existing := f.domains[d.Name] + nd := &fakeDomain{name: d.Name, xml: raw, metadata: string(meta)} + if existing != nil { + nd.running = existing.running + nd.addrs = existing.addrs + } + f.domains[d.Name] = nd + return libvirt.Domain{Name: d.Name}, nil +} + +func (f *fakeAPI) DomainCreate(d libvirt.Domain) error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainCreate"); err != nil { + return err + } + dom, ok := f.domains[d.Name] + if !ok { + return lvError(libvirt.ErrNoDomain, "no domain %s", d.Name) + } + if dom.running { + return lvError(libvirt.ErrOperationInvalid, "domain is already running") + } + dom.running = true + return nil +} + +func (f *fakeAPI) DomainDestroy(d libvirt.Domain) error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainDestroy"); err != nil { + return err + } + dom, ok := f.domains[d.Name] + if !ok { + return lvError(libvirt.ErrNoDomain, "no domain %s", d.Name) + } + if !dom.running { + return lvError(libvirt.ErrOperationInvalid, "domain is not running") + } + dom.running = false + return nil +} + +func (f *fakeAPI) DomainUndefineFlags(d libvirt.Domain, _ libvirt.DomainUndefineFlagsValues) error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainUndefineFlags"); err != nil { + return err + } + if _, ok := f.domains[d.Name]; !ok { + return lvError(libvirt.ErrNoDomain, "no domain %s", d.Name) + } + delete(f.domains, d.Name) + return nil +} + +func (f *fakeAPI) DomainGetState(d libvirt.Domain, _ uint32) (int32, int32, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainGetState"); err != nil { + return 0, 0, err + } + dom, ok := f.domains[d.Name] + if !ok { + return 0, 0, lvError(libvirt.ErrNoDomain, "no domain %s", d.Name) + } + if dom.running { + return int32(libvirt.DomainRunning), 0, nil + } + return int32(libvirt.DomainShutoff), 0, nil +} + +func (f *fakeAPI) DomainGetMetadata(d libvirt.Domain, typ int32, uri libvirt.OptString, _ libvirt.DomainModificationImpact) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainGetMetadata"); err != nil { + return "", err + } + dom, ok := f.domains[d.Name] + if !ok { + return "", lvError(libvirt.ErrNoDomain, "no domain %s", d.Name) + } + if typ != int32(libvirt.DomainMetadataElement) || len(uri) != 1 || uri[0] != metadataURI || dom.metadata == "" { + return "", lvError(libvirt.ErrNoDomainMetadata, "Requested metadata element is not present") + } + return dom.metadata, nil +} + +func (f *fakeAPI) DomainInterfaceAddresses(d libvirt.Domain, source uint32, _ uint32) ([]libvirt.DomainInterface, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("DomainInterfaceAddresses"); err != nil { + return nil, err + } + dom, ok := f.domains[d.Name] + if !ok { + return nil, lvError(libvirt.ErrNoDomain, "no domain %s", d.Name) + } + if !dom.running { + return nil, lvError(libvirt.ErrOperationInvalid, "domain is not running") + } + if source == uint32(libvirt.DomainInterfaceAddressesSrcAgent) && dom.addrs[source] == nil { + return nil, lvError(libvirt.ErrAgentUnresponsive, "Guest agent is not responding") + } + return dom.addrs[source], nil +} + +func (f *fakeAPI) ConnectListAllDomains(_ int32, _ libvirt.ConnectListAllDomainsFlags) ([]libvirt.Domain, uint32, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("ConnectListAllDomains"); err != nil { + return nil, 0, err + } + names := make([]string, 0, len(f.domains)) + for n := range f.domains { + names = append(names, n) + } + slices.Sort(names) + out := make([]libvirt.Domain, 0, len(names)) + for _, n := range names { + out = append(out, libvirt.Domain{Name: n}) + } + return out, uint32(len(out)), nil +} + +// --- storage --- + +func (f *fakeAPI) StoragePoolLookupByName(name string) (libvirt.StoragePool, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StoragePoolLookupByName"); err != nil { + return libvirt.StoragePool{}, err + } + if _, ok := f.pools[name]; !ok { + return libvirt.StoragePool{}, lvError(libvirt.ErrNoStoragePool, "Storage pool not found: no storage pool with matching name '%s'", name) + } + return libvirt.StoragePool{Name: name}, nil +} + +func (f *fakeAPI) StoragePoolRefresh(pool libvirt.StoragePool, _ uint32) error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StoragePoolRefresh"); err != nil { + return err + } + if _, ok := f.pools[pool.Name]; !ok { + return lvError(libvirt.ErrNoStoragePool, "no pool %s", pool.Name) + } + return nil +} + +func (f *fakeAPI) StorageVolLookupByName(pool libvirt.StoragePool, name string) (libvirt.StorageVol, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StorageVolLookupByName"); err != nil { + return libvirt.StorageVol{}, err + } + if _, ok := f.pools[pool.Name][name]; !ok { + return libvirt.StorageVol{}, lvError(libvirt.ErrNoStorageVol, "Storage volume not found: no storage vol with matching name '%s'", name) + } + return libvirt.StorageVol{Pool: pool.Name, Name: name, Key: f.poolPath + "/" + name}, nil +} + +func (f *fakeAPI) StorageVolGetPath(vol libvirt.StorageVol) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StorageVolGetPath"); err != nil { + return "", err + } + if _, ok := f.pools[vol.Pool][vol.Name]; !ok { + return "", lvError(libvirt.ErrNoStorageVol, "no vol %s", vol.Name) + } + return f.poolPath + "/" + vol.Name, nil +} + +func (f *fakeAPI) StorageVolCreateXML(pool libvirt.StoragePool, raw string, _ libvirt.StorageVolCreateFlags) (libvirt.StorageVol, error) { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StorageVolCreateXML"); err != nil { + return libvirt.StorageVol{}, err + } + vols, ok := f.pools[pool.Name] + if !ok { + return libvirt.StorageVol{}, lvError(libvirt.ErrNoStoragePool, "no pool %s", pool.Name) + } + var v volumeXML + if err := xml.Unmarshal([]byte(raw), &v); err != nil { + return libvirt.StorageVol{}, lvError(libvirt.ErrXMLError, "%v", err) + } + if _, exists := vols[v.Name]; exists { + return libvirt.StorageVol{}, lvError(libvirt.ErrStorageVolExist, "storage volume '%s' exists already", v.Name) + } + fv := &fakeVol{name: v.Name, format: v.Target.Format.Type, capacity: v.Capacity.Value} + if v.BackingStore != nil { + fv.backing = v.BackingStore.Path + } + vols[v.Name] = fv + return libvirt.StorageVol{Pool: pool.Name, Name: v.Name, Key: f.poolPath + "/" + v.Name}, nil +} + +func (f *fakeAPI) StorageVolUpload(vol libvirt.StorageVol, r io.Reader, _, _ uint64, _ libvirt.StorageVolUploadFlags) error { + f.mu.Lock() + if err := f.enter("StorageVolUpload"); err != nil { + f.mu.Unlock() + return err + } + fv, ok := f.pools[vol.Pool][vol.Name] + hook := f.uploadHook + f.mu.Unlock() + if !ok { + return lvError(libvirt.ErrNoStorageVol, "no vol %s", vol.Name) + } + if hook != nil { + if err := hook(vol.Name, r); err != nil { + return err + } + } + // Read outside the lock: the stream may be a slow HTTP body. + data, err := io.ReadAll(r) + if err != nil { + return fmt.Errorf("reading upload stream: %w", err) + } + f.mu.Lock() + defer f.mu.Unlock() + fv.data = data + return nil +} + +func (f *fakeAPI) StorageVolDownload(vol libvirt.StorageVol, w io.Writer, _, _ uint64, _ libvirt.StorageVolDownloadFlags) error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StorageVolDownload"); err != nil { + return err + } + fv, ok := f.pools[vol.Pool][vol.Name] + if !ok { + return lvError(libvirt.ErrNoStorageVol, "no vol %s", vol.Name) + } + _, err := io.Copy(w, bytes.NewReader(fv.data)) + return err +} + +func (f *fakeAPI) StorageVolDelete(vol libvirt.StorageVol, _ libvirt.StorageVolDeleteFlags) error { + f.mu.Lock() + defer f.mu.Unlock() + if err := f.enter("StorageVolDelete"); err != nil { + return err + } + if _, ok := f.pools[vol.Pool][vol.Name]; !ok { + return lvError(libvirt.ErrNoStorageVol, "no vol %s", vol.Name) + } + delete(f.pools[vol.Pool], vol.Name) + return nil +} diff --git a/internal/provider/kvm/image.go b/internal/provider/kvm/image.go new file mode 100644 index 0000000..ffdc0e1 --- /dev/null +++ b/internal/provider/kvm/image.go @@ -0,0 +1,232 @@ +package kvm + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + + libvirt "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +// imageSource is a resolved placement.image: either an entry from the +// provider's images map (Key set) or a URL given directly. +type imageSource struct { + Key string + URL string + SHA256 string + Format string +} + +// resolveImage maps placement.image to a source. Anything that is neither a +// configured key nor an http(s) URL is a spec error the reconciler should +// surface, not retry. +func resolveImage(cfg provider.KVMConfig, image string) (imageSource, error) { + if img, ok := cfg.Images[image]; ok { + return imageSource{Key: image, URL: img.URL, SHA256: img.SHA256, Format: img.Format}, nil + } + u, err := url.Parse(image) + if err == nil && (u.Scheme == "http" || u.Scheme == "https") && u.Host != "" { + return imageSource{URL: image, Format: formatQcow2}, nil + } + return imageSource{}, fmt.Errorf("%w: placement.image %q is neither a key in kvm.images nor an http(s) URL", + provider.ErrPermanent, image) +} + +// volumeName is the cached base image's name in the pool: readable for +// configured keys, a URL hash for ad-hoc URLs. +func (s imageSource) volumeName() string { + if s.Key != "" { + return "base-" + s.Key + "." + s.Format + } + sum := sha256.Sum256([]byte(s.URL)) + return "base-" + hex.EncodeToString(sum[:8]) + "." + s.Format +} + +// markerName is the sidecar volume holding the sha256 of the verified +// upload. Its presence is what says "the base volume is complete". +func (s imageSource) markerName() string { + return s.volumeName() + ".sha256" +} + +// imageCache fetches base images into a host's pool on first use. It keeps +// no state about what is cached — the pool is the source of truth, checked +// on every call — only a per-(host, image) lock so concurrent Creates for +// the same image on the same host result in one download. +type imageCache struct { + http *http.Client + + mu sync.Mutex + locks map[string]*sync.Mutex +} + +func newImageCache(client *http.Client) *imageCache { + if client == nil { + client = http.DefaultClient + } + return &imageCache{http: client, locks: make(map[string]*sync.Mutex)} +} + +func (c *imageCache) lockFor(key string) *sync.Mutex { + c.mu.Lock() + defer c.mu.Unlock() + m, ok := c.locks[key] + if !ok { + m = &sync.Mutex{} + c.locks[key] = m + } + return m +} + +// ensure makes src present in pool and returns the base volume's path on +// the host. The image is downloaded synchronously through the libvirt +// stream when the pool does not already hold a verified copy. +func (c *imageCache) ensure(ctx context.Context, log logr.Logger, api libvirtAPI, hostName string, pool libvirt.StoragePool, src imageSource) (string, error) { + m := c.lockFor(hostName + "/" + src.volumeName()) + m.Lock() + defer m.Unlock() + + base, ready, err := c.lookup(api, pool, src) + if err != nil { + return "", err + } + if ready { + log.V(1).Info("base image already cached", "host", hostName, "volume", src.volumeName()) + return api.StorageVolGetPath(base) + } + if err := deleteIfExists(api, pool, src.markerName()); err != nil { + return "", err + } + if err := deleteIfExists(api, pool, src.volumeName()); err != nil { + return "", err + } + base, err = c.fetch(ctx, log, api, hostName, pool, src) + if err != nil { + return "", err + } + return api.StorageVolGetPath(base) +} + +// lookup reports whether the pool holds a complete, verified base volume: +// the volume and its marker both exist and, when a checksum is configured, +// the marker records that checksum. +func (c *imageCache) lookup(api libvirtAPI, pool libvirt.StoragePool, src imageSource) (libvirt.StorageVol, bool, error) { + base, err := api.StorageVolLookupByName(pool, src.volumeName()) + if hasCode(err, libvirt.ErrNoStorageVol) { + return libvirt.StorageVol{}, false, nil + } + if err != nil { + return libvirt.StorageVol{}, false, fmt.Errorf("looking up base volume: %w", err) + } + marker, err := api.StorageVolLookupByName(pool, src.markerName()) + if hasCode(err, libvirt.ErrNoStorageVol) { + return base, false, nil + } + if err != nil { + return libvirt.StorageVol{}, false, fmt.Errorf("looking up image marker: %w", err) + } + if src.SHA256 == "" { + return base, true, nil + } + var buf bytes.Buffer + if err := api.StorageVolDownload(marker, &buf, 0, 0, 0); err != nil { + return libvirt.StorageVol{}, false, fmt.Errorf("reading image marker: %w", err) + } + return base, strings.EqualFold(strings.TrimSpace(buf.String()), src.SHA256), nil +} + +// fetch streams the image from its URL straight into a new base volume, +// hashing on the way, then writes the marker. Nothing is buffered in the +// operator beyond the HTTP body's read buffer. +func (c *imageCache) fetch(ctx context.Context, log logr.Logger, api libvirtAPI, hostName string, pool libvirt.StoragePool, src imageSource) (libvirt.StorageVol, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, src.URL, nil) + if err != nil { + return libvirt.StorageVol{}, fmt.Errorf("%w: building image request: %v", provider.ErrPermanent, err) + } + resp, err := c.http.Do(req) + if err != nil { + return libvirt.StorageVol{}, fmt.Errorf("%w: fetching image %s: %v", provider.ErrTransient, src.URL, err) + } + defer resp.Body.Close() + switch { + case resp.StatusCode >= 200 && resp.StatusCode < 300: + case resp.StatusCode >= 400 && resp.StatusCode < 500: + return libvirt.StorageVol{}, fmt.Errorf("%w: image %s returned HTTP %d", provider.ErrPermanent, src.URL, resp.StatusCode) + default: + return libvirt.StorageVol{}, fmt.Errorf("%w: image %s returned HTTP %d", provider.ErrTransient, src.URL, resp.StatusCode) + } + if resp.ContentLength < 0 { + return libvirt.StorageVol{}, fmt.Errorf("%w: image %s has no Content-Length; the base volume needs a known size", provider.ErrPermanent, src.URL) + } + + log.Info("fetching base image into host pool", + "host", hostName, "pool", pool.Name, "volume", src.volumeName(), + "url", src.URL, "bytes", resp.ContentLength) + start := time.Now() + + baseXML, err := buildVolumeXML(volumeSpec{Name: src.volumeName(), SizeBytes: resp.ContentLength, Format: src.Format}) + if err != nil { + return libvirt.StorageVol{}, err + } + base, err := api.StorageVolCreateXML(pool, baseXML, 0) + if err != nil { + return libvirt.StorageVol{}, fmt.Errorf("creating base volume: %w", err) + } + hasher := sha256.New() + if err := api.StorageVolUpload(base, io.TeeReader(resp.Body, hasher), 0, 0, 0); err != nil { + _ = api.StorageVolDelete(base, 0) + return libvirt.StorageVol{}, fmt.Errorf("uploading image to base volume: %w", err) + } + digest := hex.EncodeToString(hasher.Sum(nil)) + if src.SHA256 != "" && !strings.EqualFold(digest, src.SHA256) { + _ = api.StorageVolDelete(base, 0) + return libvirt.StorageVol{}, fmt.Errorf("%w: image %s sha256 is %s, config expects %s", + provider.ErrPermanent, src.URL, digest, src.SHA256) + } + // libvirt sized the volume from Content-Length; a refresh makes it + // re-read the real virtual size from the uploaded qcow2 header. + if err := api.StoragePoolRefresh(pool, 0); err != nil { + return libvirt.StorageVol{}, fmt.Errorf("refreshing pool after upload: %w", err) + } + + markerXML, err := buildVolumeXML(rawVolume(src.markerName(), int64(len(digest)))) + if err != nil { + return libvirt.StorageVol{}, err + } + marker, err := api.StorageVolCreateXML(pool, markerXML, 0) + if err != nil { + return libvirt.StorageVol{}, fmt.Errorf("creating image marker: %w", err) + } + if err := api.StorageVolUpload(marker, strings.NewReader(digest), 0, 0, 0); err != nil { + _ = api.StorageVolDelete(marker, 0) + return libvirt.StorageVol{}, fmt.Errorf("writing image marker: %w", err) + } + log.Info("base image cached", + "host", hostName, "volume", src.volumeName(), "sha256", digest, + "bytes", resp.ContentLength, "duration", time.Since(start).Round(time.Second)) + return base, nil +} + +func deleteIfExists(api libvirtAPI, pool libvirt.StoragePool, name string) error { + vol, err := api.StorageVolLookupByName(pool, name) + if hasCode(err, libvirt.ErrNoStorageVol) { + return nil + } + if err != nil { + return fmt.Errorf("looking up volume %s: %w", name, err) + } + if err := api.StorageVolDelete(vol, 0); err != nil && !hasCode(err, libvirt.ErrNoStorageVol) { + return fmt.Errorf("deleting volume %s: %w", name, err) + } + return nil +} diff --git a/internal/provider/kvm/image_test.go b/internal/provider/kvm/image_test.go new file mode 100644 index 0000000..2493c44 --- /dev/null +++ b/internal/provider/kvm/image_test.go @@ -0,0 +1,385 @@ +package kvm + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/go-logr/logr" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" +) + +var testImage = bytes.Repeat([]byte("QFI\xfb fake qcow2 payload "), 1024) + +func testImageSHA() string { + sum := sha256.Sum256(testImage) + return hex.EncodeToString(sum[:]) +} + +// imageServer serves testImage and counts hits. Status overrides the +// response, noLength forces chunked encoding (no Content-Length), delay +// slows each response so concurrent callers overlap. +type imageServer struct { + *httptest.Server + hits atomic.Int32 + status int + noLength bool + delay time.Duration + body []byte +} + +func newImageServer(t *testing.T) *imageServer { + t.Helper() + s := &imageServer{status: http.StatusOK, body: testImage} + s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + s.hits.Add(1) + time.Sleep(s.delay) + if s.status != http.StatusOK { + w.WriteHeader(s.status) + return + } + // Bodies larger than the server's write buffer would otherwise go + // out chunked without a Content-Length, unlike real image mirrors. + if s.noLength { + w.(http.Flusher).Flush() + } else { + w.Header().Set("Content-Length", strconv.Itoa(len(s.body))) + } + _, _ = w.Write(s.body) + })) + t.Cleanup(s.Close) + return s +} + +func testSource(srv *imageServer, sha string) imageSource { + return imageSource{Key: "debian-13", URL: srv.URL + "/debian.qcow2", SHA256: sha, Format: formatQcow2} +} + +func ensureTest(t *testing.T, api *fakeAPI, src imageSource) (string, error) { + t.Helper() + c := newImageCache(nil) + pool, err := api.StoragePoolLookupByName("default") + if err != nil { + t.Fatal(err) + } + return c.ensure(context.Background(), logr.Discard(), api, "nas1", pool, src) +} + +func TestResolveImage(t *testing.T) { + t.Parallel() + cfg := withDefaults(provider.KVMConfig{Images: map[string]provider.KVMImage{ + "debian-13": {URL: "https://cloud.debian.org/d.qcow2", SHA256: "abc"}, + }}) + tests := []struct { + name string + image string + want imageSource + wantErr bool + }{ + {"configured key", "debian-13", imageSource{Key: "debian-13", URL: "https://cloud.debian.org/d.qcow2", SHA256: "abc", Format: "qcow2"}, false}, + {"direct https url", "https://x.example/img.qcow2", imageSource{URL: "https://x.example/img.qcow2", Format: "qcow2"}, false}, + {"direct http url", "http://mirror/img.qcow2", imageSource{URL: "http://mirror/img.qcow2", Format: "qcow2"}, false}, + {"unknown key", "ubuntu", imageSource{}, true}, + {"file url", "file:///tmp/x.qcow2", imageSource{}, true}, + {"empty", "", imageSource{}, true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got, err := resolveImage(cfg, tc.image) + if tc.wantErr { + if !errors.Is(err, provider.ErrPermanent) { + t.Fatalf("resolveImage(%q) error = %v, want ErrPermanent", tc.image, err) + } + return + } + if err != nil { + t.Fatalf("resolveImage(%q) error = %v", tc.image, err) + } + if got != tc.want { + t.Errorf("resolveImage(%q) = %+v, want %+v", tc.image, got, tc.want) + } + }) + } +} + +func TestImageSource_volumeNames(t *testing.T) { + t.Parallel() + keyed := imageSource{Key: "debian-13", URL: "https://a/b.qcow2", Format: "qcow2"} + if keyed.volumeName() != "base-debian-13.qcow2" || keyed.markerName() != "base-debian-13.qcow2.sha256" { + t.Errorf("keyed names = %s / %s", keyed.volumeName(), keyed.markerName()) + } + direct := imageSource{URL: "https://a/b.qcow2", Format: "qcow2"} + name := direct.volumeName() + if !strings.HasPrefix(name, "base-") || !strings.HasSuffix(name, ".qcow2") || len(name) != len("base-")+16+len(".qcow2") { + t.Errorf("direct-url name = %q, want base-<16 hex>.qcow2", name) + } + other := imageSource{URL: "https://a/c.qcow2", Format: "qcow2"} + if other.volumeName() == name { + t.Error("different URLs produced the same volume name") + } +} + +func TestImageCache_ensure_freshFetch(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + path, err := ensureTest(t, api, testSource(srv, testImageSHA())) + if err != nil { + t.Fatalf("ensure() error = %v", err) + } + if path != "/var/lib/libvirt/images/base-debian-13.qcow2" { + t.Errorf("path = %q", path) + } + base := api.vol("default", "base-debian-13.qcow2") + if base == nil || !bytes.Equal(base.data, testImage) { + t.Fatalf("base volume missing or content differs") + } + if base.format != "qcow2" || base.capacity != int64(len(testImage)) { + t.Errorf("base volume format/capacity = %s/%d, want qcow2/%d", base.format, base.capacity, len(testImage)) + } + marker := api.vol("default", "base-debian-13.qcow2.sha256") + if marker == nil || string(marker.data) != testImageSHA() { + t.Fatalf("marker missing or wrong: %+v", marker) + } + if !api.hasCall("StoragePoolRefresh") { + t.Error("pool was not refreshed after the upload") + } + if srv.hits.Load() != 1 { + t.Errorf("http hits = %d, want 1", srv.hits.Load()) + } +} + +func TestImageCache_ensure_noChecksumConfigured(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + if _, err := ensureTest(t, api, testSource(srv, "")); err != nil { + t.Fatalf("ensure() error = %v", err) + } + if marker := api.vol("default", "base-debian-13.qcow2.sha256"); marker == nil || string(marker.data) != testImageSHA() { + t.Errorf("marker should still record the computed digest, got %+v", marker) + } + // Second call: ready without a checksum to compare, so no marker read. + if _, err := ensureTest(t, api, testSource(srv, "")); err != nil { + t.Fatalf("second ensure() error = %v", err) + } + if srv.hits.Load() != 1 { + t.Errorf("http hits = %d, want 1", srv.hits.Load()) + } +} + +func TestImageCache_ensure_alreadyCached(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + api.putVol("default", "base-debian-13.qcow2", "qcow2", testImage) + api.putVol("default", "base-debian-13.qcow2.sha256", "raw", []byte(testImageSHA())) + path, err := ensureTest(t, api, testSource(srv, testImageSHA())) + if err != nil { + t.Fatalf("ensure() error = %v", err) + } + if path != "/var/lib/libvirt/images/base-debian-13.qcow2" { + t.Errorf("path = %q", path) + } + if srv.hits.Load() != 0 { + t.Errorf("http hits = %d, want 0 for a cached image", srv.hits.Load()) + } + if api.hasCall("StorageVolCreateXML") || api.hasCall("StorageVolDelete") { + t.Errorf("cached image was touched: calls = %v", api.calls) + } +} + +func TestImageCache_ensure_markerMissingRedownloads(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + api.putVol("default", "base-debian-13.qcow2", "qcow2", []byte("half an image")) + if _, err := ensureTest(t, api, testSource(srv, testImageSHA())); err != nil { + t.Fatalf("ensure() error = %v", err) + } + if srv.hits.Load() != 1 { + t.Errorf("http hits = %d, want 1: a base without its marker is a partial upload", srv.hits.Load()) + } + if base := api.vol("default", "base-debian-13.qcow2"); base == nil || !bytes.Equal(base.data, testImage) { + t.Error("partial base volume was not replaced") + } +} + +func TestImageCache_ensure_checksumChangedRedownloads(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + api.putVol("default", "base-debian-13.qcow2", "qcow2", []byte("an older release")) + api.putVol("default", "base-debian-13.qcow2.sha256", "raw", []byte(strings.Repeat("0", 64))) + if _, err := ensureTest(t, api, testSource(srv, testImageSHA())); err != nil { + t.Fatalf("ensure() error = %v", err) + } + if srv.hits.Load() != 1 { + t.Errorf("http hits = %d, want 1 after the configured checksum changed", srv.hits.Load()) + } + if marker := api.vol("default", "base-debian-13.qcow2.sha256"); marker == nil || string(marker.data) != testImageSHA() { + t.Errorf("marker not updated: %+v", marker) + } +} + +func TestImageCache_ensure_checksumMismatchIsPermanent(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + _, err := ensureTest(t, api, testSource(srv, strings.Repeat("a", 64))) + if !errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensure() error = %v, want ErrPermanent", err) + } + if names := api.volNames("default"); len(names) != 0 { + t.Errorf("volumes left behind after a checksum mismatch: %v", names) + } +} + +func TestImageCache_ensure_httpErrors(t *testing.T) { + t.Parallel() + tests := []struct { + name string + status int + want error + }{ + {"404 is permanent", http.StatusNotFound, provider.ErrPermanent}, + {"403 is permanent", http.StatusForbidden, provider.ErrPermanent}, + {"503 is transient", http.StatusServiceUnavailable, provider.ErrTransient}, + {"500 is transient", http.StatusInternalServerError, provider.ErrTransient}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + srv.status = tc.status + api := newFakeAPI("default") + _, err := ensureTest(t, api, testSource(srv, "")) + if !errors.Is(err, tc.want) { + t.Fatalf("ensure() error = %v, want %v", err, tc.want) + } + if names := api.volNames("default"); len(names) != 0 { + t.Errorf("volumes created despite HTTP %d: %v", tc.status, names) + } + }) + } +} + +func TestImageCache_ensure_noContentLengthIsPermanent(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + srv.noLength = true + api := newFakeAPI("default") + _, err := ensureTest(t, api, testSource(srv, "")) + if !errors.Is(err, provider.ErrPermanent) || !strings.Contains(err.Error(), "Content-Length") { + t.Fatalf("ensure() error = %v, want ErrPermanent about Content-Length", err) + } +} + +func TestImageCache_ensure_uploadFailureCleansUp(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + api.uploadHook = func(_ string, r io.Reader) error { + _, _ = io.CopyN(io.Discard, r, 100) + return io.ErrUnexpectedEOF + } + _, err := ensureTest(t, api, testSource(srv, testImageSHA())) + if !errors.Is(err, provider.ErrTransient) && classify(err) != provider.ErrTransient { + t.Fatalf("ensure() error = %v, want a transient failure", err) + } + if names := api.volNames("default"); len(names) != 0 { + t.Errorf("volumes left behind after a failed upload: %v", names) + } + // Next attempt starts clean and succeeds. + api.uploadHook = nil + if _, err := ensureTest(t, api, testSource(srv, testImageSHA())); err != nil { + t.Fatalf("retry ensure() error = %v", err) + } +} + +func TestImageCache_ensure_missingPoolIsPermanent(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + c := newImageCache(nil) + pool, _ := api.StoragePoolLookupByName("default") + delete(api.pools, "default") + _, err := c.ensure(context.Background(), logr.Discard(), api, "nas1", pool, testSource(srv, "")) + if classify(err) != provider.ErrPermanent { + t.Fatalf("ensure() error = %v, want one classifying as Permanent", err) + } +} + +func TestImageCache_ensure_concurrentCallersDownloadOnce(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + srv.delay = 50 * time.Millisecond + api := newFakeAPI("default") + c := newImageCache(nil) + pool, _ := api.StoragePoolLookupByName("default") + src := testSource(srv, testImageSHA()) + var wg sync.WaitGroup + paths := make([]string, 4) + errs := make([]error, 4) + for i := range 4 { + wg.Go(func() { + paths[i], errs[i] = c.ensure(context.Background(), logr.Discard(), api, "nas1", pool, src) + }) + } + wg.Wait() + for i := range 4 { + if errs[i] != nil || paths[i] != "/var/lib/libvirt/images/base-debian-13.qcow2" { + t.Errorf("caller %d: path=%q err=%v", i, paths[i], errs[i]) + } + } + if srv.hits.Load() != 1 { + t.Errorf("http hits = %d, want 1 for four concurrent callers", srv.hits.Load()) + } +} + +func TestImageCache_ensure_differentHostsFetchIndependently(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + c := newImageCache(nil) + src := testSource(srv, "") + for _, host := range []string{"nas1", "lab2"} { + api := newFakeAPI("default") + pool, _ := api.StoragePoolLookupByName("default") + if _, err := c.ensure(context.Background(), logr.Discard(), api, host, pool, src); err != nil { + t.Fatalf("ensure(%s) error = %v", host, err) + } + } + if srv.hits.Load() != 2 { + t.Errorf("http hits = %d, want one per host", srv.hits.Load()) + } +} + +func TestImageCache_ensure_cancelledContextIsTransient(t *testing.T) { + t.Parallel() + srv := newImageServer(t) + api := newFakeAPI("default") + c := newImageCache(nil) + pool, _ := api.StoragePoolLookupByName("default") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := c.ensure(ctx, logr.Discard(), api, "nas1", pool, testSource(srv, "")) + if classify(err) != provider.ErrTransient { + t.Fatalf("ensure() error = %v, want transient", err) + } + if names := api.volNames("default"); len(names) != 0 { + t.Errorf("volumes created despite cancelled context: %v", names) + } +}