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 ca523102f5..f59fa6574e 100644 --- a/internal/ateworker/register/register.go +++ b/internal/ateworker/register/register.go @@ -52,14 +52,15 @@ 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 } -// 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() } @@ -70,22 +71,24 @@ 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") + 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 @@ -244,10 +247,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 +293,21 @@ func run(ctx context.Context, cfg ReportConfig, capacity *ateletpb.WorkerResourc if err != nil { return err } - epoch.Store(registered) + // 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 + } + 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..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 { @@ -270,38 +271,55 @@ 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) + } + 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) + } + }) } } +// runtimesRecorder is a fakeSupport that hands each sandbox runtimes report to +// onReport and returns 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) {