[r389] Revert ingester Kafka record timestamps for active series (#14611, #14744) (#15038)

Backport 1989c26 from #15036

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes the time basis for active-series purging/loading in
ingest-storage clusters, which can affect metrics accuracy and
early-compaction decisions under Kafka lag or replay.
> 
> **Overview**
> Reverts the prior **Kafka record timestamp–driven** active-series
logic for ingest-storage mode, removing ingester state used to track
latest Kafka time and inline Kafka-time purges.
> 
> Active-series purging/loading now uses **wall-clock `time.Now()`
only** (including early-compaction estimations), and the
`cortex_ingester_active_series_loading` transition is handled by the
metrics updater ticker rather than Kafka-time bookkeeping.
> 
> Removes Kafka record timestamp propagation via context in the ingest
`PusherConsumer` and deletes the related unit/integration tests and
changelog entry for the reverted enhancement.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
8604f22. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Co-authored-by: Oleg Zaytsev <mail@olegzaytsev.com>

Author: mimir-github-bot[bot]

CHANGELOG.md

@@ -148,7 +148,6 @@
 * [ENHANCEMENT] Querier: Add new config flag `querier.enable-delayed-name-removal-prometheus-engine` to enable delayed name removal for Prometheus engine. #14349
 * [ENHANCEMENT] Ingester: reduce heap usage during streaming chunk queries by releasing series label memory after each batch is sent rather than holding it until chunk streaming completes. #14422
 * [ENHANCEMENT] Ingester: Eliminate 20-minute active series metrics loading period when custom tracker or cost attribution configuration changes. Active series counts are now immediately correct after a config reload. #14537
-* [ENHANCEMENT] Ingester: Use Kafka record timestamps instead of wall-clock time for active series tracking, grace period validation, and purging when ingest storage is enabled. This prevents premature purging or stale series caused by lag between record production and consumption. #14611 #14744
 * [ENHANCEMENT] Ingester: Export `cortex_ingester_active_series_loading` gauge metric that is `1` while active series counts are still warming up after ingester startup, and `0` once they are accurate (after IdleTimeout has elapsed). #14783
 * [ENHANCEMENT] Ingest storage: Skip kotel tracing hooks for unsampled traces in the franz-go Kafka client, significantly reducing CPU and memory overhead. #14852
 * [ENHANCEMENT] Distributor: Reduced CPU utilization when writing to ingest storage with a large number of partitions by batching all partitions into a single Kafka produce call instead of one per partition. #14898

pkg/ingester/ingester.go

@@ -349,22 +349,6 @@ type Ingester struct {
 	ingestPartitionID         int32
 	ingestPartitionLifecycler *ring.PartitionInstanceLifecycler
 
-	// latestKafkaRecordTimestamp tracks the most recent Kafka record timestamp
-	// seen by the ingester (unix milliseconds). Used to provide a Kafka-time-aware
-	// "now" for active series purging when ingest storage is enabled.
-	latestKafkaRecordTimestamp atomic.Int64
-
-	// lastKafkaActiveSeriesUpdate tracks the last Kafka time (unix milliseconds) at which
-	// updateActiveSeries was triggered inline during record consumption. This ensures
-	// active series are purged at approximately UpdatePeriod intervals in Kafka time,
-	// even when records are replayed faster than real-time.
-	lastKafkaActiveSeriesUpdate atomic.Int64
-
-	// activeSeriesStartMs tracks when the ingester started receiving samples (unix milliseconds).
-	// For classic mode this is wall-clock time when the ticker starts; for Kafka mode it is the
-	// first Kafka record timestamp. Used to determine when active series counts become accurate.
-	activeSeriesStartMs atomic.Int64
-
 	circuitBreaker  ingesterCircuitBreaker
 	reactiveLimiter *ingesterReactiveLimiter
 }

pkg/ingester/ingester_compaction.go

@@ -140,10 +140,10 @@ func (i *Ingester) compactionServiceRunning(ctx context.Context) error {
 			i.compactBlocks(ctx, false, 0, nil)
 
 			// Check if any TSDB Head should be compacted to reduce the number of in-memory series.
-			i.compactBlocksToReduceInMemorySeries(ctx, i.activeSeriesNow())
+			i.compactBlocksToReduceInMemorySeries(ctx, time.Now())
 
 			// Check if any TSDB Head should be compacted based on per-tenant owned series thresholds.
-			i.compactBlocksToReducePerTenantOwnedSeries(ctx, i.activeSeriesNow())
+			i.compactBlocksToReducePerTenantOwnedSeries(ctx, time.Now())
 
 			// Decrement the counter after compaction is complete
 			i.numCompactionsInProgress.Dec()

pkg/ingester/ingester_ingest_storage_test.go

@@ -1143,180 +1143,6 @@ func createTestIngesterWithIngestStorage(
 	return ingester, kafkaCluster, prw
 }
 
-func TestIngester_ActiveSeriesPurgeWithKafkaTimestamps(t *testing.T) {
-	var (
-		ctx         = context.Background()
-		cfg         = defaultIngesterTestConfig(t)
-		reg         = prometheus.NewRegistry()
-		user1       = "user-1"
-		user2       = "user-2"
-		baseTime    = time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
-		idleTimeout = 20 * time.Minute
-	)
-
-	cfg.ActiveSeriesMetrics.Enabled = true
-	cfg.ActiveSeriesMetrics.IdleTimeout = idleTimeout
-	// Use a long update period so that the wall-clock ticker never fires during this test.
-	// Only the inline Kafka-time-based purge trigger should update active series.
-	cfg.ActiveSeriesMetrics.UpdatePeriod = 10 * time.Minute
-
-	overrides := validation.NewOverrides(defaultLimitsTestConfig(), nil)
-	ingester, _, _ := createTestIngesterWithIngestStorage(t, &cfg, overrides, nil, reg, util_test.NewTestingLogger(t))
-
-	partitionID, err := ingest.IngesterPartitionID(cfg.IngesterRing.InstanceID)
-	require.NoError(t, err)
-
-	kafkaClient, err := kgo.NewClient(
-		kgo.SeedBrokers(cfg.IngestStorageConfig.KafkaConfig.Address...),
-		kgo.RecordPartitioner(kgo.ManualPartitioner()),
-	)
-	require.NoError(t, err)
-	t.Cleanup(kafkaClient.Close)
-
-	produceRecord := func(tenantID string, seriesName string, kafkaTimestamp time.Time) {
-		wreq := &mimirpb.WriteRequest{
-			Timeseries: []mimirpb.PreallocTimeseries{{
-				TimeSeries: &mimirpb.TimeSeries{
-					Labels:  mimirpb.FromLabelsToLabelAdapters(labels.FromStrings(model.MetricNameLabel, seriesName)),
-					Samples: []mimirpb.Sample{{TimestampMs: kafkaTimestamp.UnixMilli(), Value: 1}},
-				},
-			}},
-			Source: mimirpb.API,
-		}
-		data, err := wreq.Marshal()
-		require.NoError(t, err)
-
-		results := kafkaClient.ProduceSync(ctx, &kgo.Record{
-			Topic:     cfg.IngestStorageConfig.KafkaConfig.Topic,
-			Key:       []byte(tenantID),
-			Value:     data,
-			Headers:   []kgo.RecordHeader{ingest.RecordVersionHeader(1)},
-			Partition: partitionID,
-			Timestamp: kafkaTimestamp,
-		})
-		require.NoError(t, results.FirstErr())
-	}
-
-	require.NoError(t, services.StartAndAwaitRunning(ctx, ingester))
-	t.Cleanup(func() {
-		require.NoError(t, services.StopAndAwaitTerminated(ctx, ingester))
-	})
-
-	// Phase 1: Produce 20 series for user1 at T0.
-	// The first record triggers an inline purge (Kafka time jumps from 0 to T0 > UpdatePeriod),
-	// but since no series exist yet, it has no effect on counts.
-	for i := 0; i < 20; i++ {
-		produceRecord(user1, fmt.Sprintf("series_%d", i), baseTime)
-	}
-
-	// Phase 2: Produce 20 series for user2 at T0+21min (past idle timeout for user1).
-	// The first record triggers an inline purge because Kafka time advanced by 21min > UpdatePeriod (10min).
-	// This purge runs with now=T0+21min, which purges user1's series (last seen at T0, idle for 21min > 20min timeout).
-	// After the purge, user2 has 1 active series (just added), but user1's 20 series are purged.
-	futureTime := baseTime.Add(idleTimeout + 1*time.Minute)
-	for i := 0; i < 20; i++ {
-		produceRecord(user2, fmt.Sprintf("series_%d", i), futureTime)
-	}
-
-	// Phase 3: Produce one more record at T0+32min to trigger another inline purge
-	// (advances by 11min from T0+21min > UpdatePeriod of 10min). This purge updates the
-	// active series counts to reflect all 20 of user2's series.
-	finalTime := futureTime.Add(cfg.ActiveSeriesMetrics.UpdatePeriod + 1*time.Minute)
-	produceRecord(user2, "series_final", finalTime)
-
-	// Wait for all records to be consumed and the inline purge to update metrics.
-	// user1's series should be purged (0 active), user2 should have 21 active series.
-	test.Poll(t, 10*time.Second, nil, func() interface{} {
-		return testutil.GatherAndCompare(reg, strings.NewReader(fmt.Sprintf(`
-			# HELP cortex_ingester_active_series Number of currently active series per user.
-			# TYPE cortex_ingester_active_series gauge
-			cortex_ingester_active_series{user="%s"} 21
-		`, user2)), "cortex_ingester_active_series")
-	})
-}
-
-func TestIngester_ActiveSeriesLoadingWithKafkaTimestamps(t *testing.T) {
-	var (
-		ctx         = context.Background()
-		cfg         = defaultIngesterTestConfig(t)
-		reg         = prometheus.NewRegistry()
-		user1       = "user-1"
-		baseTime    = time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC)
-		idleTimeout = 20 * time.Minute
-	)
-
-	cfg.ActiveSeriesMetrics.Enabled = true
-	cfg.ActiveSeriesMetrics.IdleTimeout = idleTimeout
-	cfg.ActiveSeriesMetrics.UpdatePeriod = 10 * time.Minute
-
-	overrides := validation.NewOverrides(defaultLimitsTestConfig(), nil)
-	ingester, _, _ := createTestIngesterWithIngestStorage(t, &cfg, overrides, nil, reg, util_test.NewTestingLogger(t))
-
-	partitionID, err := ingest.IngesterPartitionID(cfg.IngesterRing.InstanceID)
-	require.NoError(t, err)
-
-	kafkaClient, err := kgo.NewClient(
-		kgo.SeedBrokers(cfg.IngestStorageConfig.KafkaConfig.Address...),
-		kgo.RecordPartitioner(kgo.ManualPartitioner()),
-	)
-	require.NoError(t, err)
-	t.Cleanup(kafkaClient.Close)
-
-	produceRecord := func(tenantID string, seriesName string, kafkaTimestamp time.Time) {
-		wreq := &mimirpb.WriteRequest{
-			Timeseries: []mimirpb.PreallocTimeseries{{
-				TimeSeries: &mimirpb.TimeSeries{
-					Labels:  mimirpb.FromLabelsToLabelAdapters(labels.FromStrings(model.MetricNameLabel, seriesName)),
-					Samples: []mimirpb.Sample{{TimestampMs: kafkaTimestamp.UnixMilli(), Value: 1}},
-				},
-			}},
-			Source: mimirpb.API,
-		}
-		data, err := wreq.Marshal()
-		require.NoError(t, err)
-
-		results := kafkaClient.ProduceSync(ctx, &kgo.Record{
-			Topic:     cfg.IngestStorageConfig.KafkaConfig.Topic,
-			Key:       []byte(tenantID),
-			Value:     data,
-			Headers:   []kgo.RecordHeader{ingest.RecordVersionHeader(1)},
-			Partition: partitionID,
-			Timestamp: kafkaTimestamp,
-		})
-		require.NoError(t, results.FirstErr())
-	}
-
-	require.NoError(t, services.StartAndAwaitRunning(ctx, ingester))
-	t.Cleanup(func() {
-		require.NoError(t, services.StopAndAwaitTerminated(ctx, ingester))
-	})
-
-	// Produce a record at baseTime. The loading metric should be 1 after the inline purge.
-	produceRecord(user1, "series_0", baseTime)
-
-	test.Poll(t, 10*time.Second, nil, func() interface{} {
-		return testutil.GatherAndCompare(reg, strings.NewReader(`
-			# HELP cortex_ingester_active_series_loading 1 if active series counts are still warming up and may be underreported, 0 once they are accurate.
-			# TYPE cortex_ingester_active_series_loading gauge
-			cortex_ingester_active_series_loading 1
-		`), "cortex_ingester_active_series_loading")
-	})
-
-	// Produce a record at baseTime + idleTimeout + UpdatePeriod + 1min.
-	// This advances Kafka time past the idle timeout from the first record's timestamp,
-	// and also triggers an inline purge (> UpdatePeriod from last purge).
-	futureTime := baseTime.Add(idleTimeout + cfg.ActiveSeriesMetrics.UpdatePeriod + 1*time.Minute)
-	produceRecord(user1, "series_1", futureTime)
-
-	test.Poll(t, 10*time.Second, nil, func() interface{} {
-		return testutil.GatherAndCompare(reg, strings.NewReader(`
-			# HELP cortex_ingester_active_series_loading 1 if active series counts are still warming up and may be underreported, 0 once they are accurate.
-			# TYPE cortex_ingester_active_series_loading gauge
-			cortex_ingester_active_series_loading 0
-		`), "cortex_ingester_active_series_loading")
-	})
-}
-
 // BenchmarkIngester_ReplayFromKafka tests the ingester replaying records from Kafka at startup.
 // Each scenario is derived from analysis of 1 partition in some production clusters in Grafana Cloud,
 // using "kafkatool dump analyse". The fixture generator simulates the WriteRequest patterns

pkg/ingester/ingester_metrics_update.go

@@ -39,8 +39,9 @@ func (i *Ingester) metricsUpdaterServiceRunning(ctx context.Context) error {
 	defer ingestionRateTicker.Stop()
 
 	var activeSeriesTickerChan <-chan time.Time
-	if i.cfg.ActiveSeriesMetrics.Enabled && !i.cfg.IngestStorageConfig.Enabled {
-		i.activeSeriesStartMs.Store(time.Now().UnixMilli())
+	var activeSeriesStartedAt time.Time
+	if i.cfg.ActiveSeriesMetrics.Enabled {
+		activeSeriesStartedAt = time.Now()
 		t := time.NewTicker(i.cfg.ActiveSeriesMetrics.UpdatePeriod)
 		activeSeriesTickerChan = t.C
 		defer t.Stop()
@@ -64,7 +65,12 @@ func (i *Ingester) metricsUpdaterServiceRunning(ctx context.Context) error {
 			}
 			i.tsdbsMtx.RUnlock()
 		case <-activeSeriesTickerChan:
-			i.updateActiveSeries(i.activeSeriesNow())
+			now := time.Now()
+			if !activeSeriesStartedAt.IsZero() && now.Sub(activeSeriesStartedAt) >= i.cfg.ActiveSeriesMetrics.IdleTimeout {
+				i.metrics.activeSeriesLoading.Set(0)
+				activeSeriesStartedAt = time.Time{} // Only flip once.
+			}
+			i.updateActiveSeries(now)
 		case <-usageStatsUpdateTicker.C:
 			i.updateUsageStats()
 		case <-limitMetricsUpdateTicker.C:
@@ -75,23 +81,7 @@ func (i *Ingester) metricsUpdaterServiceRunning(ctx context.Context) error {
 	}
 }
 
-func (i *Ingester) activeSeriesNow() time.Time {
-	if i.cfg.IngestStorageConfig.Enabled {
-		if tsMs := i.latestKafkaRecordTimestamp.Load(); tsMs > 0 {
-			return time.UnixMilli(tsMs)
-		}
-	}
-	return time.Now()
-}
-
 func (i *Ingester) updateActiveSeries(now time.Time) {
-	if startMs := i.activeSeriesStartMs.Load(); startMs > 0 &&
-		now.UnixMilli()-startMs >= i.cfg.ActiveSeriesMetrics.IdleTimeout.Milliseconds() {
-		// Active series counts has passed the warm up.
-		// Mark the loading phase off to signal the counts are now accurate.
-		i.metrics.activeSeriesLoading.Set(0)
-	}
-
 	for _, userID := range i.getTSDBUsers() {
 		userDB := i.getTSDB(userID)
 		if userDB == nil {

pkg/ingester/ingester_push.go

@@ -24,7 +24,6 @@ import (
 	"github.com/grafana/mimir/pkg/ingester/activeseries"
 	"github.com/grafana/mimir/pkg/mimirpb"
 	mimir_storage "github.com/grafana/mimir/pkg/storage"
-	"github.com/grafana/mimir/pkg/storage/ingest"
 	"github.com/grafana/mimir/pkg/util/globalerror"
 	util_log "github.com/grafana/mimir/pkg/util/log"
 	"github.com/grafana/mimir/pkg/util/spanlogger"
@@ -284,8 +283,7 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, req *mimirpb.WriteReques
 	spanlog.DebugLog("event", "acquired append lock")
 
 	var (
-		startAppend          = time.Now()
-		appendWallClockStart = startAppend
+		startAppend = time.Now()
 
 		// Keep track of some stats which are tracked only if the samples will be
 		// successfully committed
@@ -389,10 +387,6 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, req *mimirpb.WriteReques
 		)
 	)
 
-	if ts, ok := ingest.RecordTimestampFromContext(ctx); ok {
-		startAppend = ts
-	}
-
 	// Walk the samples, appending them to the users database
 	app := db.Appender(ctx).(extendedAppender)
 	spanlog.DebugLog("event", "got appender for timeseries", "series", len(req.Timeseries))
@@ -426,7 +420,7 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, req *mimirpb.WriteReques
 	}
 
 	// At this point all samples have been added to the appender, so we can track the time it took.
-	i.metrics.appenderAddDuration.Observe(time.Since(appendWallClockStart).Seconds())
+	i.metrics.appenderAddDuration.Observe(time.Since(startAppend).Seconds())
 
 	spanlog.DebugLog(
 		"event", "start commit",
@@ -460,32 +454,6 @@ func (i *Ingester) PushWithCleanup(ctx context.Context, req *mimirpb.WriteReques
 	appendedSamplesStats.Inc(int64(stats.succeededSamplesCount))
 	appendedExemplarsStats.Inc(int64(stats.succeededExemplarsCount))
 
-	if ts, ok := ingest.RecordTimestampFromContext(ctx); ok {
-		tsMs := ts.UnixMilli()
-		for {
-			cur := i.latestKafkaRecordTimestamp.Load()
-			if tsMs <= cur || i.latestKafkaRecordTimestamp.CompareAndSwap(cur, tsMs) {
-				break
-			}
-		}
-
-		i.activeSeriesStartMs.CompareAndSwap(0, tsMs)
-
-		if i.cfg.ActiveSeriesMetrics.Enabled {
-			updatePeriodMs := i.cfg.ActiveSeriesMetrics.UpdatePeriod.Milliseconds()
-			for {
-				lastUpdate := i.lastKafkaActiveSeriesUpdate.Load()
-				if tsMs-lastUpdate < updatePeriodMs {
-					break
-				}
-				if i.lastKafkaActiveSeriesUpdate.CompareAndSwap(lastUpdate, tsMs) {
-					i.updateActiveSeries(ts)
-					break
-				}
-			}
-		}
-	}
-
 	group := i.activeGroups.UpdateActiveGroupTimestamp(userID, validation.GroupLabel(i.limits, userID, req.Timeseries), startAppend)
 
 	i.updateMetricsFromPushStats(userID, group, &stats, req.Source, db, i.metrics.discarded)