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

CHANGELOG.md

@@ -4,6 +4,9 @@
 * [CHANGE] Querier: Make query time range configurations per-tenant: `query_ingesters_within`, `query_store_after`, and `shuffle_sharding_ingesters_lookback_period`. Uses `model.Duration` instead of `time.Duration` to support serialization but has minimum unit of 1ms (nanoseconds/microseconds not supported). #7160
 * [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 owned series tracking to prevent false customer throttling during ingester scale-up and ring resharding. When enabled, the ingester tracks which series it currently owns according to the ring and uses that count (instead of total in-memory series) for limit enforcement. Eliminates a up-to-2-hour window of incorrect throttling after any ring change. Controlled by `-ingester.owned-series-metrics-enabled` (metric emission) and `-ingester.owned-series-limit-enforcement-enabled` (limit enforcement). #7509
+* [ENHANCEMENT] Ring: Consolidate sharding functions (`TokenForLabels`, `ShardByMetricName`, etc.) into `pkg/ring/token.go` for reuse by both distributor and ingester. Export `SearchToken`. #7509
+* [ENHANCEMENT] Util: Consolidate FNV hash functions into `pkg/util/fnv.go`, removing duplicate from `pkg/ingester/client/fnv.go`. #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

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

@@ -3799,7 +3799,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
@@ -3818,7 +3818,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{}{}
@@ -4299,13 +4299,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)
 		}
 	}