Quorum based listing for XL (#5475)

fixes #5380

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)

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)

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