Add support for syncing replica modifications (#11104)

when bidirectional replication is set up.

If ReplicaModifications is enabled in the replication
configuration, sync metadata updates to source if
replication rules are met. By default, if this
configuration is unset, MinIO automatically sync's
metadata updates on replica back to the source.

pkg/bucket/replication/replication.go

@@ -58,6 +58,7 @@ var (
 	errReplicationUniquePriority      = Errorf("Replication configuration has duplicate priority")
 	errReplicationDestinationMismatch = Errorf("The destination bucket must be same for all rules")
 	errRoleArnMissing                 = Errorf("Missing required parameter `Role` in ReplicationConfiguration")
+	errInvalidSourceSelectionCriteria = Errorf("Invalid ReplicaModification status")
 )
 
 // Config - replication configuration specified in
@@ -78,6 +79,16 @@ func ParseConfig(reader io.Reader) (*Config, error) {
 	if err := xml.NewDecoder(io.LimitReader(reader, maxReplicationConfigSize)).Decode(&config); err != nil {
 		return nil, err
 	}
+	// By default, set replica modification to enabled if unset.
+	for i := range config.Rules {
+		if len(config.Rules[i].SourceSelectionCriteria.ReplicaModifications.Status) == 0 {
+			config.Rules[i].SourceSelectionCriteria = SourceSelectionCriteria{
+				ReplicaModifications: ReplicaModifications{
+					Status: Enabled,
+				},
+			}
+		}
+	}
 	return &config, nil
 }
 
@@ -136,6 +147,7 @@ type ObjectOpts struct {
 	DeleteMarker bool
 	SSEC         bool
 	OpType       Type
+	Replica      bool
 }
 
 // FilterActionableRules returns the rules actions that need to be executed
@@ -187,9 +199,8 @@ func (c Config) Replicate(obj ObjectOpts) bool {
 			default:
 				return rule.DeleteMarkerReplication.Status == Enabled
 			}
-		} else { // regular object/metadata replication
-			return true
-		}
+		} // regular object/metadata replication
+		return rule.MetadataReplicate(obj)
 	}
 	return false
 }

pkg/bucket/replication/replication_test.go

