utils: reduceErrs returns and validates quorum errors. (#3300)

This is needed as explained by @krisis

Lets say we have following errors.

```
[]error{nil, errFileNotFound, errDiskAccessDenied, errDiskAccesDenied}
```

Since the last two errors are filtered, the maximum is nil,
depending on map order.

Let's say we get nil from reduceErr. Clearly at this point
we don't have quorum nodes agreeing about the data and since
GetObject only requires N/2 (Read quorum) and isDiskQuorum
would have returned true. This is problematic and can lead to
undersiable consequences.

Fixes #3298
This commit is contained in:
Harshavardhana 2016-11-21 01:47:26 -08:00 committed by GitHub
parent 066f64d34a
commit 5197649081
10 changed files with 75 additions and 36 deletions

View File

@ -141,5 +141,5 @@ func appendFile(disks []StorageAPI, volume, path string, enBlocks [][]byte, hash
if !isDiskQuorum(wErrs, writeQuorum) { if !isDiskQuorum(wErrs, writeQuorum) {
return traceError(errXLWriteQuorum) return traceError(errXLWriteQuorum)
} }
return nil return reduceWriteQuorumErrs(wErrs, objectOpIgnoredErrs, writeQuorum)
} }

View File

@ -80,7 +80,7 @@ func (xl xlObjects) MakeBucket(bucket string) error {
} }
// Verify we have any other errors which should undo make bucket. // Verify we have any other errors which should undo make bucket.
if reducedErr := reduceErrs(dErrs, bucketOpIgnoredErrs); reducedErr != nil { if reducedErr := reduceWriteQuorumErrs(dErrs, bucketOpIgnoredErrs, xl.writeQuorum); reducedErr != nil {
return toObjectErr(reducedErr, bucket) return toObjectErr(reducedErr, bucket)
} }
return nil return nil
@ -289,7 +289,7 @@ func (xl xlObjects) DeleteBucket(bucket string) error {
return toObjectErr(traceError(errXLWriteQuorum), bucket) return toObjectErr(traceError(errXLWriteQuorum), bucket)
} }
if reducedErr := reduceErrs(dErrs, bucketOpIgnoredErrs); reducedErr != nil { if reducedErr := reduceWriteQuorumErrs(dErrs, bucketOpIgnoredErrs, xl.writeQuorum); reducedErr != nil {
return toObjectErr(reducedErr, bucket) return toObjectErr(reducedErr, bucket)
} }

View File

