Mtoff/span events

ddtrace/opentelemetry/span.go

@@ -7,7 +7,9 @@ package opentelemetry
 
 import (
 	"encoding/binary"
+	"encoding/json"
 	"errors"
+	"fmt"
 	"strconv"
 	"strings"
 	"sync"
@@ -19,6 +21,7 @@ import (
 
 	"go.opentelemetry.io/otel/attribute"
 	otelcodes "go.opentelemetry.io/otel/codes"
+	otelsdk "go.opentelemetry.io/otel/sdk/trace"
 	oteltrace "go.opentelemetry.io/otel/trace"
 	"go.opentelemetry.io/otel/trace/noop"
 )
@@ -35,6 +38,7 @@ type span struct {
 	finishOpts []tracer.FinishOption
 	statusInfo
 	*oteltracer
+	events []otelsdk.Event
 }
 
 func (s *span) TracerProvider() oteltrace.TracerProvider { return s.oteltracer.provider }
@@ -45,6 +49,40 @@ func (s *span) SetName(name string) {
 	s.attributes[ext.SpanName] = strings.ToLower(name)
 }
 
+func stringifySpanEvents(evts []otelsdk.Event) (s string) {
+	for i, e := range evts {
+		if i == 0 {
+			s += marshalSpanEvent(e)
+		} else {
+			s += "," + marshalSpanEvent(e)
+		}
+	}
+	return s
+}
+
+func marshalSpanEvent(evt otelsdk.Event) string {
+	type ddSpanEvent struct {
+		Name           string                 `json:"name"`
+		Time_unix_nano int64                  `json:"time_unix_nano"`
+		Attributes     map[string]interface{} `json:"attributes,omitempty"`
+	}
+	spEvt := ddSpanEvent{
+		Time_unix_nano: evt.Time.Unix(),
+		Name:           evt.Name,
+	}
+	spEvt.Attributes = make(map[string]interface{})
+	for _, a := range evt.Attributes {
+		spEvt.Attributes[string(a.Key)] = a.Value.AsInterface()
+	}
+	s, err := json.Marshal(spEvt)
+	if err != nil {
+		// log something?
+		log.Debug(fmt.Sprintf("Issue marshaling span event %v:%v", spEvt, err))
+		return ""
+	}
+	return string(s)
+}
+
 func (s *span) End(options ...oteltrace.SpanEndOption) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -67,10 +105,13 @@ func (s *span) End(options ...oteltrace.SpanEndOption) {
 	if op, ok := s.attributes[ext.SpanName]; !ok || op == "" {
 		s.DD.SetTag(ext.SpanName, strings.ToLower(s.createOperationName()))
 	}
-
 	for k, v := range s.attributes {
 		s.DD.SetTag(k, v)
 	}
+	evts := stringifySpanEvents(s.events)
+	if evts != "" {
+		s.DD.SetTag("events", evts)
+	}
 	var finishCfg = oteltrace.NewSpanEndConfig(options...)
 	var opts []tracer.FinishOption
 	if s.statusInfo.code == otelcodes.Error {
@@ -170,6 +211,15 @@ func (s *span) SetStatus(code otelcodes.Code, description string) {
 	}
 }
 
+func (s *span) AddEvent(name string, opts ...oteltrace.EventOption) {
+	if !s.IsRecording() {
+		return
+	}
+	c := oteltrace.NewEventConfig(opts...)
+	e := otelsdk.Event{Name: name, Attributes: c.Attributes(), Time: c.Timestamp()}
+	s.events = append(s.events, e)
+}
+
 // SetAttributes sets the key-value pairs as tags on the span.
 // Every value is propagated as an interface.
 // Some attribute keys are reserved and will be remapped to Datadog reserved tags.

ddtrace/opentelemetry/span_test.go

@@ -27,6 +27,7 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
+	otelsdk "go.opentelemetry.io/otel/sdk/trace"
 	oteltrace "go.opentelemetry.io/otel/trace"
 )
 
@@ -185,6 +186,73 @@ func TestSpanLink(t *testing.T) {
 	assert.Equal(uint32(0x80000001), spanLinks[0].Flags) // sampled and set
 }
 
+func TestMarshalSpanEvent(t *testing.T) {
+	assert := assert.New(t)
+	now := time.Now()
+	nowUnix := now.Unix()
+	t.Run("attributes", func(t *testing.T) {
+		s := marshalSpanEvent(otelsdk.Event{
+			Name: "evt",
+			Time: now,
+			Attributes: []attribute.KeyValue{
+				{
+					Key:   attribute.Key("attribute1"),
+					Value: attribute.StringValue("value1"),
+				},
+				{
+					Key:   attribute.Key("attribute2"),
+					Value: attribute.IntValue(123),
+				},
+				{
+					Key:   attribute.Key("attribute3"),
+					Value: attribute.Float64SliceValue([]float64{1.0, 2.0, 3.0}),
+				},
+				{
+					Key:   attribute.Key("attribute4"),
+					Value: attribute.BoolValue(true),
+				},
+				// if two attributes have same key, last-set attribute takes precedence
+				{
+					Key:   attribute.Key("attribute4"),
+					Value: attribute.BoolValue(false),
+				},
+			},
+		})
+		assert.Equal(fmt.Sprintf("{\"name\":\"evt\",\"time_unix_nano\":%v,\"attributes\":{\"attribute1\":\"value1\",\"attribute2\":123,\"attribute3\":[1,2,3],\"attribute4\":false}}", nowUnix), s)
+	})
+	t.Run("unexpected field", func(t *testing.T) {
+		// otelsdk.Event type has field `DroppedAttributeCount`, but we don't use this field
+		s := marshalSpanEvent(otelsdk.Event{
+			Name:                  "evt",
+			Time:                  now,
+			DroppedAttributeCount: 1,
+		})
+		// resulting string should discard DroppedAttributeCount
+		assert.Equal(fmt.Sprintf("{\"name\":\"evt\",\"time_unix_nano\":%v}", nowUnix), s)
+	})
+	t.Run("missing fields", func(t *testing.T) {
+		s := marshalSpanEvent(otelsdk.Event{
+			Time: now,
+		})
+		assert.Equal(fmt.Sprintf("{\"name\":\"\",\"time_unix_nano\":%v}", nowUnix), s)
+	})
+}
+
+func TestStringifySpanEvent(t *testing.T) {
+	assert := assert.New(t)
+	evt1 := otelsdk.Event{
+		Name: "abc",
+	}
+	evt2 := otelsdk.Event{
+		Name: "def",
+	}
+	evts := []otelsdk.Event{evt1, evt2}
+	want := marshalSpanEvent(evt1) + "," + marshalSpanEvent(evt2)
+
+	s := stringifySpanEvents(evts)
+	assert.Equal(want, s)
+}
+
 func TestSpanEnd(t *testing.T) {
 	assert := assert.New(t)
 	_, payloads, cleanup := mockTracerProvider(t)
@@ -203,6 +271,9 @@ func TestSpanEnd(t *testing.T) {
 		sp.SetAttributes(attribute.String(k, v))
 	}
 	assert.True(sp.IsRecording())
+	now := time.Now()
+	nowUnix := now.Unix()
+	sp.AddEvent("evt", oteltrace.WithTimestamp(now), oteltrace.WithAttributes(attribute.String("key1", "value"), attribute.Int("key2", 1234)))
 
 	sp.End()
 	assert.False(sp.IsRecording())
@@ -233,6 +304,7 @@ func TestSpanEnd(t *testing.T) {
 	for k, v := range ignoredAttributes {
 		assert.NotContains(meta, fmt.Sprintf("%s:%s", k, v))
 	}
+	assert.Contains(meta, fmt.Sprintf("events:{\"name\":\"evt\",\"time_unix_nano\":%v,\"attributes\":{\"key1\":\"value\",\"key2\":1234}", nowUnix))
 }
 
 // This test verifies that setting the status of a span
@@ -303,6 +375,71 @@ func TestSpanSetStatus(t *testing.T) {
 	}
 }
 
+func TestSpanAddEvent(t *testing.T) {
+	assert := assert.New(t)
+	_, _, cleanup := mockTracerProvider(t)
+	tr := otel.Tracer("")
+	defer cleanup()
+
+	_, sp := tr.Start(context.Background(), "span_with_events")
+	// We can't capture the exact time that the event is added, but we can create start and end "bounds" and assert
+	// that the event's eventual timestamp is between those bounds
+	timeStartBound := time.Now()
+	sp.AddEvent("My event!", oteltrace.WithAttributes(attribute.Int("pid", 4328), attribute.String("signal", "SIGHUP")))
+	timeEndBound := time.Now()
+	sp.End()
+	dd := sp.(*span)
+
+	// Assert event exists under span events
+	assert.Len(dd.events, 1)
+	e := dd.events[0]
+	assert.Equal(e.Name, "My event!")
+	// assert event timestamp is [around] the expected time
+	assert.True((e.Time).After(timeStartBound) && (e.Time).Before(timeEndBound))
+	// Assert both attributes exist on the event
+	assert.Len(e.Attributes, 2)
+	// Assert attribute key-value fields
+	wantAttrs := map[string]interface{}{
+		"pid":    4328,
+		"signal": "SIGHUP",
+	}
+	for k, v := range wantAttrs {
+		err := attributesContains(e.Attributes, k, v)
+		assert.NoError(err, err)
+	}
+}
+
+// attributesContains returns true if attrs contains an attribute.KeyValue with the provided key and val
+func attributesContains(attrs []attribute.KeyValue, key string, val interface{}) error {
+	for _, a := range attrs {
+		if string(a.Key) == key {
+			v, ok := attributeValEqual(val, a.Value)
+			if ok {
+				return nil
+			}
+			return fmt.Errorf("Expected %v as value for attribute key %v, but got %v instead", val, key, v)
+		}
+	}
+	return fmt.Errorf("Expected key %v not available in attributes", key)
+}
+
+// attributeVal returns whether the "actual" attribute Value is equal to "expected"
+func attributeValEqual(expected interface{}, actual attribute.Value) (interface{}, bool) {
+	switch expected.(type) {
+	case string:
+		v := actual.AsString()
+		return v, v == expected
+	case int:
+		v := actual.AsInt64()
+		return v, int(v) == expected
+	case bool:
+		v := actual.AsBool()
+		return v, v == expected
+	default:
+		return nil, false
+	}
+}
+
 func TestSpanContextWithStartOptions(t *testing.T) {
 	assert := assert.New(t)
 	_, payloads, cleanup := mockTracerProvider(t)

go.mod

@@ -53,7 +53,7 @@ require (
 	github.com/golang/protobuf v1.5.3
 	github.com/gomodule/redigo v1.8.9
 	github.com/google/pprof v0.0.0-20230817174616-7a8ec2ada47b
-	github.com/google/uuid v1.5.0
+	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.0
 	github.com/graph-gophers/graphql-go v1.5.0
 	github.com/graphql-go/graphql v0.8.1
@@ -80,7 +80,7 @@ require (
 	github.com/segmentio/kafka-go v0.4.42
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spaolacci/murmur3 v1.1.0
-	github.com/stretchr/testify v1.8.4
+	github.com/stretchr/testify v1.9.0
 	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d
 	github.com/tidwall/buntdb v1.3.0
 	github.com/tinylib/msgp v1.1.8
@@ -91,12 +91,12 @@ require (
 	github.com/zenazn/goji v1.0.1
 	go.mongodb.org/mongo-driver v1.12.1
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.44.0
-	go.opentelemetry.io/otel v1.20.0
-	go.opentelemetry.io/otel/trace v1.20.0
+	go.opentelemetry.io/otel v1.28.0
+	go.opentelemetry.io/otel/trace v1.28.0
 	go.uber.org/atomic v1.11.0
 	golang.org/x/net v0.23.0
 	golang.org/x/oauth2 v0.9.0
-	golang.org/x/sys v0.20.0
+	golang.org/x/sys v0.21.0
 	golang.org/x/time v0.3.0
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028
 	google.golang.org/api v0.128.0
@@ -157,7 +157,7 @@ require (
 	github.com/gabriel-vasile/mimetype v1.4.2 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-jose/go-jose/v3 v3.0.3 // indirect
-	github.com/go-logr/logr v1.3.0 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-pg/zerochecker v0.2.0 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
@@ -229,7 +229,7 @@ require (
 	github.com/ryanuber/go-glob v1.0.0 // indirect
 	github.com/secure-systems-lab/go-securesystemslib v0.7.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/stretchr/objx v0.5.1 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/tidwall/btree v1.6.0 // indirect
 	github.com/tidwall/gjson v1.16.0 // indirect
 	github.com/tidwall/grect v0.1.4 // indirect
@@ -252,7 +252,8 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	go.opencensus.io v0.24.0 // indirect
-	go.opentelemetry.io/otel/metric v1.20.0 // indirect
+	go.opentelemetry.io/otel/metric v1.28.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.28.0
 	golang.org/x/arch v0.4.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
 	golang.org/x/exp v0.0.0-20230321023759-10a507213a29 // indirect