mirror of
https://github.com/zitadel/zitadel.git
synced 2026-11-03 12:47:09 -06:00
feat(database): publish pgxpool pool stats as metrics (#12737)
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.
- 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`.
- Adds unit tests covering the new `RegisterCounterObserver` method
(`backend/v3/instrumentation/metric_test.go`) and the pool metrics
registration (`internal/database/metrics_test.go`).
- 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
(cherry picked from commit 73de633c71)
This commit is contained in:
1 parent
7bd55e3476
commit
661b41babb
7 files changed
+223
No files matched your search
@@ -33,6 +33,7 @@ type Meter struct {
|
||||
Counters sync.Map
|
||||
UpDownSumObserver sync.Map
|
||||
ValueObservers sync.Map
|
||||
CounterObservers sync.Map
|
||||
Histograms sync.Map
|
||||
}
|
||||
|
||||
@@ -136,6 +137,20 @@ func (m *Meter) RegisterValueObserver(name, description string, callbackFunc met
|
||||
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 {
|
||||
return []metric.AddOption{metric.WithAttributes(labelsToAttributes(labels)...)}
|
||||
}
|
||||
|
||||
@@ -1,10 +1,15 @@
|
||||
package instrumentation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"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"
|
||||
)
|
||||
|
||||
@@ -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) {
|
||||
resource, err := resource.New(t.Context())
|
||||
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
|
||||
RegisterUpDownSumObserver(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
|
||||
}
|
||||
|
||||
@@ -69,3 +70,7 @@ func RegisterUpDownSumObserver(name, description string, callbackFunc metric.Int
|
||||
func RegisterValueObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
||||
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
|
||||
}
|
||||
|
||||
func (m *MockMetrics) RegisterCounterObserver(name, description string, callbackFunc metric.Int64Callback) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *MockMetrics) RegisterHistogram(name, description, unit string, buckets []float64) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -180,6 +180,7 @@ func startZitadel(ctx context.Context, config *Config, masterKey string, server
|
||||
if err != nil {
|
||||
return fmt.Errorf("cannot start DB client for queries: %w", err)
|
||||
}
|
||||
database.RegisterPoolMetrics(ctx, dbClient.Pool)
|
||||
|
||||
keyStorage, err := cryptoDB.NewKeyStorage(dbClient, masterKey)
|
||||
if err != nil {
|
||||
|
||||
@@ -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