avoid replication proxy on version excluded paths (#14878)

no need to attempt proxying objects that were
never replicated, but do have local `null`
versions on them.

cmd/admin-handlers-site-replication.go

@@ -241,6 +241,8 @@ func (a adminAPIHandlers) SRPeerReplicateBucketItem(w http.ResponseWriter, r *ht
 				return
 			}
 		}
+	case madmin.SRBucketMetaTypeVersionConfig:
+		err = globalSiteReplicationSys.PeerBucketVersioningHandler(ctx, item.Bucket, item.Versioning)
 	case madmin.SRBucketMetaTypeTags:
 		err = globalSiteReplicationSys.PeerBucketTaggingHandler(ctx, item.Bucket, item.Tags)
 	case madmin.SRBucketMetaTypeObjectLockConfig:

cmd/admin-handlers.go

@@ -57,7 +57,8 @@ import (
 )
 
 const (
-	maxEConfigJSONSize = 262272
+	maxEConfigJSONSize        = 262272
+	kubernetesVersionEndpoint = "https://kubernetes.default.svc/version"
 )
 
 // Only valid query params for mgmt admin APIs.
@@ -1768,6 +1769,36 @@ func getServerInfo(ctx context.Context, r *http.Request) madmin.InfoMessage {
 	}
 }
 
+func getKubernetesInfo(dctx context.Context) madmin.KubernetesInfo {
+	ctx, cancel := context.WithCancel(dctx)
+	defer cancel()
+
+	ki := madmin.KubernetesInfo{}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, kubernetesVersionEndpoint, nil)
+	if err != nil {
+		ki.Error = err.Error()
+		return ki
+	}
+
+	client := &http.Client{
+		Transport: NewGatewayHTTPTransport(),
+		Timeout:   10 * time.Second,
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		ki.Error = err.Error()
+		return ki
+	}
+
+	decoder := json.NewDecoder(resp.Body)
+	if err := decoder.Decode(&ki); err != nil {
+		ki.Error = err.Error()
+	}
+	return ki
+}
+
 // HealthInfoHandler - GET /minio/admin/v3/healthinfo
 // ----------
 // Get server health info
@@ -1855,6 +1886,13 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 		info.SetAddr(anonAddr(info.GetAddr()))
 	}
 
