make immediate purge non-blocking up to 100,000 entries per drive (#19231)

make immediate purge non-blocking upto 100000 entries per drive

Bonus: turn-off O_DIRECT verification when FSType is 'XFS'

cmd/admin-handlers.go

@@ -159,7 +159,7 @@ func (a adminAPIHandlers) ServerUpdateV2Handler(w http.ResponseWriter, r *http.R
 				peerResults[nerr.Host.String()] = madmin.ServerPeerUpdateStatus{
 					Host:           nerr.Host.String(),
 					CurrentVersion: Version,
-					UpdatedVersion: lrTime.Format(minioReleaseTagTimeLayout),
+					UpdatedVersion: lrTime.Format(MinioReleaseTagTimeLayout),
 				}
 			}
 		}
@@ -176,7 +176,7 @@ func (a adminAPIHandlers) ServerUpdateV2Handler(w http.ResponseWriter, r *http.R
 			peerResults[local] = madmin.ServerPeerUpdateStatus{
 				Host:           local,
 				CurrentVersion: Version,
-				UpdatedVersion: lrTime.Format(minioReleaseTagTimeLayout),
+				UpdatedVersion: lrTime.Format(MinioReleaseTagTimeLayout),
 			}
 		}
 	} else {
@@ -212,7 +212,7 @@ func (a adminAPIHandlers) ServerUpdateV2Handler(w http.ResponseWriter, r *http.R
 							Host:           nerr.Host.String(),
 							Err:            nerr.Err.Error(),
 							CurrentVersion: Version,
-							UpdatedVersion: lrTime.Format(minioReleaseTagTimeLayout),
+							UpdatedVersion: lrTime.Format(MinioReleaseTagTimeLayout),
 						}
 					}
 				}
