don't remove in flight metric from cache (#538) (#539)

* don't remove in flight metric from cache

* rename

* remove only unused metrics from cache

* add push to channel counter

* lint

* more lint

* remove unneeded check

* move to used list when touching an element

* typo

* use for loop instead of recursion

* remove null check

* add comment on log

* go mod tidy

* add a print of cache size

Co-authored-by: Dina Nimrodi <[email protected]>

Co-authored-by: Dina Nimrodi <[email protected]>

Author: dinal

go.mod

@@ -6,7 +6,6 @@ require (
 	github.com/cespare/xxhash v1.1.0
 	github.com/cpuguy83/go-md2man v1.0.10 // indirect
 	github.com/ghodss/yaml v1.0.0
-	github.com/golang/groupcache v0.0.0-20191027212112-611e8accdfc9
 	github.com/imdario/mergo v0.3.7
 	github.com/nuclio/logger v0.0.1
 	github.com/nuclio/nuclio-sdk-go v0.0.0-20190205170814-3b507fbd0324

pkg/appender/appender.go

@@ -25,7 +25,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/golang/groupcache/lru"
 	"github.com/nuclio/logger"
 	"github.com/pkg/errors"
 	"github.com/v3io/v3io-go/pkg/dataplane"
@@ -107,7 +106,6 @@ func (m *MetricState) error() error {
 type MetricsCache struct {
 	cfg           *config.V3ioConfig
 	partitionMngr *partmgr.PartitionManager
-	mtx           sync.RWMutex
 	container     v3io.Container
 	logger        logger.Logger
 	started       bool
@@ -124,7 +122,7 @@ type MetricsCache struct {
 	outstandingUpdates int64
 	requestsInFlight   int64
 
-	cacheMetricMap *lru.Cache
+	cacheMetricMap *Cache
 
 	NameLabelMap map[string]bool // temp store all lable names
 
@@ -138,7 +136,8 @@ func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config
 	partMngr *partmgr.PartitionManager) *MetricsCache {
 
 	newCache := MetricsCache{container: container, logger: logger, cfg: cfg, partitionMngr: partMngr}
-	newCache.cacheMetricMap = lru.New(cfg.MetricCacheSize)
+	newCache.cacheMetricMap = NewCache(cfg.MetricCacheSize)
+	newCache.logger.DebugWith("Initializing new metric cache", "size", cfg.MetricCacheSize)
 
 	newCache.responseChan = make(chan *v3io.Response, channelSize)
 	newCache.nameUpdateChan = make(chan *v3io.Response, channelSize)
@@ -173,21 +172,15 @@ func (mc *MetricsCache) Start() error {
 
 // return metric struct by key
 func (mc *MetricsCache) getMetric(hash uint64) (*MetricState, bool) {
-	mc.mtx.RLock()
-	defer mc.mtx.RUnlock()
-
 	metric, ok := mc.cacheMetricMap.Get(hash)
 	if ok {
-		return metric.(*MetricState), ok
+		return metric, ok
 	}
 	return nil, ok
 }
 
 // create a new metric and save in the map
 func (mc *MetricsCache) addMetric(hash uint64, name string, metric *MetricState) {
-	mc.mtx.Lock()
-	defer mc.mtx.Unlock()
-
 	mc.cacheMetricMap.Add(hash, metric)
 	if _, ok := mc.NameLabelMap[name]; !ok {
 		metric.newName = true
@@ -197,6 +190,9 @@ func (mc *MetricsCache) addMetric(hash uint64, name string, metric *MetricState)
 
 // Push append to async channel
 func (mc *MetricsCache) appendTV(metric *MetricState, t int64, v interface{}) {
+	metric.Lock()
+	defer metric.Unlock()
+	metric.store.numNotProcessed++
 	mc.asyncAppendChan <- &asyncAppend{metric: metric, t: t, v: v}
 }
 
@@ -280,6 +276,7 @@ func (mc *MetricsCache) AddFast(ref uint64, t int64, v interface{}) error {
 	}
 	metric, ok := mc.getMetric(ref)
 	if !ok {
+		// do not change error msg, it's parsed by prom
 		return fmt.Errorf(fmt.Sprintf("metric not found. ref=%v", ref))
 	}
 

pkg/appender/ingest.go

@@ -76,7 +76,7 @@ func (mc *MetricsCache) metricFeed(index int) {
 						// Handle append requests (Add / AddFast)
 						metric := app.metric
 						metric.Lock()
-
+						metric.store.numNotProcessed--
 						metric.store.Append(app.t, app.v)
 						numPushed++
 						dataQueued += metric.store.samplesQueueLength()
@@ -126,7 +126,6 @@ func (mc *MetricsCache) metricFeed(index int) {
 						}
 					}
 				}
-
 				// If we have too much work, stall the queue for some time
 				if numPushed > mc.cfg.BatchSize/2 && dataQueued/numPushed > 64 {
 					switch {
@@ -331,6 +330,7 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 				metric.store = newChunkStore(mc.logger, metric.Lset.LabelNames(), metric.store.isAggr())
 				metric.retryCount = 0
 				metric.setState(storeStateInit)
+				mc.cacheMetricMap.ResetMetric(metric.hash)
 			}
 
 			// Count errors
@@ -380,6 +380,9 @@ func (mc *MetricsCache) handleResponse(metric *MetricState, resp *v3io.Response,
 		mc.metricQueue.Push(metric)
 		metric.setState(storeStateAboutToUpdate)
 	}
+	if !sent && metric.store.numNotProcessed == 0 && metric.store.pending.Len() == 0 {
+		mc.cacheMetricMap.ResetMetric(metric.hash)
+	}
 
 	return sent
 }

pkg/appender/lru_cache.go

@@ -0,0 +1,92 @@
+package appender
+
+import (
+	clist "container/list"
+	"sync"
+)
+
+// Cache is an LRU cache. It is not safe for concurrent access.
+type Cache struct {
+	maxEntries int
+	free       *clist.List
+	used       *clist.List
+	cache      map[uint64]*clist.Element
+	cond       *sync.Cond
+	mtx        sync.Mutex
+}
+
+type entry struct {
+	key   uint64
+	value *MetricState
+}
+
+func NewCache(max int) *Cache {
+	newCache := Cache{
+		maxEntries: max,
+		free:       clist.New(),
+		used:       clist.New(),
+		cache:      make(map[uint64]*clist.Element),
+	}
+	newCache.cond = sync.NewCond(&newCache.mtx)
+	return &newCache
+}
+
+// Add adds a value to the cache.
+func (c *Cache) Add(key uint64, value *MetricState) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	if ee, ok := c.cache[key]; ok {
+		c.free.Remove(ee)
+		//check if element was already in list and if not push to front
+		c.used.MoveToFront(ee)
+		if c.used.Front() != ee {
+			c.used.PushFront(ee)
+		}
+		ee.Value.(*entry).value = value
+		return
+	}
+	ele := c.used.PushFront(&entry{key, value})
+	c.cache[key] = ele
+	if c.maxEntries != 0 && c.free.Len()+c.used.Len() > c.maxEntries {
+		c.removeOldest()
+	}
+}
+
+// Get looks up a key's value from the cache.
+func (c *Cache) Get(key uint64) (value *MetricState, ok bool) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	if ele, hit := c.cache[key]; hit {
+		c.free.Remove(ele)
+		//check if element was already in list and if not push to front
+		c.used.MoveToFront(ele)
+		if c.used.Front() != ele {
+			c.used.PushFront(ele)
+		}
+		return ele.Value.(*entry).value, true
+	}
+	return
+}
+
+func (c *Cache) removeOldest() {
+	for {
+		ele := c.free.Back()
+		if ele != nil {
+			c.free.Remove(ele)
+			kv := ele.Value.(*clist.Element).Value.(*entry)
+			delete(c.cache, kv.key)
+			return
+		}
+		c.cond.Wait()
+	}
+}
+
+func (c *Cache) ResetMetric(key uint64) {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	if ele, ok := c.cache[key]; ok {
+		c.used.Remove(ele)
+		c.free.PushFront(ele)
+		c.cond.Signal()
+	}
+}

pkg/appender/store.go

@@ -66,11 +66,12 @@ type chunkStore struct {
 	lastTid  int64
 	chunks   [2]*attrAppender
 
-	labelNames    []string
-	aggrList      *aggregate.AggregatesList
-	pending       pendingList
-	maxTime       int64
-	delRawSamples bool // TODO: for metrics w aggregates only
+	labelNames      []string
+	aggrList        *aggregate.AggregatesList
+	pending         pendingList
+	maxTime         int64
+	delRawSamples   bool // TODO: for metrics w aggregates only
+	numNotProcessed int64
 }
 
 func (cs *chunkStore) isAggr() bool {