Support creating directories on erasure coded backend (#5443)

This PR continues from #5049 where we started supporting
directories for erasure coded backend

cmd/xl-v1-object.go

@@ -33,6 +33,29 @@ import (
 // list all errors which can be ignored in object operations.
 var objectOpIgnoredErrs = append(baseIgnoredErrs, errDiskAccessDenied)
 
+// putObjectDir hints the bottom layer to create a new directory.
+func (xl xlObjects) putObjectDir(bucket, object string, writeQuorum int) error {
+	var wg = &sync.WaitGroup{}
+
+	errs := make([]error, len(xl.storageDisks))
+	// Prepare object creation in all disks
+	for index, disk := range xl.storageDisks {
+		if disk == nil {
+			continue
+		}
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			if err := disk.MakeVol(pathJoin(bucket, object)); err != nil && err != errVolumeExists {
+				errs[index] = err
+			}
+		}(index, disk)
+	}
+	wg.Wait()
+
+	return reduceWriteQuorumErrs(errs, objectOpIgnoredErrs, writeQuorum)
+}
+
 // prepareFile hints the bottom layer to optimize the creation of a new object
 func (xl xlObjects) prepareFile(bucket, object string, size int64, onlineDisks []StorageAPI, blockSize int64, dataBlocks, writeQuorum int) error {
 	pErrs := make([]error, len(onlineDisks))
@@ -204,6 +227,12 @@ func (xl xlObjects) getObject(bucket, object string, startOffset int64, length i
 		return errors.Trace(errUnexpected)
 	}
 
+	// If its a directory request, we return an empty body.
+	if hasSuffix(object, slashSeparator) {
+		_, err := writer.Write([]byte(""))
+		return toObjectErr(errors.Trace(err), bucket, object)
+	}
+
 	// Read metadata associated with the object from all disks.
 	metaArr, errs := readAllXLMetadata(xl.storageDisks, bucket, object)
 
@@ -354,6 +383,42 @@ func (xl xlObjects) getObject(bucket, object string, startOffset int64, length i
 	return nil
 }
 
+// getObjectInfoDir - This getObjectInfo is specific to object directory lookup.
+func (xl xlObjects) getObjectInfoDir(bucket, object string) (oi ObjectInfo, err error) {
+	var wg = &sync.WaitGroup{}
+
+	errs := make([]error, len(xl.storageDisks))
+	// Prepare object creation in a all disks
+	for index, disk := range xl.storageDisks {
+		if disk == nil {
+			continue
+		}
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			if _, err := disk.StatVol(pathJoin(bucket, object)); err != nil {
+				// Since we are re-purposing StatVol, an object which
+				// is a directory if it doesn't exist should be
+				// returned as errFileNotFound instead, convert
+				// the error right here accordingly.
+				if err == errVolumeNotFound {
+					err = errFileNotFound
+				} else if err == errVolumeAccessDenied {
+					err = errFileAccessDenied
+				}
+
+				// Save error to reduce it later
+				errs[index] = err
+			}
+		}(index, disk)
+	}
+
+	wg.Wait()
+
+	readQuorum := len(xl.storageDisks) / 2
+	return dirObjectInfo(bucket, object, 0, map[string]string{}), reduceReadQuorumErrs(errs, objectOpIgnoredErrs, readQuorum)
+}
+
 // GetObjectInfo - reads object metadata and replies back ObjectInfo.
 func (xl xlObjects) GetObjectInfo(bucket, object string) (oi ObjectInfo, e error) {
 	// Lock the object before reading.
@@ -367,15 +432,21 @@ func (xl xlObjects) GetObjectInfo(bucket, object string) (oi ObjectInfo, e error
 		return oi, err
 	}
 
+	if hasSuffix(object, slashSeparator) {
+		return xl.getObjectInfoDir(bucket, object)
+	}
+
 	info, err := xl.getObjectInfo(bucket, object)
 	if err != nil {
 		return oi, toObjectErr(err, bucket, object)
 	}
+
 	return info, nil
 }
 
 // getObjectInfo - wrapper for reading object metadata and constructs ObjectInfo.
 func (xl xlObjects) getObjectInfo(bucket, object string) (objInfo ObjectInfo, err error) {
+
 	// Extracts xlStat and xlMetaMap.
 	xlStat, xlMetaMap, err := xl.readXLMetaStat(bucket, object)
 	if err != nil {
@@ -508,6 +579,26 @@ func (xl xlObjects) PutObject(bucket string, object string, data *hash.Reader, m
 
 // putObject wrapper for xl PutObject
 func (xl xlObjects) putObject(bucket string, object string, data *hash.Reader, metadata map[string]string) (objInfo ObjectInfo, err error) {
+	uniqueID := mustGetUUID()
+	tempObj := uniqueID
+
+	// No metadata is set, allocate a new one.
+	if metadata == nil {
+		metadata = make(map[string]string)
+	}
+
+	// Get parity and data drive count based on storage class metadata
+	dataDrives, parityDrives := getRedundancyCount(metadata[amzStorageClass], len(xl.storageDisks))
+
+	// we now know the number of blocks this object needs for data and parity.
+	// writeQuorum is dataBlocks + 1
+	writeQuorum := dataDrives + 1
+
+	// Delete temporary object in the event of failure.
+	// If PutObject succeeded there would be no temporary
+	// object to delete.
+	defer xl.deleteObject(minioMetaTmpBucket, tempObj)
+
 	// This is a special case with size as '0' and object ends with
 	// a slash separator, we treat it like a valid operation and
 	// return success.
@@ -518,6 +609,16 @@ func (xl xlObjects) putObject(bucket string, object string, data *hash.Reader, m
 		if xl.parentDirIsObject(bucket, path.Dir(object)) {
 			return ObjectInfo{}, toObjectErr(errors.Trace(errFileAccessDenied), bucket, object)
 		}
+
+		if err = xl.putObjectDir(minioMetaTmpBucket, tempObj, writeQuorum); err != nil {
+			return ObjectInfo{}, toObjectErr(errors.Trace(err), bucket, object)
+		}
+
+		// Rename the successfully written temporary object to final location.
+		if _, err = renameObject(xl.storageDisks, minioMetaTmpBucket, tempObj, bucket, object, writeQuorum); err != nil {
+			return ObjectInfo{}, toObjectErr(err, bucket, object)
+		}
+
 		return dirObjectInfo(bucket, object, data.Size(), metadata), nil
 	}
 
@@ -538,14 +639,6 @@ func (xl xlObjects) putObject(bucket string, object string, data *hash.Reader, m
 		return ObjectInfo{}, toObjectErr(errors.Trace(errFileAccessDenied), bucket, object)
 	}
 
-	// No metadata is set, allocate a new one.
-	if metadata == nil {
-		metadata = make(map[string]string)
-	}
-
-	uniqueID := mustGetUUID()
-	tempObj := uniqueID
-
 	// Limit the reader to its provided size if specified.
 	var reader io.Reader = data
 
@@ -569,12 +662,6 @@ func (xl xlObjects) putObject(bucket string, object string, data *hash.Reader, m
 			}
 		}
 	}
-	// Get parity and data drive count based on storage class metadata
-	dataDrives, parityDrives := getRedundancyCount(metadata[amzStorageClass], len(xl.storageDisks))
-
-	// we now know the number of blocks this object needs for data and parity.
-	// writeQuorum is dataBlocks + 1
-	writeQuorum := dataDrives + 1
 
 	// Initialize parts metadata
 	partsMetadata := make([]xlMetaV1, len(xl.storageDisks))
@@ -589,11 +676,6 @@ func (xl xlObjects) putObject(bucket string, object string, data *hash.Reader, m
 	// Order disks according to erasure distribution
 	onlineDisks := shuffleDisks(xl.storageDisks, partsMetadata[0].Erasure.Distribution)
 
-	// Delete temporary object in the event of failure.
-	// If PutObject succeeded there would be no temporary
-	// object to delete.
-	defer xl.deleteObject(minioMetaTmpBucket, tempObj)
-
 	// Total size of the written object
 	var sizeWritten int64
 
@@ -741,13 +823,20 @@ func (xl xlObjects) deleteObject(bucket, object string) error {
 	// Initialize sync waitgroup.
 	var wg = &sync.WaitGroup{}
 
-	// Read metadata associated with the object from all disks.
-	metaArr, errs := readAllXLMetadata(xl.storageDisks, bucket, object)
+	var writeQuorum int
+	var err error
+	// If its a directory request, no need to read metadata.
+	if !hasSuffix(object, slashSeparator) {
+		// Read metadata associated with the object from all disks.
+		metaArr, errs := readAllXLMetadata(xl.storageDisks, bucket, object)
 
-	// get Quorum for this object
-	_, writeQuorum, err := objectQuorumFromMeta(xl, metaArr, errs)
-	if err != nil {
-		return err
+		// get Quorum for this object
+		_, writeQuorum, err = objectQuorumFromMeta(xl, metaArr, errs)
+		if err != nil {
+			return err
+		}
+	} else {
+		writeQuorum = len(xl.storageDisks)/2 + 1
 	}
 
 	// Initialize list of errors.
@@ -789,14 +878,20 @@ func (xl xlObjects) DeleteObject(bucket, object string) (err error) {
 		return err
 	}
 
+	if hasSuffix(object, slashSeparator) {
+		// Delete the object on all disks.
+		if err = xl.deleteObject(bucket, object); err != nil {
+			return toObjectErr(err, bucket, object)
+		}
+	}
+
 	// Validate object exists.
 	if !xl.isObject(bucket, object) {
 		return errors.Trace(ObjectNotFound{bucket, object})
 	} // else proceed to delete the object.
 
 	// Delete the object on all disks.
-	err = xl.deleteObject(bucket, object)
-	if err != nil {
+	if err = xl.deleteObject(bucket, object); err != nil {
 		return toObjectErr(err, bucket, object)
 	}