diff --git a/docs/upgrade.md b/docs/upgrade.md index e8084c5b3a..789fb3eeec 100644 --- a/docs/upgrade.md +++ b/docs/upgrade.md @@ -1409,6 +1409,14 @@ survive it. - **During the roll:** an actor image only a pull secret can read is pulled once its node's atelet runs 1.18; a 1.17 atelet still pulls it anonymously or through the providers, as before. +- **Behavior, CSI volumes:** a read-write-many or read-only-many volume of a + driver that advertises staging is staged once per node at one path its + targets share and unstaged after its last target, so a driver that tracks + its stagings (csi-driver-host-path) accepts an actor's second mount of it. + csi-driver-nfs advertises no staging and is unaffected. +- **During the roll, CSI volumes:** a staging a 1.17 atelet made per target + is unstaged at its own path when that target is unpublished; nothing to do + by hand. - **Goldens:** kept. - **Reset:** none. **Conversations lost:** none. diff --git a/fork/existing-volume-mounted-per-actor-at-sub.md b/fork/existing-volume-mounted-per-actor-at-sub.md index eb41ad427a..b0b8c27af9 100644 --- a/fork/existing-volume-mounted-per-actor-at-sub.md +++ b/fork/existing-volume-mounted-per-actor-at-sub.md @@ -7,18 +7,19 @@ status: carried commits: - 'feat: mount an existing volume per actor at a sub-path' - 'fix(volume): publish a read-only mount with the volume''s own mode' + - 'fix(volume): stage a multi-node volume once per node' upstream: outcome: open refs: [agent-substrate/substrate#1637, agent-substrate/substrate#1988, agent-substrate/substrate#1994] exit: as#1988 merged (gated); drop the ungated fork API depends_on: [read-write-mounts-missing-sub-path-directory, resources-deepequal-unchanged-value-check-generated-declarat, giantswarm/kagent-upstream:session-the-sessions-actor-with-its-workspace-directory] -verified: 2026-10-20 +verified: 2026-10-24 --- # An existing volume mounted per actor at a sub-path ## Change -An existing volume mounted per actor at a sub-path: an ActorTemplate declares it (`Volume.existing_volume`, field 10001, an empty marker), `CreateActor` supplies it (`Actor.existing_volumes`, field 10001, immutable; `ExistingVolume` = name, CSI driver, volume handle, access mode `READ_WRITE_MANY`/`READ_ONLY_MANY` in the shape of upstream's `VolumeAccessMode`, and the `sub_path` the actor sees as the volume's root), and `VolumeMount` gains `sub_path` and `read_only` (fields 10001, 10002), so one volume mounts at several paths. `CreateActor` refuses a reference to a volume the template does not declare as existing, a driver without a `CSIDriverConfig`, a handle no PersistentVolume of the driver holds and an access mode the PersistentVolume does not permit (ate-api-server reads PersistentVolumes: a new ClusterRole rule); the PersistentVolume's `volumeAttributes` reach the driver at resume. An unsupplied existing volume contributes neither itself nor its mounts. Substrate never creates, deletes or detaches one: pause, resume and delete only unmount and mount it; a multi-node volume is staged per target, and atelet binds each sub-path from a descriptor opened beneath the volume's root without following symbolic links (`openat2` `RESOLVE_BENEATH|RESOLVE_NO_SYMLINKS`), so a missing directory or a planted link fails the mount. An actor with existing volumes boots from its image instead of the golden snapshot and cannot be cloned from a tag. Tests: unit (validation, workload spec, CSI staging, the sub-path resolver), the `existingvolumes` e2e on kind with the CSI NFS driver (`E2E_CSI_NFS=1` in pr-workflow) +An existing volume mounted per actor at a sub-path: an ActorTemplate declares it (`Volume.existing_volume`, field 10001, an empty marker), `CreateActor` supplies it (`Actor.existing_volumes`, field 10001, immutable; `ExistingVolume` = name, CSI driver, volume handle, access mode `READ_WRITE_MANY`/`READ_ONLY_MANY` in the shape of upstream's `VolumeAccessMode`, and the `sub_path` the actor sees as the volume's root), and `VolumeMount` gains `sub_path` and `read_only` (fields 10001, 10002), so one volume mounts at several paths. `CreateActor` refuses a reference to a volume the template does not declare as existing, a driver without a `CSIDriverConfig`, a handle no PersistentVolume of the driver holds and an access mode the PersistentVolume does not permit (ate-api-server reads PersistentVolumes: a new ClusterRole rule); the PersistentVolume's `volumeAttributes` reach the driver at resume. An unsupplied existing volume contributes neither itself nor its mounts. Substrate never creates, deletes or detaches one: pause, resume and delete only unmount and mount it; a multi-node volume is staged once per node at one path all its targets share (the CSI spec allows one `staging_target_path` per volume, and csi-driver-host-path refuses a second), each target recorded beside it, and unstaged after the last target that holds it, a staging an earlier release made per target unstaged at its own path; atelet binds each sub-path from a descriptor opened beneath the volume's root without following symbolic links (`openat2` `RESOLVE_BENEATH|RESOLVE_NO_SYMLINKS`), so a missing directory or a planted link fails the mount. An actor with existing volumes boots from its image instead of the golden snapshot and cannot be cloned from a tag. Tests: unit (validation, workload spec, CSI staging, the sub-path resolver), the `existingvolumes` e2e on kind with the CSI NFS driver (`E2E_CSI_NFS=1` in pr-workflow) ## Reason @@ -28,6 +29,8 @@ Sessions on one workspace each need their own directory of one read-write-many v follows agent-substrate/substrate#1637 (shared volumes across actors, open), in the shape proposed there (https://github.com/agent-substrate/substrate/issues/1637#issuecomment-6083442678), on top of #1988's access modes (open); upstream puts it behind the Preview gate of #1994 (open), which this line does not carry, so it is ungated here. Exit: drops at the re-pin that carries #1637's API -[#234](https://github.com/giantswarm/substrate/pull/234) (`feat: mount an existing volume per actor at a sub-path`); also [#342](https://github.com/giantswarm/substrate/pull/342) (`fix(volume): publish a read-only mount with the volume's own mode`) +[#234](https://github.com/giantswarm/substrate/pull/234) (`feat: mount an existing volume per actor at a sub-path`); also [#342](https://github.com/giantswarm/substrate/pull/342) (`fix(volume): publish a read-only mount with the volume's own mode`); and [#542](https://github.com/giantswarm/substrate/pull/542) (`fix(volume): stage a multi-node volume once per node`) + +2026-10-24: the per-target staging met a driver that stages: csi-driver-host-path refused an actor's second mount of one workspace volume (its directory and the mirrors) with `volume … is already staged at …`; csi-driver-nfs advertises no staging, so the NFS e2e never staged. Reworked onto one staging per node with the targets recorded; agent-substrate/substrate#1988 stages once per volume already, without the hold of the other targets. 2026-10-20: agent-substrate/substrate#1994 (merged 2026-10-15, `756bb9e1`, not in the v0.5.0-alpha2 pin) adds the Preview gate the gated shape of #1988 needs, not the exit, so the outcome stays open. diff --git a/internal/volume/csi/plugin.go b/internal/volume/csi/plugin.go index 4574f8e47b..d601c2552b 100644 --- a/internal/volume/csi/plugin.go +++ b/internal/volume/csi/plugin.go @@ -271,6 +271,9 @@ func (p *Plugin) DetachVolume(ctx context.Context, volumeID string, node string) // the volume's mode (the AWS EFS driver refuses MULTI_NODE_READER_ONLY for a // read-write-many volume). func (p *Plugin) MountVolume(ctx context.Context, req volume.MountVolumeRequest) error { + if p.stagesShared(req.Mode) { + defer lockStaging(p.volumeStagingDir(req.VolumeID))() + } stagingPath, err := p.stageVolume(ctx, req) if err != nil { return err @@ -287,19 +290,29 @@ func (p *Plugin) MountVolume(ctx context.Context, req volume.MountVolumeRequest) } if _, err := p.client.NodePublishVolume(ctx, csiReq); err != nil { - return fmt.Errorf("CSI NodePublishVolume failed: %w", err) + err = fmt.Errorf("CSI NodePublishVolume failed: %w", err) + if stagingPath != "" && req.Mode.MultiNode() { + if releaseErr := p.releaseSharedStaging(ctx, req.VolumeID, req.TargetPath); releaseErr != nil { + slog.WarnContext(ctx, "failed to release the staging of a volume whose publish failed", slog.String("volume_id", req.VolumeID), slog.Any("error", releaseErr)) + } + } + return err } return nil } // stageVolume calls NodeStageVolume if the driver supports it. It returns the staging -// path, or an empty string if the volume was not staged. +// path, or an empty string if the volume was not staged. A volume of a +// multi-node mode is staged at one path per node, which every target shares +// (the CSI spec allows one staging_target_path per volume): staging it again +// is the driver's idempotent no-op, and the target is recorded so that the +// last target's unmount unstages it. func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) (string, error) { if !p.SupportsNodeCapability(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) { return "", nil } - stagingPath := p.stagingPath(req.VolumeID, req.TargetPath, req.Mode) + stagingPath := p.stagingPath(req.VolumeID, req.Mode) if err := os.MkdirAll(stagingPath, 0750); err != nil { return "", fmt.Errorf("failed to create staging directory %q: %w", stagingPath, err) } @@ -322,12 +335,23 @@ func (p *Plugin) stageVolume(ctx context.Context, req volume.MountVolumeRequest) if err != nil { return "", fmt.Errorf("CSI NodeStageVolume failed: %w", err) } + if req.Mode.MultiNode() { + if err := p.recordTarget(req.VolumeID, req.TargetPath); err != nil { + if releaseErr := p.releaseSharedStaging(ctx, req.VolumeID, req.TargetPath); releaseErr != nil { + slog.WarnContext(ctx, "failed to release the staging of a volume whose target was not recorded", slog.String("volume_id", req.VolumeID), slog.Any("error", releaseErr)) + } + return "", err + } + } return stagingPath, nil } // UnmountVolume maps to CSI Node NodeUnpublishVolume. // It also handles NodeUnstageVolume if staging was used. func (p *Plugin) UnmountVolume(ctx context.Context, volumeID string, targetPath string, mode volume.AccessMode) error { + if p.stagesShared(mode) { + defer lockStaging(p.volumeStagingDir(volumeID))() + } req := &csi.NodeUnpublishVolumeRequest{ VolumeId: volumeID, TargetPath: targetPath, @@ -341,13 +365,21 @@ func (p *Plugin) UnmountVolume(ctx context.Context, volumeID string, targetPath } // unstageVolume calls NodeUnstageVolume and removes the staging directory if the -// driver supports staging. +// driver supports staging. A volume of a multi-node mode stays staged while +// another target on the node still has it. func (p *Plugin) unstageVolume(ctx context.Context, volumeID, targetPath string, mode volume.AccessMode) error { if !p.SupportsNodeCapability(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) { return nil } + if mode.MultiNode() { + return p.releaseSharedStaging(ctx, volumeID, targetPath) + } + return p.unstageAt(ctx, volumeID, p.stagingPath(volumeID, mode)) +} - stagingPath := p.stagingPath(volumeID, targetPath, mode) +// unstageAt calls NodeUnstageVolume for one staging path and removes its +// directory. +func (p *Plugin) unstageAt(ctx context.Context, volumeID, stagingPath string) error { req := &csi.NodeUnstageVolumeRequest{ VolumeId: volumeID, StagingTargetPath: stagingPath, @@ -362,22 +394,102 @@ func (p *Plugin) unstageVolume(ctx context.Context, volumeID, targetPath string, } removeStagingDir(ctx, stagingPath) - if mode.MultiNode() { - // The volume's directory of per-target stagings goes with its last one. - _ = os.Remove(filepath.Dir(stagingPath)) - } return nil } -// stagingPath is where a volume is staged on the node. A volume of a -// multi-node mode is staged once per target, so that one actor unmounting it -// never unstages it under another actor on the same node. -func (p *Plugin) stagingPath(volumeID, targetPath string, mode volume.AccessMode) string { +// The layout of a multi-node volume's staging directory on the node +// (volumeStagingDir): the one staging path every target shares, and a file +// per target that has the volume published, named by targetKey. A release +// before this layout staged each target at /; +// such a staging is unstaged where it is. +const ( + sharedStagingDir = "staged" + stagedTargetsDir = "targets" +) + +// stagesShared reports whether a mount of the mode goes through the shared +// staging of a multi-node volume. +func (p *Plugin) stagesShared(mode volume.AccessMode) bool { + return mode.MultiNode() && p.SupportsNodeCapability(csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME) +} + +// volumeStagingDir is the directory of a volume's staging on the node. +func (p *Plugin) volumeStagingDir(volumeID string) string { + return filepath.Join(p.stagingDirPrefix, volumeID) +} + +// stagingPath is where a volume is staged on the node: once per node, a +// multi-node volume's shared by every target. +func (p *Plugin) stagingPath(volumeID string, mode volume.AccessMode) string { if !mode.MultiNode() { - return filepath.Join(p.stagingDirPrefix, volumeID) + return p.volumeStagingDir(volumeID) } + return filepath.Join(p.volumeStagingDir(volumeID), sharedStagingDir) +} + +// targetKey names a target path within a volume's staging directory. +func targetKey(targetPath string) string { sum := sha256.Sum256([]byte(targetPath)) - return filepath.Join(p.stagingDirPrefix, volumeID, hex.EncodeToString(sum[:8])) + return hex.EncodeToString(sum[:8]) +} + +// recordTarget records that a target has a multi-node volume published. +func (p *Plugin) recordTarget(volumeID, targetPath string) error { + dir := filepath.Join(p.volumeStagingDir(volumeID), stagedTargetsDir) + if err := os.MkdirAll(dir, 0750); err != nil { + return fmt.Errorf("failed to create the staged targets directory %q: %w", dir, err) + } + record := filepath.Join(dir, targetKey(targetPath)) + if err := os.WriteFile(record, nil, 0640); err != nil { + return fmt.Errorf("failed to record the staged target %q: %w", record, err) + } + return nil +} + +// releaseSharedStaging drops a target's hold on a multi-node volume's +// staging and unstages the volume once no target holds it. A target a +// release before the shared layout staged on its own is unstaged at its own +// path. +func (p *Plugin) releaseSharedStaging(ctx context.Context, volumeID, targetPath string) error { + dir := p.volumeStagingDir(volumeID) + key := targetKey(targetPath) + if fi, err := os.Stat(filepath.Join(dir, key)); err == nil && fi.IsDir() { + if err := p.unstageAt(ctx, volumeID, filepath.Join(dir, key)); err != nil { + return err + } + _ = os.Remove(dir) + return nil + } + + targets := filepath.Join(dir, stagedTargetsDir) + if err := os.Remove(filepath.Join(targets, key)); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("failed to remove the staged target record of %q: %w", targetPath, err) + } + remaining, err := os.ReadDir(targets) + if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("failed to read the staged targets of volume %q: %w", volumeID, err) + } + if len(remaining) > 0 { + return nil + } + if err := p.unstageAt(ctx, volumeID, filepath.Join(dir, sharedStagingDir)); err != nil { + return err + } + _ = os.Remove(targets) + _ = os.Remove(dir) + return nil +} + +// stagingLocks serializes the staging of one multi-node volume across every +// Plugin of the process, by its staging directory: a mount must not stage +// while the last other target's unmount unstages. +var stagingLocks sync.Map + +// lockStaging locks a volume's staging and returns the unlock. +func lockStaging(dir string) func() { + mu, _ := stagingLocks.LoadOrStore(dir, &sync.Mutex{}) + mu.(*sync.Mutex).Lock() + return mu.(*sync.Mutex).Unlock } // mountCapability is the mount capability of an access mode. diff --git a/internal/volume/csi/plugin_test.go b/internal/volume/csi/plugin_test.go index 585259eb64..fffbccaab1 100644 --- a/internal/volume/csi/plugin_test.go +++ b/internal/volume/csi/plugin_test.go @@ -20,6 +20,7 @@ import ( "net" "os" "path/filepath" + "sync" "testing" "github.com/agent-substrate/substrate/internal/volume" @@ -833,8 +834,9 @@ func TestPlugin_MountVolumeMultiNode(t *testing.T) { t.Fatalf("MountVolume %s: %v", target, err) } } - if len(staged) != 2 || staged[0] == staged[1] { - t.Fatalf("staging paths = %v, want one per target", staged) + shared := filepath.Join(plugin.stagingDirPrefix, "shared", sharedStagingDir) + if len(staged) != 2 || staged[0] != shared || staged[1] != shared { + t.Fatalf("staging paths = %v, want %q for both targets", staged, shared) } for _, req := range published { if got := req.GetVolumeCapability().GetAccessMode().GetMode(); got != csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER { @@ -845,12 +847,22 @@ func TestPlugin_MountVolumeMultiNode(t *testing.T) { } } - // One actor's unmount unstages only its own staging. + // One actor's unmount leaves the volume staged for the other; the last + // one unstages it. if err := plugin.UnmountVolume(ctx, "shared", targetA, volume.ReadWriteMany); err != nil { t.Fatalf("UnmountVolume: %v", err) } - if len(unstaged) != 1 || unstaged[0] != staged[0] { - t.Errorf("unstaged = %v, want only %q", unstaged, staged[0]) + if len(unstaged) != 0 { + t.Errorf("unstaged = %v after the first of two targets, want nothing", unstaged) + } + if err := plugin.UnmountVolume(ctx, "shared", targetB, volume.ReadWriteMany); err != nil { + t.Fatalf("UnmountVolume: %v", err) + } + if len(unstaged) != 1 || unstaged[0] != shared { + t.Errorf("unstaged = %v after the last target, want %q", unstaged, shared) + } + if _, err := os.Stat(filepath.Join(plugin.stagingDirPrefix, "shared")); !os.IsNotExist(err) { + t.Errorf("the volume's staging directory is left after its last target: %v", err) } // A read-only-many volume is published read-only. @@ -881,3 +893,182 @@ func TestPlugin_MountVolumeMultiNode(t *testing.T) { t.Errorf("read-only publish of a read-write-many volume = %v, want read-only MULTI_NODE_MULTI_WRITER", req) } } + +// strictStagingDriver stages a volume at one path per node and refuses +// another, as csi-driver-host-path does and the CSI spec asks of the +// orchestrator. +func strictStagingDriver(staged map[string]string, unstaged *[]string, publish func(*csi.NodePublishVolumeRequest) error) *mockCSIDriver { + return &mockCSIDriver{ + nodeStageVolumeFunc: func(_ context.Context, req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { + if at, ok := staged[req.GetVolumeId()]; ok && at != req.GetStagingTargetPath() { + return nil, status.Errorf(codes.FailedPrecondition, "volume %q is already staged at %q", req.GetVolumeId(), at) + } + staged[req.GetVolumeId()] = req.GetStagingTargetPath() + return &csi.NodeStageVolumeResponse{}, nil + }, + nodeUnstageVolumeFunc: func(_ context.Context, req *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) { + *unstaged = append(*unstaged, req.GetStagingTargetPath()) + if staged[req.GetVolumeId()] == req.GetStagingTargetPath() { + delete(staged, req.GetVolumeId()) + } + return &csi.NodeUnstageVolumeResponse{}, nil + }, + nodePublishVolumeFunc: func(_ context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) { + if publish != nil { + if err := publish(req); err != nil { + return nil, err + } + } + return &csi.NodePublishVolumeResponse{}, nil + }, + } +} + +func newStagingTestPlugin(t *testing.T, driver *mockCSIDriver, stagingDir string) *Plugin { + t.Helper() + endpoint, cleanup := startMockCSIDriver(t, driver) + t.Cleanup(cleanup) + client, err := NewCSIClient(endpoint, nil) + if err != nil { + t.Fatalf("failed to create CSI client: %v", err) + } + t.Cleanup(func() { client.Close() }) + plugin := NewPlugin(client) + plugin.stagingDirPrefix = stagingDir + plugin.InitNodeCapabilities(t.Context()) + return plugin +} + +// One actor mounting a read-write-many volume twice (its own directory and +// the mirrors) and a second actor on the same node mounting it too are all +// served by one staging, which a driver that allows one staging path per +// volume accepts; the staging outlives every unmount but the last, across a +// restart of the process in between. +func TestPlugin_MultiNodeVolumeStagedOncePerNode(t *testing.T) { + staged := map[string]string{} + var unstaged []string + driver := strictStagingDriver(staged, &unstaged, nil) + dir := filepath.Join(t.TempDir(), "staging") + plugin := newStagingTestPlugin(t, driver, dir) + ctx := t.Context() + + targets := []string{"/actors/a/session", "/actors/a/mirrors", "/actors/b/session"} + for i, target := range targets { + req := volume.MountVolumeRequest{VolumeID: "workspace", TargetPath: target, Mode: volume.ReadWriteMany, ReadOnly: i == 1} + if err := plugin.MountVolume(ctx, req); err != nil { + t.Fatalf("MountVolume %s: %v", target, err) + } + } + if want := filepath.Join(dir, "workspace", sharedStagingDir); staged["workspace"] != want { + t.Fatalf("staged at %q, want %q", staged["workspace"], want) + } + + restarted := newStagingTestPlugin(t, driver, dir) + for _, target := range targets[:2] { + if err := restarted.UnmountVolume(ctx, "workspace", target, volume.ReadWriteMany); err != nil { + t.Fatalf("UnmountVolume %s: %v", target, err) + } + } + if len(unstaged) != 0 { + t.Fatalf("unstaged %v while %s still has the volume", unstaged, targets[2]) + } + if err := restarted.UnmountVolume(ctx, "workspace", targets[2], volume.ReadWriteMany); err != nil { + t.Fatalf("UnmountVolume %s: %v", targets[2], err) + } + if len(unstaged) != 1 || len(staged) != 0 { + t.Errorf("after the last target: unstaged %v, still staged %v; want one unstage and nothing staged", unstaged, staged) + } +} + +// A target a release before the shared layout staged at its own path is +// unstaged there, leaving the shared staging of the targets mounted since. +func TestPlugin_MultiNodeVolumeLegacyStagingUnstagedWhereItIs(t *testing.T) { + staged := map[string]string{} + var unstaged []string + dir := filepath.Join(t.TempDir(), "staging") + plugin := newStagingTestPlugin(t, strictStagingDriver(staged, &unstaged, nil), dir) + ctx := t.Context() + + legacy := filepath.Join(dir, "workspace", targetKey("/actors/old/session")) + if err := os.MkdirAll(legacy, 0750); err != nil { + t.Fatal(err) + } + if err := plugin.UnmountVolume(ctx, "workspace", "/actors/old/session", volume.ReadWriteMany); err != nil { + t.Fatalf("UnmountVolume: %v", err) + } + if len(unstaged) != 1 || unstaged[0] != legacy { + t.Errorf("unstaged %v, want the legacy staging %q", unstaged, legacy) + } + if _, err := os.Stat(legacy); !os.IsNotExist(err) { + t.Errorf("legacy staging directory left: %v", err) + } +} + +// A publish that fails releases the target's hold: the last holder gone, the +// volume is unstaged. +func TestPlugin_MultiNodeVolumeFailedPublishReleasesStaging(t *testing.T) { + staged := map[string]string{} + var unstaged []string + refuse := func(*csi.NodePublishVolumeRequest) error { return status.Error(codes.Internal, "refused") } + dir := filepath.Join(t.TempDir(), "staging") + plugin := newStagingTestPlugin(t, strictStagingDriver(staged, &unstaged, refuse), dir) + + err := plugin.MountVolume(t.Context(), volume.MountVolumeRequest{VolumeID: "workspace", TargetPath: "/actors/a/session", Mode: volume.ReadWriteMany}) + if err == nil { + t.Fatal("MountVolume succeeded, want the publish's refusal") + } + if len(staged) != 0 || len(unstaged) != 1 { + t.Errorf("after a failed publish: staged %v, unstaged %v; want the volume unstaged", staged, unstaged) + } + if _, err := os.Stat(filepath.Join(dir, "workspace")); !os.IsNotExist(err) { + t.Errorf("the volume's staging directory is left after a failed publish: %v", err) + } +} + +// Targets mounting and unmounting one volume at the same time stage it once +// and unstage it once, after the last. +func TestPlugin_MultiNodeVolumeConcurrentTargets(t *testing.T) { + var mu sync.Mutex + stages, unstages := 0, 0 + driver := &mockCSIDriver{ + nodeStageVolumeFunc: func(context.Context, *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) { + mu.Lock() + defer mu.Unlock() + stages++ + return &csi.NodeStageVolumeResponse{}, nil + }, + nodeUnstageVolumeFunc: func(context.Context, *csi.NodeUnstageVolumeRequest) (*csi.NodeUnstageVolumeResponse, error) { + mu.Lock() + defer mu.Unlock() + unstages++ + return &csi.NodeUnstageVolumeResponse{}, nil + }, + } + plugin := newStagingTestPlugin(t, driver, filepath.Join(t.TempDir(), "staging")) + ctx := t.Context() + + const n = 16 + run := func(op func(target string) error) { + var wg sync.WaitGroup + errs := make(chan error, n) + for i := range n { + wg.Go(func() { errs <- op(fmt.Sprintf("/actors/%d/session", i)) }) + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + } + run(func(target string) error { + return plugin.MountVolume(ctx, volume.MountVolumeRequest{VolumeID: "workspace", TargetPath: target, Mode: volume.ReadWriteMany}) + }) + run(func(target string) error { + return plugin.UnmountVolume(ctx, "workspace", target, volume.ReadWriteMany) + }) + if stages != n || unstages != 1 { + t.Errorf("stages %d, unstages %d; want %d idempotent stages and one unstage", stages, unstages, n) + } +}