mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-07-30 20:12:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: update marker/continuation token to be the azure next marker

Browse Source 
This changes the marker/continuation token from the object name
to the marker from the azure list objects pager. This is needed
because passing the object name as the token to the azure next
call causes the Azure API to throw 400 Bad Request with
InvalidQueryParameterValue. So we have to use the azure marker
for compatibility with the azure API pager.

To do this we have to align the s3 list objects request to the
Azure ListBlobsHierarchyPager. The v2 requests have an optional
startafter where we will have to page through the azure blobs
to find the correct starting point, but after this we will
only return with the single paginated results form the Azure
pager to maintain the correct markers all the way through to
Azure.

The ListObjects (non V2) assumes that the marker must be an object
name, so for this case we have to page through the azure listings
for each call to find the correct starting point. This makes the
V2 method far more efficient, but maintains correctness for the
ListObjects.

Also remove continuation token string checks in the integration
tests since this is supposed to be an opaque token that the
client should not care about. This will help to maintain the
tests for mutliple backend types.

Fixes #1457
This commit is contained in:
nitin
2025-08-25 11:28:42 -07:00
committed by Ben McClelland
parent 5d2a1527e0
commit 630651254f
3 changed files with 260 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/azure/azure.go
+138 -108
View File
@@ -68,6 +68,8 @@ const (
onameAttr key = "Objname"
onameAttrLower key = "objname"
metaTmpMultipartPrefix key = ".sgwtmp" + "/multipart"
defaultListingMaxKeys = 1000
)
func (key) Table() map[string]struct{} {
@@ -579,26 +581,6 @@ func (az *Azure) GetObjectAttributes(ctx context.Context, input *s3.GetObjectAtt
}
func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s3response.ListObjectsResult, error) {
client, err := az.getContainerClient(*input.Bucket)
if err != nil {
return s3response.ListObjectsResult{}, nil
}
pager := client.NewListBlobsHierarchyPager(*input.Delimiter, &container.ListBlobsHierarchyOptions{
Marker: input.Marker,
MaxResults: input.MaxKeys,
Prefix: input.Prefix,
})
var objects []s3response.Object
var cPrefixes []types.CommonPrefix
var nextMarker *string
var isTruncated bool
var maxKeys int32 = math.MaxInt32
if input.MaxKeys != nil {
maxKeys = *input.MaxKeys
}
// Retrieve the bucket acl to get the bucket owner
// All the objects in the bucket are owner by the bucket owner
aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
@@ -611,19 +593,49 @@ func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s
return s3response.ListObjectsResult{}, err
}
Pager:
for pager.More() {
client, err := az.getContainerClient(*input.Bucket)
if err != nil {
return s3response.ListObjectsResult{}, nil
}
var maxKeys int32 = defaultListingMaxKeys
if input.MaxKeys != nil {
maxKeys = *input.MaxKeys
}
pager := client.NewListBlobsHierarchyPager(*input.Delimiter, &container.ListBlobsHierarchyOptions{
MaxResults: &maxKeys,
Prefix: input.Prefix,
})
var objects []s3response.Object
var cPrefixes []types.CommonPrefix
var nextMarker *string
var isTruncated bool
// Convert marker to filter criteria
var markerFilter string
if input.Marker != nil && *input.Marker != "" {
markerFilter = *input.Marker
}
// Loop through pages until we have enough objects or no more pages
objectsFound := int32(0)
for pager.More() && objectsFound < maxKeys {
resp, err := pager.NextPage(ctx)
if err != nil {
return s3response.ListObjectsResult{}, azureErrToS3Err(err)
}
// Process objects from this page
var pageObjects []s3response.Object
for _, v := range resp.Segment.BlobItems {
if len(objects)+len(cPrefixes) >= int(maxKeys) {
nextMarker = objects[len(objects)-1].Key
isTruncated = true
break Pager
// Skip objects that come before or equal to marker
if markerFilter != "" && *v.Name <= markerFilter {
continue
}
objects = append(objects, s3response.Object{
pageObjects = append(pageObjects, s3response.Object{
ETag: backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
Key: v.Name,
LastModified: v.Properties.LastModified,
@@ -633,20 +645,22 @@ Pager:
ID: &acl.Owner,
},
})
}
for _, v := range resp.Segment.BlobPrefixes {
if *v.Name <= *input.Marker {
continue
}
if len(objects)+len(cPrefixes) >= int(maxKeys) {
nextMarker = cPrefixes[len(cPrefixes)-1].Prefix
isTruncated = true
break Pager
}
marker := getString(input.Marker)
pfx := strings.TrimSuffix(*v.Name, getString(input.Delimiter))
if marker != "" && strings.HasPrefix(marker, pfx) {
objectsFound++
if objectsFound >= maxKeys {
// Set next marker to the current object name for pagination
nextMarker = v.Name
isTruncated = true
break
}
}
objects = append(objects, pageObjects...)
// Process common prefixes from this page
for _, v := range resp.Segment.BlobPrefixes {
// Skip prefixes that come before or equal to marker
if markerFilter != "" && *v.Name <= markerFilter {
continue
}
@@ -654,6 +668,16 @@ Pager:
Prefix: v.Name,
})
}
// If we've reached maxKeys, break
if objectsFound >= maxKeys {
break
}
// If Azure indicates more pages but we need to continue for more objects
if resp.NextMarker != nil && *resp.NextMarker != "" && objectsFound < maxKeys {
continue
}
}
return s3response.ListObjectsResult{
@@ -670,98 +694,104 @@ Pager:
}
func (az *Azure) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input) (s3response.ListObjectsV2Result, error) {
marker := ""
if *input.ContinuationToken > *input.StartAfter {
marker = *input.ContinuationToken
} else {
marker = *input.StartAfter
// Retrieve the bucket acl to get the bucket owner
// All the objects in the bucket are owner by the bucket owner
aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
if err != nil {
return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
}
acl, err := auth.ParseACL(aclBytes)
if err != nil {
return s3response.ListObjectsV2Result{}, err
}
client, err := az.getContainerClient(*input.Bucket)
if err != nil {
return s3response.ListObjectsV2Result{}, nil
}
var maxKeys int32 = defaultListingMaxKeys
if input.MaxKeys != nil {
maxKeys = *input.MaxKeys
}
pager := client.NewListBlobsHierarchyPager(*input.Delimiter, &container.ListBlobsHierarchyOptions{
Marker: &marker,
MaxResults: input.MaxKeys,
Marker: input.ContinuationToken,
MaxResults: &maxKeys,
Prefix: input.Prefix,
})
var objects []s3response.Object
var cPrefixes []types.CommonPrefix
var nextMarker *string
var isTruncated bool
var maxKeys int32 = math.MaxInt32
var fetchOwner bool
var resp container.ListBlobsHierarchyResponse
if input.MaxKeys != nil {
maxKeys = *input.MaxKeys
}
if input.FetchOwner != nil {
fetchOwner = *input.FetchOwner
}
// Retrieve the bucket acl to get the bucket owner, if "fetchOwner" is true
// All the objects in the bucket are owner by the bucket owner
var acl auth.ACL
if fetchOwner {
aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
// Loop through pages until we find objects or no more pages
for {
resp, err = pager.NextPage(ctx)
if err != nil {
return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
}
acl, err = auth.ParseACL(aclBytes)
if err != nil {
return s3response.ListObjectsV2Result{}, err
}
}
Pager:
for pager.More() {
resp, err := pager.NextPage(ctx)
if err != nil {
return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
}
// Convert Azure objects to S3 objects
var pageObjects []s3response.Object
for _, v := range resp.Segment.BlobItems {
if len(objects)+len(cPrefixes) >= int(maxKeys) {
nextMarker = objects[len(objects)-1].Key
isTruncated = true
break Pager
}
obj := s3response.Object{
pageObjects = append(pageObjects, s3response.Object{
ETag: backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
Key: v.Name,
LastModified: v.Properties.LastModified,
Size: v.Properties.ContentLength,
StorageClass: types.ObjectStorageClassStandard,
}
if fetchOwner {
obj.Owner = &types.Owner{
Owner: &types.Owner{
ID: &acl.Owner,
}
}
objects = append(objects, obj)
}
for _, v := range resp.Segment.BlobPrefixes {
if *v.Name <= marker {
continue
}
if len(objects)+len(cPrefixes) >= int(maxKeys) {
nextMarker = cPrefixes[len(cPrefixes)-1].Prefix
isTruncated = true
break Pager
}
marker := getString(input.ContinuationToken)
pfx := strings.TrimSuffix(*v.Name, getString(input.Delimiter))
if marker != "" && strings.HasPrefix(marker, pfx) {
continue
}
cPrefixes = append(cPrefixes, types.CommonPrefix{
Prefix: v.Name,
},
})
}
// If StartAfter is specified, filter objects
if input.StartAfter != nil && *input.StartAfter != "" {
startAfter := *input.StartAfter
startIndex := -1
for i, obj := range pageObjects {
if *obj.Key > startAfter {
startIndex = i
break
}
}
if startIndex != -1 {
// Found objects after StartAfter in this page
objects = append(objects, pageObjects[startIndex:]...)
break
} else {
// No objects after StartAfter in this page
// Check if there are more pages to examine
if resp.NextMarker == nil || *resp.NextMarker == "" {
// No more pages, so no objects after StartAfter
break
}
// Continue to next page without adding any objects
continue
}
} else {
// No StartAfter specified, add all objects from this page
objects = append(objects, pageObjects...)
break
}
}
var cPrefixes []types.CommonPrefix
for _, v := range resp.Segment.BlobPrefixes {
cPrefixes = append(cPrefixes, types.CommonPrefix{
Prefix: v.Name,
})
}
var isTruncated bool
var nextMarker *string
// If Azure returned a NextMarker, set it for the next request
if resp.NextMarker != nil && *resp.NextMarker != "" {
nextMarker = resp.NextMarker
isTruncated = true
}
return s3response.ListObjectsV2Result{
tests/integration/group-tests.go
+6 -1
View File
@@ -236,7 +236,10 @@ func TestListObjects(s *S3Conf) {
func TestListObjectsV2(s *S3Conf) {
ListObjectsV2_start_after(s)
ListObjectsV2_both_start_after_and_continuation_token(s)
// posix continuation token not compatible with azure
if !s.azureTests {
ListObjectsV2_both_start_after_and_continuation_token(s)
}
ListObjectsV2_start_after_not_in_list(s)
ListObjectsV2_start_after_empty_result(s)
ListObjectsV2_both_delimiter_and_prefix(s)
@@ -246,6 +249,7 @@ func TestListObjectsV2(s *S3Conf) {
ListObjectsV2_exceeding_max_keys(s)
ListObjectsV2_list_all_objs(s)
ListObjectsV2_with_owner(s)
ListObjectsV2_non_truncated_common_prefixes(s)
//TODO: remove the condition after implementing checksums in azure
if !s.azureTests {
ListObjectsV2_with_checksum(s)
@@ -1146,6 +1150,7 @@ func GetIntTests() IntTests {
"ListObjects_nested_dir_file_objs": ListObjects_nested_dir_file_objs,
"ListObjects_check_owner": ListObjects_check_owner,
"ListObjects_non_truncated_common_prefixes": ListObjects_non_truncated_common_prefixes,
"ListObjectsV2_non_truncated_common_prefixes": ListObjectsV2_non_truncated_common_prefixes,
"ListObjects_with_checksum": ListObjects_with_checksum,
"ListObjectsV2_start_after": ListObjectsV2_start_after,
"ListObjectsV2_both_start_after_and_continuation_token": ListObjectsV2_both_start_after_and_continuation_token,
tests/integration/tests.go
+116 -5
View File
@@ -5142,6 +5142,7 @@ func ListObjects_list_all_objs(s *S3Conf) error {
return err
}
// Test 1: List all objects without pagination
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
out, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
Bucket: &bucket,
@@ -5173,6 +5174,36 @@ func ListObjects_list_all_objs(s *S3Conf) error {
contents, out.Contents)
}
// Test 2: List all objects with pagination using ListObjectsV2
var marker *string
var allObjects []types.Object
maxKeys := int32(2)
for {
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
out, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
Bucket: &bucket,
MaxKeys: &maxKeys,
Marker: marker,
})
cancel()
if err != nil {
return err
}
allObjects = append(allObjects, out.Contents...)
if out.NextMarker == nil || !*out.IsTruncated {
break
}
marker = out.NextMarker
}
if !compareObjects(contents, allObjects) {
return fmt.Errorf("expected the contents to be %v, instead got %v",
contents, allObjects)
}
return nil
})
}
@@ -5585,10 +5616,6 @@ func ListObjectsV2_truncated_common_prefixes(s *S3Conf) error {
return fmt.Errorf("expected the max-keys to be %v, instead got %v",
maxKeys, *out.MaxKeys)
}
if getString(out.NextContinuationToken) != "d3/" {
return fmt.Errorf("expected the NextContinuationToken to be d3/, instead got %v",
getString(out.NextContinuationToken))
}
if getString(out.Delimiter) != delim {
return fmt.Errorf("expected the delimiter to be %v, instead got %v",
delim, getString(out.Delimiter))
@@ -5618,6 +5645,59 @@ func ListObjectsV2_truncated_common_prefixes(s *S3Conf) error {
})
}
func ListObjectsV2_non_truncated_common_prefixes(s *S3Conf) error {
testName := "ListObjectsV2_non_truncated_common_prefixes"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
_, err := putObjects(s3client, []string{"asdf", "boo/bar", "boo/baz/xyzzy", "cquux/thud", "cquux/bla"}, bucket)
if err != nil {
return err
}
delim, marker, maxKeys := "/", "boo/", int32(1)
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
res, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
Bucket: &bucket,
StartAfter: &marker,
Delimiter: &delim,
MaxKeys: &maxKeys,
})
cancel()
if err != nil {
return err
}
if res.IsTruncated == nil {
return fmt.Errorf("expected non-nil istruncated")
}
if *res.IsTruncated {
return fmt.Errorf("expected non-truncated result")
}
if res.MaxKeys == nil {
return fmt.Errorf("expected non nil max-keys")
}
if *res.MaxKeys != maxKeys {
return fmt.Errorf("expected max-keys to be %v, instead got %v",
maxKeys, *res.MaxKeys)
}
if getString(res.Delimiter) != delim {
return fmt.Errorf("expected delimiter to be %v, instead got %v",
delim, getString(res.Delimiter))
}
if len(res.Contents) != 0 {
return fmt.Errorf("expected empty contents, instead got %+v",
res.Contents)
}
cPrefs := []string{"cquux/"}
if !comparePrefixes(cPrefs, res.CommonPrefixes) {
return fmt.Errorf("expected common prefixes to be %v, instead got %+v",
cPrefs, res.CommonPrefixes)
}
return nil
})
}
func ListObjectsV2_all_objs_max_keys(s *S3Conf) error {
testName := "ListObjectsV2_all_objs_max_keys"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5691,11 +5771,12 @@ func ListObjectsV2_exceeding_max_keys(s *S3Conf) error {
func ListObjectsV2_list_all_objs(s *S3Conf) error {
testName := "ListObjectsV2_list_all_objs"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
contents, err := putObjects(s3client, []string{"bar", "baz", "foo", "obj1", "hell/", "xyzz/quxx"}, bucket)
contents, err := putObjects(s3client, []string{"a", "aa", "aaa", "aaaa", "bar", "baz", "foo", "obj1", "hello/world", "xyzz/quxx"}, bucket)
if err != nil {
return err
}
// Test 1: List all objects without pagination
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
out, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
Bucket: &bucket,
@@ -5731,6 +5812,36 @@ func ListObjectsV2_list_all_objs(s *S3Conf) error {
contents, out.Contents)
}
// Test 2: List all objects with pagination using ListObjectsV2
var continuationToken *string
var allObjects []types.Object
maxKeys := int32(2)
for {
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
}
allObjects = append(allObjects, out.Contents...)
if out.NextContinuationToken == nil || !*out.IsTruncated {
break
}
continuationToken = out.NextContinuationToken
}
if !compareObjects(contents, allObjects) {
return fmt.Errorf("expected the paginated contents to be %v, instead got %v",
contents, allObjects)
}
return nil
})
}