Add aws-s3-csi-mounter component (#291)

This is part of
#279.

This new component, `aws-s3-csi-mounter`, will be the entry point for
the container running Mountpoint. It will be responsible for receiving
mount options and FUSE file descriptor and spawning Mountpoint process
until completion.

---

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ı <[email protected]>

Makefile

@@ -129,7 +129,7 @@ install-go-test-coverage:
 
 .PHONY: test
 test:
-	go test -v -race ./pkg/... -coverprofile=./cover.out -covermode=atomic -coverpkg=./pkg/...
+	go test -v -race ./{cmd,pkg}/... -coverprofile=./cover.out -covermode=atomic -coverpkg=./{cmd,pkg}/...
 	# skipping controller test cases because we don't implement controller for static provisioning,
 	# this is a known limitation of sanity testing package: https://github.com/kubernetes-csi/csi-test/issues/214
 	go test -v ./tests/sanity/... -ginkgo.skip="ControllerGetCapabilities" -ginkgo.skip="ValidateVolumeCapabilities"

cmd/aws-s3-csi-mounter/csimounter/csimounter.go

@@ -0,0 +1,73 @@
+package csimounter
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"slices"
+
+	"k8s.io/klog/v2"
+
+	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mountoptions"
+)
+
+// A CmdRunner is responsible for running given `cmd` until completion and returning its exit code and its error (if any).
+// This is mainly exposed for mocking in tests, `DefaultCmdRunner` is always used in non-test environments.
+type CmdRunner func(cmd *exec.Cmd) (int, error)
+
+// DefaultCmdRunner is a real CmdRunner implementation that runs given `cmd`.
+func DefaultCmdRunner(cmd *exec.Cmd) (int, error) {
+	err := cmd.Run()
+	if err != nil {
+		return 0, err
+	}
+	return cmd.ProcessState.ExitCode(), nil
+}
+
+// An Options represents options to use while mounting Mountpoint.
+type Options struct {
+	MountpointPath string
+	MountOptions   mountoptions.Options
+	CmdRunner      CmdRunner
+}
+
+// Run runs Mountpoint with given options until completion and returns its exit code and its error (if any).
+func Run(options Options) (int, error) {
+	if options.CmdRunner == nil {
+		options.CmdRunner = DefaultCmdRunner
+	}
+
+	mountOptions := options.MountOptions
+
+	fuseDev := os.NewFile(uintptr(mountOptions.Fd), "/dev/fuse")
+	if fuseDev == nil {
+		return 0, fmt.Errorf("passed file descriptor %d is invalid", mountOptions.Fd)
+	}
+
+	args := mountOptions.Args
+
+	// By default Mountpoint runs in a detached mode. Here we want to monitor it by relaying its output,
+	// and also we want to wait until it terminates. We're passing `--foreground` to achieve this.
+	const foreground = "--foreground"
+	if !slices.Contains(args, foreground) {
+		args = append(args, foreground)
+	}
+
+	args = append([]string{
+		mountOptions.BucketName,
+		// We pass FUSE fd using `ExtraFiles`, and each entry becomes as file descriptor 3+i.
+		"/dev/fd/3",
+	}, args...)
+
+	cmd := exec.Command(options.MountpointPath, args...)
+	cmd.ExtraFiles = []*os.File{fuseDev}
+	cmd.Env = options.MountOptions.Env
+	// Connect Mountpoint's stdout/stderr to this commands stdout/stderr,
+	// so Mountpoint logs can be viewable with `kubectl logs`.
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
+	klog.Info("Starting Mountpoint process")
+
+	return options.CmdRunner(cmd)
+}

cmd/aws-s3-csi-mounter/csimounter/csimounter_test.go

