[dev.scraping-service-scalability] Fork the cortex ring/kv package for use with the configstore

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)

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: <duration> | 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: <duration> | default = "30s"]
 
 # The timeout for a cluster reshard events. A timeout of 0 indicates no timeout.
 [cluster_reshard_event_timeout: <duration> | 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: <kvstore_config>
 
 # When set, allows configs pushed to the KV store to specify configuration

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

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

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

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.<flag-name>, and ring.store, going forward it would
+	// be easier to have everything under ring, so ring.consul.<flag-name>
+	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
+}