Merge branch 'main' into rachel.yang/baggage-prop

.github/actions/add-codeowners/codeowners.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+
+for file in "$@"; do
+    temp_file="tempfile.xml"
+
+    # force write a new line at the end of the gotestsum-report.xml, or else
+    # the loop will skip the last line.
+    # fixes issue with a missing </testsuites>
+    echo -e "\n" >> $file
+
+    while read p; do
+        # we might try to report gotestsum-report.xml multiple times, so don't
+        # calculate codeowners more times than we need
+        if [[ "$p" =~ \<testcase && ! "$p" =~ "file=" ]]; then
+            class=$(echo "$p" | grep -o '.v1/[^"]*"')
+            file_name=$(echo "${class:3}" | sed 's/.$//') # trim off the edges to get the path
+            new_line=$(echo "$p" | sed "s|<testcase \([^>]*\)>|<testcase \1 file=\"$file_name\">|")
+            echo "$new_line" >> "$temp_file"
+        else 
+            echo "$p" >> "$temp_file"
+        fi
+    done < $file
+
+    mv "$temp_file" $file
+done 

.github/actions/dd-ci-upload/action.yml

@@ -42,6 +42,10 @@ runs:
         curl -L --fail "https://github.com/DataDog/datadog-ci/releases/latest/download/datadog-ci_${{ env.DD_CI_CLI_BUILD }}" --output datadog-ci    
         chmod +x datadog-ci
 
+    - name: Add CodeOwners to JUnit files
+      shell: bash
+      run: cd ./.github/actions/add-codeowners && ./codeowners.sh ${{ inputs.files }}
+
     - name: Upload the JUnit files
       shell: bash
       run: |

.github/workflows/service-extensions-publish.yml

@@ -3,7 +3,7 @@ name: Publish Service Extensions Callout images packages
 on:
   push:
     tags:
-      - 'v*.*'
+      - 'v[0-9]+.[0-9]+.[0-9]+'
   workflow_dispatch:
     inputs:
       tag_name:

.github/workflows/system-tests.yml

@@ -108,6 +108,9 @@ jobs:
             weblog-variant: graphql-go
           - scenario: GRAPHQL_APPSEC
             weblog-variant: gqlgen
+          # Service Extensions System Tests
+          - scenario: EXTERNAL_PROCESSING
+          - scenario: EXTERNAL_PROCESSING_BLOCKING
 
       fail-fast: false
     env:

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

@@ -144,6 +144,12 @@ jobs:
         image: redis:3.2
         ports:
           - 6379:6379
+      valkey:
+        image: valkey/valkey:8
+        env:
+          VALKEY_EXTRA_FLAGS: "--port 6380 --requirepass password-for-default"
+        ports:
+          - 6380:6380
       elasticsearch2:
         image: elasticsearch:2
         env:

contrib/database/sql/metrics.go

@@ -33,20 +33,27 @@ var interval = 10 * time.Second
 
 // pollDBStats calls (*DB).Stats on the db at a predetermined interval. It pushes the DBStats off to the statsd client.
 // the caller should always ensure that db & statsd are non-nil
