feat: Mount existing bucket and access the data already in the bucket #…

…55

chore: updated README.md

README.md

@@ -104,6 +104,26 @@ parameters:
 
 If the bucket is specified, it will still be created if it does not exist on the backend. Every volume will get its own prefix within the bucket which matches the volume ID. When deleting a volume, also just the prefix will be deleted.
 
+#### Using an existing bucket with custom prefix
+
+If you have an existing bucket and with or without a prefix (subpath), you can specify to use a prefixed configuration by setting the parameters as:
+
+```yaml
+kind: StorageClass
+apiVersion: storage.k8s.io/v1
+metadata:
+  name: csi-s3-existing-bucket
+provisioner: ch.ctrox.csi.s3-driver
+reclaimPolicy: Retain
+parameters:
+  mounter: rclone
+  bucket: some-existing-bucket-name
+  # 'usePrefix' must be true in order to enable the prefix feature and to avoid the removal of the bucket
+  usePrefix: "true"
+  # 'prefix' can be empty (it will mount on the root of the bucket), an existing prefix or a new one.
+  prefix: custom-prefix
+```
+
 ### Mounter
 
 As S3 is not a real file system there are some limitations to consider here. Depending on what mounter you are using, you will have different levels of POSIX compability. Also depending on what S3 storage backend you are using there are not always [consistency guarantees](https://github.com/gaul/are-we-consistent-yet#observed-consistency).

pkg/driver/controllerserver.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"path"
+	"strconv"
 	"strings"
 
 	"github.com/ctrox/csi-s3/pkg/mounter"
@@ -50,6 +51,8 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	volumeID := sanitizeVolumeID(req.GetName())
 	bucketName := volumeID
 	prefix := ""
+	usePrefix := params[mounter.UsePrefix]
+	defaultFsPath := defaultFsPath
 
 	// check if bucket name is overridden
 	if nameOverride, ok := params[mounter.BucketKey]; ok {
@@ -58,6 +61,16 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		volumeID = path.Join(bucketName, prefix)
 	}
 
+	// check if volume prefix is overridden
+	if overridePrefix, err := strconv.ParseBool(usePrefix); err == nil && overridePrefix {
+		prefix = ""
+		defaultFsPath = ""
+		if prefixOverride, ok := params[mounter.VolumePrefix]; ok && prefixOverride != "" {
+			prefix = prefixOverride
+		}
+		volumeID = path.Join(bucketName, prefix)
+	}
+
 	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
 		glog.V(3).Infof("invalid create volume req: %v", req)
 		return nil, err
@@ -75,6 +88,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 
 	meta := &s3.FSMeta{
 		BucketName:    bucketName,
+		UsePrefix:     usePrefix,
 		Prefix:        prefix,
 		Mounter:       mounterType,
 		CapacityBytes: capacityBytes,
@@ -108,7 +122,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		}
 	}
 
-	if err = client.CreatePrefix(bucketName, path.Join(prefix, defaultFsPath)); err != nil {
+	if err = client.CreatePrefix(bucketName, path.Join(prefix, defaultFsPath)); err != nil && prefix != "" {
 		return nil, fmt.Errorf("failed to create prefix %s: %v", path.Join(prefix, defaultFsPath), err)
 	}
 
@@ -153,7 +167,10 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	}
 
 	var deleteErr error
-	if prefix == "" {
+	if usePrefix, err := strconv.ParseBool(client.GetFSMetaField(meta, mounter.UsePrefix)); err == nil && usePrefix {
+		// UsePrefix is true, we do not delete anything
+		glog.V(4).Infof("Nothing to remove for %s", bucketName)
+	} else if prefix == "" {
 		// prefix is empty, we delete the whole bucket
 		if err := client.RemoveBucket(bucketName); err != nil {
 			deleteErr = err

pkg/driver/driver.go

@@ -33,7 +33,7 @@ type driver struct {
 }
 
 var (
-	vendorVersion = "v1.2.0-rc.1"
+	vendorVersion = "v1.2.0-rc.2"
 	driverName    = "ch.ctrox.csi.s3-driver"
 )
 

pkg/mounter/mounter.go

@@ -31,6 +31,8 @@ const (
 	rcloneMounterType   = "rclone"
 	TypeKey             = "mounter"
 	BucketKey           = "bucket"
+	VolumePrefix        = "prefix"
+	UsePrefix           = "usePrefix"
 )
 
 // New returns a new mounter depending on the mounterType parameter

pkg/s3/client.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"net/url"
 	"path"
+	"reflect"
 
 	"github.com/golang/glog"
 	"github.com/minio/minio-go/v7"
@@ -36,6 +37,7 @@ type Config struct {
 type FSMeta struct {
 	BucketName    string `json:"Name"`
 	Prefix        string `json:"Prefix"`
+	UsePrefix     string `json:"UsePrefix"`
 	Mounter       string `json:"Mounter"`
 	FSPath        string `json:"FSPath"`
 	CapacityBytes int64  `json:"CapacityBytes"`
@@ -250,3 +252,9 @@ func (client *s3Client) GetFSMeta(bucketName, prefix string) (*FSMeta, error) {
 	err = json.Unmarshal(b, &meta)
 	return &meta, err
 }
+
+func (client *s3Client) GetFSMetaField(meta *FSMeta, field string) string {
+	r := reflect.ValueOf(meta)
+	f := reflect.Indirect(r).FieldByName(field)
+	return f.String()
+}