diff --git a/docs/METRICS_CATALOG.md b/docs/METRICS_CATALOG.md index 148d0dad7..08bdbd2e1 100644 --- a/docs/METRICS_CATALOG.md +++ b/docs/METRICS_CATALOG.md @@ -90,10 +90,16 @@ All Metrics in node-vtex-api │ │ ├── latency histogram (via recordLatency) │ │ └── graphql_field_requests_total (Counter) │ │ -│ └── HTTP Agent (HttpClient/middlewares/request/HttpAgentSingleton.ts) -│ ├── http_agent_sockets_current (Gauge) -│ ├── http_agent_free_sockets_current (Gauge) -│ └── http_agent_pending_requests_current (Gauge) +│ ├── HTTP Agent (HttpClient/middlewares/request/HttpAgentSingleton.ts) +│ │ ├── http_agent_sockets_current (Gauge) +│ │ ├── http_agent_free_sockets_current (Gauge) +│ │ └── http_agent_pending_requests_current (Gauge) +│ │ +│ └── Cache (metrics/DiagnosticsMetrics.ts, via trackCache — observable/pull, not per-request) +│ ├── io_app_cache_operations_total (Observable Counter) - attrs: cache, cache_state +│ ├── io_app_cache_items_current (Observable Gauge) - caches that expose itemCount +│ ├── io_app_cache_capacity (Observable Gauge) - caches that expose max +│ └── io_app_cache_disposed_total (Observable Counter) - caches that expose disposedItems │ └── 🏛️ Legacy Metrics (Non-Diagnostics) │ @@ -149,7 +155,7 @@ All Metrics in node-vtex-api │ │ ├── httpAgent - sockets, freeSockets, pendingRequests │ │ └── incomingRequest - total, closed, aborted │ │ - │ └── Cache Metrics (via trackCache) + │ └── Cache Metrics (via trackCache — output discarded since #676; replacement above) │ └── {cache_name}-cache │ ├── LRU: itemCount, length, disposedItems, hitRate, hits, max, total │ ├── Disk: hits, total @@ -242,6 +248,29 @@ These are operation-specific metrics recorded in middleware components. | `http_agent_free_sockets_current` | Gauge | Free sockets in pool | | `http_agent_pending_requests_current` | Gauge | Pending requests waiting for socket | +#### Cache Metrics (Observable) + +**Source:** `metrics/DiagnosticsMetrics.ts` (`trackCache()`) + +The replacement for the legacy `MetricsAccumulator.trackCache()` (see [Legacy Metrics](#legacy-metrics-non-diagnostics) below). Unlike every other metric on this page, these are **observable (pull-based)**: the app registers a cache once, and the four instruments below are read by a callback on the OTel SDK's own collection schedule, not pushed per-request. See `registerObservableGauge`/`registerObservableCounter` on `DiagnosticsMetrics` if you need the same pull model for something other than a cache. + +Reads the cache's `getCumulativeStats()`, which has no side effects — so this can run alongside the legacy `metrics.trackCache()` without either reader consuming the other's counts. A cache registered in both places reports correctly in both, which is what makes a partial migration safe. + +| Metric Name | Type | Attributes | Reported when | +|-------------|------|------------|----------------| +| `io_app_cache_operations_total` | Observable Counter | `cache`, `cache_state` (`hit` \| `miss`) | Cache reports `hits` and `total` (all four cache classes do) | +| `io_app_cache_items_current` | Observable Gauge | `cache` | Cache reports `itemCount` (`LRUCache`, `LRUDiskCache`) | +| `io_app_cache_capacity` | Observable Gauge | `cache` | Cache reports `max` (`LRUCache`, `LRUDiskCache`) | +| `io_app_cache_disposed_total` | Observable Counter | `cache` | Cache reports `disposedItems` (`LRUCache`, `LRUDiskCache`) | + +`hitRate` is not republished — derive it from `io_app_cache_operations_total` (`hit / (hit + miss)`) so it aggregates correctly across instances instead of averaging pre-computed ratios. + +**`io_app_cache_capacity` is in the cache's own units.** It reports the LRU's `max`, which is an item count for a cache built with a plain `max`, but a size budget for one built with a `length` function. Only treat `items_current / capacity` as a fill ratio when you know the cache is count-limited. + +```typescript +const dispose = global.diagnosticsMetrics?.trackCache('pages', pagesCacheStorage) +``` + --- ## Legacy Metrics (Non-Diagnostics) diff --git a/docs/METRICS_OVERVIEW.md b/docs/METRICS_OVERVIEW.md index c6aa1a535..4155d1244 100644 --- a/docs/METRICS_OVERVIEW.md +++ b/docs/METRICS_OVERVIEW.md @@ -172,6 +172,81 @@ global.diagnosticsMetrics.setGauge('cache_items_current', stats.size, { > 📌 **Production Example:** The [render-to-string's `recordCacheMetric()` function](https://github.com/vtex/render-to-string/blob/master/node/utils/metrics.ts) uses `incrementCounter('cache_operations_total', 1, { cache, cache_state })` for unified cache tracking. +**Note:** Pattern 4 above is for *push*-based tracking — you call `incrementCounter`/`setGauge` yourself, at the point in your code where a cache is read. If instead your code uses `metrics.trackCache(name, cache)` — registering a cache instance once, with `MetricsAccumulator` reading its `getStats()` on every flush — that's a different idiom with its own replacement. See Pattern 5. + +### Pattern 5: Registering a Cache for Periodic Observation (`trackCache`) + +**Before:** +```typescript +// Registers the cache once; MetricsAccumulator calls cache.getStats() on every flush +metrics.trackCache('pages', pagesCacheStorage) +``` + +**After:** +```typescript +// Same registration call, same cache instance — DiagnosticsMetrics reads getStats() +// on the OTel SDK's own collection schedule instead of on every legacy flush. +const dispose = global.diagnosticsMetrics?.trackCache('pages', pagesCacheStorage) +``` + +This is a direct replacement, not a manual re-implementation with `incrementCounter`/`setGauge` (Pattern 4's approach) — `trackCache()` reads the cache exactly once per collection cycle no matter how many metrics it produces from it. + +**Leaving the legacy call in place is harmless.** The legacy `getStats()` reports a per-flush window and consumes it on read; `getCumulativeStats()` reports the process-lifetime total and has no side effects, so the two readers don't steal counts from each other: + +```typescript +metrics.trackCache('pages', pagesCacheStorage) // harmless to keep +global.diagnosticsMetrics?.trackCache('pages', pagesCacheStorage) // add; drop the line above when convenient +``` + +This is about safety, not about comparing the two. Since #676 (v7.4.2) nothing consumes what the legacy flush returns — `statusTrack()` keeps running only because flushing also resets the metric accumulators, the CPU baseline and the request stats. So there is no legacy cache metric to compare against; what you get is that a half-migrated app still reports correct numbers. That matters when 23 call sites across three apps are migrated by different people in different PRs: under a read that consumed the counters, registering a cache in both places would have leaked an arbitrary fraction of its counts into a flush that discards them — a plausible-looking, permanently wrong number, with no second metric anywhere to reveal the discrepancy. + +Emits `io_app_cache_operations_total`, `io_app_cache_items_current`, `io_app_cache_capacity` and `io_app_cache_disposed_total` — see [METRICS_CATALOG.md](./METRICS_CATALOG.md#cache-metrics-observable) for the full attribute reference. `hitRate` is not republished; derive it from `io_app_cache_operations_total` instead. + +### Pattern 6: Metrics Computed at Flush Time (`addOnFlushMetric`) + +`metrics.addOnFlushMetric(fn)` registers a function that `MetricsAccumulator` calls on every flush, returning an object that is published as a log line. The replacement is `registerObservableGauge`/`registerObservableCounter` — same idea (a callback read on a schedule), with the OTel collection cycle in place of the legacy flush. + +The translation is not mechanical: a flush metric returns **one object with arbitrarily many fields**, while an observable instrument reports **one numeric value** per observation. So decide, per field, whether it becomes its own instrument or an attribute on a shared one. + +**Before:** +```typescript +metrics.addOnFlushMetric(() => ({ + name: 'my-queue', + size: queue.size, + oldestAgeMs: queue.oldestAgeMs(), +})) +``` + +**After** — two distinct measurements, so two instruments: +```typescript +global.diagnosticsMetrics?.registerObservableGauge( + 'my_queue_size_current', + result => result.observe(queue.size), + { description: 'Items currently queued', unit: '1' } +) + +global.diagnosticsMetrics?.registerObservableGauge( + 'my_queue_oldest_age_milliseconds', + result => result.observe(queue.oldestAgeMs()), + { description: 'Age of the oldest queued item', unit: 'ms' } +) +``` + +When the fields are the *same* measurement split by category, use one instrument and an attribute instead: +```typescript +global.diagnosticsMetrics?.registerObservableGauge('my_queue_size_current', result => { + result.observe(queue.pending, { state: 'pending' }) + result.observe(queue.running, { state: 'running' }) +}) +``` + +Notes that apply to both register methods: + +- Use `registerObservableCounter` only for values that **never decrease** and report the cumulative total — the SDK derives the per-cycle delta itself. Anything that can go down is a gauge. +- Re-registering the same name replaces the previous callback. A name already taken by the other kind is refused with an error log, since two same-named streams of different types break the collector. +- Attributes reported by the callback are subject to the same limit as the push methods; base attributes are not merged, because the callback runs outside any request. +- Both return a disposer. Call it when the thing you're observing goes away. + --- ## What Doesn't Need Migration diff --git a/package.json b/package.json index a4cdda101..f43cc7d56 100644 --- a/package.json +++ b/package.json @@ -95,6 +95,8 @@ "xss": "^1.0.6" }, "devDependencies": { + "@opentelemetry/context-async-hooks": "^1.30.1", + "@opentelemetry/sdk-metrics": "^1.30.1", "@tiagonapoli/opentracing-alternate-mock": "^0.0.3", "@types/archiver": "^2.0.1", "@types/bluebird": "^3.5.27", diff --git a/src/caches/DiskCache.ts b/src/caches/DiskCache.ts index 1f640ba05..ae897fcc5 100644 --- a/src/caches/DiskCache.ts +++ b/src/caches/DiskCache.ts @@ -1,5 +1,5 @@ import { CacheLayer } from './CacheLayer' -import { DiskStats } from './typings' +import { CumulativeStats, DiskStats } from './typings' import { outputJSON, pathExistsSync, readJSON } from 'fs-extra' import { join } from 'path' @@ -10,6 +10,7 @@ export class DiskCache implements CacheLayer{ private hits = 0 private total = 0 private lock: ReadWriteLock + private reported = { hits: 0, total: 0 } constructor(private cachePath: string, private readFile=readJSON, private writeFile=outputJSON) { this.lock = new ReadWriteLock() @@ -22,15 +23,19 @@ export class DiskCache implements CacheLayer{ public getStats = (name='disk-cache'): DiskStats => { const stats = { - hits: this.hits, + hits: this.hits - this.reported.hits, name, - total: this.total, + total: this.total - this.reported.total, } - this.hits = 0 - this.total = 0 + this.reported = { hits: this.hits, total: this.total } return stats } + public getCumulativeStats = (): CumulativeStats => ({ + hits: this.hits, + total: this.total, + }) + public get = async (key: string): Promise => { const pathKey = this.getPathKey(key) this.total += 1 diff --git a/src/caches/LRUCache.ts b/src/caches/LRUCache.ts index 0b5a93ca0..0f1790db0 100644 --- a/src/caches/LRUCache.ts +++ b/src/caches/LRUCache.ts @@ -1,7 +1,7 @@ import LRU from 'lru-cache' import { CacheLayer } from './CacheLayer' import { MultilayeredCache } from './MultilayeredCache' -import { FetchResult, LRUStats } from './typings' +import { CumulativeStats, FetchResult, LRUStats } from './typings' export class LRUCache implements CacheLayer{ private multilayer: MultilayeredCache @@ -9,11 +9,13 @@ export class LRUCache implements CacheLayer{ private hits: number private total: number private disposed: number + private reported: { hits: number, total: number, disposed: number } constructor (options: LRU.Options) { this.hits = 0 this.total = 0 this.disposed = 0 + this.reported = { disposed: 0, hits: 0, total: 0 } this.storage = new LRU({ ...options, dispose: () => this.disposed += 1, @@ -38,19 +40,28 @@ export class LRUCache implements CacheLayer{ public has = (key: K): boolean => this.storage.has(key) public getStats = (name='lru-cache'): LRUStats => { + const hits = this.hits - this.reported.hits + const total = this.total - this.reported.total const stats = { - disposedItems: this.disposed, - hitRate: this.total > 0 ? this.hits / this.total : undefined, - hits: this.hits, + disposedItems: this.disposed - this.reported.disposed, + hitRate: total > 0 ? hits / total : undefined, + hits, itemCount: this.storage.itemCount, length: this.storage.length, max: this.storage.max, name, - total: this.total, + total, } - this.hits = 0 - this.total = 0 - this.disposed = 0 + this.reported = { disposed: this.disposed, hits: this.hits, total: this.total } return stats } + + public getCumulativeStats = (): CumulativeStats => ({ + disposedItems: this.disposed, + hits: this.hits, + itemCount: this.storage.itemCount, + length: this.storage.length, + max: this.storage.max, + total: this.total, + }) } diff --git a/src/caches/LRUDiskCache.ts b/src/caches/LRUDiskCache.ts index 60c16b6d1..dd3bb031b 100644 --- a/src/caches/LRUDiskCache.ts +++ b/src/caches/LRUDiskCache.ts @@ -1,5 +1,5 @@ import { CacheLayer } from './CacheLayer' -import { LRUDiskCacheOptions, LRUStats } from './typings' +import { CumulativeStats, LRUDiskCacheOptions, LRUStats } from './typings' import { outputJSON, readJSON, remove } from 'fs-extra' import LRU from 'lru-cache' @@ -14,6 +14,7 @@ export class LRUDiskCache implements CacheLayer{ private total = 0 private lruStorage: LRU private keyToBeDeleted: string + private reported = { disposed: 0, hits: 0, total: 0 } constructor(private cachePath: string, options: LRUDiskCacheOptions, private readFile=readJSON, private writeFile=outputJSON) { this.hits = 0 @@ -40,22 +41,31 @@ export class LRUDiskCache implements CacheLayer{ public has = (key: string): boolean => this.lruStorage.has(key) public getStats = (name='disk-lru-cache'): LRUStats => { + const hits = this.hits - this.reported.hits + const total = this.total - this.reported.total const stats = { - disposedItems: this.disposed, - hitRate: this.total > 0 ? this.hits / this.total : undefined, - hits: this.hits, + disposedItems: this.disposed - this.reported.disposed, + hitRate: total > 0 ? hits / total : undefined, + hits, itemCount: this.lruStorage.itemCount, length: this.lruStorage.length, max: this.lruStorage.max, name, - total: this.total, + total, } - this.hits = 0 - this.total = 0 - this.disposed = 0 + this.reported = { disposed: this.disposed, hits: this.hits, total: this.total } return stats } + public getCumulativeStats = (): CumulativeStats => ({ + disposedItems: this.disposed, + hits: this.hits, + itemCount: this.lruStorage.itemCount, + length: this.lruStorage.length, + max: this.lruStorage.max, + total: this.total, + }) + public get = async (key: string): Promise => { const timeOfDeath = this.lruStorage.get(key) this.total += 1 diff --git a/src/caches/MultilayeredCache.ts b/src/caches/MultilayeredCache.ts index 9ec1337cd..c3c6aea23 100644 --- a/src/caches/MultilayeredCache.ts +++ b/src/caches/MultilayeredCache.ts @@ -1,11 +1,12 @@ import { any, map, slice } from 'ramda' import { CacheLayer } from './CacheLayer' -import { FetchResult, MultilayerStats } from './typings' +import { CumulativeStats, FetchResult, MultilayerStats } from './typings' export class MultilayeredCache implements CacheLayer{ private hits = 0 private total = 0 + private reported = { hits: 0, total: 0 } constructor (private caches: Array>) {} @@ -45,16 +46,23 @@ export class MultilayeredCache implements CacheLayer{ } public getStats = (name='multilayred-cache'): MultilayerStats => { + const hits = this.hits - this.reported.hits + const total = this.total - this.reported.total const multilayerStats = { - hitRate: this.total > 0 ? this.hits / this.total : undefined, - hits: this.hits, + hitRate: total > 0 ? hits / total : undefined, + hits, name, - total: this.total, + total, } - this.resetCounters() + this.reported = { hits: this.hits, total: this.total } return multilayerStats } + public getCumulativeStats = (): CumulativeStats => ({ + hits: this.hits, + total: this.total, + }) + private findIndex = async (func: (item: T) => Promise, array: T[]): Promise => { this.total += 1 for (let index = 0; index < array.length; index++) { @@ -67,8 +75,4 @@ export class MultilayeredCache implements CacheLayer{ return -1 } - private resetCounters () { - this.hits = 0 - this.total = 0 - } } diff --git a/src/caches/cacheStats.test.ts b/src/caches/cacheStats.test.ts new file mode 100644 index 000000000..15f04490f --- /dev/null +++ b/src/caches/cacheStats.test.ts @@ -0,0 +1,139 @@ +import { DiskCache } from './DiskCache' +import { LRUCache } from './LRUCache' +import { LRUDiskCache } from './LRUDiskCache' +import { MultilayeredCache } from './MultilayeredCache' + +// getStats() reports a per-window delta and has done so since the legacy +// MetricsAccumulator flushed it as a log line. getCumulativeStats() reports the +// process-lifetime total and never resets, so an observable reader and the legacy +// flush can both read the same cache without stealing counts from each other. +describe('cache stats: windowed getStats() vs cumulative getCumulativeStats()', () => { + describe('LRUCache', () => { + const primed = () => { + const cache = new LRUCache({ max: 10 }) + cache.set('a', 1) + return cache + } + + it('getStats() returns only what happened since the previous read', () => { + const cache = primed() + + cache.get('a') + cache.get('a') + cache.get('absent') + expect(pick(cache.getStats())).toEqual({ hits: 2, total: 3, hitRate: 2 / 3 }) + + cache.get('a') + cache.get('absent') + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 2, hitRate: 0.5 }) + + // nothing happened in between + expect(pick(cache.getStats())).toEqual({ hits: 0, total: 0, hitRate: undefined }) + }) + + it('getCumulativeStats() keeps growing across getStats() reads', () => { + const cache = primed() + + cache.get('a') + cache.get('absent') + cache.getStats() // legacy flush consumes the window + expect(cache.getCumulativeStats()).toMatchObject({ hits: 1, total: 2 }) + + cache.get('a') + cache.getStats() // and again + expect(cache.getCumulativeStats()).toMatchObject({ hits: 2, total: 3 }) + }) + + it('getCumulativeStats() is side-effect free: reading it twice reports the same thing', () => { + const cache = primed() + cache.get('a') + + expect(cache.getCumulativeStats()).toEqual(cache.getCumulativeStats()) + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 1, hitRate: 1 }) + }) + + it('counts disposed items in both reads', () => { + const cache = new LRUCache({ max: 1 }) + cache.set('a', 1) + cache.set('b', 2) // evicts 'a' + + expect(cache.getCumulativeStats().disposedItems).toBe(1) + expect(cache.getStats().disposedItems).toBe(1) + expect(cache.getStats().disposedItems).toBe(0) // window consumed + expect(cache.getCumulativeStats().disposedItems).toBe(1) // total survives + }) + + it('exposes itemCount, length and max on both reads', () => { + const cache = primed() + + expect(cache.getCumulativeStats()).toMatchObject({ itemCount: 1, length: 1, max: 10 }) + expect(cache.getStats()).toMatchObject({ itemCount: 1, length: 1, max: 10 }) + }) + }) + + describe('DiskCache', () => { + it('splits the window from the cumulative total', async () => { + const readFile = jest.fn().mockResolvedValue({ value: 1 }) + const cache = new DiskCache('/tmp/does-not-matter', readFile, jest.fn()) + + await cache.get('a') + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 1, hitRate: undefined }) + expect(cache.getCumulativeStats()).toEqual({ hits: 1, total: 1 }) + + await cache.get('b') + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 1, hitRate: undefined }) + expect(cache.getCumulativeStats()).toEqual({ hits: 2, total: 2 }) + }) + + it('counts a failed read as a miss', async () => { + const readFile = jest.fn().mockRejectedValue(new Error('not there')) + const cache = new DiskCache('/tmp/does-not-matter', readFile, jest.fn()) + + await cache.get('a') + + expect(cache.getCumulativeStats()).toEqual({ hits: 0, total: 1 }) + }) + }) + + describe('LRUDiskCache', () => { + it('splits the window from the cumulative total', async () => { + const readFile = jest.fn().mockResolvedValue({ value: 1 }) + const cache = new LRUDiskCache('/tmp/does-not-matter', { max: 10 }, readFile, jest.fn()) + await cache.set('a', 1, 60000) + + await cache.get('a') + await cache.get('absent') + + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 2, hitRate: 0.5 }) + expect(cache.getCumulativeStats()).toMatchObject({ hits: 1, total: 2 }) + + await cache.get('a') + + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 1, hitRate: 1 }) + expect(cache.getCumulativeStats()).toMatchObject({ hits: 2, total: 3 }) + }) + }) + + describe('MultilayeredCache', () => { + it('splits the window from the cumulative total', async () => { + const layer = new LRUCache({ max: 10 }) + layer.set('a', 1) + const cache = new MultilayeredCache([layer]) + + await cache.get('a') + await cache.get('absent') + + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 2, hitRate: 0.5 }) + expect(cache.getCumulativeStats()).toEqual({ hits: 1, total: 2 }) + + await cache.get('a') + + expect(pick(cache.getStats())).toEqual({ hits: 1, total: 1, hitRate: 1 }) + expect(cache.getCumulativeStats()).toEqual({ hits: 2, total: 3 }) + }) + }) +}) + +function pick(stats: { hits: number, total: number, hitRate?: number }) { + return { hitRate: stats.hitRate, hits: stats.hits, total: stats.total } +} diff --git a/src/caches/typings.ts b/src/caches/typings.ts index 8cd43fcc2..6814559ba 100644 --- a/src/caches/typings.ts +++ b/src/caches/typings.ts @@ -23,6 +23,16 @@ export type LRUStats = { total: number, } +// tslint:disable-next-line:interface-over-type-literal +export type CumulativeStats = { + hits: number, + total: number, + disposedItems?: number, + itemCount?: number, + length?: number, + max?: number, +} + // tslint:disable-next-line:interface-over-type-literal export type MultilayerStats = { hitRate: number | undefined, diff --git a/src/metrics/DiagnosticsMetrics.test.ts b/src/metrics/DiagnosticsMetrics.test.ts index c06ab9ede..f950dc76a 100644 --- a/src/metrics/DiagnosticsMetrics.test.ts +++ b/src/metrics/DiagnosticsMetrics.test.ts @@ -1,7 +1,10 @@ import { Types } from '@vtex/diagnostics-nodejs' -import { context } from '@opentelemetry/api' +import { context, ObservableCallback } from '@opentelemetry/api' import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks' -import { DiagnosticsMetrics } from './DiagnosticsMetrics' +import { AggregationTemporality, InMemoryMetricExporter, MeterProvider, PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics' +import { LRUCache } from '../caches/LRUCache' +import { CumulativeStats } from '../caches/typings' +import { DiagnosticsMetrics, TrackedCache } from './DiagnosticsMetrics' // Mock only the external I/O boundary (getMetricClient) jest.mock('../service/metrics/client', () => ({ @@ -751,4 +754,437 @@ describe('DiagnosticsMetrics', () => { }) }) }) + + describe('registerObservableGauge / registerObservableCounter', () => { + // These reach the SDK through metricsClient.getProvider().getMeter(...), so this + // block mocks the Meter directly instead of the createCounter/Gauge/Histogram wrapper. + let gaugeInstruments: Map + let counterInstruments: Map + let observableMeter: { + createObservableGauge: jest.Mock + createObservableCounter: jest.Mock + addBatchObservableCallback: jest.Mock + } + let observableMetricsClient: Types.MetricClient + let observableDiagnostics: DiagnosticsMetrics + + function fakeInstrumentFactory(instruments: Map) { + return jest.fn((name: string) => { + if (!instruments.has(name)) { + instruments.set(name, { addCallback: jest.fn(), removeCallback: jest.fn() }) + } + return instruments.get(name) + }) + } + + beforeEach(async () => { + gaugeInstruments = new Map() + counterInstruments = new Map() + observableMeter = { + createObservableGauge: fakeInstrumentFactory(gaugeInstruments), + createObservableCounter: fakeInstrumentFactory(counterInstruments), + addBatchObservableCallback: jest.fn(), + } + observableMetricsClient = { + createHistogram: jest.fn(), + createCounter: jest.fn(), + createGauge: jest.fn(), + getProvider: () => ({ getMeter: () => observableMeter }), + } as any + + ;(getMetricClient as jest.Mock).mockResolvedValue(observableMetricsClient) + observableDiagnostics = new DiagnosticsMetrics() + await new Promise(resolve => setTimeout(resolve, 10)) + }) + + // The callback handed to the instrument is a wrapper that applies the attribute + // limit, so these assert delegation rather than function identity. + function attachedCallbacksOf(instrument: { addCallback: jest.Mock }): Array { + return instrument.addCallback.mock.calls.map(([callback]) => callback) + } + + function fireLast(instrument: { addCallback: jest.Mock }, result: { observe: jest.Mock }) { + const callbacks = attachedCallbacksOf(instrument) + callbacks[callbacks.length - 1](result as any) + } + + it('creates the instrument (passing options through) and attaches the callback', () => { + const observe = jest.fn() + observableDiagnostics.registerObservableGauge('queue_depth_current', observe, { unit: '1' }) + + expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('queue_depth_current', { unit: '1' }) + + const instrument = gaugeInstruments.get('queue_depth_current')! + expect(instrument.addCallback).toHaveBeenCalledTimes(1) + fireLast(instrument, { observe: jest.fn() }) + expect(observe).toHaveBeenCalledTimes(1) + }) + + it('limits the attributes an observable callback reports', () => { + const eightAttributes = { a: 1, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7, h: 8 } + observableDiagnostics.registerObservableGauge('queue_depth_current', result => result.observe(1, eightAttributes)) + + const observe = jest.fn() + fireLast(gaugeInstruments.get('queue_depth_current')!, { observe }) + + // MAX_CUSTOM_ATTRIBUTES is 7; the eighth is dropped, as with the push methods. + expect(observe).toHaveBeenCalledWith(1, { a: 1, b: 2, c: 3, d: 4, e: 5, f: 6, g: 7 }) + }) + + it('leaves a within-limit attribute set untouched', () => { + observableDiagnostics.registerObservableGauge('queue_depth_current', result => result.observe(3, { queue: 'x' })) + + const observe = jest.fn() + fireLast(gaugeInstruments.get('queue_depth_current')!, { observe }) + + expect(observe).toHaveBeenCalledWith(3, { queue: 'x' }) + }) + + it('reuses the instrument and replaces the previous callback on re-registration', () => { + const first = jest.fn() + const second = jest.fn() + + observableDiagnostics.registerObservableGauge('queue_depth_current', first) + observableDiagnostics.registerObservableGauge('queue_depth_current', second) + + const instrument = gaugeInstruments.get('queue_depth_current')! + expect(observableMeter.createObservableGauge).toHaveBeenCalledTimes(1) + // the wrapper for `first` was detached, and only `second` still fires + expect(instrument.removeCallback).toHaveBeenCalledWith(attachedCallbacksOf(instrument)[0]) + fireLast(instrument, { observe: jest.fn() }) + expect(first).not.toHaveBeenCalled() + expect(second).toHaveBeenCalledTimes(1) + }) + + it('refuses a name already registered as the other kind, instead of publishing two streams', () => { + const errorSpy = jest.spyOn(console, 'error').mockImplementation() + const gauge = jest.fn() + const counter = jest.fn() + + observableDiagnostics.registerObservableGauge('dup_metric', gauge) + const dispose = observableDiagnostics.registerObservableCounter('dup_metric', counter) + + expect(observableMeter.createObservableCounter).not.toHaveBeenCalledWith('dup_metric', undefined) + expect(errorSpy).toHaveBeenCalledWith(expect.stringContaining('dup_metric')) + expect(() => dispose()).not.toThrow() + + errorSpy.mockRestore() + }) + + it('detaches on dispose; the disposer is a no-op if called again', () => { + const observe = jest.fn() + const dispose = observableDiagnostics.registerObservableGauge('queue_depth_current', observe) + + dispose() + dispose() + + expect(gaugeInstruments.get('queue_depth_current')!.removeCallback).toHaveBeenCalledTimes(1) + }) + + it('registerObservableCounter creates a counter instrument (same code path as the gauge)', () => { + const observe = jest.fn() + observableDiagnostics.registerObservableCounter('jobs_processed_total', observe) + + expect(observableMeter.createObservableCounter).toHaveBeenCalledWith('jobs_processed_total', undefined) + fireLast(counterInstruments.get('jobs_processed_total')!, { observe: jest.fn() }) + expect(observe).toHaveBeenCalledTimes(1) + }) + + it('replaces the callback on re-registration of a counter too', () => { + const first = jest.fn() + const second = jest.fn() + + observableDiagnostics.registerObservableCounter('jobs_processed_total', first) + observableDiagnostics.registerObservableCounter('jobs_processed_total', second) + + const instrument = counterInstruments.get('jobs_processed_total')! + expect(observableMeter.createObservableCounter).toHaveBeenCalledTimes(1) + fireLast(instrument, { observe: jest.fn() }) + expect(first).not.toHaveBeenCalled() + expect(second).toHaveBeenCalledTimes(1) + }) + + it('queues the registration when the client is not ready yet, and applies it once it is', async () => { + let resolveClient!: (client: Types.MetricClient) => void + ;(getMetricClient as jest.Mock).mockReturnValueOnce( + new Promise(resolve => { resolveClient = resolve }) + ) + + const pending = new DiagnosticsMetrics() + const observe = jest.fn() + pending.registerObservableGauge('startup_queue_depth', observe) + + expect(observableMeter.createObservableGauge).not.toHaveBeenCalledWith('startup_queue_depth', undefined) + + resolveClient(observableMetricsClient) + await new Promise(resolve => setTimeout(resolve, 10)) + + fireLast(gaugeInstruments.get('startup_queue_depth')!, { observe: jest.fn() }) + expect(observe).toHaveBeenCalledTimes(1) + }) + + it('disposing a still-pending registration prevents it from being applied once ready', async () => { + let resolveClient!: (client: Types.MetricClient) => void + ;(getMetricClient as jest.Mock).mockReturnValueOnce( + new Promise(resolve => { resolveClient = resolve }) + ) + + const pending = new DiagnosticsMetrics() + const dispose = pending.registerObservableGauge('cancelled_before_ready', jest.fn()) + + dispose() + resolveClient(observableMetricsClient) + await new Promise(resolve => setTimeout(resolve, 10)) + + expect(observableMeter.createObservableGauge).not.toHaveBeenCalledWith('cancelled_before_ready', undefined) + }) + }) + + describe('trackCache', () => { + // Exercised against a real MeterProvider + MetricReader instead of a mock: the + // single-read-per-cycle accounting and the cumulative-counter contract are easy to + // get wrong in a way a mock would still pass. + let provider: MeterProvider + let reader: PeriodicExportingMetricReader + let exporter: InMemoryMetricExporter + let cacheMetricsClient: Types.MetricClient + let cacheDiagnostics: DiagnosticsMetrics + + type CollectionResult = Awaited> + + // Values are cumulative, matching what a real cache's getCumulativeStats() reports. + function fakeCache(sequence: Partial[]): TrackedCache { + let call = 0 + return { + getCumulativeStats: () => sequence[Math.min(call++, sequence.length - 1)] as CumulativeStats, + } + } + + // reader.collect() returns the data directly; it doesn't go through the exporter + // (that only happens on the reader's own timer, which never fires in this suite). + function dataPointsIn( + result: CollectionResult, + metricName: string + ): Array<{ value: number; attributes: Record }> { + const points: Array<{ value: number; attributes: Record }> = [] + for (const scopeMetrics of result.resourceMetrics.scopeMetrics) { + for (const metric of scopeMetrics.metrics) { + if (metric.descriptor.name === metricName) { + for (const dataPoint of (metric as any).dataPoints) { + points.push({ value: dataPoint.value as number, attributes: dataPoint.attributes }) + } + } + } + } + + return points + } + + function allMetricNamesIn(result: CollectionResult): string[] { + return result.resourceMetrics.scopeMetrics.flatMap(sm => sm.metrics.map(m => m.descriptor.name)) + } + + function buildClient(temporality: AggregationTemporality) { + exporter = new InMemoryMetricExporter(temporality) + reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 1000000 }) // never fires; collect() is manual + provider = new MeterProvider({ readers: [reader] }) + + return { + createCounter: jest.fn(() => ({ add: jest.fn() })), + createGauge: jest.fn(() => ({ set: jest.fn() })), + createHistogram: jest.fn(), + getProvider: () => provider, + } as any + } + + beforeEach(async () => { + cacheMetricsClient = buildClient(AggregationTemporality.CUMULATIVE) + + ;(getMetricClient as jest.Mock).mockResolvedValue(cacheMetricsClient) + cacheDiagnostics = new DiagnosticsMetrics() + await new Promise(resolve => setTimeout(resolve, 10)) + }) + + afterEach(async () => { + await provider.shutdown() + }) + + it('reports hits and misses split by cache_state', async () => { + cacheDiagnostics.trackCache('pages', fakeCache([{ hits: 3, total: 5 }])) + + const ops = dataPointsIn(await reader.collect(), 'io_app_cache_operations_total') + + expect(ops).toContainEqual({ value: 3, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(ops).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'miss' } }) + }) + + it('reports a monotonic total across collection cycles', async () => { + cacheDiagnostics.trackCache('pages', fakeCache([ + { hits: 3, total: 5 }, + { hits: 5, total: 7 }, + ])) + + await reader.collect() + const second = await reader.collect() + + const ops = dataPointsIn(second, 'io_app_cache_operations_total') + expect(ops).toContainEqual({ value: 5, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(ops).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'miss' } }) + }) + + it('shares a real cache with the legacy flush without either stealing counts', async () => { + // The whole point of reading getCumulativeStats(): MetricsAccumulator keeps + // calling getStats(), which consumes its own window, and this must not notice. + const cache = new LRUCache({ max: 10 }) + cache.set('a', 1) + cacheDiagnostics.trackCache('pages', cache) + + cache.get('a') + cache.get('absent') + cache.getStats() // legacy flush + const first = dataPointsIn(await reader.collect(), 'io_app_cache_operations_total') + + cache.get('a') + cache.getStats() // legacy flush again + const second = dataPointsIn(await reader.collect(), 'io_app_cache_operations_total') + + expect(first).toContainEqual({ value: 1, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(first).toContainEqual({ value: 1, attributes: { cache: 'pages', cache_state: 'miss' } }) + expect(second).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(second).toContainEqual({ value: 1, attributes: { cache: 'pages', cache_state: 'miss' } }) + }) + + it('reports per-cycle deltas under the delta temporality used in production', async () => { + (getMetricClient as jest.Mock).mockResolvedValue(buildClient(AggregationTemporality.DELTA)) + const deltaDiagnostics = new DiagnosticsMetrics() + await new Promise(resolve => setTimeout(resolve, 10)) + + deltaDiagnostics.trackCache('pages', fakeCache([ + { hits: 3, total: 5 }, + { hits: 5, total: 7 }, + ])) + + const first = dataPointsIn(await reader.collect(), 'io_app_cache_operations_total') + const second = dataPointsIn(await reader.collect(), 'io_app_cache_operations_total') + + expect(first).toContainEqual({ value: 3, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(second).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(second).toContainEqual({ value: 0, attributes: { cache: 'pages', cache_state: 'miss' } }) + }) + + it('reads a cache exactly once per cycle no matter how many metrics it feeds', async () => { + const getCumulativeStats = jest.fn().mockReturnValue({ hits: 1, total: 1, itemCount: 10, max: 100, disposedItems: 1 }) + cacheDiagnostics.trackCache('pages', { getCumulativeStats }) + + await reader.collect() + + expect(getCumulativeStats).toHaveBeenCalledTimes(1) + }) + + it('reports itemCount, max and disposedItems only for caches that expose them', async () => { + cacheDiagnostics.trackCache('pages', fakeCache([{ hits: 1, total: 1, itemCount: 10, max: 100, disposedItems: 2 }])) + cacheDiagnostics.trackCache('assets-disk', fakeCache([{ hits: 1, total: 1 }])) // DiskCache shape: no itemCount/max/disposedItems + + const result = await reader.collect() + + expect(dataPointsIn(result, 'io_app_cache_items_current')).toEqual([ + { value: 10, attributes: { cache: 'pages' } }, + ]) + expect(dataPointsIn(result, 'io_app_cache_capacity')).toEqual([ + { value: 100, attributes: { cache: 'pages' } }, + ]) + expect(dataPointsIn(result, 'io_app_cache_disposed_total')).toEqual([ + { value: 2, attributes: { cache: 'pages' } }, + ]) + }) + + it('queues a cache registered before the client is ready, and applies it once it is', async () => { + let resolveClient!: (client: Types.MetricClient) => void + ;(getMetricClient as jest.Mock).mockReturnValueOnce( + new Promise(resolve => { resolveClient = resolve }) + ) + + const pending = new DiagnosticsMetrics() + pending.trackCache('pages', fakeCache([{ hits: 7, total: 9 }])) + + resolveClient(cacheMetricsClient) + await new Promise(resolve => setTimeout(resolve, 10)) + + const ops = dataPointsIn(await reader.collect(), 'io_app_cache_operations_total') + expect(ops).toContainEqual({ value: 7, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(ops).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'miss' } }) + }) + + it('skips the operations counter for an object with no hit/total counters', async () => { + cacheDiagnostics.trackCache('weird', fakeCache([{ itemCount: 5 }])) + + const result = await reader.collect() + + expect(dataPointsIn(result, 'io_app_cache_operations_total')).toHaveLength(0) + expect(dataPointsIn(result, 'io_app_cache_items_current')).toEqual([ + { value: 5, attributes: { cache: 'weird' } }, + ]) + }) + + it('stops reporting a cache once its disposer is called', async () => { + const dispose = cacheDiagnostics.trackCache('pages', fakeCache([{ hits: 1, total: 1 }])) + + dispose() + const result = await reader.collect() + + expect(dataPointsIn(result, 'io_app_cache_operations_total')).toHaveLength(0) + }) + + it('does not publish hitRate, which a real cache does report', async () => { + const cache = new LRUCache({ max: 10 }) + cache.set('a', 1) + cache.get('a') + cache.get('absent') + expect(cache.getStats().hitRate).toBe(0.5) // the legacy read has it... + cacheDiagnostics.trackCache('pages', cache) + + const result = await reader.collect() + + // ...and it is deliberately not republished: derive it from the operations counter. + expect(allMetricNamesIn(result)).not.toEqual(expect.arrayContaining([expect.stringMatching(/hit.?rate/i)])) + }) + + it('never reaches for the provider in an app that uses no observable instrument', async () => { + // Spied before construction: this has to cover initialization too, not just the + // push-based calls, or the "inert if unused" guarantee isn't actually tested. + const getProviderSpy = jest.spyOn(cacheMetricsClient, 'getProvider') + ;(getMetricClient as jest.Mock).mockResolvedValue(cacheMetricsClient) + + const pushOnly = new DiagnosticsMetrics() + await new Promise(resolve => setTimeout(resolve, 10)) + pushOnly.incrementCounter('unrelated_total', 1) + pushOnly.setGauge('unrelated_current', 1) + pushOnly.recordLatency(1, { operation: 'x' }) + + expect(getProviderSpy).not.toHaveBeenCalled() + }) + + it('recovers from a cache whose read throws, without dropping other caches', async () => { + const throwingCache: TrackedCache = { + getCumulativeStats: () => { + throw new Error('boom') + }, + } + const errorSpy = jest.spyOn(console, 'error').mockImplementation() + + const disposeBroken = cacheDiagnostics.trackCache('broken', throwingCache) + cacheDiagnostics.trackCache('pages', fakeCache([{ hits: 1, total: 1 }])) + + const result = await reader.collect() + + expect(dataPointsIn(result, 'io_app_cache_operations_total')).toContainEqual({ + attributes: { cache: 'pages', cache_state: 'hit' }, + value: 1, + }) + expect(errorSpy).toHaveBeenCalled() + + disposeBroken() // avoid a second throw during afterEach's shutdown-triggered collect + errorSpy.mockRestore() + }) + }) }) diff --git a/src/metrics/DiagnosticsMetrics.ts b/src/metrics/DiagnosticsMetrics.ts index 33c533313..9ec9197b6 100644 --- a/src/metrics/DiagnosticsMetrics.ts +++ b/src/metrics/DiagnosticsMetrics.ts @@ -1,7 +1,20 @@ -import { Attributes, context, createContextKey } from '@opentelemetry/api' +import { + Attributes, + BatchObservableResult, + context, + createContextKey, + Meter, + MetricOptions, + Observable, + ObservableCallback, + ObservableCounter, + ObservableGauge, + ObservableResult, +} from '@opentelemetry/api' import { Types } from '@vtex/diagnostics-nodejs' +import { CumulativeStats } from '../caches/typings' +import { LINKED, METRIC_CLIENT_INIT_TIMEOUT_MS } from '../constants' import { getMetricClient } from '../service/metrics/client' -import { METRIC_CLIENT_INIT_TIMEOUT_MS, LINKED } from '../constants' /** * Maximum number of custom attributes allowed per metric call to control cardinality. @@ -18,6 +31,36 @@ const MAX_CUSTOM_ATTRIBUTES = 7 */ const BASE_ATTRIBUTES_KEY = createContextKey('vtex.metrics.baseAttributes') +const OBSERVABLE_METER_NAME = 'node-vtex-api' + +// trackCache() metric names — one shared instrument set, differentiated by `cache`. +const CACHE_OPERATIONS_METRIC = 'io_app_cache_operations_total' +const CACHE_ITEMS_METRIC = 'io_app_cache_items_current' +const CACHE_CAPACITY_METRIC = 'io_app_cache_capacity' +const CACHE_DISPOSED_METRIC = 'io_app_cache_disposed_total' + +// Cache instances that expose a non-resetting read (LRUCache, DiskCache, +// LRUDiskCache, MultilayeredCache). +export interface TrackedCache { + getCumulativeStats(): CumulativeStats +} + +// ObservableGauge and ObservableCounter are both just Observable in the OTel API, +// so registerObservableGauge/Counter share one implementation, keyed by kind. +type ObservableKind = 'gauge' | 'counter' +interface ObservableRegistration { + observe: ObservableCallback + options?: MetricOptions +} + +// `callback` is the app's function (the identity we key replacement/removal on); +// `attached` is the attribute-limiting wrapper actually handed to the instrument. +interface AttachedObservable { + instrument: Observable + callback: ObservableCallback + attached: ObservableCallback +} + /** * Converts an hrtime tuple [seconds, nanoseconds] to milliseconds. */ @@ -54,6 +97,17 @@ function limitCustomAttributes(customAttributes?: Attributes): Attributes | unde return Object.fromEntries(entries.slice(0, MAX_CUSTOM_ATTRIBUTES)) } +/** + * Applies the same cardinality limit the push-based methods use to whatever an + * observable callback reports. Observables run outside any request, so there are no + * base attributes to merge — every attribute here is a custom one. + */ +function limitObservableResult(result: ObservableResult): ObservableResult { + return { + observe: (value: number, attributes?: Attributes) => result.observe(value, limitCustomAttributes(attributes)), + } +} + /** * DiagnosticsMetrics provides a high-level API for recording metrics using * the @vtex/diagnostics-nodejs library. It completely abstracts instrument @@ -113,12 +167,29 @@ export class DiagnosticsMetrics { private latencyHistogram: Types.Histogram | undefined // Counters and gauges keyed by name - private counters: Map - private gauges: Map + private readonly counters: Map + private readonly gauges: Map + + // What apps registered, and what's actually attached to an OTel instrument + // (empty until the client is ready — see syncObservables), keyed by name. + private readonly observableRegistrations: Record> + private readonly observableInstruments: Record> + + // trackCache() state: registered caches and the shared instruments. + private readonly cacheRegistry: Map + private cacheInstruments: { + operations: ObservableCounter + items: ObservableGauge + capacity: ObservableGauge + disposed: ObservableCounter + } | undefined constructor() { this.counters = new Map() this.gauges = new Map() + this.observableRegistrations = { gauge: new Map(), counter: new Map() } + this.observableInstruments = { gauge: new Map(), counter: new Map() } + this.cacheRegistry = new Map() this.initMetricClient() } @@ -145,6 +216,10 @@ export class DiagnosticsMetrics { // Create the single latency histogram after client is ready this.createLatencyHistogram() + // Attach any observable registrations made before the client was ready. + // No-op if there are none. + this.flushPendingObservables() + return this.metricsClient } catch (error) { console.error('Failed to initialize metric client:', error) @@ -383,5 +458,180 @@ export class DiagnosticsMetrics { // Set the gauge value this.gauges.get(name)!.set(value, mergedAttributes) } + + // getProvider() is already part of the metrics client's type. + private getObservableMeter(): Meter | undefined { + return this.metricsClient?.getProvider().getMeter(OBSERVABLE_METER_NAME) + } + + // Registers (or replaces) a pull-based gauge: OTel calls `observe` on its own + // schedule, with `result.observe(value, attributes?)`. No base-attribute merging — + // there's no request in progress when this runs — but the attributes reported are + // held to the same MAX_CUSTOM_ATTRIBUTES limit as the push-based methods. + public registerObservableGauge(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + return this.registerObservable('gauge', name, observe, options) + } + + // Same as registerObservableGauge, but `observe` reports the cumulative total — + // the SDK derives the delta itself. + public registerObservableCounter(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + return this.registerObservable('counter', name, observe, options) + } + + private registerObservable(kind: ObservableKind, name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + // The same name registered as both kinds produces two same-named streams of + // different types in one meter, which the SDK accepts silently and the collector + // then rejects. Refuse the second one instead of publishing a broken metric. + const otherKind: ObservableKind = kind === 'gauge' ? 'counter' : 'gauge' + if (this.observableRegistrations[otherKind].has(name)) { + console.error( + `DiagnosticsMetrics: '${name}' is already registered as an observable ${otherKind}; ` + + `ignoring the ${kind} registration. Pick a distinct metric name.` + ) + return () => undefined + } + + this.observableRegistrations[kind].set(name, { observe, options }) + this.syncObservables(kind) + + return () => { + if (this.observableRegistrations[kind].get(name)?.observe === observe) { + this.observableRegistrations[kind].delete(name) + } + + const active = this.observableInstruments[kind].get(name) + if (active?.callback === observe) { + active.instrument.removeCallback(active.attached) + this.observableInstruments[kind].delete(name) + } + } + } + + // Attaches every `kind` registration to its instrument. A no-op if there's nothing + // registered or the client isn't ready — safe to call anytime, including from + // flushPendingObservables() once the client becomes ready. + private syncObservables(kind: ObservableKind): void { + if (this.observableRegistrations[kind].size === 0) { + return // skip getObservableMeter() entirely if this kind is unused + } + + const meter = this.getObservableMeter() + if (!meter) { + return + } + + for (const [name, { observe, options }] of this.observableRegistrations[kind]) { + const active = this.observableInstruments[kind].get(name) + if (active?.callback === observe) { + continue + } + + if (active) { + active.instrument.removeCallback(active.attached) + } + + const instrument: Observable = active?.instrument ?? + (kind === 'gauge' ? meter.createObservableGauge(name, options) : meter.createObservableCounter(name, options)) + + const attached: ObservableCallback = (result) => observe(limitObservableResult(result)) + instrument.addCallback(attached) + this.observableInstruments[kind].set(name, { instrument, callback: observe, attached }) + } + } + + // Runs once the client is ready. No-op for an app that never calls + // registerObservableGauge/Counter/trackCache. + private flushPendingObservables(): void { + this.syncObservables('gauge') + this.syncObservables('counter') + + if (this.cacheRegistry.size > 0) { + this.ensureCacheInstruments() + } + } + + // Replacement for the legacy MetricsAccumulator.trackCache() — same cache instances, + // see METRICS_CATALOG.md for the metrics emitted. Reads getCumulativeStats(), which + // has no side effects, so this can run alongside the legacy trackCache(). + public trackCache(name: string, cacheInstance: TrackedCache): () => void { + this.cacheRegistry.set(name, cacheInstance) + this.ensureCacheInstruments() + + return () => { + this.cacheRegistry.delete(name) + } + } + + private ensureCacheInstruments(): void { + if (this.cacheInstruments) { + return + } + + const meter = this.getObservableMeter() + if (!meter) { + return + } + + const operations = meter.createObservableCounter(CACHE_OPERATIONS_METRIC, { + description: 'Hit/miss operations for a VTEX IO app in-memory cache', + unit: '1', + }) + const items = meter.createObservableGauge(CACHE_ITEMS_METRIC, { + description: 'Current number of items held by a VTEX IO app cache', + unit: '1', + }) + const capacity = meter.createObservableGauge(CACHE_CAPACITY_METRIC, { + description: 'Capacity of a VTEX IO app cache, in the units that cache uses (item count, unless the LRU was built with a length function)', + unit: '1', + }) + const disposed = meter.createObservableCounter(CACHE_DISPOSED_METRIC, { + description: 'Items disposed (evicted) from a VTEX IO app cache', + unit: '1', + }) + + meter.addBatchObservableCallback( + (result) => this.observeCaches(result), + [operations, items, capacity, disposed] + ) + + this.cacheInstruments = { operations, items, capacity, disposed } + } + + // Reads each registered cache's cumulative counters straight into the instruments. + private observeCaches(result: BatchObservableResult): void { + if (!this.cacheInstruments) { + return + } + + const { operations, items, capacity, disposed } = this.cacheInstruments + + for (const [name, cache] of this.cacheRegistry) { + let stats: CumulativeStats + try { + stats = cache.getCumulativeStats() + } catch (error) { + console.error('DiagnosticsMetrics: failed to read stats for cache', name, error) + continue + } + + const attributes = { cache: name } + if (typeof stats.hits === 'number' && typeof stats.total === 'number') { + result.observe(operations, stats.hits, { ...attributes, cache_state: 'hit' }) + result.observe(operations, Math.max(stats.total - stats.hits, 0), { ...attributes, cache_state: 'miss' }) + } + + if (typeof stats.itemCount === 'number') { + result.observe(items, stats.itemCount, attributes) + } + + if (typeof stats.max === 'number') { + result.observe(capacity, stats.max, attributes) + } + + if (typeof stats.disposedItems === 'number') { + result.observe(disposed, stats.disposedItems, attributes) + } + } + } }