cleanup: Move code in *-multipart-common.go to *-multipart.go (#3638)

The code in *-multipart-common.go is not common anymore.

cmd/xl-v1-multipart.go

@@ -25,12 +25,260 @@ import (
 	"io/ioutil"
 	"path"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/minio/minio/pkg/mimedb"
 	"github.com/minio/sha256-simd"
 )
 
+// updateUploadJSON - add or remove upload ID info in all `uploads.json`.
+func (xl xlObjects) updateUploadJSON(bucket, object, uploadID string, initiated time.Time, isRemove bool) error {
+	uploadsPath := path.Join(bucket, object, uploadsJSONFile)
+	tmpUploadsPath := mustGetUUID()
+
+	// slice to store errors from disks
+	errs := make([]error, len(xl.storageDisks))
+	// slice to store if it is a delete operation on a disk
+	isDelete := make([]bool, len(xl.storageDisks))
+
+	wg := sync.WaitGroup{}
+	for index, disk := range xl.storageDisks {
+		if disk == nil {
+			errs[index] = traceError(errDiskNotFound)
+			continue
+		}
+		// Update `uploads.json` in a go routine.
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+
+			// read and parse uploads.json on this disk
+			uploadsJSON, err := readUploadsJSON(bucket, object, disk)
+			if errorCause(err) == errFileNotFound {
+				// If file is not found, we assume an
+				// default (empty) upload info.
+				uploadsJSON, err = newUploadsV1("xl"), nil
+			}
+			// If we have a read error, we store error and
+			// exit.
+			if err != nil {
+				errs[index] = err
+				return
+			}
+
+			if !isRemove {
+				// Add the uploadID
+				uploadsJSON.AddUploadID(uploadID, initiated)
+			} else {
+				// Remove the upload ID
+				uploadsJSON.RemoveUploadID(uploadID)
+				if len(uploadsJSON.Uploads) == 0 {
+					isDelete[index] = true
+				}
+			}
+
+			// For delete, rename to tmp, for the
+			// possibility of recovery in case of quorum
+			// failure.
+			if !isDelete[index] {
+				errs[index] = writeUploadJSON(&uploadsJSON, uploadsPath, tmpUploadsPath, disk)
+			} else {
+				wErr := disk.RenameFile(minioMetaMultipartBucket, uploadsPath, minioMetaTmpBucket, tmpUploadsPath)
+				if wErr != nil {
+					errs[index] = traceError(wErr)
+				}
+
+			}
+		}(index, disk)
+	}
+
+	// Wait for all the writes to finish.
+	wg.Wait()
+
+	// Do we have write quorum?
+	if !isDiskQuorum(errs, xl.writeQuorum) {
+		// No quorum. Perform cleanup on the minority of disks
+		// on which the operation succeeded.
+
+		// There are two cases:
+		//
+		// 1. uploads.json file was updated -> we delete the
+		//    file that we successfully overwrote on the
+		//    minority of disks, so that the failed quorum
+		//    operation is not partially visible.
+		//
+		// 2. uploads.json was deleted -> in this case since
+		//    the delete failed, we restore from tmp.
+		for index, disk := range xl.storageDisks {
+			if disk == nil || errs[index] != nil {
+				continue
+			}
+			wg.Add(1)
+			go func(index int, disk StorageAPI) {
+				defer wg.Done()
+				if !isDelete[index] {
+					_ = disk.DeleteFile(
+						minioMetaMultipartBucket,
+						uploadsPath,
+					)
+				} else {
+					_ = disk.RenameFile(
+						minioMetaTmpBucket, tmpUploadsPath,
+						minioMetaMultipartBucket, uploadsPath,
+					)
+				}
+			}(index, disk)
+		}
+		wg.Wait()
+		return traceError(errXLWriteQuorum)
+	}
+
+	// we do have quorum, so in case of delete upload.json file
+	// operation, we purge from tmp.
+	for index, disk := range xl.storageDisks {
+		if disk == nil || !isDelete[index] {
+			continue
+		}
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			// isDelete[index] = true at this point.
+			_ = disk.DeleteFile(minioMetaTmpBucket, tmpUploadsPath)
+		}(index, disk)
+	}
+	wg.Wait()
+
+	if reducedErr := reduceWriteQuorumErrs(errs, objectOpIgnoredErrs, xl.writeQuorum); reducedErr != nil {
+		return reducedErr
+	}
+	return nil
+}
+
+// addUploadID - add upload ID and its initiated time to 'uploads.json'.
+func (xl xlObjects) addUploadID(bucket, object string, uploadID string, initiated time.Time) error {
+	return xl.updateUploadJSON(bucket, object, uploadID, initiated, false)
+}
+
+// removeUploadID - remove upload ID in 'uploads.json'.
+func (xl xlObjects) removeUploadID(bucket, object string, uploadID string) error {
+	return xl.updateUploadJSON(bucket, object, uploadID, time.Time{}, true)
+}
+
+// Returns if the prefix is a multipart upload.
+func (xl xlObjects) isMultipartUpload(bucket, prefix string) bool {
+	for _, disk := range xl.getLoadBalancedDisks() {
+		if disk == nil {
+			continue
+		}
+		_, err := disk.StatFile(bucket, pathJoin(prefix, uploadsJSONFile))
+		if err == nil {
+			return true
+		}
+		// For any reason disk was deleted or goes offline, continue
+		if isErrIgnored(err, objMetadataOpIgnoredErrs...) {
+			continue
+		}
+		break
+	}
+	return false
+}
+
+// isUploadIDExists - verify if a given uploadID exists and is valid.
+func (xl xlObjects) isUploadIDExists(bucket, object, uploadID string) bool {
+	uploadIDPath := path.Join(bucket, object, uploadID)
+	return xl.isObject(minioMetaMultipartBucket, uploadIDPath)
+}
+
+// Removes part given by partName belonging to a mulitpart upload from minioMetaBucket
+func (xl xlObjects) removeObjectPart(bucket, object, uploadID, partName string) {
+	curpartPath := path.Join(bucket, object, uploadID, partName)
+	wg := sync.WaitGroup{}
+	for i, disk := range xl.storageDisks {
+		if disk == nil {
+			continue
+		}
+		wg.Add(1)
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			// Ignoring failure to remove parts that weren't present in CompleteMultipartUpload
+			// requests. xl.json is the authoritative source of truth on which parts constitute
+			// the object. The presence of parts that don't belong in the object doesn't affect correctness.
+			_ = disk.DeleteFile(minioMetaMultipartBucket, curpartPath)
+		}(i, disk)
+	}
+	wg.Wait()
+}
+
+// statPart - returns fileInfo structure for a successful stat on part file.
+func (xl xlObjects) statPart(bucket, object, uploadID, partName string) (fileInfo FileInfo, err error) {
+	partNamePath := path.Join(bucket, object, uploadID, partName)
+	for _, disk := range xl.getLoadBalancedDisks() {
+		if disk == nil {
+			continue
+		}
+		fileInfo, err = disk.StatFile(minioMetaMultipartBucket, partNamePath)
+		if err == nil {
+			return fileInfo, nil
+		}
+		err = traceError(err)
+		// For any reason disk was deleted or goes offline we continue to next disk.
+		if isErrIgnored(err, objMetadataOpIgnoredErrs...) {
+			continue
+		}
+
+		// Catastrophic error, we return.
+		break
+	}
+	return FileInfo{}, err
+}
+
+// commitXLMetadata - commit `xl.json` from source prefix to destination prefix in the given slice of disks.
+func commitXLMetadata(disks []StorageAPI, srcBucket, srcPrefix, dstBucket, dstPrefix string, quorum int) error {
+	var wg = &sync.WaitGroup{}
+	var mErrs = make([]error, len(disks))
+
+	srcJSONFile := path.Join(srcPrefix, xlMetaJSONFile)
+	dstJSONFile := path.Join(dstPrefix, xlMetaJSONFile)
+
+	// Rename `xl.json` to all disks in parallel.
+	for index, disk := range disks {
+		if disk == nil {
+			mErrs[index] = traceError(errDiskNotFound)
+			continue
+		}
+		wg.Add(1)
+		// Rename `xl.json` in a routine.
+		go func(index int, disk StorageAPI) {
+			defer wg.Done()
+			// Delete any dangling directories.
+			defer disk.DeleteFile(srcBucket, srcPrefix)
+
+			// Renames `xl.json` from source prefix to destination prefix.
+			rErr := disk.RenameFile(srcBucket, srcJSONFile, dstBucket, dstJSONFile)
+			if rErr != nil {
+				mErrs[index] = traceError(rErr)
+				return
+			}
+			mErrs[index] = nil
+		}(index, disk)
+	}
+	// Wait for all the routines.
+	wg.Wait()
+
+	// Do we have write Quorum?.
+	if !isDiskQuorum(mErrs, quorum) {
+		// Delete all `xl.json` successfully renamed.
+		deleteAllXLMetadata(disks, dstBucket, dstPrefix, mErrs)
+		return traceError(errXLWriteQuorum)
+	}
+
+	if reducedErr := reduceWriteQuorumErrs(mErrs, objectOpIgnoredErrs, quorum); reducedErr != nil {
+		return reducedErr
+	}
+	return nil
+}
+
 // listMultipartUploads - lists all multipart uploads.
 func (xl xlObjects) listMultipartUploads(bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) {
 	result := ListMultipartsInfo{