Change retention to honor compaction disablement

CHANGELOG.md

@@ -29,6 +29,7 @@
 * [CHANGE] Improve parquet readers io.ReaderAt compatibility [#4963](https://github.com/grafana/tempo/pull/4963) (@joe-elliott)
 * [CHANGE] Finish polling current tenants on poller shutdown [#4897](https://github.com/grafana/tempo/pull/4897) (@zalegrala)
 * [CHANGE] Set querier default level to INFO [#4943](https://github.com/grafana/tempo/pull/4943) (@javiermolinar)
+* [CHANGE] Change retention to honor compactor disablement [#5044](https://github.com/grafana/tempo/pull/5044) (@zalegrala)
 * [FEATURE] Add throughput SLO and metrics for the TraceByID endpoint. [#4668](https://github.com/grafana/tempo/pull/4668) (@carles-grafana)
 configurable via the throughput_bytes_slo field, and it will populate op="traces" label in slo and throughput metrics.
 * [FEATURE] Added most_recent=true query hint to TraceQL to return most recent results. [#4238](https://github.com/grafana/tempo/pull/4238) (@joe-elliott)

docs/sources/tempo/configuration/_index.md

@@ -1699,7 +1699,7 @@ overrides:
       # 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]
-      # Allow compaction to be deactivated on a per-tenant basis. Default value
+      # Allow compaction and retention to be deactivated on a per-tenant basis. Default value
       # is false (compaction active). Useful to perform operations on the backend
       # that require compaction to be disabled for a period of time.
       [compaction_disabled: <bool> | default = false]

tempodb/retention.go

@@ -42,9 +42,14 @@ func (rw *readerWriter) RetainWithConfig(ctx context.Context, compactorCfg *Comp
 	bg := boundedwaitgroup.New(compactorCfg.RetentionConcurrency)
 
 	for _, tenantID := range tenants {
+		if compactorOverrides.CompactionDisabledForTenant(tenantID) {
+			continue
+		}
+
 		bg.Add(1)
 		go func(t string) {
 			defer bg.Done()
+
 			rw.retainTenant(ctx, t, compactorCfg, compactorSharder, compactorOverrides)
 		}(tenantID)
 	}

tempodb/retention_test.go

@@ -222,6 +222,63 @@ func TestBlockRetentionOverride(t *testing.T) {
 	require.Equal(t, 0, len(rw.blocklist.Metas(testTenantID)))
 }
 
+func TestBlockRetentionOverrideDisabled(t *testing.T) {
+	tempDir := t.TempDir()
+
+	r, w, c, err := New(&Config{
+		Backend: backend.Local,
+		Local: &local.Config{
+			Path: path.Join(tempDir, "traces"),
+		},
+		Block: &common.BlockConfig{
+			IndexDownsampleBytes: 17,
+			BloomFP:              0.01,
+			BloomShardSizeBytes:  100_000,
+			Version:              encoding.DefaultEncoding().Version(),
+			Encoding:             backend.EncLZ4_256k,
+			IndexPageSizeBytes:   1000,
+		},
+		WAL: &wal.Config{
+			Filepath: path.Join(tempDir, "wal"),
+		},
+		BlocklistPoll: 0,
+	}, nil, log.NewNopLogger())
+	require.NoError(t, err)
+
+	overrides := &mockOverrides{
+		disabled: true,
+	}
+
+	ctx := context.Background()
+	err = c.EnableCompaction(ctx, &CompactorConfig{
+		ChunkSizeBytes:          10,
+		MaxCompactionRange:      time.Hour,
+		BlockRetention:          time.Hour,
+		CompactedBlockRetention: 0,
+	}, &mockSharder{}, overrides)
+	require.NoError(t, err)
+
+	r.EnablePolling(ctx, &mockJobSharder{})
+
+	cutTestBlocks(t, w, testTenantID, 10, 10)
+
+	// The test spans are all 1 second long, so we have to sleep to put all the
+	// data in the past
+	time.Sleep(time.Second)
+
+	rw := r.(*readerWriter)
+	rw.pollBlocklist(ctx)
+	require.Equal(t, 10, len(rw.blocklist.Metas(testTenantID)))
+
+	time.Sleep(10 * time.Millisecond)
+
+	// Retention = 1ns, deletes everything
+	overrides.blockRetention = time.Nanosecond
+	r.(*readerWriter).doRetention(ctx)
+	rw.pollBlocklist(ctx)
+	require.Equal(t, 10, len(rw.blocklist.Metas(testTenantID)))
+}
+
 func TestRetainWithConfig(t *testing.T) {
 	for _, enc := range encoding.AllEncodings() {
 		version := enc.Version()