lib/logstorage: keep _stream tags in sync with VL-Extra-Fields overrides

docs/victorialogs/CHANGELOG.md

@@ -23,6 +23,7 @@ according to the following docs:
 ## tip
 
 * BUGFIX: [cluster version](https://docs.victoriametrics.com/victorialogs/cluster/): evenly spread rerouted data across available `vlstorage` nodes. Previously, healthy nodes adjacent to unavailable nodes in the `-storageNode` list could receive much more data, resulting in uneven resource usage. See [#1548](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1548).
+* BUGFIX: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/): keep the [`_stream` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) in sync with [extra fields](https://docs.victoriametrics.com/victorialogs/data-ingestion/#http-parameters) passed via the `VL-Extra-Fields` header. Previously, an extra field overriding a [stream field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) value left the original `_stream` tags unchanged. See [#1571](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1571).
 * BUGFIX: [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/) and [querying](https://docs.victoriametrics.com/victorialogs/querying/): properly handle logs containing duplicate [stream field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) names. Previously, [v1.52.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.52.0) could panic when ingesting such logs in single-node VictoriaLogs, drop them during ingestion in VictoriaLogs cluster, or panic when querying such data written by earlier releases. See [#1603](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1603) and [#1604](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1604).
 
 ## [v1.52.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.52.0)

lib/logstorage/log_rows.go

@@ -310,7 +310,7 @@ func (lr *LogRows) MustAddInsertRow(r *InsertRow) {
 		return
 	}
 
-	if st.normalize(r.Fields) {
+	if st.normalize(r.Fields, lr.extraFields) {
 		bLen := len(lr.a.b)
 		lr.a.b = st.MarshalCanonical(lr.a.b)
 		streamTagsCanonical = bytesutil.ToUnsafeString(lr.a.b[bLen:])
@@ -399,7 +399,7 @@ func (lr *LogRows) MustAdd(tenantID TenantID, timestamp int64, fields []Field, s
 					invalidStreamTagsLogger.Warnf("cannot parse _stream=%s: %s; skipping the log entry; log entry: %s", f.Value, err, line)
 					return
 				}
-				st.normalize(fields)
+				st.normalize(fields, lr.extraFields)
 				// Remove _stream field, since it is re-generated from st below.
 				f.Value = ""
 			case "_stream_id":

lib/logstorage/log_rows_timing_test.go

@@ -257,7 +257,7 @@ func BenchmarkStreamTagsNormalize(b *testing.B) {
 		for pb.Next() {
 			for _, c := range casesReal {
 				st.CopyFrom(&c.streamTags)
-				st.normalize(c.fields)
+				st.normalize(c.fields, nil)
 			}
 		}
 	})

lib/logstorage/stream_tags.go

@@ -58,8 +58,8 @@ func (st *StreamTags) String() string {
 
 // normalize synchronizes st with the provided fields and returns true if st was updated.
 //
-// This function updates or keeps tags that exist in fields and removes missing ones.
-func (st *StreamTags) normalize(fields []Field) bool {
+// This function updates or keeps tags that exist in fields or extraFields and removes missing ones.
+func (st *StreamTags) normalize(fields, extraFields []Field) bool {
 	updated := false
 
 	tags := st.tags
@@ -74,6 +74,11 @@ func (st *StreamTags) normalize(fields []Field) bool {
 				// break is skipped intentionally in order to get the last matching field
 			}
 		}
+		for j := range extraFields {
+			if getCanonicalFieldName(extraFields[j].Name) == tagName {
+				f = &extraFields[j]
+			}
+		}
 		if f == nil {
 			updated = true
 			continue

lib/logstorage/stream_tags_test.go

@@ -46,7 +46,7 @@ func TestStreamTagsUnmarshalStringInplace_Failure(t *testing.T) {
 }
 
 func TestStreamTagsNormalize(t *testing.T) {
-	f := func(streamTags, fieldsStr, streamTagsExpected string, isNormalizedExpected bool) {
+	f := func(streamTags, fieldsStr string, extraFields []Field, streamTagsExpected string, isNormalizedExpected bool) {
 		t.Helper()
 
 		st := GetStreamTags()
@@ -59,7 +59,7 @@ func TestStreamTagsNormalize(t *testing.T) {
 		defer putLogfmtParser(p)
 		p.parse(fieldsStr)
 
-		isNormalized := st.normalize(p.fields)
+		isNormalized := st.normalize(p.fields, extraFields)
 
 		result := st.String()
 		if result != streamTagsExpected {
@@ -71,22 +71,26 @@ func TestStreamTagsNormalize(t *testing.T) {
 		}
 	}
 
-	f(`{}`, ``, `{}`, false)
-	f(`{}`, `a=b c=d`, `{}`, false)
-	f(`{a="b"}`, `a=b`, `{a="b"}`, false)
-	f(`{a="b"}`, `x=y a=b q=w`, `{a="b"}`, false)
-	f(`{a="b",c="d"}`, `c=d x=y a=b`, `{a="b",c="d"}`, false)
-	f(`{a="b"}`, `a=b x=y a=b`, `{a="b"}`, false)
+	f(`{}`, ``, nil, `{}`, false)
+	f(`{}`, `a=b c=d`, nil, `{}`, false)
+	f(`{a="b"}`, `a=b`, nil, `{a="b"}`, false)
+	f(`{a="b"}`, `x=y a=b q=w`, nil, `{a="b"}`, false)
+	f(`{a="b",c="d"}`, `c=d x=y a=b`, nil, `{a="b",c="d"}`, false)
+	f(`{a="b"}`, `a=b x=y a=b`, nil, `{a="b"}`, false)
 
 	// missing value
-	f(`{a="b"}`, ``, `{}`, true)
-	f(`{a="b",x="y"}`, `x=y`, `{x="y"}`, true)
+	f(`{a="b"}`, ``, nil, `{}`, true)
+	f(`{a="b",x="y"}`, `x=y`, nil, `{x="y"}`, true)
 
 	// value mismatch
-	f(`{a="b"}`, `a=c`, `{a="c"}`, true)
-	f(`{c="d",a="b"}`, `c=d x=y a=c`, `{a="c",c="d"}`, true)
+	f(`{a="b"}`, `a=c`, nil, `{a="c"}`, true)
+	f(`{c="d",a="b"}`, `c=d x=y a=c`, nil, `{a="c",c="d"}`, true)
 
 	// multiple fields with the same name
-	f(`{a="b"}`, `a=b x=y a=c`, `{a="c"}`, true)
-	f(`{a="b",q="w"}`, `a=c a=c q=w`, `{a="c",q="w"}`, true)
+	f(`{a="b"}`, `a=b x=y a=c`, nil, `{a="c"}`, true)
+	f(`{a="b",q="w"}`, `a=c a=c q=w`, nil, `{a="c",q="w"}`, true)
+
+	// extra fields override the matching fields
+	f(`{a="b"}`, `a=b`, []Field{{Name: "a", Value: "c"}}, `{a="c"}`, true)
+	f(`{a="b"}`, `a=b`, []Field{{Name: "x", Value: "y"}}, `{a="b"}`, false)
 }