Merge pull request #91 from pgsty/codex/issue-77-status-accounting

fix: report site replication metadata per site

cmd/site-replication-status-accounting_test.go

@@ -0,0 +1,191 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+)
+
+func TestSiteReplicationStatusAccountsPerSiteAndSurvivesMalformedConfig(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testSiteReplicationStatusAccountsPerSiteAndSurvivesMalformedConfig,
+	})
+}
+
+func testSiteReplicationStatusAccountsPerSiteAndSurvivesMalformedConfig(obj ObjectLayer, instanceType, localBucket string,
+	_ http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+	ctx := t.Context()
+	remoteBucket := getRandomBucketName()
+	if err := obj.MakeBucket(ctx, remoteBucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	remoteBucketMeta, err := loadBucketMetadata(ctx, obj, remoteBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	globalBucketMetadataSys.Set(remoteBucket, remoteBucketMeta)
+	globalNotificationSys.LoadBucketMetadata(ctx, remoteBucket)
+
+	tagXML := []byte(`<Tagging><TagSet><Tag><Key>key</Key><Value>value</Value></Tag></TagSet></Tagging>`)
+	versioningXML := []byte(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Enabled</Status></VersioningConfiguration>`)
+	objectLockXML := []byte(`<ObjectLockConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><ObjectLockEnabled>Enabled</ObjectLockEnabled><Rule><DefaultRetention><Mode>GOVERNANCE</Mode><Days>30</Days></DefaultRetention></Rule></ObjectLockConfiguration>`)
+	sseXML := []byte(`<ServerSideEncryptionConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Rule><ApplyServerSideEncryptionByDefault><SSEAlgorithm>AES256</SSEAlgorithm></ApplyServerSideEncryptionByDefault></Rule></ServerSideEncryptionConfiguration>`)
+	quotaJSON, err := json.Marshal(madmin.BucketQuota{Type: madmin.HardQuota, Quota: 1024})
+	if err != nil {
+		t.Fatal(err)
+	}
+	policyJSON := []byte(fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["*"]},"Action":["s3:GetObject"],"Resource":["arn:aws:s3:::%s/*"]}]}`, localBucket))
+
+	for configFile, data := range map[string][]byte{
+		bucketTaggingConfig:    tagXML,
+		bucketVersioningConfig: versioningXML,
+		objectLockConfig:       objectLockXML,
+		bucketSSEConfig:        sseXML,
+		bucketQuotaConfigFile:  quotaJSON,
+		bucketPolicyConfig:     policyJSON,
+		bucketCorsConfig:       []byte(testSiteReplicationCORSDoc),
+	} {
+		if _, err := globalBucketMetadataSys.Update(ctx, localBucket, configFile, data); err != nil {
+			t.Fatalf("%s: update %s: %v", instanceType, configFile, err)
+		}
+	}
+
+	encode := func(data []byte) *string {
+		encoded := base64.StdEncoding.EncodeToString(data)
+		return &encoded
+	}
+	remotePolicy := []byte(fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["*"]},"Action":["s3:GetObject"],"Resource":["arn:aws:s3:::%s/*"]}]}`, remoteBucket))
+	localMeta, err := globalBucketMetadataSys.Get(localBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	remoteInfo := madmin.SRInfo{
+		DeploymentID: "remote-status-accounting",
+		Buckets: map[string]madmin.SRBucketInfo{
+			localBucket: {
+				Bucket:    localBucket,
+				CreatedAt: localMeta.Created,
+			},
+			remoteBucket: {
+				Bucket:           remoteBucket,
+				CreatedAt:        remoteBucketMeta.Created,
+				Tags:             encode(tagXML),
+				Versioning:       encode(versioningXML),
+				ObjectLockConfig: encode(objectLockXML),
+				SSEConfig:        encode(sseXML),
+				QuotaConfig:      encode(quotaJSON),
+				Policy:           remotePolicy,
+				CorsConfig:       encode([]byte(testSiteReplicationCORSDoc)),
+			},
+		},
+	}
+	remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		if err := json.NewEncoder(w).Encode(remoteInfo); err != nil {
+			t.Errorf("%s: encode remote metadata: %v", instanceType, err)
+		}
+	}))
+	defer remote.Close()
+
+	serviceCred, err := auth.CreateCredentials("status-accounting-svc", "status-accounting-service-secret")
+	if err != nil {
+		t.Fatal(err)
+	}
+	serviceCred.ParentUser = credentials.AccessKey
+	if _, err = globalIAMSys.store.AddServiceAccount(ctx, serviceCred); err != nil {
+		t.Fatal(err)
+	}
+	defer globalIAMSys.DeleteServiceAccount(ctx, serviceCred.AccessKey, false)
+
+	localID := globalDeploymentID()
+	remoteID := remoteInfo.DeploymentID
+	globalSiteReplicationSys.Lock()
+	oldEnabled := globalSiteReplicationSys.enabled
+	oldState := globalSiteReplicationSys.state
+	globalSiteReplicationSys.enabled = true
+	globalSiteReplicationSys.state = srState{
+		Name:                    "status-accounting-test",
+		ServiceAccountAccessKey: serviceCred.AccessKey,
+		Peers: map[string]madmin.PeerInfo{
+			localID:  {Name: "local", DeploymentID: localID},
+			remoteID: {Name: "remote", DeploymentID: remoteID, Endpoint: remote.URL},
+		},
+	}
+	globalSiteReplicationSys.Unlock()
+	defer func() {
+		globalSiteReplicationSys.Lock()
+		globalSiteReplicationSys.enabled = oldEnabled
+		globalSiteReplicationSys.state = oldState
+		globalSiteReplicationSys.Unlock()
+	}()
+
+	check := func(name string, wantRemoteTags, wantRemoteQuota int) {
+		t.Helper()
+		status, err := globalSiteReplicationSys.siteReplicationStatus(ctx, obj, madmin.SRStatusOptions{Buckets: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+		local := status.StatsSummary[localID]
+		remote := status.StatsSummary[remoteID]
+		if local.TotalBucketsCount != 2 || remote.TotalBucketsCount != 2 {
+			t.Fatalf("%s: bucket totals = local:%d remote:%d", name, local.TotalBucketsCount, remote.TotalBucketsCount)
+		}
+		if local.TotalTagsCount != 1 || remote.TotalTagsCount != wantRemoteTags ||
+			local.TotalLockConfigCount != 1 || remote.TotalLockConfigCount != 1 ||
+			local.TotalSSEConfigCount != 1 || remote.TotalSSEConfigCount != 1 ||
+			local.TotalVersioningConfigCount != 1 || remote.TotalVersioningConfigCount != 1 ||
+			local.TotalBucketPoliciesCount != 1 || remote.TotalBucketPoliciesCount != 1 ||
+			local.TotalQuotaConfigCount != 1 || remote.TotalQuotaConfigCount != wantRemoteQuota ||
+			local.TotalCorsConfigCount != 1 || remote.TotalCorsConfigCount != 1 {
+			t.Fatalf("%s: site totals = local:%+v remote:%+v", name, local, remote)
+		}
+		if local.ReplicatedTags != 0 || remote.ReplicatedTags != 0 ||
+			local.ReplicatedBucketPolicies != 0 || remote.ReplicatedBucketPolicies != 0 ||
+			local.ReplicatedQuotaConfig != 0 || remote.ReplicatedQuotaConfig != 0 {
+			t.Fatalf("%s: asymmetric configs counted as replicated: local:%+v remote:%+v", name, local, remote)
+		}
+		remoteBucketStatus := status.BucketStats[remoteBucket][remoteID]
+		if remoteBucketStatus.HasTagsSet != (wantRemoteTags != 0) || remoteBucketStatus.HasQuotaCfgSet != (wantRemoteQuota != 0) {
+			t.Fatalf("%s: remote bucket presence = tags:%v quota:%v", name, remoteBucketStatus.HasTagsSet, remoteBucketStatus.HasQuotaCfgSet)
+		}
+	}
+
+	check("valid asymmetric configs", 1, 1)
+	invalidTags := "not-base64"
+	info := remoteInfo.Buckets[remoteBucket]
+	info.Tags = &invalidTags
+	remoteInfo.Buckets[remoteBucket] = info
+	check("malformed tags do not drop site", 0, 1)
+
+	emptyQuota := base64.StdEncoding.EncodeToString([]byte(`{}`))
+	info = remoteInfo.Buckets[remoteBucket]
+	info.QuotaConfig = &emptyQuota
+	remoteInfo.Buckets[remoteBucket] = info
+	check("empty quota is absent", 0, 0)
+}

cmd/site-replication.go

@@ -3453,80 +3453,116 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 			quotaCfgs := make([]*madmin.BucketQuota, numSites)
 			sseCfgSet := set.NewStringSet()
 			versionCfgSet := set.NewStringSet()
-			var tagCount, olockCfgCount, sseCfgCount, corsCfgCount, versionCfgCount int
+			validReplCfg := make([]bool, numSites)
+			validVersionCfg := make([]bool, numSites)
+			validQuotaCfg := make([]bool, numSites)
+			validTags := make([]bool, numSites)
+			validPolicies := make([]bool, numSites)
+			validObjectLockCfg := make([]bool, numSites)
+			validSSECfg := make([]bool, numSites)
+			validCorsCfg := make([]bool, numSites)
+			var tagCount, olockCfgCount, policyCount, quotaCfgCount, sseCfgCount, corsCfgCount, versionCfgCount int
 			for i, s := range slc {
+				logInvalid := func(configType string, err error) {
+					replLogOnceIf(ctx,
+						fmt.Errorf("unable to parse %s metadata for bucket %s from site %s: %w", configType, b, s.DeploymentID, err),
+						"site-replication-status-"+configType+"-"+b+"-"+s.DeploymentID)
+				}
 				if s.ReplicationConfig != nil {
 					cfgBytes, err := base64.StdEncoding.DecodeString(*s.ReplicationConfig)
-					if err != nil {
-						continue
+					if err == nil {
+						cfg, err := sreplication.ParseConfig(bytes.NewReader(cfgBytes))
+						if err == nil {
+							replCfgs[i] = cfg
+							validReplCfg[i] = true
+						} else {
+							logInvalid("replication", err)
+						}
+					} else {
+						logInvalid("replication", err)
 					}
-					cfg, err := sreplication.ParseConfig(bytes.NewReader(cfgBytes))
-					if err != nil {
-						continue
-					}
-					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 err == nil {
+						validVersionCfg[i] = true
+						versionCfgCount++
+						if !versionCfgSet.Contains(string(configData)) {
+							versionCfgSet.Add(string(configData))
+						}
+					} else {
+						logInvalid("versioning", err)
 					}
 				}
 				if s.QuotaConfig != nil {
 					cfgBytes, err := base64.StdEncoding.DecodeString(*s.QuotaConfig)
-					if err != nil {
-						continue
+					if err == nil {
+						cfg, err := parseBucketQuota(b, cfgBytes)
+						if err == nil {
+							if cfg != nil && *cfg != (madmin.BucketQuota{}) {
+								quotaCfgs[i] = cfg
+								validQuotaCfg[i] = true
+								quotaCfgCount++
+							}
+						} else {
+							logInvalid("quota", err)
+						}
+					} else {
+						logInvalid("quota", err)
 					}
-					cfg, err := parseBucketQuota(b, cfgBytes)
-					if err != nil {
-						continue
-					}
-					quotaCfgs[i] = cfg
 				}
 				if s.Tags != nil {
 					tagBytes, err := base64.StdEncoding.DecodeString(*s.Tags)
-					if err != nil {
-						continue
-					}
-					tagCount++
-					if !tagSet.Contains(string(tagBytes)) {
-						tagSet.Add(string(tagBytes))
+					if err == nil {
+						validTags[i] = true
+						tagCount++
+						if !tagSet.Contains(string(tagBytes)) {
+							tagSet.Add(string(tagBytes))
+						}
+					} else {
+						logInvalid("tags", err)
 					}
 				}
 				if len(s.Policy) > 0 {
 					plcy, err := policy.ParseBucketPolicyConfig(bytes.NewReader(s.Policy), b)
-					if err != nil {
-						continue
+					if err == nil {
+						policies[i] = plcy
+						validPolicies[i] = true
+						policyCount++
+					} else {
+						logInvalid("policy", err)
 					}
-					policies[i] = plcy
 				}
 				if s.ObjectLockConfig != nil {
 					configData, err := base64.StdEncoding.DecodeString(*s.ObjectLockConfig)
-					if err != nil {
-						continue
-					}
-					olockCfgCount++
-					if !olockConfigSet.Contains(string(configData)) {
-						olockConfigSet.Add(string(configData))
+					if err == nil {
+						validObjectLockCfg[i] = true
+						olockCfgCount++
+						if !olockConfigSet.Contains(string(configData)) {
+							olockConfigSet.Add(string(configData))
+						}
+					} else {
+						logInvalid("object-lock", err)
 					}
 				}
 				if s.SSEConfig != nil {
 					configData, err := base64.StdEncoding.DecodeString(*s.SSEConfig)
-					if err != nil {
-						continue
-					}
-					sseCfgCount++
-					if !sseCfgSet.Contains(string(configData)) {
-						sseCfgSet.Add(string(configData))
+					if err == nil {
+						validSSECfg[i] = true
+						sseCfgCount++
+						if !sseCfgSet.Contains(string(configData)) {
+							sseCfgSet.Add(string(configData))
+						}
+					} else {
+						logInvalid("sse", err)
 					}
 				}
 				if s.CorsConfig != nil {
 					if _, err := decodeCORSReplicationPayload(s.CorsConfig); err == nil {
+						validCorsCfg[i] = true
 						corsCfgCount++
+					} else {
+						logInvalid("cors", err)
 					}
 				}
 				ss, ok := info.StatsSummary[s.DeploymentID]
@@ -3538,24 +3574,27 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 					ss.ReplicatedBuckets++
 				}
 				ss.TotalBucketsCount++
-				if tagCount > 0 {
+				if validTags[i] {
 					ss.TotalTagsCount++
 				}
-				if olockCfgCount > 0 {
+				if validObjectLockCfg[i] {
 					ss.TotalLockConfigCount++
 				}
-				if sseCfgCount > 0 {
+				if validSSECfg[i] {
 					ss.TotalSSEConfigCount++
 				}
-				if s.CorsConfig != nil {
+				if validCorsCfg[i] {
 					ss.TotalCorsConfigCount++
 				}
-				if versionCfgCount > 0 {
+				if validVersionCfg[i] {
 					ss.TotalVersioningConfigCount++
 				}
-				if len(policies) > 0 {
+				if validPolicies[i] {
 					ss.TotalBucketPoliciesCount++
 				}
+				if validQuotaCfg[i] {
+					ss.TotalQuotaConfigCount++
+				}
 				info.StatsSummary[s.DeploymentID] = ss
 			}
 			tagMismatch := !isReplicated(tagCount, numSites, tagSet)
@@ -3589,13 +3628,13 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 					PolicyMismatch:           policyMismatch,
 					ReplicationCfgMismatch:   replCfgMismatch,
 					QuotaCfgMismatch:         quotaCfgMismatch,
-					HasReplicationCfg:        s.ReplicationConfig != nil,
-					HasTagsSet:               s.Tags != nil,
-					HasOLockConfigSet:        s.ObjectLockConfig != nil,
-					HasPolicySet:             s.Policy != nil,
+					HasReplicationCfg:        validReplCfg[i],
+					HasTagsSet:               validTags[i],
+					HasOLockConfigSet:        validObjectLockCfg[i],
+					HasPolicySet:             validPolicies[i],
 					HasQuotaCfgSet:           quotaCfgSet,
-					HasSSECfgSet:             s.SSEConfig != nil,
-					HasCorsCfgSet:            s.CorsConfig != nil,
+					HasSSECfgSet:             validSSECfg[i],
+					HasCorsCfgSet:            validCorsCfg[i],
 				}
 				var m srBucketMetaInfo
 				if len(bucketStats[s.Bucket]) > dIdx {
@@ -3621,12 +3660,15 @@ func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI O
 				if !corsCfgMismatch && corsCfgCount == numSites {
 					sum.ReplicatedCorsConfig++
 				}
-				if !policyMismatch && len(policies) == numSites {
+				if !policyMismatch && policyCount == numSites {
 					sum.ReplicatedBucketPolicies++
 				}
 				if !tagMismatch && tagCount == numSites {
 					sum.ReplicatedTags++
 				}
+				if !quotaCfgMismatch && quotaCfgCount == numSites {
+					sum.ReplicatedQuotaConfig++
+				}
 				info.StatsSummary[s.DeploymentID] = sum
 			}
 		}