mirror of
https://github.com/lukaszraczylo/graphql-monitoring-proxy.git
synced 2026-10-24 06:48:09 +00:00
fix: restore tracing endpoint validation; honor per-side client timeouts; cache eviction target; redis close
- tracing: validateEndpoint used fmt.Sscanf("%s:%d") which treats %s as
greedy, so every valid host:port endpoint failed and traced==nil disabled
tracing entirely. Validate with net.SplitHostPort (pure helper, port range
0-65535); test TestValidateEndpoint.
- proxy: createFasthttpClient used clientTimeout for every timeout slot,
ignoring the parsed CLIENT_READ_TIMEOUT/CLIENT_WRITE_TIMEOUT (silent
config loss, documented separately in README). Honour per-side read/write
timeouts, falling back to client timeout when unset.
- cache/memory: evictToFreeMemory capped candidates at maxCacheSize/5 and
stopped at the subset end, so a large write could leave the cache over
its memory limit. Collect all entries and sort.Slice (correct + O(n log n)
vs O(cap^2)), evict until the target is met or the cache is drained.
- cache/redis: the Redis cache client was never closed at shutdown (unlike
the metrics aggregator). Add RedisConfig.Close, CacheWrapper.Close, a
cache.Shutdown and register it with the ShutdownManager.
This commit is contained in:
Vendored
+10
@@ -168,6 +168,16 @@ func EnableCache(cfg *CacheConfig) {
|
|||||||
config = cfg
|
config = cfg
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Shutdown closes the underlying cache client if it owns external resources
|
||||||
|
// (the Redis connection pool). No-op for in-memory caches.
|
||||||
|
func Shutdown() {
|
||||||
|
if config != nil && config.Client != nil {
|
||||||
|
if closer, ok := config.Client.(io.Closer); ok {
|
||||||
|
_ = closer.Close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func CacheLookup(hash string) []byte {
|
func CacheLookup(hash string) []byte {
|
||||||
if !IsCacheInitialized() {
|
if !IsCacheInitialized() {
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
Vendored
+29
@@ -183,3 +183,32 @@ func TestSetMaxMemorySize(t *testing.T) {
|
|||||||
func generateKey(index int) string {
|
func generateKey(index int) string {
|
||||||
return "test-key-" + fmt.Sprintf("%d", index)
|
return "test-key-" + fmt.Sprintf("%d", index)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestEvictToFreeMemoryWithSmallMaxEntriesReachesTarget is a regression for
|
||||||
|
// under-eviction: evictToFreeMemory used to cap its candidate set at
|
||||||
|
// maxCacheSize/5 and stop at the end of that subset. With a small
|
||||||
|
// maxCacheSize the cap (maxCacheSize/5) could be lower than the number of
|
||||||
|
// entries needed to satisfy bytesToFree, so the cache stayed over its
|
||||||
|
// memory limit.
|
||||||
|
func TestEvictToFreeMemoryWithSmallMaxEntriesReachesTarget(t *testing.T) {
|
||||||
|
// maxCacheSize=10 -> old candidate cap was maxCacheSize/5 = 2 entries.
|
||||||
|
cache := NewWithSize(5*time.Second, DefaultMaxMemorySize, 10)
|
||||||
|
val := make([]byte, 1024) // ~1KB + overhead each
|
||||||
|
const count = 8
|
||||||
|
for i := 0; i < count; i++ {
|
||||||
|
cache.Set(fmt.Sprintf("e%d", i), val, time.Second)
|
||||||
|
}
|
||||||
|
before := cache.GetMemoryUsage()
|
||||||
|
// Request freeing an amount equal to ~half the entries (well beyond the
|
||||||
|
// old 2-entry cap).
|
||||||
|
need := (count / 2) * (int64(len(val)) + approxEntryOverhead)
|
||||||
|
cache.evictToFreeMemory(need)
|
||||||
|
after := cache.GetMemoryUsage()
|
||||||
|
assert.LessOrEqual(t, after, before-need,
|
||||||
|
"evictToFreeMemory must free at least the requested bytes even across more than maxCacheSize/5 entries")
|
||||||
|
|
||||||
|
// Requesting more than the total must drain the cache.
|
||||||
|
cache.evictToFreeMemory(1 << 30)
|
||||||
|
assert.Equal(t, int64(0), cache.GetMemoryUsage(),
|
||||||
|
"an oversized request should evict every entry")
|
||||||
|
}
|
||||||
|
|||||||
Vendored
+17
-19
@@ -8,6 +8,7 @@ import (
|
|||||||
"compress/gzip"
|
"compress/gzip"
|
||||||
"context"
|
"context"
|
||||||
"io"
|
"io"
|
||||||
|
"sort"
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
@@ -322,32 +323,29 @@ func (c *Cache) evictToFreeMemory(bytesToFree int64) {
|
|||||||
memorySize int64
|
memorySize int64
|
||||||
}
|
}
|
||||||
|
|
||||||
// Collect entries to consider for eviction
|
// Collect ALL entries so eviction can actually reach the target. The
|
||||||
entries := make([]keyMemorySize, 0, int(c.maxCacheSize/5))
|
// previous code capped the candidate set at maxCacheSize/5 and stopped at
|
||||||
|
// the end of that subset, so a large write could leave the cache over its
|
||||||
|
// memory limit.
|
||||||
|
entries := make([]keyMemorySize, 0, int(c.maxCacheSize))
|
||||||
c.entries.Range(func(k, v any) bool {
|
c.entries.Range(func(k, v any) bool {
|
||||||
key := k.(string)
|
key := k.(string)
|
||||||
entry := v.(CacheEntry)
|
entry := v.(CacheEntry)
|
||||||
entries = append(entries, keyMemorySize{entry.ExpiresAt, key, entry.MemorySize})
|
entries = append(entries, keyMemorySize{entry.ExpiresAt, key, entry.MemorySize})
|
||||||
return len(entries) < cap(entries)
|
return true
|
||||||
})
|
})
|
||||||
|
|
||||||
// Sort entries by expiry time (oldest first)
|
// Sort by expiry time (oldest first)
|
||||||
// Simple selection sort since we only need to find the oldest entries
|
sort.Slice(entries, func(i, j int) bool {
|
||||||
var freedBytes int64
|
return entries[i].expiresAt.Before(entries[j].expiresAt)
|
||||||
for i := 0; i < len(entries) && freedBytes < bytesToFree; i++ {
|
})
|
||||||
oldest := i
|
|
||||||
for j := i + 1; j < len(entries); j++ {
|
|
||||||
if entries[j].expiresAt.Before(entries[oldest].expiresAt) {
|
|
||||||
oldest = j
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// Swap
|
|
||||||
if oldest != i {
|
|
||||||
entries[i], entries[oldest] = entries[oldest], entries[i]
|
|
||||||
}
|
|
||||||
|
|
||||||
// Delete this entry
|
var freedBytes int64
|
||||||
if entry, exists := c.entries.LoadAndDelete(entries[i].key); exists {
|
for _, e := range entries {
|
||||||
|
if freedBytes >= bytesToFree {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
if entry, exists := c.entries.LoadAndDelete(e.key); exists {
|
||||||
cacheEntry := entry.(CacheEntry)
|
cacheEntry := entry.(CacheEntry)
|
||||||
atomic.AddInt64(&c.entryCount, -1)
|
atomic.AddInt64(&c.entryCount, -1)
|
||||||
atomic.AddInt64(&c.memoryUsage, -cacheEntry.MemorySize)
|
atomic.AddInt64(&c.memoryUsage, -cacheEntry.MemorySize)
|
||||||
|
|||||||
Vendored
+10
@@ -19,6 +19,16 @@ type RedisConfig struct {
|
|||||||
prefix string
|
prefix string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Close releases the underlying Redis connection pool. Called at process
|
||||||
|
// shutdown so the pooled connections and go-redis background goroutines are
|
||||||
|
// terminated instead of relying on process exit.
|
||||||
|
func (c *RedisConfig) Close() error {
|
||||||
|
if c == nil || c.client == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return c.client.Close()
|
||||||
|
}
|
||||||
|
|
||||||
func (c *RedisConfig) prependKeyName(key string) string {
|
func (c *RedisConfig) prependKeyName(key string) string {
|
||||||
builder := c.builderPool.Get().(*strings.Builder)
|
builder := c.builderPool.Get().(*strings.Builder)
|
||||||
defer c.builderPool.Put(builder)
|
defer c.builderPool.Put(builder)
|
||||||
|
|||||||
Vendored
+9
@@ -102,3 +102,12 @@ func (w *CacheWrapper) GetMemoryUsage() int64 {
|
|||||||
func (w *CacheWrapper) GetMaxMemorySize() int64 {
|
func (w *CacheWrapper) GetMaxMemorySize() int64 {
|
||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Close releases the underlying Redis client's connection pool at shutdown.
|
||||||
|
// Makes CacheWrapper satisfy io.Closer so the cache layer can close it.
|
||||||
|
func (w *CacheWrapper) Close() error {
|
||||||
|
if w == nil || w.redis == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return w.redis.Close()
|
||||||
|
}
|
||||||
|
|||||||
@@ -72,6 +72,28 @@ func (suite *Tests) TestFasthttpClientConfiguration() {
|
|||||||
// with the configured timeouts by testing requests
|
// with the configured timeouts by testing requests
|
||||||
assert.NotNil(suite.T(), client, "Client should be created")
|
assert.NotNil(suite.T(), client, "Client should be created")
|
||||||
|
|
||||||
|
// The read/write timeouts must honour the per-side config
|
||||||
|
// (CLIENT_READ_TIMEOUT / CLIENT_WRITE_TIMEOUT), falling back to the
|
||||||
|
// client timeout.
|
||||||
|
want := time.Duration(tc.readTimeout) * time.Second
|
||||||
|
if tc.readTimeout <= 0 {
|
||||||
|
want = time.Duration(tc.clientTimeout) * time.Second
|
||||||
|
if want <= 0 {
|
||||||
|
want = 30 * time.Second
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert.Equal(suite.T(), want, client.ReadTimeout,
|
||||||
|
"client read timeout should honour CLIENT_READ_TIMEOUT")
|
||||||
|
want = time.Duration(tc.writeTimeout) * time.Second
|
||||||
|
if tc.writeTimeout <= 0 {
|
||||||
|
want = time.Duration(tc.clientTimeout) * time.Second
|
||||||
|
if want <= 0 {
|
||||||
|
want = 30 * time.Second
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assert.Equal(suite.T(), want, client.WriteTimeout,
|
||||||
|
"client write timeout should honour CLIENT_WRITE_TIMEOUT")
|
||||||
|
|
||||||
// For non-zero configuration values, we can at least verify they were applied
|
// For non-zero configuration values, we can at least verify they were applied
|
||||||
// by checking the client isn't nil
|
// by checking the client isn't nil
|
||||||
assert.NotNil(suite.T(), client.TLSConfig, "TLS config should be created")
|
assert.NotNil(suite.T(), client.TLSConfig, "TLS config should be created")
|
||||||
|
|||||||
@@ -632,7 +632,12 @@ func main() {
|
|||||||
// instead of being cut off when the process exits.
|
// instead of being cut off when the process exits.
|
||||||
shutdownManager.RegisterComponent("http-proxy", shutdownHTTPProxy)
|
shutdownManager.RegisterComponent("http-proxy", shutdownHTTPProxy)
|
||||||
|
|
||||||
// Cache shutdown is handled internally by the cache implementation
|
// Close the Redis cache client's connection pool at shutdown (in-memory
|
||||||
|
// caches are no-ops).
|
||||||
|
shutdownManager.RegisterComponent("cache", func(context.Context) error {
|
||||||
|
libpack_cache.Shutdown()
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
// Start monitoring server
|
// Start monitoring server
|
||||||
cfg.Logger.Info(&libpack_logging.LogMessage{
|
cfg.Logger.Info(&libpack_logging.LogMessage{
|
||||||
|
|||||||
@@ -229,10 +229,18 @@ func createFasthttpClient(clientConfig *config) *fasthttp.Client {
|
|||||||
clientTimeout = 30 * time.Second // Default timeout of 30 seconds
|
clientTimeout = 30 * time.Second // Default timeout of 30 seconds
|
||||||
}
|
}
|
||||||
|
|
||||||
// For timeout behavior, use the client timeout for all timeout settings
|
// Honour the per-side read/write timeouts (CLIENT_READ_TIMEOUT /
|
||||||
// to ensure consistent behavior
|
// CLIENT_WRITE_TIMEOUT); they fall back to the client timeout when not
|
||||||
readTimeout := clientTimeout
|
// explicitly configured, so behaviour is unchanged for operators who only
|
||||||
writeTimeout := clientTimeout
|
// set CLIENT_TIMEOUT.
|
||||||
|
readTimeout := time.Duration(clientConfig.Client.ReadTimeout) * time.Second
|
||||||
|
if readTimeout <= 0 {
|
||||||
|
readTimeout = clientTimeout
|
||||||
|
}
|
||||||
|
writeTimeout := time.Duration(clientConfig.Client.WriteTimeout) * time.Second
|
||||||
|
if writeTimeout <= 0 {
|
||||||
|
writeTimeout = clientTimeout
|
||||||
|
}
|
||||||
|
|
||||||
// Create a custom dialer with timeout
|
// Create a custom dialer with timeout
|
||||||
dialer := &fasthttp.TCPDialer{
|
dialer := &fasthttp.TCPDialer{
|
||||||
|
|||||||
+22
-10
@@ -7,6 +7,8 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"net"
|
||||||
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"go.opentelemetry.io/otel"
|
"go.opentelemetry.io/otel"
|
||||||
@@ -37,6 +39,21 @@ type TraceSpanInfo struct {
|
|||||||
TraceParent string `json:"traceparent"`
|
TraceParent string `json:"traceparent"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// validateEndpoint checks that endpoint is a hostname:port pair with a port in
|
||||||
|
// [0, 65535]. Extracted from NewTracing so it can be unit-tested without
|
||||||
|
// needing a live collector.
|
||||||
|
func validateEndpoint(endpoint string) error {
|
||||||
|
_, portStr, err := net.SplitHostPort(endpoint)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("invalid endpoint format: must be 'hostname:port'")
|
||||||
|
}
|
||||||
|
port, err := strconv.Atoi(portStr)
|
||||||
|
if err != nil || port < 0 || port > 65535 {
|
||||||
|
return fmt.Errorf("invalid endpoint format: port must be between 0 and 65535")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// NewTracing creates a new tracing setup with OTLP exporter
|
// NewTracing creates a new tracing setup with OTLP exporter
|
||||||
func NewTracing(ctx context.Context, endpoint string) (*TracingSetup, error) {
|
func NewTracing(ctx context.Context, endpoint string) (*TracingSetup, error) {
|
||||||
if ctx == nil {
|
if ctx == nil {
|
||||||
@@ -46,16 +63,11 @@ func NewTracing(ctx context.Context, endpoint string) (*TracingSetup, error) {
|
|||||||
return nil, fmt.Errorf("endpoint cannot be empty")
|
return nil, fmt.Errorf("endpoint cannot be empty")
|
||||||
}
|
}
|
||||||
|
|
||||||
// Validate endpoint format
|
// Validate endpoint format: a hostname:port pair with a valid port.
|
||||||
// A simple validation to check if the endpoint has a reasonable format
|
// (Previous code used fmt.Sscanf("%s:%d") which treats %s as greedy and
|
||||||
// We're looking for hostname:port where port is a valid port number (0-65535)
|
// rejected every valid "host:port" string, silently disabling tracing.)
|
||||||
var host string
|
if err := validateEndpoint(endpoint); err != nil {
|
||||||
var port int
|
return nil, err
|
||||||
if n, err := fmt.Sscanf(endpoint, "%s:%d", &host, &port); err != nil || n != 2 {
|
|
||||||
return nil, fmt.Errorf("invalid endpoint format: must be 'hostname:port'")
|
|
||||||
}
|
|
||||||
if port < 0 || port > 65535 {
|
|
||||||
return nil, fmt.Errorf("invalid port number: must be between 0 and 65535")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create the exporter directly with the endpoint
|
// Create the exporter directly with the endpoint
|
||||||
|
|||||||
@@ -21,9 +21,7 @@ func TestNewTracing_NilContext_ReturnsError(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestNewTracing_InvalidEndpointFormats covers endpoint validation branches.
|
// TestNewTracing_InvalidEndpointFormats covers endpoint validation branches.
|
||||||
// Note: fmt.Sscanf("%s:%d") treats %s as greedy so any "host:port" string hits
|
// validateEndpoint rejects missing/host-only endpoints and out-of-range ports.
|
||||||
// the format error (n!=2). The port-range branch (port>65535) requires n==2
|
|
||||||
// which Sscanf never produces for "host:port" strings — that's a source quirk.
|
|
||||||
func TestNewTracing_InvalidEndpointFormats_ReturnsError(t *testing.T) {
|
func TestNewTracing_InvalidEndpointFormats_ReturnsError(t *testing.T) {
|
||||||
tests := []struct {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
@@ -118,3 +116,32 @@ func TestExtractSpanContext_ValidTraceparent_ReturnsValid(t *testing.T) {
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
assert.True(t, spanCtx.IsValid())
|
assert.True(t, spanCtx.IsValid())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestValidateEndpoint checks the extracted endpoint validator. Regression for
|
||||||
|
// the old fmt.Sscanf("%s:%d") which rejected every valid "host:port"
|
||||||
|
// endpoint (crushing tracing into permanent no-op).
|
||||||
|
func TestValidateEndpoint(t *testing.T) {
|
||||||
|
for _, tt := range []struct {
|
||||||
|
name string
|
||||||
|
endpoint string
|
||||||
|
wantErr bool
|
||||||
|
}{
|
||||||
|
{name: "valid host:port", endpoint: "localhost:4317", wantErr: false},
|
||||||
|
{name: "valid fqdn:port", endpoint: "otel.example.com:4317", wantErr: false},
|
||||||
|
{name: "valid high port", endpoint: "collector:65535", wantErr: false},
|
||||||
|
{name: "missing port", endpoint: "localhost", wantErr: true},
|
||||||
|
{name: "port too high", endpoint: "localhost:999999", wantErr: true},
|
||||||
|
{name: "non-numeric port", endpoint: "localhost:abc", wantErr: true},
|
||||||
|
{name: "bare number", endpoint: "12345", wantErr: true},
|
||||||
|
} {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
err := validateEndpoint(tt.endpoint)
|
||||||
|
if tt.wantErr {
|
||||||
|
assert.Error(t, err)
|
||||||
|
assert.Contains(t, err.Error(), "invalid endpoint format")
|
||||||
|
} else {
|
||||||
|
assert.NoError(t, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user