Skip to content

Enable CSI support for micro-vm #840

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
hajiler wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Enable CSI support for micro-vm #840

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
69 changes: 58 additions & 11 deletions cmd/atelet/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -466,6 +466,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 @@ -475,7 +480,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()),
CpuMilli: req.GetCpuMilli(),
Expand Down Expand Up @@ -580,6 +585,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 @@ -588,7 +598,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 @@ -601,7 +611,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
}

sandboxRec.SnapshotFiles = resp.GetSnapshotFiles()
if len(sandboxRec.SnapshotFiles) == 0 {
if len(sandboxRec.SnapshotFiles) == 0 && shouldHaveSnapshots(req) {
return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonInvalidCheckpointResult, ateerrors.ActorCrashedMetadata(), errors.New("ateom reported no snapshot files for checkpoint"))
}
sandboxRec.Atespace = req.GetAtespace()
Expand Down Expand Up @@ -693,6 +703,20 @@ func (s *AteomHerder) moveLocalCheckpoint(ctx context.Context, req *ateletpb.Che
return nil
}

// shouldHaveSnapshots returns true if the checkpoint request is expected to produce snapshot files.
func shouldHaveSnapshots(req *ateletpb.CheckpointRequest) bool {
if req.GetScope() != ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA {
return true
}

for _, vol := range req.GetSpec().GetVolumes() {
if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR {
return true
}
}
return false
}

func (s *AteomHerder) uploadExternalCheckpoint(ctx context.Context, req *ateletpb.CheckpointRequest, checkpointDir string, rec *sandboxAssetsRecord) error {
uri, err := resources.ParseSnapshotURI(req.GetExternalConfig().GetSnapshotUri())
if err != nil {
Expand Down Expand Up @@ -1105,6 +1129,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 @@ -1113,7 +1142,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 @@ -1495,32 +1524,50 @@ 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
name := vol.GetName()
if _, duplicate := volumes[name]; duplicate {
return nil, fmt.Errorf("duplicate volume name %q in workload spec", name)
}
volumes[name] = 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)
Comment thread
hajiler marked this conversation as resolved.
}

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
155 changes: 153 additions & 2 deletions cmd/atelet/main_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -693,7 +693,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 @@ -736,6 +739,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 @@ -746,12 +752,88 @@ 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`,
},
{
name: "duplicate volume names",
in: &ateletpb.WorkloadSpec{
Volumes: []*ateletpb.Volume{
{Name: "data", Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR},
{Name: "data", Type: ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL},
},
Containers: []*ateletpb.Container{
{
Name: "ctr",
VolumeMounts: []*ateletpb.VolumeMount{
{Name: "data", MountPath: "/data"},
},
},
},
},
wantErr: `duplicate volume name "data" in workload spec`,
},
}

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 Expand Up @@ -1580,3 +1662,72 @@ func TestValidateUploadPausedCheckpointRequest(t *testing.T) {
})
}
}

func TestShouldHaveSnapshots(t *testing.T) {
tests := []struct {
name string
req *ateletpb.CheckpointRequest
want bool
}{
{
name: "full scope always expects snapshots",
req: &ateletpb.CheckpointRequest{
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
},
want: true,
},
{
name: "data scope with durable volumes expects snapshots",
req: &ateletpb.CheckpointRequest{
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA,
Spec: &ateletpb.WorkloadSpec{
Volumes: []*ateletpb.Volume{
{Name: "durable", Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR},
},
},
},
want: true,
},
{
name: "data scope with only CSI volumes does not expect snapshots",
req: &ateletpb.CheckpointRequest{
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA,
Spec: &ateletpb.WorkloadSpec{
Volumes: []*ateletpb.Volume{
{Name: "csi", Type: ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL},
},
},
},
want: false,
},
{
name: "data scope with both durable and CSI volumes expects snapshots",
req: &ateletpb.CheckpointRequest{
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA,
Spec: &ateletpb.WorkloadSpec{
Volumes: []*ateletpb.Volume{
{Name: "durable", Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR},
{Name: "csi", Type: ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL},
},
},
},
want: true,
},
{
name: "data scope with no volumes does not expect snapshots",
req: &ateletpb.CheckpointRequest{
Scope: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA,
Spec: &ateletpb.WorkloadSpec{},
},
want: false,
},
}

for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
if got := shouldHaveSnapshots(tc.req); got != tc.want {
t.Errorf("shouldHaveSnapshots() = %v, want %v", got, tc.want)
}
})
}
}
18 changes: 8 additions & 10 deletions cmd/ateom-microvm/checkpoint.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"fmt"
"log/slog"
"os"
"os/exec"
"path/filepath"
"time"

Expand Down Expand Up @@ -90,13 +89,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 thread
hajiler marked this conversation as resolved.
"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 @@ -331,13 +332,10 @@ func (s *AteomService) teardownActor(ctx context.Context, id string, ra *running
_ = ra.chCmd.Process.Kill()
_, _ = ra.chCmd.Process.Wait()
}
// Kill the virtiofsds (after CH, their only client): the merged rootfs
// share's and, when the actor has durable-dir volumes, the durable share's.
for _, cmd := range []*exec.Cmd{ra.vfsdCmd, ra.durableVfsdCmd} {
if cmd != nil && cmd.Process != nil {
_ = cmd.Process.Kill()
_, _ = cmd.Process.Wait()
}
// Kill the virtiofsd (after CH, its only client).
if ra.vfsdCmd != nil && ra.vfsdCmd.Process != nil {
_ = ra.vfsdCmd.Process.Kill()
_, _ = ra.vfsdCmd.Process.Wait()
}
}

Expand Down
Loading
Loading