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
68 changes: 56 additions & 12 deletions cmd/atelet/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,11 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *
return nil, err
}

spec, err := buildAteomWorkloadSpec(req.GetSpec())
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "invalid workload spec: %v", err)
}

// Tell ateom to start the workload. gVisor uses RunscPath; the micro-VM
// runtime uses the full RuntimeAssetPaths set.
if _, err := client.RunWorkload(ctx, &ateompb.RunWorkloadRequest{
Expand All @@ -441,7 +446,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *
ActorTemplateName: req.GetActorTemplateName(),
RunscPath: runscPathFor(assetPaths),
RuntimeAssetPaths: assetPaths,
Spec: buildAteomWorkloadSpec(req.GetSpec()),
Spec: spec,
ActorUid: actorUID,
EgressGateway: toAteomEgressGateway(req.GetEgressGateway()),
}); err != nil {
Expand Down Expand Up @@ -544,6 +549,11 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
// Tell ateom to take the checkpoint and delete containers. ateom reports the
// exact files it wrote so we ship precisely that set (gVisor's image files,
// cloud-hypervisor's snapshot set, ...) rather than a hardcoded list.
spec, err := buildAteomWorkloadSpec(req.GetSpec())
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "invalid workload spec: %v", err)
}

tAteom := time.Now()
resp, err := client.CheckpointWorkload(ctx, &ateompb.CheckpointWorkloadRequest{
Atespace: actorRef.Atespace,
Expand All @@ -552,7 +562,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
ActorTemplateName: req.GetActorTemplateName(),
RunscPath: runscPathFor(assetPaths),
RuntimeAssetPaths: assetPaths,
Spec: buildAteomWorkloadSpec(req.GetSpec()),
Spec: spec,
Scope: toAteomSnapshotScope(req.GetScope()),
ActorUid: actorUID,
})
Expand All @@ -566,7 +576,22 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe

sandboxRec.SnapshotFiles = resp.GetSnapshotFiles()
if len(sandboxRec.SnapshotFiles) == 0 {
return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonInvalidCheckpointResult, ateerrors.ActorCrashedMetadata(), errors.New("ateom reported no snapshot files for checkpoint"))
hasDurable := false
hasCsi := false
for _, vol := range req.GetSpec().GetVolumes() {
switch vol.GetType() {
case ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR:
hasDurable = true
case ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL:
hasCsi = true
}
}
isDataScope := req.GetScope() == ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA
if isDataScope && hasCsi && !hasDurable {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, we do not snapshot csi volumes since they are external and persisted beyond pause/suspend. So we are basically quitely ignoring Data snapshot config when only CSI volumes are present.

Instead should we either:

  1. Disallow Data scope snapshot when only CSI volumes are present?
  2. Take snapshots of CSI volumes?
  3. Add some additional scope specific to CSI volumes?

IMO maybe we just add a note to the the Data snapshot scope that it is particular to durable dir, and external volumes will not be snapshotted.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that's correct for now. Dmitry Berkovich (@dberkov) FYI.

// OK: CSI volumes don't produce snapshot files in DATA scope.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 question 🟢 – Minor: the empty if body with the real path in else reads backwards. Inverting to if !(isDataScope && hasCsi && !hasDurable) { return ... } — or a small helper — would say it plainly.

} else {
return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonInvalidCheckpointResult, ateerrors.ActorCrashedMetadata(), errors.New("ateom reported no snapshot files for checkpoint"))
}
}
sandboxRec.Atespace = req.GetAtespace()
sandboxRec.ActorName = req.GetActorName()
Expand Down Expand Up @@ -1069,6 +1094,11 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)

// Tell ateom to do runsc create + runsc restore for pause container and
// all application containers.
spec, err := buildAteomWorkloadSpec(req.GetSpec())
if err != nil {
return nil, status.Errorf(codes.InvalidArgument, "invalid workload spec: %v", err)
}

