From 5c81d0d89aef63b908e54388e69ff93bfded6ead Mon Sep 17 00:00:00 2001 From: Poorna Date: Thu, 26 May 2022 17:57:23 -0700 Subject: [PATCH] site replication: heal missing/invalid replication config (#14979) Validate remote target ARNs and heal any stale rules in the replication config --- cmd/bucket-handlers.go | 2 +- cmd/bucket-replication.go | 40 ++++++++++++++++++++++++--------------- cmd/site-replication.go | 28 +++++++++++++++++++++++++-- 3 files changed, 52 insertions(+), 18 deletions(-) diff --git a/cmd/bucket-handlers.go b/cmd/bucket-handlers.go index e65772872..ceed0320f 100644 --- a/cmd/bucket-handlers.go +++ b/cmd/bucket-handlers.go @@ -1636,7 +1636,7 @@ func (api objectAPIHandlers) PutBucketReplicationConfigHandler(w http.ResponseWr writeErrorResponse(ctx, w, apiErr, r.URL) return } - sameTarget, apiErr := validateReplicationDestination(ctx, bucket, replicationConfig) + sameTarget, apiErr := validateReplicationDestination(ctx, bucket, replicationConfig, true) if apiErr != noError { writeErrorResponse(ctx, w, apiErr, r.URL) return diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go index 6b668873f..dd89af7fe 100644 --- a/cmd/bucket-replication.go +++ b/cmd/bucket-replication.go @@ -87,7 +87,7 @@ func getReplicationConfig(ctx context.Context, bucketName string) (rc *replicati // validateReplicationDestination returns error if replication destination bucket missing or not configured // It also returns true if replication destination is same as this server. -func validateReplicationDestination(ctx context.Context, bucket string, rCfg *replication.Config) (bool, APIError) { +func validateReplicationDestination(ctx context.Context, bucket string, rCfg *replication.Config, checkRemote bool) (bool, APIError) { var arns []string if rCfg.RoleArn != "" { arns = append(arns, rCfg.RoleArn) @@ -96,26 +96,29 @@ func validateReplicationDestination(ctx context.Context, bucket string, rCfg *re arns = append(arns, rule.Destination.String()) } } + var sameTarget bool for _, arnStr := range arns { arn, err := madmin.ParseARN(arnStr) if err != nil { - return false, errorCodes.ToAPIErrWithErr(ErrBucketRemoteArnInvalid, err) + return sameTarget, errorCodes.ToAPIErrWithErr(ErrBucketRemoteArnInvalid, err) } if arn.Type != madmin.ReplicationService { - return false, toAPIError(ctx, BucketRemoteArnTypeInvalid{Bucket: bucket}) + return sameTarget, toAPIError(ctx, BucketRemoteArnTypeInvalid{Bucket: bucket}) } clnt := globalBucketTargetSys.GetRemoteTargetClient(ctx, arnStr) if clnt == nil { - return false, toAPIError(ctx, BucketRemoteTargetNotFound{Bucket: bucket}) + return sameTarget, toAPIError(ctx, BucketRemoteTargetNotFound{Bucket: bucket}) } - if found, err := clnt.BucketExists(ctx, arn.Bucket); !found { - return false, errorCodes.ToAPIErrWithErr(ErrRemoteDestinationNotFoundError, err) - } - if ret, err := globalBucketObjectLockSys.Get(bucket); err == nil { - if ret.LockEnabled { - lock, _, _, _, err := clnt.GetObjectLockConfig(ctx, arn.Bucket) - if err != nil || lock != "Enabled" { - return false, errorCodes.ToAPIErrWithErr(ErrReplicationDestinationMissingLock, err) + if checkRemote { // validate remote bucket + if found, err := clnt.BucketExists(ctx, arn.Bucket); !found { + return sameTarget, errorCodes.ToAPIErrWithErr(ErrRemoteDestinationNotFoundError, err) + } + if ret, err := globalBucketObjectLockSys.Get(bucket); err == nil { + if ret.LockEnabled { + lock, _, _, _, err := clnt.GetObjectLockConfig(ctx, arn.Bucket) + if err != nil || lock != "Enabled" { + return sameTarget, errorCodes.ToAPIErrWithErr(ErrReplicationDestinationMissingLock, err) + } } } } @@ -123,12 +126,19 @@ func validateReplicationDestination(ctx context.Context, bucket string, rCfg *re c, ok := globalBucketTargetSys.arnRemotesMap[arnStr] if ok { if c.EndpointURL().String() == clnt.EndpointURL().String() { - sameTarget, _ := isLocalHost(clnt.EndpointURL().Hostname(), clnt.EndpointURL().Port(), globalMinioPort) - return sameTarget, toAPIError(ctx, nil) + selfTarget, _ := isLocalHost(clnt.EndpointURL().Hostname(), clnt.EndpointURL().Port(), globalMinioPort) + if !sameTarget { + sameTarget = selfTarget + } + continue } } } - return false, toAPIError(ctx, BucketRemoteTargetNotFound{Bucket: bucket}) + + if len(arns) == 0 { + return false, toAPIError(ctx, BucketRemoteTargetNotFound{Bucket: bucket}) + } + return sameTarget, toAPIError(ctx, nil) } type mustReplicateOptions struct { diff --git a/cmd/site-replication.go b/cmd/site-replication.go index 9ff77a79a..fb2754466 100644 --- a/cmd/site-replication.go +++ b/cmd/site-replication.go @@ -876,14 +876,28 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, ExistingObjectReplicate: "enable", } ) + ruleARN := targetARN for _, r := range replicationConfig.Rules { if r.ID == ruleID { hasRule = true + ruleARN = r.Destination.Bucket } } switch { case hasRule: - err = replicationConfig.EditRule(opts) + if ruleARN != opts.DestBucket { + // remove stale replication rule and replace rule with correct target ARN + if len(replicationConfig.Rules) > 1 { + err = replicationConfig.RemoveRule(opts) + } else { + replicationConfig = replication.Config{} + } + if err == nil { + err = replicationConfig.AddRule(opts) + } + } else { + err = replicationConfig.EditRule(opts) + } default: err = replicationConfig.AddRule(opts) } @@ -902,7 +916,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, if err != nil { return err } - sameTarget, apiErr := validateReplicationDestination(ctx, bucket, newReplicationConfig) + sameTarget, apiErr := validateReplicationDestination(ctx, bucket, newReplicationConfig, true) if apiErr != noError { return fmt.Errorf("bucket replication config validation error: %#v", apiErr) } @@ -3948,6 +3962,16 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob break } } + rcfg, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket) + if err != nil { + replMismatch = true + } + + // validate remote targets on current cluster for this bucket + _, apiErr := validateReplicationDestination(ctx, bucket, rcfg, false) + if apiErr != noError { + replMismatch = true + } if replMismatch { err := c.PeerBucketConfigureReplHandler(ctx, bucket) if err != nil {