app/vlinsert: consistently increase vl_http_errors_total on failed inserts

app/vlinsert/datadog/datadog.go

@@ -52,6 +52,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 		var err error
 		ts, err = strconv.ParseInt(tsValue, 10, 64)
 		if err != nil {
+			v2LogsErrorsTotal.Inc()
 			httpserver.Errorf(w, r, "could not parse dd-message-timestamp header value: %s", err)
 			return true
 		}
@@ -62,12 +63,14 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
+		v2LogsErrorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return true
 	}
 
 	if len(cp.StreamFields) == 0 {
 		if err := logstorage.CheckStreamFieldNames(*datadogStreamFields); err != nil {
+			v2LogsErrorsTotal.Inc()
 			httpserver.Errorf(w, r, "invalid stream field names at -datadog.streamFields=%s: %s", datadogStreamFields, err)
 			return true
 		}
@@ -79,6 +82,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 	}
 
 	if err := insertutil.CanWriteData(); err != nil {
+		v2LogsErrorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return true
 	}
@@ -91,6 +95,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 		return err
 	})
 	if err != nil {
+		v2LogsErrorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot read DataDog protocol data: %s", err)
 		return true
 	}
@@ -106,6 +111,7 @@ func datadogLogsIngestion(w http.ResponseWriter, r *http.Request) bool {
 
 var (
 	v2LogsRequestsTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/datadog/api/v2/logs"}`)
+	v2LogsErrorsTotal     = metrics.NewCounter(`vl_http_errors_total{path="/insert/datadog/api/v2/logs"}`)
 	v2LogsRequestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/datadog/api/v2/logs"}`)
 )
 

app/vlinsert/elasticsearch/elasticsearch.go

@@ -99,10 +99,12 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 
 		cp, err := insertutil.GetCommonParams(r)
 		if err != nil {
+			bulkErrorsTotal.Inc()
 			httpserver.Errorf(w, r, "%s", err)
 			return true
 		}
 		if err := insertutil.CanWriteData(); err != nil {
+			bulkErrorsTotal.Inc()
 			httpserver.Errorf(w, r, "%s", err)
 			return true
 		}
@@ -112,6 +114,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 		n, err := readBulkRequest(streamName, r.Body, encoding, cp.TimeFields, cp.MsgFields, cp.PreserveJSONKeys, lmp)
 		lmp.MustClose()
 		if err != nil {
+			bulkErrorsTotal.Inc()
 			httpserver.Errorf(w, r, "cannot decode log message #%d in /_bulk request: %s, stream fields: %s", n, err, cp.StreamFields)
 			return true
 		}
@@ -135,6 +138,7 @@ func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 
 var (
 	bulkRequestsTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/elasticsearch/_bulk"}`)
+	bulkErrorsTotal     = metrics.NewCounter(`vl_http_errors_total{path="/insert/elasticsearch/_bulk"}`)
 	bulkRequestDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/elasticsearch/_bulk"}`)
 )
 

app/vlinsert/internalinsert/internalinsert.go

@@ -37,10 +37,12 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}

app/vlinsert/jsonline/jsonline.go

@@ -30,16 +30,19 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 
 	wcr, err := writeconcurrencylimiter.GetReader(r.Body)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot start reading jsonline request: %s", err)
 		return
 	}
@@ -48,6 +51,7 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 	encoding := r.Header.Get("Content-Encoding")
 	reader, err := protoparserutil.GetUncompressedReader(wcr, encoding)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot decode jsonline request: %s", err)
 		return
 	}

app/vlinsert/loki/loki_json.go

@@ -29,10 +29,12 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 
 	cp, err := getCommonParams(r)
 	if err != nil {
+		errorsJSONTotal.Inc()
 		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsJSONTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
@@ -46,6 +48,7 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 		return err
 	})
 	if err != nil {
+		errorsJSONTotal.Inc()
 		httpserver.Errorf(w, r, "cannot read Loki json data: %s", err)
 		return
 	}
@@ -61,6 +64,7 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 
 var (
 	requestsJSONTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="json"}`)
+	errorsJSONTotal     = metrics.NewCounter(`vl_http_errors_total{path="/insert/loki/api/v1/push",format="json"}`)
 	requestJSONDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="json"}`)
 )
 

app/vlinsert/loki/loki_protobuf.go

@@ -19,10 +19,12 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 
 	cp, err := getCommonParams(r)
 	if err != nil {
+		errorsProtobufTotal.Inc()
 		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsProtobufTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
@@ -41,6 +43,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 		return err
 	})
 	if err != nil {
+		errorsProtobufTotal.Inc()
 		httpserver.Errorf(w, r, "cannot read Loki protobuf data: %s", err)
 		return
 	}
