sambuc/feat merge restart pr 1

cmd/csi-rclone-plugin/main.go

@@ -48,34 +48,10 @@ func main() {
 		Use:   "run",
 		Short: "Start the CSI driver.",
 	}
-	root.AddCommand(runCmd)
+	nodeCommandLineParameters(runCmd)
+	controllerCommandLineParameters(runCmd)
 
-	runNode := &cobra.Command{
-		Use:   "node",
-		Short: "Start the CSI driver node service - expected to run in a daemonset on every node.",
-		Run: func(cmd *cobra.Command, args []string) {
-			handleNode()
-		},
-	}
-	runNode.PersistentFlags().StringVar(&nodeID, "nodeid", "", "node id")
-	runNode.MarkPersistentFlagRequired("nodeid")
-	runNode.PersistentFlags().StringVar(&endpoint, "endpoint", "", "CSI endpoint")
-	runNode.MarkPersistentFlagRequired("endpoint")
-	runNode.PersistentFlags().StringVar(&cacheDir, "cachedir", "", "cache dir")
-	runNode.PersistentFlags().StringVar(&cacheSize, "cachesize", "", "cache size")
-	runCmd.AddCommand(runNode)
-	runController := &cobra.Command{
-		Use:   "controller",
-		Short: "Start the CSI driver controller.",
-		Run: func(cmd *cobra.Command, args []string) {
-			handleController()
-		},
-	}
-	runController.PersistentFlags().StringVar(&nodeID, "nodeid", "", "node id")
-	runController.MarkPersistentFlagRequired("nodeid")
-	runController.PersistentFlags().StringVar(&endpoint, "endpoint", "", "CSI endpoint")
-	runController.MarkPersistentFlagRequired("endpoint")
-	runCmd.AddCommand(runController)
+	root.AddCommand(runCmd)
 
 	versionCmd := &cobra.Command{
 		Use:   "version",
@@ -123,6 +99,23 @@ func handleNode() {
 	}
 }
 
+func nodeCommandLineParameters(runCmd *cobra.Command) {
+	runNode := &cobra.Command{
+		Use:   "node",
+		Short: "Start the CSI driver node service - expected to run in a daemonset on every node.",
+		Run: func(cmd *cobra.Command, args []string) {
+			handleNode()
+		},
+	}
+	runNode.PersistentFlags().StringVar(&nodeID, "nodeid", "", "node id")
+	runNode.MarkPersistentFlagRequired("nodeid")
+	runNode.PersistentFlags().StringVar(&endpoint, "endpoint", "", "CSI endpoint")
+	runNode.MarkPersistentFlagRequired("endpoint")
+	runNode.PersistentFlags().StringVar(&cacheDir, "cachedir", "", "cache dir")
+	runNode.PersistentFlags().StringVar(&cacheSize, "cachesize", "", "cache size")
+	runCmd.AddCommand(runNode)
+}
+
 func handleController() {
 	d := rclone.NewDriver(nodeID, endpoint)
 	cs := rclone.NewControllerServer(d.CSIDriver)
@@ -134,6 +127,21 @@ func handleController() {
 	}
 }
 
