// 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 (	"context"	"fmt"	"io"	"net/http"	"strconv"	"strings"	"sync"	"sync/atomic"	"time" 	objectlock "github.com/minio/minio/internal/bucket/object/lock"	"github.com/minio/minio/internal/color"	"github.com/minio/minio/internal/config/cache"	"github.com/minio/minio/internal/disk"	"github.com/minio/minio/internal/hash"	"github.com/minio/minio/internal/logger"	"github.com/minio/minio/internal/sync/errgroup"	"github.com/minio/pkg/wildcard") const (	cacheBlkSize          = 1 << 20	cacheGCInterval       = time.Minute * 30	writeBackStatusHeader = ReservedMetadataPrefixLower + "write-back-status"	writeBackRetryHeader  = ReservedMetadataPrefixLower + "write-back-retry") type cacheCommitStatus string const (	// CommitPending - cache writeback with backend is pending.	CommitPending cacheCommitStatus = "pending" 	// CommitComplete - cache writeback completed ok.	CommitComplete cacheCommitStatus = "complete" 	// CommitFailed - cache writeback needs a retry.	CommitFailed cacheCommitStatus = "failed") const (	// CommitWriteBack allows staging and write back of cached content for single object uploads	CommitWriteBack string = "writeback"	// CommitWriteThrough allows caching multipart uploads to disk synchronously	CommitWriteThrough string = "writethrough") // String returns string representation of statusfunc (s cacheCommitStatus) String() string {	return string(s)} // CacheStorageInfo - represents total, free capacity of// underlying cache storage.type CacheStorageInfo struct {	Total uint64 // Total cache disk space.	Free  uint64 // Free cache available space.} // CacheObjectLayer implements primitives for cache object API layer.type CacheObjectLayer interface {	// Object operations.	GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error)	GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)	DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error)	DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error)	PutObject(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error)	CopyObject(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error)	// Multipart operations.	NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error)	PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error)	AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error	CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error)	CopyObjectPart(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject, uploadID string, partID int, startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (pi PartInfo, e error) 	// Storage operations.	StorageInfo(ctx context.Context) CacheStorageInfo	CacheStats() *CacheStats} // Abstracts disk caching - used by the S3 layertype cacheObjects struct {	// slice of cache drives	cache []*diskCache	// file path patterns to exclude from cache	exclude []string	// number of accesses after which to cache an object	after int	// commit objects in async manner	commitWriteback    bool	commitWritethrough bool 	// if true migration is in progress from v1 to v2	migrating bool	// retry queue for writeback cache mode to reattempt upload to backend	wbRetryCh chan ObjectInfo	// Cache stats	cacheStats *CacheStats 	InnerGetObjectNInfoFn          func(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error)	InnerGetObjectInfoFn           func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)	InnerDeleteObjectFn            func(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error)	InnerPutObjectFn               func(ctx context.Context, bucket, object string, data *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error)	InnerCopyObjectFn              func(ctx context.Context, srcBucket, srcObject, destBucket, destObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error)	InnerNewMultipartUploadFn      func(ctx context.Context, bucket, object string, opts ObjectOptions) (uploadID string, err error)	InnerPutObjectPartFn           func(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error)	InnerAbortMultipartUploadFn    func(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) error	InnerCompleteMultipartUploadFn func(ctx context.Context, bucket, object, uploadID string, uploadedParts []CompletePart, opts ObjectOptions) (objInfo ObjectInfo, err error)	InnerCopyObjectPartFn          func(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject, uploadID string, partID int, startOffset int64, length int64, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (pi PartInfo, e error)} func (c *cacheObjects) incHitsToMeta(ctx context.Context, dcache *diskCache, bucket, object string, size int64, eTag string, rs *HTTPRangeSpec) error {	metadata := map[string]string{"etag": eTag}	return dcache.SaveMetadata(ctx, bucket, object, metadata, size, rs, "", true, false)} // Backend metadata could have changed through server side copy - reset cache metadata if that is the casefunc (c *cacheObjects) updateMetadataIfChanged(ctx context.Context, dcache *diskCache, bucket, object string, bkObjectInfo, cacheObjInfo ObjectInfo, rs *HTTPRangeSpec) error {	bkMeta := make(map[string]string, len(bkObjectInfo.UserDefined))	cacheMeta := make(map[string]string, len(cacheObjInfo.UserDefined))	for k, v := range bkObjectInfo.UserDefined {		if strings.HasPrefix(strings.ToLower(k), ReservedMetadataPrefixLower) {			// Do not need to send any internal metadata			continue		}		bkMeta[http.CanonicalHeaderKey(k)] = v	}	for k, v := range cacheObjInfo.UserDefined {		if strings.HasPrefix(strings.ToLower(k), ReservedMetadataPrefixLower) {			// Do not need to send any internal metadata			continue		}		cacheMeta[http.CanonicalHeaderKey(k)] = v	} 	if !isMetadataSame(bkMeta, cacheMeta) ||		bkObjectInfo.ETag != cacheObjInfo.ETag ||		bkObjectInfo.ContentType != cacheObjInfo.ContentType ||		!bkObjectInfo.Expires.Equal(cacheObjInfo.Expires) {		return dcache.SaveMetadata(ctx, bucket, object, getMetadata(bkObjectInfo), bkObjectInfo.Size, nil, "", false, false)	}	return c.incHitsToMeta(ctx, dcache, bucket, object, cacheObjInfo.Size, cacheObjInfo.ETag, rs)} // DeleteObject clears cache entry if backend delete operation succeedsfunc (c *cacheObjects) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {	if objInfo, err = c.InnerDeleteObjectFn(ctx, bucket, object, opts); err != nil {		return	}	if c.isCacheExclude(bucket, object) || c.skipCache() {		return	} 	dcache, cerr := c.getCacheLoc(bucket, object)	if cerr != nil {		return objInfo, cerr	}	dcache.Delete(ctx, bucket, object)	return} // DeleteObjects batch deletes objects in slice, and clears any cached entriesfunc (c *cacheObjects) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {	errs := make([]error, len(objects))	objInfos := make([]ObjectInfo, len(objects))	for idx, object := range objects {		opts.VersionID = object.VersionID		objInfos[idx], errs[idx] = c.DeleteObject(ctx, bucket, object.ObjectName, opts)	}	deletedObjects := make([]DeletedObject, len(objInfos))	for idx := range errs {		if errs[idx] != nil {			continue		}		if objInfos[idx].DeleteMarker {			deletedObjects[idx] = DeletedObject{				DeleteMarker:          objInfos[idx].DeleteMarker,				DeleteMarkerVersionID: objInfos[idx].VersionID,			}			continue		}		deletedObjects[idx] = DeletedObject{			ObjectName: objInfos[idx].Name,			VersionID:  objInfos[idx].VersionID,		}	}	return deletedObjects, errs} // construct a metadata k-v mapfunc getMetadata(objInfo ObjectInfo) map[string]string {	metadata := make(map[string]string, len(objInfo.UserDefined)+4)	metadata["etag"] = objInfo.ETag	metadata["content-type"] = objInfo.ContentType	if objInfo.ContentEncoding != "" {		metadata["content-encoding"] = objInfo.ContentEncoding	}	if !objInfo.Expires.Equal(timeSentinel) {		metadata["expires"] = objInfo.Expires.Format(http.TimeFormat)	}	metadata["last-modified"] = objInfo.ModTime.Format(http.TimeFormat)	for k, v := range objInfo.UserDefined {		metadata[k] = v	}	return metadata} // marks cache hitfunc (c *cacheObjects) incCacheStats(size int64) {	c.cacheStats.incHit()	c.cacheStats.incBytesServed(size)} func (c *cacheObjects) GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (gr *GetObjectReader, err error) {	if c.isCacheExclude(bucket, object) || c.skipCache() {		return c.InnerGetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)	}	var cc *cacheControl	var cacheObjSize int64	// fetch diskCache if object is currently cached or nearest available cache drive	dcache, err := c.getCacheToLoc(ctx, bucket, object)	if err != nil {		return c.InnerGetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)	} 	cacheReader, numCacheHits, cacheErr := dcache.Get(ctx, bucket, object, rs, h, opts)	if cacheErr == nil {		cacheObjSize = cacheReader.ObjInfo.Size		if rs != nil {			if _, len, err := rs.GetOffsetLength(cacheObjSize); err == nil {				cacheObjSize = len			}		}		cc = cacheControlOpts(cacheReader.ObjInfo)		if cc != nil && (!cc.isStale(cacheReader.ObjInfo.ModTime) ||			cc.onlyIfCached) {			// This is a cache hit, mark it so			bytesServed := cacheReader.ObjInfo.Size			if rs != nil {				if _, len, err := rs.GetOffsetLength(bytesServed); err == nil {					bytesServed = len				}			}			c.cacheStats.incHit()			c.cacheStats.incBytesServed(bytesServed)			c.incHitsToMeta(ctx, dcache, bucket, object, cacheReader.ObjInfo.Size, cacheReader.ObjInfo.ETag, rs)			return cacheReader, nil		}		if cc != nil && cc.noStore {			cacheReader.Close()			c.cacheStats.incMiss()			bReader, err := c.InnerGetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)			bReader.ObjInfo.CacheLookupStatus = CacheHit			bReader.ObjInfo.CacheStatus = CacheMiss			return bReader, err		}		// serve cached content without ETag verification if writeback commit is not yet complete		if writebackInProgress(cacheReader.ObjInfo.UserDefined) {			return cacheReader, nil		}	} 	objInfo, err := c.InnerGetObjectInfoFn(ctx, bucket, object, opts)	if backendDownError(err) && cacheErr == nil {		c.incCacheStats(cacheObjSize)		return cacheReader, nil	} else if err != nil {		if cacheErr == nil {			cacheReader.Close()		}		if _, ok := err.(ObjectNotFound); ok {			if cacheErr == nil {				// Delete cached entry if backend object				// was deleted.				dcache.Delete(ctx, bucket, object)			}		}		c.cacheStats.incMiss()		return nil, err	} 	if !objInfo.IsCacheable() {		if cacheErr == nil {			cacheReader.Close()		}		c.cacheStats.incMiss()		return c.InnerGetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)	}	// skip cache for objects with locks	objRetention := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)	legalHold := objectlock.GetObjectLegalHoldMeta(objInfo.UserDefined)	if objRetention.Mode.Valid() || legalHold.Status.Valid() {		if cacheErr == nil {			cacheReader.Close()		}		c.cacheStats.incMiss()		return c.InnerGetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts)	}	if cacheErr == nil {		// if ETag matches for stale cache entry, serve from cache		if cacheReader.ObjInfo.ETag == objInfo.ETag {			// Update metadata in case server-side copy might have changed object metadata			c.updateMetadataIfChanged(ctx, dcache, bucket, object, objInfo, cacheReader.ObjInfo, rs)			c.incCacheStats(cacheObjSize)			return cacheReader, nil		}		cacheReader.Close()		// Object is stale, so delete from cache		dcache.Delete(ctx, bucket, object)	} 	// Reaching here implies cache miss	c.cacheStats.incMiss() 	bkReader, bkErr := c.InnerGetObjectNInfoFn(ctx, bucket, object, rs, h, lockType, opts) 	if bkErr != nil {		return bkReader, bkErr	}	// If object has less hits than configured cache after, just increment the hit counter	// but do not cache it.	if numCacheHits < c.after {		c.incHitsToMeta(ctx, dcache, bucket, object, objInfo.Size, objInfo.ETag, rs)		return bkReader, bkErr	} 	// Record if cache has a hit that was invalidated by ETag verification	if cacheErr == nil {		bkReader.ObjInfo.CacheLookupStatus = CacheHit	} 	// Check if we can add it without exceeding total cache size.	if !dcache.diskSpaceAvailable(objInfo.Size) {		return bkReader, bkErr	} 	if rs != nil && !dcache.enableRange {		go func() {			// if range caching is disabled, download entire object.			rs = nil			// fill cache in the background for range GET requests			bReader, bErr := c.InnerGetObjectNInfoFn(GlobalContext, bucket, object, rs, h, lockType, opts)			if bErr != nil {				return			}			defer bReader.Close()			oi, _, _, err := dcache.statRange(GlobalContext, bucket, object, rs)			// avoid cache overwrite if another background routine filled cache			if err != nil || oi.ETag != bReader.ObjInfo.ETag {				// use a new context to avoid locker prematurely timing out operation when the GetObjectNInfo returns.				dcache.Put(GlobalContext, bucket, object, bReader, bReader.ObjInfo.Size, rs, ObjectOptions{					UserDefined: getMetadata(bReader.ObjInfo),				}, false, false)				return			}		}()		return bkReader, bkErr	} 	// Initialize pipe.	pr, pw := io.Pipe()	var wg sync.WaitGroup	teeReader := io.TeeReader(bkReader, pw)	userDefined := getMetadata(bkReader.ObjInfo)	wg.Add(1)	go func() {		_, putErr := dcache.Put(ctx, bucket, object,			io.LimitReader(pr, bkReader.ObjInfo.Size),			bkReader.ObjInfo.Size, rs, ObjectOptions{				UserDefined: userDefined,			}, false, false)		// close the read end of the pipe, so the error gets		// propagated to teeReader		pr.CloseWithError(putErr)		wg.Done()	}()	cleanupBackend := func() {		pw.CloseWithError(bkReader.Close())		wg.Wait()	}	return NewGetObjectReaderFromReader(teeReader, bkReader.ObjInfo, opts, cleanupBackend)} // Returns ObjectInfo from cache if available.func (c *cacheObjects) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {	getObjectInfoFn := c.InnerGetObjectInfoFn 	if c.isCacheExclude(bucket, object) || c.skipCache() {		return getObjectInfoFn(ctx, bucket, object, opts)	} 	// fetch diskCache if object is currently cached or nearest available cache drive	dcache, err := c.getCacheToLoc(ctx, bucket, object)	if err != nil {		return getObjectInfoFn(ctx, bucket, object, opts)	}	var cc *cacheControl	// if cache control setting is valid, avoid HEAD operation to backend	cachedObjInfo, _, cerr := dcache.Stat(ctx, bucket, object)	if cerr == nil {		cc = cacheControlOpts(cachedObjInfo)		if cc == nil || (cc != nil && !cc.isStale(cachedObjInfo.ModTime)) {			// This is a cache hit, mark it so			c.cacheStats.incHit()			return cachedObjInfo, nil		}		// serve cache metadata without ETag verification if writeback commit is not yet complete		if writebackInProgress(cachedObjInfo.UserDefined) {			return cachedObjInfo, nil		}	} 	objInfo, err := getObjectInfoFn(ctx, bucket, object, opts)	if err != nil {		if _, ok := err.(ObjectNotFound); ok {			// Delete the cached entry if backend object was deleted.			dcache.Delete(ctx, bucket, object)			c.cacheStats.incMiss()			return ObjectInfo{}, err		}		if !backendDownError(err) {			c.cacheStats.incMiss()			return ObjectInfo{}, err		}		if cerr == nil {			// This is a cache hit, mark it so			c.cacheStats.incHit()			return cachedObjInfo, nil		}		c.cacheStats.incMiss()		return ObjectInfo{}, BackendDown{}	}	// Reaching here implies cache miss	c.cacheStats.incMiss()	// when backend is up, do a sanity check on cached object	if cerr != nil {		return objInfo, nil	}	if cachedObjInfo.ETag != objInfo.ETag {		// Delete the cached entry if the backend object was replaced.		dcache.Delete(ctx, bucket, object)	}	return objInfo, nil} // CopyObject reverts to backend after evicting any stale cache entriesfunc (c *cacheObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (objInfo ObjectInfo, err error) {	copyObjectFn := c.InnerCopyObjectFn	if c.isCacheExclude(srcBucket, srcObject) || c.skipCache() {		return copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)	}	if srcBucket != dstBucket || srcObject != dstObject {		return copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)	}	// fetch diskCache if object is currently cached or nearest available cache drive	dcache, err := c.getCacheToLoc(ctx, srcBucket, srcObject)	if err != nil {		return copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)	}	// if currently cached, evict old entry and revert to backend.	if cachedObjInfo, _, cerr := dcache.Stat(ctx, srcBucket, srcObject); cerr == nil {		cc := cacheControlOpts(cachedObjInfo)		if cc == nil || !cc.isStale(cachedObjInfo.ModTime) {			dcache.Delete(ctx, srcBucket, srcObject)		}	}	return copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)} // StorageInfo - returns underlying storage statistics.func (c *cacheObjects) StorageInfo(ctx context.Context) (cInfo CacheStorageInfo) {	var total, free uint64	for _, cache := range c.cache {		if cache == nil {			continue		}		info, err := getDiskInfo(cache.dir)		logger.GetReqInfo(ctx).AppendTags("cachePath", cache.dir)		logger.LogIf(ctx, err)		total += info.Total

Lines 1–500 of 1205. Load more