From 43a61d87b2492ac11557d28894e04aee887f589b Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Tue, 17 May 2022 15:26:10 -0400 Subject: [PATCH] return HTTP status code: 400 (Bad Request) when ingest values are (#2047) out of range. Previously internal server error was returned. This is to allow for ingest to continue while logging bad values instead of stopping ingest as we do when there is a server error. --- client/batch.go | 4 +++- client/client.go | 6 ++---- field.go | 4 ++-- http_handler.go | 9 ++++++++- 4 files changed, 15 insertions(+), 8 deletions(-) diff --git a/client/batch.go b/client/batch.go index 87a8952ee..4fa326f80 100644 --- a/client/batch.go +++ b/client/batch.go @@ -700,7 +700,9 @@ func (b *Batch) Import() error { // import int data. err = b.doImport(frags, clearFrags) if err != nil { - return errors.Wrap(err, "doing import") + // doImport actually imports the ingested data. An error here, means an error with the value that is trying to be ingested. + // We log the error, so we can continue ingesting. Returning would stop the ingest. + b.log.Printf("error importing batch: %v", err) } b.log.Printf("importing fragments took %v", time.Since(makeTime)) } diff --git a/client/client.go b/client/client.go index 8da2414a3..fd2063eff 100644 --- a/client/client.go +++ b/client/client.go @@ -921,7 +921,6 @@ func (c *Client) doRequest(host *pnet.URI, method, path string, headers map[stri if err != nil { return resp.StatusCode, nil, errors.Wrap(err, "reading response body") } - switch { case resp.StatusCode >= 200 && resp.StatusCode < 300: // [200, 300): OK @@ -944,9 +943,8 @@ func (c *Client) doRequest(host *pnet.URI, method, path string, headers map[stri sleepTime = time.Duration(1< 400 && resp.StatusCode < 500: - // Pilosa nodes sometimes return 400, we retry in that case. - // (400, 500): No need to retry in other 4xx cases. + case resp.StatusCode >= 400 && resp.StatusCode < 500: + // don't retry any 400 level errors return resp.StatusCode, nil, errors.New(strings.TrimSpace(buf.String())) case resp.StatusCode == 503: diff --git a/field.go b/field.go index 5eae96ca6..6bc2f55e7 100644 --- a/field.go +++ b/field.go @@ -1764,9 +1764,9 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, shard for i := range columnIDs { columnID, value := columnIDs[i], values[i] if value > bsig.Max { - return fmt.Errorf("%v, imported value = %v, columnID = %v", ErrBSIGroupValueTooHigh, value, columnID) + return errors.Wrap(ErrBSIGroupValueTooHigh, fmt.Sprintf("value = %v, columnID = %v", value, columnID)) } else if value < bsig.Min { - return fmt.Errorf("%v, imported value = %v, columnID = %v", ErrBSIGroupValueTooLow, value, columnID) + return errors.Wrap(ErrBSIGroupValueTooLow, fmt.Sprintf("value = %v, columnID = %v", value, columnID)) } if value > max { max = value diff --git a/http_handler.go b/http_handler.go index 757c811e5..8d96a9a80 100644 --- a/http_handler.go +++ b/http_handler.go @@ -1133,7 +1133,6 @@ func (h *Handler) handlePostQuery(w http.ResponseWriter, r *http.Request) { var err error err, _ = qerr.(error) - if err != nil || !ok { w.WriteHeader(http.StatusBadRequest) e := h.writeQueryResponse(w, r, &QueryResponse{Err: err}) @@ -2972,6 +2971,8 @@ func (h *Handler) handlePostImportAtomicRecord(w http.ResponseWriter, r *http.Re switch errors.Cause(err) { case ErrClusterDoesNotOwnShard, ErrPreconditionFailed: http.Error(w, err.Error(), http.StatusPreconditionFailed) + case ErrBSIGroupValueTooLow, ErrBSIGroupValueTooHigh, ErrDecimalOutOfRange: + http.Error(w, err.Error(), http.StatusBadRequest) default: http.Error(w, err.Error(), http.StatusInternalServerError) } @@ -3045,9 +3046,12 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { switch errors.Cause(err) { case ErrClusterDoesNotOwnShard, ErrPreconditionFailed: http.Error(w, err.Error(), http.StatusPreconditionFailed) + case ErrBSIGroupValueTooLow, ErrBSIGroupValueTooHigh, ErrDecimalOutOfRange: + http.Error(w, err.Error(), http.StatusBadRequest) default: http.Error(w, err.Error(), http.StatusInternalServerError) } + return } err := qcx.Finish() @@ -3071,6 +3075,8 @@ func (h *Handler) handlePostImport(w http.ResponseWriter, r *http.Request) { switch errors.Cause(err) { case ErrClusterDoesNotOwnShard, ErrPreconditionFailed: http.Error(w, err.Error(), http.StatusPreconditionFailed) + case ErrBSIGroupValueTooLow, ErrBSIGroupValueTooHigh, ErrDecimalOutOfRange: + http.Error(w, err.Error(), http.StatusBadRequest) default: http.Error(w, err.Error(), http.StatusInternalServerError) } @@ -3626,6 +3632,7 @@ func (h *Handler) handleResetIDAlloc(w http.ResponseWriter, r *http.Request) { http.Error(w, "index name is required", http.StatusBadRequest) return } + err := h.api.ResetIDAlloc(indexName) if err != nil { http.Error(w, fmt.Sprintf("resetting ID allocation: %v", err.Error()), http.StatusBadRequest)