allow forcibly creating metadata on buckets (#14820)

introduce x-minio-force-create environment variable
to force create a bucket and its metadata as required,
it is useful in some situations when bucket metadata
needs recovery.

cmd/admin-handlers-site-replication.go

@@ -115,10 +115,12 @@ func (a adminAPIHandlers) SRPeerBucketOps(w http.ResponseWriter, r *http.Request
 	case madmin.MakeWithVersioningBktOp:
 		_, isLockEnabled := r.Form["lockEnabled"]
 		_, isVersioningEnabled := r.Form["versioningEnabled"]
+		_, isForceCreate := r.Form["forceCreate"]
 		opts := BucketOptions{
 			Location:          r.Form.Get("location"),
 			LockEnabled:       isLockEnabled,
 			VersioningEnabled: isVersioningEnabled,
+			ForceCreate:       isForceCreate,
 		}
 		err = globalSiteReplicationSys.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts)
 	case madmin.ConfigureReplBktOp:

cmd/bucket-handlers.go

@@ -706,13 +706,27 @@ func (api objectAPIHandlers) PutBucketHandler(w http.ResponseWriter, r *http.Req
 	bucket := vars["bucket"]
 
 	objectLockEnabled := false
-	if vs, found := r.Header[http.CanonicalHeaderKey("x-amz-bucket-object-lock-enabled")]; found {
-		v := strings.ToLower(strings.Join(vs, ""))
-		if v != "true" && v != "false" {
+	if vs := r.Header.Get(xhttp.AmzObjectLockEnabled); len(vs) > 0 {
+		v := strings.ToLower(vs)
+		switch v {
+		case "true", "false":
+			objectLockEnabled = v == "true"
+		default:
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
+			return
+		}
+	}
+
+	forceCreate := false
+	if vs := r.Header.Get(xhttp.MinIOForceCreate); len(vs) > 0 {
+		v := strings.ToLower(vs)
+		switch v {
+		case "true", "false":
+			forceCreate = v == "true"
+		default:
 			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
 			return
 		}
-		objectLockEnabled = v == "true"
 	}
 
 	if s3Error := checkRequestAuthType(ctx, r, policy.CreateBucketAction, bucket, ""); s3Error != ErrNone {
@@ -737,6 +751,7 @@ func (api objectAPIHandlers) PutBucketHandler(w http.ResponseWriter, r *http.Req
 	opts := BucketOptions{
 		Location:    location,
 		LockEnabled: objectLockEnabled,
+		ForceCreate: forceCreate,
 	}
 
 	if globalDNSConfig != nil {

cmd/erasure-bucket.go

@@ -55,6 +55,11 @@ func (er erasureObjects) MakeBucketWithLocation(ctx context.Context, bucket stri
 		g.Go(func() error {
 			if storageDisks[index] != nil {
 				if err := storageDisks[index].MakeVol(ctx, bucket); err != nil {
+					if opts.ForceCreate && errors.Is(err, errVolumeExists) {
+						// No need to return error when force create was
+						// requested.
+						return nil
+					}
 					if !errors.Is(err, errVolumeExists) {
 						logger.LogIf(ctx, err)
 					}

cmd/object-api-interface.go

@@ -98,6 +98,7 @@ type BucketOptions struct {
 	Location          string
 	LockEnabled       bool
 	VersioningEnabled bool
+	ForceCreate       bool // Create buckets even if they are already created.
 }
 
 // DeleteBucketOptions provides options for DeleteBucket calls.

cmd/site-replication.go

@@ -625,6 +625,9 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string,
 	if opts.VersioningEnabled {
 		optsMap["versioningEnabled"] = "true"
 	}
+	if opts.ForceCreate {
+		optsMap["forceCreate"] = "true"
+	}
 
 	// Create bucket and enable versioning on all peers.
 	makeBucketConcErr := c.concDo(