mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1602 from pikasTech/fix/1585-workbench-monitoring-lag

Browse Source 
fix: 修正 Workbench projection lag 监控口径
This commit is contained in:
Lyon
2026-06-19 14:37:51 +08:00
committed by GitHub
parent 065088427e cd6ee0e6f0
commit db518feac3
4 changed files with 109 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/backend-performance.test.ts
+28
View File
@@ -221,6 +221,34 @@ test("backend performance store records Workbench projection and AgentRun result
assert.doesNotMatch(text, /trc_secret|0123456789abcdef|run_secret|cmd_secret|ses_secret|cnv_secret/u);
});
test("backend performance reports zero projection lag for caught-up terminal traces", () => {
const store = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03", HWLAB_RUNTIME_LANE: "v03", HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601" } });
store.recordWorkbenchProjection({
sourceLatestSeq: 501,
lastProjectedSeq: 501,
sourceLatestEventAt: "2026-06-19T03:00:00.000Z",
projectedLatestEventAt: "2026-06-19T02:00:00.000Z",
terminalSourceAt: "2026-06-19T02:00:00.000Z",
terminalProjectedAt: "2026-06-19T02:00:00.020Z",
status: "completed",
reason: "none",
projectionStatus: "caught-up",
source: "agentrun-v01",
});
const text = store.metricsText();
const lagSumLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_sum") && line.includes('projection_status="caught_up"'));
const lagBucketLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_bucket") && line.includes('projection_status="caught_up"') && line.includes('le="0.05"'));
const stuckLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_stuck_traces") && line.includes('projection_status="caught_up"'));
const terminalDelayBucketLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_terminal_projection_delay_seconds_bucket") && line.includes('projection_status="caught_up"') && line.includes('le="0.05"'));
assert.ok(lagSumLine);
assert.match(lagSumLine, / 0\.000000$/u);
assert.ok(lagBucketLine?.endsWith(" 1"));
assert.ok(stuckLine?.endsWith(" 0"));
assert.ok(terminalDelayBucketLine?.endsWith(" 1"));
});
test("AgentRun read-side sync advances event cursor without forcing result aggregation before terminal evidence", async () => {
const calls: string[] = [];
const agentRunServer = createHttpServer(async (request, response) => {
internal/cloud/backend-performance.ts
+1 -1
View File
@@ -246,7 +246,7 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp
const sourceAt = unixMs(input.sourceLatestEventAt);
const projectedAt = unixMs(input.projectedLatestEventAt);
if (sourceAt !== null && projectedAt !== null) {
const lagSeconds = Math.max(0, (sourceAt - projectedAt) / 1000);
const lagSeconds = isTerminalProjectionStatus(labels.projection_status) ? 0 : Math.max(0, (sourceAt - projectedAt) / 1000);
if (!recordHistogram(projectionLagSecondsSeries, labels, lagSeconds, PROJECTION_LAG_SECONDS_BUCKETS, maxSeries)) incrementDropped("projection_lag_seconds_series_limit");
}
const terminalSourceAt = unixMs(input.terminalSourceAt);
internal/cloud/server-workbench-http.test.ts
+8 -1
View File
@@ -5,6 +5,7 @@ import assert from "node:assert/strict";
import { test } from "bun:test";
import { createCloudApiServer } from "./server.ts";
import { createBackendPerformanceStore } from "./backend-performance.ts";
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts";
import { createWorkbenchTurnProjection, projectionDiagnostics } from "./workbench-turn-projection.ts";
@@ -475,7 +476,8 @@ test("workbench read model projects completed current turn for idle session summ
async ensureBootstrap() {},
async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
};
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results });
const backendPerformanceStore = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03", HWLAB_RUNTIME_LANE: "v03", HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601" } });
const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results, backendPerformanceStore });
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
@@ -501,6 +503,11 @@ test("workbench read model projects completed current turn for idle session summ
assert.equal(turn.body.projectionStatus, "caught-up");
assert.equal(turn.body.sourceRunId, "run_workbench_idle_completed");
assert.equal(turn.body.sourceCommandId, "cmd_workbench_idle_completed");
const metricsText = backendPerformanceStore.metricsText();
const lagCountLine = metricsText.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_count") && line.includes('projection_status="caught_up"'));
const turnGetCountLine = metricsText.split("\n").find((line) => line.startsWith("hwlab_workbench_turn_get_duration_seconds_count") && line.includes('route="/v1/workbench/turns/:traceId"'));
assert.ok(lagCountLine?.endsWith(" 1"));
assert.ok(turnGetCountLine?.endsWith(" 1"));
const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
assert.equal(trace.status, 200);
internal/cloud/server-workbench-http.ts
+72
View File
@@ -201,6 +201,76 @@ function recordWorkbenchTurnReadMetric(options = {}, url, statusCode, body = {},
});
}
function recordWorkbenchProjectionMetric(options = {}, { result = null, trace = null, projection = null, diagnostic = null } = {}) {
const performanceStore = options.backendPerformanceStore ?? options.backendPerformance;
if (!performanceStore?.recordWorkbenchProjection || !projection) return;
const sourceLatestSeq = workbenchProjectionSourceLatestSeq(result, trace, projection);
const lastProjectedSeq = Number(diagnostic?.lastProjectedSeq ?? projection?.lastProjectedSeq ?? 0);
if (!Number.isFinite(sourceLatestSeq) && !Number.isFinite(lastProjectedSeq)) return;
performanceStore.recordWorkbenchProjection({
sourceLatestSeq: Number.isFinite(sourceLatestSeq) ? sourceLatestSeq : lastProjectedSeq,
lastProjectedSeq: Number.isFinite(lastProjectedSeq) ? lastProjectedSeq : 0,
sourceLatestEventAt: workbenchProjectionSourceLatestAt(result, trace, projection),
projectedLatestEventAt: diagnostic?.updatedAt ?? projection?.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null,
terminalSourceAt: workbenchTerminalSourceAt(result, projection),
terminalProjectedAt: diagnostic?.projectionStatus === "caught-up" ? diagnostic?.updatedAt ?? projection?.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null : null,
status: projection?.status ?? result?.status ?? trace?.status ?? "unknown",
reason: workbenchProjectionReason(diagnostic),
projectionStatus: diagnostic?.projectionStatus ?? "unknown",
source: result?.agentRun ? "agentrun_v01" : "workbench_read_model"
});
}
function workbenchProjectionSourceLatestSeq(result = null, trace = null, projection = null) {
const candidates = [
result?.agentRun?.lastSeq,
result?.traceSummary?.agentRun?.lastSeq,
result?.traceSummary?.lastSeq,
result?.providerTrace?.lastSeq,
result?.runnerTrace?.lastEvent?.sourceSeq,
result?.runnerTrace?.lastEvent?.seq,
result?.runnerTrace?.eventCount,
trace?.lastSeq,
trace?.lastEvent?.sourceSeq,
trace?.lastEvent?.seq,
trace?.eventCount,
projection?.lastProjectedSeq,
projection?.eventCount
];
for (const value of candidates) {
const seq = Number(value);
if (Number.isFinite(seq) && seq >= 0) return Math.floor(seq);
}
return NaN;
}
function workbenchProjectionSourceLatestAt(result = null, trace = null, projection = null) {
return result?.traceSummary?.updatedAt
?? result?.agentRun?.updatedAt
?? result?.providerTrace?.updatedAt
?? result?.runnerTrace?.updatedAt
?? trace?.updatedAt
?? result?.updatedAt
?? projection?.updatedAt
?? null;
}
function workbenchTerminalSourceAt(result = null, projection = null) {
if (projection?.terminal !== true) return null;
return result?.traceSummary?.updatedAt
?? result?.agentRun?.updatedAt
?? result?.updatedAt
?? projection?.updatedAt
?? null;
}
function workbenchProjectionReason(diagnostic = null) {
if (diagnostic?.blocker) return diagnostic.blocker.code ?? "projection_blocked";
if (diagnostic?.projectionStatus === "projecting") return "projecting";
if (diagnostic?.projectionStatus === "caught-up") return "none";
return diagnostic?.projectionStatus ?? "unknown";
}
function workbenchTurnDegradedReason(body = {}) {
if (body?.projection?.blocker?.code) return body.projection.blocker.code;
if (body?.blocker?.code) return body.blocker.code;
@@ -342,6 +412,7 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur
return sendJson(response, 404, body);
}
const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection });
recordWorkbenchProjectionMetric(options, { result, trace, projection: turnProjection, diagnostic: projection });
const body = {
ok: true,
status,
@@ -379,6 +450,7 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
const page = traceEventPage(trace, tracePageOptions(url));
const turnProjection = createWorkbenchTurnProjection({ traceId, result, session, trace });
const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection });
recordWorkbenchProjectionMetric(options, { result, trace, projection: turnProjection, diagnostic: projection });
sendJson(response, 200, {
ok: true,
status: "succeeded",