From 8ed74213a6c020fd0978c81d6b0cab6ea11bca68 Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Wed, 9 Sep 2026 08:14:07 +0000 Subject: [PATCH 1/6] feat(telemetry): flag event loop blocking --- .changeset/slow-loops-report.md | 5 + agents/etc/agents.api.md | 263 ++++++++++++++++- agents/package.json | 2 + agents/src/ipc/job_proc_lazy_main.ts | 15 +- agents/src/telemetry/index.ts | 1 + agents/src/telemetry/loop_monitor.test.ts | 309 ++++++++++++++++++++ agents/src/telemetry/loop_monitor.ts | 334 ++++++++++++++++++++++ agents/src/telemetry/otel_metrics.ts | 36 +++ agents/src/telemetry/session_context.ts | 27 ++ agents/src/telemetry/trace_types.test.ts | 13 + agents/src/telemetry/trace_types.ts | 17 ++ agents/src/telemetry/traces.ts | 94 +++++- agents/src/telemetry/upload_gate.ts | 26 ++ agents/src/voice/agent_session.ts | 29 ++ agents/src/worker.ts | 9 + pnpm-lock.yaml | 37 +++ 16 files changed, 1210 insertions(+), 7 deletions(-) create mode 100644 .changeset/slow-loops-report.md create mode 100644 agents/src/telemetry/loop_monitor.test.ts create mode 100644 agents/src/telemetry/loop_monitor.ts create mode 100644 agents/src/telemetry/otel_metrics.ts create mode 100644 agents/src/telemetry/session_context.ts diff --git a/.changeset/slow-loops-report.md b/.changeset/slow-loops-report.md new file mode 100644 index 000000000..19cc93329 --- /dev/null +++ b/.changeset/slow-loops-report.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': minor +--- + +Flag synchronous code that blocks agent event loops in telemetry and logs. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index 9ff12afb4..51600b358 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -562,6 +562,8 @@ export class AgentSession extends AgentSession_base // // @internal (undocumented) _recorderIO?: RecorderIO; + // @internal (undocumented) + _recordLoopStall(durationInS: number, timestampMs: number): void; // @internal _redactionEnabled: boolean; // (undocumented) @@ -1142,6 +1144,66 @@ const ATTR_AMD_SPEECH_DURATION = "lk.amd.speech_duration"; // @public (undocumented) const ATTR_AMD_TRANSCRIPT = "lk.pii.amd.transcript"; +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_COUNT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_COUNT = "lk.blocking.count"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_CPU_TIME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const ATTR_BLOCKING_CPU_TIME = "lk.blocking.cpu_time"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_DURATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const ATTR_BLOCKING_DURATION = "lk.blocking.duration"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_GC_TIME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_GC_TIME = "lk.blocking.gc_time"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_IMPORT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_IMPORT = "lk.blocking.import"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_MAX_DURATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_MAX_DURATION = "lk.blocking.max_duration"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_SEVERITY" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_SEVERITY = "lk.blocking.severity"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_STACK" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_STACK = "lk.blocking.stack"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_SUPPRESSED" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_SUPPRESSED = "lk.blocking.suppressed"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_TASK" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_TASK = "lk.blocking.task"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_THRESHOLD" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_THRESHOLD = "lk.blocking.threshold"; + +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_TOTAL_DURATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_BLOCKING_TOTAL_DURATION = "lk.blocking.total_duration"; + // Warning: (ae-missing-release-tag) "ATTR_CHAT_CTX" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -2073,6 +2135,19 @@ declare namespace beta { } } +// Warning: (ae-missing-release-tag) "BlockedReport" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface BlockedReport { + cpuTime: number; + duration: number; + // (undocumented) + severity: LoopMonitorSeverity; + startedAt: number; + // (undocumented) + warnThreshold: number; +} + // Warning: (ae-missing-release-tag) "BufferedSentenceStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -3080,6 +3155,11 @@ type DeepSeekModels = 'deepseek-ai/deepseek-v3' | 'deepseek-ai/deepseek-v3.2'; // @public (undocumented) export const DEFAULT_API_CONNECT_OPTIONS: APIConnectOptions; +// Warning: (ae-missing-release-tag) "DEFAULT_ERROR_THRESHOLD" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const DEFAULT_ERROR_THRESHOLD = 500; + // Warning: (ae-missing-release-tag) "DEFAULT_EXPRESSIVE_OPTIONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -3111,6 +3191,16 @@ const DEFAULT_SPEECH_STEERING_OPTIONS: SpeechSteeringOptions; // @public (undocumented) export const DEFAULT_TEXT_INPUT_CALLBACK: TextInputCallback; +// Warning: (ae-missing-release-tag) "DEFAULT_TICK_INTERVAL" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const DEFAULT_TICK_INTERVAL = 20; + +// Warning: (ae-missing-release-tag) "DEFAULT_WARN_THRESHOLD" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const DEFAULT_WARN_THRESHOLD = 100; + // Warning: (ae-missing-release-tag) "defaultEndpointingOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -3288,6 +3378,16 @@ export interface EndpointingOptions { mode: 'fixed' | 'dynamic'; } +// Warning: (ae-missing-release-tag) "ENV_ERROR_THRESHOLD_MS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ENV_ERROR_THRESHOLD_MS = "LIVEKIT_AGENTS_LOOP_BLOCK_ERROR_MS"; + +// Warning: (ae-missing-release-tag) "ENV_WARN_THRESHOLD_MS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ENV_WARN_THRESHOLD_MS = "LIVEKIT_AGENTS_LOOP_BLOCK_WARN_MS"; + declare namespace eot { export { TurnDetector, @@ -3428,6 +3528,44 @@ class EventAssert { protected _raise(message: string): never; } +// Warning: (ae-missing-release-tag) "EventLoopMonitor" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +class EventLoopMonitor { + constructor(options?: EventLoopMonitorOptions); + // @internal (undocumented) + _buildReport(duration: number, cpuTime: number): BlockedReport; + // (undocumented) + readonly errorThreshold: number; + // @internal + _onReport?: (report: BlockedReport) => void; + // @internal (undocumented) + _report(report: BlockedReport): void; + setReportContext(context: Context | undefined, runner?: ReportContextRunner): void; + start(): void; + stop(): void; + // (undocumented) + readonly tickInterval: number; + // (undocumented) + readonly warnThreshold: number; +} + +// Warning: (ae-missing-release-tag) "EventLoopMonitorOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface EventLoopMonitorOptions { + // (undocumented) + emitSpans?: boolean; + // (undocumented) + errorThreshold?: number; + // (undocumented) + name?: string; + // (undocumented) + tickInterval?: number; + // (undocumented) + warnThreshold?: number; +} + // Warning: (ae-missing-release-tag) "EventRangeAssert" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -4237,6 +4375,11 @@ export function getJobContext>(require // @public (undocumented) export function getLanguageRegion(language: string): string | undefined; +// Warning: (ae-missing-release-tag) "getMonitor" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +function getMonitor(): EventLoopMonitor | undefined; + // Warning: (ae-missing-release-tag) "GoogleModels" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -5326,6 +5469,49 @@ export const logMetrics: (metrics: AgentMetrics) => void; // @public export function loopAudioFramesFromFile(filePath: string, options?: AudioDecodeOptions): AsyncGenerator; +declare namespace loopMonitor { + export { + _tickIntervalFor, + startMonitoring, + stopMonitoring, + getMonitor, + DEFAULT_WARN_THRESHOLD, + DEFAULT_ERROR_THRESHOLD, + DEFAULT_TICK_INTERVAL, + ENV_WARN_THRESHOLD_MS, + ENV_ERROR_THRESHOLD_MS, + MAX_SPANS_PER_MINUTE, + MAX_LOGS_PER_MINUTE, + SPAN_NAME, + LoopMonitorSeverity, + BlockedReport, + LoopMonitorThresholds, + _RateLimiter, + EventLoopMonitorOptions, + ReportContextRunner, + EventLoopMonitor, + StartMonitoringOptions + } +} + +// Warning: (ae-missing-release-tag) "LoopMonitorSeverity" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +type LoopMonitorSeverity = 'warning' | 'error'; + +// Warning: (ae-missing-release-tag) "LoopMonitorThresholds" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +class LoopMonitorThresholds { + constructor(warn: number, error: number); + // (undocumented) + readonly error: number; + // (undocumented) + static fromEnv(env?: NodeJS.ProcessEnv): LoopMonitorThresholds | undefined; + // (undocumented) + readonly warn: number; +} + // Warning: (ae-missing-release-tag) "markInferenceSpanRecorded" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -5345,6 +5531,16 @@ interface MarkupInfo { // @public function matchMood(label: string, fallback?: AgentMood | null): AgentMood | null; +// Warning: (ae-missing-release-tag) "MAX_LOGS_PER_MINUTE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const MAX_LOGS_PER_MINUTE = 5; + +// Warning: (ae-missing-release-tag) "MAX_SPANS_PER_MINUTE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const MAX_SPANS_PER_MINUTE = 30; + // Warning: (ae-missing-release-tag) "maxInputLen" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -5973,6 +6169,15 @@ export interface QueueAudioOutputClearEvent { // @public export type QueueAudioOutputItem = AudioFrame | AudioSegmentEnd; +// @internal (undocumented) +class _RateLimiter { + constructor(limit: number); + // (undocumented) + allow(now: number): boolean; + // (undocumented) + takeSuppressed(): number; +} + // Warning: (ae-missing-release-tag) "readStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -6258,6 +6463,11 @@ export interface ReplyPromptArgs { callIds: string[]; } +// Warning: (ae-missing-release-tag) "ReportContextRunner" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +type ReportContextRunner = (fn: () => T) => T; + // Warning: (ae-missing-release-tag) "resampleStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -7071,6 +7281,11 @@ export function sortedToolEntries(toolCtx: ToolConte // @internal export function sortedToolNames(toolCtx: ToolContext | undefined): string[]; +// Warning: (ae-missing-release-tag) "SPAN_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const SPAN_NAME = "event_loop_blocked"; + // Warning: (ae-missing-release-tag) "SpanProcessorLike" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @deprecated (undocumented) @@ -7357,6 +7572,23 @@ function splitAllMarkup(text: string): [string, ExpressiveTag[]]; // @public const splitWords: (text: string, ignorePunctuation?: boolean) => [string, number, number][]; +// Warning: (ae-missing-release-tag) "startMonitoring" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function startMonitoring(options?: StartMonitoringOptions): EventLoopMonitor | undefined; + +// Warning: (ae-missing-release-tag) "StartMonitoringOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface StartMonitoringOptions { + // (undocumented) + emitSpans?: boolean; + // (undocumented) + name?: string; + // (undocumented) + thresholds?: LoopMonitorThresholds; +} + // Warning: (ae-missing-release-tag) "startSoon" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -7379,6 +7611,11 @@ interface StartSpanOptions { // @public function steeringInstructions(provider: string, steering: SpeechSteeringOptions): string; +// Warning: (ae-missing-release-tag) "stopMonitoring" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +function stopMonitoring(monitor?: EventLoopMonitor): void; + // Warning: (ae-missing-release-tag) "StopResponse" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -7950,6 +8187,7 @@ declare namespace telemetry { PinoLogObject, genAI, REDACTED_EXCEPTION_MESSAGE, + loopMonitor, traceTypes, FanoutSpanProcessor, flushOtelLogs, @@ -8121,6 +8359,9 @@ class ThresholdOptions { // @public type ThresholdOverride = number | Record | undefined; +// @internal (undocumented) +function _tickIntervalFor(warnThreshold: number): number; + // Warning: (ae-missing-release-tag) "TimedString" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -8507,6 +8748,18 @@ declare namespace traceTypes { ATTR_TTS_METRICS, ATTR_REALTIME_MODEL_METRICS, ATTR_E2E_LATENCY, + ATTR_BLOCKING_DURATION, + ATTR_BLOCKING_THRESHOLD, + ATTR_BLOCKING_SEVERITY, + ATTR_BLOCKING_TASK, + ATTR_BLOCKING_STACK, + ATTR_BLOCKING_GC_TIME, + ATTR_BLOCKING_CPU_TIME, + ATTR_BLOCKING_IMPORT, + ATTR_BLOCKING_SUPPRESSED, + ATTR_BLOCKING_COUNT, + ATTR_BLOCKING_TOTAL_DURATION, + ATTR_BLOCKING_MAX_DURATION, ATTR_GEN_AI_OPERATION_NAME, ATTR_GEN_AI_PROVIDER_NAME, ATTR_GEN_AI_REQUEST_MODEL, @@ -9696,11 +9949,11 @@ export const zipFunctionCallsAndOutputs: (event: FunctionToolsExecutedEvent) => // src/metrics/base.ts:198:3 - (ae-forgotten-export) The symbol "RealtimeModelMetricsOutputTokenDetails" needs to be exported by the entry point index.d.ts // src/stt/stt.ts:361:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "STT" // src/utils.ts:550:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "cancelled" -// src/voice/agent_session.ts:380:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver -// src/voice/agent_session.ts:998:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1651:5 - (ae-forgotten-export) The symbol "STTError" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1651:5 - (ae-forgotten-export) The symbol "TTSError" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1651:5 - (ae-forgotten-export) The symbol "LLMError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:381:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver +// src/voice/agent_session.ts:1002:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1658:5 - (ae-forgotten-export) The symbol "STTError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1658:5 - (ae-forgotten-export) The symbol "TTSError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1658:5 - (ae-forgotten-export) The symbol "LLMError" needs to be exported by the entry point index.d.ts // src/voice/amd.ts:314:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "waitForTrackPublication" has more than one declaration; you need to add a TSDoc member reference selector // src/voice/amd.ts:314:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "gateListening" // src/voice/amd.ts:322:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "aclose" diff --git a/agents/package.json b/agents/package.json index ba708f7b8..0d13cffc6 100644 --- a/agents/package.json +++ b/agents/package.json @@ -62,11 +62,13 @@ "@opentelemetry/api-logs": "^0.220.0", "@opentelemetry/core": "^2.8.0", "@opentelemetry/exporter-logs-otlp-proto": "^0.220.0", + "@opentelemetry/exporter-metrics-otlp-proto": "^0.220.0", "@opentelemetry/exporter-trace-otlp-proto": "^0.220.0", "@opentelemetry/instrumentation-pino": "^0.66.0", "@opentelemetry/otlp-exporter-base": "^0.220.0", "@opentelemetry/resources": "^2.8.0", "@opentelemetry/sdk-logs": "^0.220.0", + "@opentelemetry/sdk-metrics": "^2.8.0", "@opentelemetry/sdk-trace-base": "^2.8.0", "@opentelemetry/sdk-trace-node": "^2.8.0", "@opentelemetry/semantic-conventions": "^1.28.0", diff --git a/agents/src/ipc/job_proc_lazy_main.ts b/agents/src/ipc/job_proc_lazy_main.ts index 16cabb403..23b400816 100644 --- a/agents/src/ipc/job_proc_lazy_main.ts +++ b/agents/src/ipc/job_proc_lazy_main.ts @@ -3,11 +3,18 @@ // SPDX-License-Identifier: Apache-2.0 import { Room, RoomEvent, dispose } from '@livekit/rtc-node'; import { ThrowsPromise } from '@livekit/throws-transformer/throws'; +import { context as otelContext } from '@opentelemetry/api'; import { EventEmitter, once } from 'node:events'; import { pathToFileURL } from 'node:url'; import type { Logger } from 'pino'; import { type Agent, isAgent } from '../generator.js'; -import { JobContext, JobProcess, type RunningJobInfo, runWithJobContextAsync } from '../job.js'; +import { + JobContext, + JobProcess, + type RunningJobInfo, + runWithJobContext, + runWithJobContextAsync, +} from '../job.js'; import { finalizeSession, flushJobLogs, @@ -17,6 +24,7 @@ import { } from '../job_lifecycle.js'; import { initializeLogger, log } from '../log.js'; import type { SimulationContext } from '../simulation.js'; +import { getMonitor, startMonitoring, stopMonitoring } from '../telemetry/loop_monitor.js'; import { Future, shortuuid } from '../utils.js'; import { defaultInitializeProcessFunc } from '../worker.js'; import type { InferenceExecutor } from './inference_executor.js'; @@ -165,6 +173,9 @@ const startJob = ( span.setAttribute(traceTypes.ATTR_JOB_ID, info.job.id); span.setAttribute(traceTypes.ATTR_AGENT_NAME, info.job.agentName); span.setAttribute(traceTypes.ATTR_ROOM_NAME, info.job.room?.name ?? ''); + getMonitor()?.setReportContext(otelContext.active(), (fn) => + runWithJobContext(ctx, fn), + ); return func(ctx); }, { name: 'job_entrypoint' }, @@ -260,6 +271,7 @@ const startJob = ( logger.debug('initializing job runner'); await agent.prewarm(proc); logger.debug('job runner initialized'); + const loopMonitor = startMonitoring({ name: 'job' }); safeSend({ case: 'initializeResponse', value: undefined }); let job: JobTask | undefined = undefined; @@ -315,6 +327,7 @@ const startJob = ( process.on('message', messageHandler); await join.await; + if (loopMonitor) stopMonitoring(loopMonitor); clearTimeout(orphanedTimeout); process.off('message', messageHandler); diff --git a/agents/src/telemetry/index.ts b/agents/src/telemetry/index.ts index cb687e084..01597f305 100644 --- a/agents/src/telemetry/index.ts +++ b/agents/src/telemetry/index.ts @@ -25,6 +25,7 @@ export { } from './pino_otel_transport.js'; export * as genAI from './gen_ai.js'; export { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; +export * as loopMonitor from './loop_monitor.js'; export * as traceTypes from './trace_types.js'; export { FanoutSpanProcessor, diff --git a/agents/src/telemetry/loop_monitor.test.ts b/agents/src/telemetry/loop_monitor.test.ts new file mode 100644 index 000000000..749c2cb77 --- /dev/null +++ b/agents/src/telemetry/loop_monitor.test.ts @@ -0,0 +1,309 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { ROOT_CONTEXT, SpanStatusCode, trace } from '@opentelemetry/api'; +import { InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; +import { Worker } from 'node:worker_threads'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { type JobContext, getJobContext, runWithJobContext } from '../job.js'; +import { + type BlockedReport, + DEFAULT_ERROR_THRESHOLD, + DEFAULT_WARN_THRESHOLD, + ENV_ERROR_THRESHOLD_MS, + ENV_WARN_THRESHOLD_MS, + EventLoopMonitor, + LoopMonitorThresholds, + MAX_SPANS_PER_MINUTE, + SPAN_NAME, + _RateLimiter, + _tickIntervalFor, + getMonitor, + startMonitoring, + stopMonitoring, +} from './loop_monitor.js'; +import * as otelMetrics from './otel_metrics.js'; +import { + ATTR_BLOCKING_CPU_TIME, + ATTR_BLOCKING_DURATION, + ATTR_BLOCKING_SEVERITY, + ATTR_BLOCKING_THRESHOLD, +} from './trace_types.js'; +import { setTracerProvider, tracer } from './traces.js'; + +const WARN = 30; +const ERROR = 150; +const TICK = 5; + +function blockLoop(duration: number): void { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, duration); +} + +function settle(): Promise { + return new Promise((resolve) => setTimeout(resolve, TICK * 4)); +} + +function fakeJob(session?: unknown): JobContext { + return { + _primaryAgentSession: session, + job: { id: 'AJ_test', room: { sid: 'RM_test' } }, + } as unknown as JobContext; +} + +describe.sequential('event loop monitor', () => { + let exporter: InMemorySpanExporter; + let provider: NodeTracerProvider; + let originalProvider: ReturnType; + let monitor: EventLoopMonitor; + let reports: BlockedReport[]; + let sessionRoot: ReturnType; + + beforeEach(async () => { + originalProvider = tracer.getProvider(); + exporter = new InMemorySpanExporter(); + provider = new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + setTracerProvider(provider); + reports = []; + monitor = new EventLoopMonitor({ + warnThreshold: WARN, + errorThreshold: ERROR, + tickInterval: TICK, + }); + monitor._onReport = (report) => reports.push(report); + sessionRoot = tracer.startSpan({ name: 'agent_session' }); + const session = { rootSpanContext: trace.setSpan(ROOT_CONTEXT, sessionRoot) }; + const job = fakeJob(session); + monitor.setReportContext(trace.setSpan(ROOT_CONTEXT, sessionRoot), (fn) => + runWithJobContext(job, fn), + ); + monitor.start(); + await new Promise((resolve) => setTimeout(resolve, WARN)); + }); + + afterEach(async () => { + monitor.stop(); + sessionRoot.end(); + setTracerProvider(originalProvider); + await provider.shutdown(); + vi.restoreAllMocks(); + stopMonitoring(); + }); + + function blockedSpans() { + return exporter.getFinishedSpans().filter((span) => span.name === SPAN_NAME); + } + + it('reports a blocking call as a backdated error span', async () => { + blockLoop(200); + await settle(); + const [span] = blockedSpans(); + expect(span).toBeDefined(); + const duration = span!.attributes[ATTR_BLOCKING_DURATION] as number; + expect(duration).toBeGreaterThanOrEqual((200 - TICK - 10) / 1000); + expect(duration).toBeLessThanOrEqual(0.3); + expect(span!.attributes[ATTR_BLOCKING_THRESHOLD]).toBe(WARN / 1000); + expect(span!.attributes[ATTR_BLOCKING_SEVERITY]).toBe('error'); + expect(span!.status.code).toBe(SpanStatusCode.ERROR); + const elapsed = + (span!.endTime[0] - span!.startTime[0]) * 1000 + + (span!.endTime[1] - span!.startTime[1]) / 1e6; + expect(elapsed).toBeCloseTo(duration * 1000, 3); + expect(span!.attributes[ATTR_BLOCKING_CPU_TIME]).toBeTypeOf('number'); + }); + + it('reports a block between thresholds as a warning', async () => { + blockLoop(70); + await settle(); + const [span] = blockedSpans(); + expect(span!.attributes[ATTR_BLOCKING_SEVERITY]).toBe('warning'); + expect(span!.status.code).toBe(SpanStatusCode.UNSET); + }); + + it('does not report cooperative work', async () => { + for (let i = 0; i < 40; i++) { + blockLoop(2); + await new Promise((resolve) => setImmediate(resolve)); + } + await new Promise((resolve) => setTimeout(resolve, WARN * 2)); + expect(reports).toEqual([]); + }); + + it('parents a pre-session stall to the report context', async () => { + const entrypoint = tracer.startSpan({ name: 'job_entrypoint' }); + const context = trace.setSpan(ROOT_CONTEXT, entrypoint); + monitor.setReportContext(context, (fn) => runWithJobContext(fakeJob(), fn)); + blockLoop(70); + await settle(); + entrypoint.end(); + const [span] = blockedSpans(); + expect(span!.parentSpanContext?.spanId).toBe(entrypoint.spanContext().spanId); + expect( + span!.endTime[0] < entrypoint.endTime![0] || span!.endTime[1] <= entrypoint.endTime![1], + ).toBe(true); + }); + + it('does not emit a span without a job', async () => { + monitor.setReportContext(undefined); + blockLoop(70); + await settle(); + expect(blockedSpans()).toEqual([]); + expect(reports).toHaveLength(1); + }); + + it('records every stall on the active session', () => { + monitor.stop(); + const seen: number[] = []; + const job = fakeJob({ _recordLoopStall: (duration: number) => seen.push(duration) }); + monitor.setReportContext(undefined, (fn) => runWithJobContext(job, fn)); + const report = monitor._buildReport(100, 20); + for (let i = 0; i < 40; i++) monitor._report(report); + expect(seen).toHaveLength(40); + expect(seen.every((duration) => duration === 0.1)).toBe(true); + }); + + it('stops idempotently and stays quiet', async () => { + monitor.stop(); + monitor.stop(); + blockLoop(70); + await settle(); + expect(blockedSpans()).toEqual([]); + }); + + it('does not report an idle loop', async () => { + await new Promise((resolve) => setTimeout(resolve, WARN * 12)); + expect(reports).toEqual([]); + }); + + it('does not report blocking work outside the event loop', async () => { + const worker = new Worker( + `Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ${WARN * 6})`, + { eval: true }, + ); + await new Promise((resolve, reject) => { + worker.once('error', reject); + worker.once('exit', () => resolve()); + }); + await settle(); + expect(reports).toEqual([]); + }); + + it('keeps spans disabled in worker mode', async () => { + monitor.stop(); + exporter.reset(); + const workerMonitor = new EventLoopMonitor({ + warnThreshold: WARN, + errorThreshold: ERROR, + tickInterval: TICK, + emitSpans: false, + }); + const workerReports: BlockedReport[] = []; + workerMonitor._onReport = (report) => workerReports.push(report); + workerMonitor.start(); + await new Promise((resolve) => setTimeout(resolve, WARN)); + blockLoop(80); + await settle(); + workerMonitor.stop(); + expect(blockedSpans()).toEqual([]); + expect(workerReports).toHaveLength(1); + }); + + it('records the metric for every stall past span rate limits in seconds', () => { + monitor.stop(); + const record = vi.spyOn(otelMetrics, 'recordEventLoopBlocked').mockImplementation(() => {}); + const report = monitor._buildReport(100, 20); + for (let i = 0; i < 40; i++) monitor._report(report); + expect(record).toHaveBeenCalledTimes(40); + expect(record).toHaveBeenCalledWith(0.1, 'warning'); + expect(reports).toHaveLength(MAX_SPANS_PER_MINUTE); + }); + + it('restores report context even for rate-limited metrics', () => { + monitor.stop(); + const jobs: (JobContext | undefined)[] = []; + const job = fakeJob(); + vi.spyOn(otelMetrics, 'recordEventLoopBlocked').mockImplementation(() => { + jobs.push(getJobContext(false)); + }); + const context = trace.setSpan(ROOT_CONTEXT, sessionRoot); + monitor.setReportContext(context, (fn) => runWithJobContext(job, fn)); + for (let i = 0; i < 40; i++) monitor._report(monitor._buildReport(100, 20)); + expect(jobs).toHaveLength(40); + expect(jobs.every((value) => value === job)).toBe(true); + }); +}); + +describe.sequential('event loop monitor helpers', () => { + afterEach(() => { + stopMonitoring(); + vi.unstubAllEnvs(); + }); + + it('rate limiter counts suppressed reports and uses a rolling window', () => { + const limiter = new _RateLimiter(2); + expect(limiter.allow(100_000)).toBe(true); + expect(limiter.allow(100_100)).toBe(true); + expect(limiter.allow(100_200)).toBe(false); + expect(limiter.allow(100_300)).toBe(false); + expect(limiter.takeSuppressed()).toBe(2); + expect(limiter.takeSuppressed()).toBe(0); + expect(limiter.allow(161_000)).toBe(true); + }); + + it('reads, validates, disables, and clamps environment thresholds', () => { + expect(LoopMonitorThresholds.fromEnv({})).toEqual( + new LoopMonitorThresholds(DEFAULT_WARN_THRESHOLD, DEFAULT_ERROR_THRESHOLD), + ); + expect( + LoopMonitorThresholds.fromEnv({ + [ENV_WARN_THRESHOLD_MS]: '100', + [ENV_ERROR_THRESHOLD_MS]: '1000', + }), + ).toEqual(new LoopMonitorThresholds(100, 1000)); + expect(LoopMonitorThresholds.fromEnv({ [ENV_WARN_THRESHOLD_MS]: '0' })).toBeUndefined(); + for (const bad of ['fast', 'NaN', 'Infinity', '-Infinity']) { + expect(LoopMonitorThresholds.fromEnv({ [ENV_WARN_THRESHOLD_MS]: bad })?.warn).toBe(100); + } + expect( + LoopMonitorThresholds.fromEnv({ + [ENV_WARN_THRESHOLD_MS]: '200', + [ENV_ERROR_THRESHOLD_MS]: '20', + }), + ).toEqual(new LoopMonitorThresholds(200, 200)); + }); + + it('starts one process monitor and stops it', () => { + const thresholds = new LoopMonitorThresholds(WARN, ERROR); + const monitor = startMonitoring({ thresholds }); + expect(monitor).toBeDefined(); + expect(getMonitor()).toBe(monitor); + expect(startMonitoring({ thresholds })).toBeUndefined(); + stopMonitoring(); + expect(getMonitor()).toBeUndefined(); + vi.stubEnv(ENV_WARN_THRESHOLD_MS, '0'); + expect(startMonitoring()).toBeUndefined(); + }); + + it('validates constructor thresholds', () => { + expect(() => new EventLoopMonitor({ warnThreshold: 0 })).toThrow(); + expect(() => new EventLoopMonitor({ warnThreshold: 100, errorThreshold: 50 })).toThrow(); + expect(() => new EventLoopMonitor({ warnThreshold: 10, tickInterval: 50 })).toThrow(); + }); + + it('uses a bounded fifth of the warning threshold', () => { + expect(_tickIntervalFor(100)).toBe(20); + expect(_tickIntervalFor(250)).toBe(50); + expect(_tickIntervalFor(1000)).toBe(50); + expect(_tickIntervalFor(50)).toBe(20); + }); + + it('raises a warning threshold below the tick floor', () => { + const monitor = startMonitoring({ + thresholds: new LoopMonitorThresholds(5, 10), + emitSpans: false, + }); + expect(monitor?.warnThreshold).toBe(20); + expect(monitor?.errorThreshold).toBe(20); + expect(monitor?.tickInterval).toBe(20); + }); +}); diff --git a/agents/src/telemetry/loop_monitor.ts b/agents/src/telemetry/loop_monitor.ts new file mode 100644 index 000000000..a83db363b --- /dev/null +++ b/agents/src/telemetry/loop_monitor.ts @@ -0,0 +1,334 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { + type Attributes, + type Context, + SpanStatusCode, + context as otelContext, +} from '@opentelemetry/api'; +import { getJobContext } from '../job.js'; +import { log } from '../log.js'; +import { recordEventLoopBlocked } from './otel_metrics.js'; +import { recordLoopStall, sessionRootContext } from './session_context.js'; +import { + ATTR_BLOCKING_CPU_TIME, + ATTR_BLOCKING_DURATION, + ATTR_BLOCKING_SEVERITY, + ATTR_BLOCKING_SUPPRESSED, + ATTR_BLOCKING_THRESHOLD, +} from './trace_types.js'; +import { tracer } from './traces.js'; + +export const DEFAULT_WARN_THRESHOLD = 100; +export const DEFAULT_ERROR_THRESHOLD = 500; +export const DEFAULT_TICK_INTERVAL = 20; +const TICKS_PER_WARN_THRESHOLD = 5; +const MIN_TICK_INTERVAL = 20; +const MAX_TICK_INTERVAL = 50; + +export const ENV_WARN_THRESHOLD_MS = 'LIVEKIT_AGENTS_LOOP_BLOCK_WARN_MS'; +export const ENV_ERROR_THRESHOLD_MS = 'LIVEKIT_AGENTS_LOOP_BLOCK_ERROR_MS'; +export const MAX_SPANS_PER_MINUTE = 30; +export const MAX_LOGS_PER_MINUTE = 5; +export const SPAN_NAME = 'event_loop_blocked'; + +export type LoopMonitorSeverity = 'warning' | 'error'; + +export interface BlockedReport { + /** Heartbeat lag in milliseconds. */ + duration: number; + /** Approximate wall-clock start in milliseconds since the Unix epoch. */ + startedAt: number; + warnThreshold: number; + severity: LoopMonitorSeverity; + /** Process CPU consumed during the heartbeat interval, in milliseconds. */ + cpuTime: number; +} + +export class LoopMonitorThresholds { + constructor( + readonly warn: number, + readonly error: number, + ) {} + + static fromEnv(env: NodeJS.ProcessEnv = process.env): LoopMonitorThresholds | undefined { + const warn = envMilliseconds(env, ENV_WARN_THRESHOLD_MS, DEFAULT_WARN_THRESHOLD); + let error = envMilliseconds(env, ENV_ERROR_THRESHOLD_MS, DEFAULT_ERROR_THRESHOLD); + if (warn <= 0) return undefined; + if (error < warn) { + log().warn( + { errorThreshold: error, warnThreshold: warn }, + `${ENV_ERROR_THRESHOLD_MS} is below ${ENV_WARN_THRESHOLD_MS}; using the warn threshold for both`, + ); + error = warn; + } + return new LoopMonitorThresholds(warn, error); + } +} + +function envMilliseconds(env: NodeJS.ProcessEnv, name: string, defaultValue: number): number { + const raw = env[name]; + if (raw === undefined || raw.trim() === '') return defaultValue; + const value = Number(raw); + if (!Number.isFinite(value) || value < 0) { + log().warn( + { value: raw }, + `invalid ${name}, expected a finite number >= 0; using ${defaultValue}ms`, + ); + return defaultValue; + } + return value; +} + +/** @internal */ +export class _RateLimiter { + readonly #limit: number; + readonly #events: number[] = []; + #suppressed = 0; + + constructor(limit: number) { + this.#limit = limit; + } + + allow(now: number): boolean { + const windowStart = now - 60_000; + while (this.#events.length && this.#events[0]! < windowStart) this.#events.shift(); + if (this.#events.length >= this.#limit) { + this.#suppressed++; + return false; + } + this.#events.push(now); + return true; + } + + takeSuppressed(): number { + const suppressed = this.#suppressed; + this.#suppressed = 0; + return suppressed; + } +} + +export interface EventLoopMonitorOptions { + warnThreshold?: number; + errorThreshold?: number; + tickInterval?: number; + name?: string; + emitSpans?: boolean; +} + +export type ReportContextRunner = (fn: () => T) => T; + +/** Detect synchronous work that prevents the Node event loop from servicing timers. */ +export class EventLoopMonitor { + readonly warnThreshold: number; + readonly errorThreshold: number; + readonly tickInterval: number; + readonly #name: string; + readonly #emitSpans: boolean; + readonly #spanLimiter = new _RateLimiter(MAX_SPANS_PER_MINUTE); + readonly #logLimiter = new _RateLimiter(MAX_LOGS_PER_MINUTE); + #timer?: NodeJS.Timeout; + #started = false; + #closed = false; + #lastTickAt = 0; + #lastCpuUsage: NodeJS.CpuUsage = { user: 0, system: 0 }; + #reportContext?: Context; + #reportContextRunner?: ReportContextRunner; + + /** Tests and integrations may observe reports without going through OpenTelemetry. @internal */ + _onReport?: (report: BlockedReport) => void; + + constructor(options: EventLoopMonitorOptions = {}) { + this.warnThreshold = options.warnThreshold ?? DEFAULT_WARN_THRESHOLD; + this.errorThreshold = options.errorThreshold ?? DEFAULT_ERROR_THRESHOLD; + this.tickInterval = options.tickInterval ?? DEFAULT_TICK_INTERVAL; + this.#name = options.name ?? 'event-loop'; + this.#emitSpans = options.emitSpans ?? true; + if (this.warnThreshold <= 0) throw new Error('warnThreshold must be > 0'); + if (this.errorThreshold < this.warnThreshold) { + throw new Error('errorThreshold must be >= warnThreshold'); + } + if (this.tickInterval <= 0 || this.tickInterval > this.warnThreshold) { + throw new Error('tickInterval must be > 0 and <= warnThreshold'); + } + } + + /** Set the OTel parent and, optionally, a runner that restores job AsyncLocalStorage. */ + setReportContext(context: Context | undefined, runner?: ReportContextRunner): void { + this.#reportContext = context; + this.#reportContextRunner = runner; + } + + /** Start the heartbeat. Idempotent. */ + start(): void { + if (this.#started || this.#closed) return; + this.#started = true; + this.#lastTickAt = performance.now(); + this.#lastCpuUsage = process.cpuUsage(); + this.#scheduleTick(); + } + + /** Stop the heartbeat. Idempotent. */ + stop(): void { + if (this.#closed) return; + this.#closed = true; + if (this.#timer) clearTimeout(this.#timer); + this.#timer = undefined; + } + + #scheduleTick(): void { + this.#timer = setTimeout(() => this.#onTick(), this.tickInterval); + this.#timer.unref(); + } + + #onTick(): void { + if (this.#closed) return; + const now = performance.now(); + const lag = now - (this.#lastTickAt + this.tickInterval); + const cpu = process.cpuUsage(); + const cpuTime = + (cpu.user - this.#lastCpuUsage.user + cpu.system - this.#lastCpuUsage.system) / 1000; + this.#lastTickAt = now; + this.#lastCpuUsage = cpu; + this.#scheduleTick(); + if (lag < this.warnThreshold) return; + this._report(this._buildReport(lag, cpuTime)); + } + + /** @internal */ + _buildReport(duration: number, cpuTime: number): BlockedReport { + return { + duration, + startedAt: Date.now() - duration, + warnThreshold: this.warnThreshold, + severity: duration >= this.errorThreshold ? 'error' : 'warning', + cpuTime, + }; + } + + /** @internal */ + _report(report: BlockedReport): void { + const now = performance.now(); + const emitSpan = this.#spanLimiter.allow(now); + const emitLog = this.#logLimiter.allow(now); + const emit = () => { + try { + recordEventLoopBlocked(report.duration / 1000, report.severity); + } catch (error) { + log().error({ error }, 'failed to record the blocked event loop metric'); + } + recordLoopStall(report.duration / 1000, report.startedAt + report.duration); + if (!emitSpan && !emitLog) return; + if (emitSpan && this.#emitSpans) { + this.#emitSpan(report, this.#spanLimiter.takeSuppressed()); + } + if (emitLog) this.#emitLog(report); + this._onReport?.(report); + }; + + try { + const run = () => + this.#reportContext ? otelContext.with(this.#reportContext, emit) : emit(); + this.#reportContextRunner ? this.#reportContextRunner(run) : run(); + } catch (error) { + log().error({ error }, 'failed to report a blocked event loop'); + } + } + + #emitSpan(report: BlockedReport, suppressed: number): void { + if (!getJobContext(false)) return; + const attributes: Attributes = { + [ATTR_BLOCKING_DURATION]: report.duration / 1000, + [ATTR_BLOCKING_THRESHOLD]: report.warnThreshold / 1000, + [ATTR_BLOCKING_SEVERITY]: report.severity, + [ATTR_BLOCKING_CPU_TIME]: report.cpuTime / 1000, + }; + if (suppressed) attributes[ATTR_BLOCKING_SUPPRESSED] = suppressed; + const span = tracer.startSpan({ + name: SPAN_NAME, + context: sessionRootContext() ?? this.#reportContext, + startTime: report.startedAt, + attributes, + }); + if (report.severity === 'error') { + span.setStatus({ + code: SpanStatusCode.ERROR, + message: `event loop blocked for ${report.duration.toFixed(0)}ms`, + }); + } + span.end(report.startedAt + report.duration); + } + + #emitLog(report: BlockedReport): void { + log().warn( + { + duration: Math.round(report.duration * 10) / 10, + threshold: report.warnThreshold, + cpuTime: Math.round(report.cpuTime * 10) / 10, + loop: this.#name, + }, + 'event loop blocked; synchronous work on the agent loop delays audio and turn handling, move it to a worker or an async client', + ); + } +} + +/** @internal */ +export function _tickIntervalFor(warnThreshold: number): number { + return Math.min( + Math.max(warnThreshold / TICKS_PER_WARN_THRESHOLD, MIN_TICK_INTERVAL), + MAX_TICK_INTERVAL, + ); +} + +interface MonitorGlobalState { + monitor?: EventLoopMonitor; +} + +const MONITOR_KEY = Symbol.for('@livekit/agents:eventLoopMonitor'); +const globals = globalThis as typeof globalThis & { [MONITOR_KEY]?: MonitorGlobalState }; +const monitorState = (globals[MONITOR_KEY] ??= {}); + +export interface StartMonitoringOptions { + thresholds?: LoopMonitorThresholds; + name?: string; + emitSpans?: boolean; +} + +/** Start the process event-loop monitor, or return undefined if disabled/already running. */ +export function startMonitoring( + options: StartMonitoringOptions = {}, +): EventLoopMonitor | undefined { + let thresholds = options.thresholds ?? LoopMonitorThresholds.fromEnv(); + if (!thresholds || monitorState.monitor) return undefined; + if (thresholds.warn < MIN_TICK_INTERVAL) { + log().warn( + { warnThreshold: thresholds.warn }, + 'loop monitor warn threshold raised to 20ms, the smallest it measures', + ); + thresholds = new LoopMonitorThresholds( + MIN_TICK_INTERVAL, + Math.max(thresholds.error, MIN_TICK_INTERVAL), + ); + } + const monitor = new EventLoopMonitor({ + warnThreshold: thresholds.warn, + errorThreshold: thresholds.error, + tickInterval: _tickIntervalFor(thresholds.warn), + name: options.name, + emitSpans: options.emitSpans, + }); + monitorState.monitor = monitor; + monitor.start(); + return monitor; +} + +export function stopMonitoring(monitor?: EventLoopMonitor): void { + if (monitor && monitorState.monitor !== monitor) return; + monitorState.monitor?.stop(); + monitorState.monitor = undefined; +} + +export function getMonitor(): EventLoopMonitor | undefined { + return monitorState.monitor; +} diff --git a/agents/src/telemetry/otel_metrics.ts b/agents/src/telemetry/otel_metrics.ts new file mode 100644 index 000000000..9e4c65fdf --- /dev/null +++ b/agents/src/telemetry/otel_metrics.ts @@ -0,0 +1,36 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { type Attributes, type Histogram, type MeterProvider, metrics } from '@opentelemetry/api'; +import { getJobContext } from '../job.js'; + +let meterProvider: MeterProvider | undefined; +let blockedDuration: Histogram | undefined; + +function eventLoopBlockedHistogram(): Histogram { + const currentProvider = metrics.getMeterProvider(); + if (currentProvider !== meterProvider || !blockedDuration) { + meterProvider = currentProvider; + blockedDuration = metrics + .getMeter('livekit-agents') + .createHistogram('lk.agents.event_loop.blocked_duration', { + unit: 's', + description: 'Duration of synchronous blocks detected on an agent event loop', + }); + } + return blockedDuration; +} + +/** Record a synchronous event-loop block in seconds. */ +export function recordEventLoopBlocked(duration: number, severity: string): void { + const ctx = getJobContext(false); + const attributes: Attributes = { severity }; + if (ctx) { + Object.assign(attributes, ctx._otelMetadata()); + const roomId = ctx.job.room?.sid; + if (roomId) attributes.room_id = roomId; + if (ctx.job.id) attributes.job_id = ctx.job.id; + if (ctx.job.agentName) attributes['lk.agent_name'] = ctx.job.agentName; + } + eventLoopBlockedHistogram().record(duration, attributes); +} diff --git a/agents/src/telemetry/session_context.ts b/agents/src/telemetry/session_context.ts new file mode 100644 index 000000000..b4237245f --- /dev/null +++ b/agents/src/telemetry/session_context.ts @@ -0,0 +1,27 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Context } from '@opentelemetry/api'; +import { getJobContext } from '../job.js'; +import type { AgentSession } from '../voice/agent_session.js'; + +type SessionWithLoopTelemetry = AgentSession & { + _recordLoopStall?: (durationInS: number, timestamp: number) => void; +}; + +/** The primary agent session in the active job, if one exists. @internal */ +export function primarySession(): AgentSession | undefined { + return getJobContext(false)?._primaryAgentSession; +} + +/** The primary session's root trace context, if the session is running. @internal */ +export function sessionRootContext(): Context | undefined { + return primarySession()?.rootSpanContext; +} + +/** Record a stall on the active session when a session integration is available. @internal */ +export function recordLoopStall(durationInS: number, timestamp: number): void { + const session = primarySession(); + if (!session) return; + (session as SessionWithLoopTelemetry)._recordLoopStall?.(durationInS, timestamp); +} diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts index 82a670bef..d5a43dbe6 100644 --- a/agents/src/telemetry/trace_types.test.ts +++ b/agents/src/telemetry/trace_types.test.ts @@ -163,6 +163,19 @@ const SAFE_KEYS = new Set([ 'lk.tts_metrics', 'lk.realtime_model_metrics', 'lk.e2e_latency', + // Event loop blocking + 'lk.blocking.duration', + 'lk.blocking.threshold', + 'lk.blocking.severity', + 'lk.blocking.task', + 'lk.blocking.stack', + 'lk.blocking.gc_time', + 'lk.blocking.cpu_time', + 'lk.blocking.import', + 'lk.blocking.suppressed', + 'lk.blocking.count', + 'lk.blocking.total_duration', + 'lk.blocking.max_duration', // OpenTelemetry GenAI attributes and event names 'gen_ai.operation.name', 'gen_ai.request.model', diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index 22d0c1830..c4d147e01 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -116,6 +116,23 @@ export const ATTR_REALTIME_MODEL_METRICS = 'lk.realtime_model_metrics'; /** End-to-end latency in seconds. */ export const ATTR_E2E_LATENCY = 'lk.e2e_latency'; +// Event loop blocking +/** Heartbeat lag in seconds. */ +export const ATTR_BLOCKING_DURATION = 'lk.blocking.duration'; +export const ATTR_BLOCKING_THRESHOLD = 'lk.blocking.threshold'; +export const ATTR_BLOCKING_SEVERITY = 'lk.blocking.severity'; +export const ATTR_BLOCKING_TASK = 'lk.blocking.task'; +export const ATTR_BLOCKING_STACK = 'lk.blocking.stack'; +export const ATTR_BLOCKING_GC_TIME = 'lk.blocking.gc_time'; +/** Process CPU consumed during the heartbeat interval, in seconds. */ +export const ATTR_BLOCKING_CPU_TIME = 'lk.blocking.cpu_time'; +export const ATTR_BLOCKING_IMPORT = 'lk.blocking.import'; +export const ATTR_BLOCKING_SUPPRESSED = 'lk.blocking.suppressed'; +// Summary attributes on agent_session. +export const ATTR_BLOCKING_COUNT = 'lk.blocking.count'; +export const ATTR_BLOCKING_TOTAL_DURATION = 'lk.blocking.total_duration'; +export const ATTR_BLOCKING_MAX_DURATION = 'lk.blocking.max_duration'; + // OpenTelemetry GenAI semantic conventions, mirroring the attribute registry of // https://github.com/open-telemetry/semantic-conventions-genai. Backends ingest these // directly, so the names must stay byte-for-byte identical to the registry. The ones the diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 9451a8614..1ccfbf9a3 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -11,6 +11,7 @@ import { type SpanOptions, type Tracer, type TracerProvider, + metrics, context as otelContext, trace, } from '@opentelemetry/api'; @@ -22,6 +23,11 @@ import { envDetector, resourceFromAttributes, } from '@opentelemetry/resources'; +import { + AggregationTemporality, + MeterProvider, + PeriodicExportingMetricReader, +} from '@opentelemetry/sdk-metrics'; import type { ReadableSpan, Span as SdkSpan, @@ -52,7 +58,7 @@ import { flushPinoLogs, initPinoCloudExporter } from './pino_otel_transport.js'; import { uploadRecording } from './recording_upload.js'; import { allowPiiFromEnv } from './redaction.js'; import { ATTR_AGENT_NAME, ATTR_CLOUD_AGENT_ID, ATTR_DEPLOYMENT_ID } from './trace_types.js'; -import { UploadGateTraceExporter, uploadGate } from './upload_gate.js'; +import { UploadGateMetricExporter, UploadGateTraceExporter, uploadGate } from './upload_gate.js'; export interface StartSpanOptions { /** Name of the span */ @@ -287,6 +293,72 @@ const customProviderConfigs = new WeakMap( /** Providers that already carry the in-process PII stripper — installed at most once. */ const piiRedactionInstalled = new WeakSet(); +let cloudMeterProvider: MeterProvider | undefined; +let cloudMetricsUnavailable = false; +let cloudMeterShutdownRegistered = false; + +function isNoopMeterProvider(provider: ReturnType): boolean { + // The API does not publicly export its singleton NoopMeterProvider. The constructor is the + // stable distinction available in OTel API 1.x, equivalent to checking the private proxy/no-op + // provider in the Python SDK. + return provider.constructor.name === 'NoopMeterProvider'; +} + +function setupCloudMetrics( + observabilityUrl: string, + headers: Record, + resource: ReturnType, +): MeterProvider | undefined { + if (cloudMeterProvider || cloudMetricsUnavailable) return cloudMeterProvider; + + const currentProvider = metrics.getMeterProvider(); + if (!isNoopMeterProvider(currentProvider)) { + // Metric readers are fixed when an SDK 2.x MeterProvider is constructed. Preserve an + // application-installed global provider rather than replacing it and breaking its exporter. + cloudMetricsUnavailable = true; + return undefined; + } + + const exporter = new UploadGateMetricExporter({ + url: `${observabilityUrl}/observability/metrics/otlp/v0`, + headers, + compression: CompressionAlgorithm.GZIP, + temporalityPreference: AggregationTemporality.DELTA, + }); + const provider = new MeterProvider({ + resource, + readers: [ + new PeriodicExportingMetricReader({ + exporter, + exportIntervalMillis: 30_000, + }), + ], + }); + + if (!metrics.setGlobalMeterProvider(provider)) { + // Another provider won the set-once global between the check and registration. Do not leave + // the orphaned periodic reader running. + void provider.shutdown().catch(() => undefined); + cloudMetricsUnavailable = true; + return undefined; + } + + cloudMeterProvider = provider; + if (!cloudMeterShutdownRegistered) { + cloudMeterShutdownRegistered = true; + process.once('beforeExit', async () => { + const ownedProvider = cloudMeterProvider; + cloudMeterProvider = undefined; + try { + await ownedProvider?.shutdown({ timeoutMillis: 10_000 }); + } catch (error) { + console.error('Failed to shut down cloud metrics:', error); + } + }); + } + return provider; +} + /** * Installs {@link PIIFilteringSpanProcessor} on a provider LiveKit does not own. * @@ -501,6 +573,26 @@ export async function setupCloudTracer( }), ); + // A meter provider has process lifetime and cannot carry room/job identity safely. Those + // fields are attached to each measurement by otel_metrics instead. + const meterResource = defaultResource() + .merge(detectResources({ detectors: [envDetector] })) + .merge( + resourceFromAttributes({ + [ATTR_SERVICE_NAME]: 'livekit-agents', + ...(agentName ? { [ATTR_AGENT_NAME]: agentName } : {}), + ...(cloudAgentId ? { [ATTR_CLOUD_AGENT_ID]: cloudAgentId } : {}), + ...(deploymentId ? { [ATTR_DEPLOYMENT_ID]: deploymentId } : {}), + }), + ); + const meterProvider = setupCloudMetrics(observabilityUrl, headers, meterResource); + if (meterProvider) { + const { getJobContext } = await import('../job.js'); + getJobContext(false)?.addShutdownCallback(() => + meterProvider.forceFlush({ timeoutMillis: 10_000 }), + ); + } + if (enableTraces) { const url = `${observabilityUrl}/observability/traces/otlp/v0`; const createCloudExporter = () => diff --git a/agents/src/telemetry/upload_gate.ts b/agents/src/telemetry/upload_gate.ts index d280224fc..32f14355b 100644 --- a/agents/src/telemetry/upload_gate.ts +++ b/agents/src/telemetry/upload_gate.ts @@ -2,8 +2,10 @@ // // SPDX-License-Identifier: Apache-2.0 import { type ExportResult, ExportResultCode } from '@opentelemetry/core'; +import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-proto'; import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto'; import { OTLPExporterError } from '@opentelemetry/otlp-exporter-base'; +import type { ResourceMetrics } from '@opentelemetry/sdk-metrics'; import type { ReadableSpan } from '@opentelemetry/sdk-trace-base'; import { log } from '../log_core.js'; import { restorePii } from './redaction.js'; @@ -120,7 +122,31 @@ export class UploadGateTraceExporter extends OTLPTraceExporter { } } +/** OTLP metric exporter that shares the recording-disabled upload gate. */ +export class UploadGateMetricExporter extends OTLPMetricExporter { + override export(items: ResourceMetrics, resultCallback: (result: ExportResult) => void): void { + const generation = uploadGate.generation; + if (uploadGate.disabled) { + resultCallback({ code: ExportResultCode.SUCCESS }); + return; + } + + super.export(items, (result) => { + if (isDisabledExport(result)) { + uploadGate.disable(generation); + resultCallback({ code: ExportResultCode.SUCCESS }); + return; + } + resultCallback(result); + }); + } +} + function isDisabledTraceExport(result: ExportResult): boolean { + return isDisabledExport(result); +} + +function isDisabledExport(result: ExportResult): boolean { const error = result.error; return ( result.code === ExportResultCode.FAILED && diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index 786bc8099..011d8a3da 100644 --- a/agents/src/voice/agent_session.ts +++ b/agents/src/voice/agent_session.ts @@ -135,6 +135,7 @@ import { setParticipantSpanAttributes } from './utils.js'; const SIP_RULE_ID_ATTR = 'sip.ruleID'; const DEFAULT_AEC_WARMUP_DURATION = 3000; +const LOOP_STALL_EVENT = 'event_loop_blocked'; export interface AgentSessionUsage { /** List of usage summaries, one per model/provider combination. */ @@ -576,6 +577,9 @@ export class AgentSession< private sessionSpan?: Span; private agentSpeakingSpan?: Span; + private loopStallCount = 0; + private loopStallTotal = 0; + private loopStallMax = 0; private _interruptionDetection?: InterruptionOptions['mode']; @@ -1003,6 +1007,9 @@ export class AgentSession< this.closing = false; this._usageCollector = new ModelUsageCollector(); + this.loopStallCount = 0; + this.loopStallTotal = 0; + this.loopStallMax = 0; const ctx = getJobContext(false); @@ -1886,6 +1893,28 @@ export class AgentSession< } } + /** @internal */ + _recordLoopStall(durationInS: number, timestampMs: number): void { + const span = this.sessionSpan; + if (!span?.isRecording()) { + return; + } + + this.loopStallCount += 1; + this.loopStallTotal += durationInS; + this.loopStallMax = Math.max(this.loopStallMax, durationInS); + span.addEvent( + LOOP_STALL_EVENT, + { [traceTypes.ATTR_BLOCKING_DURATION]: durationInS }, + timestampMs, + ); + span.setAttributes({ + [traceTypes.ATTR_BLOCKING_COUNT]: this.loopStallCount, + [traceTypes.ATTR_BLOCKING_TOTAL_DURATION]: this.loopStallTotal, + [traceTypes.ATTR_BLOCKING_MAX_DURATION]: this.loopStallMax, + }); + } + private _onUserInputTranscribed(ev: UserInputTranscribedEvent): void { if (ev.isFinal && this._userState !== 'speaking') { if (this._userState === 'away') { diff --git a/agents/src/worker.ts b/agents/src/worker.ts index e42683237..c1491b82b 100644 --- a/agents/src/worker.ts +++ b/agents/src/worker.ts @@ -30,6 +30,11 @@ import type { JobAcceptArguments, JobProcess, RunningJobInfo } from './job.js'; import { JobRequest } from './job.js'; import { DEFAULT_SESSION_END_TIMEOUT, validateSessionEndTimeout } from './job_lifecycle.js'; import { log } from './log.js'; +import { + type EventLoopMonitor, + startMonitoring, + stopMonitoring, +} from './telemetry/loop_monitor.js'; import { Future, rejectOnAbort } from './utils.js'; import { version } from './version.js'; @@ -347,6 +352,7 @@ export class AgentServer { #httpServer?: HTTPServer; #logger = log().child({ version }); #inferenceExecutor?: InferenceProcExecutor; + #loopMonitor?: EventLoopMonitor; /* @throws {@link MissingCredentialsError} if URL, API key or API secret are missing */ constructor(opts: ServerOptions) { @@ -461,6 +467,7 @@ export class AgentServer { this.#logger.info('starting worker'); this.#closed = false; + this.#loopMonitor = startMonitoring({ name: 'worker', emitSpans: false }); this.#procPool.start(); const workerWS = async () => { @@ -971,6 +978,8 @@ export class AgentServer { this.#logger.debug('shutting down worker'); this.#closed = true; + if (this.#loopMonitor) stopMonitoring(this.#loopMonitor); + this.#loopMonitor = undefined; await this.#inferenceExecutor?.close(); await this.#procPool.close(); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 03d103a0c..fb5c1a128 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -156,6 +156,9 @@ importers: '@opentelemetry/exporter-logs-otlp-proto': specifier: ^0.220.0 version: 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/exporter-metrics-otlp-proto': + specifier: ^0.220.0 + version: 0.220.0(@opentelemetry/api@1.9.0) '@opentelemetry/exporter-trace-otlp-proto': specifier: ^0.220.0 version: 0.220.0(@opentelemetry/api@1.9.0) @@ -171,6 +174,9 @@ importers: '@opentelemetry/sdk-logs': specifier: ^0.220.0 version: 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/sdk-metrics': + specifier: ^2.8.0 + version: 2.9.0(@opentelemetry/api@1.9.0) '@opentelemetry/sdk-trace-base': specifier: ^2.8.0 version: 2.9.0(@opentelemetry/api@1.9.0) @@ -2521,6 +2527,18 @@ packages: peerDependencies: '@opentelemetry/api': ^1.3.0 + '@opentelemetry/exporter-metrics-otlp-http@0.220.0': + resolution: {integrity: sha512-Yqt3RBw/bRVncaE9qIIhk4WfjbAQqXuP9FgAaU+IKPndnLEp/cUqZlSC324+bpmduRz7DoTjig8Ub0PeILWXUA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': ^1.3.0 + + '@opentelemetry/exporter-metrics-otlp-proto@0.220.0': + resolution: {integrity: sha512-lyO+IQBdSvqHN/ZOW/OzrSWemtfD+HgWngn+HBNLhjy0YrCQQTz0OE/kSekH2Pl340dn9DWzhqHdz5Eftr+HLA==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': ^1.3.0 + '@opentelemetry/exporter-trace-otlp-http@0.220.0': resolution: {integrity: sha512-/+ExB3lRkf+erv4PnoywyL7RHKITidxtUpUTS55k7OQ0dB42S7gEF1gry7swb9MSm1hYLUhJg4QQh9W8SpwwqA==} engines: {node: ^18.19.0 || >=20.6.0} @@ -6403,6 +6421,25 @@ snapshots: '@opentelemetry/otlp-transformer': 0.220.0(@opentelemetry/api@1.9.0) '@opentelemetry/sdk-logs': 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/exporter-metrics-otlp-http@0.220.0(@opentelemetry/api@1.9.0)': + dependencies: + '@opentelemetry/api': 1.9.0 + '@opentelemetry/core': 2.9.0(@opentelemetry/api@1.9.0) + '@opentelemetry/otlp-exporter-base': 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/otlp-transformer': 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/resources': 2.9.0(@opentelemetry/api@1.9.0) + '@opentelemetry/sdk-metrics': 2.9.0(@opentelemetry/api@1.9.0) + + '@opentelemetry/exporter-metrics-otlp-proto@0.220.0(@opentelemetry/api@1.9.0)': + dependencies: + '@opentelemetry/api': 1.9.0 + '@opentelemetry/core': 2.9.0(@opentelemetry/api@1.9.0) + '@opentelemetry/exporter-metrics-otlp-http': 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/otlp-exporter-base': 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/otlp-transformer': 0.220.0(@opentelemetry/api@1.9.0) + '@opentelemetry/resources': 2.9.0(@opentelemetry/api@1.9.0) + '@opentelemetry/sdk-metrics': 2.9.0(@opentelemetry/api@1.9.0) + '@opentelemetry/exporter-trace-otlp-http@0.220.0(@opentelemetry/api@1.9.0)': dependencies: '@opentelemetry/api': 1.9.0 From f8261bd5285abbcdedd9bb999a629085101f711f Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 19:00:04 -0700 Subject: [PATCH 2/6] update implementation, also flag host CPU contention --- agents/etc/agents.api.md | 48 ++-- agents/src/telemetry/loop_monitor.test.ts | 192 ++++++++++++-- agents/src/telemetry/loop_monitor.ts | 295 +++++++++++++++++----- agents/src/telemetry/otel_metrics.test.ts | 71 ++++++ agents/src/telemetry/otel_metrics.ts | 6 +- agents/src/telemetry/rate_limiter.ts | 31 +++ agents/src/telemetry/session_context.ts | 6 +- agents/src/telemetry/trace_types.test.ts | 1 + agents/src/telemetry/trace_types.ts | 12 +- agents/src/telemetry/traces.ts | 22 +- agents/src/voice/agent_session.ts | 7 +- 11 files changed, 558 insertions(+), 133 deletions(-) create mode 100644 agents/src/telemetry/otel_metrics.test.ts create mode 100644 agents/src/telemetry/rate_limiter.ts diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index 51600b358..4d341a7d5 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -563,7 +563,7 @@ export class AgentSession extends AgentSession_base // @internal (undocumented) _recorderIO?: RecorderIO; // @internal (undocumented) - _recordLoopStall(durationInS: number, timestampMs: number): void; + _recordLoopStall(durationInS: number, timestampMs: number, cause: string): void; // @internal _redactionEnabled: boolean; // (undocumented) @@ -1144,6 +1144,11 @@ const ATTR_AMD_SPEECH_DURATION = "lk.amd.speech_duration"; // @public (undocumented) const ATTR_AMD_TRANSCRIPT = "lk.pii.amd.transcript"; +// Warning: (ae-missing-release-tag) "ATTR_BLOCKING_CAUSE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const ATTR_BLOCKING_CAUSE = "lk.blocking.cause"; + // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_COUNT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1161,12 +1166,12 @@ const ATTR_BLOCKING_DURATION = "lk.blocking.duration"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_GC_TIME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_BLOCKING_GC_TIME = "lk.blocking.gc_time"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_IMPORT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_BLOCKING_IMPORT = "lk.blocking.import"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_MAX_DURATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1181,7 +1186,7 @@ const ATTR_BLOCKING_SEVERITY = "lk.blocking.severity"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_STACK" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_BLOCKING_STACK = "lk.blocking.stack"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_SUPPRESSED" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1191,7 +1196,7 @@ const ATTR_BLOCKING_SUPPRESSED = "lk.blocking.suppressed"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_TASK" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_BLOCKING_TASK = "lk.blocking.task"; // Warning: (ae-missing-release-tag) "ATTR_BLOCKING_THRESHOLD" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -2139,13 +2144,16 @@ declare namespace beta { // // @public (undocumented) interface BlockedReport { + cause: LoopStallCause; cpuTime: number; duration: number; + gcTime: number; // (undocumented) severity: LoopMonitorSeverity; startedAt: number; // (undocumented) warnThreshold: number; + watchdogGap: number; } // Warning: (ae-missing-release-tag) "BufferedSentenceStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -3533,14 +3541,9 @@ class EventAssert { // @public class EventLoopMonitor { constructor(options?: EventLoopMonitorOptions); - // @internal (undocumented) - _buildReport(duration: number, cpuTime: number): BlockedReport; // (undocumented) readonly errorThreshold: number; - // @internal - _onReport?: (report: BlockedReport) => void; - // @internal (undocumented) - _report(report: BlockedReport): void; + onReport?: (report: BlockedReport) => void; setReportContext(context: Context | undefined, runner?: ReportContextRunner): void; start(): void; stop(): void; @@ -3548,6 +3551,7 @@ class EventLoopMonitor { readonly tickInterval: number; // (undocumented) readonly warnThreshold: number; + get watchdogActive(): boolean; } // Warning: (ae-missing-release-tag) "EventLoopMonitorOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -3564,6 +3568,7 @@ interface EventLoopMonitorOptions { tickInterval?: number; // (undocumented) warnThreshold?: number; + watchdog?: boolean; } // Warning: (ae-missing-release-tag) "EventRangeAssert" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -5471,7 +5476,6 @@ export function loopAudioFramesFromFile(filePath: string, options?: AudioDecodeO declare namespace loopMonitor { export { - _tickIntervalFor, startMonitoring, stopMonitoring, getMonitor, @@ -5484,9 +5488,9 @@ declare namespace loopMonitor { MAX_LOGS_PER_MINUTE, SPAN_NAME, LoopMonitorSeverity, + LoopStallCause, BlockedReport, LoopMonitorThresholds, - _RateLimiter, EventLoopMonitorOptions, ReportContextRunner, EventLoopMonitor, @@ -5512,6 +5516,9 @@ class LoopMonitorThresholds { readonly warn: number; } +// @public +type LoopStallCause = 'code' | 'host'; + // Warning: (ae-missing-release-tag) "markInferenceSpanRecorded" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -6169,15 +6176,6 @@ export interface QueueAudioOutputClearEvent { // @public export type QueueAudioOutputItem = AudioFrame | AudioSegmentEnd; -// @internal (undocumented) -class _RateLimiter { - constructor(limit: number); - // (undocumented) - allow(now: number): boolean; - // (undocumented) - takeSuppressed(): number; -} - // Warning: (ae-missing-release-tag) "readStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -7587,6 +7585,8 @@ interface StartMonitoringOptions { name?: string; // (undocumented) thresholds?: LoopMonitorThresholds; + // (undocumented) + watchdog?: boolean; } // Warning: (ae-missing-release-tag) "startSoon" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -8359,9 +8359,6 @@ class ThresholdOptions { // @public type ThresholdOverride = number | Record | undefined; -// @internal (undocumented) -function _tickIntervalFor(warnThreshold: number): number; - // Warning: (ae-missing-release-tag) "TimedString" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -8751,6 +8748,7 @@ declare namespace traceTypes { ATTR_BLOCKING_DURATION, ATTR_BLOCKING_THRESHOLD, ATTR_BLOCKING_SEVERITY, + ATTR_BLOCKING_CAUSE, ATTR_BLOCKING_TASK, ATTR_BLOCKING_STACK, ATTR_BLOCKING_GC_TIME, diff --git a/agents/src/telemetry/loop_monitor.test.ts b/agents/src/telemetry/loop_monitor.test.ts index 749c2cb77..a4284cb00 100644 --- a/agents/src/telemetry/loop_monitor.test.ts +++ b/agents/src/telemetry/loop_monitor.test.ts @@ -7,6 +7,7 @@ import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; import { Worker } from 'node:worker_threads'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { type JobContext, getJobContext, runWithJobContext } from '../job.js'; +import { log } from '../log.js'; import { type BlockedReport, DEFAULT_ERROR_THRESHOLD, @@ -15,18 +16,20 @@ import { ENV_WARN_THRESHOLD_MS, EventLoopMonitor, LoopMonitorThresholds, + MAX_LOGS_PER_MINUTE, MAX_SPANS_PER_MINUTE, SPAN_NAME, - _RateLimiter, - _tickIntervalFor, getMonitor, startMonitoring, stopMonitoring, } from './loop_monitor.js'; import * as otelMetrics from './otel_metrics.js'; +import { RateLimiter } from './rate_limiter.js'; import { + ATTR_BLOCKING_CAUSE, ATTR_BLOCKING_CPU_TIME, ATTR_BLOCKING_DURATION, + ATTR_BLOCKING_GC_TIME, ATTR_BLOCKING_SEVERITY, ATTR_BLOCKING_THRESHOLD, } from './trace_types.js'; @@ -40,10 +43,29 @@ function blockLoop(duration: number): void { Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, duration); } +/** Block the loop while allocating heavily, so V8 has to collect during the block. */ +function blockLoopWithGarbage(duration: number): void { + const until = performance.now() + duration; + let garbage: unknown[] = []; + while (performance.now() < until) { + garbage.push(new Array(1000).fill({ x: 1 })); + if (garbage.length > 20_000) garbage = []; + } +} + function settle(): Promise { return new Promise((resolve) => setTimeout(resolve, TICK * 4)); } +/** Wait until the watchdog thread has woken at least once, so it can vouch for the process. */ +async function watchdogReady(monitor: EventLoopMonitor): Promise { + const deadline = Date.now() + 2000; + while (!monitor.watchdogActive && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + await new Promise((resolve) => setTimeout(resolve, WARN)); +} + function fakeJob(session?: unknown): JobContext { return { _primaryAgentSession: session, @@ -51,6 +73,13 @@ function fakeJob(session?: unknown): JobContext { } as unknown as JobContext; } +/** Reports that blame code on the loop. A noisy CI host can deschedule the test process too. */ +function codeReports(reports: BlockedReport[]): BlockedReport[] { + return reports.filter((report) => report.cause === 'code'); +} + +const timings = (cpuTime = 20, watchdogGap = 0, gcTime = 0) => ({ cpuTime, gcTime, watchdogGap }); + describe.sequential('event loop monitor', () => { let exporter: InMemorySpanExporter; let provider: NodeTracerProvider; @@ -70,7 +99,7 @@ describe.sequential('event loop monitor', () => { errorThreshold: ERROR, tickInterval: TICK, }); - monitor._onReport = (report) => reports.push(report); + monitor.onReport = (report) => reports.push(report); sessionRoot = tracer.startSpan({ name: 'agent_session' }); const session = { rootSpanContext: trace.setSpan(ROOT_CONTEXT, sessionRoot) }; const job = fakeJob(session); @@ -78,7 +107,7 @@ describe.sequential('event loop monitor', () => { runWithJobContext(job, fn), ); monitor.start(); - await new Promise((resolve) => setTimeout(resolve, WARN)); + await watchdogReady(monitor); }); afterEach(async () => { @@ -110,6 +139,12 @@ describe.sequential('event loop monitor', () => { (span!.endTime[1] - span!.startTime[1]) / 1e6; expect(elapsed).toBeCloseTo(duration * 1000, 3); expect(span!.attributes[ATTR_BLOCKING_CPU_TIME]).toBeTypeOf('number'); + expect(span!.attributes[ATTR_BLOCKING_GC_TIME]).toBe(0); + expect(span!.attributes[ATTR_BLOCKING_CAUSE]).toBe('code'); + // the watchdog thread kept running while the loop thread waited: the process was scheduled + const [report] = reports; + expect(report!.cause).toBe('code'); + expect(report!.watchdogGap).toBeLessThan(report!.duration * 0.5); }); it('reports a block between thresholds as a warning', async () => { @@ -120,13 +155,27 @@ describe.sequential('event loop monitor', () => { expect(span!.status.code).toBe(SpanStatusCode.UNSET); }); + it('attributes garbage collection pauses inside a block', async () => { + blockLoopWithGarbage(200); + await settle(); + // the longest stall is the block above; a loaded host can add shorter ones around it + const report = codeReports(reports).sort((a, b) => b.duration - a.duration)[0]; + expect(report).toBeDefined(); + expect(report!.gcTime).toBeGreaterThan(0); + expect(report!.gcTime).toBeLessThanOrEqual(report!.duration); + const span = blockedSpans().find( + (candidate) => candidate.attributes[ATTR_BLOCKING_DURATION] === report!.duration / 1000, + ); + expect(span!.attributes[ATTR_BLOCKING_GC_TIME]).toBeCloseTo(report!.gcTime / 1000, 6); + }); + it('does not report cooperative work', async () => { for (let i = 0; i < 40; i++) { blockLoop(2); await new Promise((resolve) => setImmediate(resolve)); } await new Promise((resolve) => setTimeout(resolve, WARN * 2)); - expect(reports).toEqual([]); + expect(codeReports(reports)).toEqual([]); }); it('parents a pre-session stall to the report context', async () => { @@ -153,13 +202,16 @@ describe.sequential('event loop monitor', () => { it('records every stall on the active session', () => { monitor.stop(); - const seen: number[] = []; - const job = fakeJob({ _recordLoopStall: (duration: number) => seen.push(duration) }); + const seen: [number, string][] = []; + const job = fakeJob({ + _recordLoopStall: (duration: number, _timestamp: number, cause: string) => + seen.push([duration, cause]), + }); monitor.setReportContext(undefined, (fn) => runWithJobContext(job, fn)); - const report = monitor._buildReport(100, 20); - for (let i = 0; i < 40; i++) monitor._report(report); + const report = monitor['buildReport'](100, timings()); + for (let i = 0; i < 40; i++) monitor['report'](report); expect(seen).toHaveLength(40); - expect(seen.every((duration) => duration === 0.1)).toBe(true); + expect(seen.every(([duration, cause]) => duration === 0.1 && cause === 'code')).toBe(true); }); it('stops idempotently and stays quiet', async () => { @@ -172,7 +224,7 @@ describe.sequential('event loop monitor', () => { it('does not report an idle loop', async () => { await new Promise((resolve) => setTimeout(resolve, WARN * 12)); - expect(reports).toEqual([]); + expect(codeReports(reports)).toEqual([]); }); it('does not report blocking work outside the event loop', async () => { @@ -185,7 +237,7 @@ describe.sequential('event loop monitor', () => { worker.once('exit', () => resolve()); }); await settle(); - expect(reports).toEqual([]); + expect(codeReports(reports)).toEqual([]); }); it('keeps spans disabled in worker mode', async () => { @@ -198,9 +250,9 @@ describe.sequential('event loop monitor', () => { emitSpans: false, }); const workerReports: BlockedReport[] = []; - workerMonitor._onReport = (report) => workerReports.push(report); + workerMonitor.onReport = (report) => workerReports.push(report); workerMonitor.start(); - await new Promise((resolve) => setTimeout(resolve, WARN)); + await watchdogReady(workerMonitor); blockLoop(80); await settle(); workerMonitor.stop(); @@ -211,10 +263,10 @@ describe.sequential('event loop monitor', () => { it('records the metric for every stall past span rate limits in seconds', () => { monitor.stop(); const record = vi.spyOn(otelMetrics, 'recordEventLoopBlocked').mockImplementation(() => {}); - const report = monitor._buildReport(100, 20); - for (let i = 0; i < 40; i++) monitor._report(report); + const report = monitor['buildReport'](100, timings()); + for (let i = 0; i < 40; i++) monitor['report'](report); expect(record).toHaveBeenCalledTimes(40); - expect(record).toHaveBeenCalledWith(0.1, 'warning'); + expect(record).toHaveBeenCalledWith(0.1, 'warning', 'code'); expect(reports).toHaveLength(MAX_SPANS_PER_MINUTE); }); @@ -227,10 +279,87 @@ describe.sequential('event loop monitor', () => { }); const context = trace.setSpan(ROOT_CONTEXT, sessionRoot); monitor.setReportContext(context, (fn) => runWithJobContext(job, fn)); - for (let i = 0; i < 40; i++) monitor._report(monitor._buildReport(100, 20)); + for (let i = 0; i < 40; i++) monitor['report'](monitor['buildReport'](100, timings())); expect(jobs).toHaveLength(40); expect(jobs.every((value) => value === job)).toBe(true); }); + + describe('host contention', () => { + it('is told apart from blocking code by the watchdog and CPU time', () => { + monitor.stop(); + // watchdog late by most of the stall and (nearly) no CPU burned: nothing ran + const descheduled = monitor['buildReport'](400, timings(10, 300)); + expect(descheduled.cause).toBe('host'); + // watchdog on time: the loop thread alone was stuck, in a blocking wait + expect(monitor['buildReport'](400, timings(10, 4)).cause).toBe('code'); + // watchdog starved but the process was busy the whole time: code kept the loop + expect(monitor['buildReport'](400, timings(380, 300)).cause).toBe('code'); + }); + + it('keeps the severity of its impact', () => { + monitor.stop(); + // a stall past the error threshold delays audio the same however it came about + expect(monitor['buildReport'](400, timings(10, 300)).severity).toBe('error'); + expect(monitor['buildReport'](70, timings(1, 60)).severity).toBe('warning'); + }); + + it('is flagged as host contention on the span, the metric, the session, and the log', () => { + monitor.stop(); + const warn = vi.spyOn(log(), 'warn').mockImplementation(() => undefined); + const record = vi.spyOn(otelMetrics, 'recordEventLoopBlocked').mockImplementation(() => {}); + const stalls: string[] = []; + const session = { + rootSpanContext: trace.setSpan(ROOT_CONTEXT, sessionRoot), + _recordLoopStall: (_duration: number, _timestamp: number, cause: string) => + stalls.push(cause), + }; + monitor.setReportContext(trace.setSpan(ROOT_CONTEXT, sessionRoot), (fn) => + runWithJobContext(fakeJob(session), fn), + ); + monitor['report'](monitor['buildReport'](400, timings(10, 300))); + const [span] = blockedSpans(); + expect(span!.attributes[ATTR_BLOCKING_CAUSE]).toBe('host'); + expect(span!.status.code).toBe(SpanStatusCode.ERROR); + expect(span!.status.message).toContain('not scheduled'); + expect(record).toHaveBeenCalledWith(0.4, 'error', 'host'); + expect(stalls).toEqual(['host']); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0]![0]).toMatchObject({ cause: 'host' }); + expect(warn.mock.calls[0]![1]).toContain('CPU contention or a container CPU quota'); + expect(warn.mock.calls[0]![1]).not.toContain('synchronous work'); + }); + + it('has its own log quota so neither cause silences the other', () => { + monitor.stop(); + const warn = vi.spyOn(log(), 'warn').mockImplementation(() => undefined); + const host = monitor['buildReport'](400, timings(10, 300)); + for (let i = 0; i < 10; i++) monitor['report'](host); + expect(warn).toHaveBeenCalledTimes(MAX_LOGS_PER_MINUTE); + // a genuine block right after still has its full quota + monitor['report'](monitor['buildReport'](400, timings(10, 4))); + expect(warn).toHaveBeenCalledTimes(MAX_LOGS_PER_MINUTE + 1); + expect(warn.mock.calls.at(-1)![1]).toContain('synchronous work'); + }); + + it('is named as a possible cause when no watchdog is running', async () => { + monitor.stop(); + const warn = vi.spyOn(log(), 'warn').mockImplementation(() => undefined); + const bare = new EventLoopMonitor({ + warnThreshold: WARN, + errorThreshold: ERROR, + tickInterval: TICK, + watchdog: false, + }); + bare.start(); + expect(bare.watchdogActive).toBe(false); + const report = bare['buildReport'](400, timings(10, 0)); + expect(report.cause).toBe('code'); + bare['report'](report); + bare.stop(); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0]![1]).toContain('did not schedule the process'); + }); + }); }); describe.sequential('event loop monitor helpers', () => { @@ -240,7 +369,7 @@ describe.sequential('event loop monitor helpers', () => { }); it('rate limiter counts suppressed reports and uses a rolling window', () => { - const limiter = new _RateLimiter(2); + const limiter = new RateLimiter(2); expect(limiter.allow(100_000)).toBe(true); expect(limiter.allow(100_100)).toBe(true); expect(limiter.allow(100_200)).toBe(false); @@ -288,19 +417,34 @@ describe.sequential('event loop monitor helpers', () => { expect(() => new EventLoopMonitor({ warnThreshold: 0 })).toThrow(); expect(() => new EventLoopMonitor({ warnThreshold: 100, errorThreshold: 50 })).toThrow(); expect(() => new EventLoopMonitor({ warnThreshold: 10, tickInterval: 50 })).toThrow(); + // a NaN tick interval would become a zero-delay timer and spin the loop + expect(() => new EventLoopMonitor({ warnThreshold: NaN })).toThrow(); + expect(() => new EventLoopMonitor({ warnThreshold: 100, errorThreshold: NaN })).toThrow(); + expect(() => new EventLoopMonitor({ warnThreshold: 100, tickInterval: NaN })).toThrow(); + expect(() => new EventLoopMonitor({ warnThreshold: Infinity })).toThrow(); }); - it('uses a bounded fifth of the warning threshold', () => { - expect(_tickIntervalFor(100)).toBe(20); - expect(_tickIntervalFor(250)).toBe(50); - expect(_tickIntervalFor(1000)).toBe(50); - expect(_tickIntervalFor(50)).toBe(20); + it('uses a bounded fifth of the warning threshold as the tick interval', () => { + const tickFor = (warn: number) => { + const monitor = startMonitoring({ + thresholds: new LoopMonitorThresholds(warn, warn * 5), + watchdog: false, + }); + const tick = monitor!.tickInterval; + stopMonitoring(); + return tick; + }; + expect(tickFor(100)).toBe(20); + expect(tickFor(250)).toBe(50); + expect(tickFor(1000)).toBe(50); + expect(tickFor(50)).toBe(20); }); it('raises a warning threshold below the tick floor', () => { const monitor = startMonitoring({ thresholds: new LoopMonitorThresholds(5, 10), emitSpans: false, + watchdog: false, }); expect(monitor?.warnThreshold).toBe(20); expect(monitor?.errorThreshold).toBe(20); diff --git a/agents/src/telemetry/loop_monitor.ts b/agents/src/telemetry/loop_monitor.ts index a83db363b..28d302498 100644 --- a/agents/src/telemetry/loop_monitor.ts +++ b/agents/src/telemetry/loop_monitor.ts @@ -7,13 +7,18 @@ import { SpanStatusCode, context as otelContext, } from '@opentelemetry/api'; +import { PerformanceObserver } from 'node:perf_hooks'; +import { Worker } from 'node:worker_threads'; import { getJobContext } from '../job.js'; import { log } from '../log.js'; import { recordEventLoopBlocked } from './otel_metrics.js'; +import { RateLimiter } from './rate_limiter.js'; import { recordLoopStall, sessionRootContext } from './session_context.js'; import { + ATTR_BLOCKING_CAUSE, ATTR_BLOCKING_CPU_TIME, ATTR_BLOCKING_DURATION, + ATTR_BLOCKING_GC_TIME, ATTR_BLOCKING_SEVERITY, ATTR_BLOCKING_SUPPRESSED, ATTR_BLOCKING_THRESHOLD, @@ -29,12 +34,18 @@ const MAX_TICK_INTERVAL = 50; export const ENV_WARN_THRESHOLD_MS = 'LIVEKIT_AGENTS_LOOP_BLOCK_WARN_MS'; export const ENV_ERROR_THRESHOLD_MS = 'LIVEKIT_AGENTS_LOOP_BLOCK_ERROR_MS'; -export const MAX_SPANS_PER_MINUTE = 30; +export const MAX_SPANS_PER_MINUTE = 6; export const MAX_LOGS_PER_MINUTE = 5; export const SPAN_NAME = 'event_loop_blocked'; export type LoopMonitorSeverity = 'warning' | 'error'; +/** + * What kept the loop from running: `code` is synchronous work on the loop thread, `host` is the + * whole process not being scheduled (CPU contention on the host, a container CPU quota). + */ +export type LoopStallCause = 'code' | 'host'; + export interface BlockedReport { /** Heartbeat lag in milliseconds. */ duration: number; @@ -42,8 +53,22 @@ export interface BlockedReport { startedAt: number; warnThreshold: number; severity: LoopMonitorSeverity; - /** Process CPU consumed during the heartbeat interval, in milliseconds. */ + /** Garbage-collection pause time observed during the stall, in milliseconds. */ + gcTime: number; + /** + * CPU consumed during the stall, in milliseconds. This is process-wide (every thread, including + * the libuv pool and the native media threads), not the event loop thread alone: Node does not + * expose per-thread CPU accounting. It can therefore exceed the stall duration. + */ cpuTime: number; + /** How late the watchdog thread woke during the stall, in milliseconds. 0 without a watchdog. */ + watchdogGap: number; + /** + * `host` when the whole process stopped running (host CPU contention, a container CPU quota, a + * suspended machine) rather than code blocking the loop. Without a watchdog every stall is + * attributed to `code`. + */ + cause: LoopStallCause; } export class LoopMonitorThresholds { @@ -81,44 +106,48 @@ function envMilliseconds(env: NodeJS.ProcessEnv, name: string, defaultValue: num return value; } -/** @internal */ -export class _RateLimiter { - readonly #limit: number; - readonly #events: number[] = []; - #suppressed = 0; - - constructor(limit: number) { - this.#limit = limit; - } - - allow(now: number): boolean { - const windowStart = now - 60_000; - while (this.#events.length && this.#events[0]! < windowStart) this.#events.shift(); - if (this.#events.length >= this.#limit) { - this.#suppressed++; - return false; - } - this.#events.push(now); - return true; - } - - takeSuppressed(): number { - const suppressed = this.#suppressed; - this.#suppressed = 0; - return suppressed; - } -} - export interface EventLoopMonitorOptions { warnThreshold?: number; errorThreshold?: number; tickInterval?: number; name?: string; emitSpans?: boolean; + /** + * Run a watchdog worker thread whose own late wake-ups reveal when the whole process was not + * scheduled, so host contention is reported as such. Without it every stall is attributed to + * code on the loop. Default true. + */ + watchdog?: boolean; } export type ReportContextRunner = (fn: () => T) => T; +// Slots of the SharedArrayBuffer shared with the watchdog thread. Values are Date.now() +// milliseconds as BigInt, since performance.now() has a different origin in every thread. +const WD_LAST_WAKE = 0; +const WD_LATE_AT = 1; +const WD_LATE_GAP = 2; + +// The watchdog is an independent event loop, so a synchronous block on the main thread does not +// delay it. When it wakes late too, the process itself was not running. It keeps the latest +// wake-up time and the largest late wake-up it has seen since the main thread last looked. +const WATCHDOG_SOURCE = ` +const { workerData } = require('node:worker_threads'); +const state = new BigInt64Array(workerData.shared); +const interval = workerData.interval; +let before = Date.now(); +setInterval(() => { + const now = Date.now(); + const gap = now - before - interval; + before = now; + if (gap > Number(Atomics.load(state, ${WD_LATE_GAP}))) { + Atomics.store(state, ${WD_LATE_AT}, BigInt(now)); + Atomics.store(state, ${WD_LATE_GAP}, BigInt(gap)); + } + Atomics.store(state, ${WD_LAST_WAKE}, BigInt(now)); +}, interval); +`; + /** Detect synchronous work that prevents the Node event loop from servicing timers. */ export class EventLoopMonitor { readonly warnThreshold: number; @@ -126,18 +155,24 @@ export class EventLoopMonitor { readonly tickInterval: number; readonly #name: string; readonly #emitSpans: boolean; - readonly #spanLimiter = new _RateLimiter(MAX_SPANS_PER_MINUTE); - readonly #logLimiter = new _RateLimiter(MAX_LOGS_PER_MINUTE); + readonly #useWatchdog: boolean; + readonly #spanLimiter = new RateLimiter(MAX_SPANS_PER_MINUTE); + readonly #logLimiter = new RateLimiter(MAX_LOGS_PER_MINUTE); + readonly #hostLogLimiter = new RateLimiter(MAX_LOGS_PER_MINUTE); #timer?: NodeJS.Timeout; #started = false; #closed = false; #lastTickAt = 0; #lastCpuUsage: NodeJS.CpuUsage = { user: 0, system: 0 }; + #gcObserver?: PerformanceObserver; + #gcTime = 0; + #watchdog?: Worker; + #watchdogState?: BigInt64Array; #reportContext?: Context; #reportContextRunner?: ReportContextRunner; - /** Tests and integrations may observe reports without going through OpenTelemetry. @internal */ - _onReport?: (report: BlockedReport) => void; + /** Observe every report that produced a span or a log, without going through OpenTelemetry. */ + onReport?: (report: BlockedReport) => void; constructor(options: EventLoopMonitorOptions = {}) { this.warnThreshold = options.warnThreshold ?? DEFAULT_WARN_THRESHOLD; @@ -145,15 +180,27 @@ export class EventLoopMonitor { this.tickInterval = options.tickInterval ?? DEFAULT_TICK_INTERVAL; this.#name = options.name ?? 'event-loop'; this.#emitSpans = options.emitSpans ?? true; - if (this.warnThreshold <= 0) throw new Error('warnThreshold must be > 0'); - if (this.errorThreshold < this.warnThreshold) { - throw new Error('errorThreshold must be >= warnThreshold'); + this.#useWatchdog = options.watchdog ?? true; + if (!Number.isFinite(this.warnThreshold) || this.warnThreshold <= 0) { + throw new Error('warnThreshold must be finite and > 0'); + } + if (!Number.isFinite(this.errorThreshold) || this.errorThreshold < this.warnThreshold) { + throw new Error('errorThreshold must be finite and >= warnThreshold'); } - if (this.tickInterval <= 0 || this.tickInterval > this.warnThreshold) { - throw new Error('tickInterval must be > 0 and <= warnThreshold'); + if ( + !Number.isFinite(this.tickInterval) || + this.tickInterval <= 0 || + this.tickInterval > this.warnThreshold + ) { + throw new Error('tickInterval must be finite, > 0, and <= warnThreshold'); } } + /** Whether the watchdog thread is running, i.e. host contention can be told apart from code. */ + get watchdogActive(): boolean { + return this.#watchdogState !== undefined; + } + /** Set the OTel parent and, optionally, a runner that restores job AsyncLocalStorage. */ setReportContext(context: Context | undefined, runner?: ReportContextRunner): void { this.#reportContext = context; @@ -166,15 +213,63 @@ export class EventLoopMonitor { this.#started = true; this.#lastTickAt = performance.now(); this.#lastCpuUsage = process.cpuUsage(); + this.#startGcObserver(); + if (this.#useWatchdog) this.#startWatchdog(); this.#scheduleTick(); } - /** Stop the heartbeat. Idempotent. */ + /** Stop the heartbeat and the watchdog. Idempotent. */ stop(): void { if (this.#closed) return; this.#closed = true; if (this.#timer) clearTimeout(this.#timer); this.#timer = undefined; + this.#gcObserver?.disconnect(); + this.#gcObserver = undefined; + const watchdog = this.#watchdog; + this.#watchdog = undefined; + this.#watchdogState = undefined; + void watchdog?.terminate().catch(() => undefined); + } + + #startGcObserver(): void { + try { + this.#gcObserver = new PerformanceObserver((list) => { + for (const entry of list.getEntries()) this.#gcTime += entry.duration; + }); + this.#gcObserver.observe({ entryTypes: ['gc'] }); + } catch (error) { + this.#gcObserver = undefined; + log().debug({ error }, 'event loop monitor cannot observe garbage collection'); + } + } + + #startWatchdog(): void { + const shared = new SharedArrayBuffer(3 * BigInt64Array.BYTES_PER_ELEMENT); + try { + const watchdog = new Worker(WATCHDOG_SOURCE, { + eval: true, + name: `livekit-loop-monitor-${this.#name}`, + workerData: { shared, interval: this.tickInterval }, + }); + // the watchdog must not keep the process alive, nor take it down + watchdog.unref(); + watchdog.on('error', (error) => { + log().debug({ error }, 'event loop watchdog failed'); + this.#dropWatchdog(watchdog); + }); + watchdog.on('exit', () => this.#dropWatchdog(watchdog)); + this.#watchdog = watchdog; + this.#watchdogState = new BigInt64Array(shared); + } catch (error) { + log().debug({ error }, 'event loop watchdog could not start'); + } + } + + #dropWatchdog(watchdog: Worker): void { + if (this.#watchdog !== watchdog) return; + this.#watchdog = undefined; + this.#watchdogState = undefined; } #scheduleTick(): void { @@ -189,42 +284,84 @@ export class EventLoopMonitor { const cpu = process.cpuUsage(); const cpuTime = (cpu.user - this.#lastCpuUsage.user + cpu.system - this.#lastCpuUsage.system) / 1000; + const gcTime = this.#gcTime; + this.#gcTime = 0; + // the window opens at the last on-time tick, expressed on the watchdog's wall clock + const windowStart = Date.now() - (now - this.#lastTickAt); this.#lastTickAt = now; this.#lastCpuUsage = cpu; + const watchdogGap = this.#consumeWatchdogGap(windowStart); this.#scheduleTick(); if (lag < this.warnThreshold) return; - this._report(this._buildReport(lag, cpuTime)); + this.report(this.buildReport(lag, { cpuTime, gcTime, watchdogGap })); + } + + /** + * How long the watchdog thread was kept from running during this tick's window. + * + * Two sources, since the two threads race when a descheduled process resumes: the watchdog's + * recorded late wake-up if it fell inside the window (an older one belongs to a stall already + * reported), and the time since its last run if it has not run since before the window opened. + */ + #consumeWatchdogGap(windowStart: number): number { + const state = this.#watchdogState; + if (!state) return 0; + const lastWake = Number(Atomics.load(state, WD_LAST_WAKE)); + const lateAt = Number(Atomics.exchange(state, WD_LATE_AT, 0n)); + const lateGap = Number(Atomics.exchange(state, WD_LATE_GAP, 0n)); + if (lastWake === 0) return 0; // the watchdog has not woken yet (it was just started) + let gap = 0; + if (lateAt >= windowStart) gap = lateGap; + if (lastWake < windowStart) gap = Math.max(gap, Date.now() - lastWake - this.tickInterval); + return Math.max(gap, 0); } - /** @internal */ - _buildReport(duration: number, cpuTime: number): BlockedReport { + private buildReport( + lag: number, + timings: { cpuTime: number; gcTime: number; watchdogGap: number }, + ): BlockedReport { + // the watchdog is an independent thread: if it too woke late by most of the stall, the + // process was not being scheduled (host contention, CPU quota, a suspended machine). A + // descheduled process burns no CPU, which rules out the loop having been busy instead. + const watchdogStarved = timings.watchdogGap >= lag * 0.5; + const processDescheduled = watchdogStarved && timings.cpuTime < lag * 0.5; return { - duration, - startedAt: Date.now() - duration, + duration: lag, + // the block started no earlier than the last on-time tick + startedAt: Date.now() - lag, warnThreshold: this.warnThreshold, - severity: duration >= this.errorThreshold ? 'error' : 'warning', - cpuTime, + // severity measures the impact on the session, whatever the cause: audio and turn handling + // were delayed either way. The cause says who can fix it. + severity: lag >= this.errorThreshold ? 'error' : 'warning', + gcTime: Math.min(timings.gcTime, lag), + cpuTime: timings.cpuTime, + watchdogGap: timings.watchdogGap, + cause: processDescheduled ? 'host' : 'code', }; } - /** @internal */ - _report(report: BlockedReport): void { + private report(report: BlockedReport): void { const now = performance.now(); const emitSpan = this.#spanLimiter.allow(now); - const emitLog = this.#logLimiter.allow(now); + // host contention and blocking code are fixed by different people: each has its own log + // quota so a noisy neighbor cannot silence a report about the agent's code, or the reverse + const emitLog = + report.cause === 'host' ? this.#hostLogLimiter.allow(now) : this.#logLimiter.allow(now); const emit = () => { + // the metric and the session summary count every stall, including the ones the span and + // log limiters drop try { - recordEventLoopBlocked(report.duration / 1000, report.severity); + recordEventLoopBlocked(report.duration / 1000, report.severity, report.cause); } catch (error) { log().error({ error }, 'failed to record the blocked event loop metric'); } - recordLoopStall(report.duration / 1000, report.startedAt + report.duration); + recordLoopStall(report.duration / 1000, report.startedAt + report.duration, report.cause); if (!emitSpan && !emitLog) return; if (emitSpan && this.#emitSpans) { this.#emitSpan(report, this.#spanLimiter.takeSuppressed()); } if (emitLog) this.#emitLog(report); - this._onReport?.(report); + this.onReport?.(report); }; try { @@ -237,11 +374,13 @@ export class EventLoopMonitor { } #emitSpan(report: BlockedReport, suppressed: number): void { - if (!getJobContext(false)) return; + if (!getJobContext(false)) return; // no job, no trace to belong to const attributes: Attributes = { [ATTR_BLOCKING_DURATION]: report.duration / 1000, [ATTR_BLOCKING_THRESHOLD]: report.warnThreshold / 1000, [ATTR_BLOCKING_SEVERITY]: report.severity, + [ATTR_BLOCKING_CAUSE]: report.cause, + [ATTR_BLOCKING_GC_TIME]: report.gcTime / 1000, [ATTR_BLOCKING_CPU_TIME]: report.cpuTime / 1000, }; if (suppressed) attributes[ATTR_BLOCKING_SUPPRESSED] = suppressed; @@ -254,27 +393,47 @@ export class EventLoopMonitor { if (report.severity === 'error') { span.setStatus({ code: SpanStatusCode.ERROR, - message: `event loop blocked for ${report.duration.toFixed(0)}ms`, + message: + report.cause === 'host' + ? `process not scheduled for ${report.duration.toFixed(0)}ms` + : `event loop blocked for ${report.duration.toFixed(0)}ms`, }); } span.end(report.startedAt + report.duration); } #emitLog(report: BlockedReport): void { - log().warn( - { - duration: Math.round(report.duration * 10) / 10, - threshold: report.warnThreshold, - cpuTime: Math.round(report.cpuTime * 10) / 10, - loop: this.#name, - }, - 'event loop blocked; synchronous work on the agent loop delays audio and turn handling, move it to a worker or an async client', - ); + const round = (value: number) => Math.round(value * 10) / 10; + const fields = { + duration: round(report.duration), + threshold: report.warnThreshold, + gcTime: round(report.gcTime), + processCpuTime: round(report.cpuTime), + watchdogGap: round(report.watchdogGap), + cause: report.cause, + loop: this.#name, + }; + if (report.cause === 'host') { + log().warn( + fields, + 'process not scheduled; the host did not run the agent for the whole stall (CPU contention or a container CPU quota), which delays audio and turn handling like blocking code would. Check the CPU limit and co-located load', + ); + } else if (this.watchdogActive) { + log().warn( + fields, + 'event loop blocked; synchronous work on the agent loop delays audio and turn handling, move it to a worker thread or an async API', + ); + } else { + // without the watchdog a stall could as well be the host not scheduling the process + log().warn( + fields, + 'event loop stalled; either synchronous work on the agent loop (move it to a worker thread or an async API) or the host did not schedule the process (CPU contention or quota)', + ); + } } } -/** @internal */ -export function _tickIntervalFor(warnThreshold: number): number { +function tickIntervalFor(warnThreshold: number): number { return Math.min( Math.max(warnThreshold / TICKS_PER_WARN_THRESHOLD, MIN_TICK_INTERVAL), MAX_TICK_INTERVAL, @@ -293,6 +452,7 @@ export interface StartMonitoringOptions { thresholds?: LoopMonitorThresholds; name?: string; emitSpans?: boolean; + watchdog?: boolean; } /** Start the process event-loop monitor, or return undefined if disabled/already running. */ @@ -314,9 +474,10 @@ export function startMonitoring( const monitor = new EventLoopMonitor({ warnThreshold: thresholds.warn, errorThreshold: thresholds.error, - tickInterval: _tickIntervalFor(thresholds.warn), + tickInterval: tickIntervalFor(thresholds.warn), name: options.name, emitSpans: options.emitSpans, + watchdog: options.watchdog, }); monitorState.monitor = monitor; monitor.start(); diff --git a/agents/src/telemetry/otel_metrics.test.ts b/agents/src/telemetry/otel_metrics.test.ts new file mode 100644 index 000000000..a58f4bbbe --- /dev/null +++ b/agents/src/telemetry/otel_metrics.test.ts @@ -0,0 +1,71 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { metrics, context as otelContext, trace } from '@opentelemetry/api'; +import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-proto'; +import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto'; +import { MeterProvider, type ResourceMetrics } from '@opentelemetry/sdk-metrics'; +import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { recordEventLoopBlocked } from './otel_metrics.js'; +import { setupCloudTracer, tracer } from './traces.js'; + +describe('cloud metrics pipeline', () => { + let prevKey: string | undefined; + let prevSecret: string | undefined; + + beforeEach(() => { + prevKey = process.env.LIVEKIT_API_KEY; + prevSecret = process.env.LIVEKIT_API_SECRET; + process.env.LIVEKIT_API_KEY = 'devkey'; + process.env.LIVEKIT_API_SECRET = 'secretsecretsecretsecretsecretsecret'; + vi.spyOn(OTLPTraceExporter.prototype, 'export').mockImplementation((_spans, callback) => + callback({ code: 0 }), + ); + }); + + afterEach(async () => { + const provider = tracer.getProvider(); + if (provider instanceof NodeTracerProvider) await provider.shutdown(); + await (metrics.getMeterProvider() as Partial).shutdown?.(); + metrics.disable(); + otelContext.disable(); + trace.disable(); + vi.restoreAllMocks(); + if (prevKey === undefined) delete process.env.LIVEKIT_API_KEY; + else process.env.LIVEKIT_API_KEY = prevKey; + if (prevSecret === undefined) delete process.env.LIVEKIT_API_SECRET; + else process.env.LIVEKIT_API_SECRET = prevSecret; + }); + + it('installs a meter provider on a fresh process and exports the blocked-loop histogram', async () => { + // a fresh process: the API hands out its no-op provider and nothing is registered globally + expect(metrics.getMeterProvider()).not.toBeInstanceOf(MeterProvider); + const exported: ResourceMetrics[] = []; + vi.spyOn(OTLPMetricExporter.prototype, 'export').mockImplementation((items, callback) => { + exported.push(items); + callback({ code: 0 }); + }); + + await setupCloudTracer({ + roomId: 'room1', + jobId: 'job1', + observabilityUrl: 'https://example.livekit.cloud', + enableLogs: false, + }); + + const provider = metrics.getMeterProvider(); + expect(provider).toBeInstanceOf(MeterProvider); + recordEventLoopBlocked(0.25, 'error'); + await (provider as MeterProvider).forceFlush(); + + const points = exported + .flatMap((rm) => rm.scopeMetrics) + .flatMap((sm) => sm.metrics) + .filter((m) => m.descriptor.name === 'lk.agents.event_loop.blocked_duration') + .flatMap((m) => m.dataPoints); + expect(points).toHaveLength(1); + expect(points[0]!.attributes).toMatchObject({ severity: 'error' }); + expect((points[0]!.value as { sum?: number }).sum).toBeCloseTo(0.25); + }); +}); diff --git a/agents/src/telemetry/otel_metrics.ts b/agents/src/telemetry/otel_metrics.ts index 9e4c65fdf..ecf06d9ea 100644 --- a/agents/src/telemetry/otel_metrics.ts +++ b/agents/src/telemetry/otel_metrics.ts @@ -21,10 +21,10 @@ function eventLoopBlockedHistogram(): Histogram { return blockedDuration; } -/** Record a synchronous event-loop block in seconds. */ -export function recordEventLoopBlocked(duration: number, severity: string): void { +/** Record an event-loop stall in seconds, with the severity and what caused it. */ +export function recordEventLoopBlocked(duration: number, severity: string, cause: string): void { const ctx = getJobContext(false); - const attributes: Attributes = { severity }; + const attributes: Attributes = { severity, cause }; if (ctx) { Object.assign(attributes, ctx._otelMetadata()); const roomId = ctx.job.room?.sid; diff --git a/agents/src/telemetry/rate_limiter.ts b/agents/src/telemetry/rate_limiter.ts new file mode 100644 index 000000000..b08161e3e --- /dev/null +++ b/agents/src/telemetry/rate_limiter.ts @@ -0,0 +1,31 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** Allows at most `limit` events per rolling minute and counts the ones it drops. @internal */ +export class RateLimiter { + readonly #limit: number; + readonly #events: number[] = []; + #suppressed = 0; + + constructor(limit: number) { + this.#limit = limit; + } + + allow(now: number): boolean { + const windowStart = now - 60_000; + while (this.#events.length && this.#events[0]! < windowStart) this.#events.shift(); + if (this.#events.length >= this.#limit) { + this.#suppressed++; + return false; + } + this.#events.push(now); + return true; + } + + takeSuppressed(): number { + const suppressed = this.#suppressed; + this.#suppressed = 0; + return suppressed; + } +} diff --git a/agents/src/telemetry/session_context.ts b/agents/src/telemetry/session_context.ts index b4237245f..8e5ea6e08 100644 --- a/agents/src/telemetry/session_context.ts +++ b/agents/src/telemetry/session_context.ts @@ -6,7 +6,7 @@ import { getJobContext } from '../job.js'; import type { AgentSession } from '../voice/agent_session.js'; type SessionWithLoopTelemetry = AgentSession & { - _recordLoopStall?: (durationInS: number, timestamp: number) => void; + _recordLoopStall?: (durationInS: number, timestamp: number, cause: string) => void; }; /** The primary agent session in the active job, if one exists. @internal */ @@ -20,8 +20,8 @@ export function sessionRootContext(): Context | undefined { } /** Record a stall on the active session when a session integration is available. @internal */ -export function recordLoopStall(durationInS: number, timestamp: number): void { +export function recordLoopStall(durationInS: number, timestamp: number, cause: string): void { const session = primarySession(); if (!session) return; - (session as SessionWithLoopTelemetry)._recordLoopStall?.(durationInS, timestamp); + (session as SessionWithLoopTelemetry)._recordLoopStall?.(durationInS, timestamp, cause); } diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts index d5a43dbe6..bfd01349b 100644 --- a/agents/src/telemetry/trace_types.test.ts +++ b/agents/src/telemetry/trace_types.test.ts @@ -167,6 +167,7 @@ const SAFE_KEYS = new Set([ 'lk.blocking.duration', 'lk.blocking.threshold', 'lk.blocking.severity', + 'lk.blocking.cause', 'lk.blocking.task', 'lk.blocking.stack', 'lk.blocking.gc_time', diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index c4d147e01..7679b85dd 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -121,11 +121,21 @@ export const ATTR_E2E_LATENCY = 'lk.e2e_latency'; export const ATTR_BLOCKING_DURATION = 'lk.blocking.duration'; export const ATTR_BLOCKING_THRESHOLD = 'lk.blocking.threshold'; export const ATTR_BLOCKING_SEVERITY = 'lk.blocking.severity'; +/** `code` for synchronous work on the loop, `host` when the process itself was not scheduled. */ +export const ATTR_BLOCKING_CAUSE = 'lk.blocking.cause'; +/** Not populated by the Node runtime: it cannot sample another thread's JavaScript stack. */ export const ATTR_BLOCKING_TASK = 'lk.blocking.task'; +/** Not populated by the Node runtime: it cannot sample another thread's JavaScript stack. */ export const ATTR_BLOCKING_STACK = 'lk.blocking.stack'; +/** Garbage-collection pause time inside the stall, in seconds. */ export const ATTR_BLOCKING_GC_TIME = 'lk.blocking.gc_time'; -/** Process CPU consumed during the heartbeat interval, in seconds. */ +/** + * CPU consumed during the stall, in seconds. In Node this is process-wide (every thread, + * including the libuv pool and native media threads), not the event-loop thread as in the Python + * SDK, so it can exceed `lk.blocking.duration`. + */ export const ATTR_BLOCKING_CPU_TIME = 'lk.blocking.cpu_time'; +/** Not populated by the Node runtime: there is no lazy-import equivalent to attribute. */ export const ATTR_BLOCKING_IMPORT = 'lk.blocking.import'; export const ATTR_BLOCKING_SUPPRESSED = 'lk.blocking.suppressed'; // Summary attributes on agent_session. diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 1ccfbf9a3..0ad5f900d 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -297,11 +297,17 @@ let cloudMeterProvider: MeterProvider | undefined; let cloudMetricsUnavailable = false; let cloudMeterShutdownRegistered = false; -function isNoopMeterProvider(provider: ReturnType): boolean { - // The API does not publicly export its singleton NoopMeterProvider. The constructor is the - // stable distinction available in OTel API 1.x, equivalent to checking the private proxy/no-op - // provider in the Python SDK. - return provider.constructor.name === 'NoopMeterProvider'; +// @opentelemetry/api keeps every registered global under this well-known symbol, keyed by API +// major version; the default no-op meter provider is never stored there. Reading the registry +// directly also sees a provider installed through a second copy of the API package, which an +// `instanceof` or constructor-name check on `metrics.getMeterProvider()` would not. +const OTEL_API_GLOBAL_KEY = Symbol.for('opentelemetry.js.api.1'); + +function hasGlobalMeterProvider(): boolean { + const registry = (globalThis as { [OTEL_API_GLOBAL_KEY]?: { metrics?: unknown } })[ + OTEL_API_GLOBAL_KEY + ]; + return registry?.metrics !== undefined; } function setupCloudMetrics( @@ -311,10 +317,10 @@ function setupCloudMetrics( ): MeterProvider | undefined { if (cloudMeterProvider || cloudMetricsUnavailable) return cloudMeterProvider; - const currentProvider = metrics.getMeterProvider(); - if (!isNoopMeterProvider(currentProvider)) { + if (hasGlobalMeterProvider()) { // Metric readers are fixed when an SDK 2.x MeterProvider is constructed. Preserve an - // application-installed global provider rather than replacing it and breaking its exporter. + // application-installed global provider rather than replacing it and breaking its exporter; + // measurements still reach it through the API's global meter. cloudMetricsUnavailable = true; return undefined; } diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index 011d8a3da..b6a14a3b0 100644 --- a/agents/src/voice/agent_session.ts +++ b/agents/src/voice/agent_session.ts @@ -1894,7 +1894,7 @@ export class AgentSession< } /** @internal */ - _recordLoopStall(durationInS: number, timestampMs: number): void { + _recordLoopStall(durationInS: number, timestampMs: number, cause: string): void { const span = this.sessionSpan; if (!span?.isRecording()) { return; @@ -1905,7 +1905,10 @@ export class AgentSession< this.loopStallMax = Math.max(this.loopStallMax, durationInS); span.addEvent( LOOP_STALL_EVENT, - { [traceTypes.ATTR_BLOCKING_DURATION]: durationInS }, + { + [traceTypes.ATTR_BLOCKING_DURATION]: durationInS, + [traceTypes.ATTR_BLOCKING_CAUSE]: cause, + }, timestampMs, ); span.setAttributes({ From 98d7965c857c8980e1a02bfb917cbd2f6f1f2047 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 19:11:21 -0700 Subject: [PATCH 3/6] fix CI --- agents/src/telemetry/traces.ts | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 0ad5f900d..55e35bd05 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -352,14 +352,13 @@ function setupCloudMetrics( cloudMeterProvider = provider; if (!cloudMeterShutdownRegistered) { cloudMeterShutdownRegistered = true; - process.once('beforeExit', async () => { + process.once('beforeExit', () => { const ownedProvider = cloudMeterProvider; cloudMeterProvider = undefined; - try { - await ownedProvider?.shutdown({ timeoutMillis: 10_000 }); - } catch (error) { + // the pending export keeps the loop alive, so the shutdown completes before exit + void ownedProvider?.shutdown({ timeoutMillis: 10_000 }).catch((error: unknown) => { console.error('Failed to shut down cloud metrics:', error); - } + }); }); } return provider; From 272aa6668270bcf57cd436c7366e2a8125afb81a Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 19:34:28 -0700 Subject: [PATCH 4/6] address comments --- agents/etc/agents.api.md | 11 ++++ agents/src/ipc/job_proc_lazy_main.ts | 4 ++ agents/src/job_lifecycle.test.ts | 38 ++++++++++++- agents/src/job_lifecycle.ts | 12 +++- agents/src/telemetry/index.ts | 1 + agents/src/telemetry/loop_monitor.test.ts | 68 ++++++++++++++++++---- agents/src/telemetry/loop_monitor.ts | 69 +++++++++++++++++++---- agents/src/telemetry/trace_types.ts | 6 +- agents/src/telemetry/traces.ts | 21 ++++--- 9 files changed, 197 insertions(+), 33 deletions(-) diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index 897df9cfc..ec6f879dd 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -2187,6 +2187,7 @@ declare namespace beta { // @public (undocumented) interface BlockedReport { cause: LoopStallCause; + cpuScope: LoopCpuScope; cpuTime: number; duration: number; gcTime: number; @@ -4170,6 +4171,9 @@ export class FixedGate implements AudioGate { update(frame: AudioFrame): boolean; } +// @internal +function flushCloudMetrics(): Promise; + // @internal function flushOtelLogs(): Promise; @@ -5652,6 +5656,11 @@ export const logMetrics: (metrics: AgentMetrics) => void; // @public export function loopAudioFramesFromFile(filePath: string, options?: AudioDecodeOptions): AsyncGenerator; +// Warning: (ae-missing-release-tag) "LoopCpuScope" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +type LoopCpuScope = 'thread' | 'process'; + declare namespace loopMonitor { export { startMonitoring, @@ -5667,6 +5676,7 @@ declare namespace loopMonitor { SPAN_NAME, LoopMonitorSeverity, LoopStallCause, + LoopCpuScope, BlockedReport, LoopMonitorThresholds, EventLoopMonitorOptions, @@ -8382,6 +8392,7 @@ declare namespace telemetry { loopMonitor, traceTypes, FanoutSpanProcessor, + flushCloudMetrics, flushOtelLogs, setTracerProvider, setupCloudTracer, diff --git a/agents/src/ipc/job_proc_lazy_main.ts b/agents/src/ipc/job_proc_lazy_main.ts index f2f2503c7..654a26d06 100644 --- a/agents/src/ipc/job_proc_lazy_main.ts +++ b/agents/src/ipc/job_proc_lazy_main.ts @@ -18,6 +18,7 @@ import { import { finalizeSession, flushJobLogs, + flushJobMetrics, runShutdownCallbacks, validateSessionEndTimeout, waitForEntrypointShutdown, @@ -332,7 +333,10 @@ const startJob = ( process.on('message', messageHandler); await join.await; + // stop the monitor first so a stall from the shutdown callbacks is recorded, then export: + // the periodic reader gets no further turn before process.exit() below if (loopMonitor) stopMonitoring(loopMonitor); + await flushJobMetrics(logger); clearTimeout(orphanedTimeout); process.off('message', messageHandler); diff --git a/agents/src/job_lifecycle.test.ts b/agents/src/job_lifecycle.test.ts index b72b6a23e..88e7a3958 100644 --- a/agents/src/job_lifecycle.test.ts +++ b/agents/src/job_lifecycle.test.ts @@ -7,17 +7,20 @@ import { type JobContext, getJobContext } from './job.js'; import { finalizeSession, flushJobLogs, + flushJobMetrics, runShutdownCallbacks, waitForEntrypointShutdown, } from './job_lifecycle.js'; -import { flushOtelLogs } from './telemetry/index.js'; +import { flushCloudMetrics, flushOtelLogs } from './telemetry/index.js'; import type { AgentSession } from './voice/agent_session.js'; vi.mock('./telemetry/index.js', () => ({ flushOtelLogs: vi.fn(), + flushCloudMetrics: vi.fn(), })); const flushOtelLogsMock = vi.mocked(flushOtelLogs); +const flushCloudMetricsMock = vi.mocked(flushCloudMetrics); function createLogger(): Logger { return { @@ -339,3 +342,36 @@ describe('flushJobLogs', () => { expect(logger.error).toHaveBeenCalledWith({ error }, 'Failed to flush OTEL logs'); }); }); + +describe('flushJobMetrics', () => { + it('exports the pending cloud metrics', async () => { + flushCloudMetricsMock.mockResolvedValue(); + await flushJobMetrics(createLogger()); + expect(flushCloudMetricsMock).toHaveBeenCalledOnce(); + }); + + it('stops waiting after the metric flush timeout', async () => { + vi.useFakeTimers(); + flushCloudMetricsMock.mockReturnValue(new Promise(() => {})); + const logger = createLogger(); + const completion = flushJobMetrics(logger); + + await vi.advanceTimersByTimeAsync(10_000); + await completion; + + expect(logger.error).toHaveBeenCalledWith( + { timeout: 10_000 }, + 'OTEL metric flush timed out; proceeding with job shutdown', + ); + }); + + it('continues after logging metric exporter errors', async () => { + const error = new Error('exporter failed'); + const logger = createLogger(); + flushCloudMetricsMock.mockRejectedValue(error); + + await flushJobMetrics(logger); + + expect(logger.error).toHaveBeenCalledWith({ error }, 'Failed to flush OTEL metrics'); + }); +}); diff --git a/agents/src/job_lifecycle.ts b/agents/src/job_lifecycle.ts index db6dca693..8fba3d84a 100644 --- a/agents/src/job_lifecycle.ts +++ b/agents/src/job_lifecycle.ts @@ -3,13 +3,14 @@ // SPDX-License-Identifier: Apache-2.0 import type { Logger } from 'pino'; import { type JobContext, runWithJobContextAsync } from './job.js'; -import { flushOtelLogs } from './telemetry/index.js'; +import { flushCloudMetrics, flushOtelLogs } from './telemetry/index.js'; import { IdleTimeoutError, waitUntilTimeout } from './utils.js'; export const DEFAULT_SESSION_END_TIMEOUT = 300 * 1000; const ENTRYPOINT_SHUTDOWN_TIMEOUT = 15 * 1000; const SESSION_CLOSE_TIMEOUT = 60 * 1000; const OTEL_LOG_FLUSH_TIMEOUT = 10 * 1000; +const OTEL_METRIC_FLUSH_TIMEOUT = 10 * 1000; const MAX_TIMER_TIMEOUT = 2_147_483_647; type SessionEndCallback = (ctx: JobContext) => unknown; @@ -124,3 +125,12 @@ export async function flushJobLogs(logger: Logger): Promise { lateReject: 'OTEL log flush rejected after shutdown timeout', }); } + +/** Export the last metrics of the job. Run it after everything that can still record one. */ +export async function flushJobMetrics(logger: Logger): Promise { + await waitOrContinue(() => flushCloudMetrics(), OTEL_METRIC_FLUSH_TIMEOUT, logger, { + timeout: 'OTEL metric flush timed out; proceeding with job shutdown', + error: 'Failed to flush OTEL metrics', + lateReject: 'OTEL metric flush rejected after shutdown timeout', + }); +} diff --git a/agents/src/telemetry/index.ts b/agents/src/telemetry/index.ts index 01597f305..80029b00c 100644 --- a/agents/src/telemetry/index.ts +++ b/agents/src/telemetry/index.ts @@ -29,6 +29,7 @@ export * as loopMonitor from './loop_monitor.js'; export * as traceTypes from './trace_types.js'; export { FanoutSpanProcessor, + flushCloudMetrics, flushOtelLogs, setTracerProvider, setupCloudTracer, diff --git a/agents/src/telemetry/loop_monitor.test.ts b/agents/src/telemetry/loop_monitor.test.ts index a4284cb00..bb44c1dd0 100644 --- a/agents/src/telemetry/loop_monitor.test.ts +++ b/agents/src/telemetry/loop_monitor.test.ts @@ -123,10 +123,19 @@ describe.sequential('event loop monitor', () => { return exporter.getFinishedSpans().filter((span) => span.name === SPAN_NAME); } + /** The span of the longest stall: a loaded host can add shorter ones around a test's block. */ + function longestBlockedSpan() { + return blockedSpans().sort( + (a, b) => + (b.attributes[ATTR_BLOCKING_DURATION] as number) - + (a.attributes[ATTR_BLOCKING_DURATION] as number), + )[0]; + } + it('reports a blocking call as a backdated error span', async () => { blockLoop(200); await settle(); - const [span] = blockedSpans(); + const span = longestBlockedSpan(); expect(span).toBeDefined(); const duration = span!.attributes[ATTR_BLOCKING_DURATION] as number; expect(duration).toBeGreaterThanOrEqual((200 - TICK - 10) / 1000); @@ -142,15 +151,43 @@ describe.sequential('event loop monitor', () => { expect(span!.attributes[ATTR_BLOCKING_GC_TIME]).toBe(0); expect(span!.attributes[ATTR_BLOCKING_CAUSE]).toBe('code'); // the watchdog thread kept running while the loop thread waited: the process was scheduled - const [report] = reports; + const [report] = codeReports(reports).sort((a, b) => b.duration - a.duration); expect(report!.cause).toBe('code'); expect(report!.watchdogGap).toBeLessThan(report!.duration * 0.5); + // Atomics.wait parks the loop thread: per-thread accounting sees (almost) no CPU + expect(report!.cpuScope).toBe('thread'); + expect(report!.cpuTime).toBeLessThan(report!.duration * 0.5); + }); + + it('measures the CPU of the loop thread, not of the process', async () => { + // a worker thread burns CPU for the whole block while the loop thread only waits + const burner = new Worker(`const t = Date.now(); while (Date.now() - t < 300) {}`, { + eval: true, + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + blockLoop(200); + await settle(); + await new Promise((resolve) => burner.once('exit', () => resolve())); + const [report] = codeReports(reports).sort((a, b) => b.duration - a.duration); + expect(report).toBeDefined(); + expect(report!.cpuTime).toBeLessThan(report!.duration * 0.5); + }); + + it('holds a stall one heartbeat and flushes it on stop', async () => { + blockLoop(70); + // the late tick has run (it expired during the block) but the next one has not + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(reports).toEqual([]); + monitor.stop(); + expect(reports).toHaveLength(1); + expect(reports[0]!.duration).toBeGreaterThanOrEqual(70 - TICK - 10); }); it('reports a block between thresholds as a warning', async () => { blockLoop(70); await settle(); - const [span] = blockedSpans(); + const span = longestBlockedSpan(); + expect(span!.attributes[ATTR_BLOCKING_DURATION]).toBeLessThan(ERROR / 1000); expect(span!.attributes[ATTR_BLOCKING_SEVERITY]).toBe('warning'); expect(span!.status.code).toBe(SpanStatusCode.UNSET); }); @@ -163,9 +200,8 @@ describe.sequential('event loop monitor', () => { expect(report).toBeDefined(); expect(report!.gcTime).toBeGreaterThan(0); expect(report!.gcTime).toBeLessThanOrEqual(report!.duration); - const span = blockedSpans().find( - (candidate) => candidate.attributes[ATTR_BLOCKING_DURATION] === report!.duration / 1000, - ); + const span = longestBlockedSpan(); + expect(span!.attributes[ATTR_BLOCKING_DURATION]).toBe(report!.duration / 1000); expect(span!.attributes[ATTR_BLOCKING_GC_TIME]).toBeCloseTo(report!.gcTime / 1000, 6); }); @@ -185,7 +221,7 @@ describe.sequential('event loop monitor', () => { blockLoop(70); await settle(); entrypoint.end(); - const [span] = blockedSpans(); + const span = longestBlockedSpan(); expect(span!.parentSpanContext?.spanId).toBe(entrypoint.spanContext().spanId); expect( span!.endTime[0] < entrypoint.endTime![0] || span!.endTime[1] <= entrypoint.endTime![1], @@ -285,17 +321,29 @@ describe.sequential('event loop monitor', () => { }); describe('host contention', () => { - it('is told apart from blocking code by the watchdog and CPU time', () => { + it('is told apart from blocking code by the watchdog and the loop thread CPU', () => { monitor.stop(); - // watchdog late by most of the stall and (nearly) no CPU burned: nothing ran + expect(monitor['cpuScope']).toBe('thread'); + // watchdog late by most of the stall and the loop thread burned (nearly) nothing: it + // was not running const descheduled = monitor['buildReport'](400, timings(10, 300)); expect(descheduled.cause).toBe('host'); // watchdog on time: the loop thread alone was stuck, in a blocking wait expect(monitor['buildReport'](400, timings(10, 4)).cause).toBe('code'); - // watchdog starved but the process was busy the whole time: code kept the loop + // watchdog starved but the loop thread was busy the whole time: contention delayed the + // watchdog, synchronous code held the loop expect(monitor['buildReport'](400, timings(380, 300)).cause).toBe('code'); }); + it('is decided by the watchdog alone when CPU time is process-wide', () => { + monitor.stop(); + monitor['cpuScope'] = 'process'; + // native media threads burned CPU while both monitored threads starved: process-wide + // CPU cannot pin that on the loop thread, so it must not override the watchdog + expect(monitor['buildReport'](400, timings(250, 300)).cause).toBe('host'); + expect(monitor['buildReport'](400, timings(250, 4)).cause).toBe('code'); + }); + it('keeps the severity of its impact', () => { monitor.stop(); // a stall past the error threshold delays audio the same however it came about diff --git a/agents/src/telemetry/loop_monitor.ts b/agents/src/telemetry/loop_monitor.ts index 28d302498..d3d1495dd 100644 --- a/agents/src/telemetry/loop_monitor.ts +++ b/agents/src/telemetry/loop_monitor.ts @@ -46,6 +46,9 @@ export type LoopMonitorSeverity = 'warning' | 'error'; */ export type LoopStallCause = 'code' | 'host'; +/** What a report's `cpuTime` covers: the event-loop thread, or the whole process on older Node. */ +export type LoopCpuScope = 'thread' | 'process'; + export interface BlockedReport { /** Heartbeat lag in milliseconds. */ duration: number; @@ -53,14 +56,20 @@ export interface BlockedReport { startedAt: number; warnThreshold: number; severity: LoopMonitorSeverity; - /** Garbage-collection pause time observed during the stall, in milliseconds. */ + /** + * Garbage-collection pause time observed during the stall, in milliseconds. Node delivers GC + * entries to observers a couple of loop turns after the fact, so a report is emitted one + * heartbeat after its stall to include them. + */ gcTime: number; /** - * CPU consumed during the stall, in milliseconds. This is process-wide (every thread, including - * the libuv pool and the native media threads), not the event loop thread alone: Node does not - * expose per-thread CPU accounting. It can therefore exceed the stall duration. + * CPU consumed during the stall, in milliseconds. Scoped to the event-loop thread where Node + * offers `process.threadCpuUsage()` (22.15 / 23.9 and later); process-wide before that, which + * also counts the libuv pool and the native media threads and can exceed the stall duration. */ cpuTime: number; + /** What `cpuTime` covers. */ + cpuScope: LoopCpuScope; /** How late the watchdog thread woke during the stall, in milliseconds. 0 without a watchdog. */ watchdogGap: number; /** @@ -148,6 +157,11 @@ setInterval(() => { }, interval); `; +// Per-thread CPU accounting (Node 22.15 / 23.9+). Older runtimes fall back to the process total. +const threadCpuUsage: (() => NodeJS.CpuUsage) | undefined = ( + process as { threadCpuUsage?: () => NodeJS.CpuUsage } +).threadCpuUsage?.bind(process); + /** Detect synchronous work that prevents the Node event loop from servicing timers. */ export class EventLoopMonitor { readonly warnThreshold: number; @@ -166,6 +180,9 @@ export class EventLoopMonitor { #lastCpuUsage: NodeJS.CpuUsage = { user: 0, system: 0 }; #gcObserver?: PerformanceObserver; #gcTime = 0; + /** A stall detected on the previous tick, held back one heartbeat for its GC entries. */ + #pending?: BlockedReport; + private cpuScope: LoopCpuScope = threadCpuUsage ? 'thread' : 'process'; #watchdog?: Worker; #watchdogState?: BigInt64Array; #reportContext?: Context; @@ -212,7 +229,7 @@ export class EventLoopMonitor { if (this.#started || this.#closed) return; this.#started = true; this.#lastTickAt = performance.now(); - this.#lastCpuUsage = process.cpuUsage(); + this.#lastCpuUsage = this.#cpuUsage(); this.#startGcObserver(); if (this.#useWatchdog) this.#startWatchdog(); this.#scheduleTick(); @@ -224,6 +241,9 @@ export class EventLoopMonitor { this.#closed = true; if (this.#timer) clearTimeout(this.#timer); this.#timer = undefined; + // a stall waiting for its GC entries must not be lost to the shutdown that follows + this.#flushPending(this.#gcTime); + this.#gcTime = 0; this.#gcObserver?.disconnect(); this.#gcObserver = undefined; const watchdog = this.#watchdog; @@ -277,11 +297,15 @@ export class EventLoopMonitor { this.#timer.unref(); } + #cpuUsage(): NodeJS.CpuUsage { + return this.cpuScope === 'thread' && threadCpuUsage ? threadCpuUsage() : process.cpuUsage(); + } + #onTick(): void { if (this.#closed) return; const now = performance.now(); const lag = now - (this.#lastTickAt + this.tickInterval); - const cpu = process.cpuUsage(); + const cpu = this.#cpuUsage(); const cpuTime = (cpu.user - this.#lastCpuUsage.user + cpu.system - this.#lastCpuUsage.system) / 1000; const gcTime = this.#gcTime; @@ -292,8 +316,25 @@ export class EventLoopMonitor { this.#lastCpuUsage = cpu; const watchdogGap = this.#consumeWatchdogGap(windowStart); this.#scheduleTick(); + // GC entries for the previous stall reach the observer through two immediates, which the + // late tick can run ahead of; they have landed by now, so the previous stall gets them + const gcClaimed = this.#flushPending(gcTime); if (lag < this.warnThreshold) return; - this.report(this.buildReport(lag, { cpuTime, gcTime, watchdogGap })); + this.#pending = this.buildReport(lag, { + cpuTime, + gcTime: gcClaimed ? 0 : gcTime, + watchdogGap, + }); + } + + /** Emit the stall held from the previous tick, crediting it the GC time seen since. */ + #flushPending(gcTime: number): boolean { + const pending = this.#pending; + if (!pending) return false; + this.#pending = undefined; + pending.gcTime = Math.min(pending.gcTime + gcTime, pending.duration); + this.report(pending); + return true; } /** @@ -321,10 +362,14 @@ export class EventLoopMonitor { timings: { cpuTime: number; gcTime: number; watchdogGap: number }, ): BlockedReport { // the watchdog is an independent thread: if it too woke late by most of the stall, the - // process was not being scheduled (host contention, CPU quota, a suspended machine). A - // descheduled process burns no CPU, which rules out the loop having been busy instead. + // process was not being scheduled (host contention, CPU quota, a suspended machine). Under + // contention the scheduler can also starve only the watchdog while the loop thread runs + // synchronous code, which is still code's fault: the loop thread's own CPU time tells, since + // a thread that was not running burns none. Process-wide CPU cannot say which thread was + // busy, so without per-thread accounting the watchdog alone decides. const watchdogStarved = timings.watchdogGap >= lag * 0.5; - const processDescheduled = watchdogStarved && timings.cpuTime < lag * 0.5; + const loopThreadBusy = this.cpuScope === 'thread' && timings.cpuTime >= lag * 0.5; + const processDescheduled = watchdogStarved && !loopThreadBusy; return { duration: lag, // the block started no earlier than the last on-time tick @@ -335,6 +380,7 @@ export class EventLoopMonitor { severity: lag >= this.errorThreshold ? 'error' : 'warning', gcTime: Math.min(timings.gcTime, lag), cpuTime: timings.cpuTime, + cpuScope: this.cpuScope, watchdogGap: timings.watchdogGap, cause: processDescheduled ? 'host' : 'code', }; @@ -408,7 +454,8 @@ export class EventLoopMonitor { duration: round(report.duration), threshold: report.warnThreshold, gcTime: round(report.gcTime), - processCpuTime: round(report.cpuTime), + cpuTime: round(report.cpuTime), + cpuScope: report.cpuScope, watchdogGap: round(report.watchdogGap), cause: report.cause, loop: this.#name, diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index 7679b85dd..befebe68e 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -130,9 +130,9 @@ export const ATTR_BLOCKING_STACK = 'lk.blocking.stack'; /** Garbage-collection pause time inside the stall, in seconds. */ export const ATTR_BLOCKING_GC_TIME = 'lk.blocking.gc_time'; /** - * CPU consumed during the stall, in seconds. In Node this is process-wide (every thread, - * including the libuv pool and native media threads), not the event-loop thread as in the Python - * SDK, so it can exceed `lk.blocking.duration`. + * CPU consumed by the event-loop thread during the stall, in seconds. On Node runtimes without + * `process.threadCpuUsage()` (before 22.15 / 23.9) it is process-wide instead, counting the libuv + * pool and native media threads too, and can then exceed `lk.blocking.duration`. */ export const ATTR_BLOCKING_CPU_TIME = 'lk.blocking.cpu_time'; /** Not populated by the Node runtime: there is no lazy-import equivalent to attribute. */ diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 55e35bd05..6614b2bbe 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -590,13 +590,10 @@ export async function setupCloudTracer( ...(deploymentId ? { [ATTR_DEPLOYMENT_ID]: deploymentId } : {}), }), ); - const meterProvider = setupCloudMetrics(observabilityUrl, headers, meterResource); - if (meterProvider) { - const { getJobContext } = await import('../job.js'); - getJobContext(false)?.addShutdownCallback(() => - meterProvider.forceFlush({ timeoutMillis: 10_000 }), - ); - } + // The final export belongs to the job bootstrap (flushCloudMetrics), after every shutdown + // callback has run: a stall inside one of them is only recorded once it returns, and the + // periodic reader would not get another turn before process.exit(). + setupCloudMetrics(observabilityUrl, headers, meterResource); if (enableTraces) { const url = `${observabilityUrl}/observability/traces/otlp/v0`; @@ -690,6 +687,16 @@ export async function flushOtelLogs(): Promise { await flushPinoLogs(); } +/** + * Export every measurement the cloud meter provider holds. Call it once all work that could + * record a metric is done: the job process exits explicitly, so no later flush would run. + * + * @internal + */ +export async function flushCloudMetrics(): Promise { + await cloudMeterProvider?.forceFlush({ timeoutMillis: 10_000 }); +} + /** Proto field names and shapes, matching what livekit/agents emits for the same log body. */ function chatItemSpanAttribute(item: ChatItem): Record { return encodeChatItem(item).toJson({ useProtoFieldName: true }) as Record; From fdc737ea4a8d5cb9dd1b287228751fe5207c2a17 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 21:23:44 -0700 Subject: [PATCH 5/6] improve CI speed --- agents/src/inference/stt.test.ts | 4 +++- agents/src/inference/test_utils.ts | 2 +- plugins/test/src/stt.ts | 25 ++++++++++++++++++++++--- turbo.json | 1 + vitest.config.ts | 3 +++ 5 files changed, 30 insertions(+), 5 deletions(-) diff --git a/agents/src/inference/stt.test.ts b/agents/src/inference/stt.test.ts index 3b75b4693..d8ca6f9ce 100644 --- a/agents/src/inference/stt.test.ts +++ b/agents/src/inference/stt.test.ts @@ -1075,7 +1075,9 @@ describeLiveKitInference('LiveKit Inference STT integration', agents, async (har 'assemblyai/universal-streaming', 'xai/stt-1', ] as const) { - describe(model, { retry: 1 }, async () => { + // each model is an independent gateway session: run the models, and both sample rates of + // each, at the same time instead of one 50 s clip after another + describe(model, { retry: 1, concurrent: true }, async () => { const stt = model === 'assemblyai/universal-streaming' ? new STT({ model, modelOptions: { format_turns: true } }) diff --git a/agents/src/inference/test_utils.ts b/agents/src/inference/test_utils.ts index 86f4c6c11..812ef9a15 100644 --- a/agents/src/inference/test_utils.ts +++ b/agents/src/inference/test_utils.ts @@ -13,7 +13,7 @@ interface InferenceTestHarness { stt: ( model: STT, vad: VAD, - supports?: Partial<{ streaming: boolean; nonStreaming: boolean }>, + supports?: Partial<{ streaming: boolean; nonStreaming: boolean; streamSpeed: number }>, ) => Promise; tts: ( model: TTS, diff --git a/plugins/test/src/stt.ts b/plugins/test/src/stt.ts index 7f2761275..aedec4423 100644 --- a/plugins/test/src/stt.ts +++ b/plugins/test/src/stt.ts @@ -27,6 +27,22 @@ const TRANSCRIPT = const STREAM_CHUNK_DURATION_MS = 10; const TRAILING_SILENCE_DURATION_MS = 2_000; +// Audio is streamed on a fixed schedule, so a test takes the clip's 50 s at real time rather +// than drifting past it by a timer's overhead per 10 ms chunk. A multiple of real time can be +// set per harness call (`streamSpeed`) or globally with STT_TEST_STREAM_SPEED. Providers cap +// this: AssemblyAI and xAI fail at 4x, and at 2x they finalize no sooner than at 1x, since they +// process audio at its own pace. The wall-clock win comes from running the suites concurrently. +const DEFAULT_STREAM_SPEED = Number(process.env.STT_TEST_STREAM_SPEED) || 1; + +/** Resolve when chunk `index` is due on a schedule running `speed` times real time. */ +const paceChunk = (startedAt: number, index: number, speed: number): Promise => { + const due = startedAt + ((index + 1) * STREAM_CHUNK_DURATION_MS) / speed; + const wait = due - performance.now(); + // when behind schedule, still yield to the loop so the output side can drain the stream + return new Promise((resolve) => + wait > 0 ? setTimeout(resolve, wait) : setImmediate(() => resolve()), + ); +}; const validate = async (text: string, transcript: string, threshold: number) => { text = text.toLowerCase().replace(/\s/g, ' ').trim(); @@ -37,10 +53,11 @@ const validate = async (text: string, transcript: string, threshold: number) => export const stt = async ( stt: sttlib.STT, vad: VAD, - supports: Partial<{ streaming: boolean; nonStreaming: boolean }> = {}, + supports: Partial<{ streaming: boolean; nonStreaming: boolean; streamSpeed: number }> = {}, ) => { initializeLogger({ pretty: false }); supports = { streaming: true, nonStreaming: true, ...supports }; + const streamSpeed = supports.streamSpeed ?? DEFAULT_STREAM_SPEED; describe('STT', () => { it.skipIf(!supports.nonStreaming).each([24000, 44100])( 'should properly transcribe speech at %i Hz', @@ -74,9 +91,11 @@ export const stt = async ( stt.on('error', onStreamError); const input = async () => { + const startedAt = performance.now(); + let chunk = 0; for (const frame of frames) { stream.pushFrame(frame); - await new Promise((resolve) => setTimeout(resolve, STREAM_CHUNK_DURATION_MS)); + await paceChunk(startedAt, chunk++, streamSpeed); } const silence = new AudioFrame( @@ -91,7 +110,7 @@ export const stt = async ( elapsed += STREAM_CHUNK_DURATION_MS ) { stream.pushFrame(silence); - await new Promise((resolve) => setTimeout(resolve, STREAM_CHUNK_DURATION_MS)); + await paceChunk(startedAt, chunk++, streamSpeed); } stream.endInput(); }; diff --git a/turbo.json b/turbo.json index 8babfa9f0..e288e5491 100644 --- a/turbo.json +++ b/turbo.json @@ -117,6 +117,7 @@ "MISTRAL_API_KEY", "META_API_KEY", "MODEL_API_KEY", + "STT_TEST_STREAM_SPEED", "VITEST" ], "tasks": { diff --git a/vitest.config.ts b/vitest.config.ts index cdae0c52e..d66ebdbe3 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -15,6 +15,9 @@ export default defineConfig(({ mode }) => ({ environment: 'node', // Default timeout for unit tests (5s), integration tests override this per-suite testTimeout: 5_000, + // only tests marked concurrent are affected: lets the live inference STT suites (4 models x 2 + // sample rates) stream at the same time instead of two rounds of five + maxConcurrency: 8, env: loadEnv(mode, process.cwd(), ''), setupFiles: ['./vitest.setup.ts'], }, From 51ca6045f27d62074e78ca90c5c873c48cb694a9 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 21:36:19 -0700 Subject: [PATCH 6/6] fix comment --- agents/src/telemetry/loop_monitor.test.ts | 19 +++++++++++++++++++ agents/src/telemetry/loop_monitor.ts | 11 ++++++----- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/agents/src/telemetry/loop_monitor.test.ts b/agents/src/telemetry/loop_monitor.test.ts index bb44c1dd0..f90c8d389 100644 --- a/agents/src/telemetry/loop_monitor.test.ts +++ b/agents/src/telemetry/loop_monitor.test.ts @@ -31,6 +31,7 @@ import { ATTR_BLOCKING_DURATION, ATTR_BLOCKING_GC_TIME, ATTR_BLOCKING_SEVERITY, + ATTR_BLOCKING_SUPPRESSED, ATTR_BLOCKING_THRESHOLD, } from './trace_types.js'; import { setTracerProvider, tracer } from './traces.js'; @@ -236,6 +237,24 @@ describe.sequential('event loop monitor', () => { expect(reports).toHaveLength(1); }); + it('does not charge the span quota for stalls that cannot have a span', () => { + monitor.stop(); + // an idle child before its job: reports, but no job context to parent a span to + monitor.setReportContext(undefined); + for (let i = 0; i < MAX_SPANS_PER_MINUTE * 2; i++) { + monitor['report'](monitor['buildReport'](100, timings())); + } + expect(blockedSpans()).toEqual([]); + // the job starts: its very first stall still gets a span, with nothing counted as suppressed + monitor.setReportContext(trace.setSpan(ROOT_CONTEXT, sessionRoot), (fn) => + runWithJobContext(fakeJob(), fn), + ); + monitor['report'](monitor['buildReport'](100, timings())); + const [span] = blockedSpans(); + expect(span).toBeDefined(); + expect(span!.attributes[ATTR_BLOCKING_SUPPRESSED]).toBeUndefined(); + }); + it('records every stall on the active session', () => { monitor.stop(); const seen: [number, string][] = []; diff --git a/agents/src/telemetry/loop_monitor.ts b/agents/src/telemetry/loop_monitor.ts index d3d1495dd..d84e82db2 100644 --- a/agents/src/telemetry/loop_monitor.ts +++ b/agents/src/telemetry/loop_monitor.ts @@ -388,7 +388,6 @@ export class EventLoopMonitor { private report(report: BlockedReport): void { const now = performance.now(); - const emitSpan = this.#spanLimiter.allow(now); // host contention and blocking code are fixed by different people: each has its own log // quota so a noisy neighbor cannot silence a report about the agent's code, or the reverse const emitLog = @@ -402,10 +401,13 @@ export class EventLoopMonitor { log().error({ error }, 'failed to record the blocked event loop metric'); } recordLoopStall(report.duration / 1000, report.startedAt + report.duration, report.cause); + // a span needs a job to belong to (a root span here would be a stray trace), so the span + // quota is only charged once the job context is restored and a span can exist: stalls of + // an idle child before its job, or of the worker process, must not use up the job's budget + const spanEligible = this.#emitSpans && getJobContext(false) !== undefined; + const emitSpan = spanEligible && this.#spanLimiter.allow(now); if (!emitSpan && !emitLog) return; - if (emitSpan && this.#emitSpans) { - this.#emitSpan(report, this.#spanLimiter.takeSuppressed()); - } + if (emitSpan) this.#emitSpan(report, this.#spanLimiter.takeSuppressed()); if (emitLog) this.#emitLog(report); this.onReport?.(report); }; @@ -420,7 +422,6 @@ export class EventLoopMonitor { } #emitSpan(report: BlockedReport, suppressed: number): void { - if (!getJobContext(false)) return; // no job, no trace to belong to const attributes: Attributes = { [ATTR_BLOCKING_DURATION]: report.duration / 1000, [ATTR_BLOCKING_THRESHOLD]: report.warnThreshold / 1000,