Shuffle-sharding support in Ruler.

CHANGELOG.md

@@ -63,12 +63,14 @@
 * [ENHANCEMENT] Reduce tail latency by smoothing out spikes in rate of chunk flush operations. #3191
 * [ENHANCEMENT] Experimental Ruler API: Fetch rule groups from object storage in parallel. #3218
 * [ENHANCEMENT] Chunks GCS object storage client uses the `fields` selector to limit the payload size when listing objects in the bucket. #3218
+* [ENHANCEMENT] Added shuffle sharding support to ruler. Added new metric `cortex_ruler_sync_rules_total`. #3235
 * [BUGFIX] No-longer-needed ingester operations for queries triggered by queriers and rulers are now canceled. #3178
 * [BUGFIX] Ruler: directories in the configured `rules-path` will be removed on startup and shutdown in order to ensure they don't persist between runs. #3195
 * [BUGFIX] Handle hash-collisions in the query path. #3192
 * [BUGFIX] Check for postgres rows errors. #3197
 * [BUGFIX] Ruler Experimental API: Don't allow rule groups without names or empty rule groups. #3210
 * [BUGFIX] Experimental Alertmanager API: Do not allow empty Alertmanager configurations or bad template filenames to be submitted through the configuration API. #3185
+* [BUGFIX] When using ruler sharding, moving all user rule groups from ruler to a different one and then back could end up with some user groups not being evaluated at all. #3235
 
 ## 1.4.0-rc.0 in progress
 

development/tsdb-blocks-storage-s3/.data-minio/.gitignore

@@ -2,3 +2,4 @@
 !.gitignore
 !cortex-tsdb
 !cortex-alertmanager
+!cortex-ruler

development/tsdb-blocks-storage-s3/.data-minio/cortex-ruler/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

development/tsdb-blocks-storage-s3/config/cortex.yaml

@@ -70,9 +70,19 @@ blocks_storage:
 ruler:
   enable_api: true
   storage:
-    type: configdb
-    configdb:
-      configs_api_url: http://configstore:80/
+    type: s3
+    s3:
+      bucketnames: cortex-ruler
+      s3forcepathstyle: true
+      s3: http://cortex:supersecret@minio.:9000
+  enable_sharding: true
+  ring:
+    heartbeat_period:   5s
+    heartbeat_timeout:  15s
+    kvstore:
+      store: consul
+      consul:
+        host: consul:8500
 
 alertmanager:
   enable_api: true

development/tsdb-blocks-storage-s3/config/grafana-agent.yaml

@@ -31,7 +31,7 @@ prometheus:
                 namespace: 'tsdb-blocks-storage-s3'
         - job_name: tsdb-blocks-storage-s3/ruler
           static_configs:
-            - targets: ['ruler:8005']
+            - targets: ['ruler-1:8021', 'ruler-2:8022']
               labels:
                 cluster: 'docker-compose'
                 namespace: 'tsdb-blocks-storage-s3'

development/tsdb-blocks-storage-s3/config/prometheus.yaml

@@ -24,7 +24,7 @@ scrape_configs:
           namespace: 'tsdb-blocks-storage-s3'
   - job_name: tsdb-blocks-storage-s3/ruler
     static_configs:
-      - targets: ['ruler:8005']
+      - targets: ['ruler-1:8021', 'ruler-2:8022']
         labels:
           cluster: 'docker-compose'
           namespace: 'tsdb-blocks-storage-s3'

development/tsdb-blocks-storage-s3/docker-compose.yml

@@ -21,11 +21,6 @@ services:
   memcached:
     image: memcached:1.6
 
-  configstore:
-    image: nginx
-    volumes:
-      - .data-configstore:/usr/share/nginx/html/private:ro
-
   prometheus:
     image: prom/prometheus:v2.16.0
     command: ["--config.file=/etc/prometheus/prometheus.yaml"]
@@ -178,80 +173,101 @@ services:
     volumes:
       - ./config:/cortex/config
 
