contrib/database/sql: Submit DBStats as Datadog metrics

contrib/database/sql/example_test.go

@@ -106,3 +106,19 @@ func Example_dbmPropagation() {
 	}
 	defer rows.Close()
 }
+
+func Example_dbStats() {
+	sqltrace.Register("postgres", &pq.Driver{}, sqltrace.WithDBStats())
+	db, err := sqltrace.Open("postgres", "postgres://pqgotest:password@localhost/pqgotest?sslmode=disable")
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// Tracing is now enabled. Continue to use the database/sql package as usual
+	rows, err := db.Query("SELECT name FROM users WHERE age=?", 27)
+	if err != nil {
+		log.Fatal(err)
+	}
+	rows.Close()
+}

contrib/database/sql/metrics.go

@@ -0,0 +1,54 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package sql // import "gopkg.in/DataDog/dd-trace-go.v1/contrib/database/sql"
+
+import (
+	"database/sql"
+	"time"
+
+	"gopkg.in/DataDog/dd-trace-go.v1/internal"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/globalconfig"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
+)
+
+const tracerPrefix = "datadog.tracer."
+
+// ref: https://pkg.go.dev/database/sql#DBStats
+const (
+	MaxOpenConnections = tracerPrefix + "sql.db.connections.max_open"
+	OpenConnections    = tracerPrefix + "sql.db.connections.open"
+	InUse              = tracerPrefix + "sql.db.connections.in_use"
+	Idle               = tracerPrefix + "sql.db.connections.idle"
+	WaitCount          = tracerPrefix + "sql.db.connections.waiting"
+	WaitDuration       = tracerPrefix + "sql.db.connections.wait_duration"
+	MaxIdleClosed      = tracerPrefix + "sql.db.connections.closed.max_idle_conns"
+	MaxIdleTimeClosed  = tracerPrefix + "sql.db.connections.closed.max_idle_time"
+	MaxLifetimeClosed  = tracerPrefix + "sql.db.connections.closed.max_lifetime"
+)
+
+const interval = 10 * time.Second
+
+// pollDBStats calls (*DB).Stats on the db at a predetermined interval. It pushes the DBStats off to the StatsCarrier which ultimately sends them through a statsd client.
+// TODO: Perhaps grant a way for pollDBStats to grab the drivername so that it doesn't have to be passed in as a param
+func pollDBStats(db *sql.DB, tags []string) {
+	if db == nil {
+		log.Debug("No traced DB connection found; cannot pull DB stats.")
+		return
+	}
+	log.Debug("Traced DB connection found: DB stats will be gathered and sent every %v.", interval)
+	for range time.NewTicker(interval).C {
+		stat := db.Stats()
+		globalconfig.PushStat(internal.NewGauge(MaxOpenConnections, float64(stat.MaxOpenConnections), tags, 1))
+		globalconfig.PushStat(internal.NewGauge(OpenConnections, float64(stat.OpenConnections), tags, 1))
+		globalconfig.PushStat(internal.NewGauge(InUse, float64(stat.InUse), tags, 1))
+		globalconfig.PushStat(internal.NewGauge(Idle, float64(stat.Idle), tags, 1))
+		globalconfig.PushStat(internal.NewGauge(WaitCount, float64(stat.WaitCount), tags, 1))
+		globalconfig.PushStat(internal.NewTiming(WaitDuration, stat.WaitDuration, tags, 1))
+		globalconfig.PushStat(internal.NewCount(MaxIdleClosed, int64(stat.MaxIdleClosed), tags, 1))
+		globalconfig.PushStat(internal.NewCount(MaxIdleTimeClosed, int64(stat.MaxIdleTimeClosed), tags, 1))
+		globalconfig.PushStat(internal.NewCount(MaxLifetimeClosed, int64(stat.MaxLifetimeClosed), tags, 1))
+	}
+}

contrib/database/sql/option.go

