pgx.go

package obs

import (
	"context"

	"github.com/jackc/pgx/v5/pgxpool"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
)

// PoolStats wires async gauges that expose pgxpool connection-pool
// counters under db.pool.*. Call once after the pool is created.
// Multiple pools may be registered with distinct names (e.g.
// "hopper", "replica"); the name becomes a "pool" attribute.
//
// Safe to call before [Init]: the global no-op meter accepts the
// instrument creation and the callbacks do nothing useful until Init
// installs the real provider.
func PoolStats(name string, pool *pgxpool.Pool) error {
	m := Meter()

	open, err := m.Int64ObservableGauge("db.pool.open",
		metric.WithDescription("Open connections (acquired + idle)."))
	if err != nil {
		return err
	}
	idle, err := m.Int64ObservableGauge("db.pool.idle",
		metric.WithDescription("Idle connections."))
	if err != nil {
		return err
	}
	acquired, err := m.Int64ObservableGauge("db.pool.acquired",
		metric.WithDescription("Connections currently checked out."))
	if err != nil {
		return err
	}
	maxConns, err := m.Int64ObservableGauge("db.pool.max",
		metric.WithDescription("Maximum size of the pool. Saturation is acquired/max."))
	if err != nil {
		return err
	}
	constructing, err := m.Int64ObservableGauge("db.pool.constructing",
		metric.WithDescription("Connections currently being established."))
	if err != nil {
		return err
	}
	waitCount, err := m.Int64ObservableCounter("db.pool.wait_count",
		metric.WithDescription("Total acquires that had to wait for an empty pool."))
	if err != nil {
		return err
	}
	waitDur, err := m.Float64ObservableCounter("db.pool.wait_duration",
		metric.WithUnit("s"),
		metric.WithDescription("Total time blocked waiting for a connection."))
	if err != nil {
		return err
	}
	canceled, err := m.Int64ObservableCounter("db.pool.canceled_acquire",
		metric.WithDescription("Acquires canceled by context (timeout/cancellation) while waiting; rises under contention."))
	if err != nil {
		return err
	}
	newConns, err := m.Int64ObservableCounter("db.pool.new_conns",
		metric.WithDescription("Total new connections opened over the pool's lifetime."))
	if err != nil {
		return err
	}

	attrs := metric.WithAttributes(attribute.String("pool", name))
	_, err = m.RegisterCallback(func(_ context.Context, o metric.Observer) error {
		s := pool.Stat()
		o.ObserveInt64(open, int64(s.TotalConns()), attrs)
		o.ObserveInt64(idle, int64(s.IdleConns()), attrs)
		o.ObserveInt64(acquired, int64(s.AcquiredConns()), attrs)
		o.ObserveInt64(maxConns, int64(s.MaxConns()), attrs)
		o.ObserveInt64(constructing, int64(s.ConstructingConns()), attrs)
		o.ObserveInt64(waitCount, s.EmptyAcquireCount(), attrs)
		o.ObserveFloat64(waitDur, s.EmptyAcquireWaitTime().Seconds(), attrs)
		o.ObserveInt64(canceled, s.CanceledAcquireCount(), attrs)
		o.ObserveInt64(newConns, s.NewConnsCount(), attrs)
		return nil
	}, open, idle, acquired, maxConns, constructing, waitCount, waitDur, canceled, newConns)
	return err
}