adds stats to ocsql

Author: basvanbeek

dbstats_go1.11.go

@@ -0,0 +1,46 @@
+// +build go1.11
+
+package ocsql
+
+import (
+	"context"
+	"database/sql"
+	"time"
+
+	"go.opencensus.io/stats"
+)
+
+// RecordStats records database statistics for provided sql.DB at the provided
+// interval.
+func RecordStats(db *sql.DB, interval time.Duration) (fnStop func()) {
+	var (
+		ctx    = context.Background()
+		ticker = time.NewTicker(interval)
+		done   = make(chan struct{})
+	)
+
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				dbStats := db.Stats()
+				stats.Record(ctx,
+					MeasureOpenConnections.M(int64(dbStats.OpenConnections)),
+					MeasureIdleConnections.M(int64(dbStats.Idle)),
+					MeasureActiveConnections.M(int64(dbStats.InUse)),
+					MeasureWaitCount.M(dbStats.WaitCount),
+					MeasureWaitDuration.M(float64(dbStats.WaitDuration.Nanoseconds())/1e6),
+					MeasureIdleClosed.M(dbStats.MaxIdleClosed),
+					MeasureLifetimeClosed.M(dbStats.MaxLifetimeClosed),
+				)
+			case <-done:
+				ticker.Stop()
+				return
+			}
+		}
+	}()
+
+	return func() {
+		close(done)
+	}
+}

driver.go