tAteom := time.Now()
_, err = client.RestoreWorkload(ctx, &ateompb.RestoreWorkloadRequest{
Atespace: actorRef.Atespace,
Expand All @@ -1077,7 +1107,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
ActorTemplateName: req.GetActorTemplateName(),
RunscPath: runscPathFor(assetPaths),
RuntimeAssetPaths: assetPaths,
Spec: buildAteomWorkloadSpec(req.GetSpec()),
Spec: spec,
Scope: toAteomSnapshotScope(req.GetScope()),
ActorUid: req.GetActorUid(),
EgressGateway: toAteomEgressGateway(req.GetEgressGateway()),
Expand Down Expand Up @@ -1457,32 +1487,46 @@ func (s *AteomHerder) dialAteom(ctx context.Context, targetAteomUid string) (ate

// buildAteomWorkloadSpec projects the atelet-facing workload spec onto
// the ateom-facing one.
func buildAteomWorkloadSpec(spec *ateletpb.WorkloadSpec) *ateompb.WorkloadSpec {
ddVolumes := make(map[string]bool)
func buildAteomWorkloadSpec(spec *ateletpb.WorkloadSpec) (*ateompb.WorkloadSpec, error) {
volumes := make(map[string]ateletpb.VolumeType)
for _, vol := range spec.GetVolumes() {
if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR {
ddVolumes[vol.GetName()] = true
}
volumes[vol.GetName()] = vol.GetType()
}

out := &ateompb.WorkloadSpec{}
for _, ctr := range spec.GetContainers() {
var ddMounts []*ateompb.DurableDirVolumeMount
var csiMounts []*ateompb.VolumeMount
for _, vm := range ctr.GetVolumeMounts() {
if ddVolumes[vm.GetName()] {
volName := vm.GetName()
volType, ok := volumes[volName]
if !ok {
return nil, fmt.Errorf("container %q mounts volume %q which is not defined in workload volumes", ctr.GetName(), volName)

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 should-fix 🟡 – This answers your own question from the last round about validating mounts against volumes, but it only covers one half of it and only at runtime.

The uniqueness half is still open: volumes is a map keyed by name, so two entries in spec.volumes sharing a name silently collapse to the last one. If they differ in type, the mount is classified by whichever came last in the slice.

For the dangling-reference half, the error is right but late. The ActorTemplate has a CEL rule requiring every volume to be mounted by some container, but not the converse, so a template with a volumeMounts entry naming no volume passes admission and then fails at Run with InvalidArgument — a template that can never start an actor. The mirror-image CEL rule alongside the existing one would reject it at create time.

}

switch volType {
case ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR:
ddMounts = append(ddMounts, &ateompb.DurableDirVolumeMount{
VolumeName: vm.GetName(),
VolumeName: volName,
MountPath: vm.GetMountPath(),
})
case ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL:
csiMounts = append(csiMounts, &ateompb.VolumeMount{
VolumeName: volName,
MountPath: vm.GetMountPath(),
})
default:
return nil, fmt.Errorf("container %q mounts volume %q with unsupported type %v", ctr.GetName(), volName, volType)
}
}
out.Containers = append(out.Containers, &ateompb.Container{
Name: ctr.GetName(),
DurableDirVolumeMounts: ddMounts,
CsiVolumeMounts: csiMounts,
Readyz: toAteomReadyz(ctr.GetReadyz()),
})
}
return out
return out, nil
}

func toAteomEgressGateway(gateway *ateletpb.EgressGateway) *ateompb.EgressGateway {
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 @@ -679,7 +679,10 @@ func TestBuildAteomWorkloadSpecForwardsReadyz(t *testing.T) {
{Name: "without-probe"},
},
}
got := buildAteomWorkloadSpec(in)
got, err := buildAteomWorkloadSpec(in)
if err != nil {
t.Fatalf("buildAteomWorkloadSpec failed: %v", err)
}
if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" {
t.Errorf("buildAteomWorkloadSpec mismatch (-want +got):\n%s", diff)
}
Expand Down Expand Up @@ -722,6 +725,9 @@ func TestBuildAteomWorkloadSpecForwardsDurableDirMounts(t *testing.T) {
{VolumeName: "data", MountPath: "/home/counter"},
{VolumeName: "cache", MountPath: "/var/cache"},
},
CsiVolumeMounts: []*ateompb.VolumeMount{
{VolumeName: "scratch", MountPath: "/scratch"},
},
},
{
Name: "sidecar",
Expand All @@ -732,12 +738,70 @@ func TestBuildAteomWorkloadSpecForwardsDurableDirMounts(t *testing.T) {
{Name: "no-volumes"},
},
}
got := buildAteomWorkloadSpec(in)
got, err := buildAteomWorkloadSpec(in)
if err != nil {
t.Fatalf("buildAteomWorkloadSpec failed: %v", err)
}
if diff := cmp.Diff(want, got, protocmp.Transform()); diff != "" {
t.Errorf("buildAteomWorkloadSpec mismatch (-want +got):\n%s", diff)
}
}

func TestBuildAteomWorkloadSpecValidation(t *testing.T) {
tests := []struct {
name string
in *ateletpb.WorkloadSpec
wantErr string
}{
{
name: "missing volume definition",
in: &ateletpb.WorkloadSpec{
Volumes: []*ateletpb.Volume{
{Name: "data", Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR},
},
Containers: []*ateletpb.Container{
{
Name: "ctr",
VolumeMounts: []*ateletpb.VolumeMount{
{Name: "missing-vol", MountPath: "/data"},
},
},
},
},
wantErr: `container "ctr" mounts volume "missing-vol" which is not defined in workload volumes`,
},
{
name: "unsupported volume type",
in: &ateletpb.WorkloadSpec{
Volumes: []*ateletpb.Volume{
{Name: "data", Type: ateletpb.VolumeType_VOLUME_TYPE_UNSPECIFIED},
},
Containers: []*ateletpb.Container{
{
Name: "ctr",
VolumeMounts: []*ateletpb.VolumeMount{
{Name: "data", MountPath: "/data"},
},
},
},
},
wantErr: `container "ctr" mounts volume "data" with unsupported type VOLUME_TYPE_UNSPECIFIED`,
},
}

for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
_, err := buildAteomWorkloadSpec(tc.in)
if err == nil {
t.Fatal("expected error, got nil")
}
if got, want := err.Error(), tc.wantErr; !strings.Contains(got, want) {
t.Errorf("error mismatch:\nwant: %s\ngot: %s", want, got)
}
})
}
}

