Skip to content

Commit cef7fe7

Browse files
SungJin1212SungJin1212
committed
rollback bucket store metrics
Signed-off-by: SungJin1212 <[email protected]>
1 parent c54bc5d commit cef7fe7

File tree

5 files changed

+63
-116
lines changed

5 files changed

+63
-116
lines changed

‎pkg/storegateway/bucket_store_metrics.go‎

Lines changed: 1 addition & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -2,41 +2,10 @@ package storegateway
22

33
import (
44
"github.com/prometheus/client_golang/prometheus"
5-
"github.com/prometheus/client_golang/prometheus/promauto"
65

76
"github.com/cortexproject/cortex/pkg/util"
87
)
98

10-
// CortexBucketStoreMetrics common metrics in thanos and parquet block stores (in future)
11-
type CortexBucketStoreMetrics struct {
12-
syncTimes prometheus.Histogram
13-
syncLastSuccess prometheus.Gauge
14-
tenantsDiscovered prometheus.Gauge
15-
tenantsSynced prometheus.Gauge
16-
}
17-
18-
func NewCortexBucketStoreMetrics(reg prometheus.Registerer) *CortexBucketStoreMetrics {
19-
return &CortexBucketStoreMetrics{
20-
syncTimes: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
21-
Name: "cortex_bucket_stores_blocks_sync_seconds",
22-
Help: "The total time it takes to perform a sync stores",
23-
Buckets: []float64{0.1, 1, 10, 30, 60, 120, 300, 600, 900},
24-
}),
25-
syncLastSuccess: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
26-
Name: "cortex_bucket_stores_blocks_last_successful_sync_timestamp_seconds",
27-
Help: "Unix timestamp of the last successful blocks sync.",
28-
}),
29-
tenantsDiscovered: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
30-
Name: "cortex_bucket_stores_tenants_discovered",
31-
Help: "Number of tenants discovered in the bucket.",
32-
}),
33-
tenantsSynced: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
34-
Name: "cortex_bucket_stores_tenants_synced",
35-
Help: "Number of tenants synced.",
36-
}),
37-
}
38-
}
39-
409
// BucketStoreMetrics aggregates metrics exported by Thanos Bucket Store
4110
// and re-exports those aggregates as Cortex metrics.
4211
type BucketStoreMetrics struct {
@@ -91,7 +60,7 @@ type BucketStoreMetrics struct {
9160
}
9261

9362
func NewBucketStoreMetrics() *BucketStoreMetrics {
94-
m := &BucketStoreMetrics{
63+
return &BucketStoreMetrics{
9564
regs: util.NewUserRegistries(),
9665

9766
blockLoads: prometheus.NewDesc(
@@ -264,8 +233,6 @@ func NewBucketStoreMetrics() *BucketStoreMetrics {
264233
"Total number of series size in bytes overfetched due to posting lazy expansion.",
265234
nil, nil),
266235
}
267-
268-
return m
269236
}
270237

271238
func (m *BucketStoreMetrics) AddUserRegistry(user string, reg *prometheus.Registry) {

‎pkg/storegateway/bucket_store_metrics_test.go‎

Lines changed: 0 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -11,45 +11,6 @@ import (
1111
"github.com/stretchr/testify/require"
1212
)
1313

14-
func TestCortexBucketStoreMetrics(t *testing.T) {
15-
t.Parallel()
16-
reg := prometheus.NewPedanticRegistry()
17-
18-
metrics := NewCortexBucketStoreMetrics(reg)
19-
metrics.syncTimes.Observe(0.1)
20-
metrics.syncLastSuccess.Set(1759923308)
21-
metrics.tenantsSynced.Set(1)
22-
metrics.tenantsDiscovered.Set(1)
23-
24-
err := testutil.GatherAndCompare(reg, bytes.NewBufferString(`
25-
# HELP cortex_bucket_stores_blocks_last_successful_sync_timestamp_seconds Unix timestamp of the last successful blocks sync.
26-
# TYPE cortex_bucket_stores_blocks_last_successful_sync_timestamp_seconds gauge
27-
cortex_bucket_stores_blocks_last_successful_sync_timestamp_seconds 1.759923308e+09
28-
# HELP cortex_bucket_stores_blocks_sync_seconds The total time it takes to perform a sync stores
29-
# TYPE cortex_bucket_stores_blocks_sync_seconds histogram
30-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="0.1"} 1
31-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="1"} 1
32-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="10"} 1
33-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="30"} 1
34-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="60"} 1
35-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="120"} 1
36-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="300"} 1
37-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="600"} 1
38-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="900"} 1
39-
cortex_bucket_stores_blocks_sync_seconds_bucket{le="+Inf"} 1
40-
cortex_bucket_stores_blocks_sync_seconds_sum 0.1
41-
cortex_bucket_stores_blocks_sync_seconds_count 1
42-
# HELP cortex_bucket_stores_tenants_discovered Number of tenants discovered in the bucket.
43-
# TYPE cortex_bucket_stores_tenants_discovered gauge
44-
cortex_bucket_stores_tenants_discovered 1
45-
# HELP cortex_bucket_stores_tenants_synced Number of tenants synced.
46-
# TYPE cortex_bucket_stores_tenants_synced gauge
47-
cortex_bucket_stores_tenants_synced 1
48-
`,
49-
))
50-
require.NoError(t, err)
51-
}
52-
5314
func TestBucketStoreMetrics(t *testing.T) {
5415
t.Parallel()
5516
mainReg := prometheus.NewPedanticRegistry()

‎pkg/storegateway/bucket_stores.go‎

Lines changed: 49 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -53,15 +53,14 @@ type BucketStores interface {
5353

5454
// ThanosBucketStores is a multi-tenant wrapper of Thanos BucketStore.
5555
type ThanosBucketStores struct {
56-
logger log.Logger
57-
cfg tsdb.BlocksStorageConfig
58-
limits *validation.Overrides
59-
bucket objstore.Bucket
60-
logLevel logging.Level
61-
bucketStoreMetrics *BucketStoreMetrics
62-
cortexBucketStoreMetrics *CortexBucketStoreMetrics
63-
metaFetcherMetrics *MetadataFetcherMetrics
64-
shardingStrategy ShardingStrategy
56+
logger log.Logger
57+
cfg tsdb.BlocksStorageConfig
58+
limits *validation.Overrides
59+
bucket objstore.Bucket
60+
logLevel logging.Level
61+
bucketStoreMetrics *BucketStoreMetrics
62+
metaFetcherMetrics *MetadataFetcherMetrics
63+
shardingStrategy ShardingStrategy
6564

6665
// Index cache shared across all tenants.
6766
indexCache storecache.IndexCache
@@ -95,6 +94,12 @@ type ThanosBucketStores struct {
9594

9695
// Keeps number of inflight requests
9796
inflightRequests *util.InflightRequestTracker
97+
98+
// Metrics.
99+
syncTimes prometheus.Histogram
100+
syncLastSuccess prometheus.Gauge
101+
tenantsDiscovered prometheus.Gauge
102+
tenantsSynced prometheus.Gauge
98103
}
99104

100105
var ErrTooManyInflightRequests = status.Error(codes.ResourceExhausted, "too many inflight requests in store gateway")
@@ -128,21 +133,37 @@ func newThanosBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy Shardi
128133
}).Set(float64(cfg.BucketStore.MaxConcurrent))
129134

130135
u := &ThanosBucketStores{
131-
logger: logger,
132-
cfg: cfg,
133-
limits: limits,
134-
bucket: cachingBucket,
135-
shardingStrategy: shardingStrategy,
136-
stores: map[string]*store.BucketStore{},
137-
storesErrors: map[string]error{},
138-
logLevel: logLevel,
139-
bucketStoreMetrics: NewBucketStoreMetrics(),
140-
cortexBucketStoreMetrics: NewCortexBucketStoreMetrics(reg),
141-
metaFetcherMetrics: NewMetadataFetcherMetrics(),
142-
queryGate: queryGate,
143-
partitioner: newGapBasedPartitioner(cfg.BucketStore.PartitionerMaxGapBytes, reg),
144-
userTokenBuckets: make(map[string]*util.TokenBucket),
145-
inflightRequests: util.NewInflightRequestTracker(),
136+
logger: logger,
137+
cfg: cfg,
138+
limits: limits,
139+
bucket: cachingBucket,
140+
shardingStrategy: shardingStrategy,
141+
stores: map[string]*store.BucketStore{},
142+
storesErrors: map[string]error{},
143+
logLevel: logLevel,
144+
bucketStoreMetrics: NewBucketStoreMetrics(),
145+
metaFetcherMetrics: NewMetadataFetcherMetrics(),
146+
queryGate: queryGate,
147+
partitioner: newGapBasedPartitioner(cfg.BucketStore.PartitionerMaxGapBytes, reg),
148+
userTokenBuckets: make(map[string]*util.TokenBucket),
149+
inflightRequests: util.NewInflightRequestTracker(),
150+
syncTimes: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
151+
Name: "cortex_bucket_stores_blocks_sync_seconds",
152+
Help: "The total time it takes to perform a sync stores",
153+
Buckets: []float64{0.1, 1, 10, 30, 60, 120, 300, 600, 900},
154+
}),
155+
syncLastSuccess: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
156+
Name: "cortex_bucket_stores_blocks_last_successful_sync_timestamp_seconds",
157+
Help: "Unix timestamp of the last successful blocks sync.",
158+
}),
159+
tenantsDiscovered: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
160+
Name: "cortex_bucket_stores_tenants_discovered",
161+
Help: "Number of tenants discovered in the bucket.",
162+
}),
163+
tenantsSynced: promauto.With(reg).NewGauge(prometheus.GaugeOpts{
164+
Name: "cortex_bucket_stores_tenants_synced",
165+
Help: "Number of tenants synced.",
166+
}),
146167
}
147168
u.userScanner, err = users.NewScanner(cfg.UsersScanner, bucketClient, logger, reg)
148169
if err != nil {
@@ -232,9 +253,9 @@ func (u *ThanosBucketStores) syncUsersBlocksWithRetries(ctx context.Context, f f
232253

233254
func (u *ThanosBucketStores) syncUsersBlocks(ctx context.Context, f func(context.Context, *store.BucketStore) error) (returnErr error) {
234255
defer func(start time.Time) {
235-
u.cortexBucketStoreMetrics.syncTimes.Observe(time.Since(start).Seconds())
256+
u.syncTimes.Observe(time.Since(start).Seconds())
236257
if returnErr == nil {
237-
u.cortexBucketStoreMetrics.syncLastSuccess.SetToCurrentTime()
258+
u.syncLastSuccess.SetToCurrentTime()
238259
}
239260
}(time.Now())
240261

@@ -259,8 +280,8 @@ func (u *ThanosBucketStores) syncUsersBlocks(ctx context.Context, f func(context
259280
includeUserIDs[userID] = struct{}{}
260281
}
261282

262-
u.cortexBucketStoreMetrics.tenantsDiscovered.Set(float64(len(userIDs)))
263-
u.cortexBucketStoreMetrics.tenantsSynced.Set(float64(len(includeUserIDs)))
283+
u.tenantsDiscovered.Set(float64(len(userIDs)))
284+
u.tenantsSynced.Set(float64(len(includeUserIDs)))
264285

265286
// Create a pool of workers which will synchronize blocks. The pool size
266287
// is limited in order to avoid to concurrently sync a lot of tenants in

‎pkg/storegateway/bucket_stores_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -266,7 +266,7 @@ func TestBucketStores_InitialSync(t *testing.T) {
266266
))
267267

268268
thanosStores := stores.(*ThanosBucketStores)
269-
assert.Greater(t, testutil.ToFloat64(thanosStores.cortexBucketStoreMetrics.syncLastSuccess), float64(0))
269+
assert.Greater(t, testutil.ToFloat64(thanosStores.syncLastSuccess), float64(0))
270270
}
271271

272272
func TestBucketStores_InitialSyncShouldRetryOnFailure(t *testing.T) {
@@ -327,7 +327,7 @@ func TestBucketStores_InitialSyncShouldRetryOnFailure(t *testing.T) {
327327
))
328328

329329
thanosStores := stores.(*ThanosBucketStores)
330-
assert.Greater(t, testutil.ToFloat64(thanosStores.cortexBucketStoreMetrics.syncLastSuccess), float64(0))
330+
assert.Greater(t, testutil.ToFloat64(thanosStores.syncLastSuccess), float64(0))
331331
}
332332

333333
func TestBucketStores_SyncBlocks(t *testing.T) {
@@ -398,7 +398,7 @@ func TestBucketStores_SyncBlocks(t *testing.T) {
398398
))
399399

400400
thanosStores := stores.(*ThanosBucketStores)
401-
assert.Greater(t, testutil.ToFloat64(thanosStores.cortexBucketStoreMetrics.syncLastSuccess), float64(0))
401+
assert.Greater(t, testutil.ToFloat64(thanosStores.syncLastSuccess), float64(0))
402402
}
403403

