fix: alow changes to readAllData to decrement activeCount()

There was an io.LimitReader was missing for the 'length'
parameter for ranged requests, that would cause client to
get truncated responses and errors.

fixes #11651

Dockerfile.release

@@ -5,8 +5,8 @@ ARG TARGETARCH
 LABEL name="MinIO" \
       vendor="MinIO Inc <dev@min.io>" \
       maintainer="MinIO Inc <dev@min.io>" \
-      version="RELEASE.2021-02-23T20-05-01Z" \
-      release="RELEASE.2021-02-23T20-05-01Z" \
+      version="RELEASE.2021-02-24T18-44-45Z" \
+      release="RELEASE.2021-02-24T18-44-45Z" \
       summary="MinIO is a High Performance Object Storage, API compatible with Amazon S3 cloud storage service." \
       description="MinIO object storage is fundamentally different. Designed for performance and the S3 API, it is 100% open-source. MinIO is ideal for large, private cloud environments with stringent security requirements and delivers mission-critical availability across a diverse range of workloads."
 

cmd/admin-handlers-users.go

@@ -68,7 +68,7 @@ func (a adminAPIHandlers) RemoveUser(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
 	accessKey := vars["accessKey"]
 
-	ok, err := globalIAMSys.IsTempUser(accessKey)
+	ok, _, err := globalIAMSys.IsTempUser(accessKey)
 	if err != nil {
 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 		return
@@ -485,7 +485,7 @@ func (a adminAPIHandlers) AddServiceAccount(w http.ResponseWriter, r *http.Reque
 		parentUser = cred.ParentUser
 	}
 
-	newCred, err := globalIAMSys.NewServiceAccount(ctx, parentUser, createReq.Policy)
+	newCred, err := globalIAMSys.NewServiceAccount(ctx, parentUser, cred.Groups, createReq.Policy)
 	if err != nil {
 		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 		return
@@ -982,7 +982,7 @@ func (a adminAPIHandlers) SetPolicyForUserOrGroup(w http.ResponseWriter, r *http
 	isGroup := vars["isGroup"] == "true"
 
 	if !isGroup {
-		ok, err := globalIAMSys.IsTempUser(entityName)
+		ok, _, err := globalIAMSys.IsTempUser(entityName)
 		if err != nil && err != errNoSuchUser {
 			writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 			return

cmd/benchmark-utils_test.go

@@ -19,7 +19,6 @@ package cmd
 import (
 	"bytes"
 	"context"
-	"io/ioutil"
 	"math"
 	"math/rand"
 	"strconv"
@@ -175,56 +174,6 @@ func benchmarkPutObjectParallel(b *testing.B, instanceType string, objSize int)
 	runPutObjectBenchmarkParallel(b, objLayer, objSize)
 }
 
-// Benchmark utility functions for ObjectLayer.GetObject().
-// Creates Object layer setup ( MakeBucket, PutObject) and then runs the benchmark.
-func runGetObjectBenchmark(b *testing.B, obj ObjectLayer, objSize int) {
-	// obtains random bucket name.
-	bucket := getRandomBucketName()
-	// create bucket.
-	err := obj.MakeBucketWithLocation(context.Background(), bucket, BucketOptions{})
-	if err != nil {
-		b.Fatal(err)
-	}
-
-	textData := generateBytesData(objSize)
-
-	// generate etag for the generated data.
-	// etag of the data to written is required as input for PutObject.
-	// PutObject is the functions which writes the data onto the FS/Erasure backend.
-
-	// get text data generated for number of bytes equal to object size.
-	md5hex := getMD5Hash(textData)
-	sha256hex := ""
-
-	for i := 0; i < 10; i++ {
-		// insert the object.
-		var objInfo ObjectInfo
-		objInfo, err = obj.PutObject(context.Background(), bucket, "object"+strconv.Itoa(i),
-			mustGetPutObjReader(b, bytes.NewReader(textData), int64(len(textData)), md5hex, sha256hex), ObjectOptions{})
-		if err != nil {
-			b.Fatal(err)
-		}
-		if objInfo.ETag != md5hex {
-			b.Fatalf("Write no: %d: Md5Sum mismatch during object write into the bucket: Expected %s, got %s", i+1, objInfo.ETag, md5hex)
-		}
-	}
-
-	// benchmark utility which helps obtain number of allocations and bytes allocated per ops.
-	b.ReportAllocs()
-	// the actual benchmark for GetObject starts here. Reset the benchmark timer.
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		var buffer = new(bytes.Buffer)
-		err = obj.GetObject(context.Background(), bucket, "object"+strconv.Itoa(i%10), 0, int64(objSize), buffer, "", ObjectOptions{})
-		if err != nil {
-			b.Error(err)
-		}
-	}
-	// Benchmark ends here. Stop timer.
-	b.StopTimer()
-
-}
-
 // randomly picks a character and returns its equivalent byte array.
 func getRandomByte() []byte {
 	const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
@@ -240,38 +189,6 @@ func generateBytesData(size int) []byte {
 	return bytes.Repeat(getRandomByte(), size)
 }
 
-// creates Erasure/FS backend setup, obtains the object layer and calls the runGetObjectBenchmark function.
-func benchmarkGetObject(b *testing.B, instanceType string, objSize int) {
-	// create a temp Erasure/FS backend.
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	objLayer, disks, err := prepareTestBackend(ctx, instanceType)
-	if err != nil {
-		b.Fatalf("Failed obtaining Temp Backend: <ERROR> %s", err)
-	}
-	// cleaning up the backend by removing all the directories and files created.
-	defer removeRoots(disks)
-
-	//  uses *testing.B and the object Layer to run the benchmark.
-	runGetObjectBenchmark(b, objLayer, objSize)
-}
-
-// creates Erasure/FS backend setup, obtains the object layer and runs parallel benchmark for ObjectLayer.GetObject() .
-func benchmarkGetObjectParallel(b *testing.B, instanceType string, objSize int) {
-	// create a temp Erasure/FS backend.
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	objLayer, disks, err := prepareTestBackend(ctx, instanceType)
-	if err != nil {
-		b.Fatalf("Failed obtaining Temp Backend: <ERROR> %s", err)
-	}
-	// cleaning up the backend by removing all the directories and files created.
-	defer removeRoots(disks)
-
-	//  uses *testing.B and the object Layer to run the benchmark.
-	runGetObjectBenchmarkParallel(b, objLayer, objSize)
-}
-
 // Parallel benchmark utility functions for ObjectLayer.PutObject().
 // Creates Object layer setup ( MakeBucket ) and then runs the PutObject benchmark.
 func runPutObjectBenchmarkParallel(b *testing.B, obj ObjectLayer, objSize int) {
@@ -315,58 +232,3 @@ func runPutObjectBenchmarkParallel(b *testing.B, obj ObjectLayer, objSize int) {
 	// Benchmark ends here. Stop timer.
 	b.StopTimer()
 }
-
-// Parallel benchmark utility functions for ObjectLayer.GetObject().
-// Creates Object layer setup ( MakeBucket, PutObject) and then runs the benchmark.
-func runGetObjectBenchmarkParallel(b *testing.B, obj ObjectLayer, objSize int) {
-	// obtains random bucket name.
-	bucket := getRandomBucketName()
-	// create bucket.
-	err := obj.MakeBucketWithLocation(context.Background(), bucket, BucketOptions{})
-	if err != nil {
-		b.Fatal(err)
-	}
-
-	// get text data generated for number of bytes equal to object size.
-	textData := generateBytesData(objSize)
-	// generate md5sum for the generated data.
-	// md5sum of the data to written is required as input for PutObject.
-	// PutObject is the functions which writes the data onto the FS/Erasure backend.
-
-	md5hex := getMD5Hash([]byte(textData))
-	sha256hex := ""
-
-	for i := 0; i < 10; i++ {
-		// insert the object.
-		var objInfo ObjectInfo
-		objInfo, err = obj.PutObject(context.Background(), bucket, "object"+strconv.Itoa(i),
-			mustGetPutObjReader(b, bytes.NewReader(textData), int64(len(textData)), md5hex, sha256hex), ObjectOptions{})
-		if err != nil {
-			b.Fatal(err)
-		}
-		if objInfo.ETag != md5hex {
-			b.Fatalf("Write no: %d: Md5Sum mismatch during object write into the bucket: Expected %s, got %s", i+1, objInfo.ETag, md5hex)
-		}
-	}
-
-	// benchmark utility which helps obtain number of allocations and bytes allocated per ops.
-	b.ReportAllocs()
-	// the actual benchmark for GetObject starts here. Reset the benchmark timer.
-	b.ResetTimer()
-	b.RunParallel(func(pb *testing.PB) {
-		i := 0
-		for pb.Next() {
-			err = obj.GetObject(context.Background(), bucket, "object"+strconv.Itoa(i), 0, int64(objSize), ioutil.Discard, "", ObjectOptions{})
-			if err != nil {
-				b.Error(err)
-			}
-			i++
-			if i == 10 {
-				i = 0
-			}
-		}
-	})
-	// Benchmark ends here. Stop timer.
-	b.StopTimer()
-
-}

cmd/config-current.go

@@ -118,7 +118,7 @@ func initHelp() {
 		},
 		config.HelpKV{
 			Key:         config.ScannerSubSys,
-			Description: "manage scanner for usage calculation, lifecycle, healing and more",
+			Description: "manage namespace scanning for usage calculation, lifecycle, healing and more",
 		},
 		config.HelpKV{
 			Key:             config.LoggerWebhookSubSys,

cmd/config/heal/heal.go

@@ -66,7 +66,7 @@ var (
 	Help = config.HelpKVS{
 		config.HelpKV{
 			Key:         Bitrot,
-			Description: `perform bitrot scan on disks when checking objects during crawl`,
+			Description: `perform bitrot scan on disks when checking objects during scanner`,
 			Optional:    true,
 			Type:        "on|off",
 		},

cmd/config/identity/ldap/config.go

@@ -180,6 +180,9 @@ func getGroups(conn *ldap.Conn, sreq *ldap.SearchRequest) ([]string, error) {
 }
 
 func (l *Config) lookupBind(conn *ldap.Conn) error {
+	if l.LookupBindPassword == "" {
+		return conn.UnauthenticatedBind(l.LookupBindDN)
+	}
 	return conn.Bind(l.LookupBindDN, l.LookupBindPassword)
 }
 
@@ -422,7 +425,7 @@ func Lookup(kvs config.KVS, rootCAs *x509.CertPool) (l Config, err error) {
 	// Lookup bind user configuration
 	lookupBindDN := env.Get(EnvLookupBindDN, kvs.Get(LookupBindDN))
 	lookupBindPassword := env.Get(EnvLookupBindPassword, kvs.Get(LookupBindPassword))
-	if lookupBindDN != "" && lookupBindPassword != "" {
+	if lookupBindDN != "" {
 		l.LookupBindDN = lookupBindDN
 		l.LookupBindPassword = lookupBindPassword
 		l.isUsingLookupBind = true

cmd/data-scanner.go

@@ -43,9 +43,9 @@ import (
 )
 
 const (
-	dataCrawlSleepPerFolder  = time.Millisecond // Time to wait between folders.
-	dataCrawlStartDelay      = 1 * time.Minute  // Time to wait on startup and between cycles.
-	dataUsageUpdateDirCycles = 16               // Visit all folders every n cycles.
+	dataScannerSleepPerFolder = time.Millisecond // Time to wait between folders.
+	dataScannerStartDelay     = 1 * time.Minute  // Time to wait on startup and between cycles.
+	dataUsageUpdateDirCycles  = 16               // Visit all folders every n cycles.
 
 	healDeleteDangling    = true
 	healFolderIncludeProb = 32  // Include a clean folder one in n cycles.
@@ -76,7 +76,7 @@ func runDataScanner(ctx context.Context, objAPI ObjectLayer) {
 	for {
 		err := locker.GetLock(ctx, dataScannerLeaderLockTimeout)
 		if err != nil {
-			time.Sleep(time.Duration(r.Float64() * float64(dataCrawlStartDelay)))
+			time.Sleep(time.Duration(r.Float64() * float64(dataScannerStartDelay)))
 			continue
 		}
 		break
@@ -100,16 +100,16 @@ func runDataScanner(ctx context.Context, objAPI ObjectLayer) {
 		br.Close()
 	}
 
-	crawlTimer := time.NewTimer(dataCrawlStartDelay)
-	defer crawlTimer.Stop()
+	scannerTimer := time.NewTimer(dataScannerStartDelay)
+	defer scannerTimer.Stop()
 
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-crawlTimer.C:
+		case <-scannerTimer.C:
 			// Reset the timer for next cycle.
-			crawlTimer.Reset(dataCrawlStartDelay)
+			scannerTimer.Reset(dataScannerStartDelay)
 
 			if intDataUpdateTracker.debug {
 				console.Debugln("starting scanner cycle")
@@ -120,7 +120,7 @@ func runDataScanner(ctx context.Context, objAPI ObjectLayer) {
 			go storeDataUsageInBackend(ctx, objAPI, results)
 			bf, err := globalNotificationSys.updateBloomFilter(ctx, nextBloomCycle)
 			logger.LogIf(ctx, err)
-			err = objAPI.CrawlAndGetDataUsage(ctx, bf, results)
+			err = objAPI.NSScanner(ctx, bf, results)
 			close(results)
 			logger.LogIf(ctx, err)
 			if err == nil {
@@ -156,27 +156,27 @@ type folderScanner struct {
 	newCache   dataUsageCache
 	withFilter *bloomFilter
 
-	dataUsageCrawlDebug 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
+	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
 
 	newFolders      []cachedFolder
 	existingFolders []cachedFolder
 	disks           []StorageAPI
 }
 
-// crawlDataFolder will crawl the basepath+cache.Info.Name and return an updated cache.
+// 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 crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache, getSize getSizeFn) (dataUsageCache, error) {
+func scanDataFolder(ctx context.Context, basePath string, cache dataUsageCache, getSize getSizeFn) (dataUsageCache, error) {
 	t := UTCNow()
 
 	logPrefix := color.Green("data-usage: ")
 	logSuffix := color.Blue("- %v + %v", basePath, cache.Info.Name)
 	if intDataUpdateTracker.debug {
 		defer func() {
-			console.Debugf(logPrefix+" Crawl time: %v %s\n", time.Since(t), logSuffix)
+			console.Debugf(logPrefix+" Scanner time: %v %s\n", time.Since(t), logSuffix)
 		}()
 
 	}
@@ -189,15 +189,15 @@ func crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache,
 	skipHeal := cache.Info.SkipHealing
 
 	s := folderScanner{
-		root:                basePath,
-		getSize:             getSize,
-		oldCache:            cache,
-		newCache:            dataUsageCache{Info: cache.Info},
-		newFolders:          nil,
-		existingFolders:     nil,
-		dataUsageCrawlDebug: intDataUpdateTracker.debug,
-		healFolderInclude:   0,
-		healObjectSelect:    0,
+		root:                  basePath,
+		getSize:               getSize,
+		oldCache:              cache,
+		newCache:              dataUsageCache{Info: cache.Info},
+		newFolders:            nil,
+		existingFolders:       nil,
+		dataUsageScannerDebug: intDataUpdateTracker.debug,
+		healFolderInclude:     0,
+		healObjectSelect:      0,
 	}
 
 	// Add disks for set healing.
@@ -227,21 +227,21 @@ func crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache,
 			s.withFilter = nil
 		}
 	}
-	if s.dataUsageCrawlDebug {
-		console.Debugf(logPrefix+"Start crawling. Bloom filter: %v %s\n", s.withFilter != nil, logSuffix)
+	if s.dataUsageScannerDebug {
+		console.Debugf(logPrefix+"Start scanning. Bloom filter: %v %s\n", s.withFilter != nil, logSuffix)
 	}
 
 	done := ctx.Done()
 	var flattenLevels = 2
 
-	if s.dataUsageCrawlDebug {
+	if s.dataUsageScannerDebug {
 		console.Debugf(logPrefix+"Cycle: %v, Entries: %v %s\n", cache.Info.NextCycle, len(cache.Cache), logSuffix)
 	}
 
 	// Always scan flattenLevels deep. Cache root is level 0.
 	todo := []cachedFolder{{name: cache.Info.Name, objectHealProbDiv: 1}}
 	for i := 0; i < flattenLevels; i++ {
-		if s.dataUsageCrawlDebug {
+		if s.dataUsageScannerDebug {
 			console.Debugf(logPrefix+"Level %v, scanning %v directories. %s\n", i, len(todo), logSuffix)
 		}
 		select {
@@ -257,7 +257,7 @@ func crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache,
 		}
 	}
 
-	if s.dataUsageCrawlDebug {
+	if s.dataUsageScannerDebug {
 		console.Debugf(logPrefix+"New folders: %v %s\n", s.newFolders, logSuffix)
 	}
 
@@ -286,7 +286,7 @@ func crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache,
 		}
 	}
 
-	if s.dataUsageCrawlDebug {
+	if s.dataUsageScannerDebug {
 		console.Debugf(logPrefix+"Existing folders: %v %s\n", len(s.existingFolders), logSuffix)
 	}
 
@@ -313,13 +313,13 @@ func crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache,
 				// If folder isn't in filter, skip it completely.
 				if !s.withFilter.containsDir(folder.name) {
 					if !h.mod(s.oldCache.Info.NextCycle, s.healFolderInclude/folder.objectHealProbDiv) {
-						if s.dataUsageCrawlDebug {
+						if s.dataUsageScannerDebug {
 							console.Debugf(logPrefix+"Skipping non-updated folder: %v %s\n", folder, logSuffix)
 						}
 						s.newCache.replaceHashed(h, folder.parent, s.oldCache.Cache[h.Key()])
 						continue
 					} else {
-						if s.dataUsageCrawlDebug {
+						if s.dataUsageScannerDebug {
 							console.Debugf(logPrefix+"Adding non-updated folder to heal check: %v %s\n", folder.name, logSuffix)
 						}
 						// Update probability of including objects
@@ -341,8 +341,8 @@ func crawlDataFolder(ctx context.Context, basePath string, cache dataUsageCache,
 		}
 		s.newCache.replaceHashed(h, folder.parent, *du)
 	}
-	if s.dataUsageCrawlDebug {
-		console.Debugf(logPrefix+"Finished crawl, %v entries %s\n", len(s.newCache.Cache), logSuffix)
+	if s.dataUsageScannerDebug {
+		console.Debugf(logPrefix+"Finished scanner, %v entries %s\n", len(s.newCache.Cache), logSuffix)
 	}
 	s.newCache.Info.LastUpdate = UTCNow()
 	s.newCache.Info.NextCycle++
@@ -371,7 +371,7 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 		_, 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.dataUsageCrawlDebug {
+			if f.dataUsageScannerDebug {
 				console.Debugf(scannerLogPrefix+" Prefix %q has active rules\n", prefix)
 			}
 			activeLifeCycle = f.oldCache.Info.lifeCycle
@@ -382,12 +382,12 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 			if folder.name != dataUsageRoot && !filter.containsDir(folder.name) {
 				if !thisHash.mod(f.oldCache.Info.NextCycle, f.healFolderInclude/folder.objectHealProbDiv) {
 					f.newCache.copyWithChildren(&f.oldCache, thisHash, folder.parent)
-					if f.dataUsageCrawlDebug {
+					if f.dataUsageScannerDebug {
 						console.Debugf(scannerLogPrefix+" Skipping non-updated folder: %v\n", folder.name)
 					}
 					continue
 				} else {
-					if f.dataUsageCrawlDebug {
+					if f.dataUsageScannerDebug {
 						console.Debugf(scannerLogPrefix+" Adding non-updated folder to heal check: %v\n", folder.name)
 					}
 					// If probability was already scannerHealFolderInclude, keep it.
@@ -395,7 +395,7 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 				}
 			}
 		}
-		scannerSleeper.Sleep(ctx, dataCrawlSleepPerFolder)
+		scannerSleeper.Sleep(ctx, dataScannerSleepPerFolder)
 
 		cache := dataUsageEntry{}
 
@@ -404,14 +404,14 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 			entName = path.Clean(path.Join(folder.name, entName))
 			bucket, prefix := path2BucketObjectWithBasePath(f.root, entName)
 			if bucket == "" {
-				if f.dataUsageCrawlDebug {
+				if f.dataUsageScannerDebug {
 					console.Debugf(scannerLogPrefix+" no bucket (%s,%s)\n", f.root, entName)
 				}
 				return errDoneForNow
 			}
 
 			if isReservedOrInvalidBucket(bucket, false) {
-				if f.dataUsageCrawlDebug {
+				if f.dataUsageScannerDebug {
 					console.Debugf(scannerLogPrefix+" invalid bucket: %v, entry: %v\n", bucket, entName)
 				}
 				return errDoneForNow
@@ -450,13 +450,13 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 			wait := scannerSleeper.Timer(ctx)
 
 			// Get file size, ignore errors.
-			item := crawlItem{
+			item := scannerItem{
 				Path:       path.Join(f.root, entName),
 				Typ:        typ,
 				bucket:     bucket,
 				prefix:     path.Dir(prefix),
 				objectName: path.Base(entName),
-				debug:      f.dataUsageCrawlDebug,
+				debug:      f.dataUsageScannerDebug,
 				lifeCycle:  activeLifeCycle,
 				heal:       thisHash.mod(f.oldCache.Info.NextCycle, f.healObjectSelect/folder.objectHealProbDiv) && globalIsErasure,
 			}
@@ -532,7 +532,7 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 		healObjectsPrefix := color.Green("healObjects:")
 		for k := range existing {
 			bucket, prefix := path2BucketObject(k)
-			if f.dataUsageCrawlDebug {
+			if f.dataUsageScannerDebug {
 				console.Debugf(scannerLogPrefix+" checking disappeared folder: %v/%v\n", bucket, prefix)
 			}
 
@@ -552,13 +552,13 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 				minDisks:       len(f.disks), // We want full consistency.
 				// Weird, maybe transient error.
 				agreed: func(entry metaCacheEntry) {
-					if f.dataUsageCrawlDebug {
+					if f.dataUsageScannerDebug {
 						console.Debugf(healObjectsPrefix+" got agreement: %v\n", entry.name)
 					}
 				},
 				// Some disks have data for this.
 				partial: func(entries metaCacheEntries, nAgreed int, errs []error) {
-					if f.dataUsageCrawlDebug {
+					if f.dataUsageScannerDebug {
 						console.Debugf(healObjectsPrefix+" got partial, %d agreed, errs: %v\n", nAgreed, errs)
 					}
 
@@ -580,7 +580,7 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 						entry, _ = entries.firstFound()
 					}
 
-					if f.dataUsageCrawlDebug {
+					if f.dataUsageScannerDebug {
 						console.Debugf(healObjectsPrefix+" resolved to: %v, dir: %v\n", entry.name, entry.isDir())
 					}
 
@@ -618,20 +618,20 @@ func (f *folderScanner) scanQueuedLevels(ctx context.Context, folders []cachedFo
 				},
 				// Too many disks failed.
 				finished: func(errs []error) {
-					if f.dataUsageCrawlDebug {
+					if f.dataUsageScannerDebug {
 						console.Debugf(healObjectsPrefix+" too many errors: %v\n", errs)
 					}
 					cancel()
 				},
 			})
 
-			if f.dataUsageCrawlDebug && err != nil && err != errFileNotFound {
+			if f.dataUsageScannerDebug && err != nil && err != errFileNotFound {
 				console.Debugf(healObjectsPrefix+" checking returned value %v (%T)\n", err, err)
 			}
 
 			// If we found one or more disks with this folder, delete it.
 			if err == nil && dangling {
-				if f.dataUsageCrawlDebug {
+				if f.dataUsageScannerDebug {
 					console.Debugf(healObjectsPrefix+" deleting dangling directory %s\n", prefix)
 				}
 
@@ -690,7 +690,7 @@ func (f *folderScanner) deepScanFolder(ctx context.Context, folder cachedFolder,
 			dirStack = append(dirStack, entName)
 			err := readDirFn(path.Join(dirStack...), addDir)
 			dirStack = dirStack[:len(dirStack)-1]
-			scannerSleeper.Sleep(ctx, dataCrawlSleepPerFolder)
+			scannerSleeper.Sleep(ctx, dataScannerSleepPerFolder)
 			return err
 		}
 
@@ -705,19 +705,19 @@ func (f *folderScanner) deepScanFolder(ctx context.Context, folder cachedFolder,
 		bucket, prefix := path2BucketObjectWithBasePath(f.root, fileName)
 		var activeLifeCycle *lifecycle.Lifecycle
 		if f.oldCache.Info.lifeCycle != nil && f.oldCache.Info.lifeCycle.HasActiveRules(prefix, false) {
-			if f.dataUsageCrawlDebug {
+			if f.dataUsageScannerDebug {
 				console.Debugf(deepScannerLogPrefix+" Prefix %q has active rules\n", prefix)
 			}
 			activeLifeCycle = f.oldCache.Info.lifeCycle
 		}
 
-		item := crawlItem{
+		item := scannerItem{
 			Path:       fileName,
 			Typ:        typ,
 			bucket:     bucket,
 			prefix:     path.Dir(prefix),
 			objectName: path.Base(entName),
-			debug:      f.dataUsageCrawlDebug,
+			debug:      f.dataUsageScannerDebug,
 			lifeCycle:  activeLifeCycle,
 			heal:       hashPath(path.Join(prefix, entName)).mod(f.oldCache.Info.NextCycle, f.healObjectSelect/folder.objectHealProbDiv) && globalIsErasure,
 		}
@@ -752,8 +752,8 @@ func (f *folderScanner) deepScanFolder(ctx context.Context, folder cachedFolder,
 	return &cache, nil
 }
 
-// crawlItem represents each file while walking.
-type crawlItem struct {
+// scannerItem represents each file while walking.
+type scannerItem struct {
 	Path string
 	Typ  os.FileMode
 
@@ -773,10 +773,10 @@ type sizeSummary struct {
 	replicaSize    int64
 }
 
-type getSizeFn func(item crawlItem) (sizeSummary, error)
+type getSizeFn func(item scannerItem) (sizeSummary, error)
 
 // transformMetaDir will transform a directory to prefix/file.ext
-func (i *crawlItem) transformMetaDir() {
+func (i *scannerItem) transformMetaDir() {
 	split := strings.Split(i.prefix, SlashSeparator)
 	if len(split) > 1 {
 		i.prefix = path.Join(split[:len(split)-1]...)
@@ -799,7 +799,7 @@ var applyActionsLogPrefix = color.Green("applyActions:")
 // The resulting size on disk will always be returned.
 // The metadata will be compared to consensus on the object layer before any changes are applied.
 // If no metadata is supplied, -1 is returned if no action is taken.
-func (i *crawlItem) applyActions(ctx context.Context, o ObjectLayer, meta actionMeta) (size int64) {
+func (i *scannerItem) applyActions(ctx context.Context, o ObjectLayer, meta actionMeta) (size int64) {
 	size, err := meta.oi.GetActualSize()
 	if i.debug {
 		logger.LogIf(ctx, err)
@@ -1052,12 +1052,12 @@ func applyLifecycleAction(ctx context.Context, action lifecycle.Action, objLayer
 }
 
 // objectPath returns the prefix and object name.
-func (i *crawlItem) objectPath() string {
+func (i *scannerItem) objectPath() string {
 	return path.Join(i.prefix, i.objectName)
 }
 
 // healReplication will heal a scanned item that has failed replication.
-func (i *crawlItem) healReplication(ctx context.Context, o ObjectLayer, oi ObjectInfo, sizeS *sizeSummary) {
+func (i *scannerItem) healReplication(ctx context.Context, o ObjectLayer, oi ObjectInfo, sizeS *sizeSummary) {
 	if oi.DeleteMarker || !oi.VersionPurgeStatus.Empty() {
 		// heal delete marker replication failure or versioned delete replication failure
 		if oi.ReplicationStatus == replication.Pending ||
@@ -1082,7 +1082,7 @@ func (i *crawlItem) healReplication(ctx context.Context, o ObjectLayer, oi Objec
 }
 
 // healReplicationDeletes will heal a scanned deleted item that failed to replicate deletes.
-func (i *crawlItem) healReplicationDeletes(ctx context.Context, o ObjectLayer, oi ObjectInfo) {
+func (i *scannerItem) healReplicationDeletes(ctx context.Context, o ObjectLayer, oi ObjectInfo) {
 	// handle soft delete and permanent delete failures here.
 	if oi.DeleteMarker || !oi.VersionPurgeStatus.Empty() {
 		versionID := ""

cmd/data-update-tracker.go

@@ -80,7 +80,7 @@ func newDataUpdateTracker() *dataUpdateTracker {
 		Current: dataUpdateFilter{
 			idx: 1,
 		},
-		debug:      env.Get(envDataUsageCrawlDebug, config.EnableOff) == config.EnableOn || serverDebugLog,
+		debug:      env.Get(envDataUsageScannerDebug, config.EnableOff) == config.EnableOn || serverDebugLog,
 		input:      make(chan string, dataUpdateTrackerQueueSize),
 		save:       make(chan struct{}, 1),
 		saveExited: make(chan struct{}),

cmd/data-usage.go

@@ -28,7 +28,7 @@ import (
 )
 
 const (
-	envDataUsageCrawlDebug = "MINIO_DISK_USAGE_CRAWL_DEBUG"
+	envDataUsageScannerDebug = "MINIO_DISK_USAGE_SCANNER_DEBUG"
 
 	dataUsageRoot   = SlashSeparator
 	dataUsageBucket = minioMetaBucket + SlashSeparator + bucketMetaPrefix

cmd/data-usage_test.go

@@ -51,7 +51,7 @@ func TestDataUsageUpdate(t *testing.T) {
 	}
 	createUsageTestFiles(t, base, bucket, files)
 
-	getSize := func(item crawlItem) (sizeS sizeSummary, err error) {
+	getSize := func(item scannerItem) (sizeS sizeSummary, err error) {
 		if item.Typ&os.ModeDir == 0 {
 			var s os.FileInfo
 			s, err = os.Stat(item.Path)
@@ -64,7 +64,7 @@ func TestDataUsageUpdate(t *testing.T) {
 		return
 	}
 
-	got, err := crawlDataFolder(context.Background(), base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize)
+	got, err := scanDataFolder(context.Background(), base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -185,7 +185,7 @@ func TestDataUsageUpdate(t *testing.T) {
 		},
 	}
 	createUsageTestFiles(t, base, bucket, files)
-	got, err = crawlDataFolder(context.Background(), base, got, getSize)
+	got, err = scanDataFolder(context.Background(), base, got, getSize)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -270,7 +270,7 @@ func TestDataUsageUpdate(t *testing.T) {
 	}
 	// Changed dir must be picked up in this many cycles.
 	for i := 0; i < dataUsageUpdateDirCycles; i++ {
-		got, err = crawlDataFolder(context.Background(), base, got, getSize)
+		got, err = scanDataFolder(context.Background(), base, got, getSize)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -347,7 +347,7 @@ func TestDataUsageUpdatePrefix(t *testing.T) {
 	}
 	createUsageTestFiles(t, base, "", files)
 
-	getSize := func(item crawlItem) (sizeS sizeSummary, err error) {
+	getSize := func(item scannerItem) (sizeS sizeSummary, err error) {
 		if item.Typ&os.ModeDir == 0 {
 			var s os.FileInfo
 			s, err = os.Stat(item.Path)
@@ -359,7 +359,7 @@ func TestDataUsageUpdatePrefix(t *testing.T) {
 		}
 		return
 	}
-	got, err := crawlDataFolder(context.Background(), base, dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket"}}, getSize)
+	got, err := scanDataFolder(context.Background(), base, dataUsageCache{Info: dataUsageCacheInfo{Name: "bucket"}}, getSize)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -469,7 +469,7 @@ func TestDataUsageUpdatePrefix(t *testing.T) {
 		},
 	}
 	createUsageTestFiles(t, base, "", files)
-	got, err = crawlDataFolder(context.Background(), base, got, getSize)
+	got, err = scanDataFolder(context.Background(), base, got, getSize)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -552,7 +552,7 @@ func TestDataUsageUpdatePrefix(t *testing.T) {
 	}
 	// Changed dir must be picked up in this many cycles.
 	for i := 0; i < dataUsageUpdateDirCycles; i++ {
-		got, err = crawlDataFolder(context.Background(), base, got, getSize)
+		got, err = scanDataFolder(context.Background(), base, got, getSize)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -646,7 +646,7 @@ func TestDataUsageCacheSerialize(t *testing.T) {
 	}
 	createUsageTestFiles(t, base, bucket, files)
 
-	getSize := func(item crawlItem) (sizeS sizeSummary, err error) {
+	getSize := func(item scannerItem) (sizeS sizeSummary, err error) {
 		if item.Typ&os.ModeDir == 0 {
 			var s os.FileInfo
 			s, err = os.Stat(item.Path)
@@ -658,7 +658,7 @@ func TestDataUsageCacheSerialize(t *testing.T) {
 		}
 		return
 	}
-	want, err := crawlDataFolder(context.Background(), base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize)
+	want, err := scanDataFolder(context.Background(), base, dataUsageCache{Info: dataUsageCacheInfo{Name: bucket}}, getSize)
 	if err != nil {
 		t.Fatal(err)
 	}

cmd/erasure-metadata.go

@@ -102,7 +102,7 @@ func (fi FileInfo) IsValid() bool {
 func (fi FileInfo) ToObjectInfo(bucket, object string) ObjectInfo {
 	object = decodeDirObject(object)
 	versionID := fi.VersionID
-	if globalBucketVersioningSys.Enabled(bucket) && versionID == "" {
+	if (globalBucketVersioningSys.Enabled(bucket) || globalBucketVersioningSys.Suspended(bucket)) && versionID == "" {
 		versionID = nullVersionID
 	}
 

cmd/erasure-multipart.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"os"
 	"path"
 	"sort"
 	"strconv"
@@ -138,39 +139,39 @@ func (er erasureObjects) deleteAll(ctx context.Context, bucket, prefix string) {
 // Remove the old multipart uploads on the given disk.
 func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk StorageAPI, expiry time.Duration) {
 	now := time.Now()
-	shaDirs, err := disk.ListDir(ctx, minioMetaMultipartBucket, "", -1)
-	if err != nil {
-		return
-	}
-	for _, shaDir := range shaDirs {
-		uploadIDDirs, err := disk.ListDir(ctx, minioMetaMultipartBucket, shaDir, -1)
-		if err != nil {
-			continue
-		}
-		for _, uploadIDDir := range uploadIDDirs {
+	diskPath := disk.Endpoint().Path
+
+	readDirFn(pathJoin(diskPath, minioMetaMultipartBucket), func(shaDir string, typ os.FileMode) error {
+		return readDirFn(pathJoin(diskPath, minioMetaMultipartBucket, shaDir), func(uploadIDDir string, typ os.FileMode) error {
 			uploadIDPath := pathJoin(shaDir, uploadIDDir)
 			fi, err := disk.ReadVersion(ctx, minioMetaMultipartBucket, uploadIDPath, "", false)
 			if err != nil {
-				continue
+				return nil
 			}
+			wait := er.deletedCleanupSleeper.Timer(ctx)
 			if now.Sub(fi.ModTime) > expiry {
 				er.renameAll(ctx, minioMetaMultipartBucket, uploadIDPath)
 			}
+			wait()
+			return nil
+		})
+	})
+
+	readDirFn(pathJoin(diskPath, minioMetaTmpBucket), func(tmpDir string, typ os.FileMode) error {
+		if tmpDir == ".trash/" { // do not remove .trash/ here, it has its own routines
+			return nil
 		}
-	}
-	tmpDirs, err := disk.ListDir(ctx, minioMetaTmpBucket, "", -1)
-	if err != nil {
-		return
-	}
-	for _, tmpDir := range tmpDirs {
 		vi, err := disk.StatVol(ctx, pathJoin(minioMetaTmpBucket, tmpDir))
 		if err != nil {
-			continue
+			return nil
 		}
+		wait := er.deletedCleanupSleeper.Timer(ctx)
 		if now.Sub(vi.Created) > expiry {
 			er.deleteAll(ctx, minioMetaTmpBucket, tmpDir)
 		}
-	}
+		wait()
+		return nil
+	})
 }
 
 // ListMultipartUploads - lists all the pending multipart
@@ -283,7 +284,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	}
 
 	dataBlocks := len(onlineDisks) - parityBlocks
-	fi := newFileInfo(object, dataBlocks, parityBlocks)
+	fi := newFileInfo(pathJoin(bucket, object), dataBlocks, parityBlocks)
 
 	// we now know the number of blocks this object needs for data and parity.
 	// establish the writeQuorum using this data

cmd/erasure-object.go

@@ -654,7 +654,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 	// Initialize parts metadata
 	partsMetadata := make([]FileInfo, len(storageDisks))
 
-	fi := newFileInfo(object, dataDrives, parityDrives)
+	fi := newFileInfo(pathJoin(bucket, object), dataDrives, parityDrives)
 
 	if opts.Versioned {
 		fi.VersionID = opts.VersionID

cmd/erasure-server-pool.go

@@ -360,7 +360,7 @@ func (z *erasureServerPools) StorageInfo(ctx context.Context) (StorageInfo, []er
 	return storageInfo, errs
 }
 
-func (z *erasureServerPools) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error {
+func (z *erasureServerPools) NSScanner(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -379,7 +379,7 @@ func (z *erasureServerPools) CrawlAndGetDataUsage(ctx context.Context, bf *bloom
 		return nil
 	}
 
-	// Crawl latest allBuckets first.
+	// Scanner latest allBuckets first.
 	sort.Slice(allBuckets, func(i, j int) bool {
 		return allBuckets[i].Created.After(allBuckets[j].Created)
 	})
@@ -402,7 +402,7 @@ func (z *erasureServerPools) CrawlAndGetDataUsage(ctx context.Context, bf *bloom
 					}
 				}()
 				// Start scanner. Blocks until done.
-				err := erObj.crawlAndGetDataUsage(ctx, allBuckets, bf, updates)
+				err := erObj.nsScanner(ctx, allBuckets, bf, updates)
 				if err != nil {
 					logger.LogIf(ctx, err)
 					mu.Lock()
@@ -595,27 +595,6 @@ func (z *erasureServerPools) GetObjectNInfo(ctx context.Context, bucket, object
 	return gr, ObjectNotFound{Bucket: bucket, Object: object}
 }
 
-func (z *erasureServerPools) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) error {
-	if err := checkGetObjArgs(ctx, bucket, object); err != nil {
-		return err
-	}
-
-	object = encodeDirObject(object)
-	for _, pool := range z.serverPools {
-		if err := pool.GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts); err != nil {
-			if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
-				continue
-			}
-			return err
-		}
-		return nil
-	}
-	if opts.VersionID != "" {
-		return VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
-	}
-	return ObjectNotFound{Bucket: bucket, Object: object}
-}
-
 func (z *erasureServerPools) GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err error) {
 	if err = checkGetObjArgs(ctx, bucket, object); err != nil {
 		return objInfo, err

cmd/erasure-sets.go

@@ -18,10 +18,10 @@ package cmd
 
 import (
 	"context"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"hash/crc32"
-	"io"
 	"math/rand"
 	"net/http"
 	"sort"
@@ -404,21 +404,28 @@ func newErasureSets(ctx context.Context, endpoints Endpoints, storageDisks []Sto
 
 		// Initialize erasure objects for a given set.
 		s.sets[i] = &erasureObjects{
-			setNumber:          i,
-			setDriveCount:      setDriveCount,
-			defaultParityCount: defaultParityCount,
-			getDisks:           s.GetDisks(i),
-			getLockers:         s.GetLockers(i),
-			getEndpoints:       s.GetEndpoints(i),
-			nsMutex:            mutex,
-			bp:                 bp,
-			mrfOpCh:            make(chan partialOperation, 10000),
+			setNumber:             i,
+			setDriveCount:         setDriveCount,
+			defaultParityCount:    defaultParityCount,
+			getDisks:              s.GetDisks(i),
+			getLockers:            s.GetLockers(i),
+			getEndpoints:          s.GetEndpoints(i),
+			deletedCleanupSleeper: newDynamicSleeper(10, 10*time.Second),
+			nsMutex:               mutex,
+			bp:                    bp,
+			mrfOpCh:               make(chan partialOperation, 10000),
 		}
 	}
 
+	// cleanup ".trash/" folder every 30 minutes with sufficient sleep cycles.
+	const deletedObjectsCleanupInterval = 30 * time.Minute
+
 	// start cleanup stale uploads go-routine.
 	go s.cleanupStaleUploads(ctx, GlobalStaleUploadsCleanupInterval, GlobalStaleUploadsExpiry)
 
+	// start cleanup of deleted objects.
+	go s.cleanupDeletedObjects(ctx, deletedObjectsCleanupInterval)
+
 	// Start the disk monitoring and connect routine.
 	go s.monitorAndConnectEndpoints(ctx, defaultMonitorConnectEndpointInterval)
 	go s.maintainMRFList()
@@ -427,6 +434,25 @@ func newErasureSets(ctx context.Context, endpoints Endpoints, storageDisks []Sto
 	return s, nil
 }
 
+func (s *erasureSets) cleanupDeletedObjects(ctx context.Context, cleanupInterval time.Duration) {
+	timer := time.NewTimer(cleanupInterval)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			// Reset for the next interval
+			timer.Reset(cleanupInterval)
+
+			for _, set := range s.sets {
+				set.cleanupDeletedObjects(ctx)
+			}
+		}
+	}
+}
+
 func (s *erasureSets) cleanupStaleUploads(ctx context.Context, cleanupInterval, expiry time.Duration) {
 	timer := time.NewTimer(cleanupInterval)
 	defer timer.Stop()
@@ -638,9 +664,11 @@ func sipHashMod(key string, cardinality int, id [16]byte) int {
 	if cardinality <= 0 {
 		return -1
 	}
-	sip := siphash.New(id[:])
-	sip.Write([]byte(key))
-	return int(sip.Sum64() % uint64(cardinality))
+	// use the faster version as per siphash docs
+	// https://github.com/dchest/siphash#usage
+	k0, k1 := binary.LittleEndian.Uint64(id[0:8]), binary.LittleEndian.Uint64(id[8:16])
+	sum64 := siphash.Hash(k0, k1, []byte(key))
+	return int(sum64 % uint64(cardinality))
 }
 
 func crcHashMod(key string, cardinality int) int {
@@ -784,13 +812,6 @@ func (s *erasureSets) GetObjectNInfo(ctx context.Context, bucket, object string,
 	return set.GetObjectNInfo(ctx, bucket, object, rs, h, lockType, opts)
 }
 
-// GetObject - reads an object from the hashedSet based on the object name.
-func (s *erasureSets) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) error {
-	set := s.getHashedSet(object)
-	auditObjectErasureSet(ctx, object, set, s.poolNumber)
-	return set.GetObject(ctx, bucket, object, startOffset, length, writer, etag, opts)
-}
-
 func (s *erasureSets) parentDirIsObject(ctx context.Context, bucket, parent string) bool {
 	if parent == "." {
 		return false

cmd/erasure.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
+	"os"
 	"sort"
 	"sync"
 	"time"
@@ -71,6 +72,8 @@ type erasureObjects struct {
 	bp *bpool.BytePoolCap
 
 	mrfOpCh chan partialOperation
+
+	deletedCleanupSleeper *dynamicSleeper
 }
 
 // NewNSLock - initialize a new namespace RWLocker instance.
@@ -273,9 +276,31 @@ func (er erasureObjects) getOnlineDisksWithHealing() (newDisks []StorageAPI, hea
 	return newDisks, healing
 }
 
-// CrawlAndGetDataUsage will start crawling buckets and send updated totals as they are traversed.
+// Clean-up previously deleted objects. from .minio.sys/tmp/.trash/
+func (er erasureObjects) cleanupDeletedObjects(ctx context.Context) {
+	// run multiple cleanup's local to this server.
+	var wg sync.WaitGroup
+	for _, disk := range er.getLoadBalancedLocalDisks() {
+		if disk != nil {
+			wg.Add(1)
+			go func(disk StorageAPI) {
+				defer wg.Done()
+				diskPath := disk.Endpoint().Path
+				readDirFn(pathJoin(diskPath, minioMetaTmpDeletedBucket), func(ddir string, typ os.FileMode) error {
+					wait := er.deletedCleanupSleeper.Timer(ctx)
+					removeAll(pathJoin(diskPath, minioMetaTmpDeletedBucket, ddir))
+					wait()
+					return nil
+				})
+			}(disk)
+		}
+	}
+	wg.Wait()
+}
+
+// nsScanner will start scanning buckets and send updated totals as they are traversed.
 // Updates are sent on a regular basis and the caller *must* consume them.
-func (er erasureObjects) crawlAndGetDataUsage(ctx context.Context, buckets []BucketInfo, bf *bloomFilter, updates chan<- dataUsageCache) error {
+func (er erasureObjects) nsScanner(ctx context.Context, buckets []BucketInfo, bf *bloomFilter, updates chan<- dataUsageCache) error {
 	if len(buckets) == 0 {
 		return nil
 	}
@@ -283,7 +308,7 @@ func (er erasureObjects) crawlAndGetDataUsage(ctx context.Context, buckets []Buc
 	// Collect disks we can use.
 	disks, healing := er.getOnlineDisksWithHealing()
 	if len(disks) == 0 {
-		logger.Info(color.Green("data-crawl:") + " all disks are offline or being healed, skipping crawl")
+		logger.Info(color.Green("data-scanner:") + " all disks are offline or being healed, skipping scanner")
 		return nil
 	}
 
@@ -417,7 +442,7 @@ func (er erasureObjects) crawlAndGetDataUsage(ctx context.Context, buckets []Buc
 				// Calc usage
 				before := cache.Info.LastUpdate
 				var err error
-				cache, err = disk.CrawlAndGetDataUsage(ctx, cache)
+				cache, err = disk.NSScanner(ctx, cache)
 				cache.Info.BloomFilter = nil
 				if err != nil {
 					if !cache.Info.LastUpdate.IsZero() && cache.Info.LastUpdate.After(before) {

cmd/fs-v1-helpers.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	pathutil "path"
 	"runtime"
+	"strings"
 
 	"github.com/minio/minio/cmd/logger"
 	"github.com/minio/minio/pkg/lock"
@@ -392,6 +393,55 @@ func fsRenameFile(ctx context.Context, sourcePath, destPath string) error {
 	return nil
 }
 
+func deleteFile(basePath, deletePath string, recursive bool) error {
+	if basePath == "" || deletePath == "" {
+		return nil
+	}
+	isObjectDir := HasSuffix(deletePath, SlashSeparator)
+	basePath = pathutil.Clean(basePath)
+	deletePath = pathutil.Clean(deletePath)
+	if !strings.HasPrefix(deletePath, basePath) || deletePath == basePath {
+		return nil
+	}
+
+	var err error
+	if recursive {
+		os.RemoveAll(deletePath)
+	} else {
+		err = os.Remove(deletePath)
+	}
+	if err != nil {
+		switch {
+		case isSysErrNotEmpty(err):
+			// if object is a directory, but if its not empty
+			// return FileNotFound to indicate its an empty prefix.
+			if isObjectDir {
+				return errFileNotFound
+			}
+			// Ignore errors if the directory is not empty. The server relies on
+			// this functionality, and sometimes uses recursion that should not
+			// error on parent directories.
+			return nil
+		case osIsNotExist(err):
+			return errFileNotFound
+		case osIsPermission(err):
+			return errFileAccessDenied
+		case isSysErrIO(err):
+			return errFaultyDisk
+		default:
+			return err
+		}
+	}
+
+	deletePath = pathutil.Dir(deletePath)
+
+	// Delete parent directory obviously not recursively. Errors for
+	// parent directories shouldn't trickle down.
+	deleteFile(basePath, deletePath, false)
+
+	return nil
+}
+
 // fsDeleteFile is a wrapper for deleteFile(), after checking the path length.
 func fsDeleteFile(ctx context.Context, basePath, deletePath string) error {
 	if err := checkPathLength(basePath); err != nil {

cmd/fs-v1.go

@@ -231,8 +231,8 @@ func (fs *FSObjects) StorageInfo(ctx context.Context) (StorageInfo, []error) {
 	return storageInfo, nil
 }
 
-// CrawlAndGetDataUsage returns data usage stats of the current FS deployment
-func (fs *FSObjects) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error {
+// NSScanner returns data usage stats of the current FS deployment
+func (fs *FSObjects) NSScanner(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error {
 	// Load bucket totals
 	var totalCache dataUsageCache
 	err := totalCache.load(ctx, fs, dataUsageCacheName)
@@ -268,7 +268,7 @@ func (fs *FSObjects) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter,
 		}
 		bCache.Info.BloomFilter = totalCache.Info.BloomFilter
 
-		cache, err := fs.crawlBucket(ctx, b.Name, bCache)
+		cache, err := fs.scanBucket(ctx, b.Name, bCache)
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
@@ -279,13 +279,13 @@ func (fs *FSObjects) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter,
 
 		if cache.root() == nil {
 			if intDataUpdateTracker.debug {
-				logger.Info(color.Green("CrawlAndGetDataUsage:") + " No root added. Adding empty")
+				logger.Info(color.Green("NSScanner:") + " No root added. Adding empty")
 			}
 			cache.replace(cache.Info.Name, dataUsageRoot, dataUsageEntry{})
 		}
 		if cache.Info.LastUpdate.After(bCache.Info.LastUpdate) {
 			if intDataUpdateTracker.debug {
-				logger.Info(color.Green("CrawlAndGetDataUsage:")+" Saving bucket %q cache with %d entries", b.Name, len(cache.Cache))
+				logger.Info(color.Green("NSScanner:")+" Saving bucket %q cache with %d entries", b.Name, len(cache.Cache))
 			}
 			logger.LogIf(ctx, cache.save(ctx, fs, path.Join(b.Name, dataUsageCacheName)))
 		}
@@ -295,7 +295,7 @@ func (fs *FSObjects) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter,
 		entry := cl.flatten(*cl.root())
 		totalCache.replace(cl.Info.Name, dataUsageRoot, entry)
 		if intDataUpdateTracker.debug {
-			logger.Info(color.Green("CrawlAndGetDataUsage:")+" Saving totals cache with %d entries", len(totalCache.Cache))
+			logger.Info(color.Green("NSScanner:")+" Saving totals cache with %d entries", len(totalCache.Cache))
 		}
 		totalCache.Info.LastUpdate = time.Now()
 		logger.LogIf(ctx, totalCache.save(ctx, fs, dataUsageCacheName))
@@ -307,27 +307,27 @@ func (fs *FSObjects) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter,
 	return nil
 }
 
-// crawlBucket crawls a single bucket in FS mode.
+// scanBucket scans a single bucket in FS mode.
 // The updated cache for the bucket is returned.
 // A partially updated bucket may be returned.
-func (fs *FSObjects) crawlBucket(ctx context.Context, bucket string, cache dataUsageCache) (dataUsageCache, error) {
+func (fs *FSObjects) scanBucket(ctx context.Context, bucket string, cache dataUsageCache) (dataUsageCache, error) {
 	// Get bucket policy
 	// Check if the current bucket has a configured lifecycle policy
 	lc, err := globalLifecycleSys.Get(bucket)
 	if err == nil && lc.HasActiveRules("", true) {
 		if intDataUpdateTracker.debug {
-			logger.Info(color.Green("crawlBucket:") + " lifecycle: Active rules found")
+			logger.Info(color.Green("scanBucket:") + " lifecycle: Active rules found")
 		}
 		cache.Info.lifeCycle = lc
 	}
 
 	// Load bucket info.
-	cache, err = crawlDataFolder(ctx, fs.fsPath, cache, func(item crawlItem) (sizeSummary, error) {
+	cache, err = scanDataFolder(ctx, fs.fsPath, cache, func(item scannerItem) (sizeSummary, error) {
 		bucket, object := item.bucket, item.objectPath()
 		fsMetaBytes, err := xioutil.ReadFile(pathJoin(fs.fsPath, minioMetaBucket, bucketMetaPrefix, bucket, object, fs.metaJSONFile))
 		if err != nil && !osIsNotExist(err) {
 			if intDataUpdateTracker.debug {
-				logger.Info(color.Green("crawlBucket:")+" object return unexpected error: %v/%v: %w", item.bucket, item.objectPath(), err)
+				logger.Info(color.Green("scanBucket:")+" object return unexpected error: %v/%v: %w", item.bucket, item.objectPath(), err)
 			}
 			return sizeSummary{}, errSkipFile
 		}
@@ -348,7 +348,7 @@ func (fs *FSObjects) crawlBucket(ctx context.Context, bucket string, cache dataU
 		fi, fiErr := os.Stat(item.Path)
 		if fiErr != nil {
 			if intDataUpdateTracker.debug {
-				logger.Info(color.Green("crawlBucket:")+" object path missing: %v: %w", item.Path, fiErr)
+				logger.Info(color.Green("scanBucket:")+" object path missing: %v: %w", item.Path, fiErr)
 			}
 			return sizeSummary{}, errSkipFile
 		}
@@ -768,41 +768,6 @@ func (fs *FSObjects) GetObjectNInfo(ctx context.Context, bucket, object string,
 	return objReaderFn(reader, h, opts.CheckPrecondFn, closeFn)
 }
 
-// GetObject - reads an object from the disk.
-// Supports additional parameters like offset and length
-// which are synonymous with HTTP Range requests.
-//
-// startOffset indicates the starting read location of the object.
-// length indicates the total length of the object.
-func (fs *FSObjects) GetObject(ctx context.Context, bucket, object string, offset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) {
-	if opts.VersionID != "" && opts.VersionID != nullVersionID {
-		return VersionNotFound{
-			Bucket:    bucket,
-			Object:    object,
-			VersionID: opts.VersionID,
-		}
-	}
-
-	if err = checkGetObjArgs(ctx, bucket, object); err != nil {
-		return err
-	}
-
-	// Lock the object before reading.
-	lk := fs.NewNSLock(bucket, object)
-	if err := lk.GetRLock(ctx, globalOperationTimeout); err != nil {
-		logger.LogIf(ctx, err)
-		return err
-	}
-	defer lk.RUnlock()
-
-	atomic.AddInt64(&fs.activeIOCount, 1)
-	defer func() {
-		atomic.AddInt64(&fs.activeIOCount, -1)
-	}()
-
-	return fs.getObject(ctx, bucket, object, offset, length, writer, etag, true)
-}
-
 // getObject - wrapper for GetObject
 func (fs *FSObjects) getObject(ctx context.Context, bucket, object string, offset int64, length int64, writer io.Writer, etag string, lock bool) (err error) {
 	if _, err = fs.statBucketDir(ctx, bucket); err != nil {

cmd/gateway-unsupported.go

@@ -39,8 +39,8 @@ func (a GatewayUnsupported) BackendInfo() BackendInfo {
 	return BackendInfo{Type: BackendGateway}
 }
 
-// CrawlAndGetDataUsage - crawl is not implemented for gateway
-func (a GatewayUnsupported) CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error {
+// NSScanner - scanner is not implemented for gateway
+func (a GatewayUnsupported) NSScanner(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error {
 	logger.CriticalIf(ctx, errors.New("not implemented"))
 	return NotImplemented{}
 }

cmd/gateway/azure/gateway-azure.go

@@ -799,7 +799,7 @@ func (a *azureObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 
 	pr, pw := io.Pipe()
 	go func() {
-		err := a.GetObject(ctx, bucket, object, startOffset, length, pw, objInfo.InnerETag, opts)
+		err := a.getObject(ctx, bucket, object, startOffset, length, pw, objInfo.InnerETag, opts)
 		pw.CloseWithError(err)
 	}()
 	// Setup cleanup function to cause the above go-routine to
@@ -814,7 +814,7 @@ func (a *azureObjects) GetObjectNInfo(ctx context.Context, bucket, object string
 //
 // startOffset indicates the starting read location of the object.
 // length indicates the total length of the object.
-func (a *azureObjects) GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
+func (a *azureObjects) getObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
 	// startOffset cannot be negative.
 	if startOffset < 0 {
 		return azureToObjectError(minio.InvalidRange{}, bucket, object)

cmd/gateway/gcs/gateway-gcs.go

@@ -751,7 +751,7 @@ func (l *gcsGateway) GetObjectNInfo(ctx context.Context, bucket, object string,
 
 	pr, pw := io.Pipe()
 	go func() {
-		err := l.GetObject(ctx, bucket, object, startOffset, length, pw, objInfo.ETag, opts)
+		err := l.getObject(ctx, bucket, object, startOffset, length, pw, objInfo.ETag, opts)
 		pw.CloseWithError(err)
 	}()
 	// Setup cleanup function to cause the above go-routine to
@@ -766,7 +766,7 @@ func (l *gcsGateway) GetObjectNInfo(ctx context.Context, bucket, object string,
 //
 // startOffset indicates the starting read location of the object.
 // length indicates the total length of the object.
-func (l *gcsGateway) GetObject(ctx context.Context, bucket string, key string, startOffset int64, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
+func (l *gcsGateway) getObject(ctx context.Context, bucket string, key string, startOffset int64, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
 	// if we want to mimic S3 behavior exactly, we need to verify if bucket exists first,
 	// otherwise gcs will just return object not exist in case of non-existing bucket
 	if _, err := l.client.Bucket(bucket).Attrs(ctx); err != nil {

cmd/gateway/hdfs/gateway-hdfs.go

@@ -591,7 +591,7 @@ func (n *hdfsObjects) GetObjectNInfo(ctx context.Context, bucket, object string,
 
 	pr, pw := io.Pipe()
 	go func() {
-		nerr := n.GetObject(ctx, bucket, object, startOffset, length, pw, objInfo.ETag, opts)
+		nerr := n.getObject(ctx, bucket, object, startOffset, length, pw, objInfo.ETag, opts)
 		pw.CloseWithError(nerr)
 	}()
 
@@ -614,7 +614,7 @@ func (n *hdfsObjects) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
 	})
 }
 
-func (n *hdfsObjects) GetObject(ctx context.Context, bucket, key string, startOffset, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
+func (n *hdfsObjects) getObject(ctx context.Context, bucket, key string, startOffset, length int64, writer io.Writer, etag string, opts minio.ObjectOptions) error {
 	if _, err := n.clnt.Stat(n.hdfsPathJoin(bucket)); err != nil {
 		return hdfsToObjectErr(ctx, err, bucket)
 	}

cmd/gateway/s3/gateway-s3-sse.go

@@ -226,7 +226,7 @@ func (l *s3EncObjects) getGWMetadata(ctx context.Context, bucket, metaFileName s
 		return m, err1
 	}
 	var buffer bytes.Buffer
-	err = l.s3Objects.GetObject(ctx, bucket, metaFileName, 0, oi.Size, &buffer, oi.ETag, minio.ObjectOptions{})
+	err = l.s3Objects.getObject(ctx, bucket, metaFileName, 0, oi.Size, &buffer, oi.ETag, minio.ObjectOptions{})
 	if err != nil {
 		return m, err
 	}
@@ -272,7 +272,7 @@ func (l *s3EncObjects) getObject(ctx context.Context, bucket string, key string,
 	dmeta, err := l.getGWMetadata(ctx, bucket, getDareMetaPath(key))
 	if err != nil {
 		// unencrypted content
-		return l.s3Objects.GetObject(ctx, bucket, key, startOffset, length, writer, etag, o)
+		return l.s3Objects.getObject(ctx, bucket, key, startOffset, length, writer, etag, o)
 	}
 	if startOffset < 0 {
 		logger.LogIf(ctx, minio.InvalidRange{})
@@ -303,7 +303,7 @@ func (l *s3EncObjects) getObject(ctx context.Context, bucket string, key string,
 	if _, _, err := dmeta.ObjectToPartOffset(ctx, endOffset); err != nil {
 		return minio.InvalidRange{OffsetBegin: startOffset, OffsetEnd: length, ResourceSize: dmeta.Stat.Size}
 	}
-	return l.s3Objects.GetObject(ctx, bucket, key, partOffset, endOffset, writer, dmeta.ETag, o)
+	return l.s3Objects.getObject(ctx, bucket, key, partOffset, endOffset, writer, dmeta.ETag, o)
 }
 
 // GetObjectNInfo - returns object info and locked object ReadCloser

cmd/gateway/s3/gateway-s3.go

@@ -403,7 +403,7 @@ func (l *s3Objects) GetObjectNInfo(ctx context.Context, bucket, object string, r
 
 	pr, pw := io.Pipe()
 	go func() {
-		err := l.GetObject(ctx, bucket, object, off, length, pw, objInfo.ETag, opts)
+		err := l.getObject(ctx, bucket, object, off, length, pw, objInfo.ETag, opts)
 		pw.CloseWithError(err)
 	}()
 
@@ -419,7 +419,7 @@ func (l *s3Objects) GetObjectNInfo(ctx context.Context, bucket, object string, r
 //
 // startOffset indicates the starting read location of the object.
 // length indicates the total length of the object.
-func (l *s3Objects) GetObject(ctx context.Context, bucket string, key string, startOffset int64, length int64, writer io.Writer, etag string, o minio.ObjectOptions) error {
+func (l *s3Objects) getObject(ctx context.Context, bucket string, key string, startOffset int64, length int64, writer io.Writer, etag string, o minio.ObjectOptions) error {
 	if length < 0 && length != -1 {
 		return minio.ErrorRespToObjectError(minio.InvalidRange{}, bucket, key)
 	}

cmd/global-heal.go

@@ -32,7 +32,7 @@ const (
 )
 
 // NewBgHealSequence creates a background healing sequence
-// operation which crawls all objects and heal them.
+// operation which scans all objects and heal them.
 func newBgHealSequence() *healSequence {
 	reqInfo := &logger.ReqInfo{API: "BackgroundHeal"}
 	ctx, cancelCtx := context.WithCancel(logger.SetReqInfo(GlobalContext, reqInfo))

cmd/globals.go

@@ -87,8 +87,9 @@ const (
 
 	// GlobalStaleUploadsExpiry - Expiry duration after which the uploads in multipart, tmp directory are deemed stale.
 	GlobalStaleUploadsExpiry = time.Hour * 24 // 24 hrs.
+
 	// GlobalStaleUploadsCleanupInterval - Cleanup interval when the stale uploads cleanup is initiated.
-	GlobalStaleUploadsCleanupInterval = time.Hour * 24 // 24 hrs.
+	GlobalStaleUploadsCleanupInterval = time.Hour * 12 // 12 hrs.
 
 	// GlobalServiceExecutionInterval - Executes the Lifecycle events.
 	GlobalServiceExecutionInterval = time.Hour * 24 // 24 hrs.
@@ -96,7 +97,7 @@ const (
 	// Refresh interval to update in-memory iam config cache.
 	globalRefreshIAMInterval = 5 * time.Minute
 
-	// Limit of location constraint XML for unauthenticted PUT bucket operations.
+	// Limit of location constraint XML for unauthenticated PUT bucket operations.
 	maxLocationConstraintSize = 3 * humanize.MiByte
 
 	// Maximum size of default bucket encryption configuration allowed

cmd/iam.go

@@ -948,9 +948,9 @@ func (sys *IAMSys) ListUsers() (map[string]madmin.UserInfo, error) {
 }
 
 // IsTempUser - returns if given key is a temporary user.
-func (sys *IAMSys) IsTempUser(name string) (bool, error) {
+func (sys *IAMSys) IsTempUser(name string) (bool, string, error) {
 	if !sys.Initialized() {
-		return false, errServerNotInitialized
+		return false, "", errServerNotInitialized
 	}
 
 	sys.store.rlock()
@@ -958,10 +958,14 @@ func (sys *IAMSys) IsTempUser(name string) (bool, error) {
 
 	cred, found := sys.iamUsersMap[name]
 	if !found {
-		return false, errNoSuchUser
+		return false, "", errNoSuchUser
 	}
 
-	return cred.IsTemp(), nil
+	if cred.IsTemp() {
+		return true, cred.ParentUser, nil
+	}
+
+	return false, "", nil
 }
 
 // IsServiceAccount - returns if given key is a service account
@@ -1085,7 +1089,7 @@ func (sys *IAMSys) SetUserStatus(accessKey string, status madmin.AccountStatus)
 }
 
 // NewServiceAccount - create a new service account
-func (sys *IAMSys) NewServiceAccount(ctx context.Context, parentUser string, sessionPolicy *iampolicy.Policy) (auth.Credentials, error) {
+func (sys *IAMSys) NewServiceAccount(ctx context.Context, parentUser string, groups []string, sessionPolicy *iampolicy.Policy) (auth.Credentials, error) {
 	if !sys.Initialized() {
 		return auth.Credentials{}, errServerNotInitialized
 	}
@@ -1114,7 +1118,24 @@ func (sys *IAMSys) NewServiceAccount(ctx context.Context, parentUser string, ses
 
 	cr, ok := sys.iamUsersMap[parentUser]
 	if !ok {
-		return auth.Credentials{}, errNoSuchUser
+		// For LDAP users we would need this fallback
+		if sys.usersSysType != MinIOUsersSysType {
+			_, ok = sys.iamUserPolicyMap[parentUser]
+			if !ok {
+				var found bool
+				for _, group := range groups {
+					_, ok = sys.iamGroupPolicyMap[group]
+					if !ok {
+						continue
+					}
+					found = true
+					break
+				}
+				if !found {
+					return auth.Credentials{}, errNoSuchUser
+				}
+			}
+		}
 	}
 
 	// Disallow service accounts to further create more service accounts.
@@ -1138,6 +1159,7 @@ func (sys *IAMSys) NewServiceAccount(ctx context.Context, parentUser string, ses
 		return auth.Credentials{}, err
 	}
 	cred.ParentUser = parentUser
+	cred.Groups = groups
 
 	u := newUserIdentity(cred)
 
@@ -1751,6 +1773,18 @@ func (sys *IAMSys) policyDBGet(name string, isGroup bool) ([]string, error) {
 		p := sys.iamGroupPolicyMap[group]
 		policies = append(policies, p.toSlice()...)
 	}
+
+	for _, group := range u.Groups {
+		// Skip missing or disabled groups
+		gi, ok := sys.iamGroupsMap[group]
+		if !ok || gi.Status == statusDisabled {
+			continue
+		}
+
+		p := sys.iamGroupPolicyMap[group]
+		policies = append(policies, p.toSlice()...)
+	}
+
 	return policies, nil
 }
 
@@ -1777,13 +1811,13 @@ func (sys *IAMSys) IsAllowedServiceAccount(args iampolicy.Args, parent string) b
 		return false
 	}
 
-	// Check if the parent is allowed to perform this action, reject if not
-	parentUserPolicies, err := sys.PolicyDBGet(parent, false)
+	// Check policy for this service account.
+	svcPolicies, err := sys.PolicyDBGet(args.AccountName, false)
 	if err != nil {
 		return false
 	}
 
-	if len(parentUserPolicies) == 0 {
+	if len(svcPolicies) == 0 {
 		return false
 	}
 
@@ -1791,7 +1825,7 @@ func (sys *IAMSys) IsAllowedServiceAccount(args iampolicy.Args, parent string) b
 
 	// Policies were found, evaluate all of them.
 	sys.store.rlock()
-	for _, pname := range parentUserPolicies {
+	for _, pname := range svcPolicies {
 		p, found := sys.iamPolicyDocsMap[pname]
 		if found {
 			availablePolicies = append(availablePolicies, p)
@@ -1858,74 +1892,68 @@ func (sys *IAMSys) IsAllowedServiceAccount(args iampolicy.Args, parent string) b
 }
 
 // IsAllowedLDAPSTS - checks for LDAP specific claims and values
-func (sys *IAMSys) IsAllowedLDAPSTS(args iampolicy.Args) bool {
-	userIface, ok := args.Claims[ldapUser]
-	if !ok {
-		return false
-	}
-	user, ok := userIface.(string)
-	if !ok {
-		return false
-	}
-
-	sys.store.rlock()
-	defer sys.store.runlock()
-
-	var groups []string
-	cred, ok := sys.iamUsersMap[args.AccountName]
-	if !ok {
-		return false
-	}
-	groups = cred.Groups
-
-	// We look up the policy mapping directly to bypass
-	// users exists, group exists validations that do not
-	// apply here.
-	var policies []iampolicy.Policy
-	if mp, ok := sys.iamUserPolicyMap[user]; ok {
-		for _, pname := range mp.toSlice() {
-			p, found := sys.iamPolicyDocsMap[pname]
-			if !found {
-				logger.LogIf(GlobalContext, fmt.Errorf("expected policy (%s) missing for the LDAPUser %s, rejecting the request", pname, user))
-				return false
-			}
-			policies = append(policies, p)
-		}
-	}
-	for _, group := range groups {
-		mp, ok := sys.iamGroupPolicyMap[group]
+func (sys *IAMSys) IsAllowedLDAPSTS(args iampolicy.Args, parentUser string) bool {
+	parentInClaimIface, ok := args.Claims[ldapUser]
+	if ok {
+		parentInClaim, ok := parentInClaimIface.(string)
 		if !ok {
-			continue
+			// ldap parentInClaim name is not a string reject it.
+			return false
 		}
-		for _, pname := range mp.toSlice() {
-			p, found := sys.iamPolicyDocsMap[pname]
-			if !found {
-				logger.LogIf(GlobalContext, fmt.Errorf("expected policy (%s) missing for the LDAPGroup %s, rejecting the request", pname, group))
-				return false
-			}
-			policies = append(policies, p)
+
+		if parentInClaim != parentUser {
+			// ldap claim has been modified maliciously reject it.
+			return false
 		}
-	}
-	if len(policies) == 0 {
+	} else {
+		// no ldap parentInClaim claim present reject it.
 		return false
 	}
-	combinedPolicy := policies[0]
-	for i := 1; i < len(policies); i++ {
+
+	// Check policy for this LDAP user.
+	ldapPolicies, err := sys.PolicyDBGet(args.AccountName, false)
+	if err != nil {
+		return false
+	}
+
+	if len(ldapPolicies) == 0 {
+		return false
+	}
+
+	var availablePolicies []iampolicy.Policy
+
+	// Policies were found, evaluate all of them.
+	sys.store.rlock()
+	for _, pname := range ldapPolicies {
+		p, found := sys.iamPolicyDocsMap[pname]
+		if found {
+			availablePolicies = append(availablePolicies, p)
+		}
+	}
+	sys.store.runlock()
+
+	if len(availablePolicies) == 0 {
+		return false
+	}
+
+	combinedPolicy := availablePolicies[0]
+	for i := 1; i < len(availablePolicies); i++ {
 		combinedPolicy.Statements =
 			append(combinedPolicy.Statements,
-				policies[i].Statements...)
+				availablePolicies[i].Statements...)
 	}
+
 	return combinedPolicy.IsAllowed(args)
 }
 
 // IsAllowedSTS is meant for STS based temporary credentials,
 // which implements claims validation and verification other than
 // applying policies.
-func (sys *IAMSys) IsAllowedSTS(args iampolicy.Args) bool {
+func (sys *IAMSys) IsAllowedSTS(args iampolicy.Args, parentUser string) bool {
 	// If it is an LDAP request, check that user and group
 	// policies allow the request.
 	if sys.usersSysType == LDAPUsersSysType {
-		return sys.IsAllowedLDAPSTS(args)
+		return sys.IsAllowedLDAPSTS(args, parentUser)
 	}
 
 	policies, ok := args.GetPolicies(iamPolicyClaimNameOpenID())
@@ -2050,16 +2078,16 @@ func (sys *IAMSys) IsAllowed(args iampolicy.Args) bool {
 	}
 
 	// If the credential is temporary, perform STS related checks.
-	ok, err := sys.IsTempUser(args.AccountName)
+	ok, parentUser, err := sys.IsTempUser(args.AccountName)
 	if err != nil {
 		return false
 	}
 	if ok {
-		return sys.IsAllowedSTS(args)
+		return sys.IsAllowedSTS(args, parentUser)
 	}
 
 	// If the credential is for a service account, perform related check
-	ok, parentUser, err := sys.IsServiceAccount(args.AccountName)
+	ok, parentUser, err = sys.IsServiceAccount(args.AccountName)
 	if err != nil {
 		return false
 	}

cmd/jwt.go

@@ -104,7 +104,7 @@ func webTokenCallback(claims *xjwt.MapClaims) ([]byte, error) {
 	if claims.AccessKey == globalActiveCred.AccessKey {
 		return []byte(globalActiveCred.SecretKey), nil
 	}
-	ok, err := globalIAMSys.IsTempUser(claims.AccessKey)
+	ok, _, err := globalIAMSys.IsTempUser(claims.AccessKey)
 	if err != nil {
 		if err == errNoSuchUser {
 			return nil, errInvalidAccessKeyID

cmd/metacache-server-pool.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"os"
 	"path"
+	pathutil "path"
 	"strings"
 	"sync"
 	"time"
@@ -35,7 +36,7 @@ func renameAllBucketMetacache(epPath string) error {
 	// to `.minio.sys/tmp/` for deletion.
 	return readDirFn(pathJoin(epPath, minioMetaBucket, bucketMetaPrefix), func(name string, typ os.FileMode) error {
 		if typ == os.ModeDir {
-			tmpMetacacheOld := pathJoin(epPath, minioMetaTmpBucket+"-old", mustGetUUID())
+			tmpMetacacheOld := pathutil.Join(epPath, minioMetaTmpDeletedBucket, mustGetUUID())
 			if err := renameAll(pathJoin(epPath, minioMetaBucket, metacachePrefixForID(name, slashSeparator)),
 				tmpMetacacheOld); err != nil && err != errFileNotFound {
 				return fmt.Errorf("unable to rename (%s -> %s) %w",

cmd/metacache-walk.go

@@ -34,7 +34,7 @@ import (
 
 // WalkDirOptions provides options for WalkDir operations.
 type WalkDirOptions struct {
-	// Bucket to crawl
+	// Bucket to scanner
 	Bucket string
 
 	// Directory inside the bucket.

cmd/naughty-disk_test.go

@@ -98,8 +98,8 @@ func (d *naughtyDisk) SetDiskID(id string) {
 	d.disk.SetDiskID(id)
 }
 
-func (d *naughtyDisk) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCache) (info dataUsageCache, err error) {
-	return d.disk.CrawlAndGetDataUsage(ctx, cache)
+func (d *naughtyDisk) NSScanner(ctx context.Context, cache dataUsageCache) (info dataUsageCache, err error) {
+	return d.disk.NSScanner(ctx, cache)
 }
 
 func (d *naughtyDisk) DiskInfo(ctx context.Context) (info DiskInfo, err error) {

cmd/object-api-getobject_test.go

@@ -1,613 +0,0 @@
-/*
- * MinIO Cloud Storage, (C) 2016, 2017 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.
- */
-
-package cmd
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"io"
-	"os"
-	"runtime"
-	"strings"
-	"testing"
-
-	humanize "github.com/dustin/go-humanize"
-)
-
-// Wrapper for calling GetObject tests for both Erasure multiple disks and single node setup.
-func TestGetObject(t *testing.T) {
-	ExecExtendedObjectLayerTest(t, testGetObject)
-}
-
-// ObjectLayer.GetObject is called with series of cases for valid and erroneous inputs and the result is validated.
-func testGetObject(obj ObjectLayer, instanceType string, t TestErrHandler) {
-	// Setup for the tests.
-	bucketName := getRandomBucketName()
-	objectName := "test-object"
-	emptyDirName := "test-empty-dir/"
-
-	// create bucket.
-	err := obj.MakeBucketWithLocation(context.Background(), bucketName, BucketOptions{})
-	// Stop the test if creation of the bucket fails.
-	if err != nil {
-		t.Fatalf("%s : %s", instanceType, err.Error())
-	}
-
-	// set of byte data for PutObject.
-	// object has to be created before running tests for GetObject.
-	// this is required even to assert the GetObject data,
-	// since dataInserted === dataFetched back is a primary criteria for any object storage this assertion is critical.
-	bytesData := []struct {
-		byteData []byte
-	}{
-		// Regular data
-		{generateBytesData(6 * humanize.MiByte)},
-		// Empty data for empty directory
-		{},
-	}
-	// set of inputs for uploading the objects before tests for downloading is done.
-	putObjectInputs := []struct {
-		bucketName    string
-		objectName    string
-		contentLength int64
-		textData      []byte
-		metaData      map[string]string
-	}{
-		// case - 1.
-		{bucketName, objectName, int64(len(bytesData[0].byteData)), bytesData[0].byteData, make(map[string]string)},
-		{bucketName, emptyDirName, int64(len(bytesData[1].byteData)), bytesData[1].byteData, make(map[string]string)},
-	}
-	// iterate through the above set of inputs and upkoad the object.
-	for i, input := range putObjectInputs {
-		// uploading the object.
-		_, err = obj.PutObject(context.Background(), input.bucketName, input.objectName, mustGetPutObjReader(t, bytes.NewBuffer(input.textData), input.contentLength, input.metaData["etag"], ""), ObjectOptions{UserDefined: input.metaData})
-		// if object upload fails stop the test.
-		if err != nil {
-			t.Fatalf("Put Object case %d:  Error uploading object: <ERROR> %v", i+1, err)
-		}
-	}
-	// set of empty buffers used to fill GetObject data.
-	buffers := []*bytes.Buffer{
-		new(bytes.Buffer),
-		new(bytes.Buffer),
-		new(bytes.Buffer),
-	}
-
-	// test cases with set of inputs
-	testCases := []struct {
-		bucketName  string
-		objectName  string
-		startOffset int64
-		length      int64
-		// data obtained/fetched from GetObject.
-		getObjectData *bytes.Buffer
-		// writer which governs the write into the `getObjectData`.
-		writer io.Writer
-		// flag indicating whether the test for given ase should pass.
-		shouldPass bool
-		// expected Result.
-		expectedData []byte
-		err          error
-	}{
-		// Test case 1-4.
-		// Cases with invalid bucket names.
-		{".test", "obj", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Bucket name invalid: .test")},
-		{"------", "obj", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Bucket name invalid: ------")},
-		{"$this-is-not-valid-too", "obj", 0, 0, nil, nil, false,
-			[]byte(""), fmt.Errorf("%s", "Bucket name invalid: $this-is-not-valid-too")},
-		{"a", "obj", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Bucket name invalid: a")},
-		// Test case - 5.
-		// Case with invalid object names.
-		{bucketName, "", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Object name invalid: "+bucketName+"/")},
-		//	Test case - 6.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData)), buffers[0], NewEOFWriter(buffers[0], 100), false, []byte{}, io.EOF},
-		// Test case with start offset set to 0 and length set to size of the object.
-		// Fetching the entire object.
-		// 	Test case - 7.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData)), buffers[1], buffers[1], true, bytesData[0].byteData, nil},
-		// Test case with `length` parameter set to a negative value.
-		// Test case - 8.
-		{bucketName, objectName, 0, int64(-1), buffers[1], buffers[1], true, bytesData[0].byteData, nil},
-		// Test case with content-range 1 to objectSize .
-		// Test case - 9.
-		{bucketName, objectName, 1, int64(len(bytesData[0].byteData) - 1), buffers[1], buffers[1], true, bytesData[0].byteData[1:], nil},
-		// Test case with content-range 100 to objectSize - 100.
-		// Test case - 10.
-		{bucketName, objectName, 100, int64(len(bytesData[0].byteData) - 200), buffers[1], buffers[1], true,
-			bytesData[0].byteData[100 : len(bytesData[0].byteData)-100], nil},
-		// Test case with offset greater than the size of the object
-		// Test case - 11.
-		{bucketName, objectName, int64(len(bytesData[0].byteData) + 1), int64(len(bytesData[0].byteData)), buffers[0],
-			NewEOFWriter(buffers[0], 100), false, []byte{},
-			InvalidRange{int64(len(bytesData[0].byteData) + 1), int64(len(bytesData[0].byteData)), int64(len(bytesData[0].byteData))}},
-		// Test case with offset greater than the size of the object.
-		// Test case - 12.
-		{bucketName, objectName, -1, int64(len(bytesData[0].byteData)), buffers[0], new(bytes.Buffer), false, []byte{}, errUnexpected},
-		// Test case length parameter is more than the object size.
-		// Test case - 13.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData) + 1), buffers[1], buffers[1], false, bytesData[0].byteData,
-			InvalidRange{0, int64(len(bytesData[0].byteData) + 1), int64(len(bytesData[0].byteData))}},
-		// Test case with offset + length > objectSize parameter set to a negative value.
-		// Test case - 14.
-		{bucketName, objectName, 2, int64(len(bytesData[0].byteData)), buffers[1], buffers[1], false, bytesData[0].byteData,
-			InvalidRange{2, int64(len(bytesData[0].byteData)), int64(len(bytesData[0].byteData))}},
-		// Test case with the writer set to nil.
-		// Test case - 15.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData)), buffers[1], nil, false, bytesData[0].byteData, errUnexpected},
-		// Test case - 16.
-		// Test case when it is an empty directory
-		{bucketName, emptyDirName, 0, int64(len(bytesData[1].byteData)), buffers[2], buffers[2], true, bytesData[1].byteData, nil},
-	}
-
-	for i, testCase := range testCases {
-		err = obj.GetObject(context.Background(), testCase.bucketName, testCase.objectName, testCase.startOffset, testCase.length, testCase.writer, "", ObjectOptions{})
-		if err != nil && testCase.shouldPass {
-			t.Errorf("Test %d: %s:  Expected to pass, but failed with: <ERROR> %s", i+1, instanceType, err.Error())
-		}
-		if err == nil && !testCase.shouldPass {
-			t.Errorf("Test %d: %s: Expected to fail with <ERROR> \"%s\", but passed instead.", i+1, instanceType, testCase.err.Error())
-		}
-		// Failed as expected, but does it fail for the expected reason.
-		if err != nil && !testCase.shouldPass {
-			if !strings.Contains(err.Error(), testCase.err.Error()) {
-				t.Errorf("Test %d: %s: Expected to fail with error \"%s\", but instead failed with error \"%s\" instead.", i+1, instanceType, testCase.err.Error(), err.Error())
-			}
-		}
-		// Since there are cases for which GetObject fails, this is
-		// necessary. Test passes as expected, but the output values
-		// are verified for correctness here.
-		if err == nil && testCase.shouldPass {
-			if !bytes.Equal(testCase.expectedData, testCase.getObjectData.Bytes()) {
-				t.Errorf("Test %d: %s: Data Mismatch: Expected data and the fetched data from GetObject doesn't match.", i+1, instanceType)
-			}
-			// empty the buffer so that it can be used to further cases.
-			testCase.getObjectData.Reset()
-		}
-	}
-}
-
-// Wrapper for calling GetObject with permission denied expected
-func TestGetObjectPermissionDenied(t *testing.T) {
-	// Windows doesn't support Chmod under golang
-	if runtime.GOOS != globalWindowsOSName {
-		ExecObjectLayerDiskAlteredTest(t, testGetObjectPermissionDenied)
-	}
-}
-
-// Test GetObject when we are allowed to access some dirs and objects
-func testGetObjectPermissionDenied(obj ObjectLayer, instanceType string, disks []string, t *testing.T) {
-	// Setup for the tests.
-	bucketName := getRandomBucketName()
-	// create bucket.
-	err := obj.MakeBucketWithLocation(context.Background(), bucketName, BucketOptions{})
-	// Stop the test if creation of the bucket fails.
-	if err != nil {
-		t.Fatalf("%s : %s", instanceType, err.Error())
-	}
-
-	bytesData := []struct {
-		byteData []byte
-	}{
-		{generateBytesData(6 * humanize.MiByte)},
-	}
-	// set of inputs for uploading the objects before tests for downloading is done.
-	putObjectInputs := []struct {
-		bucketName    string
-		objectName    string
-		contentLength int64
-		textData      []byte
-		metaData      map[string]string
-	}{
-		{bucketName, "test-object1", int64(len(bytesData[0].byteData)), bytesData[0].byteData, make(map[string]string)},
-		{bucketName, "test-object2", int64(len(bytesData[0].byteData)), bytesData[0].byteData, make(map[string]string)},
-		{bucketName, "dir/test-object3", int64(len(bytesData[0].byteData)), bytesData[0].byteData, make(map[string]string)},
-	}
-	// iterate through the above set of inputs and upkoad the object.
-	for i, input := range putObjectInputs {
-		// uploading the object.
-		_, err = obj.PutObject(context.Background(), input.bucketName, input.objectName, mustGetPutObjReader(t, bytes.NewBuffer(input.textData), input.contentLength, input.metaData["etag"], ""), ObjectOptions{UserDefined: input.metaData})
-		// if object upload fails stop the test.
-		if err != nil {
-			t.Fatalf("Put Object case %d:  Error uploading object: <ERROR> %v", i+1, err)
-		}
-	}
-
-	// set of empty buffers used to fill GetObject data.
-	buffers := []*bytes.Buffer{
-		new(bytes.Buffer),
-	}
-
-	// test cases with set of inputs
-	testCases := []struct {
-		bucketName  string
-		objectName  string
-		chmodPath   string
-		startOffset int64
-		length      int64
-		// data obtained/fetched from GetObject.
-		getObjectData *bytes.Buffer
-		// writer which governs the write into the `getObjectData`.
-		writer io.Writer
-		// flag indicating whether the test for given ase should pass.
-		shouldPass bool
-		// expected Result.
-		expectedData []byte
-		err          error
-	}{
-		// Test 1 - chmod 000 bucket/test-object1
-		{bucketName, "test-object1", "test-object1", 0, int64(len(bytesData[0].byteData)), buffers[0], buffers[0], false, bytesData[0].byteData, PrefixAccessDenied{Bucket: bucketName, Object: "test-object1"}},
-		// Test 2 - chmod 000 bucket/dir/
-		{bucketName, "dir/test-object2", "dir", 0, int64(len(bytesData[0].byteData)), buffers[0], buffers[0], false, bytesData[0].byteData, PrefixAccessDenied{Bucket: bucketName, Object: "dir/test-object2"}},
-		// Test 3 - chmod 000 bucket/
-		{bucketName, "test-object3", "", 0, int64(len(bytesData[0].byteData)), buffers[0], buffers[0], false, bytesData[0].byteData, PrefixAccessDenied{Bucket: bucketName, Object: "test-object3"}},
-	}
-
-	for i, testCase := range testCases {
-		for _, d := range disks {
-			err = os.Chmod(d+SlashSeparator+testCase.bucketName+SlashSeparator+testCase.chmodPath, 0)
-			if err != nil {
-				t.Fatalf("Test %d, Unable to chmod: %v", i+1, err)
-			}
-		}
-
-		err = obj.GetObject(context.Background(), testCase.bucketName, testCase.objectName, testCase.startOffset, testCase.length, testCase.writer, "", ObjectOptions{})
-		if err != nil && testCase.shouldPass {
-			t.Errorf("Test %d: %s:  Expected to pass, but failed with: <ERROR> %s", i+1, instanceType, err.Error())
-		}
-		if err == nil && !testCase.shouldPass {
-			t.Errorf("Test %d: %s: Expected to fail with <ERROR> \"%s\", but passed instead.", i+1, instanceType, testCase.err.Error())
-		}
-		// Failed as expected, but does it fail for the expected reason.
-		if err != nil && !testCase.shouldPass {
-			if !strings.Contains(err.Error(), testCase.err.Error()) {
-				t.Errorf("Test %d: %s: Expected to fail with error \"%s\", but instead failed with error \"%s\" instead.", i+1, instanceType, testCase.err.Error(), err.Error())
-			}
-		}
-		// Since there are cases for which GetObject fails, this is
-		// necessary. Test passes as expected, but the output values
-		// are verified for correctness here.
-		if err == nil && testCase.shouldPass {
-			if !bytes.Equal(testCase.expectedData, testCase.getObjectData.Bytes()) {
-				t.Errorf("Test %d: %s: Data Mismatch: Expected data and the fetched data from GetObject doesn't match.", i+1, instanceType)
-			}
-			// empty the buffer so that it can be used to further cases.
-			testCase.getObjectData.Reset()
-		}
-	}
-
-}
-
-// Wrapper for calling GetObject tests for both Erasure multiple disks and single node setup.
-func TestGetObjectDiskNotFound(t *testing.T) {
-	ExecObjectLayerDiskAlteredTest(t, testGetObjectDiskNotFound)
-}
-
-// ObjectLayer.GetObject is called with series of cases for valid and erroneous inputs and the result is validated.
-// Before the Get Object call Erasure disks are moved so that the quorum just holds.
-func testGetObjectDiskNotFound(obj ObjectLayer, instanceType string, disks []string, t *testing.T) {
-	// Setup for the tests.
-	bucketName := getRandomBucketName()
-	objectName := "test-object"
-	// create bucket.
-	err := obj.MakeBucketWithLocation(context.Background(), bucketName, BucketOptions{})
-	// Stop the test if creation of the bucket fails.
-	if err != nil {
-		t.Fatalf("%s : %s", instanceType, err.Error())
-	}
-
-	// set of byte data for PutObject.
-	// object has to be created before running tests for GetObject.
-	// this is required even to assert the GetObject data,
-	// since dataInserted === dataFetched back is a primary criteria for any object storage this assertion is critical.
-	bytesData := []struct {
-		byteData []byte
-	}{
-		{generateBytesData(6 * humanize.MiByte)},
-	}
-	// set of inputs for uploading the objects before tests for downloading is done.
-	putObjectInputs := []struct {
-		bucketName    string
-		objectName    string
-		contentLength int64
-		textData      []byte
-		metaData      map[string]string
-	}{
-		// case - 1.
-		{bucketName, objectName, int64(len(bytesData[0].byteData)), bytesData[0].byteData, make(map[string]string)},
-	}
-	// iterate through the above set of inputs and upkoad the object.
-	for i, input := range putObjectInputs {
-		// uploading the object.
-		_, err = obj.PutObject(context.Background(), input.bucketName, input.objectName, mustGetPutObjReader(t, bytes.NewBuffer(input.textData), input.contentLength, input.metaData["etag"], ""), ObjectOptions{UserDefined: input.metaData})
-		// if object upload fails stop the test.
-		if err != nil {
-			t.Fatalf("Put Object case %d:  Error uploading object: <ERROR> %v", i+1, err)
-		}
-	}
-
-	// Take 4 disks down before GetObject is called, one more we loose quorum on 16 disk node.
-	for _, disk := range disks[:4] {
-		os.RemoveAll(disk)
-	}
-
-	// set of empty buffers used to fill GetObject data.
-	buffers := []*bytes.Buffer{
-		new(bytes.Buffer),
-		new(bytes.Buffer),
-	}
-
-	// test cases with set of inputs
-	testCases := []struct {
-		bucketName  string
-		objectName  string
-		startOffset int64
-		length      int64
-		// data obtained/fetched from GetObject.
-		getObjectData *bytes.Buffer
-		// writer which governs the write into the `getObjectData`.
-		writer io.Writer
-		// flag indicating whether the test for given ase should pass.
-		shouldPass bool
-		// expected Result.
-		expectedData []byte
-		err          error
-	}{
-		// Test case 1-4.
-		// Cases with invalid bucket names.
-		{".test", "obj", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Bucket name invalid: .test")},
-		{"------", "obj", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Bucket name invalid: ------")},
-		{"$this-is-not-valid-too", "obj", 0, 0, nil, nil, false,
-			[]byte(""), fmt.Errorf("%s", "Bucket name invalid: $this-is-not-valid-too")},
-		{"a", "obj", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Bucket name invalid: a")},
-		// Test case - 5.
-		// Case with invalid object names.
-		{bucketName, "", 0, 0, nil, nil, false, []byte(""), fmt.Errorf("%s", "Object name invalid: "+bucketName+"/")},
-		//	Test case - 7.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData)), buffers[0], NewEOFWriter(buffers[0], 100), false, []byte{}, io.EOF},
-		// Test case with start offset set to 0 and length set to size of the object.
-		// Fetching the entire object.
-		// 	Test case - 8.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData)), buffers[1], buffers[1], true, bytesData[0].byteData, nil},
-		// Test case with `length` parameter set to a negative value.
-		// Test case - 9.
-		{bucketName, objectName, 0, int64(-1), buffers[1], buffers[1], true, bytesData[0].byteData, nil},
-		// Test case with `length` parameter set to a negative value and offset is positive.
-		// Test case - 10.
-		{bucketName, objectName, 1, int64(-1), buffers[1], buffers[1], true, bytesData[0].byteData[1:], nil},
-		// Test case with content-range 1 to objectSize .
-		// Test case - 11.
-		{bucketName, objectName, 1, int64(len(bytesData[0].byteData) - 1), buffers[1], buffers[1], true, bytesData[0].byteData[1:], nil},
-		// Test case with content-range 100 to objectSize - 100.
-		// Test case - 12.
-		{bucketName, objectName, 100, int64(len(bytesData[0].byteData) - 200), buffers[1], buffers[1], true,
-			bytesData[0].byteData[100 : len(bytesData[0].byteData)-100], nil},
-		// Test case with offset greater than the size of the object
-		// Test case - 13.
-		{bucketName, objectName, int64(len(bytesData[0].byteData) + 1), int64(len(bytesData[0].byteData)), buffers[0],
-			NewEOFWriter(buffers[0], 100), false, []byte{},
-			InvalidRange{int64(len(bytesData[0].byteData) + 1), int64(len(bytesData[0].byteData)), int64(len(bytesData[0].byteData))}},
-		// Test case with offset greater than the size of the object.
-		// Test case - 14.
-		{bucketName, objectName, -1, int64(len(bytesData[0].byteData)), buffers[0], new(bytes.Buffer), false, []byte{}, errUnexpected},
-		// Test case length parameter is more than the object size.
-		// Test case - 15.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData) + 1), buffers[1], buffers[1], false, bytesData[0].byteData,
-			InvalidRange{0, int64(len(bytesData[0].byteData) + 1), int64(len(bytesData[0].byteData))}},
-		// Test case with offset + length > objectSize parameter set to a negative value.
-		// Test case - 16.
-		{bucketName, objectName, 2, int64(len(bytesData[0].byteData)), buffers[1], buffers[1], false, bytesData[0].byteData,
-			InvalidRange{2, int64(len(bytesData[0].byteData)), int64(len(bytesData[0].byteData))}},
-		// Test case with the writer set to nil.
-		// Test case - 17.
-		{bucketName, objectName, 0, int64(len(bytesData[0].byteData)), buffers[1], nil, false, bytesData[0].byteData, errUnexpected},
-	}
-
-	for i, testCase := range testCases {
-		err = obj.GetObject(context.Background(), testCase.bucketName, testCase.objectName, testCase.startOffset, testCase.length, testCase.writer, "", ObjectOptions{})
-		if err != nil && testCase.shouldPass {
-			t.Errorf("Test %d: %s:  Expected to pass, but failed with: <ERROR> %s", i+1, instanceType, err.Error())
-		}
-		if err == nil && !testCase.shouldPass {
-			t.Errorf("Test %d: %s: Expected to fail with <ERROR> \"%s\", but passed instead.", i+1, instanceType, testCase.err.Error())
-		}
-		// Failed as expected, but does it fail for the expected reason.
-		if err != nil && !testCase.shouldPass {
-			if !strings.Contains(err.Error(), testCase.err.Error()) {
-				t.Errorf("Test %d: %s: Expected to fail with error \"%s\", but instead failed with error \"%s\" instead.", i+1, instanceType, testCase.err.Error(), err.Error())
-			}
-		}
-		// Since there are cases for which GetObject fails, this is
-		// necessary. Test passes as expected, but the output values
-		// are verified for correctness here.
-		if err == nil && testCase.shouldPass {
-			if !bytes.Equal(testCase.expectedData, testCase.getObjectData.Bytes()) {
-				t.Errorf("Test %d: %s: Data Mismatch: Expected data and the fetched data from GetObject doesn't match.", i+1, instanceType)
-			}
-			// empty the buffer so that it can be used to further cases.
-			testCase.getObjectData.Reset()
-		}
-	}
-}
-
-// Benchmarks for ObjectLayer.GetObject().
-// The intent is to benchmark GetObject for various sizes ranging from few bytes to 100MB.
-// Also each of these Benchmarks are run both Erasure and FS backends.
-
-// BenchmarkGetObjectVerySmallFS - Benchmark FS.GetObject() for object size of 10 bytes.
-func BenchmarkGetObjectVerySmallFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 10)
-}
-
-// BenchmarkGetObjectVerySmallErasure - Benchmark Erasure.GetObject() for object size of 10 bytes.
-func BenchmarkGetObjectVerySmallErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 10)
-}
-
-// BenchmarkGetObject10KbFS - Benchmark FS.GetObject() for object size of 10KB.
-func BenchmarkGetObject10KbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 10*humanize.KiByte)
-}
-
-// BenchmarkGetObject10KbErasure - Benchmark Erasure.GetObject() for object size of 10KB.
-func BenchmarkGetObject10KbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 10*humanize.KiByte)
-}
-
-// BenchmarkGetObject100KbFS - Benchmark FS.GetObject() for object size of 100KB.
-func BenchmarkGetObject100KbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 100*humanize.KiByte)
-}
-
-// BenchmarkGetObject100KbErasure - Benchmark Erasure.GetObject() for object size of 100KB.
-func BenchmarkGetObject100KbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 100*humanize.KiByte)
-}
-
-// BenchmarkGetObject1MbFS - Benchmark FS.GetObject() for object size of 1MB.
-func BenchmarkGetObject1MbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 1*humanize.MiByte)
-}
-
-// BenchmarkGetObject1MbErasure - Benchmark Erasure.GetObject() for object size of 1MB.
-func BenchmarkGetObject1MbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 1*humanize.MiByte)
-}
-
-// BenchmarkGetObject5MbFS - Benchmark FS.GetObject() for object size of 5MB.
-func BenchmarkGetObject5MbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 5*humanize.MiByte)
-}
-
-// BenchmarkGetObject5MbErasure - Benchmark Erasure.GetObject() for object size of 5MB.
-func BenchmarkGetObject5MbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 5*humanize.MiByte)
-}
-
-// BenchmarkGetObject10MbFS - Benchmark FS.GetObject() for object size of 10MB.
-func BenchmarkGetObject10MbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 10*humanize.MiByte)
-}
-
-// BenchmarkGetObject10MbErasure - Benchmark Erasure.GetObject() for object size of 10MB.
-func BenchmarkGetObject10MbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 10*humanize.MiByte)
-}
-
-// BenchmarkGetObject25MbFS - Benchmark FS.GetObject() for object size of 25MB.
-func BenchmarkGetObject25MbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 25*humanize.MiByte)
-
-}
-
-// BenchmarkGetObject25MbErasure - Benchmark Erasure.GetObject() for object size of 25MB.
-func BenchmarkGetObject25MbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 25*humanize.MiByte)
-}
-
-// BenchmarkGetObject50MbFS - Benchmark FS.GetObject() for object size of 50MB.
-func BenchmarkGetObject50MbFS(b *testing.B) {
-	benchmarkGetObject(b, "FS", 50*humanize.MiByte)
-}
-
-// BenchmarkGetObject50MbErasure - Benchmark Erasure.GetObject() for object size of 50MB.
-func BenchmarkGetObject50MbErasure(b *testing.B) {
-	benchmarkGetObject(b, "Erasure", 50*humanize.MiByte)
-}
-
-// parallel benchmarks for ObjectLayer.GetObject() .
-
-// BenchmarkGetObjectParallelVerySmallFS - Benchmark FS.GetObject() for object size of 10 bytes.
-func BenchmarkGetObjectParallelVerySmallFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 10)
-}
-
-// BenchmarkGetObjectParallelVerySmallErasure - Benchmark Erasure.GetObject() for object size of 10 bytes.
-func BenchmarkGetObjectParallelVerySmallErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 10)
-}
-
-// BenchmarkGetObjectParallel10KbFS - Benchmark FS.GetObject() for object size of 10KB.
-func BenchmarkGetObjectParallel10KbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 10*humanize.KiByte)
-}
-
-// BenchmarkGetObjectParallel10KbErasure - Benchmark Erasure.GetObject() for object size of 10KB.
-func BenchmarkGetObjectParallel10KbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 10*humanize.KiByte)
-}
-
-// BenchmarkGetObjectParallel100KbFS - Benchmark FS.GetObject() for object size of 100KB.
-func BenchmarkGetObjectParallel100KbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 100*humanize.KiByte)
-}
-
-// BenchmarkGetObjectParallel100KbErasure - Benchmark Erasure.GetObject() for object size of 100KB.
-func BenchmarkGetObjectParallel100KbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 100*humanize.KiByte)
-}
-
-// BenchmarkGetObjectParallel1MbFS - Benchmark FS.GetObject() for object size of 1MB.
-func BenchmarkGetObjectParallel1MbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 1*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel1MbErasure - Benchmark Erasure.GetObject() for object size of 1MB.
-func BenchmarkGetObjectParallel1MbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 1*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel5MbFS - Benchmark FS.GetObject() for object size of 5MB.
-func BenchmarkGetObjectParallel5MbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 5*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel5MbErasure - Benchmark Erasure.GetObject() for object size of 5MB.
-func BenchmarkGetObjectParallel5MbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 5*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel10MbFS - Benchmark FS.GetObject() for object size of 10MB.
-func BenchmarkGetObjectParallel10MbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 10*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel10MbErasure - Benchmark Erasure.GetObject() for object size of 10MB.
-func BenchmarkGetObjectParallel10MbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 10*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel25MbFS - Benchmark FS.GetObject() for object size of 25MB.
-func BenchmarkGetObjectParallel25MbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 25*humanize.MiByte)
-
-}
-
-// BenchmarkGetObjectParallel25MbErasure - Benchmark Erasure.GetObject() for object size of 25MB.
-func BenchmarkGetObjectParallel25MbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 25*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel50MbFS - Benchmark FS.GetObject() for object size of 50MB.
-func BenchmarkGetObjectParallel50MbFS(b *testing.B) {
-	benchmarkGetObjectParallel(b, "FS", 50*humanize.MiByte)
-}
-
-// BenchmarkGetObjectParallel50MbErasure - Benchmark Erasure.GetObject() for object size of 50MB.
-func BenchmarkGetObjectParallel50MbErasure(b *testing.B) {
-	benchmarkGetObjectParallel(b, "Erasure", 50*humanize.MiByte)
-}

cmd/object-api-interface.go

@@ -87,7 +87,7 @@ type ObjectLayer interface {
 
 	// Storage operations.
 	Shutdown(context.Context) error
-	CrawlAndGetDataUsage(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error
+	NSScanner(ctx context.Context, bf *bloomFilter, updates chan<- DataUsageInfo) error
 
 	BackendInfo() BackendInfo
 	StorageInfo(ctx context.Context) (StorageInfo, []error) // local queries only local disks
@@ -112,7 +112,6 @@ type ObjectLayer interface {
 	// IMPORTANTLY, when implementations return err != nil, this
 	// function MUST NOT return a non-nil ReadCloser.
 	GetObjectNInfo(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, lockType LockType, opts ObjectOptions) (reader *GetObjectReader, err error)
-	GetObject(ctx context.Context, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error)
 	GetObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (objInfo ObjectInfo, err 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)
@@ -162,3 +161,23 @@ type ObjectLayer interface {
 	GetObjectTags(context.Context, string, string, ObjectOptions) (*tags.Tags, error)
 	DeleteObjectTags(context.Context, string, string, ObjectOptions) (ObjectInfo, error)
 }
+
+// GetObject - TODO(aead): This function just acts as an adapter for GetObject tests and benchmarks
+// since the GetObject method of the ObjectLayer interface has been removed. Once, the
+// tests are adjusted to use GetObjectNInfo this function can be removed.
+func GetObject(ctx context.Context, api ObjectLayer, bucket, object string, startOffset int64, length int64, writer io.Writer, etag string, opts ObjectOptions) (err error) {
+	var header http.Header
+	if etag != "" {
+		header.Set("ETag", etag)
+	}
+	Range := &HTTPRangeSpec{Start: startOffset, End: startOffset + length}
+
+	reader, err := api.GetObjectNInfo(ctx, bucket, object, Range, header, readLock, opts)
+	if err != nil {
+		return err
+	}
+	defer reader.Close()
+
+	_, err = io.Copy(writer, reader)
+	return err
+}

cmd/object-api-putobject_test.go

@@ -340,8 +340,19 @@ func testObjectAPIPutObjectStaleFiles(obj ObjectLayer, instanceType string, disk
 
 	for _, disk := range disks {
 		tmpMetaDir := path.Join(disk, minioMetaTmpBucket)
-		if !isDirEmpty(tmpMetaDir) {
-			t.Fatalf("%s: expected: empty, got: non-empty", minioMetaTmpBucket)
+		files, err := ioutil.ReadDir(tmpMetaDir)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var found bool
+		for _, fi := range files {
+			if fi.Name() == ".trash" {
+				continue
+			}
+			found = true
+		}
+		if found {
+			t.Fatalf("%s: expected: empty, got: non-empty %#v", minioMetaTmpBucket, files)
 		}
 	}
 }
@@ -418,8 +429,17 @@ func testObjectAPIMultipartPutObjectStaleFiles(obj ObjectLayer, instanceType str
 			t.Errorf("%s", err)
 		}
 
-		if len(files) != 0 {
-			t.Fatalf("%s: expected: empty, got: non-empty. content: %s", tmpMetaDir, files)
+		var found bool
+		for _, fi := range files {
+			if fi.Name() == ".trash" {
+				continue
+			}
+			found = true
+			break
+		}
+
+		if found {
+			t.Fatalf("%s: expected: empty, got: non-empty. content: %#v", tmpMetaDir, files)
 		}
 	}
 }

cmd/object-api-utils.go

@@ -58,8 +58,11 @@ const (
 	mpartMetaPrefix = "multipart"
 	// MinIO Multipart meta prefix.
 	minioMetaMultipartBucket = minioMetaBucket + SlashSeparator + mpartMetaPrefix
-	// MinIO Tmp meta prefix.
+	// MinIO tmp meta prefix.
 	minioMetaTmpBucket = minioMetaBucket + "/tmp"
+	// MinIO tmp meta prefix for deleted objects.
+	minioMetaTmpDeletedBucket = minioMetaTmpBucket + "/.trash"
+
 	// DNS separator (period), used for bucket name validation.
 	dnsDelimiter = "."
 	// On compressed files bigger than this;

cmd/object-handlers.go

@@ -54,6 +54,7 @@ import (
 	"github.com/minio/minio/pkg/hash"
 	iampolicy "github.com/minio/minio/pkg/iam/policy"
 	"github.com/minio/minio/pkg/ioutil"
+	xnet "github.com/minio/minio/pkg/net"
 	"github.com/minio/minio/pkg/s3select"
 	"github.com/minio/sio"
 )
@@ -507,6 +508,10 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
 		if !httpWriter.HasWritten() && !statusCodeWritten {
 			// write error response only if no data or headers has been written to client yet
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
+			return
+		}
+		if !xnet.IsNetworkOrHostDown(err, true) { // do not need to log disconnected clients
+			logger.LogIf(ctx, fmt.Errorf("Unable to write all the data to client %w", err))
 		}
 		return
 	}
@@ -516,6 +521,10 @@ func (api objectAPIHandlers) GetObjectHandler(w http.ResponseWriter, r *http.Req
 			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
 			return
 		}
+		if !xnet.IsNetworkOrHostDown(err, true) { // do not need to log disconnected clients
+			logger.LogIf(ctx, fmt.Errorf("Unable to write all the data to client %w", err))
+		}
+		return
 	}
 
 	// Notify object accessed via a GET request.

cmd/object_api_suite_test.go

@@ -210,7 +210,7 @@ func testMultipleObjectCreation(obj ObjectLayer, instanceType string, t TestErrH
 
 	for key, value := range objects {
 		var byteBuffer bytes.Buffer
-		err = obj.GetObject(context.Background(), "bucket", key, 0, int64(len(value)), &byteBuffer, "", opts)
+		err = GetObject(context.Background(), obj, "bucket", key, 0, int64(len(value)), &byteBuffer, "", opts)
 		if err != nil {
 			t.Fatalf("%s: <ERROR> %s", instanceType, err)
 		}
@@ -461,7 +461,7 @@ func testObjectOverwriteWorks(obj ObjectLayer, instanceType string, t TestErrHan
 	}
 
 	var bytesBuffer bytes.Buffer
-	err = obj.GetObject(context.Background(), "bucket", "object", 0, length, &bytesBuffer, "", opts)
+	err = GetObject(context.Background(), obj, "bucket", "object", 0, length, &bytesBuffer, "", opts)
 	if err != nil {
 		t.Fatalf("%s: <ERROR> %s", instanceType, err)
 	}
@@ -614,7 +614,7 @@ func testPutObject(obj ObjectLayer, instanceType string, t TestErrHandler) {
 	if err != nil {
 		t.Fatalf("%s: <ERROR> %s", instanceType, err)
 	}
-	err = obj.GetObject(context.Background(), "bucket", "object", 0, length, &bytesBuffer1, "", opts)
+	err = GetObject(context.Background(), obj, "bucket", "object", 0, length, &bytesBuffer1, "", opts)
 	if err != nil {
 		t.Fatalf("%s: <ERROR> %s", instanceType, err)
 	}
@@ -627,7 +627,7 @@ func testPutObject(obj ObjectLayer, instanceType string, t TestErrHandler) {
 	if err != nil {
 		t.Fatalf("%s: <ERROR> %s", instanceType, err)
 	}
-	err = obj.GetObject(context.Background(), "bucket", "object", 0, length, &bytesBuffer2, "", opts)
+	err = GetObject(context.Background(), obj, "bucket", "object", 0, length, &bytesBuffer2, "", opts)
 	if err != nil {
 		t.Fatalf("%s: <ERROR> %s", instanceType, err)
 	}
@@ -658,7 +658,7 @@ func testPutObjectInSubdir(obj ObjectLayer, instanceType string, t TestErrHandle
 	}
 
 	var bytesBuffer bytes.Buffer
-	err = obj.GetObject(context.Background(), "bucket", "dir1/dir2/object", 0, length, &bytesBuffer, "", opts)
+	err = GetObject(context.Background(), obj, "bucket", "dir1/dir2/object", 0, length, &bytesBuffer, "", opts)
 	if err != nil {
 		t.Fatalf("%s: <ERROR> %s", instanceType, err)
 	}

cmd/storage-interface.go

@@ -39,7 +39,7 @@ type StorageAPI interface {
 	Healing() bool // Returns if disk is healing.
 
 	DiskInfo(ctx context.Context) (info DiskInfo, err error)
-	CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCache) (dataUsageCache, error)
+	NSScanner(ctx context.Context, cache dataUsageCache) (dataUsageCache, error)
 
 	// Volume operations.
 	MakeVol(ctx context.Context, volume string) (err error)

cmd/storage-rest-client.go

@@ -170,12 +170,12 @@ func (client *storageRESTClient) Healing() bool {
 	return false
 }
 
-func (client *storageRESTClient) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCache) (dataUsageCache, error) {
+func (client *storageRESTClient) NSScanner(ctx context.Context, cache dataUsageCache) (dataUsageCache, error) {
 	pr, pw := io.Pipe()
 	go func() {
 		pw.CloseWithError(cache.serializeTo(pw))
 	}()
-	respBody, err := client.call(ctx, storageRESTMethodCrawlAndGetDataUsage, url.Values{}, pr, -1)
+	respBody, err := client.call(ctx, storageRESTMethodNSScanner, url.Values{}, pr, -1)
 	defer http.DrainBody(respBody)
 	if err != nil {
 		pr.Close()

cmd/storage-rest-common.go

@@ -17,20 +17,20 @@
 package cmd
 
 const (
-	storageRESTVersion       = "v27" // Add force-delete-marker to DeleteVersion
+	storageRESTVersion       = "v28" // Renamed crawl -> scanner
 	storageRESTVersionPrefix = SlashSeparator + storageRESTVersion
 	storageRESTPrefix        = minioReservedBucketPath + "/storage"
 )
 
 const (
-	storageRESTMethodHealth               = "/health"
-	storageRESTMethodDiskInfo             = "/diskinfo"
-	storageRESTMethodCrawlAndGetDataUsage = "/crawlandgetdatausage"
-	storageRESTMethodMakeVol              = "/makevol"
-	storageRESTMethodMakeVolBulk          = "/makevolbulk"
-	storageRESTMethodStatVol              = "/statvol"
-	storageRESTMethodDeleteVol            = "/deletevol"
-	storageRESTMethodListVols             = "/listvols"
+	storageRESTMethodHealth      = "/health"
+	storageRESTMethodDiskInfo    = "/diskinfo"
+	storageRESTMethodNSScanner   = "/nsscanner"
+	storageRESTMethodMakeVol     = "/makevol"
+	storageRESTMethodMakeVolBulk = "/makevolbulk"
+	storageRESTMethodStatVol     = "/statvol"
+	storageRESTMethodDeleteVol   = "/deletevol"
+	storageRESTMethodListVols    = "/listvols"
 
 	storageRESTMethodAppendFile     = "/appendfile"
 	storageRESTMethodCreateFile     = "/createfile"

cmd/storage-rest-server.go

@@ -40,6 +40,7 @@ import (
 	xhttp "github.com/minio/minio/cmd/http"
 	xjwt "github.com/minio/minio/cmd/jwt"
 	"github.com/minio/minio/cmd/logger"
+	xnet "github.com/minio/minio/pkg/net"
 )
 
 var errDiskStale = errors.New("disk stale")
@@ -156,7 +157,7 @@ func (s *storageRESTServer) DiskInfoHandler(w http.ResponseWriter, r *http.Reque
 	logger.LogIf(r.Context(), msgp.Encode(w, &info))
 }
 
-func (s *storageRESTServer) CrawlAndGetDataUsageHandler(w http.ResponseWriter, r *http.Request) {
+func (s *storageRESTServer) NSScannerHandler(w http.ResponseWriter, r *http.Request) {
 	if !s.IsValid(w, r) {
 		return
 	}
@@ -172,7 +173,7 @@ func (s *storageRESTServer) CrawlAndGetDataUsageHandler(w http.ResponseWriter, r
 	}
 
 	resp := streamHTTPResponse(w)
-	usageInfo, err := s.storage.CrawlAndGetDataUsage(r.Context(), cache)
+	usageInfo, err := s.storage.NSScanner(r.Context(), cache)
 	if err != nil {
 		resp.CloseWithError(err)
 		return
@@ -527,7 +528,12 @@ func (s *storageRESTServer) ReadFileStreamHandler(w http.ResponseWriter, r *http
 	defer rc.Close()
 
 	w.Header().Set(xhttp.ContentLength, strconv.Itoa(length))
-	io.Copy(w, rc)
+	if _, err = io.Copy(w, rc); err != nil {
+		if !xnet.IsNetworkOrHostDown(err, true) { // do not need to log disconnected clients
+			logger.LogIf(r.Context(), err)
+		}
+		return
+	}
 	w.(http.Flusher).Flush()
 }
 
@@ -684,7 +690,7 @@ func (s *storageRESTServer) RenameFileHandler(w http.ResponseWriter, r *http.Req
 }
 
 // keepHTTPResponseAlive can be used to avoid timeouts with long storage
-// operations, such as bitrot verification or data usage crawling.
+// operations, such as bitrot verification or data usage scanning.
 // Every 10 seconds a space character is sent.
 // The returned function should always be called to release resources.
 // An optional error can be sent which will be picked as text only error,
@@ -808,7 +814,7 @@ func (h *httpStreamResponse) CloseWithError(err error) {
 }
 
 // streamHTTPResponse can be used to avoid timeouts with long storage
-// operations, such as bitrot verification or data usage crawling.
+// operations, such as bitrot verification or data usage scanning.
 // Every 10 seconds a space character is sent.
 // The returned function should always be called to release resources.
 // An optional error can be sent which will be picked as text only error,
@@ -1029,7 +1035,7 @@ func registerStorageRESTHandlers(router *mux.Router, endpointServerPools Endpoin
 
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodHealth).HandlerFunc(httpTraceHdrs(server.HealthHandler))
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodDiskInfo).HandlerFunc(httpTraceHdrs(server.DiskInfoHandler))
-			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodCrawlAndGetDataUsage).HandlerFunc(httpTraceHdrs(server.CrawlAndGetDataUsageHandler))
+			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodNSScanner).HandlerFunc(httpTraceHdrs(server.NSScannerHandler))
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodMakeVol).HandlerFunc(httpTraceHdrs(server.MakeVolHandler)).Queries(restQueries(storageRESTVolume)...)
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodMakeVolBulk).HandlerFunc(httpTraceHdrs(server.MakeVolBulkHandler)).Queries(restQueries(storageRESTVolumes)...)
 			subrouter.Methods(http.MethodPost).Path(storageRESTVersionPrefix + storageRESTMethodStatVol).HandlerFunc(httpTraceHdrs(server.StatVolHandler)).Queries(restQueries(storageRESTVolume)...)

cmd/utils.go

@@ -322,8 +322,7 @@ func startProfiler(profilerType string) (minioProfiler, error) {
 			return buf.Bytes(), err
 		}
 	case madmin.ProfilerBlock:
-		prof.recordBase("block", 0)
-		runtime.SetBlockProfileRate(1)
+		runtime.SetBlockProfileRate(100)
 		prof.stopFn = func() ([]byte, error) {
 			var buf bytes.Buffer
 			err := pprof.Lookup("block").WriteTo(&buf, 0)

cmd/web-handlers_test.go

@@ -789,7 +789,7 @@ func testUploadWebHandler(obj ObjectLayer, instanceType string, t TestErrHandler
 	}
 
 	var byteBuffer bytes.Buffer
-	err = obj.GetObject(context.Background(), bucketName, objectName, 0, int64(len(content)), &byteBuffer, "", ObjectOptions{})
+	err = GetObject(context.Background(), obj, bucketName, objectName, 0, int64(len(content)), &byteBuffer, "", ObjectOptions{})
 	if err != nil {
 		t.Fatalf("Failed, %v", err)
 	}

cmd/xl-storage-disk-id-check.go

@@ -55,11 +55,11 @@ func (p *xlStorageDiskIDCheck) Healing() bool {
 	return p.storage.Healing()
 }
 
-func (p *xlStorageDiskIDCheck) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCache) (dataUsageCache, error) {
+func (p *xlStorageDiskIDCheck) NSScanner(ctx context.Context, cache dataUsageCache) (dataUsageCache, error) {
 	if err := p.checkDiskStale(); err != nil {
 		return dataUsageCache{}, err
 	}
-	return p.storage.CrawlAndGetDataUsage(ctx, cache)
+	return p.storage.NSScanner(ctx, cache)
 }
 
 func (p *xlStorageDiskIDCheck) Close() error {

cmd/xl-storage.go

@@ -353,7 +353,7 @@ func (s *xlStorage) Healing() bool {
 	return err == nil
 }
 
-func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCache) (dataUsageCache, error) {
+func (s *xlStorage) NSScanner(ctx context.Context, cache dataUsageCache) (dataUsageCache, error) {
 	var lc *lifecycle.Lifecycle
 	var err error
 
@@ -363,7 +363,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
 		if err == nil && lc.HasActiveRules("", true) {
 			cache.Info.lifeCycle = lc
 			if intDataUpdateTracker.debug {
-				console.Debugln(color.Green("crawlDisk:") + " lifecycle: Active rules found")
+				console.Debugln(color.Green("scannerDisk:") + " lifecycle: Active rules found")
 			}
 		}
 	}
@@ -375,7 +375,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
 	healOpts := globalHealConfig
 	globalHealConfigMu.Unlock()
 
-	dataUsageInfo, err := crawlDataFolder(ctx, s.diskPath, cache, func(item crawlItem) (sizeSummary, error) {
+	dataUsageInfo, err := scanDataFolder(ctx, s.diskPath, cache, func(item scannerItem) (sizeSummary, error) {
 		// Look for `xl.meta/xl.json' at the leaf.
 		if !strings.HasSuffix(item.Path, SlashSeparator+xlStorageFormatFile) &&
 			!strings.HasSuffix(item.Path, SlashSeparator+xlStorageFormatFileV1) {
@@ -386,7 +386,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
 		buf, err := xioutil.ReadFile(item.Path)
 		if err != nil {
 			if intDataUpdateTracker.debug {
-				console.Debugf(color.Green("crawlBucket:")+" object path missing: %v: %w\n", item.Path, err)
+				console.Debugf(color.Green("scannerBucket:")+" object path missing: %v: %w\n", item.Path, err)
 			}
 			return sizeSummary{}, errSkipFile
 		}
@@ -397,7 +397,7 @@ func (s *xlStorage) CrawlAndGetDataUsage(ctx context.Context, cache dataUsageCac
 		fivs, err := getFileInfoVersions(buf, item.bucket, item.objectPath())
 		if err != nil {
 			if intDataUpdateTracker.debug {
-				console.Debugf(color.Green("crawlBucket:")+" reading xl.meta failed: %v: %w\n", item.Path, err)
+				console.Debugf(color.Green("scannerBucket:")+" reading xl.meta failed: %v: %w\n", item.Path, err)
 			}
 			return sizeSummary{}, errSkipFile
 		}
@@ -940,7 +940,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
 	if !isXL2V1Format(buf) {
 		// Delete the meta file, if there are no more versions the
 		// top level parent is automatically removed.
-		return deleteFile(volumeDir, pathJoin(volumeDir, path), true)
+		return s.deleteFile(volumeDir, pathJoin(volumeDir, path), true)
 	}
 
 	var xlMeta xlMetaV2
@@ -967,7 +967,8 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
 			return err
 		}
 
-		if err = removeAll(filePath); err != nil {
+		tmpuuid := mustGetUUID()
+		if err = renameAll(filePath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, tmpuuid)); err != nil {
 			return err
 		}
 	}
@@ -985,7 +986,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
 		return err
 	}
 
-	return deleteFile(volumeDir, filePath, false)
+	return s.deleteFile(volumeDir, filePath, false)
 }
 
 // WriteMetadata - writes FileInfo metadata for path at `xl.meta`
@@ -1189,7 +1190,16 @@ func (s *xlStorage) readAllData(volumeDir string, filePath string, requireDirect
 	}
 
 	atomic.AddInt32(&s.activeIOCount, 1)
-	rd := &odirectReader{f, nil, nil, true, true, s, nil}
+	or := &odirectReader{f, nil, nil, true, true, s, nil}
+	rd := struct {
+		io.Reader
+		io.Closer
+	}{Reader: or, Closer: closeWrapper(func() error {
+		defer func() {
+			atomic.AddInt32(&s.activeIOCount, -1)
+		}()
+		return or.Close()
+	})}
 	defer rd.Close() // activeIOCount is decremented in Close()
 
 	buf, err = ioutil.ReadAll(rd)
@@ -1422,9 +1432,6 @@ func (o *odirectReader) Close() error {
 	} else {
 		o.s.poolLarge.Put(o.bufp)
 	}
-	defer func() {
-		atomic.AddInt32(&o.s.activeIOCount, -1)
-	}()
 	return o.f.Close()
 }
 
@@ -1448,10 +1455,10 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 	var file *os.File
 	// O_DIRECT only supported if offset is zero
 	if offset == 0 && globalStorageClass.GetDMA() == storageclass.DMAReadWrite && s.readODirectSupported {
-		file, err = disk.OpenFileDirectIO(filePath, os.O_RDONLY, 0666)
+		file, err = disk.OpenFileDirectIO(filePath, readMode, 0666)
 	} else {
 		// Open the file for reading.
-		file, err = os.Open(filePath)
+		file, err = os.OpenFile(filePath, readMode, 0666)
 	}
 	if err != nil {
 		switch {
@@ -1478,27 +1485,33 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 
 	st, err := file.Stat()
 	if err != nil {
+		file.Close()
 		return nil, err
 	}
 
 	// Verify it is a regular file, otherwise subsequent Seek is
 	// undefined.
 	if !st.Mode().IsRegular() {
+		file.Close()
 		return nil, errIsNotRegular
 	}
 
 	atomic.AddInt32(&s.activeIOCount, 1)
 	if offset == 0 && globalStorageClass.GetDMA() == storageclass.DMAReadWrite && s.readODirectSupported {
+		or := &odirectReader{file, nil, nil, true, false, s, nil}
 		if length <= smallFileThreshold {
-			return &odirectReader{file, nil, nil, true, true, s, nil}, nil
-		}
-		return &odirectReader{file, nil, nil, true, false, s, nil}, nil
-	}
-
-	if offset > 0 {
-		if _, err = file.Seek(offset, io.SeekStart); err != nil {
-			return nil, err
+			or = &odirectReader{file, nil, nil, true, true, s, nil}
 		}
+		r := struct {
+			io.Reader
+			io.Closer
+		}{Reader: io.LimitReader(or, length), Closer: closeWrapper(func() error {
+			defer func() {
+				atomic.AddInt32(&s.activeIOCount, -1)
+			}()
+			return or.Close()
+		})}
+		return r, nil
 	}
 
 	r := struct {
@@ -1511,6 +1524,13 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 		return file.Close()
 	})}
 
+	if offset > 0 {
+		if _, err = file.Seek(offset, io.SeekStart); err != nil {
+			r.Close()
+			return nil, err
+		}
+	}
+
 	// Add readahead to big reads
 	if length >= readAheadSize {
 		rc, err := readahead.NewReadCloserSize(r, readAheadBuffers, readAheadBufSize)
@@ -1828,7 +1848,7 @@ func (s *xlStorage) CheckFile(ctx context.Context, volume string, path string) e
 // move up the tree, deleting empty parent directories until it finds one
 // with files in it. Returns nil for a non-empty directory even when
 // recursive is set to false.
-func deleteFile(basePath, deletePath string, recursive bool) error {
+func (s *xlStorage) deleteFile(basePath, deletePath string, recursive bool) error {
 	if basePath == "" || deletePath == "" {
 		return nil
 	}
@@ -1841,7 +1861,8 @@ func deleteFile(basePath, deletePath string, recursive bool) error {
 
 	var err error
 	if recursive {
-		err = os.RemoveAll(deletePath)
+		tmpuuid := mustGetUUID()
+		err = renameAll(deletePath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, tmpuuid))
 	} else {
 		err = os.Remove(deletePath)
 	}
@@ -1872,7 +1893,7 @@ func deleteFile(basePath, deletePath string, recursive bool) error {
 
 	// Delete parent directory obviously not recursively. Errors for
 	// parent directories shouldn't trickle down.
-	deleteFile(basePath, deletePath, false)
+	s.deleteFile(basePath, deletePath, false)
 
 	return nil
 }
@@ -1910,46 +1931,7 @@ func (s *xlStorage) Delete(ctx context.Context, volume string, path string, recu
 	}
 
 	// Delete file and delete parent directory as well if it's empty.
-	return deleteFile(volumeDir, filePath, recursive)
-}
-
-func (s *xlStorage) DeleteFileBulk(volume string, paths []string) (errs []error, err error) {
-	atomic.AddInt32(&s.activeIOCount, 1)
-	defer func() {
-		atomic.AddInt32(&s.activeIOCount, -1)
-	}()
-
-	volumeDir, err := s.getVolDir(volume)
-	if err != nil {
-		return nil, err
-	}
-
-	// Stat a volume entry.
-	_, err = os.Lstat(volumeDir)
-	if err != nil {
-		if osIsNotExist(err) {
-			return nil, errVolumeNotFound
-		} else if osIsPermission(err) {
-			return nil, errVolumeAccessDenied
-		} else if isSysErrIO(err) {
-			return nil, errFaultyDisk
-		}
-		return nil, err
-	}
-
-	errs = make([]error, len(paths))
-	// Following code is needed so that we retain SlashSeparator
-	// suffix if any in path argument.
-	for idx, path := range paths {
-		filePath := pathJoin(volumeDir, path)
-		errs[idx] = checkPathLength(filePath)
-		if errs[idx] != nil {
-			continue
-		}
-		// Delete file and delete parent directory as well if its empty.
-		errs[idx] = deleteFile(volumeDir, filePath, false)
-	}
-	return
+	return s.deleteFile(volumeDir, filePath, recursive)
 }
 
 // RenameData - rename source path to destination path atomically, metadata and data directory.
@@ -2180,8 +2162,10 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath, dataDir,
 
 	// Commit data
 	if srcDataPath != "" {
-		removeAll(oldDstDataPath)
-		removeAll(dstDataPath)
+		tmpuuid := mustGetUUID()
+		renameAll(oldDstDataPath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, tmpuuid))
+		tmpuuid = mustGetUUID()
+		renameAll(dstDataPath, pathutil.Join(s.diskPath, minioMetaTmpDeletedBucket, tmpuuid))
 		if err = renameAll(srcDataPath, dstDataPath); err != nil {
 			return osErrToFileErr(err)
 		}
@@ -2194,12 +2178,12 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath, dataDir,
 
 	// Remove parent dir of the source file if empty
 	if parentDir := pathutil.Dir(srcFilePath); isDirEmpty(parentDir) {
-		deleteFile(srcVolumeDir, parentDir, false)
+		s.deleteFile(srcVolumeDir, parentDir, false)
 	}
 
 	if srcDataPath != "" {
 		if parentDir := pathutil.Dir(srcDataPath); isDirEmpty(parentDir) {
-			deleteFile(srcVolumeDir, parentDir, false)
+			s.deleteFile(srcVolumeDir, parentDir, false)
 		}
 	}
 
@@ -2286,7 +2270,7 @@ func (s *xlStorage) RenameFile(ctx context.Context, srcVolume, srcPath, dstVolum
 
 	// Remove parent dir of the source file if empty
 	if parentDir := pathutil.Dir(srcFilePath); isDirEmpty(parentDir) {
-		deleteFile(srcVolumeDir, parentDir, false)
+		s.deleteFile(srcVolumeDir, parentDir, false)
 	}
 
 	return nil

docs/bucket/replication/README.md

@@ -139,7 +139,7 @@ The replication configuration follows [AWS S3 Spec](https://docs.aws.amazon.com/
 
 When object locking is used in conjunction with replication, both source and destination buckets needs to have [object locking](https://docs.min.io/docs/minio-bucket-object-lock-guide.html) enabled. Similarly objects encrypted on the server side, will be replicated if destination also supports encryption.
 
-Replication status can be seen in the metadata on the source and destination objects. On the source side, the `X-Amz-Replication-Status` changes from `PENDING` to `COMPLETED` or `FAILED` after replication attempt either succeeded or failed respectively. On the destination side, a `X-Amz-Replication-Status` status of `REPLICA` indicates that the object was replicated successfully. Any replication failures are automatically re-attempted during a periodic disk crawl cycle.
+Replication status can be seen in the metadata on the source and destination objects. On the source side, the `X-Amz-Replication-Status` changes from `PENDING` to `COMPLETED` or `FAILED` after replication attempt either succeeded or failed respectively. On the destination side, a `X-Amz-Replication-Status` status of `REPLICA` indicates that the object was replicated successfully. Any replication failures are automatically re-attempted during a periodic disk scanner cycle.
 
 To perform bi-directional replication, repeat the above process on the target site - this time setting the source bucket as the replication target. It is recommended that replication be run in a system with atleast two CPU's available to the process, so that replication can run in its own thread.
 

docs/config/README.md

@@ -263,21 +263,21 @@ The following sub-systems are dynamic i.e., configuration parameters for each su
 ```
 api                   manage global HTTP API call specific features, such as throttling, authentication types, etc.
 heal                  manage object healing frequency and bitrot verification checks
-scanner               manage crawling for usage calculation, lifecycle, healing and more
+scanner               manage namespace scanning for usage calculation, lifecycle, healing and more
 ```
 
 > NOTE: if you set any of the following sub-system configuration using ENVs, dynamic behavior is not supported.
 
 ### Usage scanner
 
-Data usage scanner is enabled by default. The following configuration settings allow for more staggered delay in terms of usage calculation. The scanner adapts to the system speed and completely pauses when the system is under load. It is possible to adjust the speed of the scanner and thereby the latency of updates being reflected. The delays between each operation of the crawl can be adjusted by the `mc admin config set alias/ delay=15.0`. By default the value is `10.0`. This means the scanner will sleep *10x* the time each operation takes.
+Data usage scanner is enabled by default. The following configuration settings allow for more staggered delay in terms of usage calculation. The scanner adapts to the system speed and completely pauses when the system is under load. It is possible to adjust the speed of the scanner and thereby the latency of updates being reflected. The delays between each operation of the scanner can be adjusted by the `mc admin config set alias/ delay=15.0`. By default the value is `10.0`. This means the scanner will sleep *10x* the time each operation takes.
 
 In most setups this will keep the scanner slow enough to not impact overall system performance. Setting the `delay` key to a *lower* value will make the scanner faster and setting it to 0 will make the scanner run at full speed (not recommended in production). Setting it to a higher value will make the scanner slower, consuming less resources with the trade off of not collecting metrics for operations like healing and disk usage as fast.
 
 ```
 ~ mc admin config set alias/ scanner
 KEY:
-scanner  manage crawling for usage calculation, lifecycle, healing and more
+scanner  manage namespace scanning for usage calculation, lifecycle, healing and more
 
 ARGS:
 delay     (float)     scanner delay multiplier, defaults to '10.0'
@@ -306,7 +306,7 @@ KEY:
 heal  manage object healing frequency and bitrot verification checks
 
 ARGS:
-bitrotscan  (on|off)    perform bitrot scan on disks when checking objects during crawl
+bitrotscan  (on|off)    perform bitrot scan on disks when checking objects during scanner
 max_sleep   (duration)  maximum sleep duration between objects to slow down heal operation. eg. 2s
 max_io      (int)       maximum IO requests allowed between objects to slow down heal operation. eg. 3
 ```

docs/orchestration/docker-compose/docker-compose.yaml

@@ -5,7 +5,7 @@ version: '3.7'
 # it through port 9000.
 services:
   minio1:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     volumes:
       - data1-1:/data1
       - data1-2:/data2
@@ -22,7 +22,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     volumes:
       - data2-1:/data1
       - data2-2:/data2
@@ -39,7 +39,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     volumes:
       - data3-1:/data1
       - data3-2:/data2
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     volumes:
       - data4-1:/data1
       - data4-2:/data2

docs/orchestration/docker-swarm/docker-compose-secrets.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -29,7 +29,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -56,7 +56,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -83,7 +83,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/orchestration/docker-swarm/docker-compose.yaml

@@ -2,7 +2,7 @@ version: '3.7'
 
 services:
   minio1:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio1
     volumes:
       - minio1-data:/export
@@ -33,7 +33,7 @@ services:
       retries: 3
 
   minio2:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio2
     volumes:
       - minio2-data:/export
@@ -64,7 +64,7 @@ services:
       retries: 3
 
   minio3:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio3
     volumes:
       - minio3-data:/export
@@ -95,7 +95,7 @@ services:
       retries: 3
 
   minio4:
-    image: minio/minio:RELEASE.2021-02-23T20-05-01Z
+    image: minio/minio:RELEASE.2021-02-24T18-44-45Z
     hostname: minio4
     volumes:
       - minio4-data:/export

docs/sts/ldap.md

@@ -85,6 +85,8 @@ MINIO_IDENTITY_LDAP_USER_DN_SEARCH_BASE_DN  (string)    Base LDAP DN to search f
 MINIO_IDENTITY_LDAP_USER_DN_SEARCH_FILTER   (string)    Search filter to lookup user DN
 ```
 
+If you set an empty lookup bind password, the lookup bind will use the unauthenticated authentication mechanism, as described in [RFC 4513 Section 5.1.2](https://tools.ietf.org/html/rfc4513#section-5.1.2).
+
 #### Username-Format Mode ####
 
 In this mode, the server does not use a separate LDAP service account. Instead, the username and password provided in the STS API call are used to login to the LDAP server and also to lookup the user's groups. This mode preserves older behavior for compatibility, but users are encouraged to use the Lookup-Bind mode.

docs/zh_CN/config/README.md

@@ -267,15 +267,15 @@ MINIO_ETCD_COMMENT          (sentence)  optionally add a comment to this setting
 
 数据使用情况采集器默认是启用的，通过Envs可以设置更多的交错延迟。
 
-采集器能适应系统速度，并在系统负载时完全暂停。 可以调整采集器的速度，从而达到延迟更新的效果。 每次采集操作之间的延迟都可以通过环境变量`MINIO_DISK_USAGE_CRAWL_DELAY`来调整。 默认情况下，该值为10。 这意味着采集每次操作都将休眠*10x*的时间。
+采集器能适应系统速度，并在系统负载时完全暂停。 可以调整采集器的速度，从而达到延迟更新的效果。 每次采集操作之间的延迟都可以通过环境变量`MINIO_SCANNER_DELAY`来调整。 默认情况下，该值为10。 这意味着采集每次操作都将休眠*10x*的时间。
 
 大多数设置要让采集器足够慢，这样不会影响整体的系统性能。
-设置 `MINIO_DISK_USAGE_CRAWL_DELAY` 为一个 *较低* 的值可以让采集器更快，并且设置为0的时候，可以让采集器全速运行（不推荐）。 设置一个较高的值可以让采集器变慢，进一步减少资源的消耗。
+设置 `MINIO_SCANNER_DELAY` 为一个 *较低* 的值可以让采集器更快，并且设置为0的时候，可以让采集器全速运行（不推荐）。 设置一个较高的值可以让采集器变慢，进一步减少资源的消耗。
 
 示例: 如下设置将使采集器的速度降低三倍, 减少了系统资源的使用，但是反映到更新的延迟会增加。
 
 ```sh
-export MINIO_DISK_USAGE_CRAWL_DELAY=30
+export MINIO_SCANNER_DELAY=30
 minio server /data
 ```