-  ruler:
+  compactor:
     build:
       context:    .
       dockerfile: dev.dockerfile
     image: cortex
-    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18005 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=ruler -server.http-listen-port=8005 -server.grpc-listen-port=9005"]
+    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18006 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=compactor -server.http-listen-port=8006 -server.grpc-listen-port=9006"]
     depends_on:
       - consul
       - minio
     environment:
       - JAEGER_AGENT_HOST=jaeger
       - JAEGER_AGENT_PORT=6831
-      - JAEGER_TAGS=app=ruler
+      - JAEGER_TAGS=app=compactor
       - JAEGER_SAMPLER_TYPE=const
       - JAEGER_SAMPLER_PARAM=1
     ports:
-      - 8005:8005
-      - 18005:18005
+      - 8006:8006
+      - 18006:18006
     volumes:
       - ./config:/cortex/config
 
-  compactor:
+  query-frontend:
     build:
       context:    .
       dockerfile: dev.dockerfile
     image: cortex
-    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18006 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=compactor -server.http-listen-port=8006 -server.grpc-listen-port=9006"]
+    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18007 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=query-frontend -server.http-listen-port=8007 -server.grpc-listen-port=9007 -store.max-query-length=8760h"]
     depends_on:
       - consul
       - minio
     environment:
       - JAEGER_AGENT_HOST=jaeger
       - JAEGER_AGENT_PORT=6831
-      - JAEGER_TAGS=app=compactor
+      - JAEGER_TAGS=app=query-frontend
       - JAEGER_SAMPLER_TYPE=const
       - JAEGER_SAMPLER_PARAM=1
     ports:
-      - 8006:8006
-      - 18006:18006
+      - 8007:8007
+      - 18007:18007
     volumes:
       - ./config:/cortex/config
 
-  query-frontend:
+  alertmanager:
     build:
       context:    .
       dockerfile: dev.dockerfile
     image: cortex
-    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18007 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=query-frontend -server.http-listen-port=8007 -server.grpc-listen-port=9007 -store.max-query-length=8760h"]
+    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18010 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=alertmanager -server.http-listen-port=8010 -server.grpc-listen-port=9010"]
+    depends_on:
+      - consul
+      - minio
+    ports:
+      - 8010:8010
+      - 18010:18010
+    volumes:
+      - ./config:/cortex/config
+
+  ruler-1:
+    build:
+      context:    .
+      dockerfile: dev.dockerfile
+    image: cortex
+    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18021 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=ruler -server.http-listen-port=8021 -server.grpc-listen-port=9021"]
     depends_on:
       - consul
       - minio
     environment:
       - JAEGER_AGENT_HOST=jaeger
       - JAEGER_AGENT_PORT=6831
-      - JAEGER_TAGS=app=query-frontend
+      - JAEGER_TAGS=app=ruler-1
       - JAEGER_SAMPLER_TYPE=const
       - JAEGER_SAMPLER_PARAM=1
     ports:
-      - 8007:8007
-      - 18007:18007
+      - 8021:8021
+      - 18021:18021
     volumes:
       - ./config:/cortex/config
 
-  alertmanager:
+  ruler-2:
     build:
       context:    .
       dockerfile: dev.dockerfile
     image: cortex
-    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18010 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=alertmanager -server.http-listen-port=8010 -server.grpc-listen-port=9010"]
+    command: ["sh", "-c", "sleep 3 && exec ./dlv exec ./cortex --listen=:18022 --headless=true --api-version=2 --accept-multiclient --continue -- -config.file=./config/cortex.yaml -target=ruler -server.http-listen-port=8022 -server.grpc-listen-port=9022"]
     depends_on:
       - consul
       - minio
+    environment:
+      - JAEGER_AGENT_HOST=jaeger
+      - JAEGER_AGENT_PORT=6831
+      - JAEGER_TAGS=app=ruler-2
+      - JAEGER_SAMPLER_TYPE=const
+      - JAEGER_SAMPLER_PARAM=1
     ports:
