pkg/pgxpoolobserver/observer.go

package pgxpoolobserver

import (
	"context"

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

// ObservePoolMetrics registers a callback that observes the metrics of the provided pgxpool.Pool.
// the implementation is based on https://github.com/cmackenzie1/pgxpool-prometheus
func ObservePoolMetrics(meter metric.Meter, pool *pgxpool.Pool, additionalAttributes ...attribute.KeyValue) error {
	allMetrics := []metric.Observable{}

	acquireCountMetric, err := meter.Int64ObservableCounter(
		"pgxpool.acquire_count",
		metric.WithDescription("The cumulative count of successful acquires from the pool."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, acquireCountMetric)

	acquiredDurationMetric, err := meter.Int64ObservableGauge(
		"pgxpool.acquire_duration",
		metric.WithDescription("The total duration of all successful acquires from the pool in ms."),
		metric.WithUnit("ms"),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, acquiredDurationMetric)

	avgAcquiredDurationMetric, err := meter.Int64ObservableGauge(
		"pgxpool.acquire_duration_avg",
		metric.WithDescription("The average duration of all successful acquires from the pool in ms."),
		metric.WithUnit("ms"),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, avgAcquiredDurationMetric)

	acquiredConnsMetric, err := meter.Int64ObservableGauge(
		"pgxpool.acquired_conns",
		metric.WithDescription("The number of currently acquired connections in the pool."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, acquiredConnsMetric)

	canceledAcquireCountMetric, err := meter.Int64ObservableCounter(
		"pgxpool.canceled_acquire_count",
		metric.WithDescription("The cumulative count of acquires from the pool that were canceled by a context."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, canceledAcquireCountMetric)

	constructingConnsMetric, err := meter.Int64ObservableGauge(
		"pgxpool.constructing_conns",
		metric.WithDescription("The number of conns with construction in progress in the pool."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, constructingConnsMetric)

	emptyAcquireCountMetric, err := meter.Int64ObservableCounter(
		"pgxpool.empty_acquire_count",
		metric.WithDescription("The cumulative count of successful acquires from the pool that waited for a resource to be released or constructed because the pool was empty."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, emptyAcquireCountMetric)

	idleConnsMetric, err := meter.Int64ObservableGauge(
		"pgxpool.idle_conns",
		metric.WithDescription("The number of currently idle conns in the pool."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, idleConnsMetric)

	maxConns, err := meter.Int64ObservableGauge(
		"pgxpool.max_conns",
		metric.WithDescription("The maximum size of the pool."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, maxConns)

	totalConns, err := meter.Int64ObservableGauge(
		"pgxpool.total_conns",
		metric.WithDescription("The total number of resources currently in the pool. The value is the sum of ConstructingConns, AcquiredConns, and IdleConns."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, totalConns)

	newConnsCount, err := meter.Int64ObservableCounter(
		"pgxpool.new_conns_count",
		metric.WithDescription("The cumulative count of new connections opened."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, newConnsCount)

	maxLifetimeDestroyCount, err := meter.Int64ObservableCounter(
		"pgxpool.max_lifetime_destroy_count",
		metric.WithDescription("The cumulative count of connections closed due to reaching their maximum lifetime (MaxConnLifetime)."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, maxLifetimeDestroyCount)

	maxIdleDestroyCount, err := meter.Int64ObservableCounter(
		"pgxpool.max_idle_destroy_count",
		metric.WithDescription("The cumulative count of connections closed due to reaching their maximum idle time (MaxConnIdleTime)."),
	)
	if err != nil {
		return err
	}
	allMetrics = append(allMetrics, maxIdleDestroyCount)

	_, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error {
		stat := pool.Stat()

		acquireCount := stat.AcquireCount()
		acquireDurationMS := stat.AcquireDuration().Milliseconds()

		o.ObserveInt64(acquireCountMetric, acquireCount, metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(acquiredDurationMetric, acquireDurationMS, metric.WithAttributes(additionalAttributes...))

		if acquireCount > 0 {
			o.ObserveInt64(avgAcquiredDurationMetric, acquireDurationMS/acquireCount, metric.WithAttributes(additionalAttributes...))
		}
		o.ObserveInt64(acquiredConnsMetric, int64(stat.AcquiredConns()), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(canceledAcquireCountMetric, stat.CanceledAcquireCount(), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(constructingConnsMetric, int64(stat.ConstructingConns()), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(emptyAcquireCountMetric, stat.EmptyAcquireCount(), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(idleConnsMetric, int64(stat.IdleConns()), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(maxConns, int64(stat.MaxConns()), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(totalConns, int64(stat.TotalConns()), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(newConnsCount, stat.NewConnsCount(), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(maxLifetimeDestroyCount, stat.MaxLifetimeDestroyCount(), metric.WithAttributes(additionalAttributes...))
		o.ObserveInt64(maxIdleDestroyCount, stat.MaxIdleDestroyCount(), metric.WithAttributes(additionalAttributes...))

		return nil
	}, allMetrics...)
	if err != nil {
		return err
	}

	return nil
}