Creating new Metric cortex_compactor_blocks_marked_for_no_compaction_on_storage_total

Author: alanprot

pkg/compactor/blocks_cleaner.go

@@ -45,17 +45,18 @@ type BlocksCleaner struct {
 	lastOwnedUsers []string
 
 	// Metrics.
-	runsStarted                 prometheus.Counter
-	runsCompleted               prometheus.Counter
-	runsFailed                  prometheus.Counter
-	runsLastSuccess             prometheus.Gauge
-	blocksCleanedTotal          prometheus.Counter
-	blocksFailedTotal           prometheus.Counter
-	blocksMarkedForDeletion     prometheus.Counter
-	tenantBlocks                *prometheus.GaugeVec
-	tenantMarkedBlocks          *prometheus.GaugeVec
-	tenantPartialBlocks         *prometheus.GaugeVec
-	tenantBucketIndexLastUpdate *prometheus.GaugeVec
+	runsStarted                          prometheus.Counter
+	runsCompleted                        prometheus.Counter
+	runsFailed                           prometheus.Counter
+	runsLastSuccess                      prometheus.Gauge
+	blocksCleanedTotal                   prometheus.Counter
+	blocksFailedTotal                    prometheus.Counter
+	blocksMarkedForDeletion              prometheus.Counter
+	tenantBlocks                         *prometheus.GaugeVec
+	tenantMarkedBlocks                   *prometheus.GaugeVec
+	tenantPartialBlocks                  *prometheus.GaugeVec
+	tenantBucketIndexLastUpdate          *prometheus.GaugeVec
+	blocksMarkedForNoCompactionOnStorage *prometheus.GaugeVec
 }
 
 func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, usersScanner *cortex_tsdb.UsersScanner, cfgProvider ConfigProvider, logger log.Logger, reg prometheus.Registerer) *BlocksCleaner {
@@ -69,6 +70,10 @@ func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, use
 			Name: "cortex_compactor_block_cleanup_started_total",
 			Help: "Total number of blocks cleanup runs started.",
 		}),
+		blocksMarkedForNoCompactionOnStorage: promauto.With(reg).NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cortex_compactor_blocks_marked_for_no_compaction_on_storage_total",
+			Help: "Total number of blocks marked for on storage.",
+		}, []string{"user"}),
 		runsCompleted: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_block_cleanup_completed_total",
 			Help: "Total number of blocks cleanup runs successfully completed.",
