Make each DNS provider register a diff metric (#2303)

* Make each DNS provider register a diff metric

Signed-off-by: Goutham Veeramachaneni <gouthamve@gmail.com>

* Add tests to test multi register.

Also incorporate review feedback.

Signed-off-by: Goutham Veeramachaneni <gouthamve@gmail.com>

Author: gouthamve

integration/querier_test.go

@@ -3,10 +3,13 @@
 package main
 
 import (
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -251,3 +254,89 @@ func TestQuerierWithBlocksStorageOnMissingBlocksFromStorage(t *testing.T) {
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "500")
 }
+
+func TestQuerierWithChunksStorage(t *testing.T) {
+	const numUsers = 10
+	const numQueriesPerUser = 10
+
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	require.NoError(t, writeFileToSharedDir(s, cortexSchemaConfigFile, []byte(cortexSchemaConfigYaml)))
+	flags := mergeFlags(ChunksStorageFlags, map[string]string{})
+
+	// Start dependencies.
+	dynamo := e2edb.NewDynamoDB()
+
+	consul := e2edb.NewConsul()
+	require.NoError(t, s.StartAndWaitReady(consul, dynamo))
+
+	tableManager := e2ecortex.NewTableManager("table-manager", ChunksStorageFlags, "")
+	require.NoError(t, s.StartAndWaitReady(tableManager))
+
+	// Wait until the first table-manager sync has completed, so that we're
+	// sure the tables have been created.
+	require.NoError(t, tableManager.WaitSumMetrics(e2e.Greater(0), "cortex_table_manager_sync_success_timestamp_seconds"))
+
+	// Start Cortex components for the write path.
+	distributor := e2ecortex.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), flags, "")
+	ingester := e2ecortex.NewIngester("ingester", consul.NetworkHTTPEndpoint(), flags, "")
+	require.NoError(t, s.StartAndWaitReady(distributor, ingester))
+
+	// Wait until the distributor has updated the ring.
+	require.NoError(t, distributor.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	// Push a series for each user to Cortex.
+	now := time.Now()
+	expectedVectors := make([]model.Vector, numUsers)
+
+	for u := 0; u < numUsers; u++ {
+		c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", fmt.Sprintf("user-%d", u))
+		require.NoError(t, err)
+
+		var series []prompb.TimeSeries
+		series, expectedVectors[u] = generateSeries("series_1", now)
+
+		res, err := c.Push(series)
+		require.NoError(t, err)
+		require.Equal(t, 200, res.StatusCode)
+	}
+
+	// Add 2 memcache instances to test for: https://github.com/cortexproject/cortex/issues/2302
+	// Note these are not running but added to trigger the behaviour.
+	querierFlags := mergeFlags(flags, map[string]string{
+		"-store.index-cache-read.memcached.addresses":  "dns+memcached0:11211",
+		"-store.index-cache-write.memcached.addresses": "dns+memcached1:11211",
+	})
+
+	querier := e2ecortex.NewQuerier("querier", consul.NetworkHTTPEndpoint(), querierFlags, "")
+	require.NoError(t, s.StartAndWaitReady(querier))
+
+	// Wait until the querier has updated the ring.
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(512), "cortex_ring_tokens_total"))
+
+	// Query the series for each user in parallel.
+	wg := sync.WaitGroup{}
+	wg.Add(numUsers * numQueriesPerUser)
+
+	for u := 0; u < numUsers; u++ {
+		userID := u
+
+		c, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", fmt.Sprintf("user-%d", userID))
+		require.NoError(t, err)
+
+		for q := 0; q < numQueriesPerUser; q++ {
+			go func() {
+				defer wg.Done()
+
+				result, err := c.Query("series_1", now)
+				require.NoError(t, err)
+				require.Equal(t, model.ValVector, result.Type())
+				assert.Equal(t, expectedVectors[userID], result.(model.Vector))
+			}()
+		}
+	}
+
+	wg.Wait()
+}

integration/query_frontend_test.go

@@ -106,6 +106,7 @@ func runQueryFrontendTest(t *testing.T, setup queryFrontendSetup) {
 
 	// Check if we're discovering memcache or not.
 	require.NoError(t, queryFrontend.WaitSumMetrics(e2e.Equals(1), "cortex_memcache_client_servers"))
+	require.NoError(t, queryFrontend.WaitSumMetrics(e2e.Greater(0), "cortex_dns_lookups_total"))
 
 	// Start the querier after the query-frontend otherwise we're not
 	// able to get the query-frontend network endpoint.

pkg/chunk/cache/cache.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"flag"
 	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // Cache byte arrays by key.
@@ -79,7 +81,7 @@ func New(cfg Config) (Cache, error) {
 			cfg.Memcache.Expiration = cfg.DefaultValidity
 		}
 
-		client := NewMemcachedClient(cfg.MemcacheClient, cfg.Prefix)
+		client := NewMemcachedClient(cfg.MemcacheClient, cfg.Prefix, prometheus.DefaultRegisterer)
 		cache := NewMemcached(cfg.Memcache, client, cfg.Prefix)
 
 		cacheName := cfg.Prefix + "memcache"

pkg/chunk/cache/memcached_client.go

@@ -80,7 +80,7 @@ func (cfg *MemcachedClientConfig) RegisterFlagsWithPrefix(prefix, description st
 
 // NewMemcachedClient creates a new MemcacheClient that gets its server list
 // from SRV and updates the server list on a regular basis.
-func NewMemcachedClient(cfg MemcachedClientConfig, name string) MemcachedClient {
+func NewMemcachedClient(cfg MemcachedClientConfig, name string, r prometheus.Registerer) MemcachedClient {
 	var selector serverSelector
 	if cfg.ConsistentHash {
 		selector = &MemcachedJumpHashSelector{}
@@ -92,13 +92,17 @@ func NewMemcachedClient(cfg MemcachedClientConfig, name string) MemcachedClient
 	client.Timeout = cfg.Timeout
 	client.MaxIdleConns = cfg.MaxIdleConns
 
+	dnsProviderRegisterer := prometheus.WrapRegistererWithPrefix("cortex_", prometheus.WrapRegistererWith(prometheus.Labels{
+		"name": name,
+	}, r))
+
 	newClient := &memcachedClient{
 		Client:     client,
 		serverList: selector,
 		hostname:   cfg.Host,
 		service:    cfg.Service,
 		addresses:  strings.Split(cfg.Addresses, ","),
-		provider:   dns.NewProvider(util.Logger, prometheus.DefaultRegisterer, dns.GolangResolverType),
+		provider:   dns.NewProvider(util.Logger, dnsProviderRegisterer, dns.GolangResolverType),
 		quit:       make(chan struct{}),
 
 		numServers: memcacheServersDiscovered.WithLabelValues(name),