// Copyright (c) 2015-2021 MinIO, Inc.//// This file is part of MinIO Object Storage stack//// This program is free software: you can redistribute it and/or modify// it under the terms of the GNU Affero General Public License as published by// the Free Software Foundation, either version 3 of the License, or// (at your option) any later version.//// This program is distributed in the hope that it will be useful// but WITHOUT ANY WARRANTY; without even the implied warranty of// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the// GNU Affero General Public License for more details.//// You should have received a copy of the GNU Affero General Public License// along with this program.  If not, see <http://www.gnu.org/licenses/>. package cmd import (	"bytes"	"context"	"errors"	"fmt"	"io"	"net/http"	"path"	"strconv"	"strings"	"sync" 	"github.com/klauspost/readahead"	"github.com/minio/madmin-go"	"github.com/minio/minio-go/v7/pkg/tags"	"github.com/minio/minio/internal/bucket/lifecycle"	"github.com/minio/minio/internal/bucket/object/lock"	"github.com/minio/minio/internal/bucket/replication"	"github.com/minio/minio/internal/event"	"github.com/minio/minio/internal/hash"	xhttp "github.com/minio/minio/internal/http"	xioutil "github.com/minio/minio/internal/ioutil"	"github.com/minio/minio/internal/logger"	"github.com/minio/minio/internal/sync/errgroup"	"github.com/minio/pkg/mimedb"	uatomic "go.uber.org/atomic") // list all errors which can be ignored in object operations.var objectOpIgnoredErrs = append(baseIgnoredErrs, errDiskAccessDenied, errUnformattedDisk) // Object Operations func countOnlineDisks(onlineDisks []StorageAPI) (online int) {	for _, onlineDisk := range onlineDisks {		if onlineDisk != nil && onlineDisk.IsOnline() {			online++		}	}	return online} // CopyObject - copy object source object to destination object.// if source object and destination object are same we only// update metadata.func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (oi ObjectInfo, err error) {	auditObjectErasureSet(ctx, dstObject, &er) 	// This call shouldn't be used for anything other than metadata updates or adding self referential versions.	if !srcInfo.metadataOnly {		return oi, NotImplemented{}	} 	defer NSUpdated(dstBucket, dstObject) 	if !dstOpts.NoLock {		lk := er.NewNSLock(dstBucket, dstObject)		lkctx, err := lk.GetLock(ctx, globalOperationTimeout)		if err != nil {			return oi, err		}		ctx = lkctx.Context()		defer lk.Unlock(lkctx.Cancel)	}	// Read metadata associated with the object from all disks.	storageDisks := er.getDisks() 	var metaArr []FileInfo	var errs []error 	// Read metadata associated with the object from all disks.	if srcOpts.VersionID != "" {		metaArr, errs = readAllFileInfo(ctx, storageDisks, srcBucket, srcObject, srcOpts.VersionID, true)	} else {		metaArr, errs = readAllXL(ctx, storageDisks, srcBucket, srcObject, true)	} 	readQuorum, writeQuorum, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount)	if err != nil {		if errors.Is(err, errErasureReadQuorum) && !strings.HasPrefix(srcBucket, minioMetaBucket) {			_, derr := er.deleteIfDangling(ctx, srcBucket, srcObject, metaArr, errs, nil, srcOpts)			if derr != nil {				err = derr			}		}		return ObjectInfo{}, toObjectErr(err, srcBucket, srcObject)	} 	// List all online disks.	onlineDisks, modTime := listOnlineDisks(storageDisks, metaArr, errs) 	// Pick latest valid metadata.	fi, err := pickValidFileInfo(ctx, metaArr, modTime, readQuorum)	if err != nil {		return oi, toObjectErr(err, srcBucket, srcObject)	}	if fi.Deleted {		if srcOpts.VersionID == "" {			return oi, toObjectErr(errFileNotFound, srcBucket, srcObject)		}		return fi.ToObjectInfo(srcBucket, srcObject, srcOpts.Versioned || srcOpts.VersionSuspended), toObjectErr(errMethodNotAllowed, srcBucket, srcObject)	} 	filterOnlineDisksInplace(fi, metaArr, onlineDisks) 	versionID := srcInfo.VersionID	if srcInfo.versionOnly {		versionID = dstOpts.VersionID		// preserve destination versionId if specified.		if versionID == "" {			versionID = mustGetUUID()			fi.IsLatest = true // we are creating a new version so this is latest.		}		modTime = UTCNow()	} 	fi.VersionID = versionID // set any new versionID we might have created	fi.ModTime = modTime     // set modTime for the new versionID	if !dstOpts.MTime.IsZero() {		modTime = dstOpts.MTime		fi.ModTime = dstOpts.MTime	}	fi.Metadata = srcInfo.UserDefined	srcInfo.UserDefined["etag"] = srcInfo.ETag 	inlineData := fi.InlineData()	freeVersionID := fi.TierFreeVersionID()	freeVersionMarker := fi.TierFreeVersion() 	// Update `xl.meta` content on each disks.	for index := range metaArr {		if metaArr[index].IsValid() {			metaArr[index].ModTime = modTime			metaArr[index].VersionID = versionID			if !metaArr[index].InlineData() {				// If the data is not inlined, we may end up incorrectly				// inlining the data here, that leads to an inconsistent				// situation where some objects are were not inlined				// were now inlined, make sure to `nil` the Data such				// that xl.meta is written as expected.				metaArr[index].Data = nil			}			metaArr[index].Metadata = srcInfo.UserDefined			// Preserve existing values			if inlineData {				metaArr[index].SetInlineData()			}			if freeVersionID != "" {				metaArr[index].SetTierFreeVersionID(freeVersionID)			}			if freeVersionMarker {				metaArr[index].SetTierFreeVersion()			}		}	} 	// Write unique `xl.meta` for each disk.	if _, err = writeUniqueFileInfo(ctx, onlineDisks, srcBucket, srcObject, metaArr, writeQuorum); err != nil {		return oi, toObjectErr(err, srcBucket, srcObject)	} 	return fi.ToObjectInfo(srcBucket, srcObject, srcOpts.Versioned || srcOpts.VersionSuspended), nil} // GetObjectNInfo - returns object info and an object// Read(Closer). When err != nil, the returned reader is always nil.func (er erasureObjects) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) {	auditObjectErasureSet(ctx, object, &er) 	var unlockOnDefer bool	nsUnlocker := func() {}	defer func() {		if unlockOnDefer {			nsUnlocker()		}	}() 	// Acquire lock	if lockType != noLock {		lock := er.NewNSLock(bucket, object)		switch lockType {		case writeLock:			lkctx, err := lock.GetLock(ctx, globalOperationTimeout)			if err != nil {				return nil, err			}			ctx = lkctx.Context()			nsUnlocker = func() { lock.Unlock(lkctx.Cancel) }		case readLock:			lkctx, err := lock.GetRLock(ctx, globalOperationTimeout)			if err != nil {				return nil, err			}			ctx = lkctx.Context()			nsUnlocker = func() { lock.RUnlock(lkctx.Cancel) }		}		unlockOnDefer = true	} 	fi, metaArr, onlineDisks, err := er.getObjectFileInfo(ctx, bucket, object, opts, true)	if err != nil {		return nil, toObjectErr(err, bucket, object)	} 	if !fi.DataShardFixed() {		diskMTime := pickValidDiskTimeWithQuorum(metaArr, fi.Erasure.DataBlocks)		if !diskMTime.Equal(timeSentinel) && !diskMTime.IsZero() {			for index := range onlineDisks {				if onlineDisks[index] == OfflineDisk {					continue				}				if !metaArr[index].IsValid() {					continue				}				if !metaArr[index].AcceptableDelta(diskMTime, shardDiskTimeDelta) {					// If disk mTime mismatches it is considered outdated					// https://github.com/minio/minio/pull/13803					//					// This check only is active if we could find maximally					// occurring disk mtimes that are somewhat same across					// the quorum. Allowing to skip those shards which we					// might think are wrong.					onlineDisks[index] = OfflineDisk				}			}		}	} 	objInfo := fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended)	if objInfo.DeleteMarker {		if opts.VersionID == "" {			return &GetObjectReader{				ObjInfo: objInfo,			}, toObjectErr(errFileNotFound, bucket, object)		}		// Make sure to return object info to provide extra information.		return &GetObjectReader{			ObjInfo: objInfo,		}, toObjectErr(errMethodNotAllowed, bucket, object)	}	if objInfo.IsRemote() {		gr, err := getTransitionedObjectReader(ctx, bucket, object, rs, h, objInfo, opts)		if err != nil {			return nil, err		}		unlockOnDefer = false		return gr.WithCleanupFuncs(nsUnlocker), nil	} 	fn, off, length, err := NewGetObjectReader(rs, objInfo, opts)	if err != nil {		return nil, err	}	unlockOnDefer = false 	pr, pw := xioutil.WaitPipe()	go func() {		pw.CloseWithError(er.getObjectWithFileInfo(ctx, bucket, object, off, length, pw, fi, metaArr, onlineDisks))	}() 	// Cleanup function to cause the go routine above to exit, in	// case of incomplete read.	pipeCloser := func() {		pr.CloseWithError(nil)	} 	return fn(pr, h, pipeCloser, nsUnlocker)} func (er erasureObjects) getObjectWithFileInfo(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, fi FileInfo, metaArr []FileInfo, onlineDisks []StorageAPI) error {	// Reorder online disks based on erasure distribution order.	// Reorder parts metadata based on erasure distribution order.	onlineDisks, metaArr = shuffleDisksAndPartsMetadataByIndex(onlineDisks, metaArr, fi) 	// For negative length read everything.	if length < 0 {		length = fi.Size - startOffset	} 	// Reply back invalid range if the input offset and length fall out of range.	if startOffset > fi.Size || startOffset+length > fi.Size {		logger.LogIf(ctx, InvalidRange{startOffset, length, fi.Size}, logger.Application)		return InvalidRange{startOffset, length, fi.Size}	} 	// Get start part index and offset.	partIndex, partOffset, err := fi.ObjectToPartOffset(ctx, startOffset)	if err != nil {		return InvalidRange{startOffset, length, fi.Size}	} 	// Calculate endOffset according to length	endOffset := startOffset	if length > 0 {		endOffset += length - 1	} 	// Get last part index to read given length.	lastPartIndex, _, err := fi.ObjectToPartOffset(ctx, endOffset)	if err != nil {		return InvalidRange{startOffset, length, fi.Size}	} 	var totalBytesRead int64	erasure, err := NewErasure(ctx, fi.Erasure.DataBlocks, fi.Erasure.ParityBlocks, fi.Erasure.BlockSize)	if err != nil {		return toObjectErr(err, bucket, object)	} 	var healOnce sync.Once 	// once we have obtained a common FileInfo i.e latest, we should stick	// to single dataDir to read the content to avoid reading from some other	// dataDir that has stale FileInfo{} to ensure that we fail appropriately	// during reads and expect the same dataDir everywhere.	dataDir := fi.DataDir	for ; partIndex <= lastPartIndex; partIndex++ {		if length == totalBytesRead {			break		} 		partNumber := fi.Parts[partIndex].Number 		// Save the current part name and size.		partSize := fi.Parts[partIndex].Size 		partLength := partSize - partOffset		// partLength should be adjusted so that we don't write more data than what was requested.		if partLength > (length - totalBytesRead) {			partLength = length - totalBytesRead		} 		tillOffset := erasure.ShardFileOffset(partOffset, partLength, partSize)		// Get the checksums of the current part.		readers := make([]io.ReaderAt, len(onlineDisks))		prefer := make([]bool, len(onlineDisks))		for index, disk := range onlineDisks {			if disk == OfflineDisk {				continue			}			if !metaArr[index].IsValid() {				continue			}			checksumInfo := metaArr[index].Erasure.GetChecksumInfo(partNumber)			partPath := pathJoin(object, dataDir, fmt.Sprintf("part.%d", partNumber))			readers[index] = newBitrotReader(disk, metaArr[index].Data, bucket, partPath, tillOffset,				checksumInfo.Algorithm, checksumInfo.Hash, erasure.ShardSize()) 			// Prefer local disks			prefer[index] = disk.Hostname() == ""		} 		written, err := erasure.Decode(ctx, writer, readers, partOffset, partLength, partSize, prefer)		// Note: we should not be defer'ing the following closeBitrotReaders() call as		// we are inside a for loop i.e if we use defer, we would accumulate a lot of open files by the time		// we return from this function.		closeBitrotReaders(readers)		if err != nil {			// If we have successfully written all the content that was asked			// by the client, but we still see an error - this would mean			// that we have some parts or data blocks missing or corrupted			// - attempt a heal to successfully heal them for future calls.			if written == partLength {				var scan madmin.HealScanMode				switch {				case errors.Is(err, errFileNotFound):					scan = madmin.HealNormalScan				case errors.Is(err, errFileCorrupt):					scan = madmin.HealDeepScan				}				switch scan {				case madmin.HealNormalScan, madmin.HealDeepScan:					healOnce.Do(func() {						if _, healing := er.getOnlineDisksWithHealing(); !healing {							go healObject(bucket, object, fi.VersionID, scan)						}					})					// Healing is triggered and we have written					// successfully the content to client for					// the specific part, we should `nil` this error					// and proceed forward, instead of throwing errors.					err = nil				}			}			if err != nil {				return toObjectErr(err, bucket, object)			}		}		for i, r := range readers {			if r == nil {				onlineDisks[i] = OfflineDisk			}		}		// Track total bytes read from disk and written to the client.		totalBytesRead += partLength		// partOffset will be valid only for the first part, hence reset it to 0 for		// the remaining parts.		partOffset = 0	} // End of read all parts loop.	// Return success.	return nil} // GetObjectInfo - reads object metadata and replies back ObjectInfo.func (er erasureObjects) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (info ObjectInfo, err error) {	auditObjectErasureSet(ctx, object, &er) 	if !opts.NoLock {		// Lock the object before reading.		lk := er.NewNSLock(bucket, object)		lkctx, err := lk.GetRLock(ctx, globalOperationTimeout)		if err != nil {			return ObjectInfo{}, err		}		ctx = lkctx.Context()		defer lk.RUnlock(lkctx.Cancel)	} 	return er.getObjectInfo(ctx, bucket, object, opts)} func (er erasureObjects) deleteIfDangling(ctx context.Context, bucket, object string, metaArr []FileInfo, errs []error, dataErrs []error, opts ObjectOptions) (FileInfo, error) {	var err error	m, ok := isObjectDangling(metaArr, errs, dataErrs)	if ok {		err = errFileNotFound		if opts.VersionID != "" {			err = errFileVersionNotFound		}		defer NSUpdated(bucket, object) 		if opts.VersionID != "" {			er.deleteObjectVersion(ctx, bucket, object, 1, FileInfo{				VersionID: opts.VersionID,			}, false)		} else {			er.deleteObjectVersion(ctx, bucket, object, 1, FileInfo{				VersionID: m.VersionID,			}, false)		}	}	return m, err} func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, readData bool) ([]FileInfo, []error) {	metadataArray := make([]*xlMetaV2, len(disks))	metaFileInfos := make([]FileInfo, len(metadataArray))	metadataShallowVersions := make([][]xlMetaV2ShallowVersion, len(disks)) 	g := errgroup.WithNErrs(len(disks))	// Read `xl.meta` in parallel across disks.	for index := range disks {		index := index		g.Go(func() (err error) {			if disks[index] == nil {				return errDiskNotFound			}			rf, err := disks[index].ReadXL(ctx, bucket, object, readData)			if err != nil {				return err			} 			var xl xlMetaV2			if err = xl.LoadOrConvert(rf.Buf); err != nil {				return err			}			metadataArray[index] = &xl			metaFileInfos[index] = FileInfo{				DiskMTime: rf.DiskMTime,			}			return nil		}, index)	} 	errs := g.Wait()	for index, err := range errs {		if err == nil {			continue		}		if !IsErr(err, []error{			errFileNotFound,

Lines 1–500 of 1985. Load more