fix remote read error in query frontend (cortexproject#5257)

* fix remote read error in query frontend

Signed-off-by: Ben Ye <[email protected]>

* fix integration test

Signed-off-by: Ben Ye <[email protected]>

* add extra one query

Signed-off-by: Ben Ye <[email protected]>

---------

Signed-off-by: Ben Ye <[email protected]>

Author: yeya24

CHANGELOG.md

@@ -2,9 +2,9 @@
 
 ## master / unreleased
 * [CHANGE] Store gateways summary metrics have been converted to histograms `cortex_bucket_store_series_blocks_queried`, `cortex_bucket_store_series_data_fetched`, `cortex_bucket_store_series_data_size_touched_bytes`, `cortex_bucket_store_series_data_size_fetched_bytes`, `cortex_bucket_store_series_data_touched`, `cortex_bucket_store_series_result_series` #5239
-
 * [ENHANCEMENT] Querier: Batch Iterator optimization to prevent transversing it multiple times query ranges steps does not overlap. #5237
 * [BUGFIX] Catch context error in the s3 bucket client. #5240
+* [BUGFIX] Fix query frontend remote read empty body. #5257
 
 ## 1.15.0 2023-04-19
 

integration/e2ecortex/client.go

@@ -300,6 +300,72 @@ func (c *Client) RemoteRead(matchers []*labels.Matcher, start, end time.Time, st
 	return &resp, nil
 }
 
+// RemoteRead runs a remote read query.
+func (c *Client) RemoteRead(matchers []*labels.Matcher, start, end time.Time, step time.Duration) (*prompb.ReadResponse, error) {
+	startMs := start.UnixMilli()
+	endMs := end.UnixMilli()
+	stepMs := step.Milliseconds()
+
+	q, err := remote.ToQuery(startMs, endMs, matchers, &storage.SelectHints{
+		Step:  stepMs,
+		Start: startMs,
+		End:   endMs,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	req := &prompb.ReadRequest{
+		Queries:               []*prompb.Query{q},
+		AcceptedResponseTypes: []prompb.ReadRequest_ResponseType{prompb.ReadRequest_STREAMED_XOR_CHUNKS},
+	}
+
+	data, err := proto.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+	compressed := snappy.Encode(nil, data)
+
+	// Call the remote read API endpoint with a timeout.
+	httpReqCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	httpReq, err := http.NewRequestWithContext(httpReqCtx, "POST", "http://"+c.querierAddress+"/prometheus/api/v1/read", bytes.NewReader(compressed))
+	if err != nil {
+		return nil, err
+	}
+	httpReq.Header.Set("X-Scope-OrgID", "user-1")
+	httpReq.Header.Add("Content-Encoding", "snappy")
+	httpReq.Header.Add("Accept-Encoding", "snappy")
+	httpReq.Header.Set("Content-Type", "application/x-protobuf")
+	httpReq.Header.Set("User-Agent", "Prometheus/1.8.2")
+	httpReq.Header.Set("X-Prometheus-Remote-Read-Version", "0.1.0")
+
+	httpResp, err := c.httpClient.Do(httpReq)
+	if err != nil {
+		return nil, err
+	}
+	if httpResp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status code %d", httpResp.StatusCode)
+	}
+
+	compressed, err = io.ReadAll(httpResp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	uncompressed, err := snappy.Decode(nil, compressed)
+	if err != nil {
+		return nil, err
+	}
+
+	var resp prompb.ReadResponse
+	if err = proto.Unmarshal(uncompressed, &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
 func (c *Client) query(addr string) (*http.Response, []byte, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
 	defer cancel()