diff --git a/packages/apps/kubernetes/images/kubevirt-csi-driver/node.go b/packages/apps/kubernetes/images/kubevirt-csi-driver/node.go index 87ad74a82..7bac6f967 100644 --- a/packages/apps/kubernetes/images/kubevirt-csi-driver/node.go +++ b/packages/apps/kubernetes/images/kubevirt-csi-driver/node.go @@ -2,9 +2,13 @@ package main import ( "context" + "errors" "fmt" "os" + "os/exec" "path/filepath" + "strings" + "time" csi "github.com/container-storage-interface/spec/lib/go/csi" "google.golang.org/grpc/codes" @@ -81,21 +85,23 @@ func (w *WrappedNodeService) NodePublishVolume(ctx context.Context, req *csi.Nod if err != nil { return nil, status.Errorf(codes.Internal, "failed to create temp mount dir: %v", err) } - // Logged, not returned: a leftover temp dir does not make the volume - // unusable, and failing an otherwise good publish only makes kubelet retry. - defer func() { - if err := os.Remove(tmpMount); err != nil { - klog.Warningf("Failed to remove temp dir %s: %v", tmpMount, err) - } - }() rootSource := fmt.Sprintf("%s:%s", host, path) - rootOpts := []string{"nfsvers=4.2", fmt.Sprintf("port=%s", port)} + // nosharecache gives the temp mount a superblock of its own: the forced + // unmount below kills every RPC in flight on its superblock, which would + // otherwise be the one the pods on this node use for the same export. + // The price is a cache of its own too: pods already holding /data may not + // see the entries the migration moves in until their attribute cache + // expires (acdirmax). + rootOpts := []string{"nfsvers=4.2", fmt.Sprintf("port=%s", port), "nosharecache"} if err := w.mounter.Mount(rootSource, tmpMount, "nfs", rootOpts); err != nil { + if rmErr := os.Remove(tmpMount); rmErr != nil { + klog.Warningf("Failed to remove temp dir %s: %v", tmpMount, rmErr) + } return nil, status.Errorf(codes.Internal, "NFS temp mount failed: %v", err) } defer func() { - if err := w.mounter.Unmount(tmpMount); err != nil { + if err := unmountTempMount(w.mounter, tmpMount); err != nil { klog.Warningf("Failed to unmount temp dir %s: %v", tmpMount, err) } }() @@ -140,6 +146,98 @@ func (w *WrappedNodeService) NodePublishVolume(ctx context.Context, req *csi.Nod return &csi.NodePublishVolumeResponse{}, nil } +const ( + // Well inside the 2m kubelet puts on NodePublishVolume (csiTimeout in + // kubernetes pkg/volume/csi/csi_plugin.go). Bounds the unmount only: the + // temp mount is a hard mount, so the mount and the migration before it + // still block for as long as the server is gone. + tempUnmountTimeout = 30 * time.Second + // How long umount's output pipe may stay open once umount has exited or + // been killed; the umount.nfs helper it forked can hold it. + umountPipeGrace = time.Second +) + +// unmountTempMount bounds the unmount: a plain umount of an NFS mount whose +// server is gone blocks forever, which hangs the publish and leaks the mount. +func unmountTempMount(m mount.Interface, dir string) error { + return unmountWithin(m, dir, tempUnmountTimeout) +} + +// unmountWithin gives the whole unmount one deadline. On timeout the unmount +// keeps running in the background and logs its result when it ends; an umount +// stuck in uninterruptible sleep cannot be killed, so no deadline on the +// process itself can be relied on. +// +// The dir is removed by the worker once the unmount succeeds, so an unmount +// that finishes after the deadline still cleans up; removed by the caller it +// would still be a mount point at that moment. A failed unmount keeps the dir. +// Plain Remove rather than RemoveAll: if the unmount did not take, the dir's +// contents are the export's. +func unmountWithin(m mount.Interface, dir string, budget time.Duration) error { + done := make(chan error) + abandoned := make(chan struct{}) + go func() { + var err error + if _, ok := m.(*mount.Mounter); ok { + err = umountForcedAfter(dir, budget/3) + } else { + err = m.Unmount(dir) + } + if err == nil { + if rmErr := os.Remove(dir); rmErr != nil && !os.IsNotExist(rmErr) { + klog.Warningf("Failed to remove temp dir %s: %v", dir, rmErr) + } + } + select { + case done <- err: + case <-abandoned: + if err != nil { + klog.Warningf("unmount of %s failed after the deadline: %v", dir, err) + } else { + klog.Infof("unmount of %s finished after the deadline", dir) + } + } + }() + select { + case err := <-done: + return err + case <-time.After(budget): + close(abandoned) + return fmt.Errorf("unmount of %s still running after %v", dir, budget) + } +} + +// umountForcedAfter replaces mount-utils' UnmountWithForce, which retries with +// -f only once its first umount has returned. That umount forks umount.nfs; +// killed on the deadline, it leaves the helper stuck on the vanished server +// holding the output pipe, and CombinedOutput waits for that pipe, so the -f +// never runs. WaitDelay closes the pipe instead. +func umountForcedAfter(dir string, timeout time.Duration) error { + err := runUmount(timeout, dir) + if !errors.Is(err, context.DeadlineExceeded) { + return err + } + klog.V(2).Infof("Timed out waiting for unmount of %s, trying with -f", dir) + return runUmount(timeout, "-f", dir) +} + +func runUmount(timeout time.Duration, args ...string) error { + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + cmd := exec.CommandContext(ctx, "umount", args...) + cmd.WaitDelay = umountPipeGrace + out, err := cmd.CombinedOutput() + if ctx.Err() != nil { + return fmt.Errorf("umount %s: %w", strings.Join(args, " "), ctx.Err()) + } + // umount2 detaches the mount before it blocks tearing the superblock down, + // so the -f after a killed umount can find nothing left to unmount. + if err != nil && !strings.Contains(string(out), "not mounted") { + return fmt.Errorf("umount %s: %v: %s", strings.Join(args, " "), err, out) + } + return nil +} + // NodeExpandVolume for NFS volumes is a no-op (LINSTOR handles NFS resize automatically). // This should not normally be called for NFS since ControllerExpandVolume returns // NodeExpansionRequired=false, but we handle it gracefully as a safety net. diff --git a/packages/apps/kubernetes/images/kubevirt-csi-driver/node_unmount_test.go b/packages/apps/kubernetes/images/kubevirt-csi-driver/node_unmount_test.go new file mode 100644 index 000000000..77c046e56 --- /dev/null +++ b/packages/apps/kubernetes/images/kubevirt-csi-driver/node_unmount_test.go @@ -0,0 +1,338 @@ +package main + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "testing" + "time" + + csi "github.com/container-storage-interface/spec/lib/go/csi" + "k8s.io/klog/v2" + mount "k8s.io/mount-utils" +) + +// A path that does not exist, so the fake mounter's symlink resolution keeps +// it as is and matches it against its mount table. +const tempMountPath = "/nonexistent/nfs-init-pvc-1-123" + +// On Linux mount.New runs umount on a scratch dir to detect how it reports +// "not mounted", so it is built once, before any test puts a fake umount on +// PATH. +var execMounter = mount.New("") + +func mountedFake(unmountErr error) *mount.FakeMounter { + m := mount.NewFakeMounter([]mount.MountPoint{{Path: tempMountPath, Type: "nfs"}}) + if unmountErr != nil { + m.UnmountFunc = func(string) error { return unmountErr } + } + return m +} + +// fakeUmount puts an umount on PATH that appends its arguments to the returned +// log and then runs plainBody for a plain unmount or forceBody for umount -f. +func fakeUmount(t *testing.T, plainBody string) string { + return fakeUmountForce(t, plainBody, "exit 0") +} + +func fakeUmountForce(t *testing.T, plainBody, forceBody string) string { + t.Helper() + bin := t.TempDir() + log := filepath.Join(bin, "calls") + script := "#!/bin/sh\necho \"$*\" >> " + log + "\n" + + "if [ \"$1\" = \"-f\" ]; then " + forceBody + "; fi\n" + plainBody + "\n" + if err := os.WriteFile(filepath.Join(bin, "umount"), []byte(script), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + return log +} + +func umountCalls(t *testing.T, log string) []string { + t.Helper() + b, err := os.ReadFile(log) + if err != nil && !os.IsNotExist(err) { + t.Fatal(err) + } + return strings.FieldsFunc(string(b), func(r rune) bool { return r == '\n' }) +} + +func tempDir(t *testing.T) string { + t.Helper() + dir := filepath.Join(t.TempDir(), "nfs-init-pvc-1-123") + if err := os.Mkdir(dir, 0o700); err != nil { + t.Fatal(err) + } + return dir +} + +// main wires mount.New(""); unmountWithin runs its own umount only for that +// type, so a different one would silently lose the forced retry. +func TestMountNewIsTheExecMounter(t *testing.T) { + if _, ok := execMounter.(*mount.Mounter); !ok { + t.Fatal("mount.New(\"\") is not *mount.Mounter") + } +} + +func TestUnmountWithinTriesAPlainUmountFirst(t *testing.T) { + log := fakeUmount(t, "exit 0") + dir := tempDir(t) + + if err := unmountWithin(execMounter, dir, 6*time.Second); err != nil { + t.Fatalf("unmountWithin: %v", err) + } + if calls := umountCalls(t, log); !slices.Equal(calls, []string{dir}) { + t.Errorf("umount calls = %v, want just the plain one", calls) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Errorf("temp dir still present after a successful unmount: %v", err) + } +} + +// The plain umount runs the umount.nfs helper as a child. Against a vanished +// server the helper never returns from umount2(), and killing umount on the +// deadline orphans it with the output pipe still open. +func TestUnmountWithinForcesWhenThePlainUmountHangsWithAnOrphan(t *testing.T) { + log := fakeUmount(t, "sleep 30 &\nwait") + dir := tempDir(t) + budget := 6 * time.Second + + start := time.Now() + if err := unmountWithin(execMounter, dir, budget); err != nil { + t.Fatalf("unmountWithin: %v", err) + } + if elapsed := time.Since(start); elapsed >= budget { + t.Errorf("unmountWithin took %v, want under its %v budget", elapsed, budget) + } + if calls := umountCalls(t, log); !slices.Equal(calls, []string{dir, "-f " + dir}) { + t.Errorf("umount calls = %v, want the plain one then -f", calls) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Errorf("temp dir still present after the forced unmount: %v", err) + } +} + +func TestUnmountWithinTakesNotMountedFromTheForcedRetryAsDone(t *testing.T) { + fakeUmountForce(t, "sleep 30 &\nwait", "echo \"umount: $2: not mounted.\" >&2; exit 32") + dir := tempDir(t) + + if err := unmountWithin(execMounter, dir, 6*time.Second); err != nil { + t.Fatalf("unmountWithin: %v", err) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Errorf("temp dir still present after the forced retry found it unmounted: %v", err) + } +} + +func TestUnmountWithinReportsAFailedForcedRetry(t *testing.T) { + log := fakeUmountForce(t, "sleep 30 &\nwait", "echo busy >&2; exit 32") + dir := tempDir(t) + + err := unmountWithin(execMounter, dir, 6*time.Second) + if err == nil || !strings.Contains(err.Error(), "busy") { + t.Fatalf("unmountWithin error = %v, want the umount -f failure with its output", err) + } + if calls := umountCalls(t, log); !slices.Equal(calls, []string{dir, "-f " + dir}) { + t.Errorf("umount calls = %v, want the plain one then -f", calls) + } + if _, err := os.Stat(dir); err != nil { + t.Errorf("temp dir removed although it may still be mounted: %v", err) + } +} + +func TestUnmountWithinDoesNotForceAfterAFailedUmount(t *testing.T) { + log := fakeUmount(t, "echo busy >&2; exit 32") + dir := tempDir(t) + + err := unmountWithin(execMounter, dir, 6*time.Second) + if err == nil || !strings.Contains(err.Error(), "busy") { + t.Fatalf("unmountWithin error = %v, want the umount failure with its output", err) + } + if calls := umountCalls(t, log); !slices.Equal(calls, []string{dir}) { + t.Errorf("umount calls = %v, want no -f after a failure that was not a timeout", calls) + } + if _, err := os.Stat(dir); err != nil { + t.Errorf("temp dir removed although it may still be mounted: %v", err) + } +} + +func TestUnmountTempMountFallsBackToPlainUnmount(t *testing.T) { + m := mountedFake(nil) + if err := unmountTempMount(m, tempMountPath); err != nil { + t.Fatalf("unmountTempMount: %v", err) + } + if mps, _ := m.List(); len(mps) != 0 { + t.Errorf("mount points after unmount = %v, want none", mps) + } + + if err := unmountTempMount(mountedFake(errors.New("device busy")), tempMountPath); err == nil { + t.Error("unmountTempMount succeeded although Unmount failed") + } +} + +// blockingMounter stands for an unmount that does not return on its own, such +// as an umount stuck in uninterruptible sleep, which no signal ends. +type blockingMounter struct { + *mount.FakeMounter + release chan struct{} + err error +} + +func (b *blockingMounter) Unmount(string) error { + <-b.release + return b.err +} + +func TestUnmountWithinGivesUpOnAStalledUnmount(t *testing.T) { + m := &blockingMounter{FakeMounter: mountedFake(nil), release: make(chan struct{})} + defer close(m.release) + + done := make(chan error, 1) + go func() { done <- unmountWithin(m, tempMountPath, 50*time.Millisecond) }() + select { + case err := <-done: + if err == nil { + t.Fatal("unmountWithin reported success for an unmount still running") + } + case <-time.After(5 * time.Second): + t.Fatal("unmountWithin did not return within its budget while the unmount stalled") + } +} + +// An unmount that outlives its budget still finishes in the background, and the +// temp dir can only go once it has: removed any earlier it is still a mount +// point and the remove fails, and nothing would try again. +func TestUnmountWithinRemovesTheDirAfterALateUnmount(t *testing.T) { + dir := tempDir(t) + m := &blockingMounter{FakeMounter: mountedFake(nil), release: make(chan struct{})} + + if err := unmountWithin(m, dir, 50*time.Millisecond); err == nil { + t.Fatal("unmountWithin reported success for an unmount still running") + } + if _, err := os.Stat(dir); err != nil { + t.Fatalf("temp dir gone while its unmount was still running: %v", err) + } + + close(m.release) + waitFor(t, "temp dir removed after the late unmount succeeded", func() bool { + _, err := os.Stat(dir) + return os.IsNotExist(err) + }) +} + +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (s *syncBuffer) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.Write(p) +} + +func (s *syncBuffer) String() string { + s.mu.Lock() + defer s.mu.Unlock() + return s.buf.String() +} + +// The caller has already logged "still running"; without a line from the +// worker nobody can tell a mount that cleaned up late from one that leaked. +func TestUnmountWithinLogsALateResult(t *testing.T) { + var logs syncBuffer + klog.LogToStderr(false) + klog.SetOutput(&logs) + t.Cleanup(func() { + klog.SetOutput(os.Stderr) + klog.LogToStderr(true) + }) + + for _, late := range []error{nil, errors.New("device busy")} { + m := &blockingMounter{FakeMounter: mountedFake(nil), release: make(chan struct{}), err: late} + if err := unmountWithin(m, tempMountPath, 50*time.Millisecond); err == nil { + t.Fatal("unmountWithin reported success for an unmount still running") + } + close(m.release) + want := "unmount of " + tempMountPath + " finished after the deadline" + if late != nil { + want = "unmount of " + tempMountPath + " failed after the deadline: device busy" + } + waitFor(t, "log line "+want, func() bool { return strings.Contains(logs.String(), want) }) + } +} + +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for: %s", what) + } + time.Sleep(10 * time.Millisecond) + } +} + +func TestNodePublishVolumeNFSMountsTheTempRootWithNosharecache(t *testing.T) { + tmpRoot := t.TempDir() + t.Setenv("TMPDIR", tmpRoot) + + var tempOpts []string + m := mount.NewFakeMounter(nil) + m.UnmountFunc = func(path string) error { + for _, mp := range m.MountPoints { + if mp.Path == path { + tempOpts = mp.Opts + } + } + return nil + } + w := &WrappedNodeService{mounter: m} + req := &csi.NodePublishVolumeRequest{ + VolumeId: "pvc-1", + TargetPath: filepath.Join(t.TempDir(), "target"), + PublishContext: map[string]string{nfsExportKey: "nfs://192.0.2.10:2049/export"}, + } + if _, err := w.NodePublishVolume(context.Background(), req); err != nil { + t.Fatalf("NodePublishVolume: %v", err) + } + + // umount -f runs nfs_umount_begin on the superblock, killing every RPC in + // flight on it. Without nosharecache the temp mount shares that superblock + // with the pods' mounts of the same export on this node. + if !slices.Contains(tempOpts, "nosharecache") { + t.Errorf("temp mount options = %v, want nosharecache so a forced unmount stays on its own superblock", tempOpts) + } +} + +type failingMounter struct{ *mount.FakeMounter } + +func (failingMounter) Mount(string, string, string, []string) error { + return errors.New("connection refused") +} + +// The dir is removed only after a successful unmount, so the path that never +// mounts has to remove it on its own. The mounted path cannot be checked the same +// way: the fake mounts nothing, so /data lands in the temp dir itself and a +// plain remove, deliberately not RemoveAll, leaves it. +func TestNodePublishVolumeNFSRemovesTheTempDirWhenTheMountFails(t *testing.T) { + tmpRoot := t.TempDir() + t.Setenv("TMPDIR", tmpRoot) + w := &WrappedNodeService{mounter: failingMounter{mount.NewFakeMounter(nil)}} + req := &csi.NodePublishVolumeRequest{ + VolumeId: "pvc-1", + TargetPath: filepath.Join(t.TempDir(), "target"), + PublishContext: map[string]string{nfsExportKey: "nfs://192.0.2.10:2049/export"}, + } + if _, err := w.NodePublishVolume(context.Background(), req); err == nil { + t.Fatal("NodePublishVolume succeeded although the temp mount failed") + } + if left, _ := filepath.Glob(filepath.Join(tmpRoot, "nfs-init-*")); len(left) != 0 { + t.Errorf("temp dirs left behind: %v", left) + } +}