csi: Use default csi node hostpath

cmd/atecontroller/internal/controllers/workerpool_apply.go

@@ -111,6 +111,10 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin
 				WithName("run-ateom").
 				WithMountPath(ateompath.BasePath).
 				WithMountPropagation(corev1.MountPropagationHostToContainer),
+			corev1ac.VolumeMount().
+				WithName("kubelet-pods").
+				WithMountPath(ateompath.KubeletPodsDir).
+				WithMountPropagation(corev1.MountPropagationHostToContainer),
 			corev1ac.VolumeMount().
 				WithName(atunnelIdentityVolume).
 				WithMountPath(atunnelIdentityMountPath).
@@ -131,6 +135,11 @@ func buildDeploymentApplyConfig(wp *atev1alpha1.WorkerPool, otel ateomOTelSettin
 				WithHostPath(corev1ac.HostPathVolumeSource().
 					WithPath(ateompath.BasePath).
 					WithType(corev1.HostPathDirectoryOrCreate)),
+			corev1ac.Volume().
+				WithName("kubelet-pods").
+				WithHostPath(corev1ac.HostPathVolumeSource().
+					WithPath(ateompath.KubeletPodsDir).
+					WithType(corev1.HostPathDirectoryOrCreate)),
 			corev1ac.Volume().
 				WithName(atunnelIdentityVolume).
 				WithProjected(corev1ac.ProjectedVolumeSource().

cmd/atecontroller/internal/controllers/workerpool_apply_test.go

@@ -735,6 +735,11 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 				WithHostPath(corev1ac.HostPathVolumeSource().
 					WithPath(ateompath.BasePath).
 					WithType(corev1.HostPathDirectoryOrCreate)),
+			corev1ac.Volume().
+				WithName("kubelet-pods").
+				WithHostPath(corev1ac.HostPathVolumeSource().
+					WithPath(ateompath.KubeletPodsDir).
+					WithType(corev1.HostPathDirectoryOrCreate)),
 			corev1ac.Volume().
 				WithName(atunnelIdentityVolume).
 				WithProjected(corev1ac.ProjectedVolumeSource().
@@ -806,6 +811,10 @@ func expectedDeploymentApplyConfig(mutatePodSpec func(*corev1ac.PodSpecApplyConf
 					WithName("run-ateom").
 					WithMountPath(ateompath.BasePath).
 					WithMountPropagation(corev1.MountPropagationHostToContainer),
+				corev1ac.VolumeMount().
+					WithName("kubelet-pods").
+					WithMountPath(ateompath.KubeletPodsDir).
+					WithMountPropagation(corev1.MountPropagationHostToContainer),
 				corev1ac.VolumeMount().
 					WithName(atunnelIdentityVolume).
 					WithMountPath(atunnelIdentityMountPath).

cmd/atecontroller/internal/controllers/workerpool_controller_test.go

@@ -137,10 +137,11 @@ func TestWorkerPoolCreatesDeployment(t *testing.T) {
 		if len(dep.OwnerReferences) == 0 || dep.OwnerReferences[0].Name != wp.Name {
 			return false, nil
 		}
-		return len(dep.Spec.Template.Spec.Volumes) == 3 &&
+		return len(dep.Spec.Template.Spec.Volumes) == 4 &&
 			dep.Spec.Template.Spec.Volumes[0].Name == "run-ateom" &&
-			dep.Spec.Template.Spec.Volumes[1].Name == atunnelIdentityVolume &&
-			dep.Spec.Template.Spec.Volumes[2].Name == atunnelEgressTrustVolume, nil
+			dep.Spec.Template.Spec.Volumes[1].Name == "kubelet-pods" &&
+			dep.Spec.Template.Spec.Volumes[2].Name == atunnelIdentityVolume &&
+			dep.Spec.Template.Spec.Volumes[3].Name == atunnelEgressTrustVolume, nil
 	})
 }
 

cmd/atelet/main.go

@@ -461,7 +461,7 @@ func (s *AteomHerder) Run(ctx context.Context, req *ateletpb.RunRequest) (resp *
 		return nil, fmt.Errorf("while resetting actor dirs: %w", err)
 	}
 
-	if err := s.mountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil {
+	if err := s.mountExternalVolumes(ctx, req.GetTargetAteomUid(), actorUID, req.GetSpec().GetVolumes()); err != nil {
 		return nil, err
 	}
 
@@ -667,7 +667,7 @@ func (s *AteomHerder) Checkpoint(ctx context.Context, req *ateletpb.CheckpointRe
 	}
 	dPersist = time.Since(tPersist)
 
-	if err := s.unmountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil {
+	if err := s.unmountExternalVolumes(ctx, req.GetTargetAteomUid(), actorUID, req.GetSpec().GetVolumes()); err != nil {
 		return nil, ateerrors.NewGRPCError(ctx, codes.DataLoss, ateerrors.ReasonTerminalFileSystemError, ateerrors.ActorCrashedMetadata(), fmt.Errorf("while unmounting external volumes: %w", err))
 	}
 
@@ -951,7 +951,7 @@ func (s *AteomHerder) Restore(ctx context.Context, req *ateletpb.RestoreRequest)
 	}
 
 	tMount := time.Now()
-	mountErr := s.mountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes())
+	mountErr := s.mountExternalVolumes(ctx, req.GetTargetAteomUid(), actorUID, req.GetSpec().GetVolumes())
 	dMount = time.Since(tMount)
 	if mountErr != nil {
 		op.failedPhase = ateattr.SnapshotPhaseVolumeMount
@@ -1242,7 +1242,7 @@ func (s *AteomHerder) Terminate(ctx context.Context, req *ateletpb.TerminateRequ
 	}
 
 	// Unmount external volumes
-	if err := s.unmountExternalVolumes(ctx, actorUID, req.GetSpec().GetVolumes()); err != nil {
+	if err := s.unmountExternalVolumes(ctx, req.GetTargetAteomUid(), actorUID, req.GetSpec().GetVolumes()); err != nil {
 		return nil, fmt.Errorf("failed to unmount external volumes during terminate (actor: %s, actorUID: %s): %w", actorRef, actorUID, err)
 	}
 
@@ -1532,6 +1532,7 @@ func (s *AteomHerder) prepareOCIBundles(
 		if err := prepareOCIDirectory(
 			gCtx,
 			s.imageCache,
+			targetAteomUid,
 			actorUID,
 			"pause",
 			pauseImage,
@@ -1560,6 +1561,7 @@ func (s *AteomHerder) prepareOCIBundles(
 			if err := prepareOCIDirectory(
 				gCtx,
 				s.imageCache,
+				targetAteomUid,
 				actorUID,
 				ctr.GetName(),
 				ctr.GetImage(),
@@ -2097,23 +2099,6 @@ func resetActorDirs(actorUID string) error {
 		return wrapFileSystemErr("while creating system-info volume roots 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(actorUID)
-	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

@@ -77,7 +77,7 @@ func resolveCapabilities(caps *ateletpb.Capabilities) []string {
 	return out
 }
 
-func prepareOCIDirectory(ctx context.Context, imageCache *imagecache.Store, actorUID, containerName, ref string, command, args []string, env []string, annotations map[string]string, netns string, volumes []*ateletpb.Volume, volumeMounts []*ateletpb.VolumeMount, capabilities []string) error {
+func prepareOCIDirectory(ctx context.Context, imageCache *imagecache.Store, workerPodUID, actorUID, containerName, ref string, command, args []string, env []string, annotations map[string]string, netns string, volumes []*ateletpb.Volume, volumeMounts []*ateletpb.VolumeMount, capabilities []string) error {
 	tracer := otel.Tracer("prepareOCIDirectory")
 
 	ctx, span := tracer.Start(ctx, "prepareOCIDirectory")
@@ -149,7 +149,7 @@ func prepareOCIDirectory(ctx context.Context, imageCache *imagecache.Store, acto
 		return fmt.Errorf("while writing overlay spec: %w", err)
 	}
 
-	ociSpec := buildActorOCISpec(actorUID, containerName, resolvedArgs, resolvedEnv, annotations, netns, volumes, volumeMounts, capabilities)
+	ociSpec := buildActorOCISpec(workerPodUID, actorUID, containerName, resolvedArgs, resolvedEnv, annotations, netns, volumes, volumeMounts, capabilities)
 	ociSpecBytes, err := json.MarshalIndent(ociSpec, "", "  ")
 	if err != nil {
 		return fmt.Errorf("while marshaling OCI spec: %w", err)
@@ -265,7 +265,7 @@ func resolveProcessArgs(imageCfg *v1.Config, command, args []string) ([]string,
 // already-resolved args, env and capabilities (see resolveProcessArgs,
 // resolveActorEnv and resolveCapabilities). An empty capabilities set means the
 // process runs with none, which is what the pause container gets.
-func buildActorOCISpec(actorUID, containerName string, args []string, env []string, annotations map[string]string, netns string, volumes []*ateletpb.Volume, volumeMounts []*ateletpb.VolumeMount, capabilities []string) *specs.Spec {
+func buildActorOCISpec(workerPodUID, actorUID, containerName string, args []string, env []string, annotations map[string]string, netns string, volumes []*ateletpb.Volume, volumeMounts []*ateletpb.VolumeMount, capabilities []string) *specs.Spec {
 	mounts := []specs.Mount{
 		{
 			Destination: "/proc",
@@ -367,7 +367,7 @@ func buildActorOCISpec(actorUID, containerName string, args []string, env []stri
 		case *ateletpb.Volume_DurableDir:
 			srcPath = ateompath.DurableDirVolumeMountPoint(actorUID, vm.GetName())
 		case *ateletpb.Volume_External:
-			srcPath = ateompath.VolumeHostPath(actorUID, vm.GetName())
+			srcPath = ateompath.VolumeHostPath(workerPodUID, actorUID, vm.GetName())
 		case *ateletpb.Volume_SystemInfo:
 			// System-info contents are generated by atelet; the workload only
 			// reads them.

cmd/atelet/oci_test.go

@@ -41,7 +41,9 @@ func TestBuildActorOCISpec_SystemInfoVolumeMounts(t *testing.T) {
 		{Name: "sysinfo", Source: &ateletpb.Volume_SystemInfo{SystemInfo: &ateletpb.SystemInfoVolume{}}},
 	}
 	spec := buildActorOCISpec(
-		actorUID, "app",
+		"worker_pod_uid",
+		actorUID,
+		"app",
 		[]string{"/app"},
 		[]string{"FOO=bar"},
 		map[string]string{"k": "v"},
@@ -208,6 +210,7 @@ func TestResolveProcessArgs(t *testing.T) {
 // Each durable-dir volume mount becomes a bind mount whose source is the
 // per-actor on-host DurableDirVolumeMountPoint for that volume name.
 func TestBuildActorOCISpec_DurableDirVolumeMounts(t *testing.T) {
+	const workerPodUID = "worker_pod_uid"
 	const actorUID = "actor_uid"
 	durableDirs := []*ateletpb.VolumeMount{
 		{Name: "data", MountPath: "/var/data"},
@@ -218,7 +221,9 @@ func TestBuildActorOCISpec_DurableDirVolumeMounts(t *testing.T) {
 		{Name: "cache", Source: &ateletpb.Volume_DurableDir{DurableDir: &ateletpb.DurableDirVolume{}}},
 	}
 	spec := buildActorOCISpec(
-		actorUID, "app",
+		workerPodUID,
+		actorUID,
+		"app",
 		[]string{"/app"}, nil, nil,
 		"/run/netns/x",
 		volumes,
@@ -258,6 +263,7 @@ func TestBuildActorOCISpec_ImageVolumeMounts(t *testing.T) {
 		{Name: "data", MountPath: "/var/data"},
 	}
 	spec := buildActorOCISpec(
+		"worker_pod_uid",
 		"actor_uid", "app",
 		[]string{"/ate/payload-binary"}, nil, nil,
 		"/run/netns/x",
@@ -367,7 +373,7 @@ func TestResolveCapabilities(t *testing.T) {
 // ambient stay empty — see the comment in buildActorOCISpec.
 func TestBuildActorOCISpec_Capabilities(t *testing.T) {
 	want := []string{"CAP_CHOWN", "CAP_KILL"}
-	spec := buildActorOCISpec("actor_uid", "app", []string{"/app"}, nil, nil, "/run/netns/x", nil, nil, want)
+	spec := buildActorOCISpec("worker_pod_uid", "actor_uid", "app", []string{"/app"}, nil, nil, "/run/netns/x", nil, nil, want)
 
 	caps := spec.Process.Capabilities
 	if caps == nil {
@@ -400,7 +406,7 @@ func TestBuildActorOCISpec_Capabilities(t *testing.T) {
 
 // The pause container only reaps, so it is built with no capabilities at all.
 func TestBuildActorOCISpec_NoCapabilitiesForPause(t *testing.T) {
-	spec := buildActorOCISpec("actor_uid", "pause", []string{"/pause"}, nil, nil, "/run/netns/x", nil, nil, nil)
+	spec := buildActorOCISpec("worker_pod_uid", "actor_uid", "pause", []string{"/pause"}, nil, nil, "/run/netns/x", nil, nil, nil)
 
 	caps := spec.Process.Capabilities
 	if caps == nil {
@@ -421,3 +427,50 @@ func TestBuildActorOCISpec_NoCapabilitiesForPause(t *testing.T) {
 		}
 	}
 }
+
+func TestBuildActorOCISpec_ExternalVolumeMounts(t *testing.T) {
+	const workerPodUID = "worker_pod_uid"
+	const actorUID = "actor_uid"
+	extMounts := []*ateletpb.VolumeMount{
+		{Name: "shared-nfs", MountPath: "/mnt/nfs"},
+	}
+	volumes := []*ateletpb.Volume{
+		{
+			Name: "shared-nfs",
+			Source: &ateletpb.Volume_External{
+				External: &ateletpb.ExternalVolumeSource{
+					StorageVolumeId: "pvc-12345",
+					VolumeType:      "nfs.csi.k8s.io",
+				},
+			},
+		},
+	}
+	spec := buildActorOCISpec(
+		workerPodUID,
+		actorUID,
+		"app",
+		[]string{"/app"}, nil, nil,
+		"/run/netns/x",
+		volumes,
+		extMounts,
+		nil,
+	)
+
+	wantSrc := ateompath.VolumeHostPath(workerPodUID, actorUID, "shared-nfs")
+	found := false
+	for _, m := range spec.Mounts {
+		if m.Destination != "/mnt/nfs" {
+			continue
+		}
+		found = true
+		if m.Source != wantSrc {
+			t.Errorf("external volume source = %q, want %q", m.Source, wantSrc)
+		}
+		if m.Type != "bind" {
+			t.Errorf("external volume type = %q, want bind", m.Type)
+		}
+	}
+	if !found {
+		t.Fatalf("external volume mount for /mnt/nfs missing; mounts=%v", spec.Mounts)
+	}
+}

cmd/atelet/volumes.go

@@ -29,13 +29,13 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-func (s *AteomHerder) mountExternalVolumes(ctx context.Context, actorUID string, volumes []*ateletpb.Volume) error {
+func (s *AteomHerder) mountExternalVolumes(ctx context.Context, workerPodUID, actorUID string, volumes []*ateletpb.Volume) error {
 	for _, vol := range volumes {
 		ext := vol.GetExternal()
 		if ext == nil {
 			continue
 		}
-		hostPath := ateompath.VolumeHostPath(actorUID, vol.GetName())
+		hostPath := ateompath.VolumeHostPath(workerPodUID, actorUID, vol.GetName())
 		if err := os.MkdirAll(hostPath, 0o750); err != nil {
 			return fmt.Errorf("failed to create mount point %q: %w", hostPath, err)
 		}
@@ -51,14 +51,14 @@ func (s *AteomHerder) mountExternalVolumes(ctx context.Context, actorUID string,
 	return nil
 }
 
-func (s *AteomHerder) unmountExternalVolumes(ctx context.Context, actorUID string, volumes []*ateletpb.Volume) error {
+func (s *AteomHerder) unmountExternalVolumes(ctx context.Context, workerPodUID, actorUID string, volumes []*ateletpb.Volume) error {
 	var errs []error
 	for _, vol := range volumes {
 		ext := vol.GetExternal()
 		if ext == nil {
 			continue
 		}
-		hostPath := ateompath.VolumeHostPath(actorUID, vol.GetName())
+		hostPath := ateompath.VolumeHostPath(workerPodUID, actorUID, vol.GetName())
 		slog.InfoContext(ctx, "Unmounting volume", slog.String("volume_id", ext.GetStorageVolumeId()), slog.String("host_path", hostPath), slog.String("volume_type", ext.GetVolumeType()))
 		// TODO: Standardize volume plugin lookup and error handling across control plane
 		// and worker plane (e.g. via a shared helper).

cmd/atelet/volumes_test.go

@@ -70,6 +70,8 @@ func TestUnmountExternalVolumes(t *testing.T) {
 		},
 	}
 
+	workerPodUID := "test-worker-pod-123"
+
 	t.Run("success", func(t *testing.T) {
 		fake := &fakeWorkerPlugin{}
 		s := &AteomHerder{
@@ -78,7 +80,7 @@ func TestUnmountExternalVolumes(t *testing.T) {
 			},
 		}
 
-		err := s.unmountExternalVolumes(ctx, actorUID, []*ateletpb.Volume{extVol1, durableVol, extVol2})
+		err := s.unmountExternalVolumes(ctx, workerPodUID, actorUID, []*ateletpb.Volume{extVol1, durableVol, extVol2})
 		if err != nil {
 			t.Fatalf("unmountExternalVolumes failed unexpectedly: %v", err)
 		}
@@ -97,7 +99,7 @@ func TestUnmountExternalVolumes(t *testing.T) {
 			},
 		}
 
-		err := s.unmountExternalVolumes(ctx, actorUID, []*ateletpb.Volume{extVol1})
+		err := s.unmountExternalVolumes(ctx, workerPodUID, actorUID, []*ateletpb.Volume{extVol1})
 		if err == nil {
 			t.Fatal("unmountExternalVolumes returned nil, want blocking error")
 		}
@@ -116,7 +118,7 @@ func TestUnmountExternalVolumes(t *testing.T) {
 			},
 		}
 
-		err := s.unmountExternalVolumes(ctx, actorUID, []*ateletpb.Volume{extVol1, extVol2})
+		err := s.unmountExternalVolumes(ctx, workerPodUID, actorUID, []*ateletpb.Volume{extVol1, extVol2})
 		if err == nil {
 			t.Fatal("unmountExternalVolumes returned nil, want blocking error")
 		}

cmd/ateom-microvm/csi.go

@@ -60,7 +60,7 @@ func csiMounts(mounts []*ateompb.VolumeMount) []specs.Mount {
 // stageCsiVolumes bind-mounts the actor's host CSI volumes directory
 // into the sandbox's shared virtio-fs tree at SharedDir(actorUID)/csi.
 func (s *AteomService) stageCsiVolumes(ctx context.Context, actorUID string) error {
-	src := ateompath.VolumesDir(actorUID)
+	src := ateompath.VolumesDir(s.podUID, actorUID)
 	if _, err := os.Stat(src); err != nil {
 		return fmt.Errorf("while checking CSI volumes dir %q: %w", src, err)
 	}

hack/setup-csi-nfs-kind.sh

@@ -40,26 +40,7 @@ kubectl apply -f "${ROOT}/hack/third_party/csi-driver-nfs/deploy/csi-nfs-driveri
 kubectl apply -f "${ROOT}/hack/third_party/csi-driver-nfs/deploy/csi-nfs-controller.yaml"
 kubectl apply -f "${ROOT}/hack/third_party/csi-driver-nfs/deploy/csi-nfs-node.yaml"
 
-# 3. Patch CSI NFS Node DaemonSet to propagate mounts
-echo "Patching CSI NFS Node DaemonSet..."
-kubectl patch daemonset csi-nfs-node -n kube-system --patch '
-spec:
-  template:
-    spec:
-      containers:
-      - name: nfs
-        volumeMounts:
-        - name: ateom-dir
-          mountPath: /var/lib/ateom-gvisor
-          mountPropagation: Bidirectional
-      volumes:
-      - name: ateom-dir
-        hostPath:
-          path: /var/lib/ateom-gvisor
-          type: DirectoryOrCreate
-'
-
-# 4. Patch CSI NFS Controller Deployment to add socat proxy
+# 3. Patch CSI NFS Controller Deployment to add socat proxy
 echo "Patching CSI NFS Controller Deployment..."
 kubectl patch deployment csi-nfs-controller -n kube-system --patch '
 spec: