diff --git a/pkg/rclone/driver.go b/pkg/rclone/driver.go index 60fd38b..873e40b 100644 --- a/pkg/rclone/driver.go +++ b/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 } diff --git a/pkg/rclone/nodeserver.go b/pkg/rclone/nodeserver.go index ef03501..14eb4d2 100644 --- a/pkg/rclone/nodeserver.go +++ b/pkg/rclone/nodeserver.go @@ -7,10 +7,12 @@ package rclone import ( "bytes" + "context" "errors" "fmt" "os" "strings" + "sync" "time" "gopkg.in/ini.v1" @@ -22,7 +24,6 @@ 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" @@ -30,13 +31,19 @@ import ( 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") diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index 9b32508..dcc7e64 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -3,6 +3,7 @@ package rclone import ( "bufio" "bytes" + "context" "encoding/json" "errors" "fmt" @@ -10,11 +11,10 @@ import ( "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