Added KV Store client that mirrors one backend to another

CHANGELOG.md

@@ -12,11 +12,13 @@
 * [CHANGE] Use relative links from /ring page to make it work when used behind reverse proxy. #1896
 * [CHANGE] Deprecated `-distributor.limiter-reload-period` flag. #1766
 * [CHANGE] Ingesters now write only normalised tokens to the ring, although they can still read denormalised tokens used by other ingesters. `-ingester.normalise-tokens` is now deprecated, and ignored. If you want to switch back to using denormalised tokens, you need to downgrade to Cortex 0.4.0. Previous versions don't handle claiming tokens from normalised ingesters correctly. #1809
+* [CHANGE] Overrides mechanism has been renamed to "runtime config", and is now separate from limits. Runtime config is simply a file that is reloaded by Cortex every couple of seconds. Limits and now also multi KV use this mechanism.<br />New arguments were introduced: `-runtime-config.file` (defaults to empty) and `-runtime-config.reload-period` (defaults to 10 seconds), which replace previously used `-limits.per-user-override-config` and `-limits.per-user-override-period` options. Old options are still used if `-runtime-config.file` is not specified. This change is also reflected in YAML configuration, where old `limits.per_tenant_override_config` and `limits.per_tenant_override_period` fields are replaced with `runtime_config.file` and `runtime_config.period` respectively. #1749
 * [FEATURE] The distributor can now drop labels from samples (similar to the removal of the replica label for HA ingestion) per user via the `distributor.drop-label` flag. #1726
 * [FEATURE] Added `global` ingestion rate limiter strategy. Deprecated `-distributor.limiter-reload-period` flag. #1766
 * [FEATURE] Added support for Microsoft Azure blob storage to be used for storing chunk data. #1913
 * [FEATURE] Added readiness probe endpoint`/ready` to queriers. #1934
 * [FEATURE] EXPERIMENTAL: Added `/series` API endpoint support with TSDB blocks storage. #1830
+* [FEATURE] Added "multi" KV store that can interact with two other KV stores, primary one for all reads and writes, and secondary one, which only receives writes. Primary/secondary store can be modified in runtime via runtime-config mechanism (previously "overrides"). #1749
 * [ENHANCEMENT] Added `password` and `enable_tls` options to redis cache configuration. Enables usage of Microsoft Azure Cache for Redis service.
 * [BUGFIX] Fixed unnecessary CAS operations done by the HA tracker when the jitter is enabled. #1861
 * [BUGFIX] Fixed #1904 ingesters getting stuck in a LEAVING state after coming up from an ungraceful exit. #1921

docs/configuration/arguments.md

@@ -122,7 +122,7 @@ The KVStore client is used by both the Ring and HA Tracker.
 - `{ring,distributor.ha-tracker}.prefix`
    The prefix for the keys in the store. Should end with a /. For example with a prefix of foo/, the key bar would be stored under foo/bar.
 - `{ring,distributor.ha-tracker}.store`
-   Backend storage to use for the ring (consul, etcd, inmemory).
+   Backend storage to use for the ring (consul, etcd, inmemory, memberlist, multi).
 
 #### Consul
 
@@ -182,6 +182,35 @@ Flags for configuring KV store based on memberlist library. This feature is expe
    Timeout for writing 'packet' data.
 - `memberlist.transport-debug`
    Log debug transport messages. Note: global log.level must be at debug level as well.
