replication: add validation API (#1844)

api-bucket-replication.go

@@ -325,3 +325,31 @@ func (c *Client) GetBucketReplicationMetricsV2(ctx context.Context, bucketName s
 	}
 	return s, nil
 }
+
+// CheckBucketReplication validates if replication is set up properly for a bucket
+func (c *Client) CheckBucketReplication(ctx context.Context, bucketName string) (err error) {
+	// Input validation.
+	if err := s3utils.CheckValidBucketName(bucketName); err != nil {
+		return err
+	}
+	// Get resources properly escaped and lined up before
+	// using them in http request.
+	urlValues := make(url.Values)
+	urlValues.Set("replication-check", "")
+
+	// Execute GET on bucket to get replication config.
+	resp, err := c.executeMethod(ctx, http.MethodGet, requestMetadata{
+		bucketName:  bucketName,
+		queryValues: urlValues,
+	})
+
+	defer closeResponse(resp)
+	if err != nil {
+		return err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return httpRespToErrorResponse(resp, bucketName, "")
+	}
+	return nil
+}

api-compose-object.go

@@ -222,6 +222,9 @@ func (c *Client) copyObjectDo(ctx context.Context, srcBucket, srcObject, destBuc
 	if dstOpts.Internal.ReplicationRequest {
 		headers.Set(minIOBucketReplicationRequest, "true")
 	}
+	if dstOpts.Internal.ReplicationValidityCheck {
+		headers.Set(minIOBucketReplicationCheck, "true")
+	}
 	if !dstOpts.Internal.LegalholdTimestamp.IsZero() {
 		headers.Set(minIOBucketReplicationObjectLegalHoldTimestamp, dstOpts.Internal.LegalholdTimestamp.Format(time.RFC3339Nano))
 	}

api-put-object.go

@@ -56,14 +56,15 @@ func (r ReplicationStatus) Empty() bool {
 // AdvancedPutOptions for internal use - to be utilized by replication, ILM transition
 // implementation on MinIO server
 type AdvancedPutOptions struct {
-	SourceVersionID    string
-	SourceETag         string
-	ReplicationStatus  ReplicationStatus
-	SourceMTime        time.Time
-	ReplicationRequest bool
-	RetentionTimestamp time.Time
-	TaggingTimestamp   time.Time
-	LegalholdTimestamp time.Time
+	SourceVersionID          string
+	SourceETag               string
+	ReplicationStatus        ReplicationStatus
+	SourceMTime              time.Time
+	ReplicationRequest       bool
+	RetentionTimestamp       time.Time
+	TaggingTimestamp         time.Time
+	LegalholdTimestamp       time.Time
+	ReplicationValidityCheck bool
 }
 
 // PutObjectOptions represents options specified by user for PutObject call
@@ -188,6 +189,9 @@ func (opts PutObjectOptions) Header() (header http.Header) {
 	if opts.Internal.ReplicationRequest {
 		header.Set(minIOBucketReplicationRequest, "true")
 	}
+	if opts.Internal.ReplicationValidityCheck {
+		header.Set(minIOBucketReplicationCheck, "true")
+	}
 	if !opts.Internal.LegalholdTimestamp.IsZero() {
 		header.Set(minIOBucketReplicationObjectLegalHoldTimestamp, opts.Internal.LegalholdTimestamp.Format(time.RFC3339Nano))
 	}

api-remove.go

@@ -112,10 +112,11 @@ func (c *Client) RemoveBucket(ctx context.Context, bucketName string) error {
 
 // AdvancedRemoveOptions intended for internal use by replication
 type AdvancedRemoveOptions struct {
-	ReplicationDeleteMarker bool
-	ReplicationStatus       ReplicationStatus
-	ReplicationMTime        time.Time
-	ReplicationRequest      bool
+	ReplicationDeleteMarker  bool
+	ReplicationStatus        ReplicationStatus
+	ReplicationMTime         time.Time
+	ReplicationRequest       bool
+	ReplicationValidityCheck bool // check permissions
 }
 
 // RemoveObjectOptions represents options specified by user for RemoveObject call
@@ -168,6 +169,9 @@ func (c *Client) removeObject(ctx context.Context, bucketName, objectName string
 	if opts.Internal.ReplicationRequest {
 		headers.Set(minIOBucketReplicationRequest, "true")
 	}
+	if opts.Internal.ReplicationValidityCheck {
+		headers.Set(minIOBucketReplicationCheck, "true")
+	}
 	if opts.ForceDelete {
 		headers.Set(minIOForceDelete, "true")
 	}

constants.go

@@ -94,6 +94,8 @@ const (
 	minIOBucketReplicationDeleteMarker = "X-Minio-Source-DeleteMarker"
 	minIOBucketReplicationProxyRequest = "X-Minio-Source-Proxy-Request"
 	minIOBucketReplicationRequest      = "X-Minio-Source-Replication-Request"
+	minIOBucketReplicationCheck        = "X-Minio-Source-Replication-Check"
+
 	// Header indicates last tag update time on source
 	minIOBucketReplicationTaggingTimestamp = "X-Minio-Source-Replication-Tagging-Timestamp"
 	// Header indicates last retention update time on source

examples/minio/checkbucketreplication.go

@@ -0,0 +1,54 @@
+//go:build example
+// +build example
+
+/*
+ * MinIO Go Library for Amazon S3 Compatible Cloud Storage
+ * Copyright 2020 MinIO, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+)
+
+func main() {
+	// Note: YOUR-ACCESSKEYID, YOUR-SECRETACCESSKEY and my-bucketname are
+	// dummy values, please replace them with original values.
+
+	// Requests are always secure (HTTPS) by default. Set secure=false to enable insecure (HTTP) access.
+	// This boolean value is the last argument for New().
+
+	// New returns an Amazon S3 compatible client object. API compatibility (v2 or v4) is automatically
+	// determined based on the Endpoint value.
+	s3Client, err := minio.New("play.min.io", &minio.Options{
+		Creds:  credentials.NewStaticV4("YOUR-ACCESSKEYID", "YOUR-SECRETACCESSKEY", ""),
+		Secure: true,
+	})
+	if err != nil {
+		log.Fatalln(err)
+	}
+	// s3Client.TraceOn(os.Stderr)
+	// Get bucket replication configuration from S3
+	err = s3Client.CheckBucketReplication(context.Background(), "bucket")
+	if err != nil {
+		log.Fatalln(err)
+	}
+	log.Println("Bucket replication configuration is valid")
+}