@@ -140,7 +140,7 @@ func TestParseAndValidateReplicationConfig(t *testing.T) {
 }
 func TestReplicate(t *testing.T) {
 	cfgs := []Config{
-		{ //Config0 - Replication config has no filters, all replication enabled
+		{ // Config0 - Replication config has no filters, all replication enabled
 			Rules: []Rule{
 				{
 					Status:                  Enabled,
@@ -151,7 +151,7 @@ func TestReplicate(t *testing.T) {
 				},
 			},
 		},
-		{ //Config1 - Replication config has no filters, delete,delete-marker replication disabled
+		{ // Config1 - Replication config has no filters, delete,delete-marker replication disabled
 			Rules: []Rule{
 				{
 					Status:                  Enabled,
@@ -162,7 +162,7 @@ func TestReplicate(t *testing.T) {
 				},
 			},
 		},
-		{ //Config2 - Replication config has filters and more than 1 matching rule, delete,delete-marker replication disabled
+		{ // Config2 - Replication config has filters and more than 1 matching rule, delete,delete-marker replication disabled
 			Rules: []Rule{
 				{
 					Status:                  Enabled,
@@ -180,7 +180,7 @@ func TestReplicate(t *testing.T) {
 				},
 			},
 		},
-		{ //Config3 - Replication config has filters and no overlapping rules
+		{ // Config3 - Replication config has filters and no overlapping rules
 			Rules: []Rule{
 				{
 					Status:                  Enabled,
@@ -198,6 +198,17 @@ func TestReplicate(t *testing.T) {
 				},
 			},
 		},
+		{ // Config4 - Replication config has filters and SourceSelectionCriteria Disabled
+			Rules: []Rule{
+				{
+					Status:                  Enabled,
+					Priority:                2,
+					DeleteMarkerReplication: DeleteMarkerReplication{Status: Enabled},
+					DeleteReplication:       DeleteReplication{Status: Enabled},
+					SourceSelectionCriteria: SourceSelectionCriteria{ReplicaModifications: ReplicaModifications{Status: Disabled}},
+				},
+			},
+		},
 	}
 	testCases := []struct {
 		opts           ObjectOpts
@@ -249,7 +260,9 @@ func TestReplicate(t *testing.T) {
 		{ObjectOpts{Name: "abc/c4test", DeleteMarker: true, OpType: DeleteReplicationType}, cfgs[3], true},                                      //36. matches rule 2 - DeleteMarker replication allowed by rule
 		{ObjectOpts{Name: "abc/c4test", DeleteMarker: true, VersionID: "vid", OpType: DeleteReplicationType}, cfgs[3], false},                   //37. matches rule 2 - DeleteReplication disallowed by rule for permanent delete of DeleteMarker
 		{ObjectOpts{Name: "abc/c4test", VersionID: "vid", OpType: DeleteReplicationType}, cfgs[3], false},                                       //38. matches rule 2 - DeleteReplication disallowed by rule for permanent delete of version
-
+		//  using config 4 - with replica modification sync disabled.
+		{ObjectOpts{Name: "xy/c5test", UserTags: "k1=v1", Replica: true}, cfgs[4], false}, //39. replica syncing disabled, this object is a replica
+		{ObjectOpts{Name: "xa/c5test", UserTags: "k1=v1", Replica: false}, cfgs[4], true}, //40. replica syncing disabled, this object is NOT a replica
 	}
 
 	for i, testCase := range testCases {

pkg/bucket/replication/rule.go

@@ -98,9 +98,10 @@ type Rule struct {
 	Priority                int                     `xml:"Priority" json:"Priority"`
 	DeleteMarkerReplication DeleteMarkerReplication `xml:"DeleteMarkerReplication" json:"DeleteMarkerReplication"`
 	// MinIO extension to replicate versioned deletes
-	DeleteReplication DeleteReplication `xml:"DeleteReplication" json:"DeleteReplication"`
-	Destination       Destination       `xml:"Destination" json:"Destination"`
-	Filter            Filter            `xml:"Filter" json:"Filter"`
+	DeleteReplication       DeleteReplication       `xml:"DeleteReplication" json:"DeleteReplication"`
+	Destination             Destination             `xml:"Destination" json:"Destination"`
+	SourceSelectionCriteria SourceSelectionCriteria `xml:"SourceSelectionCriteria" json:"SourceSelectionCriteria"`
+	Filter                  Filter                  `xml:"Filter" json:"Filter"`
 }
 
 var (
@@ -192,6 +193,10 @@ func (r Rule) Validate(bucket string, sameTarget bool) error {
 	if err := r.DeleteReplication.Validate(); err != nil {
 		return err
 	}
+	if err := r.SourceSelectionCriteria.Validate(); err != nil {
+		return err
+	}
+
 	if r.Priority < 0 {
 		return errPriorityMissing
 	}
@@ -200,3 +205,12 @@ func (r Rule) Validate(bucket string, sameTarget bool) error {
 	}
 	return nil
 }
+
+// MetadataReplicate  returns true if object is not a replica or in the case of replicas,
+// replica modification sync is enabled.
+func (r Rule) MetadataReplicate(obj ObjectOpts) bool {
+	if !obj.Replica {
+		return true
+	}
+	return obj.Replica && r.SourceSelectionCriteria.ReplicaModifications.Status == Enabled
+}

pkg/bucket/replication/rule_test.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2015-2021 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 replication
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+)
+
+func TestMetadataReplicate(t *testing.T) {
+	testCases := []struct {
+		inputConfig    string
+		opts           ObjectOpts
+		expectedResult bool
+	}{
+		// case 1 - rule with replica modification enabled; not a replica
+		{inputConfig: `<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Role>arn:aws:iam::AcctID:role/role-name</Role><Rule><Status>Enabled</Status><DeleteMarkerReplication><Status>Disabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Disabled</Status></DeleteReplication><Prefix>key-prefix</Prefix><Destination><Bucket>arn:aws:s3:::destinationbucket</Bucket></Destination><SourceSelectionCriteria><ReplicaModifications><Status>Enabled</Status></ReplicaModifications></SourceSelectionCriteria></Rule></ReplicationConfiguration>`,
+			opts:           ObjectOpts{Name: "c1test", DeleteMarker: false, OpType: ObjectReplicationType, Replica: false}, //1. Replica mod sync enabled; not a replica
+			expectedResult: true,
+		},
+		// case 2 - rule with replica modification disabled; a replica
+		{inputConfig: `<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Role>arn:aws:iam::AcctID:role/role-name</Role><Rule><Status>Enabled</Status><DeleteMarkerReplication><Status>Disabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Disabled</Status></DeleteReplication><Prefix>key-prefix</Prefix><Destination><Bucket>arn:aws:s3:::destinationbucket</Bucket></Destination><SourceSelectionCriteria><ReplicaModifications><Status>Disabled</Status></ReplicaModifications></SourceSelectionCriteria></Rule></ReplicationConfiguration>`,
+			opts:           ObjectOpts{Name: "c2test", DeleteMarker: false, OpType: ObjectReplicationType, Replica: true}, //1. Replica mod sync enabled; a replica
+			expectedResult: false,
+		},
+		// case 3 - rule with replica modification disabled; not a replica
+		{inputConfig: `<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Role>arn:aws:iam::AcctID:role/role-name</Role><Rule><Status>Enabled</Status><DeleteMarkerReplication><Status>Disabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Disabled</Status></DeleteReplication><Prefix>key-prefix</Prefix><Destination><Bucket>arn:aws:s3:::destinationbucket</Bucket></Destination><SourceSelectionCriteria><ReplicaModifications><Status>Disabled</Status></ReplicaModifications></SourceSelectionCriteria></Rule></ReplicationConfiguration>`,
+			opts:           ObjectOpts{Name: "c2test", DeleteMarker: false, OpType: ObjectReplicationType, Replica: false}, //1. Replica mod sync disabled; not a replica
+			expectedResult: true,
+		},
+
+		// case 4 - rule with replica modification enabled; a replica
+		{inputConfig: `<ReplicationConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Role>arn:aws:iam::AcctID:role/role-name</Role><Rule><Status>Enabled</Status><DeleteMarkerReplication><Status>Disabled</Status></DeleteMarkerReplication><DeleteReplication><Status>Disabled</Status></DeleteReplication><Prefix>key-prefix</Prefix><Destination><Bucket>arn:aws:s3:::destinationbucket</Bucket></Destination><SourceSelectionCriteria><ReplicaModifications><Status>Enabled</Status></ReplicaModifications></SourceSelectionCriteria></Rule></ReplicationConfiguration>`,
+			opts:           ObjectOpts{Name: "c2test", DeleteMarker: false, OpType: MetadataReplicationType, Replica: true}, //1. Replica mod sync enabled;  a replica
+			expectedResult: true,
+		},
+	}
+
+	for i, tc := range testCases {
+		tc := tc
+		t.Run(fmt.Sprintf("Test_%d", i+1), func(t *testing.T) {
+			cfg, err := ParseConfig(bytes.NewReader([]byte(tc.inputConfig)))
+			if err != nil {
+				t.Fatalf("Got unexpected error: %v", err)
+			}
+			if got := cfg.Rules[0].MetadataReplicate(tc.opts); got != tc.expectedResult {
+				t.Fatalf("Expected result with recursive set to false: `%v`, got: `%v`", tc.expectedResult, got)
+			}
+		})
+
+	}
+}

pkg/bucket/replication/sourceselectioncriteria.go

@@ -0,0 +1,76 @@
+// Copyright (c) 2015-2021 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 replication
+
+import (
+	"encoding/xml"
+)
+
+// ReplicaModifications specifies if replica modification sync is enabled
+type ReplicaModifications struct {
+	Status Status `xml:"Status" json:"Status"`
+}
+
+// SourceSelectionCriteria - specifies additional source selection criteria in ReplicationConfiguration.
+type SourceSelectionCriteria struct {
+	ReplicaModifications ReplicaModifications `xml:"ReplicaModifications" json:"ReplicaModifications"`
+}
+
+// IsValid - checks whether SourceSelectionCriteria is valid or not.
+func (s SourceSelectionCriteria) IsValid() bool {
+	return s.ReplicaModifications.Status == Enabled || s.ReplicaModifications.Status == Disabled
+}
+
+// Validate source selection criteria
+func (s SourceSelectionCriteria) Validate() error {
+	if (s == SourceSelectionCriteria{}) {
+		return nil
+	}
+	if !s.IsValid() {
+		return errInvalidSourceSelectionCriteria
+	}
+	return nil
+}
+
+// UnmarshalXML - decodes XML data.
+func (s *SourceSelectionCriteria) UnmarshalXML(dec *xml.Decoder, start xml.StartElement) (err error) {
+	// Make subtype to avoid recursive UnmarshalXML().
+	type sourceSelectionCriteria SourceSelectionCriteria
+	ssc := sourceSelectionCriteria{}
+	if err := dec.DecodeElement(&ssc, &start); err != nil {
+		return err
+	}
+	if len(ssc.ReplicaModifications.Status) == 0 {
+		ssc.ReplicaModifications.Status = Enabled
+	}
+	*s = SourceSelectionCriteria(ssc)
+	return nil
+}
+
+// MarshalXML - encodes to XML data.
+func (s SourceSelectionCriteria) MarshalXML(e *xml.Encoder, start xml.StartElement) error {
+	if err := e.EncodeToken(start); err != nil {
+		return err
+	}
+	if s.IsValid() {
+		if err := e.EncodeElement(s.ReplicaModifications, xml.StartElement{Name: xml.Name{Local: "ReplicaModifications"}}); err != nil {
+			return err
+		}
+	}
+	return e.EncodeToken(xml.EndElement{Name: start.Name})
+}