Function TestIngester_QueryStream

pkg/ingester/ingester_test.go:3843–3926

(t *testing.T)

}

func TestIngester_QueryStream(t *testing.T) {
	// Create ingester.
	cfg := defaultIngesterTestConfig(t)

	for _, enc := range encodings {
		t.Run(enc.String(), func(t *testing.T) {
			i, err := prepareIngesterWithBlocksStorage(t, cfg, prometheus.NewRegistry())
			require.NoError(t, err)
			require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
			defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck

			// Wait until it's ACTIVE.
			test.Poll(t, 1*time.Second, ring.ACTIVE, func() any {
				return i.lifecycler.GetState()
			})

			// Push series.
			ctx := user.InjectOrgID(context.Background(), userID)
			lbls := labels.FromStrings(labels.MetricName, "foo")
			var (
				req                    *cortexpb.WriteRequest
				expectedResponseChunks *client.QueryStreamResponse
			)
			switch enc {
			case encoding.PrometheusXorChunk:
				req, expectedResponseChunks = mockWriteRequest(t, lbls, 123000, 456)
			case encoding.PrometheusHistogramChunk:
				req, expectedResponseChunks = mockHistogramWriteRequest(t, lbls, 123000, 456, false)
			case encoding.PrometheusFloatHistogramChunk:
				req, expectedResponseChunks = mockHistogramWriteRequest(t, lbls, 123000, 456, true)
			}
			_, err = i.Push(ctx, req)
			require.NoError(t, err)

			// Create a GRPC server used to query back the data.
			serv := grpc.NewServer(grpc.StreamInterceptor(middleware.StreamServerUserHeaderInterceptor))
			defer serv.GracefulStop()
			client.RegisterIngesterServer(serv, i)

			listener, err := net.Listen("tcp", "localhost:0")
			require.NoError(t, err)

			go func() {
				require.NoError(t, serv.Serve(listener))
			}()

			// Query back the series using GRPC streaming.
			c, err := client.MakeIngesterClient(listener.Addr().String(), defaultClientTestConfig(), false)
			require.NoError(t, err)
			defer c.Close()

			queryRequest := &client.QueryRequest{
				StartTimestampMs: 0,
				EndTimestampMs:   200000,
				Matchers: []*client.LabelMatcher{{
					Type:  client.EQUAL,
					Name:  model.MetricNameLabel,
					Value: "foo",