fix: add cgroup CFS throttling metrics to OTel meter (#1114)

Register four observable instruments on the existing MeterProvider to
surface container CPU bandwidth-control state so GC pauses can be
correlated with throttling in our observability backend:

- process.cpu.cfs.periods (counter)
- process.cpu.cfs.throttled_periods (counter)
- process.cpu.cfs.throttled_time (counter, ns)
- process.cpu.cfs.throttled_ratio (gauge, delta-based)

Supports cgroup v1 and v2, short-circuits cleanly on non-Linux and when
cpu.stat is unreadable, and reads the file from the OTel observable
callback (no separate timer).

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mcollina

src/internal/monitoring/cgroup-cpu-metrics.test.ts

@@ -0,0 +1,183 @@
+import * as fs from 'node:fs'
+import { Meter } from '@opentelemetry/api'
+import { afterEach, beforeEach, describe, expect, Mock, test, vi } from 'vitest'
+import { installCgroupCpuMetrics, parseCpuStat } from './cgroup-cpu-metrics'
+
+vi.mock('@internal/monitoring/logger', () => ({
+  logger: { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+  logSchema: {
+    info: vi.fn(),
+    warning: vi.fn(),
+    error: vi.fn(),
+    event: vi.fn(),
+    request: vi.fn(),
+  },
+}))
+
+vi.mock('node:fs', async () => {
+  const actual = await vi.importActual<typeof import('node:fs')>('node:fs')
+  return {
+    ...actual,
+    existsSync: vi.fn(),
+    readFileSync: vi.fn(),
+  }
+})
+
+const existsSync = fs.existsSync as unknown as Mock
+const readFileSync = fs.readFileSync as unknown as Mock
+
+const V2_BLOB = `usage_usec 1234567
+user_usec 1000000
+system_usec 234567
+nr_periods 1000
+nr_throttled 25
+throttled_usec 5000
+`
+
+const V1_BLOB = `nr_periods 500
+nr_throttled 12
+throttled_time 7500000
+`
+
+interface CapturedInstrument {
+  name: string
+  callback: (observable: { observe: (value: number) => void }) => void
+}
+
+interface CapturedObservation {
+  name: string
+  value: number
+}
+
+function createMockMeter(): {
+  meter: Meter
+  instruments: CapturedInstrument[]
+  invoke: (name: string) => CapturedObservation[]
+} {
+  const instruments: CapturedInstrument[] = []
+
+  const make = () => (name: string) => ({
+    addCallback(callback: CapturedInstrument['callback']) {
+      instruments.push({ name, callback })
+      return this
+    },
+  })
+
+  const meter = {
+    createObservableCounter: make(),
+    createObservableGauge: make(),
+  } as unknown as Meter
+
+  const invoke = (name: string): CapturedObservation[] => {
+    const observations: CapturedObservation[] = []
+    const observable = { observe: (value: number) => observations.push({ name, value }) }
+    for (const inst of instruments) {
+      if (inst.name === name) inst.callback(observable)
+    }
+    return observations
+  }
+
+  return { meter, instruments, invoke }
+}
+
+describe('parseCpuStat', () => {
+  test('parses cgroup v2 blob and converts throttled_usec to ns', () => {
+    expect(parseCpuStat(V2_BLOB, 'v2')).toEqual({
+      nr_periods: 1000,
+      nr_throttled: 25,
+      throttled_time_ns: 5_000_000,
+    })
+  })
+
+  test('parses cgroup v1 blob and keeps throttled_time as ns', () => {
+    expect(parseCpuStat(V1_BLOB, 'v1')).toEqual({
+      nr_periods: 500,
+      nr_throttled: 12,
+      throttled_time_ns: 7_500_000,
+    })
+  })
+
+  test('returns null when required fields are missing', () => {
+    expect(parseCpuStat('nr_periods 10\nnr_throttled 1\n', 'v2')).toBeNull()
+  })
+})
+
+describe('installCgroupCpuMetrics', () => {
+  const originalPlatform = process.platform
+
+  beforeEach(() => {
+    existsSync.mockReset()
+    readFileSync.mockReset()
+  })
+
+  afterEach(() => {
+    Object.defineProperty(process, 'platform', { value: originalPlatform })
+  })
+
+  test('short-circuits on non-Linux platforms and registers no instruments', () => {
+    Object.defineProperty(process, 'platform', { value: 'darwin' })
+
+    const { meter, instruments } = createMockMeter()
+    installCgroupCpuMetrics(meter)
+
+    expect(instruments).toHaveLength(0)
+    expect(readFileSync).not.toHaveBeenCalled()
+    expect(existsSync).not.toHaveBeenCalled()
+  })
+
+  test('emits 0 for throttled_ratio on the first sample (divide-by-zero guard)', () => {
+    Object.defineProperty(process, 'platform', { value: 'linux' })
+    existsSync.mockReturnValue(true)
+    readFileSync.mockReturnValue(V2_BLOB)
+
+    const { meter, invoke } = createMockMeter()
+    installCgroupCpuMetrics(meter)
+
+    expect(invoke('process.cpu.cfs.throttled_ratio')).toEqual([
+      { name: 'process.cpu.cfs.throttled_ratio', value: 0 },
+    ])
+  })
+
+  test('computes throttled_ratio between observations and avoids NaN when no new periods', () => {
+    Object.defineProperty(process, 'platform', { value: 'linux' })
+    existsSync.mockReturnValue(true)
+    readFileSync.mockReturnValue(V2_BLOB)
+
+    const { meter, invoke } = createMockMeter()
+    installCgroupCpuMetrics(meter)
+
+    // First sample → 0
+    invoke('process.cpu.cfs.throttled_ratio')
+
+    // Second sample: +100 periods, +5 throttled → 0.05
+    readFileSync.mockReturnValue(`nr_periods 1100
+nr_throttled 30
+throttled_usec 5000
+`)
+    const second = invoke('process.cpu.cfs.throttled_ratio')
+    expect(second).toEqual([{ name: 'process.cpu.cfs.throttled_ratio', value: 0.05 }])
+
+    // Third sample identical → no new periods → 0, not NaN
+    const third = invoke('process.cpu.cfs.throttled_ratio')
+    expect(third).toEqual([{ name: 'process.cpu.cfs.throttled_ratio', value: 0 }])
+  })
+
+  test('observes counter values parsed from cpu.stat', () => {
+    Object.defineProperty(process, 'platform', { value: 'linux' })
+    existsSync.mockReturnValue(true)
+    readFileSync.mockReturnValue(V2_BLOB)
+
+    const { meter, invoke } = createMockMeter()
+    installCgroupCpuMetrics(meter)
+
+    expect(invoke('process.cpu.cfs.periods')).toEqual([
+      { name: 'process.cpu.cfs.periods', value: 1000 },
+    ])
+    expect(invoke('process.cpu.cfs.throttled_periods')).toEqual([
+      { name: 'process.cpu.cfs.throttled_periods', value: 25 },
+    ])
+    expect(invoke('process.cpu.cfs.throttled_time')).toEqual([
+      { name: 'process.cpu.cfs.throttled_time', value: 5_000_000 },
+    ])
+  })
+})

src/internal/monitoring/cgroup-cpu-metrics.ts

@@ -0,0 +1,164 @@
+// Project-local metric names — these are NOT part of the OpenTelemetry semantic
+// conventions. The names `process.cpu.cfs.*` describe Linux cgroup CFS (Completely
+// Fair Scheduler) bandwidth-control state and are kept stable for our backends.
+
+import * as fs from 'node:fs'
+import { logger, logSchema } from '@internal/monitoring/logger'
+import { Meter } from '@opentelemetry/api'
+
+const V2_CONTROLLERS = '/sys/fs/cgroup/cgroup.controllers'
+const V2_STAT = '/sys/fs/cgroup/cpu.stat'
+const V1_PRIMARY = '/sys/fs/cgroup/cpu,cpuacct/cpu.stat'
+const V1_FALLBACK = '/sys/fs/cgroup/cpu/cpu.stat'
+
+export type CgroupVersion = 'v1' | 'v2'
+
+export interface CgroupCpuStatSample {
+  nr_periods: number
+  nr_throttled: number
+  throttled_time_ns: number
+}
+
+export interface CgroupSource {
+  version: CgroupVersion
+  path: string
+}
+
+export function parseCpuStat(content: string, version: CgroupVersion): CgroupCpuStatSample | null {
+  let nrPeriods: number | null = null
+  let nrThrottled: number | null = null
+  let throttledNs: number | null = null
+
+  for (const rawLine of content.split('\n')) {
+    const line = rawLine.trim()
+    if (!line) continue
+    const space = line.indexOf(' ')
+    if (space === -1) continue
+    const key = line.slice(0, space)
+    const value = Number(line.slice(space + 1).trim())
+    if (!Number.isFinite(value)) continue
+
+    if (key === 'nr_periods') {
+      nrPeriods = value
+    } else if (key === 'nr_throttled') {
+      nrThrottled = value
+    } else if (version === 'v2' && key === 'throttled_usec') {
+      throttledNs = value * 1000
+    } else if (version === 'v1' && key === 'throttled_time') {
+      throttledNs = value
+    }
+  }
+
+  if (nrPeriods === null || nrThrottled === null || throttledNs === null) {
+    return null
+  }
+  return {
+    nr_periods: nrPeriods,
+    nr_throttled: nrThrottled,
+    throttled_time_ns: throttledNs,
+  }
+}
+
+export function detectCgroupSource(): CgroupSource | null {
+  if (process.platform !== 'linux') return null
+
+  try {
+    if (fs.existsSync(V2_CONTROLLERS)) {
+      fs.readFileSync(V2_STAT, 'utf8')
+      return { version: 'v2', path: V2_STAT }
+    }
+  } catch {
+    // Fall through to v1 candidates
+  }
+
+  for (const path of [V1_PRIMARY, V1_FALLBACK]) {
+    try {
+      fs.readFileSync(path, 'utf8')
+      return { version: 'v1', path }
+    } catch {
+      // Try next
+    }
+  }
+
+  return null
+}
+
+export function installCgroupCpuMetrics(meter: Meter): void {
+  const source = detectCgroupSource()
+  if (!source) {
+    logger.debug(
+      { type: 'cgroup-cpu-metrics', platform: process.platform },
+      '[cgroup CPU metrics] cgroup cpu.stat not available, skipping'
+    )
+    return
+  }
+
+  let previous: { nr_periods: number; nr_throttled: number } | null = null
+  let readErrorLogged = false
+
+  const readSample = (): CgroupCpuStatSample | null => {
+    try {
+      const content = fs.readFileSync(source.path, 'utf8')
+      return parseCpuStat(content, source.version)
+    } catch (error) {
+      if (!readErrorLogged) {
+        readErrorLogged = true
+        logSchema.warning(logger, '[cgroup CPU metrics] Failed to read cpu.stat', {
+          type: 'cgroup-cpu-metrics',
+          error,
+        })
+      }
+      return null
+    }
+  }
+
+  meter
+    .createObservableCounter('process.cpu.cfs.periods', {
+      description: 'Total CFS periods elapsed for the cgroup',
+      unit: '{period}',
+    })
+    .addCallback((observable) => {
+      const sample = readSample()
+      if (sample) observable.observe(sample.nr_periods)
+    })
+
+  meter
+    .createObservableCounter('process.cpu.cfs.throttled_periods', {
+      description: 'CFS periods where the cgroup was throttled',
+      unit: '{period}',
+    })
+    .addCallback((observable) => {
+      const sample = readSample()
+      if (sample) observable.observe(sample.nr_throttled)
+    })
+
+  meter
+    .createObservableCounter('process.cpu.cfs.throttled_time', {
+      description: 'Total time the cgroup was throttled, in nanoseconds',
+      unit: 'ns',
+    })
+    .addCallback((observable) => {
+      const sample = readSample()
+      if (sample) observable.observe(sample.throttled_time_ns)
+    })
+
+  meter
+    .createObservableGauge('process.cpu.cfs.throttled_ratio', {
+      description: 'Fraction of recent CFS periods that were throttled',
+      unit: '1',
+    })
+    .addCallback((observable) => {
+      const sample = readSample()
+      if (!sample) return
+      if (!previous) {
+        previous = { nr_periods: sample.nr_periods, nr_throttled: sample.nr_throttled }
+        observable.observe(0)
+        return
+      }
+      const dPeriods = sample.nr_periods - previous.nr_periods
+      const dThrottled = sample.nr_throttled - previous.nr_throttled
+      previous = { nr_periods: sample.nr_periods, nr_throttled: sample.nr_throttled }
+      const ratio = dPeriods > 0 ? dThrottled / dPeriods : 0
+      observable.observe(Number.isFinite(ratio) ? ratio : 0)
+    })
+}