@@ -404,7 +404,7 @@ func (a adminAPIHandlers) ServerUpdateHandler(w http.ResponseWriter, r *http.Req
 
 	updateStatus := madmin.ServerUpdateStatus{
 		CurrentVersion: Version,
-		UpdatedVersion: lrTime.Format(minioReleaseTagTimeLayout),
+		UpdatedVersion: lrTime.Format(MinioReleaseTagTimeLayout),
 	}
 
 	// Marshal API response
@@ -2280,8 +2280,6 @@ func getServerInfo(ctx context.Context, pools, metrics bool, r *http.Request) ma
 	servers := globalNotificationSys.ServerInfo(metrics)
 	servers = append(servers, local)
 
-	assignPoolNumbers(servers)
-
 	var poolsInfo map[int]map[int]madmin.ErasureSetInfo
 	var backend madmin.ErasureBackend
 
@@ -2718,14 +2716,20 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
 			for _, server := range infoMessage.Servers {
 				anonEndpoint := anonAddr(server.Endpoint)
 				servers = append(servers, madmin.ServerInfo{
-					State:      server.State,
-					Endpoint:   anonEndpoint,
-					Uptime:     server.Uptime,
-					Version:    server.Version,
-					CommitID:   server.CommitID,
-					Network:    anonymizeNetwork(server.Network),
-					Drives:     anonymizeDrives(server.Disks),
-					PoolNumber: server.PoolNumber,
+					State:    server.State,
+					Endpoint: anonEndpoint,
+					Uptime:   server.Uptime,
+					Version:  server.Version,
+					CommitID: server.CommitID,
+					Network:  anonymizeNetwork(server.Network),
+					Drives:   anonymizeDrives(server.Disks),
+					PoolNumber: func() int {
+						if len(server.PoolNumbers) == 1 {
+							return server.PoolNumbers[0]
+						}
+						return math.MaxInt // this indicates that its unset.
+					}(),
+					PoolNumbers: server.PoolNumbers,
 					MemStats: madmin.MemStats{
 						Alloc:      server.MemStats.Alloc,
 						TotalAlloc: server.MemStats.TotalAlloc,
@@ -2915,22 +2919,6 @@ func (a adminAPIHandlers) ServerInfoHandler(w http.ResponseWriter, r *http.Reque
 	writeSuccessResponseJSON(w, jsonBytes)
 }
 
-func assignPoolNumbers(servers []madmin.ServerProperties) {
-	for i := range servers {
-		for idx, ge := range globalEndpoints {
-			for _, endpoint := range ge.Endpoints {
-				if servers[i].Endpoint == endpoint.Host {
-					servers[i].PoolNumber = idx + 1
-				} else if host, err := xnet.ParseHost(servers[i].Endpoint); err == nil {
-					if host.Name == endpoint.Hostname() {
-						servers[i].PoolNumber = idx + 1
-					}
-				}
-			}
-		}
-	}
-}
-
 func fetchLambdaInfo() []map[string][]madmin.TargetIDStatus {
 	lambdaMap := make(map[string][]madmin.TargetIDStatus)
 

cmd/admin-server-info.go

@@ -19,10 +19,12 @@ package cmd
 
 import (
 	"context"
+	"math"
 	"net/http"
 	"os"
 	"runtime"
 	"runtime/debug"
+	"sort"
 	"strings"
 	"time"
 
@@ -42,9 +44,13 @@ func getLocalServerProperty(endpointServerPools EndpointServerPools, r *http.Req
 	if globalIsDistErasure {
 		addr = globalLocalNodeName
 	}
+	poolNumbers := make(map[int]struct{})
 	network := make(map[string]string)
 	for _, ep := range endpointServerPools {
 		for _, endpoint := range ep.Endpoints {
+			if endpoint.IsLocal {
+				poolNumbers[endpoint.PoolIdx+1] = struct{}{}
+			}
 			nodeName := endpoint.Host
 			if nodeName == "" {
 				nodeName = addr
@@ -112,6 +118,17 @@ func getLocalServerProperty(endpointServerPools EndpointServerPools, r *http.Req
 		MinioEnvVars: make(map[string]string, 10),
 	}
 
+	for poolNumber := range poolNumbers {
+		props.PoolNumbers = append(props.PoolNumbers, poolNumber)
+	}
+	sort.Ints(props.PoolNumbers)
+	props.PoolNumber = func() int {
+		if len(props.PoolNumbers) == 1 {
+			return props.PoolNumbers[0]
+		}
+		return math.MaxInt // this indicates that its unset.
+	}()
+
 	sensitive := map[string]struct{}{
 		config.EnvAccessKey:         {},
 		config.EnvSecretKey:         {},

cmd/api-headers.go

@@ -50,7 +50,7 @@ func setEventStreamHeaders(w http.ResponseWriter) {
 // Write http common headers
 func setCommonHeaders(w http.ResponseWriter) {
 	// Set the "Server" http header.
-	w.Header().Set(xhttp.ServerInfo, "MinIO")
+	w.Header().Set(xhttp.ServerInfo, MinioStoreName)
 
 	// Set `x-amz-bucket-region` only if region is set on the server
 	// by default minio uses an empty region.

cmd/auth-handler.go

@@ -298,15 +298,15 @@ func checkClaimsFromToken(r *http.Request, cred auth.Credentials) (map[string]in
 	if cred.IsTemp() && cred.IsExpired() {
 		return nil, toAPIErrorCode(r.Context(), errInvalidAccessKeyID)
 	}
-
 	secret := globalActiveCred.SecretKey
-	var err error
 	if globalSiteReplicationSys.isEnabled() && cred.AccessKey != siteReplicatorSvcAcc {
-		if cred.ParentUser != globalActiveCred.AccessKey {
-			secret, err = getTokenSigningKey()
-			if err != nil {
-				return nil, toAPIErrorCode(r.Context(), err)
-			}
+		nsecret, err := getTokenSigningKey()
+		if err != nil {
+			return nil, toAPIErrorCode(r.Context(), err)
+		}
+		// sign root's temporary accounts also with site replicator creds
+		if cred.ParentUser != globalActiveCred.AccessKey || cred.IsTemp() {
+			secret = nsecret
 		}
 	}
 	if cred.IsServiceAccount() {

cmd/batch-handlers.go

@@ -63,7 +63,6 @@ type BatchJobRequest struct {
 	ID        string               `yaml:"-" json:"name"`
 	User      string               `yaml:"-" json:"user"`
 	Started   time.Time            `yaml:"-" json:"started"`
-	Location  string               `yaml:"-" json:"location"`
 	Replicate *BatchJobReplicateV1 `yaml:"replicate" json:"replicate"`
 	KeyRotate *BatchJobKeyRotateV1 `yaml:"keyrotate" json:"keyrotate"`
 	Expire    *BatchJobExpire      `yaml:"expire" json:"expire"`
@@ -710,38 +709,52 @@ type batchJobInfo struct {
 }
 
 const (
-	batchReplName      = "batch-replicate.bin"
-	batchReplFormat    = 1
-	batchReplVersionV1 = 1
-	batchReplVersion   = batchReplVersionV1
-	batchJobName       = "job.bin"
-	batchJobPrefix     = "batch-jobs"
+	batchReplName         = "batch-replicate.bin"
+	batchReplFormat       = 1
+	batchReplVersionV1    = 1
+	batchReplVersion      = batchReplVersionV1
+	batchJobName          = "job.bin"
+	batchJobPrefix        = "batch-jobs"
+	batchJobReportsPrefix = batchJobPrefix + "/reports"
 
 	batchReplJobAPIVersion        = "v1"
 	batchReplJobDefaultRetries    = 3
 	batchReplJobDefaultRetryDelay = 250 * time.Millisecond
 )
 
-func (ri *batchJobInfo) load(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
+func getJobReportPath(job BatchJobRequest) string {
 	var fileName string
-	var format, version uint16
 	switch {
 	case job.Replicate != nil:
 		fileName = batchReplName
+	case job.KeyRotate != nil:
+		fileName = batchKeyRotationName
+	case job.Expire != nil:
+		fileName = batchExpireName
+	}
+	return pathJoin(batchJobReportsPrefix, job.ID, fileName)
+}
+
+func getJobPath(job BatchJobRequest) string {
+	return pathJoin(batchJobPrefix, job.ID)
+}
+
+func (ri *batchJobInfo) load(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
+	var format, version uint16
+	switch {
+	case job.Replicate != nil:
 		version = batchReplVersionV1
 		format = batchReplFormat
 	case job.KeyRotate != nil:
-		fileName = batchKeyRotationName
 		version = batchKeyRotateVersionV1
 		format = batchKeyRotationFormat
 	case job.Expire != nil:
-		fileName = batchExpireName
 		version = batchExpireVersionV1
 		format = batchExpireFormat
 	default:
 		return errors.New("no supported batch job request specified")
 	}
-	data, err := readConfig(ctx, api, pathJoin(job.Location, fileName))
+	data, err := readConfig(ctx, api, getJobReportPath(job))
 	if err != nil {
 		if errors.Is(err, errConfigNotFound) || isErrObjectNotFound(err) {
 			ri.Version = int(version)
@@ -852,8 +865,8 @@ func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, durati
 	now := UTCNow()
 	ri.mu.Lock()
 	var (
-		format, version  uint16
-		jobTyp, fileName string
+		format, version uint16
+		jobTyp          string
 	)
 
 	if now.Sub(ri.LastUpdate) >= duration {
@@ -862,19 +875,16 @@ func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, durati
 			format = batchReplFormat
 			version = batchReplVersion
 			jobTyp = string(job.Type())
-			fileName = batchReplName
 			ri.Version = batchReplVersionV1
 		case madmin.BatchJobKeyRotate:
 			format = batchKeyRotationFormat
 			version = batchKeyRotateVersion
 			jobTyp = string(job.Type())
-			fileName = batchKeyRotationName
 			ri.Version = batchKeyRotateVersionV1
 		case madmin.BatchJobExpire:
 			format = batchExpireFormat
 			version = batchExpireVersion
 			jobTyp = string(job.Type())
-			fileName = batchExpireName
 			ri.Version = batchExpireVersionV1
 		default:
 			return errInvalidArgument
@@ -895,7 +905,7 @@ func (ri *batchJobInfo) updateAfter(ctx context.Context, api ObjectLayer, durati
 		if err != nil {
 			return err
 		}
-		return saveConfig(ctx, api, pathJoin(job.Location, fileName), buf)
+		return saveConfig(ctx, api, getJobReportPath(job), buf)
 	}
 	ri.mu.Unlock()
 	return nil
@@ -1417,15 +1427,8 @@ func (j BatchJobRequest) Validate(ctx context.Context, o ObjectLayer) error {
 }
 
 func (j BatchJobRequest) delete(ctx context.Context, api ObjectLayer) {
-	switch {
-	case j.Replicate != nil:
-		deleteConfig(ctx, api, pathJoin(j.Location, batchReplName))
-	case j.KeyRotate != nil:
-		deleteConfig(ctx, api, pathJoin(j.Location, batchKeyRotationName))
-	case j.Expire != nil:
-		deleteConfig(ctx, api, pathJoin(j.Location, batchExpireName))
-	}
-	deleteConfig(ctx, api, j.Location)
+	deleteConfig(ctx, api, getJobReportPath(j))
+	deleteConfig(ctx, api, getJobPath(j))
 }
 
 func (j *BatchJobRequest) save(ctx context.Context, api ObjectLayer) error {
@@ -1437,13 +1440,12 @@ func (j *BatchJobRequest) save(ctx context.Context, api ObjectLayer) error {
 		return err
 	}
 
-	j.Location = pathJoin(batchJobPrefix, j.ID)
 	job, err := j.MarshalMsg(nil)
 	if err != nil {
 		return err
 	}
 
-	return saveConfig(ctx, api, j.Location, job)
+	return saveConfig(ctx, api, getJobPath(*j), job)
 }
 
 func (j *BatchJobRequest) load(ctx context.Context, api ObjectLayer, name string) error {
@@ -1671,8 +1673,7 @@ func (a adminAPIHandlers) CancelBatchJob(w http.ResponseWriter, r *http.Request)
 	}
 
 	j := BatchJobRequest{
-		ID:       jobID,
-		Location: pathJoin(batchJobPrefix, jobID),
+		ID: jobID,
 	}
 
 	j.delete(ctx, objectAPI)

cmd/batch-handlers_gen.go

@@ -42,12 +42,6 @@ func (z *BatchJobRequest) DecodeMsg(dc *msgp.Reader) (err error) {
 				err = msgp.WrapError(err, "Started")
 				return
 			}
-		case "Location":
-			z.Location, err = dc.ReadString()
-			if err != nil {
-				err = msgp.WrapError(err, "Location")
-				return
-			}
 		case "Replicate":
 			if dc.IsNil() {
 				err = dc.ReadNil()
@@ -115,9 +109,9 @@ func (z *BatchJobRequest) DecodeMsg(dc *msgp.Reader) (err error) {
 
 // EncodeMsg implements msgp.Encodable
 func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) {
-	// map header, size 7
+	// map header, size 6
 	// write "ID"
-	err = en.Append(0x87, 0xa2, 0x49, 0x44)
+	err = en.Append(0x86, 0xa2, 0x49, 0x44)
 	if err != nil {
 		return
 	}
@@ -146,16 +140,6 @@ func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) {
 		err = msgp.WrapError(err, "Started")
 		return
 	}
-	// write "Location"
-	err = en.Append(0xa8, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e)
-	if err != nil {
-		return
-	}
-	err = en.WriteString(z.Location)
-	if err != nil {
-		err = msgp.WrapError(err, "Location")
-		return
-	}
 	// write "Replicate"
 	err = en.Append(0xa9, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x65)
 	if err != nil {
@@ -213,9 +197,9 @@ func (z *BatchJobRequest) EncodeMsg(en *msgp.Writer) (err error) {
 // MarshalMsg implements msgp.Marshaler
 func (z *BatchJobRequest) MarshalMsg(b []byte) (o []byte, err error) {
 	o = msgp.Require(b, z.Msgsize())
-	// map header, size 7
+	// map header, size 6
 	// string "ID"
-	o = append(o, 0x87, 0xa2, 0x49, 0x44)
+	o = append(o, 0x86, 0xa2, 0x49, 0x44)
 	o = msgp.AppendString(o, z.ID)
 	// string "User"
 	o = append(o, 0xa4, 0x55, 0x73, 0x65, 0x72)
@@ -223,9 +207,6 @@ func (z *BatchJobRequest) MarshalMsg(b []byte) (o []byte, err error) {
 	// string "Started"
 	o = append(o, 0xa7, 0x53, 0x74, 0x61, 0x72, 0x74, 0x65, 0x64)
 	o = msgp.AppendTime(o, z.Started)
-	// string "Location"
-	o = append(o, 0xa8, 0x4c, 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e)
-	o = msgp.AppendString(o, z.Location)
 	// string "Replicate"
 	o = append(o, 0xa9, 0x52, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x65)
 	if z.Replicate == nil {
@@ -298,12 +279,6 @@ func (z *BatchJobRequest) UnmarshalMsg(bts []byte) (o []byte, err error) {
 				err = msgp.WrapError(err, "Started")
 				return
 			}
-		case "Location":
-			z.Location, bts, err = msgp.ReadStringBytes(bts)
-			if err != nil {
-				err = msgp.WrapError(err, "Location")
-				return
-			}
 		case "Replicate":
 			if msgp.IsNil(bts) {
 				bts, err = msgp.ReadNilBytes(bts)
@@ -369,7 +344,7 @@ func (z *BatchJobRequest) UnmarshalMsg(bts []byte) (o []byte, err error) {
 
 // Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
 func (z *BatchJobRequest) Msgsize() (s int) {
-	s = 1 + 3 + msgp.StringPrefixSize + len(z.ID) + 5 + msgp.StringPrefixSize + len(z.User) + 8 + msgp.TimeSize + 9 + msgp.StringPrefixSize + len(z.Location) + 10
+	s = 1 + 3 + msgp.StringPrefixSize + len(z.ID) + 5 + msgp.StringPrefixSize + len(z.User) + 8 + msgp.TimeSize + 10
 	if z.Replicate == nil {
 		s += msgp.NilSize
 	} else {

cmd/build-constants.go

@@ -17,6 +17,8 @@
 
 package cmd
 
+import "runtime"
+
 // DO NOT EDIT THIS FILE DIRECTLY. These are build-time constants
 // set through ‘buildscripts/gen-ldflags.go’.
 var (
@@ -40,4 +42,25 @@ var (
 
 	// CopyrightYear - dynamic value of the copyright end year
 	CopyrightYear = "0000"
+
+	// MinioReleaseTagTimeLayout - release tag time layout.
+	MinioReleaseTagTimeLayout = "2006-01-02T15-04-05Z"
+
+	// MinioOSARCH - OS and ARCH.
+	minioOSARCH = runtime.GOOS + "-" + runtime.GOARCH
+
+	// MinioReleaseURL - release URL.
+	MinioReleaseURL = "https://dl.min.io/server/minio/release/" + minioOSARCH + SlashSeparator
+
+	// MinioStoreName - MinIO store name.
+	MinioStoreName = "MinIO"
+
+	// MinioUAName - MinIO user agent name.
+	MinioUAName = "MinIO"
+
+	// MinioBannerName - MinIO banner name for startup message.
+	MinioBannerName = "MinIO Object Storage Server"
+
+	// MinioLicense - MinIO server license.
+	MinioLicense = "GNU AGPLv3 <https://www.gnu.org/licenses/agpl-3.0.html>"
 )

cmd/erasure-object.go

@@ -1791,20 +1791,12 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
 func (er erasureObjects) deletePrefix(ctx context.Context, bucket, prefix string) error {
 	disks := er.getDisks()
 	g := errgroup.WithNErrs(len(disks))
-	dirPrefix := encodeDirObject(prefix)
 	for index := range disks {
 		index := index
 		g.Go(func() error {
 			if disks[index] == nil {
 				return nil
 			}
-			// Deletes
-			// - The prefix and its children
-			// - The prefix__XLDIR__
-			defer disks[index].Delete(ctx, bucket, dirPrefix, DeleteOptions{
-				Recursive: true,
-				Immediate: true,
-			})
 			return disks[index].Delete(ctx, bucket, prefix, DeleteOptions{
 				Recursive: true,
 				Immediate: true,
@@ -1828,9 +1820,6 @@ func (er erasureObjects) DeleteObject(ctx context.Context, bucket, object string
 	}
 
 	if opts.DeletePrefix {
-		if globalCacheConfig.Enabled() {
-			return ObjectInfo{}, toObjectErr(errMethodNotAllowed, bucket, object)
-		}
 		return ObjectInfo{}, toObjectErr(er.deletePrefix(ctx, bucket, object), bucket, object)
 	}
 

cmd/erasure-server-pool.go

@@ -92,15 +92,15 @@ func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServ
 		n = 2048
 	}
 
-	if globalIsCICD {
-		n = 256 // 256MiB for CI/CD environments is sufficient
-	}
-
 	// Avoid allocating more than half of the available memory
 	if maxN := availableMemory() / (blockSizeV2 * 2); n > maxN {
 		n = maxN
 	}
 
+	if globalIsCICD || strconv.IntSize == 32 {
+		n = 256 // 256MiB for CI/CD environments is sufficient or on 32bit platforms.
+	}
+
 	// Initialize byte pool once for all sets, bpool size is set to
 	// setCount * setDriveCount with each memory upto blockSizeV2.
 	globalBytePoolCap = bpool.NewBytePoolCap(n, blockSizeV2, blockSizeV2*2)

cmd/ftp-server.go

@@ -136,7 +136,7 @@ func startFTPServer(args []string) {
 
 	ftpServer, err := ftp.NewServer(&ftp.Options{
 		Name:           name,
-		WelcomeMessage: fmt.Sprintf("Welcome to MinIO FTP Server Version='%s' License='GNU AGPLv3'", Version),
+		WelcomeMessage: fmt.Sprintf("Welcome to '%s' FTP Server Version='%s' License='%s'", MinioStoreName, MinioLicense, Version),
 		Driver:         NewFTPDriver(),
 		Port:           port,
 		Perm:           ftp.NewSimplePerm("nobody", "nobody"),

cmd/handler-api.go

@@ -87,14 +87,14 @@ func cgroupMemLimit() (limit uint64) {
 }
 
 func availableMemory() (available uint64) {
-	available = 8 << 30 // Default to 8 GiB when we can't find the limits.
+	available = 2048 * blockSizeV2 * 2 // Default to 4 GiB when we can't find the limits.
 
 	if runtime.GOOS == "linux" {
 		// Useful in container mode
 		limit := cgroupMemLimit()
 		if limit > 0 {
-			// A valid value is found
-			available = limit
+			// A valid value is found, return its 75%
+			available = (limit * 3) / 4
 			return
 		}
 	} // for all other platforms limits are based on virtual memory.
@@ -103,7 +103,8 @@ func availableMemory() (available uint64) {
 	if err != nil {
 		return
 	}
-	available = memStats.Available / 2
+	// A valid value is available return its 75%
+	available = (memStats.Available * 3) / 4
 	return
 }
 
@@ -132,6 +133,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int) {
 
 	var apiRequestsMaxPerNode int
 	if cfg.RequestsMax <= 0 {
+		// Returns 75% of max memory allowed
 		maxMem := availableMemory()
 
 		// max requests per node is calculated as

cmd/handler-utils.go

@@ -366,12 +366,6 @@ func errorResponseHandler(w http.ResponseWriter, r *http.Request) {
 	}
 	desc := "Do not upgrade one server at a time - please follow the recommended guidelines mentioned here https://github.com/minio/minio#upgrading-minio for your environment"
 	switch {
-	case strings.HasPrefix(r.URL.Path, peerS3Prefix):
-		writeErrorResponseString(r.Context(), w, APIError{
-			Code:           "XMinioPeerS3VersionMismatch",
-			Description:    desc,
-			HTTPStatusCode: http.StatusUpgradeRequired,
-		}, r.URL)
 	case strings.HasPrefix(r.URL.Path, peerRESTPrefix):
 		writeErrorResponseString(r.Context(), w, APIError{
 			Code:           "XMinioPeerVersionMismatch",

cmd/main.go

@@ -25,7 +25,9 @@ import (
 	"runtime"
 	"runtime/debug"
 	"sort"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/minio/cli"
 	"github.com/minio/minio/internal/color"
@@ -169,8 +171,9 @@ func newApp(name string) *cli.App {
 }
 
 func startupBanner(banner io.Writer) {
+	CopyrightYear = strconv.Itoa(time.Now().Year())
 	fmt.Fprintln(banner, color.Blue("Copyright:")+color.Bold(" 2015-%s MinIO, Inc.", CopyrightYear))
-	fmt.Fprintln(banner, color.Blue("License:")+color.Bold(" GNU AGPLv3 <https://www.gnu.org/licenses/agpl-3.0.html>"))
+	fmt.Fprintln(banner, color.Blue("License:")+color.Bold(" "+MinioLicense))
 	fmt.Fprintln(banner, color.Blue("Version:")+color.Bold(" %s (%s %s/%s)", ReleaseTag, runtime.Version(), runtime.GOOS, runtime.GOARCH))
 }
 

cmd/metacache-entries.go

@@ -745,10 +745,10 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 
 		// Merge any unmerged
 		if len(toMerge) > 0 {
-			versions := make([]xlMetaV2ShallowVersion, 0, len(toMerge)+1)
+			versions := make([][]xlMetaV2ShallowVersion, 0, len(toMerge)+1)
 			xl, err := best.xlmeta()
 			if err == nil {
-				versions = append(versions, xl.versions...)
+				versions = append(versions, xl.versions)
 			}
 			for _, idx := range toMerge {
 				other := top[idx]
@@ -776,11 +776,12 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 						return err
 					}
 				}
-				versions = append(versions, xl2.versions...)
+				versions = append(versions, xl2.versions)
 			}
+
 			if xl != nil && len(versions) > 0 {
 				// Merge all versions. 'strict' doesn't matter since we only need one.
-				xl.versions = mergeXLV2Versions(readQuorum, true, 0, versions)
+				xl.versions = mergeXLV2Versions(readQuorum, true, 0, versions...)
 				if meta, err := xl.AppendTo(metaDataPoolGet()); err == nil {
 					if best.reusable {
 						metaDataPoolPut(best.metadata)

cmd/object-api-datatypes.go

@@ -30,6 +30,8 @@ import (
 	"github.com/minio/minio/internal/logger"
 )
 
+//go:generate msgp -file $GOFILE -io=false -tests=false -unexported=false
+
 // BackendType - represents different backend types.
 type BackendType int
 
@@ -187,9 +189,9 @@ type ObjectInfo struct {
 	Parts []ObjectPartInfo `json:"-"`
 
 	// Implements writer and reader used by CopyObject API
-	Writer       io.WriteCloser `json:"-"`
-	Reader       *hash.Reader   `json:"-"`
-	PutObjReader *PutObjReader  `json:"-"`
+	Writer       io.WriteCloser `json:"-" msg:"-"`
+	Reader       *hash.Reader   `json:"-" msg:"-"`
+	PutObjReader *PutObjReader  `json:"-" msg:"-"`
 
 	metadataOnly bool
 	versionOnly  bool // adds a new version, only used by CopyObject

cmd/object-api-interface.go

@@ -32,6 +32,10 @@ import (
 	xioutil "github.com/minio/minio/internal/ioutil"
 )
 
+//go:generate msgp -file $GOFILE -io=false -tests=false -unexported=false
+
+//msgp:ignore ObjectOptions TransitionOptions DeleteBucketOptions
+
 // CheckPreconditionFn returns true if precondition check failed.
 type CheckPreconditionFn func(o ObjectInfo) bool
 

cmd/object-api-interface_gen.go

@@ -0,0 +1,319 @@
+package cmd
+
+// Code generated by github.com/tinylib/msgp DO NOT EDIT.
+
+import (
+	"github.com/tinylib/msgp/msgp"
+)
+
+// MarshalMsg implements msgp.Marshaler
+func (z BucketOptions) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	// map header, size 2
+	// string "Deleted"
+	o = append(o, 0x82, 0xa7, 0x44, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x64)
+	o = msgp.AppendBool(o, z.Deleted)
+	// string "Cached"
+	o = append(o, 0xa6, 0x43, 0x61, 0x63, 0x68, 0x65, 0x64)
+	o = msgp.AppendBool(o, z.Cached)
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *BucketOptions) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, bts, err = msgp.ReadMapKeyZC(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "Deleted":
+			z.Deleted, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Deleted")
+				return
+			}
+		case "Cached":
+			z.Cached, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Cached")
+				return
+			}
+		default:
+			bts, err = msgp.Skip(bts)
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z BucketOptions) Msgsize() (s int) {
+	s = 1 + 8 + msgp.BoolSize + 7 + msgp.BoolSize
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z ExpirationOptions) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	// map header, size 1
+	// string "Expire"
+	o = append(o, 0x81, 0xa6, 0x45, 0x78, 0x70, 0x69, 0x72, 0x65)
+	o = msgp.AppendBool(o, z.Expire)
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *ExpirationOptions) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, bts, err = msgp.ReadMapKeyZC(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "Expire":
+			z.Expire, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Expire")
+				return
+			}
+		default:
+			bts, err = msgp.Skip(bts)
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z ExpirationOptions) Msgsize() (s int) {
+	s = 1 + 7 + msgp.BoolSize
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z *MakeBucketOptions) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	// map header, size 5
+	// string "LockEnabled"
+	o = append(o, 0x85, 0xab, 0x4c, 0x6f, 0x63, 0x6b, 0x45, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64)
+	o = msgp.AppendBool(o, z.LockEnabled)
+	// string "VersioningEnabled"
+	o = append(o, 0xb1, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x69, 0x6e, 0x67, 0x45, 0x6e, 0x61, 0x62, 0x6c, 0x65, 0x64)
+	o = msgp.AppendBool(o, z.VersioningEnabled)
+	// string "ForceCreate"
+	o = append(o, 0xab, 0x46, 0x6f, 0x72, 0x63, 0x65, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65)
+	o = msgp.AppendBool(o, z.ForceCreate)
+	// string "CreatedAt"
+	o = append(o, 0xa9, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x64, 0x41, 0x74)
+	o = msgp.AppendTime(o, z.CreatedAt)
+	// string "NoLock"
+	o = append(o, 0xa6, 0x4e, 0x6f, 0x4c, 0x6f, 0x63, 0x6b)
+	o = msgp.AppendBool(o, z.NoLock)
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *MakeBucketOptions) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, bts, err = msgp.ReadMapKeyZC(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "LockEnabled":
+			z.LockEnabled, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "LockEnabled")
+				return
+			}
+		case "VersioningEnabled":
+			z.VersioningEnabled, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "VersioningEnabled")
+				return
+			}
+		case "ForceCreate":
+			z.ForceCreate, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "ForceCreate")
+				return
+			}
+		case "CreatedAt":
+			z.CreatedAt, bts, err = msgp.ReadTimeBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "CreatedAt")
+				return
+			}
+		case "NoLock":
+			z.NoLock, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "NoLock")
+				return
+			}
+		default:
+			bts, err = msgp.Skip(bts)
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z *MakeBucketOptions) Msgsize() (s int) {
+	s = 1 + 12 + msgp.BoolSize + 18 + msgp.BoolSize + 12 + msgp.BoolSize + 10 + msgp.TimeSize + 7 + msgp.BoolSize
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z *WalkOptions) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	// map header, size 4
+	// string "Marker"
+	o = append(o, 0x84, 0xa6, 0x4d, 0x61, 0x72, 0x6b, 0x65, 0x72)
+	o = msgp.AppendString(o, z.Marker)
+	// string "LatestOnly"
+	o = append(o, 0xaa, 0x4c, 0x61, 0x74, 0x65, 0x73, 0x74, 0x4f, 0x6e, 0x6c, 0x79)
+	o = msgp.AppendBool(o, z.LatestOnly)
+	// string "AskDisks"
+	o = append(o, 0xa8, 0x41, 0x73, 0x6b, 0x44, 0x69, 0x73, 0x6b, 0x73)
+	o = msgp.AppendString(o, z.AskDisks)
+	// string "VersionsSort"
+	o = append(o, 0xac, 0x56, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x53, 0x6f, 0x72, 0x74)
+	o = msgp.AppendUint8(o, uint8(z.VersionsSort))
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *WalkOptions) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	var field []byte
+	_ = field
+	var zb0001 uint32
+	zb0001, bts, err = msgp.ReadMapHeaderBytes(bts)
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	for zb0001 > 0 {
+		zb0001--
+		field, bts, err = msgp.ReadMapKeyZC(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		switch msgp.UnsafeString(field) {
+		case "Marker":
+			z.Marker, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "Marker")
+				return
+			}
+		case "LatestOnly":
+			z.LatestOnly, bts, err = msgp.ReadBoolBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "LatestOnly")
+				return
+			}
+		case "AskDisks":
+			z.AskDisks, bts, err = msgp.ReadStringBytes(bts)
+			if err != nil {
+				err = msgp.WrapError(err, "AskDisks")
+				return
+			}
+		case "VersionsSort":
+			{
+				var zb0002 uint8
+				zb0002, bts, err = msgp.ReadUint8Bytes(bts)
+				if err != nil {
+					err = msgp.WrapError(err, "VersionsSort")
+					return
+				}
+				z.VersionsSort = WalkVersionsSortOrder(zb0002)
+			}
+		default:
+			bts, err = msgp.Skip(bts)
+			if err != nil {
+				err = msgp.WrapError(err)
+				return
+			}
+		}
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z *WalkOptions) Msgsize() (s int) {
+	s = 1 + 7 + msgp.StringPrefixSize + len(z.Marker) + 11 + msgp.BoolSize + 9 + msgp.StringPrefixSize + len(z.AskDisks) + 13 + msgp.Uint8Size
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z WalkVersionsSortOrder) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	o = msgp.AppendUint8(o, uint8(z))
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *WalkVersionsSortOrder) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	{
+		var zb0001 uint8
+		zb0001, bts, err = msgp.ReadUint8Bytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		(*z) = WalkVersionsSortOrder(zb0001)
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z WalkVersionsSortOrder) Msgsize() (s int) {
+	s = msgp.Uint8Size
+	return
+}

cmd/peer-rest-common.go

@@ -25,22 +25,16 @@ const (
 )
 
 const (
-	peerRESTMethodHealth                      = "/health"
-	peerRESTMethodVerifyBinary                = "/verifybinary"
-	peerRESTMethodCommitBinary                = "/commitbinary"
-	peerRESTMethodSignalService               = "/signalservice"
-	peerRESTMethodBackgroundHealStatus        = "/backgroundhealstatus"
-	peerRESTMethodGetLocks                    = "/getlocks"
-	peerRESTMethodStartProfiling              = "/startprofiling"
-	peerRESTMethodDownloadProfilingData       = "/downloadprofilingdata"
-	peerRESTMethodGetBandwidth                = "/bandwidth"
-	peerRESTMethodSpeedTest                   = "/speedtest"
-	peerRESTMethodDriveSpeedTest              = "/drivespeedtest"
-	peerRESTMethodReloadSiteReplicationConfig = "/reloadsitereplicationconfig"
-	peerRESTMethodGetLastDayTierStats         = "/getlastdaytierstats"
-	peerRESTMethodDevNull                     = "/devnull"
-	peerRESTMethodNetperf                     = "/netperf"
-	peerRESTMethodGetReplicationMRF           = "/getreplicationmrf"
+	peerRESTMethodHealth                = "/health"
+	peerRESTMethodVerifyBinary          = "/verifybinary"
+	peerRESTMethodCommitBinary          = "/commitbinary"
+	peerRESTMethodStartProfiling        = "/startprofiling"
+	peerRESTMethodDownloadProfilingData = "/downloadprofilingdata"
+	peerRESTMethodSpeedTest             = "/speedtest"
+	peerRESTMethodDriveSpeedTest        = "/drivespeedtest"
+	peerRESTMethodDevNull               = "/devnull"
+	peerRESTMethodNetperf               = "/netperf"
+	peerRESTMethodGetReplicationMRF     = "/getreplicationmrf"
 )
 
 const (

cmd/peer-rest-server.go

@@ -53,6 +53,7 @@ type peerRESTServer struct{}
 
 var (
 	// Types & Wrappers
+	aoBucketInfo           = grid.NewArrayOf[*BucketInfo](func() *BucketInfo { return &BucketInfo{} })
 	aoMetricsGroup         = grid.NewArrayOf[*Metric](func() *Metric { return &Metric{} })
 	madminBgHealState      = grid.NewJSONPool[madmin.BgHealState]()
 	madminCPUs             = grid.NewJSONPool[madmin.CPUs]()
@@ -97,6 +98,7 @@ var (
 	getSysServicesRPC              = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.SysServices]](grid.HandlerGetSysServices, grid.NewMSS, madminSysServices.NewJSON)
 	headBucketRPC                  = grid.NewSingleHandler[*grid.MSS, *VolInfo](grid.HandlerHeadBucket, grid.NewMSS, func() *VolInfo { return &VolInfo{} })
 	healBucketRPC                  = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerHealBucket, grid.NewMSS, grid.NewNoPayload)
+	listBucketsRPC                 = grid.NewSingleHandler[*BucketOptions, *grid.Array[*BucketInfo]](grid.HandlerListBuckets, func() *BucketOptions { return &BucketOptions{} }, aoBucketInfo.New)
 	loadBucketMetadataRPC          = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerLoadBucketMetadata, grid.NewMSS, grid.NewNoPayload).IgnoreNilConn()
 	loadGroupRPC                   = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerLoadGroup, grid.NewMSS, grid.NewNoPayload)
 	loadPolicyMappingRPC           = grid.NewSingleHandler[*grid.MSS, grid.NoPayload](grid.HandlerLoadPolicyMapping, grid.NewMSS, grid.NewNoPayload).IgnoreNilConn()
@@ -1226,7 +1228,20 @@ func (s *peerRESTServer) HealBucketHandler(mss *grid.MSS) (np grid.NoPayload, ne
 	return np, nil
 }
 
-// HeadBucketHandler implements peer BuckeInfo call, returns bucket create date.
+func (s *peerRESTServer) ListBucketsHandler(opts *BucketOptions) (*grid.Array[*BucketInfo], *grid.RemoteErr) {
+	buckets, err := listBucketsLocal(context.Background(), *opts)
+	if err != nil {
+		return nil, grid.NewRemoteErr(err)
+	}
+	res := aoBucketInfo.New()
+	for i := range buckets {
+		bucket := buckets[i]
+		res.Append(&bucket)
+	}
+	return res, nil
+}
+
+// HeadBucketHandler implements peer BucketInfo call, returns bucket create date.
 func (s *peerRESTServer) HeadBucketHandler(mss *grid.MSS) (info *VolInfo, nerr *grid.RemoteErr) {
 	bucket := mss.Get(peerS3Bucket)
 	if isMinioMetaBucket(bucket) {
@@ -1332,6 +1347,7 @@ func registerPeerRESTHandlers(router *mux.Router, gm *grid.Manager) {
 	logger.FatalIf(getSysServicesRPC.Register(gm, server.GetSysServicesHandler), "unable to register handler")
 	logger.FatalIf(headBucketRPC.Register(gm, server.HeadBucketHandler), "unable to register handler")
 	logger.FatalIf(healBucketRPC.Register(gm, server.HealBucketHandler), "unable to register handler")
+	logger.FatalIf(listBucketsRPC.Register(gm, server.ListBucketsHandler), "unable to register handler")
 	logger.FatalIf(listenRPC.RegisterNoInput(gm, server.ListenHandler), "unable to register handler")
 	logger.FatalIf(loadBucketMetadataRPC.Register(gm, server.LoadBucketMetadataHandler), "unable to register handler")
 	logger.FatalIf(loadGroupRPC.Register(gm, server.LoadGroupHandler), "unable to register handler")

cmd/peer-s3-client.go

@@ -19,11 +19,8 @@ package cmd
 
 import (
 	"context"
-	"encoding/gob"
 	"errors"
 	"fmt"
-	"io"
-	"net/url"
 	"sort"
 	"strconv"
 	"sync/atomic"
@@ -31,9 +28,7 @@ import (
 
 	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/grid"
-	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
-	"github.com/minio/minio/internal/rest"
 	"github.com/minio/pkg/v2/sync/errgroup"
 	"golang.org/x/exp/slices"
 )
@@ -92,39 +87,14 @@ func (l localPeerS3Client) DeleteBucket(ctx context.Context, bucket string, opts
 
 // client to talk to peer Nodes.
 type remotePeerS3Client struct {
-	node       Node
-	pools      []int
-	restClient *rest.Client
+	node  Node
+	pools []int
 
 	// Function that returns the grid connection for this peer when initialized.
 	// Will return nil if the grid connection is not initialized yet.
 	gridConn func() *grid.Connection
 }
 
-// Wrapper to restClient.Call to handle network errors, in case of network error the connection is marked disconnected
-// permanently. The only way to restore the connection is at the xl-sets layer by xlsets.monitorAndConnectEndpoints()
-// after verifying format.json
-func (client *remotePeerS3Client) call(method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
-	return client.callWithContext(GlobalContext, method, values, body, length)
-}
-
-// Wrapper to restClient.Call to handle network errors, in case of network error the connection is marked disconnected
-// permanently. The only way to restore the connection is at the xl-sets layer by xlsets.monitorAndConnectEndpoints()
-// after verifying format.json
-func (client *remotePeerS3Client) callWithContext(ctx context.Context, method string, values url.Values, body io.Reader, length int64) (respBody io.ReadCloser, err error) {
-	if values == nil {
-		values = make(url.Values)
-	}
-
-	respBody, err = client.restClient.Call(ctx, method, values, body, length)
-	if err == nil {
-		return respBody, nil
-	}
-
-	err = toStorageErr(err)
-	return nil, err
-}
-
 // S3PeerSys - S3 peer call system.
 type S3PeerSys struct {
 	peerClients []peerS3Client // Excludes self
@@ -351,18 +321,18 @@ func (sys *S3PeerSys) GetBucketInfo(ctx context.Context, bucket string, opts Buc
 }
 
 func (client *remotePeerS3Client) ListBuckets(ctx context.Context, opts BucketOptions) ([]BucketInfo, error) {
-	v := url.Values{}
-	v.Set(peerS3BucketDeleted, strconv.FormatBool(opts.Deleted))
-
-	respBody, err := client.call(peerS3MethodListBuckets, v, nil, -1)
+	bi, err := listBucketsRPC.Call(ctx, client.gridConn(), &opts)
 	if err != nil {
-		return nil, err
+		return nil, toStorageErr(err)
 	}
-	defer xhttp.DrainBody(respBody)
-
-	var buckets []BucketInfo
-	err = gob.NewDecoder(respBody).Decode(&buckets)
-	return buckets, err
+	buckets := make([]BucketInfo, 0, len(bi.Value()))
+	for _, b := range bi.Value() {
+		if b != nil {
+			buckets = append(buckets, *b)
+		}
+	}
+	bi.Recycle() // BucketInfo has no internal pointers, so it's safe to recycle.
+	return buckets, nil
 }
 
 func (client *remotePeerS3Client) HealBucket(ctx context.Context, bucket string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
@@ -533,35 +503,10 @@ func newPeerS3Clients(endpoints EndpointServerPools) (peers []peerS3Client) {
 
 // Returns a peer S3 client.
 func newPeerS3Client(node Node) peerS3Client {
-	scheme := "http"
-	if globalIsTLS {
-		scheme = "https"
-	}
-
-	serverURL := &url.URL{
-		Scheme: scheme,
-		Host:   node.Host,
-		Path:   peerS3Path,
-	}
-
-	restClient := rest.NewClient(serverURL, globalInternodeTransport, newCachedAuthToken())
-	// Use a separate client to avoid recursive calls.
-	healthClient := rest.NewClient(serverURL, globalInternodeTransport, newCachedAuthToken())
-	healthClient.NoMetrics = true
-
-	// Construct a new health function.
-	restClient.HealthCheckFn = func() bool {
-		ctx, cancel := context.WithTimeout(context.Background(), restClient.HealthCheckTimeout)
-		defer cancel()
-		respBody, err := healthClient.Call(ctx, peerS3MethodHealth, nil, nil, -1)
-		xhttp.DrainBody(respBody)
-		return !isNetworkError(err)
-	}
-
 	var gridConn atomic.Pointer[grid.Connection]
 
 	return &remotePeerS3Client{
-		node: node, restClient: restClient,
+		node: node,
 		gridConn: func() *grid.Connection {
 			// Lazy initialization of grid connection.
 			// When we create this peer client, the grid connection is likely not yet initialized.

cmd/peer-s3-server.go

@@ -19,33 +19,12 @@ package cmd
 
 import (
 	"context"
-	"encoding/gob"
 	"errors"
-	"net/http"
 
 	"github.com/minio/madmin-go/v3"
-	"github.com/minio/minio/internal/logger"
-	"github.com/minio/mux"
 	"github.com/minio/pkg/v2/sync/errgroup"
 )
 
-const (
-	peerS3Version = "v1" // First implementation
-
-	peerS3VersionPrefix = SlashSeparator + peerS3Version
-	peerS3Prefix        = minioReservedBucketPath + "/peer-s3"
-	peerS3Path          = peerS3Prefix + peerS3VersionPrefix
-)
-
-const (
-	peerS3MethodHealth        = "/health"
-	peerS3MethodMakeBucket    = "/make-bucket"
-	peerS3MethodGetBucketInfo = "/get-bucket-info"
-	peerS3MethodDeleteBucket  = "/delete-bucket"
-	peerS3MethodListBuckets   = "/list-buckets"
-	peerS3MethodHealBucket    = "/heal-bucket"
-)
-
 const (
 	peerS3Bucket            = "bucket"
 	peerS3BucketDeleted     = "bucket-deleted"
@@ -53,33 +32,6 @@ const (
 	peerS3BucketForceDelete = "force-delete"
 )
 
-type peerS3Server struct{}
-
-func (s *peerS3Server) writeErrorResponse(w http.ResponseWriter, err error) {
-	w.WriteHeader(http.StatusForbidden)
-	w.Write([]byte(err.Error()))
-}
-
-// IsValid - To authenticate and verify the time difference.
-func (s *peerS3Server) IsValid(w http.ResponseWriter, r *http.Request) bool {
-	objAPI := newObjectLayerFn()
-	if objAPI == nil {
-		s.writeErrorResponse(w, errServerNotInitialized)
-		return false
-	}
-
-	if err := storageServerRequestValidate(r); err != nil {
-		s.writeErrorResponse(w, err)
-		return false
-	}
-	return true
-}
-
-// HealthHandler - returns true of health
-func (s *peerS3Server) HealthHandler(w http.ResponseWriter, r *http.Request) {
-	s.IsValid(w, r)
-}
-
 func healBucketLocal(ctx context.Context, bucket string, opts madmin.HealOpts) (res madmin.HealResultItem, err error) {
 	globalLocalDrivesMu.RLock()
 	localDrives := cloneDrives(globalLocalDrives)
@@ -372,34 +324,3 @@ func makeBucketLocal(ctx context.Context, bucket string, opts MakeBucketOptions)
 	errs := g.Wait()
 	return reduceWriteQuorumErrs(ctx, errs, bucketOpIgnoredErrs, (len(localDrives)/2)+1)
 }
-
-func (s *peerS3Server) ListBucketsHandler(w http.ResponseWriter, r *http.Request) {
-	if !s.IsValid(w, r) {
-		return
-	}
-
-	bucketDeleted := r.Form.Get(peerS3BucketDeleted) == "true"
-
-	buckets, err := listBucketsLocal(r.Context(), BucketOptions{
-		Deleted: bucketDeleted,
-	})
-	if err != nil {
-		s.writeErrorResponse(w, err)
-		return
-	}
-
-	logger.LogIf(r.Context(), gob.NewEncoder(w).Encode(buckets))
-}
-
-// registerPeerS3Handlers - register peer s3 router.
-func registerPeerS3Handlers(router *mux.Router) {
-	server := &peerS3Server{}
-	subrouter := router.PathPrefix(peerS3Prefix).Subrouter()
-
-	h := func(f http.HandlerFunc) http.HandlerFunc {
-		return collectInternodeStats(httpTraceHdrs(f))
-	}
-
-	subrouter.Methods(http.MethodPost).Path(peerS3VersionPrefix + peerS3MethodHealth).HandlerFunc(h(server.HealthHandler))
-	subrouter.Methods(http.MethodPost).Path(peerS3VersionPrefix + peerS3MethodListBuckets).HandlerFunc(h(server.ListBucketsHandler))
-}

cmd/routers.go

@@ -32,9 +32,6 @@ func registerDistErasureRouters(router *mux.Router, endpointServerPools Endpoint
 	// Register peer REST router only if its a distributed setup.
 	registerPeerRESTHandlers(router, globalGrid.Load())
 
-	// Register peer S3 router only if its a distributed setup.
-	registerPeerS3Handlers(router)
-
 	// Register bootstrap REST router for distributed setups.
 	registerBootstrapRESTHandlers(globalGrid.Load())
 

cmd/server-main.go

@@ -652,6 +652,8 @@ func getServerListenAddrs() []string {
 
 // serverMain handler called for 'minio server' command.
 func serverMain(ctx *cli.Context) {
+	var warnings []string
+
 	signal.Notify(globalOSSignalCh, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)
 
 	go handleSignals()
@@ -733,13 +735,13 @@ func serverMain(ctx *cli.Context) {
 
 	// 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"))
+		warnings = append(warnings, color.YellowBold("- 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"))
 	}
 
 	maxProcs := runtime.GOMAXPROCS(0)
 	cpuProcs := runtime.NumCPU()
 	if maxProcs < cpuProcs {
-		logger.Info(color.RedBoldf("WARNING: Detected GOMAXPROCS(%d) < NumCPU(%d), please make sure to provide all PROCS to MinIO for optimal performance", maxProcs, cpuProcs))
+		warnings = append(warnings, color.YellowBold("- Detected GOMAXPROCS(%d) < NumCPU(%d), please make sure to provide all PROCS to MinIO for optimal performance", maxProcs, cpuProcs))
 	}
 
 	var getCert certs.GetCertificateFunc
@@ -840,14 +842,13 @@ func serverMain(ctx *cli.Context) {
 		}
 
 		if !globalServerCtxt.StrictS3Compat {
-			logger.Info(color.RedBold("WARNING: Strict AWS S3 compatible incoming PUT, POST content payload validation is turned off, caution is advised do not use in production"))
+			warnings = append(warnings, color.YellowBold("- Strict AWS S3 compatible incoming PUT, POST content payload validation is turned off, caution is advised do not use in production"))
 		}
 	})
-
 	if globalActiveCred.Equal(auth.DefaultCredentials) {
-		msg := fmt.Sprintf("WARNING: Detected default credentials '%s', we recommend that you change these values with 'MINIO_ROOT_USER' and 'MINIO_ROOT_PASSWORD' environment variables",
+		msg := fmt.Sprintf("- Detected default credentials '%s', we recommend that you change these values with 'MINIO_ROOT_USER' and 'MINIO_ROOT_PASSWORD' environment variables",
 			globalActiveCred)
-		logger.Info(color.RedBold(msg))
+		warnings = append(warnings, color.YellowBold(msg))
 	}
 
 	// Initialize users credentials and policies in background right after config has initialized.
@@ -893,7 +894,7 @@ func serverMain(ctx *cli.Context) {
 		if !globalDisableFreezeOnBoot {
 			defer bootstrapTrace("unfreezeServices", unfreezeServices)
 			t := time.AfterFunc(5*time.Minute, func() {
-				logger.Info(color.Yellow("WARNING: Initializing the config subsystem is taking longer than 5 minutes. Please set '_MINIO_DISABLE_API_FREEZE_ON_BOOT=true' to not freeze the APIs"))
+				warnings = append(warnings, color.YellowBold("- Initializing the config subsystem is taking longer than 5 minutes. Please set '_MINIO_DISABLE_API_FREEZE_ON_BOOT=true' to not freeze the APIs"))
 			})
 			defer t.Stop()
 		}
@@ -994,7 +995,17 @@ func serverMain(ctx *cli.Context) {
 
 		// Print a warning at the end of the startup banner so it is more noticeable
 		if newObject.BackendInfo().StandardSCParity == 0 {
-			logger.Error("Warning: The standard parity is set to 0. This can lead to data loss.")
+			warnings = append(warnings, color.YellowBold("- The standard parity is set to 0. This can lead to data loss."))
+		}
+		objAPI := newObjectLayerFn()
+		if objAPI != nil {
+			printStorageInfo(objAPI.StorageInfo(GlobalContext, true))
+		}
+		if len(warnings) > 0 {
+			logger.Info(color.Yellow("STARTUP WARNINGS:"))
+			for _, warn := range warnings {
+				logger.Info(warn)
+			}
 		}
 	}()
 

cmd/server-startup-msg.go

@@ -37,7 +37,7 @@ func getFormatStr(strLen int, padding int) string {
 
 // Prints the formatted startup message.
 func printStartupMessage(apiEndpoints []string, err error) {
-	logger.Info(color.Bold("MinIO Object Storage Server"))
+	logger.Info(color.Bold(MinioBannerName))
 	if err != nil {
 		if globalConsoleSys != nil {
 			globalConsoleSys.Send(GlobalContext, fmt.Sprintf("Server startup failed with '%v', some features may be missing", err))
@@ -52,12 +52,6 @@ func printStartupMessage(apiEndpoints []string, err error) {
 
 	strippedAPIEndpoints := stripStandardPorts(apiEndpoints, globalMinioHost)
 
-	// Object layer is initialized then print StorageInfo.
-	objAPI := newObjectLayerFn()
-	if objAPI != nil {
-		printStorageInfo(objAPI.StorageInfo(GlobalContext, true))
-	}
-
 	// Prints credential, region and browser access.
 	printServerCommonMsg(strippedAPIEndpoints)
 
@@ -117,24 +111,23 @@ func printServerCommonMsg(apiEndpoints []string) {
 	// Get saved region.
 	region := globalSite.Region
 
-	apiEndpointStr := strings.Join(apiEndpoints, "  ")
-
+	apiEndpointStr := strings.TrimSpace(strings.Join(apiEndpoints, "  "))
 	// Colorize the message and print.
-	logger.Info(color.Blue("S3-API: ") + color.Bold(fmt.Sprintf("%s ", apiEndpointStr)))
+	logger.Info(color.Blue("API: ") + color.Bold(fmt.Sprintf("%s ", apiEndpointStr)))
 	if color.IsTerminal() && (!globalServerCtxt.Anonymous && !globalServerCtxt.JSON && globalAPIConfig.permitRootAccess()) {
-		logger.Info(color.Blue("RootUser: ") + color.Bold("%s ", cred.AccessKey))
-		logger.Info(color.Blue("RootPass: ") + color.Bold("%s \n", cred.SecretKey))
+		logger.Info(color.Blue("   RootUser: ") + color.Bold("%s ", cred.AccessKey))
+		logger.Info(color.Blue("   RootPass: ") + color.Bold("%s \n", cred.SecretKey))
 		if region != "" {
-			logger.Info(color.Blue("Region: ") + color.Bold("%s", fmt.Sprintf(getFormatStr(len(region), 2), region)))
+			logger.Info(color.Blue("   Region: ") + color.Bold("%s", fmt.Sprintf(getFormatStr(len(region), 2), region)))
 		}
 	}
 
 	if globalBrowserEnabled {
 		consoleEndpointStr := strings.Join(stripStandardPorts(getConsoleEndpoints(), globalMinioConsoleHost), " ")
-		logger.Info(color.Blue("Console: ") + color.Bold(fmt.Sprintf("%s ", consoleEndpointStr)))
+		logger.Info(color.Blue("WebUI: ") + color.Bold(fmt.Sprintf("%s ", consoleEndpointStr)))
 		if color.IsTerminal() && (!globalServerCtxt.Anonymous && !globalServerCtxt.JSON && globalAPIConfig.permitRootAccess()) {
-			logger.Info(color.Blue("RootUser: ") + color.Bold("%s ", cred.AccessKey))
-			logger.Info(color.Blue("RootPass: ") + color.Bold("%s ", cred.SecretKey))
+			logger.Info(color.Blue("   RootUser: ") + color.Bold("%s ", cred.AccessKey))
+			logger.Info(color.Blue("   RootPass: ") + color.Bold("%s ", cred.SecretKey))
 		}
 	}
 
@@ -144,7 +137,7 @@ func printServerCommonMsg(apiEndpoints []string) {
 
 // Prints startup message for Object API access, prints link to our SDK documentation.
 func printObjectAPIMsg() {
-	logger.Info(color.Blue("\nDocumentation: ") + "https://min.io/docs/minio/linux/index.html")
+	logger.Info(color.Blue("\nDocs: ") + "https://min.io/docs/minio/linux/index.html")
 }
 
 func printLambdaTargets() {
@@ -188,7 +181,7 @@ func printCLIAccessMsg(endPoint string, alias string) {
 
 	// Configure 'mc', following block prints platform specific information for minio client.
 	if color.IsTerminal() && (!globalServerCtxt.Anonymous && globalAPIConfig.permitRootAccess()) {
-		logger.Info(color.Blue("\nCommand-line: ") + mcQuickStartGuide)
+		logger.Info(color.Blue("\nCLI: ") + mcQuickStartGuide)
 		mcMessage := fmt.Sprintf("$ mc alias set '%s' '%s' '%s' '%s'", alias,
 			endPoint, cred.AccessKey, cred.SecretKey)
 		logger.Info(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))

cmd/update-notifier.go

@@ -95,8 +95,8 @@ func colorizeUpdateMessage(updateString string, newerThan string) string {
 
 	lines := []string{
 		color.YellowBold(topLeftChar + strings.Repeat(horizBarChar, maxContentWidth) + topRightChar),
-		vertBarChar + line1InColor + strings.Repeat(" ", maxContentWidth-line1Length) + vertBarChar,
-		vertBarChar + line2InColor + strings.Repeat(" ", maxContentWidth-line2Length) + vertBarChar,
+		color.YellowBold(vertBarChar) + line1InColor + strings.Repeat(" ", maxContentWidth-line1Length) + color.YellowBold(vertBarChar),
+		color.YellowBold(vertBarChar) + line2InColor + strings.Repeat(" ", maxContentWidth-line2Length) + color.YellowBold(vertBarChar),
 		color.YellowBold(bottomLeftChar + strings.Repeat(horizBarChar, maxContentWidth) + bottomRightChar),
 	}
 	return "\n" + strings.Join(lines, "\n") + "\n"

cmd/update.go

@@ -46,16 +46,12 @@ import (
 )
 
 const (
-	minioReleaseTagTimeLayout = "2006-01-02T15-04-05Z"
-	minioOSARCH               = runtime.GOOS + "-" + runtime.GOARCH
-	minioReleaseURL           = "https://dl.min.io/server/minio/release/" + minioOSARCH + SlashSeparator
-
 	envMinisignPubKey = "MINIO_UPDATE_MINISIGN_PUBKEY"
 	updateTimeout     = 10 * time.Second
 )
 
 // For windows our files have .exe additionally.
-var minioReleaseWindowsInfoURL = minioReleaseURL + "minio.exe.sha256sum"
+var minioReleaseWindowsInfoURL = MinioReleaseURL + "minio.exe.sha256sum"
 
 // minioVersionToReleaseTime - parses a standard official release
 // MinIO version string.
@@ -72,7 +68,7 @@ func minioVersionToReleaseTime(version string) (releaseTime time.Time, err error
 // An official minio release tag looks like:
 // `RELEASE.2017-09-29T19-16-56Z`
 func releaseTimeToReleaseTag(releaseTime time.Time) string {
-	return "RELEASE." + releaseTime.Format(minioReleaseTagTimeLayout)
+	return "RELEASE." + releaseTime.Format(MinioReleaseTagTimeLayout)
 }
 
 // releaseTagToReleaseTime - reverse of `releaseTimeToReleaseTag()`
@@ -84,7 +80,7 @@ func releaseTagToReleaseTime(releaseTag string) (releaseTime time.Time, err erro
 	if fields[0] != "RELEASE" {
 		return releaseTime, fmt.Errorf("%s is not a valid release tag", releaseTag)
 	}
-	return time.Parse(minioReleaseTagTimeLayout, fields[1])
+	return time.Parse(MinioReleaseTagTimeLayout, fields[1])
 }
 
 // getModTime - get the file modification time of `path`
@@ -236,8 +232,8 @@ func getUserAgent(mode string) string {
 	uaAppend := func(p, q string) {
 		userAgentParts = append(userAgentParts, p, q)
 	}
-
-	uaAppend("MinIO (", runtime.GOOS)
+	uaAppend(MinioUAName, " (")
+	uaAppend("", runtime.GOOS)
 	uaAppend("; ", runtime.GOARCH)
 	if mode != "" {
 		uaAppend("; ", mode)
@@ -258,14 +254,14 @@ func getUserAgent(mode string) string {
 		uaAppend("; ", "source")
 	}
 
-	uaAppend(") MinIO/", Version)
-	uaAppend(" MinIO/", ReleaseTag)
-	uaAppend(" MinIO/", CommitID)
+	uaAppend(" ", Version)
+	uaAppend(" ", ReleaseTag)
+	uaAppend(" ", CommitID)
 	if IsDCOS() {
 		universePkgVersion := env.Get("MARATHON_APP_LABEL_DCOS_PACKAGE_VERSION", "")
 		// On DC/OS environment try to the get universe package version.
 		if universePkgVersion != "" {
-			uaAppend(" MinIO/universe-", universePkgVersion)
+			uaAppend(" universe-", universePkgVersion)
 		}
 	}
 
@@ -273,25 +269,26 @@ func getUserAgent(mode string) string {
 		// In Kubernetes environment, try to fetch the helm package version
 		helmChartVersion := getHelmVersion("/podinfo/labels")
 		if helmChartVersion != "" {
-			uaAppend(" MinIO/helm-", helmChartVersion)
+			uaAppend(" helm-", helmChartVersion)
 		}
 		// In Kubernetes environment, try to fetch the Operator, VSPHERE plugin version
 		opVersion := env.Get("MINIO_OPERATOR_VERSION", "")
 		if opVersion != "" {
-			uaAppend(" MinIO/operator-", opVersion)
+			uaAppend(" operator-", opVersion)
 		}
 		vsphereVersion := env.Get("MINIO_VSPHERE_PLUGIN_VERSION", "")
 		if vsphereVersion != "" {
-			uaAppend(" MinIO/vsphere-plugin-", vsphereVersion)
+			uaAppend(" vsphere-plugin-", vsphereVersion)
 		}
 	}
 
 	if IsPCFTile() {
 		pcfTileVersion := env.Get("MINIO_PCF_TILE_VERSION", "")
 		if pcfTileVersion != "" {
-			uaAppend(" MinIO/pcf-tile-", pcfTileVersion)
+			uaAppend(" pcf-tile-", pcfTileVersion)
 		}
 	}
+	uaAppend("; ", "")
 
 	if cpus, err := gopsutilcpu.Info(); err == nil && len(cpus) > 0 {
 		cpuMap := make(map[string]struct{}, len(cpus))
@@ -304,6 +301,7 @@ func getUserAgent(mode string) string {
 		uaAppend(" CPU ", fmt.Sprintf("(total_cpus:%d, total_cores:%d; vendor:%s; family:%s; model:%s; stepping:%d; model_name:%s)",
 			len(cpuMap), len(coreMap), cpu.VendorID, cpu.Family, cpu.Model, cpu.Stepping, cpu.ModelName))
 	}
+	uaAppend(")", "")
 
 	return strings.Join(userAgentParts, "")
 }
@@ -474,10 +472,10 @@ func getDownloadURL(releaseTag string) (downloadURL string) {
 
 	// For binary only installations, we return link to the latest binary.
 	if runtime.GOOS == "windows" {
-		return minioReleaseURL + "minio.exe"
+		return MinioReleaseURL + "minio.exe"
 	}
 
-	return minioReleaseURL + "minio"
+	return MinioReleaseURL + "minio"
 }
 
 func getUpdateReaderFromURL(u *url.URL, transport http.RoundTripper, mode string) (io.ReadCloser, error) {

cmd/update_fips.go

@@ -21,4 +21,4 @@
 package cmd
 
 // Newer official download info URLs appear earlier below.
-var minioReleaseInfoURL = minioReleaseURL + "minio.fips.sha256sum"
+var minioReleaseInfoURL = MinioReleaseURL + "minio.fips.sha256sum"

cmd/update_nofips.go

@@ -21,4 +21,4 @@
 package cmd
 
 // Newer official download info URLs appear earlier below.
-var minioReleaseInfoURL = minioReleaseURL + "minio.sha256sum"
+var minioReleaseInfoURL = MinioReleaseURL + "minio.sha256sum"

cmd/update_test.go

@@ -112,12 +112,12 @@ func TestDownloadURL(t *testing.T) {
 		}
 	} else {
 		if runtime.GOOS == "windows" {
-			if durl != minioReleaseURL+"minio.exe" {
-				t.Errorf("Expected %s, got %s", minioReleaseURL+"minio.exe", durl)
+			if durl != MinioReleaseURL+"minio.exe" {
+				t.Errorf("Expected %s, got %s", MinioReleaseURL+"minio.exe", durl)
 			}
 		} else {
-			if durl != minioReleaseURL+"minio" {
-				t.Errorf("Expected %s, got %s", minioReleaseURL+"minio", durl)
+			if durl != MinioReleaseURL+"minio" {
+				t.Errorf("Expected %s, got %s", MinioReleaseURL+"minio", durl)
 			}
 		}
 	}
@@ -147,19 +147,19 @@ func TestUserAgent(t *testing.T) {
 			envName:     "",
 			envValue:    "",
 			mode:        globalMinioModeFS,
-			expectedStr: fmt.Sprintf("MinIO (%s; %s; %s; source) MinIO/DEVELOPMENT.GOGET MinIO/DEVELOPMENT.GOGET MinIO/DEVELOPMENT.GOGET", runtime.GOOS, runtime.GOARCH, globalMinioModeFS),
+			expectedStr: fmt.Sprintf("MinIO (%s; %s; %s; source DEVELOPMENT.GOGET DEVELOPMENT.GOGET DEVELOPMENT.GOGET", runtime.GOOS, runtime.GOARCH, globalMinioModeFS),
 		},
 		{
 			envName:     "MESOS_CONTAINER_NAME",
 			envValue:    "mesos-11111",
 			mode:        globalMinioModeErasure,
-			expectedStr: fmt.Sprintf("MinIO (%s; %s; %s; %s; source) MinIO/DEVELOPMENT.GOGET MinIO/DEVELOPMENT.GOGET MinIO/DEVELOPMENT.GOGET MinIO/universe-%s", runtime.GOOS, runtime.GOARCH, globalMinioModeErasure, "dcos", "mesos-1111"),
+			expectedStr: fmt.Sprintf("MinIO (%s; %s; %s; %s; source DEVELOPMENT.GOGET DEVELOPMENT.GOGET DEVELOPMENT.GOGET universe-%s", runtime.GOOS, runtime.GOARCH, globalMinioModeErasure, "dcos", "mesos-1111"),
 		},
 		{
 			envName:     "KUBERNETES_SERVICE_HOST",
 			envValue:    "10.11.148.5",
 			mode:        globalMinioModeErasure,
-			expectedStr: fmt.Sprintf("MinIO (%s; %s; %s; %s; source) MinIO/DEVELOPMENT.GOGET MinIO/DEVELOPMENT.GOGET MinIO/DEVELOPMENT.GOGET", runtime.GOOS, runtime.GOARCH, globalMinioModeErasure, "kubernetes"),
+			expectedStr: fmt.Sprintf("MinIO (%s; %s; %s; %s; source DEVELOPMENT.GOGET DEVELOPMENT.GOGET DEVELOPMENT.GOGET", runtime.GOOS, runtime.GOARCH, globalMinioModeErasure, "kubernetes"),
 		},
 	}
 

cmd/xl-storage-format-v2.go

@@ -1879,6 +1879,7 @@ func (x xlMetaV2) ListVersions(volume, path string, allParts bool) ([]FileInfo,
 
 // mergeXLV2Versions will merge all versions, typically from different disks
 // that have at least quorum entries in all metas.
+// Each version slice should be sorted.
 // Quorum must be the minimum number of matching metadata files.
 // Quorum should be > 1 and <= len(versions).
 // If strict is set to false, entries that match type

cmd/xl-storage-format-v2_test.go

@@ -21,6 +21,7 @@ import (
 	"bufio"
 	"bytes"
 	"compress/gzip"
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
@@ -1011,6 +1012,77 @@ func Test_mergeXLV2Versions2(t *testing.T) {
 	}
 }
 
+func Test_mergeEntryChannels(t *testing.T) {
+	dataZ, err := os.ReadFile("testdata/xl-meta-merge.zip")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var vers []metaCacheEntry
+	zr, err := zip.NewReader(bytes.NewReader(dataZ), int64(len(dataZ)))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	for _, file := range zr.File {
+		if file.UncompressedSize64 == 0 {
+			continue
+		}
+		in, err := file.Open()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer in.Close()
+		buf, err := io.ReadAll(in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		buf = xlMetaV2TrimData(buf)
+
+		vers = append(vers, metaCacheEntry{
+			name:     "a",
+			metadata: buf,
+		})
+	}
+
+	// Shuffle...
+	for i := 0; i < 100; i++ {
+		rng := rand.New(rand.NewSource(int64(i)))
+		rng.Shuffle(len(vers), func(i, j int) {
+			vers[i], vers[j] = vers[j], vers[i]
+		})
+		var entries []chan metaCacheEntry
+		for _, v := range vers {
+			v.cached = nil
+			ch := make(chan metaCacheEntry, 1)
+			ch <- v
+			close(ch)
+			entries = append(entries, ch)
+		}
+		out := make(chan metaCacheEntry, 1)
+		err := mergeEntryChannels(context.Background(), entries, out, 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		got, ok := <-out
+		if !ok {
+			t.Fatal("Got no result")
+		}
+
+		xl, err := got.xlmeta()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if len(xl.versions) != 3 {
+			t.Fatal("Got wrong number of versions, want 3, got", len(xl.versions))
+		}
+		if !sort.SliceIsSorted(xl.versions, func(i, j int) bool {
+			return xl.versions[i].header.sortsBefore(xl.versions[j].header)
+		}) {
+			t.Errorf("Got unsorted result")
+		}
+	}
+}
+
 func TestXMinIOHealingSkip(t *testing.T) {
 	xl := xlMetaV2{}
 	failOnErr := func(err error) {

cmd/xl-storage.go

@@ -120,6 +120,8 @@ type xlStorage struct {
 	nrRequests   uint64
 	major, minor uint32
 
+	immediatePurge chan string
+
 	// mutex to prevent concurrent read operations overloading walks.
 	rotational bool
 	walkMu     *sync.Mutex
@@ -221,15 +223,26 @@ func makeFormatErasureMetaVolumes(disk StorageAPI) error {
 
 // Initialize a new storage disk.
 func newXLStorage(ep Endpoint, cleanUp bool) (s *xlStorage, err error) {
-	s = &xlStorage{
-		drivePath:     ep.Path,
-		endpoint:      ep,
-		globalSync:    globalFSOSync,
-		diskInfoCache: cachevalue.New[DiskInfo](),
-		poolIndex:     -1,
-		setIndex:      -1,
-		diskIndex:     -1,
+	immediatePurgeQueue := 100000
+	if globalIsTesting || globalIsCICD {
+		immediatePurgeQueue = 1
 	}
+	s = &xlStorage{
+		drivePath:      ep.Path,
+		endpoint:       ep,
+		globalSync:     globalFSOSync,
+		diskInfoCache:  cachevalue.New[DiskInfo](),
+		poolIndex:      -1,
+		setIndex:       -1,
+		diskIndex:      -1,
+		immediatePurge: make(chan string, immediatePurgeQueue),
+	}
+
+	defer func() {
+		if err == nil {
+			go s.cleanupTrashImmediateCallers(GlobalContext)
+		}
+	}()
 
 	s.drivePath, err = getValidPath(ep.Path)
 	if err != nil {
@@ -311,7 +324,7 @@ func newXLStorage(ep Endpoint, cleanUp bool) (s *xlStorage, err error) {
 	// oDirect off.
 	if globalIsErasureSD || !disk.ODirectPlatform {
 		s.oDirect = false
-	} else if err := s.checkODirectDiskSupport(); err == nil {
+	} else if err := s.checkODirectDiskSupport(info.FSType); err == nil {
 		s.oDirect = true
 	} else {
 		return s, err
@@ -405,11 +418,18 @@ func (s *xlStorage) Healing() *healingTracker {
 // checkODirectDiskSupport asks the disk to write some data
 // with O_DIRECT support, return an error if any and return
 // errUnsupportedDisk if there is no O_DIRECT support
-func (s *xlStorage) checkODirectDiskSupport() error {
+func (s *xlStorage) checkODirectDiskSupport(fsType string) error {
 	if !disk.ODirectPlatform {
 		return errUnsupportedDisk
 	}
 
+	// We know XFS already supports O_DIRECT no need to check.
+	if fsType == "XFS" {
+		return nil
+	}
+
+	// For all other FS pay the price of not using our recommended filesystem.
+
 	// Check if backend is writable and supports O_DIRECT
 	uuid := mustGetUUID()
 	filePath := pathJoin(s.drivePath, minioMetaTmpDeletedBucket, ".writable-check-"+uuid+".tmp")
@@ -1165,6 +1185,17 @@ func (s *xlStorage) DeleteVersions(ctx context.Context, volume string, versions
 	return errs
 }
 
+func (s *xlStorage) cleanupTrashImmediateCallers(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case entry := <-s.immediatePurge:
+			removeAll(entry)
+		}
+	}
+}
+
 func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool) (err error) {
 	pathUUID := mustGetUUID()
 	targetPath := pathutil.Join(s.drivePath, minioMetaTmpDeletedBucket, pathUUID)
@@ -1175,6 +1206,13 @@ func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool)
 		err = Rename(filePath, targetPath)
 	}
 
+	var targetPath2 string
+	if immediatePurge && HasSuffix(filePath, SlashSeparator) {
+		// With immediate purge also attempt deleting for `__XL_DIR__` folder/directory objects.
+		targetPath2 = pathutil.Join(s.drivePath, minioMetaTmpDeletedBucket, mustGetUUID())
+		renameAll(encodeDirObject(filePath), targetPath2, pathutil.Join(s.drivePath, minioMetaBucket))
+	}
+
 	// ENOSPC is a valid error from rename(); remove instead of rename in that case
 	if errors.Is(err, errDiskFull) || isSysErrNoSpace(err) {
 		if recursive {
@@ -1191,7 +1229,21 @@ func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool)
 
 	// immediately purge the target
 	if immediatePurge {
-		removeAll(targetPath)
+		for _, target := range []string{
+			targetPath,
+			targetPath2,
+		} {
+			if target == "" {
+				continue
+			}
+			select {
+			case s.immediatePurge <- target:
+			default:
+				// Too much back pressure, we will perform the delete
+				// blocking at this point we need to serialize operations.
+				removeAll(target)
+			}
+		}
 	}
 
 	return nil
@@ -2261,10 +2313,10 @@ func (s *xlStorage) deleteFile(basePath, deletePath string, recursive, immediate
 	if basePath == "" || deletePath == "" {
 		return nil
 	}
-	isObjectDir := HasSuffix(deletePath, SlashSeparator)
-	basePath = pathutil.Clean(basePath)
-	deletePath = pathutil.Clean(deletePath)
-	if !strings.HasPrefix(deletePath, basePath) || deletePath == basePath {
+
+	bp := pathutil.Clean(basePath) // do not override basepath / or deletePath /
+	dp := pathutil.Clean(deletePath)
+	if !strings.HasPrefix(dp, bp) || dp == bp {
 		return nil
 	}
 
@@ -2279,7 +2331,7 @@ func (s *xlStorage) deleteFile(basePath, deletePath string, recursive, immediate
 		case isSysErrNotEmpty(err):
 			// if object is a directory, but if its not empty
 			// return FileNotFound to indicate its an empty prefix.
-			if isObjectDir {
+			if HasSuffix(deletePath, SlashSeparator) {
 				return errFileNotFound
 			}
 			// if we have .DS_Store only on macOS
@@ -2311,11 +2363,9 @@ func (s *xlStorage) deleteFile(basePath, deletePath string, recursive, immediate
 		}
 	}
 
-	deletePath = pathutil.Dir(deletePath)
-
 	// Delete parent directory obviously not recursively. Errors for
 	// parent directories shouldn't trickle down.
-	s.deleteFile(basePath, deletePath, false, false)
+	s.deleteFile(basePath, pathutil.Dir(pathutil.Clean(deletePath)), false, false)
 
 	return nil
 }

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.2024-03-05T04-48-44Z
+  image: quay.io/minio/minio:RELEASE.2024-03-07T00-43-48Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

go.mod

@@ -51,7 +51,7 @@ require (
 	github.com/minio/dperf v0.5.3
 	github.com/minio/highwayhash v1.0.2
 	github.com/minio/kms-go/kes v0.3.0
-	github.com/minio/madmin-go/v3 v3.0.49
+	github.com/minio/madmin-go/v3 v3.0.50-0.20240307080957-112c599cb563
 	github.com/minio/minio-go/v7 v7.0.68
 	github.com/minio/mux v1.9.0
 	github.com/minio/pkg/v2 v2.0.9-0.20240209124402-7990a27fd79d
@@ -91,10 +91,10 @@ require (
 	go.uber.org/atomic v1.11.0
 	go.uber.org/zap v1.26.0
 	goftp.io/server/v2 v2.0.1
-	golang.org/x/crypto v0.19.0
+	golang.org/x/crypto v0.21.0
 	golang.org/x/exp v0.0.0-20240205201215-2c58cdc269a3
 	golang.org/x/oauth2 v0.17.0
-	golang.org/x/sys v0.17.0
+	golang.org/x/sys v0.18.0
 	golang.org/x/time v0.5.0
 	google.golang.org/api v0.164.0
 	gopkg.in/yaml.v2 v2.4.0
@@ -182,7 +182,7 @@ require (
 	github.com/lestrrat-go/blackmagic v1.0.2 // indirect
 	github.com/lestrrat-go/httpcc v1.0.1 // indirect
 	github.com/lestrrat-go/iter v1.0.2 // indirect
-	github.com/lestrrat-go/jwx v1.2.28 // indirect
+	github.com/lestrrat-go/jwx v1.2.29 // indirect
 	github.com/lestrrat-go/option v1.0.1 // indirect
 	github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20231016141302-07b5767bb0ed // indirect
@@ -244,7 +244,7 @@ require (
 	golang.org/x/mod v0.15.0 // indirect
 	golang.org/x/net v0.21.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect
-	golang.org/x/term v0.17.0 // indirect
+	golang.org/x/term v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/tools v0.18.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect

go.sum

@@ -385,8 +385,8 @@ github.com/lestrrat-go/httpcc v1.0.1 h1:ydWCStUeJLkpYyjLDHihupbn2tYmZ7m22BGkcvZZ
 github.com/lestrrat-go/httpcc v1.0.1/go.mod h1:qiltp3Mt56+55GPVCbTdM9MlqhvzyuL6W/NMDA8vA5E=
 github.com/lestrrat-go/iter v1.0.2 h1:gMXo1q4c2pHmC3dn8LzRhJfP1ceCbgSiT9lUydIzltI=
 github.com/lestrrat-go/iter v1.0.2/go.mod h1:Momfcq3AnRlRjI5b5O8/G5/BvpzrhoFTZcn06fEOPt4=
-github.com/lestrrat-go/jwx v1.2.28 h1:uadI6o0WpOVrBSf498tRXZIwPpEtLnR9CvqPFXeI5sA=
-github.com/lestrrat-go/jwx v1.2.28/go.mod h1:nF+91HEMh/MYFVwKPl5HHsBGMPscqbQb+8IDQdIazP8=
+github.com/lestrrat-go/jwx v1.2.29 h1:QT0utmUJ4/12rmsVQrJ3u55bycPkKqGYuGT4tyRhxSQ=
+github.com/lestrrat-go/jwx v1.2.29/go.mod h1:hU8k2l6WF0ncx20uQdOmik/Gjg6E3/wIRtXSNFeZuB8=
 github.com/lestrrat-go/option v1.0.0/go.mod h1:5ZHFbivi4xwXxhxY9XHDe2FHo6/Z7WWmtT7T5nBBp3I=
 github.com/lestrrat-go/option v1.0.1 h1:oAzP2fvZGQKWkvHa1/SAcFolBEca1oN+mQ7eooNBEYU=
 github.com/lestrrat-go/option v1.0.1/go.mod h1:5ZHFbivi4xwXxhxY9XHDe2FHo6/Z7WWmtT7T5nBBp3I=
@@ -446,8 +446,8 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/minio/kms-go/kes v0.3.0 h1:SU8VGVM/Hk9w1OiSby3OatkcojooUqIdDHl6dtM6NkY=
 github.com/minio/kms-go/kes v0.3.0/go.mod h1:w6DeVT878qEOU3nUrYVy1WOT5H1Ig9hbDIh698NYJKY=
-github.com/minio/madmin-go/v3 v3.0.49 h1:Ag5eyYUf9K1MvW9hiErEJhGfqlf//pOtlhdoepb9AwY=
-github.com/minio/madmin-go/v3 v3.0.49/go.mod h1:ZDF7kf5fhmxLhbGTqyq5efs4ao0v4eWf7nOuef/ljJs=
+github.com/minio/madmin-go/v3 v3.0.50-0.20240307080957-112c599cb563 h1:Sewy11CUNITOocZ2KUfW0l2zm0wNp+d8tb22ivztGRg=
+github.com/minio/madmin-go/v3 v3.0.50-0.20240307080957-112c599cb563/go.mod h1:ZDF7kf5fhmxLhbGTqyq5efs4ao0v4eWf7nOuef/ljJs=
 github.com/minio/mc v0.0.0-20240209221824-669cb0a9a475 h1:yfLzMougcV2xkVlWgwYwVRoT8pnXrcCV4oOQW+pI2EQ=
 github.com/minio/mc v0.0.0-20240209221824-669cb0a9a475/go.mod h1:MmDLdb7NWd/OYhcKcXKvwErq2GNa/Zq6xtTWuhdC4II=
 github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
@@ -634,6 +634,7 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
@@ -643,8 +644,9 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/tidwall/gjson v1.17.0 h1:/Jocvlh98kcTfpN2+JzGQWQcqrPQwDrVEMApx/M5ZwM=
 github.com/tidwall/gjson v1.17.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
@@ -744,9 +746,9 @@ golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw
 golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
 golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
 golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
-golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
-golang.org/x/crypto v0.19.0 h1:ENy+Az/9Y1vSrlrvBSyna3PITt4tiZLf7sgCjZBX7Wo=
 golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
+golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=
+golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20240205201215-2c58cdc269a3 h1:/RIbNt/Zr7rVhIkQhooTxCxFcdWLGIKnZA4IXNFSrvo=
 golang.org/x/exp v0.0.0-20240205201215-2c58cdc269a3/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08=
@@ -842,8 +844,9 @@ golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
 golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
+golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -853,8 +856,9 @@ golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
 golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
 golang.org/x/term v0.15.0/go.mod h1:BDl952bC7+uMoWR75FIrCDx79TPU9oHkTZ9yRbYOrX0=
-golang.org/x/term v0.17.0 h1:mkTF7LCd6WGJNL3K1Ad7kwxNfYAW6a8a8QqtMblp/4U=
 golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
+golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8=
+golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=

internal/bucket/lifecycle/lifecycle.go

@@ -471,9 +471,11 @@ func (lc Lifecycle) eval(obj ObjectOpts, now time.Time) Event {
 
 	if len(events) > 0 {
 		sort.Slice(events, func(i, j int) bool {
-			if events[i].Due.Equal(events[j].Due) {
-				// Prefer Expiration over Transition for both current
-				// and noncurrent versions
+			// Prefer Expiration over Transition for both current
+			// and noncurrent versions when,
+			// - now is past the expected time to action
+			// - expected time to action is the same for both actions
+			if now.After(events[i].Due) && now.After(events[j].Due) || events[i].Due.Equal(events[j].Due) {
 				switch events[i].Action {
 				case DeleteAction, DeleteVersionAction:
 					return true

internal/bucket/lifecycle/lifecycle_test.go

@@ -539,6 +539,46 @@ func TestEval(t *testing.T) {
 			objectModTime:  time.Now().UTC().Add(-15 * 24 * time.Hour),
 			expectedAction: DeleteAction,
 		},
+		{
+			inputConfig: `<LifecycleConfiguration>
+                            <Rule>
+                              <ID>Rule 1</ID>
+                              <Status>Enabled</Status>
+                              <Filter></Filter>
+                              <Transition>
+                                <StorageClass>WARM-1</StorageClass>
+                                <Days>30</Days>
+                              </Transition>
+                              <Expiration>
+                                <Days>60</Days>
+                              </Expiration>
+                            </Rule>
+                       </LifecycleConfiguration>`,
+			objectName:     "obj-1",
+			objectModTime:  time.Now().UTC().Add(-90 * 24 * time.Hour),
+			expectedAction: DeleteAction,
+		},
+		{
+			inputConfig: `<LifecycleConfiguration>
+                            <Rule>
+                              <ID>Rule 2</ID>
+                              <Filter></Filter>
+                              <Status>Enabled</Status>
+                              <NoncurrentVersionExpiration>
+                                <NoncurrentDays>60</NoncurrentDays>
+                              </NoncurrentVersionExpiration>
+	                      <NoncurrentVersionTransition>
+                                <StorageClass>WARM-1</StorageClass>
+                                <NoncurrentDays>30</NoncurrentDays>
+                              </NoncurrentVersionTransition>
+                             </Rule>
+                       </LifecycleConfiguration>`,
+			objectName:             "obj-1",
+			isNoncurrent:           true,
+			objectModTime:          time.Now().UTC().Add(-90 * 24 * time.Hour),
+			objectSuccessorModTime: time.Now().UTC().Add(-90 * 24 * time.Hour),
+			expectedAction:         DeleteVersionAction,
+		},
 	}
 
 	for _, tc := range testCases {

internal/bucket/replication/datatypes.go

@@ -22,6 +22,9 @@ package replication
 // StatusType of Replication for x-amz-replication-status header
 type StatusType string
 
+// Type - replication type enum
+type Type int
+
 const (
 	// Pending - replication is pending.
 	Pending StatusType = "PENDING"

internal/bucket/replication/datatypes_gen.go

@@ -57,3 +57,55 @@ func (z StatusType) Msgsize() (s int) {
 	s = msgp.StringPrefixSize + len(string(z))
 	return
 }
+
+// DecodeMsg implements msgp.Decodable
+func (z *Type) DecodeMsg(dc *msgp.Reader) (err error) {
+	{
+		var zb0001 int
+		zb0001, err = dc.ReadInt()
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		(*z) = Type(zb0001)
+	}
+	return
+}
+
+// EncodeMsg implements msgp.Encodable
+func (z Type) EncodeMsg(en *msgp.Writer) (err error) {
+	err = en.WriteInt(int(z))
+	if err != nil {
+		err = msgp.WrapError(err)
+		return
+	}
+	return
+}
+
+// MarshalMsg implements msgp.Marshaler
+func (z Type) MarshalMsg(b []byte) (o []byte, err error) {
+	o = msgp.Require(b, z.Msgsize())
+	o = msgp.AppendInt(o, int(z))
+	return
+}
+
+// UnmarshalMsg implements msgp.Unmarshaler
+func (z *Type) UnmarshalMsg(bts []byte) (o []byte, err error) {
+	{
+		var zb0001 int
+		zb0001, bts, err = msgp.ReadIntBytes(bts)
+		if err != nil {
+			err = msgp.WrapError(err)
+			return
+		}
+		(*z) = Type(zb0001)
+	}
+	o = bts
+	return
+}
+
+// Msgsize returns an upper bound estimate of the number of bytes occupied by the serialized message
+func (z Type) Msgsize() (s int) {
+	s = msgp.IntSize
+	return
+}

internal/bucket/replication/replication.go

@@ -118,9 +118,6 @@ func (c Config) Validate(bucket string, sameTarget bool) error {
 	return nil
 }
 
-// Type - replication type enum
-type Type int
-
 // Types of replication
 const (
 	UnsetReplicationType Type = 0 + iota

internal/grid/handlers.go

@@ -110,6 +110,7 @@ const (
 	HandlerSignalService
 	HandlerGetBandwidth
 	HandlerWriteAll
+	HandlerListBuckets
 
 	// Add more above here ^^^
 	// If all handlers are used, the type of Handler can be changed.
@@ -187,6 +188,7 @@ var handlerPrefixes = [handlerLast]string{
 	HandlerStorageInfo:                 peerPrefix,
 	HandlerConsoleLog:                  peerPrefix,
 	HandlerListDir:                     storagePrefix,
+	HandlerListBuckets:                 peerPrefixS3,
 }
 
 const (

internal/grid/handlers_string.go

@@ -79,14 +79,15 @@ func _() {
 	_ = x[HandlerSignalService-68]
 	_ = x[HandlerGetBandwidth-69]
 	_ = x[HandlerWriteAll-70]
-	_ = x[handlerTest-71]
-	_ = x[handlerTest2-72]
-	_ = x[handlerLast-73]
+	_ = x[HandlerListBuckets-71]
+	_ = x[handlerTest-72]
+	_ = x[handlerTest2-73]
+	_ = x[handlerLast-74]
 }
 
-const _HandlerID_name = "handlerInvalidLockLockLockRLockLockUnlockLockRUnlockLockRefreshLockForceUnlockWalkDirStatVolDiskInfoNSScannerReadXLReadVersionDeleteFileDeleteVersionUpdateMetadataWriteMetadataCheckPartsRenameDataRenameFileReadAllServerVerifyTraceListenDeleteBucketMetadataLoadBucketMetadataReloadSiteReplicationConfigReloadPoolMetaStopRebalanceLoadRebalanceMetaLoadTransitionTierConfigDeletePolicyLoadPolicyLoadPolicyMappingDeleteServiceAccountLoadServiceAccountDeleteUserLoadUserLoadGroupHealBucketMakeBucketHeadBucketDeleteBucketGetMetricsGetResourceMetricsGetMemInfoGetProcInfoGetOSInfoGetPartitionsGetNetInfoGetCPUsServerInfoGetSysConfigGetSysServicesGetSysErrorsGetAllBucketStatsGetBucketStatsGetSRMetricsGetPeerMetricsGetMetacacheListingUpdateMetacacheListingGetPeerBucketMetricsStorageInfoConsoleLogListDirGetLocksBackgroundHealStatusGetLastDayTierStatsSignalServiceGetBandwidthWriteAllhandlerTesthandlerTest2handlerLast"
+const _HandlerID_name = "handlerInvalidLockLockLockRLockLockUnlockLockRUnlockLockRefreshLockForceUnlockWalkDirStatVolDiskInfoNSScannerReadXLReadVersionDeleteFileDeleteVersionUpdateMetadataWriteMetadataCheckPartsRenameDataRenameFileReadAllServerVerifyTraceListenDeleteBucketMetadataLoadBucketMetadataReloadSiteReplicationConfigReloadPoolMetaStopRebalanceLoadRebalanceMetaLoadTransitionTierConfigDeletePolicyLoadPolicyLoadPolicyMappingDeleteServiceAccountLoadServiceAccountDeleteUserLoadUserLoadGroupHealBucketMakeBucketHeadBucketDeleteBucketGetMetricsGetResourceMetricsGetMemInfoGetProcInfoGetOSInfoGetPartitionsGetNetInfoGetCPUsServerInfoGetSysConfigGetSysServicesGetSysErrorsGetAllBucketStatsGetBucketStatsGetSRMetricsGetPeerMetricsGetMetacacheListingUpdateMetacacheListingGetPeerBucketMetricsStorageInfoConsoleLogListDirGetLocksBackgroundHealStatusGetLastDayTierStatsSignalServiceGetBandwidthWriteAllListBucketshandlerTesthandlerTest2handlerLast"
 
-var _HandlerID_index = [...]uint16{0, 14, 22, 31, 41, 52, 63, 78, 85, 92, 100, 109, 115, 126, 136, 149, 163, 176, 186, 196, 206, 213, 225, 230, 236, 256, 274, 301, 315, 328, 345, 369, 381, 391, 408, 428, 446, 456, 464, 473, 483, 493, 503, 515, 525, 543, 553, 564, 573, 586, 596, 603, 613, 625, 639, 651, 668, 682, 694, 708, 727, 749, 769, 780, 790, 797, 805, 825, 844, 857, 869, 877, 888, 900, 911}
+var _HandlerID_index = [...]uint16{0, 14, 22, 31, 41, 52, 63, 78, 85, 92, 100, 109, 115, 126, 136, 149, 163, 176, 186, 196, 206, 213, 225, 230, 236, 256, 274, 301, 315, 328, 345, 369, 381, 391, 408, 428, 446, 456, 464, 473, 483, 493, 503, 515, 525, 543, 553, 564, 573, 586, 596, 603, 613, 625, 639, 651, 668, 682, 694, 708, 727, 749, 769, 780, 790, 797, 805, 825, 844, 857, 869, 877, 888, 899, 911, 922}
 
 func (i HandlerID) String() string {
 	if i >= HandlerID(len(_HandlerID_index)-1) {

internal/logger/config.go

@@ -43,6 +43,7 @@ const (
 	AuthToken  = "auth_token"
 	ClientCert = "client_cert"
 	ClientKey  = "client_key"
+	BatchSize  = "batch_size"
 	QueueSize  = "queue_size"
 	QueueDir   = "queue_dir"
 	Proxy      = "proxy"
@@ -68,6 +69,7 @@ const (
 	EnvLoggerWebhookClientCert = "MINIO_LOGGER_WEBHOOK_CLIENT_CERT"
 	EnvLoggerWebhookClientKey  = "MINIO_LOGGER_WEBHOOK_CLIENT_KEY"
 	EnvLoggerWebhookProxy      = "MINIO_LOGGER_WEBHOOK_PROXY"
+	EnvLoggerWebhookBatchSize  = "MINIO_LOGGER_WEBHOOK_BATCH_SIZE"
 	EnvLoggerWebhookQueueSize  = "MINIO_LOGGER_WEBHOOK_QUEUE_SIZE"
 	EnvLoggerWebhookQueueDir   = "MINIO_LOGGER_WEBHOOK_QUEUE_DIR"
 
@@ -76,6 +78,7 @@ const (
 	EnvAuditWebhookAuthToken  = "MINIO_AUDIT_WEBHOOK_AUTH_TOKEN"
 	EnvAuditWebhookClientCert = "MINIO_AUDIT_WEBHOOK_CLIENT_CERT"
 	EnvAuditWebhookClientKey  = "MINIO_AUDIT_WEBHOOK_CLIENT_KEY"
+	EnvAuditWebhookBatchSize  = "MINIO_AUDIT_WEBHOOK_BATCH_SIZE"
 	EnvAuditWebhookQueueSize  = "MINIO_AUDIT_WEBHOOK_QUEUE_SIZE"
 	EnvAuditWebhookQueueDir   = "MINIO_AUDIT_WEBHOOK_QUEUE_DIR"
 
@@ -99,7 +102,10 @@ const (
 	auditTargetNamePrefix  = "audit-"
 )
 
-var errInvalidQueueSize = errors.New("invalid queue_size value")
+var (
+	errInvalidQueueSize = errors.New("invalid queue_size value")
+	errInvalidBatchSize = errors.New("invalid batch_size value")
+)
 
 // Default KVS for loggerHTTP and loggerAuditHTTP
 var (
@@ -128,6 +134,10 @@ var (
 			Key:   Proxy,
 			Value: "",
 		},
+		config.KV{
+			Key:   BatchSize,
+			Value: "1",
+		},
 		config.KV{
 			Key:   QueueSize,
 			Value: "100000",
@@ -159,6 +169,10 @@ var (
 			Key:   ClientKey,
 			Value: "",
 		},
+		config.KV{
+			Key:   BatchSize,
+			Value: "1",
+		},
 		config.KV{
 			Key:   QueueSize,
 			Value: "100000",
@@ -435,6 +449,14 @@ func lookupLoggerWebhookConfig(scfg config.Config, cfg Config) (Config, error) {
 		if queueSize <= 0 {
 			return cfg, errInvalidQueueSize
 		}
+		batchSizeCfgVal := getCfgVal(EnvLoggerWebhookBatchSize, k, kv.Get(BatchSize))
+		batchSize, err := strconv.Atoi(batchSizeCfgVal)
+		if err != nil {
+			return cfg, err
+		}
+		if batchSize <= 0 {
+			return cfg, errInvalidBatchSize
+		}
 		cfg.HTTP[k] = http.Config{
 			Enabled:    true,
 			Endpoint:   url,
@@ -442,6 +464,7 @@ func lookupLoggerWebhookConfig(scfg config.Config, cfg Config) (Config, error) {
 			ClientCert: clientCert,
 			ClientKey:  clientKey,
 			Proxy:      getCfgVal(EnvLoggerWebhookProxy, k, kv.Get(Proxy)),
+			BatchSize:  batchSize,
 			QueueSize:  queueSize,
 			QueueDir:   getCfgVal(EnvLoggerWebhookQueueDir, k, kv.Get(QueueDir)),
 			Name:       loggerTargetNamePrefix + k,
@@ -488,12 +511,21 @@ func lookupAuditWebhookConfig(scfg config.Config, cfg Config) (Config, error) {
 		if queueSize <= 0 {
 			return cfg, errInvalidQueueSize
 		}
+		batchSizeCfgVal := getCfgVal(EnvAuditWebhookBatchSize, k, kv.Get(BatchSize))
+		batchSize, err := strconv.Atoi(batchSizeCfgVal)
+		if err != nil {
+			return cfg, err
+		}
+		if batchSize <= 0 {
+			return cfg, errInvalidBatchSize
+		}
 		cfg.AuditWebhook[k] = http.Config{
 			Enabled:    true,
 			Endpoint:   url,
 			AuthToken:  getCfgVal(EnvAuditWebhookAuthToken, k, kv.Get(AuthToken)),
 			ClientCert: clientCert,
 			ClientKey:  clientKey,
+			BatchSize:  batchSize,
 			QueueSize:  queueSize,
 			QueueDir:   getCfgVal(EnvAuditWebhookQueueDir, k, kv.Get(QueueDir)),
 			Name:       auditTargetNamePrefix + k,

internal/logger/help.go

@@ -52,6 +52,12 @@ var (
 			Type:        "string",
 			Sensitive:   true,
 		},
+		config.HelpKV{
+			Key:         BatchSize,
+			Description: "Number of events per HTTP send to webhook target",
+			Optional:    true,
+			Type:        "number",
+		},
 		config.HelpKV{
 			Key:         QueueSize,
 			Description: "configure channel queue size for webhook targets",
@@ -107,6 +113,12 @@ var (
 			Type:        "string",
 			Sensitive:   true,
 		},
+		config.HelpKV{
+			Key:         BatchSize,
+			Description: "Number of events per HTTP send to webhook target",
+			Optional:    true,
+			Type:        "number",
+		},
 		config.HelpKV{
 			Key:         QueueSize,
 			Description: "configure channel queue size for webhook targets",

internal/logger/target/http/http.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2015-2023 MinIO, Inc.
+// Copyright (c) 2015-2024 MinIO, Inc.
 //
 // This file is part of MinIO Object Storage stack
 //
@@ -33,12 +33,14 @@ import (
 	"sync/atomic"
 	"time"
 
+	jsoniter "github.com/json-iterator/go"
 	xhttp "github.com/minio/minio/internal/http"
 	xioutil "github.com/minio/minio/internal/ioutil"
 	"github.com/minio/minio/internal/logger/target/types"
 	"github.com/minio/minio/internal/once"
 	"github.com/minio/minio/internal/store"
 	xnet "github.com/minio/pkg/v2/net"
+	"github.com/valyala/bytebufferpool"
 )
 
 const (
@@ -48,6 +50,9 @@ const (
 	// maxWorkers is the maximum number of concurrent http loggers
 	maxWorkers = 16
 
+	// maxWorkers is the maximum number of concurrent batch http loggers
+	maxWorkersWithBatchEvents = 4
+
 	// the suffix for the configured queue dir where the logs will be persisted.
 	httpLoggerExtension = ".http.log"
 )
@@ -67,6 +72,7 @@ type Config struct {
 	AuthToken  string            `json:"authToken"`
 	ClientCert string            `json:"clientCert"`
 	ClientKey  string            `json:"clientKey"`
+	BatchSize  int               `json:"batchSize"`
 	QueueSize  int               `json:"queueSize"`
 	QueueDir   string            `json:"queueDir"`
 	Proxy      string            `json:"string"`
@@ -99,6 +105,13 @@ type Target struct {
 	logCh   chan interface{}
 	logChMu sync.RWMutex
 
+	// Number of events per HTTP send to webhook target
+	// this is ideally useful only if your endpoint can
+	// support reading multiple events on a stream for example
+	// like : Splunk HTTP Event collector, if you are unsure
+	// set this to '1'.
+	batchSize int
+
 	// If the first init fails, this starts a goroutine that
 	// will attempt to establish the connection.
 	revive sync.Once
@@ -135,7 +148,7 @@ func (h *Target) IsOnline(ctx context.Context) bool {
 
 // ping returns true if the target is reachable.
 func (h *Target) ping(ctx context.Context) bool {
-	if err := h.send(ctx, []byte(`{}`), webhookCallTimeout); err != nil {
+	if err := h.send(ctx, []byte(`{}`), "application/json", webhookCallTimeout); err != nil {
 		return !xnet.IsNetworkOrHostDown(err, false) && !xnet.IsConnRefusedErr(err)
 	}
 	// We are online.
@@ -213,7 +226,7 @@ func (h *Target) init(ctx context.Context) (err error) {
 	return nil
 }
 
-func (h *Target) send(ctx context.Context, payload []byte, timeout time.Duration) (err error) {
+func (h *Target) send(ctx context.Context, payload []byte, payloadType string, timeout time.Duration) (err error) {
 	defer func() {
 		if err != nil {
 			atomic.StoreInt32(&h.status, statusOffline)
@@ -229,7 +242,9 @@ func (h *Target) send(ctx context.Context, payload []byte, timeout time.Duration
 	if err != nil {
 		return fmt.Errorf("invalid configuration for '%s'; %v", h.Endpoint(), err)
 	}
-	req.Header.Set(xhttp.ContentType, "application/json")
+	if payloadType != "" {
+		req.Header.Set(xhttp.ContentType, payloadType)
+	}
 	req.Header.Set(xhttp.MinIOVersion, xhttp.GlobalMinIOVersion)
 	req.Header.Set(xhttp.MinioDeploymentID, xhttp.GlobalDeploymentID)
 
@@ -260,13 +275,7 @@ func (h *Target) send(ctx context.Context, payload []byte, timeout time.Duration
 	}
 }
 
-func (h *Target) logEntry(ctx context.Context, entry interface{}) {
-	logJSON, err := json.Marshal(&entry)
-	if err != nil {
-		atomic.AddInt64(&h.failedMessages, 1)
-		return
-	}
-
+func (h *Target) logEntry(ctx context.Context, payload []byte, payloadType string) {
 	const maxTries = 3
 	tries := 0
 	for tries < maxTries {
@@ -281,7 +290,7 @@ func (h *Target) logEntry(ctx context.Context, entry interface{}) {
 		}
 		time.Sleep(sleep)
 		tries++
-		err := h.send(ctx, logJSON, webhookCallTimeout)
+		err := h.send(ctx, payload, payloadType, webhookCallTimeout)
 		if err == nil {
 			return
 		}
@@ -309,9 +318,36 @@ func (h *Target) startHTTPLogger(ctx context.Context) {
 		return
 	}
 
+	buf := bytebufferpool.Get()
+	defer bytebufferpool.Put(buf)
+
+	json := jsoniter.ConfigCompatibleWithStandardLibrary
+	enc := json.NewEncoder(buf)
+	batchSize := h.batchSize
+	if batchSize <= 0 {
+		batchSize = 1
+	}
+
+	payloadType := "application/json"
+	if batchSize > 1 {
+		payloadType = ""
+	}
+
+	var nevents int
 	// Send messages until channel is closed.
 	for entry := range logCh {
-		h.logEntry(ctx, entry)
+		atomic.AddInt64(&h.totalMessages, 1)
+		nevents++
+		if err := enc.Encode(&entry); err != nil {
+			atomic.AddInt64(&h.failedMessages, 1)
+			nevents--
+			continue
+		}
+		if (nevents == batchSize || len(logCh) == 0) && buf.Len() > 0 {
+			h.logEntry(ctx, buf.Bytes(), payloadType)
+			buf.Reset()
+			nevents = 0
+		}
 	}
 }
 
@@ -319,9 +355,10 @@ func (h *Target) startHTTPLogger(ctx context.Context) {
 // sends log over http to the specified endpoint
 func New(config Config) *Target {
 	h := &Target{
-		logCh:  make(chan interface{}, config.QueueSize),
-		config: config,
-		status: statusOffline,
+		logCh:     make(chan interface{}, config.QueueSize),
+		config:    config,
+		status:    statusOffline,
+		batchSize: config.BatchSize,
 	}
 
 	// If proxy available, set the same
@@ -353,7 +390,7 @@ func (h *Target) SendFromStore(key store.Key) (err error) {
 		atomic.AddInt64(&h.failedMessages, 1)
 		return
 	}
-	if err := h.send(context.Background(), logJSON, webhookCallTimeout); err != nil {
+	if err := h.send(context.Background(), logJSON, "application/json", webhookCallTimeout); err != nil {
 		atomic.AddInt64(&h.failedMessages, 1)
 		if xnet.IsNetworkOrHostDown(err, true) {
 			return store.ErrNotConnected
@@ -382,6 +419,11 @@ func (h *Target) Send(ctx context.Context, entry interface{}) error {
 		return nil
 	}
 
+	mworkers := maxWorkers
+	if h.batchSize > 100 {
+		mworkers = maxWorkersWithBatchEvents
+	}
+
 retry:
 	select {
 	case h.logCh <- entry:
@@ -394,7 +436,7 @@ retry:
 		return nil
 	default:
 		nWorkers := atomic.LoadInt64(&h.workers)
-		if nWorkers < maxWorkers {
+		if nWorkers < int64(mworkers) {
 			// Only have one try to start at the same time.
 			h.workerStartMu.Lock()
 			if time.Since(h.lastStarted) > time.Second {