MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-11-20 01:50:24 -05:00
Code Issues Packages Projects Releases Wiki Activity

Revamp bucket metadata healing (#7208)

Browse Source 
Bucket metadata healing in the current code was executed multiple
times each time for a given set. Bucket metadata just like
objects are hashed in accordance with its name on any given set,
to allow hashing to play a role we should let the top level
code decide where to navigate.

Current code also had 3 bucket metadata files hardcoded, whereas
we should make it generic by listing and navigating the .minio.sys
to heal such objects.

We also had another bug where due to isObjectDangling changes
without pre-existing bucket metadata files, we were erroneously
reporting it as grey/corrupted objects.

This PR fixes all of the above items.
This commit is contained in:
Harshavardhana
2019-02-10 19:53:13 -08:00
committed by Nitish Tiwari
parent 9600e2b35e
commit 082f777281
7 changed files with 96 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
cmd/xl-sets.go
View File

@@ -1232,15 +1232,15 @@ func (s *xlSets) HealFormat(ctx context.Context, dryRun bool) (res madmin.HealRe
}
// HealBucket - heals inconsistent buckets and bucket metadata on all sets.
func (s *xlSets) HealBucket(ctx context.Context, bucket string, dryRun, remove bool) (results []madmin.HealResultItem, err error) {
func (s *xlSets) HealBucket(ctx context.Context, bucket string, dryRun, remove bool) (result madmin.HealResultItem, err error) {
bucketLock := globalNSMutex.NewNSLock(bucket, "")
if err := bucketLock.GetLock(globalHealingTimeout); err != nil {
return nil, err
return result, err
}
defer bucketLock.Unlock()
// Initialize heal result info
res := madmin.HealResultItem{
result = madmin.HealResultItem{
Type: madmin.HealItemBucket,
Bucket: bucket,
DiskCount: s.setCount * s.drivesPerSet,
@@ -1248,25 +1248,22 @@ func (s *xlSets) HealBucket(ctx context.Context, bucket string, dryRun, remove b
}
for _, s := range s.sets {
var setResults []madmin.HealResultItem
setResults, _ = s.HealBucket(ctx, bucket, dryRun, remove)
for _, setResult := range setResults {
if setResult.Type == madmin.HealItemBucket {
for _, v := range setResult.Before.Drives {
res.Before.Drives = append(res.Before.Drives, v)
}
for _, v := range setResult.After.Drives {
res.After.Drives = append(res.After.Drives, v)
}
continue
}
results = append(results, setResult)
var healResult madmin.HealResultItem
healResult, err = s.HealBucket(ctx, bucket, dryRun, remove)
if err != nil {
return result, err
}
for _, v := range healResult.Before.Drives {
result.Before.Drives = append(result.Before.Drives, v)
}
for _, v := range healResult.After.Drives {
result.After.Drives = append(result.After.Drives, v)
}
}
for _, endpoint := range s.endpoints {
var foundBefore bool
for _, v := range res.Before.Drives {
for _, v := range result.Before.Drives {
if endpoint.IsLocal {
if v.Endpoint == endpoint.Path {
foundBefore = true
@@ -1278,14 +1275,14 @@ func (s *xlSets) HealBucket(ctx context.Context, bucket string, dryRun, remove b
}
}
if !foundBefore {
res.Before.Drives = append(res.Before.Drives, madmin.HealDriveInfo{
result.Before.Drives = append(result.Before.Drives, madmin.HealDriveInfo{
UUID: "",
Endpoint: endpoint.String(),
State: madmin.DriveStateOffline,
})
}
var foundAfter bool
for _, v := range res.After.Drives {
for _, v := range result.After.Drives {
if endpoint.IsLocal {
if v.Endpoint == endpoint.Path {
foundAfter = true
@@ -1297,7 +1294,7 @@ func (s *xlSets) HealBucket(ctx context.Context, bucket string, dryRun, remove b
}
}
if !foundAfter {
res.After.Drives = append(res.After.Drives, madmin.HealDriveInfo{
result.After.Drives = append(result.After.Drives, madmin.HealDriveInfo{
UUID: "",
Endpoint: endpoint.String(),
State: madmin.DriveStateOffline,
@@ -1306,14 +1303,12 @@ func (s *xlSets) HealBucket(ctx context.Context, bucket string, dryRun, remove b
}
// Check if we had quorum to write, if not return an appropriate error.
_, afterDriveOnline := res.GetOnlineCounts()
_, afterDriveOnline := result.GetOnlineCounts()
if afterDriveOnline < ((s.setCount*s.drivesPerSet)/2)+1 {
return nil, toObjectErr(errXLWriteQuorum, bucket)
return result, toObjectErr(errXLWriteQuorum, bucket)
}
results = append(results, res)
return results, nil
return result, nil
}
// HealObject - heals inconsistent object on a hashedSet based on object name.