feat: Expose pgxpool metrics

README.md

@@ -38,6 +38,10 @@ conn, err := pgxpool.NewConfig(ctx, cfg)
 if err != nil {
     return nil, fmt.Errorf("connect to database: %w", err)
 }
+
+if err := otelpgx.RecordStats(conn); err != nil {
+    return nil, fmt.Errorf("unable to record database stats: %w", err)
+}
 ```
 
 See [options.go](options.go) for the full list of options.

go.mod

@@ -4,15 +4,19 @@ go 1.18
 
 require (
 	github.com/jackc/pgx/v5 v5.2.0
-	go.opentelemetry.io/otel v1.12.0
-	go.opentelemetry.io/otel/trace v1.12.0
+	go.opentelemetry.io/otel v1.13.0
+	go.opentelemetry.io/otel/metric v0.36.0
+	go.opentelemetry.io/otel/trace v1.13.0
 )
 
 require (
 	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/puddle/v2 v2.1.2 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/crypto v0.5.0 // indirect
+	golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 // indirect
 	golang.org/x/text v0.6.0 // indirect
 )

go.sum

@@ -8,38 +8,32 @@ github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
-github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b h1:C8S2+VttkHFdOOCXJe+YGfa4vHYwlt4Zx+IVXQ97jYg=
-github.com/jackc/pgservicefile v0.0.0-20200714003250-2b9c44734f2b/go.mod h1:vsD4gTJCa9TptPL8sPkXrLZ+hDuNrZCnj29CQpr4X1E=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
 github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.0.4 h1:r5O6y84qHX/z/HZV40JBdx2obsHz7/uRj5b+CcYEdeY=
-github.com/jackc/pgx/v5 v5.0.4/go.mod h1:U0ynklHtgg43fue9Ly30w3OCSTDPlXjig9ghrNGaguQ=
 github.com/jackc/pgx/v5 v5.2.0 h1:NdPpngX0Y6z6XDFKqmFQaE+bCtkqzvQIOt1wvBlAqs8=
 github.com/jackc/pgx/v5 v5.2.0/go.mod h1:Ptn7zmohNsWEsdxRawMzk3gaKma2obW+NWTnKa0S4nk=
+github.com/jackc/puddle/v2 v2.1.2 h1:0f7vaaXINONKTsxYDn4otOAiJanX/BMeAtY//BXqzlg=
+github.com/jackc/puddle/v2 v2.1.2/go.mod h1:2lpufsF5mRHO6SuZkm0fNYxM6SWHfvyFj62KwNzgels=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
-go.opentelemetry.io/otel v1.11.1 h1:4WLLAmcfkmDk2ukNXJyq3/kiz/3UzCaYq6PskJsaou4=
-go.opentelemetry.io/otel v1.11.1/go.mod h1:1nNhXBbWSD0nsL38H6btgnFN2k4i0sNLHNNMZMSbUGE=
-go.opentelemetry.io/otel v1.12.0 h1:IgfC7kqQrRccIKuB7Cl+SRUmsKbEwSGPr0Eu+/ht1SQ=
-go.opentelemetry.io/otel v1.12.0/go.mod h1:geaoz0L0r1BEOR81k7/n9W4TCXYCJ7bPO7K374jQHG0=
-go.opentelemetry.io/otel/trace v1.11.1 h1:ofxdnzsNrGBYXbP7t7zpUK281+go5rF7dvdIZXF8gdQ=
-go.opentelemetry.io/otel/trace v1.11.1/go.mod h1:f/Q9G7vzk5u91PhbmKbg1Qn0rzH1LJ4vbPHFGkTPtOk=
-go.opentelemetry.io/otel/trace v1.12.0 h1:p28in++7Kd0r2d8gSt931O57fdjUyWxkVbESuILAeUc=
-go.opentelemetry.io/otel/trace v1.12.0/go.mod h1:pHlgBynn6s25qJ2szD+Bv+iwKJttjHSI3lUAyf0GNuQ=
-golang.org/x/crypto v0.1.0 h1:MDRAIl0xIo9Io2xV565hzXHw3zVseKrJKodhohM5CjU=
-golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+go.opentelemetry.io/otel v1.13.0 h1:1ZAKnNQKwBBxFtww/GwxNUyTf0AxkZzrukO8MeXqe4Y=
+go.opentelemetry.io/otel v1.13.0/go.mod h1:FH3RtdZCzRkJYFTCsAKDy9l/XYjMdNv6QrkFFB8DvVg=
+go.opentelemetry.io/otel/metric v0.36.0 h1:t0lgGI+L68QWt3QtOIlqM9gXoxqxWLhZ3R/e5oOAY0Q=
+go.opentelemetry.io/otel/metric v0.36.0/go.mod h1:wKVw57sd2HdSZAzyfOM9gTqqE8v7CbqWsYL6AyrH9qk=
+go.opentelemetry.io/otel/trace v1.13.0 h1:CBgRZ6ntv+Amuj1jDsMhZtlAPT6gbyIRdaIzFhfBSdY=
+go.opentelemetry.io/otel/trace v1.13.0/go.mod h1:muCvmmO9KKpvuXSf3KKAXXB2ygNYHQ+ZfI5X08d3tds=
+go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
+go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 golang.org/x/crypto v0.5.0 h1:U/0M97KRkSFvyD/3FSmdP5W5swImpNgle/EHFhOsQPE=
 golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
-golang.org/x/text v0.4.0 h1:BrVqGRd7+k1DiOgtnFvAkoQEWQvBc25ouMJM6429SFg=
-golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7 h1:ZrnxWX62AgTKOSagEqxvb3ffipvEDX2pl7E1TdqLqIc=
+golang.org/x/sync v0.0.0-20220923202941-7f9b1623fab7/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/text v0.6.0 h1:3XmdazWV+ubf7QgHSTWeykHOci5oeekaGJBLkrkaw4k=
 golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

internal/constants.go

@@ -5,6 +5,9 @@ const (
 	// on each span.
 	TracerName = "github.com/exaring/otelpgx"
 
+	// MeterName is the name of the metric meter.
+	MeterName = "github.com/exaring/otelpgx"
+
 	// InstrumentationVersion is the version of the otelpgx library. This will
 	// be used as an attribute on each span.
 	InstrumentationVersion = "v0.4.1"

metric.go

@@ -0,0 +1,16 @@
+package otelpgx
+
+const (
+	pgxPoolAcquireCount            = "pgxpool_acquires"
+	pgxpoolAcquireDuration         = "pgxpool_acquire_duration"
+	pgxpoolAcquiredConns           = "pgxpool_acquired_conns"
+	pgxpoolCancelledAcquires       = "pgxpool_canceled_acquires"
+	pgxpoolConstructingConns       = "pgxpool_constructing_conns"
+	pgxpoolEmptyAcquire            = "pgxpool_empty_acquire"
+	pgxpoolIdleConns               = "pgxpool_idle_conns"
+	pgxpoolMaxConns                = "pgxpool_max_conns"
+	pgxpoolMaxIdleDestroyCount     = "pgxpool_max_idle_destroys"
+	pgxpoolMaxLifetimeDestroyCount = "pgxpool_max_lifetime_destroys"
+	pgxpoolNewConnsCount           = "pgxpool_new_conns"
+	pgxpoolTotalConns              = "pgxpool_total_conns"
+)

options.go

@@ -1,7 +1,10 @@
 package otelpgx
 
 import (
+	"time"
+
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
 )
 
@@ -57,3 +60,43 @@ func WithIncludeQueryParameters() Option {
 		cfg.includeParams = true
 	})
 }
+
+// StatsOption allows for managing otelsql configuration using functional options.
+type StatsOption interface {
+	applyStatsOptions(o *statsOptions)
+}
+
+type statsOptions struct {
+	// meterProvider sets the metric.MeterProvider. If nil, the global Provider will be used.
+	meterProvider metric.MeterProvider
+
+	// minimumReadDBStatsInterval sets the minimum interval between calls to db.Stats(). Negative values are ignored.
+	minimumReadDBStatsInterval time.Duration
+
+	// defaultAttributes will be set to each metrics as default.
+	defaultAttributes []attribute.KeyValue
+}
+
+type statsOptionFunc func(o *statsOptions)
+
+func (f statsOptionFunc) applyStatsOptions(o *statsOptions) {
+	f(o)
+}
+
+// WithMeterProvider sets meter provider.
+func WithMeterProvider(p metric.MeterProvider) StatsOption {
+	return struct {
+		statsOptionFunc
+	}{
+		statsOptionFunc: func(o *statsOptions) {
+			o.meterProvider = p
+		},
+	}
+}
+
+// WithMinimumReadDBStatsInterval sets the minimum interval between calls to db.Stats(). Negative values are ignored.
+func WithMinimumReadDBStatsInterval(interval time.Duration) StatsOption {
+	return statsOptionFunc(func(o *statsOptions) {
+		o.minimumReadDBStatsInterval = interval
+	})
+}

stats.go

@@ -0,0 +1,209 @@
+package otelpgx
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/exaring/otelpgx/internal"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/metric/global"
+	"go.opentelemetry.io/otel/metric/instrument"
+	"go.opentelemetry.io/otel/metric/unit"
+	semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
+)
+
+// defaultMinimumReadDBStatsInterval is the default minimum interval between calls to db.Stats().
+const defaultMinimumReadDBStatsInterval = time.Second
+
+// RecordStats records database statistics for provided pgxpool.Pool at the provided interval.
+func RecordStats(db *pgxpool.Pool, opts ...StatsOption) error {
+	o := statsOptions{
+		meterProvider:              global.MeterProvider(),
+		minimumReadDBStatsInterval: defaultMinimumReadDBStatsInterval,
+		defaultAttributes: []attribute.KeyValue{
+			semconv.DBSystemPostgreSQL,
+		},
+	}
+
+	for _, opt := range opts {
+		opt.applyStatsOptions(&o)
+	}
+
+	meter := o.meterProvider.Meter(internal.MeterName)
+
+	return recordStats(meter, db, o.minimumReadDBStatsInterval, o.defaultAttributes...)
+}
+
+func recordStats(
+	meter metric.Meter,
+	db *pgxpool.Pool,
+	minimumReadDBStatsInterval time.Duration,
+	attrs ...attribute.KeyValue,
+) error {
+	var (
+		err error
+
+		acquireCount                         instrument.Int64ObservableCounter
+		acquireDuration                      instrument.Float64ObservableCounter
+		acquiredConns                        instrument.Int64ObservableUpDownCounter
+		cancelledAcquires                    instrument.Int64ObservableCounter
+		constructingConns                    instrument.Int64ObservableUpDownCounter
+		emptyAcquires                        instrument.Int64ObservableCounter
+		idleConns                            instrument.Int64ObservableUpDownCounter
+		maxConns                             instrument.Int64ObservableGauge
+		maxIdleDestroyCount                  instrument.Int64ObservableCounter
+		maxLifetimeDestroyCountifetimeClosed instrument.Int64ObservableCounter
+		newConnsCount                        instrument.Int64ObservableCounter
+		totalConns                           instrument.Int64ObservableUpDownCounter
+
+		dbStats     *pgxpool.Stat
+		lastDBStats time.Time
+
+		// lock prevents a race between batch observer and instrument registration.
+		lock sync.Mutex
+	)
+
+	lock.Lock()
+	defer lock.Unlock()
+
+	if acquireCount, err = meter.Int64ObservableCounter(
+		pgxPoolAcquireCount,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Cumulative count of successful acquires from the pool."),
+	); err != nil {
+		return err
+	}
+
+	if acquireDuration, err = meter.Float64ObservableCounter(
+		pgxpoolAcquireDuration,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Total duration of all successful acquires from the pool in nanoseconds."),
+	); err != nil {
+		return err
+	}
+
+	if acquiredConns, err = meter.Int64ObservableUpDownCounter(
+		pgxpoolAcquiredConns,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Number of currently acquired connections in the pool."),
+	); err != nil {
+		return err
+	}
+
+	if cancelledAcquires, err = meter.Int64ObservableCounter(
+		pgxpoolCancelledAcquires,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Cumulative count of acquires from the pool that were canceled by a context."),
+	); err != nil {
+		return err
+	}
+
+	if constructingConns, err = meter.Int64ObservableUpDownCounter(
+		pgxpoolConstructingConns,
+		instrument.WithUnit(unit.Milliseconds),
+		instrument.WithDescription("Number of conns with construction in progress in the pool."),
+	); err != nil {
+		return err
+	}
+
+	if emptyAcquires, err = meter.Int64ObservableCounter(
+		pgxpoolEmptyAcquire,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Cumulative count of successful acquires from the pool that waited for a resource to be released or constructed because the pool was empty."),
+	); err != nil {
+		return err
+	}
+
+	if idleConns, err = meter.Int64ObservableUpDownCounter(
+		pgxpoolIdleConns,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Number of currently idle conns in the pool."),
+	); err != nil {
+		return err
+	}
+
+	if maxConns, err = meter.Int64ObservableGauge(
+		pgxpoolMaxConns,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Maximum size of the pool."),
+	); err != nil {
+		return err
+	}
+
+	if maxIdleDestroyCount, err = meter.Int64ObservableCounter(
+		pgxpoolMaxIdleDestroyCount,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Cumulative count of connections destroyed because they exceeded MaxConnIdleTime."),
+	); err != nil {
+		return err
+	}
+
+	if maxLifetimeDestroyCountifetimeClosed, err = meter.Int64ObservableCounter(
+		pgxpoolMaxLifetimeDestroyCount,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Cumulative count of connections destroyed because they exceeded MaxConnLifetime."),
+	); err != nil {
+		return err
+	}
+
+	if newConnsCount, err = meter.Int64ObservableCounter(
+		pgxpoolNewConnsCount,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Cumulative count of new connections opened."),
+	); err != nil {
+		return err
+	}
+
+	if totalConns, err = meter.Int64ObservableUpDownCounter(
+		pgxpoolTotalConns,
+		instrument.WithUnit(unit.Dimensionless),
+		instrument.WithDescription("Total number of resources currently in the pool. The value is the sum of ConstructingConns, AcquiredConns, and IdleConns."),
+	); err != nil {
+		return err
+	}
+
+	_, err = meter.RegisterCallback(
+		func(ctx context.Context, o metric.Observer) error {
+			lock.Lock()
+			defer lock.Unlock()
+
+			now := time.Now()
+			if now.Sub(lastDBStats) >= minimumReadDBStatsInterval {
+				dbStats = db.Stat()
+				lastDBStats = now
+			}
+
+			o.ObserveInt64(acquireCount, dbStats.AcquireCount(), attrs...)
+			o.ObserveFloat64(acquireDuration, float64(dbStats.AcquireDuration())/1e6, attrs...)
+			o.ObserveInt64(acquiredConns, int64(dbStats.AcquiredConns()), attrs...)
+			o.ObserveInt64(cancelledAcquires, dbStats.CanceledAcquireCount(), attrs...)
+			o.ObserveInt64(constructingConns, int64(dbStats.ConstructingConns()), attrs...)
+			o.ObserveInt64(emptyAcquires, dbStats.EmptyAcquireCount(), attrs...)
+			o.ObserveInt64(idleConns, int64(dbStats.IdleConns()), attrs...)
+			o.ObserveInt64(maxConns, int64(dbStats.MaxConns()), attrs...)
+			o.ObserveInt64(maxIdleDestroyCount, dbStats.MaxIdleDestroyCount(), attrs...)
+			o.ObserveInt64(maxLifetimeDestroyCountifetimeClosed, dbStats.MaxLifetimeDestroyCount(), attrs...)
+			o.ObserveInt64(newConnsCount, dbStats.NewConnsCount(), attrs...)
+			o.ObserveInt64(totalConns, int64(dbStats.TotalConns()), attrs...)
+
+			return nil
+		},
+		acquireCount,
+		acquireDuration,
+		acquiredConns,
+		cancelledAcquires,
+		constructingConns,
+		emptyAcquires,
+		idleConns,
+		maxConns,
+		maxIdleDestroyCount,
+		maxLifetimeDestroyCountifetimeClosed,
+		newConnsCount,
+		totalConns,
+	)
+
+	return err
+}