diff --git a/CHANGELOG.md b/CHANGELOG.md index 29006a2..620802d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,49 @@ All notable changes to this project will be documented in this file. +## v1.3.0 + +This release simplifies cache construction, makes background cleanup lifecycle explicit, and streamlines metrics +integration. + +### Added + +* **Cache Naming:** Added `WithName` for assigning an optional cache name used by observability integrations. +* **Cleanup Lifecycle:** Added `StartCleanup` and `StopCleanup` for explicitly controlling periodic expiration cleanup. +* **Metrics Sources:** Added `MetricsSource` for exposing cache statistics to metrics integrations. + +### Changed + +* **Constructors:** `New` no longer accepts a cache name. `NewWithDefaultLoader` now accepts the loader directly, + followed by options. +* **Metrics API:** Replaced `StatsProvider` and `MetricsRegistration` with `Metrics.Register(MetricsSource) error`. +* **Background Cleanup:** Cleanup is no longer started automatically during cache construction. `WithCleanupInterval` + configures the cleanup interval, while `StartCleanup` runs the blocking cleanup loop explicitly. +* **Cleanup Defaults:** Background cleanup uses a one-minute interval by default. +* **Benchmarks:** Updated benchmark suites and comparisons for the current cache API. + +### Removed + +* **Cache Close:** Removed `Cache.Close`; background cleanup is stopped explicitly with `StopCleanup`. + +## extra/paceotel/v1.3.0 + +This release aligns `paceotel` with the updated `pacecache` metrics API and simplifies metrics registration across +multiple caches. + +### Added + +* **Reusable Metrics:** A single `Metrics` instance can be shared across multiple caches. +* **Source Validation:** Duplicate cache names are rejected within a `Metrics` instance, while a single unnamed cache is + supported. + +### Changed + +* **Metrics Registration:** Each cache now uses its own OpenTelemetry callback registration. +* **Unregister Lifecycle:** `Metrics.Unregister` removes all registered callbacks, releases references to registered + caches, and prevents further registrations. +* **Core Dependency:** Updated `github.com/mkbeh/pacecache` to v1.3.0. + ## v1.2.1 This release streamlines cache removal and makes cache-aside loading configurable through default and per-call loaders. diff --git a/README.md b/README.md index 2825c7b..f0fed01 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ The library provides an intuitive API with predictable behavior under high concu * **Generic API:** Type-safe caching with comparable keys and arbitrary value types. * **Bounded LRU:** Exact per-segment LRU within a fixed total capacity. * **Expiration:** Default and per-entry TTLs, jitter, sliding expiration, refresh, and no-expiration entries. -* **Cleanup:** Lazy expiration, explicit cleanup, and an optional background worker. +* **Cleanup:** Lazy expiration, explicit cleanup, and optional background cleanup. * **Cache-Aside:** Coalesces concurrent misses for the same key into a single load. * **Safe Updates:** Publication barriers prevent stale loads from overwriting newer cache state. * **Observability:** Built-in statistics with optional OpenTelemetry metrics. @@ -42,15 +42,14 @@ go get github.com/mkbeh/pacecache/extra/paceotel ## Usage -Create a cache with `pacecache.New` and close it when it is no longer needed: +Create a cache with `pacecache.New`: ```go -cache, err := pacecache.New[string, string]("cache") +cache, err := pacecache.New[string, string]() if err != nil { panic(err) } -defer cache.Close() ``` @@ -64,11 +63,9 @@ deadlines and reduce synchronized expiration bursts. Individual entries can use ```go cache, _ := pacecache.New[string, string]( - "cache", pacecache.WithTTL(5*time.Minute), pacecache.WithJitter(30*time.Second), ) -defer cache.Close() ``` @@ -86,6 +83,7 @@ value, found := cache.Get("key1") // Read a value together with its expiration metadata. entry, found := cache.GetEntry("key1") +fmt.Println(entry.Value(), entry.ExpiresAt()) // Check existence without updating LRU or TTL. exists := cache.Exists("key2") @@ -131,14 +129,12 @@ If the same loader is reused across calls, configure it once with `NewWithDefaul ```go cache, _ := pacecache.NewWithDefaultLoader[string, string]( - "cache", func(ctx context.Context, key string) (string, bool, error) { // Fetch data from a database, file, or remote service. return "loaded value", true, nil }, pacecache.WithTTL(5*time.Minute), ) -defer cache.Close() // Return the cached value or invoke the configured loader on a miss. value, found, err := cache.GetOrLoad(ctx, "key") @@ -155,22 +151,25 @@ if found { Missing results and loader errors are returned without being cached. Concurrent misses for the same key share a single loader execution, avoiding duplicate requests to the upstream source. -Expired entries are never returned and are removed lazily when encountered. Periodic background cleanup can be enabled -for entries that may remain untouched: +Expired entries are removed lazily when encountered. Background cleanup can be started with `StartCleanup`. Since +`StartCleanup` blocks until `StopCleanup` is called, it is usually launched in a separate goroutine: ```go cache, _ := pacecache.New[string, string]( - "cache", pacecache.WithTTL(5*time.Minute), - pacecache.WithCleanupInterval(time.Minute), // background cleanup ) -defer cache.Close() + +// Start automatic deletion of expired items. +go cache.StartCleanup() + +// Stop automatic deletion of expired items. +cache.StopCleanup() ``` -Background cleanup is optional. Expired entries can also be reclaimed explicitly with `DeleteExpired`. `Close` stops -the cleanup worker and waits for it to exit. +Background cleanup is optional. Expired entries can also be reclaimed explicitly with `DeleteExpired`. + ## Concurrency semantics The cache coordinates concurrent loads and mutations to prevent duplicate upstream work and stale values from diff --git a/benchmarks/performance/hitratio/assets/hit-ratio.png b/benchmarks/performance/hitratio/assets/hit-ratio.png index 2d7e61a..c661000 100644 Binary files a/benchmarks/performance/hitratio/assets/hit-ratio.png and b/benchmarks/performance/hitratio/assets/hit-ratio.png differ diff --git a/benchmarks/performance/hitratio/internal/policy/policy.go b/benchmarks/performance/hitratio/internal/policy/policy.go index 82658c4..c26c1b1 100644 --- a/benchmarks/performance/hitratio/internal/policy/policy.go +++ b/benchmarks/performance/hitratio/internal/policy/policy.go @@ -15,7 +15,6 @@ type Policy struct { func New(capacity int, segments int) (*Policy, error) { cache, err := pacecache.New[uint64, uint64]( - "hit-ratio", pacecache.WithMaxEntries(capacity), pacecache.WithSegmentCount(segments), ) @@ -60,7 +59,3 @@ func (p *Policy) Ratio() float64 { return 100 * float64(p.hits) / float64(total) } - -func (p *Policy) Close() { - p.cache.Close() -} diff --git a/benchmarks/performance/hitratio/internal/simulator/simulator.go b/benchmarks/performance/hitratio/internal/simulator/simulator.go index 5990855..7c483d9 100644 --- a/benchmarks/performance/hitratio/internal/simulator/simulator.go +++ b/benchmarks/performance/hitratio/internal/simulator/simulator.go @@ -61,7 +61,6 @@ func (s Simulator) simulateCapacity(capacity int) (result, error) { if err != nil { return result{}, fmt.Errorf("create policy for capacity %d: %w", capacity, err) } - defer p.Close() generator := trace.NewZipf( s.cfg.Zipf.S, diff --git a/benchmarks/performance/memory/assets/memory.png b/benchmarks/performance/memory/assets/memory.png index b6598f5..9a89417 100644 Binary files a/benchmarks/performance/memory/assets/memory.png and b/benchmarks/performance/memory/assets/memory.png differ diff --git a/benchmarks/performance/memory/main.go b/benchmarks/performance/memory/main.go index a84dbd7..e9a20fd 100644 --- a/benchmarks/performance/memory/main.go +++ b/benchmarks/performance/memory/main.go @@ -38,7 +38,6 @@ func main() { runtime.ReadMemStats(&before) cache, err := pacecache.New[string, string]( - "memory", pacecache.WithMaxEntries(*capacity), pacecache.WithSegmentCount(segmentCount), pacecache.WithTTL(expiration), @@ -46,7 +45,6 @@ func main() { if err != nil { log.Fatalf("create cache: %v", err) } - defer cache.Close() for index := range *capacity { key := keys[index] diff --git a/benchmarks/performance/throughput/assets/throughput.png b/benchmarks/performance/throughput/assets/throughput.png index 8617dba..fafb011 100644 Binary files a/benchmarks/performance/throughput/assets/throughput.png and b/benchmarks/performance/throughput/assets/throughput.png differ diff --git a/benchmarks/performance/throughput/throughput_test.go b/benchmarks/performance/throughput/throughput_test.go index b08af3d..622b0b1 100644 --- a/benchmarks/performance/throughput/throughput_test.go +++ b/benchmarks/performance/throughput/throughput_test.go @@ -78,7 +78,6 @@ func runThroughputBenchmark( b.Helper() cache := newThroughputCache(b, maxEntries, data) - b.Cleanup(cache.Close) var workers atomic.Uint64 @@ -133,7 +132,6 @@ func newThroughputCache( for range throughputPopulationAttempts { cache, err := pacecache.New[string, string]( - "throughput", pacecache.WithMaxEntries(maxEntries), pacecache.WithSegmentCount(throughputSegments), ) @@ -153,7 +151,6 @@ func newThroughputCache( return cache } - cache.Close() } b.Fatalf( diff --git a/cache.go b/cache.go index 74ac090..3a0428f 100644 --- a/cache.go +++ b/cache.go @@ -24,49 +24,42 @@ const ( // // Cache is safe for concurrent use. A Cache must not be copied after creation. type Cache[K comparable, V any] struct { - name string loader Loader[K, V] store *storage[K, V] states []cacheState[K, V] stats *statsCollector - cleanupPolicy cleanupPolicy - cleanup *cleanupWorker[K, V] - - metrics MetricsRegistration - closeOnce sync.Once - ttl time.Duration jitter time.Duration + + cleanupPolicy cleanupPolicy + cleanupInterval time.Duration + + cleanupMu sync.Mutex + cleanup *cleanupWorker[K, V] } -// New creates a Cache with the given logical name. -// -// Name is used by diagnostics and metrics and must not be empty. +// New creates a Cache. // // Unless overridden by options, New uses the default cache capacity, a single // storage segment, and no time-based expiration. No default loader is -// configured. Metrics and background cleanup are disabled by default. -// -// If metrics or background cleanup are configured, Close must be called to -// release the associated resources. +// configured. Metrics are disabled by default, and background cleanup is not +// started automatically. func New[K comparable, V any]( - name string, options ...Option, ) (*Cache[K, V], error) { - return newCache[K, V](name, nil, options...) + return newCache[K, V](nil, options...) } -// NewWithDefaultLoader creates a Cache with the given logical name and default loader. +// NewWithDefaultLoader creates a Cache with the given default loader. // // The loader is used by GetOrLoad and GetOrLoadEntry when no live cache entry // exists. Per-call loaders may be supplied through GetOrLoadFunc and // GetOrLoadEntryFunc. Loader must not be nil. // -// Name, options, metrics, and background cleanup have the same semantics as New. +// Options, metrics, and background cleanup have the same semantics as New. func NewWithDefaultLoader[K comparable, V any]( - name string, loader Loader[K, V], options ...Option, ) (*Cache[K, V], error) { @@ -74,15 +67,14 @@ func NewWithDefaultLoader[K comparable, V any]( return nil, ErrNoLoader } - return newCache[K, V](name, loader, options...) + return newCache[K, V](loader, options...) } func newCache[K comparable, V any]( - name string, loader Loader[K, V], options ...Option, ) (*Cache[K, V], error) { - settings, err := newCacheSettings(name, options...) + settings, err := newSettings(options...) if err != nil { return nil, fmt.Errorf("pacecache: %w", err) } @@ -93,74 +85,75 @@ func newCache[K comparable, V any]( settings.slidingExpiration, ) - policy := cleanupPolicy{ - batchSize: settings.cleanupBatchSize, - entryBudget: settings.cleanupEntryBudget, - } - cache := &Cache[K, V]{ - name: settings.name, loader: loader, store: store, states: make([]cacheState[K, V], len(store.segments)), stats: newStatsCollector(len(store.segments)), - cleanupPolicy: policy, - ttl: settings.ttl, jitter: settings.jitter, - } - if err := cache.registerMetrics(settings.metrics); err != nil { - return nil, fmt.Errorf("pacecache: register metrics: %w", err) + cleanupPolicy: cleanupPolicy{ + batchSize: settings.cleanupBatchSize, + entryBudget: settings.cleanupEntryBudget, + }, + cleanupInterval: settings.cleanupInterval, } - if settings.cleanupInterval > 0 { - cache.cleanup = newCleanupWorker( - cache.store, - cache.stats, - cache.cleanupPolicy, - settings.cleanupInterval, - ) - cache.cleanup.start() + if err := cache.registerMetrics(settings.name, settings.metrics); err != nil { + return nil, fmt.Errorf("pacecache: register metrics: %w", err) } return cache, nil } -// Name returns the logical cache name. -func (cache *Cache[K, V]) Name() string { - if cache == nil { - return "" +// StartCleanup runs periodic expiration cleanup until StopCleanup is called. +// StartCleanup blocks for the lifetime of the cleanup loop; callers that want +// background cleanup should start it in a goroutine. If cleanup is already +// running, StartCleanup returns immediately. +func (cache *Cache[K, V]) StartCleanup() { + if !cache.initialized() { + return } - return cache.name + cache.cleanupMu.Lock() + if cache.cleanup != nil { + cache.cleanupMu.Unlock() + return + } + + worker := newCleanupWorker( + cache.store, + cache.stats, + cache.cleanupPolicy, + cache.cleanupInterval, + ) + cache.cleanup = worker + cache.cleanupMu.Unlock() + + worker.run() } -// Close releases background resources associated with the cache. Close is -// idempotent. -// -// If background cleanup is configured, Close stops it and waits for the cleaner -// goroutine to exit. If metrics are configured, Close also releases their -// registration. -// -// Close does not clear or disable the cache. Cache operations remain available, -// but stopped background resources are not restarted. -func (cache *Cache[K, V]) Close() { +// StopCleanup stops a running cleanup loop. It blocks until the cleanup loop +// accepts the stop signal. Repeated calls are safe. StopCleanup is a no-op +// when cleanup is not running or Cache is nil. +func (cache *Cache[K, V]) StopCleanup() { if cache == nil { return } - cache.closeOnce.Do(func() { - if cache.cleanup != nil { - cache.cleanup.close() - } + cache.cleanupMu.Lock() + defer cache.cleanupMu.Unlock() - if cache.metrics != nil { - cache.metrics.Close() - } - }) + worker := cache.cleanup + if worker == nil { + return + } + + worker.stopCh <- struct{}{} + cache.cleanup = nil } func (cache *Cache[K, V]) effectiveTTL(expiration time.Duration) time.Duration { @@ -177,23 +170,17 @@ func (cache *Cache[K, V]) effectiveTTL(expiration time.Duration) time.Duration { return jitteredTTL(ttl, cache.jitter) } -func (cache *Cache[K, V]) registerMetrics(metrics Metrics) error { +func (cache *Cache[K, V]) registerMetrics(name string, metrics Metrics) error { if metrics == nil { return nil } - registration, err := metrics.RegisterCache( - cacheStatsProvider[K, V]{ + return metrics.Register( + metricsSource[K, V]{ + name: name, cache: cache, }, ) - if err != nil { - return err - } - - cache.metrics = registration - - return nil } func (cache *Cache[K, V]) initialized() bool { diff --git a/cache_test.go b/cache_test.go index 2b2e55d..402a137 100644 --- a/cache_test.go +++ b/cache_test.go @@ -11,9 +11,6 @@ import ( func TestNilCacheIsSafe(t *testing.T) { var cache *Cache[string, int] - if got := cache.Name(); got != "" { - t.Fatalf("nil Cache.Name() = %q, want empty", got) - } if cache.Exists("key") { t.Fatal("nil Cache.Exists() = true, want false") } @@ -32,8 +29,6 @@ func TestNilCacheIsSafe(t *testing.T) { if _, _, err := cache.GetOrLoadEntry(context.Background(), "key"); !errors.Is(err, ErrNotInitialized) { t.Fatalf("nil Cache.GetOrLoadEntry() error = %v, want ErrNotInitialized", err) } - - cache.Close() } func TestZeroValueCacheIsSafe(t *testing.T) { @@ -56,7 +51,6 @@ func TestZeroValueCacheIsSafe(t *testing.T) { cache.Delete() cache.Delete("key") cache.Clear() - cache.Close() if got := cache.Stats(); got != (Stats{}) { t.Fatalf("zero Cache.Stats() = %+v, want zero Stats", got) @@ -83,20 +77,6 @@ func TestZeroValueCacheLoadReturnsNotInitialized(t *testing.T) { } } -func TestCloseIsIdempotentAndCacheRemainsUsable(t *testing.T) { - cache, err := New[string, int]("users") - if err != nil { - t.Fatalf("New() error = %v", err) - } - - cache.Close() - cache.Close() - cache.Set("key", 1, NoExpiration) - if value, found := cache.Get("key"); value != 1 || !found { - t.Fatalf("Get after Close = (%d, %t), want usable cache", value, found) - } -} - func TestEffectiveTTLAndDeadlineHelpers(t *testing.T) { cache := &Cache[string, int]{ttl: 10 * time.Second} @@ -140,7 +120,7 @@ func TestEffectiveTTLAndDeadlineHelpers(t *testing.T) { } func TestNewEnablesSlidingExpiration(t *testing.T) { - cache := mustNewCache[int](t, "users", WithSlidingExpiration()) + cache := mustNewCache[int](t, WithSlidingExpiration()) for index := range cache.store.segments { if !cache.store.segments[index].slidingExpiration { t.Fatalf("segment %d sliding expiration disabled", index) @@ -149,11 +129,11 @@ func TestNewEnablesSlidingExpiration(t *testing.T) { } func TestNewWrapsConfigurationError(t *testing.T) { - cache, err := New[string, int]("") + cache, err := New[string, int](WithMaxEntries(2), WithSegmentCount(3)) if cache != nil { t.Fatal("cache must be nil for invalid configuration") } - if err == nil || err.Error() != "pacecache: invalid configuration: cache name must not be empty" { + if err == nil || err.Error() != "pacecache: invalid configuration: segment count must not exceed max entries" { t.Fatalf("New() error = %v", err) } } @@ -164,11 +144,10 @@ type testCompositeKey struct { } func TestCacheSupportsInt64Keys(t *testing.T) { - cache, err := New[int64, string]("users", WithMaxEntries(8), WithSegmentCount(2)) + cache, err := New[int64, string](WithMaxEntries(8), WithSegmentCount(2)) if err != nil { t.Fatalf("New() error = %v", err) } - t.Cleanup(cache.Close) cache.Set(42, "Ada", NoExpiration) @@ -189,11 +168,10 @@ func TestCacheSupportsInt64Keys(t *testing.T) { } func TestCacheSupportsComparableStructKeys(t *testing.T) { - cache, err := New[testCompositeKey, int]("users", WithMaxEntries(8), WithSegmentCount(2)) + cache, err := New[testCompositeKey, int](WithMaxEntries(8), WithSegmentCount(2)) if err != nil { t.Fatalf("New() error = %v", err) } - t.Cleanup(cache.Close) stored := testCompositeKey{TenantID: 7, UserID: 42} equal := testCompositeKey{TenantID: 7, UserID: 42} @@ -255,16 +233,14 @@ func (ctx *observedWaitContext) cancel() { const testTimeout = 5 * time.Second -func mustNewCache[V any](t *testing.T, name string, options ...Option) *Cache[string, V] { +func mustNewCache[V any](t *testing.T, options ...Option) *Cache[string, V] { t.Helper() - cache, err := New[string, V](name, options...) + cache, err := New[string, V](options...) if err != nil { t.Fatalf("New() error = %v", err) } - t.Cleanup(cache.Close) - return cache } diff --git a/cleanup.go b/cleanup.go index 947ce05..7ae0f10 100644 --- a/cleanup.go +++ b/cleanup.go @@ -3,6 +3,7 @@ package pacecache import "time" const ( + defaultCleanupInterval = time.Minute defaultCleanupBatchSize = 256 defaultCleanupEntryBudget = 16 * 1024 @@ -21,8 +22,7 @@ type cleanupWorker[K comparable, V any] struct { policy cleanupPolicy interval time.Duration - stop chan struct{} - done chan struct{} + stopCh chan struct{} // scratchSegments is reusable temporary storage for segment indexes that // still have due entries after a multi-segment cleanup pass. @@ -41,8 +41,7 @@ func newCleanupWorker[K comparable, V any]( stats: stats, policy: policy, interval: interval, - stop: make(chan struct{}), - done: make(chan struct{}), + stopCh: make(chan struct{}), } if len(store.segments) > 1 { @@ -52,23 +51,22 @@ func newCleanupWorker[K comparable, V any]( return worker } -func (worker *cleanupWorker[K, V]) start() { - go worker.run() -} - -func (worker *cleanupWorker[K, V]) close() { - close(worker.stop) - <-worker.done -} - func (worker *cleanupWorker[K, V]) run() { timer := time.NewTimer(worker.interval) defer timer.Stop() - defer close(worker.done) for { select { + case <-worker.stopCh: + return + case <-timer.C: + select { + case <-worker.stopCh: + return + default: + } + cutoff := worker.store.now() pending := worker.cleanupQuantum(cutoff) worker.stats.recordCleanupWorker( @@ -82,9 +80,6 @@ func (worker *cleanupWorker[K, V]) run() { } timer.Reset(next) - - case <-worker.stop: - return } } } @@ -108,10 +103,6 @@ func (worker *cleanupWorker[K, V]) cleanupSingleSegment(cutoff int64) bool { stats := worker.stats.segment(0) for { - if worker.stopped() { - return false - } - if remaining == 0 || cleanupTimeBudgetExceeded(startedAt) { return true } @@ -140,11 +131,6 @@ func (worker *cleanupWorker[K, V]) cleanupSegments(cutoff int64) bool { start := worker.nextSegment for offset := range segmentCount { - if worker.stopped() { - worker.scratchSegments = pending[:0] - return false - } - if remaining == 0 || cleanupTimeBudgetExceeded(startedAt) { worker.nextSegment = (start + offset) % segmentCount worker.scratchSegments = pending[:0] @@ -173,11 +159,6 @@ func (worker *cleanupWorker[K, V]) cleanupSegments(cutoff int64) bool { next := pending[:0] for _, index := range pending { - if worker.stopped() { - worker.scratchSegments = pending[:0] - return false - } - if remaining == 0 || cleanupTimeBudgetExceeded(startedAt) { // Resume from the first pending segment that this quantum did // not get a chance to process. @@ -211,15 +192,6 @@ func (worker *cleanupWorker[K, V]) nextDelay() time.Duration { return min(worker.interval, cleanupNextDelay) } -func (worker *cleanupWorker[K, V]) stopped() bool { - select { - case <-worker.stop: - return true - default: - return false - } -} - func cleanupTimeBudgetExceeded(startedAt time.Time) bool { return time.Since(startedAt) >= cleanupTimeBudget } diff --git a/cleanup_test.go b/cleanup_test.go index 2e19c8a..66b70d3 100644 --- a/cleanup_test.go +++ b/cleanup_test.go @@ -2,6 +2,7 @@ package pacecache import ( "runtime" + "sync" "testing" "time" ) @@ -37,22 +38,6 @@ func TestCleanupWorkerNextDelay(t *testing.T) { } } -func TestCleanupWorkerStopped(t *testing.T) { - store := newStorageWithExpirationResolution[string, int](1, 1, time.Nanosecond) - worker := newTestCleanupWorker(store, newStatsCollector(1), time.Second) - - if worker.stopped() { - t.Fatal("new worker unexpectedly stopped") - } - close(worker.stop) - if !worker.stopped() { - t.Fatal("closed worker stop channel not observed") - } - if worker.cleanupQuantum(10) { - t.Fatal("cleanup reported pending work after stop") - } -} - func TestCleanupWorkerEmptyStorage(t *testing.T) { store := &storage[string, int]{} worker := newTestCleanupWorker(store, newStatsCollector(0), time.Second) @@ -61,43 +46,145 @@ func TestCleanupWorkerEmptyStorage(t *testing.T) { } } -func TestCleanupWorkerBackgroundRemovesExpiredEntry(t *testing.T) { - store := newStorageWithExpirationResolution[string, int](1, 1, time.Millisecond) - stats := newStatsCollector(1) - now := store.now() - store.setAt(0, "key", 1, 2*time.Millisecond, deadlineAfter(now, 2*time.Millisecond), stats.segment(0)) +func TestNewDoesNotStartCleanup(t *testing.T) { + cache, err := New[string, int](WithCleanupInterval(time.Millisecond)) + if err != nil { + t.Fatalf("New() error = %v", err) + } - worker := newTestCleanupWorker(store, stats, time.Millisecond) - worker.start() - t.Cleanup(worker.close) + if cache.cleanup != nil { + t.Fatal("cleanup unexpectedly running after New") + } +} + +func TestStartCleanupRemovesExpiredEntry(t *testing.T) { + cache, err := New[string, int]( + WithMaxEntries(1), + WithCleanupInterval(time.Millisecond), + ) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + cache.store.enableExpirationIndex(time.Nanosecond) + cache.Set("key", 1, time.Millisecond) + time.Sleep(2 * time.Millisecond) + + done := startTestCleanup(t, cache) eventually(t, 500*time.Millisecond, func() bool { - segment := &store.segments[0] + segment := &cache.store.segments[0] segment.mu.Lock() defer segment.mu.Unlock() + return len(segment.entries) == 0 }) - if stats.segment(0).expirationCount != 1 { - t.Fatalf("expirationCount = %d, want 1", stats.segment(0).expirationCount) + cache.StopCleanup() + waitTestSignal(t, done) + + if cache.stats.segment(0).expirationCount != 1 { + t.Fatalf( + "expirationCount = %d, want 1", + cache.stats.segment(0).expirationCount, + ) } - if stats.cleanupWorkerRunCount.Load() == 0 { + if cache.stats.cleanupWorkerRunCount.Load() == 0 { t.Fatal("cleanup worker run was not recorded") } } -func TestCacheCloseStopsBackgroundCleanup(t *testing.T) { - cache, err := New[string, int]("users", WithCleanupInterval(time.Millisecond)) +func TestStopCleanupWithoutStartIsNoop(t *testing.T) { + cache := mustNewCache[int](t) + + done := make(chan struct{}) + go func() { + cache.StopCleanup() + close(done) + }() + + waitTestSignal(t, done) +} + +func TestStopCleanupStopsRunningWorker(t *testing.T) { + cache, err := New[string, int](WithCleanupInterval(time.Hour)) if err != nil { t.Fatalf("New() error = %v", err) } - if cache.cleanup == nil { - t.Fatal("cleanup worker is nil") + + done := startTestCleanup(t, cache) + + cache.StopCleanup() + waitTestSignal(t, done) + + cache.StopCleanup() +} + +func TestStopCleanupConcurrent(t *testing.T) { + cache, err := New[string, int](WithCleanupInterval(time.Hour)) + if err != nil { + t.Fatalf("New() error = %v", err) } - cache.Close() - waitTestSignal(t, cache.cleanup.done) - cache.Close() + done := startTestCleanup(t, cache) + + const callers = 16 + var group sync.WaitGroup + for range callers { + group.Add(1) + go func() { + defer group.Done() + cache.StopCleanup() + }() + } + + waitTestGroup(t, &group) + waitTestSignal(t, done) +} + +func TestStartCleanupCanRestart(t *testing.T) { + cache, err := New[string, int](WithCleanupInterval(time.Hour)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + firstDone := startTestCleanup(t, cache) + cache.StopCleanup() + waitTestSignal(t, firstDone) + + secondDone := startTestCleanup(t, cache) + cache.StopCleanup() + waitTestSignal(t, secondDone) +} + +func TestStartCleanupReturnsWhenAlreadyRunning(t *testing.T) { + cache, err := New[string, int](WithCleanupInterval(time.Hour)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + + firstDone := startTestCleanup(t, cache) + + secondDone := make(chan struct{}) + go func() { + cache.StartCleanup() + close(secondDone) + }() + + waitTestSignal(t, secondDone) + + cache.StopCleanup() + waitTestSignal(t, firstDone) +} + +func TestCleanupLifecycleNilAndZeroValueSafe(_ *testing.T) { + var nilCache *Cache[string, int] + nilCache.StartCleanup() + nilCache.StopCleanup() + + var zero Cache[string, int] + zero.StartCleanup() + zero.StopCleanup() } func TestCleanupWorkerDrainsActiveSegmentAcrossBatches(t *testing.T) { @@ -180,17 +267,23 @@ func TestCleanupWorkerRunSchedulesContinuationForBacklog(t *testing.T) { } worker := newTestCleanupWorker(store, stats, time.Millisecond) - worker.start() - t.Cleanup(worker.close) + done := make(chan struct{}) + go func() { + worker.run() + close(done) + }() + t.Cleanup(func() { + worker.stopCh <- struct{}{} + waitTestSignal(t, done) + }) eventually(t, time.Second, func() bool { return stats.cleanupWorkerPendingCount.Load() > 0 && stats.cleanupWorkerRunCount.Load() > 1 }) } -func TestCacheCleanupWorkerUsesConfiguredLimits(t *testing.T) { +func TestCacheCleanupUsesConfiguredLimits(t *testing.T) { cache, err := New[string, int]( - "users", WithCleanupInterval(time.Hour), WithCleanupBatchSize(7), WithCleanupEntryBudget(11), @@ -198,13 +291,15 @@ func TestCacheCleanupWorkerUsesConfiguredLimits(t *testing.T) { if err != nil { t.Fatalf("New() error = %v", err) } - t.Cleanup(cache.Close) - if cache.cleanup == nil { - t.Fatal("cleanup worker is nil") + if cache.cleanupInterval != time.Hour { + t.Fatalf("cleanup interval = %v, want 1h", cache.cleanupInterval) } - if cache.cleanup.policy.batchSize != 7 || cache.cleanup.policy.entryBudget != 11 { - t.Fatalf("cleanup policy = %+v, want batch=7 budget=11", cache.cleanup.policy) + if cache.cleanupPolicy.batchSize != 7 || cache.cleanupPolicy.entryBudget != 11 { + t.Fatalf( + "cleanup policy = %+v, want batch=7 budget=11", + cache.cleanupPolicy, + ) } } @@ -242,6 +337,30 @@ func TestCleanupWorkerHonorsConfiguredEntryBudget(t *testing.T) { } } +func startTestCleanup[K comparable, V any]( + t *testing.T, + cache *Cache[K, V], +) <-chan struct{} { + t.Helper() + + done := make(chan struct{}) + go func() { + cache.StartCleanup() + close(done) + }() + + t.Cleanup(cache.StopCleanup) + + eventually(t, testTimeout, func() bool { + cache.cleanupMu.Lock() + defer cache.cleanupMu.Unlock() + + return cache.cleanup != nil + }) + + return done +} + func eventually(t *testing.T, timeout time.Duration, condition func() bool) { t.Helper() diff --git a/delete.go b/delete.go index 9013a00..c43185b 100644 --- a/delete.go +++ b/delete.go @@ -196,7 +196,7 @@ func (cache *Cache[K, V]) Clear() { // index and returns the number of entries removed. // // DeleteExpired is always available; background cleanup does not need to be -// enabled. Logical expiration is independent of physical cleanup: an expired +// running. Logical expiration is independent of physical cleanup: an expired // entry is never returned even if it has not yet been reclaimed. Nearby // expiration deadlines are grouped internally. Bucket eligibility may trail // the exact TTL deadline by up to the internal bucket resolution; actual diff --git a/delete_test.go b/delete_test.go index a5326b2..cf1ef7e 100644 --- a/delete_test.go +++ b/delete_test.go @@ -9,7 +9,7 @@ import ( ) func TestCacheGetAndDelete(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) cache.Set("key", 42, NoExpiration) before := cache.Stats() @@ -40,7 +40,7 @@ func TestCacheGetAndDelete(t *testing.T) { } func TestCacheGetAndDeleteIsAtomic(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) cache.Set("key", 42, NoExpiration) const callers = 32 @@ -94,7 +94,7 @@ func TestCacheGetAndDeleteIsAtomic(t *testing.T) { } func TestCacheGetAndDeleteExpiredEntry(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) index := cache.store.segmentIndex("expired") cache.store.setAt( @@ -126,7 +126,7 @@ func TestCacheGetAndDeleteExpiredEntry(t *testing.T) { } func TestCacheGetAndDeleteSupersedesInflightLoad(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) started := make(chan struct{}) release := make(chan struct{}) @@ -200,7 +200,7 @@ func TestDeleteMultipleKeysAndDuplicates(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - cache := mustNewCache[int](t, "users", test.options...) + cache := mustNewCache[int](t, test.options...) cache.Set("a", 1, NoExpiration) cache.Set("b", 2, NoExpiration) cache.Set("c", 3, NoExpiration) @@ -224,7 +224,7 @@ func TestDeleteMultipleKeysAndDuplicates(t *testing.T) { } func TestConcurrentMultiKeyDeletionLockOrderDoesNotDeadlock(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(16), WithSegmentCount(8)) + cache := mustNewCache[int](t, WithMaxEntries(16), WithSegmentCount(8)) cache.Set("a", 1, NoExpiration) cache.Set("b", 2, NoExpiration) @@ -243,7 +243,7 @@ func TestConcurrentMultiKeyDeletionLockOrderDoesNotDeadlock(t *testing.T) { } func TestCacheClear(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(2)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(2)) cache.Set("a", 1, NoExpiration) cache.Set("b", 2, NoExpiration) @@ -279,7 +279,6 @@ func TestCacheClear(t *testing.T) { func TestCacheDeleteExpired(t *testing.T) { store := newStorageWithExpirationResolution[string, int](4, 1, time.Nanosecond) cache := &Cache[string, int]{ - name: "test", store: store, states: make([]cacheState[string, int], 1), stats: newStatsCollector(1), diff --git a/doc.go b/doc.go index d4c00aa..03c9467 100644 --- a/doc.go +++ b/doc.go @@ -5,12 +5,17 @@ // loading with configurable default or per-call loaders, duplicate load // suppression, and explicit removal. // +// Background expiration cleanup is optional and is not started automatically. +// StartCleanup runs the cleanup loop and blocks until StopCleanup is called. +// Callers that want background cleanup should start it in a goroutine. +// WithCleanupInterval configures the regular cleanup interval. +// // Cache mutations act as publication barriers for concurrent loads, preventing // superseded loader results from overwriting newer cache state. // // Cache statistics are collected locally and exposed through Cache.Stats. -// Optional metrics integrations register during New and observe those snapshots -// without adding telemetry calls to the cache request path. +// Optional metrics integrations register when a cache is created and observe +// those snapshots without adding telemetry calls to the cache request path. // // The cache is local to one application process. It does not provide // distributed cache coherence between application instances. diff --git a/examples/basic/main.go b/examples/basic/main.go index e39bae7..3e6e17d 100644 --- a/examples/basic/main.go +++ b/examples/basic/main.go @@ -36,7 +36,6 @@ func run(ctx context.Context) error { } users, err := pacecache.NewWithDefaultLoader[int64, user]( - "users", repository.find, pacecache.WithMaxEntries(128), pacecache.WithTTL(30*time.Second), @@ -45,7 +44,6 @@ func run(ctx context.Context) error { if err != nil { return fmt.Errorf("create users cache: %w", err) } - defer users.Close() // The first lookup loads the user from the underlying repository. first, found, err := users.GetOrLoad(ctx, 42) diff --git a/examples/otel/README.md b/examples/otel/README.md index 0215412..9e09174 100644 --- a/examples/otel/README.md +++ b/examples/otel/README.md @@ -12,7 +12,7 @@ OpenTelemetry SDK exporter without changing the cache integration. * Observing hits, misses, loader outcomes, and deletions * Handling not-found loader results without caching them * Flushing telemetry before a short-lived process exits -* Managing cache and OpenTelemetry lifecycles correctly +* Unregistering cache metrics before shutting down the OpenTelemetry SDK ## Run @@ -44,4 +44,4 @@ cache misses and two `not_found` loader outcomes because results returned with ` cache. Each subsequent lookup therefore invokes the repository again. Because this example is short-lived, it calls `ForceFlush` before exiting to flush pending telemetry. Long-running -applications normally rely on their configured OpenTelemetry metric reader to collect and export metrics continuously. +applications normally rely on their configured OpenTelemetry metric reader to collect and export metrics continuously. \ No newline at end of file diff --git a/examples/otel/main.go b/examples/otel/main.go index 34b1e19..7135892 100644 --- a/examples/otel/main.go +++ b/examples/otel/main.go @@ -51,6 +51,11 @@ func run(ctx context.Context) error { metrics := paceotel.New( paceotel.WithMeterProvider(meterProvider), ) + defer func() { + if err := metrics.Unregister(); err != nil { + log.Printf("unregister cache metrics: %v", err) + } + }() repository := &userRepository{ users: map[int64]user{ @@ -62,8 +67,8 @@ func run(ctx context.Context) error { } users, err := pacecache.NewWithDefaultLoader[int64, user]( - "users", repository.find, + pacecache.WithName("users"), pacecache.WithMaxEntries(128), pacecache.WithTTL(time.Minute), pacecache.WithMetrics(metrics), @@ -71,7 +76,6 @@ func run(ctx context.Context) error { if err != nil { return fmt.Errorf("create users cache: %w", err) } - defer users.Close() // 3. Generate representative cache activity. lookups := []struct { diff --git a/extra/paceotel/README.md b/extra/paceotel/README.md index b9ad83a..c299258 100644 --- a/extra/paceotel/README.md +++ b/extra/paceotel/README.md @@ -14,6 +14,7 @@ go get github.com/mkbeh/pacecache/extra/paceotel ## Usage + ```go import ( "context" @@ -30,14 +31,15 @@ defer provider.Shutdown(context.Background()) metrics := paceotel.New( paceotel.WithMeterProvider(provider), ) +defer metrics.Unregister() // Attach metrics when creating the cache. cache, _ := pacecache.New[string, string]( - "users", + pacecache.WithName("users"), pacecache.WithMetrics(metrics), ) -defer cache.Close() ``` + -For a complete runnable setup using the stdout exporter, see the [example](../../examples/otel). \ No newline at end of file +For a complete runnable setup using the stdout exporter, see the [example](../../examples/otel). diff --git a/extra/paceotel/cache.go b/extra/paceotel/cache.go index 62928ac..150c4f2 100644 --- a/extra/paceotel/cache.go +++ b/extra/paceotel/cache.go @@ -1,13 +1,10 @@ package paceotel import ( - "context" - "errors" "fmt" "slices" "github.com/mkbeh/pacecache" - "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" ) @@ -49,7 +46,7 @@ const ( removalOperationClear = "clear" ) -type cacheMetricInstruments struct { +type metricInstruments struct { entryCount metric.Int64ObservableGauge entryLimit metric.Int64ObservableGauge segmentCount metric.Int64ObservableGauge @@ -67,7 +64,7 @@ type cacheMetricInstruments struct { expirationCount metric.Int64ObservableCounter } -type cacheMetricAttributes struct { +type metricAttributes struct { base metric.ObserveOption hit metric.ObserveOption @@ -81,66 +78,10 @@ type cacheMetricAttributes struct { clearOperation metric.ObserveOption } -// RegisterCache registers OpenTelemetry metrics for one cache. -func (metrics *Metrics) RegisterCache( - cache pacecache.StatsProvider, -) (pacecache.MetricsRegistration, error) { - if metrics == nil { - return nil, errors.New("paceotel: metrics is nil") - } - - if cache == nil { - return nil, errors.New("paceotel: cache is nil") - } - - name := cache.Name() - if name == "" { - return nil, errors.New("paceotel: cache name is empty") - } - - meterProvider := metrics.meterProvider - if meterProvider == nil { - meterProvider = otel.GetMeterProvider() - } - - return registerCacheMetrics(cache, name, meterProvider) -} - -func registerCacheMetrics( - cache pacecache.StatsProvider, - name string, - provider metric.MeterProvider, -) (pacecache.MetricsRegistration, error) { - meter := provider.Meter(instrumentationName) - - instruments, err := newCacheMetricInstruments(meter) - if err != nil { - return nil, err - } - - attributes := newCacheMetricAttributes(name) - - registration, err := meter.RegisterCallback( - func(_ context.Context, observer metric.Observer) error { - instruments.observe(observer, cache.Stats(), attributes) - - return nil - }, - instruments.observables()..., - ) - if err != nil { - return nil, fmt.Errorf("paceotel: register metrics callback: %w", err) - } - - return &metricsRegistration{ - registration: registration, - }, nil -} - -func (instruments cacheMetricInstruments) observe( +func (instruments metricInstruments) observe( observer metric.Observer, stats pacecache.Stats, - attributes cacheMetricAttributes, + attributes metricAttributes, ) { observer.ObserveInt64( instruments.entryCount, @@ -257,7 +198,7 @@ func (instruments cacheMetricInstruments) observe( ) } -func (instruments cacheMetricInstruments) observables() []metric.Observable { +func (instruments metricInstruments) observables() []metric.Observable { return []metric.Observable{ instruments.entryCount, instruments.entryLimit, @@ -277,10 +218,8 @@ func (instruments cacheMetricInstruments) observables() []metric.Observable { } } -func newCacheMetricInstruments( - meter metric.Meter, -) (cacheMetricInstruments, error) { - var instruments cacheMetricInstruments +func newMetricInstruments(meter metric.Meter) (metricInstruments, error) { + var instruments metricInstruments var err error instruments.entryCount, err = meter.Int64ObservableGauge( @@ -291,7 +230,7 @@ func newCacheMetricInstruments( metric.WithUnit("{entry}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(entryCountMetricName, err) } @@ -303,7 +242,7 @@ func newCacheMetricInstruments( metric.WithUnit("{entry}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(entryLimitMetricName, err) } @@ -315,7 +254,7 @@ func newCacheMetricInstruments( metric.WithUnit("{segment}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(segmentCountMetricName, err) } @@ -327,7 +266,7 @@ func newCacheMetricInstruments( metric.WithUnit("{lookup}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(lookupCountMetricName, err) } @@ -339,7 +278,7 @@ func newCacheMetricInstruments( metric.WithUnit("{load}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(loadCountMetricName, err) } @@ -351,7 +290,7 @@ func newCacheMetricInstruments( metric.WithUnit("s"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(loadTimeMetricName, err) } @@ -363,7 +302,7 @@ func newCacheMetricInstruments( metric.WithUnit("{request}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(loadSharedCountMetricName, err) } @@ -375,7 +314,7 @@ func newCacheMetricInstruments( metric.WithUnit("{load}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(loadSupersededCountMetricName, err) } @@ -387,7 +326,7 @@ func newCacheMetricInstruments( metric.WithUnit("{entry}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(removedCountMetricName, err) } @@ -399,7 +338,7 @@ func newCacheMetricInstruments( metric.WithUnit("{cleanup}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(cleanupCountMetricName, err) } @@ -411,7 +350,7 @@ func newCacheMetricInstruments( metric.WithUnit("{run}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(cleanupWorkerRunCountMetricName, err) } @@ -423,7 +362,7 @@ func newCacheMetricInstruments( metric.WithUnit("{run}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(cleanupWorkerPendingCountMetricName, err) } @@ -435,7 +374,7 @@ func newCacheMetricInstruments( metric.WithUnit("s"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(cleanupWorkerTimeMetricName, err) } @@ -447,7 +386,7 @@ func newCacheMetricInstruments( metric.WithUnit("{entry}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(evictionCountMetricName, err) } @@ -459,19 +398,21 @@ func newCacheMetricInstruments( metric.WithUnit("{entry}"), ) if err != nil { - return cacheMetricInstruments{}, + return metricInstruments{}, newMetricError(expirationCountMetricName, err) } return instruments, nil } -func newCacheMetricAttributes(name string) cacheMetricAttributes { - base := []attribute.KeyValue{ - attribute.String( - cacheNameAttribute, - name, - ), +func newMetricAttributes(name string) metricAttributes { + var base []attribute.KeyValue + + if name != "" { + base = append( + base, + attribute.String(cacheNameAttribute, name), + ) } option := func(extra ...attribute.KeyValue) metric.ObserveOption { @@ -481,7 +422,7 @@ func newCacheMetricAttributes(name string) cacheMetricAttributes { ) } - return cacheMetricAttributes{ + return metricAttributes{ base: option(), hit: option( diff --git a/extra/paceotel/cache_test.go b/extra/paceotel/cache_test.go index 73f19f5..f594fcd 100644 --- a/extra/paceotel/cache_test.go +++ b/extra/paceotel/cache_test.go @@ -3,6 +3,7 @@ package paceotel import ( "context" "errors" + "sync/atomic" "testing" "time" @@ -12,77 +13,72 @@ import ( "go.opentelemetry.io/otel/sdk/metric/metricdata" ) -type statsProviderStub struct { - name string - stats pacecache.Stats -} - -func (provider *statsProviderStub) Name() string { - if provider == nil { - return "" - } +const ( + wantScopeName = "github.com/mkbeh/pacecache/extra/paceotel" + + wantEntryCountMetricName = "pacecache.entry.count" + wantEntryLimitMetricName = "pacecache.entry.limit" + wantSegmentCountMetricName = "pacecache.segment.count" + wantLookupCountMetricName = "pacecache.lookup.count" + wantLoadCountMetricName = "pacecache.load.count" + wantLoadTimeMetricName = "pacecache.load.time" + wantLoadSharedCountMetricName = "pacecache.load.shared.count" + wantLoadSupersededCountMetricName = "pacecache.load.superseded.count" + wantRemovedCountMetricName = "pacecache.entry.removed.count" + wantCleanupCountMetricName = "pacecache.cleanup.count" + wantCleanupWorkerRunCountMetricName = "pacecache.cleanup.worker.run.count" + wantCleanupWorkerPendingCountMetricName = "pacecache.cleanup.worker.pending.count" + wantCleanupWorkerTimeMetricName = "pacecache.cleanup.worker.time" + wantEvictionCountMetricName = "pacecache.entry.eviction.count" + wantExpirationCountMetricName = "pacecache.entry.expiration.count" + + wantCacheNameAttribute = "pacecache.name" + wantLookupResultAttribute = "pacecache.lookup.result" + wantLoadResultAttribute = "pacecache.load.result" + wantRemovalOperationAttribute = "pacecache.removal.operation" +) - return provider.name -} +func TestMetricsSchema(t *testing.T) { + metrics, reader := newTestMetrics(t) -func (provider *statsProviderStub) Stats() pacecache.Stats { - if provider == nil { - return pacecache.Stats{} + if err := metrics.Register(&metricsSourceStub{name: "users"}); err != nil { + t.Fatalf("Register() error = %v", err) } - return provider.stats -} - -func TestRegisterCacheValidation(t *testing.T) { - provider := &statsProviderStub{name: "users"} - - t.Run("nil metrics", func(t *testing.T) { - var metrics *Metrics - - registration, err := metrics.RegisterCache(provider) - if registration != nil { - t.Fatalf("RegisterCache() registration = %v, want nil", registration) - } - if err == nil || err.Error() != "paceotel: metrics is nil" { - t.Fatalf("RegisterCache() error = %v, want metrics is nil", err) - } - }) - - t.Run("nil cache", func(t *testing.T) { - metrics := New() - - registration, err := metrics.RegisterCache(nil) - if registration != nil { - t.Fatalf("RegisterCache() registration = %v, want nil", registration) - } - if err == nil || err.Error() != "paceotel: cache is nil" { - t.Fatalf("RegisterCache() error = %v, want cache is nil", err) - } - }) + collected := collectMetricsByName(t, collectTestMetrics(t, reader)) + + expected := map[string]string{ + wantEntryCountMetricName: "{entry}", + wantEntryLimitMetricName: "{entry}", + wantSegmentCountMetricName: "{segment}", + wantLookupCountMetricName: "{lookup}", + wantLoadCountMetricName: "{load}", + wantLoadTimeMetricName: "s", + wantLoadSharedCountMetricName: "{request}", + wantLoadSupersededCountMetricName: "{load}", + wantRemovedCountMetricName: "{entry}", + wantCleanupCountMetricName: "{cleanup}", + wantCleanupWorkerRunCountMetricName: "{run}", + wantCleanupWorkerPendingCountMetricName: "{run}", + wantCleanupWorkerTimeMetricName: "s", + wantEvictionCountMetricName: "{entry}", + wantExpirationCountMetricName: "{entry}", + } - t.Run("blank cache name", func(t *testing.T) { - metrics := New() + if len(collected) != len(expected) { + t.Fatalf("collected metrics = %d, want %d", len(collected), len(expected)) + } - registration, err := metrics.RegisterCache(&statsProviderStub{}) - if registration != nil { - t.Fatalf("RegisterCache() registration = %v, want nil", registration) - } - if err == nil || err.Error() != "paceotel: cache name is empty" { - t.Fatalf("RegisterCache() error = %v, want cache name is empty", err) + for name, unit := range expected { + current := requireMetric(t, collected, name) + if current.Unit != unit { + t.Fatalf("metric %q unit = %q, want %q", name, current.Unit, unit) } - }) + } } -func TestRegisterCacheCollectsMetrics(t *testing.T) { - reader := sdkmetric.NewManualReader() - meterProvider := sdkmetric.NewMeterProvider( - sdkmetric.WithReader(reader), - ) - t.Cleanup(func() { - if err := meterProvider.Shutdown(context.Background()); err != nil { - t.Fatalf("Shutdown() error = %v", err) - } - }) +func TestMetricsCollectsSource(t *testing.T) { + metrics, reader := newTestMetrics(t) stats := pacecache.Stats{ EntryCount: 3, @@ -111,204 +107,326 @@ func TestRegisterCacheCollectsMetrics(t *testing.T) { ExpirationCount: 14, } - metrics := New( - WithMeterProvider(meterProvider), - ) + var statsCalls atomic.Int64 - registration, err := metrics.RegisterCache( - &statsProviderStub{ - name: "users", - stats: stats, - }, - ) - if err != nil { - t.Fatalf("RegisterCache() error = %v", err) - } - if registration == nil { - t.Fatal("RegisterCache() registration = nil") - } - t.Cleanup(registration.Close) + if err := metrics.Register( + &metricsSourceStub{ + name: "users", + statsFn: func() pacecache.Stats { + statsCalls.Add(1) - var resourceMetrics metricdata.ResourceMetrics - if err := reader.Collect( - context.Background(), - &resourceMetrics, + return stats + }, + }, ); err != nil { - t.Fatalf("Collect() error = %v", err) + t.Fatalf("Register() error = %v", err) } - collected := collectMetricsByName(t, resourceMetrics) + collected := collectMetricsByName(t, collectTestMetrics(t, reader)) + + if got := statsCalls.Load(); got != 1 { + t.Fatalf("Stats() calls = %d, want 1", got) + } - requireInt64MetricPoint( + requireInt64GaugePoint( t, collected, - entryCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantEntryCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 3, ) - requireInt64MetricPoint( + requireInt64GaugePoint( t, collected, - entryLimitMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantEntryLimitMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 128, ) - requireInt64MetricPoint( + requireInt64GaugePoint( t, collected, - segmentCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantSegmentCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 4, ) - requireInt64MetricPoint( + lookupPoints := int64CounterPoints( + t, + requireMetric(t, collected, wantLookupCountMetricName), + ) + if len(lookupPoints) != 2 { + t.Fatalf("lookup count points = %d, want 2", len(lookupPoints)) + } + requireInt64CounterPoint( t, collected, - lookupCountMetricName, + wantLookupCountMetricName, map[string]string{ - cacheNameAttribute: "users", - lookupResultAttribute: lookupResultHit, + wantCacheNameAttribute: "users", + wantLookupResultAttribute: "hit", }, 11, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - lookupCountMetricName, + wantLookupCountMetricName, map[string]string{ - cacheNameAttribute: "users", - lookupResultAttribute: lookupResultMiss, + wantCacheNameAttribute: "users", + wantLookupResultAttribute: "miss", }, 7, ) - requireNoInt64MetricPoint( + + loadPoints := int64CounterPoints( t, - collected, - lookupCountMetricName, - map[string]string{ - cacheNameAttribute: "users", - lookupResultAttribute: "negative_hit", - }, + requireMetric(t, collected, wantLoadCountMetricName), ) - - requireInt64MetricPoint( + if len(loadPoints) != 3 { + t.Fatalf("load count points = %d, want 3", len(loadPoints)) + } + requireInt64CounterPoint( t, collected, - loadCountMetricName, + wantLoadCountMetricName, map[string]string{ - cacheNameAttribute: "users", - loadResultAttribute: loadResultFound, + wantCacheNameAttribute: "users", + wantLoadResultAttribute: "found", }, 5, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - loadCountMetricName, + wantLoadCountMetricName, map[string]string{ - cacheNameAttribute: "users", - loadResultAttribute: loadResultNotFound, + wantCacheNameAttribute: "users", + wantLoadResultAttribute: "not_found", }, 3, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - loadCountMetricName, + wantLoadCountMetricName, map[string]string{ - cacheNameAttribute: "users", - loadResultAttribute: loadResultError, + wantCacheNameAttribute: "users", + wantLoadResultAttribute: "error", }, 2, ) - requireFloat64MetricPoint( + requireFloat64CounterPoint( t, collected, - loadTimeMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantLoadTimeMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 1.5, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - loadSharedCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantLoadSharedCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 4, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - loadSupersededCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantLoadSupersededCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 1, ) - requireInt64MetricPoint( + removedPoints := int64CounterPoints( + t, + requireMetric(t, collected, wantRemovedCountMetricName), + ) + if len(removedPoints) != 2 { + t.Fatalf("removed count points = %d, want 2", len(removedPoints)) + } + requireInt64CounterPoint( t, collected, - removedCountMetricName, + wantRemovedCountMetricName, map[string]string{ - cacheNameAttribute: "users", - removalOperationAttribute: removalOperationDelete, + wantCacheNameAttribute: "users", + wantRemovalOperationAttribute: "delete", }, 8, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - removedCountMetricName, + wantRemovedCountMetricName, map[string]string{ - cacheNameAttribute: "users", - removalOperationAttribute: removalOperationClear, + wantCacheNameAttribute: "users", + wantRemovalOperationAttribute: "clear", }, 9, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - cleanupCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantCleanupCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 10, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - cleanupWorkerRunCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantCleanupWorkerRunCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 11, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - cleanupWorkerPendingCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantCleanupWorkerPendingCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 12, ) - requireFloat64MetricPoint( + requireFloat64CounterPoint( t, collected, - cleanupWorkerTimeMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantCleanupWorkerTimeMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 2.5, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - evictionCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantEvictionCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 13, ) - requireInt64MetricPoint( + requireInt64CounterPoint( t, collected, - expirationCountMetricName, - map[string]string{cacheNameAttribute: "users"}, + wantExpirationCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, 14, ) } +func TestMetricsCollectsMultipleSources(t *testing.T) { + metrics, reader := newTestMetrics(t) + + if err := metrics.Register( + &metricsSourceStub{ + name: "users", + stats: pacecache.Stats{ + EntryCount: 3, + MaxEntries: 128, + HitCount: 11, + }, + }, + ); err != nil { + t.Fatalf("Register(users) error = %v", err) + } + + if err := metrics.Register( + &metricsSourceStub{ + name: "sessions", + stats: pacecache.Stats{ + EntryCount: 7, + MaxEntries: 64, + MissCount: 5, + }, + }, + ); err != nil { + t.Fatalf("Register(sessions) error = %v", err) + } + + collected := collectMetricsByName(t, collectTestMetrics(t, reader)) + + entryPoints := int64GaugePoints( + t, + requireMetric(t, collected, wantEntryCountMetricName), + ) + if len(entryPoints) != 2 { + t.Fatalf("entry count points = %d, want 2", len(entryPoints)) + } + + requireInt64GaugePoint( + t, + collected, + wantEntryCountMetricName, + map[string]string{wantCacheNameAttribute: "users"}, + 3, + ) + requireInt64GaugePoint( + t, + collected, + wantEntryCountMetricName, + map[string]string{wantCacheNameAttribute: "sessions"}, + 7, + ) + requireInt64GaugePoint( + t, + collected, + wantEntryLimitMetricName, + map[string]string{wantCacheNameAttribute: "users"}, + 128, + ) + requireInt64GaugePoint( + t, + collected, + wantEntryLimitMetricName, + map[string]string{wantCacheNameAttribute: "sessions"}, + 64, + ) + requireInt64CounterPoint( + t, + collected, + wantLookupCountMetricName, + map[string]string{ + wantCacheNameAttribute: "users", + wantLookupResultAttribute: "hit", + }, + 11, + ) + requireInt64CounterPoint( + t, + collected, + wantLookupCountMetricName, + map[string]string{ + wantCacheNameAttribute: "sessions", + wantLookupResultAttribute: "miss", + }, + 5, + ) +} + +func TestMetricsCollectsUnnamedSource(t *testing.T) { + metrics, reader := newTestMetrics(t) + + if err := metrics.Register( + &metricsSourceStub{ + stats: pacecache.Stats{EntryCount: 3}, + }, + ); err != nil { + t.Fatalf("Register() error = %v", err) + } + + collected := collectMetricsByName(t, collectTestMetrics(t, reader)) + points := int64GaugePoints( + t, + requireMetric(t, collected, wantEntryCountMetricName), + ) + + if len(points) != 1 { + t.Fatalf("entry count points = %d, want 1", len(points)) + } + if points[0].Value != 3 { + t.Fatalf("entry count = %d, want 3", points[0].Value) + } + if got := points[0].Attributes.Len(); got != 0 { + t.Fatalf("unnamed source attributes = %d, want 0", got) + } +} + func TestNewMetricError(t *testing.T) { sentinel := errors.New("instrument failed") @@ -323,6 +441,20 @@ func TestNewMetricError(t *testing.T) { } } +func collectTestMetrics( + t *testing.T, + reader *sdkmetric.ManualReader, +) metricdata.ResourceMetrics { + t.Helper() + + var resourceMetrics metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &resourceMetrics); err != nil { + t.Fatalf("Collect() error = %v", err) + } + + return resourceMetrics +} + func collectMetricsByName( t *testing.T, resourceMetrics metricdata.ResourceMetrics, @@ -330,53 +462,63 @@ func collectMetricsByName( t.Helper() collected := make(map[string]metricdata.Metrics) + foundScope := false - var foundScope bool for _, scopeMetrics := range resourceMetrics.ScopeMetrics { - if scopeMetrics.Scope.Name != instrumentationName { + if scopeMetrics.Scope.Name != wantScopeName { continue } + if foundScope { + t.Fatalf("instrumentation scope %q collected more than once", wantScopeName) + } foundScope = true + for _, current := range scopeMetrics.Metrics { + if _, exists := collected[current.Name]; exists { + t.Fatalf("metric %q collected more than once", current.Name) + } + collected[current.Name] = current } } if !foundScope { - t.Fatalf( - "instrumentation scope %q not found", - instrumentationName, - ) + t.Fatalf("instrumentation scope %q not found", wantScopeName) } - expected := []string{ - entryCountMetricName, - entryLimitMetricName, - segmentCountMetricName, - lookupCountMetricName, - loadCountMetricName, - loadTimeMetricName, - loadSharedCountMetricName, - loadSupersededCountMetricName, - removedCountMetricName, - cleanupCountMetricName, - cleanupWorkerRunCountMetricName, - cleanupWorkerPendingCountMetricName, - cleanupWorkerTimeMetricName, - evictionCountMetricName, - expirationCountMetricName, - } - for _, name := range expected { - if _, ok := collected[name]; !ok { - t.Fatalf("metric %q not collected", name) - } + return collected +} + +func requireMetric( + t *testing.T, + collected map[string]metricdata.Metrics, + name string, +) metricdata.Metrics { + t.Helper() + + current, ok := collected[name] + if !ok { + t.Fatalf("metric %q not collected", name) } - return collected + return current +} + +func requireInt64GaugePoint( + t *testing.T, + collected map[string]metricdata.Metrics, + name string, + attributes map[string]string, + want int64, +) { + t.Helper() + + points := int64GaugePoints(t, requireMetric(t, collected, name)) + requireInt64Point(t, name, points, attributes, want) } -func requireInt64MetricPoint( +func requireInt64CounterPoint( t *testing.T, collected map[string]metricdata.Metrics, name string, @@ -385,7 +527,20 @@ func requireInt64MetricPoint( ) { t.Helper() - points := int64MetricPoints(t, collected[name]) + points := int64CounterPoints(t, requireMetric(t, collected, name)) + requireInt64Point(t, name, points, attributes, want) +} + +func requireFloat64CounterPoint( + t *testing.T, + collected map[string]metricdata.Metrics, + name string, + attributes map[string]string, + want float64, +) { + t.Helper() + + points := float64CounterPoints(t, requireMetric(t, collected, name)) for _, point := range points { if !metricAttributesMatch(point.Attributes, attributes) { continue @@ -393,7 +548,7 @@ func requireInt64MetricPoint( if point.Value != want { t.Fatalf( - "metric %q value = %d, want %d; attributes=%v", + "metric %q value = %v, want %v; attributes=%v", name, point.Value, want, @@ -411,36 +566,15 @@ func requireInt64MetricPoint( ) } -func requireNoInt64MetricPoint( +func requireInt64Point( t *testing.T, - collected map[string]metricdata.Metrics, - name string, - attributes map[string]string, -) { - t.Helper() - - points := int64MetricPoints(t, collected[name]) - for _, point := range points { - if metricAttributesMatch(point.Attributes, attributes) { - t.Fatalf( - "metric %q unexpectedly contains attributes %v", - name, - attributes, - ) - } - } -} - -func requireFloat64MetricPoint( - t *testing.T, - collected map[string]metricdata.Metrics, name string, + points []metricdata.DataPoint[int64], attributes map[string]string, - want float64, + want int64, ) { t.Helper() - points := float64MetricPoints(t, collected[name]) for _, point := range points { if !metricAttributesMatch(point.Attributes, attributes) { continue @@ -448,7 +582,7 @@ func requireFloat64MetricPoint( if point.Value != want { t.Fatalf( - "metric %q value = %v, want %v; attributes=%v", + "metric %q value = %d, want %d; attributes=%v", name, point.Value, want, @@ -466,52 +600,87 @@ func requireFloat64MetricPoint( ) } -func int64MetricPoints( +func int64GaugePoints( + t *testing.T, + current metricdata.Metrics, +) []metricdata.DataPoint[int64] { + t.Helper() + + data, ok := current.Data.(metricdata.Gauge[int64]) + if !ok { + t.Fatalf( + "metric %q data type = %T, want int64 gauge", + current.Name, + current.Data, + ) + } + + return data.DataPoints +} + +func int64CounterPoints( t *testing.T, current metricdata.Metrics, ) []metricdata.DataPoint[int64] { t.Helper() - switch data := current.Data.(type) { - case metricdata.Gauge[int64]: - return data.DataPoints - case metricdata.Sum[int64]: - return data.DataPoints - default: + data, ok := current.Data.(metricdata.Sum[int64]) + if !ok { t.Fatalf( - "metric %q data type = %T, want int64 gauge or sum", + "metric %q data type = %T, want int64 sum", current.Name, current.Data, ) - return nil } + if !data.IsMonotonic { + t.Fatalf("metric %q is not monotonic", current.Name) + } + + return data.DataPoints } -func float64MetricPoints( +func float64CounterPoints( t *testing.T, current metricdata.Metrics, ) []metricdata.DataPoint[float64] { t.Helper() - switch data := current.Data.(type) { - case metricdata.Gauge[float64]: - return data.DataPoints - case metricdata.Sum[float64]: - return data.DataPoints - default: + data, ok := current.Data.(metricdata.Sum[float64]) + if !ok { t.Fatalf( - "metric %q data type = %T, want float64 gauge or sum", + "metric %q data type = %T, want float64 sum", current.Name, current.Data, ) - return nil } + if !data.IsMonotonic { + t.Fatalf("metric %q is not monotonic", current.Name) + } + + return data.DataPoints +} + +func hasMetricAttributes( + points []metricdata.DataPoint[int64], + want map[string]string, +) bool { + for _, point := range points { + if metricAttributesMatch(point.Attributes, want) { + return true + } + } + + return false } func metricAttributesMatch( set attribute.Set, want map[string]string, ) bool { + if set.Len() != len(want) { + return false + } + for key, value := range want { current, ok := set.Value(attribute.Key(key)) if !ok || current.AsString() != value { diff --git a/extra/paceotel/metrics.go b/extra/paceotel/metrics.go index f2f63fd..61c231d 100644 --- a/extra/paceotel/metrics.go +++ b/extra/paceotel/metrics.go @@ -1,36 +1,167 @@ package paceotel import ( + "context" + "errors" "fmt" + "sync" "github.com/mkbeh/pacecache" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/metric" ) -const instrumentationName = "github.com/mkbeh/pacecache/extra/paceotel" - // Metrics exports pacecache statistics through OpenTelemetry. // -// Metrics is immutable and safe to reuse across multiple caches. +// Metrics is safe to reuse across multiple caches. Each registered source must +// have a unique name within a Metrics instance; only one unnamed source may be +// registered. Unregister removes the OpenTelemetry callbacks for all sources. type Metrics struct { - meterProvider metric.MeterProvider + meter metric.Meter + + mu sync.Mutex + instruments *metricInstruments + registrations map[string]metric.Registration + closed bool } var _ pacecache.Metrics = (*Metrics)(nil) -type metricsRegistration struct { - registration metric.Registration +// New creates an OpenTelemetry metrics implementation. +// +// By default, metrics use the global OpenTelemetry MeterProvider. The returned +// value may be reused across multiple caches. Call Unregister to remove its +// OpenTelemetry callbacks and release references to registered sources. +func New(options ...Option) *Metrics { + settings := settings{} + + for _, option := range options { + if option != nil { + option(&settings) + } + } + + meterProvider := settings.meterProvider + if meterProvider == nil { + meterProvider = otel.GetMeterProvider() + } + + return &Metrics{ + meter: meterProvider.Meter( + ScopeName, + metric.WithInstrumentationVersion(Version()), + ), + registrations: make(map[string]metric.Registration), + } +} + +// Register adds one cache metrics source to this OpenTelemetry integration. +// The source name must be unique within Metrics. An empty name is allowed, but +// only one unnamed source may be registered. +func (metrics *Metrics) Register(source pacecache.MetricsSource) error { + if metrics == nil { + return errors.New("paceotel: metrics is nil") + } + + if source == nil { + return errors.New("paceotel: source is nil") + } + + name := source.Name() + + metrics.mu.Lock() + defer metrics.mu.Unlock() + + if metrics.closed { + return errors.New("paceotel: metrics is unregistered") + } + + if _, exists := metrics.registrations[name]; exists { + return fmt.Errorf("paceotel: metrics source %q already registered", name) + } + + if metrics.instruments == nil { + instruments, err := newMetricInstruments(metrics.meter) + if err != nil { + return err + } + + metrics.instruments = &instruments + } + + registration, err := metrics.registerCallback( + source, + newMetricAttributes(name), + ) + if err != nil { + return err + } + + metrics.registrations[name] = registration + + return nil } -func (registration *metricsRegistration) Close() { - if registration == nil || registration.registration == nil { - return +// Unregister removes the OpenTelemetry callbacks associated with Metrics and +// releases references to all successfully unregistered sources. Repeated calls +// are safe and retry callbacks that previously failed to unregister. +func (metrics *Metrics) Unregister() error { + if metrics == nil { + return nil } - if err := registration.registration.Unregister(); err != nil { - otel.Handle( - fmt.Errorf("paceotel: unregister metrics: %w", err), - ) + metrics.mu.Lock() + defer metrics.mu.Unlock() + + metrics.closed = true + metrics.instruments = nil + + var unregisterErr error + + for name, registration := range metrics.registrations { + if err := registration.Unregister(); err != nil { + unregisterErr = errors.Join( + unregisterErr, + fmt.Errorf("source %q: %w", name, err), + ) + continue + } + + delete(metrics.registrations, name) } + + if len(metrics.registrations) == 0 { + metrics.registrations = nil + } + + if unregisterErr != nil { + return fmt.Errorf("paceotel: unregister metrics: %w", unregisterErr) + } + + return nil +} + +func (metrics *Metrics) registerCallback( + source pacecache.MetricsSource, + attributes metricAttributes, +) (metric.Registration, error) { + instruments := *metrics.instruments + + registration, err := metrics.meter.RegisterCallback( + func(_ context.Context, observer metric.Observer) error { + instruments.observe( + observer, + source.Stats(), + attributes, + ) + + return nil + }, + instruments.observables()..., + ) + if err != nil { + return nil, fmt.Errorf("paceotel: register metrics callback: %w", err) + } + + return registration, nil } diff --git a/extra/paceotel/metrics_test.go b/extra/paceotel/metrics_test.go index 0ae7ce0..303a977 100644 --- a/extra/paceotel/metrics_test.go +++ b/extra/paceotel/metrics_test.go @@ -1,14 +1,48 @@ package paceotel import ( + "context" "errors" + "fmt" + "strings" + "sync" "sync/atomic" "testing" + "time" - "go.opentelemetry.io/otel" + "github.com/mkbeh/pacecache" + "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/metric/embedded" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" ) +type metricsSourceStub struct { + name string + stats pacecache.Stats + statsFn func() pacecache.Stats +} + +func (source *metricsSourceStub) Name() string { + if source == nil { + return "" + } + + return source.name +} + +func (source *metricsSourceStub) Stats() pacecache.Stats { + if source == nil { + return pacecache.Stats{} + } + + if source.statsFn != nil { + return source.statsFn() + } + + return source.stats +} + type registrationStub struct { embedded.Registration @@ -22,64 +56,459 @@ func (registration *registrationStub) Unregister() error { return registration.err } -func TestMetricsRegistrationClose(t *testing.T) { - underlying := ®istrationStub{} - registration := &metricsRegistration{ - registration: underlying, +func TestNewIgnoresNilOption(t *testing.T) { + metrics := New(nil) + if metrics == nil { + t.Fatal("New(nil) = nil") } - registration.Close() + if err := metrics.Unregister(); err != nil { + t.Fatalf("Unregister() error = %v", err) + } +} + +func TestMetricsRegisterValidation(t *testing.T) { + t.Run("nil metrics", func(t *testing.T) { + var metrics *Metrics - if got := underlying.calls.Load(); got != 1 { - t.Fatalf("Unregister() calls = %d, want 1", got) + err := metrics.Register(&metricsSourceStub{name: "users"}) + if err == nil || err.Error() != "paceotel: metrics is nil" { + t.Fatalf("Register() error = %v, want metrics is nil", err) + } + }) + + t.Run("nil source", func(t *testing.T) { + metrics, _ := newTestMetrics(t) + + err := metrics.Register(nil) + if err == nil || err.Error() != "paceotel: source is nil" { + t.Fatalf("Register() error = %v, want source is nil", err) + } + }) +} + +func TestMetricsRegisterDuplicateName(t *testing.T) { + metrics, _ := newTestMetrics(t) + + if err := metrics.Register(&metricsSourceStub{name: "users"}); err != nil { + t.Fatalf("Register() error = %v", err) + } + + err := metrics.Register(&metricsSourceStub{name: "users"}) + const want = `paceotel: metrics source "users" already registered` + if err == nil || err.Error() != want { + t.Fatalf("Register() error = %v, want %q", err, want) + } +} + +func TestMetricsRegisterDuplicateUnnamedSource(t *testing.T) { + metrics, _ := newTestMetrics(t) + + if err := metrics.Register(&metricsSourceStub{}); err != nil { + t.Fatalf("Register() error = %v", err) + } + + err := metrics.Register(&metricsSourceStub{}) + const want = `paceotel: metrics source "" already registered` + if err == nil || err.Error() != want { + t.Fatalf("Register() error = %v, want %q", err, want) } } -func TestMetricsRegistrationCloseNilSafe(_ *testing.T) { - var registration *metricsRegistration - registration.Close() +func TestMetricsRegisterAfterUnregister(t *testing.T) { + metrics, _ := newTestMetrics(t) - registration = &metricsRegistration{} - registration.Close() + if err := metrics.Unregister(); err != nil { + t.Fatalf("Unregister() error = %v", err) + } + + err := metrics.Register(&metricsSourceStub{name: "users"}) + if err == nil || err.Error() != "paceotel: metrics is unregistered" { + t.Fatalf("Register() error = %v, want metrics is unregistered", err) + } } -func TestMetricsRegistrationCloseReportsUnregisterError(t *testing.T) { - sentinel := errors.New("unregister failed") - underlying := ®istrationStub{err: sentinel} - registration := &metricsRegistration{ - registration: underlying, +func TestMetricsUnregisterNilSafe(t *testing.T) { + var metrics *Metrics + + if err := metrics.Unregister(); err != nil { + t.Fatalf("Unregister() error = %v", err) } +} - previous := otel.GetErrorHandler() - t.Cleanup(func() { - otel.SetErrorHandler(previous) - }) +func TestMetricsUnregisterRepeated(t *testing.T) { + registration := ®istrationStub{} + metrics := &Metrics{ + registrations: map[string]metric.Registration{ + "users": registration, + }, + } + + if err := metrics.Unregister(); err != nil { + t.Fatalf("first Unregister() error = %v", err) + } + if err := metrics.Unregister(); err != nil { + t.Fatalf("second Unregister() error = %v", err) + } + + if got := registration.calls.Load(); got != 1 { + t.Fatalf("Registration.Unregister() calls = %d, want 1", got) + } +} + +func TestMetricsUnregisterRetriesFailures(t *testing.T) { + firstErr := errors.New("first unregister failed") + secondErr := errors.New("second unregister failed") + + firstFailed := ®istrationStub{err: firstErr} + secondFailed := ®istrationStub{err: secondErr} + successful := ®istrationStub{} + + metrics := &Metrics{ + registrations: map[string]metric.Registration{ + "first": firstFailed, + "second": secondFailed, + "successful": successful, + }, + } + + err := metrics.Unregister() + if !errors.Is(err, firstErr) { + t.Fatalf("Unregister() error = %v, want wrapped first error", err) + } + if !errors.Is(err, secondErr) { + t.Fatalf("Unregister() error = %v, want wrapped second error", err) + } + if !strings.Contains(err.Error(), `source "first"`) { + t.Fatalf("Unregister() error = %q, want first source name", err) + } + if !strings.Contains(err.Error(), `source "second"`) { + t.Fatalf("Unregister() error = %q, want second source name", err) + } + + if got := firstFailed.calls.Load(); got != 1 { + t.Fatalf("first failed Unregister() calls = %d, want 1", got) + } + if got := secondFailed.calls.Load(); got != 1 { + t.Fatalf("second failed Unregister() calls = %d, want 1", got) + } + if got := successful.calls.Load(); got != 1 { + t.Fatalf("successful Unregister() calls = %d, want 1", got) + } + + registerErr := metrics.Register(&metricsSourceStub{name: "new"}) + if registerErr == nil || registerErr.Error() != "paceotel: metrics is unregistered" { + t.Fatalf( + "Register() after failed Unregister() error = %v, want metrics is unregistered", + registerErr, + ) + } + + firstFailed.err = nil + secondFailed.err = nil + + if err := metrics.Unregister(); err != nil { + t.Fatalf("retry Unregister() error = %v", err) + } + + if got := firstFailed.calls.Load(); got != 2 { + t.Fatalf("first failed Unregister() calls after retry = %d, want 2", got) + } + if got := secondFailed.calls.Load(); got != 2 { + t.Fatalf("second failed Unregister() calls after retry = %d, want 2", got) + } + if got := successful.calls.Load(); got != 1 { + t.Fatalf("successful Unregister() calls after retry = %d, want 1", got) + } + + if err := metrics.Unregister(); err != nil { + t.Fatalf("third Unregister() error = %v", err) + } + if got := firstFailed.calls.Load(); got != 2 { + t.Fatalf("first failed Unregister() calls after third call = %d, want 2", got) + } + if got := secondFailed.calls.Load(); got != 2 { + t.Fatalf("second failed Unregister() calls after third call = %d, want 2", got) + } + if got := successful.calls.Load(); got != 1 { + t.Fatalf("successful Unregister() calls after third call = %d, want 1", got) + } +} + +func TestMetricsConcurrentRegister(t *testing.T) { + metrics, reader := newTestMetrics(t) + + const count = 16 - var handled atomic.Value - otel.SetErrorHandler( - otel.ErrorHandlerFunc( - func(err error) { - handled.Store(err) - }, - ), + start := make(chan struct{}) + errorsCh := make(chan error, count) + + var waitGroup sync.WaitGroup + for index := range count { + waitGroup.Add(1) + + go func() { + defer waitGroup.Done() + <-start + + errorsCh <- metrics.Register( + &metricsSourceStub{name: fmt.Sprintf("cache-%d", index)}, + ) + }() + } + + close(start) + waitGroup.Wait() + close(errorsCh) + + for err := range errorsCh { + if err != nil { + t.Fatalf("Register() error = %v", err) + } + } + + collected := collectMetricsByName(t, collectTestMetrics(t, reader)) + points := int64GaugePoints( + t, + requireMetric(t, collected, wantEntryCountMetricName), ) + if len(points) != count { + t.Fatalf("entry count points = %d, want %d", len(points), count) + } + + for index := range count { + attributes := map[string]string{ + wantCacheNameAttribute: fmt.Sprintf("cache-%d", index), + } + if !hasMetricAttributes(points, attributes) { + t.Fatalf("entry count point with attributes %v not found", attributes) + } + } +} + +func TestMetricsConcurrentDuplicateRegister(t *testing.T) { + metrics, _ := newTestMetrics(t) + + const count = 16 + + start := make(chan struct{}) + errorsCh := make(chan error, count) + + var waitGroup sync.WaitGroup + for range count { + waitGroup.Add(1) + + go func() { + defer waitGroup.Done() + <-start + + errorsCh <- metrics.Register(&metricsSourceStub{name: "users"}) + }() + } + + close(start) + waitGroup.Wait() + close(errorsCh) + + var successful int + var duplicates int + + for err := range errorsCh { + switch { + case err == nil: + successful++ + case err.Error() == `paceotel: metrics source "users" already registered`: + duplicates++ + default: + t.Fatalf("Register() error = %v", err) + } + } + + if successful != 1 { + t.Fatalf("successful registrations = %d, want 1", successful) + } + if duplicates != count-1 { + t.Fatalf("duplicate registrations = %d, want %d", duplicates, count-1) + } +} + +func TestMetricsConcurrentRegisterWithUnregister(t *testing.T) { + metrics, _ := newTestMetrics(t) + + start := make(chan struct{}) + registerDone := make(chan error, 1) + unregisterDone := make(chan error, 1) + + go func() { + <-start + registerDone <- metrics.Register(&metricsSourceStub{name: "users"}) + }() + + go func() { + <-start + unregisterDone <- metrics.Unregister() + }() + + close(start) + + select { + case registerErr := <-registerDone: + if registerErr != nil && registerErr.Error() != "paceotel: metrics is unregistered" { + t.Fatalf("Register() error = %v", registerErr) + } + case <-time.After(5 * time.Second): + t.Fatal("Register() did not complete") + } + + select { + case err := <-unregisterDone: + if err != nil { + t.Fatalf("Unregister() error = %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Unregister() did not complete") + } + + err := metrics.Register(&metricsSourceStub{name: "after"}) + if err == nil || err.Error() != "paceotel: metrics is unregistered" { + t.Fatalf("Register() after concurrent Unregister() error = %v, want metrics is unregistered", err) + } +} + +func TestMetricsConcurrentUnregister(t *testing.T) { + registration := ®istrationStub{} + metrics := &Metrics{ + registrations: map[string]metric.Registration{ + "users": registration, + }, + } + + const count = 16 - registration.Close() + start := make(chan struct{}) + errorsCh := make(chan error, count) - if got := underlying.calls.Load(); got != 1 { - t.Fatalf("Unregister() calls = %d, want 1", got) + var waitGroup sync.WaitGroup + for range count { + waitGroup.Add(1) + + go func() { + defer waitGroup.Done() + <-start + + errorsCh <- metrics.Unregister() + }() } - value := handled.Load() - if value == nil { - t.Fatal("OpenTelemetry error handler was not called") + close(start) + waitGroup.Wait() + close(errorsCh) + + for err := range errorsCh { + if err != nil { + t.Fatalf("Unregister() error = %v", err) + } } - err, ok := value.(error) - if !ok { - t.Fatalf("handled value = %T, want error", value) + if got := registration.calls.Load(); got != 1 { + t.Fatalf("Registration.Unregister() calls = %d, want 1", got) } - if !errors.Is(err, sentinel) { - t.Fatalf("handled error = %v, want wrapped sentinel", err) +} + +func TestMetricsCollectConcurrentWithUnregister(t *testing.T) { + metrics, reader := newTestMetrics(t) + + started := make(chan struct{}) + release := make(chan struct{}) + + var startOnce sync.Once + source := &metricsSourceStub{ + name: "users", + statsFn: func() pacecache.Stats { + startOnce.Do(func() { + close(started) + }) + <-release + + return pacecache.Stats{EntryCount: 1} + }, + } + + if err := metrics.Register(source); err != nil { + t.Fatalf("Register() error = %v", err) } + + collectDone := make(chan error, 1) + go func() { + var resourceMetrics metricdata.ResourceMetrics + collectDone <- reader.Collect(context.Background(), &resourceMetrics) + }() + + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("metric callback did not start") + } + + unregisterDone := make(chan error, 1) + go func() { + unregisterDone <- metrics.Unregister() + }() + + close(release) + + select { + case err := <-collectDone: + if err != nil { + t.Fatalf("Collect() error = %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Collect() did not complete") + } + + select { + case err := <-unregisterDone: + if err != nil { + t.Fatalf("Unregister() error = %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Unregister() did not complete") + } + + var resourceMetrics metricdata.ResourceMetrics + if err := reader.Collect(context.Background(), &resourceMetrics); err != nil { + t.Fatalf("Collect() after Unregister() error = %v", err) + } + + for _, scopeMetrics := range resourceMetrics.ScopeMetrics { + if scopeMetrics.Scope.Name == wantScopeName && len(scopeMetrics.Metrics) != 0 { + t.Fatalf( + "metrics after Unregister() = %d, want 0", + len(scopeMetrics.Metrics), + ) + } + } +} + +func newTestMetrics(t *testing.T) (*Metrics, *sdkmetric.ManualReader) { + t.Helper() + + reader := sdkmetric.NewManualReader() + meterProvider := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + ) + metrics := New( + WithMeterProvider(meterProvider), + ) + + t.Cleanup(func() { + if err := metrics.Unregister(); err != nil { + t.Errorf("Unregister() error = %v", err) + } + if err := meterProvider.Shutdown(context.Background()); err != nil { + t.Errorf("Shutdown() error = %v", err) + } + }) + + return metrics, reader } diff --git a/extra/paceotel/options.go b/extra/paceotel/options.go index 5aac26c..6f40f24 100644 --- a/extra/paceotel/options.go +++ b/extra/paceotel/options.go @@ -9,24 +9,6 @@ type settings struct { meterProvider metric.MeterProvider } -// New creates an OpenTelemetry metrics implementation. -// -// By default, metrics use the global OpenTelemetry MeterProvider. The returned -// value is immutable and may be reused across multiple caches. -func New(options ...Option) *Metrics { - settings := settings{} - - for _, option := range options { - if option != nil { - option(&settings) - } - } - - return &Metrics{ - meterProvider: settings.meterProvider, - } -} - // WithMeterProvider configures the MeterProvider used for cache metrics. // // The caller owns the provider and is responsible for its lifecycle. diff --git a/extra/paceotel/options_test.go b/extra/paceotel/options_test.go index b636534..f90a4bc 100644 --- a/extra/paceotel/options_test.go +++ b/extra/paceotel/options_test.go @@ -7,48 +7,33 @@ import ( sdkmetric "go.opentelemetry.io/otel/sdk/metric" ) -func TestNew(t *testing.T) { - metrics := New(nil) - if metrics == nil { - t.Fatal("New() = nil") - } - if metrics.meterProvider != nil { - t.Fatalf( - "New() meterProvider = %T, want nil", - metrics.meterProvider, - ) - } -} - func TestWithMeterProvider(t *testing.T) { provider := sdkmetric.NewMeterProvider() t.Cleanup(func() { if err := provider.Shutdown(context.Background()); err != nil { - t.Fatalf("Shutdown() error = %v", err) + t.Errorf("Shutdown() error = %v", err) } }) - metrics := New( - WithMeterProvider(provider), - ) + var settings settings + WithMeterProvider(provider)(&settings) - if metrics.meterProvider != provider { + if settings.meterProvider != provider { t.Fatalf( - "New() meterProvider = %T, want provided MeterProvider", - metrics.meterProvider, + "WithMeterProvider() provider = %T, want provided MeterProvider", + settings.meterProvider, ) } } func TestWithMeterProviderNil(t *testing.T) { - metrics := New( - WithMeterProvider(nil), - ) + var settings settings + WithMeterProvider(nil)(&settings) - if metrics.meterProvider != nil { + if settings.meterProvider != nil { t.Fatalf( - "New() meterProvider = %T, want nil", - metrics.meterProvider, + "WithMeterProvider(nil) provider = %T, want nil", + settings.meterProvider, ) } } diff --git a/extra/paceotel/version.go b/extra/paceotel/version.go new file mode 100644 index 0000000..88cd554 --- /dev/null +++ b/extra/paceotel/version.go @@ -0,0 +1,31 @@ +package paceotel + +import "runtime/debug" + +// ScopeName is the OpenTelemetry instrumentation scope name used by this +// package. +const ScopeName = "github.com/mkbeh/pacecache/extra/paceotel" + +// Version returns the version of the paceotel module from Go build +// information. +// +// It returns "unknown" when build information is unavailable or does not +// contain this module. +func Version() string { + info, ok := debug.ReadBuildInfo() + if !ok { + return "unknown" + } + + if info.Main.Path == ScopeName { + return info.Main.Version + } + + for _, dep := range info.Deps { + if dep.Path == ScopeName { + return dep.Version + } + } + + return "unknown" +} diff --git a/load_test.go b/load_test.go index e93138a..1906b6a 100644 --- a/load_test.go +++ b/load_test.go @@ -10,7 +10,7 @@ import ( ) func TestNewWithDefaultLoaderRejectsNilLoader(t *testing.T) { - cache, err := NewWithDefaultLoader[string, int]("users", nil) + cache, err := NewWithDefaultLoader[string, int](nil) if cache != nil { t.Fatal("cache must be nil for nil default loader") } @@ -24,7 +24,6 @@ func TestGetOrLoadUsesDefaultLoader(t *testing.T) { var loadedKey string cache, err := NewWithDefaultLoader[string, int]( - "users", func(_ context.Context, key string) (int, bool, error) { calls.Add(1) loadedKey = key @@ -35,7 +34,6 @@ func TestGetOrLoadUsesDefaultLoader(t *testing.T) { if err != nil { t.Fatalf("NewWithDefaultLoader() error = %v", err) } - t.Cleanup(cache.Close) for range 2 { value, found, err := cache.GetOrLoad(context.Background(), "key") @@ -53,7 +51,7 @@ func TestGetOrLoadUsesDefaultLoader(t *testing.T) { } func TestGetOrLoadWithoutDefaultLoaderReturnsErrNoLoaderOnMiss(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) cache.Set("cached", 7, NoExpiration) value, found, err := cache.GetOrLoad(context.Background(), "cached") @@ -72,7 +70,6 @@ func TestGetOrLoadWithOverridesDefaultLoader(t *testing.T) { var overrideCalls atomic.Int64 cache, err := NewWithDefaultLoader[string, int]( - "users", func(context.Context, string) (int, bool, error) { defaultCalls.Add(1) @@ -82,7 +79,6 @@ func TestGetOrLoadWithOverridesDefaultLoader(t *testing.T) { if err != nil { t.Fatalf("NewWithDefaultLoader() error = %v", err) } - t.Cleanup(cache.Close) value, found, err := cache.GetOrLoadFunc( context.Background(), @@ -108,7 +104,6 @@ func TestGetOrLoadWithOverridesDefaultLoader(t *testing.T) { func TestGetOrLoadEntryUsesDefaultLoader(t *testing.T) { cache, err := NewWithDefaultLoader[string, int]( - "users", func(context.Context, string) (int, bool, error) { return 42, true, nil }, @@ -117,7 +112,6 @@ func TestGetOrLoadEntryUsesDefaultLoader(t *testing.T) { if err != nil { t.Fatalf("NewWithDefaultLoader() error = %v", err) } - t.Cleanup(cache.Close) entry, found, err := cache.GetOrLoadEntry(context.Background(), "key") if err != nil || !found || entry.Value() != 42 || entry.ExpiresAt().IsZero() { @@ -132,7 +126,6 @@ func TestDefaultAndExplicitLoadersShareOneWave(t *testing.T) { var overrideCalls atomic.Int64 cache, err := NewWithDefaultLoader[string, int]( - "users", func(context.Context, string) (int, bool, error) { defaultCalls.Add(1) close(started) @@ -146,7 +139,6 @@ func TestDefaultAndExplicitLoadersShareOneWave(t *testing.T) { if err != nil { t.Fatalf("NewWithDefaultLoader() error = %v", err) } - t.Cleanup(cache.Close) ownerDone := make(chan error, 1) go func() { @@ -194,7 +186,7 @@ func TestDefaultAndExplicitLoadersShareOneWave(t *testing.T) { } func TestGetOrLoadCachesPositiveResult(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) var calls atomic.Int64 loader := func(context.Context, string) (int, bool, error) { @@ -218,7 +210,7 @@ func TestGetOrLoadCachesPositiveResult(t *testing.T) { } func TestGetOrLoadNotFoundIsNotCached(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) var calls atomic.Int64 loader := func(context.Context, string) (int, bool, error) { @@ -248,7 +240,6 @@ func TestGetOrLoadNotFoundIsNotCached(t *testing.T) { func TestGetOrLoadEntryPositiveAndNotFound(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(time.Minute), @@ -277,7 +268,7 @@ func TestGetOrLoadEntryPositiveAndNotFound(t *testing.T) { } func TestGetOrLoadErrorsAreNotCached(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) sentinel := errors.New("load failed") var calls atomic.Int64 @@ -302,7 +293,7 @@ func TestGetOrLoadErrorsAreNotCached(t *testing.T) { } func TestGetOrLoadLoaderPanicPropagatesToCallerAndDoesNotPoisonKey(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) wantErr := errors.New("boom") var recovered any @@ -344,7 +335,7 @@ func TestGetOrLoadLoaderPanicPropagatesToCallerAndDoesNotPoisonKey(t *testing.T) } func TestGetOrLoadValidatesContextAndLoader(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) if _, _, err := cache.GetOrLoadFunc(nil, "key", func(context.Context, string) (int, bool, error) { return 1, true, nil @@ -364,7 +355,7 @@ func TestGetOrLoadValidatesContextAndLoader(t *testing.T) { } func TestGetOrLoadCanceledMissDoesNotStartLoader(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) ctx, cancel := context.WithCancel(context.Background()) cancel() @@ -382,7 +373,7 @@ func TestGetOrLoadCanceledMissDoesNotStartLoader(t *testing.T) { } func TestGetOrLoadCachedHitIgnoresCanceledContext(t *testing.T) { - cache := mustNewCache[int](t, "users") + cache := mustNewCache[int](t) cache.Set("key", 7, NoExpiration) ctx, cancel := context.WithCancel(context.Background()) @@ -398,7 +389,7 @@ func TestGetOrLoadCachedHitIgnoresCanceledContext(t *testing.T) { } func TestGetOrLoadCoalescesConcurrentMisses(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(32), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(32), WithSegmentCount(1)) started := make(chan struct{}) release := make(chan struct{}) @@ -474,7 +465,7 @@ func TestGetOrLoadCoalescesConcurrentMisses(t *testing.T) { } func TestGetOrLoadWaiterCanCancelWithoutCancelingSharedLoad(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) started := make(chan struct{}) release := make(chan struct{}) @@ -524,7 +515,6 @@ func TestGetOrLoadWaiterCanCancelWithoutCancelingSharedLoad(t *testing.T) { func TestGetOrLoadAndGetOrLoadEntryShareOneWave(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(time.Minute), @@ -607,11 +597,10 @@ func TestGetOrLoadAndGetOrLoadEntryShareOneWave(t *testing.T) { } func TestGetOrLoadUsesGenericKeyIdentity(t *testing.T) { - cache, err := New[testCompositeKey, int]("users", WithMaxEntries(8), WithSegmentCount(2)) + cache, err := New[testCompositeKey, int](WithMaxEntries(8), WithSegmentCount(2)) if err != nil { t.Fatalf("New() error = %v", err) } - t.Cleanup(cache.Close) key := testCompositeKey{TenantID: 7, UserID: 42} equal := testCompositeKey{TenantID: 7, UserID: 42} @@ -906,7 +895,6 @@ func newPublicationTestCache(t *testing.T) *Cache[string, string] { t.Helper() cache, err := New[string, string]( - "publication-test", WithMaxEntries(64), WithSegmentCount(1), ) @@ -914,6 +902,5 @@ func newPublicationTestCache(t *testing.T) *Cache[string, string] { t.Fatalf("New() error = %v", err) } - t.Cleanup(cache.Close) return cache } diff --git a/lookup.go b/lookup.go index 4426b5c..7642af0 100644 --- a/lookup.go +++ b/lookup.go @@ -45,7 +45,7 @@ func (cache *Cache[K, V]) RefreshTTL(key K) bool { // The returned bool reports whether a live entry exists. A live hit updates LRU // recency and contributes to lookup statistics. Expired entries are always // treated as misses and are removed when observed, by DeleteExpired, or by -// background cleanup when it is enabled. When sliding expiration is enabled, +// background cleanup when it is running. When sliding expiration is enabled, // a live hit refreshes the entry using the TTL with which it was stored. func (cache *Cache[K, V]) Get(key K) (V, bool) { var zero V diff --git a/lookup_test.go b/lookup_test.go index 8e35d23..0f5b323 100644 --- a/lookup_test.go +++ b/lookup_test.go @@ -8,7 +8,6 @@ import ( func TestCacheExists(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), ) @@ -32,7 +31,6 @@ func TestCacheExists(t *testing.T) { func TestCacheRefreshTTL(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(time.Minute), @@ -92,7 +90,7 @@ func TestCacheExistsAndRefreshTTLRemoveExpiredEntries(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(2), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(2), WithSegmentCount(1)) stats := cache.stats.segment(0) deadline := cache.store.now() @@ -125,7 +123,7 @@ func TestCacheExistsAndRefreshTTLRemoveExpiredEntries(t *testing.T) { } func TestCacheExistsAndRefreshTTLDoNotUpdateLRU(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(2), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(2), WithSegmentCount(1)) cache.Set("a", 1, time.Minute) cache.Set("b", 2, time.Minute) @@ -149,7 +147,7 @@ func TestCacheExistsAndRefreshTTLDoNotUpdateLRU(t *testing.T) { } func TestCacheSetGetAndLRU(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(2), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(2), WithSegmentCount(1)) if value, found := cache.Get("missing"); value != 0 || found { t.Fatalf("Get(missing) = (%d, %t), want (0, false)", value, found) @@ -182,7 +180,6 @@ func TestCacheSetGetAndLRU(t *testing.T) { func TestCacheGetEntry(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(4), WithSegmentCount(1), WithTTL(time.Minute), diff --git a/metrics.go b/metrics.go index bb96af3..0a26e27 100644 --- a/metrics.go +++ b/metrics.go @@ -1,38 +1,32 @@ package pacecache -// StatsProvider exposes one cache's identity and statistics to a metrics -// implementation. -type StatsProvider interface { +// MetricsSource exposes cache identity and statistics to a metrics implementation. +type MetricsSource interface { Name() string Stats() Stats } -// Metrics registers metrics for a Cache. +// Metrics registers cache statistics with a metrics implementation. // -// Implementations must be safe to reuse across multiple caches. RegisterCache -// may be called concurrently. +// Implementations must be safe to reuse across multiple caches. Register may +// be called concurrently. // -// If RegisterCache returns an error, the implementation must release any -// resources created during the registration attempt. +// If Register returns an error, the implementation must release any resources +// created during the registration attempt. type Metrics interface { - RegisterCache(cache StatsProvider) (MetricsRegistration, error) + Register(source MetricsSource) error } -// MetricsRegistration owns a metrics registration associated with one Cache. -// Close is called once when the Cache is closed. -type MetricsRegistration interface { - Close() -} - -// cacheStatsProvider exposes only the capabilities required by Metrics. -type cacheStatsProvider[K comparable, V any] struct { +// metricsSource exposes only the capabilities required by Metrics. +type metricsSource[K comparable, V any] struct { + name string cache *Cache[K, V] } -func (provider cacheStatsProvider[K, V]) Name() string { - return provider.cache.Name() +func (source metricsSource[K, V]) Name() string { + return source.name } -func (provider cacheStatsProvider[K, V]) Stats() Stats { - return provider.cache.Stats() +func (source metricsSource[K, V]) Stats() Stats { + return source.cache.Stats() } diff --git a/metrics_test.go b/metrics_test.go index 98c9eb4..26b5e08 100644 --- a/metrics_test.go +++ b/metrics_test.go @@ -2,98 +2,197 @@ package pacecache import ( "errors" - "sync/atomic" + "slices" + "sync" "testing" ) type testMetrics struct { - registerCalls atomic.Int64 - provider StatsProvider - registration MetricsRegistration + mu sync.Mutex + + registerCalls int + sources []MetricsSource err error - providerClose bool } -func (metrics *testMetrics) RegisterCache(provider StatsProvider) (MetricsRegistration, error) { - metrics.registerCalls.Add(1) - metrics.provider = provider - _, metrics.providerClose = provider.(interface{ Close() }) - return metrics.registration, metrics.err -} +func (metrics *testMetrics) Register(source MetricsSource) error { + metrics.mu.Lock() + defer metrics.mu.Unlock() + + metrics.registerCalls++ + + if metrics.err != nil { + return metrics.err + } + + metrics.sources = append(metrics.sources, source) -type testMetricsRegistration struct { - closeCalls atomic.Int64 + return nil } -func (registration *testMetricsRegistration) Close() { - registration.closeCalls.Add(1) +func (metrics *testMetrics) snapshot() (int, []MetricsSource) { + metrics.mu.Lock() + defer metrics.mu.Unlock() + + return metrics.registerCalls, slices.Clone(metrics.sources) } -func TestMetricsRegistrationLifecycle(t *testing.T) { - registration := &testMetricsRegistration{} - metrics := &testMetrics{registration: registration} +func TestMetricsRegistersSource(t *testing.T) { + metrics := &testMetrics{} - cache, err := New[string, int]("users", WithMetrics(metrics)) + cache, err := New[string, int]( + WithName("users"), + WithMetrics(metrics), + ) if err != nil { t.Fatalf("New() error = %v", err) } - if metrics.registerCalls.Load() != 1 { - t.Fatalf("RegisterCache calls = %d, want 1", metrics.registerCalls.Load()) + registerCalls, sources := metrics.snapshot() + if registerCalls != 1 { + t.Fatalf("Register() calls = %d, want 1", registerCalls) + } + if len(sources) != 1 { + t.Fatalf("registered sources = %d, want 1", len(sources)) } - if metrics.provider == nil { - t.Fatal("metrics provider is nil") + + source := sources[0] + if source.Name() != "users" { + t.Fatalf("source name = %q, want users", source.Name()) } - if metrics.provider.Name() != "users" { - t.Fatalf("provider name = %q, want users", metrics.provider.Name()) + if _, ok := source.(interface{ StartCleanup() }); ok { + t.Fatal("metrics source unexpectedly exposes Cache.StartCleanup") } - if metrics.providerClose { - t.Fatal("metrics provider unexpectedly exposes Cache.Close") + if _, ok := source.(interface{ StopCleanup() }); ok { + t.Fatal("metrics source unexpectedly exposes Cache.StopCleanup") } cache.Set("a", 1, NoExpiration) - if got := metrics.provider.Stats().EntryCount; got != 1 { - t.Fatalf("provider Stats().EntryCount = %d, want 1", got) + if got := source.Stats().EntryCount; got != 1 { + t.Fatalf("source Stats().EntryCount = %d, want 1", got) + } +} + +func TestMetricsRegistersMultipleCaches(t *testing.T) { + metrics := &testMetrics{} + + users, err := New[string, int]( + WithName("users"), + WithMetrics(metrics), + ) + if err != nil { + t.Fatalf("New(users) error = %v", err) + } + + sessions, err := New[string, int]( + WithName("sessions"), + WithMetrics(metrics), + ) + if err != nil { + t.Fatalf("New(sessions) error = %v", err) } - cache.Close() - cache.Close() - if registration.closeCalls.Load() != 1 { - t.Fatalf("registration Close calls = %d, want 1", registration.closeCalls.Load()) + users.Set("a", 1, NoExpiration) + sessions.Set("a", 1, NoExpiration) + sessions.Set("b", 2, NoExpiration) + + registerCalls, sources := metrics.snapshot() + if registerCalls != 2 { + t.Fatalf("Register() calls = %d, want 2", registerCalls) + } + if len(sources) != 2 { + t.Fatalf("registered sources = %d, want 2", len(sources)) + } + + byName := make(map[string]MetricsSource, len(sources)) + for _, source := range sources { + if _, ok := source.(interface{ StartCleanup() }); ok { + t.Fatalf( + "metrics source %q unexpectedly exposes Cache.StartCleanup", + source.Name(), + ) + } + if _, ok := source.(interface{ StopCleanup() }); ok { + t.Fatalf( + "metrics source %q unexpectedly exposes Cache.StopCleanup", + source.Name(), + ) + } + + if _, exists := byName[source.Name()]; exists { + t.Fatalf("metrics source %q registered more than once", source.Name()) + } + + byName[source.Name()] = source + } + + usersSource, ok := byName["users"] + if !ok { + t.Fatal("users metrics source not registered") + } + if got := usersSource.Stats().EntryCount; got != 1 { + t.Fatalf("users Stats().EntryCount = %d, want 1", got) + } + + sessionsSource, ok := byName["sessions"] + if !ok { + t.Fatal("sessions metrics source not registered") + } + if got := sessionsSource.Stats().EntryCount; got != 2 { + t.Fatalf("sessions Stats().EntryCount = %d, want 2", got) } } -func TestMetricsRegistrationError(t *testing.T) { +func TestMetricsRegistersUnnamedSource(t *testing.T) { + metrics := &testMetrics{} + + if _, err := New[string, int](WithMetrics(metrics)); err != nil { + t.Fatalf("New() error = %v", err) + } + + _, sources := metrics.snapshot() + if len(sources) != 1 { + t.Fatalf("registered sources = %d, want 1", len(sources)) + } + if sources[0].Name() != "" { + t.Fatalf("source name = %q, want empty", sources[0].Name()) + } +} + +func TestMetricsRegisterError(t *testing.T) { sentinel := errors.New("register failed") metrics := &testMetrics{err: sentinel} - cache, err := New[string, int]("users", WithMetrics(metrics)) + cache, err := New[string, int]( + WithName("users"), + WithMetrics(metrics), + ) if cache != nil { t.Fatal("cache must be nil when metrics registration fails") } if !errors.Is(err, sentinel) { t.Fatalf("error = %v, want wrapped sentinel", err) } - if metrics.registerCalls.Load() != 1 { - t.Fatalf("RegisterCache calls = %d, want 1", metrics.registerCalls.Load()) + + registerCalls, sources := metrics.snapshot() + if registerCalls != 1 { + t.Fatalf("Register() calls = %d, want 1", registerCalls) + } + if len(sources) != 0 { + t.Fatalf("registered sources = %d, want 0", len(sources)) } } -func TestMetricsMayReturnNilRegistration(t *testing.T) { - metrics := &testMetrics{} - cache, err := New[string, int]("users", WithMetrics(metrics)) +func TestMetricsNilIsNoop(t *testing.T) { + cache, err := New[string, int](WithMetrics(nil)) if err != nil { t.Fatalf("New() error = %v", err) } - cache.Close() -} -func TestRegisterMetricsNilIsNoop(t *testing.T) { - cache := &Cache[string, int]{} - if err := cache.registerMetrics(nil); err != nil { - t.Fatalf("registerMetrics(nil) error = %v", err) - } - if cache.metrics != nil { - t.Fatal("metrics registration must remain nil") + cache.Set("a", 1, NoExpiration) + + value, found := cache.Get("a") + if !found || value != 1 { + t.Fatalf("Get(a) = %d, %t, want 1, true", value, found) } } diff --git a/options.go b/options.go index 47b99ae..9c26a1e 100644 --- a/options.go +++ b/options.go @@ -14,9 +14,9 @@ const ( ) // Option configures a Cache created by New or NewWithDefaultLoader. -type Option func(*cacheSettings) error +type Option func(*settings) error -type cacheSettings struct { +type settings struct { name string maxEntries int @@ -33,9 +33,8 @@ type cacheSettings struct { metrics Metrics } -func newCacheSettings(name string, options ...Option) (*cacheSettings, error) { - settings := defaultCacheSettings() - settings.name = name +func newSettings(options ...Option) (*settings, error) { + settings := defaultSettings() for index, option := range options { if option == nil { @@ -54,16 +53,29 @@ func newCacheSettings(name string, options ...Option) (*cacheSettings, error) { return settings, nil } -func defaultCacheSettings() *cacheSettings { - return &cacheSettings{ +func defaultSettings() *settings { + return &settings{ maxEntries: defaultMaxEntries, segmentCount: defaultStorageSegmentCount, ttl: defaultTTL, + cleanupInterval: defaultCleanupInterval, cleanupBatchSize: defaultCleanupBatchSize, cleanupEntryBudget: defaultCleanupEntryBudget, } } +// WithName configures an optional logical cache name. +// +// Metrics implementations may use the name to distinguish cache instances. +// An empty name leaves the cache unnamed. +func WithName(name string) Option { + return func(settings *settings) error { + settings.name = name + + return nil + } +} + // WithMaxEntries configures the total cache entry budget. // // With one segment, the full budget is shared by the cache. When multiple @@ -71,7 +83,7 @@ func defaultCacheSettings() *cacheSettings { // capacity utilization may be slightly lower because each segment enforces its // own local budget. func WithMaxEntries(maxEntries int) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if maxEntries <= 0 { return errors.New("max entries must be positive") } @@ -89,7 +101,7 @@ func WithMaxEntries(maxEntries int) Option { // segment has its own entry budget. Benchmark segment counts against the // application's actual workload. func WithSegmentCount(count int) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if count <= 0 { return errors.New("segment count must be positive") } @@ -105,7 +117,7 @@ func WithSegmentCount(count int) Option { // A positive TTL enables time-based expiration. NoExpiration disables // time-based expiration for entries using the default expiration. func WithTTL(ttl time.Duration) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if ttl <= 0 && ttl != NoExpiration { return errors.New("ttl must be positive or NoExpiration") } @@ -123,7 +135,7 @@ func WithTTL(ttl time.Duration) Option { // expiration, the resulting effective TTL is reused on every refresh instead of // selecting another jitter value. Zero disables jitter. func WithJitter(jitter time.Duration) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if jitter < 0 { return errors.New("jitter must not be negative") } @@ -143,28 +155,23 @@ func WithJitter(jitter time.Duration) Option { // jitter is selected once when the entry is stored and reused by subsequent // refreshes. Entries using NoExpiration are not refreshed. func WithSlidingExpiration() Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { settings.slidingExpiration = true return nil } } -// WithCleanupInterval enables periodic background physical removal of expired -// entries. +// WithCleanupInterval configures the interval between regular cleanup wakeups. // -// The interval controls regular background cleanup wakeups. While expired -// backlog remains, the cleaner may schedule bounded continuation work sooner. -// The interval does not affect logical TTL precision or the internal -// expiration bucket resolution. +// The default is one minute. While expired backlog remains, the cleaner may +// schedule bounded continuation work sooner. The interval does not affect +// logical TTL precision or the internal expiration bucket resolution. // -// Background cleanup is disabled by default. This is the only cleanup option -// that starts a background worker; cleanup batch size and entry budget only -// configure cleanup behavior. Manual cleanup through Cache.DeleteExpired is -// always available without this option. When background cleanup is enabled, -// Close must be called to stop the cleaner goroutine. +// Background cleanup must be started explicitly with StartCleanup. Manual +// cleanup through Cache.DeleteExpired is always available. func WithCleanupInterval(interval time.Duration) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if interval <= 0 { return errors.New("cleanup interval must be positive") } @@ -183,7 +190,7 @@ func WithCleanupInterval(interval time.Duration) Option { // Values larger than a segment or the remaining cleanup budget are safe and // are naturally limited by the available work. The default is 256. func WithCleanupBatchSize(size int) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if size <= 0 { return errors.New("cleanup batch size must be positive") } @@ -204,7 +211,7 @@ func WithCleanupBatchSize(size int) Option { // all entries due at the start of the call are drained. Values larger than the // cache size are safe. The default is 16384. func WithCleanupEntryBudget(entries int) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { if entries <= 0 { return errors.New("cleanup entry budget must be positive") } @@ -217,21 +224,17 @@ func WithCleanupEntryBudget(entries int) Option { // WithMetrics configures optional cache metrics. // -// The Metrics implementation may be reused by multiple caches. Its -// registration is released when Cache.Close is called. +// The Metrics implementation may be reused by multiple caches. The cache +// does not manage the lifecycle of metrics registrations. func WithMetrics(metrics Metrics) Option { - return func(settings *cacheSettings) error { + return func(settings *settings) error { settings.metrics = metrics return nil } } -func (settings *cacheSettings) validate() error { - if settings.name == "" { - return errors.New("cache name must not be empty") - } - +func (settings *settings) validate() error { if settings.ttl > 0 && settings.ttl > maxDuration-settings.jitter { return errors.New("ttl plus jitter exceeds maximum duration") } diff --git a/options_test.go b/options_test.go index 999de5d..4392b95 100644 --- a/options_test.go +++ b/options_test.go @@ -2,14 +2,16 @@ package pacecache import ( "errors" - "strings" "testing" "time" ) -func TestDefaultCacheSettings(t *testing.T) { - settings := defaultCacheSettings() +func TestDefaultSettings(t *testing.T) { + settings := defaultSettings() + if settings.name != "" { + t.Fatalf("name = %q, want empty", settings.name) + } if settings.maxEntries != defaultMaxEntries { t.Fatalf("maxEntries = %d, want %d", settings.maxEntries, defaultMaxEntries) } @@ -19,8 +21,15 @@ func TestDefaultCacheSettings(t *testing.T) { if settings.ttl != defaultTTL { t.Fatalf("ttl = %v, want %v", settings.ttl, defaultTTL) } - if settings.jitter != 0 || settings.cleanupInterval != 0 { - t.Fatalf("optional durations must be disabled by default: %+v", settings) + if settings.jitter != 0 { + t.Fatalf("jitter = %v, want 0", settings.jitter) + } + if settings.cleanupInterval != defaultCleanupInterval { + t.Fatalf( + "cleanupInterval = %v, want %v", + settings.cleanupInterval, + defaultCleanupInterval, + ) } if settings.cleanupBatchSize != defaultCleanupBatchSize { t.Fatalf("cleanupBatchSize = %d, want %d", settings.cleanupBatchSize, defaultCleanupBatchSize) @@ -36,11 +45,11 @@ func TestDefaultCacheSettings(t *testing.T) { } } -func TestNewCacheSettingsAppliesOptions(t *testing.T) { +func TestNewSettingsAppliesOptions(t *testing.T) { metrics := &testMetrics{} - settings, err := newCacheSettings( - "users", + got, err := newSettings( + WithName("users"), WithMaxEntries(128), WithSegmentCount(8), WithTTL(2*time.Minute), @@ -52,93 +61,94 @@ func TestNewCacheSettingsAppliesOptions(t *testing.T) { WithMetrics(metrics), ) if err != nil { - t.Fatalf("newCacheSettings() error = %v", err) + t.Fatalf("newSettings() error = %v", err) } - if settings.name != "users" || - settings.maxEntries != 128 || - settings.segmentCount != 8 || - settings.ttl != 2*time.Minute || - settings.jitter != 15*time.Second || - !settings.slidingExpiration || - settings.cleanupInterval != time.Second || - settings.cleanupBatchSize != 1024 || - settings.cleanupEntryBudget != 64*1024 || - settings.metrics != metrics { - t.Fatalf("unexpected settings: %+v", settings) + want := settings{ + name: "users", + maxEntries: 128, + segmentCount: 8, + ttl: 2 * time.Minute, + jitter: 15 * time.Second, + slidingExpiration: true, + cleanupInterval: time.Second, + cleanupBatchSize: 1024, + cleanupEntryBudget: 64 * 1024, + metrics: metrics, + } + + if *got != want { + t.Fatalf("newSettings() = %+v, want %+v", *got, want) } } -func TestNewCacheSettingsRejectsNilOption(t *testing.T) { - _, err := newCacheSettings("users", nil) - if err == nil || !strings.Contains(err.Error(), "option 0 is nil") { - t.Fatalf("error = %v, want nil-option error", err) +func TestNewSettingsRejectsNilOption(t *testing.T) { + _, err := newSettings(nil) + const want = "option 0 is nil" + if err == nil || err.Error() != want { + t.Fatalf("error = %v, want %q", err, want) } } -func TestNewCacheSettingsWrapsOptionError(t *testing.T) { +func TestNewSettingsWrapsOptionError(t *testing.T) { sentinel := errors.New("sentinel") - option := func(*cacheSettings) error { return sentinel } + option := func(*settings) error { return sentinel } - _, err := newCacheSettings("users", option) + _, err := newSettings(option) if !errors.Is(err, sentinel) { t.Fatalf("error = %v, want wrapped sentinel", err) } } -func TestCacheSettingsValidation(t *testing.T) { +func TestSettingsValidation(t *testing.T) { tests := []struct { name string - cache string options []Option want string }{ - {name: "empty name", cache: "", want: "cache name must not be empty"}, - {name: "max entries zero", cache: "users", options: []Option{WithMaxEntries(0)}, want: "max entries must be positive"}, - {name: "max entries negative", cache: "users", options: []Option{WithMaxEntries(-1)}, want: "max entries must be positive"}, - {name: "segment count zero", cache: "users", options: []Option{WithSegmentCount(0)}, want: "segment count must be positive"}, - {name: "segment count negative", cache: "users", options: []Option{WithSegmentCount(-1)}, want: "segment count must be positive"}, - {name: "ttl zero", cache: "users", options: []Option{WithTTL(0)}, want: "ttl must be positive or NoExpiration"}, - {name: "ttl invalid negative", cache: "users", options: []Option{WithTTL(-2)}, want: "ttl must be positive or NoExpiration"}, - {name: "negative jitter", cache: "users", options: []Option{WithJitter(-1)}, want: "jitter must not be negative"}, - {name: "cleanup interval zero", cache: "users", options: []Option{WithCleanupInterval(0)}, want: "cleanup interval must be positive"}, - {name: "cleanup interval negative", cache: "users", options: []Option{WithCleanupInterval(-1)}, want: "cleanup interval must be positive"}, - {name: "cleanup batch size zero", cache: "users", options: []Option{WithCleanupBatchSize(0)}, want: "cleanup batch size must be positive"}, - {name: "cleanup batch size negative", cache: "users", options: []Option{WithCleanupBatchSize(-1)}, want: "cleanup batch size must be positive"}, - {name: "cleanup entry budget zero", cache: "users", options: []Option{WithCleanupEntryBudget(0)}, want: "cleanup entry budget must be positive"}, - {name: "cleanup entry budget negative", cache: "users", options: []Option{WithCleanupEntryBudget(-1)}, want: "cleanup entry budget must be positive"}, - {name: "segments exceed max entries", cache: "users", options: []Option{WithMaxEntries(2), WithSegmentCount(3)}, want: "segment count must not exceed max entries"}, + {name: "max entries zero", options: []Option{WithMaxEntries(0)}, want: "apply option 0: max entries must be positive"}, + {name: "max entries negative", options: []Option{WithMaxEntries(-1)}, want: "apply option 0: max entries must be positive"}, + {name: "segment count zero", options: []Option{WithSegmentCount(0)}, want: "apply option 0: segment count must be positive"}, + {name: "segment count negative", options: []Option{WithSegmentCount(-1)}, want: "apply option 0: segment count must be positive"}, + {name: "ttl zero", options: []Option{WithTTL(0)}, want: "apply option 0: ttl must be positive or NoExpiration"}, + {name: "ttl invalid negative", options: []Option{WithTTL(-2)}, want: "apply option 0: ttl must be positive or NoExpiration"}, + {name: "negative jitter", options: []Option{WithJitter(-1)}, want: "apply option 0: jitter must not be negative"}, + {name: "cleanup interval zero", options: []Option{WithCleanupInterval(0)}, want: "apply option 0: cleanup interval must be positive"}, + {name: "cleanup interval negative", options: []Option{WithCleanupInterval(-1)}, want: "apply option 0: cleanup interval must be positive"}, + {name: "cleanup batch size zero", options: []Option{WithCleanupBatchSize(0)}, want: "apply option 0: cleanup batch size must be positive"}, + {name: "cleanup batch size negative", options: []Option{WithCleanupBatchSize(-1)}, want: "apply option 0: cleanup batch size must be positive"}, + {name: "cleanup entry budget zero", options: []Option{WithCleanupEntryBudget(0)}, want: "apply option 0: cleanup entry budget must be positive"}, + {name: "cleanup entry budget negative", options: []Option{WithCleanupEntryBudget(-1)}, want: "apply option 0: cleanup entry budget must be positive"}, + {name: "segments exceed max entries", options: []Option{WithMaxEntries(2), WithSegmentCount(3)}, want: "invalid configuration: segment count must not exceed max entries"}, { - name: "ttl plus jitter overflow", - cache: "users", + name: "ttl plus jitter overflow", options: []Option{ WithTTL(maxDuration), WithJitter(time.Nanosecond), }, - want: "ttl plus jitter exceeds maximum duration", + want: "invalid configuration: ttl plus jitter exceeds maximum duration", }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { - _, err := newCacheSettings(test.cache, test.options...) - if err == nil || !strings.Contains(err.Error(), test.want) { - t.Fatalf("error = %v, want substring %q", err, test.want) + _, err := newSettings(test.options...) + if err == nil || err.Error() != test.want { + t.Fatalf("error = %v, want %q", err, test.want) } }) } } -func TestCacheSettingsAcceptsIndependentCleanupLimits(t *testing.T) { - settings, err := newCacheSettings( - "users", +func TestSettingsAcceptsIndependentCleanupLimits(t *testing.T) { + settings, err := newSettings( WithMaxEntries(4), WithSegmentCount(1), WithCleanupBatchSize(10_000), WithCleanupEntryBudget(3), ) if err != nil { - t.Fatalf("newCacheSettings() error = %v", err) + t.Fatalf("newSettings() error = %v", err) } if settings.cleanupBatchSize != 10_000 || settings.cleanupEntryBudget != 3 { @@ -146,18 +156,21 @@ func TestCacheSettingsAcceptsIndependentCleanupLimits(t *testing.T) { } } -func TestCacheSettingsAcceptsBoundaryValues(t *testing.T) { - settings, err := newCacheSettings( - "users", +func TestSettingsAcceptsBoundaryValues(t *testing.T) { + settings, err := newSettings( + WithName(""), WithMaxEntries(1), WithTTL(NoExpiration), WithJitter(maxDuration), WithMetrics(nil), ) if err != nil { - t.Fatalf("newCacheSettings() error = %v", err) + t.Fatalf("newSettings() error = %v", err) } + if settings.name != "" { + t.Fatalf("name = %q, want empty", settings.name) + } if settings.ttl != NoExpiration { t.Fatalf("ttl = %v, want NoExpiration", settings.ttl) } diff --git a/set_test.go b/set_test.go index 3e8c551..0002476 100644 --- a/set_test.go +++ b/set_test.go @@ -9,7 +9,7 @@ import ( ) func TestCacheSetOverwritesExistingValue(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(4), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(4), WithSegmentCount(1)) cache.Set("key", 1, time.Minute) cache.Set("key", 2, NoExpiration) @@ -26,7 +26,6 @@ func TestCacheSetOverwritesExistingValue(t *testing.T) { func TestCacheStoresTTLPolicyForDefaultCustomAndNoExpiration(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(10*time.Second), @@ -61,7 +60,6 @@ func TestCacheStoresTTLPolicyForDefaultCustomAndNoExpiration(t *testing.T) { func TestCacheGetOrSet(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(10*time.Second), @@ -120,7 +118,6 @@ func TestCacheGetOrSet(t *testing.T) { func TestCacheGetOrSetExpirationPolicies(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(10*time.Second), @@ -160,7 +157,7 @@ func TestCacheGetOrSetExpirationPolicies(t *testing.T) { } func TestCacheGetOrSetUsesNoExpirationByDefault(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) value, found := cache.GetOrSet("key", 42, DefaultExpiration) if found || value != 42 { @@ -177,7 +174,7 @@ func TestCacheGetOrSetUsesNoExpirationByDefault(t *testing.T) { } func TestCacheGetOrSetExpiredEntry(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) index := cache.store.segmentIndex("key") cache.store.setAt( @@ -211,7 +208,7 @@ func TestCacheGetOrSetExpiredEntry(t *testing.T) { } func TestCacheGetOrSetIsAtomic(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) const callers = 32 @@ -288,7 +285,7 @@ func TestCacheGetOrSetIsAtomic(t *testing.T) { } func TestCacheGetOrSetSupersedesInflightLoad(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) started := make(chan struct{}) release := make(chan struct{}) @@ -347,7 +344,6 @@ func TestZeroValueCacheGetOrSet(t *testing.T) { func TestCacheGetOrSetEntry(t *testing.T) { cache := mustNewCache[int]( t, - "users", WithMaxEntries(8), WithSegmentCount(1), WithTTL(10*time.Second), @@ -396,7 +392,7 @@ func TestCacheGetOrSetEntry(t *testing.T) { } func TestCacheGetOrSetEntryNoExpirationMetadata(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) entry, found := cache.GetOrSetEntry("key", 42, DefaultExpiration) if found || entry.Value() != 42 { @@ -412,7 +408,7 @@ func TestCacheGetOrSetEntryNoExpirationMetadata(t *testing.T) { } func TestCacheGetOrSetEntryIsAtomic(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) const callers = 32 @@ -487,7 +483,7 @@ func TestCacheGetOrSetEntryIsAtomic(t *testing.T) { } func TestCacheGetOrSetEntrySupersedesInflightLoad(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(8), WithSegmentCount(1)) + cache := mustNewCache[int](t, WithMaxEntries(8), WithSegmentCount(1)) started := make(chan struct{}) release := make(chan struct{}) diff --git a/stats_test.go b/stats_test.go index d497e9a..59a9af7 100644 --- a/stats_test.go +++ b/stats_test.go @@ -11,7 +11,6 @@ func TestStatsSnapshotAggregatesSegments(t *testing.T) { store := newStorageWithExpirationResolution[string, int](4, 2, time.Nanosecond) collector := newStatsCollector(2) cache := &Cache[string, int]{ - name: "users", store: store, states: make([]cacheState[string, int], 2), stats: collector, @@ -46,24 +45,34 @@ func TestStatsSnapshotAggregatesSegments(t *testing.T) { collector.cleanupWorkerPendingCount.Store(16) collector.cleanupWorkerDurationNanos.Store(17) - got := cache.Stats() - if got.EntryCount != 2 || got.MaxEntries != 4 || got.SegmentCount != 2 { - t.Fatalf("state stats = %+v", got) - } - if got.HitCount != 3 || got.MissCount != 7 || got.EvictionCount != 9 || got.ExpirationCount != 11 { - t.Fatalf("storage counters = %+v", got) - } - if got.LoadFoundCount != 13 || got.LoadNotFoundCount != 15 || got.LoadErrorCount != 17 || got.LoadSupersededCount != 19 { - t.Fatalf("load counters = %+v", got) - } - if got.LoadDuration != 2_100*time.Nanosecond { - t.Fatalf("LoadDuration = %v, want 2100ns", got.LoadDuration) - } - if got.SharedCount != 23 || got.DeletedEntryCount != 25 || got.ClearedEntryCount != 13 { - t.Fatalf("atomic counters = %+v", got) + want := Stats{ + EntryCount: 2, + MaxEntries: 4, + SegmentCount: 2, + + HitCount: 3, + MissCount: 7, + EvictionCount: 9, + ExpirationCount: 11, + + LoadFoundCount: 13, + LoadNotFoundCount: 15, + LoadErrorCount: 17, + LoadSupersededCount: 19, + LoadDuration: 2_100 * time.Nanosecond, + SharedCount: 23, + + DeletedEntryCount: 25, + ClearedEntryCount: 13, + + CleanupCount: 14, + CleanupWorkerRunCount: 15, + CleanupWorkerPendingCount: 16, + CleanupWorkerDuration: 17 * time.Nanosecond, } - if got.CleanupCount != 14 || got.CleanupWorkerRunCount != 15 || got.CleanupWorkerPendingCount != 16 || got.CleanupWorkerDuration != 17*time.Nanosecond { - t.Fatalf("cleanup counters = %+v", got) + + if got := cache.Stats(); got != want { + t.Fatalf("Stats() = %+v, want %+v", got, want) } } @@ -130,7 +139,7 @@ func TestStatsRecordHelpers(t *testing.T) { } func TestStatsConcurrentWithCacheOperations(t *testing.T) { - cache := mustNewCache[int](t, "users", WithMaxEntries(128), WithSegmentCount(8)) + cache := mustNewCache[int](t, WithMaxEntries(128), WithSegmentCount(8)) var group sync.WaitGroup for worker := range 8 {