Improve metrics integration tests (#2308)

* Improve metrics integration tests

Signed-off-by: Marco Pracucci <marco@pracucci.com>

* Added alertmanager metric prefix to integration tests

Signed-off-by: Marco Pracucci <marco@pracucci.com>

* Fixed linter and improved tests

Signed-off-by: Marco Pracucci <marco@pracucci.com>

* Remove cortex_alertmanager prefix check (I've been too much optimistic)

Signed-off-by: Marco Pracucci <marco@pracucci.com>

* Fixed integration tests after rebasing master

Signed-off-by: Marco Pracucci <marco@pracucci.com>

Author: pracucci

Makefile

@@ -123,7 +123,7 @@ protos: $(PROTO_GOS)
 
 lint:
 	misspell -error docs
-	golangci-lint run --build-tags netgo --timeout=5m --enable golint --enable misspell --enable gofmt
+	golangci-lint run --build-tags netgo,require_docker --timeout=5m --enable golint --enable misspell --enable gofmt
 
 	# Ensure no blacklisted package is imported.
 	faillint -paths "github.com/bmizerany/assert=github.com/stretchr/testify/assert" ./pkg/... ./cmd/... ./tools/... ./integration/...

integration/alertmanager_test.go

@@ -36,4 +36,7 @@ func TestAlertmanager(t *testing.T) {
 	require.Equal(t, "example_groupby", cfg.Route.GroupByStr[0])
 	require.Len(t, cfg.Receivers, 1)
 	require.Equal(t, "example_receiver", cfg.Receivers[0].Name)
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, AlertManager, alertmanager)
 }

integration/api_ruler_test.go

@@ -70,4 +70,7 @@ func TestRulerAPI(t *testing.T) {
 	// Check to ensure the rule groups are no longer active
 	_, err = c.GetRuleGroups()
 	require.Error(t, err)
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Ruler, ruler)
 }

integration/asserts.go

@@ -0,0 +1,78 @@
+// +build requires_docker
+
+package main
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/integration/e2ecortex"
+)
+
+type ServiceType int
+
+const (
+	Distributor = iota
+	Ingester
+	Querier
+	QueryFrontend
+	TableManager
+	AlertManager
+	Ruler
+)
+
+var (
+	// Service-specific metrics prefixes which shouldn't be used by any other service.
+	serviceMetricsPrefixes = map[ServiceType][]string{
+		Distributor:   []string{},
+		Ingester:      []string{},
+		Querier:       []string{},
+		QueryFrontend: []string{"cortex_frontend", "cortex_query_frontend"},
+		TableManager:  []string{},
+		AlertManager:  []string{},
+		Ruler:         []string{},
+	}
+
+	// Blacklisted metrics prefixes across any Cortex service.
+	blacklistedMetricsPrefixes = []string{
+		"cortex_alert_manager", // It should be "cortex_alertmanager"
+	}
+)
+
+func assertServiceMetricsPrefixes(t *testing.T, serviceType ServiceType, service *e2ecortex.CortexService) {
+	metrics, err := service.Metrics()
+	require.NoError(t, err)
+
+	// Build the list of blacklisted metrics prefixes for this specific service.
+	blacklist := getBlacklistedMetricsPrefixesByService(serviceType)
+
+	// Ensure no metric name matches the blacklisted prefixes.
+	for _, metricLine := range strings.Split(metrics, "\n") {
+		metricLine = strings.TrimSpace(metricLine)
+		if metricLine == "" || strings.HasPrefix(metricLine, "#") {
+			continue
+		}
+
+		for _, prefix := range blacklist {
+			assert.NotRegexp(t, "^"+prefix, metricLine, "service: %s", service.Name())
+		}
+	}
+}
+
+func getBlacklistedMetricsPrefixesByService(serviceType ServiceType) []string {
+	blacklist := append([]string{}, blacklistedMetricsPrefixes...)
+
+	// Add any service-specific metrics prefix excluding the service itself.
+	for t, prefixes := range serviceMetricsPrefixes {
+		if t == serviceType {
+			continue
+		}
+
+		blacklist = append(blacklist, prefixes...)
+	}
+
+	return blacklist
+}

integration/chunks_storage_backends_test.go

@@ -108,4 +108,9 @@ func TestChunksStorageAllIndexBackends(t *testing.T) {
 		require.Equal(t, model.ValVector, result.Type())
 		assert.Equal(t, expectedVector, result.(model.Vector))
 	}
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Distributor, distributor)
+	assertServiceMetricsPrefixes(t, Ingester, ingester)
+	assertServiceMetricsPrefixes(t, Querier, querier)
 }

integration/e2ecortex/services.go

@@ -224,7 +224,7 @@ func NewRuler(name string, flags map[string]string, image string) *CortexService
 			"-log.level": "warn",
 		}, flags))...),
 		// The alertmanager doesn't expose a readiness probe, so we just check if the / returns 200
-		e2e.NewReadinessProbe(httpPort, "/", 200),
+		e2e.NewHTTPReadinessProbe(httpPort, "/", 200),
 		httpPort,
 		grpcPort,
 	)

integration/ingester_flush_test.go

