Merge branch 'main' into dario.castane/langplat-286/origin-detection

Author: darccio

.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/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/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),