fix: expose workbench cache observability
This commit is contained in:
lyon
@@ -21,6 +21,9 @@ import (
|
||||
|
||||
const cacheKeySchemaVersion = "wb-derived-v1"
|
||||
|
||||
var cacheOperationDurationBucketsSeconds = []float64{0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1}
|
||||
var cachePayloadBucketsBytes = []float64{512, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304}
|
||||
|
||||
var errCacheDisabled = errors.New("workbench redis cache disabled")
|
||||
|
||||
type cacheConfig struct {
|
||||
@@ -60,15 +63,38 @@ type derivedCache struct {
|
||||
}
|
||||
|
||||
type cacheMetrics struct {
|
||||
mu sync.Mutex
|
||||
Hits uint64
|
||||
Misses uint64
|
||||
Sets uint64
|
||||
Deletes uint64
|
||||
Errors uint64
|
||||
LastErrorKind string
|
||||
LastErrorOp string
|
||||
LastErrorAt time.Time
|
||||
mu sync.Mutex
|
||||
Hits uint64
|
||||
Misses uint64
|
||||
Sets uint64
|
||||
Deletes uint64
|
||||
Errors uint64
|
||||
LastErrorKind string
|
||||
LastErrorOp string
|
||||
LastErrorAt time.Time
|
||||
Requests map[string]cacheCounterSeries
|
||||
Stale map[string]cacheCounterSeries
|
||||
DBQueryAvoided map[string]cacheCounterSeries
|
||||
OperationDurations map[string]*cacheHistogramSeries
|
||||
PayloadBytes map[string]*cacheHistogramSeries
|
||||
}
|
||||
|
||||
type cacheMetricLabels struct {
|
||||
Class string
|
||||
Operation string
|
||||
Status string
|
||||
}
|
||||
|
||||
type cacheCounterSeries struct {
|
||||
Labels cacheMetricLabels
|
||||
Value uint64
|
||||
}
|
||||
|
||||
type cacheHistogramSeries struct {
|
||||
Labels cacheMetricLabels
|
||||
Buckets []uint64
|
||||
Sum float64
|
||||
Count uint64
|
||||
}
|
||||
|
||||
type cacheError struct {
|
||||
@@ -144,12 +170,15 @@ func (c *derivedCache) BuildKey(parts cacheKeyParts) (string, error) {
|
||||
}
|
||||
|
||||
func (c *derivedCache) Get(ctx context.Context, key string, target any) (bool, error) {
|
||||
startedAt := time.Now()
|
||||
class := cacheKeyClass(key)
|
||||
if !c.usable() {
|
||||
return false, nil
|
||||
}
|
||||
if err := validateCacheKey(c.config.MaxKeyBytes, key); err != nil {
|
||||
wrapped := cacheError{Op: "get", Kind: classifyCacheError(err), Err: err}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "get", "error", time.Since(startedAt), 0)
|
||||
return false, wrapped
|
||||
}
|
||||
var payload []byte
|
||||
@@ -160,42 +189,51 @@ func (c *derivedCache) Get(ctx context.Context, key string, target any) (bool, e
|
||||
})
|
||||
if errors.Is(err, redis.Nil) {
|
||||
c.metrics.recordMiss()
|
||||
c.metrics.recordCacheObservation(class, "get", "miss", time.Since(startedAt), 0)
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
wrapped := cacheError{Op: "get", Kind: classifyCacheError(err), Err: err}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "get", "unavailable", time.Since(startedAt), 0)
|
||||
return false, wrapped
|
||||
}
|
||||
if len(payload) > c.config.MaxPayloadBytes {
|
||||
wrapped := cacheError{Op: "get", Kind: "cache_payload_too_large", Err: fmt.Errorf("payload bytes %d exceeds %d", len(payload), c.config.MaxPayloadBytes)}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "get", "error", time.Since(startedAt), len(payload))
|
||||
return false, wrapped
|
||||
}
|
||||
if target != nil {
|
||||
if err := json.Unmarshal(payload, target); err != nil {
|
||||
wrapped := cacheError{Op: "get", Kind: "cache_payload_decode_failed", Err: err}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "get", "error", time.Since(startedAt), len(payload))
|
||||
return false, wrapped
|
||||
}
|
||||
}
|
||||
c.metrics.recordHit()
|
||||
c.metrics.recordCacheObservation(class, "get", "hit", time.Since(startedAt), len(payload))
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (c *derivedCache) Set(ctx context.Context, key string, value any, ttl time.Duration) error {
|
||||
startedAt := time.Now()
|
||||
class := cacheKeyClass(key)
|
||||
if !c.usable() {
|
||||
return nil
|
||||
}
|
||||
if err := validateCacheKey(c.config.MaxKeyBytes, key); err != nil {
|
||||
wrapped := cacheError{Op: "set", Kind: classifyCacheError(err), Err: err}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "set", "error", time.Since(startedAt), 0)
|
||||
return wrapped
|
||||
}
|
||||
payload, err := cachePayloadBytes(value, c.config.MaxPayloadBytes)
|
||||
if err != nil {
|
||||
wrapped := cacheError{Op: "set", Kind: classifyCacheError(err), Err: err}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "set", "error", time.Since(startedAt), 0)
|
||||
return wrapped
|
||||
}
|
||||
err = c.withTimeout(ctx, func(opCtx context.Context) error {
|
||||
@@ -204,12 +242,36 @@ func (c *derivedCache) Set(ctx context.Context, key string, value any, ttl time.
|
||||
if err != nil {
|
||||
wrapped := cacheError{Op: "set", Kind: classifyCacheError(err), Err: err}
|
||||
c.metrics.recordError(wrapped)
|
||||
c.metrics.recordCacheObservation(class, "set", "unavailable", time.Since(startedAt), len(payload))
|
||||
return wrapped
|
||||
}
|
||||
c.metrics.recordSet()
|
||||
c.metrics.recordCacheObservation(class, "set", "stored", time.Since(startedAt), len(payload))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *derivedCache) RecordStale(class string) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
c.metrics.recordStale(class)
|
||||
}
|
||||
|
||||
func (c *derivedCache) RecordDBQueryAvoided(class string) {
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
c.metrics.recordDBQueryAvoided(class)
|
||||
}
|
||||
|
||||
func (c *derivedCache) PrometheusText(service string) string {
|
||||
if c == nil {
|
||||
var metrics cacheMetrics
|
||||
return metrics.prometheusText(service)
|
||||
}
|
||||
return c.metrics.prometheusText(service)
|
||||
}
|
||||
|
||||
func (c *derivedCache) Delete(ctx context.Context, keys ...string) error {
|
||||
if !c.usable() || len(keys) == 0 {
|
||||
return nil
|
||||
@@ -520,6 +582,15 @@ func cacheHash(value string) string {
|
||||
return hex.EncodeToString(digest[:])[:24]
|
||||
}
|
||||
|
||||
func cacheKeyClass(key string) string {
|
||||
for _, part := range strings.Split(key, ":") {
|
||||
if strings.HasPrefix(part, "class=") {
|
||||
return safeCacheToken(strings.TrimPrefix(part, "class="))
|
||||
}
|
||||
}
|
||||
return "unknown"
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordHit() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
@@ -556,16 +627,79 @@ func (m *cacheMetrics) recordError(err error) {
|
||||
m.LastErrorAt = time.Now().UTC()
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordCacheObservation(class string, operation string, status string, duration time.Duration, payloadBytes int) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
labels := cacheLabels(class, operation, status)
|
||||
m.incrementCounterLocked(&m.Requests, labels)
|
||||
m.recordHistogramLocked(&m.OperationDurations, labels, duration.Seconds(), cacheOperationDurationBucketsSeconds)
|
||||
if payloadBytes > 0 {
|
||||
m.recordHistogramLocked(&m.PayloadBytes, labels, float64(payloadBytes), cachePayloadBucketsBytes)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordStale(class string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.incrementCounterLocked(&m.Stale, cacheLabels(class, "read", "stale"))
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordDBQueryAvoided(class string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.incrementCounterLocked(&m.DBQueryAvoided, cacheLabels(class, "read", "hit"))
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) incrementCounterLocked(target *map[string]cacheCounterSeries, labels cacheMetricLabels) {
|
||||
if *target == nil {
|
||||
*target = map[string]cacheCounterSeries{}
|
||||
}
|
||||
key := cacheMetricKey(labels)
|
||||
entry := (*target)[key]
|
||||
if entry.Labels.Class == "" {
|
||||
entry.Labels = labels
|
||||
}
|
||||
entry.Value++
|
||||
(*target)[key] = entry
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) recordHistogramLocked(target *map[string]*cacheHistogramSeries, labels cacheMetricLabels, value float64, buckets []float64) {
|
||||
if value < 0 {
|
||||
value = 0
|
||||
}
|
||||
if *target == nil {
|
||||
*target = map[string]*cacheHistogramSeries{}
|
||||
}
|
||||
key := cacheMetricKey(labels)
|
||||
entry := (*target)[key]
|
||||
if entry == nil {
|
||||
entry = &cacheHistogramSeries{Labels: labels, Buckets: make([]uint64, len(buckets))}
|
||||
(*target)[key] = entry
|
||||
}
|
||||
for index, bucket := range buckets {
|
||||
if value <= bucket {
|
||||
entry.Buckets[index]++
|
||||
}
|
||||
}
|
||||
entry.Sum += value
|
||||
entry.Count++
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) snapshot() map[string]any {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
result := map[string]any{
|
||||
"hits": int64(m.Hits),
|
||||
"misses": int64(m.Misses),
|
||||
"sets": int64(m.Sets),
|
||||
"deletes": int64(m.Deletes),
|
||||
"errors": int64(m.Errors),
|
||||
"valuesRedacted": true,
|
||||
"hits": int64(m.Hits),
|
||||
"misses": int64(m.Misses),
|
||||
"sets": int64(m.Sets),
|
||||
"deletes": int64(m.Deletes),
|
||||
"errors": int64(m.Errors),
|
||||
"requestSeries": len(m.Requests),
|
||||
"operationDurationSeries": len(m.OperationDurations),
|
||||
"payloadBytesSeries": len(m.PayloadBytes),
|
||||
"staleSeries": len(m.Stale),
|
||||
"dbQueryAvoidedSeries": len(m.DBQueryAvoided),
|
||||
"valuesRedacted": true,
|
||||
}
|
||||
if m.LastErrorKind != "" {
|
||||
result["lastErrorKind"] = m.LastErrorKind
|
||||
@@ -575,6 +709,132 @@ func (m *cacheMetrics) snapshot() map[string]any {
|
||||
return result
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) prometheusText(service string) string {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if service == "" {
|
||||
service = "hwlab-workbench-runtime"
|
||||
}
|
||||
lines := []string{
|
||||
"# HELP workbench_cache_requests_total Workbench derived Redis cache requests by low-cardinality class/status.",
|
||||
"# TYPE workbench_cache_requests_total counter",
|
||||
}
|
||||
for _, entry := range m.Requests {
|
||||
lines = append(lines, prometheusCounterLine("workbench_cache_requests_total", service, entry.Labels, entry.Value))
|
||||
}
|
||||
lines = append(lines,
|
||||
"# HELP workbench_cache_hit_ratio Workbench derived Redis cache hit ratio by class and operation.",
|
||||
"# TYPE workbench_cache_hit_ratio gauge",
|
||||
)
|
||||
for _, line := range m.prometheusHitRatioLinesLocked(service) {
|
||||
lines = append(lines, line)
|
||||
}
|
||||
lines = append(lines,
|
||||
"# HELP workbench_cache_operation_duration_seconds Workbench derived Redis cache operation duration in seconds.",
|
||||
"# TYPE workbench_cache_operation_duration_seconds histogram",
|
||||
)
|
||||
lines = append(lines, prometheusHistogramLines("workbench_cache_operation_duration_seconds", service, m.OperationDurations, cacheOperationDurationBucketsSeconds)...)
|
||||
lines = append(lines,
|
||||
"# HELP workbench_cache_payload_bytes Workbench derived Redis cache payload size in bytes.",
|
||||
"# TYPE workbench_cache_payload_bytes histogram",
|
||||
)
|
||||
lines = append(lines, prometheusHistogramLines("workbench_cache_payload_bytes", service, m.PayloadBytes, cachePayloadBucketsBytes)...)
|
||||
lines = append(lines,
|
||||
"# HELP workbench_cache_stale_total Workbench derived Redis cache stale reads detected by payload freshness metadata.",
|
||||
"# TYPE workbench_cache_stale_total counter",
|
||||
)
|
||||
for _, entry := range m.Stale {
|
||||
lines = append(lines, prometheusCounterLine("workbench_cache_stale_total", service, entry.Labels, entry.Value))
|
||||
}
|
||||
lines = append(lines,
|
||||
"# HELP workbench_cache_unavailable_total Workbench derived Redis cache unavailable/error requests.",
|
||||
"# TYPE workbench_cache_unavailable_total counter",
|
||||
)
|
||||
for _, entry := range m.Requests {
|
||||
if entry.Labels.Status == "unavailable" || entry.Labels.Status == "error" {
|
||||
lines = append(lines, prometheusCounterLine("workbench_cache_unavailable_total", service, entry.Labels, entry.Value))
|
||||
}
|
||||
}
|
||||
lines = append(lines,
|
||||
"# HELP workbench_db_query_avoided_total Workbench database reads avoided by derived Redis cache hits.",
|
||||
"# TYPE workbench_db_query_avoided_total counter",
|
||||
)
|
||||
for _, entry := range m.DBQueryAvoided {
|
||||
lines = append(lines, prometheusCounterLine("workbench_db_query_avoided_total", service, entry.Labels, entry.Value))
|
||||
}
|
||||
lines = append(lines, "")
|
||||
return strings.Join(lines, "\n")
|
||||
}
|
||||
|
||||
func (m *cacheMetrics) prometheusHitRatioLinesLocked(service string) []string {
|
||||
type totals struct{ hit, total uint64 }
|
||||
groups := map[string]totals{}
|
||||
labels := map[string]cacheMetricLabels{}
|
||||
for _, entry := range m.Requests {
|
||||
if entry.Labels.Operation != "get" {
|
||||
continue
|
||||
}
|
||||
key := entry.Labels.Class + "\x00" + entry.Labels.Operation
|
||||
group := groups[key]
|
||||
if entry.Labels.Status == "hit" {
|
||||
group.hit += entry.Value
|
||||
}
|
||||
group.total += entry.Value
|
||||
groups[key] = group
|
||||
labels[key] = cacheMetricLabels{Class: entry.Labels.Class, Operation: entry.Labels.Operation, Status: "all"}
|
||||
}
|
||||
lines := []string{}
|
||||
for key, group := range groups {
|
||||
ratio := 0.0
|
||||
if group.total > 0 {
|
||||
ratio = float64(group.hit) / float64(group.total)
|
||||
}
|
||||
lines = append(lines, prometheusGaugeLine("workbench_cache_hit_ratio", service, labels[key], ratio))
|
||||
}
|
||||
return lines
|
||||
}
|
||||
|
||||
func prometheusHistogramLines(name string, service string, series map[string]*cacheHistogramSeries, buckets []float64) []string {
|
||||
lines := []string{}
|
||||
for _, entry := range series {
|
||||
for index, bucket := range buckets {
|
||||
lines = append(lines, fmt.Sprintf("%s_bucket{%s,le=\"%s\"} %d", name, prometheusLabels(service, entry.Labels), prometheusFloat(bucket), entry.Buckets[index]))
|
||||
}
|
||||
lines = append(lines, fmt.Sprintf("%s_bucket{%s,le=\"+Inf\"} %d", name, prometheusLabels(service, entry.Labels), entry.Count))
|
||||
lines = append(lines, fmt.Sprintf("%s_sum{%s} %.6f", name, prometheusLabels(service, entry.Labels), entry.Sum))
|
||||
lines = append(lines, fmt.Sprintf("%s_count{%s} %d", name, prometheusLabels(service, entry.Labels), entry.Count))
|
||||
}
|
||||
return lines
|
||||
}
|
||||
|
||||
func prometheusCounterLine(name string, service string, labels cacheMetricLabels, value uint64) string {
|
||||
return fmt.Sprintf("%s{%s} %d", name, prometheusLabels(service, labels), value)
|
||||
}
|
||||
|
||||
func prometheusGaugeLine(name string, service string, labels cacheMetricLabels, value float64) string {
|
||||
return fmt.Sprintf("%s{%s} %.6f", name, prometheusLabels(service, labels), value)
|
||||
}
|
||||
|
||||
func prometheusLabels(service string, labels cacheMetricLabels) string {
|
||||
return fmt.Sprintf("service=\"%s\",cache_key_class=\"%s\",operation=\"%s\",cache_status=\"%s\"", prometheusEscape(service), prometheusEscape(labels.Class), prometheusEscape(labels.Operation), prometheusEscape(labels.Status))
|
||||
}
|
||||
|
||||
func prometheusEscape(value string) string {
|
||||
return strings.ReplaceAll(strings.ReplaceAll(strings.ReplaceAll(value, "\\", "\\\\"), "\n", "\\n"), "\"", "\\\"")
|
||||
}
|
||||
|
||||
func prometheusFloat(value float64) string {
|
||||
return strconv.FormatFloat(value, 'f', -1, 64)
|
||||
}
|
||||
|
||||
func cacheLabels(class string, operation string, status string) cacheMetricLabels {
|
||||
return cacheMetricLabels{Class: safeCacheToken(first(class, "unknown")), Operation: safeCacheToken(first(operation, "unknown")), Status: safeCacheToken(first(status, "unknown"))}
|
||||
}
|
||||
|
||||
func cacheMetricKey(labels cacheMetricLabels) string {
|
||||
return labels.Class + "\x00" + labels.Operation + "\x00" + labels.Status
|
||||
}
|
||||
|
||||
func envBool(name string, fallback bool) bool {
|
||||
value := strings.ToLower(strings.TrimSpace(os.Getenv(name)))
|
||||
if value == "" {
|
||||
|
||||
Reference in New Issue
Block a user