-func pollDBStats(statsd internal.StatsdClient, db *sql.DB) {
+func pollDBStats(statsd internal.StatsdClient, db *sql.DB, stop chan struct{}) {
 	log.Debug("DB stats will be gathered and sent every %v.", interval)
-	for range time.NewTicker(interval).C {
-		log.Debug("Reporting DB.Stats metrics...")
-		stat := db.Stats()
-		statsd.Gauge(MaxOpenConnections, float64(stat.MaxOpenConnections), []string{}, 1)
-		statsd.Gauge(OpenConnections, float64(stat.OpenConnections), []string{}, 1)
-		statsd.Gauge(InUse, float64(stat.InUse), []string{}, 1)
-		statsd.Gauge(Idle, float64(stat.Idle), []string{}, 1)
-		statsd.Gauge(WaitCount, float64(stat.WaitCount), []string{}, 1)
-		statsd.Timing(WaitDuration, stat.WaitDuration, []string{}, 1)
-		statsd.Gauge(MaxIdleClosed, float64(stat.MaxIdleClosed), []string{}, 1)
-		statsd.Gauge(MaxIdleTimeClosed, float64(stat.MaxIdleTimeClosed), []string{}, 1)
-		statsd.Gauge(MaxLifetimeClosed, float64(stat.MaxLifetimeClosed), []string{}, 1)
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			log.Debug("Reporting DB.Stats metrics...")
+			stat := db.Stats()
+			statsd.Gauge(MaxOpenConnections, float64(stat.MaxOpenConnections), []string{}, 1)
+			statsd.Gauge(OpenConnections, float64(stat.OpenConnections), []string{}, 1)
+			statsd.Gauge(InUse, float64(stat.InUse), []string{}, 1)
+			statsd.Gauge(Idle, float64(stat.Idle), []string{}, 1)
+			statsd.Gauge(WaitCount, float64(stat.WaitCount), []string{}, 1)
+			statsd.Timing(WaitDuration, stat.WaitDuration, []string{}, 1)
+			statsd.Gauge(MaxIdleClosed, float64(stat.MaxIdleClosed), []string{}, 1)
+			statsd.Gauge(MaxIdleTimeClosed, float64(stat.MaxIdleTimeClosed), []string{}, 1)
+			statsd.Gauge(MaxLifetimeClosed, float64(stat.MaxLifetimeClosed), []string{}, 1)
+		case <-stop:
+			return
+		}
 	}
 }
 

contrib/database/sql/metrics_test.go

@@ -6,9 +6,13 @@
 package sql
 
 import (
+	"sync"
 	"testing"
 
+	"github.com/DataDog/datadog-go/v5/statsd"
+	"github.com/lib/pq"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/globalconfig"
 )
 
@@ -64,3 +68,22 @@ func TestStatsTags(t *testing.T) {
 	})
 	resetGlobalConfig()
 }
+
+func TestPollDBStatsStop(t *testing.T) {
+	driverName := "postgres"
+	Register(driverName, &pq.Driver{}, WithServiceName("postgres-test"), WithAnalyticsRate(0.2))
+	defer unregister(driverName)
+	db, err := Open(driverName, "postgres://postgres:[email protected]:5432/postgres?sslmode=disable")
+	require.NoError(t, err)
+	defer db.Close()
+
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		pollDBStats(&statsd.NoOpClientDirect{}, db, stop)
+	}()
+	close(stop)
+	wg.Wait()
+}

contrib/database/sql/sql.go

