Make Git slowdowns visible in daemon metrics (#2366)
* feat(server): expose daemon Git pressure metrics Separate limiter queue wait from Git execution time and report subscription ownership so accumulating work is visible in the existing runtime log. * fix(server): collect subscriptions in agent metrics Reuse the existing agent snapshot during runtime flushes so WebSocket shutdown does not require an additional AgentManager method call. * test(server): retry transient hub cleanup
This commit is contained in:
Mohamed Boudra
@@ -0,0 +1,74 @@
|
||||
import { describe, expect, test } from "vitest";
|
||||
import { GitCommandRuntimeMetricsWindow } from "./git-command-runtime-metrics.js";
|
||||
|
||||
function createMetricsWindow(concurrencyLimit = 2) {
|
||||
let now = 1_000;
|
||||
return {
|
||||
metrics: new GitCommandRuntimeMetricsWindow(concurrencyLimit, () => now),
|
||||
advance(ms: number) {
|
||||
now += ms;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("GitCommandRuntimeMetricsWindow", () => {
|
||||
test("separates queue wait from execution time", () => {
|
||||
const { metrics, advance } = createMetricsWindow();
|
||||
const command = metrics.submit("status");
|
||||
advance(40);
|
||||
metrics.start(command);
|
||||
advance(15);
|
||||
metrics.finish(command, { success: true, timedOut: false });
|
||||
|
||||
expect(metrics.snapshotAndReset()).toMatchObject({
|
||||
submitted: 1,
|
||||
started: 1,
|
||||
completed: 1,
|
||||
failed: 0,
|
||||
timedOut: 0,
|
||||
queueWaitMs: { count: 1, p50Ms: 40, p95Ms: 40, maxMs: 40 },
|
||||
executionMs: { count: 1, p50Ms: 15, p95Ms: 15, maxMs: 15 },
|
||||
operationsTop: [["status", 1]],
|
||||
});
|
||||
});
|
||||
|
||||
test("reports live queue pressure across window resets", () => {
|
||||
const { metrics, advance } = createMetricsWindow(1);
|
||||
const active = metrics.submit("fetch");
|
||||
metrics.start(active);
|
||||
const pending = metrics.submit("rev-parse");
|
||||
metrics.observeLimiter(1, 1);
|
||||
advance(25);
|
||||
|
||||
expect(metrics.snapshotAndReset()).toMatchObject({
|
||||
concurrencyLimit: 1,
|
||||
active: 1,
|
||||
pending: 1,
|
||||
peakActive: 1,
|
||||
peakPending: 1,
|
||||
oldestPendingMs: 25,
|
||||
submitted: 2,
|
||||
started: 1,
|
||||
});
|
||||
|
||||
advance(10);
|
||||
metrics.finish(active, { success: true, timedOut: false });
|
||||
metrics.start(pending);
|
||||
metrics.observeLimiter(1, 0);
|
||||
advance(5);
|
||||
metrics.finish(pending, { success: false, timedOut: true });
|
||||
|
||||
expect(metrics.snapshotAndReset()).toMatchObject({
|
||||
active: 0,
|
||||
pending: 0,
|
||||
peakActive: 1,
|
||||
peakPending: 1,
|
||||
submitted: 0,
|
||||
started: 1,
|
||||
completed: 2,
|
||||
failed: 1,
|
||||
timedOut: 1,
|
||||
queueWaitMs: { count: 1, p50Ms: 35, p95Ms: 35, maxMs: 35 },
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,147 @@
|
||||
export interface GitCommandDurationStats {
|
||||
count: number;
|
||||
p50Ms: number;
|
||||
p95Ms: number;
|
||||
maxMs: number;
|
||||
}
|
||||
|
||||
export interface GitCommandRuntimeMetricsSnapshot {
|
||||
concurrencyLimit: number;
|
||||
active: number;
|
||||
pending: number;
|
||||
peakActive: number;
|
||||
peakPending: number;
|
||||
oldestPendingMs: number;
|
||||
submitted: number;
|
||||
started: number;
|
||||
completed: number;
|
||||
failed: number;
|
||||
timedOut: number;
|
||||
queueWaitMs: GitCommandDurationStats;
|
||||
executionMs: GitCommandDurationStats;
|
||||
operationsTop: Array<[string, number]>;
|
||||
}
|
||||
|
||||
interface GitCommandRuntimeMetric {
|
||||
queuedAtMs: number;
|
||||
startedAtMs: number | null;
|
||||
}
|
||||
|
||||
type Clock = () => number;
|
||||
|
||||
export class GitCommandRuntimeMetricsWindow {
|
||||
private readonly pendingCommands = new Set<GitCommandRuntimeMetric>();
|
||||
private active = 0;
|
||||
private peakActive = 0;
|
||||
private peakPending = 0;
|
||||
private submittedCount = 0;
|
||||
private startedCount = 0;
|
||||
private completedCount = 0;
|
||||
private failedCount = 0;
|
||||
private timedOutCount = 0;
|
||||
private readonly queueWaitSamples: number[] = [];
|
||||
private readonly executionSamples: number[] = [];
|
||||
private readonly operationCounts = new Map<string, number>();
|
||||
|
||||
constructor(
|
||||
private readonly concurrencyLimit: number,
|
||||
private readonly clock: Clock = Date.now,
|
||||
) {}
|
||||
|
||||
submit(operation: string): GitCommandRuntimeMetric {
|
||||
const metric = { queuedAtMs: this.clock(), startedAtMs: null };
|
||||
this.pendingCommands.add(metric);
|
||||
this.submittedCount += 1;
|
||||
this.operationCounts.set(operation, (this.operationCounts.get(operation) ?? 0) + 1);
|
||||
return metric;
|
||||
}
|
||||
|
||||
observeLimiter(active: number, pending: number): void {
|
||||
this.peakActive = Math.max(this.peakActive, active);
|
||||
this.peakPending = Math.max(this.peakPending, pending);
|
||||
}
|
||||
|
||||
start(metric: GitCommandRuntimeMetric): void {
|
||||
if (!this.pendingCommands.delete(metric)) {
|
||||
return;
|
||||
}
|
||||
const now = this.clock();
|
||||
metric.startedAtMs = now;
|
||||
this.active += 1;
|
||||
this.startedCount += 1;
|
||||
this.peakActive = Math.max(this.peakActive, this.active);
|
||||
this.queueWaitSamples.push(Math.max(0, now - metric.queuedAtMs));
|
||||
}
|
||||
|
||||
finish(metric: GitCommandRuntimeMetric, outcome: { success: boolean; timedOut: boolean }): void {
|
||||
if (metric.startedAtMs === null) {
|
||||
return;
|
||||
}
|
||||
const startedAtMs = metric.startedAtMs;
|
||||
metric.startedAtMs = null;
|
||||
this.active = Math.max(0, this.active - 1);
|
||||
this.completedCount += 1;
|
||||
if (!outcome.success) {
|
||||
this.failedCount += 1;
|
||||
}
|
||||
if (outcome.timedOut) {
|
||||
this.timedOutCount += 1;
|
||||
}
|
||||
this.executionSamples.push(Math.max(0, this.clock() - startedAtMs));
|
||||
}
|
||||
|
||||
snapshotAndReset(
|
||||
limiter = { active: this.active, pending: this.pendingCommands.size },
|
||||
): GitCommandRuntimeMetricsSnapshot {
|
||||
const now = this.clock();
|
||||
const oldestPendingAtMs = Math.min(
|
||||
...Array.from(this.pendingCommands, (metric) => metric.queuedAtMs),
|
||||
);
|
||||
const snapshot: GitCommandRuntimeMetricsSnapshot = {
|
||||
concurrencyLimit: this.concurrencyLimit,
|
||||
active: limiter.active,
|
||||
pending: limiter.pending,
|
||||
peakActive: this.peakActive,
|
||||
peakPending: this.peakPending,
|
||||
oldestPendingMs:
|
||||
limiter.pending > 0 && Number.isFinite(oldestPendingAtMs)
|
||||
? Math.max(0, now - oldestPendingAtMs)
|
||||
: 0,
|
||||
submitted: this.submittedCount,
|
||||
started: this.startedCount,
|
||||
completed: this.completedCount,
|
||||
failed: this.failedCount,
|
||||
timedOut: this.timedOutCount,
|
||||
queueWaitMs: summarizeDurations(this.queueWaitSamples),
|
||||
executionMs: summarizeDurations(this.executionSamples),
|
||||
operationsTop: [...this.operationCounts.entries()]
|
||||
.sort((left, right) => right[1] - left[1] || left[0].localeCompare(right[0]))
|
||||
.slice(0, 12),
|
||||
};
|
||||
|
||||
this.peakActive = limiter.active;
|
||||
this.peakPending = limiter.pending;
|
||||
this.submittedCount = 0;
|
||||
this.startedCount = 0;
|
||||
this.completedCount = 0;
|
||||
this.failedCount = 0;
|
||||
this.timedOutCount = 0;
|
||||
this.queueWaitSamples.length = 0;
|
||||
this.executionSamples.length = 0;
|
||||
this.operationCounts.clear();
|
||||
return snapshot;
|
||||
}
|
||||
}
|
||||
|
||||
function summarizeDurations(samples: number[]): GitCommandDurationStats {
|
||||
if (samples.length === 0) {
|
||||
return { count: 0, p50Ms: 0, p95Ms: 0, maxMs: 0 };
|
||||
}
|
||||
const sorted = [...samples].sort((left, right) => left - right);
|
||||
return {
|
||||
count: sorted.length,
|
||||
p50Ms: Math.round(sorted[Math.floor(sorted.length / 2)] ?? 0),
|
||||
p95Ms: Math.round(sorted[Math.ceil(sorted.length * 0.95) - 1] ?? 0),
|
||||
maxMs: Math.round(sorted[sorted.length - 1] ?? 0),
|
||||
};
|
||||
}
|
||||
@@ -2,6 +2,10 @@ import { existsSync } from "node:fs";
|
||||
import pLimit from "p-limit";
|
||||
import type { Logger } from "pino";
|
||||
import type { ProcessEnvRecord } from "../server/paseo-env.js";
|
||||
import {
|
||||
GitCommandRuntimeMetricsWindow,
|
||||
type GitCommandRuntimeMetricsSnapshot,
|
||||
} from "./git-command-runtime-metrics.js";
|
||||
import { spawnProcess } from "./spawn.js";
|
||||
|
||||
const DEFAULT_TIMEOUT_MS = 30_000;
|
||||
@@ -10,6 +14,7 @@ const DEFAULT_STDERR_LIMIT = 2048;
|
||||
|
||||
const gitConcurrency = parseInt(process.env.PASEO_GIT_CONCURRENCY ?? "8", 10) || 8;
|
||||
const gitLimit = pLimit(gitConcurrency);
|
||||
const gitRuntimeMetrics = new GitCommandRuntimeMetricsWindow(gitConcurrency);
|
||||
|
||||
export interface GitCommandOptions {
|
||||
cwd: string;
|
||||
@@ -81,6 +86,13 @@ export function stopGitCommandMetrics(): GitCommandMetricsSnapshot {
|
||||
};
|
||||
}
|
||||
|
||||
export function snapshotGitCommandRuntimeMetrics(): GitCommandRuntimeMetricsSnapshot {
|
||||
return gitRuntimeMetrics.snapshotAndReset({
|
||||
active: gitLimit.activeCount,
|
||||
pending: gitLimit.pendingCount,
|
||||
});
|
||||
}
|
||||
|
||||
function beginGitCommandMetric(): GitCommandMetricsState | null {
|
||||
const state = gitCommandMetricsState;
|
||||
if (!state) {
|
||||
@@ -123,9 +135,11 @@ export function runGitCommand(
|
||||
args: string[],
|
||||
options: GitCommandOptions,
|
||||
): Promise<GitCommandResult> {
|
||||
return gitLimit(
|
||||
const runtimeMetric = gitRuntimeMetrics.submit(getGitOperation(args));
|
||||
const promise = gitLimit(
|
||||
() =>
|
||||
new Promise<GitCommandResult>((resolve, reject) => {
|
||||
gitRuntimeMetrics.start(runtimeMetric);
|
||||
const timeout = options.timeout ?? DEFAULT_TIMEOUT_MS;
|
||||
const maxOutputBytes = options.maxOutputBytes ?? DEFAULT_MAX_OUTPUT_BYTES;
|
||||
const acceptExitCodes = options.acceptExitCodes ?? [0];
|
||||
@@ -175,24 +189,28 @@ export function runGitCommand(
|
||||
callback();
|
||||
};
|
||||
|
||||
const finishMetricOnce = (metric: GitCommandMetric) => {
|
||||
const finishMetricOnce = (metric: GitCommandMetric, timedOut = false) => {
|
||||
if (metricFinished) return;
|
||||
metricFinished = true;
|
||||
finishGitCommandMetric(metricsState, metric);
|
||||
gitRuntimeMetrics.finish(runtimeMetric, { success: metric.success, timedOut });
|
||||
};
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
const error = new Error(`Git command timed out after ${timeout}ms: ${command}`);
|
||||
child.kill("SIGKILL");
|
||||
finishMetricOnce({
|
||||
args,
|
||||
cwd: options.cwd,
|
||||
startedAtMs: startedAt,
|
||||
durationMs: Date.now() - startedAt,
|
||||
exitCode: null,
|
||||
signal: "SIGKILL",
|
||||
success: false,
|
||||
});
|
||||
finishMetricOnce(
|
||||
{
|
||||
args,
|
||||
cwd: options.cwd,
|
||||
startedAtMs: startedAt,
|
||||
durationMs: Date.now() - startedAt,
|
||||
exitCode: null,
|
||||
signal: "SIGKILL",
|
||||
success: false,
|
||||
},
|
||||
true,
|
||||
);
|
||||
settle(() => reject(error));
|
||||
}, timeout);
|
||||
|
||||
@@ -318,8 +336,14 @@ export function runGitCommand(
|
||||
});
|
||||
}),
|
||||
);
|
||||
gitRuntimeMetrics.observeLimiter(gitLimit.activeCount, gitLimit.pendingCount);
|
||||
return promise;
|
||||
}
|
||||
|
||||
function formatGitCommand(args: string[]): string {
|
||||
return ["git", ...args].join(" ");
|
||||
}
|
||||
|
||||
function getGitOperation(args: string[]): string {
|
||||
return args[0] === "-c" ? (args[2] ?? "unknown") : (args[0] ?? "unknown");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user