[coordinator] Make <500 endpoint errors as invalid params error

src/query/server/query_test.go

@@ -237,18 +237,37 @@ tagOptions:
 
 	promReq := test.GeneratePromWriteRequest()
 	promReqBody := test.GeneratePromWriteRequestBody(t, promReq)
-	req, err := http.NewRequestWithContext(
-		context.TODO(),
-		http.MethodPost,
-		fmt.Sprintf("http://%s%s", addr, remote.PromWriteURL),
-		promReqBody,
-	)
-	require.NoError(t, err)
+	requestURL := fmt.Sprintf("http://%s%s", addr, remote.PromWriteURL)
+	newRequest := func() *http.Request {
+		req, err := http.NewRequestWithContext(
+			context.TODO(),
+			http.MethodPost,
+			requestURL,
+			promReqBody,
+		)
+		require.NoError(t, err)
+		return req
+	}
 
-	resp, err := http.DefaultClient.Do(req)
-	require.NoError(t, err)
-	require.NoError(t, resp.Body.Close())
-	assert.NotNil(t, externalFakePromServer.GetLastWriteRequest())
+	t.Run("write request", func(t *testing.T) {
+		defer externalFakePromServer.Reset()
+		resp, err := http.DefaultClient.Do(newRequest())
+		require.NoError(t, err)
+
+		assert.NotNil(t, externalFakePromServer.GetLastWriteRequest())
+		require.NoError(t, resp.Body.Close())
+	})
+
+	t.Run("bad request propagates", func(t *testing.T) {
+		defer externalFakePromServer.Reset()
+		externalFakePromServer.SetError("badRequest", http.StatusBadRequest)
+
+		resp, err := http.DefaultClient.Do(newRequest())
+		require.NoError(t, err)
+
+		assert.Equal(t, 400, resp.StatusCode)
+		require.NoError(t, resp.Body.Close())
+	})
 }
 
 func TestGRPCBackend(t *testing.T) {

src/query/storage/promremote/promremotetest/test_server.go

@@ -37,11 +37,16 @@ import (
 type TestPromServer struct {
 	mu               sync.Mutex
 	lastWriteRequest *prompb.WriteRequest
-	respErr          error
+	respErr          *respErr
 	t                *testing.T
 	svr              *httptest.Server
 }
 
+type respErr struct {
+	error  string
+	status int
+}
+
 // NewServer creates new instance of a fake server.
 func NewServer(t *testing.T) *TestPromServer {
 	testPromServer := &TestPromServer{t: t}
@@ -67,7 +72,7 @@ func (s *TestPromServer) handleWrite(w http.ResponseWriter, r *http.Request) {
 	}
 	s.lastWriteRequest = req
 	if s.respErr != nil {
-		http.Error(w, s.respErr.Error(), http.StatusInternalServerError)
+		http.Error(w, s.respErr.error, s.respErr.status)
 		return
 	}
 }
@@ -85,10 +90,10 @@ func (s *TestPromServer) WriteAddr() string {
 }
 
 // SetError sets error that will be returned for all incoming requests.
-func (s *TestPromServer) SetError(err error) {
+func (s *TestPromServer) SetError(body string, status int) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.respErr = err
+	s.respErr = &respErr{error: body, status: status}
 }
 
 // Reset resets state to default.

src/query/storage/promremote/storage.go

@@ -164,8 +164,14 @@ func (p *promStorage) writeSingle(
 			p.logger.Error("error reading body", zap.Error(err))
 			response = errorReadingBody
 		}
-		return fmt.Errorf("expected status code 2XX: actual=%v, address=%v, resp=%s",
-			resp.StatusCode, address, response)
+		genericError := fmt.Errorf(
+			"expected status code 2XX: actual=%v, address=%v, resp=%s",
+			resp.StatusCode, address, response,
+		)
+		if resp.StatusCode < 500 && resp.StatusCode != http.StatusTooManyRequests {
+			return xerrors.NewInvalidParamsError(genericError)
+		}
+		return genericError
 	}
 	metrics.ReportSuccess(methodDuration)
 	return nil

src/query/storage/promremote/storage_test.go

@@ -22,9 +22,9 @@ package promremote
 
 import (
 	"context"
-	"errors"
 	"io"
 	"math/rand"
+	"net/http"
 	"testing"
 	"time"
 
@@ -39,6 +39,7 @@ import (
 	"github.com/m3db/m3/src/query/storage/m3/storagemetadata"
 	"github.com/m3db/m3/src/query/storage/promremote/promremotetest"
 	"github.com/m3db/m3/src/query/ts"
+	xerrors "github.com/m3db/m3/src/x/errors"
 	"github.com/m3db/m3/src/x/tallytest"
 	xtime "github.com/m3db/m3/src/x/time"
 )
@@ -239,7 +240,7 @@ func TestWriteBasedOnRetention(t *testing.T) {
 
 	t.Run("error should not prevent sending to other instances", func(t *testing.T) {
 		reset()
-		promLongRetention.SetError(errors.New("test err"))
+		promLongRetention.SetError("test err", http.StatusInternalServerError)
 		err := sendWrite(storagemetadata.Attributes{
 			Resolution: 10 * time.Minute,
 			Retention:  8760 * time.Hour,
@@ -248,6 +249,28 @@ func TestWriteBasedOnRetention(t *testing.T) {
 		assert.Contains(t, err.Error(), "test err")
 		assert.NotNil(t, promLongRetention2.GetLastWriteRequest())
 	})
+
+	t.Run("wrap non 5xx errors as invalid params error", func(t *testing.T) {
+		reset()
+		promLongRetention.SetError("test err", http.StatusForbidden)
+		err := sendWrite(storagemetadata.Attributes{
+			Resolution: 10 * time.Minute,
+			Retention:  8760 * time.Hour,
+		})
+		require.Error(t, err)
+		assert.True(t, xerrors.IsInvalidParams(err))
+	})
+
+	t.Run("429 should not be wrapped as invalid params", func(t *testing.T) {
+		reset()
+		promLongRetention.SetError("test err", http.StatusTooManyRequests)
+		err := sendWrite(storagemetadata.Attributes{
+			Resolution: 10 * time.Minute,
+			Retention:  8760 * time.Hour,
+		})
+		require.Error(t, err)
+		assert.False(t, xerrors.IsInvalidParams(err))
+	})
 }
 
 func closeWithCheck(t *testing.T, c io.Closer) {