@@ -139,6 +139,7 @@ type tracedConnector struct {
 	connector  driver.Connector
 	driverName string
 	cfg        *config
+	dbClose    chan struct{}
 }
 
 func (t *tracedConnector) Connect(ctx context.Context) (driver.Conn, error) {
@@ -171,6 +172,13 @@ func (t *tracedConnector) Driver() driver.Driver {
 	return t.connector.Driver()
 }
 
+// Close closes the dbClose channel
+// This method will be invoked when DB.Close() is called, which we expect to occur only once: https://cs.opensource.google/go/go/+/refs/tags/go1.23.4:src/database/sql/sql.go;l=918-950
+func (t *tracedConnector) Close() error {
+	close(t.dbClose)
+	return nil
+}
+
 // from Go stdlib implementation of sql.Open
 type dsnConnector struct {
 	dsn    string
@@ -208,10 +216,11 @@ func OpenDB(c driver.Connector, opts ...Option) *sql.DB {
 		connector:  c,
 		driverName: driverName,
 		cfg:        cfg,
+		dbClose:    make(chan struct{}),
 	}
 	db := sql.OpenDB(tc)
 	if cfg.dbStats && cfg.statsdClient != nil {
-		go pollDBStats(cfg.statsdClient, db)
+		go pollDBStats(cfg.statsdClient, db, tc.dbClose)
 	}
 	return db
 }

contrib/database/sql/sql_test.go

@@ -281,12 +281,13 @@ func TestOpenOptions(t *testing.T) {
 		var tg statsdtest.TestStatsdClient
 		Register(driverName, &pq.Driver{})
 		defer unregister(driverName)
-		_, err := Open(driverName, dsn, withStatsdClient(&tg), WithDBStats())
+		db, err := Open(driverName, dsn, withStatsdClient(&tg), WithDBStats())
 		require.NoError(t, err)
 
 		// The polling interval has been reduced to 500ms for the sake of this test, so at least one round of `pollDBStats` should be complete in 1s
 		deadline := time.Now().Add(1 * time.Second)
 		wantStats := []string{MaxOpenConnections, OpenConnections, InUse, Idle, WaitCount, WaitDuration, MaxIdleClosed, MaxIdleTimeClosed, MaxLifetimeClosed}
+		var calls1 []string
 		for {
 			if time.Now().After(deadline) {
 				t.Fatalf("Stats not collected in expected interval of %v", interval)
@@ -300,11 +301,16 @@ func TestOpenOptions(t *testing.T) {
 					}
 				}
 				// all expected stats have been collected; exit out of loop, test should pass
+				calls1 = calls
 				break
 			}
 			// not all stats have been collected yet, try again in 50ms
 			time.Sleep(50 * time.Millisecond)
 		}
+		// Close DB and assert the no further stats have been collected; db.Close should stop the pollDBStats goroutine.
+		db.Close()
+		time.Sleep(50 * time.Millisecond)
+		assert.Equal(t, calls1, tg.CallNames())
 	})
 }
 

contrib/emicklei/go-restful.v3/restful.go

@@ -46,9 +46,9 @@ func FilterFunc(configOpts ...Option) restful.FilterFunction {
 			spanOpts = append(spanOpts, tracer.Tag(ext.EventSampleRate, cfg.analyticsRate))
 		}
 		spanOpts = append(spanOpts, httptrace.HeaderTagsFromRequest(req.Request, cfg.headerTags))
-		span, ctx := httptrace.StartRequestSpan(req.Request, spanOpts...)
+		_, ctx, finishSpans := httptrace.StartRequestSpan(req.Request, spanOpts...)
 		defer func() {
-			httptrace.FinishRequestSpan(span, resp.StatusCode(), nil, tracer.WithError(resp.Error()))
+			finishSpans(resp.StatusCode(), nil, tracer.WithError(resp.Error()))
 		}()
 
 		// pass the span through the request context

contrib/emicklei/go-restful/restful.go

@@ -46,9 +46,9 @@ func FilterFunc(configOpts ...Option) restful.FilterFunction {
 			spanOpts = append(spanOpts, tracer.Tag(ext.EventSampleRate, cfg.analyticsRate))
 		}
 		spanOpts = append(spanOpts, httptrace.HeaderTagsFromRequest(req.Request, cfg.headerTags))
-		span, ctx := httptrace.StartRequestSpan(req.Request, spanOpts...)
+		_, ctx, finishSpans := httptrace.StartRequestSpan(req.Request, spanOpts...)
 		defer func() {
-			httptrace.FinishRequestSpan(span, resp.StatusCode(), nil, tracer.WithError(resp.Error()))
+			finishSpans(resp.StatusCode(), nil, tracer.WithError(resp.Error()))
 		}()
 
 		// pass the span through the request context
