mirror of
https://github.com/pgsty/minio.git
synced 2026-09-20 01:28:24 +03:00
fix(replication): converge bucket metadata using deterministic source states
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
Feng Ruohang
+6
-415
@@ -4883,366 +4883,23 @@ func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI Obj
|
||||
}
|
||||
|
||||
func (c *SiteReplicationSys) healTagMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||
bs := info.BucketStats[bucket]
|
||||
|
||||
c.RLock()
|
||||
defer c.RUnlock()
|
||||
if !c.enabled {
|
||||
return nil
|
||||
}
|
||||
var (
|
||||
latestID, latestPeerName string
|
||||
lastUpdate time.Time
|
||||
latestTaggingConfig *string
|
||||
)
|
||||
|
||||
for dID, ss := range bs {
|
||||
if lastUpdate.IsZero() {
|
||||
lastUpdate = ss.meta.TagConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestTaggingConfig = ss.meta.Tags
|
||||
}
|
||||
// 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.TagConfigUpdatedAt) {
|
||||
continue
|
||||
}
|
||||
if ss.meta.TagConfigUpdatedAt.After(lastUpdate) {
|
||||
lastUpdate = ss.meta.TagConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestTaggingConfig = ss.meta.Tags
|
||||
}
|
||||
}
|
||||
latestPeerName = info.Sites[latestID].Name
|
||||
var latestTaggingConfigBytes []byte
|
||||
var err error
|
||||
if latestTaggingConfig != nil {
|
||||
latestTaggingConfigBytes, err = base64.StdEncoding.DecodeString(*latestTaggingConfig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
for dID, bStatus := range bs {
|
||||
if !bStatus.TagMismatch {
|
||||
continue
|
||||
}
|
||||
if isBucketMetadataEqual(latestTaggingConfig, bStatus.meta.Tags) {
|
||||
continue
|
||||
}
|
||||
if dID == globalDeploymentID() {
|
||||
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketTaggingConfig, latestTaggingConfigBytes); err != nil {
|
||||
replLogIf(ctx, fmt.Errorf("Unable to heal tagging 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.SRBucketMetaTypeTags,
|
||||
Bucket: bucket,
|
||||
Tags: latestTaggingConfig,
|
||||
})
|
||||
if err != nil {
|
||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||
fmt.Errorf("Unable to heal tagging metadata for peer %s from peer %s : %w", peerName, latestPeerName, err)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return c.healBucketConfig(ctx, bucket, bucketTaggingConfig, info)
|
||||
}
|
||||
|
||||
func (c *SiteReplicationSys) healBucketPolicies(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||
bs := info.BucketStats[bucket]
|
||||
|
||||
c.RLock()
|
||||
defer c.RUnlock()
|
||||
if !c.enabled {
|
||||
return nil
|
||||
}
|
||||
var (
|
||||
latestID, latestPeerName string
|
||||
lastUpdate time.Time
|
||||
latestIAMPolicy json.RawMessage
|
||||
)
|
||||
|
||||
for dID, ss := range bs {
|
||||
if lastUpdate.IsZero() {
|
||||
lastUpdate = ss.meta.PolicyUpdatedAt
|
||||
latestID = dID
|
||||
latestIAMPolicy = ss.meta.Policy
|
||||
}
|
||||
// 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.PolicyUpdatedAt) {
|
||||
continue
|
||||
}
|
||||
if ss.meta.PolicyUpdatedAt.After(lastUpdate) {
|
||||
lastUpdate = ss.meta.PolicyUpdatedAt
|
||||
latestID = dID
|
||||
latestIAMPolicy = ss.meta.Policy
|
||||
}
|
||||
}
|
||||
latestPeerName = info.Sites[latestID].Name
|
||||
for dID, bStatus := range bs {
|
||||
if !bStatus.PolicyMismatch {
|
||||
continue
|
||||
}
|
||||
if strings.EqualFold(string(latestIAMPolicy), string(bStatus.meta.Policy)) {
|
||||
continue
|
||||
}
|
||||
if dID == globalDeploymentID() {
|
||||
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketPolicyConfig, latestIAMPolicy); err != nil {
|
||||
replLogIf(ctx, fmt.Errorf("Unable to heal bucket policy 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
|
||||
if err = admClient.SRPeerReplicateBucketMeta(ctx, madmin.SRBucketMeta{
|
||||
Type: madmin.SRBucketMetaTypePolicy,
|
||||
Bucket: bucket,
|
||||
Policy: latestIAMPolicy,
|
||||
UpdatedAt: lastUpdate,
|
||||
}); err != nil {
|
||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||
fmt.Errorf("Unable to heal bucket policy metadata for peer %s from peer %s : %w",
|
||||
peerName, latestPeerName, err)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return c.healBucketConfig(ctx, bucket, bucketPolicyConfig, info)
|
||||
}
|
||||
|
||||
func (c *SiteReplicationSys) healBucketQuotaConfig(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||
bs := info.BucketStats[bucket]
|
||||
|
||||
c.RLock()
|
||||
defer c.RUnlock()
|
||||
if !c.enabled {
|
||||
return nil
|
||||
}
|
||||
var (
|
||||
latestID, latestPeerName string
|
||||
lastUpdate time.Time
|
||||
latestQuotaConfig *string
|
||||
latestQuotaConfigBytes []byte
|
||||
)
|
||||
|
||||
for dID, ss := range bs {
|
||||
if lastUpdate.IsZero() {
|
||||
lastUpdate = ss.meta.QuotaConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestQuotaConfig = ss.meta.QuotaConfig
|
||||
}
|
||||
// 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.QuotaConfigUpdatedAt) {
|
||||
continue
|
||||
}
|
||||
if ss.meta.QuotaConfigUpdatedAt.After(lastUpdate) {
|
||||
lastUpdate = ss.meta.QuotaConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestQuotaConfig = ss.meta.QuotaConfig
|
||||
}
|
||||
}
|
||||
|
||||
var err error
|
||||
if latestQuotaConfig != nil {
|
||||
latestQuotaConfigBytes, err = base64.StdEncoding.DecodeString(*latestQuotaConfig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
latestPeerName = info.Sites[latestID].Name
|
||||
for dID, bStatus := range bs {
|
||||
if !bStatus.QuotaCfgMismatch {
|
||||
continue
|
||||
}
|
||||
if isBucketMetadataEqual(latestQuotaConfig, bStatus.meta.QuotaConfig) {
|
||||
continue
|
||||
}
|
||||
if dID == globalDeploymentID() {
|
||||
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketQuotaConfigFile, latestQuotaConfigBytes); err != nil {
|
||||
replLogIf(ctx, fmt.Errorf("Unable to heal quota 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
|
||||
|
||||
if err = admClient.SRPeerReplicateBucketMeta(ctx, madmin.SRBucketMeta{
|
||||
Type: madmin.SRBucketMetaTypeQuotaConfig,
|
||||
Bucket: bucket,
|
||||
Quota: latestQuotaConfigBytes,
|
||||
UpdatedAt: lastUpdate,
|
||||
}); err != nil {
|
||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||
fmt.Errorf("Unable to heal quota config metadata for peer %s from peer %s : %w",
|
||||
peerName, latestPeerName, err)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return c.healBucketConfig(ctx, bucket, bucketQuotaConfigFile, info)
|
||||
}
|
||||
|
||||
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, bucketVersioningConfig, latestVersioningConfigBytes); err != nil {
|
||||
replLogIf(ctx, fmt.Errorf("Unable to heal versioning 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,
|
||||
UpdatedAt: lastUpdate,
|
||||
})
|
||||
if err != nil {
|
||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||
fmt.Errorf("Unable to heal versioning config metadata for peer %s from peer %s : %w",
|
||||
peerName, latestPeerName, err)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return c.healBucketConfig(ctx, bucket, bucketVersioningConfig, info)
|
||||
}
|
||||
|
||||
func (c *SiteReplicationSys) healSSEMetadata(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
|
||||
latestSSEConfig *string
|
||||
)
|
||||
|
||||
bs := info.BucketStats[bucket]
|
||||
for dID, ss := range bs {
|
||||
if lastUpdate.IsZero() {
|
||||
lastUpdate = ss.meta.SSEConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestSSEConfig = ss.meta.SSEConfig
|
||||
}
|
||||
// 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.SSEConfigUpdatedAt) {
|
||||
continue
|
||||
}
|
||||
if ss.meta.SSEConfigUpdatedAt.After(lastUpdate) {
|
||||
lastUpdate = ss.meta.SSEConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestSSEConfig = ss.meta.SSEConfig
|
||||
}
|
||||
}
|
||||
|
||||
latestPeerName = info.Sites[latestID].Name
|
||||
var latestSSEConfigBytes []byte
|
||||
var err error
|
||||
if latestSSEConfig != nil {
|
||||
latestSSEConfigBytes, err = base64.StdEncoding.DecodeString(*latestSSEConfig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for dID, bStatus := range bs {
|
||||
if !bStatus.SSEConfigMismatch {
|
||||
continue
|
||||
}
|
||||
if isBucketMetadataEqual(latestSSEConfig, bStatus.meta.SSEConfig) {
|
||||
continue
|
||||
}
|
||||
if dID == globalDeploymentID() {
|
||||
if _, err := globalBucketMetadataSys.Update(ctx, bucket, bucketSSEConfig, latestSSEConfigBytes); err != nil {
|
||||
replLogIf(ctx, fmt.Errorf("Unable to heal 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.SRBucketMetaTypeSSEConfig,
|
||||
Bucket: bucket,
|
||||
SSEConfig: latestSSEConfig,
|
||||
UpdatedAt: lastUpdate,
|
||||
})
|
||||
if err != nil {
|
||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||
fmt.Errorf("Unable to heal SSE config metadata for peer %s from peer %s : %w",
|
||||
peerName, latestPeerName, err)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return c.healBucketConfig(ctx, bucket, bucketSSEConfig, info)
|
||||
}
|
||||
|
||||
func latestCORSConfig(bs map[string]srBucketStatsSummary) (latestID string, latest corsReplicationState, ok bool) {
|
||||
@@ -5310,73 +4967,7 @@ func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI Object
|
||||
}
|
||||
|
||||
func (c *SiteReplicationSys) healOLockConfigMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||
bs := info.BucketStats[bucket]
|
||||
|
||||
c.RLock()
|
||||
defer c.RUnlock()
|
||||
if !c.enabled {
|
||||
return nil
|
||||
}
|
||||
var (
|
||||
latestID, latestPeerName string
|
||||
lastUpdate time.Time
|
||||
latestObjLockConfig *string
|
||||
)
|
||||
|
||||
for dID, ss := range bs {
|
||||
if lastUpdate.IsZero() {
|
||||
lastUpdate = ss.meta.ObjectLockConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestObjLockConfig = ss.meta.ObjectLockConfig
|
||||
}
|
||||
// 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.ObjectLockConfigUpdatedAt) {
|
||||
continue
|
||||
}
|
||||
if ss.meta.ObjectLockConfig != nil && ss.meta.ObjectLockConfigUpdatedAt.After(lastUpdate) {
|
||||
lastUpdate = ss.meta.ObjectLockConfigUpdatedAt
|
||||
latestID = dID
|
||||
latestObjLockConfig = ss.meta.ObjectLockConfig
|
||||
}
|
||||
}
|
||||
latestPeerName = info.Sites[latestID].Name
|
||||
var latestObjLockConfigBytes []byte
|
||||
var err error
|
||||
if latestObjLockConfig != nil {
|
||||
latestObjLockConfigBytes, err = base64.StdEncoding.DecodeString(*latestObjLockConfig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for dID, bStatus := range bs {
|
||||
if !bStatus.OLockConfigMismatch {
|
||||
continue
|
||||
}
|
||||
if isBucketMetadataEqual(latestObjLockConfig, bStatus.meta.ObjectLockConfig) {
|
||||
continue
|
||||
}
|
||||
if dID == globalDeploymentID() {
|
||||
if _, err := globalBucketMetadataSys.Update(ctx, bucket, objectLockConfig, latestObjLockConfigBytes); err != nil {
|
||||
replLogIf(ctx, fmt.Errorf("Unable to heal objectlock config 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, newSRBucketObjectLockMeta(bucket, latestObjLockConfig, lastUpdate))
|
||||
if err != nil {
|
||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||
fmt.Errorf("Unable to heal object lock config metadata for peer %s from peer %s : %w",
|
||||
peerName, latestPeerName, err)))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return c.healBucketConfig(ctx, bucket, objectLockConfig, info)
|
||||
}
|
||||
|
||||
func (c *SiteReplicationSys) purgeDeletedBucket(ctx context.Context, objAPI ObjectLayer, bucket string) {
|
||||
|
||||
Reference in New Issue
Block a user