MyFavMirrors/minio
1
0
Fork 0
mirror of https://github.com/minio/minio.git synced 2025-04-12 07:22:18 -04:00
Code Issues Packages Projects Releases Wiki Activity

metadata: Invalidate cache if unreadable and not updating (#10844)

Browse Source 
If a scanning server shuts down unexpectedly we may have "successful" caches that are incomplete on a set.

In this case mark the cache with an error so it will no longer be handed out.
This commit is contained in:
Klaus Post 2020-11-06 08:54:09 -08:00 committed by GitHub
parent 74f7cf24ae
commit 9bf5990ea9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
cmd/metacache-manager.go
View File

@ -233,9 +233,7 @@ func (o listPathOptions) checkMetacacheState(ctx context.Context, rpc *peerRESTC
// We operate on a copy... // We operate on a copy...
o.Create = false o.Create = false
var cache metacache var cache metacache
if !o.Transient {
if rpc == nil || o.Transient { if rpc == nil || o.Transient {
// Local
cache = localMetacacheMgr.findCache(ctx, o) cache = localMetacacheMgr.findCache(ctx, o)
} else { } else {
c, err := rpc.GetMetacacheListing(ctx, o) c, err := rpc.GetMetacacheListing(ctx, o)
@ -244,14 +242,22 @@ func (o listPathOptions) checkMetacacheState(ctx context.Context, rpc *peerRESTC
} }
cache = *c cache = *c
} }
}
if cache.status == scanStateNone || cache.fileNotFound { if cache.status == scanStateNone || cache.fileNotFound {
return errFileNotFound return errFileNotFound
} }
if cache.status == scanStateSuccess || cache.status == scanStateStarted { if cache.status == scanStateSuccess || cache.status == scanStateStarted {
if time.Since(cache.lastUpdate) > metacacheMaxRunningAge { if time.Since(cache.lastUpdate) > metacacheMaxRunningAge {
return fmt.Errorf("timeout: list %s not updated for 1 minute", cache.id) // We got a stale entry, mark error on handling server.
err := fmt.Errorf("timeout: list %s not updated", cache.id)
cache.error = err.Error()
cache.status = scanStateError
if rpc == nil || o.Transient {
localMetacacheMgr.updateCacheEntry(cache)
} else {
rpc.UpdateMetacacheListing(ctx, cache)
}
return err
} }
return nil return nil
} }