Bump github.com/nats-io/nats-server/v2 from 2.7.2 to 2.7.4

Fixes: #14531

.github/workflows/go-healing.yml

@@ -37,10 +37,7 @@ jobs:
         env:
           CGO_ENABLED: 0
           GO111MODULE: on
-          MINIO_KMS_KES_CERT_FILE: /home/runner/work/minio/minio/.github/workflows/root.cert
-          MINIO_KMS_KES_KEY_FILE: /home/runner/work/minio/minio/.github/workflows/root.key
-          MINIO_KMS_KES_ENDPOINT: "https://play.min.io:7373"
-          MINIO_KMS_KES_KEY_NAME: "my-minio-key"
+          MINIO_KMS_SECRET_KEY: "my-minio-key:oyArl7zlPECEduNbB1KXgdzDn2Bdpvvw0l8VO51HQnY="
           MINIO_KMS_AUTO_ENCRYPTION: on
         run: |
           sudo sysctl net.ipv6.conf.all.disable_ipv6=0

cmd/admin-handlers.go

@@ -936,6 +936,50 @@ func (a adminAPIHandlers) BackgroundHealStatusHandler(w http.ResponseWriter, r *
 	}
 }
 
+// NetperfHandler - perform mesh style network throughput test
+func (a adminAPIHandlers) NetperfHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "NetperfHandler")
+
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.HealthInfoAdminAction)
+	if objectAPI == nil {
+		return
+	}
+
+	if !globalIsDistErasure {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
+		return
+	}
+
+	nsLock := objectAPI.NewNSLock(minioMetaBucket, "netperf")
+	lkctx, err := nsLock.GetLock(ctx, globalOperationTimeout)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(toAPIErrorCode(ctx, err)), r.URL)
+		return
+	}
+	defer nsLock.Unlock(lkctx.Cancel)
+
+	durationStr := r.Form.Get(peerRESTDuration)
+	duration, err := time.ParseDuration(durationStr)
+	if err != nil {
+		duration = globalNetPerfMinDuration
+	}
+
+	if duration < globalNetPerfMinDuration {
+		// We need sample size of minimum 10 secs.
+		duration = globalNetPerfMinDuration
+	}
+
+	duration = duration.Round(time.Second)
+
+	results := globalNotificationSys.Netperf(ctx, duration)
+	enc := json.NewEncoder(w)
+	if err := enc.Encode(madmin.NetperfResult{NodeResults: results}); err != nil {
+		return
+	}
+}
+
 // SpeedtestHandler - Deprecated. See ObjectSpeedtestHandler
 func (a adminAPIHandlers) SpeedtestHandler(w http.ResponseWriter, r *http.Request) {
 	a.ObjectSpeedtestHandler(w, r)

cmd/admin-router.go

@@ -224,7 +224,7 @@ func registerAdminRouter(router *mux.Router, enableConfigOps bool) {
 		adminRouter.Methods(http.MethodPost).Path(adminVersion + "/speedtest").HandlerFunc(httpTraceHdrs(adminAPI.SpeedtestHandler))
 		adminRouter.Methods(http.MethodPost).Path(adminVersion + "/speedtest/object").HandlerFunc(httpTraceHdrs(adminAPI.ObjectSpeedtestHandler))
 		adminRouter.Methods(http.MethodPost).Path(adminVersion + "/speedtest/drive").HandlerFunc(httpTraceHdrs(adminAPI.DriveSpeedtestHandler))
-		adminRouter.Methods(http.MethodPost).Path(adminVersion + "/speedtest/net").HandlerFunc(httpTraceHdrs(adminAPI.NetSpeedtestHandler))
+		adminRouter.Methods(http.MethodPost).Path(adminVersion + "/speedtest/net").HandlerFunc(httpTraceHdrs(adminAPI.NetperfHandler))
 
 		// HTTP Trace
 		adminRouter.Methods(http.MethodGet).Path(adminVersion + "/trace").HandlerFunc(gz(http.HandlerFunc(adminAPI.TraceHandler)))

cmd/background-newdisks-heal-ops.go

@@ -277,14 +277,6 @@ func initAutoHeal(ctx context.Context, objAPI ObjectLayer) {
 		}
 	}
 
-	if err := bgSeq.healDiskMeta(objAPI); err != nil {
-		if newObjectLayerFn() != nil {
-			// log only in situations, when object layer
-			// has fully initialized.
-			logger.LogIf(bgSeq.ctx, err)
-		}
-	}
-
 	go monitorLocalDisksAndHeal(ctx, z, bgSeq)
 }
 

cmd/bucket-encryption-handlers.go

@@ -206,6 +206,15 @@ func (api objectAPIHandlers) DeleteBucketEncryptionHandler(w http.ResponseWriter
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
 		return
 	}
-
+	// Call site replication hook.
+	//
+	if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
+		Type:      madmin.SRBucketMetaTypeSSEConfig,
+		Bucket:    bucket,
+		SSEConfig: nil,
+	}); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
 	writeSuccessNoContent(w)
 }

cmd/bucket-handlers.go

@@ -1270,6 +1270,17 @@ func (api objectAPIHandlers) DeleteBucketHandler(w http.ResponseWriter, r *http.
 				writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL)
 				return
 			}
+			rcfg, err := getReplicationConfig(ctx, bucket)
+			switch {
+			case err != nil:
+				if _, ok := err.(BucketReplicationConfigNotFound); !ok {
+					writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL)
+					return
+				}
+			case rcfg.HasActiveRules("", true):
+				writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL)
+				return
+			}
 		}
 	}
 

cmd/bucket-replication-utils.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"bytes"
 	"fmt"
+	"net/http"
 	"regexp"
 	"strconv"
 	"strings"
@@ -27,6 +28,7 @@ import (
 	"time"
 
 	"github.com/minio/minio/internal/bucket/replication"
+	xhttp "github.com/minio/minio/internal/http"
 )
 
 //go:generate msgp -file=$GOFILE
@@ -699,3 +701,26 @@ func newBucketResyncStatus(bucket string) BucketReplicationResyncStatus {
 		Version:    resyncMetaVersion,
 	}
 }
+
+var contentRangeRegexp = regexp.MustCompile(`bytes ([0-9]+)-([0-9]+)/([0-9]+|\\*)`)
+
+// parse size from content-range header
+func parseSizeFromContentRange(h http.Header) (sz int64, err error) {
+	cr := h.Get(xhttp.ContentRange)
+	if cr == "" {
+		return sz, fmt.Errorf("Content-Range not set")
+	}
+	parts := contentRangeRegexp.FindStringSubmatch(cr)
+	if len(parts) != 4 {
+		return sz, fmt.Errorf("invalid Content-Range header %s", cr)
+	}
+	if parts[3] == "*" {
+		return -1, nil
+	}
+	var usz uint64
+	usz, err = strconv.ParseUint(parts[3], 10, 64)
+	if err != nil {
+		return sz, err
+	}
+	return int64(usz), nil
+}

cmd/bucket-replication.go

@@ -1531,16 +1531,21 @@ func initBackgroundReplication(ctx context.Context, objectAPI ObjectLayer) {
 	go globalReplicationStats.loadInitialReplicationMetrics(ctx)
 }
 
+type proxyResult struct {
+	Proxy bool
+	Err   error
+}
+
 // get Reader from replication target if active-active replication is in place and
 // this node returns a 404
