Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 66a72265ce | |||
| 94b718a925 | |||
| 283d3936f6 |
@@ -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{}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user