From 2d38b9e9881ccd95ae16832f2799aa1d494bf75c Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 22 Jun 2026 14:47:48 +0800 Subject: [PATCH 1/9] feat: add node-local overlayfs rootfs cache for actor resume Implements issue #228: cache extracted rootfs per image digest on each node, and materialize per-actor bundles as overlayfs mounts instead of re-untarring on every restore. Changes: - New rootfscache package (cmd/atelet/internal/rootfscache) - New overlay.go with mount/unmount helpers - Modified prepareOCIDirectory for overlayfs integration - Updated resetActorDirs to unmount before cleanup - Added rootfs cache paths to ateompath - Unit tests for cache hit/miss, concurrent access, eviction --- .../internal/rootfscache/rootfscache.go | 624 ++++++++++++++++++ .../internal/rootfscache/rootfscache_test.go | 336 ++++++++++ cmd/atelet/main.go | 25 + cmd/atelet/oci.go | 241 +++---- cmd/atelet/overlay.go | 87 +++ internal/ateompath/ateompath.go | 13 + 6 files changed, 1170 insertions(+), 156 deletions(-) create mode 100644 cmd/atelet/internal/rootfscache/rootfscache.go create mode 100644 cmd/atelet/internal/rootfscache/rootfscache_test.go create mode 100644 cmd/atelet/overlay.go diff --git a/cmd/atelet/internal/rootfscache/rootfscache.go b/cmd/atelet/internal/rootfscache/rootfscache.go new file mode 100644 index 000000000..ef7163603 --- /dev/null +++ b/cmd/atelet/internal/rootfscache/rootfscache.go @@ -0,0 +1,624 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package rootfscache provides a node-local, digest-keyed cache of extracted +// OCI rootfs directories. On a cache hit the caller can set up an overlayfs +// mount instead of re-extracting the image tarball, reducing per-restore +// latency from seconds to sub-second. +package rootfscache + +import ( + "archive/tar" + "context" + "errors" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// DefaultMaxCacheBytes is the default disk budget for the rootfs cache (20 GiB). +const DefaultMaxCacheBytes int64 = 20 * 1024 * 1024 * 1024 + +// readySentinel is the name of the sentinel file written atomically after a +// rootfs has been fully extracted. Its presence is the cache-hit signal. +const readySentinel = ".ready" + +// lastAccessFile is the name of the file that stores the unix-timestamp of the +// most recent cache hit, used for LRU eviction. +const lastAccessFile = ".last_access" + +// entryState tracks one cached digest. All fields are immutable after +// construction except lastAccess, which is updated on cache hits. +type entryState struct { + digest string + lowerDir string + sizeBytes int64 + lastAccess time.Time +} + +// Cache is a node-local, digest-keyed rootfs cache. It is safe for +// concurrent use. +type Cache struct { + basePath string + maxCacheBytes int64 + + mu sync.Mutex + entries map[string]*entryState // keyed by digest + + // inflight deduplicates concurrent EnsureRootfs calls for the same + // digest: the first goroutine extracts while others wait. + inflight map[string]*inflightEntry + + // metrics + cacheHits metric.Int64Counter + cacheMisses metric.Int64Counter +} + +type inflightEntry struct { + done chan struct{} + // result is set before done is closed. + lowerDir string + cached bool + err error +} + +// New creates a Cache rooted at basePath. The directory is created if it does +// not exist. maxCacheBytes caps total disk usage; pass 0 for +// DefaultMaxCacheBytes. +func New(ctx context.Context, basePath string, maxCacheBytes int64) (*Cache, error) { + if maxCacheBytes <= 0 { + maxCacheBytes = DefaultMaxCacheBytes + } + if err := os.MkdirAll(basePath, 0o700); err != nil { + return nil, fmt.Errorf("creating rootfs cache dir: %w", err) + } + + meter := otel.Meter("atelet") + cacheHits, err := meter.Int64Counter("atelet.rootfs_cache.hit", + metric.WithDescription("Rootfs cache hits (overlay path taken)"), + ) + if err != nil { + return nil, fmt.Errorf("creating hit metric: %w", err) + } + cacheMisses, err := meter.Int64Counter("atelet.rootfs_cache.miss", + metric.WithDescription("Rootfs cache misses (untar required)"), + ) + if err != nil { + return nil, fmt.Errorf("creating miss metric: %w", err) + } + + c := &Cache{ + basePath: basePath, + maxCacheBytes: maxCacheBytes, + entries: make(map[string]*entryState), + inflight: make(map[string]*inflightEntry), + cacheHits: cacheHits, + cacheMisses: cacheMisses, + } + + // Load existing cache entries from disk. + if err := c.loadIndex(ctx); err != nil { + slog.WarnContext(ctx, "Failed to load rootfs cache index, starting empty", slog.Any("err", err)) + } + + return c, nil +} + +// LowerDir returns the read-only rootfs directory for the given digest, or "" +// if the digest is not cached. +func (c *Cache) LowerDir(digest string) string { + c.mu.Lock() + defer c.mu.Unlock() + if e, ok := c.entries[digest]; ok { + return e.lowerDir + } + return "" +} + +// EnsureRootfs guarantees that the rootfs for digest is extracted into the +// cache. On a cache hit it returns the lowerDir immediately without reading +// tarData. On a miss it consumes tarData to populate the cache. +// +// Returns: +// +// lowerDir – the read-only rootfs path (non-empty on success) +// cached – true if the cache was hit (tarData was NOT consumed) +// err – any error +// +// The digest MUST be a valid directory name (hex-encoded sha256, no slashes). +func (c *Cache) EnsureRootfs(ctx context.Context, digest string, tarData io.Reader) (string, bool, error) { + tracer := otel.Tracer("rootfscache") + ctx, span := tracer.Start(ctx, "EnsureRootfs") + span.SetAttributes(attribute.String("digest", digest)) + defer span.End() + + if err := validateDigest(digest); err != nil { + return "", false, err + } + + // Fast path: already cached. + c.mu.Lock() + if e, ok := c.entries[digest]; ok { + c.mu.Unlock() + c.cacheHits.Add(ctx, 1) + span.SetAttributes(attribute.Bool("hit", true)) + _ = c.touchAccess(digest) + return e.lowerDir, true, nil + } + + // Deduplicate concurrent requests for the same digest. + if infl, ok := c.inflight[digest]; ok { + c.mu.Unlock() + select { + case <-infl.done: + return infl.lowerDir, infl.cached, infl.err + case <-ctx.Done(): + return "", false, ctx.Err() + } + } + + // We are the first goroutine for this digest — set up the inflight entry + // so others can wait on us. + infl := &inflightEntry{done: make(chan struct{})} + c.inflight[digest] = infl + c.mu.Unlock() + + // Do the actual extraction outside the lock. + lowerDir, err := c.extract(ctx, digest, tarData) + + c.mu.Lock() + delete(c.inflight, digest) + c.mu.Unlock() + + if err != nil { + infl.lowerDir = "" + infl.cached = false + infl.err = err + close(infl.done) + return "", false, err + } + + c.cacheMisses.Add(ctx, 1) + span.SetAttributes(attribute.Bool("hit", false)) + + infl.lowerDir = lowerDir + infl.cached = false + infl.err = nil + close(infl.done) + return lowerDir, false, nil +} + +// extract untars tarData into the cache directory for digest, writes the +// .ready sentinel and .last_access file, and returns the lowerDir path. +func (c *Cache) extract(ctx context.Context, digest string, tarData io.Reader) (string, error) { + lowerDir := filepath.Join(c.basePath, digest, "lower") + + // Clean up any partial extraction from a previous crash. + if err := os.RemoveAll(filepath.Join(c.basePath, digest)); err != nil { + return "", fmt.Errorf("cleaning partial cache entry: %w", err) + } + if err := os.MkdirAll(lowerDir, 0o700); err != nil { + return "", fmt.Errorf("creating cache lower dir: %w", err) + } + + slog.InfoContext(ctx, "Rootfs cache miss, extracting", + slog.String("digest", digest), + slog.String("lowerDir", lowerDir), + ) + + if err := Untar(ctx, tarData, lowerDir); err != nil { + // Clean up on failure so the next attempt starts fresh. + _ = os.RemoveAll(filepath.Join(c.basePath, digest)) + return "", fmt.Errorf("extracting rootfs: %w", err) + } + + // Make the lower dir read-only (best effort; overlayfs lowerdir is + // inherently read-only, but this adds a layer of defense). + _ = chmodRecursive(lowerDir, 0o555) + + // Write the .ready sentinel atomically. + readyPath := filepath.Join(c.basePath, digest, readySentinel) + if err := os.WriteFile(readyPath, []byte(time.Now().Format(time.RFC3339)), 0o444); err != nil { + return "", fmt.Errorf("writing ready sentinel: %w", err) + } + + // Write the .last_access file. + if err := c.touchAccess(digest); err != nil { + return "", fmt.Errorf("writing last_access: %w", err) + } + + // Register in the in-memory index. + size := dirSize(lowerDir) + c.mu.Lock() + c.entries[digest] = &entryState{ + digest: digest, + lowerDir: lowerDir, + sizeBytes: size, + lastAccess: time.Now(), + } + c.mu.Unlock() + + // Best-effort eviction. + go c.evictIfNeeded(context.Background()) + + return lowerDir, nil +} + +// loadIndex scans the basePath for completed cache entries (those with a +// .ready sentinel) and populates the in-memory index. +func (c *Cache) loadIndex(ctx context.Context) error { + dirs, err := os.ReadDir(c.basePath) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + + var totalSize int64 + for _, d := range dirs { + if !d.IsDir() { + continue + } + digest := d.Name() + readyPath := filepath.Join(c.basePath, digest, readySentinel) + if _, err := os.Stat(readyPath); err != nil { + // Incomplete entry from a previous crash — remove it. + _ = os.RemoveAll(filepath.Join(c.basePath, digest)) + continue + } + lowerDir := filepath.Join(c.basePath, digest, "lower") + size := dirSize(lowerDir) + + // Read last_access time; fall back to ready file mtime. + lastAccess := readAccessTime(filepath.Join(c.basePath, digest)) + + c.entries[digest] = &entryState{ + digest: digest, + lowerDir: lowerDir, + sizeBytes: size, + lastAccess: lastAccess, + } + totalSize += size + } + + slog.InfoContext(ctx, "Loaded rootfs cache index", + slog.Int("entries", len(c.entries)), + slog.Int64("totalBytes", totalSize), + ) + return nil +} + +// touchAccess updates the .last_access file and in-memory timestamp for +// digest. +func (c *Cache) touchAccess(digest string) error { + now := time.Now() + path := filepath.Join(c.basePath, digest, lastAccessFile) + if err := os.WriteFile(path, []byte(now.Format(time.RFC3339Nano)), 0o644); err != nil { + return err + } + c.mu.Lock() + if e, ok := c.entries[digest]; ok { + e.lastAccess = now + } + c.mu.Unlock() + return nil +} + +// evictIfNeeded removes the oldest entries until total cache size is within +// the budget. It is called asynchronously after each extraction. +func (c *Cache) evictIfNeeded(ctx context.Context) { + c.mu.Lock() + defer c.mu.Unlock() + + var total int64 + for _, e := range c.entries { + total += e.sizeBytes + } + + for total > c.maxCacheBytes && len(c.entries) > 0 { + // Find the entry with the oldest lastAccess. + var oldest *entryState + for _, e := range c.entries { + if oldest == nil || e.lastAccess.Before(oldest.lastAccess) { + oldest = e + } + } + if oldest == nil { + break + } + + slog.InfoContext(ctx, "Evicting rootfs cache entry", + slog.String("digest", oldest.digest), + slog.Int64("sizeBytes", oldest.sizeBytes), + slog.Time("lastAccess", oldest.lastAccess), + ) + + entryDir := filepath.Join(c.basePath, oldest.digest) + if err := os.RemoveAll(entryDir); err != nil { + slog.WarnContext(ctx, "Failed to evict cache entry", slog.String("digest", oldest.digest), slog.Any("err", err)) + break + } + total -= oldest.sizeBytes + delete(c.entries, oldest.digest) + } +} + +// EvictLRU removes the least-recently-used cache entry and returns its digest +// and size, or ("", 0) if the cache is empty. This is exported for tests. +func (c *Cache) EvictLRU() (string, int64) { + c.mu.Lock() + defer c.mu.Unlock() + + var oldest *entryState + for _, e := range c.entries { + if oldest == nil || e.lastAccess.Before(oldest.lastAccess) { + oldest = e + } + } + if oldest == nil { + return "", 0 + } + + entryDir := filepath.Join(c.basePath, oldest.digest) + size := oldest.sizeBytes + _ = os.RemoveAll(entryDir) + delete(c.entries, oldest.digest) + return oldest.digest, size +} + +// Size returns the total size of all cached entries in bytes. +func (c *Cache) Size() int64 { + c.mu.Lock() + defer c.mu.Unlock() + var total int64 + for _, e := range c.entries { + total += e.sizeBytes + } + return total +} + +// Count returns the number of cached entries. +func (c *Cache) Count() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.entries) +} + +// --- Untar implementation ------------------------------------------------- + +// Untar extracts a tar stream into rootPath. It is a self-contained copy of +// the untar logic from cmd/atelet/oci.go, using os.OpenRoot for path-traversal +// safety. +func Untar(ctx context.Context, tarData io.Reader, rootPath string) error { + tracer := otel.Tracer("rootfscache") + _, span := tracer.Start(ctx, "Untar") + defer span.End() + + root, err := os.OpenRoot(rootPath) + if err != nil { + return fmt.Errorf("opening rootfs %q as os.Root: %w", rootPath, err) + } + defer root.Close() + + tarReader := tar.NewReader(tarData) + for { + hdr, err := tarReader.Next() + if errors.Is(err, io.EOF) { + break + } else if err != nil { + return fmt.Errorf("in tarReader.Next: %w", err) + } + + name, skip, err := ValidateTarName(hdr.Name) + if err != nil { + return fmt.Errorf("invalid tar entry: %w", err) + } + if skip { + continue + } + + mode := hdr.FileInfo().Mode().Perm() + + switch hdr.Typeflag { + case tar.TypeReg: + if _, err := root.Lstat(name); err == nil { + if err := root.RemoveAll(name); err != nil { + return fmt.Errorf("while replacing existing path at %q before regular file: %w", name, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("while checking existing path at %q before regular file: %w", name, err) + } + + outFile, err := root.OpenFile(name, os.O_CREATE|os.O_RDWR|os.O_TRUNC, mode) + if err != nil { + return fmt.Errorf("while creating file %q: %w", name, err) + } + _, err = io.Copy(outFile, tarReader) + closeErr := outFile.Close() + if err != nil { + return fmt.Errorf("while writing contents of %q from tar stream: %w", name, err) + } + if closeErr != nil { + return fmt.Errorf("while closing file %q: %w", name, closeErr) + } + + case tar.TypeDir: + err := root.Mkdir(name, mode) + if errors.Is(err, os.ErrExist) { + // Tolerate repeated directory entries. + } else if err != nil { + return fmt.Errorf("while creating directory=%q, mode=%v: %w", name, mode, err) + } + + case tar.TypeSymlink: + if existing, err := root.Lstat(name); err == nil { + if existing.Mode()&os.ModeSymlink != 0 { + if cur, rerr := root.Readlink(name); rerr == nil && cur == hdr.Linkname { + continue + } + } + if err := root.RemoveAll(name); err != nil { + return fmt.Errorf("while replacing existing path at %q before symlink: %w", name, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("while checking existing path at %q before symlink: %w", name, err) + } + if err := root.Symlink(hdr.Linkname, name); err != nil { + return fmt.Errorf("while creating symlink src=%q target=%q: %w", name, hdr.Linkname, err) + } + + case tar.TypeLink: + linkname, linkSkip, err := ValidateTarName(hdr.Linkname) + if err != nil { + return fmt.Errorf("invalid hardlink target for %q: %w", name, err) + } + if linkSkip { + return fmt.Errorf("invalid hardlink target for %q: empty", name) + } + if _, err := root.Lstat(name); err == nil { + if err := root.RemoveAll(name); err != nil { + return fmt.Errorf("while replacing existing path at %q before hardlink: %w", name, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("while checking existing path at %q before hardlink: %w", name, err) + } + if err := root.Link(linkname, name); err != nil { + return fmt.Errorf("while creating hardlink src=%q target=%q: %w", name, linkname, err) + } + + default: + tfStr := string([]byte{hdr.Typeflag}) + slog.ErrorContext(ctx, "Unhandled tar entry typeflag", slog.String("typeflag", tfStr), slog.Any("hdr", hdr)) + return fmt.Errorf("unhandled tar entry typeflag %q", tfStr) + } + } + + return nil +} + +// --- helpers -------------------------------------------------------------- + +// ValidateTarName cleans and validates a tar entry name. It is exported so +// the main package's tests can call it without importing the internals. +func ValidateTarName(name string) (cleaned string, skip bool, err error) { + if name == "" { + return "", true, nil + } + cleaned = filepath.Clean(name) + if cleaned == "." { + return "", true, nil + } + cleaned = strings.TrimPrefix(cleaned, "/") + if cleaned == "" || cleaned == "." { + return "", true, nil + } + if !filepath.IsLocal(cleaned) { + return "", false, fmt.Errorf("not a local path: %q", name) + } + return cleaned, false, nil +} + +func validateDigest(digest string) error { + if digest == "" { + return fmt.Errorf("digest must not be empty") + } + if strings.ContainsAny(digest, "/\\..") { + return fmt.Errorf("digest contains invalid characters: %q", digest) + } + return nil +} + +// dirSize returns the total size of all regular files under dir. +func dirSize(dir string) int64 { + var size int64 + _ = filepath.WalkDir(dir, func(_ string, d os.DirEntry, err error) error { + if err != nil { + return nil + } + if !d.IsDir() { + if info, err := d.Info(); err == nil { + size += info.Size() + } + } + return nil + }) + return size +} + +// readAccessTime reads the .last_access file or falls back to the directory +// mtime. +func readAccessTime(entryDir string) time.Time { + data, err := os.ReadFile(filepath.Join(entryDir, lastAccessFile)) + if err == nil { + s := strings.TrimSpace(string(data)) + if t, perr := time.Parse(time.RFC3339Nano, s); perr == nil { + return t + } + if t, perr := time.Parse(time.RFC3339, s); perr == nil { + return t + } + } + // Fallback: directory mtime. + if info, err := os.Stat(entryDir); err == nil { + return info.ModTime() + } + return time.Time{} +} + +// chmodRecursive sets the permission bits on every file and directory under +// root. Errors are silently ignored (best-effort). +func chmodRecursive(root string, perm os.FileMode) error { + return filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err != nil { + return nil + } + _ = os.Chmod(path, perm) + return nil + }) +} + +// SortedDigests returns the cached digests sorted by last-access time +// (oldest first). Useful for diagnostics and eviction. +func (c *Cache) SortedDigests() []string { + c.mu.Lock() + defer c.mu.Unlock() + + type entry struct { + digest string + lastAccess time.Time + } + var entries []entry + for _, e := range c.entries { + entries = append(entries, entry{e.digest, e.lastAccess}) + } + sort.Slice(entries, func(i, j int) bool { + return entries[i].lastAccess.Before(entries[j].lastAccess) + }) + out := make([]string, len(entries)) + for i, e := range entries { + out[i] = e.digest + } + return out +} diff --git a/cmd/atelet/internal/rootfscache/rootfscache_test.go b/cmd/atelet/internal/rootfscache/rootfscache_test.go new file mode 100644 index 000000000..bc346c825 --- /dev/null +++ b/cmd/atelet/internal/rootfscache/rootfscache_test.go @@ -0,0 +1,336 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rootfscache + +import ( + "archive/tar" + "bytes" + "context" + "os" + "path/filepath" + "sync" + "testing" +) + +const testDigest = "sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890" + +func buildTar(t *testing.T, entries []struct{ name, body string; typeflag byte; mode int64 }) []byte { + t.Helper() + var buf bytes.Buffer + tw := tar.NewWriter(&buf) + for _, e := range entries { + mode := e.mode + if mode == 0 { + if e.typeflag == tar.TypeDir { + mode = 0o755 + } else { + mode = 0o644 + } + } + hdr := &tar.Header{ + Name: e.name, + Typeflag: e.typeflag, + Mode: mode, + Size: int64(len(e.body)), + } + if err := tw.WriteHeader(hdr); err != nil { + t.Fatalf("tar.WriteHeader: %v", err) + } + if e.body != "" { + if _, err := tw.Write([]byte(e.body)); err != nil { + t.Fatalf("tar.Write: %v", err) + } + } + } + if err := tw.Close(); err != nil { + t.Fatalf("tar.Close: %v", err) + } + return buf.Bytes() +} + +func TestEnsureRootfs_CacheMiss(t *testing.T) { + base := t.TempDir() + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + + tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "etc/", typeflag: tar.TypeDir}, + {name: "etc/hostname", typeflag: tar.TypeReg, body: "test-host\n"}, + }) + + lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)) + if err != nil { + t.Fatalf("EnsureRootfs: %v", err) + } + if cached { + t.Errorf("expected cache miss, got hit") + } + if lowerDir == "" { + t.Fatalf("lowerDir is empty") + } + + // Verify the rootfs was extracted correctly. + data, err := os.ReadFile(filepath.Join(lowerDir, "etc/hostname")) + if err != nil { + t.Fatalf("read etc/hostname: %v", err) + } + if string(data) != "test-host\n" { + t.Errorf("etc/hostname = %q, want %q", data, "test-host\n") + } + + // Verify sentinel file exists. + readyPath := filepath.Join(base, testDigest, readySentinel) + if _, err := os.Stat(readyPath); err != nil { + t.Fatalf("ready sentinel missing: %v", err) + } + + if c.Count() != 1 { + t.Errorf("count = %d, want 1", c.Count()) + } +} + +func TestEnsureRootfs_CacheHit(t *testing.T) { + base := t.TempDir() + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + + tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "hello", typeflag: tar.TypeReg, body: "world"}, + }) + + // First call: cache miss. + if _, _, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)); err != nil { + t.Fatalf("EnsureRootfs (miss): %v", err) + } + + // Second call: cache hit. Pass nil reader — it must not be read. + lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, nil) + if err != nil { + t.Fatalf("EnsureRootfs (hit): %v", err) + } + if !cached { + t.Errorf("expected cache hit, got miss") + } + if lowerDir == "" { + t.Fatalf("lowerDir is empty on hit") + } + + // Verify content still accessible. + data, err := os.ReadFile(filepath.Join(lowerDir, "hello")) + if err != nil { + t.Fatalf("read hello: %v", err) + } + if string(data) != "world" { + t.Errorf("hello = %q, want %q", data, "world") + } +} + +func TestEnsureRootfs_ConcurrentMisses(t *testing.T) { + base := t.TempDir() + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + + tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "concurrent", typeflag: tar.TypeReg, body: "ok"}, + }) + + const goroutines = 10 + var wg sync.WaitGroup + errs := make([]error, goroutines) + lowerDirs := make([]string, goroutines) + cachedFlags := make([]bool, goroutines) + + for i := 0; i < goroutines; i++ { + i := i + wg.Add(1) + go func() { + defer wg.Done() + // Each goroutine gets its own reader over the same data. + lowerDirs[i], cachedFlags[i], errs[i] = c.EnsureRootfs( + context.Background(), testDigest, bytes.NewReader(tarData), + ) + }() + } + wg.Wait() + + // At least one goroutine should have done the extraction (miss). + // All should succeed with the same lowerDir. + anyMiss := false + for i := 0; i < goroutines; i++ { + if errs[i] != nil { + t.Errorf("goroutine %d: %v", i, errs[i]) + continue + } + if !cachedFlags[i] { + anyMiss = true + } + if lowerDirs[i] == "" { + t.Errorf("goroutine %d: empty lowerDir", i) + } + if i > 0 && lowerDirs[i] != lowerDirs[0] { + t.Errorf("goroutine %d: lowerDir %q != goroutine 0 lowerDir %q", i, lowerDirs[i], lowerDirs[0]) + } + } + if !anyMiss { + t.Errorf("expected at least one cache miss among %d goroutines", goroutines) + } + + // Only one cache entry should exist. + if c.Count() != 1 { + t.Errorf("count = %d, want 1", c.Count()) + } +} + +func TestEnsureRootfs_PartialEntryCleanup(t *testing.T) { + base := t.TempDir() + // Simulate a crash: create the digest directory but no .ready sentinel. + partialDir := filepath.Join(base, testDigest, "lower") + if err := os.MkdirAll(partialDir, 0o700); err != nil { + t.Fatalf("mkdir partial: %v", err) + } + if err := os.WriteFile(filepath.Join(partialDir, "stale"), []byte("data"), 0o644); err != nil { + t.Fatalf("write stale: %v", err) + } + + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + // The partial entry should have been cleaned up during loadIndex. + if c.Count() != 0 { + t.Errorf("count = %d, want 0 (partial entry should be removed)", c.Count()) + } + + // Now a fresh extraction should succeed. + tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "fresh", typeflag: tar.TypeReg, body: "data"}, + }) + lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)) + if err != nil { + t.Fatalf("EnsureRootfs: %v", err) + } + if cached { + t.Errorf("expected miss after cleanup, got hit") + } + if _, err := os.Stat(filepath.Join(lowerDir, "fresh")); err != nil { + t.Errorf("fresh file missing: %v", err) + } +} + +func TestEvictLRU(t *testing.T) { + base := t.TempDir() + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + + digest1 := "sha256:1111111111111111111111111111111111111111111111111111111111111111" + digest2 := "sha256:2222222222222222222222222222222222222222222222222222222222222222" + + tarData1 := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "d1", typeflag: tar.TypeReg, body: "data1"}, + }) + tarData2 := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "d2", typeflag: tar.TypeReg, body: "data2"}, + }) + + if _, _, err := c.EnsureRootfs(context.Background(), digest1, bytes.NewReader(tarData1)); err != nil { + t.Fatalf("EnsureRootfs d1: %v", err) + } + if _, _, err := c.EnsureRootfs(context.Background(), digest2, bytes.NewReader(tarData2)); err != nil { + t.Fatalf("EnsureRootfs d2: %v", err) + } + + if c.Count() != 2 { + t.Fatalf("count = %d, want 2", c.Count()) + } + + // Evict the oldest (digest1, loaded first). + evicted, size := c.EvictLRU() + if evicted != digest1 { + t.Errorf("evicted = %q, want %q", evicted, digest1) + } + if size <= 0 { + t.Errorf("evicted size = %d, want > 0", size) + } + if c.Count() != 1 { + t.Errorf("count = %d, want 1 after eviction", c.Count()) + } + + // The evicted directory should be gone. + if _, err := os.Stat(filepath.Join(base, digest1)); !os.IsNotExist(err) { + t.Errorf("evicted dir still exists: %v", err) + } +} + +func TestLowerDir(t *testing.T) { + base := t.TempDir() + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + + // Before caching. + if got := c.LowerDir(testDigest); got != "" { + t.Errorf("LowerDir before cache = %q, want empty", got) + } + + tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + {name: ".", typeflag: tar.TypeDir}, + }) + if _, _, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)); err != nil { + t.Fatalf("EnsureRootfs: %v", err) + } + + // After caching. + got := c.LowerDir(testDigest) + if got == "" { + t.Fatalf("LowerDir after cache is empty") + } + if got != filepath.Join(base, testDigest, "lower") { + t.Errorf("LowerDir = %q, want %q", got, filepath.Join(base, testDigest, "lower")) + } +} + +func TestValidateDigest(t *testing.T) { + tests := []struct { + input string + want bool + }{ + {"sha256:abc123", false}, + {"", true}, + {"../escape", true}, + {"sha256:abc/def", true}, + {"sha256:abc..def", true}, + } + for _, tc := range tests { + err := validateDigest(tc.input) + if (err != nil) != tc.want { + t.Errorf("validateDigest(%q) err=%v, wantErr=%v", tc.input, err, tc.want) + } + } +} diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index d01b2db04..e7fa8a80b 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -31,6 +31,7 @@ import ( "cloud.google.com/go/storage" "github.com/agent-substrate/substrate/cmd/atelet/internal/ategcs" "github.com/agent-substrate/substrate/cmd/atelet/internal/memorypullcache" + "github.com/agent-substrate/substrate/cmd/atelet/internal/rootfscache" "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/proto/ateletpb" @@ -115,6 +116,11 @@ func main() { serverboot.Fatal(ctx, "Failed to create pull cache", err) } + rootfsDiskCache, err := rootfscache.New(ctx, ateompath.RootfsCacheDir, 0) + if err != nil { + serverboot.Fatal(ctx, "Failed to create rootfs cache", err) + } + anonGCSClient, err := storage.NewClient(ctx, option.WithoutAuthentication()) if err != nil { serverboot.Fatal(ctx, "Failed to create anonymous GCS client", err) @@ -163,6 +169,7 @@ func main() { wrappedAnonGCS, wrappedGCS, pullCache, + rootfsDiskCache, ) lis, err := net.Listen("tcp", ":"+strconv.Itoa(*port)) @@ -186,6 +193,7 @@ type AteomHerder struct { ateomDialer *AteomDialer pullCache *memorypullcache.MemoryPullCache + rootfsCache *rootfscache.Cache anonGCSClient ategcs.ObjectStorage gcsClient ategcs.ObjectStorage } @@ -199,10 +207,12 @@ func NewService( anonGCSClient ategcs.ObjectStorage, gcsClient ategcs.ObjectStorage, pullCache *memorypullcache.MemoryPullCache, + rootfsCache *rootfscache.Cache, ) *AteomHerder { wms := &AteomHerder{ ateomDialer: ateomDialer, pullCache: pullCache, + rootfsCache: rootfsCache, anonGCSClient: anonGCSClient, gcsClient: gcsClient, } @@ -682,6 +692,7 @@ func (s *AteomHerder) prepareOCIBundles( if err := prepareOCIDirectory( gCtx, s.pullCache, + s.rootfsCache, actorTemplateNamespace, actorTemplateName, actorID, "pause", spec.GetPauseImage(), @@ -714,6 +725,7 @@ func (s *AteomHerder) prepareOCIBundles( if err := prepareOCIDirectory( gCtx, s.pullCache, + s.rootfsCache, actorTemplateNamespace, actorTemplateName, actorID, ctr.GetName(), ctr.GetImage(), @@ -948,6 +960,19 @@ func resetActorDirs(actorTemplateNamespace, actorTemplateName, actorID string) e // Explicitly leave runsc logs dir untouched. bundleDir := ateompath.OCIBundleDir(actorTemplateNamespace, actorTemplateName, actorID) + + // Unmount any overlayfs rootfs mounts before deleting the bundle + // directory. Each container's rootfs/ may be an overlayfs mountpoint; + // unmounting with MNT_DETACH ensures the mount is released even if + // something still holds a reference (e.g. a lingering process). + if entries, err := os.ReadDir(bundleDir); err == nil { + for _, e := range entries { + if e.IsDir() { + unmountActorRootfs(bundleDir, e.Name()) + } + } + } + if err := os.RemoveAll(bundleDir); err != nil { return fmt.Errorf("while deleting bundle dir: %w", err) } diff --git a/cmd/atelet/oci.go b/cmd/atelet/oci.go index a00df99e9..7487253ee 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -15,19 +15,17 @@ package main import ( - "archive/tar" "context" "encoding/json" - "errors" "fmt" "io" "log/slog" "os" "path" - "path/filepath" "strings" "github.com/agent-substrate/substrate/cmd/atelet/internal/memorypullcache" + "github.com/agent-substrate/substrate/cmd/atelet/internal/rootfscache" "github.com/agent-substrate/substrate/internal/ateompath" "github.com/opencontainers/runtime-spec/specs-go" "go.opentelemetry.io/otel" @@ -53,7 +51,12 @@ const ( ActorIDFileName = "actor-id" ) -func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryPullCache, actorTemplateNamespace, actorTemplateName, actorID, containerName, ref string, args []string, env []string, annotations map[string]string, netns string, identityDir string, durableDirVolumeMounts []*ateletpb.VolumeMount) error { +// prepareOCIDirectory assembles the OCI bundle for one container inside an +// actor. When a rootfsCache is available and the image ref contains a digest, +// the rootfs is materialized via an overlayfs mount over a node-local cache +// instead of re-extracting the tarball — reducing per-restore latency from +// seconds to sub-millisecond on cache hits. +func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryPullCache, rootfsCache *rootfscache.Cache, actorTemplateNamespace, actorTemplateName, actorID, containerName, ref string, args []string, env []string, annotations map[string]string, netns string, identityDir string, durableDirVolumeMounts []*ateletpb.VolumeMount) error { tracer := otel.Tracer("prepareOCIDirectory") ctx, span := tracer.Start(ctx, "prepareOCIDirectory") @@ -63,22 +66,56 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP bundlePath := ateompath.OCIBundlePath(actorTemplateNamespace, actorTemplateName, actorID, containerName) rootPath := path.Join(bundlePath, "rootfs") - if err := os.RemoveAll(rootPath); err != nil { - return fmt.Errorf("while clearing rootfs %q: %w", rootPath, err) - } + // Try the overlayfs cache path first. This succeeds when: + // 1. rootfsCache is non-nil, AND + // 2. the image ref includes a digest (@sha256:…). + // On a cache hit, tarData is NOT consumed, so we can skip the untar + // entirely. On a miss, the cache extracts and caches for next time. + digest := extractDigestFromRef(ref) + if rootfsCache != nil && digest != "" { + tarData, err := pullCache.Fetch(ctx, ref) + if err != nil { + return fmt.Errorf("in pullCache.Fetch: %w", err) + } + defer tarData.Close() - if err := os.MkdirAll(rootPath, 0o700); err != nil { - return fmt.Errorf("in os.MkdirAll for container bundle dir: %w", err) - } + lowerDir, _, err := rootfsCache.EnsureRootfs(ctx, digest, tarData) + if err != nil { + return fmt.Errorf("in rootfsCache.EnsureRootfs: %w", err) + } - tarData, err := pullCache.Fetch(ctx, ref) - if err != nil { - return fmt.Errorf("in pullCache.Fetch: %w", err) - } - defer tarData.Close() + // Create the overlay mount target. + if err := os.MkdirAll(rootPath, 0o700); err != nil { + return fmt.Errorf("in os.MkdirAll for rootfs mount target: %w", err) + } + + upperDir := path.Join(bundlePath, "upper") + workDir := path.Join(bundlePath, "work") + if err := setupOverlayfs(rootPath, lowerDir, upperDir, workDir); err != nil { + return fmt.Errorf("setting up overlayfs (lower=%s, target=%s): %w", lowerDir, rootPath, err) + } + + span.SetAttributes(attribute.String("rootfs_method", "overlay")) + } else { + // Fallback: no digest or no cache — extract directly (original path). + if err := os.RemoveAll(rootPath); err != nil { + return fmt.Errorf("while clearing rootfs %q: %w", rootPath, err) + } + if err := os.MkdirAll(rootPath, 0o700); err != nil { + return fmt.Errorf("in os.MkdirAll for container bundle dir: %w", err) + } - if err := untar(ctx, tarData, rootPath); err != nil { - return fmt.Errorf("in untar: %w", err) + tarData, err := pullCache.Fetch(ctx, ref) + if err != nil { + return fmt.Errorf("in pullCache.Fetch: %w", err) + } + defer tarData.Close() + + if err := rootfscache.Untar(ctx, tarData, rootPath); err != nil { + return fmt.Errorf("in untar: %w", err) + } + + span.SetAttributes(attribute.String("rootfs_method", "untar")) } // Bind-mount the per-actor identity directory so the workload can read its @@ -103,6 +140,19 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP return nil } +// extractDigestFromRef extracts the sha256 digest from an image reference. +// Returns "" if the ref does not contain a digest. +// - "registry/image@sha256:abc123" → "sha256:abc123" +// - "registry/image:latest" → "" +func extractDigestFromRef(ref string) string { + const prefix = "@sha256:" + idx := strings.LastIndex(ref, prefix) + if idx < 0 { + return "" + } + return strings.TrimPrefix(ref[idx:], "@") +} + // buildActorOCISpec assembles the OCI runtime spec for an actor container. // When identityDir is non-empty it adds a read-only bind mount of that host // directory at IdentityMountPath so the actor can read its own ID (see @@ -250,148 +300,27 @@ func createMountPoint(rootPath, mountPath string) error { return nil } -func validateTarName(name string) (cleaned string, skip bool, err error) { - if name == "" { - return "", true, nil - } - cleaned = filepath.Clean(name) - if cleaned == "." { - return "", true, nil +// unmountActorRootfs attempts to unmount the rootfs overlay for a single +// container inside an actor's bundle directory. Returns nil if the rootfs +// is not a mountpoint (i.e. was produced by direct untar). +func unmountActorRootfs(bundleDir, containerName string) error { + rootfsPath := path.Join(bundleDir, containerName, "rootfs") + if err := teardownOverlayfs(rootfsPath); err != nil { + // ENOTDIR/ENOENT/EINVAL — not a mountpoint, nothing to do. + slog.Debug("rootfs unmount skipped (not a mountpoint)", "path", rootfsPath, "err", err) } - cleaned = strings.TrimPrefix(cleaned, "/") - if cleaned == "" || cleaned == "." { - return "", true, nil - } - if !filepath.IsLocal(cleaned) { - return "", false, fmt.Errorf("not a local path: %q", name) - } - return cleaned, false, nil + return nil } +// untar is a thin wrapper around rootfscache.Untar kept in this package so +// that existing tests (package main) continue to compile without importing +// the rootfscache package directly. func untar(ctx context.Context, tarData io.Reader, rootPath string) error { - tracer := otel.Tracer("ateom-gvisor") - ctx, span := tracer.Start(ctx, "untar") - defer span.End() - - // os.Root confines file operations to rootPath: ".." components and - // out-of-tree symlinks are refused by the kernel. - root, err := os.OpenRoot(rootPath) - if err != nil { - return fmt.Errorf("while opening rootfs %q as os.Root: %w", rootPath, err) - } - defer root.Close() - - tarReader := tar.NewReader(tarData) - for { - hdr, err := tarReader.Next() - if errors.Is(err, io.EOF) { - break - } else if err != nil { - return fmt.Errorf("in tarReader.Next: %w", err) - } - - name, skip, err := validateTarName(hdr.Name) - if err != nil { - return fmt.Errorf("invalid tar entry: %w", err) - } - if skip { - continue - } - - mode := hdr.FileInfo().Mode().Perm() - - switch hdr.Typeflag { - case tar.TypeReg: // Regular file - // Same "later entry wins" handling: if any entry exists at the target path, - // remove it first. This ensures that: - // 1. If it's a symlink, we don't write through it (security vulnerability / incorrectness). - // 2. If it's a hardlink, we unlink it instead of truncating the shared inode. - // 3. If it's a directory, we recursively remove it so we can write the file. - if _, err := root.Lstat(name); err == nil { - if err := root.RemoveAll(name); err != nil { - return fmt.Errorf("while replacing existing path at %q before regular file: %w", name, err) - } - } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("while checking existing path at %q before regular file: %w", name, err) - } - - // Stream directly from tarReader to target file to avoid buffering in memory. - outFile, err := root.OpenFile(name, os.O_CREATE|os.O_RDWR|os.O_TRUNC, mode) - if err != nil { - return fmt.Errorf("while creating file %q: %w", name, err) - } - - _, err = io.Copy(outFile, tarReader) - closeErr := outFile.Close() - - if err != nil { - return fmt.Errorf("while writing contents of %q from tar stream: %w", name, err) - } - if closeErr != nil { - return fmt.Errorf("while closing file %q: %w", name, closeErr) - } - - case tar.TypeDir: - err := root.Mkdir(name, mode) - if errors.Is(err, os.ErrExist) { - // Ignore --- real images produced by ko seem to have directory entries placed multiple times? - } else if err != nil { - return fmt.Errorf("while creating directory=%q, mode=%v: %w", name, mode, err) - } - - case tar.TypeSymlink: - // OCI image layers may re-define the same path across layers (e.g. - // an earlier layer creates /var/run as a directory and a later - // layer re-declares it as a symlink to /run). Standard tar-extract - // semantics are "later entry wins": replace any existing entry. - if existing, err := root.Lstat(name); err == nil { - // If it's already the same symlink, skip the unlink+symlink pair. - if existing.Mode()&os.ModeSymlink != 0 { - if cur, rerr := root.Readlink(name); rerr == nil && cur == hdr.Linkname { - continue - } - } - // Root.RemoveAll removes the symlink entry itself; it does NOT - // traverse and remove the directory the symlink points to. - // That's the desired semantic here — replace this path's - // entry without touching whatever the prior symlink targeted. - if err := root.RemoveAll(name); err != nil { - return fmt.Errorf("while replacing existing path at %q before symlink: %w", name, err) - } - } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("while checking existing path at %q before symlink: %w", name, err) - } - if err := root.Symlink(hdr.Linkname, name); err != nil { - return fmt.Errorf("while creating symlink src=%q target=%q: %w", name, hdr.Linkname, err) - } - - case tar.TypeLink: - linkname, linkSkip, err := validateTarName(hdr.Linkname) - if err != nil { - return fmt.Errorf("invalid hardlink target for %q: %w", name, err) - } - if linkSkip { - return fmt.Errorf("invalid hardlink target for %q: empty", name) - } - // Same "later entry wins" handling as TypeSymlink: replace existing entry. - if _, err := root.Lstat(name); err == nil { - if err := root.RemoveAll(name); err != nil { - return fmt.Errorf("while replacing existing path at %q before hardlink: %w", name, err) - } - } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("while checking existing path at %q before hardlink: %w", name, err) - } - if err := root.Link(linkname, name); err != nil { - return fmt.Errorf("while creating hardlink src=%q target=%q: %w", name, linkname, err) - } - - default: - tfStr := string([]byte{hdr.Typeflag}) - slog.ErrorContext(ctx, "Unhandled tar entry typeflag", slog.String("typeflag", tfStr), slog.Any("hdr", hdr)) - return fmt.Errorf("unhandled tar entry typeflag %q", tfStr) - } - - } + return rootfscache.Untar(ctx, tarData, rootPath) +} - return nil +// validateTarName is re-exported here for the same reason as untar: tests in +// package main call it directly. +func validateTarName(name string) (cleaned string, skip bool, err error) { + return rootfscache.ValidateTarName(name) } diff --git a/cmd/atelet/overlay.go b/cmd/atelet/overlay.go new file mode 100644 index 000000000..bc2a85479 --- /dev/null +++ b/cmd/atelet/overlay.go @@ -0,0 +1,87 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "fmt" + "os" + + "golang.org/x/sys/unix" +) + +// setupOverlayfs mounts an overlayfs at target using the given layers. +// - lowerDir: read-only shared rootfs from the node-level cache +// - upperDir: per-actor writable layer (created if absent) +// - workDir: overlayfs work directory (created if absent) +// +// The target directory must already exist (it is the bundle's rootfs/ dir). +func setupOverlayfs(target, lowerDir, upperDir, workDir string) error { + for _, d := range []string{upperDir, workDir} { + if err := os.MkdirAll(d, 0o700); err != nil { + return fmt.Errorf("creating overlay dir %s: %w", d, err) + } + } + + opts := fmt.Sprintf("lowerdir=%s,upperdir=%s,workdir=%s", lowerDir, upperDir, workDir) + if err := unix.Mount("overlay", target, "overlay", 0, opts); err != nil { + return fmt.Errorf("mounting overlayfs at %s: %w", target, err) + } + return nil +} + +// teardownOverlayfs unmounts the overlayfs at target. The caller is +// responsible for removing the upper/work directories (typically done by +// resetActorDirs). +func teardownOverlayfs(target string) error { + if err := unix.Unmount(target, unix.MNT_DETACH); err != nil { + return fmt.Errorf("unmounting overlayfs at %s: %w", target, err) + } + return nil +} + +// isOverlayfsAvailable checks whether the overlayfs kernel module is available +// by attempting a mount on a temporary directory. +func isOverlayfsAvailable() bool { + tmpLower, err := os.MkdirTemp("", "overlay-check-lower-") + if err != nil { + return false + } + defer os.RemoveAll(tmpLower) + + tmpUpper, err := os.MkdirTemp("", "overlay-check-upper-") + if err != nil { + return false + } + defer os.RemoveAll(tmpUpper) + + tmpWork, err := os.MkdirTemp("", "overlay-check-work-") + if err != nil { + return false + } + defer os.RemoveAll(tmpWork) + + tmpTarget, err := os.MkdirTemp("", "overlay-check-target-") + if err != nil { + return false + } + defer os.RemoveAll(tmpTarget) + + opts := fmt.Sprintf("lowerdir=%s,upperdir=%s,workdir=%s", tmpLower, tmpUpper, tmpWork) + if err := unix.Mount("overlay", tmpTarget, "overlay", 0, opts); err != nil { + return false + } + _ = unix.Unmount(tmpTarget, 0) + return true +} diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index d9beb2d6c..1dc8eb3c9 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -28,8 +28,21 @@ const ( var ( // StaticFilesDir holds things like downloaded runsc binaries. StaticFilesDir = filepath.Join(BasePath, "static-files") + + // RootfsCacheDir is the node-level directory that holds extracted, + // read-only rootfs directories keyed by image digest. On a cache hit the + // per-actor rootfs is materialized as an overlayfs mount instead of + // re-extracting the tarball. + RootfsCacheDir = filepath.Join(BasePath, "rootfs-cache") ) +// RootfsCacheLowerDir returns the read-only rootfs directory for a given image +// digest inside the node-level cache. This is the "lowerdir" for an overlayfs +// mount. +func RootfsCacheLowerDir(digest string) string { + return filepath.Join(RootfsCacheDir, digest, "lower") +} + func RunSCBinaryPath(sha256 string) string { return filepath.Join(StaticFilesDir, "runsc-"+sha256) } From db0e31be1244720c592e24b680e76f1835b7656b Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Fri, 3 Jul 2026 12:18:54 +0800 Subject: [PATCH 2/9] perf(atelet): make rootfs cache skip image pull on hit EnsureRootfs took an io.Reader that oci.go always populated via pullCache.Fetch before the call, so every resume pulled and extracted the image into the memory pull cache even when the on-disk rootfs cache already had it. Change the parameter to a lazy tar provider closure that EnsureRootfs invokes only on a cache miss; on a hit it returns the on-disk lowerDir without pulling or extracting anything. --- .../internal/rootfscache/rootfscache.go | 36 +++++++++++++++---- .../internal/rootfscache/rootfscache_test.go | 23 ++++++++---- cmd/atelet/oci.go | 14 ++++---- 3 files changed, 52 insertions(+), 21 deletions(-) diff --git a/cmd/atelet/internal/rootfscache/rootfscache.go b/cmd/atelet/internal/rootfscache/rootfscache.go index ef7163603..38a3ccd25 100644 --- a/cmd/atelet/internal/rootfscache/rootfscache.go +++ b/cmd/atelet/internal/rootfscache/rootfscache.go @@ -137,17 +137,23 @@ func (c *Cache) LowerDir(digest string) string { } // EnsureRootfs guarantees that the rootfs for digest is extracted into the -// cache. On a cache hit it returns the lowerDir immediately without reading -// tarData. On a miss it consumes tarData to populate the cache. +// cache. On a cache hit it returns the lowerDir immediately WITHOUT invoking +// tarProvider — so the caller pays no image pull/extract cost when the rootfs +// is already on disk. On a miss it calls tarProvider to obtain the extracted +// image tar stream and consumes it to populate the cache. +// +// tarProvider is a lazy source of the image's extracted rootfs tar (e.g. a +// closure over MemoryPullCache.Fetch). It is invoked at most once, and only on +// a cache miss; the returned reader is closed by EnsureRootfs. // // Returns: // // lowerDir – the read-only rootfs path (non-empty on success) -// cached – true if the cache was hit (tarData was NOT consumed) +// cached – true if the cache was hit (tarProvider was NOT invoked) // err – any error // // The digest MUST be a valid directory name (hex-encoded sha256, no slashes). -func (c *Cache) EnsureRootfs(ctx context.Context, digest string, tarData io.Reader) (string, bool, error) { +func (c *Cache) EnsureRootfs(ctx context.Context, digest string, tarProvider func() (io.ReadCloser, error)) (string, bool, error) { tracer := otel.Tracer("rootfscache") ctx, span := tracer.Start(ctx, "EnsureRootfs") span.SetAttributes(attribute.String("digest", digest)) @@ -157,7 +163,8 @@ func (c *Cache) EnsureRootfs(ctx context.Context, digest string, tarData io.Read return "", false, err } - // Fast path: already cached. + // Fast path: already cached. tarProvider is never invoked, so on a hit the + // caller neither pulls nor extracts the image. c.mu.Lock() if e, ok := c.entries[digest]; ok { c.mu.Unlock() @@ -184,8 +191,9 @@ func (c *Cache) EnsureRootfs(ctx context.Context, digest string, tarData io.Read c.inflight[digest] = infl c.mu.Unlock() - // Do the actual extraction outside the lock. - lowerDir, err := c.extract(ctx, digest, tarData) + // Cache miss: obtain the tar stream lazily (only now that we know we need + // it) and extract, outside the lock. + lowerDir, err := c.fetchAndExtract(ctx, digest, tarProvider) c.mu.Lock() delete(c.inflight, digest) @@ -209,6 +217,20 @@ func (c *Cache) EnsureRootfs(ctx context.Context, digest string, tarData io.Read return lowerDir, false, nil } +// fetchAndExtract invokes tarProvider to obtain the image tar stream and +// extracts it into the cache. It is only called on a cache miss. +func (c *Cache) fetchAndExtract(ctx context.Context, digest string, tarProvider func() (io.ReadCloser, error)) (string, error) { + if tarProvider == nil { + return "", fmt.Errorf("rootfs cache miss for digest %q but no tar provider supplied", digest) + } + tarData, err := tarProvider() + if err != nil { + return "", fmt.Errorf("obtaining image tar for digest %q: %w", digest, err) + } + defer tarData.Close() + return c.extract(ctx, digest, tarData) +} + // extract untars tarData into the cache directory for digest, writes the // .ready sentinel and .last_access file, and returns the lowerDir path. func (c *Cache) extract(ctx context.Context, digest string, tarData io.Reader) (string, error) { diff --git a/cmd/atelet/internal/rootfscache/rootfscache_test.go b/cmd/atelet/internal/rootfscache/rootfscache_test.go index bc346c825..ba653fe63 100644 --- a/cmd/atelet/internal/rootfscache/rootfscache_test.go +++ b/cmd/atelet/internal/rootfscache/rootfscache_test.go @@ -18,6 +18,7 @@ import ( "archive/tar" "bytes" "context" + "io" "os" "path/filepath" "sync" @@ -60,6 +61,14 @@ func buildTar(t *testing.T, entries []struct{ name, body string; typeflag byte; return buf.Bytes() } +// tarProviderFor returns an EnsureRootfs tar provider that yields a fresh +// reader over data on each call. +func tarProviderFor(data []byte) func() (io.ReadCloser, error) { + return func() (io.ReadCloser, error) { + return io.NopCloser(bytes.NewReader(data)), nil + } +} + func TestEnsureRootfs_CacheMiss(t *testing.T) { base := t.TempDir() c, err := New(context.Background(), base, 0) @@ -73,7 +82,7 @@ func TestEnsureRootfs_CacheMiss(t *testing.T) { {name: "etc/hostname", typeflag: tar.TypeReg, body: "test-host\n"}, }) - lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)) + lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, tarProviderFor(tarData)) if err != nil { t.Fatalf("EnsureRootfs: %v", err) } @@ -117,7 +126,7 @@ func TestEnsureRootfs_CacheHit(t *testing.T) { }) // First call: cache miss. - if _, _, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)); err != nil { + if _, _, err := c.EnsureRootfs(context.Background(), testDigest, tarProviderFor(tarData)); err != nil { t.Fatalf("EnsureRootfs (miss): %v", err) } @@ -168,7 +177,7 @@ func TestEnsureRootfs_ConcurrentMisses(t *testing.T) { defer wg.Done() // Each goroutine gets its own reader over the same data. lowerDirs[i], cachedFlags[i], errs[i] = c.EnsureRootfs( - context.Background(), testDigest, bytes.NewReader(tarData), + context.Background(), testDigest, tarProviderFor(tarData), ) }() } @@ -227,7 +236,7 @@ func TestEnsureRootfs_PartialEntryCleanup(t *testing.T) { {name: ".", typeflag: tar.TypeDir}, {name: "fresh", typeflag: tar.TypeReg, body: "data"}, }) - lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)) + lowerDir, cached, err := c.EnsureRootfs(context.Background(), testDigest, tarProviderFor(tarData)) if err != nil { t.Fatalf("EnsureRootfs: %v", err) } @@ -258,10 +267,10 @@ func TestEvictLRU(t *testing.T) { {name: "d2", typeflag: tar.TypeReg, body: "data2"}, }) - if _, _, err := c.EnsureRootfs(context.Background(), digest1, bytes.NewReader(tarData1)); err != nil { + if _, _, err := c.EnsureRootfs(context.Background(), digest1, tarProviderFor(tarData1)); err != nil { t.Fatalf("EnsureRootfs d1: %v", err) } - if _, _, err := c.EnsureRootfs(context.Background(), digest2, bytes.NewReader(tarData2)); err != nil { + if _, _, err := c.EnsureRootfs(context.Background(), digest2, tarProviderFor(tarData2)); err != nil { t.Fatalf("EnsureRootfs d2: %v", err) } @@ -302,7 +311,7 @@ func TestLowerDir(t *testing.T) { tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ {name: ".", typeflag: tar.TypeDir}, }) - if _, _, err := c.EnsureRootfs(context.Background(), testDigest, bytes.NewReader(tarData)); err != nil { + if _, _, err := c.EnsureRootfs(context.Background(), testDigest, tarProviderFor(tarData)); err != nil { t.Fatalf("EnsureRootfs: %v", err) } diff --git a/cmd/atelet/oci.go b/cmd/atelet/oci.go index 7487253ee..9bb63e11d 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -73,13 +73,13 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP // entirely. On a miss, the cache extracts and caches for next time. digest := extractDigestFromRef(ref) if rootfsCache != nil && digest != "" { - tarData, err := pullCache.Fetch(ctx, ref) - if err != nil { - return fmt.Errorf("in pullCache.Fetch: %w", err) - } - defer tarData.Close() - - lowerDir, _, err := rootfsCache.EnsureRootfs(ctx, digest, tarData) + // Pass Fetch as a lazy provider: on a cache hit EnsureRootfs returns the + // on-disk lowerDir without invoking it, so we do NOT pull or extract the + // image (and never buffer it in the memory pull cache) on the hot path. + // The pull only happens on a genuine cache miss. + lowerDir, _, err := rootfsCache.EnsureRootfs(ctx, digest, func() (io.ReadCloser, error) { + return pullCache.Fetch(ctx, ref) + }) if err != nil { return fmt.Errorf("in rootfsCache.EnsureRootfs: %w", err) } From 97ff534d7e8aa516d714f551a94a97c88f4f3dac Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 15:26:19 +0800 Subject: [PATCH 3/9] refactor: move overlay rootfs mount from atelet to privileged ateom atelet runs with all capabilities dropped and can no longer call mount(2), so it cannot perform the overlayfs rootfs mount itself. On a rootfs-cache hit atelet now writes a per-bundle "overlay-lower" marker recording the read-only lowerdir; the privileged ateom worker reads the marker and mounts the overlay just before `runsc create`, in ateom's mount namespace (shared by the runsc child, invisible to atelet). - ateompath: add ContainerRootfsDir/OverlayUpperDir/OverlayWorkDir/ OverlayLowerMarkerFile so both sides agree on paths by convention. - atelet: write the marker instead of mounting; drop unmountActorRootfs and the reset-time unmount loop (mounts live in ateom's ns). - ateom: mount overlays in Run/Restore (with failure-path unmount) and unmount in Checkpoint so a long-lived ateom does not leak mounts. --- cmd/atelet/main.go | 18 ++-- cmd/atelet/oci.go | 29 +++---- cmd/atelet/overlay.go | 87 ------------------- cmd/ateom-gvisor/main.go | 33 ++++++++ cmd/ateom-gvisor/overlay.go | 122 +++++++++++++++++++++++++++ internal/ateompath/ateompath.go | 40 +++++++++ internal/ateompath/ateompath_test.go | 36 ++++++++ 7 files changed, 247 insertions(+), 118 deletions(-) delete mode 100644 cmd/atelet/overlay.go create mode 100644 cmd/ateom-gvisor/overlay.go diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index e7fa8a80b..0c3254b13 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -961,18 +961,12 @@ func resetActorDirs(actorTemplateNamespace, actorTemplateName, actorID string) e bundleDir := ateompath.OCIBundleDir(actorTemplateNamespace, actorTemplateName, actorID) - // Unmount any overlayfs rootfs mounts before deleting the bundle - // directory. Each container's rootfs/ may be an overlayfs mountpoint; - // unmounting with MNT_DETACH ensures the mount is released even if - // something still holds a reference (e.g. a lingering process). - if entries, err := os.ReadDir(bundleDir); err == nil { - for _, e := range entries { - if e.IsDir() { - unmountActorRootfs(bundleDir, e.Name()) - } - } - } - + // Any overlayfs rootfs mounts live in the privileged ateom worker's mount + // namespace (atelet cannot mount(2) after its capabilities were dropped), + // so they are not visible here and are torn down by ateom on + // checkpoint/reset. atelet only needs to remove the on-disk bundle tree; + // with hostPath propagation=None, removing the (host-view) mountpoint dir + // does not disturb ateom's mount. if err := os.RemoveAll(bundleDir); err != nil { return fmt.Errorf("while deleting bundle dir: %w", err) } diff --git a/cmd/atelet/oci.go b/cmd/atelet/oci.go index 9bb63e11d..845a22ff9 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -19,7 +19,6 @@ import ( "encoding/json" "fmt" "io" - "log/slog" "os" "path" "strings" @@ -84,15 +83,19 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP return fmt.Errorf("in rootfsCache.EnsureRootfs: %w", err) } - // Create the overlay mount target. + // Create the overlay mount target. The actual overlayfs mount is + // performed by the privileged ateom worker just before `runsc create`, + // because atelet runs with all capabilities dropped and cannot call + // mount(2). We record the read-only lowerdir in a per-bundle marker file + // that ateom reads; its presence is ateom's signal to mount an overlay + // (upperdir/workdir/target are derived from the bundle path by + // convention on both sides). if err := os.MkdirAll(rootPath, 0o700); err != nil { return fmt.Errorf("in os.MkdirAll for rootfs mount target: %w", err) } - - upperDir := path.Join(bundlePath, "upper") - workDir := path.Join(bundlePath, "work") - if err := setupOverlayfs(rootPath, lowerDir, upperDir, workDir); err != nil { - return fmt.Errorf("setting up overlayfs (lower=%s, target=%s): %w", lowerDir, rootPath, err) + markerPath := ateompath.OverlayLowerMarkerFile(actorTemplateNamespace, actorTemplateName, actorID, containerName) + if err := os.WriteFile(markerPath, []byte(lowerDir), 0o600); err != nil { + return fmt.Errorf("while writing overlay lower marker: %w", err) } span.SetAttributes(attribute.String("rootfs_method", "overlay")) @@ -300,18 +303,6 @@ func createMountPoint(rootPath, mountPath string) error { return nil } -// unmountActorRootfs attempts to unmount the rootfs overlay for a single -// container inside an actor's bundle directory. Returns nil if the rootfs -// is not a mountpoint (i.e. was produced by direct untar). -func unmountActorRootfs(bundleDir, containerName string) error { - rootfsPath := path.Join(bundleDir, containerName, "rootfs") - if err := teardownOverlayfs(rootfsPath); err != nil { - // ENOTDIR/ENOENT/EINVAL — not a mountpoint, nothing to do. - slog.Debug("rootfs unmount skipped (not a mountpoint)", "path", rootfsPath, "err", err) - } - return nil -} - // untar is a thin wrapper around rootfscache.Untar kept in this package so // that existing tests (package main) continue to compile without importing // the rootfscache package directly. diff --git a/cmd/atelet/overlay.go b/cmd/atelet/overlay.go deleted file mode 100644 index bc2a85479..000000000 --- a/cmd/atelet/overlay.go +++ /dev/null @@ -1,87 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package main - -import ( - "fmt" - "os" - - "golang.org/x/sys/unix" -) - -// setupOverlayfs mounts an overlayfs at target using the given layers. -// - lowerDir: read-only shared rootfs from the node-level cache -// - upperDir: per-actor writable layer (created if absent) -// - workDir: overlayfs work directory (created if absent) -// -// The target directory must already exist (it is the bundle's rootfs/ dir). -func setupOverlayfs(target, lowerDir, upperDir, workDir string) error { - for _, d := range []string{upperDir, workDir} { - if err := os.MkdirAll(d, 0o700); err != nil { - return fmt.Errorf("creating overlay dir %s: %w", d, err) - } - } - - opts := fmt.Sprintf("lowerdir=%s,upperdir=%s,workdir=%s", lowerDir, upperDir, workDir) - if err := unix.Mount("overlay", target, "overlay", 0, opts); err != nil { - return fmt.Errorf("mounting overlayfs at %s: %w", target, err) - } - return nil -} - -// teardownOverlayfs unmounts the overlayfs at target. The caller is -// responsible for removing the upper/work directories (typically done by -// resetActorDirs). -func teardownOverlayfs(target string) error { - if err := unix.Unmount(target, unix.MNT_DETACH); err != nil { - return fmt.Errorf("unmounting overlayfs at %s: %w", target, err) - } - return nil -} - -// isOverlayfsAvailable checks whether the overlayfs kernel module is available -// by attempting a mount on a temporary directory. -func isOverlayfsAvailable() bool { - tmpLower, err := os.MkdirTemp("", "overlay-check-lower-") - if err != nil { - return false - } - defer os.RemoveAll(tmpLower) - - tmpUpper, err := os.MkdirTemp("", "overlay-check-upper-") - if err != nil { - return false - } - defer os.RemoveAll(tmpUpper) - - tmpWork, err := os.MkdirTemp("", "overlay-check-work-") - if err != nil { - return false - } - defer os.RemoveAll(tmpWork) - - tmpTarget, err := os.MkdirTemp("", "overlay-check-target-") - if err != nil { - return false - } - defer os.RemoveAll(tmpTarget) - - opts := fmt.Sprintf("lowerdir=%s,upperdir=%s,workdir=%s", tmpLower, tmpUpper, tmpWork) - if err := unix.Mount("overlay", tmpTarget, "overlay", 0, opts); err != nil { - return false - } - _ = unix.Unmount(tmpTarget, 0) - return true -} diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 30010eacc..6b632df14 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -204,6 +204,20 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload actorID: req.GetActorId(), } + // Mount the overlayfs rootfs for any container atelet flagged via a + // per-bundle marker. atelet cannot mount(2) (its capabilities are dropped), + // so the privileged ateom worker performs it here — in ateom's mount + // namespace, which the runsc child shares. On failure the deferred + // unmount tears down any partial mounts. + if err := mountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()); err != nil { + return nil, fmt.Errorf("while mounting overlay rootfs: %w", err) + } + defer func() { + if retErr != nil { + unmountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()) + } + }() + // Create and start pause container if err := rcmd.cmdCreate(ctx, os.Stdout, "pause", nil); err != nil { return nil, fmt.Errorf("while creating pause container: %w", err) @@ -296,6 +310,12 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec "err", err) } + // Tear down the overlay rootfs mounts before this ateom resets to + // available. The mounts live in ateom's mount namespace (invisible to + // atelet), so atelet's bundle RemoveAll cannot unmount them; a live, + // long-lived ateom would otherwise accumulate leaked mounts across actors. + unmountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()) + s.cleanupActorNetworkOrExit(ctx, "Failed to clean up actor network after checkpoint") // Report exactly the files runsc wrote so atelet ships precisely this set @@ -381,6 +401,19 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore actorID: req.GetActorId(), } + // Mount the overlayfs rootfs for any container atelet flagged via a + // per-bundle marker, mirroring RunWorkload. The mount happens in ateom's + // mount namespace (shared by the runsc child); on failure the deferred + // unmount tears down any partial mounts. + if err := mountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()); err != nil { + return nil, fmt.Errorf("while mounting overlay rootfs: %w", err) + } + defer func() { + if retErr != nil { + unmountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()) + } + }() + checkpointDir := ateompath.RestoreStateDir(req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId()) switch req.GetScope() { diff --git a/cmd/ateom-gvisor/overlay.go b/cmd/ateom-gvisor/overlay.go new file mode 100644 index 000000000..6818d5bb6 --- /dev/null +++ b/cmd/ateom-gvisor/overlay.go @@ -0,0 +1,122 @@ +//go:build linux + +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "strings" + + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/proto/ateompb" + "golang.org/x/sys/unix" +) + +// mountWorkloadOverlays mounts the overlayfs rootfs for every container in a +// workload (the hardcoded "pause" sandbox container plus each application +// container) that atelet flagged for overlay via a per-bundle marker. It is +// called just before `runsc create`, so the union rootfs is visible to the +// runsc child (which shares ateom's mount namespace). Containers without a +// marker are skipped: atelet populated their rootfs/ by direct untar. +func mountWorkloadOverlays(ctx context.Context, ns, tmpl, id string, spec *ateompb.WorkloadSpec) error { + if err := mountOverlayRootfsIfRequested(ctx, ns, tmpl, id, "pause"); err != nil { + return err + } + for _, ac := range spec.GetContainers() { + if err := mountOverlayRootfsIfRequested(ctx, ns, tmpl, id, ac.GetName()); err != nil { + return err + } + } + return nil +} + +// unmountWorkloadOverlays tears down every overlay mount created by +// mountWorkloadOverlays. Best-effort and safe to call even when a container had +// no overlay. Application containers are unmounted before "pause" to mirror the +// create order in reverse. +func unmountWorkloadOverlays(ctx context.Context, ns, tmpl, id string, spec *ateompb.WorkloadSpec) { + for _, ac := range spec.GetContainers() { + unmountOverlayRootfs(ctx, ns, tmpl, id, ac.GetName()) + } + unmountOverlayRootfs(ctx, ns, tmpl, id, "pause") +} + +// mountOverlayRootfsIfRequested mounts an overlayfs rootfs for one container if +// atelet left an overlay-lower marker in its bundle. atelet records the +// read-only lowerdir in the marker but cannot perform the mount itself (its +// capabilities were dropped); the privileged ateom worker mounts here. The +// upperdir/workdir/target are derived from the bundle path by the same +// convention atelet used when creating them. Returns nil (no-op) when there is +// no marker. +func mountOverlayRootfsIfRequested(ctx context.Context, ns, tmpl, id, container string) error { + marker := ateompath.OverlayLowerMarkerFile(ns, tmpl, id, container) + data, err := os.ReadFile(marker) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("while reading overlay lower marker for %q: %w", container, err) + } + lowerDir := strings.TrimSpace(string(data)) + if lowerDir == "" { + return fmt.Errorf("overlay lower marker for %q is empty", container) + } + + target := ateompath.ContainerRootfsDir(ns, tmpl, id, container) + upperDir := ateompath.OverlayUpperDir(ns, tmpl, id, container) + workDir := ateompath.OverlayWorkDir(ns, tmpl, id, container) + + for _, d := range []string{upperDir, workDir} { + if err := os.MkdirAll(d, 0o700); err != nil { + return fmt.Errorf("while creating overlay dir %s: %w", d, err) + } + } + + opts := fmt.Sprintf("lowerdir=%s,upperdir=%s,workdir=%s", lowerDir, upperDir, workDir) + if err := unix.Mount("overlay", target, "overlay", 0, opts); err != nil { + return fmt.Errorf("while mounting overlayfs rootfs for %q at %s (lower=%s): %w", container, target, lowerDir, err) + } + + slog.InfoContext(ctx, "Mounted overlay rootfs", + slog.String("container", container), + slog.String("lowerDir", lowerDir), + slog.String("target", target), + ) + return nil +} + +// unmountOverlayRootfs tears down the overlay rootfs mount for one container. +// It is best-effort: containers without an overlay marker are skipped, and a +// lazy MNT_DETACH unmount tolerates a target that is not (or is no longer) a +// mountpoint. +func unmountOverlayRootfs(ctx context.Context, ns, tmpl, id, container string) { + marker := ateompath.OverlayLowerMarkerFile(ns, tmpl, id, container) + if _, err := os.Stat(marker); errors.Is(err, os.ErrNotExist) { + return + } + target := ateompath.ContainerRootfsDir(ns, tmpl, id, container) + if err := unix.Unmount(target, unix.MNT_DETACH); err != nil { + slog.DebugContext(ctx, "overlay rootfs unmount skipped (not a mountpoint)", + slog.String("container", container), + slog.String("target", target), + slog.Any("err", err), + ) + } +} diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 1dc8eb3c9..7f0c51144 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -127,6 +127,46 @@ func OCIBundlePath(actorTemplateNamespace, actorTemplateName, actorID, container ) } +// ContainerRootfsDir is the rootfs directory inside a container's OCI bundle. +// It is the OCI spec's Root.Path target: either an overlayfs mountpoint (when +// the rootfs cache is used) or a directly-extracted rootfs (fallback). +func ContainerRootfsDir(actorTemplateNamespace, actorTemplateName, actorID, containerName string) string { + return filepath.Join( + OCIBundlePath(actorTemplateNamespace, actorTemplateName, actorID, containerName), + "rootfs", + ) +} + +// OverlayUpperDir is the per-container writable overlay layer (overlayfs +// upperdir); copy-ups of the read-only shared lowerdir land here. +func OverlayUpperDir(actorTemplateNamespace, actorTemplateName, actorID, containerName string) string { + return filepath.Join( + OCIBundlePath(actorTemplateNamespace, actorTemplateName, actorID, containerName), + "upper", + ) +} + +// OverlayWorkDir is the overlayfs work directory required alongside upperdir. +func OverlayWorkDir(actorTemplateNamespace, actorTemplateName, actorID, containerName string) string { + return filepath.Join( + OCIBundlePath(actorTemplateNamespace, actorTemplateName, actorID, containerName), + "work", + ) +} + +// OverlayLowerMarkerFile is the per-container marker atelet writes (containing +// the node-local read-only lowerdir path) to request that the privileged ateom +// worker mount an overlayfs rootfs for this container. atelet cannot call +// mount(2) after its capabilities were dropped, so the mount is performed by +// ateom just before `runsc create`. The file's absence means no overlay: +// atelet populated ContainerRootfsDir by direct untar instead. +func OverlayLowerMarkerFile(actorTemplateNamespace, actorTemplateName, actorID, containerName string) string { + return filepath.Join( + OCIBundlePath(actorTemplateNamespace, actorTemplateName, actorID, containerName), + "overlay-lower", + ) +} + func RunscDebugLogDir(actorTemplateNamespace, actorTemplateName, actorID, containerName string) string { return filepath.Join( ActorPath(actorTemplateNamespace, actorTemplateName, actorID), diff --git a/internal/ateompath/ateompath_test.go b/internal/ateompath/ateompath_test.go index e6978a0eb..4e5097a09 100644 --- a/internal/ateompath/ateompath_test.go +++ b/internal/ateompath/ateompath_test.go @@ -58,3 +58,39 @@ func TestAteomPathUniqueness(t *testing.T) { t.Errorf("expected different paths for different pod UIDs, got %q", path1) } } + +// TestOverlayPathHelpers checks that every per-container overlay path helper is +// rooted under the container's OCI bundle, that they are mutually distinct +// (upper/work/lower must not collide with each other or with rootfs — an +// overlayfs mount requires them separate), and that they are deterministic. +func TestOverlayPathHelpers(t *testing.T) { + const ( + ns = "team-a" + tmpl = "web" + id = "actor-1" + container = "app" + ) + + bundle := OCIBundlePath(ns, tmpl, id, container) + + cases := []struct { + name string + got string + }{ + {"rootfs", ContainerRootfsDir(ns, tmpl, id, container)}, + {"upper", OverlayUpperDir(ns, tmpl, id, container)}, + {"work", OverlayWorkDir(ns, tmpl, id, container)}, + {"marker", OverlayLowerMarkerFile(ns, tmpl, id, container)}, + } + + seen := make(map[string]string, len(cases)) + for _, tc := range cases { + if !strings.HasPrefix(tc.got, bundle+"/") { + t.Errorf("%s: %q is not under bundle %q", tc.name, tc.got, bundle) + } + if prev, dup := seen[tc.got]; dup { + t.Errorf("%s and %s collide on path %q", prev, tc.name, tc.got) + } + seen[tc.got] = tc.name + } +} From 115a918b3f5197ff10dd76faa62eb3125a1f3811 Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 15:32:00 +0800 Subject: [PATCH 4/9] fix(ateom): harden overlay mount teardown against leaks Two leak/idempotency gaps in the overlay rootfs lifecycle: - CheckpointWorkload unmounted only after the checkpoint calls, so an early-return checkpoint failure skipped the unmount and leaked the mount into a long-lived ateom. Move the unmount to a defer registered before the checkpoint so it runs on every exit path. - mountOverlayRootfsIfRequested now best-effort unmounts the target before mounting, mirroring setupActorNetwork's stale-network cleanup. This makes re-mount idempotent (no stacked overlays on actor reuse) and closes the window from any prior leaked mount at the same target. --- cmd/ateom-gvisor/main.go | 14 ++++++++------ cmd/ateom-gvisor/overlay.go | 8 ++++++++ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 6b632df14..39e8897d8 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -270,6 +270,14 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec actorID: req.GetActorId(), } + // Tear down the overlay rootfs mounts before this ateom resets to + // available, regardless of whether the checkpoint below succeeds. The + // mounts live in ateom's mount namespace (invisible to atelet), so + // atelet's bundle RemoveAll cannot unmount them; deferring here ensures an + // early-return checkpoint failure does not leak the mount into a + // long-lived ateom that goes on to serve more actors. + defer unmountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()) + checkpointPath := ateompath.CheckpointStateDir(req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId()) if err := os.MkdirAll(checkpointPath, 0o700); err != nil { return nil, fmt.Errorf("while creating checkpoint directory: %w", err) @@ -310,12 +318,6 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec "err", err) } - // Tear down the overlay rootfs mounts before this ateom resets to - // available. The mounts live in ateom's mount namespace (invisible to - // atelet), so atelet's bundle RemoveAll cannot unmount them; a live, - // long-lived ateom would otherwise accumulate leaked mounts across actors. - unmountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()) - s.cleanupActorNetworkOrExit(ctx, "Failed to clean up actor network after checkpoint") // Report exactly the files runsc wrote so atelet ships precisely this set diff --git a/cmd/ateom-gvisor/overlay.go b/cmd/ateom-gvisor/overlay.go index 6818d5bb6..caf115630 100644 --- a/cmd/ateom-gvisor/overlay.go +++ b/cmd/ateom-gvisor/overlay.go @@ -83,6 +83,14 @@ func mountOverlayRootfsIfRequested(ctx context.Context, ns, tmpl, id, container upperDir := ateompath.OverlayUpperDir(ns, tmpl, id, container) workDir := ateompath.OverlayWorkDir(ns, tmpl, id, container) + // Best-effort unmount of any stale overlay left at this target before + // mounting. Mirrors setupActorNetwork's "clean stale network before setup": + // a prior actor on this ateom whose CheckpointWorkload failed before its + // unmount, or a crash-and-reuse, can leave the target still mounted — + // re-mounting on top would stack a second overlay. MNT_DETACH tolerates a + // target that is not (or is no longer) a mountpoint. + _ = unix.Unmount(target, unix.MNT_DETACH) + for _, d := range []string{upperDir, workDir} { if err := os.MkdirAll(d, 0o700); err != nil { return fmt.Errorf("while creating overlay dir %s: %w", d, err) From 5b521be3dec6a37ddeb9c25bc097a6ad252534bd Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 15:52:05 +0800 Subject: [PATCH 5/9] feat: skip in-use lowerdirs during rootfs cache eviction LRU eviction could os.RemoveAll a lowerdir currently backing a live actor's overlayfs mount, which the kernel forbids modifying and would corrupt the running actor. atelet restarts independently of the actors whose overlays live in ateom's mount namespace, so an in-memory refcount is not restart-safe. Add a restart-safe, on-disk in-use signal: overlayLowerInUse scans the per-container overlay-lower bundle markers (written before ateom mounts, removed only at teardown) into the set of lowerdirs in use, injected into the cache via the new WithInUseFunc functional option. evictIfNeeded and EvictLRU skip pinned entries and, on provider error, skip the pass entirely (exceeding the disk budget is safe; deleting a live lowerdir is not). --- .../internal/rootfscache/rootfscache.go | 77 +++++++- .../internal/rootfscache/rootfscache_test.go | 181 +++++++++++++++++- cmd/atelet/main.go | 2 +- cmd/atelet/rootfsinuse.go | 62 ++++++ internal/ateompath/ateompath.go | 7 + 5 files changed, 314 insertions(+), 15 deletions(-) create mode 100644 cmd/atelet/rootfsinuse.go diff --git a/cmd/atelet/internal/rootfscache/rootfscache.go b/cmd/atelet/internal/rootfscache/rootfscache.go index 38a3ccd25..1a802bd84 100644 --- a/cmd/atelet/internal/rootfscache/rootfscache.go +++ b/cmd/atelet/internal/rootfscache/rootfscache.go @@ -48,6 +48,24 @@ const readySentinel = ".ready" // most recent cache hit, used for LRU eviction. const lastAccessFile = ".last_access" +// InUseFunc reports the set of lowerDir paths currently referenced by a live +// actor (keyed by absolute lowerDir path, value always true). Eviction skips +// any entry whose lowerDir is in this set: an overlayfs mount forbids modifying +// its lowerdir while mounted, so os.RemoveAll of a live lowerdir would corrupt +// a running actor. The signal must be restart-safe (derived from on-disk state, +// not an in-memory refcount) because atelet restarts independently of the +// actors whose overlays are mounted in ateom's mount namespace. +type InUseFunc func() (map[string]bool, error) + +// Option configures a Cache at construction time. +type Option func(*Cache) + +// WithInUseFunc installs a provider that reports the lowerDirs currently in use +// by live actors, so eviction can skip them. If unset, eviction pins nothing. +func WithInUseFunc(fn InUseFunc) Option { + return func(c *Cache) { c.inUse = fn } +} + // entryState tracks one cached digest. All fields are immutable after // construction except lastAccess, which is updated on cache hits. type entryState struct { @@ -60,12 +78,16 @@ type entryState struct { // Cache is a node-local, digest-keyed rootfs cache. It is safe for // concurrent use. type Cache struct { - basePath string + basePath string maxCacheBytes int64 mu sync.Mutex entries map[string]*entryState // keyed by digest + // inUse, when non-nil, reports lowerDirs currently mounted by a live + // actor. Eviction consults it to avoid deleting an in-use lowerdir. + inUse InUseFunc + // inflight deduplicates concurrent EnsureRootfs calls for the same // digest: the first goroutine extracts while others wait. inflight map[string]*inflightEntry @@ -85,8 +107,9 @@ type inflightEntry struct { // New creates a Cache rooted at basePath. The directory is created if it does // not exist. maxCacheBytes caps total disk usage; pass 0 for -// DefaultMaxCacheBytes. -func New(ctx context.Context, basePath string, maxCacheBytes int64) (*Cache, error) { +// DefaultMaxCacheBytes. Optional behavior (e.g. an in-use provider for +// eviction) is configured via Options. +func New(ctx context.Context, basePath string, maxCacheBytes int64, opts ...Option) (*Cache, error) { if maxCacheBytes <= 0 { maxCacheBytes = DefaultMaxCacheBytes } @@ -116,6 +139,9 @@ func New(ctx context.Context, basePath string, maxCacheBytes int64) (*Cache, err cacheHits: cacheHits, cacheMisses: cacheMisses, } + for _, opt := range opts { + opt(c) + } // Load existing cache entries from disk. if err := c.loadIndex(ctx); err != nil { @@ -351,6 +377,16 @@ func (c *Cache) touchAccess(digest string) error { // evictIfNeeded removes the oldest entries until total cache size is within // the budget. It is called asynchronously after each extraction. func (c *Cache) evictIfNeeded(ctx context.Context) { + // Compute the in-use set before taking c.mu: the provider does disk I/O + // (globbing bundle markers) and must not run under the cache lock. On + // provider error, skip this eviction pass entirely — exceeding the budget + // is safe; deleting a live lowerdir is not. + inUse, err := c.currentInUse() + if err != nil { + slog.WarnContext(ctx, "Skipping rootfs cache eviction: in-use lookup failed", slog.Any("err", err)) + return + } + c.mu.Lock() defer c.mu.Unlock() @@ -360,14 +396,23 @@ func (c *Cache) evictIfNeeded(ctx context.Context) { } for total > c.maxCacheBytes && len(c.entries) > 0 { - // Find the entry with the oldest lastAccess. + // Find the oldest entry that is not currently in use by a live actor. var oldest *entryState for _, e := range c.entries { + if inUse[e.lowerDir] { + continue + } if oldest == nil || e.lastAccess.Before(oldest.lastAccess) { oldest = e } } if oldest == nil { + // Every remaining entry is pinned by a live actor. Better to + // exceed the disk budget than corrupt a mounted lowerdir. + slog.WarnContext(ctx, "Rootfs cache over budget but all entries in use; deferring eviction", + slog.Int64("totalBytes", total), + slog.Int64("maxCacheBytes", c.maxCacheBytes), + ) break } @@ -387,14 +432,34 @@ func (c *Cache) evictIfNeeded(ctx context.Context) { } } -// EvictLRU removes the least-recently-used cache entry and returns its digest -// and size, or ("", 0) if the cache is empty. This is exported for tests. +// currentInUse returns the set of lowerDirs currently referenced by live +// actors, or an empty set when no provider is configured. The provider does +// disk I/O, so callers must invoke this outside c.mu. +func (c *Cache) currentInUse() (map[string]bool, error) { + if c.inUse == nil { + return nil, nil + } + return c.inUse() +} + +// EvictLRU removes the least-recently-used cache entry that is not currently in +// use by a live actor and returns its digest and size, or ("", 0) if there is +// no evictable entry (cache empty or every entry pinned). This is exported for +// tests. func (c *Cache) EvictLRU() (string, int64) { + inUse, err := c.currentInUse() + if err != nil { + return "", 0 + } + c.mu.Lock() defer c.mu.Unlock() var oldest *entryState for _, e := range c.entries { + if inUse[e.lowerDir] { + continue + } if oldest == nil || e.lastAccess.Before(oldest.lastAccess) { oldest = e } diff --git a/cmd/atelet/internal/rootfscache/rootfscache_test.go b/cmd/atelet/internal/rootfscache/rootfscache_test.go index ba653fe63..fa83205cb 100644 --- a/cmd/atelet/internal/rootfscache/rootfscache_test.go +++ b/cmd/atelet/internal/rootfscache/rootfscache_test.go @@ -27,7 +27,11 @@ import ( const testDigest = "sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890" -func buildTar(t *testing.T, entries []struct{ name, body string; typeflag byte; mode int64 }) []byte { +func buildTar(t *testing.T, entries []struct { + name, body string + typeflag byte + mode int64 +}) []byte { t.Helper() var buf bytes.Buffer tw := tar.NewWriter(&buf) @@ -76,7 +80,11 @@ func TestEnsureRootfs_CacheMiss(t *testing.T) { t.Fatalf("New: %v", err) } - tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, {name: "etc/", typeflag: tar.TypeDir}, {name: "etc/hostname", typeflag: tar.TypeReg, body: "test-host\n"}, @@ -120,7 +128,11 @@ func TestEnsureRootfs_CacheHit(t *testing.T) { t.Fatalf("New: %v", err) } - tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, {name: "hello", typeflag: tar.TypeReg, body: "world"}, }) @@ -159,7 +171,11 @@ func TestEnsureRootfs_ConcurrentMisses(t *testing.T) { t.Fatalf("New: %v", err) } - tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, {name: "concurrent", typeflag: tar.TypeReg, body: "ok"}, }) @@ -232,7 +248,11 @@ func TestEnsureRootfs_PartialEntryCleanup(t *testing.T) { } // Now a fresh extraction should succeed. - tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, {name: "fresh", typeflag: tar.TypeReg, body: "data"}, }) @@ -258,11 +278,19 @@ func TestEvictLRU(t *testing.T) { digest1 := "sha256:1111111111111111111111111111111111111111111111111111111111111111" digest2 := "sha256:2222222222222222222222222222222222222222222222222222222222222222" - tarData1 := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData1 := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, {name: "d1", typeflag: tar.TypeReg, body: "data1"}, }) - tarData2 := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData2 := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, {name: "d2", typeflag: tar.TypeReg, body: "data2"}, }) @@ -296,6 +324,139 @@ func TestEvictLRU(t *testing.T) { } } +// seedTwoEntries populates the cache with digest1 (older) and digest2 (newer) +// and returns their lowerDirs. digest1 is guaranteed to sort as the LRU victim. +func seedTwoEntries(t *testing.T, c *Cache) (digest1, digest2, lower1, lower2 string) { + t.Helper() + digest1, digest2 = evictDigest1, evictDigest2 + + tarData1 := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "d1", typeflag: tar.TypeReg, body: "data1"}, + }) + tarData2 := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "d2", typeflag: tar.TypeReg, body: "data2"}, + }) + + l1, _, err := c.EnsureRootfs(context.Background(), digest1, tarProviderFor(tarData1)) + if err != nil { + t.Fatalf("EnsureRootfs d1: %v", err) + } + l2, _, err := c.EnsureRootfs(context.Background(), digest2, tarProviderFor(tarData2)) + if err != nil { + t.Fatalf("EnsureRootfs d2: %v", err) + } + return digest1, digest2, l1, l2 +} + +const ( + evictDigest1 = "sha256:1111111111111111111111111111111111111111111111111111111111111111" + evictDigest2 = "sha256:2222222222222222222222222222222222222222222222222222222222222222" +) + +// lowerPathFor returns the deterministic lowerDir a cache rooted at base would +// use for digest. Tests use it to pre-build an in-use set before seeding, so the +// injected provider never mutates shared state concurrently with the background +// eviction goroutine (which would otherwise trip the race detector). +func lowerPathFor(base, digest string) string { + return filepath.Join(base, digest, "lower") +} + +// TestEvictLRU_SkipsInUse verifies that an in-use lowerDir (as reported by the +// injected provider) is never selected as the eviction victim, even when it is +// the least-recently-used entry — deleting a mounted lowerdir would corrupt a +// live actor. +func TestEvictLRU_SkipsInUse(t *testing.T) { + base := t.TempDir() + + // Pin the older entry (digest1) up front. Eviction must skip it and take + // digest2. The set is immutable after New, so the background eviction + // goroutine can read it race-free. + pinned := map[string]bool{lowerPathFor(base, evictDigest1): true} + c, err := New(context.Background(), base, 0, WithInUseFunc(func() (map[string]bool, error) { + return pinned, nil + })) + if err != nil { + t.Fatalf("New: %v", err) + } + + digest1, digest2, _, _ := seedTwoEntries(t, c) + + evicted, size := c.EvictLRU() + if evicted != digest2 { + t.Errorf("evicted = %q, want %q (the pinned LRU entry must be skipped)", evicted, digest2) + } + if size <= 0 { + t.Errorf("evicted size = %d, want > 0", size) + } + if c.LowerDir(digest1) == "" { + t.Errorf("pinned digest1 was evicted; it must remain") + } + if _, err := os.Stat(filepath.Join(base, digest1)); err != nil { + t.Errorf("pinned digest1 dir missing: %v", err) + } +} + +// TestEvictLRU_AllPinned verifies that when every entry is in use, EvictLRU +// evicts nothing rather than deleting a live lowerdir. +func TestEvictLRU_AllPinned(t *testing.T) { + base := t.TempDir() + + pinned := map[string]bool{ + lowerPathFor(base, evictDigest1): true, + lowerPathFor(base, evictDigest2): true, + } + c, err := New(context.Background(), base, 0, WithInUseFunc(func() (map[string]bool, error) { + return pinned, nil + })) + if err != nil { + t.Fatalf("New: %v", err) + } + + seedTwoEntries(t, c) + + evicted, size := c.EvictLRU() + if evicted != "" || size != 0 { + t.Errorf("EvictLRU = (%q, %d), want (\"\", 0) when all entries pinned", evicted, size) + } + if c.Count() != 2 { + t.Errorf("count = %d, want 2 (nothing should be evicted)", c.Count()) + } +} + +// TestEvictLRU_ProviderError verifies that a failing in-use provider makes +// eviction a no-op (conservative: exceeding the budget beats corrupting a +// possibly-live lowerdir). +func TestEvictLRU_ProviderError(t *testing.T) { + base := t.TempDir() + + c, err := New(context.Background(), base, 0, WithInUseFunc(func() (map[string]bool, error) { + return nil, io.ErrUnexpectedEOF + })) + if err != nil { + t.Fatalf("New: %v", err) + } + + seedTwoEntries(t, c) + + evicted, size := c.EvictLRU() + if evicted != "" || size != 0 { + t.Errorf("EvictLRU = (%q, %d), want (\"\", 0) on provider error", evicted, size) + } + if c.Count() != 2 { + t.Errorf("count = %d, want 2 (nothing should be evicted on provider error)", c.Count()) + } +} + func TestLowerDir(t *testing.T) { base := t.TempDir() c, err := New(context.Background(), base, 0) @@ -308,7 +469,11 @@ func TestLowerDir(t *testing.T) { t.Errorf("LowerDir before cache = %q, want empty", got) } - tarData := buildTar(t, []struct{ name, body string; typeflag byte; mode int64 }{ + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ {name: ".", typeflag: tar.TypeDir}, }) if _, _, err := c.EnsureRootfs(context.Background(), testDigest, tarProviderFor(tarData)); err != nil { diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 0c3254b13..4cf9a6f40 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -116,7 +116,7 @@ func main() { serverboot.Fatal(ctx, "Failed to create pull cache", err) } - rootfsDiskCache, err := rootfscache.New(ctx, ateompath.RootfsCacheDir, 0) + rootfsDiskCache, err := rootfscache.New(ctx, ateompath.RootfsCacheDir, 0, rootfscache.WithInUseFunc(overlayLowerInUse)) if err != nil { serverboot.Fatal(ctx, "Failed to create rootfs cache", err) } diff --git a/cmd/atelet/rootfsinuse.go b/cmd/atelet/rootfsinuse.go new file mode 100644 index 000000000..1715bb898 --- /dev/null +++ b/cmd/atelet/rootfsinuse.go @@ -0,0 +1,62 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/agent-substrate/substrate/internal/ateompath" +) + +// overlayLowerInUse reports the rootfs-cache lowerDirs currently referenced by +// a live actor on this node, by scanning every per-container overlay-lower +// marker under the actors tree. atelet writes a marker before ateom mounts the +// overlay and removes it only after teardown (unmount), so a present marker is +// a conservative, restart-safe signal that the lowerdir is mounted: it is +// derived purely from on-disk state and survives atelet restarts (unlike an +// in-memory refcount, which would be lost while the mount lives on in ateom's +// mount namespace). +// +// It is wired into the rootfs cache via rootfscache.WithInUseFunc so eviction +// never os.RemoveAll's a lowerdir backing a running actor's overlay mount +// (which the kernel forbids modifying while mounted). +func overlayLowerInUse() (map[string]bool, error) { + pattern := filepath.Join(ateompath.ActorsDir(), "*", "bundles", "*", "overlay-lower") + matches, err := filepath.Glob(pattern) + if err != nil { + return nil, fmt.Errorf("globbing overlay-lower markers: %w", err) + } + + inUse := make(map[string]bool, len(matches)) + for _, m := range matches { + data, err := os.ReadFile(m) + if err != nil { + // A marker removed between glob and read simply means that actor + // finished tearing down; treat it as no longer in use. + if errors.Is(err, os.ErrNotExist) { + continue + } + return nil, fmt.Errorf("reading overlay-lower marker %s: %w", m, err) + } + if lower := strings.TrimSpace(string(data)); lower != "" { + inUse[lower] = true + } + } + return inUse, nil +} diff --git a/internal/ateompath/ateompath.go b/internal/ateompath/ateompath.go index 7f0c51144..511b7475f 100644 --- a/internal/ateompath/ateompath.go +++ b/internal/ateompath/ateompath.go @@ -81,6 +81,13 @@ func ActorPath(actorTemplateNamespace, actorTemplateName, actorID string) string ) } +// ActorsDir is the parent directory holding every per-actor directory. It is +// the glob root for enumerating live actors on this node (e.g. to find which +// overlay lowerdirs are currently in use before evicting the rootfs cache). +func ActorsDir() string { + return filepath.Join(BasePath, "actors") +} + // ActorIdentityDirPath is the host directory atelet populates with the // actor's identity data (currently the single file "actor-id") and // bind-mounts read-only into the actor. It is per-actor and regenerated on From 8b603d4fa3ce65d99ba3ca73348a77761de51a90 Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 16:11:33 +0800 Subject: [PATCH 6/9] refactor(atelet): scope identity mount-point creation to untar path Only the untar path needs to pre-create the identity bind target: runsc resolves the destination against the extracted image there. On the overlay path the pre-created dir is shadowed by ateom's later overlay mount, and runsc auto-creates the missing target into the upperdir anyway. --- cmd/atelet/oci.go | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/cmd/atelet/oci.go b/cmd/atelet/oci.go index 845a22ff9..d00f7ca28 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -98,6 +98,13 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP return fmt.Errorf("while writing overlay lower marker: %w", err) } + // We deliberately do NOT pre-create the identity mount point here: ateom + // mounts the overlay over rootPath after prepareOCIDirectory returns, + // which would shadow anything created now. runsc auto-creates the bind + // target inside the sandbox (the new dir lands in the overlay upperdir), + // so the identity bind mount attaches without our help. (Verified: a bind + // mount to a path absent from the image lower still attaches.) + span.SetAttributes(attribute.String("rootfs_method", "overlay")) } else { // Fallback: no digest or no cache — extract directly (original path). @@ -118,16 +125,20 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP return fmt.Errorf("in untar: %w", err) } - span.SetAttributes(attribute.String("rootfs_method", "untar")) - } - - // Bind-mount the per-actor identity directory so the workload can read its - // own ID at IdentityMountPath/ActorIDFileName. The bind target must exist - // in the rootfs for the mount to attach. - if identityDir != "" { - if err := createMountPoint(rootPath, IdentityMountPath); err != nil { - return fmt.Errorf("while creating identity mount point: %w", err) + // Pre-create the identity bind-mount target. On this (untar) path the + // rootfs is the container's actual root with no overlay on top, so runsc + // resolves the bind destination against the extracted image; if the image + // lacks IdentityMountPath the mount would fail. We create it here so the + // target always exists. (On the overlay path this is unnecessary — and + // would be shadowed by ateom's later overlay mount anyway — because runsc + // auto-creates the missing target into the overlay upperdir.) + if identityDir != "" { + if err := createMountPoint(rootPath, IdentityMountPath); err != nil { + return fmt.Errorf("while creating identity mount point: %w", err) + } } + + span.SetAttributes(attribute.String("rootfs_method", "untar")) } ociSpec := buildActorOCISpec(actorTemplateNamespace, actorTemplateName, actorID, args, env, annotations, netns, identityDir, durableDirVolumeMounts) From fbb7430fde49e1c6ff10edfeec6e852c26e1c8fc Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 16:27:14 +0800 Subject: [PATCH 7/9] perf(atelet): evict rootfs cache at startup and size entries from untar Two rootfs-cache improvements: - Trigger one eviction pass in New (after loadIndex) so a node that boots already over budget reclaims disk immediately, rather than waiting for the next cache miss. Fixes a steady-state, all-hits node never shrinking. - Untar now returns the total regular-file bytes written, and extract uses it to size the cache entry, dropping the extra full-tree dirSize walk on the miss path. Hardlinks are no longer double-counted. --- .../internal/rootfscache/rootfscache.go | 63 ++++++----- .../internal/rootfscache/rootfscache_test.go | 106 ++++++++++++++++++ cmd/atelet/oci.go | 5 +- 3 files changed, 147 insertions(+), 27 deletions(-) diff --git a/cmd/atelet/internal/rootfscache/rootfscache.go b/cmd/atelet/internal/rootfscache/rootfscache.go index 1a802bd84..1127e4647 100644 --- a/cmd/atelet/internal/rootfscache/rootfscache.go +++ b/cmd/atelet/internal/rootfscache/rootfscache.go @@ -148,6 +148,13 @@ func New(ctx context.Context, basePath string, maxCacheBytes int64, opts ...Opti slog.WarnContext(ctx, "Failed to load rootfs cache index, starting empty", slog.Any("err", err)) } + // A node can boot already over budget — e.g. the budget was lowered across + // a restart, or a prior version filled the cache. Eviction is otherwise + // only triggered by a miss (extract), so without this a steady-state, + // all-hits node would never reclaim disk. Run one pass now (best-effort, + // detached from the startup ctx so it isn't cancelled when New returns). + go c.evictIfNeeded(context.Background()) + return c, nil } @@ -275,7 +282,8 @@ func (c *Cache) extract(ctx context.Context, digest string, tarData io.Reader) ( slog.String("lowerDir", lowerDir), ) - if err := Untar(ctx, tarData, lowerDir); err != nil { + size, err := Untar(ctx, tarData, lowerDir) + if err != nil { // Clean up on failure so the next attempt starts fresh. _ = os.RemoveAll(filepath.Join(c.basePath, digest)) return "", fmt.Errorf("extracting rootfs: %w", err) @@ -296,8 +304,8 @@ func (c *Cache) extract(ctx context.Context, digest string, tarData io.Reader) ( return "", fmt.Errorf("writing last_access: %w", err) } - // Register in the in-memory index. - size := dirSize(lowerDir) + // Register in the in-memory index. size came from Untar, so we avoid a + // second full-tree walk here. c.mu.Lock() c.entries[digest] = &entryState{ digest: digest, @@ -497,30 +505,34 @@ func (c *Cache) Count() int { // Untar extracts a tar stream into rootPath. It is a self-contained copy of // the untar logic from cmd/atelet/oci.go, using os.OpenRoot for path-traversal -// safety. -func Untar(ctx context.Context, tarData io.Reader, rootPath string) error { +// safety. It returns the total bytes of regular-file content written, so the +// caller can size the cache entry without a second full-tree walk (hardlinks +// share content and are not double-counted). The byte count is only meaningful +// when err is nil. +func Untar(ctx context.Context, tarData io.Reader, rootPath string) (int64, error) { tracer := otel.Tracer("rootfscache") _, span := tracer.Start(ctx, "Untar") defer span.End() root, err := os.OpenRoot(rootPath) if err != nil { - return fmt.Errorf("opening rootfs %q as os.Root: %w", rootPath, err) + return 0, fmt.Errorf("opening rootfs %q as os.Root: %w", rootPath, err) } defer root.Close() + var totalBytes int64 tarReader := tar.NewReader(tarData) for { hdr, err := tarReader.Next() if errors.Is(err, io.EOF) { break } else if err != nil { - return fmt.Errorf("in tarReader.Next: %w", err) + return 0, fmt.Errorf("in tarReader.Next: %w", err) } name, skip, err := ValidateTarName(hdr.Name) if err != nil { - return fmt.Errorf("invalid tar entry: %w", err) + return 0, fmt.Errorf("invalid tar entry: %w", err) } if skip { continue @@ -532,31 +544,32 @@ func Untar(ctx context.Context, tarData io.Reader, rootPath string) error { case tar.TypeReg: if _, err := root.Lstat(name); err == nil { if err := root.RemoveAll(name); err != nil { - return fmt.Errorf("while replacing existing path at %q before regular file: %w", name, err) + return 0, fmt.Errorf("while replacing existing path at %q before regular file: %w", name, err) } } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("while checking existing path at %q before regular file: %w", name, err) + return 0, fmt.Errorf("while checking existing path at %q before regular file: %w", name, err) } outFile, err := root.OpenFile(name, os.O_CREATE|os.O_RDWR|os.O_TRUNC, mode) if err != nil { - return fmt.Errorf("while creating file %q: %w", name, err) + return 0, fmt.Errorf("while creating file %q: %w", name, err) } - _, err = io.Copy(outFile, tarReader) + n, err := io.Copy(outFile, tarReader) closeErr := outFile.Close() if err != nil { - return fmt.Errorf("while writing contents of %q from tar stream: %w", name, err) + return 0, fmt.Errorf("while writing contents of %q from tar stream: %w", name, err) } if closeErr != nil { - return fmt.Errorf("while closing file %q: %w", name, closeErr) + return 0, fmt.Errorf("while closing file %q: %w", name, closeErr) } + totalBytes += n case tar.TypeDir: err := root.Mkdir(name, mode) if errors.Is(err, os.ErrExist) { // Tolerate repeated directory entries. } else if err != nil { - return fmt.Errorf("while creating directory=%q, mode=%v: %w", name, mode, err) + return 0, fmt.Errorf("while creating directory=%q, mode=%v: %w", name, mode, err) } case tar.TypeSymlink: @@ -567,42 +580,42 @@ func Untar(ctx context.Context, tarData io.Reader, rootPath string) error { } } if err := root.RemoveAll(name); err != nil { - return fmt.Errorf("while replacing existing path at %q before symlink: %w", name, err) + return 0, fmt.Errorf("while replacing existing path at %q before symlink: %w", name, err) } } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("while checking existing path at %q before symlink: %w", name, err) + return 0, fmt.Errorf("while checking existing path at %q before symlink: %w", name, err) } if err := root.Symlink(hdr.Linkname, name); err != nil { - return fmt.Errorf("while creating symlink src=%q target=%q: %w", name, hdr.Linkname, err) + return 0, fmt.Errorf("while creating symlink src=%q target=%q: %w", name, hdr.Linkname, err) } case tar.TypeLink: linkname, linkSkip, err := ValidateTarName(hdr.Linkname) if err != nil { - return fmt.Errorf("invalid hardlink target for %q: %w", name, err) + return 0, fmt.Errorf("invalid hardlink target for %q: %w", name, err) } if linkSkip { - return fmt.Errorf("invalid hardlink target for %q: empty", name) + return 0, fmt.Errorf("invalid hardlink target for %q: empty", name) } if _, err := root.Lstat(name); err == nil { if err := root.RemoveAll(name); err != nil { - return fmt.Errorf("while replacing existing path at %q before hardlink: %w", name, err) + return 0, fmt.Errorf("while replacing existing path at %q before hardlink: %w", name, err) } } else if !errors.Is(err, os.ErrNotExist) { - return fmt.Errorf("while checking existing path at %q before hardlink: %w", name, err) + return 0, fmt.Errorf("while checking existing path at %q before hardlink: %w", name, err) } if err := root.Link(linkname, name); err != nil { - return fmt.Errorf("while creating hardlink src=%q target=%q: %w", name, linkname, err) + return 0, fmt.Errorf("while creating hardlink src=%q target=%q: %w", name, linkname, err) } default: tfStr := string([]byte{hdr.Typeflag}) slog.ErrorContext(ctx, "Unhandled tar entry typeflag", slog.String("typeflag", tfStr), slog.Any("hdr", hdr)) - return fmt.Errorf("unhandled tar entry typeflag %q", tfStr) + return 0, fmt.Errorf("unhandled tar entry typeflag %q", tfStr) } } - return nil + return totalBytes, nil } // --- helpers -------------------------------------------------------------- diff --git a/cmd/atelet/internal/rootfscache/rootfscache_test.go b/cmd/atelet/internal/rootfscache/rootfscache_test.go index fa83205cb..dd6994595 100644 --- a/cmd/atelet/internal/rootfscache/rootfscache_test.go +++ b/cmd/atelet/internal/rootfscache/rootfscache_test.go @@ -23,6 +23,7 @@ import ( "path/filepath" "sync" "testing" + "time" ) const testDigest = "sha256:abcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890" @@ -490,6 +491,111 @@ func TestLowerDir(t *testing.T) { } } +// seedReadyEntryOnDisk writes a completed cache entry (lower/ + .ready + +// .last_access) directly to disk so a freshly constructed Cache picks it up via +// loadIndex. bodySize bytes of content are written so the entry has nonzero +// size; accessedAt sets the recorded last-access time (older sorts as the LRU +// victim). +func seedReadyEntryOnDisk(t *testing.T, base, digest string, bodySize int, accessedAt time.Time) { + t.Helper() + lower := filepath.Join(base, digest, "lower") + if err := os.MkdirAll(lower, 0o700); err != nil { + t.Fatalf("mkdir lower: %v", err) + } + if err := os.WriteFile(filepath.Join(lower, "blob"), make([]byte, bodySize), 0o644); err != nil { + t.Fatalf("write blob: %v", err) + } + if err := os.WriteFile(filepath.Join(base, digest, readySentinel), []byte(accessedAt.Format(time.RFC3339)), 0o444); err != nil { + t.Fatalf("write ready: %v", err) + } + if err := os.WriteFile(filepath.Join(base, digest, lastAccessFile), []byte(accessedAt.Format(time.RFC3339Nano)), 0o644); err != nil { + t.Fatalf("write last_access: %v", err) + } +} + +// TestUntar_ReturnsByteCount verifies Untar reports the total regular-file +// content bytes it wrote, so the cache can size an entry without a second walk. +func TestUntar_ReturnsByteCount(t *testing.T) { + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "a", typeflag: tar.TypeReg, body: "hello"}, // 5 + {name: "sub/", typeflag: tar.TypeDir}, + {name: "sub/b", typeflag: tar.TypeReg, body: "world!!"}, // 7 + }) + dir := t.TempDir() + n, err := Untar(context.Background(), bytes.NewReader(tarData), dir) + if err != nil { + t.Fatalf("Untar: %v", err) + } + if n != 12 { + t.Errorf("Untar bytes = %d, want 12", n) + } +} + +// TestEnsureRootfs_RecordsSizeFromUntar verifies the byte count Untar returns is +// threaded into the cache entry's size (rather than recomputed via a tree walk). +func TestEnsureRootfs_RecordsSizeFromUntar(t *testing.T) { + base := t.TempDir() + c, err := New(context.Background(), base, 0) + if err != nil { + t.Fatalf("New: %v", err) + } + tarData := buildTar(t, []struct { + name, body string + typeflag byte + mode int64 + }{ + {name: ".", typeflag: tar.TypeDir}, + {name: "f", typeflag: tar.TypeReg, body: "0123456789"}, // 10 + }) + if _, _, err := c.EnsureRootfs(context.Background(), testDigest, tarProviderFor(tarData)); err != nil { + t.Fatalf("EnsureRootfs: %v", err) + } + if got := c.Size(); got != 10 { + t.Errorf("Size = %d, want 10", got) + } +} + +// TestNew_EvictsWhenBootingOverBudget verifies that a Cache which loads an +// already-over-budget set of entries from disk reclaims space at startup, +// evicting the least-recently-used entry — not only on the next miss. +func TestNew_EvictsWhenBootingOverBudget(t *testing.T) { + base := t.TempDir() + older := time.Now().Add(-2 * time.Hour) + newer := time.Now().Add(-1 * time.Hour) + // Two ~4 KiB entries; a 6000-byte budget fits one but not both, so the + // older entry (evictDigest1) must be evicted at startup. + seedReadyEntryOnDisk(t, base, evictDigest1, 4096, older) + seedReadyEntryOnDisk(t, base, evictDigest2, 4096, newer) + + c, err := New(context.Background(), base, 6000) + if err != nil { + t.Fatalf("New: %v", err) + } + + // Startup eviction runs asynchronously; poll until it converges. + deadline := time.Now().Add(2 * time.Second) + for c.Size() > 6000 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if got := c.Size(); got > 6000 { + t.Fatalf("still over budget after startup eviction: size=%d", got) + } + if got := c.Count(); got != 1 { + t.Fatalf("count = %d, want 1 after startup eviction", got) + } + if c.LowerDir(evictDigest1) != "" { + t.Errorf("older digest1 still present; expected it evicted") + } + if c.LowerDir(evictDigest2) == "" { + t.Errorf("newer digest2 missing; expected it retained") + } +} + func TestValidateDigest(t *testing.T) { tests := []struct { input string diff --git a/cmd/atelet/oci.go b/cmd/atelet/oci.go index d00f7ca28..caa6e9e42 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -121,7 +121,7 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP } defer tarData.Close() - if err := rootfscache.Untar(ctx, tarData, rootPath); err != nil { + if _, err := rootfscache.Untar(ctx, tarData, rootPath); err != nil { return fmt.Errorf("in untar: %w", err) } @@ -318,7 +318,8 @@ func createMountPoint(rootPath, mountPath string) error { // that existing tests (package main) continue to compile without importing // the rootfscache package directly. func untar(ctx context.Context, tarData io.Reader, rootPath string) error { - return rootfscache.Untar(ctx, tarData, rootPath) + _, err := rootfscache.Untar(ctx, tarData, rootPath) + return err } // validateTarName is re-exported here for the same reason as untar: tests in From ebf1ea4c488c1e0dcb62a8cb18d9acbce9a0291d Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 17:31:39 +0800 Subject: [PATCH 8/9] feat: fall back to untar and retry when ateom reports overlay mount failure The overlay rootfs mount is performed by the privileged ateom worker after atelet has already returned from bundle preparation, so atelet cannot fall back in-band; it only learns of a mount failure through the RPC error. Introduce internal/overlayfallback as the cross-process contract: ateom wraps mount failures as a FailedPrecondition status carrying a marker that survives the server interceptor's status rebuild, and atelet recognizes it, re-prepares the bundles with a plain untar (clearing the overlay-lower marker), and retries the RPC once. This guarantees a working rootfs even when the overlay mount fails, addressing R5 (ensure fallback to untar on mount failure). --- cmd/atelet/main.go | 56 ++++++++++++-- cmd/atelet/oci.go | 27 ++++++- cmd/ateom-gvisor/main.go | 13 +++- internal/overlayfallback/overlayfallback.go | 65 +++++++++++++++++ .../overlayfallback/overlayfallback_test.go | 73 +++++++++++++++++++ 5 files changed, 218 insertions(+), 16 deletions(-) create mode 100644 internal/overlayfallback/overlayfallback.go create mode 100644 internal/overlayfallback/overlayfallback_test.go diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 4cf9a6f40..5489e9be0 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -34,6 +34,7 @@ import ( "github.com/agent-substrate/substrate/cmd/atelet/internal/rootfscache" "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/overlayfallback" "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/resources" @@ -249,7 +250,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*atele } if err := s.prepareOCIBundles(ctx, ns, tmpl, actorID, - req.GetSpec(), req.GetTargetAteomUid(), + req.GetSpec(), req.GetTargetAteomUid(), false, ); err != nil { return nil, err } @@ -261,15 +262,33 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*atele // Tell ateom to start the workload. gVisor uses RunscPath; the micro-VM // runtime uses the full RuntimeAssetPaths set. - if _, err := client.RunWorkload(ctx, &ateompb.RunWorkloadRequest{ + runReq := &ateompb.RunWorkloadRequest{ ActorTemplateNamespace: ns, ActorTemplateName: tmpl, ActorId: actorID, RunscPath: runscPathFor(assetPaths), RuntimeAssetPaths: assetPaths, Spec: buildAteomWorkloadSpec(req.GetSpec()), - }); err != nil { - return nil, fmt.Errorf("while calling ateom.RunWorkload: %w", err) + } + if _, err := client.RunWorkload(ctx, runReq); err != nil { + if !overlayfallback.IsMountFailure(err) { + return nil, fmt.Errorf("while calling ateom.RunWorkload: %w", err) + } + // ateom could not mount the overlay rootfs. It mounts after we returned + // from bundle preparation and holds no tar, so it cannot recover; we + // re-prepare the bundles as a plain untar (clearing the overlay marker) + // and retry once. The mount is ateom's first step and its failure defer + // tears down the network + any partial mounts, leaving a clean slate. + slog.WarnContext(ctx, "ateom overlay rootfs mount failed; falling back to untar and retrying", + slog.String("actor", actorID), slog.Any("err", err)) + if err := s.prepareOCIBundles(ctx, ns, tmpl, actorID, + req.GetSpec(), req.GetTargetAteomUid(), true, + ); err != nil { + return nil, fmt.Errorf("while re-preparing OCI bundles for untar fallback: %w", err) + } + if _, err := client.RunWorkload(ctx, runReq); err != nil { + return nil, fmt.Errorf("while calling ateom.RunWorkload after untar fallback: %w", err) + } } return &ateletpb.RunResponse{}, nil @@ -533,7 +552,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) return err } t := time.Now() - if err := s.prepareOCIBundles(gctx, ns, tmpl, actorID, req.GetSpec(), req.GetTargetAteomUid()); err != nil { + if err := s.prepareOCIBundles(gctx, ns, tmpl, actorID, req.GetSpec(), req.GetTargetAteomUid(), false); err != nil { return err } dBundles = time.Since(t) @@ -551,7 +570,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) // Tell ateom to do runsc create + runsc restore for pause container and // all application containers. tAteom := time.Now() - if _, err := client.RestoreWorkload(ctx, &ateompb.RestoreWorkloadRequest{ + restoreReq := &ateompb.RestoreWorkloadRequest{ ActorTemplateNamespace: ns, ActorTemplateName: tmpl, ActorId: actorID, @@ -559,8 +578,22 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) RuntimeAssetPaths: assetPaths, Spec: buildAteomWorkloadSpec(req.GetSpec()), Scope: toAteomSnapshotScope(req.GetScope()), - }); err != nil { - return nil, fmt.Errorf("while calling ateom.RestoreWorkload: %w", err) + } + if _, err := client.RestoreWorkload(ctx, restoreReq); err != nil { + if !overlayfallback.IsMountFailure(err) { + return nil, fmt.Errorf("while calling ateom.RestoreWorkload: %w", err) + } + // See RunWorkload: recover an overlay mount failure by re-preparing the + // bundles as a plain untar and retrying once. The image bytes are + // already downloaded, so this only re-unpacks the OCI bundle. + slog.WarnContext(ctx, "ateom overlay rootfs mount failed on restore; falling back to untar and retrying", + slog.String("actor", actorID), slog.Any("err", err)) + if err := s.prepareOCIBundles(ctx, ns, tmpl, actorID, req.GetSpec(), req.GetTargetAteomUid(), true); err != nil { + return nil, fmt.Errorf("while re-preparing OCI bundles for untar fallback: %w", err) + } + if _, err := client.RestoreWorkload(ctx, restoreReq); err != nil { + return nil, fmt.Errorf("while calling ateom.RestoreWorkload after untar fallback: %w", err) + } } dAteom = time.Since(tAteom) @@ -640,11 +673,16 @@ func (s *AteomHerder) downloadExternalCheckpoint(ctx context.Context, snapshotUr // prepareOCIBundles pulls images and assembles OCI bundles for the pause // container and every application container in spec, in parallel. +// +// forceUntar is threaded to prepareOCIDirectory: when true every container's +// rootfs is extracted directly rather than via the overlay cache. atelet sets +// it when re-preparing bundles after ateom reports an overlay mount failure. func (s *AteomHerder) prepareOCIBundles( ctx context.Context, actorTemplateNamespace, actorTemplateName, actorID string, spec *ateletpb.WorkloadSpec, targetAteomUid string, + forceUntar bool, ) error { netnsPath := ateompath.AteomNetNSPath(targetAteomUid) @@ -702,6 +740,7 @@ func (s *AteomHerder) prepareOCIBundles( netnsPath, "", // pause is sandbox infra; it gets no actor identity mount. nil, + forceUntar, ); err != nil { return fmt.Errorf("while creating pause OCI bundle: %w", err) } @@ -739,6 +778,7 @@ func (s *AteomHerder) prepareOCIBundles( netnsPath, identityDir, ddMounts, + forceUntar, ); err != nil { return fmt.Errorf("while creating %q OCI bundle: %w", ctr.GetName(), err) } diff --git a/cmd/atelet/oci.go b/cmd/atelet/oci.go index caa6e9e42..1ab15ab8d 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -55,7 +55,13 @@ const ( // the rootfs is materialized via an overlayfs mount over a node-local cache // instead of re-extracting the tarball — reducing per-restore latency from // seconds to sub-millisecond on cache hits. -func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryPullCache, rootfsCache *rootfscache.Cache, actorTemplateNamespace, actorTemplateName, actorID, containerName, ref string, args []string, env []string, annotations map[string]string, netns string, identityDir string, durableDirVolumeMounts []*ateletpb.VolumeMount) error { +// +// forceUntar bypasses the overlay path unconditionally and extracts the tarball +// directly. atelet sets it when retrying after ateom reports an overlay mount +// failure (see overlayfallback): the overlay mount is performed by ateom after +// this function has returned, so the only recovery is to re-prepare the bundle +// as a plain untar and retry the RPC. +func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryPullCache, rootfsCache *rootfscache.Cache, actorTemplateNamespace, actorTemplateName, actorID, containerName, ref string, args []string, env []string, annotations map[string]string, netns string, identityDir string, durableDirVolumeMounts []*ateletpb.VolumeMount, forceUntar bool) error { tracer := otel.Tracer("prepareOCIDirectory") ctx, span := tracer.Start(ctx, "prepareOCIDirectory") @@ -67,11 +73,12 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP // Try the overlayfs cache path first. This succeeds when: // 1. rootfsCache is non-nil, AND - // 2. the image ref includes a digest (@sha256:…). + // 2. the image ref includes a digest (@sha256:…), AND + // 3. we are not being forced onto the untar fallback. // On a cache hit, tarData is NOT consumed, so we can skip the untar // entirely. On a miss, the cache extracts and caches for next time. digest := extractDigestFromRef(ref) - if rootfsCache != nil && digest != "" { + if rootfsCache != nil && digest != "" && !forceUntar { // Pass Fetch as a lazy provider: on a cache hit EnsureRootfs returns the // on-disk lowerDir without invoking it, so we do NOT pull or extract the // image (and never buffer it in the memory pull cache) on the hot path. @@ -107,7 +114,19 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP span.SetAttributes(attribute.String("rootfs_method", "overlay")) } else { - // Fallback: no digest or no cache — extract directly (original path). + // Fallback: no digest, no cache, or forced onto untar after an overlay + // mount failure — extract directly (original path). + // + // Remove any overlay-lower marker a prior overlay attempt left behind. + // Its presence is ateom's signal to mount an overlay; on the untar + // fallback the rootfs is the real root and no overlay must be mounted, + // so the stale marker has to go or ateom would try to mount again on + // the retry. + markerPath := ateompath.OverlayLowerMarkerFile(actorTemplateNamespace, actorTemplateName, actorID, containerName) + if err := os.Remove(markerPath); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("while removing stale overlay lower marker: %w", err) + } + if err := os.RemoveAll(rootPath); err != nil { return fmt.Errorf("while clearing rootfs %q: %w", rootPath, err) } diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 39e8897d8..8cbdd8df3 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -32,6 +32,7 @@ import ( "github.com/agent-substrate/substrate/internal/ateinterceptors" "github.com/agent-substrate/substrate/internal/ateompath" "github.com/agent-substrate/substrate/internal/contextlogging" + "github.com/agent-substrate/substrate/internal/overlayfallback" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/readyz" "github.com/agent-substrate/substrate/internal/serverboot" @@ -208,9 +209,11 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload // per-bundle marker. atelet cannot mount(2) (its capabilities are dropped), // so the privileged ateom worker performs it here — in ateom's mount // namespace, which the runsc child shares. On failure the deferred - // unmount tears down any partial mounts. + // unmount tears down any partial mounts. The failure is wrapped as a + // recognizable status so atelet can fall back to per-restore untar and + // retry (the mount cannot be retried in-band here — atelet holds the tar). if err := mountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()); err != nil { - return nil, fmt.Errorf("while mounting overlay rootfs: %w", err) + return nil, overlayfallback.MountFailure(fmt.Errorf("while mounting overlay rootfs: %w", err)) } defer func() { if retErr != nil { @@ -406,9 +409,11 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore // Mount the overlayfs rootfs for any container atelet flagged via a // per-bundle marker, mirroring RunWorkload. The mount happens in ateom's // mount namespace (shared by the runsc child); on failure the deferred - // unmount tears down any partial mounts. + // unmount tears down any partial mounts. The failure is wrapped as a + // recognizable status so atelet can fall back to per-restore untar and + // retry. if err := mountWorkloadOverlays(ctx, req.GetActorTemplateNamespace(), req.GetActorTemplateName(), req.GetActorId(), req.GetSpec()); err != nil { - return nil, fmt.Errorf("while mounting overlay rootfs: %w", err) + return nil, overlayfallback.MountFailure(fmt.Errorf("while mounting overlay rootfs: %w", err)) } defer func() { if retErr != nil { diff --git a/internal/overlayfallback/overlayfallback.go b/internal/overlayfallback/overlayfallback.go new file mode 100644 index 000000000..61660f788 --- /dev/null +++ b/internal/overlayfallback/overlayfallback.go @@ -0,0 +1,65 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package overlayfallback defines the cross-process contract that lets atelet +// recognize an overlay rootfs mount failure reported by ateom and recover from +// it by re-preparing the OCI bundle with a plain untar (the pre-overlay path) +// and retrying the RPC. +// +// The overlay mount is performed by the privileged ateom worker, after atelet +// has already returned from bundle preparation, so atelet cannot fall back +// in-band: it only learns of the failure through the RPC error. To fall back +// safely atelet must distinguish "the overlay could not be mounted" (recover by +// untar) from every other RunWorkload/RestoreWorkload failure (do not retry). +// +// The signal travels as a gRPC status. ateom's server interceptor rebuilds +// returned status errors as status.Error(code, message), preserving only the +// code and message and discarding any structured details — so the marker must +// live in the message, guarded by a specific code to make a false positive +// from an unrelated FailedPrecondition error effectively impossible. +package overlayfallback + +import ( + "strings" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// marker is embedded in the status message of an overlay-mount-failure error. +// It survives the ateom interceptor's status.Error(code, message) rebuild, +// unlike status details. +const marker = "ate-overlay-mount-failed" + +// MountFailure wraps err as the gRPC status atelet recognizes as an overlay +// rootfs mount failure eligible for untar fallback. err's text is preserved for +// diagnosis; the code and marker are what atelet matches on. +func MountFailure(err error) error { + return status.Errorf(codes.FailedPrecondition, "%s: %v", marker, err) +} + +// IsMountFailure reports whether err, as received by the atelet gRPC client, +// was produced by MountFailure. It matches on both the FailedPrecondition code +// and the marker so an unrelated FailedPrecondition from either RPC is not +// mistaken for a recoverable overlay failure. +func IsMountFailure(err error) bool { + if err == nil { + return false + } + st, ok := status.FromError(err) + if !ok { + return false + } + return st.Code() == codes.FailedPrecondition && strings.Contains(st.Message(), marker) +} diff --git a/internal/overlayfallback/overlayfallback_test.go b/internal/overlayfallback/overlayfallback_test.go new file mode 100644 index 000000000..30ce650c6 --- /dev/null +++ b/internal/overlayfallback/overlayfallback_test.go @@ -0,0 +1,73 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package overlayfallback + +import ( + "errors" + "fmt" + "testing" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestIsMountFailure(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "nil", err: nil, want: false}, + {name: "plain error", err: errors.New("boom"), want: false}, + {name: "mount failure", err: MountFailure(errors.New("no such device")), want: true}, + {name: "mount failure wrapping wrapped err", err: MountFailure(fmt.Errorf("while mounting overlay rootfs: %w", errors.New("EPERM"))), want: true}, + { + name: "same code without marker", + err: status.Error(codes.FailedPrecondition, "some other precondition"), + want: false, + }, + { + name: "marker under wrong code", + err: status.Error(codes.Internal, marker), + want: false, + }, + {name: "unrelated internal error", err: status.Error(codes.Internal, "internal server error"), want: false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := IsMountFailure(tc.err); got != tc.want { + t.Errorf("IsMountFailure(%v) = %v, want %v", tc.err, got, tc.want) + } + }) + } +} + +// TestIsMountFailure_SurvivesInterceptorRebuild verifies the signal survives the +// ateom server interceptor, which reconstructs any status error as +// status.Error(code, message) — dropping details but keeping code and message. +// The marker lives in the message precisely so it survives this rebuild. +func TestIsMountFailure_SurvivesInterceptorRebuild(t *testing.T) { + orig := MountFailure(errors.New("mount(2): operation not permitted")) + + st, ok := status.FromError(orig) + if !ok { + t.Fatalf("MountFailure did not produce a status error") + } + rebuilt := status.Error(st.Code(), st.Message()) + + if !IsMountFailure(rebuilt) { + t.Errorf("IsMountFailure(rebuilt) = false, want true; rebuilt=%v", rebuilt) + } +} From 2aa165ad37e8b4a9c4cac2aa6a84e318066de44b Mon Sep 17 00:00:00 2001 From: chenggui53 Date: Mon, 6 Jul 2026 17:34:21 +0800 Subject: [PATCH 9/9] feat: make rootfs cache disk budget configurable via flag Add --rootfs-cache-max-bytes to atelet, defaulting to rootfscache.DefaultMaxCacheBytes (20 GiB), replacing the hardcoded 0. Behavior is unchanged by default; operators can now tune the node-local overlayfs rootfs cache budget without a rebuild. --- cmd/atelet/main.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 5489e9be0..5cf0688ef 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -67,6 +67,8 @@ var ( gcpAuthForImagePulls = pflag.Bool("gcp-auth-for-image-pulls", true, "Use GCP application default credentials mechanism.") localhostRegistryReplacement = pflag.String("localhost-registry-replacement", "", "The replacement registry endpoint for localhost and/or loopback IP addresses, useful for local development. for example kind-registry:5000") + rootfsCacheMaxBytes = pflag.Int64("rootfs-cache-max-bytes", rootfscache.DefaultMaxCacheBytes, "Maximum disk budget in bytes for the node-local overlayfs rootfs cache. When exceeded, least-recently-used entries are evicted.") + showVersion = pflag.Bool("version", false, "Print version and exit.") ) @@ -117,7 +119,7 @@ func main() { serverboot.Fatal(ctx, "Failed to create pull cache", err) } - rootfsDiskCache, err := rootfscache.New(ctx, ateompath.RootfsCacheDir, 0, rootfscache.WithInUseFunc(overlayLowerInUse)) + rootfsDiskCache, err := rootfscache.New(ctx, ateompath.RootfsCacheDir, *rootfsCacheMaxBytes, rootfscache.WithInUseFunc(overlayLowerInUse)) if err != nil { serverboot.Fatal(ctx, "Failed to create rootfs cache", err) }