History

feat: add native resource telemetry

Replace scheduled process shell probes with a persistent native collector, Electron power/process telemetry, Effect services, diagnostics UI, attribution, and release packaging.

Julius Marminge committed 4 months ago

Commit dbd8a8e1 parent 79ad07162 files · +7741 −1957Browse files

Showing the changes to apps/server/src/diagnostics/ProcessResourceMonitor.ts (1 of 62 files). Show all

Unified · Split

apps/server/src/diagnostics/ProcessResourceMonitor.ts+44−271
@@ -1,45 +1,13 @@
1import type {1import type {
2 ServerProcessResourceHistoryBucket,
3 ServerProcessResourceHistoryInput,2 ServerProcessResourceHistoryInput,
4 ServerProcessResourceHistoryResult,3 ServerProcessResourceHistoryResult,
5 ServerProcessResourceHistorySummary,
6} from "@t3tools/contracts";4} from "@t3tools/contracts";
7import * as Context from "effect/Context";5import * as Context from "effect/Context";
8import * as DateTime from "effect/DateTime";
9import * as Effect from "effect/Effect";6import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";7import * as Layer from "effect/Layer";
11import * as Option from "effect/Option";8import * as Option from "effect/Option";
12import * as Ref from "effect/Ref";
13import { ChildProcessSpawner } from "effect/unstable/process";
14 9
15import {10import * as ResourceTelemetry from "../resourceTelemetry/ResourceTelemetry.ts";
16 buildDescendantEntries,
17 isDiagnosticsQueryProcess,
18 type ProcessRow,
19 readProcessRows,
20} from "./ProcessDiagnostics.ts";
21
22const SAMPLE_INTERVAL_MS = 5_000;
23const RETENTION_MS = 60 * 60_000;
24const MAX_RETAINED_SAMPLES = 20_000;
25
26export interface ProcessResourceSample {
27 readonly sampledAt: DateTime.Utc;
28 readonly sampledAtMs: number;
29 readonly processKey: string;
30 readonly pid: number;
31 readonly ppid: number;
32 readonly command: string;
33 readonly cpuPercent: number;
34 readonly rssBytes: number;
35 readonly depth: number;
36 readonly isServerRoot: boolean;
37}
38
39interface MonitorState {
40 readonly samples: ReadonlyArray<ProcessResourceSample>;
41 readonly lastError: string | null;
42}
43 11
44export interface ProcessResourceMonitorShape {12export interface ProcessResourceMonitorShape {
45 readonly readHistory: (13 readonly readHistory: (
@@ -52,246 +20,51 @@ export class ProcessResourceMonitor extends Context.Service< · 6 unmodified lines
52 ProcessResourceMonitorShape20 ProcessResourceMonitorShape
53>()("t3/diagnostics/ProcessResourceMonitor") {}21>()("t3/diagnostics/ProcessResourceMonitor") {}
54 22
55function dateTimeFromMillis(ms: number): DateTime.Utc {
56 return DateTime.makeUnsafe(ms);
57}
58
59function sampleKey(row: Pick<ProcessRow, "pid" | "command">): string {
60 return `${row.pid}:${row.command}`;
61}
62
63function findServerRootRow(rows: ReadonlyArray<ProcessRow>, serverPid: number): ProcessRow | null {
64 return rows.find((row) => row.pid === serverPid) ?? null;
65}
66
67export function collectMonitoredSamples(input: {
68 readonly rows: ReadonlyArray<ProcessRow>;
69 readonly serverPid: number;
70 readonly sampledAt: DateTime.Utc;
71 readonly sampledAtMs: number;
72}): ReadonlyArray<ProcessResourceSample> {
73 const rows = input.rows.filter((row) => !isDiagnosticsQueryProcess(row, input.serverPid));
74 const root = findServerRootRow(rows, input.serverPid);
75 const descendants = buildDescendantEntries(rows, input.serverPid);
76 const samples: ProcessResourceSample[] = [];
77
78 if (root) {
79 samples.push({
80 sampledAt: input.sampledAt,
81 sampledAtMs: input.sampledAtMs,
82 processKey: sampleKey(root),
83 pid: root.pid,
84 ppid: root.ppid,
85 command: root.command,
86 cpuPercent: root.cpuPercent,
87 rssBytes: root.rssBytes,
88 depth: 0,
89 isServerRoot: true,
90 });
91 }
92
93 for (const process of descendants) {
94 samples.push({
95 sampledAt: input.sampledAt,
96 sampledAtMs: input.sampledAtMs,
97 processKey: sampleKey(process),
98 pid: process.pid,
99 ppid: process.ppid,
100 command: process.command,
101 cpuPercent: process.cpuPercent,
102 rssBytes: process.rssBytes,
103 depth: process.depth + 1,
104 isServerRoot: false,
105 });
106 }
107
108 return samples;
109}
110
111function trimSamples(
112 samples: ReadonlyArray<ProcessResourceSample>,
113 nowMs: number,
114): ReadonlyArray<ProcessResourceSample> {
115 const minSampledAtMs = nowMs - RETENTION_MS;
116 const retained = samples.filter((sample) => sample.sampledAtMs >= minSampledAtMs);
117 return retained.length <= MAX_RETAINED_SAMPLES
118 ? retained
119 : retained.slice(retained.length - MAX_RETAINED_SAMPLES);
120}
121
122function summarizeProcesses(
123 samples: ReadonlyArray<ProcessResourceSample>,
124): ReadonlyArray<ServerProcessResourceHistorySummary> {
125 const groups = new Map<string, ProcessResourceSample[]>();
126 for (const sample of samples) {
127 const processSamples = groups.get(sample.processKey) ?? [];
128 processSamples.push(sample);
129 groups.set(sample.processKey, processSamples);
130 }
131
132 return [...groups.entries()]
133 .map(([processKey, processSamples]) => {
134 const sorted = processSamples.toSorted((left, right) => left.sampledAtMs - right.sampledAtMs);
135 const first = sorted[0]!;
136 const latest = sorted[sorted.length - 1]!;
137 const cpuPercentTotal = sorted.reduce((total, sample) => total + sample.cpuPercent, 0);
138 const maxCpuPercent = Math.max(...sorted.map((sample) => sample.cpuPercent));
139 const maxRssBytes = Math.max(...sorted.map((sample) => sample.rssBytes));
140 const cpuSecondsApprox = sorted.reduce(
141 (total, sample) => total + (sample.cpuPercent / 100) * (SAMPLE_INTERVAL_MS / 1_000),
142 0,
143 );
144
145 return {
146 processKey,
147 pid: latest.pid,
148 ppid: latest.ppid,
149 command: latest.command,
150 depth: latest.depth,
151 isServerRoot: latest.isServerRoot,
152 firstSeenAt: first.sampledAt,
153 lastSeenAt: latest.sampledAt,
154 currentCpuPercent: latest.cpuPercent,
155 avgCpuPercent: cpuPercentTotal / sorted.length,
156 maxCpuPercent,
157 cpuSecondsApprox,
158 currentRssBytes: latest.rssBytes,
159 maxRssBytes,
160 sampleCount: sorted.length,
161 } satisfies ServerProcessResourceHistorySummary;
162 })
163 .toSorted((left, right) => right.cpuSecondsApprox - left.cpuSecondsApprox);
164}
165
166function buildBuckets(input: {
167 readonly samples: ReadonlyArray<ProcessResourceSample>;
168 readonly nowMs: number;
169 readonly windowMs: number;
170 readonly bucketMs: number;
171}): ReadonlyArray<ServerProcessResourceHistoryBucket> {
172 const bucketMs = Math.max(1_000, input.bucketMs);
173 const windowStartMs = input.nowMs - input.windowMs;
174 const buckets: ServerProcessResourceHistoryBucket[] = [];
175
176 for (let startedAtMs = windowStartMs; startedAtMs < input.nowMs; startedAtMs += bucketMs) {
177 const endedAtMs = Math.min(input.nowMs, startedAtMs + bucketMs);
178 const bucketSamples = input.samples.filter(
179 (sample) =>
180 sample.sampledAtMs >= startedAtMs &&
181 (endedAtMs === input.nowMs
182 ? sample.sampledAtMs <= endedAtMs
183 : sample.sampledAtMs < endedAtMs),
184 );
185 const samplesByRead = new Map<number, ProcessResourceSample[]>();
186 for (const sample of bucketSamples) {
187 const samplesAtTime = samplesByRead.get(sample.sampledAtMs) ?? [];
188 samplesAtTime.push(sample);
189 samplesByRead.set(sample.sampledAtMs, samplesAtTime);
190 }
191
192 const readTotals = [...samplesByRead.values()].map((samplesAtTime) => ({
193 cpuPercent: samplesAtTime.reduce((total, sample) => total + sample.cpuPercent, 0),
194 rssBytes: samplesAtTime.reduce((total, sample) => total + sample.rssBytes, 0),
195 processCount: samplesAtTime.length,
196 }));
197 const avgCpuPercent =
198 readTotals.length === 0
199 ? 0
200 : readTotals.reduce((total, read) => total + read.cpuPercent, 0) / readTotals.length;
201
202 buckets.push({
203 startedAt: dateTimeFromMillis(startedAtMs),
204 endedAt: dateTimeFromMillis(endedAtMs),
205 avgCpuPercent,
206 maxCpuPercent: readTotals.length ? Math.max(...readTotals.map((read) => read.cpuPercent)) : 0,
207 maxRssBytes: readTotals.length ? Math.max(...readTotals.map((read) => read.rssBytes)) : 0,
208 maxProcessCount: readTotals.length
209 ? Math.max(...readTotals.map((read) => read.processCount))
210 : 0,
211 });
212 }
213
214 return buckets;
215}
216
217export function aggregateProcessResourceHistory(input: {
218 readonly samples: ReadonlyArray<ProcessResourceSample>;
219 readonly readAt: DateTime.Utc;
220 readonly readAtMs: number;
221 readonly windowMs: number;
222 readonly bucketMs: number;
223 readonly lastError: string | null;
224}): ServerProcessResourceHistoryResult {
225 const windowMs = Math.max(1_000, input.windowMs);
226 const bucketMs = Math.max(1_000, input.bucketMs);
227 const minSampledAtMs = input.readAtMs - windowMs;
228 const samples = input.samples.filter((sample) => sample.sampledAtMs >= minSampledAtMs);
229 const topProcesses = summarizeProcesses(samples);
230 const totalCpuSecondsApprox = samples.reduce(
231 (total, sample) => total + (sample.cpuPercent / 100) * (SAMPLE_INTERVAL_MS / 1_000),
232 0,
233 );
234
235 return {
236 readAt: input.readAt,
237 windowMs,
238 bucketMs,
239 sampleIntervalMs: SAMPLE_INTERVAL_MS,
240 retainedSampleCount: input.samples.length,
241 totalCpuSecondsApprox,
242 buckets: buildBuckets({ samples, nowMs: input.readAtMs, windowMs, bucketMs }),
243 topProcesses,
244 error: input.lastError ? Option.some({ message: input.lastError }) : Option.none(),
245 };
246}
247
248export const make = Effect.fn("makeProcessResourceMonitor")(function* () {23export const make = Effect.fn("makeProcessResourceMonitor")(function* () {
249 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;24 const telemetry = yield* ResourceTelemetry.ResourceTelemetry;
250 const state = yield* Ref.make<MonitorState>({ samples: [], lastError: null });
251
252 const sampleOnce = Effect.gen(function* () {
253 const sampledAt = yield* DateTime.now;
254 const sampledAtMs = DateTime.toEpochMillis(sampledAt);
255 const rows = yield* readProcessRows.pipe(
256 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
257 );
258 const samples = collectMonitoredSamples({
259 rows,
260 serverPid: process.pid,
261 sampledAt,
262 sampledAtMs,
263 });
264 yield* Ref.update(state, (current) => ({
265 samples: trimSamples([...current.samples, ...samples], sampledAtMs),
266 lastError: null,
267 }));
268 }).pipe(
269 Effect.catch((error: unknown) =>
270 Ref.update(state, (current) => ({
271 ...current,
272 lastError: error instanceof Error ? error.message : "Failed to sample process resources.",
273 })),
274 ),
275 );
276
277 yield* Effect.forever(sampleOnce.pipe(Effect.andThen(Effect.sleep(SAMPLE_INTERVAL_MS)))).pipe(
278 Effect.forkScoped,
279 );
280
281 const readHistory: ProcessResourceMonitorShape["readHistory"] = (input) =>25 const readHistory: ProcessResourceMonitorShape["readHistory"] = (input) =>
282 Effect.gen(function* () {26 telemetry.readHistory(input).pipe(
283 const readAt = yield* DateTime.now;27 Effect.map((history) => {
284 const readAtMs = DateTime.toEpochMillis(readAt);28 const topProcesses = history.topProcesses.map((entry) => ({
285 const current = yield* Ref.get(state);29 processKey: `${entry.identity.pid}:${entry.identity.startTimeMs}`,
286 return aggregateProcessResourceHistory({30 pid: entry.identity.pid,
287 samples: current.samples,31 ppid: entry.ppid,
288 readAt,32 command: entry.command || entry.name || "unknown",
289 readAtMs,33 depth: entry.depth,
290 windowMs: input.windowMs,34 isServerRoot: entry.category === "server",
291 bucketMs: input.bucketMs,35 firstSeenAt: entry.firstSeenAt,
292 lastError: current.lastError,36 lastSeenAt: entry.lastSeenAt,
293 });37 currentCpuPercent: entry.currentCpuPercent,
294 });38 avgCpuPercent: entry.avgCpuPercent,
39 maxCpuPercent: entry.maxCpuPercent,
40 cpuSecondsApprox: entry.cpuTimeMs / 1_000,
41 currentRssBytes: entry.currentRssBytes,
42 maxRssBytes: entry.peakRssBytes,
43 sampleCount: entry.sampleCount,
44 }));
45 return {
46 readAt: history.readAt,
47 windowMs: history.windowMs,
48 bucketMs: history.bucketMs,
49 sampleIntervalMs: history.sampleIntervalMs,
50 retainedSampleCount: history.retainedSampleCount,
51 totalCpuSecondsApprox: topProcesses.reduce(
52 (total, entry) => total + entry.cpuSecondsApprox,
53 0,
54 ),
55 buckets: history.buckets.map((bucket) => ({
56 startedAt: bucket.startedAt,
57 endedAt: bucket.endedAt,
58 avgCpuPercent: bucket.avgCpuPercent,
59 maxCpuPercent: bucket.maxCpuPercent,
60 maxRssBytes: bucket.maxRssBytes,
61 maxProcessCount: bucket.maxProcessCount,
62 })),
63 topProcesses,
64 error: history.health.native.lastError.pipe(Option.map((message) => ({ message }))),
65 };
66 }),
67 );
295 68
296 return ProcessResourceMonitor.of({ readHistory });69 return ProcessResourceMonitor.of({ readHistory });
297});70});