[query] Fix query metrics assigning wrong status codes for errors (#2960

)

src/query/api/experimental/annotated/handler.go

@@ -72,16 +72,18 @@ func NewHandler(
 
 func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	if r.Body == nil {
-		h.metrics.writeErrorsClient.Inc(1)
-		xhttp.WriteError(w, errEmptyBody)
+		err := errEmptyBody
+		h.metrics.incError(err)
+		xhttp.WriteError(w, err)
 		return
 	}
 	defer r.Body.Close()
 
 	req, err := parseRequest(r.Body)
 	if err != nil {
-		h.metrics.writeErrorsClient.Inc(1)
-		xhttp.WriteError(w, xhttp.NewError(err, http.StatusBadRequest))
+		resultError := xhttp.NewError(err, http.StatusBadRequest)
+		h.metrics.incError(resultError)
+		xhttp.WriteError(w, resultError)
 		return
 	}
 
@@ -97,20 +99,16 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			break
 		}
 
-		var (
-			status  = http.StatusBadRequest
-			counter = h.metrics.writeErrorsClient
-		)
+		status := http.StatusBadRequest
 		if foundInternalErr {
 			status = http.StatusInternalServerError
-			counter = h.metrics.writeErrorsServer
 		}
-		counter.Inc(1)
 
-		err = fmt.Errorf(
-			"unable to write metric batch, encountered %d errors: %v", len(batchErr.Errors()), batchErr.Error(),
-		)
-		xhttp.WriteError(w, xhttp.NewError(err, status))
+		err = fmt.Errorf("unable to write metric batch, encountered %d errors: %w",
+			len(batchErr.Errors()), batchErr)
+		responseError := xhttp.NewError(err, status)
+		h.metrics.incError(responseError)
+		xhttp.WriteError(w, responseError)
 		return
 	}
 
@@ -150,3 +148,11 @@ func newHandlerMetrics(s tally.Scope) handlerMetrics {
 		writeErrorsClient: s.SubScope("write").Tagged(map[string]string{"code": "4XX"}).Counter("errors"),
 	}
 }
+
+func (m *handlerMetrics) incError(err error) {
+	if xhttp.IsClientError(err) {
+		m.writeErrorsClient.Inc(1)
+	} else {
+		m.writeErrorsServer.Inc(1)
+	}
+}

src/query/api/v1/handler/prometheus/native/read.go

