xl: quickHeal heal bucket only when needed. (#3854)

This improves the startup time significantly
for clusters which have lot of buckets.

Also fixes a bug where `.minio.sys` is created
on disks which do not have `format.json`

cmd/xl-v1-healing.go

@@ -252,6 +252,7 @@ func (xl xlObjects) ListBucketsHeal() ([]BucketInfo, error) {
 	if err != nil {
 		return listBuckets, err
 	}
+
 	// Iterate over all buckets
 	for _, currBucket := range buckets {
 		// Check the status of bucket metadata
@@ -275,6 +276,7 @@ func (xl xlObjects) ListBucketsHeal() ([]BucketInfo, error) {
 		}
 
 	}
+
 	// Sort found buckets
 	sort.Sort(byBucketName(listBuckets))
 	return listBuckets, nil
@@ -284,19 +286,18 @@ func (xl xlObjects) ListBucketsHeal() ([]BucketInfo, error) {
 // during startup i.e healing of buckets, bucket metadata (policy.json,
 // notification.xml, listeners.json) etc. Currently this function
 // supports quick healing of buckets, bucket metadata.
-//
-// TODO :-
-// - add support for healing dangling `uploads.json`.
-// - add support for healing dangling `xl.json`.
 func quickHeal(storageDisks []StorageAPI, writeQuorum int, readQuorum int) error {
-	// List all bucket names from all disks.
+	// List all bucket name occurrence from all disks.
-	bucketNames, _, err := listAllBuckets(storageDisks)
+	_, bucketOcc, err := listAllBuckets(storageDisks)
 	if err != nil {
 		return err
 	}
-	// All bucket names and bucket metadata should be healed.
+
-	for bucketName := range bucketNames {
+	// All bucket names and bucket metadata that should be healed.
-		// Heal bucket and then proceed to heal bucket metadata.
+	for bucketName, occCount := range bucketOcc {
+		// Heal bucket only if healing is needed.
+		if occCount != len(storageDisks) {
+			// Heal bucket and then proceed to heal bucket metadata if any.
 			if err = healBucket(storageDisks, bucketName, writeQuorum); err == nil {
 				if err = healBucketMetadata(storageDisks, bucketName, readQuorum); err == nil {
 					continue
@@ -305,6 +306,9 @@ func quickHeal(storageDisks []StorageAPI, writeQuorum int, readQuorum int) error
 			}
 			return err
 		}
+	}
+
+	// Success.
 	return nil
 }
 

cmd/xl-v1.go

@@ -151,7 +151,7 @@ func newXLObjects(storageDisks []StorageAPI) (ObjectLayer, error) {
 	}
 
 	// Initialize meta volume, if volume already exists ignores it.
-	if err = initMetaVolume(storageDisks); err != nil {
+	if err = initMetaVolume(xl.storageDisks); err != nil {
 		return nil, fmt.Errorf("Unable to initialize '.minio.sys' meta volume, %s", err)
 	}