Report bad samples as 400s (ingester)

cmd/cortex/main.go

@@ -197,12 +197,26 @@ func setupDistributor(
 	prometheus.MustRegister(distributor)
 
 	prefix := "/api/prom"
-	http.Handle(prefix+"/push", instrument(logSuccess, cortex.AppenderHandler(distributor)))
+	http.Handle(prefix+"/push", instrument(logSuccess, cortex.AppenderHandler(distributor, handleDistributorError)))
 
 	// TODO: Move querier to separate binary.
 	setupQuerier(distributor, chunkStore, prefix, logSuccess)
 }
 
+func handleDistributorError(w http.ResponseWriter, err error) {
+	switch e := err.(type) {
+	case cortex.IngesterError:
+		switch {
+		case 400 <= e.StatusCode && e.StatusCode < 500:
+			log.Warnf("append err: %v", err)
+			http.Error(w, err.Error(), http.StatusBadRequest)
+			return
+		}
+	}
+	log.Errorf("append err: %v", err)
+	http.Error(w, err.Error(), http.StatusInternalServerError)
+}
+
 // setupQuerier sets up a complete querying pipeline:
 //
 // PromQL -> MergeQuerier -> Distributor -> IngesterQuerier -> Ingester
@@ -256,12 +270,23 @@ func setupIngester(
 	}
 	prometheus.MustRegister(ingester)
 
-	http.Handle("/push", instrument(logSuccess, cortex.AppenderHandler(ingester)))
+	http.Handle("/push", instrument(logSuccess, cortex.AppenderHandler(ingester, handleIngesterError)))
 	http.Handle("/query", instrument(logSuccess, cortex.QueryHandler(ingester)))
 	http.Handle("/label_values", instrument(logSuccess, cortex.LabelValuesHandler(ingester)))
 	return ingester
 }
 
+func handleIngesterError(w http.ResponseWriter, err error) {
+	switch err {
+	case ingester.ErrOutOfOrderSample, ingester.ErrDuplicateSampleForTimestamp:
+		log.Warnf("append err: %v", err)
+		http.Error(w, err.Error(), http.StatusBadRequest)
+	default:
+		log.Errorf("append err: %v", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+	}
+}
+
 // instrument instruments a handler.
 func instrument(logSuccess bool, handler http.Handler) http.Handler {
 	return middleware.Merge(

ingester_client.go

@@ -38,6 +38,23 @@ type IngesterClient struct {
 	timeout time.Duration
 }
 
+// IngesterError is an error we got from an ingester.
+type IngesterError struct {
+	StatusCode int
+	err        error
+}
+
+func errStatusCode(code int, status string) IngesterError {
+	return IngesterError{
+		StatusCode: code,
+		err:        fmt.Errorf("server returned HTTP status %s", status),
+	}
+}
+
+func (i IngesterError) Error() string {
+	return i.err.Error()
+}
+
 // NewIngesterClient makes a new IngesterClient.  This client is careful to
 // propagate the user ID from Distributor -> Ingester.
 func NewIngesterClient(address string, timeout time.Duration) (*IngesterClient, error) {
@@ -185,7 +202,7 @@ func (c *IngesterClient) doRequest(ctx context.Context, endpoint string, req pro
 	}
 	defer httpResp.Body.Close()
 	if httpResp.StatusCode/100 != 2 {
-		return fmt.Errorf("server returned HTTP status %s", httpResp.Status)
+		return errStatusCode(httpResp.StatusCode, httpResp.Status)
 	}
 	if resp == nil {
 		return nil

server.go

@@ -97,7 +97,7 @@ func getSamples(req *remote.WriteRequest) []*model.Sample {
 }
 
 // AppenderHandler returns a http.Handler that accepts proto encoded samples.
-func AppenderHandler(appender SampleAppender) http.Handler {
+func AppenderHandler(appender SampleAppender, errorHandler func(http.ResponseWriter, error)) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		userID := r.Header.Get(userIDHeaderName)
 		if userID == "" {
@@ -122,9 +122,7 @@ func AppenderHandler(appender SampleAppender) http.Handler {
 
 		err = appender.Append(ctx, getSamples(&req))
 		if err != nil {
-			log.Errorf("append err: %v", err)
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
+			errorHandler(w, err)
 		}
 	})
 }