atelet handling of external volumes.

Volume operations are integrated into the actor workflows:

- Run/Restore: MountVolume
- Checkpoint: UnmountVolume

Volumes are mounted under the actor directory and passed as
bind mounts to OCI.

Author: msau42

cmd/atelet/main.go

@@ -212,7 +212,7 @@ func NewService(
 	return wms
 }
 
-func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*ateletpb.RunResponse, error) {
+func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *ateletpb.RunResponse, err error) {
 	if err := validateRunRequest(req); err != nil {
 		// status.Error so the interceptor surfaces InvalidArgument and the
 		// message instead of masking both as Internal.
@@ -234,6 +234,16 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (*atele
 		return nil, fmt.Errorf("while resetting actor dirs: %w", err)
 	}
 
+	if err := s.mountExternalVolumes(ctx, atespace, actorID, req.GetSpec().GetVolumes()); err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err != nil {
+			// TODO cleanup orphaned volumes
+			_ = s.unmountExternalVolumes(ctx, atespace, actorID, req.GetSpec().GetVolumes())
+		}
+	}()
+
 	// Record the sandbox binaries this actor is running so a later Checkpoint
 	// (whose request no longer carries the sandbox config) can re-fetch the same
 	// version and pin it into the snapshot manifest.
@@ -350,6 +360,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
 		// in the metadata if the error is not retriable.
 		return nil, fmt.Errorf("while calling ateom.CheckpointWorkload: %w", err)
 	}
+
 	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"))
@@ -369,6 +380,9 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
 		return nil, fmt.Errorf("unexpected checkpoint type: %v", req.GetType())
 	}
 
+	// TODO cleanup orphaned volumes
+	_ = s.unmountExternalVolumes(ctx, atespace, actorID, req.GetSpec().GetVolumes())
+
 	// Note: we do not crash the actor if resetting the directory fails.
 	if err := resetActorDirs(atespace, actorID); err != nil {
 		return nil, fmt.Errorf("while resetting actor dirs: %w", err)
@@ -449,7 +463,7 @@ func (s *AteomHerder) uploadExternalCheckpoint(ctx context.Context, req *ateletp
 	return nil
 }
 
-func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (*ateletpb.RestoreResponse, error) {
+func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest) (resp *ateletpb.RestoreResponse, err error) {
 	if err := validateRestoreRequest(req); err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -462,6 +476,16 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
 		return nil, fmt.Errorf("while resetting actor dirs: %w", err)
 	}
 
+	if err := s.mountExternalVolumes(ctx, atespace, actorID, req.GetSpec().GetVolumes()); err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err != nil {
+			// TODO cleanup orphaned mounts
+			_ = s.unmountExternalVolumes(ctx, atespace, actorID, req.GetSpec().GetVolumes())
+		}
+	}()
+
 	checkpointDir := ateompath.RestoreStateDir(atespace, actorID)
 
 	// Per-step timing so we can attribute resume latency between the rustfs
