draft: implement v1 trace protocol serialization

ddtrace/tracer/payload.go

@@ -46,11 +46,13 @@ type payload interface {
 	payloadReader
 }
 
-// newpayload returns a ready to use unsafe payload.
+// newPayload returns a ready to use payload.
 func newPayload(protocol float64) payload {
-	// TODO(hannahkm): add support for v1 protocol
-	// if protocol == traceProtocolV1 {
-	// }
+	if protocol == traceProtocolV1 {
+		return &safePayload{
+			p: newPayloadV1(),
+		}
+	}
 	return &safePayload{
 		p: newPayloadV04(),
 	}
@@ -155,24 +157,24 @@ func (sp *safePayload) protocol() float64 {
 // i.e. a chunk of a trace
 type traceChunk struct {
 	// the sampling priority of the trace
-	priority int32
+	priority int32 `msg:"priority"`
 
 	// the optional string origin ("lambda", "rum", etc.) of the trace chunk
-	origin uint32
+	origin string `msg:"origin,omitempty"`
 
 	// a collection of key to value pairs common in all `spans`
-	attributes map[uint32]anyValue
+	attributes keyValueList `msg:"attributes,omitempty"`
 
 	// a list of spans in this chunk
-	spans []Span
+	spans spanList `msg:"spans,omitempty"`
 
 	// whether the trace only contains analyzed spans
 	// (not required by tracers and set by the agent)
-	droppedTrace bool
+	droppedTrace bool `msg:"droppedTrace"`
 
 	// the ID of the trace to which all spans in this chunk belong
-	traceID uint8
+	traceID []byte `msg:"traceID"`
 
 	// the optional string decision maker (previously span tag _dd.p.dm)
-	decisionMaker uint32
+	samplingMechanism string `msg:"samplingMechanism,omitempty"`
 }

ddtrace/tracer/payload_test.go

@@ -60,9 +60,9 @@ func TestPayloadIntegrity(t *testing.T) {
 	}
 }
 
-// TestPayloadDecode ensures that whatever we push into the payload can
+// TestPayloadV04Decode ensures that whatever we push into a v0.4 payload can
 // be decoded by the codec.
-func TestPayloadDecode(t *testing.T) {
+func TestPayloadV04Decode(t *testing.T) {
 	for _, n := range []int{10, 1 << 10} {
 		t.Run(strconv.Itoa(n), func(t *testing.T) {
 			assert := assert.New(t)
@@ -77,6 +77,46 @@ func TestPayloadDecode(t *testing.T) {
 	}
 }
 
+// TestPayloadV1Decode ensures that whatever we push into a v1 payload can
+// be decoded by the codec, and that it matches the original payload.
+func TestPayloadV1Decode(t *testing.T) {
+	for _, n := range []int{10, 1 << 10} {
+		t.Run(strconv.Itoa(n), func(t *testing.T) {
+			assert := assert.New(t)
+			p := newPayloadV1()
+
+			p.containerID = "containerID"
+			p.languageName = "languageName"
+			p.languageVersion = "languageVersion"
+			p.tracerVersion = "tracerVersion"
+			p.runtimeID = "runtimeID"
+			p.env = "env"
+			p.hostname = "hostname"
+			p.appVersion = "appVersion"
+
+			for i := 0; i < n; i++ {
+				_, _ = p.push(newSpanList(i%5 + 1))
+			}
+
+			encoded, err := io.ReadAll(p)
+			assert.NoError(err)
+
+			got := newPayloadV1()
+			_, err = got.Decode(encoded)
+			assert.NoError(err)
+
+			assert.Equal(p.containerID, got.containerID)
+			assert.Equal(p.languageName, got.languageName)
+			assert.Equal(p.languageVersion, got.languageVersion)
+			assert.Equal(p.tracerVersion, got.tracerVersion)
+			assert.Equal(p.runtimeID, got.runtimeID)
+			assert.Equal(p.env, got.env)
+			assert.Equal(p.hostname, got.hostname)
+			assert.Equal(p.appVersion, got.appVersion)
+		})
+	}
+}
+
 func BenchmarkPayloadThroughput(b *testing.B) {
 	b.Run("10K", benchmarkPayloadThroughput(1))
 	b.Run("100K", benchmarkPayloadThroughput(10))