mirror of
https://github.com/versity/versitygw.git
synced 2026-08-28 20:06:02 +00:00
feat(azure): carry Azure blob marker in continuation token for pagination
* feat(azure): carry Azure blob marker in continuation token for pagination Azure blob markers are opaque values only Azure may mint, so an S3 object key can never be passed back to Azure as a listing marker. Paging the underlying Azure listing with an S3 key therefore failed or re-scanned the whole prefix on every page. Introduce azMarkerToken, an opaque S3 continuation token that carries the Azure marker alongside the last returned key: the Azure marker resumes the blob listing where it stopped, and the last key filters out already-returned entries. Tokens are versioned with a "vgw1." prefix; anything without it is treated as a plain key, so tokens from older versions and hand-crafted markers keep working. The shared listBlobs helper now backs both ListObjects and ListObjectsV2, applying the S3 marker and delimiter client side. ListObjectsV2 pages efficiently via the token; ListObjects (v1) has no token to carry state and walks the prefix from the start each page. Add unit tests (token round-trip, marker-resumed pagination, delimiter and common-prefix handling, multipart filtering) driven by a fake Azure container, and integration tests covering full and delimited pagination. Signed-off-by: Nils Leger <nils.leger@getflip.com> * fix: token is now bound to delimiter too Signed-off-by: Nils Leger <nils.leger@getflip.com> --------- Signed-off-by: Nils Leger <nils.leger@getflip.com>
This commit is contained in:
Nils Leger
+220
-188
@@ -754,6 +754,176 @@ func (az *Azure) GetObjectAttributes(ctx context.Context, input *s3.GetObjectAtt
|
||||
}, nil
|
||||
}
|
||||
|
||||
// azMarkerToken is the decoded form of an S3 continuation token. Azure blob
|
||||
// markers are opaque values that only Azure may mint, so an S3 key can never be
|
||||
// used as one. The token therefore carries the Azure marker alongside the last
|
||||
// key of the previous page: the Azure marker resumes the blob listing close to
|
||||
// where it stopped, and the last key filters out the entries of that Azure page
|
||||
// which were already returned.
|
||||
type azMarkerToken struct {
|
||||
Prefix string `json:"p,omitempty"`
|
||||
Delimiter string `json:"d,omitempty"`
|
||||
Marker string `json:"m,omitempty"`
|
||||
LastKey string `json:"k,omitempty"`
|
||||
}
|
||||
|
||||
// azTokenPrefix marks continuation tokens minted by this backend. Anything
|
||||
// without it is treated as a plain key, which keeps tokens issued by older
|
||||
// versions and hand-crafted markers working.
|
||||
const azTokenPrefix = "vgw1."
|
||||
|
||||
func encodeAzMarkerToken(t azMarkerToken) string {
|
||||
data, err := json.Marshal(t)
|
||||
if err != nil {
|
||||
// the struct only holds strings, so this can't fail
|
||||
return t.LastKey
|
||||
}
|
||||
return azTokenPrefix + base64.RawURLEncoding.EncodeToString(data)
|
||||
}
|
||||
|
||||
// decodeAzMarkerToken returns the Azure marker and the last key encoded in the
|
||||
// continuation token. The Azure marker is only returned for a token minted for
|
||||
// the same prefix and delimiter, as the marker resumes the underlying blob
|
||||
// listing and the last key was computed under that request's delimiter; reusing
|
||||
// either across a different prefix or delimiter would skip the wrong entries.
|
||||
func decodeAzMarkerToken(token, prefix, delimiter string) (azureMarker, lastKey string) {
|
||||
if !strings.HasPrefix(token, azTokenPrefix) {
|
||||
return "", token
|
||||
}
|
||||
|
||||
data, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(token, azTokenPrefix))
|
||||
if err != nil {
|
||||
return "", token
|
||||
}
|
||||
|
||||
var t azMarkerToken
|
||||
if err := json.Unmarshal(data, &t); err != nil {
|
||||
return "", token
|
||||
}
|
||||
if t.Prefix != prefix || t.Delimiter != delimiter {
|
||||
return "", t.LastKey
|
||||
}
|
||||
|
||||
return t.Marker, t.LastKey
|
||||
}
|
||||
|
||||
// azListingOpts describes a single page of an S3 object listing.
|
||||
type azListingOpts struct {
|
||||
prefix string
|
||||
delimiter string
|
||||
// marker is the S3 key to start after, exclusive.
|
||||
marker string
|
||||
// azureMarker optionally resumes the underlying Azure listing instead of
|
||||
// walking the prefix from the beginning.
|
||||
azureMarker string
|
||||
maxKeys int32
|
||||
owner string
|
||||
}
|
||||
|
||||
// azListingPage is a page of objects and common prefixes gathered from Azure.
|
||||
type azListingPage struct {
|
||||
objects []s3response.Object
|
||||
commonPrefixes []types.CommonPrefix
|
||||
isTruncated bool
|
||||
// lastKey is the last object key or common prefix of the page.
|
||||
lastKey string
|
||||
// resumeMarker is the Azure marker of the blob page the listing stopped in.
|
||||
resumeMarker string
|
||||
}
|
||||
|
||||
// listBlobs collects one S3 listing page from Azure. Azure is always paged with
|
||||
// its own markers and the delimiter is applied client side, both to match S3
|
||||
// semantics and because Azure has no equivalent of "start after".
|
||||
func (az *Azure) listBlobs(ctx context.Context, client *container.Client, opts azListingOpts) (azListingPage, error) {
|
||||
var page azListingPage
|
||||
|
||||
pager := client.NewListBlobsHierarchyPager("", &container.ListBlobsHierarchyOptions{
|
||||
Prefix: backend.GetPtrFromString(opts.prefix),
|
||||
Marker: backend.GetPtrFromString(opts.azureMarker),
|
||||
})
|
||||
|
||||
// the Azure marker that started the blob page currently being processed
|
||||
blobPageMarker := opts.azureMarker
|
||||
cpSet := make(map[string]struct{})
|
||||
var pastMax bool
|
||||
var totalFound int32
|
||||
|
||||
loop:
|
||||
for pager.More() {
|
||||
resp, err := pager.NextPage(ctx)
|
||||
if err != nil {
|
||||
return azListingPage{}, azureErrToS3Err(err)
|
||||
}
|
||||
|
||||
for _, v := range resp.Segment.BlobItems {
|
||||
name := backend.GetStringFromPtr(v.Name)
|
||||
|
||||
// Filter out multipart upload blobs
|
||||
if strings.HasPrefix(name, string(metaTmpMultipartPrefix)) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Apply delimiter logic to determine if this blob contributes to
|
||||
// a common prefix or is a regular object
|
||||
key := name
|
||||
isCP := false
|
||||
if opts.delimiter != "" {
|
||||
suffix := strings.TrimPrefix(name, opts.prefix)
|
||||
before, _, found := strings.Cut(suffix, opts.delimiter)
|
||||
if found {
|
||||
isCP = true
|
||||
key = opts.prefix + before + opts.delimiter
|
||||
}
|
||||
}
|
||||
|
||||
// Skip everything at or before the marker
|
||||
if key <= opts.marker {
|
||||
continue
|
||||
}
|
||||
// Deduplicate: multiple blobs can map to the same common prefix
|
||||
if isCP {
|
||||
if _, exists := cpSet[key]; exists {
|
||||
continue
|
||||
}
|
||||
}
|
||||
// A further entry beyond maxKeys means the listing is truncated
|
||||
if pastMax {
|
||||
page.isTruncated = true
|
||||
page.resumeMarker = blobPageMarker
|
||||
break loop
|
||||
}
|
||||
|
||||
if isCP {
|
||||
cpSet[key] = struct{}{}
|
||||
page.commonPrefixes = append(page.commonPrefixes, types.CommonPrefix{
|
||||
Prefix: backend.GetPtrFromString(key),
|
||||
})
|
||||
} else {
|
||||
page.objects = append(page.objects, s3response.Object{
|
||||
ETag: backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
|
||||
Key: v.Name,
|
||||
LastModified: v.Properties.LastModified,
|
||||
Size: v.Properties.ContentLength,
|
||||
StorageClass: types.ObjectStorageClassStandard,
|
||||
Owner: &types.Owner{
|
||||
ID: &opts.owner,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
page.lastKey = key
|
||||
totalFound++
|
||||
if totalFound == opts.maxKeys {
|
||||
pastMax = true
|
||||
}
|
||||
}
|
||||
|
||||
blobPageMarker = backend.GetStringFromPtr(resp.NextMarker)
|
||||
}
|
||||
|
||||
return page, nil
|
||||
}
|
||||
|
||||
func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s3response.ListObjectsResult, error) {
|
||||
// Retrieve the bucket acl to get the bucket owner
|
||||
// All the objects in the bucket are owner by the bucket owner
|
||||
@@ -794,108 +964,36 @@ func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Use flat listing (empty delimiter) and handle delimiter logic client-side,
|
||||
// matching S3 semantics. Only pass Prefix and Marker to Azure.
|
||||
pager := client.NewListBlobsHierarchyPager("", &container.ListBlobsHierarchyOptions{
|
||||
Prefix: input.Prefix,
|
||||
Marker: input.Marker,
|
||||
// The S3 marker is an object key, which Azure rejects as a blob listing
|
||||
// marker, so it is applied client side. ListObjects has no opaque token to
|
||||
// carry an Azure marker in, hence every page walks the prefix from the
|
||||
// start; use ListObjectsV2 to page large listings efficiently.
|
||||
page, err := az.listBlobs(ctx, client, azListingOpts{
|
||||
prefix: prefix,
|
||||
delimiter: delimiter,
|
||||
marker: effectiveMarker,
|
||||
maxKeys: maxKeys,
|
||||
owner: acl.Owner,
|
||||
})
|
||||
|
||||
var objects []s3response.Object
|
||||
var cPrefixes []types.CommonPrefix
|
||||
cpSet := make(map[string]struct{})
|
||||
var pastMax, isTruncated bool
|
||||
var candidateMarker string
|
||||
var totalFound int32
|
||||
|
||||
loop:
|
||||
for pager.More() {
|
||||
resp, err := pager.NextPage(ctx)
|
||||
if err != nil {
|
||||
return s3response.ListObjectsResult{}, azureErrToS3Err(err)
|
||||
}
|
||||
|
||||
for _, v := range resp.Segment.BlobItems {
|
||||
name := backend.GetStringFromPtr(v.Name)
|
||||
|
||||
// Filter out multipart upload blobs
|
||||
if strings.HasPrefix(name, string(metaTmpMultipartPrefix)) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Apply delimiter logic to determine if this blob contributes to
|
||||
// a common prefix or is a regular object
|
||||
isCP := false
|
||||
cpKey := ""
|
||||
if delimiter != "" {
|
||||
suffix := strings.TrimPrefix(name, prefix)
|
||||
before, _, found := strings.Cut(suffix, delimiter)
|
||||
if found {
|
||||
isCP = true
|
||||
cpKey = prefix + before + delimiter
|
||||
}
|
||||
}
|
||||
|
||||
if isCP {
|
||||
// Skip common prefixes at or before the marker
|
||||
if cpKey <= effectiveMarker {
|
||||
continue
|
||||
}
|
||||
// Deduplicate: multiple blobs can map to the same common prefix
|
||||
if _, exists := cpSet[cpKey]; exists {
|
||||
continue
|
||||
}
|
||||
// If we already reached maxKeys, this new unique CP means truncation
|
||||
if pastMax {
|
||||
isTruncated = true
|
||||
break loop
|
||||
}
|
||||
cp := cpKey
|
||||
cPrefixes = append(cPrefixes, types.CommonPrefix{Prefix: &cp})
|
||||
cpSet[cpKey] = struct{}{}
|
||||
candidateMarker = cpKey
|
||||
totalFound++
|
||||
if totalFound == maxKeys {
|
||||
pastMax = true
|
||||
}
|
||||
} else {
|
||||
if pastMax {
|
||||
isTruncated = true
|
||||
break loop
|
||||
}
|
||||
objects = append(objects, s3response.Object{
|
||||
ETag: backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
|
||||
Key: v.Name,
|
||||
LastModified: v.Properties.LastModified,
|
||||
Size: v.Properties.ContentLength,
|
||||
StorageClass: types.ObjectStorageClassStandard,
|
||||
Owner: &types.Owner{
|
||||
ID: &acl.Owner,
|
||||
},
|
||||
})
|
||||
candidateMarker = name
|
||||
totalFound++
|
||||
if totalFound == maxKeys {
|
||||
pastMax = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return s3response.ListObjectsResult{}, err
|
||||
}
|
||||
|
||||
if !isTruncated {
|
||||
candidateMarker = ""
|
||||
var nextMarker string
|
||||
if page.isTruncated {
|
||||
nextMarker = page.lastKey
|
||||
}
|
||||
|
||||
return s3response.ListObjectsResult{
|
||||
Contents: objects,
|
||||
Contents: page.objects,
|
||||
Marker: backend.GetPtrFromString(effectiveMarker),
|
||||
MaxKeys: &maxKeys,
|
||||
Name: input.Bucket,
|
||||
NextMarker: backend.GetPtrFromString(candidateMarker),
|
||||
NextMarker: backend.GetPtrFromString(nextMarker),
|
||||
Prefix: backend.GetPtrFromString(prefix),
|
||||
IsTruncated: &isTruncated,
|
||||
IsTruncated: &page.isTruncated,
|
||||
Delimiter: backend.GetPtrFromString(delimiter),
|
||||
CommonPrefixes: cPrefixes,
|
||||
CommonPrefixes: page.commonPrefixes,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -927,11 +1025,13 @@ func (az *Azure) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input
|
||||
startAfterVal := backend.GetStringFromPtr(input.StartAfter)
|
||||
continuationTokenVal := backend.GetStringFromPtr(input.ContinuationToken)
|
||||
|
||||
// Take the lexicographically larger of startAfter and continuationToken so
|
||||
// listing starts strictly after both constraints.
|
||||
azureMarker, tokenKey := decodeAzMarkerToken(continuationTokenVal, prefix, delimiter)
|
||||
|
||||
// Take the lexicographically larger of startAfter and the continuation
|
||||
// token key so listing starts strictly after both constraints.
|
||||
effectiveMarker := startAfterVal
|
||||
if continuationTokenVal > effectiveMarker {
|
||||
effectiveMarker = continuationTokenVal
|
||||
if tokenKey > effectiveMarker {
|
||||
effectiveMarker = tokenKey
|
||||
}
|
||||
|
||||
if maxKeys == 0 {
|
||||
@@ -948,112 +1048,44 @@ func (az *Azure) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Use flat listing (empty delimiter) and handle delimiter logic client-side,
|
||||
// matching S3 semantics. Only pass Prefix and Marker to Azure.
|
||||
// effectiveMarker is passed as Marker so Azure skips blobs before it.
|
||||
pager := client.NewListBlobsHierarchyPager("", &container.ListBlobsHierarchyOptions{
|
||||
Prefix: input.Prefix,
|
||||
Marker: backend.GetPtrFromString(effectiveMarker),
|
||||
// The S3 marker is an object key, which Azure rejects as a blob listing
|
||||
// marker, so it is applied client side while Azure is paged with the marker
|
||||
// carried in the continuation token.
|
||||
page, err := az.listBlobs(ctx, client, azListingOpts{
|
||||
prefix: prefix,
|
||||
delimiter: delimiter,
|
||||
marker: effectiveMarker,
|
||||
azureMarker: azureMarker,
|
||||
maxKeys: maxKeys,
|
||||
owner: acl.Owner,
|
||||
})
|
||||
|
||||
var objects []s3response.Object
|
||||
var cPrefixes []types.CommonPrefix
|
||||
cpSet := make(map[string]struct{})
|
||||
var pastMax, isTruncated bool
|
||||
var candidateMarker string
|
||||
var totalFound int32
|
||||
|
||||
loop:
|
||||
for pager.More() {
|
||||
resp, err := pager.NextPage(ctx)
|
||||
if err != nil {
|
||||
return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
|
||||
}
|
||||
|
||||
for _, v := range resp.Segment.BlobItems {
|
||||
name := backend.GetStringFromPtr(v.Name)
|
||||
|
||||
// Filter out multipart upload blobs
|
||||
if strings.HasPrefix(name, string(metaTmpMultipartPrefix)) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Apply delimiter logic to determine if this blob contributes to
|
||||
// a common prefix or is a regular object
|
||||
isCP := false
|
||||
cpKey := ""
|
||||
if delimiter != "" {
|
||||
suffix := strings.TrimPrefix(name, prefix)
|
||||
before, _, found := strings.Cut(suffix, delimiter)
|
||||
if found {
|
||||
isCP = true
|
||||
cpKey = prefix + before + delimiter
|
||||
}
|
||||
}
|
||||
|
||||
if isCP {
|
||||
// Skip common prefixes at or before the effective marker
|
||||
if cpKey <= effectiveMarker {
|
||||
continue
|
||||
}
|
||||
// Deduplicate: multiple blobs can map to the same common prefix
|
||||
if _, exists := cpSet[cpKey]; exists {
|
||||
continue
|
||||
}
|
||||
// If we already reached maxKeys, this new unique CP means truncation
|
||||
if pastMax {
|
||||
isTruncated = true
|
||||
break loop
|
||||
}
|
||||
cp := cpKey
|
||||
cPrefixes = append(cPrefixes, types.CommonPrefix{Prefix: &cp})
|
||||
cpSet[cpKey] = struct{}{}
|
||||
candidateMarker = cpKey
|
||||
totalFound++
|
||||
if totalFound == maxKeys {
|
||||
pastMax = true
|
||||
}
|
||||
} else {
|
||||
if pastMax {
|
||||
isTruncated = true
|
||||
break loop
|
||||
}
|
||||
objects = append(objects, s3response.Object{
|
||||
ETag: backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
|
||||
Key: v.Name,
|
||||
LastModified: v.Properties.LastModified,
|
||||
Size: v.Properties.ContentLength,
|
||||
StorageClass: types.ObjectStorageClassStandard,
|
||||
Owner: &types.Owner{
|
||||
ID: &acl.Owner,
|
||||
},
|
||||
})
|
||||
candidateMarker = name
|
||||
totalFound++
|
||||
if totalFound == maxKeys {
|
||||
pastMax = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
return s3response.ListObjectsV2Result{}, err
|
||||
}
|
||||
|
||||
if !isTruncated {
|
||||
candidateMarker = ""
|
||||
var nextToken string
|
||||
if page.isTruncated {
|
||||
nextToken = encodeAzMarkerToken(azMarkerToken{
|
||||
Prefix: prefix,
|
||||
Delimiter: delimiter,
|
||||
Marker: page.resumeMarker,
|
||||
LastKey: page.lastKey,
|
||||
})
|
||||
}
|
||||
|
||||
keyCount := int32(len(objects) + len(cPrefixes))
|
||||
keyCount := int32(len(page.objects) + len(page.commonPrefixes))
|
||||
|
||||
return s3response.ListObjectsV2Result{
|
||||
Contents: objects,
|
||||
Contents: page.objects,
|
||||
ContinuationToken: backend.GetPtrFromString(continuationTokenVal),
|
||||
KeyCount: &keyCount,
|
||||
MaxKeys: &maxKeys,
|
||||
Name: input.Bucket,
|
||||
NextContinuationToken: backend.GetPtrFromString(candidateMarker),
|
||||
NextContinuationToken: backend.GetPtrFromString(nextToken),
|
||||
Prefix: backend.GetPtrFromString(prefix),
|
||||
IsTruncated: &isTruncated,
|
||||
IsTruncated: &page.isTruncated,
|
||||
Delimiter: backend.GetPtrFromString(delimiter),
|
||||
CommonPrefixes: cPrefixes,
|
||||
CommonPrefixes: page.commonPrefixes,
|
||||
StartAfter: backend.GetPtrFromString(startAfterVal),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,324 @@
|
||||
// Copyright 2026 Versity Software
|
||||
// This file is licensed under the Apache License, Version 2.0
|
||||
// (the "License"); you may not use this file except in compliance
|
||||
// with the License. You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing,
|
||||
// software distributed under the License is distributed on an
|
||||
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
// KIND, either express or implied. See the License for the
|
||||
// specific language governing permissions and limitations
|
||||
// under the License.
|
||||
|
||||
package azure
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob/container"
|
||||
"github.com/versity/versitygw/backend"
|
||||
)
|
||||
|
||||
func TestDecodeAzMarkerToken(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
token string
|
||||
prefix string
|
||||
delimiter string
|
||||
wantMarker string
|
||||
wantKey string
|
||||
}{
|
||||
{
|
||||
name: "round trip",
|
||||
token: encodeAzMarkerToken(azMarkerToken{Prefix: "test/", Delimiter: "/", Marker: "2!68!MDAwMDI4", LastKey: "test/a.js"}),
|
||||
prefix: "test/",
|
||||
delimiter: "/",
|
||||
wantMarker: "2!68!MDAwMDI4",
|
||||
wantKey: "test/a.js",
|
||||
},
|
||||
{
|
||||
name: "plain key from an older token or a hand written marker",
|
||||
token: "test/a.js",
|
||||
prefix: "test/",
|
||||
wantKey: "test/a.js",
|
||||
},
|
||||
{
|
||||
// the Azure marker belongs to the listing that produced it, so only
|
||||
// the key survives a prefix change
|
||||
name: "prefix mismatch drops the azure marker",
|
||||
token: encodeAzMarkerToken(azMarkerToken{Prefix: "test/", Marker: "2!68!MDAwMDI4", LastKey: "test/a.js"}),
|
||||
prefix: "media/",
|
||||
wantKey: "test/a.js",
|
||||
},
|
||||
{
|
||||
// the last key was collapsed under the token's delimiter, so a
|
||||
// delimiter change drops the azure marker and restarts the listing
|
||||
name: "delimiter mismatch drops the azure marker",
|
||||
token: encodeAzMarkerToken(azMarkerToken{Prefix: "test/", Delimiter: "/", Marker: "2!68!MDAwMDI4", LastKey: "test/a.js"}),
|
||||
prefix: "test/",
|
||||
delimiter: "",
|
||||
wantKey: "test/a.js",
|
||||
},
|
||||
{
|
||||
name: "corrupt token",
|
||||
token: azTokenPrefix + "!!!not base64!!!",
|
||||
prefix: "test/",
|
||||
wantKey: azTokenPrefix + "!!!not base64!!!",
|
||||
},
|
||||
{
|
||||
name: "empty token",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
marker, key := decodeAzMarkerToken(tt.token, tt.prefix, tt.delimiter)
|
||||
if marker != tt.wantMarker {
|
||||
t.Errorf("azure marker: got %q, want %q", marker, tt.wantMarker)
|
||||
}
|
||||
if key != tt.wantKey {
|
||||
t.Errorf("last key: got %q, want %q", key, tt.wantKey)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestListBlobsPagination(t *testing.T) {
|
||||
keys := make([]string, 0, 2500)
|
||||
for i := range 2500 {
|
||||
keys = append(keys, fmt.Sprintf("test/asset-%04d.js", i))
|
||||
}
|
||||
|
||||
client, srv := fakeAzureContainer(t, keys, 1000)
|
||||
|
||||
var got []string
|
||||
var marker, azureMarker string
|
||||
for page := 0; ; page++ {
|
||||
if page > 10 {
|
||||
t.Fatal("listing did not terminate")
|
||||
}
|
||||
|
||||
res, err := (&Azure{}).listBlobs(context.Background(), client, azListingOpts{
|
||||
prefix: "test/",
|
||||
marker: marker,
|
||||
azureMarker: azureMarker,
|
||||
maxKeys: 1000,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("list blobs: %v", err)
|
||||
}
|
||||
|
||||
for _, o := range res.objects {
|
||||
got = append(got, backend.GetStringFromPtr(o.Key))
|
||||
}
|
||||
if !res.isTruncated {
|
||||
break
|
||||
}
|
||||
if len(res.objects) != 1000 {
|
||||
t.Fatalf("truncated page holds %v objects, want 1000", len(res.objects))
|
||||
}
|
||||
marker, azureMarker = res.lastKey, res.resumeMarker
|
||||
}
|
||||
|
||||
if len(got) != len(keys) {
|
||||
t.Fatalf("listed %v objects, want %v", len(got), len(keys))
|
||||
}
|
||||
for i, key := range keys {
|
||||
if got[i] != key {
|
||||
t.Fatalf("object %v: got %q, want %q", i, got[i], key)
|
||||
}
|
||||
}
|
||||
|
||||
// resuming from an Azure marker must keep the number of blob listings
|
||||
// linear in the number of objects
|
||||
if srv.requests > 5 {
|
||||
t.Errorf("listing took %v azure requests, want at most 5", srv.requests)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListBlobsMarkerAndDelimiter(t *testing.T) {
|
||||
keys := []string{
|
||||
"test/a/1.js",
|
||||
"test/a/2.js",
|
||||
"test/b/1.js",
|
||||
"test/root.js",
|
||||
"test/z/1.js",
|
||||
}
|
||||
client, _ := fakeAzureContainer(t, keys, 1000)
|
||||
|
||||
res, err := (&Azure{}).listBlobs(context.Background(), client, azListingOpts{
|
||||
prefix: "test/",
|
||||
delimiter: "/",
|
||||
marker: "test/a/",
|
||||
maxKeys: 1000,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("list blobs: %v", err)
|
||||
}
|
||||
|
||||
var cps []string
|
||||
for _, cp := range res.commonPrefixes {
|
||||
cps = append(cps, backend.GetStringFromPtr(cp.Prefix))
|
||||
}
|
||||
if strings.Join(cps, ",") != "test/b/,test/z/" {
|
||||
t.Errorf("common prefixes: got %v, want [test/b/ test/z/]", cps)
|
||||
}
|
||||
if len(res.objects) != 1 || backend.GetStringFromPtr(res.objects[0].Key) != "test/root.js" {
|
||||
t.Errorf("objects: got %v, want [test/root.js]", res.objects)
|
||||
}
|
||||
if res.isTruncated {
|
||||
t.Error("listing reported as truncated")
|
||||
}
|
||||
}
|
||||
|
||||
func TestListBlobsTruncatedCommonPrefixes(t *testing.T) {
|
||||
// common prefixes deliberately span Azure page boundaries so paging must
|
||||
// resume from an Azure marker and dedupe prefixes both within a page and
|
||||
// across the truncation boundary
|
||||
keys := []string{
|
||||
"test/a/1.js",
|
||||
"test/a/2.js",
|
||||
"test/b/1.js",
|
||||
"test/c/1.js",
|
||||
"test/d/1.js",
|
||||
}
|
||||
client, _ := fakeAzureContainer(t, keys, 2)
|
||||
|
||||
var got []string
|
||||
var marker, azureMarker string
|
||||
for page := 0; ; page++ {
|
||||
if page > 10 {
|
||||
t.Fatal("listing did not terminate")
|
||||
}
|
||||
|
||||
res, err := (&Azure{}).listBlobs(context.Background(), client, azListingOpts{
|
||||
prefix: "test/",
|
||||
delimiter: "/",
|
||||
marker: marker,
|
||||
azureMarker: azureMarker,
|
||||
maxKeys: 2,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("list blobs: %v", err)
|
||||
}
|
||||
if len(res.objects) != 0 {
|
||||
t.Fatalf("expected no objects, got %v", res.objects)
|
||||
}
|
||||
|
||||
for _, cp := range res.commonPrefixes {
|
||||
got = append(got, backend.GetStringFromPtr(cp.Prefix))
|
||||
}
|
||||
if !res.isTruncated {
|
||||
break
|
||||
}
|
||||
marker, azureMarker = res.lastKey, res.resumeMarker
|
||||
}
|
||||
|
||||
want := []string{"test/a/", "test/b/", "test/c/", "test/d/"}
|
||||
if strings.Join(got, ",") != strings.Join(want, ",") {
|
||||
t.Fatalf("common prefixes: got %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListBlobsSkipsMultipartUploads(t *testing.T) {
|
||||
// multipart staging blobs must be filtered out before they count against
|
||||
// maxKeys, otherwise the listing would truncate early and hide real objects
|
||||
keys := []string{
|
||||
string(metaTmpMultipartPrefix) + "/upload-1/part",
|
||||
string(metaTmpMultipartPrefix) + "/upload-2/part",
|
||||
"a.js",
|
||||
"b.js",
|
||||
}
|
||||
client, _ := fakeAzureContainer(t, keys, 1000)
|
||||
|
||||
res, err := (&Azure{}).listBlobs(context.Background(), client, azListingOpts{
|
||||
maxKeys: 2,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("list blobs: %v", err)
|
||||
}
|
||||
|
||||
var got []string
|
||||
for _, o := range res.objects {
|
||||
got = append(got, backend.GetStringFromPtr(o.Key))
|
||||
}
|
||||
if strings.Join(got, ",") != "a.js,b.js" {
|
||||
t.Errorf("objects: got %v, want [a.js b.js]", got)
|
||||
}
|
||||
if res.isTruncated {
|
||||
t.Error("listing reported as truncated; multipart blobs counted against maxKeys")
|
||||
}
|
||||
}
|
||||
|
||||
type fakeAzureServer struct {
|
||||
requests int
|
||||
}
|
||||
|
||||
// fakeAzureContainer serves blob listings the way Azure does: paged with opaque
|
||||
// markers that only it mints, rejecting anything else with the same error real
|
||||
// Azure returns for an S3 key passed as marker.
|
||||
func fakeAzureContainer(t *testing.T, keys []string, pageSize int) (*container.Client, *fakeAzureServer) {
|
||||
t.Helper()
|
||||
|
||||
state := &fakeAzureServer{}
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
state.requests++
|
||||
|
||||
q := r.URL.Query()
|
||||
prefix := q.Get("prefix")
|
||||
|
||||
start := 0
|
||||
if marker := q.Get("marker"); marker != "" {
|
||||
idx, err := strconv.Atoi(strings.TrimPrefix(marker, "azmarker-"))
|
||||
if !strings.HasPrefix(marker, "azmarker-") || err != nil {
|
||||
w.Header().Set("x-ms-error-code", "InvalidQueryParameterValue")
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
start = idx
|
||||
}
|
||||
|
||||
var matched []string
|
||||
for _, key := range keys {
|
||||
if strings.HasPrefix(key, prefix) {
|
||||
matched = append(matched, key)
|
||||
}
|
||||
}
|
||||
|
||||
end := min(start+pageSize, len(matched))
|
||||
var nextMarker string
|
||||
if end < len(matched) {
|
||||
nextMarker = fmt.Sprintf("azmarker-%d", end)
|
||||
}
|
||||
|
||||
var body strings.Builder
|
||||
body.WriteString(`<?xml version="1.0" encoding="utf-8"?><EnumerationResults><Blobs>`)
|
||||
for _, key := range matched[start:end] {
|
||||
fmt.Fprintf(&body, `<Blob><Name>%s</Name><Properties>`+
|
||||
`<Last-Modified>Mon, 02 Jan 2006 15:04:05 GMT</Last-Modified>`+
|
||||
`<Etag>0x8DEADBEEF</Etag><Content-Length>7</Content-Length>`+
|
||||
`</Properties></Blob>`, key)
|
||||
}
|
||||
fmt.Fprintf(&body, `</Blobs><NextMarker>%s</NextMarker></EnumerationResults>`, nextMarker)
|
||||
|
||||
w.Header().Set("Content-Type", "application/xml")
|
||||
//nolint:errcheck
|
||||
w.Write([]byte(body.String()))
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
client, err := container.NewClientWithNoCredential(srv.URL+"/testbucket", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("init container client: %v", err)
|
||||
}
|
||||
|
||||
return client, state
|
||||
}
|
||||
@@ -86,7 +86,13 @@ func ListObjectsV2_both_start_after_and_continuation_token(s *S3Conf) error {
|
||||
maxKeys, out.MaxKeys)
|
||||
}
|
||||
|
||||
if getString(out.NextContinuationToken) != "bar" {
|
||||
// the azure backend returns an opaque token, as it has to carry the
|
||||
// azure blob listing marker along with the last key
|
||||
if s.azureTests {
|
||||
if getString(out.NextContinuationToken) == "" {
|
||||
return fmt.Errorf("expected non-empty NextContinuationToken")
|
||||
}
|
||||
} else if getString(out.NextContinuationToken) != "bar" {
|
||||
return fmt.Errorf("expected next-marker to be baz, instead got %v",
|
||||
getString(out.NextContinuationToken))
|
||||
}
|
||||
@@ -858,3 +864,161 @@ func ListObjectsV2_mp_masking_delimiter(s *S3Conf) error {
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// ListObjectsV2_full_pagination walks a listing that spans many pages and
|
||||
// checks that every object is returned exactly once, in order, with the
|
||||
// continuation token driving the walk. This exercises backends whose
|
||||
// continuation token is opaque (e.g. azure) rather than the last key.
|
||||
func ListObjectsV2_full_pagination(s *S3Conf) error {
|
||||
testName := "ListObjectsV2_full_pagination"
|
||||
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
|
||||
const objectCount = 25
|
||||
keys := make([]string, 0, objectCount)
|
||||
for i := 0; i < objectCount; i++ {
|
||||
keys = append(keys, fmt.Sprintf("obj-%03d", i))
|
||||
}
|
||||
contents, err := putObjects(s3client, keys, bucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
maxKeys := int32(4)
|
||||
var continuationToken *string
|
||||
var allObjects []types.Object
|
||||
seen := map[string]bool{}
|
||||
|
||||
for pages := 0; ; pages++ {
|
||||
if pages > objectCount+1 {
|
||||
return fmt.Errorf("pagination did not terminate after %v pages", pages)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
|
||||
out, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
|
||||
Bucket: &bucket,
|
||||
MaxKeys: &maxKeys,
|
||||
ContinuationToken: continuationToken,
|
||||
})
|
||||
cancel()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if int32(len(out.Contents)) > maxKeys {
|
||||
return fmt.Errorf("page returned %v objects, exceeding max-keys %v",
|
||||
len(out.Contents), maxKeys)
|
||||
}
|
||||
|
||||
for _, obj := range out.Contents {
|
||||
key := getString(obj.Key)
|
||||
if seen[key] {
|
||||
return fmt.Errorf("object %q returned on more than one page", key)
|
||||
}
|
||||
seen[key] = true
|
||||
}
|
||||
allObjects = append(allObjects, out.Contents...)
|
||||
|
||||
if out.IsTruncated != nil && *out.IsTruncated {
|
||||
if getString(out.NextContinuationToken) == "" {
|
||||
return fmt.Errorf("truncated page returned an empty NextContinuationToken")
|
||||
}
|
||||
continuationToken = out.NextContinuationToken
|
||||
continue
|
||||
}
|
||||
|
||||
// the final page must not advertise a continuation token
|
||||
if getString(out.NextContinuationToken) != "" {
|
||||
return fmt.Errorf("non-truncated page returned a NextContinuationToken %q",
|
||||
getString(out.NextContinuationToken))
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
if !compareObjects(contents, allObjects) {
|
||||
return fmt.Errorf("expected the paginated contents to be %v, instead got %v",
|
||||
contents, allObjects)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// ListObjectsV2_pagination_with_delimiter walks a delimited listing across
|
||||
// several pages and checks that common prefixes and objects are each returned
|
||||
// once and in order, with no prefix repeated across page boundaries.
|
||||
func ListObjectsV2_pagination_with_delimiter(s *S3Conf) error {
|
||||
testName := "ListObjectsV2_pagination_with_delimiter"
|
||||
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
|
||||
_, err := putObjects(s3client, []string{
|
||||
"a/1", "a/2", "b/1", "c/1", "d/1", "e/1", "root1", "root2",
|
||||
}, bucket)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
delim, maxKeys := "/", int32(2)
|
||||
var continuationToken *string
|
||||
var gotObjects []string
|
||||
var gotPrefixes []types.CommonPrefix
|
||||
seenPrefix := map[string]bool{}
|
||||
|
||||
for pages := 0; ; pages++ {
|
||||
if pages > 10 {
|
||||
return fmt.Errorf("pagination did not terminate after %v pages", pages)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
|
||||
out, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
|
||||
Bucket: &bucket,
|
||||
Delimiter: &delim,
|
||||
MaxKeys: &maxKeys,
|
||||
ContinuationToken: continuationToken,
|
||||
})
|
||||
cancel()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if kc := int32(len(out.Contents) + len(out.CommonPrefixes)); kc > maxKeys {
|
||||
return fmt.Errorf("page returned %v keys, exceeding max-keys %v", kc, maxKeys)
|
||||
}
|
||||
|
||||
for _, obj := range out.Contents {
|
||||
gotObjects = append(gotObjects, getString(obj.Key))
|
||||
}
|
||||
for _, cp := range out.CommonPrefixes {
|
||||
prefix := getString(cp.Prefix)
|
||||
if seenPrefix[prefix] {
|
||||
return fmt.Errorf("common prefix %q returned on more than one page", prefix)
|
||||
}
|
||||
seenPrefix[prefix] = true
|
||||
gotPrefixes = append(gotPrefixes, cp)
|
||||
}
|
||||
|
||||
if out.IsTruncated == nil || !*out.IsTruncated {
|
||||
break
|
||||
}
|
||||
if getString(out.NextContinuationToken) == "" {
|
||||
return fmt.Errorf("truncated page returned an empty NextContinuationToken")
|
||||
}
|
||||
continuationToken = out.NextContinuationToken
|
||||
}
|
||||
|
||||
wantPrefixes := []string{"a/", "b/", "c/", "d/", "e/"}
|
||||
if !comparePrefixes(wantPrefixes, gotPrefixes) {
|
||||
return fmt.Errorf("expected common prefixes %v, instead got %v",
|
||||
wantPrefixes, sprintPrefixes(gotPrefixes))
|
||||
}
|
||||
|
||||
wantObjects := []string{"root1", "root2"}
|
||||
if len(gotObjects) != len(wantObjects) {
|
||||
return fmt.Errorf("expected objects %v, instead got %v", wantObjects, gotObjects)
|
||||
}
|
||||
for i, key := range wantObjects {
|
||||
if gotObjects[i] != key {
|
||||
return fmt.Errorf("expected objects %v, instead got %v", wantObjects, gotObjects)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
@@ -319,6 +319,8 @@ func TestListObjectsV2(ts *TestState) {
|
||||
ts.Run(ListObjectsV2_all_objs_max_keys)
|
||||
ts.Run(ListObjectsV2_exceeding_max_keys)
|
||||
ts.Run(ListObjectsV2_list_all_objs)
|
||||
ts.Run(ListObjectsV2_full_pagination)
|
||||
ts.Run(ListObjectsV2_pagination_with_delimiter)
|
||||
ts.Run(ListObjectsV2_with_owner)
|
||||
ts.Run(ListObjectsV2_non_truncated_common_prefixes)
|
||||
//TODO: remove the condition after implementing checksums in azure
|
||||
@@ -2623,6 +2625,8 @@ func GetIntTests() IntTests {
|
||||
"ListObjectsV2_truncated_common_prefixes": ListObjectsV2_truncated_common_prefixes,
|
||||
"ListObjectsV2_all_objs_max_keys": ListObjectsV2_all_objs_max_keys,
|
||||
"ListObjectsV2_list_all_objs": ListObjectsV2_list_all_objs,
|
||||
"ListObjectsV2_full_pagination": ListObjectsV2_full_pagination,
|
||||
"ListObjectsV2_pagination_with_delimiter": ListObjectsV2_pagination_with_delimiter,
|
||||
"ListObjectsV2_with_owner": ListObjectsV2_with_owner,
|
||||
"ListObjectsV2_with_checksum": ListObjectsV2_with_checksum,
|
||||
"ListObjectVersions_VD_success": ListObjectVersions_VD_success,
|
||||
|
||||
Reference in New Issue
Block a user