APMSP-1241 Directly import trace-agent stats code for client-side stats

.github/workflows/unit-integration-tests.yml

@@ -261,7 +261,7 @@ jobs:
 
       - name: Testing outlier google.golang.org/api
         run: |
-              go get google.golang.org/api@v0.121.0 # version used to generate code
+              go get google.golang.org/api@v0.192.0 # version used to generate code
               go mod tidy # Go1.16 doesn't update the sum file correctly after the go get, this tidy fixes it
               go test -v ./contrib/google.golang.org/api/...
 

.gitlab/macrobenchmarks.yml

@@ -56,10 +56,10 @@ variables:
 # 2. Rebuild image in Gitlab CI (build-images CI job in https://gitlab.ddbuild.io/DataDog/apm-reliability/benchmarking-platform/-/pipelines?page=1&scope=all&ref=go%2Fgo-prof-app)
 #
 
-.go121-benchmarks:
+.go123-benchmarks:
   extends: .benchmarks
   variables:
-    GO_VERSION: "1.21.12"
+    GO_VERSION: "1.23.0"
 
 .go122-benchmarks:
   extends: .benchmarks
@@ -123,53 +123,53 @@ go122-profile-trace-asm:
     ENABLE_APPSEC: "true"
     DD_PROFILING_EXECUTION_TRACE_ENABLED: "false"
 
-go121-baseline:
-  extends: .go121-benchmarks
+go123-baseline:
+  extends: .go123-benchmarks
   variables:
     ENABLE_DDPROF: "false"
     ENABLE_TRACING: "false"
     ENABLE_PROFILING: "false"
     ENABLE_APPSEC: "false"
     DD_PROFILING_EXECUTION_TRACE_ENABLED: "false"
 
-go121-only-trace:
-  extends: .go121-benchmarks
+go123-only-trace:
+  extends: .go123-benchmarks
   variables:
     ENABLE_DDPROF: "false"
     ENABLE_TRACING: "true"
     ENABLE_PROFILING: "false"
     ENABLE_APPSEC: "false"
     DD_PROFILING_EXECUTION_TRACE_ENABLED: "false"
 
-go121-only-profile:
-  extends: .go121-benchmarks
+go123-only-profile:
+  extends: .go123-benchmarks
   variables:
     ENABLE_DDPROF: "false"
     ENABLE_TRACING: "false"
     ENABLE_PROFILING: "true"
     ENABLE_APPSEC: "false"
     DD_PROFILING_EXECUTION_TRACE_ENABLED: "false"
 
-go121-profile-trace:
-  extends: .go121-benchmarks
+go123-profile-trace:
+  extends: .go123-benchmarks
   variables:
     ENABLE_DDPROF: "false"
     ENABLE_TRACING: "true"
     ENABLE_PROFILING: "true"
     ENABLE_APPSEC: "false"
     DD_PROFILING_EXECUTION_TRACE_ENABLED: "false"
 
-go121-trace-asm:
-  extends: .go121-benchmarks
+go123-trace-asm:
+  extends: .go123-benchmarks
   variables:
     ENABLE_DDPROF: "false"
     ENABLE_TRACING: "true"
     ENABLE_PROFILING: "false"
     ENABLE_APPSEC: "true"
     DD_PROFILING_EXECUTION_TRACE_ENABLED: "false"
 
-go121-profile-trace-asm:
-  extends: .go121-benchmarks
+go123-profile-trace-asm:
+  extends: .go123-benchmarks
   variables:
     ENABLE_DDPROF: "false"
     ENABLE_TRACING: "true"

contrib/google.golang.org/api/internal/gen_endpoints/main.go

@@ -14,7 +14,6 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
-	"github.com/yosida95/uritemplate/v3"
 	"io"
 	"log"
 	"net/http"
