[dev.scraping-service-scalability] Move #802 fix to kv fork

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`,

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.

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.

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.

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)
 	})
 }

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)

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) {

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)
 }

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{}
+}

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,