support process tags in apm stats - first pass

Author: raphaelgavache

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/proto/pbgo/trace/stats.pb.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"
+	processTags       = "_dd.tags.process"
 )
 
 // TagStats returns the stats and tags coinciding with the information found in header.
@@ -664,6 +665,7 @@ func (r *HTTPReceiver) handleTraces(v Version, w http.ResponseWriter, req *http.
 		ClientComputedTopLevel: isHeaderTrue(header.ComputedTopLevel, req.Header.Get(header.ComputedTopLevel)),
 		ClientComputedStats:    isHeaderTrue(header.ComputedStats, req.Header.Get(header.ComputedStats)),
 		ClientDroppedP0s:       droppedTracesFromHeader(req.Header, ts),
+		ProcessTags:            getProcessTagsFromHeader(req.Header, ts),
 	}
 	r.out <- payload
 }
@@ -700,6 +702,10 @@ func droppedTracesFromHeader(h http.Header, ts *info.TagStats) int64 {
 	return dropped
 }
 
+func getProcessTagsFromHeader(h http.Header, ts *info.TagStats) 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/proto/pbgo/trace/stats_gen.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/proto/pbgo/trace/stats_vtproto.pb.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,10 @@ func NewAggregationFromSpan(s *StatSpan, origin string, aggKey PayloadAggregatio
 	return agg
 }
 
+func processTagsHash(processTags string) uint64 {
+	return peerTagsHash(strings.Split(processTags, ","))
+}
+
 func peerTagsHash(tags []string) uint64 {
 	if len(tags) == 0 {
 		return 0

pkg/trace/api/api.go

@@ -41,6 +41,7 @@ type Concentrator struct {
 	exit          chan struct{}
 	exitWG        sync.WaitGroup
 	cidStats      bool
+	processStats  bool
 	agentEnv      string
 	agentHostname string
 	agentVersion  string
@@ -56,11 +57,13 @@ func NewConcentrator(conf *config.AgentConfig, writer Writer, now time.Time, sta
 		BucketInterval:         bsize,
 	}, now)
 	_, disabledCIDStats := conf.Features["disable_cid_stats"]
+	_, disabledProcessStats := conf.Features["disable_process_stats"]
 	c := Concentrator{
 		spanConcentrator: sc,
 		Writer:           writer,
 		exit:             make(chan struct{}),
 		cidStats:         !disabledCIDStats,
+		processStats:     !disabledProcessStats,
 		agentEnv:         conf.DefaultEnv,
 		agentHostname:    conf.Hostname,
 		agentVersion:     conf.AgentVersion,
@@ -113,6 +116,7 @@ type Input struct {
 	Traces        []traceutil.ProcessedTrace
 	ContainerID   string
 	ContainerTags []string
+	ProcessTags   string
 }
 
 // NewStatsInput allocates a stats input for an incoming trace payload
@@ -125,16 +129,33 @@ func NewStatsInput(numChunks int, containerID string, clientComputedStats bool)
 
 // Add applies the given input to the concentrator.
 func (c *Concentrator) Add(t Input) {
+	tags := infraTags{
+		containerID:     t.ContainerID,
+		containerTags:   t.ContainerTags,
+		processTagsHash: processTagsHash(t.ProcessTags),
+		processTags:     t.ProcessTags,
+	}
 	for _, trace := range t.Traces {
-		c.addNow(&trace, t.ContainerID, t.ContainerTags)
+		c.addNow(&trace, tags)
 	}
 }
 
+type infraTags struct {
+	containerID     string
+	containerTags   []string
+	processTagsHash uint64
+	processTags     string
+}
+
 // addNow adds the given input into the concentrator.
 // Callers must guard!
-func (c *Concentrator) addNow(pt *traceutil.ProcessedTrace, containerID string, containerTags []string) {
+func (c *Concentrator) addNow(pt *traceutil.ProcessedTrace, tags infraTags) {
 	if !c.cidStats {
-		containerID = ""
+		tags.containerID = ""
+	}
+	if !c.processStats {
+		tags.processTagsHash = 0
+		tags.processTags = ""
 	}
 	hostname := pt.TracerHostname
 	if hostname == "" {
@@ -146,17 +167,18 @@ func (c *Concentrator) addNow(pt *traceutil.ProcessedTrace, containerID string,
 	}
 	weight := weight(pt.Root)
 	aggKey := PayloadAggregationKey{
-		Env:          env,
-		Hostname:     hostname,
-		Version:      pt.AppVersion,
-		ContainerID:  containerID,
-		GitCommitSha: pt.GitCommitSha,
-		ImageTag:     pt.ImageTag,
+		Env:             env,
+		Hostname:        hostname,
+		Version:         pt.AppVersion,
+		ContainerID:     tags.containerID,
+		GitCommitSha:    pt.GitCommitSha,
+		ImageTag:        pt.ImageTag,
+		ProcessTagsHash: tags.processTagsHash,
 	}
 	for _, s := range pt.TraceChunk.Spans {
 		statSpan, ok := c.spanConcentrator.NewStatSpanFromPB(s, c.peerTagKeys)
 		if ok {
-			c.spanConcentrator.addSpan(statSpan, aggKey, containerID, containerTags, pt.TraceChunk.Origin, weight)
+			c.spanConcentrator.addSpan(statSpan, aggKey, tags, pt.TraceChunk.Origin, weight)
 		}
 	}
 }