diff --git a/cmd/atelet/internal/rootfscache/rootfscache.go b/cmd/atelet/internal/rootfscache/rootfscache.go new file mode 100644 index 000000000..1127e4647 --- /dev/null +++ b/cmd/atelet/internal/rootfscache/rootfscache.go @@ -0,0 +1,724 @@ +// 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" + +// 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 { + 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 + + // 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 + + // 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. 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 + } + 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, + } + for _, opt := range opts { + opt(c) + } + + // 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)) + } + + // 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 +} + +// 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 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 (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, tarProvider func() (io.ReadCloser, error)) (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. 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() + 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() + + // 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) + 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 +} + +// 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) { + 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), + ) + + 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) + } + + // 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 came from Untar, so we avoid a + // second full-tree walk here. + 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) { + // 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() + + var total int64 + for _, e := range c.entries { + total += e.sizeBytes + } + + for total > c.maxCacheBytes && len(c.entries) > 0 { + // 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 + } + + 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) + } +} + +// 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 + } + } + 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. 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 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 0, fmt.Errorf("in tarReader.Next: %w", err) + } + + name, skip, err := ValidateTarName(hdr.Name) + if err != nil { + return 0, 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 0, fmt.Errorf("while replacing existing path at %q before regular file: %w", name, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + 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 0, fmt.Errorf("while creating file %q: %w", name, err) + } + n, err := io.Copy(outFile, tarReader) + closeErr := outFile.Close() + if err != nil { + return 0, fmt.Errorf("while writing contents of %q from tar stream: %w", name, err) + } + if closeErr != nil { + 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 0, 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 0, fmt.Errorf("while replacing existing path at %q before symlink: %w", name, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + 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 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 0, fmt.Errorf("invalid hardlink target for %q: %w", name, err) + } + if linkSkip { + 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 0, fmt.Errorf("while replacing existing path at %q before hardlink: %w", name, err) + } + } else if !errors.Is(err, os.ErrNotExist) { + return 0, fmt.Errorf("while checking existing path at %q before hardlink: %w", name, err) + } + if err := root.Link(linkname, name); err != nil { + 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 0, fmt.Errorf("unhandled tar entry typeflag %q", tfStr) + } + } + + return totalBytes, 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..dd6994595 --- /dev/null +++ b/cmd/atelet/internal/rootfscache/rootfscache_test.go @@ -0,0 +1,616 @@ +// 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" + "io" + "os" + "path/filepath" + "sync" + "testing" + "time" +) + +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() +} + +// 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) + 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, tarProviderFor(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, tarProviderFor(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, tarProviderFor(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, tarProviderFor(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, tarProviderFor(tarData1)); err != nil { + t.Fatalf("EnsureRootfs d1: %v", err) + } + if _, _, err := c.EnsureRootfs(context.Background(), digest2, tarProviderFor(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) + } +} + +// 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) + 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, tarProviderFor(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")) + } +} + +// 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 + 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..5cf0688ef 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -31,8 +31,10 @@ 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/overlayfallback" "github.com/agent-substrate/substrate/internal/proto/ateletpb" "github.com/agent-substrate/substrate/internal/proto/ateompb" "github.com/agent-substrate/substrate/internal/resources" @@ -65,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.") ) @@ -115,6 +119,11 @@ func main() { serverboot.Fatal(ctx, "Failed to create pull cache", err) } + rootfsDiskCache, err := rootfscache.New(ctx, ateompath.RootfsCacheDir, *rootfsCacheMaxBytes, rootfscache.WithInUseFunc(overlayLowerInUse)) + 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 +172,7 @@ func main() { wrappedAnonGCS, wrappedGCS, pullCache, + rootfsDiskCache, ) lis, err := net.Listen("tcp", ":"+strconv.Itoa(*port)) @@ -186,6 +196,7 @@ type AteomHerder struct { ateomDialer *AteomDialer pullCache *memorypullcache.MemoryPullCache + rootfsCache *rootfscache.Cache anonGCSClient ategcs.ObjectStorage gcsClient ategcs.ObjectStorage } @@ -199,10 +210,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, } @@ -239,7 +252,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 } @@ -251,15 +264,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 @@ -523,7 +554,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) @@ -541,7 +572,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, @@ -549,8 +580,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) @@ -630,11 +675,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) @@ -682,6 +732,7 @@ func (s *AteomHerder) prepareOCIBundles( if err := prepareOCIDirectory( gCtx, s.pullCache, + s.rootfsCache, actorTemplateNamespace, actorTemplateName, actorID, "pause", spec.GetPauseImage(), @@ -691,6 +742,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) } @@ -714,6 +766,7 @@ func (s *AteomHerder) prepareOCIBundles( if err := prepareOCIDirectory( gCtx, s.pullCache, + s.rootfsCache, actorTemplateNamespace, actorTemplateName, actorID, ctr.GetName(), ctr.GetImage(), @@ -727,6 +780,7 @@ func (s *AteomHerder) prepareOCIBundles( netnsPath, identityDir, ddMounts, + forceUntar, ); err != nil { return fmt.Errorf("while creating %q OCI bundle: %w", ctr.GetName(), err) } @@ -948,6 +1002,13 @@ func resetActorDirs(actorTemplateNamespace, actorTemplateName, actorID string) e // Explicitly leave runsc logs dir untouched. bundleDir := ateompath.OCIBundleDir(actorTemplateNamespace, actorTemplateName, actorID) + + // 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 a00df99e9..1ab15ab8d 100644 --- a/cmd/atelet/oci.go +++ b/cmd/atelet/oci.go @@ -15,19 +15,16 @@ 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 +50,18 @@ 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. +// +// 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") @@ -63,31 +71,93 @@ 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:…), 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 != "" && !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. + // 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) + } - if err := os.MkdirAll(rootPath, 0o700); err != nil { - return fmt.Errorf("in os.MkdirAll for container bundle dir: %w", err) - } + // 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) + } + 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) + } - tarData, err := pullCache.Fetch(ctx, ref) - if err != nil { - return fmt.Errorf("in pullCache.Fetch: %w", err) - } - defer tarData.Close() + // 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, 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 := untar(ctx, tarData, rootPath); err != nil { - return fmt.Errorf("in untar: %w", err) - } + 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) + } - // 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) + 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) } + + // 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) @@ -103,6 +173,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 +333,16 @@ 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 - } - 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 -} - +// 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) - } - - } + _, err := rootfscache.Untar(ctx, tarData, rootPath) + return err +} - 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/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/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index 30010eacc..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" @@ -204,6 +205,22 @@ 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. 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, overlayfallback.MountFailure(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) @@ -256,6 +273,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) @@ -381,6 +406,21 @@ 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. 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, overlayfallback.MountFailure(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..caf115630 --- /dev/null +++ b/cmd/ateom-gvisor/overlay.go @@ -0,0 +1,130 @@ +//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) + + // 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) + } + } + + 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 d9beb2d6c..511b7475f 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) } @@ -68,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 @@ -114,6 +134,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 + } +} 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) + } +}