From 8528edb578857c61a5ba6a077e4e5edf52da5450 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Fri, 7 Aug 2026 11:42:59 +0200 Subject: [PATCH 01/15] feat: wait for the VFS queue to be purged before unmounting --- go.mod | 5 -- go.sum | 5 -- pkg/rclone/rclone.go | 146 +++++++++++++++++++++++++++++-------------- 3 files changed, 99 insertions(+), 57 deletions(-) diff --git a/go.mod b/go.mod index f27cc4f..fa02cc0 100644 --- a/go.mod +++ b/go.mod @@ -26,8 +26,6 @@ require ( ) require ( - github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect - github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect github.com/beorn7/perks v1.0.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/go-logr/logr v1.4.2 // indirect @@ -42,7 +40,6 @@ require ( github.com/imdario/mergo v0.3.7 // indirect github.com/inconshreveable/mousetrap v1.0.0 // indirect github.com/json-iterator/go v1.1.10 // indirect - github.com/konsorten/go-windows-terminal-sequences v1.0.1 // indirect github.com/kubernetes-csi/csi-lib-utils v0.3.1 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect github.com/moby/sys/mountinfo v0.7.2 // indirect @@ -52,7 +49,6 @@ require ( github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 // indirect github.com/prometheus/common v0.4.0 // indirect github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084 // indirect - github.com/sirupsen/logrus v1.2.0 // indirect github.com/spf13/pflag v1.0.5 // indirect golang.org/x/crypto v0.14.0 // indirect golang.org/x/oauth2 v0.10.0 // indirect @@ -64,7 +60,6 @@ require ( google.golang.org/appengine v1.6.7 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5 // indirect google.golang.org/protobuf v1.31.0 // indirect - gopkg.in/alecthomas/kingpin.v2 v2.2.6 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 99033f0..6c1f273 100644 --- a/go.sum +++ b/go.sum @@ -37,9 +37,7 @@ github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb0 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0= github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE= -github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= -github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o= github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY= @@ -212,7 +210,6 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q= github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= @@ -300,7 +297,6 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= -github.com/sirupsen/logrus v1.2.0 h1:juTguoYk5qI21pwyTXY3B3Y5cOTH3ZUyZCg1v/mihuo= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= @@ -570,7 +566,6 @@ google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp0 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= -gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index 9b32508..5347405 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,11 @@ import ( "net/http" "os" os_exec "os/exec" - "syscall" - "strings" + "sync" + "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" @@ -44,6 +45,10 @@ type Rclone struct { port int cacheDir string cacheSize string + + // TODO: unmount queue + unmountContexts map[string]context.Context + unmountMutex sync.Mutex } type RcloneVolume struct { @@ -252,7 +257,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,52 +266,97 @@ 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} - - klog.Infof("unmounting %s", rcloneVolume.deploymentName()) - unmountArgs := UnmountRequest{ - MountPoint: targetPath, +func (r *Rclone) Unmount(ctx context.Context, volumeId string, targetPath string) error { + // Unmount in the background, with only one process per volume ID + r.unmountMutex.Lock() + unmountContext, found := r.unmountContexts[volumeId] + if !found { + unmountContext = r.unmountInBackground(volumeId, targetPath, time.Minute) + r.unmountContexts[volumeId] = unmountContext } - postBody, err := json.Marshal(unmountArgs) - if err != nil { - return fmt.Errorf("unmounting failed: couldn't create request body: %s", err) - } - requestBody := bytes.NewBuffer(postBody) - resp, err := http.Post(fmt.Sprintf("http://localhost:%d/mount/unmount", r.port), "application/json", requestBody) - if err != nil { - return fmt.Errorf("unmounting failed: couldn't send HTTP request: %w", err) - } - err = checkResponse(resp) - if err != nil { - return fmt.Errorf("unmounting failed: %w", err) - } - klog.Infof("deleted mount with volume ID %s at path %s", volumeId, targetPath) + r.unmountMutex.Unlock() - configDelete := ConfigDeleteRequest{ - Name: rcloneVolume.deploymentName(), - } - postBody, err = json.Marshal(configDelete) - if err != nil { - return fmt.Errorf("deleting config failed: couldn't create request body: %s", err) - } - requestBody = bytes.NewBuffer(postBody) - resp, err = http.Post(fmt.Sprintf("http://localhost:%d/config/delete", r.port), "application/json", requestBody) - if err != nil { - klog.Errorf("deleting config failed: couldn't send HTTP request: %v", err) - return nil + // Wait for the unmounting to finish + select { + case <-ctx.Done(): + return ctx.Err() + case <-unmountContext.Done(): } - err = checkResponse(resp) - if err != nil { - klog.Errorf("deleting config failed: %v", err) + + // Reset the unmountContext map + r.unmountMutex.Lock() + delete(r.unmountContexts, volumeId) + r.unmountMutex.Unlock() + + err := context.Cause(unmountContext) + if errors.Is(err, context.Canceled) { return nil } - klog.Infof("deleted config for volume ID %s at path %s", volumeId, targetPath) + return err +} - return nil +func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout time.Duration) context.Context { + unmountCtx, unmountCancel := context.WithCancelCause(context.Background()) + // Setup context deadline + go func() { + time.Sleep(timeout) + unmountCancel(context.DeadlineExceeded) + }() + // Perform unmounting in the background + go func() { + err := func() error { + rcloneVolume := &RcloneVolume{ID: volumeId} + + // TODO: wait for the VFS queue to be drained + + klog.Infof("unmounting %s", rcloneVolume.deploymentName()) + unmountArgs := UnmountRequest{ + MountPoint: targetPath, + } + postBody, err := json.Marshal(unmountArgs) + if err != nil { + return fmt.Errorf("unmounting failed: couldn't create request body: %s", err) + } + requestBody := bytes.NewBuffer(postBody) + resp, err := http.Post(fmt.Sprintf("http://localhost:%d/mount/unmount", r.port), "application/json", requestBody) + if err != nil { + return fmt.Errorf("unmounting failed: couldn't send HTTP request: %w", err) + } + err = checkResponse(resp) + if err != nil { + return fmt.Errorf("unmounting failed: %w", err) + } + klog.Infof("deleted mount with volume ID %s at path %s", volumeId, targetPath) + + configDelete := ConfigDeleteRequest{ + Name: rcloneVolume.deploymentName(), + } + postBody, err = json.Marshal(configDelete) + if err != nil { + return fmt.Errorf("deleting config failed: couldn't create request body: %s", err) + } + requestBody = bytes.NewBuffer(postBody) + resp, err = http.Post(fmt.Sprintf("http://localhost:%d/config/delete", r.port), "application/json", requestBody) + if err != nil { + klog.Errorf("deleting config failed: couldn't send HTTP request: %v", err) + return nil + } + err = checkResponse(resp) + if err != nil { + klog.Errorf("deleting config failed: %v", err) + return nil + } + klog.Infof("deleted config for volume ID %s at path %s", volumeId, targetPath) + + return nil + }() + unmountCancel(err) + }() + + return unmountCtx } -func (r Rclone) GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolume, error) { +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 @@ -355,11 +405,13 @@ func (r Rclone) GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolu func NewRclone(kubeClient *kubernetes.Clientset, port int, cacheDir string, cacheSize string) Operations { rclone := &Rclone{ - execute: exec.New(), - kubeClient: kubeClient, - port: port, - cacheDir: cacheDir, - cacheSize: cacheSize, + execute: exec.New(), + kubeClient: kubeClient, + port: port, + cacheDir: cacheDir, + cacheSize: cacheSize, + unmountContexts: map[string]context.Context{}, + unmountMutex: sync.Mutex{}, } return rclone } From add99db082d40eb5f00b7dbd1e2942f7db1bcbe1 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Thu, 12 Jun 2025 09:05:27 +0200 Subject: [PATCH 02/15] build: add action to build the container image (dev) --- .github/workflows/build.yaml | 71 ++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 .github/workflows/build.yaml diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml new file mode 100644 index 0000000..3fe5df7 --- /dev/null +++ b/.github/workflows/build.yaml @@ -0,0 +1,71 @@ +name: Build dev version + +on: + push: + workflow_dispatch: + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + CHART_NAME: ${{ github.repository }}/helm-chart + +defaults: + run: + shell: bash + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + build-image: + runs-on: ubuntu-24.04 + outputs: + image: ${{ steps.docker_image.outputs.image }} + image_repository: ${{ steps.docker_image.outputs.image_repository }} + image_tag: ${{ steps.docker_image.outputs.image_tag }} + permissions: + contents: read + packages: write + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Docker image metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: type=sha + - name: Extract Docker image name + id: docker_image + env: + IMAGE_TAGS: ${{ steps.meta.outputs.tags }} + run: | + IMAGE=$(echo "$IMAGE_TAGS" | cut -d" " -f1) + IMAGE_REPOSITORY=$(echo "$IMAGE" | cut -d":" -f1) + IMAGE_TAG=$(echo "$IMAGE" | cut -d":" -f2) + echo "image=$IMAGE" >> "$GITHUB_OUTPUT" + echo "image_repository=$IMAGE_REPOSITORY" >> "$GITHUB_OUTPUT" + echo "image_tag=$IMAGE_TAG" >> "$GITHUB_OUTPUT" + - name: Set up Docker buildx + uses: docker/setup-buildx-action@v3 + - name: Set up Docker + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - name: Build and push Docker image + uses: docker/build-push-action@v6 + with: + context: . + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache + cache-to: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache,mode=max + +# TODO: add job to build and push the helm chart if needed (manual trigger) From c7488f6ec2a77daa88db9d232d2f02630817f885 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Fri, 7 Aug 2026 13:37:36 +0200 Subject: [PATCH 03/15] List VFSes --- pkg/rclone/rclone.go | 60 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index 5347405..a386875 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -63,6 +63,22 @@ type MountRequest struct { MountOpt MountOpt `json:"mountOpt"` } +type VfsListResponse struct { + VfsList []string `json:"vfses,omitempty"` +} + +type VfsStatsRequest struct { + Fs string `json:"fs,omitempty"` +} + +type VfsStatsResponse 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. @@ -308,6 +324,11 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout rcloneVolume := &RcloneVolume{ID: volumeId} // TODO: wait for the VFS queue to be drained + vfs, err := r.getVfs(unmountCtx, *rcloneVolume) + if err != nil { + klog.Errorf("Error listing VFSes: %v", err) + } + klog.Infof("Got VFS: '%s'", vfs) klog.Infof("unmounting %s", rcloneVolume.deploymentName()) unmountArgs := UnmountRequest{ @@ -356,6 +377,45 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout return unmountCtx } +func (r *Rclone) getVfs(ctx context.Context, rcloneVolume RcloneVolume) (vfsName string, err error) { + req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("http://localhost:%d/vfs/list", r.port), nil) + if err != nil { + return "", fmt.Errorf("Listing VFSes failed: %w", err) + } + res, err := http.DefaultClient.Do(req) + if err != nil { + return "", fmt.Errorf("Listing VFSes failed: %w", err) + } + err = checkResponse(res) + if err != nil { + return "", fmt.Errorf("Listing VFSes failed: %w", err) + } + body, err := io.ReadAll(res.Body) + defer res.Body.Close() + if err != nil { + return "", fmt.Errorf("Listing VFSes failed: %w", err) + } + var result VfsListResponse + err = json.Unmarshal(body, &result) + if err != nil { + return "", fmt.Errorf("Listing VFSes failed: %w", err) + } + configName := rcloneVolume.deploymentName() + searchString := fmt.Sprintf("%s:", configName) + for idx := range result.VfsList { + vfs := result.VfsList[idx] + if strings.Contains(vfs, searchString) { + return vfs, nil + } + } + return "", fmt.Errorf("Could not find the VFS for volume %s", rcloneVolume.ID) +} + +// func waitForVFSQueue(rcloneVolume RcloneVolume) { +// configName := rcloneVolume.deploymentName() + +// } + func (r *Rclone) GetVolumeById(ctx context.Context, volumeId string) (*RcloneVolume, error) { pvs, err := r.kubeClient.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{}) if err != nil { From 60a529a74eebcc8d2c98548d3985878e04440d82 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Mon, 10 Aug 2026 11:13:51 +0200 Subject: [PATCH 04/15] try to wait for the VFS queue --- pkg/rclone/rclone.go | 61 ++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 56 insertions(+), 5 deletions(-) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index a386875..284a86b 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -67,11 +67,11 @@ type VfsListResponse struct { VfsList []string `json:"vfses,omitempty"` } -type VfsStatsRequest struct { +type VfsQueueRequest struct { Fs string `json:"fs,omitempty"` } -type VfsStatsResponse struct { +type VfsQueueResponse struct { Queue []VfsQueue `json:"queue,omitempty"` } @@ -329,6 +329,12 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout klog.Errorf("Error listing VFSes: %v", err) } klog.Infof("Got VFS: '%s'", vfs) + if vfs != "" { + err = r.waitForVFSQueue(unmountCtx, vfs) + if err != nil { + klog.Infof("Error waiting for VFS: %v", err) + } + } klog.Infof("unmounting %s", rcloneVolume.deploymentName()) unmountArgs := UnmountRequest{ @@ -411,10 +417,55 @@ func (r *Rclone) getVfs(ctx context.Context, rcloneVolume RcloneVolume) (vfsName return "", fmt.Errorf("Could not find the VFS for volume %s", rcloneVolume.ID) } -// func waitForVFSQueue(rcloneVolume RcloneVolume) { -// configName := rcloneVolume.deploymentName() +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 { + 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{}) From a2d1ce0b7f1acf0629a886ae3222d06413a99fa3 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Mon, 10 Aug 2026 11:43:51 +0200 Subject: [PATCH 05/15] do not guess vfs --- pkg/rclone/nodeserver.go | 5 ++- pkg/rclone/rclone.go | 92 +++++++++++++++++++--------------------- 2 files changed, 47 insertions(+), 50 deletions(-) diff --git a/pkg/rclone/nodeserver.go b/pkg/rclone/nodeserver.go index ef03501..ad7d97a 100644 --- a/pkg/rclone/nodeserver.go +++ b/pkg/rclone/nodeserver.go @@ -314,13 +314,14 @@ 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 { + rcloneVolume, err := ns.RcloneOps.GetVolumeById(ctx, req.GetVolumeId()) + if err == ErrVolumeNotFound { klog.Warning("VolumeId not found for NodeUnpublishVolume") mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false) return &csi.NodeUnpublishVolumeResponse{}, nil } - if err := ns.RcloneOps.Unmount(ctx, req.GetVolumeId(), targetPath); err != nil { + if err := ns.RcloneOps.Unmount(ctx, rcloneVolume, targetPath); err != nil { klog.Warningf("Unmounting volume failed: %s", err) } mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index 284a86b..c176066 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -32,7 +32,7 @@ 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 @@ -282,12 +282,13 @@ func (r *Rclone) DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rclo return r.command("purge", rcloneVolume.Remote, rcloneVolume.RemotePath, flags) } -func (r *Rclone) Unmount(ctx context.Context, volumeId string, targetPath string) error { +func (r *Rclone) Unmount(ctx context.Context, rcloneVolume *RcloneVolume, targetPath string) error { // Unmount in the background, with only one process per volume ID + volumeId := rcloneVolume.ID r.unmountMutex.Lock() unmountContext, found := r.unmountContexts[volumeId] if !found { - unmountContext = r.unmountInBackground(volumeId, targetPath, time.Minute) + unmountContext = r.unmountInBackground(rcloneVolume, targetPath, time.Minute) r.unmountContexts[volumeId] = unmountContext } r.unmountMutex.Unlock() @@ -311,7 +312,7 @@ func (r *Rclone) Unmount(ctx context.Context, volumeId string, targetPath string return err } -func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout time.Duration) context.Context { +func (r *Rclone) unmountInBackground(rcloneVolume *RcloneVolume, targetPath string, timeout time.Duration) context.Context { unmountCtx, unmountCancel := context.WithCancelCause(context.Background()) // Setup context deadline go func() { @@ -321,16 +322,10 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout // Perform unmounting in the background go func() { err := func() error { - rcloneVolume := &RcloneVolume{ID: volumeId} - - // TODO: wait for the VFS queue to be drained - vfs, err := r.getVfs(unmountCtx, *rcloneVolume) - if err != nil { - klog.Errorf("Error listing VFSes: %v", err) - } - klog.Infof("Got VFS: '%s'", vfs) + configName := rcloneVolume.deploymentName() + vfs := fmt.Sprintf("%s:%s", configName, rcloneVolume.RemotePath) if vfs != "" { - err = r.waitForVFSQueue(unmountCtx, vfs) + err := r.waitForVFSQueue(unmountCtx, vfs) if err != nil { klog.Infof("Error waiting for VFS: %v", err) } @@ -353,7 +348,7 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout 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(), @@ -373,7 +368,7 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout 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 }() @@ -383,39 +378,39 @@ func (r *Rclone) unmountInBackground(volumeId string, targetPath string, timeout return unmountCtx } -func (r *Rclone) getVfs(ctx context.Context, rcloneVolume RcloneVolume) (vfsName string, err error) { - req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("http://localhost:%d/vfs/list", r.port), nil) - if err != nil { - return "", fmt.Errorf("Listing VFSes failed: %w", err) - } - res, err := http.DefaultClient.Do(req) - if err != nil { - return "", fmt.Errorf("Listing VFSes failed: %w", err) - } - err = checkResponse(res) - if err != nil { - return "", fmt.Errorf("Listing VFSes failed: %w", err) - } - body, err := io.ReadAll(res.Body) - defer res.Body.Close() - if err != nil { - return "", fmt.Errorf("Listing VFSes failed: %w", err) - } - var result VfsListResponse - err = json.Unmarshal(body, &result) - if err != nil { - return "", fmt.Errorf("Listing VFSes failed: %w", err) - } - configName := rcloneVolume.deploymentName() - searchString := fmt.Sprintf("%s:", configName) - for idx := range result.VfsList { - vfs := result.VfsList[idx] - if strings.Contains(vfs, searchString) { - return vfs, nil - } - } - return "", fmt.Errorf("Could not find the VFS for volume %s", rcloneVolume.ID) -} +// func (r *Rclone) getVfs(ctx context.Context, rcloneVolume RcloneVolume) (vfsName string, err error) { +// req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("http://localhost:%d/vfs/list", r.port), nil) +// if err != nil { +// return "", fmt.Errorf("Listing VFSes failed: %w", err) +// } +// res, err := http.DefaultClient.Do(req) +// if err != nil { +// return "", fmt.Errorf("Listing VFSes failed: %w", err) +// } +// err = checkResponse(res) +// if err != nil { +// return "", fmt.Errorf("Listing VFSes failed: %w", err) +// } +// body, err := io.ReadAll(res.Body) +// defer res.Body.Close() +// if err != nil { +// return "", fmt.Errorf("Listing VFSes failed: %w", err) +// } +// var result VfsListResponse +// err = json.Unmarshal(body, &result) +// if err != nil { +// return "", fmt.Errorf("Listing VFSes failed: %w", err) +// } +// configName := rcloneVolume.deploymentName() +// searchString := fmt.Sprintf("%s:", configName) +// for idx := range result.VfsList { +// vfs := result.VfsList[idx] +// if strings.Contains(vfs, searchString) { +// return vfs, nil +// } +// } +// return "", fmt.Errorf("Could not find the VFS for volume %s", rcloneVolume.ID) +// } func (r *Rclone) waitForVFSQueue(ctx context.Context, vfs string) error { for { @@ -431,6 +426,7 @@ func (r *Rclone) waitForVFSQueue(ctx context.Context, vfs string) error { 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 } } From 23ec2e89802b3fd4a5cd46c44bf211d523de3e47 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Mon, 10 Aug 2026 13:26:37 +0200 Subject: [PATCH 06/15] cleanup --- pkg/rclone/rclone.go | 55 ++++++++++---------------------------------- 1 file changed, 12 insertions(+), 43 deletions(-) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index c176066..895e4c6 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -28,6 +28,11 @@ var ( ErrVolumeNotFound = errors.New("volume is not found") ) +var ( + // Timeout for unmounting a volume: wait for the VFS queue to sync files until the deadline + unmountTimeout = 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 @@ -39,14 +44,12 @@ type Operations interface { } type Rclone struct { - execute exec.Interface - kubeClient *kubernetes.Clientset - daemonCmd *os_exec.Cmd - port int - cacheDir string - cacheSize string - - // TODO: unmount queue + execute exec.Interface + kubeClient *kubernetes.Clientset + daemonCmd *os_exec.Cmd + port int + cacheDir string + cacheSize string unmountContexts map[string]context.Context unmountMutex sync.Mutex } @@ -288,7 +291,7 @@ func (r *Rclone) Unmount(ctx context.Context, rcloneVolume *RcloneVolume, target r.unmountMutex.Lock() unmountContext, found := r.unmountContexts[volumeId] if !found { - unmountContext = r.unmountInBackground(rcloneVolume, targetPath, time.Minute) + unmountContext = r.unmountInBackground(rcloneVolume, targetPath, unmountTimeout) r.unmountContexts[volumeId] = unmountContext } r.unmountMutex.Unlock() @@ -378,40 +381,6 @@ func (r *Rclone) unmountInBackground(rcloneVolume *RcloneVolume, targetPath stri return unmountCtx } -// func (r *Rclone) getVfs(ctx context.Context, rcloneVolume RcloneVolume) (vfsName string, err error) { -// req, err := http.NewRequestWithContext(ctx, "POST", fmt.Sprintf("http://localhost:%d/vfs/list", r.port), nil) -// if err != nil { -// return "", fmt.Errorf("Listing VFSes failed: %w", err) -// } -// res, err := http.DefaultClient.Do(req) -// if err != nil { -// return "", fmt.Errorf("Listing VFSes failed: %w", err) -// } -// err = checkResponse(res) -// if err != nil { -// return "", fmt.Errorf("Listing VFSes failed: %w", err) -// } -// body, err := io.ReadAll(res.Body) -// defer res.Body.Close() -// if err != nil { -// return "", fmt.Errorf("Listing VFSes failed: %w", err) -// } -// var result VfsListResponse -// err = json.Unmarshal(body, &result) -// if err != nil { -// return "", fmt.Errorf("Listing VFSes failed: %w", err) -// } -// configName := rcloneVolume.deploymentName() -// searchString := fmt.Sprintf("%s:", configName) -// for idx := range result.VfsList { -// vfs := result.VfsList[idx] -// if strings.Contains(vfs, searchString) { -// return vfs, nil -// } -// } -// return "", fmt.Errorf("Could not find the VFS for volume %s", rcloneVolume.ID) -// } - func (r *Rclone) waitForVFSQueue(ctx context.Context, vfs string) error { for { queue, err := r.getVFSQueue(ctx, vfs) From c54ccffc118ad1ed328ce9ce3859cb63c185cd88 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Mon, 10 Aug 2026 13:28:55 +0200 Subject: [PATCH 07/15] cleanup --- pkg/rclone/rclone.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index 895e4c6..d120350 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -66,10 +66,6 @@ type MountRequest struct { MountOpt MountOpt `json:"mountOpt"` } -type VfsListResponse struct { - VfsList []string `json:"vfses,omitempty"` -} - type VfsQueueRequest struct { Fs string `json:"fs,omitempty"` } From 9ac3ff1478ae7585e2f4fd56e98b1504e5b71c1e Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Mon, 10 Aug 2026 13:38:30 +0200 Subject: [PATCH 08/15] cleanup --- .github/workflows/build.yaml | 71 ------------------------------------ go.mod | 5 +++ go.sum | 5 +++ 3 files changed, 10 insertions(+), 71 deletions(-) delete mode 100644 .github/workflows/build.yaml diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml deleted file mode 100644 index 3fe5df7..0000000 --- a/.github/workflows/build.yaml +++ /dev/null @@ -1,71 +0,0 @@ -name: Build dev version - -on: - push: - workflow_dispatch: - -env: - REGISTRY: ghcr.io - IMAGE_NAME: ${{ github.repository }} - CHART_NAME: ${{ github.repository }}/helm-chart - -defaults: - run: - shell: bash - -concurrency: - group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} - cancel-in-progress: true - -permissions: - contents: read - -jobs: - build-image: - runs-on: ubuntu-24.04 - outputs: - image: ${{ steps.docker_image.outputs.image }} - image_repository: ${{ steps.docker_image.outputs.image_repository }} - image_tag: ${{ steps.docker_image.outputs.image_tag }} - permissions: - contents: read - packages: write - steps: - - name: Checkout - uses: actions/checkout@v4 - - name: Docker image metadata - id: meta - uses: docker/metadata-action@v5 - with: - images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} - tags: type=sha - - name: Extract Docker image name - id: docker_image - env: - IMAGE_TAGS: ${{ steps.meta.outputs.tags }} - run: | - IMAGE=$(echo "$IMAGE_TAGS" | cut -d" " -f1) - IMAGE_REPOSITORY=$(echo "$IMAGE" | cut -d":" -f1) - IMAGE_TAG=$(echo "$IMAGE" | cut -d":" -f2) - echo "image=$IMAGE" >> "$GITHUB_OUTPUT" - echo "image_repository=$IMAGE_REPOSITORY" >> "$GITHUB_OUTPUT" - echo "image_tag=$IMAGE_TAG" >> "$GITHUB_OUTPUT" - - name: Set up Docker buildx - uses: docker/setup-buildx-action@v3 - - name: Set up Docker - uses: docker/login-action@v3 - with: - registry: ${{ env.REGISTRY }} - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - name: Build and push Docker image - uses: docker/build-push-action@v6 - with: - context: . - push: true - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache - cache-to: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache,mode=max - -# TODO: add job to build and push the helm chart if needed (manual trigger) diff --git a/go.mod b/go.mod index fa02cc0..f27cc4f 100644 --- a/go.mod +++ b/go.mod @@ -26,6 +26,8 @@ require ( ) require ( + github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc // indirect + github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf // indirect github.com/beorn7/perks v1.0.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/go-logr/logr v1.4.2 // indirect @@ -40,6 +42,7 @@ require ( github.com/imdario/mergo v0.3.7 // indirect github.com/inconshreveable/mousetrap v1.0.0 // indirect github.com/json-iterator/go v1.1.10 // indirect + github.com/konsorten/go-windows-terminal-sequences v1.0.1 // indirect github.com/kubernetes-csi/csi-lib-utils v0.3.1 // indirect github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect github.com/moby/sys/mountinfo v0.7.2 // indirect @@ -49,6 +52,7 @@ require ( github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4 // indirect github.com/prometheus/common v0.4.0 // indirect github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084 // indirect + github.com/sirupsen/logrus v1.2.0 // indirect github.com/spf13/pflag v1.0.5 // indirect golang.org/x/crypto v0.14.0 // indirect golang.org/x/oauth2 v0.10.0 // indirect @@ -60,6 +64,7 @@ require ( google.golang.org/appengine v1.6.7 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230803162519-f966b187b2e5 // indirect google.golang.org/protobuf v1.31.0 // indirect + gopkg.in/alecthomas/kingpin.v2 v2.2.6 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 6c1f273..99033f0 100644 --- a/go.sum +++ b/go.sum @@ -37,7 +37,9 @@ github.com/NYTimes/gziphandler v0.0.0-20170623195520-56545f4a5d46/go.mod h1:3wb0 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/PuerkitoBio/purell v1.1.1/go.mod h1:c11w/QuzBsJSee3cPx9rAFu61PvFxuPbtSwDGJws/X0= github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdkoq3SwY9Y+13GIhn11/XLaGBb4BfwItxLd5jeuXE= +github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc h1:cAKDfWh5VpdgMhJosfJnn5/FoN2SRZ4p7fJNX58YPaU= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= +github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf h1:qet1QNfXsQxTZqLG4oE62mJzwPIB8+Tee4RNCL9ulrY= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o= github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY= @@ -210,6 +212,7 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q= github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= @@ -297,6 +300,7 @@ github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts= github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc= github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc= +github.com/sirupsen/logrus v1.2.0 h1:juTguoYk5qI21pwyTXY3B3Y5cOTH3ZUyZCg1v/mihuo= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= @@ -566,6 +570,7 @@ google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp0 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= From 81a18edbe51b6f1fb1d7ae778acc7014f90de1af Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Tue, 11 Aug 2026 14:17:28 +0200 Subject: [PATCH 09/15] fixes --- pkg/rclone/rclone.go | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index d120350..368c2f0 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -323,11 +323,9 @@ func (r *Rclone) unmountInBackground(rcloneVolume *RcloneVolume, targetPath stri err := func() error { configName := rcloneVolume.deploymentName() vfs := fmt.Sprintf("%s:%s", configName, rcloneVolume.RemotePath) - if vfs != "" { - err := r.waitForVFSQueue(unmountCtx, vfs) - if err != nil { - klog.Infof("Error waiting for VFS: %v", err) - } + err := r.waitForVFSQueue(unmountCtx, vfs) + if err != nil { + klog.Infof("Error waiting for VFS: %v", err) } klog.Infof("unmounting %s", rcloneVolume.deploymentName()) @@ -400,30 +398,30 @@ func (r *Rclone) waitForVFSQueue(ctx context.Context, vfs string) error { 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) + 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) + 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) + 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) + 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) + 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 queue, fmt.Errorf("getting VFS queue failed: %w", err) } return result, nil } From 6b0d1ee8a0b942b54c0e64ae9236f2057153fe33 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Tue, 11 Aug 2026 14:22:47 +0200 Subject: [PATCH 10/15] fix error handling --- pkg/rclone/nodeserver.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/pkg/rclone/nodeserver.go b/pkg/rclone/nodeserver.go index ad7d97a..07ebecc 100644 --- a/pkg/rclone/nodeserver.go +++ b/pkg/rclone/nodeserver.go @@ -315,8 +315,12 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu } rcloneVolume, err := ns.RcloneOps.GetVolumeById(ctx, req.GetVolumeId()) - if err == ErrVolumeNotFound { - klog.Warning("VolumeId not found for NodeUnpublishVolume") + if err != nil { + if err == ErrVolumeNotFound { + klog.Warning("VolumeId not found for NodeUnpublishVolume") + } else { + klog.Errorf("Could not find rclone volume from volume ID %s", req.GetVolumeId()) + } mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false) return &csi.NodeUnpublishVolumeResponse{}, nil } From dadb0b7a5515d8e90dcca4127eaaf1cea4ad209a Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Tue, 11 Aug 2026 14:25:10 +0200 Subject: [PATCH 11/15] add comment --- pkg/rclone/rclone.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/rclone/rclone.go b/pkg/rclone/rclone.go index 368c2f0..f149d46 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -283,6 +283,9 @@ func (r *Rclone) DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rclo func (r *Rclone) Unmount(ctx context.Context, rcloneVolume *RcloneVolume, targetPath string) error { // Unmount in the background, with only one process per volume ID + // Note that `ctx` corresponds to the `NodeUnpublishVolume` 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. volumeId := rcloneVolume.ID r.unmountMutex.Lock() unmountContext, found := r.unmountContexts[volumeId] From a03a91654a6b36e865cf5c62d62afcbd753a8f56 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Fri, 14 Aug 2026 09:35:18 +0200 Subject: [PATCH 12/15] move background task --- pkg/rclone/driver.go | 5 +- pkg/rclone/nodeserver.go | 91 ++++++++++++++++----- pkg/rclone/rclone.go | 168 ++++++++++++++------------------------- 3 files changed, 136 insertions(+), 128 deletions(-) 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 07ebecc..d065b4a 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,22 +326,32 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Target Path must be provided") } - rcloneVolume, err := ns.RcloneOps.GetVolumeById(ctx, req.GetVolumeId()) - if err != nil { - if err == ErrVolumeNotFound { - klog.Warning("VolumeId not found for NodeUnpublishVolume") - } else { - klog.Errorf("Could not find rclone volume from volume ID %s", req.GetVolumeId()) - } - mount.CleanupMountPoint(req.GetTargetPath(), ns.mounter, false) - return &csi.NodeUnpublishVolumeResponse{}, nil + volumeId := req.GetVolumeId() + ns.unpublishMutex.Lock() + unmountContext, found := ns.unpublishContexts[volumeId] + if !found { + unmountContext = ns.unpublishInBackground(volumeId, targetPath, unpublishTimeout) + ns.unpublishContexts[volumeId] = unmountContext + } + ns.unpublishMutex.Unlock() + + // Wait for the unpublish operation to finish + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-unmountContext.Done(): } - if err := ns.RcloneOps.Unmount(ctx, rcloneVolume, targetPath); err != nil { - klog.Warningf("Unmounting volume failed: %s", err) + // Reset the unmountContext map + ns.unpublishMutex.Lock() + delete(ns.unpublishContexts, volumeId) + ns.unpublishMutex.Unlock() + + err := context.Cause(unmountContext) + 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 { @@ -344,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 f149d46..dcc7e64 100644 --- a/pkg/rclone/rclone.go +++ b/pkg/rclone/rclone.go @@ -12,7 +12,6 @@ import ( "os" os_exec "os/exec" "strings" - "sync" "syscall" "time" @@ -28,9 +27,9 @@ var ( ErrVolumeNotFound = errors.New("volume is not found") ) -var ( - // Timeout for unmounting a volume: wait for the VFS queue to sync files until the deadline - unmountTimeout = time.Minute +const ( + // Timeout for waiting on the VFS queue to sync all files before unmounting + unmountVfsQueueTimeout = time.Minute ) type Operations interface { @@ -44,14 +43,12 @@ type Operations interface { } type Rclone struct { - execute exec.Interface - kubeClient *kubernetes.Clientset - daemonCmd *os_exec.Cmd - port int - cacheDir string - cacheSize string - unmountContexts map[string]context.Context - unmountMutex sync.Mutex + execute exec.Interface + kubeClient *kubernetes.Clientset + daemonCmd *os_exec.Cmd + port int + cacheDir string + cacheSize string } type RcloneVolume struct { @@ -282,100 +279,55 @@ func (r *Rclone) DeleteVol(ctx context.Context, rcloneVolume *RcloneVolume, rclo } func (r *Rclone) Unmount(ctx context.Context, rcloneVolume *RcloneVolume, targetPath string) error { - // Unmount in the background, with only one process per volume ID - // Note that `ctx` corresponds to the `NodeUnpublishVolume` 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. - volumeId := rcloneVolume.ID - r.unmountMutex.Lock() - unmountContext, found := r.unmountContexts[volumeId] - if !found { - unmountContext = r.unmountInBackground(rcloneVolume, targetPath, unmountTimeout) - r.unmountContexts[volumeId] = unmountContext - } - r.unmountMutex.Unlock() - - // Wait for the unmounting to finish - select { - case <-ctx.Done(): - return ctx.Err() - case <-unmountContext.Done(): - } - - // Reset the unmountContext map - r.unmountMutex.Lock() - delete(r.unmountContexts, volumeId) - r.unmountMutex.Unlock() - - err := context.Cause(unmountContext) - if errors.Is(err, context.Canceled) { - return nil + 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) } - return err -} -func (r *Rclone) unmountInBackground(rcloneVolume *RcloneVolume, targetPath string, timeout time.Duration) context.Context { - unmountCtx, unmountCancel := context.WithCancelCause(context.Background()) - // Setup context deadline - go func() { - time.Sleep(timeout) - unmountCancel(context.DeadlineExceeded) - }() - // Perform unmounting in the background - go func() { - err := func() error { - configName := rcloneVolume.deploymentName() - vfs := fmt.Sprintf("%s:%s", configName, rcloneVolume.RemotePath) - err := r.waitForVFSQueue(unmountCtx, vfs) - if err != nil { - klog.Infof("Error waiting for VFS: %v", err) - } - - klog.Infof("unmounting %s", rcloneVolume.deploymentName()) - unmountArgs := UnmountRequest{ - MountPoint: targetPath, - } - postBody, err := json.Marshal(unmountArgs) - if err != nil { - return fmt.Errorf("unmounting failed: couldn't create request body: %s", err) - } - requestBody := bytes.NewBuffer(postBody) - resp, err := http.Post(fmt.Sprintf("http://localhost:%d/mount/unmount", r.port), "application/json", requestBody) - if err != nil { - return fmt.Errorf("unmounting failed: couldn't send HTTP request: %w", err) - } - err = checkResponse(resp) - if err != nil { - return fmt.Errorf("unmounting failed: %w", err) - } - klog.Infof("deleted mount with volume ID %s at path %s", rcloneVolume.ID, targetPath) - - configDelete := ConfigDeleteRequest{ - Name: rcloneVolume.deploymentName(), - } - postBody, err = json.Marshal(configDelete) - if err != nil { - return fmt.Errorf("deleting config failed: couldn't create request body: %s", err) - } - requestBody = bytes.NewBuffer(postBody) - resp, err = http.Post(fmt.Sprintf("http://localhost:%d/config/delete", r.port), "application/json", requestBody) - if err != nil { - klog.Errorf("deleting config failed: couldn't send HTTP request: %v", err) - return nil - } - err = checkResponse(resp) - if err != nil { - klog.Errorf("deleting config failed: %v", err) - return nil - } - klog.Infof("deleted config for volume ID %s at path %s", rcloneVolume.ID, targetPath) + klog.Infof("unmounting %s", rcloneVolume.deploymentName()) + unmountArgs := UnmountRequest{ + MountPoint: targetPath, + } + postBody, err := json.Marshal(unmountArgs) + if err != nil { + return fmt.Errorf("unmounting failed: couldn't create request body: %s", err) + } + requestBody := bytes.NewBuffer(postBody) + resp, err := http.Post(fmt.Sprintf("http://localhost:%d/mount/unmount", r.port), "application/json", requestBody) + if err != nil { + return fmt.Errorf("unmounting failed: couldn't send HTTP request: %w", err) + } + err = checkResponse(resp) + if err != nil { + return fmt.Errorf("unmounting failed: %w", err) + } + klog.Infof("deleted mount with volume ID %s at path %s", rcloneVolume.ID, targetPath) - return nil - }() - unmountCancel(err) - }() + configDelete := ConfigDeleteRequest{ + Name: rcloneVolume.deploymentName(), + } + postBody, err = json.Marshal(configDelete) + if err != nil { + return fmt.Errorf("deleting config failed: couldn't create request body: %s", err) + } + requestBody = bytes.NewBuffer(postBody) + resp, err = http.Post(fmt.Sprintf("http://localhost:%d/config/delete", r.port), "application/json", requestBody) + if err != nil { + klog.Errorf("deleting config failed: couldn't send HTTP request: %v", err) + return nil + } + err = checkResponse(resp) + if err != nil { + klog.Errorf("deleting config failed: %v", err) + return nil + } + klog.Infof("deleted config for volume ID %s at path %s", rcloneVolume.ID, targetPath) - return unmountCtx + return nil } func (r *Rclone) waitForVFSQueue(ctx context.Context, vfs string) error { @@ -478,13 +430,11 @@ func (r *Rclone) GetVolumeById(ctx context.Context, volumeId string) (*RcloneVol func NewRclone(kubeClient *kubernetes.Clientset, port int, cacheDir string, cacheSize string) Operations { rclone := &Rclone{ - execute: exec.New(), - kubeClient: kubeClient, - port: port, - cacheDir: cacheDir, - cacheSize: cacheSize, - unmountContexts: map[string]context.Context{}, - unmountMutex: sync.Mutex{}, + execute: exec.New(), + kubeClient: kubeClient, + port: port, + cacheDir: cacheDir, + cacheSize: cacheSize, } return rclone } From 28f45c7b41e7b210eb35d123a6e2f5684e212733 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Fri, 14 Aug 2026 09:39:25 +0200 Subject: [PATCH 13/15] temp: build image --- .github/workflows/build.yaml | 71 ++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 .github/workflows/build.yaml diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml new file mode 100644 index 0000000..c45fb34 --- /dev/null +++ b/.github/workflows/build.yaml @@ -0,0 +1,71 @@ +name: Build dev version + +on: + push: + workflow_dispatch: + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository }} + CHART_NAME: ${{ github.repository }}/helm-chart + +defaults: + run: + shell: bash + +concurrency: + group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +jobs: + build-image: + runs-on: ubuntu-24.04 + outputs: + image: ${{ steps.docker_image.outputs.image }} + image_repository: ${{ steps.docker_image.outputs.image_repository }} + image_tag: ${{ steps.docker_image.outputs.image_tag }} + permissions: + contents: read + packages: write + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Docker image metadata + id: meta + uses: docker/metadata-action@v5 + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: type=sha + - name: Extract Docker image name + id: docker_image + env: + IMAGE_TAGS: ${{ steps.meta.outputs.tags }} + run: | + IMAGE=$(echo "$IMAGE_TAGS" | cut -d" " -f1) + IMAGE_REPOSITORY=$(echo "$IMAGE" | cut -d":" -f1) + IMAGE_TAG=$(echo "$IMAGE" | cut -d":" -f2) + echo "image=$IMAGE" >> "$GITHUB_OUTPUT" + echo "image_repository=$IMAGE_REPOSITORY" >> "$GITHUB_OUTPUT" + echo "image_tag=$IMAGE_TAG" >> "$GITHUB_OUTPUT" + - name: Set up Docker buildx + uses: docker/setup-buildx-action@v3 + - name: Set up Docker + uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - name: Build and push Docker image + uses: docker/build-push-action@v6 + with: + context: . + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache + cache-to: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache,mode=max + +# TODO: add job to build and push the helm chart if needed (manual trigger) \ No newline at end of file From ed3c38498419ca0068bfa1900ea5b18eceef6711 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Fri, 14 Aug 2026 09:55:17 +0200 Subject: [PATCH 14/15] cleanup --- .github/workflows/build.yaml | 71 ------------------------------------ 1 file changed, 71 deletions(-) delete mode 100644 .github/workflows/build.yaml diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml deleted file mode 100644 index c45fb34..0000000 --- a/.github/workflows/build.yaml +++ /dev/null @@ -1,71 +0,0 @@ -name: Build dev version - -on: - push: - workflow_dispatch: - -env: - REGISTRY: ghcr.io - IMAGE_NAME: ${{ github.repository }} - CHART_NAME: ${{ github.repository }}/helm-chart - -defaults: - run: - shell: bash - -concurrency: - group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} - cancel-in-progress: true - -permissions: - contents: read - -jobs: - build-image: - runs-on: ubuntu-24.04 - outputs: - image: ${{ steps.docker_image.outputs.image }} - image_repository: ${{ steps.docker_image.outputs.image_repository }} - image_tag: ${{ steps.docker_image.outputs.image_tag }} - permissions: - contents: read - packages: write - steps: - - name: Checkout - uses: actions/checkout@v4 - - name: Docker image metadata - id: meta - uses: docker/metadata-action@v5 - with: - images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} - tags: type=sha - - name: Extract Docker image name - id: docker_image - env: - IMAGE_TAGS: ${{ steps.meta.outputs.tags }} - run: | - IMAGE=$(echo "$IMAGE_TAGS" | cut -d" " -f1) - IMAGE_REPOSITORY=$(echo "$IMAGE" | cut -d":" -f1) - IMAGE_TAG=$(echo "$IMAGE" | cut -d":" -f2) - echo "image=$IMAGE" >> "$GITHUB_OUTPUT" - echo "image_repository=$IMAGE_REPOSITORY" >> "$GITHUB_OUTPUT" - echo "image_tag=$IMAGE_TAG" >> "$GITHUB_OUTPUT" - - name: Set up Docker buildx - uses: docker/setup-buildx-action@v3 - - name: Set up Docker - uses: docker/login-action@v3 - with: - registry: ${{ env.REGISTRY }} - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - name: Build and push Docker image - uses: docker/build-push-action@v6 - with: - context: . - push: true - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache - cache-to: type=registry,ref=${{ steps.docker_image.outputs.image_repository }}:buildcache,mode=max - -# TODO: add job to build and push the helm chart if needed (manual trigger) \ No newline at end of file From dfbaa2f0c73623499e803ad5a5d4cbce59a000a6 Mon Sep 17 00:00:00 2001 From: Flora Thiebaut Date: Fri, 14 Aug 2026 09:56:13 +0200 Subject: [PATCH 15/15] rename var --- pkg/rclone/nodeserver.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/pkg/rclone/nodeserver.go b/pkg/rclone/nodeserver.go index d065b4a..14eb4d2 100644 --- a/pkg/rclone/nodeserver.go +++ b/pkg/rclone/nodeserver.go @@ -328,10 +328,10 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu volumeId := req.GetVolumeId() ns.unpublishMutex.Lock() - unmountContext, found := ns.unpublishContexts[volumeId] + unpublishContext, found := ns.unpublishContexts[volumeId] if !found { - unmountContext = ns.unpublishInBackground(volumeId, targetPath, unpublishTimeout) - ns.unpublishContexts[volumeId] = unmountContext + unpublishContext = ns.unpublishInBackground(volumeId, targetPath, unpublishTimeout) + ns.unpublishContexts[volumeId] = unpublishContext } ns.unpublishMutex.Unlock() @@ -339,7 +339,7 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu select { case <-ctx.Done(): return nil, ctx.Err() - case <-unmountContext.Done(): + case <-unpublishContext.Done(): } // Reset the unmountContext map @@ -347,7 +347,7 @@ func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpu delete(ns.unpublishContexts, volumeId) ns.unpublishMutex.Unlock() - err := context.Cause(unmountContext) + err := context.Cause(unpublishContext) if errors.Is(err, context.Canceled) { return &csi.NodeUnpublishVolumeResponse{}, nil }