feat: wait for the VFS queue to be purged before unmounting

pkg/rclone/driver.go

@@ -1,6 +1,7 @@
 package rclone
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"os"
@@ -84,7 +85,9 @@ func NewNodeServer(csiDriver *csicommon.CSIDriver, cacheDir string, cacheSize st
 			Interface: mount.New(""),
 			Exec:      utilexec.New(),
 		},
-		RcloneOps: rcloneOps,
+		RcloneOps:         rcloneOps,
+		unpublishContexts: map[string]context.Context{},
+		unpublishMutex:    sync.Mutex{},
 	}, nil
 }
 

pkg/rclone/nodeserver.go

@@ -7,10 +7,12 @@ package rclone
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"gopkg.in/ini.v1"
@@ -22,21 +24,26 @@ import (
 	"github.com/SwissDataScienceCenter/csi-rclone/pkg/kube"
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/fernet/fernet-go"
-	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/utils/mount"
 
 	csicommon "github.com/kubernetes-csi/drivers/pkg/csi-common"
 )
 
-const CSI_ANNOTATION_PREFIX = "csi-rclone.dev"
-const pvcSecretNameAnnotation = CSI_ANNOTATION_PREFIX + "/secretName"
+const (
+	CSI_ANNOTATION_PREFIX   = "csi-rclone.dev"
+	pvcSecretNameAnnotation = CSI_ANNOTATION_PREFIX + "/secretName"
+	// Timeout for the Node Unpublish Operation
+	unpublishTimeout = 90 * time.Second
+)
 
 type nodeServer struct {
 	*csicommon.DefaultNodeServer
-	mounter   *mount.SafeFormatAndMount
-	RcloneOps Operations
+	mounter           *mount.SafeFormatAndMount
+	RcloneOps         Operations
+	unpublishContexts map[string]context.Context
+	unpublishMutex    sync.Mutex
 }
 
 // Mounting Volume (Preparation)
@@ -304,6 +311,11 @@ func extractConfigData(parameters map[string]string) (string, map[string]string)
 
 // Unmounting Volumes
 func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+	// Perform  NodeUnpublishVolumein the background, with only one process per volume ID.
+	// Note that `ctx` corresponds to the gRPC handling context and may get cancelled by the
+	// client timing out. Since we want to wait and unmount even if the client cancels the gRPC call, we perform
+	// the unmount operation in the background to guarantee that we try to call unmount on rclone.
+
 	klog.Infof("NodeUnpublishVolume called with: %s", req)
 	if err := validateUnPublishVolumeRequest(req); err != nil {
 		return nil, err
@@ -314,17 +326,32 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu
 		return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Target Path must be provided")
 	}
 
-	if _, err := ns.RcloneOps.GetVolumeById(ctx, req.GetVolumeId()); err == ErrVolumeNotFound {
-		klog.Warning("VolumeId not found for NodeUnpublishVolume")
-		mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false)
-		return &csi.NodeUnpublishVolumeResponse{}, nil
+	volumeId := req.GetVolumeId()
+	ns.unpublishMutex.Lock()
+	unpublishContext, found := ns.unpublishContexts[volumeId]
+	if !found {
+		unpublishContext = ns.unpublishInBackground(volumeId, targetPath, unpublishTimeout)
+		ns.unpublishContexts[volumeId] = unpublishContext
 	}
+	ns.unpublishMutex.Unlock()
 
-	if err := ns.RcloneOps.Unmount(ctx, req.GetVolumeId(), targetPath); err != nil {
-		klog.Warningf("Unmounting volume failed: %s", err)
+	// Wait for the unpublish operation to finish
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-unpublishContext.Done():
+	}
+
+	// Reset the unmountContext map
+	ns.unpublishMutex.Lock()
+	delete(ns.unpublishContexts, volumeId)
+	ns.unpublishMutex.Unlock()
+
+	err := context.Cause(unpublishContext)
+	if errors.Is(err, context.Canceled) {
+		return &csi.NodeUnpublishVolumeResponse{}, nil
 	}
