TSDB: add bucket store metrics in querier

CHANGELOG.md

@@ -33,6 +33,7 @@ instructions below to upgrade your Postgres.
 * [ENHANCEMENT] Experimental TSDB: Open existing TSDB on startup to prevent ingester from becoming ready before it can accept writes. #1917
   * `--experimental.tsdb.max-tsdb-opening-concurrency-on-startup`
 * [ENHANCEMENT] Experimental TSDB: Added `cortex_ingester_shipper_dir_syncs_total`, `cortex_ingester_shipper_dir_sync_failures_total`, `cortex_ingester_shipper_uploads_total` and `cortex_ingester_shipper_upload_failures_total` metrics from TSDB shipper component. #1983
+* [ENHANCEMENT] Experimental TSDB: Querier now exports aggregate metrics from Thanos bucket store and in memory index cache (many metrics to list, but all have `cortex_querier_bucket_store_` or `cortex_querier_blocks_index_cache_` prefix). #1996 
 * [BUGFIX] Fixed unnecessary CAS operations done by the HA tracker when the jitter is enabled. #1861
 * [BUGFIX] Fixed #1904 ingesters getting stuck in a LEAVING state after coming up from an ungraceful exit. #1921
 * [BUGFIX] Reduce memory usage when ingester Push() errors. #1922

pkg/ingester/metrics.go

