Address comments

Signed-off-by: Daniel Deluiggi <[email protected]>

Author: danielblando

CHANGELOG.md

@@ -43,7 +43,7 @@
 * [ENHANCEMENT] Compactor: Optimize cleaner run time. #6815
 * [ENHANCEMENT] Parquet Storage: Allow percentage based dynamic shard size for Parquet Converter. #6817
 * [ENHANCEMENT] Query Frontend: Enhance the performance of the JSON codec. #6816
-* [ENHANCEMENT] Compactor: Split cleaner logic. #6827
+* [ENHANCEMENT] Compactor: Emit partition metrics separate from cleaner job. #6827
 * [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517
 * [BUGFIX] Ingester: Fix labelset data race condition. #6573
 * [BUGFIX] Compactor: Cleaner should not put deletion marker for blocks with no-compact marker. #6576

pkg/compactor/blocks_cleaner.go

@@ -249,7 +249,7 @@ func (c *BlocksCleaner) loop(ctx context.Context) error {
 		metricsChan = make(chan *cleanerJob)
 		defer close(metricsChan)
 		go func() {
-			c.runEmitMetricsWorker(ctx, metricsChan)
+			c.runEmitPartitionMetricsWorker(ctx, metricsChan)
 		}()
 	}
 
@@ -315,12 +315,12 @@ func (c *BlocksCleaner) checkRunError(runType string, err error) {
 	}
 }
 
-func (c *BlocksCleaner) runEmitMetricsWorker(ctx context.Context, jobChan <-chan *cleanerJob) {
+func (c *BlocksCleaner) runEmitPartitionMetricsWorker(ctx context.Context, jobChan <-chan *cleanerJob) {
 	for job := range jobChan {
 		err := concurrency.ForEachUser(ctx, job.users, c.cfg.CleanupConcurrency, func(ctx context.Context, userID string) error {
 			userLogger := util_log.WithUserID(userID, c.logger)
 			userBucket := bucket.NewUserBucketClient(userID, c.bucketClient, c.cfgProvider)
-			c.emitUserMetrics(ctx, userLogger, userBucket, userID)
+			c.emitUserParititionMetrics(ctx, userLogger, userBucket, userID)
 			return nil
 		})
 
@@ -819,10 +819,11 @@ func (c *BlocksCleaner) cleanPartitionedGroupInfo(ctx context.Context, userBucke
 	}
 }
 
-func (c *BlocksCleaner) emitUserMetrics(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string) {
+func (c *BlocksCleaner) emitUserParititionMetrics(ctx context.Context, userLogger log.Logger, userBucket objstore.InstrumentedBucket, userID string) {
 	existentPartitionedGroupInfo, err := c.iterPartitionGroups(ctx, userBucket, userLogger)
 	if err != nil {
-		level.Warn(userLogger).Log("msg", "error return when going through partitioned group directory", "err", err)
+		level.Warn(userLogger).Log("msg", "error listing partitioned group directory to emit metrics", "err", err)
+		return
 	}
 
 	remainingCompactions := 0
@@ -831,13 +832,11 @@ func (c *BlocksCleaner) emitUserMetrics(ctx context.Context, userLogger log.Logg
 	defer func() {
 		c.remainingPlannedCompactions.WithLabelValues(userID).Set(float64(remainingCompactions))
 		c.inProgressCompactions.WithLabelValues(userID).Set(float64(inProgressCompactions))
-		if c.oldestPartitionGroupOffset != nil {
-			if oldestPartitionGroup != nil {
-				c.oldestPartitionGroupOffset.WithLabelValues(userID).Set(float64(time.Now().Unix() - oldestPartitionGroup.CreationTime))
-				level.Debug(userLogger).Log("msg", "partition group info with oldest creation time", "partitioned_group_id", oldestPartitionGroup.PartitionedGroupID, "creation_time", oldestPartitionGroup.CreationTime)
-			} else {
-				c.oldestPartitionGroupOffset.WithLabelValues(userID).Set(0)
-			}
+		if oldestPartitionGroup != nil {
+			c.oldestPartitionGroupOffset.WithLabelValues(userID).Set(float64(time.Now().Unix() - oldestPartitionGroup.CreationTime))
+			level.Debug(userLogger).Log("msg", "partition group info with oldest creation time", "partitioned_group_id", oldestPartitionGroup.PartitionedGroupID, "creation_time", oldestPartitionGroup.CreationTime)
+		} else {
+			c.oldestPartitionGroupOffset.WithLabelValues(userID).Set(0)
 		}
 	}()
 	for partitionedGroupInfo, extraInfo := range existentPartitionedGroupInfo {

pkg/compactor/blocks_cleaner_test.go

@@ -1226,7 +1226,7 @@ func TestBlocksCleaner_EmitUserMetrics(t *testing.T) {
 	err = v4Manager.updateVisitMarker(ctx)
 	require.NoError(t, err)
 
-	cleaner.emitUserMetrics(ctx, logger, userBucket, userID)
+	cleaner.emitUserParititionMetrics(ctx, logger, userBucket, userID)
 
 	metricNames := []string{
 		"cortex_compactor_remaining_planned_compactions",

pkg/storage/tsdb/caching_bucket.go

@@ -198,7 +198,7 @@ func (cfg *MetadataCacheConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix
 	f.DurationVar(&cfg.BlockIndexAttributesTTL, prefix+"block-index-attributes-ttl", 168*time.Hour, "How long to cache attributes of the block index.")
 	f.DurationVar(&cfg.BucketIndexContentTTL, prefix+"bucket-index-content-ttl", 5*time.Minute, "How long to cache content of the bucket index.")
 	f.IntVar(&cfg.BucketIndexMaxSize, prefix+"bucket-index-max-size-bytes", 1*1024*1024, "Maximum size of bucket index content to cache in bytes. Caching will be skipped if the content exceeds this size. This is useful to avoid network round trip for large content if the configured caching backend has an hard limit on cached items size (in this case, you should set this limit to the same limit in the caching backend).")
-	f.DurationVar(&cfg.PartitionedGroupsListTTL, prefix+"partitioned-groups-list-ttl", 5*time.Minute, "How long to cache list of partitioned groups for an user.")
+	f.DurationVar(&cfg.PartitionedGroupsListTTL, prefix+"partitioned-groups-list-ttl", 0, "How long to cache list of partitioned groups for an user. 0 disables caching")
 }
 
 func (cfg *MetadataCacheConfig) Validate() error {
@@ -279,8 +279,10 @@ func CreateCachingBucketForCompactor(metadataConfig MetadataCacheConfig, cleaner
 			// Cache only GET for metadata and don't cache exists and not exists.
 			cfg.CacheGet("metafile", metadataCache, matchers.GetMetafileMatcher(), metadataConfig.MetafileMaxSize, metadataConfig.MetafileContentTTL, 0, 0)
 
-			//Avoid double iter when running cleanActiveUser and emitUserMetrics
-			cfg.CacheIter("partitioned-groups-iter", metadataCache, matchers.GetPartitionedGroupsIterMatcher(), metadataConfig.PartitionedGroupsListTTL, codec, "")
+			if metadataConfig.PartitionedGroupsListTTL > 0 {
+				//Avoid double iter when running cleanActiveUser and emitUserMetrics
+				cfg.CacheIter("partitioned-groups-iter", metadataCache, matchers.GetPartitionedGroupsIterMatcher(), metadataConfig.PartitionedGroupsListTTL, codec, "")
+			}
 		}
 	}