-func proxyGetToReplicationTarget(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (gr *GetObjectReader, proxy bool) {
-	tgt, oi, proxy := proxyHeadToRepTarget(ctx, bucket, object, opts, proxyTargets)
-	if !proxy {
-		return nil, false
+func proxyGetToReplicationTarget(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, h http.Header, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (gr *GetObjectReader, proxy proxyResult, err error) {
+	tgt, oi, proxy := proxyHeadToRepTarget(ctx, bucket, object, rs, opts, proxyTargets)
+	if !proxy.Proxy {
+		return nil, proxy, nil
 	}
-	fn, off, length, err := NewGetObjectReader(rs, oi, opts)
+	fn, _, _, err := NewGetObjectReader(nil, oi, opts)
 	if err != nil {
-		return nil, false
+		return nil, proxy, err
 	}
 	gopts := miniogo.GetObjectOptions{
 		VersionID:            opts.VersionID,
@@ -1548,30 +1553,40 @@ func proxyGetToReplicationTarget(ctx context.Context, bucket, object string, rs
 		Internal: miniogo.AdvancedGetOptions{
 			ReplicationProxyRequest: "true",
 		},
+		PartNumber: opts.PartNumber,
 	}
 	// get correct offsets for encrypted object
-	if off >= 0 && length >= 0 {
-		if err := gopts.SetRange(off, off+length-1); err != nil {
-			return nil, false
+	if rs != nil {
+		h, err := rs.ToHeader()
+		if err != nil {
+			return nil, proxy, err
 		}
+		gopts.Set(xhttp.Range, h)
 	}
 	// Make sure to match ETag when proxying.
 	if err = gopts.SetMatchETag(oi.ETag); err != nil {
-		return nil, false
+		return nil, proxy, err
 	}
 	c := miniogo.Core{Client: tgt.Client}
-	obj, _, _, err := c.GetObject(ctx, bucket, object, gopts)
+	obj, _, h, err := c.GetObject(ctx, bucket, object, gopts)
 	if err != nil {
-		return nil, false
+		return nil, proxy, err
 	}
 	closeReader := func() { obj.Close() }
-
 	reader, err := fn(obj, h, closeReader)
 	if err != nil {
-		return nil, false
+		return nil, proxy, err
 	}
 	reader.ObjInfo = oi.Clone()
-	return reader, true
+	if rs != nil {
+		contentSize, err := parseSizeFromContentRange(h)
+		if err != nil {
+			return nil, proxy, err
+		}
+		reader.ObjInfo.Size = contentSize
+	}
+
+	return reader, proxyResult{Proxy: true}, nil
 }
 
 func getproxyTargets(ctx context.Context, bucket, object string, opts ObjectOptions) (tgts *madmin.BucketTargets) {
@@ -1590,11 +1605,11 @@ func getproxyTargets(ctx context.Context, bucket, object string, opts ObjectOpti
 	return tgts
 }
 
-func proxyHeadToRepTarget(ctx context.Context, bucket, object string, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (tgt *TargetClient, oi ObjectInfo, proxy bool) {
+func proxyHeadToRepTarget(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (tgt *TargetClient, oi ObjectInfo, proxy proxyResult) {
 	// this option is set when active-active replication is in place between site A -> B,
 	// and site B does not have the object yet.
 	if opts.ProxyRequest || (opts.ProxyHeaderSet && !opts.ProxyRequest) { // true only when site B sets MinIOSourceProxyRequest header
-		return nil, oi, false
+		return nil, oi, proxy
 	}
 	for _, t := range proxyTargets.Targets {
 		tgt = globalBucketTargetSys.GetRemoteTargetClient(ctx, t.Arn)
@@ -1612,9 +1627,22 @@ func proxyHeadToRepTarget(ctx context.Context, bucket, object string, opts Objec
 			Internal: miniogo.AdvancedGetOptions{
 				ReplicationProxyRequest: "true",
 			},
+			PartNumber: opts.PartNumber,
 		}
+		if rs != nil {
+			h, err := rs.ToHeader()
+			if err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Invalid range header for %s/%s(%s) - %w", bucket, object, opts.VersionID, err))
+				continue
+			}
+			gopts.Set(xhttp.Range, h)
+		}
+
 		objInfo, err := tgt.StatObject(ctx, t.TargetBucket, object, gopts)
 		if err != nil {
+			if isErrInvalidRange(ErrorRespToObjectError(err, bucket, object)) {
+				return nil, oi, proxyResult{Err: err}
+			}
 			continue
 		}
 
@@ -1645,15 +1673,15 @@ func proxyHeadToRepTarget(ctx context.Context, bucket, object string, opts Objec
 		if ok {
 			oi.ContentEncoding = ce
 		}
-		return tgt, oi, true
+		return tgt, oi, proxyResult{Proxy: true}
 	}
-	return nil, oi, false
+	return nil, oi, proxy
 }
 
 // get object info from replication target if active-active replication is in place and
 // this node returns a 404
-func proxyHeadToReplicationTarget(ctx context.Context, bucket, object string, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (oi ObjectInfo, proxy bool) {
-	_, oi, proxy = proxyHeadToRepTarget(ctx, bucket, object, opts, proxyTargets)
+func proxyHeadToReplicationTarget(ctx context.Context, bucket, object string, rs *HTTPRangeSpec, opts ObjectOptions, proxyTargets *madmin.BucketTargets) (oi ObjectInfo, proxy proxyResult) {
+	_, oi, proxy = proxyHeadToRepTarget(ctx, bucket, object, rs, opts, proxyTargets)
 	return oi, proxy
 }
 

cmd/data-scanner.go

@@ -403,7 +403,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 		if filter != nil && ok && existing.Compacted {
 			// If folder isn't in filter and we have data, skip it completely.
 			if folder.name != dataUsageRoot && !filter.containsDir(folder.name) {
-				if f.healObjectSelect == 0 || !thisHash.mod(f.oldCache.Info.NextCycle, f.healFolderInclude/folder.objectHealProbDiv) {
+				if f.healObjectSelect == 0 || !thisHash.modAlt(f.oldCache.Info.NextCycle/folder.objectHealProbDiv, f.healFolderInclude/folder.objectHealProbDiv) {
 					f.newCache.copyWithChildren(&f.oldCache, thisHash, folder.parent)
 					f.updateCache.copyWithChildren(&f.oldCache, thisHash, folder.parent)
 					if f.dataUsageScannerDebug {
@@ -482,7 +482,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 				debug:       f.dataUsageScannerDebug,
 				lifeCycle:   activeLifeCycle,
 				replication: replicationCfg,
-				heal:        thisHash.mod(f.oldCache.Info.NextCycle, f.healObjectSelect/folder.objectHealProbDiv) && globalIsErasure,
+				heal:        thisHash.modAlt(f.oldCache.Info.NextCycle/folder.objectHealProbDiv, f.healObjectSelect/folder.objectHealProbDiv) && globalIsErasure,
 			}
 			// if the drive belongs to an erasure set
 			// that is already being healed, skip the
@@ -609,13 +609,13 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
 			// and the entry itself is compacted.
 			if !into.Compacted && f.oldCache.isCompacted(h) {
 				if !h.mod(f.oldCache.Info.NextCycle, dataUsageUpdateDirCycles) {
-					if f.healObjectSelect == 0 || !h.mod(f.oldCache.Info.NextCycle, f.healFolderInclude/folder.objectHealProbDiv) {
+					if f.healObjectSelect == 0 || !h.modAlt(f.oldCache.Info.NextCycle/folder.objectHealProbDiv, f.healFolderInclude/folder.objectHealProbDiv) {
 						// Transfer and add as child...
 						f.newCache.copyWithChildren(&f.oldCache, h, folder.parent)
 						into.addChild(h)
 						continue
 					}
-					folder.objectHealProbDiv = dataUsageUpdateDirCycles
+					folder.objectHealProbDiv = f.healFolderInclude
 				}
 			}
 			scanFolder(folder)

cmd/data-usage-cache.go

@@ -369,6 +369,17 @@ func (h dataUsageHash) mod(cycle uint32, cycles uint32) bool {
 	return uint32(xxhash.Sum64String(string(h)))%cycles == cycle%cycles
 }
 
+// modAlt returns true if the hash mod cycles == cycle.
+// This is out of sync with mod.
+// If cycles is 0 false is always returned.
+// If cycles is 1 true is always returned (as expected).
+func (h dataUsageHash) modAlt(cycle uint32, cycles uint32) bool {
+	if cycles <= 1 {
+		return cycles == 1
+	}
+	return uint32(xxhash.Sum64String(string(h))>>32)%(cycles) == cycle%cycles
+}
+
 // addChild will add a child based on its hash.
 // If it already exists it will not be added again.
 func (e *dataUsageEntry) addChild(hash dataUsageHash) {

cmd/data-usage-utils.go

@@ -92,32 +92,37 @@ type DataUsageInfo struct {
 }
 
 func (dui DataUsageInfo) tierStats() []madmin.TierInfo {
-	if globalTierConfigMgr.Empty() {
+	if dui.TierStats == nil {
 		return nil
 	}
 
-	ts := make(map[string]madmin.TierStats)
-	// Add configured remote tiers
-	for tier := range globalTierConfigMgr.Tiers {
-		ts[tier] = madmin.TierStats{}
+	cfgs := globalTierConfigMgr.ListTiers()
+	if len(cfgs) == 0 {
+		return nil
 	}
+
+	ts := make(map[string]madmin.TierStats, len(cfgs)+1)
+	infos := make([]madmin.TierInfo, 0, len(ts))
+
 	// Add STANDARD (hot-tier)
 	ts[minioHotTier] = madmin.TierStats{}
+	infos = append(infos, madmin.TierInfo{
+		Name: minioHotTier,
+		Type: "internal",
+	})
+	// Add configured remote tiers
+	for _, cfg := range cfgs {
+		ts[cfg.Name] = madmin.TierStats{}
+		infos = append(infos, madmin.TierInfo{
+			Name: cfg.Name,
+			Type: cfg.Type.String(),
+		})
+	}
 
 	ts = dui.TierStats.adminStats(ts)
-	infos := make([]madmin.TierInfo, 0, len(ts))
-	for tier, st := range ts {
-		var tierType string
-		if tier == minioHotTier {
-			tierType = "internal"
-		} else {
-			tierType = globalTierConfigMgr.Tiers[tier].Type.String()
-		}
-		infos = append(infos, madmin.TierInfo{
-			Name:  tier,
-			Type:  tierType,
-			Stats: st,
-		})
+	for i := range infos {
+		info := infos[i]
+		infos[i].Stats = ts[info.Name]
 	}
 
 	sort.Slice(infos, func(i, j int) bool {
@@ -131,3 +136,31 @@ func (dui DataUsageInfo) tierStats() []madmin.TierInfo {
 	})
 	return infos
 }
+
+func (dui DataUsageInfo) tierMetrics() (metrics []Metric) {
+	if dui.TierStats == nil {
+		return nil
+	}
+	// e.g minio_cluster_ilm_transitioned_bytes{tier="S3TIER-1"}=136314880
+	//     minio_cluster_ilm_transitioned_objects{tier="S3TIER-1"}=1
+	//     minio_cluster_ilm_transitioned_versions{tier="S3TIER-1"}=3
+	for tier, st := range dui.TierStats.Tiers {
+		metrics = append(metrics, Metric{
+			Description:    getClusterTransitionedBytesMD(),
+			Value:          float64(st.TotalSize),
+			VariableLabels: map[string]string{"tier": tier},
+		})
+		metrics = append(metrics, Metric{
+			Description:    getClusterTransitionedObjectsMD(),
+			Value:          float64(st.NumObjects),
+			VariableLabels: map[string]string{"tier": tier},
+		})
+		metrics = append(metrics, Metric{
+			Description:    getClusterTransitionedVersionsMD(),
+			Value:          float64(st.NumVersions),
+			VariableLabels: map[string]string{"tier": tier},
+		})
+	}
+
+	return metrics
+}

cmd/disk-cache.go

@@ -31,6 +31,7 @@ import (
 	objectlock "github.com/minio/minio/internal/bucket/object/lock"
 	"github.com/minio/minio/internal/color"
 	"github.com/minio/minio/internal/config/cache"
+	"github.com/minio/minio/internal/disk"
 	"github.com/minio/minio/internal/hash"
 	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
@@ -594,12 +595,23 @@ func newCache(config cache.Config) ([]*diskCache, bool, error) {
 	if err != nil {
 		return nil, false, err
 	}
+	var warningMsg string
 	for i, dir := range config.Drives {
 		// skip diskCache creation for cache drives missing a format.json
 		if formats[i] == nil {
 			caches = append(caches, nil)
 			continue
 		}
+		if !globalIsCICD && len(warningMsg) == 0 {
+			rootDsk, err := disk.IsRootDisk(dir, "/")
+			if err != nil {
+				warningMsg = fmt.Sprintf("Invalid cache dir %s err : %s", dir, err.Error())
+			}
+			if rootDsk {
+				warningMsg = fmt.Sprintf("cache dir cannot be part of root disk: %s", dir)
+			}
+		}
+
 		if err := checkAtimeSupport(dir); err != nil {
 			return nil, false, fmt.Errorf("Atime support required for disk caching, atime check failed with %w", err)
 		}
@@ -610,6 +622,9 @@ func newCache(config cache.Config) ([]*diskCache, bool, error) {
 		}
 		caches = append(caches, cache)
 	}
+	if warningMsg != "" {
+		logger.Info(color.Yellow(fmt.Sprintf("WARNING: Usage of root disk for disk caching is deprecated: %s", warningMsg)))
+	}
 	return caches, migrating, nil
 }
 

cmd/erasure-healing_test.go

@@ -382,6 +382,13 @@ func TestHealingDanglingObject(t *testing.T) {
 		copy(disks, newDisks)
 		objLayer.(*erasureServerPools).serverPools[0].erasureDisksMu.Unlock()
 	}
+	getDisk := func(n int) StorageAPI {
+		objLayer.(*erasureServerPools).serverPools[0].erasureDisksMu.Lock()
+		disk := disks[n]
+		objLayer.(*erasureServerPools).serverPools[0].erasureDisksMu.Unlock()
+		return disk
+	}
+
 	// Remove 4 disks.
 	setDisks(nil, nil, nil, nil)
 
@@ -440,8 +447,8 @@ func TestHealingDanglingObject(t *testing.T) {
 	}
 
 	setDisks(orgDisks[:4]...)
-
-	fileInfoPreHeal, err = disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	disk := getDisk(0)
+	fileInfoPreHeal, err = disk.ReadVersion(context.Background(), bucket, object, "", false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -458,7 +465,8 @@ func TestHealingDanglingObject(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	fileInfoPostHeal, err = disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	disk = getDisk(0)
+	fileInfoPostHeal, err = disk.ReadVersion(context.Background(), bucket, object, "", false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -488,7 +496,8 @@ func TestHealingDanglingObject(t *testing.T) {
 
 	setDisks(orgDisks[:4]...)
 
-	fileInfoPreHeal, err = disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	disk = getDisk(0)
+	fileInfoPreHeal, err = disk.ReadVersion(context.Background(), bucket, object, "", false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -505,7 +514,8 @@ func TestHealingDanglingObject(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	fileInfoPostHeal, err = disks[0].ReadVersion(context.Background(), bucket, object, "", false)
+	disk = getDisk(0)
+	fileInfoPostHeal, err = disk.ReadVersion(context.Background(), bucket, object, "", false)
 	if err != nil {
 		t.Fatal(err)
 	}

cmd/erasure-server-pool-decom.go

@@ -593,7 +593,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 				return
 			}
 
-			// We need a reversed order for Decommissioning,
+			// We need a reversed order for decommissioning,
 			// to create the appropriate stack.
 			versionsSorter(fivs.Versions).reverse()
 
@@ -955,6 +955,18 @@ func (z *erasureServerPools) StartDecommission(ctx context.Context, idx int) (er
 		}
 	}
 
+	// Create .minio.sys/conifg, .minio.sys/buckets paths if missing,
+	// this code is present to avoid any missing meta buckets on other
+	// pools.
+	for _, metaBucket := range []string{
+		pathJoin(minioMetaBucket, minioConfigPrefix),
+		pathJoin(minioMetaBucket, bucketMetaPrefix),
+	} {
+		if err = z.MakeBucketWithLocation(ctx, metaBucket, BucketOptions{}); err != nil {
+			return err
+		}
+	}
+
 	// Buckets data are dispersed in multiple zones/sets, make
 	// sure to decommission the necessary metadata.
 	buckets = append(buckets, BucketInfo{

cmd/erasure-server-pool.go

@@ -661,6 +661,9 @@ func (z *erasureServerPools) MakeBucketWithLocation(ctx context.Context, bucket
 	for index := range z.serverPools {
 		index := index
 		g.Go(func() error {
+			if z.IsSuspended(index) {
+				return nil
+			}
 			return z.serverPools[index].MakeBucketWithLocation(ctx, bucket, opts)
 		}, index)
 	}
@@ -672,7 +675,8 @@ func (z *erasureServerPools) MakeBucketWithLocation(ctx context.Context, bucket
 			if _, ok := err.(BucketExists); !ok {
 				// Delete created buckets, ignoring errors.
 				z.DeleteBucket(context.Background(), bucket, DeleteBucketOptions{
-					Force: false, NoRecreate: true,
+					Force:      false,
+					NoRecreate: true,
 				})
 			}
 			return err
@@ -1478,6 +1482,9 @@ func (z *erasureServerPools) DeleteBucket(ctx context.Context, bucket string, op
 	for index := range z.serverPools {
 		index := index
 		g.Go(func() error {
+			if z.IsSuspended(index) {
+				return nil
+			}
 			return z.serverPools[index].DeleteBucket(ctx, bucket, opts)
 		}, index)
 	}

cmd/globals.go

@@ -345,6 +345,11 @@ var (
 	globalIsCICD bool
 
 	globalRootDiskThreshold uint64
+
+	// Used for collecting stats for netperf
+	globalNetPerfMinDuration = time.Second * 10
+	globalNetPerfRX          netPerfRX
+
 	// Add new variable global values here.
 )
 

cmd/httprange.go

@@ -174,3 +174,26 @@ func (h *HTTPRangeSpec) String(resourceSize int64) string {
 	}
 	return fmt.Sprintf("%d-%d", off, off+length-1)
 }
+
+// ToHeader returns the Range header value.
+func (h *HTTPRangeSpec) ToHeader() (string, error) {
+	if h == nil {
+		return "", nil
+	}
+	start := strconv.Itoa(int(h.Start))
+	end := strconv.Itoa(int(h.End))
+	switch {
+	case h.Start >= 0 && h.End >= 0:
+		if h.Start > h.End {
+			return "", errInvalidRange
+		}
+	case h.IsSuffixLength:
+		end = strconv.Itoa(int(h.Start * -1))
+		start = ""
+	case h.Start > -1:
+		end = ""
+	default:
+		return "", fmt.Errorf("does not have valid range value")
+	}
+	return fmt.Sprintf("bytes=%s-%s", start, end), nil
+}

cmd/httprange_test.go

@@ -101,3 +101,46 @@ func TestHTTPRequestRangeSpec(t *testing.T) {
 		t.Errorf("Case %d: Expected errInvalidRange but: %v %v %d %d %v", i, rs, err1, o, l, err2)
 	}
 }
+
+func TestHTTPRequestRangeToHeader(t *testing.T) {
+	validRangeSpecs := []struct {
+		spec        string
+		errExpected bool
+	}{
+		{"bytes=0-", false},
+		{"bytes=1-", false},
+
+		{"bytes=0-9", false},
+		{"bytes=1-10", false},
+		{"bytes=1-1", false},
+		{"bytes=2-5", false},
+
+		{"bytes=-5", false},
+		{"bytes=-1", false},
+		{"bytes=-1000", false},
+		{"bytes=", true},
+		{"bytes= ", true},
+		{"byte=", true},
+		{"bytes=A-B", true},
+		{"bytes=1-B", true},
+		{"bytes=B-1", true},
+		{"bytes=-1-1", true},
+	}
+	for i, testCase := range validRangeSpecs {
+		rs, err := parseRequestRangeSpec(testCase.spec)
+		if err != nil {
+			if !testCase.errExpected || err == nil && testCase.errExpected {
+				t.Errorf("unexpected err: %v", err)
+			}
+			continue
+		}
+		h, err := rs.ToHeader()
+		if err != nil && !testCase.errExpected || err == nil && testCase.errExpected {
+			t.Errorf("expected error with invalid range: %v", err)
+		}
+		if h != testCase.spec {
+			t.Errorf("Case %d: translated to incorrect header: %s expected: %s",
+				i, h, testCase.spec)
+		}
+	}
+}

cmd/metacache-entries.go

@@ -389,11 +389,11 @@ func (m metaCacheEntries) resolve(r *metadataResolutionParams) (selected *metaCa
 
 // firstFound returns the first found and the number of set entries.
 func (m metaCacheEntries) firstFound() (first *metaCacheEntry, n int) {
-	for _, entry := range m {
+	for i, entry := range m {
 		if entry.name != "" {
 			n++
 			if first == nil {
-				first = &entry
+				first = &m[i]
 			}
 		}
 	}

cmd/metacache-walk.go

@@ -138,6 +138,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 			// Forward some errors?
 			return nil
 		}
+		diskHealthCheckOK(ctx, err)
 		if len(entries) == 0 {
 			return nil
 		}
@@ -179,6 +180,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 				s.walkReadMu.Lock()
 				meta.metadata, err = s.readMetadata(ctx, pathJoin(volumeDir, current, entry))
 				s.walkReadMu.Unlock()
+				diskHealthCheckOK(ctx, err)
 				if err != nil {
 					logger.LogIf(ctx, err)
 					continue
@@ -196,6 +198,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 				s.walkReadMu.Lock()
 				meta.metadata, err = xioutil.ReadFile(pathJoin(volumeDir, current, entry))
 				s.walkReadMu.Unlock()
+				diskHealthCheckOK(ctx, err)
 				if err != nil {
 					logger.LogIf(ctx, err)
 					continue
@@ -257,6 +260,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 			s.walkReadMu.Lock()
 			meta.metadata, err = s.readMetadata(ctx, pathJoin(volumeDir, meta.name, xlStorageFormatFile))
 			s.walkReadMu.Unlock()
+			diskHealthCheckOK(ctx, err)
 			switch {
 			case err == nil:
 				// It was an object
@@ -266,6 +270,7 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 				out <- meta
 			case osIsNotExist(err), isSysErrIsDir(err):
 				meta.metadata, err = xioutil.ReadFile(pathJoin(volumeDir, meta.name, xlStorageFormatFileV1))
+				diskHealthCheckOK(ctx, err)
 				if err == nil {
 					// It was an object
 					out <- meta
@@ -303,16 +308,12 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
 	return scanDir(opts.BaseDir)
 }
 
-func (p *xlStorageDiskIDCheck) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writer) error {
-	defer p.updateStorageMetrics(storageMetricWalkDir, opts.Bucket, opts.BaseDir)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writer) (err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricWalkDir, opts.Bucket, opts.BaseDir)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.WalkDir(ctx, opts, wr)
 }

cmd/metrics-v2.go

@@ -1663,33 +1663,12 @@ func getClusterTierMetrics() *MetricsGroup {
 		if err != nil {
 			return
 		}
-		// data usage has not captured any data yet.
-		if dui.LastUpdate.IsZero() {
+		// data usage has not captured any tier stats yet.
+		if dui.TierStats == nil {
 			return
 		}
 
-		// e.g minio_cluster_ilm_transitioned_bytes{tier="S3TIER-1"}=136314880
-		//     minio_cluster_ilm_transitioned_objects{tier="S3TIER-1"}=1
-		//     minio_cluster_ilm_transitioned_versions{tier="S3TIER-1"}=3
-		for tier, st := range dui.TierStats.Tiers {
-			metrics = append(metrics, Metric{
-				Description:    getClusterTransitionedBytesMD(),
-				Value:          float64(st.TotalSize),
-				VariableLabels: map[string]string{"tier": tier},
-			})
-			metrics = append(metrics, Metric{
-				Description:    getClusterTransitionedObjectsMD(),
-				Value:          float64(st.NumObjects),
-				VariableLabels: map[string]string{"tier": tier},
-			})
-			metrics = append(metrics, Metric{
-				Description:    getClusterTransitionedVersionsMD(),
-				Value:          float64(st.NumVersions),
-				VariableLabels: map[string]string{"tier": tier},
-			})
-		}
-
-		return metrics
+		return dui.tierMetrics()
 	})
 	return mg
 }

cmd/notification.go

@@ -1553,6 +1553,58 @@ func (sys *NotificationSys) ServiceFreeze(ctx context.Context, freeze bool) []No
 	return nerrs
 }
 
+// Netperf - perform mesh style network throughput test
+func (sys *NotificationSys) Netperf(ctx context.Context, duration time.Duration) []madmin.NetperfNodeResult {
+	length := len(sys.allPeerClients)
+	if length == 0 {
+		// For single node erasure setup.
+		return nil
+	}
+	results := make([]madmin.NetperfNodeResult, length)
+
+	scheme := "http"
+	if globalIsTLS {
+		scheme = "https"
+	}
+
+	var wg sync.WaitGroup
+	for index := range sys.peerClients {
+		if sys.peerClients[index] == nil {
+			continue
+		}
+		wg.Add(1)
+		go func(index int) {
+			defer wg.Done()
+			r, err := sys.peerClients[index].Netperf(ctx, duration)
+			u := &url.URL{
+				Scheme: scheme,
+				Host:   sys.peerClients[index].host.String(),
+			}
+			if err != nil {
+				results[index].Error = err.Error()
+			} else {
+				results[index] = r
+			}
+			results[index].Endpoint = u.String()
+		}(index)
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		r := netperf(ctx, duration)
+		u := &url.URL{
+			Scheme: scheme,
+			Host:   globalLocalNodeName,
+		}
+		results[len(results)-1] = r
+		results[len(results)-1].Endpoint = u.String()
+	}()
+	wg.Wait()
+
+	return results
+}
+
 // Speedtest run GET/PUT tests at input concurrency for requested object size,
 // optionally you can extend the tests longer with time.Duration.
 func (sys *NotificationSys) Speedtest(ctx context.Context, size int,

cmd/object-api-errors.go

@@ -701,3 +701,8 @@ func isErrMethodNotAllowed(err error) bool {
 	var methodNotAllowed MethodNotAllowed
 	return errors.As(err, &methodNotAllowed)
 }
+
+func isErrInvalidRange(err error) bool {
+	_, ok := err.(InvalidRange)
+	return ok
+}

cmd/object-handlers.go

@@ -412,20 +412,28 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 	if err != nil {
 		var (
 			reader *GetObjectReader
-			proxy  bool
+			proxy  proxyResult
 		)
 		proxytgts := getproxyTargets(ctx, bucket, object, opts)
 		if !proxytgts.Empty() {
 			// proxy to replication target if active-active replication is in place.
-			reader, proxy = proxyGetToReplicationTarget(ctx, bucket, object, rs, r.Header, opts, proxytgts)
-			if reader != nil && proxy {
+			reader, proxy, err = proxyGetToReplicationTarget(ctx, bucket, object, rs, r.Header, opts, proxytgts)
+			if err != nil && !isErrObjectNotFound(ErrorRespToObjectError(err, bucket, object)) &&
+				!isErrVersionNotFound(ErrorRespToObjectError(err, bucket, object)) {
+				logger.LogIf(ctx, fmt.Errorf("Replication proxy failed for %s/%s(%s) - %w", bucket, object, opts.VersionID, err))
+			}
+			if reader != nil && proxy.Proxy && err == nil {
 				gr = reader
 			}
 		}
-		if reader == nil || !proxy {
+		if reader == nil || !proxy.Proxy {
 			if isErrPreconditionFailed(err) {
 				return
 			}
+			if proxy.Err != nil {
+				writeErrorResponse(ctx, w, toAPIError(ctx, proxy.Err), r.URL)
+				return
+			}
 			if globalBucketVersioningSys.Enabled(bucket) && gr != nil {
 				if !gr.ObjInfo.VersionPurgeStatus.Empty() {
 					// Shows the replication status of a permanent delete of a version
@@ -631,22 +639,28 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 		writeErrorResponseHeadersOnly(w, errorCodes.ToAPIErr(s3Error))
 		return
 	}
+	// Get request range.
+	var rs *HTTPRangeSpec
+	rangeHeader := r.Header.Get(xhttp.Range)
 
 	objInfo, err := getObjectInfo(ctx, bucket, object, opts)
 	if err != nil {
 		var (
-			proxy bool
+			proxy proxyResult
 			oi    ObjectInfo
 		)
 		// proxy HEAD to replication target if active-active replication configured on bucket
 		proxytgts := getproxyTargets(ctx, bucket, object, opts)
 		if !proxytgts.Empty() {
-			oi, proxy = proxyHeadToReplicationTarget(ctx, bucket, object, opts, proxytgts)
-			if proxy {
+			if rangeHeader != "" {
+				rs, _ = parseRequestRangeSpec(rangeHeader)
+			}
+			oi, proxy = proxyHeadToReplicationTarget(ctx, bucket, object, rs, opts, proxytgts)
+			if proxy.Proxy {
 				objInfo = oi
 			}
 		}
-		if !proxy {
+		if !proxy.Proxy {
 			if globalBucketVersioningSys.Enabled(bucket) {
 				switch {
 				case !objInfo.VersionPurgeStatus.Empty():
@@ -703,9 +717,6 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 		return
 	}
 
-	// Get request range.
-	var rs *HTTPRangeSpec
-	rangeHeader := r.Header.Get(xhttp.Range)
 	if rangeHeader != "" {
 		// Both 'Range' and 'partNumber' cannot be specified at the same time
 		if opts.PartNumber > 0 {

cmd/peer-rest-client.go

@@ -1119,3 +1119,27 @@ func (client *peerRESTClient) GetLastDayTierStats(ctx context.Context) (dailyAll
 	}
 	return dailyAllTierStats(result), nil
 }
+
+// DevNull - Used by netperf to pump data to peer
+func (client *peerRESTClient) DevNull(ctx context.Context, r io.Reader) error {
+	respBody, err := client.callWithContext(ctx, peerRESTMethodDevNull, nil, r, -1)
+	if err != nil {
+		return err
+	}
+	defer http.DrainBody(respBody)
+	return err
+}
+
+// Netperf - To initiate netperf on peer
+func (client *peerRESTClient) Netperf(ctx context.Context, duration time.Duration) (madmin.NetperfNodeResult, error) {
+	var result madmin.NetperfNodeResult
+	values := make(url.Values)
+	values.Set(peerRESTDuration, duration.String())
+	respBody, err := client.callWithContext(context.Background(), peerRESTMethodNetperf, values, nil, -1)
+	if err != nil {
+		return result, err
+	}
+	defer http.DrainBody(respBody)
+	err = gob.NewDecoder(respBody).Decode(&result)
+	return result, err
+}

cmd/peer-rest-common.go

@@ -18,7 +18,7 @@
 package cmd
 
 const (
-	peerRESTVersion       = "v20" // Add drivespeedtest
+	peerRESTVersion       = "v21" // Add netperf
 	peerRESTVersionPrefix = SlashSeparator + peerRESTVersion
 	peerRESTPrefix        = minioReservedBucketPath + "/peer"
 	peerRESTPath          = peerRESTPrefix + peerRESTVersionPrefix
@@ -70,6 +70,8 @@ const (
 	peerRESTMethodReloadSiteReplicationConfig = "/reloadsitereplicationconfig"
 	peerRESTMethodReloadPoolMeta              = "/reloadpoolmeta"
 	peerRESTMethodGetLastDayTierStats         = "/getlastdaytierstats"
+	peerRESTMethodDevNull                     = "/devnull"
+	peerRESTMethodNetperf                     = "/netperf"
 )
 
 const (

cmd/peer-rest-server.go

@@ -27,20 +27,15 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/dustin/go-humanize"
-	"github.com/google/uuid"
 	"github.com/gorilla/mux"
 	"github.com/minio/madmin-go"
 	b "github.com/minio/minio/internal/bucket/bandwidth"
 	"github.com/minio/minio/internal/event"
-	"github.com/minio/minio/internal/hash"
-	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
-	"github.com/minio/pkg/randreader"
 	"github.com/tinylib/msgp/msgp"
 )
 
@@ -1143,148 +1138,6 @@ func (s *peerRESTServer) GetPeerMetrics(w http.ResponseWriter, r *http.Request)
 	}
 }
 
-// SpeedtestResult return value of the speedtest function
-type SpeedtestResult struct {
-	Endpoint  string
-	Uploads   uint64
-	Downloads uint64
-	Error     string
-}
-
-func newRandomReader(size int) io.Reader {
-	return io.LimitReader(randreader.New(), int64(size))
-}
-
-// Runs the speedtest on local MinIO process.
-func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Duration, storageClass string) (SpeedtestResult, error) {
-	objAPI := newObjectLayerFn()
-	if objAPI == nil {
-		return SpeedtestResult{}, errServerNotInitialized
-	}
-
-	var errOnce sync.Once
-	var retError string
-	var wg sync.WaitGroup
-	var totalBytesWritten uint64
-	var totalBytesRead uint64
-
-	objCountPerThread := make([]uint64, concurrent)
-
-	uploadsCtx, uploadsCancel := context.WithCancel(context.Background())
-	defer uploadsCancel()
-
-	go func() {
-		time.Sleep(duration)
-		uploadsCancel()
-	}()
-
-	objNamePrefix := "speedtest/objects/" + uuid.New().String()
-
-	wg.Add(concurrent)
-	for i := 0; i < concurrent; i++ {
-		go func(i int) {
-			defer wg.Done()
-			for {
-				hashReader, err := hash.NewReader(newRandomReader(size),
-					int64(size), "", "", int64(size))
-				if err != nil {
-					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
-						errOnce.Do(func() {
-							retError = err.Error()
-						})
-					}
-					uploadsCancel()
-					return
-				}
-				reader := NewPutObjReader(hashReader)
-				objInfo, err := objAPI.PutObject(uploadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d",
-					objNamePrefix, i, objCountPerThread[i]), reader, ObjectOptions{
-					UserDefined: map[string]string{
-						xhttp.AmzStorageClass: storageClass,
-					},
-					Speedtest: true,
-				})
-				if err != nil {
-					objCountPerThread[i]--
-					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
-						errOnce.Do(func() {
-							retError = err.Error()
-						})
-					}
-					uploadsCancel()
-					return
-				}
-				atomic.AddUint64(&totalBytesWritten, uint64(objInfo.Size))
-				objCountPerThread[i]++
-			}
-		}(i)
-	}
-	wg.Wait()
-
-	// We already saw write failures, no need to proceed into read's
-	if retError != "" {
-		return SpeedtestResult{Uploads: totalBytesWritten, Downloads: totalBytesRead, Error: retError}, nil
-	}
-
-	downloadsCtx, downloadsCancel := context.WithCancel(context.Background())
-	defer downloadsCancel()
-	go func() {
-		time.Sleep(duration)
-		downloadsCancel()
-	}()
-
-	wg.Add(concurrent)
-	for i := 0; i < concurrent; i++ {
-		go func(i int) {
-			defer wg.Done()
-			var j uint64
-			if objCountPerThread[i] == 0 {
-				return
-			}
-			for {
-				if objCountPerThread[i] == j {
-					j = 0
-				}
-				r, err := objAPI.GetObjectNInfo(downloadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d",
-					objNamePrefix, i, j), nil, nil, noLock, ObjectOptions{})
-				if err != nil {
-					if isErrObjectNotFound(err) {
-						continue
-					}
-					if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) {
-						errOnce.Do(func() {
-							retError = err.Error()
-						})
-					}
-					downloadsCancel()
-					return
-				}
-				n, err := io.Copy(ioutil.Discard, r)
-				r.Close()
-				if err == nil {
-					// Only capture success criteria - do not
-					// have to capture failed reads, truncated
-					// reads etc.
-					atomic.AddUint64(&totalBytesRead, uint64(n))
-				}
-				if err != nil {
-					if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) {
-						errOnce.Do(func() {
-							retError = err.Error()
-						})
-					}
-					downloadsCancel()
-					return
-				}
-				j++
-			}
-		}(i)
-	}
-	wg.Wait()
-
-	return SpeedtestResult{Uploads: totalBytesWritten, Downloads: totalBytesRead, Error: retError}, nil
-}
-
 func (s *peerRESTServer) SpeedtestHandler(w http.ResponseWriter, r *http.Request) {
 	if !s.IsValid(w, r) {
 		s.writeErrorResponse(w, errors.New("invalid request"))
@@ -1384,6 +1237,50 @@ func (s *peerRESTServer) DriveSpeedTestHandler(w http.ResponseWriter, r *http.Re
 	logger.LogIf(r.Context(), gob.NewEncoder(w).Encode(result))
 }
 
+// DevNull - everything goes to io.Discard
+func (s *peerRESTServer) DevNull(w http.ResponseWriter, r *http.Request) {
+	if !s.IsValid(w, r) {
+		s.writeErrorResponse(w, errors.New("invalid request"))
+		return
+	}
+
+	globalNetPerfRX.Connect()
+	defer globalNetPerfRX.Disconnect()
+
+	connectTime := time.Now()
+	ctx := newContext(r, w, "DevNull")
+	for {
+		n, err := io.CopyN(io.Discard, r.Body, 128*humanize.KiByte)
+		atomic.AddUint64(&globalNetPerfRX.RX, uint64(n))
+		if err != nil && err != io.EOF {
+			// If there is a disconnection before globalNetPerfMinDuration (we give a margin of error of 1 sec)
+			// would mean the network is not stable. Logging here will help in debugging network issues.
+			if time.Since(connectTime) < (globalNetPerfMinDuration - time.Second) {
+				logger.LogIf(ctx, err)
+			}
+		}
+		if err != nil {
+			break
+		}
+	}
+}
+
+// Netperf - perform netperf
+func (s *peerRESTServer) Netperf(w http.ResponseWriter, r *http.Request) {
+	if !s.IsValid(w, r) {
+		s.writeErrorResponse(w, errors.New("invalid request"))
+		return
+	}
+
+	durationStr := r.Form.Get(peerRESTDuration)
+	duration, err := time.ParseDuration(durationStr)
+	if err != nil || duration.Seconds() == 0 {
+		duration = time.Second * 10
+	}
+	result := netperf(r.Context(), duration.Round(time.Second))
+	logger.LogIf(r.Context(), gob.NewEncoder(w).Encode(result))
+}
+
 // registerPeerRESTHandlers - register peer rest router.
 func registerPeerRESTHandlers(router *mux.Router) {
 	server := &peerRESTServer{}
@@ -1431,6 +1328,8 @@ func registerPeerRESTHandlers(router *mux.Router) {
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodLoadTransitionTierConfig).HandlerFunc(httpTraceHdrs(server.LoadTransitionTierConfigHandler))
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodSpeedtest).HandlerFunc(httpTraceHdrs(server.SpeedtestHandler))
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodDriveSpeedTest).HandlerFunc(httpTraceHdrs(server.DriveSpeedTestHandler))
+	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodNetperf).HandlerFunc(httpTraceHdrs(server.Netperf))
+	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodDevNull).HandlerFunc(httpTraceHdrs(server.DevNull))
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodReloadSiteReplicationConfig).HandlerFunc(httpTraceHdrs(server.ReloadSiteReplicationConfigHandler))
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodReloadPoolMeta).HandlerFunc(httpTraceHdrs(server.ReloadPoolMetaHandler))
 	subrouter.Methods(http.MethodPost).Path(peerRESTVersionPrefix + peerRESTMethodGetLastDayTierStats).HandlerFunc(httpTraceHdrs(server.GetLastDayTierStatsHandler))

cmd/perf-tests.go

@@ -0,0 +1,295 @@
+// Copyright (c) 2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/dustin/go-humanize"
+	"github.com/google/uuid"
+	"github.com/minio/madmin-go"
+	"github.com/minio/minio/internal/hash"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/pkg/randreader"
+)
+
+// SpeedtestResult return value of the speedtest function
+type SpeedtestResult struct {
+	Endpoint  string
+	Uploads   uint64
+	Downloads uint64
+	Error     string
+}
+
+func newRandomReader(size int) io.Reader {
+	return io.LimitReader(randreader.New(), int64(size))
+}
+
+// Runs the speedtest on local MinIO process.
+func selfSpeedtest(ctx context.Context, size, concurrent int, duration time.Duration, storageClass string) (SpeedtestResult, error) {
+	objAPI := newObjectLayerFn()
+	if objAPI == nil {
+		return SpeedtestResult{}, errServerNotInitialized
+	}
+
+	var errOnce sync.Once
+	var retError string
+	var wg sync.WaitGroup
+	var totalBytesWritten uint64
+	var totalBytesRead uint64
+
+	objCountPerThread := make([]uint64, concurrent)
+
+	uploadsCtx, uploadsCancel := context.WithCancel(context.Background())
+	defer uploadsCancel()
+
+	go func() {
+		time.Sleep(duration)
+		uploadsCancel()
+	}()
+
+	objNamePrefix := "speedtest/objects/" + uuid.New().String()
+
+	wg.Add(concurrent)
+	for i := 0; i < concurrent; i++ {
+		go func(i int) {
+			defer wg.Done()
+			for {
+				hashReader, err := hash.NewReader(newRandomReader(size),
+					int64(size), "", "", int64(size))
+				if err != nil {
+					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
+						errOnce.Do(func() {
+							retError = err.Error()
+						})
+					}
+					uploadsCancel()
+					return
+				}
+				reader := NewPutObjReader(hashReader)
+				objInfo, err := objAPI.PutObject(uploadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d",
+					objNamePrefix, i, objCountPerThread[i]), reader, ObjectOptions{
+					UserDefined: map[string]string{
+						xhttp.AmzStorageClass: storageClass,
+					},
+					Speedtest: true,
+				})
+				if err != nil {
+					objCountPerThread[i]--
+					if !contextCanceled(uploadsCtx) && !errors.Is(err, context.Canceled) {
+						errOnce.Do(func() {
+							retError = err.Error()
+						})
+					}
+					uploadsCancel()
+					return
+				}
+				atomic.AddUint64(&totalBytesWritten, uint64(objInfo.Size))
+				objCountPerThread[i]++
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	// We already saw write failures, no need to proceed into read's
+	if retError != "" {
+		return SpeedtestResult{Uploads: totalBytesWritten, Downloads: totalBytesRead, Error: retError}, nil
+	}
+
+	downloadsCtx, downloadsCancel := context.WithCancel(context.Background())
+	defer downloadsCancel()
+	go func() {
+		time.Sleep(duration)
+		downloadsCancel()
+	}()
+
+	wg.Add(concurrent)
+	for i := 0; i < concurrent; i++ {
+		go func(i int) {
+			defer wg.Done()
+			var j uint64
+			if objCountPerThread[i] == 0 {
+				return
+			}
+			for {
+				if objCountPerThread[i] == j {
+					j = 0
+				}
+				r, err := objAPI.GetObjectNInfo(downloadsCtx, minioMetaBucket, fmt.Sprintf("%s.%d.%d",
+					objNamePrefix, i, j), nil, nil, noLock, ObjectOptions{})
+				if err != nil {
+					if isErrObjectNotFound(err) {
+						continue
+					}
+					if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) {
+						errOnce.Do(func() {
+							retError = err.Error()
+						})
+					}
+					downloadsCancel()
+					return
+				}
+				n, err := io.Copy(ioutil.Discard, r)
+				r.Close()
+				if err == nil {
+					// Only capture success criteria - do not
+					// have to capture failed reads, truncated
+					// reads etc.
+					atomic.AddUint64(&totalBytesRead, uint64(n))
+				}
+				if err != nil {
+					if !contextCanceled(downloadsCtx) && !errors.Is(err, context.Canceled) {
+						errOnce.Do(func() {
+							retError = err.Error()
+						})
+					}
+					downloadsCancel()
+					return
+				}
+				j++
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	return SpeedtestResult{Uploads: totalBytesWritten, Downloads: totalBytesRead, Error: retError}, nil
+}
+
+// To collect RX stats during "mc support perf net"
+// RXSample holds the RX bytes for the duration between
+// the last peer to connect and the first peer to disconnect.
+// This is to improve the RX throughput accuracy.
+type netPerfRX struct {
+	RX                uint64    // RX bytes
+	lastToConnect     time.Time // time at which last peer to connect to us
+	firstToDisconnect time.Time // time at which the first peer disconnects from us
+	RXSample          uint64    // RX bytes between lastToConnect and firstToDisconnect
+	activeConnections uint64
+	sync.RWMutex
+}
+
+func (n *netPerfRX) Connect() {
+	n.Lock()
+	defer n.Unlock()
+	n.activeConnections++
+	atomic.StoreUint64(&globalNetPerfRX.RX, 0)
+	n.lastToConnect = time.Now()
+}
+
+func (n *netPerfRX) Disconnect() {
+	n.Lock()
+	defer n.Unlock()
+	n.activeConnections--
+	if n.firstToDisconnect.IsZero() {
+		n.RXSample = atomic.LoadUint64(&n.RX)
+		n.firstToDisconnect = time.Now()
+	}
+}
+
+func (n *netPerfRX) ActiveConnections() uint64 {
+	n.RLock()
+	defer n.RUnlock()
+	return n.activeConnections
+}
+
+func (n *netPerfRX) Reset() {
+	n.RLock()
+	defer n.RUnlock()
+	n.RX = 0
+	n.RXSample = 0
+	n.lastToConnect = time.Time{}
+	n.firstToDisconnect = time.Time{}
+}
+
+// Reader to read random data.
+type netperfReader struct {
+	n   uint64
+	eof chan struct{}
+	buf []byte
+}
+
+func (m *netperfReader) Read(b []byte) (int, error) {
+	select {
+	case <-m.eof:
+		return 0, io.EOF
+	default:
+	}
+	n := copy(b, m.buf)
+	atomic.AddUint64(&m.n, uint64(n))
+	return n, nil
+}
+
+func netperf(ctx context.Context, duration time.Duration) madmin.NetperfNodeResult {
+	r := &netperfReader{eof: make(chan struct{})}
+	r.buf = make([]byte, 128*humanize.KiByte)
+	rand.Read(r.buf)
+
+	connectionsPerPeer := 16
+
+	if len(globalNotificationSys.peerClients) > 16 {
+		// For a large cluster it's enough to have 1 connection per peer to saturate the network.
+		connectionsPerPeer = 1
+	}
+
+	errStr := ""
+	var wg sync.WaitGroup
+	for index := range globalNotificationSys.peerClients {
+		if globalNotificationSys.peerClients[index] == nil {
+			continue
+		}
+		go func(index int) {
+			for i := 0; i < connectionsPerPeer; i++ {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					err := globalNotificationSys.peerClients[index].DevNull(ctx, r)
+					if err != nil {
+						errStr = err.Error()
+					}
+				}()
+			}
+		}(index)
+	}
+
+	time.Sleep(duration)
+	close(r.eof)
+	wg.Wait()
+	for {
+		if globalNetPerfRX.ActiveConnections() == 0 {
+			break
+		}
+		time.Sleep(time.Second)
+	}
+	rx := float64(globalNetPerfRX.RXSample)
+	delta := globalNetPerfRX.firstToDisconnect.Sub(globalNetPerfRX.lastToConnect)
+	if delta < 0 {
+		rx = 0
+		errStr = "network disconnection issues detected"
+	}
+
+	globalNetPerfRX.Reset()
+	return madmin.NetperfNodeResult{Endpoint: "", TX: r.n / uint64(duration.Seconds()), RX: uint64(rx / delta.Seconds()), Error: errStr}
+}

cmd/server-main.go

@@ -448,6 +448,11 @@ func serverMain(ctx *cli.Context) {
 	// Set system resources to maximum.
 	setMaxResources()
 
+	// Verify kernel release and version.
+	if oldLinux() {
+		logger.Info(color.RedBold("WARNING: Detected Linux kernel version older than 4.0.0 release, there are some known potential performance problems with this kernel version. MinIO recommends a minimum of 4.x.x linux kernel version for best performance"))
+	}
+
 	// Configure server.
 	handler, err := configureServerHandler(globalEndpoints)
 	if err != nil {

cmd/server-rlimit.go

@@ -21,10 +21,28 @@ import (
 	"runtime"
 	"runtime/debug"
 
+	"github.com/minio/minio/internal/kernel"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/sys"
 )
 
+func oldLinux() bool {
+	currentKernel, err := kernel.CurrentVersion()
+	if err != nil {
+		// Could not probe the kernel version
+		return false
+	}
+
+	if currentKernel == 0 {
+		// We could not get any valid value return false
+		return false
+	}
+
+	// legacy linux indicator for printing warnings
+	// about older Linux kernels and Go runtime.
+	return currentKernel < kernel.Version(4, 0, 0)
+}
+
 func setMaxResources() (err error) {
 	// Set the Go runtime max threads threshold to 90% of kernel setting.
 	sysMaxThreads, mErr := sys.GetMaxThreads()

cmd/storage-interface.go

@@ -29,17 +29,40 @@ type StorageAPI interface {
 	String() string
 
 	// Storage operations.
-	IsOnline() bool      // Returns true if disk is online.
-	LastConn() time.Time // Returns the last time this disk (re)-connected
 
+	// Returns true if disk is online and its valid i.e valid format.json.
+	// This has nothing to do with if the drive is hung or not responding.
+	// For that individual storage API calls will fail properly. The purpose
+	// of this function is to know if the "drive" has "format.json" or not
+	// if it has a "format.json" then is it correct "format.json" or not.
+	IsOnline() bool
+
+	// Returns the last time this disk (re)-connected
+	LastConn() time.Time
+
+	// Indicates if disk is local or not.
 	IsLocal() bool
-	Hostname() string   // Returns host name if remote host.
-	Endpoint() Endpoint // Returns endpoint.
 
+	// Returns hostname if disk is remote.
+	Hostname() string
+
+	// Returns the entire endpoint.
+	Endpoint() Endpoint
+
+	// Close the disk, mark it purposefully closed, only implemented for remote disks.
 	Close() error
+
+	// Returns the unique 'uuid' of this disk.
 	GetDiskID() (string, error)
+
+	// Set a unique 'uuid' for this disk, only used when
+	// disk is replaced and formatted.
 	SetDiskID(id string)
-	Healing() *healingTracker // Returns nil if disk is not healing.
+
+	// Returns healing information for a newly replaced disk,
+	// returns 'nil' once healing is complete or if the disk
+	// has never been replaced.
+	Healing() *healingTracker
 
 	DiskInfo(ctx context.Context) (info DiskInfo, err error)
 	NSScanner(ctx context.Context, cache dataUsageCache, updates chan<- dataUsageEntry) (dataUsageCache, error)

cmd/storage-rest-server.go

@@ -52,7 +52,7 @@ var errDiskStale = errors.New("disk stale")
 
 // To abstract a disk over network.
 type storageRESTServer struct {
-	storage *xlStorage
+	storage *xlStorageDiskIDCheck
 }
 
 func (s *storageRESTServer) writeErrorResponse(w http.ResponseWriter, err error) {
@@ -1233,7 +1233,8 @@ func registerStorageRESTHandlers(router *mux.Router, endpointServerPools Endpoin
 
 			endpoint := storage.Endpoint()
 
-			server := &storageRESTServer{storage: storage}
+			server := &storageRESTServer{storage: newXLStorageDiskIDCheck(storage)}
+			server.storage.SetDiskID(storage.diskID)
 
 			subrouter := router.PathPrefix(path.Join(storageRESTPrefix, endpoint.Path)).Subrouter()
 

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

@@ -19,13 +19,19 @@ package cmd
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"io"
+	"math/rand"
+	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/minio/madmin-go"
+	"github.com/minio/minio/internal/logger"
 )
 
 //go:generate stringer -type=storageMetric -trimprefix=storageMetric $GOFILE
@@ -69,10 +75,11 @@ type xlStorageDiskIDCheck struct {
 	// do not re-order them, if you add new fields
 	// please use `fieldalignment ./...` to check
 	// if your changes are not causing any problems.
-	storage      StorageAPI
+	storage      *xlStorage
 	apiLatencies [storageMetricLast]*lockedLastMinuteLatency
 	diskID       string
 	apiCalls     [storageMetricLast]uint64
+	health       *diskHealthTracker
 }
 
 func (p *xlStorageDiskIDCheck) getMetrics() DiskMetrics {
@@ -109,6 +116,7 @@ func (e *lockedLastMinuteLatency) value() uint64 {
 func newXLStorageDiskIDCheck(storage *xlStorage) *xlStorageDiskIDCheck {
 	xl := xlStorageDiskIDCheck{
 		storage: storage,
+		health:  newDiskHealthTracker(),
 	}
 	for i := range xl.apiLatencies[:] {
 		xl.apiLatencies[i] = &lockedLastMinuteLatency{}
@@ -215,25 +223,31 @@ func (p *xlStorageDiskIDCheck) DiskInfo(ctx context.Context) (info DiskInfo, err
 			return info, errDiskNotFound
 		}
 	}
+
+	if p.health.isFaulty() {
+		// if disk is already faulty return faulty for 'mc admin info' output and prometheus alerts.
+		return info, errFaultyDisk
+	}
+
 	return info, nil
 }
 
 func (p *xlStorageDiskIDCheck) MakeVolBulk(ctx context.Context, volumes ...string) (err error) {
-	defer p.updateStorageMetrics(storageMetricMakeVolBulk, volumes...)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricMakeVolBulk, volumes...)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
+
 	return p.storage.MakeVolBulk(ctx, volumes...)
 }
 
 func (p *xlStorageDiskIDCheck) MakeVol(ctx context.Context, volume string) (err error) {
-	defer p.updateStorageMetrics(storageMetricMakeVol, volume)()
-
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricMakeVol, volume)
+	if err != nil {
+		return err
+	}
+	defer done(&err)
 	if contextCanceled(ctx) {
 		return ctx.Err()
 	}
@@ -244,167 +258,122 @@ func (p *xlStorageDiskIDCheck) MakeVol(ctx context.Context, volume string) (err
 	return p.storage.MakeVol(ctx, volume)
 }
 
-func (p *xlStorageDiskIDCheck) ListVols(ctx context.Context) ([]VolInfo, error) {
-	defer p.updateStorageMetrics(storageMetricListVols, "/")()
-
-	if contextCanceled(ctx) {
-		return nil, ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) ListVols(ctx context.Context) (vi []VolInfo, err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricListVols, "/")
+	if err != nil {
 		return nil, err
 	}
+	defer done(&err)
+
 	return p.storage.ListVols(ctx)
 }
 
 func (p *xlStorageDiskIDCheck) StatVol(ctx context.Context, volume string) (vol VolInfo, err error) {
-	defer p.updateStorageMetrics(storageMetricStatVol, volume)()
-
-	if contextCanceled(ctx) {
-		return VolInfo{}, ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricStatVol, volume)
+	if err != nil {
 		return vol, err
 	}
+	defer done(&err)
+
 	return p.storage.StatVol(ctx, volume)
 }
 
 func (p *xlStorageDiskIDCheck) DeleteVol(ctx context.Context, volume string, forceDelete bool) (err error) {
-	defer p.updateStorageMetrics(storageMetricDeleteVol, volume)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricDeleteVol, volume)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
+
 	return p.storage.DeleteVol(ctx, volume, forceDelete)
 }
 
-func (p *xlStorageDiskIDCheck) ListDir(ctx context.Context, volume, dirPath string, count int) ([]string, error) {
-	defer p.updateStorageMetrics(storageMetricListDir, volume, dirPath)()
-
-	if contextCanceled(ctx) {
-		return nil, ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) ListDir(ctx context.Context, volume, dirPath string, count int) (s []string, err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricListDir, volume, dirPath)
+	if err != nil {
 		return nil, err
 	}
+	defer done(&err)
 
 	return p.storage.ListDir(ctx, volume, dirPath, count)
 }
 
 func (p *xlStorageDiskIDCheck) ReadFile(ctx context.Context, volume string, path string, offset int64, buf []byte, verifier *BitrotVerifier) (n int64, err error) {
-	defer p.updateStorageMetrics(storageMetricReadFile, volume, path)()
-
-	if contextCanceled(ctx) {
-		return 0, ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricReadFile, volume, path)
+	if err != nil {
 		return 0, err
 	}
+	defer done(&err)
 
 	return p.storage.ReadFile(ctx, volume, path, offset, buf, verifier)
 }
 
 func (p *xlStorageDiskIDCheck) AppendFile(ctx context.Context, volume string, path string, buf []byte) (err error) {
-	defer p.updateStorageMetrics(storageMetricAppendFile, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricAppendFile, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.AppendFile(ctx, volume, path, buf)
 }
 
-func (p *xlStorageDiskIDCheck) CreateFile(ctx context.Context, volume, path string, size int64, reader io.Reader) error {
-	defer p.updateStorageMetrics(storageMetricCreateFile, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) CreateFile(ctx context.Context, volume, path string, size int64, reader io.Reader) (err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricCreateFile, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.CreateFile(ctx, volume, path, size, reader)
 }
 
 func (p *xlStorageDiskIDCheck) ReadFileStream(ctx context.Context, volume, path string, offset, length int64) (io.ReadCloser, error) {
-	defer p.updateStorageMetrics(storageMetricReadFileStream, volume, path)()
-
-	if contextCanceled(ctx) {
-		return nil, ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricReadFileStream, volume, path)
+	if err != nil {
 		return nil, err
 	}
+	defer done(&err)
 
 	return p.storage.ReadFileStream(ctx, volume, path, offset, length)
 }
 
-func (p *xlStorageDiskIDCheck) RenameFile(ctx context.Context, srcVolume, srcPath, dstVolume, dstPath string) error {
-	defer p.updateStorageMetrics(storageMetricRenameFile, srcVolume, srcPath, dstVolume, dstPath)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) RenameFile(ctx context.Context, srcVolume, srcPath, dstVolume, dstPath string) (err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricRenameFile, srcVolume, srcPath, dstVolume, dstPath)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.RenameFile(ctx, srcVolume, srcPath, dstVolume, dstPath)
 }
 
-func (p *xlStorageDiskIDCheck) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string) error {
-	defer p.updateStorageMetrics(storageMetricRenameData, srcPath, fi.DataDir, dstVolume, dstPath)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string) (err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricRenameData, srcPath, fi.DataDir, dstVolume, dstPath)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.RenameData(ctx, srcVolume, srcPath, fi, dstVolume, dstPath)
 }
 
 func (p *xlStorageDiskIDCheck) CheckParts(ctx context.Context, volume string, path string, fi FileInfo) (err error) {
-	defer p.updateStorageMetrics(storageMetricCheckParts, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricCheckParts, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.CheckParts(ctx, volume, path, fi)
 }
 
 func (p *xlStorageDiskIDCheck) Delete(ctx context.Context, volume string, path string, recursive bool) (err error) {
-	defer p.updateStorageMetrics(storageMetricDelete, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricDelete, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.Delete(ctx, volume, path, recursive)
 }
@@ -417,136 +386,102 @@ func (p *xlStorageDiskIDCheck) DeleteVersions(ctx context.Context, volume string
 	if len(versions) > 0 {
 		path = versions[0].Name
 	}
-
-	defer p.updateStorageMetrics(storageMetricDeleteVersions, volume, path)()
-
 	errs = make([]error, len(versions))
-
-	if contextCanceled(ctx) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricDeleteVersions, volume, path)
+	if err != nil {
 		for i := range errs {
 			errs[i] = ctx.Err()
 		}
 		return errs
 	}
-
-	if err := p.checkDiskStale(); err != nil {
-		for i := range errs {
-			errs[i] = err
+	defer done(&err)
+	errs = p.storage.DeleteVersions(ctx, volume, versions)
+	for i := range errs {
+		if errs[i] != nil {
+			err = errs[i]
+			break
 		}
-		return errs
 	}
 
-	return p.storage.DeleteVersions(ctx, volume, versions)
+	return errs
 }
 
-func (p *xlStorageDiskIDCheck) VerifyFile(ctx context.Context, volume, path string, fi FileInfo) error {
-	defer p.updateStorageMetrics(storageMetricVerifyFile, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err := p.checkDiskStale(); err != nil {
+func (p *xlStorageDiskIDCheck) VerifyFile(ctx context.Context, volume, path string, fi FileInfo) (err error) {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricVerifyFile, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.VerifyFile(ctx, volume, path, fi)
 }
 
 func (p *xlStorageDiskIDCheck) WriteAll(ctx context.Context, volume string, path string, b []byte) (err error) {
-	defer p.updateStorageMetrics(storageMetricWriteAll, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricWriteAll, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.WriteAll(ctx, volume, path, b)
 }
 
 func (p *xlStorageDiskIDCheck) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool) (err error) {
-	defer p.updateStorageMetrics(storageMetricDeleteVersion, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricDeleteVersion, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.DeleteVersion(ctx, volume, path, fi, forceDelMarker)
 }
 
 func (p *xlStorageDiskIDCheck) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo) (err error) {
-	defer p.updateStorageMetrics(storageMetricUpdateMetadata, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricUpdateMetadata, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.UpdateMetadata(ctx, volume, path, fi)
 }
 
 func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, volume, path string, fi FileInfo) (err error) {
-	defer p.updateStorageMetrics(storageMetricWriteMetadata, volume, path)()
-
-	if contextCanceled(ctx) {
-		return ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricWriteMetadata, volume, path)
+	if err != nil {
 		return err
 	}
+	defer done(&err)
 
 	return p.storage.WriteMetadata(ctx, volume, path, fi)
 }
 
 func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, volume, path, versionID string, readData bool) (fi FileInfo, err error) {
-	defer p.updateStorageMetrics(storageMetricReadVersion, volume, path)()
-
-	if contextCanceled(ctx) {
-		return fi, ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricReadVersion, volume, path)
+	if err != nil {
 		return fi, err
 	}
+	defer done(&err)
 
 	return p.storage.ReadVersion(ctx, volume, path, versionID, readData)
 }
 
 func (p *xlStorageDiskIDCheck) ReadAll(ctx context.Context, volume string, path string) (buf []byte, err error) {
-	defer p.updateStorageMetrics(storageMetricReadAll, volume, path)()
-
-	if contextCanceled(ctx) {
-		return nil, ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricReadAll, volume, path)
+	if err != nil {
 		return nil, err
 	}
+	defer done(&err)
 
 	return p.storage.ReadAll(ctx, volume, path)
 }
 
 func (p *xlStorageDiskIDCheck) StatInfoFile(ctx context.Context, volume, path string, glob bool) (stat []StatInfo, err error) {
-	defer p.updateStorageMetrics(storageMetricStatInfoFile, volume, path)()
-
-	if contextCanceled(ctx) {
-		return nil, ctx.Err()
-	}
-
-	if err = p.checkDiskStale(); err != nil {
+	ctx, done, err := p.TrackDiskHealth(ctx, storageMetricStatInfoFile, volume, path)
+	if err != nil {
 		return nil, err
 	}
+	defer done(&err)
 
 	return p.storage.StatInfoFile(ctx, volume, path, glob)
 }
@@ -565,10 +500,10 @@ func storageTrace(s storageMetric, startTime time.Time, duration time.Duration,
 }
 
 // Update storage metrics
-func (p *xlStorageDiskIDCheck) updateStorageMetrics(s storageMetric, paths ...string) func() {
+func (p *xlStorageDiskIDCheck) updateStorageMetrics(s storageMetric, paths ...string) func(err *error) {
 	startTime := time.Now()
 	trace := globalTrace.NumSubscribers() > 0
-	return func() {
+	return func(err *error) {
 		duration := time.Since(startTime)
 
 		atomic.AddUint64(&p.apiCalls[s], 1)
@@ -579,3 +514,317 @@ func (p *xlStorageDiskIDCheck) updateStorageMetrics(s storageMetric, paths ...st
 		}
 	}
 }
+
+const (
+	diskHealthOK = iota
+	diskHealthFaulty
+)
+
+// diskMaxConcurrent is the maximum number of running concurrent operations
+// for local and (incoming) remote disk ops respectively.
+var diskMaxConcurrent = 50
+
+func init() {
+	if s, ok := os.LookupEnv("_MINIO_DISK_MAX_CONCURRENT"); ok && s != "" {
+		var err error
+		diskMaxConcurrent, err = strconv.Atoi(s)
+		if err != nil {
+			logger.Fatal(err, "invalid _MINIO_DISK_MAX_CONCURRENT value")
+		}
+	}
+}
+
+type diskHealthTracker struct {
+	// atomic time of last success
+	lastSuccess int64
+
+	// atomic time of last time a token was grabbed.
+	lastStarted int64
+
+	// Atomic status of disk.
+	status int32
+
+	// Atomic number of requests blocking for a token.
+	blocked int32
+
+	// Concurrency tokens.
+	tokens chan struct{}
+}
+
+// newDiskHealthTracker creates a new disk health tracker.
+func newDiskHealthTracker() *diskHealthTracker {
+	d := diskHealthTracker{
+		lastSuccess: time.Now().UnixNano(),
+		lastStarted: time.Now().UnixNano(),
+		status:      diskHealthOK,
+		tokens:      make(chan struct{}, diskMaxConcurrent),
+	}
+	for i := 0; i < diskMaxConcurrent; i++ {
+		d.tokens <- struct{}{}
+	}
+	return &d
+}
+
+// logSuccess will update the last successful operation time.
+func (d *diskHealthTracker) logSuccess() {
+	atomic.StoreInt64(&d.lastSuccess, time.Now().UnixNano())
+}
+
+func (d *diskHealthTracker) isFaulty() bool {
+	return atomic.LoadInt32(&d.status) == diskHealthFaulty
+}
+
+type (
+	healthDiskCtxKey   struct{}
+	healthDiskCtxValue struct {
+		lastSuccess *int64
+	}
+)
+
+// logSuccess will update the last successful operation time.
+func (h *healthDiskCtxValue) logSuccess() {
+	atomic.StoreInt64(h.lastSuccess, time.Now().UnixNano())
+}
+
+// noopDoneFunc is a no-op done func.
+// Can be reused.
+var noopDoneFunc = func(_ *error) {}
+
+// TrackDiskHealth for this request.
+// When a non-nil error is returned 'done' MUST be called
+// with the status of the response, if it corresponds to disk health.
+// If the pointer sent to done is non-nil AND the error
+// is either nil or io.EOF the disk is considered good.
+// So if unsure if the disk status is ok, return nil as a parameter to done.
+// Shadowing will work as long as return error is named: https://go.dev/play/p/sauq86SsTN2
+func (p *xlStorageDiskIDCheck) TrackDiskHealth(ctx context.Context, s storageMetric, paths ...string) (c context.Context, done func(*error), err error) {
+	done = noopDoneFunc
+	if contextCanceled(ctx) {
+		return ctx, done, ctx.Err()
+	}
+
+	// Return early if disk is faulty already.
+	if atomic.LoadInt32(&p.health.status) == diskHealthFaulty {
+		return ctx, done, errFaultyDisk
+	}
+
+	// Verify if the disk is not stale
+	// - missing format.json (unformatted drive)
+	// - format.json is valid but invalid 'uuid'
+	if err = p.checkDiskStale(); err != nil {
+		return ctx, done, err
+	}
+
+	// Disallow recursive tracking to avoid deadlocks.
+	if ctx.Value(healthDiskCtxKey{}) != nil {
+		done = p.updateStorageMetrics(s, paths...)
+		return ctx, done, nil
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx, done, ctx.Err()
+	case <-p.health.tokens:
+		// Fast path, got token.
+	default:
+		// We ran out of tokens, check health before blocking.
+		err = p.waitForToken(ctx)
+		if err != nil {
+			return ctx, done, err
+		}
+	}
+	// We only progress here if we got a token.
+
+	atomic.StoreInt64(&p.health.lastStarted, time.Now().UnixNano())
+	ctx = context.WithValue(ctx, healthDiskCtxKey{}, &healthDiskCtxValue{lastSuccess: &p.health.lastSuccess})
+	si := p.updateStorageMetrics(s, paths...)
+	t := time.Now()
+	var once sync.Once
+	return ctx, func(errp *error) {
+		once.Do(func() {
+			if false {
+				var ers string
+				if errp != nil {
+					err := *errp
+					ers = fmt.Sprint(err)
+				}
+				fmt.Println(time.Now().Format(time.RFC3339), "op", s, "took", time.Since(t), "result:", ers, "disk:", p.storage.String(), "path:", strings.Join(paths, "/"))
+			}
+			p.health.tokens <- struct{}{}
+			if errp != nil {
+				err := *errp
+				if err != nil && !errors.Is(err, io.EOF) {
+					return
+				}
+				p.health.logSuccess()
+			}
+			si(errp)
+		})
+	}, nil
+}
+
+// waitForToken will wait for a token, while periodically
+// checking the disk status.
+// If nil is returned a token was picked up.
+func (p *xlStorageDiskIDCheck) waitForToken(ctx context.Context) (err error) {
+	atomic.AddInt32(&p.health.blocked, 1)
+	defer func() {
+		atomic.AddInt32(&p.health.blocked, -1)
+	}()
+	// Avoid stampeding herd...
+	ticker := time.NewTicker(5*time.Second + time.Duration(rand.Int63n(int64(5*time.Second))))
+	defer ticker.Stop()
+	for {
+		err = p.checkHealth(ctx)
+		if err != nil {
+			return err
+		}
+		select {
+		case <-ticker.C:
+		// Ticker expired, check health again.
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-p.health.tokens:
+			return nil
+		}
+	}
+}
+
+// checkHealth should only be called when tokens have run out.
+// This will check if disk should be taken offline.
+func (p *xlStorageDiskIDCheck) checkHealth(ctx context.Context) (err error) {
+	if atomic.LoadInt32(&p.health.status) == diskHealthFaulty {
+		return errFaultyDisk
+	}
+	// Check if there are tokens.
+	if len(p.health.tokens) > 0 {
+		return nil
+	}
+
+	const maxTimeSinceLastSuccess = 30 * time.Second
+	const minTimeSinceLastOpStarted = 15 * time.Second
+
+	// To avoid stampeding herd (100s of simultaneous starting requests)
+	// there must be a delay between the last started request and now
+	// for the last lastSuccess to be useful.
+	t := time.Since(time.Unix(0, atomic.LoadInt64(&p.health.lastStarted)))
+	if t < minTimeSinceLastOpStarted {
+		return nil
+	}
+
+	// If also more than 15 seconds since last success, take disk offline.
+	t = time.Since(time.Unix(0, atomic.LoadInt64(&p.health.lastSuccess)))
+	if t > maxTimeSinceLastSuccess {
+		if atomic.CompareAndSwapInt32(&p.health.status, diskHealthOK, diskHealthFaulty) {
+			logger.LogAlwaysIf(ctx, fmt.Errorf("taking disk %s offline, time since last response %v", p.storage.String(), t.Round(time.Millisecond)))
+			go p.monitorDiskStatus()
+		}
+		return errFaultyDisk
+	}
+	return nil
+}
+
+// monitorDiskStatus should be called once when a drive has been marked offline.
+// Once the disk has been deemed ok, it will return to online status.
+func (p *xlStorageDiskIDCheck) monitorDiskStatus() {
+	t := time.NewTicker(5 * time.Second)
+	defer t.Stop()
+	fn := mustGetUUID()
+	for range t.C {
+		if len(p.health.tokens) == 0 {
+			// Queue is still full, no need to check.
+			continue
+		}
+		err := p.storage.WriteAll(context.Background(), minioMetaTmpBucket, fn, []byte{10000: 42})
+		if err != nil {
+			continue
+		}
+		b, err := p.storage.ReadAll(context.Background(), minioMetaTmpBucket, fn)
+		if err != nil || len(b) != 10001 {
+			continue
+		}
+		err = p.storage.Delete(context.Background(), minioMetaTmpBucket, fn, false)
+		if err == nil {
+			logger.Info("Able to read+write, bringing disk %s online.", p.storage.String())
+			atomic.StoreInt32(&p.health.status, diskHealthOK)
+			return
+		}
+	}
+}
+
+// diskHealthCheckOK will check if the provided error is nil
+// and update disk status if good.
+// For convenience a bool is returned to indicate any error state
+// that is not io.EOF.
+func diskHealthCheckOK(ctx context.Context, err error) bool {
+	// Check if context has a disk health check.
+	tracker, ok := ctx.Value(healthDiskCtxKey{}).(*healthDiskCtxValue)
+	if !ok {
+		// No tracker, return
+		return err == nil || errors.Is(err, io.EOF)
+	}
+	if err == nil || errors.Is(err, io.EOF) {
+		tracker.logSuccess()
+		return true
+	}
+	return false
+}
+
+// diskHealthWrapper provides either a io.Reader or io.Writer
+// that updates status of the provided tracker.
+// Use through diskHealthReader or diskHealthWriter.
+type diskHealthWrapper struct {
+	tracker *healthDiskCtxValue
+	r       io.Reader
+	w       io.Writer
+}
+
+func (d *diskHealthWrapper) Read(p []byte) (int, error) {
+	if d.r == nil {
+		return 0, fmt.Errorf("diskHealthWrapper: Read with no reader")
+	}
+	n, err := d.r.Read(p)
+	if err == nil || err == io.EOF && n > 0 {
+		d.tracker.logSuccess()
+	}
+	return n, err
+}
+
+func (d *diskHealthWrapper) Write(p []byte) (int, error) {
+	if d.w == nil {
+		return 0, fmt.Errorf("diskHealthWrapper: Write with no writer")
+	}
+	n, err := d.w.Write(p)
+	if err == nil && n == len(p) {
+		d.tracker.logSuccess()
+	}
+	return n, err
+}
+
+// diskHealthReader provides a wrapper that will update disk health on
+// ctx, on every successful read.
+// This should only be used directly at the os/syscall level,
+// otherwise buffered operations may return false health checks.
+func diskHealthReader(ctx context.Context, r io.Reader) io.Reader {
+	// Check if context has a disk health check.
+	tracker, ok := ctx.Value(healthDiskCtxKey{}).(*healthDiskCtxValue)
+	if !ok {
+		// No need to wrap
+		return r
+	}
+	return &diskHealthWrapper{r: r, tracker: tracker}
+}
+
+// diskHealthWriter provides a wrapper that will update disk health on
+// ctx, on every successful write.
+// This should only be used directly at the os/syscall level,
+// otherwise buffered operations may return false health checks.
+func diskHealthWriter(ctx context.Context, w io.Writer) io.Writer {
+	// Check if context has a disk health check.
+	tracker, ok := ctx.Value(healthDiskCtxKey{}).(*healthDiskCtxValue)
+	if !ok {
+		// No need to wrap
+		return w
+	}
+	return &diskHealthWrapper{w: w, tracker: tracker}
+}

cmd/xl-storage.go

@@ -49,11 +49,12 @@ import (
 
 const (
 	nullVersionID = "null"
-	// Really large streams threshold per shard.
-	reallyLargeFileThreshold = 64 * humanize.MiByte // Optimized for HDDs
+	// Largest streams threshold per shard.
+	largestFileThreshold = 64 * humanize.MiByte // Optimized for HDDs
 
 	// Small file threshold below which data accompanies metadata from storage layer.
 	smallFileThreshold = 128 * humanize.KiByte // Optimized for NVMe/SSDs
+
 	// For hardrives it is possible to set this to a lower value to avoid any
 	// spike in latency. But currently we are simply keeping it optimal for SSDs.
 
@@ -683,9 +684,11 @@ func (s *xlStorage) SetDiskID(id string) {
 
 func (s *xlStorage) MakeVolBulk(ctx context.Context, volumes ...string) error {
 	for _, volume := range volumes {
-		if err := s.MakeVol(ctx, volume); err != nil && !errors.Is(err, errVolumeExists) {
+		err := s.MakeVol(ctx, volume)
+		if err != nil && !errors.Is(err, errVolumeExists) {
 			return err
 		}
+		diskHealthCheckOK(ctx, err)
 	}
 	return nil
 }
@@ -943,6 +946,7 @@ func (s *xlStorage) DeleteVersions(ctx context.Context, volume string, versions
 		if err := s.deleteVersions(ctx, volume, fiv.Name, fiv.Versions...); err != nil {
 			errs[i] = err
 		}
+		diskHealthCheckOK(ctx, errs[i])
 	}
 
 	return errs
@@ -1382,7 +1386,7 @@ func (s *xlStorage) readAllData(ctx context.Context, volumeDir string, filePath
 		buf = make([]byte, sz)
 	}
 	// Read file...
-	_, err = io.ReadFull(r, buf)
+	_, err = io.ReadFull(diskHealthReader(ctx, r), buf)
 
 	return buf, stat.ModTime().UTC(), osErrToFileErr(err)
 }
@@ -1693,7 +1697,7 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 	r := struct {
 		io.Reader
 		io.Closer
-	}{Reader: io.LimitReader(or, length), Closer: closeWrapper(func() error {
+	}{Reader: io.LimitReader(diskHealthReader(ctx, or), length), Closer: closeWrapper(func() error {
 		if !alignment || offset+length%xioutil.DirectioAlignSize != 0 {
 			// invalidate page-cache for unaligned reads.
 			if !globalAPIConfig.isDisableODirect() {
@@ -1783,8 +1787,8 @@ func (s *xlStorage) CreateFile(ctx context.Context, volume, path string, fileSiz
 	}()
 
 	var bufp *[]byte
-	if fileSize > 0 && fileSize >= reallyLargeFileThreshold {
-		// use a larger 4MiB buffer for really large streams.
+	if fileSize > 0 && fileSize >= largestFileThreshold {
+		// use a larger 4MiB buffer for a really large streams.
 		bufp = xioutil.ODirectPoolXLarge.Get().(*[]byte)
 		defer xioutil.ODirectPoolXLarge.Put(bufp)
 	} else {
@@ -1792,7 +1796,7 @@ func (s *xlStorage) CreateFile(ctx context.Context, volume, path string, fileSiz
 		defer xioutil.ODirectPoolLarge.Put(bufp)
 	}
 
-	written, err := xioutil.CopyAligned(w, r, *bufp, fileSize)
+	written, err := xioutil.CopyAligned(diskHealthWriter(ctx, w), r, *bufp, fileSize, w)
 	if err != nil {
 		return err
 	}
@@ -2269,6 +2273,7 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
 			}
 			return osErrToFileErr(err)
 		}
+		diskHealthCheckOK(ctx, err)
 
 		// renameAll only for objects that have xl.meta not saved inline.
 		if len(fi.Data) == 0 && fi.Size > 0 {
@@ -2406,7 +2411,7 @@ func (s *xlStorage) RenameFile(ctx context.Context, srcVolume, srcPath, dstVolum
 	return nil
 }
 
-func (s *xlStorage) bitrotVerify(partPath string, partSize int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
+func (s *xlStorage) bitrotVerify(ctx context.Context, partPath string, partSize int64, algo BitrotAlgorithm, sum []byte, shardSize int64) error {
 	// Open the file for reading.
 	file, err := Open(partPath)
 	if err != nil {
@@ -2421,7 +2426,7 @@ func (s *xlStorage) bitrotVerify(partPath string, partSize int64, algo BitrotAlg
 		// for healing code to fix this file.
 		return err
 	}
-	return bitrotVerify(file, fi.Size(), partSize, algo, sum, shardSize)
+	return bitrotVerify(diskHealthReader(ctx, file), fi.Size(), partSize, algo, sum, shardSize)
 }
 
 func (s *xlStorage) VerifyFile(ctx context.Context, volume, path string, fi FileInfo) (err error) {
@@ -2446,7 +2451,7 @@ func (s *xlStorage) VerifyFile(ctx context.Context, volume, path string, fi File
 	for _, part := range fi.Parts {
 		checksumInfo := erasure.GetChecksumInfo(part.Number)
 		partPath := pathJoin(volumeDir, path, fi.DataDir, fmt.Sprintf("part.%d", part.Number))
-		if err := s.bitrotVerify(partPath,
+		if err := s.bitrotVerify(ctx, partPath,
 			erasure.ShardFileSize(part.Size),
 			checksumInfo.Algorithm,
 			checksumInfo.Hash, erasure.ShardSize()); err != nil {

cmd/xl-storage_test.go

@@ -1884,7 +1884,7 @@ func TestXLStorageVerifyFile(t *testing.T) {
 	if err := storage.WriteAll(context.Background(), volName, fileName, data); err != nil {
 		t.Fatal(err)
 	}
-	if err := storage.storage.(*xlStorage).bitrotVerify(pathJoin(path, volName, fileName), size, algo, hashBytes, 0); err != nil {
+	if err := storage.storage.bitrotVerify(context.Background(), pathJoin(path, volName, fileName), size, algo, hashBytes, 0); err != nil {
 		t.Fatal(err)
 	}
 
@@ -1894,12 +1894,12 @@ func TestXLStorageVerifyFile(t *testing.T) {
 	}
 
 	// Check if VerifyFile reports the incorrect file length (the correct length is `size+1`)
-	if err := storage.storage.(*xlStorage).bitrotVerify(pathJoin(path, volName, fileName), size, algo, hashBytes, 0); err == nil {
+	if err := storage.storage.bitrotVerify(context.Background(), pathJoin(path, volName, fileName), size, algo, hashBytes, 0); err == nil {
 		t.Fatal("expected to fail bitrot check")
 	}
 
 	// Check if bitrot fails
-	if err := storage.storage.(*xlStorage).bitrotVerify(pathJoin(path, volName, fileName), size+1, algo, hashBytes, 0); err == nil {
+	if err := storage.storage.bitrotVerify(context.Background(), pathJoin(path, volName, fileName), size+1, algo, hashBytes, 0); err == nil {
 		t.Fatal("expected to fail bitrot check")
 	}
 
@@ -1928,7 +1928,7 @@ func TestXLStorageVerifyFile(t *testing.T) {
 		t.Fatal(err)
 	}
 	w.(io.Closer).Close()
-	if err := storage.storage.(*xlStorage).bitrotVerify(pathJoin(path, volName, fileName), size, algo, nil, shardSize); err != nil {
+	if err := storage.storage.bitrotVerify(context.Background(), pathJoin(path, volName, fileName), size, algo, nil, shardSize); err != nil {
 		t.Fatal(err)
 	}
 
@@ -1943,10 +1943,10 @@ func TestXLStorageVerifyFile(t *testing.T) {
 		t.Fatal(err)
 	}
 	f.Close()
-	if err := storage.storage.(*xlStorage).bitrotVerify(pathJoin(path, volName, fileName), size, algo, nil, shardSize); err == nil {
+	if err := storage.storage.bitrotVerify(context.Background(), pathJoin(path, volName, fileName), size, algo, nil, shardSize); err == nil {
 		t.Fatal("expected to fail bitrot check")
 	}
-	if err := storage.storage.(*xlStorage).bitrotVerify(pathJoin(path, volName, fileName), size+1, algo, nil, shardSize); err == nil {
+	if err := storage.storage.bitrotVerify(context.Background(), pathJoin(path, volName, fileName), size+1, algo, nil, shardSize); err == nil {
 		t.Fatal("expected to fail bitrot check")
 	}
 }

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

@@ -2,7 +2,7 @@ version: '3.7'
 
 # Settings and configurations that are common for all containers
 x-minio-common: &minio-common
-  image: quay.io/minio/minio:RELEASE.2022-03-03T21-21-16Z
+  image: quay.io/minio/minio:RELEASE.2022-03-11T11-08-23Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

docs/site-replication/run-multi-site-minio-idp.sh

@@ -173,6 +173,17 @@ if [ $? -ne 0 ]; then
     exit_1;
 fi
 
+err_minio2=$(./mc stat minio2/newbucket/xxx --json | jq -r .error.cause.message)
+if [ $? -ne 0 ]; then
+    echo "expecting object to be missing. exiting.."
+    exit_1;
+fi
+
+if [ "${err_minio2}" != "Object does not exist" ]; then
+    echo "expected to see Object does not exist error, exiting..."
+    exit_1;
+fi
+
 ./mc cp README.md minio2/newbucket/
 
 sleep 5

docs/sts/ldap.md

@@ -34,17 +34,17 @@ KEY:
 identity_ldap  enable LDAP SSO support
 
 ARGS:
-MINIO_IDENTITY_LDAP_SERVER_ADDR*            (address)   AD/LDAP server address e.g. "myldapserver.com:636"
-MINIO_IDENTITY_LDAP_LOOKUP_BIND_DN          (string)    DN for LDAP read-only service account used to perform DN and group lookups
-MINIO_IDENTITY_LDAP_LOOKUP_BIND_PASSWORD    (string)    Password for LDAP read-only service account used to perform DN and group lookups
-MINIO_IDENTITY_LDAP_USER_DN_SEARCH_BASE_DN  (string)    ";" separated list of user search base DNs e.g. "dc=myldapserver,dc=com"
-MINIO_IDENTITY_LDAP_USER_DN_SEARCH_FILTER   (string)    Search filter to lookup user DN
-MINIO_IDENTITY_LDAP_GROUP_SEARCH_FILTER     (string)    search filter for groups e.g. "(&(objectclass=groupOfNames)(memberUid=%s))"
-MINIO_IDENTITY_LDAP_GROUP_SEARCH_BASE_DN    (list)      ";" separated list of group search base DNs e.g. "dc=myldapserver,dc=com"
-MINIO_IDENTITY_LDAP_TLS_SKIP_VERIFY         (on|off)    trust server TLS without verification, defaults to "off" (verify)
-MINIO_IDENTITY_LDAP_SERVER_INSECURE         (on|off)    allow plain text connection to AD/LDAP server, defaults to "off"
-MINIO_IDENTITY_LDAP_SERVER_STARTTLS         (on|off)    use StartTLS connection to AD/LDAP server, defaults to "off"
-MINIO_IDENTITY_LDAP_COMMENT                 (sentence)  optionally add a comment to this setting
+MINIO_IDENTITY_LDAP_SERVER_ADDR*             (address)   AD/LDAP server address e.g. "myldapserver.com:636"
+MINIO_IDENTITY_LDAP_LOOKUP_BIND_DN*          (string)    DN for LDAP read-only service account used to perform DN and group lookups
+MINIO_IDENTITY_LDAP_LOOKUP_BIND_PASSWORD     (string)    Password for LDAP read-only service account used to perform DN and group lookups
+MINIO_IDENTITY_LDAP_USER_DN_SEARCH_BASE_DN*  (list)      ";" separated list of user search base DNs e.g. "dc=myldapserver,dc=com"
+MINIO_IDENTITY_LDAP_USER_DN_SEARCH_FILTER*   (string)    Search filter to lookup user DN
+MINIO_IDENTITY_LDAP_GROUP_SEARCH_FILTER      (string)    search filter for groups e.g. "(&(objectclass=groupOfNames)(memberUid=%s))"
+MINIO_IDENTITY_LDAP_GROUP_SEARCH_BASE_DN     (list)      ";" separated list of group search base DNs e.g. "dc=myldapserver,dc=com"
+MINIO_IDENTITY_LDAP_TLS_SKIP_VERIFY          (on|off)    trust server TLS without verification, defaults to "off" (verify)
+MINIO_IDENTITY_LDAP_SERVER_INSECURE          (on|off)    allow plain text connection to AD/LDAP server, defaults to "off"
+MINIO_IDENTITY_LDAP_SERVER_STARTTLS          (on|off)    use StartTLS connection to AD/LDAP server, defaults to "off"
+MINIO_IDENTITY_LDAP_COMMENT                  (sentence)  optionally add a comment to this setting
 ```
 
 ### LDAP server connectivity
@@ -69,8 +69,8 @@ If a self-signed certificate is being used, the certificate can be added to MinI
 A low-privilege read-only LDAP service account is configured in the MinIO server by providing the account's Distinguished Name (DN) and password. This service account is used to perform directory lookups as needed.
 
 ```
-MINIO_IDENTITY_LDAP_LOOKUP_BIND_DN          (string)    DN for LDAP read-only service account used to perform DN and group lookups
-MINIO_IDENTITY_LDAP_LOOKUP_BIND_PASSWORD    (string)    Password for LDAP read-only service account used to perform DN and group lookups
+MINIO_IDENTITY_LDAP_LOOKUP_BIND_DN*          (string)    DN for LDAP read-only service account used to perform DN and group lookups
+MINIO_IDENTITY_LDAP_LOOKUP_BIND_PASSWORD     (string)    Password for LDAP read-only service account used to perform DN and group lookups
 ```
 
 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).
@@ -80,8 +80,8 @@ If you set an empty lookup bind password, the lookup bind will use the unauthent
 When a user provides their LDAP credentials, MinIO runs a lookup query to find the user's Distinguished Name (DN). The search filter and base DN used in this lookup query are configured via the following variables:
 
 ```
-MINIO_IDENTITY_LDAP_USER_DN_SEARCH_BASE_DN  (string)    Base LDAP DN to search for user DN
-MINIO_IDENTITY_LDAP_USER_DN_SEARCH_FILTER   (string)    Search filter to lookup user DN
+MINIO_IDENTITY_LDAP_USER_DN_SEARCH_BASE_DN*  (list)      ";" separated list of user search base DNs e.g. "dc=myldapserver,dc=com"
+MINIO_IDENTITY_LDAP_USER_DN_SEARCH_FILTER*   (string)    Search filter to lookup user DN
 ```
 
 The search filter must use the LDAP username to find the user DN. This is done via [variable substitution](#variable-substitution-in-configuration-strings).

go.mod

@@ -27,9 +27,9 @@ require (
 	github.com/fatih/color v1.13.0
 	github.com/felixge/fgprof v0.9.2
 	github.com/go-ldap/ldap/v3 v3.2.4
-	github.com/go-openapi/loads v0.21.0
+	github.com/go-openapi/loads v0.21.1
 	github.com/go-sql-driver/mysql v1.6.0
-	github.com/golang-jwt/jwt/v4 v4.2.0
+	github.com/golang-jwt/jwt/v4 v4.3.0
 	github.com/gomodule/redigo v1.8.8
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/mux v1.8.0
@@ -45,9 +45,9 @@ require (
 	github.com/lib/pq v1.10.4
 	github.com/miekg/dns v1.1.46
 	github.com/minio/cli v1.22.0
-	github.com/minio/console v0.15.0
+	github.com/minio/console v0.15.2
 	github.com/minio/csvparser v1.0.0
-	github.com/minio/dperf v0.3.2
+	github.com/minio/dperf v0.3.4
 	github.com/minio/highwayhash v1.0.2
 	github.com/minio/kes v0.18.0
 	github.com/minio/madmin-go v1.3.5
@@ -61,8 +61,8 @@ require (
 	github.com/minio/zipindex v0.2.1
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/montanaflynn/stats v0.6.6
-	github.com/nats-io/nats-server/v2 v2.7.2
-	github.com/nats-io/nats.go v1.13.1-0.20220121202836-972a071d373d
+	github.com/nats-io/nats-server/v2 v2.7.4
+	github.com/nats-io/nats.go v1.13.1-0.20220308171302-2f2f6968e98d
 	github.com/nats-io/stan.go v0.10.2
 	github.com/ncw/directio v1.0.5
 	github.com/nsqio/go-nsq v1.0.8
@@ -75,7 +75,7 @@ require (
 	github.com/rs/cors v1.7.0
 	github.com/rs/dnscache v0.0.0-20211102005908-e0241e321417
 	github.com/secure-io/sio-go v0.3.1
-	github.com/shirou/gopsutil/v3 v3.21.12
+	github.com/shirou/gopsutil/v3 v3.22.2
 	github.com/streadway/amqp v1.0.0
 	github.com/tinylib/msgp v1.1.7-0.20211026165309-e818a1881b0e
 	github.com/valyala/bytebufferpool v1.0.0
@@ -85,11 +85,11 @@ require (
 	go.etcd.io/etcd/api/v3 v3.5.2
 	go.etcd.io/etcd/client/v3 v3.5.2
 	go.uber.org/atomic v1.9.0
-	go.uber.org/zap v1.19.1
-	golang.org/x/crypto v0.0.0-20220128200615-198e4374d7ed
-	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8
-	golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27
-	golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11
+	go.uber.org/zap v1.21.0
+	golang.org/x/crypto v0.0.0-20220214200702-86341886e292
+	golang.org/x/oauth2 v0.0.0-20220223155221-ee480838109b
+	golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9
+	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65
 	google.golang.org/api v0.67.0
 	gopkg.in/yaml.v2 v2.4.0
 )
@@ -103,13 +103,13 @@ require (
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
 	github.com/apache/thrift v0.15.0 // indirect
 	github.com/armon/go-metrics v0.3.3 // indirect
-	github.com/asaskevich/govalidator v0.0.0-20200907205600-7a23bdc65eef // indirect
+	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bits-and-blooms/bitset v1.2.0 // indirect
-	github.com/charmbracelet/bubbles v0.10.0 // indirect
-	github.com/charmbracelet/bubbletea v0.19.3 // indirect
+	github.com/charmbracelet/bubbles v0.10.3 // indirect
+	github.com/charmbracelet/bubbletea v0.20.0 // indirect
 	github.com/charmbracelet/lipgloss v0.5.0 // indirect
-	github.com/containerd/console v1.0.2 // indirect
+	github.com/containerd/console v1.0.3 // indirect
 	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.3.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -120,32 +120,30 @@ require (
 	github.com/eapache/queue v1.1.0 // indirect
 	github.com/fatih/structs v1.1.0 // indirect
 	github.com/frankban/quicktest v1.14.0 // indirect
+	github.com/gdamore/encoding v1.0.0 // indirect
+	github.com/gdamore/tcell/v2 v2.4.1-0.20210905002822-f057f0a857a1 // indirect
 	github.com/go-asn1-ber/asn1-ber v1.5.1 // indirect
-	github.com/go-logr/logr v1.2.0 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-openapi/analysis v0.21.2 // indirect
 	github.com/go-openapi/errors v0.20.2 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.19.6 // indirect
-	github.com/go-openapi/runtime v0.21.1 // indirect
+	github.com/go-openapi/runtime v0.23.1 // indirect
 	github.com/go-openapi/spec v0.20.4 // indirect
-	github.com/go-openapi/strfmt v0.21.1 // indirect
-	github.com/go-openapi/swag v0.19.15 // indirect
-	github.com/go-openapi/validate v0.20.3 // indirect
-	github.com/go-stack/stack v1.8.0 // indirect
-	github.com/goccy/go-json v0.8.1 // indirect
+	github.com/go-openapi/strfmt v0.21.2 // indirect
+	github.com/go-openapi/swag v0.21.1 // indirect
+	github.com/go-openapi/validate v0.21.0 // indirect
+	github.com/go-stack/stack v1.8.1 // indirect
+	github.com/goccy/go-json v0.9.4 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
-	github.com/google/go-cmp v0.5.7 // indirect
-	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/pprof v0.0.0-20211214055906-6f57359322fd // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/googleapis/gax-go/v2 v2.1.1 // indirect
-	github.com/googleapis/gnostic v0.5.5 // indirect
-	github.com/gorilla/websocket v1.4.2 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-immutable-radix v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
@@ -155,13 +153,13 @@ require (
 	github.com/jcmturner/gofork v1.0.0 // indirect
 	github.com/jcmturner/goidentity/v6 v6.0.1 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
-	github.com/jessevdk/go-flags v1.4.0 // indirect
+	github.com/jessevdk/go-flags v1.5.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/lestrrat-go/backoff/v2 v2.0.8 // indirect
 	github.com/lestrrat-go/blackmagic v1.0.0 // indirect
 	github.com/lestrrat-go/httpcc v1.0.0 // indirect
 	github.com/lestrrat-go/iter v1.0.1 // indirect
-	github.com/lestrrat-go/jwx v1.2.14 // indirect
+	github.com/lestrrat-go/jwx v1.2.19 // indirect
 	github.com/lestrrat-go/option v1.0.0 // indirect
 	github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
@@ -174,73 +172,59 @@ require (
 	github.com/minio/argon2 v1.0.0 // indirect
 	github.com/minio/colorjson v1.0.1 // indirect
 	github.com/minio/filepath v1.0.0 // indirect
-	github.com/minio/mc v0.0.0-20220204044644-e048c85d71a7 // indirect
+	github.com/minio/mc v0.0.0-20220302011226-f13defa54577 // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
-	github.com/minio/operator v0.0.0-20220110040724-a5d59a342b7f // indirect
-	github.com/mitchellh/mapstructure v1.4.2 // indirect
+	github.com/mitchellh/mapstructure v1.4.3 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/muesli/ansi v0.0.0-20211018074035-2e021307bc4b // indirect
+	github.com/muesli/ansi v0.0.0-20211031195517-c9f0611b6c70 // indirect
 	github.com/muesli/reflow v0.3.0 // indirect
-	github.com/muesli/termenv v0.11.1-0.20220204035834-5ac8409525e0 // indirect
+	github.com/muesli/termenv v0.11.1-0.20220212125758-44cd13922739 // indirect
 	github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296 // indirect
 	github.com/nats-io/nats-streaming-server v0.24.1 // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/navidys/tvxwidgets v0.1.0 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
-	github.com/onsi/ginkgo v1.16.5 // indirect
-	github.com/onsi/gomega v1.16.0 // indirect
-	github.com/pkg/profile v1.6.0 // indirect
-	github.com/pkg/xattr v0.4.4 // indirect
+	github.com/pkg/xattr v0.4.5 // indirect
 	github.com/posener/complete v1.2.3 // indirect
-	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
+	github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c // indirect
 	github.com/pquerna/cachecontrol v0.0.0-20171018203845-0dec1b30a021 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	github.com/rivo/tview v0.0.0-20220216162559-96063d6082f3 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/rjeczalik/notify v0.9.2 // indirect
 	github.com/rogpeppe/go-internal v1.8.1 // indirect
 	github.com/rs/xid v1.3.0 // indirect
 	github.com/sirupsen/logrus v1.8.1 // indirect
-	github.com/tidwall/gjson v1.12.1 // indirect
+	github.com/tidwall/gjson v1.14.0 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/tidwall/sjson v1.2.3 // indirect
-	github.com/tklauser/go-sysconf v0.3.9 // indirect
-	github.com/tklauser/numcpus v0.3.0 // indirect
-	github.com/unrolled/secure v1.0.9 // indirect
+	github.com/tklauser/go-sysconf v0.3.10 // indirect
+	github.com/tklauser/numcpus v0.4.0 // indirect
+	github.com/unrolled/secure v1.10.0 // indirect
 	github.com/xdg/stringprep v1.0.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.2 // indirect
-	go.mongodb.org/mongo-driver v1.7.5 // indirect
+	go.mongodb.org/mongo-driver v1.8.4 // indirect
 	go.opencensus.io v0.23.0 // indirect
-	go.uber.org/multierr v1.7.0 // indirect
+	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/mod v0.5.1 // indirect
-	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
+	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.8 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220207164111-0872dc986b00 // indirect
+	google.golang.org/genproto v0.0.0-20220302033224-9aa15565e42a // indirect
 	google.golang.org/grpc v1.44.0 // indirect
 	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/h2non/filetype.v1 v1.0.5 // indirect
-	gopkg.in/inf.v0 v0.9.1 // indirect
-	gopkg.in/ini.v1 v1.66.3 // indirect
+	gopkg.in/ini.v1 v1.66.4 // indirect
 	gopkg.in/square/go-jose.v2 v2.5.1 // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
-	k8s.io/api v0.23.3 // indirect
-	k8s.io/apimachinery v0.23.3 // indirect
-	k8s.io/client-go v0.23.3 // indirect
-	k8s.io/klog/v2 v2.40.1 // indirect
-	k8s.io/kube-openapi v0.0.0-20211115234752-e816edb12b65 // indirect
-	k8s.io/utils v0.0.0-20211116205334-6203023598ed // indirect
 	maze.io/x/duration v0.0.0-20160924141736-faac084b6075 // indirect
-	sigs.k8s.io/controller-runtime v0.8.0 // indirect
-	sigs.k8s.io/json v0.0.0-20211020170558-c049b76a60c6 // indirect
-	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
-	sigs.k8s.io/yaml v1.2.0 // indirect
 )

helm/minio/templates/deployment.yaml

@@ -60,6 +60,9 @@ spec:
         runAsUser: {{ .Values.securityContext.runAsUser }}
         runAsGroup: {{ .Values.securityContext.runAsGroup }}
         fsGroup: {{ .Values.securityContext.fsGroup }}
+        {{- if and (ge .Capabilities.KubeVersion.Major "1") (ge .Capabilities.KubeVersion.Minor 20) }}
+        fsGroupChangePolicy: {{ .Values.securityContext.fsGroupChangePolicy }}
+        {{- end }}
 {{- end }}
 {{ if .Values.serviceAccount.create }}
       serviceAccountName: {{ .Values.serviceAccount.name }}

helm/minio/templates/statefulset.yaml

@@ -38,6 +38,7 @@ apiVersion: {{ template "minio.statefulset.apiVersion" . }}
 kind: StatefulSet
 metadata:
   name: {{ template "minio.fullname" . }}
+  namespace: {{ .Release.Namespace | quote }}
   labels:
     app: {{ template "minio.name" . }}
     chart: {{ template "minio.chart" . }}
@@ -86,6 +87,9 @@ spec:
         runAsUser: {{ .Values.securityContext.runAsUser }}
         runAsGroup: {{ .Values.securityContext.runAsGroup }}
         fsGroup: {{ .Values.securityContext.fsGroup }}
+        {{- if and (ge .Capabilities.KubeVersion.Major "1") (ge .Capabilities.KubeVersion.Minor 20) }}
+        fsGroupChangePolicy: {{ .Values.securityContext.fsGroupChangePolicy }}
+        {{- end }}
 {{- end }}
 {{ if .Values.serviceAccount.create }}
       serviceAccountName: {{ .Values.serviceAccount.name }}

helm/minio/values.yaml

@@ -237,6 +237,7 @@ securityContext:
   runAsUser: 1000
   runAsGroup: 1000
   fsGroup: 1000
+  fsGroupChangePolicy: "OnRootMismatch"
 
 # Additational pod annotations
 podAnnotations: {}

internal/config/cache/config.go

@@ -115,7 +115,6 @@ func parseCacheDrives(drives string) ([]string, error) {
 			endpoints = append(endpoints, d)
 		}
 	}
-
 	for _, d := range endpoints {
 		if !filepath.IsAbs(d) {
 			return nil, config.ErrInvalidCacheDrivesValue(nil).Msg("cache dir should be absolute path: %s", d)

internal/config/identity/ldap/help.go

@@ -31,7 +31,6 @@ var (
 		config.HelpKV{
 			Key:         LookupBindDN,
 			Description: `DN for LDAP read-only service account used to perform DN and group lookups`,
-			Optional:    true,
 			Type:        "string",
 			Sensitive:   true,
 		},
@@ -45,13 +44,11 @@ var (
 		config.HelpKV{
 			Key:         UserDNSearchBaseDN,
 			Description: `";" separated list of user search base DNs e.g. "dc=myldapserver,dc=com"`,
-			Optional:    true,
 			Type:        "list",
 		},
 		config.HelpKV{
 			Key:         UserDNSearchFilter,
 			Description: `Search filter to lookup user DN`,
-			Optional:    true,
 			Type:        "string",
 		},
 		config.HelpKV{

internal/event/target/webhook.go

@@ -119,10 +119,17 @@ func (target *WebhookTarget) IsActive() (bool, error) {
 		}
 		return false, err
 	}
+	tokens := strings.Fields(target.args.AuthToken)
+	switch len(tokens) {
+	case 2:
+		req.Header.Set("Authorization", target.args.AuthToken)
+	case 1:
+		req.Header.Set("Authorization", "Bearer "+target.args.AuthToken)
+	}
 
 	resp, err := target.httpClient.Do(req)
 	if err != nil {
-		if xnet.IsNetworkOrHostDown(err, false) || errors.Is(err, context.DeadlineExceeded) {
+		if xnet.IsNetworkOrHostDown(err, true) {
 			return false, errNotConnected
 		}
 		return false, err
@@ -133,7 +140,8 @@ func (target *WebhookTarget) IsActive() (bool, error) {
 	return true, nil
 }
 
-// Save - saves the events to the store if queuestore is configured, which will be replayed when the wenhook connection is active.
+// Save - saves the events to the store if queuestore is configured,
+// which will be replayed when the webhook connection is active.
 func (target *WebhookTarget) Save(eventData event.Event) error {
 	if target.store != nil {
 		return target.store.Put(eventData)

internal/ioutil/ioutil.go

@@ -254,14 +254,14 @@ const DirectioAlignSize = 4096
 // used with DIRECT I/O based file descriptor and it is expected that
 // input writer *os.File not a generic io.Writer. Make sure to have
 // the file opened for writes with syscall.O_DIRECT flag.
-func CopyAligned(w *os.File, r io.Reader, alignedBuf []byte, totalSize int64) (int64, error) {
+func CopyAligned(w io.Writer, r io.Reader, alignedBuf []byte, totalSize int64, file *os.File) (int64, error) {
 	// Writes remaining bytes in the buffer.
-	writeUnaligned := func(w *os.File, buf []byte) (remainingWritten int64, err error) {
+	writeUnaligned := func(w io.Writer, buf []byte) (remainingWritten int64, err error) {
 		// Disable O_DIRECT on fd's on unaligned buffer
 		// perform an amortized Fdatasync(fd) on the fd at
 		// the end, this is performed by the caller before
 		// closing 'w'.
-		if err = disk.DisableDirectIO(w); err != nil {
+		if err = disk.DisableDirectIO(file); err != nil {
 			return remainingWritten, err
 		}
 		// Since w is *os.File io.Copy shall use ReadFrom() call.

internal/kernel/kernel.go

@@ -0,0 +1,130 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build linux
+// +build linux
+
+package kernel
+
+import (
+	"fmt"
+	"io/ioutil"
+	"regexp"
+	"strconv"
+	"strings"
+	"syscall"
+)
+
+var versionRegex = regexp.MustCompile(`^(\d+)\.(\d+).(\d+).*$`)
+
+// VersionFromRelease converts a release string with format
+// 4.4.2[-1] to a kernel version number in LINUX_VERSION_CODE format.
+// That is, for kernel "a.b.c", the version number will be (a<<16 + b<<8 + c)
+func VersionFromRelease(releaseString string) (uint32, error) {
+	versionParts := versionRegex.FindStringSubmatch(releaseString)
+	if len(versionParts) != 4 {
+		return 0, fmt.Errorf("got invalid release version %q (expected format '4.3.2-1')", releaseString)
+	}
+	major, err := strconv.Atoi(versionParts[1])
+	if err != nil {
+		return 0, err
+	}
+
+	minor, err := strconv.Atoi(versionParts[2])
+	if err != nil {
+		return 0, err
+	}
+
+	patch, err := strconv.Atoi(versionParts[3])
+	if err != nil {
+		return 0, err
+	}
+	return Version(major, minor, patch), nil
+}
+
+// Version implements KERNEL_VERSION equivalent macro
+// #define KERNEL_VERSION(a,b,c) (((a) << 16) + ((b) << 8) + ((c) > 255 ? 255 : (c)))
+func Version(major, minor, patch int) uint32 {
+	if patch > 255 {
+		patch = 255
+	}
+	out := major<<16 + minor<<8 + patch
+	return uint32(out)
+}
+
+func currentVersionUname() (uint32, error) {
+	var buf syscall.Utsname
+	if err := syscall.Uname(&buf); err != nil {
+		return 0, err
+	}
+	releaseString := strings.Trim(utsnameStr(buf.Release[:]), "\x00")
+	return VersionFromRelease(releaseString)
+}
+
+func currentVersionUbuntu() (uint32, error) {
+	procVersion, err := ioutil.ReadFile("/proc/version_signature")
+	if err != nil {
+		return 0, err
+	}
+	var u1, u2, releaseString string
+	_, err = fmt.Sscanf(string(procVersion), "%s %s %s", &u1, &u2, &releaseString)
+	if err != nil {
+		return 0, err
+	}
+	return VersionFromRelease(releaseString)
+}
+
+var debianVersionRegex = regexp.MustCompile(`.* SMP Debian (\d+\.\d+.\d+-\d+)(?:\+[[:alnum:]]*)?.*`)
+
+func parseDebianVersion(str string) (uint32, error) {
+	match := debianVersionRegex.FindStringSubmatch(str)
+	if len(match) != 2 {
+		return 0, fmt.Errorf("failed to parse kernel version from /proc/version: %s", str)
+	}
+	return VersionFromRelease(match[1])
+}
+
+func currentVersionDebian() (uint32, error) {
+	procVersion, err := ioutil.ReadFile("/proc/version")
+	if err != nil {
+		return 0, fmt.Errorf("error reading /proc/version: %s", err)
+	}
+
+	return parseDebianVersion(string(procVersion))
+}
+
+// CurrentVersion returns the current kernel version in
+// LINUX_VERSION_CODE format (see VersionFromRelease())
+func CurrentVersion() (uint32, error) {
+	// We need extra checks for Debian and Ubuntu as they modify
+	// the kernel version patch number for compatibility with
+	// out-of-tree modules. Linux perf tools do the same for Ubuntu
+	// systems: https://github.com/torvalds/linux/commit/d18acd15c
+	//
+	// See also:
+	// https://kernel-team.pages.debian.net/kernel-handbook/ch-versions.html
+	// https://wiki.ubuntu.com/Kernel/FAQ
+	version, err := currentVersionUbuntu()
+	if err == nil {
+		return version, nil
+	}
+	version, err = currentVersionDebian()
+	if err == nil {
+		return version, nil
+	}
+	return currentVersionUname()
+}

internal/kernel/kernel_other.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build !linux
+// +build !linux
+
+package kernel
+
+// VersionFromRelease only implemented on Linux.
+func VersionFromRelease(_ string) (uint32, error) {
+	return 0, nil
+}
+
+// Version only implemented on Linux.
+func Version(_, _, _ int) uint32 {
+	return 0
+}
+
+// CurrentVersion only implemented on Linux.
+func CurrentVersion() (uint32, error) {
+	return 0, nil
+}

internal/kernel/kernel_test.go

@@ -0,0 +1,88 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build linux
+// +build linux
+
+package kernel
+
+import "testing"
+
+var testData = []struct {
+	success       bool
+	releaseString string
+	kernelVersion uint32
+}{
+	{true, "4.1.2-3", 262402},
+	{true, "4.8.14-200.fc24.x86_64", 264206},
+	{true, "4.1.2-3foo", 262402},
+	{true, "4.1.2foo-1", 262402},
+	{true, "4.1.2-rkt-v1", 262402},
+	{true, "4.1.2rkt-v1", 262402},
+	{true, "4.1.2-3 foo", 262402},
+	{true, "3.10.0-1062.el7.x86_64", 199168},
+	{true, "3.0.0", 196608},
+	{true, "2.6.32", 132640},
+	{true, "5.13.0-30-generic", 331008},
+	{true, "5.10.0-1052-oem", 330240},
+	{false, "foo 4.1.2-3", 0},
+	{true, "4.1.2", 262402},
+	{false, ".4.1.2", 0},
+	{false, "4.1.", 0},
+	{false, "4.1", 0},
+}
+
+func TestVersionFromRelease(t *testing.T) {
+	for _, test := range testData {
+		version, err := VersionFromRelease(test.releaseString)
+		if err != nil && test.success {
+			t.Errorf("expected %q to success: %s", test.releaseString, err)
+		} else if err == nil && !test.success {
+			t.Errorf("expected %q to fail", test.releaseString)
+		}
+		if version != test.kernelVersion {
+			t.Errorf("expected kernel version %d, got %d", test.kernelVersion, version)
+		}
+	}
+}
+
+func TestParseDebianVersion(t *testing.T) {
+	for _, tc := range []struct {
+		success       bool
+		releaseString string
+		kernelVersion uint32
+	}{
+		// 4.9.168
+		{true, "Linux version 4.9.0-9-amd64 (debian-kernel@lists.debian.org) (gcc version 6.3.0 20170516 (Debian 6.3.0-18+deb9u1) ) #1 SMP Debian 4.9.168-1+deb9u3 (2019-06-16)", 264616},
+		// 4.9.88
+		{true, "Linux ip-10-0-75-49 4.9.0-6-amd64 #1 SMP Debian 4.9.88-1+deb9u1 (2018-05-07) x86_64 GNU/Linux", 264536},
+		// 3.0.4
+		{true, "Linux version 3.16.0-9-amd64 (debian-kernel@lists.debian.org) (gcc version 4.9.2 (Debian 4.9.2-10+deb8u2) ) #1 SMP Debian 3.16.68-1 (2019-05-22)", 200772},
+		// Invalid
+		{false, "Linux version 4.9.125-linuxkit (root@659b6d51c354) (gcc version 6.4.0 (Alpine 6.4.0) ) #1 SMP Fri Sep 7 08:20:28 UTC 2018", 0},
+	} {
+		version, err := parseDebianVersion(tc.releaseString)
+		if err != nil && tc.success {
+			t.Errorf("expected %q to success: %s", tc.releaseString, err)
+		} else if err == nil && !tc.success {
+			t.Errorf("expected %q to fail", tc.releaseString)
+		}
+		if version != tc.kernelVersion {
+			t.Errorf("expected kernel version %d, got %d", tc.kernelVersion, version)
+		}
+	}
+}

internal/kernel/kernel_utsname_int8.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build (linux && 386) || (linux && amd64) || (linux && arm64) || (linux && mips64) || (linux && mips)
+// +build linux,386 linux,amd64 linux,arm64 linux,mips64 linux,mips
+
+package kernel
+
+func utsnameStr(in []int8) string {
+	out := make([]byte, 0, len(in))
+	for i := 0; i < len(in); i++ {
+		if in[i] == 0x00 {
+			break
+		}
+		out = append(out, byte(in[i]))
+	}
+	return string(out)
+}

internal/kernel/kernel_utsname_uint8.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+//go:build (linux && arm) || (linux && ppc64) || (linux && ppc64le) || (linux && s390x)
+// +build linux,arm linux,ppc64 linux,ppc64le linux,s390x
+
+package kernel
+
+func utsnameStr(in []uint8) string {
+	out := make([]byte, 0, len(in))
+	for i := 0; i < len(in); i++ {
+		if in[i] == 0x00 {
+			break
+		}
+		out = append(out, byte(in[i]))
+	}
+	return string(out)
+}