Skip to content

ateom: prepare gVisor sandbox during image pull #984

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Du Bin (dubin555) wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ateom: prepare gVisor sandbox during image pull #984

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
157 changes: 106 additions & 51 deletions cmd/atelet/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand All @@ -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.
Expand All @@ -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)
}

Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand All @@ -1456,7 +1493,7 @@ func (s *AteomHerder) prepareOCIBundles(
}
g.Go(func() error {
if err := prepareOCIDirectory(
gCtx,
gctx,
s.imageCache,
actorUID,
ctr.GetName(),
Expand All @@ -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(),
Expand All @@ -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) {
Expand Down
14 changes: 14 additions & 0 deletions cmd/atelet/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading