Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1009 (2021-01-12 06:01:34)

Fixes error handling for promscrape.streamParse
properly return error if client cannot read data,
properly suppress scraper errors

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=h1) Report
> Merging [#1009](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=desc) (3d02749) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/89611fa48ce91ea895f330693319728b8a3cecad?el=desc) (89611fa) will **decrease** coverage by `0.14%`.
> The diff coverage is `42.85%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master    #1009      +/-   ##
==========================================
- Coverage   60.57%   60.42%   -0.15%     
==========================================
  Files         202      202              
  Lines       22983    22988       +5     
==========================================
- Hits        13921    13890      -31     
- Misses       7736     7778      +42     
+ Partials     1326     1320       -6     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [lib/promscrape/scrapework.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/diff?src=pr&el=tree#diff-bGliL3Byb21zY3JhcGUvc2NyYXBld29yay5nbw==) | `60.86% <0.00%> (ø)` | |
| [lib/protoparser/prometheus/streamparser.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/diff?src=pr&el=tree#diff-bGliL3Byb3RvcGFyc2VyL3Byb21ldGhldXMvc3RyZWFtcGFyc2VyLmdv) | `68.67% <50.00%> (-1.84%)` | :arrow_down: |
| [lib/mergeset/block\_stream\_reader.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L2Jsb2NrX3N0cmVhbV9yZWFkZXIuZ28=) | `49.30% <0.00%> (-13.20%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `63.44% <0.00%> (-3.08%)` | :arrow_down: |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `63.96% <0.00%> (-0.11%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `87.98% <0.00%> (+3.10%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=footer). Last update [89611fa...3d02749](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/1009?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/app/vmagent/prometheusimport/request_handler.go b/app/vmagent/prometheusimport/request_handler.go
index aac41d944..45daa860b 100644
--- a/app/vmagent/prometheusimport/request_handler.go
+++ b/app/vmagent/prometheusimport/request_handler.go
@@ -31,7 +31,7 @@ func InsertHandler(req *http.Request) error {
 		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
 		return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
 			return insertRows(rows, extraLabels)
-		})
+		}, nil)
 	})
 }
 
diff --git a/app/vminsert/prometheusimport/request_handler.go b/app/vminsert/prometheusimport/request_handler.go
index c499600cd..cc916e8b5 100644
--- a/app/vminsert/prometheusimport/request_handler.go
+++ b/app/vminsert/prometheusimport/request_handler.go
@@ -31,7 +31,7 @@ func InsertHandler(req *http.Request) error {
 		isGzipped := req.Header.Get("Content-Encoding") == "gzip"
 		return parser.ParseStream(req.Body, defaultTimestamp, isGzipped, func(rows []parser.Row) error {
 			return insertRows(rows, extraLabels)
-		})
+		}, nil)
 	})
 }
 
diff --git a/lib/promscrape/scrapework.go b/lib/promscrape/scrapework.go
index 30f28e3a3..6bc351444 100644
--- a/lib/promscrape/scrapework.go
+++ b/lib/promscrape/scrapework.go
@@ -343,7 +343,7 @@ func (sw *scrapeWork) scrapeStream(scrapeTimestamp, realTimestamp int64) error {
 			pushDataDuration.UpdateDuration(startTime)
 			wc.resetNoRows()
 			return nil
-		})
+		}, sw.logError)
 		responseSize = sr.bytesRead
 		sr.MustClose()
 	}
@@ -373,7 +373,7 @@ func (sw *scrapeWork) scrapeStream(scrapeTimestamp, realTimestamp int64) error {
 	wc.reset()
 	writeRequestCtxPool.Put(wc)
 	tsmGlobal.Update(sw.Config, sw.ScrapeGroup, up == 1, realTimestamp, int64(duration*1000), err)
-	return nil
+	return err
 }
 
 // leveledWriteRequestCtxPool allows reducing memory usage when writeRequesCtx
diff --git a/lib/protoparser/prometheus/streamparser.go b/lib/protoparser/prometheus/streamparser.go
index 5e7563a29..91009692f 100644
--- a/lib/protoparser/prometheus/streamparser.go
+++ b/lib/protoparser/prometheus/streamparser.go
@@ -18,7 +18,7 @@ import (
 // The callback can be called concurrently multiple times for streamed data from r.
 //
 // callback shouldn't hold rows after returning.
-func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback func(rows []Row) error) error {
+func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback func(rows []Row) error, errLogger func(string)) error {
 	if isGzipped {
 		zr, err := common.GetGzipReader(r)
 		if err != nil {
@@ -31,6 +31,7 @@ func ParseStream(r io.Reader, defaultTimestamp int64, isGzipped bool, callback f
 	defer putStreamContext(ctx)
 	for ctx.Read() {
 		uw := getUnmarshalWork()
+		uw.errLogger = errLogger
 		uw.callback = func(rows []Row) {
 			if err := callback(rows); err != nil {
 				ctx.callbackErrLock.Lock()
@@ -133,6 +134,7 @@ var streamContextPoolCh = make(chan *streamContext, cgroup.AvailableCPUs())
 type unmarshalWork struct {
 	rows             Rows
 	callback         func(rows []Row)
+	errLogger        func(string)
 	defaultTimestamp int64
 	reqBuf           []byte
 }
@@ -140,13 +142,18 @@ type unmarshalWork struct {
 func (uw *unmarshalWork) reset() {
 	uw.rows.Reset()
 	uw.callback = nil
+	uw.errLogger = nil
 	uw.defaultTimestamp = 0
 	uw.reqBuf = uw.reqBuf[:0]
 }
 
 // Unmarshal implements common.UnmarshalWork
 func (uw *unmarshalWork) Unmarshal() {
-	uw.rows.Unmarshal(bytesutil.ToUnsafeString(uw.reqBuf))
+	if uw.errLogger != nil {
+		uw.rows.UnmarshalWithErrLogger(bytesutil.ToUnsafeString(uw.reqBuf), uw.errLogger)
+	} else {
+		uw.rows.Unmarshal(bytesutil.ToUnsafeString(uw.reqBuf))
+	}
 	rows := uw.rows.Rows
 	rowsRead.Add(len(rows))
 


Test patch:
diff --git a/lib/protoparser/prometheus/streamparser_test.go b/lib/protoparser/prometheus/streamparser_test.go
index d2b88ede6..3e0a0048f 100644
--- a/lib/protoparser/prometheus/streamparser_test.go
+++ b/lib/protoparser/prometheus/streamparser_test.go
@@ -31,7 +31,7 @@ func TestParseStream(t *testing.T) {
 			}
 			lock.Unlock()
 			return nil
-		})
+		}, nil)
 		if err != nil {
 			t.Fatalf("unexpected error when parsing %q: %s", s, err)
 		}
@@ -64,7 +64,7 @@ func TestParseStream(t *testing.T) {
 			}
 			lock.Unlock()
 			return nil
-		})
+		}, nil)
 		if err != nil {
 			t.Fatalf("unexpected error when parsing compressed %q: %s", s, err)
 		}