Add support for json export

telemetry/dd/json/attr.go

@@ -0,0 +1,18 @@
+package json
+
+import "go.opentelemetry.io/otel/attribute"
+
+func attrSliceToMap(attributes []attribute.KeyValue) map[string]any {
+	if len(attributes) == 0 {
+		return nil
+	}
+	attrs := make(map[string]any, len(attributes))
+	for _, kv := range attributes {
+		attrs[string(kv.Key)] = kv.Value.AsInterface()
+	}
+	return attrs
+}
+
+func attrSetToMap(attributes attribute.Set) map[string]any {
+	return attrSliceToMap(attributes.ToSlice())
+}

telemetry/dd/json/attr_test.go

@@ -0,0 +1,117 @@
+package json
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/otel/attribute"
+)
+
+func Test_attrSliceToMap(t *testing.T) {
+	type args struct {
+		attributes []attribute.KeyValue
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]any
+	}{
+		{
+			name: "string attribute",
+			args: args{attributes: []attribute.KeyValue{
+				attribute.String("key", "value"),
+			}},
+			want: map[string]any{
+				"key": "value",
+			},
+		},
+		{
+			name: "int64 attribute",
+			args: args{attributes: []attribute.KeyValue{
+				attribute.Int64("key", 123),
+			}},
+			want: map[string]any{
+				"key": int64(123),
+			},
+		},
+		{
+			name: "bool attribute",
+			args: args{attributes: []attribute.KeyValue{
+				attribute.Bool("key", true),
+			}},
+			want: map[string]any{
+				"key": true,
+			},
+		},
+		{
+			name: "float64 attribute",
+			args: args{attributes: []attribute.KeyValue{
+				attribute.Float64("key", 123.456),
+			}},
+			want: map[string]any{
+				"key": 123.456,
+			},
+		},
+		{
+			name: "multiple attributes",
+			args: args{attributes: []attribute.KeyValue{
+				attribute.String("str", "value"),
+				attribute.Int64("int", 123),
+				attribute.Bool("bool", true),
+				attribute.Float64("float", 123.456),
+			}},
+			want: map[string]any{
+				"str":   "value",
+				"int":   int64(123),
+				"bool":  true,
+				"float": 123.456,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.EqualValuesf(t, tt.want, attrSliceToMap(tt.args.attributes), "attrSliceToMap(%v)", tt.args.attributes)
+		})
+	}
+}
+
+func Test_attrSetToMap(t *testing.T) {
+	type args struct {
+		attributes attribute.Set
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]any
+	}{
+		{
+			name: "set with single attribute",
+			args: args{attributes: attribute.NewSet(
+				attribute.String("key", "value"),
+			)},
+			want: map[string]any{
+				"key": "value",
+			},
+		},
+		{
+			name: "set with multiple attributes",
+			args: args{attributes: attribute.NewSet(
+				attribute.String("str", "value"),
+				attribute.Int64("int", 123),
+				attribute.Bool("bool", true),
+				attribute.Float64("float", 123.456),
+			)},
+			want: map[string]any{
+				"str":   "value",
+				"int":   int64(123),
+				"bool":  true,
+				"float": 123.456,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equalf(t, tt.want, attrSetToMap(tt.args.attributes), "attrSetToMap(%v)", tt.args.attributes)
+		})
+	}
+}

telemetry/dd/json/exporter.go

@@ -0,0 +1,121 @@
+package json
+
+import (
+	"bufio"
+	"context"
+	"encoding/binary"
+	"encoding/json"
+	"io"
+	"strconv"
+
+	"go.opentelemetry.io/otel/log"
+	sdklog "go.opentelemetry.io/otel/sdk/log"
+	"go.opentelemetry.io/otel/sdk/resource"
+)
+
+var _ sdklog.Exporter = &jsonExporter{}
+
+type jsonExporter struct {
+	flush   func() error
+	encoder json.Encoder
+}
+
+func NewJSONExporter(w io.Writer) sdklog.Exporter {
+	b := bufio.NewWriter(w)
+	return &jsonExporter{
+		flush:   b.Flush,
+		encoder: *json.NewEncoder(b),
+	}
+}
+
+// Export implements log.Exporter.
+func (j *jsonExporter) Export(_ context.Context, records []sdklog.Record) error {
+	for i := 0; i < len(records); i++ {
+		err := j.encoder.Encode(convert(&records[i]))
+		if err != nil {
+			return err
+		}
+	}
+	return j.flush()
+}
+
+func convert(record *sdklog.Record) jsonRecord {
+	entry := jsonRecord{
+		Msg:          record.Body().AsString(),
+		Level:        record.Severity().String(),
+		Time:         LogTime(record.Timestamp()),
+		ObservedTime: LogTime(record.ObservedTimestamp()),
+		Scope: jsonScope{
+			SchemaURL:  record.InstrumentationScope().SchemaURL,
+			Name:       record.InstrumentationScope().Name,
+			Version:    record.InstrumentationScope().Version,
+			Attributes: attrSetToMap(record.InstrumentationScope().Attributes),
+		},
+	}
+
+	res := record.Resource()
+	if &res != resource.Empty() {
+		entry.Resource = &jsonResource{
+			SchemaURL: res.SchemaURL(),
+		}
+		entry.Resource.Attributes = attrSliceToMap(res.Attributes())
+	}
+
+	if record.TraceID().IsValid() {
+		id := record.TraceID()
+		entry.DDTraceID = strconv.FormatUint(binary.BigEndian.Uint64(id[8:]), 10)
+	}
+
+	if record.SpanID().IsValid() {
+		id := record.SpanID()
+		entry.DDSpanID = strconv.FormatUint(binary.BigEndian.Uint64(id[:]), 10)
+	}
+
+	if record.AttributesLen() > 0 {
+		attrs := make(map[string]value, record.AttributesLen())
+		record.WalkAttributes(func(kv log.KeyValue) bool {
+			attrs[kv.Key] = newValue(kv.Value)
+			return true
+		})
+		entry.Attributes = attrs
+	}
+	return entry
+}
+
+type jsonScope struct {
+	SchemaURL  string         `json:"schema_url"`
+	Name       string         `json:"name"`
+	Version    string         `json:"version"`
+	Attributes map[string]any `json:"attributes,omitempty"`
+}
+
+type jsonRecord struct {
+	Level        string  `json:"level"`
+	Msg          string  `json:"msg"`
+	Time         LogTime `json:"time,omitempty"`
+	ObservedTime LogTime `json:"observed_time,omitempty"`
+
+	Scope    jsonScope     `json:"scope,omitempty"`
+	Resource *jsonResource `json:"resource,omitempty"`
+
+	DDTraceID string `json:"dd.trace_id,omitempty"`
+	DDSpanID  string `json:"dd.span_id,omitempty"`
+
+	Attributes map[string]value `json:"attributes,omitempty"`
+}
+
+type jsonResource struct {
+	SchemaURL  string         `json:"schema.url"`
+	Attributes map[string]any `json:"attributes,omitempty"`
+}
+
+// ForceFlush implements log.Exporter.
+func (j *jsonExporter) ForceFlush(_ context.Context) error {
+	return j.flush()
+}
+
+// Shutdown implements log.Exporter.
+func (j *jsonExporter) Shutdown(_ context.Context) error {
+	j.encoder = *json.NewEncoder(io.Discard)
+	return j.flush()
+}