@@ -24,11 +23,13 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+
+	"github.com/yosida95/uritemplate/v3"
 )
 
 const (
 	// The github.com/googleapis/google-api-go-client version to use.
-	version = "v0.121.0"
+	version = "v0.192.0"
 )
 
 var (

ddtrace/mocktracer/main_test.go

@@ -6,10 +6,12 @@
 package mocktracer
 
 import (
-	"go.uber.org/goleak"
 	"testing"
+
+	"go.uber.org/goleak"
 )
 
 func TestMain(m *testing.M) {
-	goleak.VerifyTestMain(m)
+	// TODO: seelog (indirect dependency) has a known goroutine leak where it leaks a single goroutine on init (https://github.com/cihub/seelog/issues/182)
+	goleak.VerifyTestMain(m, goleak.IgnoreAnyFunction("github.com/cihub/seelog.(*asyncLoopLogger).processQueue"))
 }

ddtrace/tracer/civisibility_transport.go

@@ -15,6 +15,7 @@ import (
 	"runtime"
 	"strings"
 
+	pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/civisibility/constants"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
@@ -186,7 +187,7 @@ func (t *ciVisibilityTransport) send(p *payload) (body io.ReadCloser, err error)
 // Returns:
 //
 //	An error indicating that stats are not supported.
-func (t *ciVisibilityTransport) sendStats(*statsPayload) error {
+func (t *ciVisibilityTransport) sendStats(*pb.ClientStatsPayload) error {
 	// Stats are not supported by CI Visibility agentless / EVP proxy.
 	return nil
 }

ddtrace/tracer/option.go

@@ -628,6 +628,12 @@ type agentFeatures struct {
 
 	// featureFlags specifies all the feature flags reported by the trace-agent.
 	featureFlags map[string]struct{}
+
+	// peerTags specifies precursor tags to aggregate stats on when client stats is enabled
+	peerTags []string
+
+	// defaultEnv is the trace-agent's default env, used for stats calculation if no env override is present
+	defaultEnv string
 }
 
 // HasFlag reports whether the agent has set the feat feature flag.
@@ -653,12 +659,18 @@ func loadAgentFeatures(agentDisabled bool, agentURL *url.URL, httpClient *http.C
 		return
 	}
 	defer resp.Body.Close()
+	type agentConfig struct {
+		defaultEnv string `json:"default_env"`
+	}
 	type infoResponse struct {
-		Endpoints     []string `json:"endpoints"`
-		ClientDropP0s bool     `json:"client_drop_p0s"`
-		StatsdPort    int      `json:"statsd_port"`
-		FeatureFlags  []string `json:"feature_flags"`
+		Endpoints     []string    `json:"endpoints"`
+		ClientDropP0s bool        `json:"client_drop_p0s"`
+		StatsdPort    int         `json:"statsd_port"`
+		FeatureFlags  []string    `json:"feature_flags"`
+		PeerTags      []string    `json:"peer_tags"`
+		Config        agentConfig `json:"config"`
 	}
+
 	var info infoResponse
 	if err := json.NewDecoder(resp.Body).Decode(&info); err != nil {
 		log.Error("Decoding features: %v", err)

ddtrace/tracer/span.go

@@ -11,7 +11,6 @@ import (
 	"context"
 	"encoding/base64"
 	"fmt"
-	"math"
 	"os"
 	"reflect"
 	"runtime"
@@ -33,9 +32,10 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/samplernames"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/traceprof"
 
-	"github.com/DataDog/datadog-agent/pkg/obfuscate"
 	"github.com/tinylib/msgp/msgp"
 	"golang.org/x/xerrors"
+
+	"github.com/DataDog/datadog-agent/pkg/obfuscate"
 )
 
 type (
@@ -552,13 +552,16 @@ func (s *span) finish(finishTime int64) {
 			return
 		}
 		// we have an active tracer
-		if t.config.canComputeStats() && shouldComputeStats(s) {
-			// the agent supports computed stats
-			select {
-			case t.stats.In <- newAggregableSpan(s, t.obfuscator):
-				// ok
-			default:
-				log.Error("Stats channel full, disregarding span.")
+		if t.config.canComputeStats() {
+			statSpan, shouldCalc := t.stats.newTracerStatSpan(s, t.obfuscator)
+			if shouldCalc {
+				// the agent supports computed stats
+				select {
+				case t.stats.In <- statSpan:
+					// ok
+				default:
+					log.Error("Stats channel full, disregarding span.")
+				}
 			}
 		}
 		if t.config.canDropP0s() {
@@ -593,40 +596,6 @@ func (s *span) finish(finishTime int64) {
 	}
 }
 
-// newAggregableSpan creates a new summary for the span s, within an application
-// version version.
-func newAggregableSpan(s *span, obfuscator *obfuscate.Obfuscator) *aggregableSpan {
-	var statusCode uint32
-	if sc, ok := s.Meta["http.status_code"]; ok && sc != "" {
-		if c, err := strconv.Atoi(sc); err == nil && c > 0 && c <= math.MaxInt32 {
-			statusCode = uint32(c)
-		}
-	}
-	var isTraceRoot trilean
-	if s.ParentID == 0 {
-		isTraceRoot = trileanTrue
-	} else {
-		isTraceRoot = trileanFalse
-	}
-
-	key := aggregation{
-		Name:        s.Name,
-		Resource:    obfuscatedResource(obfuscator, s.Type, s.Resource),
-		Service:     s.Service,
-		Type:        s.Type,
-		Synthetics:  strings.HasPrefix(s.Meta[keyOrigin], "synthetics"),
-		StatusCode:  statusCode,
-		IsTraceRoot: isTraceRoot,
-	}
-	return &aggregableSpan{
-		key:      key,
-		Start:    s.Start,
-		Duration: s.Duration,
-		TopLevel: s.Metrics[keyTopLevel] == 1,
-		Error:    s.Error,
-	}
-}
-
 // textNonParsable specifies the text that will be assigned to resources for which the resource
 // can not be parsed due to an obfuscation error.
 const textNonParsable = "Non-parsable SQL query"

ddtrace/tracer/span_test.go

@@ -22,7 +22,6 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/samplernames"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/traceprof"
 
-	"github.com/DataDog/datadog-agent/pkg/obfuscate"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -165,41 +164,6 @@ func TestShouldComputeStats(t *testing.T) {
 	}
 }
 
-func TestNewAggregableSpan(t *testing.T) {
-	t.Run("obfuscating", func(t *testing.T) {
-		o := obfuscate.NewObfuscator(obfuscate.Config{})
-		aggspan := newAggregableSpan(&span{
-			Name:     "name",
-			Resource: "SELECT * FROM table WHERE password='secret'",
-			Service:  "service",
-			Type:     "sql",
-		}, o)
-		assert.Equal(t, aggregation{
-			Name:        "name",
-			Type:        "sql",
-			Resource:    "SELECT * FROM table WHERE password = ?",
-			Service:     "service",
-			IsTraceRoot: 1,
-		}, aggspan.key)
-	})
-
-	t.Run("nil-obfuscator", func(t *testing.T) {
-		aggspan := newAggregableSpan(&span{
-			Name:     "name",
-			Resource: "SELECT * FROM table WHERE password='secret'",
-			Service:  "service",
-			Type:     "sql",
-		}, nil)
-		assert.Equal(t, aggregation{
-			Name:        "name",
-			Type:        "sql",
-			Resource:    "SELECT * FROM table WHERE password='secret'",
-			Service:     "service",
-			IsTraceRoot: 1,
-		}, aggspan.key)
-	})
-}
-
 func TestSpanFinishWithTime(t *testing.T) {
 	assert := assert.New(t)
 

ddtrace/tracer/spancontext_test.go

@@ -40,7 +40,7 @@ func TestNewSpanContextPushError(t *testing.T) {
 
 	tp := new(log.RecordLogger)
 	tp.Ignore("appsec: ", telemetry.LogPrefix)
-	_, _, _, stop := startTestTracer(t, WithLogger(tp), WithLambdaMode(true))
+	_, _, _, stop := startTestTracer(t, WithLogger(tp), WithLambdaMode(true), WithEnv("testEnv"))
 	defer stop()
 	parent := newBasicSpan("test1")                  // 1st span in trace
 	parent.context.trace.push(newBasicSpan("test2")) // 2nd span in trace
@@ -51,6 +51,7 @@ func TestNewSpanContextPushError(t *testing.T) {
 	child.context = newSpanContext(child, parent.context)
 
 	log.Flush()
+
 	assert.Contains(t, tp.Logs()[0], "ERROR: trace buffer full (2)")
 }
 
@@ -242,7 +243,7 @@ func TestSpanTracePushNoFinish(t *testing.T) {
 
 	tp := new(log.RecordLogger)
 	tp.Ignore("appsec: ", telemetry.LogPrefix)
-	_, _, _, stop := startTestTracer(t, WithLogger(tp), WithLambdaMode(true))
+	_, _, _, stop := startTestTracer(t, WithLogger(tp), WithLambdaMode(true), WithEnv("testEnv"))
 	defer stop()
 
 	buffer := newTrace()
@@ -756,7 +757,7 @@ func TestSpanContextPushFull(t *testing.T) {
 	traceMaxSize = 2
 	tp := new(log.RecordLogger)
 	tp.Ignore("appsec: ", telemetry.LogPrefix)
-	_, _, _, stop := startTestTracer(t, WithLogger(tp), WithLambdaMode(true))
+	_, _, _, stop := startTestTracer(t, WithLogger(tp), WithLambdaMode(true), WithEnv("testEnv"))
 	defer stop()
 
 	span1 := newBasicSpan("span1")