refactor(mounter): reimplement interceptors following gRPC pattern

Author: iltyty

pkg/mounter/interceptors/alinas_secret.go

@@ -1,6 +1,7 @@
 package interceptors
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path"
@@ -9,33 +10,29 @@ import (
 	"k8s.io/klog/v2"
 )
 
-type AlinasSecretInterceptor struct{}
+var _ mounter.MountInterceptor = AlinasSecretInterceptor
 
-var _ mounter.MountInterceptor = AlinasSecretInterceptor{}
-
-func NewAlinasSecretInterceptor() mounter.MountInterceptor {
-	return AlinasSecretInterceptor{}
-}
-
-func (AlinasSecretInterceptor) BeforeMount(op *mounter.MountOperation, _ error) (*mounter.MountOperation, error) {
+func AlinasSecretInterceptor(ctx context.Context, op *mounter.MountOperation, handler mounter.MountHandler) error {
 	if op == nil || op.Secrets == nil {
-		return op, nil
+		return handler(ctx, op)
 	}
 
 	tmpDir, err := os.MkdirTemp("", "alinas-")
 	if err != nil {
-		return op, err
+		return err
 	}
 
 	credFileContent := makeCredFileContent(op.Secrets)
 	credFilePath := path.Join(tmpDir, op.VolumeID+".credentials")
 	if err = os.WriteFile(credFilePath, credFileContent, 0o600); err != nil {
-		return op, err
+		os.RemoveAll(tmpDir) // cleanup in case of error
+		return err
 	}
 
 	klog.V(4).InfoS("Created alinas credential file", "path", credFilePath)
 	op.Options = append(op.Options, "ram_config_file="+credFilePath)
-	return op, nil
+
+	return handler(ctx, op)
 }
 
 func makeCredFileContent(secrets map[string]string) []byte {
@@ -46,7 +43,3 @@ func makeCredFileContent(secrets map[string]string) []byte {
 		secrets["akSecret"],
 	)
 }
-
-func (AlinasSecretInterceptor) AfterMount(op *mounter.MountOperation, err error) error {
-	return nil
-}

pkg/mounter/interceptors/ossfs2_secret.go

@@ -1,67 +1,52 @@
 package interceptors
 
 import (
+	"context"
 	"errors"
-	"fmt"
 
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/proxy/server"
+	"k8s.io/klog/v2"
 	"k8s.io/mount-utils"
 )
 
-type OssfsMonitorInterceptor struct {
-	raw            mount.Interface
-	monitorManager *server.MountMonitorManager
-}
+var _ mounter.MountInterceptor = OssfsMonitorInterceptor
 
-var _ mounter.MountInterceptor = OssfsMonitorInterceptor{}
+var (
+	raw            = mount.NewWithoutSystemd("")
+	monitorManager = server.NewMountMonitorManager()
+)
 
-func NewOssfsMonitorInterceptor() mounter.MountInterceptor {
-	return OssfsMonitorInterceptor{
-		raw:            mount.NewWithoutSystemd(""),
-		monitorManager: server.NewMountMonitorManager(),
+func OssfsMonitorInterceptor(ctx context.Context, op *mounter.MountOperation, handler mounter.MountHandler) error {
+	if op == nil || op.MetricsPath == "" {
+		return handler(ctx, op)
 	}
-}
 
-func (i OssfsMonitorInterceptor) BeforeMount(req *mounter.MountOperation, err error) (*mounter.MountOperation, error) {
-	if req.MetricsPath == "" {
-		return req, nil
-	}
 	// Get or create monitor for this target
-	monitor, found := i.monitorManager.GetMountMonitor(req.Target, req.MetricsPath, i.raw, true)
+	monitor, found := monitorManager.GetMountMonitor(op.Target, op.MetricsPath, raw, true)
 	if monitor == nil {
-		return req, fmt.Errorf("failed to get mount monitor for %s, stop monitoring mountpoint status", req.Target)
+		klog.ErrorS(errors.New("failed to get mount monitor"), "stop monitoring mountpoint status", "mountpoint", op.Target)
+		return handler(ctx, op)
 	}
 	if found {
 		monitor.IncreaseMountRetryCount()
 	}
-	if err != nil {
-		monitor.HandleMountFailureOrExit(err)
-	}
-	return req, nil
-}
 
-func (i OssfsMonitorInterceptor) AfterMount(op *mounter.MountOperation, err error) error {
-	if op.MetricsPath == "" {
-		return nil
-	}
-	monitor, _ := i.monitorManager.GetMountMonitor(op.Target, op.MetricsPath, i.raw, true)
-	if monitor == nil {
-		return fmt.Errorf("failed to get mount monitor for %s", op.Target)
-	}
+	err := handler(ctx, op)
 
 	// Immediate process-exit handling during mount attempt
 	// Assume the process exits with no error upon receiving SIGTERM,
 	// and exits with an error in case of unexpected failures.
 	monitor.HandleMountFailureOrExit(err)
 
 	if op.MountResult == nil {
-		return nil
+		return err
 	}
 
 	res, ok := op.MountResult.(server.OssfsMountResult)
 	if !ok {
-		return errors.New("failed to parse ossfs mount result")
+		klog.ErrorS(errors.New("failed to assert ossfs mount result type"), "skipping monitoring of mountpoint", "mountpoint", op.Target)
+		return err
 	}
 
 	go func() {
@@ -70,11 +55,11 @@ func (i OssfsMonitorInterceptor) AfterMount(op *mounter.MountOperation, err erro
 	}()
 
 	if err != nil {
-		return nil
+		return err
 	}
 
 	monitor.HandleMountSuccess(res.PID)
 	// Start monitoring goroutine (ticker based only)
-	i.monitorManager.StartMonitoring(op.Target)
+	monitorManager.StartMonitoring(op.Target)
 	return nil
 }

pkg/mounter/interceptors/ossfs_monitor.go

@@ -1,41 +1,60 @@
 package interceptors
 
 import (
-	"fmt"
+	"context"
+	"errors"
 	"os"
 
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/proxy/server"
 	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+	"k8s.io/klog/v2"
 )
 
-type OssfsSecretInterceptor struct {
-	passwdFile string
-}
+var _ mounter.MountInterceptor = OssfsSecretInterceptor
 
-var _ mounter.MountInterceptor = OssfsSecretInterceptor{}
+func OssfsSecretInterceptor(ctx context.Context, op *mounter.MountOperation, handler mounter.MountHandler) error {
+	return ossfsSecretInterceptor(ctx, op, handler, "ossfs")
+}
 
-func NewOssfsSecretInterceptor() mounter.MountInterceptor {
-	return OssfsSecretInterceptor{}
+func Ossfs2SecretInterceptor(ctx context.Context, op *mounter.MountOperation, handler mounter.MountHandler) error {
+	return ossfsSecretInterceptor(ctx, op, handler, "ossfs2")
 }
 
-func (i OssfsSecretInterceptor) BeforeMount(req *mounter.MountOperation, _ error) (*mounter.MountOperation, error) {
-	var err error
-	i.passwdFile, err = utils.SaveOssSecretsToFile(req.Secrets, req.FsType)
+func ossfsSecretInterceptor(ctx context.Context, op *mounter.MountOperation, handler mounter.MountHandler, fuseType string) error {
+	passwdFile, err := utils.SaveOssSecretsToFile(op.Secrets, op.FsType)
 	if err != nil {
-		return req, err
+		return err
 	}
-	if i.passwdFile != "" {
-		req.Options = append(req.Options, "passwd_file="+i.passwdFile)
+	if passwdFile != "" {
+		if fuseType == "ossfs" {
+			op.Args = append(op.Args, "passwd_file="+passwdFile)
+		} else {
+			op.Args = append(op.Args, []string{"-c", passwdFile}...)
+		}
 	}
-	return req, nil
-}
 
-func (i OssfsSecretInterceptor) AfterMount(op *mounter.MountOperation, err error) error {
-	if i.passwdFile == "" {
+	if err = handler(ctx, op); err != nil {
+		return err
+	}
+
+	if passwdFile == "" || op.MountResult == nil {
 		return nil
 	}
-	if err := os.Remove(i.passwdFile); err != nil {
-		return fmt.Errorf("error removing passwd file: %w, mountpoint=%s, path=%s", err, op.Target, i.passwdFile)
+	result, ok := op.MountResult.(*server.OssfsMountResult)
+	if !ok {
+		klog.ErrorS(
+			errors.New("failed to assert ossfs mount result"),
+			"skipping cleanup of passwd file", "mountpoint", op.Target, "path", passwdFile,
+		)
+		return nil
 	}
+
+	go func() {
+		<-result.ExitChan
+		if err := os.Remove(passwdFile); err != nil {
+			klog.ErrorS(err, "failed to cleanup ossfs passwd file", "mountpoint", op.Target, "path", passwdFile)
+		}
+	}()
 	return nil
 }

pkg/mounter/interceptors/ossfs_secret.go

@@ -2,9 +2,7 @@ package mounter
 
 import (
 	"context"
-	"fmt"
 
-	"k8s.io/klog/v2"
 	mountutils "k8s.io/mount-utils"
 )
 
@@ -26,46 +24,46 @@ type MountOperation struct {
 	MountResult any
 }
 
-type MountInterceptor interface {
-	BeforeMount(op *MountOperation, err error) (*MountOperation, error)
-	AfterMount(op *MountOperation, err error) error
-}
+type MountHandler func(ctx context.Context, op *MountOperation) error
+
+type MountInterceptor func(ctx context.Context, op *MountOperation, handler MountHandler) error
 
 type MountWorkflow struct {
 	Mounter
-	interceptors []MountInterceptor
+	chainedHandler MountHandler
 }
 
 var _ Mounter = &MountWorkflow{}
 
 func (w *MountWorkflow) ExtendedMount(ctx context.Context, op *MountOperation) error {
-	var err error
-	for i, interceptor := range w.interceptors {
-		if op, err = interceptor.BeforeMount(op, err); err != nil && i != len(w.interceptors)-1 {
-			// Log error but continue to the next interceptor, since some interceptors may
-			// want to handle the error, e.g. the OssMonitorInterceptor.
-			// Only log for the intermediate interceptors, otherwise the final error will be printed twice.
-			klog.ErrorS(err, "error executing BeforeMount interceptor")
-		}
+	return w.chainedHandler(ctx, op)
+}
+
+// chainInterceptors creates a chain of interceptors similar to gRPC
+func chainInterceptors(interceptors []MountInterceptor, finalHandler MountHandler) MountHandler {
+	if len(interceptors) == 0 {
+		return finalHandler
 	}
-	if err != nil {
-		return fmt.Errorf("error executing BeforeMount interceptor: %w", err)
+
+	return func(ctx context.Context, op *MountOperation) error {
+		return interceptors[0](ctx, op, getChainHandler(interceptors, 0, finalHandler))
+	}
+}
+
+// getChainHandler creates a handler that chains interceptors recursively
+func getChainHandler(interceptors []MountInterceptor, curr int, finalHandler MountHandler) MountHandler {
+	if curr == len(interceptors)-1 {
+		return finalHandler
 	}
 
-	err = w.Mounter.ExtendedMount(ctx, op)
-	for _, interceptor := range w.interceptors {
-		if afterErr := interceptor.AfterMount(op, err); afterErr != nil {
-			// Log error but continue to the next interceptor.
-			// Do not override the original error from mount operation.
-			klog.ErrorS(afterErr, "error executing AfterMount interceptor")
-		}
+	return func(ctx context.Context, op *MountOperation) error {
+		return interceptors[curr+1](ctx, op, getChainHandler(interceptors, curr+1, finalHandler))
 	}
-	return err
 }
 
 func NewForMounter(m Mounter, interceptors ...MountInterceptor) Mounter {
 	return &MountWorkflow{
-		Mounter:      m,
-		interceptors: interceptors,
+		Mounter:        m,
+		chainedHandler: chainInterceptors(interceptors, m.ExtendedMount),
 	}
 }

pkg/mounter/mounter.go

@@ -30,7 +30,7 @@ func init() {
 	server.RegisterDriver(&Driver{
 		Mounter: mounter.NewForMounter(
 			&extendedMounter{Interface: mount.New("")},
-			interceptors.NewAlinasSecretInterceptor(),
+			interceptors.AlinasSecretInterceptor,
 		),
 	})
 }