From 91bf3d84408858bd58c0803ddd7c64d1c2f67abe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 14:55:49 -0300 Subject: [PATCH 1/8] feat(metrics): add observable instruments to DiagnosticsMetrics, with trackCache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the approved RFC (vtex/stable-storefronts-specs#8): DiagnosticsMetrics only exposes push-based instruments (recordLatency, incrementCounter, setGauge), so metrics.trackCache() and metrics.addOnFlushMetric() — both pull-based — have had no replacement. Apps migrating off MetricsAccumulator keep those two calls untouched (see vtex/render-server#861), which blocks ever removing the legacy API from apps that use them. Adds three public methods, reaching the OTel SDK through metricsClient.getProvider() — already part of DiagnosticsMetrics' declared type, and the same access node-vtex-api already uses for HostMetricsInstrumentation in service/telemetry/client.ts. No new dependency. - registerObservableGauge(name, observe, options?) / registerObservableCounter(...): generic pull-based instruments, for anything read on a schedule rather than pushed per request (queue depth, pool size, addOnFlushMetric replacements). Re-registering a name replaces the previous callback on the same instrument, matching the precedent in vtex/faststore-cloud's lib/diagnostics.ts. - trackCache(name, cacheInstance): direct replacement for the legacy trackCache, accepting the same LRUCache/DiskCache/LRUDiskCache/ MultilayeredCache instances already in use. Emits io_app_cache_operations_total, io_app_cache_items_current, io_app_cache_capacity and io_app_cache_disposed_total. The one subtlety: these caches' getStats() resets hits/total/disposedItems on every read (so the legacy flush-based model wouldn't double-count between flushes). The RFC's approved decision was corte seco — no dual-write path, no non-resetting read added to the cache classes — so trackCache reads each cache exactly once per collection cycle, via a single addBatchObservableCallback covering all four instruments, and folds each delta into a running cumulative total before observing it (an ObservableCounter must report the cumulative value; the SDK derives the delta itself). Reading the same cache from both the legacy trackCache and this one would split its counts between them — replace, don't add alongside. Also fixed while building this: pending registrations. DiagnosticsMetrics' underlying client initializes asynchronously, but apps call trackCache/ registerObservableGauge/registerObservableCounter synchronously at module load — frequently before that initialization finishes. Registrations made before the client is ready are queued and replayed once it is; this is the one existing method (initMetricClient) this change touches, and it's a no-op for any app that never calls the new APIs — the existing 34 tests still pass without needing to know it exists, since their mock client has no getProvider() at all and it's simply never called. Everything else about this change is additive: new imports, a new exported TrackedCache interface, new private fields initialized in the constructor alongside the existing ones, and new methods appended to the class. No existing method's behavior changes. hitRate is intentionally not republished — it's derivable from io_app_cache_operations_total, and publishing a pre-computed ratio prevents correct aggregation across instances. A Grafana/alert search across ~1,565 dashboards and ~1,769 alert rules (done during the RFC) found no consumer of hitRate/itemCount/disposedItems to preserve. Verified: tsc clean, tslint at baseline (0 errors), 51/51 tests in DiagnosticsMetrics.test.ts (17 new), 223/223 across the full repo suite. trackCache's tests run against a real MeterProvider + MetricReader rather than a mock, specifically to catch the single-read-per-cycle and delta-to-cumulative accounting a mock could get "passing" while still wrong. Docs: added a Cache Metrics (Observable) section to METRICS_CATALOG.md and a Pattern 5 to METRICS_OVERVIEW.md, cross-referenced with the existing Pattern 4 (which is the different, already-supported push-based idiom — hand-rolled incrementCounter/setGauge calls at the point a cache is read, not a registered callback). Co-Authored-By: Claude Sonnet 5 --- docs/METRICS_CATALOG.md | 35 ++- docs/METRICS_OVERVIEW.md | 31 ++ src/metrics/DiagnosticsMetrics.test.ts | 337 +++++++++++++++++++++- src/metrics/DiagnosticsMetrics.ts | 379 ++++++++++++++++++++++++- 4 files changed, 775 insertions(+), 7 deletions(-) diff --git a/docs/METRICS_CATALOG.md b/docs/METRICS_CATALOG.md index 148d0dad7..be6414cb5 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) - only if getStats() has itemCount +│ ├── io_app_cache_capacity (Observable Gauge) - only if getStats() has max +│ └── io_app_cache_disposed_total (Observable Counter) - only if getStats() has 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 — replacement available, see Diagnostics Cache Metrics above) │ └── {cache_name}-cache │ ├── LRU: itemCount, length, disposedItems, hitRate, hits, max, total │ ├── Disk: hits, total @@ -242,6 +248,25 @@ 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. + +| Metric Name | Type | Attributes | Reported when | +|-------------|------|------------|----------------| +| `io_app_cache_operations_total` | Observable Counter | `cache`, `cache_state` (`hit` \| `miss`) | Always | +| `io_app_cache_items_current` | Observable Gauge | `cache` | Cache's `getStats()` returns `itemCount` | +| `io_app_cache_capacity` | Observable Gauge | `cache` | Cache's `getStats()` returns `max` | +| `io_app_cache_disposed_total` | Observable Counter | `cache` | Cache's `getStats()` returns `disposedItems` | + +`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. + +```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..cbbb34387 100644 --- a/docs/METRICS_OVERVIEW.md +++ b/docs/METRICS_OVERVIEW.md @@ -172,6 +172,37 @@ 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 `getStats()` exactly once per collection cycle no matter how many metrics it produces from that one cache, which matters because `hits`/`total`/`disposedItems` reset on every read: reading the same cache from two places (e.g. the legacy `trackCache` and a hand-rolled `incrementCounter` call) would split its counts between them. Migrate a cache by **replacing** the legacy `metrics.trackCache(...)` call, not by adding this alongside it. + +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. + +If you have a periodic value to report that isn't a cache — a queue depth, a connection pool size, anything read on a schedule rather than pushed per-request — use the lower-level `registerObservableGauge`/`registerObservableCounter` that `trackCache` is built on: + +```typescript +const dispose = global.diagnosticsMetrics?.registerObservableGauge( + 'queue_depth_current', + (result) => result.observe(queue.length), + { description: 'Items currently queued', unit: '1' } +) +``` + --- ## What Doesn't Need Migration diff --git a/src/metrics/DiagnosticsMetrics.test.ts b/src/metrics/DiagnosticsMetrics.test.ts index c06ab9ede..2a93d0648 100644 --- a/src/metrics/DiagnosticsMetrics.test.ts +++ b/src/metrics/DiagnosticsMetrics.test.ts @@ -1,7 +1,8 @@ import { Types } from '@vtex/diagnostics-nodejs' import { context } from '@opentelemetry/api' import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks' -import { DiagnosticsMetrics } from './DiagnosticsMetrics' +import { AggregationTemporality, InMemoryMetricExporter, MeterProvider, PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics' +import { DiagnosticsMetrics, TrackedCache } from './DiagnosticsMetrics' // Mock only the external I/O boundary (getMetricClient) jest.mock('../service/metrics/client', () => ({ @@ -751,4 +752,338 @@ describe('DiagnosticsMetrics', () => { }) }) }) + + describe('registerObservableGauge / registerObservableCounter', () => { + // These two methods reach the OTel SDK through metricsClient.getProvider().getMeter(...), + // one level below the createCounter/createGauge/createHistogram wrapper the rest of this + // file mocks. So this block mocks the Meter itself instead — tracking addCallback/ + // removeCallback calls per instrument name, the same way the outer mock tracks + // add()/set()/record() calls per counter/gauge/histogram name. + let observableGaugeInstruments: Map + let observableCounterInstruments: Map + let observableMeter: { + createObservableGauge: jest.Mock + createObservableCounter: jest.Mock + addBatchObservableCallback: jest.Mock + } + let observableMetricsClient: Types.MetricClient + let observableDiagnostics: DiagnosticsMetrics + + beforeEach(async () => { + observableGaugeInstruments = new Map() + observableCounterInstruments = new Map() + + observableMeter = { + createObservableGauge: jest.fn((name: string) => { + if (!observableGaugeInstruments.has(name)) { + observableGaugeInstruments.set(name, { addCallback: jest.fn(), removeCallback: jest.fn() }) + } + return observableGaugeInstruments.get(name) + }), + createObservableCounter: jest.fn((name: string) => { + if (!observableCounterInstruments.has(name)) { + observableCounterInstruments.set(name, { addCallback: jest.fn(), removeCallback: jest.fn() }) + } + return observableCounterInstruments.get(name) + }), + 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)) + }) + + it('creates the instrument and attaches the callback', () => { + const observe = jest.fn() + observableDiagnostics.registerObservableGauge('queue_depth_current', observe) + + expect(observableMeter.createObservableGauge).toHaveBeenCalledTimes(1) + expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('queue_depth_current', undefined) + expect(observableGaugeInstruments.get('queue_depth_current')!.addCallback).toHaveBeenCalledWith(observe) + }) + + it('passes through instrument options (description, unit)', () => { + const observe = jest.fn() + const options = { description: 'Items currently queued', unit: '1' } + observableDiagnostics.registerObservableGauge('queue_depth_current', observe, options) + + expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('queue_depth_current', options) + }) + + 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 = observableGaugeInstruments.get('queue_depth_current')! + expect(observableMeter.createObservableGauge).toHaveBeenCalledTimes(1) + expect(instrument.removeCallback).toHaveBeenCalledWith(first) + expect(instrument.addCallback).toHaveBeenCalledWith(second) + }) + + it('detaches the callback when the returned disposer is called', () => { + const observe = jest.fn() + const dispose = observableDiagnostics.registerObservableGauge('queue_depth_current', observe) + + dispose() + + expect(observableGaugeInstruments.get('queue_depth_current')!.removeCallback).toHaveBeenCalledWith(observe) + }) + + it('disposer is a no-op the second time it is called', () => { + const observe = jest.fn() + const dispose = observableDiagnostics.registerObservableGauge('queue_depth_current', observe) + + dispose() + dispose() + + expect(observableGaugeInstruments.get('queue_depth_current')!.removeCallback).toHaveBeenCalledTimes(1) + }) + + it('creates an observable counter and attaches the callback', () => { + const observe = jest.fn() + observableDiagnostics.registerObservableCounter('jobs_processed_total', observe) + + expect(observableMeter.createObservableCounter).toHaveBeenCalledTimes(1) + expect(observableCounterInstruments.get('jobs_processed_total')!.addCallback).toHaveBeenCalledWith(observe) + }) + + it('reuses the counter instrument and replaces the previous callback on re-registration', () => { + const first = jest.fn() + const second = jest.fn() + + observableDiagnostics.registerObservableCounter('jobs_processed_total', first) + observableDiagnostics.registerObservableCounter('jobs_processed_total', second) + + const instrument = observableCounterInstruments.get('jobs_processed_total')! + expect(observableMeter.createObservableCounter).toHaveBeenCalledTimes(1) + expect(instrument.removeCallback).toHaveBeenCalledWith(first) + expect(instrument.addCallback).toHaveBeenCalledWith(second) + }) + + 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 pendingInstance = new DiagnosticsMetrics() + const observe = jest.fn() + pendingInstance.registerObservableGauge('startup_queue_depth', observe) + + // Not created yet: the client this new instance is waiting on hasn't resolved. + expect(observableMeter.createObservableGauge).not.toHaveBeenCalledWith('startup_queue_depth', undefined) + + resolveClient(observableMetricsClient) + await new Promise(resolve => setTimeout(resolve, 10)) + + expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('startup_queue_depth', undefined) + expect(observableGaugeInstruments.get('startup_queue_depth')!.addCallback).toHaveBeenCalledWith(observe) + }) + + 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 pendingInstance = new DiagnosticsMetrics() + const observe = jest.fn() + const dispose = pendingInstance.registerObservableGauge('cancelled_before_ready', observe) + + dispose() + + resolveClient(observableMetricsClient) + await new Promise(resolve => setTimeout(resolve, 10)) + + expect(observableMeter.createObservableGauge).not.toHaveBeenCalledWith('cancelled_before_ready', undefined) + }) + }) + + describe('trackCache', () => { + // Unlike the rest of this file, trackCache is exercised against the real OTel SDK + // (a real MeterProvider + MetricReader), not a hand-rolled mock. The behavior worth + // trusting here — one read of getStats() per cycle feeding four instruments, and + // cumulative totals correctly accumulated from a delta-on-read source — is exactly + // the kind of thing a mock could get "passing" while still being wrong. + let provider: MeterProvider + let reader: PeriodicExportingMetricReader + let exporter: InMemoryMetricExporter + let cacheMetricsClient: Types.MetricClient + let cacheDiagnostics: DiagnosticsMetrics + + interface CacheStatsFixture { + [key: string]: number | boolean | string | undefined + } + type CollectionResult = Awaited> + + function fakeCache(sequence: CacheStatsFixture[]): TrackedCache { + let call = 0 + return { + getStats: () => sequence[Math.min(call++, sequence.length - 1)], + } + } + + // `reader.collect()` returns the collected data directly — it does not go through + // the configured exporter (that only happens on the reader's own periodic timer, + // which this suite deliberately never lets fire). `InMemoryMetricExporter` is only + // here because PeriodicExportingMetricReader requires some exporter to construct; + // assertions read straight from collect()'s own return value instead. + 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)) + } + + beforeEach(async () => { + exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE) + // Large interval: this suite only ever triggers collection manually via + // reader.collect(); the periodic timer itself must never fire during a test. + reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 1000000 }) + provider = new MeterProvider({ readers: [reader] }) + + cacheMetricsClient = { + createHistogram: jest.fn(), + createCounter: jest.fn(() => ({ add: jest.fn() })), + createGauge: jest.fn(() => ({ set: jest.fn() })), + getProvider: () => provider, + } as any + + ;(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('accumulates across collection cycles instead of treating each delta as the total', async () => { + cacheDiagnostics.trackCache('pages', fakeCache([ + { hits: 3, total: 5 }, + { hits: 2, total: 2 }, + ])) + + await reader.collect() + const second = await reader.collect() + + const ops = dataPointsIn(second, 'io_app_cache_operations_total') + // cumulative hits: 3 + 2 = 5; cumulative misses: (5-3) + (2-2) = 2 + 0 = 2 + expect(ops).toContainEqual({ value: 5, attributes: { cache: 'pages', cache_state: 'hit' } }) + expect(ops).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'miss' } }) + }) + + it('reads a cache exactly once per cycle no matter how many metrics it feeds', async () => { + const getStats = jest.fn().mockReturnValue({ hits: 1, total: 1, itemCount: 10, max: 100, disposedItems: 1 }) + cacheDiagnostics.trackCache('pages', { getStats }) + + await reader.collect() + + expect(getStats).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('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', async () => { + cacheDiagnostics.trackCache('pages', fakeCache([{ hits: 3, total: 5, hitRate: 0.6 }])) + + const result = await reader.collect() + + expect(allMetricNamesIn(result)).not.toEqual(expect.arrayContaining([expect.stringMatching(/hit.?rate/i)])) + }) + + it('does not create the meter for apps that never call trackCache', () => { + const getProviderSpy = jest.spyOn(cacheMetricsClient, 'getProvider') + + // A DiagnosticsMetrics instance that only ever uses the synchronous APIs. + cacheDiagnostics.incrementCounter('unrelated_total', 1) + + expect(getProviderSpy).not.toHaveBeenCalled() + }) + + it('recovers from a cache whose getStats() throws, without dropping other caches', async () => { + const throwingCache: TrackedCache = { + getStats: () => { + 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({ + value: 1, + attributes: { cache: 'pages', cache_state: 'hit' }, + }) + expect(errorSpy).toHaveBeenCalled() + + // Dispose the throwing cache before afterEach's provider.shutdown() triggers one + // more collection cycle — otherwise it throws again through the (by-then-restored) + // real console.error, which is harmless but noisy in the test output. + disposeBroken() + errorSpy.mockRestore() + }) + }) }) diff --git a/src/metrics/DiagnosticsMetrics.ts b/src/metrics/DiagnosticsMetrics.ts index 33c533313..403800044 100644 --- a/src/metrics/DiagnosticsMetrics.ts +++ b/src/metrics/DiagnosticsMetrics.ts @@ -1,4 +1,14 @@ -import { Attributes, context, createContextKey } from '@opentelemetry/api' +import { + Attributes, + BatchObservableResult, + context, + createContextKey, + Meter, + MetricOptions, + ObservableCallback, + ObservableCounter, + ObservableGauge, +} from '@opentelemetry/api' import { Types } from '@vtex/diagnostics-nodejs' import { getMetricClient } from '../service/metrics/client' import { METRIC_CLIENT_INIT_TIMEOUT_MS, LINKED } from '../constants' @@ -18,6 +28,40 @@ const MAX_CUSTOM_ATTRIBUTES = 7 */ const BASE_ATTRIBUTES_KEY = createContextKey('vtex.metrics.baseAttributes') +/** + * Name of the meter used for observable (pull-based) instruments, i.e. instruments + * whose value is read by a callback on the SDK's own collection schedule rather than + * pushed by application code. Kept separate from per-app instrumentation names since + * these instruments live at the node-vtex-api level. + */ +const OBSERVABLE_METER_NAME = 'node-vtex-api' + +/** + * Metric names for the trackCache() cache-visibility instruments. One shared set of + * instruments differentiated by a `cache` attribute, following the same "single + * instrument, many operations" pattern as the latency histogram. + */ +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' + +/** + * The subset of a VTEX IO cache's stats surface that trackCache() understands. + * Matches the shape already returned by the LRUCache, DiskCache, LRUDiskCache and + * MultilayeredCache classes' `getStats()` — see `src/caches/*.ts` and the `GetStats` + * interface in `MetricsAccumulator.ts`, which this mirrors so the same cache instance + * can be passed to either API. + * + * `hits` and `total` are expected to be a delta since the last read (all four cache + * classes reset them on every `getStats()` call); `itemCount`/`length`/`max` are read + * as the current state and are not reset. Fields absent from a given cache type + * (e.g. DiskCache has no `itemCount`) are simply not reported. + */ +export interface TrackedCache { + getStats(): { [key: string]: number | boolean | string | undefined } +} + /** * Converts an hrtime tuple [seconds, nanoseconds] to milliseconds. */ @@ -116,9 +160,43 @@ export class DiagnosticsMetrics { private counters: Map private gauges: Map + // Observable (pull-based) instruments, keyed by name. Each entry tracks the + // OTel instrument handle alongside the callback currently attached to it, so a + // second registration under the same name can detach the old callback before + // attaching the new one instead of accumulating callbacks on the same instrument. + private observableGauges: Map + private observableCounters: Map + + // Observable registrations requested before the metrics client finished initializing. + // The metrics client initializes asynchronously (see initMetricClient), while apps + // typically call trackCache/registerObservableGauge/registerObservableCounter + // synchronously at module load time — often before that initialization completes. + // Without this, those early registrations would be silently dropped. Replayed by + // flushPendingObservables() once the client becomes available. + private pendingObservableGauges: Map + private pendingObservableCounters: Map + + // trackCache() state: caches registered for observation, the running cumulative + // totals derived from their delta-on-read stats (see TrackedCache), and the shared + // instruments + batch callback created once on first use. + private cacheRegistry: Map + private cacheCumulative: Map + private cacheInstruments: { + operations: ObservableCounter + items: ObservableGauge + capacity: ObservableGauge + disposed: ObservableCounter + } | undefined + constructor() { this.counters = new Map() this.gauges = new Map() + this.observableGauges = new Map() + this.observableCounters = new Map() + this.pendingObservableGauges = new Map() + this.pendingObservableCounters = new Map() + this.cacheRegistry = new Map() + this.cacheCumulative = new Map() this.initMetricClient() } @@ -145,6 +223,11 @@ export class DiagnosticsMetrics { // Create the single latency histogram after client is ready this.createLatencyHistogram() + // Replay any registerObservableGauge/registerObservableCounter/trackCache + // calls that arrived before the client was ready. No-op if none arrived — + // apps that never call these APIs are unaffected by this step. + this.flushPendingObservables() + return this.metricsClient } catch (error) { console.error('Failed to initialize metric client:', error) @@ -383,5 +466,299 @@ export class DiagnosticsMetrics { // Set the gauge value this.gauges.get(name)!.set(value, mergedAttributes) } + + /** + * Get the meter used for observable instruments, if the metrics client is ready. + * Reaches the OpenTelemetry MeterProvider through `getProvider()`, which is already + * part of the metrics client's public surface (the same access node-vtex-api uses + * for HostMetricsInstrumentation in service/telemetry/client.ts). + */ + private getObservableMeter(): Meter | undefined { + return this.metricsClient?.getProvider().getMeter(OBSERVABLE_METER_NAME) + } + + /** + * Register (or replace) the callback for a named observable gauge instrument. + * + * Base attributes from `runWithBaseAttributes` are NOT merged here: observable + * callbacks run on the SDK's own collection schedule, not within a request, so + * there is no request-scoped context to merge in. The `observe` callback is + * responsible for supplying whatever attributes it needs directly. + * + * @param name Instrument name (e.g. 'queue_depth_current') + * @param observe Called by the OTel SDK on each collection cycle; use + * `result.observe(value, attributes?)` to report the current value + * @param options Optional instrument metadata (description, unit) + * @returns A disposer that detaches this callback. Safe to call more than once. + * + * @example + * ```typescript + * const dispose = diagnosticsMetrics.registerObservableGauge( + * 'queue_depth_current', + * (result) => result.observe(queue.length), + * { description: 'Items currently queued', unit: '1' } + * ) + * // later, if the queue goes away: + * dispose() + * ``` + */ + public registerObservableGauge(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + if (!this.metricsClient) { + this.pendingObservableGauges.set(name, { observe, options }) + return () => this.detachPendingOrActiveObservableGauge(name, observe) + } + + return this.attachObservableGauge(name, observe, options) + } + + /** + * Register (or replace) the callback for a named observable counter instrument. + * Unlike `incrementCounter`, the callback must report the current cumulative total + * on each collection cycle (not a delta) — the SDK computes the delta itself. + * + * See `registerObservableGauge` for the base-attributes caveat and the "replace on + * re-registration" behavior. + * + * @param name Instrument name (e.g. 'jobs_processed_total') + * @param observe Called by the OTel SDK on each collection cycle; use + * `result.observe(cumulativeValue, attributes?)` + * @param options Optional instrument metadata (description, unit) + * @returns A disposer that detaches this callback. Safe to call more than once. + */ + public registerObservableCounter(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + if (!this.metricsClient) { + this.pendingObservableCounters.set(name, { observe, options }) + return () => this.detachPendingOrActiveObservableCounter(name, observe) + } + + return this.attachObservableCounter(name, observe, options) + } + + private attachObservableGauge(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + const meter = this.getObservableMeter() + if (!meter) { + console.warn('DiagnosticsMetrics not initialized. Call initialize() first.') + return () => {} + } + + const existing = this.observableGauges.get(name) + const instrument = existing?.instrument ?? meter.createObservableGauge(name, options) + if (existing) { + existing.instrument.removeCallback(existing.callback) + } + + instrument.addCallback(observe) + this.observableGauges.set(name, { instrument, callback: observe }) + + return () => this.detachPendingOrActiveObservableGauge(name, observe) + } + + private attachObservableCounter(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + const meter = this.getObservableMeter() + if (!meter) { + console.warn('DiagnosticsMetrics not initialized. Call initialize() first.') + return () => {} + } + + const existing = this.observableCounters.get(name) + const instrument = existing?.instrument ?? meter.createObservableCounter(name, options) + if (existing) { + existing.instrument.removeCallback(existing.callback) + } + + instrument.addCallback(observe) + this.observableCounters.set(name, { instrument, callback: observe }) + + return () => this.detachPendingOrActiveObservableCounter(name, observe) + } + + private detachPendingOrActiveObservableGauge(name: string, observe: ObservableCallback): void { + if (this.pendingObservableGauges.get(name)?.observe === observe) { + this.pendingObservableGauges.delete(name) + } + + const active = this.observableGauges.get(name) + if (active?.callback === observe) { + active.instrument.removeCallback(observe) + this.observableGauges.delete(name) + } + } + + private detachPendingOrActiveObservableCounter(name: string, observe: ObservableCallback): void { + if (this.pendingObservableCounters.get(name)?.observe === observe) { + this.pendingObservableCounters.delete(name) + } + + const active = this.observableCounters.get(name) + if (active?.callback === observe) { + active.instrument.removeCallback(observe) + this.observableCounters.delete(name) + } + } + + /** + * Replay observable registrations that arrived before the metrics client was ready. + * Called once, right after the client finishes initializing. A no-op for any app + * that never calls registerObservableGauge/registerObservableCounter/trackCache. + */ + private flushPendingObservables(): void { + for (const [name, { observe, options }] of this.pendingObservableGauges) { + this.attachObservableGauge(name, observe, options) + } + + this.pendingObservableGauges.clear() + + for (const [name, { observe, options }] of this.pendingObservableCounters) { + this.attachObservableCounter(name, observe, options) + } + + this.pendingObservableCounters.clear() + + // Only touch the cache instruments if trackCache() actually registered something + // before the client was ready. Calling ensureCacheInstruments() unconditionally + // here would call getProvider() on every DiagnosticsMetrics instance, including + // apps that never call trackCache() — the opposite of the "inert unless used" + // guarantee this feature is meant to keep. + if (this.cacheRegistry.size > 0) { + this.ensureCacheInstruments() + } + } + + /** + * Register a cache for periodic, pull-based observation — the DiagnosticsMetrics + * replacement for the legacy MetricsAccumulator.trackCache(). Accepts the same + * cache instances already in use today (LRUCache, DiskCache, LRUDiskCache, + * MultilayeredCache from `../caches`). + * + * Unlike the legacy trackCache, this does not read `cacheInstance.getStats()` + * immediately or on any fixed schedule of its own — it is read once per OTel + * collection cycle, from a single shared callback covering every registered cache, + * so that a cache's delta-on-read counters (`hits`, `total`, `disposedItems`) are + * never read twice in the same cycle and split between two callers. + * + * There is deliberately no dual-write path with the legacy `trackCache`: reading + * the same cache from both would divide its hit/miss counts between them. Replace + * the legacy call with this one in the same change, not alongside it. + * + * Emits, per registered cache (attribute `cache` = the name passed here): + * - `io_app_cache_operations_total` (counter, attribute `cache_state`: 'hit' | 'miss') + * - `io_app_cache_items_current` (gauge) — only if the cache reports `itemCount` + * - `io_app_cache_capacity` (gauge) — only if the cache reports `max` + * - `io_app_cache_disposed_total` (counter) — only if the cache reports `disposedItems` + * + * `hitRate` is intentionally not republished — it is derivable from the operations + * counter, and publishing it directly would prevent correct aggregation across + * instances. + * + * @param name Cache name (e.g. 'pages') — becomes the `cache` attribute + * @param cacheInstance Any cache exposing `getStats()` in the legacy shape + * @returns A disposer that stops observing this cache. Safe to call more than once. + * + * @example + * ```typescript + * const dispose = diagnosticsMetrics.trackCache('pages', pagesCacheStorage) + * ``` + */ + public trackCache(name: string, cacheInstance: TrackedCache): () => void { + this.cacheRegistry.set(name, cacheInstance) + this.ensureCacheInstruments() + + return () => { + this.cacheRegistry.delete(name) + this.cacheCumulative.delete(name) + } + } + + /** + * Lazily create the shared cache instruments and the single batch callback that + * reads every registered cache once per collection cycle. Idempotent: safe to call + * from both `trackCache()` (in case the client is already ready) and + * `flushPendingObservables()` (in case it was not). + */ + private ensureCacheInstruments(): void { + if (this.cacheInstruments) { + return + } + + const meter = this.getObservableMeter() + if (!meter) { + // Not ready yet. trackCache() already recorded the cache in cacheRegistry; + // flushPendingObservables() will call this again once the client is ready. + 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: 'Maximum number of items a VTEX IO app cache can hold', + 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 } + } + + /** + * The single callback backing every registered cache's instruments. Reads each + * cache's `getStats()` exactly once per collection cycle and folds the delta into + * a running cumulative total (see the class-level note on trackCache), since + * `hits`/`total`/`disposedItems` reset on every read. + */ + 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: { [key: string]: number | boolean | string | undefined } + try { + stats = cache.getStats() + } catch (error) { + console.error(`DiagnosticsMetrics: failed to read stats for cache '${name}':`, error) + continue + } + + const running = this.cacheCumulative.get(name) ?? { hits: 0, misses: 0, disposed: 0 } + const hits = typeof stats.hits === 'number' ? stats.hits : 0 + const total = typeof stats.total === 'number' ? stats.total : 0 + running.hits += hits + running.misses += Math.max(total - hits, 0) + + const attributes = { cache: name } + result.observe(operations, running.hits, { ...attributes, cache_state: 'hit' }) + result.observe(operations, running.misses, { ...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') { + running.disposed += stats.disposedItems + result.observe(disposed, running.disposed, attributes) + } + + this.cacheCumulative.set(name, running) + } + } } From 5d8314f7e0436a4306803983456a5b92f5cb42dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 15:29:17 -0300 Subject: [PATCH 2/8] refactor(metrics): unify gauge/counter into one code path, cut comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two things this branch was getting called out for: 1. registerObservableGauge and registerObservableCounter, plus their pending- registration handling, were four near-identical methods and four separate Maps. ObservableGauge and ObservableCounter are the same OTel type (Observable) — createObservableGauge/createObservableCounter only differ in which one they call. Collapsed into one registerObservable/ syncObservables pair keyed by 'gauge' | 'counter'; the two public methods are now one line each. Fixed a bug this surfaced: syncObservables() must check whether there's anything registered for that kind before calling getObservableMeter() — calling it unconditionally (as the merged flushPendingObservables did at first) reintroduces the exact "calls getProvider() even when unused" bug already fixed for the cache path, just on the gauge/counter path instead. 2. JSDoc had grown a multi-paragraph block with an @example per method. Cut to what the existing methods in this file already do: one to a few lines, only the part that isn't obvious from the signature. Also: fixed an unsafe format string flagged on the PR — string-interpolating a variable into console.error's first argument. Passed it as a separate argument instead. Net effect: DiagnosticsMetrics.ts's diff against master goes from +379 to +276 lines; the test file from +337 to +285, after also merging the gauge/counter tests that were exercising the same code path twice. Verified: tsc clean, tslint at baseline, 48/48 in DiagnosticsMetrics.test.ts (down from 51 — 3 tests merged, not dropped: coverage is the same code paths, fewer near-duplicate assertions), 220/220 across the full repo suite. Co-Authored-By: Claude Sonnet 5 --- src/metrics/DiagnosticsMetrics.test.ts | 130 ++++------- src/metrics/DiagnosticsMetrics.ts | 311 +++++++++---------------- 2 files changed, 143 insertions(+), 298 deletions(-) diff --git a/src/metrics/DiagnosticsMetrics.test.ts b/src/metrics/DiagnosticsMetrics.test.ts index 2a93d0648..74df98f4f 100644 --- a/src/metrics/DiagnosticsMetrics.test.ts +++ b/src/metrics/DiagnosticsMetrics.test.ts @@ -754,13 +754,10 @@ describe('DiagnosticsMetrics', () => { }) describe('registerObservableGauge / registerObservableCounter', () => { - // These two methods reach the OTel SDK through metricsClient.getProvider().getMeter(...), - // one level below the createCounter/createGauge/createHistogram wrapper the rest of this - // file mocks. So this block mocks the Meter itself instead — tracking addCallback/ - // removeCallback calls per instrument name, the same way the outer mock tracks - // add()/set()/record() calls per counter/gauge/histogram name. - let observableGaugeInstruments: Map - let observableCounterInstruments: Map + // 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 @@ -769,26 +766,23 @@ describe('DiagnosticsMetrics', () => { let observableMetricsClient: Types.MetricClient let observableDiagnostics: DiagnosticsMetrics - beforeEach(async () => { - observableGaugeInstruments = new Map() - observableCounterInstruments = new Map() + 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: jest.fn((name: string) => { - if (!observableGaugeInstruments.has(name)) { - observableGaugeInstruments.set(name, { addCallback: jest.fn(), removeCallback: jest.fn() }) - } - return observableGaugeInstruments.get(name) - }), - createObservableCounter: jest.fn((name: string) => { - if (!observableCounterInstruments.has(name)) { - observableCounterInstruments.set(name, { addCallback: jest.fn(), removeCallback: jest.fn() }) - } - return observableCounterInstruments.get(name) - }), + createObservableGauge: fakeInstrumentFactory(gaugeInstruments), + createObservableCounter: fakeInstrumentFactory(counterInstruments), addBatchObservableCallback: jest.fn(), } - observableMetricsClient = { createHistogram: jest.fn(), createCounter: jest.fn(), @@ -801,21 +795,12 @@ describe('DiagnosticsMetrics', () => { await new Promise(resolve => setTimeout(resolve, 10)) }) - it('creates the instrument and attaches the callback', () => { + it('creates the instrument (passing options through) and attaches the callback', () => { const observe = jest.fn() - observableDiagnostics.registerObservableGauge('queue_depth_current', observe) + observableDiagnostics.registerObservableGauge('queue_depth_current', observe, { unit: '1' }) - expect(observableMeter.createObservableGauge).toHaveBeenCalledTimes(1) - expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('queue_depth_current', undefined) - expect(observableGaugeInstruments.get('queue_depth_current')!.addCallback).toHaveBeenCalledWith(observe) - }) - - it('passes through instrument options (description, unit)', () => { - const observe = jest.fn() - const options = { description: 'Items currently queued', unit: '1' } - observableDiagnostics.registerObservableGauge('queue_depth_current', observe, options) - - expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('queue_depth_current', options) + expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('queue_depth_current', { unit: '1' }) + expect(gaugeInstruments.get('queue_depth_current')!.addCallback).toHaveBeenCalledWith(observe) }) it('reuses the instrument and replaces the previous callback on re-registration', () => { @@ -825,50 +810,28 @@ describe('DiagnosticsMetrics', () => { observableDiagnostics.registerObservableGauge('queue_depth_current', first) observableDiagnostics.registerObservableGauge('queue_depth_current', second) - const instrument = observableGaugeInstruments.get('queue_depth_current')! + const instrument = gaugeInstruments.get('queue_depth_current')! expect(observableMeter.createObservableGauge).toHaveBeenCalledTimes(1) expect(instrument.removeCallback).toHaveBeenCalledWith(first) expect(instrument.addCallback).toHaveBeenCalledWith(second) }) - it('detaches the callback when the returned disposer is called', () => { - const observe = jest.fn() - const dispose = observableDiagnostics.registerObservableGauge('queue_depth_current', observe) - - dispose() - - expect(observableGaugeInstruments.get('queue_depth_current')!.removeCallback).toHaveBeenCalledWith(observe) - }) - - it('disposer is a no-op the second time it is called', () => { + 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(observableGaugeInstruments.get('queue_depth_current')!.removeCallback).toHaveBeenCalledTimes(1) + expect(gaugeInstruments.get('queue_depth_current')!.removeCallback).toHaveBeenCalledTimes(1) }) - it('creates an observable counter and attaches the callback', () => { + 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).toHaveBeenCalledTimes(1) - expect(observableCounterInstruments.get('jobs_processed_total')!.addCallback).toHaveBeenCalledWith(observe) - }) - - it('reuses the counter instrument and replaces the previous callback on re-registration', () => { - const first = jest.fn() - const second = jest.fn() - - observableDiagnostics.registerObservableCounter('jobs_processed_total', first) - observableDiagnostics.registerObservableCounter('jobs_processed_total', second) - - const instrument = observableCounterInstruments.get('jobs_processed_total')! - expect(observableMeter.createObservableCounter).toHaveBeenCalledTimes(1) - expect(instrument.removeCallback).toHaveBeenCalledWith(first) - expect(instrument.addCallback).toHaveBeenCalledWith(second) + expect(observableMeter.createObservableCounter).toHaveBeenCalledWith('jobs_processed_total', undefined) + expect(counterInstruments.get('jobs_processed_total')!.addCallback).toHaveBeenCalledWith(observe) }) it('queues the registration when the client is not ready yet, and applies it once it is', async () => { @@ -877,18 +840,16 @@ describe('DiagnosticsMetrics', () => { new Promise(resolve => { resolveClient = resolve }) ) - const pendingInstance = new DiagnosticsMetrics() + const pending = new DiagnosticsMetrics() const observe = jest.fn() - pendingInstance.registerObservableGauge('startup_queue_depth', observe) + pending.registerObservableGauge('startup_queue_depth', observe) - // Not created yet: the client this new instance is waiting on hasn't resolved. expect(observableMeter.createObservableGauge).not.toHaveBeenCalledWith('startup_queue_depth', undefined) resolveClient(observableMetricsClient) await new Promise(resolve => setTimeout(resolve, 10)) - expect(observableMeter.createObservableGauge).toHaveBeenCalledWith('startup_queue_depth', undefined) - expect(observableGaugeInstruments.get('startup_queue_depth')!.addCallback).toHaveBeenCalledWith(observe) + expect(gaugeInstruments.get('startup_queue_depth')!.addCallback).toHaveBeenCalledWith(observe) }) it('disposing a still-pending registration prevents it from being applied once ready', async () => { @@ -897,12 +858,10 @@ describe('DiagnosticsMetrics', () => { new Promise(resolve => { resolveClient = resolve }) ) - const pendingInstance = new DiagnosticsMetrics() - const observe = jest.fn() - const dispose = pendingInstance.registerObservableGauge('cancelled_before_ready', observe) + const pending = new DiagnosticsMetrics() + const dispose = pending.registerObservableGauge('cancelled_before_ready', jest.fn()) dispose() - resolveClient(observableMetricsClient) await new Promise(resolve => setTimeout(resolve, 10)) @@ -911,11 +870,9 @@ describe('DiagnosticsMetrics', () => { }) describe('trackCache', () => { - // Unlike the rest of this file, trackCache is exercised against the real OTel SDK - // (a real MeterProvider + MetricReader), not a hand-rolled mock. The behavior worth - // trusting here — one read of getStats() per cycle feeding four instruments, and - // cumulative totals correctly accumulated from a delta-on-read source — is exactly - // the kind of thing a mock could get "passing" while still being wrong. + // Exercised against a real MeterProvider + MetricReader instead of a mock: the + // single-read-per-cycle and delta-to-cumulative accounting are easy to get wrong + // in a way a mock would still pass. let provider: MeterProvider let reader: PeriodicExportingMetricReader let exporter: InMemoryMetricExporter @@ -934,11 +891,8 @@ describe('DiagnosticsMetrics', () => { } } - // `reader.collect()` returns the collected data directly — it does not go through - // the configured exporter (that only happens on the reader's own periodic timer, - // which this suite deliberately never lets fire). `InMemoryMetricExporter` is only - // here because PeriodicExportingMetricReader requires some exporter to construct; - // assertions read straight from collect()'s own return value instead. + // 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 @@ -963,9 +917,7 @@ describe('DiagnosticsMetrics', () => { beforeEach(async () => { exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE) - // Large interval: this suite only ever triggers collection manually via - // reader.collect(); the periodic timer itself must never fire during a test. - reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 1000000 }) + reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 1000000 }) // never fires; collect() is manual provider = new MeterProvider({ readers: [reader] }) cacheMetricsClient = { @@ -1054,7 +1006,6 @@ describe('DiagnosticsMetrics', () => { it('does not create the meter for apps that never call trackCache', () => { const getProviderSpy = jest.spyOn(cacheMetricsClient, 'getProvider') - // A DiagnosticsMetrics instance that only ever uses the synchronous APIs. cacheDiagnostics.incrementCounter('unrelated_total', 1) expect(getProviderSpy).not.toHaveBeenCalled() @@ -1079,10 +1030,7 @@ describe('DiagnosticsMetrics', () => { }) expect(errorSpy).toHaveBeenCalled() - // Dispose the throwing cache before afterEach's provider.shutdown() triggers one - // more collection cycle — otherwise it throws again through the (by-then-restored) - // real console.error, which is harmless but noisy in the test output. - disposeBroken() + 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 403800044..2094f7114 100644 --- a/src/metrics/DiagnosticsMetrics.ts +++ b/src/metrics/DiagnosticsMetrics.ts @@ -5,6 +5,7 @@ import { createContextKey, Meter, MetricOptions, + Observable, ObservableCallback, ObservableCounter, ObservableGauge, @@ -28,40 +29,35 @@ const MAX_CUSTOM_ATTRIBUTES = 7 */ const BASE_ATTRIBUTES_KEY = createContextKey('vtex.metrics.baseAttributes') -/** - * Name of the meter used for observable (pull-based) instruments, i.e. instruments - * whose value is read by a callback on the SDK's own collection schedule rather than - * pushed by application code. Kept separate from per-app instrumentation names since - * these instruments live at the node-vtex-api level. - */ +// Meter for observable (pull-based) instruments — read by the SDK on its own +// collection schedule, unlike the push-based ones above. const OBSERVABLE_METER_NAME = 'node-vtex-api' -/** - * Metric names for the trackCache() cache-visibility instruments. One shared set of - * instruments differentiated by a `cache` attribute, following the same "single - * instrument, many operations" pattern as the latency histogram. - */ +// 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' /** - * The subset of a VTEX IO cache's stats surface that trackCache() understands. - * Matches the shape already returned by the LRUCache, DiskCache, LRUDiskCache and - * MultilayeredCache classes' `getStats()` — see `src/caches/*.ts` and the `GetStats` - * interface in `MetricsAccumulator.ts`, which this mirrors so the same cache instance - * can be passed to either API. - * - * `hits` and `total` are expected to be a delta since the last read (all four cache - * classes reset them on every `getStats()` call); `itemCount`/`length`/`max` are read - * as the current state and are not reset. Fields absent from a given cache type - * (e.g. DiskCache has no `itemCount`) are simply not reported. + * The stats shape trackCache() reads — matches LRUCache/DiskCache/LRUDiskCache/ + * MultilayeredCache's getStats() (see `../caches` and `GetStats` in + * MetricsAccumulator.ts). `hits`/`total`/`disposedItems` reset on every read; + * `itemCount`/`max` don't. Fields a cache doesn't have are just not reported. */ export interface TrackedCache { getStats(): { [key: string]: number | boolean | string | undefined } } +// ObservableGauge and ObservableCounter are the same type in the OTel API +// (Observable), so registerObservableGauge/Counter share one implementation below, +// keyed by which kind of instrument to create. +type ObservableKind = 'gauge' | 'counter' +interface ObservableRegistration { + observe: ObservableCallback + options?: MetricOptions +} + /** * Converts an hrtime tuple [seconds, nanoseconds] to milliseconds. */ @@ -160,25 +156,14 @@ export class DiagnosticsMetrics { private counters: Map private gauges: Map - // Observable (pull-based) instruments, keyed by name. Each entry tracks the - // OTel instrument handle alongside the callback currently attached to it, so a - // second registration under the same name can detach the old callback before - // attaching the new one instead of accumulating callbacks on the same instrument. - private observableGauges: Map - private observableCounters: Map - - // Observable registrations requested before the metrics client finished initializing. - // The metrics client initializes asynchronously (see initMetricClient), while apps - // typically call trackCache/registerObservableGauge/registerObservableCounter - // synchronously at module load time — often before that initialization completes. - // Without this, those early registrations would be silently dropped. Replayed by - // flushPendingObservables() once the client becomes available. - private pendingObservableGauges: Map - private pendingObservableCounters: Map - - // trackCache() state: caches registered for observation, the running cumulative - // totals derived from their delta-on-read stats (see TrackedCache), and the shared - // instruments + batch callback created once on first use. + // Observable (pull-based) instruments: what apps asked to register, and what's + // actually attached to an OTel instrument (empty until the client is ready — see + // syncObservables). Re-registering a name replaces its callback. + private observableRegistrations: Record> + private observableInstruments: Record> + + // trackCache(): registered caches, their running cumulative totals (getStats() + // resets on read — see observeCaches), and the shared instruments, created once. private cacheRegistry: Map private cacheCumulative: Map private cacheInstruments: { @@ -191,10 +176,8 @@ export class DiagnosticsMetrics { constructor() { this.counters = new Map() this.gauges = new Map() - this.observableGauges = new Map() - this.observableCounters = new Map() - this.pendingObservableGauges = new Map() - this.pendingObservableCounters = new Map() + this.observableRegistrations = { gauge: new Map(), counter: new Map() } + this.observableInstruments = { gauge: new Map(), counter: new Map() } this.cacheRegistry = new Map() this.cacheCumulative = new Map() this.initMetricClient() @@ -223,9 +206,8 @@ export class DiagnosticsMetrics { // Create the single latency histogram after client is ready this.createLatencyHistogram() - // Replay any registerObservableGauge/registerObservableCounter/trackCache - // calls that arrived before the client was ready. No-op if none arrived — - // apps that never call these APIs are unaffected by this step. + // Attach any observable registrations made before the client was ready. + // No-op if there are none. this.flushPendingObservables() return this.metricsClient @@ -473,191 +455,114 @@ export class DiagnosticsMetrics { * part of the metrics client's public surface (the same access node-vtex-api uses * for HostMetricsInstrumentation in service/telemetry/client.ts). */ + // Reaches the OTel MeterProvider via getProvider(), already part of the metrics + // client's type — the same access node-vtex-api uses for HostMetricsInstrumentation + // in service/telemetry/client.ts. private getObservableMeter(): Meter | undefined { return this.metricsClient?.getProvider().getMeter(OBSERVABLE_METER_NAME) } /** - * Register (or replace) the callback for a named observable gauge instrument. - * - * Base attributes from `runWithBaseAttributes` are NOT merged here: observable - * callbacks run on the SDK's own collection schedule, not within a request, so - * there is no request-scoped context to merge in. The `observe` callback is - * responsible for supplying whatever attributes it needs directly. + * Register (or replace, if `name` is already registered) a pull-based gauge: OTel + * calls `observe` on its own collection schedule and expects `result.observe(value, + * attributes?)`. Unlike the push-based methods above, base attributes are not + * merged — there's no request in progress when this runs, so `observe` must supply + * whatever attributes it needs. * - * @param name Instrument name (e.g. 'queue_depth_current') - * @param observe Called by the OTel SDK on each collection cycle; use - * `result.observe(value, attributes?)` to report the current value - * @param options Optional instrument metadata (description, unit) - * @returns A disposer that detaches this callback. Safe to call more than once. - * - * @example - * ```typescript - * const dispose = diagnosticsMetrics.registerObservableGauge( - * 'queue_depth_current', - * (result) => result.observe(queue.length), - * { description: 'Items currently queued', unit: '1' } - * ) - * // later, if the queue goes away: - * dispose() - * ``` + * @returns A disposer that detaches the callback. Safe to call more than once, and + * safe to call before the metrics client is ready. */ public registerObservableGauge(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { - if (!this.metricsClient) { - this.pendingObservableGauges.set(name, { observe, options }) - return () => this.detachPendingOrActiveObservableGauge(name, observe) - } - - return this.attachObservableGauge(name, observe, options) + return this.registerObservable('gauge', name, observe, options) } /** - * Register (or replace) the callback for a named observable counter instrument. - * Unlike `incrementCounter`, the callback must report the current cumulative total - * on each collection cycle (not a delta) — the SDK computes the delta itself. - * - * See `registerObservableGauge` for the base-attributes caveat and the "replace on - * re-registration" behavior. - * - * @param name Instrument name (e.g. 'jobs_processed_total') - * @param observe Called by the OTel SDK on each collection cycle; use - * `result.observe(cumulativeValue, attributes?)` - * @param options Optional instrument metadata (description, unit) - * @returns A disposer that detaches this callback. Safe to call more than once. + * Same as `registerObservableGauge`, but for a monotonically increasing total: + * `observe` must report the current cumulative value (the SDK derives the delta), + * the same way `io_app_cache_operations_total` does below. */ public registerObservableCounter(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { - if (!this.metricsClient) { - this.pendingObservableCounters.set(name, { observe, options }) - return () => this.detachPendingOrActiveObservableCounter(name, observe) - } - - return this.attachObservableCounter(name, observe, options) + return this.registerObservable('counter', name, observe, options) } - private attachObservableGauge(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { - const meter = this.getObservableMeter() - if (!meter) { - console.warn('DiagnosticsMetrics not initialized. Call initialize() first.') - return () => {} - } + private registerObservable(kind: ObservableKind, name: string, observe: ObservableCallback, options?: MetricOptions): () => void { + this.observableRegistrations[kind].set(name, { observe, options }) + this.syncObservables(kind) - const existing = this.observableGauges.get(name) - const instrument = existing?.instrument ?? meter.createObservableGauge(name, options) - if (existing) { - existing.instrument.removeCallback(existing.callback) - } - - instrument.addCallback(observe) - this.observableGauges.set(name, { instrument, callback: observe }) + return () => { + if (this.observableRegistrations[kind].get(name)?.observe === observe) { + this.observableRegistrations[kind].delete(name) + } - return () => this.detachPendingOrActiveObservableGauge(name, observe) + const active = this.observableInstruments[kind].get(name) + if (active?.callback === observe) { + active.instrument.removeCallback(observe) + this.observableInstruments[kind].delete(name) + } + } } - private attachObservableCounter(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { - const meter = this.getObservableMeter() - if (!meter) { - console.warn('DiagnosticsMetrics not initialized. Call initialize() first.') - return () => {} + // Attaches every `kind` registration to its instrument. Safe to call anytime — + // a no-op if the client isn't ready yet, or if nothing changed since last call. + // This is what makes registering before the client is ready work: the caller + // gets its disposer immediately, and the actual OTel wiring happens here, once, + // whenever the meter becomes available (see flushPendingObservables). + private syncObservables(kind: ObservableKind): void { + // Checked before getObservableMeter(): an app that never registers anything of + // this kind must never touch getProvider(), or the "inert unless used" guarantee + // breaks for it. + if (this.observableRegistrations[kind].size === 0) { + return } - const existing = this.observableCounters.get(name) - const instrument = existing?.instrument ?? meter.createObservableCounter(name, options) - if (existing) { - existing.instrument.removeCallback(existing.callback) + const meter = this.getObservableMeter() + if (!meter) { + return } - instrument.addCallback(observe) - this.observableCounters.set(name, { instrument, callback: observe }) - - return () => this.detachPendingOrActiveObservableCounter(name, observe) - } - - private detachPendingOrActiveObservableGauge(name: string, observe: ObservableCallback): void { - if (this.pendingObservableGauges.get(name)?.observe === observe) { - this.pendingObservableGauges.delete(name) - } + for (const [name, { observe, options }] of this.observableRegistrations[kind]) { + const active = this.observableInstruments[kind].get(name) + if (active?.callback === observe) { + continue + } - const active = this.observableGauges.get(name) - if (active?.callback === observe) { - active.instrument.removeCallback(observe) - this.observableGauges.delete(name) - } - } + if (active) { + active.instrument.removeCallback(active.callback) + } - private detachPendingOrActiveObservableCounter(name: string, observe: ObservableCallback): void { - if (this.pendingObservableCounters.get(name)?.observe === observe) { - this.pendingObservableCounters.delete(name) - } + const instrument: Observable = active?.instrument ?? + (kind === 'gauge' ? meter.createObservableGauge(name, options) : meter.createObservableCounter(name, options)) - const active = this.observableCounters.get(name) - if (active?.callback === observe) { - active.instrument.removeCallback(observe) - this.observableCounters.delete(name) + instrument.addCallback(observe) + this.observableInstruments[kind].set(name, { instrument, callback: observe }) } } - /** - * Replay observable registrations that arrived before the metrics client was ready. - * Called once, right after the client finishes initializing. A no-op for any app - * that never calls registerObservableGauge/registerObservableCounter/trackCache. - */ + // Attaches whatever was registered before the client was ready. No-op for an app + // that never calls registerObservableGauge/Counter/trackCache. private flushPendingObservables(): void { - for (const [name, { observe, options }] of this.pendingObservableGauges) { - this.attachObservableGauge(name, observe, options) - } - - this.pendingObservableGauges.clear() - - for (const [name, { observe, options }] of this.pendingObservableCounters) { - this.attachObservableCounter(name, observe, options) - } + this.syncObservables('gauge') + this.syncObservables('counter') - this.pendingObservableCounters.clear() - - // Only touch the cache instruments if trackCache() actually registered something - // before the client was ready. Calling ensureCacheInstruments() unconditionally - // here would call getProvider() on every DiagnosticsMetrics instance, including - // apps that never call trackCache() — the opposite of the "inert unless used" - // guarantee this feature is meant to keep. if (this.cacheRegistry.size > 0) { this.ensureCacheInstruments() } } /** - * Register a cache for periodic, pull-based observation — the DiagnosticsMetrics - * replacement for the legacy MetricsAccumulator.trackCache(). Accepts the same - * cache instances already in use today (LRUCache, DiskCache, LRUDiskCache, - * MultilayeredCache from `../caches`). - * - * Unlike the legacy trackCache, this does not read `cacheInstance.getStats()` - * immediately or on any fixed schedule of its own — it is read once per OTel - * collection cycle, from a single shared callback covering every registered cache, - * so that a cache's delta-on-read counters (`hits`, `total`, `disposedItems`) are - * never read twice in the same cycle and split between two callers. - * - * There is deliberately no dual-write path with the legacy `trackCache`: reading - * the same cache from both would divide its hit/miss counts between them. Replace - * the legacy call with this one in the same change, not alongside it. - * - * Emits, per registered cache (attribute `cache` = the name passed here): - * - `io_app_cache_operations_total` (counter, attribute `cache_state`: 'hit' | 'miss') - * - `io_app_cache_items_current` (gauge) — only if the cache reports `itemCount` - * - `io_app_cache_capacity` (gauge) — only if the cache reports `max` - * - `io_app_cache_disposed_total` (counter) — only if the cache reports `disposedItems` + * Replacement for the legacy `MetricsAccumulator.trackCache()`, taking the same + * cache instances (LRUCache, DiskCache, LRUDiskCache, MultilayeredCache — see + * `../caches`). Reads `getStats()` once per OTel collection cycle, not on any + * schedule of its own, and folds the delta into a running total (see + * `observeCaches`) since `hits`/`total`/`disposedItems` reset on every read. * - * `hitRate` is intentionally not republished — it is derivable from the operations - * counter, and publishing it directly would prevent correct aggregation across - * instances. + * Migrate a cache by replacing the legacy `trackCache` call, not adding this one + * alongside it — reading the same cache from both would split its counts between + * them. Emits `io_app_cache_operations_total`, `_items_current`, `_capacity` and + * `_disposed_total` (see METRICS_CATALOG.md); `hitRate` is not republished, derive + * it from `_operations_total` instead. * - * @param name Cache name (e.g. 'pages') — becomes the `cache` attribute - * @param cacheInstance Any cache exposing `getStats()` in the legacy shape * @returns A disposer that stops observing this cache. Safe to call more than once. - * - * @example - * ```typescript - * const dispose = diagnosticsMetrics.trackCache('pages', pagesCacheStorage) - * ``` */ public trackCache(name: string, cacheInstance: TrackedCache): () => void { this.cacheRegistry.set(name, cacheInstance) @@ -669,12 +574,9 @@ export class DiagnosticsMetrics { } } - /** - * Lazily create the shared cache instruments and the single batch callback that - * reads every registered cache once per collection cycle. Idempotent: safe to call - * from both `trackCache()` (in case the client is already ready) and - * `flushPendingObservables()` (in case it was not). - */ + // Lazily creates the shared cache instruments and the one batch callback that + // reads every registered cache per cycle. Idempotent — called from trackCache() + // and, in case the client wasn't ready yet, from flushPendingObservables(). private ensureCacheInstruments(): void { if (this.cacheInstruments) { return @@ -682,8 +584,6 @@ export class DiagnosticsMetrics { const meter = this.getObservableMeter() if (!meter) { - // Not ready yet. trackCache() already recorded the cache in cacheRegistry; - // flushPendingObservables() will call this again once the client is ready. return } @@ -712,12 +612,9 @@ export class DiagnosticsMetrics { this.cacheInstruments = { operations, items, capacity, disposed } } - /** - * The single callback backing every registered cache's instruments. Reads each - * cache's `getStats()` exactly once per collection cycle and folds the delta into - * a running cumulative total (see the class-level note on trackCache), since - * `hits`/`total`/`disposedItems` reset on every read. - */ + // The one callback backing every cache's instruments — reads each cache exactly + // once per cycle (never twice, never split across two callbacks) and turns its + // delta-on-read stats into a running cumulative total before observing it. private observeCaches(result: BatchObservableResult): void { if (!this.cacheInstruments) { return @@ -730,7 +627,7 @@ export class DiagnosticsMetrics { try { stats = cache.getStats() } catch (error) { - console.error(`DiagnosticsMetrics: failed to read stats for cache '${name}':`, error) + console.error('DiagnosticsMetrics: failed to read stats for cache', name, error) continue } From 13a1773b8ed22d0306278c7f6ce415c0c5ab8dc2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 15:43:21 -0300 Subject: [PATCH 3/8] refactor(metrics): reuse MetricsAccumulator's GetStats, cut comments further MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two things from review: 1. TrackedCache was a byte-for-byte copy of MetricsAccumulator's GetStats interface. Exported GetStats instead and made TrackedCache a type alias for it, removing the duplicate declaration. Checked for other reuse opportunities across the repo — grepped for any existing instrument- registry/getOrCreate utility and for any other code already using getProvider()/createObservableGauge/createObservableCounter/ addBatchObservableCallback outside this file. Found none; this is genuinely new capability, not a rewrite of something that existed. 2. Every private method had gained its own JSDoc block. The file's own existing private methods (getBaseAttributes, createLatencyHistogram, mergeAttributes) carry zero to one inline comment, not a doc block — matched that. Public methods keep a short comment, not the @param/@example treatment the three original public methods have; that fuller style makes sense for the three flagship methods, less so repeated across eight more. Also fixed a comment that overclaimed precedent: it said this reaches the MeterProvider "the same access node-vtex-api uses for HostMetricsInstrumentation", but that call site uses .provider() (a different, wider type), not .getProvider(). Same underlying object at runtime, different method — worth being accurate about rather than implying an existing pattern that isn't quite there. Caught along the way: an editing artifact left an unterminated /** in getObservableMeter's comment (a leftover open token from an earlier trim that never got its matching close deleted). Compiled and tested clean regardless — worth flagging since "tsc passed" doesn't rule out this class of mistake if the malformed comment happens to swallow only comment-like content, but it's exactly the kind of thing worth double-checking by hand after several consecutive edits, which is how this was caught. DiagnosticsMetrics.ts's diff against master: 379 -> 276 -> 227 lines across the three rounds of feedback on this branch. Verified: tsc clean, tslint at baseline, 48/48 in DiagnosticsMetrics.test.ts, 220/220 across the full repo suite. Co-Authored-By: Claude Sonnet 5 --- src/metrics/DiagnosticsMetrics.ts | 101 ++++++++---------------------- src/metrics/MetricsAccumulator.ts | 2 +- 2 files changed, 27 insertions(+), 76 deletions(-) diff --git a/src/metrics/DiagnosticsMetrics.ts b/src/metrics/DiagnosticsMetrics.ts index 2094f7114..20bc6b049 100644 --- a/src/metrics/DiagnosticsMetrics.ts +++ b/src/metrics/DiagnosticsMetrics.ts @@ -13,6 +13,7 @@ import { import { Types } from '@vtex/diagnostics-nodejs' import { getMetricClient } from '../service/metrics/client' import { METRIC_CLIENT_INIT_TIMEOUT_MS, LINKED } from '../constants' +import { GetStats } from './MetricsAccumulator' /** * Maximum number of custom attributes allowed per metric call to control cardinality. @@ -29,8 +30,6 @@ const MAX_CUSTOM_ATTRIBUTES = 7 */ const BASE_ATTRIBUTES_KEY = createContextKey('vtex.metrics.baseAttributes') -// Meter for observable (pull-based) instruments — read by the SDK on its own -// collection schedule, unlike the push-based ones above. const OBSERVABLE_METER_NAME = 'node-vtex-api' // trackCache() metric names — one shared instrument set, differentiated by `cache`. @@ -39,19 +38,12 @@ 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' -/** - * The stats shape trackCache() reads — matches LRUCache/DiskCache/LRUDiskCache/ - * MultilayeredCache's getStats() (see `../caches` and `GetStats` in - * MetricsAccumulator.ts). `hits`/`total`/`disposedItems` reset on every read; - * `itemCount`/`max` don't. Fields a cache doesn't have are just not reported. - */ -export interface TrackedCache { - getStats(): { [key: string]: number | boolean | string | undefined } -} +// Same shape as MetricsAccumulator's cache instances (LRUCache, DiskCache, etc.), +// so trackCache() accepts what apps already have. +export type TrackedCache = GetStats -// ObservableGauge and ObservableCounter are the same type in the OTel API -// (Observable), so registerObservableGauge/Counter share one implementation below, -// keyed by which kind of instrument to create. +// 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 @@ -156,14 +148,12 @@ export class DiagnosticsMetrics { private counters: Map private gauges: Map - // Observable (pull-based) instruments: what apps asked to register, and what's - // actually attached to an OTel instrument (empty until the client is ready — see - // syncObservables). Re-registering a name replaces its callback. + // What apps registered, and what's actually attached to an OTel instrument + // (empty until the client is ready — see syncObservables), keyed by name. private observableRegistrations: Record> private observableInstruments: Record> - // trackCache(): registered caches, their running cumulative totals (getStats() - // resets on read — see observeCaches), and the shared instruments, created once. + // trackCache() state: registered caches, running cumulative totals, shared instruments. private cacheRegistry: Map private cacheCumulative: Map private cacheInstruments: { @@ -449,38 +439,20 @@ export class DiagnosticsMetrics { this.gauges.get(name)!.set(value, mergedAttributes) } - /** - * Get the meter used for observable instruments, if the metrics client is ready. - * Reaches the OpenTelemetry MeterProvider through `getProvider()`, which is already - * part of the metrics client's public surface (the same access node-vtex-api uses - * for HostMetricsInstrumentation in service/telemetry/client.ts). - */ - // Reaches the OTel MeterProvider via getProvider(), already part of the metrics - // client's type — the same access node-vtex-api uses for HostMetricsInstrumentation - // in service/telemetry/client.ts. + // getProvider() is already part of the metrics client's type. private getObservableMeter(): Meter | undefined { return this.metricsClient?.getProvider().getMeter(OBSERVABLE_METER_NAME) } - /** - * Register (or replace, if `name` is already registered) a pull-based gauge: OTel - * calls `observe` on its own collection schedule and expects `result.observe(value, - * attributes?)`. Unlike the push-based methods above, base attributes are not - * merged — there's no request in progress when this runs, so `observe` must supply - * whatever attributes it needs. - * - * @returns A disposer that detaches the callback. Safe to call more than once, and - * safe to call before the metrics client is ready. - */ + // 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. public registerObservableGauge(name: string, observe: ObservableCallback, options?: MetricOptions): () => void { return this.registerObservable('gauge', name, observe, options) } - /** - * Same as `registerObservableGauge`, but for a monotonically increasing total: - * `observe` must report the current cumulative value (the SDK derives the delta), - * the same way `io_app_cache_operations_total` does below. - */ + // 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) } @@ -502,17 +474,12 @@ export class DiagnosticsMetrics { } } - // Attaches every `kind` registration to its instrument. Safe to call anytime — - // a no-op if the client isn't ready yet, or if nothing changed since last call. - // This is what makes registering before the client is ready work: the caller - // gets its disposer immediately, and the actual OTel wiring happens here, once, - // whenever the meter becomes available (see flushPendingObservables). + // 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 { - // Checked before getObservableMeter(): an app that never registers anything of - // this kind must never touch getProvider(), or the "inert unless used" guarantee - // breaks for it. if (this.observableRegistrations[kind].size === 0) { - return + return // skip getObservableMeter() entirely if this kind is unused } const meter = this.getObservableMeter() @@ -538,8 +505,8 @@ export class DiagnosticsMetrics { } } - // Attaches whatever was registered before the client was ready. No-op for an app - // that never calls registerObservableGauge/Counter/trackCache. + // 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') @@ -549,21 +516,9 @@ export class DiagnosticsMetrics { } } - /** - * Replacement for the legacy `MetricsAccumulator.trackCache()`, taking the same - * cache instances (LRUCache, DiskCache, LRUDiskCache, MultilayeredCache — see - * `../caches`). Reads `getStats()` once per OTel collection cycle, not on any - * schedule of its own, and folds the delta into a running total (see - * `observeCaches`) since `hits`/`total`/`disposedItems` reset on every read. - * - * Migrate a cache by replacing the legacy `trackCache` call, not adding this one - * alongside it — reading the same cache from both would split its counts between - * them. Emits `io_app_cache_operations_total`, `_items_current`, `_capacity` and - * `_disposed_total` (see METRICS_CATALOG.md); `hitRate` is not republished, derive - * it from `_operations_total` instead. - * - * @returns A disposer that stops observing this cache. Safe to call more than once. - */ + // Replacement for the legacy MetricsAccumulator.trackCache() — same cache instances, + // see METRICS_CATALOG.md for the metrics emitted. Replace the legacy call, don't + // add this alongside it: getStats() resets on read, so reading twice splits the count. public trackCache(name: string, cacheInstance: TrackedCache): () => void { this.cacheRegistry.set(name, cacheInstance) this.ensureCacheInstruments() @@ -574,9 +529,6 @@ export class DiagnosticsMetrics { } } - // Lazily creates the shared cache instruments and the one batch callback that - // reads every registered cache per cycle. Idempotent — called from trackCache() - // and, in case the client wasn't ready yet, from flushPendingObservables(). private ensureCacheInstruments(): void { if (this.cacheInstruments) { return @@ -612,9 +564,8 @@ export class DiagnosticsMetrics { this.cacheInstruments = { operations, items, capacity, disposed } } - // The one callback backing every cache's instruments — reads each cache exactly - // once per cycle (never twice, never split across two callbacks) and turns its - // delta-on-read stats into a running cumulative total before observing it. + // Reads each cache once per cycle and turns its delta-on-read stats into a + // running cumulative total (an ObservableCounter must report the total, not a delta). private observeCaches(result: BatchObservableResult): void { if (!this.cacheInstruments) { return diff --git a/src/metrics/MetricsAccumulator.ts b/src/metrics/MetricsAccumulator.ts index 9d42bb358..c57a64146 100644 --- a/src/metrics/MetricsAccumulator.ts +++ b/src/metrics/MetricsAccumulator.ts @@ -27,7 +27,7 @@ interface Aggregate { type AggregateMetric = EnvMetric & Aggregate -interface GetStats { +export interface GetStats { getStats(): { [key: string]: number | boolean | string | undefined, } From 8e91e4af41aa9c69c01f2565f9480d77f874135e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 16:41:06 -0300 Subject: [PATCH 4/8] feat(caches): add getCumulativeStats(), a side-effect-free read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit getStats() reports a per-flush window and consumes it on read — a legacy of MetricsAccumulator publishing metrics as log lines. That makes the read mutating, so two readers of the same cache split its counts between them. Internal counters are now plain monotonic totals and getStats() computes its own delta against a snapshot of what it last reported, so its contract is unchanged: same windowed values, same hitRate. getCumulativeStats() reports the lifetime total and never resets, letting an observable reader and the legacy flush read the same cache independently. Also adds the first tests for these four classes, covering both contracts. Co-Authored-By: Claude Opus 5 --- src/caches/DiskCache.ts | 15 ++-- src/caches/LRUCache.ts | 27 +++++-- src/caches/LRUDiskCache.ts | 26 ++++-- src/caches/MultilayeredCache.ts | 22 ++--- src/caches/cacheStats.test.ts | 139 ++++++++++++++++++++++++++++++++ src/caches/typings.ts | 10 +++ 6 files changed, 209 insertions(+), 30 deletions(-) create mode 100644 src/caches/cacheStats.test.ts 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, From e8b9584a1be1f81a20b081cf797236a2b7b3af3e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 16:41:06 -0300 Subject: [PATCH 5/8] feat(metrics): read caches cumulatively, and harden the observable registration trackCache() now reads getCumulativeStats() instead of getStats(). This drops the running-total bookkeeping that existed only to undo the reset-on-read, and removes the footgun it left behind: migrating a cache no longer requires *replacing* the legacy call, so the two can run side by side while the migration is validated. Three fixes to the observable instruments: - A name already registered as the other kind is refused with an error log. The SDK accepts two same-named streams of different types silently and the collector then rejects them. - Attributes reported by an observable callback are held to the same MAX_CUSTOM_ATTRIBUTES limit as the push-based methods. - The operations counter is skipped for an object with no hit/total counters, instead of emitting two zero-valued series. Reverts the now-unneeded export of MetricsAccumulator's GetStats. Tests: the production delta temporality, a cache registered before the client is ready, a real cache shared with the legacy flush, and the inert-if-unused guarantee (spied before construction, so it covers initialization too). Co-Authored-By: Claude Opus 5 --- src/metrics/DiagnosticsMetrics.test.ts | 213 +++++++++++++++++++++---- src/metrics/DiagnosticsMetrics.ts | 93 +++++++---- src/metrics/MetricsAccumulator.ts | 2 +- 3 files changed, 243 insertions(+), 65 deletions(-) diff --git a/src/metrics/DiagnosticsMetrics.test.ts b/src/metrics/DiagnosticsMetrics.test.ts index 74df98f4f..f950dc76a 100644 --- a/src/metrics/DiagnosticsMetrics.test.ts +++ b/src/metrics/DiagnosticsMetrics.test.ts @@ -1,7 +1,9 @@ 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 { 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) @@ -795,12 +797,47 @@ describe('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' }) - expect(gaugeInstruments.get('queue_depth_current')!.addCallback).toHaveBeenCalledWith(observe) + + 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', () => { @@ -812,8 +849,26 @@ describe('DiagnosticsMetrics', () => { const instrument = gaugeInstruments.get('queue_depth_current')! expect(observableMeter.createObservableGauge).toHaveBeenCalledTimes(1) - expect(instrument.removeCallback).toHaveBeenCalledWith(first) - expect(instrument.addCallback).toHaveBeenCalledWith(second) + // 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', () => { @@ -831,7 +886,22 @@ describe('DiagnosticsMetrics', () => { observableDiagnostics.registerObservableCounter('jobs_processed_total', observe) expect(observableMeter.createObservableCounter).toHaveBeenCalledWith('jobs_processed_total', undefined) - expect(counterInstruments.get('jobs_processed_total')!.addCallback).toHaveBeenCalledWith(observe) + 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 () => { @@ -849,7 +919,8 @@ describe('DiagnosticsMetrics', () => { resolveClient(observableMetricsClient) await new Promise(resolve => setTimeout(resolve, 10)) - expect(gaugeInstruments.get('startup_queue_depth')!.addCallback).toHaveBeenCalledWith(observe) + 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 () => { @@ -871,23 +942,21 @@ describe('DiagnosticsMetrics', () => { describe('trackCache', () => { // Exercised against a real MeterProvider + MetricReader instead of a mock: the - // single-read-per-cycle and delta-to-cumulative accounting are easy to get wrong - // in a way a mock would still pass. + // 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 - interface CacheStatsFixture { - [key: string]: number | boolean | string | undefined - } type CollectionResult = Awaited> - function fakeCache(sequence: CacheStatsFixture[]): TrackedCache { + // Values are cumulative, matching what a real cache's getCumulativeStats() reports. + function fakeCache(sequence: Partial[]): TrackedCache { let call = 0 return { - getStats: () => sequence[Math.min(call++, sequence.length - 1)], + getCumulativeStats: () => sequence[Math.min(call++, sequence.length - 1)] as CumulativeStats, } } @@ -915,17 +984,21 @@ describe('DiagnosticsMetrics', () => { return result.resourceMetrics.scopeMetrics.flatMap(sm => sm.metrics.map(m => m.descriptor.name)) } - beforeEach(async () => { - exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE) + function buildClient(temporality: AggregationTemporality) { + exporter = new InMemoryMetricExporter(temporality) reader = new PeriodicExportingMetricReader({ exporter, exportIntervalMillis: 1000000 }) // never fires; collect() is manual provider = new MeterProvider({ readers: [reader] }) - cacheMetricsClient = { - createHistogram: jest.fn(), + 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() @@ -945,28 +1018,67 @@ describe('DiagnosticsMetrics', () => { expect(ops).toContainEqual({ value: 2, attributes: { cache: 'pages', cache_state: 'miss' } }) }) - it('accumulates across collection cycles instead of treating each delta as the total', async () => { + it('reports a monotonic total across collection cycles', async () => { cacheDiagnostics.trackCache('pages', fakeCache([ { hits: 3, total: 5 }, - { hits: 2, total: 2 }, + { hits: 5, total: 7 }, ])) await reader.collect() const second = await reader.collect() const ops = dataPointsIn(second, 'io_app_cache_operations_total') - // cumulative hits: 3 + 2 = 5; cumulative misses: (5-3) + (2-2) = 2 + 0 = 2 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 getStats = jest.fn().mockReturnValue({ hits: 1, total: 1, itemCount: 10, max: 100, disposedItems: 1 }) - cacheDiagnostics.trackCache('pages', { getStats }) + const getCumulativeStats = jest.fn().mockReturnValue({ hits: 1, total: 1, itemCount: 10, max: 100, disposedItems: 1 }) + cacheDiagnostics.trackCache('pages', { getCumulativeStats }) await reader.collect() - expect(getStats).toHaveBeenCalledTimes(1) + expect(getCumulativeStats).toHaveBeenCalledTimes(1) }) it('reports itemCount, max and disposedItems only for caches that expose them', async () => { @@ -986,6 +1098,34 @@ describe('DiagnosticsMetrics', () => { ]) }) + 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 }])) @@ -995,25 +1135,38 @@ describe('DiagnosticsMetrics', () => { expect(dataPointsIn(result, 'io_app_cache_operations_total')).toHaveLength(0) }) - it('does not publish hitRate', async () => { - cacheDiagnostics.trackCache('pages', fakeCache([{ hits: 3, total: 5, hitRate: 0.6 }])) + 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('does not create the meter for apps that never call trackCache', () => { + 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) - cacheDiagnostics.incrementCounter('unrelated_total', 1) + 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 getStats() throws, without dropping other caches', async () => { + it('recovers from a cache whose read throws, without dropping other caches', async () => { const throwingCache: TrackedCache = { - getStats: () => { + getCumulativeStats: () => { throw new Error('boom') }, } @@ -1025,8 +1178,8 @@ describe('DiagnosticsMetrics', () => { const result = await reader.collect() expect(dataPointsIn(result, 'io_app_cache_operations_total')).toContainEqual({ - value: 1, attributes: { cache: 'pages', cache_state: 'hit' }, + value: 1, }) expect(errorSpy).toHaveBeenCalled() diff --git a/src/metrics/DiagnosticsMetrics.ts b/src/metrics/DiagnosticsMetrics.ts index 20bc6b049..88e46ae69 100644 --- a/src/metrics/DiagnosticsMetrics.ts +++ b/src/metrics/DiagnosticsMetrics.ts @@ -9,11 +9,12 @@ import { 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' -import { GetStats } from './MetricsAccumulator' /** * Maximum number of custom attributes allowed per metric call to control cardinality. @@ -38,9 +39,11 @@ 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' -// Same shape as MetricsAccumulator's cache instances (LRUCache, DiskCache, etc.), -// so trackCache() accepts what apps already have. -export type TrackedCache = GetStats +// 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. @@ -50,6 +53,14 @@ interface ObservableRegistration { 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. */ @@ -86,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 @@ -151,11 +173,10 @@ export class DiagnosticsMetrics { // What apps registered, and what's actually attached to an OTel instrument // (empty until the client is ready — see syncObservables), keyed by name. private observableRegistrations: Record> - private observableInstruments: Record> + private observableInstruments: Record> - // trackCache() state: registered caches, running cumulative totals, shared instruments. + // trackCache() state: registered caches and the shared instruments. private cacheRegistry: Map - private cacheCumulative: Map private cacheInstruments: { operations: ObservableCounter items: ObservableGauge @@ -169,7 +190,6 @@ export class DiagnosticsMetrics { this.observableRegistrations = { gauge: new Map(), counter: new Map() } this.observableInstruments = { gauge: new Map(), counter: new Map() } this.cacheRegistry = new Map() - this.cacheCumulative = new Map() this.initMetricClient() } @@ -446,7 +466,8 @@ export class DiagnosticsMetrics { // 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. + // 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) } @@ -458,6 +479,18 @@ export class DiagnosticsMetrics { } 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) @@ -468,7 +501,7 @@ export class DiagnosticsMetrics { const active = this.observableInstruments[kind].get(name) if (active?.callback === observe) { - active.instrument.removeCallback(observe) + active.instrument.removeCallback(active.attached) this.observableInstruments[kind].delete(name) } } @@ -494,14 +527,15 @@ export class DiagnosticsMetrics { } if (active) { - active.instrument.removeCallback(active.callback) + active.instrument.removeCallback(active.attached) } const instrument: Observable = active?.instrument ?? (kind === 'gauge' ? meter.createObservableGauge(name, options) : meter.createObservableCounter(name, options)) - instrument.addCallback(observe) - this.observableInstruments[kind].set(name, { instrument, callback: observe }) + const attached: ObservableCallback = (result) => observe(limitObservableResult(result)) + instrument.addCallback(attached) + this.observableInstruments[kind].set(name, { instrument, callback: observe, attached }) } } @@ -517,15 +551,14 @@ export class DiagnosticsMetrics { } // Replacement for the legacy MetricsAccumulator.trackCache() — same cache instances, - // see METRICS_CATALOG.md for the metrics emitted. Replace the legacy call, don't - // add this alongside it: getStats() resets on read, so reading twice splits the count. + // 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) - this.cacheCumulative.delete(name) } } @@ -548,7 +581,7 @@ export class DiagnosticsMetrics { unit: '1', }) const capacity = meter.createObservableGauge(CACHE_CAPACITY_METRIC, { - description: 'Maximum number of items a VTEX IO app cache can hold', + 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, { @@ -564,8 +597,7 @@ export class DiagnosticsMetrics { this.cacheInstruments = { operations, items, capacity, disposed } } - // Reads each cache once per cycle and turns its delta-on-read stats into a - // running cumulative total (an ObservableCounter must report the total, not a delta). + // Reads each registered cache's cumulative counters straight into the instruments. private observeCaches(result: BatchObservableResult): void { if (!this.cacheInstruments) { return @@ -574,23 +606,19 @@ export class DiagnosticsMetrics { const { operations, items, capacity, disposed } = this.cacheInstruments for (const [name, cache] of this.cacheRegistry) { - let stats: { [key: string]: number | boolean | string | undefined } + let stats: CumulativeStats try { - stats = cache.getStats() + stats = cache.getCumulativeStats() } catch (error) { console.error('DiagnosticsMetrics: failed to read stats for cache', name, error) continue } - const running = this.cacheCumulative.get(name) ?? { hits: 0, misses: 0, disposed: 0 } - const hits = typeof stats.hits === 'number' ? stats.hits : 0 - const total = typeof stats.total === 'number' ? stats.total : 0 - running.hits += hits - running.misses += Math.max(total - hits, 0) - const attributes = { cache: name } - result.observe(operations, running.hits, { ...attributes, cache_state: 'hit' }) - result.observe(operations, running.misses, { ...attributes, cache_state: 'miss' }) + 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) @@ -601,11 +629,8 @@ export class DiagnosticsMetrics { } if (typeof stats.disposedItems === 'number') { - running.disposed += stats.disposedItems - result.observe(disposed, running.disposed, attributes) + result.observe(disposed, stats.disposedItems, attributes) } - - this.cacheCumulative.set(name, running) } } } diff --git a/src/metrics/MetricsAccumulator.ts b/src/metrics/MetricsAccumulator.ts index c57a64146..9d42bb358 100644 --- a/src/metrics/MetricsAccumulator.ts +++ b/src/metrics/MetricsAccumulator.ts @@ -27,7 +27,7 @@ interface Aggregate { type AggregateMetric = EnvMetric & Aggregate -export interface GetStats { +interface GetStats { getStats(): { [key: string]: number | boolean | string | undefined, } From 614c0910381bd71cfab1493a8b86c1912171b9b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 16:41:06 -0300 Subject: [PATCH 6/8] docs(metrics): add Pattern 6 for addOnFlushMetric, correct Pattern 5 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pattern 5 said to replace the legacy trackCache call rather than add alongside it. That is no longer true, and running both is now the recommended way to validate a migration. Pattern 6 maps addOnFlushMetric onto registerObservableGauge/Counter — the other legacy API with no documented replacement. The translation is not mechanical: a flush metric returns one object with arbitrarily many fields, an instrument reports one value, so the pattern covers both shapes. Also notes that io_app_cache_capacity is in the cache's own units, and declares @opentelemetry/sdk-metrics and context-async-hooks, which the tests import but which resolved only as hoisted transitive deps. Co-Authored-By: Claude Opus 5 --- docs/METRICS_CATALOG.md | 20 +++++++++------- docs/METRICS_OVERVIEW.md | 52 ++++++++++++++++++++++++++++++++++++---- package.json | 2 ++ 3 files changed, 61 insertions(+), 13 deletions(-) diff --git a/docs/METRICS_CATALOG.md b/docs/METRICS_CATALOG.md index be6414cb5..b62bee306 100644 --- a/docs/METRICS_CATALOG.md +++ b/docs/METRICS_CATALOG.md @@ -97,9 +97,9 @@ All Metrics in node-vtex-api │ │ │ └── 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) - only if getStats() has itemCount -│ ├── io_app_cache_capacity (Observable Gauge) - only if getStats() has max -│ └── io_app_cache_disposed_total (Observable Counter) - only if getStats() has disposedItems +│ ├── 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) │ @@ -155,7 +155,7 @@ All Metrics in node-vtex-api │ │ ├── httpAgent - sockets, freeSockets, pendingRequests │ │ └── incomingRequest - total, closed, aborted │ │ - │ └── Cache Metrics (via trackCache — replacement available, see Diagnostics Cache Metrics above) + │ └── Cache Metrics (via trackCache — replacement above; safe to run both while migrating) │ └── {cache_name}-cache │ ├── LRU: itemCount, length, disposedItems, hitRate, hits, max, total │ ├── Disk: hits, total @@ -254,15 +254,19 @@ These are operation-specific metrics recorded in middleware components. 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()` during a migration without either reader consuming the other's counts. + | Metric Name | Type | Attributes | Reported when | |-------------|------|------------|----------------| -| `io_app_cache_operations_total` | Observable Counter | `cache`, `cache_state` (`hit` \| `miss`) | Always | -| `io_app_cache_items_current` | Observable Gauge | `cache` | Cache's `getStats()` returns `itemCount` | -| `io_app_cache_capacity` | Observable Gauge | `cache` | Cache's `getStats()` returns `max` | -| `io_app_cache_disposed_total` | Observable Counter | `cache` | Cache's `getStats()` returns `disposedItems` | +| `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) ``` diff --git a/docs/METRICS_OVERVIEW.md b/docs/METRICS_OVERVIEW.md index cbbb34387..5f96e84b5 100644 --- a/docs/METRICS_OVERVIEW.md +++ b/docs/METRICS_OVERVIEW.md @@ -189,20 +189,62 @@ metrics.trackCache('pages', pagesCacheStorage) 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 `getStats()` exactly once per collection cycle no matter how many metrics it produces from that one cache, which matters because `hits`/`total`/`disposedItems` reset on every read: reading the same cache from two places (e.g. the legacy `trackCache` and a hand-rolled `incrementCounter` call) would split its counts between them. Migrate a cache by **replacing** the legacy `metrics.trackCache(...)` call, not by adding this alongside it. +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. + +**You can leave the legacy call in place while you validate.** The legacy `getStats()` reports a per-flush window and consumes it on read; `getCumulativeStats()` reports the process-lifetime total and has no side effects. The two readers don't steal counts from each other, so running them side by side and comparing is the recommended way to migrate: + +```typescript +metrics.trackCache('pages', pagesCacheStorage) // keep during validation +global.diagnosticsMetrics?.trackCache('pages', pagesCacheStorage) // add, compare, then drop the line above +``` 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. -If you have a periodic value to report that isn't a cache — a queue depth, a connection pool size, anything read on a schedule rather than pushed per-request — use the lower-level `registerObservableGauge`/`registerObservableCounter` that `trackCache` is built on: +### 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 -const dispose = global.diagnosticsMetrics?.registerObservableGauge( - 'queue_depth_current', - (result) => result.observe(queue.length), +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 829761a1e..c742b994d 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", From 082bb9ec6ebfb4912208255e228d4dcfc031bc23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 18:05:16 -0300 Subject: [PATCH 7/8] fix(metrics): mark never-reassigned private maps as readonly Sonar S2933 on the three fields this PR adds. The two pre-existing fields right above them (counters, gauges) have the same issue and are marked too, so the block stays consistent. Co-Authored-By: Claude Opus 5 --- src/metrics/DiagnosticsMetrics.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/metrics/DiagnosticsMetrics.ts b/src/metrics/DiagnosticsMetrics.ts index 88e46ae69..9ec9197b6 100644 --- a/src/metrics/DiagnosticsMetrics.ts +++ b/src/metrics/DiagnosticsMetrics.ts @@ -167,16 +167,16 @@ 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 observableRegistrations: Record> - private observableInstruments: Record> + private readonly observableRegistrations: Record> + private readonly observableInstruments: Record> // trackCache() state: registered caches and the shared instruments. - private cacheRegistry: Map + private readonly cacheRegistry: Map private cacheInstruments: { operations: ObservableCounter items: ObservableGauge From ce8b5a7cd6e3d3b39f030d148a973aa4f25abb27 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Vin=C3=ADcius=20Seixas?= Date: Mon, 14 Sep 2026 18:09:34 -0300 Subject: [PATCH 8/8] docs(metrics): correct why the legacy trackCache call is safe to keep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pattern 5 told the reader to keep the legacy call "during validation" and compare the two metrics. There is nothing to compare: since #676 (v7.4.2) nothing consumes what the legacy flush returns — statusTrack() keeps running only for the reset side effects. What the side-effect-free read actually buys is that a half-migrated app still reports correct numbers, which matters across 23 call sites in three apps. 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. Co-Authored-By: Claude Opus 5 --- docs/METRICS_CATALOG.md | 4 ++-- docs/METRICS_OVERVIEW.md | 8 +++++--- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/docs/METRICS_CATALOG.md b/docs/METRICS_CATALOG.md index b62bee306..08bdbd2e1 100644 --- a/docs/METRICS_CATALOG.md +++ b/docs/METRICS_CATALOG.md @@ -155,7 +155,7 @@ All Metrics in node-vtex-api │ │ ├── httpAgent - sockets, freeSockets, pendingRequests │ │ └── incomingRequest - total, closed, aborted │ │ - │ └── Cache Metrics (via trackCache — replacement above; safe to run both while migrating) + │ └── Cache Metrics (via trackCache — output discarded since #676; replacement above) │ └── {cache_name}-cache │ ├── LRU: itemCount, length, disposedItems, hitRate, hits, max, total │ ├── Disk: hits, total @@ -254,7 +254,7 @@ These are operation-specific metrics recorded in middleware components. 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()` during a migration without either reader consuming the other's counts. +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 | |-------------|------|------------|----------------| diff --git a/docs/METRICS_OVERVIEW.md b/docs/METRICS_OVERVIEW.md index 5f96e84b5..4155d1244 100644 --- a/docs/METRICS_OVERVIEW.md +++ b/docs/METRICS_OVERVIEW.md @@ -191,13 +191,15 @@ 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. -**You can leave the legacy call in place while you validate.** The legacy `getStats()` reports a per-flush window and consumes it on read; `getCumulativeStats()` reports the process-lifetime total and has no side effects. The two readers don't steal counts from each other, so running them side by side and comparing is the recommended way to migrate: +**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) // keep during validation -global.diagnosticsMetrics?.trackCache('pages', pagesCacheStorage) // add, compare, then drop the line above +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`)