chore: add tx metrics and logs for serialization errors (#15215)

Before db_metrics were all or nothing. Now `InTx` metrics are always recorded, and query metrics are opt in.


Adds instrumentation & logging around serialization failures in the database.

coderd/database/db.go

@@ -28,7 +28,7 @@ type Store interface {
 	wrapper
 
 	Ping(ctx context.Context) (time.Duration, error)
-	InTx(func(Store) error, *sql.TxOptions) error
+	InTx(func(Store) error, *TxOptions) error
 }
 
 type wrapper interface {
@@ -57,6 +57,43 @@ func New(sdb *sql.DB) Store {
 	}
 }
 
+// TxOptions is used to pass some execution metadata to the callers.
+// Ideally we could throw this into a context, but no context is used for
+// transactions. So instead, the return context is attached to the options
+// passed in.
+// This metadata should not be returned in the method signature, because it
+// is only used for metric tracking. It should never be used by business logic.
+type TxOptions struct {
+	// Isolation is the transaction isolation level.
+	// If zero, the driver or database's default level is used.
+	Isolation sql.IsolationLevel
+	ReadOnly  bool
+
+	// -- Coder specific metadata --
+	// TxIdentifier is a unique identifier for the transaction to be used
+	// in metrics. Can be any string.
+	TxIdentifier string
+
+	// Set by InTx
+	executionCount int
+}
+
+// IncrementExecutionCount is a helper function for external packages
+// to increment the unexported count.
+// Mainly for `dbmem`.
+func IncrementExecutionCount(opts *TxOptions) {
+	opts.executionCount++
+}
+
+func (o TxOptions) ExecutionCount() int {
+	return o.executionCount
+}
+
+func (o *TxOptions) WithID(id string) *TxOptions {
+	o.TxIdentifier = id
+	return o
+}
+
 // queries encompasses both are sqlc generated
 // queries and our custom queries.
 type querier interface {
@@ -80,11 +117,24 @@ func (q *sqlQuerier) Ping(ctx context.Context) (time.Duration, error) {
 	return time.Since(start), err
 }
 
-func (q *sqlQuerier) InTx(function func(Store) error, txOpts *sql.TxOptions) error {
+func DefaultTXOptions() *TxOptions {
+	return &TxOptions{
+		Isolation: sql.LevelDefault,
+		ReadOnly:  false,
+	}
+}
+
+func (q *sqlQuerier) InTx(function func(Store) error, txOpts *TxOptions) error {
 	_, inTx := q.db.(*sqlx.Tx)
-	isolation := sql.LevelDefault
-	if txOpts != nil {
-		isolation = txOpts.Isolation
+
+	if txOpts == nil {
+		// create a default txOpts if left to nil
+		txOpts = DefaultTXOptions()
+	}
+
+	sqlOpts := &sql.TxOptions{
+		Isolation: txOpts.Isolation,
+		ReadOnly:  txOpts.ReadOnly,
 	}
 
 	// If we are not already in a transaction, and we are running in serializable
@@ -92,13 +142,14 @@ func (q *sqlQuerier) InTx(function func(Store) error, txOpts *sql.TxOptions) err
 	// prepared to allow retries if using serializable mode.
 	// If we are in a transaction already, the parent InTx call will handle the retry.
 	// We do not want to duplicate those retries.
-	if !inTx && isolation == sql.LevelSerializable {
+	if !inTx && sqlOpts.Isolation == sql.LevelSerializable {
 		// This is an arbitrarily chosen number.
 		const retryAmount = 3
 		var err error
 		attempts := 0
 		for attempts = 0; attempts < retryAmount; attempts++ {
-			err = q.runTx(function, txOpts)
+			txOpts.executionCount++
+			err = q.runTx(function, sqlOpts)
 			if err == nil {
 				// Transaction succeeded.
 				return nil
@@ -111,7 +162,9 @@ func (q *sqlQuerier) InTx(function func(Store) error, txOpts *sql.TxOptions) err
 		// Transaction kept failing in serializable mode.
 		return xerrors.Errorf("transaction failed after %d attempts: %w", attempts, err)
 	}
-	return q.runTx(function, txOpts)
+
+	txOpts.executionCount++
+	return q.runTx(function, sqlOpts)
 }
 
 // InTx performs database operations inside a transaction.

coderd/database/db_test.go

@@ -27,7 +27,7 @@ func TestSerializedRetry(t *testing.T) {
 	db := database.New(sqlDB)
 
 	called := 0
-	txOpts := &sql.TxOptions{Isolation: sql.LevelSerializable}
+	txOpts := &database.TxOptions{Isolation: sql.LevelSerializable}
 	err := db.InTx(func(tx database.Store) error {
 		// Test nested error
 		return tx.InTx(func(tx database.Store) error {

coderd/database/dbauthz/dbauthz.go

@@ -558,7 +558,7 @@ func (q *querier) Ping(ctx context.Context) (time.Duration, error) {
 }
 
 // InTx runs the given function in a transaction.
-func (q *querier) InTx(function func(querier database.Store) error, txOpts *sql.TxOptions) error {
+func (q *querier) InTx(function func(querier database.Store) error, txOpts *database.TxOptions) error {
 	return q.db.InTx(func(tx database.Store) error {
 		// Wrap the transaction store in a querier.
 		wrapped := New(tx, q.auth, q.log, q.acs)

coderd/database/dbmem/dbmem.go

@@ -365,7 +365,7 @@ func (tx *fakeTx) releaseLocks() {
 }
 
 // InTx doesn't rollback data properly for in-memory yet.
-func (q *FakeQuerier) InTx(fn func(database.Store) error, _ *sql.TxOptions) error {
+func (q *FakeQuerier) InTx(fn func(database.Store) error, opts *database.TxOptions) error {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()
 	tx := &fakeTx{
@@ -374,6 +374,9 @@ func (q *FakeQuerier) InTx(fn func(database.Store) error, _ *sql.TxOptions) erro
 	}
 	defer tx.releaseLocks()
 
+	if opts != nil {
+		database.IncrementExecutionCount(opts)
+	}
 	return fn(tx)
 }
 

coderd/database/dbmetrics/dbmetrics_test.go

@@ -0,0 +1,109 @@
+package dbmetrics_test
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/xerrors"
+
+	"cdr.dev/slog"
+	"cdr.dev/slog/sloggers/sloghuman"
+	"cdr.dev/slog/sloggers/slogtest"
+	"github.com/coder/coder/v2/coderd/coderdtest/promhelp"
+	"github.com/coder/coder/v2/coderd/database"
+	"github.com/coder/coder/v2/coderd/database/dbmem"
+	"github.com/coder/coder/v2/coderd/database/dbmetrics"
+)
+
+func TestInTxMetrics(t *testing.T) {
+	t.Parallel()
+
+	successLabels := prometheus.Labels{
+		"success": "true",
+		"id":      "",
+	}
+	const inTxHistMetricName = "coderd_db_tx_duration_seconds"
+	const inTxCountMetricName = "coderd_db_tx_executions_count"
+	t.Run("QueryMetrics", func(t *testing.T) {
+		t.Parallel()
+
+		db := dbmem.New()
+		reg := prometheus.NewRegistry()
+		db = dbmetrics.NewQueryMetrics(db, slogtest.Make(t, nil), reg)
+
+		err := db.InTx(func(s database.Store) error {
+			return nil
+		}, nil)
+		require.NoError(t, err)
+
+		// Check that the metrics are registered
+		inTxMetric := promhelp.HistogramValue(t, reg, inTxHistMetricName, successLabels)
+		require.NotNil(t, inTxMetric)
+		require.Equal(t, uint64(1), inTxMetric.GetSampleCount())
+	})
+
+	t.Run("DBMetrics", func(t *testing.T) {
+		t.Parallel()
+
+		db := dbmem.New()
+		reg := prometheus.NewRegistry()
+		db = dbmetrics.NewDBMetrics(db, slogtest.Make(t, nil), reg)
+
+		err := db.InTx(func(s database.Store) error {
+			return nil
+		}, nil)
+		require.NoError(t, err)
+
+		// Check that the metrics are registered
+		inTxMetric := promhelp.HistogramValue(t, reg, inTxHistMetricName, successLabels)
+		require.NotNil(t, inTxMetric)
+		require.Equal(t, uint64(1), inTxMetric.GetSampleCount())
+	})
+
+	// Test log output and metrics on failures
+	// Log example:
+	//  [erro]  database transaction hit serialization error and had to retry  success=false  executions=2  id=foobar_factory
+	t.Run("SerializationError", func(t *testing.T) {
+		t.Parallel()
+
+		var output bytes.Buffer
+		logger := slog.Make(sloghuman.Sink(&output))
+
+		reg := prometheus.NewRegistry()
+		db := dbmetrics.NewDBMetrics(dbmem.New(), logger, reg)
+		const id = "foobar_factory"
+
+		txOpts := database.DefaultTXOptions().WithID(id)
+		database.IncrementExecutionCount(txOpts) // 2 executions
+
+		err := db.InTx(func(s database.Store) error {
+			return xerrors.Errorf("some dumb error")
+		}, txOpts)
+		require.Error(t, err)
+
+		// Check that the metrics are registered
+		inTxHistMetric := promhelp.HistogramValue(t, reg, inTxHistMetricName, prometheus.Labels{
+			"success": "false",
+			"id":      id,
+		})
+		require.NotNil(t, inTxHistMetric)
+		require.Equal(t, uint64(1), inTxHistMetric.GetSampleCount())
+
+		inTxCountMetric := promhelp.CounterValue(t, reg, inTxCountMetricName, prometheus.Labels{
+			"success": "false",
+			"retries": "1",
+			"id":      id,
+		})
+		require.NotNil(t, inTxCountMetric)
+		require.Equal(t, 1, inTxCountMetric)
+
+		// Also check the logs
+		require.Contains(t, output.String(), "some dumb error")
+		require.Contains(t, output.String(), "database transaction hit serialization error and had to retry")
+		require.Contains(t, output.String(), "success=false")
+		require.Contains(t, output.String(), "executions=2")
+		require.Contains(t, output.String(), "id="+id)
+	})
+}

coderd/database/dbmock/dbmock.go

@@ -11,7 +11,6 @@ package dbmock
 
 import (
 	context "context"
-	sql "database/sql"
 	reflect "reflect"
 	time "time"
 
@@ -3489,7 +3488,7 @@ func (mr *MockStoreMockRecorder) GetWorkspacesEligibleForTransition(arg0, arg1 a
 }
 
 // InTx mocks base method.
-func (m *MockStore) InTx(arg0 func(database.Store) error, arg1 *sql.TxOptions) error {
+func (m *MockStore) InTx(arg0 func(database.Store) error, arg1 *database.TxOptions) error {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "InTx", arg0, arg1)
 	ret0, _ := ret[0].(error)

coderd/database/dbpurge/dbpurge.go

@@ -66,7 +66,7 @@ func New(ctx context.Context, logger slog.Logger, db database.Store, clk quartz.
 			logger.Info(ctx, "purged old database entries", slog.F("duration", clk.Since(start)))
 
 			return nil
-		}, nil); err != nil {
+		}, database.DefaultTXOptions().WithID("db_purge")); err != nil {
 			logger.Error(ctx, "failed to purge old database entries", slog.Error(err))
 			return
 		}

coderd/database/dbrollup/dbrollup.go

@@ -108,7 +108,7 @@ func (r *Rolluper) start(ctx context.Context) {
 
 				ev.TemplateUsageStats = true
 				return tx.UpsertTemplateUsageStats(ctx)
-			}, nil)
+			}, database.DefaultTXOptions().WithID("db_rollup"))
 		})
 
 		err := eg.Wait()

coderd/database/dbrollup/dbrollup_test.go

@@ -38,7 +38,7 @@ type wrapUpsertDB struct {
 	resume <-chan struct{}
 }
 
-func (w *wrapUpsertDB) InTx(fn func(database.Store) error, opts *sql.TxOptions) error {
+func (w *wrapUpsertDB) InTx(fn func(database.Store) error, opts *database.TxOptions) error {
 	return w.Store.InTx(func(tx database.Store) error {
 		return fn(&wrapUpsertDB{Store: tx, resume: w.resume})
 	}, opts)

coderd/database/tx.go

@@ -33,7 +33,7 @@ func ReadModifyUpdate(db Store, f func(tx Store) error,
 ) error {
 	var err error
 	for retries := 0; retries < maxRetries; retries++ {
-		err = db.InTx(f, &sql.TxOptions{
+		err = db.InTx(f, &TxOptions{
 			Isolation: sql.LevelRepeatableRead,
 		})
 		var pqe *pq.Error

coderd/database/tx_test.go

@@ -19,7 +19,7 @@ func TestReadModifyUpdate_OK(t *testing.T) {
 	mDB := dbmock.NewMockStore(gomock.NewController(t))
 
 	mDB.EXPECT().
-		InTx(gomock.Any(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead}).
+		InTx(gomock.Any(), &database.TxOptions{Isolation: sql.LevelRepeatableRead}).
 		Times(1).
 		Return(nil)
 	err := database.ReadModifyUpdate(mDB, func(tx database.Store) error {
@@ -34,11 +34,11 @@ func TestReadModifyUpdate_RetryOK(t *testing.T) {
 	mDB := dbmock.NewMockStore(gomock.NewController(t))
 
 	firstUpdate := mDB.EXPECT().
-		InTx(gomock.Any(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead}).
+		InTx(gomock.Any(), &database.TxOptions{Isolation: sql.LevelRepeatableRead}).
 		Times(1).
 		Return(&pq.Error{Code: pq.ErrorCode("40001")})
 	mDB.EXPECT().
-		InTx(gomock.Any(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead}).
+		InTx(gomock.Any(), &database.TxOptions{Isolation: sql.LevelRepeatableRead}).
 		After(firstUpdate).
 		Times(1).
 		Return(nil)
@@ -55,7 +55,7 @@ func TestReadModifyUpdate_HardError(t *testing.T) {
 	mDB := dbmock.NewMockStore(gomock.NewController(t))
 
 	mDB.EXPECT().
-		InTx(gomock.Any(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead}).
+		InTx(gomock.Any(), &database.TxOptions{Isolation: sql.LevelRepeatableRead}).
 		Times(1).
 		Return(xerrors.New("a bad thing happened"))
 
@@ -71,7 +71,7 @@ func TestReadModifyUpdate_TooManyRetries(t *testing.T) {
 	mDB := dbmock.NewMockStore(gomock.NewController(t))
 
 	mDB.EXPECT().
-		InTx(gomock.Any(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead}).
+		InTx(gomock.Any(), &database.TxOptions{Isolation: sql.LevelRepeatableRead}).
 		Times(5).
 		Return(&pq.Error{Code: pq.ErrorCode("40001")})
 	err := database.ReadModifyUpdate(mDB, func(tx database.Store) error {