From d9530e6577bbde1c8d291f1c07bae1b123b7500b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=B3=E6=9C=A8?= Date: Fri, 10 Jul 2026 13:10:38 +0800 Subject: [PATCH 1/2] fix(sls): add structured flush failure diagnostics Classify SLS send failures across AK and webtracking paths, attach safe diagnostic context to alarms, failed logs, and flusher metrics, and document the new fields with focused unit coverage. Co-Authored-By: Claude Opus 4.6 (1M context) --- assets/skills/references/sls-diagnostics.md | 41 +++- docs/sls-output.md | 17 ++ docs/zh-CN/sls-output.md | 17 ++ src/flushers/sls-flusher.ts | 145 ++++++++++++--- src/flushers/sls-transport.ts | 186 +++++++++++++++++-- src/metrics/alarm-manager.ts | 50 ++++- src/metrics/metrics-collector.ts | 13 +- tests/unit/flushers/sls-flusher.test.ts | 100 ++++++++++ tests/unit/flushers/sls-transport.test.ts | 77 +++++++- tests/unit/metrics/alarm-manager.test.ts | 42 +++++ tests/unit/metrics/metrics-collector.test.ts | 26 +++ tests/unit/metrics/metrics-writer.test.ts | 48 +++++ 12 files changed, 718 insertions(+), 44 deletions(-) diff --git a/assets/skills/references/sls-diagnostics.md b/assets/skills/references/sls-diagnostics.md index 32fafce2..21c5d5db 100644 --- a/assets/skills/references/sls-diagnostics.md +++ b/assets/skills/references/sls-diagnostics.md @@ -19,6 +19,7 @@ SlsFlusher 按 endpoint 分别发送 ↓ 发送失败（3 次重试后仍失败） → 写入 ~/.loongsuite-pilot/sls-failed-logs/{endpoint-name}.jsonl + → 记录 FLUSH_SEND_ALARM（含 failure_class / status_code / endpoint_host 等诊断字段） ``` 关键事实： @@ -27,6 +28,7 @@ SlsFlusher 按 endpoint 分别发送 - 失败自动重试 3 次（指数退避：1s → 2s → 4s），可重试的状态码：408、429、500、502、503、504 - 最终失败的日志持久化到 `sls-failed-logs/` 目录（不会丢失） - 支持多 endpoint 同时发送（互不阻塞） +- `FLUSH_SEND_ALARM` 保持原 alarm_type，同时附加结构化诊断字段；`endpoint_host` 只包含安全解析后的 hostname，不包含 path/query/userinfo；`reason` 已限长并去敏 --- @@ -156,11 +158,17 @@ tail -5 ~/.loongsuite-pilot/sls-failed-logs/*.jsonl 2>/dev/null { "ts": 1716000000000, // 失败时间戳 "endpoint": "internal-sls", // endpoint 名称 + "endpoint_host": "...", // 仅 hostname，不含完整 endpoint URL + "mode": "webtracking", // webtracking 或 ak "kind": "agentActivity", // 数据类型 "project": "...", // SLS project "logstore": "...", // SLS logstore + "failure_class": "permission_denied", + "status_code": 403, + "retryable": false, + "reason": "HTTP 403 ...", // 限长、去敏后的错误摘要 "logGroup": { ... }, // 完整的日志内容（可用于手动重放） - "error": "..." // 错误详情 + "error": "..." // 与 reason 一致的安全错误摘要 } ``` @@ -168,6 +176,34 @@ tail -5 ~/.loongsuite-pilot/sls-failed-logs/*.jsonl 2>/dev/null ### 4.1 常见失败原因分析 +`FLUSH_SEND_ALARM` 和 failed-log 都包含以下诊断字段： + +| 字段 | 含义 | +|------|------| +| `endpoint_name` / `endpoint` | SLS endpoint 配置名，也是 failed-log 文件名来源 | +| `endpoint_host` | 从 endpoint 安全解析出的 hostname，不透传完整 URL | +| `mode` | `webtracking` 或 `ak` | +| `project` / `logstore` | 目标 SLS project / logstore | +| `failure_class` | 归一化失败类型 | +| `status_code` | HTTP 或 SDK 返回的状态码；无状态码时为空 | +| `retryable` | 该错误是否属于重试类错误 | +| `reason` | 限长、去敏后的错误摘要 | + +`failure_class` 常见取值： + +| failure_class | 常见原因 | 建议处理 | +|---|---|---| +| `auth_failed` | 401、AK/SK 无效、签名错误 | 检查 AK/SK 是否正确、是否过期 | +| `permission_denied` | 403、RAM 权限不足 | 确认具备 SLS 写入权限 | +| `not_found` | 404、project/logstore 不存在 | 核对 project/logstore 拼写和地域 | +| `quota_throttle` | 429、quota/throttle/server busy | 检查写入限额、QPS、SLS 侧限流 | +| `network_timeout` | 408、AbortError、TimeoutError、ETIMEDOUT | 检查网络连通性、代理、防火墙 | +| `network_refused` | ECONNREFUSED、ECONNRESET、socket hang up | 检查 endpoint 可达性和网络链路 | +| `server_error` | 5xx、InternalServerError | 关注 SLS 服务状态，必要时重试观察 | +| `payload_too_large` | 413、请求体过大 | 检查单批大小和单条日志体积 | +| `bad_request` | 其他 4xx | 检查 endpoint/project/logstore 和请求格式 | +| `unknown` | 未识别错误 | 结合服务日志和 failed-log 的 `reason` 排查 | + ```bash # 提取失败原因统计 python3 -c " @@ -178,7 +214,8 @@ for f in glob.glob('$HOME/.loongsuite-pilot/sls-failed-logs/*.jsonl'): for line in open(f): try: r = json.loads(line) - errors[r.get('error','unknown')[:80]] += 1 + key = f"{r.get('failure_class','unknown')} status={r.get('status_code','')}" + errors[key] += 1 except: pass for err, cnt in errors.most_common(10): print(f'{cnt:5d} {err}') diff --git a/docs/sls-output.md b/docs/sls-output.md index 974c7546..79d3e5a2 100644 --- a/docs/sls-output.md +++ b/docs/sls-output.md @@ -114,6 +114,23 @@ Local JSONL output can help confirm whether collection itself is working before tail -f ~/.loongsuite-pilot/logs/output/*.jsonl ``` +## Failure Diagnostics + +When an SLS endpoint still fails after the built-in retries, Pilot keeps the existing `FLUSH_SEND_ALARM` alarm type and adds structured diagnostic fields: + +| Field | Description | +|-------|-------------| +| `endpoint_name` | Configured endpoint name. | +| `endpoint_host` | Hostname safely parsed from the endpoint URL. It does not include path, query, userinfo, or the full endpoint string. | +| `mode` | `webtracking` or `ak`. | +| `project` / `logstore` | Target SLS project and logstore. | +| `failure_class` | Normalized class such as `auth_failed`, `permission_denied`, `not_found`, `quota_throttle`, `network_timeout`, `network_refused`, `server_error`, `bad_request`, `payload_too_large`, or `unknown`. | +| `status_code` | HTTP/SDK status code when available. | +| `retryable` | Whether the failure matches Pilot's retryable set. | +| `reason` | Redacted, length-limited failure summary. | + +The same diagnostic fields are written to `~/.loongsuite-pilot/sls-failed-logs/.jsonl` together with the failed `logGroup`. `FLUSH_QUOTA_ALARM` is still emitted for 429 throttling; retry count, batch size, failed-log path, and multi-endpoint isolation are unchanged. + ## Privacy Notes SLS is a remote destination. Review [Agent Configuration](agents.md) for content capture controls and [Data Masking](masking.md) for secret masking before enabling SLS in sensitive environments. diff --git a/docs/zh-CN/sls-output.md b/docs/zh-CN/sls-output.md index 1d51a107..c43278a7 100644 --- a/docs/zh-CN/sls-output.md +++ b/docs/zh-CN/sls-output.md @@ -114,6 +114,23 @@ ls ~/.loongsuite-pilot/sls-failed-logs/ tail -f ~/.loongsuite-pilot/logs/output/*.jsonl ``` +## 失败诊断 + +SLS endpoint 在内置重试后仍失败时，Pilot 会保留既有 `FLUSH_SEND_ALARM` alarm_type，并附加结构化诊断字段： + +| 字段 | 说明 | +|------|------| +| `endpoint_name` | 配置中的 endpoint 名称。 | +| `endpoint_host` | 从 endpoint URL 安全解析出的 hostname，不包含 path、query、userinfo 或完整 endpoint 字符串。 | +| `mode` | `webtracking` 或 `ak`。 | +| `project` / `logstore` | 目标 SLS project 和 logstore。 | +| `failure_class` | 归一化失败类型，例如 `auth_failed`、`permission_denied`、`not_found`、`quota_throttle`、`network_timeout`、`network_refused`、`server_error`、`bad_request`、`payload_too_large`、`unknown`。 | +| `status_code` | 可获得时记录 HTTP/SDK 状态码。 | +| `retryable` | 是否属于 Pilot 的可重试错误集合。 | +| `reason` | 已去敏、限长的失败摘要。 | + +同样的诊断字段会写入 `~/.loongsuite-pilot/sls-failed-logs/.jsonl`，并保留失败的 `logGroup` 便于排查。429 限流仍会同时产生 `FLUSH_QUOTA_ALARM`；重试次数、batch 大小、失败落盘路径和多 endpoint 隔离语义均不变。 + ## 隐私说明 SLS 是远端输出目标。敏感环境中开启前，请先查看 [Agent 配置](agents.md) 的内容采集控制和 [数据脱敏](masking.md)。 diff --git a/src/flushers/sls-flusher.ts b/src/flushers/sls-flusher.ts index b895a7fd..7baed157 100644 --- a/src/flushers/sls-flusher.ts +++ b/src/flushers/sls-flusher.ts @@ -17,12 +17,15 @@ import { HttpError, postWebtracking, isRetryable, + classifySlsFailure, + extractEndpointHost, RETRY_MAX_ATTEMPTS, RETRY_BASE_DELAY_MS, WEBTRACKING_TIMEOUT_MS, WEBTRACKING_MAX_BODY_BYTES, WEBTRACKING_MAX_LOGS, RETRYABLE_STATUS_CODES, + type SlsFailureDiagnostics, } from './sls-transport.js'; const HOSTNAME = os.hostname(); @@ -36,6 +39,12 @@ interface QueuedLog { agentType?: string; } +interface FlushResult { + sentEntries: number; + failedEntries: number; + lastFailure?: SlsFailureDiagnostics; +} + const logger = createLogger('SlsFlusher'); export interface EndpointCounter { @@ -50,6 +59,10 @@ export interface EndpointCounter { endpoint: string; project: string; logstore: string; + lastFailureClass: string; + lastFailureStatusCode: string; + lastFailureTime: string; + consecutiveFailures: number; } export class SlsFlusher extends BaseFlusher { @@ -76,6 +89,8 @@ export class SlsFlusher extends BaseFlusher { inEntries: 0, inBytes: 0, outEntries: 0, outFailed: 0, totalDelayMs: 0, lastFlushTime: '', startTime: '', mode: ep.mode, endpoint: ep.endpoint, project: ep.project, logstore: ep.logstore, + lastFailureClass: '', lastFailureStatusCode: '', lastFailureTime: '', + consecutiveFailures: 0, }); } } @@ -148,26 +163,47 @@ export class SlsFlusher extends BaseFlusher { const send = endpoint.mode === 'ak' ? this.flushViaAk(endpoint, logs) : this.flushViaWebtracking(endpoint, logs); - return send.then(() => { + return send.then((result) => { if (counter) { - counter.outEntries += logs.length; + counter.outEntries += result.sentEntries; + counter.outFailed += result.failedEntries; counter.totalDelayMs += Date.now() - startMs; - counter.lastFlushTime = formatTime(new Date()); + if (result.sentEntries > 0) { + counter.lastFlushTime = formatTime(new Date()); + } + if (result.failedEntries > 0 && result.lastFailure) { + this.recordCounterFailure(counter, result.lastFailure); + } else if (result.failedEntries === 0) { + counter.consecutiveFailures = 0; + } } }).catch(err => { + const diagnostics = classifySlsFailure(err); if (counter) { counter.outFailed += logs.length; counter.totalDelayMs += Date.now() - startMs; + this.recordCounterFailure(counter, diagnostics); } logger.error('SLS endpoint flush failed', { endpoint: endpoint.name, - error: String(err), + failureClass: diagnostics.failure_class, + statusCode: diagnostics.status_code, + error: diagnostics.reason, }); }); }); await Promise.all(tasks); } + private recordCounterFailure(counter: EndpointCounter, diagnostics: SlsFailureDiagnostics): void { + counter.lastFailureClass = diagnostics.failure_class; + counter.lastFailureStatusCode = diagnostics.status_code === undefined + ? '' + : String(diagnostics.status_code); + counter.lastFailureTime = formatTime(new Date()); + counter.consecutiveFailures++; + } + private resolveServiceName(agentType?: string): string { if (!this.serviceName) return ''; return agentType ? `${this.serviceName}-${agentType}` : this.serviceName; @@ -194,7 +230,7 @@ export class SlsFlusher extends BaseFlusher { } } - private async flushViaAk(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { + private async flushViaAk(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { this.warnIfMixedAgentTypes(logs); const now = Math.floor(Date.now() / 1000); const agentType = logs[0]?.agentType; @@ -223,45 +259,57 @@ export class SlsFlusher extends BaseFlusher { logstore: endpoint.logstore, count: logs.length, }); - return; + return { sentEntries: logs.length, failedEntries: 0 }; } catch (err) { lastErr = err; if (!isRetryable(err) || attempt === RETRY_MAX_ATTEMPTS - 1) break; + const diagnostics = classifySlsFailure(err); const delay = RETRY_BASE_DELAY_MS * 2 ** attempt; logger.warn('SLS ak send retrying', { endpoint: endpoint.name, attempt: attempt + 1, delayMs: delay, - error: String(err), + failureClass: diagnostics.failure_class, + statusCode: diagnostics.status_code, + error: diagnostics.reason, }); await this.sleep(delay); } } + const diagnostics = classifySlsFailure(lastErr); logger.error('SLS send failed after retries', { endpoint: endpoint.name, - error: String(lastErr), + failureClass: diagnostics.failure_class, + statusCode: diagnostics.status_code, + error: diagnostics.reason, }); this.alarmManager?.record( 'FLUSH_SEND_ALARM', '2', - `SLS ak send failed: ${String(lastErr)}`, - { endpoint_name: endpoint.name }, + `SLS ak send failed: ${diagnostics.reason}`, + this.buildFailureContext(endpoint, diagnostics), ); - if (lastErr instanceof HttpError && lastErr.status === 429) { + if (diagnostics.status_code === 429) { this.alarmManager?.record( 'FLUSH_QUOTA_ALARM', '2', `SLS endpoint throttled (429)`, - { endpoint_name: endpoint.name }, + this.buildFailureContext(endpoint, diagnostics), ); } - await this.persistFailedLogs(endpoint, logGroup, lastErr); + await this.persistFailedLogs(endpoint, logGroup, diagnostics); + return { sentEntries: 0, failedEntries: logs.length, lastFailure: diagnostics }; } - private async flushViaWebtracking(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { + private async flushViaWebtracking(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { const chunks = this.splitForWebtracking(logs); + const result: FlushResult = { sentEntries: 0, failedEntries: 0 }; for (const chunk of chunks) { - await this.postWebtracking(endpoint, chunk); + const chunkResult = await this.postWebtracking(endpoint, chunk); + result.sentEntries += chunkResult.sentEntries; + result.failedEntries += chunkResult.failedEntries; + if (chunkResult.lastFailure) result.lastFailure = chunkResult.lastFailure; } + return result; } private splitForWebtracking(logs: QueuedLog[]): QueuedLog[][] { @@ -290,7 +338,7 @@ export class SlsFlusher extends BaseFlusher { return chunks; } - private async postWebtracking(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { + private async postWebtracking(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { this.warnIfMixedAgentTypes(logs); const agentType = logs[0]?.agentType; const body = { @@ -334,7 +382,7 @@ export class SlsFlusher extends BaseFlusher { logstore: endpoint.logstore, count: logs.length, }); - return; + return { sentEntries: logs.length, failedEntries: 0 }; } } catch (err) { lastErr = err; @@ -342,46 +390,91 @@ export class SlsFlusher extends BaseFlusher { if (attempt === RETRY_MAX_ATTEMPTS - 1) break; } + const diagnostics = classifySlsFailure(lastErr); const delay = RETRY_BASE_DELAY_MS * 2 ** attempt; logger.warn('SLS webtracking retrying', { endpoint: endpoint.name, attempt: attempt + 1, delayMs: delay, - error: String(lastErr), + failureClass: diagnostics.failure_class, + statusCode: diagnostics.status_code, + error: diagnostics.reason, }); await this.sleep(delay); } + const diagnostics = classifySlsFailure(lastErr); logger.error('SLS webtracking send failed after retries', { endpoint: endpoint.name, - error: String(lastErr), + failureClass: diagnostics.failure_class, + statusCode: diagnostics.status_code, + error: diagnostics.reason, }); this.alarmManager?.record( 'FLUSH_SEND_ALARM', '2', - `SLS webtracking send failed: ${String(lastErr)}`, - { endpoint_name: endpoint.name }, + `SLS webtracking send failed: ${diagnostics.reason}`, + this.buildFailureContext(endpoint, diagnostics), ); - if (lastErr instanceof HttpError && lastErr.status === 429) { + if (diagnostics.status_code === 429) { this.alarmManager?.record( 'FLUSH_QUOTA_ALARM', '2', `SLS endpoint throttled (429)`, - { endpoint_name: endpoint.name }, + this.buildFailureContext(endpoint, diagnostics), ); } - await this.persistFailedLogs(endpoint, body, lastErr); + await this.persistFailedLogs(endpoint, body, diagnostics); + return { sentEntries: 0, failedEntries: logs.length, lastFailure: diagnostics }; + } + + private buildFailureContext(endpoint: SlsEndpoint, diagnostics: SlsFailureDiagnostics): { + endpoint_name: string; + mode: string; + endpoint_host: string; + project: string; + logstore: string; + failure_class: string; + status_code?: string; + retryable: string; + reason: string; + } { + return { + endpoint_name: endpoint.name, + mode: endpoint.mode, + endpoint_host: extractEndpointHost(endpoint.endpoint), + project: endpoint.project, + logstore: endpoint.logstore, + failure_class: diagnostics.failure_class, + ...(diagnostics.status_code === undefined + ? {} + : { status_code: String(diagnostics.status_code) }), + retryable: String(diagnostics.retryable), + reason: diagnostics.reason, + }; } - private async persistFailedLogs(endpoint: SlsEndpoint, logGroup: unknown, err: unknown): Promise { + private async persistFailedLogs( + endpoint: SlsEndpoint, + logGroup: unknown, + diagnostics: SlsFailureDiagnostics, + ): Promise { await ensureDir(this.failedLogDir); const filePath = path.join(this.failedLogDir, `${endpoint.name}.jsonl`); const line = JSON.stringify({ ts: Date.now(), endpoint: endpoint.name, + endpoint_host: extractEndpointHost(endpoint.endpoint), + mode: endpoint.mode, project: endpoint.project, logstore: endpoint.logstore, kind: endpoint.kind, + failure_class: diagnostics.failure_class, + ...(diagnostics.status_code === undefined + ? {} + : { status_code: diagnostics.status_code }), + retryable: diagnostics.retryable, + reason: diagnostics.reason, logGroup, - error: String(err), + error: diagnostics.reason, }); await appendLine(filePath, line); } diff --git a/src/flushers/sls-transport.ts b/src/flushers/sls-transport.ts index b9c7741d..841e6ab2 100644 --- a/src/flushers/sls-transport.ts +++ b/src/flushers/sls-transport.ts @@ -11,9 +11,29 @@ export const RETRY_MAX_ATTEMPTS = 3; export const RETRY_BASE_DELAY_MS = 1000; export const RETRYABLE_STATUS_CODES = new Set([408, 429, 500, 502, 503, 504]); +const SLS_FAILURE_REASON_MAX_LENGTH = 240; + +export type SlsFailureClass = + | 'network_timeout' + | 'network_refused' + | 'quota_throttle' + | 'auth_failed' + | 'permission_denied' + | 'not_found' + | 'server_error' + | 'bad_request' + | 'payload_too_large' + | 'unknown'; + +export interface SlsFailureDiagnostics { + failure_class: SlsFailureClass; + status_code?: number; + retryable: boolean; + reason: string; +} export class HttpError extends Error { - constructor(readonly status: number, body: string) { + constructor(readonly status: number, readonly body: string) { super(`${status} ${body}`); } } @@ -66,17 +86,161 @@ export function splitForWebtracking( } export function isRetryable(err: unknown): boolean { - if (err instanceof HttpError) return RETRYABLE_STATUS_CODES.has(err.status); - const msg = String(err); + return classifySlsFailure(err).retryable; +} + +export function extractEndpointHost(endpoint: string): string { + const raw = endpoint.trim(); + if (!raw) return ''; + const candidate = /^https?:\/\//i.test(raw) ? raw : `https://${raw}`; + try { + return new URL(candidate).hostname; + } catch { + return ''; + } +} + +export function classifySlsFailure(err: unknown): SlsFailureDiagnostics { + const statusCode = extractStatusCode(err); + const text = extractFailureText(err); + const lower = text.toLowerCase(); + const failureClass = classifyByStatus(statusCode) ?? classifyByText(lower); + const retryable = isRetryableFailure(failureClass, statusCode, lower); + + return { + failure_class: failureClass, + ...(statusCode === undefined ? {} : { status_code: statusCode }), + retryable, + reason: sanitizeFailureReason(text || 'unknown error'), + }; +} + +export function sanitizeFailureReason(reason: string): string { + const redacted = reason + .replace(/\s+/g, ' ') + .replace(/https?:\/\/[^\s"'<>]+/gi, (match) => { + try { + const url = new URL(match); + return `[url:${url.hostname}]`; + } catch { + return '[url:redacted]'; + } + }) + .replace(/\b(Bearer|Basic)\s+[A-Za-z0-9._~+/=-]+/gi, '$1 [REDACTED]') + .replace(/\b(access[-_]?key(?:id|secret)?|accessKeyId|accessKeySecret|securityToken|authorization|signature|password|secret|token)\b["']?\s*[:=]\s*["']?[^"',\s&}\]]+/gi, '$1=[REDACTED]') + .replace(/\b(?:AKIA|LTAI)[A-Z0-9]{12,}\b/gi, '[ACCESS_KEY_REDACTED]') + .trim(); + + if (redacted.length <= SLS_FAILURE_REASON_MAX_LENGTH) return redacted; + return `${redacted.slice(0, SLS_FAILURE_REASON_MAX_LENGTH - 3)}...`; +} + +function extractStatusCode(err: unknown): number | undefined { + if (err instanceof HttpError) return err.status; + if (!err || typeof err !== 'object') return undefined; + + const record = err as Record; + for (const key of ['statusCode', 'status', 'httpStatusCode']) { + const raw = record[key]; + if (typeof raw === 'number') return raw; + if (typeof raw === 'string' && /^\d+$/.test(raw)) return Number(raw); + } + return undefined; +} + +function extractFailureText(err: unknown): string { + if (err instanceof HttpError) { + return `HTTP ${err.status}: ${err.body}`; + } + if (!err || typeof err !== 'object') return String(err); + + const record = err as Record; + const parts: string[] = []; + for (const key of ['name', 'code', 'statusCode', 'status', 'message']) { + const value = record[key]; + if (value !== undefined && value !== null && value !== '') { + parts.push(`${key}=${String(value)}`); + } + } + if (parts.length > 0) return parts.join(' '); + return String(err); +} + +function classifyByStatus(statusCode?: number): SlsFailureClass | undefined { + if (statusCode === undefined) return undefined; + if (statusCode === 408) return 'network_timeout'; + if (statusCode === 401) return 'auth_failed'; + if (statusCode === 403) return 'permission_denied'; + if (statusCode === 404) return 'not_found'; + if (statusCode === 413) return 'payload_too_large'; + if (statusCode === 429) return 'quota_throttle'; + if (statusCode >= 500 && statusCode <= 599) return 'server_error'; + if (statusCode >= 400 && statusCode <= 499) return 'bad_request'; + return 'unknown'; +} + +function classifyByText(lower: string): SlsFailureClass { + if ( + lower.includes('aborterror') || + lower.includes('timeouterror') || + lower.includes('etimedout') || + lower.includes('timeout') + ) return 'network_timeout'; + if ( + lower.includes('econnrefused') || + lower.includes('econnreset') || + lower.includes('socket hang up') || + lower.includes('network') + ) return 'network_refused'; + if ( + lower.includes('throttl') || + lower.includes('quota') || + lower.includes('serverbusy') + ) return 'quota_throttle'; + if ( + lower.includes('invalidaccesskey') || + lower.includes('signature') || + lower.includes('unauthorized') + ) return 'auth_failed'; + if ( + lower.includes('accessdenied') || + lower.includes('forbidden') || + lower.includes('permission') + ) return 'permission_denied'; + if ( + lower.includes('notfound') || + lower.includes('not found') || + lower.includes('logstorenotexist') || + lower.includes('projectnotexist') + ) return 'not_found'; + if ( + lower.includes('payloadtoolarge') || + lower.includes('entity too large') || + lower.includes('request body too large') + ) return 'payload_too_large'; + if ( + lower.includes('internalservererror') || + lower.includes('internal server error') || + lower.includes('serviceunavailable') + ) return 'server_error'; + return 'unknown'; +} + +function isRetryableFailure( + _failureClass: SlsFailureClass, + statusCode: number | undefined, + lower: string, +): boolean { + if (statusCode !== undefined) return RETRYABLE_STATUS_CODES.has(statusCode); return ( - msg.includes('ECONNRESET') || - msg.includes('ETIMEDOUT') || - msg.includes('ECONNREFUSED') || - msg.includes('socket hang up') || - msg.includes('network') || - msg.includes('TimeoutError') || - msg.includes('InternalServerError') || - msg.includes('ServerBusy') + lower.includes('econnreset') || + lower.includes('etimedout') || + lower.includes('econnrefused') || + lower.includes('socket hang up') || + lower.includes('network') || + lower.includes('timeouterror') || + lower.includes('internalservererror') || + lower.includes('serverbusy') ); } diff --git a/src/metrics/alarm-manager.ts b/src/metrics/alarm-manager.ts index 6e20a70f..b4563e9a 100644 --- a/src/metrics/alarm-manager.ts +++ b/src/metrics/alarm-manager.ts @@ -22,6 +22,14 @@ export type AlarmType = export interface AlarmContext { input_name?: string; endpoint_name?: string; + mode?: string; + endpoint_host?: string; + project?: string; + logstore?: string; + failure_class?: string; + status_code?: string | number; + retryable?: string | boolean; + reason?: string; } export interface AlarmEntry { @@ -34,6 +42,14 @@ export interface AlarmEntry { ver: string; input_name?: string; endpoint_name?: string; + mode?: string; + endpoint_host?: string; + project?: string; + logstore?: string; + failure_class?: string; + status_code?: string; + retryable?: string; + reason?: string; __time__: number; } @@ -58,11 +74,18 @@ export class AlarmManager { } record(type: AlarmType, level: AlarmLevel, message: string, context?: AlarmContext): void { - const key = `${type}_${context?.input_name ?? ''}_${context?.endpoint_name ?? ''}`; + const key = [ + type, + context?.input_name ?? '', + context?.endpoint_name ?? '', + context?.failure_class ?? '', + context?.status_code ?? '', + ].join('_'); const existing = this.alarms.get(key); if (existing) { existing.count++; existing.message = message; + existing.context = context; } else { this.alarms.set(key, { alarmType: type, level, message, count: 1, context }); } @@ -86,12 +109,33 @@ export class AlarmManager { ver: this.version, __time__: now, }; - if (item.context?.input_name) entry.input_name = item.context.input_name; - if (item.context?.endpoint_name) entry.endpoint_name = item.context.endpoint_name; + this.copyContext(entry, item.context); entries.push(entry); } this.alarms.clear(); return entries; } + + private copyContext(entry: AlarmEntry, context?: AlarmContext): void { + if (!context) return; + const target = entry as unknown as Record; + for (const key of [ + 'input_name', + 'endpoint_name', + 'mode', + 'endpoint_host', + 'project', + 'logstore', + 'failure_class', + 'status_code', + 'retryable', + 'reason', + ] as const) { + const value = context[key]; + if (value !== undefined && value !== '') { + target[key] = String(value); + } + } + } } diff --git a/src/metrics/metrics-collector.ts b/src/metrics/metrics-collector.ts index c19067c2..98966cdf 100644 --- a/src/metrics/metrics-collector.ts +++ b/src/metrics/metrics-collector.ts @@ -92,6 +92,10 @@ export interface FlusherMetrics { out_failed_entries_total: string; total_delay_ms: string; last_flush_time: string; + last_failure_class: string; + last_failure_status_code: string; + last_failure_time: string; + consecutive_failures: string; start_time: string; __time__: number; } @@ -104,6 +108,10 @@ export interface FlusherStats { totalDelayMs: number; lastFlushTime: string; startTime: string; + lastFailureClass?: string; + lastFailureStatusCode?: string; + lastFailureTime?: string; + consecutiveFailures?: number; } export interface InputStats { @@ -293,6 +301,10 @@ export class MetricsCollector { out_failed_entries_total: String(stats.outFailed), total_delay_ms: String(stats.totalDelayMs), last_flush_time: stats.lastFlushTime, + last_failure_class: stats.lastFailureClass ?? '', + last_failure_status_code: stats.lastFailureStatusCode ?? '', + last_failure_time: stats.lastFailureTime ?? '', + consecutive_failures: String(stats.consecutiveFailures ?? 0), start_time: stats.startTime, __time__: now, }); @@ -474,4 +486,3 @@ function countVersions(dataDir: string): number { return 0; } } - diff --git a/tests/unit/flushers/sls-flusher.test.ts b/tests/unit/flushers/sls-flusher.test.ts index a094773d..07a1bb02 100644 --- a/tests/unit/flushers/sls-flusher.test.ts +++ b/tests/unit/flushers/sls-flusher.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import { ClientType, ActionType } from '../../../src/types/index.js'; import type { SlsFlusherConfig, SlsEndpoint } from '../../../src/types/index.js'; import { buildTestEntry } from '../../helpers/fixture-builder.js'; +import { AlarmManager } from '../../../src/metrics/alarm-manager.js'; const mockPostLogStoreLogs = vi.fn().mockResolvedValue(undefined); const mockAppendLine = vi.fn().mockResolvedValue(undefined); @@ -192,6 +193,105 @@ describe('SlsFlusher', () => { expect(parsed.kind).toBe('agentActivity'); expect(parsed.endpoint).toBe('activity'); }); + + it('adds sanitized AK failure diagnostics to alarm, failed log, and counters', async () => { + const alarmManager = new AlarmManager({ ip: '127.0.0.1', version: '1.0.0', userId: 'u1' }); + flusher.setAlarmManager(alarmManager); + const err = Object.assign( + new Error('accessKeySecret=plainsecret missing log:PostLogStoreLogs permission'), + { statusCode: 403, code: 'AccessDenied', name: 'SlsError' }, + ); + mockPostLogStoreLogs.mockRejectedValueOnce(err); + + await flusher.send(buildTestEntry()); + await flusher.flush(); + + const alarms = alarmManager.serialize(); + expect(alarms).toHaveLength(1); + expect(alarms[0]).toMatchObject({ + alarm_type: 'FLUSH_SEND_ALARM', + endpoint_name: 'activity', + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + mode: 'ak', + project: 'proj-a', + logstore: 'store-a', + failure_class: 'permission_denied', + status_code: '403', + retryable: 'false', + }); + expect(alarms[0].reason).not.toContain('plainsecret'); + expect(JSON.stringify(alarms[0])).not.toContain('https://cn-hangzhou.log.aliyuncs.com'); + + expect(mockAppendLine).toHaveBeenCalledOnce(); + const parsed = JSON.parse(mockAppendLine.mock.calls[0][1]); + expect(parsed).toMatchObject({ + endpoint: 'activity', + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + mode: 'ak', + failure_class: 'permission_denied', + status_code: 403, + retryable: false, + }); + expect(parsed.reason).not.toContain('plainsecret'); + expect(parsed.error).toBe(parsed.reason); + + const counter = flusher.getEndpointCounters().get('activity')!; + expect(counter.outEntries).toBe(0); + expect(counter.outFailed).toBe(1); + expect(counter.lastFailureClass).toBe('permission_denied'); + expect(counter.lastFailureStatusCode).toBe('403'); + expect(counter.lastFailureTime).not.toBe(''); + expect(counter.consecutiveFailures).toBe(1); + }); + + it('keeps 429 quota alarm and records webtracking diagnostics', async () => { + const alarmManager = new AlarmManager({ ip: '127.0.0.1', version: '1.0.0', userId: 'u1' }); + const fetchSpy = vi.fn().mockResolvedValue({ + ok: false, + status: 429, + text: async () => `quota exceeded AccessKeyId=LTAI1234567890SECRET ${'x'.repeat(400)}`, + }); + vi.stubGlobal('fetch', fetchSpy); + + flusher = new SlsFlusher(makeConfig({ + endpoints: [ + { + name: 'wt', + endpoint: 'https://cn-hangzhou.log.aliyuncs.com/path?token=abc', + project: 'proj-a', + logstore: 'store-a', + kind: 'agentActivity', + mode: 'webtracking', + }, + ], + }), '/tmp/data'); + flusher.setAlarmManager(alarmManager); + + await flusher.send(buildTestEntry()); + const flushPromise = flusher.flush(); + await vi.runAllTimersAsync(); + await flushPromise; + + const alarms = alarmManager.serialize(); + const sendAlarm = alarms.find(a => a.alarm_type === 'FLUSH_SEND_ALARM')!; + const quotaAlarm = alarms.find(a => a.alarm_type === 'FLUSH_QUOTA_ALARM')!; + expect(sendAlarm).toBeDefined(); + expect(quotaAlarm).toBeDefined(); + expect(sendAlarm.failure_class).toBe('quota_throttle'); + expect(sendAlarm.status_code).toBe('429'); + expect(sendAlarm.retryable).toBe('true'); + expect(sendAlarm.endpoint_host).toBe('cn-hangzhou.log.aliyuncs.com'); + expect(sendAlarm.reason.length).toBeLessThanOrEqual(240); + expect(sendAlarm.reason).not.toContain('LTAI1234567890SECRET'); + expect(JSON.stringify(sendAlarm)).not.toContain('/path?token=abc'); + + const parsed = JSON.parse(mockAppendLine.mock.calls[0][1]); + expect(parsed.failure_class).toBe('quota_throttle'); + expect(parsed.status_code).toBe(429); + expect(parsed.reason).not.toContain('LTAI1234567890SECRET'); + + vi.unstubAllGlobals(); + }); }); describe('shutdown (T016)', () => { diff --git a/tests/unit/flushers/sls-transport.test.ts b/tests/unit/flushers/sls-transport.test.ts index bbb261b8..6b916ad5 100644 --- a/tests/unit/flushers/sls-transport.test.ts +++ b/tests/unit/flushers/sls-transport.test.ts @@ -1,5 +1,11 @@ import { describe, it, expect } from 'vitest'; -import { splitForWebtracking, isRetryable, HttpError } from '../../../src/flushers/sls-transport.js'; +import { + splitForWebtracking, + isRetryable, + HttpError, + classifySlsFailure, + extractEndpointHost, +} from '../../../src/flushers/sls-transport.js'; describe('splitForWebtracking', () => { it('returns single chunk when under limits', () => { @@ -63,3 +69,72 @@ describe('isRetryable', () => { expect(isRetryable('string error')).toBe(false); }); }); + +describe('classifySlsFailure', () => { + it.each([ + [401, 'auth_failed', false], + [403, 'permission_denied', false], + [404, 'not_found', false], + [408, 'network_timeout', true], + [413, 'payload_too_large', false], + [429, 'quota_throttle', true], + [500, 'server_error', true], + ] as const)('classifies HTTP %s as %s', (status, failureClass, retryable) => { + const diagnostics = classifySlsFailure(new HttpError(status, 'body')); + expect(diagnostics.failure_class).toBe(failureClass); + expect(diagnostics.status_code).toBe(status); + expect(diagnostics.retryable).toBe(retryable); + }); + + it('classifies timeout and refused network errors', () => { + expect(classifySlsFailure(Object.assign(new Error('operation aborted'), { name: 'AbortError' })).failure_class).toBe('network_timeout'); + expect(classifySlsFailure(Object.assign(new Error('connect ETIMEDOUT'), { code: 'ETIMEDOUT' })).failure_class).toBe('network_timeout'); + expect(classifySlsFailure(Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })).failure_class).toBe('network_refused'); + }); + + it('classifies common AK SDK error shape', () => { + const diagnostics = classifySlsFailure({ + statusCode: 403, + code: 'AccessDenied', + name: 'SlsError', + message: 'missing log:PostLogStoreLogs permission', + }); + + expect(diagnostics.failure_class).toBe('permission_denied'); + expect(diagnostics.status_code).toBe(403); + expect(diagnostics.retryable).toBe(false); + expect(diagnostics.reason).toContain('AccessDenied'); + }); + + it('sanitizes and limits failure reason', () => { + const diagnostics = classifySlsFailure(new HttpError( + 403, + `AccessKeyId=LTAI1234567890SECRET "accessKeySecret":"plainsecret" Authorization: Bearer tokenvalue url=https://user:pass@example.com/path?token=abc ${'x'.repeat(400)}`, + )); + + expect(diagnostics.reason.length).toBeLessThanOrEqual(240); + expect(diagnostics.reason).not.toContain('LTAI1234567890SECRET'); + expect(diagnostics.reason).not.toContain('plainsecret'); + expect(diagnostics.reason).not.toContain('tokenvalue'); + expect(diagnostics.reason).not.toContain('/path?token=abc'); + expect(diagnostics.reason).toContain('[url:example.com]'); + }); + + it('falls back to unknown for unrecognized errors', () => { + const diagnostics = classifySlsFailure(new Error('some random error')); + expect(diagnostics.failure_class).toBe('unknown'); + expect(diagnostics.status_code).toBeUndefined(); + expect(diagnostics.retryable).toBe(false); + }); +}); + +describe('extractEndpointHost', () => { + it('extracts only host from URL endpoint', () => { + expect(extractEndpointHost('https://user:pass@cn-hangzhou.log.aliyuncs.com/path?q=1')).toBe('cn-hangzhou.log.aliyuncs.com'); + }); + + it('supports host-only endpoints and returns empty for invalid input', () => { + expect(extractEndpointHost('cn-hangzhou.log.aliyuncs.com')).toBe('cn-hangzhou.log.aliyuncs.com'); + expect(extractEndpointHost('')).toBe(''); + }); +}); diff --git a/tests/unit/metrics/alarm-manager.test.ts b/tests/unit/metrics/alarm-manager.test.ts index 2a9c3a9a..4ad735e4 100644 --- a/tests/unit/metrics/alarm-manager.test.ts +++ b/tests/unit/metrics/alarm-manager.test.ts @@ -49,6 +49,48 @@ describe('AlarmManager', () => { expect(entries[1].endpoint_name).toBe('ep2'); }); + it('splits SLS send alarms by failure class and status code', () => { + manager.record('FLUSH_SEND_ALARM', '2', 'forbidden', { + endpoint_name: 'ep1', + failure_class: 'permission_denied', + status_code: 403, + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + mode: 'ak', + project: 'project-a', + logstore: 'store-a', + retryable: false, + reason: 'HTTP 403', + }); + manager.record('FLUSH_SEND_ALARM', '2', 'server error', { + endpoint_name: 'ep1', + failure_class: 'server_error', + status_code: 500, + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + mode: 'ak', + project: 'project-a', + logstore: 'store-a', + retryable: true, + reason: 'HTTP 500', + }); + + const entries = manager.serialize(); + expect(entries).toHaveLength(2); + expect(entries.map(e => e.failure_class).sort()).toEqual(['permission_denied', 'server_error']); + expect(entries.map(e => e.status_code).sort()).toEqual(['403', '500']); + expect(entries[0].endpoint_host).toBe('cn-hangzhou.log.aliyuncs.com'); + expect(entries[0].retryable).toMatch(/^(true|false)$/); + }); + + it('keeps legacy aggregation when diagnostic fields are absent', () => { + manager.record('FLUSH_SEND_ALARM', '2', 'fail 1', { endpoint_name: 'ep1' }); + manager.record('FLUSH_SEND_ALARM', '2', 'fail 2', { endpoint_name: 'ep1' }); + + const entries = manager.serialize(); + expect(entries).toHaveLength(1); + expect(entries[0].alarm_count).toBe('2'); + expect(entries[0].alarm_message).toBe('fail 2'); + }); + it('clears alarms after serialize', () => { manager.record('HOOK_INSTALL_ALARM', '2', 'install failed'); manager.serialize(); diff --git a/tests/unit/metrics/metrics-collector.test.ts b/tests/unit/metrics/metrics-collector.test.ts index ed12f7f1..53eb64cb 100644 --- a/tests/unit/metrics/metrics-collector.test.ts +++ b/tests/unit/metrics/metrics-collector.test.ts @@ -189,9 +189,35 @@ describe('MetricsCollector', () => { expect(ep.out_failed_entries_total).toBe('5'); expect(ep.total_delay_ms).toBe('3500'); expect(ep.last_flush_time).toBe('2026-05-19 10:05:00'); + expect(ep.last_failure_class).toBe(''); + expect(ep.last_failure_status_code).toBe(''); + expect(ep.last_failure_time).toBe(''); + expect(ep.consecutive_failures).toBe('0'); expect(ep.start_time).toBe('2026-05-19 09:00:00'); }); + it('includes additive SLS failure diagnostics in flusher metrics', () => { + const flushers = new Map(); + flushers.set('internal-default', { + inEntries: 200, inBytes: 50000, outEntries: 195, outFailed: 5, + totalDelayMs: 3500, lastFlushTime: '2026-05-19 10:05:00', + startTime: '2026-05-19 09:00:00', flusherName: 'sls', mode: 'webtracking', + endpoint: 'https://cn-heyuan.log.aliyuncs.com', project: 'my-project', logstore: 'my-logstore', + lastFailureClass: 'quota_throttle', + lastFailureStatusCode: '429', + lastFailureTime: '2026-05-19 10:06:00', + consecutiveFailures: 3, + }); + + const result = collector.collectL2Flushers(buildSnapshot({ flushers })); + + expect(result).toHaveLength(1); + expect(result[0].last_failure_class).toBe('quota_throttle'); + expect(result[0].last_failure_status_code).toBe('429'); + expect(result[0].last_failure_time).toBe('2026-05-19 10:06:00'); + expect(result[0].consecutive_failures).toBe('3'); + }); + it('returns empty array when no flushers', () => { const result = collector.collectL2Flushers(buildSnapshot()); expect(result).toEqual([]); diff --git a/tests/unit/metrics/metrics-writer.test.ts b/tests/unit/metrics/metrics-writer.test.ts index 32b2c27d..cb5e24cf 100644 --- a/tests/unit/metrics/metrics-writer.test.ts +++ b/tests/unit/metrics/metrics-writer.test.ts @@ -104,6 +104,54 @@ describe('MetricsWriter', () => { expect(call![1]).not.toHaveProperty('__topic__'); }); + it('writes and sends additive SLS alarm diagnostics', async () => { + fs.mkdirSync(path.join(tmpDir, 'logs', 'metric_alarm'), { recursive: true }); + const alarmManager = new AlarmManager({ ip: '127.0.0.1', version: '2.0.0', userId: 'test-user' }); + alarmManager.record('FLUSH_SEND_ALARM', '2', 'SLS webtracking send failed: HTTP 403', { + endpoint_name: 'internal-sls', + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + mode: 'webtracking', + project: 'project-a', + logstore: 'store-a', + failure_class: 'permission_denied', + status_code: 403, + retryable: false, + reason: 'HTTP 403 forbidden', + }); + writer = new MetricsWriter({ + dataDir: tmpDir, + version: '2.0.0', + userId: 'u1', + getSnapshot: buildSnapshot, + alarmManager, + }); + + await (writer as any).writeAlarms(); + + const alarmPath = path.join(tmpDir, 'logs', 'metric_alarm', 'pilot-alarms.jsonl'); + const entry = JSON.parse(fs.readFileSync(alarmPath, 'utf-8').trim()); + expect(entry).toMatchObject({ + alarm_type: 'FLUSH_SEND_ALARM', + endpoint_name: 'internal-sls', + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + mode: 'webtracking', + project: 'project-a', + logstore: 'store-a', + failure_class: 'permission_denied', + status_code: '403', + retryable: 'false', + reason: 'HTTP 403 forbidden', + }); + + expect(mockSendAlarm).toHaveBeenCalledWith('pilot_alarm', expect.objectContaining({ + failure_class: 'permission_denied', + status_code: '403', + retryable: 'false', + endpoint_host: 'cn-hangzhou.log.aliyuncs.com', + })); + expect(JSON.stringify(mockSendAlarm.mock.calls[0][1])).not.toContain('https://cn-hangzhou.log.aliyuncs.com'); + }); + it('writes L2 input/flusher metrics on stop (final flush)', async () => { writer = new MetricsWriter({ dataDir: tmpDir, From 544c50b13161282e7c58b9c0c7247394a71f3fd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=B3=E6=9C=A8?= Date: Fri, 10 Jul 2026 14:04:07 +0800 Subject: [PATCH 2/2] fix(sls): classify nested fetch network failures Extract status and errno details from nested fetch causes so refused and timeout transport errors use network failure classes with retryable diagnostics. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/flushers/sls-transport.ts | 123 ++++++++++++++++++++-- tests/unit/flushers/sls-transport.test.ts | 50 +++++++++ 2 files changed, 165 insertions(+), 8 deletions(-) diff --git a/src/flushers/sls-transport.ts b/src/flushers/sls-transport.ts index 841e6ab2..fdd69669 100644 --- a/src/flushers/sls-transport.ts +++ b/src/flushers/sls-transport.ts @@ -1,6 +1,7 @@ import { createLogger } from '../utils/logger.js'; import { appendLine, ensureDir } from '../utils/fs-utils.js'; import * as path from 'node:path'; +import { constants as osConstants } from 'node:os'; const logger = createLogger('SlsTransport'); @@ -12,6 +13,41 @@ export const RETRY_BASE_DELAY_MS = 1000; export const RETRYABLE_STATUS_CODES = new Set([408, 429, 500, 502, 503, 504]); const SLS_FAILURE_REASON_MAX_LENGTH = 240; +const ERROR_SUMMARY_KEYS = [ + 'name', + 'code', + 'errno', + 'syscall', + 'address', + 'port', + 'statusCode', + 'status', + 'httpStatusCode', + 'message', +] as const; +const MAX_ERROR_CAUSE_DEPTH = 6; +const NETWORK_TIMEOUT_ERRNOS = new Set([ + osConstants.errno.ETIMEDOUT, + -osConstants.errno.ETIMEDOUT, + 60, + -60, + 110, + -110, +].map(String)); +const NETWORK_REFUSED_ERRNOS = new Set([ + osConstants.errno.ECONNREFUSED, + -osConstants.errno.ECONNREFUSED, + osConstants.errno.ECONNRESET, + -osConstants.errno.ECONNRESET, + 54, + -54, + 61, + -61, + 104, + -104, + 111, + -111, +].map(String)); export type SlsFailureClass = | 'network_timeout' @@ -135,16 +171,33 @@ export function sanitizeFailureReason(reason: string): string { return `${redacted.slice(0, SLS_FAILURE_REASON_MAX_LENGTH - 3)}...`; } -function extractStatusCode(err: unknown): number | undefined { +function extractStatusCode( + err: unknown, + seen: Set = new Set(), +): number | undefined { if (err instanceof HttpError) return err.status; if (!err || typeof err !== 'object') return undefined; + if (seen.has(err)) return undefined; + seen.add(err); + const record = err as Record; for (const key of ['statusCode', 'status', 'httpStatusCode']) { const raw = record[key]; if (typeof raw === 'number') return raw; if (typeof raw === 'string' && /^\d+$/.test(raw)) return Number(raw); } + + const causeStatus = extractStatusCode(record.cause, seen); + if (causeStatus !== undefined) return causeStatus; + + if (Array.isArray(record.errors)) { + for (const nested of record.errors) { + const nestedStatus = extractStatusCode(nested, seen); + if (nestedStatus !== undefined) return nestedStatus; + } + } + return undefined; } @@ -154,16 +207,50 @@ function extractFailureText(err: unknown): string { } if (!err || typeof err !== 'object') return String(err); + const parts = collectErrorSummaryParts(err); + if (parts.length > 0) return parts.join(' '); + return String(err); +} + +function collectErrorSummaryParts( + err: unknown, + seen: Set = new Set(), + depth = 0, +): string[] { + if (!err || typeof err !== 'object' || depth > MAX_ERROR_CAUSE_DEPTH) { + return []; + } + + if (seen.has(err)) return []; + seen.add(err); + const record = err as Record; const parts: string[] = []; - for (const key of ['name', 'code', 'statusCode', 'status', 'message']) { + const prefix = depth === 0 ? '' : `${'cause.'.repeat(depth)}`; + + for (const key of ERROR_SUMMARY_KEYS) { const value = record[key]; if (value !== undefined && value !== null && value !== '') { - parts.push(`${key}=${String(value)}`); + parts.push(`${prefix}${key}=${String(value)}`); } } - if (parts.length > 0) return parts.join(' '); - return String(err); + + const cause = record.cause; + if (cause !== undefined && cause !== null) { + if (typeof cause === 'object') { + parts.push(...collectErrorSummaryParts(cause, seen, depth + 1)); + } else { + parts.push(`${prefix}cause=${String(cause)}`); + } + } + + if (Array.isArray(record.errors)) { + for (const nested of record.errors.slice(0, 3)) { + parts.push(...collectErrorSummaryParts(nested, seen, depth + 1)); + } + } + + return parts; } function classifyByStatus(statusCode?: number): SlsFailureClass | undefined { @@ -184,13 +271,15 @@ function classifyByText(lower: string): SlsFailureClass { lower.includes('aborterror') || lower.includes('timeouterror') || lower.includes('etimedout') || - lower.includes('timeout') + lower.includes('timeout') || + textHasErrno(lower, NETWORK_TIMEOUT_ERRNOS) ) return 'network_timeout'; if ( lower.includes('econnrefused') || lower.includes('econnreset') || lower.includes('socket hang up') || - lower.includes('network') + lower.includes('network') || + textHasErrno(lower, NETWORK_REFUSED_ERRNOS) ) return 'network_refused'; if ( lower.includes('throttl') || @@ -227,11 +316,15 @@ function classifyByText(lower: string): SlsFailureClass { } function isRetryableFailure( - _failureClass: SlsFailureClass, + failureClass: SlsFailureClass, statusCode: number | undefined, lower: string, ): boolean { if (statusCode !== undefined) return RETRYABLE_STATUS_CODES.has(statusCode); + if ( + failureClass === 'network_timeout' || + failureClass === 'network_refused' + ) return true; return ( lower.includes('econnreset') || lower.includes('etimedout') || @@ -239,11 +332,25 @@ function isRetryableFailure( lower.includes('socket hang up') || lower.includes('network') || lower.includes('timeouterror') || + lower.includes('aborterror') || lower.includes('internalservererror') || lower.includes('serverbusy') ); } +function textHasErrno(text: string, errnoValues: Set): boolean { + for (const errno of errnoValues) { + if (new RegExp(`\\berrno[:=]\\s*${escapeRegExp(errno)}\\b`).test(text)) { + return true; + } + } + return false; +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); +} + export async function postWebtracking( config: SlsTransportConfig, logs: Record[], diff --git a/tests/unit/flushers/sls-transport.test.ts b/tests/unit/flushers/sls-transport.test.ts index 6b916ad5..fb1c9d8d 100644 --- a/tests/unit/flushers/sls-transport.test.ts +++ b/tests/unit/flushers/sls-transport.test.ts @@ -1,4 +1,5 @@ import { describe, it, expect } from 'vitest'; +import { constants as osConstants } from 'node:os'; import { splitForWebtracking, isRetryable, @@ -62,6 +63,7 @@ describe('isRetryable', () => { expect(isRetryable(new Error('ECONNREFUSED'))).toBe(true); expect(isRetryable(new Error('socket hang up'))).toBe(true); expect(isRetryable(new Error('TimeoutError'))).toBe(true); + expect(isRetryable(Object.assign(new Error('operation aborted'), { name: 'AbortError' }))).toBe(true); }); it('returns false for unknown errors', () => { @@ -92,6 +94,54 @@ describe('classifySlsFailure', () => { expect(classifySlsFailure(Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })).failure_class).toBe('network_refused'); }); + it('classifies Node fetch TypeError by nested cause code and errno', () => { + const cause = Object.assign(new Error('connect ECONNREFUSED 127.0.0.1:18082'), { + code: 'ECONNREFUSED', + errno: -osConstants.errno.ECONNREFUSED, + syscall: 'connect', + address: '127.0.0.1', + port: 18082, + }); + const err = new TypeError('fetch failed'); + Object.assign(err, { cause }); + + const diagnostics = classifySlsFailure(err); + expect(diagnostics.failure_class).toBe('network_refused'); + expect(diagnostics.status_code).toBeUndefined(); + expect(diagnostics.retryable).toBe(true); + expect(diagnostics.reason).toContain('name=TypeError'); + expect(diagnostics.reason).toContain('message=fetch failed'); + expect(diagnostics.reason).toContain('cause.code=ECONNREFUSED'); + expect(diagnostics.reason).toContain(`cause.errno=${-osConstants.errno.ECONNREFUSED}`); + expect(diagnostics.reason).toContain('cause.message=connect ECONNREFUSED 127.0.0.1:18082'); + }); + + it('classifies nested cause timeout and reset errors', () => { + const timeoutCause = Object.assign(new Error('connect timeout'), { + errno: -osConstants.errno.ETIMEDOUT, + }); + const timeoutWrapper = Object.assign(new TypeError('fetch failed'), { + cause: Object.assign(new Error('wrapped transport error'), { + cause: timeoutCause, + }), + }); + expect(classifySlsFailure(timeoutWrapper)).toMatchObject({ + failure_class: 'network_timeout', + retryable: true, + }); + + const resetErr = Object.assign(new TypeError('fetch failed'), { + cause: Object.assign(new Error('socket closed'), { + code: 'ECONNRESET', + errno: -osConstants.errno.ECONNRESET, + }), + }); + expect(classifySlsFailure(resetErr)).toMatchObject({ + failure_class: 'network_refused', + retryable: true, + }); + }); + it('classifies common AK SDK error shape', () => { const diagnostics = classifySlsFailure({ statusCode: 403,