This repository was archived by the owner on Aug 23, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 107
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1536 from grafana/sync-fakemetrics
bring in sync with upstream fakemetrics
- Loading branch information
Showing
9 changed files
with
474 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| package keycache | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| "github.com/grafana/metrictank/schema" | ||
| ) | ||
|
|
||
| // Cache is a single-tenant keycache | ||
| // it is sharded for 2 reasons: | ||
| // * more granular GC (eg. less latency perceived by caller) | ||
| // * mild space savings cause keys are 1 byte shorter | ||
| // We shard on the first byte of the metric key, which we assume | ||
| // is evenly distributed. | ||
| type Cache struct { | ||
| shards [256]Shard | ||
| } | ||
|
|
||
| // NewCache creates a new cache | ||
| func NewCache(ref Ref) *Cache { | ||
| c := Cache{} | ||
| for i := 0; i < 256; i++ { | ||
| c.shards[i] = NewShard(ref) | ||
| } | ||
| return &c | ||
| } | ||
|
|
||
| // Touch marks the key as seen and returns whether it was seen before | ||
| // callers should assure that t >= ref and t-ref <= 42 hours | ||
| func (c *Cache) Touch(key schema.Key, t time.Time) bool { | ||
| shard := int(key[0]) | ||
| return c.shards[shard].Touch(key, t) | ||
| } | ||
|
|
||
| // Len returns the length of the cache | ||
| func (c *Cache) Len() int { | ||
| var sum int | ||
| for i := 0; i < 256; i++ { | ||
| sum += c.shards[i].Len() | ||
| } | ||
| return sum | ||
| } | ||
|
|
||
| // Prune makes sure all shards are pruned | ||
| func (c *Cache) Prune(now time.Time, staleThresh Duration) int { | ||
| var remaining int | ||
| for i := 0; i < 256; i++ { | ||
| remaining += c.shards[i].Prune(now, staleThresh) | ||
| } | ||
| return remaining | ||
| } |
103 changes: 103 additions & 0 deletions
103
stacktest/fakemetrics/out/kafkamdm/keycache/keycache.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| package keycache | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/grafana/metrictank/schema" | ||
| ) | ||
|
|
||
| // KeyCache tracks for all orgs, which keys have been seen, and when was the last time | ||
| type KeyCache struct { | ||
| staleThresh Duration // number of 10-minutely periods | ||
| pruneInterval time.Duration | ||
|
|
||
| sync.RWMutex | ||
| caches map[uint32]*Cache | ||
| } | ||
|
|
||
| // NewKeyCache creates a new KeyCache | ||
| func NewKeyCache(staleThresh, pruneInterval time.Duration) *KeyCache { | ||
| if staleThresh.Hours() > 42 { | ||
| panic("stale time may not exceed 42 hours due to resolution of internal bookkeeping") | ||
| } | ||
| if pruneInterval.Hours() > 42 { | ||
| panic("prune interval may not exceed 42 hours due to resolution of internal bookkeeping") | ||
| } | ||
| if pruneInterval.Minutes() < 10 { | ||
| panic("prune interval less than 10 minutes is useless due to resolution of internal bookkeeping") | ||
| } | ||
| k := &KeyCache{ | ||
| pruneInterval: pruneInterval, | ||
| staleThresh: Duration(int(staleThresh.Seconds()) / 600), | ||
| caches: make(map[uint32]*Cache), | ||
| } | ||
| go k.prune() | ||
| return k | ||
| } | ||
|
|
||
| // Touch marks the key as seen and returns whether it was seen before | ||
| // callers should assure that t >= ref and t-ref <= 42 hours | ||
| func (k *KeyCache) Touch(key schema.MKey, t time.Time) bool { | ||
| k.RLock() | ||
| cache, ok := k.caches[key.Org] | ||
| k.RUnlock() | ||
| // most likely this branch won't execute | ||
| if !ok { | ||
| k.Lock() | ||
| // check again in case another routine has just added it | ||
| cache, ok = k.caches[key.Org] | ||
| if !ok { | ||
| cache = NewCache(NewRef(t)) | ||
| k.caches[key.Org] = cache | ||
| } | ||
| k.Unlock() | ||
| } | ||
| return cache.Touch(key.Key, t) | ||
| } | ||
|
|
||
| // Len returns the size across all orgs | ||
| func (k *KeyCache) Len() int { | ||
| var sum int | ||
| k.RLock() | ||
| caches := make([]*Cache, 0, len(k.caches)) | ||
| for _, c := range k.caches { | ||
| caches = append(caches, c) | ||
| } | ||
| k.RUnlock() | ||
| for _, c := range caches { | ||
| sum += c.Len() | ||
| } | ||
| return sum | ||
| } | ||
|
|
||
| // prune makes sure each org's cache is pruned | ||
| func (k *KeyCache) prune() { | ||
| tick := time.NewTicker(k.pruneInterval) | ||
| for now := range tick.C { | ||
|
|
||
| type target struct { | ||
| org uint32 | ||
| cache *Cache | ||
| } | ||
|
|
||
| k.RLock() | ||
| targets := make([]target, 0, len(k.caches)) | ||
| for org, c := range k.caches { | ||
| targets = append(targets, target{ | ||
| org, | ||
| c, | ||
| }) | ||
| } | ||
| k.RUnlock() | ||
|
|
||
| for _, t := range targets { | ||
| size := t.cache.Prune(now, k.staleThresh) | ||
| if size == 0 { | ||
| k.Lock() | ||
| delete(k.caches, t.org) | ||
| k.Unlock() | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.