[WIP] fix: labeled cache result metrics + stop Vary-sentinel from polluting L2 hit-rate

docs/operation/operation.md

@@ -322,6 +322,14 @@ To monitor skipper we recommend the following queries:
 - P99 response filter duration (depends on label selector): `histogram_quantile(0.99, sum(rate(skipper_filter_response_duration_seconds_bucket{application="skipper-ingress"}[1m])) by (le) )`
 - If you use Kubernetes limits or Linux cgroup CFS quotas (depends on label selector): `sum(rate(container_cpu_cfs_throttled_periods_total{container_name="skipper-ingress"}[1m]))`
 
+#### Cache filter metrics
+
+When the cache filter is active, the following labeled metrics are exposed:
+
+- `skipper_cache_result_total{result="...", code_class="2xx|4xx|5xx|other|n/a"}` - Counter for cache-related outcomes by result type and HTTP status code class. `result` covers the same outcomes as the flat `skipper_custom_total{key="cache.*"}` counters below — `hit`, `miss`, `stale`, `storage_error`, `coalesce_error`, `coalesce_bypass`, `coalesce_bypass_error`, and, when an L2 (Valkey) store is configured, `l1_hit`, `l2_hit`, `l2_miss`, `l2_get_error`, `l2_set_fallback`. `code_class` is only meaningful for outcomes with a cached response in hand (`hit`, `miss`, `stale`, `l1_hit`, `l2_hit`, `l2_set_fallback`); for outcomes with no response to classify (`storage_error`, `coalesce_error`, `coalesce_bypass`, `coalesce_bypass_error`, `l2_miss`, `l2_get_error`) it is the literal string `n/a`. A rising `hit`/`4xx` or `hit`/`5xx` series indicates the cache is serving an error response as a hit — investigate the upstream response and cache-control directives for that route. Cardinality is bounded by the number of distinct (result, code_class) pairs actually emitted.
+- `skipper_cache_lru_bytes{route="..."}` - Gauge for per-route LRU (L1) cache storage consumption in bytes, updated on every hit, miss, and stale serve. Cardinality bounded by the number of distinct routes.
+- The flat-key counters `skipper_custom_total{key="cache.hit"}`, `cache.miss`, `cache.stale`, and the rest of the `cache.*` keys above are the primary, directly-emitted metrics — `skipper_cache_result_total`/`skipper_cache_lru_bytes` are an additional labeled breakdown of the same underlying events, not a replacement for them.
+
 You may add static metrics labels like `version` using Prometheus [relabeling feature](https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config).
 
 ### Prometheus native histograms
@@ -2123,6 +2131,9 @@ flowchart TD
   it performed its own fetch instead of reusing another request's in-flight response
 - `cache.coalesce_bypass_error`: Counter, an independent bypass fetch (see
   `cache.coalesce_bypass`) returned an error
+- `cache.coalesce_wait_duration`: Histogram, time from issuing a cold-miss singleflight
+  call to the result being available — near-zero for the leader (it computes the result
+  itself), the leader's real fetch time for followers waiting on it
 
 **LRU (always active):**
 
@@ -2150,7 +2161,10 @@ flowchart TD
 - `cache.l2_get_error`: Counter, L2 Get errors — request treated as a cache miss, fetched from origin
 - `cache.l2_set_fallback`: Counter, L2 Set errors — entry written to L1 only (not L2)
 - `cache.l2_hit`: Counter, successful L2 Get (entry returned from L2); L1 is warmed as a side-effect when `--cache-l1-ttl > 0`
+- `cache.l2_get_duration`: Histogram, time of the underlying Valkey/Redis `Get` round-trip, on every call (hit, miss, or error) — including internal lookups like the Vary-sentinel check
+- `cache.l2_set_duration`: Histogram, time of the underlying Valkey/Redis `SetWithExpire` round-trip, on every call (success or fallback)
 - `cache.storage_error`: Counter, any storage operation (Set or Delete) failed — covers both L2 failures and L1 eviction-path errors; the request is still served correctly
