HealBucket: create the bucket if it is missing in one of the disks. (#2924)

cmd/control-heal-main.go

@@ -117,6 +117,16 @@ func healControl(ctx *cli.Context) {
 		return
 	}
 
+	// If object is "" then heal the bucket first.
+	if objectName == "" {
+		fmt.Printf("Healing : /%s\n", bucketName)
+		args := &HealObjectArgs{Bucket: bucketName, Object: ""}
+		reply := &HealObjectReply{}
+		err = client.Call("Control.HealObjectHandler", args, reply)
+		fatalIf(err, "Healing bucket %s failed.", bucketName)
+		// Continue to heal the objects in the bucket.
+	}
+
 	// Recursively list and heal the objects.
 	prefix := objectName
 	marker := ""

cmd/xl-v1-bucket.go

@@ -305,3 +305,27 @@ func (xl xlObjects) DeleteBucket(bucket string) error {
 	// Success.
 	return nil
 }
+
+// Heal bucket - create buckets on disks where it does not exist.
+func healBucket(disks []StorageAPI, bucket string) error {
+	bucketFound := false
+	for _, disk := range disks {
+		_, err := disk.StatVol(bucket)
+		if err == nil {
+			bucketFound = true
+		}
+	}
+	if !bucketFound {
+		return traceError(errVolumeNotFound)
+	}
+	for _, disk := range disks {
+		err := disk.MakeVol(bucket)
+		if err == nil {
+			continue
+		}
+		if err != errVolumeExists {
+			return traceError(err)
+		}
+	}
+	return nil
+}

cmd/xl-v1-object.go

@@ -226,6 +226,11 @@ func (xl xlObjects) HealObject(bucket, object string) error {
 	if !IsValidBucketName(bucket) {
 		return traceError(BucketNameInvalid{Bucket: bucket})
 	}
+	if object == "" {
+		// Empty object name indicates that bucket should be healed.
+		return healBucket(xl.storageDisks, bucket)
+	}
+
 	// Verify if object is valid.
 	if !IsValidObjectName(object) {
 		// FIXME: return Invalid prefix.

cmd/xl-v1-object_test.go

@@ -345,4 +345,21 @@ func TestHealObject(t *testing.T) {
 	if !reflect.DeepEqual(xlMetaPreHeal, xlMetaPostHeal) {
 		t.Fatal("HealObject failed")
 	}
+
+	// Remove the bucket - to simulate the case where bucket was created when the
+	// disk was down.
+	err = os.RemoveAll(path.Join(fsDirs[0], bucket))
+	if err != nil {
+		t.Fatal(err)
+	}
+	// This would create the bucket.
+	err = xl.HealObject(bucket, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Stat the bucket to make sure that it was created.
+	_, err = xl.storageDisks[0].StatVol(bucket)
+	if err != nil {
+		t.Fatal(err)
+	}
 }