fix global prefix handling with marker

Author: ahouene

backends/s3/marker.go

@@ -18,7 +18,7 @@ func (b *Backend) setMarker(ctx context.Context, name, etag string, isDel bool)
 	}
 	nanos := time.Now().UnixNano()
 	s := fmt.Sprintf("%s:%s:%d:%v", name, etag, nanos, isDel)
-	_, err := b.doStore(ctx, UpdateMarkerFilename, []byte(s))
+	_, err := b.doStore(ctx, b.markerName, []byte(s))
 	if err != nil {
 		return err
 	}

backends/s3/s3.go

@@ -106,23 +106,24 @@ func (o Options) Check() error {
 }
 
 type Backend struct {
-	opt    Options
-	config *minio.Options
-	client *minio.Client
-	log    logr.Logger
+	opt        Options
+	config     *minio.Options
+	client     *minio.Client
+	log        logr.Logger
+	markerName string
 
 	mu         sync.Mutex
 	lastMarker string
 	lastList   simpleblob.BlobList
 	lastTime   time.Time
 }
 
-func (b *Backend) List(ctx context.Context, prefix string) (simpleblob.BlobList, error) {
-	// Prepend global prefix
-	prefix = b.prependGlobalPrefix(prefix)
+func (b *Backend) List(ctx context.Context, prefix string) (blobList simpleblob.BlobList, err error) {
+	// Handle global prefix
+	combinedPrefix := b.prependGlobalPrefix(prefix)
 
 	if !b.opt.UseUpdateMarker {
-		return b.doList(ctx, prefix)
+		return b.doList(ctx, combinedPrefix)
 	}
 
 	m, err := b.Load(ctx, UpdateMarkerFilename)
@@ -144,7 +145,7 @@ func (b *Backend) List(ctx context.Context, prefix string) (simpleblob.BlobList,
 		return blobs.WithPrefix(prefix), nil
 	}
 
-	blobs, err = b.doList(ctx, "") // We want to cache all, so no prefix
+	blobs, err = b.doList(ctx, b.opt.GlobalPrefix) // We want to cache all, so no prefix
 	if err != nil {
 		return nil, err
 	}
@@ -162,7 +163,9 @@ func (b *Backend) doList(ctx context.Context, prefix string) (simpleblob.BlobLis
 	var blobs simpleblob.BlobList
 
 	// Runes to strip from blob names for GlobalPrefix
-	gpEndRune := len(b.opt.GlobalPrefix)
+	// This is fine, because we can trust the API to only return with the prefix.
+	// TODO: trust but verify
+	gpEndIndex := len(b.opt.GlobalPrefix)
 
 	objCh := b.client.ListObjects(ctx, b.opt.Bucket, minio.ListObjectsOptions{
 		Prefix:    prefix,
@@ -177,14 +180,14 @@ func (b *Backend) doList(ctx context.Context, prefix string) (simpleblob.BlobLis
 
 		metricCalls.WithLabelValues("list").Inc()
 		metricLastCallTimestamp.WithLabelValues("list").SetToCurrentTime()
-		if obj.Key == UpdateMarkerFilename {
+		if obj.Key == b.markerName {
 			continue
 		}
 
 		// Strip global prefix from blob
 		blobName := obj.Key
-		if gpEndRune > 0 {
-			blobName = blobName[gpEndRune:]
+		if gpEndIndex > 0 {
+			blobName = blobName[gpEndIndex:]
 		}
 
 		blobs = append(blobs, simpleblob.Blob{Name: blobName, Size: obj.Size})
@@ -378,10 +381,18 @@ func New(ctx context.Context, opt Options) (*Backend, error) {
 		client: client,
 		log:    log,
 	}
+	b.setGlobalPrefix(opt.GlobalPrefix)
 
 	return b, nil
 }
 
+// setGlobalPrefix updates the global prefix in b and the cached marker name,
+// so it can be dynamically changed in tests.
+func (b *Backend) setGlobalPrefix(prefix string) {
+	b.opt.GlobalPrefix = prefix
+	b.markerName = b.prependGlobalPrefix(UpdateMarkerFilename)
+}
+
 // convertMinioError takes an error, possibly a minio.ErrorResponse
 // and turns it into a well known error when possible.
 // If error is not well known, it is returned as is.

backends/s3/s3_test.go

@@ -66,7 +66,7 @@ func getBackend(ctx context.Context, t *testing.T) (b *Backend) {
 			}
 		}
 		// This one is not returned by the List command
-		err = b.client.RemoveObject(ctx, b.opt.Bucket, UpdateMarkerFilename, minio.RemoveObjectOptions{})
+		err = b.client.RemoveObject(ctx, b.opt.Bucket, b.markerName, minio.RemoveObjectOptions{})
 		require.NoError(t, err)
 	}
 	t.Cleanup(cleanup)
@@ -107,13 +107,19 @@ func TestBackend_globalprefix(t *testing.T) {
 	defer cancel()
 
 	b := getBackend(ctx, t)
-	b.opt.GlobalPrefix = "v5/"
+	b.setGlobalPrefix("v5/")
 
 	tester.DoBackendTests(t, b)
 	assert.Empty(t, b.lastMarker)
+}
 
-	b = getBackend(ctx, t)
-	b.opt.GlobalPrefix = "v6/"
+func TestBackend_globalPrefixAndMarker(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// Start the backend over
+	b := getBackend(ctx, t)
+	b.setGlobalPrefix("v6/")
 	b.opt.UseUpdateMarker = true
 
 	tester.DoBackendTests(t, b)