app/vminsert/graphite: allow skipping timestamps in Graphite plaintext protocol

In this case VictoriaMetrics uses the ingestion time as a timestamp.

app/vminsert/graphite/parser.go

@@ -86,7 +86,9 @@ func (r *Row) unmarshal(s string, tagsPool []Tag) ([]Tag, error) {
 
 	n = strings.IndexByte(tail, ' ')
 	if n < 0 {
-		return tagsPool, fmt.Errorf("cannot find whitespace between value and timestamp in %q", s)
+		// There is no timestamp. Use default timestamp instead.
+		r.Value = fastfloat.ParseBestEffort(tail)
+		return tagsPool, nil
 	}
 	r.Value = fastfloat.ParseBestEffort(tail[:n])
 	r.Timestamp = fastfloat.ParseInt64BestEffort(tail[n+1:])

app/vminsert/graphite/parser_test.go

@@ -22,9 +22,6 @@ func TestRowsUnmarshalFailure(t *testing.T) {
 	// Missing value
 	f("aaa")
 
-	// Missing timestamp
-	f("aaa 1123")
-
 	// Invalid multiline
 	f("aaa\nbbb 123 34")
 
@@ -81,6 +78,14 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 		}},
 	})
 
+	// Missing timestamp
+	f("aaa 1123", &Rows{
+		Rows: []Row{{
+			Metric: "aaa",
+			Value:  1123,
+		}},
+	})
+
 	// Tags
 	f("foo;bar=baz 1 2", &Rows{
 		Rows: []Row{{
@@ -116,13 +121,17 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 	})
 
 	// Multi lines
-	f("foo 0.3 2\nbar.baz 0.34 43\n", &Rows{
+	f("foo 0.3 2\naaa 3\nbar.baz 0.34 43\n", &Rows{
 		Rows: []Row{
 			{
 				Metric:    "foo",
 				Value:     0.3,
 				Timestamp: 2,
 			},
+			{
+				Metric: "aaa",
+				Value:  3,
+			},
 			{
 				Metric:    "bar.baz",
 				Value:     0.34,

app/vminsert/graphite/request_handler.go

@@ -105,10 +105,21 @@ func (ctx *pushCtx) Read(r io.Reader) bool {
 		return false
 	}
 
-	// Convert timestamps from seconds to milliseconds
+	// Fill missing timestamps with the current timestamp rounded to seconds.
-	for i := range ctx.Rows.Rows {
+	currentTimestamp := time.Now().Unix()
-		ctx.Rows.Rows[i].Timestamp *= 1e3
+	rows := ctx.Rows.Rows
+	for i := range rows {
+		r := &rows[i]
+		if r.Timestamp == 0 {
+			r.Timestamp = currentTimestamp
+		}
 	}
+
+	// Convert timestamps from seconds to milliseconds.
+	for i := range rows {
+		rows[i].Timestamp *= 1e3
+	}
+
 	return true
 }