mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-07-20 15:02:27 +00:00
s3: keep listing when empty directories fill the listing window (#10280)
* s3: keep listing when empty directories fill the listing window doListFilerEntries issued a single ListEntries request per directory with Limit = maxKeys+2. Entries that emit nothing - empty directories, the .uploads folder, the marker echo - consume that window without consuming maxKeys, so a bucket whose first window held only empty directories was reported empty and not truncated, and a subdirectory whose window filled up silently dropped the entries behind it. Keep requesting from the last received entry until the quota is filled or a short window shows the directory is exhausted. * s3: test listing across windows of empty directories The test filer client now honors StartFromFileName so repeated windows advance like the real filer. * s3: propagate the request context into directory listing RPCs A disconnected client now cancels the ListEntries streams instead of letting the listing keep issuing requests against the filer. * s3: group per-directory listing parameters into a request struct doListFilerEntries took ten positional parameters; call sites read as string and bool soup. Wrap the per-directory arguments in listDirectoryRequest so recursions and tests name what they pass. * s3: group list request parameters into a struct listFilerEntries took eight positional parameters ending in two bare booleans. Wrap them in listObjectsRequest so the V1 and V2 handlers name what they pass.
This commit is contained in:
@@ -109,7 +109,15 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
|
||||
// Adjust marker if it ends with delimiter to skip all entries with that prefix
|
||||
marker = adjustMarkerForDelimiter(marker, delimiter)
|
||||
|
||||
response, err := s3a.listFilerEntries(r.Context(), bucket, originalPrefix, maxKeys, marker, delimiter, encodingTypeUrl, fetchOwner)
|
||||
response, err := s3a.listFilerEntries(r.Context(), listObjectsRequest{
|
||||
bucket: bucket,
|
||||
prefix: originalPrefix,
|
||||
marker: marker,
|
||||
delimiter: delimiter,
|
||||
maxKeys: maxKeys,
|
||||
encodingTypeUrl: encodingTypeUrl,
|
||||
fetchOwner: fetchOwner,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
@@ -173,7 +181,15 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
|
||||
// Adjust marker if it ends with delimiter to skip all entries with that prefix
|
||||
marker = adjustMarkerForDelimiter(marker, delimiter)
|
||||
|
||||
response, err := s3a.listFilerEntries(r.Context(), bucket, originalPrefix, uint16(maxKeys), marker, delimiter, encodingTypeUrl, true)
|
||||
response, err := s3a.listFilerEntries(r.Context(), listObjectsRequest{
|
||||
bucket: bucket,
|
||||
prefix: originalPrefix,
|
||||
marker: marker,
|
||||
delimiter: delimiter,
|
||||
maxKeys: uint16(maxKeys),
|
||||
encodingTypeUrl: encodingTypeUrl,
|
||||
fetchOwner: true,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
@@ -232,7 +248,20 @@ func sanitizeV1MarkerEcho(response *ListBucketResult, marker string, encodingTyp
|
||||
}
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, bucket string, originalPrefix string, maxKeys uint16, originalMarker string, delimiter string, encodingTypeUrl bool, fetchOwner bool) (response ListBucketResult, err error) {
|
||||
type listObjectsRequest struct {
|
||||
bucket string
|
||||
prefix string
|
||||
marker string
|
||||
delimiter string
|
||||
maxKeys uint16
|
||||
encodingTypeUrl bool
|
||||
fetchOwner bool
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, req listObjectsRequest) (response ListBucketResult, err error) {
|
||||
bucket, originalPrefix, originalMarker := req.bucket, req.prefix, req.marker
|
||||
maxKeys, delimiter := req.maxKeys, req.delimiter
|
||||
encodingTypeUrl, fetchOwner := req.encodingTypeUrl, req.fetchOwner
|
||||
// convert full path prefix into directory name and prefix for entry name
|
||||
requestDir, prefix, marker := normalizePrefixMarker(originalPrefix, originalMarker)
|
||||
bucketPrefix := s3a.bucketPrefix(bucket)
|
||||
@@ -298,7 +327,7 @@ func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, bucket string, ori
|
||||
for {
|
||||
empty := true
|
||||
|
||||
nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, cursor, marker, delimiter, false, bucket, func(dir string, entry *filer_pb.Entry) {
|
||||
nextMarker, doErr = s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: reqDir, prefix: prefix, marker: marker, delimiter: delimiter, bucket: bucket}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
empty = false
|
||||
dirName, entryName, _ := entryUrlEncode(dir, entry.Name, encodingTypeUrl)
|
||||
if entry.IsDirectory {
|
||||
@@ -557,7 +586,18 @@ func buildTruncatedNextMarker(requestDir, prefix, nextMarker string, lastEntryWa
|
||||
return nextMarker
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, cursor *ListingCursor, marker, delimiter string, inclusiveStartFrom bool, bucket string, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
|
||||
type listDirectoryRequest struct {
|
||||
dir string
|
||||
prefix string
|
||||
marker string
|
||||
delimiter string
|
||||
bucket string
|
||||
inclusiveStartFrom bool
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) doListFilerEntries(ctx context.Context, client filer_pb.SeaweedFilerClient, req listDirectoryRequest, cursor *ListingCursor, eachEntryFn func(dir string, entry *filer_pb.Entry)) (nextMarker string, err error) {
|
||||
dir, prefix, bucket := req.dir, req.prefix, req.bucket
|
||||
marker, delimiter, inclusiveStartFrom := req.marker, req.delimiter, req.inclusiveStartFrom
|
||||
// invariants
|
||||
// prefix and marker should be under dir, marker may contain "/"
|
||||
// maxKeys should be updated for each recursion
|
||||
@@ -571,7 +611,7 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
|
||||
if strings.Contains(marker, "/") {
|
||||
subDir, subMarker := toParentAndDescendants(marker)
|
||||
// println("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker)
|
||||
subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+subDir, "", cursor, subMarker, delimiter, false, bucket, eachEntryFn)
|
||||
subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + subDir, marker: subMarker, delimiter: delimiter, bucket: bucket}, cursor, eachEntryFn)
|
||||
if subErr != nil {
|
||||
err = subErr
|
||||
return
|
||||
@@ -585,156 +625,170 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
|
||||
}
|
||||
|
||||
// now marker is also a direct child of dir
|
||||
request := &filer_pb.ListEntriesRequest{
|
||||
Directory: dir,
|
||||
Prefix: prefix,
|
||||
Limit: uint32(cursor.maxKeys) + 2, // bucket root directory needs to skip additional s3_constants.MultipartUploadsFolder folder
|
||||
StartFromFileName: marker,
|
||||
InclusiveStartFrom: inclusiveStartFrom,
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
stream, listErr := client.ListEntries(ctx, request)
|
||||
if listErr != nil {
|
||||
if errors.Is(listErr, filer_pb.ErrNotFound) {
|
||||
return
|
||||
}
|
||||
err = fmt.Errorf("list entries %+v: %w", request, listErr)
|
||||
return
|
||||
}
|
||||
|
||||
// Entries that emit nothing (empty directories, the .uploads folder, the marker
|
||||
// echo) consume the request window without consuming maxKeys, so one window may
|
||||
// end before maxKeys is satisfied. Keep requesting from the last received entry
|
||||
// until the quota is filled or a short window shows the directory is exhausted.
|
||||
for {
|
||||
resp, recvErr := stream.Recv()
|
||||
if recvErr != nil {
|
||||
if recvErr == io.EOF {
|
||||
break
|
||||
} else {
|
||||
err = fmt.Errorf("iterating entries %+v: %v", request, recvErr)
|
||||
request := &filer_pb.ListEntriesRequest{
|
||||
Directory: dir,
|
||||
Prefix: prefix,
|
||||
Limit: uint32(cursor.maxKeys) + 2, // bucket root directory needs to skip additional s3_constants.MultipartUploadsFolder folder
|
||||
StartFromFileName: marker,
|
||||
InclusiveStartFrom: inclusiveStartFrom,
|
||||
}
|
||||
|
||||
stream, listErr := client.ListEntries(ctx, request)
|
||||
if listErr != nil {
|
||||
if errors.Is(listErr, filer_pb.ErrNotFound) {
|
||||
return
|
||||
}
|
||||
}
|
||||
entry := resp.Entry
|
||||
if entry == nil {
|
||||
continue
|
||||
}
|
||||
// listFilerEntries always calls doListFilerEntries with inclusiveStartFrom=false
|
||||
// (S3 marker semantics are exclusive), but keep the guard explicit to preserve
|
||||
// behavior if inclusive callers are introduced in the future.
|
||||
if !inclusiveStartFrom && marker != "" && entry.Name == marker {
|
||||
continue
|
||||
err = fmt.Errorf("list entries %+v: %w", request, listErr)
|
||||
return
|
||||
}
|
||||
|
||||
if cursor.maxKeys <= 0 {
|
||||
cursor.isTruncated = true
|
||||
break
|
||||
}
|
||||
|
||||
// Set nextMarker only when we have quota to process this entry
|
||||
nextMarker = entry.Name
|
||||
// Track whether this entry is the exact directory targeted by a trailing-slash prefix
|
||||
// (e.g., prefix "foo" from original prefix "foo/"). After recursing into this directory,
|
||||
// we must stop processing siblings to avoid matching unrelated entries like "foo1000".
|
||||
matchedPrefixDir := cursor.prefixEndsOnDelimiter && entry.Name == prefix && entry.IsDirectory
|
||||
if cursor.prefixEndsOnDelimiter {
|
||||
if entry.Name == prefix && entry.IsDirectory {
|
||||
if delimiter != "/" {
|
||||
cursor.prefixEndsOnDelimiter = false
|
||||
}
|
||||
} else {
|
||||
continue
|
||||
}
|
||||
}
|
||||
if entry.IsDirectory {
|
||||
// glog.V(4).Infof("List Dir Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
|
||||
if entry.Name == s3_constants.MultipartUploadsFolder { // FIXME no need to apply to all directories. this extra also affects maxKeys
|
||||
continue
|
||||
}
|
||||
|
||||
// Process .versions directories immediately to create logical versioned object entries
|
||||
// These directories are never traversed (we continue here), so each is only encountered once
|
||||
if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
|
||||
// Extract object name from .versions directory name
|
||||
baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder)
|
||||
// Construct full object path relative to bucket
|
||||
bucketFullPath := s3a.bucketDir(bucket)
|
||||
bucketRelativePath := strings.TrimPrefix(dir, bucketFullPath)
|
||||
bucketRelativePath = strings.TrimPrefix(bucketRelativePath, "/")
|
||||
var fullObjectPath string
|
||||
if bucketRelativePath == "" {
|
||||
fullObjectPath = baseObjectName
|
||||
var entriesReceived uint32
|
||||
var lastEntryName string
|
||||
for {
|
||||
resp, recvErr := stream.Recv()
|
||||
if recvErr != nil {
|
||||
if recvErr == io.EOF {
|
||||
break
|
||||
} else {
|
||||
fullObjectPath = bucketRelativePath + "/" + baseObjectName
|
||||
}
|
||||
// Use metadata from the already-fetched .versions directory entry
|
||||
if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, entry); err == nil {
|
||||
eachEntryFn(dir, latestVersionEntry)
|
||||
} else if !errors.Is(err, ErrDeleteMarker) {
|
||||
// Log unexpected errors (delete markers are expected)
|
||||
glog.V(2).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err)
|
||||
err = fmt.Errorf("iterating entries %+v: %v", request, recvErr)
|
||||
return
|
||||
}
|
||||
}
|
||||
entry := resp.Entry
|
||||
if entry == nil {
|
||||
continue
|
||||
}
|
||||
entriesReceived++
|
||||
lastEntryName = entry.Name
|
||||
// listFilerEntries always calls doListFilerEntries with inclusiveStartFrom=false
|
||||
// (S3 marker semantics are exclusive), but keep the guard explicit to preserve
|
||||
// behavior if inclusive callers are introduced in the future.
|
||||
if !inclusiveStartFrom && marker != "" && entry.Name == marker {
|
||||
continue
|
||||
}
|
||||
|
||||
if delimiter != "/" || cursor.prefixEndsOnDelimiter {
|
||||
// A trailing-slash prefix (e.g. "logs/") names one directory and asks
|
||||
// whether it exists, so a real but empty directory must be reported for
|
||||
// that probe.
|
||||
explicitDirProbe := cursor.prefixEndsOnDelimiter
|
||||
if cursor.prefixEndsOnDelimiter {
|
||||
cursor.prefixEndsOnDelimiter = false
|
||||
if cursor.maxKeys <= 0 {
|
||||
cursor.isTruncated = true
|
||||
break
|
||||
}
|
||||
|
||||
// Set nextMarker only when we have quota to process this entry
|
||||
nextMarker = entry.Name
|
||||
// Track whether this entry is the exact directory targeted by a trailing-slash prefix
|
||||
// (e.g., prefix "foo" from original prefix "foo/"). After recursing into this directory,
|
||||
// we must stop processing siblings to avoid matching unrelated entries like "foo1000".
|
||||
matchedPrefixDir := cursor.prefixEndsOnDelimiter && entry.Name == prefix && entry.IsDirectory
|
||||
if cursor.prefixEndsOnDelimiter {
|
||||
if entry.Name == prefix && entry.IsDirectory {
|
||||
if delimiter != "/" {
|
||||
cursor.prefixEndsOnDelimiter = false
|
||||
}
|
||||
} else {
|
||||
continue
|
||||
}
|
||||
isKeyObject := entry.IsDirectoryKeyObject()
|
||||
if isKeyObject {
|
||||
// Directory key objects (created via PutObject with trailing "/")
|
||||
// must appear as regular keys in recursive listing mode.
|
||||
}
|
||||
if entry.IsDirectory {
|
||||
// glog.V(4).Infof("List Dir Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
|
||||
if entry.Name == s3_constants.MultipartUploadsFolder { // FIXME no need to apply to all directories. this extra also affects maxKeys
|
||||
continue
|
||||
}
|
||||
|
||||
// Process .versions directories immediately to create logical versioned object entries
|
||||
// These directories are never traversed (we continue here), so each is only encountered once
|
||||
if strings.HasSuffix(entry.Name, s3_constants.VersionsFolder) {
|
||||
// Extract object name from .versions directory name
|
||||
baseObjectName := strings.TrimSuffix(entry.Name, s3_constants.VersionsFolder)
|
||||
// Construct full object path relative to bucket
|
||||
bucketFullPath := s3a.bucketDir(bucket)
|
||||
bucketRelativePath := strings.TrimPrefix(dir, bucketFullPath)
|
||||
bucketRelativePath = strings.TrimPrefix(bucketRelativePath, "/")
|
||||
var fullObjectPath string
|
||||
if bucketRelativePath == "" {
|
||||
fullObjectPath = baseObjectName
|
||||
} else {
|
||||
fullObjectPath = bucketRelativePath + "/" + baseObjectName
|
||||
}
|
||||
// Use metadata from the already-fetched .versions directory entry
|
||||
if latestVersionEntry, err := s3a.getLatestVersionEntryFromDirectoryEntry(bucket, fullObjectPath, entry); err == nil {
|
||||
eachEntryFn(dir, latestVersionEntry)
|
||||
} else if !errors.Is(err, ErrDeleteMarker) {
|
||||
// Log unexpected errors (delete markers are expected)
|
||||
glog.V(2).Infof("Skipping versioned object %s due to error: %v", fullObjectPath, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
if delimiter != "/" || cursor.prefixEndsOnDelimiter {
|
||||
// A trailing-slash prefix (e.g. "logs/") names one directory and asks
|
||||
// whether it exists, so a real but empty directory must be reported for
|
||||
// that probe.
|
||||
explicitDirProbe := cursor.prefixEndsOnDelimiter
|
||||
if cursor.prefixEndsOnDelimiter {
|
||||
cursor.prefixEndsOnDelimiter = false
|
||||
}
|
||||
isKeyObject := entry.IsDirectoryKeyObject()
|
||||
if isKeyObject {
|
||||
// Directory key objects (created via PutObject with trailing "/")
|
||||
// must appear as regular keys in recursive listing mode.
|
||||
eachEntryFn(dir, entry)
|
||||
}
|
||||
// Recurse into subdirectory to list any children, noting whether the
|
||||
// subtree produced any entries.
|
||||
childEmitted := false
|
||||
subNextMarker, subErr := s3a.doListFilerEntries(ctx, client, listDirectoryRequest{dir: dir + "/" + entry.Name, delimiter: delimiter, bucket: bucket}, cursor, func(d string, e *filer_pb.Entry) {
|
||||
childEmitted = true
|
||||
eachEntryFn(d, e)
|
||||
})
|
||||
if subErr != nil {
|
||||
err = fmt.Errorf("doListFilerEntries2: %w", subErr)
|
||||
return
|
||||
}
|
||||
// A real but empty directory (created out of band via mount, mkdir or
|
||||
// the filer API, so it carries no MIME) is otherwise invisible to S3
|
||||
// clients that detect directories by listing it under its own "<dir>/"
|
||||
// prefix. Surface it as a directory marker for that explicit probe,
|
||||
// identical to a directory created via PutObject with a trailing "/", so
|
||||
// tools like hadoop-aws can find it. Plain listings are left untouched, so
|
||||
// empty directories left behind by deleted objects are not shown as keys.
|
||||
if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && entry.Attributes != nil {
|
||||
entry.Attributes.Mime = s3_constants.FolderMimeType
|
||||
eachEntryFn(dir, entry)
|
||||
}
|
||||
// println("doListFilerEntries2 dir", dir+"/"+entry.Name, "subNextMarker", subNextMarker)
|
||||
nextMarker = entry.Name + "/" + subNextMarker
|
||||
if cursor.isTruncated {
|
||||
return
|
||||
}
|
||||
if matchedPrefixDir {
|
||||
return
|
||||
}
|
||||
// println("doListFilerEntries2 nextMarker", nextMarker)
|
||||
} else {
|
||||
eachEntryFn(dir, entry)
|
||||
}
|
||||
// Recurse into subdirectory to list any children, noting whether the
|
||||
// subtree produced any entries.
|
||||
childEmitted := false
|
||||
subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", cursor, "", delimiter, false, bucket, func(d string, e *filer_pb.Entry) {
|
||||
childEmitted = true
|
||||
eachEntryFn(d, e)
|
||||
})
|
||||
if subErr != nil {
|
||||
err = fmt.Errorf("doListFilerEntries2: %w", subErr)
|
||||
return
|
||||
}
|
||||
// A real but empty directory (created out of band via mount, mkdir or
|
||||
// the filer API, so it carries no MIME) is otherwise invisible to S3
|
||||
// clients that detect directories by listing it under its own "<dir>/"
|
||||
// prefix. Surface it as a directory marker for that explicit probe,
|
||||
// identical to a directory created via PutObject with a trailing "/", so
|
||||
// tools like hadoop-aws can find it. Plain listings are left untouched, so
|
||||
// empty directories left behind by deleted objects are not shown as keys.
|
||||
if explicitDirProbe && !isKeyObject && !childEmitted && !cursor.isTruncated && entry.Attributes != nil {
|
||||
entry.Attributes.Mime = s3_constants.FolderMimeType
|
||||
eachEntryFn(dir, entry)
|
||||
}
|
||||
// println("doListFilerEntries2 dir", dir+"/"+entry.Name, "subNextMarker", subNextMarker)
|
||||
nextMarker = entry.Name + "/" + subNextMarker
|
||||
if cursor.isTruncated {
|
||||
return
|
||||
}
|
||||
if matchedPrefixDir {
|
||||
return
|
||||
}
|
||||
// println("doListFilerEntries2 nextMarker", nextMarker)
|
||||
} else {
|
||||
eachEntryFn(dir, entry)
|
||||
// glog.V(4).Infof("List File Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
|
||||
}
|
||||
if cursor.prefixEndsOnDelimiter {
|
||||
cursor.prefixEndsOnDelimiter = false
|
||||
}
|
||||
} else {
|
||||
eachEntryFn(dir, entry)
|
||||
// glog.V(4).Infof("List File Entries %s, file: %s, maxKeys %d", dir, entry.Name, cursor.maxKeys)
|
||||
}
|
||||
if cursor.prefixEndsOnDelimiter {
|
||||
cursor.prefixEndsOnDelimiter = false
|
||||
}
|
||||
}
|
||||
|
||||
// Versioned directories are processed above (lines 524-546)
|
||||
return
|
||||
if cursor.isTruncated || entriesReceived < request.Limit {
|
||||
return
|
||||
}
|
||||
marker = lastEntryName
|
||||
inclusiveStartFrom = false
|
||||
}
|
||||
}
|
||||
|
||||
func getListObjectsV2Args(values url.Values) (prefix, startAfter, delimiter string, token OptionalString, encodingTypeUrl bool, fetchOwner bool, maxkeys uint16, allowUnordered bool, errCode s3err.ErrorCode) {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package s3api
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
@@ -38,14 +39,14 @@ func TestDirectoryListedAsCommonPrefix(t *testing.T) {
|
||||
var seenFiles []string
|
||||
|
||||
_, err := s3a.doListFilerEntries(
|
||||
context.Background(),
|
||||
client,
|
||||
"/buckets/xoa-bucket/xo-vm-backups/data",
|
||||
"", // prefix
|
||||
listDirectoryRequest{
|
||||
dir: "/buckets/xoa-bucket/xo-vm-backups/data",
|
||||
delimiter: "/", // should yield directory for CommonPrefix processing
|
||||
bucket: "xoa-bucket",
|
||||
},
|
||||
cursor,
|
||||
"", // marker
|
||||
"/", // delimiter="/" - should yield directory for CommonPrefix processing
|
||||
false,
|
||||
"xoa-bucket",
|
||||
func(dir string, entry *filer_pb.Entry) {
|
||||
if entry.IsDirectory {
|
||||
seenDirs = append(seenDirs, entry.Name)
|
||||
@@ -88,7 +89,7 @@ func TestEmptyDirectorySurfacedAsMarker(t *testing.T) {
|
||||
// Mirrors the getFileStatus probe: prefix "logs/" with delimiter "/".
|
||||
cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
|
||||
var seen []*filer_pb.Entry
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/test", "logs", cursor, "", "/", false, "test",
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", prefix: "logs", delimiter: "/", bucket: "test"}, cursor,
|
||||
func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, entry)
|
||||
})
|
||||
@@ -129,7 +130,7 @@ func TestNonEmptyDirectoryGetsNoPhantomMarker(t *testing.T) {
|
||||
// Recursive listing under the "data/" prefix.
|
||||
cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
|
||||
var seen []string
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/test", "data", cursor, "", "", false, "test",
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", prefix: "data", bucket: "test"}, cursor,
|
||||
func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, entry.Name)
|
||||
})
|
||||
@@ -173,7 +174,7 @@ func TestEmptyDirectoryHiddenInFlatListing(t *testing.T) {
|
||||
// Plain flat listing: no prefix, no delimiter.
|
||||
cursor := &ListingCursor{maxKeys: 1000}
|
||||
var seen []string
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/test", "", cursor, "", "", false, "test",
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test", bucket: "test"}, cursor,
|
||||
func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, entry.Name)
|
||||
})
|
||||
|
||||
@@ -44,8 +44,7 @@ type testFilerClient struct {
|
||||
|
||||
func (c *testFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) {
|
||||
entries := c.entriesByDir[in.Directory]
|
||||
// Simplified mock: implements basic prefix filtering but ignores Limit, StartFromFileName, and InclusiveStartFrom
|
||||
// to keep test logic focused. Prefix "/" is treated as no filter for bucket root compatibility.
|
||||
// Simplified mock: prefix "/" is treated as no filter for bucket root compatibility.
|
||||
if in.Prefix != "" && in.Prefix != "/" {
|
||||
filtered := make([]*filer_pb.Entry, 0)
|
||||
for _, e := range entries {
|
||||
@@ -56,7 +55,16 @@ func (c *testFilerClient) ListEntries(ctx context.Context, in *filer_pb.ListEntr
|
||||
entries = filtered
|
||||
}
|
||||
|
||||
// Respect Limit
|
||||
if in.StartFromFileName != "" {
|
||||
filtered := make([]*filer_pb.Entry, 0)
|
||||
for _, e := range entries {
|
||||
if e.Name > in.StartFromFileName || (in.InclusiveStartFrom && e.Name == in.StartFromFileName) {
|
||||
filtered = append(filtered, e)
|
||||
}
|
||||
}
|
||||
entries = filtered
|
||||
}
|
||||
|
||||
if in.Limit > 0 && int(in.Limit) < len(entries) {
|
||||
entries = entries[:in.Limit]
|
||||
}
|
||||
@@ -300,7 +308,7 @@ func TestDoListFilerEntries_BucketRootPrefixSlashDelimiterSlash_ListsDirectories
|
||||
|
||||
cursor := &ListingCursor{maxKeys: 1000}
|
||||
seen := make([]string, 0)
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "/", cursor, "", "/", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", prefix: "/", delimiter: "/", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if entry.IsDirectory {
|
||||
seen = append(seen, entry.Name)
|
||||
}
|
||||
@@ -322,7 +330,7 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEcho(t *testing.T) {
|
||||
|
||||
cursor := &ListingCursor{maxKeys: 1000}
|
||||
var seen []string
|
||||
nextMarker, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "", cursor, "test.txt", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: "test.txt", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, entry.Name)
|
||||
})
|
||||
|
||||
@@ -346,7 +354,7 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEchoWithSubsequentEntries(t
|
||||
|
||||
cursor := &ListingCursor{maxKeys: 1000}
|
||||
var seen []string
|
||||
nextMarker, err := s3a.doListFilerEntries(client, "/buckets/test-bucket", "", cursor, "test.txt", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: "test.txt", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, entry.Name)
|
||||
})
|
||||
|
||||
@@ -355,6 +363,116 @@ func TestDoListFilerEntries_ExclusiveStartSkipsMarkerEchoWithSubsequentEntries(t
|
||||
assert.Equal(t, "zebra.txt", nextMarker)
|
||||
}
|
||||
|
||||
func TestDoListFilerEntries_EmptyDirectoriesDoNotHideLaterEntries(t *testing.T) {
|
||||
// With maxKeys=1 the listing window is 3 entries. The first window holds only
|
||||
// empty directories, so the real object further down must still be found.
|
||||
s3a := &S3ApiServer{}
|
||||
client := &testFilerClient{
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/test-bucket": {
|
||||
{Name: "00-empty-1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-2", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-3", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-4", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-5", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "zz-real", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
"/buckets/test-bucket/zz-real": {
|
||||
{Name: "file.log", Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
cursor := &ListingCursor{maxKeys: 1}
|
||||
var seen []string
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, dir+"/"+entry.Name)
|
||||
cursor.maxKeys--
|
||||
})
|
||||
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, []string{"/buckets/test-bucket/zz-real/file.log"}, seen)
|
||||
assert.False(t, cursor.isTruncated)
|
||||
}
|
||||
|
||||
func TestDoListFilerEntries_EmptyDirectoriesInSubdirectoryDoNotHideSiblings(t *testing.T) {
|
||||
// The recursion into "logs" gets its own listing window; empty subdirectories
|
||||
// filling that window must not skip the object behind them.
|
||||
s3a := &S3ApiServer{}
|
||||
client := &testFilerClient{
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/test-bucket": {
|
||||
{Name: "logs", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
"/buckets/test-bucket/logs": {
|
||||
{Name: "00-empty-1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-2", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-3", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-4", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-5", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "zz.log", Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
cursor := &ListingCursor{maxKeys: 1}
|
||||
var seen []string
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, dir+"/"+entry.Name)
|
||||
cursor.maxKeys--
|
||||
})
|
||||
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, []string{"/buckets/test-bucket/logs/zz.log"}, seen)
|
||||
assert.False(t, cursor.isTruncated)
|
||||
}
|
||||
|
||||
func TestDoListFilerEntries_TruncationAcrossEmptyDirectories(t *testing.T) {
|
||||
// Two real objects separated by empty directories: the first fills the quota
|
||||
// and the listing must report truncation with a resumable marker.
|
||||
s3a := &S3ApiServer{}
|
||||
client := &testFilerClient{
|
||||
entriesByDir: map[string][]*filer_pb.Entry{
|
||||
"/buckets/test-bucket": {
|
||||
{Name: "00-empty-1", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-2", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "00-empty-3", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "mm-real", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "pp-empty", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
{Name: "zz-real", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
"/buckets/test-bucket/mm-real": {
|
||||
{Name: "file.log", Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
"/buckets/test-bucket/zz-real": {
|
||||
{Name: "file.log", Attributes: &filer_pb.FuseAttributes{}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
cursor := &ListingCursor{maxKeys: 1}
|
||||
var seen []string
|
||||
nextMarker, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, dir+"/"+entry.Name)
|
||||
cursor.maxKeys--
|
||||
})
|
||||
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, []string{"/buckets/test-bucket/mm-real/file.log"}, seen)
|
||||
assert.True(t, cursor.isTruncated)
|
||||
|
||||
cursor = &ListingCursor{maxKeys: 1}
|
||||
seen = nil
|
||||
_, err = s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/test-bucket", marker: nextMarker, bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
seen = append(seen, dir+"/"+entry.Name)
|
||||
cursor.maxKeys--
|
||||
})
|
||||
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, []string{"/buckets/test-bucket/zz-real/file.log"}, seen)
|
||||
assert.False(t, cursor.isTruncated)
|
||||
}
|
||||
|
||||
func TestAllowUnorderedWithDelimiterValidation(t *testing.T) {
|
||||
t.Run("should return error when allow-unordered=true and delimiter are both present", func(t *testing.T) {
|
||||
// Create a request with both allow-unordered=true and delimiter
|
||||
@@ -765,7 +883,7 @@ func TestListObjectsV2_Regression(t *testing.T) {
|
||||
// Call doListFilerEntries directly to unit test listing logic in isolation,
|
||||
// simulating parameters passed from listFilerEntries for prefix "reports/".
|
||||
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/reports", "reports", cursor, "", "", false, "reports", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/reports", prefix: "reports", bucket: "reports"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if !entry.IsDirectory {
|
||||
results = append(results, entry.Name)
|
||||
}
|
||||
@@ -803,7 +921,7 @@ func TestListObjectsV2_Regression_Sorting(t *testing.T) {
|
||||
// Without the fix, Limit=1 would cause the lister to stop after "reports-archive",
|
||||
// missing the intended "reports" directory.
|
||||
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/reports", "reports", cursor, "", "", false, "reports", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/reports", prefix: "reports", bucket: "reports"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if !entry.IsDirectory {
|
||||
results = append(results, entry.Name)
|
||||
}
|
||||
@@ -846,7 +964,7 @@ func TestListObjectsV2_PrefixEndingWithSlash_DoesNotMatchSiblings(t *testing.T)
|
||||
cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
|
||||
var results []string
|
||||
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/bucket/path/to/list", "1", cursor, "", "", false, "bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/bucket/path/to/list", prefix: "1", bucket: "bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if !entry.IsDirectory {
|
||||
results = append(results, entry.Name)
|
||||
}
|
||||
@@ -878,7 +996,7 @@ func TestListObjectsV2_PrefixEndingWithSlash_WithDelimiter(t *testing.T) {
|
||||
cursor := &ListingCursor{maxKeys: 1000, prefixEndsOnDelimiter: true}
|
||||
var results []string
|
||||
|
||||
_, err := s3a.doListFilerEntries(client, "/buckets/bucket/path/to/list", "1", cursor, "", "/", false, "bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), client, listDirectoryRequest{dir: "/buckets/bucket/path/to/list", prefix: "1", delimiter: "/", bucket: "bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
results = append(results, entry.Name)
|
||||
})
|
||||
|
||||
|
||||
@@ -136,7 +136,7 @@ func TestListObjectsWithVersionedObjects(t *testing.T) {
|
||||
commonPrefixes := []PrefixEntry{}
|
||||
bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, tt.bucket)
|
||||
|
||||
_, err := s3a.doListFilerEntries(filerClient, bucketPrefix[:len(bucketPrefix)-1], tt.prefix, cursor, "", tt.delimiter, false, tt.bucket, func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: bucketPrefix[:len(bucketPrefix)-1], prefix: tt.prefix, delimiter: tt.delimiter, bucket: tt.bucket}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if cursor.maxKeys <= 0 {
|
||||
return
|
||||
}
|
||||
@@ -237,7 +237,7 @@ func TestVersionedObjectsNoDuplication(t *testing.T) {
|
||||
|
||||
cursor := &ListingCursor{maxKeys: uint16(1000)}
|
||||
contents := []ListEntry{}
|
||||
_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if cursor.maxKeys <= 0 {
|
||||
return
|
||||
}
|
||||
@@ -295,7 +295,7 @@ func TestVersionedObjectsWithDeleteMarker(t *testing.T) {
|
||||
|
||||
cursor := &ListingCursor{maxKeys: uint16(1000)}
|
||||
contents := []ListEntry{}
|
||||
_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if cursor.maxKeys <= 0 {
|
||||
return
|
||||
}
|
||||
@@ -344,7 +344,7 @@ func TestVersionedObjectsMaxKeys(t *testing.T) {
|
||||
|
||||
cursor := &ListingCursor{maxKeys: uint16(3)}
|
||||
contents := []ListEntry{}
|
||||
_, err := s3a.doListFilerEntries(filerClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), filerClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if cursor.maxKeys <= 0 {
|
||||
return
|
||||
}
|
||||
@@ -406,7 +406,7 @@ func TestVersionsDirectoryNotTraversed(t *testing.T) {
|
||||
|
||||
cursor := &ListingCursor{maxKeys: uint16(1000)}
|
||||
contents := []ListEntry{}
|
||||
_, err := s3a.doListFilerEntries(customClient, "/buckets/test-bucket", "", cursor, "", "", false, "test-bucket", func(dir string, entry *filer_pb.Entry) {
|
||||
_, err := s3a.doListFilerEntries(context.Background(), customClient, listDirectoryRequest{dir: "/buckets/test-bucket", bucket: "test-bucket"}, cursor, func(dir string, entry *filer_pb.Entry) {
|
||||
if cursor.maxKeys <= 0 {
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user