mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-06-03 01:00:04 +02:00
Code Issues Projects Releases Wiki Activity

fix s3 tests

Browse source 
bucket_list_delimiter_prefix
bucket_list_delimiter_prefix_underscore
bucket_list_delimiter_prefix_ends_with_delimiter
This commit is contained in:
Konstantin Lebedev 2022-06-07 14:43:10 +05:00
parent b86628f85d
commit c07820178f
3 changed files with 53 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
weed/pb/filer_pb/filer_pb_helper.go
View file

@ -17,6 +17,10 @@ func (entry *Entry) IsInRemoteOnly() bool {
return len(entry.Chunks) == 0 && entry.RemoteEntry != nil && entry.RemoteEntry.RemoteSize > 0 return len(entry.Chunks) == 0 && entry.RemoteEntry != nil && entry.RemoteEntry.RemoteSize > 0
} }
func (entry *Entry) IsDirectoryKeyObject() bool {
return entry.IsDirectory && entry.Attributes != nil && entry.Attributes.Mime != ""
}
func (entry *Entry) FileMode() (fileMode os.FileMode) { func (entry *Entry) FileMode() (fileMode os.FileMode) {
if entry != nil && entry.Attributes != nil { if entry != nil && entry.Attributes != nil {
fileMode = os.FileMode(entry.Attributes.FileMode) fileMode = os.FileMode(entry.Attributes.FileMode)

8
weed/s3api/s3api_object_handlers.go
View file

@ -92,16 +92,20 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
} }
defer dataReader.Close() defer dataReader.Close()
objectContentType := r.Header.Get("Content-Type")
if strings.HasSuffix(object, "/") { if strings.HasSuffix(object, "/") {
if err := s3a.mkdir(s3a.option.BucketsPath, bucket+strings.TrimSuffix(object, "/"), func(entry *filer_pb.Entry) { if err := s3a.mkdir(s3a.option.BucketsPath, bucket+strings.TrimSuffix(object, "/"), func(entry *filer_pb.Entry) {
entry.Attributes.Mime = r.Header.Get("Content-Type") if objectContentType == "" {
objectContentType = "httpd/unix-directory"
}
entry.Attributes.Mime = objectContentType
}); err != nil { }); err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
return return
} }
} else { } else {
uploadUrl := s3a.toFilerUrl(bucket, object) uploadUrl := s3a.toFilerUrl(bucket, object)
if r.Header.Get("Content-Type") == "" { if objectContentType == "" {
dataReader = mimeDetect(r, dataReader) dataReader = mimeDetect(r, dataReader)
} }

72
weed/s3api/s3api_objects_list_handlers.go
View file

@ -133,6 +133,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
reqDir = reqDir[1:] reqDir = reqDir[1:]
} }
bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, bucket) bucketPrefix := fmt.Sprintf("%s/%s/", s3a.option.BucketsPath, bucket)
bucketPrefixLen := len(bucketPrefix)
reqDir = fmt.Sprintf("%s%s", bucketPrefix, reqDir) reqDir = fmt.Sprintf("%s%s", bucketPrefix, reqDir)
if strings.HasSuffix(reqDir, "/") { if strings.HasSuffix(reqDir, "/") {
reqDir = strings.TrimSuffix(reqDir, "/") reqDir = strings.TrimSuffix(reqDir, "/")
@ -147,14 +148,14 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
// check filer // check filer
err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, isTruncated, nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, maxKeys, marker, delimiter, false, func(dir string, entry *filer_pb.Entry) { _, isTruncated, nextMarker, doErr = s3a.doListFilerEntries(client, reqDir, prefix, maxKeys, marker, delimiter, false, false, bucketPrefixLen, func(dir string, entry *filer_pb.Entry) {
if entry.IsDirectory { if entry.IsDirectory {
if delimiter == "/" { if delimiter == "/" {
commonPrefixes = append(commonPrefixes, PrefixEntry{ commonPrefixes = append(commonPrefixes, PrefixEntry{
Prefix: fmt.Sprintf("%s/%s/", dir, entry.Name)[len(bucketPrefix):], Prefix: fmt.Sprintf("%s/%s/", dir, entry.Name)[bucketPrefixLen:],
}) })
} }
if entry.Attributes.Mime == "" || !strings.HasSuffix(entry.Name, "/") { if !(entry.IsDirectoryKeyObject() && strings.HasSuffix(entry.Name, "/")) {
return return
} }
} }
@ -163,7 +164,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
storageClass = string(v) storageClass = string(v)
} }
contents = append(contents, ListEntry{ contents = append(contents, ListEntry{
Key: fmt.Sprintf("%s/%s", dir, entry.Name)[len(bucketPrefix):], Key: fmt.Sprintf("%s/%s", dir, entry.Name)[bucketPrefixLen:],
LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(), LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(),
ETag: "\"" + filer.ETag(entry) + "\"", ETag: "\"" + filer.ETag(entry) + "\"",
Size: int64(filer.FileSize(entry)), Size: int64(filer.FileSize(entry)),
@ -174,6 +175,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
StorageClass: StorageClass(storageClass), StorageClass: StorageClass(storageClass),
}) })
}) })
glog.V(4).Infof("end doListFilerEntries isTruncated:%v nextMarker:%v reqDir: %v prefix: %v", isTruncated, nextMarker, reqDir, prefix)
if doErr != nil { if doErr != nil {
return doErr return doErr
} }
@ -187,14 +189,14 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
reqDir, prefix = filepath.Split(strings.TrimSuffix(reqDir, "/")) reqDir, prefix = filepath.Split(strings.TrimSuffix(reqDir, "/"))
reqDir = strings.TrimSuffix(reqDir, "/") reqDir = strings.TrimSuffix(reqDir, "/")
} }
_, _, _, doErr = s3a.doListFilerEntries(client, reqDir, prefix, 1, prefix, delimiter, true, func(dir string, entry *filer_pb.Entry) { _, _, _, doErr = s3a.doListFilerEntries(client, reqDir, prefix, 1, prefix, delimiter, true, false, bucketPrefixLen, func(dir string, entry *filer_pb.Entry) {
if entry.IsDirectory && entry.Attributes.Mime != "" && entry.Name == prefix { if entry.IsDirectoryKeyObject() && entry.Name == prefix {
storageClass := "STANDARD" storageClass := "STANDARD"
if v, ok := entry.Extended[s3_constants.AmzStorageClass]; ok { if v, ok := entry.Extended[s3_constants.AmzStorageClass]; ok {
storageClass = string(v) storageClass = string(v)
} }
contents = append(contents, ListEntry{ contents = append(contents, ListEntry{
Key: fmt.Sprintf("%s/%s/", dir, entry.Name)[len(bucketPrefix):], Key: fmt.Sprintf("%s/%s/", dir, entry.Name)[bucketPrefixLen:],
LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(), LastModified: time.Unix(entry.Attributes.Mtime, 0).UTC(),
ETag: "\"" + fmt.Sprintf("%x", entry.Attributes.Md5) + "\"", ETag: "\"" + fmt.Sprintf("%x", entry.Attributes.Md5) + "\"",
Size: int64(filer.FileSize(entry)), Size: int64(filer.FileSize(entry)),
@ -211,6 +213,10 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
} }
} }
if len(nextMarker) > 0 {
nextMarker = nextMarker[bucketPrefixLen:]
}
response = ListBucketResult{ response = ListBucketResult{
Name: bucket, Name: bucket,
Prefix: originalPrefix, Prefix: originalPrefix,
@ -229,7 +235,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
return return
} }
func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, maxKeys int, marker, delimiter string, inclusiveStartFrom bool, eachEntryFn func(dir string, entry *filer_pb.Entry)) (counter int, isTruncated bool, nextMarker string, err error) { func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, dir, prefix string, maxKeys int, marker, delimiter string, inclusiveStartFrom bool, subEntries bool, bucketPrefixLen int, eachEntryFn func(dir string, entry *filer_pb.Entry)) (counter int, isTruncated bool, nextMarker string, err error) {
// invariants // invariants
// prefix and marker should be under dir, marker may contain "/" // prefix and marker should be under dir, marker may contain "/"
// maxKeys should be updated for each recursion // maxKeys should be updated for each recursion
@ -242,20 +248,27 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
} }
if strings.Contains(marker, "/") { if strings.Contains(marker, "/") {
sepIndex := strings.Index(marker, "/") if strings.HasSuffix(marker, "/") {
subDir, subMarker := marker[0:sepIndex], marker[sepIndex+1:] marker = strings.TrimSuffix(marker, "/")
glog.V(4).Infoln("doListFilerEntries dir", dir+"/"+subDir, "subMarker", subMarker, "maxKeys", maxKeys) }
subCounter, subIsTruncated, subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+subDir, "", maxKeys, subMarker, delimiter, false, eachEntryFn) sepIndex := strings.Index(marker, "/")
if subErr != nil { if sepIndex != -1 {
err = subErr subPrefix, subMarker := marker[0:sepIndex], marker[sepIndex+1:]
return subDir := fmt.Sprintf("%s/%s", dir[0:bucketPrefixLen-1], subPrefix)
if strings.HasPrefix(subDir, dir) {
subCounter, subIsTruncated, subNextMarker, subErr := s3a.doListFilerEntries(client, subDir, "", maxKeys, subMarker, delimiter, false, false, bucketPrefixLen, eachEntryFn)
if subErr != nil {
err = subErr
return
}
counter += subCounter
isTruncated = isTruncated || subIsTruncated
maxKeys -= subCounter
nextMarker = subNextMarker
// finished processing this sub directory
marker = subPrefix
}
} }
counter += subCounter
isTruncated = isTruncated || subIsTruncated
maxKeys -= subCounter
nextMarker = subDir + "/" + subNextMarker
// finished processing this sub directory
marker = subDir
} }
if maxKeys <= 0 { if maxKeys <= 0 {
return return
@ -293,45 +306,46 @@ func (s3a *S3ApiServer) doListFilerEntries(client filer_pb.SeaweedFilerClient, d
return return
} }
entry := resp.Entry entry := resp.Entry
nextMarker = entry.Name nextMarker = dir + "/" + entry.Name
if entry.IsDirectory { if entry.IsDirectory {
// println("ListEntries", dir, "dir:", entry.Name) // println("ListEntries", dir, "dir:", entry.Name)
if entry.Name == ".uploads" { // FIXME no need to apply to all directories. this extra also affects maxKeys if entry.Name == ".uploads" { // FIXME no need to apply to all directories. this extra also affects maxKeys
continue continue
} }
if delimiter != "/" { if delimiter == "" {
eachEntryFn(dir, entry) eachEntryFn(dir, entry)
// println("doListFilerEntries2 dir", dir+"/"+entry.Name, "maxKeys", maxKeys-counter) // println("doListFilerEntries2 dir", dir+"/"+entry.Name, "maxKeys", maxKeys-counter)
subCounter, subIsTruncated, subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", maxKeys-counter, "", delimiter, false, eachEntryFn) subCounter, subIsTruncated, subNextMarker, subErr := s3a.doListFilerEntries(client, dir+"/"+entry.Name, "", maxKeys-counter, "", delimiter, false, true, bucketPrefixLen, eachEntryFn)
if subErr != nil { if subErr != nil {
err = fmt.Errorf("doListFilerEntries2: %v", subErr) err = fmt.Errorf("doListFilerEntries2: %v", subErr)
return return
} }
// println("doListFilerEntries2 dir", dir+"/"+entry.Name, "maxKeys", maxKeys-counter, "subCounter", subCounter, "subNextMarker", subNextMarker, "subIsTruncated", subIsTruncated) // println("doListFilerEntries2 dir", dir+"/"+entry.Name, "maxKeys", maxKeys-counter, "subCounter", subCounter, "subNextMarker", subNextMarker, "subIsTruncated", subIsTruncated)
if subCounter == 0 && entry.Attributes.Mime != "" { if subCounter == 0 && entry.IsDirectoryKeyObject() {
entry.Name += "/" entry.Name += "/"
eachEntryFn(dir, entry) eachEntryFn(dir, entry)
counter++ counter++
} }
counter += subCounter counter += subCounter
nextMarker = entry.Name + "/" + subNextMarker nextMarker = subNextMarker
if subIsTruncated { if subIsTruncated {
isTruncated = true isTruncated = true
return return
} }
} else { } else if delimiter == "/" {
var isEmpty bool var isEmpty bool
if !s3a.option.AllowEmptyFolder && entry.Attributes.Mime == "" { if !s3a.option.AllowEmptyFolder && !entry.IsDirectoryKeyObject() {
if isEmpty, err = s3a.isDirectoryAllEmpty(client, dir, entry.Name); err != nil { if isEmpty, err = s3a.isDirectoryAllEmpty(client, dir, entry.Name); err != nil {
glog.Errorf("check empty folder %s: %v", dir, err) glog.Errorf("check empty folder %s: %v", dir, err)
} }
} }
if !isEmpty { if !isEmpty {
nextMarker += "/"
eachEntryFn(dir, entry) eachEntryFn(dir, entry)
counter++ counter++
} }
} }
} else { } else if !(delimiter == "/" && subEntries) {
// println("ListEntries", dir, "file:", entry.Name) // println("ListEntries", dir, "file:", entry.Name)
eachEntryFn(dir, entry) eachEntryFn(dir, entry)
counter++ counter++