4 Commits

Author SHA1 Message Date
66a72265ce feat(pgdb): export config discovery functions for display purposes
Add FindConfig, ParseURIConfig, and GetConfigFiles as exported functions
to allow callers to retrieve database configuration info without
establishing a connection. Useful for version/startup info display.
2025-11-29 13:04:36 -08:00
94b718a925 fix(database): correct metrics and improve error handling
- Fix metrics double-counting: track deltas for WaitCount/WaitDuration
  instead of adding cumulative values each tick
- Replace fmt.Printf with structured logging in pool monitor
- Add PoolOptions validation (MaxConns > 0, MinConns >= 0)
- Warn when DATABASE_URI overrides non-default PoolOptions
- Improve findAndParseConfig to report all tried files and errors
- Remove dead code in pgdb/config.go (unreachable host default)
- Fix errcheck lint issues for file.Close() calls
- Add context parameter to OpenDBMonitor() (breaking change)
2025-11-29 12:56:49 -08:00
283d3936f6 feat(pgdb): add DATABASE_URI environment variable support
Add support for PostgreSQL connection URIs via DATABASE_URI env var.
When set, it takes precedence over config files and PoolOptions are
ignored (pool settings can be specified in URI query string).
2025-11-29 11:52:45 -08:00
7291f00f48 feat(metricsserver): add Gatherer method
Add explicit Gatherer() method to improve API discoverability
and prevent users from accidentally using prometheus.DefaultGatherer
instead of the custom registry.

Changes:
- Add Gatherer() method returning prometheus.Gatherer interface
- Add NewWithDefaultGatherer() constructor for opt-in default usage
- Update package docs with usage examples
- Add tests for both gatherer modes
2025-10-12 16:13:19 -07:00
10 changed files with 352 additions and 57 deletions

View File

