MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2024-12-24 06:05:55 -05:00
Code Issues Packages Projects Releases Wiki Activity

fs: Fix non-progressing scanner (#13218)

Browse Source 
Scanner would keep doing the same cycle in FS mode leading to missed updates.

Add a few sanity checks and handle errors better.
This commit is contained in:
Klaus Post 2021-09-15 09:24:41 -07:00 committed by GitHub
parent bf409936e7
commit f98f115ac2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 19 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
cmd/data-scanner.go
View File

@ -405,11 +405,11 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
err := readDirFn(path.Join(f.root, folder.name), func(entName string, typ os.FileMode) error { err := readDirFn(path.Join(f.root, folder.name), func(entName string, typ os.FileMode) error {
// Parse // Parse
entName = pathClean(path.Join(folder.name, entName)) entName = pathClean(path.Join(folder.name, entName))
if entName == "" { if entName == "" || entName == folder.name {
if f.dataUsageScannerDebug { if f.dataUsageScannerDebug {
console.Debugf(scannerLogPrefix+" no bucket (%s,%s)\n", f.root, entName) console.Debugf(scannerLogPrefix+" no entity (%s,%s)\n", f.root, entName)
} }
return errDoneForNow return nil
} }
bucket, prefix := path2BucketObjectWithBasePath(f.root, entName) bucket, prefix := path2BucketObjectWithBasePath(f.root, entName)
if bucket == "" { if bucket == "" {
@ -435,7 +435,9 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
if typ&os.ModeDir != 0 { if typ&os.ModeDir != 0 {
h := hashPath(entName) h := hashPath(entName)
_, exists := f.oldCache.Cache[h.Key()] _, exists := f.oldCache.Cache[h.Key()]
if h == thisHash {
return nil
}
this := cachedFolder{name: entName, parent: &thisHash, objectHealProbDiv: folder.objectHealProbDiv} this := cachedFolder{name: entName, parent: &thisHash, objectHealProbDiv: folder.objectHealProbDiv}
delete(abandonedChildren, h.Key()) // h.Key() already accounted for. delete(abandonedChildren, h.Key()) // h.Key() already accounted for.
if exists { if exists {
@ -468,11 +470,14 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
item.heal = item.heal && f.healObjectSelect > 0 item.heal = item.heal && f.healObjectSelect > 0
sz, err := f.getSize(item) sz, err := f.getSize(item)
if err == errSkipFile { if err != nil {
wait() // wait to proceed to next entry. wait() // wait to proceed to next entry.
if err != errSkipFile && f.dataUsageScannerDebug {
console.Debugf(scannerLogPrefix+" getSize \"%v/%v\" returned err: %v\n", bucket, item.objectPath(), err)
}
return nil return nil
} }
// successfully read means we have a valid object. // successfully read means we have a valid object.
foundObjects = true foundObjects = true
// Remove filename i.e is the meta file to construct object name // Remove filename i.e is the meta file to construct object name

9
cmd/fs-v1.go
View File

@ -258,6 +258,12 @@ func (fs *FSObjects) NSScanner(ctx context.Context, bf *bloomFilter, updates cha
updates <- totalCache.dui(dataUsageRoot, buckets) updates <- totalCache.dui(dataUsageRoot, buckets)
return nil return nil
} }
for i, b := range buckets {
if isReservedOrInvalidBucket(b.Name, false) {
// Delete bucket...
buckets = append(buckets[:i], buckets[i+1:]...)
}
}
totalCache.Info.BloomFilter = bf.bytes() totalCache.Info.BloomFilter = bf.bytes()
@ -282,6 +288,7 @@ func (fs *FSObjects) NSScanner(ctx context.Context, bf *bloomFilter, updates cha
bCache.Info.Name = b.Name bCache.Info.Name = b.Name
} }
bCache.Info.BloomFilter = totalCache.Info.BloomFilter bCache.Info.BloomFilter = totalCache.Info.BloomFilter
bCache.Info.NextCycle = wantCycle
upds := make(chan dataUsageEntry, 1) upds := make(chan dataUsageEntry, 1)
var wg sync.WaitGroup var wg sync.WaitGroup
wg.Add(1) wg.Add(1)
@ -290,7 +297,7 @@ func (fs *FSObjects) NSScanner(ctx context.Context, bf *bloomFilter, updates cha
for update := range upds { for update := range upds {
totalCache.replace(b.Name, dataUsageRoot, update) totalCache.replace(b.Name, dataUsageRoot, update)
if intDataUpdateTracker.debug { if intDataUpdateTracker.debug {
logger.Info(color.Green("NSScanner:")+" Got update:", len(totalCache.Cache)) logger.Info(color.Green("NSScanner:")+" Got update: %v", len(totalCache.Cache))
} }
cloned := totalCache.clone() cloned := totalCache.clone()
updates <- cloned.dui(dataUsageRoot, buckets) updates <- cloned.dui(dataUsageRoot, buckets)