Move mount, unmount and isMountPoint functionalities to github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mounter (#444)

This is part of the effort of consolidating Mountpoint related
functionalities in `github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint`
package.

---

By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

---------

Signed-off-by: Burak Varlı <burakvar@amazon.co.uk>

Author: unexge

pkg/driver/driver.go

@@ -23,18 +23,19 @@ import (
 	"os"
 	"time"
 
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
+
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/credentialprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/version"
+	mpmounter "github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mppod/watcher"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/util"
-	"github.com/container-storage-interface/spec/lib/go/csi"
-	"google.golang.org/grpc"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
-	"k8s.io/klog/v2"
-	"k8s.io/mount-utils"
 )
 
 const (
@@ -90,20 +91,21 @@ func NewDriver(endpoint string, mpVersion string, nodeID string) (*Driver, error
 	stopCh := make(chan struct{})
 
 	var mounterImpl mounter.Mounter
+	mpMounter := mpmounter.New()
 	if util.UsePodMounter() {
 		podWatcher := watcher.New(clientset, mountpointPodNamespace, podWatcherResyncPeriod)
 		err = podWatcher.Start(stopCh)
 		if err != nil {
 			klog.Fatalf("Failed to start Pod watcher: %v\n", err)
 		}
 
-		mounterImpl, err = mounter.NewPodMounter(podWatcher, credProvider, mount.New(""), nil, kubernetesVersion)
+		mounterImpl, err = mounter.NewPodMounter(podWatcher, credProvider, mpMounter, nil, kubernetesVersion)
 		if err != nil {
 			klog.Fatalln(err)
 		}
 		klog.Infoln("Using pod mounter")
 	} else {
-		mounterImpl, err = mounter.NewSystemdMounter(credProvider, mpVersion, kubernetesVersion)
+		mounterImpl, err = mounter.NewSystemdMounter(credProvider, mpMounter, mpVersion, kubernetesVersion)
 		if err != nil {
 			klog.Fatalln(err)
 		}

pkg/driver/node/mounter/mounter.go

@@ -3,20 +3,13 @@ package mounter
 
 import (
 	"context"
-	"fmt"
 	"os"
 
-	"k8s.io/klog/v2"
-	"k8s.io/mount-utils"
-
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/credentialprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/system"
 )
 
-// https://github.com/awslabs/mountpoint-s3/blob/9ed8b6243f4511e2013b2f4303a9197c3ddd4071/mountpoint-s3/src/cli.rs#L421
-const mountpointDeviceName = "mountpoint-s3"
-
 type ServiceRunner interface {
 	StartService(ctx context.Context, config *system.ExecConfig) (string, error)
 	RunOneshot(ctx context.Context, config *system.ExecConfig) (string, error)
@@ -39,29 +32,3 @@ func MountS3Path() string {
 	}
 	return mountS3Path
 }
-
-// isMountPoint returns whether given `target` is a `mount-s3` mount.
-// We implement additional check on top of `mounter.IsMountPoint` because we need
-// to verify not only that the target is a mount point but also that it is specifically a mount-s3 mount point.
-// This is achieved by calling the `mounter.List()` method to enumerate all mount points.
-func isMountPoint(mounter mount.Interface, target string) (bool, error) {
-	if _, err := os.Stat(target); os.IsNotExist(err) {
-		return false, err
-	}
-
-	mountPoints, err := mounter.List()
-	if err != nil {
-		return false, fmt.Errorf("Failed to list mounts: %w", err)
-	}
-	for _, mp := range mountPoints {
-		if mp.Path == target {
-			if mp.Device != mountpointDeviceName {
-				klog.V(4).Infof("IsMountPoint: %s is not a `mount-s3` mount. Expected device type to be %s but got %s, skipping unmount", target, mountpointDeviceName, mp.Device)
-				continue
-			}
-
-			return true, nil
-		}
-	}
-	return false, nil
-}

pkg/driver/node/mounter/pod_mounter.go

@@ -7,18 +7,17 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
-	"syscall"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
-	"k8s.io/mount-utils"
 
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/credentialprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/envprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/targetpath"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint"
+	mpmounter "github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mountoptions"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mppod"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mppod/watcher"
@@ -33,21 +32,21 @@ const targetDirPerm = fs.FileMode(0755)
 
 // mountSyscall is the function that performs `mount` operation for given `target` with given Mountpoint `args`.
 // It returns mounted FUSE file descriptor as a result.
-// This is mainly exposed for testing, in production platform-native function (`mountSyscallDefault`) will be used.
+// This is mainly exposed for testing, in production platform-native function (`mpmounter.Mount`) will be used.
 type mountSyscall func(target string, args mountpoint.Args) (fd int, err error)
 
 // A PodMounter is a [Mounter] that mounts Mountpoint on pre-created Kubernetes Pod running in the same node.
 type PodMounter struct {
 	podWatcher        *watcher.Watcher
-	mount             mount.Interface
+	mount             *mpmounter.Mounter
 	kubeletPath       string
 	mountSyscall      mountSyscall
 	kubernetesVersion string
 	credProvider      *credentialprovider.Provider
 }
 
 // NewPodMounter creates a new [PodMounter] with given Kubernetes client.
-func NewPodMounter(podWatcher *watcher.Watcher, credProvider *credentialprovider.Provider, mount mount.Interface, mountSyscall mountSyscall, kubernetesVersion string) (*PodMounter, error) {
+func NewPodMounter(podWatcher *watcher.Watcher, credProvider *credentialprovider.Provider, mount *mpmounter.Mounter, mountSyscall mountSyscall, kubernetesVersion string) (*PodMounter, error) {
 	return &PodMounter{
 		podWatcher:        podWatcher,
 		credProvider:      credProvider,
@@ -77,14 +76,12 @@ func (pm *PodMounter) Mount(ctx context.Context, bucketName string, target strin
 
 	podID := credentialCtx.PodID
 
-	err = pm.verifyOrSetupMountTarget(target)
-	if err != nil {
-		return fmt.Errorf("Failed to verify target path can be used as a mount point %q: %w", target, err)
-	}
-
 	isMountPoint, err := pm.IsMountPoint(target)
 	if err != nil {
-		return fmt.Errorf("Could not check if %q is already a mount point: %w", target, err)
+		err = pm.verifyOrSetupMountTarget(target, err)
+		if err != nil {
+			return fmt.Errorf("Failed to verify target path can be used as a mount point %q: %w", target, err)
+		}
 	}
 
 	// TODO: If `target` is a `systemd`-mounted Mountpoint, this would return an error,
@@ -232,8 +229,7 @@ func (pm *PodMounter) Unmount(ctx context.Context, target string, credentialCtx
 
 // IsMountPoint returns whether given `target` is a `mount-s3` mount.
 func (pm *PodMounter) IsMountPoint(target string) (bool, error) {
-	// TODO: Can we just use regular `IsMountPoint` check from `mounter` with containerization?
-	return isMountPoint(pm.mount, target)
+	return pm.mount.CheckMountPoint(target)
 }
 
 // waitForMountpointPod waints until Mountpoint Pod for given `podID` and `volumeName` is in `Running` state.
@@ -300,7 +296,7 @@ func (pm *PodMounter) waitForMount(parentCtx context.Context, target, podName, p
 
 // closeFUSEDevFD closes given FUSE file descriptor.
 func (pm *PodMounter) closeFUSEDevFD(fd int) {
-	err := syscall.Close(fd)
+	err := mpmounter.CloseFD(fd)
 	if err != nil {
 		klog.V(4).Infof("Mount: Failed to close /dev/fuse file descriptor %d: %v\n", fd, err)
 	}
@@ -309,20 +305,15 @@ func (pm *PodMounter) closeFUSEDevFD(fd int) {
 // verifyOrSetupMountTarget checks target path for existence and corrupted mount error.
 // If the target dir does not exists it tries to create it.
 // If the target dir is corrupted (decided with `mount.IsCorruptedMnt`) it tries to unmount it to have a clean mount.
-func (pm *PodMounter) verifyOrSetupMountTarget(target string) error {
-	err := verifyMountPointStatx(target)
-	if err == nil {
-		return nil
-	}
-
+func (pm *PodMounter) verifyOrSetupMountTarget(target string, err error) error {
 	if errors.Is(err, fs.ErrNotExist) {
 		klog.V(5).Infof("Target path does not exists %s, trying to create", target)
 		if err := os.MkdirAll(target, targetDirPerm); err != nil {
 			return fmt.Errorf("Failed to create target directory: %w", err)
 		}
 
 		return nil
-	} else if mount.IsCorruptedMnt(err) {
+	} else if pm.mount.IsMountPointCorrupted(err) {
 		klog.V(4).Infof("Target path %q is a corrupted mount. Trying to unmount", target)
 		if unmountErr := pm.unmountTarget(target); unmountErr != nil {
 			klog.V(4).Infof("Failed to unmount target path %q: %v, original failure of stat: %v", target, unmountErr, err)
@@ -332,6 +323,7 @@ func (pm *PodMounter) verifyOrSetupMountTarget(target string) error {
 		return nil
 	}
 
+	// Some other error that we cannot recover from, just propagate it.
 	return err
 }
 
@@ -358,13 +350,17 @@ func (pm *PodMounter) podPath(pod *corev1.Pod) string {
 	return filepath.Join(pm.kubeletPath, "pods", string(pod.UID))
 }
 
-// mountSyscallWithDefault delegates to `mountSyscall` if set, or fallbacks to platform-native `mountSyscallDefault`.
+// mountSyscallWithDefault delegates to `mountSyscall` if set, or fallbacks to platform-native `mpmounter.Mount`.
 func (pm *PodMounter) mountSyscallWithDefault(target string, args mountpoint.Args) (int, error) {
 	if pm.mountSyscall != nil {
 		return pm.mountSyscall(target, args)
 	}
 
-	return pm.mountSyscallDefault(target, args)
+	opts := mpmounter.MountOptions{
+		ReadOnly:   args.Has(mountpoint.ArgReadOnly),
+		AllowOther: args.Has(mountpoint.ArgAllowOther) || args.Has(mountpoint.ArgAllowRoot),
+	}
+	return pm.mount.Mount(target, opts)
 }
 
 // unmountTarget calls `unmount` syscall on `target`.

pkg/driver/node/mounter/pod_mounter_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/mounter/mountertest"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint"
+	mpmounter "github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mountoptions"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mppod"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mppod/watcher"
@@ -119,7 +120,7 @@ func setup(t *testing.T) *testCtx {
 	err = podWatcher.Start(stopCh)
 	assert.NoError(t, err)
 
-	podMounter, err := mounter.NewPodMounter(podWatcher, credProvider, mount, mountSyscall, testK8sVersion)
+	podMounter, err := mounter.NewPodMounter(podWatcher, credProvider, mpmounter.NewWithMount(mount), mountSyscall, testK8sVersion)
 	assert.NoError(t, err)
 
 	testCtx.podMounter = podMounter

pkg/driver/node/mounter/systemd_mounter.go

@@ -13,26 +13,29 @@ import (
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/credentialprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/envprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint"
+	mpmounter "github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/system"
 )
 
 type SystemdMounter struct {
 	Runner            ServiceRunner
 	Mounter           mount.Interface
+	MpMounter         *mpmounter.Mounter
 	MpVersion         string
 	MountS3Path       string
 	kubernetesVersion string
 	credProvider      *credentialprovider.Provider
 }
 
-func NewSystemdMounter(credProvider *credentialprovider.Provider, mpVersion string, kubernetesVersion string) (*SystemdMounter, error) {
+func NewSystemdMounter(credProvider *credentialprovider.Provider, mpMounter *mpmounter.Mounter, mpVersion string, kubernetesVersion string) (*SystemdMounter, error) {
 	runner, err := system.StartOsSystemdSupervisor()
 	if err != nil {
 		return nil, fmt.Errorf("failed to start systemd supervisor: %w", err)
 	}
 	return &SystemdMounter{
 		Runner:            runner,
 		Mounter:           mount.New(""),
+		MpMounter:         mpMounter,
 		MpVersion:         mpVersion,
 		MountS3Path:       MountS3Path(),
 		kubernetesVersion: kubernetesVersion,
@@ -42,7 +45,7 @@ func NewSystemdMounter(credProvider *credentialprovider.Provider, mpVersion stri
 
 // IsMountPoint returns whether given `target` is a `mount-s3` mount.
 func (m *SystemdMounter) IsMountPoint(target string) (bool, error) {
-	return isMountPoint(m.Mounter, target)
+	return m.MpMounter.CheckMountPoint(target)
 }
 
 // Mount mounts the given bucket at the target path using provided credentials.
@@ -69,11 +72,11 @@ func (m *SystemdMounter) Mount(ctx context.Context, bucketName string, target st
 
 	cleanupDir := false
 
+	isMountPoint, err := m.IsMountPoint(target)
 	// check if the target path exists and is a directory
-	mountpointErr := verifyMountPointStatx(target)
-	if mountpointErr != nil {
+	if err != nil {
 		// does not exist, create the directory
-		if os.IsNotExist(mountpointErr) {
+		if os.IsNotExist(err) {
 			if err := os.MkdirAll(target, 0755); err != nil {
 				return fmt.Errorf("Failed to create target directory: %w", err)
 			}
@@ -85,25 +88,21 @@ func (m *SystemdMounter) Mount(ctx context.Context, bucketName string, target st
 					}
 				}
 			}()
-		}
-		// Corrupted mount, try unmounting
-		if mount.IsCorruptedMnt(mountpointErr) {
+			// Corrupted mount, try unmounting
+		} else if m.MpMounter.IsMountPointCorrupted(err) {
 			klog.V(4).Infof("Mount: Target path %q is a corrupted mount. Trying to unmount.", target)
 			if mntErr := m.Unmount(ctx, target, credentialprovider.CleanupContext{
 				WritePath: credentialCtx.WritePath,
 				PodID:     credentialCtx.PodID,
 				VolumeID:  credentialCtx.VolumeID,
 			}); mntErr != nil {
-				return fmt.Errorf("Unable to unmount the target %q : %v, %v", target, mountpointErr, mntErr)
+				return fmt.Errorf("Unable to unmount the target %q : %v, %v", target, err, mntErr)
 			}
+		} else {
+			return fmt.Errorf("Could not check if %q is a mount point: %v", target, err)
 		}
 	}
 
-	isMountPoint, err := m.IsMountPoint(target)
-	if err != nil {
-		return fmt.Errorf("Could not check if %q is a mount point: %v, %v", target, mountpointErr, err)
-	}
-
 	credEnv, authenticationSource, err := m.credProvider.Provide(ctx, credentialCtx)
 	if err != nil {
 		klog.V(4).Infof("NodePublishVolume: Failed to provide credentials for %s: %v", target, err)

pkg/driver/node/mounter/systemd_mounter_test.go

@@ -8,15 +8,18 @@ import (
 	"strings"
 	"testing"
 
+	"slices"
+
+	"github.com/golang/mock/gomock"
+	"k8s.io/mount-utils"
+
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/credentialprovider"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/mounter"
 	mock_driver "github.com/awslabs/aws-s3-csi-driver/pkg/driver/node/mounter/mocks"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint"
+	mpmounter "github.com/awslabs/aws-s3-csi-driver/pkg/mountpoint/mounter"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/system"
 	"github.com/awslabs/aws-s3-csi-driver/pkg/util/testutil/assert"
-	"github.com/golang/mock/gomock"
-	"k8s.io/mount-utils"
-	"slices"
 )
 
 type mounterTestEnv struct {
@@ -40,6 +43,7 @@ func initMounterTestEnv(t *testing.T) *mounterTestEnv {
 		mounter: &mounter.SystemdMounter{
 			Runner:      mockRunner,
 			Mounter:     mount.NewFakeMounter(nil),
+			MpMounter:   mpmounter.NewWithMount(mount.NewFakeMounter(nil)),
 			MpVersion:   mountpointVersion,
 			MountS3Path: mounter.MountS3Path(),
 		},
@@ -235,7 +239,7 @@ func TestIsMountPoint(t *testing.T) {
 
 	for name, test := range tests {
 		t.Run(name, func(t *testing.T) {
-			mounter := &mounter.SystemdMounter{Mounter: mount.NewFakeMounter(test.procMountsContent)}
+			mounter := &mounter.SystemdMounter{MpMounter: mpmounter.NewWithMount(mount.NewFakeMounter(test.procMountsContent))}
 			isMountPoint, err := mounter.IsMountPoint(test.target)
 			assert.Equals(t, test.isMountPoint, isMountPoint)
 			assert.Equals(t, test.expectErr, err != nil)