e2e: add dedup cases for pushdown

Add some cases for pushdown with deduplication to test how it works. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com>
thanos-io · Dec 22, 2021 · 89f78f4 · 89f78f4
1 parent 53ec210
commit 89f78f4
Showing 1 changed file with 129 additions and 0 deletions.
diff --git a/test/e2e/query_test.go b/test/e2e/query_test.go
@@ -1006,3 +1006,132 @@ func synthesizeSamples(ctx context.Context, prometheus *e2e.InstrumentedRunnable
 	compressed := snappy.Encode(buf, pBuf.Bytes())
 	return client.Store(ctx, compressed)
 }
+
+func TestSidecarQueryEvaluationWithDedup(t *testing.T) {
+	t.Parallel()
+
+	ts := []struct {
+		prom1Samples []fakeMetricSample
+		prom2Samples []fakeMetricSample
+		query        string
+		result       model.Vector
+	}{
+		{
+			query:        "max (my_fake_metric)",
+			prom1Samples: []fakeMetricSample{{"i1", 1}, {"i2", 5}, {"i3", 9}},
+			prom2Samples: []fakeMetricSample{{"i1", 3}, {"i2", 4}, {"i3", 10}},
+			result: []*model.Sample{
+				{
+					Metric: map[model.LabelName]model.LabelValue{},
+					Value:  9, // Why? Very weird.
+				},
+			},
+		},
+		{
+			query:        "max by (instance) (my_fake_metric)",
+			prom1Samples: []fakeMetricSample{{"i1", 1}, {"i2", 5}, {"i3", 9}},
+			prom2Samples: []fakeMetricSample{{"i1", 3}, {"i2", 4}, {"i3", 10}},
+			result: []*model.Sample{
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i1"},
+					Value:  1, // Weird results.
+				},
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i2"},
+					Value:  5,
+				},
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i3"},
+					Value:  9, // Weird again :/
+				},
+			},
+		},
+		{
+			query:        "group by (instance) (my_fake_metric)",
+			prom1Samples: []fakeMetricSample{{"i1", 1}, {"i2", 5}, {"i3", 9}},
+			prom2Samples: []fakeMetricSample{{"i1", 3}, {"i2", 4}},
+			result: []*model.Sample{
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i1"},
+					Value:  1,
+				},
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i2"},
+					Value:  1,
+				},
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i3"},
+					Value:  1,
+				},
+			},
+		},
+		{
+			query:        "max_over_time(my_fake_metric[10m])",
+			prom1Samples: []fakeMetricSample{{"i1", 1}, {"i2", 5}},
+			prom2Samples: []fakeMetricSample{{"i1", 3}},
+			result: []*model.Sample{
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i1", "prometheus": "p1"},
+					Value:  1, // I expect to have 3 here?
+				},
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i2", "prometheus": "p1"},
+					Value:  5,
+				},
+			},
+		},
+		{
+			query:        "min_over_time(my_fake_metric[10m])",
+			prom1Samples: []fakeMetricSample{{"i1", 1}, {"i2", 5}},
+			prom2Samples: []fakeMetricSample{{"i1", 3}},
+			result: []*model.Sample{
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i1", "prometheus": "p1"},
+					Value:  1,
+				},
+				{
+					Metric: map[model.LabelName]model.LabelValue{"instance": "i2", "prometheus": "p1"},
+					Value:  5,
+				},
+			},
+		},
+	}
+
+	for _, tc := range ts {
+		t.Run(tc.query, func(t *testing.T) {
+			e, err := e2e.NewDockerEnvironment("e2e_test_query_pushdown")
+			testutil.Ok(t, err)
+			t.Cleanup(e2ethanos.CleanScenario(t, e))
+
+			prom1, sidecar1, err := e2ethanos.NewPrometheusWithSidecar(e, "p1", defaultPromConfig("p1", 0, "", ""), "", e2ethanos.DefaultPrometheusImage(), "remote-write-receiver")
+			testutil.Ok(t, err)
+			testutil.Ok(t, e2e.StartAndWaitReady(prom1, sidecar1))
+
+			prom2, sidecar2, err := e2ethanos.NewPrometheusWithSidecar(e, "p2", defaultPromConfig("p1", 1, "", ""), "", e2ethanos.DefaultPrometheusImage(), "remote-write-receiver")
+			testutil.Ok(t, err)
+			testutil.Ok(t, e2e.StartAndWaitReady(prom2, sidecar2))
+
+			endpoints := []string{
+				sidecar1.InternalEndpoint("grpc"),
+				sidecar2.InternalEndpoint("grpc"),
+			}
+			q, err := e2ethanos.
+				NewQuerierBuilder(e, "1", endpoints...).
+				WithEnabledFeatures([]string{"query-pushdown"}).
+				Build()
+			testutil.Ok(t, err)
+			testutil.Ok(t, e2e.StartAndWaitReady(q))
+
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
+			t.Cleanup(cancel)
+
+			testutil.Ok(t, synthesizeSamples(ctx, prom1, tc.prom1Samples))
+			testutil.Ok(t, synthesizeSamples(ctx, prom2, tc.prom2Samples))
+
+			testQuery := func() string { return tc.query }
+			queryAndAssert(t, ctx, q.Endpoint("http"), testQuery, time.Now, promclient.QueryOptions{
+				Deduplicate: true,
+			}, tc.result)
+		})
+	}
+}