diff --git a/backend/v3/instrumentation/metric.go b/backend/v3/instrumentation/metric.go index 2221cdfc3..d80db9818 100644 --- a/backend/v3/instrumentation/metric.go +++ b/backend/v3/instrumentation/metric.go @@ -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)...)} } diff --git a/backend/v3/instrumentation/metric_test.go b/backend/v3/instrumentation/metric_test.go index 46cf3d7e8..a72727024 100644 --- a/backend/v3/instrumentation/metric_test.go +++ b/backend/v3/instrumentation/metric_test.go @@ -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) diff --git a/backend/v3/instrumentation/metrics/metric.go b/backend/v3/instrumentation/metrics/metric.go index 3fe5aa7fb..65cd3edaf 100644 --- a/backend/v3/instrumentation/metrics/metric.go +++ b/backend/v3/instrumentation/metrics/metric.go @@ -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) +} diff --git a/backend/v3/instrumentation/metrics/mock.go b/backend/v3/instrumentation/metrics/mock.go index b28a6f5d4..f805c99f4 100644 --- a/backend/v3/instrumentation/metrics/mock.go +++ b/backend/v3/instrumentation/metrics/mock.go @@ -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 } diff --git a/cmd/start/start.go b/cmd/start/start.go index 8d04e57bd..ed2366688 100644 --- a/cmd/start/start.go +++ b/cmd/start/start.go @@ -183,6 +183,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) new_domain.SetPool(v3_postgres.PGxPool(dbClient.Pool)) keyStorage, err := cryptoDB.NewKeyStorage(dbClient, masterKey) diff --git a/internal/database/metrics.go b/internal/database/metrics.go new file mode 100644 index 000000000..ce6f71fb4 --- /dev/null +++ b/internal/database/metrics.go @@ -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() }) +} diff --git a/internal/database/metrics_test.go b/internal/database/metrics_test.go new file mode 100644 index 000000000..45f1eb367 --- /dev/null +++ b/internal/database/metrics_test.go @@ -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{} +}