+- `skipper_cache_storage_error_total{backend="l1|l2", op="get|set|delete"}`: Counter, labeled breakdown of storage errors by backend and operation, attributed at the point in `LRUStorage`/`L2Storage` where the backend is actually known. Also increments the flat `cache.storage_error` counter above and the labeled `skipper_cache_result_total{result="storage_error",code_class="n/a"}`, so existing consumers of either see no change — this is purely an additional breakdown. `backend="l2", op="delete"` always reports the real underlying failure, since `L2Storage.Delete` only ever returns `nil` to its caller regardless of whether the Valkey delete succeeded (L1 delete always runs); a generic `Storage.Set`/`Storage.Delete` failure from code that isn't `LRUStorage` or `L2Storage` (e.g. a test double) cannot be attributed and records nothing here.
 
 **OpenTracing span tags (set on every request when a span is active):**
 

filters/cache/filter.go

@@ -88,7 +88,26 @@ type cacheSpec struct {
 	filters    map[filterCacheKey]*cacheFilter
 }
 
-// NewCacheFilter returns a Spec for the cache() filter.
+// statusCodeClass buckets an HTTP status code for the cache.result_total
+// code_class label.
+func statusCodeClass(code int) string {
+	switch {
+	case code >= 200 && code < 300:
+		return "2xx"
+	case code >= 400 && code < 500:
+		return "4xx"
+	case code >= 500 && code < 600:
+		return "5xx"
+	default:
+		return "other"
+	}
+}
+
+// NewCacheFilter returns a Spec for the cache() filter. maxBytes is the
+// in-memory storage budget for the shared LRU cache backing all filter
+// instances created from this Spec.
+// listenAddr is Skipper's own listener address (e.g. ":9090"); revalidation
+// requests are sent back through Skipper so the full filter chain runs.
 //
 // Route usage (RFC mode — upstream Cache-Control is fully authoritative):
 //
@@ -372,6 +391,18 @@ func tagSpan(ctx filters.FilterContext, status string, ttlRemainingMs int64) {
 	}
 }
 
+// updateLRUBytesGauge records the current per-route L1 (LRU) footprint. It
+// reads f.lruStorage directly rather than f.storage, since f.storage may be
+// an *L2Storage wrapping L1 — f.lruStorage is always a direct, non-nil
+// reference to L1 regardless of whether L2 is configured for this route.
+// Called once per request outcome (hit/miss/stale) so the gauge stays
+// accurate under any traffic mix, not just hit-heavy routes.
+func (f *cacheFilter) updateLRUBytesGauge(ctx filters.FilterContext) {
+	if p, ok := f.metrics.(interface{ SetCacheLRUBytes(string, float64) }); ok {
+		p.SetCacheLRUBytes(ctx.RouteId(), float64(f.lruStorage.lru.Bytes()))
+	}
+}
+
 // Request checks the cache. On a hit it calls ctx.Serve() to short-circuit the
 // backend roundtrip. On a stale hit it serves stale and fires a background
 // revalidation. On a miss it stores the computed key in the state bag so
