From 97cf896a17b6acbb583670c9a22d30125bbac310 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Tue, 4 Nov 2025 00:32:53 +0000 Subject: [PATCH] Add pool management. --- README.md | 61 +++++++++++++ http.go | 14 ++- main.go | 37 +++++--- pool.go | 251 +++++++++++++++++++++++++++++++++++++++++++++++++++ pool_test.go | 135 +++++++++++++++++++++++++++ structs.go | 29 +++--- 6 files changed, 500 insertions(+), 27 deletions(-) create mode 100644 pool.go create mode 100644 pool_test.go diff --git a/README.md b/README.md index 96eb5fc..1da2d95 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,10 @@ Therefore, I present you the simple client to which you can copy & paste your gr * `GRAPHQL_RETRIES_NUMBER` - Number of retries: Default: `3` * `GRAPHQL_RETRIES_DELAY` - Delay in retries in milliseconds. Default: `250` * `GRAPHQL_RETRIES_PATTERNS` - Comma-separated error patterns that trigger retries. Default: `postgres,connection,timeout,transaction,could not,temporarily unavailable,deadlock` +* `GRAPHQL_POOL_WARMUP_ENABLED` - Enable intelligent connection pool warmup. Default: `false` +* `GRAPHQL_POOL_SIZE` - Number of connections to pre-create and maintain. Default: `5` +* `GRAPHQL_POOL_WARMUP_QUERY` - Query used for warming up connections. Default: `query{__typename}` +* `GRAPHQL_POOL_HEALTH_INTERVAL` - Interval in seconds for pool health checks. Default: `30` ### Modifiers on the fly @@ -85,6 +89,63 @@ export GRAPHQL_RETRIES_DELAY=250 This will retry GraphQL queries up to 3 times when postgres transaction errors occur, with exponential backoff starting at 250ms. +### Intelligent Connection Pooling + +The library supports intelligent connection pooling with pre-warmed connections and automatic health monitoring. This significantly improves performance during traffic spikes by keeping hot standby connections ready. + +**How it works:** +- **Pre-warming**: Connections are created and warmed up during initialization +- **Health monitoring**: Background process periodically checks connection health +- **Auto-refresh**: Automatically replaces stale or broken connections +- **Hot standbys**: Ready connections eliminate cold-start latency + +**Configuration:** + +Enable connection pool warmup for production environments with variable traffic: + +```bash +export GRAPHQL_POOL_WARMUP_ENABLED=true +export GRAPHQL_POOL_SIZE=5 # Number of hot standby connections +export GRAPHQL_POOL_HEALTH_INTERVAL=30 # Health check every 30 seconds +export GRAPHQL_POOL_WARMUP_QUERY="query{__typename}" # Lightweight health check query +``` + +**Benefits:** +- ⚡ **Instant response** - Connections already established during traffic spikes +- 🔄 **Auto-healing** - Detects and replaces broken connections automatically +- 📊 **Predictable performance** - Eliminates connection establishment latency +- 🛡️ **Resilience** - Maintains pool health even during network instability + +**Recommended settings by use case:** + +```bash +# High-traffic API (e.g., production web app) +GRAPHQL_POOL_WARMUP_ENABLED=true +GRAPHQL_POOL_SIZE=10 +GRAPHQL_POOL_HEALTH_INTERVAL=15 + +# Medium-traffic service (e.g., background jobs) +GRAPHQL_POOL_WARMUP_ENABLED=true +GRAPHQL_POOL_SIZE=5 +GRAPHQL_POOL_HEALTH_INTERVAL=30 + +# Low-traffic/development +GRAPHQL_POOL_WARMUP_ENABLED=false # Default: on-demand connections +``` + +**Graceful shutdown:** + +When shutting down your application, stop the pool monitor gracefully: + +```go +func main() { + gql := graphql.NewConnection() + defer gql.StopPoolMonitor() // Ensure clean shutdown + + // Your application logic +} +``` + ### Cache You have two options to enable the cache: diff --git a/http.go b/http.go index 45fdb12..492a9db 100644 --- a/http.go +++ b/http.go @@ -24,6 +24,9 @@ func (b *BaseClient) createHttpClient() (http_client *http.Client) { DisableCompression: true, // Disable automatic compression to prevent trailing garbage errors WriteBufferSize: 4096, ReadBufferSize: 4096, + // Additional settings for connection health + ExpectContinueTimeout: 1 * time.Second, // Timeout for 100-Continue + TLSHandshakeTimeout: 10 * time.Second, // Timeout for TLS handshake } http_client = &http.Client{ @@ -47,10 +50,13 @@ func (b *BaseClient) createHttpClient() (http_client *http.Client) { http2Transport := &http2.Transport{ AllowHTTP: true, TLSClientConfig: tlsClientConfig, - ReadIdleTimeout: 30 * time.Second, - PingTimeout: 10 * time.Second, - WriteByteTimeout: 10 * time.Second, - DisableCompression: true, // Disable automatic compression to prevent trailing garbage errors + ReadIdleTimeout: 30 * time.Second, // Close idle connections after 30s + PingTimeout: 10 * time.Second, // Detect dead connections with PING + WriteByteTimeout: 10 * time.Second, // Timeout for write operations + DisableCompression: true, // Disable automatic compression to prevent trailing garbage errors + // StrictMaxConcurrentStreams forces the HTTP/2 connection to wait for SETTINGS frame + // This helps with connection health detection + StrictMaxConcurrentStreams: true, } http_client = &http.Client{ diff --git a/main.go b/main.go index 14c465f..69148a1 100644 --- a/main.go +++ b/main.go @@ -47,24 +47,39 @@ func NewConnection() (b *BaseClient) { }) b = &BaseClient{ - endpoint: envutil.Getenv("GRAPHQL_ENDPOINT", "https://api.github.com/graphql"), - responseType: envutil.Getenv("GRAPHQL_OUTPUT", "string"), - Logger: logger, - cache: cache.New(time.Duration(envutil.GetInt("GRAPHQL_CACHE_TTL", 5)) * time.Second), - cache_global: envutil.GetBool("GRAPHQL_CACHE_ENABLED", false), - retries_enable: envutil.GetBool("GRAPHQL_RETRIES_ENABLE", false), - retries_delay: time.Duration(envutil.GetInt("GRAPHQL_RETRIES_DELAY", 250) * int(time.Millisecond)), - retries_number: envutil.GetInt("GRAPHQL_RETRIES_NUMBER", 3), - retries_patterns: parseRetryPatterns(envutil.Getenv("GRAPHQL_RETRIES_PATTERNS", "postgres,connection,timeout,transaction,could not,temporarily unavailable,deadlock")), - minify_queries: envutil.GetBool("GRAPHQL_MINIFY_QUERIES", true), // Default: enabled for production efficiency + endpoint: envutil.Getenv("GRAPHQL_ENDPOINT", "https://api.github.com/graphql"), + responseType: envutil.Getenv("GRAPHQL_OUTPUT", "string"), + Logger: logger, + cache: cache.New(time.Duration(envutil.GetInt("GRAPHQL_CACHE_TTL", 5)) * time.Second), + cache_global: envutil.GetBool("GRAPHQL_CACHE_ENABLED", false), + retries_enable: envutil.GetBool("GRAPHQL_RETRIES_ENABLE", false), + retries_delay: time.Duration(envutil.GetInt("GRAPHQL_RETRIES_DELAY", 250) * int(time.Millisecond)), + retries_number: envutil.GetInt("GRAPHQL_RETRIES_NUMBER", 3), + retries_patterns: parseRetryPatterns(envutil.Getenv("GRAPHQL_RETRIES_PATTERNS", "postgres,connection,timeout,transaction,could not,temporarily unavailable,deadlock")), + minify_queries: envutil.GetBool("GRAPHQL_MINIFY_QUERIES", true), // Default: enabled for production efficiency + pool_warmup_enabled: envutil.GetBool("GRAPHQL_POOL_WARMUP_ENABLED", false), + pool_size: envutil.GetInt("GRAPHQL_POOL_SIZE", 5), + pool_warmup_query: envutil.Getenv("GRAPHQL_POOL_WARMUP_QUERY", "query{__typename}"), + pool_health_interval: time.Duration(envutil.GetInt("GRAPHQL_POOL_HEALTH_INTERVAL", 30)) * time.Second, + pool_stop: make(chan bool, 1), } b.client = b.createHttpClient() b.Logger.Debug(&logging.LogMessage{ Message: "Created new GraphQL client connection", Pairs: map[string]interface{}{ - "values": b, + "endpoint": b.endpoint, + "pool_warmup": b.pool_warmup_enabled, + "pool_size": b.pool_size, + "pool_health_check": b.pool_health_interval, }, }) + + // Initialize connection pool if warmup is enabled + if b.pool_warmup_enabled { + b.warmupConnectionPool() + b.startPoolHealthMonitor() + } + return b } diff --git a/pool.go b/pool.go new file mode 100644 index 0000000..66aca67 --- /dev/null +++ b/pool.go @@ -0,0 +1,251 @@ +package gql + +import ( + "context" + "sync" + "time" + + libpack_logger "github.com/lukaszraczylo/go-simple-graphql/logging" +) + +// warmupConnectionPool pre-creates connections by executing warmup queries +// This establishes connections in the pool before actual traffic arrives +func (b *BaseClient) warmupConnectionPool() { + b.Logger.Info(&libpack_logger.LogMessage{ + Message: "Warming up connection pool", + Pairs: map[string]interface{}{ + "pool_size": b.pool_size, + "warmup_query": b.pool_warmup_query, + "endpoint": b.endpoint, + }, + }) + + startTime := time.Now() + var wg sync.WaitGroup + successCount := 0 + failCount := 0 + var mu sync.Mutex + + // Pre-create connections concurrently + for i := 0; i < b.pool_size; i++ { + wg.Add(1) + go func(index int) { + defer wg.Done() + + // Execute warmup query to establish connection + _, err := b.Query(b.pool_warmup_query, nil, nil) + + mu.Lock() + if err != nil { + failCount++ + b.Logger.Warning(&libpack_logger.LogMessage{ + Message: "Failed to warm up connection", + Pairs: map[string]interface{}{ + "connection_index": index, + "error": err.Error(), + }, + }) + } else { + successCount++ + b.Logger.Debug(&libpack_logger.LogMessage{ + Message: "Connection warmed up successfully", + Pairs: map[string]interface{}{ + "connection_index": index, + }, + }) + } + mu.Unlock() + }(i) + } + + wg.Wait() + duration := time.Since(startTime) + + b.Logger.Info(&libpack_logger.LogMessage{ + Message: "Connection pool warmup completed", + Pairs: map[string]interface{}{ + "successful": successCount, + "failed": failCount, + "total": b.pool_size, + "duration_ms": duration.Milliseconds(), + }, + }) +} + +// startPoolHealthMonitor starts a background goroutine that periodically +// checks and maintains pool health by sending health check queries +func (b *BaseClient) startPoolHealthMonitor() { + b.Logger.Info(&libpack_logger.LogMessage{ + Message: "Starting connection pool health monitor", + Pairs: map[string]interface{}{ + "check_interval": b.pool_health_interval, + "pool_size": b.pool_size, + }, + }) + + go func() { + ticker := time.NewTicker(b.pool_health_interval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + b.performPoolHealthCheck() + case <-b.pool_stop: + b.Logger.Info(&libpack_logger.LogMessage{ + Message: "Stopping connection pool health monitor", + Pairs: nil, + }) + return + } + } + }() +} + +// performPoolHealthCheck executes health check queries to maintain pool connections +// This keeps connections alive and detects/replaces dead connections +func (b *BaseClient) performPoolHealthCheck() { + b.Logger.Debug(&libpack_logger.LogMessage{ + Message: "Performing pool health check", + Pairs: map[string]interface{}{ + "pool_size": b.pool_size, + }, + }) + + startTime := time.Now() + var wg sync.WaitGroup + healthyCount := 0 + unhealthyCount := 0 + var mu sync.Mutex + + // Check a subset of connections to ensure pool health + // We check pool_size/2 to balance between thoroughness and overhead + checkCount := max(1, b.pool_size/2) + + for i := 0; i < checkCount; i++ { + wg.Add(1) + go func(index int) { + defer wg.Done() + + // Create a context with timeout for health check + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + // Use a channel to handle timeout + done := make(chan error, 1) + go func() { + _, err := b.Query(b.pool_warmup_query, nil, nil) + done <- err + }() + + select { + case err := <-done: + mu.Lock() + if err != nil { + unhealthyCount++ + b.Logger.Warning(&libpack_logger.LogMessage{ + Message: "Pool health check failed", + Pairs: map[string]interface{}{ + "check_index": index, + "error": err.Error(), + }, + }) + } else { + healthyCount++ + } + mu.Unlock() + case <-ctx.Done(): + mu.Lock() + unhealthyCount++ + b.Logger.Warning(&libpack_logger.LogMessage{ + Message: "Pool health check timeout", + Pairs: map[string]interface{}{ + "check_index": index, + }, + }) + mu.Unlock() + } + }(i) + } + + wg.Wait() + duration := time.Since(startTime) + + b.Logger.Debug(&libpack_logger.LogMessage{ + Message: "Pool health check completed", + Pairs: map[string]interface{}{ + "healthy": healthyCount, + "unhealthy": unhealthyCount, + "checked": checkCount, + "duration_ms": duration.Milliseconds(), + }, + }) + + // If more than 50% of checks failed, trigger a pool refresh + if unhealthyCount > checkCount/2 { + b.Logger.Warning(&libpack_logger.LogMessage{ + Message: "Pool health degraded, triggering refresh", + Pairs: map[string]interface{}{ + "unhealthy_ratio": float64(unhealthyCount) / float64(checkCount), + }, + }) + b.refreshConnectionPool() + } +} + +// refreshConnectionPool attempts to refresh the connection pool +// by executing warmup queries to replace stale connections +func (b *BaseClient) refreshConnectionPool() { + b.Logger.Info(&libpack_logger.LogMessage{ + Message: "Refreshing connection pool", + Pairs: map[string]interface{}{ + "pool_size": b.pool_size, + }, + }) + + var wg sync.WaitGroup + refreshCount := max(1, b.pool_size/3) // Refresh 1/3 of pool + + for i := 0; i < refreshCount; i++ { + wg.Add(1) + go func(index int) { + defer wg.Done() + + _, err := b.Query(b.pool_warmup_query, nil, nil) + if err != nil { + b.Logger.Debug(&libpack_logger.LogMessage{ + Message: "Pool refresh connection failed", + Pairs: map[string]interface{}{ + "connection_index": index, + "error": err.Error(), + }, + }) + } else { + b.Logger.Debug(&libpack_logger.LogMessage{ + Message: "Pool connection refreshed", + Pairs: map[string]interface{}{ + "connection_index": index, + }, + }) + } + }(i) + } + + wg.Wait() +} + +// StopPoolMonitor gracefully stops the pool health monitor +// Call this when shutting down the application +func (b *BaseClient) StopPoolMonitor() { + if b.pool_warmup_enabled { + b.Logger.Info(&libpack_logger.LogMessage{ + Message: "Stopping pool monitor", + Pairs: nil, + }) + select { + case b.pool_stop <- true: + default: + // Channel already has a value or monitor already stopped + } + } +} diff --git a/pool_test.go b/pool_test.go new file mode 100644 index 0000000..09d5c0f --- /dev/null +++ b/pool_test.go @@ -0,0 +1,135 @@ +package gql + +import ( + "os" + "testing" + "time" +) + +func (suite *Tests) TestPoolWarmup() { + suite.T().Run("should create client with pool warmup disabled by default", func(t *testing.T) { + // Save original env vars + originalPoolEnabled := os.Getenv("GRAPHQL_POOL_WARMUP_ENABLED") + originalEndpoint := os.Getenv("GRAPHQL_ENDPOINT") + defer func() { + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", originalPoolEnabled) + os.Setenv("GRAPHQL_ENDPOINT", originalEndpoint) + }() + + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", "false") + os.Setenv("GRAPHQL_ENDPOINT", mockServer.URL) + + client := NewConnection() + defer client.StopPoolMonitor() + + assert.False(client.pool_warmup_enabled) + assert.Equal(5, client.pool_size) // Default size + }) + + suite.T().Run("should warmup pool when enabled", func(t *testing.T) { + originalPoolEnabled := os.Getenv("GRAPHQL_POOL_WARMUP_ENABLED") + originalPoolSize := os.Getenv("GRAPHQL_POOL_SIZE") + originalEndpoint := os.Getenv("GRAPHQL_ENDPOINT") + defer func() { + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", originalPoolEnabled) + os.Setenv("GRAPHQL_POOL_SIZE", originalPoolSize) + os.Setenv("GRAPHQL_ENDPOINT", originalEndpoint) + }() + + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", "true") + os.Setenv("GRAPHQL_POOL_SIZE", "3") + os.Setenv("GRAPHQL_ENDPOINT", mockServer.URL) + + startTime := time.Now() + client := NewConnection() + defer client.StopPoolMonitor() + + duration := time.Since(startTime) + + assert.True(client.pool_warmup_enabled) + assert.Equal(3, client.pool_size) + // Warmup should complete relatively quickly for 3 connections + assert.Less(duration.Seconds(), 5.0, "Pool warmup took too long") + }) + + suite.T().Run("should use custom warmup query", func(t *testing.T) { + originalPoolEnabled := os.Getenv("GRAPHQL_POOL_WARMUP_ENABLED") + originalQuery := os.Getenv("GRAPHQL_POOL_WARMUP_QUERY") + originalEndpoint := os.Getenv("GRAPHQL_ENDPOINT") + defer func() { + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", originalPoolEnabled) + os.Setenv("GRAPHQL_POOL_WARMUP_QUERY", originalQuery) + os.Setenv("GRAPHQL_ENDPOINT", originalEndpoint) + }() + + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", "false") // Don't actually warmup, just test configuration + os.Setenv("GRAPHQL_POOL_WARMUP_QUERY", "query{__schema{queryType{name}}}") + os.Setenv("GRAPHQL_ENDPOINT", mockServer.URL) + + client := NewConnection() + defer client.StopPoolMonitor() + + assert.Equal("query{__schema{queryType{name}}}", client.pool_warmup_query) + }) +} + +func (suite *Tests) TestPoolHealthMonitor() { + suite.T().Run("should stop pool monitor gracefully", func(t *testing.T) { + client := CreateTestClient() + client.pool_warmup_enabled = false + + // Should not panic + client.StopPoolMonitor() + + assert.True(true, "Handled stop gracefully when monitor not running") + }) + + suite.T().Run("should handle multiple stop calls", func(t *testing.T) { + originalPoolEnabled := os.Getenv("GRAPHQL_POOL_WARMUP_ENABLED") + originalPoolSize := os.Getenv("GRAPHQL_POOL_SIZE") + originalEndpoint := os.Getenv("GRAPHQL_ENDPOINT") + defer func() { + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", originalPoolEnabled) + os.Setenv("GRAPHQL_POOL_SIZE", originalPoolSize) + os.Setenv("GRAPHQL_ENDPOINT", originalEndpoint) + }() + + os.Setenv("GRAPHQL_POOL_WARMUP_ENABLED", "true") + os.Setenv("GRAPHQL_POOL_SIZE", "2") + os.Setenv("GRAPHQL_ENDPOINT", mockServer.URL) + + client := NewConnection() + + // Call stop multiple times - should not panic + client.StopPoolMonitor() + client.StopPoolMonitor() + client.StopPoolMonitor() + + assert.True(true, "Handled multiple stop calls gracefully") + }) +} + +func (suite *Tests) TestPoolHealthCheck() { + suite.T().Run("should perform health check without panic", func(t *testing.T) { + client := CreateTestClient() + client.pool_warmup_enabled = true + client.pool_size = 2 + client.pool_warmup_query = "query{__typename}" + + // Perform a health check - should not panic + client.performPoolHealthCheck() + + assert.True(true, "Health check completed") + }) + + suite.T().Run("should refresh pool without panic", func(t *testing.T) { + client := CreateTestClient() + client.pool_size = 2 + client.pool_warmup_query = "query{__typename}" + + // Perform pool refresh - should not panic + client.refreshConnectionPool() + + assert.True(true, "Pool refresh completed") + }) +} diff --git a/structs.go b/structs.go index b798626..6bec523 100644 --- a/structs.go +++ b/structs.go @@ -9,18 +9,23 @@ import ( ) type BaseClient struct { - cache *cache.Cache - Logger *logging.Logger - client *http.Client - endpoint string - responseType string - retries_delay time.Duration - retries_number int - retries_patterns []string // Error patterns that should trigger retries - MaxGoRoutines int - cache_global bool - retries_enable bool - minify_queries bool // Enable GraphQL query minification (default: true) + cache *cache.Cache + Logger *logging.Logger + client *http.Client + endpoint string + responseType string + retries_delay time.Duration + retries_number int + retries_patterns []string // Error patterns that should trigger retries + pool_size int // Number of connections to pre-warm + pool_warmup_enabled bool // Enable connection pool warmup + pool_warmup_query string // Query to use for warming up connections + pool_health_interval time.Duration // Interval for pool health checks + pool_stop chan bool // Channel to stop pool monitor + MaxGoRoutines int + cache_global bool + retries_enable bool + minify_queries bool // Enable GraphQL query minification (default: true) } type Query struct {