Add number of offline disks in quorum errors (#16822)

cmd/admin-handler-utils.go

@@ -226,12 +226,10 @@ func toAdminAPIErr(ctx context.Context, err error) APIError {
 // toAdminAPIErrCode - converts errErasureWriteQuorum error to admin API
 // specific error.
 func toAdminAPIErrCode(ctx context.Context, err error) APIErrorCode {
-	switch err {
+	if errors.Is(err, errErasureWriteQuorum) {
-	case errErasureWriteQuorum:
 		return ErrAdminConfigNoQuorum
-	default:
-		return toAPIErrorCode(ctx, err)
 	}
+	return toAPIErrorCode(ctx, err)
 }
 
 // wraps export error for more context

cmd/api-errors.go

@@ -2026,6 +2026,9 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) {
 		return ErrClientDisconnected
 	}
 
+	// Unwrap the error first
+	err = unwrapAll(err)
+
 	switch err {
 	case errInvalidArgument:
 		apiErr = ErrAdminInvalidArgument

cmd/api-errors_test.go

@@ -20,8 +20,6 @@ package cmd
 import (
 	"context"
 	"errors"
-	"os"
-	"path/filepath"
 	"testing"
 
 	"github.com/minio/minio/internal/crypto"
@@ -67,11 +65,6 @@ var toAPIErrorTests = []struct {
 }
 
 func TestAPIErrCode(t *testing.T) {
-	disk := filepath.Join(globalTestTmpDir, "minio-"+nextSuffix())
-	defer os.RemoveAll(disk)
-
-	initFSObjects(disk, t)
-
 	ctx := context.Background()
 	for i, testCase := range toAPIErrorTests {
 		errCode := toAPIErrorCode(ctx, testCase.err)

cmd/erasure-decode.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"sync"
 	"sync/atomic"
@@ -124,6 +125,7 @@ func (p *parallelReader) Read(dst [][]byte) ([][]byte, error) {
 		readTriggerCh <- true
 	}
 
+	disksNotFound := int32(0)
 	bitrotHeal := int32(0)       // Atomic bool flag.
 	missingPartsHeal := int32(0) // Atomic bool flag.
 	readerIndex := 0
@@ -164,10 +166,13 @@ func (p *parallelReader) Read(dst [][]byte) ([][]byte, error) {
 			p.buf[bufIdx] = p.buf[bufIdx][:p.shardSize]
 			n, err := rr.ReadAt(p.buf[bufIdx], p.offset)
 			if err != nil {
-				if errors.Is(err, errFileNotFound) {
+				switch {
+				case errors.Is(err, errFileNotFound):
 					atomic.StoreInt32(&missingPartsHeal, 1)
-				} else if errors.Is(err, errFileCorrupt) {
+				case errors.Is(err, errFileCorrupt):
 					atomic.StoreInt32(&bitrotHeal, 1)
+				case errors.Is(err, errDiskNotFound):
+					atomic.AddInt32(&disksNotFound, 1)
 				}
 
 				// This will be communicated upstream.
@@ -189,16 +194,16 @@ func (p *parallelReader) Read(dst [][]byte) ([][]byte, error) {
 	wg.Wait()
 	if p.canDecode(newBuf) {
 		p.offset += p.shardSize
-		if atomic.LoadInt32(&missingPartsHeal) == 1 {
+		if missingPartsHeal == 1 {
 			return newBuf, errFileNotFound
-		} else if atomic.LoadInt32(&bitrotHeal) == 1 {
+		} else if bitrotHeal == 1 {
 			return newBuf, errFileCorrupt
 		}
 		return newBuf, nil
 	}
 
 	// If we cannot decode, just return read quorum error.
-	return nil, errErasureReadQuorum
+	return nil, fmt.Errorf("%w (offline-disks=%d/%d)", errErasureReadQuorum, disksNotFound, len(p.readers))
 }
 
 // Decode reads from readers, reconstructs data if needed and writes the data to the writer.

cmd/erasure-encode.go

@@ -19,6 +19,7 @@ package cmd
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"sync"
 
@@ -70,7 +71,9 @@ func (p *parallelWriter) Write(ctx context.Context, blocks [][]byte) error {
 	if nilCount >= p.writeQuorum {
 		return nil
 	}
-	return reduceWriteQuorumErrs(ctx, p.errs, objectOpIgnoredErrs, p.writeQuorum)
+
+	writeErr := reduceWriteQuorumErrs(ctx, p.errs, objectOpIgnoredErrs, p.writeQuorum)
+	return fmt.Errorf("%w (offline-disks=%d/%d)", writeErr, countErrs(p.errs, errDiskNotFound), len(p.writers))
 }
 
 // Encode reads from the reader, erasure-encodes the data and writes to the writers.

cmd/erasure-multipart.go

@@ -88,7 +88,7 @@ func (er erasureObjects) checkUploadIDExists(ctx context.Context, bucket, object
 
 	if write {
 		reducedErr := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
-		if reducedErr == errErasureWriteQuorum {
+		if errors.Is(reducedErr, errErasureWriteQuorum) {
 			return fi, nil, reducedErr
 		}
 	} else {
@@ -563,7 +563,7 @@ func writeAllDisks(ctx context.Context, disks []StorageAPI, dstBucket, dstEntry
 	// We can safely allow RenameFile errors up to len(er.getDisks()) - writeQuorum
 	// otherwise return failure. Cleanup successful renames.
 	err := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
-	if err == errErasureWriteQuorum {
+	if errors.Is(err, errErasureWriteQuorum) {
 		// Remove all written
 		g := errgroup.WithNErrs(len(disks))
 		for index := range disks {

cmd/object-api-errors.go

@@ -31,9 +31,14 @@ func toObjectErr(err error, params ...string) error {
 	if err == nil {
 		return nil
 	}
-	if errors.Is(err, context.Canceled) {
+
+	// Unwarp the error first
+	err = unwrapAll(err)
+
+	if err == context.Canceled {
 		return context.Canceled
 	}
+
 	switch err.Error() {
 	case errVolumeNotFound.Error():
 		apiErr := BucketNotFound{}

cmd/test-utils_test.go

@@ -224,19 +224,6 @@ func prepareErasure16(ctx context.Context) (ObjectLayer, []string, error) {
 	return prepareErasure(ctx, 16)
 }
 
-// Initialize FS objects.
-func initFSObjects(disk string, t *testing.T) (obj ObjectLayer) {
-	obj, _, err := initObjectLayer(context.Background(), mustGetPoolEndpoints(disk))
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	newTestConfig(globalMinioDefaultRegion, obj)
-
-	initAllSubsystems(GlobalContext)
-	return obj
-}
-
 // TestErrHandler - Go testing.T satisfy this interface.
 // This makes it easy to run the TestServer from any of the tests.
 // Using this interface, functionalities to be used in tests can be

cmd/utils.go

@@ -1261,3 +1261,14 @@ func MockOpenIDTestUserInteraction(ctx context.Context, pro OpenIDClientAppParam
 	// fmt.Printf("TOKEN: %s\n", rawIDToken)
 	return rawIDToken, nil
 }
+
+// unwrapAll will unwrap the returned error completely.
+func unwrapAll(err error) error {
+	for {
+		werr := errors.Unwrap(err)
+		if werr == nil {
+			return err
+		}
+		err = werr
+	}
+}

internal/crypto/error.go

@@ -18,6 +18,7 @@
 package crypto
 
 import (
+	"errors"
 	"fmt"
 )
 
@@ -25,24 +26,29 @@ import (
 // an object. It indicates that the object itself or its metadata was
 // modified accidentally or maliciously.
 type Error struct {
-	err error
+	msg   string
+	cause error
 }
 
 // Errorf - formats according to a format specifier and returns
 // the string as a value that satisfies error of type crypto.Error
 func Errorf(format string, a ...interface{}) error {
-	return Error{err: fmt.Errorf(format, a...)}
+	e := fmt.Errorf(format, a...)
+	ee := Error{}
+	ee.msg = e.Error()
+	ee.cause = errors.Unwrap(e)
+	return ee
 }
 
 // Unwrap the internal error.
-func (e Error) Unwrap() error { return e.err }
+func (e Error) Unwrap() error { return e.cause }
 
 // Error 'error' compatible method.
 func (e Error) Error() string {
-	if e.err == nil {
+	if e.msg == "" {
 		return "crypto: cause <nil>"
 	}
-	return e.err.Error()
+	return e.msg
 }
 
 var (