fix: close http body when webhook send (#18487)

Signed-off-by: Shubhendu Ram Tripathi <shubhendu@minio.io>

.github/PULL_REQUEST_TEMPLATE.md

@@ -1,6 +1,6 @@
 ## Community Contribution License
 All community contributions in this pull request are licensed to the project maintainers
-under the terms of the [Apache 2 license] (https://www.apache.org/licenses/LICENSE-2.0). 
+under the terms of the [Apache 2 license](https://www.apache.org/licenses/LICENSE-2.0). 
 By creating this pull request I represent that I have the right to license the 
 contributions to the project maintainers under the Apache 2 license.
 

.github/workflows/vulncheck.yml

@@ -22,7 +22,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v3
         with:
-          go-version: 1.21.3
+          go-version: 1.21.4
           check-latest: true
       - name: Get official govulncheck
         run: go install golang.org/x/vuln/cmd/govulncheck@latest

cmd/api-response.go

@@ -561,6 +561,7 @@ func generateListVersionsResponse(bucket, prefix, marker, versionIDMarker, delim
 				}
 				content.UserMetadata.Set(k, v)
 			}
+			content.UserMetadata.Set("expires", object.Expires.Format(http.TimeFormat))
 			content.Internal = &ObjectInternalInfo{
 				K: object.DataBlocks,
 				M: object.ParityBlocks,
@@ -704,6 +705,7 @@ func generateListObjectsV2Response(bucket, prefix, token, nextToken, startAfter,
 					}
 					content.UserMetadata.Set(k, v)
 				}
+				content.UserMetadata.Set("expires", object.Expires.Format(http.TimeFormat))
 				content.Internal = &ObjectInternalInfo{
 					K: object.DataBlocks,
 					M: object.ParityBlocks,

cmd/batch-handlers.go

@@ -163,7 +163,7 @@ func (r *BatchJobReplicateV1) ReplicateFromSource(ctx context.Context, api Objec
 	}
 	rd, objInfo, _, err := core.GetObject(ctx, srcBucket, srcObject, gopts)
 	if err != nil {
-		return err
+		return ErrorRespToObjectError(err, srcBucket, srcObject, srcObjInfo.VersionID)
 	}
 	defer rd.Close()
 
@@ -226,7 +226,7 @@ func (r *BatchJobReplicateV1) copyWithMultipartfromSource(ctx context.Context, a
 		}
 		rd, objInfo, _, err := c.GetObject(ctx, srcBucket, srcObject, gopts)
 		if err != nil {
-			return err
+			return ErrorRespToObjectError(err, srcBucket, srcObject, srcObjInfo.VersionID)
 		}
 		defer rd.Close()
 
@@ -262,6 +262,9 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 	if err := ri.load(ctx, api, job); err != nil {
 		return err
 	}
+	if ri.Complete {
+		return nil
+	}
 	globalBatchJobsMetrics.save(job.ID, ri)
 
 	delay := job.Replicate.Flags.Retry.Delay
@@ -517,6 +520,10 @@ func toObjectInfo(bucket, object string, objInfo miniogo.ObjectInfo) ObjectInfo
 		oi.UserDefined[xhttp.AmzStorageClass] = objInfo.StorageClass
 	}
 
+	for k, v := range objInfo.UserMetadata {
+		oi.UserDefined[k] = v
+	}
+
 	return oi
 }
 
@@ -837,6 +844,9 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	if err := ri.load(ctx, api, job); err != nil {
 		return err
 	}
+	if ri.Complete {
+		return nil
+	}
 	globalBatchJobsMetrics.save(job.ID, ri)
 	lastObject := ri.Object
 

cmd/batch-rotate.go

@@ -261,6 +261,9 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	if err := ri.load(ctx, api, job); err != nil {
 		return err
 	}
+	if ri.Complete {
+		return nil
+	}
 
 	globalBatchJobsMetrics.save(job.ID, ri)
 	lastObject := ri.Object

cmd/bucket-lifecycle.go

@@ -43,6 +43,7 @@ import (
 	"github.com/minio/minio/internal/s3select"
 	"github.com/minio/pkg/v2/env"
 	"github.com/minio/pkg/v2/workers"
+	"github.com/zeebo/xxh3"
 )
 
 const (
@@ -223,11 +224,19 @@ type transitionState struct {
 	lastDayStats map[string]*lastDayTierStats
 }
 
-func (t *transitionState) queueTransitionTask(oi ObjectInfo, event lifecycle.Event, src lcEventSrc) {
-	select {
-	case <-t.ctx.Done():
-	case t.transitionCh <- transitionTask{objInfo: oi, event: event, src: src}:
-	default:
+func (t *transitionState) queueTransitionTask(oi ObjectInfo, event lifecycle.Event, src lcEventSrc, blocking bool) {
+	task := transitionTask{objInfo: oi, event: event, src: src}
+	if blocking {
+		select {
+		case <-t.ctx.Done():
+		case t.transitionCh <- task:
+		}
+	} else {
+		select {
+		case <-t.ctx.Done():
+		case t.transitionCh <- task:
+		default:
+		}
 	}
 }
 
@@ -237,7 +246,7 @@ var globalTransitionState *transitionState
 // via its Init method.
 func newTransitionState(ctx context.Context) *transitionState {
 	return &transitionState{
-		transitionCh: make(chan transitionTask, 10000),
+		transitionCh: make(chan transitionTask, 100000),
 		ctx:          ctx,
 		killCh:       make(chan struct{}),
 		lastDayStats: make(map[string]*lastDayTierStats),
@@ -367,7 +376,7 @@ func enqueueTransitionImmediate(obj ObjectInfo, src lcEventSrc) {
 	if lc, err := globalLifecycleSys.Get(obj.Bucket); err == nil {
 		switch event := lc.Eval(obj.ToLifecycleOpts()); event.Action {
 		case lifecycle.TransitionAction, lifecycle.TransitionVersionAction:
-			globalTransitionState.queueTransitionTask(obj, event, src)
+			globalTransitionState.queueTransitionTask(obj, event, src, true)
 		}
 	}
 }
@@ -444,7 +453,8 @@ func genTransitionObjName(bucket string) (string, error) {
 		return "", err
 	}
 	us := u.String()
-	obj := fmt.Sprintf("%s/%s/%s/%s/%s", globalDeploymentID(), bucket, us[0:2], us[2:4], us)
+	hash := xxh3.HashString(pathJoin(globalDeploymentID(), bucket))
+	obj := fmt.Sprintf("%s/%s/%s/%s", strconv.FormatUint(hash, 16), us[0:2], us[2:4], us)
 	return obj, nil
 }
 

cmd/bucket-metadata-sys.go

@@ -437,21 +437,8 @@ func (sys *BucketMetadataSys) Init(ctx context.Context, buckets []BucketInfo, ob
 	return nil
 }
 
-func (sys *BucketMetadataSys) loadBucketMetadata(ctx context.Context, bucket BucketInfo) error {
-	meta, err := loadBucketMetadata(ctx, sys.objAPI, bucket.Name)
-	if err != nil {
-		return err
-	}
-
-	sys.Lock()
-	sys.metadataMap[bucket.Name] = meta
-	sys.Unlock()
-
-	return nil
-}
-
 // concurrently load bucket metadata to speed up loading bucket metadata.
-func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []BucketInfo) {
+func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []BucketInfo, failedBuckets map[string]struct{}) {
 	g := errgroup.WithNErrs(len(buckets))
 	bucketMetas := make([]BucketMetadata, len(buckets))
 	for index := range buckets {
@@ -491,6 +478,10 @@ func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []Buck
 
 	for i, meta := range bucketMetas {
 		if errs[i] != nil {
+			if failedBuckets == nil {
+				failedBuckets = make(map[string]struct{})
+			}
+			failedBuckets[buckets[i].Name] = struct{}{}
 			continue
 		}
 		globalEventNotifier.set(buckets[i], meta)   // set notification targets
@@ -498,7 +489,7 @@ func (sys *BucketMetadataSys) concurrentLoad(ctx context.Context, buckets []Buck
 	}
 }
 
-func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context) {
+func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context, failedBuckets map[string]struct{}) {
 	const bucketMetadataRefresh = 15 * time.Minute
 
 	sleeper := newDynamicSleeper(2, 150*time.Millisecond, false)
@@ -513,7 +504,7 @@ func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context) {
 			buckets, err := sys.objAPI.ListBuckets(ctx, BucketOptions{})
 			if err != nil {
 				logger.LogIf(ctx, err)
-				continue
+				break
 			}
 
 			// Handle if we have some buckets in-memory those are stale.
@@ -525,38 +516,49 @@ func (sys *BucketMetadataSys) refreshBucketsMetadataLoop(ctx context.Context) {
 			}
 			sys.RemoveStaleBuckets(diskBuckets)
 
-			for _, bucket := range buckets {
+			for i := range buckets {
 				wait := sleeper.Timer(ctx)
 
-				err := sys.loadBucketMetadata(ctx, bucket)
+				meta, err := loadBucketMetadata(ctx, sys.objAPI, buckets[i].Name)
 				if err != nil {
 					logger.LogIf(ctx, err)
 					wait() // wait to proceed to next entry.
 					continue
 				}
 
+				sys.Lock()
+				sys.metadataMap[buckets[i].Name] = meta
+				sys.Unlock()
+
+				// Initialize the failed buckets
+				if _, ok := failedBuckets[buckets[i].Name]; ok {
+					globalEventNotifier.set(buckets[i], meta)
+					globalBucketTargetSys.set(buckets[i], meta)
+					delete(failedBuckets, buckets[i].Name)
+				}
+
 				wait() // wait to proceed to next entry.
 			}
-
-			t.Reset(bucketMetadataRefresh)
 		}
+		t.Reset(bucketMetadataRefresh)
 	}
 }
 
 // Loads bucket metadata for all buckets into BucketMetadataSys.
 func (sys *BucketMetadataSys) init(ctx context.Context, buckets []BucketInfo) {
 	count := 100 // load 100 bucket metadata at a time.
+	failedBuckets := make(map[string]struct{})
 	for {
 		if len(buckets) < count {
-			sys.concurrentLoad(ctx, buckets)
+			sys.concurrentLoad(ctx, buckets, failedBuckets)
 			break
 		}
-		sys.concurrentLoad(ctx, buckets[:count])
+		sys.concurrentLoad(ctx, buckets[:count], failedBuckets)
 		buckets = buckets[count:]
 	}
 
 	if globalIsDistErasure {
-		go sys.refreshBucketsMetadataLoop(ctx)
+		go sys.refreshBucketsMetadataLoop(ctx, failedBuckets)
 	}
 }
 

cmd/bucket-replication.go

@@ -777,6 +777,7 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put
 		UserMetadata:    meta,
 		ContentType:     objInfo.ContentType,
 		ContentEncoding: objInfo.ContentEncoding,
+		Expires:         objInfo.Expires,
 		StorageClass:    sc,
 		Internal: minio.AdvancedPutOptions{
 			SourceVersionID:    objInfo.VersionID,
@@ -3187,12 +3188,16 @@ func (p *ReplicationPool) queueMRFSave(entry MRFReplicateEntry) {
 	}
 }
 
-func (p *ReplicationPool) persistToDrive(ctx context.Context, v MRFReplicateEntries, data []byte) {
+func (p *ReplicationPool) persistToDrive(ctx context.Context, v MRFReplicateEntries) {
 	newReader := func() io.ReadCloser {
 		r, w := io.Pipe()
 		go func() {
+			// Initialize MRF meta header.
+			var data [4]byte
+			binary.LittleEndian.PutUint16(data[0:2], mrfMetaFormat)
+			binary.LittleEndian.PutUint16(data[2:4], mrfMetaVersion)
 			mw := msgp.NewWriter(w)
-			n, err := mw.Write(data)
+			n, err := mw.Write(data[:])
 			if err != nil {
 				w.CloseWithError(err)
 				return
@@ -3230,15 +3235,10 @@ func (p *ReplicationPool) saveMRFEntries(ctx context.Context, entries map[string
 
 	v := MRFReplicateEntries{
 		Entries: entries,
-		Version: mrfMetaVersionV1,
+		Version: mrfMetaVersion,
 	}
-	data := make([]byte, 4, v.Msgsize()+4)
 
-	// Initialize the resync meta header.
-	binary.LittleEndian.PutUint16(data[0:2], mrfMetaFormat)
-	binary.LittleEndian.PutUint16(data[2:4], mrfMetaVersion)
-
-	p.persistToDrive(ctx, v, data)
+	p.persistToDrive(ctx, v)
 }
 
 // load mrf entries from disk

cmd/data-scanner.go

@@ -1173,7 +1173,7 @@ func applyTransitionRule(event lifecycle.Event, src lcEventSrc, obj ObjectInfo)
 	if obj.DeleteMarker {
 		return false
 	}
-	globalTransitionState.queueTransitionTask(obj, event, src)
+	globalTransitionState.queueTransitionTask(obj, event, src, false)
 	return true
 }
 

cmd/erasure-object.go

@@ -666,6 +666,32 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r
 	return metaFileInfos, errs
 }
 
+// Checking if an object is dangling costs some IOPS; hence implementing this function
+// which decides which condition it is useful to check if an object is dangling
+//
+//	  errs: errors from reading xl.meta in all disks
+//	   err: reduced errs
+//	bucket: the object name in question
+func shouldCheckForDangling(err error, errs []error, bucket string) bool {
+	// Avoid data in .minio.sys for now
+	if bucket == minioMetaBucket {
+		return false
+	}
+	switch {
+	// Check if we have a read quorum issue
+	case errors.Is(err, errErasureReadQuorum):
+		return true
+	// Check if the object is inexistent in most disks but not all of them
+	case errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound):
+		for i := range errs {
+			if errs[i] == nil {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object string, opts ObjectOptions, readData bool) (fi FileInfo, metaArr []FileInfo, onlineDisks []StorageAPI, err error) {
 	disks := er.getDisks()
 
@@ -680,7 +706,7 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
 
 	readQuorum, _, err := objectQuorumFromMeta(ctx, metaArr, errs, er.defaultParityCount)
 	if err != nil {
-		if errors.Is(err, errErasureReadQuorum) && !strings.HasPrefix(bucket, minioMetaBucket) {
+		if shouldCheckForDangling(err, errs, bucket) {
 			_, derr := er.deleteIfDangling(ctx, bucket, object, metaArr, errs, nil, opts)
 			if derr != nil {
 				err = derr
@@ -690,7 +716,7 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s
 	}
 
 	if reducedErr := reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, readQuorum); reducedErr != nil {
-		if errors.Is(reducedErr, errErasureReadQuorum) && !strings.HasPrefix(bucket, minioMetaBucket) {
+		if shouldCheckForDangling(err, errs, bucket) {
 			_, derr := er.deleteIfDangling(ctx, bucket, object, metaArr, errs, nil, opts)
 			if derr != nil {
 				reducedErr = derr

cmd/erasure-server-pool-decom.go

@@ -701,10 +701,12 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec
 		return fmt.Errorf("no online drives found for set with endpoints %s", set.getEndpoints())
 	}
 
+	listQuorum := (len(disks) + 1) / 2
+
 	// How to resolve partial results.
 	resolver := metadataResolutionParams{
-		dirQuorum: len(disks) / 2, // make sure to capture all quorum ratios
-		objQuorum: len(disks) / 2, // make sure to capture all quorum ratios
+		dirQuorum: listQuorum, // make sure to capture all quorum ratios
+		objQuorum: listQuorum, // make sure to capture all quorum ratios
 		bucket:    bi.Name,
 	}
 
@@ -714,7 +716,7 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec
 		path:           bi.Prefix,
 		recursive:      true,
 		forwardTo:      "",
-		minDisks:       len(disks) / 2, // to capture all quorum ratios
+		minDisks:       len(disks) / 2,
 		reportNotFound: false,
 		agreed:         fn,
 		partial: func(entries metaCacheEntries, _ []error) {

cmd/erasure-server-pool.go

@@ -970,7 +970,7 @@ func (z *erasureServerPools) GetObjectInfo(ctx context.Context, bucket, object s
 // PutObject - writes an object to least used erasure pool.
 func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, object string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
 	// Validate put object input args.
-	if err := checkPutObjectArgs(ctx, bucket, object, z); err != nil {
+	if err := checkPutObjectArgs(ctx, bucket, object); err != nil {
 		return ObjectInfo{}, err
 	}
 
@@ -2253,11 +2253,12 @@ type HealthOptions struct {
 // additionally with any specific heuristic information which
 // was queried
 type HealthResult struct {
-	Healthy        bool
-	HealingDrives  int
-	UnhealthyPools []struct {
+	Healthy       bool
+	HealingDrives int
+	ESHealth      []struct {
 		Maintenance   bool
 		PoolID, SetID int
+		HealthyDrives int
 		WriteQuorum   int
 	}
 	WriteQuorum   int
@@ -2372,50 +2373,40 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 	}
 
 	result := HealthResult{
-		HealingDrives: len(aggHealStateResult.HealDisks),
+		Healthy:       true,
 		WriteQuorum:   maximumWriteQuorum,
 		UsingDefaults: usingDefaults, // indicates if config was not initialized and we are using defaults on this node.
 	}
 
 	for poolIdx := range erasureSetUpCount {
 		for setIdx := range erasureSetUpCount[poolIdx] {
+			result.ESHealth = append(result.ESHealth, struct {
+				Maintenance                bool
+				PoolID, SetID              int
+				HealthyDrives, WriteQuorum int
+			}{
+				Maintenance:   opts.Maintenance,
+				SetID:         setIdx,
+				PoolID:        poolIdx,
+				HealthyDrives: erasureSetUpCount[poolIdx][setIdx],
+				WriteQuorum:   poolWriteQuorums[poolIdx],
+			})
+
 			if erasureSetUpCount[poolIdx][setIdx] < poolWriteQuorums[poolIdx] {
 				logger.LogIf(logger.SetReqInfo(ctx, reqInfo),
 					fmt.Errorf("Write quorum may be lost on pool: %d, set: %d, expected write quorum: %d",
 						poolIdx, setIdx, poolWriteQuorums[poolIdx]))
-				result.UnhealthyPools = append(result.UnhealthyPools, struct {
-					Maintenance                bool
-					PoolID, SetID, WriteQuorum int
-				}{
-					Maintenance: opts.Maintenance,
-					SetID:       setIdx,
-					PoolID:      poolIdx,
-					WriteQuorum: poolWriteQuorums[poolIdx],
-				})
+				result.Healthy = false
 			}
 		}
-		if len(result.UnhealthyPools) > 0 {
-			// We have unhealthy pools return error.
-			return result
-		}
 	}
 
-	// when maintenance is not specified we don't have
-	// to look at the healing side of the code.
-	if !opts.Maintenance {
-		return HealthResult{
-			Healthy:       true,
-			WriteQuorum:   maximumWriteQuorum,
-			UsingDefaults: usingDefaults, // indicates if config was not initialized and we are using defaults on this node.
-		}
+	if opts.Maintenance {
+		result.Healthy = result.Healthy && len(aggHealStateResult.HealDisks) == 0
+		result.HealingDrives = len(aggHealStateResult.HealDisks)
 	}
 
-	return HealthResult{
-		Healthy:       len(aggHealStateResult.HealDisks) == 0,
-		HealingDrives: len(aggHealStateResult.HealDisks),
-		WriteQuorum:   maximumWriteQuorum,
-		UsingDefaults: usingDefaults, // indicates if config was not initialized and we are using defaults on this node.
-	}
+	return result
 }
 
 // PutObjectMetadata - replace or add tags to an existing object

cmd/erasure-sets.go

@@ -217,6 +217,10 @@ func (s *erasureSets) connectDisks() {
 				continue
 			}
 		}
+		if cdisk != nil {
+			// Close previous offline disk.
+			cdisk.Close()
+		}
 
 		wg.Add(1)
 		go func(endpoint Endpoint) {

cmd/event-notification.go

@@ -128,40 +128,6 @@ func (evnot *EventNotifier) AddRulesMap(bucketName string, rulesMap event.RulesM
 	}
 }
 
-// RemoveRulesMap - removes rules map for bucket name.
-func (evnot *EventNotifier) RemoveRulesMap(bucketName string, rulesMap event.RulesMap) {
-	evnot.Lock()
-	defer evnot.Unlock()
-
-	evnot.bucketRulesMap[bucketName].Remove(rulesMap)
-	if len(evnot.bucketRulesMap[bucketName]) == 0 {
-		delete(evnot.bucketRulesMap, bucketName)
-	}
-}
-
-// ConfiguredTargetIDs - returns list of configured target id's
-func (evnot *EventNotifier) ConfiguredTargetIDs() []event.TargetID {
-	if evnot == nil {
-		return nil
-	}
-
-	evnot.RLock()
-	defer evnot.RUnlock()
-
-	var targetIDs []event.TargetID
-	for _, rmap := range evnot.bucketRulesMap {
-		for _, rules := range rmap {
-			for _, targetSet := range rules {
-				for id := range targetSet {
-					targetIDs = append(targetIDs, id)
-				}
-			}
-		}
-	}
-
-	return targetIDs
-}
-
 // RemoveNotification - removes all notification configuration for bucket name.
 func (evnot *EventNotifier) RemoveNotification(bucketName string) {
 	evnot.Lock()

cmd/ftp-server.go

@@ -18,20 +18,14 @@
 package cmd
 
 import (
-	"context"
-	"crypto/subtle"
 	"fmt"
 	"net"
-	"os"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/minio/cli"
 	"github.com/minio/minio/internal/logger"
-	"github.com/pkg/sftp"
 	ftp "goftp.io/server/v2"
-	"golang.org/x/crypto/ssh"
 )
 
 var globalRemoteFTPClientTransport = NewRemoteTargetHTTPTransport(true)()
@@ -75,196 +69,6 @@ func (log *minioLogger) PrintResponse(sessionID string, code int, message string
 	}
 }
 
-func acceptSFTPConnection(conn net.Conn, config *ssh.ServerConfig) {
-	// For SSH handshake keep a 2 minute deadline, as OpensSSH default.
-	conn.SetDeadline(time.Now().Add(2 * time.Minute))
-
-	// Before use, a handshake must be performed on the incoming net.Conn.
-	sconn, chans, reqs, err := ssh.NewServerConn(conn, config)
-	if err != nil {
-		return
-	}
-
-	// Once we are done with SSH handshake, remove deadline.
-	conn.SetDeadline(time.Time{})
-
-	// The incoming Request channel must be serviced.
-	go ssh.DiscardRequests(reqs)
-
-	// Service the incoming Channel channel.
-	for newChannel := range chans {
-		// Channels have a type, depending on the application level
-		// protocol intended. In the case of an SFTP session, this is "subsystem"
-		// with a payload string of "<length=4>sftp"
-		if newChannel.ChannelType() != "session" {
-			newChannel.Reject(ssh.UnknownChannelType, "unknown channel type")
-			continue
-		}
-
-		channel, requests, err := newChannel.Accept()
-		if err != nil {
-			logger.LogOnceIf(context.Background(), fmt.Errorf("unable to accept the connection requests channel: %w", err), "accept-channel-sftp")
-			continue
-		}
-
-		// Sessions have out-of-band requests such as "shell",
-		// "pty-req" and "env".  Here we handle only the
-		// "subsystem" request.
-		go func(in <-chan *ssh.Request) {
-			for req := range in {
-				ok := false
-				if req.Type == "subsystem" {
-					if len(req.Payload) > 4 && string(req.Payload[4:]) == "sftp" {
-						ok = true
-						go handleSFTPConnection(channel, sconn)
-					}
-				}
-
-				if req.WantReply {
-					// We only reply to SSH packets that have `sftp` payload, all other
-					// packets are rejected
-					req.Reply(ok, nil)
-				}
-			}
-		}(requests)
-	}
-}
-
-func handleSFTPConnection(channel ssh.Channel, sconn *ssh.ServerConn) {
-	// Create the server instance for the channel using the handler we created above.
-	server := sftp.NewRequestServer(channel, NewSFTPDriver(sconn.Permissions), sftp.WithRSAllocator())
-	defer server.Close()
-
-	server.Serve()
-}
-
-func startSFTPServer(c *cli.Context) {
-	args := c.StringSlice("sftp")
-
-	var (
-		port          int
-		publicIP      string
-		sshPrivateKey string
-	)
-
-	var err error
-	for _, arg := range args {
-		tokens := strings.SplitN(arg, "=", 2)
-		if len(tokens) != 2 {
-			logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s", arg), "unable to start SFTP server")
-		}
-		switch tokens[0] {
-		case "address":
-			host, portStr, err := net.SplitHostPort(tokens[1])
-			if err != nil {
-				logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s (%v)", arg, err), "unable to start SFTP server")
-			}
-			port, err = strconv.Atoi(portStr)
-			if err != nil {
-				logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s (%v)", arg, err), "unable to start SFTP server")
-			}
-			if port < 1 || port > 65535 {
-				logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s, (port number must be between 1 to 65535)", arg), "unable to start SFTP server")
-			}
-			publicIP = host
-		case "ssh-private-key":
-			sshPrivateKey = tokens[1]
-		}
-	}
-
-	if port == 0 {
-		port = 8022 // Default SFTP port, since no port was given.
-	}
-
-	if sshPrivateKey == "" {
-		logger.Fatal(fmt.Errorf("invalid arguments passed, private key file is mandatory for --sftp='ssh-private-key=path/to/id_ecdsa'"), "unable to start SFTP server")
-	}
-
-	privateBytes, err := os.ReadFile(sshPrivateKey)
-	if err != nil {
-		logger.Fatal(fmt.Errorf("invalid arguments passed, private key file is not accessible: %v", err), "unable to start SFTP server")
-	}
-
-	private, err := ssh.ParsePrivateKey(privateBytes)
-	if err != nil {
-		logger.Fatal(fmt.Errorf("invalid arguments passed, private key file is not parseable: %v", err), "unable to start SFTP server")
-	}
-
-	// An SSH server is represented by a ServerConfig, which holds
-	// certificate details and handles authentication of ServerConns.
-	config := &ssh.ServerConfig{
-		PasswordCallback: func(c ssh.ConnMetadata, pass []byte) (*ssh.Permissions, error) {
-			if globalIAMSys.LDAPConfig.Enabled() {
-				targetUser, targetGroups, err := globalIAMSys.LDAPConfig.Bind(c.User(), string(pass))
-				if err != nil {
-					return nil, err
-				}
-				ldapPolicies, _ := globalIAMSys.PolicyDBGet(targetUser, false, targetGroups...)
-				if len(ldapPolicies) == 0 {
-					return nil, errAuthentication
-				}
-				return &ssh.Permissions{
-					CriticalOptions: map[string]string{
-						ldapUser:  targetUser,
-						ldapUserN: c.User(),
-					},
-					Extensions: make(map[string]string),
-				}, nil
-			}
-
-			ui, ok := globalIAMSys.GetUser(context.Background(), c.User())
-			if !ok {
-				return nil, errNoSuchUser
-			}
-
-			if subtle.ConstantTimeCompare([]byte(ui.Credentials.SecretKey), pass) == 1 {
-				return &ssh.Permissions{
-					CriticalOptions: map[string]string{
-						"accessKey": c.User(),
-					},
-					Extensions: make(map[string]string),
-				}, nil
-			}
-			return nil, errAuthentication
-		},
-	}
-
-	config.AddHostKey(private)
-
-	// Once a ServerConfig has been configured, connections can be accepted.
-	listener, err := net.Listen("tcp", net.JoinHostPort(publicIP, strconv.Itoa(port)))
-	if err != nil {
-		logger.Fatal(err, "unable to start listening on --sftp='port=%d'", port)
-	}
-
-	logger.Info(fmt.Sprintf("MinIO SFTP Server listening on %s", net.JoinHostPort(publicIP, strconv.Itoa(port))))
-
-	var tempDelay time.Duration // how long to sleep on accept failure
-
-	for {
-		conn, err := listener.Accept()
-		if err != nil {
-			// From https://github.com/golang/go/blob/4aa1efed4853ea067d665a952eee77c52faac774/src/net/http/server.go#L3046
-			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				if tempDelay == 0 {
-					tempDelay = 5 * time.Millisecond
-				} else {
-					tempDelay *= 2
-				}
-				if max := 1 * time.Second; tempDelay > max {
-					tempDelay = max
-				}
-				logger.LogOnceIf(context.Background(), fmt.Errorf("error while accepting connections: %w, retrying in %s", err, tempDelay), "accept-limit-sftp")
-				time.Sleep(tempDelay)
-				continue
-			}
-			logger.Fatal(err, "unrecoverable error while accepting new connections")
-		}
-
-		go acceptSFTPConnection(conn, config)
-	}
-}
-
 func startFTPServer(c *cli.Context) {
 	args := c.StringSlice("ftp")
 

cmd/healthcheck-handler.go

@@ -90,21 +90,15 @@ func ClusterReadCheckHandler(w http.ResponseWriter, r *http.Request) {
 	writeResponse(w, http.StatusOK, nil, mimeNone)
 }
 
-// ReadinessCheckHandler Checks if the process is up. Always returns success.
+// ReadinessCheckHandler checks whether MinIO is up and ready to serve requests.
+// It also checks whether the KMS is available and whether etcd is reachable,
+// if configured.
 func ReadinessCheckHandler(w http.ResponseWriter, r *http.Request) {
-	LivenessCheckHandler(w, r)
-}
-
-// LivenessCheckHandler - Checks if the process is up. Always returns success.
-func LivenessCheckHandler(w http.ResponseWriter, r *http.Request) {
-	objLayer := newObjectLayerFn()
-	if objLayer == nil {
-		// Service not initialized yet
-		w.Header().Set(xhttp.MinIOServerStatus, unavailable)
+	if objLayer := newObjectLayerFn(); objLayer == nil {
+		w.Header().Set(xhttp.MinIOServerStatus, unavailable) // Service not initialized yet
 	}
-
-	peerCall := r.Header.Get(xhttp.MinIOPeerCall) != ""
-	if peerCall {
+	if r.Header.Get(xhttp.MinIOPeerCall) != "" {
+		writeResponse(w, http.StatusOK, nil, mimeNone)
 		return
 	}
 
@@ -120,7 +114,7 @@ func LivenessCheckHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// Verify if KMS is reachable if its configured
-	if GlobalKMS != nil && !peerCall {
+	if GlobalKMS != nil {
 		ctx, cancel := context.WithTimeout(r.Context(), time.Minute)
 		defer cancel()
 
@@ -153,6 +147,30 @@ func LivenessCheckHandler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	}
-
+	writeResponse(w, http.StatusOK, nil, mimeNone)
+}
+
+// LivenessCheckHandler checks whether MinIO is up. It differs from the
+// readiness handler since a failing liveness check causes pod restarts
+// in K8S enviromnents. Therefore, it does not contact external systems.
+func LivenessCheckHandler(w http.ResponseWriter, r *http.Request) {
+	if objLayer := newObjectLayerFn(); objLayer == nil {
+		w.Header().Set(xhttp.MinIOServerStatus, unavailable) // Service not initialized yet
+	}
+	if r.Header.Get(xhttp.MinIOPeerCall) != "" {
+		writeResponse(w, http.StatusOK, nil, mimeNone)
+		return
+	}
+
+	if int(globalHTTPStats.loadRequestsInQueue()) > globalAPIConfig.getRequestsPoolCapacity() {
+		apiErr := getAPIError(ErrBusy)
+		switch r.Method {
+		case http.MethodHead:
+			writeResponse(w, apiErr.HTTPStatusCode, nil, mimeNone)
+		case http.MethodGet:
+			writeErrorResponse(r.Context(), w, apiErr, r.URL)
+		}
+		return
+	}
 	writeResponse(w, http.StatusOK, nil, mimeNone)
 }

cmd/iam.go

@@ -1663,18 +1663,25 @@ func (sys *IAMSys) PolicyDBUpdateLDAP(ctx context.Context, isAttach bool,
 			return
 		}
 		if dn == "" {
-			err = errNoSuchUser
-			return
+			// Still attempt to detach if provided user is a DN.
+			if !isAttach && sys.LDAPConfig.IsLDAPUserDN(r.User) {
+				dn = r.User
+			} else {
+				err = errNoSuchUser
+				return
+			}
 		}
 		isGroup = false
 	} else {
-		var exists bool
-		if exists, err = sys.LDAPConfig.DoesGroupDNExist(r.Group); err != nil {
-			logger.LogIf(ctx, err)
-			return
-		} else if !exists {
-			err = errNoSuchGroup
-			return
+		if isAttach {
+			var exists bool
+			if exists, err = sys.LDAPConfig.DoesGroupDNExist(r.Group); err != nil {
+				logger.LogIf(ctx, err)
+				return
+			} else if !exists {
+				err = errNoSuchGroup
+				return
+			}
 		}
 		dn = r.Group
 		isGroup = true

cmd/metrics-v2.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"net/http"
 	"runtime"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -3187,6 +3188,16 @@ func getClusterHealthStatusMD() MetricDescription {
 	}
 }
 
+func getClusterErasureSetToleranceMD() MetricDescription {
+	return MetricDescription{
+		Namespace: clusterMetricNamespace,
+		Subsystem: "health",
+		Name:      "erasure_set_tolerance",
+		Help:      "Get erasure set tolerance status",
+		Type:      gaugeMetric,
+	}
+}
+
 func getClusterHealthMetrics() *MetricsGroup {
 	mg := &MetricsGroup{
 		cacheInterval: 10 * time.Second,
@@ -3218,6 +3229,18 @@ func getClusterHealthMetrics() *MetricsGroup {
 			Value:       float64(health),
 		})
 
+		for _, h := range result.ESHealth {
+			labels := map[string]string{
+				"pool": strconv.Itoa(h.PoolID),
+				"set":  strconv.Itoa(h.SetID),
+			}
+			metrics = append(metrics, Metric{
+				Description:    getClusterErasureSetToleranceMD(),
+				VariableLabels: labels,
+				Value:          float64(h.HealthyDrives - h.WriteQuorum),
+			})
+		}
+
 		return
 	})
 

cmd/object-api-input-checks.go

@@ -149,7 +149,6 @@ func checkObjectArgs(ctx context.Context, bucket, object string, obj ObjectLayer
 	// This is done on purpose since the order of errors is
 	// important here bucket does not exist error should
 	// happen before we return an error for invalid object name.
-	// FIXME: should be moved to handler layer.
 	if err := checkBucketExist(ctx, bucket, obj); err != nil {
 		return err
 	}
@@ -170,16 +169,7 @@ func checkObjectArgs(ctx context.Context, bucket, object string, obj ObjectLayer
 }
 
 // Checks for PutObject arguments validity, also validates if bucket exists.
-func checkPutObjectArgs(ctx context.Context, bucket, object string, obj getBucketInfoI) error {
-	// Verify if bucket exists before validating object name.
-	// This is done on purpose since the order of errors is
-	// important here bucket does not exist error should
-	// happen before we return an error for invalid object name.
-	// FIXME: should be moved to handler layer.
-	if err := checkBucketExist(ctx, bucket, obj); err != nil {
-		return err
-	}
-
+func checkPutObjectArgs(ctx context.Context, bucket, object string) error {
 	if err := checkObjectNameForLengthAndSlash(bucket, object); err != nil {
 		return err
 	}

cmd/object-api-utils.go

@@ -1198,7 +1198,7 @@ func hasSpaceFor(di []*DiskInfo, size int64) (bool, error) {
 		if disk == nil || disk.Total == 0 {
 			continue
 		}
-		if disk.FreeInodes < diskMinInodes && disk.UsedInodes > 0 {
+		if !globalIsErasureSD && disk.FreeInodes < diskMinInodes && disk.UsedInodes > 0 {
 			// We have an inode count, but not enough inodes.
 			return false, nil
 		}

cmd/sftp-server-driver.go

@@ -201,19 +201,74 @@ func (f *sftpDriver) Fileread(r *sftp.Request) (ra io.ReaderAt, err error) {
 	return obj, nil
 }
 
-type writerAt struct {
-	w  *io.PipeWriter
-	wg *sync.WaitGroup
+// TransferError will catch network errors during transfer.
+// When TransferError() is called Close() will also
+// be called, so we do not need to Wait() here.
+func (w *writerAt) TransferError(err error) {
+	_ = w.w.CloseWithError(err)
+	_ = w.r.CloseWithError(err)
+	w.err = err
 }
 
-func (w *writerAt) Close() error {
-	err := w.w.Close()
+func (w *writerAt) Close() (err error) {
+	switch {
+	case len(w.buffer) > 0:
+		err = errors.New("some file segments were not flushed from the queue")
+		_ = w.w.CloseWithError(err)
+	case w.err != nil:
+		// No need to close here since both pipes were
+		// closing inside TransferError()
+		err = w.err
+	default:
+		err = w.w.Close()
+	}
+	for i := range w.buffer {
+		delete(w.buffer, i)
+	}
 	w.wg.Wait()
 	return err
 }
 
+type writerAt struct {
+	w      *io.PipeWriter
+	r      *io.PipeReader
+	wg     *sync.WaitGroup
+	buffer map[int64][]byte
+	err    error
+
+	nextOffset int64
+	m          sync.Mutex
+}
+
 func (w *writerAt) WriteAt(b []byte, offset int64) (n int, err error) {
-	return w.w.Write(b)
+	w.m.Lock()
+	defer w.m.Unlock()
+
+	if w.nextOffset == offset {
+		n, err = w.w.Write(b)
+		w.nextOffset += int64(n)
+	} else {
+		w.buffer[offset] = make([]byte, len(b))
+		copy(w.buffer[offset], b)
+		n = len(b)
+	}
+
+again:
+	nextOut, ok := w.buffer[w.nextOffset]
+	if ok {
+		n, err = w.w.Write(nextOut)
+		delete(w.buffer, w.nextOffset)
+		w.nextOffset += int64(n)
+		if n != len(nextOut) {
+			return 0, fmt.Errorf("expected write size %d but wrote %d bytes", len(nextOut), n)
+		}
+		if err != nil {
+			return 0, err
+		}
+		goto again
+	}
+
+	return len(b), nil
 }
 
 func (f *sftpDriver) Filewrite(r *sftp.Request) (w io.WriterAt, err error) {
@@ -238,7 +293,12 @@ func (f *sftpDriver) Filewrite(r *sftp.Request) (w io.WriterAt, err error) {
 
 	pr, pw := io.Pipe()
 
-	wa := &writerAt{w: pw, wg: &sync.WaitGroup{}}
+	wa := &writerAt{
+		buffer: make(map[int64][]byte),
+		w:      pw,
+		r:      pr,
+		wg:     &sync.WaitGroup{},
+	}
 	wa.wg.Add(1)
 	go func() {
 		_, err := clnt.PutObject(r.Context(), bucket, object, pr, -1, minio.PutObjectOptions{SendContentMd5: true})
@@ -298,6 +358,7 @@ func (f *sftpDriver) Filecmd(r *sftp.Request) (err error) {
 				return err.Err
 			}
 		}
+		return err
 
 	case "Remove":
 		bucket, object := path2BucketObject(r.Filepath)

cmd/sftp-server.go

@@ -0,0 +1,172 @@
+// Copyright (c) 2015-2023 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"
+	"crypto/subtle"
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/minio/cli"
+	"github.com/minio/minio/internal/logger"
+	xsftp "github.com/minio/pkg/v2/sftp"
+	"github.com/pkg/sftp"
+	"golang.org/x/crypto/ssh"
+)
+
+type sftpLogger struct{}
+
+func (s *sftpLogger) Info(tag xsftp.LogType, msg string) {
+	logger.Info(msg)
+}
+
+func (s *sftpLogger) Error(tag xsftp.LogType, err error) {
+	switch tag {
+	case xsftp.AcceptNetworkError:
+		logger.LogOnceIf(context.Background(), err, "accept-limit-sftp")
+	case xsftp.AcceptChannelError:
+		logger.LogOnceIf(context.Background(), err, "accept-channel-sftp")
+	case xsftp.SSHKeyExchangeError:
+		logger.LogOnceIf(context.Background(), err, "key-exchange-sftp")
+	default:
+		logger.LogOnceIf(context.Background(), err, "unknown-error-sftp")
+	}
+}
+
+func startSFTPServer(c *cli.Context) {
+	args := c.StringSlice("sftp")
+
+	var (
+		port          int
+		publicIP      string
+		sshPrivateKey string
+	)
+
+	var err error
+	for _, arg := range args {
+		tokens := strings.SplitN(arg, "=", 2)
+		if len(tokens) != 2 {
+			logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s", arg), "unable to start SFTP server")
+		}
+		switch tokens[0] {
+		case "address":
+			host, portStr, err := net.SplitHostPort(tokens[1])
+			if err != nil {
+				logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s (%v)", arg, err), "unable to start SFTP server")
+			}
+			port, err = strconv.Atoi(portStr)
+			if err != nil {
+				logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s (%v)", arg, err), "unable to start SFTP server")
+			}
+			if port < 1 || port > 65535 {
+				logger.Fatal(fmt.Errorf("invalid arguments passed to --sftp=%s, (port number must be between 1 to 65535)", arg), "unable to start SFTP server")
+			}
+			publicIP = host
+		case "ssh-private-key":
+			sshPrivateKey = tokens[1]
+		}
+	}
+
+	if port == 0 {
+		port = 8022 // Default SFTP port, since no port was given.
+	}
+
+	if sshPrivateKey == "" {
+		logger.Fatal(fmt.Errorf("invalid arguments passed, private key file is mandatory for --sftp='ssh-private-key=path/to/id_ecdsa'"), "unable to start SFTP server")
+	}
+
+	privateBytes, err := os.ReadFile(sshPrivateKey)
+	if err != nil {
+		logger.Fatal(fmt.Errorf("invalid arguments passed, private key file is not accessible: %v", err), "unable to start SFTP server")
+	}
+
+	private, err := ssh.ParsePrivateKey(privateBytes)
+	if err != nil {
+		logger.Fatal(fmt.Errorf("invalid arguments passed, private key file is not parseable: %v", err), "unable to start SFTP server")
+	}
+
+	// An SSH server is represented by a ServerConfig, which holds
+	// certificate details and handles authentication of ServerConns.
+	sshConfig := &ssh.ServerConfig{
+		PasswordCallback: func(c ssh.ConnMetadata, pass []byte) (*ssh.Permissions, error) {
+			if globalIAMSys.LDAPConfig.Enabled() {
+				targetUser, targetGroups, err := globalIAMSys.LDAPConfig.Bind(c.User(), string(pass))
+				if err != nil {
+					return nil, err
+				}
+				ldapPolicies, _ := globalIAMSys.PolicyDBGet(targetUser, false, targetGroups...)
+				if len(ldapPolicies) == 0 {
+					return nil, errAuthentication
+				}
+				return &ssh.Permissions{
+					CriticalOptions: map[string]string{
+						ldapUser:  targetUser,
+						ldapUserN: c.User(),
+					},
+					Extensions: make(map[string]string),
+				}, nil
+			}
+
+			ui, ok := globalIAMSys.GetUser(context.Background(), c.User())
+			if !ok {
+				return nil, errNoSuchUser
+			}
+
+			if subtle.ConstantTimeCompare([]byte(ui.Credentials.SecretKey), pass) == 1 {
+				return &ssh.Permissions{
+					CriticalOptions: map[string]string{
+						"accessKey": c.User(),
+					},
+					Extensions: make(map[string]string),
+				}, nil
+			}
+			return nil, errAuthentication
+		},
+	}
+
+	sshConfig.AddHostKey(private)
+
+	handleSFTPSession := func(channel ssh.Channel, sconn *ssh.ServerConn) {
+		server := sftp.NewRequestServer(channel, NewSFTPDriver(sconn.Permissions), sftp.WithRSAllocator())
+		defer server.Close()
+		server.Serve()
+	}
+
+	sftpServer, err := xsftp.NewServer(&xsftp.Options{
+		PublicIP: publicIP,
+		Port:     port,
+		// OpensSSH default handshake timeout is 2 minutes.
+		SSHHandshakeDeadline: 2 * time.Minute,
+		Logger:               new(sftpLogger),
+		SSHConfig:            sshConfig,
+		HandleSFTPSession:    handleSFTPSession,
+	})
+	if err != nil {
+		logger.Fatal(err, "Unable to start SFTP Server")
+	}
+
+	err = sftpServer.Listen()
+	if err != nil {
+		logger.Fatal(err, "SFTP Server had an unrecoverable error while accepting connections")
+	}
+}

cmd/site-replication.go

@@ -3787,14 +3787,6 @@ func (c *SiteReplicationSys) PeerEditReq(ctx context.Context, arg madmin.PeerInf
 		p := c.state.Peers[i]
 		if p.DeploymentID == arg.DeploymentID {
 			p.Endpoint = arg.Endpoint
-			if arg.DefaultBandwidth.IsSet {
-				if arg.DefaultBandwidth.UpdatedAt.After(p.DefaultBandwidth.UpdatedAt) {
-					p.DefaultBandwidth = arg.DefaultBandwidth
-				}
-			}
-			if !arg.SyncState.Empty() {
-				p.SyncState = arg.SyncState
-			}
 			c.state.Peers[arg.DeploymentID] = p
 		}
 		if p.DeploymentID == globalDeploymentID() {
@@ -5239,6 +5231,9 @@ func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer
 	if len(res.Buckets) > 0 {
 		res.ErrDetail = "partial failure in starting site resync"
 	}
+	if len(buckets) != 0 && len(res.Buckets) == len(buckets) {
+		return res, fmt.Errorf("all buckets resync failed")
+	}
 	return res, nil
 }
 

cmd/speedtest.go

@@ -238,9 +238,15 @@ func driveSpeedTest(ctx context.Context, opts madmin.DriveSpeedTestOpts) madmin.
 	}
 
 	localPaths := globalEndpoints.LocalDisksPaths()
+	var ignoredPaths []string
 	paths := func() (tmpPaths []string) {
 		for _, lp := range localPaths {
-			tmpPaths = append(tmpPaths, pathJoin(lp, minioMetaTmpBucket))
+			if _, err := Lstat(pathJoin(lp, minioMetaBucket, formatConfigFile)); err == nil {
+				tmpPaths = append(tmpPaths, pathJoin(lp, minioMetaTmpBucket))
+			} else {
+				// Use dperf on only formatted drives.
+				ignoredPaths = append(ignoredPaths, lp)
+			}
 		}
 		return tmpPaths
 	}()
@@ -274,6 +280,12 @@ func driveSpeedTest(ctx context.Context, opts madmin.DriveSpeedTestOpts) madmin.
 				}
 				results = append(results, result)
 			}
+			for _, inp := range ignoredPaths {
+				results = append(results, madmin.DrivePerf{
+					Path:  inp,
+					Error: errFaultyDisk.Error(),
+				})
+			}
 			return results
 		}(),
 		Error: func() string {

cmd/storage-datatypes.go

@@ -296,7 +296,11 @@ func (fi FileInfo) GetDataDir() string {
 // InlineData returns true if object contents are inlined alongside its metadata.
 func (fi FileInfo) InlineData() bool {
 	_, ok := fi.Metadata[ReservedMetadataPrefixLower+"inline-data"]
-	return ok
+	// Earlier MinIO versions didn't reset "x-minio-internal-inline-data"
+	// from fi.Metadata when the object was tiered. So, tiered objects
+	// would return true for InlineData() in these versions even though the
+	// object isn't inlined in xl.meta
+	return ok && !fi.IsRemote()
 }
 
 // SetInlineData marks object (version) as inline.

cmd/storage-rest-server.go

@@ -611,6 +611,8 @@ func (s *storageRESTServer) ReadFileStreamHandler(w http.ResponseWriter, r *http
 		return
 	}
 
+	w.Header().Set(xhttp.ContentLength, strconv.Itoa(length))
+
 	rc, err := s.storage.ReadFileStream(r.Context(), volume, filePath, int64(offset), int64(length))
 	if err != nil {
 		s.writeErrorResponse(w, err)
@@ -618,12 +620,28 @@ func (s *storageRESTServer) ReadFileStreamHandler(w http.ResponseWriter, r *http
 	}
 	defer rc.Close()
 
-	w.Header().Set(xhttp.ContentLength, strconv.Itoa(length))
-	if _, err = xioutil.Copy(w, rc); err != nil {
-		if !xnet.IsNetworkOrHostDown(err, true) { // do not need to log disconnected clients
-			logger.LogIf(r.Context(), err)
+	rf, ok := w.(io.ReaderFrom)
+	if ok {
+		// Attempt to use splice/sendfile() optimization, A very specific behavior mentioned below is necessary.
+		// See https://github.com/golang/go/blob/f7c5cbb82087c55aa82081e931e0142783700ce8/src/net/sendfile_linux.go#L20
+		dr, ok := rc.(*xioutil.DeadlineReader)
+		if ok {
+			sr, ok := dr.ReadCloser.(*sendFileReader)
+			if ok {
+				_, err = rf.ReadFrom(sr.Reader)
+				if !xnet.IsNetworkOrHostDown(err, true) { // do not need to log disconnected clients
+					logger.LogIf(r.Context(), err)
+				}
+				if err == nil || !errors.Is(err, xhttp.ErrNotImplemented) {
+					return
+				}
+			}
 		}
-		return
+	} // Fallback to regular copy
+
+	_, err = xioutil.Copy(w, rc)
+	if !xnet.IsNetworkOrHostDown(err, true) { // do not need to log disconnected clients
+		logger.LogIf(r.Context(), err)
 	}
 }
 

cmd/utils.go

@@ -99,12 +99,16 @@ func ErrorRespToObjectError(err error, params ...string) error {
 
 	bucket := ""
 	object := ""
+	versionID := ""
 	if len(params) >= 1 {
 		bucket = params[0]
 	}
 	if len(params) == 2 {
 		object = params[1]
 	}
+	if len(params) == 3 {
+		versionID = params[2]
+	}
 
 	if xnet.IsNetworkOrHostDown(err, false) {
 		return BackendDown{Err: err.Error()}
@@ -142,6 +146,12 @@ func ErrorRespToObjectError(err error, params ...string) error {
 		} else {
 			err = BucketNotFound{Bucket: bucket}
 		}
+	case "NoSuchVersion":
+		if object != "" {
+			err = ObjectNotFound{Bucket: bucket, Object: object, VersionID: versionID}
+		} else {
+			err = BucketNotFound{Bucket: bucket}
+		}
 	case "XMinioInvalidObjectName":
 		err = ObjectNameInvalid{}
 	case "AccessDenied":

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

@@ -421,7 +421,13 @@ func (p *xlStorageDiskIDCheck) ReadFile(ctx context.Context, volume string, path
 	}
 	defer done(&err)
 
-	return p.storage.ReadFile(ctx, volume, path, offset, buf, verifier)
+	w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+	err = w.Run(func() error {
+		n, err = p.storage.ReadFile(ctx, volume, path, offset, buf, verifier)
+		return err
+	})
+
+	return n, err
 }
 
 // Legacy API - does not have any deadlines
@@ -432,7 +438,10 @@ func (p *xlStorageDiskIDCheck) AppendFile(ctx context.Context, volume string, pa
 	}
 	defer done(&err)
 
-	return p.storage.AppendFile(ctx, volume, path, buf)
+	w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+	return w.Run(func() error {
+		return p.storage.AppendFile(ctx, volume, path, buf)
+	})
 }
 
 func (p *xlStorageDiskIDCheck) CreateFile(ctx context.Context, volume, path string, size int64, reader io.Reader) (err error) {
@@ -442,7 +451,7 @@ func (p *xlStorageDiskIDCheck) CreateFile(ctx context.Context, volume, path stri
 	}
 	defer done(&err)
 
-	return p.storage.CreateFile(ctx, volume, path, size, reader)
+	return p.storage.CreateFile(ctx, volume, path, size, xioutil.NewDeadlineReader(io.NopCloser(reader), diskMaxTimeout))
 }
 
 func (p *xlStorageDiskIDCheck) ReadFileStream(ctx context.Context, volume, path string, offset, length int64) (io.ReadCloser, error) {
@@ -452,9 +461,16 @@ func (p *xlStorageDiskIDCheck) ReadFileStream(ctx context.Context, volume, path
 	}
 	defer done(&err)
 
-	rc, err := p.storage.ReadFileStream(ctx, volume, path, offset, length)
+	w := xioutil.NewDeadlineWorker(diskMaxTimeout)
+
+	var rc io.ReadCloser
+	err = w.Run(func() error {
+		var ierr error
+		rc, ierr = p.storage.ReadFileStream(ctx, volume, path, offset, length)
+		return ierr
+	})
 	if err != nil {
-		return rc, err
+		return nil, err
 	}
 
 	return xioutil.NewDeadlineReader(rc, diskMaxTimeout), nil

cmd/xl-storage-format-v2.go

@@ -536,6 +536,10 @@ func (j xlMetaV2Object) InlineData() bool {
 	return ok
 }
 
+func (j *xlMetaV2Object) ResetInlineData() {
+	delete(j.MetaSys, ReservedMetadataPrefixLower+"inline-data")
+}
+
 const (
 	metaTierStatus    = ReservedMetadataPrefixLower + TransitionStatus
 	metaTierObjName   = ReservedMetadataPrefixLower + TransitionedObjectName
@@ -1440,6 +1444,7 @@ func (x *xlMetaV2) DeleteVersion(fi FileInfo) (string, error) {
 			err = x.setIdx(i, *ver)
 		case fi.TransitionStatus == lifecycle.TransitionComplete:
 			ver.ObjectV2.SetTransition(fi)
+			ver.ObjectV2.ResetInlineData()
 			err = x.setIdx(i, *ver)
 		default:
 			x.versions = append(x.versions[:i], x.versions[i+1:]...)

cmd/xl-storage.go

@@ -1779,6 +1779,11 @@ func (s *xlStorage) openFile(filePath string, mode int) (f *os.File, err error)
 	return w, nil
 }
 
+type sendFileReader struct {
+	io.Reader
+	io.Closer
+}
+
 // ReadFileStream - Returns the read stream of the file.
 func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, offset, length int64) (io.ReadCloser, error) {
 	if offset < 0 {
@@ -1796,14 +1801,7 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 		return nil, err
 	}
 
-	odirectEnabled := globalAPIConfig.odirectEnabled() && s.oDirect && length >= 0
-
-	var file *os.File
-	if odirectEnabled {
-		file, err = OpenFileDirectIO(filePath, readMode, 0o666)
-	} else {
-		file, err = OpenFile(filePath, readMode, 0o666)
-	}
+	file, err := OpenFile(filePath, readMode, 0o666)
 	if err != nil {
 		switch {
 		case osIsNotExist(err):
@@ -1852,14 +1850,6 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 		return nil, errFileCorrupt
 	}
 
-	alignment := offset%xioutil.DirectioAlignSize == 0
-	if !alignment && odirectEnabled {
-		if err = disk.DisableDirectIO(file); err != nil {
-			file.Close()
-			return nil, err
-		}
-	}
-
 	if offset > 0 {
 		if _, err = file.Seek(offset, io.SeekStart); err != nil {
 			file.Close()
@@ -1867,26 +1857,7 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 		}
 	}
 
-	or := &xioutil.ODirectReader{
-		File: file,
-		// Select bigger blocks when reading at least 50% of a big block.
-		SmallFile: length <= xioutil.BlockSizeLarge/2,
-	}
-
-	r := struct {
-		io.Reader
-		io.Closer
-	}{Reader: io.LimitReader(diskHealthReader(ctx, or), length), Closer: closeWrapper(func() error {
-		if (!alignment || offset+length%xioutil.DirectioAlignSize != 0) && odirectEnabled {
-			// invalidate page-cache for unaligned reads.
-			// skip removing from page-cache only
-			// if O_DIRECT was disabled.
-			disk.FadviseDontNeed(file)
-		}
-		return or.Close()
-	})}
-
-	return r, nil
+	return &sendFileReader{Reader: io.LimitReader(file, length), Closer: file}, nil
 }
 
 // closeWrapper converts a function to an io.Closer

docs/metrics/prometheus/README.md

@@ -151,3 +151,7 @@ curl https://play.min.io/minio/v2/metrics/cluster
 ### List of metrics reported Cluster and Bucket level
 
 [The list of metrics reported can be here](https://github.com/minio/minio/blob/master/docs/metrics/prometheus/list.md)
+
+### Configure Alerts for Prometheus
+
+[The Prometheus AlertManager and alerts can be configured following this](https://github.com/minio/minio/blob/master/docs/metrics/prometheus/alerts.md)

docs/metrics/prometheus/alerts.md

@@ -0,0 +1,120 @@
+# How to configure Prometheus AlertManager
+
+Alerting with prometheus is two step process. First we setup alerts in Prometheus server and then we need to send alerts to the AlertManager.
+Prometheus AlertManager is the component that manages sending, inhibition and silencing of the alerts generated from Prometheus. The AlertManager can be configured to send alerts to variety of receivers. Refer [Prometheus AlertManager receivers](https://prometheus.io/docs/alerting/latest/configuration/#receiver) for more details.
+
+Follow below steps to enable and use AlertManager.
+
+## Deploy and start AlertManager
+Install Prometheus AlertManager from https://prometheus.io/download/ and create configuration as below
+
+```yaml
+route:
+  group_by: ['alertname']
+  group_wait: 30s
+  group_interval: 5m
+  repeat_interval: 1h
+  receiver: 'web.hook'
+receivers:
+  - name: 'web.hook'
+    webhook_configs:
+      - url: 'http://127.0.0.1:8010/webhook'
+inhibit_rules:
+  - source_match:
+      severity: 'critical'
+    target_match:
+      severity: 'warning'
+    equal: ['alertname', 'dev', 'instance']
+```
+
+This sample confoguration uses a `webhook` at http://127.0.0.1:8010/webhook to post the alerts.
+Start the AlertManager and it listens on port `9093` by default. Make sure your webhook is up and listening for the alerts.
+
+## Configure Prometheus to use AlertManager
+
+Add below section to your `prometheus.yml`
+```yaml
+alerting:
+  alertmanagers:
+  - static_configs:
+    - targets: ['localhost:9093']
+rule_files:
+  - rules.yml
+```
+Here `rules.yml` is the file which should contain the alerting rules defined.
+
+## Add rules for your deployment
+Below is a sample alerting rules configuration for MinIO. Refer https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/ for more instructions on writing alerting rules for Prometheus.
+
+```yaml
+groups:
+- name: example
+  rules:
+  - alert: MinIOClusterTolerance
+    expr: minio_cluster_health_erasure_set_tolerance <= 0
+    for: 5m
+    labels:
+      severity: critical
+    annotations:
+      summary: "Instance {{ $labels.server }} unable to tolerate node failures"
+      description: "MinIO instance {{ $labels.server }} of job {{ $labels.job }} has tolerance <=0 for more than 5 minutes."
+```
+
+## Verify the configuration and alerts
+To verify the above sample alert follow below steps
+
+1. Start a distributed MinIO instance (4 nodes setup)
+2. Start Prometheus server and AlertManager
+3. Bring down couple of MinIO instances to bring down the Erasure Set tolerance to -1 and verify the same with `mc admin prometheus metrics ALIAS | grep minio_cluster_health_erasure_set_tolerance`
+4. Wait for 5 mins (as alert is configured to be firing after 5 mins), and verify that you see an entry in webhook for the alert as well as in Prometheus console as shown below
+
+```json
+{
+  "receiver": "web\\.hook",
+  "status": "firing",
+  "alerts": [
+    {
+      "status": "firing",
+      "labels": {
+        "alertname": "MinIOClusterTolerance",
+        "instance": "localhost:9000",
+        "job": "minio-job-node",
+        "pool": "0",
+        "server": "127.0.0.1:9000",
+        "set": "0",
+        "severity": "critical"
+      },
+      "annotations": {
+        "description": "MinIO instance 127.0.0.1:9000 of job minio-job has tolerance <=0 for more than 5 minutes.",
+        "summary": "Instance 127.0.0.1:9000 unable to tolerate node failures"
+      },
+      "startsAt": "2023-11-18T06:20:09.456Z",
+      "endsAt": "0001-01-01T00:00:00Z",
+      "generatorURL": "http://fedora-shubhendu:9090/graph?g0.expr=minio_cluster_health_erasure_set_tolerance+%3C%3D+0&g0.tab=1",
+      "fingerprint": "2255608b0da28ca3"
+    }
+  ],
+  "groupLabels": {
+    "alertname": "MinIOClusterTolerance"
+  },
+  "commonLabels": {
+    "alertname": "MinIOClusterTolerance",
+    "instance": "localhost:9000",
+    "job": "minio-job-node",
+    "pool": "0",
+    "server": "127.0.0.1:9000",
+    "set": "0",
+    "severity": "critical"
+  },
+  "commonAnnotations": {
+    "description": "MinIO instance 127.0.0.1:9000 of job minio-job has tolerance <=0 for more than 5 minutes.",
+    "summary": "Instance 127.0.0.1:9000 unable to tolerate node failures"
+  },
+  "externalURL": "http://fedora-shubhendu:9093",
+  "version": "4",
+  "groupKey": "{}:{alertname=\"MinIOClusterTolerance\"}",
+  "truncatedAlerts": 0
+}
+```
+
+![Prometheus](https://raw.githubusercontent.com/minio/minio/master/docs/metrics/prometheus/minio-es-tolerance-alert.png)

docs/metrics/prometheus/grafana/minio-dashboard.json

@@ -4540,7 +4540,7 @@
       "fill": 1,
       "fillGradient": 0,
       "gridPos": {
-        "h": 6,
+        "h": 5,
         "w": 6,
         "x": 18,
         "y": 109
@@ -4616,6 +4616,94 @@
       "yaxis": {
         "align": false
       }
+    },
+    {
+      "aliasColors": {},
+      "bars": false,
+      "dashLength": 10,
+      "dashes": false,
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
+      "fill": 1,
+      "fillGradient": 0,
+      "gridPos": {
+        "h": 6,
+        "w": 12,
+        "x": 0,
+        "y": 115
+      },
+      "hiddenSeries": false,
+      "id": 94,
+      "legend": {
+        "avg": false,
+        "current": false,
+        "max": false,
+        "min": false,
+        "show": true,
+        "total": false,
+        "values": false
+      },
+      "lines": true,
+      "linewidth": 1,
+      "nullPointMode": "null",
+      "options": {
+        "alertThreshold": true
+      },
+      "percentage": false,
+      "pluginVersion": "10.0.2",
+      "pointradius": 2,
+      "points": false,
+      "renderer": "flot",
+      "seriesOverrides": [],
+      "spaceLength": 10,
+      "stack": false,
+      "steppedLine": false,
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "exemplar": true,
+          "expr": "minio_cluster_health_erasure_set_tolerance{job=\"$scrape_jobs\"}",
+          "interval": "",
+          "legendFormat": "Pool: {{pool}} Set: {{set}}",
+          "refId": "A"
+        }
+      ],
+      "thresholds": [],
+      "timeRegions": [],
+      "title": "Erasure Set Tolerance",
+      "tooltip": {
+        "shared": true,
+        "sort": 0,
+        "value_type": "individual"
+      },
+      "type": "graph",
+      "xaxis": {
+        "mode": "time",
+        "show": true,
+        "values": []
+      },
+      "yaxes": [
+        {
+          "$$hashKey": "object:1043",
+          "format": "none",
+          "logBase": 1,
+          "show": true
+        },
+        {
+          "$$hashKey": "object:1044",
+          "format": "short",
+          "logBase": 1,
+          "show": true
+        }
+      ],
+      "yaxis": {
+        "align": false
+      }
     }
   ],
   "refresh": "",

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.2023-11-01T18-37-25Z
+  image: quay.io/minio/minio:RELEASE.2023-11-15T20-43-25Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

go.mod

@@ -49,9 +49,9 @@ require (
 	github.com/minio/highwayhash v1.0.2
 	github.com/minio/kes-go v0.2.0
 	github.com/minio/madmin-go/v3 v3.0.29
-	github.com/minio/minio-go/v7 v7.0.64-0.20230920204636-e783c9ba11b3
+	github.com/minio/minio-go/v7 v7.0.64-0.20231119012610-6eebdd6d5eba
 	github.com/minio/mux v1.9.0
-	github.com/minio/pkg/v2 v2.0.2
+	github.com/minio/pkg/v2 v2.0.3-0.20231107172951-8a60b89ec9b4
 	github.com/minio/selfupdate v0.6.0
 	github.com/minio/sha256-simd v1.0.1
 	github.com/minio/simdjson-go v0.4.5

go.sum

@@ -485,14 +485,14 @@ github.com/minio/mc v0.0.0-20231030184332-9f2fb2b6a9f8/go.mod h1:SoPU55ntH5d6IEq
 github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
 github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
 github.com/minio/minio-go/v6 v6.0.46/go.mod h1:qD0lajrGW49lKZLtXKtCB4X/qkMf0a5tBvN2PaZg7Gg=
-github.com/minio/minio-go/v7 v7.0.64-0.20230920204636-e783c9ba11b3 h1:0DtfDxg67S/IRcGnIBKgzyjZ0GXyk3jN1Fy5/+8CQlM=
-github.com/minio/minio-go/v7 v7.0.64-0.20230920204636-e783c9ba11b3/go.mod h1:Q6X7Qjb7WMhvG65qKf4gUgA5XaiSox74kR1uAEjxRS4=
+github.com/minio/minio-go/v7 v7.0.64-0.20231119012610-6eebdd6d5eba h1:OF4FT9fUnXReMpMLQyoydIYJYPN3W3omFuD/TTL8cZg=
+github.com/minio/minio-go/v7 v7.0.64-0.20231119012610-6eebdd6d5eba/go.mod h1:R4WVUR6ZTedlCcGwZRauLMIKjgyaWxhs4Mqi/OMPmEc=
 github.com/minio/mux v1.9.0 h1:dWafQFyEfGhJvK6AwLOt83bIG5bxKxKJnKMCi0XAaoA=
 github.com/minio/mux v1.9.0/go.mod h1:1pAare17ZRL5GpmNL+9YmqHoWnLmMZF9C/ioUCfy0BQ=
 github.com/minio/pkg v1.7.5 h1:UOUJjewE5zoaDPlCMJtNx/swc1jT1ZR+IajT7hrLd44=
 github.com/minio/pkg v1.7.5/go.mod h1:mEfGMTm5Z0b5EGxKNuPwyb5A2d+CC/VlUyRj6RJtIwo=
-github.com/minio/pkg/v2 v2.0.2 h1:cytXmC21fBNS+0NVKEE5FuYmQfY+HFTqis6Kkj3U9ac=
-github.com/minio/pkg/v2 v2.0.2/go.mod h1:6xTAr5M9yobpUroXAAaTrGJ9fhOZIqKYOT0I87u2yZ4=
+github.com/minio/pkg/v2 v2.0.3-0.20231107172951-8a60b89ec9b4 h1:5eHjHtFZrrCQ3eO0sesXomdAUTtcGh0Fpp7Qa6dtjrY=
+github.com/minio/pkg/v2 v2.0.3-0.20231107172951-8a60b89ec9b4/go.mod h1:6xTAr5M9yobpUroXAAaTrGJ9fhOZIqKYOT0I87u2yZ4=
 github.com/minio/selfupdate v0.6.0 h1:i76PgT0K5xO9+hjzKcacQtO7+MjJ4JKA8Ak8XQ9DDwU=
 github.com/minio/selfupdate v0.6.0/go.mod h1:bO02GTIPCMQFTEvE5h4DjYB58bCoZ35XLeBf0buTDdM=
 github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=

internal/config/api/help.go

@@ -94,7 +94,7 @@ var (
 		},
 		config.HelpKV{
 			Key:         apiODirect,
-			Description: "set to enable or disable O_DIRECT for read and writes under special conditions. NOTE: do not disable O_DIRECT without prior testing" + defaultHelpPostfix(apiODirect),
+			Description: "set to enable or disable O_DIRECT for writes under special conditions. NOTE: do not disable O_DIRECT without prior testing" + defaultHelpPostfix(apiODirect),
 			Optional:    true,
 			Type:        "boolean",
 		},

internal/config/constants.go

@@ -83,3 +83,11 @@ const (
 	EnvRegion     = "MINIO_REGION"      // legacy
 	EnvRegionName = "MINIO_REGION_NAME" // legacy
 )
+
+// Expiration Token durations
+// These values are used to validate the expiration time range from
+// either the exp claim or MINI_STS_DURATION value
+const (
+	MinExpiration = 900
+	MaxExpiration = 31536000
+)

internal/config/identity/openid/jwt.go

@@ -114,8 +114,7 @@ func updateClaimsExpiry(dsecs string, claims map[string]interface{}) error {
 		return nil
 	}
 
-	expAt, err := auth.ExpToInt64(expStr)
-	if err != nil {
+	if _, err := auth.ExpToInt64(expStr); err != nil {
 		return err
 	}
 
@@ -124,13 +123,6 @@ func updateClaimsExpiry(dsecs string, claims map[string]interface{}) error {
 		return err
 	}
 
-	// Verify if JWT expiry is lesser than default expiry duration,
-	// if that is the case then set the default expiration to be
-	// from the JWT expiry claim.
-	if time.Unix(expAt, 0).UTC().Sub(time.Now().UTC()) < defaultExpiryDuration {
-		defaultExpiryDuration = time.Unix(expAt, 0).UTC().Sub(time.Now().UTC())
-	} // else honor the specified expiry duration.
-
 	claims["exp"] = time.Now().UTC().Add(defaultExpiryDuration).Unix() // update with new expiry.
 	return nil
 }

internal/config/identity/openid/openid.go

@@ -603,9 +603,9 @@ func GetDefaultExpiration(dsecs string) (time.Duration, error) {
 	timeout := env.Get(config.EnvMinioStsDuration, "")
 	defaultExpiryDuration, err := time.ParseDuration(timeout)
 	if err != nil {
-		defaultExpiryDuration = time.Duration(60) * time.Minute
+		defaultExpiryDuration = time.Hour
 	}
-	if dsecs != "" {
+	if timeout == "" && dsecs != "" {
 		expirySecs, err := strconv.ParseInt(dsecs, 10, 64)
 		if err != nil {
 			return 0, auth.ErrInvalidDuration
@@ -614,11 +614,18 @@ func GetDefaultExpiration(dsecs string) (time.Duration, error) {
 		// The duration, in seconds, of the role session.
 		// The value can range from 900 seconds (15 minutes)
 		// up to 365 days.
-		if expirySecs < 900 || expirySecs > 31536000 {
+		if expirySecs < config.MinExpiration || expirySecs > config.MaxExpiration {
 			return 0, auth.ErrInvalidDuration
 		}
 
 		defaultExpiryDuration = time.Duration(expirySecs) * time.Second
+	} else if timeout == "" && dsecs == "" {
+		return time.Hour, nil
 	}
+
+	if defaultExpiryDuration.Seconds() < config.MinExpiration || defaultExpiryDuration.Seconds() > config.MaxExpiration {
+		return 0, auth.ErrInvalidDuration
+	}
+
 	return defaultExpiryDuration, nil
 }

internal/config/lambda/target/webhook.go

@@ -30,7 +30,7 @@ import (
 	"time"
 
 	"github.com/minio/minio/internal/config/lambda/event"
-	xioutil "github.com/minio/minio/internal/ioutil"
+	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/v2/certs"
 	xnet "github.com/minio/pkg/v2/net"
@@ -133,7 +133,7 @@ func (target *WebhookTarget) isActive() (bool, error) {
 		}
 		return false, err
 	}
-	xioutil.DiscardReader(resp.Body)
+	xhttp.DrainBody(resp.Body)
 	// No network failure i.e response from the target means its up
 	return true, nil
 }

internal/event/target/kafka_scram_client_contrib.go

@@ -18,6 +18,7 @@ package target
 
 import (
 	"crypto/sha512"
+	"strings"
 
 	"github.com/IBM/sarama"
 	"github.com/xdg/scram"
@@ -26,7 +27,7 @@ import (
 )
 
 func initScramClient(args KafkaArgs, config *sarama.Config) {
-	switch args.SASL.Mechanism {
+	switch strings.ToLower(args.SASL.Mechanism) {
 	case "sha512":
 		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: KafkaSHA512} }
 		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)

internal/event/target/webhook.go

@@ -34,7 +34,7 @@ import (
 	"time"
 
 	"github.com/minio/minio/internal/event"
-	xioutil "github.com/minio/minio/internal/ioutil"
+	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/minio/internal/once"
 	"github.com/minio/minio/internal/store"
@@ -196,7 +196,7 @@ func (target *WebhookTarget) send(eventData event.Event) error {
 	if err != nil {
 		return err
 	}
-	defer xioutil.DiscardReader(resp.Body)
+	defer xhttp.DrainBody(resp.Body)
 
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
 		return fmt.Errorf("sending event failed with %v", resp.Status)

internal/http/response-recorder.go

@@ -19,6 +19,7 @@ package http
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -29,6 +30,7 @@ import (
 // status code and to record the response body
 type ResponseRecorder struct {
 	http.ResponseWriter
+	io.ReaderFrom
 	StatusCode int
 	// Log body of 4xx or 5xx responses
 	LogErrBody bool
@@ -51,13 +53,27 @@ type ResponseRecorder struct {
 // NewResponseRecorder - returns a wrapped response writer to trap
 // http status codes for auditing purposes.
 func NewResponseRecorder(w http.ResponseWriter) *ResponseRecorder {
+	rf, _ := w.(io.ReaderFrom)
 	return &ResponseRecorder{
 		ResponseWriter: w,
+		ReaderFrom:     rf,
 		StatusCode:     http.StatusOK,
 		StartTime:      time.Now().UTC(),
 	}
 }
 
+// ErrNotImplemented when a functionality is not implemented
+var ErrNotImplemented = errors.New("not implemented")
+
+// ReadFrom implements support for calling internal io.ReaderFrom implementations
+// returns an error if the underlying ResponseWriter does not implement io.ReaderFrom
+func (lrw *ResponseRecorder) ReadFrom(r io.Reader) (int64, error) {
+	if lrw.ReaderFrom != nil {
+		return lrw.ReaderFrom.ReadFrom(r)
+	}
+	return 0, ErrNotImplemented
+}
+
 func (lrw *ResponseRecorder) Write(p []byte) (int, error) {
 	if !lrw.headersLogged {
 		// We assume the response code to be '200 OK' when WriteHeader() is not called,

internal/logger/target/http/http.go

@@ -379,10 +379,6 @@ func (h *Target) SendFromStore(key store.Key) (err error) {
 // Messages are queued in the disk if the store is enabled
 // If Cancel has been called the message is ignored.
 func (h *Target) Send(ctx context.Context, entry interface{}) error {
-	if atomic.LoadInt32(&h.status) == statusOffline {
-		h.config.LogOnce(ctx, fmt.Errorf("target %s is offline", h.Endpoint()), h.Endpoint())
-		return nil
-	}
 	if atomic.LoadInt32(&h.status) == statusClosed {
 		return nil
 	}

internal/logger/target/kafka/kafka_scram_client_contrib.go

@@ -19,6 +19,7 @@ package kafka
 
 import (
 	"crypto/sha512"
+	"strings"
 
 	"github.com/IBM/sarama"
 	"github.com/xdg/scram"
@@ -27,7 +28,7 @@ import (
 )
 
 func initScramClient(cfg Config, config *sarama.Config) {
-	switch cfg.SASL.Mechanism {
+	switch strings.ToLower(cfg.SASL.Mechanism) {
 	case "sha512":
 		config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: KafkaSHA512} }
 		config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)