diff --git a/CHANGELOG.md b/CHANGELOG.md index 90e1ef2e072a..5b4f3279eb65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,6 +72,12 @@ for specific instructions. - [CHANGE] Breaking change: `prom_instance` in the spanmetrics config is now named `metrics_instance`. (@rfratto) +- [CHANGE] Scraping service: metrics for the config store have been changed to + `agent_kv_request_duration_seconds`. The temporary + `agent_configstore_consul_request_duration_seconds` metric has been removed + and will roll up into `agent_kv_request_duration_seconds` now with the + appropriate List operation. + - [DEPRECATION] The `loki` key at the root of the config file has been deprecated in favor of `logs`. `loki`-named fields in `automatic_logging` have been renamed accordinly: `loki_name` is now `logs_instance_name`, diff --git a/pkg/metrics/instance/configstore/kv/client.go b/pkg/metrics/instance/configstore/kv/client.go index 229b21741499..90f024df2b6c 100644 --- a/pkg/metrics/instance/configstore/kv/client.go +++ b/pkg/metrics/instance/configstore/kv/client.go @@ -11,6 +11,7 @@ import ( "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" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" ) const ( @@ -89,7 +90,7 @@ func (cfg *Config) RegisterFlagsWithPrefix(flagsPrefix, defaultPrefix string, f 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) + List(ctx context.Context, prefix string) ([]pair.KVP, 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. diff --git a/pkg/metrics/instance/configstore/kv/consul/client.go b/pkg/metrics/instance/configstore/kv/consul/client.go index 1c39a473ced7..7d48c8ce00b7 100644 --- a/pkg/metrics/instance/configstore/kv/consul/client.go +++ b/pkg/metrics/instance/configstore/kv/consul/client.go @@ -10,6 +10,7 @@ import ( "time" "github.com/go-kit/kit/log/level" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" consul "github.com/hashicorp/consul/api" "github.com/hashicorp/go-cleanhttp" "github.com/weaveworks/common/instrument" @@ -323,7 +324,7 @@ func (c *Client) WatchPrefix(ctx context.Context, prefix string, f func(string, } // List implements kv.List. -func (c *Client) List(ctx context.Context, prefix string) ([]string, error) { +func (c *Client) List(ctx context.Context, prefix string) ([]pair.KVP, error) { options := &consul.QueryOptions{ AllowStale: !c.cfg.ConsistentReads, RequireConsistent: c.cfg.ConsistentReads, @@ -333,11 +334,20 @@ func (c *Client) List(ctx context.Context, prefix string) ([]string, error) { return nil, err } - keys := make([]string, 0, len(pairs)) + res := make([]pair.KVP, 0, len(pairs)) for _, kvp := range pairs { - keys = append(keys, kvp.Key) + value, 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 + } + + res = append(res, pair.KVP{ + Key: kvp.Key, + Value: value, + }) } - return keys, nil + return res, nil } // Get implements kv.Get. diff --git a/pkg/metrics/instance/configstore/kv/etcd/etcd.go b/pkg/metrics/instance/configstore/kv/etcd/etcd.go index d981c00320ad..0ca3fe8c2ed0 100644 --- a/pkg/metrics/instance/configstore/kv/etcd/etcd.go +++ b/pkg/metrics/instance/configstore/kv/etcd/etcd.go @@ -8,6 +8,7 @@ import ( "time" "github.com/go-kit/kit/log/level" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" "github.com/pkg/errors" clientv3 "go.etcd.io/etcd/client/v3" "go.etcd.io/etcd/pkg/transport" @@ -247,16 +248,26 @@ outer: } // 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()) +func (c *Client) List(ctx context.Context, prefix string) ([]pair.KVP, error) { + resp, err := c.cli.Get(ctx, prefix, clientv3.WithPrefix()) if err != nil { return nil, err } - keys := make([]string, 0, len(resp.Kvs)) + + res := make([]pair.KVP, 0, len(resp.Kvs)) for _, kv := range resp.Kvs { - keys = append(keys, string(kv.Key)) + out, err := c.codec.Decode(kv.Value) + if err != nil { + level.Error(util_log.Logger).Log("msg", "error decoding key", "key", string(kv.Key), "err", err) + continue + } + + res = append(res, pair.KVP{ + Key: string(kv.Key), + Value: out, + }) } - return keys, nil + return res, nil } // Get implements kv.Client. diff --git a/pkg/metrics/instance/configstore/kv/kv_test.go b/pkg/metrics/instance/configstore/kv/kv_test.go index b58c5b701156..8e43242b0a00 100644 --- a/pkg/metrics/instance/configstore/kv/kv_test.go +++ b/pkg/metrics/instance/configstore/kv/kv_test.go @@ -15,6 +15,7 @@ import ( "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" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" ) func withFixtures(t *testing.T, f func(*testing.T, Client)) { @@ -28,6 +29,14 @@ func withFixtures(t *testing.T, f func(*testing.T, Client)) { {"etcd", func() (Client, io.Closer, error) { return etcd.Mock(codec.String{}) }}, + {"prefixed/etcd", func() (cli Client, closer io.Closer, err error) { + cli, closer, err = etcd.Mock(codec.String{}) + if err != nil { + return + } + cli = PrefixClient(cli, "prefix/") + return + }}, } { t.Run(fixture.name, func(t *testing.T) { client, closer, err := fixture.factory() @@ -94,7 +103,7 @@ func TestNilCAS(t *testing.T) { func TestWatchKey(t *testing.T) { const key = "test" const max = 100 - const sleep = 15 * time.Millisecond + const sleep = 50 * time.Millisecond const totalTestTimeout = 3 * max * sleep const expectedFactor = 0.75 // we may not see every single value @@ -257,20 +266,24 @@ func TestWatchPrefix(t *testing.T) { // TestList makes sure stored keys are listed back. func TestList(t *testing.T) { - keysToCreate := []string{"a", "b", "c"} + kvpsToCreate := []pair.KVP{ + {Key: "a", Value: "value_a"}, + {Key: "b", Value: "value_b"}, + {Key: "c", Value: "value_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 + for _, kvp := range kvpsToCreate { + err := client.CAS(context.Background(), kvp.Key, func(in interface{}) (out interface{}, retry bool, err error) { + return kvp.Value, false, nil }) require.NoError(t, err) } - storedKeys, err := client.List(context.Background(), "") + storedKVPs, err := client.List(context.Background(), "") require.NoError(t, err) - sort.Strings(storedKeys) + sort.Slice(storedKVPs, func(i, j int) bool { return storedKVPs[i].Key < storedKVPs[j].Key }) - require.Equal(t, keysToCreate, storedKeys) + require.Equal(t, kvpsToCreate, storedKVPs) }) } diff --git a/pkg/metrics/instance/configstore/kv/metrics.go b/pkg/metrics/instance/configstore/kv/metrics.go index ac29f1de650f..ea188c96e89f 100644 --- a/pkg/metrics/instance/configstore/kv/metrics.go +++ b/pkg/metrics/instance/configstore/kv/metrics.go @@ -4,6 +4,7 @@ import ( "context" "strconv" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "github.com/weaveworks/common/httpgrpc" @@ -61,8 +62,8 @@ func newMetricsClient(backend string, c Client, reg prometheus.Registerer) Clien } } -func (m metrics) List(ctx context.Context, prefix string) ([]string, error) { - var result []string +func (m metrics) List(ctx context.Context, prefix string) ([]pair.KVP, error) { + var result []pair.KVP err := instrument.CollectedRequest(ctx, "List", m.requestDuration, instrument.ErrorCode, func(ctx context.Context) error { var err error result, err = m.c.List(ctx, prefix) diff --git a/pkg/metrics/instance/configstore/kv/mock.go b/pkg/metrics/instance/configstore/kv/mock.go index ac7ae011df71..c2f195a9dd03 100644 --- a/pkg/metrics/instance/configstore/kv/mock.go +++ b/pkg/metrics/instance/configstore/kv/mock.go @@ -4,6 +4,7 @@ import ( "context" "github.com/go-kit/kit/log/level" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" util_log "github.com/cortexproject/cortex/pkg/util/log" ) @@ -17,8 +18,8 @@ func buildMockClient() (Client, error) { return mockClient{}, nil } -func (m mockClient) List(ctx context.Context, prefix string) ([]string, error) { - return []string{}, nil +func (m mockClient) List(ctx context.Context, prefix string) ([]pair.KVP, error) { + return []pair.KVP{}, nil } func (m mockClient) Get(ctx context.Context, key string) (interface{}, error) { diff --git a/pkg/metrics/instance/configstore/kv/multi.go b/pkg/metrics/instance/configstore/kv/multi.go index 9d37e03d2c02..908796ddd049 100644 --- a/pkg/metrics/instance/configstore/kv/multi.go +++ b/pkg/metrics/instance/configstore/kv/multi.go @@ -8,6 +8,7 @@ import ( "time" "github.com/go-kit/kit/log" + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" "github.com/prometheus/client_golang/prometheus" "go.uber.org/atomic" @@ -285,7 +286,7 @@ func (m *MultiClient) runWithPrimaryClient(origCtx context.Context, fn func(newC } // List is a part of the kv.Client interface. -func (m *MultiClient) List(ctx context.Context, prefix string) ([]string, error) { +func (m *MultiClient) List(ctx context.Context, prefix string) ([]pair.KVP, error) { _, kv := m.getPrimaryClient() return kv.client.List(ctx, prefix) } diff --git a/pkg/metrics/instance/configstore/kv/pair/kvp.go b/pkg/metrics/instance/configstore/kv/pair/kvp.go new file mode 100644 index 000000000000..8063aee4996a --- /dev/null +++ b/pkg/metrics/instance/configstore/kv/pair/kvp.go @@ -0,0 +1,9 @@ +package pair + +// KVP is a Key-Value Pair for a key in a kv.Client. +type KVP struct { + Key string + + // Value should be deserialised through the Client's codec. + Value interface{} +} diff --git a/pkg/metrics/instance/configstore/kv/prefix.go b/pkg/metrics/instance/configstore/kv/prefix.go index 5775cf17b947..e2d14d664743 100644 --- a/pkg/metrics/instance/configstore/kv/prefix.go +++ b/pkg/metrics/instance/configstore/kv/prefix.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "strings" + + "github.com/grafana/agent/pkg/metrics/instance/configstore/kv/pair" ) type prefixedKVClient struct { @@ -17,8 +19,8 @@ func PrefixClient(client Client, prefix string) 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) +func (c *prefixedKVClient) List(ctx context.Context, prefix string) ([]pair.KVP, error) { + pairs, err := c.client.List(ctx, c.prefix+prefix) if err != nil { return nil, err } @@ -27,11 +29,11 @@ func (c *prefixedKVClient) List(ctx context.Context, prefix string) ([]string, e // 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) + for i := range pairs { + pairs[i].Key = strings.TrimPrefix(pairs[i].Key, c.prefix) } - return keys, nil + return pairs, nil } // CAS atomically modifies a value in a callback. If the value doesn't exist, diff --git a/pkg/metrics/instance/configstore/remote.go b/pkg/metrics/instance/configstore/remote.go index 3e808fd1e1c0..4c3044ebd0c6 100644 --- a/pkg/metrics/instance/configstore/remote.go +++ b/pkg/metrics/instance/configstore/remote.go @@ -2,39 +2,18 @@ package configstore import ( "context" - "errors" "fmt" - "net/http" "strings" "sync" - "github.com/weaveworks/common/instrument" - - "github.com/hashicorp/go-cleanhttp" - - "github.com/hashicorp/consul/api" - "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" ) -/*********************************************************************************************************************** -The consul code skipping the cortex handler is due to performance issue with a large number of configs and overloading -consul. See issue https://github.com/grafana/agent/issues/789. The long term method will be to refactor and extract -the cortex code so other stores can also benefit from this. @mattdurham -***********************************************************************************************************************/ - -var consulRequestDuration = instrument.NewHistogramCollector(promauto.NewHistogramVec(prometheus.HistogramOpts{ - Name: "agent_configstore_consul_request_duration_seconds", - Help: "Time spent on consul requests when listing configs.", - Buckets: prometheus.DefBuckets, -}, []string{"operation", "status_code"})) - // Remote loads instance files from a remote KV store. The KV store // can be swapped out in real time. type Remote struct { @@ -42,7 +21,7 @@ type Remote struct { reg *util.Unregisterer kvMut sync.RWMutex - kv *agentRemoteClient + kv kv.Client reloadKV chan struct{} cancelCtx context.Context @@ -52,14 +31,6 @@ type Remote struct { configsCh chan WatchEvent } -// agentRemoteClient is a simple wrapper to allow the shortcircuit of consul, while being backwards compatible with non -// consul kv stores -type agentRemoteClient struct { - kv.Client - consul *api.Client - config kv.Config -} - // NewRemote creates a new Remote store that uses a Key-Value client to store // and retrieve configs. If enable is true, the store will be immediately // connected to. Otherwise, it can be lazily loaded by enabling later through @@ -99,49 +70,23 @@ func (r *Remote) ApplyConfig(cfg kv.Config, enable bool) error { r.reg.UnregisterAll() if !enable { - r.setClient(nil, nil, kv.Config{}) + r.setClient(nil) return nil } cli, err := kv.NewClient(cfg, GetCodec(), kv.RegistererWithKVName(r.reg, "agent_configs")) - // This is a hack to get a consul client, the client above has it embedded but its not exposed - var consulClient *api.Client - if cfg.Store == "consul" { - consulClient, err = api.NewClient(&api.Config{ - Address: cfg.Consul.Host, - Token: cfg.Consul.ACLToken, - Scheme: "http", - HttpClient: &http.Client{ - Transport: cleanhttp.DefaultPooledTransport(), - // See https://blog.cloudflare.com/the-complete-guide-to-golang-net-http-timeouts/ - Timeout: cfg.Consul.HTTPClientTimeout, - }, - }) - if err != nil { - return err - } - - } if err != nil { return fmt.Errorf("failed to create kv client: %w", err) } - r.setClient(cli, consulClient, cfg) + r.setClient(cli) return nil } // setClient sets the active client and notifies run to restart the // kv watcher. -func (r *Remote) setClient(client kv.Client, consulClient *api.Client, config kv.Config) { - if client == nil && consulClient == nil { - r.kv = nil - } else { - r.kv = &agentRemoteClient{ - Client: client, - consul: consulClient, - config: config, - } - } +func (r *Remote) setClient(client kv.Client) { + r.kv = client r.reloadKV <- struct{}{} } @@ -174,7 +119,7 @@ Outer: } } -func (r *Remote) watchKV(ctx context.Context, client *agentRemoteClient) { +func (r *Remote) watchKV(ctx context.Context, client kv.Client) { // Edge case: client was unset, nothing to do here. if client == nil { level.Info(r.log).Log("msg", "not watching the KV, none set") @@ -214,39 +159,15 @@ func (r *Remote) List(ctx context.Context) ([]string, error) { return nil, ErrNotConnected } - return r.kv.List(ctx, "") -} - -// listConsul returns Key Value Pairs instead of []string -func (r *Remote) listConsul(ctx context.Context) (api.KVPairs, error) { - if r.kv == nil { - return nil, ErrNotConnected - } - - var pairs api.KVPairs - options := &api.QueryOptions{ - AllowStale: !r.kv.config.Consul.ConsistentReads, - RequireConsistent: r.kv.config.Consul.ConsistentReads, - } - // This is copied from cortex list so that stats stay the same - err := instrument.CollectedRequest(ctx, "List", consulRequestDuration, instrument.ErrorCode, func(ctx context.Context) error { - var err error - pairs, _, err = r.kv.consul.KV().List(r.kv.config.Prefix, options.WithContext(ctx)) - return err - }) - + kvps, err := r.kv.List(ctx, "") if err != nil { return nil, err } - // This mirrors the previous behavior of returning a blank array as opposed to nil. - if pairs == nil { - blankPairs := make(api.KVPairs, 0) - return blankPairs, nil + keys := make([]string, len(kvps)) + for i, kvp := range kvps { + keys[i] = kvp.Key } - for _, kvp := range pairs { - kvp.Key = strings.TrimPrefix(kvp.Key, r.kv.config.Prefix) - } - return pairs, nil + return keys, nil } // Get retrieves an individual config from the KV store. @@ -347,111 +268,32 @@ func (r *Remote) all(ctx context.Context, keep func(key string) bool) (<-chan in return nil, ErrNotConnected } - // If we are using a consul client then do the short circuit way, this is done so that we receive all the key value pairs - // in one call then, operate on them in memory. Previously we retrieved the list (which stripped the values) - // then ran a goroutine to get each individual value from consul. In situations with an extremely large number of - // configs this overloaded the consul instances. This reduces that to one call, that was being made anyways. - if r.kv.consul != nil { - return r.allConsul(ctx, keep) - } - return r.allOther(ctx, keep) - -} - -// allConsul is ONLY usable when consul is the keystore. This is a performance improvement in using the client directly -// instead of the cortex multi store kv interface. That interface returns the list then each value must be retrieved -// individually. This returns all the keys and values in one call and works on them in memory -func (r *Remote) allConsul(ctx context.Context, keep func(key string) bool) (<-chan instance.Config, error) { - if r.kv.consul == nil { - level.Error(r.log).Log("err", "allConsul called but consul client nil") - return nil, errors.New("allConsul called but consul client nil") + kvps, err := r.kv.List(ctx, "") + if err != nil { + return nil, fmt.Errorf("failed to list configs: %w", err) } - var configs []*instance.Config - c := GetCodec() - pairs, err := r.listConsul(ctx) + // NOTE(rfratto): ch must be buffered for everything we're going to write to + // it, otherwise this goroutine will deadlock. + // + // Data written to the buffered channel will still be available for reading + // even after the channel is closed. + ch := make(chan instance.Config, len(kvps)) + defer close(ch) - if err != nil { - return nil, err - } - for _, kvp := range pairs { + for _, kvp := range kvps { if keep != nil && !keep(kvp.Key) { level.Debug(r.log).Log("msg", "skipping key that was filtered out", "key", kvp.Key) continue } - value, err := c.Decode(kvp.Value) - if err != nil { - level.Error(r.log).Log("msg", "failed to decode config from store", "key", kvp.Key, "err", err) - continue - } - if value == nil { - // Config was deleted since we called list, skip it. - level.Debug(r.log).Log("msg", "skipping key that was deleted after list was called", "key", kvp.Key) - continue - } - cfg, err := instance.UnmarshalConfig(strings.NewReader(value.(string))) + cfg, err := instance.UnmarshalConfig(strings.NewReader(kvp.Value.(string))) if err != nil { level.Error(r.log).Log("msg", "failed to unmarshal config from store", "key", kvp.Key, "err", err) continue } - configs = append(configs, cfg) - } - ch := make(chan instance.Config, len(configs)) - for _, cfg := range configs { ch <- *cfg } - close(ch) - return ch, nil -} - -func (r *Remote) allOther(ctx context.Context, keep func(key string) bool) (<-chan instance.Config, error) { - if r.kv == nil { - return nil, ErrNotConnected - } - - keys, err := r.kv.List(ctx, "") - if err != nil { - return nil, fmt.Errorf("failed to list configs: %w", err) - } - - ch := make(chan instance.Config) - - var wg sync.WaitGroup - wg.Add(len(keys)) - go func() { - wg.Wait() - close(ch) - }() - - for _, key := range keys { - go func(key string) { - defer wg.Done() - - if keep != nil && !keep(key) { - level.Debug(r.log).Log("msg", "skipping key that was filtered out", "key", key) - return - } - - // TODO(rfratto): retries might be useful here - v, err := r.kv.Get(ctx, key) - if err != nil { - level.Error(r.log).Log("msg", "failed to get config with key", "key", key, "err", err) - return - } else if v == nil { - // Config was deleted since we called list, skip it. - level.Debug(r.log).Log("msg", "skipping key that was deleted after list was called", "key", key) - return - } - - cfg, err := instance.UnmarshalConfig(strings.NewReader(v.(string))) - if err != nil { - level.Error(r.log).Log("msg", "failed to unmarshal config from store", "key", key, "err", err) - return - } - ch <- *cfg - }(key) - } return ch, nil }