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

fix: bound web-probe analyzer tail reads (#785)

Browse Source 
Co-authored-by: Codex <codex@noreply.local>
This commit is contained in:
Lyon
2026-06-24 06:21:07 +08:00
committed by GitHub
parent a223079b20
commit cf5c782997
1 changed files with 77 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/hwlab-node-web-observe-runner-source.ts
+77 -7
View File
@@ -2059,14 +2059,16 @@ const analysisDir = path.join(stateDir, "analysis");
const reportJsonPath = path.join(analysisDir, "report.json");
const reportMdPath = path.join(analysisDir, "report.md");
const jsonlReadIssues = [];
const sourceSamples = await readJsonl(dataFile("samples.jsonl"), { compact: compactSampleForAnalysis });
const samples = analyzeTailSamples > 0 ? sourceSamples.slice(-analyzeTailSamples) : sourceSamples;
const sourceSamples = await readJsonl(dataFile("samples.jsonl"), { compact: compactSampleForAnalysis, tail: analyzeTailSamples });
const samples = sourceSamples;
const sampleWindow = sampleTimeWindow(samples, 60_000);
const control = filterRowsByTimeWindow(await readJsonl(dataFile("control.jsonl")), sampleWindow);
const network = filterRowsByTimeWindow(await readJsonl(dataFile("network.jsonl")), sampleWindow);
const consoleEvents = filterRowsByTimeWindow(await readJsonl(dataFile("console.jsonl")), sampleWindow);
const errors = filterRowsByTimeWindow(await readJsonl(dataFile("errors.jsonl")), sampleWindow);
const artifacts = filterRowsByTimeWindow(await readJsonl(dataFile("artifacts.jsonl")), sampleWindow);
const relatedJsonlTailLimit = analyzeTailSamples > 0 ? Math.max(1200, analyzeTailSamples * 50) : 0;
const smallJsonlTailLimit = analyzeTailSamples > 0 ? Math.max(500, analyzeTailSamples * 8) : 0;
const control = filterRowsByTimeWindow(await readJsonl(dataFile("control.jsonl"), { tail: relatedJsonlTailLimit }), sampleWindow);
const network = filterRowsByTimeWindow(await readJsonl(dataFile("network.jsonl"), { tail: relatedJsonlTailLimit }), sampleWindow);
const consoleEvents = filterRowsByTimeWindow(await readJsonl(dataFile("console.jsonl"), { tail: smallJsonlTailLimit }), sampleWindow);
const errors = filterRowsByTimeWindow(await readJsonl(dataFile("errors.jsonl"), { tail: smallJsonlTailLimit }), sampleWindow);
const artifacts = filterRowsByTimeWindow(await readJsonl(dataFile("artifacts.jsonl"), { tail: smallJsonlTailLimit }), sampleWindow);
const manifest = await readJson(path.join(stateDir, "manifest.json"));
const heartbeat = await readJson(path.join(stateDir, "heartbeat.json"));
@@ -2420,6 +2422,8 @@ function parseAlertThresholds(value) {
}
async function readJsonl(file, options = {}) {
const tailLimit = Number.isFinite(Number(options.tail)) && Number(options.tail) > 0 ? Math.floor(Number(options.tail)) : 0;
if (tailLimit > 0) return readJsonlTail(file, tailLimit, options);
const rows = [];
try {
const input = createReadStream(file, { encoding: "utf8" });
@@ -2446,6 +2450,72 @@ async function readJsonl(file, options = {}) {
}
}
async function readJsonlTail(file, limit, options = {}) {
try {
const lines = await readTailLines(file, limit);
const rows = [];
let lineNo = 0;
for (const rawLine of lines) {
lineNo += 1;
const line = String(rawLine || "").trim();
if (!line) continue;
try {
const parsed = JSON.parse(line);
rows.push(typeof options.compact === "function" ? options.compact(parsed) : parsed);
} catch (error) {
const item = { parseError: true, lineNo, tail: true, rawHash: sha256(line), errorMessage: limitText(error && error.message ? error.message : String(error), 240) };
rows.push(item);
if (jsonlReadIssues.length < 50) jsonlReadIssues.push({ file: path.basename(file), kind: "parse-error", lineNo, tail: true, rawHash: item.rawHash, errorMessage: item.errorMessage });
}
}
return rows;
} catch (error) {
if (error && error.code === "ENOENT") return [];
if (jsonlReadIssues.length < 50) jsonlReadIssues.push({ file: path.basename(file), kind: "read-error", tail: true, code: error && error.code ? String(error.code) : null, errorMessage: limitText(error && error.message ? error.message : String(error), 240) });
return [];
}
}
async function readTailLines(file, limit) {
const info = await stat(file);
if (!info.size || limit <= 0) return [];
const chunkSize = 64 * 1024;
const maxBytes = 32 * 1024 * 1024;
const chunks = [];
let position = info.size;
let readBytes = 0;
let newlineCount = 0;
while (position > 0 && newlineCount <= limit && readBytes < maxBytes) {
const readSize = Math.min(chunkSize, position, maxBytes - readBytes);
position -= readSize;
const chunk = await readFileSlice(file, position, readSize);
chunks.unshift(chunk);
readBytes += chunk.length;
for (let index = 0; index < chunk.length; index += 1) {
if (chunk[index] === 10) newlineCount += 1;
}
}
if (position > 0 && newlineCount <= limit && jsonlReadIssues.length < 50) {
jsonlReadIssues.push({ file: path.basename(file), kind: "tail-scan-truncated", limit, readBytes, fileBytes: info.size });
}
const text = Buffer.concat(chunks).toString("utf8");
let lines = text.split(/\r?\n/u);
if (lines.length > 0 && lines[lines.length - 1] === "") lines.pop();
if (position > 0 && lines.length > 0) lines.shift();
return lines.slice(-limit);
}
async function readFileSlice(file, start, length) {
const chunks = [];
await new Promise((resolve, reject) => {
const stream = createReadStream(file, { start, end: start + length - 1 });
stream.on("data", (chunk) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));
stream.on("error", reject);
stream.on("end", resolve);
});
return Buffer.concat(chunks);
}
function sampleTimeWindow(samples, paddingMs) {
const times = samples
.map((item) => Date.parse(String(item && item.ts || "")))