// 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"	"encoding/binary"	"encoding/json"	"errors"	"fmt"	"io/fs"	"math"	"math/rand"	"os"	"path"	"strings"	"sync"	"time" 	"github.com/bits-and-blooms/bloom/v3"	"github.com/dustin/go-humanize"	"github.com/minio/madmin-go"	"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/color"	"github.com/minio/minio/internal/config/heal"	"github.com/minio/minio/internal/event"	"github.com/minio/minio/internal/logger"	"github.com/minio/pkg/console"	uatomic "go.uber.org/atomic") const (	dataScannerSleepPerFolder        = time.Millisecond                 // Time to wait between folders.	dataUsageUpdateDirCycles         = 16                               // Visit all folders every n cycles.	dataScannerCompactLeastObject    = 500                              // Compact when there is less than this many objects in a branch.	dataScannerCompactAtChildren     = 10000                            // Compact when there are this many children in a branch.	dataScannerCompactAtFolders      = dataScannerCompactAtChildren / 4 // Compact when this many subfolders in a single folder.	dataScannerForceCompactAtFolders = 1_000_000                        // Compact when this many subfolders in a single folder (even top level).	dataScannerStartDelay            = 1 * time.Minute                  // Time to wait on startup and between cycles. 	healDeleteDangling    = true	healFolderIncludeProb = 32  // Include a clean folder one in n cycles.	healObjectSelectProb  = 512 // Overall probability of a file being scanned; one in n.) var (	globalHealConfig heal.Config 	dataScannerLeaderLockTimeout = newDynamicTimeout(30*time.Second, 10*time.Second)	// Sleeper values are updated when config is loaded.	scannerSleeper = newDynamicSleeper(10, 10*time.Second, true)	scannerCycle   = uatomic.NewDuration(dataScannerStartDelay)) // initDataScanner will start the scanner in the background.func initDataScanner(ctx context.Context, objAPI ObjectLayer) {	go func() {		r := rand.New(rand.NewSource(time.Now().UnixNano()))		// Run the data scanner in a loop		for {			runDataScanner(ctx, objAPI)			duration := time.Duration(r.Float64() * float64(scannerCycle.Load()))			if duration < time.Second {				// Make sure to sleep atleast a second to avoid high CPU ticks.				duration = time.Second			}			time.Sleep(duration)		}	}()} func getCycleScanMode(currentCycle, bitrotStartCycle uint64, bitrotStartTime time.Time) madmin.HealScanMode {	bitrotCycle := globalHealConfig.BitrotScanCycle()	switch bitrotCycle {	case -1:		return madmin.HealNormalScan	case 0:		return madmin.HealDeepScan	} 	if currentCycle-bitrotStartCycle < healObjectSelectProb {		return madmin.HealDeepScan	} 	if time.Since(bitrotStartTime) > bitrotCycle {		return madmin.HealDeepScan	} 	return madmin.HealNormalScan} type backgroundHealInfo struct {	BitrotStartTime  time.Time           `json:"bitrotStartTime"`	BitrotStartCycle uint64              `json:"bitrotStartCycle"`	CurrentScanMode  madmin.HealScanMode `json:"currentScanMode"`} func readBackgroundHealInfo(ctx context.Context, objAPI ObjectLayer) backgroundHealInfo {	// Get last healing information	buf, err := readConfig(ctx, objAPI, backgroundHealInfoPath)	if err != nil {		if !errors.Is(err, errConfigNotFound) {			logger.LogIf(ctx, err)		}		return backgroundHealInfo{}	}	var info backgroundHealInfo	err = json.Unmarshal(buf, &info)	if err != nil {		logger.LogIf(ctx, err)		return backgroundHealInfo{}	}	return info} func saveBackgroundHealInfo(ctx context.Context, objAPI ObjectLayer, info backgroundHealInfo) {	b, err := json.Marshal(info)	if err != nil {		logger.LogIf(ctx, err)		return	}	// Get last healing information	err = saveConfig(ctx, objAPI, backgroundHealInfoPath, b)	if err != nil {		logger.LogIf(ctx, err)	}} // runDataScanner will start a data scanner.// The function will block until the context is canceled.// There should only ever be one scanner running per cluster.func runDataScanner(pctx context.Context, objAPI ObjectLayer) {	// Make sure only 1 scanner is running on the cluster.	locker := objAPI.NewNSLock(minioMetaBucket, "scanner/runDataScanner.lock")	lkctx, err := locker.GetLock(pctx, dataScannerLeaderLockTimeout)	if err != nil {		if intDataUpdateTracker.debug {			logger.LogIf(pctx, err)		}		return	}	ctx := lkctx.Context()	defer lkctx.Cancel()	// No unlock for "leader" lock. 	// Load current bloom cycle	var cycleInfo currentScannerCycle	cycleInfo.next = intDataUpdateTracker.current() + 1 	buf, _ := readConfig(ctx, objAPI, dataUsageBloomNamePath)	if len(buf) == 8 {		cycleInfo.next = binary.LittleEndian.Uint64(buf)	} else if len(buf) > 8 {		cycleInfo.next = binary.LittleEndian.Uint64(buf[:8])		buf = buf[8:]		_, err = cycleInfo.UnmarshalMsg(buf)		logger.LogIf(ctx, err)	} 	scannerTimer := time.NewTimer(scannerCycle.Load())	defer scannerTimer.Stop()	defer globalScannerMetrics.setCycle(nil) 	for {		select {		case <-ctx.Done():			return		case <-scannerTimer.C:			// Reset the timer for next cycle.			// If scanner takes longer we start at once.			scannerTimer.Reset(scannerCycle.Load()) 			stopFn := globalScannerMetrics.log(scannerMetricScanCycle)			cycleInfo.current = cycleInfo.next			cycleInfo.started = time.Now()			globalScannerMetrics.setCycle(&cycleInfo) 			bgHealInfo := readBackgroundHealInfo(ctx, objAPI)			scanMode := getCycleScanMode(cycleInfo.current, bgHealInfo.BitrotStartCycle, bgHealInfo.BitrotStartTime)			if bgHealInfo.CurrentScanMode != scanMode {				newHealInfo := bgHealInfo				newHealInfo.CurrentScanMode = scanMode				if scanMode == madmin.HealDeepScan {					newHealInfo.BitrotStartTime = time.Now().UTC()					newHealInfo.BitrotStartCycle = cycleInfo.current				}				saveBackgroundHealInfo(ctx, objAPI, newHealInfo)			} 			// Wait before starting next cycle and wait on startup.			results := make(chan DataUsageInfo, 1)			go storeDataUsageInBackend(ctx, objAPI, results)			bf, err := globalNotificationSys.updateBloomFilter(ctx, cycleInfo.current)			logger.LogIf(ctx, err)			err = objAPI.NSScanner(ctx, bf, results, uint32(cycleInfo.current), scanMode)			logger.LogIf(ctx, err)			stopFn()			if err == nil {				// Store new cycle...				cycleInfo.next++				cycleInfo.current = 0				cycleInfo.cycleCompleted = append(cycleInfo.cycleCompleted, time.Now())				if len(cycleInfo.cycleCompleted) > dataUsageUpdateDirCycles {					cycleInfo.cycleCompleted = cycleInfo.cycleCompleted[len(cycleInfo.cycleCompleted)-dataUsageUpdateDirCycles:]				}				globalScannerMetrics.setCycle(&cycleInfo)				tmp := make([]byte, 8, 8+cycleInfo.Msgsize())				// Cycle for backward compat.				binary.LittleEndian.PutUint64(tmp, cycleInfo.next)				tmp, _ = cycleInfo.MarshalMsg(tmp)				err = saveConfig(ctx, objAPI, dataUsageBloomNamePath, tmp)				logger.LogIf(ctx, err)			}		}	}} type cachedFolder struct {	name              string	parent            *dataUsageHash	objectHealProbDiv uint32} type folderScanner struct {	root        string	getSize     getSizeFn	oldCache    dataUsageCache	newCache    dataUsageCache	updateCache dataUsageCache	withFilter  *bloomFilter 	dataUsageScannerDebug bool	healFolderInclude     uint32 // Include a clean folder one in n cycles.	healObjectSelect      uint32 // Do a heal check on an object once every n cycles. Must divide into healFolderInclude	scanMode              madmin.HealScanMode 	disks       []StorageAPI	disksQuorum int 	// If set updates will be sent regularly to this channel.	// Will not be closed when returned.	updates    chan<- dataUsageEntry	lastUpdate time.Time 	// updateCurrentPath should be called whenever a new path is scanned.	updateCurrentPath func(string)} // Cache structure and compaction://// A cache structure will be kept with a tree of usages.// The cache is a tree structure where each keeps track of its children.//// An uncompacted branch contains a count of the files only directly at the// branch level, and contains link to children branches or leaves.//// The leaves are "compacted" based on a number of properties.// A compacted leaf contains the totals of all files beneath it.//// A leaf is only scanned once every dataUsageUpdateDirCycles,// rarer if the bloom filter for the path is clean and no lifecycles are applied.// Skipped leaves have their totals transferred from the previous cycle.//// A clean leaf will be included once every healFolderIncludeProb for partial heal scans.// When selected there is a one in healObjectSelectProb that any object will be chosen for heal scan.//// Compaction happens when either://// 1) The folder (and subfolders) contains less than dataScannerCompactLeastObject objects.// 2) The folder itself contains more than dataScannerCompactAtFolders folders.// 3) The folder only contains objects and no subfolders.//// A bucket root will never be compacted.//// Furthermore if a has more than dataScannerCompactAtChildren recursive children (uncompacted folders)// the tree will be recursively scanned and the branches with the least number of objects will be// compacted until the limit is reached.//// This ensures that any branch will never contain an unreasonable amount of other branches,// and also that small branches with few objects don't take up unreasonable amounts of space.// This keeps the cache size at a reasonable size for all buckets.//// Whenever a branch is scanned, it is assumed that it will be un-compacted// before it hits any of the above limits.// This will make the branch rebalance itself when scanned if the distribution of objects has changed. // scanDataFolder will scanner the basepath+cache.Info.Name and return an updated cache.// The returned cache will always be valid, but may not be updated from the existing.// Before each operation sleepDuration is called which can be used to temporarily halt the scanner.// If the supplied context is canceled the function will return at the first chance.func scanDataFolder(ctx context.Context, poolIdx, setIdx int, basePath string, cache dataUsageCache, getSize getSizeFn, scanMode madmin.HealScanMode) (dataUsageCache, error) {	logPrefix := color.Green("data-usage: ") 	switch cache.Info.Name {	case "", dataUsageRoot:		return cache, errors.New("internal error: root scan attempted")	}	updatePath, closeDisk := globalScannerMetrics.currentPathUpdater(basePath, cache.Info.Name)	defer closeDisk() 	s := folderScanner{		root:                  basePath,		getSize:               getSize,		oldCache:              cache,		newCache:              dataUsageCache{Info: cache.Info},		updateCache:           dataUsageCache{Info: cache.Info},		dataUsageScannerDebug: intDataUpdateTracker.debug,		healFolderInclude:     0,		healObjectSelect:      0,		scanMode:              scanMode,		updates:               cache.Info.updates,		updateCurrentPath:     updatePath,	} 	// Add disks for set healing.	if poolIdx >= 0 && setIdx >= 0 {		objAPI, ok := newObjectLayerFn().(*erasureServerPools)		if ok {			if poolIdx < len(objAPI.serverPools) && setIdx < len(objAPI.serverPools[poolIdx].sets) {				// Pass the disks belonging to the set.				s.disks = objAPI.serverPools[poolIdx].sets[setIdx].getDisks()				s.disksQuorum = len(s.disks) / 2			} else {				logger.LogIf(ctx, fmt.Errorf("Matching pool %s, set %s not found", humanize.Ordinal(poolIdx+1), humanize.Ordinal(setIdx+1)))			}		}	} 	// Enable healing in XL mode.	if globalIsErasure && !cache.Info.SkipHealing {		// Include a clean folder one in n cycles.		s.healFolderInclude = healFolderIncludeProb		// Do a heal check on an object once every n cycles. Must divide into healFolderInclude		s.healObjectSelect = healObjectSelectProb	}	if len(cache.Info.BloomFilter) > 0 {		s.withFilter = &bloomFilter{BloomFilter: &bloom.BloomFilter{}}		_, err := s.withFilter.ReadFrom(bytes.NewReader(cache.Info.BloomFilter))		if err != nil {			logger.LogIf(ctx, err, logPrefix+"Error reading bloom filter")			s.withFilter = nil		}	} 	done := ctx.Done() 	// Read top level in bucket.	select {	case <-done:		return cache, ctx.Err()	default:	}	root := dataUsageEntry{}	folder := cachedFolder{name: cache.Info.Name, objectHealProbDiv: 1}	err := s.scanFolder(ctx, folder, &root)	if err != nil {		// No useful information...		return cache, err	} 	s.newCache.Info.LastUpdate = UTCNow()	s.newCache.Info.NextCycle = cache.Info.NextCycle	return s.newCache, nil} // sendUpdate() should be called on a regular basis when the newCache contains more recent total than previously.// May or may not send an update upstream.func (f *folderScanner) sendUpdate() {	// Send at most an update every minute.	if f.updates == nil || time.Since(f.lastUpdate) < time.Minute {		return	}	if flat := f.updateCache.sizeRecursive(f.newCache.Info.Name); flat != nil {		select {		case f.updates <- *flat:		default:		}		f.lastUpdate = time.Now()	}} // scanFolder will scan the provided folder.// Files found in the folders will be added to f.newCache.// If final is provided folders will be put into f.newFolders or f.existingFolders.// If final is not provided the folders found are returned from the function.func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, into *dataUsageEntry) error {	done := ctx.Done()	scannerLogPrefix := color.Green("folder-scanner:") 	thisHash := hashPath(folder.name)	// Store initial compaction state.	wasCompacted := into.Compacted 	for {		select {		case <-done:			return ctx.Err()		default:		}		existing, ok := f.oldCache.Cache[thisHash.Key()]		var abandonedChildren dataUsageHashMap		if !into.Compacted {			abandonedChildren = f.oldCache.findChildrenCopy(thisHash)		} 		// If there are lifecycle rules for the prefix, remove the filter.		filter := f.withFilter		_, prefix := path2BucketObjectWithBasePath(f.root, folder.name)		var activeLifeCycle *lifecycle.Lifecycle		if f.oldCache.Info.lifeCycle != nil && f.oldCache.Info.lifeCycle.HasActiveRules(prefix, true) {			if f.dataUsageScannerDebug {				console.Debugf(scannerLogPrefix+" Prefix %q has active rules\n", prefix)			}			activeLifeCycle = f.oldCache.Info.lifeCycle			filter = nil		}		// If there are replication rules for the prefix, remove the filter.		var replicationCfg replicationConfig		if !f.oldCache.Info.replication.Empty() && f.oldCache.Info.replication.Config.HasActiveRules(prefix, true) {			replicationCfg = f.oldCache.Info.replication			filter = nil		}		// Check if we can skip it due to bloom filter...		if filter != nil && ok && existing.Compacted {			// If folder isn't in filter and we have data, skip it completely.			if folder.name != dataUsageRoot && !filter.containsDir(folder.name) {				if f.healObjectSelect == 0 || !thisHash.modAlt(f.oldCache.Info.NextCycle/folder.objectHealProbDiv, f.healFolderInclude/folder.objectHealProbDiv) {					f.newCache.copyWithChildren(&f.oldCache, thisHash, folder.parent)					f.updateCache.copyWithChildren(&f.oldCache, thisHash, folder.parent)					if f.dataUsageScannerDebug {						console.Debugf(scannerLogPrefix+" Skipping non-updated folder: %v\n", folder.name)					}					return nil				}				if f.dataUsageScannerDebug {					console.Debugf(scannerLogPrefix+" Adding non-updated folder to heal check: %v\n", folder.name)				}				// If probability was already scannerHealFolderInclude, keep it.				folder.objectHealProbDiv = f.healFolderInclude			}		}		scannerSleeper.Sleep(ctx, dataScannerSleepPerFolder) 		var existingFolders, newFolders []cachedFolder		var foundObjects bool		err := readDirFn(path.Join(f.root, folder.name), func(entName string, typ os.FileMode) error {			// Parse			entName = pathClean(path.Join(folder.name, entName))			if entName == "" || entName == folder.name {				if f.dataUsageScannerDebug {					console.Debugf(scannerLogPrefix+" no entity (%s,%s)\n", f.root, entName)				}				return nil			}			bucket, prefix := path2BucketObjectWithBasePath(f.root, entName)			if bucket == "" {				if f.dataUsageScannerDebug {					console.Debugf(scannerLogPrefix+" no bucket (%s,%s)\n", f.root, entName)				}				return errDoneForNow			} 			if isReservedOrInvalidBucket(bucket, false) {				if f.dataUsageScannerDebug {					console.Debugf(scannerLogPrefix+" invalid bucket: %v, entry: %v\n", bucket, entName)				}				return errDoneForNow			} 			select {			case <-done:				return errDoneForNow			default:			} 			if typ&os.ModeDir != 0 {				h := hashPath(entName)				_, exists := f.oldCache.Cache[h.Key()]				if h == thisHash {					return nil				}				this := cachedFolder{name: entName, parent: &thisHash, objectHealProbDiv: folder.objectHealProbDiv}

Lines 1–500 of 1498. Load more