Add per-tenant compaction window

docs/sources/tempo/configuration/_index.md

@@ -1323,6 +1323,9 @@ overrides:
       # Per-user block retention. If this value is set to 0 (default),
       # then block_retention in the compactor configuration is used.
       [block_retention: <duration> | default = 0s]
+      # Per-user compaction window. If this value is set to 0 (default),
+      # then block_retention in the compactor configuration is used.
+      [compaction_window: <duration> | default = 0s]
 
     # Metrics-generator related overrides
     metrics_generator:

modules/compactor/compactor.go

@@ -258,6 +258,10 @@ func (c *Compactor) MaxBytesPerTraceForTenant(tenantID string) int {
 	return c.overrides.MaxBytesPerTrace(tenantID)
 }
 
+func (c *Compactor) MaxCompactionRangeForTenant(tenantID string) time.Duration {
+	return c.overrides.MaxCompactionRange(tenantID)
+}
+
 func (c *Compactor) isSharded() bool {
 	return c.cfg.ShardingRing.KVStore.Store != ""
 }

modules/overrides/config.go

@@ -8,9 +8,10 @@ import (
 
 	"github.com/grafana/tempo/tempodb/backend"
 
+	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/grafana/tempo/pkg/sharedconfig"
 	filterconfig "github.com/grafana/tempo/pkg/spanfilter/config"
-	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/prometheus/common/model"
 )
@@ -44,6 +45,7 @@ const (
 	MetricIngestionRateLimitBytes         = "ingestion_rate_limit_bytes"
 	MetricIngestionBurstSizeBytes         = "ingestion_burst_size_bytes"
 	MetricBlockRetention                  = "block_retention"
+	MetricCompactionWindow                = "compaction_window"
 	MetricMetricsGeneratorMaxActiveSeries = "metrics_generator_max_active_series"
 	MetricsGeneratorDryRunEnabled         = "metrics_generator_dry_run_enabled"
 )