@@ -29,6 +29,7 @@ type config struct {
 	errCheck           func(err error) bool
 	tags               map[string]interface{}
 	dbmPropagationMode tracer.DBMPropagationMode
+	dbStats            bool
 }
 
 func (c *config) checkDBMPropagation(driverName string, driver driver.Driver, dsn string) {
@@ -140,6 +141,7 @@ func defaults(cfg *config, driverName string, rc *registerConfig) {
 		cfg.errCheck = rc.errCheck
 		cfg.ignoreQueryTypes = rc.ignoreQueryTypes
 		cfg.childSpansOnly = rc.childSpansOnly
+		cfg.dbStats = rc.dbStats
 	}
 }
 
@@ -262,3 +264,12 @@ func WithDBMPropagation(mode tracer.DBMPropagationMode) Option {
 		cfg.dbmPropagationMode = mode
 	}
 }
+
+// WithDBStats enables polling of DBStats metrics
+// ref: https://pkg.go.dev/database/sql#DBStats
+// These metrics are submitted to Datadog and are not billed as custom metrics
+func WithDBStats() Option {
+	return func(cfg *config) {
+		cfg.dbStats = true
+	}
+}

contrib/database/sql/option_test.go

@@ -43,3 +43,17 @@ func TestAnalyticsSettings(t *testing.T) {
 		assert.Equal(t, 0.2, cfg.analyticsRate)
 	})
 }
+
+func TestWithDBStats(t *testing.T) {
+	t.Run("default off", func(t *testing.T) {
+		cfg := new(config)
+		defaults(cfg, "", nil)
+		assert.False(t, cfg.dbStats)
+	})
+	t.Run("on", func(t *testing.T) {
+		cfg := new(config)
+		defaults(cfg, "", nil)
+		WithDBStats()(cfg)
+		assert.True(t, cfg.dbStats)
+	})
+}

contrib/database/sql/sql.go

@@ -19,11 +19,12 @@ import (
 	"context"
 	"database/sql"
 	"database/sql/driver"
+	"fmt"
 	"reflect"
 	"sync"
 	"time"
 
-	"gopkg.in/DataDog/dd-trace-go.v1/contrib/database/sql/internal"
+	sqlinternal "gopkg.in/DataDog/dd-trace-go.v1/contrib/database/sql/internal"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/telemetry"
@@ -154,7 +155,7 @@ func (t *tracedConnector) Connect(ctx context.Context) (driver.Conn, error) {
 		cfg:        t.cfg,
 	}
 	if dsn != "" {
-		tp.meta, _ = internal.ParseDSN(t.driverName, dsn)
+		tp.meta, _ = sqlinternal.ParseDSN(t.driverName, dsn)
 	}
 	start := time.Now()
 	ctx, end := startTraceTask(ctx, string(QueryTypeConnect))
@@ -209,7 +210,11 @@ func OpenDB(c driver.Connector, opts ...Option) *sql.DB {
 		driverName: driverName,
 		cfg:        cfg,
 	}
-	return sql.OpenDB(tc)
+	db := sql.OpenDB(tc)
+	if cfg.dbStats {
+		go pollDBStats(db, []string{fmt.Sprintf("drivername:%v", driverName)})
+	}
+	return db
 }
 
 // Open returns connection to a DB using the traced version of the given driver. The driver may

contrib/database/sql/sql_test.go

@@ -21,6 +21,9 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/contrib/internal/sqltest"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/mocktracer"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/globalconfig"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/statsdtest"
 
 	mssql "github.com/denisenkom/go-mssqldb"
 	"github.com/go-sql-driver/mysql"
@@ -268,6 +271,56 @@ func TestOpenOptions(t *testing.T) {
 		s0 := spans[0]
 		assert.Equal(t, "register-override", s0.Tag(ext.ServiceName))
 	})
