Implement unmarshal Logs with jsoniter

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 - Fix reading resource attributes for trace JSON, remove duplicate code. (#6023)
+- Add support to unmarshalls bytes into plogs.Logs with `jsoniter` in jsonUnmarshaler(#5935)
+
 
 ### 🛑 Breaking changes 🛑
 

pdata/internal/json/attribute.go

@@ -32,7 +32,7 @@ func ReadAttribute(iter *jsoniter.Iterator) otlpcommon.KeyValue {
 			kv.Key = iter.ReadString()
 		case "value":
 			iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
-				kv.Value = readAnyValue(iter, f)
+				kv.Value = ReadAnyValue(iter, f)
 				return true
 			})
 		default:
@@ -43,7 +43,8 @@ func ReadAttribute(iter *jsoniter.Iterator) otlpcommon.KeyValue {
 	return kv
 }
 
-func readAnyValue(iter *jsoniter.Iterator, f string) otlpcommon.AnyValue {
+// ReadAnyValue Unmarshal JSON data and return otlpcommon.AnyValue
+func ReadAnyValue(iter *jsoniter.Iterator, f string) otlpcommon.AnyValue {
 	switch f {
 	case "stringValue", "string_value":
 		return otlpcommon.AnyValue{
@@ -104,7 +105,7 @@ func readArray(iter *jsoniter.Iterator) *otlpcommon.ArrayValue {
 		case "values":
 			iter.ReadArrayCB(func(iter *jsoniter.Iterator) bool {
 				iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
-					v.Values = append(v.Values, readAnyValue(iter, f))
+					v.Values = append(v.Values, ReadAnyValue(iter, f))
 					return true
 				})
 				return true

pdata/internal/json/attribute_test.go

@@ -170,7 +170,7 @@ func TestReadAnyValueUnknownField(t *testing.T) {
 	jsonStr := `{"extra":""}`
 	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
 	defer jsoniter.ConfigFastest.ReturnIterator(iter)
-	value := readAnyValue(iter, "")
+	value := ReadAnyValue(iter, "")
 	assert.NoError(t, iter.Error)
 	assert.EqualValues(t, otlpcommon.AnyValue{}, value)
 }
@@ -179,7 +179,7 @@ func TestReadAnyValueInvliadBytesValue(t *testing.T) {
 	jsonStr := `"--"`
 	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
 	defer jsoniter.ConfigFastest.ReturnIterator(iter)
-	readAnyValue(iter, "bytesValue")
+	ReadAnyValue(iter, "bytesValue")
 	if assert.Error(t, iter.Error) {
 		assert.Contains(t, iter.Error.Error(), "base64")
 	}
@@ -208,7 +208,7 @@ func TestReadArrayValueeInvalidArrayValue(t *testing.T) {
 	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
 	defer jsoniter.ConfigFastest.ReturnIterator(iter)
 
-	value := readAnyValue(iter, "arrayValue")
+	value := ReadAnyValue(iter, "arrayValue")
 	assert.NoError(t, iter.Error)
 	assert.EqualValues(t, otlpcommon.AnyValue{
 		Value: &otlpcommon.AnyValue_ArrayValue{
@@ -221,7 +221,7 @@ func TestReadKvlistValueInvalidArrayValue(t *testing.T) {
 	jsonStr := `{"extra":""}`
 	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
 	defer jsoniter.ConfigFastest.ReturnIterator(iter)
-	value := readAnyValue(iter, "kvlistValue")
+	value := ReadAnyValue(iter, "kvlistValue")
 	assert.NoError(t, iter.Error)
 	assert.EqualValues(t, otlpcommon.AnyValue{
 		Value: &otlpcommon.AnyValue_KvlistValue{

pdata/plog/json.go

@@ -16,11 +16,14 @@ package plog // import "go.opentelemetry.io/collector/pdata/plog"
 
 import (
 	"bytes"
+	"fmt"
 
 	"github.com/gogo/protobuf/jsonpb"
+	jsoniter "github.com/json-iterator/go"
 
 	"go.opentelemetry.io/collector/pdata/internal"
 	otlplogs "go.opentelemetry.io/collector/pdata/internal/data/protogen/logs/v1"
+	"go.opentelemetry.io/collector/pdata/internal/json"
 	"go.opentelemetry.io/collector/pdata/internal/otlp"
 )
 
@@ -45,7 +48,6 @@ func (e *jsonMarshaler) MarshalLogs(ld Logs) ([]byte, error) {
 }
 
 type jsonUnmarshaler struct {
-	delegate jsonpb.Unmarshaler
 }
 
 // NewJSONUnmarshaler returns a model.Unmarshaler. Unmarshals from OTLP json bytes.
@@ -54,14 +56,122 @@ func NewJSONUnmarshaler() Unmarshaler {
 }
 
 func newJSONUnmarshaler() *jsonUnmarshaler {
-	return &jsonUnmarshaler{delegate: jsonpb.Unmarshaler{}}
+	return &jsonUnmarshaler{}
 }
 
 func (d *jsonUnmarshaler) UnmarshalLogs(buf []byte) (Logs, error) {
-	ld := otlplogs.LogsData{}
-	if err := d.delegate.Unmarshal(bytes.NewReader(buf), &ld); err != nil {
-		return Logs{}, err
+	iter := jsoniter.ConfigFastest.BorrowIterator(buf)
+	defer jsoniter.ConfigFastest.ReturnIterator(iter)
+	ld := d.readLogsData(iter)
+	if iter.Error != nil {
+		return Logs{}, iter.Error
 	}
 	otlp.MigrateLogs(ld.ResourceLogs)
 	return Logs(internal.LogsFromProto(ld)), nil
 }
+
+func (d *jsonUnmarshaler) readLogsData(iter *jsoniter.Iterator) otlplogs.LogsData {
+	ld := otlplogs.LogsData{}
+	iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
+		switch f {
+		case "resource_logs", "resourceLogs":
+			iter.ReadArrayCB(func(iterator *jsoniter.Iterator) bool {
+				ld.ResourceLogs = append(ld.ResourceLogs, d.readResourceLogs(iter))
+				return true
+			})
+		default:
+			iter.Skip()
+		}
+		return true
+	})
+	return ld
+}
+
+func (d *jsonUnmarshaler) readResourceLogs(iter *jsoniter.Iterator) *otlplogs.ResourceLogs {
+	rs := &otlplogs.ResourceLogs{}
+	iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
+		switch f {
+		case "resource":
+			json.ReadResource(iter, &rs.Resource)
+		case "scope_logs", "scopeLogs":
+			iter.ReadArrayCB(func(iter *jsoniter.Iterator) bool {
+				rs.ScopeLogs = append(rs.ScopeLogs,
+					d.readScopeLogs(iter))
+				return true
+			})
+		case "schemaUrl", "schema_url":
+			rs.SchemaUrl = iter.ReadString()
+		default:
+			iter.Skip()
+		}
+		return true
+	})
+	return rs
+}
+
+func (d *jsonUnmarshaler) readScopeLogs(iter *jsoniter.Iterator) *otlplogs.ScopeLogs {
+	ils := &otlplogs.ScopeLogs{}
+	iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
+		switch f {
+		case "scope":
+			json.ReadScope(iter, &ils.Scope)
+		case "log_records", "logRecords":
+			iter.ReadArrayCB(func(iter *jsoniter.Iterator) bool {
+				ils.LogRecords = append(ils.LogRecords, d.readLog(iter))
+				return true
+			})
+		case "schemaUrl", "schema_url":
+			ils.SchemaUrl = iter.ReadString()
+		default:
+			iter.Skip()
+		}
+		return true
+	})
+	return ils
+}
+
+func (d *jsonUnmarshaler) readLog(iter *jsoniter.Iterator) *otlplogs.LogRecord {
+	lr := &otlplogs.LogRecord{}
+	iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
+		switch f {
+		case "timeUnixNano", "time_unix_nano":
+			lr.TimeUnixNano = json.ReadUint64(iter)
+		case "observed_time_unix_nano", "observedTimeUnixNano":
+			lr.ObservedTimeUnixNano = json.ReadUint64(iter)
+		case "severity_number", "severityNumber":
+			lr.SeverityNumber = d.readSeverityNumber(iter)
+		case "severity_text", "severityText":
+			lr.SeverityText = iter.ReadString()
+		case "body":
+			iter.ReadObjectCB(func(iter *jsoniter.Iterator, f string) bool {
+				lr.Body = json.ReadAnyValue(iter, f)
+				return true
+			})
+		case "attributes":
+			iter.ReadArrayCB(func(iter *jsoniter.Iterator) bool {
+				lr.Attributes = append(lr.Attributes, json.ReadAttribute(iter))
+				return true
+			})
+		case "droppedAttributesCount", "dropped_attributes_count":
+			lr.DroppedAttributesCount = json.ReadUint32(iter)
+		case "flags":
+			lr.Flags = json.ReadUint32(iter)
+		case "traceId", "trace_id":
+			if err := lr.TraceId.UnmarshalJSON([]byte(iter.ReadString())); err != nil {
+				iter.ReportError("readLog.traceId", fmt.Sprintf("parse trace_id:%v", err))
+			}
+		case "spanId", "span_id":
+			if err := lr.SpanId.UnmarshalJSON([]byte(iter.ReadString())); err != nil {
+				iter.ReportError("readLog.spanId", fmt.Sprintf("parse span_id:%v", err))
+			}
+		default:
+			iter.Skip()
+		}
+		return true
+	})
+	return lr
+}
+
+func (d *jsonUnmarshaler) readSeverityNumber(iter *jsoniter.Iterator) otlplogs.SeverityNumber {
+	return otlplogs.SeverityNumber(json.ReadEnumValue(iter, otlplogs.SeverityNumber_value))
+}

pdata/plog/json_test.go

@@ -16,39 +16,146 @@ package plog
 
 import (
 	"testing"
+	"time"
 
+	"github.com/gogo/protobuf/jsonpb"
+	jsoniter "github.com/json-iterator/go"
 	"github.com/stretchr/testify/assert"
+
+	otlplogs "go.opentelemetry.io/collector/pdata/internal/data/protogen/logs/v1"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
 var logsOTLP = func() Logs {
 	ld := NewLogs()
 	rl := ld.ResourceLogs().AppendEmpty()
 	rl.Resource().Attributes().UpsertString("host.name", "testHost")
+	rl.Resource().SetDroppedAttributesCount(1)
+	rl.SetSchemaUrl("testSchemaURL")
 	il := rl.ScopeLogs().AppendEmpty()
 	il.Scope().SetName("name")
 	il.Scope().SetVersion("version")
-	il.LogRecords().AppendEmpty().SetSeverityText("Error")
+	il.Scope().SetDroppedAttributesCount(1)
+	il.SetSchemaUrl("ScopeLogsSchemaURL")
+	lg := il.LogRecords().AppendEmpty()
+	lg.SetSeverityNumber(SeverityNumber(otlplogs.SeverityNumber_SEVERITY_NUMBER_ERROR))
+	lg.SetSeverityText("Error")
+	lg.SetDroppedAttributesCount(1)
+	lg.SetFlags(LogRecordFlags(otlplogs.LogRecordFlags_LOG_RECORD_FLAG_UNSPECIFIED))
+	traceID := pcommon.TraceID([16]byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x0E, 0x0F, 0x10})
+	spanID := pcommon.SpanID([8]byte{0x11, 0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18})
+	lg.SetTraceID(traceID)
+	lg.SetSpanID(spanID)
+	lg.Body().SetStringVal("hello world")
+	t := pcommon.NewTimestampFromTime(time.Now())
+	lg.SetTimestamp(t)
+	lg.SetObservedTimestamp(t)
+	lg.Attributes().UpsertString("sdkVersion", "1.0.1")
 	return ld
-}()
+}
 
-var logsJSON = `{"resourceLogs":[{"resource":{"attributes":[{"key":"host.name","value":{"stringValue":"testHost"}}]},"scopeLogs":[{"scope":{"name":"name","version":"version"},"logRecords":[{"severityText":"Error","body":{},"traceId":"","spanId":""}]}]}]}`
+func TestReadLogsDataUnknownField(t *testing.T) {
+	jsonStr := `{"extra":""}`
+	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
+	defer jsoniter.ConfigFastest.ReturnIterator(iter)
+	unmarshaler := &jsonUnmarshaler{}
+	value := unmarshaler.readLogsData(iter)
+	assert.NoError(t, iter.Error)
+	assert.EqualValues(t, otlplogs.LogsData{}, value)
+}
+
+func TestReadResourceLogsUnknownField(t *testing.T) {
+	jsonStr := `{"extra":"","resource":{"extra":""}}`
+	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
+	defer jsoniter.ConfigFastest.ReturnIterator(iter)
+	unmarshaler := &jsonUnmarshaler{}
+	value := unmarshaler.readResourceLogs(iter)
+	assert.NoError(t, iter.Error)
+	assert.EqualValues(t, &otlplogs.ResourceLogs{}, value)
+}
+
+func TestReadScopeLogs(t *testing.T) {
+	jsonStr := `{"extra":""}`
+	iter := jsoniter.ConfigFastest.BorrowIterator([]byte(jsonStr))
+	defer jsoniter.ConfigFastest.ReturnIterator(iter)
+	unmarshaler := &jsonUnmarshaler{}
+	value := unmarshaler.readScopeLogs(iter)
+	assert.NoError(t, iter.Error)
+	assert.EqualValues(t, &otlplogs.ScopeLogs{}, value)
+}
 
 func TestLogsJSON(t *testing.T) {
-	encoder := NewJSONMarshaler()
-	jsonBuf, err := encoder.MarshalLogs(logsOTLP)
-	assert.NoError(t, err)
+	type args struct {
+		logFunc func() Logs
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "otlp",
+			args: args{
+				logFunc: logsOTLP,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			for _, opEnumsAsInts := range []bool{true, false} {
+				for _, opEmitDefaults := range []bool{true, false} {
+					for _, opOrigName := range []bool{true, false} {
+						marshaller := &jsonMarshaler{
+							delegate: jsonpb.Marshaler{
+								EnumsAsInts:  opEnumsAsInts,
+								EmitDefaults: opEmitDefaults,
+								OrigName:     opOrigName,
+							}}
+						ld := tt.args.logFunc()
+						jsonBuf, err := marshaller.MarshalLogs(ld)
+						assert.NoError(t, err)
+						decoder := NewJSONUnmarshaler()
+						var got interface{}
+						got, err = decoder.UnmarshalLogs(jsonBuf)
+						assert.NoError(t, err)
+						assert.EqualValues(t, ld, got)
+					}
+				}
+			}
+		})
+	}
+}
 
+var logsJSON = `{"resourceLogs":[{"resource":{"attributes":[{"key":"host.name","value":{"stringValue":"testHost"}}]},"scopeLogs":[{"scope":{"name":"name","version":"version"},"logRecords":[{"severityText":"Error","body":{},"traceId":"","spanId":""}]}]}]}`
+
+func TestLogsJSON_WithoutTraceIdAndSpanId(t *testing.T) {
 	decoder := NewJSONUnmarshaler()
-	var got interface{}
-	got, err = decoder.UnmarshalLogs(jsonBuf)
+	_, err := decoder.UnmarshalLogs([]byte(logsJSON))
 	assert.NoError(t, err)
+}
+
+var logsJSONWrongTraceID = `{"resourceLogs":[{"resource":{"attributes":[{"key":"host.name","value":{"stringValue":"testHost"}}]},"scopeLogs":[{"scope":{"name":"name","version":"version"},"logRecords":[{"severityText":"Error","body":{},"traceId":"--","spanId":""}]}]}]}`
+
+func TestLogsJSON_WrongTraceID(t *testing.T) {
+	decoder := NewJSONUnmarshaler()
+	_, err := decoder.UnmarshalLogs([]byte(logsJSONWrongTraceID))
+	assert.Error(t, err)
+	if assert.Error(t, err) {
+		assert.Contains(t, err.Error(), "parse trace_id")
+	}
+}
+
+var LogsJSONWrongSpanID = `{"resourceLogs":[{"resource":{"attributes":[{"key":"host.name","value":{"stringValue":"testHost"}}]},"scopeLogs":[{"scope":{"name":"name","version":"version"},"logRecords":[{"severityText":"Error","body":{},"traceId":"","spanId":"--"}]}]}]}`
 
-	assert.EqualValues(t, logsOTLP, got)
+func TestLogsJSON_WrongSpanID(t *testing.T) {
+	decoder := NewJSONUnmarshaler()
+	_, err := decoder.UnmarshalLogs([]byte(LogsJSONWrongSpanID))
+	assert.Error(t, err)
+	if assert.Error(t, err) {
+		assert.Contains(t, err.Error(), "parse span_id")
+	}
 }
 
-func TestLogsJSON_Marshal(t *testing.T) {
-	encoder := NewJSONMarshaler()
-	jsonBuf, err := encoder.MarshalLogs(logsOTLP)
-	assert.NoError(t, err)
-	assert.Equal(t, logsJSON, string(jsonBuf))
+func TestLogsJSON_MarshalJSON(t *testing.T) {
+	marshal := NewJSONMarshaler()
+	assert.NotNil(t, marshal)
 }