stop retrying chunk pool exhaustion at query frontend, retry at queri…

…er level

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

pkg/querier/blocks_store_queryable.go

@@ -3,6 +3,7 @@ package querier
 import (
 	"context"
 	"fmt"
+	"github.com/thanos-io/thanos/pkg/pool"
 	"io"
 	"sort"
 	"strings"
@@ -1113,6 +1114,8 @@ func countSamplesAndChunks(series ...*storepb.Series) (samplesCount, chunksCount
 
 // only retry connection issues
 func isRetryableError(err error) bool {
+	s := status.Convert(err)
+	s.Err()
 	switch status.Code(err) {
 	case codes.Unavailable:
 		return true
@@ -1122,6 +1125,9 @@ func isRetryableError(err error) bool {
 	// https://github.com/grpc/grpc-go/blob/03172006f5d168fc646d87928d85cb9c4a480291/clientconn.go#L67
 	case codes.Canceled:
 		return strings.Contains(err.Error(), "grpc: the client connection is closing")
+	case codes.Unknown:
+		// Catch chunks pool exhaustion error only.
+		return strings.Contains(err.Error(), pool.ErrPoolExhausted.Error())
 	default:
 		return false
 	}

pkg/querier/blocks_store_queryable_test.go

@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/promql-engine/engine"
 	"github.com/thanos-io/promql-engine/logicalplan"
+	"github.com/thanos-io/thanos/pkg/pool"
 	"github.com/thanos-io/thanos/pkg/store/hintspb"
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
@@ -668,6 +669,35 @@ func TestBlocksStoreQuerier_Select(t *testing.T) {
 				},
 			},
 		},
+		"multiple store-gateways has the block, but one of them fails to return due to chunk pool exhaustion": {
+			finderResult: bucketindex.Blocks{
+				{ID: block1},
+			},
+			storeSetResponses: []interface{}{
+				map[BlocksStoreClient][]ulid.ULID{
+					&storeGatewayClientMock{
+						remoteAddr:      "1.1.1.1",
+						mockedSeriesErr: status.Error(codes.Unknown, pool.ErrPoolExhausted.Error()),
+					}: {block1},
+				},
+				map[BlocksStoreClient][]ulid.ULID{
+					&storeGatewayClientMock{remoteAddr: "2.2.2.2", mockedSeriesResponses: []*storepb.SeriesResponse{
+						mockSeriesResponse(labels.Labels{metricNameLabel, series1Label}, minT, 2),
+						mockHintsResponse(block1),
+					}}: {block1},
+				},
+			},
+			limits:       &blocksStoreLimitsMock{},
+			queryLimiter: noOpQueryLimiter,
+			expectedSeries: []seriesResult{
+				{
+					lbls: labels.New(metricNameLabel, series1Label),
+					values: []valueResult{
+						{t: minT, v: 2},
+					},
+				},
+			},
+		},
 		"all store-gateways return PermissionDenied": {
 			finderResult: bucketindex.Blocks{
 				{ID: block1},

pkg/querier/tripperware/queryrange/retry.go

@@ -3,11 +3,14 @@ package queryrange
 import (
 	"context"
 	"errors"
+	"strings"
+	"unsafe"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/thanos-io/thanos/pkg/pool"
 	"github.com/weaveworks/common/httpgrpc"
 
 	"github.com/cortexproject/cortex/pkg/querier/tripperware"
@@ -72,9 +75,9 @@ func (r retry) Do(ctx context.Context, req tripperware.Request) (tripperware.Res
 			return nil, err
 		}
 
-		// Retry if we get a HTTP 500 or a non-HTTP error.
+		// Retry if we get an HTTP 500 or a non-HTTP error.
 		httpResp, ok := httpgrpc.HTTPResponseFromError(err)
-		if !ok || httpResp.Code/100 == 5 {
+		if !ok || isRetryableResponse(httpResp) {
 			lastErr = err
 			level.Error(util_log.WithContext(ctx, r.log)).Log("msg", "error processing request", "try", tries, "err", err)
 			continue
@@ -84,3 +87,20 @@ func (r retry) Do(ctx context.Context, req tripperware.Request) (tripperware.Res
 	}
 	return nil, lastErr
 }
+
+func isRetryableResponse(resp *httpgrpc.HTTPResponse) bool {
+	if resp.Code/100 != 5 {
+		return false
+	}
+
+	// If pool exhausted, retry at query frontend might make things worse.
+	// Rely on retries at querier level only.
+	if strings.Contains(yoloString(resp.Body), pool.ErrPoolExhausted.Error()) {
+		return false
+	}
+	return true
+}
+
+func yoloString(buf []byte) string {
+	return *((*string)(unsafe.Pointer(&buf)))
+}

pkg/querier/tripperware/queryrange/retry_test.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/thanos/pkg/pool"
 	"github.com/weaveworks/common/httpgrpc"
 	"go.uber.org/atomic"
 
@@ -38,16 +39,32 @@ func TestRetry(t *testing.T) {
 		{
 			name: "don't retry 400s",
 			handler: tripperware.HandlerFunc(func(_ context.Context, req tripperware.Request) (tripperware.Response, error) {
+				if try.Inc() == 5 {
+					return &PrometheusResponse{Status: "Hello World"}, nil
+				}
 				return nil, httpgrpc.Errorf(http.StatusBadRequest, "Bad Request")
 			}),
 			err: httpgrpc.Errorf(http.StatusBadRequest, "Bad Request"),
 		},
 		{
 			name: "retry 500s",
 			handler: tripperware.HandlerFunc(func(_ context.Context, req tripperware.Request) (tripperware.Response, error) {
+				if try.Inc() == 5 {
+					return &PrometheusResponse{Status: "Hello World"}, nil
+				}
 				return nil, httpgrpc.Errorf(http.StatusInternalServerError, "Internal Server Error")
 			}),
-			err: httpgrpc.Errorf(http.StatusInternalServerError, "Internal Server Error"),
+			resp: &PrometheusResponse{Status: "Hello World"},
+		},
+		{
+			name: "don't retry chunk pool exhaustion",
+			handler: tripperware.HandlerFunc(func(_ context.Context, req tripperware.Request) (tripperware.Response, error) {
+				if try.Inc() == 5 {
+					return &PrometheusResponse{Status: "Hello World"}, nil
+				}
+				return nil, httpgrpc.Errorf(http.StatusInternalServerError, pool.ErrPoolExhausted.Error())
+			}),
+			err: httpgrpc.Errorf(http.StatusInternalServerError, pool.ErrPoolExhausted.Error()),
 		},
 		{
 			name: "last error",