Don't retry 400 and 413

plugin/output/elasticsearch/elasticsearch.go

@@ -5,6 +5,7 @@ import (
 	"encoding/base64"
 	"fmt"
 	"net/http"
+	"strconv"
 	"sync"
 	"time"
 
@@ -48,6 +49,7 @@ type Plugin struct {
 
 	// plugin metrics
 	sendErrorMetric      prometheus.Counter
+	sendSemiErrorMetric  *prometheus.CounterVec
 	indexingErrorsMetric prometheus.Counter
 }
 
@@ -282,6 +284,7 @@ func (p *Plugin) Out(event *pipeline.Event) {
 
 func (p *Plugin) registerMetrics(ctl *metric.Ctl) {
 	p.sendErrorMetric = ctl.RegisterCounter("output_elasticsearch_send_error", "Total elasticsearch send errors")
+	p.sendSemiErrorMetric = ctl.RegisterCounterVec("output_elasticsearch_send_semi_error", "Total elasticsearch send semi errors", "status_code")
 	p.indexingErrorsMetric = ctl.RegisterCounter("output_elasticsearch_index_error", "Number of elasticsearch indexing errors")
 }
 
@@ -346,15 +349,23 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err
 		data.outBuf = p.appendEvent(data.outBuf, event)
 	})
 
-	_, err := p.client.DoTimeout(http.MethodPost, NDJSONContentType, data.outBuf,
+	statusCode, err := p.client.DoTimeout(http.MethodPost, NDJSONContentType, data.outBuf,
 		p.config.ConnectionTimeout_, p.reportESErrors)
 
 	if err != nil {
-		p.sendErrorMetric.Inc()
-		p.logger.Error("can't send to the elastic, will try other endpoint", zap.Error(err))
+		switch statusCode {
+		case http.StatusBadRequest, http.StatusRequestEntityTooLarge:
+			p.sendSemiErrorMetric.WithLabelValues(strconv.Itoa(statusCode)).Inc()
+			p.logger.Error("semi error occurred during sending to elastic", zap.Error(err))
+			return nil
+		default:
+			p.sendErrorMetric.Inc()
+			p.logger.Error("can't send to the elastic, will try other endpoint", zap.Error(err))
+			return err
+		}
 	}
 
-	return err
+	return nil
 }
 
 func (p *Plugin) appendEvent(outBuf []byte, event *pipeline.Event) []byte {

xhttp/client.go

@@ -102,7 +102,7 @@ func (c *Client) DoTimeout(
 	respContent := resp.Body()
 	statusCode := resp.Header.StatusCode()
 
-	if statusCode < http.StatusOK || statusCode > http.StatusAccepted {
+	if !(http.StatusOK <= statusCode && statusCode <= http.StatusAccepted) {
 		return statusCode, fmt.Errorf("response status from %s isn't OK: status=%d, body=%s", endpoint.String(), statusCode, string(respContent))
 	}