Handle grpc code resource exhausted for store gateway (cortexproject#…

…5286)

* handle grpc code resource exhausted for store gateway

Signed-off-by: Ben Ye <benye@amazon.com>

* fix lint

Signed-off-by: Ben Ye <benye@amazon.com>

* update changelog

Signed-off-by: Ben Ye <benye@amazon.com>

* try fixing test

Signed-off-by: Ben Ye <benye@amazon.com>

* try to fix E2E test

Signed-off-by: Ben Ye <benye@amazon.com>

* lint

Signed-off-by: Ben Ye <benye@amazon.com>

* try again

Signed-off-by: Ben Ye <benye@amazon.com>

* fix message

Signed-off-by: Ben Ye <benye@amazon.com>

* remove labels API

Signed-off-by: Ben Ye <benye@amazon.com>

* remove logic to check string contains

Signed-off-by: Ben Ye <benye@amazon.com>

* make limiter vars private

Signed-off-by: Ben Ye <benye@amazon.com>

---------

Signed-off-by: Ben Ye <benye@amazon.com>

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## master / unreleased
 * [BUGFIX] Ruler: Validate if rule group can be safely converted back to rule group yaml from protobuf message #5265
+* [BUGFIX] Querier: Convert gRPC `ResourceExhausted` status code from store gateway to 422 limit error. #5286
 
 ## 1.15.0 2023-04-19
 

integration/e2ecortex/client.go

@@ -300,72 +300,6 @@ 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()

integration/query_fuzz_test.go

@@ -133,7 +133,6 @@ func TestVerticalShardingFuzz(t *testing.T) {
 	opts := []promqlsmith.Option{
 		promqlsmith.WithEnableOffset(true),
 		promqlsmith.WithEnableAtModifier(true),
-		promqlsmith.WithEnableVectorMatching(true),
 	}
 	ps := promqlsmith.New(rnd, lbls, opts...)