404404
func TestBucketStores_syncUsersBlocks(t *testing.T) {

‎pkg/storegateway/parquet_bucket_stores.go‎

Lines changed: 10 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,7 @@ type ParquetBucketStores struct {
5757

5858
inflightRequests *cortex_util.InflightRequestTracker
5959

60-
cortexBucketStoreMetrics *CortexBucketStoreMetrics
61-
userScanner users.Scanner
60+
userScanner users.Scanner
6261

6362
userTokenBuckets map[string]*cortex_util.TokenBucket
6463
}
@@ -72,16 +71,15 @@ func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, bucketClient objstore.
7271
}
7372

7473
u := &ParquetBucketStores{
75-
logger: logger,
76-
cfg: cfg,
77-
limits: limits,
78-
bucket: cachingBucket,
79-
stores: map[string]*parquetBucketStore{},
80-
storesErrors: map[string]error{},
81-
chunksDecoder: schema.NewPrometheusParquetChunksDecoder(chunkenc.NewPool()),
82-
inflightRequests: cortex_util.NewInflightRequestTracker(),
83-
cortexBucketStoreMetrics: NewCortexBucketStoreMetrics(reg),
84-
userTokenBuckets: make(map[string]*cortex_util.TokenBucket),
74+
logger: logger,
75+
cfg: cfg,
76+
limits: limits,
77+
bucket: cachingBucket,
78+
stores: map[string]*parquetBucketStore{},
79+
storesErrors: map[string]error{},
80+
chunksDecoder: schema.NewPrometheusParquetChunksDecoder(chunkenc.NewPool()),
81+
inflightRequests: cortex_util.NewInflightRequestTracker(),
82+
userTokenBuckets: make(map[string]*cortex_util.TokenBucket),
8583
}
8684
u.userScanner, err = users.NewScanner(cfg.UsersScanner, bucketClient, logger, reg)
8785
if err != nil {

0 commit comments

Comments
 (0)