From 14be2398e8906c60769165592373584b306430f3 Mon Sep 17 00:00:00 2001 From: Lukasz Raczylo Date: Thu, 13 Aug 2026 07:16:52 +0100 Subject: [PATCH] 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. --- cache/cache.go | 10 +++++++++ cache/memory/eviction_test.go | 29 +++++++++++++++++++++++++ cache/memory/memory.go | 36 +++++++++++++++----------------- cache/redis/redis.go | 10 +++++++++ cache/redis/wrapper.go | 9 ++++++++ fasthttp_client_test.go | 22 +++++++++++++++++++ main.go | 7 ++++++- proxy.go | 16 ++++++++++---- tracing/tracing.go | 32 +++++++++++++++++++--------- tracing/tracing_coverage_test.go | 33 ++++++++++++++++++++++++++--- 10 files changed, 167 insertions(+), 37 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index 08c9927..41a1908 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -168,6 +168,16 @@ func EnableCache(cfg *CacheConfig) { 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 { if !IsCacheInitialized() { return nil diff --git a/cache/memory/eviction_test.go b/cache/memory/eviction_test.go index 76b5a0f..94a31c2 100644 --- a/cache/memory/eviction_test.go +++ b/cache/memory/eviction_test.go @@ -183,3 +183,32 @@ func TestSetMaxMemorySize(t *testing.T) { func generateKey(index int) string { 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") +} diff --git a/cache/memory/memory.go b/cache/memory/memory.go index cab74f4..43b1a8b 100644 --- a/cache/memory/memory.go +++ b/cache/memory/memory.go @@ -8,6 +8,7 @@ import ( "compress/gzip" "context" "io" + "sort" "sync" "sync/atomic" "time" @@ -322,32 +323,29 @@ func (c *Cache) evictToFreeMemory(bytesToFree int64) { memorySize int64 } - // Collect entries to consider for eviction - entries := make([]keyMemorySize, 0, int(c.maxCacheSize/5)) + // Collect ALL entries so eviction can actually reach the target. The + // 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 { key := k.(string) entry := v.(CacheEntry) entries = append(entries, keyMemorySize{entry.ExpiresAt, key, entry.MemorySize}) - return len(entries) < cap(entries) + return true }) - // Sort entries by expiry time (oldest first) - // Simple selection sort since we only need to find the oldest entries - var freedBytes int64 - 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] - } + // Sort by expiry time (oldest first) + sort.Slice(entries, func(i, j int) bool { + return entries[i].expiresAt.Before(entries[j].expiresAt) + }) - // Delete this entry - if entry, exists := c.entries.LoadAndDelete(entries[i].key); exists { + var freedBytes int64 + for _, e := range entries { + if freedBytes >= bytesToFree { + break + } + if entry, exists := c.entries.LoadAndDelete(e.key); exists { cacheEntry := entry.(CacheEntry) atomic.AddInt64(&c.entryCount, -1) atomic.AddInt64(&c.memoryUsage, -cacheEntry.MemorySize) diff --git a/cache/redis/redis.go b/cache/redis/redis.go index ba814f3..f97e91a 100644 --- a/cache/redis/redis.go +++ b/cache/redis/redis.go @@ -19,6 +19,16 @@ type RedisConfig struct { 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 { builder := c.builderPool.Get().(*strings.Builder) defer c.builderPool.Put(builder) diff --git a/cache/redis/wrapper.go b/cache/redis/wrapper.go index 813a0e0..20cfc4e 100644 --- a/cache/redis/wrapper.go +++ b/cache/redis/wrapper.go @@ -102,3 +102,12 @@ func (w *CacheWrapper) GetMemoryUsage() int64 { func (w *CacheWrapper) GetMaxMemorySize() int64 { 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() +} diff --git a/fasthttp_client_test.go b/fasthttp_client_test.go index 33d3a7a..05850e2 100644 --- a/fasthttp_client_test.go +++ b/fasthttp_client_test.go @@ -72,6 +72,28 @@ func (suite *Tests) TestFasthttpClientConfiguration() { // with the configured timeouts by testing requests 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 // by checking the client isn't nil assert.NotNil(suite.T(), client.TLSConfig, "TLS config should be created") diff --git a/main.go b/main.go index 0ab248e..0c15d19 100644 --- a/main.go +++ b/main.go @@ -632,7 +632,12 @@ func main() { // instead of being cut off when the process exits. 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 cfg.Logger.Info(&libpack_logging.LogMessage{ diff --git a/proxy.go b/proxy.go index ea5d5e1..614b0f2 100644 --- a/proxy.go +++ b/proxy.go @@ -229,10 +229,18 @@ func createFasthttpClient(clientConfig *config) *fasthttp.Client { clientTimeout = 30 * time.Second // Default timeout of 30 seconds } - // For timeout behavior, use the client timeout for all timeout settings - // to ensure consistent behavior - readTimeout := clientTimeout - writeTimeout := clientTimeout + // Honour the per-side read/write timeouts (CLIENT_READ_TIMEOUT / + // CLIENT_WRITE_TIMEOUT); they fall back to the client timeout when not + // explicitly configured, so behaviour is unchanged for operators who only + // 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 dialer := &fasthttp.TCPDialer{ diff --git a/tracing/tracing.go b/tracing/tracing.go index c6edc56..cbbb3d8 100644 --- a/tracing/tracing.go +++ b/tracing/tracing.go @@ -7,6 +7,8 @@ import ( "context" "encoding/json" "fmt" + "net" + "strconv" "time" "go.opentelemetry.io/otel" @@ -37,6 +39,21 @@ type TraceSpanInfo struct { 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 func NewTracing(ctx context.Context, endpoint string) (*TracingSetup, error) { if ctx == nil { @@ -46,16 +63,11 @@ func NewTracing(ctx context.Context, endpoint string) (*TracingSetup, error) { return nil, fmt.Errorf("endpoint cannot be empty") } - // Validate endpoint format - // A simple validation to check if the endpoint has a reasonable format - // We're looking for hostname:port where port is a valid port number (0-65535) - var host string - var port int - 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") + // Validate endpoint format: a hostname:port pair with a valid port. + // (Previous code used fmt.Sscanf("%s:%d") which treats %s as greedy and + // rejected every valid "host:port" string, silently disabling tracing.) + if err := validateEndpoint(endpoint); err != nil { + return nil, err } // Create the exporter directly with the endpoint diff --git a/tracing/tracing_coverage_test.go b/tracing/tracing_coverage_test.go index 0f05ad4..8fbe780 100644 --- a/tracing/tracing_coverage_test.go +++ b/tracing/tracing_coverage_test.go @@ -21,9 +21,7 @@ func TestNewTracing_NilContext_ReturnsError(t *testing.T) { } // TestNewTracing_InvalidEndpointFormats covers endpoint validation branches. -// Note: fmt.Sscanf("%s:%d") treats %s as greedy so any "host:port" string hits -// 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. +// validateEndpoint rejects missing/host-only endpoints and out-of-range ports. func TestNewTracing_InvalidEndpointFormats_ReturnsError(t *testing.T) { tests := []struct { name string @@ -118,3 +116,32 @@ func TestExtractSpanContext_ValidTraceparent_ReturnsValid(t *testing.T) { require.NoError(t, err) 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) + } + }) + } +}