@@ -0,0 +1,151 @@
+package csimounter_test
+
+import (
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"syscall"
+	"testing"
+
+	"github.com/google/go-cmp/cmp/cmpopts"
+
+	"github.com/awslabs/aws-s3-csi-driver/cmd/aws-s3-csi-mounter/csimounter"
+	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mountoptions"
+	"github.com/awslabs/aws-s3-csi-driver/pkg/util/testutil/assert"
+)
+
+func TestRunningMountpoint(t *testing.T) {
+	mountpointPath := filepath.Join(t.TempDir(), "mount-s3")
+
+	t.Run("Passes bucket name and FUSE device as mount point", func(t *testing.T) {
+		dev := openDevNull(t)
+
+		runner := func(c *exec.Cmd) (int, error) {
+			assertSameFile(t, dev, c.ExtraFiles[0])
+			assert.Equals(t, mountpointPath, c.Path)
+			assert.Equals(t, []string{mountpointPath, "test-bucket", "/dev/fd/3"}, c.Args[:3])
+			return 0, nil
+		}
+
+		exitCode, err := csimounter.Run(csimounter.Options{
+			MountpointPath: mountpointPath,
+			MountOptions: mountoptions.Options{
+				Fd:         int(dev.Fd()),
+				BucketName: "test-bucket",
+			},
+			CmdRunner: runner,
+		})
+		assert.NoError(t, err)
+		assert.Equals(t, 0, exitCode)
+	})
+
+	t.Run("Passes bucket name", func(t *testing.T) {
+		runner := func(c *exec.Cmd) (int, error) {
+			assert.Equals(t, mountpointPath, c.Path)
+			assert.Equals(t, []string{mountpointPath, "test-bucket"}, c.Args[:2])
+			return 0, nil
+		}
+
+		exitCode, err := csimounter.Run(csimounter.Options{
+			MountpointPath: mountpointPath,
+			MountOptions: mountoptions.Options{
+				Fd:         int(openDevNull(t).Fd()),
+				BucketName: "test-bucket",
+			},
+			CmdRunner: runner,
+		})
+		assert.NoError(t, err)
+		assert.Equals(t, 0, exitCode)
+	})
+
+	t.Run("Passes environment variables", func(t *testing.T) {
+		env := []string{"FOO=bar", "BAZ=qux"}
+
+		runner := func(c *exec.Cmd) (int, error) {
+			assert.Equals(t, env, c.Env)
+			return 0, nil
+		}
+
+		exitCode, err := csimounter.Run(csimounter.Options{
+			MountpointPath: mountpointPath,
+			MountOptions: mountoptions.Options{
+				Fd:  int(openDevNull(t).Fd()),
+				Env: env,
+			},
+			CmdRunner: runner,
+		})
+		assert.NoError(t, err)
+		assert.Equals(t, 0, exitCode)
+	})
+
+	t.Run("Adds `--foreground` argument if not passed", func(t *testing.T) {
+		runner := func(c *exec.Cmd) (int, error) {
+			assert.Equals(t, []string{
+				mountpointPath,
+				"test-bucket", "/dev/fd/3",
+				"--foreground",
+			}, c.Args)
+			return 0, nil
+		}
+
+		exitCode, err := csimounter.Run(csimounter.Options{
+			MountpointPath: mountpointPath,
+			MountOptions: mountoptions.Options{
+				Fd:         int(openDevNull(t).Fd()),
+				BucketName: "test-bucket",
+			},
+			CmdRunner: runner,
+		})
+		assert.NoError(t, err)
+		assert.Equals(t, 0, exitCode)
+
+		exitCode, err = csimounter.Run(csimounter.Options{
+			MountpointPath: mountpointPath,
+			MountOptions: mountoptions.Options{
+				Fd:         int(openDevNull(t).Fd()),
+				BucketName: "test-bucket",
+				Args:       []string{"--foreground"},
+			},
+			CmdRunner: runner,
+		})
+		assert.NoError(t, err)
+		assert.Equals(t, 0, exitCode)
+	})
+
+	t.Run("Fails if file descriptor is invalid", func(t *testing.T) {
+		_, err := csimounter.Run(csimounter.Options{
+			MountpointPath: mountpointPath,
+			MountOptions: mountoptions.Options{
+				Fd:         -1,
+				BucketName: "test-bucket",
+			},
+		})
+		assert.Equals(t, cmpopts.AnyError, err)
+	})
+}
+
+func openDevNull(t *testing.T) *os.File {
+	file, err := os.Open(os.DevNull)
+	assert.NoError(t, err)
+	t.Cleanup(func() {
+		err = file.Close()
+		if err != nil {
+			log.Printf("Failed to close file handle: %v\n", err)
+		}
+	})
+	return file
+}
+
+func assertSameFile(t *testing.T, want *os.File, got *os.File) {
+	var wantStat = &syscall.Stat_t{}
+	err := syscall.Fstat(int(want.Fd()), wantStat)
+	assert.NoError(t, err)
+
+	var gotStat = &syscall.Stat_t{}
+	err = syscall.Fstat(int(got.Fd()), gotStat)
+	assert.NoError(t, err)
+
+	assert.Equals(t, wantStat.Dev, gotStat.Dev)
+	assert.Equals(t, wantStat.Ino, gotStat.Ino)
+}