-      - 8010:8010
-      - 18010:18010
+      - 8022:8022
+      - 18022:18022
     volumes:
       - ./config:/cortex/config

development/tsdb-blocks-storage-s3/goland/ruler-1.run.xml

@@ -0,0 +1,5 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="TSDB Ruler-1" type="GoRemoteDebugConfigurationType" factoryName="Go Remote" port="18021">
+    <method v="2" />
+  </configuration>
+</component>

development/tsdb-blocks-storage-s3/goland/ruler-2.run.xml

@@ -0,0 +1,5 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="TSDB Ruler-2" type="GoRemoteDebugConfigurationType" factoryName="Go Remote" port="18022">
+    <method v="2" />
+  </configuration>
+</component>

docs/configuration/config-file-reference.md

@@ -1117,6 +1117,10 @@ storage:
 # CLI flag: -ruler.enable-sharding
 [enable_sharding: <boolean> | default = false]
 
+# The sharding strategy to use. Supported values are: default, shuffle-sharding.
+# CLI flag: -ruler.sharding-strategy
+[sharding_strategy: <string> | default = "default"]
+
 # Time to spend searching for a pending ruler when shutting down.
 # CLI flag: -ruler.search-pending-for
 [search_pending_for: <duration> | default = 5m]
@@ -2866,6 +2870,12 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s
 # CLI flag: -ruler.evaluation-delay-duration
 [ruler_evaluation_delay_duration: <duration> | default = 0s]
 
+# The default tenant's shard size when the shuffle-sharding strategy is used by
+# ruler. When this setting is specified in the per-tenant overrides, a value of
+# 0 disables shuffle sharding for the tenant.
+# CLI flag: -ruler.tenant-shard-size
+[ruler_tenant_shard_size: <int> | default = 0]
+
 # The default tenant's shard size when the shuffle-sharding strategy is used.
 # Must be set when the store-gateway sharding is enabled with the
 # shuffle-sharding strategy. When this setting is specified in the per-tenant

docs/guides/shuffle-sharding.md