@ -74,7 +74,7 @@ func (xl xlObjects) HealBucket(bucket string) error {
} }
// Proceed to heal bucket metadata. // Proceed to heal bucket metadata.
return healBucketMetadata(xl.storageDisks, bucket) return healBucketMetadata(xl.storageDisks, bucket, xl.readQuorum)
} }
// Heal bucket - create buckets on disks where it does not exist. // Heal bucket - create buckets on disks where it does not exist.
@ -122,7 +122,7 @@ func healBucket(storageDisks []StorageAPI, bucket string, writeQuorum int) error
} }
// Verify we have any other errors which should be returned as failure. // Verify we have any other errors which should be returned as failure.
if reducedErr := reduceErrs(dErrs, bucketOpIgnoredErrs); reducedErr != nil { if reducedErr := reduceWriteQuorumErrs(dErrs, bucketOpIgnoredErrs, writeQuorum); reducedErr != nil {
return toObjectErr(reducedErr, bucket) return toObjectErr(reducedErr, bucket)
} }
return nil return nil
@ -130,13 +130,13 @@ func healBucket(storageDisks []StorageAPI, bucket string, writeQuorum int) error
// Heals all the metadata associated for a given bucket, this function // Heals all the metadata associated for a given bucket, this function
// heals `policy.json`, `notification.xml` and `listeners.json`. // heals `policy.json`, `notification.xml` and `listeners.json`.
func healBucketMetadata(storageDisks []StorageAPI, bucket string) error { func healBucketMetadata(storageDisks []StorageAPI, bucket string, readQuorum int) error {
healBucketMetaFn := func(metaPath string) error { healBucketMetaFn := func(metaPath string) error {
metaLock := nsMutex.NewNSLock(minioMetaBucket, metaPath) metaLock := nsMutex.NewNSLock(minioMetaBucket, metaPath)
metaLock.RLock() metaLock.RLock()
defer metaLock.RUnlock() defer metaLock.RUnlock()
// Heals the given file at metaPath. // Heals the given file at metaPath.
if err := healObject(storageDisks, minioMetaBucket, metaPath); err != nil && !isErrObjectNotFound(err) { if err := healObject(storageDisks, minioMetaBucket, metaPath, readQuorum); err != nil && !isErrObjectNotFound(err) {
return err return err
} // Success. } // Success.
return nil return nil
@ -200,7 +200,7 @@ func listBucketNames(storageDisks []StorageAPI) (bucketNames map[string]struct{}
// TODO :- // TODO :-
// - add support for healing dangling `uploads.json`. // - add support for healing dangling `uploads.json`.
// - add support for healing dangling `xl.json`. // - add support for healing dangling `xl.json`.
func quickHeal(storageDisks []StorageAPI, writeQuorum int) error { func quickHeal(storageDisks []StorageAPI, writeQuorum int, readQuorum int) error {
// List all bucket names from all disks. // List all bucket names from all disks.
bucketNames, err := listBucketNames(storageDisks) bucketNames, err := listBucketNames(storageDisks)
if err != nil { if err != nil {
@ -210,7 +210,7 @@ func quickHeal(storageDisks []StorageAPI, writeQuorum int) error {
for bucketName := range bucketNames { for bucketName := range bucketNames {
// Heal bucket and then proceed to heal bucket metadata. // Heal bucket and then proceed to heal bucket metadata.
if err = healBucket(storageDisks, bucketName, writeQuorum); err == nil { if err = healBucket(storageDisks, bucketName, writeQuorum); err == nil {
if err = healBucketMetadata(storageDisks, bucketName); err == nil { if err = healBucketMetadata(storageDisks, bucketName, readQuorum); err == nil {
continue continue
} }
return err return err
@ -221,10 +221,10 @@ func quickHeal(storageDisks []StorageAPI, writeQuorum int) error {
} }
// Heals an object only the corrupted/missing erasure blocks. // Heals an object only the corrupted/missing erasure blocks.
func healObject(storageDisks []StorageAPI, bucket string, object string) error { func healObject(storageDisks []StorageAPI, bucket string, object string, quorum int) error {
partsMetadata, errs := readAllXLMetadata(storageDisks, bucket, object) partsMetadata, errs := readAllXLMetadata(storageDisks, bucket, object)
if err := reduceErrs(errs, nil); err != nil { if reducedErr := reduceReadQuorumErrs(errs, nil, quorum); reducedErr != nil {
return toObjectErr(err, bucket, object) return toObjectErr(reducedErr, bucket, object)
} }
if !xlShouldHeal(partsMetadata, errs) { if !xlShouldHeal(partsMetadata, errs) {
@ -363,5 +363,5 @@ func (xl xlObjects) HealObject(bucket, object string) error {
defer objectLock.RUnlock() defer objectLock.RUnlock()
// Heal the object. // Heal the object.
return healObject(xl.storageDisks, bucket, object) return healObject(xl.storageDisks, bucket, object, xl.readQuorum)
} }

View File

@ -353,7 +353,7 @@ func TestQuickHeal(t *testing.T) {
} }
// Heal the missing buckets. // Heal the missing buckets.
if err = quickHeal(xl.storageDisks, xl.writeQuorum); err != nil { if err = quickHeal(xl.storageDisks, xl.writeQuorum, xl.readQuorum); err != nil {
t.Fatal(err) t.Fatal(err)
} }
@ -370,7 +370,7 @@ func TestQuickHeal(t *testing.T) {
t.Fatal("storage disk is not *posix type") t.Fatal("storage disk is not *posix type")
} }
xl.storageDisks[0] = newNaughtyDisk(posixDisk, nil, errUnformattedDisk) xl.storageDisks[0] = newNaughtyDisk(posixDisk, nil, errUnformattedDisk)
if err = quickHeal(xl.storageDisks, xl.writeQuorum); err != errUnformattedDisk { if err = quickHeal(xl.storageDisks, xl.writeQuorum, xl.readQuorum); err != errUnformattedDisk {
t.Fatal(err) t.Fatal(err)
} }
@ -392,7 +392,7 @@ func TestQuickHeal(t *testing.T) {
} }
xl = obj.(*xlObjects) xl = obj.(*xlObjects)
xl.storageDisks[0] = nil xl.storageDisks[0] = nil
if err = quickHeal(xl.storageDisks, xl.writeQuorum); err != nil { if err = quickHeal(xl.storageDisks, xl.writeQuorum, xl.readQuorum); err != nil {
t.Fatal("Got an unexpected error: ", err) t.Fatal("Got an unexpected error: ", err)
} }
@ -419,7 +419,7 @@ func TestQuickHeal(t *testing.T) {
t.Fatal("storage disk is not *posix type") t.Fatal("storage disk is not *posix type")
} }
xl.storageDisks[0] = newNaughtyDisk(posixDisk, nil, errDiskNotFound) xl.storageDisks[0] = newNaughtyDisk(posixDisk, nil, errDiskNotFound)
if err = quickHeal(xl.storageDisks, xl.writeQuorum); err != nil { if err = quickHeal(xl.storageDisks, xl.writeQuorum, xl.readQuorum); err != nil {
t.Fatal("Got an unexpected error: ", err) t.Fatal("Got an unexpected error: ", err)
} }
} }

View File

@ -335,8 +335,7 @@ func writeUniqueXLMetadata(disks []StorageAPI, bucket, prefix string, xlMetas []
deleteAllXLMetadata(disks, bucket, prefix, mErrs) deleteAllXLMetadata(disks, bucket, prefix, mErrs)
return traceError(errXLWriteQuorum) return traceError(errXLWriteQuorum)
} }
return reduceWriteQuorumErrs(mErrs, objectOpIgnoredErrs, quorum)
return reduceErrs(mErrs, objectOpIgnoredErrs)
} }
// writeSameXLMetadata - write `xl.json` on all disks in order. // writeSameXLMetadata - write `xl.json` on all disks in order.
@ -375,6 +374,5 @@ func writeSameXLMetadata(disks []StorageAPI, bucket, prefix string, xlMeta xlMet
deleteAllXLMetadata(disks, bucket, prefix, mErrs) deleteAllXLMetadata(disks, bucket, prefix, mErrs)
return traceError(errXLWriteQuorum) return traceError(errXLWriteQuorum)
} }
return reduceWriteQuorumErrs(mErrs, objectOpIgnoredErrs, writeQuorum)
return reduceErrs(mErrs, objectOpIgnoredErrs)
} }

View File

@ -140,7 +140,10 @@ func (xl xlObjects) updateUploadJSON(bucket, object string, uCh uploadIDChange)
} }
wg.Wait() wg.Wait()
return reduceErrs(errs, objectOpIgnoredErrs) if reducedErr := reduceWriteQuorumErrs(errs, objectOpIgnoredErrs, xl.writeQuorum); reducedErr != nil {
return reducedErr
}
return nil
} }
// Returns if the prefix is a multipart upload. // Returns if the prefix is a multipart upload.
@ -251,5 +254,8 @@ func commitXLMetadata(disks []StorageAPI, srcPrefix, dstPrefix string, quorum in
return traceError(errXLWriteQuorum) return traceError(errXLWriteQuorum)
} }
return reduceErrs(mErrs, objectOpIgnoredErrs) if reducedErr := reduceWriteQuorumErrs(mErrs, objectOpIgnoredErrs, quorum); reducedErr != nil {
return reducedErr
}
return nil
} }

