Reset inverted labels cache on epoch change

The database cache epoch advances when series are deleted from the
database. When the connector detects a cache epoch change, it
invalidates the series cache, as it may contain values which reference
the deleted rows in the series table.

During the database series deletion, unused labels may also be deleted.
This means that we must also invalidate the inverted label cache, as it
could also reference deleted rows in the `_prom_catalog.label` table.

Author: JamesGuthrie

CHANGELOG.md

@@ -25,6 +25,7 @@ We use the following categories for changes:
 
 ### Fixed
 - Make Jaeger Event queryable using name and tags [#1553]
+- Reset inverted labels cache on epoch change [#1561]
 
 ## [0.13.0] - 2022-07-20
 

pkg/pgmodel/Readme.md

@@ -15,7 +15,7 @@ TODO (@ante do you want to write this?)
 
 ## Write Path
 
-A `WriteRequest` is made out of a `[]TimeSeries` each of which contains a set of lables, each a `{string, string}`, and a set of samples, each a `{Timestamp, Value}`. Logically, the write path deals with this write request multiple stages. In Go notation, this can be thought of as follows
+A `WriteRequest` is made out of a `[]TimeSeries` each of which contains a set of labels, each a `{string, string}`, and a set of samples, each a `{Timestamp, Value}`. Logically, the write path deals with this write request multiple stages. In Go notation, this can be thought of as follows
 
 ```go
 type WriteRequest = []TimeSeries

pkg/pgmodel/cache/inverted_labels_cache.go

@@ -42,7 +42,7 @@ type InvertedLabelsCache struct {
 }
 
 // Cache is thread-safe
-func NewInvertedLablesCache(size uint64) (*InvertedLabelsCache, error) {
+func NewInvertedLabelsCache(size uint64) (*InvertedLabelsCache, error) {
 	if size <= 0 {
 		return nil, fmt.Errorf("labels cache size must be > 0")
 	}
@@ -62,3 +62,7 @@ func (c *InvertedLabelsCache) Put(key LabelKey, val LabelInfo) bool {
 	_, added := c.cache.Insert(key, val, uint64(key.len())+uint64(val.len())+17)
 	return added
 }
+
+func (c *InvertedLabelsCache) Reset() {
+	c.cache.Reset()
+}

pkg/pgmodel/ingestor/dispatcher.go

@@ -38,6 +38,7 @@ type pgxDispatcher struct {
 	conn                   pgxconn.PgxConn
 	metricTableNames       cache.MetricCache
 	scache                 cache.SeriesCache
+	invertedLabelsCache    *cache.InvertedLabelsCache
 	exemplarKeyPosCache    cache.PositionCache
 	batchers               sync.Map
 	completeMetricCreation chan struct{}
@@ -76,7 +77,7 @@ func newPgxDispatcher(conn pgxconn.PgxConn, mCache cache.MetricCache, scache cac
 	}
 
 	labelArrayOID := model.GetCustomTypeOID(model.LabelArray)
-	labelsCache, err := cache.NewInvertedLablesCache(cfg.InvertedLabelsCacheSize)
+	labelsCache, err := cache.NewInvertedLabelsCache(cfg.InvertedLabelsCacheSize)
 	if err != nil {
 		return nil, err
 	}
@@ -91,6 +92,7 @@ func newPgxDispatcher(conn pgxconn.PgxConn, mCache cache.MetricCache, scache cac
 		conn:                   conn,
 		metricTableNames:       mCache,
 		scache:                 scache,
+		invertedLabelsCache:    labelsCache,
 		exemplarKeyPosCache:    eCache,
 		completeMetricCreation: make(chan struct{}, 1),
 		asyncAcks:              cfg.AsyncAcks,
@@ -156,10 +158,14 @@ func (p *pgxDispatcher) refreshSeriesEpoch(existingEpoch model.SeriesEpoch) (mod
 	if err != nil {
 		// Trash the cache just in case an epoch change occurred, seems safer
 		p.scache.Reset()
+		// Also trash the inverted labels cache, which can also be invalidated when the series cache is
+		p.invertedLabelsCache.Reset()
 		return model.InvalidSeriesEpoch, err
 	}
 	if existingEpoch == model.InvalidSeriesEpoch || dbEpoch != existingEpoch {
 		p.scache.Reset()
+		// If the series cache needs to be invalidated, so does the inverted labels cache
+		p.invertedLabelsCache.Reset()
 	}
 	return dbEpoch, nil
 }

pkg/pgmodel/ingestor/ingestor_sql_test.go

@@ -276,7 +276,7 @@ func TestPGXInserterInsertSeries(t *testing.T) {
 			mock := model.NewSqlRecorder(c.sqlQueries, t)
 			scache := cache.NewSeriesCache(cache.DefaultConfig, nil)
 			scache.Reset()
-			lCache, _ := cache.NewInvertedLablesCache(10)
+			lCache, _ := cache.NewInvertedLabelsCache(10)
 			sw := NewSeriesWriter(mock, 0, lCache)
 
 			lsi := make([]model.Insertable, 0)
@@ -393,6 +393,21 @@ func TestPGXInserterCacheReset(t *testing.T) {
 		},
 		{Sql: "COMMIT;"},
 		{Sql: "BEGIN;"},
+		{
+			Sql: "SELECT * FROM _prom_catalog.get_or_create_label_ids($1, $2, $3, $4)",
+			Args: []interface{}{
+				"metric_1",
+				tableName,
+				[]string{"__name__", "name_1", "name_1"},
+				[]string{"metric_1", "value_1", "value_2"},
+			},
+			Results: model.RowResults{
+				{[]int32{1, 2, 2}, []int32{1, 2, 3}, []string{"__name__", "name_1", "name_1"}, []string{"metric_1", "value_1", "value_2"}},
+			},
+			Err: error(nil),
+		},
+		{Sql: "COMMIT;"},
+		{Sql: "BEGIN;"},
 		{
 			Sql: seriesInsertSQL,
 			Args: []interface{}{
@@ -419,11 +434,12 @@ func TestPGXInserterCacheReset(t *testing.T) {
 
 	mock := model.NewSqlRecorder(sqlQueries, t)
 	scache := cache.NewSeriesCache(cache.DefaultConfig, nil)
-	lcache, _ := cache.NewInvertedLablesCache(10)
+	lcache, _ := cache.NewInvertedLabelsCache(10)
 	sw := NewSeriesWriter(mock, 0, lcache)
 	inserter := pgxDispatcher{
-		conn:   mock,
-		scache: scache,
+		conn:                mock,
+		scache:              scache,
+		invertedLabelsCache: lcache,
 	}
 
 	makeSamples := func(series []labels.Labels) []model.Insertable {
@@ -460,7 +476,7 @@ func TestPGXInserterCacheReset(t *testing.T) {
 		}
 	}
 
-	// refreshing during the same epoch givesthe same IDs without checking the DB
+	// refreshing during the same epoch gives the same IDs without checking the DB
 	_, err = inserter.refreshSeriesEpoch(1)
 	require.NoError(t, err)