From 71abc9e75140f9e6848ff5e9d8070b0bfa559cbf Mon Sep 17 00:00:00 2001 From: Shrutiya Date: Wed, 28 Oct 2026 22:31:13 +0000 Subject: [PATCH 1/2] ateworker: report ready only after sandbox runtimes are registered run stored the worker epoch as soon as RegisterWorker returned, before reconciler.sync read sandbox-runtimes.json and called SetWorkerSandboxRuntimes. When atelet had not written sandbox-runtimes.json yet or SetWorkerSandboxRuntimes failed, Epoch.Registered still returned nil, so /readyz served 200 OK and the Worker moved to WORKER_STATE_ACTIVE with no SandboxClasses recorded on the control plane. Store the epoch in the SetWorkerSandboxRuntimes callback after the first report succeeds. Epoch.Registered and Epoch.Require then hold the pod unready and refuse workloads with Unavailable until the control plane holds both capacity and sandbox runtimes. Signed-off-by: Shrutiya --- internal/ateworker/register/register.go | 35 +++++++---- internal/ateworker/register/register_test.go | 65 +++++++++++++------- 2 files changed, 68 insertions(+), 32 deletions(-) diff --git a/internal/ateworker/register/register.go b/internal/ateworker/register/register.go index ca523102f5..04372737fd 100644 --- a/internal/ateworker/register/register.go +++ b/internal/ateworker/register/register.go @@ -52,9 +52,10 @@ const ( resyncInterval = 5 * time.Minute ) -// Epoch is the epoch this worker process registered with. A worker process -// takes no workloads until it has registered, and reports this epoch with each -// one it starts, so the control plane can tell which process ran it. +// Epoch is the epoch this worker process registered with, set once the control +// plane holds its capacity and its sandbox runtimes. A worker process takes no +// workloads until then, and reports this epoch with each one it starts, so the +// control plane can tell which process ran it. type Epoch struct { v atomic.Int64 } @@ -70,8 +71,9 @@ func (e *Epoch) Store(epoch int64) { } // Registered is a readiness check that fails until this worker process has -// registered. The pod is then not Ready, so the Worker takes no placements -// while a restarted worker process has yet to register. +// registered and reported its sandbox runtimes, which is when the control +// plane can first place on it. The pod is not Ready before then, so the +// Deployment does not count a worker with no usable runtime as Ready. func (e *Epoch) Registered(context.Context) error { if e.Load() == 0 { return errors.New("worker has not registered yet") @@ -244,10 +246,10 @@ type supportClient interface { SetWorkerSandboxRuntimes(ctx context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error } -// Report registers this worker process with the node-local atelet, stores the -// epoch it registered with in epoch, then keeps its sandbox runtimes reported -// until ctx ends: it reads its sandbox runtimes file at startup and again -// whenever atelet rewrites it. +// Report registers this worker process with the node-local atelet, then keeps +// its sandbox runtimes reported until ctx ends: it reads its sandbox runtimes +// file at startup and again whenever atelet rewrites it. It stores the epoch it +// registered with in epoch once the first report is accepted. // // It returns early on a configuration error, which no retry would fix, and // when the control plane refuses this process's epoch, which means another @@ -290,12 +292,23 @@ func run(ctx context.Context, cfg ReportConfig, capacity *ateletpb.WorkerResourc if err != nil { return err } - epoch.Store(registered) + // The epoch is stored once the control plane holds this process' sandbox + // runtimes as well as its capacity: it places nothing here before then, + // and Registered keeps the pod not Ready until it could. A workload that + // arrives before the response does is refused with Unavailable and + // retried, the same as one before registration. + report := func(ctx context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error { + if err := client.SetWorkerSandboxRuntimes(ctx, req); err != nil { + return err + } + epoch.Store(registered) + return nil + } r := newReconciler( cfg, registered, func() (*sandboxruntimes.File, error) { return sandboxruntimes.Read(cfg.RuntimesFile) }, - client.SetWorkerSandboxRuntimes, + report, ) return r.run(ctx, changes) } diff --git a/internal/ateworker/register/register_test.go b/internal/ateworker/register/register_test.go index beefb454a6..a1a64846ff 100644 --- a/internal/ateworker/register/register_test.go +++ b/internal/ateworker/register/register_test.go @@ -270,38 +270,61 @@ func TestRegisterOnce(t *testing.T) { } } -func TestRunStoresEpochBeforeReportingRuntimes(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() +// run stores the epoch once the control plane has accepted the sandbox +// runtimes, so a worker whose report has not landed stays not Ready and +// refuses workloads. +func TestRunStoresEpochAfterFirstReport(t *testing.T) { + for _, tc := range []struct { + name string + reportErr error + wantEpoch int64 + }{ + {name: "report accepted", wantEpoch: 1}, + {name: "report failed", reportErr: status.Error(codes.Unavailable, "ateapi is down")}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() - path := filepath.Join(t.TempDir(), nodepath.SandboxRuntimesFileName) - if _, err := sandboxruntimes.Write(path, gvisorFile(sandboxruntimes.Attributes{hardware.AttrSandboxBundleHash: "sha256:v1"})); err != nil { - t.Fatal(err) - } - var epoch Epoch - client := &runtimesRecorder{fakeSupport: &fakeSupport{}, onReport: func(req *ateletpb.SetWorkerSandboxRuntimesRequest) { - if got := epoch.Load(); got != req.GetEpoch() { - t.Errorf("epoch stored = %d when runtimes were reported with %d", got, req.GetEpoch()) - } - cancel() - }} - err := run(ctx, ReportConfig{SandboxClass: "gvisor", RuntimesFile: path, PodIPs: []string{"10.0.0.1"}}, &ateletpb.WorkerResources{Actors: testActors}, client, &epoch, nil) - if !errors.Is(err, context.Canceled) { - t.Fatalf("run() = %v, want context.Canceled", err) - } - if epoch.Load() != 1 { - t.Errorf("epoch = %d, want 1", epoch.Load()) + path := filepath.Join(t.TempDir(), nodepath.SandboxRuntimesFileName) + if _, err := sandboxruntimes.Write(path, gvisorFile(sandboxruntimes.Attributes{hardware.AttrSandboxBundleHash: "sha256:v1"})); err != nil { + t.Fatal(err) + } + var epoch Epoch + client := &runtimesRecorder{fakeSupport: &fakeSupport{}, err: tc.reportErr, onReport: func(*ateletpb.SetWorkerSandboxRuntimesRequest) { + if got := epoch.Load(); got != 0 { + t.Errorf("epoch = %d while runtimes were being reported, want 0", got) + } + if err := epoch.Registered(ctx); err == nil { + t.Error("Registered() = nil while runtimes were being reported, want an error") + } + cancel() + }} + err := run(ctx, ReportConfig{SandboxClass: "gvisor", RuntimesFile: path, PodIPs: []string{"10.0.0.1"}}, &ateletpb.WorkerResources{Actors: testActors}, client, &epoch, nil) + if !errors.Is(err, context.Canceled) { + t.Fatalf("run() = %v, want context.Canceled", err) + } + if got := epoch.Load(); got != tc.wantEpoch { + t.Errorf("epoch = %d, want %d", got, tc.wantEpoch) + } + if err := epoch.Registered(context.Background()); (err == nil) != (tc.wantEpoch != 0) { + t.Errorf("Registered() = %v, want ready %v", err, tc.wantEpoch != 0) + } + }) } } +// runtimesRecorder is a fakeSupport that hands each sandbox runtimes report to +// onReport and fails it with err. type runtimesRecorder struct { *fakeSupport onReport func(*ateletpb.SetWorkerSandboxRuntimesRequest) + err error } func (r *runtimesRecorder) SetWorkerSandboxRuntimes(_ context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error { r.onReport(req) - return nil + return r.err } func TestReconcilerSync(t *testing.T) { From a73d028a7f017fbf46810181b2434483e6fb284e Mon Sep 17 00:00:00 2001 From: Shrutiya Date: Wed, 28 Oct 2026 22:56:38 +0000 Subject: [PATCH 2/2] ateworker: update Epoch readiness messages and comments Signed-off-by: Shrutiya --- cmd/ateworker-gvisor/main.go | 2 +- cmd/ateworker-microvm/main.go | 2 +- internal/ateworker/register/register.go | 17 ++++++++--------- internal/ateworker/register/register_test.go | 11 +++-------- 4 files changed, 13 insertions(+), 19 deletions(-) diff --git a/cmd/ateworker-gvisor/main.go b/cmd/ateworker-gvisor/main.go index 7f2ec9ff81..2d36c8023c 100644 --- a/cmd/ateworker-gvisor/main.go +++ b/cmd/ateworker-gvisor/main.go @@ -369,7 +369,7 @@ type AteWorkerService struct { shuttingDown atomic.Bool // epoch is the epoch this process registered with. It starts no workload - // before registering. + // before reporting its sandbox runtimes. epoch register.Epoch // cgroupRoot is where the sandbox's cgroup v2 leaves live: the worker pod's diff --git a/cmd/ateworker-microvm/main.go b/cmd/ateworker-microvm/main.go index b93b575031..f74dad77bc 100644 --- a/cmd/ateworker-microvm/main.go +++ b/cmd/ateworker-microvm/main.go @@ -366,7 +366,7 @@ type AteWorkerService struct { shuttingDown atomic.Bool // epoch is the epoch this process registered with. It starts no workload - // before registering. + // before reporting its sandbox runtimes. epoch register.Epoch podUID string diff --git a/internal/ateworker/register/register.go b/internal/ateworker/register/register.go index 04372737fd..f59fa6574e 100644 --- a/internal/ateworker/register/register.go +++ b/internal/ateworker/register/register.go @@ -60,7 +60,7 @@ type Epoch struct { v atomic.Int64 } -// Load returns the registered epoch, or 0 before registration. +// Load returns the registered epoch, or 0 until sandbox runtimes are reported. func (e *Epoch) Load() int64 { return e.v.Load() } @@ -76,18 +76,19 @@ func (e *Epoch) Store(epoch int64) { // Deployment does not count a worker with no usable runtime as Ready. func (e *Epoch) Registered(context.Context) error { if e.Load() == 0 { - return errors.New("worker has not registered yet") + return errors.New("worker has not reported its sandbox runtimes yet") } return nil } // Require returns the registered epoch for a workload about to start, or -// Unavailable before registration, so the caller retries once it has. +// Unavailable until sandbox runtimes are reported, so the caller retries once +// they are. func (e *Epoch) Require() (int64, error) { if epoch := e.Load(); epoch != 0 { return epoch, nil } - return 0, apierror.Unavailable("worker has not registered yet: not accepting workloads") + return 0, apierror.Unavailable("worker has not reported its sandbox runtimes yet: not accepting workloads") } // FromFiles combines the downward API compute limits with the worker's own @@ -292,11 +293,9 @@ func run(ctx context.Context, cfg ReportConfig, capacity *ateletpb.WorkerResourc if err != nil { return err } - // The epoch is stored once the control plane holds this process' sandbox - // runtimes as well as its capacity: it places nothing here before then, - // and Registered keeps the pod not Ready until it could. A workload that - // arrives before the response does is refused with Unavailable and - // retried, the same as one before registration. + // Stored once the control plane holds this process's sandbox runtimes as + // well as its capacity. A workload that arrives before the response does is + // refused with Unavailable and retried, the same as one before registration. report := func(ctx context.Context, req *ateletpb.SetWorkerSandboxRuntimesRequest) error { if err := client.SetWorkerSandboxRuntimes(ctx, req); err != nil { return err diff --git a/internal/ateworker/register/register_test.go b/internal/ateworker/register/register_test.go index a1a64846ff..5642420ec3 100644 --- a/internal/ateworker/register/register_test.go +++ b/internal/ateworker/register/register_test.go @@ -172,7 +172,8 @@ func TestEpochRequire(t *testing.T) { } // The readiness check holds the pod not Ready until the worker process has -// registered, so a restarted one takes no placements before it can serve them. +// reported its sandbox runtimes, so it takes no placements before it can serve +// them. func TestEpochRegistered(t *testing.T) { var e Epoch if err := e.Registered(context.Background()); err == nil { @@ -295,9 +296,6 @@ func TestRunStoresEpochAfterFirstReport(t *testing.T) { if got := epoch.Load(); got != 0 { t.Errorf("epoch = %d while runtimes were being reported, want 0", got) } - if err := epoch.Registered(ctx); err == nil { - t.Error("Registered() = nil while runtimes were being reported, want an error") - } cancel() }} err := run(ctx, ReportConfig{SandboxClass: "gvisor", RuntimesFile: path, PodIPs: []string{"10.0.0.1"}}, &ateletpb.WorkerResources{Actors: testActors}, client, &epoch, nil) @@ -307,15 +305,12 @@ func TestRunStoresEpochAfterFirstReport(t *testing.T) { if got := epoch.Load(); got != tc.wantEpoch { t.Errorf("epoch = %d, want %d", got, tc.wantEpoch) } - if err := epoch.Registered(context.Background()); (err == nil) != (tc.wantEpoch != 0) { - t.Errorf("Registered() = %v, want ready %v", err, tc.wantEpoch != 0) - } }) } } // runtimesRecorder is a fakeSupport that hands each sandbox runtimes report to -// onReport and fails it with err. +// onReport and returns err. type runtimesRecorder struct { *fakeSupport onReport func(*ateletpb.SetWorkerSandboxRuntimesRequest)