Configuration Management:

Optimized the getDetailsFromEnv function to reduce redundant lookups and improve type handling
Added direct environment variable access for better performance

Memory Cache Optimization:

Implemented a size-based compression threshold (1KB) to avoid compressing small payloads
Added cache size limits (10,000 entries) to prevent memory leaks
Implemented efficient eviction strategies for the oldest entries
Added atomic counter for thread-safe cache size tracking
Improved cleanup routines with GC triggering for large caches

Proxy Implementation:

Refactored the proxy code into smaller, focused functions for better maintainability
Optimized gzip handling for better performance
Improved error handling and logging
Enhanced tracing integration

GraphQL Processing:

Optimized introspection query checking with fast-path returns
Improved object pool usage
Added detailed comments for better code understanding
Split complex functions into smaller, more focused ones
Fixed test compatibility issues with introspection checking

Request Processing:

Refactored the request processing logic into smaller, focused functions
Separated user extraction, caching, and request handling for better maintainability
Improved error handling and response generation

Tracing Enhancements:

Added better span context management
Implemented custom attributes for more detailed tracing
Added sampling configuration to reduce overhead
Improved resource attribution with host and process information
Added timeout handling for tracing operations

Application Lifecycle:

Implemented graceful shutdown with proper signal handling
Added goroutine management with wait groups
Improved startup sequence with better error handling
Added timeout handling for shutdown operations
This commit is contained in:
2025-02-25 23:34:39 +00:00
parent da577e8a02
commit 2ab78d35ce
6 changed files with 510 additions and 249 deletions
+114 -33
View File
@@ -4,14 +4,22 @@ import (
"bytes"
"compress/gzip"
"io"
"log"
"runtime"
"sync"
"sync/atomic"
"time"
)
// CompressionThreshold is the minimum size in bytes before a value is compressed
const CompressionThreshold = 1024 // 1KB
// MaxCacheSize is the maximum number of entries in the cache
const MaxCacheSize = 10000
type CacheEntry struct {
ExpiresAt time.Time
Value []byte
Compressed bool
}
type Cache struct {
@@ -19,6 +27,7 @@ type Cache struct {
decompressPool sync.Pool
entries sync.Map
globalTTL time.Duration
entryCount int64
sync.RWMutex
}
@@ -38,32 +47,66 @@ func New(globalTTL time.Duration) *Cache {
},
}
// Start cleanup routine
go cache.cleanupRoutine(globalTTL)
return cache
}
func (c *Cache) cleanupRoutine(globalTTL time.Duration) {
ticker := time.NewTicker(globalTTL / 2)
// Clean up more frequently when the cache is large
ticker := time.NewTicker(globalTTL / 4)
defer ticker.Stop()
for range ticker.C {
c.CleanExpiredEntries()
// Trigger GC if we have a lot of entries
if atomic.LoadInt64(&c.entryCount) > MaxCacheSize/2 {
runtime.GC()
}
}
}
func (c *Cache) Set(key string, value []byte, ttl time.Duration) {
// Check if we've reached the maximum cache size
if atomic.LoadInt64(&c.entryCount) >= MaxCacheSize {
c.evictOldest(MaxCacheSize / 10) // Evict 10% of entries
}
expiresAt := time.Now().Add(ttl)
compressedValue, err := c.compress(value)
if err != nil {
log.Printf("Error compressing value for key %s: %v", key, err)
return
// Only compress if the value is larger than the threshold
var entry CacheEntry
if len(value) > CompressionThreshold {
compressedValue, err := c.compress(value)
if err == nil && len(compressedValue) < len(value) {
entry = CacheEntry{
Value: compressedValue,
ExpiresAt: expiresAt,
Compressed: true,
}
} else {
// If compression failed or didn't reduce size, store uncompressed
entry = CacheEntry{
Value: value,
ExpiresAt: expiresAt,
Compressed: false,
}
}
} else {
entry = CacheEntry{
Value: value,
ExpiresAt: expiresAt,
Compressed: false,
}
}
entry := CacheEntry{
Value: compressedValue,
ExpiresAt: expiresAt,
// Check if this is a new entry
_, exists := c.entries.Load(key)
if !exists {
atomic.AddInt64(&c.entryCount, 1)
}
c.entries.Store(key, entry)
}
@@ -76,19 +119,25 @@ func (c *Cache) Get(key string) ([]byte, bool) {
cacheEntry := entry.(CacheEntry)
if cacheEntry.ExpiresAt.Before(time.Now()) {
c.entries.Delete(key)
atomic.AddInt64(&c.entryCount, -1)
return nil, false
}
value, err := c.decompress(cacheEntry.Value)
if err != nil {
log.Printf("Error decompressing value for key %s: %v", key, err)
return nil, false
if cacheEntry.Compressed {
value, err := c.decompress(cacheEntry.Value)
if err != nil {
return nil, false
}
return value, true
}
return value, true
return cacheEntry.Value, true
}
func (c *Cache) Delete(key string) {
c.entries.Delete(key)
if _, exists := c.entries.LoadAndDelete(key); exists {
atomic.AddInt64(&c.entryCount, -1)
}
}
func (c *Cache) Clear() {
@@ -96,24 +145,18 @@ func (c *Cache) Clear() {
c.entries.Delete(key)
return true
})
atomic.StoreInt64(&c.entryCount, 0)
}
func (c *Cache) CountQueries() int64 {
var count int
c.entries.Range(func(_, _ interface{}) bool {
count++
return true
})
return int64(count)
return atomic.LoadInt64(&c.entryCount)
}
func (c *Cache) compress(data []byte) ([]byte, error) {
var buf bytes.Buffer
w := c.compressPool.Get().(*gzip.Writer)
defer func() {
w.Close()
c.compressPool.Put(w)
}()
defer c.compressPool.Put(w)
w.Reset(&buf)
if _, err := w.Write(data); err != nil {
return nil, err
@@ -126,6 +169,8 @@ func (c *Cache) compress(data []byte) ([]byte, error) {
func (c *Cache) decompress(data []byte) ([]byte, error) {
r, ok := c.decompressPool.Get().(*gzip.Reader)
defer c.decompressPool.Put(r)
if !ok || r == nil {
var err error
r, err = gzip.NewReader(bytes.NewReader(data))
@@ -137,11 +182,8 @@ func (c *Cache) decompress(data []byte) ([]byte, error) {
return nil, err
}
}
defer func() {
r.Close()
c.decompressPool.Put(r)
}()
defer r.Close()
return io.ReadAll(r)
}
@@ -150,8 +192,47 @@ func (c *Cache) CleanExpiredEntries() {
c.entries.Range(func(key, value interface{}) bool {
entry := value.(CacheEntry)
if entry.ExpiresAt.Before(now) {
c.entries.Delete(key)
if _, exists := c.entries.LoadAndDelete(key); exists {
atomic.AddInt64(&c.entryCount, -1)
}
}
return true
})
}
// evictOldest removes the oldest n entries from the cache
func (c *Cache) evictOldest(n int) {
type keyExpiry struct {
key string
expiresAt time.Time
}
// Collect all entries with their expiry times
entries := make([]keyExpiry, 0, n*2)
c.entries.Range(func(k, v interface{}) bool {
key := k.(string)
entry := v.(CacheEntry)
entries = append(entries, keyExpiry{key, entry.ExpiresAt})
return len(entries) < cap(entries)
})
// Sort by expiry time (oldest first)
// Using a simple selection sort since we only need to find the n oldest
for i := 0; i < n && i < len(entries); 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
if _, exists := c.entries.LoadAndDelete(entries[i].key); exists {
atomic.AddInt64(&c.entryCount, -1)
}
}
}