@@ -129,7 +131,8 @@ type ReadOverrides struct {
 
 type CompactionOverrides struct {
 	// Compactor enforced overrides.
-	BlockRetention model.Duration `yaml:"block_retention,omitempty" json:"block_retention,omitempty"`
+	BlockRetention   model.Duration `yaml:"block_retention,omitempty" json:"block_retention,omitempty"`
+	CompactionWindow model.Duration `yaml:"compaction_window,omitempty" json:"compaction_window,omitempty"`
 }
 
 type GlobalOverrides struct {
@@ -247,5 +250,6 @@ func (c *Config) Collect(ch chan<- prometheus.Metric) {
 	ch <- prometheus.MustNewConstMetric(metricLimitsDesc, prometheus.GaugeValue, float64(c.Defaults.Read.MaxBlocksPerTagValuesQuery), MetricMaxBlocksPerTagValuesQuery)
 	ch <- prometheus.MustNewConstMetric(metricLimitsDesc, prometheus.GaugeValue, float64(c.Defaults.Global.MaxBytesPerTrace), MetricMaxBytesPerTrace)
 	ch <- prometheus.MustNewConstMetric(metricLimitsDesc, prometheus.GaugeValue, float64(c.Defaults.Compaction.BlockRetention), MetricBlockRetention)
+	ch <- prometheus.MustNewConstMetric(metricLimitsDesc, prometheus.GaugeValue, float64(c.Defaults.Compaction.CompactionWindow), MetricCompactionWindow)
 	ch <- prometheus.MustNewConstMetric(metricLimitsDesc, prometheus.GaugeValue, float64(c.Defaults.MetricsGenerator.MaxActiveSeries), MetricMetricsGeneratorMaxActiveSeries)
 }

modules/overrides/config_legacy.go

@@ -48,7 +48,8 @@ func (c *Overrides) toLegacy() LegacyOverrides {
 		MetricsGeneratorProcessorLocalBlocksCompleteBlockTimeout:         c.MetricsGenerator.Processor.LocalBlocks.CompleteBlockTimeout,
 		MetricsGeneratorIngestionSlack:                                   c.MetricsGenerator.IngestionSlack,
 
-		BlockRetention: c.Compaction.BlockRetention,
+		BlockRetention:   c.Compaction.BlockRetention,
+		CompactionWindow: c.Compaction.CompactionWindow,
 
 		MaxBytesPerTagValuesQuery:  c.Read.MaxBytesPerTagValuesQuery,
 		MaxBlocksPerTagValuesQuery: c.Read.MaxBlocksPerTagValuesQuery,
@@ -103,7 +104,8 @@ type LegacyOverrides struct {
 	MetricsGeneratorIngestionSlack                                   time.Duration                    `yaml:"metrics_generator_ingestion_time_range_slack" json:"metrics_generator_ingestion_time_range_slack"`
 
 	// Compactor enforced limits.
-	BlockRetention model.Duration `yaml:"block_retention" json:"block_retention"`
+	BlockRetention   model.Duration `yaml:"block_retention" json:"block_retention"`
+	CompactionWindow model.Duration `yaml:"compaction_window" json:"compaction_window"`
 
 	// Querier and Ingester enforced limits.
 	MaxBytesPerTagValuesQuery  int `yaml:"max_bytes_per_tag_values_query" json:"max_bytes_per_tag_values_query"`
@@ -135,7 +137,8 @@ func (l *LegacyOverrides) toNewLimits() Overrides {
 			MaxSearchDuration:          l.MaxSearchDuration,
 		},
 		Compaction: CompactionOverrides{
-			BlockRetention: l.BlockRetention,
+			BlockRetention:   l.BlockRetention,
+			CompactionWindow: l.CompactionWindow,
 		},
 		MetricsGenerator: MetricsGeneratorOverrides{
 			RingSize:           l.MetricsGeneratorRingSize,

modules/overrides/interface.go

@@ -26,6 +26,7 @@ type Interface interface {
 	MaxLocalTracesPerUser(userID string) int
 	MaxGlobalTracesPerUser(userID string) int
 	MaxBytesPerTrace(userID string) int
+	MaxCompactionRange(userID string) time.Duration
 	Forwarders(userID string) []string
 	MaxBytesPerTagValuesQuery(userID string) int
 	MaxBlocksPerTagValuesQuery(userID string) int

modules/overrides/runtime_config_overrides.go

@@ -261,6 +261,11 @@ func (o *runtimeConfigOverridesManager) MaxGlobalTracesPerUser(userID string) in
 	return o.getOverridesForUser(userID).Ingestion.MaxGlobalTracesPerUser
 }
 
+// MaxCompactionRange returns the maximum compaction window for this tenant.
+func (o *runtimeConfigOverridesManager) MaxCompactionRange(userID string) time.Duration {
+	return time.Duration(o.getOverridesForUser(userID).Compaction.CompactionWindow)
+}
+
 // IngestionRateLimitBytes is the number of spans per second allowed for this tenant.
 func (o *runtimeConfigOverridesManager) IngestionRateLimitBytes(userID string) float64 {
 	return float64(o.getOverridesForUser(userID).Ingestion.RateLimitBytes)

tempodb/compactor.go

@@ -107,6 +107,11 @@ func (rw *readerWriter) doCompaction(ctx context.Context) {
 	// Get the meta file of all non-compacted blocks for the given tenant
 	blocklist := rw.blocklist.Metas(tenantID)
 
+	window := rw.compactorOverrides.MaxCompactionRangeForTenant(tenantID)
+	if window == 0 {
+		window = rw.compactorCfg.MaxCompactionRange
+	}
+
 	// Select which blocks to compact.
 	//
 	// Blocks are firstly divided by the active compaction window (default: most recent 24h)
@@ -115,7 +120,7 @@ func (rw *readerWriter) doCompaction(ctx context.Context) {
 	//  2. If blocks are outside the active window, they're grouped only by windows, ignoring compaction level.
 	//   It picks more recent windows first, and compacting blocks only from the same tenant.
 	blockSelector := newTimeWindowBlockSelector(blocklist,
-		rw.compactorCfg.MaxCompactionRange,
+		window,
 		rw.compactorCfg.MaxCompactionObjects,
 		rw.compactorCfg.MaxBlockBytes,
 		defaultMinInputBlocks,

tempodb/compactor_test.go

@@ -46,8 +46,9 @@ type mockJobSharder struct{}
 func (m *mockJobSharder) Owns(string) bool { return true }
 
 type mockOverrides struct {
-	blockRetention   time.Duration
-	maxBytesPerTrace int
+	blockRetention      time.Duration
+	maxBytesPerTrace    int
+	maxCompactionWindow time.Duration
 }
 
 func (m *mockOverrides) BlockRetentionForTenant(_ string) time.Duration {
@@ -58,6 +59,10 @@ func (m *mockOverrides) MaxBytesPerTraceForTenant(_ string) int {
 	return m.maxBytesPerTrace
 }
 
+func (m *mockOverrides) MaxCompactionRangeForTenant(_ string) time.Duration {
+	return m.maxCompactionWindow
+}
+
 func TestCompactionRoundtrip(t *testing.T) {
 	for _, enc := range encoding.AllEncodings() {
 		version := enc.Version()

tempodb/tempodb.go

@@ -96,6 +96,7 @@ type CompactorSharder interface {
 type CompactorOverrides interface {
 	BlockRetentionForTenant(tenantID string) time.Duration
 	MaxBytesPerTraceForTenant(tenantID string) int
+	MaxCompactionRangeForTenant(tenantID string) time.Duration
 }
 
 type WriteableBlock interface {