+
+	t.Run("WithDBStats", func(t *testing.T) {
+		Register(driverName, &pq.Driver{})
+		defer unregister(driverName)
+		_, err := Open(driverName, dsn, WithDBStats())
+		require.NoError(t, err)
+
+		var tg statsdtest.TestStatsdClient
+		sc := internal.NewStatsCarrier(&tg)
+		sc.Start()
+		defer sc.Stop()
+		globalconfig.SetStatsCarrier(sc)
+
+		// check that within 13s, at least one round of `pollDBStats` has completed
+		deadline := time.Now().Add(13 * time.Second)
+		for {
+			if time.Now().After(deadline) {
+				t.Fatalf("Stats not collected in expected interval of %v", interval)
+			}
+			calls := tg.CallNames()
+			if len(calls) < 9 {
+				time.Sleep(100 * time.Millisecond)
+				continue
+			}
+			if containsAllStats(calls) {
+				break
+			}
+			t.Fatalf("Some stats missing")
+		}
+	})
+}
+
+func containsAllStats(calls []string) bool {
+	wantStats := []string{MaxOpenConnections, OpenConnections, InUse, Idle, WaitCount, WaitDuration, MaxIdleClosed, MaxIdleTimeClosed, MaxLifetimeClosed}
+	for _, s := range wantStats {
+		if !contains(calls, s) {
+			return false
+		}
+	}
+	return true
+}
+
+func contains(s []string, str string) bool {
+	for _, v := range s {
+		if v == str {
+			return true
+		}
+	}
+
+	return false
 }
 
 func TestMySQLUint64(t *testing.T) {

ddtrace/tracer/option.go

@@ -260,6 +260,8 @@ type config struct {
 
 	// headerAsTags holds the header as tags configuration.
 	headerAsTags dynamicConfig[[]string]
+
+	contribStats bool
 }
 
 // orchestrionConfig contains Orchestrion configuration.
@@ -340,6 +342,7 @@ func newConfig(opts ...StartOption) *config {
 		c.logToStdout = true
 	}
 	c.logStartup = internal.BoolEnv("DD_TRACE_STARTUP_LOGS", true)
+	c.contribStats = internal.BoolEnv("DD_TRACE_CONTRIB_STATS_ENABLED", true)
 	c.runtimeMetrics = internal.BoolEnv("DD_RUNTIME_METRICS_ENABLED", false)
 	c.debug = internal.BoolEnv("DD_TRACE_DEBUG", false)
 	c.enabled = newDynamicConfig("tracing_enabled", internal.BoolEnv("DD_TRACE_ENABLED", true), func(b bool) bool { return true }, equal[bool])
@@ -1265,6 +1268,15 @@ func setHeaderTags(headerAsTags []string) bool {
 	return true
 }
 
+// WithContribStats opens up a channel of communication between tracer and contrib libraries
+// for submitting stats from contribs to Datadog via the tracer's statsd client
+// It is enabled by default but can be disabled with `WithContribStats(false)`
+func WithContribStats(enabled bool) StartOption {
+	return func(c *config) {
+		c.contribStats = enabled
+	}
+}
+
 // UserMonitoringConfig is used to configure what is used to identify a user.
 // This configuration can be set by combining one or several UserMonitoringOption with a call to SetUser().
 type UserMonitoringConfig struct {

ddtrace/tracer/option_test.go

@@ -1235,6 +1235,27 @@ func TestWithHeaderTags(t *testing.T) {
 	assert.Equal(t, 0, globalconfig.HeaderTagsLen())
 }
 
+func TestContribStatsEnabled(t *testing.T) {
+	t.Run("Default", func(t *testing.T) {
+		c := newConfig()
+		assert.True(t, c.contribStats)
+	})
+	t.Run("Disable", func(t *testing.T) {
+		c := newConfig(WithContribStats(false))
+		assert.False(t, c.contribStats)
+	})
+	t.Run("Disable with ENV", func(t *testing.T) {
+		t.Setenv("DD_TRACE_CONTRIB_STATS_ENABLED", "false")
+		c := newConfig()
+		assert.False(t, c.contribStats)
+	})
+	t.Run("Env override", func(t *testing.T) {
+		t.Setenv("DD_TRACE_CONTRIB_STATS_ENABLED", "false")
+		c := newConfig(WithContribStats(true))
+		assert.True(t, c.contribStats)
+	})
+}
+
 func TestHostnameDisabled(t *testing.T) {
 	t.Run("DisabledWithUDS", func(t *testing.T) {
 		t.Setenv("DD_TRACE_AGENT_URL", "unix://somefakesocket")

ddtrace/tracer/tracer.go

@@ -23,6 +23,7 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/appsec"
 	appsecConfig "gopkg.in/DataDog/dd-trace-go.v1/internal/appsec/config"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/datastreams"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal/globalconfig"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/hostname"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/log"
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/remoteconfig"
@@ -105,6 +106,8 @@ type tracer struct {
 	// abandonedSpansDebugger specifies where and how potentially abandoned spans are stored
 	// when abandoned spans debugging is enabled.
 	abandonedSpansDebugger *abandonedSpansDebugger
+
+	statsCarrier *globalinternal.StatsCarrier
 }
 
 const (
@@ -259,6 +262,10 @@ func newUnstartedTracer(opts ...StartOption) *tracer {
 			return f.DataStreams
 		})
 	}
+	var statsCarrier *globalinternal.StatsCarrier
+	if c.contribStats {
+		statsCarrier = globalinternal.NewStatsCarrier(statsd)
+	}
 	t := &tracer{
 		config:           c,
 		traceWriter:      writer,
@@ -278,8 +285,9 @@ func newUnstartedTracer(opts ...StartOption) *tracer {
 				Cache:            c.agent.HasFlag("sql_cache"),
 			},
 		}),
-		statsd:      statsd,
-		dataStreams: dataStreamsProcessor,
+		statsd:       statsd,
+		dataStreams:  dataStreamsProcessor,
+		statsCarrier: statsCarrier,
 	}
 	return t
 }
@@ -323,6 +331,10 @@ func newTracer(opts ...StartOption) *tracer {
 		t.reportHealthMetrics(statsInterval)
 	}()
 	t.stats.Start()
+	if sc := t.statsCarrier; sc != nil {
+		sc.Start()
+		globalconfig.SetStatsCarrier(sc)
+	}
 	return t
 }
 