View File

@ -79,7 +79,7 @@ func (xl xlObjects) GetObject(bucket, object string, startOffset int64, length i
return traceError(InsufficientReadQuorum{}, errs...) return traceError(InsufficientReadQuorum{}, errs...)
} }
if reducedErr := reduceErrs(errs, objectOpIgnoredErrs); reducedErr != nil { if reducedErr := reduceReadQuorumErrs(errs, objectOpIgnoredErrs, xl.readQuorum); reducedErr != nil {
return toObjectErr(reducedErr, bucket, object) return toObjectErr(reducedErr, bucket, object)
} }
@ -339,8 +339,7 @@ func rename(disks []StorageAPI, srcBucket, srcEntry, dstBucket, dstEntry string,
undoRename(disks, srcBucket, srcEntry, dstBucket, dstEntry, isPart, errs) undoRename(disks, srcBucket, srcEntry, dstBucket, dstEntry, isPart, errs)
return traceError(errXLWriteQuorum) return traceError(errXLWriteQuorum)
} }
// Return on first error, also undo any partially successful rename operations. return reduceWriteQuorumErrs(errs, objectOpIgnoredErrs, quorum)
return reduceErrs(errs, objectOpIgnoredErrs)
} }
// renamePart - renames a part of the source object to the destination // renamePart - renames a part of the source object to the destination

View File

