From 7972f16cc217ac920e6e7c9b6c02b915739028a3 Mon Sep 17 00:00:00 2001 From: Snssn <1502062504@qq.com> Date: Thu, 13 Aug 2026 15:46:45 +0800 Subject: [PATCH 1/2] fix(sls-flusher): propagate send failures and classify by actionability Retry-exhausted sends previously recorded an alarm, persisted, then returned normally, so flush() credited failed batches as succeeded and outFailed stayed 0 (send-failure was invisible in L1/L2 metrics). All three send paths (ak/apiKey/webtracking) now rethrow, and flush() splits succeeded vs failed counts, including partial webtracking chunks. Failures are classified (transient/quota/config/payload) which drives alarm_level and recovery: - transient no longer alarms per-occurrence (that volume already lives in the out_failed metric); only a sustained per-endpoint outage escalates to a single cooldown-gated alarm. - config (project not-exist/forbidden/recycled) is cooldown-gated and trips a per-endpoint circuit breaker with exponential backoff (<=10min) and a half-open probe that auto-recovers. - oversize entries are truncated to fit the body cap, or dropped and counted, instead of being sent as a request that is guaranteed to 413. Adds an optional failure_class field to the alarm schema; alarm_type stays FLUSH_SEND_ALARM so downstream filters are unaffected. --- docs/zh-CN/sls-output.md | 13 + src/flushers/sls-flusher.ts | 296 ++++++++++++++--- src/flushers/sls-transport.ts | 63 ++++ src/metrics/alarm-manager.ts | 5 +- .../sls-flusher.failure-handling.test.ts | 310 ++++++++++++++++++ tests/unit/flushers/sls-transport.test.ts | 43 ++- tests/unit/metrics/alarm-manager.test.ts | 19 ++ 7 files changed, 701 insertions(+), 48 deletions(-) create mode 100644 tests/unit/flushers/sls-flusher.failure-handling.test.ts diff --git a/docs/zh-CN/sls-output.md b/docs/zh-CN/sls-output.md index 2e25e8261..4d7a04a25 100644 --- a/docs/zh-CN/sls-output.md +++ b/docs/zh-CN/sls-output.md @@ -153,6 +153,19 @@ ls ~/.loongsuite-pilot/logs/sls-failed-logs/ 这些 JSONL 记录只包含 endpoint、错误摘要、batch 条数和 batch 字节数估算，不包含失败 batch payload、消息正文、请求 headers 或凭证，因此不能用于重放失败数据。日志按本地日期和单文件 10MiB 轮转，目录总量限制为 50MiB，同时遵循 `retention.slsFailedDays`（默认 7 天）。 +## 发送失败的分类、冷却与熔断 + +重试耗尽后的发送失败会按可操作性分类，并驱动告警级别与恢复行为： + +| 分类 | 判据 | 告警行为 | +|------|------|----------| +| `transient` | 超时 / 网络失败 / 5xx | 默认不逐条告警，只计入失败指标；同一 endpoint 连续多个周期全批失败才升级一条告警 | +| `quota` | 429 限流 | 逐条上报（聚合） | +| `config` | 404 / 403 / project 不存在、被禁用、在回收站 | 上报受冷却窗口抑制（每小时一条），并对该 endpoint 触发熔断退避 | +| `payload` | 413 / 单条超过单请求体积上限 | 上报（超限条目先尝试截断最大字段，仍超限则丢弃） | + +熔断说明：当某 endpoint 连续出现 `config` 类终态失败时，Pilot 会停止对它高频重试，按指数退避（上界 10 分钟）降低尝试频率；退避到达后放行一次探测，成功即自动恢复。熔断为内存态，不跨重启保留。无论是否告警，失败条数都会计入 flusher 指标的 `out_failed_entries_total`。 + 调试 SLS 前，可以先通过本地 JSONL 确认采集本身是否正常： ```bash diff --git a/src/flushers/sls-flusher.ts b/src/flushers/sls-flusher.ts index 7fdac39f6..f8600572b 100644 --- a/src/flushers/sls-flusher.ts +++ b/src/flushers/sls-flusher.ts @@ -19,6 +19,9 @@ import { postWebtracking, postApiKeyLogStoreLogs, isRetryable, + classifyFailure, + FAILURE_CLASS_ALARM_LEVEL, + type FailureClass, RETRY_MAX_ATTEMPTS, RETRY_BASE_DELAY_MS, WEBTRACKING_TIMEOUT_MS, @@ -37,6 +40,24 @@ const DEFAULT_HEADERS_TIMEOUT_MS = 30_000; const DEFAULT_BODY_TIMEOUT_MS = 15_000; const DEFAULT_OVERALL_TIMEOUT_MS = 30_000; +// transient (network) failures don't alarm per-occurrence — the out_failed metric +// already carries the volume. Only a sustained outage (this many consecutive +// all-batch-failed flush cycles on one endpoint) escalates to a single alarm. +const TRANSIENT_ESCALATE_THRESHOLD = 3; + +// A terminal (config) failure trips the breaker after this many consecutive +// occurrences, then backs off exponentially to stop pointless retries. +const CIRCUIT_FAILURE_THRESHOLD = 3; +const CIRCUIT_BASE_BACKOFF_MS = 2_000; +const CIRCUIT_MAX_BACKOFF_MS = 600_000; // 10 min upper bound + +// config + escalated-transient alarms are cooldown-gated so a persistent fault +// reports once per window instead of every cycle. Re-arms after the window (not a once-guard). +const FLUSH_ALARM_COOLDOWN_MS = 3_600_000; + +// Marker appended to a field truncated to fit the single-request body cap. +const TRUNCATION_MARKER = '...[TRUNCATED]'; + interface QueuedLog { content: Record; endpoint: SlsEndpoint; @@ -44,6 +65,29 @@ interface QueuedLog { byteSize: number; } +interface CircuitState { + configFails: number; + openUntil: number; + backoffMs: number; +} + +/** + * Raised by a send path when a batch (or part of it) finally fails. Carries the + * split so flush() can credit succeeded entries and debit failed ones accurately, + * instead of charging the whole batch to one side. + */ +export class FlushFailure extends Error { + constructor( + readonly succeededEntries: number, + readonly failedEntries: number, + readonly failureClass: FailureClass, + readonly cause?: unknown, + ) { + super(`flush failed: ${failedEntries} entries (${failureClass})`); + this.name = 'FlushFailure'; + } +} + const logger = createLogger('SlsFlusher'); export interface EndpointCounter { @@ -79,6 +123,12 @@ export class SlsFlusher extends BaseFlusher { private readonly endpointCounters: Map = new Map(); private alarmManager: AlarmManager | null = null; + // Per-endpoint failure state (keyed by endpoint.name). + private readonly transientFailStreak: Map = new Map(); + private readonly circuits: Map = new Map(); + // Cooldown gate for config + escalated-transient alarms, keyed `${endpoint}_${class}`. + private readonly lastAlarmAt: Map = new Map(); + private readonly serviceName: string; private readonly serviceNamePrefix: string; private readonly userAgent: string; @@ -194,6 +244,15 @@ export class SlsFlusher extends BaseFlusher { .map(([, logs]) => () => { const endpoint = logs[0].endpoint; const counter = this.endpointCounters.get(endpoint.name); + + // Circuit open (terminal endpoint, still within backoff): skip the send + // entirely. Count the drop but do NOT re-send, re-persist, or re-alarm — + // that is exactly the pointless-request/write loop we are stopping. + if (this.isCircuitOpen(endpoint.name, Date.now())) { + if (counter) counter.outFailed += logs.length; + return Promise.resolve(); + } + const startMs = Date.now(); const send = this.flushEndpoint(endpoint, logs); return send.then(() => { @@ -202,15 +261,24 @@ export class SlsFlusher extends BaseFlusher { counter.totalDelayMs += Date.now() - startMs; counter.lastFlushTime = formatTime(new Date()); } + this.onEndpointSuccess(endpoint.name); }).catch(err => { + const succeeded = err instanceof FlushFailure ? err.succeededEntries : 0; + const failed = err instanceof FlushFailure ? err.failedEntries : logs.length; + const failureClass: FailureClass = + err instanceof FlushFailure ? err.failureClass : classifyFailure(err); if (counter) { - counter.outFailed += logs.length; + counter.outEntries += succeeded; + counter.outFailed += failed; counter.totalDelayMs += Date.now() - startMs; } logger.error('SLS endpoint flush failed', { endpoint: endpoint.name, - error: String(err), + failureClass, + failed, + succeeded, }); + this.onEndpointFailure(endpoint, succeeded, failureClass, err instanceof FlushFailure ? err.cause : err); }); }); @@ -238,6 +306,91 @@ export class SlsFlusher extends BaseFlusher { await Promise.all(workers); } + // --- per-endpoint failure handling ------------------------------------- + + private isCircuitOpen(name: string, now: number): boolean { + const c = this.circuits.get(name); + return !!c && now < c.openUntil; + } + + private onEndpointSuccess(name: string): void { + // Any success (including a half-open probe) clears streak + breaker. + this.transientFailStreak.delete(name); + this.circuits.delete(name); + } + + private onEndpointFailure( + endpoint: SlsEndpoint, + succeeded: number, + failureClass: FailureClass, + cause: unknown, + ): void { + if (failureClass === 'transient') { + // Only a whole-batch failure counts toward "sustained outage". Any partial + // success means the endpoint is reachable → reset and stay silent. + if (succeeded > 0) { + this.transientFailStreak.delete(endpoint.name); + return; + } + const streak = (this.transientFailStreak.get(endpoint.name) ?? 0) + 1; + this.transientFailStreak.set(endpoint.name, streak); + if (streak >= TRANSIENT_ESCALATE_THRESHOLD) { + this.recordFailureAlarm( + endpoint, 'transient', + `SLS send failing continuously (${streak} cycles): ${String(cause)}`, + true, + ); + } + return; + } + + // Non-transient failure this cycle: not a clean transient outage → reset streak. + this.transientFailStreak.delete(endpoint.name); + + if (failureClass === 'config') { + this.recordFailureAlarm(endpoint, 'config', `SLS terminal config error: ${String(cause)}`, true); + this.tripCircuit(endpoint.name); + return; + } + // quota / payload: report per-occurrence (aggregated by AlarmManager, no cooldown). + this.recordFailureAlarm(endpoint, failureClass, `SLS ${failureClass} failure: ${String(cause)}`, false); + } + + private recordFailureAlarm( + endpoint: SlsEndpoint, + failureClass: FailureClass, + message: string, + cooldown: boolean, + ): void { + if (!this.alarmManager) return; + if (cooldown) { + const key = `${endpoint.name}_${failureClass}`; + const now = Date.now(); + const last = this.lastAlarmAt.get(key) ?? 0; + if (now - last < FLUSH_ALARM_COOLDOWN_MS) return; + this.lastAlarmAt.set(key, now); + } + this.alarmManager.record( + 'FLUSH_SEND_ALARM', + FAILURE_CLASS_ALARM_LEVEL[failureClass], + message, + { endpoint_name: endpoint.name, failure_class: failureClass }, + ); + } + + private tripCircuit(name: string): void { + const c = this.circuits.get(name) ?? { configFails: 0, openUntil: 0, backoffMs: CIRCUIT_BASE_BACKOFF_MS }; + c.configFails++; + if (c.configFails >= CIRCUIT_FAILURE_THRESHOLD) { + // First trip uses base backoff; each subsequent probe failure doubles it (capped). + c.backoffMs = c.openUntil === 0 + ? CIRCUIT_BASE_BACKOFF_MS + : Math.min(c.backoffMs * 2, CIRCUIT_MAX_BACKOFF_MS); + c.openUntil = Date.now() + c.backoffMs; + } + this.circuits.set(name, c); + } + /** Exact global name wins; otherwise managed endpoints may override the shared prefix. */ private effectiveServiceName(endpoint?: SlsEndpoint): string { return this.serviceName || endpoint?.serviceName || this.serviceNamePrefix; @@ -325,30 +478,44 @@ export class SlsFlusher extends BaseFlusher { endpoint: endpoint.name, error: String(lastErr), }); - this.alarmManager?.record( - 'FLUSH_SEND_ALARM', '2', - `SLS ak send failed: ${String(lastErr)}`, - { endpoint_name: endpoint.name }, - ); - if (lastErr instanceof HttpError && lastErr.status === 429) { - this.alarmManager?.record( - 'FLUSH_QUOTA_ALARM', '2', - `SLS endpoint throttled (429)`, - { endpoint_name: endpoint.name }, - ); - } await this.persistFailedLogs( endpoint, logs.length, logs.reduce((sum, log) => sum + log.byteSize, 0), lastErr, ); + throw lastErr; } private async flushViaWebtracking(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { - const chunks = this.splitForWebtracking(logs); + const { chunks, dropped } = this.splitForWebtracking(logs); + let succeeded = 0; + let failed = 0; + let sendErr: unknown; + + // Oversize entries the splitter could not fit even after truncation: count as + // failed and persist a payload record. They never reach the wire. + if (dropped > 0) { + failed += dropped; + await this.persistFailedLogs( + endpoint, dropped, 0, + new Error(`payload dropped: ${dropped} entr${dropped === 1 ? 'y' : 'ies'} exceed WEBTRACKING_MAX_BODY_BYTES`), + ); + } + for (const chunk of chunks) { - await this.postWebtracking(endpoint, chunk); + try { + await this.postWebtracking(endpoint, chunk); + succeeded += chunk.length; + } catch (err) { + failed += chunk.length; + sendErr = err; + } + } + + if (failed > 0) { + const failureClass: FailureClass = sendErr ? classifyFailure(sendErr) : 'payload'; + throw new FlushFailure(succeeded, failed, failureClass, sendErr ?? new Error('payload oversize')); } } @@ -406,37 +573,46 @@ export class SlsFlusher extends BaseFlusher { endpoint: endpoint.name, error: String(lastErr), }); - this.alarmManager?.record( - 'FLUSH_SEND_ALARM', '2', - `SLS apiKey send failed: ${String(lastErr)}`, - { endpoint_name: endpoint.name }, - ); - if (lastErr instanceof HttpError && lastErr.status === 429) { - this.alarmManager?.record( - 'FLUSH_QUOTA_ALARM', '2', - `SLS endpoint throttled (429)`, - { endpoint_name: endpoint.name }, - ); - } await this.persistFailedLogs( endpoint, logs.length, logs.reduce((sum, log) => sum + log.byteSize, 0), lastErr, ); + throw lastErr; } - private splitForWebtracking(logs: QueuedLog[]): QueuedLog[][] { + private splitForWebtracking(logs: QueuedLog[]): { chunks: QueuedLog[][]; dropped: number } { + const maxBytes = WEBTRACKING_MAX_BODY_BYTES; const chunks: QueuedLog[][] = []; let current: QueuedLog[] = []; let currentSize = 0; - - for (const log of logs) { - const logSize = Buffer.byteLength(JSON.stringify(log.content)); + let dropped = 0; + + for (const raw of logs) { + let log = raw; + let logSize = Buffer.byteLength(JSON.stringify(log.content)); + + // A single entry over the cap can never fit any chunk. Try trimming its + // largest field; if it still won't fit, drop it rather than emit a request + // that is guaranteed to be rejected. + if (logSize > maxBytes) { + const trimmed = this.truncateOversizeEntry(log, maxBytes); + if (!trimmed) { + dropped++; + continue; + } + log = trimmed; + logSize = Buffer.byteLength(JSON.stringify(log.content)); + if (logSize > maxBytes) { + dropped++; + continue; + } + } if (current.length > 0 && (current.length >= WEBTRACKING_MAX_LOGS || - currentSize + logSize > WEBTRACKING_MAX_BODY_BYTES)) { + currentSize + logSize > maxBytes)) { chunks.push(current); current = []; currentSize = 0; @@ -449,7 +625,39 @@ export class SlsFlusher extends BaseFlusher { if (current.length > 0) { chunks.push(current); } - return chunks; + return { chunks, dropped }; + } + + /** + * Shrink an oversize entry by truncating its largest string field so the whole + * content serializes under maxBytes, preserving JSON validity, UTF-8 boundaries, + * and leaving a marker. Returns null when trimming that one field can't get it + * under the cap (caller then drops the entry). + */ + private truncateOversizeEntry(log: QueuedLog, maxBytes: number): QueuedLog | null { + const content: Record = { ...log.content }; + let largestKey = ''; + let largestBytes = 0; + for (const [k, v] of Object.entries(content)) { + const len = Buffer.byteLength(v); + if (len > largestBytes) { + largestBytes = len; + largestKey = k; + } + } + if (!largestKey) return null; + + const overshoot = Buffer.byteLength(JSON.stringify(content)) - maxBytes; + if (overshoot <= 0) return log; + + const markerBytes = Buffer.byteLength(TRUNCATION_MARKER); + // Headroom absorbs JSON escaping of the retained slice. + const targetBytes = largestBytes - overshoot - markerBytes - 256; + if (targetBytes <= 0) return null; + + content[largestKey] = truncateUtf8Bytes(content[largestKey], targetBytes) + TRUNCATION_MARKER; + const byteSize = Buffer.byteLength(JSON.stringify(content)); + return { ...log, content, byteSize }; } private async postWebtracking(endpoint: SlsEndpoint, logs: QueuedLog[]): Promise { @@ -522,19 +730,8 @@ export class SlsFlusher extends BaseFlusher { endpoint: endpoint.name, error: String(lastErr), }); - this.alarmManager?.record( - 'FLUSH_SEND_ALARM', '2', - `SLS webtracking send failed: ${String(lastErr)}`, - { endpoint_name: endpoint.name }, - ); - if (lastErr instanceof HttpError && lastErr.status === 429) { - this.alarmManager?.record( - 'FLUSH_QUOTA_ALARM', '2', - `SLS endpoint throttled (429)`, - { endpoint_name: endpoint.name }, - ); - } await this.persistFailedLogs(endpoint, logs.length, Buffer.byteLength(raw), lastErr); + throw lastErr; } private async persistFailedLogs( @@ -656,3 +853,10 @@ export class SlsFlusher extends BaseFlusher { return new Promise(resolve => setTimeout(resolve, ms)); } } + +/** UTF-8-safe truncation to at most maxBytes, without splitting a multibyte char. */ +function truncateUtf8Bytes(value: string, maxBytes: number): string { + const bytes = Buffer.from(value, 'utf8'); + if (bytes.length <= maxBytes) return value; + return bytes.subarray(0, maxBytes).toString('utf8').replace(/\uFFFD$/u, ''); +} diff --git a/src/flushers/sls-transport.ts b/src/flushers/sls-transport.ts index 5d57b9f4c..49d9ba469 100644 --- a/src/flushers/sls-transport.ts +++ b/src/flushers/sls-transport.ts @@ -24,6 +24,69 @@ export class HttpError extends Error { } } +/** + * Actionability classes for a final send failure. Drives alarm level and + * whether the failure is terminal (needs human config change) vs self-healing. + */ +export type FailureClass = 'transient' | 'quota' | 'config' | 'payload'; + +/** alarm_level per failure class: terminal/payload are most actionable. */ +export const FAILURE_CLASS_ALARM_LEVEL: Record = { + transient: '3', + quota: '2', + config: '1', + payload: '1', +}; + +// Terminal SLS API error codes: the target project/logstore is gone, forbidden, +// or recycled — retrying never succeeds until an operator fixes configuration. +const CONFIG_ERROR_CODES = [ + 'ProjectNotExist', + 'ProjectForbidden', + 'ProjectInRecycleBin', + 'LogStoreNotExist', +]; + +function extractStatus(err: unknown): number | undefined { + if (err instanceof HttpError) return err.status; + if (err && typeof err === 'object' && 'status' in err) { + const s = (err as { status: unknown }).status; + if (typeof s === 'number') return s; + } + return undefined; +} + +/** + * Classify a final send failure by actionability. Works across all send modes: + * webtracking throws HttpError (has .status), while the @alicloud/log SDK (ak + * mode) throws objects whose code lives in `.code`/`.errorCode` or the message + * string — so we inspect both, unlike the historical `instanceof HttpError` + * check that silently never matched on the ak path. + */ +export function classifyFailure(err: unknown): FailureClass { + const status = extractStatus(err); + const codeField = + err && typeof err === 'object' + ? String( + (err as { code?: unknown }).code ?? + (err as { errorCode?: unknown }).errorCode ?? + '', + ) + : ''; + const msg = `${codeField} ${String(err)}`; + + if (status === 413 || /PostBodyTooLarge|Request Entity Too Large|body size/i.test(msg)) { + return 'payload'; + } + if (status === 429 || /\bServerBusy\b|Throttl/i.test(msg)) { + return 'quota'; + } + if (status === 404 || status === 403 || CONFIG_ERROR_CODES.some(c => msg.includes(c))) { + return 'config'; + } + return 'transient'; +} + export interface SlsTransportConfig { endpoint: string; project: string; diff --git a/src/metrics/alarm-manager.ts b/src/metrics/alarm-manager.ts index aba46fe87..307664314 100644 --- a/src/metrics/alarm-manager.ts +++ b/src/metrics/alarm-manager.ts @@ -25,6 +25,7 @@ export type AlarmType = export interface AlarmContext { input_name?: string; endpoint_name?: string; + failure_class?: string; } export interface AlarmEntry { @@ -37,6 +38,7 @@ export interface AlarmEntry { ver: string; input_name?: string; endpoint_name?: string; + failure_class?: string; __time__: number; } @@ -61,7 +63,7 @@ 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 ?? ''}`; const existing = this.alarms.get(key); if (existing) { existing.count++; @@ -91,6 +93,7 @@ export class AlarmManager { }; if (item.context?.input_name) entry.input_name = item.context.input_name; if (item.context?.endpoint_name) entry.endpoint_name = item.context.endpoint_name; + if (item.context?.failure_class) entry.failure_class = item.context.failure_class; entries.push(entry); } diff --git a/tests/unit/flushers/sls-flusher.failure-handling.test.ts b/tests/unit/flushers/sls-flusher.failure-handling.test.ts new file mode 100644 index 000000000..f035dbe7d --- /dev/null +++ b/tests/unit/flushers/sls-flusher.failure-handling.test.ts @@ -0,0 +1,310 @@ +/** + * SLS flusher failure handling: failure propagation & counting (D1/D2a), + * transient escalation (D4 方案B), config cooldown (D3/D4), circuit breaker (D3/D5), + * and oversize payload guard (D2b/D6). + */ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { SlsFlusherConfig, SlsEndpoint } from '../../../src/types/index.js'; +import { buildTestEntry } from '../../helpers/fixture-builder.js'; + +const mockPostLogStoreLogs = vi.fn().mockResolvedValue(undefined); +const mockFailureWrite = vi.fn().mockResolvedValue(true); + +vi.mock('@alicloud/log', () => ({ + default: vi.fn().mockImplementation(() => ({ postLogStoreLogs: mockPostLogStoreLogs })), +})); + +const fetchSpy = vi.fn().mockResolvedValue({ ok: true, status: 200, text: async () => '' }); +vi.stubGlobal('fetch', fetchSpy); + +vi.mock('../../../src/utils/fs-utils.js', () => ({ + getTodayDateString: () => '2026-04-27', + readInstalledVersion: () => '0.0.0-test', +})); + +vi.mock('../../../src/flushers/sls-failure-log-writer.js', () => ({ + SlsFailureLogWriter: vi.fn().mockImplementation(() => ({ + start: vi.fn().mockResolvedValue(undefined), + write: mockFailureWrite, + })), +})); + +vi.mock('../../../src/utils/logger.js', () => ({ + createLogger: () => ({ info: vi.fn(), debug: vi.fn(), warn: vi.fn(), error: vi.fn() }), +})); + +import { SlsFlusher } from '../../../src/flushers/sls-flusher.js'; +import { AlarmManager } from '../../../src/metrics/alarm-manager.js'; + +function akEndpoint(name: string, url: string, project: string): SlsEndpoint { + return { + name, endpoint: url, project, logstore: `${project}-store`, + kind: 'agentActivity', mode: 'ak', + accessKeyId: `${name}-ak`, accessKeySecret: `${name}-sk`, redact: false, + }; +} + +function wtEndpoint(name: string, url: string, project: string): SlsEndpoint { + return { + name, endpoint: url, project, logstore: `${project}-store`, + kind: 'agentActivity', mode: 'webtracking', redact: false, + }; +} + +function makeConfig(endpoints: SlsEndpoint[]): SlsFlusherConfig { + const primary = endpoints[0]; + return { + enabled: true, + accessKeyId: primary.accessKeyId ?? '', + accessKeySecret: primary.accessKeySecret ?? '', + apiKey: primary.apiKey ?? '', + endpoint: primary.endpoint, + mode: primary.mode, + endpoints, + batchMaxSize: 20, + flushIntervalMs: 99999, + serviceNamePrefix: '', + }; +} + +// A non-retryable error (so the ak send path breaks immediately, no retry sleeps) +// that classifyFailure maps to `transient` (matches none of payload/quota/config). +function transientErr(msg = 'boom'): Error { + return new Error(msg); +} +// Non-retryable + classified as config (terminal). +function configErr(): Error { + return new Error('{"errorCode":"ProjectNotExist","errorMessage":"gone"}'); +} + +function counters(flusher: SlsFlusher, name: string) { + return flusher.getEndpointCounters().get(name)!; +} + +describe('SlsFlusher failure handling', () => { + beforeEach(() => { + vi.clearAllMocks(); + fetchSpy.mockReset().mockResolvedValue({ ok: true, status: 200, text: async () => '' }); + mockPostLogStoreLogs.mockReset().mockResolvedValue(undefined); + mockFailureWrite.mockReset().mockResolvedValue(true); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + // --- D1 / D2a: failure propagation & counting ------------------------- + + it('counts failed entries in outFailed, not outEntries, on send failure', async () => { + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + mockPostLogStoreLogs.mockRejectedValue(transientErr()); + + await flusher.send(buildTestEntry()); + await flusher.flush(); + + const c = counters(flusher, 'user'); + expect(c.outFailed).toBe(1); + expect(c.outEntries).toBe(0); + }); + + it('counts succeeded entries in outEntries on success', async () => { + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + + await flusher.send(buildTestEntry()); + await flusher.flush(); + + const c = counters(flusher, 'user'); + expect(c.outEntries).toBe(1); + expect(c.outFailed).toBe(0); + }); + + it('splits succeeded vs failed counts across webtracking chunks (partial failure)', async () => { + const flusher = new SlsFlusher(makeConfig([wtEndpoint('internal', 'https://cn-heyuan.log.aliyuncs.com', 'p')]), '/tmp/data'); + + // Two ~1.6MB entries force a 2-chunk split (cap is 2.8MB). One chunk succeeds, one 404s. + const big = 'x'.repeat(1_600_000); + await flusher.send(buildTestEntry({ 'gen_ai.completion': big })); + await flusher.send(buildTestEntry({ 'gen_ai.completion': big })); + + fetchSpy + .mockResolvedValueOnce({ ok: true, status: 200, text: async () => '' }) + .mockResolvedValueOnce({ ok: false, status: 404, text: async () => '{"errorCode":"ProjectNotExist"}' }); + + await flusher.flush(); + + const c = counters(flusher, 'internal'); + expect(c.outEntries).toBe(1); + expect(c.outFailed).toBe(1); + }); + + // --- D4 方案B: transient escalation ----------------------------------- + + it('does not alarm on a single transient failure, escalates only after threshold', async () => { + const alarm = new AlarmManager({ ip: '1.1.1.1', version: 'v', userId: 'u' }); + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + flusher.setAlarmManager(alarm); + mockPostLogStoreLogs.mockRejectedValue(transientErr()); + + // Threshold is 3 consecutive all-failed cycles. + for (let i = 0; i < 2; i++) { + await flusher.send(buildTestEntry()); + await flusher.flush(); + } + expect(alarm.serialize()).toHaveLength(0); // still below threshold + + await flusher.send(buildTestEntry()); + await flusher.flush(); + const entries = alarm.serialize(); + expect(entries).toHaveLength(1); + expect(entries[0].alarm_type).toBe('FLUSH_SEND_ALARM'); + expect(entries[0].failure_class).toBe('transient'); + expect(entries[0].alarm_level).toBe('3'); + }); + + it('resets the transient streak after a success (intermittent failures stay silent)', async () => { + const alarm = new AlarmManager({ ip: '1.1.1.1', version: 'v', userId: 'u' }); + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + flusher.setAlarmManager(alarm); + + for (let i = 0; i < 5; i++) { + // fail, then succeed — never 3 in a row + mockPostLogStoreLogs.mockRejectedValueOnce(transientErr()); + await flusher.send(buildTestEntry()); + await flusher.flush(); + await flusher.send(buildTestEntry()); + await flusher.flush(); // success resets streak + } + expect(alarm.serialize()).toHaveLength(0); + }); + + it('always counts transient failures in outFailed regardless of alarming', async () => { + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + mockPostLogStoreLogs.mockRejectedValue(transientErr()); + + await flusher.send(buildTestEntry()); + await flusher.flush(); + expect(counters(flusher, 'user').outFailed).toBe(1); + }); + + // --- D3 / D4: config cooldown ----------------------------------------- + + it('alarms once for config failure within the cooldown window', async () => { + const alarm = new AlarmManager({ ip: '1.1.1.1', version: 'v', userId: 'u' }); + const record = vi.spyOn(alarm, 'record'); + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + flusher.setAlarmManager(alarm); + mockPostLogStoreLogs.mockRejectedValue(configErr()); + + // Several config failures in quick succession → only one alarm (cooldown). + for (let i = 0; i < 3; i++) { + await flusher.send(buildTestEntry()); + await flusher.flush(); + } + const configAlarms = record.mock.calls.filter(c => (c[3] as { failure_class?: string })?.failure_class === 'config'); + expect(configAlarms).toHaveLength(1); + expect(configAlarms[0][1]).toBe('1'); // level 1 + }); + + // --- D3 / D5: circuit breaker ----------------------------------------- + + it('trips the breaker after consecutive config failures and stops sending', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + mockPostLogStoreLogs.mockRejectedValue(configErr()); + + // 3 config failures trip the breaker. + for (let i = 0; i < 3; i++) { + await flusher.send(buildTestEntry()); + await flusher.flush(); + } + expect(mockPostLogStoreLogs).toHaveBeenCalledTimes(3); + const failWritesAfterTrip = mockFailureWrite.mock.calls.length; + + // Next cycle: breaker open, send is skipped, but still counted as failed. + await flusher.send(buildTestEntry()); + await flusher.flush(); + expect(mockPostLogStoreLogs).toHaveBeenCalledTimes(3); // no new send + expect(mockFailureWrite.mock.calls.length).toBe(failWritesAfterTrip); // no new persist + expect(counters(flusher, 'user').outFailed).toBe(4); // still counted + }); + + it('recovers automatically when a half-open probe succeeds', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const flusher = new SlsFlusher(makeConfig([akEndpoint('user', 'https://cn-shanghai.log.aliyuncs.com', 'p')]), '/tmp/data'); + mockPostLogStoreLogs.mockRejectedValue(configErr()); + + for (let i = 0; i < 3; i++) { + await flusher.send(buildTestEntry()); + await flusher.flush(); + } + // Advance past the backoff window; next flush is a half-open probe. + vi.setSystemTime(10_000); + mockPostLogStoreLogs.mockReset().mockResolvedValue(undefined); + await flusher.send(buildTestEntry()); + await flusher.flush(); + expect(mockPostLogStoreLogs).toHaveBeenCalledTimes(1); // probe went through + + // Breaker cleared: subsequent sends flow normally. + await flusher.send(buildTestEntry()); + await flusher.flush(); + expect(mockPostLogStoreLogs).toHaveBeenCalledTimes(2); + }); + + it('circuit breaker on one endpoint does not affect another', async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const flusher = new SlsFlusher(makeConfig([ + akEndpoint('bad', 'https://cn-shanghai.log.aliyuncs.com', 'bad'), + akEndpoint('good', 'https://cn-heyuan.log.aliyuncs.com', 'good'), + ]), '/tmp/data'); + // bad endpoint always config-fails; good always succeeds. + mockPostLogStoreLogs.mockImplementation((project: string) => + project === 'bad' ? Promise.reject(configErr()) : Promise.resolve(undefined), + ); + + for (let i = 0; i < 5; i++) { + await flusher.send(buildTestEntry()); + await flusher.flush(); + } + // good endpoint kept flushing every cycle. + expect(counters(flusher, 'good').outEntries).toBe(5); + expect(counters(flusher, 'good').outFailed).toBe(0); + }); + + // --- D2b / D6: oversize payload guard --------------------------------- + + it('drops an entry that stays oversize even after truncation, without blocking the batch', async () => { + const flusher = new SlsFlusher(makeConfig([wtEndpoint('internal', 'https://cn-heyuan.log.aliyuncs.com', 'p')]), '/tmp/data'); + + // One normal entry + one whose many fields are each huge (no single field to trim under cap). + await flusher.send(buildTestEntry()); + const oversize: Record = {}; + for (let i = 0; i < 40; i++) oversize[`f${i}`] = 'y'.repeat(100_000); // ~4MB spread across fields + await flusher.send(buildTestEntry(oversize)); + + await flusher.flush(); + + // Normal entry still sent; oversize entry dropped + persisted. + expect(fetchSpy).toHaveBeenCalled(); + const persistedPayload = mockFailureWrite.mock.calls.some(c => String((c[0] as { error: unknown }).error).includes('payload dropped')); + expect(persistedPayload).toBe(true); + expect(counters(flusher, 'internal').outFailed).toBeGreaterThanOrEqual(1); + }); + + it('truncates an entry with one oversize field so it fits and still sends', async () => { + const flusher = new SlsFlusher(makeConfig([wtEndpoint('internal', 'https://cn-heyuan.log.aliyuncs.com', 'p')]), '/tmp/data'); + + // Single field just over the 2.8MB cap → truncated, then sent. + await flusher.send(buildTestEntry({ 'gen_ai.completion': 'z'.repeat(3_000_000) })); + await flusher.flush(); + + expect(fetchSpy).toHaveBeenCalledOnce(); + const body = JSON.parse(String(fetchSpy.mock.calls[0][1].body)); + const sent = body.__logs__[0]['gen_ai.completion'] as string; + expect(sent.endsWith('...[TRUNCATED]')).toBe(true); + expect(Buffer.byteLength(JSON.stringify(body.__logs__[0]))).toBeLessThanOrEqual(2_800_000); + expect(counters(flusher, 'internal').outEntries).toBe(1); + }); +}); diff --git a/tests/unit/flushers/sls-transport.test.ts b/tests/unit/flushers/sls-transport.test.ts index bbb261b8c..13e6bf934 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, + classifyFailure, + FAILURE_CLASS_ALARM_LEVEL, +} from '../../../src/flushers/sls-transport.js'; describe('splitForWebtracking', () => { it('returns single chunk when under limits', () => { @@ -63,3 +69,38 @@ describe('isRetryable', () => { expect(isRetryable('string error')).toBe(false); }); }); + +describe('classifyFailure', () => { + it('classifies 413 / body-too-large as payload', () => { + expect(classifyFailure(new HttpError(413, 'Request Entity Too Large'))).toBe('payload'); + // ak SDK style: no status, message carries the API error + expect(classifyFailure(new Error('PostBodyTooLarge: body size must little than 10485760'))).toBe('payload'); + }); + + it('classifies 429 / ServerBusy as quota', () => { + expect(classifyFailure(new HttpError(429, 'rate limited'))).toBe('quota'); + expect(classifyFailure({ code: 'ServerBusy', message: 'slow down' })).toBe('quota'); + }); + + it('classifies terminal config errors as config', () => { + expect(classifyFailure(new HttpError(404, '{"errorCode":"ProjectNotExist"}'))).toBe('config'); + expect(classifyFailure(new HttpError(403, 'forbidden'))).toBe('config'); + // ak SDK style: errorCode field, no HTTP status + expect(classifyFailure({ errorCode: 'ProjectForbidden', message: 'forbidden' })).toBe('config'); + expect(classifyFailure({ code: 'ProjectInRecycleBin' })).toBe('config'); + }); + + it('classifies timeouts / network / unknown as transient', () => { + expect(classifyFailure(new Error('TimeoutError'))).toBe('transient'); + expect(classifyFailure(new Error('fetch failed'))).toBe('transient'); + expect(classifyFailure(new HttpError(500, 'internal'))).toBe('transient'); + expect(classifyFailure('some unknown error')).toBe('transient'); + }); + + it('maps each class to the expected alarm level', () => { + expect(FAILURE_CLASS_ALARM_LEVEL.transient).toBe('3'); + expect(FAILURE_CLASS_ALARM_LEVEL.quota).toBe('2'); + expect(FAILURE_CLASS_ALARM_LEVEL.config).toBe('1'); + expect(FAILURE_CLASS_ALARM_LEVEL.payload).toBe('1'); + }); +}); diff --git a/tests/unit/metrics/alarm-manager.test.ts b/tests/unit/metrics/alarm-manager.test.ts index 2a9c3a9a4..3103206cc 100644 --- a/tests/unit/metrics/alarm-manager.test.ts +++ b/tests/unit/metrics/alarm-manager.test.ts @@ -54,4 +54,23 @@ describe('AlarmManager', () => { manager.serialize(); expect(manager.serialize()).toEqual([]); }); + + it('carries failure_class and keeps distinct classes on the same endpoint separate', () => { + manager.record('FLUSH_SEND_ALARM', '3', 'timeout', { endpoint_name: 'ep1', failure_class: 'transient' }); + manager.record('FLUSH_SEND_ALARM', '1', 'project gone', { endpoint_name: 'ep1', failure_class: 'config' }); + + const entries = manager.serialize().sort((a, b) => (a.failure_class ?? '').localeCompare(b.failure_class ?? '')); + expect(entries).toHaveLength(2); + expect(entries[0].failure_class).toBe('config'); + expect(entries[0].alarm_level).toBe('1'); + expect(entries[1].failure_class).toBe('transient'); + expect(entries[1].alarm_level).toBe('3'); + }); + + it('omits failure_class when not provided (backward compatible)', () => { + manager.record('INPUT_STOP_ALARM', '3', 'timeout', { input_name: 'cursor-hook' }); + const entries = manager.serialize(); + expect(entries).toHaveLength(1); + expect(entries[0].failure_class).toBeUndefined(); + }); }); From 697f897f1beee451e7d419ad7569329e38e2c633 Mon Sep 17 00:00:00 2001 From: Snssn <1502062504@qq.com> Date: Thu, 13 Aug 2026 16:47:20 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(sls-flusher):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20proxy-404=20guard,=20breaker=20logging,=20byteSize?= =?UTF-8?q?=20reuse,=20EN=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - F1: only classify 404/403 as `config` when the body carries an SLS error code, so a proxy/CDN/WAF 404 no longer falsely trips the circuit breaker. - F2: log circuit breaker trip (warn), open-skip (debug), and recovery (info) so operators can reconstruct incident timelines. - F3: reuse the byteSize precomputed in enqueue() instead of re-serializing every entry in splitForWebtracking; only re-serialize after truncation. - F4: add the Send Failure Classification / Cooldown / Circuit Breaker section to the English docs/sls-output.md for parity with zh-CN. --- docs/sls-output.md | 13 +++++++++++++ src/flushers/sls-flusher.ts | 19 ++++++++++++++++--- src/flushers/sls-transport.ts | 8 +++++++- tests/unit/flushers/sls-transport.test.ts | 8 +++++++- 4 files changed, 43 insertions(+), 5 deletions(-) diff --git a/docs/sls-output.md b/docs/sls-output.md index ada3ab57e..4e8827329 100644 --- a/docs/sls-output.md +++ b/docs/sls-output.md @@ -153,6 +153,19 @@ ls ~/.loongsuite-pilot/logs/sls-failed-logs/ These JSONL records contain the endpoint, error summary, batch count, and batch byte estimate. They do **not** contain the failed batch payload, message content, request headers, or credentials, so they cannot be used to replay failed uploads. Files rotate by local date and at 10 MiB; the directory is limited to 50 MiB and also follows `retention.slsFailedDays` (7 days by default). +## Send Failure Classification, Cooldown & Circuit Breaker + +After retries are exhausted, a send failure is classified by actionability, which drives the alarm level and recovery behavior: + +| Class | Trigger | Alarm behavior | +|-------|---------|----------------| +| `transient` | Timeout / network failure / 5xx | Not alarmed per-occurrence; failures are only reflected in the failure metric. A single alarm is raised only when one endpoint fails whole batches for several consecutive flush cycles. | +| `quota` | 429 throttling | Alarmed per-occurrence (aggregated). | +| `config` | 404 / 403 with an SLS error code, or project not-exist / forbidden / in-recycle-bin | Alarm is cooldown-gated (once per hour) and trips a circuit breaker for that endpoint. | +| `payload` | 413, or a single entry larger than the per-request body cap | Alarmed (an oversize entry's largest field is truncated to fit, or the entry is dropped). | + +Circuit breaker: when an endpoint keeps returning terminal `config` failures, Pilot stops retrying it at high frequency and backs off exponentially (capped at 10 minutes). When the backoff elapses, one probe request is allowed through, and success clears the breaker automatically. Breaker state is in-memory and does not survive a restart. Regardless of whether an alarm is raised, failed entries are always counted in the flusher metric `out_failed_entries_total`. + Local JSONL output can help confirm whether collection itself is working before debugging SLS delivery: ```bash diff --git a/src/flushers/sls-flusher.ts b/src/flushers/sls-flusher.ts index f8600572b..4e3460050 100644 --- a/src/flushers/sls-flusher.ts +++ b/src/flushers/sls-flusher.ts @@ -250,6 +250,10 @@ export class SlsFlusher extends BaseFlusher { // that is exactly the pointless-request/write loop we are stopping. if (this.isCircuitOpen(endpoint.name, Date.now())) { if (counter) counter.outFailed += logs.length; + logger.debug('SLS circuit open, skipping send', { + endpoint: endpoint.name, + dropped: logs.length, + }); return Promise.resolve(); } @@ -316,7 +320,9 @@ export class SlsFlusher extends BaseFlusher { private onEndpointSuccess(name: string): void { // Any success (including a half-open probe) clears streak + breaker. this.transientFailStreak.delete(name); - this.circuits.delete(name); + if (this.circuits.delete(name)) { + logger.info('SLS circuit breaker recovered', { endpoint: name }); + } } private onEndpointFailure( @@ -387,6 +393,11 @@ export class SlsFlusher extends BaseFlusher { ? CIRCUIT_BASE_BACKOFF_MS : Math.min(c.backoffMs * 2, CIRCUIT_MAX_BACKOFF_MS); c.openUntil = Date.now() + c.backoffMs; + logger.warn('SLS circuit breaker tripped', { + endpoint: name, + configFails: c.configFails, + backoffMs: c.backoffMs, + }); } this.circuits.set(name, c); } @@ -591,7 +602,9 @@ export class SlsFlusher extends BaseFlusher { for (const raw of logs) { let log = raw; - let logSize = Buffer.byteLength(JSON.stringify(log.content)); + // byteSize was already computed in enqueue() from the same JSON.stringify — + // reuse it here and only re-serialize after a truncation mutates content. + let logSize = raw.byteSize; // A single entry over the cap can never fit any chunk. Try trimming its // largest field; if it still won't fit, drop it rather than emit a request @@ -603,7 +616,7 @@ export class SlsFlusher extends BaseFlusher { continue; } log = trimmed; - logSize = Buffer.byteLength(JSON.stringify(log.content)); + logSize = trimmed.byteSize; if (logSize > maxBytes) { dropped++; continue; diff --git a/src/flushers/sls-transport.ts b/src/flushers/sls-transport.ts index 49d9ba469..4c21eaff3 100644 --- a/src/flushers/sls-transport.ts +++ b/src/flushers/sls-transport.ts @@ -81,7 +81,13 @@ export function classifyFailure(err: unknown): FailureClass { if (status === 429 || /\bServerBusy\b|Throttl/i.test(msg)) { return 'quota'; } - if (status === 404 || status === 403 || CONFIG_ERROR_CODES.some(c => msg.includes(c))) { + // A known SLS error code is terminal regardless of status. A bare 404/403 + // only counts as config when the body looks like a structured SLS error — + // otherwise a proxy/CDN/WAF 404 would falsely trip the circuit breaker. + if (CONFIG_ERROR_CODES.some(c => msg.includes(c))) { + return 'config'; + } + if ((status === 404 || status === 403) && /errorCode/i.test(msg)) { return 'config'; } return 'transient'; diff --git a/tests/unit/flushers/sls-transport.test.ts b/tests/unit/flushers/sls-transport.test.ts index 13e6bf934..8f28a95b9 100644 --- a/tests/unit/flushers/sls-transport.test.ts +++ b/tests/unit/flushers/sls-transport.test.ts @@ -84,12 +84,18 @@ describe('classifyFailure', () => { it('classifies terminal config errors as config', () => { expect(classifyFailure(new HttpError(404, '{"errorCode":"ProjectNotExist"}'))).toBe('config'); - expect(classifyFailure(new HttpError(403, 'forbidden'))).toBe('config'); + expect(classifyFailure(new HttpError(403, '{"errorCode":"ProjectForbidden"}'))).toBe('config'); // ak SDK style: errorCode field, no HTTP status expect(classifyFailure({ errorCode: 'ProjectForbidden', message: 'forbidden' })).toBe('config'); expect(classifyFailure({ code: 'ProjectInRecycleBin' })).toBe('config'); }); + it('does not classify a bare (proxy) 404/403 as config', () => { + // No SLS errorCode in the body → likely a proxy/CDN/WAF, not a terminal SLS error. + expect(classifyFailure(new HttpError(404, 'nginx not found'))).toBe('transient'); + expect(classifyFailure(new HttpError(403, 'Forbidden'))).toBe('transient'); + }); + it('classifies timeouts / network / unknown as transient', () => { expect(classifyFailure(new Error('TimeoutError'))).toBe('transient'); expect(classifyFailure(new Error('fetch failed'))).toBe('transient');