+func controllerCommandLineParameters(runCmd *cobra.Command) {
+	runController := &cobra.Command{
+		Use:   "controller",
+		Short: "Start the CSI driver controller.",
+		Run: func(cmd *cobra.Command, args []string) {
+			handleController()
+		},
+	}
+	runController.PersistentFlags().StringVar(&nodeID, "nodeid", "", "node id")
+	runController.MarkPersistentFlagRequired("nodeid")
+	runController.PersistentFlags().StringVar(&endpoint, "endpoint", "", "CSI endpoint")
+	runController.MarkPersistentFlagRequired("endpoint")
+	runCmd.AddCommand(runController)
+}
+
 // unmountOldVols is used to unmount volumes after a restart on a node
 func unmountOldVols() error {
 	const mountType = "fuse.rclone"

pkg/rclone/driver.go

@@ -1,6 +1,7 @@
 package rclone
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"os"
@@ -78,14 +79,27 @@ func NewNodeServer(csiDriver *csicommon.CSIDriver, cacheDir string, cacheSize st
 	}
 	rcloneOps := NewRclone(kubeClient, rclonePort, cacheDir, cacheSize)
 
-	return &nodeServer{
+	// Use kubelet plugin directory for state persistence
+	stateFile := "/var/lib/kubelet/plugins/csi-rclone/mounted_volumes.json"
+
+	ns := &nodeServer{
 		DefaultNodeServer: csicommon.NewDefaultNodeServer(csiDriver),
 		mounter: &mount.SafeFormatAndMount{
 			Interface: mount.New(""),
 			Exec:      utilexec.New(),
 		},
-		RcloneOps: rcloneOps,
-	}, nil
+		RcloneOps:      rcloneOps,
+		mountedVolumes: make(map[string]*MountedVolume),
+		mutex:          &sync.Mutex{},
+		stateFile:      stateFile,
+	}
+
+	// Load persisted state on startup
+	if err := ns.loadState(); err != nil {
+		klog.Warningf("Failed to load persisted volume state: %v", err)
+	}
+
+	return ns, nil
 }
 
 func NewControllerServer(csiDriver *csicommon.CSIDriver) *controllerServer {
@@ -139,7 +153,13 @@ func (d *Driver) Run() error {
 	)
 	d.server = s
 	if d.ns != nil && d.ns.RcloneOps != nil {
-		return d.ns.RcloneOps.Run()
+		onDaemonReady := func() error {
+			if d.ns != nil {
+				return d.ns.remountTrackedVolumes(context.Background())
+			}
+			return nil
+		}
+		return d.ns.RcloneOps.Run(onDaemonReady)
 	}
 	s.Wait()
 	return nil

pkg/rclone/nodeserver.go

@@ -7,10 +7,13 @@ package rclone
 
 import (
 	"bytes"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"gopkg.in/ini.v1"
@@ -37,6 +40,23 @@ type nodeServer struct {
 	*csicommon.DefaultNodeServer
 	mounter   *mount.SafeFormatAndMount
 	RcloneOps Operations
+
+	// Track mounted volumes for automatic remounting
+	mountedVolumes map[string]*MountedVolume
+	mutex          *sync.Mutex
+	stateFile      string
+}
+
+type MountedVolume struct {
+	VolumeId        string
+	TargetPath      string
+	Remote          string
+	RemotePath      string
+	ConfigData      string
+	ReadOnly        bool
+	Parameters      map[string]string
+	SecretName      string
+	SecretNamespace string
 }
 
 // Mounting Volume (Preparation)
@@ -141,6 +161,10 @@ func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 		}
 		return nil, status.Error(codes.Internal, err.Error())
 	}
+
+	// Track the mounted volume for automatic remounting
+	ns.trackMountedVolume(volumeId, targetPath, remote, remotePath, configData, readOnly, flags, secretName, secretNamespace)
+
 	// err = ns.WaitForMountAvailable(targetPath)
 	// if err != nil {
 	// 	return nil, status.Error(codes.Internal, err.Error())
@@ -323,6 +347,10 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu
 	if err := ns.RcloneOps.Unmount(ctx, req.GetVolumeId(), targetPath); err != nil {
 		klog.Warningf("Unmounting volume failed: %s", err)
 	}
+
+	// Remove the volume from tracking
+	ns.removeTrackedVolume(req.GetVolumeId())
+
 	mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false)
 	return &csi.NodeUnpublishVolumeResponse{}, nil
 }
@@ -344,6 +372,84 @@ func (*nodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolu
 	return nil, status.Errorf(codes.Unimplemented, "method NodeExpandVolume not implemented")
 }
 