@@ -51,6 +51,7 @@ Cortex currently supports shuffle sharding in the following services:
 - [Ingesters](#ingesters-shuffle-sharding)
 - [Query-frontend](#query-frontend-shuffle-sharding)
 - [Store-gateway](#store-gateway-shuffle-sharding)
+- [Ruler](#ruler-shuffle-sharding)
 
 Shuffle sharding is **disabled by default** and needs to be explicitly enabled in the configuration.
 
@@ -92,3 +93,13 @@ When shuffle sharding is **enabled** via `-store-gateway.sharding-strategy=shuff
 _The shard size can be overridden on a per-tenant basis setting `store_gateway_tenant_shard_size` in the limits overrides configuration._
 
 _Please check out the [store-gateway documentation](../blocks-storage/store-gateway.md) for more information about how it works._
+
+### Ruler shuffle sharding
+
+Cortex ruler can run in three modes:
+
+1. **No sharding at all.** This is the most basic mode of the ruler. It is activated by using `-ruler.enable-sharding=false` (default) and works correctly only if single ruler is running. In this mode the Ruler loads all rules for all tenants.
+2. **Default sharding**, activated by using `-ruler.enable-sharding=true` and `-ruler.sharding-strategy=default` (default). In this mode rulers register themselves into the ring. Each ruler will then select and evaluate only those rules that it "owns".
+3. **Shuffle sharding**, activated by using `-ruler.enable-sharding=true` and `-ruler.sharding-strategy=shuffle-sharding`. Similarly to default sharding, rulers use the ring to distribute workload, but rule groups for each tenant can only be evaluated on limited number of rulers (`-ruler.tenant-shard-size`, can also be set per tenant as `ruler_tenant_shard_size` in overrides).
+
+Note that when using sharding strategy, each rule group is evaluated by single ruler only, there is no replication.

integration/chunks_storage_backends_test.go

@@ -260,7 +260,7 @@ func TestSwiftRuleStorage(t *testing.T) {
 	require.NoError(t, err)
 
 	// Get rules back.
-	rls, err := store.ListAllRuleGroups(ctx)
+	rls, err := store.LoadAllRuleGroups(ctx)
 	require.NoError(t, err)
 	require.Equal(t, 2, len(rls[userID]))
 
@@ -274,7 +274,7 @@ func TestSwiftRuleStorage(t *testing.T) {
 	require.NoError(t, err)
 
 	//Verify we only have the second rule group
-	rls, err = store.ListAllRuleGroups(ctx)
+	rls, err = store.LoadAllRuleGroups(ctx)
 	require.NoError(t, err)
 	require.Equal(t, 1, len(rls[userID]))
 	require.Equal(t, r2, rls[userID][0])

pkg/chunk/storage_client.go

@@ -58,7 +58,7 @@ type ReadBatchIterator interface {
 	Value() []byte
 }
 
-// ObjectClient is used to store arbitrary data in Object Store (S3/GCS/Azure/Etc)
+// ObjectClient is used to store arbitrary data in Object Store (S3/GCS/Azure/...)
 type ObjectClient interface {
 	PutObject(ctx context.Context, objectKey string, object io.ReadSeeker) error
 	GetObject(ctx context.Context, objectKey string) (io.ReadCloser, error)
@@ -85,4 +85,5 @@ type StorageObject struct {
 }
 
 // StorageCommonPrefix represents a common prefix aka a synthetic directory in Object Store.
+// It is guaranteed to always end with delimiter passed to List method.
 type StorageCommonPrefix string

pkg/cortex/cortex.go

@@ -165,7 +165,7 @@ func (c *Config) Validate(log log.Logger) error {
 	if err := c.ChunkStore.Validate(); err != nil {
 		return errors.Wrap(err, "invalid chunk store config")
 	}
-	if err := c.Ruler.Validate(); err != nil {
+	if err := c.Ruler.Validate(c.LimitsConfig); err != nil {
 		return errors.Wrap(err, "invalid ruler config")
 	}
 	if err := c.BlocksStorage.Validate(); err != nil {

pkg/cortex/modules.go

@@ -542,6 +542,7 @@ func (t *Cortex) initRuler() (serv services.Service, err error) {
 		prometheus.DefaultRegisterer,
 		util.Logger,
 		t.RulerStorage,
+		t.Overrides,
 	)
 	if err != nil {
 		return

pkg/querier/frontend/frontend_querier_queues.go

@@ -1,10 +1,10 @@
 package frontend
 
 import (
-	"crypto/md5"
-	"encoding/binary"
 	"math/rand"
 	"sort"
+
+	"github.com/cortexproject/cortex/pkg/util"
 )
 
 // This struct holds user queues for pending requests. It also keeps track of connected queriers,
@@ -89,7 +89,7 @@ func (q *queues) getOrAddQueue(userID string, maxQueriers int) chan *request {
 	if uq == nil {
 		uq = &userQueue{
 			ch:    make(chan *request, q.maxUserQueueSize),
-			seed:  getSeedForUser(userID),
+			seed:  util.ShuffleShardSeed(userID),
 			index: -1,
 		}
 		q.userQueues[userID] = uq
@@ -222,10 +222,3 @@ func shuffleQueriersForUser(userSeed int64, queriersToSelect int, allSortedQueri
 
 	return result
 }
-
-func getSeedForUser(user string) int64 {
-	d := md5.New()
-	_, _ = d.Write([]byte(user))
-	buf := d.Sum(nil)
-	return int64(binary.BigEndian.Uint64(buf))
-}

pkg/ring/model.go

@@ -162,6 +162,9 @@ func (i *IngesterDesc) IsHealthy(op Operation, heartbeatTimeout time.Duration) b
 
 	case BlocksRead:
 		healthy = i.State == ACTIVE
+
+	case Ruler:
+		healthy = i.State == ACTIVE
 	}
 
 	return healthy && time.Since(time.Unix(i.Timestamp, 0)) <= heartbeatTimeout