func TestToAteomEgressGateway(t *testing.T) {
if got := toAteomEgressGateway(nil); got != nil {
t.Fatalf("toAteomEgressGateway(nil) = %v, want nil", got)
Expand Down
8 changes: 5 additions & 3 deletions cmd/ateom-microvm/checkpoint.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,13 +81,15 @@ func (s *AteomService) CheckpointWorkload(ctx context.Context, req *ateompb.Chec
// captures. DATA_ON_GOLDEN is restore-only (a DataOnGolden commit arrives
// here as plain DATA) and lands in the default rejection.
durable := hasDurableVolumes(req.GetSpec().GetContainers())
csi := hasCsiVolumes(req.GetSpec().GetContainers())
scope := req.GetScope()
switch scope {
case ateompb.SnapshotScope_SNAPSHOT_SCOPE_FULL:
case ateompb.SnapshotScope_SNAPSHOT_SCOPE_DATA:
if !durable {
// TODO: Revisit handling for CSI volumes since snapshots are currently quietly ignored.
if !durable && !csi {
return nil, status.Error(codes.FailedPrecondition,
Comment on lines +90 to 91

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to the commend above, should we continue to quietly ignore snapshots for CSI volumes?

"no durable-dir volumes found for a Data-scope snapshot")
"no durable-dir or CSI volumes found for a Data-scope snapshot")
}
default:
return nil, status.Errorf(codes.InvalidArgument, "unsupported snapshot scope: %v", scope)
Expand Down Expand Up @@ -284,7 +286,7 @@ func (s *AteomService) teardownActor(ctx context.Context, id string, ra *running
}
// Kill the virtiofsds (after CH, their only client): the overlay RO lower's
// and, when the actor has durable-dir volumes, the writable share's.
for _, cmd := range []*exec.Cmd{ra.vfsdCmd, ra.durableVfsdCmd} {
for _, cmd := range []*exec.Cmd{ra.vfsdCmd, ra.durableVfsdCmd, ra.csiVfsdCmd} {
if cmd != nil && cmd.Process != nil {
_ = cmd.Process.Kill()
_, _ = cmd.Process.Wait()
Expand Down
84 changes: 84 additions & 0 deletions cmd/ateom-microvm/csi.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
//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"
"fmt"
"os"
"os/exec"
"path/filepath"

"github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/kata"
"github.com/agent-substrate/substrate/internal/ateompath"
"github.com/agent-substrate/substrate/internal/proto/ateompb"
specs "github.com/opencontainers/runtime-spec/specs-go"
)

// hasCsiVolumes reports whether any container mounts a CSI volume.
func hasCsiVolumes(containers []*ateompb.Container) bool {
Comment thread
hajiler marked this conversation as resolved.
for _, c := range containers {
if len(c.GetCsiVolumeMounts()) > 0 {
return true
}
}
return false
}

// csiMounts returns the OCI mounts that expose a container's CSI
// volumes at the paths it declared. Each source is that volume's directory
// inside the guest's CSI share, which the agent mounts at sandbox creation.
func csiMounts(mounts []*ateompb.VolumeMount) []specs.Mount {
out := make([]specs.Mount, 0, len(mounts))
for _, m := range mounts {
out = append(out, specs.Mount{
Destination: m.GetMountPath(),
Source: kata.GuestCsiVolumeDir(m.GetVolumeName()),
Type: "bind",
Options: []string{"rbind", "rw"},
})
}
return out
Comment thread
hajiler marked this conversation as resolved.
}

func csiVirtiofsdLogPath(id string) string {
return filepath.Join(kata.VMDir(id), "virtiofsd-csi.log")
}
Comment thread
hajiler marked this conversation as resolved.

// stageCsiShare starts the virtiofsd serving the actor's CSI volumes.
func (s *AteomService) stageCsiShare(ctx context.Context, rr resolvedRuntime, actorUID string) (*exec.Cmd, error) {
shared := ateompath.VolumesDir(actorUID)
if _, err := os.Stat(shared); err != nil {
return nil, fmt.Errorf("while checking CSI volumes dir %q: %w", shared, err)
}
log, err := os.OpenFile(csiVirtiofsdLogPath(actorUID), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
if err != nil {
return nil, fmt.Errorf("while opening CSI virtiofsd log file: %w", err)
}
defer log.Close()
cmd, err := kata.StartVirtiofsd(ctx, kata.VirtiofsdOptions{
Binary: rr.virtiofsd,
SocketPath: kata.CsiVirtiofsdSocketPath(actorUID),
SharedDir: shared,
Cache: "auto",
Log: log,
})
if err != nil {
return nil, fmt.Errorf("while starting CSI virtiofsd: %w", err)
}
return cmd, nil
}
Loading
Loading