+// Track mounted volume for automatic remounting
+func (ns *nodeServer) trackMountedVolume(volumeId, targetPath, remote, remotePath, configData string, readOnly bool, parameters map[string]string, secretName, secretNamespace string) {
+	ns.mutex.Lock()
+
+	ns.mountedVolumes[volumeId] = &MountedVolume{
+		VolumeId:        volumeId,
+		TargetPath:      targetPath,
+		Remote:          remote,
+		RemotePath:      remotePath,
+		ConfigData:      configData,
+		ReadOnly:        readOnly,
+		Parameters:      parameters,
+		SecretName:      secretName,
+		SecretNamespace: secretNamespace,
+	}
+	// Can't use defer here, as persistState also takes the mutex, and would fail as the Unlock happens after it returns
+	ns.mutex.Unlock()
+	klog.Infof("Tracked mounted volume %s at path %s", volumeId, targetPath)
+
+	if err := ns.persistState(); err != nil {
+		klog.Errorf("Failed to persist volume state: %v", err)
+	}
+}
+
+// Remove tracked volume when unmounted
+func (ns *nodeServer) removeTrackedVolume(volumeId string) {
+	ns.mutex.Lock()
+
+	delete(ns.mountedVolumes, volumeId)
+	// Can't use defer here, as persistState also takes the mutex, and would fail as the Unlock happens after it returns
+	ns.mutex.Unlock()
+	klog.Infof("Removed tracked volume %s", volumeId)
+
+	if err := ns.persistState(); err != nil {
+		klog.Errorf("Failed to persist volume state: %v", err)
+	}
+}
+
+// Automatically remount all tracked volumes after daemon restart
+func (ns *nodeServer) remountTrackedVolumes(ctx context.Context) error {
+	ns.mutex.Lock()
+	defer ns.mutex.Unlock()
+
+	if len(ns.mountedVolumes) == 0 {
+		klog.Info("No tracked volumes to remount")
+		return nil
+	}
+
+	klog.Infof("Remounting %d tracked volumes", len(ns.mountedVolumes))
+
+	for volumeId, mv := range ns.mountedVolumes {
+		klog.Infof("Remounting volume %s to %s", volumeId, mv.TargetPath)
+
+		// Create the mount directory if it doesn't exist
+		if err := os.MkdirAll(mv.TargetPath, 0750); err != nil {
+			klog.Errorf("Failed to create mount directory %s: %v", mv.TargetPath, err)
+			continue
+		}
+
+		// Remount the volume
+		rcloneVol := &RcloneVolume{
+			ID:         mv.VolumeId,
+			Remote:     mv.Remote,
+			RemotePath: mv.RemotePath,
+		}
+
+		err := ns.RcloneOps.Mount(ctx, rcloneVol, mv.TargetPath, mv.ConfigData, mv.ReadOnly, mv.Parameters)
+		if err != nil {
+			klog.Errorf("Failed to remount volume %s: %v", volumeId, err)
+			// Don't return error here - continue with other volumes
+		} else {
+			klog.Infof("Successfully remounted volume %s", volumeId)
+		}
+	}
+
+	return nil
+}
+
 func (ns *nodeServer) WaitForMountAvailable(mountpoint string) error {
 	for {
 		select {
@@ -357,3 +463,57 @@ func (ns *nodeServer) WaitForMountAvailable(mountpoint string) error {
 		}
 	}
 }
+
+// Persist volume state to disk
+func (ns *nodeServer) persistState() error {
+	if ns.stateFile == "" {
+		return nil
+	}
+
+	if err := os.MkdirAll(filepath.Dir(ns.stateFile), 0755); err != nil {
+		return fmt.Errorf("failed to create state directory: %v", err)
+	}
+
+	ns.mutex.Lock()
+	defer ns.mutex.Unlock()
+
+	data, err := json.Marshal(ns.mountedVolumes)
+	if err != nil {
+		return fmt.Errorf("failed to marshal volume state: %v", err)
+	}
+
+	if err := os.WriteFile(ns.stateFile, data, 0600); err != nil {
+		return fmt.Errorf("failed to write state file: %v", err)
+	}
+
+	klog.Infof("Persisted volume state to %s", ns.stateFile)
+	return nil
+}
+
+// Load volume state from disk
+func (ns *nodeServer) loadState() error {
+	if ns.stateFile == "" {
+		return nil
+	}
+
+	ns.mutex.Lock()
+	defer ns.mutex.Unlock()
+
+	data, err := os.ReadFile(ns.stateFile)
+	if err != nil {
+		if os.IsNotExist(err) {
+			klog.Info("No persisted volume state found, starting fresh")
+			return nil
+		}
+		return fmt.Errorf("failed to read state file: %v", err)
+	}
+
+	var volumes map[string]*MountedVolume
+	if err := json.Unmarshal(data, &volumes); err != nil {
+		return fmt.Errorf("failed to unmarshal volume state: %v", err)
+	}
+
+	ns.mountedVolumes = volumes
+	klog.Infof("Loaded %d tracked volumes from %s", len(ns.mountedVolumes), ns.stateFile)
+	return nil
+}

pkg/rclone/rclone.go

@@ -34,7 +34,7 @@ type Operations interface {
 	Unmount(ctx context.Context, volumeId string, targetPath string) error
 	GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolume, error)
 	Cleanup() error
-	Run() error
+	Run(onDaemonReady func() error) error
 }
 
 type Rclone struct {
@@ -472,11 +472,16 @@ func (r *Rclone) start_daemon() error {
 	return nil
 }
 
-func (r *Rclone) Run() error {
+func (r *Rclone) Run(onDaemonReady func() error) error {
 	err := r.start_daemon()
 	if err != nil {
 		return err
 	}
+	if onDaemonReady != nil {
+		if err := onDaemonReady(); err != nil {
+			klog.Warningf("Error in onDaemonReady callback: %v", err)
+		}
+	}
 	// blocks until the rclone daemon is stopped
 	return r.daemonCmd.Wait()
 }