@@ -56,6 +59,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 
 var (
 	requestsProtobufTotal   = metrics.NewCounter(`vl_http_requests_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
+	errorsProtobufTotal     = metrics.NewCounter(`vl_http_errors_total{path="/insert/loki/api/v1/push",format="protobuf"}`)
 	requestProtobufDuration = metrics.NewSummary(`vl_http_request_duration_seconds{path="/insert/loki/api/v1/push",format="protobuf"}`)
 )
 

app/vlinsert/nativeinsert/nativeinsert.go

@@ -39,10 +39,12 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}

app/vlinsert/nativeinsert/nativemultitenant/multitenant.go

@@ -33,10 +33,12 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) {
 
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}

app/vlinsert/opentelemetry/opentelemetry.go

@@ -40,10 +40,12 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
@@ -57,6 +59,7 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 		return err
 	})
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot read OpenTelemetry protocol data: %s", err)
 		return
 	}
@@ -86,7 +89,6 @@ func pushProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFie
 	}
 
 	if err := decodeLogsData(data, pushLogs); err != nil {
-		errorsTotal.Inc()
 		return fmt.Errorf("cannot decode LogsData request from %d bytes: %w", len(data), err)
 	}
 	return nil

app/vlinsert/splunk/splunk.go

@@ -135,10 +135,12 @@ func requestHandler(w http.ResponseWriter, r *http.Request) {
 
 	cp, err := getCommonParams(r)
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 	if err := insertutil.CanWriteData(); err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
@@ -150,6 +152,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) {
 		return processEvent(data, lmp, cp.TimeFields, cp.MsgFields, cp.PreserveJSONKeys)
 	})
 	if err != nil {
+		errorsTotal.Inc()
 		httpserver.Errorf(w, r, "cannot read Splunk request: %s", err)
 		return
 	}
@@ -177,8 +180,8 @@ func processEvent(data []byte, lmp insertutil.LogMessageProcessor, timeFields, m
 		n++
 	}
 	if err := s.Error(); err != nil {
-		errorsTotal.Add(1)
 		if n > 0 {
+			errorsTotal.Add(1)
 			logger.Warnf("splunk: failed to parse JSON message #%d: %s", n+1, err)
 			return nil
 		}

docs/victorialogs/CHANGELOG.md

@@ -59,6 +59,7 @@ according to the following docs:
 * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): evenly spread rerouted data across available `vlstorage` nodes. Previously, healthy nodes adjacent to unavailable nodes in the `-storageNode` list could receive much more data, resulting in uneven resource usage. See [#1548](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1548).
 * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): properly cancel queries already running on `vlstorage` when the corresponding query is canceled or times out on `vlselect`. Previously, `vlstorage` could fail to detect the disconnected `vlselect`, causing these queries to continue running and waste CPU and disk IO. See [#1672](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1672).
 * BUGFIX: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/) and [querying](https://docs.victoriametrics.com/victorialogs/querying/): properly handle logs containing duplicate [stream field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) names. Previously, [v1.52.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.52.0) could panic when ingesting such logs in single-node VictoriaLogs, drop them during ingestion in VictoriaLogs cluster, or panic when querying such data written by earlier releases. See [#1603](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1603) and [#1604](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1604).
+* BUGFIX: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/): increment [`vl_http_errors_total`](https://docs.victoriametrics.com/victorialogs/metrics/#vl_http_errors_total) for failed insert requests, such as requests with invalid params or requests rejected because the storage is in read-only mode (e.g. because of lack of free disk space). Also expose the missing `vl_http_errors_total` for Loki, Elasticsearch and DataDog ingestion endpoints. See [#1811](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1811).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): fix [`week_range[Sun,Sun]` filter](https://docs.victoriametrics.com/victorialogs/logsql/#week-range-filter) when it is used inside the [`filter` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#filter-pipe). Previously, it could fail to match rows on Sunday. See [#1335](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1335).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): properly execute queries returning the last `N` logs (such as queries ending with [`| sort by (_time desc) limit N`](https://docs.victoriametrics.com/victorialogs/logsql/#sort-pipe)) when they contain pipes writing to the `_time` field, such as `math ... as _time`, `replace (...) at _time` or `extract "<_time>"`. Previously such queries could fail with the `missing _time field in the query results` error or return logs in the wrong order when they were executed on wide time ranges. See [#1727](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1727).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): fix [`sort by (_time) limit N`](https://docs.victoriametrics.com/victorialogs/logsql/#sort-pipe) returning logs out of order when the query pipeline included pipes like [`unpack_json`](https://docs.victoriametrics.com/victorialogs/logsql/#unpack_json-pipe) that overwrite `_time`. See [#1360](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1360).

docs/victorialogs/metrics.md

@@ -52,7 +52,16 @@ These metrics follow the Prometheus exposition format and can be used for monito
 - `path`: endpoint path
 - `format`: request format when applicable (e.g. `protobuf`)
 
-**Description:** Errors encountered while processing requests for the given endpoint. The counter is incremented by endpoint handlers on non-trivial processing errors (e.g. request decoding/parsing failures or query execution errors) and may differ from the number of HTTP responses with error status. For line-oriented ingestion (e.g. `/insert/jsonline`), it is incremented per invalid log line inside a request.
+**Description:** The number of errors at the given endpoint.
+
+For data ingestion endpoints, the counter is incremented in two cases:
+
+- A request is counted at `vl_http_requests_total` and then rejected, for example because of invalid request args, broken request body or because the storage is in read-only mode. Requests rejected earlier, such as requests with unsupported HTTP method, content type or protocol version, are counted neither at `vl_http_requests_total` nor at `vl_http_errors_total`.
+- A request is accepted, but some of its log lines or events are invalid and skipped. In this case the counter is incremented per every skipped log line or event, e.g. at `/insert/jsonline`.
+
+So the counter can be bigger than the number of failed requests.
+
+For `/internal/select/*` endpoints, the counter is incremented per every failed query, except for queries canceled by the client.
 
 ### vl_http_request_duration_seconds
 **Type:** Summary