Read-write mode integration test improvements

integration/read_write_mode_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/grafana/e2e"
 	e2edb "github.com/grafana/e2e/db"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/require"
 
@@ -41,6 +42,13 @@ func TestReadWriteMode(t *testing.T) {
 	c, err := e2emimir.NewClient(writeInstance.HTTPEndpoint(), readInstance.HTTPEndpoint(), "", "", "user-1")
 	require.NoError(t, err)
 
+	// Wait for the ingester to join the ring and become active - this prevents "empty ring" errors later when we try to query data.
+	require.NoError(t, readInstance.WaitSumMetricsWithOptions(
+		e2e.Equals(1),
+		[]string{"cortex_ring_members"},
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "name", "ingester"), labels.MustNewMatcher(labels.MatchEqual, "state", "ACTIVE")),
+	))
+
 	// Push some data to the cluster.
 	now := time.Now()
 	series, expectedVector := generateSeries("test_series_1", now, prompb.Label{Name: "foo", Value: "bar"})
@@ -50,13 +58,15 @@ func TestReadWriteMode(t *testing.T) {
 	require.Equal(t, 200, res.StatusCode)
 
 	// Verify we can read the data we just pushed, both with an instant query and a range query.
-	result, err := c.Query("test_series_1", now)
+	queryResult, err := c.Query("test_series_1", now)
 	require.NoError(t, err)
-	require.Equal(t, model.ValVector, result.Type())
-	require.Equal(t, expectedVector, result.(model.Vector))
+	require.Equal(t, model.ValVector, queryResult.Type())
+	require.Equal(t, expectedVector, queryResult.(model.Vector))
 
-	_, err = c.QueryRange("test_series_1", now.Add(-5*time.Minute), now, 15*time.Second)
+	rangeResult, err := c.QueryRange("test_series_1", now.Add(-5*time.Minute), now, 15*time.Second)
 	require.NoError(t, err)
+	require.Equal(t, model.ValMatrix, rangeResult.Type())
+	require.Equal(t, expectedMatrix(series), rangeResult.(model.Matrix))
 
 	minTime := time.Unix(math.MinInt64/1000+62135596801, 0).UTC()
 	maxTime := time.Unix(math.MaxInt64/1000-62135596801, 999999999).UTC()
@@ -70,3 +80,30 @@ func TestReadWriteMode(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []string{"__name__", "foo"}, labelNames)
 }
+
+func expectedMatrix(series []prompb.TimeSeries) model.Matrix {
+	m := make(model.Matrix, 0, len(series))
+
+	for _, s := range series {
+
+		m = append(m, &model.SampleStream{
+			Metric: prompbLabelsToModelMetric(s.Labels),
+			Values: prompbSamplesToModelSamplePairs(s.Samples),
+		})
+	}
+
+	return m
+}
+
+func prompbSamplesToModelSamplePairs(samples []prompb.Sample) []model.SamplePair {
+	v := make([]model.SamplePair, 0, len(samples))
+
+	for _, s := range samples {
+		v = append(v, model.SamplePair{
+			Timestamp: model.Time(s.Timestamp),
+			Value:     model.SampleValue(s.Value),
+		})
+	}
+
+	return v
+}