Decouple blocks meta fetcher metrics mapping from bucket store metrics (#2375)

* De-couple blocks meta fetcher metrics mapping from bucket store metrics

Signed-off-by: Marco Pracucci <[email protected]>

* Added PR number to CHANGELOG

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

CHANGELOG.md

@@ -73,6 +73,11 @@
 * [CHANGE] Experimental TSDB: Added `-compactor.deletion-delay`, which is time before a block marked for deletion is deleted from bucket. If not 0, blocks will be marked for deletion and compactor component will delete blocks marked for deletion from the bucket. If delete-delay is 0, blocks will be deleted straight away. Note that deleting blocks immediately can cause query failures, if store gateway / querier still has the block loaded, or compactor is ignoring the deletion because it's compacting the block at the same time. Default value is 48h. #2335
 * [CHANGE] Experimental TSDB: Added `-experimental.tsdb.bucket-store.index-cache.postings-compression-enabled`, to set duration after which the blocks marked for deletion will be filtered out while fetching blocks used for querying. This option allows querier to ignore blocks that are marked for deletion with some delay. This ensures store can still serve blocks that are meant to be deleted but do not have a replacement yet. Default is 24h, half of the default value for `-compactor.deletion-delay`. #2335
 * [CHANGE] Experimental TSDB: Added `-experimental.tsdb.bucket-store.index-cache.memcached.max-item-size` to control maximum size of item that is stored to memcached. Defaults to 1 MiB. #2335
+* [CHANGE] Experimental TSDB: renamed blocks meta fetcher metrics: #2375
+  * `cortex_querier_bucket_store_blocks_meta_syncs_total` > `cortex_querier_blocks_meta_syncs_total`
+  * `cortex_querier_bucket_store_blocks_meta_sync_failures_total` > `cortex_querier_blocks_meta_sync_failures_total`
+  * `cortex_querier_bucket_store_blocks_meta_sync_duration_seconds` > `cortex_querier_blocks_meta_sync_duration_seconds`
+  * `cortex_querier_bucket_store_blocks_meta_sync_consistency_delay_seconds` > `cortex_querier_blocks_meta_sync_consistency_delay_seconds`
 * [FEATURE] Added experimental storage API to the ruler service that is enabled when the `-experimental.ruler.enable-api` is set to true #2269
   * `-ruler.storage.type` flag now allows `s3`,`gcs`, and `azure` values
   * `-ruler.storage.(s3|gcs|azure)` flags exist to allow the configuration of object clients set for rule storage

development/tsdb-blocks-storage-s3/config/cortex.yaml

@@ -37,6 +37,7 @@ tsdb:
 
   bucket_store:
     sync_dir: /tmp/cortex-tsdb-querier
+    consistency_delay: 5s
 
     index_cache:
       backend: memcached
@@ -52,7 +53,7 @@ tsdb:
 
 ruler:
   enable_api: true
-  storeconfig:
+  storage:
     type: configdb
     configdb:
       configs_api_url: http://configstore:80/
@@ -72,4 +73,4 @@ compactor:
         host: consul:8500
 
 frontend_worker:
-  address: "query-frontend:9007"
+  frontend_address: "query-frontend:9007"

pkg/querier/block_store.go

@@ -37,6 +37,7 @@ type UserStore struct {
 	bucket             objstore.Bucket
 	logLevel           logging.Level
 	bucketStoreMetrics *tsdbBucketStoreMetrics
+	metaFetcherMetrics *metaFetcherMetrics
 	indexCacheMetrics  prometheus.Collector
 
 	// Index cache shared across all tenants.
@@ -61,6 +62,7 @@ func NewUserStore(cfg tsdb.Config, bucketClient objstore.Bucket, logLevel loggin
 		stores:             map[string]*store.BucketStore{},
 		logLevel:           logLevel,
 		bucketStoreMetrics: newTSDBBucketStoreMetrics(),
+		metaFetcherMetrics: newMetaFetcherMetrics(),
 		indexCacheMetrics:  tsdb.MustNewIndexCacheMetrics(cfg.BucketStore.IndexCache.Backend, indexCacheRegistry),
 		syncTimes: promauto.With(registerer).NewHistogram(prometheus.HistogramOpts{
 			Name:    "cortex_querier_blocks_sync_seconds",
@@ -76,7 +78,7 @@ func NewUserStore(cfg tsdb.Config, bucketClient objstore.Bucket, logLevel loggin
 	}
 
 	if registerer != nil {
-		registerer.MustRegister(u.bucketStoreMetrics, u.indexCacheMetrics)
+		registerer.MustRegister(u.bucketStoreMetrics, u.metaFetcherMetrics, u.indexCacheMetrics)
 	}
 
 	u.Service = services.NewBasicService(u.starting, u.syncStoresLoop, nil)
@@ -259,16 +261,16 @@ func (u *UserStore) getOrCreateStore(userID string) (*store.BucketStore, error)
 
 	userBkt := tsdb.NewUserBucketClient(userID, u.bucket)
 
-	reg := prometheus.NewRegistry()
+	fetcherReg := prometheus.NewRegistry()
 	fetcher, err := block.NewMetaFetcher(
 		userLogger,
 		u.cfg.BucketStore.MetaSyncConcurrency,
 		userBkt,
 		filepath.Join(u.cfg.BucketStore.SyncDir, userID), // The fetcher stores cached metas in the "meta-syncer/" sub directory
-		reg,
+		fetcherReg,
 		[]block.MetadataFilter{
 			// List of filters to apply (order matters).
-			block.NewConsistencyDelayMetaFilter(userLogger, u.cfg.BucketStore.ConsistencyDelay, reg),
+			block.NewConsistencyDelayMetaFilter(userLogger, u.cfg.BucketStore.ConsistencyDelay, fetcherReg),
 			block.NewIgnoreDeletionMarkFilter(userLogger, userBkt, u.cfg.BucketStore.IgnoreDeletionMarksDelay),
 			// Filters out duplicate blocks that can be formed from two or more overlapping
 			// blocks that fully submatches the source blocks of the older blocks.
@@ -279,9 +281,10 @@ func (u *UserStore) getOrCreateStore(userID string) (*store.BucketStore, error)
 		return nil, err
 	}
 
+	bucketStoreReg := prometheus.NewRegistry()
 	bs, err = store.NewBucketStore(
 		userLogger,
-		reg,
+		bucketStoreReg,
 		userBkt,
 		fetcher,
 		filepath.Join(u.cfg.BucketStore.SyncDir, userID),
@@ -301,7 +304,8 @@ func (u *UserStore) getOrCreateStore(userID string) (*store.BucketStore, error)
 	}
 
 	u.stores[userID] = bs
-	u.bucketStoreMetrics.addUserRegistry(userID, reg)
+	u.metaFetcherMetrics.addUserRegistry(userID, fetcherReg)
+	u.bucketStoreMetrics.addUserRegistry(userID, bucketStoreReg)
 
 	return bs, nil
 }

pkg/querier/block_store_metrics.go

@@ -16,34 +16,26 @@ type tsdbBucketStoreMetrics struct {
 	regs   map[string]*prometheus.Registry
 
 	// exported metrics, gathered from Thanos BucketStore
-	blockLoads               *prometheus.Desc
-	blockLoadFailures        *prometheus.Desc
-	blockDrops               *prometheus.Desc
-	blockDropFailures        *prometheus.Desc
-	blocksLoaded             *prometheus.Desc
-	seriesDataTouched        *prometheus.Desc
-	seriesDataFetched        *prometheus.Desc
-	seriesDataSizeTouched    *prometheus.Desc
-	seriesDataSizeFetched    *prometheus.Desc
-	seriesBlocksQueried      *prometheus.Desc
-	seriesGetAllDuration     *prometheus.Desc
-	seriesMergeDuration      *prometheus.Desc
-	seriesRefetches          *prometheus.Desc
-	resultSeriesCount        *prometheus.Desc
-	metaSyncs                *prometheus.Desc
-	metaSyncFailures         *prometheus.Desc
-	metaSyncDuration         *prometheus.Desc
-	metaSyncConsistencyDelay *prometheus.Desc
+	blockLoads            *prometheus.Desc
+	blockLoadFailures     *prometheus.Desc
+	blockDrops            *prometheus.Desc
+	blockDropFailures     *prometheus.Desc
+	blocksLoaded          *prometheus.Desc
+	seriesDataTouched     *prometheus.Desc
+	seriesDataFetched     *prometheus.Desc
+	seriesDataSizeTouched *prometheus.Desc
+	seriesDataSizeFetched *prometheus.Desc
+	seriesBlocksQueried   *prometheus.Desc
+	seriesGetAllDuration  *prometheus.Desc
+	seriesMergeDuration   *prometheus.Desc
+	seriesRefetches       *prometheus.Desc
+	resultSeriesCount     *prometheus.Desc
 
 	cachedPostingsCompressions           *prometheus.Desc
 	cachedPostingsCompressionErrors      *prometheus.Desc
 	cachedPostingsCompressionTimeSeconds *prometheus.Desc
 	cachedPostingsOriginalSizeBytes      *prometheus.Desc
 	cachedPostingsCompressedSizeBytes    *prometheus.Desc
-
-	// Ignored:
-	// blocks_meta_synced
-	// blocks_meta_modified
 }
 
 func newTSDBBucketStoreMetrics() *tsdbBucketStoreMetrics {
@@ -107,22 +99,6 @@ func newTSDBBucketStoreMetrics() *tsdbBucketStoreMetrics {
 			"cortex_querier_bucket_store_series_result_series",
 			"TSDB: Number of series observed in the final result of a query.",
 			nil, nil),
-		metaSyncs: prometheus.NewDesc(
-			"cortex_querier_bucket_store_blocks_meta_syncs_total",
-			"TSDB: Total blocks metadata synchronization attempts",
-			nil, nil),
-		metaSyncFailures: prometheus.NewDesc(
-			"cortex_querier_bucket_store_blocks_meta_sync_failures_total",
-			"TSDB: Total blocks metadata synchronization failures",
-			nil, nil),
-		metaSyncDuration: prometheus.NewDesc(
-			"cortex_querier_bucket_store_blocks_meta_sync_duration_seconds",
-			"TSDB: Duration of the blocks metadata synchronization in seconds",
-			nil, nil),
-		metaSyncConsistencyDelay: prometheus.NewDesc(
-			"cortex_querier_bucket_store_blocks_meta_sync_consistency_delay_seconds",
-			"TSDB: Configured consistency delay in seconds.",
-			nil, nil),
 
 		cachedPostingsCompressions: prometheus.NewDesc(
 			"cortex_querier_bucket_store_cached_postings_compressions_total",
@@ -181,11 +157,6 @@ func (m *tsdbBucketStoreMetrics) Describe(out chan<- *prometheus.Desc) {
 	out <- m.seriesRefetches
 	out <- m.resultSeriesCount
 
-	out <- m.metaSyncs
-	out <- m.metaSyncFailures
-	out <- m.metaSyncDuration
-	out <- m.metaSyncConsistencyDelay
-
 	out <- m.cachedPostingsCompressions
 	out <- m.cachedPostingsCompressionErrors
 	out <- m.cachedPostingsCompressionTimeSeconds
@@ -214,11 +185,6 @@ func (m *tsdbBucketStoreMetrics) Collect(out chan<- prometheus.Metric) {
 	data.SendSumOfCounters(out, m.seriesRefetches, "thanos_bucket_store_series_refetches_total")
 	data.SendSumOfSummaries(out, m.resultSeriesCount, "thanos_bucket_store_series_result_series")
 
-	data.SendSumOfCounters(out, m.metaSyncs, "blocks_meta_syncs_total")
-	data.SendSumOfCounters(out, m.metaSyncFailures, "blocks_meta_sync_failures_total")
-	data.SendSumOfHistograms(out, m.metaSyncDuration, "blocks_meta_sync_duration_seconds")
-	data.SendMaxOfGauges(out, m.metaSyncConsistencyDelay, "consistency_delay_seconds")
-
 	data.SendSumOfCountersWithLabels(out, m.cachedPostingsCompressions, "thanos_bucket_store_cached_postings_compressions_total", "op")
 	data.SendSumOfCountersWithLabels(out, m.cachedPostingsCompressionErrors, "thanos_bucket_store_cached_postings_compression_errors_total", "op")
 	data.SendSumOfCountersWithLabels(out, m.cachedPostingsCompressionTimeSeconds, "thanos_bucket_store_cached_postings_compression_time_seconds", "op")

pkg/querier/blocks_meta_fetcher_metrics.go

@@ -0,0 +1,85 @@
+package querier
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/cortexproject/cortex/pkg/util"
+)
+
+// This struct aggregates metrics exported by Thanos MetaFetcher
+// and re-exports those aggregates as Cortex metrics.
+type metaFetcherMetrics struct {
+	// Maps userID -> registry
+	regsMu sync.Mutex
+	regs   map[string]*prometheus.Registry
+
+	// Exported metrics, gathered from Thanos MetaFetcher
+	syncs                *prometheus.Desc
+	syncFailures         *prometheus.Desc
+	syncDuration         *prometheus.Desc
+	syncConsistencyDelay *prometheus.Desc
+
+	// Ignored:
+	// blocks_meta_synced
+	// blocks_meta_modified
+}
+
+func newMetaFetcherMetrics() *metaFetcherMetrics {
+	return &metaFetcherMetrics{
+		regs: map[string]*prometheus.Registry{},
+
+		syncs: prometheus.NewDesc(
+			"cortex_querier_blocks_meta_syncs_total",
+			"Total blocks metadata synchronization attempts",
+			nil, nil),
+		syncFailures: prometheus.NewDesc(
+			"cortex_querier_blocks_meta_sync_failures_total",
+			"Total blocks metadata synchronization failures",
+			nil, nil),
+		syncDuration: prometheus.NewDesc(
+			"cortex_querier_blocks_meta_sync_duration_seconds",
+			"Duration of the blocks metadata synchronization in seconds",
+			nil, nil),
+		syncConsistencyDelay: prometheus.NewDesc(
+			"cortex_querier_blocks_meta_sync_consistency_delay_seconds",
+			"Configured consistency delay in seconds.",
+			nil, nil),
+	}
+}
+
+func (m *metaFetcherMetrics) addUserRegistry(user string, reg *prometheus.Registry) {
+	m.regsMu.Lock()
+	m.regs[user] = reg
+	m.regsMu.Unlock()
+}
+
+func (m *metaFetcherMetrics) registries() map[string]*prometheus.Registry {
+	regs := map[string]*prometheus.Registry{}
+
+	m.regsMu.Lock()
+	defer m.regsMu.Unlock()
+	for uid, r := range m.regs {
+		regs[uid] = r
+	}
+
+	return regs
+}
+
+func (m *metaFetcherMetrics) Describe(out chan<- *prometheus.Desc) {
+
+	out <- m.syncs
+	out <- m.syncFailures
+	out <- m.syncDuration
+	out <- m.syncConsistencyDelay
+}
+
+func (m *metaFetcherMetrics) Collect(out chan<- prometheus.Metric) {
+	data := util.BuildMetricFamiliesPerUserFromUserRegistries(m.registries())
+
+	data.SendSumOfCounters(out, m.syncs, "blocks_meta_syncs_total")
+	data.SendSumOfCounters(out, m.syncFailures, "blocks_meta_sync_failures_total")
+	data.SendSumOfHistograms(out, m.syncDuration, "blocks_meta_sync_duration_seconds")
+	data.SendMaxOfGauges(out, m.syncConsistencyDelay, "consistency_delay_seconds")
+}

pkg/querier/blocks_meta_fetcher_metrics_test.go

@@ -0,0 +1,95 @@
+package querier
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetaFetcherMetrics(t *testing.T) {
+	mainReg := prometheus.NewPedanticRegistry()
+
+	metrics := newMetaFetcherMetrics()
+	mainReg.MustRegister(metrics)
+
+	metrics.addUserRegistry("user1", populateMetaFetcherMetrics(3))
+	metrics.addUserRegistry("user2", populateMetaFetcherMetrics(5))
+	metrics.addUserRegistry("user3", populateMetaFetcherMetrics(7))
+
+	//noinspection ALL
+	err := testutil.GatherAndCompare(mainReg, bytes.NewBufferString(`
+		# HELP cortex_querier_blocks_meta_sync_duration_seconds Duration of the blocks metadata synchronization in seconds
+		# TYPE cortex_querier_blocks_meta_sync_duration_seconds histogram
+		cortex_querier_blocks_meta_sync_duration_seconds_bucket{le="0.01"} 0
+		cortex_querier_blocks_meta_sync_duration_seconds_bucket{le="1"} 0
+		cortex_querier_blocks_meta_sync_duration_seconds_bucket{le="10"} 3
+		cortex_querier_blocks_meta_sync_duration_seconds_bucket{le="100"} 3
+		cortex_querier_blocks_meta_sync_duration_seconds_bucket{le="1000"} 3
+		cortex_querier_blocks_meta_sync_duration_seconds_bucket{le="+Inf"} 3
+		cortex_querier_blocks_meta_sync_duration_seconds_sum 9
+		cortex_querier_blocks_meta_sync_duration_seconds_count 3
+
+		# HELP cortex_querier_blocks_meta_sync_failures_total Total blocks metadata synchronization failures
+		# TYPE cortex_querier_blocks_meta_sync_failures_total counter
+		cortex_querier_blocks_meta_sync_failures_total 30
+
+		# HELP cortex_querier_blocks_meta_syncs_total Total blocks metadata synchronization attempts
+		# TYPE cortex_querier_blocks_meta_syncs_total counter
+		cortex_querier_blocks_meta_syncs_total 15
+
+		# HELP cortex_querier_blocks_meta_sync_consistency_delay_seconds Configured consistency delay in seconds.
+		# TYPE cortex_querier_blocks_meta_sync_consistency_delay_seconds gauge
+		cortex_querier_blocks_meta_sync_consistency_delay_seconds 300
+`))
+	require.NoError(t, err)
+}
+
+func populateMetaFetcherMetrics(base float64) *prometheus.Registry {
+	reg := prometheus.NewRegistry()
+	m := newMetaFetcherMetricsMock(reg)
+
+	m.syncs.Add(base * 1)
+	m.syncFailures.Add(base * 2)
+	m.syncDuration.Observe(3)
+	m.syncConsistencyDelay.Set(300)
+
+	return reg
+}
+
+type metaFetcherMetricsMock struct {
+	syncs                prometheus.Counter
+	syncFailures         prometheus.Counter
+	syncDuration         prometheus.Histogram
+	syncConsistencyDelay prometheus.Gauge
+}
+
+func newMetaFetcherMetricsMock(reg prometheus.Registerer) *metaFetcherMetricsMock {
+	var m metaFetcherMetricsMock
+
+	m.syncs = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Subsystem: "blocks_meta",
+		Name:      "syncs_total",
+		Help:      "Total blocks metadata synchronization attempts",
+	})
+	m.syncFailures = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Subsystem: "blocks_meta",
+		Name:      "sync_failures_total",
+		Help:      "Total blocks metadata synchronization failures",
+	})
+	m.syncDuration = promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
+		Subsystem: "blocks_meta",
+		Name:      "sync_duration_seconds",
+		Help:      "Duration of the blocks metadata synchronization in seconds",
+		Buckets:   []float64{0.01, 1, 10, 100, 1000},
+	})
+	m.syncConsistencyDelay = promauto.With(reg).NewGauge(prometheus.GaugeOpts{
+		Name: "consistency_delay_seconds",
+		Help: "Configured consistency delay in seconds.",
+	})
+
+	return &m
+}