Repository: cortexproject/cortex
Pull request #2113 (2020-02-11 14:53:03)

Add support to horizontally scale blocks compactor
**What this PR does**:
This PR introduces the support to horizontally scale the compactor, when using the experimental blocks storage.

I've also took the change to remove `ingester` references from the ring HTTP page, given it's now used by multiple components (ingesters, ruler, compactor). The refactoring done here should be safe, given we've recently removed support for denormalised tokens.

**Which issue(s) this PR fixes**:
N/A

**Checklist**
- [x] Tests updated
- [x] Documentation added
- [x] `CHANGELOG.md` updated - the order of entries should be `[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`


pracucci: > I guess my only question is during a topology change is there not a way that two compactors can run on the same user? Ex. first compactor comes up, adds itself to the ring and starts compacting user 0. Next compactor starts up adds itself to the ring, and it now owns user 0 and will start compaction on that user, resulting int two compactors working on the same user.

@thorfour Definitely. This can happen "by design". After looking at the bucket compactor logic, doing some manual experiments and talking to Bartek (back in December) this looked to be not a real issue. If the same blocks for the same tenant are simultaneously compacted by two instances, the next compactor run will get rid of the extra blocks created due to simultaneous runs (at this point only 1 of the two compactors will run the next compaction).
pracucci: > Overall PR looks good to me, but I'm concerned about resharding. What if one compactor is working on a user, and at the same time, new compactor is added – and suddenly it's new compactor who owns user 1. Does first compactor detect and cancels the compaction? Or does it not upload compacted blocks in such case?

@pstibrany It doesn't detect or cancel the on-going compaction. There's no strong guarantee two replicas will not compact the same blocks. My take is this one:
https://github.com/cortexproject/cortex/pull/2113#issuecomment-584737468
pracucci: @pstibrany I should have addressed your comments. Do you have any other feedback?

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 002e25974e..efce19ece0 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
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/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..07dab65361 100644
--- a/pkg/cortex/modules.go
+++ b/pkg/cortex/modules.go
@@ -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.


Test patch:
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 {