metrics-generator: Add per-label limiter to control cardinality (#6414)

* metrics-generator: Add per-label cardinality sanitizer

Adds a per-label cardinality cap that replaces only the high-cardinality
label value with "__overflow__" while preserving all other dimensions.

Implementation:
- CardinalitySanitizer: per-label HLL sketch tracking with cached
  overLimit flag updated every 15s. Always inserts original value hash
  to prevent estimate oscillation.
- ChainSanitizer: composes sanitizers in order (DRAIN first, then
  cardinality sanitizer as hard backstop).
- New config field: max_cardinality_per_label (0 = disabled).

* metrics-generator: Add docs for per-label cardinality sanitizer

Rename overflow value to __cardinality_overflow__ for clarity and add
troubleshooting docs explaining the per-label cardinality limit, its
PromQL query, configuration, and how it interacts with the series and
entity limiters.

* metrics-generator: Move max_cardinality_per_label to per-tenant override

Move max_cardinality_per_label from static registry config to a
per-tenant runtime override, consistent with max_active_series and
max_active_entities. The CardinalitySanitizer now receives the
Overrides interface and refreshes the cached value every maintenance
tick (15s), allowing operators to tune it per tenant without restart.

* move from piggybacking on sanitizer and have a standalone per label limiter

sanitizer is there to sanitize the labels, so we shouldn't piggyback and rely
on the chained sanitizer hack to ensure per label limiter runs after all sanitizer.

we create a new LabelLimiter interface and run it after Sanitizer and decouple from Sanitizer

* extract maxSeriesFunc and maxEntityFunc to type

* test the overflow -> recovery -> overflow liftcycle

* make the test more strict

* rework docs

* track label_names and add a demand estimate metric as well

* Add tests for MetricsGeneratorMaxCardinalityPerLabel override

* optimize hot path

* Update TestPerLabelLimiter_ConcurrentAccess test

* Add BenchmarkPerLabelLimiter to check perf and allocs

* Add CHANGELOG.md

* Update config index page

* drop noopLabelLimiter from tests

* always run s.doPeriodicMaintenance first to refresh the config value

CHANGELOG.md

@@ -11,6 +11,8 @@
 * [FEATURE] Add span_multiplier_key to overrides. This allows tenants to specify the attribute key used for span multiplier values to compensate for head-based sampling. [#6260](https://github.com/grafana/tempo/pull/6260) (@carles-grafana)
 * [FEATURE] **BREAKING CHANGE** Optimize TraceQL AST by rewriting conditions on the same attribute to their array equivalent [#6353](https://github.com/grafana/tempo/pull/6353) (@stoewer)
   Slightly changes the array matching semantics of != and !~ operators and introduces stricter rules for regex literals.
+* [FEATURE] metrics-generator: Add per-label limiter to control cardinality [#6414](https://github.com/grafana/tempo/pull/6414) (@electron0zero)
+  Adds `max_cardinality_per_label` per per-tenant override and new metrics to estimate per label cardinality demand estimate.
 * [ENHANCEMENT] Add new alerts and runbooks entries [#6276](https://github.com/grafana/tempo/pull/6276) (@javiermolinar)
 * [ENHANCEMENT] Double the maximum number of dedicated string columns in vParquet5 and update tempo-cli to determine the optimum number for the data [#6282](https://github.com/grafana/tempo/pull/6282) (@mdisibio)
 * [ENHANCEMENT] Improve attribute truncating observability [#6400](https://github.com/grafana/tempo/pull/6400) (@javiermolinar)

docs/sources/tempo/configuration/_index.md

@@ -1959,6 +1959,18 @@ overrides:
       # This setting only applies when limiter_type is set to "entity".
       [max_active_entities: <int>]
 
+      # Maximum number of distinct values any single label can have. When a label exceeds the
+      # configured threshold, all new label value is replaced with `__cardinality_overflow__`.
+      # All other labels that is under the limit are preserved
+      # If the limit is reached, no new label values will be added to the limit label.
+      # The amount of limited entities can be observed with the metric:
+      #   tempo_metrics_generator_registry_label_values_limited_total
+      # To view the estimated cardinality demand per label:
+      #   tempo_metrics_generator_registry_label_cardinality_demand_estimate
+      # This setting only applies when limiter_type is set to "entity".
+      # A value of 0 disables this limiter.
+      [max_cardinality_per_label:  <uint64> | default = 0]
+
       # Per-user configuration of the collection interval. A value of 0 means the global default is
       # used set in the metrics_generator config block.
       [collection_interval: <duration>]

docs/sources/tempo/troubleshooting/metrics-generator.md

@@ -121,6 +121,65 @@ overrides:
       max_active_entities: 0
 ```
 
+### Per-label cardinality limiting
+
+The per-label cardinality limiter caps the number of distinct values any single label can have. When a label exceeds the configured threshold, its value is replaced with `__cardinality_overflow__` while all other labels that are under the limit are preserved.
+
+For example, if the `url` label exceeds the cardinality limit:
+
+Before:
+```
+{service="foo", method="GET", url="/users/1"}
+{service="foo", method="GET", url="/users/2"}
+{service="foo", method="GET", url="/users/3"}
+...
+```
+
+After:
+```
+{service="foo", method="GET", url="__cardinality_overflow__"}
+```
+
+Once the limiter kicks in, new `url` values are replaced with `__cardinality_overflow__`. Labels that remain under the limit, like `method`, are unaffected.
+
+To detect if per-label cardinality limiting is active:
+
+```promql
+sum by (tenant, label_name) (rate(tempo_metrics_generator_registry_label_values_limited_total{}[5m]))
+```
+
+To view the estimated cardinality demand per label:
+
+```promql
+tempo_metrics_generator_registry_label_cardinality_demand_estimate{}
+```
+
+Use this metric to identify which labels have high cardinality, how far they exceed the configured limit, and to choose an appropriate
+`max_cardinality_per_label` value. To observe actual demand before enforcing a limit, deploy with a high `max_cardinality_per_label` value first.
+
+Configure the per-label cardinality limit:
+
+```yaml
+overrides:
+  defaults:
+    metrics_generator:
+      max_cardinality_per_label: 0
+```
+
+A value of `0` (default) disables the limit.
+
+This setting works alongside both active series limiting (`max_active_series`) and entity-based limiting (`max_active_entities`).
+The per-label limiter runs during label construction, preventing any single high-cardinality label from consuming the entire active series or entity budget.
+
+The per-label limiter uses HyperLogLog sketches to estimate cardinality, so the limit is approximate with a 3.25% standard error. Estimates are
+re-evaluated every few seconds, which means there may be a brief delay between a label crossing the threshold and the limiter taking effect.
+
+If a high-cardinality label's cardinality is later reduced (for example, by fixing instrumentation), the limiter automatically recovers
+and allows label values through again. No configuration changes are needed.
+
+Recovery is not immediate. The limiter tracks cardinality over a sliding window (based on the registry's `stale_duration`). It takes at least that 
+duration or longer for existing high-cardinality labels to age out before the label values are allowed through again.
+
 ### Estimate active series demand
 
 When the active series limit is reached, the `tempo_metrics_generator_registry_active_series` metric no longer reflects the true demand. Use the `tempo_metrics_generator_registry_active_series_demand_estimate` metric to estimate what the active series count would be without the limit:

modules/generator/localentitylimiter/local_entity_limiter.go

@@ -52,11 +52,13 @@ func newMetrics(reg prometheus.Registerer) localEntityLimiterMetrics {
 
 var metrics = newMetrics(prometheus.DefaultRegisterer)
 
+type maxEntityFunc func(tenant string) uint32
+
 type LocalEntityLimiter struct {
 	tenant             string
 	entityActiveSeries map[uint64]uint32
 	mtx                sync.Mutex
-	maxEntityFunc      func(tenant string) uint32
+	maxEntityFunc      maxEntityFunc
 	limitLogger        *tempo_log.RateLimitedLogger
 
 	metricTotalEntitiesLimited prometheus.Counter
@@ -69,11 +71,11 @@ type LocalEntityLimiter struct {
 	overflowEntityHash uint64
 }
 
-func New(maxEntityFunc func(tenant string) uint32, tenant string, limitLogger *tempo_log.RateLimitedLogger) *LocalEntityLimiter {
+func New(maxEntityF maxEntityFunc, tenant string, limitLogger *tempo_log.RateLimitedLogger) *LocalEntityLimiter {
 	l := &LocalEntityLimiter{
 		tenant:             tenant,
 		entityActiveSeries: make(map[uint64]uint32),
-		maxEntityFunc:      maxEntityFunc,
+		maxEntityFunc:      maxEntityF,
 		limitLogger:        limitLogger,
 
 		metricTotalEntitiesLimited: metrics.totalEntitiesLimited.WithLabelValues(tenant),

modules/generator/localserieslimiter/local_series_limiter.go

@@ -50,10 +50,12 @@ func newMetrics(reg prometheus.Registerer) localSeriesLimiterMetrics {
 
 var metrics = newMetrics(prometheus.DefaultRegisterer)
 
+type maxSeriesFunc func(tenant string) uint32
+
 type LocalSeriesLimiter struct {
 	tenant                   string
 	activeSeries             atomic.Uint32
-	maxSeriesFunc            func(tenant string) uint32
+	maxSeriesFunc            maxSeriesFunc
 	limitLogger              *tempo_log.RateLimitedLogger
 	metricTotalSeriesLimited prometheus.Counter
 	metricActiveSeries       prometheus.Gauge
@@ -67,10 +69,10 @@ type LocalSeriesLimiter struct {
 
 var _ registry.Limiter = (*LocalSeriesLimiter)(nil)
 
-func New(maxSeriesFunc func(tenant string) uint32, tenant string, limitLogger *tempo_log.RateLimitedLogger) *LocalSeriesLimiter {
+func New(maxSeriesF maxSeriesFunc, tenant string, limitLogger *tempo_log.RateLimitedLogger) *LocalSeriesLimiter {
 	l := &LocalSeriesLimiter{
 		tenant:                   tenant,
-		maxSeriesFunc:            maxSeriesFunc,
+		maxSeriesFunc:            maxSeriesF,
 		limitLogger:              limitLogger,
 		metricTotalSeriesLimited: metrics.totalSeriesLimited.WithLabelValues(tenant),
 		metricActiveSeries:       metrics.activeSeries.WithLabelValues(tenant),

modules/generator/overrides_test.go

@@ -156,6 +156,10 @@ func (m *mockOverrides) MetricsGeneratorSpanNameSanitization(string) string {
 	return ""
 }
 
+func (m *mockOverrides) MetricsGeneratorMaxCardinalityPerLabel(string) uint64 {
+	return 0
+}
+
 // MetricsGeneratorProcessorSpanMetricsEnableTargetInfo enables target_info metrics
 func (m *mockOverrides) MetricsGeneratorProcessorSpanMetricsEnableTargetInfo(string) (bool, bool) {
 	spanMetricsEnableTargetInfo := m.spanMetricsEnableTargetInfo

modules/generator/registry/builder.go

@@ -33,8 +33,9 @@ func (p *safeBuilderPool) Put(builder *labels.Builder) {
 var builderPool = newSafeBuilderPool()
 
 type labelBuilder struct {
-	builder   *labels.Builder
-	sanitizer Sanitizer
+	builder         *labels.Builder
+	sanitizer       Sanitizer
+	perLabelLimiter LabelLimiter
 
 	maxLabelNameLength  int
 	maxLabelValueLength int
@@ -42,11 +43,12 @@ type labelBuilder struct {
 
 var _ LabelBuilder = (*labelBuilder)(nil)
 
-func NewLabelBuilder(maxLabelNameLength int, maxLabelValueLength int, sanitizer Sanitizer) LabelBuilder {
+func NewLabelBuilder(maxLabelNameLength int, maxLabelValueLength int, sanitizer Sanitizer, perLabelLimiter LabelLimiter) LabelBuilder {
 	builder := builderPool.Get()
 	return &labelBuilder{
 		builder:             builder,
 		sanitizer:           sanitizer,
+		perLabelLimiter:     perLabelLimiter,
 		maxLabelNameLength:  maxLabelNameLength,
 		maxLabelValueLength: maxLabelValueLength,
 	}
@@ -63,16 +65,20 @@ func (b *labelBuilder) Add(name, value string) {
 }
 
 func (b *labelBuilder) CloseAndBuildLabels() (labels.Labels, bool) {
-	labels := b.sanitizer.Sanitize(b.builder.Labels())
+	// We always run sanitizer first and then run per label limiter to ensure that
+	// per label limits are always applied after sanitizer.
+	// Pipeline: sanitize labels --> per-label cardinality limit --> entity/series limit
+	lbls := b.sanitizer.Sanitize(b.builder.Labels())
+	lbls = b.perLabelLimiter.Limit(lbls)
 	// it's no longer safe to use the builder after this point, so we drop our
 	// reference to it. this may cause a nil panic if the builder is used after
 	// this point, but it's better than memory corruption.
 	builderPool.Put(b.builder)
 	b.builder = nil
 
-	if !labels.IsValid(model.UTF8Validation) {
-		return labels, false
+	if !lbls.IsValid(model.UTF8Validation) {
+		return lbls, false
 	}
 
-	return labels, true
+	return lbls, true
 }

modules/generator/registry/builder_test.go

@@ -8,7 +8,7 @@ import (
 )
 
 func TestLabelBuilder(t *testing.T) {
-	builder := NewLabelBuilder(0, 0, newTestDrainSanitizer(SpanNameSanitizationDisabled))
+	builder := NewLabelBuilder(0, 0, newTestDrainSanitizer(SpanNameSanitizationDisabled), newTestLabelLimiter())
 	builder.Add("name", "value")
 	lbls, ok := builder.CloseAndBuildLabels()
 
@@ -23,7 +23,7 @@ func TestLabelBuilder(t *testing.T) {
 }
 
 func TestLabelBuilder_MaxLabelNameLength(t *testing.T) {
-	builder := NewLabelBuilder(10, 10, newTestDrainSanitizer(SpanNameSanitizationDisabled))
+	builder := NewLabelBuilder(10, 10, newTestDrainSanitizer(SpanNameSanitizationDisabled), newTestLabelLimiter())
 	builder.Add("name", "very_long_value")
 	builder.Add("very_long_name", "value")
 
@@ -34,7 +34,7 @@ func TestLabelBuilder_MaxLabelNameLength(t *testing.T) {
 }
 
 func TestLabelBuilder_InvalidUTF8(t *testing.T) {
-	builder := NewLabelBuilder(0, 0, newTestDrainSanitizer(SpanNameSanitizationDisabled))
+	builder := NewLabelBuilder(0, 0, newTestDrainSanitizer(SpanNameSanitizationDisabled), newTestLabelLimiter())
 	builder.Add("name", "svc-\xc3\x28") // Invalid UTF-8
 
 	_, ok := builder.CloseAndBuildLabels()
@@ -69,7 +69,7 @@ func (s sanitizerFunc) Sanitize(lbls labels.Labels) labels.Labels {
 func TestLabelBuilder_Sanitizer(t *testing.T) {
 	builder := NewLabelBuilder(0, 0, sanitizerFunc(func(_ labels.Labels) labels.Labels {
 		return labels.FromStrings("name", "sanitized_value")
-	}))
+	}), newTestLabelLimiter())
 	builder.Add("name", "value")
 	lbls, ok := builder.CloseAndBuildLabels()
 

modules/generator/registry/overrides.go

@@ -18,6 +18,7 @@ type Overrides interface {
 	MetricsGeneratorNativeHistogramMaxBucketNumber(userID string) uint32
 	MetricsGeneratorNativeHistogramMinResetDuration(userID string) time.Duration
 	MetricsGeneratorSpanNameSanitization(userID string) string
+	MetricsGeneratorMaxCardinalityPerLabel(userID string) uint64
 }
 
 var _ Overrides = (overrides.Interface)(nil)

modules/generator/registry/per_label_limiter.go

@@ -0,0 +1,183 @@
+package registry
+
+import (
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/cespare/xxhash/v2"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/schema"
+)
+
+const (
+	overflowValue = "__cardinality_overflow__"
+	// demandUpdateInterval controls how often the cardinality estimate from HLL
+	// is refreshed and the overLimit flag and demand gauge are updated.
+	// kept at 15s to limit lock contention with Limit() (hot path) which shares the same mutex
+	demandUpdateInterval = 15 * time.Second
+)
+
+var metricLabelValuesLimited = promauto.NewCounterVec(prometheus.CounterOpts{
+	Namespace: "tempo",
+	Name:      "metrics_generator_registry_label_values_limited_total",
+	Help:      "Total number of times a label value was limited due to exceeding the per-label cardinality limit",
+}, []string{"tenant", "label_name"})
+
+var metricLabelCardinalityDemand = promauto.NewGaugeVec(prometheus.GaugeOpts{
+	Namespace: "tempo",
+	Name:      "metrics_generator_registry_label_cardinality_demand_estimate",
+	Help:      "Estimated cardinality demand (distinct value count) for each label, each tenant",
+}, []string{"tenant", "label_name"})
+
+// maxCardinalityFunc returns the MaxCardinalityPerLabel config value for the tenant.
+type maxCardinalityFunc func(tenant string) uint64
+
+type labelCardinalityState struct {
+	sketch    *Cardinality
+	overLimit bool // cached flag, updated periodically in maintenance tick
+}
+
+// PerLabelLimiter caps the number of distinct values any single label can have.
+// When a label's estimated cardinality exceeds maxCardinality, its value is replaced
+// with '__cardinality_overflow__' while all other labels are preserved.
+//
+// This is conceptually a limiter, not a sanitizer - it enforces a cardinality ceiling
+// rather than normalizing label values (like DrainSanitizer does for span names).
+// It runs in the label-building pipeline after sanitization but before the global
+// entity limiter, making the processing order: sanitize -> per-label limit -> entity limit.
+type PerLabelLimiter struct {
+	mtx                sync.Mutex
+	tenant             string
+	maxCardinalityFunc maxCardinalityFunc
+	maxCardinality     atomic.Uint64 // refreshed on demand update tick, read atomically in Limit() hot path
+
+	labelsState   map[string]*labelCardinalityState
+	staleDuration time.Duration
+
+	demandUpdateChan <-chan time.Time
+	pruneChan        <-chan time.Time
+}
+
+func NewPerLabelLimiter(tenant string, maxCardinalityF maxCardinalityFunc, staleDuration time.Duration) *PerLabelLimiter {
+	pll := &PerLabelLimiter{
+		tenant:             tenant,
+		maxCardinalityFunc: maxCardinalityF,
+		labelsState:        make(map[string]*labelCardinalityState),
+		staleDuration:      staleDuration,
+		demandUpdateChan:   time.Tick(demandUpdateInterval),
+		pruneChan:          time.Tick(removeStaleSeriesInterval),
+	}
+	// init on New, config is refreshed on demand update tick
+	pll.maxCardinality.Store(maxCardinalityF(tenant))
+	return pll
+}
+
+// Limit applies the per-label cardinality limit to the given labels.
+// Labels whose estimated cardinality exceeds the configured max have their
+// value replaced with __cardinality_overflow__.
+func (s *PerLabelLimiter) Limit(lbls labels.Labels) labels.Labels {
+	// do maintenance check as the first thing to ensure maxCardinality
+	// is refreshed from runtime overrides. without this,
+	// a limiter that starts disabled would never be enabled without restart.
+	s.doPeriodicMaintenance()
+
+	// maxCardinality is zero, so limiter is disabled, return labels as is
+	if s.maxCardinality.Load() == 0 {
+		return lbls
+	}
+
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	// Defer builder creation until we actually need to modify a label.
+	// In the common case (no overflow), we avoid the allocations entirely.
+	var builder *labels.Builder
+	lbls.Range(func(l labels.Label) {
+		// skip over the metadata labels
+		if schema.IsMetadataLabel(l.Name) {
+			return
+		}
+
+		state := s.getOrCreateState(l.Name)
+
+		// we always insert the ORIGINAL value to hash even while overflowing,
+		// which prevents the estimate from artificially dropping.
+		// It will make sure that recovery (label going back under limit) only happens when the
+		// actual incoming data has lower cardinality AND the old sketches have been rotated out.
+		//
+		// If we inserted the overflowValue, then the estimate would drop and cause oscillation:
+		// over limit -> Add overflowValue -> estimate drops -> under limit -> real values -> over limit ->...
+		// Insert acquires its own internal mu lock on the sketch.
+		state.sketch.Insert(xxhash.Sum64String(l.Value))
+
+		// we are over the limit, replace label value and capture the metric
+		if state.overLimit {
+			// Lazy init: only create once, so previous Set calls are preserved
+			// when multiple labels overflow in the same series
+			if builder == nil {
+				builder = labels.NewBuilder(lbls)
+			}
+			builder.Set(l.Name, overflowValue)
+			metricLabelValuesLimited.WithLabelValues(s.tenant, l.Name).Inc()
+		}
+	})
+
+	// No labels were limited, return the original labels as is.
+	if builder == nil {
+		return lbls
+	}
+	return builder.Labels()
+}
+
+func (s *PerLabelLimiter) getOrCreateState(labelName string) *labelCardinalityState {
+	state, ok := s.labelsState[labelName]
+	if !ok {
+		state = &labelCardinalityState{
+			sketch: NewCardinality(s.staleDuration, removeStaleSeriesInterval),
+		}
+		s.labelsState[labelName] = state
+	}
+	return state
+}
+
+// doPeriodicMaintenance holds s.mtx while iterating labelsState for both
+// demand updates (every 15s) and pruning (every 5m). This blocks Limit()
+// callers for the duration. In practice this is fast, so it's acceptable.
+// If it becomes a problem, snapshot the labelsState slice under the lock
+// and do sketch operations outside it, and then update the labelsState under lock.
+func (s *PerLabelLimiter) doPeriodicMaintenance() {
+	select {
+	case <-s.demandUpdateChan:
+		// step 1: refresh maxCardinality config from override
+		// fetch once per tick and cache atomically, the limit is the same for all labels in a tenant
+		maxCardinality := s.maxCardinalityFunc(s.tenant)
+		s.maxCardinality.Store(maxCardinality)
+
+		// if the check is disabled, skip the demand update and, exit early.
+		// no data is being inserted into the sketch, so nothing to estimate or publish
+		if maxCardinality == 0 {
+			return
+		}
+
+		// step 2: update estimate and publish demand estimate metric
+		s.mtx.Lock()
+		for labelName, state := range s.labelsState {
+			estimate := state.sketch.Estimate()
+			state.overLimit = estimate > maxCardinality
+			metricLabelCardinalityDemand.WithLabelValues(s.tenant, labelName).Set(float64(estimate))
+		}
+		s.mtx.Unlock()
+	case <-s.pruneChan:
+		s.mtx.Lock()
+		// label names come from config and are mostly stable, so stale entries
+		// in labelsState are unlikely to grow unboundedly, so we don't clean up.
+		for _, state := range s.labelsState {
+			state.sketch.Advance()
+		}
+		s.mtx.Unlock()
+	default:
+	}
+}

modules/generator/registry/per_label_limiter_test.go

@@ -0,0 +1,450 @@
+package registry
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	io_prometheus_client "github.com/prometheus/client_model/go"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/require"
+)
+
+// testMaxCardinality returns a maxCardinalityFunc that always returns the given value.
+func testMaxCardinality(value uint64) maxCardinalityFunc {
+	return func(string) uint64 { return value }
+}
+
+func TestPerLabelLimiter_Disabled(t *testing.T) {
+	s := NewPerLabelLimiter("test", testMaxCardinality(0), 15*time.Minute)
+
+	lbls := labels.FromStrings("__name__", "foo", "method", "GET", "url", "/api/users/123")
+	result := s.Limit(lbls)
+	require.Equal(t, lbls, result)
+}
+
+// TestPerLabelLimiter_RuntimeEnableDisable verifies that toggling max_cardinality_per_label at runtime
+// via overrides takes effect without restarting the generator.
+func TestPerLabelLimiter_RuntimeEnableDisable(t *testing.T) {
+	tenant := "test-runtime-toggle"
+	var maxCardinality atomic.Uint64
+	maxCardinality.Store(0) // start disabled
+
+	s := NewPerLabelLimiter(tenant, func(string) uint64 {
+		return maxCardinality.Load()
+	}, 15*time.Minute)
+
+	// Phase 1: Disabled - all labels pass through
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "m", "url", fmt.Sprintf("/path/%d", i))
+		result := s.Limit(lbls)
+		require.Equal(t, fmt.Sprintf("/path/%d", i), result.Get("url"), "should pass through when disabled")
+	}
+
+	// when disabled, Limit() returns before inserting into sketches, so labelsState
+	// is empty, which means no demand gauge was published as well.
+	triggerDemandUpdate(s)
+	s.mtx.Lock()
+	labelsStateLen := len(s.labelsState)
+	s.mtx.Unlock()
+	require.Equal(t, 0, labelsStateLen, "no label state should exist when disabled")
+
+	// Phase 2: Enable at runtime by changing the override
+	maxCardinality.Store(5)
+	triggerDemandUpdate(s)
+
+	// Push more distinct values - should overflow now
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "m", "url", fmt.Sprintf("/path/%d", i))
+		s.Limit(lbls)
+	}
+	triggerDemandUpdate(s)
+
+	// demand gauge should be published when enabled
+	var g io_prometheus_client.Metric
+	require.NoError(t, metricLabelCardinalityDemand.WithLabelValues(tenant, "url").Write(&g))
+	require.Greater(t, g.GetGauge().GetValue(), float64(5), "demand gauge should be published when enabled")
+
+	lbls := labels.FromStrings("__name__", "m", "url", "/path/999")
+	result := s.Limit(lbls)
+	require.Equal(t, overflowValue, result.Get("url"), "should overflow after runtime enable")
+
+	// capture the demand gauge value before disabling
+	var before io_prometheus_client.Metric
+	require.NoError(t, metricLabelCardinalityDemand.WithLabelValues(tenant, "url").Write(&before))
+	demandBefore := before.GetGauge().GetValue()
+
+	// Phase 3: Disable again at runtime
+	maxCardinality.Store(0)
+	triggerDemandUpdate(s)
+
+	lbls = labels.FromStrings("__name__", "m", "url", "/path/new/10000")
+	result = s.Limit(lbls)
+	require.Equal(t, "/path/new/10000", result.Get("url"), "should pass through after runtime disable")
+
+	// demand gauge should be stale (not updated) after disabling
+	var after io_prometheus_client.Metric
+	require.NoError(t, metricLabelCardinalityDemand.WithLabelValues(tenant, "url").Write(&after))
+	require.Equal(t, demandBefore, after.GetGauge().GetValue(), "demand gauge should not be updated when disabled")
+}
+
+func TestPerLabelLimiter_UnderLimit(t *testing.T) {
+	s := NewPerLabelLimiter("test", testMaxCardinality(100), 15*time.Minute)
+
+	// Insert a few distinct values - well under the limit
+	for i := 0; i < 5; i++ {
+		lbls := labels.FromStrings("__name__", "foo", "method", fmt.Sprintf("m%d", i))
+		result := s.Limit(lbls)
+		// Before the first maintenance tick, overLimit is false (default), so everything passes through
+		require.Equal(t, fmt.Sprintf("m%d", i), result.Get("method"))
+	}
+
+	// Trigger maintenance to update overLimit flags
+	triggerDemandUpdate(s)
+
+	// Still under limit, should pass through
+	lbls := labels.FromStrings("__name__", "foo", "method", "GET")
+	result := s.Limit(lbls)
+	require.Equal(t, "GET", result.Get("method"))
+}
+
+func TestPerLabelLimiter_HighCardinalityOverflows(t *testing.T) {
+	s := NewPerLabelLimiter("test", testMaxCardinality(5), 15*time.Minute)
+
+	// Push distinct url values but few method values
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "http_requests", "method", "GET", "url", fmt.Sprintf("/users/%d", i))
+		s.Limit(lbls)
+	}
+
+	// Trigger maintenance to update overLimit flags
+	triggerDemandUpdate(s)
+
+	// Now the url should overflow but the method should be preserved
+	lbls := labels.FromStrings("__name__", "http_requests", "method", "GET", "url", "/users/999")
+	result := s.Limit(lbls)
+	require.Equal(t, "GET", result.Get("method"), "low-cardinality label should be preserved")
+	require.Equal(t, overflowValue, result.Get("url"), "high-cardinality label should have overflow value")
+	require.Equal(t, "http_requests", result.Get("__name__"), "__name__ should be preserved")
+}
+
+func TestPerLabelLimiter_MultipleHighCardinalityOverflows(t *testing.T) {
+	s := NewPerLabelLimiter("test", testMaxCardinality(5), 15*time.Minute)
+
+	// Push many distinct values for BOTH url and user_id
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "m", "method", "GET", "url", fmt.Sprintf("/p/%d", i), "user_id", fmt.Sprintf("u%d", i))
+		s.Limit(lbls)
+	}
+
+	triggerDemandUpdate(s)
+
+	lbls := labels.FromStrings("__name__", "m", "method", "GET", "url", "/p/999", "user_id", "u999")
+	result := s.Limit(lbls)
+	require.Equal(t, "GET", result.Get("method"), "low-cardinality label preserved")
+	require.Equal(t, overflowValue, result.Get("url"), "high-cardinality url overflows")
+	require.Equal(t, overflowValue, result.Get("user_id"), "high-cardinality user_id overflows")
+}
+
+func TestPerLabelLimiter_MetadataLabelsNeverOverflows(t *testing.T) {
+	s := NewPerLabelLimiter("test", testMaxCardinality(5), 15*time.Minute)
+
+	// Push many distinct values for all metadata labels to exceed the limit
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings(
+			"__name__", fmt.Sprintf("metric_%d", i),
+			"__type__", fmt.Sprintf("type_%d", i),
+			"__unit__", fmt.Sprintf("unit_%d", i),
+			"url", fmt.Sprintf("/path/%d", i),
+		)
+		s.Limit(lbls)
+	}
+
+	triggerDemandUpdate(s)
+
+	lbls := labels.FromStrings("__name__", "metric_999", "__type__", "type_999", "__unit__", "unit_999", "url", "/path/999")
+	result := s.Limit(lbls)
+	require.Equal(t, "metric_999", result.Get("__name__"), "__name__ should never overflow")
+	require.Equal(t, "type_999", result.Get("__type__"), "__type__ should never overflow")
+	require.Equal(t, "unit_999", result.Get("__unit__"), "__unit__ should never overflow")
+	require.Equal(t, overflowValue, result.Get("url"), "non-metadata label should overflow")
+}
+
+// TestPerLabelLimiter_RecoveryAfterOverflow verifies that a label
+// recovers from overflow once the user reduces cardinality and the old
+// high-cardinality sketches rotate out of the sliding window.
+func TestPerLabelLimiter_RecoveryAfterOverflow(t *testing.T) {
+	staleDuration := 15 * time.Minute
+	s := NewPerLabelLimiter("test", testMaxCardinality(5), staleDuration)
+
+	// Phase 1: Push high-cardinality data to trigger overflow
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "http_requests", "url", fmt.Sprintf("/users/%d", i))
+		s.Limit(lbls)
+	}
+	triggerDemandUpdate(s)
+
+	// Verify that overflow happens
+	lbls := labels.FromStrings("__name__", "http_requests", "url", "/users/999")
+	result := s.Limit(lbls)
+	require.Equal(t, overflowValue, result.Get("url"), "should overflow while cardinality is high")
+
+	// Phase 2: Simulate time passing - Advance the sketch enough times to fully rotate
+	// out all old high-cardinality data from the sketch ring and prune it.
+	// With staleDuration=15m and sketchDuration=5m, sketchesLength=4, so prune it 4 times
+	for i := 0; i < 4; i++ {
+		triggerPrune(s)
+	}
+
+	// Phase 3: Push only low-cardinality data (within limit)
+	for i := 0; i < 3; i++ {
+		lbls := labels.FromStrings("__name__", "http_requests", "url", fmt.Sprintf("/api/v1/resource_%d", i))
+		s.Limit(lbls)
+	}
+
+	// Trigger maintenance to re-evaluate overLimit from the now-low estimate
+	triggerDemandUpdate(s)
+
+	// Verify recovery - label values should pass through again
+	lbls = labels.FromStrings("__name__", "http_requests", "url", "/api/v1/healthy")
+	result = s.Limit(lbls)
+	require.Equal(t, "/api/v1/healthy", result.Get("url"), "should recover after cardinality drops below limit")
+
+	// Phase 4: Cardinality explodes again - verify overflow kicks back in
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "http_requests", "url", fmt.Sprintf("/new_endpoint/%d", i))
+		s.Limit(lbls)
+	}
+	triggerDemandUpdate(s)
+
+	lbls = labels.FromStrings("__name__", "http_requests", "url", "/new_endpoint/999")
+	result = s.Limit(lbls)
+	require.Equal(t, overflowValue, result.Get("url"), "should overflow again after cardinality increases")
+}
+
+func TestPerLabelLimiter_OverflowMetrics(t *testing.T) {
+	tenant := "test-overflow-metrics"
+	s := NewPerLabelLimiter(tenant, testMaxCardinality(5), 15*time.Minute)
+
+	// Push enough distinct values to exceed the limit
+	for i := 0; i < 10; i++ {
+		lbls := labels.FromStrings("__name__", "m", "url", fmt.Sprintf("/path/%d", i))
+		s.Limit(lbls)
+	}
+
+	triggerDemandUpdate(s)
+
+	// Now limit - should increment the counter
+	lbls := labels.FromStrings("__name__", "m", "url", "/path/new")
+	result := s.Limit(lbls)
+	require.Equal(t, overflowValue, result.Get("url"))
+
+	var m io_prometheus_client.Metric
+	require.NoError(t, metricLabelValuesLimited.WithLabelValues(tenant, "url").Write(&m))
+	require.Equal(t, float64(1), m.GetCounter().GetValue())
+
+	// Verify demand estimate gauge was updated during triggerDemandUpdate
+	var g io_prometheus_client.Metric
+	require.NoError(t, metricLabelCardinalityDemand.WithLabelValues(tenant, "url").Write(&g))
+	require.GreaterOrEqual(t, g.GetGauge().GetValue(), float64(10), "demand estimate should reflect the distinct values pushed")
+}
+
+// TestPerLabelLimiter_ConcurrentAccess verifies Limit() is safe to call
+// from multiple goroutines while doPeriodicMaintenance fires concurrently.
+// Run with -race to detect unsynchronized access.
+func TestPerLabelLimiter_ConcurrentAccess(t *testing.T) {
+	s := NewPerLabelLimiter("test", testMaxCardinality(10), 15*time.Minute)
+
+	// Replace tickers with channels we control, so doPeriodicMaintenance
+	// actually runs its demand-update and prune paths during the test.
+	demandCh := make(chan time.Time, 10)
+	pruneCh := make(chan time.Time, 10)
+	s.demandUpdateChan = demandCh
+	s.pruneChan = pruneCh
+
+	var wg sync.WaitGroup
+	wg.Add(12) // 10 Limit() goroutines + 1 demand update ticker + 1 prune ticker
+
+	for g := 0; g < 10; g++ {
+		go func(id int) {
+			defer wg.Done()
+			for i := 0; i < 100; i++ {
+				lbls := labels.FromStrings("__name__", "m", "label", fmt.Sprintf("g%d-v%d", id, i))
+				result := s.Limit(lbls)
+				require.Equal(t, "m", result.Get("__name__"), "metric name must be preserved")
+				val := result.Get("label")
+				require.True(t, val == fmt.Sprintf("g%d-v%d", id, i) || val == overflowValue,
+					"label value must be original or overflow, got: %s", val)
+			}
+		}(g)
+	}
+
+	// Feed demand update ticks concurrently - picked up by doPeriodicMaintenance
+	// inside Limit() calls, exercising the full code path.
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 10; i++ {
+			demandCh <- time.Now()
+			time.Sleep(time.Millisecond)
+		}
+	}()
+
+	// Feed prune ticks concurrently
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 5; i++ {
+			pruneCh <- time.Now()
+			time.Sleep(time.Millisecond)
+		}
+	}()
+
+	wg.Wait()
+
+	// After all goroutines finish, trigger maintenance and verify the state is consistent
+	triggerDemandUpdate(s)
+
+	s.mtx.Lock()
+	state, ok := s.labelsState["label"]
+	s.mtx.Unlock()
+	require.True(t, ok, "label state should exist after concurrent inserts")
+	// Estimate may be less than 1000 because prune ticks rotate out sketch data during the test
+	require.Greater(t, state.sketch.Estimate(), uint64(0), "sketch should have recorded values")
+}
+
+func BenchmarkPerLabelLimiter_Limit(b *testing.B) {
+	b.Run("disabled", func(b *testing.B) {
+		s := NewPerLabelLimiter("bench", testMaxCardinality(0), 15*time.Minute)
+		lbls := labels.FromStrings("__name__", "http_requests", "method", "GET", "url", "/api/v1/users")
+		b.ReportAllocs()
+		// Reset timer so setup (limiter creation, label generation, warmup) isn't measured
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			s.Limit(lbls)
+		}
+	})
+
+	b.Run("under_limit", func(b *testing.B) {
+		s := NewPerLabelLimiter("bench", testMaxCardinality(1000), 15*time.Minute)
+		// Pre-generate distinct labels to simulate real traffic with unique values
+		n := 500
+		allLbls := make([]labels.Labels, n)
+		for i := 0; i < n; i++ {
+			allLbls[i] = labels.FromStrings("__name__", "http_requests", "method", "GET", "url", fmt.Sprintf("/api/v1/users/%d", i))
+		}
+		s.Limit(allLbls[0])
+		triggerDemandUpdate(s)
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			s.Limit(allLbls[i%n])
+		}
+	})
+
+	b.Run("over_limit", func(b *testing.B) {
+		s := NewPerLabelLimiter("bench", testMaxCardinality(5), 15*time.Minute)
+		n := 500
+		allLbls := make([]labels.Labels, n)
+		for i := 0; i < n; i++ {
+			allLbls[i] = labels.FromStrings("__name__", "http_requests", "method", "GET", "url", fmt.Sprintf("/api/v1/users/%d", i))
+		}
+		for i := 0; i < 20; i++ {
+			s.Limit(allLbls[i])
+		}
+		triggerDemandUpdate(s)
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			s.Limit(allLbls[i%n])
+		}
+	})
+
+	b.Run("many_labels_under_limit", func(b *testing.B) {
+		s := NewPerLabelLimiter("bench", testMaxCardinality(1000), 15*time.Minute)
+		n := 500
+		allLbls := make([]labels.Labels, n)
+		for i := 0; i < n; i++ {
+			allLbls[i] = labels.FromStrings(
+				"__name__", "http_requests",
+				"method", "GET",
+				"url", fmt.Sprintf("/api/v1/users/%d", i),
+				"status_code", "200",
+				"service", "frontend",
+				"region", "us-east-1",
+				"instance", fmt.Sprintf("pod-%d", i),
+			)
+		}
+		s.Limit(allLbls[0])
+		triggerDemandUpdate(s)
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			s.Limit(allLbls[i%n])
+		}
+	})
+
+	b.Run("many_labels_over_limit", func(b *testing.B) {
+		s := NewPerLabelLimiter("bench", testMaxCardinality(5), 15*time.Minute)
+		n := 500
+		allLbls := make([]labels.Labels, n)
+		for i := 0; i < n; i++ {
+			allLbls[i] = labels.FromStrings(
+				"__name__", "http_requests",
+				"method", fmt.Sprintf("m%d", i),
+				"url", fmt.Sprintf("/path/%d", i),
+				"status_code", fmt.Sprintf("%d", i),
+				"service", fmt.Sprintf("svc%d", i),
+				"region", fmt.Sprintf("r%d", i),
+				"instance", fmt.Sprintf("pod%d", i),
+			)
+		}
+		// Warm up to trigger overflow on all labels
+		for i := 0; i < 20; i++ {
+			s.Limit(allLbls[i])
+		}
+		triggerDemandUpdate(s)
+		b.ReportAllocs()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			s.Limit(allLbls[i%n])
+		}
+	})
+
+	b.Run("parallel", func(b *testing.B) {
+		s := NewPerLabelLimiter("bench", testMaxCardinality(1000), 15*time.Minute)
+		n := 500
+		allLbls := make([]labels.Labels, n)
+		for i := 0; i < n; i++ {
+			allLbls[i] = labels.FromStrings("__name__", "http_requests", "method", "GET", "url", fmt.Sprintf("/api/v1/users/%d", i))
+		}
+		s.Limit(allLbls[0])
+		triggerDemandUpdate(s)
+		b.ReportAllocs()
+		b.ResetTimer()
+		var counter atomic.Int64
+		b.RunParallel(func(pb *testing.PB) {
+			for pb.Next() {
+				i := int(counter.Add(1))
+				s.Limit(allLbls[i%n])
+			}
+		})
+	})
+}
+
+// triggerDemandUpdate force runs the demand-update path of doPeriodicMaintenance,
+// re-evaluating overLimit from current sketch estimates.
+func triggerDemandUpdate(s *PerLabelLimiter) {
+	ch := make(chan time.Time, 1)
+	s.demandUpdateChan = ch
+	ch <- time.Now()
+	s.doPeriodicMaintenance()
+}
+
+// triggerPrune force runs the prune path of doPeriodicMaintenance, advancing the sketch ring by one step.
+func triggerPrune(s *PerLabelLimiter) {
+	ch := make(chan time.Time, 1)
+	s.pruneChan = ch
+	ch <- time.Now()
+	s.doPeriodicMaintenance()
+}

modules/generator/registry/registry.go

@@ -50,8 +50,10 @@ type ManagedRegistry struct {
 	metricsMtx   sync.RWMutex
 	metrics      map[string]metric
 	entityDemand *Cardinality
-	limiter      Limiter
-	sanitizer    Sanitizer
+
+	sanitizer       Sanitizer
+	perLabelLimiter LabelLimiter
+	limiter         Limiter
 
 	appendable storage.Appendable
 
@@ -98,6 +100,11 @@ type Limiter interface {
 	OnDelete(labelHash uint64, seriesCount uint32)
 }
 
+// LabelLimiter caps label cardinality by replacing high-cardinality values.
+type LabelLimiter interface {
+	Limit(lbls labels.Labels) labels.Labels
+}
+
 // Sanitizer applies a transformation to all non-constant labels.
 type Sanitizer interface {
 	Sanitize(lbls labels.Labels) labels.Labels
@@ -119,7 +126,8 @@ func New(cfg *Config, overrides Overrides, tenant string, appendable storage.App
 		externalLabels[cfg.InjectTenantIDAs] = tenant
 	}
 
-	sanitizer := NewDrainSanitizer(tenant, overrides.MetricsGeneratorSpanNameSanitization, cfg.StaleDuration)
+	drainSanitizer := NewDrainSanitizer(tenant, overrides.MetricsGeneratorSpanNameSanitization, cfg.StaleDuration)
+	perLabelLimiter := NewPerLabelLimiter(tenant, overrides.MetricsGeneratorMaxCardinalityPerLabel, cfg.StaleDuration)
 
 	r := &ManagedRegistry{
 		onShutdown: cancel,
@@ -131,10 +139,11 @@ func New(cfg *Config, overrides Overrides, tenant string, appendable storage.App
 
 		metrics: map[string]metric{},
 
-		appendable:   appendable,
-		limiter:      limiter,
-		sanitizer:    sanitizer,
-		entityDemand: NewCardinality(cfg.StaleDuration, removeStaleSeriesInterval),
+		appendable:      appendable,
+		sanitizer:       drainSanitizer,
+		perLabelLimiter: perLabelLimiter,
+		limiter:         limiter,
+		entityDemand:    NewCardinality(cfg.StaleDuration, removeStaleSeriesInterval),
 
 		logger:                 logger,
 		limitLogger:            tempo_log.NewRateLimitedLogger(1, level.Warn(logger)),
@@ -150,7 +159,7 @@ func New(cfg *Config, overrides Overrides, tenant string, appendable storage.App
 }
 
 func (r *ManagedRegistry) NewLabelBuilder() LabelBuilder {
-	return NewLabelBuilder(r.cfg.MaxLabelNameLength, r.cfg.MaxLabelValueLength, r.sanitizer)
+	return NewLabelBuilder(r.cfg.MaxLabelNameLength, r.cfg.MaxLabelValueLength, r.sanitizer, r.perLabelLimiter)
 }
 
 func (r *ManagedRegistry) OnAdd(labelHash uint64, seriesCount uint32, lbls labels.Labels) (labels.Labels, uint64) {

modules/generator/registry/registry_test.go

@@ -57,7 +57,7 @@ func (m *mockLimiter) OnPruneStaleSeries() {
 }
 
 func buildTestLabels(names []string, values []string) labels.Labels {
-	builder := NewLabelBuilder(0, 0, newTestDrainSanitizer(SpanNameSanitizationDisabled))
+	builder := NewLabelBuilder(0, 0, newTestDrainSanitizer(SpanNameSanitizationDisabled), newTestLabelLimiter())
 	for i := range names {
 		builder.Add(names[i], values[i])
 	}
@@ -484,6 +484,7 @@ type mockOverrides struct {
 	nativeHistogramMaxBucketNumber  uint32
 	nativeHistogramBucketFactor     float64
 	nativeHistogramMinResetDuration time.Duration
+	maxCardinalityPerLabel          uint64
 }
 
 var _ Overrides = (*mockOverrides)(nil)
@@ -528,6 +529,10 @@ func (m *mockOverrides) MetricsGeneratorSpanNameSanitization(string) string {
 	return SpanNameSanitizationDisabled
 }
 
+func (m *mockOverrides) MetricsGeneratorMaxCardinalityPerLabel(string) uint64 {
+	return m.maxCardinalityPerLabel
+}
+
 func mustGetHostname() string {
 	hostname, _ := os.Hostname()
 	return hostname
@@ -811,3 +816,60 @@ func TestManagedRegistry_entityDemandWithMultipleMetrics(t *testing.T) {
 	assert.Less(t, math.Abs(diff), 0.15, "entity demand should be ~50 since same entities used across metrics")
 	assert.Less(t, entityDemand, uint64(70), "entity demand should not triple-count entities across metrics")
 }
+
+func TestManagedRegistry_cardinalitySanitizer(t *testing.T) {
+	appender := &capturingAppender{}
+
+	cfg := &Config{
+		StaleDuration: 15 * time.Minute,
+	}
+	reg := New(cfg, &mockOverrides{maxCardinalityPerLabel: 5}, "test", appender, log.NewNopLogger(), noopLimiter)
+	defer reg.Close()
+
+	counter := reg.NewCounter("http_requests")
+
+	// Helper to build labels through the registry's label builder (which applies the sanitizer)
+	buildLabels := func(method, url string) labels.Labels {
+		b := reg.NewLabelBuilder()
+		b.Add("method", method)
+		b.Add("url", url)
+		lbls, _ := b.CloseAndBuildLabels()
+		return lbls
+	}
+
+	// Push 10 series with high-cardinality url but low-cardinality method
+	for i := 0; i < 10; i++ {
+		counter.Inc(buildLabels("GET", fmt.Sprintf("/users/%d", i)), 1.0)
+	}
+
+	// Force the per-label limiter to re-evaluate overLimit flags.
+	triggerDemandUpdate(reg.perLabelLimiter.(*PerLabelLimiter))
+
+	// Before the overflow kicks in, we should have exactly 10 series
+	// we got 10 active series while the `maxCardinalityPerLabel` is 5 because we added 10 active series
+	// before demand update was executed.
+	require.Equal(t, uint32(10), reg.activeSeries(), "10 pre-overflow series")
+
+	// Push 10 more series after maintenance has flagged url as over limit
+	// no new values of label `url` will be added.
+	for i := 0; i < 10; i++ {
+		counter.Inc(buildLabels("GET", fmt.Sprintf("/users/%d", i)), 1.0)
+	}
+
+	reg.CollectMetrics(context.Background())
+
+	// Verify: 'url' should have overflowed to __cardinality_overflow__ for post-maintenance series
+	// while the method remains "GET"
+	foundOverflow := false
+	for _, s := range appender.samples {
+		if s.l.Get("url") == "__cardinality_overflow__" {
+			foundOverflow = true
+			require.Equal(t, "GET", s.l.Get("method"), "method should be preserved when url overflows")
+		}
+	}
+	require.True(t, foundOverflow, "expected at least one series with url=__cardinality_overflow__")
+
+	// Verify active series: 10 pre-overflow series + 1 collapsed overflow series = 11
+	activeSeries := reg.activeSeries()
+	require.Equal(t, uint32(11), activeSeries, "10 pre-overflow + 1 overflow series")
+}

modules/generator/registry/test.go

@@ -10,6 +10,11 @@ import (
 	"github.com/prometheus/prometheus/storage"
 )
 
+// newTestLabelLimiter returns a PerLabelLimiter with limiting disabled (maxCardinality=0).
+func newTestLabelLimiter() *PerLabelLimiter {
+	return NewPerLabelLimiter("test", func(string) uint64 { return 0 }, 0)
+}
+
 // TestRegistry is a simple implementation of Registry intended for tests. It is not concurrent-safe.
 type TestRegistry struct {
 	// "metric{labels}" -> value
@@ -41,7 +46,7 @@ func (t *TestRegistry) NewGauge(name string) Gauge {
 func (t *TestRegistry) NewLabelBuilder() LabelBuilder {
 	nds := NewDrainSanitizer("test", func(string) string { return SpanNameSanitizationDisabled }, 0)
 
-	return NewLabelBuilder(0, 0, nds)
+	return NewLabelBuilder(0, 0, nds, newTestLabelLimiter())
 }
 
 func (t *TestRegistry) NewHistogram(name string, buckets []float64, histogramOverrides HistogramMode) Histogram {

modules/overrides/config.go

@@ -159,6 +159,7 @@ type MetricsGeneratorOverrides struct {
 	NativeHistogramMaxBucketNumber  uint32        `yaml:"native_histogram_max_bucket_number,omitempty" json:"native_histogram_max_bucket_number,omitempty"`
 	NativeHistogramMinResetDuration time.Duration `yaml:"native_histogram_min_reset_duration,omitempty" json:"native_histogram_min_reset_duration,omitempty"`
 	SpanNameSanitization            string        `yaml:"span_name_sanitization,omitempty" json:"span_name_sanitization,omitempty"`
+	MaxCardinalityPerLabel          uint64        `yaml:"max_cardinality_per_label,omitempty" json:"max_cardinality_per_label,omitempty"`
 }
 
 type ReadOverrides struct {

modules/overrides/config_legacy.go

@@ -67,6 +67,7 @@ func (c *Overrides) toLegacy() LegacyOverrides {
 		MetricsGeneratorNativeHistogramMaxBucketNumber:                              c.MetricsGenerator.NativeHistogramMaxBucketNumber,
 		MetricsGeneratorNativeHistogramMinResetDuration:                             c.MetricsGenerator.NativeHistogramMinResetDuration,
 		MetricsGeneratorSpanNameSanitization:                                        c.MetricsGenerator.SpanNameSanitization,
+		MetricsGeneratorMaxCardinalityPerLabel:                                      c.MetricsGenerator.MaxCardinalityPerLabel,
 
 		BlockRetention:     c.Compaction.BlockRetention,
 		CompactionWindow:   c.Compaction.CompactionWindow,
@@ -119,6 +120,7 @@ type LegacyOverrides struct {
 	MetricsGeneratorNativeHistogramMaxBucketNumber                              uint32                           `yaml:"metrics_generator_native_histogram_max_bucket_number,omitempty" json:"metrics_generator_native_histogram_max_bucket_number,omitempty"`
 	MetricsGeneratorNativeHistogramMinResetDuration                             time.Duration                    `yaml:"metrics_generator_native_histogram_min_reset_duration,omitempty" json:"native_histogram_min_reset_duration,omitempty"`
 	MetricsGeneratorSpanNameSanitization                                        string                           `yaml:"metrics_generator_span_name_sanitization" json:"metrics_generator_span_name_sanitization"`
+	MetricsGeneratorMaxCardinalityPerLabel                                      uint64                           `yaml:"metrics_generator_max_cardinality_per_label,omitempty" json:"metrics_generator_max_cardinality_per_label,omitempty"`
 	MetricsGeneratorTraceIDLabelName                                            string                           `yaml:"metrics_generator_trace_id_label_name" json:"metrics_generator_trace_id_label_name"`
 	MetricsGeneratorForwarderQueueSize                                          int                              `yaml:"metrics_generator_forwarder_queue_size" json:"metrics_generator_forwarder_queue_size"`
 	MetricsGeneratorForwarderWorkers                                            int                              `yaml:"metrics_generator_forwarder_workers" json:"metrics_generator_forwarder_workers"`
@@ -251,6 +253,7 @@ func (l *LegacyOverrides) toNewLimits() Overrides {
 			NativeHistogramMaxBucketNumber:  l.MetricsGeneratorNativeHistogramMaxBucketNumber,
 			NativeHistogramMinResetDuration: l.MetricsGeneratorNativeHistogramMinResetDuration,
 			SpanNameSanitization:            l.MetricsGeneratorSpanNameSanitization,
+			MaxCardinalityPerLabel:          l.MetricsGeneratorMaxCardinalityPerLabel,
 		},
 		Forwarders: l.Forwarders,
 		Global: GlobalOverrides{

modules/overrides/config_test.go

@@ -420,6 +420,7 @@ func generateTestLegacyOverrides() LegacyOverrides {
 		MetricsGeneratorProcessors:                                                  makeListToMap([]string{"processor-1", "processor-2"}),
 		MetricsGeneratorMaxActiveSeries:                                             1000,
 		MetricsGeneratorMaxActiveEntities:                                           100,
+		MetricsGeneratorMaxCardinalityPerLabel:                                      500,
 		MetricsGeneratorCollectionInterval:                                          10 * time.Second,
 		MetricsGeneratorDisableCollection:                                           false,
 		MetricsGeneratorGenerateNativeHistograms:                                    histograms.HistogramMethodNative,

modules/overrides/interface.go

@@ -84,6 +84,7 @@ type Interface interface {
 	MetricsGeneratorNativeHistogramMaxBucketNumber(userID string) uint32
 	MetricsGeneratorNativeHistogramMinResetDuration(userID string) time.Duration
 	MetricsGeneratorSpanNameSanitization(userID string) string
+	MetricsGeneratorMaxCardinalityPerLabel(userID string) uint64
 	BlockRetention(userID string) time.Duration
 	CompactionDisabled(userID string) bool
 	MaxSearchDuration(userID string) time.Duration

modules/overrides/runtime_config_overrides.go

@@ -420,7 +420,7 @@ func (o *runtimeConfigOverridesManager) MetricsGeneratorMaxActiveSeries(userID s
 	return o.getOverridesForUser(userID).MetricsGenerator.MaxActiveSeries
 }
 
-// MetricsGeneratorMaxEntities is the maximum amount of entities in the metrics-generator registry
+// MetricsGeneratorMaxActiveEntities is the maximum number of entities in the metrics-generator registry
 // for this tenant. Note this is a local limit enforced in every instance separately.
 // Requires the generator's limiter type to be set to "entity".
 func (o *runtimeConfigOverridesManager) MetricsGeneratorMaxActiveEntities(userID string) uint32 {
@@ -470,6 +470,13 @@ func (o *runtimeConfigOverridesManager) MetricsGeneratorSpanNameSanitization(use
 	return o.defaultLimits.MetricsGenerator.SpanNameSanitization
 }
 
+// MetricsGeneratorMaxCardinalityPerLabel is the maximum number of distinct values any single
+// label can have before values are replaced with __cardinality_overflow__.
+// 0 disables the limit.
+func (o *runtimeConfigOverridesManager) MetricsGeneratorMaxCardinalityPerLabel(userID string) uint64 {
+	return o.getOverridesForUser(userID).MetricsGenerator.MaxCardinalityPerLabel
+}
+
 // MetricsGeneratorTraceIDLabelName is the label name used for the trace ID in metrics.
 // "TraceID" is used if no value is provided.
 func (o *runtimeConfigOverridesManager) MetricsGeneratorTraceIDLabelName(userID string) string {

modules/overrides/runtime_config_overrides_test.go

@@ -683,6 +683,86 @@ func TestNativeHistogramOverrides(t *testing.T) {
 	}
 }
 
+func TestMetricsGeneratorMaxCardinalityPerLabel(t *testing.T) {
+	tests := []struct {
+		name               string
+		defaultLimits      Overrides
+		perTenantOverrides *perTenantOverrides
+		expected           map[string]uint64
+	}{
+		{
+			name: "default enabled, no tenant override",
+			defaultLimits: Overrides{
+				MetricsGenerator: MetricsGeneratorOverrides{
+					MaxCardinalityPerLabel: 100,
+				},
+			},
+			expected: map[string]uint64{"user1": 100, "user2": 100},
+		},
+		{
+			name:          "default disabled, tenant enables",
+			defaultLimits: Overrides{},
+			perTenantOverrides: &perTenantOverrides{
+				TenantLimits: map[string]*Overrides{
+					"user1": {
+						MetricsGenerator: MetricsGeneratorOverrides{
+							MaxCardinalityPerLabel: 50,
+						},
+					},
+				},
+			},
+			expected: map[string]uint64{"user1": 50, "user2": 0},
+		},
+		{
+			name: "default enabled, tenant disables with 0",
+			defaultLimits: Overrides{
+				MetricsGenerator: MetricsGeneratorOverrides{
+					MaxCardinalityPerLabel: 100,
+				},
+			},
+			perTenantOverrides: &perTenantOverrides{
+				TenantLimits: map[string]*Overrides{
+					"user1": {
+						MetricsGenerator: MetricsGeneratorOverrides{
+							MaxCardinalityPerLabel: 0,
+						},
+					},
+				},
+			},
+			expected: map[string]uint64{"user1": 0, "user2": 100},
+		},
+		{
+			name: "default enabled, tenant overrides with higher value",
+			defaultLimits: Overrides{
+				MetricsGenerator: MetricsGeneratorOverrides{
+					MaxCardinalityPerLabel: 100,
+				},
+			},
+			perTenantOverrides: &perTenantOverrides{
+				TenantLimits: map[string]*Overrides{
+					"user1": {
+						MetricsGenerator: MetricsGeneratorOverrides{
+							MaxCardinalityPerLabel: 500,
+						},
+					},
+				},
+			},
+			expected: map[string]uint64{"user1": 500, "user2": 100},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			overrides, cleanup := createAndInitializeRuntimeOverridesManager(t, tt.defaultLimits, toYamlBytes(t, tt.perTenantOverrides))
+			defer cleanup()
+
+			for user, expected := range tt.expected {
+				require.Equal(t, expected, overrides.MetricsGeneratorMaxCardinalityPerLabel(user), "user: %s", user)
+			}
+		})
+	}
+}
+
 func createAndInitializeRuntimeOverridesManager(t *testing.T, defaultLimits Overrides, perTenantOverrides []byte) (Service, func()) {
 	cfg := Config{
 		Defaults: defaultLimits,