@@ -4,9 +4,7 @@ import (
 	"sync"
 
 	"github.com/cortexproject/cortex/pkg/util"
-	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/client_golang/prometheus"
-	dto "github.com/prometheus/client_model/go"
 )
 
 const (
@@ -122,10 +120,6 @@ type tsdbMetrics struct {
 	memSeriesCreatedTotal *prometheus.Desc
 	memSeriesRemovedTotal *prometheus.Desc
 
-	// These maps drive the collection output. Key = original metric name to group.
-	sumCountersGlobally map[string]*prometheus.Desc
-	sumCountersPerUser  map[string]*prometheus.Desc
-
 	regsMu sync.RWMutex                    // custom mutex for shipper registry, to avoid blocking main user state mutex on collection
 	regs   map[string]*prometheus.Registry // One prometheus registry per tenant
 }
@@ -155,18 +149,6 @@ func newTSDBMetrics(r prometheus.Registerer) *tsdbMetrics {
 		memSeriesRemovedTotal: prometheus.NewDesc(memSeriesRemovedTotalName, memSeriesRemovedTotalHelp, []string{"user"}, nil),
 	}
 
-	m.sumCountersGlobally = map[string]*prometheus.Desc{
-		"thanos_shipper_dir_syncs_total":         m.dirSyncs,
-		"thanos_shipper_dir_sync_failures_total": m.dirSyncFailures,
-		"thanos_shipper_uploads_total":           m.uploads,
-		"thanos_shipper_upload_failures_total":   m.uploadFailures,
-	}
-
-	m.sumCountersPerUser = map[string]*prometheus.Desc{
-		"prometheus_tsdb_head_series_created_total": m.memSeriesCreatedTotal,
-		"prometheus_tsdb_head_series_removed_total": m.memSeriesRemovedTotal,
-	}
-
 	if r != nil {
 		r.MustRegister(m)
 	}
@@ -183,30 +165,16 @@ func (sm *tsdbMetrics) Describe(out chan<- *prometheus.Desc) {
 }
 
 func (sm *tsdbMetrics) Collect(out chan<- prometheus.Metric) {
-	regs := sm.registries()
-	data := gatheredMetricsPerUser{}
-
-	for userID, r := range regs {
-		m, err := r.Gather()
-		if err != nil {
-			level.Warn(util.Logger).Log("msg", "failed to gather metrics from TSDB shipper", "user", userID, "err", err)
-			continue
-		}
-
-		data.addGatheredDataForUser(userID, m)
-	}
+	data := util.BuildMetricFamiliesPerUserFromUserRegistries(sm.registries())
 
 	// OK, we have it all. Let's build results.
-	for metric, desc := range sm.sumCountersGlobally {
-		out <- prometheus.MustNewConstMetric(desc, prometheus.CounterValue, data.sumCountersAcrossAllUsers(metric))
-	}
+	data.SendSumOfCounters(out, sm.dirSyncs, "thanos_shipper_dir_syncs_total")
+	data.SendSumOfCounters(out, sm.dirSyncFailures, "thanos_shipper_dir_sync_failures_total")
+	data.SendSumOfCounters(out, sm.uploads, "thanos_shipper_uploads_total")
+	data.SendSumOfCounters(out, sm.uploadFailures, "thanos_shipper_upload_failures_total")
 
-	for metric, desc := range sm.sumCountersPerUser {
-		userValues := data.sumCountersPerUser(metric)
-		for user, val := range userValues {
-			out <- prometheus.MustNewConstMetric(desc, prometheus.CounterValue, val, user)
-		}
-	}
+	data.SendSumOfCountersPerUser(out, sm.memSeriesCreatedTotal, "prometheus_tsdb_head_series_created_total")
+	data.SendSumOfCountersPerUser(out, sm.memSeriesRemovedTotal, "prometheus_tsdb_head_series_removed_total")
 }
 
 // make a copy of the map, so that metrics can be gathered while the new registry is being added.
@@ -226,56 +194,3 @@ func (sm *tsdbMetrics) setRegistryForUser(userID string, registry *prometheus.Re
 	sm.regs[userID] = registry
 	sm.regsMu.Unlock()
 }
-
-func sumCounters(mfs []*dto.MetricFamily) float64 {
-	result := float64(0)
-	for _, mf := range mfs {
-		if mf.Type == nil || *mf.Type != dto.MetricType_COUNTER {
-			continue
-		}
-
-		for _, m := range mf.Metric {
-			if m == nil || m.Counter == nil || m.Counter.Value == nil {
-				continue
-			}
-
-			result += *m.Counter.Value
-		}
-	}
-	return result
-}
-
-// first key = userID, second key = metric name. Value = slice of gathered values with the same metric name.
-type gatheredMetricsPerUser map[string]map[string][]*dto.MetricFamily
-
-func (d gatheredMetricsPerUser) addGatheredDataForUser(userID string, metrics []*dto.MetricFamily) {
-	// first, create new map which maps metric names to a slice of MetricFamily instances.
-	// That makes it easier to do searches later.
-	perMetricName := map[string][]*dto.MetricFamily{}
-
-	for _, m := range metrics {
-		if m.Name == nil {
-			continue
-		}
-		perMetricName[*m.Name] = append(perMetricName[*m.Name], m)
-	}
-
-	d[userID] = perMetricName
-}
-
-func (d gatheredMetricsPerUser) sumCountersAcrossAllUsers(counter string) float64 {
-	result := float64(0)
-	for _, perMetric := range d {
-		result += sumCounters(perMetric[counter])
-	}
-	return result
-}
-
-func (d gatheredMetricsPerUser) sumCountersPerUser(counter string) map[string]float64 {
-	result := map[string]float64{}
-	for user, perMetric := range d {
-		v := sumCounters(perMetric[counter])
-		result[user] = v
-	}
-	return result
-}

pkg/ingester/metrics_test.go

@@ -10,23 +10,14 @@ import (
 )
 
 func TestTSDBMetrics(t *testing.T) {
-	mainReg := prometheus.NewRegistry()
+	mainReg := prometheus.NewPedanticRegistry()
 
 	tsdbMetrics := newTSDBMetrics(mainReg)
 
 	tsdbMetrics.setRegistryForUser("user1", populateTSDBMetrics(12345))
 	tsdbMetrics.setRegistryForUser("user2", populateTSDBMetrics(85787))
 	tsdbMetrics.setRegistryForUser("user3", populateTSDBMetrics(999))
 
-	metricNames := []string{
-		"cortex_ingester_shipper_dir_syncs_total",
-		"cortex_ingester_shipper_dir_sync_failures_total",
-		"cortex_ingester_shipper_uploads_total",
-		"cortex_ingester_shipper_upload_failures_total",
-		"cortex_ingester_memory_series_created_total",
-		"cortex_ingester_memory_series_removed_total",
-	}
-
 	err := testutil.GatherAndCompare(mainReg, bytes.NewBufferString(`
 			# HELP cortex_ingester_shipper_dir_syncs_total TSDB: Total dir sync attempts
 			# TYPE cortex_ingester_shipper_dir_syncs_total counter
@@ -61,7 +52,7 @@ func TestTSDBMetrics(t *testing.T) {
 			cortex_ingester_memory_series_removed_total{user="user1"} 74070
 			cortex_ingester_memory_series_removed_total{user="user2"} 514722
 			cortex_ingester_memory_series_removed_total{user="user3"} 5994
-	`), metricNames...)
+	`))
 	require.NoError(t, err)
 }
 

pkg/querier/block.go

@@ -39,14 +39,16 @@ func NewBlockQuerier(cfg tsdb.Config, logLevel logging.Level, r prometheus.Regis
 		}),
 	}
 
-	r.MustRegister(b.syncTimes)
-
 	us, err := NewUserStore(cfg, logLevel, util.Logger)
 	if err != nil {
 		return nil, err
 	}
 	b.us = us
 
+	if r != nil {
+		r.MustRegister(b.syncTimes, us.tsdbMetrics)
+	}
+
 	level.Info(util.Logger).Log("msg", "synchronizing TSDB blocks for all users")
 	if err := us.InitialSync(context.Background()); err != nil {
 		level.Warn(util.Logger).Log("msg", "failed to synchronize TSDB blocks", "err", err)

pkg/querier/block_store.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/thanos/pkg/model"
 	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/store"
@@ -24,12 +25,13 @@ import (
 
 // UserStore is a multi-tenant version of Thanos BucketStore
 type UserStore struct {
-	logger   log.Logger
-	cfg      tsdb.Config
-	bucket   objstore.BucketReader
-	stores   map[string]*store.BucketStore
-	client   storepb.StoreClient
-	logLevel logging.Level
+	logger      log.Logger
+	cfg         tsdb.Config
+	bucket      objstore.BucketReader
+	stores      map[string]*store.BucketStore
+	client      storepb.StoreClient
+	logLevel    logging.Level
+	tsdbMetrics *tsdbBucketStoreMetrics
 }
 
 // NewUserStore returns a new UserStore
@@ -40,11 +42,12 @@ func NewUserStore(cfg tsdb.Config, logLevel logging.Level, logger log.Logger) (*
 	}
 
 	u := &UserStore{
-		logger:   logger,
-		cfg:      cfg,
-		bucket:   bkt,
-		stores:   make(map[string]*store.BucketStore),
-		logLevel: logLevel,
+		logger:      logger,
+		cfg:         cfg,
+		bucket:      bkt,
+		stores:      map[string]*store.BucketStore{},
+		logLevel:    logLevel,
+		tsdbMetrics: newTSDBBucketStoreMetrics(),
 	}
 
 	serv := grpc.NewServer()
@@ -115,9 +118,11 @@ func (u *UserStore) syncUserStores(ctx context.Context, f func(context.Context,
 				Bucket: bkt,
 			}
 
+			reg := prometheus.NewRegistry()
+
 			indexCacheSizeBytes := u.cfg.BucketStore.IndexCacheSizeBytes
 			maxItemSizeBytes := indexCacheSizeBytes / 2
-			indexCache, err := storecache.NewInMemoryIndexCache(u.logger, nil, storecache.Opts{
+			indexCache, err := storecache.NewInMemoryIndexCache(u.logger, reg, storecache.Opts{
 				MaxSizeBytes:     indexCacheSizeBytes,
 				MaxItemSizeBytes: maxItemSizeBytes,
 			})
@@ -126,7 +131,7 @@ func (u *UserStore) syncUserStores(ctx context.Context, f func(context.Context,
 			}
 			bs, err = store.NewBucketStore(
 				u.logger,
-				nil,
+				reg,
 				userBkt,
 				filepath.Join(u.cfg.BucketStore.SyncDir, user),
 				indexCache,
@@ -147,6 +152,7 @@ func (u *UserStore) syncUserStores(ctx context.Context, f func(context.Context,
 			}
 
 			u.stores[user] = bs
+			u.tsdbMetrics.addUserRegistry(user, reg)
 		}
 
 		wg.Add(1)