fix response error to be ungzipped when status code is not 2xx (#4975)

* fix response to be gzipped when status code is not 2xx

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

* adding tests

Signed-off-by: Alan Protasio <[email protected]>

* lint

Signed-off-by: Alan Protasio <[email protected]>

* changelog

Signed-off-by: Alan Protasio <[email protected]>

Signed-off-by: Ben Ye <[email protected]>
Signed-off-by: Alan Protasio <[email protected]>
Co-authored-by: Alan Protasio <[email protected]>

Author: yeya24

CHANGELOG.md

@@ -78,6 +78,7 @@
 * [BUGFIX] Ingester: fixed incorrect logging at the start of ingester block shipping logic. #4934
 * [BUGFIX] Storage/Bucket: fixed global mark missing on deletion. #4949
 * [BUGFIX] QueryFrontend/Querier: fixed regression added by #4863 where we stopped compressing the response between querier and query frontend. #4960
+* [BUGFIX] QueryFrontend/Querier: fixed fix response error to be ungzipped when status code is not 2xx. #4975
 
 ## 1.13.0 2022-07-14
 

pkg/querier/tripperware/instantquery/instant_query.go

@@ -152,11 +152,6 @@ func (c instantQueryCodec) DecodeRequest(_ context.Context, r *http.Request, for
 }
 
 func (instantQueryCodec) DecodeResponse(ctx context.Context, r *http.Response, _ tripperware.Request) (tripperware.Response, error) {
-	if r.StatusCode/100 != 2 {
-		body, _ := io.ReadAll(r.Body)
-		return nil, httpgrpc.Errorf(r.StatusCode, string(body))
-	}
-
 	log, ctx := spanlogger.New(ctx, "PrometheusInstantQueryResponse") //nolint:ineffassign,staticcheck
 	defer log.Finish()
 
@@ -165,6 +160,10 @@ func (instantQueryCodec) DecodeResponse(ctx context.Context, r *http.Response, _
 		log.Error(err)
 		return nil, err
 	}
+	if r.StatusCode/100 != 2 {
+		return nil, httpgrpc.Errorf(r.StatusCode, string(buf))
+	}
+
 	var resp PrometheusInstantQueryResponse
 	if err := json.Unmarshal(buf, &resp); err != nil {
 		return nil, httpgrpc.Errorf(http.StatusInternalServerError, "error decoding response: %v", err)

pkg/querier/tripperware/instantquery/instant_query_test.go

@@ -2,6 +2,7 @@ package instantquery
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
 	"fmt"
 	"io"
@@ -12,6 +13,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/weaveworks/common/httpgrpc"
 	"github.com/weaveworks/common/user"
 
 	"github.com/cortexproject/cortex/pkg/querier/tripperware"
@@ -93,6 +95,62 @@ func TestRequest(t *testing.T) {
 	}
 }
 
+func TestGzippedResponse(t *testing.T) {
+	for _, tc := range []struct {
+		body   string
+		status int
+		err    error
+	}{
+		{
+			body:   `{"status":"success","data":{"resultType":"string","result":[1,"foo"]}}`,
+			status: 200,
+		},
+		{
+			body:   `error generic 400`,
+			status: 400,
+			err:    httpgrpc.Errorf(400, "error generic 400"),
+		},
+		{
+			status: 400,
+			err:    httpgrpc.Errorf(400, ""),
+		},
+	} {
+		for _, c := range []bool{true, false} {
+			t.Run(fmt.Sprintf("compressed %t [%s]", c, tc.body), func(t *testing.T) {
+				h := http.Header{
+					"Content-Type": []string{"application/json"},
+				}
+
+				responseBody := bytes.NewBuffer([]byte(tc.body))
+				if c {
+					h.Set("Content-Encoding", "gzip")
+					var buf bytes.Buffer
+					w := gzip.NewWriter(&buf)
+					_, err := w.Write([]byte(tc.body))
+					require.NoError(t, err)
+					w.Close()
+					responseBody = &buf
+				}
+
+				response := &http.Response{
+					StatusCode: tc.status,
+					Header:     h,
+					Body:       io.NopCloser(responseBody),
+				}
+				r, err := InstantQueryCodec.DecodeResponse(context.Background(), response, nil)
+				require.Equal(t, tc.err, err)
+
+				if err == nil {
+					resp, err := json.Marshal(r)
+					require.NoError(t, err)
+
+					require.Equal(t, tc.body, string(resp))
+				}
+			})
+		}
+	}
+}
+
 func TestResponse(t *testing.T) {
 	for i, tc := range []struct {
 		body string

pkg/querier/tripperware/queryrange/query_range.go

@@ -258,10 +258,6 @@ func (prometheusCodec) EncodeRequest(ctx context.Context, r tripperware.Request)
 }
 
 func (prometheusCodec) DecodeResponse(ctx context.Context, r *http.Response, _ tripperware.Request) (tripperware.Response, error) {
-	if r.StatusCode/100 != 2 {
-		body, _ := io.ReadAll(r.Body)
-		return nil, httpgrpc.Errorf(r.StatusCode, string(body))
-	}
 	log, ctx := spanlogger.New(ctx, "ParseQueryRangeResponse") //nolint:ineffassign,staticcheck
 	defer log.Finish()
 
@@ -270,6 +266,9 @@ func (prometheusCodec) DecodeResponse(ctx context.Context, r *http.Response, _ t
 		log.Error(err)
 		return nil, err
 	}
+	if r.StatusCode/100 != 2 {
+		return nil, httpgrpc.Errorf(r.StatusCode, string(buf))
+	}
 	log.LogFields(otlog.Int("bytes", len(buf)))
 
 	var resp PrometheusResponse

pkg/querier/tripperware/queryrange/query_range_test.go

@@ -2,7 +2,9 @@ package queryrange
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
+	"fmt"
 	io "io"
 	"net/http"
 	"strconv"
@@ -657,6 +659,62 @@ func TestMergeAPIResponses(t *testing.T) {
 	}
 }
 
+func TestGzippedResponse(t *testing.T) {
+	for _, tc := range []struct {
+		body   string
+		status int
+		err    error
+	}{
+		{
+			body:   `{"status":"success","data":{"resultType":"matrix","result":[{"metric":{"a":"b","c":"d"},"values":[[2,"2"],[3,"3"]]}],"stats":{"samples":{"totalQueryableSamples":20,"totalQueryableSamplesPerStep":[[2,2],[3,3]]}}}}`,
+			status: 200,
+		},
+		{
+			body:   `error generic 400`,
+			status: 400,
+			err:    httpgrpc.Errorf(400, `error generic 400`),
+		},
+		{
+			status: 400,
+			err:    httpgrpc.Errorf(400, ""),
+		},
+	} {
+		for _, c := range []bool{true, false} {
+			t.Run(fmt.Sprintf("compressed %t [%s]", c, tc.body), func(t *testing.T) {
+				h := http.Header{
+					"Content-Type": []string{"application/json"},
+				}
+
+				responseBody := bytes.NewBuffer([]byte(tc.body))
+				if c {
+					h.Set("Content-Encoding", "gzip")
+					var buf bytes.Buffer
+					w := gzip.NewWriter(&buf)
+					_, err := w.Write([]byte(tc.body))
+					require.NoError(t, err)
+					w.Close()
+					responseBody = &buf
+				}
+
+				response := &http.Response{
+					StatusCode: tc.status,
+					Header:     h,
+					Body:       io.NopCloser(responseBody),
+				}
+				r, err := PrometheusCodec.DecodeResponse(context.Background(), response, nil)
+				require.Equal(t, tc.err, err)
+
+				if err == nil {
+					resp, err := json.Marshal(r)
+					require.NoError(t, err)
+
+					require.Equal(t, tc.body, string(resp))
+				}
+			})
+		}
+	}
+}
+
 func mustParse(t *testing.T, response string) tripperware.Response {
 	var resp PrometheusResponse
 	// Needed as goimports automatically add a json import otherwise.