refactor(mounter): use ossfs secret and monitor interceptors in ossfs mount proxy servers

Author: iltyty

pkg/mounter/cmd_mounter.go

@@ -51,6 +51,7 @@ func getArgs(op *MountOperation) []string {
 		return mount.MakeMountArgs(op.Source, op.Target, "", op.Options)
 	case "ossfs2":
 		args := []string{"mount", op.Target}
+		args = append(args, op.Args...)
 		for _, o := range op.Options {
 			args = append(args, fmt.Sprintf("--%s", o))
 		}

pkg/mounter/interceptors/alinas_secret.go

@@ -17,7 +17,7 @@ func NewAlinasSecretInterceptor() mounter.MountInterceptor {
 	return AlinasSecretInterceptor{}
 }
 
-func (AlinasSecretInterceptor) BeforeMount(op *mounter.MountOperation) (*mounter.MountOperation, error) {
+func (AlinasSecretInterceptor) BeforeMount(op *mounter.MountOperation, _ error) (*mounter.MountOperation, error) {
 	if op == nil || op.Secrets == nil {
 		return op, nil
 	}
@@ -46,3 +46,7 @@ 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

@@ -0,0 +1,41 @@
+package interceptors
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter"
+	"github.com/kubernetes-sigs/alibaba-cloud-csi-driver/pkg/mounter/utils"
+)
+
+type Ossfs2SecretInterceptor struct {
+	passwdFile string
+}
+
+var _ mounter.MountInterceptor = Ossfs2SecretInterceptor{}
+
+func NewOssfs2SecretInterceptor() mounter.MountInterceptor {
+	return Ossfs2SecretInterceptor{}
+}
+
+func (i Ossfs2SecretInterceptor) BeforeMount(req *mounter.MountOperation, _ error) (*mounter.MountOperation, error) {
+	var err error
+	i.passwdFile, err = utils.SaveOssSecretsToFile(req.Secrets, req.FsType)
+	if err != nil {
+		return req, err
+	}
+	if i.passwdFile != "" {
+		req.Args = append(req.Args, []string{"-c", i.passwdFile}...)
+	}
+	return req, nil
+}
+
+func (i Ossfs2SecretInterceptor) AfterMount(op *mounter.MountOperation, err error) error {
+	if i.passwdFile == "" {
+		return nil
+	}
+	if err := os.Remove(i.passwdFile); err != nil {
+		return fmt.Errorf("error removing configuration file: %w, mountpoint=%s, path=%s", err, op.Target, i.passwdFile)
+	}
+	return nil
+}

pkg/mounter/interceptors/ossfs_monitor.go

@@ -0,0 +1,80 @@
+package interceptors
+
+import (
+	"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/mount-utils"
+)
+
+type OssfsMonitorInterceptor struct {
+	raw            mount.Interface
+	monitorManager *server.MountMonitorManager
+}
+
+var _ mounter.MountInterceptor = OssfsMonitorInterceptor{}
+
+func NewOssfsMonitorInterceptor() mounter.MountInterceptor {
+	return OssfsMonitorInterceptor{
+		raw:            mount.NewWithoutSystemd(""),
+		monitorManager: server.NewMountMonitorManager(),
+	}
+}
+
+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)
+	if monitor == nil {
+		return req, fmt.Errorf("failed to get mount monitor for %s, stop monitoring mountpoint status", req.Target)
+	}
+	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)
+	}
+
+	// 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
+	}
+
+	res, ok := op.MountResult.(server.OssfsMountResult)
+	if !ok {
+		return errors.New("failed to parse ossfs mount result")
+	}
+
+	go func() {
+		err := <-res.ExitChan
+		monitor.HandleMountFailureOrExit(err)
+	}()
+
+	if err != nil {
+		return nil
+	}
+
+	monitor.HandleMountSuccess(res.PID)
+	// Start monitoring goroutine (ticker based only)
+	i.monitorManager.StartMonitoring(op.Target)
+	return nil
+}

pkg/mounter/interceptors/ossfs_secret.go

