Split handleMessages (cosmetic) (#20095 )

Split the read and write sides of handleMessages into two separate functions Cosmetic. The only non-copy-and-paste change is that `cancel(ErrDisconnected)` is moved into the defer on `readStream`.
add validation test for v3 metrics for all its endpoints (#20094 )
2026-11-07 00:02:11 +03:00 · 2024-07-15 12:02:30 -07:00 · 2024-07-15 09:28:02 -07:00 · 2024-07-14 11:12:07 -07:00 · 2024-07-14 11:11:42 -07:00 · 2024-07-13 02:08:28 +00:00
47 changed files with 733 additions and 458 deletions
@@ -3,6 +3,8 @@ extend-exclude = [
    ".git/",
    "docs/",
    "CREDITS",
+    "go.mod",
+    "go.sum",
 ]
 ignore-hidden = false

@@ -17,6 +19,7 @@ extend-ignore-re = [
    "MIIDBTCCAe2gAwIBAgIQWHw7h.*",
    'http\.Header\{"X-Amz-Server-Side-Encryptio":',
    "ZoEoZdLlzVbOlT9rbhD7ZN7TLyiYXSAlB79uGEge",
+    "ERRO:",
 ]

 [default.extend-words]
@@ -101,7 +101,7 @@ function fail() {
 }

 function check_online() {
-	if ! grep -q 'Status:' ${WORK_DIR}/dist-minio-*.log; then
+	if ! grep -q 'API:' ${WORK_DIR}/dist-minio-*.log; then
 		echo "1"
 	fi
 }
@@ -78,7 +78,7 @@ function start_minio_3_node() {
 }

 function check_heal() {
-	if ! grep -q 'Status:' ${WORK_DIR}/dist-minio-*.log; then
+	if ! grep -q 'API:' ${WORK_DIR}/dist-minio-*.log; then
 		return 1
 	fi

@@ -552,7 +552,7 @@ func (a adminAPIHandlers) ListAccessKeysLDAPBulk(w http.ResponseWriter, r *http.
 		dnList = append(dnList, selfDN)
 	}

-	accessKeyMap := make(map[string]madmin.ListAccessKeysLDAPResp)
+	var ldapUserList []string
 	if isAll {
 		ldapUsers, err := globalIAMSys.ListLDAPUsers(ctx)
 		if err != nil {
@@ -560,7 +560,7 @@ func (a adminAPIHandlers) ListAccessKeysLDAPBulk(w http.ResponseWriter, r *http.
 			return
 		}
 		for user := range ldapUsers {
-			accessKeyMap[user] = madmin.ListAccessKeysLDAPResp{}
+			ldapUserList = append(ldapUserList, user)
 		}
 	} else {
 		for _, userDN := range dnList {
@@ -573,7 +573,7 @@ func (a adminAPIHandlers) ListAccessKeysLDAPBulk(w http.ResponseWriter, r *http.
 			if foundResult == nil {
 				continue
 			}
-			accessKeyMap[foundResult.NormDN] = madmin.ListAccessKeysLDAPResp{}
+			ldapUserList = append(ldapUserList, foundResult.NormDN)
 		}
 	}

@@ -598,9 +598,12 @@ func (a adminAPIHandlers) ListAccessKeysLDAPBulk(w http.ResponseWriter, r *http.
 		return
 	}

-	for dn, accessKeys := range accessKeyMap {
+	accessKeyMap := make(map[string]madmin.ListAccessKeysLDAPResp)
+	for _, internalDN := range ldapUserList {
+		externalDN := globalIAMSys.LDAPConfig.DecodeDN(internalDN)
+		accessKeys := madmin.ListAccessKeysLDAPResp{}
 		if listSTSKeys {
-			stsKeys, err := globalIAMSys.ListSTSAccounts(ctx, dn)
+			stsKeys, err := globalIAMSys.ListSTSAccounts(ctx, internalDN)
 			if err != nil {
 				writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 				return
@@ -614,13 +617,12 @@ func (a adminAPIHandlers) ListAccessKeysLDAPBulk(w http.ResponseWriter, r *http.
 			}
 			// if only STS keys, skip if user has no STS keys
 			if !listServiceAccounts && len(stsKeys) == 0 {
-				delete(accessKeyMap, dn)
 				continue
 			}
 		}

 		if listServiceAccounts {
-			serviceAccounts, err := globalIAMSys.ListServiceAccounts(ctx, dn)
+			serviceAccounts, err := globalIAMSys.ListServiceAccounts(ctx, internalDN)
 			if err != nil {
 				writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
 				return
@@ -634,11 +636,10 @@ func (a adminAPIHandlers) ListAccessKeysLDAPBulk(w http.ResponseWriter, r *http.
 			}
 			// if only service accounts, skip if user has no service accounts
 			if !listSTSKeys && len(serviceAccounts) == 0 {
-				delete(accessKeyMap, dn)
 				continue
 			}
 		}
-		accessKeyMap[dn] = accessKeys
+		accessKeyMap[externalDN] = accessKeys
 	}

 	data, err := json.Marshal(accessKeyMap)
@@ -2308,7 +2308,7 @@ func (a adminAPIHandlers) ImportIAM(w http.ResponseWriter, r *http.Request) {
 					// clean import.
 					err := globalIAMSys.DeleteServiceAccount(ctx, svcAcctReq.AccessKey, true)
 					if err != nil {
-						delErr := fmt.Errorf("failed to delete existing service account(%s) before importing it: %w", svcAcctReq.AccessKey, err)
+						delErr := fmt.Errorf("failed to delete existing service account (%s) before importing it: %w", svcAcctReq.AccessKey, err)
 						writeErrorResponseJSON(ctx, w, importError(ctx, delErr, allSvcAcctsFile, user), r.URL)
 						return
 					}
@@ -389,6 +389,17 @@ func (r *BatchJobKeyRotateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 					stopFn(result, err)
 					batchLogIf(ctx, err)
 					success = false
+					if attempts >= retryAttempts {
+						auditOptions := AuditLogOptions{
+							Event:     "KeyRotate",
+							APIName:   "StartBatchJob",
+							Bucket:    result.Bucket,
+							Object:    result.Name,
+							VersionID: result.VersionID,
+							Error:     err.Error(),
+						}
+						auditLogInternal(ctx, auditOptions)
+					}
 				} else {
 					stopFn(result, nil)
 				}
@@ -1803,15 +1803,18 @@ var (
 type ReplicationPool struct {
 	// atomic ops:
 	activeWorkers    int32
+	activeLrgWorkers int32
 	activeMRFWorkers int32

-	objLayer   ObjectLayer
-	ctx        context.Context
-	priority   string
-	maxWorkers int
-	mu         sync.RWMutex
-	mrfMU      sync.Mutex
-	resyncer   *replicationResyncer
+	objLayer    ObjectLayer
+	ctx         context.Context
+	priority    string
+	maxWorkers  int
+	maxLWorkers int
+
+	mu       sync.RWMutex
+	mrfMU    sync.Mutex
+	resyncer *replicationResyncer

 	// workers:
 	workers    []chan ReplicationWorkerOperation
@@ -1882,9 +1885,13 @@ func NewReplicationPool(ctx context.Context, o ObjectLayer, opts replicationPool
 	if maxWorkers > 0 && failedWorkers > maxWorkers {
 		failedWorkers = maxWorkers
 	}
+	maxLWorkers := LargeWorkerCount
+	if opts.MaxLWorkers > 0 {
+		maxLWorkers = opts.MaxLWorkers
+	}
 	pool := &ReplicationPool{
 		workers:         make([]chan ReplicationWorkerOperation, 0, workers),
-		lrgworkers:      make([]chan ReplicationWorkerOperation, 0, LargeWorkerCount),
+		lrgworkers:      make([]chan ReplicationWorkerOperation, 0, maxLWorkers),
 		mrfReplicaCh:    make(chan ReplicationWorkerOperation, 100000),
 		mrfWorkerKillCh: make(chan struct{}, failedWorkers),
 		resyncer:        newresyncer(),
@@ -1894,9 +1901,10 @@ func NewReplicationPool(ctx context.Context, o ObjectLayer, opts replicationPool
 		objLayer:        o,
 		priority:        priority,
 		maxWorkers:      maxWorkers,
+		maxLWorkers:     maxLWorkers,
 	}

-	pool.AddLargeWorkers()
+	pool.ResizeLrgWorkers(maxLWorkers, 0)
 	pool.ResizeWorkers(workers, 0)
 	pool.ResizeFailedWorkers(failedWorkers)
 	go pool.resyncer.PersistToDisk(ctx, o)
@@ -1975,23 +1983,8 @@ func (p *ReplicationPool) AddWorker(input <-chan ReplicationWorkerOperation, opT
 	}
 }

-// AddLargeWorkers adds a static number of workers to handle large uploads
-func (p *ReplicationPool) AddLargeWorkers() {
-	for i := 0; i < LargeWorkerCount; i++ {
-		p.lrgworkers = append(p.lrgworkers, make(chan ReplicationWorkerOperation, 100000))
-		i := i
-		go p.AddLargeWorker(p.lrgworkers[i])
-	}
-	go func() {
-		<-p.ctx.Done()
-		for i := 0; i < LargeWorkerCount; i++ {
-			xioutil.SafeClose(p.lrgworkers[i])
-		}
-	}()
-}
-
 // AddLargeWorker adds a replication worker to the static pool for large uploads.
-func (p *ReplicationPool) AddLargeWorker(input <-chan ReplicationWorkerOperation) {
+func (p *ReplicationPool) AddLargeWorker(input <-chan ReplicationWorkerOperation, opTracker *int32) {
 	for {
 		select {
 		case <-p.ctx.Done():
@@ -2002,11 +1995,23 @@ func (p *ReplicationPool) AddLargeWorker(input <-chan ReplicationWorkerOperation
 			}
 			switch v := oi.(type) {
 			case ReplicateObjectInfo:
+				if opTracker != nil {
+					atomic.AddInt32(opTracker, 1)
+				}
 				globalReplicationStats.incQ(v.Bucket, v.Size, v.DeleteMarker, v.OpType)
 				replicateObject(p.ctx, v, p.objLayer)
 				globalReplicationStats.decQ(v.Bucket, v.Size, v.DeleteMarker, v.OpType)
+				if opTracker != nil {
+					atomic.AddInt32(opTracker, -1)
+				}
 			case DeletedObjectReplicationInfo:
+				if opTracker != nil {
+					atomic.AddInt32(opTracker, 1)
+				}
 				replicateDelete(p.ctx, v, p.objLayer)
+				if opTracker != nil {
+					atomic.AddInt32(opTracker, -1)
+				}
 			default:
 				bugLogIf(p.ctx, fmt.Errorf("unknown replication type: %T", oi), "unknown-replicate-type")
 			}
@@ -2014,6 +2019,30 @@ func (p *ReplicationPool) AddLargeWorker(input <-chan ReplicationWorkerOperation
 	}
 }

+// ResizeLrgWorkers sets replication workers pool for large transfers(>=128MiB) to new size.
+// checkOld can be set to an expected value.
+// If the worker count changed
+func (p *ReplicationPool) ResizeLrgWorkers(n, checkOld int) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if (checkOld > 0 && len(p.lrgworkers) != checkOld) || n == len(p.lrgworkers) || n < 1 {
+		// Either already satisfied or worker count changed while we waited for the lock.
+		return
+	}
+	for len(p.lrgworkers) < n {
+		input := make(chan ReplicationWorkerOperation, 100000)
+		p.lrgworkers = append(p.lrgworkers, input)
+
+		go p.AddLargeWorker(input, &p.activeLrgWorkers)
+	}
+	for len(p.lrgworkers) > n {
+		worker := p.lrgworkers[len(p.lrgworkers)-1]
+		p.lrgworkers = p.lrgworkers[:len(p.lrgworkers)-1]
+		xioutil.SafeClose(worker)
+	}
+}
+
 // ActiveWorkers returns the number of active workers handling replication traffic.
 func (p *ReplicationPool) ActiveWorkers() int {
 	return int(atomic.LoadInt32(&p.activeWorkers))
@@ -2024,6 +2053,11 @@ func (p *ReplicationPool) ActiveMRFWorkers() int {
 	return int(atomic.LoadInt32(&p.activeMRFWorkers))
 }

+// ActiveLrgWorkers returns the number of active workers handling traffic > 128MiB object size.
+func (p *ReplicationPool) ActiveLrgWorkers() int {
+	return int(atomic.LoadInt32(&p.activeLrgWorkers))
+}
+
 // ResizeWorkers sets replication workers pool to new size.
 // checkOld can be set to an expected value.
 // If the worker count changed
@@ -2049,7 +2083,7 @@ func (p *ReplicationPool) ResizeWorkers(n, checkOld int) {
 }

 // ResizeWorkerPriority sets replication failed workers pool size
-func (p *ReplicationPool) ResizeWorkerPriority(pri string, maxWorkers int) {
+func (p *ReplicationPool) ResizeWorkerPriority(pri string, maxWorkers, maxLWorkers int) {
 	var workers, mrfWorkers int
 	p.mu.Lock()
 	switch pri {
@@ -2076,11 +2110,15 @@ func (p *ReplicationPool) ResizeWorkerPriority(pri string, maxWorkers int) {
 	if maxWorkers > 0 && mrfWorkers > maxWorkers {
 		mrfWorkers = maxWorkers
 	}
+	if maxLWorkers <= 0 {
+		maxLWorkers = LargeWorkerCount
+	}
 	p.priority = pri
 	p.maxWorkers = maxWorkers
 	p.mu.Unlock()
 	p.ResizeWorkers(workers, 0)
 	p.ResizeFailedWorkers(mrfWorkers)
+	p.ResizeLrgWorkers(maxLWorkers, 0)
 }

 // ResizeFailedWorkers sets replication failed workers pool size
@@ -2127,6 +2165,15 @@ func (p *ReplicationPool) queueReplicaTask(ri ReplicateObjectInfo) {
 		case p.lrgworkers[h%LargeWorkerCount] <- ri:
 		default:
 			globalReplicationPool.queueMRFSave(ri.ToMRFEntry())
+			p.mu.RLock()
+			maxLWorkers := p.maxLWorkers
+			existing := len(p.lrgworkers)
+			p.mu.RUnlock()
+			maxLWorkers = min(maxLWorkers, LargeWorkerCount)
+			if p.ActiveLrgWorkers() < maxLWorkers {
+				workers := min(existing+1, maxLWorkers)
+				p.ResizeLrgWorkers(workers, existing)
+			}
 		}
 		return
 	}
@@ -2229,8 +2276,9 @@ func (p *ReplicationPool) queueReplicaDeleteTask(doi DeletedObjectReplicationInf
 }

 type replicationPoolOpts struct {
-	Priority   string
-	MaxWorkers int
+	Priority    string
+	MaxWorkers  int
+	MaxLWorkers int
 }

 func initBackgroundReplication(ctx context.Context, objectAPI ObjectLayer) {
@@ -101,7 +101,7 @@ func initHelp() {
 		config.HelpKV{
 			Key:         config.SubnetSubSys,
 			Type:        "string",
-			Description: "register the cluster to MinIO SUBNET",
+			Description: "register Enterprise license for the cluster",
 			Optional:    true,
 		},
 		config.HelpKV{
@@ -2447,6 +2447,7 @@ const (
 type HealthOptions struct {
 	Maintenance    bool
 	DeploymentType string
+	NoLogging      bool
 }

 // HealthResult returns the current state of the system, also
@@ -2483,7 +2484,7 @@ func (hr HealthResult) String() string {
 		if i == 0 {
 			str.WriteString(")")
 		} else {
-			str.WriteString("), ")
+			str.WriteString(") | ")
 		}
 	}
 	return str.String()
@@ -2606,7 +2607,7 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 			})

 			healthy := erasureSetUpCount[poolIdx][setIdx].online >= poolWriteQuorums[poolIdx]
-			if !healthy {
+			if !healthy && !opts.NoLogging {
 				storageLogIf(logger.SetReqInfo(ctx, reqInfo),
 					fmt.Errorf("Write quorum could not be established on pool: %d, set: %d, expected write quorum: %d, drives-online: %d",
 						poolIdx, setIdx, poolWriteQuorums[poolIdx], erasureSetUpCount[poolIdx][setIdx].online), logger.FatalKind)
@@ -2614,7 +2615,7 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 			result.Healthy = result.Healthy && healthy

 			healthyRead := erasureSetUpCount[poolIdx][setIdx].online >= poolReadQuorums[poolIdx]
-			if !healthyRead {
+			if !healthyRead && !opts.NoLogging {
 				storageLogIf(logger.SetReqInfo(ctx, reqInfo),
 					fmt.Errorf("Read quorum could not be established on pool: %d, set: %d, expected read quorum: %d, drives-online: %d",
 						poolIdx, setIdx, poolReadQuorums[poolIdx], erasureSetUpCount[poolIdx][setIdx].online))
@@ -189,7 +189,7 @@ func (s *erasureSets) Legacy() (ok bool) {

 // connectDisks - attempt to connect all the endpoints, loads format
 // and re-arranges the disks in proper position.
-func (s *erasureSets) connectDisks() {
+func (s *erasureSets) connectDisks(log bool) {
 	defer func() {
 		s.lastConnectDisksOpTime = time.Now()
 	}()
@@ -224,7 +224,9 @@ func (s *erasureSets) connectDisks() {
 				if endpoint.IsLocal && errors.Is(err, errUnformattedDisk) {
 					globalBackgroundHealState.pushHealLocalDisks(endpoint)
 				} else if !errors.Is(err, errDriveIsRoot) {
-					printEndpointError(endpoint, err, true)
+					if log {
+						printEndpointError(endpoint, err, true)
+					}
 				}
 				return
 			}
@@ -285,7 +287,7 @@ func (s *erasureSets) monitorAndConnectEndpoints(ctx context.Context, monitorInt
 	time.Sleep(time.Duration(r.Float64() * float64(time.Second)))

 	// Pre-emptively connect the disks if possible.
-	s.connectDisks()
+	s.connectDisks(false)

 	monitor := time.NewTimer(monitorInterval)
 	defer monitor.Stop()
@@ -299,7 +301,7 @@ func (s *erasureSets) monitorAndConnectEndpoints(ctx context.Context, monitorInt
 				console.Debugln("running drive monitoring")
 			}

-			s.connectDisks()
+			s.connectDisks(true)

 			// Reset the timer for next interval
 			monitor.Reset(monitorInterval)
@@ -39,14 +39,15 @@ import (
 type apiConfig struct {
 	mu sync.RWMutex

-	requestsDeadline      time.Duration
-	requestsPool          chan struct{}
-	clusterDeadline       time.Duration
-	listQuorum            string
-	corsAllowOrigins      []string
-	replicationPriority   string
-	replicationMaxWorkers int
-	transitionWorkers     int
+	requestsDeadline       time.Duration
+	requestsPool           chan struct{}
+	clusterDeadline        time.Duration
+	listQuorum             string
+	corsAllowOrigins       []string
+	replicationPriority    string
+	replicationMaxWorkers  int
+	replicationMaxLWorkers int
+	transitionWorkers      int

 	staleUploadsExpiry          time.Duration
 	staleUploadsCleanupInterval time.Duration
@@ -170,11 +171,12 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) {
 	}
 	t.listQuorum = listQuorum
 	if globalReplicationPool != nil &&
-		(cfg.ReplicationPriority != t.replicationPriority || cfg.ReplicationMaxWorkers != t.replicationMaxWorkers) {
-		globalReplicationPool.ResizeWorkerPriority(cfg.ReplicationPriority, cfg.ReplicationMaxWorkers)
+		(cfg.ReplicationPriority != t.replicationPriority || cfg.ReplicationMaxWorkers != t.replicationMaxWorkers || cfg.ReplicationMaxLWorkers != t.replicationMaxLWorkers) {
+		globalReplicationPool.ResizeWorkerPriority(cfg.ReplicationPriority, cfg.ReplicationMaxWorkers, cfg.ReplicationMaxLWorkers)
 	}
 	t.replicationPriority = cfg.ReplicationPriority
 	t.replicationMaxWorkers = cfg.ReplicationMaxWorkers
+	t.replicationMaxLWorkers = cfg.ReplicationMaxLWorkers

 	// N B api.transition_workers will be deprecated
 	if globalTransitionState != nil {
@@ -381,14 +383,16 @@ func (t *apiConfig) getReplicationOpts() replicationPoolOpts {

 	if t.replicationPriority == "" {
 		return replicationPoolOpts{
-			Priority:   "auto",
-			MaxWorkers: WorkerMaxLimit,
+			Priority:    "auto",
+			MaxWorkers:  WorkerMaxLimit,
+			MaxLWorkers: LargeWorkerCount,
 		}
 	}

 	return replicationPoolOpts{
-		Priority:   t.replicationPriority,
-		MaxWorkers: t.replicationMaxWorkers,
+		Priority:    t.replicationPriority,
+		MaxWorkers:  t.replicationMaxWorkers,
+		MaxLWorkers: t.replicationMaxLWorkers,
 	}
 }

@@ -495,7 +495,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 		for _, item := range regUsersList {
 			userName := path.Dir(item)
 			if err := iamOS.loadUser(ctx, userName, regUser, cache.iamUsersMap); err != nil && err != errNoSuchUser {
-				return fmt.Errorf("unable to load the user `%s`: %w", userName, err)
+				return fmt.Errorf("unable to load the user: %w", err)
 			}
 		}
 		if took := time.Since(regUsersLoadStartTime); took > maxIAMLoadOpTime {
@@ -510,7 +510,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 		for _, item := range groupsList {
 			group := path.Dir(item)
 			if err := iamOS.loadGroup(ctx, group, cache.iamGroupsMap); err != nil && err != errNoSuchGroup {
-				return fmt.Errorf("unable to load the group `%s`: %w", group, err)
+				return fmt.Errorf("unable to load the group: %w", err)
 			}
 		}
 		if took := time.Since(groupsLoadStartTime); took > maxIAMLoadOpTime {
@@ -524,7 +524,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 	for _, item := range userPolicyMappingsList {
 		userName := strings.TrimSuffix(item, ".json")
 		if err := iamOS.loadMappedPolicy(ctx, userName, regUser, false, cache.iamUserPolicyMap); err != nil && !errors.Is(err, errNoSuchPolicy) {
-			return fmt.Errorf("unable to load the policy mapping for the user `%s`: %w", userName, err)
+			return fmt.Errorf("unable to load the policy mapping for the user: %w", err)
 		}
 	}
 	if took := time.Since(userPolicyMappingLoadStartTime); took > maxIAMLoadOpTime {
@@ -537,7 +537,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 	for _, item := range groupPolicyMappingsList {
 		groupName := strings.TrimSuffix(item, ".json")
 		if err := iamOS.loadMappedPolicy(ctx, groupName, regUser, true, cache.iamGroupPolicyMap); err != nil && !errors.Is(err, errNoSuchPolicy) {
-			return fmt.Errorf("unable to load the policy mapping for the group `%s`: %w", groupName, err)
+			return fmt.Errorf("unable to load the policy mapping for the group: %w", err)
 		}
 	}
 	if took := time.Since(groupPolicyMappingLoadStartTime); took > maxIAMLoadOpTime {
@@ -551,7 +551,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 	for _, item := range svcAccList {
 		userName := path.Dir(item)
 		if err := iamOS.loadUser(ctx, userName, svcUser, svcUsersMap); err != nil && err != errNoSuchUser {
-			return fmt.Errorf("unable to load the service account `%s`: %w", userName, err)
+			return fmt.Errorf("unable to load the service account: %w", err)
 		}
 	}
 	if took := time.Since(svcAccLoadStartTime); took > maxIAMLoadOpTime {
@@ -580,7 +580,7 @@ func (iamOS *IAMObjectStore) loadAllFromObjStore(ctx context.Context, cache *iam
 			// OIDC/AssumeRoleWithCustomToken/AssumeRoleWithCertificate).
 			err := iamOS.loadMappedPolicy(ctx, svcParent, stsUser, false, cache.iamSTSPolicyMap)
 			if err != nil && !errors.Is(err, errNoSuchPolicy) {
-				return fmt.Errorf("unable to load the policy mapping for the STS user `%s`: %w", svcParent, err)
+				return fmt.Errorf("unable to load the policy mapping for the STS user: %w", err)
 			}
 		}
 	}
@@ -2090,7 +2090,7 @@ func (store *IAMStoreSys) GetAllSTSUserMappings(userPredicate func(string) bool)

 // Assumes store is locked by caller. If userMap is empty, returns all user mappings.
 func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, userMap map[string]set.StringSet,
-	userPredicate func(string) bool,
+	userPredicate func(string) bool, decodeFunc func(string) string,
 ) []madmin.UserPolicyEntities {
 	stsMap := xsync.NewMapOf[string, MappedPolicy]()
 	resMap := make(map[string]madmin.UserPolicyEntities, len(userMap))
@@ -2098,9 +2098,13 @@ func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, userMap map[st
 	for user, groupSet := range userMap {
 		// Attempt to load parent user mapping for STS accounts
 		store.loadMappedPolicy(context.TODO(), user, stsUser, false, stsMap)
-		blankEntities := madmin.UserPolicyEntities{User: user}
+		decodeUser := user
+		if decodeFunc != nil {
+			decodeUser = decodeFunc(user)
+		}
+		blankEntities := madmin.UserPolicyEntities{User: decodeUser}
 		if !groupSet.IsEmpty() {
-			blankEntities.MemberOfMappings = store.listGroupPolicyMappings(cache, groupSet, nil)
+			blankEntities.MemberOfMappings = store.listGroupPolicyMappings(cache, groupSet, nil, decodeFunc)
 		}
 		resMap[user] = blankEntities
 	}
@@ -2116,7 +2120,11 @@ func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, userMap map[st
 			if len(userMap) > 0 {
 				return true
 			}
-			entitiesWithMemberOf = madmin.UserPolicyEntities{User: user}
+			decodeUser := user
+			if decodeFunc != nil {
+				decodeUser = decodeFunc(user)
+			}
+			entitiesWithMemberOf = madmin.UserPolicyEntities{User: decodeUser}
 		}

 		ps := mappedPolicy.toSlice()
@@ -2155,7 +2163,7 @@ func (store *IAMStoreSys) listUserPolicyMappings(cache *iamCache, userMap map[st

 // Assumes store is locked by caller. If groups is empty, returns all group mappings.
 func (store *IAMStoreSys) listGroupPolicyMappings(cache *iamCache, groupsSet set.StringSet,
-	groupPredicate func(string) bool,
+	groupPredicate func(string) bool, decodeFunc func(string) string,
 ) []madmin.GroupPolicyEntities {
 	var r []madmin.GroupPolicyEntities

@@ -2168,10 +2176,15 @@ func (store *IAMStoreSys) listGroupPolicyMappings(cache *iamCache, groupsSet set
 			return true
 		}

+		decodeGroup := group
+		if decodeFunc != nil {
+			decodeGroup = decodeFunc(group)
+		}
+
 		ps := mappedPolicy.toSlice()
 		sort.Strings(ps)
 		r = append(r, madmin.GroupPolicyEntities{
-			Group:    group,
+			Group:    decodeGroup,
 			Policies: ps,
 		})
 		return true
@@ -2186,7 +2199,7 @@ func (store *IAMStoreSys) listGroupPolicyMappings(cache *iamCache, groupsSet set

 // Assumes store is locked by caller. If policies is empty, returns all policy mappings.
 func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.StringSet,
-	userPredicate, groupPredicate func(string) bool,
+	userPredicate, groupPredicate func(string) bool, decodeFunc func(string) string,
 ) []madmin.PolicyEntities {
 	policyToUsersMap := make(map[string]set.StringSet)
 	cache.iamUserPolicyMap.Range(func(user string, mappedPolicy MappedPolicy) bool {
@@ -2194,6 +2207,11 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 			return true
 		}

+		decodeUser := user
+		if decodeFunc != nil {
+			decodeUser = decodeFunc(user)
+		}
+
 		commonPolicySet := mappedPolicy.policySet()
 		if !queryPolSet.IsEmpty() {
 			commonPolicySet = commonPolicySet.Intersection(queryPolSet)
@@ -2201,9 +2219,9 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 		for _, policy := range commonPolicySet.ToSlice() {
 			s, ok := policyToUsersMap[policy]
 			if !ok {
-				policyToUsersMap[policy] = set.CreateStringSet(user)
+				policyToUsersMap[policy] = set.CreateStringSet(decodeUser)
 			} else {
-				s.Add(user)
+				s.Add(decodeUser)
 				policyToUsersMap[policy] = s
 			}
 		}
@@ -2217,6 +2235,11 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 				continue
 			}

+			decodeUser := user
+			if decodeFunc != nil {
+				decodeUser = decodeFunc(user)
+			}
+
 			var mappedPolicy MappedPolicy
 			store.loadIAMConfig(context.Background(), &mappedPolicy, getMappedPolicyPath(user, stsUser, false))

@@ -2227,9 +2250,9 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 			for _, policy := range commonPolicySet.ToSlice() {
 				s, ok := policyToUsersMap[policy]
 				if !ok {
-					policyToUsersMap[policy] = set.CreateStringSet(user)
+					policyToUsersMap[policy] = set.CreateStringSet(decodeUser)
 				} else {
-					s.Add(user)
+					s.Add(decodeUser)
 					policyToUsersMap[policy] = s
 				}
 			}
@@ -2244,6 +2267,11 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 					return true
 				}

+				decodeUser := user
+				if decodeFunc != nil {
+					decodeUser = decodeFunc(user)
+				}
+
 				commonPolicySet := mappedPolicy.policySet()
 				if !queryPolSet.IsEmpty() {
 					commonPolicySet = commonPolicySet.Intersection(queryPolSet)
@@ -2251,9 +2279,9 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 				for _, policy := range commonPolicySet.ToSlice() {
 					s, ok := policyToUsersMap[policy]
 					if !ok {
-						policyToUsersMap[policy] = set.CreateStringSet(user)
+						policyToUsersMap[policy] = set.CreateStringSet(decodeUser)
 					} else {
-						s.Add(user)
+						s.Add(decodeUser)
 						policyToUsersMap[policy] = s
 					}
 				}
@@ -2268,6 +2296,11 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 			return true
 		}

+		decodeGroup := group
+		if decodeFunc != nil {
+			decodeGroup = decodeFunc(group)
+		}
+
 		commonPolicySet := mappedPolicy.policySet()
 		if !queryPolSet.IsEmpty() {
 			commonPolicySet = commonPolicySet.Intersection(queryPolSet)
@@ -2275,9 +2308,9 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St
 		for _, policy := range commonPolicySet.ToSlice() {
 			s, ok := policyToGroupsMap[policy]
 			if !ok {
-				policyToGroupsMap[policy] = set.CreateStringSet(group)
+				policyToGroupsMap[policy] = set.CreateStringSet(decodeGroup)
 			} else {
-				s.Add(group)
+				s.Add(decodeGroup)
 				policyToGroupsMap[policy] = s
 			}
 		}
@@ -2318,7 +2351,7 @@ func (store *IAMStoreSys) listPolicyMappings(cache *iamCache, queryPolSet set.St

 // ListPolicyMappings - return users/groups mapped to policies.
 func (store *IAMStoreSys) ListPolicyMappings(q cleanEntitiesQuery,
-	userPredicate, groupPredicate func(string) bool,
+	userPredicate, groupPredicate func(string) bool, decodeFunc func(string) string,
 ) madmin.PolicyEntitiesResult {
 	cache := store.rlock()
 	defer store.runlock()
@@ -2328,13 +2361,13 @@ func (store *IAMStoreSys) ListPolicyMappings(q cleanEntitiesQuery,
 	isAllPoliciesQuery := len(q.Users) == 0 && len(q.Groups) == 0 && len(q.Policies) == 0

 	if len(q.Users) > 0 {
-		result.UserMappings = store.listUserPolicyMappings(cache, q.Users, userPredicate)
+		result.UserMappings = store.listUserPolicyMappings(cache, q.Users, userPredicate, decodeFunc)
 	}
 	if len(q.Groups) > 0 {
-		result.GroupMappings = store.listGroupPolicyMappings(cache, q.Groups, groupPredicate)
+		result.GroupMappings = store.listGroupPolicyMappings(cache, q.Groups, groupPredicate, decodeFunc)
 	}
 	if len(q.Policies) > 0 || isAllPoliciesQuery {
-		result.PolicyMappings = store.listPolicyMappings(cache, q.Policies, userPredicate, groupPredicate)
+		result.PolicyMappings = store.listPolicyMappings(cache, q.Policies, userPredicate, groupPredicate, decodeFunc)
 	}
 	return result
 }
@@ -217,6 +217,9 @@ func (sys *IAMSys) Load(ctx context.Context, firstTime bool) error {

 	if firstTime {
 		bootstrapTraceMsg(fmt.Sprintf("globalIAMSys.Load(): (duration: %s)", loadDuration))
+		if globalIsDistErasure {
+			logger.Info("IAM load(startup) finished. (duration: %s)", loadDuration)
+		}
 	}

 	select {
@@ -400,12 +403,12 @@ func (sys *IAMSys) periodicRoutines(ctx context.Context, baseInterval time.Durat
 			// Load all IAM items (except STS creds) periodically.
 			refreshStart := time.Now()
 			if err := sys.Load(ctx, false); err != nil {
-				iamLogIf(ctx, fmt.Errorf("Failure in periodic refresh for IAM (took %.2fs): %v", time.Since(refreshStart).Seconds(), err), logger.WarningKind)
+				iamLogIf(ctx, fmt.Errorf("Failure in periodic refresh for IAM (duration: %s): %v", time.Since(refreshStart), err), logger.WarningKind)
 			} else {
 				took := time.Since(refreshStart).Seconds()
 				if took > maxDurationSecondsForLog {
 					// Log if we took a lot of time to load.
-					logger.Info("IAM refresh took %.2fs", took)
+					logger.Info("IAM refresh took (duration: %s)", took)
 				}
 			}

@@ -841,7 +844,7 @@ func (sys *IAMSys) createCleanEntitiesQuery(q madmin.PolicyEntitiesQuery, ldap b
 		// Validate and normalize groups.
 		for _, group := range q.Groups {
 			lookupRes, underDN, _ := sys.LDAPConfig.GetValidatedGroupDN(nil, group)
-			if lookupRes != nil && !underDN {
+			if lookupRes != nil && underDN {
 				cleanQ.Groups.Add(lookupRes.NormDN)
 			}
 		}
@@ -871,7 +874,7 @@ func (sys *IAMSys) QueryLDAPPolicyEntities(ctx context.Context, q madmin.PolicyE
 	select {
 	case <-sys.configLoaded:
 		cleanQuery := sys.createCleanEntitiesQuery(q, true)
-		pe := sys.store.ListPolicyMappings(cleanQuery, sys.LDAPConfig.IsLDAPUserDN, sys.LDAPConfig.IsLDAPGroupDN)
+		pe := sys.store.ListPolicyMappings(cleanQuery, sys.LDAPConfig.IsLDAPUserDN, sys.LDAPConfig.IsLDAPGroupDN, sys.LDAPConfig.DecodeDN)
 		pe.Timestamp = UTCNow()
 		return &pe, nil
 	case <-ctx.Done():
@@ -955,7 +958,7 @@ func (sys *IAMSys) QueryPolicyEntities(ctx context.Context, q madmin.PolicyEntit
 				return !sys.LDAPConfig.IsLDAPGroupDN(s)
 			}
 		}
-		pe := sys.store.ListPolicyMappings(cleanQuery, userPredicate, groupPredicate)
+		pe := sys.store.ListPolicyMappings(cleanQuery, userPredicate, groupPredicate, nil)
 		pe.Timestamp = UTCNow()
 		return &pe, nil
 	case <-ctx.Done():
@@ -1571,11 +1574,11 @@ func (sys *IAMSys) NormalizeLDAPAccessKeypairs(ctx context.Context, accessKeyMap
 		// server and is under a configured base DN.
 		validatedParent, isUnderBaseDN, err := sys.LDAPConfig.GetValidatedUserDN(conn, parent)
 		if err != nil {
-			collectedErrors = append(collectedErrors, fmt.Errorf("could not validate `%s` exists in LDAP directory: %w", parent, err))
+			collectedErrors = append(collectedErrors, fmt.Errorf("could not validate parent exists in LDAP directory: %w", err))
 			continue
 		}
 		if validatedParent == nil || !isUnderBaseDN {
-			err := fmt.Errorf("DN `%s` was not found in the LDAP directory", parent)
+			err := fmt.Errorf("DN parent was not found in the LDAP directory")
 			collectedErrors = append(collectedErrors, err)
 			continue
 		}
@@ -1590,11 +1593,11 @@ func (sys *IAMSys) NormalizeLDAPAccessKeypairs(ctx context.Context, accessKeyMap
 			// configured base DN.
 			validatedGroup, _, err := sys.LDAPConfig.GetValidatedGroupDN(conn, group)
 			if err != nil {
-				collectedErrors = append(collectedErrors, fmt.Errorf("could not validate `%s` exists in LDAP directory: %w", group, err))
+				collectedErrors = append(collectedErrors, fmt.Errorf("could not validate group exists in LDAP directory: %w", err))
 				continue
 			}
 			if validatedGroup == nil {
-				err := fmt.Errorf("DN `%s` was not found in the LDAP directory", group)
+				err := fmt.Errorf("DN group was not found in the LDAP directory")
 				collectedErrors = append(collectedErrors, err)
 				continue
 			}
@@ -1684,7 +1687,7 @@ func (sys *IAMSys) NormalizeLDAPMappingImport(ctx context.Context, isGroup bool,
 			continue
 		}
 		if validatedDN == nil || !underBaseDN {
-			err := fmt.Errorf("DN `%s` was not found in the LDAP directory", k)
+			err := fmt.Errorf("DN was not found in the LDAP directory")
 			collectedErrors = append(collectedErrors, err)
 			continue
 		}
@@ -2027,7 +2030,7 @@ func (sys *IAMSys) PolicyDBUpdateLDAP(ctx context.Context, isAttach bool,
 		if dnResult == nil {
 			// dn not found - still attempt to detach if provided user is a DN.
 			if !isAttach && sys.LDAPConfig.IsLDAPUserDN(r.User) {
-				dn = r.User
+				dn = sys.LDAPConfig.QuickNormalizeDN(r.User)
 			} else {
 				err = errNoSuchUser
 				return
@@ -2044,7 +2047,7 @@ func (sys *IAMSys) PolicyDBUpdateLDAP(ctx context.Context, isAttach bool,
 		}
 		if dnResult == nil || !underBaseDN {
 			if !isAttach {
-				dn = r.Group
+				dn = sys.LDAPConfig.QuickNormalizeDN(r.Group)
 			} else {
 				err = errNoSuchGroup
 				return
@@ -153,19 +153,7 @@ func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (
 			} else {
 				// Continue listing
 				o.ID = c.id
-				go func(meta metacache) {
-					// Continuously update while we wait.
-					t := time.NewTicker(metacacheMaxClientWait / 10)
-					defer t.Stop()
-					select {
-					case <-ctx.Done():
-						// Request is done, stop updating.
-						return
-					case <-t.C:
-						meta.lastHandout = time.Now()
-						meta, _ = rpc.UpdateMetacacheListing(ctx, meta)
-					}
-				}(*c)
+				go c.keepAlive(ctx, rpc)
 			}
 		}
 	}
@@ -219,6 +207,9 @@ func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (
 		o.ID = ""
 	}

+	if contextCanceled(ctx) {
+		return entries, ctx.Err()
+	}
 	// Do listing in-place.
 	// Create output for our results.
 	// Create filter for results.
@@ -449,5 +440,10 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
 		xioutil.SafeClose(saveCh)
 	}()

-	return filteredResults()
+	entries, err = filteredResults()
+	if err == nil {
+		// Check if listing recorded an error.
+		err = meta.getErr()
+	}
+	return entries, err
 }
@@ -805,6 +805,17 @@ func (m *metaCacheRPC) setErr(err string) {
 	*m.meta = meta
 }

+// getErr will return an error if the listing failed.
+// The error is not type safe.
+func (m *metaCacheRPC) getErr() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.meta.status == scanStateError {
+		return errors.New(m.meta.error)
+	}
+	return nil
+}
+
 func (er *erasureObjects) saveMetaCacheStream(ctx context.Context, mc *metaCacheRPC, entries <-chan metaCacheEntry) (err error) {
 	o := mc.o
 	o.debugf(color.Green("saveMetaCacheStream:")+" with options: %#v", o)
@@ -24,6 +24,8 @@ import (
 	"path"
 	"strings"
 	"time"
+
+	"github.com/minio/pkg/v3/console"
 )

 type scanStatus uint8
@@ -97,6 +99,37 @@ func (m *metacache) worthKeeping() bool {
 	return true
 }

+// keepAlive will continuously update lastHandout until ctx is canceled.
+func (m metacache) keepAlive(ctx context.Context, rpc *peerRESTClient) {
+	// we intentionally operate on a copy of m, so we can update without locks.
+	t := time.NewTicker(metacacheMaxClientWait / 10)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			// Request is done, stop updating.
+			return
+		case <-t.C:
+			m.lastHandout = time.Now()
+
+			if m2, err := rpc.UpdateMetacacheListing(ctx, m); err == nil {
+				if m2.status != scanStateStarted {
+					if serverDebugLog {
+						console.Debugln("returning", m.id, "due to scan state", m2.status, time.Now().Format(time.RFC3339))
+					}
+					return
+				}
+				m = m2
+				if serverDebugLog {
+					console.Debugln("refreshed", m.id, time.Now().Format(time.RFC3339))
+				}
+			} else if serverDebugLog {
+				console.Debugln("error refreshing", m.id, time.Now().Format(time.RFC3339))
+			}
+		}
+	}
+}
+
 // baseDirFromPrefix will return the base directory given an object path.
 // For example an object with name prefix/folder/object.ext will return `prefix/folder/`.
 func baseDirFromPrefix(prefix string) string {
@@ -116,13 +149,17 @@ func baseDirFromPrefix(prefix string) string {
 // update cache with new status.
 // The updates are conditional so multiple callers can update with different states.
 func (m *metacache) update(update metacache) {
-	m.lastUpdate = UTCNow()
+	now := UTCNow()
+	m.lastUpdate = now

-	if m.lastHandout.After(m.lastHandout) {
-		m.lastHandout = UTCNow()
+	if update.lastHandout.After(m.lastHandout) {
+		m.lastHandout = update.lastUpdate
+		if m.lastHandout.After(now) {
+			m.lastHandout = now
+		}
 	}
 	if m.status == scanStateStarted && update.status == scanStateSuccess {
-		m.ended = UTCNow()
+		m.ended = now
 	}

 	if m.status == scanStateStarted && update.status != scanStateStarted {
@@ -138,7 +175,7 @@ func (m *metacache) update(update metacache) {
 	if m.error == "" && update.error != "" {
 		m.error = update.error
 		m.status = scanStateError
-		m.ended = UTCNow()
+		m.ended = now
 	}
 	m.fileNotFound = m.fileNotFound || update.fileNotFound
 }
@@ -38,7 +38,8 @@ const (

 // Standard env prometheus auth type
 const (
-	EnvPrometheusAuthType = "MINIO_PROMETHEUS_AUTH_TYPE"
+	EnvPrometheusAuthType    = "MINIO_PROMETHEUS_AUTH_TYPE"
+	EnvPrometheusOpenMetrics = "MINIO_PROMETHEUS_OPEN_METRICS"
 )

 type prometheusAuthType string
@@ -58,14 +59,15 @@ func registerMetricsRouter(router *mux.Router) {
 	if authType == prometheusPublic {
 		auth = NoAuthMiddleware
 	}
+
 	metricsRouter.Handle(prometheusMetricsPathLegacy, auth(metricsHandler()))
 	metricsRouter.Handle(prometheusMetricsV2ClusterPath, auth(metricsServerHandler()))
 	metricsRouter.Handle(prometheusMetricsV2BucketPath, auth(metricsBucketHandler()))
 	metricsRouter.Handle(prometheusMetricsV2NodePath, auth(metricsNodeHandler()))
 	metricsRouter.Handle(prometheusMetricsV2ResourcePath, auth(metricsResourceHandler()))

-	// Metrics v3!
-	metricsV3Server := newMetricsV3Server(authType)
+	// Metrics v3
+	metricsV3Server := newMetricsV3Server(auth)

 	// Register metrics v3 handler. It also accepts an optional query
 	// parameter `?list` - see handler for details.
@@ -139,7 +139,7 @@ var (
 )

 // loadClusterUsageBucketMetrics - `MetricsLoaderFn` to load bucket usage metrics.
-func loadClusterUsageBucketMetrics(ctx context.Context, m MetricValues, c *metricsCache, buckets []string) error {
+func loadClusterUsageBucketMetrics(ctx context.Context, m MetricValues, c *metricsCache) error {
 	dataUsageInfo, err := c.dataUsageInfo.Get()
 	if err != nil {
 		metricsLogIf(ctx, err)
@@ -153,11 +153,7 @@ func loadClusterUsageBucketMetrics(ctx context.Context, m MetricValues, c *metri

 	m.Set(usageSinceLastUpdateSeconds, float64(time.Since(dataUsageInfo.LastUpdate)))

-	for _, bucket := range buckets {
-		usage, ok := dataUsageInfo.BucketsUsage[bucket]
-		if !ok {
-			continue
-		}
+	for bucket, usage := range dataUsageInfo.BucketsUsage {
 		quota, err := globalBucketQuotaSys.Get(ctx, bucket)
 		if err != nil {
 			// Log and continue if we are unable to retrieve metrics for this
@@ -23,9 +23,12 @@ import (
 	"net/http"
 	"slices"
 	"strings"
+	"sync"

+	"github.com/minio/minio/internal/config"
 	"github.com/minio/minio/internal/mcontext"
 	"github.com/minio/mux"
+	"github.com/minio/pkg/v3/env"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
@@ -33,41 +36,39 @@ import (
 type promLogger struct{}

 func (p promLogger) Println(v ...interface{}) {
-	s := make([]string, 0, len(v))
-	for _, val := range v {
-		s = append(s, fmt.Sprintf("%v", val))
-	}
-	err := fmt.Errorf("metrics handler error: %v", strings.Join(s, " "))
-	metricsLogIf(GlobalContext, err)
+	metricsLogIf(GlobalContext, fmt.Errorf("metrics handler error: %v", v))
 }

 type metricsV3Server struct {
 	registry *prometheus.Registry
 	opts     promhttp.HandlerOpts
-	authFn   func(http.Handler) http.Handler
+	auth     func(http.Handler) http.Handler

 	metricsData *metricsV3Collection
 }

-func newMetricsV3Server(authType prometheusAuthType) *metricsV3Server {
+var (
+	globalMetricsV3CollectorPaths []collectorPath
+	globalMetricsV3Once           sync.Once
+)
+
+func newMetricsV3Server(auth func(h http.Handler) http.Handler) *metricsV3Server {
 	registry := prometheus.NewRegistry()
-	authFn := AuthMiddleware
-	if authType == prometheusPublic {
-		authFn = NoAuthMiddleware
-	}
-
 	metricGroups := newMetricGroups(registry)
-
+	globalMetricsV3Once.Do(func() {
+		globalMetricsV3CollectorPaths = metricGroups.collectorPaths
+	})
 	return &metricsV3Server{
 		registry: registry,
 		opts: promhttp.HandlerOpts{
 			ErrorLog:            promLogger{},
-			ErrorHandling:       promhttp.HTTPErrorOnError,
+			ErrorHandling:       promhttp.ContinueOnError,
 			Registry:            registry,
 			MaxRequestsInFlight: 2,
+			EnableOpenMetrics:   env.Get(EnvPrometheusOpenMetrics, config.EnableOff) == config.EnableOn,
+			ProcessStartTime:    globalBootTime,
 		},
-		authFn: authFn,
-
+		auth:        auth,
 		metricsData: metricGroups,
 	}
 }
@@ -163,7 +164,7 @@ func (h *metricsV3Server) handle(path string, isListingRequest bool, buckets []s
 		http.Error(w, "Metrics Resource Not found", http.StatusNotFound)
 	})

-	// Require that metrics path has at least component.
+	// Require that metrics path has one component at least.
 	if path == "/" {
 		return notFoundHandler
 	}
@@ -221,7 +222,7 @@ func (h *metricsV3Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	pathComponents := mux.Vars(r)["pathComps"]
 	isListingRequest := r.Form.Has("list")

-	buckets := []string{}
+	var buckets []string
 	if strings.HasPrefix(pathComponents, "/bucket/") {
 		// bucket specific metrics, extract the bucket name from the path.
 		// it's the last part of the path. e.g. /bucket/api/<bucket-name>
@@ -246,5 +247,5 @@ func (h *metricsV3Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	})

 	// Add authentication
-	h.authFn(tracedHandler).ServeHTTP(w, r)
+	h.auth(tracedHandler).ServeHTTP(w, r)
 }
@@ -35,8 +35,8 @@ type collectorPath string
 // converted to snake-case (by replaced '/' and '-' with '_') and prefixed with
 // `minio_`.
 func (cp collectorPath) metricPrefix() string {
-	s := strings.TrimPrefix(string(cp), "/")
-	s = strings.ReplaceAll(s, "/", "_")
+	s := strings.TrimPrefix(string(cp), SlashSeparator)
+	s = strings.ReplaceAll(s, SlashSeparator, "_")
 	s = strings.ReplaceAll(s, "-", "_")
 	return "minio_" + s
 }
@@ -56,8 +56,8 @@ func (cp collectorPath) isDescendantOf(arg string) bool {
 	if len(arg) >= len(descendant) {
 		return false
 	}
-	if !strings.HasSuffix(arg, "/") {
-		arg += "/"
+	if !strings.HasSuffix(arg, SlashSeparator) {
+		arg += SlashSeparator
 	}
 	return strings.HasPrefix(descendant, arg)
 }
@@ -72,10 +72,11 @@ const (
 	GaugeMT
 	// HistogramMT - represents a histogram metric.
 	HistogramMT
-	// rangeL - represents a range label.
-	rangeL = "range"
 )

+// rangeL - represents a range label.
+const rangeL = "range"
+
 func (mt MetricType) String() string {
 	switch mt {
 	case CounterMT:
@@ -270,7 +270,7 @@ func newMetricGroups(r *prometheus.Registry) *metricsV3Collection {
 		loadClusterUsageObjectMetrics,
 	)

-	clusterUsageBucketsMG := NewBucketMetricsGroup(clusterUsageBucketsCollectorPath,
+	clusterUsageBucketsMG := NewMetricsGroup(clusterUsageBucketsCollectorPath,
 		[]MetricDescriptor{
 			usageSinceLastUpdateSecondsMD,
 			usageBucketTotalBytesMD,
@@ -166,13 +166,13 @@ func connectLoadInitFormats(verboseLogging bool, firstDisk bool, storageDisks []
 		if err != nil && !errors.Is(err, errXLBackend) && !errors.Is(err, errUnformattedDisk) {
 			if errors.Is(err, errDiskNotFound) && verboseLogging {
 				if globalEndpoints.NEndpoints() > 1 {
-					logger.Error("Unable to connect to %s: %v", endpoints[i], isServerResolvable(endpoints[i], time.Second))
+					logger.Info("Unable to connect to %s: %v, will be retried", endpoints[i], isServerResolvable(endpoints[i], time.Second))
 				} else {
 					logger.Fatal(err, "Unable to connect to %s: %v", endpoints[i], isServerResolvable(endpoints[i], time.Second))
 				}
 			} else {
 				if globalEndpoints.NEndpoints() > 1 {
-					logger.Error("Unable to use the drive %s: %v", endpoints[i], err)
+					logger.Info("Unable to use the drive %s: %v, will be retried", endpoints[i], err)
 				} else {
 					logger.Fatal(errInvalidArgument, "Unable to use the drive %s: %v", endpoints[i], err)
 				}
@@ -841,13 +841,14 @@ func serverMain(ctx *cli.Context) {

 	// Verify kernel release and version.
 	if oldLinux() {
-		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"))
+		warnings = append(warnings, color.YellowBold("Detected Linux kernel version older than 4.0 release, there are some known potential performance problems with this kernel version. MinIO recommends a minimum of 4.x linux kernel version for best performance"))
 	}

 	maxProcs := runtime.GOMAXPROCS(0)
 	cpuProcs := runtime.NumCPU()
 	if 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))
+		warnings = append(warnings, color.YellowBold("Detected GOMAXPROCS(%d) < NumCPU(%d), please make sure to provide all PROCS to MinIO for optimal performance",
+			maxProcs, cpuProcs))
 	}

 	// Initialize grid
@@ -921,16 +922,18 @@ func serverMain(ctx *cli.Context) {
 	}

 	bootstrapTrace("waitForQuorum", func() {
-		result := newObject.Health(context.Background(), HealthOptions{})
+		result := newObject.Health(context.Background(), HealthOptions{NoLogging: true})
 		for !result.HealthyRead {
 			if debugNoExit {
-				logger.Info("Not waiting for quorum since we are debugging.. possible cause unhealthy sets (%s)", result)
+				logger.Info("Not waiting for quorum since we are debugging.. possible cause unhealthy sets")
+				logger.Info(result.String())
 				break
 			}
 			d := time.Duration(r.Float64() * float64(time.Second))
-			logger.Info("Waiting for quorum READ healthcheck to succeed.. possible cause unhealthy sets (%s), retrying in %s", result, d)
+			logger.Info("Waiting for quorum READ healthcheck to succeed retrying in %s.. possible cause unhealthy sets", d)
+			logger.Info(result.String())
 			time.Sleep(d)
-			result = newObject.Health(context.Background(), HealthOptions{})
+			result = newObject.Health(context.Background(), HealthOptions{NoLogging: true})
 		}
 	})

@@ -953,11 +956,11 @@ func serverMain(ctx *cli.Context) {
 		}

 		if !globalServerCtxt.StrictS3Compat {
-			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"))
+			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("- 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)
 		warnings = append(warnings, color.YellowBold(msg))
 	}
@@ -1103,18 +1106,12 @@ func serverMain(ctx *cli.Context) {
 		printStartupMessage(getAPIEndpoints(), err)

 		// Print a warning at the end of the startup banner so it is more noticeable
-		if newObject.BackendInfo().StandardSCParity == 0 {
-			warnings = append(warnings, color.YellowBold("- The standard parity is set to 0. This can lead to data loss."))
+		if newObject.BackendInfo().StandardSCParity == 0 && !globalIsErasureSD {
+			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)
-			}
+
+		for _, warn := range warnings {
+			logger.Warning(warn)
 		}
 	}()

@@ -23,7 +23,6 @@ import (
 	"net/url"
 	"strings"

-	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/color"
 	"github.com/minio/minio/internal/logger"
 	xnet "github.com/minio/pkg/v3/net"
@@ -37,7 +36,11 @@ func getFormatStr(strLen int, padding int) string {

 // Prints the formatted startup message.
 func printStartupMessage(apiEndpoints []string, err error) {
-	logger.Info(color.Bold(MinioBannerName))
+	banner := strings.Repeat("-", len(MinioBannerName))
+	if globalIsDistErasure {
+		logger.Startup(color.Bold(banner))
+	}
+	logger.Startup(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))
@@ -47,7 +50,7 @@ func printStartupMessage(apiEndpoints []string, err error) {
 	if !globalSubnetConfig.Registered() {
 		var builder strings.Builder
 		startupBanner(&builder)
-		logger.Info(builder.String())
+		logger.Startup(builder.String())
 	}

 	strippedAPIEndpoints := stripStandardPorts(apiEndpoints, globalMinioHost)
@@ -61,6 +64,9 @@ func printStartupMessage(apiEndpoints []string, err error) {

 	// Prints documentation message.
 	printObjectAPIMsg()
+	if globalIsDistErasure {
+		logger.Startup(color.Bold(banner))
+	}
 }

 // Returns true if input is IPv6
@@ -113,21 +119,21 @@ func printServerCommonMsg(apiEndpoints []string) {

 	apiEndpointStr := strings.TrimSpace(strings.Join(apiEndpoints, "  "))
 	// Colorize the message and print.
-	logger.Info(color.Blue("API: ") + color.Bold(fmt.Sprintf("%s ", apiEndpointStr)))
+	logger.Startup(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.Startup(color.Blue("   RootUser: ") + color.Bold("%s ", cred.AccessKey))
+		logger.Startup(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.Startup(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("WebUI: ") + color.Bold(fmt.Sprintf("%s ", consoleEndpointStr)))
+		logger.Startup(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.Startup(color.Blue("   RootUser: ") + color.Bold("%s ", cred.AccessKey))
+			logger.Startup(color.Blue("   RootPass: ") + color.Bold("%s ", cred.SecretKey))
 		}
 	}

@@ -137,7 +143,7 @@ func printServerCommonMsg(apiEndpoints []string) {

 // Prints startup message for Object API access, prints link to our SDK documentation.
 func printObjectAPIMsg() {
-	logger.Info(color.Blue("\nDocs: ") + "https://min.io/docs/minio/linux/index.html")
+	logger.Startup(color.Blue("\nDocs: ") + "https://min.io/docs/minio/linux/index.html")
 }

 func printLambdaTargets() {
@@ -149,7 +155,7 @@ func printLambdaTargets() {
 	for _, arn := range globalLambdaTargetList.List(globalSite.Region()) {
 		arnMsg += color.Bold(fmt.Sprintf("%s ", arn))
 	}
-	logger.Info(arnMsg + "\n")
+	logger.Startup(arnMsg + "\n")
 }

 // Prints bucket notification configurations.
@@ -168,7 +174,7 @@ func printEventNotifiers() {
 		arnMsg += color.Bold(fmt.Sprintf("%s ", arn))
 	}

-	logger.Info(arnMsg + "\n")
+	logger.Startup(arnMsg + "\n")
 }

 // Prints startup message for command line access. Prints link to our documentation
@@ -181,35 +187,9 @@ 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("\nCLI: ") + mcQuickStartGuide)
+		logger.Startup(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))
-	}
-}
-
-// Get formatted disk/storage info message.
-func getStorageInfoMsg(storageInfo StorageInfo) string {
-	var msg string
-	var mcMessage string
-	onlineDisks, offlineDisks := getOnlineOfflineDisksStats(storageInfo.Disks)
-	if storageInfo.Backend.Type == madmin.Erasure {
-		if offlineDisks.Sum() > 0 {
-			mcMessage = "Use `mc admin info` to look for latest server/drive info\n"
-		}
-
-		diskInfo := fmt.Sprintf(" %d Online, %d Offline. ", onlineDisks.Sum(), offlineDisks.Sum())
-		msg += color.Blue("Status:") + fmt.Sprintf(getFormatStr(len(diskInfo), 8), diskInfo)
-		if len(mcMessage) > 0 {
-			msg = fmt.Sprintf("%s %s", mcMessage, msg)
-		}
-	}
-	return msg
-}
-
-// Prints startup message of storage capacity and erasure information.
-func printStorageInfo(storageInfo StorageInfo) {
-	if msg := getStorageInfoMsg(storageInfo); msg != "" {
-		logger.Info(msg)
+		logger.Startup(fmt.Sprintf(getFormatStr(len(mcMessage), 3), mcMessage))
 	}
 }
@@ -21,32 +21,9 @@ import (
 	"context"
 	"os"
 	"reflect"
-	"strings"
 	"testing"
-
-	"github.com/minio/madmin-go/v3"
 )

-// Tests if we generate storage info.
-func TestStorageInfoMsg(t *testing.T) {
-	infoStorage := StorageInfo{}
-	infoStorage.Disks = []madmin.Disk{
-		{Endpoint: "http://127.0.0.1:9000/data/1/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9000/data/2/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9000/data/3/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9000/data/4/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9001/data/1/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9001/data/2/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9001/data/3/", State: madmin.DriveStateOk},
-		{Endpoint: "http://127.0.0.1:9001/data/4/", State: madmin.DriveStateOffline},
-	}
-	infoStorage.Backend.Type = madmin.Erasure
-
-	if msg := getStorageInfoMsg(infoStorage); !strings.Contains(msg, "7 Online, 1 Offline") {
-		t.Fatal("Unexpected storage info message, found:", msg)
-	}
-}
-
 // Tests stripping standard ports from apiEndpoints.
 func TestStripStandardPorts(t *testing.T) {
 	apiEndpoints := []string{"http://127.0.0.1:9000", "http://127.0.0.2:80", "https://127.0.0.3:443"}
@@ -35,6 +35,7 @@ import (
 	"time"

 	"github.com/dustin/go-humanize"
+	jwtgo "github.com/golang-jwt/jwt/v4"
 	"github.com/minio/minio-go/v7/pkg/set"
 	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/pkg/v3/policy"
@@ -122,6 +123,9 @@ func runAllTests(suite *TestSuiteCommon, c *check) {
 	suite.TestObjectMultipartListError(c)
 	suite.TestObjectValidMD5(c)
 	suite.TestObjectMultipart(c)
+	suite.TestMetricsV3Handler(c)
+	suite.TestBucketSQSNotificationWebHook(c)
+	suite.TestBucketSQSNotificationAMQP(c)
 	suite.TearDownSuite(c)
 }

@@ -189,6 +193,36 @@ func (s *TestSuiteCommon) TearDownSuite(c *check) {
 	s.testServer.Stop()
 }

+const (
+	defaultPrometheusJWTExpiry = 100 * 365 * 24 * time.Hour
+)
+
+func (s *TestSuiteCommon) TestMetricsV3Handler(c *check) {
+	jwt := jwtgo.NewWithClaims(jwtgo.SigningMethodHS512, jwtgo.StandardClaims{
+		ExpiresAt: time.Now().UTC().Add(defaultPrometheusJWTExpiry).Unix(),
+		Subject:   s.accessKey,
+		Issuer:    "prometheus",
+	})
+
+	token, err := jwt.SignedString([]byte(s.secretKey))
+	c.Assert(err, nil)
+
+	for _, cpath := range globalMetricsV3CollectorPaths {
+		request, err := newTestSignedRequest(http.MethodGet, s.endPoint+minioReservedBucketPath+metricsV3Path+string(cpath),
+			0, nil, s.accessKey, s.secretKey, s.signer)
+		c.Assert(err, nil)
+
+		request.Header.Set("Authorization", "Bearer "+token)
+
+		// execute the request.
+		response, err := s.client.Do(request)
+		c.Assert(err, nil)
+
+		// assert the http response status code.
+		c.Assert(response.StatusCode, http.StatusOK)
+	}
+}
+
 func (s *TestSuiteCommon) TestBucketSQSNotificationWebHook(c *check) {
 	// Sample bucket notification.
 	bucketNotificationBuf := `<NotificationConfiguration><QueueConfiguration><Event>s3:ObjectCreated:Put</Event><Filter><S3Key><FilterRule><Name>prefix</Name><Value>images/</Value></FilterRule></S3Key></Filter><Id>1</Id><Queue>arn:minio:sqs:us-east-1:444455556666:webhook</Queue></QueueConfiguration></NotificationConfiguration>`
@@ -5390,7 +5390,7 @@ func (c *SiteReplicationSys) healUserPolicies(ctx context.Context, objAPI Object
 			UpdatedAt: lastUpdate,
 		})
 		if err != nil {
-			replLogOnceIf(ctx, fmt.Errorf("Unable to heal IAM user policy mapping for %s from peer site %s -> site %s : %w", user, latestPeerName, peerName, err),
+			replLogOnceIf(ctx, fmt.Errorf("Unable to heal IAM user policy mapping from peer site %s -> site %s : %w", latestPeerName, peerName, err),
 				fmt.Sprintf("heal-user-policy-%s", user))
 		}
 	}
@@ -5455,7 +5455,7 @@ func (c *SiteReplicationSys) healGroupPolicies(ctx context.Context, objAPI Objec
 		})
 		if err != nil {
 			replLogOnceIf(ctx,
-				fmt.Errorf("Unable to heal IAM group policy mapping for %s from peer site %s -> site %s : %w", group, latestPeerName, peerName, err),
+				fmt.Errorf("Unable to heal IAM group policy mapping for from peer site %s -> site %s : %w", latestPeerName, peerName, err),
 				fmt.Sprintf("heal-group-policy-%s", group))
 		}
 	}
@@ -5518,7 +5518,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 			claims, err := globalIAMSys.GetClaimsForSvcAcc(ctx, creds.AccessKey)
 			if err != nil {
 				replLogOnceIf(ctx,
-					fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Errorf("Unable to heal service account from peer site %s -> %s : %w", latestPeerName, peerName, err),
 					fmt.Sprintf("heal-user-%s", user))
 				continue
 			}
@@ -5526,7 +5526,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 			_, policy, err := globalIAMSys.GetServiceAccount(ctx, creds.AccessKey)
 			if err != nil {
 				replLogOnceIf(ctx,
-					fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Errorf("Unable to heal service account from peer site %s -> %s : %w", latestPeerName, peerName, err),
 					fmt.Sprintf("heal-user-%s", user))
 				continue
 			}
@@ -5536,7 +5536,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				policyJSON, err = json.Marshal(policy)
 				if err != nil {
 					replLogOnceIf(ctx,
-						fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+						fmt.Errorf("Unable to heal service account from peer site %s -> %s : %w", latestPeerName, peerName, err),
 						fmt.Sprintf("heal-user-%s", user))
 					continue
 				}
@@ -5561,7 +5561,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				UpdatedAt: lastUpdate,
 			}); err != nil {
 				replLogOnceIf(ctx,
-					fmt.Errorf("Unable to heal service account %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Errorf("Unable to heal service account from peer site %s -> %s : %w", latestPeerName, peerName, err),
 					fmt.Sprintf("heal-user-%s", user))
 			}
 			continue
@@ -5576,7 +5576,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				// be applied.
 				if !errors.Is(err, errNoSuchUser) {
 					replLogOnceIf(ctx,
-						fmt.Errorf("Unable to heal temporary credentials %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+						fmt.Errorf("Unable to heal temporary credentials from peer site %s -> %s : %w", latestPeerName, peerName, err),
 						fmt.Sprintf("heal-user-%s", user))
 					continue
 				}
@@ -5596,7 +5596,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 				UpdatedAt: lastUpdate,
 			}); err != nil {
 				replLogOnceIf(ctx,
-					fmt.Errorf("Unable to heal temporary credentials %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+					fmt.Errorf("Unable to heal temporary credentials from peer site %s -> %s : %w", latestPeerName, peerName, err),
 					fmt.Sprintf("heal-user-%s", user))
 			}
 			continue
@@ -5614,7 +5614,7 @@ func (c *SiteReplicationSys) healUsers(ctx context.Context, objAPI ObjectLayer,
 			UpdatedAt: lastUpdate,
 		}); err != nil {
 			replLogOnceIf(ctx,
-				fmt.Errorf("Unable to heal user %s from peer site %s -> %s : %w", user, latestPeerName, peerName, err),
+				fmt.Errorf("Unable to heal user from peer site %s -> %s : %w", latestPeerName, peerName, err),
 				fmt.Sprintf("heal-user-%s", user))
 		}
 	}
@@ -5680,7 +5680,7 @@ func (c *SiteReplicationSys) healGroups(ctx context.Context, objAPI ObjectLayer,
 			UpdatedAt: lastUpdate,
 		}); err != nil {
 			replLogOnceIf(ctx,
-				fmt.Errorf("Unable to heal group %s from peer site %s -> site %s : %w", group, latestPeerName, peerName, err),
+				fmt.Errorf("Unable to heal group from peer site %s -> site %s : %w", latestPeerName, peerName, err),
 				fmt.Sprintf("heal-group-%s", group))
 		}
 	}
@@ -100,7 +100,7 @@ func TestMain(m *testing.M) {
 	// Disable printing console messages during tests.
 	color.Output = io.Discard
 	// Disable Error logging in testing.
-	logger.DisableErrorLog = true
+	logger.DisableLog = true

 	// Uncomment the following line to see trace logs during unit tests.
 	// logger.AddTarget(console.New())
@@ -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-07-04T14-25-45Z
+  image: quay.io/minio/minio:RELEASE.2024-07-13T01-46-15Z
  command: server --console-address ":9001" http://minio{1...4}/data{1...2}
  expose:
    - "9000"
@@ -31,8 +31,6 @@ MINIO_IDENTITY_OPENID_CLAIM_USERINFO        (on|off)    Enable fetching claims f
 MINIO_IDENTITY_OPENID_KEYCLOAK_REALM        (string)    Specify Keycloak 'realm' name, only honored if vendor was set to 'keycloak' as value, if no realm is specified 'master' is default
 MINIO_IDENTITY_OPENID_KEYCLOAK_ADMIN_URL    (string)    Specify Keycloak 'admin' REST API endpoint e.g. http://localhost:8080/auth/admin/
 MINIO_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC  (on|off)    Enable 'Host' header based dynamic redirect URI (default: 'off')
-MINIO_IDENTITY_OPENID_CLAIM_PREFIX          (string)    [DEPRECATED use 'claim_name'] JWT claim namespace prefix e.g. "customer1/"
-MINIO_IDENTITY_OPENID_REDIRECT_URI          (string)    [DEPRECATED use env 'MINIO_BROWSER_REDIRECT_URL'] Configure custom redirect_uri for OpenID login flow callback
 MINIO_IDENTITY_OPENID_COMMENT               (sentence)  optionally add a comment to this setting
 ```

@@ -3,8 +3,10 @@ summary=Maximum server performance for MinIO

 [vm]
 transparent_hugepage=madvise
-transparent_hugepage.defrag=defer+madvise
-transparent_hugepage.khugepaged.max_ptes_none=0
+
+[sysfs]
+/sys/kernel/mm/transparent_hugepage/defrag=defer+madvise
+/sys/kernel/mm/transparent_hugepage/khugepaged/max_ptes_none=0

 [cpu]
 force_latency=1
@@ -1,6 +1,6 @@
 module github.com/minio/minio

-go 1.21
+go 1.22

 require (
 	cloud.google.com/go/storage v1.42.0
@@ -40,22 +40,22 @@ require (
 	github.com/klauspost/filepathx v1.1.1
 	github.com/klauspost/pgzip v1.2.6
 	github.com/klauspost/readahead v1.4.0
-	github.com/klauspost/reedsolomon v1.12.1
+	github.com/klauspost/reedsolomon v1.12.2
 	github.com/lib/pq v1.10.9
 	github.com/lithammer/shortuuid/v4 v4.0.0
 	github.com/miekg/dns v1.1.61
 	github.com/minio/cli v1.24.2
-	github.com/minio/console v1.6.1
+	github.com/minio/console v1.6.3
 	github.com/minio/csvparser v1.0.0
 	github.com/minio/dnscache v0.1.1
 	github.com/minio/dperf v0.5.3
-	github.com/minio/highwayhash v1.0.2
+	github.com/minio/highwayhash v1.0.3
 	github.com/minio/kms-go/kes v0.3.0
 	github.com/minio/kms-go/kms v0.4.0
 	github.com/minio/madmin-go/v3 v3.0.58
 	github.com/minio/minio-go/v7 v7.0.73
 	github.com/minio/mux v1.9.0
-	github.com/minio/pkg/v3 v3.0.3
+	github.com/minio/pkg/v3 v3.0.7
 	github.com/minio/selfupdate v0.6.0
 	github.com/minio/simdjson-go v0.4.5
 	github.com/minio/sio v0.4.0
@@ -218,7 +218,7 @@ require (
 	github.com/muesli/termenv v0.15.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nats-io/jwt/v2 v2.5.0 // indirect
-	github.com/nats-io/nats-streaming-server v0.24.3 // indirect
+	github.com/nats-io/nats-streaming-server v0.24.6 // indirect
 	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/oklog/ulid v1.3.1 // indirect
@@ -325,8 +325,8 @@ github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uG
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/hashicorp/raft v1.3.6 h1:v5xW5KzByoerQlN/o31VJrFNiozgzGyDoMgDJgXpsto=
-github.com/hashicorp/raft v1.3.6/go.mod h1:4Ak7FSPnuvmb0GV6vgIAJ4vYT4bek9bb6Q+7HVbyzqM=
+github.com/hashicorp/raft v1.3.9 h1:9yuo1aR0bFTr1cw7pj3S2Bk6MhJCsnr2NAxvIBrP2x4=
+github.com/hashicorp/raft v1.3.9/go.mod h1:4Ak7FSPnuvmb0GV6vgIAJ4vYT4bek9bb6Q+7HVbyzqM=
 github.com/ianlancetaylor/demangle v0.0.0-20230524184225-eabc099b10ab/go.mod h1:gx7rwoVhcfuVKG5uya9Hs3Sxj7EIvldVofAWIUtGouw=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
@@ -373,8 +373,8 @@ github.com/klauspost/pgzip v1.2.6 h1:8RXeL5crjEUFnR2/Sn6GJNWtSQ3Dk8pq4CL3jvdDyjU
 github.com/klauspost/pgzip v1.2.6/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
 github.com/klauspost/readahead v1.4.0 h1:w4hQ3BpdLjBnRQkZyNi+nwdHU7eGP9buTexWK9lU7gY=
 github.com/klauspost/readahead v1.4.0/go.mod h1:7bolpMKhT5LKskLwYXGSDOyA2TYtMFgdgV0Y8gy7QhA=
-github.com/klauspost/reedsolomon v1.12.1 h1:NhWgum1efX1x58daOBGCFWcxtEhOhXKKl1HAPQUp03Q=
-github.com/klauspost/reedsolomon v1.12.1/go.mod h1:nEi5Kjb6QqtbofI6s+cbG/j1da11c96IBYBSnVGtuBs=
+github.com/klauspost/reedsolomon v1.12.2 h1:TC0hlL/tTRxiMNnqHCzKsY11E0fIIKGCoZ2vQoPKIEM=
+github.com/klauspost/reedsolomon v1.12.2/go.mod h1:nEi5Kjb6QqtbofI6s+cbG/j1da11c96IBYBSnVGtuBs=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8=
 github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
@@ -442,8 +442,8 @@ github.com/minio/cli v1.24.2 h1:J+fCUh9mhPLjN3Lj/YhklXvxj8mnyE/D6FpFduXJ2jg=
 github.com/minio/cli v1.24.2/go.mod h1:bYxnK0uS629N3Bq+AOZZ+6lwF77Sodk4+UL9vNuXhOY=
 github.com/minio/colorjson v1.0.8 h1:AS6gEQ1dTRYHmC4xuoodPDRILHP/9Wz5wYUGDQfPLpg=
 github.com/minio/colorjson v1.0.8/go.mod h1:wrs39G/4kqNlGjwqHvPlAnXuc2tlPszo6JKdSBCLN8w=
-github.com/minio/console v1.6.1 h1:/rlXITBdZeDcX33PCjhEdA2vufeMMFMufj1XVukbu+0=
-github.com/minio/console v1.6.1/go.mod h1:XJ3HKHmigs1MgjaNjUwpyuOAJnwqlSMB+QnZCZ+BROY=
+github.com/minio/console v1.6.3 h1:XGI/Oyq3J2vs+a1cobE87m4L059jr3q1Scej7hrEcbM=
+github.com/minio/console v1.6.3/go.mod h1:yFhhM3Y3uT4N1WtphcYr3QAd7WYLU8CEuTcIiDpksWs=
 github.com/minio/csvparser v1.0.0 h1:xJEHcYK8ZAjeW4hNV9Zu30u+/2o4UyPnYgyjWp8b7ZU=
 github.com/minio/csvparser v1.0.0/go.mod h1:lKXskSLzPgC5WQyzP7maKH7Sl1cqvANXo9YCto8zbtM=
 github.com/minio/dnscache v0.1.1 h1:AMYLqomzskpORiUA1ciN9k7bZT1oB3YZN4cEIi88W5o=
@@ -452,8 +452,9 @@ github.com/minio/dperf v0.5.3 h1:D58ZrMfxrRw83EvAhr4FggvRT0DwWXsWrvsM8Xne+EM=
 github.com/minio/dperf v0.5.3/go.mod h1:WrI7asRe/kv5zmnZ4XwHY74PV8OyUN+efeKINRgk5UI=
 github.com/minio/filepath v1.0.0 h1:fvkJu1+6X+ECRA6G3+JJETj4QeAYO9sV43I79H8ubDY=
 github.com/minio/filepath v1.0.0/go.mod h1:/nRZA2ldl5z6jT9/KQuvZcQlxZIMQoFFQPvEXx9T/Bw=
-github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
+github.com/minio/highwayhash v1.0.3 h1:kbnuUMoHYyVl7szWjSxJnxw11k2U709jqFPPmIUyD6Q=
+github.com/minio/highwayhash v1.0.3/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
 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/kms-go/kms v0.4.0 h1:cLPZceEp+05xHotVBaeFJrgL7JcXM4lBy6PU0idkE7I=
@@ -471,8 +472,8 @@ github.com/minio/mux v1.9.0 h1:dWafQFyEfGhJvK6AwLOt83bIG5bxKxKJnKMCi0XAaoA=
 github.com/minio/mux v1.9.0/go.mod h1:1pAare17ZRL5GpmNL+9YmqHoWnLmMZF9C/ioUCfy0BQ=
 github.com/minio/pkg/v2 v2.0.19 h1:r187/k/oVH9H0DDwvLY5WipkJaZ4CLd4KI3KgIUExR0=
 github.com/minio/pkg/v2 v2.0.19/go.mod h1:luK9LAhQlAPzSuF6F326XSCKjMc1G3Tbh+a9JYwqh8M=
-github.com/minio/pkg/v3 v3.0.3 h1:PUJVi5a6Hdn5mIhffC24koFMQwucvTyBHsIOjsisI+U=
-github.com/minio/pkg/v3 v3.0.3/go.mod h1:53gkSUVHcfYoskOs5YAJ3D99nsd2SKru90rdE9whlXU=
+github.com/minio/pkg/v3 v3.0.7 h1:1I2CbFKO+brioB6Pbnw0jLlFxo+YPy6hCTTXTSitgI8=
+github.com/minio/pkg/v3 v3.0.7/go.mod h1:njlf539caYrgXqn/CXewqvkqBIMDTQo9oBBEL34LzY0=
 github.com/minio/selfupdate v0.6.0 h1:i76PgT0K5xO9+hjzKcacQtO7+MjJ4JKA8Ak8XQ9DDwU=
 github.com/minio/selfupdate v0.6.0/go.mod h1:bO02GTIPCMQFTEvE5h4DjYB58bCoZ35XLeBf0buTDdM=
 github.com/minio/sha256-simd v0.1.1/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM=
@@ -511,16 +512,17 @@ github.com/muesli/termenv v0.15.2/go.mod h1:Epx+iuz8sNs7mNKhxzH4fWXGNpZwUaJKRS1n
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
-github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
+github.com/nats-io/jwt/v2 v2.2.1-0.20220330180145-442af02fd36a/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
 github.com/nats-io/jwt/v2 v2.5.0 h1:WQQ40AAlqqfx+f6ku+i0pOVm+ASirD4fUh+oQsiE9Ak=
 github.com/nats-io/jwt/v2 v2.5.0/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
-github.com/nats-io/nats-server/v2 v2.7.4/go.mod h1:1vZ2Nijh8tcyNe8BDVyTviCd9NYzRbubQYiEHsvOQWc=
+github.com/nats-io/nats-server/v2 v2.8.2/go.mod h1:vIdpKz3OG+DCg4q/xVPdXHoztEyKDWRtykQ4N7hd7C4=
 github.com/nats-io/nats-server/v2 v2.9.23 h1:6Wj6H6QpP9FMlpCyWUaNu2yeZ/qGj+mdRkZ1wbikExU=
 github.com/nats-io/nats-server/v2 v2.9.23/go.mod h1:wEjrEy9vnqIGE4Pqz4/c75v9Pmaq7My2IgFmnykc4C0=
-github.com/nats-io/nats-streaming-server v0.24.3 h1:uZez8jBkXscua++jaDsK7DhpSAkizdetar6yWbPMRco=
-github.com/nats-io/nats-streaming-server v0.24.3/go.mod h1:rqWfyCbxlhKj//fAp8POdQzeADwqkVhZcoWlbhkuU5w=
+github.com/nats-io/nats-streaming-server v0.24.6 h1:iIZXuPSznnYkiy0P3L0AP9zEN9Etp+tITbbX1KKeq4Q=
+github.com/nats-io/nats-streaming-server v0.24.6/go.mod h1:tdKXltY3XLeBJ21sHiZiaPl+j8sK3vcCKBWVyxeQs10=
 github.com/nats-io/nats.go v1.13.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
-github.com/nats-io/nats.go v1.13.1-0.20220308171302-2f2f6968e98d/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.14.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.15.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nats.go v1.22.1/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA=
 github.com/nats-io/nats.go v1.36.0 h1:suEUPuWzTSse/XhESwqLxXGuj8vGRuPRoG7MoRN/qyU=
 github.com/nats-io/nats.go v1.36.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
@@ -737,8 +739,7 @@ golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWP
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211209193657-4570a0811e8b/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20220307211146-efcb8507fb70/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 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=
@@ -830,9 +831,9 @@ golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220307203707-22a9840ba4d7/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220408201424-a24fb2fb8a0f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -74,12 +74,16 @@ func (r *MonitoredReader) Read(buf []byte) (n int, err error) {
 		need = int(math.Min(float64(b), float64(need)))
 		tokens = need
 	}
-
+	// reduce tokens requested according to availability
+	av := int(r.throttle.Tokens())
+	if av < tokens && av > 0 {
+		tokens = av
+		need = int(math.Min(float64(tokens), float64(need)))
+	}
 	err = r.throttle.WaitN(r.ctx, tokens)
 	if err != nil {
 		return
 	}
-
 	n, err = r.r.Read(buf[:need])
 	if err != nil {
 		r.lastErr = err
@@ -40,6 +40,7 @@ const (
 	apiListQuorum              = "list_quorum"
 	apiReplicationPriority     = "replication_priority"
 	apiReplicationMaxWorkers   = "replication_max_workers"
+	apiReplicationMaxLWorkers  = "replication_max_lrg_workers"

 	apiTransitionWorkers           = "transition_workers"
 	apiStaleUploadsCleanupInterval = "stale_uploads_cleanup_interval"
@@ -52,16 +53,18 @@ const (
 	apiSyncEvents                  = "sync_events"
 	apiObjectMaxVersions           = "object_max_versions"

-	EnvAPIRequestsMax                 = "MINIO_API_REQUESTS_MAX"
-	EnvAPIRequestsDeadline            = "MINIO_API_REQUESTS_DEADLINE"
-	EnvAPIClusterDeadline             = "MINIO_API_CLUSTER_DEADLINE"
-	EnvAPICorsAllowOrigin             = "MINIO_API_CORS_ALLOW_ORIGIN"
-	EnvAPIRemoteTransportDeadline     = "MINIO_API_REMOTE_TRANSPORT_DEADLINE"
-	EnvAPITransitionWorkers           = "MINIO_API_TRANSITION_WORKERS"
-	EnvAPIListQuorum                  = "MINIO_API_LIST_QUORUM"
-	EnvAPISecureCiphers               = "MINIO_API_SECURE_CIPHERS" // default config.EnableOn
-	EnvAPIReplicationPriority         = "MINIO_API_REPLICATION_PRIORITY"
-	EnvAPIReplicationMaxWorkers       = "MINIO_API_REPLICATION_MAX_WORKERS"
+	EnvAPIRequestsMax             = "MINIO_API_REQUESTS_MAX"
+	EnvAPIRequestsDeadline        = "MINIO_API_REQUESTS_DEADLINE"
+	EnvAPIClusterDeadline         = "MINIO_API_CLUSTER_DEADLINE"
+	EnvAPICorsAllowOrigin         = "MINIO_API_CORS_ALLOW_ORIGIN"
+	EnvAPIRemoteTransportDeadline = "MINIO_API_REMOTE_TRANSPORT_DEADLINE"
+	EnvAPITransitionWorkers       = "MINIO_API_TRANSITION_WORKERS"
+	EnvAPIListQuorum              = "MINIO_API_LIST_QUORUM"
+	EnvAPISecureCiphers           = "MINIO_API_SECURE_CIPHERS" // default config.EnableOn
+	EnvAPIReplicationPriority     = "MINIO_API_REPLICATION_PRIORITY"
+	EnvAPIReplicationMaxWorkers   = "MINIO_API_REPLICATION_MAX_WORKERS"
+	EnvAPIReplicationMaxLWorkers  = "MINIO_API_REPLICATION_MAX_LRG_WORKERS"
+
 	EnvAPIStaleUploadsCleanupInterval = "MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL"
 	EnvAPIStaleUploadsExpiry          = "MINIO_API_STALE_UPLOADS_EXPIRY"
 	EnvAPIDeleteCleanupInterval       = "MINIO_API_DELETE_CLEANUP_INTERVAL"
@@ -117,6 +120,10 @@ var (
 			Key:   apiReplicationMaxWorkers,
 			Value: "500",
 		},
+		config.KV{
+			Key:   apiReplicationMaxLWorkers,
+			Value: "10",
+		},
 		config.KV{
 			Key:   apiTransitionWorkers,
 			Value: "100",
@@ -171,6 +178,7 @@ type Config struct {
 	ListQuorum                  string        `json:"list_quorum"`
 	ReplicationPriority         string        `json:"replication_priority"`
 	ReplicationMaxWorkers       int           `json:"replication_max_workers"`
+	ReplicationMaxLWorkers      int           `json:"replication_max_lrg_workers"`
 	TransitionWorkers           int           `json:"transition_workers"`
 	StaleUploadsCleanupInterval time.Duration `json:"stale_uploads_cleanup_interval"`
 	StaleUploadsExpiry          time.Duration `json:"stale_uploads_expiry"`
@@ -280,11 +288,21 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 	if err != nil {
 		return cfg, err
 	}
-
 	if replicationMaxWorkers <= 0 || replicationMaxWorkers > 500 {
 		return cfg, config.ErrInvalidReplicationWorkersValue(nil).Msg("Number of replication workers should be between 1 and 500")
 	}
 	cfg.ReplicationMaxWorkers = replicationMaxWorkers
+
+	replicationMaxLWorkers, err := strconv.Atoi(env.Get(EnvAPIReplicationMaxLWorkers, kvs.GetWithDefault(apiReplicationMaxLWorkers, DefaultKVS)))
+	if err != nil {
+		return cfg, err
+	}
+	if replicationMaxLWorkers <= 0 || replicationMaxLWorkers > 10 {
+		return cfg, config.ErrInvalidReplicationWorkersValue(nil).Msg("Number of replication workers for transfers >=128MiB should be between 1 and 10 per node")
+	}
+
+	cfg.ReplicationMaxLWorkers = replicationMaxLWorkers
+
 	transitionWorkers, err := strconv.Atoi(env.Get(EnvAPITransitionWorkers, kvs.GetWithDefault(apiTransitionWorkers, DefaultKVS)))
 	if err != nil {
 		return cfg, err
@@ -402,3 +402,21 @@ func (l *Config) LookupGroupMemberships(userDistNames []string, userDNToUsername

 	return res, nil
 }
+
+// QuickNormalizeDN - normalizes the given DN without checking if it is valid or
+// exists in the LDAP directory. Returns input if error
+func (l Config) QuickNormalizeDN(dn string) string {
+	if normDN, err := xldap.NormalizeDN(dn); err == nil {
+		return normDN
+	}
+	return dn
+}
+
+// DecodeDN - denormalizes the given DN by unescaping any escaped characters.
+// Returns input if error
+func (l Config) DecodeDN(dn string) string {
+	if decodedDN, err := xldap.DecodeDN(dn); err == nil {
+		return decodedDN
+	}
+	return dn
+}
@@ -101,12 +101,14 @@ var (
 			Value: "",
 		},
 		config.KV{
-			Key:   ClaimPrefix,
-			Value: "",
+			Key:           ClaimPrefix,
+			Value:         "",
+			HiddenIfEmpty: true,
 		},
 		config.KV{
-			Key:   RedirectURI,
-			Value: "",
+			Key:           RedirectURI,
+			Value:         "",
+			HiddenIfEmpty: true,
 		},
 		config.KV{
 			Key:   RedirectURIDynamic,
@@ -117,7 +117,7 @@ func (k *KeycloakProvider) LookupUser(userid string) (User, error) {
 	case http.StatusUnauthorized:
 		return User{}, ErrAccessTokenExpired
 	}
-	return User{}, fmt.Errorf("Unable to lookup %s - keycloak user lookup returned %v", userid, resp.Status)
+	return User{}, fmt.Errorf("Unable to lookup - keycloak user lookup returned %v", resp.Status)
 }

 // Option is a function type that accepts a pointer Target
@@ -861,20 +861,24 @@ var (
 			Value: config.EnableOff,
 		},
 		config.KV{
-			Key:   target.NATSStreaming,
-			Value: config.EnableOff,
+			Key:           target.NATSStreaming,
+			Value:         config.EnableOff,
+			HiddenIfEmpty: true,
 		},
 		config.KV{
-			Key:   target.NATSStreamingAsync,
-			Value: config.EnableOff,
+			Key:           target.NATSStreamingAsync,
+			Value:         config.EnableOff,
+			HiddenIfEmpty: true,
 		},
 		config.KV{
-			Key:   target.NATSStreamingMaxPubAcksInFlight,
-			Value: "0",
+			Key:           target.NATSStreamingMaxPubAcksInFlight,
+			Value:         "0",
+			HiddenIfEmpty: true,
 		},
 		config.KV{
-			Key:   target.NATSStreamingClusterID,
-			Value: "",
+			Key:           target.NATSStreamingClusterID,
+			Value:         "",
+			HiddenIfEmpty: true,
 		},
 		config.KV{
 			Key:   target.NATSQueueDir,
@@ -42,12 +42,14 @@ const (
 var (
 	DefaultKVS = config.KVS{
 		config.KV{
-			Key:   URL,
-			Value: "",
+			Key:           URL,
+			Value:         "",
+			HiddenIfEmpty: true,
 		},
 		config.KV{
-			Key:   AuthToken,
-			Value: "",
+			Key:           AuthToken,
+			Value:         "",
+			HiddenIfEmpty: true,
 		},
 	}
 )
@@ -1,4 +1,4 @@
-// Copyright (c) 2015-2021 MinIO, Inc.
+// Copyright (c) 2015-2024 MinIO, Inc.
 //
 // This file is part of MinIO Object Storage stack
 //
@@ -27,23 +27,23 @@ var (
 	// HelpSubnet - provides help for subnet api key config
 	HelpSubnet = config.HelpKVS{
 		config.HelpKV{
-			Key:         config.License, // Deprecated Dec 2021
+			Key:         config.License,
 			Type:        "string",
-			Description: "[DEPRECATED use api_key] Subnet license token for the cluster" + defaultHelpPostfix(config.License),
+			Description: "Enterprise license for the cluster" + defaultHelpPostfix(config.License),
 			Optional:    true,
 			Sensitive:   true,
 		},
 		config.HelpKV{
 			Key:         config.APIKey,
 			Type:        "string",
-			Description: "Subnet api key for the cluster" + defaultHelpPostfix(config.APIKey),
+			Description: "Enterprise license API key for the cluster" + defaultHelpPostfix(config.APIKey),
 			Optional:    true,
 			Sensitive:   true,
 		},
 		config.HelpKV{
 			Key:         config.Proxy,
 			Type:        "string",
-			Description: "HTTP(S) proxy URL to use for connecting to SUBNET" + defaultHelpPostfix(config.Proxy),
+			Description: "HTTP(s) proxy URL to use for connecting to SUBNET" + defaultHelpPostfix(config.Proxy),
 			Optional:    true,
 			Sensitive:   true,
 		},
@@ -49,8 +49,8 @@ var generateKeyTests = []struct {
 }

 func TestGenerateKey(t *testing.T) {
-	defer func(l bool) { logger.DisableErrorLog = l }(logger.DisableErrorLog)
-	logger.DisableErrorLog = true
+	defer func(l bool) { logger.DisableLog = l }(logger.DisableLog)
+	logger.DisableLog = true

 	for i, test := range generateKeyTests {
 		i, test := i, test
@@ -75,8 +75,8 @@ var generateIVTests = []struct {
 }

 func TestGenerateIV(t *testing.T) {
-	defer func(l bool) { logger.DisableErrorLog = l }(logger.DisableErrorLog)
-	logger.DisableErrorLog = true
+	defer func(l bool) { logger.DisableLog = l }(logger.DisableLog)
+	logger.DisableLog = true

 	for i, test := range generateIVTests {
 		i, test := i, test
@@ -313,8 +313,8 @@ var s3CreateMetadataTests = []struct {
 }

 func TestS3CreateMetadata(t *testing.T) {
-	defer func(l bool) { logger.DisableErrorLog = l }(logger.DisableErrorLog)
-	logger.DisableErrorLog = true
+	defer func(l bool) { logger.DisableLog = l }(logger.DisableLog)
+	logger.DisableLog = true
 	for i, test := range s3CreateMetadataTests {
 		metadata := S3.CreateMetadata(nil, test.KeyID, test.SealedDataKey, test.SealedKey)
 		keyID, kmsKey, sealedKey, err := S3.ParseMetadata(metadata)
@@ -358,8 +358,8 @@ var ssecCreateMetadataTests = []struct {
 }

 func TestSSECCreateMetadata(t *testing.T) {
-	defer func(l bool) { logger.DisableErrorLog = l }(logger.DisableErrorLog)
-	logger.DisableErrorLog = true
+	defer func(l bool) { logger.DisableLog = l }(logger.DisableLog)
+	logger.DisableLog = true
 	for i, test := range ssecCreateMetadataTests {
 		metadata := SSEC.CreateMetadata(nil, test.SealedKey)
 		sealedKey, err := SSEC.ParseMetadata(metadata)
@@ -925,137 +925,141 @@ func (c *Connection) handleMessages(ctx context.Context, conn net.Conn) {
 	c.handleMsgWg.Add(2)
 	c.reconnectMu.Unlock()

-	// Read goroutine
-	go func() {
-		defer func() {
-			if rec := recover(); rec != nil {
-				gridLogIf(ctx, fmt.Errorf("handleMessages: panic recovered: %v", rec))
-				debug.PrintStack()
-			}
-			c.connChange.L.Lock()
-			if atomic.CompareAndSwapUint32((*uint32)(&c.state), StateConnected, StateConnectionError) {
-				c.connChange.Broadcast()
-			}
-			c.connChange.L.Unlock()
-			conn.Close()
-			c.handleMsgWg.Done()
-		}()
+	// Start reader and writer
+	go c.readStream(ctx, conn, cancel)
+	c.writeStream(ctx, conn, cancel)
+}

-		controlHandler := wsutil.ControlFrameHandler(conn, c.side)
-		wsReader := wsutil.Reader{
-			Source:          conn,
-			State:           c.side,
-			CheckUTF8:       true,
-			SkipHeaderCheck: false,
-			OnIntermediate:  controlHandler,
+// readStream handles the read side of the connection.
+// It will read messages and send them to c.handleMsg.
+// If an error occurs the cancel function will be called and conn be closed.
+// The function will block until the connection is closed or an error occurs.
+func (c *Connection) readStream(ctx context.Context, conn net.Conn, cancel context.CancelCauseFunc) {
+	defer func() {
+		if rec := recover(); rec != nil {
+			gridLogIf(ctx, fmt.Errorf("handleMessages: panic recovered: %v", rec))
+			debug.PrintStack()
 		}
-		readDataInto := func(dst []byte, rw io.ReadWriter, s ws.State, want ws.OpCode) ([]byte, error) {
-			dst = dst[:0]
-			for {
-				hdr, err := wsReader.NextFrame()
-				if err != nil {
-					return nil, err
-				}
-				if hdr.OpCode.IsControl() {
-					if err := controlHandler(hdr, &wsReader); err != nil {
-						return nil, err
-					}
-					continue
-				}
-				if hdr.OpCode&want == 0 {
-					if err := wsReader.Discard(); err != nil {
-						return nil, err
-					}
-					continue
-				}
-				if int64(cap(dst)) < hdr.Length+1 {
-					dst = make([]byte, 0, hdr.Length+hdr.Length>>3)
-				}
-				return readAllInto(dst[:0], &wsReader)
-			}
-		}
-
-		// Keep reusing the same buffer.
-		var msg []byte
-		for {
-			if atomic.LoadUint32((*uint32)(&c.state)) != StateConnected {
-				cancel(ErrDisconnected)
-				return
-			}
-			if cap(msg) > readBufferSize*4 {
-				// Don't keep too much memory around.
-				msg = nil
-			}
-
-			var err error
-			msg, err = readDataInto(msg, conn, c.side, ws.OpBinary)
-			if err != nil {
-				cancel(ErrDisconnected)
-				if !xnet.IsNetworkOrHostDown(err, true) {
-					gridLogIfNot(ctx, fmt.Errorf("ws read: %w", err), net.ErrClosed, io.EOF)
-				}
-				return
-			}
-			block := c.blockMessages.Load()
-			if block != nil && *block != nil {
-				<-*block
-			}
-
-			if c.incomingBytes != nil {
-				c.incomingBytes(int64(len(msg)))
-			}
-
-			// Parse the received message
-			var m message
-			subID, remain, err := m.parse(msg)
-			if err != nil {
-				if !xnet.IsNetworkOrHostDown(err, true) {
-					gridLogIf(ctx, fmt.Errorf("ws parse package: %w", err))
-				}
-				cancel(ErrDisconnected)
-				return
-			}
-			if debugPrint {
-				fmt.Printf("%s Got msg: %v\n", c.Local, m)
-			}
-			if m.Op != OpMerged {
-				c.inMessages.Add(1)
-				c.handleMsg(ctx, m, subID)
-				continue
-			}
-			// Handle merged messages.
-			messages := int(m.Seq)
-			c.inMessages.Add(int64(messages))
-			for i := 0; i < messages; i++ {
-				if atomic.LoadUint32((*uint32)(&c.state)) != StateConnected {
-					cancel(ErrDisconnected)
-					return
-				}
-				var next []byte
-				next, remain, err = msgp.ReadBytesZC(remain)
-				if err != nil {
-					if !xnet.IsNetworkOrHostDown(err, true) {
-						gridLogIf(ctx, fmt.Errorf("ws read merged: %w", err))
-					}
-					cancel(ErrDisconnected)
-					return
-				}
-
-				m.Payload = nil
-				subID, _, err = m.parse(next)
-				if err != nil {
-					if !xnet.IsNetworkOrHostDown(err, true) {
-						gridLogIf(ctx, fmt.Errorf("ws parse merged: %w", err))
-					}
-					cancel(ErrDisconnected)
-					return
-				}
-				c.handleMsg(ctx, m, subID)
-			}
+		cancel(ErrDisconnected)
+		c.connChange.L.Lock()
+		if atomic.CompareAndSwapUint32((*uint32)(&c.state), StateConnected, StateConnectionError) {
+			c.connChange.Broadcast()
 		}
+		c.connChange.L.Unlock()
+		conn.Close()
+		c.handleMsgWg.Done()
 	}()

-	// Write function.
+	controlHandler := wsutil.ControlFrameHandler(conn, c.side)
+	wsReader := wsutil.Reader{
+		Source:          conn,
+		State:           c.side,
+		CheckUTF8:       true,
+		SkipHeaderCheck: false,
+		OnIntermediate:  controlHandler,
+	}
+	readDataInto := func(dst []byte, rw io.ReadWriter, s ws.State, want ws.OpCode) ([]byte, error) {
+		dst = dst[:0]
+		for {
+			hdr, err := wsReader.NextFrame()
+			if err != nil {
+				return nil, err
+			}
+			if hdr.OpCode.IsControl() {
+				if err := controlHandler(hdr, &wsReader); err != nil {
+					return nil, err
+				}
+				continue
+			}
+			if hdr.OpCode&want == 0 {
+				if err := wsReader.Discard(); err != nil {
+					return nil, err
+				}
+				continue
+			}
+			if int64(cap(dst)) < hdr.Length+1 {
+				dst = make([]byte, 0, hdr.Length+hdr.Length>>3)
+			}
+			return readAllInto(dst[:0], &wsReader)
+		}
+	}
+
+	// Keep reusing the same buffer.
+	var msg []byte
+	for atomic.LoadUint32((*uint32)(&c.state)) == StateConnected {
+		if cap(msg) > readBufferSize*4 {
+			// Don't keep too much memory around.
+			msg = nil
+		}
+
+		var err error
+		msg, err = readDataInto(msg, conn, c.side, ws.OpBinary)
+		if err != nil {
+			if !xnet.IsNetworkOrHostDown(err, true) {
+				gridLogIfNot(ctx, fmt.Errorf("ws read: %w", err), net.ErrClosed, io.EOF)
+			}
+			return
+		}
+		block := c.blockMessages.Load()
+		if block != nil && *block != nil {
+			<-*block
+		}
+
+		if c.incomingBytes != nil {
+			c.incomingBytes(int64(len(msg)))
+		}
+
+		// Parse the received message
+		var m message
+		subID, remain, err := m.parse(msg)
+		if err != nil {
+			if !xnet.IsNetworkOrHostDown(err, true) {
+				gridLogIf(ctx, fmt.Errorf("ws parse package: %w", err))
+			}
+			return
+		}
+		if debugPrint {
+			fmt.Printf("%s Got msg: %v\n", c.Local, m)
+		}
+		if m.Op != OpMerged {
+			c.inMessages.Add(1)
+			c.handleMsg(ctx, m, subID)
+			continue
+		}
+		// Handle merged messages.
+		messages := int(m.Seq)
+		c.inMessages.Add(int64(messages))
+		for i := 0; i < messages; i++ {
+			if atomic.LoadUint32((*uint32)(&c.state)) != StateConnected {
+				return
+			}
+			var next []byte
+			next, remain, err = msgp.ReadBytesZC(remain)
+			if err != nil {
+				if !xnet.IsNetworkOrHostDown(err, true) {
+					gridLogIf(ctx, fmt.Errorf("ws read merged: %w", err))
+				}
+				return
+			}
+
+			m.Payload = nil
+			subID, _, err = m.parse(next)
+			if err != nil {
+				if !xnet.IsNetworkOrHostDown(err, true) {
+					gridLogIf(ctx, fmt.Errorf("ws parse merged: %w", err))
+				}
+				return
+			}
+			c.handleMsg(ctx, m, subID)
+		}
+	}
+}
+
+// writeStream handles the read side of the connection.
+// It will grab messages from c.outQueue and write them to the connection.
+// If an error occurs the cancel function will be called and conn be closed.
+// The function will block until the connection is closed or an error occurs.
+func (c *Connection) writeStream(ctx context.Context, conn net.Conn, cancel context.CancelCauseFunc) {
 	defer func() {
 		if rec := recover(); rec != nil {
 			gridLogIf(ctx, fmt.Errorf("handleMessages: panic recovered: %v", rec))
@@ -1,4 +1,4 @@
-// Copyright (c) 2015-2021 MinIO, Inc.
+// Copyright (c) 2015-2024 MinIO, Inc.
 //
 // This file is part of MinIO Object Storage stack
 //
@@ -110,7 +110,7 @@ func (f fatalMsg) quiet(msg string, args ...interface{}) {
 }

 var (
-	logTag      = "ERROR"
+	logTag      = "FATAL"
 	logBanner   = color.BgRed(color.FgWhite(color.Bold(logTag))) + " "
 	emptyBanner = color.BgRed(strings.Repeat(" ", len(logTag))) + " "
 	bannerWidth = len(logTag) + 1
@@ -187,7 +187,7 @@ func (i infoMsg) pretty(msg string, args ...interface{}) {
 	if msg == "" {
 		fmt.Fprintln(Output, args...)
 	} else {
-		fmt.Fprintf(Output, msg, args...)
+		fmt.Fprintf(Output, `INFO: `+msg, args...)
 	}
 }

@@ -222,13 +222,13 @@ func (i errorMsg) pretty(msg string, args ...interface{}) {
 	if msg == "" {
 		fmt.Fprintln(Output, args...)
 	} else {
-		fmt.Fprintf(Output, msg, args...)
+		fmt.Fprintf(Output, `ERRO: `+msg, args...)
 	}
 }

 // Error :
 func Error(msg string, data ...interface{}) {
-	if DisableErrorLog {
+	if DisableLog {
 		return
 	}
 	consoleLog(errorMessage, msg, data...)
@@ -236,8 +236,92 @@ func Error(msg string, data ...interface{}) {

 // Info :
 func Info(msg string, data ...interface{}) {
-	if DisableErrorLog {
+	if DisableLog {
 		return
 	}
 	consoleLog(info, msg, data...)
 }
+
+// Startup :
+func Startup(msg string, data ...interface{}) {
+	if DisableLog {
+		return
+	}
+	consoleLog(startup, msg, data...)
+}
+
+type startupMsg struct{}
+
+var startup startupMsg
+
+func (i startupMsg) json(msg string, args ...interface{}) {
+	var message string
+	if msg != "" {
+		message = fmt.Sprintf(msg, args...)
+	} else {
+		message = fmt.Sprint(args...)
+	}
+	logJSON, err := json.Marshal(&log.Entry{
+		Level:   InfoKind,
+		Message: message,
+		Time:    time.Now().UTC(),
+	})
+	if err != nil {
+		panic(err)
+	}
+	fmt.Fprintln(Output, string(logJSON))
+}
+
+func (i startupMsg) quiet(msg string, args ...interface{}) {
+}
+
+func (i startupMsg) pretty(msg string, args ...interface{}) {
+	if msg == "" {
+		fmt.Fprintln(Output, args...)
+	} else {
+		fmt.Fprintf(Output, msg, args...)
+	}
+}
+
+type warningMsg struct{}
+
+var warningMessage warningMsg
+
+func (i warningMsg) json(msg string, args ...interface{}) {
+	var message string
+	if msg != "" {
+		message = fmt.Sprintf(msg, args...)
+	} else {
+		message = fmt.Sprint(args...)
+	}
+	logJSON, err := json.Marshal(&log.Entry{
+		Level:   WarningKind,
+		Message: message,
+		Time:    time.Now().UTC(),
+		Trace:   &log.Trace{Message: message, Source: []string{getSource(6)}},
+	})
+	if err != nil {
+		panic(err)
+	}
+	fmt.Fprintln(Output, string(logJSON))
+}
+
+func (i warningMsg) quiet(msg string, args ...interface{}) {
+	i.pretty(msg, args...)
+}
+
+func (i warningMsg) pretty(msg string, args ...interface{}) {
+	if msg == "" {
+		fmt.Fprintln(Output, args...)
+	} else {
+		fmt.Fprintf(Output, `WARN: `+msg, args...)
+	}
+}
+
+// Warning :
+func Warning(msg string, data ...interface{}) {
+	if DisableLog {
+		return
+	}
+	consoleLog(warningMessage, msg, data...)
+}
@@ -53,8 +53,8 @@ const (
 )

 var (
-	// DisableErrorLog avoids printing error/event/info kind of logs
-	DisableErrorLog = false
+	// DisableLog avoids printing error/event/info kind of logs
+	DisableLog = false
 	// Output allows configuring custom writer, defaults to os.Stderr
 	Output io.Writer = os.Stderr
 )
@@ -386,7 +386,7 @@ func buildLogEntry(ctx context.Context, subsystem, message string, trace []strin
 // consoleLogIf prints a detailed error message during
 // the execution of the server.
 func consoleLogIf(ctx context.Context, subsystem string, err error, errKind ...interface{}) {
-	if DisableErrorLog {
+	if DisableLog {
 		return
 	}
 	if err == nil {
@@ -401,7 +401,7 @@ func consoleLogIf(ctx context.Context, subsystem string, err error, errKind ...i
 // logIf prints a detailed error message during
 // the execution of the server.
 func logIf(ctx context.Context, subsystem string, err error, errKind ...interface{}) {
-	if DisableErrorLog {
+	if DisableLog {
 		return
 	}
 	if err == nil {
@@ -430,7 +430,7 @@ func sendLog(ctx context.Context, entry log.Entry) {

 // Event sends a event log to  log targets
 func Event(ctx context.Context, subsystem, msg string, args ...interface{}) {
-	if DisableErrorLog {
+	if DisableLog {
 		return
 	}
 	entry := logToEntry(ctx, subsystem, fmt.Sprintf(msg, args...), EventKind)
Author	SHA1	Message	Date
Klaus Post	ded373e600	Split handleMessages (cosmetic) (#20095 ) Split the read and write sides of handleMessages into two separate functions Cosmetic. The only non-copy-and-paste change is that `cancel(ErrDisconnected)` is moved into the defer on `readStream`.	2024-07-15 12:02:30 -07:00
Harshavardhana	e8c54c3d6c	add validation test for v3 metrics for all its endpoints (#20094 ) add unit test for v3 metrics for all its exposed endpoints Bonus: - support OpenMetrics encoding - adds boot time for prometheus - continueOnError is better to serve as much metrics as possible.	2024-07-15 09:28:02 -07:00
Shubhendu	f944a42886	Removed user and group details from logs (#20072 ) Signed-off-by: Shubhendu Ram Tripathi <shubhendu@minio.io>	2024-07-14 11:12:07 -07:00
Harshavardhana	eff0ea43aa	fix: typo in BucketUsageMetrics group registration in v3 metrics (#20090 ) ``` curl http://localhost:9000/minio/metrics/v3/cluster/usage/buckets ``` Did not work as documented, due to the fact that there was a typo in the bucket usage metrics registration group. This endpoint is a cluster endpoint and does not require any `buckets` argument.	2024-07-14 11:11:42 -07:00
Minio Trusted	3b602bb532	Update yaml files to latest version RELEASE.2024-07-13T01-46-15Z	2024-07-13 02:08:28 +00:00
Alex	459985f0fa	Update Console to v1.6.3 (#20084 ) Signed-off-by: Benjamin Perez <benjamin@bexsoft.net>	2024-07-12 18:46:15 -07:00
Harshavardhana	d0080046c2	allow sysfs tuning from tuned	2024-07-12 16:31:18 -07:00
Harshavardhana	7fcb428622	do not print unexpected logs (#20083 )	2024-07-12 13:51:54 -07:00
dependabot[bot]	4ea6f94ed8	Bump github.com/nats-io/nats-streaming-server from 0.24.3 to 0.24.6 (#20082 ) Bumps [github.com/nats-io/nats-streaming-server](https://github.com/nats-io/nats-streaming-server) from 0.24.3 to 0.24.6. - [Release notes](https://github.com/nats-io/nats-streaming-server/releases) - [Changelog](https://github.com/nats-io/nats-streaming-server/blob/main/.goreleaser.yml) - [Commits](https://github.com/nats-io/nats-streaming-server/compare/v0.24.3...v0.24.6) --- updated-dependencies: - dependency-name: github.com/nats-io/nats-streaming-server dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2024-07-12 11:45:09 -07:00
Klaus Post	83adc2eebf	Fix ListObjects aborting after 3 minute on async request (#20074 ) When creating the async listing, if the first request does not return within 3 minutes, it is stopped, since it isn't being kept alive. Keep updating `lastHandout` while we are waiting for the initial request to be fulfilled.	2024-07-12 09:23:16 -07:00
Poorna	989c318a28	replication: make large workers configurable (#20077 ) This PR also improves throttling by reducing tokens requested from rate limiter based on available tokens to avoid exceeding throttle wait deadlines	2024-07-12 07:57:31 -07:00
Frank Wessels	ef802f2b2c	Updated dependencies for ARM SVE support (#20081 )	2024-07-12 07:36:29 -07:00
Taran Pelkey	f5d2fbc84c	Add DecodeDN and QuickNormalizeDN functions to LDAP config (#20076 )	2024-07-11 18:04:53 -07:00
Allan Roger Reid	e139673969	Audit failure in batch job key rotate (#20073 )	2024-07-11 16:13:15 -07:00
Harshavardhana	a8c6465f22	hide some deprecated fields from 'get' output (#20069 ) also update wording on `subnet license="" api_key=""`	2024-07-10 13:16:44 -07:00
Minio Trusted	27538e2d22	Update yaml files to latest version RELEASE.2024-07-10T18-41-49Z	2024-07-10 19:27:17 +00:00