Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
15 changes: 15 additions & 0 deletions cmd/atelet/imagevolume_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -75,11 +75,26 @@ func singleFileLayer(t *testing.T, path, body string) v1.Layer {
// pushTestImage pushes the image and returns its digest-pinned reference,
// which is what validated workload specs carry.
func pushTestImage(t *testing.T, ref string, layers ...v1.Layer) string {
t.Helper()
return pushTestImageWithConfig(t, ref, v1.Config{}, layers...)
}

// pushTestImageWithConfig is pushTestImage with the image config set.
func pushTestImageWithConfig(t *testing.T, ref string, cfg v1.Config, layers ...v1.Layer) string {
t.Helper()
img, err := mutate.AppendLayers(empty.Image, layers...)
if err != nil {
t.Fatalf("mutate.AppendLayers: %v", err)
}
cf, err := img.ConfigFile()
if err != nil {
t.Fatalf("img.ConfigFile: %v", err)
}
cf = cf.DeepCopy()
cf.Config = cfg
if img, err = mutate.ConfigFile(img, cf); err != nil {
t.Fatalf("mutate.ConfigFile: %v", err)
}
tag, err := name.ParseReference(ref, name.Insecure)
if err != nil {
t.Fatalf("name.ParseReference(%q): %v", ref, err)
Expand Down
62 changes: 42 additions & 20 deletions cmd/atelet/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -771,7 +771,7 @@ func (s *Atelet) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *atele
if registration, err = s.systemInfoVolumes.Register(actorUID, actorRef, systemInfoVolumesFor(actorUID, req.GetSpec())); err != nil {
return nil, err
}
containerSpecs, err := s.prepareOCIBundles(ctx, actorUID, actorRef,
bundles, err := s.prepareOCIBundles(ctx, actorUID, actorRef,
req.GetSpec(), sandboxRec.PauseImage,
)
if err != nil {
Expand All @@ -783,7 +783,7 @@ func (s *Atelet) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *atele
return nil, err
}

spec, err := buildWorkerWorkloadSpec(req.GetSpec(), containerSpecs)
spec, err := buildWorkerWorkloadSpec(req.GetSpec(), bundles)
if err != nil {
return nil, apierror.InvalidArgument("invalid workload spec: %v", err)
}
Expand Down Expand Up @@ -1489,7 +1489,7 @@ func (s *Atelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (res
// fetch + image unpack hides whichever leg is shorter, and on a cold node
// (uncached assets + image, ~2.5s unpack) that overlap is large.
var assetPaths map[string]string
var containerSpecs []*ateworkerpb.ContainerSpec
var bundles *bundleSpecs
// One per leg: a single field written from both goroutines would race.
var downloadErr, prepErr error
var prepFailedPhase string
Expand Down Expand Up @@ -1523,7 +1523,7 @@ func (s *Atelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (res
return err
}
t := time.Now()
containerSpecs, err = s.prepareOCIBundles(gctx, actorUID, actorRef, req.GetSpec(), runtimeRec.PauseImage)
bundles, err = s.prepareOCIBundles(gctx, actorUID, actorRef, req.GetSpec(), runtimeRec.PauseImage)
dBundles = time.Since(t)
if err != nil {
prepFailedPhase = ateattr.SnapshotPhaseOCIUnpack
Expand All @@ -1548,7 +1548,7 @@ func (s *Atelet) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (res

// Tell worker to do runsc create + runsc restore for pause container and
// all application containers.
spec, err := buildWorkerWorkloadSpec(req.GetSpec(), containerSpecs)
spec, err := buildWorkerWorkloadSpec(req.GetSpec(), bundles)
if err != nil {
return nil, apierror.InvalidArgument("invalid workload spec: %v", err)
}
Expand Down Expand Up @@ -1808,14 +1808,14 @@ func prepareDurableDirVolume(path string) error {
// configuration, and on a restore it must be the image the snapshot was taken
// with. It is empty for sandboxes without a pause container, which get no
// pause bundle. It returns each application container's spec, indexed like
// spec.GetContainers().
// spec.GetContainers(), and the pause container's identity.
func (s *Atelet) prepareOCIBundles(
ctx context.Context,
actorUID string,
actorRef resources.ActorRef,
spec *ateletpb.WorkloadSpec,
pauseImage string,
) ([]*ateworkerpb.ContainerSpec, error) {
) (*bundleSpecs, error) {
// Prepare host folders for volume types that need them.
for _, vol := range spec.GetVolumes() {
if vol.GetDurableDir() != nil {
Expand All @@ -1827,45 +1827,65 @@ func (s *Atelet) prepareOCIBundles(

g, gCtx := errgroup.WithContext(ctx)

out := &bundleSpecs{containers: make([]*ateworkerpb.ContainerSpec, len(spec.GetContainers()))}
if pauseImage != "" {
g.Go(func() error {
// pause is sandbox infra; it mounts no volumes.
if _, err := prepareOCIDirectory(gCtx, s.imageCache, actorUID, ocispec.PauseContainer, pauseImage, nil, nil); err != nil {
img, _, err := prepareOCIDirectory(gCtx, s.imageCache, actorUID, ocispec.PauseContainer, pauseImage, nil, nil)
if err != nil {
return wrapFileSystemErr("while creating pause OCI bundle", err)
}
if out.pauseUser, err = resolveImageUser(img); err != nil {
return fmt.Errorf("while resolving the pause container's user: %w", err)
}
return nil
})
}

// Application containers.
containerSpecs := make([]*ateworkerpb.ContainerSpec, len(spec.GetContainers()))
for i, ctr := range spec.GetContainers() {
var envs []string
for _, env := range ctr.GetEnv() {
envs = append(envs, fmt.Sprintf("%s=%s", env.GetName(), env.GetValue()))
}
g.Go(func() error {
imageConfig, err := prepareOCIDirectory(gCtx, s.imageCache, actorUID, ctr.GetName(), ctr.GetImage(), spec.GetVolumes(), ctr.GetVolumeMounts())
img, cwd, err := prepareOCIDirectory(gCtx, s.imageCache, actorUID, ctr.GetName(), ctr.GetImage(), spec.GetVolumes(), ctr.GetVolumeMounts())
if err != nil {
return wrapFileSystemErr(fmt.Sprintf("while creating %q OCI bundle", ctr.GetName()), err)
}
args, err := resolveProcessArgs(imageConfig, ctr.GetCommand(), ctr.GetArgs())
args, err := resolveProcessArgs(&img.Config, ctr.GetCommand(), ctr.GetArgs())
if err != nil {
return fmt.Errorf("while resolving process args for container %q: %w", ctr.GetName(), err)
}
containerSpecs[i] = &ateworkerpb.ContainerSpec{
user, err := resolveImageUser(img)
if err != nil {
return fmt.Errorf("while resolving user for container %q: %w", ctr.GetName(), err)
}
out.containers[i] = &ateworkerpb.ContainerSpec{
Args: args,
Env: resolveActorEnv(imageConfig, envs),
Env: resolveActorEnv(&img.Config, envs),
Capabilities: resolveCapabilities(ctr.GetSecurityContext().GetCapabilities()),
Resources: toWorkerResourceLimits(ctr.GetResources()),
User: user,
Cwd: cwd,
}
return nil
})
}
if err := g.Wait(); err != nil {
return nil, err
}
return containerSpecs, nil
return out, nil
}

// bundleSpecs is what prepareOCIBundles resolved from the images.
type bundleSpecs struct {
// containers are the application containers' specs, indexed like
// WorkloadSpec.containers.
containers []*ateworkerpb.ContainerSpec
// pauseUser is the pause image's USER. nil is root, and a sandbox without
// a pause container.
pauseUser *ateworkerpb.ProcessUser
}

// dialWorker opens (or reuses) the gRPC connection to the target worker
Expand All @@ -1879,10 +1899,9 @@ func (s *Atelet) dialWorker(ctx context.Context, workerPodUID string) (ateworker
}

// buildWorkerWorkloadSpec projects the atelet-facing workload spec onto
// the worker-facing one. containerSpecs holds each container's spec from
// prepareOCIBundles, indexed like spec.GetContainers(); it is nil for RPCs that
// start no process.
func buildWorkerWorkloadSpec(spec *ateletpb.WorkloadSpec, containerSpecs []*ateworkerpb.ContainerSpec) (*ateworkerpb.WorkloadSpec, error) {
// the worker-facing one. bundles holds what prepareOCIBundles resolved from
// the images; it is nil for RPCs that start no process.
func buildWorkerWorkloadSpec(spec *ateletpb.WorkloadSpec, bundles *bundleSpecs) (*ateworkerpb.WorkloadSpec, error) {
volumes := make(map[string]*ateletpb.Volume)
for _, vol := range spec.GetVolumes() {
name := vol.GetName()
Expand All @@ -1893,6 +1912,9 @@ func buildWorkerWorkloadSpec(spec *ateletpb.WorkloadSpec, containerSpecs []*atew
}

out := &ateworkerpb.WorkloadSpec{}
if bundles != nil {
out.PauseUser = bundles.pauseUser
}
for i, ctr := range spec.GetContainers() {
var ddMounts []*ateworkerpb.VolumeMount
var extMounts []*ateworkerpb.VolumeMount
Expand Down Expand Up @@ -1938,8 +1960,8 @@ func buildWorkerWorkloadSpec(spec *ateletpb.WorkloadSpec, containerSpecs []*atew
ImageVolumeMounts: imgMounts,
WakeupProbe: toWorkerWakeupProbe(ctr.GetWakeupProbe()),
}
if containerSpecs != nil {
container.ContainerSpec = containerSpecs[i]
if bundles != nil {
container.ContainerSpec = bundles.containers[i]
}
out.Containers = append(out.Containers, container)
}
Expand Down
68 changes: 66 additions & 2 deletions cmd/atelet/main_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import (
"time"

"github.com/google/go-cmp/cmp"
v1 "github.com/google/go-containerregistry/pkg/v1"
"github.com/spf13/pflag"
"go.ate.dev/substrate/cmd/atelet/internal/ateletpath"
"go.ate.dev/substrate/cmd/atelet/internal/snapshottracker"
Expand Down Expand Up @@ -386,12 +387,75 @@ func TestPrepareOCIBundlesFillsContainerSpec(t *testing.T) {
Env: []string{"GREETING=hello", "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"},
Capabilities: []string{"CAP_AUDIT_WRITE", "CAP_NET_ADMIN", "CAP_NET_BIND_SERVICE"},
Resources: &ateworkerpb.ResourceLimits{MemoryBytes: 1 << 30, CpuMillis: 500},
Cwd: "/",
}}
if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" {
if diff := cmp.Diff(want, got.containers, protocmp.Transform()); diff != "" {
t.Errorf("prepareOCIBundles container specs mismatch (-want +got):\n%s", diff)
}
}

// TestPrepareOCIBundlesResolvesUsers pins that a container runs as its image's
// USER, resolved against the image's own passwd and group files, and the pause
// container as the pause image's; a USER the image cannot resolve fails the
// run.
func TestPrepareOCIBundlesResolvesUsers(t *testing.T) {
useTempNodeDirs(t)
host := imageVolumeTestRegistry(t)
etc := []v1.Layer{
singleFileLayer(t, "etc/passwd", "root:x:0:0:root:/root:/bin/sh\nnonroot:x:65532:65532::/home/nonroot:/sbin/nologin\n"),
singleFileLayer(t, "etc/group", "root:x:0:\nnonroot:x:65532:\nvideo:x:44:nonroot\n"),
}
app := pushTestImageWithConfig(t, host+"/app:v1", v1.Config{User: "nonroot", Cmd: []string{"/bin/app"}}, etc...)
pause := pushTestImageWithConfig(t, host+"/pause:v1", v1.Config{User: "65535", Cmd: []string{"/pause"}})
s := &Atelet{imageCache: newImageVolumeStore(t)}

spec := &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "app", Image: app}}}
got, err := s.prepareOCIBundles(t.Context(), "actor-uid-1", resources.ActorRef{}, spec, pause)
if err != nil {
t.Fatalf("prepareOCIBundles: %v", err)
}
want := &ateworkerpb.ProcessUser{Uid: 65532, Gid: 65532, AdditionalGids: []uint32{44}}
if diff := cmp.Diff(want, got.containers[0].GetUser(), protocmp.Transform()); diff != "" {
t.Errorf("app user mismatch (-want +got):\n%s", diff)
}
// A bare uid without a passwd entry gets gid 0, as under Docker.
if diff := cmp.Diff(&ateworkerpb.ProcessUser{Uid: 65535}, got.pauseUser, protocmp.Transform()); diff != "" {
t.Errorf("pause user mismatch (-want +got):\n%s", diff)
}

unresolvable := pushTestImageWithConfig(t, host+"/nobody:v1", v1.Config{User: "nobody", Cmd: []string{"/bin/app"}})
spec = &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "app", Image: unresolvable}}}
if _, err := s.prepareOCIBundles(t.Context(), "actor-uid-2", resources.ActorRef{}, spec, ""); err == nil {
t.Error("prepareOCIBundles succeeded with a USER the image has no passwd entry for, want an error")
}
}

// TestPrepareOCIBundlesStartsInTheImageWorkdir pins that a container starts in
// its image's WORKDIR, created through the overlay when the image lacks it.
func TestPrepareOCIBundlesStartsInTheImageWorkdir(t *testing.T) {
useTempNodeDirs(t)
host := imageVolumeTestRegistry(t)
image := pushTestImageWithConfig(t, host+"/app:v1", v1.Config{Cmd: []string{"/bin/app"}, WorkingDir: "/home/nonroot"})
spec := &ateletpb.WorkloadSpec{Containers: []*ateletpb.Container{{Name: "app", Image: image}}}

const actorUID = "actor-uid-1"
s := &Atelet{imageCache: newImageVolumeStore(t)}
got, err := s.prepareOCIBundles(t.Context(), actorUID, resources.ActorRef{}, spec, "")
if err != nil {
t.Fatalf("prepareOCIBundles: %v", err)
}
if cwd := got.containers[0].GetCwd(); cwd != "/home/nonroot" {
t.Errorf("Cwd = %q, want /home/nonroot", cwd)
}
overlay, err := imagecache.ReadSpec(ateletpath.OCIBundlePath(actorUID, "app"))
if err != nil {
t.Fatalf("imagecache.ReadSpec: %v", err)
}
if !slices.Contains(overlay.ExtraDirs, "/home/nonroot") {
t.Errorf("ExtraDirs = %v, want the working directory", overlay.ExtraDirs)
}
}

func TestSnapshotManifestRejectsNonLocalFile(t *testing.T) {
manifest, err := json.Marshal(sandboxAssetsRecord{
SandboxClass: "gvisor",
Expand Down Expand Up @@ -866,7 +930,7 @@ func TestBuildWorkerWorkloadSpecForwardsContainerSpecs(t *testing.T) {
{Name: "sidecar", ContainerSpec: &ateworkerpb.ContainerSpec{Args: []string{"/sidecar"}}},
},
}
got, err := buildWorkerWorkloadSpec(in, containerSpecs)
got, err := buildWorkerWorkloadSpec(in, &bundleSpecs{containers: containerSpecs})
if err != nil {
t.Fatalf("buildWorkerWorkloadSpec failed: %v", err)
}
Expand Down
Loading
Loading