a73x

8c746ce1

feat(agent): gzipped images are decompressed in-process, sparsely

a73x   2026-08-06 09:12

Commit message
feat(agent): gzipped images are decompressed in-process, sparsely

Gzip wraps a disk image rather than being one, so a compressed download is
decompressed and the result treated as the image. The writer punches holes
rather than writing runs of zeros, so a sparse image occupies what it uses.

The decompressed head is sniffed before a byte lands, so a gzipped qcow2 is
refused outright instead of landing in the cache under a `.raw` name to fail
later at boot. Corrupt gzip is permanent for the same reason a checksum mismatch
is: the sha already matched, so re-fetching returns the same bytes.
docs/assumptions.md
Old New
@@ -123,3 +123,23 @@ reaches it over the LAN rather than hairpinning out through the uplink, and
123 that hosts stay on that LAN. 123 that hosts stay on that LAN.
124 **Unverified**: one fleet host is a laptop that leaves the network, and a 124 **Unverified**: one fleet host is a laptop that leaves the network, and a
125 public hostname resolving to a public address would route the long way round. 125 public hostname resolving to a public address would route the long way round.
126
127 ### A gzipped image wraps a raw one
128
129 Gzip is transport, not a disk format, so a `.gz` download is decompressed and
130 the result treated as the image. Underpins publishing guest images gzipped—802
131 MB on the wire against 3.5 GB raw, and smaller than the qcow2 it replaces.
132 **Proven** for the failure that matters: the decompressed head is sniffed
133 before a byte is written, so a gzipped qcow2 is rejected permanently instead of
134 landing in the cache under a `.raw` name and failing later at boot. Corrupt
135 gzip is permanent for the same reason a checksum mismatch is—the sha already
136 matched, so re-fetching returns the same bytes.
137
138 ### Decompression is cheaper than the download it saves
139
140 Measured on the real image: 802 MB gzipped, 6.8s to decompress with C gunzip
141 and 14.4s writing sparsely. Go's `compress/gzip` is slower, so budget 15-25s.
142 Underpins choosing gzipped raw over plain raw.
143 **Partly proven**: measured on one host (6 vCPU, nested). A slow host pays more
144 CPU for less transfer, which is the right trade on a LAN and the wrong one only
145 if a host is very slow and very well connected.
internal/agent/imagecache/format.go
Old New
@@ -43,6 +43,20 @@ func sniff(head []byte) imageFormat {
43 } 43 }
44 } 44 }
45 45
46 // permanentError marks a failure no retry can fix. reconcile matches the
47 // Permanent() method structurally (errors.As against an anonymous interface),
48 // so this mirrors cloudhv's and inert's markers instead of sharing one: what a
49 // VMM driver and an image cache call permanent have nothing else in common.
50 type permanentError struct{ err error }
51
52 func (e permanentError) Error() string { return e.err.Error() }
53 func (e permanentError) Unwrap() error { return e.err }
54 func (e permanentError) Permanent() bool { return true }
55
56 func permanentf(format string, args ...any) error {
57 return permanentError{err: fmt.Errorf(format, args...)}
58 }
59
46 // sniffFile classifies the image file at path. A file shorter than magicLen is 60 // sniffFile classifies the image file at path. A file shorter than magicLen is
47 // read in full and classified on what it has, so a truncated or empty image is 61 // read in full and classified on what it has, so a truncated or empty image is
48 // raw rather than an error — the checksum has already vouched for the bytes by 62 // raw rather than an error — the checksum has already vouched for the bytes by
internal/agent/imagecache/imagecache.go
Old New
@@ -15,9 +15,12 @@
15 package imagecache 15 package imagecache
16 16
17 import ( 17 import (
18 "bytes"
19 "compress/gzip"
18 "context" 20 "context"
19 "crypto/sha256" 21 "crypto/sha256"
20 "encoding/hex" 22 "encoding/hex"
23 "errors"
21 "fmt" 24 "fmt"
22 "io" 25 "io"
23 "log/slog" 26 "log/slog"
@@ -176,8 +179,8 @@ func (c *Cache) ensureOnce(ctx context.Context, url, sha string) (string, error)
176 return final, nil 179 return final, nil
177 } 180 }
178 181
179 // Convert to a UNIQUE temp file first, then rename onto final atomically so a 182 // Materialize into a UNIQUE temp file first, then rename onto final atomically
180 // crash mid-convert cannot leave a corrupt file at the final path. The temp 183 // so a crash mid-write cannot leave a corrupt file at the final path. The temp
181 // name is unique (not a fixed "<sha>.converting") so two concurrent Ensure 184 // name is unique (not a fixed "<sha>.converting") so two concurrent Ensure
182 // calls for the same sha can never converge on one file and corrupt it. The 185 // calls for the same sha can never converge on one file and corrupt it. The
183 // suffix deliberately does NOT end in ".raw", so evict()'s "*.raw" glob never 186 // suffix deliberately does NOT end in ".raw", so evict()'s "*.raw" glob never
@@ -188,10 +191,19 @@ func (c *Cache) ensureOnce(ctx context.Context, url, sha string) (string, error)
188 } 191 }
189 converting := cf.Name() 192 converting := cf.Name()
190 _ = cf.Close() 193 _ = cf.Close()
191 _ = os.Remove(converting) // reserve the unique name; let qemu-img create it fresh 194 _ = os.Remove(converting) // reserve the unique name; let the writer create it fresh
192 defer os.Remove(converting) 195 defer os.Remove(converting)
193 if _, err := c.run(ctx, "qemu-img", "convert", "-O", "raw", tmp, converting); err != nil { 196
194 return "", fmt.Errorf("qemu-img convert: %w", err) 197 // sniff returns exactly three formats and raw returned above, so these two
198 // cases are the whole space; a fourth format means editing both.
199 switch format {
200 case formatGzip:
201 err = decompress(url, tmp, converting)
202 case formatQcow2:
203 err = c.convert(ctx, url, tmp, converting)
204 }
205 if err != nil {
206 return "", err
195 } 207 }
196 if err := os.Rename(converting, final); err != nil { 208 if err := os.Rename(converting, final); err != nil {
197 return "", fmt.Errorf("imagecache: rename to final: %w", err) 209 return "", fmt.Errorf("imagecache: rename to final: %w", err)
@@ -200,6 +212,65 @@ func (c *Cache) ensureOnce(ctx context.Context, url, sha string) (string, error)
200 return final, nil 212 return final, nil
201 } 213 }
202 214
215 // decompress writes the gzipped raw image at src to dst, punching the long runs
216 // of zeros a cloud image is mostly made of into holes.
217 func decompress(url, src, dst string) error {
218 in, err := os.Open(src)
219 if err != nil {
220 return fmt.Errorf("imagecache: open download: %w", err)
221 }
222 defer in.Close()
223 zr, err := gzip.NewReader(in)
224 if err != nil {
225 // The download's sha already matched, so a broken gzip header is a
226 // property of the published artifact: re-fetching returns the same bytes
227 // and fails the same way.
228 return permanentf("imagecache: gzip header: %v", err)
229 }
230 defer zr.Close()
231
232 // Gzip wraps a RAW image. Sniff what is inside before writing a byte: a
233 // .qcow2.gz would otherwise decompress into a cache entry named <sha>.raw
234 // holding qcow2 bytes, and the guest would fail to boot for a reason nothing
235 // in the error names — exactly the illegible failure this package exists to
236 // prevent. The peeked bytes go back in front of the stream so the writer
237 // still sees the whole image. Failing before the output file is created is
238 // what leaves no temp behind.
239 head := make([]byte, magicLen)
240 n, err := io.ReadFull(zr, head)
241 if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
242 return fmt.Errorf("imagecache: read decompressed header: %w", err)
243 }
244 head = head[:n]
245 if sniff(head) == formatQcow2 {
246 // Permanent for the same reason a broken header is: the sha matched, so
247 // these bytes are what the publisher published.
248 return permanentf("imagecache: %s decompresses to a qcow2 image; eitri guest images must be raw or gzipped raw (.raw.gz) — convert it with `qemu-img convert -O raw` where you publish it", url)
249 }
250
251 out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
252 if err != nil {
253 return fmt.Errorf("imagecache: create image: %w", err)
254 }
255 if err := writeSparse(out, io.MultiReader(bytes.NewReader(head), zr)); err != nil {
256 out.Close()
257 return fmt.Errorf("imagecache: decompress: %w", err)
258 }
259 if err := out.Close(); err != nil {
260 return fmt.Errorf("imagecache: close image: %w", err)
261 }
262 return nil
263 }
264
265 // convert turns a qcow2 download into a raw image with qemu-img — the one
266 // format that still needs a tool, and the one a host without it cannot boot.
267 func (c *Cache) convert(ctx context.Context, url, src, dst string) error {
268 if _, err := c.run(ctx, "qemu-img", "convert", "-O", "raw", src, dst); err != nil {
269 return fmt.Errorf("qemu-img convert %s: %w", url, err)
270 }
271 return nil
272 }
273
203 // SweepTemps removes abandoned download and convert temporaries — the two temp 274 // SweepTemps removes abandoned download and convert temporaries — the two temp
204 // forms Ensure writes before its atomic rename. Neither matches evict's "*.raw" 275 // forms Ensure writes before its atomic rename. Neither matches evict's "*.raw"
205 // glob, so nothing else ever reclaims them, and an agent killed mid-fetch (the 276 // glob, so nothing else ever reclaims them, and an agent killed mid-fetch (the
internal/agent/imagecache/imagecache_test.go
Old New
@@ -2,6 +2,7 @@ package imagecache
2 2
3 import ( 3 import (
4 "bytes" 4 "bytes"
5 "compress/gzip"
5 "context" 6 "context"
6 "crypto/sha256" 7 "crypto/sha256"
7 "encoding/hex" 8 "encoding/hex"
@@ -342,3 +343,85 @@ func TestConcurrentEnsureFetchesOnce(t *testing.T) {
342 require.NoError(t, err) 343 require.NoError(t, err)
343 assert.Equal(t, body, data, "the shared result must be the real image") 344 assert.Equal(t, body, data, "the shared result must be the real image")
344 } 345 }
346
347 // gzipped returns body compressed, the shape a published .raw.gz has.
348 func gzipped(t *testing.T, body []byte) []byte {
349 t.Helper()
350 var buf bytes.Buffer
351 zw := gzip.NewWriter(&buf)
352 _, err := zw.Write(body)
353 require.NoError(t, err)
354 require.NoError(t, zw.Close())
355 return buf.Bytes()
356 }
357
358 // qcow2 prefixes body with the qcow2 magic, so a test image is classified the
359 // way a real one would be.
360 func qcow2(body []byte) []byte {
361 return append([]byte{'Q', 'F', 'I', 0xfb}, body...)
362 }
363
364 // TestEnsureDecompressesGzippedRaw pins the format that keeps a published raw
365 // image smaller than the qcow2 it replaces: gzip is undone in-process, byte for
366 // byte, with no external tool.
367 func TestEnsureDecompressesGzippedRaw(t *testing.T) {
368 raw := concat(
369 bytes.Repeat([]byte{0xeb}, 4096),
370 make([]byte, 2<<20),
371 bytes.Repeat([]byte{0x77}, 4096),
372 )
373 ts, sum := serve(t, gzipped(t, raw))
374 dir := t.TempDir()
375 c := New(dir, noRunner(t))
376
377 p, err := c.Ensure(context.Background(), ts.URL+"/img.raw.gz", sum)
378 require.NoError(t, err)
379
380 got, err := os.ReadFile(p)
381 require.NoError(t, err)
382 assert.True(t, bytes.Equal(raw, got), "decompressed image must be byte-identical")
383
384 fi, err := os.Stat(p)
385 require.NoError(t, err)
386 assert.Equal(t, int64(len(raw)), fi.Size())
387
388 entries, err := os.ReadDir(dir)
389 require.NoError(t, err)
390 assert.Len(t, entries, 1, "no temp may survive a successful decompress")
391 }
392
393 // TestEnsureRejectsCorruptGzipPermanently pins the classification: the sha
394 // already matched, so the bytes are exactly what the publisher published and
395 // re-downloading them cannot produce a different outcome.
396 func TestEnsureRejectsCorruptGzipPermanently(t *testing.T) {
397 ts, sum := serve(t, []byte{0x1f, 0x8b, 'g', 'a', 'r', 'b', 'a', 'g', 'e', '!'})
398 c := New(t.TempDir(), noRunner(t))
399
400 _, err := c.Ensure(context.Background(), ts.URL+"/img.raw.gz", sum)
401 require.Error(t, err)
402 var p interface{ Permanent() bool }
403 require.ErrorAs(t, err, &p, "a broken gzip header must be marked permanent")
404 assert.True(t, p.Permanent())
405 }
406
407 // TestEnsureRejectsGzippedQcow2Permanently pins the one trap gzip opens: gzip
408 // wraps a RAW image, and a .qcow2.gz would otherwise decompress into a cache
409 // entry named <sha>.raw holding qcow2 bytes — a guest that fails to boot for a
410 // reason no error names. Permanent because the sha already matched: the bytes
411 // are what the publisher published, and re-fetching cannot change them.
412 func TestEnsureRejectsGzippedQcow2Permanently(t *testing.T) {
413 ts, sum := serve(t, gzipped(t, qcow2(bytes.Repeat([]byte{0x42}, 4096))))
414 dir := t.TempDir()
415 c := New(dir, noRunner(t))
416
417 _, err := c.Ensure(context.Background(), ts.URL+"/img.qcow2.gz", sum)
418 require.Error(t, err)
419 var p interface{ Permanent() bool }
420 require.ErrorAs(t, err, &p, "a gzipped qcow2 can never decompress to a raw image")
421 assert.True(t, p.Permanent())
422 assert.Contains(t, err.Error(), "qcow2", "the operator must learn what the image actually is")
423
424 entries, err := os.ReadDir(dir)
425 require.NoError(t, err)
426 assert.Empty(t, entries, "a rejected image must leave nothing in the cache")
427 }
internal/agent/imagecache/sparse.go
Old New
@@ -0,0 +1,69 @@
1 package imagecache
2
3 import (
4 "bytes"
5 "errors"
6 "io"
7 "os"
8 )
9
10 const (
11 // blockSize is the granularity at which zeros are detected. A hole can only
12 // start on a filesystem block boundary, so 4 KiB — the block size on ext4,
13 // xfs, btrfs and APFS — is the finest granularity that buys anything.
14 blockSize = 4096
15 // bufSize keeps reads chunky. It is a multiple of blockSize so every buffer
16 // boundary is also a block boundary, and a zero run crossing two buffers is
17 // still punched as one hole.
18 bufSize = 1 << 20
19 )
20
21 // zeroBlock is the comparand for hole detection: bytes.Equal against it is one
22 // memequal, which is why scanning a multi-gigabyte image block by block costs
23 // nothing next to the decompression feeding it.
24 var zeroBlock = make([]byte, blockSize)
25
26 // writeSparse copies src into dst, seeking past all-zero blocks instead of
27 // writing them, so a decompressed cloud image — which is mostly zeros — costs
28 // only the disk its real data occupies.
29 //
30 // dst ends at the stream's exact apparent size. That needs the closing
31 // Truncate: seeking past the end of a file does not extend it, so a stream
32 // ending in zeros would otherwise leave a file shorter than the image it
33 // claims to be, and PrepareRootDisk sizes a guest's disk against that number.
34 func writeSparse(dst *os.File, src io.Reader) error {
35 buf := make([]byte, bufSize)
36 var size int64 // logical bytes consumed, holes included
37 var hole int64 // consecutive zero bytes not yet seeked past
38 for {
39 n, readErr := io.ReadFull(src, buf)
40 for chunk := buf[:n]; len(chunk) > 0; {
41 b := chunk
42 if len(b) > blockSize {
43 b = b[:blockSize]
44 }
45 chunk = chunk[len(b):]
46 size += int64(len(b))
47 if bytes.Equal(b, zeroBlock[:len(b)]) {
48 hole += int64(len(b))
49 continue
50 }
51 if hole > 0 {
52 if _, err := dst.Seek(hole, io.SeekCurrent); err != nil {
53 return err
54 }
55 hole = 0
56 }
57 if _, err := dst.Write(b); err != nil {
58 return err
59 }
60 }
61 if errors.Is(readErr, io.EOF) || errors.Is(readErr, io.ErrUnexpectedEOF) {
62 break
63 }
64 if readErr != nil {
65 return readErr
66 }
67 }
68 return dst.Truncate(size)
69 }
internal/agent/imagecache/sparse_test.go
Old New
@@ -0,0 +1,127 @@
1 package imagecache
2
3 import (
4 "bytes"
5 "os"
6 "path/filepath"
7 "syscall"
8 "testing"
9
10 "github.com/stretchr/testify/assert"
11 "github.com/stretchr/testify/require"
12 )
13
14 // allocatedBytes reports how much disk a file actually occupies, as opposed to
15 // the size it claims. st_blocks is in 512-byte units by POSIX definition,
16 // regardless of the filesystem's own block size.
17 func allocatedBytes(t *testing.T, path string) int64 {
18 t.Helper()
19 fi, err := os.Stat(path)
20 require.NoError(t, err)
21 st, ok := fi.Sys().(*syscall.Stat_t)
22 require.True(t, ok, "no syscall.Stat_t for %s", path)
23 return st.Blocks * 512
24 }
25
26 // requireSparseFS skips the calling test when the temp filesystem cannot hold
27 // holes at all. Without this probe a sparseness assertion is really an
28 // assertion about the machine it runs on.
29 func requireSparseFS(t *testing.T, dir string) {
30 t.Helper()
31 p := filepath.Join(dir, "sparse-probe")
32 f, err := os.Create(p)
33 require.NoError(t, err)
34 require.NoError(t, f.Truncate(8<<20))
35 require.NoError(t, f.Close())
36 alloc := allocatedBytes(t, p)
37 require.NoError(t, os.Remove(p))
38 if alloc > 1<<20 {
39 t.Skipf("filesystem under %s does not support sparse files (8 MiB hole allocated %d bytes)", dir, alloc)
40 }
41 }
42
43 // TestWriteSparse pins both halves of the contract at once: the file that comes
44 // out is byte-identical to the stream that went in AND its apparent size is
45 // exact, while the runs of zeros cost no disk.
46 func TestWriteSparse(t *testing.T) {
47 data := bytes.Repeat([]byte{0xAB}, blockSize)
48
49 tests := []struct {
50 name string
51 body []byte
52 sparse bool // must occupy far less disk than it claims
53 }{
54 {
55 name: "solid data only",
56 body: bytes.Repeat([]byte{0x5A}, 3*blockSize),
57 },
58 {
59 name: "hole between two data blocks",
60 body: concat(data, make([]byte, 8<<20), data),
61 sparse: true,
62 },
63 {
64 name: "leading hole",
65 body: concat(make([]byte, 8<<20), data),
66 sparse: true,
67 },
68 {
69 name: "trailing hole: apparent size must survive it",
70 body: concat(data, make([]byte, 8<<20)),
71 sparse: true,
72 },
73 {
74 name: "zero run spanning several read buffers",
75 body: concat(data, make([]byte, 3*bufSize), data),
76 sparse: true,
77 },
78 {
79 name: "sub-block tail",
80 body: concat(data, []byte("tail")),
81 },
82 {
83 name: "unaligned data after a hole",
84 body: concat(make([]byte, 8<<20), []byte("x")),
85 },
86 {
87 name: "empty stream",
88 body: nil,
89 },
90 }
91
92 for _, tc := range tests {
93 t.Run(tc.name, func(t *testing.T) {
94 dir := t.TempDir()
95 if tc.sparse {
96 requireSparseFS(t, dir)
97 }
98 p := filepath.Join(dir, "out.raw")
99 f, err := os.Create(p)
100 require.NoError(t, err)
101 require.NoError(t, writeSparse(f, bytes.NewReader(tc.body)))
102 require.NoError(t, f.Close())
103
104 got, err := os.ReadFile(p)
105 require.NoError(t, err)
106 assert.True(t, bytes.Equal(tc.body, got), "decompressed bytes must be identical")
107
108 fi, err := os.Stat(p)
109 require.NoError(t, err)
110 assert.Equal(t, int64(len(tc.body)), fi.Size(), "apparent size must be exact")
111
112 if tc.sparse {
113 alloc := allocatedBytes(t, p)
114 assert.Less(t, alloc, fi.Size()/2,
115 "zero runs must be holes: %d bytes allocated for an apparent %d", alloc, fi.Size())
116 }
117 })
118 }
119 }
120
121 func concat(parts ...[]byte) []byte {
122 var out []byte
123 for _, p := range parts {
124 out = append(out, p...)
125 }
126 return out
127 }