Add counters for the various cache sizes

Add prometheus counters for the number of elements stored int the
various caches we have. This should help us debug cache-sizing issues,
should any occur.

Author: JLockerman

cmd/timescale-prometheus/main.go

@@ -10,13 +10,14 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"github.com/prometheus/common/route"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
 	"sync/atomic"
 	"time"
 
+	"github.com/prometheus/common/route"
+
 	"github.com/jackc/pgx/v4/pgxpool"
 	_ "github.com/jackc/pgx/v4/stdlib"
 
@@ -194,6 +195,61 @@ func main() {
 	}
 	defer client.Close()
 
+	cachedMetricNames := prometheus.NewCounterFunc(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "metric_name_cache_elements_stored",
+		Help:      "Total number of metric names in the metric name cache.",
+	}, func() float64 {
+		return float64(client.NumCachedMetricNames())
+	})
+
+	metricNamesCacheCap := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Namespace: promNamespace,
+		Name:      "metric_name_cache_capacity",
+		Help:      "Maximum number of elements in the metric names cache.",
+	}, func() float64 {
+		return float64(client.MetricNamesCacheCapacity())
+	})
+
+	cachedSeriesSets := prometheus.NewCounterFunc(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "series_set_cache_elements_stored",
+		Help:      "Total number of labels to series-set id mappings cached.",
+	}, func() float64 {
+		return float64(client.NumCachedSeries())
+	})
+
+	seriesSetCacheCap := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Namespace: promNamespace,
+		Name:      "series_set_cache_capacity",
+		Help:      "Maximum number of elements in the series-set cache.",
+	}, func() float64 {
+		return float64(client.SeriesCacheCapacity())
+	})
+
+	cachedLabels := prometheus.NewCounterFunc(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Name:      "label_cache_elements_stored",
+		Help:      "Total number of label-id to label mappings cache.",
+	}, func() float64 {
+		return float64(client.NumCachedLabels())
+	})
+
+	labelsCacheCap := prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Namespace: promNamespace,
+		Name:      "label_cache_capacity",
+		Help:      "Total number of label-id to label mappings cache.",
+	}, func() float64 {
+		return float64(client.LabelsCacheCapacity())
+	})
+
+	prometheus.MustRegister(cachedMetricNames)
+	prometheus.MustRegister(metricNamesCacheCap)
+	prometheus.MustRegister(cachedSeriesSets)
+	prometheus.MustRegister(seriesSetCacheCap)
+	prometheus.MustRegister(cachedLabels)
+	prometheus.MustRegister(labelsCacheCap)
+
 	router := route.New()
 
 	promMetrics := api.Metrics{
@@ -207,6 +263,9 @@ func main() {
 		QueryBatchDuration:  queryBatchDuration,
 		FailedQueries:       failedQueries,
 		ReceivedQueries:     receivedQueries,
+		CachedMetricNames:   cachedMetricNames,
+		CachedSeriesSets:    cachedSeriesSets,
+		CachedLabels:        cachedLabels,
 	}
 	writeHandler := timeHandler(httpRequestDuration, "write", api.Write(client, elector, &promMetrics))
 	router.Post("/write", writeHandler)

pkg/api/metrics.go

@@ -18,4 +18,7 @@ type Metrics struct {
 	ReceivedQueries     prometheus.Counter
 	FailedQueries       prometheus.Counter
 	QueryBatchDuration  prometheus.Histogram
+	CachedMetricNames   prometheus.CounterFunc
+	CachedSeriesSets    prometheus.CounterFunc
+	CachedLabels        prometheus.CounterFunc
 }

pkg/api/query_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/prometheus/prometheus/promql/parser"
 	"github.com/prometheus/prometheus/storage"
 	"github.com/timescale/timescale-prometheus/pkg/log"
+	"github.com/timescale/timescale-prometheus/pkg/pgmodel"
 	"github.com/timescale/timescale-prometheus/pkg/prompb"
 	"github.com/timescale/timescale-prometheus/pkg/promql"
 	"github.com/timescale/timescale-prometheus/pkg/query"
@@ -42,6 +43,8 @@ type mockQuerier struct {
 	labelNamesErr       error
 }
 
+var _ pgmodel.Querier = (*mockQuerier)(nil)
+
 func (m mockQuerier) LabelNames() ([]string, error) {
 	return m.labelNames, m.labelNamesErr
 }
@@ -60,6 +63,14 @@ func (m mockQuerier) Select(int64, int64, bool, *storage.SelectHints, []parser.N
 	return &mockSeriesSet{}, nil, nil, m.selectErr
 }
 
+func (m mockQuerier) NumCachedLabels() int {
+	return 0
+}
+
+func (m mockQuerier) LabelsCacheCapacity() int {
+	return 0
+}
+
 func TestParseDuration(t *testing.T) {
 	testCase := []struct {
 		in          string

pkg/clockcache/cache.go

@@ -273,6 +273,12 @@ func (self *Cache) Len() int {
 	return len(self.storage)
 }
 
+func (self *Cache) Cap() int {
+	self.elementsLock.RLock()
+	defer self.elementsLock.RUnlock()
+	return cap(self.storage)
+}
+
 func (self *Cache) debugString() string {
 	self.elementsLock.RLock()
 	defer self.elementsLock.RUnlock()

pkg/pgclient/client.go

@@ -60,6 +60,7 @@ type Client struct {
 	queryable     *query.Queryable
 	cfg           *Config
 	ConnectionStr string
+	metricCache   *pgmodel.MetricNameCache
 }
 
 // NewClient creates a new PostgreSQL client
@@ -98,7 +99,14 @@ func NewClient(cfg *Config, readHist prometheus.ObserverVec) (*Client, error) {
 
 	queryable := query.NewQueryable(reader.GetQuerier())
 
-	return &Client{Connection: connectionPool, ingestor: ingestor, reader: reader, queryable: queryable, cfg: cfg}, nil
+	return &Client{
+		Connection:  connectionPool,
+		ingestor:    ingestor,
+		reader:      reader,
+		queryable:   queryable,
+		cfg:         cfg,
+		metricCache: cache,
+	}, nil
 }
 
 // GetConnectionStr returns a Postgres connection string
@@ -122,6 +130,30 @@ func (c *Client) Read(req *prompb.ReadRequest) (*prompb.ReadResponse, error) {
 	return c.reader.Read(req)
 }
 
+func (c *Client) NumCachedSeries() int {
+	return c.ingestor.NumCachedSeries()
+}
+
+func (c *Client) SeriesCacheCapacity() int {
+	return c.ingestor.SeriesCacheCapacity()
+}
+
+func (c *Client) NumCachedMetricNames() int {
+	return c.metricCache.NumElements()
+}
+
+func (c *Client) MetricNamesCacheCapacity() int {
+	return c.metricCache.Capacity()
+}
+
+func (c *Client) NumCachedLabels() int {
+	return c.reader.GetQuerier().NumCachedLabels()
+}
+
+func (c *Client) LabelsCacheCapacity() int {
+	return c.reader.GetQuerier().LabelsCacheCapacity()
+}
+
 // HealthCheck checks that the client is properly connected
 func (c *Client) HealthCheck() error {
 	return c.reader.HealthCheck()

pkg/pgmodel/cache.go

@@ -25,6 +25,8 @@ type seriesCache struct {
 	series *clockcache.Cache
 }
 
+var _ SeriesCache = (*seriesCache)(nil)
+
 func (b *seriesCache) GetSeries(lset Labels) (SeriesID, error) {
 	result, ok := b.series.Get(lset.String())
 	if !ok {
@@ -38,6 +40,14 @@ func (b *seriesCache) SetSeries(lset Labels, id SeriesID) error {
 	return nil
 }
 
+func (b *seriesCache) NumElements() int {
+	return b.series.Len()
+}
+
+func (b *seriesCache) Capacity() int {
+	return b.series.Cap()
+}
+
 // MetricNameCache stores and retrieves metric table names in a in-memory cache.
 type MetricNameCache struct {
 	Metrics *clockcache.Cache
@@ -64,3 +74,11 @@ func (m *MetricNameCache) Set(metric string, tableName string) error {
 	m.Metrics.Insert(metricBuilder.String(), tableBuilder.String())
 	return nil
 }
+
+func (m *MetricNameCache) NumElements() int {
+	return m.Metrics.Len()
+}
+
+func (m *MetricNameCache) Capacity() int {
+	return m.Metrics.Cap()
+}

pkg/pgmodel/ingestor.go

@@ -34,10 +34,12 @@ type seriesWithCallback struct {
 	Callback func(l Labels, id SeriesID) error
 }
 
-// Cache provides a caching mechanism for labels and series.
-type Cache interface {
+// SeriesCache provides a caching mechanism for labels and series.
+type SeriesCache interface {
 	GetSeries(lset Labels) (SeriesID, error)
 	SetSeries(lset Labels, id SeriesID) error
+	NumElements() int
+	Capacity() int
 }
 
 type samplesInfo struct {
@@ -48,7 +50,7 @@ type samplesInfo struct {
 
 // DBIngestor ingest the TimeSeries data into Timescale database.
 type DBIngestor struct {
-	cache Cache
+	cache SeriesCache
 	db    inserter
 }
 
@@ -105,6 +107,14 @@ func (i *DBIngestor) parseData(tts []prompb.TimeSeries, req *prompb.WriteRequest
 	return dataSamples, rows, nil
 }
 
+func (i *DBIngestor) NumCachedSeries() int {
+	return i.cache.NumElements()
+}
+
+func (i *DBIngestor) SeriesCacheCapacity() int {
+	return i.cache.Capacity()
+}
+
 // Close closes the ingestor
 func (i *DBIngestor) Close() {
 	i.db.Close()

pkg/pgmodel/ingestor_test.go

@@ -16,6 +16,8 @@ type mockCache struct {
 	setSeriesErr error
 }
 
+var _ SeriesCache = (*mockCache)(nil)
+
 func (m *mockCache) GetSeries(lset Labels) (SeriesID, error) {
 	if m.getSeriesErr != nil {
 		return 0, m.getSeriesErr
@@ -34,6 +36,14 @@ func (m *mockCache) SetSeries(lset Labels, id SeriesID) error {
 	return m.setSeriesErr
 }
 
+func (m *mockCache) NumElements() int {
+	return len(m.seriesCache)
+}
+
+func (m *mockCache) Capacity() int {
+	return len(m.seriesCache)
+}
+
 type mockInserter struct {
 	insertedSeries  map[string]SeriesID
 	insertedData    []map[string][]samplesInfo

pkg/pgmodel/pgx.go

@@ -216,7 +216,6 @@ 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) {
-	//TODO what size should we use?
 	cache := &MetricNameCache{clockcache.WithMax(DefaultMetricCacheSize)}
 	return NewPgxIngestorWithMetricCache(c, cache, &Cfg{})
 }
@@ -913,6 +912,8 @@ type pgxQuerier struct {
 	labels *clockcache.Cache
 }
 
+var _ Querier = (*pgxQuerier)(nil)
+
 // HealthCheck implements the healtchecker interface
 func (q *pgxQuerier) HealthCheck() error {
 	rows, err := q.conn.Query(context.Background(), "SELECT")
@@ -925,6 +926,14 @@ func (q *pgxQuerier) HealthCheck() error {
 	return nil
 }
 
+func (q *pgxQuerier) NumCachedLabels() int {
+	return q.labels.Len()
+}
+
+func (q *pgxQuerier) LabelsCacheCapacity() int {
+	return q.labels.Cap()
+}
+
 func (q *pgxQuerier) Select(mint int64, maxt int64, sortSeries bool, hints *storage.SelectHints, path []parser.Node, ms ...*labels.Matcher) (storage.SeriesSet, parser.Node, storage.Warnings, error) {
 	rows, topNode, err := q.getResultRows(mint, maxt, hints, path, ms)
 

pkg/pgmodel/querier.go

@@ -23,6 +23,8 @@ type Querier interface {
 	Select(mint int64, maxt int64, sortSeries bool, hints *storage.SelectHints, path []parser.Node, ms ...*labels.Matcher) (storage.SeriesSet, parser.Node, storage.Warnings, error)
 	LabelNames() ([]string, error)
 	LabelValues(labelName string) ([]string, error)
+	NumCachedLabels() int
+	LabelsCacheCapacity() int
 }
 
 //HealthChecker allows checking for proper operations.