-	mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false)
-	return &csi.NodeUnpublishVolumeResponse{}, nil
+	return nil, err
 }
 
 func validateUnPublishVolumeRequest(req *csi.NodeUnpublishVolumeRequest) error {
@@ -339,6 +366,39 @@ func validateUnPublishVolumeRequest(req *csi.NodeUnpublishVolumeRequest) error {
 	return nil
 }
 
+func (ns *nodeServer) unpublishInBackground(volumeId string, targetPath string, timeout time.Duration) context.Context {
+	unpublishCtx, unpublishCancel := context.WithCancelCause(context.Background())
+	// Setup context deadline
+	go func() {
+		time.Sleep(timeout)
+		unpublishCancel(context.DeadlineExceeded)
+	}()
+	// Perform unpublish in the background
+	go func() {
+		err := func() error {
+			rcloneVolume, err := ns.RcloneOps.GetVolumeById(unpublishCtx, volumeId)
+			if err != nil {
+				if err == ErrVolumeNotFound {
+					klog.Warningf("VolumeId %s not found for NodeUnpublishVolume", volumeId)
+				} else {
+					klog.Errorf("Could not find rclone volume from volume ID %s", volumeId)
+				}
+				mount.CleanupMountPoint(targetPath, ns.mounter, false)
+				return nil
+			}
+
+			if err := ns.RcloneOps.Unmount(unpublishCtx, rcloneVolume, targetPath); err != nil {
+				klog.Warningf("Unmounting volume %s failed: %s", volumeId, err)
+			}
+			mount.CleanupMountPoint(targetPath, ns.mounter, false)
+			return nil
+		}()
+		unpublishCancel(err)
+	}()
+
+	return unpublishCtx
+}
+
 // Resizing Volume
 func (*nodeServer) NodeExpandVolume(ctx context.Context, req *csi.NodeExpandVolumeRequest) (*csi.NodeExpandVolumeResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method NodeExpandVolume not implemented")

pkg/rclone/rclone.go

@@ -3,18 +3,18 @@ package rclone
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"os"
 	os_exec "os/exec"
-	"syscall"
-
 	"strings"
+	"syscall"
+	"time"
 
-	"golang.org/x/net/context"
 	"gopkg.in/ini.v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -27,11 +27,16 @@ var (
 	ErrVolumeNotFound = errors.New("volume is not found")
 )
 
+const (
+	// Timeout for waiting on the VFS queue to sync all files before unmounting
+	unmountVfsQueueTimeout = time.Minute
+)
+
 type Operations interface {
 	CreateVol(ctx context.Context, volumeName, remote, remotePath, rcloneConfigPath string, pameters map[string]string) error
 	DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rcloneConfigPath string, pameters map[string]string) error
 	Mount(ctx context.Context, rcloneVolume *RcloneVolume, targetPath string, rcloneConfigData string, readOnly bool, pameters map[string]string) error
-	Unmount(ctx context.Context, volumeId string, targetPath string) error
+	Unmount(ctx context.Context, rcloneVolume *RcloneVolume, targetPath string) error
 	GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolume, error)
 	Cleanup() error
 	Run() error
@@ -58,6 +63,18 @@ type MountRequest struct {
 	MountOpt   MountOpt `json:"mountOpt"`
 }
 
+type VfsQueueRequest struct {
+	Fs string `json:"fs,omitempty"`
+}
+
+type VfsQueueResponse struct {
+	Queue []VfsQueue `json:"queue,omitempty"`
+}
+
+type VfsQueue struct {
+	Name string `json:"name"`
+}
+
 // VfsOpt is options for creating the vfs
 //
 // Note that the `Daemon` option has been removed as it is not accepted for rc calls.
@@ -252,7 +269,7 @@ func (r *Rclone) CreateVol(ctx context.Context, volumeName, remote, remotePath,
 	return r.command("mkdir", remote, path, flags)
 }
 
