feat(ingester): Add owned series tracking to prevent false throttling…

CHANGELOG.md

@@ -22,6 +22,7 @@
 * [CHANGE] Alertmanager: Remove the obsolete startup migration of local state files into per-tenant directories (scheduled for removal in 1.11.0). Upgrading from a release older than 1.9.0 with a persisted local state directory now requires upgrading to an intermediate release first, so the migration can run. #7513
 * [CHANGE] Cache: Setting `-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl` to 0 will disable the bucket-index cache. #7446
 * [CHANGE] HA Tracker: Move `-distributor.ha-tracker.failover-timeout` from a global config to a per-tenant runtime config. The flag name and default value (30s) remain the same. #7481
+* [FEATURE] Ingester: Add experimental owned series tracking to prevent false throttling during ingester scale-up and ring resharding. The ingester counts only the series the ring assigns to it, rather than everything left in the TSDB head, closing a window of up to one head compaction cycle in which a tenant could be throttled against a limit that had already shrunk. Enabled via `-ingester.owned-series-metrics-enabled` and `-ingester.owned-series-limit-enforcement-enabled`. #7509
 * [FEATURE] Parquet: Support sharded parquet file conversion and querying. #7610
 * [FEATURE] Parquet Converter: Add experimental `-parquet-converter.max-num-columns` flag to automatically shard parquet files when the number of columns exceeds the configured limit. This prevents failures when a TSDB block has more unique label names than the parquet library's column limit (32767). #7624
 * [FEATURE] Distributor: Add experimental `-distributor.num-query-workers` flag to use a goroutine worker pool for query fan-out calls to ingesters. Reuses pre-grown goroutine stacks to eliminate the `runtime.copystack` overhead (~8% CPU) observed on rulers with wide ingester fan-out. Falls back to spawning a new goroutine when no worker is available. #7623

docs/configuration/config-file-reference.md

@@ -4193,6 +4193,30 @@ lifecycler:
 # CLI flag: -ingester.head-queried-series-metrics-windows
 [head_queried_series_metrics_windows: <list of duration> | default = 2h0m0s]
 
+# Enable tracking of owned series per user. When enabled, the ingester computes
+# which series the ring assigns to it and emits the cortex_ingester_owned_series
+# metric. This also changes how long active series entries are retained: they
+# are released at head compaction rather than after the idle timeout, so that
+# the owned count reflects series still held in memory rather than only recently
+# active ones. Ingester memory therefore increases roughly in proportion to the
+# ratio between series in the TSDB head and recently active series. The value of
+# cortex_ingester_active_series is unchanged, but cortex_ingester_owned_series
+# may exceed it for a tenant with high churn, because it counts idle series
+# which are still held.
+# CLI flag: -ingester.owned-series-metrics-enabled
+[owned_series_metrics_enabled: <boolean> | default = false]
+
+# Use the owned series count for limit enforcement instead of the total number
+# of series in the TSDB head. Requires owned-series-metrics-enabled. This
+# prevents tenants being throttled against a limit that has already shrunk after
+# a ring change, while the head still holds series that have moved elsewhere.
+# Note this applies to the instance-wide max_series limit as well as to
+# per-tenant limits, so max_series stops counting series which have been
+# reassigned to another ingester but are still resident until the next head
+# compaction.
+# CLI flag: -ingester.owned-series-limit-enforcement-enabled
+[owned_series_limit_enforcement_enabled: <boolean> | default = false]
+
 # Enable uploading compacted blocks.
 # CLI flag: -ingester.upload-compacted-blocks-enabled
 [upload_compacted_blocks_enabled: <boolean> | default = true]

docs/configuration/v1-guarantees.md

@@ -174,3 +174,9 @@ Currently experimental features are:
   - `-querier.optimizers` / `-ruler.optimizers` (string) CLI flags
   - `-querier.decoding-concurrency` / `-ruler.decoding-concurrency` (int) CLI flags
   - `-querier.selector-batch-size` / `-ruler.selector-batch-size` (int) CLI flags
+- Ingester: Owned Series Tracking
+  - Enable on Ingester via `-ingester.owned-series-metrics-enabled=true`
+  - Counts only the series the ring assigns to this ingester and exposes the `cortex_ingester_owned_series` metric
+  - Active series entries are then retained until head compaction instead of being released after `-ingester.active-series-metrics-idle-timeout`, so that the owned count reflects what is held in memory. This increases ingester memory roughly in proportion to the ratio between series in the TSDB head and recently active series
+  - The value of `cortex_ingester_active_series` is unchanged, but `cortex_ingester_owned_series` may exceed it for a tenant with high churn, since it counts idle series which are still held
+  - `-ingester.owned-series-limit-enforcement-enabled` additionally uses that count for per-tenant and instance series limits, instead of the total number of series in the TSDB head. Note the instance-wide `max_series` limit then stops counting series which have been reassigned to another ingester but are still resident until the next head compaction