@@ -59,9 +59,9 @@ func FilterFunc(configOpts ...Option) restful.FilterFunction {
 
 // Filter is deprecated. Please use FilterFunc.
 func Filter(req *restful.Request, resp *restful.Response, chain *restful.FilterChain) {
-	span, ctx := httptrace.StartRequestSpan(req.Request, tracer.ResourceName(req.SelectedRoutePath()))
+	_, ctx, finishSpans := httptrace.StartRequestSpan(req.Request, tracer.ResourceName(req.SelectedRoutePath()))
 	defer func() {
-		httptrace.FinishRequestSpan(span, resp.StatusCode(), nil, tracer.WithError(resp.Error()))
+		finishSpans(resp.StatusCode(), nil, tracer.WithError(resp.Error()))
 	}()
 
 	// pass the span through the request context

contrib/gin-gonic/gin/gintrace.go

@@ -52,9 +52,9 @@ func Middleware(service string, opts ...Option) gin.HandlerFunc {
 		}
 		opts = append(opts, tracer.Tag(ext.HTTPRoute, c.FullPath()))
 		opts = append(opts, httptrace.HeaderTagsFromRequest(c.Request, cfg.headerTags))
-		span, ctx := httptrace.StartRequestSpan(c.Request, opts...)
+		span, ctx, finishSpans := httptrace.StartRequestSpan(c.Request, opts...)
 		defer func() {
-			httptrace.FinishRequestSpan(span, c.Writer.Status(), nil)
+			finishSpans(c.Writer.Status(), nil)
 		}()
 
 		// pass the span through the request context

contrib/go-chi/chi.v5/chi.go

@@ -51,11 +51,11 @@ func Middleware(opts ...Option) func(next http.Handler) http.Handler {
 				opts = append(opts, tracer.Tag(ext.EventSampleRate, cfg.analyticsRate))
 			}
 			opts = append(opts, httptrace.HeaderTagsFromRequest(r, cfg.headerTags))
-			span, ctx := httptrace.StartRequestSpan(r, opts...)
+			span, ctx, finishSpans := httptrace.StartRequestSpan(r, opts...)
 			ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor)
 			defer func() {
 				status := ww.Status()
-				httptrace.FinishRequestSpan(span, status, cfg.isStatusError)
+				finishSpans(status, cfg.isStatusError)
 			}()
 
 			// pass the span through the request context

contrib/go-chi/chi/chi.go

@@ -51,11 +51,11 @@ func Middleware(opts ...Option) func(next http.Handler) http.Handler {
 				opts = append(opts, tracer.Tag(ext.EventSampleRate, cfg.analyticsRate))
 			}
 			opts = append(opts, httptrace.HeaderTagsFromRequest(r, cfg.headerTags))
-			span, ctx := httptrace.StartRequestSpan(r, opts...)
+			span, ctx, finishSpans := httptrace.StartRequestSpan(r, opts...)
 			ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor)
 			defer func() {
 				status := ww.Status()
-				httptrace.FinishRequestSpan(span, status, cfg.isStatusError)
+				finishSpans(status, cfg.isStatusError)
 			}()
 
 			// pass the span through the request context

contrib/go-redis/redis.v7/redis.go

@@ -79,7 +79,7 @@ func additionalTagOptions(client redis.UniversalClient) []ddtrace.StartSpanOptio
 		opt := clientOptions.Options()
 		if opt.Addr == "FailoverClient" {
 			additionalTags = []ddtrace.StartSpanOption{
-				tracer.Tag("out.db", strconv.Itoa(opt.DB)),
+				tracer.Tag(ext.TargetDB, strconv.Itoa(opt.DB)),
 				tracer.Tag(ext.RedisDatabaseIndex, opt.DB),
 			}
 		} else {
@@ -91,7 +91,7 @@ func additionalTagOptions(client redis.UniversalClient) []ddtrace.StartSpanOptio
 			additionalTags = []ddtrace.StartSpanOption{
 				tracer.Tag(ext.TargetHost, host),
 				tracer.Tag(ext.TargetPort, port),
-				tracer.Tag("out.db", strconv.Itoa(opt.DB)),
+				tracer.Tag(ext.TargetDB, strconv.Itoa(opt.DB)),
 				tracer.Tag(ext.RedisDatabaseIndex, opt.DB),
 			}
 		}

contrib/go-redis/redis.v8/redis.go

