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,45 @@ func (s *span) SetName(name string) {
  s.attributes[ext.SpanName] = strings.ToLower(name)
 }
 
+// stringifySpanEvents transforms a slice of otelsdk.Events into a comma separated string
+func stringifySpanEvents(evts []otelsdk.Event) (s string) {
+ for i, e := range evts {
+ if i == 0 {
+ s += marshalSpanEvent(e)
+ } else {
+ s += "," + marshalSpanEvent(e)
+ }
+ }
+ return s
+}
+
+// ddSpanEvent holds information about otelsdk.Event types, with some fields altered and renamed to fit Datadog needs
+// along with json tags for easy marshaling
+type ddSpanEvent struct {
+ Name string `json:"name"`
+ Time_unix_nano int64 `json:"time_unix_nano"`
+ Attributes map[string]interface{} `json:"attributes,omitempty"`
+}
+
+// marshalSpanEvent transforms an otelsdk.Event into a JSON-encoded object with "name" and "time_unix_nano" fields, and an optional "attributes" field
+func marshalSpanEvent(evt otelsdk.Event) string {
+ 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 +110,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 +216,16 @@ func (s *span) SetStatus(code otelcodes.Code, description string) {
  }
 }
 
+// AddEvent adds a span event onto the span with the provided name and EventOptions
+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,81 @@ 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) {
+ want := fmt.Sprintf("{\"name\":\"evt\",\"time_unix_nano\":%v,\"attributes\":{\"attribute1\":\"value1\",\"attribute2\":123,\"attribute3\":[1,2,3],\"attribute4\":false}}", nowUnix)
+ 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(want, 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) {
+ // ensure the `name` field is still included in the json result with value of empty string, even if not set in the struct object
+ 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)
+ t.Run("multiple events", func(t *testing.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)
+ })
+ t.Run("no events", func(t *testing.T) {
+ s := stringifySpanEvents(nil)
+ assert.Equal("", s)
+ })
+}
+
 func TestSpanEnd(t *testing.T) {
  assert := assert.New(t)
  _, payloads, cleanup := mockTracerProvider(t)
@@ -203,6 +279,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 +312,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 +383,66 @@ func TestSpanSetStatus(t *testing.T) {
  }
 }
 
+func TestSpanAddEvent(t *testing.T) {
+ assert := assert.New(t)
+ _, _, cleanup := mockTracerProvider(t)
+ tr := otel.Tracer("")
+ defer cleanup()
+
+ t.Run("event with attributes", func(t *testing.T) {
+ _, sp := tr.Start(context.Background(), "span_event")
+ // When no timestamp option is provided, otel will generate a timestamp for the event
+ // We can't know 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
+ // note that attribute.Int("pid", 4328) created an attribute with value int64(4328), hence why the `want` is in int64 format
+ wantAttrs := map[string]interface{}{
+ "pid": int64(4328),
+ "signal": "SIGHUP",
+ }
+ for k, v := range wantAttrs {
+ assert.True(attributesContains(e.Attributes, k, v))
+ }
+ })
+ t.Run("event with timestamp", func(t *testing.T) {
+ _, sp := tr.Start(context.Background(), "span_event")
+ now := time.Now()
+ sp.AddEvent("My event!", oteltrace.WithTimestamp(now))
+ sp.End()
+
+ dd := sp.(*span)
+ assert.Len(dd.events, 1)
+ e := dd.events[0]
+ assert.Equal(e.Time, now)
+ })
+}
+
+// attributesContains returns true if attrs contains an attribute.KeyValue with the provided key and val
+func attributesContains(attrs []attribute.KeyValue, key string, val interface{}) bool {
+ for _, a := range attrs {
+ fmt.Printf("Looking for %v:%v with types %T:%T; have %v:%v with types %T:%T\n", key, val, key, val, string(a.Key), a.Value.AsInterface(), string(a.Key), a.Value.AsInterface())
+ if string(a.Key) == key && a.Value.AsInterface() == val {
+ fmt.Println("hello?")
+ return true
+ }
+ }
+ return 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