refactor: split non-mounting actions into interceptors

Author: iltyty

pkg/mounter/cmd_mounter.go

@@ -7,41 +7,32 @@ import (
 	"os/exec"
 	"time"
 
-	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
 	"k8s.io/mount-utils"
 )
 
 const timeout = time.Second * 10
 
 type OssCmdMounter struct {
-	mount.Interface
-	volumeId string
 	execPath string
+	volumeID string
+	mount.Interface
 }
 
-func NewOssCmdMounter(execPath, volumeId string, inner mount.Interface) Mounter {
+var _ Mounter = &OssCmdMounter{}
+
+func NewOssCmdMounter(execPath, volumeID string, inner mount.Interface) Mounter {
 	return &OssCmdMounter{
 		execPath:  execPath,
-		volumeId:  volumeId,
+		volumeID:  volumeID,
 		Interface: inner,
 	}
 }
 
-func (m *OssCmdMounter) ExtendedMount(source, target, fstype string, options []string, params *ExtendedMountParams) error {
+func (m *OssCmdMounter) ExtendedMount(_ context.Context, op *MountOperation) error {
 	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(timeout))
 	defer cancel()
 
-	// Parameters in ExtendedMountParams are optional
-	if params == nil {
-		params = &ExtendedMountParams{}
-	}
-	passwd, err := utils.SaveOssSecretsToFile(params.Secrets, fstype)
-	if err != nil {
-		return err
-	}
-
-	args := getArgs(source, target, fstype, passwd, options)
-	cmd := exec.CommandContext(ctx, m.execPath, args...)
+	cmd := exec.CommandContext(ctx, m.execPath, getArgs(op)...)
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 
@@ -51,20 +42,20 @@ func (m *OssCmdMounter) ExtendedMount(source, target, fstype string, options []s
 	return nil
 }
 
-func getArgs(source, target, fstype, passwdFile string, options []string) []string {
-	if fstype == "ossfs" {
-		if passwdFile != "" {
-			options = append(options, "passwd_file="+passwdFile)
-		}
-		return mount.MakeMountArgs(source, target, "", options)
+func getArgs(op *MountOperation) []string {
+	if op == nil {
+		return nil
 	}
-	// ossfs2
-	args := []string{"mount", target}
-	if passwdFile != "" {
-		args = append(args, []string{"-c", passwdFile}...)
-	}
-	for _, o := range options {
-		args = append(args, fmt.Sprintf("--%s", o))
+	switch op.FsType {
+	case "ossfs":
+		return mount.MakeMountArgs(op.Source, op.Target, "", op.Options)
+	case "ossfs2":
+		args := []string{"mount", op.Target}
+		for _, o := range op.Options {
+			args = append(args, fmt.Sprintf("--%s", o))
+		}
+		return args
+	default:
+		return nil
 	}
-	return args
 }

pkg/mounter/interceptors/alinas_secret.go

@@ -0,0 +1,17 @@
+package interceptors
+
+import (
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
+)
+
+type AlinasSecretInterceptor struct{}
+
+var _ mounter.MountInterceptor = AlinasSecretInterceptor{}
+
+func NewAlinasSecretInterceptor() mounter.MountInterceptor {
+	return AlinasSecretInterceptor{}
+}
+
+func (AlinasSecretInterceptor) BeforeMount(req *mounter.MountOperation) (*mounter.MountOperation, error) {
+	return req, nil
+}

pkg/mounter/interceptors/ossfs_secret.go

@@ -0,0 +1,35 @@
+package interceptors
+
+import (
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+)
+
+type OssfsSecretInterceptor struct {
+	credFileKey string
+}
+
+var _ mounter.MountInterceptor = OssfsSecretInterceptor{}
+
+func NewOssfsSecretInterceptor() mounter.MountInterceptor {
+	return OssfsSecretInterceptor{
+		credFileKey: "passwd_file",
+	}
+}
+
+func NewOssfs2SecretInterceptor() mounter.MountInterceptor {
+	return OssfsSecretInterceptor{
+		credFileKey: "-c",
+	}
+}
+
+func (i OssfsSecretInterceptor) BeforeMount(req *mounter.MountOperation) (*mounter.MountOperation, error) {
+	filePath, err := utils.SaveOssSecretsToFile(req.Secrets, req.FsType)
+	if err != nil {
+		return req, err
+	}
+	if filePath != "" {
+		req.Options = append(req.Options, i.credFileKey+"="+filePath)
+	}
+	return req, nil
+}

pkg/mounter/mounter.go

@@ -1,15 +1,49 @@
 package mounter
 
 import (
+	"context"
+
 	mountutils "k8s.io/mount-utils"
 )
 
-type ExtendedMountParams struct {
+type Mounter interface {
+	mountutils.Interface
+	ExtendedMount(ctx context.Context, op *MountOperation) error
+}
+
+type MountOperation struct {
+	Source      string
+	Target      string
+	FsType      string
+	Options     []string
 	Secrets     map[string]string
 	MetricsPath string
+	VolumeID    string
 }
 
-type Mounter interface {
-	mountutils.Interface
-	ExtendedMount(source, target, fstype string, options []string, params *ExtendedMountParams) error
+type MountInterceptor interface {
+	BeforeMount(op *MountOperation) (*MountOperation, error)
+}
+
+type MountWorkflow struct {
+	Mounter
+	interceptors []MountInterceptor
+}
+
+var _ Mounter = &MountWorkflow{}
+
+func (w *MountWorkflow) ExtendedMount(ctx context.Context, op *MountOperation) (err error) {
+	for _, interceptor := range w.interceptors {
+		if op, err = interceptor.BeforeMount(op); err != nil {
+			return err
+		}
+	}
+	return w.Mounter.ExtendedMount(ctx, op)
+}
+
+func NewForMounter(m Mounter, interceptors ...MountInterceptor) Mounter {
+	return &MountWorkflow{
+		Mounter:      m,
+		interceptors: interceptors,
+	}
 }

pkg/mounter/proxy/protocol.go

@@ -43,4 +43,5 @@ type MountRequest struct {
 	MountFlags  []string          `json:"mountFlags,omitempty"`
 	Secrets     map[string]string `json:"secrets,omitempty"`
 	MetricsPath string            `json:"metricsPath,omitempty"`
+	VolumeID    string            `json:"volumeID,omitempty"`
 }

pkg/mounter/proxy/server/alinas/driver.go

@@ -11,9 +11,11 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/interceptors"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/proxy"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/proxy/server"
-	mounter "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+	mounterutils "github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
@@ -25,11 +27,16 @@ const (
 )
 
 func init() {
-	server.RegisterDriver(&Driver{mounter: mount.New("")})
+	server.RegisterDriver(&Driver{
+		Mounter: mounter.NewForMounter(
+			&extendedMounter{Interface: mount.New("")},
+			interceptors.NewAlinasSecretInterceptor(),
+		),
+	})
 }
 
 type Driver struct {
-	mounter mount.Interface
+	mounter.Mounter
 }
 
 func (h *Driver) Name() string {
@@ -41,15 +48,15 @@ func (h *Driver) Fstypes() []string {
 }
 
 func (h *Driver) Mount(ctx context.Context, req *proxy.MountRequest) error {
-	klog.InfoS("Mounting", "fstype", req.Fstype, "source", req.Source, "target", req.Target, "options", req.Options)
-	options := append(req.Options, "no_start_watchdog")
-	if req.Fstype == fstypeAlinas {
-		// options = append(options, "no_atomic_move", "auto_fallback_nfs")
-		options = append(options, "no_atomic_move")
-		options = addAutoFallbackNFSMountOptions(options)
-	}
-
-	return h.mounter.Mount(req.Source, req.Target, req.Fstype, options)
+	return h.ExtendedMount(ctx, &mounter.MountOperation{
+		Source:      req.Source,
+		Target:      req.Target,
+		FsType:      req.Fstype,
+		Options:     req.Options,
+		Secrets:     req.Secrets,
+		MetricsPath: req.MetricsPath,
+		VolumeID:    req.VolumeID,
+	})
 }
 
 func (h *Driver) Init() {
@@ -79,7 +86,7 @@ func addAutoFallbackNFSMountOptions(mountOptions []string) []string {
 	isEFC := false
 	isVSC := false
 	for _, options := range mountOptions {
-		for _, option := range mounter.SplitMountOptions(options) {
+		for _, option := range mounterutils.SplitMountOptions(options) {
 			if option == "" {
 				continue
 			}
@@ -158,3 +165,20 @@ func copyFile(src, dst string) error {
 
 	return dstFile.Sync()
 }
+
+type extendedMounter struct {
+	mount.Interface
+}
+
+var _ mounter.Mounter = &extendedMounter{}
+
+func (m *extendedMounter) ExtendedMount(ctx context.Context, op *mounter.MountOperation) error {
+	klog.InfoS("Mounting", "fstype", op.FsType, "source", op.Source, "target", op.Target, "options", op.Options)
+	op.Options = append(op.Options, "no_start_watchdog")
+	if op.FsType == fstypeAlinas {
+		// options = append(options, "no_atomic_move", "auto_fallback_nfs")
+		op.Options = append(op.Options, "no_atomic_move")
+		op.Options = addAutoFallbackNFSMountOptions(op.Options)
+	}
+	return m.Mount(op.Source, op.Target, op.FsType, op.Options)
+}

pkg/mounter/proxy_mounter.go

@@ -1,6 +1,7 @@
 package mounter
 
 import (
+	"context"
 	"errors"
 	"fmt"
 
@@ -14,27 +15,25 @@ type ProxyMounter struct {
 	mountutils.Interface
 }
 
+var _ Mounter = &ProxyMounter{}
+
 func NewProxyMounter(socketPath string, inner mountutils.Interface) Mounter {
 	return &ProxyMounter{
 		socketPath: socketPath,
 		Interface:  inner,
 	}
 }
 
-func (m *ProxyMounter) ExtendedMount(source, target, fstype string, options []string, params *ExtendedMountParams) error {
-	// Parameters in ExtendedMountParams are optional
-	if params == nil {
-		params = &ExtendedMountParams{}
-	}
-
+func (m *ProxyMounter) ExtendedMount(ctx context.Context, op *MountOperation) error {
 	dclient := client.NewClient(m.socketPath)
 	resp, err := dclient.Mount(&proxy.MountRequest{
-		Source:      source,
-		Target:      target,
-		Fstype:      fstype,
-		Options:     options,
-		Secrets:     params.Secrets,
-		MetricsPath: params.MetricsPath,
+		Source:      op.Source,
+		Target:      op.Target,
+		Fstype:      op.FsType,
+		Options:     op.Options,
+		Secrets:     op.Secrets,
+		MetricsPath: op.MetricsPath,
+		VolumeID:    op.VolumeID,
 	})
 	if err != nil {
 		return fmt.Errorf("call mounter daemon: %w", err)
@@ -43,7 +42,7 @@ func (m *ProxyMounter) ExtendedMount(source, target, fstype string, options []st
 	if err != nil {
 		return fmt.Errorf("failed to mount: %w", err)
 	}
-	notMnt, err := m.IsLikelyNotMountPoint(target)
+	notMnt, err := m.IsLikelyNotMountPoint(op.Target)
 	if err != nil {
 		return err
 	}
@@ -54,5 +53,10 @@ func (m *ProxyMounter) ExtendedMount(source, target, fstype string, options []st
 }
 
 func (m *ProxyMounter) Mount(source string, target string, fstype string, options []string) error {
-	return m.ExtendedMount(source, target, fstype, options, nil)
+	return m.ExtendedMount(context.Background(), &MountOperation{
+		Source:  source,
+		Target:  target,
+		FsType:  fstype,
+		Options: options,
+	})
 }

pkg/nas/mounter.go

@@ -32,22 +32,17 @@ func (m *NasMounter) Mount(source string, target string, fstype string, options
 	return err
 }
 
-func newNasMounter(agentMode bool) mountutils.Interface {
+func newNasMounter(agentMode bool, socketPath string) mountutils.Interface {
 	inner := mountutils.NewWithoutSystemd("")
 	m := &NasMounter{
 		Interface:     inner,
 		alinasMounter: inner,
 	}
-	if !agentMode {
+	switch {
+	case socketPath != "":
+		m.alinasMounter = mounter.NewProxyMounter(socketPath, inner)
+	case !agentMode: // normal case, use connector mounter to ensure backward compatability
 		m.alinasMounter = mounter.NewConnectorMounter(inner, "")
 	}
 	return m
 }
-
-func newNasMounterWithProxy(socketPath string) mountutils.Interface {
-	inner := mountutils.NewWithoutSystemd("")
-	return &NasMounter{
-		Interface:     inner,
-		alinasMounter: mounter.NewProxyMounter(socketPath, inner),
-	}
-}

pkg/nas/mounter_test.go

@@ -25,7 +25,7 @@ func (m *errorMockMounter) Mount(source string, target string, fstype string, op
 }
 
 func TestNewNasMounter(t *testing.T) {
-	actual := newNasMounter(true)
+	actual := newNasMounter(true, "")
 	assert.NotNil(t, actual)
 }