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

Backport of #15036 to `r387`.

On `r387` the ingester code hasn't been split into separate files
(`ingester_compaction.go`, `ingester_metrics_update.go`,
`ingester_push.go`) yet, so the automated cherry-pick failed with
conflicts. The equivalent changes were applied directly to
`ingester.go`.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes affect ingester active-series purging/loading timing and
early-compaction decision inputs, which can alter memory/metric behavior
under ingest-storage replay or Kafka lag; scope is contained to
ingester/ingest-storage plumbing.
> 
> **Overview**
> This PR reverts the ingest-storage behavior that drove *active series
purging/loading and related compaction heuristics* off Kafka record
timestamps, returning those paths to wall-clock time.
> 
> It removes the ingester’s Kafka-timestamp tracking fields and the
inline, Kafka-time-based `updateActiveSeries` triggering during record
consumption, and updates the ingest pusher to stop propagating record
timestamps via context. Related tests and the changelog entry covering
Kafka-timestamp-based active series behavior are removed/updated
accordingly.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
966ef95. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: colega

CHANGELOG.md

@@ -139,7 +139,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: Allow configuring multiple Kafka seed brokers via `-ingest-storage.kafka.address` (comma-separated). #14328
 * [ENHANCEMENT] MQE: Add experimental support for eliminating selectors that are a subset of another selector. Enable with `-querier.mimir-query-engine.enable-subset-selector-elimination=true`. #14456 #14457 #14546 #14559 #14561 #14621

pkg/ingester/ingester.go

@@ -378,22 +378,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
 }
@@ -880,8 +864,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()
@@ -905,7 +890,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:
@@ -916,23 +906,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 {
@@ -1387,8 +1361,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
@@ -1492,10 +1465,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))
@@ -1529,7 +1498,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",
@@ -1563,32 +1532,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)
@@ -3383,10 +3326,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