app/vminsert/influx: allow empty measurement in Influx line protocol

In this case metric names are mapped directly from field names without any prefixes.

app/vminsert/influx/parser.go

@@ -85,9 +85,7 @@ func (r *Row) unmarshal(s string, tagsPool []Tag, fieldsPool []Field, noEscapeCh
 		measurementTags = measurementTags[:n]
 	}
 	r.Measurement = unescapeTagValue(measurementTags, noEscapeChars)
-	if len(r.Measurement) == 0 {
-		return tagsPool, fieldsPool, fmt.Errorf("measurement cannot be empty. measurementTags=%q", s)
-	}
+	// Allow empty r.Measurement. In this case metric name is constructed directly from field keys.
 
 	// Parse fields
 	fieldsStart := len(fieldsPool)

app/vminsert/influx/parser_test.go

@@ -86,9 +86,6 @@ func TestRowsUnmarshalFailure(t *testing.T) {
 		}
 	}
 
-	// Missing measurement
-	f(",foo=bar baz=123")
-
 	// No fields
 	f("foo")
 	f("foo,bar=baz 1234")
@@ -147,6 +144,30 @@ func TestRowsUnmarshalSuccess(t *testing.T) {
 	f("#foobar baz", &Rows{})
 	f("#foobar baz\n#sss", &Rows{})
 
+	// Missing measurement
+	f(" baz=123", &Rows{
+		Rows: []Row{{
+			Measurement: "",
+			Fields: []Field{{
+				Key:   "baz",
+				Value: 123,
+			}},
+		}},
+	})
+	f(",foo=bar baz=123", &Rows{
+		Rows: []Row{{
+			Measurement: "",
+			Tags: []Tag{{
+				Key:   "foo",
+				Value: "bar",
+			}},
+			Fields: []Field{{
+				Key:   "baz",
+				Value: 123,
+			}},
+		}},
+	})
+
 	// Minimal line without tags and timestamp
 	f("foo bar=123", &Rows{
 		Rows: []Row{{

app/vminsert/influx/request_handler.go

@@ -104,7 +104,7 @@ func (ctx *pushCtx) InsertRows(at *auth.Token, db string) error {
 		metricNameBufLen := len(ic.MetricNameBuf)
 		ctx.metricGroupBuf = append(ctx.metricGroupBuf[:0], r.Measurement...)
 		skipFieldKey := len(r.Fields) == 1 && *skipSingleField
-		if !skipFieldKey {
+		if len(ctx.metricGroupBuf) > 0 && !skipFieldKey {
 			ctx.metricGroupBuf = append(ctx.metricGroupBuf, *measurementFieldSeparator...)
 		}
 		metricGroupPrefixLen := len(ctx.metricGroupBuf)