@@ -117,7 +117,7 @@ func (h *promReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	parsedOptions, rErr := ParseRequest(ctx, r, h.instant, h.opts)
 	if rErr != nil {
-		h.promReadMetrics.fetchErrorsClient.Inc(1)
+		h.promReadMetrics.incError(rErr)
 		logger.Error("could not parse request", zap.Error(rErr))
 		xhttp.WriteError(w, rErr)
 		return
@@ -143,7 +143,7 @@ func (h *promReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		logger.Error("range query error",
 			zap.Error(err),
 			zap.Any("parsedOptions", parsedOptions))
-		h.promReadMetrics.fetchErrorsServer.Inc(1)
+		h.promReadMetrics.incError(err)
 
 		xhttp.WriteError(w, err)
 		return

src/query/api/v1/handler/prometheus/native/read_common.go

@@ -35,6 +35,7 @@ import (
 	"github.com/m3db/m3/src/query/storage"
 	"github.com/m3db/m3/src/query/ts"
 	xerrors "github.com/m3db/m3/src/x/errors"
+	xhttp "github.com/m3db/m3/src/x/net/http"
 	xopentracing "github.com/m3db/m3/src/x/opentracing"
 
 	opentracinglog "github.com/opentracing/opentracing-go/log"
@@ -59,6 +60,14 @@ func newPromReadMetrics(scope tally.Scope) promReadMetrics {
 	}
 }
 
+func (m *promReadMetrics) incError(err error) {
+	if xhttp.IsClientError(err) {
+		m.fetchErrorsClient.Inc(1)
+	} else {
+		m.fetchErrorsServer.Inc(1)
+	}
+}
+
 // ReadResponse is the response that gets returned to the user
 type ReadResponse struct {
 	Results []ts.Series `json:"results,omitempty"`

src/query/api/v1/handler/prometheus/remote/read.go

@@ -101,6 +101,14 @@ func newPromReadMetrics(scope tally.Scope) promReadMetrics {
 	}
 }
 
+func (m *promReadMetrics) incError(err error) {
+	if xhttp.IsClientError(err) {
+		m.fetchErrorsClient.Inc(1)
+	} else {
+		m.fetchErrorsServer.Inc(1)
+	}
+}
+
 func (h *promReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	timer := h.promReadMetrics.fetchTimerSuccess.Start()
 	defer timer.Stop()
@@ -109,7 +117,7 @@ func (h *promReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	logger := logging.WithContext(ctx, h.opts.InstrumentOpts())
 	req, fetchOpts, rErr := ParseRequest(ctx, r, h.opts)
 	if rErr != nil {
-		h.promReadMetrics.fetchErrorsClient.Inc(1)
+		h.promReadMetrics.incError(rErr)
 		logger.Error("remote read query parse error",
 			zap.Error(rErr),
 			zap.Any("req", req),
@@ -121,7 +129,7 @@ func (h *promReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	cancelWatcher := handler.NewResponseWriterCanceller(w, h.opts.InstrumentOpts())
 	readResult, err := Read(ctx, cancelWatcher, req, fetchOpts, h.opts)
 	if err != nil {
-		h.promReadMetrics.fetchErrorsServer.Inc(1)
+		h.promReadMetrics.incError(err)
 		logger.Error("remote read query error",
 			zap.Error(err),
 			zap.Any("req", req),
@@ -183,7 +191,7 @@ func (h *promReadHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if err != nil {
-		h.promReadMetrics.fetchErrorsServer.Inc(1)
+		h.promReadMetrics.incError(err)
 	} else {
 		h.promReadMetrics.fetchSuccess.Inc(1)
 	}

src/query/api/v1/handler/prometheus/remote/write.go

@@ -196,6 +196,14 @@ type promWriteMetrics struct {
 	forwardLatency           tally.Histogram
 }
 
+func (m *promWriteMetrics) incError(err error) {
+	if xhttp.IsClientError(err) {
+		m.writeErrorsClient.Inc(1)
+	} else {
+		m.writeErrorsServer.Inc(1)
+	}
+}
+
 func newPromWriteMetrics(scope tally.Scope) (promWriteMetrics, error) {
 	upTo1sBuckets, err := tally.LinearDurationBuckets(0, 100*time.Millisecond, 10)
 	if err != nil {
@@ -263,7 +271,7 @@ func (h *PromWriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	checkedReq, err := h.checkedParseRequest(r)
 	if err != nil {
-		h.metrics.writeErrorsClient.Inc(1)
+		h.metrics.incError(err)
 		xhttp.WriteError(w, err)
 		return
 	}
@@ -359,10 +367,8 @@ func (h *PromWriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		switch {
 		case numBadRequest == len(errs):
 			status = http.StatusBadRequest
-			h.metrics.writeErrorsClient.Inc(1)
 		default:
 			status = http.StatusInternalServerError
-			h.metrics.writeErrorsServer.Inc(1)
 		}
 
 		logger := logging.WithContext(r.Context(), h.instrumentOpts)
@@ -374,20 +380,23 @@ func (h *PromWriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			zap.String("lastRegularError", lastRegularErr),
 			zap.String("lastBadRequestErr", lastBadRequestErr))
 
-		var resultErr string
+		var resultErrMessage string
 		if lastRegularErr != "" {
-			resultErr = fmt.Sprintf("retryable_errors: count=%d, last=%s",
+			resultErrMessage = fmt.Sprintf("retryable_errors: count=%d, last=%s",
 				numRegular, lastRegularErr)
 		}
 		if lastBadRequestErr != "" {
 			var sep string
 			if lastRegularErr != "" {
 				sep = ", "
 			}
-			resultErr = fmt.Sprintf("%s%sbad_request_errors: count=%d, last=%s",
-				resultErr, sep, numBadRequest, lastBadRequestErr)
+			resultErrMessage = fmt.Sprintf("%s%sbad_request_errors: count=%d, last=%s",
+				resultErrMessage, sep, numBadRequest, lastBadRequestErr)
 		}
-		xhttp.WriteError(w, xhttp.NewError(errors.New(resultErr), status))
+
+		resultError := xhttp.NewError(errors.New(resultErrMessage), status)
+		h.metrics.incError(resultError)
+		xhttp.WriteError(w, resultError)
 		return
 	}
 

src/x/net/http/errors.go

@@ -116,3 +116,9 @@ func getStatusCode(err error) int {
 	}
 	return http.StatusInternalServerError
 }
+
+// IsClientError returns true if this error would result in 4xx status code
+func IsClientError(err error) bool {
+	code := getStatusCode(err)
+	return code >= 400 && code < 500
+}

src/x/net/http/errors_test.go

@@ -0,0 +1,54 @@
+// Copyright (c) 2018 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package xhttp
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	xerrors "github.com/m3db/m3/src/x/errors"
+)
+
+func TestIsClientError(t *testing.T) {
+	tests := []struct {
+		err      error
+		expected bool
+	}{
+		{NewError(fmt.Errorf("xhttp.Error(400)"), 400), true},
+		{NewError(fmt.Errorf("xhttp.Error(499)"), 499), true},
+		{xerrors.NewInvalidParamsError(fmt.Errorf("InvalidParamsError")), true},
+		{xerrors.NewRetryableError(xerrors.NewInvalidParamsError(
+			fmt.Errorf("InvalidParamsError insde RetyrableError"))), true},
+
+		{NewError(fmt.Errorf("xhttp.Error(399)"), 399), false},
+		{NewError(fmt.Errorf("xhttp.Error(500)"), 500), false},
+		{xerrors.NewRetryableError(fmt.Errorf("any error inside RetryableError")), false},
+		{fmt.Errorf("any error"), false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.err.Error(), func(t *testing.T) {
+			require.Equal(t, tt.expected, IsClientError(tt.err))
+		})
+	}
+}