Replaced Subring() with ShuffleShard()

Signed-off-by: Marco Pracucci <marco@pracucci.com>

Author: pracucci

docs/configuration/config-file-reference.md

@@ -393,6 +393,10 @@ ha_tracker:
 # CLI flag: -distributor.extra-query-delay
 [extra_queue_delay: <duration> | default = 0s]
 
+# The sharding strategy to use. Supported values are: default, shuffle-sharding.
+# CLI flag: -distributor.sharding-strategy
+[sharding_strategy: <string> | default = "default"]
+
 # Distribute samples based on all labels, as opposed to solely by user and
 # metric name.
 # CLI flag: -distributor.shard-by-all-labels
@@ -2714,9 +2718,12 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s
 # CLI flag: -validation.enforce-metric-name
 [enforce_metric_name: <boolean> | default = true]
 
-# Per-user subring to shard metrics to ingesters. 0 is disabled.
-# CLI flag: -experimental.distributor.user-subring-size
-[user_subring_size: <int> | default = 0]
+# The default tenant's shard size when the shuffle-sharding strategy is used.
+# Must be set both on ingesters and distributors. When this setting is specified
+# in the per-tenant overrides, a value of 0 disables shuffle sharding for the
+# tenant.
+# CLI flag: -distribution.ingestion-tenant-shard-size
+[ingestion_tenant_shard_size: <int> | default = 0]
 
 # The maximum number of series for which a query can fetch samples from each
 # ingester. This limit is enforced only in the ingesters (when querying samples

pkg/cortex/cortex.go

