MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-04-22 11:26:36 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix object deletion where quorum is not available (#5709)

Browse Source 
Object deletion should not be possible if quorum is not
available. This PR updates deleteObject() to check for
quorum errors before proceeding with object deletion.

Fixes #5535
This commit is contained in:
Nitish Tiwari 2018-03-27 05:09:28 +05:30 committed by Dee Koder
parent de44be86d0
commit 58291acb8d
1 changed files with 11 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
cmd/xl-v1-object.go
View File

@ -730,17 +730,20 @@ func (xl xlObjects) deleteObject(bucket, object string) error {
isDir := hasSuffix(object, slashSeparator) isDir := hasSuffix(object, slashSeparator)
// If its a directory request, no need to read metadata.
if !isDir { if !isDir {
// Read metadata associated with the object from all disks. // Read metadata associated with the object from all disks.
metaArr, errs := readAllXLMetadata(xl.getDisks(), bucket, object) metaArr, errs := readAllXLMetadata(xl.getDisks(), bucket, object)
// get Quorum for this object // get Quorum for this object
_, writeQuorum, err = objectQuorumFromMeta(xl, metaArr, errs) _, writeQuorum, err = objectQuorumFromMeta(xl, metaArr, errs)
if err != nil { if err != nil {
return err return err
} }
err = reduceWriteQuorumErrs(errs, objectOpIgnoredErrs, writeQuorum)
if err != nil {
return err
}
} else { } else {
// WriteQuorum is defaulted to N/2 + 1 for directories
writeQuorum = len(xl.getDisks())/2 + 1 writeQuorum = len(xl.getDisks())/2 + 1
} }
@ -755,16 +758,16 @@ func (xl xlObjects) deleteObject(bucket, object string) error {
wg.Add(1) wg.Add(1)
go func(index int, disk StorageAPI, isDir bool) { go func(index int, disk StorageAPI, isDir bool) {
defer wg.Done() defer wg.Done()
var err error var e error
if isDir { if isDir {
// DeleteFile() simply tries to remove a directory // DeleteFile() simply tries to remove a directory
// and will succeed only if that directory is empty. // and will succeed only if that directory is empty.
err = disk.DeleteFile(bucket, object) e = disk.DeleteFile(bucket, object)
} else { } else {
err = cleanupDir(disk, bucket, object) e = cleanupDir(disk, bucket, object)
} }
if err != nil && errors.Cause(err) != errVolumeNotFound { if e != nil && errors.Cause(e) != errVolumeNotFound {
dErrs[index] = err dErrs[index] = e
} }
}(index, disk, isDir) }(index, disk, isDir)
} }
@ -772,6 +775,7 @@ func (xl xlObjects) deleteObject(bucket, object string) error {
// Wait for all routines to finish. // Wait for all routines to finish.
wg.Wait() wg.Wait()
// return errors if any during deletion
return reduceWriteQuorumErrs(dErrs, objectOpIgnoredErrs, writeQuorum) return reduceWriteQuorumErrs(dErrs, objectOpIgnoredErrs, writeQuorum)
} }