cmd/aws-s3-csi-mounter/main.go

@@ -0,0 +1,56 @@
+// WIP: Part of https://github.com/awslabs/mountpoint-s3-csi-driver/issues/279.
+//
+// `aws-s3-csi-mounter` is the entrypoint binary running on Mountpoint Pods.
+// It is responsible for receiving mount options from the CSI Driver Node Pod,
+// and spawning a Mountpoint instance in turn.
+// It will then wait until Mountpoint process terminates (which normally happens as a result of `unmount`).
+package main
+
+import (
+	"context"
+	"flag"
+	"os"
+	"path/filepath"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	"github.com/awslabs/aws-s3-csi-driver/cmd/aws-s3-csi-mounter/csimounter"
+	"github.com/awslabs/aws-s3-csi-driver/pkg/podmounter/mountoptions"
+)
+
+var mountSockPath = flag.String("mount-sock-path", "/comm/mount.sock", "Path of the Unix socket to receive mount options from.")
+var mountSockRecvTimeout = flag.Duration("mount-sock-recv-timeout", 2*time.Minute, "Timeout for receiving mount options from passed Unix socket.")
+var mountpointBinDir = flag.String("mountpoint-bin-dir", os.Getenv("MOUNTPOINT_BIN_DIR"), "Directory of mount-s3 binary.")
+
+const mountpointBin = "mount-s3"
+
+func main() {
+	klog.InitFlags(nil)
+	flag.Parse()
+
+	mountpointBinFullPath := filepath.Join(*mountpointBinDir, mountpointBin)
+	mountOptions := recvMountOptions()
+
+	exitCode, err := csimounter.Run(csimounter.Options{
+		MountpointPath: mountpointBinFullPath,
+		MountOptions:   mountOptions,
+	})
+	if err != nil {
+		klog.Fatalf("Failed to run Mountpoint: %v\n", err)
+	}
+	klog.Infof("Mountpoint exited with %d exit code\n", exitCode)
+	os.Exit(exitCode)
+}
+
+func recvMountOptions() mountoptions.Options {
+	ctx, cancel := context.WithTimeout(context.Background(), *mountSockRecvTimeout)
+	defer cancel()
+	klog.Infof("Trying to receive mount options from %s", *mountSockPath)
+	options, err := mountoptions.Recv(ctx, *mountSockPath)
+	if err != nil {
+		klog.Fatalf("Failed to receive mount options from %s: %v\n", *mountSockPath, err)
+	}
+	klog.Infof("Mount options has been received from %s", *mountSockPath)
+	return options
+}

go.mod

@@ -63,7 +63,7 @@ require (
 	github.com/fsnotify/fsnotify v1.4.9 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.6.0
 	github.com/moby/sys/mountinfo v0.7.1 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect