2016-05-21 05:48:47 +02:00
|
|
|
/*
|
|
|
|
* Minio Cloud Storage, (C) 2016 Minio, Inc.
|
|
|
|
*
|
|
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
* you may not use this file except in compliance with the License.
|
|
|
|
* You may obtain a copy of the License at
|
|
|
|
*
|
|
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
*
|
|
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
* See the License for the specific language governing permissions and
|
|
|
|
* limitations under the License.
|
|
|
|
*/
|
|
|
|
|
2016-08-19 01:23:42 +02:00
|
|
|
package cmd
|
2016-05-21 05:48:47 +02:00
|
|
|
|
|
|
|
import (
|
|
|
|
"path"
|
|
|
|
"sync"
|
2016-11-23 00:29:39 +01:00
|
|
|
"time"
|
2016-05-21 05:48:47 +02:00
|
|
|
)
|
|
|
|
|
2016-11-23 00:29:39 +01:00
|
|
|
// 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 {
|
2016-11-22 22:15:06 +01:00
|
|
|
uploadsPath := path.Join(bucket, object, uploadsJSONFile)
|
2016-11-23 01:52:37 +01:00
|
|
|
tmpUploadsPath := mustGetUUID()
|
2016-05-21 05:48:47 +02:00
|
|
|
|
2016-10-30 17:27:29 +01:00
|
|
|
// 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{}
|
2016-06-29 11:10:40 +02:00
|
|
|
for index, disk := range xl.storageDisks {
|
2016-06-03 01:34:15 +02:00
|
|
|
if disk == nil {
|
2016-10-30 17:27:29 +01:00
|
|
|
errs[index] = traceError(errDiskNotFound)
|
2016-06-03 01:34:15 +02:00
|
|
|
continue
|
|
|
|
}
|
2016-10-30 17:27:29 +01:00
|
|
|
// Update `uploads.json` in a go routine.
|
2016-05-21 05:48:47 +02:00
|
|
|
wg.Add(1)
|
|
|
|
go func(index int, disk StorageAPI) {
|
|
|
|
defer wg.Done()
|
2016-10-30 17:27:29 +01:00
|
|
|
|
|
|
|
// 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
|
2016-05-21 05:48:47 +02:00
|
|
|
}
|
2016-10-30 17:27:29 +01:00
|
|
|
// If we have a read error, we store error and
|
|
|
|
// exit.
|
|
|
|
if err != nil {
|
2016-11-02 04:04:32 +01:00
|
|
|
errs[index] = err
|
2016-05-21 05:48:47 +02:00
|
|
|
return
|
|
|
|
}
|
2016-10-30 17:27:29 +01:00
|
|
|
|
2016-11-23 00:29:39 +01:00
|
|
|
if !isRemove {
|
2016-10-30 17:27:29 +01:00
|
|
|
// Add the uploadID
|
2016-11-23 00:29:39 +01:00
|
|
|
uploadsJSON.AddUploadID(uploadID, initiated)
|
2016-10-30 17:27:29 +01:00
|
|
|
} else {
|
|
|
|
// Remove the upload ID
|
2016-11-23 00:29:39 +01:00
|
|
|
uploadsJSON.RemoveUploadID(uploadID)
|
2016-10-30 17:27:29 +01:00
|
|
|
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 {
|
2016-11-22 22:15:06 +01:00
|
|
|
wErr := disk.RenameFile(minioMetaMultipartBucket, uploadsPath, minioMetaTmpBucket, tmpUploadsPath)
|
2016-10-30 17:27:29 +01:00
|
|
|
if wErr != nil {
|
|
|
|
errs[index] = traceError(wErr)
|
|
|
|
}
|
|
|
|
|
2016-05-21 05:48:47 +02:00
|
|
|
}
|
|
|
|
}(index, disk)
|
|
|
|
}
|
|
|
|
|
2016-10-30 17:27:29 +01:00
|
|
|
// Wait for all the writes to finish.
|
2016-05-21 05:48:47 +02:00
|
|
|
wg.Wait()
|
|
|
|
|
2016-10-30 17:27:29 +01:00
|
|
|
// Do we have write quorum?
|
2016-07-13 09:29:48 +02:00
|
|
|
if !isDiskQuorum(errs, xl.writeQuorum) {
|
2016-10-30 17:27:29 +01:00
|
|
|
// 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.
|
2016-06-29 11:10:40 +02:00
|
|
|
for index, disk := range xl.storageDisks {
|
2016-10-30 17:27:29 +01:00
|
|
|
if disk == nil || errs[index] != nil {
|
2016-06-03 01:34:15 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
wg.Add(1)
|
|
|
|
go func(index int, disk StorageAPI) {
|
|
|
|
defer wg.Done()
|
2016-10-30 17:27:29 +01:00
|
|
|
if !isDelete[index] {
|
|
|
|
_ = disk.DeleteFile(
|
2016-11-22 22:15:06 +01:00
|
|
|
minioMetaMultipartBucket,
|
2016-10-30 17:27:29 +01:00
|
|
|
uploadsPath,
|
|
|
|
)
|
|
|
|
} else {
|
|
|
|
_ = disk.RenameFile(
|
2016-11-20 23:25:43 +01:00
|
|
|
minioMetaTmpBucket, tmpUploadsPath,
|
2016-11-22 22:15:06 +01:00
|
|
|
minioMetaMultipartBucket, uploadsPath,
|
2016-10-30 17:27:29 +01:00
|
|
|
)
|
2016-06-03 01:34:15 +02:00
|
|
|
}
|
|
|
|
}(index, disk)
|
|
|
|
}
|
|
|
|
wg.Wait()
|
2016-08-25 18:39:01 +02:00
|
|
|
return traceError(errXLWriteQuorum)
|
2016-06-03 01:34:15 +02:00
|
|
|
}
|
2016-05-21 05:48:47 +02:00
|
|
|
|
2016-10-30 17:27:29 +01:00
|
|
|
// we do have quorum, so in case of delete upload.json file
|
|
|
|
// operation, we purge from tmp.
|
2016-06-29 11:10:40 +02:00
|
|
|
for index, disk := range xl.storageDisks {
|
2016-10-30 17:27:29 +01:00
|
|
|
if disk == nil || !isDelete[index] {
|
2016-06-03 01:34:15 +02:00
|
|
|
continue
|
|
|
|
}
|
2016-05-21 05:48:47 +02:00
|
|
|
wg.Add(1)
|
|
|
|
go func(index int, disk StorageAPI) {
|
|
|
|
defer wg.Done()
|
2016-10-30 17:27:29 +01:00
|
|
|
// isDelete[index] = true at this point.
|
2016-11-20 23:25:43 +01:00
|
|
|
_ = disk.DeleteFile(minioMetaTmpBucket, tmpUploadsPath)
|
2016-05-21 05:48:47 +02:00
|
|
|
}(index, disk)
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
|
2016-11-21 10:47:26 +01:00
|
|
|
if reducedErr := reduceWriteQuorumErrs(errs, objectOpIgnoredErrs, xl.writeQuorum); reducedErr != nil {
|
|
|
|
return reducedErr
|
|
|
|
}
|
|
|
|
return nil
|
2016-05-21 05:48:47 +02:00
|
|
|
}
|
|
|
|
|
2016-11-23 00:29:39 +01:00
|
|
|
// 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)
|
|
|
|
}
|
|
|
|
|
2016-05-26 23:43:17 +02:00
|
|
|
// Returns if the prefix is a multipart upload.
|
|
|
|
func (xl xlObjects) isMultipartUpload(bucket, prefix string) bool {
|
2016-07-21 09:27:08 +02:00
|
|
|
for _, disk := range xl.getLoadBalancedDisks() {
|
2016-06-03 01:34:15 +02:00
|
|
|
if disk == nil {
|
|
|
|
continue
|
|
|
|
}
|
2016-06-02 01:43:31 +02:00
|
|
|
_, err := disk.StatFile(bucket, pathJoin(prefix, uploadsJSONFile))
|
2016-07-08 07:10:27 +02:00
|
|
|
if err == nil {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
// For any reason disk was deleted or goes offline, continue
|
|
|
|
if isErrIgnored(err, objMetadataOpIgnoredErrs) {
|
|
|
|
continue
|
2016-05-21 05:48:47 +02:00
|
|
|
}
|
2016-06-02 01:43:31 +02:00
|
|
|
break
|
2016-05-21 05:48:47 +02:00
|
|
|
}
|
2016-07-08 07:10:27 +02:00
|
|
|
return false
|
2016-06-02 01:43:31 +02:00
|
|
|
}
|
2016-05-21 05:48:47 +02:00
|
|
|
|
|
|
|
// isUploadIDExists - verify if a given uploadID exists and is valid.
|
|
|
|
func (xl xlObjects) isUploadIDExists(bucket, object, uploadID string) bool {
|
2016-11-22 22:15:06 +01:00
|
|
|
uploadIDPath := path.Join(bucket, object, uploadID)
|
|
|
|
return xl.isObject(minioMetaMultipartBucket, uploadIDPath)
|
2016-05-21 05:48:47 +02:00
|
|
|
}
|
2016-05-28 22:23:08 +02:00
|
|
|
|
|
|
|
// Removes part given by partName belonging to a mulitpart upload from minioMetaBucket
|
|
|
|
func (xl xlObjects) removeObjectPart(bucket, object, uploadID, partName string) {
|
2016-11-22 22:15:06 +01:00
|
|
|
curpartPath := path.Join(bucket, object, uploadID, partName)
|
2016-05-28 22:23:08 +02:00
|
|
|
wg := sync.WaitGroup{}
|
|
|
|
for i, disk := range xl.storageDisks {
|
2016-06-03 01:34:15 +02:00
|
|
|
if disk == nil {
|
|
|
|
continue
|
|
|
|
}
|
2016-05-28 22:23:08 +02:00
|
|
|
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.
|
2016-11-22 22:15:06 +01:00
|
|
|
_ = disk.DeleteFile(minioMetaMultipartBucket, curpartPath)
|
2016-05-28 22:23:08 +02:00
|
|
|
}(i, disk)
|
|
|
|
}
|
|
|
|
wg.Wait()
|
|
|
|
}
|
2016-06-08 03:15:04 +02:00
|
|
|
|
|
|
|
// statPart - returns fileInfo structure for a successful stat on part file.
|
|
|
|
func (xl xlObjects) statPart(bucket, object, uploadID, partName string) (fileInfo FileInfo, err error) {
|
2016-11-22 22:15:06 +01:00
|
|
|
partNamePath := path.Join(bucket, object, uploadID, partName)
|
2016-07-21 09:27:08 +02:00
|
|
|
for _, disk := range xl.getLoadBalancedDisks() {
|
2016-06-08 03:15:04 +02:00
|
|
|
if disk == nil {
|
|
|
|
continue
|
|
|
|
}
|
2016-11-22 22:15:06 +01:00
|
|
|
fileInfo, err = disk.StatFile(minioMetaMultipartBucket, partNamePath)
|
2016-07-08 07:10:27 +02:00
|
|
|
if err == nil {
|
|
|
|
return fileInfo, nil
|
|
|
|
}
|
2016-08-25 18:39:01 +02:00
|
|
|
err = traceError(err)
|
2016-07-25 03:08:15 +02:00
|
|
|
// For any reason disk was deleted or goes offline we continue to next disk.
|
2016-07-08 07:10:27 +02:00
|
|
|
if isErrIgnored(err, objMetadataOpIgnoredErrs) {
|
|
|
|
continue
|
2016-06-08 03:15:04 +02:00
|
|
|
}
|
2016-07-25 03:08:15 +02:00
|
|
|
|
|
|
|
// Catastrophic error, we return.
|
2016-06-08 03:15:04 +02:00
|
|
|
break
|
|
|
|
}
|
2016-07-08 07:10:27 +02:00
|
|
|
return FileInfo{}, err
|
2016-06-08 03:15:04 +02:00
|
|
|
}
|
2016-06-17 20:57:51 +02:00
|
|
|
|
2016-07-12 02:24:49 +02:00
|
|
|
// commitXLMetadata - commit `xl.json` from source prefix to destination prefix in the given slice of disks.
|
2016-11-21 21:45:02 +01:00
|
|
|
func commitXLMetadata(disks []StorageAPI, srcBucket, srcPrefix, dstBucket, dstPrefix string, quorum int) error {
|
2016-06-17 20:57:51 +02:00
|
|
|
var wg = &sync.WaitGroup{}
|
2016-07-12 02:24:49 +02:00
|
|
|
var mErrs = make([]error, len(disks))
|
2016-06-17 20:57:51 +02:00
|
|
|
|
|
|
|
srcJSONFile := path.Join(srcPrefix, xlMetaJSONFile)
|
|
|
|
dstJSONFile := path.Join(dstPrefix, xlMetaJSONFile)
|
|
|
|
|
|
|
|
// Rename `xl.json` to all disks in parallel.
|
2016-07-12 02:24:49 +02:00
|
|
|
for index, disk := range disks {
|
2016-06-17 20:57:51 +02:00
|
|
|
if disk == nil {
|
2016-08-25 18:39:01 +02:00
|
|
|
mErrs[index] = traceError(errDiskNotFound)
|
2016-06-17 20:57:51 +02:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
wg.Add(1)
|
|
|
|
// Rename `xl.json` in a routine.
|
|
|
|
go func(index int, disk StorageAPI) {
|
|
|
|
defer wg.Done()
|
2016-07-01 01:37:51 +02:00
|
|
|
// Delete any dangling directories.
|
2016-11-21 21:45:02 +01:00
|
|
|
defer disk.DeleteFile(srcBucket, srcPrefix)
|
2016-07-01 01:37:51 +02:00
|
|
|
|
2016-06-17 20:57:51 +02:00
|
|
|
// Renames `xl.json` from source prefix to destination prefix.
|
2016-11-21 21:45:02 +01:00
|
|
|
rErr := disk.RenameFile(srcBucket, srcJSONFile, dstBucket, dstJSONFile)
|
2016-06-17 20:57:51 +02:00
|
|
|
if rErr != nil {
|
2016-08-25 18:39:01 +02:00
|
|
|
mErrs[index] = traceError(rErr)
|
2016-06-17 20:57:51 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
mErrs[index] = nil
|
|
|
|
}(index, disk)
|
|
|
|
}
|
|
|
|
// Wait for all the routines.
|
|
|
|
wg.Wait()
|
|
|
|
|
2016-07-13 22:44:33 +02:00
|
|
|
// Do we have write Quorum?.
|
2016-07-20 04:24:32 +02:00
|
|
|
if !isDiskQuorum(mErrs, quorum) {
|
2016-07-13 22:44:33 +02:00
|
|
|
// Delete all `xl.json` successfully renamed.
|
2016-11-21 21:45:02 +01:00
|
|
|
deleteAllXLMetadata(disks, dstBucket, dstPrefix, mErrs)
|
2016-08-25 18:39:01 +02:00
|
|
|
return traceError(errXLWriteQuorum)
|
2016-06-17 20:57:51 +02:00
|
|
|
}
|
2016-07-13 09:29:48 +02:00
|
|
|
|
2016-11-21 10:47:26 +01:00
|
|
|
if reducedErr := reduceWriteQuorumErrs(mErrs, objectOpIgnoredErrs, quorum); reducedErr != nil {
|
|
|
|
return reducedErr
|
|
|
|
}
|
|
|
|
return nil
|
2016-06-17 20:57:51 +02:00
|
|
|
}
|