@@ -1,35 +1,41 @@
 package interceptors
 
 import (
+	"fmt"
+	"os"
+
 	"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
+	passwdFile string
 }
 
 var _ mounter.MountInterceptor = OssfsSecretInterceptor{}
 
 func NewOssfsSecretInterceptor() mounter.MountInterceptor {
-	return OssfsSecretInterceptor{
-		credFileKey: "passwd_file",
-	}
-}
-
-func NewOssfs2SecretInterceptor() mounter.MountInterceptor {
-	return OssfsSecretInterceptor{
-		credFileKey: "-c",
-	}
+	return OssfsSecretInterceptor{}
 }
 
-func (i OssfsSecretInterceptor) BeforeMount(req *mounter.MountOperation) (*mounter.MountOperation, error) {
-	filePath, err := utils.SaveOssSecretsToFile(req.Secrets, req.FsType)
+func (i OssfsSecretInterceptor) BeforeMount(req *mounter.MountOperation, _ error) (*mounter.MountOperation, error) {
+	var err error
+	i.passwdFile, err = utils.SaveOssSecretsToFile(req.Secrets, req.FsType)
 	if err != nil {
 		return req, err
 	}
-	if filePath != "" {
-		req.Options = append(req.Options, i.credFileKey+"="+filePath)
+	if i.passwdFile != "" {
+		req.Options = append(req.Options, "passwd_file="+i.passwdFile)
 	}
 	return req, nil
 }
+
+func (i OssfsSecretInterceptor) AfterMount(op *mounter.MountOperation, err error) error {
+	if i.passwdFile == "" {
+		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)
+	}
+	return nil
+}

pkg/mounter/mounter.go

@@ -2,7 +2,9 @@ package mounter
 
 import (
 	"context"
+	"fmt"
 
+	"k8s.io/klog/v2"
 	mountutils "k8s.io/mount-utils"
 )
 
@@ -16,13 +18,17 @@ type MountOperation struct {
 	Target      string
 	FsType      string
 	Options     []string
+	Args        []string
 	Secrets     map[string]string
 	MetricsPath string
 	VolumeID    string
+
+	MountResult any
 }
 
 type MountInterceptor interface {
-	BeforeMount(op *MountOperation) (*MountOperation, error)
+	BeforeMount(op *MountOperation, err error) (*MountOperation, error)
+	AfterMount(op *MountOperation, err error) error
 }
 
 type MountWorkflow struct {
@@ -32,13 +38,29 @@ type MountWorkflow struct {
 
 var _ Mounter = &MountWorkflow{}
 
-func (w *MountWorkflow) ExtendedMount(ctx context.Context, op *MountOperation) (err error) {
+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")
+		}
+	}
+	if err != nil {
+		return fmt.Errorf("error executing BeforeMount interceptor: %w", err)
+	}
+
+	err = w.Mounter.ExtendedMount(ctx, op)
 	for _, interceptor := range w.interceptors {
-		if op, err = interceptor.BeforeMount(op); err != nil {
-			return err
+		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 w.Mounter.ExtendedMount(ctx, op)
+	return err
 }
 
 func NewForMounter(m Mounter, interceptors ...MountInterceptor) Mounter {

pkg/mounter/proxy/server/metrics.go

@@ -3,7 +3,6 @@ package server
 import (
 	"fmt"
 	"os"
-	"os/exec"
 	"path/filepath"
 	"strconv"
 	"sync"
@@ -125,15 +124,15 @@ func (m *MountMonitor) HandleMountFailureOrExit(err error) {
 }
 
 // HandleMountSuccess handles the case when mount operation succeeds
-func (m *MountMonitor) HandleMountSuccess(process *exec.Cmd) {
+func (m *MountMonitor) HandleMountSuccess(pid int) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
 	// Update metrics for mount success
 	m.updateMountPointMetrics(&m.retryCount, nil, nil)
 
+	m.Pid = pid
 	m.State = MonitorStateMonitoring
-	m.Pid = process.Process.Pid
 
 	klog.InfoS("Mount succeeded", "target", m.Target, "pid", m.Pid)
 }

pkg/mounter/proxy/server/mount_result.go

@@ -0,0 +1,6 @@
+package server
+
+type OssfsMountResult struct {
+	PID      int
+	ExitChan chan error
+}