diff --git a/CHANGELOG.md b/CHANGELOG.md index ae81ccc410dd..90e1ef2e072a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,9 @@ for specific instructions. - [BUGFIX] Fix warn-level logging of dropped targets. (@james-callahan) +- [BUGFIX] Reloading the scraping service kvstore config for loading instance + configs will no longer use the clustering config instead. (@rfratto) + - [CHANGE] Breaking change: reduced verbosity of tracing autologging by not logging `STATUS_CODE_UNSET` status codes. (@mapno) @@ -79,7 +82,7 @@ for specific instructions. deprecated in favor of `metrics`. Flag names starting with `prometheus.` have also been deprecated in favor of the same flags with the `metrics.` prefix. (@rfratto) - + - [DEPRECATION] Rename Tempo to Traces (@mattdurham) # v0.18.4 (2021-09-14) diff --git a/docs/configuration/prometheus-config.md b/docs/configuration/prometheus-config.md index f16a9a8bdc4d..4f9bf17ecd5a 100644 --- a/docs/configuration/prometheus-config.md +++ b/docs/configuration/prometheus-config.md @@ -68,14 +68,15 @@ agents distribute discovery and scrape load between nodes. # events are not sent by an agent. [reshard_interval: | default = "1m"] -# The timeout for configuration refreshes. This can occur on cluster events or +# The timeout for configuration refreshes. This can occur on cluster events or # on the reshard interval. A timeout of 0 indicates no timeout. [reshard_timeout: | default = "30s"] # The timeout for a cluster reshard events. A timeout of 0 indicates no timeout. [cluster_reshard_event_timeout: | default = "30s"] -# Configuration for the KV store to store configurations. +# Configuration for the KV store to store configurations. Note that gossip +# cannot be configured for the scraping service config. kvstore: # When set, allows configs pushed to the KV store to specify configuration diff --git a/go.mod b/go.mod index 40fa7119c83c..b367eb1fe206 100644 --- a/go.mod +++ b/go.mod @@ -63,6 +63,9 @@ require ( github.com/stretchr/testify v1.7.0 github.com/uber/jaeger-client-go v2.29.1+incompatible github.com/weaveworks/common v0.0.0-20210419092856-009d1eebd624 + go.etcd.io/etcd v0.5.0-alpha.5.0.20200910180754-dd1b699fc489 + go.etcd.io/etcd/client/v3 v3.5.0 + go.etcd.io/etcd/server/v3 v3.5.0-alpha.0.0.20210225194612-fa82d11a958a go.mongodb.org/mongo-driver v1.5.3 go.opencensus.io v0.23.0 go.opentelemetry.io/collector v0.30.0 @@ -72,6 +75,7 @@ require ( go.uber.org/zap v1.18.1 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97 // indirect golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 + golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba google.golang.org/grpc v1.39.0 gopkg.in/alecthomas/kingpin.v2 v2.2.6 gopkg.in/yaml.v2 v2.4.0 diff --git a/pkg/metrics/cluster/cluster.go b/pkg/metrics/cluster/cluster.go index fa3a80ad9261..f84d34c31725 100644 --- a/pkg/metrics/cluster/cluster.go +++ b/pkg/metrics/cluster/cluster.go @@ -130,7 +130,7 @@ func (c *Cluster) ApplyConfig( return fmt.Errorf("failed to apply config to node membership: %w", err) } - if err := c.store.ApplyConfig(cfg.Lifecycler.RingConfig.KVStore, cfg.Enabled); err != nil { + if err := c.store.ApplyConfig(cfg.KVStore, cfg.Enabled); err != nil { return fmt.Errorf("failed to apply config to config store: %w", err) } diff --git a/pkg/metrics/cluster/config.go b/pkg/metrics/cluster/config.go index e8266bd1efba..0b7ab96da94e 100644 --- a/pkg/metrics/cluster/config.go +++ b/pkg/metrics/cluster/config.go @@ -5,8 +5,8 @@ import ( "time" "github.com/cortexproject/cortex/pkg/ring" - "github.com/cortexproject/cortex/pkg/ring/kv" "github.com/grafana/agent/pkg/metrics/cluster/client" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv" flagutil "github.com/grafana/agent/pkg/util" ) diff --git a/pkg/metrics/instance/configstore/kv/client.go b/pkg/metrics/instance/configstore/kv/client.go new file mode 100644 index 000000000000..229b21741499 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/client.go @@ -0,0 +1,201 @@ +package kv + +import ( + "context" + "flag" + "fmt" + "sync" + + "github.com/prometheus/client_golang/prometheus" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/consul" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/etcd" +) + +const ( + // Primary is a role to use KV store primarily. + Primary = role("primary") + // Secondary is a role for KV store used by "multi" KV store. + Secondary = role("secondary") +) + +const ( + clientConsul = "consul" + clientEtcd = "etcd" + clientInmemory = "inmemory" + clientMulti = "multi" + clientMock = "mock" +) + +// The role type indicates a role of KV store. +type role string + +// Labels method returns Prometheus labels relevant to itself. +func (r *role) Labels() prometheus.Labels { + return prometheus.Labels{"role": string(*r)} +} + +// The NewInMemoryKVClient returned by NewClient() is a singleton, so +// that distributors and ingesters started in the same process can +// find themselves. +var inmemoryStoreInit sync.Once +var inmemoryStore Client + +// StoreConfig is a configuration used for building single store client, either +// Consul, Etcd, or MultiClient. It was extracted from Config to keep +// single-client config separate from final client-config (with all the wrappers) +type StoreConfig struct { + Consul consul.Config `yaml:"consul"` + Etcd etcd.Config `yaml:"etcd"` + Multi MultiConfig `yaml:"multi"` +} + +// Config is config for a KVStore currently used by ring and HA tracker, +// where store can be consul or inmemory. +type Config struct { + Store string `yaml:"store"` + Prefix string `yaml:"prefix"` + StoreConfig `yaml:",inline"` + + Mock Client `yaml:"-"` +} + +// RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet. +// If prefix is an empty string we will register consul flags with no prefix and the +// store flag with the prefix ring, so ring.store. For everything else we pass the prefix +// to the Consul flags. +// If prefix is not an empty string it should end with a period. +func (cfg *Config) RegisterFlagsWithPrefix(flagsPrefix, defaultPrefix string, f *flag.FlagSet) { + // We need Consul flags to not have the ring prefix to maintain compatibility. + // This needs to be fixed in the future (1.0 release maybe?) when we normalize flags. + // At the moment we have consul., and ring.store, going forward it would + // be easier to have everything under ring, so ring.consul. + cfg.Consul.RegisterFlags(f, flagsPrefix) + cfg.Etcd.RegisterFlagsWithPrefix(f, flagsPrefix) + cfg.Multi.RegisterFlagsWithPrefix(f, flagsPrefix) + + if flagsPrefix == "" { + flagsPrefix = "ring." + } + f.StringVar(&cfg.Prefix, flagsPrefix+"prefix", defaultPrefix, "The prefix for the keys in the store. Should end with a /.") + f.StringVar(&cfg.Store, flagsPrefix+"store", "consul", "Backend storage to use for the ring. Supported values are: consul, etcd, inmemory, multi.") +} + +// Client is a high-level client for key-value stores (such as Etcd and +// Consul) that exposes operations such as CAS and Watch which take callbacks. +// It also deals with serialisation by using a Codec and having a instance of +// the the desired type passed in to methods ala json.Unmarshal. +type Client interface { + // List returns a list of keys under the given prefix. Returned keys will + // include the prefix. + List(ctx context.Context, prefix string) ([]string, error) + + // Get a specific key. Will use a codec to deserialise key to appropriate type. + // If the key does not exist, Get will return nil and no error. + Get(ctx context.Context, key string) (interface{}, error) + + // Delete a specific key. Deletions are best-effort and no error will + // be returned if the key does not exist. + Delete(ctx context.Context, key string) error + + // CAS stands for Compare-And-Swap. Will call provided callback f with the + // current value of the key and allow callback to return a different value. + // Will then attempt to atomically swap the current value for the new value. + // If that doesn't succeed will try again - callback will be called again + // with new value etc. Guarantees that only a single concurrent CAS + // succeeds. Callback can return nil to indicate it is happy with existing + // value. + CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error + + // WatchKey calls f whenever the value stored under key changes. + WatchKey(ctx context.Context, key string, f func(interface{}) bool) + + // WatchPrefix calls f whenever any value stored under prefix changes. + WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) +} + +// NewClient creates a new Client (consul, etcd or inmemory) based on the config, +// encodes and decodes data for storage using the codec. +func NewClient(cfg Config, codec codec.Codec, reg prometheus.Registerer) (Client, error) { + if cfg.Mock != nil { + return cfg.Mock, nil + } + + return createClient(cfg.Store, cfg.Prefix, cfg.StoreConfig, codec, Primary, reg) +} + +func createClient(backend string, prefix string, cfg StoreConfig, codec codec.Codec, role role, reg prometheus.Registerer) (Client, error) { + var client Client + var err error + + switch backend { + case clientConsul: + client, err = consul.NewClient(cfg.Consul, codec) + + case clientEtcd: + client, err = etcd.New(cfg.Etcd, codec) + + case clientInmemory: + // If we use the in-memory store, make sure everyone gets the same instance + // within the same process. + inmemoryStoreInit.Do(func() { + inmemoryStore = consul.NewInMemoryClient(codec) + }) + client = inmemoryStore + + case clientMulti: + client, err = buildMultiClient(cfg, codec, reg) + + // This case is for testing. The mock KV client does not do anything internally. + case clientMock: + client, err = buildMockClient() + + default: + return nil, fmt.Errorf("invalid KV store type: %s", backend) + } + + if err != nil { + return nil, err + } + + if prefix != "" { + client = PrefixClient(client, prefix) + } + + // If no Registerer is provided return the raw client. + if reg == nil { + return client, nil + } + + return newMetricsClient(backend, client, prometheus.WrapRegistererWith(role.Labels(), reg)), nil +} + +func buildMultiClient(cfg StoreConfig, codec codec.Codec, reg prometheus.Registerer) (Client, error) { + if cfg.Multi.Primary == "" || cfg.Multi.Secondary == "" { + return nil, fmt.Errorf("primary or secondary store not set") + } + if cfg.Multi.Primary == clientMulti || cfg.Multi.Secondary == clientMulti { + return nil, fmt.Errorf("primary and secondary stores cannot be multi-stores") + } + if cfg.Multi.Primary == cfg.Multi.Secondary { + return nil, fmt.Errorf("primary and secondary stores must be different") + } + + primary, err := createClient(cfg.Multi.Primary, "", cfg, codec, Primary, reg) + if err != nil { + return nil, err + } + + secondary, err := createClient(cfg.Multi.Secondary, "", cfg, codec, Secondary, reg) + if err != nil { + return nil, err + } + + clients := []kvclient{ + {client: primary, name: cfg.Multi.Primary}, + {client: secondary, name: cfg.Multi.Secondary}, + } + + return NewMultiClient(cfg.Multi, clients), nil +} diff --git a/pkg/metrics/instance/configstore/kv/client_test.go b/pkg/metrics/instance/configstore/kv/client_test.go new file mode 100644 index 000000000000..c0043efe3fd8 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/client_test.go @@ -0,0 +1,149 @@ +package kv + +import ( + "context" + "testing" + "time" + + "github.com/gogo/protobuf/proto" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v2" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" +) + +func TestParseConfig(t *testing.T) { + conf := ` +store: consul +consul: + host: "consul:8500" + consistentreads: true +prefix: "test/" +multi: + primary: consul + secondary: etcd +` + + cfg := Config{} + + err := yaml.Unmarshal([]byte(conf), &cfg) + require.NoError(t, err) + require.Equal(t, "consul", cfg.Store) + require.Equal(t, "test/", cfg.Prefix) + require.Equal(t, "consul:8500", cfg.Consul.Host) + require.Equal(t, "consul", cfg.Multi.Primary) + require.Equal(t, "etcd", cfg.Multi.Secondary) +} + +func Test_createClient_multiBackend_withSingleRing(t *testing.T) { + storeCfg, testCodec := newConfigsForTest() + require.NotPanics(t, func() { + _, err := createClient("multi", "/collector", storeCfg, testCodec, Primary, prometheus.NewRegistry()) + require.NoError(t, err) + }) +} + +func Test_createClient_multiBackend_withMultiRing(t *testing.T) { + storeCfg1, testCodec := newConfigsForTest() + storeCfg2 := StoreConfig{} + reg := prometheus.NewRegistry() + + require.NotPanics(t, func() { + _, err := createClient("multi", "/test", storeCfg1, testCodec, Primary, reg) + require.NoError(t, err) + }, "First client for KV store must not panic") + require.NotPanics(t, func() { + _, err := createClient("mock", "/test", storeCfg2, testCodec, Primary, reg) + require.NoError(t, err) + }, "Second client for KV store must not panic") +} + +func Test_createClient_singleBackend_mustContainRoleAndTypeLabels(t *testing.T) { + storeCfg, testCodec := newConfigsForTest() + reg := prometheus.NewRegistry() + client, err := createClient("mock", "/test1", storeCfg, testCodec, Primary, reg) + require.NoError(t, err) + require.NoError(t, client.CAS(context.Background(), "/test", func(_ interface{}) (out interface{}, retry bool, err error) { + out = &mockMessage{id: "inCAS"} + retry = false + return + })) + + actual := typeToRoleMap(t, reg) + require.Len(t, actual, 1) + require.Equal(t, "primary", actual["mock"]) +} + +func Test_createClient_multiBackend_mustContainRoleAndTypeLabels(t *testing.T) { + storeCfg, testCodec := newConfigsForTest() + storeCfg.Multi.MirrorEnabled = true + storeCfg.Multi.MirrorTimeout = 10 * time.Second + reg := prometheus.NewRegistry() + client, err := createClient("multi", "/test1", storeCfg, testCodec, Primary, reg) + require.NoError(t, err) + require.NoError(t, client.CAS(context.Background(), "/test", func(_ interface{}) (out interface{}, retry bool, err error) { + out = &mockMessage{id: "inCAS"} + retry = false + return + })) + + actual := typeToRoleMap(t, reg) + // expected multi-primary, inmemory-primary and mock-secondary + require.Len(t, actual, 3) + require.Equal(t, "primary", actual["multi"]) + require.Equal(t, "primary", actual["inmemory"]) + require.Equal(t, "secondary", actual["mock"]) + +} + +func typeToRoleMap(t *testing.T, reg prometheus.Gatherer) map[string]string { + mfs, err := reg.Gather() + require.NoError(t, err) + result := map[string]string{} + for _, mf := range mfs { + for _, m := range mf.GetMetric() { + backendType := "" + role := "" + for _, l := range m.GetLabel() { + if l.GetName() == "role" { + role = l.GetValue() + } else if l.GetName() == "type" { + backendType = l.GetValue() + } + } + require.NotEmpty(t, backendType) + require.NotEmpty(t, role) + result[backendType] = role + } + } + return result +} +func newConfigsForTest() (cfg StoreConfig, c codec.Codec) { + cfg = StoreConfig{ + Multi: MultiConfig{ + Primary: "inmemory", + Secondary: "mock", + }, + } + c = codec.NewProtoCodec("test", func() proto.Message { + return &mockMessage{id: "inCodec"} + }) + return +} + +type mockMessage struct { + id string +} + +func (m *mockMessage) Reset() { + panic("do not use") +} + +func (m *mockMessage) String() string { + panic("do not use") +} + +func (m *mockMessage) ProtoMessage() { + panic("do not use") +} diff --git a/pkg/metrics/instance/configstore/kv/consul/client.go b/pkg/metrics/instance/configstore/kv/consul/client.go new file mode 100644 index 000000000000..1c39a473ced7 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/consul/client.go @@ -0,0 +1,392 @@ +package consul + +import ( + "context" + "errors" + "flag" + "fmt" + "math/rand" + "net/http" + "time" + + "github.com/go-kit/kit/log/level" + consul "github.com/hashicorp/consul/api" + "github.com/hashicorp/go-cleanhttp" + "github.com/weaveworks/common/instrument" + "golang.org/x/time/rate" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" + "github.com/cortexproject/cortex/pkg/util" + util_log "github.com/cortexproject/cortex/pkg/util/log" +) + +const ( + longPollDuration = 10 * time.Second +) + +var ( + writeOptions = &consul.WriteOptions{} + + // ErrNotFound is returned by ConsulClient.Get. + ErrNotFound = fmt.Errorf("Not found") + + backoffConfig = util.BackoffConfig{ + MinBackoff: 1 * time.Second, + MaxBackoff: 1 * time.Minute, + } +) + +// Config to create a ConsulClient +type Config struct { + Host string `yaml:"host"` + ACLToken string `yaml:"acl_token"` + HTTPClientTimeout time.Duration `yaml:"http_client_timeout"` + ConsistentReads bool `yaml:"consistent_reads"` + WatchKeyRateLimit float64 `yaml:"watch_rate_limit"` // Zero disables rate limit + WatchKeyBurstSize int `yaml:"watch_burst_size"` // Burst when doing rate-limit, defaults to 1 + + // Used in tests only. + MaxCasRetries int `yaml:"-"` + CasRetryDelay time.Duration `yaml:"-"` +} + +type kv interface { + CAS(p *consul.KVPair, q *consul.WriteOptions) (bool, *consul.WriteMeta, error) + Get(key string, q *consul.QueryOptions) (*consul.KVPair, *consul.QueryMeta, error) + List(path string, q *consul.QueryOptions) (consul.KVPairs, *consul.QueryMeta, error) + Delete(key string, q *consul.WriteOptions) (*consul.WriteMeta, error) + Put(p *consul.KVPair, q *consul.WriteOptions) (*consul.WriteMeta, error) +} + +// Client is a KV.Client for Consul. +type Client struct { + kv + codec codec.Codec + cfg Config +} + +// RegisterFlags adds the flags required to config this to the given FlagSet +// If prefix is not an empty string it should end with a period. +func (cfg *Config) RegisterFlags(f *flag.FlagSet, prefix string) { + f.StringVar(&cfg.Host, prefix+"consul.hostname", "localhost:8500", "Hostname and port of Consul.") + f.StringVar(&cfg.ACLToken, prefix+"consul.acl-token", "", "ACL Token used to interact with Consul.") + f.DurationVar(&cfg.HTTPClientTimeout, prefix+"consul.client-timeout", 2*longPollDuration, "HTTP timeout when talking to Consul") + f.BoolVar(&cfg.ConsistentReads, prefix+"consul.consistent-reads", false, "Enable consistent reads to Consul.") + f.Float64Var(&cfg.WatchKeyRateLimit, prefix+"consul.watch-rate-limit", 1, "Rate limit when watching key or prefix in Consul, in requests per second. 0 disables the rate limit.") + f.IntVar(&cfg.WatchKeyBurstSize, prefix+"consul.watch-burst-size", 1, "Burst size used in rate limit. Values less than 1 are treated as 1.") +} + +// NewClient returns a new Client. +func NewClient(cfg Config, codec codec.Codec) (*Client, error) { + client, err := consul.NewClient(&consul.Config{ + Address: cfg.Host, + Token: cfg.ACLToken, + Scheme: "http", + HttpClient: &http.Client{ + Transport: cleanhttp.DefaultPooledTransport(), + // See https://blog.cloudflare.com/the-complete-guide-to-golang-net-http-timeouts/ + Timeout: cfg.HTTPClientTimeout, + }, + }) + if err != nil { + return nil, err + } + c := &Client{ + kv: consulMetrics{client.KV()}, + codec: codec, + cfg: cfg, + } + return c, nil +} + +// Put is mostly here for testing. +func (c *Client) Put(ctx context.Context, key string, value interface{}) error { + bytes, err := c.codec.Encode(value) + if err != nil { + return err + } + + return instrument.CollectedRequest(ctx, "Put", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + _, err := c.kv.Put(&consul.KVPair{ + Key: key, + Value: bytes, + }, nil) + return err + }) +} + +// CAS atomically modifies a value in a callback. +// If value doesn't exist you'll get nil as an argument to your callback. +func (c *Client) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + return instrument.CollectedRequest(ctx, "CAS loop", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + return c.cas(ctx, key, f) + }) +} + +func (c *Client) cas(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + retries := c.cfg.MaxCasRetries + if retries == 0 { + retries = 10 + } + + sleepBeforeRetry := time.Duration(0) + if c.cfg.CasRetryDelay > 0 { + sleepBeforeRetry = time.Duration(rand.Int63n(c.cfg.CasRetryDelay.Nanoseconds())) + } + + index := uint64(0) + for i := 0; i < retries; i++ { + if i > 0 && sleepBeforeRetry > 0 { + time.Sleep(sleepBeforeRetry) + } + + // Get with default options - don't want stale data to compare with + options := &consul.QueryOptions{} + kvp, _, err := c.kv.Get(key, options.WithContext(ctx)) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error getting key", "key", key, "err", err) + continue + } + var intermediate interface{} + if kvp != nil { + out, err := c.codec.Decode(kvp.Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding key", "key", key, "err", err) + continue + } + // If key doesn't exist, index will be 0. + index = kvp.ModifyIndex + intermediate = out + } + + intermediate, retry, err := f(intermediate) + if err != nil { + if !retry { + return err + } + continue + } + + // Treat the callback returning nil for intermediate as a decision to + // not actually write to Consul, but this is not an error. + if intermediate == nil { + return nil + } + + bytes, err := c.codec.Encode(intermediate) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error serialising value", "key", key, "err", err) + continue + } + ok, _, err := c.kv.CAS(&consul.KVPair{ + Key: key, + Value: bytes, + ModifyIndex: index, + }, writeOptions.WithContext(ctx)) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error CASing", "key", key, "err", err) + continue + } + if !ok { + level.Debug(util_log.Logger).Log("msg", "error CASing, trying again", "key", key, "index", index) + continue + } + return nil + } + return fmt.Errorf("failed to CAS %s", key) +} + +// WatchKey will watch a given key in consul for changes. When the value +// under said key changes, the f callback is called with the deserialised +// value. To construct the deserialised value, a factory function should be +// supplied which generates an empty struct for WatchKey to deserialise +// into. This function blocks until the context is cancelled or f returns false. +func (c *Client) WatchKey(ctx context.Context, key string, f func(interface{}) bool) { + var ( + backoff = util.NewBackoff(ctx, backoffConfig) + index = uint64(0) + limiter = c.createRateLimiter() + ) + + for backoff.Ongoing() { + err := limiter.Wait(ctx) + if err != nil { + if errors.Is(err, context.Canceled) { + break + } + level.Error(util_log.Logger).Log("msg", "error while rate-limiting", "key", key, "err", err) + backoff.Wait() + continue + } + + queryOptions := &consul.QueryOptions{ + AllowStale: !c.cfg.ConsistentReads, + RequireConsistent: c.cfg.ConsistentReads, + WaitIndex: index, + WaitTime: longPollDuration, + } + + kvp, meta, err := c.kv.Get(key, queryOptions.WithContext(ctx)) + + // Don't backoff if value is not found (kvp == nil). In that case, Consul still returns index value, + // and next call to Get will block as expected. We handle missing value below. + if err != nil { + level.Error(util_log.Logger).Log("msg", "error getting path", "key", key, "err", err) + backoff.Wait() + continue + } + backoff.Reset() + + skip := false + index, skip = checkLastIndex(index, meta.LastIndex) + if skip { + continue + } + + if kvp == nil { + level.Info(util_log.Logger).Log("msg", "value is nil", "key", key, "index", index) + continue + } + + out, err := c.codec.Decode(kvp.Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding key", "key", key, "err", err) + continue + } + if !f(out) { + return + } + } +} + +// WatchPrefix will watch a given prefix in Consul for new keys and changes to existing keys under that prefix. +// When the value under said key changes, the f callback is called with the deserialised value. +// Values in Consul are assumed to be JSON. This function blocks until the context is cancelled. +func (c *Client) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) { + var ( + backoff = util.NewBackoff(ctx, backoffConfig) + index = uint64(0) + limiter = c.createRateLimiter() + ) + for backoff.Ongoing() { + err := limiter.Wait(ctx) + if err != nil { + if errors.Is(err, context.Canceled) { + break + } + level.Error(util_log.Logger).Log("msg", "error while rate-limiting", "prefix", prefix, "err", err) + backoff.Wait() + continue + } + + queryOptions := &consul.QueryOptions{ + AllowStale: !c.cfg.ConsistentReads, + RequireConsistent: c.cfg.ConsistentReads, + WaitIndex: index, + WaitTime: longPollDuration, + } + + kvps, meta, err := c.kv.List(prefix, queryOptions.WithContext(ctx)) + // kvps being nil here is not an error -- quite the opposite. Consul returns index, + // which makes next query blocking, so there is no need to detect this and act on it. + if err != nil { + level.Error(util_log.Logger).Log("msg", "error getting path", "prefix", prefix, "err", err) + backoff.Wait() + continue + } + backoff.Reset() + + newIndex, skip := checkLastIndex(index, meta.LastIndex) + if skip { + continue + } + + for _, kvp := range kvps { + // We asked for values newer than 'index', but Consul returns all values below given prefix, + // even those that haven't changed. We don't need to report all of them as updated. + if index > 0 && kvp.ModifyIndex <= index && kvp.CreateIndex <= index { + continue + } + + out, err := c.codec.Decode(kvp.Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding list of values for prefix:key", "prefix", prefix, "key", kvp.Key, "err", err) + continue + } + if !f(kvp.Key, out) { + return + } + } + + index = newIndex + } +} + +// List implements kv.List. +func (c *Client) List(ctx context.Context, prefix string) ([]string, error) { + options := &consul.QueryOptions{ + AllowStale: !c.cfg.ConsistentReads, + RequireConsistent: c.cfg.ConsistentReads, + } + pairs, _, err := c.kv.List(prefix, options.WithContext(ctx)) + if err != nil { + return nil, err + } + + keys := make([]string, 0, len(pairs)) + for _, kvp := range pairs { + keys = append(keys, kvp.Key) + } + return keys, nil +} + +// Get implements kv.Get. +func (c *Client) Get(ctx context.Context, key string) (interface{}, error) { + options := &consul.QueryOptions{ + AllowStale: !c.cfg.ConsistentReads, + RequireConsistent: c.cfg.ConsistentReads, + } + kvp, _, err := c.kv.Get(key, options.WithContext(ctx)) + if err != nil { + return nil, err + } else if kvp == nil { + return nil, nil + } + return c.codec.Decode(kvp.Value) +} + +// Delete implements kv.Delete. +func (c *Client) Delete(ctx context.Context, key string) error { + _, err := c.kv.Delete(key, writeOptions.WithContext(ctx)) + return err +} + +func checkLastIndex(index, metaLastIndex uint64) (newIndex uint64, skip bool) { + // See https://www.consul.io/api/features/blocking.html#implementation-details for logic behind these checks. + if metaLastIndex == 0 { + // Don't just keep using index=0. + // After blocking request, returned index must be at least 1. + return 1, false + } else if metaLastIndex < index { + // Index reset. + return 0, false + } else if index == metaLastIndex { + // Skip if the index is the same as last time, because the key value is + // guaranteed to be the same as last time + return metaLastIndex, true + } else { + return metaLastIndex, false + } +} + +func (c *Client) createRateLimiter() *rate.Limiter { + if c.cfg.WatchKeyRateLimit <= 0 { + // burst is ignored when limit = rate.Inf + return rate.NewLimiter(rate.Inf, 0) + } + burst := c.cfg.WatchKeyBurstSize + if burst < 1 { + burst = 1 + } + return rate.NewLimiter(rate.Limit(c.cfg.WatchKeyRateLimit), burst) +} diff --git a/pkg/metrics/instance/configstore/kv/consul/client_test.go b/pkg/metrics/instance/configstore/kv/consul/client_test.go new file mode 100644 index 000000000000..cc741c0db2c3 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/consul/client_test.go @@ -0,0 +1,165 @@ +package consul + +import ( + "context" + "fmt" + "strconv" + "testing" + "time" + + "github.com/go-kit/kit/log/level" + consul "github.com/hashicorp/consul/api" + "github.com/stretchr/testify/require" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" + util_log "github.com/cortexproject/cortex/pkg/util/log" +) + +func writeValuesToKV(client *Client, key string, start, end int, sleep time.Duration) <-chan struct{} { + ch := make(chan struct{}) + go func() { + defer close(ch) + for i := start; i <= end; i++ { + level.Debug(util_log.Logger).Log("ts", time.Now(), "msg", "writing value", "val", i) + _, _ = client.kv.Put(&consul.KVPair{Key: key, Value: []byte(fmt.Sprintf("%d", i))}, nil) + time.Sleep(sleep) + } + }() + return ch +} + +func TestWatchKeyWithRateLimit(t *testing.T) { + c := NewInMemoryClientWithConfig(codec.String{}, Config{ + WatchKeyRateLimit: 5.0, + WatchKeyBurstSize: 1, + }) + + const key = "test" + const max = 100 + + ch := writeValuesToKV(c, key, 0, max, 10*time.Millisecond) + + // Only observe for a little over 1.5 rate limits (5 req/s) + observed := observeValueForSomeTime(c, key, 1700*time.Millisecond) + + // wait until updater finishes + <-ch + + if testing.Verbose() { + t.Log(observed) + } + // Let's see how many updates we have observed. Given the rate limit and our observing time, it should be 6 + // We should also have seen one of the later values, as we're observing for longer than a second, so rate limit should allow + // us to see it. + if len(observed) < 5 || len(observed) > 10 { + t.Error("Expected ~6 observed values, got", observed) + } + last := observed[len(observed)-1] + n, _ := strconv.Atoi(last) + if n < max/2 { + t.Error("Expected to see high last observed value, got", observed) + } +} + +func TestWatchKeyNoRateLimit(t *testing.T) { + c := NewInMemoryClientWithConfig(codec.String{}, Config{ + WatchKeyRateLimit: 0, + }) + + const key = "test" + const max = 100 + + ch := writeValuesToKV(c, key, 0, max, time.Millisecond) + observed := observeValueForSomeTime(c, key, 2400*time.Millisecond) + + // wait until updater finishes + <-ch + + // With no limit, we should see most written values (we can lose some values if watching + // code is busy while multiple new values are written) + if len(observed) < 3*max/4 { + t.Error("Expected at least 3/4 of all values, got", observed) + } +} + +func TestReset(t *testing.T) { + c := NewInMemoryClient(codec.String{}) + + const key = "test" + const max = 5 + + ch := make(chan error) + go func() { + defer close(ch) + for i := 0; i <= max; i++ { + level.Debug(util_log.Logger).Log("ts", time.Now(), "msg", "writing value", "val", i) + _, _ = c.kv.Put(&consul.KVPair{Key: key, Value: []byte(fmt.Sprintf("%d", i))}, nil) + if i == 1 { + c.kv.(*mockKV).ResetIndex() + } + if i == 2 { + c.kv.(*mockKV).ResetIndexForKey(key) + } + time.Sleep(10 * time.Millisecond) + } + }() + + observed := observeValueForSomeTime(c, key, 2400*time.Millisecond) + + // wait until updater finishes + <-ch + + // Let's see how many updates we have observed. Given the rate limit and our observing time, we should see all numeric values + if testing.Verbose() { + t.Log(observed) + } + if len(observed) < max { + t.Error("Expected all values, got", observed) + } else if observed[len(observed)-1] != fmt.Sprintf("%d", max) { + t.Error("Expected to see last written value, got", observed) + } +} + +func observeValueForSomeTime(client *Client, key string, timeout time.Duration) []string { + observed := []string(nil) + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + client.WatchKey(ctx, key, func(i interface{}) bool { + s, ok := i.(string) + if !ok { + return false + } + level.Debug(util_log.Logger).Log("ts", time.Now(), "msg", "observed value", "val", s) + observed = append(observed, s) + return true + }) + return observed +} + +func TestWatchKeyWithNoStartValue(t *testing.T) { + c := NewInMemoryClient(codec.String{}) + + const key = "test" + + go func() { + time.Sleep(100 * time.Millisecond) + _, err := c.kv.Put(&consul.KVPair{Key: key, Value: []byte("start")}, nil) + require.NoError(t, err) + + time.Sleep(100 * time.Millisecond) + _, err = c.kv.Put(&consul.KVPair{Key: key, Value: []byte("end")}, nil) + require.NoError(t, err) + }() + + ctx, fn := context.WithTimeout(context.Background(), 750*time.Millisecond) + defer fn() + + reported := 0 + c.WatchKey(ctx, key, func(i interface{}) bool { + reported++ + return reported != 2 + }) + + // we should see both start and end values. + require.Equal(t, 2, reported) +} diff --git a/pkg/metrics/instance/configstore/kv/consul/metrics.go b/pkg/metrics/instance/configstore/kv/consul/metrics.go new file mode 100644 index 000000000000..8cb45463ddaf --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/consul/metrics.go @@ -0,0 +1,82 @@ +package consul + +import ( + "context" + + consul "github.com/hashicorp/consul/api" + "github.com/prometheus/client_golang/prometheus" + "github.com/weaveworks/common/instrument" +) + +var consulRequestDuration = instrument.NewHistogramCollector(prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "agent", + Name: "consul_request_duration_seconds", + Help: "Time spent on consul requests.", + Buckets: prometheus.DefBuckets, +}, []string{"operation", "status_code"})) + +func init() { + consulRequestDuration.Register() +} + +type consulMetrics struct { + kv +} + +func (c consulMetrics) CAS(p *consul.KVPair, options *consul.WriteOptions) (bool, *consul.WriteMeta, error) { + var ok bool + var result *consul.WriteMeta + err := instrument.CollectedRequest(options.Context(), "CAS", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + options = options.WithContext(ctx) + var err error + ok, result, err = c.kv.CAS(p, options) + return err + }) + return ok, result, err +} + +func (c consulMetrics) Get(key string, options *consul.QueryOptions) (*consul.KVPair, *consul.QueryMeta, error) { + var kvp *consul.KVPair + var meta *consul.QueryMeta + err := instrument.CollectedRequest(options.Context(), "Get", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + options = options.WithContext(ctx) + var err error + kvp, meta, err = c.kv.Get(key, options) + return err + }) + return kvp, meta, err +} + +func (c consulMetrics) List(path string, options *consul.QueryOptions) (consul.KVPairs, *consul.QueryMeta, error) { + var kvps consul.KVPairs + var meta *consul.QueryMeta + err := instrument.CollectedRequest(options.Context(), "List", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + options = options.WithContext(ctx) + var err error + kvps, meta, err = c.kv.List(path, options) + return err + }) + return kvps, meta, err +} + +func (c consulMetrics) Delete(key string, options *consul.WriteOptions) (*consul.WriteMeta, error) { + var meta *consul.WriteMeta + err := instrument.CollectedRequest(options.Context(), "Delete", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + options = options.WithContext(ctx) + var err error + meta, err = c.kv.Delete(key, options) + return err + }) + return meta, err +} + +func (c consulMetrics) Put(p *consul.KVPair, options *consul.WriteOptions) (*consul.WriteMeta, error) { + var result *consul.WriteMeta + err := instrument.CollectedRequest(options.Context(), "Put", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { + options = options.WithContext(ctx) + var err error + result, err = c.kv.Put(p, options) + return err + }) + return result, err +} diff --git a/pkg/metrics/instance/configstore/kv/consul/mock.go b/pkg/metrics/instance/configstore/kv/consul/mock.go new file mode 100644 index 000000000000..5d1e4557395b --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/consul/mock.go @@ -0,0 +1,230 @@ +package consul + +import ( + "fmt" + "strings" + "sync" + "time" + + "github.com/go-kit/kit/log/level" + consul "github.com/hashicorp/consul/api" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" + util_log "github.com/cortexproject/cortex/pkg/util/log" +) + +type mockKV struct { + mtx sync.Mutex + cond *sync.Cond + kvps map[string]*consul.KVPair + current uint64 // the current 'index in the log' +} + +// NewInMemoryClient makes a new mock consul client. +func NewInMemoryClient(codec codec.Codec) *Client { + return NewInMemoryClientWithConfig(codec, Config{}) +} + +// NewInMemoryClientWithConfig makes a new mock consul client with supplied Config. +func NewInMemoryClientWithConfig(codec codec.Codec, cfg Config) *Client { + m := mockKV{ + kvps: map[string]*consul.KVPair{}, + // Always start from 1, we NEVER want to report back index 0 in the responses. + // This is in line with Consul, and our new checks for index return value in client.go. + current: 1, + } + m.cond = sync.NewCond(&m.mtx) + go m.loop() + return &Client{ + kv: &m, + codec: codec, + cfg: cfg, + } +} + +func copyKVPair(in *consul.KVPair) *consul.KVPair { + out := *in + out.Value = make([]byte, len(in.Value)) + copy(out.Value, in.Value) + return &out +} + +// periodic loop to wake people up, so they can honour timeouts +func (m *mockKV) loop() { + for range time.Tick(1 * time.Second) { + m.mtx.Lock() + m.cond.Broadcast() + m.mtx.Unlock() + } +} + +func (m *mockKV) Put(p *consul.KVPair, q *consul.WriteOptions) (*consul.WriteMeta, error) { + m.mtx.Lock() + defer m.mtx.Unlock() + + m.current++ + existing, ok := m.kvps[p.Key] + if ok { + existing.Value = p.Value + existing.ModifyIndex = m.current + } else { + m.kvps[p.Key] = &consul.KVPair{ + Key: p.Key, + Value: p.Value, + CreateIndex: m.current, + ModifyIndex: m.current, + } + } + + m.cond.Broadcast() + + level.Debug(util_log.Logger).Log("msg", "Put", "key", p.Key, "value", fmt.Sprintf("%.40q", p.Value), "modify_index", m.current) + return nil, nil +} + +func (m *mockKV) CAS(p *consul.KVPair, q *consul.WriteOptions) (bool, *consul.WriteMeta, error) { + level.Debug(util_log.Logger).Log("msg", "CAS", "key", p.Key, "modify_index", p.ModifyIndex, "value", fmt.Sprintf("%.40q", p.Value)) + + m.mtx.Lock() + defer m.mtx.Unlock() + existing, ok := m.kvps[p.Key] + if ok && existing.ModifyIndex != p.ModifyIndex { + return false, nil, nil + } + + m.current++ + if ok { + existing.Value = p.Value + existing.ModifyIndex = m.current + } else { + m.kvps[p.Key] = &consul.KVPair{ + Key: p.Key, + Value: p.Value, + CreateIndex: m.current, + ModifyIndex: m.current, + } + } + + m.cond.Broadcast() + return true, nil, nil +} + +func (m *mockKV) Get(key string, q *consul.QueryOptions) (*consul.KVPair, *consul.QueryMeta, error) { + level.Debug(util_log.Logger).Log("msg", "Get", "key", key, "wait_index", q.WaitIndex) + + m.mtx.Lock() + defer m.mtx.Unlock() + + value := m.kvps[key] + if value == nil && q.WaitIndex == 0 { + level.Debug(util_log.Logger).Log("msg", "Get - not found", "key", key) + return nil, &consul.QueryMeta{LastIndex: m.current}, nil + } + + var valueModifyIndex uint64 + if value != nil { + valueModifyIndex = value.ModifyIndex + } else { + valueModifyIndex = m.current + } + + if q.WaitIndex >= valueModifyIndex && q.WaitTime > 0 { + deadline := time.Now().Add(mockedMaxWaitTime(q.WaitTime)) + if ctxDeadline, ok := q.Context().Deadline(); ok && ctxDeadline.Before(deadline) { + // respect deadline from context, if set. + deadline = ctxDeadline + } + + // simply wait until value.ModifyIndex changes. This allows us to test reporting old index values by resetting them. + startModify := valueModifyIndex + for startModify == valueModifyIndex && time.Now().Before(deadline) { + m.cond.Wait() + value = m.kvps[key] + + if value != nil { + valueModifyIndex = value.ModifyIndex + } + } + if time.Now().After(deadline) { + level.Debug(util_log.Logger).Log("msg", "Get - deadline exceeded", "key", key) + return nil, &consul.QueryMeta{LastIndex: q.WaitIndex}, nil + } + } + + if value == nil { + level.Debug(util_log.Logger).Log("msg", "Get - not found", "key", key) + return nil, &consul.QueryMeta{LastIndex: m.current}, nil + } + + level.Debug(util_log.Logger).Log("msg", "Get", "key", key, "modify_index", value.ModifyIndex, "value", fmt.Sprintf("%.40q", value.Value)) + return copyKVPair(value), &consul.QueryMeta{LastIndex: value.ModifyIndex}, nil +} + +func (m *mockKV) List(prefix string, q *consul.QueryOptions) (consul.KVPairs, *consul.QueryMeta, error) { + m.mtx.Lock() + defer m.mtx.Unlock() + + if q.WaitTime > 0 { + deadline := time.Now().Add(mockedMaxWaitTime(q.WaitTime)) + if ctxDeadline, ok := q.Context().Deadline(); ok && ctxDeadline.Before(deadline) { + // respect deadline from context, if set. + deadline = ctxDeadline + } + + for q.WaitIndex >= m.current && time.Now().Before(deadline) { + m.cond.Wait() + } + if time.Now().After(deadline) { + return nil, &consul.QueryMeta{LastIndex: q.WaitIndex}, nil + } + } + + result := consul.KVPairs{} + for _, kvp := range m.kvps { + if strings.HasPrefix(kvp.Key, prefix) && kvp.ModifyIndex >= q.WaitIndex { + // unfortunately real consul doesn't do index check and returns everything with given prefix. + result = append(result, copyKVPair(kvp)) + } + } + return result, &consul.QueryMeta{LastIndex: m.current}, nil +} + +func (m *mockKV) Delete(key string, q *consul.WriteOptions) (*consul.WriteMeta, error) { + m.mtx.Lock() + defer m.mtx.Unlock() + delete(m.kvps, key) + return nil, nil +} + +func (m *mockKV) ResetIndex() { + m.mtx.Lock() + defer m.mtx.Unlock() + + m.current = 0 + m.cond.Broadcast() + + level.Debug(util_log.Logger).Log("msg", "Reset") +} + +func (m *mockKV) ResetIndexForKey(key string) { + m.mtx.Lock() + defer m.mtx.Unlock() + + if value, ok := m.kvps[key]; ok { + value.ModifyIndex = 0 + } + + m.cond.Broadcast() + level.Debug(util_log.Logger).Log("msg", "ResetIndexForKey", "key", key) +} + +// mockedMaxWaitTime returns the minimum duration between the input duration +// and the max wait time allowed in this mock, in order to have faster tests. +func mockedMaxWaitTime(queryWaitTime time.Duration) time.Duration { + const maxWaitTime = time.Second + if queryWaitTime > maxWaitTime { + return maxWaitTime + } + + return queryWaitTime +} diff --git a/pkg/metrics/instance/configstore/kv/etcd/etcd.go b/pkg/metrics/instance/configstore/kv/etcd/etcd.go new file mode 100644 index 000000000000..d981c00320ad --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/etcd/etcd.go @@ -0,0 +1,280 @@ +package etcd + +import ( + "context" + "crypto/tls" + "flag" + "fmt" + "time" + + "github.com/go-kit/kit/log/level" + "github.com/pkg/errors" + clientv3 "go.etcd.io/etcd/client/v3" + "go.etcd.io/etcd/pkg/transport" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" + "github.com/cortexproject/cortex/pkg/util" + "github.com/cortexproject/cortex/pkg/util/flagext" + util_log "github.com/cortexproject/cortex/pkg/util/log" + cortex_tls "github.com/cortexproject/cortex/pkg/util/tls" +) + +// Config for a new etcd.Client. +type Config struct { + Endpoints []string `yaml:"endpoints"` + DialTimeout time.Duration `yaml:"dial_timeout"` + MaxRetries int `yaml:"max_retries"` + EnableTLS bool `yaml:"tls_enabled"` + TLS cortex_tls.ClientConfig `yaml:",inline"` +} + +// Client implements ring.KVClient for etcd. +type Client struct { + cfg Config + codec codec.Codec + cli *clientv3.Client +} + +// RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet. +func (cfg *Config) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) { + cfg.Endpoints = []string{} + f.Var((*flagext.StringSlice)(&cfg.Endpoints), prefix+"etcd.endpoints", "The etcd endpoints to connect to.") + f.DurationVar(&cfg.DialTimeout, prefix+"etcd.dial-timeout", 10*time.Second, "The dial timeout for the etcd connection.") + f.IntVar(&cfg.MaxRetries, prefix+"etcd.max-retries", 10, "The maximum number of retries to do for failed ops.") + f.BoolVar(&cfg.EnableTLS, prefix+"etcd.tls-enabled", false, "Enable TLS.") + cfg.TLS.RegisterFlagsWithPrefix(prefix+"etcd", f) +} + +// GetTLS sets the TLS config field with certs +func (cfg *Config) GetTLS() (*tls.Config, error) { + if !cfg.EnableTLS { + return nil, nil + } + tlsInfo := &transport.TLSInfo{ + CertFile: cfg.TLS.CertPath, + KeyFile: cfg.TLS.KeyPath, + TrustedCAFile: cfg.TLS.CAPath, + ServerName: cfg.TLS.ServerName, + InsecureSkipVerify: cfg.TLS.InsecureSkipVerify, + } + return tlsInfo.ClientConfig() +} + +// New makes a new Client. +func New(cfg Config, codec codec.Codec) (*Client, error) { + tlsConfig, err := cfg.GetTLS() + if err != nil { + return nil, errors.Wrapf(err, "unable to initialise TLS configuration for etcd") + } + cli, err := clientv3.New(clientv3.Config{ + Endpoints: cfg.Endpoints, + DialTimeout: cfg.DialTimeout, + // Configure the keepalive to make sure that the client reconnects + // to the etcd service endpoint(s) in case the current connection is + // dead (ie. the node where etcd is running is dead or a network + // partition occurs). + // + // The settings: + // - DialKeepAliveTime: time before the client pings the server to + // see if transport is alive (10s hardcoded) + // - DialKeepAliveTimeout: time the client waits for a response for + // the keep-alive probe (set to 2x dial timeout, in order to avoid + // exposing another config option which is likely to be a factor of + // the dial timeout anyway) + // - PermitWithoutStream: whether the client should send keepalive pings + // to server without any active streams (enabled) + DialKeepAliveTime: 10 * time.Second, + DialKeepAliveTimeout: 2 * cfg.DialTimeout, + PermitWithoutStream: true, + TLS: tlsConfig, + }) + if err != nil { + return nil, err + } + + return &Client{ + cfg: cfg, + codec: codec, + cli: cli, + }, nil +} + +// CAS implements kv.Client. +func (c *Client) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + var revision int64 + var lastErr error + + for i := 0; i < c.cfg.MaxRetries; i++ { + resp, err := c.cli.Get(ctx, key) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error getting key", "key", key, "err", err) + lastErr = err + continue + } + + var intermediate interface{} + if len(resp.Kvs) > 0 { + intermediate, err = c.codec.Decode(resp.Kvs[0].Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding key", "key", key, "err", err) + lastErr = err + continue + } + revision = resp.Kvs[0].Version + } + + var retry bool + intermediate, retry, err = f(intermediate) + if err != nil { + if !retry { + return err + } + lastErr = err + continue + } + + // Callback returning nil means it doesn't want to CAS anymore. + if intermediate == nil { + return nil + } + + buf, err := c.codec.Encode(intermediate) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error serialising value", "key", key, "err", err) + lastErr = err + continue + } + + result, err := c.cli.Txn(ctx). + If(clientv3.Compare(clientv3.Version(key), "=", revision)). + Then(clientv3.OpPut(key, string(buf))). + Commit() + if err != nil { + level.Error(util_log.Logger).Log("msg", "error CASing", "key", key, "err", err) + lastErr = err + continue + } + // result is not Succeeded if the the comparison was false, meaning if the modify indexes did not match. + if !result.Succeeded { + level.Debug(util_log.Logger).Log("msg", "failed to CAS, revision and version did not match in etcd", "key", key, "revision", revision) + continue + } + + return nil + } + + if lastErr != nil { + return lastErr + } + return fmt.Errorf("failed to CAS %s", key) +} + +// WatchKey implements kv.Client. +func (c *Client) WatchKey(ctx context.Context, key string, f func(interface{}) bool) { + backoff := util.NewBackoff(ctx, util.BackoffConfig{ + MinBackoff: 1 * time.Second, + MaxBackoff: 1 * time.Minute, + }) + + // Ensure the context used by the Watch is always cancelled. + watchCtx, cancel := context.WithCancel(ctx) + defer cancel() + +outer: + for backoff.Ongoing() { + for resp := range c.cli.Watch(watchCtx, key) { + if err := resp.Err(); err != nil { + level.Error(util_log.Logger).Log("msg", "watch error", "key", key, "err", err) + continue outer + } + + backoff.Reset() + + for _, event := range resp.Events { + out, err := c.codec.Decode(event.Kv.Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding key", "key", key, "err", err) + continue + } + + if !f(out) { + return + } + } + } + } +} + +// WatchPrefix implements kv.Client. +func (c *Client) WatchPrefix(ctx context.Context, key string, f func(string, interface{}) bool) { + backoff := util.NewBackoff(ctx, util.BackoffConfig{ + MinBackoff: 1 * time.Second, + MaxBackoff: 1 * time.Minute, + }) + + // Ensure the context used by the Watch is always cancelled. + watchCtx, cancel := context.WithCancel(ctx) + defer cancel() + +outer: + for backoff.Ongoing() { + for resp := range c.cli.Watch(watchCtx, key, clientv3.WithPrefix()) { + if err := resp.Err(); err != nil { + level.Error(util_log.Logger).Log("msg", "watch error", "key", key, "err", err) + continue outer + } + + backoff.Reset() + + for _, event := range resp.Events { + if event.Kv.Version == 0 && event.Kv.Value == nil { + // Delete notification. Since not all KV store clients (and Cortex codecs) support this, we ignore it. + continue + } + + out, err := c.codec.Decode(event.Kv.Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding key", "key", key, "err", err) + continue + } + + if !f(string(event.Kv.Key), out) { + return + } + } + } + } +} + +// List implements kv.Client. +func (c *Client) List(ctx context.Context, prefix string) ([]string, error) { + resp, err := c.cli.Get(ctx, prefix, clientv3.WithPrefix(), clientv3.WithKeysOnly()) + if err != nil { + return nil, err + } + keys := make([]string, 0, len(resp.Kvs)) + for _, kv := range resp.Kvs { + keys = append(keys, string(kv.Key)) + } + return keys, nil +} + +// Get implements kv.Client. +func (c *Client) Get(ctx context.Context, key string) (interface{}, error) { + resp, err := c.cli.Get(ctx, key) + if err != nil { + return nil, err + } + if len(resp.Kvs) == 0 { + return nil, nil + } else if len(resp.Kvs) != 1 { + return nil, fmt.Errorf("got %d kvs, expected 1 or 0", len(resp.Kvs)) + } + return c.codec.Decode(resp.Kvs[0].Value) +} + +// Delete implements kv.Client. +func (c *Client) Delete(ctx context.Context, key string) error { + _, err := c.cli.Delete(ctx, key) + return err +} diff --git a/pkg/metrics/instance/configstore/kv/etcd/mock.go b/pkg/metrics/instance/configstore/kv/etcd/mock.go new file mode 100644 index 000000000000..b5322a03fa9b --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/etcd/mock.go @@ -0,0 +1,82 @@ +package etcd + +import ( + "flag" + "fmt" + "io" + "io/ioutil" + "net/url" + "time" + + "github.com/cortexproject/cortex/pkg/util/flagext" + + "go.etcd.io/etcd/server/v3/embed" + "go.etcd.io/etcd/server/v3/etcdserver/api/v3client" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" +) + +const etcdStartTimeout = 30 * time.Second + +// Mock returns a Mock Etcd client. +// Inspired by https://github.com/ligato/cn-infra/blob/master/db/keyval/etcd/mocks/embeded_etcd.go. +func Mock(codec codec.Codec) (*Client, io.Closer, error) { + dir, err := ioutil.TempDir("", "etcd") + if err != nil { + return nil, nil, err + } + + cfg := embed.NewConfig() + cfg.Logger = "zap" + cfg.Dir = dir + lpurl, _ := url.Parse("http://localhost:0") + lcurl, _ := url.Parse("http://localhost:0") + cfg.LPUrls = []url.URL{*lpurl} + cfg.LCUrls = []url.URL{*lcurl} + + etcd, err := embed.StartEtcd(cfg) + if err != nil { + return nil, nil, err + } + + select { + case <-etcd.Server.ReadyNotify(): + case <-time.After(etcdStartTimeout): + etcd.Server.Stop() // trigger a shutdown + return nil, nil, fmt.Errorf("server took too long to start") + } + + closer := CloserFunc(func() error { + etcd.Server.Stop() + return nil + }) + + var config Config + flagext.DefaultValues(&config) + + client := &Client{ + cfg: config, + codec: codec, + cli: v3client.New(etcd.Server), + } + + return client, closer, nil +} + +// CloserFunc is like http.HandlerFunc but for io.Closers. +type CloserFunc func() error + +// Close implements io.Closer. +func (f CloserFunc) Close() error { + return f() +} + +// NopCloser does nothing. +var NopCloser = CloserFunc(func() error { + return nil +}) + +// RegisterFlags adds the flags required to config this to the given FlagSet. +func (cfg *Config) RegisterFlags(f *flag.FlagSet) { + cfg.RegisterFlagsWithPrefix(f, "") +} diff --git a/pkg/metrics/instance/configstore/kv/kv_test.go b/pkg/metrics/instance/configstore/kv/kv_test.go new file mode 100644 index 000000000000..b58c5b701156 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/kv_test.go @@ -0,0 +1,276 @@ +package kv + +import ( + "context" + "fmt" + "io" + "sort" + "strconv" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/cortexproject/cortex/pkg/ring/kv/codec" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/consul" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/etcd" +) + +func withFixtures(t *testing.T, f func(*testing.T, Client)) { + for _, fixture := range []struct { + name string + factory func() (Client, io.Closer, error) + }{ + {"consul", func() (Client, io.Closer, error) { + return consul.NewInMemoryClient(codec.String{}), etcd.NopCloser, nil + }}, + {"etcd", func() (Client, io.Closer, error) { + return etcd.Mock(codec.String{}) + }}, + } { + t.Run(fixture.name, func(t *testing.T) { + client, closer, err := fixture.factory() + require.NoError(t, err) + defer closer.Close() + f(t, client) + }) + } +} + +var ( + ctx = context.Background() + key = "/key" +) + +func TestCAS(t *testing.T) { + withFixtures(t, func(t *testing.T, client Client) { + // Blindly set key to "0". + err := client.CAS(ctx, key, func(in interface{}) (interface{}, bool, error) { + return "0", true, nil + }) + require.NoError(t, err) + + // Swap key to i+1 iff its i. + for i := 0; i < 10; i++ { + err = client.CAS(ctx, key, func(in interface{}) (interface{}, bool, error) { + require.EqualValues(t, strconv.Itoa(i), in) + return strconv.Itoa(i + 1), true, nil + }) + require.NoError(t, err) + } + + // Make sure the CASes left the right value - "10". + value, err := client.Get(ctx, key) + require.NoError(t, err) + require.EqualValues(t, "10", value) + }) +} + +// TestNilCAS ensures we can return nil from the CAS callback when we don't +// want to modify the value. +func TestNilCAS(t *testing.T) { + withFixtures(t, func(t *testing.T, client Client) { + // Blindly set key to "0". + err := client.CAS(ctx, key, func(in interface{}) (interface{}, bool, error) { + return "0", true, nil + }) + require.NoError(t, err) + + // Ensure key is "0" and don't set it. + err = client.CAS(ctx, key, func(in interface{}) (interface{}, bool, error) { + require.EqualValues(t, "0", in) + return nil, false, nil + }) + require.NoError(t, err) + + // Make sure value is still 0. + value, err := client.Get(ctx, key) + require.NoError(t, err) + require.EqualValues(t, "0", value) + }) +} + +func TestWatchKey(t *testing.T) { + const key = "test" + const max = 100 + const sleep = 15 * time.Millisecond + const totalTestTimeout = 3 * max * sleep + const expectedFactor = 0.75 // we may not see every single value + + withFixtures(t, func(t *testing.T, client Client) { + observedValuesCh := make(chan string, max) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + // Start watching before we even start generating values. + // Values will be buffered in the channel. + client.WatchKey(ctx, key, func(value interface{}) bool { + observedValuesCh <- value.(string) + return true + }) + }() + + // update value for the key + go func() { + for i := 0; i < max; i++ { + // Start with sleeping, so that watching client see empty KV store at the beginning. + time.Sleep(sleep) + + err := client.CAS(ctx, key, func(in interface{}) (out interface{}, retry bool, err error) { + return fmt.Sprintf("%d", i), true, nil + }) + + if ctx.Err() != nil { + break + } + require.NoError(t, err) + } + }() + + lastObservedValue := -1 + observedCount := 0 + + totalDeadline := time.After(totalTestTimeout) + + for watching := true; watching; { + select { + case <-totalDeadline: + watching = false + case valStr := <-observedValuesCh: + val, err := strconv.Atoi(valStr) + if err != nil { + t.Fatal("Unexpected value observed:", valStr) + } + + if val <= lastObservedValue { + t.Fatal("Unexpected value observed:", val, "previous:", lastObservedValue) + } + lastObservedValue = val + observedCount++ + + if observedCount >= expectedFactor*max { + watching = false + } + } + } + + if observedCount < expectedFactor*max { + t.Errorf("expected at least %.0f%% observed values, got %.0f%% (observed count: %d)", 100*expectedFactor, 100*float64(observedCount)/max, observedCount) + } + }) +} + +func TestWatchPrefix(t *testing.T) { + withFixtures(t, func(t *testing.T, client Client) { + const prefix = "test/" + const prefix2 = "ignore/" + + // We are going to generate this number of updates, sleeping between each update. + const max = 100 + const sleep = time.Millisecond * 10 + // etcd seems to be quite slow. If we finish faster, test will end sooner. + // (We regularly see generators taking up to 5 seconds to produce all messages on some platforms!) + const totalTestTimeout = 10 * time.Second + + observedKeysCh := make(chan string, max) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + wg := sync.WaitGroup{} + + wg.Add(1) + go func() { + defer wg.Done() + + // start watching before we even start generating values. values will be buffered + client.WatchPrefix(ctx, prefix, func(key string, val interface{}) bool { + observedKeysCh <- key + return true + }) + }() + + gen := func(p string) { + defer wg.Done() + + start := time.Now() + for i := 0; i < max && ctx.Err() == nil; i++ { + // Start with sleeping, so that watching client can see empty KV store at the beginning. + time.Sleep(sleep) + + key := fmt.Sprintf("%s%d", p, i) + err := client.CAS(ctx, key, func(in interface{}) (out interface{}, retry bool, err error) { + return key, true, nil + }) + + if ctx.Err() != nil { + break + } + require.NoError(t, err) + } + t.Log("Generator finished in", time.Since(start)) + } + + wg.Add(2) + go gen(prefix) + go gen(prefix2) // we don't want to see these keys reported + + observedKeys := map[string]int{} + + totalDeadline := time.After(totalTestTimeout) + + start := time.Now() + for watching := true; watching; { + select { + case <-totalDeadline: + watching = false + case key := <-observedKeysCh: + observedKeys[key]++ + if len(observedKeys) == max { + watching = false + } + } + } + t.Log("Watching finished in", time.Since(start)) + + // Stop all goroutines and wait until terminated. + cancel() + wg.Wait() + + // verify that each key was reported once, and keys outside prefix were not reported + for i := 0; i < max; i++ { + key := fmt.Sprintf("%s%d", prefix, i) + + if observedKeys[key] != 1 { + t.Errorf("key %s has incorrect value %d", key, observedKeys[key]) + } + delete(observedKeys, key) + } + + if len(observedKeys) > 0 { + t.Errorf("unexpected keys reported: %v", observedKeys) + } + }) +} + +// TestList makes sure stored keys are listed back. +func TestList(t *testing.T) { + keysToCreate := []string{"a", "b", "c"} + + withFixtures(t, func(t *testing.T, client Client) { + for _, key := range keysToCreate { + err := client.CAS(context.Background(), key, func(in interface{}) (out interface{}, retry bool, err error) { + return key, false, nil + }) + require.NoError(t, err) + } + + storedKeys, err := client.List(context.Background(), "") + require.NoError(t, err) + sort.Strings(storedKeys) + + require.Equal(t, keysToCreate, storedKeys) + }) +} diff --git a/pkg/metrics/instance/configstore/kv/metrics.go b/pkg/metrics/instance/configstore/kv/metrics.go new file mode 100644 index 000000000000..ac29f1de650f --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/metrics.go @@ -0,0 +1,109 @@ +package kv + +import ( + "context" + "strconv" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/weaveworks/common/httpgrpc" + "github.com/weaveworks/common/instrument" +) + +// RegistererWithKVName wraps the provided Registerer with the KV name label. If a nil reg +// is provided, a nil registry is returned +func RegistererWithKVName(reg prometheus.Registerer, name string) prometheus.Registerer { + if reg == nil { + return nil + } + + return prometheus.WrapRegistererWith(prometheus.Labels{"kv_name": name}, reg) +} + +// getCasErrorCode converts the provided CAS error into the code that should be used to track the operation +// in metrics. +func getCasErrorCode(err error) string { + if err == nil { + return "200" + } + if resp, ok := httpgrpc.HTTPResponseFromError(err); ok { + return strconv.Itoa(int(resp.GetCode())) + } + + // If the error has been returned to abort the CAS operation, then we shouldn't + // consider it an error when tracking metrics. + if casErr, ok := err.(interface{ IsOperationAborted() bool }); ok && casErr.IsOperationAborted() { + return "200" + } + + return "500" +} + +type metrics struct { + c Client + requestDuration *instrument.HistogramCollector +} + +func newMetricsClient(backend string, c Client, reg prometheus.Registerer) Client { + return &metrics{ + c: c, + requestDuration: instrument.NewHistogramCollector( + promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "agent", + Name: "kv_request_duration_seconds", + Help: "Time spent on kv store requests.", + Buckets: prometheus.DefBuckets, + ConstLabels: prometheus.Labels{ + "type": backend, + }, + }, []string{"operation", "status_code"}), + ), + } +} + +func (m metrics) List(ctx context.Context, prefix string) ([]string, error) { + var result []string + err := instrument.CollectedRequest(ctx, "List", m.requestDuration, instrument.ErrorCode, func(ctx context.Context) error { + var err error + result, err = m.c.List(ctx, prefix) + return err + }) + return result, err +} + +func (m metrics) Get(ctx context.Context, key string) (interface{}, error) { + var result interface{} + err := instrument.CollectedRequest(ctx, "GET", m.requestDuration, instrument.ErrorCode, func(ctx context.Context) error { + var err error + result, err = m.c.Get(ctx, key) + return err + }) + return result, err +} + +func (m metrics) Delete(ctx context.Context, key string) error { + err := instrument.CollectedRequest(ctx, "Delete", m.requestDuration, instrument.ErrorCode, func(ctx context.Context) error { + return m.c.Delete(ctx, key) + }) + return err +} + +func (m metrics) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + return instrument.CollectedRequest(ctx, "CAS", m.requestDuration, getCasErrorCode, func(ctx context.Context) error { + return m.c.CAS(ctx, key, f) + }) +} + +func (m metrics) WatchKey(ctx context.Context, key string, f func(interface{}) bool) { + _ = instrument.CollectedRequest(ctx, "WatchKey", m.requestDuration, instrument.ErrorCode, func(ctx context.Context) error { + m.c.WatchKey(ctx, key, f) + return nil + }) +} + +func (m metrics) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) { + _ = instrument.CollectedRequest(ctx, "WatchPrefix", m.requestDuration, instrument.ErrorCode, func(ctx context.Context) error { + m.c.WatchPrefix(ctx, prefix, f) + return nil + }) +} diff --git a/pkg/metrics/instance/configstore/kv/mock.go b/pkg/metrics/instance/configstore/kv/mock.go new file mode 100644 index 000000000000..ac7ae011df71 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/mock.go @@ -0,0 +1,40 @@ +package kv + +import ( + "context" + + "github.com/go-kit/kit/log/level" + + util_log "github.com/cortexproject/cortex/pkg/util/log" +) + +// The mockClient does not anything. +// This is used for testing only. +type mockClient struct{} + +func buildMockClient() (Client, error) { + level.Warn(util_log.Logger).Log("msg", "created mockClient for testing only") + return mockClient{}, nil +} + +func (m mockClient) List(ctx context.Context, prefix string) ([]string, error) { + return []string{}, nil +} + +func (m mockClient) Get(ctx context.Context, key string) (interface{}, error) { + return "", nil +} + +func (m mockClient) Delete(ctx context.Context, key string) error { + return nil +} + +func (m mockClient) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + return nil +} + +func (m mockClient) WatchKey(ctx context.Context, key string, f func(interface{}) bool) { +} + +func (m mockClient) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) { +} diff --git a/pkg/metrics/instance/configstore/kv/multi.go b/pkg/metrics/instance/configstore/kv/multi.go new file mode 100644 index 000000000000..9d37e03d2c02 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/multi.go @@ -0,0 +1,365 @@ +package kv + +import ( + "context" + "flag" + "fmt" + "sync" + "time" + + "github.com/go-kit/kit/log" + "github.com/prometheus/client_golang/prometheus" + "go.uber.org/atomic" + + util_log "github.com/cortexproject/cortex/pkg/util/log" + + "github.com/go-kit/kit/log/level" +) + +var ( + primaryStoreGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "agent_multikv_primary_store", + Help: "Selected primary KV store", + }, []string{"store"}) + + mirrorEnabledGauge = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "agent_multikv_mirror_enabled", + Help: "Is mirroring to secondary store enabled", + }) + + mirrorWritesCounter = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "agent_multikv_mirror_writes_total", + Help: "Number of mirror-writes to secondary store", + }) + + mirrorFailuresCounter = prometheus.NewCounter(prometheus.CounterOpts{ + Name: "agent_multikv_mirror_write_errors_total", + Help: "Number of failures to mirror-write to secondary store", + }) +) + +func init() { + prometheus.MustRegister(primaryStoreGauge, mirrorEnabledGauge, mirrorWritesCounter, mirrorFailuresCounter) +} + +// MultiConfig is a configuration for MultiClient. +type MultiConfig struct { + Primary string `yaml:"primary"` + Secondary string `yaml:"secondary"` + + MirrorEnabled bool `yaml:"mirror_enabled"` + MirrorTimeout time.Duration `yaml:"mirror_timeout"` + + // ConfigProvider returns channel with MultiRuntimeConfig updates. + ConfigProvider func() <-chan MultiRuntimeConfig `yaml:"-"` +} + +// RegisterFlagsWithPrefix registers flags with prefix. +func (cfg *MultiConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) { + f.StringVar(&cfg.Primary, prefix+"multi.primary", "", "Primary backend storage used by multi-client.") + f.StringVar(&cfg.Secondary, prefix+"multi.secondary", "", "Secondary backend storage used by multi-client.") + f.BoolVar(&cfg.MirrorEnabled, prefix+"multi.mirror-enabled", false, "Mirror writes to secondary store.") + f.DurationVar(&cfg.MirrorTimeout, prefix+"multi.mirror-timeout", 2*time.Second, "Timeout for storing value to secondary store.") +} + +// MultiRuntimeConfig has values that can change in runtime (via overrides) +type MultiRuntimeConfig struct { + // Primary store used by MultiClient. Can be updated in runtime to switch to + // a different store (eg. consul -> etcd). Doing this allows nice migration + // between stores. Empty values are ignored. + PrimaryStore string `yaml:"primary"` + + // Mirroring enabled or not. Nil = no change. + Mirroring *bool `yaml:"mirror_enabled"` +} + +type kvclient struct { + client Client + name string +} + +type clientInProgress struct { + client int + cancel context.CancelFunc +} + +// MultiClient implements kv.Client by forwarding all API calls to primary client. +// Writes performed via CAS method are also (optionally) forwarded to secondary clients. +type MultiClient struct { + // Available KV clients + clients []kvclient + + mirrorTimeout time.Duration + mirroringEnabled *atomic.Bool + + // logger with "multikv" component + logger log.Logger + + // The primary client used for interaction. + primaryID *atomic.Int32 + + cancel context.CancelFunc + + inProgressMu sync.Mutex + // Cancel functions for ongoing operations. key is a value from inProgressCnt. + // What we really need is a []context.CancelFunc, but functions cannot be compared against each other using ==, + // so we use this map instead. + inProgress map[int]clientInProgress + inProgressCnt int +} + +// NewMultiClient creates new MultiClient with given KV Clients. +// First client in the slice is the primary client. +func NewMultiClient(cfg MultiConfig, clients []kvclient) *MultiClient { + c := &MultiClient{ + clients: clients, + primaryID: atomic.NewInt32(0), + inProgress: map[int]clientInProgress{}, + + mirrorTimeout: cfg.MirrorTimeout, + mirroringEnabled: atomic.NewBool(cfg.MirrorEnabled), + + logger: log.With(util_log.Logger, "component", "multikv"), + } + + ctx, cancelFn := context.WithCancel(context.Background()) + c.cancel = cancelFn + + if cfg.ConfigProvider != nil { + go c.watchConfigChannel(ctx, cfg.ConfigProvider()) + } + + c.updatePrimaryStoreGauge() + c.updateMirrorEnabledGauge() + return c +} + +func (m *MultiClient) watchConfigChannel(ctx context.Context, configChannel <-chan MultiRuntimeConfig) { + for { + select { + case cfg, ok := <-configChannel: + if !ok { + return + } + + if cfg.Mirroring != nil { + enabled := *cfg.Mirroring + old := m.mirroringEnabled.Swap(enabled) + if old != enabled { + level.Info(m.logger).Log("msg", "toggled mirroring", "enabled", enabled) + } + m.updateMirrorEnabledGauge() + } + + if cfg.PrimaryStore != "" { + switched, err := m.setNewPrimaryClient(cfg.PrimaryStore) + if switched { + level.Info(m.logger).Log("msg", "switched primary KV store", "primary", cfg.PrimaryStore) + } + if err != nil { + level.Error(m.logger).Log("msg", "failed to switch primary KV store", "primary", cfg.PrimaryStore, "err", err) + } + } + + case <-ctx.Done(): + return + } + } +} + +func (m *MultiClient) getPrimaryClient() (int, kvclient) { + v := m.primaryID.Load() + return int(v), m.clients[v] +} + +// returns true, if primary client has changed +func (m *MultiClient) setNewPrimaryClient(store string) (bool, error) { + newPrimaryIx := -1 + for ix, c := range m.clients { + if c.name == store { + newPrimaryIx = ix + break + } + } + + if newPrimaryIx < 0 { + return false, fmt.Errorf("KV store not found") + } + + prev := int(m.primaryID.Swap(int32(newPrimaryIx))) + if prev == newPrimaryIx { + return false, nil + } + + defer m.updatePrimaryStoreGauge() // do as the last thing, after releasing the lock + + // switching to new primary... cancel clients using previous one + m.inProgressMu.Lock() + defer m.inProgressMu.Unlock() + + for _, inp := range m.inProgress { + if inp.client == prev { + inp.cancel() + } + } + return true, nil +} + +func (m *MultiClient) updatePrimaryStoreGauge() { + _, pkv := m.getPrimaryClient() + + for _, kv := range m.clients { + value := float64(0) + if pkv == kv { + value = 1 + } + + primaryStoreGauge.WithLabelValues(kv.name).Set(value) + } +} + +func (m *MultiClient) updateMirrorEnabledGauge() { + if m.mirroringEnabled.Load() { + mirrorEnabledGauge.Set(1) + } else { + mirrorEnabledGauge.Set(0) + } +} + +func (m *MultiClient) registerCancelFn(clientID int, fn context.CancelFunc) int { + m.inProgressMu.Lock() + defer m.inProgressMu.Unlock() + + m.inProgressCnt++ + id := m.inProgressCnt + m.inProgress[id] = clientInProgress{client: clientID, cancel: fn} + return id +} + +func (m *MultiClient) unregisterCancelFn(id int) { + m.inProgressMu.Lock() + defer m.inProgressMu.Unlock() + + delete(m.inProgress, id) +} + +// Runs supplied fn with current primary client. If primary client changes, fn is restarted. +// When fn finishes (with or without error), this method returns given error value. +func (m *MultiClient) runWithPrimaryClient(origCtx context.Context, fn func(newCtx context.Context, primary kvclient) error) error { + cancelFn := context.CancelFunc(nil) + cancelFnID := 0 + + cleanup := func() { + if cancelFn != nil { + cancelFn() + } + if cancelFnID > 0 { + m.unregisterCancelFn(cancelFnID) + } + } + + defer cleanup() + + // This only loops if switchover to a new primary backend happens while calling 'fn', which is very rare. + for { + cleanup() + pid, kv := m.getPrimaryClient() + + var cancelCtx context.Context + cancelCtx, cancelFn = context.WithCancel(origCtx) + cancelFnID = m.registerCancelFn(pid, cancelFn) + + err := fn(cancelCtx, kv) + + if err == nil { + return nil + } + + if cancelCtx.Err() == context.Canceled && origCtx.Err() == nil { + // our context was cancelled, but outer context is not done yet. retry + continue + } + + return err + } +} + +// List is a part of the kv.Client interface. +func (m *MultiClient) List(ctx context.Context, prefix string) ([]string, error) { + _, kv := m.getPrimaryClient() + return kv.client.List(ctx, prefix) +} + +// Get is a part of kv.Client interface. +func (m *MultiClient) Get(ctx context.Context, key string) (interface{}, error) { + _, kv := m.getPrimaryClient() + return kv.client.Get(ctx, key) +} + +// Delete is a part of the kv.Client interface. +func (m *MultiClient) Delete(ctx context.Context, key string) error { + _, kv := m.getPrimaryClient() + return kv.client.Delete(ctx, key) +} + +// CAS is a part of kv.Client interface. +func (m *MultiClient) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + _, kv := m.getPrimaryClient() + + updatedValue := interface{}(nil) + err := kv.client.CAS(ctx, key, func(in interface{}) (interface{}, bool, error) { + out, retry, err := f(in) + updatedValue = out + return out, retry, err + }) + + if err == nil && updatedValue != nil && m.mirroringEnabled.Load() { + m.writeToSecondary(ctx, kv, key, updatedValue) + } + + return err +} + +// WatchKey is a part of kv.Client interface. +func (m *MultiClient) WatchKey(ctx context.Context, key string, f func(interface{}) bool) { + _ = m.runWithPrimaryClient(ctx, func(newCtx context.Context, primary kvclient) error { + primary.client.WatchKey(newCtx, key, f) + return newCtx.Err() + }) +} + +// WatchPrefix is a part of kv.Client interface. +func (m *MultiClient) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) { + _ = m.runWithPrimaryClient(ctx, func(newCtx context.Context, primary kvclient) error { + primary.client.WatchPrefix(newCtx, prefix, f) + return newCtx.Err() + }) +} + +func (m *MultiClient) writeToSecondary(ctx context.Context, primary kvclient, key string, newValue interface{}) { + if m.mirrorTimeout > 0 { + var cfn context.CancelFunc + ctx, cfn = context.WithTimeout(ctx, m.mirrorTimeout) + defer cfn() + } + + // let's propagate new value to all remaining clients + for _, kvc := range m.clients { + if kvc == primary { + continue + } + + mirrorWritesCounter.Inc() + err := kvc.client.CAS(ctx, key, func(in interface{}) (out interface{}, retry bool, err error) { + // try once + return newValue, false, nil + }) + + if err != nil { + mirrorFailuresCounter.Inc() + level.Warn(m.logger).Log("msg", "failed to update value in secondary store", "key", key, "err", err, "primary", primary.name, "secondary", kvc.name) + } else { + level.Debug(m.logger).Log("msg", "stored updated value to secondary store", "key", key, "primary", primary.name, "secondary", kvc.name) + } + } +} diff --git a/pkg/metrics/instance/configstore/kv/multi_test.go b/pkg/metrics/instance/configstore/kv/multi_test.go new file mode 100644 index 000000000000..9385ed5ff8e3 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/multi_test.go @@ -0,0 +1,32 @@ +package kv + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "gopkg.in/yaml.v2" +) + +func boolPtr(b bool) *bool { + return &b +} + +func TestMultiRuntimeConfigWithVariousEnabledValues(t *testing.T) { + testcases := map[string]struct { + yaml string + expected *bool + }{ + "nil": {"primary: test", nil}, + "true": {"primary: test\nmirror_enabled: true", boolPtr(true)}, + "false": {"mirror_enabled: false", boolPtr(false)}, + } + + for name, tc := range testcases { + t.Run(name, func(t *testing.T) { + c := MultiRuntimeConfig{} + err := yaml.Unmarshal([]byte(tc.yaml), &c) + assert.NoError(t, err, tc.yaml) + assert.Equal(t, tc.expected, c.Mirroring, tc.yaml) + }) + } +} diff --git a/pkg/metrics/instance/configstore/kv/prefix.go b/pkg/metrics/instance/configstore/kv/prefix.go new file mode 100644 index 000000000000..5775cf17b947 --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/prefix.go @@ -0,0 +1,63 @@ +package kv + +import ( + "context" + "fmt" + "strings" +) + +type prefixedKVClient struct { + prefix string + client Client +} + +// PrefixClient takes a KVClient and forces a prefix on all its operations. +func PrefixClient(client Client, prefix string) Client { + return &prefixedKVClient{prefix, client} +} + +// List returns a list of keys under a given prefix. +func (c *prefixedKVClient) List(ctx context.Context, prefix string) ([]string, error) { + keys, err := c.client.List(ctx, c.prefix+prefix) + if err != nil { + return nil, err + } + + // Remove the prefix from the returned key. The prefix attached to the + // prefixed client is supposed to be transparent and the values returned + // by List should be able to be immediately inserted into the Get + // function, which means that our injected prefix needs to be removed. + for i := range keys { + keys[i] = strings.TrimPrefix(keys[i], c.prefix) + } + + return keys, nil +} + +// CAS atomically modifies a value in a callback. If the value doesn't exist, +// you'll get 'nil' as an argument to your callback. +func (c *prefixedKVClient) CAS(ctx context.Context, key string, f func(in interface{}) (out interface{}, retry bool, err error)) error { + return c.client.CAS(ctx, c.prefix+key, f) +} + +// WatchKey watches a key. +func (c *prefixedKVClient) WatchKey(ctx context.Context, key string, f func(interface{}) bool) { + c.client.WatchKey(ctx, c.prefix+key, f) +} + +// WatchPrefix watches a prefix. For a prefix client it appends the prefix argument to the clients prefix. +func (c *prefixedKVClient) WatchPrefix(ctx context.Context, prefix string, f func(string, interface{}) bool) { + c.client.WatchPrefix(ctx, fmt.Sprintf("%s%s", c.prefix, prefix), func(k string, i interface{}) bool { + return f(strings.TrimPrefix(k, c.prefix), i) + }) +} + +// Get looks up a given object from its key. +func (c *prefixedKVClient) Get(ctx context.Context, key string) (interface{}, error) { + return c.client.Get(ctx, c.prefix+key) +} + +// Delete removes a given object from its key. +func (c *prefixedKVClient) Delete(ctx context.Context, key string) error { + return c.client.Delete(ctx, c.prefix+key) +} diff --git a/pkg/metrics/instance/configstore/remote.go b/pkg/metrics/instance/configstore/remote.go index f5d5e3dd4ef5..3e808fd1e1c0 100644 --- a/pkg/metrics/instance/configstore/remote.go +++ b/pkg/metrics/instance/configstore/remote.go @@ -14,10 +14,10 @@ import ( "github.com/hashicorp/consul/api" - "github.com/cortexproject/cortex/pkg/ring/kv" "github.com/go-kit/kit/log" "github.com/go-kit/kit/log/level" "github.com/grafana/agent/pkg/metrics/instance" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv" "github.com/grafana/agent/pkg/util" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" diff --git a/pkg/metrics/instance/configstore/remote_test.go b/pkg/metrics/instance/configstore/remote_test.go index f9c1f583ed39..dbfa79b62267 100644 --- a/pkg/metrics/instance/configstore/remote_test.go +++ b/pkg/metrics/instance/configstore/remote_test.go @@ -8,9 +8,9 @@ import ( "testing" "time" - "github.com/cortexproject/cortex/pkg/ring/kv" "github.com/go-kit/kit/log" "github.com/grafana/agent/pkg/metrics/instance" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv" "github.com/grafana/agent/pkg/util" "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/require"