Repository: cortexproject/cortex
Pull request #2114 (2020-02-11 17:46:23)

Fix panic when alertmanager applies configs
After rebasing #1686, I noticed a major bug in the Alertmanager. The `notify.NewPipelineBuilder` function is called every time a change is applied causing a panic:
```
--- FAIL: TestMultitenantAlertmanager_loadAllConfigs (0.02s)
panic: runtime error: invalid memory address or nil pointer dereference [recovered]
	panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x16c72e6]

goroutine 15 [running]:
testing.tRunner.func1(0xc0001bca00)
	/usr/local/Cellar/go/1.13.6/libexec/src/testing/testing.go:874 +0x3a3
panic(0x1a2fbe0, 0x2440fe0)
	/usr/local/Cellar/go/1.13.6/libexec/src/runtime/panic.go:679 +0x1b2
github.com/prometheus/alertmanager/dispatch.(*Dispatcher).Stop(0x0)
	/Users/jtlisi/go/pkg/mod/github.com/prometheus/alertmanager@v0.19.0/dispatch/dispatch.go:219 +0x26
github.com/cortexproject/cortex/pkg/alertmanager.(*Alertmanager).ApplyConfig(0xc00036b200, 0x1b69ac3, 0x5, 0xc00039a3f0, 0x6, 0x0)
	/Users/jtlisi/GrafanaLabs/projects/cortex/pkg/alertmanager/alertmanager.go:192 +0x263
github.com/cortexproject/cortex/pkg/alertmanager.(*MultitenantAlertmanager).newAlertmanager(0xc000091800, 0x1b69ac3, 0x5, 0xc00039a3f0, 0x0, 0x0, 0xc00051ba40)
	/Users/jtlisi/GrafanaLabs/projects/cortex/pkg/alertmanager/multitenant.go:434 +0x21e
github.com/cortexproject/cortex/pkg/alertmanager.(*MultitenantAlertmanager).setConfig(0xc000091800, 0x1b69ac3, 0x5, 0x1b9a6c1, 0x34, 0x247cd18, 0x0, 0x0, 0x501de00, 0x100)
	/Users/jtlisi/GrafanaLabs/projects/cortex/pkg/alertmanager/multitenant.go:401 +0xad1
github.com/cortexproject/cortex/pkg/alertmanager.(*MultitenantAlertmanager).syncConfigs(0xc000091800, 0xc00051b770)
	/Users/jtlisi/GrafanaLabs/projects/cortex/pkg/alertmanager/multitenant.go:289 +0x2cd
github.com/cortexproject/cortex/pkg/alertmanager.(*MultitenantAlertmanager).updateConfigs(0xc000091800, 0xc0003bc000, 0xc0000406c0)
	/Users/jtlisi/GrafanaLabs/projects/cortex/pkg/alertmanager/multitenant.go:271 +0x5c
github.com/cortexproject/cortex/pkg/alertmanager.TestMultitenantAlertmanager_loadAllConfigs(0xc0001bca00)
	/Users/jtlisi/GrafanaLabs/projects/cortex/pkg/alertmanager/multitenant_test.go:77 +0x4f8
testing.tRunner(0xc0001bca00, 0x1c646f0)
	/usr/local/Cellar/go/1.13.6/libexec/src/testing/testing.go:909 +0xc9
created by testing.(*T).Run
	/usr/local/Cellar/go/1.13.6/libexec/src/testing/testing.go:960 +0x350
FAIL	github.com/cortexproject/cortex/pkg/alertmanager	0.380s
FAIL
Error: Tests failed.
```
This PR fixes the issue and builds a single PipelineBuilder for each per-tenant alertmanager.


**Checklist**
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`



Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index b703095a11..ba08a6a256 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -25,6 +25,12 @@ workflows:
         filters:
           tags:
             only: /^v[0-9]+(\.[0-9]+){2}(-.+|[^-.]*)$/
+    - integration-configs-db:
+        requires:
+        - build
+        filters:
+          tags:
+            only: /^v[0-9]+(\.[0-9]+){2}(-.+|[^-.]*)$/
     - build:
         filters:
           tags:
@@ -42,6 +48,7 @@ workflows:
         - test
         - lint
         - integration
+        - integration-configs-db
         filters:
           branches:
             only: master
@@ -90,7 +97,7 @@ jobs:
         name: Test
         command: make BUILD_IN_CONTAINER=false test
 
-  integration:
+  integration-configs-db:
     machine:
       image: ubuntu-1604:201903-01
 
@@ -104,17 +111,37 @@ jobs:
         command: |
           touch build-image/.uptodate
           MIGRATIONS_DIR=$(pwd)/cmd/cortex/migrations make BUILD_IMAGE=cortexproject/build-image:master-d74af5958 configs-integration-test
+
+  integration:
+    machine:
+      image: ubuntu-1604:201903-01
+
+    # Make sure the working directory is within the GOPATH, otherwise
+    # "go test" will not download module dependencies.
+    working_directory: ~/.go_workspace/src/github.com/cortexproject/cortex
+    steps:
+    - checkout
+    - restore_cache:
+        key: v1-cortex-{{ .Branch }}-{{ .Revision }}
     - run:
         name: Load Images
         command: ln -s /tmp/images ./docker-images; make BUILD_IN_CONTAINER=false load-images
     - run:
-        name: Ingester hand-over test
-        command: ./integration-tests/test-handover.sh
+        name: Preload Docker Images
+        # We download docker images used by integration tests so that all images are available
+        # locally and the download time doesn't account in the test execution time, which is subject
+        # to a timeout.
+        command: |
+          docker pull minio/minio:RELEASE.2019-12-30T05-45-39Z
+          docker pull amazon/dynamodb-local:1.11.477
+          docker pull consul:0.9
+          docker pull quay.io/cortexproject/cortex:v0.6.0
     - run:
-        name: Ingester flush test
+        name: Integration Tests
         command: |
-          ./integration-tests/stop-components.sh
-          ./integration-tests/test-flush.sh
+          export CORTEX_IMAGE="quay.io/cortexproject/cortex:${CIRCLE_TAG:-$(./tools/image-tag)}"
+          export CORTEX_INTEGRATION_DIR="/home/circleci/.go_workspace/src/github.com/cortexproject/cortex/integration"
+          go test -timeout 300s -v -count=1 ./integration
 
   build:
     <<: *defaults
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 002e25974e..c6f5aed839 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@
 * [FEATURE] Added user sub rings to distribute users to a subset of ingesters. #1947
   * `--experimental.distributor.user-subring-size`
 * [FEATURE] Added flag `-experimental.ruler.enable-api` to enable the ruler api which implements the Prometheus API `/api/v1/rules` and `/api/v1/alerts` endpoints under the configured `-http.prefix`. #1999
+* [FEATURE] Added sharding support to compactor when using the experimental TSDB blocks storage. #2113
 * [ENHANCEMENT] Experimental TSDB: Export TSDB Syncer metrics from Compactor component, they are prefixed with `cortex_compactor_`. #2023
 * [ENHANCEMENT] Experimental TSDB: Added dedicated flag `-experimental.tsdb.bucket-store.tenant-sync-concurrency` to configure the maximum number of concurrent tenants for which blocks are synched. #2026
 * [ENHANCEMENT] Experimental TSDB: Expose metrics for objstore operations (prefixed with `cortex_<component>_thanos_objstore_`, component being one of `ingester`, `querier` and `compactor`). #2027
@@ -28,9 +29,11 @@
   * `-server.grpc.keepalive.timeout`
 * [ENHANCEMENT] PostgreSQL: Bump up `github.com/lib/pq` from `v1.0.0` to `v1.3.0` to support PostgreSQL SCRAM-SHA-256 authentication. #2097
 * [ENHANCEMENT] Casandra: User no longer need `CREATE` privilege on `<all keyspaces>` if given keyspace exists. #2032
+* [BUGFIX] Alertmanager: fixed panic upon applying a new config, caused by duplicate metrics registration in the `NewPipelineBuilder` function. #211
 * [BUGFIX] Experimental TSDB: fixed `/all_user_stats` and `/api/prom/user_stats` endpoints when using the experimental TSDB blocks storage. #2042
 * [BUGFIX] Experimental TSDB: fixed ruler to correctly work with the experimental TSDB blocks storage. #2101
 * [BUGFIX] Azure Blob ChunkStore: Fixed issue causing `invalid chunk checksum` errors. #2074
+* [BUGFIX] The gauge `cortex_overrides_last_reload_successful` is now only exported by components that use a `RuntimeConfigManager`. Previously, for components that do not initialize a `RuntimeConfigManager` (such as the compactor) the gauge was initialized with 0 (indicating error state) and then never updated, resulting in a false-negative permanent error state. #2092
 
 Cortex 0.4.0 is the last version that can *write* denormalised tokens. Cortex 0.5.0 and above always write normalised tokens.
 
diff --git a/development/tsdb-blocks-storage-s3/config/cortex.yaml b/development/tsdb-blocks-storage-s3/config/cortex.yaml
index 48f6485e5e..4d80166b81 100644
--- a/development/tsdb-blocks-storage-s3/config/cortex.yaml
+++ b/development/tsdb-blocks-storage-s3/config/cortex.yaml
@@ -55,3 +55,14 @@ ruler:
 
 storage:
   engine: tsdb
+
+compactor:
+  compaction_interval: 30s
+  data_dir:            /tmp/cortex-compactor
+  consistency_delay:   1m
+  sharding_enabled:    true
+  sharding_ring:
+    kvstore:
+      store: consul
+      consul:
+        host: consul:8500
diff --git a/development/tsdb-blocks-storage-s3/docker-compose.yml b/development/tsdb-blocks-storage-s3/docker-compose.yml
index 9e94dee313..9104f6b4fe 100644
--- a/development/tsdb-blocks-storage-s3/docker-compose.yml
+++ b/development/tsdb-blocks-storage-s3/docker-compose.yml
@@ -94,3 +94,17 @@ services:
       - 8005:8005
     volumes:
       - ./config:/cortex/config
+
+  compactor:
+    build:
+      context:    .
+      dockerfile: dev.dockerfile
+    image: cortex
+    command: ["sh", "-c", "sleep 3 && exec ./cortex -config.file=./config/cortex.yaml -target=compactor -server.http-listen-port=8006 -server.grpc-listen-port=9006"]
+    depends_on:
+      - consul
+      - minio
+    ports:
+      - 8006:8006
+    volumes:
+      - ./config:/cortex/config
diff --git a/docs/operations/blocks-storage.md b/docs/operations/blocks-storage.md
index 93f24bf007..592366ef7c 100644
--- a/docs/operations/blocks-storage.md
+++ b/docs/operations/blocks-storage.md
@@ -41,7 +41,7 @@ The blocks chunks and the entire index is never fully downloaded by the queriers
 
 The index header is also stored to the local disk, in order to avoid to re-download it on subsequent restarts of a querier. For this reason, it's recommended - but not required - to run the querier with a persistent local disk. For example, if you're running the Cortex cluster in Kubernetes, you may use a StatefulSet with a persistent volume claim for the queriers.
 
-### Sharding and Replication
+### Series sharding and replication
 
 The series sharding and replication doesn't change based on the storage engine, so the general overview provided by the "[Cortex architecture](../architecture.md)" documentation applies to the blocks storage as well.
 
@@ -60,6 +60,17 @@ The **horizontal compaction** triggers after the vertical compaction and compact
 
 The compactor is **stateless**.
 
+#### Compactor sharding
+
+The compactor optionally supports sharding. When sharding is enabled, multiple compactor instances can coordinate to split the workload and shard blocks by tenant. All the blocks of a tenant are processed by a single compactor instance at any given time, but compaction for different tenants may simultaneously run on different compactor instances.
+
+Whenever the pool of compactors increase or decrease (ie. following up a scale up/down), tenants are resharded across the available compactor instances without any manual intervention. Compactors coordinate via the Cortex [hash ring](../architecture.md#the-hash-ring).
+
+#### Compactor HTTP endpoints
+
+- `GET /compactor_ring`<br />
+  Displays the status of the compactors ring, including the tokens owned by each compactor and an option to remove (forget) instances from the ring.
+
 ## Configuration
 
 The general [configuration documentation](../configuration/_index.md) also applied to a Cortex cluster running the blocks storage, with few differences:
@@ -251,14 +262,67 @@ compactor:
     # interval.
     # CLI flag: -compactor.compaction-retries
     [compaction_retries: <int> | default = 3]
+
+    # Shard tenants across multiple compactor instances. Sharding is required if
+    # you run multiple compactor instances, in order to coordinate compactions
+    # and avoid race conditions leading to the same tenant blocks simultaneously
+    # compacted by different instances.
+    # CLI flag: -compactor.sharding-enabled
+    [sharding_enabled: <bool> | default = false]
+
+    # Configures the ring used when sharding is enabled.
+    sharding_ring:
+      kvstore:
+        # Backend storage to use for the ring. Supported values are: consul, etcd,
+        # inmemory, multi, memberlist (experimental).
+        # CLI flag: -compactor.ring.store
+        [store: <string> | default = "consul"]
+
+        # The prefix for the keys in the store. Should end with a /.
+        # CLI flag: -compactor.ring.prefix
+        [prefix: <string> | default = "collectors/"]
+
+        # The consul_config configures the consul client.
+        # The CLI flags prefix for this block config is: compactor.ring
+        [consul: <consul_config>]
+
+        # The etcd_config configures the etcd client.
+        # The CLI flags prefix for this block config is: compactor.ring
+        [etcd: <etcd_config>]
+
+        # The memberlist_config configures the Gossip memberlist.
+        # The CLI flags prefix for this block config is: compactor.ring
+        [memberlist: <memberlist_config>]
+
+        multi:
+          # Primary backend storage used by multi-client.
+          # CLI flag: -compactor.ring.multi.primary
+          [primary: <string> | default = ""]
+
+          # Secondary backend storage used by multi-client.
+          # CLI flag: -compactor.ring.multi.secondary
+          [secondary: <string> | default = ""]
+
+          # Mirror writes to secondary store.
+          # CLI flag: -compactor.ring.multi.mirror-enabled
+          [mirror_enabled: <boolean> | default = false]
+
+          # Timeout for storing value to secondary store.
+          # CLI flag: -compactor.ring.multi.mirror-timeout
+          [mirror_timeout: <duration> | default = 2s]
+
+      # Period at which to heartbeat to the ring.
+      # CLI flag: -compactor.ring.heartbeat-period
+      [heartbeat_period: <duration> | default = 5m]
+
+      # The heartbeat timeout after which compactors are considered unhealthy
+      # within the ring.
+      # CLI flag: -compactor.ring.heartbeat-timeout
+      [heartbeat_timeout: <duration> | default = 1m]
 ```
 
 ## Known issues
 
