pike
50ca0a071c
Operators could see attach-ready and capacity-pressure events but not how full the configured disk (or memory) tier was without reading filesystems. Expose size/capacity gauges and disk_cache_full_ratio next to disk_tier_ready. Capacity is a config read, so it stays available while attach is pending.
443 lines
17 KiB
Go
443 lines
17 KiB
Go
// steamcache/metrics/metrics.go
|
|
package metrics
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"s1d3sw1ped/steamcache2/steamcache/logger"
|
|
)
|
|
|
|
// Metrics tracks various performance and operational metrics
|
|
type Metrics struct {
|
|
// Request metrics
|
|
TotalRequests int64
|
|
CacheHits int64
|
|
CacheMisses int64
|
|
CacheCoalesced int64
|
|
NegativeCacheHits int64 // 404/410 served from a still-valid negative cache entry
|
|
RangeCache int64 // Range requests served as 206 from an already-cached object (HIT)
|
|
RangeUpstream int64 // Range requests that required an upstream fetch (full object), served as 206
|
|
Errors int64
|
|
RateLimited int64
|
|
|
|
// Performance metrics
|
|
TotalResponseTime int64 // in nanoseconds
|
|
TotalBytesServed int64
|
|
TotalBytesSaved int64 // bytes served from cache instead of being re-downloaded from upstream
|
|
|
|
// Cache metrics
|
|
MemoryCacheSize int64
|
|
DiskCacheSize int64
|
|
MemoryCacheCapacity int64 // configured memory capacity (bytes)
|
|
DiskCacheCapacity int64 // configured disk capacity (bytes); 0 when no disk
|
|
MemoryCacheHits int64
|
|
DiskCacheHits int64
|
|
Promotions int64
|
|
Evictions int64
|
|
CapacityPressureEvents int64 // soft eviction under cap and/or disk ENOSPC
|
|
DiskTierReady int64 // 0=pending (or unset), 1=ready or no-disk (N/A)
|
|
|
|
// Expanded observability (upstream breakdowns, cache write failures, per-service errors)
|
|
UpstreamErrors int64
|
|
CacheWriteFailures int64
|
|
ServiceErrors map[string]int64
|
|
serviceErrorsMutex sync.RWMutex
|
|
|
|
// Service metrics
|
|
ServiceRequests map[string]int64
|
|
serviceMutex sync.RWMutex
|
|
|
|
// Time tracking
|
|
StartTime time.Time
|
|
LastResetTime time.Time
|
|
}
|
|
|
|
// NewMetrics creates a new metrics instance
|
|
func NewMetrics() *Metrics {
|
|
now := time.Now()
|
|
return &Metrics{
|
|
ServiceRequests: make(map[string]int64),
|
|
ServiceErrors: make(map[string]int64),
|
|
StartTime: now,
|
|
LastResetTime: now,
|
|
}
|
|
}
|
|
|
|
// IncrementTotalRequests increments the total request counter
|
|
func (m *Metrics) IncrementTotalRequests() {
|
|
atomic.AddInt64(&m.TotalRequests, 1)
|
|
}
|
|
|
|
// IncrementCacheHits increments the cache hit counter
|
|
func (m *Metrics) IncrementCacheHits() {
|
|
atomic.AddInt64(&m.CacheHits, 1)
|
|
}
|
|
|
|
// IncrementCacheMisses increments the cache miss counter
|
|
func (m *Metrics) IncrementCacheMisses() {
|
|
atomic.AddInt64(&m.CacheMisses, 1)
|
|
}
|
|
|
|
// IncrementNegativeCacheHits increments hits served from a 404/410 negative entry.
|
|
func (m *Metrics) IncrementNegativeCacheHits() {
|
|
atomic.AddInt64(&m.NegativeCacheHits, 1)
|
|
}
|
|
|
|
// IncrementCacheCoalesced increments the coalesced request counter
|
|
func (m *Metrics) IncrementCacheCoalesced() {
|
|
atomic.AddInt64(&m.CacheCoalesced, 1)
|
|
}
|
|
|
|
// IncrementRangeCache increments the Range-from-cache counter (HIT served as 206)
|
|
func (m *Metrics) IncrementRangeCache() {
|
|
atomic.AddInt64(&m.RangeCache, 1)
|
|
}
|
|
|
|
// IncrementRangeUpstream increments the Range-from-upstream counter (full object
|
|
// fetched upstream, requested slice served as 206)
|
|
func (m *Metrics) IncrementRangeUpstream() {
|
|
atomic.AddInt64(&m.RangeUpstream, 1)
|
|
}
|
|
|
|
// IncrementErrors increments the error counter
|
|
func (m *Metrics) IncrementErrors() {
|
|
atomic.AddInt64(&m.Errors, 1)
|
|
}
|
|
|
|
// IncrementRateLimited increments the rate limited counter
|
|
func (m *Metrics) IncrementRateLimited() {
|
|
atomic.AddInt64(&m.RateLimited, 1)
|
|
}
|
|
|
|
// AddResponseTime adds response time to the total
|
|
func (m *Metrics) AddResponseTime(duration time.Duration) {
|
|
atomic.AddInt64(&m.TotalResponseTime, int64(duration))
|
|
}
|
|
|
|
// AddBytesServed adds bytes served to the total
|
|
func (m *Metrics) AddBytesServed(bytes int64) {
|
|
atomic.AddInt64(&m.TotalBytesServed, bytes)
|
|
}
|
|
|
|
// AddBytesSaved records bytes that were served from the cache instead of being
|
|
// fetched again from the upstream (the main value metric for a cache).
|
|
func (m *Metrics) AddBytesSaved(bytes int64) {
|
|
atomic.AddInt64(&m.TotalBytesSaved, bytes)
|
|
}
|
|
|
|
// SetMemoryCacheSize sets the current memory cache size
|
|
func (m *Metrics) SetMemoryCacheSize(size int64) {
|
|
atomic.StoreInt64(&m.MemoryCacheSize, size)
|
|
}
|
|
|
|
// SetDiskCacheSize sets the current disk cache size
|
|
func (m *Metrics) SetDiskCacheSize(size int64) {
|
|
atomic.StoreInt64(&m.DiskCacheSize, size)
|
|
}
|
|
|
|
// SetMemoryCacheCapacity sets the configured memory cache capacity in bytes.
|
|
func (m *Metrics) SetMemoryCacheCapacity(capacity int64) {
|
|
atomic.StoreInt64(&m.MemoryCacheCapacity, capacity)
|
|
}
|
|
|
|
// SetDiskCacheCapacity sets the configured disk cache capacity in bytes
|
|
// (0 when no disk is configured).
|
|
func (m *Metrics) SetDiskCacheCapacity(capacity int64) {
|
|
atomic.StoreInt64(&m.DiskCacheCapacity, capacity)
|
|
}
|
|
|
|
// SetDiskTierReady sets whether the disk slow tier is attached (1) or still pending (0).
|
|
// Memory-only (no disk) also uses 1 — meaning "not waiting on disk attach". Reset does not clear this.
|
|
func (m *Metrics) SetDiskTierReady(ready int64) {
|
|
atomic.StoreInt64(&m.DiskTierReady, ready)
|
|
}
|
|
|
|
// GetDiskTierReady returns 1 if disk tier is ready (or no disk configured), else 0 while attach pending.
|
|
func (m *Metrics) GetDiskTierReady() int64 {
|
|
return atomic.LoadInt64(&m.DiskTierReady)
|
|
}
|
|
|
|
// IncrementMemoryCacheHits increments memory cache hits
|
|
func (m *Metrics) IncrementMemoryCacheHits() {
|
|
atomic.AddInt64(&m.MemoryCacheHits, 1)
|
|
}
|
|
|
|
// IncrementDiskCacheHits increments disk cache hits
|
|
func (m *Metrics) IncrementDiskCacheHits() {
|
|
atomic.AddInt64(&m.DiskCacheHits, 1)
|
|
}
|
|
|
|
// IncrementServiceRequests increments requests for a specific service
|
|
func (m *Metrics) IncrementServiceRequests(service string) {
|
|
m.serviceMutex.Lock()
|
|
defer m.serviceMutex.Unlock()
|
|
m.ServiceRequests[service]++
|
|
}
|
|
|
|
// GetServiceRequests returns the number of requests for a service
|
|
func (m *Metrics) GetServiceRequests(service string) int64 {
|
|
m.serviceMutex.RLock()
|
|
defer m.serviceMutex.RUnlock()
|
|
return m.ServiceRequests[service]
|
|
}
|
|
|
|
func (m *Metrics) IncrementPromotions() { atomic.AddInt64(&m.Promotions, 1) }
|
|
func (m *Metrics) IncrementEvictions() { atomic.AddInt64(&m.Evictions, 1) }
|
|
func (m *Metrics) IncrementCapacityPressureEvents() { atomic.AddInt64(&m.CapacityPressureEvents, 1) }
|
|
|
|
// NoteSoftEviction records one cap-pressure eviction batch that freed bytes.
|
|
// Keeps the existing evictions counter and also increments capacity_pressure_events.
|
|
// Nil m is safe: the log still fires so ops can grep without metrics wired.
|
|
func NoteSoftEviction(m *Metrics, tier string, evicted uint) {
|
|
if evicted == 0 {
|
|
return
|
|
}
|
|
if m != nil {
|
|
m.IncrementEvictions()
|
|
m.IncrementCapacityPressureEvents()
|
|
}
|
|
logger.Logger.Info().
|
|
Str("tier", tier).
|
|
Str("reason", "eviction").
|
|
Uint("bytes_evicted", evicted).
|
|
Msg("cache capacity pressure")
|
|
}
|
|
|
|
// NoteNoSpace records a disk Create/Write/Mkdir ENOSPC (or equivalent) event.
|
|
func NoteNoSpace(m *Metrics, err error) {
|
|
if m != nil {
|
|
m.IncrementCapacityPressureEvents()
|
|
}
|
|
logger.Logger.Warn().
|
|
Str("tier", "disk").
|
|
Str("reason", "enospc").
|
|
Err(err).
|
|
Msg("cache capacity pressure")
|
|
}
|
|
|
|
// Additional observability counters
|
|
func (m *Metrics) IncrementUpstreamErrors() { atomic.AddInt64(&m.UpstreamErrors, 1) }
|
|
func (m *Metrics) IncrementCacheWriteFailures() { atomic.AddInt64(&m.CacheWriteFailures, 1) }
|
|
func (m *Metrics) IncrementServiceError(service string) {
|
|
m.serviceErrorsMutex.Lock()
|
|
defer m.serviceErrorsMutex.Unlock()
|
|
if m.ServiceErrors == nil {
|
|
m.ServiceErrors = make(map[string]int64)
|
|
}
|
|
m.ServiceErrors[service]++
|
|
}
|
|
|
|
// GetStats returns a snapshot of current metrics
|
|
func (m *Metrics) GetStats() *Stats {
|
|
totalRequests := atomic.LoadInt64(&m.TotalRequests)
|
|
cacheHits := atomic.LoadInt64(&m.CacheHits)
|
|
cacheMisses := atomic.LoadInt64(&m.CacheMisses)
|
|
|
|
var hitRate float64
|
|
if totalRequests > 0 {
|
|
hitRate = float64(cacheHits) / float64(totalRequests)
|
|
}
|
|
|
|
var avgResponseTime time.Duration
|
|
if totalRequests > 0 {
|
|
avgResponseTime = time.Duration(atomic.LoadInt64(&m.TotalResponseTime) / totalRequests)
|
|
}
|
|
|
|
m.serviceMutex.RLock()
|
|
serviceRequests := make(map[string]int64)
|
|
for k, v := range m.ServiceRequests {
|
|
serviceRequests[k] = v
|
|
}
|
|
m.serviceMutex.RUnlock()
|
|
|
|
serviceErrors := make(map[string]int64)
|
|
m.serviceErrorsMutex.RLock()
|
|
defer m.serviceErrorsMutex.RUnlock()
|
|
for k, v := range m.ServiceErrors {
|
|
serviceErrors[k] = v
|
|
}
|
|
|
|
memoryCacheSize := atomic.LoadInt64(&m.MemoryCacheSize)
|
|
diskCacheSize := atomic.LoadInt64(&m.DiskCacheSize)
|
|
memoryCacheCapacity := atomic.LoadInt64(&m.MemoryCacheCapacity)
|
|
diskCacheCapacity := atomic.LoadInt64(&m.DiskCacheCapacity)
|
|
|
|
return &Stats{
|
|
TotalRequests: totalRequests,
|
|
CacheHits: cacheHits,
|
|
CacheMisses: cacheMisses,
|
|
CacheCoalesced: atomic.LoadInt64(&m.CacheCoalesced),
|
|
NegativeCacheHits: atomic.LoadInt64(&m.NegativeCacheHits),
|
|
RangeCache: atomic.LoadInt64(&m.RangeCache),
|
|
RangeUpstream: atomic.LoadInt64(&m.RangeUpstream),
|
|
Errors: atomic.LoadInt64(&m.Errors),
|
|
RateLimited: atomic.LoadInt64(&m.RateLimited),
|
|
HitRate: hitRate,
|
|
AvgResponseTime: avgResponseTime,
|
|
TotalBytesServed: atomic.LoadInt64(&m.TotalBytesServed),
|
|
TotalBytesSaved: atomic.LoadInt64(&m.TotalBytesSaved),
|
|
MemoryCacheSize: memoryCacheSize,
|
|
DiskCacheSize: diskCacheSize,
|
|
MemoryCacheCapacity: memoryCacheCapacity,
|
|
DiskCacheCapacity: diskCacheCapacity,
|
|
DiskCacheFullRatio: diskFullRatio(diskCacheSize, diskCacheCapacity),
|
|
DiskTierReady: atomic.LoadInt64(&m.DiskTierReady),
|
|
MemoryCacheHits: atomic.LoadInt64(&m.MemoryCacheHits),
|
|
DiskCacheHits: atomic.LoadInt64(&m.DiskCacheHits),
|
|
Promotions: atomic.LoadInt64(&m.Promotions),
|
|
Evictions: atomic.LoadInt64(&m.Evictions),
|
|
CapacityPressureEvents: atomic.LoadInt64(&m.CapacityPressureEvents),
|
|
ServiceRequests: serviceRequests,
|
|
UpstreamErrors: atomic.LoadInt64(&m.UpstreamErrors),
|
|
CacheWriteFailures: atomic.LoadInt64(&m.CacheWriteFailures),
|
|
ServiceErrors: serviceErrors,
|
|
Uptime: time.Since(m.StartTime),
|
|
LastResetTime: m.LastResetTime,
|
|
}
|
|
}
|
|
|
|
// diskFullRatio is size / capacity clamped to [0,1].
|
|
// It is 0 when no disk is configured or the capacity is 0.
|
|
func diskFullRatio(size, capacity int64) float64 {
|
|
if size <= 0 || capacity <= 0 {
|
|
return 0
|
|
}
|
|
ratio := float64(size) / float64(capacity)
|
|
if ratio > 1 {
|
|
return 1
|
|
}
|
|
return ratio
|
|
}
|
|
|
|
// Reset resets all metrics to zero
|
|
func (m *Metrics) Reset() {
|
|
atomic.StoreInt64(&m.TotalRequests, 0)
|
|
atomic.StoreInt64(&m.CacheHits, 0)
|
|
atomic.StoreInt64(&m.CacheMisses, 0)
|
|
atomic.StoreInt64(&m.CacheCoalesced, 0)
|
|
atomic.StoreInt64(&m.NegativeCacheHits, 0)
|
|
atomic.StoreInt64(&m.RangeCache, 0)
|
|
atomic.StoreInt64(&m.RangeUpstream, 0)
|
|
atomic.StoreInt64(&m.Errors, 0)
|
|
atomic.StoreInt64(&m.RateLimited, 0)
|
|
atomic.StoreInt64(&m.TotalResponseTime, 0)
|
|
atomic.StoreInt64(&m.TotalBytesServed, 0)
|
|
atomic.StoreInt64(&m.TotalBytesSaved, 0)
|
|
atomic.StoreInt64(&m.MemoryCacheHits, 0)
|
|
atomic.StoreInt64(&m.DiskCacheHits, 0)
|
|
atomic.StoreInt64(&m.Promotions, 0)
|
|
atomic.StoreInt64(&m.Evictions, 0)
|
|
atomic.StoreInt64(&m.CapacityPressureEvents, 0)
|
|
atomic.StoreInt64(&m.UpstreamErrors, 0)
|
|
atomic.StoreInt64(&m.CacheWriteFailures, 0)
|
|
|
|
m.serviceMutex.Lock()
|
|
m.ServiceRequests = make(map[string]int64)
|
|
m.serviceMutex.Unlock()
|
|
|
|
m.serviceErrorsMutex.Lock()
|
|
defer m.serviceErrorsMutex.Unlock()
|
|
m.ServiceErrors = make(map[string]int64)
|
|
|
|
m.LastResetTime = time.Now()
|
|
}
|
|
|
|
// Stats represents a snapshot of metrics
|
|
type Stats struct {
|
|
TotalRequests int64
|
|
CacheHits int64
|
|
CacheMisses int64
|
|
CacheCoalesced int64
|
|
NegativeCacheHits int64
|
|
RangeCache int64
|
|
RangeUpstream int64
|
|
Errors int64
|
|
RateLimited int64
|
|
HitRate float64
|
|
AvgResponseTime time.Duration
|
|
TotalBytesServed int64
|
|
TotalBytesSaved int64
|
|
MemoryCacheSize int64
|
|
|
|
DiskCacheSize int64
|
|
MemoryCacheCapacity int64 // configured memory capacity (bytes)
|
|
DiskCacheCapacity int64 // configured disk capacity (bytes); 0 when no disk
|
|
DiskCacheFullRatio float64 // disk_cache_size / disk_cache_capacity, clamped to [0,1]; 0 when no disk or capacity is 0
|
|
DiskTierReady int64
|
|
MemoryCacheHits int64
|
|
DiskCacheHits int64
|
|
Promotions int64
|
|
Evictions int64
|
|
CapacityPressureEvents int64
|
|
UpstreamErrors int64
|
|
CacheWriteFailures int64
|
|
ServiceErrors map[string]int64
|
|
ServiceRequests map[string]int64
|
|
Uptime time.Duration
|
|
LastResetTime time.Time
|
|
}
|
|
|
|
// WriteText emits Prometheus text exposition format 0.0.4 to the ResponseWriter.
|
|
// Each metric family is # HELP, then # TYPE, then one or more sample lines.
|
|
// Metric names are stable; labeled series keep service=%q (Prometheus-valid quotes).
|
|
// All fmt.Fprintf errors are intentionally discarded via _ = : this is a best-effort
|
|
// read-only debug endpoint; client disconnects or write errors during metrics dump
|
|
// are not actionable (do not affect cache correctness or require retries).
|
|
func WriteText(w http.ResponseWriter, stats *Stats) {
|
|
writeInt(w, "total_requests", "Total HTTP requests handled.", "counter", stats.TotalRequests)
|
|
writeInt(w, "cache_hits", "Requests served from cache.", "counter", stats.CacheHits)
|
|
writeInt(w, "cache_misses", "Requests not found in cache.", "counter", stats.CacheMisses)
|
|
writeInt(w, "negative_cache_hits", "404/410 served from a still-valid negative cache entry.", "counter", stats.NegativeCacheHits)
|
|
writeInt(w, "cache_coalesced", "Requests coalesced onto an in-flight upstream fetch.", "counter", stats.CacheCoalesced)
|
|
writeInt(w, "range_cache", "Range requests served as 206 from an already-cached object.", "counter", stats.RangeCache)
|
|
writeInt(w, "range_upstream", "Range requests that required an upstream fetch, served as 206.", "counter", stats.RangeUpstream)
|
|
writeInt(w, "errors", "Request errors.", "counter", stats.Errors)
|
|
writeInt(w, "rate_limited", "Requests rejected by rate limiting.", "counter", stats.RateLimited)
|
|
writeInt(w, "upstream_errors", "Errors talking to upstream.", "counter", stats.UpstreamErrors)
|
|
writeInt(w, "cache_write_failures", "Failures writing objects into cache.", "counter", stats.CacheWriteFailures)
|
|
writeInt(w, "memory_cache_hits", "Hits served from the memory tier.", "counter", stats.MemoryCacheHits)
|
|
writeInt(w, "disk_cache_hits", "Hits served from the disk tier.", "counter", stats.DiskCacheHits)
|
|
writeInt(w, "promotions", "Objects promoted from disk to memory.", "counter", stats.Promotions)
|
|
writeInt(w, "evictions", "Objects evicted from cache.", "counter", stats.Evictions)
|
|
writeInt(w, "capacity_pressure_events", "Soft eviction under the memory or disk cap, and/or disk ENOSPC.", "counter", stats.CapacityPressureEvents)
|
|
|
|
writeHelpType(w, "service_errors", "Errors attributed to a named service.", "counter")
|
|
for svc, cnt := range stats.ServiceErrors {
|
|
_, _ = fmt.Fprintf(w, "service_errors{service=%q} %d\n", svc, cnt)
|
|
}
|
|
writeHelpType(w, "service_requests", "Requests attributed to a named service.", "counter")
|
|
for svc, cnt := range stats.ServiceRequests {
|
|
_, _ = fmt.Fprintf(w, "service_requests{service=%q} %d\n", svc, cnt)
|
|
}
|
|
|
|
writeFloat(w, "hit_rate", "Cache hits divided by total requests.", "gauge", "%.4f", stats.HitRate)
|
|
writeFloat(w, "avg_response_time_ms", "Average response time in milliseconds.", "gauge", "%.2f", float64(stats.AvgResponseTime.Nanoseconds())/1e6)
|
|
writeInt(w, "total_bytes_served", "Total bytes sent to clients.", "counter", stats.TotalBytesServed)
|
|
writeInt(w, "total_bytes_saved", "Bytes served from cache instead of being re-downloaded from upstream.", "counter", stats.TotalBytesSaved)
|
|
writeInt(w, "memory_cache_size", "Current memory cache size in bytes.", "gauge", stats.MemoryCacheSize)
|
|
writeInt(w, "memory_cache_capacity", "Configured memory cache capacity in bytes.", "gauge", stats.MemoryCacheCapacity)
|
|
writeInt(w, "disk_cache_size", "Current disk cache size in bytes.", "gauge", stats.DiskCacheSize)
|
|
writeInt(w, "disk_cache_capacity", "Configured disk cache capacity in bytes; 0 when no disk is configured.", "gauge", stats.DiskCacheCapacity)
|
|
writeFloat(w, "disk_cache_full_ratio", "disk_cache_size / disk_cache_capacity in [0,1]; 0 when no disk or capacity is 0.", "gauge", "%.4f", stats.DiskCacheFullRatio)
|
|
writeInt(w, "disk_tier_ready", "1 if the disk tier is attached or no disk is configured; 0 while attach is pending.", "gauge", stats.DiskTierReady)
|
|
writeFloat(w, "uptime_seconds", "Process uptime in seconds.", "gauge", "%.2f", stats.Uptime.Seconds())
|
|
}
|
|
|
|
func writeHelpType(w http.ResponseWriter, name, help, typ string) {
|
|
_, _ = fmt.Fprintf(w, "# HELP %s %s\n# TYPE %s %s\n", name, help, name, typ)
|
|
}
|
|
|
|
func writeInt(w http.ResponseWriter, name, help, typ string, v int64) {
|
|
writeHelpType(w, name, help, typ)
|
|
_, _ = fmt.Fprintf(w, "%s %d\n", name, v)
|
|
}
|
|
|
|
func writeFloat(w http.ResponseWriter, name, help, typ, valFmt string, v float64) {
|
|
writeHelpType(w, name, help, typ)
|
|
_, _ = fmt.Fprintf(w, "%s "+valFmt+"\n", name, v)
|
|
}
|