xl: add quorum support for DeleteFile() (#1426)

Fixes #1396

xl-erasure-v1-metadata.go

@@ -29,6 +29,7 @@ type xlMetaV1 struct {
 		Size    int64     `json:"size"`
 		ModTime time.Time `json:"modTime"`
 		Version int64     `json:"version"`
+		Deleted bool      `json:"deleted"`
 	} `json:"stat"`
 	Erasure struct {
 		DataBlocks   int   `json:"data"`

xl-erasure-v1.go

@@ -18,6 +18,7 @@ package main
 
 import (
 	"fmt"
+	"io"
 	"os"
 	slashpath "path"
 	"sort"
@@ -547,10 +548,54 @@ func (xl XL) DeleteFile(volume, path string) error {
 	if !isValidPath(path) {
 		return errInvalidArgument
 	}
+
+	// Lock right before reading from disk.
+	nsMutex.RLock(volume, path)
+	partsMetadata, errs := xl.getPartsMetadata(volume, path)
+	nsMutex.RUnlock(volume, path)
+
+	// List all the file versions on existing files.
+	versions, err := listFileVersions(partsMetadata, errs)
+	// Get highest file version.
+	higherVersion := highestInt(versions)
+	// Increment to have next higher version.
+	higherVersion++
+
+	// Take last meta data to use later
+	var mdata xlMetaV1
+	onlineDisksCount := 0
+	for index, metadata := range partsMetadata {
+		if errs[index] == nil {
+			mdata = metadata
+			onlineDisksCount++
+		}
+	}
+
+	if onlineDisksCount < xl.writeQuorum {
+		return errWriteQuorum
+	}
+
+	xlMetaV1FilePath := slashpath.Join(path, xlMetaV1File)
+	deleteMetaData := (onlineDisksCount == len(xl.storageDisks))
+
+	// Set higher version to indicate file operation
+	mdata.Stat.Version = higherVersion
+	mdata.Stat.Deleted = true
+
+	nsMutex.Lock(volume, path)
+	defer nsMutex.Unlock(volume, path)
+
 	// Loop through and delete each chunks.
 	for index, disk := range xl.storageDisks {
-		erasureFilePart := slashpath.Join(path, fmt.Sprintf("file.%d", index))
-		err := disk.DeleteFile(volume, erasureFilePart)
+		if errs[index] != nil {
+			continue
+		}
+
+		// TODO: errors can be allowed up to len(xl.storageDisks) - xl.writeQuorum
+		// TODO: Fix: meta data will be lost if deleteMetaData is true and anyone DeleteFile failure.
+
+		if deleteMetaData {
+			err = disk.DeleteFile(volume, xlMetaV1FilePath)
 			if err != nil {
 				log.WithFields(logrus.Fields{
 					"volume": volume,
@@ -558,8 +603,30 @@ func (xl XL) DeleteFile(volume, path string) error {
 				}).Errorf("DeleteFile failed with %s", err)
 				return err
 			}
-		xlMetaV1FilePath := slashpath.Join(path, xlMetaV1File)
-		err = disk.DeleteFile(volume, xlMetaV1FilePath)
+		} else {
+			var metadataWriter io.WriteCloser
+			metadataWriter, err = disk.CreateFile(volume, xlMetaV1FilePath)
+			if err != nil {
+				log.WithFields(logrus.Fields{
+					"volume": volume,
+					"path":   path,
+				}).Errorf("CreateFile failed with %s", err)
+				return err
+			}
+
+			err = mdata.Write(metadataWriter)
+			if err != nil {
+				log.WithFields(logrus.Fields{
+					"volume":    volume,
+					"path":      path,
+					"diskIndex": index,
+				}).Errorf("Writing metadata failed with %s", err)
+				return err
+			}
+		}
+
+		erasureFilePart := slashpath.Join(path, fmt.Sprintf("file.%d", index))
+		err = disk.DeleteFile(volume, erasureFilePart)
 		if err != nil {
 			log.WithFields(logrus.Fields{
 				"volume": volume,