Add integration test using ingest storage (#7160)

* Add integration test using ingest storage Signed-off-by: Marco Pracucci <[email protected]> * Re-vendored grafana/e2e Signed-off-by: Marco Pracucci <[email protected]> --------- Signed-off-by: Marco Pracucci <[email protected]>
grafana · Jan 18, 2024 · fadc22f · fadc22f
1 parent befb428
commit fadc22f
Show file tree

Hide file tree

Showing 8 changed files with 152 additions and 6 deletions.
diff --git a/go.mod b/go.mod
@@ -21,7 +21,7 @@ require (
 	github.com/google/gopacket v1.1.19
 	github.com/gorilla/mux v1.8.1
 	github.com/grafana/dskit v0.0.0-20240117144322-b9a439dedeb8
-	github.com/grafana/e2e v0.1.1
+	github.com/grafana/e2e v0.1.2-0.20240118170847-db90b84177fc
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/json-iterator/go v1.1.12
 	github.com/minio/minio-go/v7 v7.0.66

diff --git a/go.sum b/go.sum
@@ -545,8 +545,8 @@ github.com/grafana-tools/sdk v0.0.0-20220919052116-6562121319fc h1:PXZQA2WCxe85T
 github.com/grafana-tools/sdk v0.0.0-20220919052116-6562121319fc/go.mod h1:AHHlOEv1+GGQ3ktHMlhuTUwo3zljV3QJbC0+8o2kn+4=
 github.com/grafana/dskit v0.0.0-20240117144322-b9a439dedeb8 h1:6sqXQFpxe/VO2VsBwhdbGd6TGSULFf9Mpzij7zXQ66o=
 github.com/grafana/dskit v0.0.0-20240117144322-b9a439dedeb8/go.mod h1:x5DMwyr1kyirtHOxoFSZ7RnyOgHdGh03ZruupdPetQM=
-github.com/grafana/e2e v0.1.1 h1:/b6xcv5BtoBnx8cZnCiey9DbjEc8z7gXHO5edoeRYxc=
-github.com/grafana/e2e v0.1.1/go.mod h1:RpNLgae5VT+BUHvPE+/zSypmOXKwEu4t+tnEMS1ATaE=
+github.com/grafana/e2e v0.1.2-0.20240118170847-db90b84177fc h1:BW+LjKJDz0So5LI8UZfW5neWeKpSkWqhmGjQFzcFfLM=
+github.com/grafana/e2e v0.1.2-0.20240118170847-db90b84177fc/go.mod h1:JVmqPBe8A/pZWwRoJW5ZjyALeY5OXMzPl7LrVXOdZAI=
 github.com/grafana/goautoneg v0.0.0-20231010094147-47ce5e72a9ae h1:Yxbw9jKGJVC6qAK5Ubzzb/qZwM6rRMMqaDc/d4Vp3pM=
 github.com/grafana/goautoneg v0.0.0-20231010094147-47ce5e72a9ae/go.mod h1:GFAN9Jn9t1cX7sNfc6ZoFyc4f7i8jtm3SajrWdZM2EE=
 github.com/grafana/gomemcache v0.0.0-20231023152154-6947259a0586 h1:/of8Z8taCPftShATouOrBVy6GaTTjgQd/VfNiZp/VXQ=

diff --git a/integration/configs.go b/integration/configs.go
@@ -16,6 +16,8 @@ import (
 
 	e2e "github.com/grafana/e2e"
 	e2edb "github.com/grafana/e2e/db"
+
+	"github.com/grafana/mimir/pkg/querier/api"
 )
 
 const (
@@ -208,6 +210,23 @@ blocks_storage:
 		MinioSecretKey: e2edb.MinioSecretKey,
 		MinioEndpoint:  fmt.Sprintf("%s-minio-9000:9000", networkName),
 	})
+
+	IngestStorageFlags = func() map[string]string {
+		return map[string]string{
+			"-ingest-storage.enabled":       "true",
+			"-ingest-storage.kafka.topic":   "ingest",
+			"-ingest-storage.kafka.address": fmt.Sprintf("%s-kafka:9092", networkName),
+
+			// To simplify integration tests, we use strong read consistency by default.
+			// Integration tests that want to test the eventual consistency can override it.
+			"-ingest-storage.read-consistency": api.ReadConsistencyStrong,
+
+			// Frequently poll last produced offset in order to have a low end-to-end latency
+			// and faster integration tests.
+			"-ingest-storage.kafka.last-produced-offset-poll-interval": "50ms",
+			"-ingest-storage.kafka.last-produced-offset-retry-timeout": "1s",
+		}
+	}
 )
 
 func buildConfigFromTemplate(tmpl string, data interface{}) string {

diff --git a/integration/query_frontend_test.go b/integration/query_frontend_test.go
@@ -229,6 +229,28 @@ func TestQueryFrontendWithQueryResultPayloadFormats(t *testing.T) {
 	}
 }
 
+func TestQueryFrontendWithIngestStorageViaFlagsAndWithQuerySchedulerAndQueryStatsEnabled(t *testing.T) {
+	runQueryFrontendTest(t, queryFrontendTestConfig{
+		querySchedulerEnabled:       true,
+		querySchedulerDiscoveryMode: "dns",
+		queryStatsEnabled:           true,
+		setup: func(t *testing.T, s *e2e.Scenario) (configFile string, flags map[string]string) {
+			flags = mergeFlags(
+				BlocksStorageFlags(),
+				BlocksStorageS3Flags(),
+				IngestStorageFlags(),
+			)
+
+			kafka := e2edb.NewKafka()
+			minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
+			require.NoError(t, s.StartAndWaitReady(minio, kafka))
+
+			return "", flags
+		},
+		withHistograms: true,
+	})
+}
+
 func runQueryFrontendTest(t *testing.T, cfg queryFrontendTestConfig) {
 	const numUsers = 10
 	const numQueriesPerUser = 10
@@ -275,7 +297,7 @@ func runQueryFrontendTest(t *testing.T, cfg queryFrontendTestConfig) {
 	}
 
 	// Start all other services.
-	ingester := e2emimir.NewIngester("ingester", consul.NetworkHTTPEndpoint(), flags, e2emimir.WithConfigFile(configFile))
+	ingester := e2emimir.NewIngester("ingester-0", consul.NetworkHTTPEndpoint(), flags, e2emimir.WithConfigFile(configFile))
 	distributor := e2emimir.NewDistributor("distributor", consul.NetworkHTTPEndpoint(), flags, e2emimir.WithConfigFile(configFile))
 	querier := e2emimir.NewQuerier("querier", consul.NetworkHTTPEndpoint(), flags, e2emimir.WithConfigFile(configFile))
 

diff --git a/tools/pre-pull-images/main.go b/tools/pre-pull-images/main.go
@@ -18,6 +18,7 @@ func main() {
 	fmt.Println(images.Consul)
 	fmt.Println(images.ETCD)
 	fmt.Println(images.Memcached)
+	fmt.Println(images.Kafka)
 
 	// images from previous releases
 	for image := range integration.DefaultPreviousVersionImages {

diff --git a/vendor/github.com/grafana/e2e/db/kafka.go b/vendor/github.com/grafana/e2e/db/kafka.go
diff --git a/vendor/github.com/grafana/e2e/images/images.go b/vendor/github.com/grafana/e2e/images/images.go
diff --git a/vendor/modules.txt b/vendor/modules.txt