@@ -172,7 +172,7 @@ func (c *Config) Validate(log log.Logger) error {
 	if err := c.LimitsConfig.Validate(c.Distributor.ShardByAllLabels); err != nil {
 		return errors.Wrap(err, "invalid limits config")
 	}
-	if err := c.Distributor.Validate(); err != nil {
+	if err := c.Distributor.Validate(c.LimitsConfig); err != nil {
 		return errors.Wrap(err, "invalid distributor config")
 	}
 	if err := c.Querier.Validate(); err != nil {

pkg/distributor/distributor.go

@@ -3,6 +3,7 @@ package distributor
 import (
 	"context"
 	"flag"
+	"fmt"
 	"net/http"
 	"sort"
 	"strings"
@@ -104,11 +105,21 @@ var (
 		Help: "Unix timestamp of latest received sample per user.",
 	}, []string{"user"})
 	emptyPreallocSeries = ingester_client.PreallocTimeseries{}
+
+	supportedShardingStrategies = []string{ShardingStrategyDefault, ShardingStrategyShuffle}
+
+	// Validation errors.
+	errInvalidShardingStrategy = errors.New("invalid sharding strategy")
+	errInvalidTenantShardSize  = errors.New("invalid tenant shard size, the value must be greater than 0")
 )
 
 const (
 	typeSamples  = "samples"
 	typeMetadata = "metadata"
+
+	// Supported sharding strategies.
+	ShardingStrategyDefault = "default"
+	ShardingStrategyShuffle = "shuffle-sharding"
 )
 
 // Distributor is a storage.SampleAppender and a client.Querier which
@@ -147,7 +158,8 @@ type Config struct {
 	RemoteTimeout   time.Duration `yaml:"remote_timeout"`
 	ExtraQueryDelay time.Duration `yaml:"extra_queue_delay"`
 
-	ShardByAllLabels bool `yaml:"shard_by_all_labels"`
+	ShardingStrategy string `yaml:"sharding_strategy"`
+	ShardByAllLabels bool   `yaml:"shard_by_all_labels"`
 
 	// Distributors ring
 	DistributorRing RingConfig `yaml:"ring"`
@@ -170,10 +182,19 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.RemoteTimeout, "distributor.remote-timeout", 2*time.Second, "Timeout for downstream ingesters.")
 	f.DurationVar(&cfg.ExtraQueryDelay, "distributor.extra-query-delay", 0, "Time to wait before sending more than the minimum successful query requests.")
 	f.BoolVar(&cfg.ShardByAllLabels, "distributor.shard-by-all-labels", false, "Distribute samples based on all labels, as opposed to solely by user and metric name.")
+	f.StringVar(&cfg.ShardingStrategy, "distributor.sharding-strategy", ShardingStrategyDefault, fmt.Sprintf("The sharding strategy to use. Supported values are: %s.", strings.Join(supportedShardingStrategies, ", ")))
 }
 
 // Validate config and returns error on failure
-func (cfg *Config) Validate() error {
+func (cfg *Config) Validate(limits validation.Limits) error {
+	if !util.StringsContain(supportedShardingStrategies, cfg.ShardingStrategy) {
+		return errInvalidShardingStrategy
+	}
+
+	if cfg.ShardingStrategy == ShardingStrategyShuffle && limits.IngestionTenantShardSize <= 0 {
+		return errInvalidTenantShardSize
+	}
+
 	return cfg.HATrackerConfig.Validate()
 }
 
@@ -508,13 +529,11 @@ func (d *Distributor) Push(ctx context.Context, req *client.WriteRequest) (*clie
 		return nil, httpgrpc.Errorf(http.StatusTooManyRequests, "ingestion rate limit (%v) exceeded while adding %d samples and %d metadata", d.ingestionRateLimiter.Limit(now, userID), validatedSamples, len(validatedMetadata))
 	}
 
-	var subRing ring.ReadRing
-	subRing = d.ingestersRing
+	subRing := d.ingestersRing.(ring.ReadRing)
 
-	// Obtain a subring if required
-	if size := d.limits.SubringSize(userID); size > 0 {
-		h := client.HashAdd32a(client.HashNew32a(), userID)
-		subRing = d.ingestersRing.Subring(h, size)
+	// Obtain a subring if required.
+	if d.cfg.ShardingStrategy == ShardingStrategyShuffle {
+		subRing = d.ingestersRing.ShuffleShard(userID, d.limits.IngestionTenantShardSize(userID))
 	}
 
 	keys := append(seriesKeys, metadataKeys...)

pkg/querier/blocks_store_replicated_set_test.go

@@ -37,10 +37,7 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) {
 	block3Hash := cortex_tsdb.HashBlockID(block3)
 	block4Hash := cortex_tsdb.HashBlockID(block4)
 
-	// Ensure the user ID we use belongs to the instances holding the token for the block 1
-	// (it's expected by the assertions below).
 	userID := "user-A"
-	require.LessOrEqual(t, cortex_tsdb.HashTenantID(userID), block1Hash)
 
 	tests := map[string]struct {
 		shardingStrategy  string
@@ -250,7 +247,7 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) {
 			queryBlocks: []ulid.ULID{block1, block2, block4},
 			expectedClients: map[string][]ulid.ULID{
 				"127.0.0.1": {block1, block4},
-				"127.0.0.2": {block2},
+				"127.0.0.3": {block2},
 			},
 		},
 		"shuffle sharding, multiple instances in the ring with RF = 1, SS = 4": {
@@ -286,7 +283,7 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) {
 				block2: {"127.0.0.1"},
 			},
 			expectedClients: map[string][]ulid.ULID{
-				"127.0.0.2": {block1, block2},
+				"127.0.0.3": {block1, block2},
 			},
 		},
 		"shuffle sharding, multiple instances in the ring with RF = 2, SS = 2 with excluded blocks and no replacement available": {
@@ -301,7 +298,7 @@ func TestBlocksStoreReplicationSet_GetClientsFor(t *testing.T) {
 			},
 			queryBlocks: []ulid.ULID{block1, block2},
 			exclude: map[ulid.ULID][]string{
-				block1: {"127.0.0.1", "127.0.0.2"},
+				block1: {"127.0.0.1", "127.0.0.3"},
 				block2: {"127.0.0.1"},
 			},
 			expectedErr: fmt.Errorf("no store-gateway instance left after checking exclude for block %s", block1.String()),

pkg/ring/ring.go

@@ -49,7 +49,10 @@ type ReadRing interface {
 	GetAll(op Operation) (ReplicationSet, error)
 	ReplicationFactor() int
 	IngesterCount() int
-	Subring(key uint32, n int) ReadRing
+
+	// ShuffleShard returns a subring for the provided identifier (eg. a tenant ID)
+	// and size (number of instances).
+	ShuffleShard(identifier string, size int) ReadRing
 
 	// HasInstance returns whether the ring contains an instance matching the provided instanceID.
 	HasInstance(instanceID string) bool

pkg/storage/tsdb/util.go

@@ -15,9 +15,3 @@ func HashBlockID(id ulid.ULID) uint32 {
 	}
 	return h
 }
-
-// HashTenantID returns a 32-bit hash of the tenant ID useful for
-// ring-based sharding.
-func HashTenantID(id string) uint32 {
-	return client.HashAdd32a(client.HashNew32a(), id)
-}

pkg/storegateway/sharding_strategy.go

@@ -154,7 +154,7 @@ func GetShuffleShardingSubring(ring *ring.Ring, userID string, limits ShardingLi
 		return ring
 	}
 
-	return ring.Subring(cortex_tsdb.HashTenantID(userID), shardSize)
+	return ring.ShuffleShard(userID, shardSize)
 }
 
 type shardingMetadataFilterAdapter struct {

pkg/storegateway/sharding_strategy_test.go

@@ -291,10 +291,7 @@ func TestShuffleShardingStrategy(t *testing.T) {
 	block3Hash := cortex_tsdb.HashBlockID(block3)
 	block4Hash := cortex_tsdb.HashBlockID(block4)
 
-	// Ensure the user ID we use belongs to the instances holding the token for the block 1
-	// (it's expected by the assertions below).
 	userID := "user-A"
-	require.LessOrEqual(t, cortex_tsdb.HashTenantID(userID), block1Hash)
 
 	type usersExpectation struct {
 		instanceID   string
@@ -499,62 +496,62 @@ func TestShuffleShardingStrategy(t *testing.T) {
 			limits:            &shardingLimitsMock{storeGatewayTenantShardSize: 2},
 			setupRing: func(r *ring.Desc) {
 				r.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1, block4Hash + 1}, ring.ACTIVE)
-				r.AddIngester("instance-3", "127.0.0.3", "", []uint32{block3Hash + 1}, ring.ACTIVE)
+				r.AddIngester("instance-2", "127.0.0.2", "", []uint32{block2Hash + 1}, ring.ACTIVE)
 
-				r.Ingesters["instance-2"] = ring.IngesterDesc{
-					Addr:      "127.0.0.2",
+				r.Ingesters["instance-3"] = ring.IngesterDesc{
+					Addr:      "127.0.0.3",
 					Timestamp: time.Now().Add(-time.Hour).Unix(),
 					State:     ring.ACTIVE,
-					Tokens:    []uint32{block2Hash + 1},
+					Tokens:    []uint32{block3Hash + 1},
 				}
 			},
 			expectedUsers: []usersExpectation{
 				{instanceID: "instance-1", instanceAddr: "127.0.0.1", users: []string{userID}},
-				{instanceID: "instance-2", instanceAddr: "127.0.0.2", users: []string{userID}},
-				{instanceID: "instance-3", instanceAddr: "127.0.0.3", users: nil},
+				{instanceID: "instance-2", instanceAddr: "127.0.0.2", users: nil},
+				{instanceID: "instance-3", instanceAddr: "127.0.0.3", users: []string{userID}},
 			},
 			expectedBlocks: []blocksExpectation{
 				{instanceID: "instance-1", instanceAddr: "127.0.0.1", blocks: []ulid.ULID{block1, block2, block3, block4}},
-				{instanceID: "instance-2", instanceAddr: "127.0.0.2", blocks: []ulid.ULID{ /* no blocks because unhealthy */ }},
-				{instanceID: "instance-3", instanceAddr: "127.0.0.3", blocks: []ulid.ULID{ /* no blocks because not belonging to the shard */ }},
+				{instanceID: "instance-2", instanceAddr: "127.0.0.2", blocks: []ulid.ULID{ /* no blocks because not belonging to the shard */ }},
+				{instanceID: "instance-3", instanceAddr: "127.0.0.3", blocks: []ulid.ULID{ /* no blocks because unhealthy */ }},
 			},
 		},
 		"LEAVING instance in the ring should continue to keep its shard blocks but they should also be replicated to another instance": {
 			replicationFactor: 1,
 			limits:            &shardingLimitsMock{storeGatewayTenantShardSize: 2},
 			setupRing: func(r *ring.Desc) {
 				r.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1, block3Hash + 1}, ring.ACTIVE)
-				r.AddIngester("instance-2", "127.0.0.2", "", []uint32{block2Hash + 1}, ring.LEAVING)
-				r.AddIngester("instance-3", "127.0.0.3", "", []uint32{block4Hash + 1}, ring.ACTIVE)
+				r.AddIngester("instance-2", "127.0.0.2", "", []uint32{block2Hash + 1}, ring.ACTIVE)
+				r.AddIngester("instance-3", "127.0.0.3", "", []uint32{block4Hash + 1}, ring.LEAVING)
 			},
 			expectedUsers: []usersExpectation{
 				{instanceID: "instance-1", instanceAddr: "127.0.0.1", users: []string{userID}},
-				{instanceID: "instance-2", instanceAddr: "127.0.0.2", users: []string{userID}},
-				{instanceID: "instance-3", instanceAddr: "127.0.0.3", users: nil},
+				{instanceID: "instance-2", instanceAddr: "127.0.0.2", users: nil},
+				{instanceID: "instance-3", instanceAddr: "127.0.0.3", users: []string{userID}},
 			},
 			expectedBlocks: []blocksExpectation{
-				{instanceID: "instance-1", instanceAddr: "127.0.0.1", blocks: []ulid.ULID{block1, block3, block4 /* replicated: */, block2}},
-				{instanceID: "instance-2", instanceAddr: "127.0.0.2", blocks: []ulid.ULID{block2}},
-				{instanceID: "instance-3", instanceAddr: "127.0.0.3", blocks: []ulid.ULID{ /* no blocks because not belonging to the shard */ }},
+				{instanceID: "instance-1", instanceAddr: "127.0.0.1", blocks: []ulid.ULID{block1, block2, block3 /* replicated: */, block4}},
+				{instanceID: "instance-2", instanceAddr: "127.0.0.2", blocks: []ulid.ULID{ /* no blocks because not belonging to the shard */ }},
+				{instanceID: "instance-3", instanceAddr: "127.0.0.3", blocks: []ulid.ULID{block4}},
 			},
 		},
 		"JOINING instance in the ring should get its shard blocks but they should also be replicated to another instance": {
 			replicationFactor: 1,
 			limits:            &shardingLimitsMock{storeGatewayTenantShardSize: 2},
 			setupRing: func(r *ring.Desc) {
 				r.AddIngester("instance-1", "127.0.0.1", "", []uint32{block1Hash + 1, block3Hash + 1}, ring.ACTIVE)
-				r.AddIngester("instance-2", "127.0.0.2", "", []uint32{block2Hash + 1}, ring.JOINING)
-				r.AddIngester("instance-3", "127.0.0.3", "", []uint32{block4Hash + 1}, ring.ACTIVE)
+				r.AddIngester("instance-2", "127.0.0.2", "", []uint32{block2Hash + 1}, ring.ACTIVE)
+				r.AddIngester("instance-3", "127.0.0.3", "", []uint32{block4Hash + 1}, ring.JOINING)
 			},
 			expectedUsers: []usersExpectation{
 				{instanceID: "instance-1", instanceAddr: "127.0.0.1", users: []string{userID}},
-				{instanceID: "instance-2", instanceAddr: "127.0.0.2", users: []string{userID}},
-				{instanceID: "instance-3", instanceAddr: "127.0.0.3", users: nil},
+				{instanceID: "instance-2", instanceAddr: "127.0.0.2", users: nil},
+				{instanceID: "instance-3", instanceAddr: "127.0.0.3", users: []string{userID}},
 			},
 			expectedBlocks: []blocksExpectation{
-				{instanceID: "instance-1", instanceAddr: "127.0.0.1", blocks: []ulid.ULID{block1, block3, block4 /* replicated: */, block2}},
-				{instanceID: "instance-2", instanceAddr: "127.0.0.2", blocks: []ulid.ULID{block2}},
-				{instanceID: "instance-3", instanceAddr: "127.0.0.3", blocks: []ulid.ULID{ /* no blocks because not belonging to the shard */ }},
+				{instanceID: "instance-1", instanceAddr: "127.0.0.1", blocks: []ulid.ULID{block1, block2, block3 /* replicated: */, block4}},
+				{instanceID: "instance-2", instanceAddr: "127.0.0.2", blocks: []ulid.ULID{ /* no blocks because not belonging to the shard */ }},
+				{instanceID: "instance-3", instanceAddr: "127.0.0.3", blocks: []ulid.ULID{block4}},
 			},
 		},
 		"SS = 0 disables shuffle sharding": {

pkg/util/validation/limits.go

@@ -38,7 +38,7 @@ type Limits struct {
 	CreationGracePeriod       time.Duration       `yaml:"creation_grace_period"`
 	EnforceMetadataMetricName bool                `yaml:"enforce_metadata_metric_name"`
 	EnforceMetricName         bool                `yaml:"enforce_metric_name"`
-	SubringSize               int                 `yaml:"user_subring_size"`
+	IngestionTenantShardSize  int                 `yaml:"ingestion_tenant_shard_size"`
 
 	// Ingester enforced limits.
 	// Series
@@ -72,7 +72,7 @@ type Limits struct {
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (l *Limits) RegisterFlags(f *flag.FlagSet) {
-	f.IntVar(&l.SubringSize, "experimental.distributor.user-subring-size", 0, "Per-user subring to shard metrics to ingesters. 0 is disabled.")
+	f.IntVar(&l.IngestionTenantShardSize, "distribution.ingestion-tenant-shard-size", 0, "The default tenant's shard size when the shuffle-sharding strategy is used. Must be set both on ingesters and distributors. When this setting is specified in the per-tenant overrides, a value of 0 disables shuffle sharding for the tenant.")
 	f.Float64Var(&l.IngestionRate, "distributor.ingestion-rate-limit", 25000, "Per-user ingestion rate limit in samples per second.")
 	f.StringVar(&l.IngestionRateStrategy, "distributor.ingestion-rate-limit-strategy", "local", "Whether the ingestion rate limit should be applied individually to each distributor instance (local), or evenly shared across the cluster (global).")
 	f.IntVar(&l.IngestionBurstSize, "distributor.ingestion-burst-size", 50000, "Per-user allowed ingestion burst size (in number of samples).")
@@ -341,12 +341,12 @@ func (o *Overrides) MaxGlobalMetadataPerMetric(userID string) int {
 	return o.getOverridesForUser(userID).MaxGlobalMetadataPerMetric
 }
 
-// SubringSize returns the size of the subring for a given user.
-func (o *Overrides) SubringSize(userID string) int {
-	return o.getOverridesForUser(userID).SubringSize
+// IngestionTenantShardSize returns the ingesters shard size for a given user.
+func (o *Overrides) IngestionTenantShardSize(userID string) int {
+	return o.getOverridesForUser(userID).IngestionTenantShardSize
 }
 
-// StoreGatewayTenantShardSize returns the size of the store-gateway shard size for a given user.
+// StoreGatewayTenantShardSize returns the store-gateway shard size for a given user.
 func (o *Overrides) StoreGatewayTenantShardSize(userID string) int {
 	return o.getOverridesForUser(userID).StoreGatewayTenantShardSize
 }