+
+#### Multi KV
+
+This is a special key-value implementation that uses two different KV stores (eg. consul, etcd or memberlist). One of them is always marked as primary, and all reads and writes go to primary store. Other one, secondary, is only used for writes. The idea is that operator can use multi KV store to migrate from primary to secondary store in runtime.
+
+For example, migration from Consul to Etcd would look like this:
+
+- Set `ring.store` to use `multi` store. Set `-multi.primary=consul` and `-multi.secondary=etcd`. All consul and etcd settings must still be specified.
+- Start all Cortex microservices. They will still use Consul as primary KV, but they will also write share ring via etcd.
+- Operator can now use "runtime config" mechanism to switch primary store to etcd.
+- After all Cortex microservices have picked up new primary store, and everything looks correct, operator can now shut down Consul, and modify Cortex configuration to use `-ring.store=etcd` only.
+- At this point, Consul can be shut down.
+
+Multi KV has following parameters:
+
+- `multi.primary` - name of primary KV store. Same values as in `ring.store` are supported, except `multi`.
+- `multi.secondary` - name of secondary KV store.
+- `multi.mirror-enabled` - enable mirroring of values to secondary store, defaults to true
+- `multi.mirror-timeout` - wait max this time to write to secondary store to finish. Default to 2 seconds. Errors writing to secondary store are not reported to caller, but are logged and also reported via `cortex_multikv_mirror_write_errors_total` metric.
+
+Multi KV also reacts on changes done via runtime configuration. It uses this section:
+
+```yaml
+multi_kv_config:
+    mirror-enabled: false
+    primary: memberlist
+```
+
+Note that runtime configuration values take precedence over command line options.
 
 ### HA Tracker
 
@@ -276,11 +305,13 @@ It also talks to a KVStore and has it's own copies of the same flags used by the
    Where you don't want to cache every chunk written by ingesters, but you do want to take advantage of chunk write deduplication, this option will make ingesters write a placeholder to the cache for each chunk.
    Make sure you configure ingesters with a different cache to queriers, which need the whole value.
 
-## Ingester, Distributor & Querier limits.
+## Runtime Configuration file
 
-Cortex implements various limits on the requests it can process, in order to prevent a single tenant overwhelming the cluster.  There are various default global limits which apply to all tenants which can be set on the command line.  These limits can also be overridden on a per-tenant basis, using a configuration file.  Specify the filename for the override configuration file using the `-limits.per-user-override-config=<filename>` flag.  The override file will be re-read every 10 seconds by default - this can also be controlled using the `-limits.per-user-override-period=10s` flag.
+Cortex has a concept of "runtime config" file, which is simply a file that is reloaded while Cortex is running. It is used by some Cortex components to allow operator to change some aspects of Cortex configuration without restarting it. File is specified by using `-runtime-config.file=<filename>` flag and reload period (which defaults to 10 seconds) can be changed by `-runtime-config.reload-period=<duration>` flag. Previously this mechanism was only used by limits overrides, and flags were called `-limits.per-user-override-config=<filename>` and `-limits.per-user-override-period=10s` respectively. These are still used, if `-runtime-config.file=<filename>` is not specified.
 
-The override file should be in YAML format and contain a single `overrides` field, which itself is a map of tenant ID (same values as passed in the `X-Scope-OrgID` header) to the various limits.  An example `overrides.yml` could look like:
+At the moment, two components use runtime configuration: limits and multi KV store.
+
+Example runtime configuration file:
 
 ```yaml
 overrides:
@@ -292,11 +323,33 @@ overrides:
     max_samples_per_query: 1000000
     max_series_per_metric: 100000
     max_series_per_query: 100000
+
+multi_kv_config:
+    mirror-enabled: false
+    primary: memberlist
 ```
 
 When running Cortex on Kubernetes, store this file in a config map and mount it in each services' containers.  When changing the values there is no need to restart the services, unless otherwise specified.
 
-Valid fields are (with their corresponding flags for default values):
+## Ingester, Distributor & Querier limits.
+
+Cortex implements various limits on the requests it can process, in order to prevent a single tenant overwhelming the cluster.  There are various default global limits which apply to all tenants which can be set on the command line.  These limits can also be overridden on a per-tenant basis by using `overrides` field of runtime configuration file.
+
+The `overrides` field is a map of tenant ID (same values as passed in the `X-Scope-OrgID` header) to the various limits.  An example could look like:
+
+```yaml
+overrides:
+  tenant1:
+    ingestion_rate: 10000
+    max_series_per_metric: 100000
+    max_series_per_query: 100000
+  tenant2:
+    max_samples_per_query: 1000000
+    max_series_per_metric: 100000
+    max_series_per_query: 100000
+```
+
+Valid per-tenant limits are (with their corresponding flags for default values):
 
 - `ingestion_rate_strategy` / `-distributor.ingestion-rate-limit-strategy`
 - `ingestion_rate` / `-distributor.ingestion-rate-limit`