@@ -78,7 +78,7 @@ func additionalTagOptions(client redis.UniversalClient) []ddtrace.StartSpanOptio
 		opt := clientOptions.Options()
 		if opt.Addr == "FailoverClient" {
 			additionalTags = []ddtrace.StartSpanOption{
-				tracer.Tag("out.db", strconv.Itoa(opt.DB)),
+				tracer.Tag(ext.TargetDB, strconv.Itoa(opt.DB)),
 				tracer.Tag(ext.RedisDatabaseIndex, opt.DB),
 			}
 		} else {
@@ -90,7 +90,7 @@ func additionalTagOptions(client redis.UniversalClient) []ddtrace.StartSpanOptio
 			additionalTags = []ddtrace.StartSpanOption{
 				tracer.Tag(ext.TargetHost, host),
 				tracer.Tag(ext.TargetPort, port),
-				tracer.Tag("out.db", strconv.Itoa(opt.DB)),
+				tracer.Tag(ext.TargetDB, strconv.Itoa(opt.DB)),
 				tracer.Tag(ext.RedisDatabaseIndex, opt.DB),
 			}
 		}

contrib/go-redis/redis/redis.go

@@ -130,7 +130,7 @@ func (c *Pipeliner) execWithContext(ctx context.Context) ([]redis.Cmder, error)
 		tracer.ResourceName("redis"),
 		tracer.Tag(ext.TargetHost, p.host),
 		tracer.Tag(ext.TargetPort, p.port),
-		tracer.Tag("out.db", strconv.Itoa(p.db)),
+		tracer.Tag(ext.TargetDB, strconv.Itoa(p.db)),
 		tracer.Tag(ext.Component, componentName),
 		tracer.Tag(ext.SpanKind, ext.SpanKindClient),
 		tracer.Tag(ext.DBSystem, ext.DBSystemRedis),
@@ -202,7 +202,7 @@ func createWrapperFromClient(tc *Client) func(oldProcess func(cmd redis.Cmder) e
 				tracer.ResourceName(parts[0]),
 				tracer.Tag(ext.TargetHost, p.host),
 				tracer.Tag(ext.TargetPort, p.port),
-				tracer.Tag("out.db", strconv.Itoa(p.db)),
+				tracer.Tag(ext.TargetDB, strconv.Itoa(p.db)),
 				tracer.Tag("redis.raw_command", raw),
 				tracer.Tag("redis.args_length", strconv.Itoa(length)),
 				tracer.Tag(ext.Component, componentName),

contrib/gorm.io/gorm.v1/orchestrion.yml

@@ -13,4 +13,18 @@ aspects:
     join-point:
       function-call: gorm.io/gorm.Open
     advice:
-      - replace-function: gopkg.in/DataDog/dd-trace-go.v1/contrib/gorm.io/gorm.v1.Open
+      - wrap-expression:
+          imports:
+            gorm: gorm.io/gorm
+            gormtrace: gopkg.in/DataDog/dd-trace-go.v1/contrib/gorm.io/gorm.v1
+          template: |-
+            func() (*gorm.DB, error) {
+              db, err := {{ . }}
+              if err != nil {
+                return nil, err
+              }
+              if err := db.Use(gormtrace.NewTracePlugin()); err != nil {
+                return nil, err
+              }
+              return db, nil
+            }()

contrib/internal/httptrace/before_handle.go

@@ -57,11 +57,11 @@ func BeforeHandle(cfg *ServeConfig, w http.ResponseWriter, r *http.Request) (htt
 	if cfg.Route != "" {
 		opts = append(opts, tracer.Tag(ext.HTTPRoute, cfg.Route))
 	}
-	span, ctx := StartRequestSpan(r, opts...)
+	span, ctx, finishSpans := StartRequestSpan(r, opts...)
 	rw, ddrw := wrapResponseWriter(w)
 	rt := r.WithContext(ctx)
 	closeSpan := func() {
-		FinishRequestSpan(span, ddrw.status, cfg.IsStatusError, cfg.FinishOpts...)
+		finishSpans(ddrw.status, cfg.IsStatusError, cfg.FinishOpts...)
 	}
 	afterHandle := closeSpan
 	handled := false