pkg/distributor/distributor.go

@@ -36,7 +36,6 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	ring_client "github.com/cortexproject/cortex/pkg/ring/client"
 	"github.com/cortexproject/cortex/pkg/util"
-	"github.com/cortexproject/cortex/pkg/util/extract"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/labelset"
 	"github.com/cortexproject/cortex/pkg/util/limiter"
@@ -587,49 +586,11 @@ func (d *Distributor) stopping(_ error) error {
 }
 
 func (d *Distributor) tokenForLabels(userID string, labels []cortexpb.LabelAdapter) (uint32, error) {
-	if d.cfg.ShardByAllLabels {
-		return shardByAllLabels(userID, labels), nil
-	}
-
-	unsafeMetricName, err := extract.UnsafeMetricNameFromLabelAdapters(labels)
-	if err != nil {
-		return 0, err
-	}
-	return shardByMetricName(userID, unsafeMetricName), nil
+	return ring.TokenForLabels(userID, labels, d.cfg.ShardByAllLabels)
 }
 
 func (d *Distributor) tokenForMetadata(userID string, metricName string) uint32 {
-	if d.cfg.ShardByAllLabels {
-		return shardByMetricName(userID, metricName)
-	}
-
-	return shardByUser(userID)
-}
-
-// shardByMetricName returns the token for the given metric. The provided metricName
-// is guaranteed to not be retained.
-func shardByMetricName(userID string, metricName string) uint32 {
-	h := shardByUser(userID)
-	h = ingester_client.HashAdd32(h, metricName)
-	return h
-}
-
-func shardByUser(userID string) uint32 {
-	h := ingester_client.HashNew32()
-	h = ingester_client.HashAdd32(h, userID)
-	return h
-}
-
-// This function generates different values for different order of same labels.
-func shardByAllLabels(userID string, labels []cortexpb.LabelAdapter) uint32 {
-	h := shardByUser(userID)
-	for _, label := range labels {
-		if len(label.Value) > 0 {
-			h = ingester_client.HashAdd32(h, label.Name)
-			h = ingester_client.HashAdd32(h, label.Value)
-		}
-	}
-	return h
+	return ring.TokenForMetadata(userID, metricName, d.cfg.ShardByAllLabels)
 }
 
 // Remove the label labelname from a slice of LabelPairs if it exists.

pkg/distributor/distributor_test.go

@@ -3812,7 +3812,7 @@ func (i *mockIngester) Push(ctx context.Context, req *cortexpb.WriteRequest, opt
 
 	for j := range req.Timeseries {
 		series := req.Timeseries[j]
-		hash := shardByAllLabels(orgid, series.Labels)
+		hash := ring.ShardByAllLabels(orgid, series.Labels)
 		existing, ok := i.timeseries[hash]
 		if !ok {
 			// Make a copy because the request Timeseries are reused
@@ -3831,7 +3831,7 @@ func (i *mockIngester) Push(ctx context.Context, req *cortexpb.WriteRequest, opt
 	}
 
 	for _, m := range req.Metadata {
-		hash := shardByMetricName(orgid, m.MetricFamilyName)
+		hash := ring.ShardByMetricName(orgid, m.MetricFamilyName)
 		set, ok := i.metadata[hash]
 		if !ok {
 			set = map[cortexpb.MetricMetadata]struct{}{}
@@ -4312,13 +4312,13 @@ func TestRemoveReplicaLabel(t *testing.T) {
 // This is not great, but we deal with unsorted labels when validating labels.
 func TestShardByAllLabelsReturnsWrongResultsForUnsortedLabels(t *testing.T) {
 	t.Parallel()
-	val1 := shardByAllLabels("test", []cortexpb.LabelAdapter{
+	val1 := ring.ShardByAllLabels("test", []cortexpb.LabelAdapter{
 		{Name: "__name__", Value: "foo"},
 		{Name: "bar", Value: "baz"},
 		{Name: "sample", Value: "1"},
 	})
 
-	val2 := shardByAllLabels("test", []cortexpb.LabelAdapter{
+	val2 := ring.ShardByAllLabels("test", []cortexpb.LabelAdapter{
 		{Name: "__name__", Value: "foo"},
 		{Name: "sample", Value: "1"},
 		{Name: "bar", Value: "baz"},

pkg/distributor/query.go

@@ -104,7 +104,7 @@ func (d *Distributor) GetIngestersForQuery(ctx context.Context, matchers ...*lab
 		metricNameMatcher, _, ok := extract.MetricNameMatcherFromMatchers(matchers)
 
 		if ok && metricNameMatcher.Type == labels.MatchEqual {
-			return d.ingestersRing.Get(shardByMetricName(userID, metricNameMatcher.Value), ring.Read, nil, nil, nil)
+			return d.ingestersRing.Get(ring.ShardByMetricName(userID, metricNameMatcher.Value), ring.Read, nil, nil, nil)
 		}
 	}