Merge pull request #543 from dinal/master

dev -> master

Author: dinal

pkg/appender/appender.go

@@ -56,7 +56,7 @@ type MetricState struct {
 	store      *chunkStore
 	err        error
 	retryCount uint8
-	newName    bool
+	created    bool
 	isVariant  bool
 
 	shouldGetState bool
@@ -147,7 +147,6 @@ func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config
 	newCache.newUpdates = make(chan int, 1000)
 	newCache.stopChan = make(chan int, 3)
 
-	newCache.NameLabelMap = map[string]bool{}
 	newCache.performanceReporter = performance.ReporterInstanceFromConfig(cfg)
 
 	return &newCache
@@ -182,10 +181,6 @@ func (mc *MetricsCache) getMetric(hash uint64) (*MetricState, bool) {
 // create a new metric and save in the map
 func (mc *MetricsCache) addMetric(hash uint64, name string, metric *MetricState) {
 	mc.cacheMetricMap.Add(hash, metric)
-	if _, ok := mc.NameLabelMap[name]; !ok {
-		metric.newName = true
-		mc.NameLabelMap[name] = true
-	}
 }
 
 // Push append to async channel

pkg/appender/ingest.go

@@ -241,6 +241,9 @@ func (mc *MetricsCache) postMetricUpdates(metric *MetricState) {
 		if !sent {
 			if metric.store.samplesQueueLength() == 0 {
 				metric.setState(storeStateReady)
+				if metric.store.numNotProcessed == 0 {
+					mc.cacheMetricMap.ResetMetric(metric.hash)
+				}
 			} else {
 				if mc.metricQueue.length() > 0 {
 					atomic.AddInt64(&mc.outstandingUpdates, 1)

pkg/appender/lru_cache.go

@@ -35,34 +35,31 @@ func NewCache(max int) *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
+	if ele, ok := c.cache[key]; ok {
+		//deleting from both lists as a workaround to not knowing which list actually contains the elem
+		//if the elem is not present in the list Remove won't do anything
+		// Remove clears ele's head,prev and next pointers so it can't be reused
+		c.free.Remove(ele)
+		c.used.Remove(ele)
 	}
-	ele := c.used.PushFront(&entry{key, value})
-	c.cache[key] = ele
+	c.cache[key] = c.used.PushFront(&entry{key, value})
 	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 {
+		//deleting from both lists as a workaround to not knowing which list actually contains the elem
+		//if the elem is not present in the list Remove won't do anything
+		// Remove clears ele's head,prev and next pointers so it can't be reused
 		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)
-		}
+		c.used.Remove(ele)
+		c.cache[key] = c.used.PushFront(&entry{key, ele.Value.(*entry).value})
 		return ele.Value.(*entry).value, true
 	}
 	return
@@ -73,8 +70,7 @@ func (c *Cache) removeOldest() {
 		ele := c.free.Back()
 		if ele != nil {
 			c.free.Remove(ele)
-			kv := ele.Value.(*clist.Element).Value.(*entry)
-			delete(c.cache, kv.key)
+			delete(c.cache, ele.Value.(*entry).key)
 			return
 		}
 		c.cond.Wait()
@@ -85,8 +81,12 @@ func (c *Cache) ResetMetric(key uint64) {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 	if ele, ok := c.cache[key]; ok {
+		//deleting from both lists as a workaround to not knowing which list actually contains the elem
+		//if the elem is not present in the list Remove won't do anything
+		// Remove clears ele's head,prev and next pointers so it can't be reused
 		c.used.Remove(ele)
-		c.free.PushFront(ele)
+		c.free.Remove(ele)
+		c.cache[key] = c.free.PushFront(&entry{key, ele.Value.(*entry).value})
 		c.cond.Signal()
 	}
 }

pkg/appender/store.go

@@ -186,7 +186,8 @@ func (cs *chunkStore) processGetResp(mc *MetricsCache, metric *MetricState, resp
 
 	if resp.Error != nil {
 		if utils.IsNotExistsError(resp.Error) {
-			if metric.newName {
+			if !metric.created {
+				metric.created = true
 				path := filepath.Join(mc.cfg.TablePath, config.NamesDirectory, metric.name)
 				putInput := v3io.PutItemInput{Path: path, Attributes: map[string]interface{}{}}