-func (r Rclone) DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rcloneConfigPath string, parameters map[string]string) error {
+func (r *Rclone) DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rcloneConfigPath string, parameters map[string]string) error {
 	flags := make(map[string]string)
 	for key, value := range parameters {
 		flags[key] = value
@@ -261,8 +278,15 @@ func (r Rclone) DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rclon
 	return r.command("purge", rcloneVolume.Remote, rcloneVolume.RemotePath, flags)
 }
 
-func (r Rclone) Unmount(ctx context.Context, volumeId string, targetPath string) error {
-	rcloneVolume := &RcloneVolume{ID: volumeId}
+func (r *Rclone) Unmount(ctx context.Context, rcloneVolume *RcloneVolume, targetPath string) error {
+	configName := rcloneVolume.deploymentName()
+	vfs := fmt.Sprintf("%s:%s", configName, rcloneVolume.RemotePath)
+	queueCtx, queueCtxCancel := context.WithTimeout(ctx, unmountVfsQueueTimeout)
+	defer queueCtxCancel()
+	err := r.waitForVFSQueue(queueCtx, vfs)
+	if err != nil {
+		klog.Infof("Error waiting for VFS: %v", err)
+	}
 
 	klog.Infof("unmounting %s", rcloneVolume.deploymentName())
 	unmountArgs := UnmountRequest{
@@ -281,7 +305,7 @@ func (r Rclone) Unmount(ctx context.Context, volumeId string, targetPath string)
 	if err != nil {
 		return fmt.Errorf("unmounting failed: %w", err)
 	}
-	klog.Infof("deleted mount with volume ID %s at path %s", volumeId, targetPath)
+	klog.Infof("deleted mount with volume ID %s at path %s", rcloneVolume.ID, targetPath)
 
 	configDelete := ConfigDeleteRequest{
 		Name: rcloneVolume.deploymentName(),
@@ -301,12 +325,63 @@ func (r Rclone) Unmount(ctx context.Context, volumeId string, targetPath string)
 		klog.Errorf("deleting config failed: %v", err)
 		return nil
 	}
-	klog.Infof("deleted config for volume ID %s at path %s", volumeId, targetPath)
+	klog.Infof("deleted config for volume ID %s at path %s", rcloneVolume.ID, targetPath)
 
 	return nil
 }
 
-func (r Rclone) GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolume, error) {
+func (r *Rclone) waitForVFSQueue(ctx context.Context, vfs string) error {
+	for {
+		queue, err := r.getVFSQueue(ctx, vfs)
+		if err != nil {
+			return err
+		}
+		if len(queue.Queue) > 0 {
+			files := []string{}
+			for idx := range queue.Queue {
+				files = append(files, queue.Queue[idx].Name)
+			}
+			klog.Infof("Unmounting VFS '%s' still waiting for files: %s", vfs, strings.Join(files, ", "))
+			time.Sleep(time.Second)
+		} else {
+			klog.Infof("VFS '%s' queue is empty", vfs)
+			return nil
+		}
+	}
+}
+
+func (r *Rclone) getVFSQueue(ctx context.Context, vfs string) (queue VfsQueueResponse, err error) {
+	postBody, err := json.Marshal(VfsQueueRequest{Fs: vfs})
+	if err != nil {
+		return queue, fmt.Errorf("getting VFS queue failed: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("http://localhost:%d/vfs/queue", r.port), bytes.NewBuffer(postBody))
+	if err != nil {
+		return queue, fmt.Errorf("getting VFS queue failed: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return queue, fmt.Errorf("getting VFS queue failed: %w", err)
+	}
+	err = checkResponse(res)
+	if err != nil {
+		return queue, fmt.Errorf("getting VFS queue failed: %w", err)
+	}
+	body, err := io.ReadAll(res.Body)
+	defer res.Body.Close()
+	if err != nil {
+		return queue, fmt.Errorf("getting VFS queue failed: %w", err)
+	}
+	var result VfsQueueResponse
+	err = json.Unmarshal(body, &result)
+	if err != nil {
+		return queue, fmt.Errorf("getting VFS queue failed: %w", err)
+	}
+	return result, nil
+}
+
+func (r *Rclone) GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolume, error) {
 	pvs, err := r.kubeClient.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
 	if err != nil {
 		return nil, err