From 3f1ef8f7fa4721ea236c6adfabd3e30199f8bf7d Mon Sep 17 00:00:00 2001 From: dberkov Date: Mon, 31 Aug 2026 15:28:17 -0700 Subject: [PATCH 01/10] atelet: pre-download sandbox assets from SandboxConfigs Sandbox assets were fetched inside the first Run/Restore on a node, putting the gVisor release download+extract on that actor's critical path. atelet now watches SandboxConfig objects and pre-downloads their assets for the node's architecture in the background: gvisor configs on every node, microvm configs only where /dev/kvm exists (the same signal the device plugin advertises, and known at startup before any WorkerPool schedules here). Downloads are jittered by up to 30s so a config rollout does not hit the bucket from every node at once. Prewarming is best-effort and shares the content-addressed cache with the on-demand fetch, which remains the correctness path; racing the two is safe because both install files via atomic rename. Retry-with-backoff on prewarm failure and GC of unreferenced cached assets are left as TODOs. The atelet ClusterRole gains get/list/watch on sandboxconfigs. --- cmd/atelet/deviceplugin.go | 15 +++ cmd/atelet/main.go | 9 ++ cmd/atelet/sandbox_prewarm.go | 174 ++++++++++++++++++++++++++++ cmd/atelet/sandbox_prewarm_test.go | 179 +++++++++++++++++++++++++++++ manifests/ate-install/atelet.yaml | 5 + 5 files changed, 382 insertions(+) create mode 100644 cmd/atelet/sandbox_prewarm.go create mode 100644 cmd/atelet/sandbox_prewarm_test.go diff --git a/cmd/atelet/deviceplugin.go b/cmd/atelet/deviceplugin.go index 1cb74dc8b7..e65b9b2016 100644 --- a/cmd/atelet/deviceplugin.go +++ b/cmd/atelet/deviceplugin.go @@ -30,6 +30,21 @@ import ( // exist; workers are handed the real host paths, which kubelet resolves. const hostDevRoot = "/host/dev" +// microvmNodeCapable reports whether this node can host micro-VM workers: +// cloud-hypervisor needs /dev/kvm (VmCreate fails with EPERM without it), and +// worker pods request the matching extended resource, so they only schedule to +// nodes where the device exists. Device presence is therefore the earliest +// reliable eligibility signal — known at atelet startup, before any WorkerPool +// schedules here. +func microvmNodeCapable(devRoot string) bool { + for _, d := range deviceplugin.SandboxDevices { + if d.ResourceName == deviceplugin.ResourceKVM { + return d.Present(devRoot) + } + } + return false +} + // startDevicePlugins advertises the sandbox host devices present on this node to // kubelet as extended resources, in the background for the lifetime of ctx. This // is what lets a worker be granted /dev/kvm without running privileged. atelet diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index c1994865b3..4d8dcc2db2 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -279,6 +279,9 @@ func main() { // is unlikely to be used with frequency. ateFactory := externalversions.NewSharedInformerFactory(ateClient, 0) csiDriverConfigLister := ateFactory.Api().V1alpha1().CSIDriverConfigs().Lister() + // Requested before Start so the factory runs it; the prewarm handler is + // attached after the herder exists (see startSandboxAssetPrewarm below). + sandboxConfigInformer := ateFactory.Api().V1alpha1().SandboxConfigs().Informer() // Start an informer on the ClusterTrustBundle we care about (currently // only the egress trust bundle). The v1beta1 API is feature-gated: on a @@ -308,6 +311,12 @@ func main() { csiDriverConfigLister, clusterTrustBundleLister, ) + // Pre-download sandbox assets as SandboxConfigs appear/change so the first + // Run/Restore on this node hits the cache. Best-effort: on failure the + // on-demand fetch in ensureSandboxAssets still covers correctness. + if err := startSandboxAssetPrewarm(ctx, sandboxConfigInformer, wmService, microvmNodeCapable(hostDevRoot)); err != nil { + slog.ErrorContext(ctx, "Sandbox asset prewarm disabled", slog.Any("err", err)) + } dialOpts, err := ateapiauth.DialOptions(ateapiauth.ClientConfig{ K8sClient: k8sClient, CAFile: *ateapiCAFile, diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go new file mode 100644 index 0000000000..51432c446b --- /dev/null +++ b/cmd/atelet/sandbox_prewarm.go @@ -0,0 +1,174 @@ +// 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" + "fmt" + "log/slog" + "math/rand/v2" + "runtime" + "time" + + "k8s.io/client-go/tools/cache" + + "github.com/agent-substrate/substrate/pkg/api/v1alpha1" +) + +// prewarmMaxJitter spreads the fleet's asset downloads after a SandboxConfig +// change. Every atelet observes a create/update within about a second, and +// without jitter they would all open the same bucket objects at once. A var so +// tests can zero it. +var prewarmMaxJitter = 30 * time.Second + +// sandboxPrewarmer downloads SandboxConfig assets into the node's +// content-addressed static-files cache before any actor asks for them, so the +// fetch inside the first Run/Restore on the node is a cache hit instead of a +// download+extract on the critical path. +type sandboxPrewarmer struct { + herder *AteomHerder + // queue decouples informer event handlers (which must not block) from the + // downloads. A single worker drains it, which also serializes downloads so + // concurrent prewarms never compete for node bandwidth. + queue chan *v1alpha1.SandboxConfig + // microvmCapable gates micro-VM configs: their guest images run to + // hundreds of MiB, and a node without /dev/kvm can never run that class + // (workers request the ate.dev/kvm extended resource, so they only + // schedule where the device exists). See microvmNodeCapable. + microvmCapable bool +} + +// startSandboxAssetPrewarm registers an event handler on the SandboxConfig +// informer and starts a background worker that pre-downloads each config's +// sandbox assets for this node's architecture. Prewarming is purely a latency +// optimization: every failure is logged and left to the on-demand fetch in +// ensureSandboxAssets, which remains the correctness path. +// +// TODO: the static-files cache is never pruned, and prewarming every config +// revision makes stale releases accumulate faster. Add a GC that removes +// assets referenced by no current SandboxConfig and no on-node actor record. +func startSandboxAssetPrewarm(ctx context.Context, informer cache.SharedIndexInformer, herder *AteomHerder, microvmCapable bool) error { + p := &sandboxPrewarmer{ + herder: herder, + // SandboxConfigs are cluster-scoped and number a handful; 64 buffered + // events is far beyond any realistic burst. + queue: make(chan *v1alpha1.SandboxConfig, 64), + microvmCapable: microvmCapable, + } + // The handler is registered after the informer cache has synced, so it + // replays every existing SandboxConfig as a synthetic Add: a freshly booted + // node prewarms the current configs, not only future changes. + if _, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj any) { p.enqueue(ctx, obj) }, + UpdateFunc: func(_, obj any) { p.enqueue(ctx, obj) }, + }); err != nil { + return fmt.Errorf("while registering sandbox config prewarm handler: %w", err) + } + go p.run(ctx) + slog.InfoContext(ctx, "Sandbox asset prewarm started", slog.Bool("microvmCapable", microvmCapable)) + return nil +} + +func (p *sandboxPrewarmer) enqueue(ctx context.Context, obj any) { + cfg, ok := obj.(*v1alpha1.SandboxConfig) + if !ok { + return + } + switch cfg.Spec.SandboxClass { + case v1alpha1.SandboxClassGvisor: + // Every node runs gVisor workers; always prewarm. + case v1alpha1.SandboxClassMicroVM: + if !p.microvmCapable { + slog.DebugContext(ctx, "Skipping sandbox asset prewarm: node has no /dev/kvm, cannot run micro-VM workers", + slog.String("config", cfg.Name)) + return + } + default: + // An unknown class has no backend in this atelet (likely version skew + // with a newer control plane); nothing to prewarm. + slog.InfoContext(ctx, "Skipping sandbox asset prewarm: unknown sandbox class", + slog.String("config", cfg.Name), + slog.String("sandboxClass", string(cfg.Spec.SandboxClass))) + return + } + + select { + case p.queue <- cfg: + default: + // Best-effort: dropping an event only costs a download at first use. + slog.WarnContext(ctx, "Sandbox asset prewarm queue full; skipping config", slog.String("config", cfg.Name)) + } +} + +func (p *sandboxPrewarmer) run(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case cfg := <-p.queue: + if prewarmMaxJitter > 0 { + select { + case <-ctx.Done(): + return + case <-time.After(rand.N(prewarmMaxJitter)): + } + } + if err := p.prewarm(ctx, cfg); err != nil { + // TODO: retry with backoff (e.g. a rate-limited workqueue). + // Until then a transient failure leaves the asset cold until + // the next config event or first use. + slog.WarnContext(ctx, "Sandbox asset prewarm failed", slog.String("config", cfg.Name), slog.Any("err", err)) + } + } + } +} + +// prewarm fetches every asset of one SandboxConfig into the static-files +// cache. Racing an on-demand ensureSandboxAssets for the same assets is safe: +// both paths install content-addressed files via atomic rename. +func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxConfig) error { + rec, err := recordFromSandboxConfig(cfg) + if err != nil { + return err + } + t := time.Now() + if _, err := p.herder.ensureSandboxAssets(ctx, rec); err != nil { + return err + } + slog.InfoContext(ctx, "Sandbox assets prewarmed", + slog.String("config", cfg.Name), + slog.Int("assets", len(rec.Assets)), + slog.Duration("duration", time.Since(t))) + return nil +} + +// recordFromSandboxConfig projects a SandboxConfig's per-architecture assets +// onto the local node's architecture, mirroring recordFromRequest. +func recordFromSandboxConfig(cfg *v1alpha1.SandboxConfig) (*sandboxAssetsRecord, error) { + arch := runtime.GOARCH + files := cfg.Spec.Assets[arch] + if len(files) == 0 { + return nil, fmt.Errorf("sandbox config %q has no assets for architecture %q", cfg.Name, arch) + } + rec := &sandboxAssetsRecord{ + SandboxClass: string(cfg.Spec.SandboxClass), + PauseImage: cfg.Spec.PauseImage, + Assets: make(map[string]assetEntry, len(files)), + } + for name, f := range files { + rec.Assets[name] = assetEntry{URL: f.URL, SHA256: f.SHA256} + } + return rec, nil +} diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go new file mode 100644 index 0000000000..c998b11751 --- /dev/null +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -0,0 +1,179 @@ +// 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" + "crypto/sha256" + "errors" + "fmt" + "os" + "runtime" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" + + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + "github.com/agent-substrate/substrate/pkg/client/clientset/versioned/fake" + "github.com/agent-substrate/substrate/pkg/client/informers/externalversions" +) + +func gvisorConfig(name, url, sha string) *v1alpha1.SandboxConfig { + return &v1alpha1.SandboxConfig{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.SandboxConfigSpec{ + SandboxClass: v1alpha1.SandboxClassGvisor, + PauseImage: "registry.k8s.io/pause@sha256:abc", + Assets: map[string]map[string]v1alpha1.AssetFile{ + runtime.GOARCH: { + runscAssetName: {URL: url, SHA256: sha}, + }, + }, + }, + } +} + +func TestRecordFromSandboxConfig(t *testing.T) { + sha := fmt.Sprintf("%x", sha256.Sum256([]byte("runsc"))) + cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", sha) + + rec, err := recordFromSandboxConfig(cfg) + if err != nil { + t.Fatalf("recordFromSandboxConfig: %v", err) + } + if rec.SandboxClass != string(v1alpha1.SandboxClassGvisor) { + t.Errorf("SandboxClass = %q, want %q", rec.SandboxClass, v1alpha1.SandboxClassGvisor) + } + if rec.PauseImage != cfg.Spec.PauseImage { + t.Errorf("PauseImage = %q, want %q", rec.PauseImage, cfg.Spec.PauseImage) + } + want := assetEntry{URL: "gs://bucket/runsc", SHA256: sha} + if got := rec.Assets[runscAssetName]; got != want { + t.Errorf("Assets[%q] = %+v, want %+v", runscAssetName, got, want) + } + + // A config with no assets for this node's architecture cannot be projected. + cfg.Spec.Assets = map[string]map[string]v1alpha1.AssetFile{ + "other-arch": {runscAssetName: {URL: "gs://bucket/runsc", SHA256: sha}}, + } + if _, err := recordFromSandboxConfig(cfg); err == nil { + t.Error("recordFromSandboxConfig accepted a config with no assets for the local architecture") + } +} + +func TestPrewarmEnqueueFilters(t *testing.T) { + ctx := context.Background() + microvm := &v1alpha1.SandboxConfig{ + ObjectMeta: metav1.ObjectMeta{Name: "microvm-default"}, + Spec: v1alpha1.SandboxConfigSpec{SandboxClass: v1alpha1.SandboxClassMicroVM}, + } + gvisor := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("runsc")))) + + t.Run("node without KVM", func(t *testing.T) { + p := &sandboxPrewarmer{queue: make(chan *v1alpha1.SandboxConfig, 1)} + + p.enqueue(ctx, "not a sandbox config") + p.enqueue(ctx, microvm) + p.enqueue(ctx, &v1alpha1.SandboxConfig{ + ObjectMeta: metav1.ObjectMeta{Name: "future-class"}, + Spec: v1alpha1.SandboxConfigSpec{SandboxClass: "future-class"}, + }) + if len(p.queue) != 0 { + t.Fatalf("queue holds %d configs after filtered enqueues, want 0", len(p.queue)) + } + + p.enqueue(ctx, gvisor) + if len(p.queue) != 1 { + t.Fatalf("queue holds %d configs after gvisor enqueue, want 1", len(p.queue)) + } + // A full queue must drop rather than block the informer handler. + p.enqueue(ctx, gvisor) + if len(p.queue) != 1 { + t.Errorf("queue holds %d configs after enqueue on a full queue, want 1", len(p.queue)) + } + }) + + t.Run("node with KVM", func(t *testing.T) { + p := &sandboxPrewarmer{queue: make(chan *v1alpha1.SandboxConfig, 2), microvmCapable: true} + p.enqueue(ctx, microvm) + p.enqueue(ctx, gvisor) + if len(p.queue) != 2 { + t.Errorf("queue holds %d configs, want both microvm and gvisor queued", len(p.queue)) + } + }) +} + +// TestMicrovmNodeCapable covers the detectable negative cases; the positive +// case needs a /dev/kvm character device, which a test cannot mknod. +func TestMicrovmNodeCapable(t *testing.T) { + devRoot := t.TempDir() + if microvmNodeCapable(devRoot) { + t.Error("microvmNodeCapable = true for a dev root without kvm") + } + // A plain file named kvm is not a character device and must not count. + if err := os.WriteFile(devRoot+"/kvm", []byte("not a device"), 0o600); err != nil { + t.Fatal(err) + } + if microvmNodeCapable(devRoot) { + t.Error("microvmNodeCapable = true for a regular file named kvm") + } +} + +// TestSandboxAssetPrewarmDownloads runs the whole path: a SandboxConfig in a +// fake clientset flows through the informer into the prewarm worker, which +// lands the asset in the static-files cache without any Run/Restore request. +func TestSandboxAssetPrewarmDownloads(t *testing.T) { + origDir, origJitter := ateompath.StaticFilesDir, prewarmMaxJitter + ateompath.StaticFilesDir = t.TempDir() + prewarmMaxJitter = 0 + t.Cleanup(func() { ateompath.StaticFilesDir, prewarmMaxJitter = origDir, origJitter }) + + content := []byte("runsc binary bytes") + sha := fmt.Sprintf("%x", sha256.Sum256(content)) + cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", sha) + + ctx := t.Context() + client := fake.NewSimpleClientset(cfg) + factory := externalversions.NewSharedInformerFactory(client, 0) + informer := factory.Api().V1alpha1().SandboxConfigs().Informer() + stopCh := make(chan struct{}) + defer close(stopCh) + factory.Start(stopCh) + if !cache.WaitForCacheSync(stopCh, informer.HasSynced) { + t.Fatal("informer cache never synced") + } + + herder := &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}} + if err := startSandboxAssetPrewarm(ctx, informer, herder, false); err != nil { + t.Fatalf("startSandboxAssetPrewarm: %v", err) + } + + wantPath := ateompath.RunSCBinaryPath(sha) + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := os.Stat(wantPath); err == nil { + return + } else if !errors.Is(err, os.ErrNotExist) { + t.Fatalf("stat %s: %v", wantPath, err) + } + if time.Now().After(deadline) { + t.Fatalf("asset never prewarmed to %s", wantPath) + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/manifests/ate-install/atelet.yaml b/manifests/ate-install/atelet.yaml index 573501a0d3..9dcaa07fc0 100644 --- a/manifests/ate-install/atelet.yaml +++ b/manifests/ate-install/atelet.yaml @@ -31,6 +31,11 @@ rules: - apiGroups: ["ate.dev"] resources: ["csidriverconfigs"] verbs: ["get", "list", "watch"] +# SandboxConfigs are watched to pre-download sandbox assets into the node's +# cache before the first actor needs them (see cmd/atelet/sandbox_prewarm.go). +- apiGroups: ["ate.dev"] + resources: ["sandboxconfigs"] + verbs: ["get", "list", "watch"] # ClusterTrustBundles referenced by SystemInfo trustBundle data sources are # resolved on the node: atelet reads them through an informer and projects # the sanitized PEM into actors (see cmd/atelet/trustbundle.go). From 0c7b71e68c8783dd743e9abfbd25c5c31c6a8abe Mon Sep 17 00:00:00 2001 From: dberkov Date: Tue, 1 Sep 2026 05:57:19 -0700 Subject: [PATCH 02/10] atelet: prewarm the pause image alongside sandbox assets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sandbox asset prewarmer downloaded the runtime binaries but not the SandboxConfig's pause image, so every node pulled it from the registry inside its first Run/Restore. At benchmark start that is a synchronized fleet-wide pull burst — registry.k8s.io answered a 1000-node run with 429s, failing Restores — and a failed pull writes no cache record, so each subsequent Restore on the node re-pinged the registry and fed the rate limiter. The prewarm worker now pulls the pause image into the image cache concurrently with the asset downloads (they live in different backends, so neither fetch waits on or fails the other), inheriting the existing per-config jitter that spreads the fleet's pulls after a config rollout. Prewarming remains best-effort: the pull inside prepareOCIBundles is still the correctness path. --- cmd/atelet/sandbox_prewarm.go | 39 +++++++++-- cmd/atelet/sandbox_prewarm_test.go | 106 ++++++++++++++++++++++++++++- 2 files changed, 138 insertions(+), 7 deletions(-) diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index 51432c446b..cf7de66872 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -16,10 +16,12 @@ package main import ( "context" + "errors" "fmt" "log/slog" "math/rand/v2" "runtime" + "sync" "time" "k8s.io/client-go/tools/cache" @@ -52,9 +54,10 @@ type sandboxPrewarmer struct { // startSandboxAssetPrewarm registers an event handler on the SandboxConfig // informer and starts a background worker that pre-downloads each config's -// sandbox assets for this node's architecture. Prewarming is purely a latency -// optimization: every failure is logged and left to the on-demand fetch in -// ensureSandboxAssets, which remains the correctness path. +// sandbox assets for this node's architecture, and pulls its pause image into +// the image cache. Prewarming is purely a latency optimization: every failure +// is logged and left to the on-demand fetch in ensureSandboxAssets and the +// pull inside prepareOCIBundles, which remain the correctness path. // // TODO: the static-files cache is never pruned, and prewarming every config // revision makes stale releases accumulate faster. Add a GC that removes @@ -136,20 +139,44 @@ func (p *sandboxPrewarmer) run(ctx context.Context) { } // prewarm fetches every asset of one SandboxConfig into the static-files -// cache. Racing an on-demand ensureSandboxAssets for the same assets is safe: -// both paths install content-addressed files via atomic rename. +// cache and its pause image into the image cache. Racing an on-demand +// ensureSandboxAssets for the same assets is safe: both paths install +// content-addressed files via atomic rename, and the image cache collapses +// concurrent pulls of one digest. func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxConfig) error { rec, err := recordFromSandboxConfig(cfg) if err != nil { return err } t := time.Now() - if _, err := p.herder.ensureSandboxAssets(ctx, rec); err != nil { + // The pause image is a sandbox prerequisite like the runtime binaries: it + // is the root container of every actor, and without prewarming each node + // pulls it inside its first Run/Restore — which at fleet scale is a + // synchronized stampede on the image registry. The two fetches live in + // different backends (bucket vs. registry), so they run concurrently and + // fail independently: either being warm still shortens the first actor's + // critical path. + var assetErr, imageErr error + var wg sync.WaitGroup + wg.Go(func() { + _, assetErr = p.herder.ensureSandboxAssets(ctx, rec) + }) + wg.Go(func() { + if rec.PauseImage == "" { + return + } + if _, err := p.herder.imageCache.EnsureImage(ctx, rec.PauseImage); err != nil { + imageErr = fmt.Errorf("while prewarming pause image %q: %w", rec.PauseImage, err) + } + }) + wg.Wait() + if err := errors.Join(assetErr, imageErr); err != nil { return err } slog.InfoContext(ctx, "Sandbox assets prewarmed", slog.String("config", cfg.Name), slog.Int("assets", len(rec.Assets)), + slog.String("pauseImage", rec.PauseImage), slog.Duration("duration", time.Since(t))) return nil } diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index c998b11751..5aeafb0781 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -19,6 +19,10 @@ import ( "crypto/sha256" "errors" "fmt" + "io" + "log" + "net/http/httptest" + "net/url" "os" "runtime" "testing" @@ -28,11 +32,40 @@ import ( "k8s.io/client-go/tools/cache" "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/imagecache" "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/client/clientset/versioned/fake" "github.com/agent-substrate/substrate/pkg/client/informers/externalversions" + "github.com/google/go-containerregistry/pkg/name" + "github.com/google/go-containerregistry/pkg/registry" + v1 "github.com/google/go-containerregistry/pkg/v1" + "github.com/google/go-containerregistry/pkg/v1/empty" + "github.com/google/go-containerregistry/pkg/v1/mutate" + "github.com/google/go-containerregistry/pkg/v1/remote" ) +// pushPauseImage pushes a tiny image to ref and returns its manifest digest, +// so tests can later assert a cache hit by digest with the registry gone. +func pushPauseImage(t *testing.T, ref string) v1.Hash { + t.Helper() + img, err := mutate.AppendLayers(empty.Image, singleFileLayer(t, "pause", "pause bytes")) + if err != nil { + t.Fatalf("mutate.AppendLayers: %v", err) + } + digest, err := img.Digest() + if err != nil { + t.Fatalf("img.Digest: %v", err) + } + tag, err := name.ParseReference(ref, name.Insecure) + if err != nil { + t.Fatalf("name.ParseReference(%q): %v", ref, err) + } + if err := remote.Write(tag, img); err != nil { + t.Fatalf("remote.Write(%q): %v", ref, err) + } + return digest +} + func gvisorConfig(name, url, sha string) *v1alpha1.SandboxConfig { return &v1alpha1.SandboxConfig{ ObjectMeta: metav1.ObjectMeta{Name: name}, @@ -134,6 +167,68 @@ func TestMicrovmNodeCapable(t *testing.T) { } } +// TestPrewarmPauseImage covers the pause-image half of prewarm: the image +// lands in the image cache, and an asset fetch failure does not stop the +// pull (the two live in different backends). +func TestPrewarmPauseImage(t *testing.T) { + origDir := ateompath.StaticFilesDir + ateompath.StaticFilesDir = t.TempDir() + t.Cleanup(func() { ateompath.StaticFilesDir = origDir }) + + srv := httptest.NewServer(registry.New(registry.Logger(log.New(io.Discard, "", 0)))) + defer srv.Close() + u, err := url.Parse(srv.URL) + if err != nil { + t.Fatalf("parsing registry URL: %v", err) + } + pauseRef := u.Host + "/pause:3.10" + pauseDigest := pushPauseImage(t, pauseRef) + + newStore := func() *imagecache.Store { + s, err := imagecache.New(t.TempDir()) + if err != nil { + t.Fatalf("imagecache.New: %v", err) + } + return s + } + okStore, failStore := newStore(), newStore() + + ctx := context.Background() + content := []byte("runsc binary bytes") + cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256(content))) + cfg.Spec.PauseImage = pauseRef + + p := &sandboxPrewarmer{herder: &AteomHerder{ + anonGCSClient: fakeObjectStorage{data: content}, + imageCache: okStore, + }} + if err := p.prewarm(ctx, cfg); err != nil { + t.Fatalf("prewarm: %v", err) + } + + // A different asset hash misses the shared static-files cache, so the + // failing object storage is actually consulted — and must not keep the + // pause image from being pulled. + failCfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("other runsc")))) + failCfg.Spec.PauseImage = pauseRef + p = &sandboxPrewarmer{herder: &AteomHerder{ + anonGCSClient: fakeObjectStorage{err: errors.New("bucket unavailable")}, + imageCache: failStore, + }} + if err := p.prewarm(ctx, failCfg); err == nil { + t.Error("prewarm returned nil despite the asset fetch failing") + } + + // With the registry gone, only a cache hit can satisfy a digest ref. + srv.Close() + digestRef := u.Host + "/pause@" + pauseDigest.String() + for name, store := range map[string]*imagecache.Store{"ok": okStore, "asset-failure": failStore} { + if _, err := store.EnsureImage(ctx, digestRef); err != nil { + t.Errorf("pause image not prewarmed into the %s store: %v", name, err) + } + } +} + // TestSandboxAssetPrewarmDownloads runs the whole path: a SandboxConfig in a // fake clientset flows through the informer into the prewarm worker, which // lands the asset in the static-files cache without any Run/Restore request. @@ -143,9 +238,14 @@ func TestSandboxAssetPrewarmDownloads(t *testing.T) { prewarmMaxJitter = 0 t.Cleanup(func() { ateompath.StaticFilesDir, prewarmMaxJitter = origDir, origJitter }) + host := imageVolumeTestRegistry(t) + pauseRef := host + "/pause:3.10" + pushPauseImage(t, pauseRef) + content := []byte("runsc binary bytes") sha := fmt.Sprintf("%x", sha256.Sum256(content)) cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", sha) + cfg.Spec.PauseImage = pauseRef ctx := t.Context() client := fake.NewSimpleClientset(cfg) @@ -158,7 +258,11 @@ func TestSandboxAssetPrewarmDownloads(t *testing.T) { t.Fatal("informer cache never synced") } - herder := &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}} + store, err := imagecache.New(t.TempDir()) + if err != nil { + t.Fatalf("imagecache.New: %v", err) + } + herder := &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}, imageCache: store} if err := startSandboxAssetPrewarm(ctx, informer, herder, false); err != nil { t.Fatalf("startSandboxAssetPrewarm: %v", err) } From 224f35a2a39a09a8ddb91b2e8d8cbc2a461ad686 Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 09:08:46 -0700 Subject: [PATCH 03/10] atelet: dedupe sandbox prewarm work with a rate-limited workqueue A relist or event burst used to stack duplicate SandboxConfigs in the buffered channel, each copy costing its own jitter wait and fetch pass. Queue config names in a rate-limited workqueue instead: duplicate events coalesce into one entry, the worker resolves the name to the config's latest revision through the informer's lister at processing time, and a failed prewarm retries with download-scale backoff (1s to 5min, 8 attempts) instead of staying cold until the next config event. The per-config jitter moves from a sleep inside the worker to the enqueue delay, so one config's jitter window no longer serializes unrelated configs, and duplicates arriving inside the window collapse too. The queue-full drop path is gone: the workqueue is unbounded but deduped, so a handful of cluster-scoped configs can never grow it. --- cmd/atelet/sandbox_prewarm.go | 116 ++++++++++++++++++++--------- cmd/atelet/sandbox_prewarm_test.go | 60 ++++++++++++--- 2 files changed, 131 insertions(+), 45 deletions(-) diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index cf7de66872..c0ed75a12d 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -24,27 +24,44 @@ import ( "sync" "time" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/client-go/tools/cache" + "k8s.io/client-go/util/workqueue" "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" ) // prewarmMaxJitter spreads the fleet's asset downloads after a SandboxConfig // change. Every atelet observes a create/update within about a second, and -// without jitter they would all open the same bucket objects at once. A var so -// tests can zero it. +// without jitter they would all open the same bucket objects at once. The +// jitter is applied as the enqueue delay, so duplicate events arriving inside +// the window collapse into one queue entry instead of stacking waits in the +// worker. A var so tests can zero it. var prewarmMaxJitter = 30 * time.Second +// prewarmMaxRetries bounds how often one config's failed prewarm is retried +// before it is dropped until the next config event (or first use, which stays +// the correctness path). The backoff below caps at 5 minutes, so this covers +// transient bucket or registry outages of several minutes without hammering a +// permanently broken config forever. +const prewarmMaxRetries = 8 + // sandboxPrewarmer downloads SandboxConfig assets into the node's // content-addressed static-files cache before any actor asks for them, so the // fetch inside the first Run/Restore on the node is a cache hit instead of a // download+extract on the critical path. type sandboxPrewarmer struct { herder *AteomHerder + // lister resolves a queued config name to its latest revision at + // processing time, so coalesced events never prewarm a stale spec. + lister listersv1alpha1.SandboxConfigLister // queue decouples informer event handlers (which must not block) from the - // downloads. A single worker drains it, which also serializes downloads so - // concurrent prewarms never compete for node bandwidth. - queue chan *v1alpha1.SandboxConfig + // downloads, dedupes by config name so relists cannot stack duplicate + // work, and rate-limits retries after failures. A single worker drains + // it, which also serializes downloads so concurrent prewarms never + // compete for node bandwidth. + queue workqueue.TypedRateLimitingInterface[string] // microvmCapable gates micro-VM configs: their guest images run to // hundreds of MiB, and a node without /dev/kvm can never run that class // (workers request the ate.dev/kvm extended resource, so they only @@ -52,6 +69,19 @@ type sandboxPrewarmer struct { microvmCapable bool } +func newSandboxPrewarmer(herder *AteomHerder, lister listersv1alpha1.SandboxConfigLister, microvmCapable bool) *sandboxPrewarmer { + return &sandboxPrewarmer{ + herder: herder, + lister: lister, + // Downloads fail on the scale of network timeouts, not API conflicts, + // so back off in seconds and cap in minutes rather than the + // millisecond-based controller default. + queue: workqueue.NewTypedRateLimitingQueue( + workqueue.NewTypedItemExponentialFailureRateLimiter[string](time.Second, 5*time.Minute)), + microvmCapable: microvmCapable, + } +} + // startSandboxAssetPrewarm registers an event handler on the SandboxConfig // informer and starts a background worker that pre-downloads each config's // sandbox assets for this node's architecture, and pulls its pause image into @@ -63,13 +93,7 @@ type sandboxPrewarmer struct { // revision makes stale releases accumulate faster. Add a GC that removes // assets referenced by no current SandboxConfig and no on-node actor record. func startSandboxAssetPrewarm(ctx context.Context, informer cache.SharedIndexInformer, herder *AteomHerder, microvmCapable bool) error { - p := &sandboxPrewarmer{ - herder: herder, - // SandboxConfigs are cluster-scoped and number a handful; 64 buffered - // events is far beyond any realistic burst. - queue: make(chan *v1alpha1.SandboxConfig, 64), - microvmCapable: microvmCapable, - } + p := newSandboxPrewarmer(herder, listersv1alpha1.NewSandboxConfigLister(informer.GetIndexer()), microvmCapable) // The handler is registered after the informer cache has synced, so it // replays every existing SandboxConfig as a synthetic Add: a freshly booted // node prewarms the current configs, not only future changes. @@ -106,36 +130,60 @@ func (p *sandboxPrewarmer) enqueue(ctx context.Context, obj any) { slog.String("sandboxClass", string(cfg.Spec.SandboxClass))) return } - - select { - case p.queue <- cfg: - default: - // Best-effort: dropping an event only costs a download at first use. - slog.WarnContext(ctx, "Sandbox asset prewarm queue full; skipping config", slog.String("config", cfg.Name)) + if prewarmMaxJitter > 0 { + p.queue.AddAfter(cfg.Name, rand.N(prewarmMaxJitter)) + return } + p.queue.Add(cfg.Name) } func (p *sandboxPrewarmer) run(ctx context.Context) { + go func() { + <-ctx.Done() + p.queue.ShutDown() + }() for { - select { - case <-ctx.Done(): + name, shutdown := p.queue.Get() + if shutdown { return - case cfg := <-p.queue: - if prewarmMaxJitter > 0 { - select { - case <-ctx.Done(): - return - case <-time.After(rand.N(prewarmMaxJitter)): - } - } - if err := p.prewarm(ctx, cfg); err != nil { - // TODO: retry with backoff (e.g. a rate-limited workqueue). - // Until then a transient failure leaves the asset cold until - // the next config event or first use. - slog.WarnContext(ctx, "Sandbox asset prewarm failed", slog.String("config", cfg.Name), slog.Any("err", err)) - } } + p.process(ctx, name) + } +} + +// process prewarms the named config's current revision, requeueing with +// backoff on failure. The queue holds names, not objects, so an event that +// arrives while its config is being processed is simply requeued by the +// workqueue and prewarms the newer revision afterwards. +func (p *sandboxPrewarmer) process(ctx context.Context, name string) { + defer p.queue.Done(name) + cfg, err := p.lister.Get(name) + if apierrors.IsNotFound(err) { + // Deleted since it was enqueued; nothing to prewarm anymore. + p.queue.Forget(name) + return + } + if err == nil { + err = p.prewarm(ctx, cfg) + } + if err == nil { + p.queue.Forget(name) + return + } + if ctx.Err() != nil { + // Shutting down, not a prewarm failure; drop without retry noise. + return + } + if retries := p.queue.NumRequeues(name); retries < prewarmMaxRetries { + slog.WarnContext(ctx, "Sandbox asset prewarm failed; will retry", + slog.String("config", name), slog.Int("retries", retries), slog.Any("err", err)) + p.queue.AddRateLimited(name) + return } + // Best-effort: give up until the next config event or first use. + slog.WarnContext(ctx, "Sandbox asset prewarm failed; giving up", + slog.String("config", name), slog.Int("retries", prewarmMaxRetries), slog.Any("err", err)) + p.queue.Forget(name) } // prewarm fetches every asset of one SandboxConfig into the static-files diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index 5aeafb0781..53e681f247 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -36,6 +36,7 @@ import ( "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/client/clientset/versioned/fake" "github.com/agent-substrate/substrate/pkg/client/informers/externalversions" + listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" "github.com/google/go-containerregistry/pkg/name" "github.com/google/go-containerregistry/pkg/registry" v1 "github.com/google/go-containerregistry/pkg/v1" @@ -110,6 +111,10 @@ func TestRecordFromSandboxConfig(t *testing.T) { } func TestPrewarmEnqueueFilters(t *testing.T) { + origJitter := prewarmMaxJitter + prewarmMaxJitter = 0 // enqueue synchronously so Len is observable + t.Cleanup(func() { prewarmMaxJitter = origJitter }) + ctx := context.Background() microvm := &v1alpha1.SandboxConfig{ ObjectMeta: metav1.ObjectMeta{Name: "microvm-default"}, @@ -118,7 +123,7 @@ func TestPrewarmEnqueueFilters(t *testing.T) { gvisor := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("runsc")))) t.Run("node without KVM", func(t *testing.T) { - p := &sandboxPrewarmer{queue: make(chan *v1alpha1.SandboxConfig, 1)} + p := newSandboxPrewarmer(nil, nil, false) p.enqueue(ctx, "not a sandbox config") p.enqueue(ctx, microvm) @@ -126,31 +131,64 @@ func TestPrewarmEnqueueFilters(t *testing.T) { ObjectMeta: metav1.ObjectMeta{Name: "future-class"}, Spec: v1alpha1.SandboxConfigSpec{SandboxClass: "future-class"}, }) - if len(p.queue) != 0 { - t.Fatalf("queue holds %d configs after filtered enqueues, want 0", len(p.queue)) + if p.queue.Len() != 0 { + t.Fatalf("queue holds %d configs after filtered enqueues, want 0", p.queue.Len()) } p.enqueue(ctx, gvisor) - if len(p.queue) != 1 { - t.Fatalf("queue holds %d configs after gvisor enqueue, want 1", len(p.queue)) + if p.queue.Len() != 1 { + t.Fatalf("queue holds %d configs after gvisor enqueue, want 1", p.queue.Len()) } - // A full queue must drop rather than block the informer handler. + // Duplicate events (e.g. a relist) must coalesce, not stack work. p.enqueue(ctx, gvisor) - if len(p.queue) != 1 { - t.Errorf("queue holds %d configs after enqueue on a full queue, want 1", len(p.queue)) + if p.queue.Len() != 1 { + t.Errorf("queue holds %d configs after duplicate enqueue, want 1", p.queue.Len()) } }) t.Run("node with KVM", func(t *testing.T) { - p := &sandboxPrewarmer{queue: make(chan *v1alpha1.SandboxConfig, 2), microvmCapable: true} + p := newSandboxPrewarmer(nil, nil, true) p.enqueue(ctx, microvm) p.enqueue(ctx, gvisor) - if len(p.queue) != 2 { - t.Errorf("queue holds %d configs, want both microvm and gvisor queued", len(p.queue)) + if p.queue.Len() != 2 { + t.Errorf("queue holds %d configs, want both microvm and gvisor queued", p.queue.Len()) } }) } +// TestPrewarmProcessRetries covers the worker's failure handling: a failed +// prewarm is requeued with backoff, and a config deleted between enqueue and +// processing is forgotten without retries. +func TestPrewarmProcessRetries(t *testing.T) { + origDir := ateompath.StaticFilesDir + ateompath.StaticFilesDir = t.TempDir() + t.Cleanup(func() { ateompath.StaticFilesDir = origDir }) + + ctx := context.Background() + cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("runsc")))) + // No pause image, so a failing prewarm exercises only the asset path. + cfg.Spec.PauseImage = "" + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := indexer.Add(cfg); err != nil { + t.Fatalf("indexer.Add: %v", err) + } + herder := &AteomHerder{anonGCSClient: fakeObjectStorage{err: errors.New("bucket unavailable")}} + p := newSandboxPrewarmer(herder, listersv1alpha1.NewSandboxConfigLister(indexer), false) + + p.queue.Add(cfg.Name) + name, _ := p.queue.Get() + p.process(ctx, name) + if got := p.queue.NumRequeues(cfg.Name); got != 1 { + t.Errorf("NumRequeues after failed prewarm = %d, want 1 (requeued with backoff)", got) + } + + p.process(ctx, "deleted-config") + if got := p.queue.NumRequeues("deleted-config"); got != 0 { + t.Errorf("NumRequeues for a deleted config = %d, want 0 (forgotten)", got) + } +} + // TestMicrovmNodeCapable covers the detectable negative cases; the positive // case needs a /dev/kvm character device, which a test cannot mknod. func TestMicrovmNodeCapable(t *testing.T) { From 8ec2d95d246540c003a66719de1829a06741a986 Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 09:14:48 -0700 Subject: [PATCH 04/10] atelet: bound each prewarm attempt and decouple the pause image pull MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A prewarm attempt ran on the daemon's never-cancelled context, so a download that hung without failing (a registry that accepts the connection and stalls, a wedged bucket read) would block the single prewarm worker — and with it every other queued config — forever. Each attempt now runs under a 5-minute timeout; a timed-out attempt requeues with backoff like any other failure, and other configs process between its retries. The pause image pull no longer depends on the per-architecture asset projection: it needs no arch lookup, so it is scheduled first and runs even when the config has no assets for the node's architecture. The projection error is still returned so the failure stays visible. --- cmd/atelet/sandbox_prewarm.go | 53 ++++++++++++++++-------------- cmd/atelet/sandbox_prewarm_test.go | 52 +++++++++++++++++++++++++++-- 2 files changed, 79 insertions(+), 26 deletions(-) diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index c0ed75a12d..a51e27a7b6 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -40,6 +40,15 @@ import ( // worker. A var so tests can zero it. var prewarmMaxJitter = 30 * time.Second +// prewarmTimeout bounds a single prewarm attempt. The queue is drained by one +// worker, so without a deadline a download that hangs without failing (a +// registry that accepts the connection and stalls, a wedged bucket read) +// would block every other config forever — prewarm runs on the daemon's +// never-cancelled context. Generous enough for multi-hundred-MiB micro-VM +// guest images on a busy node; a timed-out attempt requeues with backoff like +// any other failure. A var so tests can shorten it. +var prewarmTimeout = 5 * time.Minute + // prewarmMaxRetries bounds how often one config's failed prewarm is retried // before it is dropped until the next config event (or first use, which stays // the correctness path). The backoff below caps at 5 minutes, so this covers @@ -192,31 +201,27 @@ func (p *sandboxPrewarmer) process(ctx context.Context, name string) { // content-addressed files via atomic rename, and the image cache collapses // concurrent pulls of one digest. func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxConfig) error { - rec, err := recordFromSandboxConfig(cfg) - if err != nil { - return err - } + ctx, cancel := context.WithTimeout(ctx, prewarmTimeout) + defer cancel() t := time.Now() - // The pause image is a sandbox prerequisite like the runtime binaries: it - // is the root container of every actor, and without prewarming each node - // pulls it inside its first Run/Restore — which at fleet scale is a - // synchronized stampede on the image registry. The two fetches live in - // different backends (bucket vs. registry), so they run concurrently and - // fail independently: either being warm still shortens the first actor's - // critical path. - var assetErr, imageErr error + + var imageErr error var wg sync.WaitGroup - wg.Go(func() { - _, assetErr = p.herder.ensureSandboxAssets(ctx, rec) - }) - wg.Go(func() { - if rec.PauseImage == "" { - return - } - if _, err := p.herder.imageCache.EnsureImage(ctx, rec.PauseImage); err != nil { - imageErr = fmt.Errorf("while prewarming pause image %q: %w", rec.PauseImage, err) - } - }) + // schedule prewarm pause image if provided + if cfg.Spec.PauseImage != "" { + wg.Go(func() { + if _, err := p.herder.imageCache.EnsureImage(ctx, cfg.Spec.PauseImage); err != nil { + imageErr = fmt.Errorf("while prewarming pause image %q: %w", cfg.Spec.PauseImage, err) + } + }) + } + // schedule prewarm sandbox assets if provided + rec, assetErr := recordFromSandboxConfig(cfg) + if assetErr == nil { + wg.Go(func() { + _, assetErr = p.herder.ensureSandboxAssets(ctx, rec) + }) + } wg.Wait() if err := errors.Join(assetErr, imageErr); err != nil { return err @@ -224,7 +229,7 @@ func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxCon slog.InfoContext(ctx, "Sandbox assets prewarmed", slog.String("config", cfg.Name), slog.Int("assets", len(rec.Assets)), - slog.String("pauseImage", rec.PauseImage), + slog.String("pauseImage", cfg.Spec.PauseImage), slog.Duration("duration", time.Since(t))) return nil } diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index 53e681f247..f7d83a57cf 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -229,7 +229,7 @@ func TestPrewarmPauseImage(t *testing.T) { } return s } - okStore, failStore := newStore(), newStore() + okStore, failStore, archStore := newStore(), newStore(), newStore() ctx := context.Background() content := []byte("runsc binary bytes") @@ -257,16 +257,64 @@ func TestPrewarmPauseImage(t *testing.T) { t.Error("prewarm returned nil despite the asset fetch failing") } + // A config with no assets for this node's architecture still gets its + // pause image pulled: the image needs no per-architecture projection. + archCfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256(content))) + archCfg.Spec.Assets = map[string]map[string]v1alpha1.AssetFile{ + "other-arch": {runscAssetName: archCfg.Spec.Assets[runtime.GOARCH][runscAssetName]}, + } + archCfg.Spec.PauseImage = pauseRef + p = &sandboxPrewarmer{herder: &AteomHerder{imageCache: archStore}} + if err := p.prewarm(ctx, archCfg); err == nil { + t.Error("prewarm returned nil despite the config having no assets for the local architecture") + } + // With the registry gone, only a cache hit can satisfy a digest ref. srv.Close() digestRef := u.Host + "/pause@" + pauseDigest.String() - for name, store := range map[string]*imagecache.Store{"ok": okStore, "asset-failure": failStore} { + for name, store := range map[string]*imagecache.Store{"ok": okStore, "asset-failure": failStore, "no-local-arch": archStore} { if _, err := store.EnsureImage(ctx, digestRef); err != nil { t.Errorf("pause image not prewarmed into the %s store: %v", name, err) } } } +// hangingObjectStorage blocks GetObject until the caller's context ends, +// simulating a bucket read that stalls without failing. +type hangingObjectStorage struct{} + +func (hangingObjectStorage) GetObject(ctx context.Context, _, _ string) (io.ReadCloser, error) { + <-ctx.Done() + return nil, ctx.Err() +} + +func (hangingObjectStorage) PutObject(_ context.Context, _, _ string, _ io.Reader) error { return nil } + +// TestPrewarmTimeout verifies a single prewarm attempt is bounded by +// prewarmTimeout: the queue has one worker, so an attempt that never returned +// would block every other config's prewarm. +func TestPrewarmTimeout(t *testing.T) { + origDir, origTimeout := ateompath.StaticFilesDir, prewarmTimeout + ateompath.StaticFilesDir = t.TempDir() + prewarmTimeout = 50 * time.Millisecond + t.Cleanup(func() { ateompath.StaticFilesDir, prewarmTimeout = origDir, origTimeout }) + + cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("hung runsc")))) + cfg.Spec.PauseImage = "" + p := &sandboxPrewarmer{herder: &AteomHerder{anonGCSClient: hangingObjectStorage{}}} + + done := make(chan error, 1) + go func() { done <- p.prewarm(context.Background(), cfg) }() + select { + case err := <-done: + if err == nil { + t.Error("prewarm returned nil despite the download hanging") + } + case <-time.After(10 * time.Second): + t.Fatal("prewarm never returned; a hung download would block the worker forever") + } +} + // TestSandboxAssetPrewarmDownloads runs the whole path: a SandboxConfig in a // fake clientset flows through the informer into the prewarm worker, which // lands the asset in the static-files cache without any Run/Restore request. From 3d7654c9d8f525179df4c6134cda3b23b1466df3 Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 09:21:02 -0700 Subject: [PATCH 05/10] atelet: keep the SandboxConfig informer off the startup-critical sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prewarm informer was requested on the shared factory before its blocking WaitForCacheSync, so a failing list/watch — most plausibly Forbidden while the ClusterRole rollout lags the binary — would hang atelet startup forever, turning a best-effort optimization into a node outage. Request the informer only after WaitForCacheSync returns and start it with a second factory Start call, which launches only informers added since the first. A failing list/watch now degrades prewarm instead of blocking the node: the reflector retries in the background and prewarm recovers with it. A watch error handler names the degradation in the log instead of leaving only generic reflector noise. The prewarm handler is now registered before its informer starts; the initial List still replays every existing SandboxConfig into it as an Add, so a freshly booted node prewarms current configs as before. --- cmd/atelet/main.go | 15 ++++++++++++--- cmd/atelet/sandbox_prewarm.go | 17 ++++++++++++++--- cmd/atelet/sandbox_prewarm_test.go | 12 ++++++------ 3 files changed, 32 insertions(+), 12 deletions(-) diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 4d8dcc2db2..209e2ecc56 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -279,9 +279,6 @@ func main() { // is unlikely to be used with frequency. ateFactory := externalversions.NewSharedInformerFactory(ateClient, 0) csiDriverConfigLister := ateFactory.Api().V1alpha1().CSIDriverConfigs().Lister() - // Requested before Start so the factory runs it; the prewarm handler is - // attached after the herder exists (see startSandboxAssetPrewarm below). - sandboxConfigInformer := ateFactory.Api().V1alpha1().SandboxConfigs().Informer() // Start an informer on the ClusterTrustBundle we care about (currently // only the egress trust bundle). The v1beta1 API is feature-gated: on a @@ -314,9 +311,21 @@ func main() { // Pre-download sandbox assets as SandboxConfigs appear/change so the first // Run/Restore on this node hits the cache. Best-effort: on failure the // on-demand fetch in ensureSandboxAssets still covers correctness. + // + // The informer is requested only now, after the factory's blocking + // WaitForCacheSync above, so it cannot hold up atelet startup when its + // list/watch fails (e.g. Forbidden while the ClusterRole rollout lags the + // binary): the reflector retries in the background and prewarm stays cold + // until it recovers. + sandboxConfigInformer := ateFactory.Api().V1alpha1().SandboxConfigs().Informer() if err := startSandboxAssetPrewarm(ctx, sandboxConfigInformer, wmService, microvmNodeCapable(hostDevRoot)); err != nil { slog.ErrorContext(ctx, "Sandbox asset prewarm disabled", slog.Any("err", err)) } + // The factory only runs informers that exist when Start is called: the + // Start above predates the SandboxConfigs informer, so without this call + // it would never list or watch. Start is idempotent per informer — this + // launches the new one and leaves the already-running ones untouched. + ateFactory.Start(stopCh) dialOpts, err := ateapiauth.DialOptions(ateapiauth.ClientConfig{ K8sClient: k8sClient, CAFile: *ateapiCAFile, diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index a51e27a7b6..67df7d8a4c 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -103,9 +103,20 @@ func newSandboxPrewarmer(herder *AteomHerder, lister listersv1alpha1.SandboxConf // assets referenced by no current SandboxConfig and no on-node actor record. func startSandboxAssetPrewarm(ctx context.Context, informer cache.SharedIndexInformer, herder *AteomHerder, microvmCapable bool) error { p := newSandboxPrewarmer(herder, listersv1alpha1.NewSandboxConfigLister(informer.GetIndexer()), microvmCapable) - // The handler is registered after the informer cache has synced, so it - // replays every existing SandboxConfig as a synthetic Add: a freshly booted - // node prewarms the current configs, not only future changes. + // Atelet startup never waits for this informer to sync: prewarm is + // best-effort, so a failing list/watch (e.g. Forbidden while an RBAC + // rollout lags the binary) must degrade prewarm, not hang the node. The + // handler makes that degradation visible in the log; the reflector keeps + // retrying and prewarm recovers with it. Setting it fails only on an + // already-started informer, where the default reflector logging applies. + if err := informer.SetWatchErrorHandler(func(_ *cache.Reflector, err error) { + slog.WarnContext(ctx, "SandboxConfig list/watch failed; sandbox asset prewarm degraded until it recovers", slog.Any("err", err)) + }); err != nil { + slog.InfoContext(ctx, "Could not set sandbox config watch error handler", slog.Any("err", err)) + } + // The initial List replays every existing SandboxConfig into the handler + // as an Add: a freshly booted node prewarms the current configs, not only + // future changes. if _, err := informer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj any) { p.enqueue(ctx, obj) }, UpdateFunc: func(_, obj any) { p.enqueue(ctx, obj) }, diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index f7d83a57cf..aeb2c3c589 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -337,21 +337,21 @@ func TestSandboxAssetPrewarmDownloads(t *testing.T) { client := fake.NewSimpleClientset(cfg) factory := externalversions.NewSharedInformerFactory(client, 0) informer := factory.Api().V1alpha1().SandboxConfigs().Informer() - stopCh := make(chan struct{}) - defer close(stopCh) - factory.Start(stopCh) - if !cache.WaitForCacheSync(stopCh, informer.HasSynced) { - t.Fatal("informer cache never synced") - } store, err := imagecache.New(t.TempDir()) if err != nil { t.Fatalf("imagecache.New: %v", err) } herder := &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}, imageCache: store} + // Handler first, informer start second, mirroring main: atelet startup + // must never wait on this informer's sync, and the initial List replays + // the pre-existing config into the handler as an Add. if err := startSandboxAssetPrewarm(ctx, informer, herder, false); err != nil { t.Fatalf("startSandboxAssetPrewarm: %v", err) } + stopCh := make(chan struct{}) + defer close(stopCh) + factory.Start(stopCh) wantPath := ateompath.RunSCBinaryPath(sha) deadline := time.Now().Add(10 * time.Second) From f7e3392b3ef416bfe009fa5e61cd5388e30b050c Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 09:25:29 -0700 Subject: [PATCH 06/10] atelet: narrow the prewarmer's herder dependency to what it uses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prewarmer took the whole AteomHerder while calling exactly two things on it: the asset fetch and the pause image pull. It now takes a one-method sandboxAssetFetcher interface — still implemented by AteomHerder, so prewarm and the on-demand Run/Restore fetch cannot diverge on cache layout — and the image cache store directly, which main already owns. The prewarmer's real dependencies are visible at the call site instead of implied by a service-wide handle. --- cmd/atelet/main.go | 2 +- cmd/atelet/sandbox_prewarm.go | 24 +++++++++++++++++------- cmd/atelet/sandbox_prewarm_test.go | 30 +++++++++++++++--------------- 3 files changed, 33 insertions(+), 23 deletions(-) diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 209e2ecc56..911d659feb 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -318,7 +318,7 @@ func main() { // binary): the reflector retries in the background and prewarm stays cold // until it recovers. sandboxConfigInformer := ateFactory.Api().V1alpha1().SandboxConfigs().Informer() - if err := startSandboxAssetPrewarm(ctx, sandboxConfigInformer, wmService, microvmNodeCapable(hostDevRoot)); err != nil { + if err := startSandboxAssetPrewarm(ctx, sandboxConfigInformer, wmService, imageCache, microvmNodeCapable(hostDevRoot)); err != nil { slog.ErrorContext(ctx, "Sandbox asset prewarm disabled", slog.Any("err", err)) } // The factory only runs informers that exist when Start is called: the diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index 67df7d8a4c..854ae8cf1f 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -28,6 +28,7 @@ import ( "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" + "github.com/agent-substrate/substrate/internal/imagecache" "github.com/agent-substrate/substrate/pkg/api/v1alpha1" listersv1alpha1 "github.com/agent-substrate/substrate/pkg/client/listers/api/v1alpha1" ) @@ -56,12 +57,20 @@ var prewarmTimeout = 5 * time.Minute // permanently broken config forever. const prewarmMaxRetries = 8 +// sandboxAssetFetcher is the one slice of AteomHerder the prewarmer needs. +// Prewarming through the same method as the Run/Restore path keeps the two +// fetches from ever diverging on cache layout or validation. +type sandboxAssetFetcher interface { + ensureSandboxAssets(ctx context.Context, rec *sandboxAssetsRecord) (map[string]string, error) +} + // sandboxPrewarmer downloads SandboxConfig assets into the node's // content-addressed static-files cache before any actor asks for them, so the // fetch inside the first Run/Restore on the node is a cache hit instead of a // download+extract on the critical path. type sandboxPrewarmer struct { - herder *AteomHerder + assets sandboxAssetFetcher + images *imagecache.Store // lister resolves a queued config name to its latest revision at // processing time, so coalesced events never prewarm a stale spec. lister listersv1alpha1.SandboxConfigLister @@ -78,9 +87,10 @@ type sandboxPrewarmer struct { microvmCapable bool } -func newSandboxPrewarmer(herder *AteomHerder, lister listersv1alpha1.SandboxConfigLister, microvmCapable bool) *sandboxPrewarmer { +func newSandboxPrewarmer(assets sandboxAssetFetcher, images *imagecache.Store, lister listersv1alpha1.SandboxConfigLister, microvmCapable bool) *sandboxPrewarmer { return &sandboxPrewarmer{ - herder: herder, + assets: assets, + images: images, lister: lister, // Downloads fail on the scale of network timeouts, not API conflicts, // so back off in seconds and cap in minutes rather than the @@ -101,8 +111,8 @@ func newSandboxPrewarmer(herder *AteomHerder, lister listersv1alpha1.SandboxConf // TODO: the static-files cache is never pruned, and prewarming every config // revision makes stale releases accumulate faster. Add a GC that removes // assets referenced by no current SandboxConfig and no on-node actor record. -func startSandboxAssetPrewarm(ctx context.Context, informer cache.SharedIndexInformer, herder *AteomHerder, microvmCapable bool) error { - p := newSandboxPrewarmer(herder, listersv1alpha1.NewSandboxConfigLister(informer.GetIndexer()), microvmCapable) +func startSandboxAssetPrewarm(ctx context.Context, informer cache.SharedIndexInformer, assets sandboxAssetFetcher, images *imagecache.Store, microvmCapable bool) error { + p := newSandboxPrewarmer(assets, images, listersv1alpha1.NewSandboxConfigLister(informer.GetIndexer()), microvmCapable) // Atelet startup never waits for this informer to sync: prewarm is // best-effort, so a failing list/watch (e.g. Forbidden while an RBAC // rollout lags the binary) must degrade prewarm, not hang the node. The @@ -221,7 +231,7 @@ func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxCon // schedule prewarm pause image if provided if cfg.Spec.PauseImage != "" { wg.Go(func() { - if _, err := p.herder.imageCache.EnsureImage(ctx, cfg.Spec.PauseImage); err != nil { + if _, err := p.images.EnsureImage(ctx, cfg.Spec.PauseImage); err != nil { imageErr = fmt.Errorf("while prewarming pause image %q: %w", cfg.Spec.PauseImage, err) } }) @@ -230,7 +240,7 @@ func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxCon rec, assetErr := recordFromSandboxConfig(cfg) if assetErr == nil { wg.Go(func() { - _, assetErr = p.herder.ensureSandboxAssets(ctx, rec) + _, assetErr = p.assets.ensureSandboxAssets(ctx, rec) }) } wg.Wait() diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index aeb2c3c589..34442d8fa1 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -123,7 +123,7 @@ func TestPrewarmEnqueueFilters(t *testing.T) { gvisor := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("runsc")))) t.Run("node without KVM", func(t *testing.T) { - p := newSandboxPrewarmer(nil, nil, false) + p := newSandboxPrewarmer(nil, nil, nil, false) p.enqueue(ctx, "not a sandbox config") p.enqueue(ctx, microvm) @@ -147,7 +147,7 @@ func TestPrewarmEnqueueFilters(t *testing.T) { }) t.Run("node with KVM", func(t *testing.T) { - p := newSandboxPrewarmer(nil, nil, true) + p := newSandboxPrewarmer(nil, nil, nil, true) p.enqueue(ctx, microvm) p.enqueue(ctx, gvisor) if p.queue.Len() != 2 { @@ -174,7 +174,7 @@ func TestPrewarmProcessRetries(t *testing.T) { t.Fatalf("indexer.Add: %v", err) } herder := &AteomHerder{anonGCSClient: fakeObjectStorage{err: errors.New("bucket unavailable")}} - p := newSandboxPrewarmer(herder, listersv1alpha1.NewSandboxConfigLister(indexer), false) + p := newSandboxPrewarmer(herder, nil, listersv1alpha1.NewSandboxConfigLister(indexer), false) p.queue.Add(cfg.Name) name, _ := p.queue.Get() @@ -236,10 +236,10 @@ func TestPrewarmPauseImage(t *testing.T) { cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256(content))) cfg.Spec.PauseImage = pauseRef - p := &sandboxPrewarmer{herder: &AteomHerder{ - anonGCSClient: fakeObjectStorage{data: content}, - imageCache: okStore, - }} + p := &sandboxPrewarmer{ + assets: &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}}, + images: okStore, + } if err := p.prewarm(ctx, cfg); err != nil { t.Fatalf("prewarm: %v", err) } @@ -249,10 +249,10 @@ func TestPrewarmPauseImage(t *testing.T) { // pause image from being pulled. failCfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("other runsc")))) failCfg.Spec.PauseImage = pauseRef - p = &sandboxPrewarmer{herder: &AteomHerder{ - anonGCSClient: fakeObjectStorage{err: errors.New("bucket unavailable")}, - imageCache: failStore, - }} + p = &sandboxPrewarmer{ + assets: &AteomHerder{anonGCSClient: fakeObjectStorage{err: errors.New("bucket unavailable")}}, + images: failStore, + } if err := p.prewarm(ctx, failCfg); err == nil { t.Error("prewarm returned nil despite the asset fetch failing") } @@ -264,7 +264,7 @@ func TestPrewarmPauseImage(t *testing.T) { "other-arch": {runscAssetName: archCfg.Spec.Assets[runtime.GOARCH][runscAssetName]}, } archCfg.Spec.PauseImage = pauseRef - p = &sandboxPrewarmer{herder: &AteomHerder{imageCache: archStore}} + p = &sandboxPrewarmer{images: archStore} if err := p.prewarm(ctx, archCfg); err == nil { t.Error("prewarm returned nil despite the config having no assets for the local architecture") } @@ -301,7 +301,7 @@ func TestPrewarmTimeout(t *testing.T) { cfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("hung runsc")))) cfg.Spec.PauseImage = "" - p := &sandboxPrewarmer{herder: &AteomHerder{anonGCSClient: hangingObjectStorage{}}} + p := &sandboxPrewarmer{assets: &AteomHerder{anonGCSClient: hangingObjectStorage{}}} done := make(chan error, 1) go func() { done <- p.prewarm(context.Background(), cfg) }() @@ -342,11 +342,11 @@ func TestSandboxAssetPrewarmDownloads(t *testing.T) { if err != nil { t.Fatalf("imagecache.New: %v", err) } - herder := &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}, imageCache: store} + herder := &AteomHerder{anonGCSClient: fakeObjectStorage{data: content}} // Handler first, informer start second, mirroring main: atelet startup // must never wait on this informer's sync, and the initial List replays // the pre-existing config into the handler as an Add. - if err := startSandboxAssetPrewarm(ctx, informer, herder, false); err != nil { + if err := startSandboxAssetPrewarm(ctx, informer, herder, store, false); err != nil { t.Fatalf("startSandboxAssetPrewarm: %v", err) } stopCh := make(chan struct{}) From b4d312bb3cd59def5de9af1f218cc8467422e3a7 Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 09:25:30 -0700 Subject: [PATCH 07/10] atelet: note that micro-VM capability will outgrow the KVM-only check microvmNodeCapable equates micro-VM capability with /dev/kvm presence. With /dev/mshv support proposed for AKS nodes, an mshv-only node would be wrongly reported incapable and skip prewarming micro-VM assets. Leave a TODO to treat presence of any micro-VM hypervisor device in SandboxDevices as capability once a second device exists. --- cmd/atelet/deviceplugin.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/cmd/atelet/deviceplugin.go b/cmd/atelet/deviceplugin.go index e65b9b2016..d4b3cce8b0 100644 --- a/cmd/atelet/deviceplugin.go +++ b/cmd/atelet/deviceplugin.go @@ -36,6 +36,12 @@ const hostDevRoot = "/host/dev" // nodes where the device exists. Device presence is therefore the earliest // reliable eligibility signal — known at atelet startup, before any WorkerPool // schedules here. +// +// TODO(https://github.com/agent-substrate/substrate/pull/1207): /dev/kvm is +// not the only micro-VM hypervisor device; once /dev/mshv support lands, an +// mshv-only node would be wrongly reported incapable here. Treat presence of +// any micro-VM hypervisor device in SandboxDevices as capability, not KVM +// alone. func microvmNodeCapable(devRoot string) bool { for _, d := range deviceplugin.SandboxDevices { if d.ResourceName == deviceplugin.ResourceKVM { From 288edfb362ac7ce011af7e8b6328e801aadb0efd Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 18:17:30 -0700 Subject: [PATCH 08/10] imagecache: run image pulls on a context detached from the initiating caller EnsureImage collapses concurrent pulls of one digest, and the winning call's context used to govern the pull for every waiter. Callers with very different lifetimes share these flights -- a serving Restore, a best-effort prewarm bounded by its own deadline, a caller whose daemon is draining after SIGTERM -- so a prewarm that won the slot could fail a Restore with its cancellation or timeout. The singleflight body now runs on a context that keeps the initiating caller's values but none of its cancellation, bounded by a store-level pull timeout (default 10m, WithPullTimeout to override). A waiter only ever sees a real pull failure; each caller stops waiting when its own context ends, while the pull runs on to warm the cache for the next attempt. --- internal/imagecache/imagecache.go | 44 ++++++++++++++++++++------ internal/imagecache/pull_gated_test.go | 36 +++++++++++++++++---- 2 files changed, 64 insertions(+), 16 deletions(-) diff --git a/internal/imagecache/imagecache.go b/internal/imagecache/imagecache.go index 5ac985fd81..7ba99e9d10 100644 --- a/internal/imagecache/imagecache.go +++ b/internal/imagecache/imagecache.go @@ -94,6 +94,10 @@ const ( // image pull. Memory use is O(stream buffers) per slot, independent of // layer size. layerPullConcurrency = 4 + + // defaultPullTimeout is the default per-pull bound (see WithPullTimeout). + // Generous enough for multi-GiB images on a busy node. + defaultPullTimeout = 10 * time.Minute ) // Store is atelet's handle to the on-disk layer pool. It is safe for @@ -121,6 +125,11 @@ type Store struct { // spec write / ateom mount that roots it. minAge time.Duration + // pullTimeout bounds each pull. Pulls run detached from the contexts of + // the callers waiting on them (see EnsureImage), so this is the only + // bound on how long one can run. + pullTimeout time.Duration + // meter, when set, is the meter the store reports on. See WithMeter. meter metric.Meter @@ -177,6 +186,11 @@ func WithMinAge(d time.Duration) Option { return func(s *Store) { s.minAge = d } } +// WithPullTimeout overrides the per-pull timeout (default 10m). +func WithPullTimeout(d time.Duration) Option { + return func(s *Store) { s.pullTimeout = d } +} + // WithMeter attaches the meter the store reports ate.imagecache.requests on. // Without it the store records nothing, so a caller with no metrics pipeline // needs no meter provider. @@ -209,7 +223,7 @@ type imageRecord struct { // startup recovery: verifying the layout version and sweeping temp dirs left // by unpacks that were in flight when a previous atelet died. func New(root string, opts ...Option) (*Store, error) { - s := &Store{root: root, minAge: defaultMinAge} + s := &Store{root: root, minAge: defaultMinAge, pullTimeout: defaultPullTimeout} for _, o := range opts { o(s) } @@ -366,16 +380,28 @@ func (s *Store) EnsureImage(ctx context.Context, ref string) (_ *Image, err erro slog.InfoContext(ctx, "Image cache miss", slog.String("ref", ref), slog.String("digest", digest.String())) // Collapse concurrent pulls of the same digest (e.g. several containers of - // one actor, or several actors landing at once). The winning call's ctx - // governs the pull; if it is cancelled the waiters fail too and retry at - // the RPC level. - v, err, _ := s.imageSF.Do(digest.String(), func() (any, error) { - return s.pull(ctx, parsedRef, digest) + // one actor, or several actors landing at once). Callers with very + // different lifetimes share these flights — a serving Restore, a + // best-effort prewarm bounded by its own deadline, a caller whose daemon + // is draining — so the pull runs on a context detached from whichever + // caller happened to start the flight, bounded only by the store's pull + // timeout: a waiter must only ever see a real pull failure, never another + // caller's cancellation. Each caller stops waiting when its own ctx ends, + // while the pull runs on to warm the cache for the next attempt. + ch := s.imageSF.DoChan(digest.String(), func() (any, error) { + pullCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), s.pullTimeout) + defer cancel() + return s.pull(pullCtx, parsedRef, digest) }) - if err != nil { - return nil, err + select { + case res := <-ch: + if res.Err != nil { + return nil, res.Err + } + return res.Val.(*Image), nil + case <-ctx.Done(): + return nil, fmt.Errorf("while waiting for pull of %s: %w", digest, context.Cause(ctx)) } - return v.(*Image), nil } // cachedImageHit is the hit side of the hitMu contract: it verifies the diff --git a/internal/imagecache/pull_gated_test.go b/internal/imagecache/pull_gated_test.go index e605cd8736..639b4fe03e 100644 --- a/internal/imagecache/pull_gated_test.go +++ b/internal/imagecache/pull_gated_test.go @@ -22,6 +22,7 @@ import ( "archive/tar" "context" "encoding/json" + "errors" "io" "log" "net/http" @@ -303,8 +304,9 @@ func TestPullReverifyFailsCleanlyOnYankedLayer(t *testing.T) { } } -// The one behavior change visible without GC: an interrupted pull leaves a -// valid partial record — resumable progress — rather than nothing. +// A mid-flight pull holds a valid partial record — resumable progress — +// rather than nothing, and a caller that stops waiting (cancellation) does +// not interrupt the pull itself, which runs on a detached context. func TestInterruptedPullLeavesResumableRecord(t *testing.T) { reg := newGatedRegistry(t) free, gatedLayer, _ := gatedTestLayers(t) @@ -328,12 +330,12 @@ func TestInterruptedPullLeavesResumableRecord(t *testing.T) { }) cancel() - if err := <-done; err == nil { - t.Fatal("EnsureImage succeeded despite cancellation") + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("EnsureImage with a cancelled caller = %v, want context.Canceled", err) } - release() - // The pre-written record survives the failure, referencing both layers. + // The caller is gone but the pull is still in flight, blocked on the + // gate. The pre-written record already references both layers. var recs []string entries, err := os.ReadDir(store.manifestsDir()) if err != nil { @@ -367,7 +369,10 @@ func TestInterruptedPullLeavesResumableRecord(t *testing.T) { t.Errorf("gated layer present after interrupted pull: %v", err) } - // A retry completes the image and yields a full cache hit. + // A retry joins the still-running flight and completes the image once + // the gate opens (which also drains the detached pull before the test's + // temp dirs are removed). + release() img, err := store.EnsureImage(context.Background(), ref) if err != nil { t.Fatalf("EnsureImage retry: %v", err) @@ -377,3 +382,20 @@ func TestInterruptedPullLeavesResumableRecord(t *testing.T) { t.Fatalf("no complete cachedImage after retry: %v, %v", cached, err) } } + +// A detached pull is not unbounded: pulls no longer end with the caller +// that started them, so the store's own pull timeout is what reclaims a +// wedged pull's flight slot. +func TestDetachedPullBoundedByPullTimeout(t *testing.T) { + reg := newGatedRegistry(t) + free, gatedLayer, _ := gatedTestLayers(t) + ref := reg.host + "/test/wedged:latest" + pushImage(t, ref, v1.Config{}, free, gatedLayer) + + store := newTestStore(t, WithPullTimeout(200*time.Millisecond)) + reg.gate(t, gatedLayer) // released only at test cleanup: the pull is wedged + + if _, err := store.EnsureImage(context.Background(), ref); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("EnsureImage on a wedged pull = %v, want context.DeadlineExceeded", err) + } +} From d785c6b26b953c83ffd42d712679c858917c5d52 Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 18:17:39 -0700 Subject: [PATCH 09/10] atelet: re-apply the sandbox class gate when prewarm processes a config enqueue filters on SandboxClass but queues only the config name, and process re-resolves the current revision at processing time. sandboxClass is mutable, so a gvisor-to-microvm edit inside the jitter or backoff window made a node without /dev/kvm download the micro-VM guest assets the gate exists to avoid. The gate now lives in skipConfig, applied both at enqueue and after the lister Get in process. --- cmd/atelet/sandbox_prewarm.go | 31 ++++++++++++++++++++++++------ cmd/atelet/sandbox_prewarm_test.go | 31 ++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 6 deletions(-) diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index 854ae8cf1f..80d1cba59b 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -138,26 +138,39 @@ func startSandboxAssetPrewarm(ctx context.Context, informer cache.SharedIndexInf return nil } -func (p *sandboxPrewarmer) enqueue(ctx context.Context, obj any) { - cfg, ok := obj.(*v1alpha1.SandboxConfig) - if !ok { - return - } +// skipConfig reports whether this node has nothing to prewarm for cfg, +// logging why. Both enqueue and process apply it: sandboxClass is mutable, +// so the class observed at enqueue time can be stale by the time the worker +// resolves the name after jitter or backoff, and the gate must hold for the +// revision actually prewarmed. +func (p *sandboxPrewarmer) skipConfig(ctx context.Context, cfg *v1alpha1.SandboxConfig) bool { switch cfg.Spec.SandboxClass { case v1alpha1.SandboxClassGvisor: // Every node runs gVisor workers; always prewarm. + return false case v1alpha1.SandboxClassMicroVM: if !p.microvmCapable { slog.DebugContext(ctx, "Skipping sandbox asset prewarm: node has no /dev/kvm, cannot run micro-VM workers", slog.String("config", cfg.Name)) - return + return true } + return false default: // An unknown class has no backend in this atelet (likely version skew // with a newer control plane); nothing to prewarm. slog.InfoContext(ctx, "Skipping sandbox asset prewarm: unknown sandbox class", slog.String("config", cfg.Name), slog.String("sandboxClass", string(cfg.Spec.SandboxClass))) + return true + } +} + +func (p *sandboxPrewarmer) enqueue(ctx context.Context, obj any) { + cfg, ok := obj.(*v1alpha1.SandboxConfig) + if !ok { + return + } + if p.skipConfig(ctx, cfg) { return } if prewarmMaxJitter > 0 { @@ -193,6 +206,12 @@ func (p *sandboxPrewarmer) process(ctx context.Context, name string) { p.queue.Forget(name) return } + if err == nil && p.skipConfig(ctx, cfg) { + // The class gate re-applies to the revision resolved now, which may + // differ from the one that passed enqueue's filter. + p.queue.Forget(name) + return + } if err == nil { err = p.prewarm(ctx, cfg) } diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index 34442d8fa1..786aee5581 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -189,6 +189,37 @@ func TestPrewarmProcessRetries(t *testing.T) { } } +// TestPrewarmProcessReappliesClassGate covers the gap between enqueue and +// processing: sandboxClass is mutable, so a config that passed the enqueue +// filter as gVisor may be micro-VM by the time the worker resolves it, and a +// node without KVM must skip it rather than download guest assets. +func TestPrewarmProcessReappliesClassGate(t *testing.T) { + origDir := ateompath.StaticFilesDir + ateompath.StaticFilesDir = t.TempDir() + t.Cleanup(func() { ateompath.StaticFilesDir = origDir }) + + ctx := context.Background() + // Enqueued while gVisor, edited to micro-VM before the worker ran. + cfg := gvisorConfig("mutating-config", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256([]byte("runsc")))) + cfg.Spec.SandboxClass = v1alpha1.SandboxClassMicroVM + + indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{}) + if err := indexer.Add(cfg); err != nil { + t.Fatalf("indexer.Add: %v", err) + } + // Any fetch would fail and requeue, so NumRequeues distinguishes + // "skipped" from "attempted and failed". + herder := &AteomHerder{anonGCSClient: fakeObjectStorage{err: errors.New("bucket unavailable")}} + p := newSandboxPrewarmer(herder, nil, listersv1alpha1.NewSandboxConfigLister(indexer), false) + + p.queue.Add(cfg.Name) + name, _ := p.queue.Get() + p.process(ctx, name) + if got := p.queue.NumRequeues(cfg.Name); got != 0 { + t.Errorf("NumRequeues after a class change to micro-VM on a non-KVM node = %d, want 0 (skipped, not retried)", got) + } +} + // TestMicrovmNodeCapable covers the detectable negative cases; the positive // case needs a /dev/kvm character device, which a test cannot mknod. func TestMicrovmNodeCapable(t *testing.T) { From 02616d527013878fc5044b78adfaab099c1049fe Mon Sep 17 00:00:00 2001 From: dberkov Date: Wed, 2 Sep 2026 18:17:49 -0700 Subject: [PATCH 10/10] atelet: stop retrying prewarm when a config has no assets for the node arch A config with no assets for this node's architecture will not grow them without a config update, and an update re-enqueues anyway. process could not tell this apart from a failed download, so every node of a non-listed architecture burned all eight rate-limited attempts with a warning each -- and every attempt redid the pause-image pull that had already succeeded, because prewarm joins both errors into one retry decision. recordFromSandboxConfig now wraps a sentinel error, and prewarm treats it as nothing-to-do: logged at debug, pause image still pulled, and the worker Forgets the config immediately. The sentinel is handled before the error join rather than in process, because errors.Is on the joined error would also match when only the pause pull failed transiently, silently dropping its retry. --- cmd/atelet/sandbox_prewarm.go | 23 ++++++++++++++++++++--- cmd/atelet/sandbox_prewarm_test.go | 15 +++++++++------ 2 files changed, 29 insertions(+), 9 deletions(-) diff --git a/cmd/atelet/sandbox_prewarm.go b/cmd/atelet/sandbox_prewarm.go index 80d1cba59b..5b360f7b19 100644 --- a/cmd/atelet/sandbox_prewarm.go +++ b/cmd/atelet/sandbox_prewarm.go @@ -257,7 +257,15 @@ func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxCon } // schedule prewarm sandbox assets if provided rec, assetErr := recordFromSandboxConfig(cfg) - if assetErr == nil { + switch { + case errors.Is(assetErr, errNoAssetsForArch): + // Permanent until the config changes, and a change re-enqueues: + // retrying would burn every attempt on a non-failure and redo the + // pause-image pull with it. Nothing to do for this node. + slog.DebugContext(ctx, "No sandbox assets to prewarm for this node's architecture", + slog.String("config", cfg.Name), slog.String("arch", runtime.GOARCH)) + assetErr = nil + case assetErr == nil: wg.Go(func() { _, assetErr = p.assets.ensureSandboxAssets(ctx, rec) }) @@ -266,21 +274,30 @@ func (p *sandboxPrewarmer) prewarm(ctx context.Context, cfg *v1alpha1.SandboxCon if err := errors.Join(assetErr, imageErr); err != nil { return err } + assets := 0 + if rec != nil { + assets = len(rec.Assets) + } slog.InfoContext(ctx, "Sandbox assets prewarmed", slog.String("config", cfg.Name), - slog.Int("assets", len(rec.Assets)), + slog.Int("assets", assets), slog.String("pauseImage", cfg.Spec.PauseImage), slog.Duration("duration", time.Since(t))) return nil } +// errNoAssetsForArch reports that a SandboxConfig lists no assets for this +// node's architecture. Unlike a failed download it is permanent until the +// config changes, so prewarm treats it as nothing-to-do rather than retrying. +var errNoAssetsForArch = errors.New("no sandbox assets for this architecture") + // recordFromSandboxConfig projects a SandboxConfig's per-architecture assets // onto the local node's architecture, mirroring recordFromRequest. func recordFromSandboxConfig(cfg *v1alpha1.SandboxConfig) (*sandboxAssetsRecord, error) { arch := runtime.GOARCH files := cfg.Spec.Assets[arch] if len(files) == 0 { - return nil, fmt.Errorf("sandbox config %q has no assets for architecture %q", cfg.Name, arch) + return nil, fmt.Errorf("sandbox config %q, architecture %q: %w", cfg.Name, arch, errNoAssetsForArch) } rec := &sandboxAssetsRecord{ SandboxClass: string(cfg.Spec.SandboxClass), diff --git a/cmd/atelet/sandbox_prewarm_test.go b/cmd/atelet/sandbox_prewarm_test.go index 786aee5581..71030d34d3 100644 --- a/cmd/atelet/sandbox_prewarm_test.go +++ b/cmd/atelet/sandbox_prewarm_test.go @@ -101,12 +101,14 @@ func TestRecordFromSandboxConfig(t *testing.T) { t.Errorf("Assets[%q] = %+v, want %+v", runscAssetName, got, want) } - // A config with no assets for this node's architecture cannot be projected. + // A config with no assets for this node's architecture cannot be + // projected, and the error carries the sentinel that keeps prewarm from + // retrying a condition only a config change can clear. cfg.Spec.Assets = map[string]map[string]v1alpha1.AssetFile{ "other-arch": {runscAssetName: {URL: "gs://bucket/runsc", SHA256: sha}}, } - if _, err := recordFromSandboxConfig(cfg); err == nil { - t.Error("recordFromSandboxConfig accepted a config with no assets for the local architecture") + if _, err := recordFromSandboxConfig(cfg); !errors.Is(err, errNoAssetsForArch) { + t.Errorf("recordFromSandboxConfig with no assets for the local architecture = %v, want errNoAssetsForArch", err) } } @@ -289,15 +291,16 @@ func TestPrewarmPauseImage(t *testing.T) { } // A config with no assets for this node's architecture still gets its - // pause image pulled: the image needs no per-architecture projection. + // pause image pulled, and prewarm succeeds: the missing assets are + // permanent until the config changes, not a retryable failure. archCfg := gvisorConfig("gvisor-default", "gs://bucket/runsc", fmt.Sprintf("%x", sha256.Sum256(content))) archCfg.Spec.Assets = map[string]map[string]v1alpha1.AssetFile{ "other-arch": {runscAssetName: archCfg.Spec.Assets[runtime.GOARCH][runscAssetName]}, } archCfg.Spec.PauseImage = pauseRef p = &sandboxPrewarmer{images: archStore} - if err := p.prewarm(ctx, archCfg); err == nil { - t.Error("prewarm returned nil despite the config having no assets for the local architecture") + if err := p.prewarm(ctx, archCfg); err != nil { + t.Errorf("prewarm with no assets for the local architecture: %v", err) } // With the registry gone, only a cache hit can satisfy a digest ref.