kacerr/egress-proxies-operator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

kvm2: base image cache with a streamed upload and the pool-space precheck

Browse Source 
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
kacerr
2026-09-16 18:19:05 +00:00
co-authored by Claude Fable 5.1
parent b45f83327a
commit 6b3c985b4a
5 changed files with 1605 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans-executions/2026-09-16-1333-kvm2-provider.md
+40 -1
View File
@@ -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 2 — Config wiring, IDs, state mapping, classify
- [x] Step 3 — Pure builders: domain and volume XML, metadata, seed - [x] Step 3 — Pure builders: domain and volume XML, metadata, seed
- [x] Step 4 — Transport, seam, decorator, fakes - [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 6 — The four contract methods
- [ ] Step 7 — Checker and the CRD comment - [ ] Step 7 — Checker and the CRD comment
- [ ] Step 8 — Deployment files, Makefile, integration test - [ ] 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`. 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".
internal/provider/kvm2/fake_test.go
+60 -3
View File
@@ -2,6 +2,8 @@ package kvm2
import ( import (
"context" "context"
"crypto/sha256"
"encoding/hex"
"encoding/xml" "encoding/xml"
"fmt" "fmt"
"io" "io"
@@ -18,7 +20,9 @@ import (
// libvirt host for the provider's Create, Get, Delete and ListByTag, with // libvirt host for the provider's Create, Get, Delete and ListByTag, with
// per-method error hooks and blocking, call counters and an ordered log. // per-method error hooks and blocking, call counters and an ordered log.
type fakeBackend struct { 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 domains map[string]*fakeDomain
pools map[string]*fakePool pools map[string]*fakePool
hostname string hostname string
@@ -28,6 +32,10 @@ type fakeBackend struct {
hooks map[string]func() error hooks map[string]func() error
// block makes a method wait on its channel (or ctx) before running. // block makes a method wait on its channel (or ctx) before running.
block map[string]chan struct{} 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 calls map[string]int
log []string log []string
@@ -55,12 +63,16 @@ type fakePool struct {
type fakeVol struct { type fakeVol struct {
content []byte content []byte
capacity uint64 capacity uint64
// uploaded and sha256 record a discarded upload (see discardUploads).
uploaded int64
sha256 string
} }
const fakePoolName = "default" const fakePoolName = "default"
func newFakeBackend() *fakeBackend { func newFakeBackend() *fakeBackend {
return &fakeBackend{ return &fakeBackend{
mu: &sync.Mutex{},
domains: make(map[string]*fakeDomain), domains: make(map[string]*fakeDomain),
pools: map[string]*fakePool{fakePoolName: newFakePool()}, pools: map[string]*fakePool{fakePoolName: newFakePool()},
hostname: "fake-host", 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 { func newFakePool() *fakePool {
return &fakePool{state: libvirt.StoragePoolRunning, capacity: 100 << 30, available: 50 << 30, vols: make(map[string]*fakeVol)} 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 { if err := f.enter(ctx, "StorageVolUpload"); err != nil {
return err return err
} }
var content []byte var (
content []byte
uploaded int64
digest string
)
if r != nil { if r != nil {
var err error 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 return err
} }
} }
@@ -436,6 +491,8 @@ func (f *fakeBackend) StorageVolUpload(ctx context.Context, vol libvirt.StorageV
return err return err
} }
v.content = content v.content = content
v.uploaded = uploaded
v.sha256 = digest
return nil return nil
} }
internal/provider/kvm2/image.go
+474
View File
@@ -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
}
internal/provider/kvm2/image_test.go
+981
View File
@@ -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)
}
}
internal/provider/kvm2/kvm2.go
+46 -3
View File
@@ -7,13 +7,16 @@ package kvm2
import ( import (
"context" "context"
"errors" "errors"
"net/http"
"os" "os"
"sync"
"time" "time"
"github.com/go-logr/logr" "github.com/go-logr/logr"
logf "sigs.k8s.io/controller-runtime/pkg/log" 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/provider"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
) )
const ( const (
@@ -29,6 +32,29 @@ type Provider struct {
cfg provider.KVMConfig cfg provider.KVMConfig
now func() time.Time now func() time.Time
hosts *hostTable 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) 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 // newProvider is the constructor tests use: d hands out the backends, the
// production one being newLibvirtDialer. // production one being newLibvirtDialer. Without options the image
func newProvider(name string, cfg provider.KVMConfig, d dialer) *Provider { // downloads use http.DefaultClient (system CA roots, no client timeout: the
return &Provider{name: name, cfg: withDefaults(cfg), now: time.Now, hosts: newHostTable(d)} // 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 // withDefaults returns cfg with every optional field filled in, so the rest