diff --git a/.changeset/wild-otters-collect-metrics.md b/.changeset/wild-otters-collect-metrics.md new file mode 100644 index 0000000000..bea8d2fa5b --- /dev/null +++ b/.changeset/wild-otters-collect-metrics.md @@ -0,0 +1,5 @@ +--- +'posthog-node': minor +--- + +Proof of concept: autocapture low-level Node runtime metrics (CPU, memory, event loop delay and utilization, GC pauses, uptime, active handles) through `posthog.metrics`, with no instrumentation. Off unless the `metrics-sdk-autocapture` feature flag opens the gate (evaluated locally only, so it costs no request and no event) or `enableMetricsAutocapture: true` is set. diff --git a/packages/node/references/posthog-node-references-latest.json b/packages/node/references/posthog-node-references-latest.json index c353c084ed..c96cb2b507 100644 --- a/packages/node/references/posthog-node-references-latest.json +++ b/packages/node/references/posthog-node-references-latest.json @@ -3456,6 +3456,16 @@ "type": "MetricsConfig", "name": "metrics" }, + { + "description": "PROOF OF CONCEPT - MAY CHANGE WITHOUT WARNING\n\nAutocapture low-level Node runtime metrics (CPU time and utilization, memory\nand heap limit, event loop delay and utilization, GC pauses, uptime, active\nhandles) through `posthog.metrics`, with no instrumentation of your own.\n\nLeave this unset to let the `metrics-sdk-autocapture` feature flag decide,\nre-evaluated every 30 seconds so it works as a remote kill switch. The flag\nis only ever evaluated locally, against cached flag definitions — so it\nrequires `secretKey`, must itself be locally evaluable, and never costs a\nrequest or a `$feature_flag_called` event. Set this option explicitly to opt\nin or out and skip flag evaluation entirely.", + "type": "boolean", + "name": "enableMetricsAutocapture" + }, + { + "description": "PROOF OF CONCEPT - MAY CHANGE WITHOUT WARNING\n\nHow often runtime metrics are sampled when autocapture is on. Values below\n1000ms are clamped, and the default matches the metrics flush interval so\neach flush window carries roughly one sample per series.", + "type": "number", + "name": "metricsAutocaptureIntervalMs" + }, { "description": "Credential that enables local feature flag evaluation and remote config.\n\nAccepts either a Personal API Key (`phx_...`) or a Project Secret API Key (`phs_...`).\nWhen provided, the client can evaluate feature flags locally and decrypt remote\nconfig payloads via `getRemoteConfigPayload`. Prefer this over the deprecated\n`personalApiKey` option; when both are set, `secretKey` takes precedence.", "type": "string", diff --git a/packages/node/src/__tests__/metrics-autocapture.spec.ts b/packages/node/src/__tests__/metrics-autocapture.spec.ts new file mode 100644 index 0000000000..c78481936f --- /dev/null +++ b/packages/node/src/__tests__/metrics-autocapture.spec.ts @@ -0,0 +1,290 @@ +import { PostHog } from '@/entrypoints/index.node' +import type { PostHogOptions } from '@/types' +import { + DEFAULT_SAMPLE_INTERVAL_MS, + GATE_POLL_INTERVAL_MS, + METRICS_AUTOCAPTURE_FLAG, +} from '@/extensions/metrics-autocapture' +import { parseCgroupCpuQuota } from '@/extensions/metrics-autocapture/runtime.node' +import { waitForPromises } from './utils' + +jest.mock('../version', () => ({ version: '1.2.3' })) +jest.spyOn(console, 'debug').mockImplementation() + +const mockedFetch = jest.spyOn(globalThis, 'fetch').mockImplementation() + +const options: PostHogOptions = { + host: 'http://example.com', + disableCompression: true, + fetchRetryCount: 0, + featureFlagsRequestMaxRetries: 0, + metrics: { serviceName: 'test-service' }, +} + +/** + * Serves flag definitions for local evaluation with the gate flag rolled out to + * either everyone or nobody, plus 200s for every write endpoint. + */ +const mockApi = (gateEnabled: boolean): void => { + mockedFetch.mockImplementation((url: any): Promise => { + if (String(url).includes('flags/definitions')) { + return Promise.resolve({ + status: 200, + text: () => Promise.resolve('ok'), + headers: { get: () => null }, + json: () => + Promise.resolve({ + flags: [ + { + id: 1, + name: 'Metrics SDK autocapture', + key: METRICS_AUTOCAPTURE_FLAG, + active: true, + filters: { groups: [{ rollout_percentage: gateEnabled ? 100 : 0 }] }, + }, + ], + group_type_mapping: {}, + cohorts: {}, + }), + }) + } + return Promise.resolve({ status: 200, text: () => Promise.resolve('ok') }) + }) +} + +const localEvaluationOptions: PostHogOptions = { ...options, secretKey: 'phx_test' } + +const flagsCalls = (): any[] => mockedFetch.mock.calls.filter((call) => String(call[0]).includes('/flags/?')) + +const metricNames = (): string[] => { + const names = new Set() + for (const call of mockedFetch.mock.calls) { + if (!String(call[0]).includes('/i/v1/metrics')) { + continue + } + const body = JSON.parse((call[1] as any).body) + for (const metric of body.resourceMetrics[0].scopeMetrics[0].metrics) { + names.add(metric.name) + } + } + return [...names] +} + +describe('PostHog Node.js metrics autocapture', () => { + let posthog: PostHog + + jest.useFakeTimers() + + beforeEach(() => { + mockedFetch.mockReset() + mockApi(true) + }) + + afterEach(async () => { + await posthog?.shutdown() + }) + + describe('when explicitly enabled', () => { + beforeEach(() => { + posthog = new PostHog('TEST_API_KEY', { ...options, enableMetricsAutocapture: true }) + }) + + it('collects runtime metrics on the sample interval without any instrumentation', async () => { + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.metrics.flush() + + const names = metricNames() + expect(names).toEqual( + expect.arrayContaining([ + 'process.cpu.time', + 'process.cpu.utilization', + 'process.memory.usage', + 'process.memory.heap_limit', + 'process.event_loop.delay', + 'process.event_loop.utilization', + 'process.uptime', + 'process.active_resources', + ]) + ) + }) + + it('breaks memory down by type and event loop delay by stat, and nothing else', async () => { + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.metrics.flush() + + const metricsCall = mockedFetch.mock.calls.find((call) => String(call[0]).includes('/i/v1/metrics'))! + const metrics = JSON.parse((metricsCall[1] as any).body).resourceMetrics[0].scopeMetrics[0].metrics + const byName = Object.fromEntries(metrics.map((m: any) => [m.name, m])) + + const attributeValues = (metric: any, key: string): string[] => { + const dataPoints: any[] = metric.gauge ? metric.gauge.dataPoints : metric.sum.dataPoints + return dataPoints.map((point) => point.attributes.find((attr: any) => attr.key === key)?.value.stringValue) + } + + expect(attributeValues(byName['process.memory.usage'], 'type').sort()).toEqual([ + 'array_buffers', + 'external', + 'heap_total', + 'heap_used', + 'rss', + ]) + // `mean` is NaN until the event loop monitor has seen a tick, and NaN + // samples are dropped rather than shipped as a bogus data point. + const delayStats = attributeValues(byName['process.event_loop.delay'], 'stat').sort() + expect(delayStats).toEqual(expect.arrayContaining(['max', 'p50', 'p90', 'p99'])) + expect(delayStats.every((stat) => ['mean', 'p50', 'p90', 'p99', 'max'].includes(stat))).toBe(true) + expect(attributeValues(byName['process.cpu.time'], 'state').sort()).toEqual(['system', 'user']) + // No per-host or per-instance attributes — those would be a series per box. + expect(byName['process.uptime'].gauge.dataPoints[0].attributes).toEqual([]) + }) + + it('does not evaluate the gate flag', async () => { + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + + expect(flagsCalls()).toHaveLength(0) + }) + + it('respects a custom sample interval', async () => { + await posthog.shutdown() + posthog = new PostHog('TEST_API_KEY', { + ...options, + enableMetricsAutocapture: true, + metricsAutocaptureIntervalMs: 60000, + }) + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.metrics.flush() + expect(metricNames()).toHaveLength(0) + + await jest.advanceTimersByTimeAsync(60000) + await posthog.metrics.flush() + expect(metricNames()).toContain('process.memory.usage') + }) + + it('stops sampling on shutdown', async () => { + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.shutdown() + mockedFetch.mockClear() + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS * 5) + + expect(metricNames()).toHaveLength(0) + }) + }) + + it('collects nothing when explicitly disabled', async () => { + posthog = new PostHog('TEST_API_KEY', { ...localEvaluationOptions, enableMetricsAutocapture: false }) + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS * 2) + await posthog.metrics.flush() + + expect(metricNames()).toHaveLength(0) + }) + + it('stays off when left to the flag but local evaluation is unavailable', async () => { + // Without the poller the gate would cost a `/flags` request per poll, so it + // stays closed rather than adding an unasked-for request to every client. + posthog = new PostHog('TEST_API_KEY', options) + await waitForPromises() + + await jest.advanceTimersByTimeAsync(GATE_POLL_INTERVAL_MS * 2) + await posthog.metrics.flush() + + expect(metricNames()).toHaveLength(0) + expect(flagsCalls()).toHaveLength(0) + }) + + describe('when left to the feature flag', () => { + it('collects runtime metrics once the flag evaluates to true', async () => { + posthog = new PostHog('TEST_API_KEY', localEvaluationOptions) + await waitForPromises() + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.metrics.flush() + + expect(metricNames()).toContain('process.memory.usage') + // Local evaluation only: the gate never costs a `/flags` request. + expect(flagsCalls()).toHaveLength(0) + }) + + it('does not capture a $feature_flag_called event for its own gate', async () => { + // The SDK evaluating a flag about itself must not bill the user for an + // event, nor attach one to the synthetic gate distinct ID. + posthog = new PostHog('TEST_API_KEY', localEvaluationOptions) + await waitForPromises() + await jest.advanceTimersByTimeAsync(GATE_POLL_INTERVAL_MS) + await posthog.flush() + + expect(mockedFetch.mock.calls.filter((call) => String(call[0]).includes('/batch/'))).toHaveLength(0) + }) + + it('collects nothing while the flag evaluates to false', async () => { + mockApi(false) + posthog = new PostHog('TEST_API_KEY', localEvaluationOptions) + await waitForPromises() + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS * 2) + await posthog.metrics.flush() + + expect(metricNames()).toHaveLength(0) + }) + + it('acts as a kill switch: stops collecting when the flag is turned off', async () => { + posthog = new PostHog('TEST_API_KEY', localEvaluationOptions) + await waitForPromises() + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.metrics.flush() + expect(metricNames()).toContain('process.memory.usage') + + mockApi(false) + // One poll interval refreshes the cached definitions, the next re-evaluates + // the gate against them and closes it. + for (let i = 0; i < 3; i++) { + await jest.advanceTimersByTimeAsync(GATE_POLL_INTERVAL_MS) + await waitForPromises() + } + mockedFetch.mockClear() + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS * 3) + await posthog.metrics.flush() + + expect(metricNames()).toHaveLength(0) + }) + + it('stays off and keeps polling when the definitions load fails', async () => { + mockedFetch.mockRejectedValue(new Error('connection refused')) + posthog = new PostHog('TEST_API_KEY', localEvaluationOptions) + await waitForPromises() + + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS * 2) + expect(metricNames()).toHaveLength(0) + + mockApi(true) + for (let i = 0; i < 3; i++) { + await jest.advanceTimersByTimeAsync(GATE_POLL_INTERVAL_MS) + await waitForPromises() + } + await jest.advanceTimersByTimeAsync(DEFAULT_SAMPLE_INTERVAL_MS) + await posthog.metrics.flush() + + expect(metricNames()).toContain('process.memory.usage') + }) + }) +}) + +describe('cgroup CPU quota parsing', () => { + it.each([ + // A pod limited to 500m: without this the utilization denominator would be + // the host's core count and the ratio would read 100x too low. + [{ cpuMax: '50000 100000' }, 0.5], + [{ cpuMax: '200000 100000\n' }, 2], + // Unlimited, in both cgroup versions — fall back to available parallelism. + [{ cpuMax: 'max 100000' }, undefined], + [{ cfsQuotaUs: '-1', cfsPeriodUs: '100000' }, undefined], + [{ cfsQuotaUs: '150000', cfsPeriodUs: '100000' }, 1.5], + // No cgroup filesystem at all. + [{}, undefined], + ])('parses %j as %s cores', (files, expected) => { + expect(parseCgroupCpuQuota(files)).toBe(expected) + }) +}) diff --git a/packages/node/src/client.ts b/packages/node/src/client.ts index dbd47b1f3d..41c744ae47 100644 --- a/packages/node/src/client.ts +++ b/packages/node/src/client.ts @@ -55,6 +55,8 @@ import { RequiresServerEvaluation, } from './extensions/feature-flags/feature-flags' import ErrorTracking from './extensions/error-tracking' +import MetricsAutocapture from './extensions/metrics-autocapture' +import type { RuntimeMetricsSampler } from './extensions/metrics-autocapture/types' import { PostHogMemoryStorage } from './storage-memory' import { ContextData, ContextOptions, IPostHogContext } from './extensions/context/types' import { type CaptureMode, resolveCaptureMode } from './capture-v1/config' @@ -140,6 +142,7 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen private featureFlagsPoller?: FeatureFlagsPoller protected errorTracking: ErrorTracking + protected metricsAutocapture: MetricsAutocapture private maxCacheSize: number public readonly options: PostHogOptions protected readonly context?: IPostHogContext @@ -266,10 +269,28 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen } this.errorTracking = new ErrorTracking(this, normalizedOptions, this._logger) + this.metricsAutocapture = new MetricsAutocapture( + this, + normalizedOptions, + this._logger, + // Its gate flag is evaluated locally only, so it can't resolve without the poller. + this.featureFlagsPoller !== undefined, + () => this.createRuntimeMetricsSampler() + ) + this.metricsAutocapture.start() this.distinctIdHasSentFlagCalls = {} this.maxCacheSize = normalizedOptions.maxCacheSize || MAX_CACHE_SIZE } + /** + * The runtime metrics sampler used by metrics autocapture, or `undefined` when + * the runtime has no low-level metrics to offer. Overridden by the Node + * entrypoint; the base (and the edge build) opts out. + */ + protected createRuntimeMetricsSampler(): RuntimeMetricsSampler | undefined { + return undefined + } + protected override enqueue( type: string, message: any, @@ -2573,6 +2594,10 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen await this.featureFlagsPoller?.stopPoller(shutdownTimeoutMs) this.errorTracking.shutdown() + // Stopped before the metrics flush below so no sample can land in a window + // that has already been drained, and so the event loop monitor and GC + // observer are torn down even if the flush times out. + this.metricsAutocapture.shutdown() if (this._metrics) { // Send whatever is aggregated in the current window, then clear the flush // timer so it can't fire after teardown. Raced against the shutdown budget: diff --git a/packages/node/src/entrypoints/index.node.ts b/packages/node/src/entrypoints/index.node.ts index 1750a78c30..e5bf1d518a 100644 --- a/packages/node/src/entrypoints/index.node.ts +++ b/packages/node/src/entrypoints/index.node.ts @@ -7,6 +7,7 @@ import { createRelativePathModifier } from '../extensions/error-tracking/modifie import { PostHogBackendClient } from '../client' import { ErrorTracking as CoreErrorTracking } from '@posthog/core' import { PostHogContext } from '../extensions/context/context' +import { RuntimeMetricsCollector } from '../extensions/metrics-autocapture/runtime.node' import { gzipCompress } from '../gzip.node' export class PostHog extends PostHogBackendClient { @@ -22,6 +23,10 @@ export class PostHog extends PostHogBackendClient { return new PostHogContext() } + protected override createRuntimeMetricsSampler(): RuntimeMetricsCollector { + return new RuntimeMetricsCollector() + } + protected override createErrorPropertiesBuilder(): CoreErrorTracking.ErrorPropertiesBuilder { return new CoreErrorTracking.ErrorPropertiesBuilder( [ diff --git a/packages/node/src/extensions/metrics-autocapture/index.ts b/packages/node/src/extensions/metrics-autocapture/index.ts new file mode 100644 index 0000000000..a20b78647c --- /dev/null +++ b/packages/node/src/extensions/metrics-autocapture/index.ts @@ -0,0 +1,270 @@ +import { safeSetTimeout, uuidv7 } from '@posthog/core' +import type { Logger, Metrics } from '@posthog/core' +import type { PostHogBackendClient } from '@/client' +import type { PostHogOptions } from '@/types' +import type { RuntimeMetricsSampler } from './types' +import { version } from '@/version' + +/** + * Feature flag that turns runtime metrics autocapture on when + * `enableMetricsAutocapture` is left unset. + */ +export const METRICS_AUTOCAPTURE_FLAG = 'metrics-sdk-autocapture' + +export const DEFAULT_SAMPLE_INTERVAL_MS = 10000 +const MINIMUM_SAMPLE_INTERVAL_MS = 1000 +/** + * How often the gate flag is re-evaluated. Re-checking is what makes the flag a + * kill switch rather than a boot-time-only decision: turning it off stops a + * running fleet from collecting within one interval, without a deploy. It's + * matched to the flag definition polling interval, and each check is a local + * evaluation against already-cached definitions, so it costs no request. + */ +export const GATE_POLL_INTERVAL_MS = 30000 + +/** + * PROOF OF CONCEPT — autocaptures low-level Node runtime metrics (CPU, memory, + * event loop delay, GC) through the `posthog.metrics` client, with no + * instrumentation from the user. + * + * Three-state gate, mirroring how a server-controlled SDK behaviour has to work + * if the eventual goal is "install the SDK and get metrics": + * + * - `enableMetricsAutocapture: true` — on, no flag evaluation at all. + * - `enableMetricsAutocapture: false` — off, no flag evaluation at all. + * - unset — decided by the `metrics-sdk-autocapture` feature flag, re-evaluated + * every {@link GATE_POLL_INTERVAL_MS} so it doubles as a remote kill switch. + * + * The gate is evaluated **locally only**, against the flag definitions the + * poller already caches (so it needs `secretKey`). No `/flags` request, no + * `$feature_flag_called` event, nothing added to the user's critical path or + * bill for a decision the SDK is making about itself. The consequence is that + * the gate flag has to be locally evaluable — a simple rollout percentage or + * property condition, not a cohort or experience-continuity flag. + * + * The flag is evaluated against a synthetic per-process distinct ID, so a + * percentage rollout buckets *processes* rather than end users, and person + * properties (`$lib`, `$lib_version`, `service_name`, `environment`) are passed + * so the flag can be targeted at one service or SDK version. + */ +export default class MetricsAutocapture { + private readonly _client: PostHogBackendClient + private readonly _options: PostHogOptions + private readonly _logger: Logger + private readonly _localEvaluationEnabled: boolean + private readonly _createSampler: () => RuntimeMetricsSampler | undefined + private readonly _intervalMs: number + + private _sampler?: RuntimeMetricsSampler + private _sampleTimer?: ReturnType + private _gateTimer?: ReturnType + private _gateId?: string + private _sampling = false + private _shutdown = false + private _sampleErrorLogged = false + + constructor( + client: PostHogBackendClient, + options: PostHogOptions, + logger: Logger, + localEvaluationEnabled: boolean, + createSampler: () => RuntimeMetricsSampler | undefined + ) { + this._client = client + this._options = options + this._logger = logger.createLogger('[Metrics autocapture]') + this._localEvaluationEnabled = localEvaluationEnabled + this._createSampler = createSampler + this._intervalMs = Math.max( + options.metricsAutocaptureIntervalMs ?? DEFAULT_SAMPLE_INTERVAL_MS, + MINIMUM_SAMPLE_INTERVAL_MS + ) + } + + /** + * Resolves the gate and starts sampling if it is open. Never throws and never + * blocks: the flag evaluation runs detached so constructing a client stays + * synchronous, and a failed evaluation just leaves autocapture off until the + * next poll. + */ + start(): void { + if (this._options.enableMetricsAutocapture === false || this._client.isDisabled || this._client.optedOut) { + return + } + + if (this._options.enableMetricsAutocapture === true) { + this._startSampling() + return + } + + if (!this._localEvaluationEnabled) { + // Without local evaluation the gate would cost a `/flags` request per + // poll, so it stays closed. Opt in explicitly instead. + this._logger.debug( + `Not evaluating ${METRICS_AUTOCAPTURE_FLAG} because local evaluation is off — ` + + 'pass `enableMetricsAutocapture: true` to collect runtime metrics without it.' + ) + return + } + + void this._pollGate() + } + + /** Whether the gate is currently open and runtime metrics are being sampled. */ + isEnabled(): boolean { + return this._sampling + } + + shutdown(): void { + this._shutdown = true + if (this._gateTimer) { + clearTimeout(this._gateTimer) + this._gateTimer = undefined + } + this._stopSampling() + } + + private async _pollGate(): Promise { + let enabled = false + try { + enabled = await this._evaluateGate() + } catch (err) { + // Evaluation is best effort — nothing here may take down the host process, + // and the next poll gets another go once definitions have loaded. + this._logger.debug(`Could not evaluate ${METRICS_AUTOCAPTURE_FLAG}:`, err) + } + + if (this._shutdown) { + return + } + + if (enabled) { + this._startSampling() + } else { + this._stopSampling() + } + + this._gateTimer = safeSetTimeout(() => void this._pollGate(), GATE_POLL_INTERVAL_MS) + } + + private async _evaluateGate(): Promise { + if (this._client.isDisabled) { + return false + } + + // `onlyEvaluateLocally` keeps this free: it reads the definitions the poller + // already caches instead of issuing a `/flags` request, and returns undefined + // (gate closed) until the first definition load lands. `sendFeatureFlagEvents: + // false` keeps it silent: the SDK evaluating a flag about itself must not bill + // the user for a `$feature_flag_called` event on every poll. + const result = await this._client.getFeatureFlagResult(METRICS_AUTOCAPTURE_FLAG, this._gateDistinctId(), { + onlyEvaluateLocally: true, + sendFeatureFlagEvents: false, + personProperties: this._gatePersonProperties(), + }) + return result?.enabled === true + } + + /** + * Stable for the lifetime of the process and unique per process, so a + * percentage rollout on the flag buckets processes consistently instead of + * flip-flopping on every poll. + * + * Deliberately random rather than derived from hostname/pid: `HOSTNAME` is + * unset outside containers and pids repeat, which would hash a whole fleet + * into the same bucket and make a 10% rollout resolve as 0% or 100%. + */ + private _gateDistinctId(): string { + this._gateId ??= `posthog-node-metrics:${this._options.metrics?.serviceName ?? 'unknown-service'}:${uuidv7()}` + return this._gateId + } + + private _gatePersonProperties(): Record { + const properties: Record = { + $lib: this._client.getLibraryId(), + $lib_version: version, + } + if (this._options.metrics?.serviceName) { + properties.service_name = this._options.metrics.serviceName + } + if (this._options.metrics?.environment) { + properties.environment = this._options.metrics.environment + } + return properties + } + + private _startSampling(): void { + if (this._sampling || this._shutdown) { + return + } + + const sampler = this._createSampler() + if (!sampler) { + // No sampler for this runtime (e.g. the edge build, where `perf_hooks` and + // most of `process` don't exist) — nothing to collect. + return + } + + const metrics = this._client.metrics + try { + sampler.start(metrics) + } catch (err) { + // A partial start may already hold an event loop monitor or a GC observer, + // and this path can be retried on every gate poll — so tear it down rather + // than leaking one set of handles per poll. + try { + sampler.stop() + } catch { + // Nothing useful to do if teardown of a failed start also fails. + } + this._logger.debug('Could not start runtime metrics sampler:', err) + return + } + + this._sampler = sampler + this._sampling = true + this._armSampleTimer(metrics) + this._logger.debug(`Collecting Node runtime metrics every ${this._intervalMs}ms`) + } + + private _armSampleTimer(metrics: Metrics): void { + // Re-armed per tick rather than setInterval so a slow sample can't stack up + // overlapping runs, and so the handle is always the current one to clear. + this._sampleTimer = safeSetTimeout(() => { + this._sampleTimer = undefined + if (this._shutdown || !this._sampler) { + return + } + if (this._client.optedOut) { + // Every sample would be dropped at capture anyway; skip the work but keep + // the timer, so opting back in resumes collection. + this._armSampleTimer(metrics) + return + } + try { + this._sampler.sample(metrics) + } catch (err) { + // One log per client, not one per interval. + if (!this._sampleErrorLogged) { + this._sampleErrorLogged = true + this._logger.warn('Failed to sample Node runtime metrics:', err) + } + } + this._armSampleTimer(metrics) + }, this._intervalMs) + } + + private _stopSampling(): void { + this._sampling = false + if (this._sampleTimer) { + clearTimeout(this._sampleTimer) + this._sampleTimer = undefined + } + try { + this._sampler?.stop() + } catch (err) { + this._logger.debug('Could not stop runtime metrics sampler:', err) + } + this._sampler = undefined + } +} diff --git a/packages/node/src/extensions/metrics-autocapture/runtime.node.ts b/packages/node/src/extensions/metrics-autocapture/runtime.node.ts new file mode 100644 index 0000000000..89d5dbeb19 --- /dev/null +++ b/packages/node/src/extensions/metrics-autocapture/runtime.node.ts @@ -0,0 +1,276 @@ +import { constants, monitorEventLoopDelay, performance, PerformanceObserver } from 'node:perf_hooks' +import { readFileSync } from 'node:fs' +import { availableParallelism } from 'node:os' +import { getHeapStatistics } from 'node:v8' +import type { Metrics } from '@posthog/core' +import type { RuntimeMetricsSampler } from './types' + +/** + * Samples low-level Node runtime metrics into the `posthog.metrics` client. + * + * Everything here comes from APIs the process already has, so there is nothing + * for the user to instrument — the same trade autocapture makes for events. + * + * Two constraints shape these series: + * + * - **Cardinality.** Every attribute combination is its own series, so the only + * attributes used are closed enums (`state`, `type`, `stat`, `kind`). Nothing + * per-host or per-instance is attached; service/environment identity comes from + * the resource attributes on the `metrics` client config. The cost of that + * choice is that gauges from several replicas of one service land on the same + * series and overwrite each other — an instance attribute is the obvious next + * step, and the obvious cardinality trade to argue about first. + * - **Cheap sampling.** Every read is an O(1) process-local counter folded into + * the pre-aggregating metrics client, so a sample is a handful of `gauge()` + * calls rather than any syscall-heavy work. + */ +export class RuntimeMetricsCollector implements RuntimeMetricsSampler { + private _lastCpuUsage?: NodeJS.CpuUsage + private _lastSampleAtMs?: number + private _lastEventLoopUtilization?: ReturnType + private _eventLoopDelay?: ReturnType + private _gcObserver?: PerformanceObserver + private _cpuCount = 1 + + /** + * Starts the collectors that have to be running *between* samples: the event + * loop delay histogram and the GC observer. Everything else is a + * point-in-time read taken during `sample()`. + */ + start(metrics: Metrics): void { + this._cpuCount = detectCpuCount() + this._lastCpuUsage = process.cpuUsage() + this._lastSampleAtMs = Date.now() + this._lastEventLoopUtilization = performance.eventLoopUtilization() + + // 10ms resolution: fine enough to see a blocked loop, coarse enough that the + // libuv timer doing the sampling isn't itself a cost. + this._eventLoopDelay = monitorEventLoopDelay({ resolution: EVENT_LOOP_RESOLUTION_MS }) + this._eventLoopDelay.enable() + + // The only unbounded-rate emission here: a busy process GCs thousands of + // times a second, all folded into one histogram series per kind. Note the + // default bucket bounds start at 5ms, so sub-ms scavenges all land in the + // first bucket — count, sum and max are what carry the signal. + this._gcObserver = new PerformanceObserver((list) => { + for (const entry of list.getEntries()) { + metrics.histogram('process.gc.duration', entry.duration, { + unit: 'ms', + attributes: { kind: gcKind((entry as PerformanceEntry & { detail?: { kind?: number } }).detail?.kind) }, + }) + } + }) + this._gcObserver.observe({ entryTypes: ['gc'] }) + } + + /** Records one sample of every available runtime series. */ + sample(metrics: Metrics): void { + const nowMs = Date.now() + + try { + this._sampleCpu(metrics, nowMs) + this._sampleMemory(metrics) + this._sampleEventLoop(metrics) + this._sampleProcess(metrics) + } finally { + // In a `finally` because `_sampleCpu` has already advanced its own cursor: a + // throw further down must not leave the two out of step, or every later CPU + // delta gets divided by a too-long window and utilization reads far too low. + this._lastSampleAtMs = nowMs + } + } + + stop(): void { + this._eventLoopDelay?.disable() + this._eventLoopDelay = undefined + this._gcObserver?.disconnect() + this._gcObserver = undefined + } + + private _sampleCpu(metrics: Metrics, nowMs: number): void { + const usage = process.cpuUsage() + const previous = this._lastCpuUsage + this._lastCpuUsage = usage + if (!previous || this._lastSampleAtMs === undefined) { + return + } + + const userDeltaMicros = usage.user - previous.user + const systemDeltaMicros = usage.system - previous.system + + // `process.cpu.time` is a counter in seconds per the OTel semantic + // convention, and per-window deltas are what delta temporality wants anyway. + metrics.count('process.cpu.time', userDeltaMicros / MICROS_PER_SECOND, { + unit: 's', + attributes: { state: 'user' }, + }) + metrics.count('process.cpu.time', systemDeltaMicros / MICROS_PER_SECOND, { + unit: 's', + attributes: { state: 'system' }, + }) + + const elapsedMs = nowMs - this._lastSampleAtMs + if (elapsedMs > 0) { + // Fraction of the CPU capacity available to this process (see + // `detectCpuCount`), so 1.0 means saturated — cpuUsage() sums every thread, + // hence the division. + const utilization = (userDeltaMicros + systemDeltaMicros) / (elapsedMs * MILLIS_TO_MICROS) / this._cpuCount + metrics.gauge('process.cpu.utilization', clampNonNegative(utilization)) + } + } + + private _sampleMemory(metrics: Metrics): void { + const memory = process.memoryUsage() + const byType: Record = { + rss: memory.rss, + heap_used: memory.heapUsed, + heap_total: memory.heapTotal, + external: memory.external, + array_buffers: memory.arrayBuffers, + } + for (const [type, value] of Object.entries(byType)) { + if (typeof value === 'number') { + metrics.gauge('process.memory.usage', value, { unit: 'byte', attributes: { type } }) + } + } + + // The heap limit is what heap_used is actually racing against, so ship it as + // its own series instead of making the reader guess the ceiling. + metrics.gauge('process.memory.heap_limit', getHeapStatistics().heap_size_limit, { unit: 'byte' }) + } + + private _sampleEventLoop(metrics: Metrics): void { + const histogram = this._eventLoopDelay + if (histogram) { + const stats: Record = { + mean: histogram.mean, + p50: histogram.percentile(50), + p90: histogram.percentile(90), + p99: histogram.percentile(99), + max: histogram.max, + } + for (const [stat, valueNanos] of Object.entries(stats)) { + if (Number.isFinite(valueNanos)) { + // Node records the full interval between monitor ticks, so an idle + // process reports ~`resolution` on every stat. Report the excess over + // the resolution instead, so 0 means "not delayed" as a reader expects. + const delayMs = Math.max(0, valueNanos / NANOS_PER_MILLI - EVENT_LOOP_RESOLUTION_MS) + metrics.gauge('process.event_loop.delay', delayMs, { unit: 'ms', attributes: { stat } }) + } + } + // Reset so each window reports the delay seen during that window, rather + // than a since-boot distribution that never recovers from one bad spike. + histogram.reset() + } + + const current = performance.eventLoopUtilization() + const delta = this._lastEventLoopUtilization + ? performance.eventLoopUtilization(current, this._lastEventLoopUtilization) + : current + this._lastEventLoopUtilization = current + if (Number.isFinite(delta.utilization)) { + metrics.gauge('process.event_loop.utilization', clampNonNegative(delta.utilization)) + } + } + + private _sampleProcess(metrics: Metrics): void { + metrics.gauge('process.uptime', process.uptime(), { unit: 's' }) + // A steadily climbing handle count is the cheapest leak signal there is. + metrics.gauge('process.active_resources', process.getActiveResourcesInfo().length) + } +} + +const NANOS_PER_MILLI = 1e6 +const MICROS_PER_SECOND = 1e6 +const MILLIS_TO_MICROS = 1000 +const EVENT_LOOP_RESOLUTION_MS = 10 + +/** + * CPU capacity available to *this process*, for the utilization ratio. + * + * `os.cpus().length` reports the host's cores, which in a container is the wrong + * denominator by one to two orders of magnitude: a pod limited to 500m on a + * 64-core node would report 0.008 while pegged at its quota. The cgroup quota is + * the number that makes 1.0 mean "saturated". + */ +function detectCpuCount(): number { + const quota = readCgroupCpuQuota() + if (quota !== undefined && quota > 0) { + return quota + } + try { + return Math.max(1, availableParallelism()) + } catch { + return 1 + } +} + +function readCgroupCpuQuota(): number | undefined { + return parseCgroupCpuQuota({ + cpuMax: readFileIfPresent('/sys/fs/cgroup/cpu.max'), + cfsQuotaUs: readFileIfPresent('/sys/fs/cgroup/cpu/cpu.cfs_quota_us'), + cfsPeriodUs: readFileIfPresent('/sys/fs/cgroup/cpu/cpu.cfs_period_us'), + }) +} + +/** + * Cores allowed by the cgroup, or `undefined` when unlimited or unreadable. + * Split out from the file reads so it can be tested without a container. + */ +export function parseCgroupCpuQuota({ + cpuMax, + cfsQuotaUs, + cfsPeriodUs, +}: { + cpuMax?: string + cfsQuotaUs?: string + cfsPeriodUs?: string +}): number | undefined { + // cgroup v2: " ", both in microseconds. + if (cpuMax) { + const [quota, period] = cpuMax.trim().split(/\s+/) + if (quota === 'max') { + return undefined + } + if (Number(quota) > 0 && Number(period) > 0) { + return Number(quota) / Number(period) + } + } + + // cgroup v1: a quota of -1 means unlimited. + const quotaV1 = Number(cfsQuotaUs) + const periodV1 = Number(cfsPeriodUs) + if (quotaV1 > 0 && periodV1 > 0) { + return quotaV1 / periodV1 + } + + return undefined +} + +function readFileIfPresent(path: string): string | undefined { + try { + return readFileSync(path, 'utf8') + } catch { + return undefined + } +} + +/** Maps the numeric perf_hooks GC kind onto a bounded attribute value. */ +function gcKind(kind: number | undefined): string { + switch (kind) { + case constants.NODE_PERFORMANCE_GC_MINOR: + return 'minor' + case constants.NODE_PERFORMANCE_GC_MAJOR: + return 'major' + case constants.NODE_PERFORMANCE_GC_INCREMENTAL: + return 'incremental' + case constants.NODE_PERFORMANCE_GC_WEAKCB: + return 'weak_callbacks' + default: + return 'unknown' + } +} + +function clampNonNegative(value: number): number { + return Number.isFinite(value) && value > 0 ? value : 0 +} diff --git a/packages/node/src/extensions/metrics-autocapture/types.ts b/packages/node/src/extensions/metrics-autocapture/types.ts new file mode 100644 index 0000000000..539b9389be --- /dev/null +++ b/packages/node/src/extensions/metrics-autocapture/types.ts @@ -0,0 +1,20 @@ +import type { Metrics } from '@posthog/core' + +/** + * A source of runtime metrics, sampled on an interval by {@link MetricsAutocapture}. + * + * The interface exists so the runtime-agnostic autocapture loop stays free of + * Node built-ins: the concrete sampler (`RuntimeMetricsCollector`, which imports + * `node:perf_hooks` and friends) is injected by the Node entrypoint, exactly how + * `PostHogContext` and the error-tracking frame modifiers are wired. + * + * @internal — an SDK-internal seam, not something users implement. + */ +export interface RuntimeMetricsSampler { + /** Starts anything that must run between samples (e.g. histograms, observers). */ + start(metrics: Metrics): void + /** Records one sample of every available series. */ + sample(metrics: Metrics): void + /** Tears down whatever `start` began. Must be safe to call twice. */ + stop(): void +} diff --git a/packages/node/src/types.ts b/packages/node/src/types.ts index 7cb6fad85a..0e9fe5fa6e 100644 --- a/packages/node/src/types.ts +++ b/packages/node/src/types.ts @@ -173,6 +173,40 @@ export type PostHogOptions = Omit