mirror of
https://github.com/minio/minio.git
synced 2024-12-24 22:25:54 -05:00
fix: refactor getReplicationDiff for safe use (#16051)
This commit is contained in:
parent
3bb82ef60d
commit
8a07000e58
@ -2604,31 +2604,38 @@ func saveResyncStatus(ctx context.Context, bucket string, brs BucketReplicationR
|
|||||||
return saveConfig(ctx, objectAPI, configFile, buf)
|
return saveConfig(ctx, objectAPI, configFile, buf)
|
||||||
}
|
}
|
||||||
|
|
||||||
// getReplicationDiff returns unreplicated objects in a channel
|
// getReplicationDiff returns un-replicated objects in a channel.
|
||||||
func getReplicationDiff(ctx context.Context, objAPI ObjectLayer, bucket string, opts madmin.ReplDiffOpts) (diffCh chan madmin.DiffInfo, err error) {
|
// If a non-nil channel is returned it must be consumed fully or
|
||||||
objInfoCh := make(chan ObjectInfo)
|
// the provided context must be canceled.
|
||||||
if err := objAPI.Walk(ctx, bucket, opts.Prefix, objInfoCh, ObjectOptions{}); err != nil {
|
func getReplicationDiff(ctx context.Context, objAPI ObjectLayer, bucket string, opts madmin.ReplDiffOpts) (chan madmin.DiffInfo, error) {
|
||||||
logger.LogIf(ctx, err)
|
|
||||||
return diffCh, err
|
|
||||||
}
|
|
||||||
cfg, err := getReplicationConfig(ctx, bucket)
|
cfg, err := getReplicationConfig(ctx, bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.LogIf(ctx, err)
|
logger.LogIf(ctx, err)
|
||||||
return diffCh, err
|
return nil, err
|
||||||
}
|
}
|
||||||
tgts, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
tgts, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logger.LogIf(ctx, err)
|
logger.LogIf(ctx, err)
|
||||||
return diffCh, err
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
objInfoCh := make(chan ObjectInfo, 10)
|
||||||
|
if err := objAPI.Walk(ctx, bucket, opts.Prefix, objInfoCh, ObjectOptions{}); err != nil {
|
||||||
|
logger.LogIf(ctx, err)
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
rcfg := replicationConfig{
|
rcfg := replicationConfig{
|
||||||
Config: cfg,
|
Config: cfg,
|
||||||
remotes: tgts,
|
remotes: tgts,
|
||||||
}
|
}
|
||||||
diffCh = make(chan madmin.DiffInfo, 4000)
|
diffCh := make(chan madmin.DiffInfo, 4000)
|
||||||
go func() {
|
go func() {
|
||||||
defer close(diffCh)
|
defer close(diffCh)
|
||||||
for obj := range objInfoCh {
|
for obj := range objInfoCh {
|
||||||
|
if contextCanceled(ctx) {
|
||||||
|
// Just consume input...
|
||||||
|
continue
|
||||||
|
}
|
||||||
// Ignore object prefixes which are excluded
|
// Ignore object prefixes which are excluded
|
||||||
// from versioning via the MinIO bucket versioning extension.
|
// from versioning via the MinIO bucket versioning extension.
|
||||||
if globalBucketVersioningSys.PrefixSuspended(bucket, obj.Name) {
|
if globalBucketVersioningSys.PrefixSuspended(bucket, obj.Name) {
|
||||||
@ -2682,7 +2689,7 @@ func getReplicationDiff(ctx context.Context, objAPI ObjectLayer, bucket string,
|
|||||||
Targets: tgtsMap,
|
Targets: tgtsMap,
|
||||||
}:
|
}:
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
continue
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user