erasure: Handle failed disks so that we initialize properly if they are missing. (#1607)

Fixes #1592
Fixes #1579

posix.go

@@ -93,25 +93,25 @@ func newPosix(diskPath string) (StorageAPI, error) {
 		log.Error("Disk cannot be empty")
 		return nil, errInvalidArgument
 	}
-	if err := checkPathLength(diskPath); err != nil {
-		return nil, err
+	fs := fsStorage{
+		diskPath:    diskPath,
+		minFreeDisk: fsMinSpacePercent, // Minimum 5% disk should be free.
 	}
 	st, err := os.Stat(diskPath)
 	if err != nil {
 		log.WithFields(logrus.Fields{
 			"diskPath": diskPath,
 		}).Debugf("Stat failed, with error %s.", err)
-		return nil, err
+		if os.IsNotExist(err) {
+			return fs, errDiskNotFound
+		}
+		return fs, err
 	}
 	if !st.IsDir() {
 		log.WithFields(logrus.Fields{
 			"diskPath": diskPath,
 		}).Debugf("Disk %s.", syscall.ENOTDIR)
-		return nil, syscall.ENOTDIR
-	}
-	fs := fsStorage{
-		diskPath:    diskPath,
-		minFreeDisk: fsMinSpacePercent, // Minimum 5% disk should be free.
+		return fs, syscall.ENOTDIR
 	}
 	log.WithFields(logrus.Fields{
 		"diskPath":    diskPath,

rpc-server.go

@@ -129,7 +129,7 @@ func (s *storageServer) RenameFileHandler(arg *RenameFileArgs, reply *GenericRep
 func newRPCServer(exportPath string) (*storageServer, error) {
 	// Initialize posix storage API.
 	storage, err := newPosix(exportPath)
-	if err != nil {
+	if err != nil && err != errDiskNotFound {
 		return nil, err
 	}
 	return &storageServer{

xl-erasure-v1-common.go

@@ -56,18 +56,24 @@ func listFileVersions(partsMetadata []xlMetaV1, errs []error) (versions []int64)
 func (xl XL) listOnlineDisks(volume, path string) (onlineDisks []StorageAPI, mdata xlMetaV1, heal bool, err error) {
 	partsMetadata, errs := xl.getPartsMetadata(volume, path)
 	notFoundCount := 0
-	// FIXME: take care of the situation when a disk has failed and been removed
-	// by looking at the error returned from the fs layer. fs-layer will have
-	// to return an error indicating that the disk is not available and should be
-	// different from ErrNotExist.
+	diskNotFoundCount := 0
 	for _, err := range errs {
 		if err == errFileNotFound {
 			notFoundCount++
-			// If we have errors with file not found equal to the number of disks.
+			// If we have errors with file not found greater than
+			// writeQuroum, return as errFileNotFound.
 			if notFoundCount > len(xl.storageDisks)-xl.readQuorum {
 				return nil, xlMetaV1{}, false, errFileNotFound
 			}
 		}
+		if err == errDiskNotFound {
+			diskNotFoundCount++
+			// If we have errors with disk not found equal to the
+			// number of disks, return as errDiskNotFound.
+			if diskNotFoundCount == len(xl.storageDisks) {
+				return nil, xlMetaV1{}, false, errDiskNotFound
+			}
+		}
 	}
 
 	highestVersion := int64(0)

xl-erasure-v1-errors.go

@@ -19,10 +19,13 @@ package main
 import "errors"
 
 // errMaxDisks - returned for reached maximum of disks.
-var errMaxDisks = errors.New("Total number of disks specified is higher than supported maximum of '16'")
+var errMaxDisks = errors.New("Number of disks are higher than supported maximum count '16'")
 
-// errNumDisks - returned for odd numebr of disks.
-var errNumDisks = errors.New("Invalid number of disks provided, should be always multiples of '2'")
+// errMinDisks - returned for minimum number of disks.
+var errMinDisks = errors.New("Number of disks are smaller than supported minimum count '8'")
+
+// errNumDisks - returned for odd number of disks.
+var errNumDisks = errors.New("Number of disks should be multiples of '2'")
 
 // errUnexpected - returned for any unexpected error.
 var errUnexpected = errors.New("Unexpected error - please report at https://github.com/minio/minio/issues")

xl-erasure-v1.go

@@ -34,6 +34,8 @@ const (
 	xlMetaV1File = "file.json"
 	// Maximum erasure blocks.
 	maxErasureBlocks = 16
+	// Minimum erasure blocks.
+	minErasureBlocks = 8
 )
 
 // XL layer structure.
@@ -51,18 +53,22 @@ func newXL(disks ...string) (StorageAPI, error) {
 	// Initialize XL.
 	xl := &XL{}
 
-	// Verify disks.
+	// Verify total number of disks.
 	totalDisks := len(disks)
 	if totalDisks > maxErasureBlocks {
 		return nil, errMaxDisks
 	}
+	if totalDisks < minErasureBlocks {
+		return nil, errMinDisks
+	}
 
 	// isEven function to verify if a given number if even.
 	isEven := func(number int) bool {
 		return number%2 == 0
 	}
 
-	// TODO: verify if this makes sense in future.
+	// Verify if we have even number of disks.
+	// only combination of 8, 10, 12, 14, 16 are supported.
 	if !isEven(totalDisks) {
 		return nil, errNumDisks
 	}
@@ -85,8 +91,13 @@ func newXL(disks ...string) (StorageAPI, error) {
 	storageDisks := make([]StorageAPI, len(disks))
 	for index, disk := range disks {
 		var err error
+		// Intentionally ignore disk not found errors while
+		// initializing POSIX, so that we have successfully
+		// initialized posix Storage.
+		// Subsequent calls to XL/Erasure will manage any errors
+		// related to disks.
 		storageDisks[index], err = newPosix(disk)
-		if err != nil {
+		if err != nil && err != errDiskNotFound {
 			return nil, err
 		}
 	}
@@ -153,6 +164,9 @@ func (xl XL) MakeVol(volume string) error {
 
 	// Make a volume entry on all underlying storage disks.
 	for index, disk := range xl.storageDisks {
+		if disk == nil {
+			continue
+		}
 		wg.Add(1)
 		// Make a volume inside a go-routine.
 		go func(index int, disk StorageAPI) {

xl-objects.go

@@ -18,6 +18,7 @@ package main
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"path/filepath"
@@ -91,6 +92,14 @@ func newXLObjects(exportPaths ...string) (ObjectLayer, error) {
 			}
 		} else {
 			log.Errorf("Unable to check backend format %s", err)
+			if err == errReadQuorum {
+				errMsg := fmt.Sprintf("Not all disks %s on command line are available to meet the read quroum.", exportPaths)
+				return nil, errors.New(errMsg)
+			}
+			if err == errDiskNotFound {
+				errMsg := fmt.Sprintf("All disks %s on command line are not available.", exportPaths)
+				return nil, errors.New(errMsg)
+			}
 			return nil, err
 		}
 	}