@@ -381,7 +412,9 @@ func (f *cacheFilter) Request(ctx filters.FilterContext) {
 	// (e.g. header stripping by auth) are intentionally reflected in the cache key.
 	baseKey := cacheKey(ctx.RouteId(), ctx.Request(), f.keyHeaders)
 	key := baseKey
-	if sentinel, _ := f.storage.Get(ctx.Request().Context(), "vary:"+baseKey); sentinel != nil && len(sentinel.VaryHeaders) > 0 {
+	// GetSilent: this is an internal plumbing lookup, not a real cache decision —
+	// it must not count toward hit/l1_hit/l2_hit/l2_miss/l2_get_error on every request.
+	if sentinel, _ := f.storage.GetSilent(ctx.Request().Context(), "vary:"+baseKey); sentinel != nil && len(sentinel.VaryHeaders) > 0 {
 		key = varyKey(baseKey, ctx.Request(), sentinel.VaryHeaders)
 	}
 	ctx.StateBag()[stateBagKey] = key
@@ -455,6 +488,10 @@ func (f *cacheFilter) Request(ctx filters.FilterContext) {
 		tagSpan(ctx, cacheStatusStale, -1)
 		setAgeHeader(rsp, entry, now)
 		f.metrics.IncCounter("cache.stale")
+		if p, ok := f.metrics.(interface{ IncCacheResult(string, string) }); ok {
+			p.IncCacheResult("stale", statusCodeClass(entry.StatusCode))
+		}
+		f.updateLRUBytesGauge(ctx)
 		method := ctx.Request().Method
 		if isCacheableMethod(method) && evaluateConditionals(ctx.Request(), entry) {
 			notModified := &http.Response{
@@ -495,6 +532,10 @@ func (f *cacheFilter) Request(ctx filters.FilterContext) {
 	tagSpan(ctx, cacheStatusHit, max(0, entry.TTL-time.Since(entry.CreatedAt)).Milliseconds())
 	setAgeHeader(rsp, entry, time.Now())
 	f.metrics.IncCounter("cache.hit")
+	if p, ok := f.metrics.(interface{ IncCacheResult(string, string) }); ok {
+		p.IncCacheResult("hit", statusCodeClass(entry.StatusCode))
+	}
+	f.updateLRUBytesGauge(ctx)
 	method := ctx.Request().Method
 	if isCacheableMethod(method) && evaluateConditionals(ctx.Request(), entry) {
 		notModified := &http.Response{
@@ -551,6 +592,7 @@ func (f *cacheFilter) storeDecision(req *http.Request, statusCode int, header ht
 func (f *cacheFilter) coalesce(ctx filters.FilterContext, key string) {
 	req := ctx.Request().Clone(context.Background())
 
+	waitStart := time.Now()
 	ch := f.coldSF.DoChan(key, func() (any, error) {
 		// Capture any existing stale-if-error eligible entry before fetching, so that a
 		// subsequent 5xx response cannot overwrite it in storage before we read it.
@@ -598,7 +640,6 @@ func (f *cacheFilter) coalesce(ctx filters.FilterContext, key string) {
 		if shouldStore {
 			if err := f.storage.Set(context.Background(), key, entry); err != nil {
 				log.WithError(err).Warn("cache: Set failed (cold-miss store)")
-				f.metrics.IncCounter("cache.storage_error")
 			}
 		}
 		return &coalesceResult{
@@ -612,8 +653,12 @@ func (f *cacheFilter) coalesce(ctx filters.FilterContext, key string) {
 
 	select {
 	case res := <-ch:
+		f.metrics.MeasureSince("cache.coalesce_wait_duration", waitStart)
 		if res.Err != nil || res.Val == nil {
 			f.metrics.IncCounter("cache.coalesce_error")
+			if p, ok := f.metrics.(interface{ IncCacheResult(string, string) }); ok {
+				p.IncCacheResult("coalesce_error", "n/a")
+			}
 			return
 		}
 		cr := res.Val.(*coalesceResult)
@@ -646,9 +691,15 @@ func (f *cacheFilter) coalesce(ctx filters.FilterContext, key string) {
 		identityMatchesLeader := len(f.keyHeaders) > 0 || sameVaryDimension
 		if req != cr.leaderReq && !cr.storeable && !identityMatchesLeader {
 			f.metrics.IncCounter("cache.coalesce_bypass")
+			if p, ok := f.metrics.(interface{ IncCacheResult(string, string) }); ok {
+				p.IncCacheResult("coalesce_bypass", "n/a")
+			}
 			resp, err := f.fetch(req)
 			if err != nil {
 				f.metrics.IncCounter("cache.coalesce_bypass_error")
+				if p, ok := f.metrics.(interface{ IncCacheResult(string, string) }); ok {
+					p.IncCacheResult("coalesce_bypass_error", "n/a")
+				}
 				return
 			}
 			defer resp.Body.Close()
@@ -664,6 +715,10 @@ func (f *cacheFilter) coalesce(ctx filters.FilterContext, key string) {
 		rsp.Header.Set(cacheStatusHeader, cacheStatusMiss)
 		tagSpan(ctx, cacheStatusMiss, -1)
 		f.metrics.IncCounter("cache.miss")
+		if p, ok := f.metrics.(interface{ IncCacheResult(string, string) }); ok {
+			p.IncCacheResult("miss", statusCodeClass(entry.StatusCode))
+		}
+		f.updateLRUBytesGauge(ctx)
 		ctx.Serve(headBodyOmitted(ctx.Request().Method, rsp))
 	case <-ctx.Request().Context().Done():
 		// Client disconnected. Remaining waiters still receive their result.
@@ -699,7 +754,6 @@ func (f *cacheFilter) Response(ctx filters.FilterContext) {
 			}
 			if err := f.storage.Set(ctx.Request().Context(), key, stored); err != nil {
 				log.WithError(err).Warn("cache: Set failed (HEAD freshen)")
-				f.metrics.IncCounter("cache.storage_error")
 			}
 		}
 		return
@@ -715,22 +769,18 @@ func (f *cacheFilter) Response(ctx filters.FilterContext) {
 	if isUnsafeMethod(ctx.Request().Method) && rsp.StatusCode < 400 {
 		if err := f.storage.Delete(ctx.Request().Context(), key); err != nil {
 			log.WithError(err).Warn("cache: Delete failed (unsafe method invalidation)")
-			f.metrics.IncCounter("cache.storage_error")
 		}
 		if err := f.storage.Delete(ctx.Request().Context(), "vary:"+key); err != nil {
 			log.WithError(err).Warn("cache: Delete failed (vary sentinel invalidation)")
-			f.metrics.IncCounter("cache.storage_error")
 		}
 		for _, hdrName := range []string{"Location", "Content-Location"} {
 			if loc := rsp.Header.Get(hdrName); loc != "" && sameOrigin(ctx.Request(), loc) {
 				if locKey := cacheKeyForURL(ctx.RouteId(), ctx.Request(), loc, f.keyHeaders); locKey != "" {
 					if err := f.storage.Delete(ctx.Request().Context(), locKey); err != nil {
 						log.WithError(err).Warn("cache: Delete failed (Location invalidation)")
-						f.metrics.IncCounter("cache.storage_error")
 					}
 					if err := f.storage.Delete(ctx.Request().Context(), "vary:"+locKey); err != nil {
 						log.WithError(err).Warn("cache: Delete failed (vary sentinel for Location)")
-						f.metrics.IncCounter("cache.storage_error")
 					}
 				}
 			}
@@ -789,7 +839,6 @@ func (f *cacheFilter) Response(ctx filters.FilterContext) {
 		}
 		if err := f.storage.Set(ctx.Request().Context(), "vary:"+baseKey, sentinel); err != nil {
 			log.WithError(err).Warn("cache: Set failed (vary sentinel)")
-			f.metrics.IncCounter("cache.storage_error")
 		}
 	}
 
@@ -819,7 +868,6 @@ func (f *cacheFilter) Response(ctx filters.FilterContext) {
 	}
 	if err := f.storage.Set(ctx.Request().Context(), storeKey, entry); err != nil {
 		log.WithError(err).Warn("cache: Set failed (response store)")
-		f.metrics.IncCounter("cache.storage_error")
 	}
 }
 
@@ -1021,7 +1069,6 @@ func (f *cacheFilter) doRevalidate(key string, req *http.Request, backendURL, ou
 		}
 		if err := f.storage.Set(context.Background(), key, entry); err != nil {
 			log.WithError(err).Warn("cache: Set failed (background revalidation)")
-			f.metrics.IncCounter("cache.storage_error")
 		}
 		return nil, nil
 	})