self-hosted/minio
self-hosted/minio
1
0
Fork 0
mirror of https://github.com/minio/minio synced 2024-10-15 04:12:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

relax write quorum requirement for ListBuckets()/HeadBucket() (#18288)

Browse source 
Also fix error handling for HeadBucket() to be pool specific
This commit is contained in:
Harshavardhana 2023-10-20 17:50:21 -07:00 committed by GitHub
parent 8cd80fec8c
commit aa703dc903
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 14 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
cmd/peer-s3-client.go
View file

@ -131,7 +131,6 @@ func (sys *S3PeerSys) ListBuckets(ctx context.Context, opts BucketOptions) ([]Bu
g := errgroup.WithNErrs(len(sys.peerClients))
nodeBuckets := make([][]BucketInfo, len(sys.peerClients))
errs := []error{nil}
for idx, client := range sys.peerClients {
idx := idx
@ -149,7 +148,7 @@ func (sys *S3PeerSys) ListBuckets(ctx context.Context, opts BucketOptions) ([]Bu
}, idx)
}
errs = append(errs, g.Wait()...)
errs := g.Wait()
// The list of buckets in a map to avoid duplication
resultMap := make(map[string]BucketInfo)
@ -161,7 +160,7 @@ func (sys *S3PeerSys) ListBuckets(ctx context.Context, opts BucketOptions) ([]Bu
perPoolErrs = append(perPoolErrs, errs[i])
}
}
quorum := len(perPoolErrs)/2 + 1
quorum := len(perPoolErrs) / 2
if poolErr := reduceWriteQuorumErrs(ctx, perPoolErrs, bucketOpIgnoredErrs, quorum); poolErr != nil {
return nil, poolErr
}
@ -181,7 +180,7 @@ func (sys *S3PeerSys) ListBuckets(ctx context.Context, opts BucketOptions) ([]Bu
continue
}
bucketsMap[bi.Name]++
if bucketsMap[bi.Name] == quorum {
if bucketsMap[bi.Name] >= quorum {
resultMap[bi.Name] = bi
}
}
@ -223,9 +222,17 @@ func (sys *S3PeerSys) GetBucketInfo(ctx context.Context, bucket string, opts Buc
errs := g.Wait()
quorum := len(sys.peerClients)/2 + 1
if err = reduceReadQuorumErrs(ctx, errs, bucketOpIgnoredErrs, quorum); err != nil {
return BucketInfo{}, toObjectErr(err, bucket)
for poolIdx := 0; poolIdx < sys.poolsCount; poolIdx++ {
perPoolErrs := make([]error, 0, len(sys.peerClients))
for i, client := range sys.peerClients {
if slices.Contains(client.GetPools(), poolIdx) {
perPoolErrs = append(perPoolErrs, errs[i])
}
}
quorum := len(perPoolErrs) / 2
if poolErr := reduceWriteQuorumErrs(ctx, perPoolErrs, bucketOpIgnoredErrs, quorum); poolErr != nil {
return BucketInfo{}, poolErr
}
}
for i, err := range errs {