// 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"	"crypto/md5"	"crypto/rand"	"encoding/base64"	"encoding/hex"	"errors"	"fmt"	"io"	"io/ioutil"	"net/http"	"os"	"path"	"strconv"	"strings"	"sync"	"sync/atomic"	"time" 	"github.com/djherbis/atime"	"github.com/minio/minio/internal/config/cache"	"github.com/minio/minio/internal/crypto"	"github.com/minio/minio/internal/disk"	"github.com/minio/minio/internal/fips"	"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/kms"	"github.com/minio/minio/internal/logger"	"github.com/minio/sio") const (	// cache.json object metadata for cached objects.	cacheMetaJSONFile   = "cache.json"	cacheDataFile       = "part.1"	cacheDataFilePrefix = "part" 	cacheMetaVersion = "1.0.0"	cacheExpiryDays  = 90 * time.Hour * 24 // defaults to 90 days	// SSECacheEncrypted is the metadata key indicating that the object	// is a cache entry encrypted with cache KMS master key in globalCacheKMS.	SSECacheEncrypted = "X-Minio-Internal-Encrypted-Cache"	cacheMultipartDir = "multipart"	cacheWritebackDir = "writeback" 	cacheStaleUploadCleanupInterval = time.Hour * 24	cacheStaleUploadExpiry          = time.Hour * 24	cacheWBStaleUploadExpiry        = time.Hour * 24 * 7) // CacheChecksumInfoV1 - carries checksums of individual blocks on disk.type CacheChecksumInfoV1 struct {	Algorithm string `json:"algorithm"`	Blocksize int64  `json:"blocksize"`} // Represents the cache metadata structtype cacheMeta struct {	Version string   `json:"version"`	Stat    StatInfo `json:"stat"` // Stat of the current object `cache.json`. 	// checksums of blocks on disk.	Checksum CacheChecksumInfoV1 `json:"checksum,omitempty"`	// Metadata map for current object.	Meta map[string]string `json:"meta,omitempty"`	// Ranges maps cached range to associated filename.	Ranges map[string]string `json:"ranges,omitempty"`	// Hits is a counter on the number of times this object has been accessed so far.	Hits   int    `json:"hits,omitempty"`	Bucket string `json:"bucket,omitempty"`	Object string `json:"object,omitempty"`	// for multipart upload	PartNumbers     []int    `json:"partNums,omitempty"`   // Part Numbers	PartETags       []string `json:"partETags,omitempty"`  // Part ETags	PartSizes       []int64  `json:"partSizes,omitempty"`  // Part Sizes	PartActualSizes []int64  `json:"partASizes,omitempty"` // Part ActualSizes (compression)} // RangeInfo has the range, file and range length information for a cached range.type RangeInfo struct {	Range string	File  string	Size  int64} // Empty returns true if this is an empty structfunc (r *RangeInfo) Empty() bool {	return r.Range == "" && r.File == "" && r.Size == 0} func (m *cacheMeta) ToObjectInfo() (o ObjectInfo) {	if len(m.Meta) == 0 {		m.Meta = make(map[string]string)		m.Stat.ModTime = timeSentinel	} 	o = ObjectInfo{		Bucket:            m.Bucket,		Name:              m.Object,		CacheStatus:       CacheHit,		CacheLookupStatus: CacheHit,	}	meta := cloneMSS(m.Meta)	// We set file info only if its valid.	o.Size = m.Stat.Size	o.ETag = extractETag(meta)	o.ContentType = meta["content-type"]	o.ContentEncoding = meta["content-encoding"]	if storageClass, ok := meta[xhttp.AmzStorageClass]; ok {		o.StorageClass = storageClass	} else {		o.StorageClass = globalMinioDefaultStorageClass	}	var (		t time.Time		e error	)	if exp, ok := meta["expires"]; ok {		if t, e = time.Parse(http.TimeFormat, exp); e == nil {			o.Expires = t.UTC()		}	}	if mtime, ok := meta["last-modified"]; ok {		if t, e = time.Parse(http.TimeFormat, mtime); e == nil {			o.ModTime = t.UTC()		}	}	o.Parts = make([]ObjectPartInfo, len(m.PartNumbers))	for i := range m.PartNumbers {		o.Parts[i].Number = m.PartNumbers[i]		o.Parts[i].Size = m.PartSizes[i]		o.Parts[i].ETag = m.PartETags[i]		o.Parts[i].ActualSize = m.PartActualSizes[i]	}	// etag/md5Sum has already been extracted. We need to	// remove to avoid it from appearing as part of user-defined metadata	o.UserDefined = cleanMetadata(meta)	return o} // represents disk cache structtype diskCache struct {	// is set to 0 if drive is offline	online       uint32 // ref: https://golang.org/pkg/sync/atomic/#pkg-note-BUG	purgeRunning int32 	triggerGC          chan struct{}	dir                string         // caching directory	stats              CacheDiskStats // disk cache stats for prometheus	quotaPct           int            // max usage in %	pool               sync.Pool	after              int // minimum accesses before an object is cached.	lowWatermark       int	highWatermark      int	enableRange        bool	commitWriteback    bool	commitWritethrough bool 	retryWritebackCh chan ObjectInfo	// nsMutex namespace lock	nsMutex *nsLockMap	// Object functions pointing to the corresponding functions of backend implementation.	NewNSLockFn func(cachePath string) RWLocker} // Inits the disk cache dir if it is not initialized already.func newDiskCache(ctx context.Context, dir string, config cache.Config) (*diskCache, error) {	quotaPct := config.MaxUse	if quotaPct == 0 {		quotaPct = config.Quota	} 	if err := os.MkdirAll(dir, 0o777); err != nil {		return nil, fmt.Errorf("Unable to initialize '%s' dir, %w", dir, err)	}	cache := diskCache{		dir:                dir,		triggerGC:          make(chan struct{}, 1),		stats:              CacheDiskStats{Dir: dir},		quotaPct:           quotaPct,		after:              config.After,		lowWatermark:       config.WatermarkLow,		highWatermark:      config.WatermarkHigh,		enableRange:        config.Range,		commitWriteback:    config.CacheCommitMode == CommitWriteBack,		commitWritethrough: config.CacheCommitMode == CommitWriteThrough, 		retryWritebackCh: make(chan ObjectInfo, 10000),		online:           1,		pool: sync.Pool{			New: func() interface{} {				b := disk.AlignedBlock(int(cacheBlkSize))				return &b			},		},		nsMutex: newNSLock(false),	}	go cache.purgeWait(ctx)	go cache.cleanupStaleUploads(ctx)	if cache.commitWriteback {		go cache.scanCacheWritebackFailures(ctx)	}	cache.diskSpaceAvailable(0) // update if cache usage is already high.	cache.NewNSLockFn = func(cachePath string) RWLocker {		return cache.nsMutex.NewNSLock(nil, cachePath, "")	}	return &cache, nil} // diskUsageLow() returns true if disk usage falls below the low watermark w.r.t configured cache quota.// Ex. for a 100GB disk, if quota is configured as 70%  and watermark_low = 80% and// watermark_high = 90% then garbage collection starts when 63% of disk is used and// stops when disk usage drops to 56%func (c *diskCache) diskUsageLow() bool {	gcStopPct := c.quotaPct * c.lowWatermark / 100	di, err := disk.GetInfo(c.dir)	if err != nil {		reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir)		ctx := logger.SetReqInfo(GlobalContext, reqInfo)		logger.LogIf(ctx, err)		return false	}	usedPercent := float64(di.Used) * 100 / float64(di.Total)	low := int(usedPercent) < gcStopPct	atomic.StoreUint64(&c.stats.UsagePercent, uint64(usedPercent))	if low {		atomic.StoreInt32(&c.stats.UsageState, 0)	}	return low} // Returns if the disk usage reaches  or exceeds configured cache quota when size is added.// If current usage without size exceeds high watermark a GC is automatically queued.func (c *diskCache) diskSpaceAvailable(size int64) bool {	reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir)	ctx := logger.SetReqInfo(GlobalContext, reqInfo) 	gcTriggerPct := c.quotaPct * c.highWatermark / 100	di, err := disk.GetInfo(c.dir)	if err != nil {		logger.LogIf(ctx, err)		return false	}	if di.Total == 0 {		logger.LogIf(ctx, errors.New("diskCache: Received 0 total disk size"))		return false	}	usedPercent := float64(di.Used) * 100 / float64(di.Total)	if usedPercent >= float64(gcTriggerPct) {		atomic.StoreInt32(&c.stats.UsageState, 1)		c.queueGC()	}	atomic.StoreUint64(&c.stats.UsagePercent, uint64(usedPercent)) 	// Recalculate percentage with provided size added.	usedPercent = float64(di.Used+uint64(size)) * 100 / float64(di.Total) 	return usedPercent < float64(c.quotaPct)} // queueGC will queue a GC.// Calling this function is always non-blocking.func (c *diskCache) queueGC() {	select {	case c.triggerGC <- struct{}{}:	default:	}} // toClear returns how many bytes should be cleared to reach the low watermark quota.// returns 0 if below quota.func (c *diskCache) toClear() uint64 {	di, err := disk.GetInfo(c.dir)	if err != nil {		reqInfo := (&logger.ReqInfo{}).AppendTags("cachePath", c.dir)		ctx := logger.SetReqInfo(GlobalContext, reqInfo)		logger.LogIf(ctx, err)		return 0	}	return bytesToClear(int64(di.Total), int64(di.Free), uint64(c.quotaPct), uint64(c.lowWatermark), uint64(c.highWatermark))} func (c *diskCache) purgeWait(ctx context.Context) {	for {		select {		case <-ctx.Done():		case <-c.triggerGC: // wait here until someone triggers.			c.purge(ctx)		}	}} // Purge cache entries that were not accessed.func (c *diskCache) purge(ctx context.Context) {	if atomic.LoadInt32(&c.purgeRunning) == 1 || c.diskUsageLow() {		return	} 	toFree := c.toClear()	if toFree == 0 {		return	} 	atomic.StoreInt32(&c.purgeRunning, 1) // do not run concurrent purge()	defer atomic.StoreInt32(&c.purgeRunning, 0) 	// expiry for cleaning up old cache.json files that	// need to be cleaned up.	expiry := UTCNow().Add(-cacheExpiryDays)	// defaulting max hits count to 100	// ignore error we know what value we are passing.	scorer, err := newFileScorer(toFree, time.Now().Unix(), 100)	if err != nil {		logger.LogIf(ctx, err)		return	} 	// this function returns FileInfo for cached range files.	fiStatRangesFn := func(ranges map[string]string, pathPrefix string) map[string]os.FileInfo {		fm := make(map[string]os.FileInfo)		for _, rngFile := range ranges {			fname := pathJoin(pathPrefix, rngFile)			if fi, err := os.Stat(fname); err == nil {				fm[fname] = fi			}		}		return fm	} 	// this function returns most recent Atime among cached part files.	lastAtimeFn := func(partNums []int, pathPrefix string) time.Time {		lastATime := timeSentinel		for _, pnum := range partNums {			fname := pathJoin(pathPrefix, fmt.Sprintf("%s.%d", cacheDataFilePrefix, pnum))			if fi, err := os.Stat(fname); err == nil {				if atime.Get(fi).After(lastATime) {					lastATime = atime.Get(fi)				}			}		}		if len(partNums) == 0 {			fname := pathJoin(pathPrefix, cacheDataFile)			if fi, err := os.Stat(fname); err == nil {				lastATime = atime.Get(fi)			}		}		return lastATime	} 	filterFn := func(name string, typ os.FileMode) error {		if name == minioMetaBucket {			// Proceed to next file.			return nil		} 		cacheDir := pathJoin(c.dir, name)		meta, _, numHits, err := c.statCachedMeta(ctx, cacheDir)		if err != nil {			// delete any partially filled cache entry left behind.			removeAll(cacheDir)			// Proceed to next file.			return nil		}		// get last access time of cache part files		lastAtime := lastAtimeFn(meta.PartNumbers, pathJoin(c.dir, name))		// stat all cached file ranges.		cachedRngFiles := fiStatRangesFn(meta.Ranges, pathJoin(c.dir, name))		objInfo := meta.ToObjectInfo()		// prevent gc from clearing un-synced commits. This metadata is present when		// cache writeback commit setting is enabled.		status, ok := objInfo.UserDefined[writeBackStatusHeader]		if ok && status != CommitComplete.String() {			return nil		}		cc := cacheControlOpts(objInfo)		switch {		case cc != nil:			if cc.isStale(objInfo.ModTime) {				removeAll(cacheDir)				scorer.adjustSaveBytes(-objInfo.Size)				// break early if sufficient disk space reclaimed.				if c.diskUsageLow() {					// if we found disk usage is already low, we return nil filtering is complete.					return errDoneForNow				}			}		case lastAtime != timeSentinel:			// cached multipart or single part			objInfo.AccTime = lastAtime			objInfo.Name = pathJoin(c.dir, name, cacheDataFile)			scorer.addFileWithObjInfo(objInfo, numHits)		} 		for fname, fi := range cachedRngFiles {			if fi == nil {				continue			}			if cc != nil {				if cc.isStale(objInfo.ModTime) {					removeAll(fname)					scorer.adjustSaveBytes(-fi.Size()) 					// break early if sufficient disk space reclaimed.					if c.diskUsageLow() {						// if we found disk usage is already low, we return nil filtering is complete.						return errDoneForNow					}				}				continue			}			scorer.addFile(fname, atime.Get(fi), fi.Size(), numHits)		}		// clean up stale cache.json files for objects that never got cached but access count was maintained in cache.json		fi, err := os.Stat(pathJoin(cacheDir, cacheMetaJSONFile))		if err != nil || (fi != nil && fi.ModTime().Before(expiry) && len(cachedRngFiles) == 0) {			removeAll(cacheDir)			if fi != nil {				scorer.adjustSaveBytes(-fi.Size())			}			// Proceed to next file.			return nil		} 		// if we found disk usage is already low, we return nil filtering is complete.		if c.diskUsageLow() {			return errDoneForNow		} 		// Proceed to next file.		return nil	} 	if err := readDirFn(c.dir, filterFn); err != nil {		logger.LogIf(ctx, err)		return	} 	scorer.purgeFunc(func(qfile queuedFile) {		fileName := qfile.name		removeAll(fileName)		slashIdx := strings.LastIndex(fileName, SlashSeparator)		if slashIdx >= 0 {			fileNamePrefix := fileName[0:slashIdx]			fname := fileName[slashIdx+1:]			if fname == cacheDataFile {				removeAll(fileNamePrefix)			}		}	}) 	scorer.reset()} // sets cache drive statusfunc (c *diskCache) setOffline() {	atomic.StoreUint32(&c.online, 0)} // returns true if cache drive is onlinefunc (c *diskCache) IsOnline() bool {	return atomic.LoadUint32(&c.online) != 0} // Stat returns ObjectInfo from disk cachefunc (c *diskCache) Stat(ctx context.Context, bucket, object string) (oi ObjectInfo, numHits int, err error) {	var partial bool	var meta *cacheMeta 	cacheObjPath := getCacheSHADir(c.dir, bucket, object)	// Stat the file to get file size.	meta, partial, numHits, err = c.statCachedMeta(ctx, cacheObjPath)	if err != nil {		return	}	if partial {		return oi, numHits, errFileNotFound	}	oi = meta.ToObjectInfo()	oi.Bucket = bucket

Lines 1–500 of 1668. Load more