diff --git a/docs/metrics.md b/docs/metrics.md index 3d563f7a..90ad8193 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -318,6 +318,36 @@ the unary byte-write fast path or the streaming fallback). Flood-safe, alertable populate-write failures; the per-block error is logged only at Debug. Healthy value is 0 — alert on any sustained rate. +#### tag_cache_block_refetch_total + +**Type:** Counter (labeled by `previous`) + +Upstream block fetches for a block whose previous populate **on this pod** did not land in +cache, so the fetch is a direct cost of a warm-caching failure rather than a cold miss. +Counted at the start of a real upstream block fetch (after the presence recheck finds the +block absent). + +| `previous` | Meaning | +|------------|---------| +| `write_skipped` | The earlier fetch was declined a populate slot (budget/concurrency pressure), so its bytes were served but never cached | +| `write_failed` | The earlier fetch's `PutBlock` failed (also counted by `tag_cache_block_populate_failed_total`) | + +Caveats: +- The memory of earlier failures is a per-pod, fixed-size table (1M slots), cleared on + restart. A block whose earlier populate failed on another pod, or whose slot was + overwritten by a colliding key, is not counted, so this is a lower bound. +- A successful `PutBlock` clears the block's entry, so a later eviction of that block is + not attributed here. + +```promql +# Re-fetches caused by warm-caching failures, by cause +sum by (previous) (rate(tag_cache_block_refetch_total[5m])) + +# Share of upstream block fetches that are re-fetches of a block that should have been warm +sum(rate(tag_cache_block_refetch_total[5m])) / +sum(rate(tag_cache_block_populated_total[5m]) + rate(tag_cache_block_populate_failed_total[5m])) +``` + #### tag_cache_block_prefetch_windows_total **Type:** Counter (labeled by `outcome`) diff --git a/metrics/metrics.go b/metrics/metrics.go index 0670551c..e1d741de 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -311,6 +311,19 @@ var ( }, ) + // CacheBlockRefetch counts upstream block fetches for a block whose previous + // populate on this pod was skipped (write_skipped: populate budget declined) + // or failed (write_failed: PutBlock error) — re-fetches caused directly by a + // warm-caching failure. Attribution uses a bounded per-pod table, so it is a + // lower bound and resets on restart. + CacheBlockRefetch = promauto.NewCounterVec( + prometheus.CounterOpts{ + Name: "tag_cache_block_refetch_total", + Help: "Upstream block fetches caused by a previous populate of the same block being skipped or failing (block-aligned caching)", + }, + []string{"previous"}, + ) + // CacheBlockPrefetchWindows counts multi-block serves by the prefetch window they were // granted from the shared staging budget: "full" (requested window granted), "shrunk" // (a smaller window than requested), or "sequential" (budget declined every window size — diff --git a/proxy/block_cache.go b/proxy/block_cache.go index 0581d568..530494d2 100644 --- a/proxy/block_cache.go +++ b/proxy/block_cache.go @@ -1273,6 +1273,9 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a s.finishBlockFetch(blockKey, state, nil) return } + // This is a real upstream fetch; count it if the previous populate of this + // block on this pod was skipped or failed. + s.blockWriteFailures.countBlockRefetch(blockKey) bStart, bEnd := blockBounds(blockIdx, meta.BlockSize, meta.ContentLength) blockLen := bEnd - bStart + 1 @@ -1283,6 +1286,7 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a reserveWeight := s.populateWeight(blockLen) if !s.acquireCacheSlot(fetchCtx, reserveWeight, priorityReadMiss) { metrics.RecordCachePopulateSkipped(priorityReadMiss.metricSource()) + s.blockWriteFailures.record(blockKey, blockWriteSkipped) fail(errCachePopulateDeclined) return } @@ -1348,11 +1352,13 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a if perr != nil { putBlockBuf(bufp) metrics.CacheBlockPopulateFailed.Inc() + s.blockWriteFailures.record(blockKey, blockWriteFailed) state.completeServe(nil, nil, perr) s.finishBlockFetch(blockKey, state, perr) return } metrics.CacheBlockPopulated.Inc() + s.blockWriteFailures.clear(blockKey) metrics.CacheBlockBytesPopulated.Add(float64(blockLen)) state.completeServe(bufp, blockBuf, nil) s.finishBlockFetch(blockKey, state, nil) @@ -1367,9 +1373,11 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a perr := s.cache.PutBlock(fetchCtx, bucket, key, meta.ETag, meta.BlockSize, blockIdx, blockBuf, ttl) if perr != nil { metrics.CacheBlockPopulateFailed.Inc() + s.blockWriteFailures.record(blockKey, blockWriteFailed) log.Debug().Err(perr).Str("bucket", bucket).Str("key", key).Int64("block", blockIdx).Msg("Detached remote block cache write failed") } else { metrics.CacheBlockPopulated.Inc() + s.blockWriteFailures.clear(blockKey) metrics.CacheBlockBytesPopulated.Add(float64(blockLen)) } s.releaseCacheSlot(reserveWeight) diff --git a/proxy/block_write_failures.go b/proxy/block_write_failures.go new file mode 100644 index 00000000..0ec29fa0 --- /dev/null +++ b/proxy/block_write_failures.go @@ -0,0 +1,84 @@ +package proxy + +import ( + "hash/maphash" + "sync/atomic" + + "github.com/tigrisdata/tag/metrics" +) + +// blockWriteFailureEntries is the slot count of a Service's block write-failure +// table (8 bytes each: 8 MiB). +const blockWriteFailureEntries = 1 << 20 + +// blockWriteFailure is why a fetched block was not cached. Codes fit in the low +// 3 bits of a table word; 0 is reserved for an empty slot. +type blockWriteFailure uint64 + +const ( + blockWriteNone blockWriteFailure = iota + blockWriteSkipped + blockWriteFailed +) + +const blockWriteFailureMask = 7 + +var blockWriteFailureSeed = maphash.MakeSeed() + +// blockWriteFailureTable remembers, per block key, that its last populate was +// skipped or failed on this pod. It is a fixed-size direct-mapped array of +// atomics: slot = hash & (len-1), word = (hash with the low 3 bits cleared) | +// reason. A colliding key overwrites the slot and a forgotten key reads as none, +// so refetch counts are a lower bound. Lock-free; a nil table records nothing. +type blockWriteFailureTable struct { + slots []atomic.Uint64 + mask uint64 +} + +// newBlockWriteFailureTable sizes the table to n, which must be a power of two. +func newBlockWriteFailureTable(n int) *blockWriteFailureTable { + return &blockWriteFailureTable{slots: make([]atomic.Uint64, n), mask: uint64(n - 1)} +} + +func (t *blockWriteFailureTable) record(blockKey string, r blockWriteFailure) { + if t == nil { + return + } + h := maphash.String(blockWriteFailureSeed, blockKey) + t.slots[h&t.mask].Store(h&^blockWriteFailureMask | uint64(r)) +} + +func (t *blockWriteFailureTable) lookup(blockKey string) blockWriteFailure { + if t == nil { + return blockWriteNone + } + h := maphash.String(blockWriteFailureSeed, blockKey) + w := t.slots[h&t.mask].Load() + if w != 0 && w&^blockWriteFailureMask == h&^blockWriteFailureMask { + return blockWriteFailure(w & blockWriteFailureMask) + } + return blockWriteNone +} + +// clear empties blockKey's slot if it still holds blockKey. +func (t *blockWriteFailureTable) clear(blockKey string) { + if t == nil { + return + } + h := maphash.String(blockWriteFailureSeed, blockKey) + slot := &t.slots[h&t.mask] + if w := slot.Load(); w != 0 && w&^blockWriteFailureMask == h&^blockWriteFailureMask { + slot.CompareAndSwap(w, 0) + } +} + +// countBlockRefetch increments tag_cache_block_refetch_total when blockKey's +// previous populate on this pod was skipped or failed. +func (t *blockWriteFailureTable) countBlockRefetch(blockKey string) { + switch t.lookup(blockKey) { + case blockWriteSkipped: + metrics.CacheBlockRefetch.WithLabelValues("write_skipped").Inc() + case blockWriteFailed: + metrics.CacheBlockRefetch.WithLabelValues("write_failed").Inc() + } +} diff --git a/proxy/service.go b/proxy/service.go index eb9e9958..59ea36e3 100644 --- a/proxy/service.go +++ b/proxy/service.go @@ -123,6 +123,10 @@ type Service struct { // already examined. Without it every tail read of a fully-warmed object re-probes // its metadata blocks, which in cluster mode are mostly remote. recentFooterWork *expirable.LRU[string, struct{}] + // blockWriteFailures remembers block keys whose populate write was skipped + // or failed, so a later re-fetch is counted as a warm-caching failure. + // Nil (as in bare test Services) records nothing. + blockWriteFailures *blockWriteFailureTable } // NewService creates a new proxy service. @@ -214,6 +218,7 @@ func NewService(forwarder RequestForwarder, cache *cache.Cache, cfg *config.Conf blockFetches: make(map[string]*blockFetchState), retierClaims: make(map[string]int), retierInflight: make(map[string]context.CancelFunc), + blockWriteFailures: newBlockWriteFailureTable(blockWriteFailureEntries), } // Allocated only when the feature that uses it is on. if cfg.Cache.ParquetOptimization {