app/vminsert/opentsdb: skip invalid rows and continue parsing the remaining rows

Invalid rows are logged and counted in `vm_rows_invalid_total{type="opentsdb"}` metric

app/vminsert/opentsdb/parser.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/metrics"
 	"github.com/valyala/fastjson/fastfloat"
 )
 
@@ -34,10 +36,8 @@ func (rs *Rows) Reset() {
 // See http://opentsdb.net/docs/build/html/api_telnet/put.html
 //
 // s must be unchanged until rs is in use.
-func (rs *Rows) Unmarshal(s string) error {
+func (rs *Rows) Unmarshal(s string) {
-	var err error
+	rs.Rows, rs.tagsPool = unmarshalRows(rs.Rows[:0], s, rs.tagsPool[:0])
-	rs.Rows, rs.tagsPool, err = unmarshalRows(rs.Rows[:0], s, rs.tagsPool[:0])
-	return err
 }
 
 // Row is a single OpenTSDB row.
@@ -89,41 +89,46 @@ func (r *Row) unmarshal(s string, tagsPool []Tag) ([]Tag, error) {
 	return tagsPool, nil
 }
 
-func unmarshalRows(dst []Row, s string, tagsPool []Tag) ([]Row, []Tag, error) {
+func unmarshalRows(dst []Row, s string, tagsPool []Tag) ([]Row, []Tag) {
 	for len(s) > 0 {
 		n := strings.IndexByte(s, '\n')
-		if n == 0 {
+		if n < 0 {
-			// Skip empty line
+			// The last line.
-			s = s[1:]
+			return unmarshalRow(dst, s, tagsPool)
-			continue
 		}
+		dst, tagsPool = unmarshalRow(dst, s[:n], tagsPool)
+		s = s[n+1:]
+	}
+	return dst, tagsPool
+}
+
+func unmarshalRow(dst []Row, s string, tagsPool []Tag) ([]Row, []Tag) {
+	if len(s) > 0 && s[len(s)-1] == '\r' {
+		s = s[:len(s)-1]
+	}
+	if len(s) == 0 {
+		// Skip empty line
+		return dst, tagsPool
+	}
+
 	if cap(dst) > len(dst) {
 		dst = dst[:len(dst)+1]
 	} else {
 		dst = append(dst, Row{})
 	}
 	r := &dst[len(dst)-1]
-		if n < 0 {
-			// The last line.
 	var err error
 	tagsPool, err = r.unmarshal(s, tagsPool)
 	if err != nil {
-				err = fmt.Errorf("cannot unmarshal OpenTSDB line %q: %s", s, err)
+		dst = dst[:len(dst)-1]
-				return dst, tagsPool, err
+		logger.Errorf("cannot unmarshal OpenTSDB line %q: %s", s, err)
+		invalidLines.Inc()
 	}
-			return dst, tagsPool, nil
+	return dst, tagsPool
-		}
-		var err error
-		tagsPool, err = r.unmarshal(s[:n], tagsPool)
-		if err != nil {
-			err = fmt.Errorf("cannot unmarshal OpenTSDB line %q: %s", s[:n], err)
-			return dst, tagsPool, err
-		}
-		s = s[n+1:]
-	}
-	return dst, tagsPool, nil
 }
 
+var invalidLines = metrics.NewCounter(`vm_rows_invalid_total{type="opentsdb"}`)
+
 func unmarshalTags(dst []Tag, s string) ([]Tag, error) {
 	for {
 		if cap(dst) > len(dst) {

app/vminsert/opentsdb/parser_test.go

@@ -9,13 +9,15 @@ func TestRowsUnmarshalFailure(t *testing.T) {
 	f := func(s string) {
 		t.Helper()
 		var rows Rows
-		if err := rows.Unmarshal(s); err == nil {
+		rows.Unmarshal(s)
-			t.Fatalf("expecting non-nil error when parsing %q", s)
+		if len(rows.Rows) != 0 {
+			t.Fatalf("unexpected number of rows parsed; got %d; want 0", len(rows.Rows))
 		}
 
 		// Try again
-		if err := rows.Unmarshal(s); err == nil {
+		rows.Unmarshal(s)
-			t.Fatalf("expecting non-nil error when parsing %q", s)
+		if len(rows.Rows) != 0 {
+			t.Fatalf("unexpected number of rows parsed; got %d; want 0", len(rows.Rows))
 		}
 	}
 
@@ -51,17 +53,13 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 	f := func(s string, rowsExpected *Rows) {
 		t.Helper()
 		var rows Rows
-		if err := rows.Unmarshal(s); err != nil {
+		rows.Unmarshal(s)
-			t.Fatalf("cannot unmarshal %q: %s", s, err)
-		}
 		if !reflect.DeepEqual(rows.Rows, rowsExpected.Rows) {
 			t.Fatalf("unexpected rows;\ngot\n%+v;\nwant\n%+v", rows.Rows, rowsExpected.Rows)
 		}
 
 		// Try unmarshaling again
-		if err := rows.Unmarshal(s); err != nil {
+		rows.Unmarshal(s)
-			t.Fatalf("cannot unmarshal %q: %s", s, err)
-		}
 		if !reflect.DeepEqual(rows.Rows, rowsExpected.Rows) {
 			t.Fatalf("unexpected rows;\ngot\n%+v;\nwant\n%+v", rows.Rows, rowsExpected.Rows)
 		}
@@ -74,7 +72,9 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 
 	// Empty line
 	f("", &Rows{})
+	f("\r", &Rows{})
 	f("\n\n", &Rows{})
+	f("\n\r\n", &Rows{})
 
 	// Single line
 	f("put foobar 789 -123.456 a=b", &Rows{
@@ -200,4 +200,27 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 			},
 		},
 	})
+	// Multi lines with invalid line
+	f("put foo 2 0.3 a=b\naaa bbb\nput bar.baz 43 0.34 a=b\n", &Rows{
+		Rows: []Row{
+			{
+				Metric:    "foo",
+				Value:     0.3,
+				Timestamp: 2,
+				Tags: []Tag{{
+					Key:   "a",
+					Value: "b",
+				}},
+			},
+			{
+				Metric:    "bar.baz",
+				Value:     0.34,
+				Timestamp: 43,
+				Tags: []Tag{{
+					Key:   "a",
+					Value: "b",
+				}},
+			},
+		},
+	})
 }

app/vminsert/opentsdb/parser_timing_test.go

@@ -16,8 +16,9 @@ put cpu.usage_irq 1234556768 0.34432 a=b
 	b.RunParallel(func(pb *testing.PB) {
 		var rows Rows
 		for pb.Next() {
-			if err := rows.Unmarshal(s); err != nil {
+			rows.Unmarshal(s)
-				panic(fmt.Errorf("cannot unmarshal %q: %s", s, err))
+			if len(rows.Rows) != 4 {
+				panic(fmt.Errorf("unexpected number of parsed rows; got %d; want 4", len(rows.Rows)))
 			}
 		}
 	})

app/vminsert/opentsdb/request_handler.go

@@ -103,11 +103,7 @@ func (ctx *pushCtx) Read(r io.Reader) bool {
 			return false
 		}
 	}
-	if err := ctx.Rows.Unmarshal(bytesutil.ToUnsafeString(ctx.reqBuf)); err != nil {
+	ctx.Rows.Unmarshal(bytesutil.ToUnsafeString(ctx.reqBuf))
-		opentsdbUnmarshalErrors.Inc()
-		ctx.err = fmt.Errorf("cannot unmarshal OpenTSDB put protocol data with size %d: %s", len(ctx.reqBuf), err)
-		return false
-	}
 
 	// Fill in missing timestamps
 	currentTimestamp := time.Now().Unix()
@@ -155,7 +151,6 @@ func (ctx *pushCtx) reset() {
 var (
 	opentsdbReadCalls  = metrics.NewCounter(`vm_read_calls_total{name="opentsdb"}`)
 	opentsdbReadErrors = metrics.NewCounter(`vm_read_errors_total{name="opentsdb"}`)
-	opentsdbUnmarshalErrors = metrics.NewCounter(`vm_unmarshal_errors_total{name="opentsdb"}`)
 )
 
 func getPushCtx() *pushCtx {