@@ -123,6 +123,8 @@ type ocConn struct {
 }
 
 func (c ocConn) Ping(ctx context.Context) (err error) {
+	defer recordCallStats(ctx, "go.sql.ping")(err)
+
 	if c.options.Ping && (c.options.AllowRoot || trace.FromContext(ctx) != nil) {
 		var span *trace.Span
 		ctx, span = trace.StartSpan(ctx, "sql:ping", trace.WithSpanKind(trace.SpanKindClient))
@@ -149,6 +151,8 @@ func (c ocConn) Ping(ctx context.Context) (err error) {
 }
 
 func (c ocConn) Exec(query string, args []driver.Value) (res driver.Result, err error) {
+	defer recordCallStats(context.Background(), "go.sql.exec")(err)
+
 	if exec, ok := c.parent.(driver.Execer); ok {
 		if !c.options.AllowRoot {
 			return exec.Exec(query, args)
@@ -188,6 +192,8 @@ func (c ocConn) Exec(query string, args []driver.Value) (res driver.Result, err
 }
 
 func (c ocConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (res driver.Result, err error) {
+	defer recordCallStats(ctx, "go.sql.exec")(err)
+
 	if execCtx, ok := c.parent.(driver.ExecerContext); ok {
 		parentSpan := trace.FromContext(ctx)
 		if !c.options.AllowRoot && parentSpan == nil {
@@ -225,6 +231,8 @@ func (c ocConn) ExecContext(ctx context.Context, query string, args []driver.Nam
 }
 
 func (c ocConn) Query(query string, args []driver.Value) (rows driver.Rows, err error) {
+	defer recordCallStats(context.Background(), "go.sql.query")(err)
+
 	if queryer, ok := c.parent.(driver.Queryer); ok {
 		if !c.options.AllowRoot {
 			return queryer.Query(query, args)
@@ -265,6 +273,8 @@ func (c ocConn) Query(query string, args []driver.Value) (rows driver.Rows, err
 }
 
 func (c ocConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (rows driver.Rows, err error) {
+	defer recordCallStats(ctx, "go.sql.query")(err)
+
 	if queryerCtx, ok := c.parent.(driver.QueryerContext); ok {
 		parentSpan := trace.FromContext(ctx)
 		if !c.options.AllowRoot && parentSpan == nil {
@@ -303,6 +313,8 @@ func (c ocConn) QueryContext(ctx context.Context, query string, args []driver.Na
 }
 
 func (c ocConn) Prepare(query string) (stmt driver.Stmt, err error) {
+	defer recordCallStats(context.Background(), "go.sql.prepare")(err)
+
 	if c.options.AllowRoot {
 		_, span := trace.StartSpan(context.Background(), "sql:prepare", trace.WithSpanKind(trace.SpanKindClient))
 		attrs := make([]trace.Attribute, 0, len(c.options.DefaultAttributes)+1)
@@ -337,6 +349,8 @@ func (c *ocConn) Begin() (driver.Tx, error) {
 }
 
 func (c *ocConn) PrepareContext(ctx context.Context, query string) (stmt driver.Stmt, err error) {
+	defer recordCallStats(ctx, "go.sql.prepare")(err)
+
 	var span *trace.Span
 	attrs := append([]trace.Attribute(nil), c.options.DefaultAttributes...)
 	if c.options.AllowRoot || trace.FromContext(ctx) != nil {
@@ -367,7 +381,9 @@ func (c *ocConn) PrepareContext(ctx context.Context, query string) (stmt driver.
 	return
 }
 
-func (c *ocConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
+func (c *ocConn) BeginTx(ctx context.Context, opts driver.TxOptions) (tx driver.Tx, err error) {
+	defer recordCallStats(ctx, "go.sql.begin")(err)
+
 	if !c.options.AllowRoot && trace.FromContext(ctx) == nil {
 		if connBeginTx, ok := c.parent.(driver.ConnBeginTx); ok {
 			return connBeginTx.BeginTx(ctx, opts)
@@ -392,7 +408,7 @@ func (c *ocConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx,
 	defer span.End()
 
 	if connBeginTx, ok := c.parent.(driver.ConnBeginTx); ok {
-		tx, err := connBeginTx.BeginTx(ctx, opts)
+		tx, err = connBeginTx.BeginTx(ctx, opts)
 		setSpanStatus(span, err)
 		if err != nil {
 			return nil, err
@@ -407,7 +423,7 @@ func (c *ocConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx,
 			"ocsql.deprecated", "driver does not support ConnBeginTx",
 		),
 	)
-	tx, err := c.parent.Begin()
+	tx, err = c.parent.Begin()
 	setSpanStatus(span, err)
 	if err != nil {
 		return nil, err
@@ -462,6 +478,8 @@ type ocStmt struct {
 }
 
 func (s ocStmt) Exec(args []driver.Value) (res driver.Result, err error) {
+	defer recordCallStats(context.Background(), "go.sql.stmt.exec")(err)
+
 	if !s.options.AllowRoot {
 		return s.parent.Exec(args)
 	}
@@ -507,6 +525,8 @@ func (s ocStmt) NumInput() int {
 }
 
 func (s ocStmt) Query(args []driver.Value) (rows driver.Rows, err error) {
+	defer recordCallStats(context.Background(), "go.sql.stmt.query")(err)
+
 	if !s.options.AllowRoot {
 		return s.parent.Query(args)
 	}
@@ -543,6 +563,8 @@ func (s ocStmt) Query(args []driver.Value) (rows driver.Rows, err error) {
 }
 
 func (s ocStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (res driver.Result, err error) {
+	defer recordCallStats(ctx, "go.sql.stmt.exec")(err)
+
 	parentSpan := trace.FromContext(ctx)
 	if !s.options.AllowRoot && parentSpan == nil {
 		// we already tested driver to implement StmtExecContext
@@ -580,6 +602,8 @@ func (s ocStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (res
 }
 
 func (s ocStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (rows driver.Rows, err error) {
+	defer recordCallStats(ctx, "go.sql.stmt.query")(err)
+
 	parentSpan := trace.FromContext(ctx)
 	if !s.options.AllowRoot && parentSpan == nil {
 		// we already tested driver to implement StmtQueryContext
@@ -775,6 +799,8 @@ type ocTx struct {
 }
 
 func (t ocTx) Commit() (err error) {
+	defer recordCallStats(context.Background(), "go.sql.commit")(err)
+
 	_, span := trace.StartSpan(t.ctx, "sql:commit", trace.WithSpanKind(trace.SpanKindClient))
 	if len(t.options.DefaultAttributes) > 0 {
 		span.AddAttributes(t.options.DefaultAttributes...)
@@ -789,6 +815,8 @@ func (t ocTx) Commit() (err error) {
 }
 
 func (t ocTx) Rollback() (err error) {
+	defer recordCallStats(context.Background(), "go.sql.rollback")(err)
+
 	_, span := trace.StartSpan(t.ctx, "sql:rollback", trace.WithSpanKind(trace.SpanKindClient))
 	if len(t.options.DefaultAttributes) > 0 {
 		span.AddAttributes(t.options.DefaultAttributes...)

observability.go

@@ -0,0 +1,176 @@
+package ocsql
+
+import (
+	"context"
+	"time"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+)
+
+// The following tags are applied to stats recorded by this package.
+var (
+	// GoSqlMethod is the SQL method called.
+	GoSqlMethod, _ = tag.NewKey("go_sql_method")
+	// GoSqlError is the error received while calling a SQL method.
+	GoSqlError, _ = tag.NewKey("go_sql_error")
+	// GoSqlStatus identifies success vs. error from the SQL method response.
+	GoSqlStatus, _ = tag.NewKey("go_sql_status")
+
+	valueOK  = tag.Insert(GoSqlStatus, "OK")
+	valueErr = tag.Insert(GoSqlStatus, "ERROR")
+)
+
+// The following measures are supported for use in custom views.
+var (
+	MeasureLatencyMs         = stats.Float64("go.sql/latency", "The latency of calls in milliseconds", stats.UnitMilliseconds)
+	MeasureOpenConnections   = stats.Int64("go.sql/connections/open", "Count of open connections in the pool", stats.UnitDimensionless)
+	MeasureIdleConnections   = stats.Int64("go.sql/connections/idle", "Count of idle connections in the pool", stats.UnitDimensionless)
+	MeasureActiveConnections = stats.Int64("go.sql/connections/active", "Count of active connections in the pool", stats.UnitDimensionless)
+	MeasureWaitCount         = stats.Int64("go.sql/connections/wait_count", "The total number of connections waited for", stats.UnitDimensionless)
+	MeasureWaitDuration      = stats.Float64("go.sql/connections/wait_duration", "The total time blocked waiting for a new connection", stats.UnitMilliseconds)
+	MeasureIdleClosed        = stats.Int64("go.sql/connections/idle_closed", "The total number of connections closed due to SetMaxIdleConns", stats.UnitDimensionless)
+	MeasureLifetimeClosed    = stats.Int64("go.sql/connections/lifetime_closed", "The total number of connections closed due to SetConnMaxLifetime", stats.UnitDimensionless)
+)
+
+// Default distributions used by views in this package
+var (
+	DefaultMillisecondsDistribution = view.Distribution(
+		0.0,
+		0.001,
+		0.005,
+		0.01,
+		0.05,
+		0.1,
+		0.5,
+		1.0,
+		1.5,
+		2.0,
+		2.5,
+		5.0,
+		10.0,
+		25.0,
+		50.0,
+		100.0,
+		200.0,
+		400.0,
+		600.0,
+		800.0,
+		1000.0,
+		1500.0,
+		2000.0,
+		2500.0,
+		5000.0,
+		10000.0,
+		20000.0,
+		40000.0,
+		100000.0,
+		200000.0,
+		500000.0)
+)
+
+// Package ocsql provides some convenience views.
+// You still need to register these views for data to actually be collected.
+// You can use the RegisterAllViews function for this.
+var (
+	SqlClientLatencyView = &view.View{
+		Name:        "go.sql/client/latency",
+		Description: "The distribution of latencies of various calls in milliseconds",
+		Measure:     MeasureLatencyMs,
+		Aggregation: DefaultMillisecondsDistribution,
+		TagKeys:     []tag.Key{GoSqlMethod, GoSqlError, GoSqlStatus},
+	}
+
+	SqlClientCallsView = &view.View{
+		Name:        "go.sql/client/calls",
+		Description: "The number of various calls of methods",
+		Measure:     MeasureLatencyMs,
+		Aggregation: view.Count(),
+		TagKeys:     []tag.Key{GoSqlMethod, GoSqlError, GoSqlStatus},
+	}
+
+	SqlClientOpenConnectionsView = &view.View{
+		Name:        "go.sql/db/connections/open",
+		Description: "The number of open connections",
+		Measure:     MeasureOpenConnections,
+		Aggregation: view.LastValue(),
+	}
+
+	SqlClientIdleConnectionsView = &view.View{
+		Name:        "go.sql/db/connections/idle",
+		Description: "The number of idle connections",
+		Measure:     MeasureIdleConnections,
+		Aggregation: view.LastValue(),
+	}
+
+	SqlClientActiveConnectionsView = &view.View{
+		Name:        "go.sql/db/connections/active",
+		Description: "The number of active connections",
+		Measure:     MeasureActiveConnections,
+		Aggregation: view.LastValue(),
+	}
+
+	SqlClientWaitCountView = &view.View{
+		Name:        "go.sql/db/connections/wait_count",
+		Description: "The total number of connections waited for",
+		Measure:     MeasureWaitCount,
+		Aggregation: view.LastValue(),
+	}
+
+	SqlClientWaitDurationView = &view.View{
+		Name:        "go.sql/db/connections/wait_duration",
+		Description: "The total time blocked waiting for a new connection",
+		Measure:     MeasureWaitDuration,
+		Aggregation: view.LastValue(),
+	}
+
+	SqlClientIdleClosedView = &view.View{
+		Name:        "go.sql/db/connections/idle_closed_count",
+		Description: "The total number of connections closed due to SetMaxIdleConns",
+		Measure:     MeasureIdleClosed,
+		Aggregation: view.LastValue(),
+	}
+
+	SqlClientLifetimeClosedView = &view.View{
+		Name:        "go.sql/db/connections/lifetime_closed_count",
+		Description: "The total number of connections closed due to SetConnMaxLifetime",
+		Measure:     MeasureLifetimeClosed,
+		Aggregation: view.LastValue(),
+	}
+
+	DefaultViews = []*view.View{
+		SqlClientLatencyView, SqlClientCallsView, SqlClientOpenConnectionsView,
+		SqlClientIdleConnectionsView, SqlClientActiveConnectionsView,
+		SqlClientWaitCountView, SqlClientWaitDurationView,
+		SqlClientIdleClosedView, SqlClientLifetimeClosedView,
+	}
+)
+
+// RegisterAllViews registers all ocsql views to enable collection of stats.
+func RegisterAllViews() {
+	if err := view.Register(DefaultViews...); err != nil {
+		panic(err)
+	}
+}
+
+func recordCallStats(ctx context.Context, method string) func(err error) {
+	var tags []tag.Mutator
+	startTime := time.Now()
+
+	return func(err error) {
+		timeSpent := float64(time.Since(startTime).Nanoseconds()) / 1e6
+
+		if err != nil {
+			tags = []tag.Mutator{
+				tag.Insert(GoSqlMethod, method), valueErr, tag.Insert(GoSqlError, err.Error()),
+			}
+		} else {
+			tags = []tag.Mutator{
+				tag.Insert(GoSqlMethod, method), valueOK,
+			}
+		}
+
+		_ = stats.RecordWithTags(ctx, tags, MeasureLatencyMs.M(timeSpent))
+	}
+}