Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions common/metrics/metrics.go
Original file line number Diff line number Diff line change
Expand Up @@ -213,6 +213,18 @@ var (
},
)

// CleanerRowsScanned counts the metadata rows each cleaner scan walked:
// "ttl" for the per-tick expiry sweep, "reconcile" for the startup and
// hourly size reconciliation. Both walk only the metadata prefix, so this
// tracks the key count rather than the whole keyspace.
CleanerRowsScanned = promauto.NewCounterVec(
prometheus.CounterOpts{
Name: "ocache_cleaner_rows_scanned_total",
Help: "Metadata rows walked by cleaner scans",
},
[]string{"scan"},
)

// Disk Usage Metrics
DiskUsageBytes = promauto.NewGaugeVec(
prometheus.GaugeOpts{
Expand Down
37 changes: 19 additions & 18 deletions storage/cleaner.go
Original file line number Diff line number Diff line change
Expand Up @@ -413,24 +413,25 @@ func (c *Cleaner) cleanupExpiredKeys() {
batch.Clear()
}

for it.SeekToFirst(); it.Valid(); it.Next() {
// Walk only the metadata rows. The eviction indexes, back-references,
// compaction and deletion queues share this database, and under LRU those
// rows outnumber the metadata rows two to one, so walking the whole
// keyspace read three rows for every one this sweep could act on, every
// tick, on a cache whose row count grows with its size.
metaPrefix := []byte(keys.MetadataPrefix)
scanned := 0
defer func() { metrics.CleanerRowsScanned.WithLabelValues("ttl").Add(float64(scanned)) }()
for it.Seek(metaPrefix); it.ValidForPrefix(metaPrefix); it.Next() {
// Check if we're shutting down
select {
case <-c.closeCh:
zlog.Info().Msg("cleaner: cleanup interrupted by shutdown")
return
default:
}
scanned++
keyBytes := it.Key().Data()

// Only process user metadata keys
if !keys.IsMetadataKey(keyBytes) {
// Skip all non-metadata keys (including other internal keys)
it.Key().Free()
it.Value().Free()
continue
}

// Extract the original user key
key := keys.ExtractUserKey(keyBytes)

Expand Down Expand Up @@ -510,6 +511,7 @@ func (c *Cleaner) cleanupExpiredKeys() {
metrics.CleanerBytesFreed.WithLabelValues("ttl").Add(float64(committedBytes))

zlog.Info().
Int("scanned", scanned).
Int("cleaned", committedCleaned).
Int64("bytes_freed", committedBytes).
Dur("duration_ms", duration).
Expand Down Expand Up @@ -615,7 +617,12 @@ func (c *Cleaner) reconcileFromMetadata() {
referencedRaw = make(map[string]int64)
}

for it.SeekToFirst(); it.Valid(); it.Next() {
// Metadata rows only, for the same reason as the TTL sweep: the other
// keyspaces would otherwise be read and discarded row by row.
metaPrefix := []byte(keys.MetadataPrefix)
scanned := 0
defer func() { metrics.CleanerRowsScanned.WithLabelValues("reconcile").Add(float64(scanned)) }()
for it.Seek(metaPrefix); it.ValidForPrefix(metaPrefix); it.Next() {
// Check if we're shutting down
select {
case <-c.closeCh:
Expand All @@ -624,16 +631,9 @@ func (c *Cleaner) reconcileFromMetadata() {
return
default:
}
scanned++
keyBytes := it.Key().Data()

// Only process user metadata keys
if !keys.IsMetadataKey(keyBytes) {
// Skip all non-metadata keys (including other internal keys)
it.Key().Free()
it.Value().Free()
continue
}

// This scan sums only value_length across every metadata row, so read
// it directly off the wire rather than fully decoding each message —
// that skips a Data-payload copy per inline row (up to the 64 KiB inline
Expand Down Expand Up @@ -703,6 +703,7 @@ func (c *Cleaner) reconcileFromMetadata() {
c.refreshSizeMetrics()

event := zlog.Info().
Int("scanned", scanned).
Int64("total_size", totalSize).
Int64("drift", tracked-totalSize).
Dur("duration_ms", time.Since(start))
Expand Down
75 changes: 75 additions & 0 deletions storage/cleaner_scan_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,75 @@
// Copyright 2026 Tigris Data, Inc.
// SPDX-License-Identifier: Apache-2.0

package storage

import (
"bytes"
"fmt"
"testing"

grocksdb "github.com/linxGnu/grocksdb"
"github.com/prometheus/client_golang/prometheus/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/tigrisdata/ocache/common/metrics"
"github.com/tigrisdata/ocache/storage/metadata"
)

// seedForeignRows writes rows in keyspaces that sort on either side of the
// metadata prefix, standing in for the eviction indexes, back-references and
// queues that share the database with the metadata rows.
func seedForeignRows(t *testing.T, s *Storage, n int) {
t.Helper()
wo := grocksdb.NewDefaultWriteOptions()
defer wo.Destroy()
for i := 0; i < n; i++ {
require.NoError(t, s.meta.Handle().Put(wo, []byte(fmt.Sprintf("!aaa/%06d", i)), []byte("x")))
require.NoError(t, s.meta.Handle().Put(wo, []byte(fmt.Sprintf("!zzz/%06d", i)), []byte("x")))
}
}

func countRows(t *testing.T, s *Storage, prefix string) int {
t.Helper()
ro := metadata.CreateReadOptions(false, false)
defer ro.Destroy()
it := s.meta.Handle().NewIterator(ro)
defer it.Close()
p := []byte(prefix)
n := 0
for it.Seek(p); it.ValidForPrefix(p); it.Next() {
n++
}
return n
}

// TestCleanerScans_WalkOnlyMetadataRows: the TTL sweep and the size
// reconciliation read exactly the metadata rows, however many rows the other
// keyspaces hold around them, and leave those rows alone.
func TestCleanerScans_WalkOnlyMetadataRows(t *testing.T) {
s := newExpiryTestStorage(t)
seedForeignRows(t, s, 500)
for i := 0; i < 3; i++ {
require.NoError(t, s.Put(fmt.Sprintf("live-%d", i), bytes.NewReader([]byte("v")), 0))
}
writeExpiredInline(t, s, "expired-a", "a")
writeExpiredInline(t, s, "expired-b", "b")

ttl := metrics.CleanerRowsScanned.WithLabelValues("ttl")
before := testutil.ToFloat64(ttl)
s.cleaner.cleanupExpiredKeys()
assert.Equal(t, 5.0, testutil.ToFloat64(ttl)-before, "the TTL sweep walks the metadata rows and nothing else")
_, found := readValue(t, s, "expired-a")
assert.False(t, found, "expired row deleted")
v, found := readValue(t, s, "live-0")
assert.True(t, found)
assert.Equal(t, "v", v)

reconcile := metrics.CleanerRowsScanned.WithLabelValues("reconcile")
before = testutil.ToFloat64(reconcile)
s.cleaner.reconcileFromMetadata()
assert.Equal(t, 3.0, testutil.ToFloat64(reconcile)-before, "reconcile walks the metadata rows and nothing else")

assert.Equal(t, 500, countRows(t, s, "!aaa/"))
assert.Equal(t, 500, countRows(t, s, "!zzz/"))
}
Loading