@@ -658,6 +670,9 @@ func (t *tracer) Stop() {
 	if t.dataStreams != nil {
 		t.dataStreams.Stop()
 	}
+	if t.statsCarrier != nil {
+		t.statsCarrier.Stop()
+	}
 	appsec.Stop()
 	remoteconfig.Stop()
 }

ddtrace/tracer/tracer_test.go

@@ -674,6 +674,37 @@ func TestTracerRuntimeMetrics(t *testing.T) {
 	})
 }
 
+func TestTracerContribStats(t *testing.T) {
+	t.Run("default on", func(t *testing.T) {
+		tp := new(log.RecordLogger)
+		tracer := newTracer(WithDebugMode(true), WithLogger(tp))
+		defer tracer.Stop()
+		assert.NotNil(t, tracer.statsCarrier)
+	})
+	t.Run("off", func(t *testing.T) {
+		tp := new(log.RecordLogger)
+		tracer := newTracer(WithContribStats(false), WithLogger(tp), WithDebugMode(true))
+		defer tracer.Stop()
+		assert.Nil(t, tracer.statsCarrier)
+	})
+	t.Run("env", func(t *testing.T) {
+		os.Setenv("DD_TRACE_CONTRIB_STATS_ENABLED", "false")
+		defer os.Unsetenv("DD_TRACE_CONTRIB_STATS_ENABLED")
+		tp := new(log.RecordLogger)
+		tracer := newTracer(WithLogger(tp), WithDebugMode(true))
+		defer tracer.Stop()
+		assert.Nil(t, tracer.statsCarrier)
+	})
+	t.Run("env override", func(t *testing.T) {
+		os.Setenv("DD_TRACE_CONTRIB_STATS_ENABLED", "false")
+		defer os.Unsetenv("DD_TRACE_CONTRIB_STATS_ENABLED")
+		tp := new(log.RecordLogger)
+		tracer := newTracer(WithLogger(tp), WithDebugMode(true), WithContribStats(true))
+		defer tracer.Stop()
+		assert.NotNil(t, tracer.statsCarrier)
+	})
+}
+
 func TestTracerStartSpanOptions(t *testing.T) {
 	tracer := newTracer()
 	defer tracer.Stop()