feat: support container tags hash in DSM (DataDog#4759)

This is a missing feature in dd-trace-go, here are the other implementations ([java](DataDog/dd-trace-java#9156), [node](DataDog/dd-trace-js#7212), [python](DataDog/dd-trace-py#15293))


This PR stores the has propagated back by the info endpoint (called every 5s) and enrich it in DSM hash computation


### Motivation

<!--
* What inspired you to submit this pull request?
* Link any related GitHub issues or PRs here.
* If this resolves a GitHub issue, include "Fixes #XXXX" to link the issue and auto-close it on merge.
-->

### Reviewer's Checklist
<!--
* Authors can use this list as a reference to ensure that there are no problems
  during the review but the signing off is to be done by the reviewer(s).
-->

- [ ] Changed code has unit tests for its functionality at or near 100% coverage.
- [ ] [System-Tests](https://github.com/DataDog/system-tests/) covering this feature have been added and enabled with the va.b.c-dev version tag.
- [ ] There is a benchmark for any new code, or changes to existing code.
- [ ] If this interacts with the agent in a new way, a system test has been added.
- [ ] New code is free of linting errors. You can check this by running `make lint` locally.
- [ ] New code doesn't break existing tests. You can check this by running `make test` locally.
- [ ] Add an appropriate team label so this PR gets put in the right place for the release notes.
- [ ] All generated files are up to date. You can check this by running `make generate` locally.
- [ ] Non-trivial go.mod changes, e.g. adding new modules, are reviewed by @DataDog/dd-trace-go-guild. Make sure all nested modules are up to date by running `make fix-modules` locally.

Unsure? Have a question? Request a review!


Co-authored-by: raphael.gavache <raphael.gavache@datadoghq.com>

Author: raphaelgavache

ddtrace/tracer/option.go

@@ -691,6 +691,7 @@ func fetchAgentFeatures(ctx context.Context, agentURL *url.URL, httpClient *http
 	if err != nil {
 		return agentFeatures{}, fmt.Errorf("creating /info request: %w", err)
 	}
+	setContainerHeaders(req.Header)
 	resp, err := httpClient.Do(req)
 	if err != nil {
 		return agentFeatures{}, fmt.Errorf("loading features: %w", err)
@@ -704,6 +705,7 @@ func fetchAgentFeatures(ctx context.Context, agentURL *url.URL, httpClient *http
 	if resp.StatusCode != http.StatusOK {
 		return agentFeatures{}, fmt.Errorf("unexpected /info status: %d", resp.StatusCode)
 	}
+	updateContainerTagsHash(resp.Header)
 	type infoResponse struct {
 		Endpoints          []string `json:"endpoints"`
 		ClientDropP0s      bool     `json:"client_drop_p0s"`

ddtrace/tracer/transport.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/DataDog/dd-trace-go/v2/ddtrace/internal/tracerstats"
 	"github.com/DataDog/dd-trace-go/v2/internal"
+	"github.com/DataDog/dd-trace-go/v2/internal/processtags"
 	"github.com/DataDog/dd-trace-go/v2/internal/version"
 
 	"github.com/tinylib/msgp/msgp"
@@ -38,6 +39,9 @@ const (
 	defaultHTTPTimeout       = 10 * time.Second              // defines the current timeout before giving up with the send process
 	traceCountHeader         = "X-Datadog-Trace-Count"       // header containing the number of traces in the payload
 	obfuscationVersionHeader = "Datadog-Obfuscation-Version" // header containing the version of obfuscation used, if any
+	containerIDHeader        = "Datadog-Container-ID"
+	entityIDHeader           = "Datadog-Entity-ID"
+	containerTagsHashHeader  = "Datadog-Container-Tags-Hash"
 
 	tracesAPIPath   = "/v0.4/traces"
 	tracesAPIPathV1 = "/v1.0/traces"
@@ -87,17 +91,32 @@ func datadogHeaders() map[string]string {
 		"Content-Type":                  "application/msgpack",
 	}
 	if cid := internal.ContainerID(); cid != "" {
-		h["Datadog-Container-ID"] = cid
+		h[containerIDHeader] = cid
 	}
 	if eid := internal.EntityID(); eid != "" {
-		h["Datadog-Entity-ID"] = eid
+		h[entityIDHeader] = eid
 	}
 	if extEnv := internal.ExternalEnvironment(); extEnv != "" {
 		h["Datadog-External-Env"] = extEnv
 	}
 	return h
 }
 
+func setContainerHeaders(h http.Header) {
+	if cid := internal.ContainerID(); cid != "" {
+		h.Set(containerIDHeader, cid)
+	}
+	if eid := internal.EntityID(); eid != "" {
+		h.Set(entityIDHeader, eid)
+	}
+}
+
+func updateContainerTagsHash(h http.Header) {
+	if hash := h.Get(containerTagsHashHeader); hash != "" {
+		processtags.SetContainerTagsHash(hash)
+	}
+}
+
 func (t *httpTransport) sendStats(p *pb.ClientStatsPayload, tracerObfuscationVersion int) error {
 	var buf bytes.Buffer
 	if err := msgp.Encode(&buf, p); err != nil {

ddtrace/tracer/transport_test.go

@@ -6,6 +6,7 @@
 package tracer
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net"
@@ -26,6 +27,7 @@ import (
 	tinternal "github.com/DataDog/dd-trace-go/v2/ddtrace/tracer/internal"
 	"github.com/DataDog/dd-trace-go/v2/internal"
 	internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config"
+	"github.com/DataDog/dd-trace-go/v2/internal/processtags"
 	"github.com/DataDog/dd-trace-go/v2/internal/statsdtest"
 
 	"github.com/stretchr/testify/assert"
@@ -183,6 +185,26 @@ func TestTransportResponse(t *testing.T) {
 	}
 }
 
+func TestFetchAgentFeaturesContainerTagsHash(t *testing.T) {
+	t.Cleanup(func() { processtags.SetContainerTagsHash("") })
+	processtags.SetContainerTagsHash("")
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, "/info", r.URL.Path)
+		w.Header().Set(containerTagsHashHeader, "info-container-hash")
+		w.Write([]byte(`{"endpoints":["/v0.6/stats"],"client_drop_p0s":true,"config":{}}`))
+	}))
+	defer srv.Close()
+
+	agentURL, err := url.Parse(srv.URL)
+	require.NoError(t, err)
+	features, err := fetchAgentFeatures(context.Background(), agentURL, srv.Client())
+	require.NoError(t, err)
+
+	assert.True(t, features.Stats)
+	assert.Equal(t, "info-container-hash", processtags.ContainerTagsHash())
+}
+
 func TestTraceCountHeader(t *testing.T) {
 	assert := assert.New(t)
 

internal/datastreams/hash_cache.go

@@ -19,7 +19,7 @@ type hashCache struct {
 	m  map[string]uint64
 }
 
-func getHashKey(edgeTags, processTags []string, parentHash uint64) string {
+func getHashKey(edgeTags, processTags []string, containerTagsHash string, parentHash uint64) string {
 	var s strings.Builder
 	l := 0
 	for _, t := range edgeTags {
@@ -28,6 +28,7 @@ func getHashKey(edgeTags, processTags []string, parentHash uint64) string {
 	for _, t := range processTags {
 		l += len(t)
 	}
+	l += len(containerTagsHash)
 	l += 8
 	s.Grow(l)
 	for _, t := range edgeTags {
@@ -36,6 +37,7 @@ func getHashKey(edgeTags, processTags []string, parentHash uint64) string {
 	for _, t := range processTags {
 		s.WriteString(t)
 	}
+	s.WriteString(containerTagsHash)
 	s.WriteByte(byte(parentHash))
 	s.WriteByte(byte(parentHash >> 8))
 	s.WriteByte(byte(parentHash >> 16))
@@ -47,8 +49,8 @@ func getHashKey(edgeTags, processTags []string, parentHash uint64) string {
 	return s.String()
 }
 
-func (c *hashCache) computeAndGet(key string, parentHash uint64, service, env string, edgeTags, processTags []string) uint64 {
-	hash := pathwayHash(nodeHash(service, env, edgeTags, processTags), parentHash)
+func (c *hashCache) computeAndGet(key string, parentHash uint64, service, env string, edgeTags, processTags []string, containerTagsHash string) uint64 {
+	hash := pathwayHash(nodeHash(service, env, edgeTags, processTags, containerTagsHash), parentHash)
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.m) >= maxHashCacheSize {
@@ -60,15 +62,15 @@ func (c *hashCache) computeAndGet(key string, parentHash uint64, service, env st
 	return hash
 }
 
-func (c *hashCache) get(service, env string, edgeTags, processTags []string, parentHash uint64) uint64 {
-	key := getHashKey(edgeTags, processTags, parentHash)
+func (c *hashCache) get(service, env string, edgeTags, processTags []string, containerTagsHash string, parentHash uint64) uint64 {
+	key := getHashKey(edgeTags, processTags, containerTagsHash, parentHash)
 	c.mu.RLock()
 	if hash, ok := c.m[key]; ok {
 		c.mu.RUnlock()
 		return hash
 	}
 	c.mu.RUnlock()
-	return c.computeAndGet(key, parentHash, service, env, edgeTags, processTags)
+	return c.computeAndGet(key, parentHash, service, env, edgeTags, processTags, containerTagsHash)
 }
 
 func newHashCache() *hashCache {

internal/datastreams/hash_cache_test.go

@@ -14,25 +14,29 @@ import (
 
 func TestHashCache(t *testing.T) {
 	cache := newHashCache()
-	assert.Equal(t, pathwayHash(nodeHash("service", "env", []string{"type:kafka"}, nil), 1234), cache.get("service", "env", []string{"type:kafka"}, nil, 1234))
+	assert.Equal(t, pathwayHash(nodeHash("service", "env", []string{"type:kafka"}, nil, ""), 1234), cache.get("service", "env", []string{"type:kafka"}, nil, "", 1234))
 	assert.Len(t, cache.m, 1)
-	assert.Equal(t, pathwayHash(nodeHash("service", "env", []string{"type:kafka"}, nil), 1234), cache.get("service", "env", []string{"type:kafka"}, nil, 1234))
+	assert.Equal(t, pathwayHash(nodeHash("service", "env", []string{"type:kafka"}, nil, ""), 1234), cache.get("service", "env", []string{"type:kafka"}, nil, "", 1234))
 	assert.Len(t, cache.m, 1)
-	assert.Equal(t, pathwayHash(nodeHash("service", "env", []string{"type:kafka2"}, nil), 1234), cache.get("service", "env", []string{"type:kafka2"}, nil, 1234))
+	assert.Equal(t, pathwayHash(nodeHash("service", "env", []string{"type:kafka2"}, nil, ""), 1234), cache.get("service", "env", []string{"type:kafka2"}, nil, "", 1234))
 	assert.Len(t, cache.m, 2)
 
 	pTags := []string{"entrypoint.name:something", "entrypoint.type:executable"}
-	h1 := pathwayHash(nodeHash("service", "env", []string{"type:kafka"}, pTags), 1234)
-	h2 := cache.get("service", "env", []string{"type:kafka"}, pTags, 1234)
+	h1 := pathwayHash(nodeHash("service", "env", []string{"type:kafka"}, pTags, "container-hash"), 1234)
+	h2 := cache.get("service", "env", []string{"type:kafka"}, pTags, "container-hash", 1234)
 
 	assert.Equal(t, h1, h2)
 	assert.Len(t, cache.m, 3)
+
+	h3 := cache.get("service", "env", []string{"type:kafka"}, pTags, "other-container-hash", 1234)
+	assert.NotEqual(t, h2, h3)
+	assert.Len(t, cache.m, 4)
 }
 
 func TestGetHashKey(t *testing.T) {
 	parentHash := uint64(87234)
-	key := getHashKey([]string{"type:kafka", "topic:topic1", "group:group1"}, []string{"entrypoint.name:something", "entrypoint.type:executable"}, parentHash)
+	key := getHashKey([]string{"type:kafka", "topic:topic1", "group:group1"}, []string{"entrypoint.name:something", "entrypoint.type:executable"}, "container-hash", parentHash)
 	hash := make([]byte, 8)
 	binary.LittleEndian.PutUint64(hash, parentHash)
-	assert.Equal(t, "type:kafkatopic:topic1group:group1entrypoint.name:somethingentrypoint.type:executable"+string(hash), key)
+	assert.Equal(t, "type:kafkatopic:topic1group:group1entrypoint.name:somethingentrypoint.type:executablecontainer-hash"+string(hash), key)
 }

internal/datastreams/pathway.go

@@ -26,7 +26,7 @@ func isWellFormedEdgeTag(t string) bool {
 	return false
 }
 
-func nodeHash(service, env string, edgeTags, processTags []string) uint64 {
+func nodeHash(service, env string, edgeTags, processTags []string, containerTagsHash string) uint64 {
 	h := fnv.New64()
 	sort.Strings(edgeTags)
 	h.Write([]byte(service))
@@ -41,6 +41,9 @@ func nodeHash(service, env string, edgeTags, processTags []string) uint64 {
 	for _, t := range processTags {
 		h.Write([]byte(t))
 	}
+	if containerTagsHash != "" {
+		h.Write([]byte(containerTagsHash))
+	}
 	return h.Sum64()
 }
 

internal/datastreams/pathway_test.go

@@ -35,9 +35,9 @@ func TestPathway(t *testing.T) {
 		end := middle.Add(time.Hour)
 		processor.timeSource = func() time.Time { return end }
 		ctx = processor.SetCheckpoint(ctx, "topic:topic2")
-		hash1 := pathwayHash(nodeHash("service-1", "env", nil, processTags), 0)
-		hash2 := pathwayHash(nodeHash("service-1", "env", []string{"topic:topic1"}, processTags), hash1)
-		hash3 := pathwayHash(nodeHash("service-1", "env", []string{"topic:topic2"}, processTags), hash2)
+		hash1 := pathwayHash(nodeHash("service-1", "env", nil, processTags, ""), 0)
+		hash2 := pathwayHash(nodeHash("service-1", "env", []string{"topic:topic1"}, processTags, ""), hash1)
+		hash3 := pathwayHash(nodeHash("service-1", "env", []string{"topic:topic2"}, processTags, ""), hash2)
 		p, _ := PathwayFromContext(ctx)
 		assert.Equal(t, hash3, p.GetHash())
 		assert.Equal(t, start, p.PathwayStart())
@@ -86,9 +86,9 @@ func TestPathway(t *testing.T) {
 		pathwayWith2EdgeTags, _ := PathwayFromContext(processor.SetCheckpoint(context.Background(), "type:internal", "some_other_key:some_other_val"))
 
 		processTags := processtags.GlobalTags().Slice()
-		hash1 := pathwayHash(nodeHash("service-1", "env", nil, processTags), 0)
-		hash2 := pathwayHash(nodeHash("service-1", "env", []string{"type:internal"}, processTags), 0)
-		hash3 := pathwayHash(nodeHash("service-1", "env", []string{"type:internal", "some_other_key:some_other_val"}, processTags), 0)
+		hash1 := pathwayHash(nodeHash("service-1", "env", nil, processTags, ""), 0)
+		hash2 := pathwayHash(nodeHash("service-1", "env", []string{"type:internal"}, processTags, ""), 0)
+		hash3 := pathwayHash(nodeHash("service-1", "env", []string{"type:internal", "some_other_key:some_other_val"}, processTags, ""), 0)
 		assert.Equal(t, hash1, pathwayWithNoEdgeTags.GetHash())
 		assert.Equal(t, hash2, pathwayWith1EdgeTag.GetHash())
 		assert.Equal(t, hash3, pathwayWith2EdgeTags.GetHash())
@@ -106,28 +106,32 @@ func TestPathway(t *testing.T) {
 
 	t.Run("test nodeHash", func(t *testing.T) {
 		assert.NotEqual(t,
-			nodeHash("service-1", "env", []string{"type:internal"}, nil),
-			nodeHash("service-1", "env", []string{"type:kafka"}, nil),
+			nodeHash("service-1", "env", []string{"type:internal"}, nil, ""),
+			nodeHash("service-1", "env", []string{"type:kafka"}, nil, ""),
 		)
 		assert.NotEqual(t,
-			nodeHash("service-1", "env", []string{"exchange:1"}, nil),
-			nodeHash("service-1", "env", []string{"exchange:2"}, nil),
+			nodeHash("service-1", "env", []string{"exchange:1"}, nil, ""),
+			nodeHash("service-1", "env", []string{"exchange:2"}, nil, ""),
 		)
 		assert.NotEqual(t,
-			nodeHash("service-1", "env", []string{"topic:1"}, nil),
-			nodeHash("service-1", "env", []string{"topic:2"}, nil),
+			nodeHash("service-1", "env", []string{"topic:1"}, nil, ""),
+			nodeHash("service-1", "env", []string{"topic:2"}, nil, ""),
 		)
 		assert.NotEqual(t,
-			nodeHash("service-1", "env", []string{"group:1"}, nil),
-			nodeHash("service-1", "env", []string{"group:2"}, nil),
+			nodeHash("service-1", "env", []string{"group:1"}, nil, ""),
+			nodeHash("service-1", "env", []string{"group:2"}, nil, ""),
 		)
 		assert.NotEqual(t,
-			nodeHash("service-1", "env", []string{"event_type:1"}, nil),
-			nodeHash("service-1", "env", []string{"event_type:2"}, nil),
+			nodeHash("service-1", "env", []string{"event_type:1"}, nil, ""),
+			nodeHash("service-1", "env", []string{"event_type:2"}, nil, ""),
 		)
 		assert.Equal(t,
-			nodeHash("service-1", "env", []string{"partition:0"}, nil),
-			nodeHash("service-1", "env", []string{"partition:1"}, nil),
+			nodeHash("service-1", "env", []string{"partition:0"}, nil, ""),
+			nodeHash("service-1", "env", []string{"partition:1"}, nil, ""),
+		)
+		assert.NotEqual(t,
+			nodeHash("service-1", "env", []string{"type:kafka"}, nil, "container-hash-1"),
+			nodeHash("service-1", "env", []string{"type:kafka"}, nil, "container-hash-2"),
 		)
 	})
 
@@ -171,7 +175,7 @@ func TestPathway(t *testing.T) {
 	})
 
 	t.Run("test GetHash", func(t *testing.T) {
-		pathway := Pathway{hash: nodeHash("service", "env", []string{"direction:in"}, nil)}
+		pathway := Pathway{hash: nodeHash("service", "env", []string{"direction:in"}, nil, "")}
 		assert.Equal(t, pathway.hash, pathway.GetHash())
 	})
 }
@@ -187,6 +191,6 @@ func BenchmarkNodeHash(b *testing.B) {
 	env := "test"
 	edgeTags := []string{"event_type:dog", "exchange:local", "group:all", "topic:off", "type:writer"}
 	for b.Loop() {
-		nodeHash(service, env, edgeTags, nil)
+		nodeHash(service, env, edgeTags, nil, "")
 	}
 }

internal/datastreams/processor.go

@@ -641,9 +641,14 @@ func (p *Processor) SetCheckpointWithParams(ctx context.Context, params options.
 	if params.ServiceOverride != "" {
 		service = params.ServiceOverride
 	}
-	processTags := processtags.GlobalTags().Slice()
+	var processTags []string
+	var containerTagsHash string
+	if pTags := processtags.GlobalTags(); pTags != nil {
+		processTags = pTags.Slice()
+		containerTagsHash = processtags.ContainerTagsHash()
+	}
 	child := Pathway{
-		hash:         p.hashCache.get(service, p.env, edgeTags, processTags, parentHash),
+		hash:         p.hashCache.get(service, p.env, edgeTags, processTags, containerTagsHash, parentHash),
 		pathwayStart: pathwayStart,
 		edgeStart:    now,
 	}