-
Notifications
You must be signed in to change notification settings - Fork 228
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.
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
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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{ | ||
|
|
@@ -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 { | ||
|
|
@@ -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, | ||
|
|
@@ -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, | ||
| }) | ||
|
|
@@ -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 { | ||
| // OK: CSI volumes don't produce snapshot files in DATA scope. | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🤖 question 🟢 – Minor: the empty |
||
| } 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() | ||
|
|
@@ -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, | ||
|
|
@@ -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()), | ||
|
|
@@ -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) | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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: 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 |
||
| } | ||
|
|
||
| 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 { | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe 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) | ||
|
|
@@ -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() | ||
|
|
||
| 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 { | ||
|
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 | ||
|
hajiler marked this conversation as resolved.
|
||
| } | ||
|
|
||
| func csiVirtiofsdLogPath(id string) string { | ||
| return filepath.Join(kata.VMDir(id), "virtiofsd-csi.log") | ||
| } | ||
|
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 | ||
| } | ||
There was a problem hiding this comment.
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:
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.
There was a problem hiding this comment.
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.