Bug: Cost attribution tracking can diverge on early head compaction

[tcard]

What is the bug?

Active series tracking keeps series's refs, while cost attribution works by labelset.

When removing a series, active series tracking looks up the labelset for a ref from the TSDB, then decrements the cost attribution tracking:

mimir/pkg/ingester/activeseries/active_series.go

Lines 636 to 640 in af48ad2

	if err := idx.Series(ref, &buf, nil); err != nil {
	s.activeSeriesAttributionFailureCounter.Add(1)
	} else {
	s.cat.Decrement(buf.Labels(), entry.numNativeHistogramBuckets)
	}

mimir/pkg/ingester/activeseries/active_series.go

Lines 696 to 700 in af48ad2

	if err := idx.Series(ref, &buf, nil); err != nil {
	s.activeSeriesAttributionFailureCounter.Add(1)
	} else {
	s.cat.Decrement(buf.Labels(), entry.numNativeHistogramBuckets)
	}

Normally, this works fine, because a series will be removed from active series after 20 minutes, while it will be kept in the TSDB head for longer than that (2 hours).

But if early head compaction kicks in, and there are no samples for the series in the new head (which is expected to happen: early head compaction is often accompanied by series resharding), then the lookup later will fail.

To avoid this, early head compaction is supposed to purge active series up to the moment it compacts before removing the series from the head, and it does:

mimir/pkg/ingester/ingester_compaction.go

Lines 561 to 564 in af48ad2

	// Purge active series to get accurate count
	idx := db.Head().MustIndex()
	db.activeSeries.Purge(now, idx)
	_ = idx.Close()

mimir/pkg/ingester/ingester_compaction.go

Lines 464 to 467 in af48ad2

	// Purge the active series so that the next call to Active() will return the up-to-date count.
	idx := db.Head().MustIndex()
	db.activeSeries.Purge(now, idx)
	idx.Close()

But we've still seen cases in which this doesn't work for some reason, and series count from cost attribution trackers diverge from active series count, while cortex_ingester_attributed_active_series_failure spikes.

A hypothesis is that this is due to Kafka consumer lag: a series is active if it received a sample in the last 20 minutes (wall-clock time), but then compaction removes it from the head based on the samples's timestamp, which may be significantly older under high Kafka consumer lag. So, active series purge may still consider a series active, while their latest sample is older than 20 minutes.

How to reproduce it?

This test reproduces the hypothesis, although it's not clear it's the actual issue:

Details

