feat: refactoring code, add adaptive bloom filter, adaptive migration and other updates

Author: arturmon

Readme.MD

@@ -33,34 +33,53 @@ import (
 	"github.com/arturmon/multi-tier-caching/storage"
 )
 
-func main() {
 
+func main() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 	cfg := config.LoadConfig()
 
+	var debugCache = false
+	
 	logger.InitLogger(cfg.LogLevel)
 	logger.Info("Launching the cache system", "memoryCacheSize", cfg.MemoryCacheSize)
 
-	dbStorage, err := storage.NewDatabaseStorage(cfg.DatabaseDSN)
+	memoryStorage, err := storage.NewRistrettoCache(ctx, int64(cfg.MemoryCacheSize))
+	if err != nil {
+		logger.Error("Failed to create Memory stoage: %v", err)
+	}
+
+	dbStorage, err := storage.NewDatabaseStorage(cfg.DatabaseDSN, debugCache)
 	if err != nil {
 		logger.Error("Failed to connect to the database", "error", err)
 		return
 	}
 	defer dbStorage.Close()
 
-	redisStorage, err := storage.NewRedisStorage(cfg.RedisAddr, cfg.RedisPassword)
+	redisStorage, err := storage.NewRedisStorage(cfg.RedisAddr, cfg.RedisPassword, 1)
 	if err != nil {
 		logger.Error("Failed to connect to Redis", "error", err)
 		return
 	}
 
-	cache := multi_tier_caching.NewMultiTierCache(
-		[]multi_tier_caching.CacheLayer{
-			multi_tier_caching.NewMemoryCache(),
-			multi_tier_caching.NewRedisCache(redisStorage),
+	logger.Info("Launching the cache system", "memoryCacheSize", cfg.MemoryCacheSize)
+	logger.Info("Launching the cache system", "databaseDSN", cfg.DatabaseDSN)
+	logger.Info("Launching the cache system", "redisAddr", cfg.RedisAddr)
+
+	cacheConfig := multi_tier_caching.MultiTierCacheConfig{
+		Layers: []multi_tier_caching.LayerInfo{
+			multi_tier_caching.NewLayerInfo(multi_tier_caching.NewMemoryCache(memoryStorage)),
+			multi_tier_caching.NewLayerInfo(multi_tier_caching.NewRedisCache(redisStorage)),
 		},
-		multi_tier_caching.NewDatabaseCache(dbStorage),
-	)
+		DB:          multi_tier_caching.NewDatabaseCache(dbStorage),
+		Thresholds:  []int{10, 5},
+		BloomSize:   100000,
+		BloomHashes: 5,
+		Debug:       debugCache,
+	}
 
+	cache := multi_tier_caching.NewMultiTierCache(ctx, cacheConfig)
+	
 	err = cache.Set(context.Background(), "key1", "value1")
 	if err != nil {
 		return
@@ -90,11 +109,57 @@ fmt.Println("Cached Value:", value)
 ```
 
 ## Features
-- **Multi-tier caching**: Uses memory, Redis, and database storage layers.
-- **Bloom filter**: Reduces unnecessary database queries.
-- **Write-behind queue**: Efficient database updates.
-- **TTL Management**: Supports automatic expiration of cached items.
-- **Cache Analytics**: Provides cache hit/miss statistics.
+
+- **Multi-tier caching architecture**:
+    - Configurable cache layers (e.g., hot, warm, cold) for optimized data access.
+    - Prioritizes checks from the hottest layer (in-memory) to colder layers (e.g., disk, remote).
+
+- **Bloom filter optimization**:
+    - Reduces unnecessary database queries by probabilistically checking key existence.
+    - Dynamically resizes based on request frequency, miss rate, and load factors.
+
+- **Adaptive TTL management**:
+    - Adjusts time-to-live (TTL) dynamically using key request frequency.
+    - Longer TTL for high-frequency keys to minimize cache churn.
+
+- **Intelligent data migration**:
+    - Automatically promotes/demotes keys between layers using frequency thresholds.
+    - Processes migrations asynchronously with adjustable intervals via background workers.
+
+- **Metrics and analytics**:
+    - Tracks cache hits, misses, migration times, and key frequency.
+    - Exposes Prometheus metrics (`cacheHits`, `migrationTime`, `keyFrequency`).
+
+- **Write-behind queue**:
+    - Batches and asynchronously persists updates to reduce database latency.
+    - Adaptive processing intervals based on queue load.
+
+- **Self-optimizing components**:
+    - **Bloom filter auto-scaling**: Dynamically adjusts size and hash functions.
+    - **TTL auto-tuning**: Balances cache efficiency and storage costs.
+
+- **Health monitoring**:
+    - Health checks for all cache layers and the underlying database.
+    - Graceful shutdown with resource cleanup.
+
+- **Debugging and observability**:
+    - Detailed logs for migrations, TTL adjustments, and cache operations.
+    - `GetDebugInfo()` method for real-time Bloom filter stats (false positive rate, load factor).
+
+- **Configurable policies**:
+    - Customizable frequency thresholds for layer transitions.
+    - Adjustable Bloom filter parameters (initial size, hash functions).
+
+- **Database integration**:
+    - Fallback to database on cache misses, with automatic cache refresh for fetched keys.
+
+- **Concurrency and scalability**:
+    - Thread-safe operations using mutexes and channels.
+    - Parallel migration workers for high-throughput scenarios.
+
+- **Prometheus integration**:
+    - Built-in support for Prometheus metrics (histograms, counters, gauges).
+
 
 ## Contributing
 Pull requests are welcome. Please open an issue first to discuss any major changes.

analytics.go

@@ -2,94 +2,69 @@ package multi_tier_caching
 
 import (
 	"sync"
-	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
 )
 
 type CacheAnalytics struct {
-	hits, misses  int
-	frequencies   map[string]int // Request frequency for each keyword
-	recentFreq    sync.Map       // Request frequency in the last minute
-	mu            sync.Mutex
-	lastResetTime time.Time
-}
-
-func NewCacheAnalytics() *CacheAnalytics {
-	return &CacheAnalytics{
-		frequencies:   make(map[string]int),
-		lastResetTime: time.Now(),
-	}
+	cacheHits      *prometheus.CounterVec
+	cacheMisses    prometheus.Counter
+	keyFrequency   *prometheus.GaugeVec
+	migrationTime  prometheus.Histogram
+	migrationCount *prometheus.CounterVec
+	mu             sync.RWMutex
+	keys           map[string]struct{}
 }
 
-func (ca *CacheAnalytics) LogHit(key string) {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
-	ca.hits++
-	ca.frequencies[key]++
-	valRecent, _ := ca.recentFreq.LoadOrStore(key, 0)
-	ca.recentFreq.Store(key, valRecent.(int)+1)
+// LogHit records a cache hit for a specific layer and key, updating the corresponding metrics.
+func (a *CacheAnalytics) LogHit(layerName, key string) {
+	a.cacheHits.WithLabelValues(layerName).Inc()
+	a.keyFrequency.WithLabelValues(key).Inc()
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.keys[key] = struct{}{}
 }
 
-func (ca *CacheAnalytics) LogMiss() {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
-	ca.misses++
+// LogMiss increments the cache miss counter.
+func (a *CacheAnalytics) LogMiss() {
+	a.cacheMisses.Inc()
 }
 
-func (ca *CacheAnalytics) GetStats() (int, int) {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
-	return ca.hits, ca.misses
-}
+// GetFrequency retrieves the request frequency of a specific cache key.
+func (a *CacheAnalytics) GetFrequency(key string) int {
+	gauge, err := a.keyFrequency.GetMetricWithLabelValues(key)
+	if err != nil {
+		return 0
+	}
 
-func (ca *CacheAnalytics) GetFrequency(key string) int {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
-	return ca.frequencies[key]
-}
+	var metric dto.Metric
+	if err = gauge.Write(&metric); err != nil {
+		return 0
+	}
 
-// GetTotalFrequency Returns the total number of times a key has been accessed.
-func (ca *CacheAnalytics) GetTotalFrequency(key string) int {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
-	return ca.frequencies[key]
+	return int(metric.Gauge.GetValue())
 }
 
-// GetFrequencyPerMinute Returns the request rate per minute and resets the counters
-func (ca *CacheAnalytics) GetFrequencyPerMinute() map[string]int {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
+// GetFrequencyPerMinute returns a map containing the request frequency of all tracked keys.
+func (a *CacheAnalytics) GetFrequencyPerMinute() map[string]int {
+	result := make(map[string]int)
 
-	now := time.Now()
-	if now.Sub(ca.lastResetTime).Minutes() < 1 {
-		return nil // Not a minute has passed yet
+	// Create a copy of the keys for secure iteration
+	a.mu.RLock()
+	keysCopy := make([]string, 0, len(a.keys))
+	for key := range a.keys {
+		keysCopy = append(keysCopy, key)
 	}
+	a.mu.RUnlock()
 
-	freqCopy := make(map[string]int)
-	// Iterating over the sync.Map using Range
-	ca.recentFreq.Range(func(k, v interface{}) bool {
-		if key, ok := k.(string); ok {
-			if count, ok := v.(int); ok {
-				freqCopy[key] = count
+	for _, key := range keysCopy {
+		if gauge, err := a.keyFrequency.GetMetricWithLabelValues(key); err == nil {
+			var metric dto.Metric
+			if err = gauge.Write(&metric); err == nil {
+				result[key] = int(*metric.Gauge.Value)
 			}
 		}
-		return true // continue iteration
-	})
-
-	// Resetting temporary data
-	ca.recentFreq = sync.Map{}
-	ca.lastResetTime = now
-
-	return freqCopy
-}
-
-// ResetFrequencies We reset the counters every minute
-func (ca *CacheAnalytics) ResetFrequencies() {
-	ca.mu.Lock()
-	defer ca.mu.Unlock()
-
-	now := time.Now()
-	if now.Sub(ca.lastResetTime).Minutes() >= 1 {
-		ca.frequencies = make(map[string]int) // Clearing the counters
-		ca.lastResetTime = now
 	}
+	return result
 }

analyticsMetrics.go

@@ -0,0 +1,38 @@
+package multi_tier_caching
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// NewCacheAnalytics initializes and returns a new CacheAnalytics instance with Prometheus metrics.
+func NewCacheAnalytics() *CacheAnalytics {
+	return &CacheAnalytics{
+		cacheHits: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "cache_hits_total",
+			Help: "Total cache hits per layer",
+		}, []string{"layer"}),
+
+		cacheMisses: promauto.NewCounter(prometheus.CounterOpts{
+			Name: "cache_misses_total",
+			Help: "Total cache misses",
+		}),
+
+		keyFrequency: promauto.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "cache_key_frequency",
+			Help: "Request frequency for keys",
+		}, []string{"key"}),
+
+		migrationTime: promauto.NewHistogram(prometheus.HistogramOpts{
+			Name:    "cache_migration_duration_seconds",
+			Help:    "Time spent on data migration",
+			Buckets: []float64{0.1, 0.5, 1, 5},
+		}),
+
+		migrationCount: promauto.NewCounterVec(prometheus.CounterOpts{
+			Name: "cache_migration_operations_total",
+			Help: "Total migration operations",
+		}, []string{"status"}),
+		keys: make(map[string]struct{}),
+	}
+}