From 2afd196c83ceccaa930dff74ed052432bfdcdecf Mon Sep 17 00:00:00 2001 From: Krishna Srinivas Date: Thu, 1 Feb 2018 10:47:49 -0800 Subject: [PATCH] Quorum based listing for XL (#5475) fixes #5380 --- cmd/tree-walk_test.go | 9 ++++++--- cmd/xl-v1-list-objects.go | 41 +++++++++++++++++++++++++++++---------- cmd/xl-v1-object.go | 36 +++++++++++++++++++++++++--------- 3 files changed, 64 insertions(+), 22 deletions(-) diff --git a/cmd/tree-walk_test.go b/cmd/tree-walk_test.go index 590bfd940..081e6f653 100644 --- a/cmd/tree-walk_test.go +++ b/cmd/tree-walk_test.go @@ -312,14 +312,17 @@ func TestListDir(t *testing.T) { if err != nil { t.Error(err) } - if len(entries) != 1 { - t.Fatal("Expected the number of entries to be 1") + if len(entries) != 2 { + t.Fatal("Expected the number of entries to be 2") } if entries[0] != file1 { t.Fatal("Expected the entry to be file1") } + if entries[1] != file2 { + t.Fatal("Expected the entry to be file2") + } - // Remove fsDir1 to test failover. + // Remove fsDir1, list should return entries from fsDir2 err = os.RemoveAll(fsDir1) if err != nil { t.Error(err) diff --git a/cmd/xl-v1-list-objects.go b/cmd/xl-v1-list-objects.go index ab02972e5..ff23fdf93 100644 --- a/cmd/xl-v1-list-objects.go +++ b/cmd/xl-v1-list-objects.go @@ -16,18 +16,24 @@ package cmd -import "github.com/minio/minio/pkg/errors" +import ( + "sort" + + "github.com/minio/minio/pkg/errors" +) // Returns function "listDir" of the type listDirFunc. // isLeaf - is used by listDir function to check if an entry is a leaf or non-leaf entry. -// disks - used for doing disk.ListDir(). FS passes single disk argument, XL passes a list of disks. +// disks - used for doing disk.ListDir() func listDirFactory(isLeaf isLeafFunc, treeWalkIgnoredErrs []error, disks ...StorageAPI) listDirFunc { - // listDir - lists all the entries at a given prefix and given entry in the prefix. - listDir := func(bucket, prefixDir, prefixEntry string) (entries []string, delayIsLeaf bool, err error) { + // Returns sorted merged entries from all the disks. + listDir := func(bucket, prefixDir, prefixEntry string) (mergedEntries []string, delayIsLeaf bool, err error) { for _, disk := range disks { if disk == nil { continue } + var entries []string + var newEntries []string entries, err = disk.ListDir(bucket, prefixDir) if err != nil { // For any reason disk was deleted or goes offline, continue @@ -38,11 +44,24 @@ func listDirFactory(isLeaf isLeafFunc, treeWalkIgnoredErrs []error, disks ...Sto return nil, false, errors.Trace(err) } - entries, delayIsLeaf = filterListEntries(bucket, prefixDir, entries, prefixEntry, isLeaf) - return entries, delayIsLeaf, nil + // Find elements in entries which are not in mergedEntries + for _, entry := range entries { + idx := sort.SearchStrings(mergedEntries, entry) + // if entry is already present in mergedEntries don't add. + if idx < len(mergedEntries) && mergedEntries[idx] == entry { + continue + } + newEntries = append(newEntries, entry) + } + + if len(newEntries) > 0 { + // Merge the entries and sort it. + mergedEntries = append(mergedEntries, newEntries...) + sort.Strings(mergedEntries) + } } - // Nothing found in all disks - return nil, false, nil + mergedEntries, delayIsLeaf = filterListEntries(bucket, prefixDir, mergedEntries, prefixEntry, isLeaf) + return mergedEntries, delayIsLeaf, nil } return listDir } @@ -90,8 +109,10 @@ func (xl xlObjects) listObjects(bucket, prefix, marker, delimiter string, maxKey var err error objInfo, err = xl.getObjectInfo(bucket, entry) if err != nil { - // Ignore errFileNotFound - if errors.Cause(err) == errFileNotFound { + // Ignore errFileNotFound as the object might have got deleted in the interim period + // of listing and getObjectInfo() + // Ignore quorum error as it might be an entry from an outdated disk. + if errors.Cause(err) == errFileNotFound || errors.Cause(err) == errXLReadQuorum { continue } return loi, toObjectErr(err, bucket, prefix) diff --git a/cmd/xl-v1-object.go b/cmd/xl-v1-object.go index bbe37d9ab..290a6c5bb 100644 --- a/cmd/xl-v1-object.go +++ b/cmd/xl-v1-object.go @@ -446,30 +446,48 @@ func (xl xlObjects) GetObjectInfo(bucket, object string) (oi ObjectInfo, e error // getObjectInfo - wrapper for reading object metadata and constructs ObjectInfo. func (xl xlObjects) getObjectInfo(bucket, object string) (objInfo ObjectInfo, err error) { + // Read metadata associated with the object from all disks. + metaArr, errs := readAllXLMetadata(xl.storageDisks, bucket, object) - // Extracts xlStat and xlMetaMap. - xlStat, xlMetaMap, err := xl.readXLMetaStat(bucket, object) + // get Quorum for this object + readQuorum, _, err := objectQuorumFromMeta(xl, metaArr, errs) if err != nil { - return ObjectInfo{}, err + return objInfo, toObjectErr(err, bucket, object) + } + + if reducedErr := reduceReadQuorumErrs(errs, objectOpIgnoredErrs, readQuorum); reducedErr != nil { + return objInfo, toObjectErr(reducedErr, bucket, object) + } + + // List all the file commit ids from parts metadata. + modTimes := listObjectModtimes(metaArr, errs) + + // Reduce list of UUIDs to a single common value. + modTime, _ := commonTime(modTimes) + + // Pick latest valid metadata. + xlMeta, err := pickValidXLMeta(metaArr, modTime) + if err != nil { + return objInfo, err } objInfo = ObjectInfo{ IsDir: false, Bucket: bucket, Name: object, - Size: xlStat.Size, - ModTime: xlStat.ModTime, - ContentType: xlMetaMap["content-type"], - ContentEncoding: xlMetaMap["content-encoding"], + Size: xlMeta.Stat.Size, + ModTime: xlMeta.Stat.ModTime, + ContentType: xlMeta.Meta["content-type"], + ContentEncoding: xlMeta.Meta["content-encoding"], } // Extract etag. - objInfo.ETag = extractETag(xlMetaMap) + objInfo.ETag = extractETag(xlMeta.Meta) // etag/md5Sum has already been extracted. We need to // remove to avoid it from appearing as part of // response headers. e.g, X-Minio-* or X-Amz-*. - objInfo.UserDefined = cleanMetadata(xlMetaMap) + objInfo.UserDefined = cleanMetadata(xlMeta.Meta) // Success. return objInfo, nil