@@ -33,12 +33,12 @@ func TestIngesterFlushWithChunksStorage(t *testing.T) {
 	require.NoError(t, writeFileToSharedDir(s, cortexSchemaConfigFile, []byte(cortexSchemaConfigYaml)))
 
 	tableManager := e2ecortex.NewTableManager("table-manager", ChunksStorageFlags, "")
-	ingester1 := e2ecortex.NewIngester("ingester-1", consul.NetworkHTTPEndpoint(), mergeFlags(ChunksStorageFlags, map[string]string{
+	ingester := e2ecortex.NewIngester("ingester-1", consul.NetworkHTTPEndpoint(), mergeFlags(ChunksStorageFlags, map[string]string{
 		"-ingester.max-transfer-retries": "0",
 	}), "")
 	querier := e2ecortex.NewQuerier("querier", consul.NetworkHTTPEndpoint(), ChunksStorageFlags, "")
 	distributor := e2ecortex.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), ChunksStorageFlags, "")
-	require.NoError(t, s.StartAndWaitReady(distributor, querier, ingester1, tableManager))
+	require.NoError(t, s.StartAndWaitReady(distributor, querier, ingester, tableManager))
 
 	// Wait until the first table-manager sync has completed, so that we're
 	// sure the tables have been created.
@@ -73,9 +73,12 @@ func TestIngesterFlushWithChunksStorage(t *testing.T) {
 	require.Equal(t, model.ValVector, result.Type())
 	assert.Equal(t, expectedVector2, result.(model.Vector))
 
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Ingester, ingester)
+
 	// Stop ingester-1, so that it will flush all chunks to the storage. This function will return
 	// once the ingester-1 is successfully stopped, which means the flushing is completed.
-	require.NoError(t, s.Stop(ingester1))
+	require.NoError(t, s.Stop(ingester))
 
 	// Ensure chunks have been uploaded to the storage (DynamoDB).
 	dynamoURL := "dynamodb://u:p@" + dynamo.Endpoint(8000)
@@ -94,4 +97,9 @@ func TestIngesterFlushWithChunksStorage(t *testing.T) {
 	out, err = dynamoClient.Scan(&dynamodb.ScanInput{TableName: aws.String(chunksTable)})
 	require.NoError(t, err)
 	assert.Equal(t, int64(2), *out.Count)
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Distributor, distributor)
+	assertServiceMetricsPrefixes(t, Querier, querier)
+	assertServiceMetricsPrefixes(t, TableManager, tableManager)
 }

integration/ingester_hand_over_test.go

@@ -89,4 +89,9 @@ func runIngesterHandOverTest(t *testing.T, flags map[string]string, setup func(t
 	result, err = c.Query("series_1", now)
 	require.NoError(t, err)
 	assert.Equal(t, expectedVector, result.(model.Vector))
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Distributor, distributor)
+	assertServiceMetricsPrefixes(t, Ingester, ingester2)
+	assertServiceMetricsPrefixes(t, Querier, querier)
 }

integration/metrics_test.go

@@ -36,7 +36,7 @@ func TestQuerierWithBlocksStorage(t *testing.T) {
 				"-experimental.tsdb.bucket-store.index-cache.backend": "inmemory",
 			}),
 		},
-		"queintegration/e2e/service.gorier running with memcached index cache": {
+		"querier running with memcached index cache": {
 			flags: mergeFlags(BlocksStorageFlags, map[string]string{
 				// The address will be inject during the test execution because it's dynamic.
 				"-experimental.tsdb.bucket-store.index-cache.backend": "memcached",
@@ -169,6 +169,11 @@ func TestQuerierWithBlocksStorage(t *testing.T) {
 			} else if indexCacheBackend == tsdb.IndexCacheBackendMemcached {
 				require.NoError(t, querier.WaitSumMetrics(e2e.Equals(11+2), "cortex_querier_blocks_index_cache_memcached_operations_total")) // as before + 2 gets
 			}
+
+			// Ensure no service-specific metrics prefix is used by the wrong service.
+			assertServiceMetricsPrefixes(t, Distributor, distributor)
+			assertServiceMetricsPrefixes(t, Ingester, ingester)
+			assertServiceMetricsPrefixes(t, Querier, querier)
 		})
 	}
 }
@@ -292,7 +297,7 @@ func TestQuerierWithChunksStorage(t *testing.T) {
 	expectedVectors := make([]model.Vector, numUsers)
 
 	for u := 0; u < numUsers; u++ {
-		c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", fmt.Sprintf("user-%d", u))
+		c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), "", "", "", fmt.Sprintf("user-%d", u))
 		require.NoError(t, err)
 
 		var series []prompb.TimeSeries
@@ -323,7 +328,7 @@ func TestQuerierWithChunksStorage(t *testing.T) {
 	for u := 0; u < numUsers; u++ {
 		userID := u
 
-		c, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", fmt.Sprintf("user-%d", userID))
+		c, err := e2ecortex.NewClient("", querier.HTTPEndpoint(), "", "", fmt.Sprintf("user-%d", userID))
 		require.NoError(t, err)
 
 		for q := 0; q < numQueriesPerUser; q++ {
@@ -339,4 +344,10 @@ func TestQuerierWithChunksStorage(t *testing.T) {
 	}
 
 	wg.Wait()
+
+	// Ensure no service-specific metrics prefix is used by the wrong service.
+	assertServiceMetricsPrefixes(t, Distributor, distributor)
+	assertServiceMetricsPrefixes(t, Ingester, ingester)
+	assertServiceMetricsPrefixes(t, Querier, querier)
+	assertServiceMetricsPrefixes(t, TableManager, tableManager)
 }