mirror of
https://github.com/pgsty/minio.git
synced 2026-11-12 02:25:34 +03:00
fix(site-replication): snapshot state before bucket healing and peer hooks
Snapshot site-replication topology and release the state read lock before bucket healing, metadata healing, IAM deletion healing, and replication hooks perform storage I/O or peer RPCs. Guard remaining peer-state reads and cloning edits so concurrent configuration updates neither deadlock waiting writers nor race on state maps. Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
+5
-6
@@ -676,9 +676,8 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) {
|
|||||||
}()
|
}()
|
||||||
c.iamHealMu.Lock()
|
c.iamHealMu.Lock()
|
||||||
defer c.iamHealMu.Unlock()
|
defer c.iamHealMu.Unlock()
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
snapshot := globalIAMSys.store.revisionIndex().snapshot()
|
snapshot := globalIAMSys.store.revisionIndex().snapshot()
|
||||||
@@ -719,17 +718,17 @@ func (c *SiteReplicationSys) healIAMDeletions(ctx context.Context) (err error) {
|
|||||||
c.iamRevisionProgress = make(map[string]iamRevisionProgress)
|
c.iamRevisionProgress = make(map[string]iamRevisionProgress)
|
||||||
}
|
}
|
||||||
for id := range c.iamRevisionProgress {
|
for id := range c.iamRevisionProgress {
|
||||||
if _, present := c.state.Peers[id]; !present {
|
if _, present := state.Peers[id]; !present {
|
||||||
delete(c.iamRevisionProgress, id)
|
delete(c.iamRevisionProgress, id)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
var progressMu sync.Mutex
|
var progressMu sync.Mutex
|
||||||
cerr := c.state.concDo(nil, func(id string, p madmin.PeerInfo) error {
|
cerr := state.concDo(nil, func(id string, p madmin.PeerInfo) error {
|
||||||
// Bound each pass, but retain acknowledgements independently of the
|
// Bound each pass, but retain acknowledgements independently of the
|
||||||
// pass deadline or unrelated changes at either site.
|
// pass deadline or unrelated changes at either site.
|
||||||
peerCtx, cancel := context.WithTimeout(ctx, defaultContextTimeout)
|
peerCtx, cancel := context.WithTimeout(ctx, defaultContextTimeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
client, err := c.getAdminClient(peerCtx, id)
|
client, err := state.getAdminClient(id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -126,9 +126,8 @@ func latestBucketConfig(bucket, file string, info srStatusInfo) (bucketConfigSta
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file string, info srStatusInfo) error {
|
func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file string, info srStatusInfo) error {
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
latest, found := latestBucketConfig(bucket, file, info)
|
latest, found := latestBucketConfig(bucket, file, info)
|
||||||
@@ -177,7 +176,7 @@ func (c *SiteReplicationSys) healBucketConfig(ctx context.Context, bucket, file
|
|||||||
_, err = globalBucketMetadataSys.updateAndParseMetadata(ctx, bucket, file, latest.data, false, false, &latest.at)
|
_, err = globalBucketMetadataSys.updateAndParseMetadata(ctx, bucket, file, latest.data, false, false, &latest.at)
|
||||||
} else {
|
} else {
|
||||||
var client *madmin.AdminClient
|
var client *madmin.AdminClient
|
||||||
client, err = c.getAdminClient(ctx, id)
|
client, err = state.getAdminClient(id)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
err = client.SRPeerReplicateBucketMeta(ctx, newBucketConfigReplicationEvent(bucket, file, incoming))
|
err = client.SRPeerReplicateBucketMeta(ctx, newBucketConfigReplicationEvent(bucket, file, incoming))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ package cmd
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/base64"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"maps"
|
"maps"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -196,3 +197,323 @@ func TestSiteReplicationMetaInfoCachedPeerState(t *testing.T) {
|
|||||||
t.Fatal("returned cached peer state that was not requested")
|
t.Fatal("returned cached peer state that was not requested")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type siteBucketDeletePausedLayer struct {
|
||||||
|
ObjectLayer
|
||||||
|
once sync.Once
|
||||||
|
deleting, resume chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *siteBucketDeletePausedLayer) DeleteBucket(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
|
||||||
|
l.once.Do(func() { close(l.deleting); <-l.resume })
|
||||||
|
return l.ObjectLayer.DeleteBucket(ctx, bucket, opts)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSiteReplicationHealAndHooksReleaseLock(t *testing.T) {
|
||||||
|
cases := []string{
|
||||||
|
"make-bucket-hook",
|
||||||
|
"delete-bucket-hook",
|
||||||
|
"bucket-meta-hook",
|
||||||
|
"iam-change-hook",
|
||||||
|
"configure-repl",
|
||||||
|
"peer-bucket-delete",
|
||||||
|
"heal-bucket-make",
|
||||||
|
"heal-bucket-delete",
|
||||||
|
"heal-bucket-repl-config",
|
||||||
|
"heal-bucket-config",
|
||||||
|
"heal-cors",
|
||||||
|
"heal-ilm-expiry-config",
|
||||||
|
"heal-bucket-ilm-expiry",
|
||||||
|
}
|
||||||
|
for _, tc := range cases {
|
||||||
|
t.Run(tc, func(t *testing.T) {
|
||||||
|
site := newSvcAccSite(t)
|
||||||
|
c := &globalSiteReplicationSys
|
||||||
|
const bucket = "sr-lock-test-bucket"
|
||||||
|
mustIAM(t, site.obj.MakeBucket(site.ctx, bucket, MakeBucketOptions{VersioningEnabled: true}))
|
||||||
|
_, err := globalBucketMetadataSys.Update(site.ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
|
||||||
|
mustIAM(t, err)
|
||||||
|
|
||||||
|
entered, proceed := make(chan struct{}), make(chan struct{})
|
||||||
|
var once sync.Once
|
||||||
|
pause := func() {
|
||||||
|
once.Do(func() { close(entered) })
|
||||||
|
<-proceed
|
||||||
|
}
|
||||||
|
release := sync.OnceFunc(func() { close(proceed) })
|
||||||
|
defer release()
|
||||||
|
|
||||||
|
peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
pause()
|
||||||
|
if strings.HasSuffix(r.URL.Path, iamRevisionPeerPath) {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_ = json.NewEncoder(w).Encode(iamRevisionResponse{
|
||||||
|
iamRevisionStatus: iamRevisionStatus{Version: iamRevisionProtocol, Node: "node", Instance: "boot", Digest: "fixture"},
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if r.Method == http.MethodGet {
|
||||||
|
if _, ok := r.URL.Query()["versioning"]; ok {
|
||||||
|
w.Header().Set("Content-Type", "application/xml")
|
||||||
|
_, _ = w.Write([]byte(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Enabled</Status></VersioningConfiguration>`))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if _, ok := r.URL.Query()["location"]; ok {
|
||||||
|
w.Header().Set("Content-Type", "application/xml")
|
||||||
|
_, _ = w.Write([]byte(`<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></LocationConstraint>`))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
}))
|
||||||
|
defer peer.Close()
|
||||||
|
|
||||||
|
pausedObj := &siteBucketDeletePausedLayer{
|
||||||
|
ObjectLayer: site.obj,
|
||||||
|
deleting: entered,
|
||||||
|
resume: proceed,
|
||||||
|
}
|
||||||
|
if tc == "peer-bucket-delete" || tc == "heal-bucket-delete" {
|
||||||
|
setObjectLayer(pausedObj)
|
||||||
|
defer setObjectLayer(site.obj)
|
||||||
|
}
|
||||||
|
|
||||||
|
local := globalDeploymentID()
|
||||||
|
now := UTCNow().Truncate(time.Second)
|
||||||
|
peers := site.peers(peer.URL)
|
||||||
|
for id, p := range peers {
|
||||||
|
p.ReplicateILMExpiry = true
|
||||||
|
peers[id] = p
|
||||||
|
}
|
||||||
|
c.Lock()
|
||||||
|
c.enabled = true
|
||||||
|
c.state = srState{
|
||||||
|
Name: "local",
|
||||||
|
ServiceAccountAccessKey: site.parent.AccessKey,
|
||||||
|
Peers: peers,
|
||||||
|
UpdatedAt: now,
|
||||||
|
}
|
||||||
|
c.Unlock()
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
switch tc {
|
||||||
|
case "make-bucket-hook":
|
||||||
|
done <- c.MakeBucketHook(site.ctx, bucket, MakeBucketOptions{CreatedAt: now})
|
||||||
|
case "delete-bucket-hook":
|
||||||
|
done <- c.DeleteBucketHook(site.ctx, bucket, true)
|
||||||
|
case "bucket-meta-hook":
|
||||||
|
done <- c.BucketMetaHook(site.ctx, madmin.SRBucketMeta{
|
||||||
|
Type: madmin.SRBucketMetaTypeTags,
|
||||||
|
Bucket: bucket,
|
||||||
|
UpdatedAt: now,
|
||||||
|
})
|
||||||
|
case "iam-change-hook":
|
||||||
|
site.create(t, "sr-lock-hook-sa")
|
||||||
|
done <- c.IAMChangeHook(site.ctx, madmin.SRIAMItem{
|
||||||
|
Type: madmin.SRIAMItemSvcAcc,
|
||||||
|
SvcAccChange: &madmin.SRSvcAccChange{Create: &madmin.SRSvcAccCreate{AccessKey: "sr-lock-hook-sa"}},
|
||||||
|
UpdatedAt: now,
|
||||||
|
})
|
||||||
|
case "configure-repl":
|
||||||
|
done <- c.PeerBucketConfigureReplHandler(site.ctx, bucket)
|
||||||
|
case "peer-bucket-delete":
|
||||||
|
done <- c.PeerBucketDeleteHandler(site.ctx, bucket, DeleteBucketOptions{Force: true})
|
||||||
|
case "heal-bucket-make":
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
BucketStats: map[string]map[string]srBucketStatsSummary{
|
||||||
|
bucket: {
|
||||||
|
local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now}}},
|
||||||
|
"remote": {HasBucket: false},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healBucket(site.ctx, site.obj, bucket, info)
|
||||||
|
case "heal-bucket-delete":
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
BucketStats: map[string]map[string]srBucketStatsSummary{
|
||||||
|
bucket: {
|
||||||
|
local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), DeletedAt: now}}},
|
||||||
|
"remote": {HasBucket: false, BucketMarkedDeleted: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{DeletedAt: now.Add(-2 * time.Minute)}}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healBucket(site.ctx, pausedObj, bucket, info)
|
||||||
|
case "heal-bucket-repl-config":
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
BucketStats: map[string]map[string]srBucketStatsSummary{
|
||||||
|
bucket: {
|
||||||
|
local: {HasBucket: true, ReplicationCfgMismatch: true},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healBucketReplicationConfig(site.ctx, site.obj, bucket, info, &validateReplicationDestinationOptions{CheckReady: true})
|
||||||
|
case "heal-bucket-config":
|
||||||
|
tagPayload := base64.StdEncoding.EncodeToString([]byte(`<Tagging><TagSet><Tag><Key>k</Key><Value>v</Value></Tag></TagSet></Tagging>`))
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
BucketStats: map[string]map[string]srBucketStatsSummary{
|
||||||
|
bucket: {
|
||||||
|
local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), Tags: &tagPayload, TagConfigUpdatedAt: now}}},
|
||||||
|
"remote": {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute)}}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healBucketConfig(site.ctx, bucket, bucketTaggingConfig, info)
|
||||||
|
case "heal-cors":
|
||||||
|
corsPayload := base64.StdEncoding.EncodeToString([]byte(`<CORSConfiguration><CORSRule><AllowedOrigin>*</AllowedOrigin><AllowedMethod>GET</AllowedMethod></CORSRule></CORSConfiguration>`))
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
BucketStats: map[string]map[string]srBucketStatsSummary{
|
||||||
|
bucket: {
|
||||||
|
local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), CorsConfig: &corsPayload, CorsConfigUpdatedAt: now}}},
|
||||||
|
"remote": {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute)}}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healCORSMetadata(site.ctx, site.obj, bucket, info)
|
||||||
|
case "heal-ilm-expiry-config":
|
||||||
|
stalePeers := site.peers(peer.URL)
|
||||||
|
for id, p := range stalePeers {
|
||||||
|
p.ReplicateILMExpiry = false
|
||||||
|
stalePeers[id] = p
|
||||||
|
}
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
PeerStates: map[string]madmin.SRStateInfo{
|
||||||
|
local: {Name: "local", Peers: peers, UpdatedAt: now},
|
||||||
|
"remote": {Name: "remote", Peers: stalePeers, UpdatedAt: now.Add(-time.Minute)},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healILMExpiryConfig(site.ctx, site.obj, info)
|
||||||
|
case "heal-bucket-ilm-expiry":
|
||||||
|
lcXML := `<LifecycleConfiguration><Rule><ID>exp</ID><Status>Enabled</Status><Expiration><Days>1</Days></Expiration></Rule></LifecycleConfiguration>`
|
||||||
|
lcPayload := base64.StdEncoding.EncodeToString([]byte(lcXML))
|
||||||
|
info := srStatusInfo{
|
||||||
|
Sites: site.peers(peer.URL),
|
||||||
|
BucketStats: map[string]map[string]srBucketStatsSummary{
|
||||||
|
bucket: {
|
||||||
|
local: {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute), ExpiryLCConfig: &lcPayload, ExpiryLCConfigUpdatedAt: now}}},
|
||||||
|
"remote": {HasBucket: true, meta: srBucketMetaInfo{SRBucketInfo: madmin.SRBucketInfo{CreatedAt: now.Add(-time.Minute)}}},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
done <- c.healBucketILMExpiry(site.ctx, site.obj, bucket, info)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-entered:
|
||||||
|
case err := <-done:
|
||||||
|
t.Fatalf("operation returned before reaching storage/peer pause point: %v", err)
|
||||||
|
case <-time.After(10 * time.Second):
|
||||||
|
t.Fatal("timed out waiting for storage/peer pause point")
|
||||||
|
}
|
||||||
|
|
||||||
|
if c.TryLock() {
|
||||||
|
c.state.Name = "mutated-while-in-flight"
|
||||||
|
delete(c.state.Peers, "remote")
|
||||||
|
c.state.UpdatedAt = now.Add(time.Second)
|
||||||
|
c.Unlock()
|
||||||
|
} else {
|
||||||
|
t.Error("SiteReplicationSys lock held across storage I/O or peer RPC")
|
||||||
|
}
|
||||||
|
|
||||||
|
release()
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("unexpected error: %v", err)
|
||||||
|
}
|
||||||
|
case <-time.After(10 * time.Second):
|
||||||
|
t.Fatal("operation deadlocked after writer mutation")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSiteReplicationConcurrentStateUpdates(t *testing.T) {
|
||||||
|
site := newSvcAccSite(t)
|
||||||
|
c := &globalSiteReplicationSys
|
||||||
|
const bucket = "sr-concurrent-state-bucket"
|
||||||
|
mustIAM(t, site.obj.MakeBucket(site.ctx, bucket, MakeBucketOptions{VersioningEnabled: true}))
|
||||||
|
_, err := globalBucketMetadataSys.Update(site.ctx, bucket, bucketVersioningConfig, enabledBucketVersioningConfig)
|
||||||
|
mustIAM(t, err)
|
||||||
|
|
||||||
|
peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
if strings.Contains(r.URL.Path, "metainfo") {
|
||||||
|
w.Header().Set("Content-Type", "application/json")
|
||||||
|
_ = json.NewEncoder(w).Encode(madmin.SRInfo{DeploymentID: "remote"})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if r.Method == http.MethodGet {
|
||||||
|
if _, ok := r.URL.Query()["versioning"]; ok {
|
||||||
|
w.Header().Set("Content-Type", "application/xml")
|
||||||
|
_, _ = w.Write([]byte(`<VersioningConfiguration xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Status>Enabled</Status></VersioningConfiguration>`))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if _, ok := r.URL.Query()["location"]; ok {
|
||||||
|
w.Header().Set("Content-Type", "application/xml")
|
||||||
|
_, _ = w.Write([]byte(`<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></LocationConstraint>`))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
}))
|
||||||
|
defer peer.Close()
|
||||||
|
|
||||||
|
basePeers := site.peers(peer.URL)
|
||||||
|
now := UTCNow().Truncate(time.Second)
|
||||||
|
c.Lock()
|
||||||
|
c.enabled = true
|
||||||
|
c.state = srState{
|
||||||
|
Name: "local",
|
||||||
|
ServiceAccountAccessKey: site.parent.AccessKey,
|
||||||
|
Peers: maps.Clone(basePeers),
|
||||||
|
UpdatedAt: now,
|
||||||
|
}
|
||||||
|
c.Unlock()
|
||||||
|
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(4)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for i := range 10 {
|
||||||
|
_ = c.PeerEditReq(site.ctx, madmin.PeerInfo{
|
||||||
|
DeploymentID: "remote",
|
||||||
|
Name: "remote",
|
||||||
|
Endpoint: peer.URL,
|
||||||
|
ReplicateILMExpiry: i%2 == 0,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for i := range 10 {
|
||||||
|
peers := site.peers(peer.URL)
|
||||||
|
p := peers["remote"]
|
||||||
|
p.ReplicateILMExpiry = i%2 == 1
|
||||||
|
peers["remote"] = p
|
||||||
|
_ = c.PeerStateEditReq(site.ctx, madmin.SRStateEditReq{
|
||||||
|
Peers: peers,
|
||||||
|
UpdatedAt: now.Add(time.Duration(i+1) * time.Second),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for range 10 {
|
||||||
|
_ = c.PeerBucketConfigureReplHandler(site.ctx, bucket)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
for range 5 {
|
||||||
|
_ = c.healBuckets(site.ctx, site.obj)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
|
|||||||
+174
-134
@@ -516,14 +516,12 @@ func (c *SiteReplicationSys) AddPeerClusters(ctx context.Context, psites []madmi
|
|||||||
}
|
}
|
||||||
// check if few peers exist already and ILM expiry replcation is set to true
|
// check if few peers exist already and ILM expiry replcation is set to true
|
||||||
replicateILMExpirySet := false
|
replicateILMExpirySet := false
|
||||||
if c.state.Peers != nil {
|
for _, pi := range c.cloneState().Peers {
|
||||||
for _, pi := range c.state.Peers {
|
|
||||||
if pi.ReplicateILMExpiry {
|
if pi.ReplicateILMExpiry {
|
||||||
replicateILMExpirySet = true
|
replicateILMExpirySet = true
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
for _, v := range sites {
|
for _, v := range sites {
|
||||||
var peerReplicateILMExpiry bool
|
var peerReplicateILMExpiry bool
|
||||||
// if peers already exist and for one of them ReplicateILMExpiry
|
// if peers already exist and for one of them ReplicateILMExpiry
|
||||||
@@ -644,17 +642,16 @@ func (c *SiteReplicationSys) PeerJoinReq(ctx context.Context, arg madmin.SRPeerJ
|
|||||||
return errSRServiceAccount(fmt.Errorf("unable to create service account on %s: %v", ourName, err))
|
return errSRServiceAccount(fmt.Errorf("unable to create service account on %s: %v", ourName, err))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
prevPeers := c.cloneState().Peers
|
||||||
peers := make(map[string]madmin.PeerInfo, len(arg.Peers))
|
peers := make(map[string]madmin.PeerInfo, len(arg.Peers))
|
||||||
for dID, pi := range arg.Peers {
|
for dID, pi := range arg.Peers {
|
||||||
if c.state.Peers != nil {
|
if existingPeer, ok := prevPeers[dID]; ok {
|
||||||
if existingPeer, ok := c.state.Peers[dID]; ok {
|
|
||||||
// retain existing ReplicateILMExpiry of peer if its already set
|
// retain existing ReplicateILMExpiry of peer if its already set
|
||||||
// and incoming arg has it false. it could be default false
|
// and incoming arg has it false. it could be default false
|
||||||
if !pi.ReplicateILMExpiry && existingPeer.ReplicateILMExpiry {
|
if !pi.ReplicateILMExpiry && existingPeer.ReplicateILMExpiry {
|
||||||
pi.ReplicateILMExpiry = existingPeer.ReplicateILMExpiry
|
pi.ReplicateILMExpiry = existingPeer.ReplicateILMExpiry
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
|
||||||
peers[dID] = pi
|
peers[dID] = pi
|
||||||
}
|
}
|
||||||
state := srState{
|
state := srState{
|
||||||
@@ -806,9 +803,8 @@ const (
|
|||||||
func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string, opts MakeBucketOptions) error {
|
func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string, opts MakeBucketOptions) error {
|
||||||
// At this point, the local bucket is created.
|
// At this point, the local bucket is created.
|
||||||
|
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -831,33 +827,33 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string,
|
|||||||
opts.CreatedAt = createdAt
|
opts.CreatedAt = createdAt
|
||||||
|
|
||||||
// Create bucket and enable versioning on all peers.
|
// Create bucket and enable versioning on all peers.
|
||||||
makeBucketConcErr := c.state.concDo(
|
makeBucketConcErr := state.concDo(
|
||||||
func() error {
|
func() error {
|
||||||
return c.annotateErr(makeBucketWithVersion, c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts))
|
return state.annotateErr(makeBucketWithVersion, c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts))
|
||||||
},
|
},
|
||||||
func(deploymentID string, p madmin.PeerInfo) error {
|
func(deploymentID string, p madmin.PeerInfo) error {
|
||||||
admClient, err := c.getAdminClient(ctx, deploymentID)
|
admClient, err := state.getAdminClient(deploymentID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.annotatePeerErr(p.Name, makeBucketWithVersion, admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap))
|
return state.annotatePeerErr(p.Name, makeBucketWithVersion, admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap))
|
||||||
},
|
},
|
||||||
makeBucketWithVersion,
|
makeBucketWithVersion,
|
||||||
)
|
)
|
||||||
|
|
||||||
// Create bucket remotes and add replication rules for the bucket on self and peers.
|
// Create bucket remotes and add replication rules for the bucket on self and peers.
|
||||||
makeRemotesConcErr := c.state.concDo(
|
makeRemotesConcErr := state.concDo(
|
||||||
func() error {
|
func() error {
|
||||||
return c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket))
|
return state.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket))
|
||||||
},
|
},
|
||||||
func(deploymentID string, p madmin.PeerInfo) error {
|
func(deploymentID string, p madmin.PeerInfo) error {
|
||||||
admClient, err := c.getAdminClient(ctx, deploymentID)
|
admClient, err := state.getAdminClient(deploymentID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.annotatePeerErr(p.Name, configureReplication, admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil))
|
return state.annotatePeerErr(p.Name, configureReplication, admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil))
|
||||||
},
|
},
|
||||||
configureReplication,
|
configureReplication,
|
||||||
)
|
)
|
||||||
@@ -874,9 +870,8 @@ func (c *SiteReplicationSys) MakeBucketHook(ctx context.Context, bucket string,
|
|||||||
func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string, forceDelete bool) error {
|
func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string, forceDelete bool) error {
|
||||||
// At this point, the local bucket is deleted.
|
// At this point, the local bucket is deleted.
|
||||||
|
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -886,13 +881,13 @@ func (c *SiteReplicationSys) DeleteBucketHook(ctx context.Context, bucket string
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Send bucket delete to other clusters.
|
// Send bucket delete to other clusters.
|
||||||
cerr := c.state.concDo(nil, func(deploymentID string, p madmin.PeerInfo) error {
|
cerr := state.concDo(nil, func(deploymentID string, p madmin.PeerInfo) error {
|
||||||
admClient, err := c.getAdminClient(ctx, deploymentID)
|
admClient, err := state.getAdminClient(deploymentID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.annotatePeerErr(p.Name, deleteBucket, admClient.SRPeerBucketOps(ctx, bucket, op, nil))
|
return state.annotatePeerErr(p.Name, deleteBucket, admClient.SRPeerBucketOps(ctx, bucket, op, nil))
|
||||||
},
|
},
|
||||||
deleteBucket,
|
deleteBucket,
|
||||||
)
|
)
|
||||||
@@ -982,7 +977,8 @@ func (c *SiteReplicationSys) PeerBucketMakeWithVersioningHandler(ctx context.Con
|
|||||||
// PeerBucketConfigureReplHandler - configures replication remote and
|
// PeerBucketConfigureReplHandler - configures replication remote and
|
||||||
// replication rules to all other peers for the local bucket.
|
// replication rules to all other peers for the local bucket.
|
||||||
func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, bucket string) error {
|
func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context, bucket string) error {
|
||||||
creds, err := c.state.getPeerCreds()
|
state := c.cloneState()
|
||||||
|
creds, err := state.getPeerCreds()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
@@ -1068,7 +1064,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context,
|
|||||||
}
|
}
|
||||||
err := globalBucketTargetSys.SetTarget(ctx, bucket, &targetToUpdate, true)
|
err := globalBucketTargetSys.SetTarget(ctx, bucket, &targetToUpdate, true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotatePeerErr(peer.Name, "Bucket target update error", err)
|
return state.annotatePeerErr(peer.Name, "Bucket target update error", err)
|
||||||
}
|
}
|
||||||
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -1105,7 +1101,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context,
|
|||||||
if !exists { // persist newly generated ARN to targets and metadata on disk
|
if !exists { // persist newly generated ARN to targets and metadata on disk
|
||||||
err := globalBucketTargetSys.SetTarget(ctx, bucket, &bucketTarget, false)
|
err := globalBucketTargetSys.SetTarget(ctx, bucket, &bucketTarget, false)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotatePeerErr(peer.Name, "Bucket target creation error", err)
|
return state.annotatePeerErr(peer.Name, "Bucket target creation error", err)
|
||||||
}
|
}
|
||||||
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -1159,7 +1155,7 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context,
|
|||||||
err = replicationConfig.AddRule(opts)
|
err = replicationConfig.AddRule(opts)
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotatePeerErr(peer.Name, "Error adding bucket replication rule", err)
|
return state.annotatePeerErr(peer.Name, "Error adding bucket replication rule", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Now convert the configuration back to server's type so we can
|
// Now convert the configuration back to server's type so we can
|
||||||
@@ -1187,27 +1183,23 @@ func (c *SiteReplicationSys) PeerBucketConfigureReplHandler(ctx context.Context,
|
|||||||
}
|
}
|
||||||
|
|
||||||
_, err = globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, replCfgData)
|
_, err = globalBucketMetadataSys.Update(ctx, bucket, bucketReplicationConfig, replCfgData)
|
||||||
return c.annotatePeerErr(peer.Name, "Error updating replication configuration", err)
|
return state.annotatePeerErr(peer.Name, "Error updating replication configuration", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
c.RLock()
|
errMap := make(map[string]error, len(state.Peers))
|
||||||
defer c.RUnlock()
|
for d, peer := range state.Peers {
|
||||||
errMap := make(map[string]error, len(c.state.Peers))
|
|
||||||
for d, peer := range c.state.Peers {
|
|
||||||
if d == globalDeploymentID() {
|
if d == globalDeploymentID() {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
errMap[d] = configurePeerFn(d, peer)
|
errMap[d] = configurePeerFn(d, peer)
|
||||||
}
|
}
|
||||||
return c.state.toErrorFromErrMap(errMap, configureReplication)
|
return state.toErrorFromErrMap(errMap, configureReplication)
|
||||||
}
|
}
|
||||||
|
|
||||||
// PeerBucketDeleteHandler - deletes bucket on local in response to a delete
|
// PeerBucketDeleteHandler - deletes bucket on local in response to a delete
|
||||||
// bucket request from a peer.
|
// bucket request from a peer.
|
||||||
func (c *SiteReplicationSys) PeerBucketDeleteHandler(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
|
func (c *SiteReplicationSys) PeerBucketDeleteHandler(ctx context.Context, bucket string, opts DeleteBucketOptions) error {
|
||||||
c.RLock()
|
if !c.isEnabled() {
|
||||||
defer c.RUnlock()
|
|
||||||
if !c.enabled {
|
|
||||||
return errSRNotEnabled
|
return errSRNotEnabled
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1286,19 +1278,18 @@ func (c *SiteReplicationSys) IAMChangeHook(ctx context.Context, item madmin.SRIA
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
cerr := c.state.concDo(nil, func(d string, p madmin.PeerInfo) error {
|
cerr := state.concDo(nil, func(d string, p madmin.PeerInfo) error {
|
||||||
admClient, err := c.getAdminClient(ctx, d)
|
admClient, err := state.getAdminClient(d)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err = executeIAMRevisionRequest(ctx, admClient, http.MethodPut, &iamRevisionBatch{Version: iamRevisionProtocol, Items: []iamReplicationItem{versioned}})
|
_, err = executeIAMRevisionRequest(ctx, admClient, http.MethodPut, &iamRevisionBatch{Version: iamRevisionProtocol, Items: []iamReplicationItem{versioned}})
|
||||||
return c.annotatePeerErr(p.Name, replicateIAMItem, err)
|
return state.annotatePeerErr(p.Name, replicateIAMItem, err)
|
||||||
},
|
},
|
||||||
replicateIAMItem,
|
replicateIAMItem,
|
||||||
)
|
)
|
||||||
@@ -1602,19 +1593,18 @@ func (c *SiteReplicationSys) BucketMetaHook(ctx context.Context, item madmin.SRB
|
|||||||
// The change has already been applied to the local cluster at this
|
// The change has already been applied to the local cluster at this
|
||||||
// point, and only needs to be updated on all remote peer clusters.
|
// point, and only needs to be updated on all remote peer clusters.
|
||||||
|
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
cerr := c.state.concDo(nil, func(d string, p madmin.PeerInfo) error {
|
cerr := state.concDo(nil, func(d string, p madmin.PeerInfo) error {
|
||||||
admClient, err := c.getAdminClient(ctx, d)
|
admClient, err := state.getAdminClient(d)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return c.annotatePeerErr(p.Name, replicateBucketMetadata, admClient.SRPeerReplicateBucketMeta(ctx, item))
|
return state.annotatePeerErr(p.Name, replicateBucketMetadata, admClient.SRPeerReplicateBucketMeta(ctx, item))
|
||||||
},
|
},
|
||||||
replicateBucketMetadata,
|
replicateBucketMetadata,
|
||||||
)
|
)
|
||||||
@@ -2107,15 +2097,40 @@ func (c *SiteReplicationSys) PeerBucketLCConfigHandler(ctx context.Context, buck
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// getAdminClient - NOTE: ensure to take at least a read lock on SiteReplicationSys
|
func (c *SiteReplicationSys) cloneState() srState {
|
||||||
// before calling this.
|
c.RLock()
|
||||||
|
defer c.RUnlock()
|
||||||
|
state := c.state
|
||||||
|
state.Peers = maps.Clone(c.state.Peers)
|
||||||
|
return state
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *SiteReplicationSys) stateSnapshot() (srState, bool) {
|
||||||
|
c.RLock()
|
||||||
|
defer c.RUnlock()
|
||||||
|
if !c.enabled {
|
||||||
|
return srState{}, false
|
||||||
|
}
|
||||||
|
state := c.state
|
||||||
|
state.Peers = maps.Clone(c.state.Peers)
|
||||||
|
return state, true
|
||||||
|
}
|
||||||
|
|
||||||
func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID string) (*madmin.AdminClient, error) {
|
func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID string) (*madmin.AdminClient, error) {
|
||||||
creds, err := c.state.getPeerCreds()
|
return c.cloneState().getAdminClient(deploymentID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *SiteReplicationSys) getAdminClientWithEndpoint(ctx context.Context, deploymentID, endpoint string) (*madmin.AdminClient, error) {
|
||||||
|
return c.cloneState().getAdminClientWithEndpoint(deploymentID, endpoint)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s srState) getAdminClient(deploymentID string) (*madmin.AdminClient, error) {
|
||||||
|
creds, err := s.getPeerCreds()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
peer, ok := c.state.Peers[deploymentID]
|
peer, ok := s.Peers[deploymentID]
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, errSRPeerNotFound
|
return nil, errSRPeerNotFound
|
||||||
}
|
}
|
||||||
@@ -2123,15 +2138,13 @@ func (c *SiteReplicationSys) getAdminClient(ctx context.Context, deploymentID st
|
|||||||
return getAdminClient(peer.Endpoint, creds.AccessKey, creds.SecretKey)
|
return getAdminClient(peer.Endpoint, creds.AccessKey, creds.SecretKey)
|
||||||
}
|
}
|
||||||
|
|
||||||
// getAdminClientWithEndpoint - NOTE: ensure to take at least a read lock on SiteReplicationSys
|
func (s srState) getAdminClientWithEndpoint(deploymentID, endpoint string) (*madmin.AdminClient, error) {
|
||||||
// before calling this.
|
creds, err := s.getPeerCreds()
|
||||||
func (c *SiteReplicationSys) getAdminClientWithEndpoint(ctx context.Context, deploymentID, endpoint string) (*madmin.AdminClient, error) {
|
|
||||||
creds, err := c.state.getPeerCreds()
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if _, ok := c.state.Peers[deploymentID]; !ok {
|
if _, ok := s.Peers[deploymentID]; !ok {
|
||||||
return nil, errSRPeerNotFound
|
return nil, errSRPeerNotFound
|
||||||
}
|
}
|
||||||
return getAdminClient(endpoint, creds.AccessKey, creds.SecretKey)
|
return getAdminClient(endpoint, creds.AccessKey, creds.SecretKey)
|
||||||
@@ -2535,18 +2548,38 @@ func (s srState) concDo(selfActionFn func() error, peerActionFn func(deploymentI
|
|||||||
return s.newConcErr(errMap, actionName)
|
return s.newConcErr(errMap, actionName)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s srState) annotateErr(annotation string, err error) error {
|
||||||
|
if err == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return fmt.Errorf("%s: %s: %w", s.Name, annotation, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s srState) annotatePeerErr(dstPeer string, annotation string, err error) error {
|
||||||
|
if err == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return fmt.Errorf("%s->%s: %s: %w", s.Name, dstPeer, annotation, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *SiteReplicationSys) siteName() string {
|
||||||
|
c.RLock()
|
||||||
|
defer c.RUnlock()
|
||||||
|
return c.state.Name
|
||||||
|
}
|
||||||
|
|
||||||
func (c *SiteReplicationSys) annotateErr(annotation string, err error) error {
|
func (c *SiteReplicationSys) annotateErr(annotation string, err error) error {
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return fmt.Errorf("%s: %s: %w", c.state.Name, annotation, err)
|
return fmt.Errorf("%s: %s: %w", c.siteName(), annotation, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *SiteReplicationSys) annotatePeerErr(dstPeer string, annotation string, err error) error {
|
func (c *SiteReplicationSys) annotatePeerErr(dstPeer string, annotation string, err error) error {
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
return fmt.Errorf("%s->%s: %s: %w", c.state.Name, dstPeer, annotation, err)
|
return fmt.Errorf("%s->%s: %s: %w", c.siteName(), dstPeer, annotation, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// isEnabled returns true if site replication is enabled
|
// isEnabled returns true if site replication is enabled
|
||||||
@@ -2568,12 +2601,14 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
|
|||||||
return st, errSRBackendIssue(err)
|
return st, errSRBackendIssue(err)
|
||||||
}
|
}
|
||||||
peerMap := make(map[string]madmin.PeerInfo)
|
peerMap := make(map[string]madmin.PeerInfo)
|
||||||
|
peerByID := make(map[string]madmin.PeerInfo, len(info.Sites))
|
||||||
var rmvEndpoints []string
|
var rmvEndpoints []string
|
||||||
siteNames := rreq.SiteNames
|
siteNames := rreq.SiteNames
|
||||||
updatedPeers := make(map[string]madmin.PeerInfo)
|
updatedPeers := make(map[string]madmin.PeerInfo)
|
||||||
|
|
||||||
for _, pi := range info.Sites {
|
for _, pi := range info.Sites {
|
||||||
updatedPeers[pi.DeploymentID] = pi
|
updatedPeers[pi.DeploymentID] = pi
|
||||||
|
peerByID[pi.DeploymentID] = pi
|
||||||
peerMap[pi.Name] = pi
|
peerMap[pi.Name] = pi
|
||||||
if rreq.RemoveAll {
|
if rreq.RemoveAll {
|
||||||
siteNames = append(siteNames, pi.Name)
|
siteNames = append(siteNames, pi.Name)
|
||||||
@@ -2588,7 +2623,8 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
|
|||||||
delete(updatedPeers, pinfo.DeploymentID)
|
delete(updatedPeers, pinfo.DeploymentID)
|
||||||
}
|
}
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
errs := make(map[string]error, len(c.state.Peers))
|
var errsMu sync.Mutex
|
||||||
|
errs := make(map[string]error, len(info.Sites))
|
||||||
|
|
||||||
for _, v := range info.Sites {
|
for _, v := range info.Sites {
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
@@ -2596,7 +2632,9 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
|
|||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
err := c.RemoveRemoteTargetsForEndpoint(ctx, objectAPI, rmvEndpoints, false)
|
err := c.RemoveRemoteTargetsForEndpoint(ctx, objectAPI, rmvEndpoints, false)
|
||||||
|
errsMu.Lock()
|
||||||
errs[globalDeploymentID()] = err
|
errs[globalDeploymentID()] = err
|
||||||
|
errsMu.Unlock()
|
||||||
}()
|
}()
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -2604,17 +2642,22 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
|
|||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
admClient, err := c.getAdminClient(ctx, pi.DeploymentID)
|
admClient, err := c.getAdminClient(ctx, pi.DeploymentID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
errsMu.Lock()
|
||||||
errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to create admin client for %s: %w", pi.Name, err))
|
errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to create admin client for %s: %w", pi.Name, err))
|
||||||
|
errsMu.Unlock()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
// set the requesting site's deploymentID for verification of peer request
|
// set the requesting site's deploymentID for verification of peer request
|
||||||
rreq.RequestingDepID = globalDeploymentID()
|
req := rreq
|
||||||
if _, err = admClient.SRPeerRemove(ctx, rreq); err != nil {
|
req.RequestingDepID = globalDeploymentID()
|
||||||
|
if _, err = admClient.SRPeerRemove(ctx, req); err != nil {
|
||||||
if errors.Is(err, errMissingSRConfig) {
|
if errors.Is(err, errMissingSRConfig) {
|
||||||
// ignore if peer is already removed.
|
// ignore if peer is already removed.
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
errsMu.Lock()
|
||||||
errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err))
|
errs[pi.DeploymentID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err))
|
||||||
|
errsMu.Unlock()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}(v)
|
}(v)
|
||||||
@@ -2630,7 +2673,7 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
|
|||||||
return madmin.ReplicateRemoveStatus{
|
return madmin.ReplicateRemoveStatus{
|
||||||
ErrDetail: err.Error(),
|
ErrDetail: err.Error(),
|
||||||
Status: madmin.ReplicateRemoveStatusPartial,
|
Status: madmin.ReplicateRemoveStatusPartial,
|
||||||
}, errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", c.state.Peers[dID].Name, err))
|
}, errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", peerByID[dID].Name, err))
|
||||||
}
|
}
|
||||||
errdID = dID
|
errdID = dID
|
||||||
}
|
}
|
||||||
@@ -2686,13 +2729,14 @@ func (c *SiteReplicationSys) RemovePeerCluster(ctx context.Context, objectAPI Ob
|
|||||||
// InternalRemoveReq - sends an unlink request to peer cluster to remove one or more sites
|
// InternalRemoveReq - sends an unlink request to peer cluster to remove one or more sites
|
||||||
// from the site replication configuration.
|
// from the site replication configuration.
|
||||||
func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI ObjectLayer, rreq madmin.SRRemoveReq) error {
|
func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI ObjectLayer, rreq madmin.SRRemoveReq) error {
|
||||||
if !c.isEnabled() {
|
snapshot, ok := c.stateSnapshot()
|
||||||
|
if !ok {
|
||||||
return errSRNotEnabled
|
return errSRNotEnabled
|
||||||
}
|
}
|
||||||
if rreq.RequestingDepID != "" {
|
if rreq.RequestingDepID != "" {
|
||||||
// validate if requesting site is still part of site replication
|
// validate if requesting site is still part of site replication
|
||||||
var foundRequestor bool
|
var foundRequestor bool
|
||||||
for _, p := range c.state.Peers {
|
for _, p := range snapshot.Peers {
|
||||||
if p.DeploymentID == rreq.RequestingDepID {
|
if p.DeploymentID == rreq.RequestingDepID {
|
||||||
foundRequestor = true
|
foundRequestor = true
|
||||||
break
|
break
|
||||||
@@ -2708,7 +2752,7 @@ func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI Ob
|
|||||||
updatedPeers := make(map[string]madmin.PeerInfo)
|
updatedPeers := make(map[string]madmin.PeerInfo)
|
||||||
siteNames := rreq.SiteNames
|
siteNames := rreq.SiteNames
|
||||||
|
|
||||||
for _, p := range c.state.Peers {
|
for _, p := range snapshot.Peers {
|
||||||
peerMap[p.Name] = p
|
peerMap[p.Name] = p
|
||||||
if p.DeploymentID == globalDeploymentID() {
|
if p.DeploymentID == globalDeploymentID() {
|
||||||
ourName = p.Name
|
ourName = p.Name
|
||||||
@@ -2739,9 +2783,9 @@ func (c *SiteReplicationSys) InternalRemoveReq(ctx context.Context, objectAPI Ob
|
|||||||
var state srState
|
var state srState
|
||||||
if !unlinkSelf {
|
if !unlinkSelf {
|
||||||
state = srState{
|
state = srState{
|
||||||
Name: c.state.Name,
|
Name: snapshot.Name,
|
||||||
Peers: updatedPeers,
|
Peers: updatedPeers,
|
||||||
ServiceAccountAccessKey: c.state.ServiceAccountAccessKey,
|
ServiceAccountAccessKey: snapshot.ServiceAccountAccessKey,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -3004,14 +3048,10 @@ const (
|
|||||||
|
|
||||||
// siteReplicationStatus returns the site replication status across clusters participating in site replication.
|
// siteReplicationStatus returns the site replication status across clusters participating in site replication.
|
||||||
func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI ObjectLayer, opts madmin.SRStatusOptions) (info srStatusInfo, err error) {
|
func (c *SiteReplicationSys) siteReplicationStatus(ctx context.Context, objAPI ObjectLayer, opts madmin.SRStatusOptions) (info srStatusInfo, err error) {
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
if !c.enabled {
|
if !ok {
|
||||||
c.RUnlock()
|
|
||||||
return info, nil
|
return info, nil
|
||||||
}
|
}
|
||||||
state := c.state
|
|
||||||
state.Peers = maps.Clone(state.Peers)
|
|
||||||
c.RUnlock()
|
|
||||||
// Local IAM reads can re-enter the replication lock. Remote work must
|
// Local IAM reads can re-enter the replication lock. Remote work must
|
||||||
// also leave topology changes free to proceed. Use one peer snapshot for
|
// also leave topology changes free to proceed. Use one peer snapshot for
|
||||||
// dispatch, result indexing and error reporting throughout this request.
|
// dispatch, result indexing and error reporting throughout this request.
|
||||||
@@ -4324,10 +4364,7 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe
|
|||||||
return madmin.ReplicateEditStatus{}, errSRInvalidRequest(fmt.Errorf("%s not found in existing replicated sites", peer.DeploymentID))
|
return madmin.ReplicateEditStatus{}, errSRInvalidRequest(fmt.Errorf("%s not found in existing replicated sites", peer.DeploymentID))
|
||||||
}
|
}
|
||||||
successMsg := "Cluster replication configuration updated successfully with:"
|
successMsg := "Cluster replication configuration updated successfully with:"
|
||||||
var state srState
|
state := c.cloneState()
|
||||||
c.RLock()
|
|
||||||
state = c.state
|
|
||||||
c.RUnlock()
|
|
||||||
|
|
||||||
// in case of --disable-ilm-expiry-replication and --enable-ilm-expiry-replication
|
// in case of --disable-ilm-expiry-replication and --enable-ilm-expiry-replication
|
||||||
// --deployment-id is not passed
|
// --deployment-id is not passed
|
||||||
@@ -4335,7 +4372,7 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe
|
|||||||
prevPeerInfo, pi madmin.PeerInfo
|
prevPeerInfo, pi madmin.PeerInfo
|
||||||
)
|
)
|
||||||
if peer.DeploymentID != "" {
|
if peer.DeploymentID != "" {
|
||||||
pi = c.state.Peers[peer.DeploymentID]
|
pi = state.Peers[peer.DeploymentID]
|
||||||
prevPeerInfo = pi
|
prevPeerInfo = pi
|
||||||
if !peer.SyncState.Empty() { // update replication to peer to be sync/async
|
if !peer.SyncState.Empty() { // update replication to peer to be sync/async
|
||||||
pi.SyncState = peer.SyncState
|
pi.SyncState = peer.SyncState
|
||||||
@@ -4387,6 +4424,7 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe
|
|||||||
state.UpdatedAt = time.Now()
|
state.UpdatedAt = time.Now()
|
||||||
|
|
||||||
errs := make(map[string]error, len(state.Peers))
|
errs := make(map[string]error, len(state.Peers))
|
||||||
|
var errsMu sync.Mutex
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
|
|
||||||
for dID, v := range state.Peers {
|
for dID, v := range state.Peers {
|
||||||
@@ -4405,16 +4443,20 @@ func (c *SiteReplicationSys) EditPeerCluster(ctx context.Context, peer madmin.Pe
|
|||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func(pi madmin.PeerInfo, dID string) {
|
go func(pi madmin.PeerInfo, dID string) {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if dID == peer.DeploymentID {
|
if dID == peer.DeploymentID {
|
||||||
admClient, err = c.getAdminClientWithEndpoint(ctx, dID, pi.Endpoint)
|
admClient, err = state.getAdminClientWithEndpoint(dID, pi.Endpoint)
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
errsMu.Lock()
|
||||||
errs[dID] = errSRPeerResp(fmt.Errorf("unable to create admin client for %s: %w", pi.Name, err))
|
errs[dID] = errSRPeerResp(fmt.Errorf("unable to create admin client for %s: %w", pi.Name, err))
|
||||||
|
errsMu.Unlock()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if err = admClient.SRPeerEdit(ctx, pi); err != nil {
|
if err = admClient.SRPeerEdit(ctx, pi); err != nil {
|
||||||
|
errsMu.Lock()
|
||||||
errs[dID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err))
|
errs[dID] = errSRPeerResp(fmt.Errorf("unable to update peer %s: %w", pi.Name, err))
|
||||||
|
errsMu.Unlock()
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}(p, dID)
|
}(p, dID)
|
||||||
@@ -4515,32 +4557,33 @@ func (c *SiteReplicationSys) updateTargetEndpoints(ctx context.Context, prevInfo
|
|||||||
// PeerEditReq - internal API handler to respond to a peer cluster's request
|
// PeerEditReq - internal API handler to respond to a peer cluster's request
|
||||||
// to edit endpoint.
|
// to edit endpoint.
|
||||||
func (c *SiteReplicationSys) PeerEditReq(ctx context.Context, arg madmin.PeerInfo) error {
|
func (c *SiteReplicationSys) PeerEditReq(ctx context.Context, arg madmin.PeerInfo) error {
|
||||||
|
state := c.cloneState()
|
||||||
ourName := ""
|
ourName := ""
|
||||||
|
|
||||||
// Set ReplicateILMExpiry for all peers
|
// Set ReplicateILMExpiry for all peers
|
||||||
currTime := time.Now()
|
currTime := time.Now()
|
||||||
for i := range c.state.Peers {
|
for i := range state.Peers {
|
||||||
p := c.state.Peers[i]
|
p := state.Peers[i]
|
||||||
if p.ReplicateILMExpiry == arg.ReplicateILMExpiry {
|
if p.ReplicateILMExpiry == arg.ReplicateILMExpiry {
|
||||||
// its already set due to previous edit req
|
// its already set due to previous edit req
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
p.ReplicateILMExpiry = arg.ReplicateILMExpiry
|
p.ReplicateILMExpiry = arg.ReplicateILMExpiry
|
||||||
c.state.UpdatedAt = currTime
|
state.UpdatedAt = currTime
|
||||||
c.state.Peers[i] = p
|
state.Peers[i] = p
|
||||||
}
|
}
|
||||||
|
|
||||||
for i := range c.state.Peers {
|
for i := range state.Peers {
|
||||||
p := c.state.Peers[i]
|
p := state.Peers[i]
|
||||||
if p.DeploymentID == arg.DeploymentID {
|
if p.DeploymentID == arg.DeploymentID {
|
||||||
p.Endpoint = arg.Endpoint
|
p.Endpoint = arg.Endpoint
|
||||||
c.state.Peers[arg.DeploymentID] = p
|
state.Peers[arg.DeploymentID] = p
|
||||||
}
|
}
|
||||||
if p.DeploymentID == globalDeploymentID() {
|
if p.DeploymentID == globalDeploymentID() {
|
||||||
ourName = p.Name
|
ourName = p.Name
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if err := c.saveToDisk(ctx, c.state); err != nil {
|
if err := c.saveToDisk(ctx, state); err != nil {
|
||||||
return errSRBackendIssue(fmt.Errorf("unable to save cluster-replication state to drive on %s: %v", ourName, err))
|
return errSRBackendIssue(fmt.Errorf("unable to save cluster-replication state to drive on %s: %v", ourName, err))
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -4549,11 +4592,11 @@ func (c *SiteReplicationSys) PeerEditReq(ctx context.Context, arg madmin.PeerInf
|
|||||||
// PeerStateEditReq - internal API handler to respond to a peer cluster's request
|
// PeerStateEditReq - internal API handler to respond to a peer cluster's request
|
||||||
// to edit state.
|
// to edit state.
|
||||||
func (c *SiteReplicationSys) PeerStateEditReq(ctx context.Context, arg madmin.SRStateEditReq) error {
|
func (c *SiteReplicationSys) PeerStateEditReq(ctx context.Context, arg madmin.SRStateEditReq) error {
|
||||||
if arg.UpdatedAt.After(c.state.UpdatedAt) {
|
state := c.cloneState()
|
||||||
state := c.state
|
if arg.UpdatedAt.After(state.UpdatedAt) {
|
||||||
// update only the ReplicateILMExpiry flag for the peers from incoming request
|
// update only the ReplicateILMExpiry flag for the peers from incoming request
|
||||||
for _, peer := range arg.Peers {
|
for _, peer := range arg.Peers {
|
||||||
currPeer := c.state.Peers[peer.DeploymentID]
|
currPeer := state.Peers[peer.DeploymentID]
|
||||||
currPeer.ReplicateILMExpiry = peer.ReplicateILMExpiry
|
currPeer.ReplicateILMExpiry = peer.ReplicateILMExpiry
|
||||||
state.Peers[peer.DeploymentID] = currPeer
|
state.Peers[peer.DeploymentID] = currPeer
|
||||||
}
|
}
|
||||||
@@ -4705,9 +4748,8 @@ func getSRBucketDeleteOp(isSiteReplicated bool) SRBucketDeleteOp {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *SiteReplicationSys) healILMExpiryConfig(ctx context.Context, objAPI ObjectLayer, info srStatusInfo) error {
|
func (c *SiteReplicationSys) healILMExpiryConfig(ctx context.Context, objAPI ObjectLayer, info srStatusInfo) error {
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
var (
|
var (
|
||||||
@@ -4749,12 +4791,12 @@ func (c *SiteReplicationSys) healILMExpiryConfig(ctx context.Context, objAPI Obj
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Send details to other sites for healing
|
// Send details to other sites for healing
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
if err = admClient.SRStateEdit(ctx, madmin.SRStateEditReq{Peers: latestPeers, UpdatedAt: lastUpdate}); err != nil {
|
if err = admClient.SRStateEdit(ctx, madmin.SRStateEditReq{Peers: latestPeers, UpdatedAt: lastUpdate}); err != nil {
|
||||||
replLogIf(ctx, c.annotatePeerErr(ps.Name, siteReplicationEdit,
|
replLogIf(ctx, state.annotatePeerErr(ps.Name, siteReplicationEdit,
|
||||||
fmt.Errorf("Unable to heal site replication state for peer %s from peer %s : %w",
|
fmt.Errorf("Unable to heal site replication state for peer %s from peer %s : %w",
|
||||||
ps.Name, latestPeerName, err)))
|
ps.Name, latestPeerName, err)))
|
||||||
}
|
}
|
||||||
@@ -4797,7 +4839,7 @@ func (c *SiteReplicationSys) healBuckets(ctx context.Context, objAPI ObjectLayer
|
|||||||
c.healILMExpiryConfig(ctx, objAPI, info)
|
c.healILMExpiryConfig(ctx, objAPI, info)
|
||||||
ilmExpiryCfgHealed = true
|
ilmExpiryCfgHealed = true
|
||||||
}
|
}
|
||||||
if ilmExpiryReplicationEnabled(c.state.Peers) {
|
if state, ok := c.stateSnapshot(); ok && ilmExpiryReplicationEnabled(state.Peers) {
|
||||||
c.healBucketILMExpiry(ctx, objAPI, bucket, info)
|
c.healBucketILMExpiry(ctx, objAPI, bucket, info)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -4809,9 +4851,8 @@ func (c *SiteReplicationSys) healBuckets(ctx context.Context, objAPI ObjectLayer
|
|||||||
func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||||
bs := info.BucketStats[bucket]
|
bs := info.BucketStats[bucket]
|
||||||
|
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4863,7 +4904,7 @@ func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI Obj
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
@@ -4874,7 +4915,7 @@ func (c *SiteReplicationSys) healBucketILMExpiry(ctx context.Context, objAPI Obj
|
|||||||
ExpiryLCConfig: latestExpLCConfig,
|
ExpiryLCConfig: latestExpLCConfig,
|
||||||
UpdatedAt: lastUpdate,
|
UpdatedAt: lastUpdate,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
replLogIf(ctx, state.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||||
fmt.Errorf("Unable to heal bucket ILM expiry data for peer %s from peer %s : %w",
|
fmt.Errorf("Unable to heal bucket ILM expiry data for peer %s from peer %s : %w",
|
||||||
peerName, latestPeerName, err)))
|
peerName, latestPeerName, err)))
|
||||||
}
|
}
|
||||||
@@ -4919,9 +4960,8 @@ func latestCORSConfig(bs map[string]srBucketStatsSummary) (latestID string, late
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4946,7 +4986,7 @@ func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI Object
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return wrapSRErr(err)
|
return wrapSRErr(err)
|
||||||
}
|
}
|
||||||
@@ -4958,7 +4998,7 @@ func (c *SiteReplicationSys) healCORSMetadata(ctx context.Context, objAPI Object
|
|||||||
UpdatedAt: latestState.updatedAt,
|
UpdatedAt: latestState.updatedAt,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
replLogIf(ctx, c.annotatePeerErr(peerName, replicateBucketMetadata,
|
replLogIf(ctx, state.annotatePeerErr(peerName, replicateBucketMetadata,
|
||||||
fmt.Errorf("Unable to heal CORS config metadata for peer %s from peer %s : %w",
|
fmt.Errorf("Unable to heal CORS config metadata for peer %s from peer %s : %w",
|
||||||
peerName, latestPeerName, err)))
|
peerName, latestPeerName, err)))
|
||||||
}
|
}
|
||||||
@@ -4981,12 +5021,11 @@ func (c *SiteReplicationSys) purgeDeletedBucket(ctx context.Context, objAPI Obje
|
|||||||
// healBucket creates/deletes the bucket according to latest state across clusters participating in site replication.
|
// healBucket creates/deletes the bucket according to latest state across clusters participating in site replication.
|
||||||
func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo) error {
|
||||||
bs := info.BucketStats[bucket]
|
bs := info.BucketStats[bucket]
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
numSites := len(c.state.Peers)
|
numSites := len(state.Peers)
|
||||||
mostRecent := func(d1, d2 time.Time) time.Time {
|
mostRecent := func(d1, d2 time.Time) time.Time {
|
||||||
if d1.IsZero() {
|
if d1.IsZero() {
|
||||||
return d2
|
return d2
|
||||||
@@ -5068,19 +5107,19 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer,
|
|||||||
if dID == globalDeploymentID() {
|
if dID == globalDeploymentID() {
|
||||||
err := c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts)
|
err := c.PeerBucketMakeWithVersioningHandler(ctx, bucket, opts)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotateErr(makeBucketWithVersion, fmt.Errorf("error healing bucket for site replication %w from %s -> %s",
|
return state.annotateErr(makeBucketWithVersion, fmt.Errorf("error healing bucket for site replication %w from %s -> %s",
|
||||||
err, latestPeerName, peerName))
|
err, latestPeerName, peerName))
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err))
|
return state.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err))
|
||||||
}
|
}
|
||||||
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap); err != nil {
|
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.MakeWithVersioningBktOp, optsMap); err != nil {
|
||||||
return c.annotatePeerErr(peerName, makeBucketWithVersion, err)
|
return state.annotatePeerErr(peerName, makeBucketWithVersion, err)
|
||||||
}
|
}
|
||||||
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil); err != nil {
|
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.ConfigureReplBktOp, nil); err != nil {
|
||||||
return c.annotatePeerErr(peerName, configureReplication, err)
|
return state.annotatePeerErr(peerName, configureReplication, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -5088,7 +5127,7 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer,
|
|||||||
// configure replication from current cluster to other clusters
|
// configure replication from current cluster to other clusters
|
||||||
err := c.PeerBucketConfigureReplHandler(ctx, bucket)
|
err := c.PeerBucketConfigureReplHandler(ctx, bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotateErr(configureReplication, err)
|
return state.annotateErr(configureReplication, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -5101,12 +5140,12 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer,
|
|||||||
if dID == globalDeploymentID() {
|
if dID == globalDeploymentID() {
|
||||||
c.purgeDeletedBucket(ctx, objAPI, bucket)
|
c.purgeDeletedBucket(ctx, objAPI, bucket)
|
||||||
} else {
|
} else {
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err))
|
return state.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err))
|
||||||
}
|
}
|
||||||
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.PurgeDeletedBucketOp, nil); err != nil {
|
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.PurgeDeletedBucketOp, nil); err != nil {
|
||||||
return c.annotatePeerErr(peerName, deleteBucket, err)
|
return state.annotatePeerErr(peerName, deleteBucket, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -5120,16 +5159,16 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer,
|
|||||||
Force: true,
|
Force: true,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotateErr(deleteBucket, fmt.Errorf("error healing bucket for site replication %w from %s -> %s",
|
return state.annotateErr(deleteBucket, fmt.Errorf("error healing bucket for site replication %w from %s -> %s",
|
||||||
err, latestPeerName, peerName))
|
err, latestPeerName, peerName))
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
admClient, err := c.getAdminClient(ctx, dID)
|
admClient, err := state.getAdminClient(dID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return c.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err))
|
return state.annotateErr(configureReplication, fmt.Errorf("unable to use admin client for %s: %w", dID, err))
|
||||||
}
|
}
|
||||||
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.ForceDeleteBucketBktOp, nil); err != nil {
|
if err = admClient.SRPeerBucketOps(ctx, bucket, madmin.ForceDeleteBucketBktOp, nil); err != nil {
|
||||||
return c.annotatePeerErr(peerName, deleteBucket, err)
|
return state.annotatePeerErr(peerName, deleteBucket, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -5141,9 +5180,8 @@ func (c *SiteReplicationSys) healBucket(ctx context.Context, objAPI ObjectLayer,
|
|||||||
func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo, opts *validateReplicationDestinationOptions) error {
|
func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, objAPI ObjectLayer, bucket string, info srStatusInfo, opts *validateReplicationDestinationOptions) error {
|
||||||
bs := info.BucketStats[bucket]
|
bs := info.BucketStats[bucket]
|
||||||
|
|
||||||
c.RLock()
|
state, ok := c.stateSnapshot()
|
||||||
defer c.RUnlock()
|
if !ok {
|
||||||
if !c.enabled {
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -5172,7 +5210,7 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob
|
|||||||
arnTgtMap[t.Arn] = t
|
arnTgtMap[t.Arn] = t
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
for _, p := range c.state.Peers {
|
for _, p := range state.Peers {
|
||||||
epDeplIDMap[p.Endpoint] = p.DeploymentID
|
epDeplIDMap[p.Endpoint] = p.DeploymentID
|
||||||
}
|
}
|
||||||
// fix stale ARN's in replication config and endpoint mismatch between site config and
|
// fix stale ARN's in replication config and endpoint mismatch between site config and
|
||||||
@@ -5199,7 +5237,7 @@ func (c *SiteReplicationSys) healBucketReplicationConfig(ctx context.Context, ob
|
|||||||
}
|
}
|
||||||
|
|
||||||
if replMismatch {
|
if replMismatch {
|
||||||
replLogOnceIf(ctx, c.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)), "heal-bucket-relication-config")
|
replLogOnceIf(ctx, state.annotateErr(configureReplication, c.PeerBucketConfigureReplHandler(ctx, bucket)), "heal-bucket-relication-config")
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -5678,7 +5716,8 @@ func (c *SiteReplicationSys) getPeerForUpload(deplID string) (pi srPeerInfo, loc
|
|||||||
func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) {
|
func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) {
|
||||||
c.resyncMu.Lock()
|
c.resyncMu.Lock()
|
||||||
defer c.resyncMu.Unlock()
|
defer c.resyncMu.Unlock()
|
||||||
if !c.isEnabled() {
|
state, ok := c.stateSnapshot()
|
||||||
|
if !ok {
|
||||||
return res, errSRNotEnabled
|
return res, errSRNotEnabled
|
||||||
}
|
}
|
||||||
if objAPI == nil {
|
if objAPI == nil {
|
||||||
@@ -5688,7 +5727,7 @@ func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer
|
|||||||
if peer.DeploymentID == globalDeploymentID() {
|
if peer.DeploymentID == globalDeploymentID() {
|
||||||
return res, errSRResyncToSelf
|
return res, errSRResyncToSelf
|
||||||
}
|
}
|
||||||
if _, ok := c.state.Peers[peer.DeploymentID]; !ok {
|
if _, ok := state.Peers[peer.DeploymentID]; !ok {
|
||||||
return res, errSRPeerNotFound
|
return res, errSRPeerNotFound
|
||||||
}
|
}
|
||||||
rs, err := globalSiteResyncMetrics.siteStatus(ctx, objAPI, peer.DeploymentID)
|
rs, err := globalSiteResyncMetrics.siteStatus(ctx, objAPI, peer.DeploymentID)
|
||||||
@@ -5799,7 +5838,8 @@ func (c *SiteReplicationSys) startResync(ctx context.Context, objAPI ObjectLayer
|
|||||||
func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) {
|
func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLayer, peer madmin.PeerInfo) (res madmin.SRResyncOpStatus, err error) {
|
||||||
c.resyncMu.Lock()
|
c.resyncMu.Lock()
|
||||||
defer c.resyncMu.Unlock()
|
defer c.resyncMu.Unlock()
|
||||||
if !c.isEnabled() {
|
state, ok := c.stateSnapshot()
|
||||||
|
if !ok {
|
||||||
return res, errSRNotEnabled
|
return res, errSRNotEnabled
|
||||||
}
|
}
|
||||||
if objAPI == nil {
|
if objAPI == nil {
|
||||||
@@ -5808,7 +5848,7 @@ func (c *SiteReplicationSys) cancelResync(ctx context.Context, objAPI ObjectLaye
|
|||||||
if peer.DeploymentID == globalDeploymentID() {
|
if peer.DeploymentID == globalDeploymentID() {
|
||||||
return res, errSRResyncToSelf
|
return res, errSRResyncToSelf
|
||||||
}
|
}
|
||||||
if _, ok := c.state.Peers[peer.DeploymentID]; !ok {
|
if _, ok := state.Peers[peer.DeploymentID]; !ok {
|
||||||
return res, errSRPeerNotFound
|
return res, errSRPeerNotFound
|
||||||
}
|
}
|
||||||
rs, err := globalSiteResyncMetrics.siteStatus(ctx, objAPI, peer.DeploymentID)
|
rs, err := globalSiteResyncMetrics.siteStatus(ctx, objAPI, peer.DeploymentID)
|
||||||
|
|||||||
Reference in New Issue
Block a user