status-im/eth-rpc-proxy
1
0
Fork 0
mirror of https://github.com/status-im/eth-rpc-proxy.git synced 2026-08-27 09:41:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(go-cache-service): metrics (+grafana) (#68)

Browse Source 
* feat(go-proxy-cache): handle when cache is not accessible

* feat(go-proxy-cache): serve metrics over http

* feat(go-proxy-cache): metrics

* feat(go-proxy-cache): fix pr comments

fixes #67

* feat(go-proxy-cache): update github actions

* feat(go-proxy-cache): fix linter
This commit is contained in:
Andrey Bocharnikov
2025-09-17 17:26:23 +04:00
committed by GitHub
parent df369f0949
commit b4ea66fa1f
29 changed files with 2254 additions and 494 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+30 -3
View File
@@ -8,8 +8,8 @@ on:
branches: [ master ]
jobs:
lint:
name: Run Linter
lint-rpc-health-checker:
name: Run Linter (RPC Health Checker)
runs-on: ubuntu-latest
permissions:
contents: read
@@ -33,4 +33,31 @@ jobs:
uses: golangci/golangci-lint-action@v3
with:
version: latest
working-directory: rpc-health-checker
working-directory: rpc-health-checker
lint-go-proxy-cache:
name: Run Linter (Go Proxy Cache)
runs-on: ubuntu-latest
permissions:
contents: read
steps:
- uses: actions/checkout@v4
- name: Set up Go
uses: actions/setup-go@v5
with:
go-version: '1.23'
cache: true
cache-dependency-path: go-proxy-cache/go.sum
- name: Install dependencies
run: |
cd go-proxy-cache
go mod download
- name: Run golangci-lint
uses: golangci/golangci-lint-action@v3
with:
version: latest
working-directory: go-proxy-cache
.github/workflows/test.yml
+29 -2
View File
@@ -8,8 +8,8 @@ on:
branches: [ master ]
jobs:
test:
name: Run Tests
test-rpc-health-checker:
name: Run Tests (RPC Health Checker)
runs-on: ubuntu-latest
permissions:
contents: read
@@ -33,4 +33,31 @@ jobs:
id: tests
run: |
cd rpc-health-checker
go test -v -race ./...
test-go-proxy-cache:
name: Run Tests (Go Proxy Cache)
runs-on: ubuntu-latest
permissions:
contents: read
steps:
- uses: actions/checkout@v4
- name: Set up Go
uses: actions/setup-go@v5
with:
go-version: '1.23'
cache: true
cache-dependency-path: go-proxy-cache/go.sum
- name: Install dependencies
run: |
cd go-proxy-cache
go mod download
- name: Run tests
id: tests
run: |
cd go-proxy-cache
go test -v -race ./...
docker-compose-local.yml
+2 -14
View File
@@ -110,19 +110,6 @@ services:
volumes:
- 'keydb-data:/data'
keydb-exporter:
image: 'oliver006/redis_exporter:v1.74.0'
container_name: 'keydb-exporter'
restart: 'always'
ports:
- '9121:9121'
networks:
- 'rpc-network'
environment:
REDIS_ADDR: 'keydb:6379'
depends_on:
- 'keydb'
cache-service:
build:
context: './go-proxy-cache'
@@ -132,7 +119,7 @@ services:
environment:
CACHE_CONFIG_FILE: '/app/cache_config.yaml'
CACHE_RULES_FILE: '/app/cache_rules.yaml'
KEYDB_URL: 'redis://keydb:6379'
CACHE_KEYDB_URL_FILE: '/app/.keydb-url'
CACHE_SOCKET_PATH: '/tmp/cache.sock'
# No ports exposed - using Unix socket only
networks:
@@ -142,6 +129,7 @@ services:
volumes:
- './go-proxy-cache/cache_config.yaml:/app/cache_config.yaml:ro'
- './go-proxy-cache/cache_rules.yaml:/app/cache_rules.yaml:ro'
- './secrets/.keydb-url:/app/.keydb-url:ro'
- 'cache_socket:/tmp'
healthcheck:
test: ['CMD-SHELL', 'test -S /tmp/cache.sock || exit 1']
docker-compose.yml
+3 -1
View File
@@ -99,7 +99,9 @@ services:
CACHE_RULES_FILE: '/app/cache_rules.yaml'
CACHE_KEYDB_URL_FILE: '/app/.keydb-url'
CACHE_SOCKET_PATH: '/tmp/cache.sock'
# No ports exposed - using Unix socket only
CACHE_METRICS_PORT: '8099'
ports:
- '8099:8099' # Expose metrics endpoint on port 8099
networks:
- 'rpc-network'
volumes:
go-proxy-cache/METRICS.md
+78
View File
@@ -0,0 +1,78 @@
# Cache Metrics Reference
## Overview
Brief description of all cache metrics with their types and cardinality.
**Note**: All cache metrics are prefixed with `eth_rpc_proxy_` in the actual Prometheus output
## Core Cache Metrics
| Metric Name | Type | Labels | Cardinality | Description |
|-------------|------|--------|-------------|-------------|
| `cache_requests_total` | Counter | `cache_type`, `level`, `network`, `rpc_method` | ~1,035 | Total number of cache requests |
| `cache_hits_total` | Counter | `cache_type`, `level`, `network`, `rpc_method` | ~1,035 | Total number of cache hits |
| `cache_misses_total` | Counter | `cache_type`, `level`, `network`, `rpc_method` | ~1,035 | Total number of cache misses |
**Total Core Metrics Cardinality**: ~3,105 time series
## Operational Metrics
| Metric Name | Type | Labels | Cardinality | Description |
|-------------|------|--------|-------------|-------------|
| `cache_sets_total` | Counter | `level`, `cache_type`, `network` | 45 | Number of cache set operations |
| `cache_evictions_total` | Counter | `level`, `cache_type`, `network` | 45 | Number of cache evictions |
| `cache_errors_total` | Counter | `level`, `kind` | 12 | Cache errors by type |
| `cache_bytes_read_total` | Counter | `level`, `cache_type`, `network` | 45 | Bytes read from cache |
| `cache_bytes_written_total` | Counter | `level`, `cache_type`, `network` | 45 | Bytes written to cache |
**Total Operational Metrics Cardinality**: 192 time series
## Performance Metrics
| Metric Name | Type | Labels | Cardinality | Description |
|-------------|------|--------|-------------|-------------|
| `cache_operation_duration_seconds` | Histogram | `operation`, `level` | 78 | Duration of cache operations (get/set) |
| `cache_item_age_seconds` | Histogram | `level`, `cache_type` | 117 | Age of items at hit time (TTL analysis) |
**Total Performance Metrics Cardinality**: 195 time series
## Capacity Metrics
| Metric Name | Type | Labels | Cardinality | Description |
|-------------|------|--------|-------------|-------------|
| `cache_keys` | Gauge | `level` | 3 | Current number of keys in cache |
| `cache_capacity_bytes` | Gauge | `level` | 1 | L1 cache capacity in bytes |
| `cache_used_bytes` | Gauge | `level` | 1 | L1 cache used space in bytes |
**Total Capacity Metrics Cardinality**: 5 time series
## Label Values
| Label | Values | Count | Description |
|-------|--------|-------|-------------|
| `cache_type` | `permanent`, `short`, `minimal` | 3 | Cache type based on TTL rules |
| `level` | `l1`, `l2`, `origin` | 3 | Cache level or origin server |
| `network` | `ethereum:mainnet`, `polygon:mainnet`, `unknown`, etc. | ~5 | Network identifier |
| `rpc_method` | Whitelisted methods + `other` | 23 | RPC method name (controlled) |
| `operation` | `get`, `set` | 2 | Cache operation type |
| `kind` | `encode`, `decode`, `upstream`, `redis` | 4 | Error type |
## RPC Method Whitelist
| Category | Methods | Count |
|----------|---------|-------|
| **Permanent Data** | `eth_getBlockByHash`, `eth_getBlockByNumber`, `eth_getTransactionByHash`, `eth_getTransactionReceipt`, `eth_getLogs` | 5 |
| **Short-lived Data** | `eth_blockNumber`, `eth_gasPrice`, `eth_getBalance`, `eth_getCode`, `eth_getStorageAt`, `eth_getTransactionCount`, `eth_call`, `eth_estimateGas` | 8 |
| **Minimal Cache** | `eth_sendRawTransaction`, `eth_sendTransaction` | 2 |
| **Web3/Net Methods** | `web3_clientVersion`, `web3_sha3`, `net_version`, `net_listening`, `net_peerCount` | 5 |
| **Other Methods** | All unlisted methods aggregated as `other` | 1 |
| **Total** | | **23** |
## Error Types
| Error Kind | Source | Description |
|------------|--------|-------------|
| `encode` | Serialization | JSON marshaling errors |
| `decode` | Deserialization | JSON unmarshaling errors |
| `upstream` | BigCache | L1 cache operation errors |
| `redis` | KeyDB/Redis | L2 cache connection/operation errors |
go-proxy-cache/cmd/rpc-cache/composition_root.go
+40 -3
View File
@@ -15,6 +15,7 @@ import (
"go-proxy-cache/internal/config"
"go-proxy-cache/internal/httpserver"
"go-proxy-cache/internal/interfaces"
"go-proxy-cache/internal/metrics"
)
// CompositionRoot holds all application dependencies and provides a centralized
@@ -36,8 +37,9 @@ type CompositionRoot struct {
KeyBuilder interfaces.KeyBuilder
// Services
CacheService *service.CacheService
HTTPServer *httpserver.Server
CacheService *service.CacheService
HTTPServer *httpserver.Server
MetricsServer *httpserver.MetricsServer
}
// NewCompositionRoot creates and initializes all application dependencies.
@@ -84,6 +86,11 @@ func NewCompositionRoot() (*CompositionRoot, error) {
return nil, fmt.Errorf("failed to initialize HTTP server: %w", err)
}
// Initialize metrics server
if err := root.initMetricsServer(); err != nil {
return nil, fmt.Errorf("failed to initialize metrics server: %w", err)
}
return root, nil
}
@@ -127,6 +134,10 @@ func (r *CompositionRoot) loadCacheRules() error {
// Create classifier from the loaded config
r.CacheRules = cache_rules.NewClassifier(r.Logger, cacheRules)
// Initialize metrics allowed methods from cache rules
r.initMetrics(cacheRules)
return nil
}
@@ -172,7 +183,11 @@ func (r *CompositionRoot) initL2Cache() error {
// Create KeyDB client
keydbClient, err := l2.NewRedisKeyDbClient(&r.Config.KeyDB, keydbURL, r.Logger)
if err != nil {
return err
r.Logger.Warn("Failed to connect to KeyDB, falling back to no L2 cache",
zap.String("keydb_url", keydbURL),
zap.Error(err))
r.L2Cache = noop.NewNoOpCache()
return nil
}
// Create L2 cache with the client
@@ -209,6 +224,12 @@ func (r *CompositionRoot) initHTTPServer() error {
return nil
}
// initMetricsServer initializes the metrics HTTP server
func (r *CompositionRoot) initMetricsServer() error {
r.MetricsServer = httpserver.NewMetricsServer(r.Logger)
return nil
}
// Cleanup performs cleanup of all resources
func (r *CompositionRoot) Cleanup() error {
var errors []error
@@ -254,3 +275,19 @@ func (r *CompositionRoot) GetSocketPath() string {
}
return socketPath
}
// initMetrics initializes metrics system with allowed methods from cache rules
func (r *CompositionRoot) initMetrics(cacheRulesConfig interfaces.CacheRulesConfig) {
methods := cacheRulesConfig.GetAllMethods()
metrics.InitializeAllowedMethods(methods)
r.Logger.Info("Metrics initialized", zap.Int("allowed_methods_count", len(methods)), zap.Strings("methods", methods))
}
// GetMetricsPort returns the port for the metrics HTTP server
func (r *CompositionRoot) GetMetricsPort() string {
port := os.Getenv("CACHE_METRICS_PORT")
if port == "" {
port = "8080"
}
return port
}
go-proxy-cache/cmd/rpc-cache/main.go
+14 -2
View File
@@ -37,6 +37,15 @@ func main() {
}
}()
// Start metrics server on HTTP port
metricsPort := root.GetMetricsPort()
root.Logger.Info("Starting metrics server", zap.String("port", metricsPort))
go func() {
if err := root.MetricsServer.Start(metricsPort); err != nil {
root.Logger.Error("Metrics server failed to start", zap.Error(err))
}
}()
// Wait for interrupt signal to gracefully shutdown
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
@@ -48,9 +57,12 @@ func main() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Shutdown server
// Shutdown servers
if err := root.HTTPServer.Stop(ctx); err != nil {
root.Logger.Error("Server forced to shutdown", zap.Error(err))
root.Logger.Error("HTTP server forced to shutdown", zap.Error(err))
}
if err := root.MetricsServer.Stop(ctx); err != nil {
root.Logger.Error("Metrics server forced to shutdown", zap.Error(err))
}
root.Logger.Info("Server exited")
go-proxy-cache/cmd/rpc-cache/utils.go
+3 -1
View File
@@ -17,7 +17,7 @@ const (
// 3. Default value
func GetKeyDBURL(logger *zap.Logger) string {
// Priority 1: Environment variable
if keydbURL := os.Getenv("KEYDB_URL"); keydbURL != "" {
if keydbURL := strings.TrimSpace(os.Getenv("KEYDB_URL")); keydbURL != "" {
logger.Info("Using KeyDB URL from environment variable", zap.String("url", keydbURL))
return keydbURL
}
@@ -33,6 +33,8 @@ func GetKeyDBURL(logger *zap.Logger) string {
if len(keydbURL) > 0 {
logger.Info("Using KeyDB URL from connection file", zap.String("file", connectionFile), zap.String("url", keydbURL))
return keydbURL
} else {
logger.Warn("KeyDB connection file is empty", zap.String("file", connectionFile))
}
} else {
logger.Debug("KeyDB connection file not found", zap.String("file", connectionFile), zap.Error(err))
go-proxy-cache/go.mod
+1 -1
View File
@@ -10,7 +10,7 @@ require (
github.com/gorilla/mux v1.8.0
github.com/prometheus/client_golang v1.23.0
github.com/stretchr/testify v1.10.0
go.uber.org/mock v0.5.2
go.uber.org/mock v0.6.0
go.uber.org/zap v1.24.0
gopkg.in/yaml.v3 v3.0.1
)
go-proxy-cache/go.sum
+2 -2
View File
@@ -58,8 +58,8 @@ go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.uber.org/mock v0.5.2 h1:LbtPTcP8A5k9WPXj54PPPbjcI4Y6lhyOZXn+VS7wNko=
go.uber.org/mock v0.5.2/go.mod h1:wLlUxC2vVTPTaE3UD51E0BGOAElKrILxhVSDYQLld5o=
go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
go-proxy-cache/internal/cache/l1/bigcache.go
Vendored
+57 -12
View File
@@ -10,7 +10,9 @@ import (
"go-proxy-cache/internal/config"
"go-proxy-cache/internal/interfaces"
"go-proxy-cache/internal/metrics"
"go-proxy-cache/internal/models"
"go-proxy-cache/internal/scheduler"
)
// Ensure BigCache implements interfaces.Cache
@@ -18,8 +20,9 @@ var _ interfaces.Cache = (*BigCache)(nil)
// BigCache implements L1 cache using BigCache
type BigCache struct {
cache *bigcache.BigCache
logger *zap.Logger
cache *bigcache.BigCache
logger *zap.Logger
metricsScheduler *scheduler.Scheduler
}
// NewBigCache creates a new BigCache instance
@@ -34,10 +37,15 @@ func NewBigCache(bigcacheCfg *config.BigCacheConfig, logger *zap.Logger) (interf
return nil, err
}
return &BigCache{
bc := &BigCache{
cache: cache,
logger: logger,
}, nil
}
// Start periodic metrics collection
bc.startMetricsCollection()
return bc, nil
}
// Get retrieves value from cache with freshness information
@@ -49,13 +57,15 @@ func (bc *BigCache) Get(key string) (*models.CacheEntry, bool) {
var entry models.CacheEntry
if err := json.Unmarshal(data, &entry); err != nil {
bc.cache.Delete(key) // Remove corrupted entry
bc.logger.Warn("Failed to unmarshal L1 cache entry", zap.String("key", key), zap.Error(err))
metrics.RecordCacheError("l1", "decode")
_ = bc.cache.Delete(key) // Remove corrupted entry
return nil, false
}
// Check if entry is expired
if entry.IsExpired() {
bc.cache.Delete(key)
_ = bc.cache.Delete(key)
return nil, false
}
@@ -71,13 +81,13 @@ func (bc *BigCache) GetStale(key string) (*models.CacheEntry, bool) {
var entry models.CacheEntry
if err := json.Unmarshal(data, &entry); err != nil {
bc.cache.Delete(key) // Remove corrupted entry
_ = bc.cache.Delete(key) // Remove corrupted entry
return nil, false
}
// Check if entry is completely expired (beyond stale time)
if entry.IsExpired() {
bc.cache.Delete(key)
_ = bc.cache.Delete(key)
return nil, false
}
@@ -98,26 +108,28 @@ func (bc *BigCache) Set(key string, val []byte, ttl models.TTL) {
data, err := json.Marshal(entry)
if err != nil {
bc.logger.Error("Failed to marshal cache entry", zap.String("key", key), zap.Error(err))
metrics.RecordCacheError("l1", "encode")
return
}
err = bc.cache.Set(key, data)
if err != nil {
bc.logger.Error("Failed to set cache entry", zap.String("key", key), zap.Error(err))
metrics.RecordCacheError("l1", "upstream")
return
}
}
// Delete removes entry from cache
func (bc *BigCache) Delete(key string) {
err := bc.cache.Delete(key)
if err != nil {
return
}
_ = bc.cache.Delete(key)
}
// Close closes the cache
func (bc *BigCache) Close() error {
// Stop metrics collection
bc.stopMetricsCollection()
return bc.cache.Close()
}
@@ -128,5 +140,38 @@ func (bc *BigCache) GetStats() (capacity, used int64) {
// Convert from MB to bytes
capacity = int64(bc.cache.Capacity()) // This returns the configured size in bytes
used = int64(stats.Hits + stats.Misses) // Approximate usage based on operations
return capacity, used
}
// startMetricsCollection starts periodic metrics collection
func (bc *BigCache) startMetricsCollection() {
bc.metricsScheduler = scheduler.New(30*time.Second, bc.updateMetrics)
bc.metricsScheduler.Start()
// Initial collection
bc.updateMetrics()
bc.logger.Debug("Started L1 cache metrics collection")
}
// stopMetricsCollection stops periodic metrics collection
func (bc *BigCache) stopMetricsCollection() {
if bc.metricsScheduler != nil {
bc.metricsScheduler.Stop()
bc.logger.Debug("Stopped L1 cache metrics collection")
}
}
// updateMetrics updates cache metrics
func (bc *BigCache) updateMetrics() {
capacity, used := bc.GetStats()
// Update capacity metrics
metrics.UpdateL1CacheCapacity(capacity, used)
// Update key count (estimated from operations)
stats := bc.cache.Stats()
totalOps := stats.Hits + stats.Misses
metrics.UpdateCacheKeys("l1", int64(totalOps))
}
go-proxy-cache/internal/cache/l1/bigcache_test.go
Vendored
+4 -4
View File
@@ -88,7 +88,7 @@ func TestBigCache_Set_And_Get_Stale(t *testing.T) {
// Manually marshal and set the entry
entryJSON, _ := json.Marshal(entry)
bigCache.cache.Set("test-key", entryJSON)
_ = bigCache.cache.Set("test-key", entryJSON)
// Get the value (should be stale but not expired)
result, found := cache.Get("test-key")
@@ -119,7 +119,7 @@ func TestBigCache_Set_And_Get_Expired(t *testing.T) {
// Manually marshal and set the entry
entryJSON, _ := json.Marshal(entry)
bigCache.cache.Set("test-key", entryJSON)
_ = bigCache.cache.Set("test-key", entryJSON)
// Get the value (should be expired and not found)
result, found := cache.Get("test-key")
@@ -148,7 +148,7 @@ func TestBigCache_GetStale_Success(t *testing.T) {
// Manually marshal and set the entry
entryJSON, _ := json.Marshal(entry)
bigCache.cache.Set("test-key", entryJSON)
_ = bigCache.cache.Set("test-key", entryJSON)
// Get stale value
result, found := cache.GetStale("test-key")
@@ -190,7 +190,7 @@ func TestBigCache_GetStale_Expired(t *testing.T) {
// Manually marshal and set the entry
entryJSON, _ := json.Marshal(entry)
bigCache.cache.Set("test-key", entryJSON)
_ = bigCache.cache.Set("test-key", entryJSON)
// Try to get stale value (should be expired)
result, found := cache.GetStale("test-key")
go-proxy-cache/internal/cache/l2/keydb.go
Vendored
+16 -4
View File
@@ -3,12 +3,15 @@ package l2
import (
"context"
"encoding/json"
"errors"
"time"
"github.com/go-redis/redis/v8"
"go.uber.org/zap"
"go-proxy-cache/internal/config"
"go-proxy-cache/internal/interfaces"
"go-proxy-cache/internal/metrics"
"go-proxy-cache/internal/models"
)
@@ -38,13 +41,17 @@ func (kc *KeyDBCache) Get(key string) (*models.CacheEntry, bool) {
data, err := kc.client.Get(ctx, key).Result()
if err != nil {
kc.logger.Error("L2 cache get error", zap.String("key", key), zap.Error(err))
if errors.Is(err, redis.Nil) { // Cache miss
return nil, false
}
kc.logger.Warn("L2 cache get failed", zap.String("key", key), zap.Error(err))
return nil, false
}
var entry models.CacheEntry
if err := json.Unmarshal([]byte(data), &entry); err != nil {
kc.logger.Error("Failed to unmarshal L2 cache entry", zap.String("key", key), zap.Error(err))
metrics.RecordCacheError("l2", "decode")
kc.client.Del(context.Background(), key)
return nil, false
}
@@ -65,7 +72,10 @@ func (kc *KeyDBCache) GetStale(key string) (*models.CacheEntry, bool) {
data, err := kc.client.Get(ctx, key).Result()
if err != nil {
kc.logger.Error("L2 cache stale get error", zap.String("key", key), zap.Error(err))
if errors.Is(err, redis.Nil) { // Cache miss
return nil, false
}
kc.logger.Warn("L2 cache stale get failed", zap.String("key", key), zap.Error(err))
return nil, false
}
@@ -102,6 +112,7 @@ func (kc *KeyDBCache) Set(key string, val []byte, ttl models.TTL) {
data, err := json.Marshal(entry)
if err != nil {
kc.logger.Error("Failed to marshal L2 cache entry", zap.String("key", key), zap.Error(err))
metrics.RecordCacheError("l2", "encode")
return
}
@@ -109,7 +120,8 @@ func (kc *KeyDBCache) Set(key string, val []byte, ttl models.TTL) {
totalTTL := ttl.Fresh + ttl.Stale
err = kc.client.Set(ctx, key, data, totalTTL).Err()
if err != nil {
kc.logger.Error("Failed to set L2 cache entry", zap.String("key", key), zap.Error(err))
kc.logger.Warn("Failed to set L2 cache entry", zap.String("key", key), zap.Error(err))
metrics.RecordCacheError("l2", "redis")
return
}
}
@@ -121,7 +133,7 @@ func (kc *KeyDBCache) Delete(key string) {
err := kc.client.Del(ctx, key).Err()
if err != nil {
kc.logger.Error("Failed to delete L2 cache entry", zap.String("key", key), zap.Error(err))
kc.logger.Warn("Failed to delete L2 cache entry", zap.String("key", key), zap.Error(err))
return
}
}
go-proxy-cache/internal/cache/l2/keydb_client.go
Vendored
+2 -1
View File
@@ -69,7 +69,8 @@ func NewRedisKeyDbClient(keydbCfg *config.KeyDBConfig, keydbURL string, logger *
defer cancel()
if err := client.Ping(ctx).Err(); err != nil {
return nil, fmt.Errorf("failed to connect to KeyDB: %w", err)
client.Close() // Clean up the client
return nil, fmt.Errorf("failed to connect to KeyDB at %s: %w", opts.Addr, err)
}
logger.Info("Connected to KeyDB",
go-proxy-cache/internal/cache/service/service.go
Vendored
+48 -4
View File
@@ -9,6 +9,7 @@ import (
"go-proxy-cache/internal/cache"
"go-proxy-cache/internal/cache/multi"
"go-proxy-cache/internal/interfaces"
"go-proxy-cache/internal/metrics"
"go-proxy-cache/internal/models"
"go-proxy-cache/internal/utils"
)
@@ -19,6 +20,7 @@ type CacheService struct {
keyBuilder interfaces.KeyBuilder
cacheClassifier interfaces.CacheRulesClassifier
logger *zap.Logger
l1Cache interfaces.Cache // Keep reference to L1 cache for metrics
}
// NewCacheService creates a new cache service instance with MultiCache
@@ -27,12 +29,15 @@ func NewCacheService(l1Cache, l2Cache interfaces.Cache, cacheClassifier interfac
caches := []interfaces.Cache{l1Cache, l2Cache}
multiCache := multi.NewMultiCache(caches, logger, enablePropagation)
return &CacheService{
service := &CacheService{
multiCache: multiCache,
keyBuilder: cache.NewKeyBuilder(),
cacheClassifier: cacheClassifier,
logger: logger,
l1Cache: l1Cache,
}
return service
}
// GetResponse represents the result of a cache get operation
@@ -63,6 +68,8 @@ func (s *CacheService) Get(chain, network, rawBody string) (*GetResponse, error)
// Check if caching should be bypassed (TTL = 0)
cacheInfo := s.cacheClassifier.GetTtl(chain, network, request)
cacheType := string(cacheInfo.CacheType)
if cacheInfo.TTL == 0 {
return &GetResponse{
Found: false,
@@ -70,15 +77,36 @@ func (s *CacheService) Get(chain, network, rawBody string) (*GetResponse, error)
Data: "",
Key: key,
Bypass: true,
CacheType: string(cacheInfo.CacheType),
CacheType: cacheType,
TTL: int(cacheInfo.TTL.Seconds()),
CacheLevel: models.CacheLevelMiss,
}, nil
}
// Start timing cache get operation
timer := metrics.TimeCacheGetOperation("multi")
defer timer()
// Try MultiCache with level information
result := s.multiCache.GetWithLevel(key)
if result.Found && result.Entry != nil {
// Record cache hit with level information
var level string
switch result.Level {
case models.CacheLevelL1:
level = "l1"
case models.CacheLevelL2:
level = "l2"
default:
level = "unknown"
}
// Calculate item age for TTL effectiveness analysis
itemAge := time.Duration(time.Now().Unix()-result.Entry.CreatedAt) * time.Second
metrics.RecordCacheHit(cacheType, level, chain, network, request.Method, itemAge)
// Record bytes read
metrics.RecordCacheBytesRead(level, cacheType, chain, network, len(result.Entry.Data))
// Fix response ID to match current request
fixedData := utils.FixResponseID(string(result.Entry.Data), request.ID)
@@ -88,19 +116,22 @@ func (s *CacheService) Get(chain, network, rawBody string) (*GetResponse, error)
Data: fixedData,
Key: key,
Bypass: false,
CacheType: string(cacheInfo.CacheType),
CacheType: cacheType,
TTL: int(cacheInfo.TTL.Seconds()),
CacheLevel: result.Level,
}, nil
}
// Record cache miss
metrics.RecordCacheMiss(cacheType, chain, network, request.Method)
return &GetResponse{
Found: false,
Fresh: false,
Data: "",
Key: key,
Bypass: false,
CacheType: string(cacheInfo.CacheType),
CacheType: cacheType,
TTL: int(cacheInfo.TTL.Seconds()),
CacheLevel: models.CacheLevelMiss,
}, nil
@@ -140,7 +171,20 @@ func (s *CacheService) Set(chain, network, rawBody, responseData string, customT
// Store using MultiCache (will store in all configured caches)
ttlStruct := models.TTL{Fresh: ttl, Stale: staleTTL}
// Time the set operation
timer := metrics.TimeCacheOperation("set", "multi")
s.multiCache.Set(key, []byte(responseData), ttlStruct)
timer()
// Record cache set operation and bytes written for each level
cacheInfo := s.cacheClassifier.GetTtl(chain, network, request)
cacheType := string(cacheInfo.CacheType)
dataSize := len(responseData)
// Record for both L1 and L2 (since MultiCache writes to both)
metrics.RecordCacheSet("l1", cacheType, chain, network, dataSize)
metrics.RecordCacheSet("l2", cacheType, chain, network, dataSize)
return nil
}
go-proxy-cache/internal/cache_rules/config.go
+9
View File
@@ -73,6 +73,15 @@ func (cr *CacheConfig) lookupTTL(key string, cacheType models.CacheType) time.Du
return 0
}
// GetAllMethods returns all configured RPC methods from cache rules
func (cr *CacheConfig) GetAllMethods() []string {
methods := make([]string, 0, len(cr.config.CacheRules))
for method := range cr.config.CacheRules {
methods = append(methods, method)
}
return methods
}
// getFallbackTTL provides fallback TTL values when config is not available
func (cr *CacheConfig) getFallbackTTL(cacheType models.CacheType) time.Duration {
fallbackTTLs := map[models.CacheType]time.Duration{
go-proxy-cache/internal/httpserver/metrics_server.go
+50
View File
@@ -0,0 +1,50 @@
package httpserver
import (
"context"
"net/http"
"time"
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.uber.org/zap"
)
// MetricsServer represents a separate HTTP server for metrics
type MetricsServer struct {
logger *zap.Logger
server *http.Server
}
// NewMetricsServer creates a new metrics HTTP server
func NewMetricsServer(logger *zap.Logger) *MetricsServer {
return &MetricsServer{
logger: logger,
}
}
// Start starts the metrics HTTP server on the specified port
func (ms *MetricsServer) Start(port string) error {
mux := http.NewServeMux()
mux.Handle("/metrics", promhttp.Handler())
ms.server = &http.Server{
Addr: ":" + port,
Handler: mux,
ReadTimeout: 30 * time.Second,
WriteTimeout: 30 * time.Second,
IdleTimeout: 60 * time.Second,
}
ms.logger.Info("Starting metrics HTTP server", zap.String("port", port))
return ms.server.ListenAndServe()
}
// Stop stops the metrics HTTP server
func (ms *MetricsServer) Stop(ctx context.Context) error {
if ms.server == nil {
return nil
}
ms.logger.Info("Stopping metrics HTTP server")
return ms.server.Shutdown(ctx)
}
go-proxy-cache/internal/httpserver/server.go
+7 -5
View File
@@ -10,7 +10,6 @@ import (
"time"
"github.com/gorilla/mux"
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.uber.org/zap"
"go-proxy-cache/internal/cache/service"
@@ -65,7 +64,13 @@ func (s *Server) StartUnixSocket(socketPath string) error {
// Stop stops the HTTP server
func (s *Server) Stop(ctx context.Context) error {
s.logger.Info("Stopping cache HTTP server")
return s.server.Shutdown(ctx)
// Stop main server
if s.server != nil {
return s.server.Shutdown(ctx)
}
return nil
}
// createRouter creates and configures the HTTP router
@@ -82,9 +87,6 @@ func (s *Server) createRouter() *mux.Router {
// Cache info endpoint (equivalent to cache rules check)
router.HandleFunc("/cache/info", s.handleCacheInfo).Methods("POST")
// Prometheus metrics endpoint
router.Handle("/metrics", promhttp.Handler()).Methods("GET")
return router
}
go-proxy-cache/internal/httpserver/server_test.go
-19
View File
@@ -61,25 +61,6 @@ func (m *mockCache) Delete(key string) {
delete(m.data, key)
}
// mockKeyBuilder implements the KeyBuilder interface for testing
type mockKeyBuilder struct{}
func (m *mockKeyBuilder) Build(chain string, network string, req *models.JSONRPCRequest) (key string, paramsHash uint32) {
if req == nil || req.Method == "" {
return "", 0
}
return chain + ":" + network + ":" + req.Method, 12345
}
func (m *mockKeyBuilder) BuildBatch(chain, network string, reqs []models.JSONRPCRequest) ([]string, []uint32) {
keys := make([]string, len(reqs))
hashes := make([]uint32, len(reqs))
for i, req := range reqs {
keys[i], hashes[i] = m.Build(chain, network, &req)
}
return keys, hashes
}
// setupMockCacheClassifier configures the mock cache classifier with common expectations
func setupMockCacheClassifier(ctrl *gomock.Controller) *mock.MockCacheRulesClassifier {
mockClassifier := mock.NewMockCacheRulesClassifier(ctrl)
go-proxy-cache/internal/interfaces/cache_rules_config.go
+2
View File
@@ -14,4 +14,6 @@ type CacheRulesConfig interface {
// for a given chain and network combination
GetTtlForCacheType(chain, network string, cacheType models.CacheType) time.Duration
GetCacheTypeForMethod(method string) models.CacheType
// GetAllMethods returns all configured RPC methods
GetAllMethods() []string
}
go-proxy-cache/internal/interfaces/mock/cache.go
+106
View File
@@ -93,3 +93,109 @@ func (mr *MockCacheMockRecorder) Set(key, val, ttl any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Set", reflect.TypeOf((*MockCache)(nil).Set), key, val, ttl)
}
// MockLevelAwareCache is a mock of LevelAwareCache interface.
type MockLevelAwareCache struct {
ctrl *gomock.Controller
recorder *MockLevelAwareCacheMockRecorder
isgomock struct{}
}
// MockLevelAwareCacheMockRecorder is the mock recorder for MockLevelAwareCache.
type MockLevelAwareCacheMockRecorder struct {
mock *MockLevelAwareCache
}
// NewMockLevelAwareCache creates a new mock instance.
func NewMockLevelAwareCache(ctrl *gomock.Controller) *MockLevelAwareCache {
mock := &MockLevelAwareCache{ctrl: ctrl}
mock.recorder = &MockLevelAwareCacheMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockLevelAwareCache) EXPECT() *MockLevelAwareCacheMockRecorder {
return m.recorder
}
// Delete mocks base method.
func (m *MockLevelAwareCache) Delete(key string) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "Delete", key)
}
// Delete indicates an expected call of Delete.
func (mr *MockLevelAwareCacheMockRecorder) Delete(key any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Delete", reflect.TypeOf((*MockLevelAwareCache)(nil).Delete), key)
}
// Get mocks base method.
func (m *MockLevelAwareCache) Get(key string) (*models.CacheEntry, bool) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Get", key)
ret0, _ := ret[0].(*models.CacheEntry)
ret1, _ := ret[1].(bool)
return ret0, ret1
}
// Get indicates an expected call of Get.
func (mr *MockLevelAwareCacheMockRecorder) Get(key any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockLevelAwareCache)(nil).Get), key)
}
// GetStale mocks base method.
func (m *MockLevelAwareCache) GetStale(key string) (*models.CacheEntry, bool) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetStale", key)
ret0, _ := ret[0].(*models.CacheEntry)
ret1, _ := ret[1].(bool)
return ret0, ret1
}
// GetStale indicates an expected call of GetStale.
func (mr *MockLevelAwareCacheMockRecorder) GetStale(key any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStale", reflect.TypeOf((*MockLevelAwareCache)(nil).GetStale), key)
}
// GetStaleWithLevel mocks base method.
func (m *MockLevelAwareCache) GetStaleWithLevel(key string) *models.CacheResult {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetStaleWithLevel", key)
ret0, _ := ret[0].(*models.CacheResult)
return ret0
}
// GetStaleWithLevel indicates an expected call of GetStaleWithLevel.
func (mr *MockLevelAwareCacheMockRecorder) GetStaleWithLevel(key any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStaleWithLevel", reflect.TypeOf((*MockLevelAwareCache)(nil).GetStaleWithLevel), key)
}
// GetWithLevel mocks base method.
func (m *MockLevelAwareCache) GetWithLevel(key string) *models.CacheResult {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetWithLevel", key)
ret0, _ := ret[0].(*models.CacheResult)
return ret0
}
// GetWithLevel indicates an expected call of GetWithLevel.
func (mr *MockLevelAwareCacheMockRecorder) GetWithLevel(key any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetWithLevel", reflect.TypeOf((*MockLevelAwareCache)(nil).GetWithLevel), key)
}
// Set mocks base method.
func (m *MockLevelAwareCache) Set(key string, val []byte, ttl models.TTL) {
m.ctrl.T.Helper()
m.ctrl.Call(m, "Set", key, val, ttl)
}
// Set indicates an expected call of Set.
func (mr *MockLevelAwareCacheMockRecorder) Set(key, val, ttl any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Set", reflect.TypeOf((*MockLevelAwareCache)(nil).Set), key, val, ttl)
}
go-proxy-cache/internal/interfaces/mock/cache_rules_config.go
+14
View File
@@ -41,6 +41,20 @@ func (m *MockCacheRulesConfig) EXPECT() *MockCacheRulesConfigMockRecorder {
return m.recorder
}
// GetAllMethods mocks base method.
func (m *MockCacheRulesConfig) GetAllMethods() []string {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetAllMethods")
ret0, _ := ret[0].([]string)
return ret0
}
// GetAllMethods indicates an expected call of GetAllMethods.
func (mr *MockCacheRulesConfigMockRecorder) GetAllMethods() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllMethods", reflect.TypeOf((*MockCacheRulesConfig)(nil).GetAllMethods))
}
// GetCacheTypeForMethod mocks base method.
func (m *MockCacheRulesConfig) GetCacheTypeForMethod(method string) models.CacheType {
m.ctrl.T.Helper()
go-proxy-cache/internal/metrics/metrics.go
+160 -42
View File
@@ -1,67 +1,116 @@
package metrics
import (
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
// White list of allowed RPC methods to prevent cardinality explosion
// Initialized from cache_rules.yaml via InitializeAllowedMethods()
var allowedMethods map[string]bool
var (
// Core request/hit/miss counters
// Core request/hit/miss counters with unified labels
CacheRequests = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "cache_requests_total",
Name: "eth_rpc_proxy_cache_requests_total",
Help: "Total number of cache requests",
},
[]string{"cache_type"},
[]string{"cache_type", "level", "network", "rpc_method"},
)
CacheHits = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "cache_hits_total",
Name: "eth_rpc_proxy_cache_hits_total",
Help: "Total number of cache hits",
},
[]string{"cache_type"},
[]string{"cache_type", "level", "network", "rpc_method"},
)
CacheMisses = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "cache_misses_total",
Name: "eth_rpc_proxy_cache_misses_total",
Help: "Total number of cache misses",
},
[]string{"cache_type"},
[]string{"cache_type", "level", "network", "rpc_method"},
)
// L1/L2 specific hits (separate counters for simplicity)
L1CacheHits = promauto.NewCounterVec(
// New metrics for enhanced monitoring
CacheSets = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "l1_cache_hits_total",
Help: "Total number of L1 cache hits",
Name: "eth_rpc_proxy_cache_sets_total",
Help: "Total number of cache set operations",
},
[]string{"cache_type"},
[]string{"level", "cache_type", "network"},
)
L2CacheHits = promauto.NewCounterVec(
CacheEvictions = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "l2_cache_hits_total",
Help: "Total number of L2 cache hits",
Name: "eth_rpc_proxy_cache_evictions_total",
Help: "Total number of cache evictions",
},
[]string{"cache_type"},
[]string{"level", "cache_type", "network"},
)
// Get operation latency only
CacheErrors = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "eth_rpc_proxy_cache_errors_total",
Help: "Cache errors by kind",
},
[]string{"level", "kind"},
)
CacheBytesRead = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "eth_rpc_proxy_cache_bytes_read_total",
Help: "Bytes read from cache",
},
[]string{"level", "cache_type", "network"},
)
CacheBytesWritten = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "eth_rpc_proxy_cache_bytes_written_total",
Help: "Bytes written to cache",
},
[]string{"level", "cache_type", "network"},
)
// Extended operation latency for get and set operations
CacheOperationDuration = promauto.NewHistogramVec(
prometheus.HistogramOpts{
Name: "cache_operation_duration_seconds",
Help: "Duration of cache get operations",
Name: "eth_rpc_proxy_cache_operation_duration_seconds",
Help: "Duration of cache operations",
Buckets: prometheus.DefBuckets,
},
[]string{"operation", "level"}, // simplified labels
[]string{"operation", "level"}, // operation: get|set, level: l1|l2|multi
)
// Cache keys count (mainly for L1)
CacheKeys = promauto.NewGaugeVec(
prometheus.GaugeOpts{
Name: "eth_rpc_proxy_cache_keys",
Help: "Current number of keys in cache",
},
[]string{"level"},
)
// Cache item age at hit time (for TTL effectiveness analysis)
CacheItemAge = promauto.NewHistogramVec(
prometheus.HistogramOpts{
Name: "eth_rpc_proxy_cache_item_age_seconds",
Help: "Age of item at hit time",
Buckets: []float64{0.1, 0.5, 1, 2, 5, 10, 30, 60, 120, 300, 600, 1800, 3600}, // up to 1 hour
},
[]string{"level", "cache_type"},
)
// L1 capacity metrics only (if L1 is in-memory)
CacheCapacity = promauto.NewGaugeVec(
prometheus.GaugeOpts{
Name: "cache_capacity_bytes",
Name: "eth_rpc_proxy_cache_capacity_bytes",
Help: "L1 cache capacity in bytes",
},
[]string{"level"}, // only "l1"
@@ -69,45 +118,114 @@ var (
CacheUsed = promauto.NewGaugeVec(
prometheus.GaugeOpts{
Name: "cache_used_bytes",
Name: "eth_rpc_proxy_cache_used_bytes",
Help: "L1 cache used space in bytes",
},
[]string{"level"}, // only "l1"
)
)
// RecordCacheRequest records a cache request
func RecordCacheRequest(cacheType string) {
CacheRequests.WithLabelValues(cacheType).Inc()
}
// InitializeAllowedMethods initializes the allowed methods whitelist from cache rules
func InitializeAllowedMethods(methods []string) {
allowedMethods = make(map[string]bool)
// RecordCacheHit records a cache hit
func RecordCacheHit(cacheType string, level string) {
CacheHits.WithLabelValues(cacheType).Inc()
switch level {
case "l1":
L1CacheHits.WithLabelValues(cacheType).Inc()
case "l2":
L2CacheHits.WithLabelValues(cacheType).Inc()
// Add all configured methods to whitelist
for _, method := range methods {
allowedMethods[method] = true
}
}
// RecordCacheMiss records a cache miss
func RecordCacheMiss(cacheType string) {
CacheMisses.WithLabelValues(cacheType).Inc()
// normalizeRPCMethod returns the method name if it's in the whitelist, otherwise "other"
func normalizeRPCMethod(method string) string {
if allowedMethods != nil && allowedMethods[method] {
return method
}
return "other"
}
// UpdateL1CacheCapacity updates L1 cache capacity metrics only
// normalizeNetwork creates a network identifier from chain and network
func normalizeNetwork(chain, network string) string {
if chain == "" || network == "" {
return "unknown"
}
return chain + ":" + network
}
// RecordCacheHit records a cache hit with enhanced labels and age tracking
func RecordCacheHit(cacheType, level, chain, network, rpcMethod string, itemAge time.Duration) {
normalizedNetwork := normalizeNetwork(chain, network)
normalizedMethod := normalizeRPCMethod(rpcMethod)
// Record request and hit with proper level
CacheRequests.WithLabelValues(cacheType, level, normalizedNetwork, normalizedMethod).Inc()
CacheHits.WithLabelValues(cacheType, level, normalizedNetwork, normalizedMethod).Inc()
// Record item age for TTL effectiveness analysis
if itemAge > 0 {
CacheItemAge.WithLabelValues(level, cacheType).Observe(itemAge.Seconds())
}
}
// RecordCacheMiss records a cache miss with enhanced labels
func RecordCacheMiss(cacheType, chain, network, rpcMethod string) {
normalizedNetwork := normalizeNetwork(chain, network)
normalizedMethod := normalizeRPCMethod(rpcMethod)
// For miss, we don't know the level, so we use "miss" as level
// This represents requests that didn't hit any cache level
CacheRequests.WithLabelValues(cacheType, "miss", normalizedNetwork, normalizedMethod).Inc()
CacheMisses.WithLabelValues(cacheType, "miss", normalizedNetwork, normalizedMethod).Inc()
}
// RecordCacheSet records a cache set operation with size tracking
func RecordCacheSet(level, cacheType, chain, network string, dataSize int) {
normalizedNetwork := normalizeNetwork(chain, network)
CacheSets.WithLabelValues(level, cacheType, normalizedNetwork).Inc()
if dataSize > 0 {
CacheBytesWritten.WithLabelValues(level, cacheType, normalizedNetwork).Add(float64(dataSize))
}
}
// RecordCacheEviction records a cache eviction
func RecordCacheEviction(level, cacheType, chain, network string) {
normalizedNetwork := normalizeNetwork(chain, network)
CacheEvictions.WithLabelValues(level, cacheType, normalizedNetwork).Inc()
}
// RecordCacheError records a cache error
func RecordCacheError(level, kind string) {
CacheErrors.WithLabelValues(level, kind).Inc()
}
// RecordCacheBytesRead records bytes read from cache
func RecordCacheBytesRead(level, cacheType, chain, network string, bytesRead int) {
if bytesRead > 0 {
normalizedNetwork := normalizeNetwork(chain, network)
CacheBytesRead.WithLabelValues(level, cacheType, normalizedNetwork).Add(float64(bytesRead))
}
}
// UpdateL1CacheCapacity updates L1 cache capacity metrics
func UpdateL1CacheCapacity(capacity, used int64) {
CacheCapacity.WithLabelValues("l1").Set(float64(capacity))
CacheUsed.WithLabelValues("l1").Set(float64(used))
}
// TimeCacheGetOperation returns a timer function for measuring cache get operation duration
func TimeCacheGetOperation(level string) func() {
timer := prometheus.NewTimer(CacheOperationDuration.WithLabelValues("get", level))
// UpdateCacheKeys updates the number of keys in cache
func UpdateCacheKeys(level string, count int64) {
CacheKeys.WithLabelValues(level).Set(float64(count))
}
// TimeCacheOperation returns a timer function for measuring cache operation duration
func TimeCacheOperation(operation, level string) func() {
timer := prometheus.NewTimer(CacheOperationDuration.WithLabelValues(operation, level))
return func() {
timer.ObserveDuration()
}
}
// TimeCacheGetOperation returns a timer function for measuring cache get operation duration (backward compatibility)
func TimeCacheGetOperation(level string) func() {
return TimeCacheOperation("get", level)
}
go-proxy-cache/internal/metrics/metrics_test.go
+54 -9
View File
@@ -2,26 +2,32 @@ package metrics
import (
"testing"
"time"
)
func TestCacheMetrics(t *testing.T) {
// Note: Metrics are now package-level variables, automatically registered
// This test just verifies the functions don't panic
t.Run("RecordCacheRequest", func(t *testing.T) {
// This should not panic
RecordCacheRequest("permanent")
})
t.Run("RecordCacheHit", func(t *testing.T) {
// This should not panic
RecordCacheHit("permanent", "l1")
RecordCacheHit("permanent", "l2")
RecordCacheHit("permanent", "l1", "ethereum", "mainnet", "eth_getBlockByHash", time.Second*30)
RecordCacheHit("permanent", "l2", "ethereum", "mainnet", "eth_getBlockByHash", time.Second*60)
})
t.Run("RecordCacheMiss", func(t *testing.T) {
// This should not panic
RecordCacheMiss("permanent")
RecordCacheMiss("permanent", "ethereum", "mainnet", "eth_getBlockByHash")
})
t.Run("RecordCacheSet", func(t *testing.T) {
// This should not panic
RecordCacheSet("l1", "permanent", "ethereum", "mainnet", 1024)
})
t.Run("RecordCacheError", func(t *testing.T) {
// This should not panic
RecordCacheError("l1", "encode")
})
t.Run("UpdateL1CacheCapacity", func(t *testing.T) {
@@ -29,9 +35,48 @@ func TestCacheMetrics(t *testing.T) {
UpdateL1CacheCapacity(1000000, 500000)
})
t.Run("TimeCacheGetOperation", func(t *testing.T) {
t.Run("UpdateCacheKeys", func(t *testing.T) {
// This should not panic
UpdateCacheKeys("l1", 1000)
})
t.Run("TimeCacheOperation", func(t *testing.T) {
// This should not panic
timer := TimeCacheOperation("get", "l1")
timer() // Call the returned function
})
t.Run("TimeCacheGetOperation", func(t *testing.T) {
// This should not panic (backward compatibility)
timer := TimeCacheGetOperation("l1")
timer() // Call the returned function
})
t.Run("NormalizeRPCMethod", func(t *testing.T) {
// Initialize test methods
testMethods := []string{"eth_getBlockByHash", "eth_call", "net_version"}
InitializeAllowedMethods(testMethods)
// Test whitelisted method
if normalizeRPCMethod("eth_getBlockByHash") != "eth_getBlockByHash" {
t.Error("Expected whitelisted method to be preserved")
}
// Test non-whitelisted method
if normalizeRPCMethod("custom_method") != "other" {
t.Error("Expected non-whitelisted method to be normalized to 'other'")
}
})
t.Run("NormalizeNetwork", func(t *testing.T) {
// Test valid network
if normalizeNetwork("ethereum", "mainnet") != "ethereum:mainnet" {
t.Error("Expected valid network to be formatted correctly")
}
// Test empty network
if normalizeNetwork("", "") != "unknown" {
t.Error("Expected empty network to be normalized to 'unknown'")
}
})
}
go-proxy-cache/internal/scheduler/scheduler.go
+76
View File
@@ -0,0 +1,76 @@
package scheduler
import (
"context"
"sync"
"time"
)
// Scheduler manages a background task that runs at regular intervals
type Scheduler struct {
interval time.Duration
task func()
cancel context.CancelFunc
wg sync.WaitGroup
mu sync.Mutex
running bool
}
// New creates a new Scheduler instance
func New(interval time.Duration, task func()) *Scheduler {
return &Scheduler{
interval: interval,
task: task,
}
}
// Start begins executing the task at the specified interval
func (s *Scheduler) Start() {
s.mu.Lock()
defer s.mu.Unlock()
if s.running {
return
}
ctx, cancel := context.WithCancel(context.Background())
s.cancel = cancel
s.running = true
s.wg.Add(1)
go func() {
defer s.wg.Done()
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
s.task()
case <-ctx.Done():
return
}
}
}()
}
// Stop terminates the periodic task execution
func (s *Scheduler) Stop() {
s.mu.Lock()
defer s.mu.Unlock()
if !s.running {
return
}
s.cancel()
s.wg.Wait()
s.running = false
}
// IsRunning returns true if the task is currently running
func (s *Scheduler) IsRunning() bool {
s.mu.Lock()
defer s.mu.Unlock()
return s.running
}
grafana/provisioning/dashboards/cache-metrics.json
+1246
View File
File diff suppressed because it is too large Load Diff
grafana/provisioning/dashboards/rpc-metrics.json
+77 -363
View File
@@ -298,7 +298,10 @@
"id": 4,
"options": {
"legend": {
"calcs": ["mean", "max"],
"calcs": [
"mean",
"max"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
@@ -369,7 +372,9 @@
"id": 5,
"options": {
"legend": {
"calcs": ["lastNotNull"],
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
@@ -440,7 +445,9 @@
"id": 6,
"options": {
"legend": {
"calcs": ["lastNotNull"],
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
@@ -472,14 +479,28 @@
},
"thresholds": {
"steps": [
{"color": "green", "value": null},
{"color": "yellow", "value": 100},
{"color": "red", "value": 1000}
{
"color": "green",
"value": null
},
{
"color": "yellow",
"value": 100
},
{
"color": "red",
"value": 1000
}
]
}
}
},
"gridPos": {"h": 6, "w": 6, "x": 0, "y": 32},
"gridPos": {
"h": 6,
"w": 6,
"x": 0,
"y": 32
},
"id": 7,
"title": "Auth: Tokens Issued",
"type": "stat",
@@ -511,7 +532,12 @@
}
}
},
"gridPos": {"h": 6, "w": 9, "x": 6, "y": 32},
"gridPos": {
"h": 6,
"w": 9,
"x": 6,
"y": 32
},
"id": 8,
"title": "Auth: Token Issuance Rate",
"type": "timeseries",
@@ -535,14 +561,28 @@
},
"thresholds": {
"steps": [
{"color": "red", "value": null},
{"color": "yellow", "value": 50},
{"color": "green", "value": 80}
{
"color": "red",
"value": null
},
{
"color": "yellow",
"value": 50
},
{
"color": "green",
"value": 80
}
]
}
}
},
"gridPos": {"h": 6, "w": 9, "x": 15, "y": 32},
"gridPos": {
"h": 6,
"w": 9,
"x": 15,
"y": 32
},
"id": 9,
"title": "Auth: Success Rate",
"type": "stat",
@@ -574,7 +614,12 @@
}
}
},
"gridPos": {"h": 6, "w": 12, "x": 0, "y": 38},
"gridPos": {
"h": 6,
"w": 12,
"x": 0,
"y": 38
},
"id": 10,
"title": "Auth: Puzzle Attempts by Status",
"type": "timeseries",
@@ -606,7 +651,12 @@
}
}
},
"gridPos": {"h": 6, "w": 12, "x": 12, "y": 38},
"gridPos": {
"h": 6,
"w": 12,
"x": 12,
"y": 38
},
"id": 11,
"title": "Auth: Token Verifications",
"type": "timeseries",
@@ -763,7 +813,10 @@
"id": 13,
"options": {
"legend": {
"calcs": ["mean", "lastNotNull"],
"calcs": [
"mean",
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
@@ -891,7 +944,10 @@
"id": 14,
"options": {
"legend": {
"calcs": ["mean", "lastNotNull"],
"calcs": [
"mean",
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
@@ -1024,7 +1080,10 @@
"id": 15,
"options": {
"legend": {
"calcs": ["mean", "lastNotNull"],
"calcs": [
"mean",
"lastNotNull"
],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
@@ -1068,351 +1127,6 @@
"refId": "F"
}
]
},
{
"datasource": {
"type": "prometheus",
"uid": "PBFA97CFB590B2093"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"vis": false
},
"lineInterpolation": "linear",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 80
}
]
},
"unit": "short"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 48
},
"id": 15,
"options": {
"legend": {
"calcs": [],
"displayMode": "list",
"placement": "bottom"
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "redis_connected_clients",
"legendFormat": "Connected Clients",
"refId": "A"
},
{
"expr": "redis_commands_processed_total",
"legendFormat": "Commands Processed",
"refId": "B"
}
],
"title": "KeyDB Connection Stats",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "PBFA97CFB590B2093"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"vis": false
},
"lineInterpolation": "linear",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 80
}
]
},
"unit": "bytes"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 48
},
"id": 16,
"options": {
"legend": {
"calcs": [],
"displayMode": "list",
"placement": "bottom"
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "redis_memory_used_bytes",
"legendFormat": "Memory Used",
"refId": "A"
},
{
"expr": "redis_memory_used_rss_bytes",
"legendFormat": "RSS Memory",
"refId": "B"
}
],
"title": "KeyDB Memory Usage",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "PBFA97CFB590B2093"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "thresholds"
},
"mappings": [
{
"options": {
"0": {
"color": "red",
"text": "Down"
},
"1": {
"color": "green",
"text": "Up"
}
},
"type": "value"
}
],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "red",
"value": null
},
{
"color": "green",
"value": 1
}
]
}
},
"overrides": []
},
"gridPos": {
"h": 4,
"w": 6,
"x": 0,
"y": 56
},
"id": 17,
"options": {
"colorMode": "background",
"graphMode": "none",
"justifyMode": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"textMode": "auto"
},
"pluginVersion": "9.0.0",
"targets": [
{
"expr": "up{job=\"keydb\"}",
"legendFormat": "KeyDB Status",
"refId": "A"
}
],
"title": "KeyDB Status",
"type": "stat"
},
{
"datasource": {
"type": "prometheus",
"uid": "PBFA97CFB590B2093"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"vis": false
},
"lineInterpolation": "linear",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 80
}
]
},
"unit": "short"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 18,
"x": 6,
"y": 56
},
"id": 18,
"options": {
"legend": {
"calcs": [],
"displayMode": "list",
"placement": "bottom"
},
"tooltip": {
"mode": "single",
"sort": "none"
}
},
"targets": [
{
"expr": "rate(redis_commands_processed_total[5m])",
"legendFormat": "Commands/sec",
"refId": "A"
},
{
"expr": "redis_keyspace_hits_total",
"legendFormat": "Keyspace Hits",
"refId": "B"
},
{
"expr": "redis_keyspace_misses_total",
"legendFormat": "Keyspace Misses",
"refId": "C"
}
],
"title": "KeyDB Performance Metrics",
"type": "timeseries"
}
],
"refresh": "10s",
@@ -1432,4 +1146,4 @@
"uid": "rpc-health-metrics",
"version": 1,
"weekStart": ""
}
}
prometheus-alerts.yml
+119
View File
@@ -0,0 +1,119 @@
groups:
- name: cache_alerts
rules:
# 1. Hit ratio degradation
- alert: CacheHitRatioLow
expr: sum(rate(cache_hits_total[10m])) / sum(rate(cache_requests_total[10m])) < 0.6
for: 10m
labels:
severity: warning
component: cache
annotations:
summary: "Cache hit ratio has degraded"
description: "Cache hit ratio is {{ $value | humanizePercentage }}, which is below the 60% threshold. This may indicate cache configuration issues or increased cache pressure."
# 2. High L1 cache latency
- alert: CacheL1LatencyHigh
expr: histogram_quantile(0.95, sum by (le) (rate(cache_operation_duration_seconds_bucket{operation="get",level="l1"}[5m]))) > 0.005
for: 10m
labels:
severity: warning
component: cache
level: l1
annotations:
summary: "L1 cache latency is high"
description: "L1 cache p95 latency is {{ $value | humanizeDuration }}, which is above 5ms threshold. This may indicate memory pressure or cache contention."
# 3. High L2 cache latency
- alert: CacheL2LatencyHigh
expr: histogram_quantile(0.95, sum by (le) (rate(cache_operation_duration_seconds_bucket{operation="get",level="l2"}[5m]))) > 0.050
for: 10m
labels:
severity: warning
component: cache
level: l2
annotations:
summary: "L2 cache latency is high"
description: "L2 cache p95 latency is {{ $value | humanizeDuration }}, which is above 50ms threshold. This may indicate Redis/KeyDB performance issues."
# 4. High eviction rate
- alert: CacheEvictionsHigh
expr: sum(rate(cache_evictions_total[5m])) > 100
for: 5m
labels:
severity: warning
component: cache
annotations:
summary: "Cache eviction rate is high"
description: "Cache evictions are occurring at {{ $value }} per second, which may indicate insufficient cache capacity or suboptimal TTL settings."
# 5. High L1 cache usage
- alert: CacheL1UsageHigh
expr: 100 * (cache_used_bytes{level="l1"} / cache_capacity_bytes{level="l1"}) > 90
for: 15m
labels:
severity: critical
component: cache
level: l1
annotations:
summary: "L1 cache usage is critically high"
description: "L1 cache usage is {{ $value }}%, which is above 90% threshold. Consider increasing cache size or reviewing TTL policies."
# 6. Cache errors occurring
- alert: CacheErrorsDetected
expr: sum(rate(cache_errors_total[5m])) > 0
for: 5m
labels:
severity: warning
component: cache
annotations:
summary: "Cache errors detected"
description: "Cache errors are occurring at {{ $value }} per second. Check cache system health and connectivity."
# 9. Cache hit ratio by level (L1 should be higher than L2)
- alert: CacheL1HitRatioLow
expr: sum(rate(cache_hits_total{level="l1"}[10m])) / sum(rate(cache_requests_total{level="l1"}[10m])) < 0.3
for: 15m
labels:
severity: warning
component: cache
level: l1
annotations:
summary: "L1 cache hit ratio is low"
description: "L1 cache hit ratio is {{ $value | humanizePercentage }}, indicating potential L1 cache sizing or configuration issues."
# 10. Bytes read/write imbalance (more reads than writes might indicate good caching)
- alert: CacheBytesImbalance
expr: sum(rate(cache_bytes_read_total[10m])) / sum(rate(cache_bytes_written_total[10m])) < 2
for: 30m
labels:
severity: info
component: cache
annotations:
summary: "Cache read/write ratio is low"
description: "Cache read/write ratio is {{ $value }}, which might indicate suboptimal cache utilization. Expected ratio should be higher for effective caching."
- name: network_specific_alerts
rules:
# Network-specific hit ratio alerts
- alert: NetworkCacheHitRatioLow
expr: sum by (network) (rate(cache_hits_total[10m])) / sum by (network) (rate(cache_requests_total[10m])) < 0.4
for: 15m
labels:
severity: warning
component: cache
annotations:
summary: "Cache hit ratio low for network {{ $labels.network }}"
description: "Cache hit ratio for network {{ $labels.network }} is {{ $value | humanizePercentage }}, which is below expected levels."
# Method-specific alerts for critical RPC methods
- alert: CriticalMethodCacheHitRatioLow
expr: sum by (rpc_method) (rate(cache_hits_total{rpc_method=~"eth_getBlockByHash|eth_getTransactionReceipt|eth_getLogs"}[10m])) / sum by (rpc_method) (rate(cache_requests_total{rpc_method=~"eth_getBlockByHash|eth_getTransactionReceipt|eth_getLogs"}[10m])) < 0.8
for: 15m
labels:
severity: warning
component: cache
annotations:
summary: "Low cache hit ratio for critical method {{ $labels.rpc_method }}"
description: "Cache hit ratio for {{ $labels.rpc_method }} is {{ $value | humanizePercentage }}. This method should have high cache effectiveness."
prometheus.yml
+5 -2
View File
@@ -1,6 +1,9 @@
global:
scrape_interval: 15s
rule_files:
- "prometheus-alerts.yml"
scrape_configs:
- job_name: 'health-checker'
static_configs:
@@ -23,9 +26,9 @@ scrape_configs:
metrics_path: '/metrics/cache'
scheme: http
- job_name: 'keydb'
- job_name: 'cache-service'
static_configs:
- targets: ['keydb-exporter:9121']
- targets: ['cache-service:8099']
scrape_interval: 15s
metrics_path: '/metrics'
scheme: http