@ -31,8 +31,7 @@ import (
// golang's map orders keys. This doesn't affect correctness as long as quorum // golang's map orders keys. This doesn't affect correctness as long as quorum
// value is greater than or equal to simple majority, since none of the equally // value is greater than or equal to simple majority, since none of the equally
// maximal values would occur quorum or more number of times. // maximal values would occur quorum or more number of times.
func reduceErrs(errs []error, ignoredErrs []error) (maxCount int, maxErr error) {
func reduceErrs(errs []error, ignoredErrs []error) error {
errorCounts := make(map[error]int) errorCounts := make(map[error]int)
errs = errorsCause(errs) errs = errorsCause(errs)
for _, err := range errs { for _, err := range errs {
@ -42,14 +41,45 @@ func reduceErrs(errs []error, ignoredErrs []error) error {
errorCounts[err]++ errorCounts[err]++
} }
max := 0 max := 0
var errMax error
for err, count := range errorCounts { for err, count := range errorCounts {
if max < count { if max < count {
max = count max = count
errMax = err maxErr = err
} }
} }
return traceError(errMax, errs...) return max, maxErr
}
// reduceQuorumErrs behaves like reduceErrs by only for returning
// values of maximally occurring errors validated against a generic
// quorum number can be read or write quorum depending on usage.
// additionally a special error is provided as well to be returned
// in case quorum is not satisfied.
func reduceQuorumErrs(errs []error, ignoredErrs []error, quorum int, quorumErr error) (maxErr error) {
maxCount, maxErr := reduceErrs(errs, ignoredErrs)
if maxErr == nil && maxCount >= quorum {
// Success in quorum.
return nil
}
if maxErr != nil && maxCount >= quorum {
// Errors in quorum.
return traceError(maxErr, errs...)
}
// No quorum satisfied.
maxErr = traceError(quorumErr, errs...)
return
}
// reduceReadQuorumErrs behaves like reduceErrs but only for returning
// values of maximally occurring errors validated against readQuorum.
func reduceReadQuorumErrs(errs []error, ignoredErrs []error, readQuorum int) (maxErr error) {
return reduceQuorumErrs(errs, ignoredErrs, readQuorum, errXLReadQuorum)
}
// reduceWriteQuorumErrs behaves like reduceErrs but only for returning
// values of maximally occurring errors validated against writeQuorum.
func reduceWriteQuorumErrs(errs []error, ignoredErrs []error, writeQuorum int) (maxErr error) {
return reduceQuorumErrs(errs, ignoredErrs, writeQuorum, errXLWriteQuorum)
} }
// List of all errors which are ignored while verifying quorum. // List of all errors which are ignored while verifying quorum.

View File

@ -63,29 +63,35 @@ func TestReduceErrs(t *testing.T) {
errDiskNotFound, errDiskNotFound,
errDiskNotFound, errDiskNotFound,
errDiskFull, errDiskFull,
}, []error{}, errDiskNotFound}, }, []error{}, errXLReadQuorum},
// Validate if have no consensus. // Validate if have no consensus.
{[]error{ {[]error{
errDiskFull, errDiskFull,
errDiskNotFound, errDiskNotFound,
nil, nil, nil, nil,
}, []error{}, nil}, }, []error{}, errXLReadQuorum},
// Validate if have consensus and errors ignored. // Validate if have consensus and errors ignored.
{[]error{ {[]error{
errVolumeNotFound,
errVolumeNotFound,
errVolumeNotFound, errVolumeNotFound,
errVolumeNotFound, errVolumeNotFound,
errVolumeNotFound, errVolumeNotFound,
errDiskNotFound, errDiskNotFound,
errDiskNotFound, errDiskNotFound,
}, []error{errDiskNotFound}, errVolumeNotFound}, }, []error{errDiskNotFound}, errVolumeNotFound},
{[]error{}, []error{}, nil}, {[]error{}, []error{}, errXLReadQuorum},
} }
// Validates list of all the testcases for returning valid errors. // Validates list of all the testcases for returning valid errors.
for i, testCase := range testCases { for i, testCase := range testCases {
gotErr := reduceErrs(testCase.errs, testCase.ignoredErrs) gotErr := reduceReadQuorumErrs(testCase.errs, testCase.ignoredErrs, 5)
if errorCause(gotErr) != testCase.err { if errorCause(gotErr) != testCase.err {
t.Errorf("Test %d : expected %s, got %s", i+1, testCase.err, gotErr) t.Errorf("Test %d : expected %s, got %s", i+1, testCase.err, gotErr)
} }
gotNewErr := reduceWriteQuorumErrs(testCase.errs, testCase.ignoredErrs, 6)
if errorCause(gotNewErr) != errXLWriteQuorum {
t.Errorf("Test %d : expected %s, got %s", i+1, errXLWriteQuorum, gotErr)
}
} }
} }

View File

@ -124,7 +124,7 @@ func newXLObjects(storageDisks []StorageAPI) (ObjectLayer, error) {
xl.writeQuorum = writeQuorum xl.writeQuorum = writeQuorum
// Do a quick heal on the buckets themselves for any discrepancies. // Do a quick heal on the buckets themselves for any discrepancies.
if err := quickHeal(xl.storageDisks, xl.writeQuorum); err != nil { if err := quickHeal(xl.storageDisks, xl.writeQuorum, xl.readQuorum); err != nil {
return xl, err return xl, err
} }