support process tags in apm stats and spans

pkg/proto/datadog/trace/stats.proto

@@ -49,6 +49,10 @@ message ClientStatsPayload {
 	string git_commit_sha = 13;
 	// The image tag is obtained from a container's set of tags.
 	string image_tag = 14;
+        // The process tags hash is used as a key for agent stats agregation.
+        uint64 process_tags_hash = 15;
+        // The process tags contains a list of tags that are specific to the process.
+        string process_tags = 16;
 }
 
 // ClientStatsBucket is a time bucket containing aggregated stats.

pkg/trace/agent/agent.go

@@ -313,7 +313,7 @@ func (a *Agent) Process(p *api.Payload) {
 	defer a.Timing.Since("datadog.trace_agent.internal.process_payload_ms", now)
 	ts := p.Source
 	sampledChunks := new(writer.SampledChunks)
-	statsInput := stats.NewStatsInput(len(p.TracerPayload.Chunks), p.TracerPayload.ContainerID, p.ClientComputedStats)
+	statsInput := stats.NewStatsInput(len(p.TracerPayload.Chunks), p.TracerPayload.ContainerID, p.ClientComputedStats, p.ProcessTags)
 
 	p.TracerPayload.Env = traceutil.NormalizeTagValue(p.TracerPayload.Env)
 

pkg/trace/agent/agent_test.go

@@ -2450,9 +2450,11 @@ func TestConvertStats(t *testing.T) {
 			name:     "containerID feature enabled, no fargate",
 			features: "enable_cid_stats",
 			in: &pb.ClientStatsPayload{
-				Hostname: "tracer_hots",
-				Env:      "tracer_env",
-				Version:  "code_version",
+				Hostname:        "tracer_hots",
+				Env:             "tracer_env",
+				Version:         "code_version",
+				ProcessTags:     "binary_name:bin",
+				ProcessTagsHash: 123456789,
 				Stats: []*pb.ClientStatsBucket{
 					{
 						Start:    1,
@@ -2487,12 +2489,14 @@ func TestConvertStats(t *testing.T) {
 			tracerVersion: "v1",
 			containerID:   "abc123",
 			out: &pb.ClientStatsPayload{
-				Hostname:      "tracer_hots",
-				Env:           "tracer_env",
-				Version:       "code_version",
-				Lang:          "java",
-				TracerVersion: "v1",
-				ContainerID:   "abc123",
+				Hostname:        "tracer_hots",
+				Env:             "tracer_env",
+				Version:         "code_version",
+				Lang:            "java",
+				TracerVersion:   "v1",
+				ContainerID:     "abc123",
+				ProcessTags:     "binary_name:bin",
+				ProcessTagsHash: 123456789,
 				Stats: []*pb.ClientStatsBucket{
 					{
 						Start:    1,

pkg/trace/api/api.go

@@ -440,6 +440,7 @@ const (
 	// tagContainersTags specifies the name of the tag which holds key/value
 	// pairs representing information about the container (Docker, EC2, etc).
 	tagContainersTags = "_dd.tags.container"
+	tagProcessTags    = "_dd.tags.process"
 )
 
 // TagStats returns the stats and tags coinciding with the information found in header.
@@ -657,13 +658,21 @@ func (r *HTTPReceiver) handleTraces(v Version, w http.ResponseWriter, req *http.
 		}
 		tp.Tags[tagContainersTags] = ctags
 	}
+	ptags := getProcessTagsFromHeader(req.Header)
+	if ptags != "" {
+		if tp.Tags == nil {
+			tp.Tags = make(map[string]string)
+		}
+		tp.Tags[tagProcessTags] = ptags
+	}
 
 	payload := &Payload{
 		Source:                 ts,
 		TracerPayload:          tp,
 		ClientComputedTopLevel: isHeaderTrue(header.ComputedTopLevel, req.Header.Get(header.ComputedTopLevel)),
 		ClientComputedStats:    isHeaderTrue(header.ComputedStats, req.Header.Get(header.ComputedStats)),
 		ClientDroppedP0s:       droppedTracesFromHeader(req.Header, ts),
+		ProcessTags:            ptags,
 	}
 	r.out <- payload
 }
@@ -700,6 +709,10 @@ func droppedTracesFromHeader(h http.Header, ts *info.TagStats) int64 {
 	return dropped
 }
 
+func getProcessTagsFromHeader(h http.Header) string {
+	return h.Get(header.ProcessTags)
+}
+
 // handleServices handle a request with a list of several services
 func (r *HTTPReceiver) handleServices(_ Version, w http.ResponseWriter, _ *http.Request) {
 	httpOK(w)

pkg/trace/api/internal/header/headers.go

@@ -11,6 +11,9 @@ const (
 	// with the number of traces contained in the payload.
 	TraceCount = "X-Datadog-Trace-Count"
 
+	// ProcessTags is a list that contains process tags split by a ','.
+	ProcessTags = "X-Datadog-Process-Tags"
+
 	// ContainerID specifies the name of the header which contains the ID of the
 	// container where the request originated.
 	// Deprecated in favor of Datadog-Entity-ID.

pkg/trace/api/payload.go

@@ -29,6 +29,9 @@ type Payload struct {
 
 	// ClientDroppedP0s specifies the number of P0 traces chunks dropped by the client.
 	ClientDroppedP0s int64
+
+	// ProcessTags is a list of tags describing an instrumented process.
+	ProcessTags string
 }
 
 // Chunks returns chunks in TracerPayload

pkg/trace/stats/aggregation.go

@@ -46,12 +46,13 @@ type BucketsAggregationKey struct {
 
 // PayloadAggregationKey specifies the key by which a payload is aggregated.
 type PayloadAggregationKey struct {
-	Env          string
-	Hostname     string
-	Version      string
-	ContainerID  string
-	GitCommitSha string
-	ImageTag     string
+	Env             string
+	Hostname        string
+	Version         string
+	ContainerID     string
+	GitCommitSha    string
+	ImageTag        string
+	ProcessTagsHash uint64
 }
 
 func getStatusCode(meta map[string]string, metrics map[string]float64) uint32 {
@@ -99,6 +100,13 @@ func NewAggregationFromSpan(s *StatSpan, origin string, aggKey PayloadAggregatio
 	return agg
 }
 
+func processTagsHash(processTags string) uint64 {
+	if processTags == "" {
+		return 0
+	}
+	return peerTagsHash(strings.Split(processTags, ","))
+}
+
 func peerTagsHash(tags []string) uint64 {
 	if len(tags) == 0 {
 		return 0