@@ -50,7 +50,7 @@ func createConnector(configFile string) CreateConnectorFunc {
if err != nil {
return nil, err
}
defer dbFile.Close()
defer func() { _ = dbFile.Close() }()
dec := yaml.NewDecoder(dbFile)
cfg := Config{}

View File

@@ -7,9 +7,7 @@ import (
)
// Mock types for testing SQLC integration patterns
type mockQueries struct {
db DBTX
}
type mockQueries struct{}
type mockQueriesTx struct {
*mockQueries
@@ -58,7 +56,7 @@ func TestSQLCIntegration(t *testing.T) {
// Verify our DB interface is compatible with what SQLC expects
var dbInterface DB[*mockQueriesTx]
var mockDB *mockQueries = &mockQueries{}
mockDB := &mockQueries{}
dbInterface = mockDB
// Test that our transaction helper can work with this pattern

View File

@@ -3,10 +3,10 @@ package database
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/prometheus/client_golang/prometheus"
"go.ntppool.org/common/logger"
)
// DatabaseMetrics holds the Prometheus metrics for database connection pool monitoring
@@ -16,6 +16,10 @@ type DatabaseMetrics struct {
ConnectionsInUse prometheus.Gauge
ConnectionsWaitCount prometheus.Counter
ConnectionsWaitDuration prometheus.Histogram
// Track last values for delta calculation (cumulative stats from sql.DBStats)
lastWaitCount int64
lastWaitDuration time.Duration
}
// NewDatabaseMetrics creates a new set of database metrics and registers them
@@ -67,26 +71,44 @@ func monitorConnectionPool(ctx context.Context, db *sql.DB, registerer prometheu
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
log := logger.FromContext(ctx)
for {
select {
case <-ctx.Done():
log.InfoContext(ctx, "database connection pool monitor stopped")
return
case <-ticker.C:
stats := db.Stats()
// Update gauge metrics (current state)
metrics.ConnectionsOpen.Set(float64(stats.OpenConnections))
metrics.ConnectionsIdle.Set(float64(stats.Idle))
metrics.ConnectionsInUse.Set(float64(stats.InUse))
metrics.ConnectionsWaitCount.Add(float64(stats.WaitCount))
if stats.WaitDuration > 0 {
metrics.ConnectionsWaitDuration.Observe(stats.WaitDuration.Seconds())
// Update counter with delta (WaitCount is cumulative in sql.DBStats)
waitCountDelta := stats.WaitCount - metrics.lastWaitCount
if waitCountDelta > 0 {
metrics.ConnectionsWaitCount.Add(float64(waitCountDelta))
metrics.lastWaitCount = stats.WaitCount
}
// Update histogram with delta (WaitDuration is cumulative in sql.DBStats)
waitDurationDelta := stats.WaitDuration - metrics.lastWaitDuration
if waitDurationDelta > 0 {
metrics.ConnectionsWaitDuration.Observe(waitDurationDelta.Seconds())
metrics.lastWaitDuration = stats.WaitDuration
}
// Log connection pool stats for high usage or waiting
if stats.OpenConnections > 20 || stats.WaitCount > 0 {
fmt.Printf("Connection pool stats: open=%d idle=%d in_use=%d wait_count=%d wait_duration=%s\n",
stats.OpenConnections, stats.Idle, stats.InUse, stats.WaitCount, stats.WaitDuration)
if stats.OpenConnections > 20 || waitCountDelta > 0 {
log.WarnContext(ctx, "high database connection usage",
"open", stats.OpenConnections,
"idle", stats.Idle,
"in_use", stats.InUse,
"wait_count", stats.WaitCount,
"wait_duration", stats.WaitDuration,
)
}
}
}

View File

@@ -93,8 +93,34 @@ sslmode: prefer
## Environment Variables
- `DATABASE_URI` - PostgreSQL connection URI (takes precedence over config files)
- `DATABASE_CONFIG_FILE` - Override config file location
### URI Format
Standard PostgreSQL URI format:
```
postgresql://user:password@host:port/database?sslmode=require&pool_max_conns=10
```
Pool settings can be included in the URI query string:
- `pool_max_conns`, `pool_min_conns`
- `pool_max_conn_lifetime`, `pool_max_conn_idle_time`
- `pool_health_check_period`
When using `DATABASE_URI`, `PoolOptions` are ignored and all settings come from the URI.
Example with CloudNativePG:
```yaml
# Mount the secret's 'uri' key as DATABASE_URI
env:
- name: DATABASE_URI
valueFrom:
secretKeyRef:
name: mydb-app
key: uri
```
## When to Use
**Use `pgdb.OpenPool()`** (this package) when:

View File

@@ -27,15 +27,10 @@ func CreatePoolConfig(cfg *database.PostgresConfig) (*pgxpool.Config, error) {
"require": true, "verify-ca": true, "verify-full": true,
}
if cfg.SSLMode != "" && !validSSLModes[cfg.SSLMode] {
return nil, fmt.Errorf("postgres: invalid sslmode: %s", cfg.SSLMode)
}
// Set defaults
host := cfg.Host
if host == "" {
host = "localhost"
return nil, fmt.Errorf("postgres: invalid sslmode: %s (valid: disable, allow, prefer, require, verify-ca, verify-full)", cfg.SSLMode)
}
// Apply defaults for optional fields (host is validated as required above)
port := cfg.Port
if port == 0 {
port = 5432
@@ -49,7 +44,7 @@ func CreatePoolConfig(cfg *database.PostgresConfig) (*pgxpool.Config, error) {
// Build connection string WITHOUT password
// We'll set the password separately in the config
connString := fmt.Sprintf("host=%s port=%d user=%s dbname=%s sslmode=%s",
host, port, cfg.User, cfg.Name, sslmode)
cfg.Host, port, cfg.User, cfg.Name, sslmode)
// Parse the connection string
poolConfig, err := pgxpool.ParseConfig(connString)

View File

@@ -2,12 +2,14 @@ package pgdb
import (
"context"
"errors"
"fmt"
"os"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"go.ntppool.org/common/database"
"go.ntppool.org/common/logger"
"gopkg.in/yaml.v3"
)
@@ -40,7 +42,7 @@ type PoolOptions struct {
// DefaultPoolOptions returns sensible defaults for pgxpool
func DefaultPoolOptions() PoolOptions {
return PoolOptions{
ConfigFiles: getConfigFiles(),
ConfigFiles: GetConfigFiles(),
MinConns: 0,
MaxConns: 25,
MaxConnLifetime: time.Hour,
@@ -51,25 +53,66 @@ func DefaultPoolOptions() PoolOptions {
// OpenPool opens a native pgx connection pool with the specified configuration
// This is the primary and recommended way to connect to PostgreSQL
//
// Configuration precedence (highest to lowest):
// 1. DATABASE_URI environment variable (pool settings can be included in URI)
// 2. DATABASE_CONFIG_FILE environment variable (YAML)
// 3. Default config files (database.yaml, /vault/secrets/database.yaml)
//
// When using DATABASE_URI, pool settings (pool_max_conns, pool_min_conns, etc.)
// can be specified in the URI query string and PoolOptions are ignored.
// When using config files, PoolOptions are applied.
func OpenPool(ctx context.Context, options PoolOptions) (*pgxpool.Pool, error) {
// Find and read config file
pgCfg, err := findAndParseConfig(options.ConfigFiles)
if err != nil {
return nil, err
log := logger.FromContext(ctx)
// Validate PoolOptions
if options.MaxConns <= 0 {
return nil, fmt.Errorf("pgdb: MaxConns must be positive, got: %d", options.MaxConns)
}
if options.MinConns < 0 {
return nil, fmt.Errorf("pgdb: MinConns must be non-negative, got: %d", options.MinConns)
}
if options.MinConns > options.MaxConns {
return nil, fmt.Errorf("pgdb: MinConns (%d) cannot exceed MaxConns (%d)", options.MinConns, options.MaxConns)
}
// Create pool config from PostgreSQL config
poolConfig, err := CreatePoolConfig(pgCfg)
if err != nil {
return nil, err
}
var poolConfig *pgxpool.Config
var err error
// Apply pool-specific settings
poolConfig.MinConns = options.MinConns
poolConfig.MaxConns = options.MaxConns
poolConfig.MaxConnLifetime = options.MaxConnLifetime
poolConfig.MaxConnIdleTime = options.MaxConnIdleTime
poolConfig.HealthCheckPeriod = options.HealthCheckPeriod
// Check DATABASE_URI environment variable first (highest priority)
if uri := os.Getenv("DATABASE_URI"); uri != "" {
poolConfig, err = pgxpool.ParseConfig(uri)
if err != nil {
return nil, fmt.Errorf("failed to parse DATABASE_URI: %w", err)
}
// Log when PoolOptions differ from defaults (they will be ignored)
defaults := DefaultPoolOptions()
if options.MaxConns != defaults.MaxConns || options.MinConns != defaults.MinConns {
log.WarnContext(ctx, "DATABASE_URI is set; PoolOptions are ignored (use URI query parameters for pool settings)",
"ignored_max_conns", options.MaxConns,
"ignored_min_conns", options.MinConns,
)
}
} else {
// Fall back to config file approach
pgCfg, _, err := FindConfig(options.ConfigFiles)
if err != nil {
return nil, err
}
poolConfig, err = CreatePoolConfig(pgCfg)
if err != nil {
return nil, err
}
// Apply pool-specific settings from PoolOptions (config files don't support these)
poolConfig.MinConns = options.MinConns
poolConfig.MaxConns = options.MaxConns
poolConfig.MaxConnLifetime = options.MaxConnLifetime
poolConfig.MaxConnIdleTime = options.MaxConnIdleTime
poolConfig.HealthCheckPeriod = options.HealthCheckPeriod
}
// Create the pool
pool, err := pgxpool.NewWithConfig(ctx, poolConfig)
@@ -79,7 +122,7 @@ func OpenPool(ctx context.Context, options PoolOptions) (*pgxpool.Pool, error) {
// Test the connection
if err := pool.Ping(ctx); err != nil {
pool.Close()
pool.Close() // pgxpool.Pool.Close() doesn't return an error
return nil, fmt.Errorf("failed to ping database: %w", err)
}
@@ -94,39 +137,54 @@ func OpenPoolWithConfigFile(ctx context.Context, configFile string) (*pgxpool.Po
return OpenPool(ctx, options)
}
// findAndParseConfig searches for and parses the first existing config file
func findAndParseConfig(configFiles []string) (*database.PostgresConfig, error) {
var firstErr error
// FindConfig searches for and parses the first existing config file.
// Returns the PostgresConfig, the path to the config file used, and any error.
// If DATABASE_URI env var is set, returns nil config with empty path (use ParseURIConfig instead).
func FindConfig(configFiles []string) (*database.PostgresConfig, string, error) {
// Check if DATABASE_URI takes precedence
if os.Getenv("DATABASE_URI") != "" {
return nil, "", nil
}
var errs []error
var triedFiles []string
for _, configFile := range configFiles {
if configFile == "" {
continue
}
triedFiles = append(triedFiles, configFile)
// Check if file exists
if _, err := os.Stat(configFile); err != nil {
if firstErr == nil {
firstErr = err
}
errs = append(errs, fmt.Errorf("%s: %w", configFile, err))
continue
}
// Try to read and parse the file
pgCfg, err := parseConfigFile(configFile)
if err != nil {
if firstErr == nil {
firstErr = err
}
errs = append(errs, fmt.Errorf("%s: %w", configFile, err))
continue
}
return pgCfg, nil
return pgCfg, configFile, nil
}
if firstErr != nil {
return nil, fmt.Errorf("no config file found: %w", firstErr)
if len(errs) > 0 {
return nil, "", fmt.Errorf("no valid config file found (tried: %v): %w", triedFiles, errors.Join(errs...))
}
return nil, fmt.Errorf("no valid config files provided")
return nil, "", fmt.Errorf("no valid config files provided")
}
// ParseURIConfig extracts connection info from DATABASE_URI environment variable.
// Returns nil if DATABASE_URI is not set.
func ParseURIConfig() (*pgxpool.Config, error) {
uri := os.Getenv("DATABASE_URI")
if uri == "" {
return nil, nil
}
return pgxpool.ParseConfig(uri)
}
// parseConfigFile reads and parses a YAML config file
@@ -135,7 +193,7 @@ func parseConfigFile(configFile string) (*database.PostgresConfig, error) {
if err != nil {
return nil, fmt.Errorf("failed to open config file: %w", err)
}
defer file.Close()
defer func() { _ = file.Close() }()
dec := yaml.NewDecoder(file)
cfg := database.Config{}
@@ -164,8 +222,9 @@ func parseConfigFile(configFile string) (*database.PostgresConfig, error) {
return nil, fmt.Errorf("no PostgreSQL configuration found in %s", configFile)
}
// getConfigFiles returns the list of config files to search
func getConfigFiles() []string {
// GetConfigFiles returns the list of config files to search for database configuration.
// Checks DATABASE_CONFIG_FILE env var first, otherwise returns default paths.
func GetConfigFiles() []string {
if configFile := os.Getenv("DATABASE_CONFIG_FILE"); configFile != "" {
return []string{configFile}
}

View File

@@ -1,6 +1,9 @@
package pgdb
import (
"context"
"os"
"strings"
"testing"
"time"
@@ -149,3 +152,63 @@ func TestCreatePoolConfigDefaults(t *testing.T) {
t.Errorf("Expected default Port=5432, got %d", poolCfg.ConnConfig.Port)
}
}
func TestOpenPoolWithDatabaseURI(t *testing.T) {
if testing.Short() {
t.Skip("skipping integration test")
}
// This test requires a running PostgreSQL instance
uri := os.Getenv("TEST_DATABASE_URI")
if uri == "" {
t.Skip("TEST_DATABASE_URI not set")
}
ctx := context.Background()
t.Setenv("DATABASE_URI", uri)
pool, err := OpenPool(ctx, DefaultPoolOptions())
if err != nil {
t.Fatalf("OpenPool failed: %v", err)
}
defer pool.Close()
// Verify connection works
var result int
err = pool.QueryRow(ctx, "SELECT 1").Scan(&result)
if err != nil {
t.Fatalf("query failed: %v", err)
}
if result != 1 {
t.Errorf("expected 1, got %d", result)
}
}
func TestDatabaseURIPrecedence(t *testing.T) {
// Test that DATABASE_URI takes precedence over config files
// We use localhost with a port that's unlikely to have postgres running
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
// Set DATABASE_URI to a parseable URI pointing to a non-listening port
t.Setenv("DATABASE_URI", "postgres://testuser:testpass@127.0.0.1:59999/testdb?connect_timeout=1")
// Set config files to a nonexistent path - if this were used, we'd get
// "config file not found" error instead of connection refused
opts := DefaultPoolOptions()
opts.ConfigFiles = []string{"/nonexistent/path/database.yaml"}
_, err := OpenPool(ctx, opts)
// Should fail with connection error (not config file error)
// This proves DATABASE_URI was used instead of config files
if err == nil {
t.Fatal("expected error, got nil")
}
// The error should be about connection failure, not about missing config file
errStr := err.Error()
if strings.Contains(errStr, "config file") || strings.Contains(errStr, "no such file") {
t.Errorf("expected connection error, got config file error: %v", err)
}
}

View File

@@ -51,9 +51,9 @@ func OpenDBWithConfigFile(ctx context.Context, configFile string) (*sql.DB, erro
// OpenDBMonitor opens a database connection with monitor-specific defaults
// This is a convenience function for Monitor package compatibility
func OpenDBMonitor() (*sql.DB, error) {
func OpenDBMonitor(ctx context.Context) (*sql.DB, error) {
options := MonitorConfigOptions()
return OpenDB(context.Background(), options)
return OpenDB(ctx, options)
}
// findConfigFile searches for the first existing config file from the list

View File

@@ -3,6 +3,24 @@
// This package implements a dedicated metrics server that exposes application metrics
// via HTTP. It uses a custom Prometheus registry to avoid conflicts with other metric
// collectors and provides graceful shutdown capabilities.
//
// # Usage
//
// Create a new metrics server and register your metrics with its Registry():
//
// m := metricsserver.New()
// myCounter := prometheus.NewCounter(...)
// m.Registry().MustRegister(myCounter)
//
// When you need a Gatherer (for example, to pass to other libraries), use the Gatherer() method
// instead of prometheus.DefaultGatherer to ensure your custom metrics are collected:
//
// gatherer := m.Gatherer() // Returns the custom registry as a Gatherer
//
// To use the default Prometheus gatherer alongside your custom registry:
//
// m := metricsserver.NewWithDefaultGatherer()
// m.Gatherer() // Returns prometheus.DefaultGatherer
package metricsserver
import (
@@ -21,15 +39,32 @@ import (
// Metrics provides a custom Prometheus registry and HTTP handlers for metrics exposure.
// It isolates application metrics from the default global registry.
type Metrics struct {
r *prometheus.Registry
r *prometheus.Registry
useDefaultGatherer bool
}
// New creates a new Metrics instance with a custom Prometheus registry.
// Use this when you want isolated metrics that don't interfere with the default registry.
func New() *Metrics {
r := prometheus.NewRegistry()
m := &Metrics{
r: r,
r: r,
useDefaultGatherer: false,
}
return m
}
// NewWithDefaultGatherer creates a new Metrics instance that uses the default Prometheus gatherer.
// This is useful when you want to expose metrics from the default registry alongside your custom metrics.
// The custom registry will still be available via Registry() but Gatherer() will return prometheus.DefaultGatherer.
func NewWithDefaultGatherer() *Metrics {
r := prometheus.NewRegistry()
m := &Metrics{
r: r,
useDefaultGatherer: true,
}
return m
@@ -41,6 +76,23 @@ func (m *Metrics) Registry() *prometheus.Registry {
return m.r
}
// Gatherer returns the Prometheus gatherer to use for collecting metrics.
// This returns the custom registry's Gatherer by default, ensuring your registered
// metrics are collected. If the instance was created with NewWithDefaultGatherer(),
// this returns prometheus.DefaultGatherer instead.
//
// Use this method when you need a prometheus.Gatherer interface, for example when
// configuring other libraries that need to collect metrics.
//
// IMPORTANT: Do not use prometheus.DefaultGatherer directly if you want to collect
// metrics registered with this instance's Registry(). Always use this Gatherer() method.
func (m *Metrics) Gatherer() prometheus.Gatherer {
if m.useDefaultGatherer {
return prometheus.DefaultGatherer
}
return m.r
}
// Handler returns an HTTP handler for the /metrics endpoint with OpenMetrics support.
func (m *Metrics) Handler() http.Handler {
log := logger.NewStdLog("prom http", false, nil)

View File

@@ -67,6 +67,47 @@ func TestRegistry(t *testing.T) {
}
}
func TestGatherer(t *testing.T) {
metrics := New()
gatherer := metrics.Gatherer()
if gatherer == nil {
t.Fatal("Gatherer() returned nil")
}
// Register a test metric
counter := prometheus.NewCounter(prometheus.CounterOpts{
Name: "test_gatherer_counter",
Help: "A test counter for gatherer",
})
metrics.Registry().MustRegister(counter)
counter.Inc()
// Test that the gatherer collects our custom metric
metricFamilies, err := gatherer.Gather()
if err != nil {
t.Errorf("failed to gather metrics: %v", err)
}
found := false
for _, mf := range metricFamilies {
if mf.GetName() == "test_gatherer_counter" {
found = true
break
}
}
if !found {
t.Error("registered metric not found via Gatherer()")
}
// Verify gatherer is the same as registry
if gatherer != metrics.r {
t.Error("Gatherer() should return the same object as the registry for custom registry mode")
}
}
func TestHandler(t *testing.T) {
metrics := New()
@@ -212,6 +253,45 @@ func TestListenAndServeContextCancellation(t *testing.T) {
}
}
func TestNewWithDefaultGatherer(t *testing.T) {
metrics := NewWithDefaultGatherer()
if metrics == nil {
t.Fatal("NewWithDefaultGatherer returned nil")
}
if !metrics.useDefaultGatherer {
t.Error("useDefaultGatherer should be true")
}
gatherer := metrics.Gatherer()
if gatherer == nil {
t.Fatal("Gatherer() returned nil")
}
// Verify it returns the default gatherer
if gatherer != prometheus.DefaultGatherer {
t.Error("Gatherer() should return prometheus.DefaultGatherer when useDefaultGatherer is true")
}
// Verify the custom registry is still available and separate
if metrics.Registry() == nil {
t.Error("Registry() should still return a custom registry")
}
// Test that registering in custom registry doesn't affect default gatherer check
counter := prometheus.NewCounter(prometheus.CounterOpts{
Name: "test_default_gatherer_counter",
Help: "A test counter",
})
metrics.Registry().MustRegister(counter)
// The gatherer should still be the default one, not our custom registry
if metrics.Gatherer() != prometheus.DefaultGatherer {
t.Error("Gatherer() should continue to return prometheus.DefaultGatherer")
}
}
// Benchmark the metrics handler response time
func BenchmarkMetricsHandler(b *testing.B) {
metrics := New()