feat(backend): Redis persistence for metrics (seed + 5min flush + graceful shutdown)

- seedMetricsFromRedis: loads persisted counters into atomics on startup
- flushMetricsToRedis: writes all 12 counters to 3 Redis hashes on demand
- startMetricsFlusher: goroutine that flushes every 5 minutes
- Graceful shutdown: SIGINT/SIGTERM flushes metrics then drains HTTP in 10s
- http.ListenAndServe replaced with http.Server for Shutdown support

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Shekhar Vaidya 2026-06-21 20:48:08 +05:30
parent cbe5ca2f9b
commit 470abbc94b

View file

@ -12,12 +12,14 @@ import (
"net/http"
"net/url"
"os"
"os/signal"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
"github.com/google/uuid"
@ -307,6 +309,7 @@ func initRedis() {
}
rdb = client
log.Println("redis: connected")
seedMetricsFromRedis()
}
func redisWriteLink(productID, skuID string, raw []byte, expiresAt time.Time) {
@ -369,6 +372,89 @@ func redisSeedLinkCache() {
}
}
// seedMetricsFromRedis loads persisted metric counters from Redis into the
// in-memory atomic vars. Called once on startup after Redis connects.
func seedMetricsFromRedis() {
if rdb == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
loadHash := func(key string) map[string]string {
vals, err := rdb.HGetAll(ctx, key).Result()
if err != nil {
return nil
}
return vals
}
parseI64 := func(m map[string]string, field string) int64 {
if m == nil {
return 0
}
v, _ := strconv.ParseInt(m[field], 10, 64)
return v
}
sku := loadHash("msdl:metrics:sku")
atomic.StoreInt64(&mSkuRequests, parseI64(sku, "requests"))
atomic.StoreInt64(&mSkuCacheHits, parseI64(sku, "cache_hits"))
atomic.StoreInt64(&mSkuFetches, parseI64(sku, "ms_fetches"))
atomic.StoreInt64(&mSkuNegHits, parseI64(sku, "neg_hits"))
link := loadHash("msdl:metrics:link")
atomic.StoreInt64(&mLinkRequests, parseI64(link, "requests"))
atomic.StoreInt64(&mLinkCacheHits, parseI64(link, "cache_hits"))
atomic.StoreInt64(&mLinkFetches, parseI64(link, "ms_fetches"))
atomic.StoreInt64(&mLinkNegHits, parseI64(link, "neg_hits"))
atomic.StoreInt64(&mLinkStale, parseI64(link, "stale"))
eval := loadHash("msdl:metrics:eval")
atomic.StoreInt64(&mEvalRequests, parseI64(eval, "requests"))
atomic.StoreInt64(&mEvalCacheHits, parseI64(eval, "cache_hits"))
atomic.StoreInt64(&mEvalStale, parseI64(eval, "stale"))
log.Println("redis: seeded in-memory metrics from Redis")
}
// flushMetricsToRedis writes current in-memory metric counters to Redis.
// Called periodically (every 5 min) and on graceful shutdown.
func flushMetricsToRedis() {
if rdb == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
rdb.HSet(ctx, "msdl:metrics:sku",
"requests", atomic.LoadInt64(&mSkuRequests),
"cache_hits", atomic.LoadInt64(&mSkuCacheHits),
"ms_fetches", atomic.LoadInt64(&mSkuFetches),
"neg_hits", atomic.LoadInt64(&mSkuNegHits),
)
rdb.HSet(ctx, "msdl:metrics:link",
"requests", atomic.LoadInt64(&mLinkRequests),
"cache_hits", atomic.LoadInt64(&mLinkCacheHits),
"ms_fetches", atomic.LoadInt64(&mLinkFetches),
"neg_hits", atomic.LoadInt64(&mLinkNegHits),
"stale", atomic.LoadInt64(&mLinkStale),
)
rdb.HSet(ctx, "msdl:metrics:eval",
"requests", atomic.LoadInt64(&mEvalRequests),
"cache_hits", atomic.LoadInt64(&mEvalCacheHits),
"stale", atomic.LoadInt64(&mEvalStale),
)
log.Println("redis: flushed in-memory metrics to Redis")
}
// startMetricsFlusher flushes in-memory metrics to Redis every 5 minutes.
func startMetricsFlusher() {
ticker := time.NewTicker(5 * time.Minute)
for range ticker.C {
flushMetricsToRedis()
}
}
// --- /contribute validation helpers ---
// validContributeProducts is the canonical product ID allow-list for POST /contribute.
@ -1606,20 +1692,20 @@ func main() {
go cleanupSessions()
go cleanupCaches()
go warmEvalCache()
go startMetricsFlusher()
mux := http.NewServeMux()
mux.HandleFunc("/skuinfo", handleSkuInfo)
mux.HandleFunc("/proxy", handleProxy)
mux.HandleFunc("/contribute", handleContribute)
mux.HandleFunc("/evallinks", handleEvalLinks)
mux.HandleFunc("/metrics", handleMetrics)
mux.HandleFunc("/telemetry", handleTelemetry)
mux.HandleFunc("/cli/version", handleCLIVersion)
mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
w.Write([]byte(`{"status":"ok"}`))
})
mux.HandleFunc("/cli/version", handleCLIVersion)
mux.HandleFunc("/telemetry", handleTelemetry)
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/" {
http.NotFound(w, r)
@ -1629,10 +1715,21 @@ func main() {
})
handler := enableCORS(mux)
srv := &http.Server{Addr: PORT, Handler: handler}
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
go func() {
<-sigCh
log.Println("shutdown: flushing metrics to Redis...")
flushMetricsToRedis()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
srv.Shutdown(ctx)
}()
log.Printf("Go Backend running on http://localhost%s\n", PORT)
err := http.ListenAndServe(PORT, handler)
if err != nil {
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatal(err)
}
}