mirror of
https://github.com/andwati/zitadel.git
synced 2026-10-18 05:52:51 +00:00
feat(database): publish pgxpool pool stats as metrics (#12737)
# Which Problems Are Solved ZITADEL opens its Postgres connection pool via `pgxpool.NewWithConfig` (`internal/database/postgres/pg.go`) but never reads `pgxpool.Pool.Stat()`, so operators have no way to observe pool saturation (acquired/idle/total conns vs. `MaxConns`) or connection churn (acquire waits, canceled/empty acquires, idle/lifetime destroys, new connections) through ZITADEL's existing metrics pipeline. # How the Problems Are Solved - Adds `RegisterCounterObserver` (backed by an OTel `Int64ObservableCounter`) to the `Metrics` interface, its implementation, and the test mock (`backend/v3/instrumentation/metric.go`, `backend/v3/instrumentation/metrics/metric.go`, `backend/v3/instrumentation/metrics/mock.go`), mirroring the existing `RegisterValueObserver`/`RegisterUpDownSumObserver` — there was previously no observable instrument for cumulative counters. - Adds `database.RegisterPoolMetrics(ctx, pool)` (`internal/database/metrics.go`), which registers 5 observable gauges (`acquired_conns`, `constructing_conns`, `idle_conns`, `max_conns`, `total_conns`) and 7 observable counters (`acquire_count`, `acquire_duration_milliseconds`, `canceled_acquire_count`, `empty_acquire_count`, `max_idle_destroy_count`, `max_lifetime_destroy_count`, `new_conns_count`) sourced directly from `pgxpool.Pool.Stat()`, published under `zitadel.database.pool.*` metric names. - Wires the registration call into `cmd/start/start.go` right after the pool is created, so the running server process exposes these series on `/metrics`. # Additional Changes - Adds unit tests covering the new `RegisterCounterObserver` method (`backend/v3/instrumentation/metric_test.go`) and the pool metrics registration (`internal/database/metrics_test.go`). # Additional Context - Scope is intentionally limited to `cmd/start` (the long-running process that serves `/metrics`); one-shot commands (`setup`, `mirror`, `initialise`, `key`) are left untouched since an exporter running for a single command invocation wouldn't produce useful series. - https://pkg.go.dev/github.com/jackc/pgx/v5/pgxpool#Stat
This commit is contained in:
@@ -33,6 +33,7 @@ type Meter struct {
|
|||||||
Counters sync.Map
|
Counters sync.Map
|
||||||
UpDownSumObserver sync.Map
|
UpDownSumObserver sync.Map
|
||||||
ValueObservers sync.Map
|
ValueObservers sync.Map
|
||||||
|
CounterObservers sync.Map
|
||||||
Histograms sync.Map
|
Histograms sync.Map
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -136,6 +137,20 @@ func (m *Meter) RegisterValueObserver(name, description string, callbackFunc met
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *Meter) RegisterCounterObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
||||||
|
if _, exists := m.CounterObservers.Load(name); exists {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
counter, err := m.Meter.Int64ObservableCounter(name, metric.WithInt64Callback(callbackFunc), metric.WithDescription(description))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
m.CounterObservers.Store(name, counter)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func MapToAddOption(labels map[string]attribute.Value) []metric.AddOption {
|
func MapToAddOption(labels map[string]attribute.Value) []metric.AddOption {
|
||||||
return []metric.AddOption{metric.WithAttributes(labelsToAttributes(labels)...)}
|
return []metric.AddOption{metric.WithAttributes(labelsToAttributes(labels)...)}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,10 +1,15 @@
|
|||||||
package instrumentation
|
package instrumentation
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
"go.opentelemetry.io/otel/metric"
|
||||||
|
sdk_metric "go.opentelemetry.io/otel/sdk/metric"
|
||||||
|
"go.opentelemetry.io/otel/sdk/metric/metricdata"
|
||||||
"go.opentelemetry.io/otel/sdk/resource"
|
"go.opentelemetry.io/otel/sdk/resource"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -201,6 +206,49 @@ func Test_newMeterProvider(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestMeter_RegisterCounterObserver(t *testing.T) {
|
||||||
|
reader := sdk_metric.NewManualReader()
|
||||||
|
provider := sdk_metric.NewMeterProvider(sdk_metric.WithReader(reader))
|
||||||
|
m := &Meter{Meter: provider.Meter(t.Name())}
|
||||||
|
|
||||||
|
callback := func(_ context.Context, o metric.Int64Observer) error {
|
||||||
|
o.Observe(42)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err := m.RegisterCounterObserver("test.counter.observer", "test counter observer", callback)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
// registering the same name again is a no-op, not an error.
|
||||||
|
err = m.RegisterCounterObserver("test.counter.observer", "test counter observer", callback)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
_, exists := m.CounterObservers.Load("test.counter.observer")
|
||||||
|
assert.True(t, exists)
|
||||||
|
|
||||||
|
var rm metricdata.ResourceMetrics
|
||||||
|
require.NoError(t, reader.Collect(t.Context(), &rm))
|
||||||
|
|
||||||
|
got := findMetric(t, rm, "test.counter.observer")
|
||||||
|
sum, ok := got.Data.(metricdata.Sum[int64])
|
||||||
|
require.True(t, ok, "expected an observable counter (Sum[int64]), got %T", got.Data)
|
||||||
|
assert.True(t, sum.IsMonotonic)
|
||||||
|
require.Len(t, sum.DataPoints, 1)
|
||||||
|
assert.Equal(t, int64(42), sum.DataPoints[0].Value)
|
||||||
|
}
|
||||||
|
|
||||||
|
// findMetric locates a metric by name across all scopes collected from a reader.
|
||||||
|
func findMetric(t *testing.T, rm metricdata.ResourceMetrics, name string) metricdata.Metrics {
|
||||||
|
t.Helper()
|
||||||
|
for _, sm := range rm.ScopeMetrics {
|
||||||
|
if i := slices.IndexFunc(sm.Metrics, func(m metricdata.Metrics) bool { return m.Name == name }); i >= 0 {
|
||||||
|
return sm.Metrics[i]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t.Fatalf("metric %q not found in collected data", name)
|
||||||
|
return metricdata.Metrics{}
|
||||||
|
}
|
||||||
|
|
||||||
func Test_newMeterProvider_autoexport(t *testing.T) {
|
func Test_newMeterProvider_autoexport(t *testing.T) {
|
||||||
resource, err := resource.New(t.Context())
|
resource, err := resource.New(t.Context())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ type Metrics interface {
|
|||||||
AddHistogramMeasurement(ctx context.Context, name string, value float64, labels map[string]attribute.Value) error
|
AddHistogramMeasurement(ctx context.Context, name string, value float64, labels map[string]attribute.Value) error
|
||||||
RegisterUpDownSumObserver(name, description string, callbackFunc metric.Int64Callback) error
|
RegisterUpDownSumObserver(name, description string, callbackFunc metric.Int64Callback) error
|
||||||
RegisterValueObserver(name, description string, callbackFunc metric.Int64Callback) error
|
RegisterValueObserver(name, description string, callbackFunc metric.Int64Callback) error
|
||||||
|
RegisterCounterObserver(name, description string, callbackFunc metric.Int64Callback) error
|
||||||
RegisterHistogram(name, description, unit string, buckets []float64) error
|
RegisterHistogram(name, description, unit string, buckets []float64) error
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -69,3 +70,7 @@ func RegisterUpDownSumObserver(name, description string, callbackFunc metric.Int
|
|||||||
func RegisterValueObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
func RegisterValueObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
||||||
return globalMeter().RegisterValueObserver(name, description, callbackFunc)
|
return globalMeter().RegisterValueObserver(name, description, callbackFunc)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func RegisterCounterObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
||||||
|
return globalMeter().RegisterCounterObserver(name, description, callbackFunc)
|
||||||
|
}
|
||||||
|
|||||||
@@ -66,6 +66,10 @@ func (m *MockMetrics) RegisterValueObserver(name, description string, callbackFu
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *MockMetrics) RegisterCounterObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
func (m *MockMetrics) RegisterHistogram(name, description, unit string, buckets []float64) error {
|
func (m *MockMetrics) RegisterHistogram(name, description, unit string, buckets []float64) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -183,6 +183,7 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("cannot start DB client for queries: %w", err)
|
return fmt.Errorf("cannot start DB client for queries: %w", err)
|
||||||
}
|
}
|
||||||
|
database.RegisterPoolMetrics(ctx, dbClient.Pool)
|
||||||
new_domain.SetPool(v3_postgres.PGxPool(dbClient.Pool))
|
new_domain.SetPool(v3_postgres.PGxPool(dbClient.Pool))
|
||||||
|
|
||||||
keyStorage, err := cryptoDB.NewKeyStorage(dbClient, masterKey)
|
keyStorage, err := cryptoDB.NewKeyStorage(dbClient, masterKey)
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
package database
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"log/slog"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
"go.opentelemetry.io/otel/metric"
|
||||||
|
|
||||||
|
"github.com/zitadel/zitadel/backend/v3/instrumentation/logging"
|
||||||
|
"github.com/zitadel/zitadel/backend/v3/instrumentation/metrics"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
PoolAcquiredConns = "zitadel.database.pool.acquired_conns"
|
||||||
|
PoolConstructingConns = "zitadel.database.pool.constructing_conns"
|
||||||
|
PoolIdleConns = "zitadel.database.pool.idle_conns"
|
||||||
|
PoolMaxConns = "zitadel.database.pool.max_conns"
|
||||||
|
PoolTotalConns = "zitadel.database.pool.total_conns"
|
||||||
|
|
||||||
|
PoolAcquireCount = "zitadel.database.pool.acquire_count"
|
||||||
|
PoolAcquireDurationMs = "zitadel.database.pool.acquire_duration_milliseconds"
|
||||||
|
PoolCanceledAcquireCount = "zitadel.database.pool.canceled_acquire_count"
|
||||||
|
PoolEmptyAcquireCount = "zitadel.database.pool.empty_acquire_count"
|
||||||
|
PoolMaxIdleDestroyCount = "zitadel.database.pool.max_idle_destroy_count"
|
||||||
|
PoolMaxLifetimeDestroyCount = "zitadel.database.pool.max_lifetime_destroy_count"
|
||||||
|
PoolNewConnsCount = "zitadel.database.pool.new_conns_count"
|
||||||
|
)
|
||||||
|
|
||||||
|
// RegisterPoolMetrics registers observable gauges and counters reporting pool's
|
||||||
|
// underlying *pgxpool.Pool state (see pgxpool.Pool.Stat) with the global metrics
|
||||||
|
// pipeline.
|
||||||
|
func RegisterPoolMetrics(ctx context.Context, pool *pgxpool.Pool) {
|
||||||
|
registerPoolMetrics(ctx, metrics.GlobalMeter(), pool)
|
||||||
|
}
|
||||||
|
|
||||||
|
func registerPoolMetrics(ctx context.Context, m metrics.Metrics, pool *pgxpool.Pool) {
|
||||||
|
gauge := func(name, description string, value func(*pgxpool.Stat) int64) {
|
||||||
|
err := m.RegisterValueObserver(name, description, func(_ context.Context, o metric.Int64Observer) error {
|
||||||
|
o.Observe(value(pool.Stat()))
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
logging.OnError(ctx, err).Error("failed to register database pool gauge metric", slog.String("metric", name))
|
||||||
|
}
|
||||||
|
counter := func(name, description string, value func(*pgxpool.Stat) int64) {
|
||||||
|
err := m.RegisterCounterObserver(name, description, func(_ context.Context, o metric.Int64Observer) error {
|
||||||
|
o.Observe(value(pool.Stat()))
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
logging.OnError(ctx, err).Error("failed to register database pool counter metric", slog.String("metric", name))
|
||||||
|
}
|
||||||
|
|
||||||
|
gauge(PoolAcquiredConns, "Number of currently acquired connections in the pool", func(s *pgxpool.Stat) int64 { return int64(s.AcquiredConns()) })
|
||||||
|
gauge(PoolConstructingConns, "Number of connections currently being established", func(s *pgxpool.Stat) int64 { return int64(s.ConstructingConns()) })
|
||||||
|
gauge(PoolIdleConns, "Number of currently idle connections in the pool", func(s *pgxpool.Stat) int64 { return int64(s.IdleConns()) })
|
||||||
|
gauge(PoolMaxConns, "Maximum size of the pool", func(s *pgxpool.Stat) int64 { return int64(s.MaxConns()) })
|
||||||
|
gauge(PoolTotalConns, "Total number of connections currently in the pool", func(s *pgxpool.Stat) int64 { return int64(s.TotalConns()) })
|
||||||
|
|
||||||
|
counter(PoolAcquireCount, "Cumulative count of successful acquires from the pool", func(s *pgxpool.Stat) int64 { return s.AcquireCount() })
|
||||||
|
counter(PoolAcquireDurationMs, "Cumulative time spent waiting for successful acquires from the pool, in milliseconds", func(s *pgxpool.Stat) int64 { return s.AcquireDuration().Milliseconds() })
|
||||||
|
counter(PoolCanceledAcquireCount, "Cumulative count of acquires from the pool that were canceled by a context", func(s *pgxpool.Stat) int64 { return s.CanceledAcquireCount() })
|
||||||
|
counter(PoolEmptyAcquireCount, "Cumulative count of successful acquires that waited for a resource to be released or constructed because the pool was empty", func(s *pgxpool.Stat) int64 { return s.EmptyAcquireCount() })
|
||||||
|
counter(PoolMaxIdleDestroyCount, "Cumulative count of connections destroyed because they exceeded MaxConnIdleTime", func(s *pgxpool.Stat) int64 { return s.MaxIdleDestroyCount() })
|
||||||
|
counter(PoolMaxLifetimeDestroyCount, "Cumulative count of connections destroyed because they exceeded MaxConnLifetime", func(s *pgxpool.Stat) int64 { return s.MaxLifetimeDestroyCount() })
|
||||||
|
counter(PoolNewConnsCount, "Cumulative count of new connections opened", func(s *pgxpool.Stat) int64 { return s.NewConnsCount() })
|
||||||
|
}
|
||||||
@@ -0,0 +1,84 @@
|
|||||||
|
package database
|
||||||
|
|
||||||
|
import (
|
||||||
|
"slices"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
sdk_metric "go.opentelemetry.io/otel/sdk/metric"
|
||||||
|
"go.opentelemetry.io/otel/sdk/metric/metricdata"
|
||||||
|
|
||||||
|
"github.com/zitadel/zitadel/backend/v3/instrumentation"
|
||||||
|
)
|
||||||
|
|
||||||
|
func Test_registerPoolMetrics(t *testing.T) {
|
||||||
|
cfg, err := pgxpool.ParseConfig("postgres://user:pass@localhost:5432/db")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
// pgxpool.NewWithConfig does not establish any connection, so this works
|
||||||
|
// without a reachable database.
|
||||||
|
pool, err := pgxpool.NewWithConfig(t.Context(), cfg)
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(pool.Close)
|
||||||
|
|
||||||
|
reader := sdk_metric.NewManualReader()
|
||||||
|
provider := sdk_metric.NewMeterProvider(sdk_metric.WithReader(reader))
|
||||||
|
m := &instrumentation.Meter{Meter: provider.Meter(t.Name())}
|
||||||
|
|
||||||
|
registerPoolMetrics(t.Context(), m, pool)
|
||||||
|
|
||||||
|
var rm metricdata.ResourceMetrics
|
||||||
|
require.NoError(t, reader.Collect(t.Context(), &rm))
|
||||||
|
|
||||||
|
gauges := map[string]int64{
|
||||||
|
PoolAcquiredConns: 0,
|
||||||
|
PoolConstructingConns: 0,
|
||||||
|
PoolIdleConns: 0,
|
||||||
|
PoolMaxConns: int64(cfg.MaxConns),
|
||||||
|
PoolTotalConns: 0,
|
||||||
|
}
|
||||||
|
for name, want := range gauges {
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
got := findMetric(t, rm, name)
|
||||||
|
gauge, ok := got.Data.(metricdata.Gauge[int64])
|
||||||
|
require.True(t, ok, "expected a gauge, got %T", got.Data)
|
||||||
|
require.Len(t, gauge.DataPoints, 1)
|
||||||
|
assert.Equal(t, want, gauge.DataPoints[0].Value)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
counters := []string{
|
||||||
|
PoolAcquireCount,
|
||||||
|
PoolAcquireDurationMs,
|
||||||
|
PoolCanceledAcquireCount,
|
||||||
|
PoolEmptyAcquireCount,
|
||||||
|
PoolMaxIdleDestroyCount,
|
||||||
|
PoolMaxLifetimeDestroyCount,
|
||||||
|
PoolNewConnsCount,
|
||||||
|
}
|
||||||
|
for _, name := range counters {
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
got := findMetric(t, rm, name)
|
||||||
|
sum, ok := got.Data.(metricdata.Sum[int64])
|
||||||
|
require.True(t, ok, "expected an observable counter (Sum[int64]), got %T", got.Data)
|
||||||
|
assert.True(t, sum.IsMonotonic)
|
||||||
|
require.Len(t, sum.DataPoints, 1)
|
||||||
|
// A freshly created, unused pool has not acquired or destroyed any connections yet.
|
||||||
|
assert.Equal(t, int64(0), sum.DataPoints[0].Value)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// findMetric locates a metric by name across all scopes collected from a reader.
|
||||||
|
func findMetric(t *testing.T, rm metricdata.ResourceMetrics, name string) metricdata.Metrics {
|
||||||
|
t.Helper()
|
||||||
|
for _, sm := range rm.ScopeMetrics {
|
||||||
|
if i := slices.IndexFunc(sm.Metrics, func(m metricdata.Metrics) bool { return m.Name == name }); i >= 0 {
|
||||||
|
return sm.Metrics[i]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
t.Fatalf("metric %q not found in collected data", name)
|
||||||
|
return metricdata.Metrics{}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user