Compare commits

...
6 Commits
Author SHA1 Message Date
lukaszraczylo 489247c42a Double the timeout. 2025-11-18 18:26:29 +00:00
lukaszraczylo 6c55bccc20 fixup! Add pool management. 2025-11-18 18:00:21 +00:00
lukaszraczylo 97cf896a17 Add pool management. 2025-11-04 00:32:53 +00:00
lukaszraczylo 71a14721c5 fixup! Add support of h2c over cleartext http 2025-11-03 23:40:04 +00:00
lukaszraczylo cd16b92dca Add support of h2c over cleartext http 2025-11-03 23:33:28 +00:00
lukaszraczylo b314a25549 Fix retry patterns 2025-11-03 21:02:50 +00:00
10 changed files with 709 additions and 114 deletions
+104 -4
View File
@@ -24,9 +24,11 @@ Therefore, I present you the simple client to which you can copy & paste your gr
## Features
* Executing GraphQL queries as they are, without types declaration
* HTTP2 support!
* HTTP/2 support for HTTPS endpoints (automatic)
* HTTP/1.1 for HTTP endpoints (compatible with all servers)
* Automatic retry with configurable error patterns
* Support for additional headers
* Query cache
* Query cache with compression
## Usage example
@@ -37,15 +39,113 @@ Therefore, I present you the simple client to which you can copy & paste your gr
* `GRAPHQL_CACHE_TTL` - Cache TTL in seconds for SELECT type of queries. Default: `5`
* `GRAPHQL_OUTPUT` - Output format. Default: `string`, available: `byte`, `string`, `mapstring`
* `LOG_LEVEL` - Logging level. Default: `info` available: `debug`, `info`, `warn`, `error`
* `GRAPHQL_RETRIES_ENABLE` - Should retries be enabled? Default: `true`
* `GRAPHQL_RETRIES_NUMBER` - Number of retries: Default: `1`
* `GRAPHQL_RETRIES_ENABLE` - Should retries be enabled? Default: `false`
* `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
* `gql.SetEndpoint('your-endpoint-url')` - modifies endpoint, without the need to set the environment variable
* `gql.SetOutput('byte')` - modifies output format, without the need to set the environment variable
### Retries
The library supports automatic retries for both HTTP-level errors and GraphQL errors that match configurable patterns.
**Retry behavior:**
- **HTTP errors** (status codes outside 200-299): Always retried if retries are enabled
- **GraphQL errors**: Only retried if the error message contains one of the configured patterns
- **Non-matching errors**: Fail immediately without retrying (e.g., validation errors, field not found)
**Configuring retryable patterns:**
By default, errors containing these patterns will trigger retries:
- `postgres` - Database connection/transaction errors
- `connection` - Network connection issues
- `timeout` - Request timeout errors
- `transaction` - Database transaction conflicts
- `could not` - Generic transient failures
- `temporarily unavailable` - Service unavailability
- `deadlock` - Database deadlocks
You can customize these patterns using the `GRAPHQL_RETRIES_PATTERNS` environment variable:
```bash
export GRAPHQL_RETRIES_PATTERNS="postgres,connection,timeout,custom-error-pattern"
```
**Example:** To handle flaky Hasura database connections, enable retries:
```bash
export GRAPHQL_RETRIES_ENABLE=true
export GRAPHQL_RETRIES_NUMBER=3
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:
+86 -39
View File
@@ -3,11 +3,11 @@ package gql
import (
"bytes"
"compress/gzip"
"crypto/tls"
"errors"
"fmt"
"io"
"net/http"
"strings"
"time"
"unicode/utf8"
@@ -32,23 +32,49 @@ func max(a, b int) int {
return b
}
var (
// Shared HTTP transport with optimized settings
defaultTransport = &http.Transport{
MaxIdleConns: 100,
MaxIdleConnsPerHost: 100,
IdleConnTimeout: 90 * time.Second,
DisableCompression: true, // Disable automatic request compression to prevent "trailing garbage" errors
ForceAttemptHTTP2: true,
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, // Skip TLS verification for test environments
// isRetryableError checks if GraphQL errors contain patterns that indicate a transient failure
// that should be retried (e.g., database connection issues, timeouts, transaction conflicts)
func isRetryableError(errors []struct {
Message interface{} `json:"message"`
}, patterns []string) bool {
if len(errors) == 0 || len(patterns) == 0 {
return false
}
// Shared HTTP client with timeouts
defaultClient = &http.Client{
Transport: defaultTransport,
Timeout: 30 * time.Second,
for _, err := range errors {
// Try to extract error message as string
var msgStr string
switch msg := err.Message.(type) {
case string:
msgStr = msg
case map[string]interface{}:
// Handle nested error structures (e.g., {"message": "...", "extensions": {...}})
if nestedMsg, ok := msg["message"].(string); ok {
msgStr = nestedMsg
}
default:
// If we can't convert to string, try fmt.Sprint as fallback
msgStr = fmt.Sprint(msg)
}
if msgStr == "" {
continue
}
// Check if error message contains any of the configured retry patterns (case-insensitive)
msgLower := strings.ToLower(msgStr)
for _, pattern := range patterns {
if strings.Contains(msgLower, strings.ToLower(pattern)) {
return true
}
}
}
)
return false
}
// Note: defaultClient removed - all clients should be created via NewConnection()
// which properly configures HTTP/2 transport with correct settings
func (qe *QueryExecutor) executeQuery() ([]byte, error) {
// Reuse buffer from pool to avoid allocations
@@ -131,12 +157,16 @@ func (qe *QueryExecutor) executeQuery() ([]byte, error) {
},
})
// Use default client if custom client is not set
client := qe.client
if client == nil {
client = defaultClient
// Client should always be set via NewConnection(), but check for safety
if qe.client == nil {
qe.Logger.Error(&libpack_logger.LogMessage{
Message: "HTTP client not initialized - use NewConnection() to create client",
Pairs: nil,
})
return fmt.Errorf("HTTP client not initialized")
}
httpResponse, err := client.Do(httpRequest)
httpResponse, err := qe.client.Do(httpRequest)
if err != nil {
return err
}
@@ -331,6 +361,38 @@ func (qe *QueryExecutor) executeQuery() ([]byte, error) {
return fmt.Errorf("error unmarshalling HTTP response: %w", err)
}
// Check for GraphQL errors in the response
if len(queryResult.Errors) > 0 {
// Determine if this is a retryable error based on configured patterns
if isRetryableError(queryResult.Errors, qe.retries_patterns) {
// Log as warning since we'll retry
qe.Logger.Warning(&libpack_logger.LogMessage{
Message: "Retryable GraphQL error detected",
Pairs: map[string]interface{}{"errors": queryResult.Errors},
})
// Return error to trigger retry mechanism
return fmt.Errorf("retryable error executing query: %s", queryResult.Errors)
}
// Non-retryable error - log and fail immediately without retrying
qe.Logger.Error(&libpack_logger.LogMessage{
Message: "Non-retryable GraphQL error",
Pairs: map[string]interface{}{"errors": queryResult.Errors},
})
// Use retry.Unrecoverable to skip remaining retry attempts
return retry.Unrecoverable(fmt.Errorf("error executing query: %s", queryResult.Errors))
}
// Check for null data in the response
if queryResult.Data == nil {
qe.Logger.Error(&libpack_logger.LogMessage{
Message: "GraphQL query returned no data",
Pairs: map[string]interface{}{"error": "data field is null"},
})
// Null data is typically not retryable (indicates query structure issue)
return retry.Unrecoverable(errors.New("error executing query: no data"))
}
return nil
},
retry.OnRetry(func(n uint, err error) {
@@ -346,33 +408,18 @@ func (qe *QueryExecutor) executeQuery() ([]byte, error) {
retry.LastErrorOnly(true),
)
if err != nil {
qe.Logger.Debug(&libpack_logger.LogMessage{
Message: "Error executing HTTP request",
qe.Logger.Error(&libpack_logger.LogMessage{
Message: "Query execution failed after retries",
Pairs: map[string]interface{}{"error": err.Error()},
})
return nil, err
}
if len(queryResult.Errors) > 0 {
qe.Logger.Debug(&libpack_logger.LogMessage{
Message: "Error executing query",
Pairs: map[string]interface{}{"error": queryResult.Errors},
})
return nil, fmt.Errorf("error executing query: %s", queryResult.Errors)
}
if queryResult.Data == nil {
qe.Logger.Debug(&libpack_logger.LogMessage{
Message: "Error executing query",
Pairs: map[string]interface{}{"error": "no data"},
})
return nil, errors.New("error executing query: no data")
}
// At this point, we have successfully executed the query and validated the response
// Marshal queryResult.Data
jsonData, err := json.Marshal(queryResult.Data)
if err != nil {
qe.Logger.Debug(&libpack_logger.LogMessage{
qe.Logger.Error(&libpack_logger.LogMessage{
Message: "Error marshalling query result",
Pairs: map[string]interface{}{"error": err.Error(), "data": queryResult.Data},
})
+5 -4
View File
@@ -185,7 +185,7 @@ func (suite *Tests) TestQueryExecutor_executeQuery() {
assert.Equal(3, attempts) // Should have retried twice before succeeding
})
suite.T().Run("should use default client when none provided", func(t *testing.T) {
suite.T().Run("should return error when client not initialized", func(t *testing.T) {
client := CreateTestClient()
qe := &QueryExecutor{
@@ -196,12 +196,13 @@ func (suite *Tests) TestQueryExecutor_executeQuery() {
Retries: false,
}
qe.endpoint = mockServer.URL
qe.client = nil // No client provided
qe.client = nil // No client provided (should always use NewConnection())
qe.cache = client.cache
result, err := qe.executeQuery()
assert.NoError(err)
assert.NotNil(result)
assert.Error(err)
assert.Nil(result)
assert.Contains(err.Error(), "HTTP client not initialized")
})
suite.T().Run("should set default Content-Type header", func(t *testing.T) {
+37 -25
View File
@@ -11,60 +11,72 @@ import (
)
func (b *BaseClient) createHttpClient() (http_client *http.Client) {
// Optimized connection pool settings for better performance
httpTransport := &http.Transport{
MaxIdleConns: 100, // Reduced from 512
MaxConnsPerHost: 50, // Reduced from 512
MaxIdleConnsPerHost: 10, // Reduced from 512
IdleConnTimeout: 30 * time.Second, // Increased for better reuse
ResponseHeaderTimeout: 10 * time.Second, // Reduced for faster timeouts
DisableKeepAlives: false,
DisableCompression: true, // Disable automatic request compression to prevent "trailing garbage" errors
WriteBufferSize: 4096, // Optimize buffer size
ReadBufferSize: 4096, // Optimize buffer size
}
if strings.HasPrefix(b.endpoint, "http://") {
// Use HTTP/1.1 for http:// endpoints
// Note: h2c (HTTP/2 Cleartext) is rarely supported by servers
httpTransport := &http.Transport{
MaxIdleConns: 100,
MaxConnsPerHost: 50,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 60 * time.Second,
ResponseHeaderTimeout: 20 * time.Second,
DisableKeepAlives: false,
DisableCompression: true, // Disable automatic compression to prevent trailing garbage errors
WriteBufferSize: 4096,
ReadBufferSize: 4096,
// Additional settings for connection health
ExpectContinueTimeout: 2 * time.Second, // Timeout for 100-Continue
TLSHandshakeTimeout: 20 * time.Second, // Timeout for TLS handshake
}
http_client = &http.Client{
Timeout: 30 * time.Second, // Increased for better reliability
Timeout: 60 * time.Second,
Transport: httpTransport,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
},
}
b.Logger.Debug(&libpack_logger.LogMessage{
Message: "Using HTTP/1.1 over http",
Message: "Using HTTP/1.1 for http:// endpoint",
Pairs: nil,
})
} else if strings.HasPrefix(b.endpoint, "https://") {
tlsClientConfig := &tls.Config{}
if strings.HasPrefix(b.endpoint, "https://") {
tlsClientConfig.InsecureSkipVerify = true
// Use HTTP/2 for https:// endpoints
tlsClientConfig := &tls.Config{
InsecureSkipVerify: true, // TODO: Make this configurable via environment variable
}
http2Transport := &http2.Transport{
AllowHTTP: true,
TLSClientConfig: tlsClientConfig,
ReadIdleTimeout: 30 * time.Second, // Increased for better reuse
PingTimeout: 10 * time.Second, // Reduced for faster detection
WriteByteTimeout: 10 * time.Second, // Add write timeout
DisableCompression: true, // Disable automatic request compression to prevent "trailing garbage" errors
ReadIdleTimeout: 60 * time.Second, // Close idle connections after 60s
PingTimeout: 20 * time.Second, // Detect dead connections with PING
WriteByteTimeout: 20 * 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{
Timeout: 30 * time.Second, // Increased for better reliability
Timeout: 60 * time.Second,
Transport: http2Transport,
CheckRedirect: func(req *http.Request, via []*http.Request) error {
return http.ErrUseLastResponse
},
}
b.Logger.Debug(&libpack_logger.LogMessage{
Message: "Using HTTP/2 over https",
Message: "Using HTTP/2 for https:// endpoint",
Pairs: nil,
})
} else {
b.Logger.Critical(&libpack_logger.LogMessage{
Message: "Invalid endpoint - neither http or https",
Message: "Invalid endpoint - must start with http:// or https://",
})
return nil
}
return http_client
}
+17 -15
View File
@@ -8,6 +8,8 @@ import (
"golang.org/x/net/http2"
)
// Note: http.Transport import is kept for http.ErrUseLastResponse usage in redirect tests
func (suite *Tests) TestBaseClient_createHttpClient() {
suite.T().Run("should create HTTP/1.1 client for http endpoints", func(t *testing.T) {
client := NewConnection()
@@ -15,17 +17,17 @@ func (suite *Tests) TestBaseClient_createHttpClient() {
httpClient := client.createHttpClient()
assert.NotNil(httpClient)
assert.Equal(30*time.Second, httpClient.Timeout)
assert.Equal(60*time.Second, httpClient.Timeout)
assert.NotNil(httpClient.Transport)
// Check that it's using regular HTTP transport
// Check that it's using HTTP/1.1 transport (h2c is rarely supported by servers)
transport, ok := httpClient.Transport.(*http.Transport)
assert.True(ok)
assert.Equal(100, transport.MaxIdleConns)
assert.Equal(50, transport.MaxConnsPerHost)
assert.Equal(10, transport.MaxIdleConnsPerHost)
assert.Equal(30*time.Second, transport.IdleConnTimeout)
assert.Equal(10*time.Second, transport.ResponseHeaderTimeout)
assert.Equal(60*time.Second, transport.IdleConnTimeout)
assert.Equal(20*time.Second, transport.ResponseHeaderTimeout)
assert.False(transport.DisableKeepAlives)
assert.True(transport.DisableCompression) // Request compression disabled to prevent "trailing garbage" errors
assert.Equal(4096, transport.WriteBufferSize)
@@ -38,7 +40,7 @@ func (suite *Tests) TestBaseClient_createHttpClient() {
httpClient := client.createHttpClient()
assert.NotNil(httpClient)
assert.Equal(30*time.Second, httpClient.Timeout)
assert.Equal(60*time.Second, httpClient.Timeout)
assert.NotNil(httpClient.Transport)
// Check that it's using HTTP/2 transport
@@ -47,9 +49,9 @@ func (suite *Tests) TestBaseClient_createHttpClient() {
assert.True(transport.AllowHTTP)
assert.NotNil(transport.TLSClientConfig)
assert.True(transport.TLSClientConfig.InsecureSkipVerify)
assert.Equal(30*time.Second, transport.ReadIdleTimeout)
assert.Equal(10*time.Second, transport.PingTimeout)
assert.Equal(10*time.Second, transport.WriteByteTimeout)
assert.Equal(60*time.Second, transport.ReadIdleTimeout)
assert.Equal(20*time.Second, transport.PingTimeout)
assert.Equal(20*time.Second, transport.WriteByteTimeout)
})
suite.T().Run("should handle invalid endpoints", func(t *testing.T) {
@@ -92,19 +94,19 @@ func (suite *Tests) TestBaseClient_createHttpClient() {
}
func (suite *Tests) TestBaseClient_createHttpClient_transportSettings() {
suite.T().Run("should configure HTTP transport correctly", func(t *testing.T) {
suite.T().Run("should configure HTTP/1.1 transport correctly for http://", func(t *testing.T) {
client := NewConnection()
client.SetEndpoint("http://example.com/graphql")
httpClient := client.createHttpClient()
transport := httpClient.Transport.(*http.Transport)
// Verify all transport settings
// Verify all HTTP/1.1 transport settings
assert.Equal(100, transport.MaxIdleConns)
assert.Equal(50, transport.MaxConnsPerHost)
assert.Equal(10, transport.MaxIdleConnsPerHost)
assert.Equal(30*time.Second, transport.IdleConnTimeout)
assert.Equal(10*time.Second, transport.ResponseHeaderTimeout)
assert.Equal(60*time.Second, transport.IdleConnTimeout)
assert.Equal(20*time.Second, transport.ResponseHeaderTimeout)
assert.False(transport.DisableKeepAlives)
assert.True(transport.DisableCompression) // Request compression disabled to prevent "trailing garbage" errors
assert.Equal(4096, transport.WriteBufferSize)
@@ -122,9 +124,9 @@ func (suite *Tests) TestBaseClient_createHttpClient_transportSettings() {
assert.True(transport.AllowHTTP)
assert.NotNil(transport.TLSClientConfig)
assert.True(transport.TLSClientConfig.InsecureSkipVerify)
assert.Equal(30*time.Second, transport.ReadIdleTimeout)
assert.Equal(10*time.Second, transport.PingTimeout)
assert.Equal(10*time.Second, transport.WriteByteTimeout)
assert.Equal(60*time.Second, transport.ReadIdleTimeout)
assert.Equal(20*time.Second, transport.PingTimeout)
assert.Equal(20*time.Second, transport.WriteByteTimeout)
})
}
+43 -10
View File
@@ -2,6 +2,7 @@ package gql
import (
"net/http"
"strings"
"time"
"github.com/gookit/goutil/envutil"
@@ -9,6 +10,22 @@ import (
logging "github.com/lukaszraczylo/go-simple-graphql/logging"
)
// parseRetryPatterns parses a comma-separated string of retry patterns into a slice
func parseRetryPatterns(patterns string) []string {
if patterns == "" {
return []string{}
}
parts := strings.Split(patterns, ",")
result := make([]string, 0, len(parts))
for _, p := range parts {
trimmed := strings.TrimSpace(p)
if trimmed != "" {
result = append(result, trimmed)
}
}
return result
}
func NewConnection() (b *BaseClient) {
// Read LOG_LEVEL environment variable
logLevelStr := envutil.Getenv("LOG_LEVEL", "info")
@@ -30,23 +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),
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
}
+250
View File
@@ -0,0 +1,250 @@
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
// Uses sync.Once to ensure it only executes once, preventing race conditions
func (b *BaseClient) StopPoolMonitor() {
if b.pool_warmup_enabled {
b.pool_stop_once.Do(func() {
b.Logger.Info(&libpack_logger.LogMessage{
Message: "Stopping pool monitor",
Pairs: nil,
})
b.pool_stop <- true
})
}
}
+135
View File
@@ -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")
})
}
+13 -6
View File
@@ -11,6 +11,7 @@ import (
assertions "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/suite"
"golang.org/x/net/http2"
)
type RequestCompressionTestSuite struct {
@@ -74,17 +75,23 @@ func (suite *RequestCompressionTestSuite) TestRequestCompression() {
httpClient := client.createHttpClient()
assert.NotNil(httpClient, "HTTP client should be created")
// Verify transport settings
// Verify transport settings based on protocol
if strings.HasPrefix(endpoint, "http://") {
// HTTP/1.1 transport
// HTTP/1.1 transport for http:// (h2c rarely supported)
transport, ok := httpClient.Transport.(*http.Transport)
assert.True(ok, "Should be http.Transport for HTTP endpoints")
assert.True(ok, "Should be http.Transport for http:// endpoints")
if ok {
assert.True(transport.DisableCompression, "DisableCompression should be true for HTTP transport")
assert.True(transport.DisableCompression, "DisableCompression should be true to prevent trailing garbage")
}
} else {
// HTTP/2 transport - this is where the issue might be
// We need to ensure HTTP/2 transport also has compression disabled
// HTTP/2 transport for https://
transport, ok := httpClient.Transport.(*http2.Transport)
assert.True(ok, "Should be http2.Transport for https:// endpoints")
if ok {
assert.True(transport.DisableCompression, "DisableCompression should be true to prevent trailing garbage")
assert.True(transport.AllowHTTP, "AllowHTTP should be true")
assert.NotNil(transport.TLSClientConfig, "TLSClientConfig should be set for https:// endpoints")
}
}
qe := &QueryExecutor{
+19 -11
View File
@@ -2,6 +2,7 @@ package gql
import (
"net/http"
"sync"
"time"
cache "github.com/lukaszraczylo/go-simple-graphql/cache"
@@ -9,17 +10,24 @@ import (
)
type BaseClient struct {
cache *cache.Cache
Logger *logging.Logger
client *http.Client
endpoint string
responseType string
retries_delay time.Duration
retries_number int
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
pool_stop_once sync.Once // Ensures StopPoolMonitor is called only once
MaxGoRoutines int
cache_global bool
retries_enable bool
minify_queries bool // Enable GraphQL query minification (default: true)
}
type Query struct {