@@ -241,6 +246,7 @@ func (c *BlocksCleaner) deleteUserMarkedForDeletion(ctx context.Context, userID
 	c.tenantBlocks.DeleteLabelValues(userID)
 	c.tenantMarkedBlocks.DeleteLabelValues(userID)
 	c.tenantPartialBlocks.DeleteLabelValues(userID)
+	c.blocksMarkedForNoCompactionOnStorage.DeleteLabelValues(userID)
 
 	if deletedBlocks > 0 {
 		level.Info(userLogger).Log("msg", "deleted blocks for tenant marked for deletion", "deletedBlocks", deletedBlocks)
@@ -331,6 +337,7 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b
 	// Generate an updated in-memory version of the bucket index.
 	w := bucketindex.NewUpdater(c.bucketClient, userID, c.cfgProvider, c.logger)
 	idx, partials, err := w.UpdateIndex(ctx, idx)
+	c.blocksMarkedForNoCompactionOnStorage.WithLabelValues(userID).Set(float64(idx.TotalBlocksBlocksMarkedForNoCompaction))
 	if err != nil {
 		return err
 	}

pkg/storage/tsdb/bucketindex/index.go

@@ -41,6 +41,9 @@ type Index struct {
 	// UpdatedAt is a unix timestamp (seconds precision) of when the index has been updated
 	// (written in the storage) the last time.
 	UpdatedAt int64 `json:"updated_at"`
+
+	// TotalBlocksBlocksMarkedForNoCompaction is then number of blocks marked for no compaction
+	TotalBlocksBlocksMarkedForNoCompaction int64 `json:"total_blocks_marked_for_no_compaction"`
 }
 
 func (idx *Index) GetUpdatedAt() time.Time {

pkg/storage/tsdb/bucketindex/markers.go

@@ -51,6 +51,24 @@ func IsBlockDeletionMarkFilename(name string) (ulid.ULID, bool) {
 	return id, err == nil
 }
 
+// IsBlockNoCompactMarkFilename returns whether the input filename matches the expected pattern
+// of block no compact markers stored in the markers location.
+func IsBlockNoCompactMarkFilename(name string) (ulid.ULID, bool) {
+	parts := strings.SplitN(name, "-", 2)
+	if len(parts) != 2 {
+		return ulid.ULID{}, false
+	}
+
+	// Ensure the 2nd part matches the block deletion mark filename.
+	if parts[1] != metadata.NoCompactMarkFilename {
+		return ulid.ULID{}, false
+	}
+
+	// Ensure the 1st part is a valid block ID.
+	id, err := ulid.Parse(filepath.Base(parts[0]))
+	return id, err == nil
+}
+
 // MigrateBlockDeletionMarksToGlobalLocation list all tenant's blocks and, for each of them, look for
 // a deletion mark in the block location. Found deletion marks are copied to the global markers location.
 // The migration continues on error and returns once all blocks have been checked.

pkg/storage/tsdb/bucketindex/markers_bucket_client.go

@@ -148,13 +148,3 @@ func (b *globalMarkersBucket) isMark(name string) (string, bool) {
 
 	return "", false
 }
-
-func (b *globalMarkersBucket) isBlockDeletionMark(name string) (ulid.ULID, bool) {
-	if path.Base(name) != metadata.DeletionMarkFilename {
-		return ulid.ULID{}, false
-	}
-
-	// Parse the block ID in the path. If there's not block ID, then it's not the per-block
-	// deletion mark.
-	return block.IsBlockDir(path.Dir(name))
-}

pkg/storage/tsdb/bucketindex/updater.go

@@ -57,16 +57,17 @@ func (w *Updater) UpdateIndex(ctx context.Context, old *Index) (*Index, map[ulid
 		return nil, nil, err
 	}
 
-	blockDeletionMarks, err := w.updateBlockDeletionMarks(ctx, oldBlockDeletionMarks)
+	blockDeletionMarks, totalBlocksBlocksMarkedForNoCompaction, err := w.updateBlockMarks(ctx, oldBlockDeletionMarks)
 	if err != nil {
 		return nil, nil, err
 	}
 
 	return &Index{
-		Version:            IndexVersion1,
-		Blocks:             blocks,
-		BlockDeletionMarks: blockDeletionMarks,
-		UpdatedAt:          time.Now().Unix(),
+		Version:                                IndexVersion1,
+		Blocks:                                 blocks,
+		BlockDeletionMarks:                     blockDeletionMarks,
+		UpdatedAt:                              time.Now().Unix(),
+		TotalBlocksBlocksMarkedForNoCompaction: totalBlocksBlocksMarkedForNoCompaction,
 	}, partials, nil
 }
 
@@ -163,19 +164,25 @@ func (w *Updater) updateBlockIndexEntry(ctx context.Context, id ulid.ULID) (*Blo
 	return block, nil
 }
 
-func (w *Updater) updateBlockDeletionMarks(ctx context.Context, old []*BlockDeletionMark) ([]*BlockDeletionMark, error) {
+func (w *Updater) updateBlockMarks(ctx context.Context, old []*BlockDeletionMark) ([]*BlockDeletionMark, int64, error) {
 	out := make([]*BlockDeletionMark, 0, len(old))
 	discovered := map[ulid.ULID]struct{}{}
+	totalBlocksBlocksMarkedForNoCompaction := int64(0)
 
 	// Find all markers in the storage.
 	err := w.bkt.Iter(ctx, MarkersPathname+"/", func(name string) error {
 		if blockID, ok := IsBlockDeletionMarkFilename(path.Base(name)); ok {
 			discovered[blockID] = struct{}{}
 		}
+
+		if _, ok := IsBlockNoCompactMarkFilename(path.Base(name)); ok {
+			totalBlocksBlocksMarkedForNoCompaction++
+		}
+
 		return nil
 	})
 	if err != nil {
-		return nil, errors.Wrap(err, "list block deletion marks")
+		return nil, totalBlocksBlocksMarkedForNoCompaction, errors.Wrap(err, "list block deletion marks")
 	}
 
 	// Since deletion marks are immutable, all markers already existing in the index can just be copied.
@@ -199,13 +206,13 @@ func (w *Updater) updateBlockDeletionMarks(ctx context.Context, old []*BlockDele
 			continue
 		}
 		if err != nil {
-			return nil, err
+			return nil, totalBlocksBlocksMarkedForNoCompaction, err
 		}
 
 		out = append(out, m)
 	}
 
-	return out, nil
+	return out, totalBlocksBlocksMarkedForNoCompaction, nil
 }
 
 func (w *Updater) updateBlockDeletionMarkIndexEntry(ctx context.Context, id ulid.ULID) (*BlockDeletionMark, error) {