app/vlinsert/opentelemetry: parse ScopeLogs.scope field and store its contents as log fields with scope. prefix

See #826

Author: valyala

app/vlinsert/opentelemetry/opentelemetry_test.go

@@ -54,6 +54,30 @@ func TestPushProtobufRequest(t *testing.T) {
 	resultsExpected := `{"_msg":"log-line-message","severity":"Trace"}`
 	f(data, timestampsExpected, resultsExpected)
 
+	// single line with scope attributes
+	data = `[{
+		"scopeLogs": [{
+			"scope": {
+				"name": "foo",
+				"version": "v1.234.5",
+				"attributes": [
+					{"key":"abc","value":{"stringValue":"de"}},
+					{"key":"x","value":{"stringValue":"aaa"}}
+				]
+			},
+			"logRecords": [{
+				"timeUnixNano": 1234,
+				"severityNumber": 1,
+				"body": {
+					"stringValue": "log-line-message"
+				}
+			}]
+		}]
+	}]`
+	timestampsExpected = []int64{1234}
+	resultsExpected = `{"scope.name":"foo","scope.version":"v1.234.5","scope.attributes.abc":"de","scope.attributes.x":"aaa","_msg":"log-line-message","severity":"Trace"}`
+	f(data, timestampsExpected, resultsExpected)
+
 	// severities mapping
 	data = `[{
 		"scopeLogs": [{
@@ -110,6 +134,11 @@ func TestPushProtobufRequest(t *testing.T) {
 			]
 		},
 		"scopeLogs": [{
+			"scope": {
+				"attributes": [
+					{"key":"abc","value":{"stringValue":"de"}}
+				]
+			},
 			"logRecords": [
 				{"timeUnixNano":1234,"severityNumber":1,"body":{"stringValue":"log-line-message"}},
 				{"timeUnixNano":1235,"severityNumber":5,"body":{"stringValue":"log-line-message-msg-2"}}
@@ -133,8 +162,8 @@ func TestPushProtobufRequest(t *testing.T) {
 		]
 	}]`
 	timestampsExpected = []int64{1234, 1235, 2345, 2346, 2347, 2348, 3333, 432}
-	resultsExpected = `{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message","severity":"Trace"}
-{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","_msg":"log-line-message-msg-2","severity":"Debug"}
+	resultsExpected = `{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","scope.name":"unknown","scope.version":"unknown","scope.attributes.abc":"de","_msg":"log-line-message","severity":"Trace"}
+{"logger":"context","instance_id":"10","node_taints.role":"dev","node_taints.cluster_load_percent":"0.55","scope.name":"unknown","scope.version":"unknown","scope.attributes.abc":"de","_msg":"log-line-message-msg-2","severity":"Debug"}
 {"_msg":"log-line-resource-scope-1-0-0","severity":"Info2"}
 {"_msg":"log-line-resource-scope-1-0-1","severity":"Info2"}
 {"_msg":"log-line-resource-scope-1-1-0","severity":"Info4"}
@@ -437,15 +466,39 @@ func (kvl *keyValueList) marshalProtobuf(mm *easyproto.MessageMarshaler) {
 
 // scopeLogs represents the corresponding OTEL protobuf message
 type scopeLogs struct {
-	LogRecords []logRecord `json:"logRecords,omitzero"`
+	Scope      *instrumentationScope `json:"scope,omitzero"`
+	LogRecords []logRecord           `json:"logRecords,omitzero"`
 }
 
 func (sl *scopeLogs) marshalProtobuf(mm *easyproto.MessageMarshaler) {
+	if sl.Scope != nil {
+		sl.Scope.marshalProtobuf(mm.AppendMessage(1))
+	}
 	for _, m := range sl.LogRecords {
 		m.marshalProtobuf(mm.AppendMessage(2))
 	}
 }
 
+// instrumentationScope represents the corresponding OTEL protobuf message.
+// See https://github.com/open-telemetry/opentelemetry-proto/blob/a5f0eac5b802f7ae51dfe41e5116fe5548955e64/opentelemetry/proto/common/v1/common.proto#L76
+type instrumentationScope struct {
+	Name       string      `json:"name,omitzero"`
+	Version    string      `json:"version,omitzero"`
+	Attributes []*keyValue `json:"attributes,omitzero"`
+}
+
+func (s *instrumentationScope) marshalProtobuf(mm *easyproto.MessageMarshaler) {
+	if s.Name != "" {
+		mm.AppendString(1, s.Name)
+	}
+	if s.Version != "" {
+		mm.AppendString(2, s.Version)
+	}
+	for _, m := range s.Attributes {
+		m.marshalProtobuf(mm.AppendMessage(3))
+	}
+}
+
 // logRecord represents the corresponding OTEL protobuf message.
 // See https://github.com/open-telemetry/oteps/blob/main/text/logs/0097-log-data-model.md
 type logRecord struct {

app/vlinsert/opentelemetry/pb.go

@@ -5,6 +5,8 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+
 	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
 	"github.com/VictoriaMetrics/easyproto"
 )
@@ -57,12 +59,12 @@ func decodeResourceLogs(src []byte, pushLogs pushLogsHandler) (err error) {
 	defer logstorage.PutFields(fs)
 
 	// Decode resource
-	data, ok, err := easyproto.GetMessageData(src, 1)
+	resourceData, ok, err := easyproto.GetMessageData(src, 1)
 	if err != nil {
 		return fmt.Errorf("cannot find Resource: %w", err)
 	}
 	if ok {
-		if err = decodeResource(data, fs, fb); err != nil {
+		if err = decodeResource(resourceData, fs, fb); err != nil {
 			return fmt.Errorf("cannot decode Resource: %w", err)
 		}
 	}
@@ -120,6 +122,7 @@ func decodeResource(src []byte, fs *logstorage.Fields, fb *fmtBuffer) (err error
 
 func decodeScopeLogs(src []byte, fs *logstorage.Fields, pushLogs pushLogsHandler) (err error) {
 	// message ScopeLogs {
+	//   InstrumentationScope scope = 1;
 	//   repeated LogRecord log_records = 2;
 	// }
 
@@ -128,6 +131,18 @@ func decodeScopeLogs(src []byte, fs *logstorage.Fields, pushLogs pushLogsHandler
 
 	streamFieldsLen := len(fs.Fields)
 
+	scopeData, ok, err := easyproto.GetMessageData(src, 1)
+	if err != nil {
+		return fmt.Errorf("cannot read InstrumentationScope: %w", err)
+	}
+	if ok {
+		if err := decodeInstrumentationScope(scopeData, fs, fb); err != nil {
+			return fmt.Errorf("cannot decode InstrumentationScope: %w", err)
+		}
+	}
+
+	commonFieldsLen := len(fs.Fields)
+
 	var fc easyproto.FieldContext
 	for len(src) > 0 {
 		src, err = fc.NextField(src)
@@ -142,7 +157,7 @@ func decodeScopeLogs(src []byte, fs *logstorage.Fields, pushLogs pushLogsHandler
 			}
 
 			fb.reset()
-			fs.Fields = fs.Fields[:streamFieldsLen]
+			fs.Fields = fs.Fields[:commonFieldsLen]
 
 			eventName, timestamp, err := decodeLogRecord(data, fs, fb)
 			if err != nil {
@@ -159,6 +174,9 @@ func decodeScopeLogs(src []byte, fs *logstorage.Fields, pushLogs pushLogsHandler
 				f.Value = eventName
 
 				pushLogs(timestamp, fs.Fields, streamFieldsLen+1)
+
+				// Return back common fields to their places before the next iteration
+				fs.Fields = append(fs.Fields[:streamFieldsLen], fs.Fields[streamFieldsLen+1:commonFieldsLen+1]...)
 			} else {
 				pushLogs(timestamp, fs.Fields, streamFieldsLen)
 			}
@@ -167,6 +185,56 @@ func decodeScopeLogs(src []byte, fs *logstorage.Fields, pushLogs pushLogsHandler
 	return nil
 }
 
+func decodeInstrumentationScope(src []byte, fs *logstorage.Fields, fb *fmtBuffer) error {
+	// See https://github.com/open-telemetry/opentelemetry-proto/blob/a5f0eac5b802f7ae51dfe41e5116fe5548955e64/opentelemetry/proto/common/v1/common.proto#L76
+	//
+	// message InstrumentationScope {
+	//   string name = 1;
+	//   string version = 2;
+	//   repeated KeyValue attributes = 3;
+	// }
+
+	nameData, ok, err := easyproto.GetMessageData(src, 1)
+	if err != nil {
+		return fmt.Errorf("cannot read name: %w", err)
+	}
+	name := "unknown"
+	if ok {
+		name = bytesutil.ToUnsafeString(nameData)
+	}
+	fs.Add("scope.name", name)
+
+	versionData, ok, err := easyproto.GetMessageData(src, 2)
+	if err != nil {
+		return fmt.Errorf("cannot read version: %w", err)
+	}
+	version := "unknown"
+	if ok {
+		version = bytesutil.ToUnsafeString(versionData)
+	}
+	fs.Add("scope.version", version)
+
+	var fc easyproto.FieldContext
+	for len(src) > 0 {
+		src, err = fc.NextField(src)
+		if err != nil {
+			return fmt.Errorf("cannot read the next field: %w", err)
+		}
+		switch fc.FieldNum {
+		case 3:
+			attributesData, ok := fc.MessageData()
+			if !ok {
+				return fmt.Errorf("cannot read Attributes data")
+			}
+			if err := decodeKeyValue(attributesData, fs, fb, "scope.attributes"); err != nil {
+				return fmt.Errorf("cannot decode Attributes: %w", err)
+			}
+		}
+	}
+
+	return nil
+}
+
 func decodeLogRecord(src []byte, fs *logstorage.Fields, fb *fmtBuffer) (string, int64, error) {
 	// See https://github.com/open-telemetry/opentelemetry-proto/blob/a5f0eac5b802f7ae51dfe41e5116fe5548955e64/opentelemetry/proto/logs/v1/logs.proto#L136
 	//
@@ -228,11 +296,11 @@ func decodeLogRecord(src []byte, fs *logstorage.Fields, fb *fmtBuffer) (string,
 				return "", 0, fmt.Errorf("cannot decode Body: %w", err)
 			}
 		case 6:
-			data, ok := fc.MessageData()
+			attributesData, ok := fc.MessageData()
 			if !ok {
 				return "", 0, fmt.Errorf("cannot read Attributes data")
 			}
-			if err := decodeKeyValue(data, fs, fb, ""); err != nil {
+			if err := decodeKeyValue(attributesData, fs, fb, ""); err != nil {
 				return "", 0, fmt.Errorf("cannot decode Attributes: %w", err)
 			}
 		case 9:
@@ -282,17 +350,17 @@ func decodeKeyValue(src []byte, fs *logstorage.Fields, fb *fmtBuffer, fieldNameP
 	// }
 
 	// Decode key
-	data, ok, err := easyproto.GetMessageData(src, 1)
+	keyData, ok, err := easyproto.GetMessageData(src, 1)
 	if err != nil {
 		return fmt.Errorf("cannot find Key in KeyValue: %w", err)
 	}
 	if !ok {
 		return fmt.Errorf("key is missing in KeyValue")
 	}
-	fieldName := fb.formatSubFieldName(fieldNamePrefix, data)
+	fieldName := fb.formatSubFieldName(fieldNamePrefix, keyData)
 
 	// Decode value
-	data, ok, err = easyproto.GetMessageData(src, 2)
+	valueData, ok, err := easyproto.GetMessageData(src, 2)
 	if err != nil {
 		return fmt.Errorf("cannot find Value in KeyValue: %w", err)
 	}
@@ -301,7 +369,7 @@ func decodeKeyValue(src []byte, fs *logstorage.Fields, fb *fmtBuffer, fieldNameP
 		return nil
 	}
 
-	if err := decodeAnyValue(data, fs, fb, fieldName); err != nil {
+	if err := decodeAnyValue(valueData, fs, fb, fieldName); err != nil {
 		return fmt.Errorf("cannot decode AnyValue: %w", err)
 	}
 

docs/victorialogs/CHANGELOG.md

@@ -21,6 +21,8 @@ according to the follosing docs:
 
 ## tip
 
+* FEATURE: [OpenTelemetry data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/opentelemetry/): parse [`scope` inside ScopeLogs](https://github.com/open-telemetry/opentelemetry-proto/blob/a5f0eac5b802f7ae51dfe41e5116fe5548955e64/opentelemetry/proto/logs/v1/logs.proto#L72) and store it into [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) starting with `scope.` prefix. See [#826](https://github.com/VictoriaMetrics/VictoriaLogs/issues/826).
+
 ## [v1.40.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.40.0)
 
 Released at 2025-12-05