diff --git a/cmd/atelet/main.go b/cmd/atelet/main.go index 46de6e0d4e..7cb7310855 100644 --- a/cmd/atelet/main.go +++ b/cmd/atelet/main.go @@ -435,11 +435,6 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp * if err != nil { return nil, status.Error(codes.InvalidArgument, err.Error()) } - assetPaths, err := s.ensureSandboxAssets(ctx, sandboxRec) - if err != nil { - return nil, err - } - if err := resetActorDirs(actorUID); err != nil { return nil, fmt.Errorf("while resetting actor dirs: %w", err) } @@ -455,16 +450,43 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp * return nil, fmt.Errorf("while recording sandbox assets: %w", err) } - if err := s.prepareOCIBundles(ctx, actorUID, actorRef.Name, - req.GetSpec(), sandboxRec.PauseImage, req.GetTargetAteomUid(), - ); err != nil { - return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidContainerConfig) - } - client, err := s.dialAteom(ctx, req.GetTargetAteomUid()) if err != nil { return nil, err } + if err := prepareOCIPrerequisites(actorUID, actorRef.Name, req.GetSpec()); err != nil { + return nil, err + } + + var assetPaths map[string]string + g, gctx := errgroup.WithContext(ctx) + g.Go(func() error { + dependencies, dependenciesCtx := errgroup.WithContext(gctx) + dependencies.Go(func() (err error) { + assetPaths, err = s.ensureSandboxAssets(dependenciesCtx, sandboxRec) + return err + }) + dependencies.Go(func() error { + return s.preparePauseOCIBundle(dependenciesCtx, actorUID, req.GetSpec(), sandboxRec.PauseImage, req.GetTargetAteomUid()) + }) + if err := dependencies.Wait(); err != nil { + return err + } + return prepareSandbox(gctx, client, &ateompb.PrepareSandboxRequest{ + ActorUid: actorUID, + RunscPath: runscPathFor(assetPaths), + RedirectEgress: req.GetEgressGateway() != nil, + CpuMilli: req.GetCpuMilli(), + MemoryBytes: req.GetMemoryBytes(), + }) + }) + g.Go(func() error { + return s.prepareApplicationOCIBundles(gctx, actorUID, req.GetSpec(), req.GetTargetAteomUid()) + }) + if err := g.Wait(); err != nil { + discardPreparedSandbox(ctx, client, actorUID) + return nil, ateerrors.CrashIfReason(ctx, err, ateerrors.ReasonInvalidContainerConfig) + } // Tell ateom to start the workload. gVisor uses RunscPath; the micro-VM // runtime uses the full RuntimeAssetPaths set. @@ -481,6 +503,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp * CpuMilli: req.GetCpuMilli(), MemoryBytes: req.GetMemoryBytes(), }); err != nil { + discardPreparedSandbox(ctx, client, actorUID) return nil, fmt.Errorf("while calling ateom.RunWorkload: %w", err) } @@ -1389,6 +1412,20 @@ func (s *AteomHerder) prepareOCIBundles( pauseImage string, targetAteomUid string, ) error { + if err := prepareOCIPrerequisites(actorUID, actorName, spec); err != nil { + return err + } + g, gctx := errgroup.WithContext(ctx) + g.Go(func() error { + return s.preparePauseOCIBundle(gctx, actorUID, spec, pauseImage, targetAteomUid) + }) + g.Go(func() error { + return s.prepareApplicationOCIBundles(gctx, actorUID, spec, targetAteomUid) + }) + return g.Wait() +} + +func prepareOCIPrerequisites(actorUID, actorName string, spec *ateletpb.WorkloadSpec) error { // Populate the per-actor identity directory that gets bind-mounted into // the application containers. Regenerated on every resume, so it carries // the correct per-actor name even when restoring from the golden snapshot. @@ -1408,46 +1445,46 @@ func (s *AteomHerder) prepareOCIBundles( } } } + return nil +} - g, gCtx := errgroup.WithContext(ctx) - - // Pause container. - g.Go(func() error { - annotations := map[string]string{ - "io.kubernetes.cri.container-type": "sandbox", - "io.kubernetes.cri.container-name": "pause", - } - // Declare durable-dir volumes to gVisor. We use the volume name as the - // mount hint name to support multiple durable-dir volumes. - for _, vol := range spec.GetVolumes() { - if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR { - annotations[fmt.Sprintf("dev.gvisor.spec.mount.%s.type", vol.GetName())] = "bind" - annotations[fmt.Sprintf("dev.gvisor.spec.mount.%s.share", vol.GetName())] = "container" - annotations[fmt.Sprintf("dev.gvisor.spec.mount.%s.source", vol.GetName())] = ateompath.DurableDirVolumeMountPoint(actorUID, vol.GetName()) - } - } - - if err := prepareOCIDirectory( - gCtx, - s.imageCache, - actorUID, - "pause", - pauseImage, - []string{"/pause"}, - nil, - nil, - annotations, - ateompath.AteomNetNSPath(targetAteomUid), - "", // pause is sandbox infra; it gets no actor identity mount. - nil, - nil, - ); err != nil { - return wrapFileSystemErr("while creating pause OCI bundle", err) +func (s *AteomHerder) preparePauseOCIBundle(ctx context.Context, actorUID string, spec *ateletpb.WorkloadSpec, pauseImage, targetAteomUID string) error { + annotations := map[string]string{ + "io.kubernetes.cri.container-type": "sandbox", + "io.kubernetes.cri.container-name": "pause", + } + // Declare durable-dir volumes to gVisor. We use the volume name as the + // mount hint name to support multiple durable-dir volumes. + for _, vol := range spec.GetVolumes() { + if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR { + annotations[fmt.Sprintf("dev.gvisor.spec.mount.%s.type", vol.GetName())] = "bind" + annotations[fmt.Sprintf("dev.gvisor.spec.mount.%s.share", vol.GetName())] = "container" + annotations[fmt.Sprintf("dev.gvisor.spec.mount.%s.source", vol.GetName())] = ateompath.DurableDirVolumeMountPoint(actorUID, vol.GetName()) } - return nil - }) + } + if err := prepareOCIDirectory( + ctx, + s.imageCache, + actorUID, + "pause", + pauseImage, + []string{"/pause"}, + nil, + nil, + annotations, + ateompath.AteomNetNSPath(targetAteomUID), + "", // pause is sandbox infra; it gets no actor identity mount. + nil, + nil, + ); err != nil { + return wrapFileSystemErr("while creating pause OCI bundle", err) + } + return nil +} - // Application containers. +func (s *AteomHerder) prepareApplicationOCIBundles(ctx context.Context, actorUID string, spec *ateletpb.WorkloadSpec, targetAteomUID string) error { + g, gctx := errgroup.WithContext(ctx) + identityDir := ateompath.ActorIdentityDirPath(actorUID) for _, ctr := range spec.GetContainers() { ctr := ctr var envs []string @@ -1456,7 +1493,7 @@ func (s *AteomHerder) prepareOCIBundles( } g.Go(func() error { if err := prepareOCIDirectory( - gCtx, + gctx, s.imageCache, actorUID, ctr.GetName(), @@ -1469,7 +1506,7 @@ func (s *AteomHerder) prepareOCIBundles( "io.kubernetes.cri.sandbox-id": "pause", "io.kubernetes.cri.container-name": ctr.GetName(), }, - ateompath.AteomNetNSPath(targetAteomUid), + ateompath.AteomNetNSPath(targetAteomUID), identityDir, spec.GetVolumes(), ctr.GetVolumeMounts(), @@ -1479,10 +1516,28 @@ func (s *AteomHerder) prepareOCIBundles( return nil }) } - return g.Wait() } +// prepareSandbox is a no-op for runtimes and older ateom versions that do not +// implement the split startup RPC; their RunWorkload path remains unchanged. +func prepareSandbox(ctx context.Context, client ateompb.AteomClient, req *ateompb.PrepareSandboxRequest) error { + _, err := client.PrepareSandbox(ctx, req) + if status.Code(err) == codes.Unimplemented { + return nil + } + return err +} + +func discardPreparedSandbox(ctx context.Context, client ateompb.AteomClient, actorUID string) { + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cancel() + _, err := client.DiscardPreparedSandbox(cleanupCtx, &ateompb.DiscardPreparedSandboxRequest{ActorUid: actorUID}) + if err != nil && status.Code(err) != codes.Unimplemented { + slog.WarnContext(cleanupCtx, "Failed to discard prepared sandbox", slog.Any("err", err)) + } +} + // dialAteom opens (or reuses) the gRPC connection to the target ateom // pod and returns an ateom client. func (s *AteomHerder) dialAteom(ctx context.Context, targetAteomUid string) (ateompb.AteomClient, error) { diff --git a/cmd/atelet/main_test.go b/cmd/atelet/main_test.go index 1bb492db44..d9dd123785 100644 --- a/cmd/atelet/main_test.go +++ b/cmd/atelet/main_test.go @@ -54,6 +54,20 @@ import ( const testPauseImage = "registry.k8s.io/pause:3.10.2@sha256:f548e0e8e3dc1896ca956272154dde3314e8cc4fde0a57577ee9fa1c63f5baf4" +type unimplementedPrepareClient struct { + ateompb.AteomClient +} + +func (unimplementedPrepareClient) PrepareSandbox(context.Context, *ateompb.PrepareSandboxRequest, ...grpc.CallOption) (*ateompb.PrepareSandboxResponse, error) { + return nil, status.Error(codes.Unimplemented, "runtime uses RunWorkload") +} + +func TestPrepareSandboxFallsBackWhenUnimplemented(t *testing.T) { + if err := prepareSandbox(context.Background(), unimplementedPrepareClient{}, &ateompb.PrepareSandboxRequest{}); err != nil { + t.Fatalf("prepareSandbox returned an error for a runtime without the split RPC: %v", err) + } +} + // TestPortFlagDefault verifies the default value of the --port flag. func TestPortFlagDefault(t *testing.T) { f := pflag.Lookup("port") diff --git a/cmd/ateom-gvisor/main.go b/cmd/ateom-gvisor/main.go index dc59c6fd84..ea953f9e16 100644 --- a/cmd/ateom-gvisor/main.go +++ b/cmd/ateom-gvisor/main.go @@ -286,6 +286,7 @@ func runAtunnel(ctx context.Context, upstream *url.URL) (*atunnel.Server, *atunn } const ( + rpcPrepareSandbox = "PrepareSandbox" rpcRunWorkload = "RunWorkload" rpcRestoreWorkload = "RestoreWorkload" rpcCheckpointWorkload = "CheckpointWorkload" @@ -304,6 +305,23 @@ type workloadSession struct { containers []string } +// preparedSandbox is the root container started by PrepareSandbox and waiting +// for the matching RunWorkload call. It is guarded by AteomService.lock. +type preparedSandbox struct { + actorUID string + runscPath string + redirectEgress bool + cpuMilli int64 + memoryBytes int64 + // rootDeleted makes DiscardPreparedSandbox retryable when later cleanup fails. + rootDeleted bool +} + +func (p *preparedSandbox) matches(actorUID, runscPath string, redirectEgress bool, cpuMilli, memoryBytes int64) bool { + return p.actorUID == actorUID && p.runscPath == runscPath && p.redirectEgress == redirectEgress && + p.cpuMilli == cpuMilli && p.memoryBytes == memoryBytes +} + type cancelableMutex struct { ch chan struct{} } @@ -387,6 +405,10 @@ type AteomService struct { // RunWorkload/RestoreWorkload, cleared by CheckpointWorkload. Guarded by lock. activeSession *workloadSession + // prepared is non-nil between PrepareSandbox and the matching RunWorkload, + // or until DiscardPreparedSandbox tears the sandbox down. + prepared *preparedSandbox + activeRPCMu sync.Mutex activeRPC *activeRPCInfo @@ -441,10 +463,10 @@ func (s *AteomService) clearActiveRPC() { s.activeRPC = nil } -func (s *AteomService) cancelActiveRestoreOrRunRPC() { +func (s *AteomService) cancelActiveStartupRPC() { s.activeRPCMu.Lock() defer s.activeRPCMu.Unlock() - if s.activeRPC != nil && (s.activeRPC.name == rpcRestoreWorkload || s.activeRPC.name == rpcRunWorkload) { + if s.activeRPC != nil && (s.activeRPC.name == rpcPrepareSandbox || s.activeRPC.name == rpcRestoreWorkload || s.activeRPC.name == rpcRunWorkload) { slog.Info("Cancelling in-progress workload startup RPC due to graceful shutdown", slog.String("rpc", s.activeRPC.name)) s.activeRPC.cancel() } @@ -454,10 +476,10 @@ func (s *AteomService) cancelActiveRestoreOrRunRPC() { // containers to exit. func (s *AteomService) gracefulShutdown(ctx context.Context) { s.shuttingDown.Store(true) - // If there is an active run or restore RPC, try to cancel it. This is considered + // If there is an active startup RPC, try to cancel it. This is considered // less disruptive than waiting for it to complete and then immediately sending // a SIGTERM. - s.cancelActiveRestoreOrRunRPC() + s.cancelActiveStartupRPC() // Attempt to acquire the lock used to serialize ateom RPCs. This will wait for any // pending RPCs to finish (suspend, resume, etc...). After the RPCs finish there @@ -578,21 +600,152 @@ func containerNames(containers []*ateompb.Container) []string { return names } -func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkloadRequest) (resp *ateompb.RunWorkloadResponse, retErr error) { +// PrepareSandbox starts gVisor's root container while atelet continues pulling +// and unpacking the application images. +func (s *AteomService) PrepareSandbox(ctx context.Context, req *ateompb.PrepareSandboxRequest) (resp *ateompb.PrepareSandboxResponse, retErr error) { s.lock.Lock() defer s.lock.Unlock() if err := s.rejectIfDraining(); err != nil { return nil, err } + if req.GetActorUid() == "" || req.GetRunscPath() == "" { + return nil, status.Error(codes.InvalidArgument, "actor_uid and runsc_path are required") + } + if s.prepared != nil { + if s.prepared.rootDeleted { + return nil, status.Error(codes.FailedPrecondition, "prepared sandbox cleanup is incomplete") + } + if s.prepared.matches(req.GetActorUid(), req.GetRunscPath(), req.GetRedirectEgress(), req.GetCpuMilli(), req.GetMemoryBytes()) { + return &ateompb.PrepareSandboxResponse{}, nil + } + return nil, status.Error(codes.FailedPrecondition, "ateom already has a different prepared sandbox") + } + if s.activeSession != nil { + return nil, status.Error(codes.FailedPrecondition, "ateom already has a running workload") + } ctx, cancel := context.WithCancel(ctx) defer cancel() - s.setActiveRPC(rpcRunWorkload, cancel) + s.setActiveRPC(rpcPrepareSandbox, cancel) defer s.clearActiveRPC() if err := s.deactivateActorNetworking(ctx); err != nil { return nil, err } + rcmd := &runsc{ + path: req.GetRunscPath(), + actorUID: req.GetActorUid(), + size: sizing.FromLimits(req.GetCpuMilli(), req.GetMemoryBytes()), + } + rootCreated := false + networkConfigured := false + defer func() { + if retErr == nil { + return + } + cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cleanupCancel() + if rootCreated { + deleteContainers(cleanupCtx, rcmd, []string{"pause"}, "PrepareSandbox") + } + if err := imagecache.UnmountAllUnder(ateompath.OCIBundleDir(req.GetActorUid())); err != nil { + slog.WarnContext(cleanupCtx, "Failed to unmount rootfs after PrepareSandbox failure", "actorUID", req.GetActorUid(), "err", err) + } + if networkConfigured { + if err := ateomnet.CleanupActorNetwork(cleanupCtx, s.interiorNetNS); err != nil { + slog.WarnContext(cleanupCtx, "Failed to clean up actor network after PrepareSandbox failure", slog.Any("err", err)) + } + } + }() + + if err := ateomnet.SetupActorNetwork(ctx, ateomnet.NetworkConfig{ + InteriorNetNS: s.interiorNetNS, + DumpNetInfo: true, + EgressRedirectPort: s.egressRedirectPort(req.GetRedirectEgress()), + }); err != nil { + return nil, fmt.Errorf("while setting up actor network: %w", err) + } + networkConfigured = true + if err := imagecache.SetupBundleRootfs(ateompath.OCIBundlePath(req.GetActorUid(), "pause")); err != nil { + return nil, fmt.Errorf("while composing pause rootfs: %w", err) + } + if err := rcmd.cmdCreate(ctx, os.Stdout, "pause", nil); err != nil { + return nil, fmt.Errorf("while creating pause container: %w", err) + } + rootCreated = true + if err := rcmd.cmdStart(ctx, os.Stdout, "pause"); err != nil { + return nil, fmt.Errorf("while starting pause container: %w", err) + } + + s.prepared = &preparedSandbox{ + actorUID: req.GetActorUid(), + runscPath: req.GetRunscPath(), + redirectEgress: req.GetRedirectEgress(), + cpuMilli: req.GetCpuMilli(), + memoryBytes: req.GetMemoryBytes(), + } + return &ateompb.PrepareSandboxResponse{}, nil +} + +// DiscardPreparedSandbox releases a root sandbox when another concurrent +// preparation step failed before RunWorkload could use it. +func (s *AteomService) DiscardPreparedSandbox(ctx context.Context, req *ateompb.DiscardPreparedSandboxRequest) (*ateompb.DiscardPreparedSandboxResponse, error) { + s.lock.Lock() + defer s.lock.Unlock() + if s.prepared == nil { + return &ateompb.DiscardPreparedSandboxResponse{}, nil + } + if req.GetActorUid() != s.prepared.actorUID { + return nil, status.Errorf(codes.FailedPrecondition, "prepared sandbox belongs to actor %q", s.prepared.actorUID) + } + + p := s.prepared + s.activeActor.Store(nil) + rcmd := &runsc{path: p.runscPath, actorUID: p.actorUID} + var cleanupErr error + if !p.rootDeleted { + if err := rcmd.cmdDelete(ctx, "pause"); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("while deleting prepared pause container: %w", err)) + } else { + p.rootDeleted = true + } + } + if err := imagecache.UnmountAllUnder(ateompath.OCIBundleDir(p.actorUID)); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("while unmounting prepared sandbox rootfs: %w", err)) + } + if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("while cleaning up prepared sandbox network: %w", err)) + } + if cleanupErr == nil { + s.prepared = nil + } + return &ateompb.DiscardPreparedSandboxResponse{}, cleanupErr +} + +func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkloadRequest) (resp *ateompb.RunWorkloadResponse, retErr error) { + s.lock.Lock() + defer s.lock.Unlock() + if err := s.rejectIfDraining(); err != nil { + return nil, err + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + s.setActiveRPC(rpcRunWorkload, cancel) + defer s.clearActiveRPC() + + prepared := s.prepared != nil + if prepared && s.prepared.rootDeleted { + return nil, status.Error(codes.FailedPrecondition, "prepared sandbox cleanup is incomplete") + } + if prepared && !s.prepared.matches(req.GetActorUid(), req.GetRunscPath(), req.GetEgressGateway() != nil, req.GetCpuMilli(), req.GetMemoryBytes()) { + return nil, status.Error(codes.FailedPrecondition, "prepared sandbox does not match RunWorkload request") + } + if !prepared { + if err := s.deactivateActorNetworking(ctx); err != nil { + return nil, err + } + } actorRef := resources.ActorRef{Atespace: req.GetAtespace(), Name: req.GetActorName()} s.actorLogger.EmitLifecycleLog("Actor starting", actorRef, req.GetActorUid(), req.GetActorTemplateNamespace(), req.GetActorTemplateName()) @@ -612,15 +765,15 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload if err != nil { return nil, err } - if err := ateomnet.SetupActorNetwork(ctx, ateomnet.NetworkConfig{ - InteriorNetNS: s.interiorNetNS, - DumpNetInfo: true, - EgressRedirectPort: s.egressRedirectPort(req.GetEgressGateway() != nil), - }); err != nil { - // Cleared here as well as in the deferred cleanup below, because that - // defer is not registered until after this check. - s.activeActor.Store(nil) - return nil, fmt.Errorf("while setting up actor network: %w", err) + if !prepared { + if err := ateomnet.SetupActorNetwork(ctx, ateomnet.NetworkConfig{ + InteriorNetNS: s.interiorNetNS, + DumpNetInfo: true, + EgressRedirectPort: s.egressRedirectPort(req.GetEgressGateway() != nil), + }); err != nil { + s.activeActor.Store(nil) + return nil, fmt.Errorf("while setting up actor network: %w", err) + } } rcmd := &runsc{ path: req.GetRunscPath(), @@ -628,21 +781,21 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload size: sizing.FromLimits(req.GetCpuMilli(), req.GetMemoryBytes()), } var containersToDelete []string + if prepared { + containersToDelete = append(containersToDelete, "pause") + s.prepared = nil + } defer func() { if retErr != nil { s.activeActor.Store(nil) - cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) - defer cancel() + cleanupCtx, cleanupCancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cleanupCancel() if err := s.deactivateActorNetworking(cleanupCtx); err != nil { slog.WarnContext(cleanupCtx, "Failed to deactivate actor networking after Run failure", slog.Any("err", err)) } deleteContainers(cleanupCtx, rcmd, containersToDelete, "Run") - // Detach any bundle rootfs overlays a partially-completed setup - // mounted, mirroring the post-checkpoint cleanup — otherwise they - // linger in this namespace until atelet wipes the bundle dirs. - // Run before the network cleanup. if err := imagecache.UnmountAllUnder(ateompath.OCIBundleDir(req.GetActorUid())); err != nil { - slog.WarnContext(ctx, "Failed to unmount bundle rootfs overlays after Run failure", + slog.WarnContext(cleanupCtx, "Failed to unmount bundle rootfs overlays after Run failure", "actorUID", req.GetActorUid(), "err", err) } if err := ateomnet.CleanupActorNetwork(cleanupCtx, s.interiorNetNS); err != nil { @@ -650,20 +803,22 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload } } }() - // Create and start pause container. The bundle rootfs is composed here — - // an overlay of the node's cached image layers plus the bundle's private - // upper — because mounting is ateom's job (atelet runs with no - // capabilities); runsc's gofer resolves the mount in this pod's mount - // namespace. - if err := imagecache.SetupBundleRootfs(ateompath.OCIBundlePath(req.GetActorUid(), "pause")); err != nil { - return nil, fmt.Errorf("while composing pause rootfs: %w", err) - } - containersToDelete = append(containersToDelete, "pause") - if err := rcmd.cmdCreate(ctx, os.Stdout, "pause", nil); err != nil { - return nil, fmt.Errorf("while creating pause container: %w", err) - } - if err := rcmd.cmdStart(ctx, os.Stdout, "pause"); err != nil { - return nil, fmt.Errorf("while starting pause container: %w", err) + if !prepared { + // Create and start pause container. The bundle rootfs is composed here — + // an overlay of the node's cached image layers plus the bundle's private + // upper — because mounting is ateom's job (atelet runs with no + // capabilities); runsc's gofer resolves the mount in this pod's mount + // namespace. + if err := imagecache.SetupBundleRootfs(ateompath.OCIBundlePath(req.GetActorUid(), "pause")); err != nil { + return nil, fmt.Errorf("while composing pause rootfs: %w", err) + } + containersToDelete = append(containersToDelete, "pause") + if err := rcmd.cmdCreate(ctx, os.Stdout, "pause", nil); err != nil { + return nil, fmt.Errorf("while creating pause container: %w", err) + } + if err := rcmd.cmdStart(ctx, os.Stdout, "pause"); err != nil { + return nil, fmt.Errorf("while starting pause container: %w", err) + } } // Create and start each application container, each with its own log pipe so @@ -708,6 +863,9 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.CheckpointWorkloadRequest) (*ateompb.CheckpointWorkloadResponse, error) { s.lock.Lock() defer s.lock.Unlock() + if s.prepared != nil { + return nil, status.Error(codes.FailedPrecondition, "cannot checkpoint a prepared sandbox") + } ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -862,6 +1020,9 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore if err := s.rejectIfDraining(); err != nil { return nil, err } + if s.prepared != nil { + return nil, status.Error(codes.FailedPrecondition, "cannot restore over a prepared sandbox") + } ctx, cancel := context.WithCancel(ctx) defer cancel() diff --git a/cmd/ateom-gvisor/prepared_test.go b/cmd/ateom-gvisor/prepared_test.go new file mode 100644 index 0000000000..4ece4635aa --- /dev/null +++ b/cmd/ateom-gvisor/prepared_test.go @@ -0,0 +1,229 @@ +//go:build linux + +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "runtime" + "testing" + + "github.com/opencontainers/runtime-spec/specs-go" + "github.com/vishvananda/netns" + "golang.org/x/sys/unix" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/agent-substrate/substrate/internal/actorlog" + "github.com/agent-substrate/substrate/internal/ateomnet" + "github.com/agent-substrate/substrate/internal/ateompath" + "github.com/agent-substrate/substrate/internal/atunnel" + "github.com/agent-substrate/substrate/internal/imagecache" + "github.com/agent-substrate/substrate/internal/proto/ateompb" + "github.com/agent-substrate/substrate/internal/roottest" +) + +func TestPreparedSandboxRejectsMismatchedRun(t *testing.T) { + s := &AteomService{ + lock: newCancelableMutex(), + prepared: &preparedSandbox{ + actorUID: "actor-a", + runscPath: "/runsc", + cpuMilli: 1000, + memoryBytes: 1 << 30, + }, + } + req := &ateompb.PrepareSandboxRequest{ActorUid: "actor-a", RunscPath: "/runsc", CpuMilli: 1000, MemoryBytes: 1 << 30} + if _, err := s.PrepareSandbox(context.Background(), req); err != nil { + t.Fatalf("idempotent PrepareSandbox: %v", err) + } + _, err := s.RunWorkload(context.Background(), &ateompb.RunWorkloadRequest{ + ActorUid: "actor-a", RunscPath: "/runsc", CpuMilli: 500, MemoryBytes: 1 << 30, + }) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("mismatched RunWorkload code = %v, want %v", status.Code(err), codes.FailedPrecondition) + } +} + +// TestPreparedSandboxLifecycleWithRunsc crosses the actual gVisor boundary: +// PrepareSandbox boots the Sentry, then RunWorkload adds an application to it. +// Set RUNSC_TEST_BINARY and run this test as root (a throwaway user/net/mount +// namespace is sufficient). +func TestPreparedSandboxLifecycleWithRunsc(t *testing.T) { + runscPath := os.Getenv("RUNSC_TEST_BINARY") + if runscPath == "" { + t.Skip("set RUNSC_TEST_BINARY to exercise the real gVisor lifecycle") + } + roottest.Require(t, "creating network namespaces, veth pairs, and nftables rules") + runscPath, err := filepath.Abs(runscPath) + if err != nil { + t.Fatal(err) + } + + withPreparedSandboxNetNS(t, func(interior netns.NsHandle) { + oldActorsDir := ateompath.ActorsDir + ateompath.ActorsDir = t.TempDir() + defer func() { ateompath.ActorsDir = oldActorsDir }() + + const actorUID = "prepared-sandbox-test" + for _, dir := range []string{ateompath.PIDFileDir(actorUID), ateompath.RunSCStateDir(actorUID)} { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + } + writeRunscTestBundle(t, actorUID, "pause", "sandbox") + writeRunscTestBundle(t, actorUID, "app", "container") + + wrapper := filepath.Join(t.TempDir(), "runsc") + script := fmt.Sprintf("#!/bin/sh\n"+ + "previous=\n"+ + "for argument in \"$@\"; do\n"+ + " if [ \"$previous\" = \"-bundle\" ]; then\n"+ + " sed -i '/\"cgroupsPath\":/d' \"$argument/config.json\"\n"+ + " fi\n"+ + " previous=\"$argument\"\n"+ + "done\n"+ + "exec %q --TESTONLY-unsafe-nonroot --network=none --ignore-cgroups \"$@\"\n", runscPath) + if err := os.WriteFile(wrapper, []byte(script), 0o700); err != nil { + t.Fatal(err) + } + + s := NewService(interior, actorlog.NewActorLogger(io.Discard, false), &atunnel.Server{}, &atunnel.Egress{}, 0, "", "", "") + ctx := context.Background() + rcmd := &runsc{path: wrapper, actorUID: actorUID} + if _, err := s.PrepareSandbox(ctx, &ateompb.PrepareSandboxRequest{ActorUid: actorUID, RunscPath: wrapper}); err != nil { + t.Fatalf("PrepareSandbox: %v", err) + } + defer func() { + _ = rcmd.cmdDelete(context.Background(), "app") + _ = rcmd.cmdDelete(context.Background(), "pause") + _ = unix.Unmount(filepath.Join(ateompath.RunSCStateDir(actorUID), "null-netns"), unix.MNT_DETACH) + _ = imagecache.UnmountAllUnder(ateompath.OCIBundleDir(actorUID)) + _ = ateomnet.CleanupActorNetwork(context.Background(), interior) + }() + if err := rcmd.cmdState(ctx, "pause"); err != nil { + t.Fatalf("prepared pause container is not running: %v", err) + } + // Fail after the pause container is deleted, then verify that retrying + // finishes the remaining cleanup without trying to delete it again. + closedNetNS, err := netns.Get() + if err != nil { + t.Fatal(err) + } + closedNetNS.Close() + s.interiorNetNS = closedNetNS + if _, err := s.DiscardPreparedSandbox(ctx, &ateompb.DiscardPreparedSandboxRequest{ActorUid: actorUID}); err == nil { + t.Fatal("DiscardPreparedSandbox succeeded with a closed network namespace") + } + s.interiorNetNS = interior + if _, err := s.DiscardPreparedSandbox(ctx, &ateompb.DiscardPreparedSandboxRequest{ActorUid: actorUID}); err != nil { + t.Fatalf("retrying DiscardPreparedSandbox: %v", err) + } + if err := rcmd.cmdState(ctx, "pause"); err == nil { + t.Fatal("pause container still exists after DiscardPreparedSandbox") + } + if _, err := s.PrepareSandbox(ctx, &ateompb.PrepareSandboxRequest{ActorUid: actorUID, RunscPath: wrapper}); err != nil { + t.Fatalf("PrepareSandbox after discard: %v", err) + } + if err := rcmd.cmdState(ctx, "app"); err == nil { + t.Fatal("application started before RunWorkload") + } + + if _, err := s.RunWorkload(ctx, &ateompb.RunWorkloadRequest{ + Atespace: "test", ActorName: "actor", ActorUid: actorUID, RunscPath: wrapper, + Spec: &ateompb.WorkloadSpec{Containers: []*ateompb.Container{{Name: "app"}}}, + }); err != nil { + t.Fatalf("RunWorkload: %v", err) + } + if err := rcmd.cmdState(ctx, "app"); err != nil { + t.Fatalf("application did not join the prepared sandbox: %v", err) + } + }) +} + +func withPreparedSandboxNetNS(t *testing.T, fn func(netns.NsHandle)) { + t.Helper() + runtime.LockOSThread() + defer runtime.UnlockOSThread() + original, err := netns.Get() + if err != nil { + t.Fatal(err) + } + defer original.Close() + defer func() { + if err := netns.Set(original); err != nil { + t.Errorf("restoring original netns: %v", err) + } + }() + pod, err := netns.New() + if err != nil { + t.Fatal(err) + } + defer pod.Close() + interior, err := netns.New() + if err != nil { + t.Fatal(err) + } + defer interior.Close() + if err := netns.Set(pod); err != nil { + t.Fatal(err) + } + fn(interior) +} + +func writeRunscTestBundle(t *testing.T, actorUID, name, containerType string) { + t.Helper() + bundle := ateompath.OCIBundlePath(actorUID, name) + binDir := filepath.Join(bundle, "rootfs", "bin") + if err := os.MkdirAll(binDir, 0o700); err != nil { + t.Fatal(err) + } + busybox, err := os.ReadFile("/bin/busybox") + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(binDir, "busybox"), busybox, 0o700); err != nil { + t.Fatal(err) + } + annotations := map[string]string{ + "io.kubernetes.cri.container-type": containerType, + "io.kubernetes.cri.container-name": name, + } + if containerType == "container" { + annotations["io.kubernetes.cri.sandbox-id"] = "pause" + } + spec := specs.Spec{ + Version: specs.Version, + Process: &specs.Process{User: specs.User{}, Args: []string{"/bin/busybox", "sh", "-c", "while :; do /bin/busybox sleep 3600; done"}, Cwd: "/"}, + Root: &specs.Root{Path: "rootfs"}, + Linux: &specs.Linux{Namespaces: []specs.LinuxNamespace{ + {Type: specs.PIDNamespace}, {Type: specs.NetworkNamespace}, {Type: specs.IPCNamespace}, {Type: specs.UTSNamespace}, {Type: specs.MountNamespace}, + }}, + Annotations: annotations, + } + data, err := json.Marshal(spec) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(bundle, "config.json"), data, 0o600); err != nil { + t.Fatal(err) + } +} diff --git a/internal/proto/ateompb/ateom.pb.go b/internal/proto/ateompb/ateom.pb.go index c865d0a1c6..634e160c96 100644 --- a/internal/proto/ateompb/ateom.pb.go +++ b/internal/proto/ateompb/ateom.pb.go @@ -1513,6 +1513,201 @@ func (*GetActiveWorkloadStatsResponse_Sample) isGetActiveWorkloadStatsResponse_R func (*GetActiveWorkloadStatsResponse_NoSampleReason) isGetActiveWorkloadStatsResponse_Result() {} +type PrepareSandboxRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + ActorUid string `protobuf:"bytes,1,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + RunscPath string `protobuf:"bytes,2,opt,name=runsc_path,json=runscPath,proto3" json:"runsc_path,omitempty"` + // Whether the actor network needs transparent egress redirection. + RedirectEgress bool `protobuf:"varint,3,opt,name=redirect_egress,json=redirectEgress,proto3" json:"redirect_egress,omitempty"` + // The actor's declared size. The pause/root container is created by this RPC, + // so it must receive the same limits as the later RunWorkload call. + CpuMilli int64 `protobuf:"varint,4,opt,name=cpu_milli,json=cpuMilli,proto3" json:"cpu_milli,omitempty"` // CPU limit in millicores (1000 = one core). + MemoryBytes int64 `protobuf:"varint,5,opt,name=memory_bytes,json=memoryBytes,proto3" json:"memory_bytes,omitempty"` // Memory limit in bytes. + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PrepareSandboxRequest) Reset() { + *x = PrepareSandboxRequest{} + mi := &file_ateom_proto_msgTypes[17] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PrepareSandboxRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PrepareSandboxRequest) ProtoMessage() {} + +func (x *PrepareSandboxRequest) ProtoReflect() protoreflect.Message { + mi := &file_ateom_proto_msgTypes[17] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PrepareSandboxRequest.ProtoReflect.Descriptor instead. +func (*PrepareSandboxRequest) Descriptor() ([]byte, []int) { + return file_ateom_proto_rawDescGZIP(), []int{17} +} + +func (x *PrepareSandboxRequest) GetActorUid() string { + if x != nil { + return x.ActorUid + } + return "" +} + +func (x *PrepareSandboxRequest) GetRunscPath() string { + if x != nil { + return x.RunscPath + } + return "" +} + +func (x *PrepareSandboxRequest) GetRedirectEgress() bool { + if x != nil { + return x.RedirectEgress + } + return false +} + +func (x *PrepareSandboxRequest) GetCpuMilli() int64 { + if x != nil { + return x.CpuMilli + } + return 0 +} + +func (x *PrepareSandboxRequest) GetMemoryBytes() int64 { + if x != nil { + return x.MemoryBytes + } + return 0 +} + +type PrepareSandboxResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *PrepareSandboxResponse) Reset() { + *x = PrepareSandboxResponse{} + mi := &file_ateom_proto_msgTypes[18] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *PrepareSandboxResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PrepareSandboxResponse) ProtoMessage() {} + +func (x *PrepareSandboxResponse) ProtoReflect() protoreflect.Message { + mi := &file_ateom_proto_msgTypes[18] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PrepareSandboxResponse.ProtoReflect.Descriptor instead. +func (*PrepareSandboxResponse) Descriptor() ([]byte, []int) { + return file_ateom_proto_rawDescGZIP(), []int{18} +} + +type DiscardPreparedSandboxRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + ActorUid string `protobuf:"bytes,1,opt,name=actor_uid,json=actorUid,proto3" json:"actor_uid,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *DiscardPreparedSandboxRequest) Reset() { + *x = DiscardPreparedSandboxRequest{} + mi := &file_ateom_proto_msgTypes[19] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *DiscardPreparedSandboxRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DiscardPreparedSandboxRequest) ProtoMessage() {} + +func (x *DiscardPreparedSandboxRequest) ProtoReflect() protoreflect.Message { + mi := &file_ateom_proto_msgTypes[19] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DiscardPreparedSandboxRequest.ProtoReflect.Descriptor instead. +func (*DiscardPreparedSandboxRequest) Descriptor() ([]byte, []int) { + return file_ateom_proto_rawDescGZIP(), []int{19} +} + +func (x *DiscardPreparedSandboxRequest) GetActorUid() string { + if x != nil { + return x.ActorUid + } + return "" +} + +type DiscardPreparedSandboxResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *DiscardPreparedSandboxResponse) Reset() { + *x = DiscardPreparedSandboxResponse{} + mi := &file_ateom_proto_msgTypes[20] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *DiscardPreparedSandboxResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DiscardPreparedSandboxResponse) ProtoMessage() {} + +func (x *DiscardPreparedSandboxResponse) ProtoReflect() protoreflect.Message { + mi := &file_ateom_proto_msgTypes[20] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DiscardPreparedSandboxResponse.ProtoReflect.Descriptor instead. +func (*DiscardPreparedSandboxResponse) Descriptor() ([]byte, []int) { + return file_ateom_proto_rawDescGZIP(), []int{20} +} + var File_ateom_proto protoreflect.FileDescriptor const file_ateom_proto_rawDesc = "" + @@ -1624,7 +1819,18 @@ const file_ateom_proto_rawDesc = "" + "\x1eGetActiveWorkloadStatsResponse\x124\n" + "\x06sample\x18\x01 \x01(\v2\x1a.ateom.WorkloadStatsSampleH\x00R\x06sample\x12A\n" + "\x10no_sample_reason\x18\x02 \x01(\x0e2\x15.ateom.NoSampleReasonH\x00R\x0enoSampleReasonB\b\n" + - "\x06result*\x84\x01\n" + + "\x06result\"\xbc\x01\n" + + "\x15PrepareSandboxRequest\x12\x1b\n" + + "\tactor_uid\x18\x01 \x01(\tR\bactorUid\x12\x1d\n" + + "\n" + + "runsc_path\x18\x02 \x01(\tR\trunscPath\x12'\n" + + "\x0fredirect_egress\x18\x03 \x01(\bR\x0eredirectEgress\x12\x1b\n" + + "\tcpu_milli\x18\x04 \x01(\x03R\bcpuMilli\x12!\n" + + "\fmemory_bytes\x18\x05 \x01(\x03R\vmemoryBytes\"\x18\n" + + "\x16PrepareSandboxResponse\"<\n" + + "\x1dDiscardPreparedSandboxRequest\x12\x1b\n" + + "\tactor_uid\x18\x01 \x01(\tR\bactorUid\" \n" + + "\x1eDiscardPreparedSandboxResponse*\x84\x01\n" + "\rSnapshotScope\x12\x1e\n" + "\x1aSNAPSHOT_SCOPE_UNSPECIFIED\x10\x00\x12\x17\n" + "\x13SNAPSHOT_SCOPE_FULL\x10\x01\x12\x17\n" + @@ -1641,8 +1847,10 @@ const file_ateom_proto_rawDesc = "" + "\x0eNoSampleReason\x12 \n" + "\x1cNO_SAMPLE_REASON_UNSPECIFIED\x10\x00\x12 \n" + "\x1cNO_SAMPLE_REASON_NO_WORKLOAD\x10\x01\x12'\n" + - "#NO_SAMPLE_REASON_NOT_MEASURABLE_YET\x10\x022\xc0\x03\n" + - "\x05Ateom\x12F\n" + + "#NO_SAMPLE_REASON_NOT_MEASURABLE_YET\x10\x022\xfa\x04\n" + + "\x05Ateom\x12O\n" + + "\x0ePrepareSandbox\x12\x1c.ateom.PrepareSandboxRequest\x1a\x1d.ateom.PrepareSandboxResponse\"\x00\x12g\n" + + "\x16DiscardPreparedSandbox\x12$.ateom.DiscardPreparedSandboxRequest\x1a%.ateom.DiscardPreparedSandboxResponse\"\x00\x12F\n" + "\vRunWorkload\x12\x19.ateom.RunWorkloadRequest\x1a\x1a.ateom.RunWorkloadResponse\"\x00\x12[\n" + "\x12CheckpointWorkload\x12 .ateom.CheckpointWorkloadRequest\x1a!.ateom.CheckpointWorkloadResponse\"\x00\x12R\n" + "\x0fRestoreWorkload\x12\x1d.ateom.RestoreWorkloadRequest\x1a\x1e.ateom.RestoreWorkloadResponse\"\x00\x12U\n" + @@ -1662,7 +1870,7 @@ func file_ateom_proto_rawDescGZIP() []byte { } var file_ateom_proto_enumTypes = make([]protoimpl.EnumInfo, 4) -var file_ateom_proto_msgTypes = make([]protoimpl.MessageInfo, 20) +var file_ateom_proto_msgTypes = make([]protoimpl.MessageInfo, 24) var file_ateom_proto_goTypes = []any{ (SnapshotScope)(0), // 0: ateom.SnapshotScope (SandboxClass)(0), // 1: ateom.SandboxClass @@ -1685,23 +1893,27 @@ var file_ateom_proto_goTypes = []any{ (*GetWorkloadStatsResponse)(nil), // 18: ateom.GetWorkloadStatsResponse (*GetActiveWorkloadStatsRequest)(nil), // 19: ateom.GetActiveWorkloadStatsRequest (*GetActiveWorkloadStatsResponse)(nil), // 20: ateom.GetActiveWorkloadStatsResponse - nil, // 21: ateom.RunWorkloadRequest.RuntimeAssetPathsEntry - nil, // 22: ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry - nil, // 23: ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry + (*PrepareSandboxRequest)(nil), // 21: ateom.PrepareSandboxRequest + (*PrepareSandboxResponse)(nil), // 22: ateom.PrepareSandboxResponse + (*DiscardPreparedSandboxRequest)(nil), // 23: ateom.DiscardPreparedSandboxRequest + (*DiscardPreparedSandboxResponse)(nil), // 24: ateom.DiscardPreparedSandboxResponse + nil, // 25: ateom.RunWorkloadRequest.RuntimeAssetPathsEntry + nil, // 26: ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry + nil, // 27: ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry } var file_ateom_proto_depIdxs = []int32{ 6, // 0: ateom.RunWorkloadRequest.spec:type_name -> ateom.WorkloadSpec - 21, // 1: ateom.RunWorkloadRequest.runtime_asset_paths:type_name -> ateom.RunWorkloadRequest.RuntimeAssetPathsEntry + 25, // 1: ateom.RunWorkloadRequest.runtime_asset_paths:type_name -> ateom.RunWorkloadRequest.RuntimeAssetPathsEntry 5, // 2: ateom.RunWorkloadRequest.egress_gateway:type_name -> ateom.EgressGateway 7, // 3: ateom.WorkloadSpec.containers:type_name -> ateom.Container 9, // 4: ateom.Container.readyz:type_name -> ateom.Readyz 8, // 5: ateom.Container.durable_dir_volume_mounts:type_name -> ateom.DurableDirVolumeMount 10, // 6: ateom.Readyz.http_get:type_name -> ateom.HTTPGetAction 6, // 7: ateom.CheckpointWorkloadRequest.spec:type_name -> ateom.WorkloadSpec - 22, // 8: ateom.CheckpointWorkloadRequest.runtime_asset_paths:type_name -> ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry + 26, // 8: ateom.CheckpointWorkloadRequest.runtime_asset_paths:type_name -> ateom.CheckpointWorkloadRequest.RuntimeAssetPathsEntry 0, // 9: ateom.CheckpointWorkloadRequest.scope:type_name -> ateom.SnapshotScope 6, // 10: ateom.RestoreWorkloadRequest.spec:type_name -> ateom.WorkloadSpec - 23, // 11: ateom.RestoreWorkloadRequest.runtime_asset_paths:type_name -> ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry + 27, // 11: ateom.RestoreWorkloadRequest.runtime_asset_paths:type_name -> ateom.RestoreWorkloadRequest.RuntimeAssetPathsEntry 0, // 12: ateom.RestoreWorkloadRequest.scope:type_name -> ateom.SnapshotScope 5, // 13: ateom.RestoreWorkloadRequest.egress_gateway:type_name -> ateom.EgressGateway 1, // 14: ateom.WorkloadStatsSample.sandbox_class:type_name -> ateom.SandboxClass @@ -1709,18 +1921,22 @@ var file_ateom_proto_depIdxs = []int32{ 17, // 16: ateom.GetWorkloadStatsResponse.sample:type_name -> ateom.WorkloadStatsSample 17, // 17: ateom.GetActiveWorkloadStatsResponse.sample:type_name -> ateom.WorkloadStatsSample 3, // 18: ateom.GetActiveWorkloadStatsResponse.no_sample_reason:type_name -> ateom.NoSampleReason - 4, // 19: ateom.Ateom.RunWorkload:input_type -> ateom.RunWorkloadRequest - 12, // 20: ateom.Ateom.CheckpointWorkload:input_type -> ateom.CheckpointWorkloadRequest - 14, // 21: ateom.Ateom.RestoreWorkload:input_type -> ateom.RestoreWorkloadRequest - 16, // 22: ateom.Ateom.GetWorkloadStats:input_type -> ateom.GetWorkloadStatsRequest - 19, // 23: ateom.Ateom.GetActiveWorkloadStats:input_type -> ateom.GetActiveWorkloadStatsRequest - 11, // 24: ateom.Ateom.RunWorkload:output_type -> ateom.RunWorkloadResponse - 13, // 25: ateom.Ateom.CheckpointWorkload:output_type -> ateom.CheckpointWorkloadResponse - 15, // 26: ateom.Ateom.RestoreWorkload:output_type -> ateom.RestoreWorkloadResponse - 18, // 27: ateom.Ateom.GetWorkloadStats:output_type -> ateom.GetWorkloadStatsResponse - 20, // 28: ateom.Ateom.GetActiveWorkloadStats:output_type -> ateom.GetActiveWorkloadStatsResponse - 24, // [24:29] is the sub-list for method output_type - 19, // [19:24] is the sub-list for method input_type + 21, // 19: ateom.Ateom.PrepareSandbox:input_type -> ateom.PrepareSandboxRequest + 23, // 20: ateom.Ateom.DiscardPreparedSandbox:input_type -> ateom.DiscardPreparedSandboxRequest + 4, // 21: ateom.Ateom.RunWorkload:input_type -> ateom.RunWorkloadRequest + 12, // 22: ateom.Ateom.CheckpointWorkload:input_type -> ateom.CheckpointWorkloadRequest + 14, // 23: ateom.Ateom.RestoreWorkload:input_type -> ateom.RestoreWorkloadRequest + 16, // 24: ateom.Ateom.GetWorkloadStats:input_type -> ateom.GetWorkloadStatsRequest + 19, // 25: ateom.Ateom.GetActiveWorkloadStats:input_type -> ateom.GetActiveWorkloadStatsRequest + 22, // 26: ateom.Ateom.PrepareSandbox:output_type -> ateom.PrepareSandboxResponse + 24, // 27: ateom.Ateom.DiscardPreparedSandbox:output_type -> ateom.DiscardPreparedSandboxResponse + 11, // 28: ateom.Ateom.RunWorkload:output_type -> ateom.RunWorkloadResponse + 13, // 29: ateom.Ateom.CheckpointWorkload:output_type -> ateom.CheckpointWorkloadResponse + 15, // 30: ateom.Ateom.RestoreWorkload:output_type -> ateom.RestoreWorkloadResponse + 18, // 31: ateom.Ateom.GetWorkloadStats:output_type -> ateom.GetWorkloadStatsResponse + 20, // 32: ateom.Ateom.GetActiveWorkloadStats:output_type -> ateom.GetActiveWorkloadStatsResponse + 26, // [26:33] is the sub-list for method output_type + 19, // [19:26] is the sub-list for method input_type 19, // [19:19] is the sub-list for extension type_name 19, // [19:19] is the sub-list for extension extendee 0, // [0:19] is the sub-list for field type_name @@ -1743,7 +1959,7 @@ func file_ateom_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_ateom_proto_rawDesc), len(file_ateom_proto_rawDesc)), NumEnums: 4, - NumMessages: 20, + NumMessages: 24, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/proto/ateompb/ateom.proto b/internal/proto/ateompb/ateom.proto index 9ec80a232d..14207250cc 100644 --- a/internal/proto/ateompb/ateom.proto +++ b/internal/proto/ateompb/ateom.proto @@ -21,7 +21,8 @@ option go_package = "github.com/agent-substrate/substrate/internal/proto/ateompb // Ateom is the interface to control a single gVisor (or, in the future microVM) // guest inside a worker pod. // -// Each ateom server has two main states, "available" and "executing". +// Each ateom server may be available, have a prepared root sandbox, or be +// executing a workload. // // When the ateom is "available", the substrate control plane is free to either // boot a new workload (using RunWorkload), or restore an existing workload from @@ -32,6 +33,14 @@ option go_package = "github.com/agent-substrate/substrate/internal/proto/ateompb // running workload (with CheckpointWorkload). This moves the ateom back to // "free" state. service Ateom { + // PrepareSandbox starts the root sandbox before application images finish + // downloading. RunWorkload later attaches and starts the applications. + rpc PrepareSandbox(PrepareSandboxRequest) returns (PrepareSandboxResponse) {} + + // DiscardPreparedSandbox tears down a sandbox that cannot be used because + // another concurrent preparation step failed. + rpc DiscardPreparedSandbox(DiscardPreparedSandboxRequest) returns (DiscardPreparedSandboxResponse) {} + // RunWorkload tells ateom to begin running a new workload (one or more // containers, potentially with shared filesystems). rpc RunWorkload(RunWorkloadRequest) returns (RunWorkloadResponse) {} @@ -401,3 +410,26 @@ message GetActiveWorkloadStatsResponse { NoSampleReason no_sample_reason = 2; } } + +message PrepareSandboxRequest { + string actor_uid = 1; + string runsc_path = 2; + + // Whether the actor network needs transparent egress redirection. + bool redirect_egress = 3; + + // The actor's declared size. The pause/root container is created by this RPC, + // so it must receive the same limits as the later RunWorkload call. + int64 cpu_milli = 4; // CPU limit in millicores (1000 = one core). + int64 memory_bytes = 5; // Memory limit in bytes. +} + +message PrepareSandboxResponse { +} + +message DiscardPreparedSandboxRequest { + string actor_uid = 1; +} + +message DiscardPreparedSandboxResponse { +} diff --git a/internal/proto/ateompb/ateom_grpc.pb.go b/internal/proto/ateompb/ateom_grpc.pb.go index f3be1ce8b8..6da31c2bd3 100644 --- a/internal/proto/ateompb/ateom_grpc.pb.go +++ b/internal/proto/ateompb/ateom_grpc.pb.go @@ -33,6 +33,8 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( + Ateom_PrepareSandbox_FullMethodName = "/ateom.Ateom/PrepareSandbox" + Ateom_DiscardPreparedSandbox_FullMethodName = "/ateom.Ateom/DiscardPreparedSandbox" Ateom_RunWorkload_FullMethodName = "/ateom.Ateom/RunWorkload" Ateom_CheckpointWorkload_FullMethodName = "/ateom.Ateom/CheckpointWorkload" Ateom_RestoreWorkload_FullMethodName = "/ateom.Ateom/RestoreWorkload" @@ -47,7 +49,8 @@ const ( // Ateom is the interface to control a single gVisor (or, in the future microVM) // guest inside a worker pod. // -// Each ateom server has two main states, "available" and "executing". +// Each ateom server may be available, have a prepared root sandbox, or be +// executing a workload. // // When the ateom is "available", the substrate control plane is free to either // boot a new workload (using RunWorkload), or restore an existing workload from @@ -58,6 +61,12 @@ const ( // running workload (with CheckpointWorkload). This moves the ateom back to // "free" state. type AteomClient interface { + // PrepareSandbox starts the root sandbox before application images finish + // downloading. RunWorkload later attaches and starts the applications. + PrepareSandbox(ctx context.Context, in *PrepareSandboxRequest, opts ...grpc.CallOption) (*PrepareSandboxResponse, error) + // DiscardPreparedSandbox tears down a sandbox that cannot be used because + // another concurrent preparation step failed. + DiscardPreparedSandbox(ctx context.Context, in *DiscardPreparedSandboxRequest, opts ...grpc.CallOption) (*DiscardPreparedSandboxResponse, error) // RunWorkload tells ateom to begin running a new workload (one or more // containers, potentially with shared filesystems). RunWorkload(ctx context.Context, in *RunWorkloadRequest, opts ...grpc.CallOption) (*RunWorkloadResponse, error) @@ -130,6 +139,26 @@ func NewAteomClient(cc grpc.ClientConnInterface) AteomClient { return &ateomClient{cc} } +func (c *ateomClient) PrepareSandbox(ctx context.Context, in *PrepareSandboxRequest, opts ...grpc.CallOption) (*PrepareSandboxResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(PrepareSandboxResponse) + err := c.cc.Invoke(ctx, Ateom_PrepareSandbox_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + +func (c *ateomClient) DiscardPreparedSandbox(ctx context.Context, in *DiscardPreparedSandboxRequest, opts ...grpc.CallOption) (*DiscardPreparedSandboxResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(DiscardPreparedSandboxResponse) + err := c.cc.Invoke(ctx, Ateom_DiscardPreparedSandbox_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + func (c *ateomClient) RunWorkload(ctx context.Context, in *RunWorkloadRequest, opts ...grpc.CallOption) (*RunWorkloadResponse, error) { cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(RunWorkloadResponse) @@ -187,7 +216,8 @@ func (c *ateomClient) GetActiveWorkloadStats(ctx context.Context, in *GetActiveW // Ateom is the interface to control a single gVisor (or, in the future microVM) // guest inside a worker pod. // -// Each ateom server has two main states, "available" and "executing". +// Each ateom server may be available, have a prepared root sandbox, or be +// executing a workload. // // When the ateom is "available", the substrate control plane is free to either // boot a new workload (using RunWorkload), or restore an existing workload from @@ -198,6 +228,12 @@ func (c *ateomClient) GetActiveWorkloadStats(ctx context.Context, in *GetActiveW // running workload (with CheckpointWorkload). This moves the ateom back to // "free" state. type AteomServer interface { + // PrepareSandbox starts the root sandbox before application images finish + // downloading. RunWorkload later attaches and starts the applications. + PrepareSandbox(context.Context, *PrepareSandboxRequest) (*PrepareSandboxResponse, error) + // DiscardPreparedSandbox tears down a sandbox that cannot be used because + // another concurrent preparation step failed. + DiscardPreparedSandbox(context.Context, *DiscardPreparedSandboxRequest) (*DiscardPreparedSandboxResponse, error) // RunWorkload tells ateom to begin running a new workload (one or more // containers, potentially with shared filesystems). RunWorkload(context.Context, *RunWorkloadRequest) (*RunWorkloadResponse, error) @@ -270,6 +306,12 @@ type AteomServer interface { // pointer dereference when methods are called. type UnimplementedAteomServer struct{} +func (UnimplementedAteomServer) PrepareSandbox(context.Context, *PrepareSandboxRequest) (*PrepareSandboxResponse, error) { + return nil, status.Error(codes.Unimplemented, "method PrepareSandbox not implemented") +} +func (UnimplementedAteomServer) DiscardPreparedSandbox(context.Context, *DiscardPreparedSandboxRequest) (*DiscardPreparedSandboxResponse, error) { + return nil, status.Error(codes.Unimplemented, "method DiscardPreparedSandbox not implemented") +} func (UnimplementedAteomServer) RunWorkload(context.Context, *RunWorkloadRequest) (*RunWorkloadResponse, error) { return nil, status.Error(codes.Unimplemented, "method RunWorkload not implemented") } @@ -306,6 +348,42 @@ func RegisterAteomServer(s grpc.ServiceRegistrar, srv AteomServer) { s.RegisterService(&Ateom_ServiceDesc, srv) } +func _Ateom_PrepareSandbox_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(PrepareSandboxRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AteomServer).PrepareSandbox(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Ateom_PrepareSandbox_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AteomServer).PrepareSandbox(ctx, req.(*PrepareSandboxRequest)) + } + return interceptor(ctx, in, info, handler) +} + +func _Ateom_DiscardPreparedSandbox_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(DiscardPreparedSandboxRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(AteomServer).DiscardPreparedSandbox(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Ateom_DiscardPreparedSandbox_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(AteomServer).DiscardPreparedSandbox(ctx, req.(*DiscardPreparedSandboxRequest)) + } + return interceptor(ctx, in, info, handler) +} + func _Ateom_RunWorkload_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(RunWorkloadRequest) if err := dec(in); err != nil { @@ -403,6 +481,14 @@ var Ateom_ServiceDesc = grpc.ServiceDesc{ ServiceName: "ateom.Ateom", HandlerType: (*AteomServer)(nil), Methods: []grpc.MethodDesc{ + { + MethodName: "PrepareSandbox", + Handler: _Ateom_PrepareSandbox_Handler, + }, + { + MethodName: "DiscardPreparedSandbox", + Handler: _Ateom_DiscardPreparedSandbox_Handler, + }, { MethodName: "RunWorkload", Handler: _Ateom_RunWorkload_Handler,