+	getAndWritePlatformInfo := func() {
+		if IsKubernetes() {
+			healthInfo.Sys.KubernetesInfo = getKubernetesInfo(deadlinedCtx)
+			partialWrite(healthInfo)
+		}
+	}
+
 	getAndWriteCPUs := func() {
 		if query.Get("syscpu") == "true" {
 			localCPUInfo := madmin.GetCPUs(deadlinedCtx, globalLocalNodeName)
@@ -2177,6 +2215,7 @@ func (a adminAPIHandlers) HealthInfoHandler(w http.ResponseWriter, r *http.Reque
 		defer close(healthInfoCh)
 
 		partialWrite(healthInfo) // Write first message with only version and deployment id populated
+		getAndWritePlatformInfo()
 		getAndWriteCPUs()
 		getAndWritePartitions()
 		getAndWriteOSInfo()

cmd/api-errors.go

@@ -280,6 +280,7 @@ const (
 	ErrSiteReplicationBucketConfigError
 	ErrSiteReplicationBucketMetaError
 	ErrSiteReplicationIAMError
+	ErrSiteReplicationConfigMissing
 
 	// Bucket Quota error codes
 	ErrAdminBucketQuotaExceeded
@@ -1340,7 +1341,11 @@ var errorCodes = errorCodeMap{
 		Description:    "Error while replicating an IAM item",
 		HTTPStatusCode: http.StatusServiceUnavailable,
 	},
-
+	ErrSiteReplicationConfigMissing: {
+		Code:           "XMinioSiteReplicationConfigMissingError",
+		Description:    "Site not found in site replication configuration",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 	ErrMaximumExpires: {
 		Code:           "AuthorizationQueryParametersError",
 		Description:    "X-Amz-Expires must be less than a week (in seconds); that is, the given X-Amz-Expires must be less than 604800 seconds",

cmd/bucket-handlers.go

@@ -471,9 +471,6 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 		hasLockEnabled = true
 	}
 
-	versioned := globalBucketVersioningSys.Enabled(bucket)
-	suspended := globalBucketVersioningSys.Suspended(bucket)
-
 	type deleteResult struct {
 		delInfo DeletedObject
 		errInfo DeleteError
@@ -481,8 +478,8 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 
 	deleteResults := make([]deleteResult, len(deleteObjectsReq.Objects))
 
+	vc, _ := globalBucketVersioningSys.Get(bucket)
 	oss := make([]*objSweeper, len(deleteObjectsReq.Objects))
-
 	for index, object := range deleteObjectsReq.Objects {
 		if apiErrCode := checkRequestAuthType(ctx, r, policy.DeleteObjectAction, bucket, object.ObjectName); apiErrCode != ErrNone {
 			if apiErrCode == ErrSignatureDoesNotMatch || apiErrCode == ErrInvalidAccessKeyID {
@@ -514,8 +511,8 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 
 		opts := ObjectOptions{
 			VersionID:        object.VersionID,
-			Versioned:        versioned,
-			VersionSuspended: suspended,
+			Versioned:        vc.PrefixEnabled(object.ObjectName),
+			VersionSuspended: vc.Suspended(),
 		}
 
 		if replicateDeletes || object.VersionID != "" && hasLockEnabled || !globalTierConfigMgr.Empty() {
@@ -526,7 +523,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 		}
 
 		if !globalTierConfigMgr.Empty() {
-			oss[index] = newObjSweeper(bucket, object.ObjectName).WithVersion(opts.VersionID).WithVersioning(versioned, suspended)
+			oss[index] = newObjSweeper(bucket, object.ObjectName).WithVersion(opts.VersionID).WithVersioning(opts.Versioned, opts.VersionSuspended)
 			oss[index].SetTransitionState(goi.TransitionedObject)
 		}
 
@@ -581,8 +578,8 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 
 	deleteList := toNames(objectsToDelete)
 	dObjects, errs := deleteObjectsFn(ctx, bucket, deleteList, ObjectOptions{
-		Versioned:        versioned,
-		VersionSuspended: suspended,
+		PrefixEnabledFn:  vc.PrefixEnabled,
+		VersionSuspended: vc.Suspended(),
 	})
 
 	for i := range errs {
@@ -638,22 +635,12 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 			continue
 		}
 
-		if replicateDeletes {
-			if dobj.DeleteMarkerReplicationStatus() == replication.Pending || dobj.VersionPurgeStatus() == Pending {
-				dv := DeletedObjectReplicationInfo{
-					DeletedObject: dobj,
-					Bucket:        bucket,
-				}
-				scheduleReplicationDelete(ctx, dv, objectAPI)
+		if replicateDeletes && (dobj.DeleteMarkerReplicationStatus() == replication.Pending || dobj.VersionPurgeStatus() == Pending) {
+			dv := DeletedObjectReplicationInfo{
+				DeletedObject: dobj,
+				Bucket:        bucket,
 			}
-		}
-
-	}
-
-	// Notify deleted event for objects.
-	for _, dobj := range deletedObjects {
-		if dobj.ObjectName == "" {
-			continue
+			scheduleReplicationDelete(ctx, dv, objectAPI)
 		}
 
 		eventName := event.ObjectRemovedDelete

cmd/bucket-lifecycle.go

@@ -330,7 +330,7 @@ const (
 // 2. when a transitioned object expires (based on an ILM rule).
 func expireTransitionedObject(ctx context.Context, objectAPI ObjectLayer, oi *ObjectInfo, lcOpts lifecycle.ObjectOpts, action expireAction) error {
 	var opts ObjectOptions
-	opts.Versioned = globalBucketVersioningSys.Enabled(oi.Bucket)
+	opts.Versioned = globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name)
 	opts.VersionID = lcOpts.VersionID
 	opts.Expiration = ExpirationOptions{Expire: true}
 	switch action {
@@ -415,8 +415,8 @@ func transitionObject(ctx context.Context, objectAPI ObjectLayer, oi ObjectInfo)
 			ETag:   oi.ETag,
 		},
 		VersionID:        oi.VersionID,
-		Versioned:        globalBucketVersioningSys.Enabled(oi.Bucket),
-		VersionSuspended: globalBucketVersioningSys.Suspended(oi.Bucket),
+		Versioned:        globalBucketVersioningSys.PrefixEnabled(oi.Bucket, oi.Name),
+		VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name),
 		MTime:            oi.ModTime,
 	}
 	return tier, objectAPI.TransitionObject(ctx, oi.Bucket, oi.Name, opts)
@@ -575,8 +575,8 @@ func (r *RestoreObjectRequest) validate(ctx context.Context, objAPI ObjectLayer)
 
 // postRestoreOpts returns ObjectOptions with version-id from the POST restore object request for a given bucket and object.
 func postRestoreOpts(ctx context.Context, r *http.Request, bucket, object string) (opts ObjectOptions, err error) {
-	versioned := globalBucketVersioningSys.Enabled(bucket)
-	versionSuspended := globalBucketVersioningSys.Suspended(bucket)
+	versioned := globalBucketVersioningSys.PrefixEnabled(bucket, object)
+	versionSuspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
 	vid := strings.TrimSpace(r.Form.Get(xhttp.VersionID))
 	if vid != "" && vid != nullVersionID {
 		_, err := uuid.Parse(vid)
@@ -627,8 +627,8 @@ func putRestoreOpts(bucket, object string, rreq *RestoreObjectRequest, objInfo O
 			meta[xhttp.AmzServerSideEncryption] = xhttp.AmzEncryptionAES
 		}
 		return ObjectOptions{
-			Versioned:        globalBucketVersioningSys.Enabled(bucket),
-			VersionSuspended: globalBucketVersioningSys.Suspended(bucket),
+			Versioned:        globalBucketVersioningSys.PrefixEnabled(bucket, object),
+			VersionSuspended: globalBucketVersioningSys.PrefixSuspended(bucket, object),
 			UserDefined:      meta,
 		}
 	}
@@ -640,8 +640,8 @@ func putRestoreOpts(bucket, object string, rreq *RestoreObjectRequest, objInfo O
 	}
 
 	return ObjectOptions{
-		Versioned:        globalBucketVersioningSys.Enabled(bucket),
-		VersionSuspended: globalBucketVersioningSys.Suspended(bucket),
+		Versioned:        globalBucketVersioningSys.PrefixEnabled(bucket, object),
+		VersionSuspended: globalBucketVersioningSys.PrefixSuspended(bucket, object),
 		UserDefined:      meta,
 		VersionID:        objInfo.VersionID,
 		MTime:            objInfo.ModTime,

cmd/bucket-metadata-sys.go

@@ -196,7 +196,7 @@ func (sys *BucketMetadataSys) Get(bucket string) (BucketMetadata, error) {
 func (sys *BucketMetadataSys) GetVersioningConfig(bucket string) (*versioning.Versioning, error) {
 	meta, err := sys.GetConfig(GlobalContext, bucket)
 	if err != nil {
-		return nil, err
+		return &versioning.Versioning{}, err
 	}
 	return meta.versioningConfig, nil
 }

cmd/bucket-replication-utils.go

@@ -509,7 +509,7 @@ func getHealReplicateObjectInfo(objInfo ObjectInfo, rcfg replicationConfig) Repl
 				ObjectName: oi.Name,
 				VersionID:  oi.VersionID,
 			},
-		}, oi, ObjectOptions{}, nil)
+		}, oi, ObjectOptions{VersionSuspended: globalBucketVersioningSys.PrefixSuspended(oi.Bucket, oi.Name)}, nil)
 	} else {
 		dsc = mustReplicate(GlobalContext, oi.Bucket, oi.Name, getMustReplicateOptions(ObjectInfo{
 			UserDefined: oi.UserDefined,

cmd/bucket-replication.go

@@ -180,6 +180,11 @@ func mustReplicate(ctx context.Context, bucket, object string, mopts mustReplica
 		return
 	}
 
+	// Disable server-side replication on object prefixes which are excluded
+	// from versioning via the MinIO bucket versioning extension.
+	if globalBucketVersioningSys.PrefixSuspended(bucket, object) {
+		return
+	}
 	replStatus := mopts.ReplicationStatus()
 	if replStatus == replication.Replica && !mopts.isMetadataReplication() {
 		return
@@ -264,6 +269,11 @@ func checkReplicateDelete(ctx context.Context, bucket string, dobj ObjectToDelet
 	if delOpts.ReplicationRequest {
 		return
 	}
+	// Skip replication if this object's prefix is excluded from being
+	// versioned.
+	if !delOpts.Versioned {
+		return
+	}
 	opts := replication.ObjectOpts{
 		Name:         dobj.ObjectName,
 		SSEC:         crypto.SSEC.IsEncrypted(oi.UserDefined),
@@ -457,8 +467,8 @@ func replicateDelete(ctx context.Context, dobj DeletedObjectReplicationInfo, obj
 		VersionID:         versionID,
 		MTime:             dobj.DeleteMarkerMTime.Time,
 		DeleteReplication: drs,
-		Versioned:         globalBucketVersioningSys.Enabled(bucket),
-		VersionSuspended:  globalBucketVersioningSys.Suspended(bucket),
+		Versioned:         globalBucketVersioningSys.PrefixEnabled(bucket, dobj.ObjectName),
+		VersionSuspended:  globalBucketVersioningSys.PrefixSuspended(bucket, dobj.ObjectName),
 	})
 	if err != nil && !isErrVersionNotFound(err) { // VersionNotFound would be reported by pool that object version is missing on.
 		logger.LogIf(ctx, fmt.Errorf("Unable to update replication metadata for %s/%s(%s): %s", bucket, dobj.ObjectName, versionID, err))
@@ -1590,7 +1600,11 @@ func proxyGetToReplicationTarget(ctx context.Context, bucket, object string, rs
 	return reader, proxyResult{Proxy: true}, nil
 }
 
-func getproxyTargets(ctx context.Context, bucket, object string, opts ObjectOptions) (tgts *madmin.BucketTargets) {
+func getProxyTargets(ctx context.Context, bucket, object string, opts ObjectOptions) (tgts *madmin.BucketTargets) {
+	if opts.VersionSuspended {
+		return &madmin.BucketTargets{}
+	}
+
 	cfg, err := getReplicationConfig(ctx, bucket)
 	if err != nil || cfg == nil {
 		return &madmin.BucketTargets{}

cmd/bucket-versioning-handler.go

@@ -18,12 +18,14 @@
 package cmd
 
 import (
+	"encoding/base64"
 	"encoding/xml"
 	"io"
 	"net/http"
 
 	humanize "github.com/dustin/go-humanize"
 	"github.com/gorilla/mux"
+	"github.com/minio/madmin-go"
 	"github.com/minio/minio/internal/bucket/versioning"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/pkg/bucket/policy"
@@ -63,16 +65,16 @@ func (api objectAPIHandlers) PutBucketVersioningHandler(w http.ResponseWriter, r
 		return
 	}
 
-	if globalSiteReplicationSys.isEnabled() {
+	if globalSiteReplicationSys.isEnabled() && !v.Enabled() {
 		writeErrorResponse(ctx, w, APIError{
 			Code:           "InvalidBucketState",
-			Description:    "Cluster replication is enabled for this site, so the versioning state cannot be changed.",
+			Description:    "Cluster replication is enabled for this site, so the versioning cannot be suspended.",
 			HTTPStatusCode: http.StatusConflict,
 		}, r.URL)
 		return
 	}
 
-	if rcfg, _ := globalBucketObjectLockSys.Get(bucket); rcfg.LockEnabled && v.Suspended() {
+	if rcfg, _ := globalBucketObjectLockSys.Get(bucket); rcfg.LockEnabled && (v.Suspended() || v.PrefixesExcluded()) {
 		writeErrorResponse(ctx, w, APIError{
 			Code:           "InvalidBucketState",
 			Description:    "An Object Lock configuration is present on this bucket, so the versioning state cannot be changed.",
@@ -100,6 +102,20 @@ func (api objectAPIHandlers) PutBucketVersioningHandler(w http.ResponseWriter, r
 		return
 	}
 
+	// Call site replication hook.
+	//
+	// We encode the xml bytes as base64 to ensure there are no encoding
+	// errors.
+	cfgStr := base64.StdEncoding.EncodeToString(configData)
+	if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
+		Type:       madmin.SRBucketMetaTypeVersionConfig,
+		Bucket:     bucket,
+		Versioning: &cfgStr,
+	}); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
 	writeSuccessResponseHeadersOnly(w)
 }
 

cmd/bucket-versioning.go

@@ -31,6 +31,17 @@ func (sys *BucketVersioningSys) Enabled(bucket string) bool {
 	return vc.Enabled()
 }
 
+// PrefixEnabled returns true is versioning is enabled at bucket level and if
+// the given prefix doesn't match any excluded prefixes pattern. This is
+// part of a MinIO versioning configuration extension.
+func (sys *BucketVersioningSys) PrefixEnabled(bucket, prefix string) bool {
+	vc, err := globalBucketMetadataSys.GetVersioningConfig(bucket)
+	if err != nil {
+		return false
+	}
+	return vc.PrefixEnabled(prefix)
+}
+
 // Suspended suspended versioning?
 func (sys *BucketVersioningSys) Suspended(bucket string) bool {
 	vc, err := globalBucketMetadataSys.GetVersioningConfig(bucket)
@@ -40,15 +51,19 @@ func (sys *BucketVersioningSys) Suspended(bucket string) bool {
 	return vc.Suspended()
 }
 
+// PrefixSuspended returns true if the given prefix matches an excluded prefix
+// pattern. This is part of a MinIO versioning configuration extension.
+func (sys *BucketVersioningSys) PrefixSuspended(bucket, prefix string) bool {
+	vc, err := globalBucketMetadataSys.GetVersioningConfig(bucket)
+	if err != nil {
+		return false
+	}
+
+	return vc.PrefixSuspended(prefix)
+}
+
 // Get returns stored bucket policy
 func (sys *BucketVersioningSys) Get(bucket string) (*versioning.Versioning, error) {
-	if globalIsGateway {
-		objAPI := newObjectLayerFn()
-		if objAPI == nil {
-			return nil, errServerNotInitialized
-		}
-		return nil, NotImplemented{}
-	}
 	return globalBucketMetadataSys.GetVersioningConfig(bucket)
 }
 

cmd/config-current.go

@@ -108,7 +108,7 @@ func initHelp() {
 		},
 		config.HelpKV{
 			Key:         config.PolicyOPASubSys,
-			Description: "[DEPRECATED] enable external OPA for policy enforcement",
+			Description: "enable external OPA for policy enforcement",
 		},
 		config.HelpKV{
 			Key:         config.APISubSys,

cmd/data-scanner.go

@@ -1184,7 +1184,7 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
 		opts.VersionID = obj.VersionID
 	}
 	if opts.VersionID == "" {
-		opts.Versioned = globalBucketVersioningSys.Enabled(obj.Bucket)
+		opts.Versioned = globalBucketVersioningSys.PrefixEnabled(obj.Bucket, obj.Name)
 	}
 
 	obj, err := objLayer.DeleteObject(ctx, obj.Bucket, obj.Name, opts)
@@ -1243,6 +1243,9 @@ func (i *scannerItem) objectPath() string {
 // healReplication will heal a scanned item that has failed replication.
 func (i *scannerItem) healReplication(ctx context.Context, o ObjectLayer, oi ObjectInfo, sizeS *sizeSummary) {
 	roi := getHealReplicateObjectInfo(oi, i.replication)
+	if !roi.Dsc.ReplicateAny() {
+		return
+	}
 
 	if oi.DeleteMarker || !oi.VersionPurgeStatus.Empty() {
 		// heal delete marker replication failure or versioned delete replication failure

cmd/disk-cache-backend.go

@@ -884,6 +884,7 @@ func (c *diskCache) put(ctx context.Context, bucket, object string, data io.Read
 	}
 
 	if err := os.MkdirAll(cachePath, 0o777); err != nil {
+		removeAll(cachePath)
 		return oi, err
 	}
 	metadata := cloneMSS(opts.UserDefined)
@@ -892,6 +893,7 @@ func (c *diskCache) put(ctx context.Context, bucket, object string, data io.Read
 	if globalCacheKMS != nil {
 		reader, err = newCacheEncryptReader(data, bucket, object, metadata)
 		if err != nil {
+			removeAll(cachePath)
 			return oi, err
 		}
 		actualSize, _ = sio.EncryptedSize(uint64(size))

cmd/disk-cache.go

@@ -33,7 +33,6 @@ import (
 	"github.com/minio/minio/internal/config/cache"
 	"github.com/minio/minio/internal/disk"
 	"github.com/minio/minio/internal/hash"
-	xhttp "github.com/minio/minio/internal/http"
 	"github.com/minio/minio/internal/logger"
 	"github.com/minio/minio/internal/sync/errgroup"
 	"github.com/minio/pkg/wildcard"
@@ -739,35 +738,30 @@ func (c *cacheObjects) PutObject(ctx context.Context, bucket, object string, r *
 	// Initialize pipe to stream data to cache
 	rPipe, wPipe := io.Pipe()
 	infoCh := make(chan ObjectInfo)
-	errorCh := make(chan error)
 	go func() {
+		defer close(infoCh)
 		info, err := putObjectFn(ctx, bucket, object, NewPutObjReader(hashReader), opts)
-		if err != nil {
-			close(infoCh)
-			pipeReader.CloseWithError(err)
-			rPipe.CloseWithError(err)
-			errorCh <- err
-			return
+		pipeReader.CloseWithError(err)
+		rPipe.CloseWithError(err)
+		if err == nil {
+			infoCh <- info
 		}
-		close(errorCh)
-		infoCh <- info
 	}()
 
 	go func() {
 		_, err := dcache.put(lkctx.Context(), bucket, object, rPipe, r.Size(), nil, opts, false, false)
 		if err != nil {
-			rPipe.CloseWithError(err)
-			return
+			logger.LogIf(lkctx.Context(), err)
 		}
+		// We do not care about errors to cached backend.
+		rPipe.Close()
 	}()
 
 	mwriter := cacheMultiWriter(pipeWriter, wPipe)
 	_, err = io.Copy(mwriter, r)
 	pipeWriter.Close()
 	wPipe.Close()
-
 	if err != nil {
-		err = <-errorCh
 		return ObjectInfo{}, err
 	}
 	info := <-infoCh
@@ -803,8 +797,7 @@ func (c *cacheObjects) uploadObject(ctx context.Context, oi ObjectInfo) {
 		return
 	}
 	var opts ObjectOptions
-	opts.UserDefined = make(map[string]string)
-	opts.UserDefined[xhttp.ContentMD5] = oi.UserDefined["content-md5"]
+	opts.UserDefined = cloneMSS(oi.UserDefined)
 	objInfo, err := c.InnerPutObjectFn(ctx, oi.Bucket, oi.Name, NewPutObjReader(hashReader), opts)
 	wbCommitStatus := CommitComplete
 	size := objInfo.Size

cmd/erasure-metadata.go

@@ -107,7 +107,7 @@ func (fi FileInfo) IsValid() bool {
 func (fi FileInfo) ToObjectInfo(bucket, object string) ObjectInfo {
 	object = decodeDirObject(object)
 	versionID := fi.VersionID
-	if (globalBucketVersioningSys.Enabled(bucket) || globalBucketVersioningSys.Suspended(bucket)) && versionID == "" {
+	if (globalBucketVersioningSys.PrefixEnabled(bucket, object) || globalBucketVersioningSys.PrefixSuspended(bucket, object)) && versionID == "" {
 		versionID = nullVersionID
 	}
 

cmd/erasure-multipart.go

@@ -286,8 +286,10 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, objec
 // disks. `uploads.json` carries metadata regarding on-going multipart
 // operation(s) on the object.
 func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string, object string, opts ObjectOptions) (string, error) {
+	userDefined := cloneMSS(opts.UserDefined)
+
 	onlineDisks := er.getDisks()
-	parityDrives := globalStorageClass.GetParityForSC(opts.UserDefined[xhttp.AmzStorageClass])
+	parityDrives := globalStorageClass.GetParityForSC(userDefined[xhttp.AmzStorageClass])
 	if parityDrives <= 0 {
 		parityDrives = er.defaultParityCount
 	}
@@ -308,7 +310,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 		}
 	}
 	if parityOrig != parityDrives {
-		opts.UserDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
+		userDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
 	}
 
 	dataDrives := len(onlineDisks) - parityDrives
@@ -336,8 +338,8 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	}
 
 	// Guess content-type from the extension if possible.
-	if opts.UserDefined["content-type"] == "" {
-		opts.UserDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
+	if userDefined["content-type"] == "" {
+		userDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
 	}
 
 	modTime := opts.MTime
@@ -352,7 +354,7 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 	for index := range partsMetadata {
 		partsMetadata[index].Fresh = true
 		partsMetadata[index].ModTime = modTime
-		partsMetadata[index].Metadata = opts.UserDefined
+		partsMetadata[index].Metadata = userDefined
 	}
 
 	uploadID := mustGetUUID()
@@ -375,10 +377,6 @@ func (er erasureObjects) newMultipartUpload(ctx context.Context, bucket string,
 func (er erasureObjects) NewMultipartUpload(ctx context.Context, bucket, object string, opts ObjectOptions) (string, error) {
 	auditObjectErasureSet(ctx, object, &er)
 
-	// No metadata is set, allocate a new one.
-	if opts.UserDefined == nil {
-		opts.UserDefined = make(map[string]string)
-	}
 	return er.newMultipartUpload(ctx, bucket, object, opts)
 }
 

cmd/erasure-object.go

@@ -842,17 +842,14 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 
 	data := r.Reader
 
-	// No metadata is set, allocate a new one.
-	if opts.UserDefined == nil {
-		opts.UserDefined = make(map[string]string)
-	}
+	userDefined := cloneMSS(opts.UserDefined)
 
 	storageDisks := er.getDisks()
 
 	parityDrives := len(storageDisks) / 2
 	if !opts.MaxParity {
 		// Get parity and data drive count based on storage class metadata
-		parityDrives = globalStorageClass.GetParityForSC(opts.UserDefined[xhttp.AmzStorageClass])
+		parityDrives = globalStorageClass.GetParityForSC(userDefined[xhttp.AmzStorageClass])
 		if parityDrives <= 0 {
 			parityDrives = er.defaultParityCount
 		}
@@ -890,7 +887,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			parityDrives = len(storageDisks) / 2
 		}
 		if parityOrig != parityDrives {
-			opts.UserDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
+			userDefined[minIOErasureUpgraded] = strconv.Itoa(parityOrig) + "->" + strconv.Itoa(parityDrives)
 		}
 	}
 	dataDrives := len(storageDisks) - parityDrives
@@ -1066,13 +1063,13 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			Hash:       bitrotWriterSum(w),
 		})
 	}
-	if opts.UserDefined["etag"] == "" {
-		opts.UserDefined["etag"] = r.MD5CurrentHexString()
+	if userDefined["etag"] == "" {
+		userDefined["etag"] = r.MD5CurrentHexString()
 	}
 
 	// Guess content-type from the extension if possible.
-	if opts.UserDefined["content-type"] == "" {
-		opts.UserDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
+	if userDefined["content-type"] == "" {
+		userDefined["content-type"] = mimedb.TypeByExtension(path.Ext(object))
 	}
 
 	modTime := opts.MTime
@@ -1083,7 +1080,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 	// Fill all the necessary metadata.
 	// Update `xl.meta` content on each disks.
 	for index := range partsMetadata {
-		partsMetadata[index].Metadata = opts.UserDefined
+		partsMetadata[index].Metadata = userDefined
 		partsMetadata[index].Size = n
 		partsMetadata[index].ModTime = modTime
 	}
@@ -1190,7 +1187,13 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
 		// VersionID is not set means delete is not specific about
 		// any version, look for if the bucket is versioned or not.
 		if objects[i].VersionID == "" {
-			if opts.Versioned || opts.VersionSuspended {
+			// MinIO extension to bucket version configuration
+			suspended := opts.VersionSuspended
+			versioned := opts.Versioned
+			if opts.PrefixEnabledFn != nil {
+				versioned = opts.PrefixEnabledFn(objects[i].ObjectName)
+			}
+			if versioned || suspended {
 				// Bucket is versioned and no version was explicitly
 				// mentioned for deletes, create a delete marker instead.
 				vr.ModTime = UTCNow()
@@ -1198,7 +1201,7 @@ func (er erasureObjects) DeleteObjects(ctx context.Context, bucket string, objec
 				// Versioning suspended means that we add a `null` version
 				// delete marker, if not add a new version for this delete
 				// marker.
-				if opts.Versioned {
+				if versioned {
 					vr.VersionID = mustGetUUID()
 				}
 			}

cmd/erasure-server-pool-decom.go

@@ -619,7 +619,6 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 
 	parallelWorkers := make(chan struct{}, workerSize)
 
-	versioned := globalBucketVersioningSys.Enabled(bName)
 	for _, set := range pool.sets {
 		set := set
 		disks := set.getOnlineDisks()
@@ -629,6 +628,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 			continue
 		}
 
+		vc, _ := globalBucketVersioningSys.Get(bName)
 		decommissionEntry := func(entry metaCacheEntry) {
 			defer func() {
 				<-parallelWorkers
@@ -667,7 +667,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 						bName,
 						version.Name,
 						ObjectOptions{
-							Versioned:         versioned,
+							Versioned:         vc.PrefixEnabled(version.Name),
 							VersionID:         version.VersionID,
 							MTime:             version.ModTime,
 							DeleteReplication: version.ReplicationState,

cmd/gateway/s3/gateway-s3-sse.go

@@ -364,6 +364,7 @@ func (l *s3EncObjects) GetObjectInfo(ctx context.Context, bucket string, object
 // CopyObject copies an object from source bucket to a destination bucket.
 func (l *s3EncObjects) CopyObject(ctx context.Context, srcBucket string, srcObject string, dstBucket string, dstObject string, srcInfo minio.ObjectInfo, s, d minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
 	cpSrcDstSame := path.Join(srcBucket, srcObject) == path.Join(dstBucket, dstObject)
+	userDefined := minio.CloneMSS(srcInfo.UserDefined)
 	if cpSrcDstSame {
 		var gwMeta gwMetaV1
 		if s.ServerSideEncryption != nil && d.ServerSideEncryption != nil &&
@@ -378,16 +379,16 @@ func (l *s3EncObjects) CopyObject(ctx context.Context, srcBucket string, srcObje
 				d.ServerSideEncryption.Marshal(header)
 			}
 			for k, v := range header {
-				srcInfo.UserDefined[k] = v[0]
+				userDefined[k] = v[0]
 			}
-			gwMeta.Meta = srcInfo.UserDefined
+			gwMeta.Meta = userDefined
 			if err = l.writeGWMetadata(ctx, dstBucket, getDareMetaPath(dstObject), gwMeta, minio.ObjectOptions{}); err != nil {
 				return objInfo, minio.ErrorRespToObjectError(err)
 			}
 			return gwMeta.ToObjectInfo(dstBucket, dstObject), nil
 		}
 	}
-	dstOpts := minio.ObjectOptions{ServerSideEncryption: d.ServerSideEncryption, UserDefined: srcInfo.UserDefined}
+	dstOpts := minio.ObjectOptions{ServerSideEncryption: d.ServerSideEncryption, UserDefined: userDefined}
 	return l.PutObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, dstOpts)
 }
 

cmd/gateway/s3/gateway-s3.go

@@ -502,17 +502,20 @@ func (l *s3Objects) GetObjectInfo(ctx context.Context, bucket string, object str
 // PutObject creates a new object with the incoming data,
 func (l *s3Objects) PutObject(ctx context.Context, bucket string, object string, r *minio.PutObjReader, opts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
 	data := r.Reader
+
+	userDefined := minio.CloneMSS(opts.UserDefined)
+
 	var tagMap map[string]string
-	if tagstr, ok := opts.UserDefined[xhttp.AmzObjectTagging]; ok && tagstr != "" {
+	if tagstr, ok := userDefined[xhttp.AmzObjectTagging]; ok && tagstr != "" {
 		tagObj, err := tags.ParseObjectTags(tagstr)
 		if err != nil {
 			return objInfo, minio.ErrorRespToObjectError(err, bucket, object)
 		}
 		tagMap = tagObj.ToMap()
-		delete(opts.UserDefined, xhttp.AmzObjectTagging)
+		delete(userDefined, xhttp.AmzObjectTagging)
 	}
 	putOpts := miniogo.PutObjectOptions{
-		UserMetadata:         opts.UserDefined,
+		UserMetadata:         userDefined,
 		ServerSideEncryption: opts.ServerSideEncryption,
 		UserTags:             tagMap,
 		// Content-Md5 is needed for buckets with object locking,
@@ -529,7 +532,7 @@ func (l *s3Objects) PutObject(ctx context.Context, bucket string, object string,
 		ETag:     ui.ETag,
 		Size:     ui.Size,
 		Key:      object,
-		Metadata: minio.ToMinioClientObjectInfoMetadata(opts.UserDefined),
+		Metadata: minio.ToMinioClientObjectInfoMetadata(userDefined),
 	}
 
 	return minio.FromMinioClientObjectInfo(bucket, oi), nil
@@ -544,8 +547,9 @@ func (l *s3Objects) CopyObject(ctx context.Context, srcBucket string, srcObject
 	// metadata input is already a trickled down value from interpreting x-amz-metadata-directive at
 	// handler layer. So what we have right now is supposed to be applied on the destination object anyways.
 	// So preserve it by adding "REPLACE" directive to save all the metadata set by CopyObject API.
-	srcInfo.UserDefined["x-amz-metadata-directive"] = "REPLACE"
-	srcInfo.UserDefined["x-amz-copy-source-if-match"] = srcInfo.ETag
+	userDefined := minio.CloneMSS(srcInfo.UserDefined)
+	userDefined["x-amz-metadata-directive"] = "REPLACE"
+	userDefined["x-amz-copy-source-if-match"] = srcInfo.ETag
 	header := make(http.Header)
 	if srcOpts.ServerSideEncryption != nil {
 		encrypt.SSECopy(srcOpts.ServerSideEncryption).Marshal(header)
@@ -556,10 +560,10 @@ func (l *s3Objects) CopyObject(ctx context.Context, srcBucket string, srcObject
 	}
 
 	for k, v := range header {
-		srcInfo.UserDefined[k] = v[0]
+		userDefined[k] = v[0]
 	}
 
-	if _, err = l.Client.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo.UserDefined, miniogo.CopySrcOptions{}, miniogo.PutObjectOptions{}); err != nil {
+	if _, err = l.Client.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, userDefined, miniogo.CopySrcOptions{}, miniogo.PutObjectOptions{}); err != nil {
 		return objInfo, minio.ErrorRespToObjectError(err, srcBucket, srcObject)
 	}
 	return l.GetObjectInfo(ctx, dstBucket, dstObject, dstOpts)
@@ -605,17 +609,18 @@ func (l *s3Objects) ListMultipartUploads(ctx context.Context, bucket string, pre
 // NewMultipartUpload upload object in multiple parts
 func (l *s3Objects) NewMultipartUpload(ctx context.Context, bucket string, object string, o minio.ObjectOptions) (uploadID string, err error) {
 	var tagMap map[string]string
-	if tagStr, ok := o.UserDefined[xhttp.AmzObjectTagging]; ok {
+	userDefined := minio.CloneMSS(o.UserDefined)
+	if tagStr, ok := userDefined[xhttp.AmzObjectTagging]; ok {
 		tagObj, err := tags.Parse(tagStr, true)
 		if err != nil {
 			return uploadID, minio.ErrorRespToObjectError(err, bucket, object)
 		}
 		tagMap = tagObj.ToMap()
-		delete(o.UserDefined, xhttp.AmzObjectTagging)
+		delete(userDefined, xhttp.AmzObjectTagging)
 	}
 	// Create PutObject options
 	opts := miniogo.PutObjectOptions{
-		UserMetadata:         o.UserDefined,
+		UserMetadata:         userDefined,
 		ServerSideEncryption: o.ServerSideEncryption,
 		UserTags:             tagMap,
 	}
@@ -645,9 +650,8 @@ func (l *s3Objects) CopyObjectPart(ctx context.Context, srcBucket, srcObject, de
 	if srcOpts.CheckPrecondFn != nil && srcOpts.CheckPrecondFn(srcInfo) {
 		return minio.PartInfo{}, minio.PreConditionFailed{}
 	}
-	srcInfo.UserDefined = map[string]string{
-		"x-amz-copy-source-if-match": srcInfo.ETag,
-	}
+	userDefined := minio.CloneMSS(srcInfo.UserDefined)
+	userDefined["x-amz-copy-source-if-match"] = srcInfo.ETag
 	header := make(http.Header)
 	if srcOpts.ServerSideEncryption != nil {
 		encrypt.SSECopy(srcOpts.ServerSideEncryption).Marshal(header)
@@ -657,11 +661,11 @@ func (l *s3Objects) CopyObjectPart(ctx context.Context, srcBucket, srcObject, de
 		dstOpts.ServerSideEncryption.Marshal(header)
 	}
 	for k, v := range header {
-		srcInfo.UserDefined[k] = v[0]
+		userDefined[k] = v[0]
 	}
 
 	completePart, err := l.Client.CopyObjectPart(ctx, srcBucket, srcObject, destBucket, destObject,
-		uploadID, partID, startOffset, length, srcInfo.UserDefined)
+		uploadID, partID, startOffset, length, userDefined)
 	if err != nil {
 		return p, minio.ErrorRespToObjectError(err, srcBucket, srcObject)
 	}

cmd/object-api-interface.go

@@ -76,6 +76,8 @@ type ObjectOptions struct {
 	// Mutate set to 'true' if the call is namespace mutation call
 	Mutate        bool
 	WalkAscending bool // return Walk results in ascending order of versions
+
+	PrefixEnabledFn func(prefix string) bool // function which returns true if versioning is enabled on prefix
 }
 
 // ExpirationOptions represents object options for object expiration at objectLayer.

cmd/object-api-listobjects_test.go

@@ -70,7 +70,7 @@ func testListObjectsVersionedFolders(obj ObjectLayer, instanceType string, t1 Te
 		md5Bytes := md5.Sum([]byte(object.content))
 		_, err = obj.PutObject(context.Background(), object.parentBucket, object.name, mustGetPutObjReader(t, bytes.NewBufferString(object.content),
 			int64(len(object.content)), hex.EncodeToString(md5Bytes[:]), ""), ObjectOptions{
-			Versioned:   globalBucketVersioningSys.Enabled(object.parentBucket),
+			Versioned:   globalBucketVersioningSys.PrefixEnabled(object.parentBucket, object.name),
 			UserDefined: object.meta,
 		})
 		if err != nil {
@@ -78,7 +78,7 @@ func testListObjectsVersionedFolders(obj ObjectLayer, instanceType string, t1 Te
 		}
 		if object.addDeleteMarker {
 			oi, err := obj.DeleteObject(context.Background(), object.parentBucket, object.name, ObjectOptions{
-				Versioned: globalBucketVersioningSys.Enabled(object.parentBucket),
+				Versioned: globalBucketVersioningSys.PrefixEnabled(object.parentBucket, object.name),
 			})
 			if err != nil {
 				t.Fatalf("%s : %s", instanceType, err.Error())
@@ -380,7 +380,7 @@ func _testListObjects(obj ObjectLayer, instanceType string, t1 TestErrHandler, v
 		_, err = obj.PutObject(context.Background(), object.parentBucket, object.name,
 			mustGetPutObjReader(t, bytes.NewBufferString(object.content),
 				int64(len(object.content)), hex.EncodeToString(md5Bytes[:]), ""), ObjectOptions{
-				Versioned:   globalBucketVersioningSys.Enabled(object.parentBucket),
+				Versioned:   globalBucketVersioningSys.PrefixEnabled(object.parentBucket, object.name),
 				UserDefined: object.meta,
 			})
 		if err != nil {
@@ -1065,16 +1065,16 @@ func testDeleteObjectVersion(obj ObjectLayer, instanceType string, t1 TestErrHan
 		_, err := obj.PutObject(context.Background(), object.parentBucket, object.name,
 			mustGetPutObjReader(t, bytes.NewBufferString(object.content),
 				int64(len(object.content)), hex.EncodeToString(md5Bytes[:]), ""), ObjectOptions{
-				Versioned:        globalBucketVersioningSys.Enabled(object.parentBucket),
-				VersionSuspended: globalBucketVersioningSys.Suspended(object.parentBucket),
+				Versioned:        globalBucketVersioningSys.PrefixEnabled(object.parentBucket, object.name),
+				VersionSuspended: globalBucketVersioningSys.PrefixSuspended(object.parentBucket, object.name),
 				UserDefined:      object.meta,
 			})
 		if err != nil {
 			t.Fatalf("%s : %s", instanceType, err)
 		}
 		obj, err := obj.DeleteObject(context.Background(), object.parentBucket, object.name, ObjectOptions{
-			Versioned:        globalBucketVersioningSys.Enabled(object.parentBucket),
-			VersionSuspended: globalBucketVersioningSys.Suspended(object.parentBucket),
+			Versioned:        globalBucketVersioningSys.PrefixEnabled(object.parentBucket, object.name),
+			VersionSuspended: globalBucketVersioningSys.PrefixSuspended(object.parentBucket, object.name),
 			VersionID:        object.versionID,
 		})
 		if err != nil {

cmd/object-api-options.go

@@ -157,16 +157,16 @@ func getOpts(ctx context.Context, r *http.Request, bucket, object string) (Objec
 			}
 		}
 	}
+	opts.VersionSuspended = globalBucketVersioningSys.PrefixSuspended(bucket, object)
 	return opts, nil
 }
 
 func delOpts(ctx context.Context, r *http.Request, bucket, object string) (opts ObjectOptions, err error) {
-	versioned := globalBucketVersioningSys.Enabled(bucket)
 	opts, err = getOpts(ctx, r, bucket, object)
 	if err != nil {
 		return opts, err
 	}
-	opts.Versioned = versioned
+	opts.Versioned = globalBucketVersioningSys.PrefixEnabled(bucket, object)
 	opts.VersionSuspended = globalBucketVersioningSys.Suspended(bucket)
 	delMarker := strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceDeleteMarker))
 	if delMarker != "" {
@@ -203,8 +203,9 @@ func delOpts(ctx context.Context, r *http.Request, bucket, object string) (opts
 
 // get ObjectOptions for PUT calls from encryption headers and metadata
 func putOpts(ctx context.Context, r *http.Request, bucket, object string, metadata map[string]string) (opts ObjectOptions, err error) {
-	versioned := globalBucketVersioningSys.Enabled(bucket)
-	versionSuspended := globalBucketVersioningSys.Suspended(bucket)
+	versioned := globalBucketVersioningSys.PrefixEnabled(bucket, object)
+	versionSuspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
+
 	vid := strings.TrimSpace(r.Form.Get(xhttp.VersionID))
 	if vid != "" && vid != nullVersionID {
 		_, err := uuid.Parse(vid)

cmd/object-handlers-common.go

@@ -266,8 +266,6 @@ func setPutObjHeaders(w http.ResponseWriter, objInfo ObjectInfo, delete bool) {
 }
 
 func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toDel []ObjectToDelete) {
-	versioned := globalBucketVersioningSys.Enabled(bucket)
-	versionSuspended := globalBucketVersioningSys.Suspended(bucket)
 	for remaining := toDel; len(remaining) > 0; toDel = remaining {
 		if len(toDel) > maxDeleteList {
 			remaining = toDel[maxDeleteList:]
@@ -275,9 +273,10 @@ func deleteObjectVersions(ctx context.Context, o ObjectLayer, bucket string, toD
 		} else {
 			remaining = nil
 		}
+		vc, _ := globalBucketVersioningSys.Get(bucket)
 		deletedObjs, errs := o.DeleteObjects(ctx, bucket, toDel, ObjectOptions{
-			Versioned:        versioned,
-			VersionSuspended: versionSuspended,
+			PrefixEnabledFn:  vc.PrefixEnabled,
+			VersionSuspended: vc.Suspended(),
 		})
 		var logged bool
 		for i, err := range errs {

cmd/object-handlers.go

@@ -204,7 +204,7 @@ func (api objectAPIHandlers) SelectObjectContentHandler(w http.ResponseWriter, r
 
 	objInfo, err := getObjectInfo(ctx, bucket, object, opts)
 	if err != nil {
-		if globalBucketVersioningSys.Enabled(bucket) {
+		if globalBucketVersioningSys.PrefixEnabled(bucket, object) {
 			// Versioning enabled quite possibly object is deleted might be delete-marker
 			// if present set the headers, no idea why AWS S3 sets these headers.
 			if objInfo.VersionID != "" && objInfo.DeleteMarker {
@@ -425,7 +425,7 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 			proxy  proxyResult
 			perr   error
 		)
-		proxytgts := getproxyTargets(ctx, bucket, object, opts)
+		proxytgts := getProxyTargets(ctx, bucket, object, opts)
 		if !proxytgts.Empty() {
 			// proxy to replication target if active-active replication is in place.
 			reader, proxy, perr = proxyGetToReplicationTarget(ctx, bucket, object, rs, r.Header, opts, proxytgts)
@@ -445,7 +445,7 @@ func (api objectAPIHandlers) getObjectHandler(ctx context.Context, objectAPI Obj
 				writeErrorResponse(ctx, w, toAPIError(ctx, proxy.Err), r.URL)
 				return
 			}
-			if globalBucketVersioningSys.Enabled(bucket) && gr != nil {
+			if globalBucketVersioningSys.PrefixEnabled(bucket, object) && gr != nil {
 				if !gr.ObjInfo.VersionPurgeStatus.Empty() {
 					// Shows the replication status of a permanent delete of a version
 					w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(gr.ObjInfo.VersionPurgeStatus)}
@@ -662,7 +662,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 			oi    ObjectInfo
 		)
 		// proxy HEAD to replication target if active-active replication configured on bucket
-		proxytgts := getproxyTargets(ctx, bucket, object, opts)
+		proxytgts := getProxyTargets(ctx, bucket, object, opts)
 		if !proxytgts.Empty() {
 			if rangeHeader != "" {
 				rs, _ = parseRequestRangeSpec(rangeHeader)
@@ -673,7 +673,7 @@ func (api objectAPIHandlers) headObjectHandler(ctx context.Context, objectAPI Ob
 			}
 		}
 		if !proxy.Proxy {
-			if globalBucketVersioningSys.Enabled(bucket) {
+			if globalBucketVersioningSys.PrefixEnabled(bucket, object) {
 				switch {
 				case !objInfo.VersionPurgeStatus.Empty():
 					w.Header()[xhttp.MinIODeleteReplicationStatus] = []string{string(objInfo.VersionPurgeStatus)}
@@ -1087,7 +1087,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		if isErrPreconditionFailed(err) {
 			return
 		}
-		if globalBucketVersioningSys.Enabled(srcBucket) && gr != nil {
+		if globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject) && gr != nil {
 			// Versioning enabled quite possibly object is deleted might be delete-marker
 			// if present set the headers, no idea why AWS S3 sets these headers.
 			if gr.ObjInfo.VersionID != "" && gr.ObjInfo.DeleteMarker {
@@ -2480,7 +2480,7 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt
 		if isErrPreconditionFailed(err) {
 			return
 		}
-		if globalBucketVersioningSys.Enabled(srcBucket) && gr != nil {
+		if globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject) && gr != nil {
 			// Versioning enabled quite possibly object is deleted might be delete-marker
 			// if present set the headers, no idea why AWS S3 sets these headers.
 			if gr.ObjInfo.VersionID != "" && gr.ObjInfo.DeleteMarker {
@@ -3211,8 +3211,8 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 		w.Write(encodedErrorResponse)
 	}
 
-	versioned := globalBucketVersioningSys.Enabled(bucket)
-	suspended := globalBucketVersioningSys.Suspended(bucket)
+	versioned := globalBucketVersioningSys.PrefixEnabled(bucket, object)
+	suspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
 	os := newObjSweeper(bucket, object).WithVersioning(versioned, suspended)
 	if !globalTierConfigMgr.Empty() {
 		// Get appropriate object info to identify the remote object to delete

cmd/peer-rest-client.go

@@ -903,43 +903,54 @@ func (client *peerRESTClient) Trace(traceCh chan interface{}, doneCh <-chan stru
 	}()
 }
 
+func (client *peerRESTClient) doConsoleLog(logCh chan interface{}, doneCh <-chan struct{}) {
+	// To cancel the REST request in case doneCh gets closed.
+	ctx, cancel := context.WithCancel(GlobalContext)
+
+	cancelCh := make(chan struct{})
+	defer close(cancelCh)
+	go func() {
+		select {
+		case <-doneCh:
+		case <-cancelCh:
+			// There was an error in the REST request.
+		}
+		cancel()
+	}()
+
+	respBody, err := client.callWithContext(ctx, peerRESTMethodLog, nil, nil, -1)
+	defer http.DrainBody(respBody)
+	if err != nil {
+		return
+	}
+
+	dec := gob.NewDecoder(respBody)
+	for {
+		var lg madmin.LogInfo
+		if err = dec.Decode(&lg); err != nil {
+			break
+		}
+		if lg.DeploymentID != "" {
+			select {
+			case logCh <- lg:
+			default:
+				// Do not block on slow receivers.
+			}
+		}
+	}
+}
+
 // ConsoleLog - sends request to peer nodes to get console logs
 func (client *peerRESTClient) ConsoleLog(logCh chan interface{}, doneCh <-chan struct{}) {
 	go func() {
 		for {
-			// get cancellation context to properly unsubscribe peers
-			ctx, cancel := context.WithCancel(GlobalContext)
-			respBody, err := client.callWithContext(ctx, peerRESTMethodLog, nil, nil, -1)
-			if err != nil {
-				// Retry the failed request.
-				time.Sleep(5 * time.Second)
-			} else {
-				dec := gob.NewDecoder(respBody)
-
-				go func() {
-					<-doneCh
-					cancel()
-				}()
-
-				for {
-					var log madmin.LogInfo
-					if err = dec.Decode(&log); err != nil {
-						break
-					}
-					select {
-					case logCh <- log:
-					default:
-					}
-				}
-			}
-
+			client.doConsoleLog(logCh, doneCh)
 			select {
 			case <-doneCh:
-				cancel()
-				http.DrainBody(respBody)
 				return
 			default:
-				// There was error in the REST request, retry.
+				// There was error in the REST request, retry after sometime as probably the peer is down.
+				time.Sleep(5 * time.Second)
 			}
 		}
 	}()

cmd/peer-rest-server.go

@@ -932,6 +932,8 @@ func (s *peerRESTServer) ListenHandler(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			w.(http.Flusher).Flush()
+		case <-r.Context().Done():
+			return
 		case <-keepAliveTicker.C:
 			if err := enc.Encode(&event.Event{}); err != nil {
 				return
@@ -993,6 +995,8 @@ func (s *peerRESTServer) TraceHandler(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			w.(http.Flusher).Flush()
+		case <-r.Context().Done():
+			return
 		case <-keepAliveTicker.C:
 			if err := enc.Encode(&madmin.TraceInfo{}); err != nil {
 				return
@@ -1059,15 +1063,15 @@ func (s *peerRESTServer) ConsoleLogHandler(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	w.Header().Set("Connection", "close")
-	w.WriteHeader(http.StatusOK)
-
 	doneCh := make(chan struct{})
 	defer close(doneCh)
 
 	ch := make(chan interface{}, 2000)
 	globalConsoleSys.Subscribe(ch, doneCh, "", 0, string(logger.All), nil)
 
+	keepAliveTicker := time.NewTicker(500 * time.Millisecond)
+	defer keepAliveTicker.Stop()
+
 	enc := gob.NewEncoder(w)
 	for {
 		select {
@@ -1076,6 +1080,11 @@ func (s *peerRESTServer) ConsoleLogHandler(w http.ResponseWriter, r *http.Reques
 				return
 			}
 			w.(http.Flusher).Flush()
+		case <-keepAliveTicker.C:
+			if err := enc.Encode(&madmin.LogInfo{}); err != nil {
+				return
+			}
+			w.(http.Flusher).Flush()
 		case <-r.Context().Done():
 			return
 		}

cmd/site-replication.go

@@ -126,6 +126,13 @@ func errSRIAMError(err error) SRError {
 	}
 }
 
+func errSRConfigMissingError(err error) SRError {
+	return SRError{
+		Cause: err,
+		Code:  ErrSiteReplicationConfigMissing,
+	}
+}
+
 var errSRObjectLayerNotReady = SRError{
 	Cause: fmt.Errorf("object layer not ready"),
 	Code:  ErrServerNotInitialized,
@@ -1209,6 +1216,23 @@ func (c *SiteReplicationSys) BucketMetaHook(ctx context.Context, item madmin.SRB
 	return cErr.summaryErr
 }
 
+// PeerBucketVersioningHandler - updates versioning config to local cluster.
+func (c *SiteReplicationSys) PeerBucketVersioningHandler(ctx context.Context, bucket string, versioning *string) error {
+	if versioning != nil {
+		configData, err := base64.StdEncoding.DecodeString(*versioning)
+		if err != nil {
+			return wrapSRErr(err)
+		}
+		err = globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, configData)
+		if err != nil {
+			return wrapSRErr(err)
+		}
+		return nil
+	}
+
+	return nil
+}
+
 // PeerBucketPolicyHandler - copies/deletes policy to local cluster.
 func (c *SiteReplicationSys) PeerBucketPolicyHandler(ctx context.Context, bucket string, policy *policy.Policy) error {
 	if policy != nil {
@@ -1848,6 +1872,8 @@ func (c *SiteReplicationSys) isEnabled() bool {
 	return c.enabled
 }
 
+var errMissingSRConfig = fmt.Errorf("Site not found in site replication configuration")
+
 // RemovePeerCluster - removes one or more clusters from site replication configuration.
 func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI ObjectLayer, rreq madmin.SRRemoveReq) (st madmin.ReplicateRemoveStatus, err error) {
 	if !c.isEnabled() {
@@ -1872,7 +1898,7 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
 	for _, s := range siteNames {
 		info, ok := peerMap[s]
 		if !ok {
-			return st, errSRInvalidRequest(fmt.Errorf("Site %s not found in site replication configuration", s))
+			return st, errSRConfigMissingError(errMissingSRConfig)
 		}
 		rmvEndpoints = append(rmvEndpoints, info.Endpoint)
 		delete(updatedPeers, info.DeploymentID)
@@ -1898,6 +1924,9 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
 				return
 			}
 			if _, err = admClient.SRPeerRemove(ctx, rreq); err != nil {
+				if errors.As(err, &errMissingSRConfig) {
+					return
+				}
 				errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err))
 				return
 			}
@@ -1937,6 +1966,10 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
 // InternalRemoveReq - sends an unlink request to peer cluster to remove one or more sites
 // from the site replication configuration.
 func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI ObjectLayer, rreq madmin.SRRemoveReq) error {
+	if !c.isEnabled() {
+		return errSRNotEnabled
+	}
+
 	ourName := ""
 	peerMap := make(map[string]madmin.PeerInfo)
 	updatedPeers := make(map[string]madmin.PeerInfo)
@@ -1958,7 +1991,7 @@ func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI Ob
 	for _, s := range siteNames {
 		info, ok := peerMap[s]
 		if !ok {
-			return fmt.Errorf("Site %s not found in site replication configuration", s)
+			return errMissingSRConfig
 		}
 		if info.DeploymentID == globalDeploymentID {
 			unlinkSelf = true
@@ -2135,6 +2168,7 @@ func (c *SiteReplicationSys) SiteReplicationStatus(ctx context.Context, objAPI O
 	for b, stat := range sinfo.BucketStats {
 		for dID, st := range stat {
 			if st.TagMismatch ||
+				st.VersioningConfigMismatch ||
 				st.OLockConfigMismatch ||
 				st.SSEConfigMismatch ||
 				st.PolicyMismatch ||
@@ -2532,7 +2566,8 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 			replCfgs := make([]*sreplication.Config, numSites)
 			quotaCfgs := make([]*madmin.BucketQuota, numSites)
 			sseCfgSet := set.NewStringSet()
-			var tagCount, olockCfgCount, sseCfgCount int
+			versionCfgSet := set.NewStringSet()
+			var tagCount, olockCfgCount, sseCfgCount, versionCfgCount int
 			for i, s := range slc {
 				if s.ReplicationConfig != nil {
 					cfgBytes, err := base64.StdEncoding.DecodeString(*s.ReplicationConfig)
@@ -2545,6 +2580,16 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 					}
 					replCfgs[i] = cfg
 				}
+				if s.Versioning != nil {
+					configData, err := base64.StdEncoding.DecodeString(*s.Versioning)
+					if err != nil {
+						continue
+					}
+					versionCfgCount++
+					if !versionCfgSet.Contains(string(configData)) {
+						versionCfgSet.Add(string(configData))
+					}
+				}
 				if s.QuotaConfig != nil {
 					cfgBytes, err := base64.StdEncoding.DecodeString(*s.QuotaConfig)
 					if err != nil {
@@ -2574,12 +2619,11 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 					policies[i] = plcy
 				}
 				if s.ObjectLockConfig != nil {
-					olockCfgCount++
 					configData, err := base64.StdEncoding.DecodeString(*s.ObjectLockConfig)
 					if err != nil {
 						continue
 					}
-
+					olockCfgCount++
 					if !olockConfigSet.Contains(string(configData)) {
 						olockConfigSet.Add(string(configData))
 					}
@@ -2589,10 +2633,10 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 					if err != nil {
 						continue
 					}
+					sseCfgCount++
 					if !sseCfgSet.Contains(string(configData)) {
 						sseCfgSet.Add(string(configData))
 					}
-					sseCfgCount++
 				}
 				ss, ok := info.StatsSummary[s.DeploymentID]
 				if !ok {
@@ -2612,6 +2656,9 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 				if sseCfgCount > 0 {
 					ss.TotalSSEConfigCount++
 				}
+				if versionCfgCount > 0 {
+					ss.TotalVersioningConfigCount++
+				}
 				if len(policies) > 0 {
 					ss.TotalBucketPoliciesCount++
 				}
@@ -2620,6 +2667,7 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 			tagMismatch := !isReplicated(tagCount, numSites, tagSet)
 			olockCfgMismatch := !isReplicated(olockCfgCount, numSites, olockConfigSet)
 			sseCfgMismatch := !isReplicated(sseCfgCount, numSites, sseCfgSet)
+			versionCfgMismatch := !isReplicated(versionCfgCount, numSites, versionCfgSet)
 			policyMismatch := !isBktPolicyReplicated(numSites, policies)
 			replCfgMismatch := !isBktReplCfgReplicated(numSites, replCfgs)
 			quotaCfgMismatch := !isBktQuotaCfgReplicated(numSites, quotaCfgs)
@@ -2632,20 +2680,21 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 				}
 				quotaCfgSet := hasBucket && *quotaCfgs[i] != madmin.BucketQuota{}
 				ss := madmin.SRBucketStatsSummary{
-					DeploymentID:           s.DeploymentID,
-					HasBucket:              hasBucket,
-					TagMismatch:            tagMismatch,
-					OLockConfigMismatch:    olockCfgMismatch,
-					SSEConfigMismatch:      sseCfgMismatch,
-					PolicyMismatch:         policyMismatch,
-					ReplicationCfgMismatch: replCfgMismatch,
-					QuotaCfgMismatch:       quotaCfgMismatch,
-					HasReplicationCfg:      s.ReplicationConfig != nil,
-					HasTagsSet:             s.Tags != nil,
-					HasOLockConfigSet:      s.ObjectLockConfig != nil,
-					HasPolicySet:           s.Policy != nil,
-					HasQuotaCfgSet:         quotaCfgSet,
-					HasSSECfgSet:           s.SSEConfig != nil,
+					DeploymentID:             s.DeploymentID,
+					HasBucket:                hasBucket,
+					TagMismatch:              tagMismatch,
+					OLockConfigMismatch:      olockCfgMismatch,
+					SSEConfigMismatch:        sseCfgMismatch,
+					VersioningConfigMismatch: versionCfgMismatch,
+					PolicyMismatch:           policyMismatch,
+					ReplicationCfgMismatch:   replCfgMismatch,
+					QuotaCfgMismatch:         quotaCfgMismatch,
+					HasReplicationCfg:        s.ReplicationConfig != nil,
+					HasTagsSet:               s.Tags != nil,
+					HasOLockConfigSet:        s.ObjectLockConfig != nil,
+					HasPolicySet:             s.Policy != nil,
+					HasQuotaCfgSet:           quotaCfgSet,
+					HasSSECfgSet:             s.SSEConfig != nil,
 				}
 				var m srBucketMetaInfo
 				if len(bucketStats[s.Bucket]) > dIdx {
@@ -2662,6 +2711,9 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 				if !olockCfgMismatch && olockCfgCount == numSites {
 					sum.ReplicatedLockConfig++
 				}
+				if !versionCfgMismatch && versionCfgCount == numSites {
+					sum.ReplicatedVersioningConfig++
+				}
 				if !sseCfgMismatch && sseCfgCount == numSites {
 					sum.ReplicatedSSEConfig++
 				}
@@ -3352,6 +3404,7 @@ func (c *SiteReplicationSys) healBuckets(ctx context.Context, objAPI ObjectLayer
 
 	for bucket := range info.BucketStats {
 		c.healCreateMissingBucket(ctx, objAPI, bucket, info)
+		c.healVersioningMetadata(ctx, objAPI, bucket, info)
 		c.healOLockConfigMetadata(ctx, objAPI, bucket, info)
 		c.healSSEMetadata(ctx, objAPI, bucket, info)
 		c.healBucketReplicationConfig(ctx, objAPI, bucket, info)
@@ -3572,6 +3625,79 @@ func (c *SiteReplicationSys) healBucketQuotaConfig(ctx context.Context, objAPI O
 	return nil
 }
 
+func (c *SiteReplicationSys) healVersioningMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
+	c.RLock()
+	defer c.RUnlock()
+	if !c.enabled {
+		return nil
+	}
+	var (
+		latestID, latestPeerName string
+		lastUpdate               time.Time
+		latestVersioningConfig   *string
+	)
+
+	bs := info.BucketStats[bucket]
+	for dID, ss := range bs {
+		if lastUpdate.IsZero() {
+			lastUpdate = ss.meta.VersioningConfigUpdatedAt
+			latestID = dID
+			latestVersioningConfig = ss.meta.Versioning
+		}
+		// avoid considering just created buckets as latest. Perhaps this site
+		// just joined cluster replication and yet to be sync'd
+		if ss.meta.CreatedAt.Equal(ss.meta.VersioningConfigUpdatedAt) {
+			continue
+		}
+		if ss.meta.VersioningConfigUpdatedAt.After(lastUpdate) {
+			lastUpdate = ss.meta.VersioningConfigUpdatedAt
+			latestID = dID
+			latestVersioningConfig = ss.meta.Versioning
+		}
+	}
+
+	latestPeerName = info.Sites[latestID].Name
+	var latestVersioningConfigBytes []byte
+	var err error
+	if latestVersioningConfig != nil {
+		latestVersioningConfigBytes, err = base64.StdEncoding.DecodeString(*latestVersioningConfig)
+		if err != nil {
+			return err
+		}
+	}
+
+	for dID, bStatus := range bs {
+		if !bStatus.VersioningConfigMismatch {
+			continue
+		}
+		if isBucketMetadataEqual(latestVersioningConfig, bStatus.meta.Versioning) {
+			continue
+		}
+		if dID == globalDeploymentID {
+			if err := globalBucketMetadataSys.Update(ctx, bucket, bucketSSEConfig, latestVersioningConfigBytes); err != nil {
+				logger.LogIf(ctx, fmt.Errorf("Error healing sse metadata from peer site %s : %w", latestPeerName, err))
+			}
+			continue
+		}
+
+		admClient, err := c.getAdminClient(ctx, dID)
+		if err != nil {
+			return wrapSRErr(err)
+		}
+		peerName := info.Sites[dID].Name
+		err = admClient.SRPeerReplicateBucketMeta(ctx, madmin.SRBucketMeta{
+			Type:       madmin.SRBucketMetaTypeVersionConfig,
+			Bucket:     bucket,
+			Versioning: latestVersioningConfig,
+		})
+		if err != nil {
+			logger.LogIf(ctx, c.annotatePeerErr(peerName, "SRPeerReplicateBucketMeta", fmt.Errorf("Error healing versioning config metadata for peer %s from peer %s : %s",
+				peerName, latestPeerName, err.Error())))
+		}
+	}
+	return nil
+}
+
 func (c *SiteReplicationSys) healSSEMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
 	c.RLock()
 	defer c.RUnlock()

cmd/utils.go

@@ -130,6 +130,9 @@ func getWriteQuorum(drive int) int {
 	return quorum
 }
 
+// CloneMSS is an exposed function of cloneMSS for gateway usage.
+var CloneMSS = cloneMSS
+
 // cloneMSS will clone a map[string]string.
 // If input is nil an empty map is returned, not nil.
 func cloneMSS(v map[string]string) map[string]string {

docs/bucket/replication/README.md

@@ -243,6 +243,26 @@ Replication from a source bucket to multiple destination buckets is supported. F
 
 Note that on the source side, the `X-Amz-Replication-Status` changes from `PENDING` to `COMPLETED` after replication succeeds to each of the targets. On the destination side, a `X-Amz-Replication-Status` status of `REPLICA` indicates that the object was replicated successfully. Any replication failures are automatically re-attempted during a periodic disk scanner cycle.
 
+### Interaction with extended Bucket Versioning configuration
+When Bucket Versioning with excluded prefixes are configured objects matching these prefixes are excluded from being versioned and replicated.
+
+```
+<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+        <Status>Enabled</Status>
+        <ExcludeFolders>true</ExcludeFolders>
+
+        <ExcludedPrefixes>
+          <Prefix>app1-jobs/*/_temporary/</Prefix>
+        </ExcludedPrefixes>
+        <ExcludedPrefixes>
+          <Prefix>app2-jobs/*/_magic/</Prefix>
+        </ExcludedPrefixes>
+
+        <!-- .. up to 10 prefixes in all -->
+</VersioningConfiguration>
+```
+In the above sample config, objects under prefixes matching any of the `ExcludedPrefixes` glob patterns will neither be versioned nor replicated.
+
 ## Explore Further
 
 - [MinIO Bucket Replication Design](https://github.com/minio/minio/blob/master/docs/bucket/replication/DESIGN.md)

docs/bucket/versioning/README.md

@@ -63,7 +63,38 @@ Similarly to suspend versioning set the configuration with Status set to `Suspen
 </VersioningConfiguration>
 ```
 
-Only users with explicit permissions or the root credential can configure the versioning state of any bucket.
+## MinIO extension to Bucket Versioning
+### Motivation
+**PLEASE READ: This feature is meant for advanced usecases only where the setup is using bucket versioning or with replicated buckets, use this feature to optimize versioning behavior for some specific applications. MinIO experts will evaluate and guide on the benefits for your application, please reach out to us on https://subnet.min.io.**
+
+Spark/Hadoop workloads which use Hadoop MR Committer v1/v2 algorithm upload objects to a temporary prefix in a bucket. These objects are 'renamed' to a different prefix on Job commit. Object storage admins are forced to configure separate ILM policies to expire these objects and their versions to reclaim space.
+
+### Solution
+To exclude objects under a list of prefix (glob) patterns from being versioned, you can send the following versioning configuration with Status set to `Enabled`.
+
+```
+<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+        <Status>Enabled</Status>
+        <ExcludeFolders>true</ExcludeFolders>
+
+        <ExcludedPrefixes>
+          <Prefix>*/_temporary</Prefix>
+        </ExcludedPrefixes>
+        <ExcludedPrefixes>
+          <Prefix>*/__magic</Prefix>
+        </ExcludedPrefixes>
+        <ExcludedPrefixes>
+          <Prefix>*/_staging</Prefix>
+        </ExcludedPrefixes>
+
+        <!-- .. up to 10 prefixes in all -->
+</VersioningConfiguration>
+```
+
+### Features
+- Objects matching these prefixes will behave as though versioning were suspended. These objects **will not** be replicated if bucket has replication configured.
+- Objects matching these prefixes will also not leave `null` delete markers, dramatically reduces namespace pollution while keeping the benefits of replication.
+- Users with explicit permissions or the root credential can configure the versioning state of any bucket.
 
 ## Examples of enabling bucket versioning using MinIO Java SDK
 

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

@@ -2,7 +2,7 @@ version: '3.7'
 
 # Settings and configurations that are common for all containers
 x-minio-common: &minio-common
-  image: quay.io/minio/minio:RELEASE.2022-05-03T20-36-08Z
+  image: quay.io/minio/minio:RELEASE.2022-05-04T07-45-27Z
   command: server --console-address ":9001" http://minio{1...4}/data{1...2}
   expose:
     - "9000"

docs/sts/opa.md

@@ -0,0 +1,82 @@
+# OPA Quickstart Guide [![Slack](https://slack.minio.io/slack?type=svg)](https://slack.minio.io)
+OPA is a lightweight general-purpose policy engine that can be co-located with MinIO server, in this document we talk about how to use OPA HTTP API to authorize requests. It can be used with any type of credentials (STS based like OpenID or LDAP, regular IAM users or service accounts).
+
+## Get started
+
+### 1. Start OPA in a container
+
+```sh
+podman run -it \
+    --name opa \
+    --publish 8181:8181 \
+    docker.io/openpolicyagent/opa:0.40.0-rootless \
+       run --server \
+           --log-format=json-pretty \
+           --log-level=debug \
+           --set=decision_logs.console=true
+```
+
+### 2. Create a sample OPA Policy
+
+In another terminal, create a policy that allows root user all access and for all other users denies `PutObject`:
+```sh
+cat > example.rego <<EOF
+package httpapi.authz
+
+import input
+
+default allow = false
+
+# Allow the root user to perform any action.
+allow {
+ input.owner == true
+}
+
+# All other users may do anything other than call PutObject
+allow {
+ input.action != "s3:PutObject"
+ input.owner == false
+}
+EOF
+```
+
+Then load the policy via OPA's REST API.
+```
+curl -X PUT --data-binary @example.rego \
+  localhost:8181/v1/policies/putobject
+```
+
+### 4. Setup MinIO with OPA
+
+Set the `MINIO_POLICY_OPA_URL` as the endpoint that MinIO should send authorization requests to. Then start the server.
+
+```sh
+export MINIO_POLICY_OPA_URL=http://localhost:8181/v1/data/httpapi/authz/allow
+export MINIO_CI_CD=1
+export MINIO_ROOT_USER=minio
+export MINIO_ROOT_PASSWORD=minio123
+minio server /mnt/data
+```
+
+### 5. Test with a regular IAM user
+
+Ensure that `mc` is installed and the configured with the above server with the alias `myminio`.
+
+```sh
+# 1. Create a bucket and a user, and upload a file. These operations will succeed.
+mc mb myminio/test
+mc admin user add myminio foo foobar123
+mc cp /etc/issue myminio/test/
+
+# 2. Now access the server as user `foo`. These operations will also succeed.
+export MC_HOST_foo=http://foo:foobar123@localhost:9000
+mc ls foo/test
+mc cat foo/test/issue
+
+# 3. Attempt to upload an object as user `foo` - this will fail with a permissions error.
+mc cp /etc/issue myminio/test/issue2
+```
+
+## Explore Further
+- [MinIO STS Quickstart Guide](https://docs.minio.io/docs/minio-sts-quickstart-guide)
+- [The MinIO documentation website](https://docs.minio.io)

go.mod

@@ -48,7 +48,7 @@ require (
 	github.com/minio/dperf v0.3.6
 	github.com/minio/highwayhash v1.0.2
 	github.com/minio/kes v0.19.2
-	github.com/minio/madmin-go v1.3.12
+	github.com/minio/madmin-go v1.3.14
 	github.com/minio/minio-go/v7 v7.0.24
 	github.com/minio/pkg v1.1.23
 	github.com/minio/selfupdate v0.4.0

go.sum

@@ -614,8 +614,8 @@ github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLT
 github.com/minio/kes v0.19.2 h1:0kdMAgLMSkiDA33k8pMHC7d6erDuseuLrZF+N3017SM=
 github.com/minio/kes v0.19.2/go.mod h1:X2fMkDbAkjbSKDGOQZvyPkHxoG7nuzP6R78Jw+TzXtM=
 github.com/minio/madmin-go v1.3.5/go.mod h1:vGKGboQgGIWx4DuDUaXixjlIEZOCIp6ivJkQoiVaACc=
-github.com/minio/madmin-go v1.3.12 h1:7SmK/KtT7+d3hn3VcYBqI/c4yETfXV9gRT1j+g/U1jE=
-github.com/minio/madmin-go v1.3.12/go.mod h1:ez87VmMtsxP7DRxjKJKD4RDNW+nhO2QF9KSzwxBDQ98=
+github.com/minio/madmin-go v1.3.14 h1:9f9ZylP5Yn/TcplE/wowsBjb+Czt2+/NRCa2IqpNLcI=
+github.com/minio/madmin-go v1.3.14/go.mod h1:ez87VmMtsxP7DRxjKJKD4RDNW+nhO2QF9KSzwxBDQ98=
 github.com/minio/mc v0.0.0-20220419155441-cc4ff3a0cc82 h1:CiTaWFwpxzjd7A3sUQ0xZEX8sWfZh3/k2qbxuPip05s=
 github.com/minio/mc v0.0.0-20220419155441-cc4ff3a0cc82/go.mod h1:h6VCl43/2AUA3RP1GWUVMqcUiXq2NWJ4+dSei+ibf70=
 github.com/minio/md5-simd v1.1.0/go.mod h1:XpBqgZULrMYD3R+M28PcmP0CkI7PEMzB3U77ZrKZ0Gw=

helm/minio/templates/pvc.yaml

@@ -10,6 +10,10 @@ metadata:
     chart: {{ template "minio.chart" . }}
     release: {{ .Release.Name }}
     heritage: {{ .Release.Service }}
+{{- if .Values.persistence.annotations }}
+  annotations:
+{{ toYaml .Values.persistence.annotations | trimSuffix "\n" | indent 4 }}
+{{- end }}
 spec:
   accessModes:
     - {{ .Values.persistence.accessMode | quote }}

internal/bucket/versioning/versioning.go

@@ -20,6 +20,9 @@ package versioning
 import (
 	"encoding/xml"
 	"io"
+	"strings"
+
+	"github.com/minio/pkg/wildcard"
 )
 
 // State - enabled/disabled/suspended states
@@ -33,12 +36,26 @@ const (
 	Suspended State = "Suspended"
 )
 
+var (
+	errExcludedPrefixNotSupported = Errorf("excluded prefixes extension supported only when versioning is enabled")
+	errTooManyExcludedPrefixes    = Errorf("too many excluded prefixes")
+)
+
+// ExcludedPrefix - holds individual prefixes excluded from being versioned.
+type ExcludedPrefix struct {
+	Prefix string
+}
+
 // Versioning - Configuration for bucket versioning.
 type Versioning struct {
 	XMLNS   string   `xml:"xmlns,attr,omitempty"`
 	XMLName xml.Name `xml:"VersioningConfiguration"`
 	// MFADelete State    `xml:"MFADelete,omitempty"` // not supported yet.
 	Status State `xml:"Status,omitempty"`
+	// MinIO extension - allows selective, prefix-level versioning exclusion.
+	// Requires versioning to be enabled
+	ExcludedPrefixes []ExcludedPrefix `xml:",omitempty"`
+	ExcludeFolders   bool             `xml:",omitempty"`
 }
 
 // Validate - validates the versioning configuration
@@ -50,7 +67,16 @@ func (v Versioning) Validate() error {
 	// 	return Errorf("unsupported MFADelete state %s", v.MFADelete)
 	// }
 	switch v.Status {
-	case Enabled, Suspended:
+	case Enabled:
+		const maxExcludedPrefixes = 10
+		if len(v.ExcludedPrefixes) > maxExcludedPrefixes {
+			return errTooManyExcludedPrefixes
+		}
+
+	case Suspended:
+		if len(v.ExcludedPrefixes) > 0 {
+			return errExcludedPrefixNotSupported
+		}
 	default:
 		return Errorf("unsupported Versioning status %s", v.Status)
 	}
@@ -62,11 +88,67 @@ func (v Versioning) Enabled() bool {
 	return v.Status == Enabled
 }
 
+// PrefixEnabled - returns true if versioning is enabled at the bucket and given
+// prefix, false otherwise.
+func (v Versioning) PrefixEnabled(prefix string) bool {
+	if v.Status != Enabled {
+		return false
+	}
+
+	if prefix == "" {
+		return true
+	}
+	if v.ExcludeFolders && strings.HasSuffix(prefix, "/") {
+		return false
+	}
+
+	for _, sprefix := range v.ExcludedPrefixes {
+		// Note: all excluded prefix patterns end with `/` (See Validate)
+		sprefix.Prefix += "*"
+
+		if matched := wildcard.MatchSimple(sprefix.Prefix, prefix); matched {
+			return false
+		}
+	}
+	return true
+}
+
 // Suspended - returns true if versioning is suspended
 func (v Versioning) Suspended() bool {
 	return v.Status == Suspended
 }
 
+// PrefixSuspended - returns true if versioning is suspended at the bucket level
+// or suspended on the given prefix.
+func (v Versioning) PrefixSuspended(prefix string) bool {
+	if v.Status == Suspended {
+		return true
+	}
+	if v.Status == Enabled {
+		if prefix == "" {
+			return false
+		}
+		if v.ExcludeFolders && strings.HasSuffix(prefix, "/") {
+			return true
+		}
+
+		for _, sprefix := range v.ExcludedPrefixes {
+			// Note: all excluded prefix patterns end with `/` (See Validate)
+			sprefix.Prefix += "*"
+			if matched := wildcard.MatchSimple(sprefix.Prefix, prefix); matched {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// PrefixesExcluded returns true if v contains one or more excluded object
+// prefixes or if ExcludeFolders is true.
+func (v Versioning) PrefixesExcluded() bool {
+	return len(v.ExcludedPrefixes) > 0 || v.ExcludeFolders
+}
+
 // ParseConfig - parses data in given reader to VersioningConfiguration.
 func ParseConfig(reader io.Reader) (*Versioning, error) {
 	var v Versioning

internal/bucket/versioning/versioning_test.go

@@ -0,0 +1,241 @@
+// 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 versioning
+
+import (
+	"encoding/xml"
+	"strings"
+	"testing"
+)
+
+func TestParseConfig(t *testing.T) {
+	testcases := []struct {
+		input            string
+		err              error
+		excludedPrefixes []string
+		excludeFolders   bool
+	}{
+		{
+			input: `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+                                  <Status>Enabled</Status>
+                                </VersioningConfiguration>`,
+			err: nil,
+		},
+		{
+			input: `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+                                  <Status>Enabled</Status>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_temporary/</Prefix>
+                                  </ExcludedPrefixes>
+                                </VersioningConfiguration>`,
+			err:              nil,
+			excludedPrefixes: []string{"path/to/my/workload/_staging/", "path/to/my/workload/_temporary/"},
+		},
+		{
+			input: `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+                                  <Status>Suspended</Status>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging</Prefix>
+                                  </ExcludedPrefixes>
+                                </VersioningConfiguration>`,
+			err: errExcludedPrefixNotSupported,
+		},
+		{
+			input: `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+                                  <Status>Enabled</Status>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/ab/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/cd/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/ef/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/gh/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/ij/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/kl/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/mn/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/op/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/qr/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/st/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/uv/</Prefix>
+                                  </ExcludedPrefixes>
+                                </VersioningConfiguration>`,
+			err: errTooManyExcludedPrefixes,
+		},
+		{
+			input: `<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+                                  <Status>Enabled</Status>
+                                  <ExcludeFolders>true</ExcludeFolders>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_staging/</Prefix>
+                                  </ExcludedPrefixes>
+                                  <ExcludedPrefixes>
+                                    <Prefix>path/to/my/workload/_temporary/</Prefix>
+                                  </ExcludedPrefixes>
+                                </VersioningConfiguration>`,
+			err:              nil,
+			excludedPrefixes: []string{"path/to/my/workload/_staging/", "path/to/my/workload/_temporary/"},
+			excludeFolders:   true,
+		},
+	}
+
+	for i, tc := range testcases {
+		var v *Versioning
+		var err error
+		v, err = ParseConfig(strings.NewReader(tc.input))
+		if tc.err != err {
+			t.Fatalf("Test %d: expected %v but got %v", i+1, tc.err, err)
+		}
+		if err != nil {
+			if tc.err == nil {
+				t.Fatalf("Test %d: failed due to %v", i+1, err)
+			}
+		} else {
+			if err := v.Validate(); tc.err != err {
+				t.Fatalf("Test %d: validation failed due to %v", i+1, err)
+			}
+			if len(tc.excludedPrefixes) > 0 {
+				var mismatch bool
+				if len(v.ExcludedPrefixes) != len(tc.excludedPrefixes) {
+					t.Fatalf("Test %d: Expected length of excluded prefixes %d but got %d", i+1, len(tc.excludedPrefixes), len(v.ExcludedPrefixes))
+				}
+				var i int
+				var eprefix string
+				for i, eprefix = range tc.excludedPrefixes {
+					if eprefix != v.ExcludedPrefixes[i].Prefix {
+						mismatch = true
+						break
+					}
+				}
+				if mismatch {
+					t.Fatalf("Test %d: Expected excluded prefix %s but got %s", i+1, tc.excludedPrefixes[i], v.ExcludedPrefixes[i].Prefix)
+				}
+			}
+			if tc.excludeFolders != v.ExcludeFolders {
+				t.Fatalf("Test %d: Expected ExcludeFoldersr=%v but got %v", i+1, tc.excludeFolders, v.ExcludeFolders)
+			}
+		}
+	}
+}
+
+func TestMarshalXML(t *testing.T) {
+	// Validates if Versioning with no excluded prefixes omits
+	// ExcludedPrefixes tags
+	v := Versioning{
+		Status: Enabled,
+	}
+	buf, err := xml.Marshal(v)
+	if err != nil {
+		t.Fatalf("Failed to marshal %v: %v", v, err)
+	}
+
+	str := string(buf)
+	if strings.Contains(str, "ExcludedPrefixes") {
+		t.Fatalf("XML shouldn't contain ExcludedPrefixes tag - %s", str)
+	}
+}
+
+func TestVersioningZero(t *testing.T) {
+	var v Versioning
+	if v.Enabled() {
+		t.Fatalf("Expected to be disabled but got enabled")
+	}
+	if v.Suspended() {
+		t.Fatalf("Expected to be disabled but got suspended")
+	}
+}
+
+func TestExcludeFolders(t *testing.T) {
+	v := Versioning{
+		Status:         Enabled,
+		ExcludeFolders: true,
+	}
+	testPrefixes := []string{"jobs/output/_temporary/", "jobs/output/", "jobs/"}
+	for i, prefix := range testPrefixes {
+		if v.PrefixEnabled(prefix) || !v.PrefixSuspended(prefix) {
+			t.Fatalf("Test %d: Expected versioning to be excluded for %s", i+1, prefix)
+		}
+	}
+
+	// Test applicability for regular objects
+	if prefix := "prefix-1/obj-1"; !v.PrefixEnabled(prefix) || v.PrefixSuspended(prefix) {
+		t.Fatalf("Expected versioning to be enabled for %s", prefix)
+	}
+
+	// Test when ExcludeFolders is disabled
+	v.ExcludeFolders = false
+	for i, prefix := range testPrefixes {
+		if !v.PrefixEnabled(prefix) || v.PrefixSuspended(prefix) {
+			t.Fatalf("Test %d: Expected versioning to be enabled for %s", i+1, prefix)
+		}
+	}
+}
+
+func TestExcludedPrefixesMatch(t *testing.T) {
+	v := Versioning{
+		Status:           Enabled,
+		ExcludedPrefixes: []ExcludedPrefix{{"*/_temporary/"}},
+	}
+
+	if err := v.Validate(); err != nil {
+		t.Fatalf("Invalid test versioning config %v: %v", v, err)
+	}
+	tests := []struct {
+		prefix   string
+		excluded bool
+	}{
+		{
+			prefix:   "app1-jobs/output/_temporary/attempt1/data.csv",
+			excluded: true,
+		},
+		{
+			prefix:   "app1-jobs/output/final/attempt1/data.csv",
+			excluded: false,
+		},
+	}
+
+	for i, test := range tests {
+		if v.PrefixSuspended(test.prefix) != test.excluded {
+			if test.excluded {
+				t.Fatalf("Test %d: Expected prefix %s to be excluded from versioning", i+1, test.prefix)
+			} else {
+				t.Fatalf("Test %d: Expected prefix %s to have versioning enabled", i+1, test.prefix)
+			}
+		}
+	}
+}

internal/config/compress/help.go

@@ -26,6 +26,13 @@ var (
 	}
 
 	Help = config.HelpKVS{
+		config.HelpKV{
+			Key:         config.Enable,
+			Description: "Enable or disable object compression",
+			Type:        "on|off",
+			Optional:    true,
+			Sensitive:   false,
+		},
 		config.HelpKV{
 			Key:         Extensions,
 			Description: `comma separated file extensions` + defaultHelpPostfix(Extensions),

internal/config/identity/openid/help.go

@@ -26,6 +26,13 @@ var (
 	}
 
 	Help = config.HelpKVS{
+		config.HelpKV{
+			Key:         config.Enable,
+			Description: "Enable or disable OpenID",
+			Type:        "on|off",
+			Optional:    true,
+			Sensitive:   false,
+		},
 		config.HelpKV{
 			Key:         DisplayName,
 			Description: "Friendly display name for this Provider/App" + defaultHelpPostfix(DisplayName),

internal/config/identity/openid/jwt.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2015-2021 MinIO, Inc.
+// Copyright (c) 2015-2022 MinIO, Inc.
 //
 // This file is part of MinIO Object Storage stack
 //
@@ -19,34 +19,21 @@ package openid
 
 import (
 	"crypto"
-	"crypto/sha1"
-	"crypto/sha256"
-	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
-	"strconv"
-	"strings"
 	"sync"
 	"time"
 
 	jwtgo "github.com/golang-jwt/jwt/v4"
-	"github.com/minio/madmin-go"
-	"github.com/minio/minio-go/v7/pkg/set"
 	"github.com/minio/minio/internal/arn"
 	"github.com/minio/minio/internal/auth"
-	"github.com/minio/minio/internal/config"
-	"github.com/minio/minio/internal/config/identity/openid/provider"
-	xhttp "github.com/minio/minio/internal/http"
-	"github.com/minio/pkg/env"
 	iampolicy "github.com/minio/pkg/iam/policy"
 	xnet "github.com/minio/pkg/net"
 )
 
-var errSingleProvider = config.Errorf("Only one OpenID provider can be configured if not using role policy mapping")
-
 type publicKeys struct {
 	*sync.RWMutex
 
@@ -79,185 +66,6 @@ func (pk *publicKeys) get(kid string) crypto.PublicKey {
 	return pk.pkMap[kid]
 }
 
-type providerCfg struct {
-	// Used for user interface like console
-	DisplayName string `json:"displayName,omitempty"`
-
-	JWKS struct {
-		URL *xnet.URL `json:"url"`
-	} `json:"jwks"`
-	URL                *xnet.URL `json:"url,omitempty"`
-	ClaimPrefix        string    `json:"claimPrefix,omitempty"`
-	ClaimName          string    `json:"claimName,omitempty"`
-	ClaimUserinfo      bool      `json:"claimUserInfo,omitempty"`
-	RedirectURI        string    `json:"redirectURI,omitempty"`
-	RedirectURIDynamic bool      `json:"redirectURIDynamic"`
-	DiscoveryDoc       DiscoveryDoc
-	ClientID           string
-	ClientSecret       string
-	RolePolicy         string
-
-	roleArn  arn.ARN
-	provider provider.Provider
-}
-
-// initializeProvider initializes if any additional vendor specific information
-// was provided, initialization will return an error initial login fails.
-func (p *providerCfg) initializeProvider(cfgGet func(string, string) string, transport http.RoundTripper) error {
-	vendor := cfgGet(EnvIdentityOpenIDVendor, Vendor)
-	if vendor == "" {
-		return nil
-	}
-	var err error
-	switch vendor {
-	case keyCloakVendor:
-		adminURL := cfgGet(EnvIdentityOpenIDKeyCloakAdminURL, KeyCloakAdminURL)
-		realm := cfgGet(EnvIdentityOpenIDKeyCloakRealm, KeyCloakRealm)
-		p.provider, err = provider.KeyCloak(
-			provider.WithAdminURL(adminURL),
-			provider.WithOpenIDConfig(provider.DiscoveryDoc(p.DiscoveryDoc)),
-			provider.WithTransport(transport),
-			provider.WithRealm(realm),
-		)
-		return err
-	default:
-		return fmt.Errorf("Unsupport vendor %s", keyCloakVendor)
-	}
-}
-
-// UserInfo returns claims for authenticated user from userInfo endpoint.
-//
-// Some OIDC implementations such as GitLab do not support
-// claims as part of the normal oauth2 flow, instead rely
-// on service providers making calls to IDP to fetch additional
-// claims available from the UserInfo endpoint
-func (p *providerCfg) UserInfo(accessToken string, transport http.RoundTripper) (map[string]interface{}, error) {
-	if p.JWKS.URL == nil || p.JWKS.URL.String() == "" {
-		return nil, errors.New("openid not configured")
-	}
-	client := &http.Client{
-		Transport: transport,
-	}
-
-	req, err := http.NewRequest(http.MethodPost, p.DiscoveryDoc.UserInfoEndpoint, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	if accessToken != "" {
-		req.Header.Set("Authorization", "Bearer "+accessToken)
-	}
-
-	resp, err := client.Do(req)
-	if err != nil {
-		return nil, err
-	}
-
-	defer xhttp.DrainBody(resp.Body)
-	if resp.StatusCode != http.StatusOK {
-		// uncomment this for debugging when needed.
-		// reqBytes, _ := httputil.DumpRequest(req, false)
-		// fmt.Println(string(reqBytes))
-		// respBytes, _ := httputil.DumpResponse(resp, true)
-		// fmt.Println(string(respBytes))
-		return nil, errors.New(resp.Status)
-	}
-
-	dec := json.NewDecoder(resp.Body)
-	claims := map[string]interface{}{}
-
-	if err = dec.Decode(&claims); err != nil {
-		// uncomment this for debugging when needed.
-		// reqBytes, _ := httputil.DumpRequest(req, false)
-		// fmt.Println(string(reqBytes))
-		// respBytes, _ := httputil.DumpResponse(resp, true)
-		// fmt.Println(string(respBytes))
-		return nil, err
-	}
-
-	return claims, nil
-}
-
-// Config - OpenID Config
-// RSA authentication target arguments
-type Config struct {
-	Enabled bool `json:"enabled"`
-
-	// map of roleARN to providerCfg's
-	arnProviderCfgsMap map[arn.ARN]*providerCfg
-
-	// map of config names to providerCfg's
-	ProviderCfgs map[string]*providerCfg
-
-	pubKeys          publicKeys
-	roleArnPolicyMap map[arn.ARN]string
-
-	transport   http.RoundTripper
-	closeRespFn func(io.ReadCloser)
-}
-
-// GetIAMPolicyClaimName - returns the policy claim name for the (at most one)
-// provider configured without a role policy.
-func (r *Config) GetIAMPolicyClaimName() string {
-	pCfg, ok := r.arnProviderCfgsMap[DummyRoleARN]
-	if !ok {
-		return ""
-	}
-	return pCfg.ClaimPrefix + pCfg.ClaimName
-}
-
-// LookupUser lookup userid for the provider
-func (r Config) LookupUser(roleArn, userid string) (provider.User, error) {
-	// Can safely ignore error here as empty or invalid ARNs will not be
-	// mapped.
-	arnVal, _ := arn.Parse(roleArn)
-	pCfg, ok := r.arnProviderCfgsMap[arnVal]
-	if ok {
-		user, err := pCfg.provider.LookupUser(userid)
-		if err != nil && err != provider.ErrAccessTokenExpired {
-			return user, err
-		}
-		if err == provider.ErrAccessTokenExpired {
-			if err = pCfg.provider.LoginWithClientID(pCfg.ClientID, pCfg.ClientSecret); err != nil {
-				return user, err
-			}
-			user, err = pCfg.provider.LookupUser(userid)
-		}
-		return user, err
-	}
-	// Without any specific logic for a provider, all accounts
-	// are always enabled.
-	return provider.User{ID: userid, Enabled: true}, nil
-}
-
-const (
-	keyCloakVendor = "keycloak"
-)
-
-// ProviderEnabled returns true if any vendor specific provider is enabled.
-func (r Config) ProviderEnabled() bool {
-	if !r.Enabled {
-		return false
-	}
-	for _, v := range r.arnProviderCfgsMap {
-		if v.provider != nil {
-			return true
-		}
-	}
-	return false
-}
-
-// GetRoleInfo - returns ARN to policies map if a role policy based openID
-// provider is configured. Otherwise returns nil.
-func (r Config) GetRoleInfo() map[arn.ARN]string {
-	for _, p := range r.arnProviderCfgsMap {
-		if p.RolePolicy != "" {
-			return r.roleArnPolicyMap
-		}
-	}
-	return nil
-}
-
 // PopulatePublicKey - populates a new publickey from the JWKS URL.
 func (r *Config) PopulatePublicKey(arn arn.ARN) error {
 	pCfg := r.arnProviderCfgsMap[arn]
@@ -280,42 +88,6 @@ func (r *Config) PopulatePublicKey(arn arn.ARN) error {
 	return r.pubKeys.parseAndAdd(resp.Body)
 }
 
-// UnmarshalJSON - decodes JSON data.
-func (r *Config) UnmarshalJSON(data []byte) error {
-	// subtype to avoid recursive call to UnmarshalJSON()
-	type subConfig Config
-	var sr subConfig
-
-	if err := json.Unmarshal(data, &sr); err != nil {
-		return err
-	}
-
-	ar := Config(sr)
-	*r = ar
-	return nil
-}
-
-// GetDefaultExpiration - returns the expiration seconds expected.
-func GetDefaultExpiration(dsecs string) (time.Duration, error) {
-	defaultExpiryDuration := time.Duration(60) * time.Minute // Defaults to 1hr.
-	if dsecs != "" {
-		expirySecs, err := strconv.ParseInt(dsecs, 10, 64)
-		if err != nil {
-			return 0, auth.ErrInvalidDuration
-		}
-
-		// The duration, in seconds, of the role session.
-		// The value can range from 900 seconds (15 minutes)
-		// up to 365 days.
-		if expirySecs < 900 || expirySecs > 31536000 {
-			return 0, auth.ErrInvalidDuration
-		}
-
-		defaultExpiryDuration = time.Duration(expirySecs) * time.Second
-	}
-	return defaultExpiryDuration, nil
-}
-
 // ErrTokenExpired - error token expired
 var (
 	ErrTokenExpired = errors.New("token expired")
@@ -461,86 +233,6 @@ func (r *Config) updateUserinfoClaims(arn arn.ARN, accessToken string, claims ma
 	return nil
 }
 
-// GetSettings - fetches OIDC settings for site-replication related validation.
-// NOTE that region must be populated by caller as this package does not know.
-func (r *Config) GetSettings() madmin.OpenIDSettings {
-	res := madmin.OpenIDSettings{}
-	if !r.Enabled {
-		return res
-	}
-
-	for arn, provCfg := range r.arnProviderCfgsMap {
-		hashedSecret := ""
-		{
-			h := sha256.New()
-			h.Write([]byte(provCfg.ClientSecret))
-			bs := h.Sum(nil)
-			hashedSecret = base64.RawURLEncoding.EncodeToString(bs)
-		}
-		if arn != DummyRoleARN {
-			if res.Roles != nil {
-				res.Roles = make(map[string]madmin.OpenIDProviderSettings)
-			}
-			res.Roles[arn.String()] = madmin.OpenIDProviderSettings{
-				ClaimUserinfoEnabled: provCfg.ClaimUserinfo,
-				RolePolicy:           provCfg.RolePolicy,
-				ClientID:             provCfg.ClientID,
-				HashedClientSecret:   hashedSecret,
-			}
-		} else {
-			res.ClaimProvider = madmin.OpenIDProviderSettings{
-				ClaimUserinfoEnabled: provCfg.ClaimUserinfo,
-				RolePolicy:           provCfg.RolePolicy,
-				ClientID:             provCfg.ClientID,
-				HashedClientSecret:   hashedSecret,
-			}
-		}
-
-	}
-
-	return res
-}
-
-// OpenID keys and envs.
-const (
-	JwksURL       = "jwks_url"
-	ConfigURL     = "config_url"
-	ClaimName     = "claim_name"
-	ClaimUserinfo = "claim_userinfo"
-	ClaimPrefix   = "claim_prefix"
-	ClientID      = "client_id"
-	ClientSecret  = "client_secret"
-	RolePolicy    = "role_policy"
-	DisplayName   = "display_name"
-
-	Vendor             = "vendor"
-	Scopes             = "scopes"
-	RedirectURI        = "redirect_uri"
-	RedirectURIDynamic = "redirect_uri_dynamic"
-
-	// Vendor specific ENV only enabled if the Vendor matches == "vendor"
-	KeyCloakRealm    = "keycloak_realm"
-	KeyCloakAdminURL = "keycloak_admin_url"
-
-	EnvIdentityOpenIDEnable             = "MINIO_IDENTITY_OPENID_ENABLE"
-	EnvIdentityOpenIDVendor             = "MINIO_IDENTITY_OPENID_VENDOR"
-	EnvIdentityOpenIDClientID           = "MINIO_IDENTITY_OPENID_CLIENT_ID"
-	EnvIdentityOpenIDClientSecret       = "MINIO_IDENTITY_OPENID_CLIENT_SECRET"
-	EnvIdentityOpenIDURL                = "MINIO_IDENTITY_OPENID_CONFIG_URL"
-	EnvIdentityOpenIDClaimName          = "MINIO_IDENTITY_OPENID_CLAIM_NAME"
-	EnvIdentityOpenIDClaimUserInfo      = "MINIO_IDENTITY_OPENID_CLAIM_USERINFO"
-	EnvIdentityOpenIDClaimPrefix        = "MINIO_IDENTITY_OPENID_CLAIM_PREFIX"
-	EnvIdentityOpenIDRolePolicy         = "MINIO_IDENTITY_OPENID_ROLE_POLICY"
-	EnvIdentityOpenIDRedirectURI        = "MINIO_IDENTITY_OPENID_REDIRECT_URI"
-	EnvIdentityOpenIDRedirectURIDynamic = "MINIO_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC"
-	EnvIdentityOpenIDScopes             = "MINIO_IDENTITY_OPENID_SCOPES"
-	EnvIdentityOpenIDDisplayName        = "MINIO_IDENTITY_OPENID_DISPLAY_NAME"
-
-	// Vendor specific ENVs only enabled if the Vendor matches == "vendor"
-	EnvIdentityOpenIDKeyCloakRealm    = "MINIO_IDENTITY_OPENID_KEYCLOAK_REALM"
-	EnvIdentityOpenIDKeyCloakAdminURL = "MINIO_IDENTITY_OPENID_KEYCLOAK_ADMIN_URL"
-)
-
 // DiscoveryDoc - parses the output from openid-configuration
 // for example https://accounts.google.com/.well-known/openid-configuration
 type DiscoveryDoc struct {
@@ -582,312 +274,3 @@ func parseDiscoveryDoc(u *xnet.URL, transport http.RoundTripper, closeRespFn fun
 	}
 	return d, nil
 }
-
-// DefaultKVS - default config for OpenID config
-var (
-	DefaultKVS = config.KVS{
-		config.KV{
-			Key:   config.Enable,
-			Value: "",
-		},
-		config.KV{
-			Key:   DisplayName,
-			Value: "",
-		},
-		config.KV{
-			Key:   ConfigURL,
-			Value: "",
-		},
-		config.KV{
-			Key:   ClientID,
-			Value: "",
-		},
-		config.KV{
-			Key:   ClientSecret,
-			Value: "",
-		},
-		config.KV{
-			Key:   ClaimName,
-			Value: iampolicy.PolicyName,
-		},
-		config.KV{
-			Key:   ClaimUserinfo,
-			Value: "",
-		},
-		config.KV{
-			Key:   RolePolicy,
-			Value: "",
-		},
-		config.KV{
-			Key:   ClaimPrefix,
-			Value: "",
-		},
-		config.KV{
-			Key:   RedirectURI,
-			Value: "",
-		},
-		config.KV{
-			Key:   RedirectURIDynamic,
-			Value: "off",
-		},
-		config.KV{
-			Key:   Scopes,
-			Value: "",
-		},
-	}
-)
-
-// Enabled returns if configURL is enabled.
-func Enabled(kvs config.KVS) bool {
-	return kvs.Get(ConfigURL) != ""
-}
-
-// DummyRoleARN is used to indicate that the user associated with it was
-// authenticated via policy-claim based OpenID provider.
-var DummyRoleARN = func() arn.ARN {
-	v, err := arn.NewIAMRoleARN("dummy-internal", "")
-	if err != nil {
-		panic("should not happen!")
-	}
-	return v
-}()
-
-// LookupConfig lookup jwks from config, override with any ENVs.
-func LookupConfig(kvsMap map[string]config.KVS, transport http.RoundTripper, closeRespFn func(io.ReadCloser), serverRegion string) (c Config, err error) {
-	openIDClientTransport := http.DefaultTransport
-	if transport != nil {
-		openIDClientTransport = transport
-	}
-	c = Config{
-		Enabled:            false,
-		arnProviderCfgsMap: map[arn.ARN]*providerCfg{},
-		ProviderCfgs:       map[string]*providerCfg{},
-		pubKeys: publicKeys{
-			RWMutex: &sync.RWMutex{},
-			pkMap:   map[string]crypto.PublicKey{},
-		},
-		roleArnPolicyMap: map[arn.ARN]string{},
-		transport:        openIDClientTransport,
-		closeRespFn:      closeRespFn,
-	}
-
-	// Make a copy of the config we received so we can mutate it safely.
-	kvsMap2 := make(map[string]config.KVS, len(kvsMap))
-	for k, v := range kvsMap {
-		kvsMap2[k] = v
-	}
-
-	// Add in each configuration name found from environment variables, i.e.
-	// if we see MINIO_IDENTITY_OPENID_CONFIG_URL_2, we add the key "2" to
-	// `kvsMap2` if it does not already exist.
-	envs := env.List(EnvIdentityOpenIDURL + config.Default)
-	for _, k := range envs {
-		cfgName := strings.TrimPrefix(k, EnvIdentityOpenIDURL+config.Default)
-		if cfgName == "" {
-			return c, config.Errorf("Environment variable must have a non-empty config name: %s", k)
-		}
-
-		// It is possible that some variables were specified via config
-		// commands and some variables are intended to be overridden
-		// from the environment, so we ensure that the key is not
-		// overwritten in `kvsMap2` as it may have existing config.
-		if _, ok := kvsMap2[cfgName]; !ok {
-			kvsMap2[cfgName] = DefaultKVS
-		}
-	}
-
-	var (
-		hasLegacyPolicyMapping = false
-		seenClientIDs          = set.NewStringSet()
-	)
-	for cfgName, kvs := range kvsMap2 {
-		// remove this since we have removed support for this already.
-		kvs.Delete(JwksURL)
-
-		if err = config.CheckValidKeys(config.IdentityOpenIDSubSys, kvs, DefaultKVS); err != nil {
-			return c, err
-		}
-
-		getCfgVal := func(envVar, cfgParam string) string {
-			if cfgName != config.Default {
-				envVar += config.Default + cfgName
-			}
-			return env.Get(envVar, kvs.Get(cfgParam))
-		}
-
-		// In the past, when only one openID provider was allowed, there
-		// was no `enable` parameter - the configuration is turned off
-		// by clearing the values. With multiple providers, we support
-		// individually enabling/disabling provider configurations. If
-		// the enable parameter's value is non-empty, we use that
-		// setting, otherwise we treat it as enabled if some important
-		// parameters are non-empty.
-		var (
-			cfgEnableVal        = getCfgVal(EnvIdentityOpenIDEnable, config.Enable)
-			isExplicitlyEnabled = false
-		)
-		if cfgEnableVal != "" {
-			isExplicitlyEnabled = true
-		}
-
-		var enabled bool
-		if isExplicitlyEnabled {
-			enabled, err = config.ParseBool(cfgEnableVal)
-			if err != nil {
-				return c, err
-			}
-			// No need to continue loading if the config is not enabled.
-			if !enabled {
-				continue
-			}
-		}
-
-		p := providerCfg{
-			DisplayName:        getCfgVal(EnvIdentityOpenIDDisplayName, DisplayName),
-			ClaimName:          getCfgVal(EnvIdentityOpenIDClaimName, ClaimName),
-			ClaimUserinfo:      getCfgVal(EnvIdentityOpenIDClaimUserInfo, ClaimUserinfo) == config.EnableOn,
-			ClaimPrefix:        getCfgVal(EnvIdentityOpenIDClaimPrefix, ClaimPrefix),
-			RedirectURI:        getCfgVal(EnvIdentityOpenIDRedirectURI, RedirectURI),
-			RedirectURIDynamic: getCfgVal(EnvIdentityOpenIDRedirectURIDynamic, RedirectURIDynamic) == config.EnableOn,
-			ClientID:           getCfgVal(EnvIdentityOpenIDClientID, ClientID),
-			ClientSecret:       getCfgVal(EnvIdentityOpenIDClientSecret, ClientSecret),
-			RolePolicy:         getCfgVal(EnvIdentityOpenIDRolePolicy, RolePolicy),
-		}
-
-		configURL := getCfgVal(EnvIdentityOpenIDURL, ConfigURL)
-
-		if !isExplicitlyEnabled {
-			enabled = true
-			if p.ClientID == "" && p.ClientSecret == "" && configURL == "" {
-				enabled = false
-			}
-		}
-
-		// No need to continue loading if the config is not enabled.
-		if !enabled {
-			continue
-		}
-
-		// Validate that client ID has not been duplicately specified.
-		if seenClientIDs.Contains(p.ClientID) {
-			return c, config.Errorf("Client ID %s is present with multiple OpenID configurations", p.ClientID)
-		}
-		seenClientIDs.Add(p.ClientID)
-
-		p.URL, err = xnet.ParseHTTPURL(configURL)
-		if err != nil {
-			return c, err
-		}
-		configURLDomain := p.URL.Hostname()
-		p.DiscoveryDoc, err = parseDiscoveryDoc(p.URL, transport, closeRespFn)
-		if err != nil {
-			return c, err
-		}
-
-		if p.ClaimUserinfo && configURL == "" {
-			return c, errors.New("please specify config_url to enable fetching claims from UserInfo endpoint")
-		}
-
-		if scopeList := getCfgVal(EnvIdentityOpenIDScopes, Scopes); scopeList != "" {
-			var scopes []string
-			for _, scope := range strings.Split(scopeList, ",") {
-				scope = strings.TrimSpace(scope)
-				if scope == "" {
-					return c, config.Errorf("empty scope value is not allowed '%s', please refer to our documentation", scopeList)
-				}
-				scopes = append(scopes, scope)
-			}
-			// Replace the discovery document scopes by client customized scopes.
-			p.DiscoveryDoc.ScopesSupported = scopes
-		}
-
-		// Check if claim name is the non-default value and role policy is set.
-		if p.ClaimName != iampolicy.PolicyName && p.RolePolicy != "" {
-			// In the unlikely event that the user specifies
-			// `iampolicy.PolicyName` as the claim name explicitly and sets
-			// a role policy, this check is thwarted, but we will be using
-			// the role policy anyway.
-			return c, config.Errorf("Role Policy (=`%s`) and Claim Name (=`%s`) cannot both be set", p.RolePolicy, p.ClaimName)
-		}
-
-		jwksURL := p.DiscoveryDoc.JwksURI
-		if jwksURL == "" {
-			return c, config.Errorf("no JWKS URI found in your provider's discovery doc (config_url=%s)", configURL)
-		}
-
-		p.JWKS.URL, err = xnet.ParseHTTPURL(jwksURL)
-		if err != nil {
-			return c, err
-		}
-
-		if p.RolePolicy != "" {
-			// RolePolicy is validated by IAM System during its
-			// initialization.
-
-			// Generate role ARN as combination of provider domain and
-			// prefix of client ID.
-			domain := configURLDomain
-			if domain == "" {
-				// Attempt to parse the JWKs URI.
-				domain = p.JWKS.URL.Hostname()
-				if domain == "" {
-					return c, config.Errorf("unable to generate a domain from the OpenID config")
-				}
-			}
-			if p.ClientID == "" {
-				return c, config.Errorf("client ID must not be empty")
-			}
-
-			// We set the resource ID of the role arn as a hash of client
-			// ID, so we can get a short roleARN that stays the same on
-			// restart.
-			var resourceID string
-			{
-				h := sha1.New()
-				h.Write([]byte(p.ClientID))
-				bs := h.Sum(nil)
-				resourceID = base64.RawURLEncoding.EncodeToString(bs)
-			}
-			p.roleArn, err = arn.NewIAMRoleARN(resourceID, serverRegion)
-			if err != nil {
-				return c, config.Errorf("unable to generate ARN from the OpenID config: %v", err)
-			}
-
-			c.roleArnPolicyMap[p.roleArn] = p.RolePolicy
-		} else if p.ClaimName == "" {
-			return c, config.Errorf("A role policy or claim name must be specified")
-		}
-
-		if err = p.initializeProvider(getCfgVal, c.transport); err != nil {
-			return c, err
-		}
-
-		arnKey := p.roleArn
-		if p.RolePolicy == "" {
-			arnKey = DummyRoleARN
-			hasLegacyPolicyMapping = true
-			// Ensure that when a JWT policy claim based provider
-			// exists, it is the only one.
-			if _, ok := c.arnProviderCfgsMap[DummyRoleARN]; ok {
-				return c, errSingleProvider
-			}
-		}
-
-		c.arnProviderCfgsMap[arnKey] = &p
-		c.ProviderCfgs[cfgName] = &p
-
-		if err = c.PopulatePublicKey(arnKey); err != nil {
-			return c, err
-		}
-	}
-
-	// Ensure that when a JWT policy claim based provider
-	// exists, it is the only one.
-	if hasLegacyPolicyMapping && len(c.ProviderCfgs) > 1 {
-		return c, errSingleProvider
-	}
-
-	c.Enabled = true
-
-	return c, nil
-}

internal/config/identity/openid/openid.go

@@ -0,0 +1,518 @@
+// 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 openid
+
+import (
+	"crypto"
+	"crypto/sha1"
+	"crypto/sha256"
+	"encoding/base64"
+	"errors"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/minio/madmin-go"
+	"github.com/minio/minio-go/v7/pkg/set"
+	"github.com/minio/minio/internal/arn"
+	"github.com/minio/minio/internal/auth"
+	"github.com/minio/minio/internal/config"
+	"github.com/minio/minio/internal/config/identity/openid/provider"
+	"github.com/minio/pkg/env"
+	iampolicy "github.com/minio/pkg/iam/policy"
+	xnet "github.com/minio/pkg/net"
+)
+
+// OpenID keys and envs.
+const (
+	JwksURL       = "jwks_url"
+	ConfigURL     = "config_url"
+	ClaimName     = "claim_name"
+	ClaimUserinfo = "claim_userinfo"
+	ClaimPrefix   = "claim_prefix"
+	ClientID      = "client_id"
+	ClientSecret  = "client_secret"
+	RolePolicy    = "role_policy"
+	DisplayName   = "display_name"
+
+	Vendor             = "vendor"
+	Scopes             = "scopes"
+	RedirectURI        = "redirect_uri"
+	RedirectURIDynamic = "redirect_uri_dynamic"
+
+	// Vendor specific ENV only enabled if the Vendor matches == "vendor"
+	KeyCloakRealm    = "keycloak_realm"
+	KeyCloakAdminURL = "keycloak_admin_url"
+
+	EnvIdentityOpenIDEnable             = "MINIO_IDENTITY_OPENID_ENABLE"
+	EnvIdentityOpenIDVendor             = "MINIO_IDENTITY_OPENID_VENDOR"
+	EnvIdentityOpenIDClientID           = "MINIO_IDENTITY_OPENID_CLIENT_ID"
+	EnvIdentityOpenIDClientSecret       = "MINIO_IDENTITY_OPENID_CLIENT_SECRET"
+	EnvIdentityOpenIDURL                = "MINIO_IDENTITY_OPENID_CONFIG_URL"
+	EnvIdentityOpenIDClaimName          = "MINIO_IDENTITY_OPENID_CLAIM_NAME"
+	EnvIdentityOpenIDClaimUserInfo      = "MINIO_IDENTITY_OPENID_CLAIM_USERINFO"
+	EnvIdentityOpenIDClaimPrefix        = "MINIO_IDENTITY_OPENID_CLAIM_PREFIX"
+	EnvIdentityOpenIDRolePolicy         = "MINIO_IDENTITY_OPENID_ROLE_POLICY"
+	EnvIdentityOpenIDRedirectURI        = "MINIO_IDENTITY_OPENID_REDIRECT_URI"
+	EnvIdentityOpenIDRedirectURIDynamic = "MINIO_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC"
+	EnvIdentityOpenIDScopes             = "MINIO_IDENTITY_OPENID_SCOPES"
+	EnvIdentityOpenIDDisplayName        = "MINIO_IDENTITY_OPENID_DISPLAY_NAME"
+
+	// Vendor specific ENVs only enabled if the Vendor matches == "vendor"
+	EnvIdentityOpenIDKeyCloakRealm    = "MINIO_IDENTITY_OPENID_KEYCLOAK_REALM"
+	EnvIdentityOpenIDKeyCloakAdminURL = "MINIO_IDENTITY_OPENID_KEYCLOAK_ADMIN_URL"
+)
+
+// DefaultKVS - default config for OpenID config
+var (
+	DefaultKVS = config.KVS{
+		config.KV{
+			Key:   config.Enable,
+			Value: "",
+		},
+		config.KV{
+			Key:   DisplayName,
+			Value: "",
+		},
+		config.KV{
+			Key:   ConfigURL,
+			Value: "",
+		},
+		config.KV{
+			Key:   ClientID,
+			Value: "",
+		},
+		config.KV{
+			Key:   ClientSecret,
+			Value: "",
+		},
+		config.KV{
+			Key:   ClaimName,
+			Value: iampolicy.PolicyName,
+		},
+		config.KV{
+			Key:   ClaimUserinfo,
+			Value: "",
+		},
+		config.KV{
+			Key:   RolePolicy,
+			Value: "",
+		},
+		config.KV{
+			Key:   ClaimPrefix,
+			Value: "",
+		},
+		config.KV{
+			Key:   RedirectURI,
+			Value: "",
+		},
+		config.KV{
+			Key:   RedirectURIDynamic,
+			Value: "off",
+		},
+		config.KV{
+			Key:   Scopes,
+			Value: "",
+		},
+	}
+)
+
+var errSingleProvider = config.Errorf("Only one OpenID provider can be configured if not using role policy mapping")
+
+// DummyRoleARN is used to indicate that the user associated with it was
+// authenticated via policy-claim based OpenID provider.
+var DummyRoleARN = func() arn.ARN {
+	v, err := arn.NewIAMRoleARN("dummy-internal", "")
+	if err != nil {
+		panic("should not happen!")
+	}
+	return v
+}()
+
+// Config - OpenID Config
+type Config struct {
+	Enabled bool
+
+	// map of roleARN to providerCfg's
+	arnProviderCfgsMap map[arn.ARN]*providerCfg
+
+	// map of config names to providerCfg's
+	ProviderCfgs map[string]*providerCfg
+
+	pubKeys          publicKeys
+	roleArnPolicyMap map[arn.ARN]string
+
+	transport   http.RoundTripper
+	closeRespFn func(io.ReadCloser)
+}
+
+// LookupConfig lookup jwks from config, override with any ENVs.
+func LookupConfig(kvsMap map[string]config.KVS, transport http.RoundTripper, closeRespFn func(io.ReadCloser), serverRegion string) (c Config, err error) {
+	openIDClientTransport := http.DefaultTransport
+	if transport != nil {
+		openIDClientTransport = transport
+	}
+	c = Config{
+		Enabled:            false,
+		arnProviderCfgsMap: map[arn.ARN]*providerCfg{},
+		ProviderCfgs:       map[string]*providerCfg{},
+		pubKeys: publicKeys{
+			RWMutex: &sync.RWMutex{},
+			pkMap:   map[string]crypto.PublicKey{},
+		},
+		roleArnPolicyMap: map[arn.ARN]string{},
+		transport:        openIDClientTransport,
+		closeRespFn:      closeRespFn,
+	}
+
+	// Make a copy of the config we received so we can mutate it safely.
+	kvsMap2 := make(map[string]config.KVS, len(kvsMap))
+	for k, v := range kvsMap {
+		kvsMap2[k] = v
+	}
+
+	// Add in each configuration name found from environment variables, i.e.
+	// if we see MINIO_IDENTITY_OPENID_CONFIG_URL_2, we add the key "2" to
+	// `kvsMap2` if it does not already exist.
+	envs := env.List(EnvIdentityOpenIDURL + config.Default)
+	for _, k := range envs {
+		cfgName := strings.TrimPrefix(k, EnvIdentityOpenIDURL+config.Default)
+		if cfgName == "" {
+			return c, config.Errorf("Environment variable must have a non-empty config name: %s", k)
+		}
+
+		// It is possible that some variables were specified via config
+		// commands and some variables are intended to be overridden
+		// from the environment, so we ensure that the key is not
+		// overwritten in `kvsMap2` as it may have existing config.
+		if _, ok := kvsMap2[cfgName]; !ok {
+			kvsMap2[cfgName] = DefaultKVS
+		}
+	}
+
+	var (
+		hasLegacyPolicyMapping = false
+		seenClientIDs          = set.NewStringSet()
+	)
+	for cfgName, kvs := range kvsMap2 {
+		// remove this since we have removed support for this already.
+		kvs.Delete(JwksURL)
+
+		if err = config.CheckValidKeys(config.IdentityOpenIDSubSys, kvs, DefaultKVS); err != nil {
+			return c, err
+		}
+
+		getCfgVal := func(envVar, cfgParam string) string {
+			if cfgName != config.Default {
+				envVar += config.Default + cfgName
+			}
+			return env.Get(envVar, kvs.Get(cfgParam))
+		}
+
+		// In the past, when only one openID provider was allowed, there
+		// was no `enable` parameter - the configuration is turned off
+		// by clearing the values. With multiple providers, we support
+		// individually enabling/disabling provider configurations. If
+		// the enable parameter's value is non-empty, we use that
+		// setting, otherwise we treat it as enabled if some important
+		// parameters are non-empty.
+		var (
+			cfgEnableVal        = getCfgVal(EnvIdentityOpenIDEnable, config.Enable)
+			isExplicitlyEnabled = false
+		)
+		if cfgEnableVal != "" {
+			isExplicitlyEnabled = true
+		}
+
+		var enabled bool
+		if isExplicitlyEnabled {
+			enabled, err = config.ParseBool(cfgEnableVal)
+			if err != nil {
+				return c, err
+			}
+			// No need to continue loading if the config is not enabled.
+			if !enabled {
+				continue
+			}
+		}
+
+		p := newProviderCfgFromConfig(getCfgVal)
+		configURL := getCfgVal(EnvIdentityOpenIDURL, ConfigURL)
+
+		if !isExplicitlyEnabled {
+			enabled = true
+			if p.ClientID == "" && p.ClientSecret == "" && configURL == "" {
+				enabled = false
+			}
+		}
+
+		// No need to continue loading if the config is not enabled.
+		if !enabled {
+			continue
+		}
+
+		// Validate that client ID has not been duplicately specified.
+		if seenClientIDs.Contains(p.ClientID) {
+			return c, config.Errorf("Client ID %s is present with multiple OpenID configurations", p.ClientID)
+		}
+		seenClientIDs.Add(p.ClientID)
+
+		p.URL, err = xnet.ParseHTTPURL(configURL)
+		if err != nil {
+			return c, err
+		}
+		configURLDomain := p.URL.Hostname()
+		p.DiscoveryDoc, err = parseDiscoveryDoc(p.URL, transport, closeRespFn)
+		if err != nil {
+			return c, err
+		}
+
+		if p.ClaimUserinfo && configURL == "" {
+			return c, errors.New("please specify config_url to enable fetching claims from UserInfo endpoint")
+		}
+
+		if scopeList := getCfgVal(EnvIdentityOpenIDScopes, Scopes); scopeList != "" {
+			var scopes []string
+			for _, scope := range strings.Split(scopeList, ",") {
+				scope = strings.TrimSpace(scope)
+				if scope == "" {
+					return c, config.Errorf("empty scope value is not allowed '%s', please refer to our documentation", scopeList)
+				}
+				scopes = append(scopes, scope)
+			}
+			// Replace the discovery document scopes by client customized scopes.
+			p.DiscoveryDoc.ScopesSupported = scopes
+		}
+
+		// Check if claim name is the non-default value and role policy is set.
+		if p.ClaimName != iampolicy.PolicyName && p.RolePolicy != "" {
+			// In the unlikely event that the user specifies
+			// `iampolicy.PolicyName` as the claim name explicitly and sets
+			// a role policy, this check is thwarted, but we will be using
+			// the role policy anyway.
+			return c, config.Errorf("Role Policy (=`%s`) and Claim Name (=`%s`) cannot both be set", p.RolePolicy, p.ClaimName)
+		}
+
+		jwksURL := p.DiscoveryDoc.JwksURI
+		if jwksURL == "" {
+			return c, config.Errorf("no JWKS URI found in your provider's discovery doc (config_url=%s)", configURL)
+		}
+
+		p.JWKS.URL, err = xnet.ParseHTTPURL(jwksURL)
+		if err != nil {
+			return c, err
+		}
+
+		if p.RolePolicy != "" {
+			// RolePolicy is validated by IAM System during its
+			// initialization.
+
+			// Generate role ARN as combination of provider domain and
+			// prefix of client ID.
+			domain := configURLDomain
+			if domain == "" {
+				// Attempt to parse the JWKs URI.
+				domain = p.JWKS.URL.Hostname()
+				if domain == "" {
+					return c, config.Errorf("unable to parse a domain from the OpenID config")
+				}
+			}
+			if p.ClientID == "" {
+				return c, config.Errorf("client ID must not be empty")
+			}
+
+			// We set the resource ID of the role arn as a hash of client
+			// ID, so we can get a short roleARN that stays the same on
+			// restart.
+			var resourceID string
+			{
+				h := sha1.New()
+				h.Write([]byte(p.ClientID))
+				bs := h.Sum(nil)
+				resourceID = base64.RawURLEncoding.EncodeToString(bs)
+			}
+			p.roleArn, err = arn.NewIAMRoleARN(resourceID, serverRegion)
+			if err != nil {
+				return c, config.Errorf("unable to generate ARN from the OpenID config: %v", err)
+			}
+
+			c.roleArnPolicyMap[p.roleArn] = p.RolePolicy
+		} else if p.ClaimName == "" {
+			return c, config.Errorf("A role policy or claim name must be specified")
+		}
+
+		if err = p.initializeProvider(getCfgVal, c.transport); err != nil {
+			return c, err
+		}
+
+		arnKey := p.roleArn
+		if p.RolePolicy == "" {
+			arnKey = DummyRoleARN
+			hasLegacyPolicyMapping = true
+			// Ensure that when a JWT policy claim based provider
+			// exists, it is the only one.
+			if _, ok := c.arnProviderCfgsMap[DummyRoleARN]; ok {
+				return c, errSingleProvider
+			}
+		}
+
+		c.arnProviderCfgsMap[arnKey] = &p
+		c.ProviderCfgs[cfgName] = &p
+
+		if err = c.PopulatePublicKey(arnKey); err != nil {
+			return c, err
+		}
+	}
+
+	// Ensure that when a JWT policy claim based provider
+	// exists, it is the only one.
+	if hasLegacyPolicyMapping && len(c.ProviderCfgs) > 1 {
+		return c, errSingleProvider
+	}
+
+	c.Enabled = true
+
+	return c, nil
+}
+
+// Enabled returns if configURL is enabled.
+func Enabled(kvs config.KVS) bool {
+	return kvs.Get(ConfigURL) != ""
+}
+
+// GetSettings - fetches OIDC settings for site-replication related validation.
+// NOTE that region must be populated by caller as this package does not know.
+func (r *Config) GetSettings() madmin.OpenIDSettings {
+	res := madmin.OpenIDSettings{}
+	if !r.Enabled {
+		return res
+	}
+
+	for arn, provCfg := range r.arnProviderCfgsMap {
+		hashedSecret := ""
+		{
+			h := sha256.New()
+			h.Write([]byte(provCfg.ClientSecret))
+			bs := h.Sum(nil)
+			hashedSecret = base64.RawURLEncoding.EncodeToString(bs)
+		}
+		if arn != DummyRoleARN {
+			if res.Roles != nil {
+				res.Roles = make(map[string]madmin.OpenIDProviderSettings)
+			}
+			res.Roles[arn.String()] = madmin.OpenIDProviderSettings{
+				ClaimUserinfoEnabled: provCfg.ClaimUserinfo,
+				RolePolicy:           provCfg.RolePolicy,
+				ClientID:             provCfg.ClientID,
+				HashedClientSecret:   hashedSecret,
+			}
+		} else {
+			res.ClaimProvider = madmin.OpenIDProviderSettings{
+				ClaimUserinfoEnabled: provCfg.ClaimUserinfo,
+				RolePolicy:           provCfg.RolePolicy,
+				ClientID:             provCfg.ClientID,
+				HashedClientSecret:   hashedSecret,
+			}
+		}
+
+	}
+
+	return res
+}
+
+// GetIAMPolicyClaimName - returns the policy claim name for the (at most one)
+// provider configured without a role policy.
+func (r *Config) GetIAMPolicyClaimName() string {
+	pCfg, ok := r.arnProviderCfgsMap[DummyRoleARN]
+	if !ok {
+		return ""
+	}
+	return pCfg.ClaimPrefix + pCfg.ClaimName
+}
+
+// LookupUser lookup userid for the provider
+func (r Config) LookupUser(roleArn, userid string) (provider.User, error) {
+	// Can safely ignore error here as empty or invalid ARNs will not be
+	// mapped.
+	arnVal, _ := arn.Parse(roleArn)
+	pCfg, ok := r.arnProviderCfgsMap[arnVal]
+	if ok {
+		user, err := pCfg.provider.LookupUser(userid)
+		if err != nil && err != provider.ErrAccessTokenExpired {
+			return user, err
+		}
+		if err == provider.ErrAccessTokenExpired {
+			if err = pCfg.provider.LoginWithClientID(pCfg.ClientID, pCfg.ClientSecret); err != nil {
+				return user, err
+			}
+			user, err = pCfg.provider.LookupUser(userid)
+		}
+		return user, err
+	}
+	// Without any specific logic for a provider, all accounts
+	// are always enabled.
+	return provider.User{ID: userid, Enabled: true}, nil
+}
+
+// ProviderEnabled returns true if any vendor specific provider is enabled.
+func (r Config) ProviderEnabled() bool {
+	if !r.Enabled {
+		return false
+	}
+	for _, v := range r.arnProviderCfgsMap {
+		if v.provider != nil {
+			return true
+		}
+	}
+	return false
+}
+
+// GetRoleInfo - returns ARN to policies map if a role policy based openID
+// provider is configured. Otherwise returns nil.
+func (r Config) GetRoleInfo() map[arn.ARN]string {
+	for _, p := range r.arnProviderCfgsMap {
+		if p.RolePolicy != "" {
+			return r.roleArnPolicyMap
+		}
+	}
+	return nil
+}
+
+// GetDefaultExpiration - returns the expiration seconds expected.
+func GetDefaultExpiration(dsecs string) (time.Duration, error) {
+	defaultExpiryDuration := time.Duration(60) * time.Minute // Defaults to 1hr.
+	if dsecs != "" {
+		expirySecs, err := strconv.ParseInt(dsecs, 10, 64)
+		if err != nil {
+			return 0, auth.ErrInvalidDuration
+		}
+
+		// The duration, in seconds, of the role session.
+		// The value can range from 900 seconds (15 minutes)
+		// up to 365 days.
+		if expirySecs < 900 || expirySecs > 31536000 {
+			return 0, auth.ErrInvalidDuration
+		}
+
+		defaultExpiryDuration = time.Duration(expirySecs) * time.Second
+	}
+	return defaultExpiryDuration, nil
+}

internal/config/identity/openid/providercfg.go

@@ -0,0 +1,148 @@
+// 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 openid
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+
+	"github.com/minio/minio/internal/arn"
+	"github.com/minio/minio/internal/config"
+	"github.com/minio/minio/internal/config/identity/openid/provider"
+	xhttp "github.com/minio/minio/internal/http"
+	xnet "github.com/minio/pkg/net"
+)
+
+type providerCfg struct {
+	// Used for user interface like console
+	DisplayName string
+
+	JWKS struct {
+		URL *xnet.URL
+	}
+	URL                *xnet.URL
+	ClaimPrefix        string
+	ClaimName          string
+	ClaimUserinfo      bool
+	RedirectURI        string
+	RedirectURIDynamic bool
+	DiscoveryDoc       DiscoveryDoc
+	ClientID           string
+	ClientSecret       string
+	RolePolicy         string
+
+	roleArn  arn.ARN
+	provider provider.Provider
+}
+
+func newProviderCfgFromConfig(getCfgVal func(env, cfgName string) string) providerCfg {
+	return providerCfg{
+		DisplayName:        getCfgVal(EnvIdentityOpenIDDisplayName, DisplayName),
+		ClaimName:          getCfgVal(EnvIdentityOpenIDClaimName, ClaimName),
+		ClaimUserinfo:      getCfgVal(EnvIdentityOpenIDClaimUserInfo, ClaimUserinfo) == config.EnableOn,
+		ClaimPrefix:        getCfgVal(EnvIdentityOpenIDClaimPrefix, ClaimPrefix),
+		RedirectURI:        getCfgVal(EnvIdentityOpenIDRedirectURI, RedirectURI),
+		RedirectURIDynamic: getCfgVal(EnvIdentityOpenIDRedirectURIDynamic, RedirectURIDynamic) == config.EnableOn,
+		ClientID:           getCfgVal(EnvIdentityOpenIDClientID, ClientID),
+		ClientSecret:       getCfgVal(EnvIdentityOpenIDClientSecret, ClientSecret),
+		RolePolicy:         getCfgVal(EnvIdentityOpenIDRolePolicy, RolePolicy),
+	}
+}
+
+const (
+	keyCloakVendor = "keycloak"
+)
+
+// initializeProvider initializes if any additional vendor specific information
+// was provided, initialization will return an error initial login fails.
+func (p *providerCfg) initializeProvider(cfgGet func(string, string) string, transport http.RoundTripper) error {
+	vendor := cfgGet(EnvIdentityOpenIDVendor, Vendor)
+	if vendor == "" {
+		return nil
+	}
+	var err error
+	switch vendor {
+	case keyCloakVendor:
+		adminURL := cfgGet(EnvIdentityOpenIDKeyCloakAdminURL, KeyCloakAdminURL)
+		realm := cfgGet(EnvIdentityOpenIDKeyCloakRealm, KeyCloakRealm)
+		p.provider, err = provider.KeyCloak(
+			provider.WithAdminURL(adminURL),
+			provider.WithOpenIDConfig(provider.DiscoveryDoc(p.DiscoveryDoc)),
+			provider.WithTransport(transport),
+			provider.WithRealm(realm),
+		)
+		return err
+	default:
+		return fmt.Errorf("Unsupport vendor %s", keyCloakVendor)
+	}
+}
+
+// UserInfo returns claims for authenticated user from userInfo endpoint.
+//
+// Some OIDC implementations such as GitLab do not support
+// claims as part of the normal oauth2 flow, instead rely
+// on service providers making calls to IDP to fetch additional
+// claims available from the UserInfo endpoint
+func (p *providerCfg) UserInfo(accessToken string, transport http.RoundTripper) (map[string]interface{}, error) {
+	if p.JWKS.URL == nil || p.JWKS.URL.String() == "" {
+		return nil, errors.New("openid not configured")
+	}
+	client := &http.Client{
+		Transport: transport,
+	}
+
+	req, err := http.NewRequest(http.MethodPost, p.DiscoveryDoc.UserInfoEndpoint, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if accessToken != "" {
+		req.Header.Set("Authorization", "Bearer "+accessToken)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	defer xhttp.DrainBody(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		// uncomment this for debugging when needed.
+		// reqBytes, _ := httputil.DumpRequest(req, false)
+		// fmt.Println(string(reqBytes))
+		// respBytes, _ := httputil.DumpResponse(resp, true)
+		// fmt.Println(string(respBytes))
+		return nil, errors.New(resp.Status)
+	}
+
+	dec := json.NewDecoder(resp.Body)
+	claims := map[string]interface{}{}
+
+	if err = dec.Decode(&claims); err != nil {
+		// uncomment this for debugging when needed.
+		// reqBytes, _ := httputil.DumpRequest(req, false)
+		// fmt.Println(string(reqBytes))
+		// respBytes, _ := httputil.DumpResponse(resp, true)
+		// fmt.Println(string(respBytes))
+		return nil, err
+	}
+
+	return claims, nil
+}

internal/config/notify/help.go

@@ -28,9 +28,18 @@ const (
 	queueLimitComment = `maximum limit for undelivered messages, defaults to '100000'`
 )
 
+var enableHelp = config.HelpKV{
+	Key:         config.Enable,
+	Description: "Enable or disable notifications",
+	Type:        "on|off",
+	Sensitive:   false,
+	Optional:    true,
+}
+
 // Help template inputs for all notification targets
 var (
 	HelpWebhook = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.WebhookEndpoint,
 			Description: "webhook server endpoint e.g. http://localhost:8080/minio/events",
@@ -79,6 +88,7 @@ var (
 	}
 
 	HelpAMQP = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.AmqpURL,
 			Description: "AMQP server endpoint e.g. `amqp://myuser:mypassword@localhost:5672`",
@@ -167,6 +177,7 @@ var (
 	}
 
 	HelpKafka = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.KafkaBrokers,
 			Description: "comma separated list of Kafka broker addresses",
@@ -263,6 +274,7 @@ var (
 	}
 
 	HelpMQTT = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.MqttBroker,
 			Description: "MQTT server endpoint e.g. `tcp://localhost:1883`",
@@ -327,6 +339,7 @@ var (
 	}
 
 	HelpPostgres = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.PostgresConnectionString,
 			Description: `Postgres server connection-string e.g. "host=localhost port=5432 dbname=minio_events user=postgres password=password sslmode=disable"`,
@@ -370,6 +383,7 @@ var (
 	}
 
 	HelpMySQL = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.MySQLDSNString,
 			Description: `MySQL data-source-name connection string e.g. "<user>:<password>@tcp(<host>:<port>)/<database>"`,
@@ -414,6 +428,7 @@ var (
 	}
 
 	HelpNATS = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.NATSAddress,
 			Description: "NATS server address e.g. '0.0.0.0:4222'",
@@ -530,6 +545,7 @@ var (
 	}
 
 	HelpNSQ = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.NSQAddress,
 			Description: "NSQ server address e.g. '127.0.0.1:4150'",
@@ -574,6 +590,7 @@ var (
 	}
 
 	HelpES = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.ElasticURL,
 			Description: "Elasticsearch server's address, with optional authentication info",
@@ -625,6 +642,7 @@ var (
 	}
 
 	HelpRedis = config.HelpKVS{
+		enableHelp,
 		config.HelpKV{
 			Key:         target.RedisAddress,
 			Description: "Redis server's address. For example: `localhost:6379`",

internal/config/policy/opa/help.go

@@ -28,13 +28,13 @@ var (
 	Help = config.HelpKVS{
 		config.HelpKV{
 			Key:         URL,
-			Description: `[DEPRECATED] OPA HTTP(s) endpoint e.g. "http://localhost:8181/v1/data/httpapi/authz/allow"` + defaultHelpPostfix(URL),
+			Description: `OPA HTTP(s) endpoint e.g. "http://localhost:8181/v1/data/httpapi/authz/allow"` + defaultHelpPostfix(URL),
 			Type:        "url",
 			Sensitive:   true,
 		},
 		config.HelpKV{
 			Key:         AuthToken,
-			Description: "[DEPRECATED] authorization token for OPA endpoint" + defaultHelpPostfix(AuthToken),
+			Description: "authorization token for OPA endpoint" + defaultHelpPostfix(AuthToken),
 			Optional:    true,
 			Type:        "string",
 			Sensitive:   true,

internal/logger/help.go

@@ -73,6 +73,13 @@ var (
 	}
 
 	HelpWebhook = config.HelpKVS{
+		config.HelpKV{
+			Key:         config.Enable,
+			Description: "set to 'on' to enable the audit logger",
+			Optional:    true,
+			Type:        "on|off",
+			Sensitive:   false,
+		},
 		config.HelpKV{
 			Key:         Endpoint,
 			Description: `HTTP(s) endpoint e.g. "http://localhost:8080/minio/logs/audit"`,
@@ -115,6 +122,13 @@ var (
 	}
 
 	HelpKafka = config.HelpKVS{
+		config.HelpKV{
+			Key:         config.Enable,
+			Description: "set to 'on' to enable kafka audit logging",
+			Optional:    true,
+			Type:        "on|off",
+			Sensitive:   false,
+		},
 		config.HelpKV{
 			Key:         KafkaBrokers,
 			Description: "comma separated list of Kafka broker addresses",