do not print errors upon 'nil' err

Update yaml files to latest version RELEASE.2022-06-06T23-14-52Z
Add periodic callhome functionality (#14918 )
2026-10-17 23:16:09 +03:00 · 2022-06-06 17:33:41 -07:00 · 2022-06-06 23:47:31 +00:00 · 2022-06-06 16:14:52 -07:00 · 2022-06-06 15:15:11 -07:00 · 2022-06-06 15:14:56 -07:00
42 changed files with 1228 additions and 533 deletions
@@ -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
@@ -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)"
@@ -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.
@@ -1409,9 +1409,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 +1489,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 +1831,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 +1861,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 +1869,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 +1890,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 +1922,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 +1937,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 +1952,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 +1967,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 +1983,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 +1999,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 +2079,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 +2308,7 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 				return
 			}
 			w.(http.Flusher).Flush()
-		case <-deadlinedCtx.Done():
+		case <-healthCtx.Done():
 			return
 		}
 	}
@@ -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
@@ -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: {
@@ -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(
@@ -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)
-}
@@ -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)
+}
@@ -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
@@ -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))
+	}
+}
@@ -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())
 	}
 }

@@ -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()
@@ -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
@@ -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())
 		}
 	}
 }
@@ -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()
@@ -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
@@ -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)
 						}
 					}
@@ -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
@@ -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)
@@ -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.
@@ -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)
 		}
@@ -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()
 	}
@@ -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 {
@@ -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
@@ -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()
@@ -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 {
@@ -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
+}
@@ -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.
@@ -2066,7 +2066,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,
@@ -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
@@ -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-06-02T02-11-04Z
+  image: quay.io/minio/minio:RELEASE.2022-06-06T23-14-52Z
  command: server --console-address ":9001" http://minio{1...4}/data{1...2}
  expose:
    - "9000"
@@ -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.0
 	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
@@ -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.0 h1:6SDDYDBqwGPhw930yTSlYWWLoH9Qj/O1mKTXBG51MhM=
+github.com/minio/console v0.18.0/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=
@@ -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 {
@@ -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
+}
@@ -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,
+		},
+	}
+)
@@ -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,
@@ -53,9 +53,13 @@ const (
 	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"
@@ -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))
@@ -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}
 }
@@ -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))
 	}
 }
Author	SHA1	Message	Date
Minio Trusted	e2d4d097e7	do not print errors upon 'nil' err	2022-06-06 17:33:41 -07:00
Minio Trusted	ac8cb6ba0d	Update yaml files to latest version RELEASE.2022-06-06T23-14-52Z	2022-06-06 23:47:31 +00:00
Shireesh Anjal	4ce81fd07f	Add periodic callhome functionality (#14918 ) * Add periodic callhome functionality Periodically (every 24hrs by default), fetch callhome information and upload it to SUBNET. New config keys under the `callhome` subsystem: enable - Set to `on` for enabling callhome. Default `off` frequency - Interval between callhome cycles. Default `24h` * Improvements based on review comments - Update `enableCallhome` safely - Rename pctx to ctx - Block during execution of callhome - Store parsed proxy URL in global subnet config - Store callhome URL(s) in constants - Use existing global transport - Pass auth token to subnetPostReq - Use `config.EnableOn` instead of `"on"` * Use atomic package instead of lock * Use uber atomic package * Use `Cancel` instead of `cancel` Co-authored-by: Harshavardhana <harsha@minio.io> Co-authored-by: Harshavardhana <harsha@minio.io> Co-authored-by: Aditya Manthramurthy <donatello@users.noreply.github.com>	2022-06-06 16:14:52 -07:00
Harshavardhana	df9eeb7f8f	fix: do not log concurrently when multiple disks return errors (#15044 ) since the values inside 'context' are mutated internally by logger, make sure to log serially upon errors not concurrently.	2022-06-06 15:15:11 -07:00
Harshavardhana	31c4fdbf79	fix: resyncing 'null' version on pre-existing content (#15043 ) PR #15041 fixed replicating 'null' version however due to a regression from #14994 caused the target versions for these 'null' versioned objects to have different 'versions', this may cause confusion with bi-directional replication and cause double replication. This PR fixes this properly by making sure we replicate the correct versions on the objects.	2022-06-06 15:14:56 -07:00
Harshavardhana	48e367ff7d	reject resync start on misconfigured replication rules (#15041 ) we expect resync to start on buckets with replication rule ExistingObjects enabled, if not we reject such calls.	2022-06-06 02:54:39 -07:00
Anis Elleuch	fd02492cb7	avoid limits on the number of parallel trace/bucket notifications listeners (#14799 ) Simplifies overall limits on the incoming listeners for notifications. Fixes #14566	2022-06-05 14:29:12 -07:00
Aditya Manthramurthy	addfa35d93	Add FIPS build to CI and add README.fips.md (#15038 )	2022-06-04 18:25:37 -07:00
Harshavardhana	5afdc56796	allow single drive mode to run on root disk (#15037 ) for practical reasons, allow root disk based installs for single drive mode.	2022-06-03 12:53:42 -07:00
Harshavardhana	fb1c333a83	update latest dperf v0.4.0	2022-06-03 11:13:20 -07:00
Harshavardhana	c3e1da8e04	honor canceled context and do not leak on mergeChannels (#15034 ) mergeEntryChannels has the potential to perpetually wait on the results channel, context might be closed and we did not honor the caller context canceling.	2022-06-03 05:59:02 -07:00
Anis Elleuch	20a753e2e5	Fix a possible service freeze after perf object (#15036 ) The S3 service can be frozen indefinitely if a client or mc asks for object perf API but quits early or has some networking issues. The reason is that partialWrite() can block indefinitely. This commit makes partialWrite() listens to context cancellation as well. It also renames deadlinedCtx to healthCtx since it covers handler context cancellation and not only not only the speedtest deadline.	2022-06-03 05:58:45 -07:00
Minio Trusted	3a398775fb	Update yaml files to latest version RELEASE.2022-06-03T01-40-53Z	2022-06-03 02:36:56 +00:00
Aditya Manthramurthy	61a7434379	Update --version option behavior (#15032 ) - Add git commit ID - Add go version	2022-06-02 18:40:53 -07:00
Aditya Manthramurthy	09f5e29327	Bump up console to v0.18.0 (#15031 )	2022-06-02 17:34:37 -07:00
Poorna	29edb4ccfe	fix: site replication bucket heal to not panic if replication config is missing (#15025 )	2022-06-02 12:34:03 -07:00
Minio Trusted	197d6fb644	Update yaml files to latest version RELEASE.2022-06-02T16-16-26Z	2022-06-02 17:46:32 +00:00