lib/logstorage: reduce the propbability of improper usage of LogRows.MustAdd() function

The LogRows.MustAdd function was accepting streamFields []Field as the last arg.
VictoriaLogs data model expects that all the stream fields exist in the log fields.
The LogRows.MustAdd() function was easy to misuse by passing streamFields, which are missing in the fields slice.
This could result in violation of VictoriaLogs data model - e.g. there could be stream fields,
which are searchable via {field_name=value} syntax, while aren't searchable via field_name:=value syntax.

Fix this by accepting streamFieldsLen arg instead of streamFields arg at LogRows.MustAdd.
The caller must ensure that the given number of stream fields exist in the beginning of the fields slice.

Author: valyala

app/vlagent/kubernetescollector/processor.go

@@ -36,8 +36,14 @@ type logFileProcessor struct {
 	lr       *logstorage.LogRows
 	tenantID logstorage.TenantID
 
+	// streamFieldsLen is the number of stream fields at the beginning of commonFields.
+	streamFieldsLen int
+
+	// commonFields are common fields for the given log file.
 	commonFields []logstorage.Field
-	streamFields []logstorage.Field
+
+	// fieldsBuf is used for constructing log fields from commonFields and the actual log line fields before sending them to VictoriaLogs.
+	fieldsBuf []logstorage.Field
 
 	// partialCRIContent accumulates the content of partial CRI log lines.
 	// Can be truncated if it exceeds maxLineSize.
@@ -47,13 +53,20 @@ type logFileProcessor struct {
 }
 
 func newLogFileProcessor(storage insertutil.LogRowsStorage, commonFields []logstorage.Field) *logFileProcessor {
-	var streamFields []logstorage.Field
+	// move stream fields to the beginning of commonFields
+
+	streamFields := make([]logstorage.Field, 0, len(commonFields))
 	for _, f := range commonFields {
-		for _, fieldName := range streamFieldNames {
-			if f.Name == fieldName {
-				streamFields = append(streamFields, f)
-				break
-			}
+		if slices.Contains(streamFieldNames, f.Name) {
+			streamFields = append(streamFields, f)
+		}
+	}
+	streamFieldsLen := len(streamFields)
+
+	fields := streamFields
+	for _, f := range commonFields {
+		if !slices.Contains(streamFieldNames, f.Name) {
+			fields = append(fields, f)
 		}
 	}
 
@@ -68,8 +81,8 @@ func newLogFileProcessor(storage insertutil.LogRowsStorage, commonFields []logst
 		lr:       lr,
 		tenantID: tenantID,
 
-		commonFields: commonFields,
-		streamFields: streamFields,
+		streamFieldsLen: streamFieldsLen,
+		commonFields:    fields,
 	}
 }
 
@@ -146,7 +159,7 @@ func (lfp *logFileProcessor) joinPartialLines(criLine criLine) (int64, []byte, b
 	lfp.partialCRIContentSize += len(criLine.content)
 	if lfp.partialCRIContentSize > maxLogLineSize {
 		// Discard the too large log line.
-		reportLogRowSizeExceeded(lfp.streamFields, lfp.partialCRIContentSize)
+		reportLogRowSizeExceeded(lfp.commonFields[:lfp.streamFieldsLen], lfp.partialCRIContentSize)
 
 		lfp.partialCRIContent.Reset()
 		lfp.partialCRIContentSize = 0
@@ -189,8 +202,6 @@ func (lfp *logFileProcessor) addLineInternal(timestamp int64, line []byte) {
 		logstorage.RenameField(parser.Fields, *msgField, "_msg")
 	}
 
-	parser.Fields = append(parser.Fields, lfp.commonFields...)
-
 	if len(parser.Fields) > 1000 {
 		line := logstorage.MarshalFieldsToJSON(nil, parser.Fields)
 		logger.Warnf("dropping log line with %d fields; %s", parser.Fields, line)
@@ -201,7 +212,11 @@ func (lfp *logFileProcessor) addLineInternal(timestamp int64, line []byte) {
 }
 
 func (lfp *logFileProcessor) addRow(timestamp int64, fields []logstorage.Field) {
-	lfp.lr.MustAdd(lfp.tenantID, timestamp, fields, lfp.streamFields)
+	clear(lfp.fieldsBuf)
+	lfp.fieldsBuf = append(lfp.fieldsBuf[:0], lfp.commonFields...)
+	lfp.fieldsBuf = append(lfp.fieldsBuf, fields...)
+
+	lfp.lr.MustAdd(lfp.tenantID, timestamp, lfp.fieldsBuf, lfp.streamFieldsLen)
 	lfp.storage.MustAddRows(lfp.lr)
 	lfp.lr.ResetKeepSettings()
 }

app/vlinsert/datadog/datadog.go

@@ -253,7 +253,7 @@ func readLogsRequest(ts int64, data []byte, lmp insertutil.LogMessageProcessor)
 		if err != nil {
 			return err
 		}
-		lmp.AddRow(ts, fields, nil)
+		lmp.AddRow(ts, fields, -1)
 		fields = fields[:0]
 	}
 	return nil

app/vlinsert/elasticsearch/elasticsearch.go

@@ -207,7 +207,7 @@ func readBulkLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp
 		ts = time.Now().UnixNano()
 	}
 	logstorage.RenameField(p.Fields, msgFields, "_msg")
-	lmp.AddRow(ts, p.Fields, nil)
+	lmp.AddRow(ts, p.Fields, -1)
 
 	return true, nil
 }

app/vlinsert/insertutil/common_params.go

@@ -176,10 +176,10 @@ func CanWriteData() error {
 type LogMessageProcessor interface {
 	// AddRow must add row to the LogMessageProcessor with the given timestamp and fields.
 	//
-	// If streamFields is non-nil, then the given streamFields must be used as log stream fields instead of pre-configured fields.
+	// If streamFieldsLen >= 0, then the given number of initial fields must be used as log stream fields instead of pre-configured fields.
 	//
 	// The LogMessageProcessor implementation cannot hold references to fields, since the caller can reuse them.
-	AddRow(timestamp int64, fields, streamFields []logstorage.Field)
+	AddRow(timestamp int64, fields []logstorage.Field, streamFieldsLen int)
 
 	// MustClose() must flush all the remaining fields and free up resources occupied by LogMessageProcessor.
 	MustClose()
@@ -227,8 +227,9 @@ func (lmp *logMessageProcessor) initPeriodicFlush() {
 
 // AddRow adds new log message to lmp with the given timestamp and fields.
 //
-// If streamFields is non-nil, then it is used as log stream fields instead of the pre-configured stream fields.
-func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []logstorage.Field) {
+// If streamFieldsLen >= 0, then the given number of the initial fields is used as log stream fields
+// instead of the pre-configured stream fields.
+func (lmp *logMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field, streamFieldsLen int) {
 	lmp.rowsIngestedTotal.Inc()
 	n := logstorage.EstimatedJSONRowLen(fields)
 	lmp.bytesIngestedTotal.Add(n)
@@ -243,7 +244,7 @@ func (lmp *logMessageProcessor) AddRow(timestamp int64, fields, streamFields []l
 	lmp.mu.Lock()
 	defer lmp.mu.Unlock()
 
-	lmp.lr.MustAdd(lmp.cp.TenantID, timestamp, fields, streamFields)
+	lmp.lr.MustAdd(lmp.cp.TenantID, timestamp, fields, streamFieldsLen)
 
 	if lmp.cp.Debug {
 		s := lmp.lr.GetRowString(0)

app/vlinsert/insertutil/testutils.go

@@ -15,9 +15,9 @@ type TestLogMessageProcessor struct {
 }
 
 // AddRow adds row with the given timestamp and fields to tlp
-func (tlp *TestLogMessageProcessor) AddRow(timestamp int64, fields, streamFields []logstorage.Field) {
-	if streamFields != nil {
-		panic(fmt.Errorf("BUG: streamFields must be nil; got %v", streamFields))
+func (tlp *TestLogMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field, streamFieldsLen int) {
+	if streamFieldsLen >= 0 {
+		panic(fmt.Errorf("BUG: streamFieldsLen must be negative; got %d", streamFieldsLen))
 	}
 	tlp.timestamps = append(tlp.timestamps, timestamp)
 	tlp.rows = append(tlp.rows, string(logstorage.MarshalFieldsToJSON(nil, fields)))
@@ -48,7 +48,7 @@ func (tlp *TestLogMessageProcessor) Verify(timestampsExpected []int64, resultExp
 type BenchmarkLogMessageProcessor struct{}
 
 // AddRow implements LogMessageProcessor interface.
-func (blp *BenchmarkLogMessageProcessor) AddRow(_ int64, _, _ []logstorage.Field) {
+func (blp *BenchmarkLogMessageProcessor) AddRow(_ int64, _ []logstorage.Field, _ int) {
 }
 
 // MustClose implements LogMessageProcessor interface.

app/vlinsert/journald/journald.go

@@ -249,7 +249,7 @@ func readJournaldLogEntry(streamName string, lr *insertutil.LineReader, lmp inse
 				if ts == 0 {
 					ts = time.Now().UnixNano()
 				}
-				lmp.AddRow(ts, fb.fields, nil)
+				lmp.AddRow(ts, fb.fields, -1)
 			}
 			return nil
 		}

app/vlinsert/jsonline/jsonline.go

@@ -111,7 +111,7 @@ func readLine(lr *insertutil.LineReader, timeFields, msgFields []string, lmp ins
 		return true, fmt.Errorf("%s; line contents: %q", err, line)
 	}
 	logstorage.RenameField(p.Fields, msgFields, "_msg")
-	lmp.AddRow(ts, p.Fields, nil)
+	lmp.AddRow(ts, p.Fields, -1)
 
 	return true, nil
 }

app/vlinsert/loki/loki_json.go

@@ -166,15 +166,16 @@ func parseJSONRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields
 				return fmt.Errorf("unexpected log message type for %q; want string", lineA[1])
 			}
 			allowMsgRenaming := addMsgField(fieldsTmp, msgParser, bytesutil.ToUnsafeString(msg))
-
-			var streamFields []logstorage.Field
-			if useDefaultStreamFields {
-				streamFields = fieldsTmp.Fields[:commonFieldsLen]
-			}
 			if allowMsgRenaming {
 				logstorage.RenameField(fieldsTmp.Fields[commonFieldsLen:], msgFields, "_msg")
 			}
-			lmp.AddRow(ts, fieldsTmp.Fields, streamFields)
+
+			streamFieldsLen := -1
+			if useDefaultStreamFields {
+				streamFieldsLen = commonFieldsLen
+			}
+
+			lmp.AddRow(ts, fieldsTmp.Fields, streamFieldsLen)
 		}
 	}
 

app/vlinsert/loki/loki_protobuf.go

@@ -72,16 +72,15 @@ func parseProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFi
 		}
 
 		allowMsgRenaming := addMsgField(fs, msgParser, line)
-
-		var streamFields []logstorage.Field
-		if useDefaultStreamFields {
-			streamFields = fs.Fields[:streamFieldsLen]
-		}
 		if allowMsgRenaming {
 			logstorage.RenameField(fs.Fields[streamFieldsLen:], msgFields, "_msg")
 		}
 
-		lmp.AddRow(timestamp, fs.Fields, streamFields)
+		if !useDefaultStreamFields {
+			streamFieldsLen = -1
+		}
+
+		lmp.AddRow(timestamp, fs.Fields, streamFieldsLen)
 	}
 
 	if err := decodePushRequest(data, pushLogs); err != nil {

app/vlinsert/loki/loki_protobuf_test.go

@@ -14,9 +14,9 @@ type testLogMessageProcessor struct {
 	pr pushRequest
 }
 
-func (tlp *testLogMessageProcessor) AddRow(timestamp int64, fields, streamFields []logstorage.Field) {
-	if streamFields != nil {
-		panic(fmt.Errorf("unexpected non-nil streamFields: %v", streamFields))
+func (tlp *testLogMessageProcessor) AddRow(timestamp int64, fields []logstorage.Field, streamFieldsLen int) {
+	if streamFieldsLen >= 0 {
+		panic(fmt.Errorf("unexpected positive streamFieldsLen: %d", streamFieldsLen))
 	}
 	msg := ""
 	for _, f := range fields {