Replace remaining uses of bigcache with clockcache

Bigcache is a TTL cache, making it largely inappropriate for our use
case where the cache is mainly a garbage collection mechanism for dead
metrics and series, both of which we expect to change in a
time-correlated fashion. This commit switches the remaining usages to
our own clockcache, whose eviction policy we expect to be better suited
to our usage.

Right now the cache sizes are constants. It may be worth keeping a
hit-rate for each cache so we can start off with a smaller initial size
and grow the caches if the miss-rate is too large.

Author: JLockerman

go.mod

@@ -5,7 +5,6 @@ go 1.14
 require (
 	github.com/NYTimes/gziphandler v1.1.1
 	github.com/OneOfOne/xxhash v1.2.5 // indirect
-	github.com/allegro/bigcache v1.2.1
 	github.com/blang/semver/v4 v4.0.0
 	github.com/docker/go-connections v0.4.0
 	github.com/edsrzf/mmap-go v1.0.0

go.sum

@@ -73,8 +73,6 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
-github.com/allegro/bigcache v1.2.1 h1:hg1sY1raCwic3Vnsvje6TT7/pnZba83LeFck5NrFKSc=
-github.com/allegro/bigcache v1.2.1/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/antihax/optional v0.0.0-20180407024304-ca021399b1a6/go.mod h1:V8iCPQYkqmusNa815XgQio277wI47sdRh1dUOLdyC6Q=
 github.com/apache/arrow/go/arrow v0.0.0-20191024131854-af6fa24be0db/go.mod h1:VTxUBvSJ3s3eHAg65PNgrsn5BtqCRPdmyXh6rAfdxN0=
@@ -419,6 +417,7 @@ github.com/jackc/chunkreader v1.0.0/go.mod h1:RT6O25fNZIuasFJRyZ4R/Y2BbhasbmZXF9
 github.com/jackc/chunkreader/v2 v2.0.0/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
 github.com/jackc/chunkreader/v2 v2.0.1 h1:i+RDz65UE+mmpjTfyz0MoVTnzeYxroil2G82ki7MGG8=
 github.com/jackc/chunkreader/v2 v2.0.1/go.mod h1:odVSm741yZoC3dpHEUXIqA9tQRhFrgOHwnPIn9lDKlk=
+github.com/jackc/fake v0.0.0-20150926172116-812a484cc733/go.mod h1:WrMFNQdiFJ80sQsxDoMokWK1W5TQtxBFNpzWTD84ibQ=
 github.com/jackc/pgerrcode v0.0.0-20190803225404-afa3381909a6 h1:geJ1mgTGd0WQo67wEd+H4OjFG5uA2e3cEBz9D5+pftU=
 github.com/jackc/pgerrcode v0.0.0-20190803225404-afa3381909a6/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgio v1.0.0 h1:g12B9UwVnzGhueNavwioyEEpAmqMe1E/BN9ES+8ovkE=

pkg/log/log.go

@@ -69,16 +69,6 @@ func Fatal(keyvals ...interface{}) {
 	os.Exit(1)
 }
 
-// CustomCacheLogger is a custom logger used for transforming cache logs
-// so that they conform the our logging setup. It also implements the
-// bigcache.Logger interface.
-type CustomCacheLogger struct{}
-
-// Printf sends the log in the debug stream of the logger.
-func (c *CustomCacheLogger) Printf(format string, v ...interface{}) {
-	_ = level.Debug(logger).Log("msg", fmt.Sprintf(format, v...))
-}
-
 func parseLogLevel(logLevel string) (level.Option, error) {
 	switch logLevel {
 	case "debug":

pkg/pgclient/client.go

@@ -8,9 +8,9 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 
-	"github.com/allegro/bigcache"
 	"github.com/jackc/pgx/v4/pgxpool"
 
+	"github.com/timescale/timescale-prometheus/pkg/clockcache"
 	"github.com/timescale/timescale-prometheus/pkg/prompb"
 
 	"github.com/timescale/timescale-prometheus/pkg/log"
@@ -31,6 +31,8 @@ type Config struct {
 	AsyncAcks        bool
 	ReportInterval   int
 	LabelsCacheSize  uint64
+	MetricsCacheSize uint64
+	SeriesCacheSize  uint64
 }
 
 // ParseFlags parses the configuration flags specific to PostgreSQL and TimescaleDB
@@ -45,6 +47,8 @@ func ParseFlags(cfg *Config) *Config {
 	flag.BoolVar(&cfg.AsyncAcks, "async-acks", false, "Ack before data is written to DB")
 	flag.IntVar(&cfg.ReportInterval, "tput-report", 0, "interval in seconds at which throughput should be reported")
 	flag.Uint64Var(&cfg.LabelsCacheSize, "labels-cache-size", 10000, "maximum number of labels to cache")
+	flag.Uint64Var(&cfg.MetricsCacheSize, "metrics-cache-size", pgmodel.DefaultMetricCacheSize, "maximum number of metric names to cache")
+	flag.Uint64Var(&cfg.SeriesCacheSize, "series-cache-size", pgmodel.DefaultSeriesCacheSize, "maximum number of series-sets to cache")
 	return cfg
 }
 
@@ -78,10 +82,13 @@ func NewClient(cfg *Config, readHist prometheus.ObserverVec) (*Client, error) {
 		return nil, err
 	}
 
-	metrics, _ := bigcache.NewBigCache(pgmodel.DefaultCacheConfig())
-	cache := &pgmodel.MetricNameCache{Metrics: metrics}
+	cache := &pgmodel.MetricNameCache{Metrics: clockcache.WithMax(cfg.MetricsCacheSize)}
 
-	c := pgmodel.Cfg{AsyncAcks: cfg.AsyncAcks, ReportInterval: cfg.ReportInterval}
+	c := pgmodel.Cfg{
+		AsyncAcks:       cfg.AsyncAcks,
+		ReportInterval:  cfg.ReportInterval,
+		SeriesCacheSize: cfg.SeriesCacheSize,
+	}
 	ingestor, err := pgmodel.NewPgxIngestorWithMetricCache(connectionPool, cache, &c)
 	if err != nil {
 		log.Error("err starting ingestor", err)

pkg/pgmodel/bigcache.go

@@ -0,0 +1,66 @@
+// This file and its contents are licensed under the Apache License 2.0.
+// Please see the included NOTICE for copyright information and
+// LICENSE for a copy of the license.
+
+package pgmodel
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/timescale/timescale-prometheus/pkg/clockcache"
+)
+
+const (
+	DefaultMetricCacheSize = 10000
+	DefaultSeriesCacheSize = 100000
+)
+
+var (
+	// ErrEntryNotFound is returned when entry is not found.
+	ErrEntryNotFound = fmt.Errorf("entry not found")
+)
+
+type seriesCache struct {
+	series *clockcache.Cache
+}
+
+func (b *seriesCache) GetSeries(lset Labels) (SeriesID, error) {
+	result, ok := b.series.Get(lset.String())
+	if !ok {
+		return 0, ErrEntryNotFound
+	}
+	return result.(SeriesID), nil
+}
+
+func (b *seriesCache) SetSeries(lset Labels, id SeriesID) error {
+	b.series.Insert(lset.String(), id)
+	return nil
+}
+
+// MetricNameCache stores and retrieves metric table names in a in-memory cache.
+type MetricNameCache struct {
+	Metrics *clockcache.Cache
+}
+
+// Get fetches the table name for specified metric.
+func (m *MetricNameCache) Get(metric string) (string, error) {
+	result, ok := m.Metrics.Get(metric)
+	if !ok {
+		return "", ErrEntryNotFound
+	}
+	return result.(string), nil
+}
+
+// Set stores table name for specified metric.
+func (m *MetricNameCache) Set(metric string, tableName string) error {
+	// deep copy the strings so the original memory doesn't need to stick around
+	metricBuilder := strings.Builder{}
+	metricBuilder.Grow(len(metric))
+	metricBuilder.WriteString(metric)
+	tableBuilder := strings.Builder{}
+	tableBuilder.Grow(len(tableName))
+	tableBuilder.WriteString(tableName)
+	m.Metrics.Insert(metricBuilder.String(), tableBuilder.String())
+	return nil
+}

pkg/pgmodel/cache.go

@@ -7,20 +7,14 @@ import (
 	"math"
 	"strings"
 	"testing"
-	"time"
 
-	"github.com/allegro/bigcache"
 	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/timescale/timescale-prometheus/pkg/clockcache"
 )
 
-func TestBigCache(t *testing.T) {
-	config := bigcache.DefaultConfig(10 * time.Minute)
-	series, err := bigcache.NewBigCache(config)
-	if err != nil {
-		t.Fatal("unable to run test, unable to create labels cache")
-	}
-	cache := bCache{
-		series: series,
+func TestCache(t *testing.T) {
+	cache := seriesCache{
+		series: clockcache.WithMax(100),
 	}
 
 	label := labels.Labels{
@@ -145,17 +139,11 @@ func TestMetricTableNameCache(t *testing.T) {
 			tableName: "",
 		},
 	}
-	config := bigcache.DefaultConfig(10 * time.Minute)
 
 	for _, c := range testCases {
 		t.Run(c.name, func(t *testing.T) {
-			metrics, err := bigcache.NewBigCache(config)
-
-			if err != nil {
-				t.Fatal("unable to run test, unable to create metrics table name cache")
-			}
 			cache := MetricNameCache{
-				Metrics: metrics,
+				Metrics: clockcache.WithMax(100),
 			}
 
 			missing, err := cache.Get(c.metric)

pkg/pgmodel/bigcache_test.go renamed to pkg/pgmodel/cache_test.go

@@ -15,7 +15,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/allegro/bigcache"
 	"github.com/jackc/pgconn"
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v4"
@@ -189,8 +188,9 @@ func (t *SampleInfoIterator) Err() error {
 }
 
 type Cfg struct {
-	AsyncAcks      bool
-	ReportInterval int
+	AsyncAcks       bool
+	ReportInterval  int
+	SeriesCacheSize uint64
 }
 
 // NewPgxIngestorWithMetricCache returns a new Ingestor that uses connection pool and a metrics cache
@@ -206,11 +206,7 @@ func NewPgxIngestorWithMetricCache(c *pgxpool.Pool, cache MetricCache, cfg *Cfg)
 		return nil, err
 	}
 
-	series, _ := bigcache.NewBigCache(DefaultCacheConfig())
-
-	bc := &bCache{
-		series: series,
-	}
+	bc := &seriesCache{clockcache.WithMax(cfg.SeriesCacheSize)}
 
 	return &DBIngestor{
 		db:    pi,
@@ -220,8 +216,8 @@ func NewPgxIngestorWithMetricCache(c *pgxpool.Pool, cache MetricCache, cfg *Cfg)
 
 // NewPgxIngestor returns a new Ingestor that write to PostgreSQL using PGX
 func NewPgxIngestor(c *pgxpool.Pool) (*DBIngestor, error) {
-	metrics, _ := bigcache.NewBigCache(DefaultCacheConfig())
-	cache := &MetricNameCache{metrics}
+	//TODO what size should we use?
+	cache := &MetricNameCache{clockcache.WithMax(DefaultMetricCacheSize)}
 	return NewPgxIngestorWithMetricCache(c, cache, &Cfg{})
 }
 
@@ -900,8 +896,7 @@ func NewPgxReaderWithMetricCache(c *pgxpool.Pool, cache MetricCache, labelsCache
 
 // NewPgxReader returns a new DBReader that reads that from PostgreSQL using PGX.
 func NewPgxReader(c *pgxpool.Pool, readHist prometheus.ObserverVec, labelsCacheSize uint64) *DBReader {
-	metrics, _ := bigcache.NewBigCache(DefaultCacheConfig())
-	cache := &MetricNameCache{metrics}
+	cache := &MetricNameCache{clockcache.WithMax(DefaultMetricCacheSize)}
 	return NewPgxReaderWithMetricCache(c, cache, labelsCacheSize)
 }