-### Horizontal scalability
-
-The compactor currently doesn't support horizontal scalability and only 1 replica of the compactor should run at any given time within a Cortex cluster.
-
 ### Migrating from the chunks to the blocks storage
 
 Currently, no smooth migration path is provided to migrate from chunks to blocks storage. For this reason, the blocks storage can only be enabled in new Cortex clusters.
diff --git a/integration/README.md b/integration/README.md
new file mode 100644
index 0000000000..20dc6ba776
--- /dev/null
+++ b/integration/README.md
@@ -0,0 +1,10 @@
+# Cortex integration tests
+
+## Supported environment variables
+
+- `CORTEX_IMAGE`: Docker image used to run Cortex in integration tests (defaults to `quay.io/cortexproject/cortex:latest`)
+- `CORTEX_INTEGRATION_DIR`: Absolute path to the integration/ directory on the host (defaults to `$GOPATH/src/github.com/cortexproject/cortex/integration`)
+
+## Mounting files to Cortex container
+
+The `integration/` directory in the repository is mounted to `/integration` within the container. If you need to add fixture files (ie. config file) you can add the files to this repository (in `integration/`) and the files will be available within the Cortex container at `/integration`.
diff --git a/integration-tests/schema1.yaml b/integration/chunks-storage-schema-dynamodb.yaml
similarity index 100%
rename from integration-tests/schema1.yaml
rename to integration/chunks-storage-schema-dynamodb.yaml
diff --git a/integration/configs.go b/integration/configs.go
new file mode 100644
index 0000000000..c7f181b95b
--- /dev/null
+++ b/integration/configs.go
@@ -0,0 +1,26 @@
+package main
+
+import "github.com/cortexproject/cortex/integration/framework"
+
+var (
+	BlocksStorage = map[string]string{
+		"-store.engine":                                 "tsdb",
+		"-experimental.tsdb.backend":                    "s3",
+		"-experimental.tsdb.block-ranges-period":        "1m",
+		"-experimental.tsdb.bucket-store.sync-interval": "5s",
+		"-experimental.tsdb.retention-period":           "5m",
+		"-experimental.tsdb.ship-interval":              "1m",
+		"-experimental.tsdb.s3.access-key-id":           framework.MinioAccessKey,
+		"-experimental.tsdb.s3.secret-access-key":       framework.MinioSecretKey,
+		"-experimental.tsdb.s3.bucket-name":             "cortex",
+		"-experimental.tsdb.s3.endpoint":                "minio:9000",
+		"-experimental.tsdb.s3.insecure":                "true",
+	}
+
+	ChunksStorage = map[string]string{
+		"-dynamodb.url":                   "dynamodb://u:p@dynamodb.:8000",
+		"-dynamodb.poll-interval":         "1m",
+		"-config-yaml":                    "/integration/chunks-storage-schema-dynamodb.yaml",
+		"-table-manager.retention-period": "168h",
+	}
+)
diff --git a/integration/framework/client.go b/integration/framework/client.go
new file mode 100644
index 0000000000..4bf879d1a2
--- /dev/null
+++ b/integration/framework/client.go
@@ -0,0 +1,97 @@
+package framework
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/golang/snappy"
+	promapi "github.com/prometheus/client_golang/api"
+	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/prompb"
+)
+
+// Client is a client used to interact with Cortex in integration tests
+type Client struct {
+	distributorAddress string
+	timeout            time.Duration
+	httpClient         *http.Client
+	querierClient      promv1.API
+	orgID              string
+}
+
+// NewClient makes a new Cortex client
+func NewClient(distributorAddress string, querierAddress string, orgID string) (*Client, error) {
+	// Create querier API client
+	querierAPIClient, err := promapi.NewClient(promapi.Config{
+		Address:      "http://" + querierAddress + "/api/prom",
+		RoundTripper: &addOrgIDRoundTripper{orgID: orgID, next: http.DefaultTransport},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	c := &Client{
+		distributorAddress: distributorAddress,
+		timeout:            5 * time.Second,
+		httpClient:         &http.Client{},
+		querierClient:      promv1.NewAPI(querierAPIClient),
+		orgID:              orgID,
+	}
+
+	return c, nil
+}
+
+// Push the input timeseries to the remote endpoint
+func (c *Client) Push(timeseries []prompb.TimeSeries) (*http.Response, error) {
+	// Create write request
+	data, err := proto.Marshal(&prompb.WriteRequest{Timeseries: timeseries})
+	if err != nil {
+		return nil, err
+	}
+
+	// Create HTTP request
+	compressed := snappy.Encode(nil, data)
+	req, err := http.NewRequest("POST", fmt.Sprintf("http://%s/api/prom/push", c.distributorAddress), bytes.NewReader(compressed))
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Add("Content-Encoding", "snappy")
+	req.Header.Set("Content-Type", "application/x-protobuf")
+	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
+	req.Header.Set("X-Scope-OrgID", c.orgID)
+
+	ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
+	defer cancel()
+
+	// Execute HTTP request
+	res, err := c.httpClient.Do(req.WithContext(ctx))
+	if err != nil {
+		return nil, err
+	}
+
+	defer res.Body.Close()
+	return res, nil
+}
+
+// Query runs a query
+func (c *Client) Query(query string, ts time.Time) (model.Value, error) {
+	value, _, err := c.querierClient.Query(context.Background(), query, ts)
+	return value, err
+}
+
+type addOrgIDRoundTripper struct {
+	orgID string
+	next  http.RoundTripper
+}
+
+func (r *addOrgIDRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	req.Header.Set("X-Scope-OrgID", r.orgID)
+
+	return r.next.RoundTrip(req)
+}
diff --git a/integration/framework/scenario.go b/integration/framework/scenario.go
new file mode 100644
index 0000000000..bafe72336a
--- /dev/null
+++ b/integration/framework/scenario.go
@@ -0,0 +1,326 @@
+package framework
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+const (
+	MinioAccessKey = "cortex"
+	MinioSecretKey = "supersecret"
+	NetworkName    = "cortex-integration-test"
+)
+
+type Scenario struct {
+	services []*Service
+}
+
+func NewScenario() (*Scenario, error) {
+	s := &Scenario{
+		services: []*Service{},
+	}
+
+	// Force a shutdown in order to cleanup from a spurious situation in case
+	// the previous tests run didn't cleanup correctly
+	s.Shutdown()
+
+	// Setup the docker network
+	if out, err := RunCommandAndGetOutput("docker", "network", "create", NetworkName); err != nil {
+		fmt.Println(string(out))
+		return nil, errors.Wrapf(err, "create docker network '%s'", NetworkName)
+	}
+
+	return s, nil
+}
+
+func (s *Scenario) Service(name string) *Service {
+	for _, service := range s.services {
+		if service.name == name {
+			return service
+		}
+	}
+
+	return nil
+}
+
+func (s *Scenario) Endpoint(name string, port int) string {
+	service := s.Service(name)
+	if service == nil {
+		return ""
+	}
+
+	return service.Endpoint(port)
+}
+
+func (s *Scenario) StartService(service *Service) error {
+	fmt.Println("Starting", service.name)
+
+	// Ensure another service with the same name doesn't exist
+	if s.Service(service.name) != nil {
+		return fmt.Errorf("Another service with the same name '%s' has already been started", service.name)
+	}
+
+	// Start the service
+	if err := service.Start(); err != nil {
+		return err
+	}
+
+	// Add to the list of services
+	s.services = append(s.services, service)
+	return nil
+}
+
+func (s *Scenario) StartConsul() error {
+	return s.StartService(NewService(
+		"consul",
+		// If you change the image tag, remember to update it in the preloading done
+		// by CircleCI too (see .circleci/config.yml).
+		"consul:0.9",
+		NetworkName,
+		[]int{},
+		nil,
+		// Run consul in "dev" mode so that the initial leader election is immediate
+		NewCommand("agent", "-server", "-client=0.0.0.0", "-dev", "-log-level=err"),
+		nil,
+	))
+}
+
+// StartMinio starts minio server, used as a local replacement for S3.
+func (s *Scenario) StartMinio() error {
+	return s.StartService(NewService(
+		"minio",
+		// If you change the image tag, remember to update it in the preloading done
+		// by CircleCI too (see .circleci/config.yml).
+		"minio/minio:RELEASE.2019-12-30T05-45-39Z",
+		NetworkName,
+		[]int{9000},
+		map[string]string{
+			"MINIO_ACCESS_KEY": MinioAccessKey,
+			"MINIO_SECRET_KEY": MinioSecretKey,
+			"MINIO_BROWSER":    "off",
+			"ENABLE_HTTPS":     "0",
+		},
+		// Create the "cortex" bucket before starting minio
+		NewCommandWithoutEntrypoint("sh", "-c", "mkdir -p /data/cortex && minio server --quiet /data"),
+		NewReadinessProbe(9000, "/minio/health/ready", 200),
+	))
+}
+
+func (s *Scenario) StartDynamoDB() error {
+	return s.StartService(NewService(
+		"dynamodb",
+		// If you change the image tag, remember to update it in the preloading done
+		// by CircleCI too (see .circleci/config.yml).
+		"amazon/dynamodb-local:1.11.477",
+		NetworkName,
+		[]int{8000},
+		nil,
+		NewCommand("-jar", "DynamoDBLocal.jar", "-inMemory", "-sharedDb"),
+		// DynamoDB doesn't have a readiness probe, so we check if the / works even if returns 400
+		NewReadinessProbe(8000, "/", 400),
+	))
+}
+
+func (s *Scenario) StartDistributor(name string, flags map[string]string, image string) error {
+	if image == "" {
+		image = getDefaultCortexImage()
+	}
+
+	return s.StartService(NewService(
+		name,
+		image,
+		NetworkName,
+		[]int{80},
+		nil,
+		NewCommandWithoutEntrypoint("cortex", BuildArgs(MergeFlags(map[string]string{
+			"-target":                         "distributor",
+			"-log.level":                      "warn",
+			"-auth.enabled":                   "true",
+			"-distributor.replication-factor": "1",
+			// Configure the ingesters ring backend
+			"-ring.store":      "consul",
+			"-consul.hostname": "consul:8500",
+		}, flags))...),
+		NewReadinessProbe(80, "/ring", 200),
+	))
+}
+
+func (s *Scenario) StartQuerier(name string, flags map[string]string, image string) error {
+	if image == "" {
+		image = getDefaultCortexImage()
+	}
+
+	return s.StartService(NewService(
+		name,
+		image,
+		NetworkName,
+		[]int{80},
+		nil,
+		NewCommandWithoutEntrypoint("cortex", BuildArgs(MergeFlags(map[string]string{
+			"-target":                         "querier",
+			"-log.level":                      "warn",
+			"-distributor.replication-factor": "1",
+			// Configure the ingesters ring backend
+			"-ring.store":      "consul",
+			"-consul.hostname": "consul:8500",
+		}, flags))...),
+		NewReadinessProbe(80, "/ready", 204),
+	))
+}
+
+func (s *Scenario) StartIngester(name string, flags map[string]string, image string) error {
+	if image == "" {
+		image = getDefaultCortexImage()
+	}
+
+	return s.StartService(NewService(
+		name,
+		image,
+		NetworkName,
+		[]int{80},
+		nil,
+		NewCommandWithoutEntrypoint("cortex", BuildArgs(MergeFlags(map[string]string{
+			"-target":                        "ingester",
+			"-log.level":                     "warn",
+			"-ingester.final-sleep":          "0s",
+			"-ingester.join-after":           "0s",
+			"-ingester.min-ready-duration":   "0s",
+			"-ingester.concurrent-flushes":   "10",
+			"-ingester.max-transfer-retries": "10",
+			"-ingester.num-tokens":           "512",
+			// Configure the ingesters ring backend
+			"-ring.store":      "consul",
+			"-consul.hostname": "consul:8500",
+		}, flags))...),
+		NewReadinessProbe(80, "/ready", 204),
+	))
+}
+
+func (s *Scenario) StartTableManager(name string, flags map[string]string, image string) error {
+	if image == "" {
+		image = getDefaultCortexImage()
+	}
+
+	return s.StartService(NewService(
+		name,
+		image,
+		NetworkName,
+		[]int{80},
+		nil,
+		NewCommandWithoutEntrypoint("cortex", BuildArgs(MergeFlags(map[string]string{
+			"-target":    "table-manager",
+			"-log.level": "warn",
+		}, flags))...),
+		// The table-manager doesn't expose a readiness probe, so we just check if the / returns 404
+		NewReadinessProbe(80, "/", 404),
+	))
+}
+
+func (s *Scenario) StopService(name string) error {
+	service := s.Service(name)
+	if service == nil {
+		return fmt.Errorf("unable to stop service %s because does not exist", name)
+	}
+
+	if err := service.Stop(); err != nil {
+		return err
+	}
+
+	// Remove the service from the list of services
+	for i, entry := range s.services {
+		if entry.name == name {
+			s.services = append(s.services[:i], s.services[i+1:]...)
+			break
+		}
+	}
+
+	return nil
+}
+
+// WaitReady waits until one or more services are ready. A service
+// is ready when its readiness probe succeed. If a service has no
+// readiness probe, it's considered ready without doing any check.
+func (s *Scenario) WaitReady(services ...string) error {
+	for _, name := range services {
+		service := s.Service(name)
+		if service == nil {
+			return fmt.Errorf("Unable to wait for service '%s' ready because the service does not exist", name)
+		}
+
+		if err := service.WaitReady(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *Scenario) Shutdown() {
+	// Kill the services in the opposite order
+	for i := len(s.services) - 1; i >= 0; i-- {
+		if err := s.services[i].Kill(); err != nil {
+			fmt.Println("Unable to kill service", s.services[i].name, ":", err.Error())
+		}
+	}
+
+	// Ensure there are no leftover containers
+	if out, err := RunCommandAndGetOutput("docker", "ps", "-a", "--quiet", "--filter", fmt.Sprintf("network=%s", NetworkName)); err == nil {
+		for _, containerID := range strings.Split(string(out), "\n") {
+			containerID = strings.TrimSpace(containerID)
+			if containerID == "" {
+				continue
+			}
+
+			if out, err = RunCommandAndGetOutput("docker", "rm", "--force", containerID); err != nil {
+				fmt.Println(string(out))
+				fmt.Println("Unable to cleanup leftover container", containerID, ":", err.Error())
+			}
+		}
+	} else {
+		fmt.Println(string(out))
+		fmt.Println("Unable to cleanup leftover containers:", err.Error())
+	}
+
+	// Teardown the docker network. In case the network does not exists (ie. this function
+	// is called during the setup of the scenario) we skip the removal in order to not log
+	// an error which may be misleading.
+	if ok, err := existDockerNetwork(); ok || err != nil {
+		if out, err := RunCommandAndGetOutput("docker", "network", "rm", NetworkName); err != nil {
+			fmt.Println(string(out))
+			fmt.Println("Unable to remove docker network", NetworkName, ":", err.Error())
+		}
+	}
+}
+
+func existDockerNetwork() (bool, error) {
+	out, err := RunCommandAndGetOutput("docker", "network", "ls", "--quiet", "--filter", fmt.Sprintf("name=%s", NetworkName))
+	if err != nil {
+		fmt.Println(string(out))
+		fmt.Println("Unable to check if docker network", NetworkName, "exists:", err.Error())
+	}
+
+	return strings.TrimSpace(string(out)) != "", nil
+}
+
+// getDefaultCortexImage returns the Docker image to use to run Cortex.
+func getDefaultCortexImage() string {
+	// Get the cortex image from the CORTEX_IMAGE env variable,
+	// falling back to "quay.io/cortexproject/cortex:latest"
+	if os.Getenv("CORTEX_IMAGE") != "" {
+		return os.Getenv("CORTEX_IMAGE")
+	}
+
+	return "quay.io/cortexproject/cortex:latest"
+}
+
+// getIntegrationDir returns the absolute path of the integration dir on the host.
+func getIntegrationDir() string {
+	if os.Getenv("CORTEX_INTEGRATION_DIR") != "" {
+		return os.Getenv("CORTEX_INTEGRATION_DIR")
+	}
+
+	return os.Getenv("GOPATH") + "/src/github.com/cortexproject/cortex/integration"
+}
diff --git a/integration/framework/service.go b/integration/framework/service.go
new file mode 100644
index 0000000000..bb759b0631
--- /dev/null
+++ b/integration/framework/service.go
@@ -0,0 +1,346 @@
+package framework
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/prometheus/common/expfmt"
+
+	"github.com/cortexproject/cortex/pkg/util"
+)
+
+var (
+	dockerPortPattern = regexp.MustCompile(`^.*:(\d+)$`)
+)
+
+type Service struct {
+	name         string
+	image        string
+	networkName  string
+	networkPorts []int
+	env          map[string]string
+	command      *Command
+	readiness    *ReadinessProbe
+
+	// Maps container ports to dynamically binded local ports.
+	networkPortsContainerToLocal map[int]int
+
+	// Generic retry backoff.
+	retryBackoff *util.Backoff
+}
+
+func NewService(
+	name string,
+	image string,
+	networkName string,
+	networkPorts []int,
+	env map[string]string,
+	command *Command,
+	readiness *ReadinessProbe,
+) *Service {
+	return &Service{
+		name:                         name,
+		image:                        image,
+		networkName:                  networkName,
+		networkPorts:                 networkPorts,
+		env:                          env,
+		command:                      command,
+		networkPortsContainerToLocal: map[int]int{},
+		readiness:                    readiness,
+		retryBackoff: util.NewBackoff(context.Background(), util.BackoffConfig{
+			MinBackoff: 100 * time.Millisecond,
+			MaxBackoff: 200 * time.Millisecond,
+			MaxRetries: 100, // Sometimes the CI is slow
+		}),
+	}
+}
+
+func (s *Service) Start() (err error) {
+	// In case of any error, if the container was already created, we
+	// have to cleanup removing it. We ignore the error of the "docker rm"
+	// because we don't know if the container was created or not.
+	defer func() {
+		if err != nil {
+			_, _ = RunCommandAndGetOutput("docker", "rm", "--force", s.name)
+		}
+	}()
+
+	cmd := exec.Command("docker", s.buildDockerRunArgs()...)
+	cmd.Stdout = &LinePrefixWriter{prefix: s.name + ": ", wrapped: os.Stdout}
+	cmd.Stderr = &LinePrefixWriter{prefix: s.name + ": ", wrapped: os.Stderr}
+	if err = cmd.Start(); err != nil {
+		return err
+	}
+
+	// Wait until the container has been started
+	if err = s.WaitStarted(); err != nil {
+		return err
+	}
+
+	// Get the dynamic local ports mapped to the container
+	for _, containerPort := range s.networkPorts {
+		var out []byte
+		var localPort int
+
+		out, err = RunCommandAndGetOutput("docker", "port", s.name, strconv.Itoa(containerPort))
+		if err != nil {
+			err = errors.Wrapf(err, "unable to get mapping for port %d", containerPort)
+			return err
+		}
+
+		stdout := strings.TrimSpace(string(out))
+		matches := dockerPortPattern.FindStringSubmatch(stdout)
+		if len(matches) != 2 {
+			err = fmt.Errorf("unable to get mapping for port %d (output: %s)", containerPort, stdout)
+			return err
+		}
+
+		localPort, err = strconv.Atoi(matches[1])
+		if err != nil {
+			err = errors.Wrapf(err, "unable to get mapping for port %d", containerPort)
+			return err
+		}
+
+		s.networkPortsContainerToLocal[containerPort] = localPort
+	}
+
+	return nil
+}
+
+func (s *Service) Stop() error {
+	fmt.Println("Stopping", s.name)
+
+	if out, err := RunCommandAndGetOutput("docker", "stop", "--time=30", s.name); err != nil {
+		fmt.Println(string(out))
+		return err
+	}
+
+	return nil
+}
+
+func (s *Service) Kill() error {
+	fmt.Println("Killing", s.name)
+
+	if out, err := RunCommandAndGetOutput("docker", "stop", "--time=0", s.name); err != nil {
+		fmt.Println(string(out))
+		return err
+	}
+
+	return nil
+}
+
+func (s *Service) Endpoint(port int) string {
+	// Map the container port to the local port
+	localPort, ok := s.networkPortsContainerToLocal[port]
+	if !ok {
+		return ""
+	}
+
+	// Do not use "localhost" cause it doesn't work with the AWS DynamoDB client.
+	return fmt.Sprintf("127.0.0.1:%d", localPort)
+}
+
+func (s *Service) Ready() (bool, error) {
+	// Ensure the service has a readiness probe configure
+	if s.readiness == nil {
+		return true, nil
+	}
+
+	// Map the container port to the local port
+	localPort, ok := s.networkPortsContainerToLocal[s.readiness.port]
+	if !ok {
+		return false, fmt.Errorf("unknown port %d configured in the readiness probe", s.readiness.port)
+	}
+
+	return s.readiness.Ready(localPort), nil
+}
+
+func (s *Service) Metrics(port int) (string, error) {
+	// Map the container port to the local port
+	localPort, ok := s.networkPortsContainerToLocal[port]
+	if !ok {
+		return "", fmt.Errorf("unknown port %d", port)
+	}
+
+	// Fetch metrics
+	res, err := GetRequest(fmt.Sprintf("http://localhost:%d/metrics", localPort))
+	if err != nil {
+		return "", err
+	}
+
+	// Check the status code
+	if res.StatusCode < 200 || res.StatusCode >= 300 {
+		return "", fmt.Errorf("unexpected status code %d while fetching metrics", res.StatusCode)
+	}
+
+	defer res.Body.Close()
+	body, err := ioutil.ReadAll(res.Body)
+
+	return string(body), err
+}
+
+func (s *Service) WaitStarted() error {
+	for s.retryBackoff.Reset(); s.retryBackoff.Ongoing(); {
+		if err := exec.Command("docker", "inspect", s.name).Run(); err == nil {
+			return nil
+		}
+
+		s.retryBackoff.Wait()
+	}
+
+	return fmt.Errorf("docker container %s failed to start", s.name)
+}
+
+func (s *Service) WaitReady() error {
+	for s.retryBackoff.Reset(); s.retryBackoff.Ongoing(); {
+		ready, err := s.Ready()
+		if ready {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+
+		s.retryBackoff.Wait()
+	}
+
+	return fmt.Errorf("the service %s is not ready", s.name)
+}
+
+func (s *Service) WaitMetric(port int, metric string, value float64) error {
+	for s.retryBackoff.Reset(); s.retryBackoff.Ongoing(); {
+		metrics, err := s.Metrics(80)
+		if err != nil {
+			return err
+		}
+
+		var tp expfmt.TextParser
+		families, err := tp.TextToMetricFamilies(strings.NewReader(metrics))
+		if err != nil {
+			return err
+		}
+
+		// Check if the metric is exported
+		mf, ok := families[metric]
+		if ok {
+			for _, m := range mf.Metric {
+				if m.GetGauge() != nil && m.GetGauge().GetValue() == value {
+					return nil
+				} else if m.GetCounter() != nil && m.GetCounter().GetValue() == value {
+					return nil
+				} else if m.GetHistogram() != nil && float64(m.GetHistogram().GetSampleCount()) == value {
+					return nil
+				}
+			}
+		}
+
+		s.retryBackoff.Wait()
+	}
+
+	return fmt.Errorf("unable to find a metric %s with value %v", metric, value)
+}
+
+func (s *Service) buildDockerRunArgs() []string {
+	args := []string{"run", "--rm", "--net=" + s.networkName, "--name=" + s.name, "--hostname=" + s.name}
+
+	// Mount the integration/ directory into the container
+	args = append(args, "-v", fmt.Sprintf("%s:/integration", getIntegrationDir()))
+
+	// Environment variables
+	for name, value := range s.env {
+		args = append(args, "-e", name+"="+value)
+	}
+
+	// Published ports
+	for _, port := range s.networkPorts {
+		args = append(args, "-p", strconv.Itoa(port))
+	}
+
+	// Disable entrypoint if required
+	if s.command.entrypointDisabled {
+		args = append(args, "--entrypoint", "")
+	}
+
+	args = append(args, s.image)
+	args = append(args, s.command.cmd)
+	args = append(args, s.command.args...)
+	return args
+}
+
+type Command struct {
+	cmd                string
+	args               []string
+	entrypointDisabled bool
+}
+
+func NewCommand(cmd string, args ...string) *Command {
+	return &Command{
+		cmd:  cmd,
+		args: args,
+	}
+}
+
+func NewCommandWithoutEntrypoint(cmd string, args ...string) *Command {
+	return &Command{
+		cmd:                cmd,
+		args:               args,
+		entrypointDisabled: true,
+	}
+}
+
+type ReadinessProbe struct {
+	port           int
+	path           string
+	expectedStatus int
+}
+
+func NewReadinessProbe(port int, path string, expectedStatus int) *ReadinessProbe {
+	return &ReadinessProbe{
+		port:           port,
+		path:           path,
+		expectedStatus: expectedStatus,
+	}
+}
+
+func (p *ReadinessProbe) Ready(localPort int) bool {
+	res, err := GetRequest(fmt.Sprintf("http://localhost:%d%s", localPort, p.path))
+	if err != nil {
+		return false
+	}
+
+	res.Body.Close()
+
+	return res.StatusCode == p.expectedStatus
+}
+
+type LinePrefixWriter struct {
+	prefix  string
+	wrapped io.Writer
+}
+
+func (w *LinePrefixWriter) Write(p []byte) (n int, err error) {
+	for _, line := range strings.Split(string(p), "\n") {
+		// Skip empty lines
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+
+		// Write the prefix + line to the wrapped writer
+		_, err := w.wrapped.Write([]byte(w.prefix + line + "\n"))
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	return len(p), nil
+}
diff --git a/integration/framework/util.go b/integration/framework/util.go
new file mode 100644
index 0000000000..8e35ec9480
--- /dev/null
+++ b/integration/framework/util.go
@@ -0,0 +1,108 @@
+package framework
+
+import (
+	"math"
+	"net/http"
+	"net/url"
+	"os/exec"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/prometheus/prometheus/prompb"
+	awscommon "github.com/weaveworks/common/aws"
+)
+
+func RunCommandAndGetOutput(name string, args ...string) ([]byte, error) {
+	cmd := exec.Command(name, args...)
+	return cmd.CombinedOutput()
+}
+
+func MergeFlags(inputs ...map[string]string) map[string]string {
+	output := map[string]string{}
+
+	for _, input := range inputs {
+		for name, value := range input {
+			output[name] = value
+		}
+	}
+
+	return output
+}
+
+func BuildArgs(flags map[string]string) []string {
+	args := make([]string, 0, len(flags))
+
+	for name, value := range flags {
+		if value != "" {
+			args = append(args, name+"="+value)
+		} else {
+			args = append(args, name)
+		}
+	}
+
+	return args
+}
+
+func GetRequest(url string) (*http.Response, error) {
+	const timeout = 1 * time.Second
+
+	client := &http.Client{Timeout: timeout}
+	return client.Get(url)
+}
+
+// timeToMilliseconds returns the input time as milliseconds, using the same
+// formula used by Prometheus in order to get the same timestamp when asserting
+// on query results.
+func TimeToMilliseconds(t time.Time) int64 {
+	// The millisecond is rounded to the nearest
+	return int64(math.Round(float64(t.UnixNano()) / 1000000))
+}
+
+func NewDynamoClient(endpoint string) (*dynamodb.DynamoDB, error) {
+	dynamoURL, err := url.Parse(endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	dynamoConfig, err := awscommon.ConfigFromURL(dynamoURL)
+	if err != nil {
+		return nil, err
+	}
+
+	dynamoConfig = dynamoConfig.WithMaxRetries(0)
+	dynamoSession, err := session.NewSession(dynamoConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return dynamodb.New(dynamoSession), nil
+}
+
+func GenerateSeries(name string, ts time.Time) (series []prompb.TimeSeries, vector model.Vector) {
+	tsMillis := TimeToMilliseconds(ts)
+
+	// Generate the series
+	series = append(series, prompb.TimeSeries{
+		Labels: []prompb.Label{
+			{Name: labels.MetricName, Value: name},
+		},
+		Samples: []prompb.Sample{
+			{Value: float64(1), Timestamp: tsMillis},
+		},
+	})
+
+	// Generate the expected vector when querying it
+	metric := model.Metric{}
+	metric[labels.MetricName] = model.LabelValue(name)
+
+	vector = append(vector, &model.Sample{
+		Metric:    metric,
+		Value:     model.SampleValue(1),
+		Timestamp: model.Time(tsMillis),
+	})
+
+	return
+}
diff --git a/integration/util.go b/integration/util.go
new file mode 100644
index 0000000000..7263884e33
--- /dev/null
+++ b/integration/util.go
@@ -0,0 +1,13 @@
+package main
+
+import (
+	"github.com/cortexproject/cortex/integration/framework"
+)
+
+var (
+	// Expose some utilities form the framework so that we don't have to prefix them
+	// with the package name in tests.
+	mergeFlags      = framework.MergeFlags
+	newDynamoClient = framework.NewDynamoClient
+	generateSeries  = framework.GenerateSeries
+)
diff --git a/pkg/alertmanager/alertmanager.go b/pkg/alertmanager/alertmanager.go
index 8deae31472..6371de0b3b 100644
--- a/pkg/alertmanager/alertmanager.go
+++ b/pkg/alertmanager/alertmanager.go
@@ -52,19 +52,20 @@ type Config struct {
 
 // An Alertmanager manages the alerts for one user.
 type Alertmanager struct {
-	cfg        *Config
-	api        *api.API
-	logger     log.Logger
-	nflog      *nflog.Log
-	silences   *silence.Silences
-	marker     types.Marker
-	alerts     *mem.Alerts
-	dispatcher *dispatch.Dispatcher
-	inhibitor  *inhibit.Inhibitor
-	stop       chan struct{}
-	wg         sync.WaitGroup
-	mux        *http.ServeMux
-	registry   *prometheus.Registry
+	cfg             *Config
+	api             *api.API
+	logger          log.Logger
+	nflog           *nflog.Log
+	silences        *silence.Silences
+	marker          types.Marker
+	alerts          *mem.Alerts
+	dispatcher      *dispatch.Dispatcher
+	inhibitor       *inhibit.Inhibitor
+	pipelineBuilder *notify.PipelineBuilder
+	stop            chan struct{}
+	wg              sync.WaitGroup
+	mux             *http.ServeMux
+	registry        *prometheus.Registry
 }
 
 var webReload = make(chan chan error)
@@ -130,6 +131,8 @@ func New(cfg *Config) (*Alertmanager, error) {
 		am.silences.SetBroadcast(c.Broadcast)
 	}
 
+	am.pipelineBuilder = notify.NewPipelineBuilder(am.registry)
+
 	am.wg.Add(1)
 	go func() {
 		am.silences.Maintenance(15*time.Minute, filepath.Join(cfg.DataDir, silencesID), am.stop)
@@ -205,8 +208,8 @@ func (am *Alertmanager) ApplyConfig(userID string, conf *config.Config) error {
 	if err != nil {
 		return nil
 	}
-	pipelineBuilder := notify.NewPipelineBuilder(am.registry)
-	pipeline := pipelineBuilder.New(
+
+	pipeline := am.pipelineBuilder.New(
 		integrationsMap,
 		waitFunc,
 		am.inhibitor,
diff --git a/pkg/compactor/compactor.go b/pkg/compactor/compactor.go
index 6c9604d72b..70119c40e0 100644
--- a/pkg/compactor/compactor.go
+++ b/pkg/compactor/compactor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"hash/fnv"
 	"path"
 	"strings"
 	"sync"
@@ -19,6 +20,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/compact/downsample"
 	"github.com/thanos-io/thanos/pkg/objstore"
 
+	"github.com/cortexproject/cortex/pkg/ring"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
 )
@@ -33,6 +35,10 @@ type Config struct {
 	CompactionInterval   time.Duration            `yaml:"compaction_interval"`
 	CompactionRetries    int                      `yaml:"compaction_retries"`
 
+	// Compactors sharding.
+	ShardingEnabled bool       `yaml:"sharding_enabled"`
+	ShardingRing    RingConfig `yaml:"sharding_ring"`
+
 	// No need to add options to customize the retry backoff,
 	// given the defaults should be fine, but allow to override
 	// it in tests.
@@ -42,6 +48,8 @@ type Config struct {
 
 // RegisterFlags registers the Compactor flags.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
+	cfg.ShardingRing.RegisterFlags(f)
+
 	cfg.BlockRanges = cortex_tsdb.DurationList{2 * time.Hour, 12 * time.Hour, 24 * time.Hour}
 	cfg.retryMinBackoff = 10 * time.Second
 	cfg.retryMaxBackoff = time.Minute
@@ -53,6 +61,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.StringVar(&cfg.DataDir, "compactor.data-dir", "./data", "Data directory in which to cache blocks and process compactions")
 	f.DurationVar(&cfg.CompactionInterval, "compactor.compaction-interval", time.Hour, "The frequency at which the compaction runs")
 	f.IntVar(&cfg.CompactionRetries, "compactor.compaction-retries", 3, "How many times to retry a failed compaction during a single compaction interval")
+	f.BoolVar(&cfg.ShardingEnabled, "compactor.sharding-enabled", false, "Shard tenants across multiple compactor instances. Sharding is required if you run multiple compactor instances, in order to coordinate compactions and avoid race conditions leading to the same tenant blocks simultaneously compacted by different instances.")
 }
 
 // Compactor is a multi-tenant TSDB blocks compactor based on Thanos.
@@ -75,6 +84,10 @@ type Compactor struct {
 	ctx       context.Context
 	cancelCtx context.CancelFunc
 
+	// Ring used for sharding compactions.
+	ringLifecycler *ring.Lifecycler
+	ring           *ring.Ring
+
 	// Metrics.
 	compactionRunsStarted   prometheus.Counter
 	compactionRunsCompleted prometheus.Counter
@@ -104,7 +117,13 @@ func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.Config, logger log
 		return nil, errors.Wrap(err, "failed to create TSDB compactor")
 	}
 
-	return newCompactor(ctx, cancelCtx, compactorCfg, storageCfg, bucketClient, tsdbCompactor, logger, registerer)
+	cortexCompactor, err := newCompactor(ctx, cancelCtx, compactorCfg, storageCfg, bucketClient, tsdbCompactor, logger, registerer)
+	if err != nil {
+		cancelCtx()
+		return nil, errors.Wrap(err, "failed to create Cortex blocks compactor")
+	}
+
+	return cortexCompactor, nil
 }
 
 func newCompactor(
@@ -139,29 +158,70 @@ func newCompactor(
 		}),
 	}
 
+	// Initialize the compactors ring if sharding is enabled.
+	if compactorCfg.ShardingEnabled {
+		lifecyclerCfg := compactorCfg.ShardingRing.ToLifecyclerConfig()
+		lifecycler, err := ring.NewLifecycler(lifecyclerCfg, ring.NewNoopFlushTransferer(), "compactor", ring.CompactorRingKey, false)
+		if err != nil {
+			return nil, errors.Wrap(err, "unable to initialize compactor ring lifecycler")
+		}
+
+		lifecycler.Start()
+		c.ringLifecycler = lifecycler
+
+		ring, err := ring.New(lifecyclerCfg.RingConfig, "compactor", ring.CompactorRingKey)
+		if err != nil {
+			return nil, errors.Wrap(err, "unable to initialize compactor ring")
+		}
+
+		c.ring = ring
+	}
+
 	// Register metrics.
 	if registerer != nil {
 		registerer.MustRegister(c.compactionRunsStarted, c.compactionRunsCompleted, c.compactionRunsFailed)
 		c.syncerMetrics = newSyncerMetrics(registerer)
 	}
 
+	return c, nil
+}
+
+// Start the compactor.
+func (c *Compactor) Start() {
 	// Start the compactor loop.
 	c.runner.Add(1)
 	go c.run()
-
-	return c, nil
 }
 
-// Shutdown the compactor and waits until done. This may take some time
+// Stop the compactor and waits until done. This may take some time
 // if there's a on-going compaction.
-func (c *Compactor) Shutdown() {
+func (c *Compactor) Stop() {
 	c.cancelCtx()
 	c.runner.Wait()
+
+	// Shutdown the ring lifecycler (if any)
+	if c.ringLifecycler != nil {
+		c.ringLifecycler.Shutdown()
+	}
+
+	if c.ring != nil {
+		c.ring.Stop()
+	}
 }
 
 func (c *Compactor) run() {
 	defer c.runner.Done()
 
+	// If sharding is enabled we should wait until this instance is
+	// ACTIVE within the ring.
+	if c.compactorCfg.ShardingEnabled {
+		level.Info(c.logger).Log("msg", "waiting until compactor is ACTIVE in the ring")
+		if err := c.waitRingActive(); err != nil {
+			return
+		}
+		level.Info(c.logger).Log("msg", "compactor is ACTIVE in the ring")
+	}
+
 	// Run an initial compaction before starting the interval.
 	c.compactUsersWithRetries(c.ctx)
 
@@ -215,6 +275,17 @@ func (c *Compactor) compactUsers(ctx context.Context) bool {
 			return false
 		}
 
+		// If sharding is enabled, ensure the user ID belongs to our shard.
+		if c.compactorCfg.ShardingEnabled {
+			if owned, err := c.ownUser(userID); err != nil {
+				level.Warn(c.logger).Log("msg", "unable to check if user is owned by this shard", "user", userID, "err", err)
+				continue
+			} else if !owned {
+				level.Debug(c.logger).Log("msg", "skipping user because not owned by this shard", "user", userID)
+				continue
+			}
+		}
+
 		level.Info(c.logger).Log("msg", "starting compaction of user blocks", "user", userID)
 
 		if err = c.compactUser(ctx, userID); err != nil {
@@ -290,3 +361,43 @@ func (c *Compactor) discoverUsers(ctx context.Context) ([]string, error) {
 
 	return users, err
 }
+
+func (c *Compactor) ownUser(userID string) (bool, error) {
+	// Hash the user ID.
+	hasher := fnv.New32a()
+	_, _ = hasher.Write([]byte(userID))
+	userHash := hasher.Sum32()
+
+	// Check whether this compactor instance owns the user.
+	rs, err := c.ring.Get(userHash, ring.Read, []ring.IngesterDesc{})
+	if err != nil {
+		return false, err
+	}
+
+	if len(rs.Ingesters) != 1 {
+		return false, fmt.Errorf("unexpected number of compactors in the shard (expected 1, got %d)", len(rs.Ingesters))
+	}
+
+	return rs.Ingesters[0].Addr == c.ringLifecycler.Addr, nil
+}
+
+func (c *Compactor) waitRingActive() error {
+	for {
+		// Check if the ingester is ACTIVE in the ring and our ring client
+		// has detected it.
+		if rs, err := c.ring.GetAll(); err == nil {
+			for _, i := range rs.Ingesters {
+				if i.GetAddr() == c.ringLifecycler.Addr && i.GetState() == ring.ACTIVE {
+					return nil
+				}
+			}
+		}
+
+		select {
+		case <-time.After(time.Second):
+			// Nothing to do
+		case <-c.ctx.Done():
+			return c.ctx.Err()
+		}
+	}
+}
diff --git a/pkg/compactor/compactor_http.go b/pkg/compactor/compactor_http.go
new file mode 100644
index 0000000000..a61295a93e
--- /dev/null
+++ b/pkg/compactor/compactor_http.go
@@ -0,0 +1,35 @@
+package compactor
+
+import (
+	"net/http"
+
+	"github.com/go-kit/kit/log/level"
+)
+
+const (
+	shardingDisabledPage = `
+	<!DOCTYPE html>
+	<html>
+		<head>
+			<meta charset="UTF-8">
+			<title>Cortex Compactor Ring</title>
+		</head>
+		<body>
+			<h1>Cortex Compactor Ring</h1>
+			<p>Compactor has no ring because sharding is disabled.</p>
+		</body>
+	</html>
+	`
+)
+
+func (c *Compactor) RingHandler(w http.ResponseWriter, req *http.Request) {
+	if c.compactorCfg.ShardingEnabled {
+		c.ring.ServeHTTP(w, req)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+	if _, err := w.Write([]byte(shardingDisabledPage)); err != nil {
+		level.Error(c.logger).Log("msg", "unable to serve compactor ring page", "err", err)
+	}
+}
diff --git a/pkg/compactor/compactor_ring.go b/pkg/compactor/compactor_ring.go
new file mode 100644
index 0000000000..379db16f1c
--- /dev/null
+++ b/pkg/compactor/compactor_ring.go
@@ -0,0 +1,91 @@
+package compactor
+
+import (
+	"flag"
+	"os"
+	"time"
+
+	"github.com/go-kit/kit/log/level"
+
+	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/ring/kv"
+	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
+)
+
+// RingConfig masks the ring lifecycler config which contains
+// many options not really required by the compactors ring. This config
+// is used to strip down the config to the minimum, and avoid confusion
+// to the user.
+type RingConfig struct {
+	KVStore          kv.Config     `yaml:"kvstore,omitempty"`
+	HeartbeatPeriod  time.Duration `yaml:"heartbeat_period,omitempty"`
+	HeartbeatTimeout time.Duration `yaml:"heartbeat_timeout,omitempty"`
+
+	// Instance details
+	InstanceID             string   `yaml:"instance_id" doc:"hidden"`
+	InstanceInterfaceNames []string `yaml:"instance_interface_names" doc:"hidden"`
+	InstancePort           int      `yaml:"instance_port" doc:"hidden"`
+	InstanceAddr           string   `yaml:"instance_addr" doc:"hidden"`
+
+	// Injected internally
+	ListenPort int `yaml:"-"`
+}
+
+// RegisterFlags adds the flags required to config this to the given FlagSet
+func (cfg *RingConfig) RegisterFlags(f *flag.FlagSet) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		level.Error(util.Logger).Log("msg", "failed to get hostname", "err", err)
+		os.Exit(1)
+	}
+
+	// Ring flags
+	cfg.KVStore.RegisterFlagsWithPrefix("compactor.ring.", "collectors/", f)
+	f.DurationVar(&cfg.HeartbeatPeriod, "compactor.ring.heartbeat-period", 5*time.Second, "Period at which to heartbeat to the ring.")
+	f.DurationVar(&cfg.HeartbeatTimeout, "compactor.ring.heartbeat-timeout", time.Minute, "The heartbeat timeout after which compactors are considered unhealthy within the ring.")
+
+	// Instance flags
+	cfg.InstanceInterfaceNames = []string{"eth0", "en0"}
+	f.Var((*flagext.Strings)(&cfg.InstanceInterfaceNames), "compactor.ring.instance-interface", "Name of network interface to read address from.")
+	f.StringVar(&cfg.InstanceAddr, "compactor.ring.instance-addr", "", "IP address to advertise in the ring.")
+	f.IntVar(&cfg.InstancePort, "compactor.ring.instance-port", 0, "Port to advertise in the ring (defaults to server.grpc-listen-port).")
+	f.StringVar(&cfg.InstanceID, "compactor.ring.instance-id", hostname, "Instance ID to register in the ring.")
+}
+
+// ToLifecyclerConfig returns a LifecyclerConfig based on the compactor
+// ring config.
+func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
+	// We have to make sure that the ring.LifecyclerConfig and ring.Config
+	// defaults are preserved
+	lc := ring.LifecyclerConfig{}
+	rc := ring.Config{}
+
+	flagext.DefaultValues(&lc)
+	flagext.DefaultValues(&rc)
+
+	// Configure ring
+	rc.KVStore = cfg.KVStore
+	rc.HeartbeatTimeout = cfg.HeartbeatTimeout
+	rc.ReplicationFactor = 1
+
+	// Configure lifecycler
+	lc.RingConfig = rc
+	lc.ListenPort = &cfg.ListenPort
+	lc.Addr = cfg.InstanceAddr
+	lc.Port = cfg.InstancePort
+	lc.ID = cfg.InstanceID
+	lc.InfNames = cfg.InstanceInterfaceNames
+	lc.SkipUnregister = false
+	lc.HeartbeatPeriod = cfg.HeartbeatPeriod
+	lc.ObservePeriod = 0
+	lc.JoinAfter = 0
+	lc.MinReadyDuration = 0
+	lc.FinalSleep = 0
+
+	// We use a safe default instead of exposing to config option to the user
+	// in order to simplify the config.
+	lc.NumTokens = 512
+
+	return lc
+}
diff --git a/pkg/cortex/modules.go b/pkg/cortex/modules.go
index 7fdc8cd94e..faf5a2d8f4 100644
--- a/pkg/cortex/modules.go
+++ b/pkg/cortex/modules.go
@@ -193,7 +193,7 @@ func (t *Cortex) initRuntimeConfig(cfg *Config) (err error) {
 	// make sure to set default limits before we start loading configuration into memory
 	validation.SetDefaultLimitsForYAMLUnmarshalling(cfg.LimitsConfig)
 
-	t.runtimeConfig, err = runtimeconfig.NewRuntimeConfigManager(cfg.RuntimeConfig)
+	t.runtimeConfig, err = runtimeconfig.NewRuntimeConfigManager(cfg.RuntimeConfig, prometheus.DefaultRegisterer)
 	return err
 }
 
@@ -490,12 +490,23 @@ func (t *Cortex) stopAlertmanager() error {
 }
 
 func (t *Cortex) initCompactor(cfg *Config) (err error) {
+	cfg.Compactor.ShardingRing.ListenPort = cfg.Server.GRPCListenPort
+
 	t.compactor, err = compactor.NewCompactor(cfg.Compactor, cfg.TSDB, util.Logger, prometheus.DefaultRegisterer)
-	return err
+	if err != nil {
+		return err
+	}
+
+	t.compactor.Start()
+
+	// Expose HTTP endpoints.
+	t.server.HTTP.HandleFunc("/compactor_ring", t.compactor.RingHandler)
+
+	return nil
 }
 
 func (t *Cortex) stopCompactor() error {
-	t.compactor.Shutdown()
+	t.compactor.Stop()
 	return nil
 }
 
diff --git a/pkg/ring/http.go b/pkg/ring/http.go
index 9ca4ce2182..12fea72d09 100644
--- a/pkg/ring/http.go
+++ b/pkg/ring/http.go
@@ -10,12 +10,11 @@ import (
 	"time"
 
 	"github.com/go-kit/kit/log/level"
-	"github.com/gogo/protobuf/proto"
 
 	"github.com/cortexproject/cortex/pkg/util"
 )
 
-const tpl = `
+const pageContent = `
 <!DOCTYPE html>
 <html>
 	<head>
@@ -30,7 +29,7 @@ const tpl = `
 			<table width="100%" border="1">
 				<thead>
 					<tr>
-						<th>Ingester</th>
+						<th>Instance ID</th>
 						<th>State</th>
 						<th>Address</th>
 						<th>Last Heartbeat</th>
@@ -50,7 +49,7 @@ const tpl = `
 						<td>{{ .State }}</td>
 						<td>{{ .Address }}</td>
 						<td>{{ .Timestamp }}</td>
-						<td>{{ .Tokens }}</td>
+						<td>{{ .NumTokens }}</td>
 						<td>{{ .Ownership }}%</td>
 						<td><button name="forget" value="{{ .ID }}" type="submit">Forget</button></td>
 					</tr>
@@ -59,21 +58,32 @@ const tpl = `
 			</table>
 			<br>
 			{{ if .ShowTokens }}
-			<input type="button" value="Hide Ingester Tokens" onclick="window.location.href = '?tokens=false' " />
+			<input type="button" value="Hide Tokens" onclick="window.location.href = '?tokens=false' " />
 			{{ else }}
-			<input type="button" value="Show Ingester Tokens" onclick="window.location.href = '?tokens=true'" />
+			<input type="button" value="Show Tokens" onclick="window.location.href = '?tokens=true'" />
+			{{ end }}
+
+			{{ if .ShowTokens }}
+				{{ range $i, $ing := .Ingesters }}
+					<h2>Instance: {{ .ID }}</h2>
+					<p>
+						Tokens:<br />
+						{{ range $token := .Tokens }}
+							{{ $token }}
+						{{ end }}
+					</p>
+				{{ end }}
 			{{ end }}
-			<pre>{{ .Ring }}</pre>
 		</form>
 	</body>
 </html>`
 
-var tmpl *template.Template
+var pageTemplate *template.Template
 
 func init() {
 	t := template.New("webpage")
 	t.Funcs(template.FuncMap{"mod": func(i, j int) bool { return i%j == 0 }})
-	tmpl = template.Must(t.Parse(tpl))
+	pageTemplate = template.Must(t.Parse(pageContent))
 }
 
 func (r *Ring) forget(ctx context.Context, id string) error {
@@ -93,7 +103,7 @@ func (r *Ring) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	if req.Method == http.MethodPost {
 		ingesterID := req.FormValue("forget")
 		if err := r.forget(req.Context(), ingesterID); err != nil {
-			level.Error(util.WithContext(req.Context(), util.Logger)).Log("msg", "error forgetting ingester", "err", err)
+			level.Error(util.WithContext(req.Context(), util.Logger)).Log("msg", "error forgetting instance", "err", err)
 		}
 
 		// Implement PRG pattern to prevent double-POST and work with CSRF middleware.
@@ -118,7 +128,7 @@ func (r *Ring) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	sort.Strings(ingesterIDs)
 
 	ingesters := []interface{}{}
-	tokens, owned := countTokens(r.ringDesc, r.ringTokens)
+	_, owned := countTokens(r.ringDesc, r.ringTokens)
 	for _, id := range ingesterIDs {
 		ing := r.ringDesc.Ingesters[id]
 		timestamp := time.Unix(ing.Timestamp, 0)
@@ -129,35 +139,30 @@ func (r *Ring) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 
 		ingesters = append(ingesters, struct {
 			ID, State, Address, Timestamp string
-			Tokens                        uint32
+			Tokens                        []uint32
+			NumTokens                     int
 			Ownership                     float64
 		}{
 			ID:        id,
 			State:     state,
 			Address:   ing.Addr,
 			Timestamp: timestamp.String(),
-			Tokens:    tokens[id],
+			Tokens:    ing.Tokens,
+			NumTokens: len(ing.Tokens),
 			Ownership: (float64(owned[id]) / float64(math.MaxUint32)) * 100,
 		})
 	}
 
 	tokensParam := req.URL.Query().Get("tokens")
-	var ringDescString string
-	showTokens := false
-	if tokensParam == "true" {
-		ringDescString = proto.MarshalTextString(r.ringDesc)
-		showTokens = true
-	}
-	if err := tmpl.Execute(w, struct {
+
+	if err := pageTemplate.Execute(w, struct {
 		Ingesters  []interface{}
 		Now        time.Time
-		Ring       string
 		ShowTokens bool
 	}{
 		Ingesters:  ingesters,
 		Now:        time.Now(),
-		Ring:       ringDescString,
-		ShowTokens: showTokens,
+		ShowTokens: tokensParam == "true",
 	}); err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
diff --git a/pkg/ring/ring.go b/pkg/ring/ring.go
index 4e08e33c2d..b944506f57 100644
--- a/pkg/ring/ring.go
+++ b/pkg/ring/ring.go
@@ -31,6 +31,9 @@ const (
 
 	// DistributorRingKey is the key under which we store the distributors ring in the KVStore.
 	DistributorRingKey = "distributor"
+
+	// CompactorRingKey is the key under which we store the compactors ring in the KVStore.
+	CompactorRingKey = "compactor"
 )
 
 // ReadRing represents the read interface to the ring.
diff --git a/pkg/util/runtimeconfig/manager.go b/pkg/util/runtimeconfig/manager.go
index 88da6f6a82..87d777037c 100644
--- a/pkg/util/runtimeconfig/manager.go
+++ b/pkg/util/runtimeconfig/manager.go
@@ -7,16 +7,10 @@ import (
 
 	"github.com/go-kit/kit/log/level"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/cortexproject/cortex/pkg/util"
 )
 
-var overridesReloadSuccess = promauto.NewGauge(prometheus.GaugeOpts{
-	Name: "cortex_overrides_last_reload_successful",
-	Help: "Whether the last config reload attempt was successful.",
-})
-
 // Loader loads the configuration from file.
 type Loader func(filename string) (interface{}, error)
 
@@ -45,13 +39,23 @@ type Manager struct {
 
 	configMtx sync.RWMutex
 	config    interface{}
+
+	configLoadSuccess prometheus.Gauge
 }
 
 // NewRuntimeConfigManager creates an instance of Manager and starts reload config loop based on config
-func NewRuntimeConfigManager(cfg ManagerConfig) (*Manager, error) {
+func NewRuntimeConfigManager(cfg ManagerConfig, registerer prometheus.Registerer) (*Manager, error) {
 	mgr := Manager{
 		cfg:  cfg,
 		quit: make(chan struct{}),
+		configLoadSuccess: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "cortex_overrides_last_reload_successful",
+			Help: "Whether the last config reload attempt was successful.",
+		}),
+	}
+
+	if registerer != nil {
+		registerer.MustRegister(mgr.configLoadSuccess)
 	}
 
 	if cfg.LoadPath != "" {
@@ -120,10 +124,10 @@ func (om *Manager) loop() {
 func (om *Manager) loadConfig() error {
 	cfg, err := om.cfg.Loader(om.cfg.LoadPath)
 	if err != nil {
-		overridesReloadSuccess.Set(0)
+		om.configLoadSuccess.Set(0)
 		return err
 	}
-	overridesReloadSuccess.Set(1)
+	om.configLoadSuccess.Set(1)
 
 	om.setConfig(cfg)
 	om.callListeners(cfg)


Test patch:
diff --git a/integration-tests/common.sh b/integration-tests/common.sh
deleted file mode 100755
index 324968c1ab..0000000000
--- a/integration-tests/common.sh
+++ /dev/null
@@ -1,30 +0,0 @@
-# file intended to be sourced by other shell scripts
-
-TEST_DIR="$( cd "$(dirname "$0")" ; pwd -P )"
-
-IMAGE_TAG=${CIRCLE_TAG:-$($TEST_DIR/../tools/image-tag)}
-IMAGE_PREFIX="${IMAGE_PREFIX:-quay.io/cortexproject/}"
-
-COMMON_ARGS="-consul.hostname=consul:8500"
-STORAGE_ARGS="-config-yaml=/tests/schema1.yaml -dynamodb.url=dynamodb://u:p@dynamodb.cortex.:8000"
-INGESTER_ARGS="$COMMON_ARGS $STORAGE_ARGS -ingester.num-tokens=4 -ingester.min-ready-duration=1s --ingester.final-sleep=0s -ingester.concurrent-flushes=5"
-RUN_ARGS="--net=cortex -v $TEST_DIR:/tests"
-AVALANCHE_IMAGE=quay.io/freshtracks.io/avalanche:master-2019-07-28-47b00e3
-
-# Execute command $1 repeatedly until it returns true; description in $2, optional repeat limit in $3
-wait_for() {
-    reps=${3:-10}
-    for i in $(seq 1 $reps); do
-        echo "Waiting for $2"
-        if $1; then
-            return
-        fi
-        sleep 1
-    done
-    echo "Timed out waiting for $2" >&2
-    exit 1
-}
-
-container_ip() {
-    docker inspect -f "{{.NetworkSettings.Networks.cortex.IPAddress}}" $@
-}
diff --git a/integration-tests/start-components.sh b/integration-tests/start-components.sh
deleted file mode 100755
index 74816579e6..0000000000
--- a/integration-tests/start-components.sh
+++ /dev/null
@@ -1,12 +0,0 @@
-#! /bin/sh
-
-set -e
-
-THIS_DIR=$(dirname "$0")
-. "$THIS_DIR/common.sh"
-
-docker network create cortex
-
-docker run $RUN_ARGS -d --name=consul --hostname=consul consul:0.9 agent -ui -server -client=0.0.0.0 -bootstrap
-docker run $RUN_ARGS -d --name=dynamodb --hostname=dynamodb amazon/dynamodb-local:1.11.477 -jar DynamoDBLocal.jar -inMemory -sharedDb
-docker run $RUN_ARGS -d --name=distributor --hostname=distributor -p 8080:80 "$IMAGE_PREFIX"cortex:$IMAGE_TAG -target=distributor $COMMON_ARGS -distributor.replication-factor=1
diff --git a/integration-tests/stop-components.sh b/integration-tests/stop-components.sh
deleted file mode 100755
index 4a5bd92e3c..0000000000
--- a/integration-tests/stop-components.sh
+++ /dev/null
@@ -1,6 +0,0 @@
-#! /bin/sh
-
-set -e
-
-docker rm -f tm i2 i1 distributor dynamodb consul || true
-docker network rm cortex || true
diff --git a/integration-tests/test-flush.sh b/integration-tests/test-flush.sh
deleted file mode 100755
index 5632f46390..0000000000
--- a/integration-tests/test-flush.sh
+++ /dev/null
@@ -1,30 +0,0 @@
-#! /bin/sh
-
-set -e
-
-. "$(dirname "$0")/common.sh"
-
-"$(dirname "$0")/start-components.sh"
-
-# TODO: wait for DynamoDB ready
-sleep 5
-echo Start table-manager
-docker run $RUN_ARGS -d --name=tm --hostname=tm --restart=always "$IMAGE_PREFIX"cortex:$IMAGE_TAG -target=table-manager $STORAGE_ARGS -table-manager.retention-period=168h -dynamodb.poll-interval=5s
-
-echo Start ingester
-docker run $RUN_ARGS -d --name=i1 --hostname=i1 "$IMAGE_PREFIX"cortex:$IMAGE_TAG -target=ingester $INGESTER_ARGS
-
-sleep 5
-I1_ADDR=$(container_ip i1)
-wait_for "curl -s -f -m 3 $I1_ADDR/ready" "ingester ready"
-
-has_tokens_owned() {
-    curl -s -f -m 3 $1/metrics | grep 'cortex_ring_tokens_owned' >/dev/null
-}
-DIST_ADDR=$(container_ip distributor)
-wait_for "has_tokens_owned $DIST_ADDR" "distributor to see ingester in ring"
-
-docker run $RUN_ARGS --rm $AVALANCHE_IMAGE --metric-count=2 --label-count=2 --series-count=2 --remote-requests-count=1 --remote-url=http://distributor/api/prom/push
-
-echo Stop ingester so it should flush
-docker stop -t=60 i1 # allow 1 minute for flush to complete
diff --git a/integration-tests/test-handover.sh b/integration-tests/test-handover.sh
deleted file mode 100755
index 3d35b0044e..0000000000
--- a/integration-tests/test-handover.sh
+++ /dev/null
@@ -1,32 +0,0 @@
-#! /bin/sh
-
-set -e
-
-. "$(dirname "$0")/common.sh"
-
-"$(dirname "$0")/start-components.sh"
-
-# Note we haven't created any tables in the DB yet - we aren't going to run long enough to flush anything
-
-echo Start first ingester
-docker run $RUN_ARGS -d --name=i1 --hostname=i1 "$IMAGE_PREFIX"cortex:$IMAGE_TAG -target=ingester $INGESTER_ARGS -ingester.claim-on-rollout=true
-
-I1_ADDR=$(container_ip i1)
-wait_for "curl -s -f -m 3 $I1_ADDR/ready" "ingester ready"
-
-has_tokens_owned() {
-    curl -s -f -m 3 $1/metrics | grep -q 'cortex_ring_tokens_owned'
-}
-DIST_ADDR=$(container_ip distributor)
-wait_for "has_tokens_owned $DIST_ADDR" "distributor to see ingester in ring"
-
-docker run $RUN_ARGS --rm $AVALANCHE_IMAGE --metric-count=2 --label-count=2 --series-count=2 --remote-requests-count=1 --remote-url=http://distributor/api/prom/push
-
-echo Start second ingester, waiting for first
-docker run $RUN_ARGS -d --name=i2 --hostname=i2 "$IMAGE_PREFIX"cortex:$IMAGE_TAG -target=ingester $INGESTER_ARGS -ingester.join-after=300s -ingester.claim-on-rollout=true
-
-echo Stop first ingester so it should hand over
-docker stop i1
-
-I2_ADDR=$(container_ip i2)
-wait_for "curl -s -f -m 3 $I2_ADDR/ready" "ingester ready"
diff --git a/integration/backward_compatibility_test.go b/integration/backward_compatibility_test.go
new file mode 100644
index 0000000000..52e4594639
--- /dev/null
+++ b/integration/backward_compatibility_test.go
@@ -0,0 +1,84 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/integration/framework"
+)
+
+const (
+	// If you change the image tag, remember to update it in the preloading done
+	// by CircleCI too (see .circleci/config.yml).
+	previousVersionImage = "quay.io/cortexproject/cortex:v0.6.0"
+)
+
+func TestBackwardCompatibilityWithChunksStorage(t *testing.T) {
+	s, err := framework.NewScenario()
+	require.NoError(t, err)
+	defer s.Shutdown()
+
+	// Start dependencies
+	require.NoError(t, s.StartDynamoDB())
+	require.NoError(t, s.StartConsul())
+	require.NoError(t, s.WaitReady("consul", "dynamodb"))
+
+	// Start Cortex components (ingester running on previous version)
+	require.NoError(t, s.StartTableManager("table-manager", ChunksStorage, ""))
+	require.NoError(t, s.StartIngester("ingester-1", ChunksStorage, previousVersionImage))
+	require.NoError(t, s.StartDistributor("distributor", ChunksStorage, ""))
+	require.NoError(t, s.WaitReady("distributor", "ingester-1", "table-manager"))
+
+	// Wait until the first table-manager sync has completed, so that we're
+	// sure the tables have been created
+	require.NoError(t, s.Service("table-manager").WaitMetric(80, "cortex_dynamo_sync_tables_seconds", 1))
+
+	// Wait until the distributor has updated the ring
+	require.NoError(t, s.Service("distributor").WaitMetric(80, "cortex_ring_tokens_total", 512))
+
+	// Push some series to Cortex
+	now := time.Now()
+	series, expectedVector := generateSeries("series_1", now)
+
+	c, err := framework.NewClient(s.Endpoint("distributor", 80), "", "user-1")
+	require.NoError(t, err)
+
+	res, err := c.Push(series)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	// Start ingester-2 on new version, to ensure the transfer is backward compatible
+	require.NoError(t, s.StartIngester("ingester-2", mergeFlags(ChunksStorage, map[string]string{
+		"-ingester.join-after": "10s",
+	}), ""))
+
+	// Stop ingester-1. This function will return once the ingester-1 is successfully
+	// stopped, which means the transfer to ingester-2 is completed.
+	require.NoError(t, s.StopService("ingester-1"))
+
+	// Query the new ingester both with the old and the new querier
+	for _, image := range []string{previousVersionImage, ""} {
+		require.NoError(t, s.StartQuerier("querier", ChunksStorage, image))
+		require.NoError(t, s.WaitReady("querier"))
+
+		// Wait until the querier has updated the ring
+		require.NoError(t, s.Service("querier").WaitMetric(80, "cortex_ring_tokens_total", 512))
+
+		// Query the series
+		c, err := framework.NewClient(s.Endpoint("distributor", 80), s.Endpoint("querier", 80), "user-1")
+		require.NoError(t, err)
+
+		result, err := c.Query("series_1", now)
+		require.NoError(t, err)
+		require.Equal(t, model.ValVector, result.Type())
+		assert.Equal(t, expectedVector, result.(model.Vector))
+
+		// Stop the querier, so that the test on the next image will work
+		err = s.StopService("querier")
+		require.NoError(t, err)
+	}
+}
diff --git a/integration/ingester_flush_test.go b/integration/ingester_flush_test.go
new file mode 100644
index 0000000000..04b93cdeaf
--- /dev/null
+++ b/integration/ingester_flush_test.go
@@ -0,0 +1,91 @@
+package main
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/prompb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/integration/framework"
+)
+
+func TestIngesterFlushWithChunksStorage(t *testing.T) {
+	s, err := framework.NewScenario()
+	require.NoError(t, err)
+	defer s.Shutdown()
+
+	// Start dependencies
+	require.NoError(t, s.StartDynamoDB())
+	require.NoError(t, s.StartConsul())
+	require.NoError(t, s.WaitReady("consul", "dynamodb"))
+
+	// Start Cortex components
+	require.NoError(t, s.StartTableManager("table-manager", ChunksStorage, ""))
+	require.NoError(t, s.StartIngester("ingester-1", mergeFlags(ChunksStorage, map[string]string{
+		"-ingester.max-transfer-retries": "0",
+	}), ""))
+	require.NoError(t, s.StartQuerier("querier", ChunksStorage, ""))
+	require.NoError(t, s.StartDistributor("distributor", ChunksStorage, ""))
+	require.NoError(t, s.WaitReady("distributor", "querier", "ingester-1", "table-manager"))
+
+	// Wait until the first table-manager sync has completed, so that we're
+	// sure the tables have been created
+	require.NoError(t, s.Service("table-manager").WaitMetric(80, "cortex_dynamo_sync_tables_seconds", 1))
+
+	// Wait until both the distributor and querier have updated the ring
+	require.NoError(t, s.Service("distributor").WaitMetric(80, "cortex_ring_tokens_total", 512))
+	require.NoError(t, s.Service("querier").WaitMetric(80, "cortex_ring_tokens_total", 512))
+
+	c, err := framework.NewClient(s.Endpoint("distributor", 80), s.Endpoint("querier", 80), "user-1")
+	require.NoError(t, err)
+
+	// Push some series to Cortex
+	now := time.Now()
+	series1, expectedVector1 := generateSeries("series_1", now)
+	series2, expectedVector2 := generateSeries("series_2", now)
+
+	for _, series := range [][]prompb.TimeSeries{series1, series2} {
+		res, err := c.Push(series)
+		require.NoError(t, err)
+		require.Equal(t, 200, res.StatusCode)
+	}
+
+	// Query the series
+	result, err := c.Query("series_1", now)
+	require.NoError(t, err)
+	require.Equal(t, model.ValVector, result.Type())
+	assert.Equal(t, expectedVector1, result.(model.Vector))
+
+	result, err = c.Query("series_2", now)
+	require.NoError(t, err)
+	require.Equal(t, model.ValVector, result.Type())
+	assert.Equal(t, expectedVector2, result.(model.Vector))
+
+	// Stop ingester-1, so that it will flush all chunks to the storage. This function will return
+	// once the ingester-1 is successfully stopped, which means the flushing is completed.
+	require.NoError(t, s.StopService("ingester-1"))
+
+	// Ensure chunks have been uploaded to the storage (DynamoDB).
+	dynamoURL := "dynamodb://u:p@" + s.Service("dynamodb").Endpoint(8000)
+	dynamoClient, err := newDynamoClient(dynamoURL)
+	require.NoError(t, err)
+
+	// We have pushed 2 series, so we do expect 2 chunks
+	period := now.Unix() / (168 * 3600)
+	indexTable := fmt.Sprintf("cortex_%d", period)
+	chunksTable := fmt.Sprintf("cortex_chunks_%d", period)
+
+	out, err := dynamoClient.Scan(&dynamodb.ScanInput{TableName: aws.String(indexTable)})
+	require.NoError(t, err)
+	assert.Equal(t, int64(2*2), *out.Count)
+
+	out, err = dynamoClient.Scan(&dynamodb.ScanInput{TableName: aws.String(chunksTable)})
+	require.NoError(t, err)
+	assert.Equal(t, int64(2), *out.Count)
+}
diff --git a/integration/ingester_hand_over_test.go b/integration/ingester_hand_over_test.go
new file mode 100644
index 0000000000..9890ae3df8
--- /dev/null
+++ b/integration/ingester_hand_over_test.go
@@ -0,0 +1,91 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/common/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/integration/framework"
+)
+
+func TestIngesterHandOverWithBlocksStorage(t *testing.T) {
+	runIngesterHandOverTest(t, BlocksStorage, func(t *testing.T, s *framework.Scenario) {
+		// Start dependencies
+		require.NoError(t, s.StartMinio())
+		require.NoError(t, s.StartConsul())
+		require.NoError(t, s.WaitReady("consul", "minio"))
+
+		// Start Cortex components
+		require.NoError(t, s.StartIngester("ingester-1", BlocksStorage, ""))
+		require.NoError(t, s.StartQuerier("querier", BlocksStorage, ""))
+		require.NoError(t, s.StartDistributor("distributor", BlocksStorage, ""))
+		require.NoError(t, s.WaitReady("distributor", "querier", "ingester-1"))
+	})
+}
+
+func TestIngesterHandOverWithChunksStorage(t *testing.T) {
+	runIngesterHandOverTest(t, ChunksStorage, func(t *testing.T, s *framework.Scenario) {
+		// Start dependencies
+		require.NoError(t, s.StartDynamoDB())
+		require.NoError(t, s.StartConsul())
+		require.NoError(t, s.WaitReady("consul", "dynamodb"))
+
+		// Start Cortex components
+		require.NoError(t, s.StartTableManager("table-manager", ChunksStorage, ""))
+		require.NoError(t, s.StartIngester("ingester-1", ChunksStorage, ""))
+		require.NoError(t, s.StartQuerier("querier", ChunksStorage, ""))
+		require.NoError(t, s.StartDistributor("distributor", ChunksStorage, ""))
+		require.NoError(t, s.WaitReady("distributor", "querier", "ingester-1", "table-manager"))
+
+		// Wait until the first table-manager sync has completed, so that we're
+		// sure the tables have been created
+		require.NoError(t, s.Service("table-manager").WaitMetric(80, "cortex_dynamo_sync_tables_seconds", 1))
+	})
+}
+
+func runIngesterHandOverTest(t *testing.T, flags map[string]string, setup func(t *testing.T, s *framework.Scenario)) {
+	s, err := framework.NewScenario()
+	require.NoError(t, err)
+	defer s.Shutdown()
+
+	// Setup
+	setup(t, s)
+
+	// Wait until both the distributor and querier have updated the ring
+	require.NoError(t, s.Service("distributor").WaitMetric(80, "cortex_ring_tokens_total", 512))
+	require.NoError(t, s.Service("querier").WaitMetric(80, "cortex_ring_tokens_total", 512))
+
+	c, err := framework.NewClient(s.Endpoint("distributor", 80), s.Endpoint("querier", 80), "user-1")
+	require.NoError(t, err)
+
+	// Push some series to Cortex
+	now := time.Now()
+	series, expectedVector := generateSeries("series_1", now)
+
+	res, err := c.Push(series)
+	require.NoError(t, err)
+	require.Equal(t, 200, res.StatusCode)
+
+	// Query the series
+	result, err := c.Query("series_1", now)
+	require.NoError(t, err)
+	require.Equal(t, model.ValVector, result.Type())
+	assert.Equal(t, expectedVector, result.(model.Vector))
+
+	// Start ingester-2
+	require.NoError(t, s.StartIngester("ingester-2", mergeFlags(flags, map[string]string{
+		"-ingester.join-after": "10s",
+	}), ""))
+
+	// Stop ingester-1. This function will return once the ingester-1 is successfully
+	// stopped, which means the transfer to ingester-2 is completed.
+	require.NoError(t, s.StopService("ingester-1"))
+
+	// Query the series again
+	result, err = c.Query("series_1", now)
+	require.NoError(t, err)
+	assert.Equal(t, expectedVector, result.(model.Vector))
+}
diff --git a/pkg/compactor/compactor_ring_test.go b/pkg/compactor/compactor_ring_test.go
new file mode 100644
index 0000000000..fd3299854a
--- /dev/null
+++ b/pkg/compactor/compactor_ring_test.go
@@ -0,0 +1,61 @@
+package compactor
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
+)
+
+func TestRingConfig_DefaultConfigToLifecyclerConfig(t *testing.T) {
+	cfg := RingConfig{}
+	expected := ring.LifecyclerConfig{}
+	flagext.DefaultValues(&cfg, &expected)
+
+	// The default config of the compactor ring must be the exact same
+	// of the default lifecycler config, except few options which are
+	// intentionally overridden
+	expected.ListenPort = &cfg.ListenPort
+	expected.RingConfig.ReplicationFactor = 1
+	expected.NumTokens = 512
+	expected.MinReadyDuration = 0
+	expected.FinalSleep = 0
+
+	assert.Equal(t, expected, cfg.ToLifecyclerConfig())
+}
+
+func TestRingConfig_CustomConfigToLifecyclerConfig(t *testing.T) {
+	cfg := RingConfig{}
+	expected := ring.LifecyclerConfig{}
+	flagext.DefaultValues(&cfg, &expected)
+
+	// Customize the compactor ring config
+	cfg.HeartbeatPeriod = 1 * time.Second
+	cfg.HeartbeatTimeout = 10 * time.Second
+	cfg.InstanceID = "test"
+	cfg.InstanceInterfaceNames = []string{"abc1"}
+	cfg.InstancePort = 10
+	cfg.InstanceAddr = "1.2.3.4"
+	cfg.ListenPort = 10
+
+	// The lifecycler config should be generated based upon the compactor
+	// ring config
+	expected.HeartbeatPeriod = cfg.HeartbeatPeriod
+	expected.RingConfig.HeartbeatTimeout = cfg.HeartbeatTimeout
+	expected.ID = cfg.InstanceID
+	expected.InfNames = cfg.InstanceInterfaceNames
+	expected.Port = cfg.InstancePort
+	expected.Addr = cfg.InstanceAddr
+	expected.ListenPort = &cfg.ListenPort
+
+	// Hardcoded config
+	expected.RingConfig.ReplicationFactor = 1
+	expected.NumTokens = 512
+	expected.MinReadyDuration = 0
+	expected.FinalSleep = 0
+
+	assert.Equal(t, expected, cfg.ToLifecyclerConfig())
+}
diff --git a/pkg/compactor/compactor_test.go b/pkg/compactor/compactor_test.go
index 3d02670759..6692fd06b3 100644
--- a/pkg/compactor/compactor_test.go
+++ b/pkg/compactor/compactor_test.go
@@ -6,6 +6,9 @@ import (
 	"encoding/json"
 	"errors"
 	"flag"
+	"fmt"
+	"io/ioutil"
+	"os"
 	"strings"
 	"testing"
 	"time"
@@ -20,6 +23,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v2"
 
+	"github.com/cortexproject/cortex/pkg/ring"
+	"github.com/cortexproject/cortex/pkg/ring/kv/consul"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	cortex_testutil "github.com/cortexproject/cortex/pkg/util/test"
@@ -74,14 +79,16 @@ func TestCompactor_ShouldDoNothingOnNoUserBlocks(t *testing.T) {
 	bucketClient := &cortex_tsdb.BucketClientMock{}
 	bucketClient.MockIter("", []string{}, nil)
 
-	c, _, logs, registry := prepare(t, bucketClient)
+	c, _, logs, registry, cleanup := prepare(t, prepareConfig(), bucketClient)
+	defer cleanup()
+	c.Start()
 
 	// Wait until a run has completed.
 	cortex_testutil.Poll(t, time.Second, 1.0, func() interface{} {
 		return prom_testutil.ToFloat64(c.compactionRunsCompleted)
 	})
 
-	c.Shutdown()
+	c.Stop()
 
 	assert.Equal(t, []string{
 		`level=info msg="discovering users from bucket"`,
@@ -162,14 +169,16 @@ func TestCompactor_ShouldRetryOnFailureWhileDiscoveringUsersFromBucket(t *testin
 	bucketClient := &cortex_tsdb.BucketClientMock{}
 	bucketClient.MockIter("", nil, errors.New("failed to iterate the bucket"))
 
-	c, _, logs, registry := prepare(t, bucketClient)
+	c, _, logs, registry, cleanup := prepare(t, prepareConfig(), bucketClient)
+	defer cleanup()
+	c.Start()
 
 	// Wait until all retry attempts have completed.
 	cortex_testutil.Poll(t, time.Second, 1.0, func() interface{} {
 		return prom_testutil.ToFloat64(c.compactionRunsFailed)
 	})
 
-	c.Shutdown()
+	c.Stop()
 
 	// Ensure the bucket iteration has been retried the configured number of times.
 	bucketClient.AssertNumberOfCalls(t, "Iter", 3)
@@ -261,7 +270,9 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 	bucketClient.MockGet("user-1/01DTVP434PA9VFXSW2JKB3392D/meta.json", mockBlockMetaJSON("01DTVP434PA9VFXSW2JKB3392D"), nil)
 	bucketClient.MockGet("user-2/01DTW0ZCPDDNV4BV83Q2SV4QAZ/meta.json", mockBlockMetaJSON("01DTW0ZCPDDNV4BV83Q2SV4QAZ"), nil)
 
-	c, tsdbCompactor, logs, registry := prepare(t, bucketClient)
+	c, tsdbCompactor, logs, registry, cleanup := prepare(t, prepareConfig(), bucketClient)
+	defer cleanup()
+	c.Start()
 
 	// Mock the compactor as if there's no compaction to do,
 	// in order to simplify tests (all in all, we just want to
@@ -274,7 +285,7 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 		return prom_testutil.ToFloat64(c.compactionRunsCompleted)
 	})
 
-	c.Shutdown()
+	c.Stop()
 
 	// Ensure a plan has been executed for the blocks of each user.
 	tsdbCompactor.AssertNumberOfCalls(t, "Plan", 2)
@@ -314,6 +325,176 @@ func TestCompactor_ShouldIterateOverUsersAndRunCompaction(t *testing.T) {
 	`), testedMetrics...))
 }
 
+func TestCompactor_ShouldCompactAllUsersOnShardingEnabledButOnlyOneInstanceRunning(t *testing.T) {
+	t.Parallel()
+
+	// Mock the bucket to contain two users, each one with one block.
+	bucketClient := &cortex_tsdb.BucketClientMock{}
+	bucketClient.MockIter("", []string{"user-1", "user-2"}, nil)
+	bucketClient.MockIter("user-1/", []string{"user-1/01DTVP434PA9VFXSW2JKB3392D"}, nil)
+	bucketClient.MockIter("user-2/", []string{"user-2/01DTW0ZCPDDNV4BV83Q2SV4QAZ"}, nil)
+	bucketClient.MockGet("user-1/01DTVP434PA9VFXSW2JKB3392D/meta.json", mockBlockMetaJSON("01DTVP434PA9VFXSW2JKB3392D"), nil)
+	bucketClient.MockGet("user-2/01DTW0ZCPDDNV4BV83Q2SV4QAZ/meta.json", mockBlockMetaJSON("01DTW0ZCPDDNV4BV83Q2SV4QAZ"), nil)
+
+	cfg := prepareConfig()
+	cfg.ShardingEnabled = true
+	cfg.ShardingRing.InstanceID = "compactor-1"
+	cfg.ShardingRing.KVStore.Mock = consul.NewInMemoryClient(ring.GetCodec())
+
+	c, tsdbCompactor, logs, _, cleanup := prepare(t, cfg, bucketClient)
+	defer cleanup()
+	c.Start()
+
+	// Mock the compactor as if there's no compaction to do,
+	// in order to simplify tests (all in all, we just want to
+	// test our logic and not TSDB compactor which we expect to
+	// be already tested).
+	tsdbCompactor.On("Plan", mock.Anything).Return([]string{}, nil)
+
+	// Wait until a run has completed.
+	cortex_testutil.Poll(t, 5*time.Second, 1.0, func() interface{} {
+		return prom_testutil.ToFloat64(c.compactionRunsCompleted)
+	})
+
+	c.Stop()
+
+	// Ensure a plan has been executed for the blocks of each user.
+	tsdbCompactor.AssertNumberOfCalls(t, "Plan", 2)
+
+	assert.Equal(t, []string{
+		`level=info msg="waiting until compactor is ACTIVE in the ring"`,
+		`level=info msg="compactor is ACTIVE in the ring"`,
+		`level=info msg="discovering users from bucket"`,
+		`level=info msg="discovered users from bucket" users=2`,
+		`level=info msg="starting compaction of user blocks" user=user-1`,
+		`level=info msg="start sync of metas"`,
+		`level=info msg="start of GC"`,
+		`level=info msg="start of compaction"`,
+		`level=info msg="compaction iterations done"`,
+		`level=info msg="successfully compacted user blocks" user=user-1`,
+		`level=info msg="starting compaction of user blocks" user=user-2`,
+		`level=info msg="start sync of metas"`,
+		`level=info msg="start of GC"`,
+		`level=info msg="start of compaction"`,
+		`level=info msg="compaction iterations done"`,
+		`level=info msg="successfully compacted user blocks" user=user-2`,
+	}, removeMetaFetcherLogs(strings.Split(strings.TrimSpace(logs.String()), "\n")))
+}
+
+func TestCompactor_ShouldCompactOnlyUsersOwnedByTheInstanceOnShardingEnabledAndMultipleInstancesRunning(t *testing.T) {
+	t.Parallel()
+
+	numUsers := 100
+
+	// Setup user IDs
+	userIDs := make([]string, 0, numUsers)
+	for i := 1; i <= numUsers; i++ {
+		userIDs = append(userIDs, fmt.Sprintf("user-%d", i))
+	}
+
+	// Mock the bucket to contain all users, each one with one block.
+	bucketClient := &cortex_tsdb.BucketClientMock{}
+	bucketClient.MockIter("", userIDs, nil)
+	for _, userID := range userIDs {
+		bucketClient.MockIter(userID+"/", []string{"user-1/01DTVP434PA9VFXSW2JKB3392D"}, nil)
+		bucketClient.MockGet(userID+"/01DTVP434PA9VFXSW2JKB3392D/meta.json", mockBlockMetaJSON("01DTVP434PA9VFXSW2JKB3392D"), nil)
+	}
+
+	// Create a shared KV Store
+	kvstore := consul.NewInMemoryClient(ring.GetCodec())
+
+	// Create two compactors
+	compactors := []*Compactor{}
+	logs := []*bytes.Buffer{}
+
+	for i := 1; i <= 2; i++ {
+		cfg := prepareConfig()
+		cfg.ShardingEnabled = true
+		cfg.ShardingRing.InstanceID = fmt.Sprintf("compactor-%d", i)
+		cfg.ShardingRing.InstanceAddr = fmt.Sprintf("127.0.0.%d", i)
+		cfg.ShardingRing.KVStore.Mock = kvstore
+
+		c, tsdbCompactor, l, _, cleanup := prepare(t, cfg, bucketClient)
+		defer c.Stop()
+		defer cleanup()
+
+		compactors = append(compactors, c)
+		logs = append(logs, l)
+
+		// Mock the compactor as if there's no compaction to do,
+		// in order to simplify tests (all in all, we just want to
+		// test our logic and not TSDB compactor which we expect to
+		// be already tested).
+		tsdbCompactor.On("Plan", mock.Anything).Return([]string{}, nil)
+	}
+
+	// Wait until each compactor sees all ACTIVE compactors in the ring
+	for _, c := range compactors {
+		cortex_testutil.Poll(t, 10*time.Second, len(compactors), func() interface{} {
+			rs, err := c.ring.GetAll()
+			if err != nil {
+				return 0
+			}
+
+			numActive := 0
+			for _, i := range rs.Ingesters {
+				if i.GetState() == ring.ACTIVE {
+					numActive++
+				}
+			}
+
+			return numActive
+		})
+	}
+
+	// Start all compactors
+	for _, c := range compactors {
+		c.Start()
+	}
+
+	// Wait until a run has been completed on each compactor
+	for _, c := range compactors {
+		cortex_testutil.Poll(t, 10*time.Second, 1.0, func() interface{} {
+			return prom_testutil.ToFloat64(c.compactionRunsCompleted)
+		})
+	}
+
+	// Ensure that each user has been compacted by the correct instance
+	for _, userID := range userIDs {
+		_, l, err := findCompactorByUserID(compactors, logs, userID)
+		require.NoError(t, err)
+		assert.Contains(t, l.String(), fmt.Sprintf(`level=info msg="successfully compacted user blocks" user=%s`, userID))
+	}
+}
+
+func findCompactorByUserID(compactors []*Compactor, logs []*bytes.Buffer, userID string) (*Compactor, *bytes.Buffer, error) {
+	var compactor *Compactor
+	var log *bytes.Buffer
+
+	for i, c := range compactors {
+		owned, err := c.ownUser(userID)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		// Ensure the user is not owned by multiple compactors
+		if owned && compactor != nil {
+			return nil, nil, fmt.Errorf("user %s owned by multiple compactors", userID)
+		}
+		if owned {
+			compactor = c
+			log = logs[i]
+		}
+	}
+
+	// Return an error if we've not been able to find it
+	if compactor == nil {
+		return nil, nil, fmt.Errorf("user %s not owned by any compactor", userID)
+	}
+
+	return compactor, log, nil
+}
+
 func removeMetaFetcherLogs(input []string) []string {
 	out := make([]string, 0, len(input))
 
@@ -326,13 +507,29 @@ func removeMetaFetcherLogs(input []string) []string {
 	return out
 }
 
-func prepare(t *testing.T, bucketClient *cortex_tsdb.BucketClientMock) (*Compactor, *tsdbCompactorMock, *bytes.Buffer, prometheus.Gatherer) {
+func prepareConfig() Config {
 	compactorCfg := Config{}
+	flagext.DefaultValues(&compactorCfg)
+
+	return compactorCfg
+}
+
+func prepare(t *testing.T, compactorCfg Config, bucketClient *cortex_tsdb.BucketClientMock) (*Compactor, *tsdbCompactorMock, *bytes.Buffer, prometheus.Gatherer, func()) {
 	storageCfg := cortex_tsdb.Config{}
-	flagext.DefaultValues(&compactorCfg, &storageCfg)
+	flagext.DefaultValues(&storageCfg)
+
 	compactorCfg.retryMinBackoff = 0
 	compactorCfg.retryMaxBackoff = 0
 
+	// Create a temporary directory for compactor data.
+	dataDir, err := ioutil.TempDir(os.TempDir(), "compactor-test")
+	require.NoError(t, err)
+
+	compactorCfg.DataDir = dataDir
+	cleanup := func() {
+		os.RemoveAll(dataDir)
+	}
+
 	tsdbCompactor := &tsdbCompactorMock{}
 	logs := &bytes.Buffer{}
 	logger := log.NewLogfmtLogger(logs)
@@ -342,7 +539,7 @@ func prepare(t *testing.T, bucketClient *cortex_tsdb.BucketClientMock) (*Compact
 	c, err := newCompactor(ctx, cancelCtx, compactorCfg, storageCfg, bucketClient, tsdbCompactor, logger, registry)
 	require.NoError(t, err)
 
-	return c, tsdbCompactor, logs, registry
+	return c, tsdbCompactor, logs, registry, cleanup
 }
 
 type tsdbCompactorMock struct {
diff --git a/pkg/util/runtimeconfig/manager_test.go b/pkg/util/runtimeconfig/manager_test.go
index c4f590d6d5..7a008af7f4 100644
--- a/pkg/util/runtimeconfig/manager_test.go
+++ b/pkg/util/runtimeconfig/manager_test.go
@@ -74,7 +74,7 @@ func TestNewOverridesManager(t *testing.T) {
 		Loader:       testLoadOverrides,
 	}
 
-	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig)
+	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
 
 	// Cleaning up
@@ -108,7 +108,7 @@ func TestOverridesManager_ListenerWithDefaultLimits(t *testing.T) {
 		Loader:       testLoadOverrides,
 	}
 
-	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig)
+	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
 
 	// need to use buffer, otherwise loadConfig will throw away update
@@ -156,7 +156,7 @@ func TestOverridesManager_ListenerChannel(t *testing.T) {
 		},
 	}
 
-	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig)
+	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
 
 	// need to use buffer, otherwise loadConfig will throw away update
@@ -205,7 +205,7 @@ func TestOverridesManager_StopClosesListenerChannels(t *testing.T) {
 		},
 	}
 
-	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig)
+	overridesManager, err := NewRuntimeConfigManager(overridesManagerConfig, nil)
 	require.NoError(t, err)
 
 	// need to use buffer, otherwise loadConfig will throw away update
diff --git a/tools/test b/tools/test
index cac70713fc..7a6a3e4140 100755
--- a/tools/test
+++ b/tools/test
@@ -83,7 +83,7 @@ fail=0
 
 if [ -z "$TESTDIRS" ]; then
     # NB: Relies on paths being prefixed with './'.
-    TESTDIRS=($(git ls-files -- '*_test.go' | grep -vE '^(vendor|experimental)/' | xargs -n1 dirname | sort -u | sed -e 's|^|./|'))
+    TESTDIRS=($(git ls-files -- '*_test.go' | grep -vE '^(vendor|experimental|integration)/' | xargs -n1 dirname | sort -u | sed -e 's|^|./|'))
 else
     # TESTDIRS on the right side is not really an array variable, it
     # is just a string with spaces, but it is written like that to