diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c319f52fe..8a25495add 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ This is the log of notable changes to EAS CLI and related packages. ### 🎉 New features - [build-tools] Pass a metrics CORS origin to serve-sim so the dashboard can read a session's live CPU/memory stream cross-origin. ([#3990](https://github.com/expo/eas-cli/pull/3990) by [@gwdp](https://github.com/gwdp)) +- [build-tools] Collect each iOS session's serve-sim CPU/memory `/metrics` to a file and upload it as a device run session artifact at teardown. ([#3995](https://github.com/expo/eas-cli/pull/3995) by [@gwdp](https://github.com/gwdp)) ### 🐛 Bug fixes diff --git a/packages/build-tools/src/steps/easFunctions.ts b/packages/build-tools/src/steps/easFunctions.ts index 97421432a6..ee99a731c0 100644 --- a/packages/build-tools/src/steps/easFunctions.ts +++ b/packages/build-tools/src/steps/easFunctions.ts @@ -49,6 +49,8 @@ import { createStartCuttlefishDeviceBuildFunction } from './functions/startCuttl import { createStartIosSimulatorBuildFunction } from './functions/startIosSimulator'; import { createStartIosSimulatorRecordingsBuildFunction } from './functions/startIosSimulatorRecordings'; import { createStartServeSimRemoteSessionBuildFunction } from './functions/startServeSimRemoteSession'; +import { createStartServeSimMetricsBuildFunction } from './functions/startServeSimMetrics'; +import { createCollectServeSimMetricsBuildFunction } from './functions/collectServeSimMetrics'; import { createUploadArtifactBuildFunction } from './functions/uploadArtifact'; import { createUploadDeviceRunSessionScreenRecordingsBuildFunction } from './functions/uploadDeviceRunSessionScreenRecordings'; import { createUploadToAscBuildFunction } from './functions/uploadToAsc'; @@ -99,6 +101,8 @@ export function getEasFunctions(ctx: CustomBuildContext): BuildFunction[] { createFinishIosSimulatorRecordingsBuildFunction(), createUploadDeviceRunSessionScreenRecordingsBuildFunction(ctx), createStartServeSimRemoteSessionBuildFunction(ctx), + createStartServeSimMetricsBuildFunction(), + createCollectServeSimMetricsBuildFunction(ctx), createInstallMaestroBuildFunction(), createInstallPodsBuildFunction(), diff --git a/packages/build-tools/src/steps/functions/collectServeSimMetrics.ts b/packages/build-tools/src/steps/functions/collectServeSimMetrics.ts new file mode 100644 index 0000000000..de84fef331 --- /dev/null +++ b/packages/build-tools/src/steps/functions/collectServeSimMetrics.ts @@ -0,0 +1,31 @@ +import { BuildFunction, BuildRuntimePlatform } from '@expo/steps'; + +import { type CustomBuildContext } from '../../customBuildContext'; +import { getDeviceRunSessionIdOrThrow } from '../utils/remoteDeviceRunSession'; +import { uploadServeSimMetricsFileAsync } from '../utils/serveSimMetricsArtifacts'; +import { ServeSimMetricsRecorder } from '../utils/serveSimMetricsRecorder'; + +export function createCollectServeSimMetricsBuildFunction(ctx: CustomBuildContext): BuildFunction { + return new BuildFunction({ + namespace: 'eas', + id: 'collect_serve_sim_metrics', + name: 'Collect serve-sim metrics', + __metricsId: 'eas/collect_serve_sim_metrics', + supportedRuntimePlatforms: [BuildRuntimePlatform.DARWIN], + fn: async ({ logger }, { env }) => { + const collected = await ServeSimMetricsRecorder.finishAsync({ logger }); + if (collected.length === 0) { + logger.info('No serve-sim metrics collected; skipping upload.'); + return; + } + try { + const deviceRunSessionId = getDeviceRunSessionIdOrThrow(env); + for (const { udid, filePath } of collected) { + await uploadServeSimMetricsFileAsync(ctx, { deviceRunSessionId, udid, filePath, logger }); + } + } catch (err) { + logger.warn({ err }, 'Could not upload serve-sim metrics.'); + } + }, + }); +} diff --git a/packages/build-tools/src/steps/functions/startServeSimMetrics.ts b/packages/build-tools/src/steps/functions/startServeSimMetrics.ts new file mode 100644 index 0000000000..e50b62cdd5 --- /dev/null +++ b/packages/build-tools/src/steps/functions/startServeSimMetrics.ts @@ -0,0 +1,16 @@ +import { BuildFunction, BuildRuntimePlatform } from '@expo/steps'; + +import { ServeSimMetricsRecorder } from '../utils/serveSimMetricsRecorder'; + +export function createStartServeSimMetricsBuildFunction(): BuildFunction { + return new BuildFunction({ + namespace: 'eas', + id: 'start_serve_sim_metrics', + name: 'Start serve-sim metrics', + __metricsId: 'eas/start_serve_sim_metrics', + supportedRuntimePlatforms: [BuildRuntimePlatform.DARWIN], + fn: async ({ logger }) => { + await ServeSimMetricsRecorder.startAsync({ logger }); + }, + }); +} diff --git a/packages/build-tools/src/steps/utils/__tests__/serveSimMetricsArtifacts.test.ts b/packages/build-tools/src/steps/utils/__tests__/serveSimMetricsArtifacts.test.ts new file mode 100644 index 0000000000..0051e7dc55 --- /dev/null +++ b/packages/build-tools/src/steps/utils/__tests__/serveSimMetricsArtifacts.test.ts @@ -0,0 +1,109 @@ +import { bunyan } from '@expo/logger'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { Readable } from 'node:stream'; + +import { CustomBuildContext } from '../../../customBuildContext'; +import { uploadDeviceRunSessionArtifactAsync } from '../deviceRunSessionArtifacts'; +import { uploadServeSimMetricsFileAsync } from '../serveSimMetricsArtifacts'; + +jest.mock('../deviceRunSessionArtifacts'); +jest.mock('../../../sentry'); + +function createLoggerMock(): bunyan { + return { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + } as unknown as bunyan; +} + +async function readStreamAsync(stream: NodeJS.ReadableStream): Promise { + let out = ''; + for await (const chunk of stream as Readable) { + out += chunk.toString(); + } + return out; +} + +const ctx = {} as CustomBuildContext; +const NDJSON = '{"t":1000,"cpuPct":1,"memBytes":2}\n'; + +let workDir: string; +let filePath: string; + +beforeEach(async () => { + jest.mocked(uploadDeviceRunSessionArtifactAsync).mockReset(); + workDir = await mkdtemp(path.join(os.tmpdir(), 'serve-sim-metrics-art-test-')); + filePath = path.join(workDir, 'AAAA.ndjson'); +}); + +afterEach(async () => { + await rm(workDir, { recursive: true, force: true }); +}); + +describe(uploadServeSimMetricsFileAsync, () => { + it('uploads the file as a metrics.ndjson artifact keyed by udid', async () => { + await writeFile(filePath, NDJSON); + let uploaded = ''; + jest + .mocked(uploadDeviceRunSessionArtifactAsync) + .mockImplementation(async (_ctx, { stream }) => { + uploaded = await readStreamAsync(stream); + }); + + await uploadServeSimMetricsFileAsync(ctx, { + deviceRunSessionId: 'session-id', + udid: 'AAAA', + filePath, + logger: createLoggerMock(), + }); + + expect(uploaded).toBe(NDJSON); + expect(uploadDeviceRunSessionArtifactAsync).toHaveBeenCalledWith( + ctx, + expect.objectContaining({ + deviceRunSessionId: 'session-id', + artifactId: 'metrics-AAAA', + filename: 'metrics.ndjson', + size: Buffer.byteLength(NDJSON), + }) + ); + }); + + it('skips the upload when the file is missing', async () => { + await uploadServeSimMetricsFileAsync(ctx, { + deviceRunSessionId: 'session-id', + udid: 'AAAA', + filePath, + logger: createLoggerMock(), + }); + expect(uploadDeviceRunSessionArtifactAsync).not.toHaveBeenCalled(); + }); + + it('skips the upload when the file is empty', async () => { + await writeFile(filePath, ''); + await uploadServeSimMetricsFileAsync(ctx, { + deviceRunSessionId: 'session-id', + udid: 'AAAA', + filePath, + logger: createLoggerMock(), + }); + expect(uploadDeviceRunSessionArtifactAsync).not.toHaveBeenCalled(); + }); + + it('swallows an upload failure (best-effort)', async () => { + await writeFile(filePath, NDJSON); + jest.mocked(uploadDeviceRunSessionArtifactAsync).mockRejectedValue(new Error('R2 down')); + await expect( + uploadServeSimMetricsFileAsync(ctx, { + deviceRunSessionId: 'session-id', + udid: 'AAAA', + filePath, + logger: createLoggerMock(), + }) + ).resolves.toBeUndefined(); + }); +}); diff --git a/packages/build-tools/src/steps/utils/__tests__/serveSimMetricsRecorder.test.ts b/packages/build-tools/src/steps/utils/__tests__/serveSimMetricsRecorder.test.ts new file mode 100644 index 0000000000..507efec528 --- /dev/null +++ b/packages/build-tools/src/steps/utils/__tests__/serveSimMetricsRecorder.test.ts @@ -0,0 +1,190 @@ +import { bunyan } from '@expo/logger'; +import fetch from 'node-fetch'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { Readable } from 'node:stream'; +import { setTimeout as delay } from 'node:timers/promises'; + +import { + ServeSimMetricsRecorder, + readServeSimServersAsync, + streamServeSimMetricsToFileAsync, +} from '../serveSimMetricsRecorder'; + +jest.mock('../../../sentry'); +jest.mock('node-fetch'); + +const { Response } = jest.requireActual('node-fetch') as typeof import('node-fetch'); + +function createLoggerMock(): bunyan { + return { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + } as unknown as bunyan; +} + +const SSE_STREAM = [ + 'event: meta', + 'data: {"schemaVersion":1,"udid":"U","hostCores":8,"sampleIntervalMs":1000}', + '', + ': heartbeat', + '', + 'data: {"t":1000,"bundleId":"dev.expo.MyApp","cpuPct":0,"memBytes":100}', + '', + 'data: {"t":2000,"bundleId":"dev.expo.MyApp","cpuPct":40,"memBytes":120}', + '', +].join('\n'); + +const EXPECTED_NDJSON = + '{"schemaVersion":1,"udid":"U","hostCores":8,"sampleIntervalMs":1000}\n' + + '{"t":1000,"bundleId":"dev.expo.MyApp","cpuPct":0,"memBytes":100}\n' + + '{"t":2000,"bundleId":"dev.expo.MyApp","cpuPct":40,"memBytes":120}\n'; + +function sseResponse(): InstanceType { + return new Response(Readable.from([Buffer.from(SSE_STREAM)])); +} + +let workDir: string; +let stateDir: string; + +beforeEach(async () => { + jest.mocked(fetch).mockReset(); + workDir = await mkdtemp(path.join(os.tmpdir(), 'serve-sim-metrics-rec-test-')); + stateDir = path.join(workDir, 'serve-sim-state'); +}); + +afterEach(async () => { + await ServeSimMetricsRecorder.finishAsync({ logger: createLoggerMock() }); + await rm(workDir, { recursive: true, force: true }); +}); + +async function writeServerStateAsync( + udid: string, + value: unknown = { pid: 1, port: 4000, device: udid, url: 'https://sim.example' } +): Promise { + await writeFile(path.join(stateDir, `server-${udid}.json`), JSON.stringify(value)); +} + +describe(readServeSimServersAsync, () => { + beforeEach(async () => { + const { mkdir } = await import('node:fs/promises'); + await mkdir(stateDir, { recursive: true }); + }); + + it('returns each server-*.json with a device + url, ignoring other files', async () => { + await writeServerStateAsync('A', { device: 'A', url: 'http://127.0.0.1:1' }); + await writeServerStateAsync('B', { device: 'B', url: 'http://127.0.0.1:2' }); + await writeFile(path.join(stateDir, 'not-a-server.json'), '{"device":"X","url":"http://x"}'); + + const servers = await readServeSimServersAsync(stateDir); + expect(servers).toEqual( + expect.arrayContaining([ + { udid: 'A', url: 'http://127.0.0.1:1' }, + { udid: 'B', url: 'http://127.0.0.1:2' }, + ]) + ); + expect(servers).toHaveLength(2); + }); + + it('tolerates malformed/incomplete state files', async () => { + await writeFile(path.join(stateDir, 'server-bad.json'), 'not json'); + await writeServerStateAsync('C', { device: 'C' }); // missing url + await writeServerStateAsync('D', { device: 'D', url: 'http://127.0.0.1:4' }); + + expect(await readServeSimServersAsync(stateDir)).toEqual([ + { udid: 'D', url: 'http://127.0.0.1:4' }, + ]); + }); + + it('returns [] when the state dir does not exist', async () => { + expect(await readServeSimServersAsync(path.join(workDir, 'nope'))).toEqual([]); + }); +}); + +describe(streamServeSimMetricsToFileAsync, () => { + it('appends the data payloads to the file as NDJSON, ignoring event/heartbeat lines', async () => { + jest.mocked(fetch).mockResolvedValue(sseResponse()); + const filePath = path.join(workDir, 'stream.ndjson'); + await streamServeSimMetricsToFileAsync({ + serveSimUrl: 'https://sim.example', + filePath, + signal: new AbortController().signal, + logger: createLoggerMock(), + }); + expect(await readFile(filePath, 'utf-8')).toBe(EXPECTED_NDJSON); + expect(String(jest.mocked(fetch).mock.calls[0][0])).toBe('https://sim.example/metrics'); + }); + + it('writes nothing on a non-ok response', async () => { + jest.mocked(fetch).mockResolvedValue(new Response('nope', { status: 502 })); + const filePath = path.join(workDir, 'empty.ndjson'); + await streamServeSimMetricsToFileAsync({ + serveSimUrl: 'https://sim.example', + filePath, + signal: new AbortController().signal, + logger: createLoggerMock(), + }); + expect(await readFile(filePath, 'utf-8')).toBe(''); + }); +}); + +describe('ServeSimMetricsRecorder', () => { + it('discovers running servers and collects a per-device NDJSON file', async () => { + const { mkdir } = await import('node:fs/promises'); + await mkdir(stateDir, { recursive: true }); + await writeServerStateAsync('AAAA', { device: 'AAAA', url: 'https://sim-a.example' }); + jest.mocked(fetch).mockResolvedValue(sseResponse()); + + await ServeSimMetricsRecorder.startAsync({ logger: createLoggerMock(), stateDir }); + await delay(200); + + const collected = await ServeSimMetricsRecorder.finishAsync({ logger: createLoggerMock() }); + expect(collected).toHaveLength(1); + expect(collected[0].udid).toBe('AAAA'); + expect(await readFile(collected[0].filePath, 'utf-8')).toBe(EXPECTED_NDJSON); + }); + + it('reconnects a device whose first /metrics attempt fails', async () => { + const { mkdir } = await import('node:fs/promises'); + await mkdir(stateDir, { recursive: true }); + await writeServerStateAsync('CCCC', { device: 'CCCC', url: 'https://sim-c.example' }); + jest + .mocked(fetch) + .mockResolvedValueOnce(new Response('nope', { status: 502 })) + .mockImplementation(async () => sseResponse()); + + await ServeSimMetricsRecorder.startAsync({ + logger: createLoggerMock(), + stateDir, + pollIntervalMs: 20, + }); + await delay(200); + + const collected = await ServeSimMetricsRecorder.finishAsync({ logger: createLoggerMock() }); + expect(collected).toHaveLength(1); + // The 502 didn't permanently stop collection — a later attempt recovered samples. + expect(await readFile(collected[0].filePath, 'utf-8')).toContain('"cpuPct"'); + }); + + it('returns [] from finishAsync when nothing is running', async () => { + expect(await ServeSimMetricsRecorder.finishAsync({ logger: createLoggerMock() })).toEqual([]); + }); + + it('is a no-op when start is called twice (single session)', async () => { + const { mkdir } = await import('node:fs/promises'); + await mkdir(stateDir, { recursive: true }); + await writeServerStateAsync('BBBB', { device: 'BBBB', url: 'https://sim-b.example' }); + jest.mocked(fetch).mockResolvedValue(sseResponse()); + + await ServeSimMetricsRecorder.startAsync({ logger: createLoggerMock(), stateDir }); + await ServeSimMetricsRecorder.startAsync({ logger: createLoggerMock(), stateDir }); + await delay(200); + + const collected = await ServeSimMetricsRecorder.finishAsync({ logger: createLoggerMock() }); + expect(collected).toHaveLength(1); + expect(await ServeSimMetricsRecorder.finishAsync({ logger: createLoggerMock() })).toEqual([]); + }); +}); diff --git a/packages/build-tools/src/steps/utils/serveSimMetricsArtifacts.ts b/packages/build-tools/src/steps/utils/serveSimMetricsArtifacts.ts new file mode 100644 index 0000000000..ab0cda3438 --- /dev/null +++ b/packages/build-tools/src/steps/utils/serveSimMetricsArtifacts.ts @@ -0,0 +1,53 @@ +import { type bunyan } from '@expo/logger'; +import { createReadStream } from 'node:fs'; +import { stat } from 'node:fs/promises'; + +import { CustomBuildContext } from '../../customBuildContext'; +import { Sentry } from '../../sentry'; +import { formatBytes } from '../../utils/artifacts'; +import { uploadDeviceRunSessionArtifactAsync } from './deviceRunSessionArtifacts'; + +const METRICS_ARTIFACT_FILENAME = 'metrics.ndjson'; + +export async function uploadServeSimMetricsFileAsync( + ctx: CustomBuildContext, + { + deviceRunSessionId, + udid, + filePath, + logger, + }: { + deviceRunSessionId: string; + udid: string; + filePath: string; + logger: bunyan; + } +): Promise { + let size: number; + try { + ({ size } = await stat(filePath)); + } catch { + logger.info(`No serve-sim metrics captured for ${udid}; skipping upload.`); + return; + } + if (size === 0) { + logger.info(`No serve-sim metrics captured for ${udid}; skipping upload.`); + return; + } + try { + logger.info(`Uploading serve-sim metrics for ${udid} (${formatBytes(size)}).`); + await uploadDeviceRunSessionArtifactAsync(ctx, { + deviceRunSessionId, + artifactId: `metrics-${udid}`, + name: METRICS_ARTIFACT_FILENAME, + filename: METRICS_ARTIFACT_FILENAME, + kind: undefined, + size, + stream: createReadStream(filePath), + }); + } catch (err) { + const error = err instanceof Error ? err : new Error(String(err)); + Sentry.capture('Could not upload serve-sim metrics artifact', error); + logger.warn({ err: error }, `Could not upload serve-sim metrics artifact for ${udid}.`); + } +} diff --git a/packages/build-tools/src/steps/utils/serveSimMetricsRecorder.ts b/packages/build-tools/src/steps/utils/serveSimMetricsRecorder.ts new file mode 100644 index 0000000000..c91b285021 --- /dev/null +++ b/packages/build-tools/src/steps/utils/serveSimMetricsRecorder.ts @@ -0,0 +1,207 @@ +import { type bunyan } from '@expo/logger'; +import fetch from 'node-fetch'; +import { createWriteStream } from 'node:fs'; +import { mkdtemp, readFile, readdir } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; + +import { Sentry } from '../../sentry'; + +const POLL_INTERVAL_MS = 2_000; +const MAX_STREAM_ATTEMPTS_PER_DEVICE = 10; +// serve-sim's own per-device registry while serving: `${tmpdir}/serve-sim/server-.json`. +const SERVE_SIM_STATE_DIR = path.join(os.tmpdir(), 'serve-sim'); + +type ServeSimServer = { udid: string; url: string }; + +type ServeSimMetricsSession = { + logger: bunyan; + stateDir: string; + pollIntervalMs: number; + outputDirectory: string; + files: Map; + attempts: Map; + activeStreams: Map }>; + pollingPromise: Promise; + abortController: AbortController; +}; + +let activeSession: ServeSimMetricsSession | null = null; + +export namespace ServeSimMetricsRecorder { + export async function startAsync({ + logger, + stateDir = SERVE_SIM_STATE_DIR, + pollIntervalMs = POLL_INTERVAL_MS, + }: { + logger: bunyan; + stateDir?: string; + pollIntervalMs?: number; + }): Promise { + if (activeSession) { + logger.info('serve-sim metrics polling is already running.'); + return; + } + const outputDirectory = await mkdtemp(path.join(os.tmpdir(), 'serve-sim-metrics-')); + const session: ServeSimMetricsSession = { + logger, + stateDir, + pollIntervalMs, + outputDirectory, + files: new Map(), + attempts: new Map(), + activeStreams: new Map(), + pollingPromise: Promise.resolve(), + abortController: new AbortController(), + }; + activeSession = session; + + logger.info('Started polling serve-sim for cpu/mem metrics.'); + session.pollingPromise = pollServeSimMetricsAsync(session).catch(err => { + const error = err instanceof Error ? err : new Error(String(err)); + Sentry.capture('serve-sim metrics poller failed', error); + logger.warn({ err: error }, 'serve-sim metrics poller failed.'); + }); + } + + export async function finishAsync({ + logger, + }: { + logger: bunyan; + }): Promise<{ udid: string; filePath: string }[]> { + const session = activeSession; + if (!session) { + logger.info('No serve-sim metrics polling is running.'); + return []; + } + activeSession = null; + + session.abortController.abort(); + await session.pollingPromise; + await Promise.all( + [...session.activeStreams.values()].map(async stream => { + stream.abortController.abort(); + await stream.donePromise; + }) + ); + return [...session.files.entries()].map(([udid, filePath]) => ({ udid, filePath })); + } +} + +async function pollServeSimMetricsAsync(session: ServeSimMetricsSession): Promise { + const { abortController, logger } = session; + while (!abortController.signal.aborted) { + for (const server of await readServeSimServersAsync(session.stateDir)) { + if (session.activeStreams.has(server.udid)) { + continue; + } + const attempts = session.attempts.get(server.udid) ?? 0; + if (attempts >= MAX_STREAM_ATTEMPTS_PER_DEVICE) { + continue; + } + session.attempts.set(server.udid, attempts + 1); + + let filePath = session.files.get(server.udid); + if (!filePath) { + filePath = path.join(session.outputDirectory, `${server.udid}.ndjson`); + session.files.set(server.udid, filePath); + } + + const streamAbortController = new AbortController(); + const streamSignal = AbortSignal.any([abortController.signal, streamAbortController.signal]); + logger.info(`Collecting serve-sim metrics for ${server.udid}.`); + // Free the slot when the stream ends so the poller reconnects if the server is still up (an + // early race or a mid-session drop), up to the per-device attempt cap. + const donePromise = streamServeSimMetricsToFileAsync({ + serveSimUrl: server.url, + filePath, + signal: streamSignal, + logger, + }).finally(() => { + session.activeStreams.delete(server.udid); + }); + session.activeStreams.set(server.udid, { + abortController: streamAbortController, + donePromise, + }); + } + await delay(session.pollIntervalMs, undefined, { signal: abortController.signal }).catch( + () => {} + ); + } +} + +export async function readServeSimServersAsync(stateDir: string): Promise { + let entries: string[]; + try { + entries = await readdir(stateDir); + } catch { + return []; + } + const servers: ServeSimServer[] = []; + for (const entry of entries) { + if (!/^server-.+\.json$/.test(entry)) { + continue; + } + try { + const state = JSON.parse(await readFile(path.join(stateDir, entry), 'utf-8')) as { + device?: unknown; + url?: unknown; + }; + if (typeof state.device === 'string' && typeof state.url === 'string') { + servers.push({ udid: state.device, url: state.url }); + } + } catch { + continue; + } + } + return servers; +} + +export async function streamServeSimMetricsToFileAsync({ + serveSimUrl, + filePath, + signal, + logger, +}: { + serveSimUrl: string; + filePath: string; + signal: AbortSignal; + logger: bunyan; +}): Promise { + const file = createWriteStream(filePath, { flags: 'a' }); + file.on('error', err => { + logger.warn({ err }, `serve-sim metrics file write failed for ${filePath}.`); + }); + try { + const response = await fetch(new URL('/metrics', serveSimUrl).toString(), { signal }); + if (!response.ok || !response.body) { + logger.warn(`serve-sim /metrics responded ${response.status} for ${serveSimUrl}.`); + return; + } + let buffer = ''; + for await (const chunk of response.body) { + buffer += chunk.toString(); + let newline: number; + while ((newline = buffer.indexOf('\n')) !== -1) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + if (line.startsWith('data:')) { + const payload = line.slice('data:'.length).trim(); + if (payload) { + file.write(payload + '\n'); + } + } + } + } + } catch (err) { + // Stream ends (teardown abort, serve-sim exit/restart) are expected and reconnected by the + // poller, so log rather than report — best-effort telemetry shouldn't page. + if (!signal.aborted) { + logger.warn({ err }, `serve-sim /metrics stream ended for ${serveSimUrl}.`); + } + } finally { + await new Promise(resolve => file.end(() => resolve())); + } +}