format: Check properly for disks in valid formats. (#3427)

There was an error in how we validated disk formats,
if one of the disk was formatted and was formatted with
FS would cause confusion and object layer would never
initialize essentially go into an infinite loop.

Validate pre-emptively and also check for FS format
properly.

cmd/format-config-v1.go

@@ -206,23 +206,24 @@ func loadAllFormats(bootstrapDisks []StorageAPI) ([]*formatConfigV1, []error) {
 	return formatConfigs, sErrs
 }
 
-// genericFormatCheck - validates and returns error.
+// genericFormatCheckFS - validates format config and returns an error if any.
+func genericFormatCheckFS(formatConfig *formatConfigV1, sErr error) (err error) {
+	if sErr != nil {
+		return sErr
+	}
+	// Successfully read, validate if FS.
+	if !isFSFormat(formatConfig) {
+		return errFSDiskFormat
+	}
+	return nil
+}
+
+// genericFormatCheckXL - validates and returns error.
 // if (no quorum) return error
 // if (any disk is corrupt) return error // phase2
 // if (jbod inconsistent) return error // phase2
 // if (disks not recognized) // Always error.
-func genericFormatCheck(formatConfigs []*formatConfigV1, sErrs []error) (err error) {
-	if len(formatConfigs) == 1 {
-		// Successfully read, validate further.
-		if sErrs[0] == nil {
-			if !isFSFormat(formatConfigs[0]) {
-				return errFSDiskFormat
-			}
-			return nil
-		} // Returns error here.
-		return sErrs[0]
-	}
-
+func genericFormatCheckXL(formatConfigs []*formatConfigV1, sErrs []error) (err error) {
 	// Calculate the errors.
 	var (
 		errCorruptFormatCount = 0
@@ -248,12 +249,12 @@ func genericFormatCheck(formatConfigs []*formatConfigV1, sErrs []error) (err err
 	// Calculate read quorum.
 	readQuorum := len(formatConfigs) / 2
 
-	// Validate the err count under tolerant limit.
+	// Validate the err count under read quorum.
 	if errCount > len(formatConfigs)-readQuorum {
 		return errXLReadQuorum
 	}
 
-	// Check if number of corrupted format under quorum
+	// Check if number of corrupted format under read quorum
 	if errCorruptFormatCount > len(formatConfigs)-readQuorum {
 		return errCorruptedFormat
 	}
@@ -793,8 +794,7 @@ func loadFormatXL(bootstrapDisks []StorageAPI, readQuorum int) (disks []StorageA
 	return reorderDisks(bootstrapDisks, formatConfigs)
 }
 
-// checkFormatXL - verifies if format.json format is intact.
-func checkFormatXL(formatConfigs []*formatConfigV1) error {
+func checkFormatXLValues(formatConfigs []*formatConfigV1) error {
 	for _, formatXL := range formatConfigs {
 		if formatXL == nil {
 			continue
@@ -813,6 +813,14 @@ func checkFormatXL(formatConfigs []*formatConfigV1) error {
 			return fmt.Errorf("Number of disks %d did not match the backend format %d", len(formatConfigs), len(formatXL.XL.JBOD))
 		}
 	}
+	return nil
+}
+
+// checkFormatXL - verifies if format.json format is intact.
+func checkFormatXL(formatConfigs []*formatConfigV1) error {
+	if err := checkFormatXLValues(formatConfigs); err != nil {
+		return err
+	}
 	if err := checkJBODConsistency(formatConfigs); err != nil {
 		return err
 	}

cmd/format-config-v1_test.go

@@ -664,36 +664,58 @@ func TestReduceFormatErrs(t *testing.T) {
 	}
 }
 
-// Tests for genericFormatCheck()
-func TestGenericFormatCheck(t *testing.T) {
+// Tests for genericFormatCheckFS()
+func TestGenericFormatCheckFS(t *testing.T) {
+	// Generate format configs for XL.
+	formatConfigs := genFormatXLInvalidJBOD()
+
+	// Validate disk format is fs, should fail.
+	if err := genericFormatCheckFS(formatConfigs[0], nil); err != errFSDiskFormat {
+		t.Fatalf("Unexpected error, expected %s, got %s", errFSDiskFormat, err)
+	}
+
+	// Validate disk is unformatted, should fail.
+	if err := genericFormatCheckFS(nil, errUnformattedDisk); err != errUnformattedDisk {
+		t.Fatalf("Unexpected error, expected %s, got %s", errUnformattedDisk, err)
+	}
+
+	// Validate when disk is in FS format.
+	format := newFSFormatV1()
+	if err := genericFormatCheckFS(format, nil); err != nil {
+		t.Fatalf("Unexpected error should pass, failed with %s", err)
+	}
+}
+
+// Tests for genericFormatCheckXL()
+func TestGenericFormatCheckXL(t *testing.T) {
 	var errs []error
 	formatConfigs := genFormatXLInvalidJBOD()
 
 	// Some disks has corrupted formats, one faulty disk
 	errs = []error{nil, nil, errCorruptedFormat, errCorruptedFormat, errCorruptedFormat, errCorruptedFormat,
 		errCorruptedFormat, errFaultyDisk}
-	if err := genericFormatCheck(formatConfigs, errs); err != errCorruptedFormat {
+	if err := genericFormatCheckXL(formatConfigs, errs); err != errCorruptedFormat {
 		t.Fatal("Got unexpected err: ", err)
 	}
 
 	// Many faulty disks
 	errs = []error{nil, nil, errFaultyDisk, errFaultyDisk, errFaultyDisk, errFaultyDisk,
 		errCorruptedFormat, errFaultyDisk}
-	if err := genericFormatCheck(formatConfigs, errs); err != errXLReadQuorum {
+	if err := genericFormatCheckXL(formatConfigs, errs); err != errXLReadQuorum {
 		t.Fatal("Got unexpected err: ", err)
 	}
 
 	// All formats successfully loaded
 	errs = []error{nil, nil, nil, nil, nil, nil, nil, nil}
-	if err := genericFormatCheck(formatConfigs, errs); err == nil {
+	if err := genericFormatCheckXL(formatConfigs, errs); err == nil {
 		t.Fatalf("Should fail here")
 	}
 	errs = []error{nil}
-	if err := genericFormatCheck([]*formatConfigV1{genFormatFS()}, errs); err != nil {
-		t.Fatal("Got unexpected err: ", err)
+	if err := genericFormatCheckXL([]*formatConfigV1{genFormatFS()}, errs); err == nil {
+		t.Fatalf("Should fail here")
 	}
 	errs = []error{errFaultyDisk}
-	if err := genericFormatCheck([]*formatConfigV1{genFormatFS()}, errs); err == nil {
+	if err := genericFormatCheckXL([]*formatConfigV1{genFormatFS()}, errs); err == nil {
 		t.Fatalf("Should fail here")
 	}
 }

cmd/prepare-storage.go

@@ -220,51 +220,59 @@ func retryFormattingDisks(firstDisk bool, endpoints []*url.URL, storageDisks []S
 				// for disks not being available.
 				printRetryMsg(sErrs, storageDisks)
 			}
-			// Check if this is a XL or distributed XL, anything > 1 is considered XL backend.
-			if len(formatConfigs) > 1 {
-				switch prepForInitXL(firstDisk, sErrs, len(storageDisks)) {
-				case Abort:
-					return errCorruptedFormat
-				case FormatDisks:
-					console.Eraseline()
-					printFormatMsg(endpoints, storageDisks, printOnceFn())
-					return initFormatXL(storageDisks)
-				case InitObjectLayer:
-					console.Eraseline()
-					// Validate formats load before proceeding forward.
-					err := genericFormatCheck(formatConfigs, sErrs)
-					if err == nil {
-						printRegularMsg(endpoints, storageDisks, printOnceFn())
+			if len(formatConfigs) == 1 {
+				err := genericFormatCheckFS(formatConfigs[0], sErrs[0])
+				if err != nil {
+					if err == errUnformattedDisk {
+						return initFormatFS(storageDisks[0])
 					}
 					return err
-				case WaitForHeal:
-					// Validate formats load before proceeding forward.
-					err := genericFormatCheck(formatConfigs, sErrs)
-					if err == nil {
-						printHealMsg(endpoints, storageDisks, printOnceFn())
-					}
-					return err
-				case WaitForQuorum:
-					console.Printf(
-						"Initializing data volume. Waiting for minimum %d servers to come online. (elapsed %s)\n",
-						len(storageDisks)/2+1, getElapsedTime(),
-					)
-				case WaitForConfig:
-					// Print configuration errors.
-					printConfigErrMsg(storageDisks, sErrs, printOnceFn())
-				case WaitForAll:
-					console.Printf("Initializing data volume for first time. Waiting for other servers to come online (elapsed %s)\n", getElapsedTime())
-				case WaitForFormatting:
-					console.Printf("Initializing data volume for first time. Waiting for first server to come online (elapsed %s)\n", getElapsedTime())
 				}
-				continue
-			} // else We have FS backend now. Check fs format as well now.
-			if isFormatFound(formatConfigs) {
+				return nil
+			} // Check if this is a XL or distributed XL, anything > 1 is considered XL backend.
+			// Pre-emptively check if one of the formatted disks
+			// is invalid. This function returns success for the
+			// most part unless one of the formats is not consistent
+			// with expected XL format. For example if a user is trying
+			// to pool FS backend.
+			if err := checkFormatXLValues(formatConfigs); err != nil {
+				return err
+			}
+			switch prepForInitXL(firstDisk, sErrs, len(storageDisks)) {
+			case Abort:
+				return errCorruptedFormat
+			case FormatDisks:
 				console.Eraseline()
-				// Validate formats load before proceeding forward.
-				return genericFormatCheck(formatConfigs, sErrs)
-			} // else initialize the format for FS.
-			return initFormatFS(storageDisks[0])
+				printFormatMsg(endpoints, storageDisks, printOnceFn())
+				return initFormatXL(storageDisks)
+			case InitObjectLayer:
+				console.Eraseline()
+				// Validate formats loaded before proceeding forward.
+				err := genericFormatCheckXL(formatConfigs, sErrs)
+				if err == nil {
+					printRegularMsg(endpoints, storageDisks, printOnceFn())
+				}
+				return err
+			case WaitForHeal:
+				// Validate formats loaded before proceeding forward.
+				err := genericFormatCheckXL(formatConfigs, sErrs)
+				if err == nil {
+					printHealMsg(endpoints, storageDisks, printOnceFn())
+				}
+				return err
+			case WaitForQuorum:
+				console.Printf(
+					"Initializing data volume. Waiting for minimum %d servers to come online. (elapsed %s)\n",
+					len(storageDisks)/2+1, getElapsedTime(),
+				)
+			case WaitForConfig:
+				// Print configuration errors.
+				printConfigErrMsg(storageDisks, sErrs, printOnceFn())
+			case WaitForAll:
+				console.Printf("Initializing data volume for first time. Waiting for other servers to come online (elapsed %s)\n", getElapsedTime())
+			case WaitForFormatting:
+				console.Printf("Initializing data volume for first time. Waiting for first server to come online (elapsed %s)\n", getElapsedTime())
+			}
 		case <-globalServiceDoneCh:
 			return errors.New("Initializing data volumes gracefully stopped")
 		}

cmd/xl-v1-healing.go

@@ -31,7 +31,7 @@ func healFormatXL(storageDisks []StorageAPI) (err error) {
 	// Generic format check.
 	// - if (no quorum) return error
 	// - if (disks not recognized) // Always error.
-	if err = genericFormatCheck(formatConfigs, sErrs); err != nil {
+	if err = genericFormatCheckXL(formatConfigs, sErrs); err != nil {
 		return err
 	}