go.mod

@@ -60,6 +60,7 @@ require (
 	github.com/thanos-io/thanos v0.8.1-0.20200102143048-a37ac093a67a
 	github.com/tinylib/msgp v0.0.0-20161221055906-38a6f61a768d // indirect
 	github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 // indirect
+	github.com/uber-go/atomic v1.4.0
 	github.com/uber/jaeger-client-go v2.20.1+incompatible
 	github.com/weaveworks/billing-client v0.0.0-20171006123215-be0d55e547b1
 	github.com/weaveworks/common v0.0.0-20190822150010-afb9996716e4

go.sum

@@ -750,6 +750,8 @@ github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5 h1:LnC5Kc/wtumK+WB441p7ynQJzVuNRJiqddSIE3IlSEQ=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
+github.com/uber-go/atomic v1.4.0 h1:yOuPqEq4ovnhEjpHmfFwsqBXDYbQeT6Nb0bwD6XnD5o=
+github.com/uber-go/atomic v1.4.0/go.mod h1:/Ct5t2lcmbJ4OSe/waGBoaVvVqtO0bmtfVNex1PFV8g=
 github.com/uber/jaeger-client-go v2.20.1+incompatible h1:HgqpYBng0n7tLJIlyT4kPCIv5XgCsF+kai1NnnrJzEU=
 github.com/uber/jaeger-client-go v2.20.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=

pkg/chunk/chunk_store_test.go

@@ -88,7 +88,7 @@ func newTestChunkStoreConfig(t require.TestingT, schemaName string, storeCfg Sto
 	var limits validation.Limits
 	flagext.DefaultValues(&limits)
 	limits.MaxQueryLength = 30 * 24 * time.Hour
-	overrides, err := validation.NewOverrides(limits)
+	overrides, err := validation.NewOverrides(limits, nil)
 	require.NoError(t, err)
 
 	store := NewCompositeStore()

pkg/chunk/storage/caching_fixtures.go

@@ -41,5 +41,5 @@ func defaultLimits() (*validation.Overrides, error) {
 	var defaults validation.Limits
 	flagext.DefaultValues(&defaults)
 	defaults.CardinalityLimit = 5
-	return validation.NewOverrides(defaults)
+	return validation.NewOverrides(defaults, nil)
 }

pkg/chunk/storage/factory_test.go

@@ -30,7 +30,7 @@ func TestFactoryStop(t *testing.T) {
 		},
 	}
 
-	limits, err := validation.NewOverrides(defaults)
+	limits, err := validation.NewOverrides(defaults, nil)
 	require.NoError(t, err)
 
 	store, err := NewStore(cfg, storeConfig, schemaConfig, limits)

pkg/cortex/cortex.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ruler"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/runtimeconfig"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -75,10 +76,11 @@ type Config struct {
 	Encoding       encoding.Config          `yaml:"-"` // No yaml for this, it only works with flags.
 	TSDB           tsdb.Config              `yaml:"tsdb"`
 
-	Ruler        ruler.Config                               `yaml:"ruler,omitempty"`
-	ConfigDB     db.Config                                  `yaml:"configdb,omitempty"`
-	ConfigStore  config_client.Config                       `yaml:"config_store,omitempty"`
-	Alertmanager alertmanager.MultitenantAlertmanagerConfig `yaml:"alertmanager,omitempty"`
+	Ruler         ruler.Config                               `yaml:"ruler,omitempty"`
+	ConfigDB      db.Config                                  `yaml:"configdb,omitempty"`
+	ConfigStore   config_client.Config                       `yaml:"config_store,omitempty"`
+	Alertmanager  alertmanager.MultitenantAlertmanagerConfig `yaml:"alertmanager,omitempty"`
+	RuntimeConfig runtimeconfig.ManagerConfig                `yaml:"runtime_config,omitempty"`
 }
 
 // RegisterFlags registers flag.
@@ -112,6 +114,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
 	c.ConfigDB.RegisterFlags(f)
 	c.ConfigStore.RegisterFlagsWithPrefix("alertmanager.", f)
 	c.Alertmanager.RegisterFlags(f)
+	c.RuntimeConfig.RegisterFlags(f)
 
 	// These don't seem to have a home.
 	flag.IntVar(&chunk_util.QueryParallelism, "querier.query-parallelism", 100, "Max subqueries run in parallel per higher-level query.")
@@ -146,16 +149,17 @@ type Cortex struct {
 	target             moduleName
 	httpAuthMiddleware middleware.Interface
 
-	server       *server.Server
-	ring         *ring.Ring
-	overrides    *validation.Overrides
-	distributor  *distributor.Distributor
-	ingester     *ingester.Ingester
-	store        chunk.Store
-	worker       frontend.Worker
-	frontend     *frontend.Frontend
-	tableManager *chunk.TableManager
-	cache        cache.Cache
+	server        *server.Server
+	ring          *ring.Ring
+	overrides     *validation.Overrides
+	distributor   *distributor.Distributor
+	ingester      *ingester.Ingester
+	store         chunk.Store
+	worker        frontend.Worker
+	frontend      *frontend.Frontend
+	tableManager  *chunk.TableManager
+	cache         cache.Cache
+	runtimeConfig *runtimeconfig.Manager
 
 	ruler        *ruler.Ruler
 	configAPI    *api.API

pkg/cortex/modules.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ruler"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/runtimeconfig"
 	"github.com/cortexproject/cortex/pkg/util/validation"
 )
 
@@ -40,6 +41,7 @@ type moduleName int
 // The various modules that make up Cortex.
 const (
 	Ring moduleName = iota
+	RuntimeConfig
 	Overrides
 	Server
 	Distributor
@@ -58,6 +60,8 @@ func (m moduleName) String() string {
 	switch m {
 	case Ring:
 		return "ring"
+	case RuntimeConfig:
+		return "runtime-config"
 	case Overrides:
 		return "overrides"
 	case Server:
@@ -152,6 +156,7 @@ func (t *Cortex) stopServer() (err error) {
 }
 
 func (t *Cortex) initRing(cfg *Config) (err error) {
+	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	t.ring, err = ring.New(cfg.Ingester.LifecyclerConfig.RingConfig, "ingester", ring.IngesterRingKey)
 	if err != nil {
 		return
@@ -161,16 +166,30 @@ func (t *Cortex) initRing(cfg *Config) (err error) {
 	return
 }
 
-func (t *Cortex) initOverrides(cfg *Config) (err error) {
-	t.overrides, err = validation.NewOverrides(cfg.LimitsConfig)
+func (t *Cortex) initRuntimeConfig(cfg *Config) (err error) {
+	if cfg.RuntimeConfig.LoadPath == "" {
+		cfg.RuntimeConfig.LoadPath = cfg.LimitsConfig.PerTenantOverrideConfig
+		cfg.RuntimeConfig.ReloadPeriod = cfg.LimitsConfig.PerTenantOverridePeriod
+	}
+	cfg.RuntimeConfig.Loader = loadRuntimeConfig
+
+	// make sure to set default limits before we start loading configuration into memory
+	validation.SetDefaultLimitsForYAMLUnmarshalling(cfg.LimitsConfig)
+
+	t.runtimeConfig, err = runtimeconfig.NewRuntimeConfigManager(cfg.RuntimeConfig)
 	return err
 }
 
-func (t *Cortex) stopOverrides() error {
-	t.overrides.Stop()
+func (t *Cortex) stopRuntimeConfig() (err error) {
+	t.runtimeConfig.Stop()
 	return nil
 }
 
+func (t *Cortex) initOverrides(cfg *Config) (err error) {
+	t.overrides, err = validation.NewOverrides(cfg.LimitsConfig, tenantLimitsFromRuntimeConfig(t.runtimeConfig))
+	return err
+}
+
 func (t *Cortex) initDistributor(cfg *Config) (err error) {
 	cfg.Distributor.DistributorRing.ListenPort = cfg.Server.GRPCListenPort
 
@@ -257,6 +276,7 @@ func (t *Cortex) stopQuerier() error {
 }
 
 func (t *Cortex) initIngester(cfg *Config) (err error) {
+	cfg.Ingester.LifecyclerConfig.RingConfig.KVStore.Multi.ConfigProvider = multiClientRuntimeConfigChannel(t.runtimeConfig)
 	cfg.Ingester.LifecyclerConfig.ListenPort = &cfg.Server.GRPCListenPort
 	cfg.Ingester.TSDBEnabled = cfg.Storage.Engine == storage.StorageEngineTSDB
 	cfg.Ingester.TSDBConfig = cfg.TSDB
@@ -446,14 +466,19 @@ var modules = map[moduleName]module{
 		stop: (*Cortex).stopServer,
 	},
 
+	RuntimeConfig: {
+		init: (*Cortex).initRuntimeConfig,
+		stop: (*Cortex).stopRuntimeConfig,
+	},
+
 	Ring: {
-		deps: []moduleName{Server},
+		deps: []moduleName{Server, RuntimeConfig},
 		init: (*Cortex).initRing,
 	},
 
 	Overrides: {
+		deps: []moduleName{RuntimeConfig},
 		init: (*Cortex).initOverrides,
-		stop: (*Cortex).stopOverrides,
 	},
 
 	Distributor: {
@@ -469,7 +494,7 @@ var modules = map[moduleName]module{
 	},
 
 	Ingester: {
-		deps: []moduleName{Overrides, Store, Server},
+		deps: []moduleName{Overrides, Store, Server, RuntimeConfig},
 		init: (*Cortex).initIngester,
 		stop: (*Cortex).stopIngester,
 	},

pkg/cortex/runtime_config.go

@@ -0,0 +1,72 @@
+package cortex
+
+import (
+	"os"
+
+	"gopkg.in/yaml.v2"
+
+	"github.com/cortexproject/cortex/pkg/ring/kv"
+	"github.com/cortexproject/cortex/pkg/util/runtimeconfig"
+	"github.com/cortexproject/cortex/pkg/util/validation"
+)
+
+// runtimeConfigValues are values that can be reloaded from configuration file while Cortex is running.
+// Reloading is done by runtime_config.Manager, which also keeps the currently loaded config.
+// These values are then pushed to the components that are interested in them.
+type runtimeConfigValues struct {
+	TenantLimits map[string]*validation.Limits `yaml:"overrides"`
+
+	Multi kv.MultiRuntimeConfig `yaml:"multi_kv_config"`
+}
+
+func loadRuntimeConfig(filename string) (interface{}, error) {
+	f, err := os.Open(filename)
+	if err != nil {
+		return nil, err
+	}
+
+	var overrides = &runtimeConfigValues{}
+
+	decoder := yaml.NewDecoder(f)
+	decoder.SetStrict(true)
+	if err := decoder.Decode(&overrides); err != nil {
+		return nil, err
+	}
+
+	return overrides, nil
+}
+
+func tenantLimitsFromRuntimeConfig(c *runtimeconfig.Manager) validation.TenantLimits {
+	return func(userID string) *validation.Limits {
+		cfg, ok := c.GetConfig().(*runtimeConfigValues)
+		if !ok || cfg == nil {
+			return nil
+		}
+
+		return cfg.TenantLimits[userID]
+	}
+}
+
+func multiClientRuntimeConfigChannel(manager *runtimeconfig.Manager) func() <-chan kv.MultiRuntimeConfig {
+	// returns function that can be used in MultiConfig.ConfigProvider
+	return func() <-chan kv.MultiRuntimeConfig {
+		outCh := make(chan kv.MultiRuntimeConfig, 1)
+
+		// push initial config to the channel
+		val := manager.GetConfig()
+		if cfg, ok := val.(*runtimeConfigValues); ok && cfg != nil {
+			outCh <- cfg.Multi
+		}
+
+		ch := manager.CreateListenerChannel(1)
+		go func() {
+			for val := range ch {
+				if cfg, ok := val.(*runtimeConfigValues); ok && cfg != nil {
+					outCh <- cfg.Multi
+				}
+			}
+		}()
+
+		return outCh
+	}
+}