@@ -657,12 +681,9 @@ func (s *AteomHerder) prepareOCIBundles(
 	if err := writeFileAtomic(filepath.Join(identityDir, ActorIDFileName), []byte(actorID), 0o644); err != nil {
 		return fmt.Errorf("while writing actor identity file: %w", err)
 	}
-
-	ddVolumes := make(map[string]bool)
 	// make directories for all durable-dir volumes
 	for _, vol := range spec.GetVolumes() {
 		if vol.GetType() == ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR {
-			ddVolumes[vol.GetName()] = true
 			volPath := ateompath.DurableDirVolumeMountPoint(atespace, actorID, vol.GetName())
 			if err := os.MkdirAll(volPath, 0o700); err != nil {
 				return fmt.Errorf("while creating %q: %w", volPath, err)
@@ -700,6 +721,7 @@ func (s *AteomHerder) prepareOCIBundles(
 			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)
 		}
@@ -713,12 +735,6 @@ func (s *AteomHerder) prepareOCIBundles(
 		for _, env := range ctr.GetEnv() {
 			envs = append(envs, fmt.Sprintf("%s=%s", env.GetName(), env.GetValue()))
 		}
-		var ddMounts []*ateletpb.VolumeMount
-		for _, vm := range ctr.GetVolumeMounts() {
-			if ddVolumes[vm.GetName()] {
-				ddMounts = append(ddMounts, vm)
-			}
-		}
 		g.Go(func() error {
 			if err := prepareOCIDirectory(
 				gCtx,
@@ -735,7 +751,8 @@ func (s *AteomHerder) prepareOCIBundles(
 				},
 				ateompath.AteomNetNSPath(targetAteomUid),
 				identityDir,
-				ddMounts,
+				spec.GetVolumes(),
+				ctr.GetVolumeMounts(),
 			); err != nil {
 				return wrapFileSystemErr(fmt.Sprintf("while creating %q OCI bundle", ctr.GetName()), err)
 			}
@@ -1058,5 +1075,22 @@ func resetActorDirs(atespace, actorID string) error {
 		return wrapFileSystemErr("while creating durable-dir volumes mount dir: %w", err)
 	}
 
+	// Do not call RemoveAll on volume directories in case the unmount failed.
+	// We do not want to delete mount content.
+	volumesDir := ateompath.VolumesDir(atespace, actorID)
+	entries, err := os.ReadDir(volumesDir)
+	if err != nil && !os.IsNotExist(err) {
+		return wrapFileSystemErr("while reading volumes dir: %w", err)
+	}
+	for _, entry := range entries {
+		volPath := filepath.Join(volumesDir, entry.Name())
+		if err := os.Remove(volPath); err != nil {
+			return wrapFileSystemErr("while removing volume dir: %w", err)
+		}
+	}
+	if err := os.MkdirAll(volumesDir, 0o755); err != nil {
+		return wrapFileSystemErr("while creating volumes dir: %w", err)
+	}
+
 	return nil
 }

cmd/atelet/oci.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/agent-substrate/substrate/cmd/atelet/internal/memorypullcache"
 	"github.com/agent-substrate/substrate/internal/ateompath"
+
 	"github.com/opencontainers/runtime-spec/specs-go"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -53,7 +54,7 @@ const (
 	ActorIDFileName = "actor-id"
 )
 
-func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryPullCache, atespace, actorID, containerName, ref string, args []string, env []string, annotations map[string]string, netns string, identityDir string, durableDirVolumeMounts []*ateletpb.VolumeMount) error {
+func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryPullCache, atespace, actorID, containerName, ref string, args []string, env []string, annotations map[string]string, netns string, identityDir string, volumes []*ateletpb.Volume, volumeMounts []*ateletpb.VolumeMount) error {
 	tracer := otel.Tracer("prepareOCIDirectory")
 
 	ctx, span := tracer.Start(ctx, "prepareOCIDirectory")
@@ -90,7 +91,14 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP
 		}
 	}
 
-	ociSpec := buildActorOCISpec(atespace, actorID, args, env, annotations, netns, identityDir, durableDirVolumeMounts)
+	// Create directories in rootfs to mount all volumes.
+	for _, vm := range volumeMounts {
+		if err := createMountPoint(rootPath, vm.GetMountPath()); err != nil {
+			return fmt.Errorf("while creating volume mount point %q: %w", vm.GetMountPath(), err)
+		}
+	}
+
+	ociSpec := buildActorOCISpec(atespace, actorID, args, env, annotations, netns, identityDir, volumes, volumeMounts)
 	ociSpecBytes, err := json.MarshalIndent(ociSpec, "", "  ")
 	if err != nil {
 		return fmt.Errorf("while marshaling OCI spec: %w", err)
@@ -107,7 +115,7 @@ func prepareOCIDirectory(ctx context.Context, pullCache *memorypullcache.MemoryP
 // When identityDir is non-empty it adds a read-only bind mount of that host
 // directory at IdentityMountPath so the actor can read its own ID (see
 // IdentityMountPath for why this is a bind mount rather than env vars).
-func buildActorOCISpec(atespace string, actorID string, args []string, env []string, annotations map[string]string, netns string, identityDir string, durableDirVolumeMounts []*ateletpb.VolumeMount) *specs.Spec {
+func buildActorOCISpec(atespace string, actorID string, args []string, env []string, annotations map[string]string, netns string, identityDir string, volumes []*ateletpb.Volume, volumeMounts []*ateletpb.VolumeMount) *specs.Spec {
 	envVars := []string{
 		"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
 	}
@@ -220,12 +228,27 @@ func buildActorOCISpec(atespace string, actorID string, args []string, env []str
 		Annotations: annotations,
 	}
 
-	// Prepare and mount durable-dir volumes.
-	for _, vm := range durableDirVolumeMounts {
+	// Prepare and mount all volumes.
+	volumeTypes := make(map[string]ateletpb.VolumeType)
+	for _, vol := range volumes {
+		volumeTypes[vol.GetName()] = vol.GetType()
+	}
+
+	for _, vm := range volumeMounts {
+		var srcPath string
+		switch volumeTypes[vm.GetName()] {
+		case ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR:
+			srcPath = ateompath.DurableDirVolumeMountPoint(atespace, actorID, vm.GetName())
+		case ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL:
+			srcPath = ateompath.VolumeHostPath(atespace, actorID, vm.GetName())
+		default:
+			continue
+		}
 		spec.Mounts = append(spec.Mounts, specs.Mount{
 			Destination: vm.GetMountPath(),
 			Type:        "bind",
-			Source:      ateompath.DurableDirVolumeMountPoint(atespace, actorID, vm.GetName()),
+			Source:      srcPath,
+			Options:     []string{"bind", "rw"},
 		})
 	}
 

cmd/atelet/oci_test.go

@@ -94,6 +94,7 @@ func TestBuildActorOCISpec_IdentityMount(t *testing.T) {
 		"/run/netns/x",
 		"/host/actors/atespace:id/identity",
 		nil,
+		nil,
 	)
 	found := false
 	for _, m := range spec.Mounts {
@@ -118,7 +119,7 @@ func TestBuildActorOCISpec_IdentityMount(t *testing.T) {
 
 // Without an identity dir (the pause container), no identity mount appears.
 func TestBuildActorOCISpec_NoIdentityMountForPause(t *testing.T) {
-	bare := buildActorOCISpec("atespace", "id", []string{"/pause"}, nil, nil, "/run/netns/x", "", nil)
+	bare := buildActorOCISpec("atespace", "id", []string{"/pause"}, nil, nil, "/run/netns/x", "", nil, nil)
 	for _, m := range bare.Mounts {
 		if m.Destination == IdentityMountPath {
 			t.Errorf("identity mount must be absent when identityDir is empty")
@@ -134,11 +135,16 @@ func TestBuildActorOCISpec_DurableDirVolumeMounts(t *testing.T) {
 		{Name: "data", MountPath: "/var/data"},
 		{Name: "cache", MountPath: "/var/cache"},
 	}
+	volumes := []*ateletpb.Volume{
+		{Name: "data", Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR},
+		{Name: "cache", Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR},
+	}
 	spec := buildActorOCISpec(
 		atespace, id,
 		[]string{"/app"}, nil, nil,
 		"/run/netns/x",
 		"",
+		volumes,
 		durableDirs,
 	)
 

cmd/atelet/volumes.go

@@ -0,0 +1,74 @@
+// 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"
+	"log/slog"
+	"os"
+
+	"github.com/agent-substrate/substrate/internal/ateompath"
+	"github.com/agent-substrate/substrate/internal/proto/ateletpb"
+	"github.com/agent-substrate/substrate/internal/volume"
+)
+
+var (
+	globalVolumePlugin = volume.NewMockVolumePlugin()
+)
+
+// TODO: Replace with actual volume plugin search
+func getVolumePlugin() volume.VolumePlugin {
+	return globalVolumePlugin
+}
+
+func (s *AteomHerder) mountExternalVolumes(ctx context.Context, atespace, actorID string, volumes []*ateletpb.Volume) error {
+	for _, vol := range volumes {
+		if vol.GetType() != ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL {
+			continue
+		}
+		ext := vol.GetExternal()
+		if ext == nil {
+			continue
+		}
+		hostPath := ateompath.VolumeHostPath(atespace, actorID, vol.GetName())
+		if err := os.MkdirAll(hostPath, 0o750); err != nil {
+			return fmt.Errorf("failed to create mount point %q: %w", hostPath, err)
+		}
+		slog.InfoContext(ctx, "Mounting volume", slog.String("volume_id", ext.GetStorageVolumeId()), slog.String("host_path", hostPath))
+		if err := getVolumePlugin().MountVolume(ctx, ext.GetStorageVolumeId(), hostPath); err != nil {
+			return fmt.Errorf("failed to mount volume %q to %q: %w", ext.GetStorageVolumeId(), hostPath, err)
+		}
+	}
+	return nil
+}
+
+func (s *AteomHerder) unmountExternalVolumes(ctx context.Context, atespace, actorID string, volumes []*ateletpb.Volume) error {
+	for _, vol := range volumes {
+		if vol.GetType() != ateletpb.VolumeType_VOLUME_TYPE_EXTERNAL {
+			continue
+		}
+		ext := vol.GetExternal()
+		if ext == nil {
+			continue
+		}
+		hostPath := ateompath.VolumeHostPath(atespace, actorID, vol.GetName())
+		slog.InfoContext(ctx, "Unmounting volume", slog.String("volume_id", ext.GetStorageVolumeId()), slog.String("host_path", hostPath))
+		if err := getVolumePlugin().UnmountVolume(ctx, ext.GetStorageVolumeId(), hostPath); err != nil {
+			slog.ErrorContext(ctx, "failed to unmount volume", slog.String("volume_id", ext.GetStorageVolumeId()), slog.String("host_path", hostPath), slog.Any("error", err))
+		}
+	}
+	return nil
+}

internal/ateompath/ateompath.go

@@ -184,3 +184,17 @@ func PIDFilePath(atespace, actorID, containerName string) string {
 		containerName+".pid",
 	)
 }
+
+func VolumesDir(atespace, actorID string) string {
+	return filepath.Join(
+		ActorPath(atespace, actorID),
+		"volumes",
+	)
+}
+
+func VolumeHostPath(atespace, actorID, volumeName string) string {
+	return filepath.Join(
+		VolumesDir(atespace, actorID),
+		volumeName,
+	)
+}