support reading systemctl config automatically on baremetal setups (#15066)

this allows for customers to use `mc admin service restart`
directly even when performing RPM, DEB upgrades. Upon such 'restart'
after upgrade MinIO will re-read the /etc/default/minio for any
newer environment variables.

As long as `MINIO_CONFIG_ENV_FILE=/etc/default/minio` is set, this
is honored.

.github/workflows/go-fips.yml

@@ -0,0 +1,51 @@
+name: FIPS Build Test
+
+on:
+  pull_request:
+    branches:
+    - master
+
+# This ensures that previous jobs for the PR are canceled when the PR is
+# updated.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    name: Go BoringCrypto ${{ matrix.go-version }} on ${{ matrix.os }}
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        go-version: [1.17.11b7, 1.18.3b7]
+        os: [ubuntu-latest]
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v2
+
+      - name: Setup dockerfile for build test
+        run: |
+          echo "FROM us-docker.pkg.dev/google.com/api-project-999119582588/go-boringcrypto/golang:${{ matrix.go-version }}" > Dockerfile.fips.test
+          echo "COPY . /minio" >> Dockerfile.fips.test
+          echo "WORKDIR /minio" >> Dockerfile.fips.test
+          echo "RUN make" >> Dockerfile.fips.test
+
+      - name: Build
+        uses: docker/build-push-action@v3
+        with:
+          context: .
+          file: Dockerfile.fips.test
+          push: false
+          load: true
+          tags: minio/fips-test:latest
+
+      # This should fail if grep returns non-zero exit
+      - name: Test binary
+        run: |
+          docker run --rm minio/fips-test:latest ./minio --version
+          docker run --rm -i minio/fips-test:latest /bin/bash -c 'go tool nm ./minio' | grep -q FIPS

.gitignore

@@ -9,8 +9,7 @@ site/
 /.idea/
 /Minio.iml
 **/access.log
-vendor/**/*.js
-vendor/**/*.json
+vendor/
 .DS_Store
 *.syso
 coverage.txt

Makefile

@@ -58,6 +58,7 @@ test-iam: build ## verify IAM (external IDP, etcd backends)
 test-replication: install ## verify multi site replication
 	@echo "Running tests for replicating three sites"
 	@(env bash $(PWD)/docs/bucket/replication/setup_3site_replication.sh)
+	@(env bash $(PWD)/docs/bucket/replication/setup_2site_existing_replication.sh)
 
 test-site-replication-ldap: install ## verify automatic site replication
 	@echo "Running tests for automatic site replication of IAM (with LDAP)"

README.fips.md

@@ -0,0 +1,7 @@
+# MinIO FIPS Builds
+
+MinIO creates FIPS builds using a patched version of the Go compiler (that uses BoringCrypto, from BoringSSL, which is [FIPS 140-2 validated](https://csrc.nist.gov/csrc/media/projects/cryptographic-module-validation-program/documents/security-policies/140sp2964.pdf)) published by the Golang Team [here](https://github.com/golang/go/tree/dev.boringcrypto/misc/boring).
+
+MinIO FIPS executables are available at http://dl.min.io - they are only published for `linux-amd64` architecture as binary files with the suffix `.fips`. We also publish corresponding container images to our official image repositories.
+
+We are not making any statements or representations about the suitability of this code or build in relation to the FIPS 140-2 standard. Interested users will have to evaluate for themselves whether this is useful for their own purposes.

README.md

@@ -236,16 +236,16 @@ Upgrades require zero downtime in MinIO, all upgrades are non-disruptive, all tr
 mc admin update <minio alias, e.g., myminio>
 ```
 
-- For deployments without external internet access (e.g. airgapped environments), download the binary from <https://dl.min.io> and replace the existing MinIO binary let's say for example `/opt/bin/minio`, apply executable permissions `chmod +x /opt/bin/minio` and do `mc admin service restart alias/`.
+- For deployments without external internet access (e.g. airgapped environments), download the binary from <https://dl.min.io> and replace the existing MinIO binary let's say for example `/opt/bin/minio`, apply executable permissions `chmod +x /opt/bin/minio` and proceed to perform `mc admin service restart alias/`.
 
-- For installations using Systemd MinIO service, upgrade via RPM/DEB packages **parallelly** on all servers or replace the binary lets say `/opt/bin/minio` on all nodes, apply executable permissions `chmod +x /opt/bin/minio`. Proceed to perform `systemctl restart minio` across all nodes in **parallel**.
+- For installations using Systemd MinIO service, upgrade via RPM/DEB packages **parallelly** on all servers or replace the binary lets say `/opt/bin/minio` on all nodes, apply executable permissions `chmod +x /opt/bin/minio` and process to perform `mc admin service restart alias/`.
 
 ### Upgrade Checklist
 
 - Test all upgrades in a lower environment (DEV, QA, UAT) before applying to production. Performing blind upgrades in production environments carries significant risk.
-- Read the release notes for the targeted MinIO release *before* performing any installation, there is no forced requirement to upgrade to latest releases every week. If it has a bug fix you are looking for then yes, else avoid actively upgrading a running production system.
-- If you plan to use `mc admin update`, MinIO process must have write access to the parent directory to provide in-place upgrades.
-- `mc admin update` is not supported in kubernetes/container environments, container environments provide their own mechanisms for container updates.
+- Read the release notes for MinIO *before* performing any upgrade, there is no forced requirement to upgrade to latest releases upon every releases. Some releases may not be relevant to your setup, avoid upgrading production environments unnecessarily.
+- If you plan to use `mc admin update`, MinIO process must have write access to the parent directory where the binary is present on the host system.
+- `mc admin update` is not supported and should be avoided in kubernetes/container environments, please upgrade containers by upgrading relevant container images.
 - **We do not recommend upgrading one MinIO server at a time, the product is designed to support parallel upgrades please follow our recommended guidelines.**
 
 ## Explore Further

cmd/admin-handlers.go

@@ -291,6 +291,8 @@ type ServerHTTPStats struct {
 	CurrentS3Requests      ServerHTTPAPIStats `json:"currentS3Requests"`
 	TotalS3Requests        ServerHTTPAPIStats `json:"totalS3Requests"`
 	TotalS3Errors          ServerHTTPAPIStats `json:"totalS3Errors"`
+	TotalS35xxErrors       ServerHTTPAPIStats `json:"totalS35xxErrors"`
+	TotalS34xxErrors       ServerHTTPAPIStats `json:"totalS34xxErrors"`
 	TotalS3Canceled        ServerHTTPAPIStats `json:"totalS3Canceled"`
 	TotalS3RejectedAuth    uint64             `json:"totalS3RejectedAuth"`
 	TotalS3RejectedTime    uint64             `json:"totalS3RejectedTime"`
@@ -1409,9 +1411,15 @@ func (a adminAPIHandlers) TraceHandler(w http.ResponseWriter, r *http.Request) {
 
 	peers, _ := newPeerRestClients(globalEndpoints)
 
-	globalTrace.Subscribe(traceCh, ctx.Done(), func(entry interface{}) bool {
+	traceFn := func(entry interface{}) bool {
 		return mustTrace(entry, traceOpts)
-	})
+	}
+
+	err = globalTrace.Subscribe(traceCh, ctx.Done(), traceFn)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrSlowDown), r.URL)
+		return
+	}
 
 	for _, peer := range peers {
 		if peer == nil {
@@ -1483,7 +1491,11 @@ func (a adminAPIHandlers) ConsoleLogHandler(w http.ResponseWriter, r *http.Reque
 
 	peers, _ := newPeerRestClients(globalEndpoints)
 
-	globalConsoleSys.Subscribe(logCh, ctx.Done(), node, limitLines, logKind, nil)
+	err = globalConsoleSys.Subscribe(logCh, ctx.Done(), node, limitLines, logKind, nil)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrSlowDown), r.URL)
+		return
+	}
 
 	for _, peer := range peers {
 		if peer == nil {
@@ -1821,9 +1833,6 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 	healthInfoCh := make(chan madmin.HealthInfo)
 
 	enc := json.NewEncoder(w)
-	partialWrite := func(oinfo madmin.HealthInfo) {
-		healthInfoCh <- oinfo
-	}
 
 	setCommonHeaders(w)
 
@@ -1854,9 +1863,6 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 		}
 	}
 
-	deadlinedCtx, deadlineCancel := context.WithTimeout(ctx, deadline)
-	defer deadlineCancel()
-
 	nsLock := objectAPI.NewNSLock(minioMetaBucket, "health-check-in-progress")
 	lkctx, err := nsLock.GetLock(ctx, newDynamicTimeout(deadline, deadline))
 	if err != nil { // returns a locked lock
@@ -1865,6 +1871,9 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 	}
 	defer nsLock.Unlock(lkctx.Cancel)
 
+	healthCtx, healthCancel := context.WithTimeout(lkctx.Context(), deadline)
+	defer healthCancel()
+
 	hostAnonymizer := createHostAnonymizer()
 	// anonAddr - Anonymizes hosts in given input string.
 	anonAddr := func(addr string) string {
@@ -1883,20 +1892,27 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 		info.SetAddr(anonAddr(info.GetAddr()))
 	}
 
+	partialWrite := func(oinfo madmin.HealthInfo) {
+		select {
+		case healthInfoCh <- oinfo:
+		case <-healthCtx.Done():
+		}
+	}
+
 	getAndWritePlatformInfo := func() {
 		if IsKubernetes() {
-			healthInfo.Sys.KubernetesInfo = getKubernetesInfo(deadlinedCtx)
+			healthInfo.Sys.KubernetesInfo = getKubernetesInfo(healthCtx)
 			partialWrite(healthInfo)
 		}
 	}
 
 	getAndWriteCPUs := func() {
 		if query.Get("syscpu") == "true" {
-			localCPUInfo := madmin.GetCPUs(deadlinedCtx, globalLocalNodeName)
+			localCPUInfo := madmin.GetCPUs(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localCPUInfo)
 			healthInfo.Sys.CPUInfo = append(healthInfo.Sys.CPUInfo, localCPUInfo)
 
-			peerCPUInfo := globalNotificationSys.GetCPUs(deadlinedCtx)
+			peerCPUInfo := globalNotificationSys.GetCPUs(healthCtx)
 			for _, cpuInfo := range peerCPUInfo {
 				anonymizeAddr(&cpuInfo)
 				healthInfo.Sys.CPUInfo = append(healthInfo.Sys.CPUInfo, cpuInfo)
@@ -1908,11 +1924,11 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWritePartitions := func() {
 		if query.Get("sysdrivehw") == "true" {
-			localPartitions := madmin.GetPartitions(deadlinedCtx, globalLocalNodeName)
+			localPartitions := madmin.GetPartitions(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localPartitions)
 			healthInfo.Sys.Partitions = append(healthInfo.Sys.Partitions, localPartitions)
 
-			peerPartitions := globalNotificationSys.GetPartitions(deadlinedCtx)
+			peerPartitions := globalNotificationSys.GetPartitions(healthCtx)
 			for _, p := range peerPartitions {
 				anonymizeAddr(&p)
 				healthInfo.Sys.Partitions = append(healthInfo.Sys.Partitions, p)
@@ -1923,11 +1939,11 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWriteOSInfo := func() {
 		if query.Get("sysosinfo") == "true" {
-			localOSInfo := madmin.GetOSInfo(deadlinedCtx, globalLocalNodeName)
+			localOSInfo := madmin.GetOSInfo(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localOSInfo)
 			healthInfo.Sys.OSInfo = append(healthInfo.Sys.OSInfo, localOSInfo)
 
-			peerOSInfos := globalNotificationSys.GetOSInfo(deadlinedCtx)
+			peerOSInfos := globalNotificationSys.GetOSInfo(healthCtx)
 			for _, o := range peerOSInfos {
 				anonymizeAddr(&o)
 				healthInfo.Sys.OSInfo = append(healthInfo.Sys.OSInfo, o)
@@ -1938,11 +1954,11 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWriteMemInfo := func() {
 		if query.Get("sysmem") == "true" {
-			localMemInfo := madmin.GetMemInfo(deadlinedCtx, globalLocalNodeName)
+			localMemInfo := madmin.GetMemInfo(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localMemInfo)
 			healthInfo.Sys.MemInfo = append(healthInfo.Sys.MemInfo, localMemInfo)
 
-			peerMemInfos := globalNotificationSys.GetMemInfo(deadlinedCtx)
+			peerMemInfos := globalNotificationSys.GetMemInfo(healthCtx)
 			for _, m := range peerMemInfos {
 				anonymizeAddr(&m)
 				healthInfo.Sys.MemInfo = append(healthInfo.Sys.MemInfo, m)
@@ -1953,12 +1969,12 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWriteSysErrors := func() {
 		if query.Get(string(madmin.HealthDataTypeSysErrors)) == "true" {
-			localSysErrors := madmin.GetSysErrors(deadlinedCtx, globalLocalNodeName)
+			localSysErrors := madmin.GetSysErrors(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localSysErrors)
 			healthInfo.Sys.SysErrs = append(healthInfo.Sys.SysErrs, localSysErrors)
 			partialWrite(healthInfo)
 
-			peerSysErrs := globalNotificationSys.GetSysErrors(deadlinedCtx)
+			peerSysErrs := globalNotificationSys.GetSysErrors(healthCtx)
 			for _, se := range peerSysErrs {
 				anonymizeAddr(&se)
 				healthInfo.Sys.SysErrs = append(healthInfo.Sys.SysErrs, se)
@@ -1969,12 +1985,12 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWriteSysConfig := func() {
 		if query.Get(string(madmin.HealthDataTypeSysConfig)) == "true" {
-			localSysConfig := madmin.GetSysConfig(deadlinedCtx, globalLocalNodeName)
+			localSysConfig := madmin.GetSysConfig(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localSysConfig)
 			healthInfo.Sys.SysConfig = append(healthInfo.Sys.SysConfig, localSysConfig)
 			partialWrite(healthInfo)
 
-			peerSysConfig := globalNotificationSys.GetSysConfig(deadlinedCtx)
+			peerSysConfig := globalNotificationSys.GetSysConfig(healthCtx)
 			for _, sc := range peerSysConfig {
 				anonymizeAddr(&sc)
 				healthInfo.Sys.SysConfig = append(healthInfo.Sys.SysConfig, sc)
@@ -1985,12 +2001,12 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWriteSysServices := func() {
 		if query.Get(string(madmin.HealthDataTypeSysServices)) == "true" {
-			localSysServices := madmin.GetSysServices(deadlinedCtx, globalLocalNodeName)
+			localSysServices := madmin.GetSysServices(healthCtx, globalLocalNodeName)
 			anonymizeAddr(&localSysServices)
 			healthInfo.Sys.SysServices = append(healthInfo.Sys.SysServices, localSysServices)
 			partialWrite(healthInfo)
 
-			peerSysServices := globalNotificationSys.GetSysServices(deadlinedCtx)
+			peerSysServices := globalNotificationSys.GetSysServices(healthCtx)
 			for _, ss := range peerSysServices {
 				anonymizeAddr(&ss)
 				healthInfo.Sys.SysServices = append(healthInfo.Sys.SysServices, ss)
@@ -2065,10 +2081,10 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 
 	getAndWriteProcInfo := func() {
 		if query.Get("sysprocess") == "true" {
-			localProcInfo := madmin.GetProcInfo(deadlinedCtx, globalLocalNodeName)
+			localProcInfo := madmin.GetProcInfo(healthCtx, globalLocalNodeName)
 			anonymizeProcInfo(&localProcInfo)
 			healthInfo.Sys.ProcInfo = append(healthInfo.Sys.ProcInfo, localProcInfo)
-			peerProcInfos := globalNotificationSys.GetProcInfo(deadlinedCtx)
+			peerProcInfos := globalNotificationSys.GetProcInfo(healthCtx)
 			for _, p := range peerProcInfos {
 				anonymizeProcInfo(&p)
 				healthInfo.Sys.ProcInfo = append(healthInfo.Sys.ProcInfo, p)
@@ -2294,7 +2310,7 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 				return
 			}
 			w.(http.Flusher).Flush()
-		case <-deadlinedCtx.Done():
+		case <-healthCtx.Done():
 			return
 		}
 	}

cmd/admin-server-info.go

@@ -31,7 +31,10 @@ import (
 // local endpoints from given list of endpoints
 func getLocalServerProperty(endpointServerPools EndpointServerPools, r *http.Request) madmin.ServerProperties {
 	var localEndpoints Endpoints
-	addr := r.Host
+	addr := globalLocalNodeName
+	if r != nil {
+		addr = r.Host
+	}
 	if globalIsDistErasure {
 		addr = globalLocalNodeName
 	}
@@ -40,7 +43,7 @@ func getLocalServerProperty(endpointServerPools EndpointServerPools, r *http.Req
 		for _, endpoint := range ep.Endpoints {
 			nodeName := endpoint.Host
 			if nodeName == "" {
-				nodeName = r.Host
+				nodeName = addr
 			}
 			if endpoint.IsLocal {
 				// Only proceed for local endpoints

cmd/api-errors.go

@@ -131,7 +131,7 @@ const (
 	ErrReplicationNeedsVersioningError
 	ErrReplicationBucketNeedsVersioningError
 	ErrReplicationDenyEditError
-	ErrReplicationNoMatchingRuleError
+	ErrReplicationNoExistingObjects
 	ErrObjectRestoreAlreadyInProgress
 	ErrNoSuchKey
 	ErrNoSuchUpload
@@ -893,9 +893,9 @@ var errorCodes = errorCodeMap{
 		Description:    "Bandwidth limit for remote target must be atleast 100MBps",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
-	ErrReplicationNoMatchingRuleError: {
-		Code:           "XMinioReplicationNoMatchingRule",
-		Description:    "No matching replication rule found for this object prefix",
+	ErrReplicationNoExistingObjects: {
+		Code:           "XMinioReplicationNoExistingObjects",
+		Description:    "No matching ExistingsObjects rule enabled",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
 	ErrReplicationDenyEditError: {

cmd/api-router.go

@@ -342,7 +342,7 @@ func registerAPIRouter(router *mux.Router) {
 			collectAPIStats("getbucketnotification", maxClients(gz(httpTraceAll(api.GetBucketNotificationHandler))))).Queries("notification", "")
 		// ListenNotification
 		router.Methods(http.MethodGet).HandlerFunc(
-			collectAPIStats("listennotification", maxClients(gz(httpTraceAll(api.ListenNotificationHandler))))).Queries("events", "{events:.*}")
+			collectAPIStats("listennotification", gz(httpTraceAll(api.ListenNotificationHandler)))).Queries("events", "{events:.*}")
 		// ResetBucketReplicationStatus - MinIO extension API
 		router.Methods(http.MethodGet).HandlerFunc(
 			collectAPIStats("resetbucketreplicationstatus", maxClients(gz(httpTraceAll(api.ResetBucketReplicationStatusHandler))))).Queries("replication-reset-status", "")
@@ -474,7 +474,7 @@ func registerAPIRouter(router *mux.Router) {
 
 	// ListenNotification
 	apiRouter.Methods(http.MethodGet).Path(SlashSeparator).HandlerFunc(
-		collectAPIStats("listennotification", maxClients(gz(httpTraceAll(api.ListenNotificationHandler))))).Queries("events", "{events:.*}")
+		collectAPIStats("listennotification", gz(httpTraceAll(api.ListenNotificationHandler)))).Queries("events", "{events:.*}")
 
 	// ListBuckets
 	apiRouter.Methods(http.MethodGet).Path(SlashSeparator).HandlerFunc(

cmd/bucket-handlers.go

@@ -21,7 +21,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/base64"
-	"encoding/json"
 	"encoding/xml"
 	"fmt"
 	"io"
@@ -33,7 +32,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/gorilla/mux"
@@ -1596,372 +1594,3 @@ func (api objectAPIHandlers) DeleteBucketTaggingHandler(w http.ResponseWriter, r
 	// Write success response.
 	writeSuccessResponseHeadersOnly(w)
 }
-
-// PutBucketReplicationConfigHandler - PUT Bucket replication configuration.
-// ----------
-// Add a replication configuration on the specified bucket as specified in https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketReplication.html
-func (api objectAPIHandlers) PutBucketReplicationConfigHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "PutBucketReplicationConfig")
-	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
-
-	vars := mux.Vars(r)
-	bucket := vars["bucket"]
-	objectAPI := api.ObjectAPI()
-	if objectAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
-		return
-	}
-	if globalIsGateway {
-		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
-		return
-	}
-	if s3Error := checkRequestAuthType(ctx, r, policy.PutReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
-		return
-	}
-	// Check if bucket exists.
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	if versioned := globalBucketVersioningSys.Enabled(bucket); !versioned {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationNeedsVersioningError), r.URL)
-		return
-	}
-	replicationConfig, err := replication.ParseConfig(io.LimitReader(r.Body, r.ContentLength))
-	if err != nil {
-		apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
-		apiErr.Description = err.Error()
-		writeErrorResponse(ctx, w, apiErr, r.URL)
-		return
-	}
-	sameTarget, apiErr := validateReplicationDestination(ctx, bucket, replicationConfig, true)
-	if apiErr != noError {
-		writeErrorResponse(ctx, w, apiErr, r.URL)
-		return
-	}
-	// Validate the received bucket replication config
-	if err = replicationConfig.Validate(bucket, sameTarget); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-	configData, err := xml.Marshal(replicationConfig)
-	if err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-	if err = globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, configData); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	// Write success response.
-	writeSuccessResponseHeadersOnly(w)
-}
-
-// GetBucketReplicationConfigHandler - GET Bucket replication configuration.
-// ----------
-// Gets the replication configuration for a bucket.
-func (api objectAPIHandlers) GetBucketReplicationConfigHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "GetBucketReplicationConfig")
-
-	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
-
-	vars := mux.Vars(r)
-	bucket := vars["bucket"]
-
-	objectAPI := api.ObjectAPI()
-	if objectAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
-		return
-	}
-
-	// check if user has permissions to perform this operation
-	if s3Error := checkRequestAuthType(ctx, r, policy.GetReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
-		return
-	}
-	// Check if bucket exists.
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	config, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket)
-	if err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-	configData, err := xml.Marshal(config)
-	if err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	// Write success response.
-	writeSuccessResponseXML(w, configData)
-}
-
-// DeleteBucketReplicationConfigHandler - DELETE Bucket replication config.
-// ----------
-func (api objectAPIHandlers) DeleteBucketReplicationConfigHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "DeleteBucketReplicationConfig")
-	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
-	vars := mux.Vars(r)
-	bucket := vars["bucket"]
-
-	objectAPI := api.ObjectAPI()
-	if objectAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
-		return
-	}
-
-	if s3Error := checkRequestAuthType(ctx, r, policy.PutReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
-		return
-	}
-	// Check if bucket exists.
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-	if globalSiteReplicationSys.isEnabled() {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationDenyEditError), r.URL)
-		return
-	}
-	if err := globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, nil); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	// Write success response.
-	writeSuccessResponseHeadersOnly(w)
-}
-
-// GetBucketReplicationMetricsHandler - GET Bucket replication metrics.
-// ----------
-// Gets the replication metrics for a bucket.
-func (api objectAPIHandlers) GetBucketReplicationMetricsHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "GetBucketReplicationMetrics")
-
-	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
-
-	vars := mux.Vars(r)
-	bucket := vars["bucket"]
-
-	objectAPI := api.ObjectAPI()
-	if objectAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
-		return
-	}
-
-	// check if user has permissions to perform this operation
-	if s3Error := checkRequestAuthType(ctx, r, policy.GetReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
-		return
-	}
-
-	// Check if bucket exists.
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-	var usageInfo BucketUsageInfo
-	dataUsageInfo, err := loadDataUsageFromBackend(ctx, objectAPI)
-	if err == nil && !dataUsageInfo.LastUpdate.IsZero() {
-		usageInfo = dataUsageInfo.BucketsUsage[bucket]
-	}
-
-	w.Header().Set(xhttp.ContentType, string(mimeJSON))
-
-	enc := json.NewEncoder(w)
-	if err = enc.Encode(getLatestReplicationStats(bucket, usageInfo)); err != nil {
-		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
-		return
-	}
-}
-
-// ResetBucketReplicationStartHandler - starts a replication reset for all objects in a bucket which
-// qualify for replication and re-sync the object(s) to target, provided ExistingObjectReplication is
-// enabled for the qualifying rule. This API is a MinIO only extension provided for situations where
-// remote target is entirely lost,and previously replicated objects need to be re-synced. If resync is
-// already in progress it returns an error
-func (api objectAPIHandlers) ResetBucketReplicationStartHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "ResetBucketReplicationStart")
-	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
-
-	vars := mux.Vars(r)
-	bucket := vars["bucket"]
-	durationStr := r.URL.Query().Get("older-than")
-	arn := r.URL.Query().Get("arn")
-	resetID := r.URL.Query().Get("reset-id")
-	if resetID == "" {
-		resetID = mustGetUUID()
-	}
-	var (
-		days time.Duration
-		err  error
-	)
-	if durationStr != "" {
-		days, err = time.ParseDuration(durationStr)
-		if err != nil {
-			writeErrorResponse(ctx, w, toAPIError(ctx, InvalidArgument{
-				Bucket: bucket,
-				Err:    fmt.Errorf("invalid query parameter older-than %s for %s : %w", durationStr, bucket, err),
-			}), r.URL)
-		}
-	}
-	resetBeforeDate := UTCNow().AddDate(0, 0, -1*int(days/24))
-
-	objectAPI := api.ObjectAPI()
-	if objectAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
-		return
-	}
-
-	if s3Error := checkRequestAuthType(ctx, r, policy.ResetBucketReplicationStateAction, bucket, ""); s3Error != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
-		return
-	}
-
-	// Check if bucket exists.
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	config, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket)
-	if err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-	if !config.HasActiveRules("", true) {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationNoMatchingRuleError), r.URL)
-		return
-	}
-	tgtArns := config.FilterTargetArns(
-		replication.ObjectOpts{
-			OpType:    replication.ResyncReplicationType,
-			TargetArn: arn,
-		})
-
-	if len(tgtArns) == 0 {
-		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
-			Bucket: bucket,
-			Err:    fmt.Errorf("Remote target ARN %s missing or ineligible for replication resync", arn),
-		}), r.URL)
-		return
-	}
-
-	if len(tgtArns) > 1 && arn == "" {
-		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
-			Bucket: bucket,
-			Err:    fmt.Errorf("ARN should be specified for replication reset"),
-		}), r.URL)
-		return
-	}
-	var rinfo ResyncTargetsInfo
-	target := globalBucketTargetSys.GetRemoteBucketTargetByArn(ctx, bucket, tgtArns[0])
-	target.ResetBeforeDate = UTCNow().AddDate(0, 0, -1*int(days/24))
-	target.ResetID = resetID
-	rinfo.Targets = append(rinfo.Targets, ResyncTarget{Arn: tgtArns[0], ResetID: target.ResetID})
-	if err = globalBucketTargetSys.SetTarget(ctx, bucket, &target, true); err != nil {
-		switch err.(type) {
-		case BucketRemoteConnectionErr:
-			writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrReplicationRemoteConnectionError, err), r.URL)
-		default:
-			writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
-		}
-	}
-	if err := startReplicationResync(ctx, bucket, arn, resetID, resetBeforeDate, objectAPI); err != nil {
-		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
-			Bucket: bucket,
-			Err:    err,
-		}), r.URL)
-		return
-	}
-
-	data, err := json.Marshal(rinfo)
-	if err != nil {
-		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
-		return
-	}
-	// Write success response.
-	writeSuccessResponseJSON(w, data)
-}
-
-// ResetBucketReplicationStatusHandler - returns the status of replication reset.
-// This API is a MinIO only extension
-func (api objectAPIHandlers) ResetBucketReplicationStatusHandler(w http.ResponseWriter, r *http.Request) {
-	ctx := newContext(r, w, "ResetBucketReplicationStatus")
-	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
-
-	vars := mux.Vars(r)
-	bucket := vars["bucket"]
-	arn := r.URL.Query().Get("arn")
-	var err error
-
-	objectAPI := api.ObjectAPI()
-	if objectAPI == nil {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
-		return
-	}
-
-	if s3Error := checkRequestAuthType(ctx, r, policy.ResetBucketReplicationStateAction, bucket, ""); s3Error != ErrNone {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
-		return
-	}
-
-	// Check if bucket exists.
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	if _, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
-	globalReplicationPool.resyncState.RLock()
-	brs, ok := globalReplicationPool.resyncState.statusMap[bucket]
-	if !ok {
-		brs, err = loadBucketResyncMetadata(ctx, bucket, objectAPI)
-		if err != nil {
-			writeErrorResponse(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
-				Bucket: bucket,
-				Err:    fmt.Errorf("No replication resync status available for %s", arn),
-			}), r.URL)
-		}
-		return
-	}
-
-	var rinfo ResyncTargetsInfo
-	for tarn, st := range brs.TargetsMap {
-		if arn != "" && tarn != arn {
-			continue
-		}
-		rinfo.Targets = append(rinfo.Targets, ResyncTarget{
-			Arn:             tarn,
-			ResetID:         st.ResyncID,
-			StartTime:       st.StartTime,
-			EndTime:         st.EndTime,
-			ResyncStatus:    st.ResyncStatus.String(),
-			ReplicatedSize:  st.ReplicatedSize,
-			ReplicatedCount: st.ReplicatedCount,
-			FailedSize:      st.FailedSize,
-			FailedCount:     st.FailedCount,
-			Bucket:          st.Bucket,
-			Object:          st.Object,
-		})
-	}
-	globalReplicationPool.resyncState.RUnlock()
-	data, err := json.Marshal(rinfo)
-	if err != nil {
-		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
-		return
-	}
-	// Write success response.
-	writeSuccessResponseJSON(w, data)
-}

cmd/bucket-replication-handlers.go

@@ -0,0 +1,404 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"encoding/json"
+	"encoding/xml"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/minio/minio/internal/bucket/replication"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/minio/internal/logger"
+	"github.com/minio/pkg/bucket/policy"
+)
+
+// PutBucketReplicationConfigHandler - PUT Bucket replication configuration.
+// ----------
+// Add a replication configuration on the specified bucket as specified in https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutBucketReplication.html
+func (api objectAPIHandlers) PutBucketReplicationConfigHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "PutBucketReplicationConfig")
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return
+	}
+	if globalIsGateway {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
+		return
+	}
+	if s3Error := checkRequestAuthType(ctx, r, policy.PutReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+	// Check if bucket exists.
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	if versioned := globalBucketVersioningSys.Enabled(bucket); !versioned {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationNeedsVersioningError), r.URL)
+		return
+	}
+	replicationConfig, err := replication.ParseConfig(io.LimitReader(r.Body, r.ContentLength))
+	if err != nil {
+		apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
+		apiErr.Description = err.Error()
+		writeErrorResponse(ctx, w, apiErr, r.URL)
+		return
+	}
+	sameTarget, apiErr := validateReplicationDestination(ctx, bucket, replicationConfig, true)
+	if apiErr != noError {
+		writeErrorResponse(ctx, w, apiErr, r.URL)
+		return
+	}
+	// Validate the received bucket replication config
+	if err = replicationConfig.Validate(bucket, sameTarget); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	configData, err := xml.Marshal(replicationConfig)
+	if err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	if err = globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, configData); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	// Write success response.
+	writeSuccessResponseHeadersOnly(w)
+}
+
+// GetBucketReplicationConfigHandler - GET Bucket replication configuration.
+// ----------
+// Gets the replication configuration for a bucket.
+func (api objectAPIHandlers) GetBucketReplicationConfigHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "GetBucketReplicationConfig")
+
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return
+	}
+
+	// check if user has permissions to perform this operation
+	if s3Error := checkRequestAuthType(ctx, r, policy.GetReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+	// Check if bucket exists.
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	config, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket)
+	if err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	configData, err := xml.Marshal(config)
+	if err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	// Write success response.
+	writeSuccessResponseXML(w, configData)
+}
+
+// DeleteBucketReplicationConfigHandler - DELETE Bucket replication config.
+// ----------
+func (api objectAPIHandlers) DeleteBucketReplicationConfigHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "DeleteBucketReplicationConfig")
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return
+	}
+
+	if s3Error := checkRequestAuthType(ctx, r, policy.PutReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+	// Check if bucket exists.
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	if globalSiteReplicationSys.isEnabled() {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationDenyEditError), r.URL)
+		return
+	}
+	if err := globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, nil); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	// Write success response.
+	writeSuccessResponseHeadersOnly(w)
+}
+
+// GetBucketReplicationMetricsHandler - GET Bucket replication metrics.
+// ----------
+// Gets the replication metrics for a bucket.
+func (api objectAPIHandlers) GetBucketReplicationMetricsHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "GetBucketReplicationMetrics")
+
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return
+	}
+
+	// check if user has permissions to perform this operation
+	if s3Error := checkRequestAuthType(ctx, r, policy.GetReplicationConfigurationAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+
+	// Check if bucket exists.
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	var usageInfo BucketUsageInfo
+	dataUsageInfo, err := loadDataUsageFromBackend(ctx, objectAPI)
+	if err == nil && !dataUsageInfo.LastUpdate.IsZero() {
+		usageInfo = dataUsageInfo.BucketsUsage[bucket]
+	}
+
+	w.Header().Set(xhttp.ContentType, string(mimeJSON))
+
+	enc := json.NewEncoder(w)
+	if err = enc.Encode(getLatestReplicationStats(bucket, usageInfo)); err != nil {
+		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
+		return
+	}
+}
+
+// ResetBucketReplicationStartHandler - starts a replication reset for all objects in a bucket which
+// qualify for replication and re-sync the object(s) to target, provided ExistingObjectReplication is
+// enabled for the qualifying rule. This API is a MinIO only extension provided for situations where
+// remote target is entirely lost,and previously replicated objects need to be re-synced. If resync is
+// already in progress it returns an error
+func (api objectAPIHandlers) ResetBucketReplicationStartHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "ResetBucketReplicationStart")
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+	durationStr := r.URL.Query().Get("older-than")
+	arn := r.URL.Query().Get("arn")
+	resetID := r.URL.Query().Get("reset-id")
+	if resetID == "" {
+		resetID = mustGetUUID()
+	}
+	var (
+		days time.Duration
+		err  error
+	)
+	if durationStr != "" {
+		days, err = time.ParseDuration(durationStr)
+		if err != nil {
+			writeErrorResponse(ctx, w, toAPIError(ctx, InvalidArgument{
+				Bucket: bucket,
+				Err:    fmt.Errorf("invalid query parameter older-than %s for %s : %w", durationStr, bucket, err),
+			}), r.URL)
+		}
+	}
+	resetBeforeDate := UTCNow().AddDate(0, 0, -1*int(days/24))
+
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return
+	}
+
+	if s3Error := checkRequestAuthType(ctx, r, policy.ResetBucketReplicationStateAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+
+	// Check if bucket exists.
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	config, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket)
+	if err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+	if !config.HasExistingObjectReplication(arn) {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationNoExistingObjects), r.URL)
+		return
+	}
+
+	tgtArns := config.FilterTargetArns(
+		replication.ObjectOpts{
+			OpType:    replication.ResyncReplicationType,
+			TargetArn: arn,
+		})
+
+	if len(tgtArns) == 0 {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
+			Bucket: bucket,
+			Err:    fmt.Errorf("Remote target ARN %s missing or ineligible for replication resync", arn),
+		}), r.URL)
+		return
+	}
+
+	if len(tgtArns) > 1 && arn == "" {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
+			Bucket: bucket,
+			Err:    fmt.Errorf("ARN should be specified for replication reset"),
+		}), r.URL)
+		return
+	}
+	var rinfo ResyncTargetsInfo
+	target := globalBucketTargetSys.GetRemoteBucketTargetByArn(ctx, bucket, tgtArns[0])
+	target.ResetBeforeDate = UTCNow().AddDate(0, 0, -1*int(days/24))
+	target.ResetID = resetID
+	rinfo.Targets = append(rinfo.Targets, ResyncTarget{Arn: tgtArns[0], ResetID: target.ResetID})
+	if err = globalBucketTargetSys.SetTarget(ctx, bucket, &target, true); err != nil {
+		switch err.(type) {
+		case BucketRemoteConnectionErr:
+			writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrReplicationRemoteConnectionError, err), r.URL)
+		default:
+			writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
+		}
+	}
+	if err := startReplicationResync(ctx, bucket, arn, resetID, resetBeforeDate, objectAPI); err != nil {
+		writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
+			Bucket: bucket,
+			Err:    err,
+		}), r.URL)
+		return
+	}
+
+	data, err := json.Marshal(rinfo)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
+		return
+	}
+	// Write success response.
+	writeSuccessResponseJSON(w, data)
+}
+
+// ResetBucketReplicationStatusHandler - returns the status of replication reset.
+// This API is a MinIO only extension
+func (api objectAPIHandlers) ResetBucketReplicationStatusHandler(w http.ResponseWriter, r *http.Request) {
+	ctx := newContext(r, w, "ResetBucketReplicationStatus")
+	defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
+
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+	arn := r.URL.Query().Get("arn")
+	var err error
+
+	objectAPI := api.ObjectAPI()
+	if objectAPI == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
+		return
+	}
+
+	if s3Error := checkRequestAuthType(ctx, r, policy.ResetBucketReplicationStateAction, bucket, ""); s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
+
+	// Check if bucket exists.
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	if _, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	globalReplicationPool.resyncState.RLock()
+	brs, ok := globalReplicationPool.resyncState.statusMap[bucket]
+	globalReplicationPool.resyncState.RUnlock()
+	if !ok {
+		brs, err = loadBucketResyncMetadata(ctx, bucket, objectAPI)
+		if err != nil {
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErrWithErr(ErrBadRequest, InvalidArgument{
+				Bucket: bucket,
+				Err:    fmt.Errorf("No replication resync status available for %s", arn),
+			}), r.URL)
+			return
+		}
+	}
+
+	var rinfo ResyncTargetsInfo
+	for tarn, st := range brs.TargetsMap {
+		if arn != "" && tarn != arn {
+			continue
+		}
+		rinfo.Targets = append(rinfo.Targets, ResyncTarget{
+			Arn:             tarn,
+			ResetID:         st.ResyncID,
+			StartTime:       st.StartTime,
+			EndTime:         st.EndTime,
+			ResyncStatus:    st.ResyncStatus.String(),
+			ReplicatedSize:  st.ReplicatedSize,
+			ReplicatedCount: st.ReplicatedCount,
+			FailedSize:      st.FailedSize,
+			FailedCount:     st.FailedCount,
+			Bucket:          st.Bucket,
+			Object:          st.Object,
+		})
+	}
+	data, err := json.Marshal(rinfo)
+	if err != nil {
+		writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
+		return
+	}
+
+	// Write success response.
+	writeSuccessResponseJSON(w, data)
+}

cmd/bucket-replication.go

@@ -1045,8 +1045,13 @@ func replicateObjectToTarget(ctx context.Context, ri ReplicateObjectInfo, object
 		return
 	}
 
+	versioned := globalBucketVersioningSys.PrefixEnabled(bucket, object)
+	versionSuspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
+
 	gr, err = objectAPI.GetObjectNInfo(ctx, bucket, object, nil, http.Header{}, readLock, ObjectOptions{
-		VersionID: objInfo.VersionID,
+		VersionID:        objInfo.VersionID,
+		Versioned:        versioned,
+		VersionSuspended: versionSuspended,
 	})
 	if err != nil {
 		sendEvent(eventArgs{
@@ -1757,10 +1762,6 @@ func (c replicationConfig) Resync(ctx context.Context, oi ObjectInfo, dsc *Repli
 	if c.Empty() {
 		return
 	}
-	// existing object replication does not apply to un-versioned objects
-	if oi.VersionID == "" {
-		return
-	}
 
 	// Now overlay existing object replication choices for target
 	if oi.DeleteMarker {
@@ -2006,6 +2007,7 @@ func resyncBucket(ctx context.Context, bucket, arn string, heal bool, objectAPI
 		st.EndTime = UTCNow()
 		st.ResyncStatus = resyncStatus
 		m.TargetsMap[arn] = st
+		globalReplicationPool.resyncState.statusMap[bucket] = m
 		globalReplicationPool.resyncState.Unlock()
 	}()
 	// Allocate new results channel to receive ObjectInfo.
@@ -2066,7 +2068,6 @@ func resyncBucket(ctx context.Context, bucket, arn string, heal bool, objectAPI
 		}
 
 		if roi.DeleteMarker || !roi.VersionPurgeStatus.Empty() {
-
 			versionID := ""
 			dmVersionID := ""
 			if roi.VersionPurgeStatus.Empty() {
@@ -2113,6 +2114,7 @@ func resyncBucket(ctx context.Context, bucket, arn string, heal bool, objectAPI
 			st.ReplicatedSize += roi.Size
 		}
 		m.TargetsMap[arn] = st
+		globalReplicationPool.resyncState.statusMap[bucket] = m
 		globalReplicationPool.resyncState.Unlock()
 	}
 	resyncStatus = ResyncCompleted

cmd/callhome.go

@@ -0,0 +1,136 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"time"
+
+	"github.com/minio/madmin-go"
+	"github.com/minio/minio/internal/logger"
+	uatomic "go.uber.org/atomic"
+)
+
+const (
+	// callhomeSchemaVersion1 is callhome schema version 1
+	callhomeSchemaVersion1 = "1"
+
+	// callhomeSchemaVersion is current callhome schema version.
+	callhomeSchemaVersion = callhomeSchemaVersion1
+
+	// callhomeCycleDefault is the default interval between two callhome cycles (24hrs)
+	callhomeCycleDefault = 24 * time.Hour
+)
+
+// CallhomeInfo - Contains callhome information
+type CallhomeInfo struct {
+	SchemaVersion string             `json:"schema_version"`
+	AdminInfo     madmin.InfoMessage `json:"admin_info"`
+}
+
+var (
+	enableCallhome            = uatomic.NewBool(false)
+	callhomeLeaderLockTimeout = newDynamicTimeout(30*time.Second, 10*time.Second)
+	callhomeFreq              = uatomic.NewDuration(callhomeCycleDefault)
+)
+
+func updateCallhomeParams(ctx context.Context, objAPI ObjectLayer) {
+	alreadyEnabled := enableCallhome.Load()
+	enableCallhome.Store(globalCallhomeConfig.Enable)
+	callhomeFreq.Store(globalCallhomeConfig.Frequency)
+
+	// If callhome was disabled earlier and has now been enabled,
+	// initialize the callhome process again.
+	if !alreadyEnabled && enableCallhome.Load() {
+		initCallhome(ctx, objAPI)
+	}
+}
+
+// initCallhome will start the callhome task in the background.
+func initCallhome(ctx context.Context, objAPI ObjectLayer) {
+	go func() {
+		r := rand.New(rand.NewSource(time.Now().UnixNano()))
+		// Leader node (that successfully acquires the lock inside runCallhome)
+		// will keep performing the callhome. If the leader goes down for some reason,
+		// the lock will be released and another node will acquire it and take over
+		// because of this loop.
+		for {
+			runCallhome(ctx, objAPI)
+			if !enableCallhome.Load() {
+				return
+			}
+
+			// callhome running on a different node.
+			// sleep for some time and try again.
+			duration := time.Duration(r.Float64() * float64(callhomeFreq.Load()))
+			if duration < time.Second {
+				// Make sure to sleep atleast a second to avoid high CPU ticks.
+				duration = time.Second
+			}
+			time.Sleep(duration)
+
+			if !enableCallhome.Load() {
+				return
+			}
+		}
+	}()
+}
+
+func runCallhome(ctx context.Context, objAPI ObjectLayer) {
+	// Make sure only 1 callhome is running on the cluster.
+	locker := objAPI.NewNSLock(minioMetaBucket, "callhome/runCallhome.lock")
+	lkctx, err := locker.GetLock(ctx, callhomeLeaderLockTimeout)
+	if err != nil {
+		return
+	}
+
+	ctx = lkctx.Context()
+	defer locker.Unlock(lkctx.Cancel)
+
+	callhomeTimer := time.NewTimer(callhomeFreq.Load())
+	defer callhomeTimer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-callhomeTimer.C:
+			if !enableCallhome.Load() {
+				// Stop the processing as callhome got disabled
+				return
+			}
+			performCallhome(ctx)
+
+			// Reset the timer for next cycle.
+			callhomeTimer.Reset(callhomeFreq.Load())
+		}
+	}
+}
+
+func performCallhome(ctx context.Context) {
+	err := sendCallhomeInfo(
+		CallhomeInfo{
+			SchemaVersion: callhomeSchemaVersion,
+			AdminInfo:     getServerInfo(ctx, nil),
+		})
+	if err != nil {
+		logger.LogIf(ctx, fmt.Errorf("Unable to perform callhome: %w", err))
+	}
+}

cmd/common-main.go

@@ -230,8 +230,8 @@ func minioConfigToConsoleFeatures() {
 	if globalSubnetConfig.APIKey != "" {
 		os.Setenv("CONSOLE_SUBNET_API_KEY", globalSubnetConfig.APIKey)
 	}
-	if globalSubnetConfig.Proxy != "" {
-		os.Setenv("CONSOLE_SUBNET_PROXY", globalSubnetConfig.Proxy)
+	if globalSubnetConfig.ProxyURL != nil {
+		os.Setenv("CONSOLE_SUBNET_PROXY", globalSubnetConfig.ProxyURL.String())
 	}
 }
 
@@ -526,9 +526,6 @@ func parsEnvEntry(envEntry string) (envKV, error) {
 func minioEnvironFromFile(envConfigFile string) ([]envKV, error) {
 	f, err := os.Open(envConfigFile)
 	if err != nil {
-		if os.IsNotExist(err) { // ignore if file doesn't exist.
-			return nil, nil
-		}
 		return nil, err
 	}
 	defer f.Close()
@@ -624,7 +621,7 @@ func loadEnvVarsFromFiles() {
 
 	if env.IsSet(config.EnvConfigEnvFile) {
 		ekvs, err := minioEnvironFromFile(env.Get(config.EnvConfigEnvFile, ""))
-		if err != nil {
+		if err != nil && !os.IsNotExist(err) {
 			logger.Fatal(err, "Unable to read the config environment file")
 		}
 		for _, ekv := range ekvs {

cmd/config-current.go

@@ -28,6 +28,7 @@ import (
 	"github.com/minio/minio/internal/config"
 	"github.com/minio/minio/internal/config/api"
 	"github.com/minio/minio/internal/config/cache"
+	"github.com/minio/minio/internal/config/callhome"
 	"github.com/minio/minio/internal/config/compress"
 	"github.com/minio/minio/internal/config/dns"
 	"github.com/minio/minio/internal/config/etcd"
@@ -70,6 +71,7 @@ func initHelp() {
 		config.HealSubSys:           heal.DefaultKVS,
 		config.ScannerSubSys:        scanner.DefaultKVS,
 		config.SubnetSubSys:         subnet.DefaultKVS,
+		config.CallhomeSubSys:       callhome.DefaultKVS,
 	}
 	for k, v := range notify.DefaultNotificationKVS {
 		kvs[k] = v
@@ -201,6 +203,12 @@ func initHelp() {
 			Description: "set subnet config for the cluster e.g. api key",
 			Optional:    true,
 		},
+		config.HelpKV{
+			Key:         config.CallhomeSubSys,
+			Type:        "string",
+			Description: "enable callhome for the cluster",
+			Optional:    true,
+		},
 	}
 
 	if globalIsErasure {
@@ -243,6 +251,7 @@ func initHelp() {
 		config.NotifyWebhookSubSys:  notify.HelpWebhook,
 		config.NotifyESSubSys:       notify.HelpES,
 		config.SubnetSubSys:         subnet.HelpSubnet,
+		config.CallhomeSubSys:       callhome.HelpCallhome,
 	}
 
 	config.RegisterHelpSubSys(helpMap)
@@ -358,6 +367,10 @@ func validateSubSysConfig(s config.Config, subSys string, objAPI ObjectLayer) er
 		if _, err := subnet.LookupConfig(s[config.SubnetSubSys][config.Default]); err != nil {
 			return err
 		}
+	case config.CallhomeSubSys:
+		if _, err := callhome.LookupConfig(s[config.CallhomeSubSys][config.Default]); err != nil {
+			return err
+		}
 	case config.PolicyOPASubSys:
 		// In case legacy OPA config is being set, we treat it as if the
 		// AuthZPlugin is being set.
@@ -649,7 +662,7 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 			return fmt.Errorf("Unable to apply scanner config: %w", err)
 		}
 		// update dynamic scanner values.
-		scannerCycle.Update(scannerCfg.Cycle)
+		scannerCycle.Store(scannerCfg.Cycle)
 		logger.LogIf(ctx, scannerSleeper.Update(scannerCfg.Delay, scannerCfg.MaxWait))
 	case config.LoggerWebhookSubSys:
 		loggerCfg, err := logger.LookupConfigForSubSys(s, config.LoggerWebhookSubSys)
@@ -719,6 +732,14 @@ func applyDynamicConfigForSubSys(ctx context.Context, objAPI ObjectLayer, s conf
 				}
 			}
 		}
+	case config.CallhomeSubSys:
+		callhomeCfg, err := callhome.LookupConfig(s[config.CallhomeSubSys][config.Default])
+		if err != nil {
+			logger.LogIf(ctx, fmt.Errorf("Unable to load callhome config: %w", err))
+		} else {
+			globalCallhomeConfig = callhomeCfg
+			updateCallhomeParams(ctx, objAPI)
+		}
 	}
 	globalServerConfigMu.Lock()
 	defer globalServerConfigMu.Unlock()

cmd/consolelogger.go

@@ -45,7 +45,7 @@ type HTTPConsoleLoggerSys struct {
 // NewConsoleLogger - creates new HTTPConsoleLoggerSys with all nodes subscribed to
 // the console logging pub sub system
 func NewConsoleLogger(ctx context.Context) *HTTPConsoleLoggerSys {
-	ps := pubsub.New()
+	ps := pubsub.New(8)
 	return &HTTPConsoleLoggerSys{
 		pubsub:  ps,
 		console: console.New(),
@@ -75,7 +75,7 @@ func (sys *HTTPConsoleLoggerSys) HasLogListeners() bool {
 }
 
 // Subscribe starts console logging for this node.
-func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan interface{}, doneCh <-chan struct{}, node string, last int, logKind string, filter func(entry interface{}) bool) {
+func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan interface{}, doneCh <-chan struct{}, node string, last int, logKind string, filter func(entry interface{}) bool) error {
 	// Enable console logging for remote client.
 	if !sys.HasLogListeners() {
 		logger.AddSystemTarget(sys)
@@ -111,11 +111,12 @@ func (sys *HTTPConsoleLoggerSys) Subscribe(subCh chan interface{}, doneCh <-chan
 			select {
 			case subCh <- entry:
 			case <-doneCh:
-				return
+				return nil
 			}
 		}
 	}
-	sys.pubsub.Subscribe(subCh, doneCh, filter)
+
+	return sys.pubsub.Subscribe(subCh, doneCh, filter)
 }
 
 // Init if HTTPConsoleLoggerSys is valid, always returns nil right now

cmd/data-scanner.go

@@ -45,6 +45,7 @@ import (
 	"github.com/minio/minio/internal/event"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/console"
+	uatomic "go.uber.org/atomic"
 )
 
 const (
@@ -66,9 +67,7 @@ var (
 	dataScannerLeaderLockTimeout = newDynamicTimeout(30*time.Second, 10*time.Second)
 	// Sleeper values are updated when config is loaded.
 	scannerSleeper = newDynamicSleeper(10, 10*time.Second)
-	scannerCycle   = &safeDuration{
-		t: dataScannerStartDelay,
-	}
+	scannerCycle   = uatomic.NewDuration(dataScannerStartDelay)
 )
 
 // initDataScanner will start the scanner in the background.
@@ -78,7 +77,7 @@ func initDataScanner(ctx context.Context, objAPI ObjectLayer) {
 		// Run the data scanner in a loop
 		for {
 			runDataScanner(ctx, objAPI)
-			duration := time.Duration(r.Float64() * float64(scannerCycle.Get()))
+			duration := time.Duration(r.Float64() * float64(scannerCycle.Load()))
 			if duration < time.Second {
 				// Make sure to sleep atleast a second to avoid high CPU ticks.
 				duration = time.Second
@@ -88,23 +87,6 @@ func initDataScanner(ctx context.Context, objAPI ObjectLayer) {
 	}()
 }
 
-type safeDuration struct {
-	sync.Mutex
-	t time.Duration
-}
-
-func (s *safeDuration) Update(t time.Duration) {
-	s.Lock()
-	defer s.Unlock()
-	s.t = t
-}
-
-func (s *safeDuration) Get() time.Duration {
-	s.Lock()
-	defer s.Unlock()
-	return s.t
-}
-
 func getCycleScanMode(currentCycle, bitrotStartCycle uint64, bitrotStartTime time.Time) madmin.HealScanMode {
 	bitrotCycle := globalHealConfig.BitrotScanCycle()
 	switch bitrotCycle {
@@ -189,7 +171,7 @@ func runDataScanner(pctx context.Context, objAPI ObjectLayer) {
 		}
 	}
 
-	scannerTimer := time.NewTimer(scannerCycle.Get())
+	scannerTimer := time.NewTimer(scannerCycle.Load())
 	defer scannerTimer.Stop()
 
 	for {
@@ -231,7 +213,7 @@ func runDataScanner(pctx context.Context, objAPI ObjectLayer) {
 			}
 
 			// Reset the timer for next cycle.
-			scannerTimer.Reset(scannerCycle.Get())
+			scannerTimer.Reset(scannerCycle.Load())
 		}
 	}
 }

cmd/erasure-metadata-utils.go

@@ -137,24 +137,29 @@ func readAllFileInfo(ctx context.Context, disks []StorageAPI, bucket, object, ve
 				return errDiskNotFound
 			}
 			metadataArray[index], err = disks[index].ReadVersion(ctx, bucket, object, versionID, readData)
-			if err != nil {
-				if !IsErr(err, []error{
-					errFileNotFound,
-					errVolumeNotFound,
-					errFileVersionNotFound,
-					errDiskNotFound,
-				}...) {
-					logger.LogOnceIf(ctx, fmt.Errorf("Drive %s, path (%s/%s) returned an error (%w)",
-						disks[index], bucket, object, err),
-						disks[index].String())
-				}
-			}
 			return err
 		}, index)
 	}
 
+	errs := g.Wait()
+	for index, err := range errs {
+		if err == nil {
+			continue
+		}
+		if !IsErr(err, []error{
+			errFileNotFound,
+			errVolumeNotFound,
+			errFileVersionNotFound,
+			errDiskNotFound,
+		}...) {
+			logger.LogOnceIf(ctx, fmt.Errorf("Drive %s, path (%s/%s) returned an error (%w)",
+				disks[index], bucket, object, err),
+				disks[index].String())
+		}
+	}
+
 	// Return all the metadata.
-	return metadataArray, g.Wait()
+	return metadataArray, errs
 }
 
 // shuffleDisksAndPartsMetadataByIndex this function should be always used by GetObjectNInfo()

cmd/erasure-object.go

@@ -476,16 +476,6 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r
 			}
 			rf, err := disks[index].ReadXL(ctx, bucket, object, readData)
 			if err != nil {
-				if !IsErr(err, []error{
-					errFileNotFound,
-					errVolumeNotFound,
-					errFileVersionNotFound,
-					errDiskNotFound,
-				}...) {
-					logger.LogOnceIf(ctx, fmt.Errorf("Drive %s, path (%s/%s) returned an error (%w)",
-						disks[index], bucket, object, err),
-						disks[index].String())
-				}
 				return err
 			}
 
@@ -502,6 +492,22 @@ func readAllXL(ctx context.Context, disks []StorageAPI, bucket, object string, r
 	}
 
 	errs := g.Wait()
+	for index, err := range errs {
+		if err == nil {
+			continue
+		}
+		if !IsErr(err, []error{
+			errFileNotFound,
+			errVolumeNotFound,
+			errFileVersionNotFound,
+			errDiskNotFound,
+		}...) {
+			logger.LogOnceIf(ctx, fmt.Errorf("Drive %s, path (%s/%s) returned an error (%w)",
+				disks[index], bucket, object, err),
+				disks[index].String())
+		}
+	}
+
 	for index := range metadataArray {
 		if metadataArray[index] != nil {
 			metadataShallowVersions[index] = metadataArray[index].versions

cmd/erasure-server-pool.go

@@ -1751,13 +1751,13 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
 		return err
 	}
 
+	vcfg, _ := globalBucketVersioningSys.Get(bucket)
+
 	ctx, cancel := context.WithCancel(ctx)
 	go func() {
 		defer cancel()
 		defer close(results)
 
-		versioned := opts.Versioned || opts.VersionSuspended
-
 		for _, erasureSet := range z.serverPools {
 			var wg sync.WaitGroup
 			for _, set := range erasureSet.sets {
@@ -1785,11 +1785,14 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
 						if opts.WalkAscending {
 							for i := len(fivs.Versions) - 1; i >= 0; i-- {
 								version := fivs.Versions[i]
+								versioned := vcfg != nil && vcfg.Versioned(version.Name)
+
 								results <- version.ToObjectInfo(bucket, version.Name, versioned)
 							}
 							return
 						}
 						for _, version := range fivs.Versions {
+							versioned := vcfg != nil && vcfg.Versioned(version.Name)
 							results <- version.ToObjectInfo(bucket, version.Name, versioned)
 						}
 					}
@@ -2079,11 +2082,14 @@ func (z *erasureServerPools) ReadHealth(ctx context.Context) bool {
 	}
 
 	b := z.BackendInfo()
-	readQuorum := b.StandardSCData[0]
+	poolReadQuorums := make([]int, len(b.StandardSCData))
+	for i, data := range b.StandardSCData {
+		poolReadQuorums[i] = data
+	}
 
 	for poolIdx := range erasureSetUpCount {
 		for setIdx := range erasureSetUpCount[poolIdx] {
-			if erasureSetUpCount[poolIdx][setIdx] < readQuorum {
+			if erasureSetUpCount[poolIdx][setIdx] < poolReadQuorums[poolIdx] {
 				return false
 			}
 		}
@@ -2120,9 +2126,12 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 	reqInfo := (&logger.ReqInfo{}).AppendTags("maintenance", strconv.FormatBool(opts.Maintenance))
 
 	b := z.BackendInfo()
-	writeQuorum := b.StandardSCData[0]
-	if writeQuorum == b.StandardSCParity {
-		writeQuorum++
+	poolWriteQuorums := make([]int, len(b.StandardSCData))
+	for i, data := range b.StandardSCData {
+		poolWriteQuorums[i] = data
+		if data == b.StandardSCParity {
+			poolWriteQuorums[i] = data + 1
+		}
 	}
 
 	var aggHealStateResult madmin.BgHealState
@@ -2146,34 +2155,44 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
 
 	for poolIdx := range erasureSetUpCount {
 		for setIdx := range erasureSetUpCount[poolIdx] {
-			if erasureSetUpCount[poolIdx][setIdx] < writeQuorum {
+			if erasureSetUpCount[poolIdx][setIdx] < poolWriteQuorums[poolIdx] {
 				logger.LogIf(logger.SetReqInfo(ctx, reqInfo),
 					fmt.Errorf("Write quorum may be lost on pool: %d, set: %d, expected write quorum: %d",
-						poolIdx, setIdx, writeQuorum))
+						poolIdx, setIdx, poolWriteQuorums[poolIdx]))
 				return HealthResult{
 					Healthy:       false,
 					HealingDrives: len(aggHealStateResult.HealDisks),
 					PoolID:        poolIdx,
 					SetID:         setIdx,
-					WriteQuorum:   writeQuorum,
+					WriteQuorum:   poolWriteQuorums[poolIdx],
 				}
 			}
 		}
 	}
 
+	var maximumWriteQuorum int
+	for _, writeQuorum := range poolWriteQuorums {
+		if maximumWriteQuorum == 0 {
+			maximumWriteQuorum = writeQuorum
+		}
+		if writeQuorum > maximumWriteQuorum {
+			maximumWriteQuorum = writeQuorum
+		}
+	}
+
 	// when maintenance is not specified we don't have
 	// to look at the healing side of the code.
 	if !opts.Maintenance {
 		return HealthResult{
 			Healthy:     true,
-			WriteQuorum: writeQuorum,
+			WriteQuorum: maximumWriteQuorum,
 		}
 	}
 
 	return HealthResult{
 		Healthy:       len(aggHealStateResult.HealDisks) == 0,
 		HealingDrives: len(aggHealStateResult.HealDisks),
-		WriteQuorum:   writeQuorum,
+		WriteQuorum:   maximumWriteQuorum,
 	}
 }
 

cmd/generic-handlers.go

@@ -115,9 +115,11 @@ func setRequestLimitHandler(h http.Handler) http.Handler {
 
 // Reserved bucket.
 const (
-	minioReservedBucket     = "minio"
-	minioReservedBucketPath = SlashSeparator + minioReservedBucket
-	loginPathPrefix         = SlashSeparator + "login"
+	minioReservedBucket              = "minio"
+	minioReservedBucketPath          = SlashSeparator + minioReservedBucket
+	minioReservedBucketPathWithSlash = SlashSeparator + minioReservedBucket + SlashSeparator
+
+	loginPathPrefix = SlashSeparator + "login"
 )
 
 func guessIsBrowserReq(r *http.Request) bool {

cmd/globals.go

@@ -37,6 +37,7 @@ import (
 	"github.com/dustin/go-humanize"
 	"github.com/minio/minio/internal/auth"
 	"github.com/minio/minio/internal/config/cache"
+	"github.com/minio/minio/internal/config/callhome"
 	"github.com/minio/minio/internal/config/compress"
 	"github.com/minio/minio/internal/config/dns"
 	xldap "github.com/minio/minio/internal/config/identity/ldap"
@@ -226,10 +227,10 @@ var (
 
 	// global Trace system to send HTTP request/response
 	// and Storage/OS calls info to registered listeners.
-	globalTrace = pubsub.New()
+	globalTrace = pubsub.New(8)
 
 	// global Listen system to send S3 API events to registered listeners
-	globalHTTPListen = pubsub.New()
+	globalHTTPListen = pubsub.New(0)
 
 	// global console system to send console logs to
 	// registered listeners
@@ -243,6 +244,9 @@ var (
 	// The global subnet config
 	globalSubnetConfig subnet.Config
 
+	// The global callhome config
+	globalCallhomeConfig callhome.Config
+
 	globalRemoteEndpoints map[string]Endpoint
 
 	// Global server's network statistics

cmd/handler-api.go

@@ -247,7 +247,12 @@ func maxClients(f http.HandlerFunc) http.HandlerFunc {
 			if val := globalServiceFreeze.Load(); val != nil {
 				if unlock, ok := val.(chan struct{}); ok && unlock != nil {
 					// Wait until unfrozen.
-					<-unlock
+					select {
+					case <-unlock:
+					case <-r.Context().Done():
+						// if client canceled we don't need to wait here forever.
+						return
+					}
 				}
 			}
 		}

cmd/http-stats.go

@@ -148,6 +148,8 @@ type HTTPStats struct {
 	currentS3Requests       HTTPAPIStats
 	totalS3Requests         HTTPAPIStats
 	totalS3Errors           HTTPAPIStats
+	totalS34xxErrors        HTTPAPIStats
+	totalS35xxErrors        HTTPAPIStats
 	totalS3Canceled         HTTPAPIStats
 }
 
@@ -178,6 +180,12 @@ func (st *HTTPStats) toServerHTTPStats() ServerHTTPStats {
 	serverStats.TotalS3Errors = ServerHTTPAPIStats{
 		APIStats: st.totalS3Errors.Load(),
 	}
+	serverStats.TotalS34xxErrors = ServerHTTPAPIStats{
+		APIStats: st.totalS34xxErrors.Load(),
+	}
+	serverStats.TotalS35xxErrors = ServerHTTPAPIStats{
+		APIStats: st.totalS35xxErrors.Load(),
+	}
 	serverStats.TotalS3Canceled = ServerHTTPAPIStats{
 		APIStats: st.totalS3Canceled.Load(),
 	}
@@ -186,27 +194,29 @@ func (st *HTTPStats) toServerHTTPStats() ServerHTTPStats {
 
 // Update statistics from http request and response data
 func (st *HTTPStats) updateStats(api string, r *http.Request, w *logger.ResponseWriter) {
-	// A successful request has a 2xx response code or < 4xx response
-	successReq := w.StatusCode >= 200 && w.StatusCode < 400
-
-	if !strings.HasSuffix(r.URL.Path, prometheusMetricsPathLegacy) ||
-		!strings.HasSuffix(r.URL.Path, prometheusMetricsV2ClusterPath) ||
-		!strings.HasSuffix(r.URL.Path, prometheusMetricsV2NodePath) {
-		st.totalS3Requests.Inc(api)
-		if !successReq {
-			switch w.StatusCode {
-			case 0:
-			case 499:
-				// 499 is a good error, shall be counted as canceled.
-				st.totalS3Canceled.Inc(api)
-			default:
-				st.totalS3Errors.Inc(api)
-			}
-		}
+	// Ignore non S3 requests
+	if strings.HasSuffix(r.URL.Path, minioReservedBucketPathWithSlash) {
+		return
 	}
 
 	// Increment the prometheus http request response histogram with appropriate label
 	httpRequestsDuration.With(prometheus.Labels{"api": api}).Observe(w.TimeToFirstByte.Seconds())
+
+	code := w.StatusCode
+
+	switch {
+	case code == 0:
+	case code == 499:
+		// 499 is a good error, shall be counted as canceled.
+		st.totalS3Canceled.Inc(api)
+	case code >= http.StatusBadRequest:
+		st.totalS3Errors.Inc(api)
+		if code >= http.StatusInternalServerError {
+			st.totalS35xxErrors.Inc(api)
+		} else {
+			st.totalS34xxErrors.Inc(api)
+		}
+	}
 }
 
 // Prepare new HTTPStats structure

cmd/listen-notification-handlers.go

@@ -127,7 +127,7 @@ func (api objectAPIHandlers) ListenNotificationHandler(w http.ResponseWriter, r
 
 	peers, _ := newPeerRestClients(globalEndpoints)
 
-	globalHTTPListen.Subscribe(listenCh, ctx.Done(), func(evI interface{}) bool {
+	listenFn := func(evI interface{}) bool {
 		ev, ok := evI.(event.Event)
 		if !ok {
 			return false
@@ -138,7 +138,13 @@ func (api objectAPIHandlers) ListenNotificationHandler(w http.ResponseWriter, r
 			}
 		}
 		return rulesMap.MatchSimple(ev.EventName, ev.S3.Object.Key)
-	})
+	}
+
+	err := globalHTTPListen.Subscribe(listenCh, ctx.Done(), listenFn)
+	if err != nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrSlowDown), r.URL)
+		return
+	}
 
 	if bucketName != "" {
 		values.Set(peerRESTListenBucket, bucketName)

cmd/main.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"os"
 	"path/filepath"
+	"runtime"
 	"sort"
 
 	"github.com/minio/cli"
@@ -68,6 +69,11 @@ var GlobalFlags = []cli.Flag{
 	},
 }
 
+var versionFlag = cli.BoolFlag{
+	Name:  "version, v",
+	Usage: "print version information",
+}
+
 // Help template for minio.
 var minioHelpTemplate = `NAME:
   {{.Name}} - {{.Usage}}
@@ -132,14 +138,19 @@ func newApp(name string) *cli.App {
 		Usage: "show help",
 	}
 
+	topLevelFlags := make([]cli.Flag, len(GlobalFlags)+1)
+	copy(topLevelFlags, GlobalFlags)
+	topLevelFlags[len(GlobalFlags)] = versionFlag
+
 	app := cli.NewApp()
 	app.Name = name
 	app.Author = "MinIO, Inc."
-	app.Version = ReleaseTag
+	app.Action = versionAndHelpAction
 	app.Usage = "High Performance Object Storage"
 	app.Description = `Build high performance data infrastructure for machine learning, analytics and application data workloads with MinIO`
-	app.Flags = GlobalFlags
+	app.Flags = topLevelFlags
 	app.HideHelpCommand = true // Hide `help, h` command, we already have `minio --help`.
+	app.HideVersion = true
 	app.Commands = commands
 	app.CustomAppHelpTemplate = minioHelpTemplate
 	app.CommandNotFound = func(ctx *cli.Context, command string) {
@@ -159,6 +170,18 @@ func newApp(name string) *cli.App {
 	return app
 }
 
+func versionAndHelpAction(ctx *cli.Context) {
+	if ctx.IsSet("version") {
+		console.Printf("%s version %s\n", ctx.App.Name, ReleaseTag)
+		console.Printf("commit: %s\n", CommitID)
+		console.Printf("go version: %s\n", runtime.Version())
+
+		return
+	}
+
+	cli.ShowAppHelpAndExit(ctx, 1)
+}
+
 // Main main for minio server.
 func Main(args []string) {
 	// Set the minio app name.

cmd/metacache-entries.go

@@ -441,7 +441,6 @@ func (m *metaCacheEntriesSorted) fileInfoVersions(bucket, prefix, delimiter, aft
 	versions = make([]ObjectInfo, 0, m.len())
 	prevPrefix := ""
 	vcfg, _ := globalBucketVersioningSys.Get(bucket)
-	versioned := vcfg != nil && vcfg.Versioned(prefix)
 
 	for _, entry := range m.o {
 		if entry.isObject() {
@@ -478,6 +477,7 @@ func (m *metaCacheEntriesSorted) fileInfoVersions(bucket, prefix, delimiter, aft
 			}
 
 			for _, version := range fiVersions {
+				versioned := vcfg != nil && vcfg.Versioned(entry.name)
 				versions = append(versions, version.ToObjectInfo(bucket, entry.name, versioned))
 			}
 
@@ -516,7 +516,6 @@ func (m *metaCacheEntriesSorted) fileInfos(bucket, prefix, delimiter string) (ob
 	prevPrefix := ""
 
 	vcfg, _ := globalBucketVersioningSys.Get(bucket)
-	versioned := vcfg != nil && vcfg.Versioned(prefix)
 
 	for _, entry := range m.o {
 		if entry.isObject() {
@@ -540,6 +539,7 @@ func (m *metaCacheEntriesSorted) fileInfos(bucket, prefix, delimiter string) (ob
 
 			fi, err := entry.fileInfo(bucket)
 			if err == nil {
+				versioned := vcfg != nil && vcfg.Versioned(entry.name)
 				objects = append(objects, fi.ToObjectInfo(bucket, entry.name, versioned))
 			}
 			continue
@@ -695,8 +695,12 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
 			}
 		}
 		if best.name > last {
-			out <- *best
-			last = best.name
+			select {
+			case <-ctxDone:
+				return ctx.Err()
+			case out <- *best:
+				last = best.name
+			}
 		} else if serverDebugLog {
 			console.Debugln("mergeEntryChannels: discarding duplicate", best.name, "<=", last)
 		}

cmd/metacache-server-pool.go

@@ -554,11 +554,11 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
 	for _, pool := range z.serverPools {
 		for _, set := range pool.sets {
 			wg.Add(1)
-			results := make(chan metaCacheEntry, 100)
-			inputs = append(inputs, results)
+			innerResults := make(chan metaCacheEntry, 100)
+			inputs = append(inputs, innerResults)
 			go func(i int, set *erasureObjects) {
 				defer wg.Done()
-				err := set.listPath(listCtx, o, results)
+				err := set.listPath(listCtx, o, innerResults)
 				mu.Lock()
 				defer mu.Unlock()
 				if err == nil {
@@ -609,6 +609,7 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
 	if err != nil {
 		return err
 	}
+
 	if contextCanceled(ctx) {
 		return ctx.Err()
 	}

cmd/metrics-v2.go

@@ -603,7 +603,27 @@ func getS3RequestsErrorsMD() MetricDescription {
 		Namespace: s3MetricNamespace,
 		Subsystem: requestsSubsystem,
 		Name:      errorsTotal,
-		Help:      "Total number S3 requests with errors",
+		Help:      "Total number S3 requests with (4xx and 5xx) errors",
+		Type:      counterMetric,
+	}
+}
+
+func getS3Requests4xxErrorsMD() MetricDescription {
+	return MetricDescription{
+		Namespace: s3MetricNamespace,
+		Subsystem: requestsSubsystem,
+		Name:      "4xx_" + errorsTotal,
+		Help:      "Total number S3 requests with (4xx) errors",
+		Type:      counterMetric,
+	}
+}
+
+func getS3Requests5xxErrorsMD() MetricDescription {
+	return MetricDescription{
+		Namespace: s3MetricNamespace,
+		Subsystem: requestsSubsystem,
+		Name:      "5xx_" + errorsTotal,
+		Help:      "Total number S3 requests with (5xx) errors",
 		Type:      counterMetric,
 	}
 }
@@ -1347,7 +1367,7 @@ func getNodeHealthMetrics() *MetricsGroup {
 			return
 		}
 		metrics = make([]Metric, 0, 16)
-		nodesUp, nodesDown := GetPeerOnlineCount()
+		nodesUp, nodesDown := globalNotificationSys.GetPeerOnlineCount()
 		metrics = append(metrics, Metric{
 			Description: getNodeOnlineTotalMD(),
 			Value:       float64(nodesUp),
@@ -1488,7 +1508,9 @@ func getHTTPMetrics() *MetricsGroup {
 		metrics = make([]Metric, 0, 3+
 			len(httpStats.CurrentS3Requests.APIStats)+
 			len(httpStats.TotalS3Requests.APIStats)+
-			len(httpStats.TotalS3Errors.APIStats))
+			len(httpStats.TotalS3Errors.APIStats)+
+			len(httpStats.TotalS35xxErrors.APIStats)+
+			len(httpStats.TotalS34xxErrors.APIStats))
 		metrics = append(metrics, Metric{
 			Description: getS3RejectedAuthRequestsTotalMD(),
 			Value:       float64(httpStats.TotalS3RejectedAuth),
@@ -1535,6 +1557,20 @@ func getHTTPMetrics() *MetricsGroup {
 				VariableLabels: map[string]string{"api": api},
 			})
 		}
+		for api, value := range httpStats.TotalS35xxErrors.APIStats {
+			metrics = append(metrics, Metric{
+				Description:    getS3Requests5xxErrorsMD(),
+				Value:          float64(value),
+				VariableLabels: map[string]string{"api": api},
+			})
+		}
+		for api, value := range httpStats.TotalS34xxErrors.APIStats {
+			metrics = append(metrics, Metric{
+				Description:    getS3Requests4xxErrorsMD(),
+				Value:          float64(value),
+				VariableLabels: map[string]string{"api": api},
+			})
+		}
 		for api, value := range httpStats.TotalS3Canceled.APIStats {
 			metrics = append(metrics, Metric{
 				Description:    getS3RequestsCanceledMD(),
@@ -1932,11 +1968,9 @@ func (c *minioClusterCollector) Collect(out chan<- prometheus.Metric) {
 	}
 
 	// Call peer api to fetch metrics
-	peerCh := globalNotificationSys.GetClusterMetrics(GlobalContext)
-	selfCh := ReportMetrics(GlobalContext, c.metricsGroups)
 	wg.Add(2)
-	go publish(peerCh)
-	go publish(selfCh)
+	go publish(ReportMetrics(GlobalContext, c.metricsGroups))
+	go publish(globalNotificationSys.GetClusterMetrics(GlobalContext))
 	wg.Wait()
 }
 

cmd/metrics.go

@@ -110,7 +110,7 @@ func nodeHealthMetricsPrometheus(ch chan<- prometheus.Metric) {
 		return
 	}
 
-	nodesUp, nodesDown := GetPeerOnlineCount()
+	nodesUp, nodesDown := globalNotificationSys.GetPeerOnlineCount()
 	ch <- prometheus.MustNewConstMetric(
 		prometheus.NewDesc(
 			prometheus.BuildFQName(minioNamespace, "nodes", "online"),

cmd/notification.go

@@ -1344,6 +1344,35 @@ func (sys *NotificationSys) restClientFromHash(s string) (client *peerRESTClient
 	return peerClients[idx]
 }
 
+// GetPeerOnlineCount gets the count of online and offline nodes.
+func (sys *NotificationSys) GetPeerOnlineCount() (nodesOnline, nodesOffline int) {
+	nodesOnline = 1 // Self is always online.
+	nodesOffline = 0
+	nodesOnlineIndex := make([]bool, len(sys.peerClients))
+	var wg sync.WaitGroup
+	for idx, client := range sys.peerClients {
+		if client == nil {
+			continue
+		}
+		wg.Add(1)
+		go func(idx int, client *peerRESTClient) {
+			defer wg.Done()
+			nodesOnlineIndex[idx] = client.restClient.HealthCheckFn()
+		}(idx, client)
+
+	}
+	wg.Wait()
+
+	for _, online := range nodesOnlineIndex {
+		if online {
+			nodesOnline++
+		} else {
+			nodesOffline++
+		}
+	}
+	return
+}
+
 // NewNotificationSys - creates new notification system object.
 func NewNotificationSys(endpoints EndpointServerPools) *NotificationSys {
 	// targetList/bucketRulesMap/bucketRemoteTargetRulesMap are populated by NotificationSys.Init()
@@ -1358,21 +1387,6 @@ func NewNotificationSys(endpoints EndpointServerPools) *NotificationSys {
 	}
 }
 
-// GetPeerOnlineCount gets the count of online and offline nodes.
-func GetPeerOnlineCount() (nodesOnline, nodesOffline int) {
-	nodesOnline = 1 // Self is always online.
-	nodesOffline = 0
-	servers := globalNotificationSys.ServerInfo()
-	for _, s := range servers {
-		if s.State == string(madmin.ItemOnline) {
-			nodesOnline++
-			continue
-		}
-		nodesOffline++
-	}
-	return
-}
-
 type eventArgs struct {
 	EventName    event.Name
 	BucketName   string
@@ -1524,7 +1538,7 @@ func (sys *NotificationSys) GetBandwidthReports(ctx context.Context, buckets ...
 }
 
 // GetClusterMetrics - gets the cluster metrics from all nodes excluding self.
-func (sys *NotificationSys) GetClusterMetrics(ctx context.Context) chan Metric {
+func (sys *NotificationSys) GetClusterMetrics(ctx context.Context) <-chan Metric {
 	if sys == nil {
 		return nil
 	}
@@ -1545,11 +1559,14 @@ func (sys *NotificationSys) GetClusterMetrics(ctx context.Context) chan Metric {
 	ch := make(chan Metric)
 	var wg sync.WaitGroup
 	for index, err := range g.Wait() {
-		reqInfo := (&logger.ReqInfo{}).AppendTags("peerAddress",
-			sys.peerClients[index].host.String())
-		ctx := logger.SetReqInfo(ctx, reqInfo)
 		if err != nil {
-			logger.LogOnceIf(ctx, err, sys.peerClients[index].host.String())
+			if sys.peerClients[index] != nil {
+				reqInfo := (&logger.ReqInfo{}).AppendTags("peerAddress",
+					sys.peerClients[index].host.String())
+				logger.LogOnceIf(logger.SetReqInfo(ctx, reqInfo), err, sys.peerClients[index].host.String())
+			} else {
+				logger.LogOnceIf(ctx, err, "peer-offline")
+			}
 			continue
 		}
 		wg.Add(1)

cmd/object-api-options.go

@@ -157,6 +157,7 @@ func getOpts(ctx context.Context, r *http.Request, bucket, object string) (Objec
 			}
 		}
 	}
+	opts.Versioned = globalBucketVersioningSys.PrefixEnabled(bucket, object)
 	opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(bucket, object)
 	return opts, nil
 }
@@ -167,7 +168,7 @@ func delOpts(ctx context.Context, r *http.Request, bucket, object string) (opts
 		return opts, err
 	}
 	opts.Versioned = globalBucketVersioningSys.PrefixEnabled(bucket, object)
-	opts.VersionSuspended = globalBucketVersioningSys.Suspended(bucket)
+	opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(bucket, object)
 	delMarker := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceDeleteMarker))
 	if delMarker != "" {
 		switch delMarker {

cmd/object-handlers.go

@@ -1041,7 +1041,11 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 	srcOpts.VersionID = vid
 
 	// convert copy src encryption options for GET calls
-	getOpts := ObjectOptions{VersionID: srcOpts.VersionID, Versioned: srcOpts.Versioned}
+	getOpts := ObjectOptions{
+		VersionID:        srcOpts.VersionID,
+		Versioned:        srcOpts.Versioned,
+		VersionSuspended: srcOpts.VersionSuspended,
+	}
 	getSSE := encrypt.SSE(srcOpts.ServerSideEncryption)
 	if getSSE != srcOpts.ServerSideEncryption {
 		getOpts.ServerSideEncryption = getSSE

cmd/peer-rest-server.go

@@ -933,7 +933,7 @@ func (s *peerRESTServer) ListenHandler(w http.ResponseWriter, r *http.Request) {
 	// Use buffered channel to take care of burst sends or slow w.Write()
 	ch := make(chan interface{}, 2000)
 
-	globalHTTPListen.Subscribe(ch, doneCh, func(evI interface{}) bool {
+	listenFn := func(evI interface{}) bool {
 		ev, ok := evI.(event.Event)
 		if !ok {
 			return false
@@ -944,7 +944,13 @@ func (s *peerRESTServer) ListenHandler(w http.ResponseWriter, r *http.Request) {
 			}
 		}
 		return rulesMap.MatchSimple(ev.EventName, ev.S3.Object.Key)
-	})
+	}
+
+	err := globalHTTPListen.Subscribe(ch, doneCh, listenFn)
+	if err != nil {
+		s.writeErrorResponse(w, err)
+		return
+	}
 
 	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
 	defer keepAliveTicker.Stop()
@@ -1005,9 +1011,15 @@ func (s *peerRESTServer) TraceHandler(w http.ResponseWriter, r *http.Request) {
 	// Use buffered channel to take care of burst sends or slow w.Write()
 	ch := make(chan interface{}, 2000)
 
-	globalTrace.Subscribe(ch, doneCh, func(entry interface{}) bool {
+	traceFn := func(entry interface{}) bool {
 		return mustTrace(entry, traceOpts)
-	})
+	}
+
+	err = globalTrace.Subscribe(ch, doneCh, traceFn)
+	if err != nil {
+		s.writeErrorResponse(w, err)
+		return
+	}
 
 	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
 	defer keepAliveTicker.Stop()
@@ -1092,7 +1104,11 @@ func (s *peerRESTServer) ConsoleLogHandler(w http.ResponseWriter, r *http.Reques
 	defer close(doneCh)
 
 	ch := make(chan interface{}, 2000)
-	globalConsoleSys.Subscribe(ch, doneCh, "", 0, string(logger.All), nil)
+	err := globalConsoleSys.Subscribe(ch, doneCh, "", 0, string(logger.All), nil)
+	if err != nil {
+		s.writeErrorResponse(w, err)
+		return
+	}
 
 	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
 	defer keepAliveTicker.Stop()
@@ -1158,13 +1174,9 @@ func (s *peerRESTServer) GetPeerMetrics(w http.ResponseWriter, r *http.Request)
 		s.writeErrorResponse(w, errors.New("invalid request"))
 	}
 
-	doneCh := make(chan struct{})
-	defer close(doneCh)
-
 	enc := gob.NewEncoder(w)
 
-	ch := ReportMetrics(r.Context(), peerMetricsGroups)
-	for m := range ch {
+	for m := range ReportMetrics(r.Context(), peerMetricsGroups) {
 		if err := enc.Encode(m); err != nil {
 			s.writeErrorResponse(w, errors.New("Encoding metric failed: "+err.Error()))
 			return

cmd/server-main.go

@@ -138,6 +138,13 @@ func serverCmdArgs(ctx *cli.Context) []string {
 		logger.FatalIf(err, "Unable to validate passed arguments in %s:%s",
 			config.EnvArgs, os.Getenv(config.EnvArgs))
 	}
+	if v == "" {
+		v, _, _, err = env.LookupEnv(config.EnvVolumes)
+		if err != nil {
+			logger.FatalIf(err, "Unable to validate passed arguments in %s:%s",
+				config.EnvVolumes, os.Getenv(config.EnvVolumes))
+		}
+	}
 	if v == "" {
 		// Fall back to older environment value MINIO_ENDPOINTS
 		v, _, _, err = env.LookupEnv(config.EnvEndpoints)
@@ -422,12 +429,12 @@ func serverMain(ctx *cli.Context) {
 	erasureSelfTest()
 	compressSelfTest()
 
-	// Handle all server command args.
-	serverHandleCmdArgs(ctx)
-
 	// Handle all server environment vars.
 	serverHandleEnvVars()
 
+	// Handle all server command args.
+	serverHandleCmdArgs(ctx)
+
 	// Set node name, only set for distributed setup.
 	globalConsoleSys.SetNodeName(globalLocalNodeName)
 

cmd/site-replication.go

@@ -3991,14 +3991,21 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob
 	}
 	rcfg, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket)
 	if err != nil {
+		_, ok := err.(BucketReplicationConfigNotFound)
+		if !ok {
+			return err
+		}
 		replMismatch = true
 	}
 
-	// validate remote targets on current cluster for this bucket
-	_, apiErr := validateReplicationDestination(ctx, bucket, rcfg, false)
-	if apiErr != noError {
-		replMismatch = true
+	if rcfg != nil {
+		// validate remote targets on current cluster for this bucket
+		_, apiErr := validateReplicationDestination(ctx, bucket, rcfg, false)
+		if apiErr != noError {
+			replMismatch = true
+		}
 	}
+
 	if replMismatch {
 		err := c.PeerBucketConfigureReplHandler(ctx, bucket)
 		if err != nil {

cmd/storage-rest-server.go

@@ -1067,10 +1067,13 @@ func waitForHTTPStream(respBody io.ReadCloser, w io.Writer) error {
 				return err
 			}
 			length := binary.LittleEndian.Uint32(tmp[:])
-			_, err = io.CopyBuffer(w, io.LimitReader(respBody, int64(length)), buf)
+			n, err := io.CopyBuffer(w, io.LimitReader(respBody, int64(length)), buf)
 			if err != nil {
 				return err
 			}
+			if n != int64(length) {
+				return io.ErrUnexpectedEOF
+			}
 			continue
 		case 32:
 			continue

cmd/subnet-utils.go

@@ -0,0 +1,102 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"time"
+
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+const (
+	subnetRespBodyLimit = 1 << 20 // 1 MiB
+	callhomeURL         = "https://subnet.min.io/api/callhome"
+	callhomeURLDev      = "http://localhost:9000/api/callhome"
+)
+
+func httpClient(timeout time.Duration) *http.Client {
+	return &http.Client{
+		Timeout:   timeout,
+		Transport: globalProxyTransport,
+	}
+}
+
+func subnetHTTPDo(req *http.Request) (*http.Response, error) {
+	client := httpClient(10 * time.Second)
+	if globalSubnetConfig.ProxyURL != nil {
+		client.Transport.(*http.Transport).Proxy = http.ProxyURL((*url.URL)(globalSubnetConfig.ProxyURL))
+	}
+	return client.Do(req)
+}
+
+func subnetReqDo(r *http.Request, authToken string) (string, error) {
+	r.Header.Set("Authorization", authToken)
+	r.Header.Set("Content-Type", "application/json")
+
+	resp, err := subnetHTTPDo(r)
+	if resp != nil {
+		defer xhttp.DrainBody(resp.Body)
+	}
+	if err != nil {
+		return "", err
+	}
+
+	respBytes, err := ioutil.ReadAll(io.LimitReader(resp.Body, subnetRespBodyLimit))
+	if err != nil {
+		return "", err
+	}
+	respStr := string(respBytes)
+
+	if resp.StatusCode == http.StatusOK {
+		return respStr, nil
+	}
+	return respStr, fmt.Errorf("SUBNET request failed with code %d and error: %s", resp.StatusCode, respStr)
+}
+
+func subnetPostReq(reqURL string, payload interface{}, authToken string) (string, error) {
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return "", err
+	}
+	r, err := http.NewRequest(http.MethodPost, reqURL, bytes.NewReader(body))
+	if err != nil {
+		return "", err
+	}
+	return subnetReqDo(r, authToken)
+}
+
+func sendCallhomeInfo(ch CallhomeInfo) error {
+	if len(globalSubnetConfig.APIKey) == 0 {
+		return errors.New("Cluster is not registered with SUBNET. Please register by running 'mc support register ALIAS'")
+	}
+
+	url := callhomeURL
+	if globalIsCICD {
+		url = callhomeURLDev
+	}
+	_, err := subnetPostReq(url, ch, globalSubnetConfig.APIKey)
+	return err
+}

cmd/xl-storage.go

@@ -94,8 +94,8 @@ type xlStorage struct {
 	endpoint Endpoint
 
 	globalSync bool
-
-	rootDisk bool
+	oDirect    bool // indicates if this disk supports ODirect
+	rootDisk   bool
 
 	diskID string
 
@@ -214,7 +214,7 @@ func newXLStorage(ep Endpoint) (s *xlStorage, err error) {
 	}
 
 	var rootDisk bool
-	if !globalIsCICD {
+	if !globalIsCICD && !globalIsErasureSD {
 		if globalRootDiskThreshold > 0 {
 			// Use MINIO_ROOTDISK_THRESHOLD_SIZE to figure out if
 			// this disk is a root disk.
@@ -264,15 +264,27 @@ func newXLStorage(ep Endpoint) (s *xlStorage, err error) {
 		filePath := pathJoin(s.diskPath, ".writable-check-"+uuid+".tmp")
 		w, err := s.openFileDirect(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL)
 		if err != nil {
-			return s, err
+			// relax single drive mode without O_DIRECT support
+			if globalIsErasureSD && errors.Is(err, errUnsupportedDisk) {
+				s.oDirect = false
+			} else {
+				return s, err
+			}
 		}
 		_, err = w.Write(alignedBuf)
 		w.Close()
 		if err != nil {
 			if isSysErrInvalidArg(err) {
-				return s, errUnsupportedDisk
+				err = errUnsupportedDisk
 			}
-			return s, err
+			// relax single drive mode without O_DIRECT support
+			if globalIsErasureSD && errors.Is(err, errUnsupportedDisk) {
+				s.oDirect = false
+			} else {
+				return s, err
+			}
+		} else {
+			s.oDirect = true
 		}
 		renameAll(filePath, pathJoin(s.diskPath, minioMetaTmpDeletedBucket, uuid))
 
@@ -289,6 +301,7 @@ func newXLStorage(ep Endpoint) (s *xlStorage, err error) {
 		s.diskID = format.Erasure.This
 		s.formatLastCheck = time.Now()
 		s.formatLegacy = format.Erasure.DistributionAlgo == formatErasureVersionV2DistributionAlgoV1
+		s.oDirect = true
 	}
 
 	// Success.
@@ -1372,7 +1385,13 @@ func (s *xlStorage) readAllData(ctx context.Context, volumeDir string, filePath
 		return nil, time.Time{}, ctx.Err()
 	}
 
-	f, err := OpenFileDirectIO(filePath, readMode, 0o666)
+	odirectEnabled := !globalAPIConfig.isDisableODirect() && s.oDirect
+	var f *os.File
+	if odirectEnabled {
+		f, err = OpenFileDirectIO(filePath, readMode, 0o666)
+	} else {
+		f, err = OpenFile(filePath, readMode, 0o666)
+	}
 	if err != nil {
 		if osIsNotExist(err) {
 			// Check if the object doesn't exist because its bucket
@@ -1662,7 +1681,14 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 		return nil, err
 	}
 
-	file, err := OpenFileDirectIO(filePath, readMode, 0o666)
+	odirectEnabled := !globalAPIConfig.isDisableODirect() && s.oDirect
+
+	var file *os.File
+	if odirectEnabled {
+		file, err = OpenFileDirectIO(filePath, readMode, 0o666)
+	} else {
+		file, err = OpenFile(filePath, readMode, 0o666)
+	}
 	if err != nil {
 		switch {
 		case osIsNotExist(err):
@@ -1706,7 +1732,7 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 	}
 
 	alignment := offset%xioutil.DirectioAlignSize == 0
-	if !alignment || globalAPIConfig.isDisableODirect() {
+	if !alignment && odirectEnabled {
 		if err = disk.DisableDirectIO(file); err != nil {
 			file.Close()
 			return nil, err
@@ -1736,13 +1762,11 @@ func (s *xlStorage) ReadFileStream(ctx context.Context, volume, path string, off
 		io.Reader
 		io.Closer
 	}{Reader: io.LimitReader(diskHealthReader(ctx, or), length), Closer: closeWrapper(func() error {
-		if !alignment || offset+length%xioutil.DirectioAlignSize != 0 {
+		if (!alignment || offset+length%xioutil.DirectioAlignSize != 0) && odirectEnabled {
 			// invalidate page-cache for unaligned reads.
-			if !globalAPIConfig.isDisableODirect() {
-				// skip removing from page-cache only
-				// if O_DIRECT was disabled.
-				disk.FadviseDontNeed(file)
-			}
+			// skip removing from page-cache only
+			// if O_DIRECT was disabled.
+			disk.FadviseDontNeed(file)
 		}
 		return or.Close()
 	})}
@@ -1814,7 +1838,13 @@ func (s *xlStorage) CreateFile(ctx context.Context, volume, path string, fileSiz
 		return osErrToFileErr(err)
 	}
 
-	w, err := OpenFileDirectIO(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0o666)
+	odirectEnabled := s.oDirect
+	var w *os.File
+	if odirectEnabled {
+		w, err = OpenFileDirectIO(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0o666)
+	} else {
+		w, err = OpenFile(filePath, os.O_CREATE|os.O_WRONLY|os.O_EXCL, 0o666)
+	}
 	if err != nil {
 		return osErrToFileErr(err)
 	}
@@ -1834,7 +1864,12 @@ func (s *xlStorage) CreateFile(ctx context.Context, volume, path string, fileSiz
 		defer xioutil.ODirectPoolLarge.Put(bufp)
 	}
 
-	written, err := xioutil.CopyAligned(diskHealthWriter(ctx, w), r, *bufp, fileSize, w)
+	var written int64
+	if odirectEnabled {
+		written, err = xioutil.CopyAligned(diskHealthWriter(ctx, w), r, *bufp, fileSize, w)
+	} else {
+		written, err = io.CopyBuffer(diskHealthWriter(ctx, w), r, *bufp)
+	}
 	if err != nil {
 		return err
 	}
@@ -2066,7 +2101,7 @@ func skipAccessChecks(volume string) (ok bool) {
 // RenameData - rename source path to destination path atomically, metadata and data directory.
 func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string) (err error) {
 	defer func() {
-		if err != nil && !contextCanceled(ctx) {
+		if err != nil && !contextCanceled(ctx) && !errors.Is(err, errFileNotFound) {
 			// Only log these errors if context is not yet canceled.
 			logger.LogIf(ctx, fmt.Errorf("srcVolume: %s, srcPath: %s, dstVolume: %s:, dstPath: %s - error %v",
 				srcVolume, srcPath,

docs/bucket/replication/setup_2site_existing_replication.sh

@@ -0,0 +1,93 @@
+#!/usr/bin/env bash
+
+trap 'catch $LINENO' ERR
+
+# shellcheck disable=SC2120
+catch() {
+    if [ $# -ne 0 ]; then
+       echo "error on line $1"
+       for site in sitea siteb; do
+           echo "$site server logs ========="
+           cat "/tmp/${site}_1.log"
+           echo "==========================="
+           cat "/tmp/${site}_2.log"
+       done
+    fi
+
+    echo "Cleaning up instances of MinIO"
+    pkill minio
+    pkill -9 minio
+    rm -rf /tmp/multisitea
+    rm -rf /tmp/multisiteb
+    rm -rf /tmp/data
+}
+
+catch
+
+set -e
+export MINIO_CI_CD=1
+export MINIO_BROWSER=off
+export MINIO_ROOT_USER="minio"
+export MINIO_ROOT_PASSWORD="minio123"
+export MINIO_KMS_AUTO_ENCRYPTION=off
+export MINIO_PROMETHEUS_AUTH_TYPE=public
+export MINIO_KMS_SECRET_KEY=my-minio-key:OSMM+vkKUTCvQs9YL/CVMIMt43HFhkUpqJxTmGl6rYw=
+unset MINIO_KMS_KES_CERT_FILE
+unset MINIO_KMS_KES_KEY_FILE
+unset MINIO_KMS_KES_ENDPOINT
+unset MINIO_KMS_KES_KEY_NAME
+
+wget -O mc https://dl.minio.io/client/mc/release/linux-amd64/mc \
+    && chmod +x mc
+
+minio server --address 127.0.0.1:9001 "http://127.0.0.1:9001/tmp/multisitea/data/disterasure/xl{1...4}" \
+      "http://127.0.0.1:9002/tmp/multisitea/data/disterasure/xl{5...8}" >/tmp/sitea_1.log 2>&1 &
+minio server --address 127.0.0.1:9002 "http://127.0.0.1:9001/tmp/multisitea/data/disterasure/xl{1...4}" \
+      "http://127.0.0.1:9002/tmp/multisitea/data/disterasure/xl{5...8}" >/tmp/sitea_2.log 2>&1 &
+
+minio server --address 127.0.0.1:9003 "http://127.0.0.1:9003/tmp/multisiteb/data/disterasure/xl{1...4}" \
+      "http://127.0.0.1:9004/tmp/multisiteb/data/disterasure/xl{5...8}" >/tmp/siteb_1.log 2>&1 &
+minio server --address 127.0.0.1:9004 "http://127.0.0.1:9003/tmp/multisiteb/data/disterasure/xl{1...4}" \
+      "http://127.0.0.1:9004/tmp/multisiteb/data/disterasure/xl{5...8}" >/tmp/siteb_2.log 2>&1 &
+
+sleep 10s
+
+export MC_HOST_sitea=http://minio:minio123@127.0.0.1:9001
+export MC_HOST_siteb=http://minio:minio123@127.0.0.1:9004
+
+./mc mb sitea/bucket
+
+## Create 100 files
+mkdir -p /tmp/data
+for i in $(seq 1 10); do
+    echo "T" > /tmp/data/file_${i}.txt
+done
+
+./mc mirror /tmp/data sitea/bucket/
+./mc version enable sitea/bucket
+
+./mc mb siteb/bucket/
+./mc version enable siteb/bucket/
+
+echo "adding replication config for site a -> site b"
+remote_arn=$(./mc admin bucket remote add sitea/bucket/ \
+   http://minio:minio123@127.0.0.1:9004/bucket \
+   --service "replication" --json | jq -r ".RemoteARN")
+echo "adding replication rule for a -> b : ${remote_arn}"
+
+./mc replicate add sitea/bucket/ \
+   --remote-bucket "${remote_arn}"
+sleep 1
+
+./mc replicate resync start sitea/bucket/ --remote-bucket "${remote_arn}"
+sleep 10s ## sleep for 10s idea is that we give 100ms per object.
+
+count=$(./mc replicate resync status sitea/bucket --remote-bucket "${remote_arn}" --json | jq .resyncInfo.target[].replicationCount)
+./mc ls --versions sitea/bucket
+./mc ls --versions siteb/bucket
+if [ $count -ne 10 ]; then
+    echo "resync not complete after 100s unexpected failure"
+    ./mc diff sitea/bucket siteb/bucket
+fi
+
+catch

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

@@ -72,7 +72,7 @@
   "gnetId": 13502,
   "graphTooltip": 0,
   "id": null,
-  "iteration": 1653756543524,
+  "iteration": 1654578559812,
   "links": [
     {
       "icon": "external link",
@@ -505,7 +505,7 @@
               },
               {
                 "color": "red",
-                "value": 80
+                "value": 2000
               }
             ]
           },
@@ -661,7 +661,7 @@
               },
               {
                 "color": "red",
-                "value": 80
+                "value": 2000
               }
             ]
           },
@@ -2696,5 +2696,5 @@
   "timezone": "",
   "title": "MinIO Dashboard",
   "uid": "TgmJnqnnk",
-  "version": 7
+  "version": 8
 }

docs/metrics/prometheus/list.md

@@ -52,7 +52,9 @@ These metrics can be from any MinIO server once per collection.
 | `minio_node_process_uptime_seconds`          | Uptime for MinIO process per node in seconds.                                                                       |
 | `minio_node_syscall_read_total`              | Total read SysCalls to the kernel. /proc/[pid]/io syscr                                                             |
 | `minio_node_syscall_write_total`             | Total write SysCalls to the kernel. /proc/[pid]/io syscw                                                            |
-| `minio_s3_requests_error_total`              | Total number S3 requests with errors                                                                                |
+| `minio_s3_requests_errors_total`             | Total number S3 requests with 4xx and 5xx errors                                                                    |
+| `minio_s3_requests_4xx_errors_total`         | Total number S3 requests with 4xx errors                                                                            |
+| `minio_s3_requests_5xx_errors_total`         | Total number S3 requests with 5xx errors                                                                            |
 | `minio_s3_requests_inflight_total`           | Total number of S3 requests currently in flight                                                                     |
 | `minio_s3_requests_total`                    | Total number S3 requests                                                                                            |
 | `minio_s3_time_ttfb_seconds_distribution`    | Distribution of the time to first byte across API calls.                                                            |

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

@@ -2,7 +2,7 @@ version: '3.7'
 
 # Settings and configurations that are common for all containers
 x-minio-common: &minio-common
-  image: quay.io/minio/minio:RELEASE.2022-05-26T05-48-41Z
+  image: quay.io/minio/minio:RELEASE.2022-06-07T00-33-41Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

go.mod

@@ -43,9 +43,9 @@ require (
 	github.com/lib/pq v1.10.4
 	github.com/miekg/dns v1.1.48
 	github.com/minio/cli v1.22.0
-	github.com/minio/console v0.17.2
+	github.com/minio/console v0.18.1
 	github.com/minio/csvparser v1.0.0
-	github.com/minio/dperf v0.3.6
+	github.com/minio/dperf v0.4.0
 	github.com/minio/highwayhash v1.0.2
 	github.com/minio/kes v0.19.2
 	github.com/minio/madmin-go v1.3.14

go.sum

@@ -611,12 +611,12 @@ github.com/minio/cli v1.22.0 h1:VTQm7lmXm3quxO917X3p+el1l0Ca5X3S4PM2ruUYO68=
 github.com/minio/cli v1.22.0/go.mod h1:bYxnK0uS629N3Bq+AOZZ+6lwF77Sodk4+UL9vNuXhOY=
 github.com/minio/colorjson v1.0.2 h1:Em3IM68MTm3h+Oxa0nxrV9VQqDgbxvC5iq5A+pqzDeI=
 github.com/minio/colorjson v1.0.2/go.mod h1:JWxcL2n8T8JVf+NY6awl6kn5nK49aAzHOeQEM33dL0k=
-github.com/minio/console v0.17.2 h1:1S6yYqWAEumwyUiHZM/GGYuzmePes4tsXYzQPqVNr7Y=
-github.com/minio/console v0.17.2/go.mod h1:yfF8zcgJtD+KPPFHHD1PMpFbhMnQRY8C7jmsrHBdoRU=
+github.com/minio/console v0.18.1 h1:2KiGwfkE52BBlJsH0QM0KCtf5B3Wfj/+FABVjIzG9Zc=
+github.com/minio/console v0.18.1/go.mod h1:yfF8zcgJtD+KPPFHHD1PMpFbhMnQRY8C7jmsrHBdoRU=
 github.com/minio/csvparser v1.0.0 h1:xJEHcYK8ZAjeW4hNV9Zu30u+/2o4UyPnYgyjWp8b7ZU=
 github.com/minio/csvparser v1.0.0/go.mod h1:lKXskSLzPgC5WQyzP7maKH7Sl1cqvANXo9YCto8zbtM=
-github.com/minio/dperf v0.3.6 h1:EAe/OEEuFRGRbcGsfV0+2KsQkrmTvh51UOoCd/h4DDI=
-github.com/minio/dperf v0.3.6/go.mod h1:mo7rPmlIl3Bgn8CrM0Yjeu3AL3xXVd7sH7Hpj4OGK1E=
+github.com/minio/dperf v0.4.0 h1:iUiHOO96S/hcRzIaqWao6NJPkbuNQTEndrq3ht+vXOQ=
+github.com/minio/dperf v0.4.0/go.mod h1:mo7rPmlIl3Bgn8CrM0Yjeu3AL3xXVd7sH7Hpj4OGK1E=
 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.1/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=

internal/bucket/replication/replication.go

@@ -146,6 +146,18 @@ type ObjectOpts struct {
 	TargetArn      string
 }
 
+// HasExistingObjectReplication returns true if any of the rule returns 'ExistingObjects' replication.
+func (c Config) HasExistingObjectReplication(arn string) bool {
+	for _, rule := range c.Rules {
+		if rule.Destination.ARN == arn || c.RoleArn == arn {
+			if rule.ExistingObjectReplication.Status == Enabled {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 // FilterActionableRules returns the rules actions that need to be executed
 // after evaluating prefix/tag filtering
 func (c Config) FilterActionableRules(obj ObjectOpts) []Rule {

internal/config/callhome/callhome.go

@@ -0,0 +1,65 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package callhome
+
+import (
+	"time"
+
+	"github.com/minio/minio/internal/config"
+	"github.com/minio/pkg/env"
+)
+
+// Callhome related keys
+const (
+	Enable    = "enable"
+	Frequency = "frequency"
+)
+
+// DefaultKVS - default KV config for subnet settings
+var DefaultKVS = config.KVS{
+	config.KV{
+		Key:   Enable,
+		Value: "off",
+	},
+	config.KV{
+		Key:   Frequency,
+		Value: "24h",
+	},
+}
+
+// Config represents the subnet related configuration
+type Config struct {
+	// Flag indicating whether callhome is enabled.
+	Enable bool `json:"enable"`
+
+	// The interval between callhome cycles
+	Frequency time.Duration `json:"frequency"`
+}
+
+// LookupConfig - lookup config and override with valid environment settings if any.
+func LookupConfig(kvs config.KVS) (cfg Config, err error) {
+	if err = config.CheckValidKeys(config.CallhomeSubSys, kvs, DefaultKVS); err != nil {
+		return cfg, err
+	}
+
+	cfg.Enable = env.Get(config.EnvMinIOCallhomeEnable,
+		kvs.GetWithDefault(Enable, DefaultKVS)) == config.EnableOn
+	cfg.Frequency, err = time.ParseDuration(env.Get(config.EnvMinIOCallhomeFrequency,
+		kvs.GetWithDefault(Frequency, DefaultKVS)))
+	return cfg, err
+}

internal/config/callhome/help.go

@@ -0,0 +1,42 @@
+// Copyright (c) 2015-2022 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package callhome
+
+import "github.com/minio/minio/internal/config"
+
+var (
+	defaultHelpPostfix = func(key string) string {
+		return config.DefaultHelpPostfix(DefaultKVS, key)
+	}
+
+	// HelpCallhome - provides help for callhome config
+	HelpCallhome = config.HelpKVS{
+		config.HelpKV{
+			Key:         Enable,
+			Type:        "on|off",
+			Description: "set to enable callhome" + defaultHelpPostfix(Enable),
+			Optional:    true,
+		},
+		config.HelpKV{
+			Key:         Frequency,
+			Type:        "duration",
+			Description: "time duration between callhome cycles e.g. 24h" + defaultHelpPostfix(Frequency),
+			Optional:    true,
+		},
+	}
+)

internal/config/config.go

@@ -88,6 +88,7 @@ const (
 	ScannerSubSys        = "scanner"
 	CrawlerSubSys        = "crawler"
 	SubnetSubSys         = "subnet"
+	CallhomeSubSys       = "callhome"
 
 	// Add new constants here if you add new fields to config.
 )
@@ -161,6 +162,7 @@ var SubSystems = set.CreateStringSet(
 	NotifyRedisSubSys,
 	NotifyWebhookSubSys,
 	SubnetSubSys,
+	CallhomeSubSys,
 )
 
 // SubSystemsDynamic - all sub-systems that have dynamic config.
@@ -170,6 +172,7 @@ var SubSystemsDynamic = set.CreateStringSet(
 	ScannerSubSys,
 	HealSubSys,
 	SubnetSubSys,
+	CallhomeSubSys,
 	LoggerWebhookSubSys,
 	AuditWebhookSubSys,
 	AuditKafkaSubSys,

internal/config/constants.go

@@ -48,14 +48,19 @@ const (
 	EnvPublicIPs  = "MINIO_PUBLIC_IPS"
 	EnvFSOSync    = "MINIO_FS_OSYNC"
 	EnvArgs       = "MINIO_ARGS"
+	EnvVolumes    = "MINIO_VOLUMES"
 	EnvDNSWebhook = "MINIO_DNS_WEBHOOK_ENDPOINT"
 
 	EnvSiteName   = "MINIO_SITE_NAME"
 	EnvSiteRegion = "MINIO_SITE_REGION"
 
-	EnvMinIOSubnetLicense      = "MINIO_SUBNET_LICENSE" // Deprecated Dec 2021
-	EnvMinIOSubnetAPIKey       = "MINIO_SUBNET_API_KEY"
-	EnvMinIOSubnetProxy        = "MINIO_SUBNET_PROXY"
+	EnvMinIOSubnetLicense = "MINIO_SUBNET_LICENSE" // Deprecated Dec 2021
+	EnvMinIOSubnetAPIKey  = "MINIO_SUBNET_API_KEY"
+	EnvMinIOSubnetProxy   = "MINIO_SUBNET_PROXY"
+
+	EnvMinIOCallhomeEnable    = "MINIO_CALLHOME_ENABLE"
+	EnvMinIOCallhomeFrequency = "MINIO_CALLHOME_FREQUENCY"
+
 	EnvMinIOServerURL          = "MINIO_SERVER_URL"
 	EnvMinIOBrowserRedirectURL = "MINIO_BROWSER_REDIRECT_URL"
 	EnvRootDiskThresholdSize   = "MINIO_ROOTDISK_THRESHOLD_SIZE"

internal/config/subnet/api-key.go

@@ -18,10 +18,9 @@
 package subnet
 
 import (
-	"net/url"
-
 	"github.com/minio/minio/internal/config"
 	"github.com/minio/pkg/env"
+	xnet "github.com/minio/pkg/net"
 )
 
 // DefaultKVS - default KV config for subnet settings
@@ -49,7 +48,7 @@ type Config struct {
 	APIKey string `json:"api_key"`
 
 	// The HTTP(S) proxy URL to use for connecting to SUBNET
-	Proxy string `json:"proxy"`
+	ProxyURL *xnet.URL `json:"proxy_url"`
 }
 
 // LookupConfig - lookup config and override with valid environment settings if any.
@@ -58,10 +57,12 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) {
 		return cfg, err
 	}
 
-	cfg.Proxy = env.Get(config.EnvMinIOSubnetProxy, kvs.Get(config.Proxy))
-	_, err = url.Parse(cfg.Proxy)
-	if err != nil {
-		return cfg, err
+	proxy := env.Get(config.EnvMinIOSubnetProxy, kvs.Get(config.Proxy))
+	if len(proxy) > 0 {
+		cfg.ProxyURL, err = xnet.ParseHTTPURL(proxy)
+		if err != nil {
+			return cfg, err
+		}
 	}
 
 	cfg.License = env.Get(config.EnvMinIOSubnetLicense, kvs.Get(config.License))

internal/pubsub/pubsub.go

@@ -18,6 +18,7 @@
 package pubsub
 
 import (
+	"fmt"
 	"sync"
 	"sync/atomic"
 )
@@ -32,6 +33,7 @@ type Sub struct {
 type PubSub struct {
 	subs           []*Sub
 	numSubscribers int32
+	maxSubscribers int32
 	sync.RWMutex
 }
 
@@ -53,13 +55,18 @@ func (ps *PubSub) Publish(item interface{}) {
 }
 
 // Subscribe - Adds a subscriber to pubsub system
-func (ps *PubSub) Subscribe(subCh chan interface{}, doneCh <-chan struct{}, filter func(entry interface{}) bool) {
+func (ps *PubSub) Subscribe(subCh chan interface{}, doneCh <-chan struct{}, filter func(entry interface{}) bool) error {
+	totalSubs := atomic.AddInt32(&ps.numSubscribers, 1)
+	if ps.maxSubscribers > 0 && totalSubs > ps.maxSubscribers {
+		atomic.AddInt32(&ps.numSubscribers, -1)
+		return fmt.Errorf("the limit of `%d` subscribers is reached", ps.maxSubscribers)
+	}
+
 	ps.Lock()
 	defer ps.Unlock()
 
 	sub := &Sub{subCh, filter}
 	ps.subs = append(ps.subs, sub)
-	atomic.AddInt32(&ps.numSubscribers, 1)
 
 	go func() {
 		<-doneCh
@@ -74,6 +81,8 @@ func (ps *PubSub) Subscribe(subCh chan interface{}, doneCh <-chan struct{}, filt
 		}
 		atomic.AddInt32(&ps.numSubscribers, -1)
 	}()
+
+	return nil
 }
 
 // NumSubscribers returns the number of current subscribers
@@ -81,7 +90,8 @@ func (ps *PubSub) NumSubscribers() int32 {
 	return atomic.LoadInt32(&ps.numSubscribers)
 }
 
-// New inits a PubSub system
-func New() *PubSub {
-	return &PubSub{}
+// New inits a PubSub system with a limit of maximum
+// subscribers unless zero is specified
+func New(maxSubscribers int32) *PubSub {
+	return &PubSub{maxSubscribers: maxSubscribers}
 }

internal/pubsub/pubsub_test.go

@@ -24,68 +24,100 @@ import (
 )
 
 func TestSubscribe(t *testing.T) {
-	ps := New()
+	ps := New(2)
 	ch1 := make(chan interface{}, 1)
 	ch2 := make(chan interface{}, 1)
 	doneCh := make(chan struct{})
 	defer close(doneCh)
-	ps.Subscribe(ch1, doneCh, nil)
-	ps.Subscribe(ch2, doneCh, nil)
+	if err := ps.Subscribe(ch1, doneCh, nil); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if err := ps.Subscribe(ch2, doneCh, nil); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 	ps.Lock()
 	defer ps.Unlock()
 	if len(ps.subs) != 2 {
-		t.Errorf("expected 2 subscribers")
+		t.Fatalf("expected 2 subscribers")
+	}
+}
+
+func TestSubscribeExceedingLimit(t *testing.T) {
+	ps := New(2)
+	ch1 := make(chan interface{}, 1)
+	ch2 := make(chan interface{}, 1)
+	ch3 := make(chan interface{}, 1)
+	doneCh := make(chan struct{})
+	defer close(doneCh)
+	if err := ps.Subscribe(ch1, doneCh, nil); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if err := ps.Subscribe(ch2, doneCh, nil); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if err := ps.Subscribe(ch3, doneCh, nil); err == nil {
+		t.Fatalf("unexpected nil err")
 	}
 }
 
 func TestUnsubscribe(t *testing.T) {
-	ps := New()
+	ps := New(2)
 	ch1 := make(chan interface{}, 1)
 	ch2 := make(chan interface{}, 1)
 	doneCh1 := make(chan struct{})
 	doneCh2 := make(chan struct{})
-	ps.Subscribe(ch1, doneCh1, nil)
-	ps.Subscribe(ch2, doneCh2, nil)
+	if err := ps.Subscribe(ch1, doneCh1, nil); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if err := ps.Subscribe(ch2, doneCh2, nil); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 
 	close(doneCh1)
 	// Allow for the above statement to take effect.
 	time.Sleep(100 * time.Millisecond)
 	ps.Lock()
 	if len(ps.subs) != 1 {
-		t.Errorf("expected 1 subscriber")
+		t.Fatal("expected 1 subscriber")
 	}
 	ps.Unlock()
 	close(doneCh2)
 }
 
 func TestPubSub(t *testing.T) {
-	ps := New()
+	ps := New(1)
 	ch1 := make(chan interface{}, 1)
 	doneCh1 := make(chan struct{})
 	defer close(doneCh1)
-	ps.Subscribe(ch1, doneCh1, func(entry interface{}) bool { return true })
+	if err := ps.Subscribe(ch1, doneCh1, func(entry interface{}) bool { return true }); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 	val := "hello"
 	ps.Publish(val)
 	msg := <-ch1
 	if msg != "hello" {
-		t.Errorf(fmt.Sprintf("expected %s , found %s", val, msg))
+		t.Fatalf(fmt.Sprintf("expected %s , found %s", val, msg))
 	}
 }
 
 func TestMultiPubSub(t *testing.T) {
-	ps := New()
+	ps := New(2)
 	ch1 := make(chan interface{}, 1)
 	ch2 := make(chan interface{}, 1)
 	doneCh := make(chan struct{})
 	defer close(doneCh)
-	ps.Subscribe(ch1, doneCh, func(entry interface{}) bool { return true })
-	ps.Subscribe(ch2, doneCh, func(entry interface{}) bool { return true })
+	if err := ps.Subscribe(ch1, doneCh, func(entry interface{}) bool { return true }); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if err := ps.Subscribe(ch2, doneCh, func(entry interface{}) bool { return true }); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
 	val := "hello"
 	ps.Publish(val)
 
 	msg1 := <-ch1
 	msg2 := <-ch2
 	if msg1 != "hello" && msg2 != "hello" {
-		t.Errorf(fmt.Sprintf("expected both subscribers to have%s , found %s and  %s", val, msg1, msg2))
+		t.Fatalf(fmt.Sprintf("expected both subscribers to have%s , found %s and  %s", val, msg1, msg2))
 	}
 }