[pull] master from cortexproject:master

CHANGELOG.md

@@ -63,6 +63,7 @@
 * [ENHANCEMENT] All: Handling CMK Access Denied errors. #5420 #5542
 * [ENHANCEMENT] Querier: Retry store gateway client connection closing gRPC error. #5558
 * [ENHANCEMENT] QueryFrontend: Add generic retry for all APIs. #5561.
+* [ENHANCEMENT] QueryFrontend: Add metric for number of series requests. #5373
 * [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
 * [BUGFIX] Alertmanager: Route web-ui requests to the alertmanager distributor when sharding is enabled. #5293

pkg/distributor/distributor_test.go

@@ -17,7 +17,6 @@ import (
 	"google.golang.org/grpc/codes"
 
 	"github.com/go-kit/log"
-	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/prometheus/common/model"
@@ -284,10 +283,7 @@ func TestDistributor_Push(t *testing.T) {
 				request := makeWriteRequest(tc.samples.startTimestampMs, tc.samples.num, tc.metadata)
 				response, err := ds[0].Push(ctx, request)
 				assert.Equal(t, tc.expectedResponse, response)
-				if err != nil {
-					err = errors.Cause(err)
-				}
-				assert.Equal(t, tc.expectedError, err)
+				assert.Equal(t, status.Code(tc.expectedError), status.Code(err))
 
 				// Check tracked Prometheus metrics. Since the Push() response is sent as soon as the quorum
 				// is reached, when we reach this point the 3rd ingester may not have received series/metadata

pkg/querier/tripperware/roundtrip.go

@@ -127,10 +127,13 @@ func NewQueryTripperware(
 			return RoundTripFunc(func(r *http.Request) (*http.Response, error) {
 				isQuery := strings.HasSuffix(r.URL.Path, "/query")
 				isQueryRange := strings.HasSuffix(r.URL.Path, "/query_range")
+				isSeries := strings.HasSuffix(r.URL.Path, "/series")
 
 				op := "query"
 				if isQueryRange {
 					op = "query_range"
+				} else if isSeries {
+					op = "series"
 				}
 
 				tenantIDs, err := tenant.TenantIDs(r.Context())

pkg/ring/batch.go

@@ -5,10 +5,10 @@ import (
 	"fmt"
 	"sync"
 
+	"go.uber.org/atomic"
 	"google.golang.org/grpc/status"
 
-	"github.com/pkg/errors"
-	"go.uber.org/atomic"
+	"github.com/cortexproject/cortex/pkg/util/httpgrpcutil"
 )
 
 type batchTracker struct {
@@ -152,7 +152,7 @@ func (b *batchTracker) record(instance instance, err error) {
 		if err != nil {
 			// Track the number of errors by error family, and if it exceeds maxFailures
 			// shortcut the waiting rpc.
-			wrappedErr := errors.Wrapf(err, "addr=%s state=%s zone=%s", instance.desc.Addr, instance.desc.State, instance.desc.Zone)
+			wrappedErr := httpgrpcutil.WrapHTTPGrpcError(err, "addr=%s state=%s zone=%s", instance.desc.Addr, instance.desc.State, instance.desc.Zone)
 			errCount := instance.itemTrackers[i].recordError(wrappedErr)
 			// We should return an error if we reach the maxFailure (quorum) on a given error family OR
 			// we dont have any remaining ingesters to try
@@ -162,13 +162,13 @@ func (b *batchTracker) record(instance instance, err error) {
 			// Ex: 5xx, _, 5xx -> return 5xx
 			if errCount > int32(sampleTrackers[i].maxFailures) {
 				if b.rpcsFailed.Inc() == 1 {
-					b.err <- errors.Wrap(sampleTrackers[i].getError(), "maxFailure (quorum) on a given error family")
+					b.err <- httpgrpcutil.WrapHTTPGrpcError(sampleTrackers[i].getError(), "maxFailure (quorum) on a given error family")
 				}
 				continue
 			}
 			if sampleTrackers[i].remaining.Dec() == 0 {
 				if b.rpcsFailed.Inc() == 1 {
-					b.err <- errors.Wrap(sampleTrackers[i].getError(), "not enough remaining instances to try")
+					b.err <- httpgrpcutil.WrapHTTPGrpcError(sampleTrackers[i].getError(), "not enough remaining instances to try")
 				}
 				continue
 			}
@@ -187,7 +187,7 @@ func (b *batchTracker) record(instance instance, err error) {
 			// Ex: 4xx, 5xx, 2xx
 			if sampleTrackers[i].remaining.Dec() == 0 {
 				if b.rpcsFailed.Inc() == 1 {
-					b.err <- errors.Wrap(sampleTrackers[i].getError(), "not enough remaining instances to try")
+					b.err <- httpgrpcutil.WrapHTTPGrpcError(sampleTrackers[i].getError(), "not enough remaining instances to try")
 				}
 			}
 		}

pkg/util/httpgrpcutil/errors.go

@@ -0,0 +1,24 @@
+package httpgrpcutil
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/weaveworks/common/httpgrpc"
+)
+
+func WrapHTTPGrpcError(err error, format string, args ...interface{}) error {
+	if err == nil {
+		return nil
+	}
+	msg := fmt.Sprintf(format, args...)
+	resp, ok := httpgrpc.HTTPResponseFromError(err)
+	if !ok {
+		return httpgrpc.Errorf(http.StatusInternalServerError, "%s, %s", msg, err)
+	}
+	return httpgrpc.ErrorFromHTTPResponse(&httpgrpc.HTTPResponse{
+		Code:    resp.Code,
+		Headers: resp.Headers,
+		Body:    []byte(fmt.Sprintf("%s, %s", msg, err)),
+	})
+}