func TestIngester_CostAttribution_ActiveSeriesLeaksWhenSamplesLagBehindWallClock(t *testing.T) {
	const idleTimeout = 20 * time.Minute

	ctx := context.Background()
	ctxWithUser := user.InjectOrgID(ctx, userID)
	now := time.Now()

	// The sample lags wall-clock by more than the idle timeout, but it is ingested now.
	laggingSampleTS := now.Add(-idleTimeout - 5*time.Minute)

	cfg := defaultIngesterTestConfig(t)
	cfg.ActiveSeriesMetrics.Enabled = true
	cfg.ActiveSeriesMetrics.IdleTimeout = idleTimeout
	cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = time.Hour // Only trigger compaction manually.
	cfg.BlocksStorageConfig.TSDB.EarlyHeadCompactionMinInMemorySeries = 1
	cfg.BlocksStorageConfig.TSDB.EarlyHeadCompactionMinEstimatedSeriesReductionPercentage = 0

	// Enable cost attribution by the "team" label. PastGracePeriod defaults to 0 (disabled),
	// so the lagging sample is accepted.
	limits := defaultLimitsTestConfig()
	limits.MaxCostAttributionCardinality = 100
	limits.CostAttributionBaseTrackers = costattributionmodel.TrackerConfigs{
		costattributionmodel.DefaultTrackerName: {Labels: costattributionmodel.Labels{{Input: "team", Output: "team"}}},
	}
	limits.CostAttributionBaseTrackers.Canonicalize()
	limits.ComputeCostAttributionConfigHash()
	overrides := validation.NewOverrides(limits, nil)

	reg := prometheus.NewRegistry()
	caReg := prometheus.NewRegistry()
	cam, err := costattribution.NewManager(5*time.Second, 10*time.Second, log.NewNopLogger(), overrides, reg, caReg)
	require.NoError(t, err)

	ingester, r, err := prepareIngesterWithBlockStorageOverridesAndCostAttribution(t, cfg, overrides, nil, "", "", reg, cam)
	require.NoError(t, err)
	startAndWaitHealthy(t, ingester, r)

	// Push a series whose sample is already older than the idle timeout, ingested now.
	require.NoError(t, pushSeriesToIngester(ctxWithUser, t, ingester, []util_test.Series{{
		Labels:  labels.FromStrings(model.MetricNameLabel, "metric_1", "team", "foo"),
		Samples: []util_test.Sample{{TS: laggingSampleTS.UnixMilli(), Val: 1}},
	}}))

	// It is active (ingested now) and attributed.
	require.NoError(t, testutil.GatherAndCompare(caReg, strings.NewReader(`
		# HELP cortex_ingester_attributed_active_series The total number of active series per user and attribution.
		# TYPE cortex_ingester_attributed_active_series gauge
		cortex_ingester_attributed_active_series{team="foo",tenant="1",tracker="cost-attribution"} 1
	`), "cortex_ingester_attributed_active_series"))

	// Trigger early head compaction now. It purges active series first (head intact), but our
	// series is still active by wall-clock, so it is NOT decremented. Then it truncates the head
	// at now-idleTimeout (sample-time), dropping the series (its only sample is older than that).
	ingester.compactBlocksToReduceInMemorySeries(ctx, now)
	require.Equal(t, uint64(0), ingester.getTSDB(userID).Head().NumSeries())

	// Time passes; the series finally becomes inactive by wall-clock. The active-series update
	// purges it, but the head index can no longer resolve its ref.
	ingester.updateActiveSeries(now.Add(idleTimeout + time.Minute))

	// Plain active self-corrects to 0.
	active, _, _, _ := ingester.getTSDB(userID).activeSeries.Active()
	assert.Equal(t, 0, active)

	// The failed decrement was recorded.
	assert.Equal(t, float64(1), testutil.ToFloat64(ingester.metrics.attributedActiveSeriesFailuresPerUser.WithLabelValues(userID)))

	// BUG: cost-attribution active series leaks — it should be 0 now that the series is gone.
	// The skipped decrement leaves it at 1. This assertion FAILS on current main.
	assert.NoError(t, testutil.GatherAndCompare(caReg, strings.NewReader(``), "cortex_ingester_attributed_active_series"))
}

What did you think would happen?

Cost attribution trackers should be kept in sync with active series tracking, even when the series disappear early from the head.

What was your environment?

Mimir r404

Any additional context to share?

Early head compactions:

Divergence between series counts:

cortex_ingester_attributed_active_series_failure spike:

[colega]

A hypothesis is that this is due to Kafka consumer lag: a series is active if it received a sample in the last 20 minutes (wall-clock time), but then compaction removes it from the head based on the samples's timestamp, which may be significantly older under high Kafka consumer lag. So, active series purge may still consider a series active, while their latest sample is older than 20 minutes.

If this hypothesis is true, then there's something really wrong in how active series are removed during early head compaction. Sample's timestamp has nothing to do with being active series.

This is not only caused by Kafka lag, but also by OOO samples (especially very old ones).

[tcard]

A hypothesis is that this is due to Kafka consumer lag: a series is active if it received a sample in the last 20 minutes (wall-clock time), but then compaction removes it from the head based on the samples's timestamp, which may be significantly older under high Kafka consumer lag. So, active series purge may still consider a series active, while their latest sample is older than 20 minutes.

If this hypothesis is true, then there's something really wrong in how active series are removed during early head compaction. Sample's timestamp has nothing to do with being active series.

This is not only caused by Kafka lag, but also by OOO samples (especially very old ones).

@colega See this repro/regression test from #16368:

mimir/pkg/ingester/ingester_early_compaction_test.go

Line 1832 in 7b3fece

func TestIngester_CostAttribution_ActiveSeriesLeaksWhenSamplesLagBehindWallClock(t *testing.T) {

The issue is definitely there under such scenario, but I'm not sure this scenario is the only way this may happen (or what happened in the incident mentioned in this issue).