diff --git a/docs/plans-executions/2026-09-16-1333-kvm2-provider.md b/docs/plans-executions/2026-09-16-1333-kvm2-provider.md index e3e142c..f4a7579 100644 --- a/docs/plans-executions/2026-09-16-1333-kvm2-provider.md +++ b/docs/plans-executions/2026-09-16-1333-kvm2-provider.md @@ -7,7 +7,7 @@ Plan: [docs/plans/2026-09-16-1333-kvm2-provider.md](../plans/2026-09-16-1333-kvm - [x] Step 2 — Config wiring, IDs, state mapping, classify - [x] Step 3 — Pure builders: domain and volume XML, metadata, seed - [x] Step 4 — Transport, seam, decorator, fakes -- [ ] Step 5 — Image cache and pool-space precheck +- [x] Step 5 — Image cache and pool-space precheck - [ ] Step 6 — The four contract methods - [ ] Step 7 — Checker and the CRD comment - [ ] Step 8 — Deployment files, Makefile, integration test @@ -171,3 +171,42 @@ grep -n 'context.Background()' internal/provider/kvm2/*.go | grep -v _test.go ``` All `ok`; the agent also ran the timing-sensitive tests at `-count=8`. + +## Step 5 — Image cache and pool-space precheck + +2026-09-16 18:11. Fourth subagent step: `image.go` and `image_test.go` +(27 tests), plus additive changes to `fake_test.go` (a `twin()` sharing +state with its own call log, so a test can tell the pooled connection from +the dedicated one) and `kvm2.go` (`http` client, budgets and tracing +options behind functional options, so `New` is unchanged). The pipeline: +marker lookup and bounded read on the pooled connection, then a GET bound to +ctx plus the 20-minute cap under a `kvm2.download` span, `StorageVolCreateXML` +and a streamed `StorageVolUpload` through sha256 on a fresh connection, +`StoragePoolRefresh`, and the marker written last; every failure after the +volume exists discards it with the cancellation stripped from ctx so the +cleanup RPC still runs. + +Streaming is asserted by counting: a hook on the dedicated fake's upload +snapshots how many bytes the 64 MiB test server had written when the upload +began, and asserts under 16 MiB. The agent chose that over a process-wide +`MemStats` delta because the tests run in parallel. + +Decisions the agent made, accepted: the base volume is created on the +dedicated connection (nothing exists if the dial fails); `Accept-Encoding: +identity` is sent so transparent gunzip cannot hide the Content-Length; a +marker whose content is not 64 hex characters counts as debris; +`Content-Length: 0` is its own Permanent error; the image lock is a +one-slot channel so a waiter gives up with its ctx instead of sitting out a +transfer; the fetch logger is derived after the span starts so its lines +carry the span ID (the agent's first test run caught that). + +Commands (agent and orchestrator): + +```sh +gofmt -l ./internal/provider/kvm2 ; go vet ./internal/provider/kvm2/ && go test -race -count=2 ./internal/provider/kvm2/ +go build ./... && go test -short ./... +grep -n 'io.ReadAll\|bytes.Buffer' internal/provider/kvm2/image.go # empty +grep -n 'context.Background()' internal/provider/kvm2/*.go | grep -v _test.go # empty +``` + +All `ok`. Attestation: "files read outside the allowed set: none". diff --git a/internal/provider/kvm2/fake_test.go b/internal/provider/kvm2/fake_test.go index 3ea5aa3..6550f66 100644 --- a/internal/provider/kvm2/fake_test.go +++ b/internal/provider/kvm2/fake_test.go @@ -2,6 +2,8 @@ package kvm2 import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/xml" "fmt" "io" @@ -18,7 +20,9 @@ import ( // libvirt host for the provider's Create, Get, Delete and ListByTag, with // per-method error hooks and blocking, call counters and an ordered log. type fakeBackend struct { - mu sync.Mutex + // mu, domains and pools are the host's state; a twin shares them, so + // two connections to the same fake host see the same volumes. + mu *sync.Mutex domains map[string]*fakeDomain pools map[string]*fakePool hostname string @@ -28,6 +32,10 @@ type fakeBackend struct { hooks map[string]func() error // block makes a method wait on its channel (or ctx) before running. block map[string]chan struct{} + // discardUploads makes StorageVolUpload stream the reader through a + // sha256 in fixed chunks instead of holding the bytes, for tests that + // push more through the fake than they want in memory. + discardUploads bool calls map[string]int log []string @@ -55,12 +63,16 @@ type fakePool struct { type fakeVol struct { content []byte capacity uint64 + // uploaded and sha256 record a discarded upload (see discardUploads). + uploaded int64 + sha256 string } const fakePoolName = "default" func newFakeBackend() *fakeBackend { return &fakeBackend{ + mu: &sync.Mutex{}, domains: make(map[string]*fakeDomain), pools: map[string]*fakePool{fakePoolName: newFakePool()}, hostname: "fake-host", @@ -72,6 +84,39 @@ func newFakeBackend() *fakeBackend { } } +// twin returns a second connection to the same fake host: it shares the +// domains and pools (and their lock) but has its own hooks, counters, log, +// close count and liveness, so a test can tell which connection did what. +func (f *fakeBackend) twin() *fakeBackend { + return &fakeBackend{ + mu: f.mu, + domains: f.domains, + pools: f.pools, + hostname: f.hostname, + version: f.version, + hooks: make(map[string]func() error), + block: make(map[string]chan struct{}), + discardUploads: f.discardUploads, + calls: make(map[string]int), + deadCh: make(chan struct{}), + } +} + +// vol returns the volume by name in the default pool, or nil. +func (f *fakeBackend) vol(name string) *fakeVol { + f.mu.Lock() + defer f.mu.Unlock() + return f.pools[fakePoolName].vols[name] +} + +// closeCount is closed read under the lock, for twins closed by another +// goroutine. +func (f *fakeBackend) closeCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.closed +} + func newFakePool() *fakePool { return &fakePool{state: libvirt.StoragePoolRunning, capacity: 100 << 30, available: 50 << 30, vols: make(map[string]*fakeVol)} } @@ -422,10 +467,20 @@ func (f *fakeBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageV if err := f.enter(ctx, "StorageVolUpload"); err != nil { return err } - var content []byte + var ( + content []byte + uploaded int64 + digest string + ) if r != nil { var err error - if content, err = io.ReadAll(r); err != nil { + if f.discardUploads { + h := sha256.New() + if uploaded, err = io.CopyBuffer(h, r, make([]byte, 64<<10)); err != nil { + return err + } + digest = hex.EncodeToString(h.Sum(nil)) + } else if content, err = io.ReadAll(r); err != nil { return err } } @@ -436,6 +491,8 @@ func (f *fakeBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageV return err } v.content = content + v.uploaded = uploaded + v.sha256 = digest return nil } diff --git a/internal/provider/kvm2/image.go b/internal/provider/kvm2/image.go new file mode 100644 index 0000000..2e1d6be --- /dev/null +++ b/internal/provider/kvm2/image.go @@ -0,0 +1,474 @@ +package kvm2 + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "regexp" + "strings" + "sync" + "time" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +const ( + // markerCapacityBytes is the marker volume's size: one lowercase hex + // sha256 digest, nothing else. + markerCapacityBytes = sha256.Size * 2 + // markerReadCap bounds what is read back from a marker, so a wrong file + // under that name cannot grow memory. + markerReadCap = 1 << 10 + // seedReserveBytes is what the quota precheck sets aside for the seed + // ISO next to the image and the overlay (spec entry 13). + seedReserveBytes = 16 << 20 +) + +var hexDigest = regexp.MustCompile(`^[0-9a-f]{64}$`) + +// imageSource is a resolved placement.image (spec entries 3 and 4): the +// catalogue entry, or a URL given directly with the catalogue defaults, plus +// the key its base volume is named after. +type imageSource struct { + Key string + provider.KVMImage +} + +func (s imageSource) volumeName() string { return baseVolumeName(s.Key, s.Format) } +func (s imageSource) markerName() string { return markerVolumeName(s.volumeName()) } + +// resolveImage maps placement.image to a source. cfg is already defaulted, +// so a catalogue entry carries its format and firmware; a URL gets qcow2 +// and uefi. +func resolveImage(cfg provider.KVMConfig, image string) (imageSource, error) { + if img, ok := cfg.Images[image]; ok { + return imageSource{Key: image, KVMImage: img}, nil + } + if provider.IsHTTPURL(image) { + return imageSource{ + Key: urlImageKey(image), + KVMImage: provider.KVMImage{URL: image, Format: provider.KVMFormatQcow2, Firmware: provider.KVMFirmwareUEFI}, + }, nil + } + return imageSource{}, permanentf("placement.image %q is neither a key in kvm2.images nor an http(s) URL", image) +} + +// lockImage takes the per-(host, volume) lock of spec entry 19, or gives up +// with ctx. The lock is a one-slot channel rather than a mutex so a caller +// whose ctx ends while another caller's download runs returns promptly +// (contract §1.5) instead of waiting out a 20-minute transfer. +func (p *Provider) lockImage(ctx context.Context, hostName, volume string) (func(), error) { + key := hostName + "/" + volume + p.imageMu.Lock() + sem, ok := p.imageLocks[key] + if !ok { + sem = make(chan struct{}, 1) + p.imageLocks[key] = sem + } + p.imageMu.Unlock() + select { + case sem <- struct{}{}: + return func() { <-sem }, nil + case <-ctx.Done(): + return nil, fmt.Errorf("%w: waiting for image %s on host %s: %w", provider.ErrTransient, volume, hostName, ctx.Err()) + } +} + +// ensureImage makes the base image present and complete in the host's pool +// and returns its path (spec entry 4). be is the caller's pooled connection +// and carries the cheap checks; the upload itself goes over a connection of +// its own. Concurrent callers for the same (host, volume) queue on the +// lock: the first fetches, the rest find the marker. +func (p *Provider) ensureImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, be backend, src imageSource) (string, error) { + unlock, err := p.lockImage(ctx, host.Name, src.volumeName()) + if err != nil { + return "", err + } + defer unlock() + log := p.logger(ctx).WithValues("host", host.Name, "volume", src.volumeName()) + path, err := cachedImage(ctx, log, pool, be, src) + if err != nil || path != "" { + return path, err + } + vol, err := p.fetchImage(ctx, host, pool, be, src) + if err != nil { + return "", err + } + path, err = be.StorageVolGetPath(ctx, vol) + if err != nil { + return "", fmt.Errorf("reading the path of base volume %s: %w", vol.Name, err) + } + return path, nil +} + +// cachedImage returns the base volume's path when the marker says the base +// is complete and its digest is the configured one (or none is configured). +// Anything else is cleared so the fetch starts from nothing, and "" is +// returned: a base without a marker is an upload that was cut off; a marker +// without a base, or one whose content is not a digest, is debris; a marker +// with a different digest means the catalogue entry changed. +func cachedImage(ctx context.Context, log logr.Logger, pool libvirt.StoragePool, be backend, src imageSource) (string, error) { + name, markerName := src.volumeName(), src.markerName() + marker, err := be.StorageVolLookupByName(ctx, pool, markerName) + switch { + case hasCode(err, libvirt.ErrNoStorageVol): + return "", deleteVolumeIfPresent(ctx, be, pool, name) + case err != nil: + return "", fmt.Errorf("looking up marker %s: %w", markerName, err) + } + digest, err := readMarker(ctx, be, marker) + if err != nil { + return "", err + } + switch { + case !hexDigest.MatchString(digest): + log.V(1).Info("base image marker holds no digest; refetching") + case src.SHA256 != "" && !strings.EqualFold(digest, src.SHA256): + log.V(1).Info("base image digest differs from the configured sha256; refetching", "cached", digest, "configured", strings.ToLower(src.SHA256)) + default: + base, err := be.StorageVolLookupByName(ctx, pool, name) + switch { + case hasCode(err, libvirt.ErrNoStorageVol): + return "", deleteVolumeIfPresent(ctx, be, pool, markerName) + case err != nil: + return "", fmt.Errorf("looking up base volume %s: %w", name, err) + } + path, err := be.StorageVolGetPath(ctx, base) + if err != nil { + return "", fmt.Errorf("reading the path of base volume %s: %w", name, err) + } + return path, nil + } + return "", errors.Join(deleteVolumeIfPresent(ctx, be, pool, markerName), deleteVolumeIfPresent(ctx, be, pool, name)) +} + +// readMarker returns the digest a marker volume holds, lowercased and +// stripped of padding. The volume is 64 bytes; the read is capped anyway. +func readMarker(ctx context.Context, be backend, marker libvirt.StorageVol) (string, error) { + w := &cappedWriter{limit: markerReadCap} + if err := be.StorageVolDownload(ctx, marker, w, 0, 0, 0); err != nil { + return "", fmt.Errorf("reading marker %s: %w", marker.Name, err) + } + return strings.ToLower(strings.TrimRight(string(w.buf), "\x00 \r\n\t")), nil +} + +// cappedWriter keeps the first limit bytes and accepts the rest without +// storing them, so a download into it is bounded whatever its length. +type cappedWriter struct { + limit int + buf []byte +} + +func (w *cappedWriter) Write(b []byte) (int, error) { + if room := w.limit - len(w.buf); room > 0 { + w.buf = append(w.buf, b[:min(room, len(b))]...) + } + return len(b), nil +} + +// deleteVolumeIfPresent removes the named volume; one that is already gone +// is success. +func deleteVolumeIfPresent(ctx context.Context, be backend, pool libvirt.StoragePool, name string) error { + vol, err := be.StorageVolLookupByName(ctx, pool, name) + if hasCode(err, libvirt.ErrNoStorageVol) { + return nil + } + if err != nil { + return fmt.Errorf("looking up volume %s: %w", name, err) + } + if err := be.StorageVolDelete(ctx, vol, 0); err != nil && !hasCode(err, libvirt.ErrNoStorageVol) { + return fmt.Errorf("deleting volume %s: %w", name, err) + } + return nil +} + +// fetchImage streams the image from its URL into a new base volume and then +// writes the marker, under the kvm2.download span (spec entry 17). The +// upload goes over a connection of its own (spec entry 18): abandoning a +// call kills the connection it ran on, and a 20-minute upload must not die +// because a poll for another Proxy was cancelled on the pooled one. The +// marker is written last, on the pooled connection, after +// StoragePoolRefresh made libvirt read the file's real format; its presence +// is what marks the base complete. +// +// Every exit between the base volume's creation and the marker deletes the +// base volume (best effort): the next attempt would delete it anyway on +// finding no marker, and removing it now frees the space at once. +func (p *Provider) fetchImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, be backend, src imageSource) (libvirt.StorageVol, error) { + name := src.volumeName() + ctx, cancel := context.WithTimeout(ctx, p.budgets.transfer) + defer cancel() + ctx, span := tracing.StartSpan(ctx, tracing.Tracer(p.tracing...), "kvm2.download", + trace.WithSpanKind(trace.SpanKindClient), + trace.WithAttributes( + attribute.String("kvm2.host", host.Name), + attribute.String("kvm2.volume", name), + attribute.String("kvm2.url_host", urlHost(src.URL)), + attribute.Bool("kvm2.sha256_configured", src.SHA256 != ""), + )) + defer span.End() + // Taken after the span starts, so the lines carry its ID. + log := p.logger(ctx).WithValues("host", host.Name, "volume", name) + start := time.Now() + + resp, err := p.do(ctx, http.MethodGet, src.URL) + if err == nil { + err = checkResponse(src.URL, resp) + } + if err != nil { + span.SetStatus(codes.Error, err.Error()) + return libvirt.StorageVol{}, err + } + defer func() { _ = resp.Body.Close() }() + size := resp.ContentLength + span.SetAttributes(attribute.Int64("kvm2.bytes", size)) + log.V(1).Info("fetching base image into host pool", "url", src.URL, "bytes", size) + + vol, digest, err := p.uploadImage(ctx, host, pool, be, src, resp.Body, size) + if err != nil { + span.SetStatus(codes.Error, err.Error()) + return libvirt.StorageVol{}, err + } + if err := be.StoragePoolRefresh(ctx, pool, 0); err != nil { + err = fmt.Errorf("refreshing storage pool %s after the upload: %w", pool.Name, err) + p.discardVolume(ctx, log, be, pool, name) + span.SetStatus(codes.Error, err.Error()) + return libvirt.StorageVol{}, err + } + if err := writeMarker(ctx, be, pool, src.markerName(), digest); err != nil { + p.discardVolume(ctx, log, be, pool, name) + span.SetStatus(codes.Error, err.Error()) + return libvirt.StorageVol{}, err + } + log.V(1).Info("base image cached", "bytes", size, "duration", time.Since(start), "sha256", digest) + return vol, nil +} + +// uploadImage creates the base volume and streams body into it over a +// dedicated connection with a running sha256, then checks length and +// digest. On any failure the volume is discarded and the error says why: +// a body that ended early is Transient, a digest that differs from the +// configured one is Permanent. +func (p *Provider) uploadImage(ctx context.Context, host provider.KVMHost, pool libvirt.StoragePool, be backend, src imageSource, body io.Reader, size int64) (libvirt.StorageVol, string, error) { + name := src.volumeName() + xmlDoc, err := buildVolumeXML(volumeSpec{Name: name, CapacityBytes: size, Format: src.Format}) + if err != nil { + return libvirt.StorageVol{}, "", err + } + fresh, err := p.hosts.dialFresh(ctx, host) + if err != nil { + return libvirt.StorageVol{}, "", fmt.Errorf("dialling host %s for the image upload: %w", host.Name, err) + } + defer func() { _ = fresh.Close() }() + vol, err := fresh.StorageVolCreateXML(ctx, pool, xmlDoc, 0) + if err != nil { + return libvirt.StorageVol{}, "", fmt.Errorf("creating base volume %s: %w", name, err) + } + + hash := sha256.New() + counted := &countingReader{r: io.TeeReader(body, hash)} + uploadErr := fresh.StorageVolUpload(ctx, vol, counted, 0, uint64(size), 0) + read, readErr := counted.state() + var digest string + if uploadErr == nil { + // Only then has the upload stopped reading; an abandoned call's + // goroutine may still be writing into the hash. + digest = hex.EncodeToString(hash.Sum(nil)) + } + switch { + case readErr != nil: + err = fmt.Errorf("%w: image %s: the response ended after %d of %d bytes: %w", provider.ErrTransient, src.URL, read, size, readErr) + case uploadErr != nil: + err = fmt.Errorf("uploading base volume %s: %w", name, uploadErr) + case read != size: + err = transientf("image %s: short read, %d of %d bytes", src.URL, read, size) + case src.SHA256 != "" && !strings.EqualFold(digest, src.SHA256): + err = permanentf("image %s: sha256 %s does not match the configured %s", src.URL, digest, strings.ToLower(src.SHA256)) + } + if err != nil { + p.discardVolume(ctx, p.logger(ctx).WithValues("host", host.Name, "volume", name), be, pool, name) + return libvirt.StorageVol{}, "", err + } + return vol, digest, nil +} + +// countingReader counts what the upload consumed and keeps the first +// non-EOF error the source returned. The adapter can return from an +// abandoned upload while go-libvirt still reads, so the fields are guarded. +type countingReader struct { + r io.Reader + mu sync.Mutex + n int64 + err error +} + +func (c *countingReader) Read(b []byte) (int, error) { + n, err := c.r.Read(b) + c.mu.Lock() + c.n += int64(n) + if err != nil && err != io.EOF && c.err == nil { + c.err = err + } + c.mu.Unlock() + return n, err +} + +func (c *countingReader) state() (int64, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.n, c.err +} + +// discardVolume removes a base volume a failed fetch left behind. It runs +// with ctx's values but not its cancellation: the fetch may have failed +// because ctx ended, and the adapter refuses calls on a finished ctx. The +// budget bounds it on its own. Failure is logged, not returned: the next +// attempt finds the volume without a marker and deletes it then. +func (p *Provider) discardVolume(ctx context.Context, log logr.Logger, be backend, pool libvirt.StoragePool, name string) { + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), p.budgets.rpc) + defer cancel() + if err := deleteVolumeIfPresent(ctx, be, pool, name); err != nil { + log.V(1).Info("incomplete base volume left behind; the next attempt removes it", "error", err.Error()) + } +} + +// writeMarker creates the raw marker volume and uploads the digest into it. +// A marker that could not be filled is removed again: an empty marker next +// to a complete base would read as debris and cost a refetch. +func writeMarker(ctx context.Context, be backend, pool libvirt.StoragePool, name, digest string) error { + xmlDoc, err := buildVolumeXML(volumeSpec{Name: name, CapacityBytes: markerCapacityBytes, Format: provider.KVMFormatRaw}) + if err != nil { + return err + } + vol, err := be.StorageVolCreateXML(ctx, pool, xmlDoc, 0) + if err != nil { + return fmt.Errorf("creating marker %s: %w", name, err) + } + if err := be.StorageVolUpload(ctx, vol, strings.NewReader(digest), 0, uint64(len(digest)), 0); err != nil { + _ = deleteVolumeIfPresent(ctx, be, pool, name) + return fmt.Errorf("writing marker %s: %w", name, err) + } + return nil +} + +// do issues one request bound to ctx. Accept-Encoding is fixed to identity +// because a transparently decompressed body reports no Content-Length. A +// transport error is Transient. +func (p *Provider) do(ctx context.Context, method, rawURL string) (*http.Response, error) { + req, err := http.NewRequestWithContext(ctx, method, rawURL, nil) + if err != nil { + return nil, permanentf("image %s: %v", rawURL, err) + } + req.Header.Set("Accept-Encoding", "identity") + resp, err := p.http.Do(req) + if err != nil { + return nil, fmt.Errorf("%w: fetching image %s: %w", provider.ErrTransient, rawURL, err) + } + return resp, nil +} + +// checkResponse classifies the response at the source (spec entry 13) and +// closes its body when it is unusable. 404, 403 and 410, and every other +// 4xx, mean the catalogue entry is wrong; 5xx and 429 are worth a retry. A +// missing Content-Length is Permanent: the volume's capacity must be +// declared before the upload, and a mirror that only chunks is one the +// catalogue must not point at. +func checkResponse(rawURL string, resp *http.Response) error { + var err error + code := resp.StatusCode + switch { + case code >= 200 && code < 300: + case code == http.StatusTooManyRequests || code >= 500: + err = transientf("image %s: HTTP %s", rawURL, resp.Status) + case code >= 400: + err = permanentf("image %s: HTTP %s", rawURL, resp.Status) + default: + err = transientf("image %s: unexpected HTTP %s", rawURL, resp.Status) + } + if err == nil && resp.ContentLength < 0 { + err = permanentf("image %s: response has no Content-Length; the volume capacity must be known before upload", rawURL) + } + if err == nil && resp.ContentLength == 0 { + err = permanentf("image %s: response is empty (Content-Length 0)", rawURL) + } + if err != nil { + _ = resp.Body.Close() + } + return err +} + +// urlHost is the only part of the URL a span carries (contract §9). +func urlHost(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil { + return "" + } + return u.Host +} + +// imageSizeIfUncached returns the image's Content-Length when the base is +// not cached on the host, and 0 when the marker is present, for the quota +// precheck of spec entry 13. It never downloads: HEAD first, and a GET whose +// body is closed at once when the server does not allow HEAD. A marker is +// taken at face value here; ensureImage decides whether it still matches. +func (p *Provider) imageSizeIfUncached(ctx context.Context, be backend, pool libvirt.StoragePool, src imageSource) (int64, error) { + _, err := be.StorageVolLookupByName(ctx, pool, src.markerName()) + switch { + case err == nil: + return 0, nil + case !hasCode(err, libvirt.ErrNoStorageVol): + return 0, fmt.Errorf("looking up marker %s: %w", src.markerName(), err) + } + ctx, cancel := context.WithTimeout(ctx, p.budgets.rpc) + defer cancel() + resp, err := p.do(ctx, http.MethodHead, src.URL) + if err != nil { + return 0, err + } + _ = resp.Body.Close() + if resp.StatusCode == http.StatusMethodNotAllowed { + if resp, err = p.do(ctx, http.MethodGet, src.URL); err != nil { + return 0, err + } + _ = resp.Body.Close() + } + if err := checkResponse(src.URL, resp); err != nil { + return 0, err + } + return resp.ContentLength, nil +} + +// checkPoolSpace is the precheck of spec entry 13: the pool must be running +// and have needed bytes available, or Create returns ErrQuotaExceeded and +// the reconciler waits QuotaRetry instead of failing the volume creation. +func checkPoolSpace(ctx context.Context, be backend, pool libvirt.StoragePool, poolName string, needed uint64) error { + state, _, _, available, err := be.StoragePoolGetInfo(ctx, pool) + if err != nil { + return fmt.Errorf("reading storage pool %s: %w", poolName, err) + } + if s := libvirt.StoragePoolState(state); s != libvirt.StoragePoolRunning { + return permanentf("storage pool %s is not running (state %s)", poolName, s) + } + if available < needed { + return fmt.Errorf("%w: storage pool %s has %d bytes available, %d needed", provider.ErrQuotaExceeded, poolName, available, needed) + } + return nil +} + +// neededBytes is what one Create commits to the pool: the image when it is +// not cached yet (0 otherwise), the overlay's capacity and the seed reserve. +func neededBytes(imageBytes int64, diskSizeGB int64) uint64 { + return uint64(max(imageBytes, 0)) + uint64(max(diskSizeGB, 0))<<30 + seedReserveBytes +} diff --git a/internal/provider/kvm2/image_test.go b/internal/provider/kvm2/image_test.go new file mode 100644 index 0000000..96e0b05 --- /dev/null +++ b/internal/provider/kvm2/image_test.go @@ -0,0 +1,981 @@ +package kvm2 + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/digitalocean/go-libvirt" + "github.com/go-logr/logr/funcr" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" +) + +const imageKey = "debian-13" + +var testPool = libvirt.StoragePool{Name: fakePoolName, UUID: fakeUUID(fakePoolName)} + +// imageFixture is one host with a pooled fake, a dialer that hands out +// twins of it for dialFresh, an image server, a span recorder and a log +// capture. Tests that need a single ordered call log point the dialer at +// the pooled fake instead (sameConnection). +type imageFixture struct { + t *testing.T + p *Provider + pooled *fakeBackend + dialer *stubDialer + srv *httptest.Server + hits atomic.Int32 + sr *tracetest.SpanRecorder + logs *logCapture + ctx context.Context + host provider.KVMHost + + mu sync.Mutex + fresh []*fakeBackend + // onFresh configures each twin before it is handed out. + onFresh func(*fakeBackend) +} + +func newImageFixture(t *testing.T, handler http.HandlerFunc) *imageFixture { + t.Helper() + fx := &imageFixture{t: t, pooled: newFakeBackend(), logs: &logCapture{}} + fx.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fx.hits.Add(1) + handler(w, r) + })) + t.Cleanup(fx.srv.Close) + fx.dialer = &stubDialer{fresh: func() backend { + tw := fx.pooled.twin() + fx.mu.Lock() + if fx.onFresh != nil { + fx.onFresh(tw) + } + fx.fresh = append(fx.fresh, tw) + fx.mu.Unlock() + return tw + }} + fx.sr = tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(fx.sr)) + t.Cleanup(func() { _ = tp.Shutdown(context.Background()) }) + cfg := provider.KVMConfig{ + Images: map[string]provider.KVMImage{imageKey: {URL: fx.srv.URL + "/images/debian-13-generic-amd64.qcow2"}}, + MachineTypes: map[string]provider.KVMMachineType{"small": {VCPUs: 1, MemoryMiB: 1024}}, + Hosts: []provider.KVMHost{{Name: "beelink", Address: "beelink:22", User: "virt", Bridge: "br0"}}, + } + fx.p = newProvider("kvm2-home", cfg, fx.dialer, withHTTPClient(fx.srv.Client()), withTracing(tracing.WithTracerProvider(tp))) + fx.host = fx.p.cfg.Hosts[0] + log := funcr.New(func(prefix, args string) { + fx.logs.mu.Lock() + defer fx.logs.mu.Unlock() + fx.logs.lines = append(fx.logs.lines, prefix+" "+args) + }, funcr.Options{Verbosity: 1}) + fx.ctx = logf.IntoContext(context.Background(), log) + return fx +} + +// sameConnection makes dialFresh return the pooled fake, so one call log +// carries every call in order. +func (fx *imageFixture) sameConnection() { + fx.dialer.fresh = nil + fx.dialer.backend = fx.pooled +} + +func (fx *imageFixture) src() imageSource { + fx.t.Helper() + src, err := resolveImage(fx.p.cfg, imageKey) + if err != nil { + fx.t.Fatalf("resolveImage(%q) error = %v", imageKey, err) + } + return src +} + +func (fx *imageFixture) ensure(ctx context.Context, src imageSource) (string, error) { + return fx.p.ensureImage(ctx, fx.host, testPool, fx.pooled, src) +} + +func (fx *imageFixture) freshBackends() []*fakeBackend { + fx.mu.Lock() + defer fx.mu.Unlock() + return append([]*fakeBackend(nil), fx.fresh...) +} + +// putVol places a volume in the pool directly, as a previous run would have. +func (fx *imageFixture) putVol(name string, content []byte) { + fx.pooled.mu.Lock() + defer fx.pooled.mu.Unlock() + fx.pooled.pools[fakePoolName].vols[name] = &fakeVol{content: content, capacity: uint64(len(content))} +} + +func randomBody(t *testing.T, n int) []byte { + t.Helper() + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + t.Fatal(err) + } + return b +} + +func digestOf(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +func serveBytes(body []byte) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(body))) + w.WriteHeader(http.StatusOK) + if r.Method != http.MethodHead { + _, _ = w.Write(body) + } + } +} + +func serveStatus(code int) http.HandlerFunc { + return func(w http.ResponseWriter, _ *http.Request) { http.Error(w, http.StatusText(code), code) } +} + +func TestResolveImage_table(t *testing.T) { + t.Parallel() + cfg := withDefaults(provider.KVMConfig{Images: map[string]provider.KVMImage{ + "debian-13": {URL: "https://cloud.debian.org/debian-13.qcow2", SHA256: "ABCD"}, + "alpine": {URL: "https://alpine.example/alpine.img", Format: provider.KVMFormatRaw, Firmware: provider.KVMFirmwareBIOS}, + }}) + const rawURL = "http://mirror.example/x/y.qcow2" + tests := []struct { + name string + image string + want imageSource + wantVolume string + wantErr string + }{ + { + name: "catalogue key with defaults applied", + image: "debian-13", + want: imageSource{Key: "debian-13", KVMImage: provider.KVMImage{URL: "https://cloud.debian.org/debian-13.qcow2", SHA256: "ABCD", Format: "qcow2", Firmware: "uefi"}}, + wantVolume: "base-debian-13.qcow2", + }, + { + name: "catalogue key with explicit format and firmware", + image: "alpine", + want: imageSource{Key: "alpine", KVMImage: provider.KVMImage{URL: "https://alpine.example/alpine.img", Format: "raw", Firmware: "bios"}}, + wantVolume: "base-alpine.raw", + }, + { + name: "http URL used directly with defaults", + image: rawURL, + want: imageSource{Key: urlImageKey(rawURL), KVMImage: provider.KVMImage{URL: rawURL, Format: "qcow2", Firmware: "uefi"}}, + wantVolume: "base-" + urlImageKey(rawURL) + ".qcow2", + }, + { + name: "unknown key is permanent and names the value", + image: "ubuntu-24", + wantErr: `placement.image "ubuntu-24" is neither a key in kvm2.images nor an http(s) URL`, + }, + { + name: "non-http scheme is permanent", + image: "file:///var/lib/images/x.qcow2", + wantErr: `placement.image "file:///var/lib/images/x.qcow2" is neither`, + }, + { + name: "empty is permanent", + image: "", + wantErr: `placement.image "" is neither`, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got, err := resolveImage(cfg, tc.image) + if tc.wantErr != "" { + if err == nil || !errors.Is(err, provider.ErrPermanent) || !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("resolveImage(%q) = %+v, %v; want ErrPermanent containing %q", tc.image, got, err, tc.wantErr) + } + 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) + } + if got.volumeName() != tc.wantVolume || got.markerName() != tc.wantVolume+".sha256" { + t.Errorf("volume names = %q, %q; want %q and its .sha256", got.volumeName(), got.markerName(), tc.wantVolume) + } + }) + } +} + +func TestEnsureImage_downloadsAndWritesMarkerLast(t *testing.T) { + t.Parallel() + body := randomBody(t, 1<<20) + fx := newImageFixture(t, serveBytes(body)) + fx.sameConnection() + src := fx.src() + + path, err := fx.ensure(fx.ctx, src) + if err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if want := fakeVolPath(fakePoolName, src.volumeName()); path != want { + t.Errorf("path = %q, want %q", path, want) + } + base := fx.pooled.vol(src.volumeName()) + if base == nil || !bytes.Equal(base.content, body) { + t.Fatalf("base volume missing or content differs (%d bytes)", len(body)) + } + if base.capacity != uint64(len(body)) { + t.Errorf("base capacity = %d, want the Content-Length %d", base.capacity, len(body)) + } + marker := fx.pooled.vol(src.markerName()) + if marker == nil || string(marker.content) != digestOf(body) { + t.Fatalf("marker = %+v, want the body's digest %s", marker, digestOf(body)) + } + if marker.capacity != markerCapacityBytes { + t.Errorf("marker capacity = %d, want %d", marker.capacity, markerCapacityBytes) + } + + log := fx.pooled.callLog() + wantTail := []string{"StorageVolCreateXML", "StorageVolUpload", "StoragePoolRefresh", "StorageVolCreateXML", "StorageVolUpload", "StorageVolGetPath"} + if len(log) < len(wantTail) || strings.Join(log[len(log)-len(wantTail):], ",") != strings.Join(wantTail, ",") { + t.Errorf("call log = %v, want it to end with %v (base create+upload, refresh, marker create+upload, path)", log, wantTail) + } + if fx.hits.Load() != 1 { + t.Errorf("server hits = %d, want 1", fx.hits.Load()) + } + + for _, want := range []string{`"msg"="fetching base image into host pool"`, `"msg"="base image cached"`} { + lines := fx.logs.matching(want) + if len(lines) != 1 { + t.Errorf("got %d lines with %s, want 1:\n%s", len(lines), want, strings.Join(fx.logs.all(), "\n")) + continue + } + for _, kv := range []string{`"provider"="kvm2-home"`, `"host"="beelink"`, `"volume"="base-debian-13.qcow2"`, `"bytes"=` + strconv.Itoa(len(body)), `"traceID"=`} { + if !strings.Contains(lines[0], kv) { + t.Errorf("line lacks %s: %s", kv, lines[0]) + } + } + } + if cached := fx.logs.matching(`"msg"="base image cached"`); len(cached) == 1 && !strings.Contains(cached[0], `"sha256"="`+digestOf(body)+`"`) { + t.Errorf("cached line lacks the digest: %s", cached[0]) + } + + spans := fx.sr.Ended() + if len(spans) != 1 || spans[0].Name() != "kvm2.download" { + t.Fatalf("spans = %v, want exactly one kvm2.download", spanNames(spans)) + } + attrs := spans[0].Attributes() + for _, want := range []attribute.KeyValue{ + attribute.String("kvm2.host", "beelink"), + attribute.String("kvm2.volume", "base-debian-13.qcow2"), + attribute.String("kvm2.url_host", strings.TrimPrefix(fx.srv.URL, "http://")), + attribute.Int64("kvm2.bytes", int64(len(body))), + attribute.Bool("kvm2.sha256_configured", false), + } { + if !hasAttr(attrs, want) { + t.Errorf("span attributes %v lack %v", attrs, want) + } + } + for _, a := range attrs { + if strings.Contains(a.Value.Emit(), src.URL) || strings.Contains(a.Value.Emit(), "/images/") { + t.Errorf("span attribute %s carries the full URL: %s", a.Key, a.Value.Emit()) + } + } +} + +func spanNames(spans []sdktrace.ReadOnlySpan) []string { + out := make([]string, len(spans)) + for i, s := range spans { + out[i] = s.Name() + } + return out +} + +func TestEnsureImage_cachedWithMarkerSkipsDownload(t *testing.T) { + t.Parallel() + body := randomBody(t, 4096) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + fx.putVol(src.volumeName(), body) + fx.putVol(src.markerName(), []byte(digestOf(body))) + + path, err := fx.ensure(fx.ctx, src) + if err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if want := fakeVolPath(fakePoolName, src.volumeName()); path != want { + t.Errorf("path = %q, want %q", path, want) + } + if fx.hits.Load() != 0 { + t.Errorf("server hits = %d, want 0", fx.hits.Load()) + } + if fx.dialer.dials.Load() != 0 { + t.Errorf("dials = %d, want 0: a cached image needs no dedicated connection", fx.dialer.dials.Load()) + } + if n := fx.pooled.callCount("StorageVolDelete"); n != 0 { + t.Errorf("StorageVolDelete called %d times, want 0", n) + } + if len(fx.sr.Ended()) != 0 { + t.Errorf("spans = %v, want none without a download", spanNames(fx.sr.Ended())) + } +} + +func TestEnsureImage_cachedMarkerMatchesConfiguredSHA256CaseInsensitively(t *testing.T) { + t.Parallel() + body := randomBody(t, 4096) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + src.SHA256 = strings.ToUpper(digestOf(body)) + fx.putVol(src.volumeName(), body) + fx.putVol(src.markerName(), []byte(digestOf(body))) + + if _, err := fx.ensure(fx.ctx, src); err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if fx.hits.Load() != 0 { + t.Errorf("server hits = %d, want 0", fx.hits.Load()) + } +} + +func TestEnsureImage_missingMarkerRedownloads(t *testing.T) { + t.Parallel() + body := randomBody(t, 8192) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + fx.putVol(src.volumeName(), []byte("cut off half way")) + + if _, err := fx.ensure(fx.ctx, src); err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if fx.hits.Load() != 1 { + t.Errorf("server hits = %d, want 1", fx.hits.Load()) + } + if n := fx.pooled.callCount("StorageVolDelete"); n != 1 { + t.Errorf("StorageVolDelete called %d times on the pooled connection, want 1 (the leftover base)", n) + } + base := fx.pooled.vol(src.volumeName()) + if base == nil || !bytes.Equal(base.content, body) { + t.Errorf("base volume was not replaced by the download") + } + marker := fx.pooled.vol(src.markerName()) + if marker == nil || string(marker.content) != digestOf(body) { + t.Errorf("marker = %+v, want %s", marker, digestOf(body)) + } +} + +func TestEnsureImage_sha256ChangeRedownloads(t *testing.T) { + t.Parallel() + body := randomBody(t, 8192) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + src.SHA256 = digestOf(body) + digestA := strings.Repeat("a", 64) + fx.putVol(src.volumeName(), []byte("the old image")) + fx.putVol(src.markerName(), []byte(digestA)) + + if _, err := fx.ensure(fx.ctx, src); err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if fx.hits.Load() != 1 { + t.Errorf("server hits = %d, want 1", fx.hits.Load()) + } + if n := fx.pooled.callCount("StorageVolDelete"); n != 2 { + t.Errorf("StorageVolDelete called %d times, want 2 (marker and base)", n) + } + base := fx.pooled.vol(src.volumeName()) + if base == nil || !bytes.Equal(base.content, body) { + t.Errorf("base volume was not replaced by the download") + } + marker := fx.pooled.vol(src.markerName()) + if marker == nil || string(marker.content) != digestOf(body) { + t.Errorf("marker = %+v, want the new digest %s", marker, digestOf(body)) + } + if lines := fx.logs.matching("differs from the configured sha256"); len(lines) != 1 { + t.Errorf("got %d refetch lines, want 1", len(lines)) + } +} + +func TestEnsureImage_markerWithoutDigestRedownloads(t *testing.T) { + t.Parallel() + body := randomBody(t, 4096) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + fx.putVol(src.volumeName(), []byte("whatever")) + fx.putVol(src.markerName(), []byte("not a digest")) + + if _, err := fx.ensure(fx.ctx, src); err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if fx.hits.Load() != 1 { + t.Errorf("server hits = %d, want 1", fx.hits.Load()) + } + if marker := fx.pooled.vol(src.markerName()); marker == nil || string(marker.content) != digestOf(body) { + t.Errorf("marker = %+v, want %s", marker, digestOf(body)) + } +} + +func TestEnsureImage_sha256MismatchIsPermanentAndRemovesVolume(t *testing.T) { + t.Parallel() + body := randomBody(t, 8192) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + src.SHA256 = strings.Repeat("f", 64) + + _, err := fx.ensure(fx.ctx, src) + if !errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensureImage() error = %v, want ErrPermanent", err) + } + if !strings.Contains(err.Error(), digestOf(body)) || !strings.Contains(err.Error(), src.SHA256) { + t.Errorf("error %q should name both digests", err) + } + if fx.pooled.vol(src.volumeName()) != nil { + t.Error("base volume left behind after a digest mismatch") + } + if fx.pooled.vol(src.markerName()) != nil { + t.Error("marker written although the digest did not match") + } + if n := fx.pooled.callCount("StoragePoolRefresh"); n != 0 { + t.Errorf("StoragePoolRefresh called %d times, want 0", n) + } +} + +func TestEnsureImage_noContentLengthIsPermanent(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + _, _ = w.Write([]byte("chunked body")) + }) + src := fx.src() + + _, err := fx.ensure(fx.ctx, src) + if !errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensureImage() error = %v, want ErrPermanent", err) + } + if !strings.Contains(err.Error(), "no Content-Length") || !strings.Contains(err.Error(), src.URL) { + t.Errorf("error %q should say Content-Length is missing and name the URL", err) + } + if fx.dialer.dials.Load() != 0 || fx.pooled.callCount("StorageVolCreateXML") != 0 { + t.Error("a rejected response must not dial or create a volume") + } +} + +func TestEnsureImage_http404IsPermanent(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, serveStatus(http.StatusNotFound)) + _, err := fx.ensure(fx.ctx, fx.src()) + if !errors.Is(err, provider.ErrPermanent) || !strings.Contains(err.Error(), "404") { + t.Fatalf("ensureImage() error = %v, want ErrPermanent naming 404", err) + } + if fx.pooled.vol(fx.src().volumeName()) != nil { + t.Error("volume created for a 404") + } +} + +func TestEnsureImage_http403And410ArePermanent(t *testing.T) { + t.Parallel() + for _, code := range []int{http.StatusForbidden, http.StatusGone, http.StatusBadRequest} { + t.Run(strconv.Itoa(code), func(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, serveStatus(code)) + if _, err := fx.ensure(fx.ctx, fx.src()); !errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensureImage() error = %v, want ErrPermanent", err) + } + }) + } +} + +func TestEnsureImage_http503IsTransient(t *testing.T) { + t.Parallel() + for _, code := range []int{http.StatusServiceUnavailable, http.StatusTooManyRequests, http.StatusBadGateway} { + t.Run(strconv.Itoa(code), func(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, serveStatus(code)) + _, err := fx.ensure(fx.ctx, fx.src()) + if !errors.Is(err, provider.ErrTransient) || errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensureImage() error = %v, want ErrTransient", err) + } + }) + } +} + +func TestEnsureImage_connectionRefusedIsTransient(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, serveBytes([]byte("never served"))) + src := fx.src() + fx.srv.Close() + + _, err := fx.ensure(fx.ctx, src) + if !errors.Is(err, provider.ErrTransient) || errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensureImage() error = %v, want ErrTransient", err) + } + if !strings.Contains(err.Error(), "fetching image") { + t.Errorf("error %q should say what it was doing", err) + } +} + +func TestEnsureImage_shortReadRemovesVolumeAndIsTransient(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", "4096") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(make([]byte, 1024)) + w.(http.Flusher).Flush() + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + return + } + _ = conn.Close() + }) + src := fx.src() + + _, err := fx.ensure(fx.ctx, src) + if !errors.Is(err, provider.ErrTransient) || errors.Is(err, provider.ErrPermanent) { + t.Fatalf("ensureImage() error = %v, want ErrTransient", err) + } + if !strings.Contains(err.Error(), "1024 of 4096") { + t.Errorf("error %q should say how far the body got", err) + } + if fx.pooled.vol(src.volumeName()) != nil { + t.Error("base volume left behind after a short read") + } + if fx.pooled.vol(src.markerName()) != nil { + t.Error("marker written after a short read") + } + if fresh := fx.freshBackends(); len(fresh) != 1 || fresh[0].closeCount() != 1 { + t.Errorf("fresh connections = %d, want 1 and closed", len(fresh)) + } +} + +func TestEnsureImage_concurrentCallersFetchOnce(t *testing.T) { + t.Parallel() + body := randomBody(t, 256<<10) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + + const n = 20 + var wg sync.WaitGroup + paths := make([]string, n) + errs := make([]error, n) + for i := range n { + wg.Add(1) + go func() { + defer wg.Done() + paths[i], errs[i] = fx.ensure(fx.ctx, src) + }() + } + wg.Wait() + want := fakeVolPath(fakePoolName, src.volumeName()) + for i := range n { + if errs[i] != nil || paths[i] != want { + t.Errorf("caller %d: path = %q, err = %v; want %q, nil", i, paths[i], errs[i], want) + } + } + if fx.hits.Load() != 1 { + t.Errorf("server hits = %d, want 1", fx.hits.Load()) + } + baseUploads := 0 + for _, b := range fx.freshBackends() { + baseUploads += b.callCount("StorageVolUpload") + } + if baseUploads != 1 || len(fx.freshBackends()) != 1 { + t.Errorf("base uploads = %d over %d dedicated connections, want 1 over 1", baseUploads, len(fx.freshBackends())) + } + if n := fx.pooled.callCount("StorageVolUpload"); n != 1 { + t.Errorf("pooled uploads = %d, want 1 (the marker)", n) + } + if got := len(fx.sr.Ended()); got != 1 { + t.Errorf("download spans = %d, want 1", got) + } +} + +func TestEnsureImage_cancelledMidStreamReturnsPromptly(t *testing.T) { + t.Parallel() + started := make(chan struct{}) + fx := newImageFixture(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(1<<20)) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(make([]byte, 1024)) + w.(http.Flusher).Flush() + close(started) + <-r.Context().Done() + }) + src := fx.src() + + ctx, cancel := context.WithCancel(fx.ctx) + defer cancel() + go func() { + <-started + cancel() + }() + begin := time.Now() + _, err := fx.ensure(ctx, src) + elapsed := time.Since(begin) + if elapsed > time.Second { + t.Errorf("ensureImage() took %v after cancellation, want under 1s", elapsed) + } + if !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.Canceled) { + t.Fatalf("ensureImage() error = %v, want ErrTransient wrapping context.Canceled", err) + } + if fx.pooled.vol(src.volumeName()) != nil { + t.Error("base volume left behind after cancellation; the best-effort delete must not use the cancelled ctx") + } + if fx.pooled.vol(src.markerName()) != nil { + t.Error("marker written after cancellation") + } +} + +// TestEnsureImage_streamsWithoutBuffering proves the body is never +// materialised: the server counts what it has sent, and the fake records +// that count the moment StorageVolUpload is entered. Had ensureImage read +// the body first (io.ReadAll, a bytes.Buffer, a temp file), the server would +// have delivered all 64 MiB before the upload began; streaming means the +// upload starts while the server has sent only what the socket buffers +// hold. This holds under -race and in parallel, unlike a process-wide +// allocation measurement. The fake discards the bytes through a sha256, so +// the test itself holds no 64 MiB either. +func TestEnsureImage_streamsWithoutBuffering(t *testing.T) { + t.Parallel() + const size = 64 << 20 + block := randomBody(t, 64<<10) + var served atomic.Int64 + fx := newImageFixture(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(size)) + w.WriteHeader(http.StatusOK) + for sent := 0; sent < size; sent += len(block) { + if _, err := w.Write(block); err != nil { + return + } + served.Add(int64(len(block))) + } + }) + fx.pooled.discardUploads = true + var servedAtUploadStart atomic.Int64 + fx.onFresh = func(tw *fakeBackend) { + tw.hooks["StorageVolUpload"] = func() error { + servedAtUploadStart.Store(served.Load()) + return nil + } + } + src := fx.src() + src.SHA256 = func() string { + h := sha256.New() + for sent := 0; sent < size; sent += len(block) { + h.Write(block) + } + return hex.EncodeToString(h.Sum(nil)) + }() + + if _, err := fx.ensure(fx.ctx, src); err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + if got := servedAtUploadStart.Load(); got >= 16<<20 { + t.Errorf("server had sent %d bytes when the upload started, want under 16 MiB: the body was buffered", got) + } + base := fx.pooled.vol(src.volumeName()) + if base == nil || base.uploaded != size || base.sha256 != src.SHA256 { + t.Fatalf("base = %+v, want %d bytes uploaded with digest %s", base, size, src.SHA256) + } + if marker := fx.pooled.vol(src.markerName()); marker == nil || marker.sha256 != digestOf([]byte(src.SHA256)) { + t.Errorf("marker not written after the streamed upload") + } +} + +func TestEnsureImage_usesDedicatedConnectionForUpload(t *testing.T) { + t.Parallel() + body := randomBody(t, 64<<10) + fx := newImageFixture(t, serveBytes(body)) + src := fx.src() + + if _, err := fx.ensure(fx.ctx, src); err != nil { + t.Fatalf("ensureImage() error = %v", err) + } + fresh := fx.freshBackends() + if len(fresh) != 1 { + t.Fatalf("dedicated connections = %d, want 1", len(fresh)) + } + dedicated := fresh[0] + if dedicated.callCount("StorageVolCreateXML") != 1 || dedicated.callCount("StorageVolUpload") != 1 { + t.Errorf("dedicated connection calls = %v, want the base create and upload", dedicated.callLog()) + } + if dedicated.closeCount() != 1 { + t.Errorf("dedicated connection closed %d times, want 1", dedicated.closeCount()) + } + if fx.pooled.closeCount() != 0 { + t.Error("pooled connection was closed") + } + if n := fx.pooled.callCount("StorageVolUpload"); n != 1 { + t.Errorf("pooled uploads = %d, want 1 (the marker only)", n) + } + if fx.pooled.callCount("StorageVolLookupByName") == 0 || fx.pooled.callCount("StoragePoolRefresh") != 1 { + t.Errorf("pooled connection calls = %v, want the lookups and one refresh", fx.pooled.callLog()) + } + if fx.pooled.callCount("StorageVolCreateXML") != 1 { + t.Errorf("pooled StorageVolCreateXML = %d, want 1 (the marker)", fx.pooled.callCount("StorageVolCreateXML")) + } + if fx.dialer.hosts[0] != "beelink" { + t.Errorf("dialled %q, want beelink", fx.dialer.hosts[0]) + } +} + +func TestEnsureImage_dialFailureIsReturnedAndCreatesNothing(t *testing.T) { + t.Parallel() + fx := newImageFixture(t, serveBytes([]byte("image"))) + fx.dialer.err = transientf("ssh: connect: no route to host") + src := fx.src() + + _, err := fx.ensure(fx.ctx, src) + if !errors.Is(err, provider.ErrTransient) || !strings.Contains(err.Error(), "no route to host") { + t.Fatalf("ensureImage() error = %v, want the dial error", err) + } + if fx.pooled.vol(src.volumeName()) != nil || fx.pooled.vol(src.markerName()) != nil { + t.Error("volumes created although the dedicated connection failed") + } +} + +func TestEnsureImage_lockWaiterGivesUpWithContext(t *testing.T) { + t.Parallel() + release := make(chan struct{}) + fx := newImageFixture(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", "4096") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + select { + case <-release: + case <-r.Context().Done(): + return + } + _, _ = w.Write(make([]byte, 4096)) + }) + src := fx.src() + firstDone := make(chan error, 1) + go func() { + _, err := fx.ensure(fx.ctx, src) + firstDone <- err + }() + for fx.hits.Load() == 0 { + time.Sleep(time.Millisecond) + } + + ctx, cancel := context.WithTimeout(fx.ctx, 50*time.Millisecond) + defer cancel() + begin := time.Now() + _, err := fx.ensure(ctx, src) + if time.Since(begin) > time.Second || !errors.Is(err, provider.ErrTransient) || !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("waiter returned after %v with %v, want promptly with ErrTransient wrapping DeadlineExceeded", time.Since(begin), err) + } + close(release) + if err := <-firstDone; err != nil { + t.Errorf("first caller error = %v", err) + } + if fx.hits.Load() != 1 { + t.Errorf("server hits = %d, want 1: the waiter must not fetch", fx.hits.Load()) + } +} + +func TestCheckPoolSpace_belowNeededIsQuota(t *testing.T) { + t.Parallel() + be := newFakeBackend() + be.pools[fakePoolName].available = 5 << 30 + err := checkPoolSpace(context.Background(), be, testPool, fakePoolName, 6<<30) + if !errors.Is(err, provider.ErrQuotaExceeded) { + t.Fatalf("checkPoolSpace() error = %v, want ErrQuotaExceeded", err) + } + for _, want := range []string{"storage pool default", strconv.Itoa(5 << 30), strconv.Itoa(6 << 30)} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error %q lacks %q", err, want) + } + } +} + +func TestCheckPoolSpace_poolNotRunningIsPermanent(t *testing.T) { + t.Parallel() + for _, state := range []libvirt.StoragePoolState{libvirt.StoragePoolInactive, libvirt.StoragePoolBuilding, libvirt.StoragePoolDegraded, libvirt.StoragePoolInaccessible} { + t.Run(state.String(), func(t *testing.T) { + t.Parallel() + be := newFakeBackend() + be.pools[fakePoolName].state = state + err := checkPoolSpace(context.Background(), be, testPool, fakePoolName, 1) + if !errors.Is(err, provider.ErrPermanent) || !strings.Contains(err.Error(), "storage pool default is not running") { + t.Fatalf("checkPoolSpace() error = %v, want ErrPermanent 'not running'", err) + } + }) + } +} + +func TestCheckPoolSpace_enoughIsNil(t *testing.T) { + t.Parallel() + be := newFakeBackend() + be.pools[fakePoolName].available = 50 << 30 + for _, needed := range []uint64{0, 1, 50 << 30} { + if err := checkPoolSpace(context.Background(), be, testPool, fakePoolName, needed); err != nil { + t.Errorf("checkPoolSpace(needed=%d) error = %v, want nil", needed, err) + } + } +} + +func TestCheckPoolSpace_backendErrorIsReturned(t *testing.T) { + t.Parallel() + be := newFakeBackend() + be.failWith("StoragePoolGetInfo", fakeErr(libvirt.ErrNoStoragePool, "no pool")) + err := checkPoolSpace(context.Background(), be, testPool, fakePoolName, 1) + if !hasCode(err, libvirt.ErrNoStoragePool) || errors.Is(err, provider.ErrQuotaExceeded) { + t.Fatalf("checkPoolSpace() error = %v, want the libvirt error unclassified", err) + } +} + +func TestNeededBytes(t *testing.T) { + t.Parallel() + tests := []struct { + name string + imageBytes int64 + diskGB int64 + want uint64 + }{ + {name: "cached image and default disk", imageBytes: 0, diskGB: 10, want: 10<<30 + seedReserveBytes}, + {name: "uncached image adds its size", imageBytes: 435 << 20, diskGB: 10, want: 435<<20 + 10<<30 + seedReserveBytes}, + {name: "zero disk keeps the seed reserve", imageBytes: 0, diskGB: 0, want: seedReserveBytes}, + {name: "negative sizes count as zero", imageBytes: -1, diskGB: -3, want: seedReserveBytes}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := neededBytes(tc.imageBytes, tc.diskGB); got != tc.want { + t.Errorf("neededBytes(%d, %d) = %d, want %d", tc.imageBytes, tc.diskGB, got, tc.want) + } + }) + } +} + +func TestImageSizeIfUncached_headThenGetFallback(t *testing.T) { + t.Parallel() + body := randomBody(t, 12345) + tests := []struct { + name string + handler http.HandlerFunc + cached bool + wantSize int64 + wantMethods []string + wantClass error + wantErr string + }{ + { + name: "HEAD answers with the length", + handler: serveBytes(body), + wantSize: int64(len(body)), + wantMethods: []string{http.MethodHead}, + }, + { + name: "HEAD not allowed falls back to GET with the body closed", + handler: func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead { + http.Error(w, "nope", http.StatusMethodNotAllowed) + return + } + serveBytes(body)(w, r) + }, + wantSize: int64(len(body)), + wantMethods: []string{http.MethodHead, http.MethodGet}, + }, + { + name: "marker present means nothing to download", + handler: serveBytes(body), + cached: true, + wantSize: 0, + wantMethods: nil, + }, + { + name: "404 is permanent", + handler: serveStatus(http.StatusNotFound), + wantMethods: []string{http.MethodHead}, + wantClass: provider.ErrPermanent, + wantErr: "404", + }, + { + name: "503 is transient", + handler: serveStatus(http.StatusServiceUnavailable), + wantMethods: []string{http.MethodHead}, + wantClass: provider.ErrTransient, + }, + { + name: "no Content-Length is permanent", + handler: func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + }, + wantMethods: []string{http.MethodHead}, + wantClass: provider.ErrPermanent, + wantErr: "no Content-Length", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + var mu sync.Mutex + var methods []string + fx := newImageFixture(t, func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + methods = append(methods, r.Method) + mu.Unlock() + tc.handler(w, r) + }) + src := fx.src() + if tc.cached { + fx.putVol(src.markerName(), []byte(strings.Repeat("0", 64))) + } + got, err := fx.p.imageSizeIfUncached(fx.ctx, fx.pooled, testPool, src) + mu.Lock() + defer mu.Unlock() + if strings.Join(methods, ",") != strings.Join(tc.wantMethods, ",") { + t.Errorf("methods = %v, want %v", methods, tc.wantMethods) + } + if tc.wantClass != nil { + if !errors.Is(err, tc.wantClass) || !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("imageSizeIfUncached() error = %v, want %v containing %q", err, tc.wantClass, tc.wantErr) + } + return + } + if err != nil { + t.Fatalf("imageSizeIfUncached() error = %v", err) + } + if got != tc.wantSize { + t.Errorf("size = %d, want %d", got, tc.wantSize) + } + }) + } +} + +func TestReadMarker_capsTheRead(t *testing.T) { + t.Parallel() + be := newFakeBackend() + huge := bytes.Repeat([]byte("x"), 1<<20) + be.pools[fakePoolName].vols["m"] = &fakeVol{content: huge} + got, err := readMarker(context.Background(), be, libvirt.StorageVol{Pool: fakePoolName, Name: "m"}) + if err != nil { + t.Fatal(err) + } + if len(got) != markerReadCap { + t.Errorf("read %d bytes, want the cap %d", len(got), markerReadCap) + } + w := &cappedWriter{limit: 4} + if _, err := io.Copy(w, strings.NewReader("abcdefgh")); err != nil || string(w.buf) != "abcd" { + t.Errorf("cappedWriter kept %q, want abcd", w.buf) + } +} diff --git a/internal/provider/kvm2/kvm2.go b/internal/provider/kvm2/kvm2.go index b3e6e72..50625c0 100644 --- a/internal/provider/kvm2/kvm2.go +++ b/internal/provider/kvm2/kvm2.go @@ -7,13 +7,16 @@ package kvm2 import ( "context" "errors" + "net/http" "os" + "sync" "time" "github.com/go-logr/logr" logf "sigs.k8s.io/controller-runtime/pkg/log" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider" + "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" ) const ( @@ -25,10 +28,33 @@ const ( // Provider is one configured kvm2 entry: a set of hosts sharing one SSH key // and one image and machine-type catalogue. type Provider struct { - name string - cfg provider.KVMConfig - now func() time.Time - hosts *hostTable + name string + cfg provider.KVMConfig + now func() time.Time + hosts *hostTable + http *http.Client + budgets budgets + tracing []tracing.Option + + // imageLocks serialises image fetches per (host, volume), spec entry 19. + imageMu sync.Mutex + imageLocks map[string]chan struct{} +} + +// providerOption adjusts what newProvider builds; tests inject an HTTP +// client bound to a test server, a span recorder and shorter budgets. +type providerOption func(*Provider) + +func withHTTPClient(c *http.Client) providerOption { + return func(p *Provider) { p.http = c } +} + +func withBudgets(b budgets) providerOption { + return func(p *Provider) { p.budgets = b } +} + +func withTracing(opts ...tracing.Option) providerOption { + return func(p *Provider) { p.tracing = opts } } var _ provider.Provider = (*Provider)(nil) @@ -59,9 +85,26 @@ func New(ctx context.Context, pc provider.ProviderConfig) (provider.Provider, er } // newProvider is the constructor tests use: d hands out the backends, the -// production one being newLibvirtDialer. -func newProvider(name string, cfg provider.KVMConfig, d dialer) *Provider { - return &Provider{name: name, cfg: withDefaults(cfg), now: time.Now, hosts: newHostTable(d)} +// production one being newLibvirtDialer. Without options the image +// downloads use http.DefaultClient (system CA roots, no client timeout: the +// per-request ctx carries the transfer budget) and the default budgets. +func newProvider(name string, cfg provider.KVMConfig, d dialer, opts ...providerOption) *Provider { + p := &Provider{ + name: name, + cfg: withDefaults(cfg), + now: time.Now, + hosts: newHostTable(d), + http: http.DefaultClient, + budgets: defaultBudgets, + imageLocks: make(map[string]chan struct{}), + } + for _, opt := range opts { + opt(p) + } + if p.http == nil { + p.http = http.DefaultClient + } + return p } // withDefaults returns cfg with every optional field filled in, so the rest