store/cache: fix broken metric and current index cache size handling (#873)

* store/cache: do not forget to increase c.current on adding new items

* store/cache: properly adjust c.curSize

* store/cache: prevent uint64 overflow by switching operands

Adding uint64(len(b)) to c.curSize might potentially overflow uint64 if
the numbers are big enough and then we might not remove enough items
from the LRU to satisfy the request. On the other hand, switching the
operands avoids this problem because we check before if uint64(len(b))
is bigger than c.maxSize so subtracting uint64(len(b)) will *never*
overflow because we know that it is less or equal to c.maxSize.

* store/cache: revert ensureFits() changes

c.curSize is lowered in onEvict.

* store/cache: add smoke tests

Add smoke tests for the index cache which check if we set curSize
properly, and if removal works.

Author: GiedriusS

pkg/store/cache.go

@@ -127,7 +127,7 @@ func (c *indexCache) ensureFits(b []byte) bool {
 	if uint64(len(b)) > c.maxSize {
 		return false
 	}
-	for c.curSize+uint64(len(b)) > c.maxSize {
+	for c.curSize > c.maxSize-uint64(len(b)) {
 		c.lru.RemoveOldest()
 	}
 	return true
@@ -151,6 +151,8 @@ func (c *indexCache) setPostings(b ulid.ULID, l labels.Label, v []byte) {
 	c.lru.Add(cacheItem{b, cacheKeyPostings(l)}, cv)
 
 	c.currentSize.WithLabelValues(cacheTypePostings).Add(float64(len(v)))
+	c.current.WithLabelValues(cacheTypePostings).Inc()
+	c.curSize += uint64(len(v))
 }
 
 func (c *indexCache) postings(b ulid.ULID, l labels.Label) ([]byte, bool) {
@@ -185,6 +187,8 @@ func (c *indexCache) setSeries(b ulid.ULID, id uint64, v []byte) {
 	c.lru.Add(cacheItem{b, cacheKeySeries(id)}, cv)
 
 	c.currentSize.WithLabelValues(cacheTypeSeries).Add(float64(len(v)))
+	c.current.WithLabelValues(cacheTypeSeries).Inc()
+	c.curSize += uint64(len(v))
 }
 
 func (c *indexCache) series(b ulid.ULID, id uint64) ([]byte, bool) {

pkg/store/cache_test.go

@@ -3,9 +3,13 @@ package store
 
 import (
 	"testing"
+	"time"
 
+	"github.com/fortytw2/leaktest"
 	"github.com/improbable-eng/thanos/pkg/testutil"
+	"github.com/oklog/ulid"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/tsdb/labels"
 )
 
 // TestIndexCacheEdge tests the index cache edge cases.
@@ -20,3 +24,38 @@ func TestIndexCacheEdge(t *testing.T) {
 	fits = cache.ensureFits([]byte{42})
 	testutil.Equals(t, fits, true)
 }
+
+// TestIndexCacheSmoke runs the smoke tests for the index cache.
+func TestIndexCacheSmoke(t *testing.T) {
+	defer leaktest.CheckTimeout(t, 10*time.Second)()
+
+	metrics := prometheus.NewRegistry()
+	cache, err := newIndexCache(metrics, 20)
+	testutil.Ok(t, err)
+
+	blid := ulid.ULID([16]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16})
+	labels := labels.Label{Name: "test", Value: "123"}
+
+	cache.setPostings(blid, labels, []byte{42})
+
+	p, ok := cache.postings(blid, labels)
+	testutil.Equals(t, ok, true)
+	testutil.Equals(t, p, []byte{42})
+	testutil.Equals(t, cache.curSize, uint64(1))
+
+	cache.setSeries(blid, 1234, []byte{42, 42})
+
+	s, ok := cache.series(blid, 1234)
+	testutil.Equals(t, ok, true)
+	testutil.Equals(t, s, []byte{42, 42})
+	testutil.Equals(t, cache.curSize, uint64(3))
+
+	cache.lru.RemoveOldest()
+	testutil.Equals(t, cache.curSize, uint64(2))
+
+	cache.lru.RemoveOldest()
+	testutil.Equals(t, cache.curSize, uint64(0))
+
+	cache.lru.RemoveOldest()
+	testutil.Equals(t, cache.curSize, uint64(0))
+}