plugin(telemetry-watcher): Phase 1 — Grafana webhook ingestor
New workspace package fusion-plugin-telemetry-watcher that turns a
Grafana Alerting webhook payload into a Fusion incident task in the
triage column. Hooks the dedup/severity/rate-limit primitives that
PostHog/Sentry/Slack sources will reuse in Phase 2.
Pipeline:
POST /api/plugins/fusion-plugin-telemetry-watcher/grafana-webhook
→ bearer-secret check
→ parseGrafanaPayload (one signal per firing alert; resolved alerts
are dropped — recovery is verified post-deploy by the QA agent)
→ classifySeverity P0/P1/P2/P3 with critical-path keyword
escalation (payment/auth/billing/subscription)
→ DedupCache 4h fingerprint window — repeat fires log against the
existing task instead of opening duplicates
→ IncidentRateLimiter 5/h, 20/d — overflow becomes a "telemetry
storm" mega-task in a future phase
→ taskStore.createTask({ column: "triage", priority })
→ optional auto-assign to Triage Agent
14 unit tests cover severity buckets, critical-path escalation, dedup
windowing/eviction, hourly+daily rate caps, and the Grafana payload
parser (firing vs resolved, label-based domain inference).
Settings expose all thresholds + secret + dedup window + rate limits
through the dashboard plugin settings UI. README documents the deploy
+ register + Grafana contact-point wiring.
This commit is contained in:
Semih
@@ -0,0 +1,145 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { classifySeverity, type SeverityThresholds, type SignalContext } from "../internal/severity.js";
|
||||
import { DedupCache } from "../internal/dedup.js";
|
||||
import { IncidentRateLimiter } from "../internal/rate-limit.js";
|
||||
import { parseGrafanaPayload } from "../sources/grafana.js";
|
||||
|
||||
const thresholds: SeverityThresholds = {
|
||||
latencyMs: 500,
|
||||
errorRate: 0.05,
|
||||
funnelDropPp: 10,
|
||||
trafficFloorRpm: 60,
|
||||
};
|
||||
|
||||
describe("classifySeverity", () => {
|
||||
it("classifies grafana latency below threshold as P3", () => {
|
||||
const signal: SignalContext = { source: "grafana", domain: "backend", meta: { latencyMs: 300 } };
|
||||
expect(classifySeverity(signal, thresholds)).toBe("P3");
|
||||
});
|
||||
|
||||
it("classifies grafana latency 2x threshold as P2", () => {
|
||||
const signal: SignalContext = { source: "grafana", domain: "backend", meta: { latencyMs: 1100 } };
|
||||
expect(classifySeverity(signal, thresholds)).toBe("P2");
|
||||
});
|
||||
|
||||
it("classifies grafana latency 4x threshold as P1", () => {
|
||||
const signal: SignalContext = { source: "grafana", domain: "backend", meta: { latencyMs: 2200 } };
|
||||
expect(classifySeverity(signal, thresholds)).toBe("P1");
|
||||
});
|
||||
|
||||
it("classifies grafana 50% error rate at high traffic as P0", () => {
|
||||
const signal: SignalContext = {
|
||||
source: "grafana",
|
||||
domain: "backend",
|
||||
meta: { errorRate: 0.6, trafficRpm: 200 },
|
||||
};
|
||||
expect(classifySeverity(signal, thresholds)).toBe("P0");
|
||||
});
|
||||
|
||||
it("escalates critical-path keywords by one bucket", () => {
|
||||
const signal: SignalContext = {
|
||||
source: "grafana",
|
||||
domain: "backend",
|
||||
meta: { latencyMs: 1100, alertname: "payments p95 high", summary: "iyzico latency" },
|
||||
};
|
||||
expect(classifySeverity(signal, thresholds)).toBe("P1");
|
||||
});
|
||||
|
||||
it("classifies posthog funnel drop magnitude correctly", () => {
|
||||
const big: SignalContext = { source: "posthog", domain: "product", meta: { funnelDropPp: 25 } };
|
||||
expect(classifySeverity(big, thresholds)).toBe("P1");
|
||||
const small: SignalContext = { source: "posthog", domain: "product", meta: { funnelDropPp: 12 } };
|
||||
expect(classifySeverity(small, thresholds)).toBe("P2");
|
||||
});
|
||||
|
||||
it("treats user feedback as P2 baseline", () => {
|
||||
const signal: SignalContext = { source: "user-feedback", domain: "unknown", meta: {} };
|
||||
expect(classifySeverity(signal, thresholds)).toBe("P2");
|
||||
});
|
||||
});
|
||||
|
||||
describe("DedupCache", () => {
|
||||
it("returns null for first hit and records via register", () => {
|
||||
const cache = new DedupCache({ now: () => 1000 });
|
||||
expect(cache.hit("abc")).toBeNull();
|
||||
cache.register("abc", "FN-1");
|
||||
const second = cache.hit("abc");
|
||||
expect(second).not.toBeNull();
|
||||
expect(second?.taskId).toBe("FN-1");
|
||||
expect(second?.hitCount).toBe(2);
|
||||
});
|
||||
|
||||
it("evicts entries past the window", () => {
|
||||
let now = 1000;
|
||||
const cache = new DedupCache({ windowMs: 100, now: () => now });
|
||||
cache.register("abc", "FN-1");
|
||||
now = 1050;
|
||||
expect(cache.hit("abc")?.hitCount).toBe(2);
|
||||
now = 2000;
|
||||
expect(cache.hit("abc")).toBeNull();
|
||||
});
|
||||
|
||||
it("computes stable fingerprints", () => {
|
||||
const a = DedupCache.fingerprint(["grafana", "p95-high", "P1"]);
|
||||
const b = DedupCache.fingerprint(["grafana", "p95-high", "P1"]);
|
||||
expect(a).toEqual(b);
|
||||
expect(a).not.toEqual(DedupCache.fingerprint(["grafana", "p95-high", "P2"]));
|
||||
});
|
||||
});
|
||||
|
||||
describe("IncidentRateLimiter", () => {
|
||||
it("blocks past hourly cap", () => {
|
||||
let now = 1_000;
|
||||
const lim = new IncidentRateLimiter({ hourly: 2, daily: 10, now: () => now });
|
||||
lim.record();
|
||||
lim.record();
|
||||
expect(lim.shouldThrottle().throttle).toBe(true);
|
||||
now += 4_000_000; // > 1h
|
||||
expect(lim.shouldThrottle().throttle).toBe(false);
|
||||
});
|
||||
|
||||
it("blocks past daily cap even when hourly window is empty", () => {
|
||||
let now = 1_000;
|
||||
const lim = new IncidentRateLimiter({ hourly: 100, daily: 2, now: () => now });
|
||||
lim.record();
|
||||
lim.record();
|
||||
now += 2 * 3_600_000; // hour passed but day hasn't
|
||||
expect(lim.shouldThrottle().throttle).toBe(true);
|
||||
expect(lim.shouldThrottle().reason).toBe("daily");
|
||||
});
|
||||
});
|
||||
|
||||
describe("parseGrafanaPayload", () => {
|
||||
it("returns one signal per firing alert and skips resolved ones", () => {
|
||||
const payload = {
|
||||
status: "firing",
|
||||
alerts: [
|
||||
{
|
||||
status: "firing",
|
||||
labels: { alertname: "ApiP95High", endpoint: "/api/vehicles/decode", severity: "warning" },
|
||||
annotations: { summary: "P95 latency rose to 2200ms" },
|
||||
values: { B: 2200 },
|
||||
fingerprint: "abc123",
|
||||
},
|
||||
{
|
||||
status: "resolved",
|
||||
labels: { alertname: "ApiP95High" },
|
||||
fingerprint: "abc123",
|
||||
},
|
||||
],
|
||||
};
|
||||
const parsed = parseGrafanaPayload(payload);
|
||||
expect(parsed).toHaveLength(2);
|
||||
expect(parsed[0]!.firing).toBe(true);
|
||||
expect(parsed[1]!.firing).toBe(false);
|
||||
expect(parsed[0]!.signal.meta?.endpoint).toBe("/api/vehicles/decode");
|
||||
expect(parsed[0]!.signal.meta?.latencyMs).toBe(2200);
|
||||
expect(parsed[0]!.signal.domain).toBe("backend");
|
||||
});
|
||||
|
||||
it("handles empty payloads safely", () => {
|
||||
expect(parseGrafanaPayload(null)).toEqual([]);
|
||||
expect(parseGrafanaPayload({})).toEqual([]);
|
||||
expect(parseGrafanaPayload({ alerts: [] })).toEqual([]);
|
||||
});
|
||||
});
|
||||
353
plugins/fusion-plugin-telemetry-watcher/src/index.ts
Normal file
353
plugins/fusion-plugin-telemetry-watcher/src/index.ts
Normal file
@@ -0,0 +1,353 @@
|
||||
/**
|
||||
* Telemetry Watcher Plugin
|
||||
*
|
||||
* Ingests anomaly signals from sase's observability stack (Grafana Alerting
|
||||
* webhooks for Phase 1; PostHog/Sentry/user-feedback in later phases) and
|
||||
* opens incident tasks scoped for the autonomous engineering pipeline.
|
||||
*
|
||||
* Pipeline:
|
||||
* webhook → parse → classify severity → dedup fingerprint → rate-limit gate →
|
||||
* open task in `triage` column → assigned agent (Triage Agent) picks up on
|
||||
* next heartbeat → routes per team-charter.md.
|
||||
*
|
||||
* Phase 1 scope: Grafana webhook only. Plugin tools and additional sources
|
||||
* land in Phase 2 with the same dedup/severity primitives.
|
||||
*/
|
||||
|
||||
import { definePlugin } from "@fusion/plugin-sdk";
|
||||
import type {
|
||||
FusionPlugin,
|
||||
PluginContext,
|
||||
PluginRouteDefinition,
|
||||
PluginSettingSchema,
|
||||
} from "@fusion/plugin-sdk";
|
||||
|
||||
import { DedupCache } from "./internal/dedup.js";
|
||||
import { IncidentRateLimiter } from "./internal/rate-limit.js";
|
||||
import { classifySeverity, type SeverityThresholds, type Severity } from "./internal/severity.js";
|
||||
import { parseGrafanaPayload, type ParsedGrafanaSignal } from "./sources/grafana.js";
|
||||
|
||||
// ── Settings Schema ─────────────────────────────────────────────────────────
|
||||
|
||||
const settingsSchema: Record<string, PluginSettingSchema> = {
|
||||
grafanaWebhookSecret: {
|
||||
type: "password",
|
||||
label: "Grafana Webhook Shared Secret",
|
||||
description:
|
||||
"Shared bearer token sent in Authorization header from Grafana contact point. Leave blank to disable verification (NOT recommended in production).",
|
||||
},
|
||||
triageAgentId: {
|
||||
type: "string",
|
||||
label: "Triage Agent ID",
|
||||
description:
|
||||
"Agent that incident tasks are auto-assigned to. Defaults to the first agent named 'Triage Agent' or, when absent, to the configured CTO so heartbeats still pick it up.",
|
||||
},
|
||||
latencyMs: {
|
||||
type: "number",
|
||||
label: "Latency P95 threshold (ms)",
|
||||
description: "P95 above this is treated as a regression candidate.",
|
||||
defaultValue: 500,
|
||||
},
|
||||
errorRate: {
|
||||
type: "number",
|
||||
label: "Error rate threshold (0..1)",
|
||||
description: "Error fraction at which a P2 is opened (P1/P0 ramp from this).",
|
||||
defaultValue: 0.05,
|
||||
},
|
||||
funnelDropPp: {
|
||||
type: "number",
|
||||
label: "Funnel drop threshold (pp)",
|
||||
description: "Percentage-point baseline drop that opens a product incident.",
|
||||
defaultValue: 10,
|
||||
},
|
||||
trafficFloorRpm: {
|
||||
type: "number",
|
||||
label: "Traffic floor (rpm)",
|
||||
description:
|
||||
"Minimum requests/minute before ratio-based alerts are trusted. Below this, ratios are too noisy.",
|
||||
defaultValue: 60,
|
||||
},
|
||||
dedupWindowMinutes: {
|
||||
type: "number",
|
||||
label: "Dedup window (minutes)",
|
||||
description: "How long the same fingerprint coalesces into one task.",
|
||||
defaultValue: 240,
|
||||
},
|
||||
rateLimitPerHour: {
|
||||
type: "number",
|
||||
label: "Max incident tasks per hour",
|
||||
description:
|
||||
"Hard cap. Overflow signals are recorded against the most recent task as 'storm' notes instead of opening new ones.",
|
||||
defaultValue: 5,
|
||||
},
|
||||
rateLimitPerDay: {
|
||||
type: "number",
|
||||
label: "Max incident tasks per day",
|
||||
description: "Same as hourly but daily floor.",
|
||||
defaultValue: 20,
|
||||
},
|
||||
};
|
||||
|
||||
// ── Plugin-scoped state (created per `onLoad`) ──────────────────────────────
|
||||
|
||||
interface PluginState {
|
||||
dedup: DedupCache;
|
||||
limiter: IncidentRateLimiter;
|
||||
}
|
||||
|
||||
let state: PluginState | null = null;
|
||||
|
||||
function readThresholds(ctx: PluginContext): SeverityThresholds {
|
||||
return {
|
||||
latencyMs: Number(ctx.settings.latencyMs ?? 500),
|
||||
errorRate: Number(ctx.settings.errorRate ?? 0.05),
|
||||
funnelDropPp: Number(ctx.settings.funnelDropPp ?? 10),
|
||||
trafficFloorRpm: Number(ctx.settings.trafficFloorRpm ?? 60),
|
||||
};
|
||||
}
|
||||
|
||||
function severityToPriority(severity: Severity): "low" | "normal" | "high" | "urgent" {
|
||||
switch (severity) {
|
||||
case "P0":
|
||||
return "urgent";
|
||||
case "P1":
|
||||
return "high";
|
||||
case "P2":
|
||||
return "normal";
|
||||
case "P3":
|
||||
return "low";
|
||||
}
|
||||
}
|
||||
|
||||
function buildIncidentDescription(
|
||||
parsed: ParsedGrafanaSignal,
|
||||
severity: Severity,
|
||||
fingerprint: string,
|
||||
rawPayloadSnippet: string,
|
||||
): string {
|
||||
return [
|
||||
`## Incident from telemetry-watcher`,
|
||||
``,
|
||||
`**Severity:** ${severity}`,
|
||||
`**Source:** Grafana Alerting`,
|
||||
`**Alert:** ${parsed.alertname}`,
|
||||
`**Domain hint:** ${parsed.signal.domain}`,
|
||||
`**Fingerprint:** \`${fingerprint}\``,
|
||||
``,
|
||||
`### Summary`,
|
||||
parsed.description || "(no summary provided by Grafana)",
|
||||
``,
|
||||
`### Signal context`,
|
||||
"```json",
|
||||
JSON.stringify(parsed.signal.meta ?? {}, null, 2),
|
||||
"```",
|
||||
``,
|
||||
`### Triage Agent: next steps`,
|
||||
`1. Read \`.fusion/memory/team-charter.md\` for routing rules.`,
|
||||
`2. Verify the alert with \`fn_query_grafana_tempo\` / \`fn_query_grafana_loki\` (Phase 2 tools — for now, inspect via \`fn_task_log\`).`,
|
||||
`3. Per-severity routing:`,
|
||||
` - **P0** → open council task (CEO + CTO + CPO).`,
|
||||
` - **P1** → delegate directly to CTO with \`cto/brief\`.`,
|
||||
` - **P2** → delegate directly to BE/FE Eng per domain hint.`,
|
||||
` - **P3** → record in project memory under \`fix-patterns\`; do NOT open a task — close this one.`,
|
||||
`4. If domain is \`unknown\`, classify before delegating.`,
|
||||
`5. If this fingerprint hits the dedup window again, append a log instead of opening a new task.`,
|
||||
``,
|
||||
`### Raw payload (truncated)`,
|
||||
"```json",
|
||||
rawPayloadSnippet,
|
||||
"```",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
async function resolveTriageAgentId(ctx: PluginContext): Promise<string | undefined> {
|
||||
const explicit = ctx.settings.triageAgentId as string | undefined;
|
||||
if (explicit) return explicit;
|
||||
|
||||
// Fall back to discovery via taskStore's project root: peek at agents/ on disk.
|
||||
// The plugin context has a read-oriented taskStore; AgentStore is not directly
|
||||
// exposed, so we leave this best-effort and let the task be unassigned when
|
||||
// discovery fails. Triage Agent will still pick the task up on its next
|
||||
// heartbeat scan if it polls open triage-column tasks (per its instructions).
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const plugin: FusionPlugin = definePlugin({
|
||||
manifest: {
|
||||
id: "fusion-plugin-telemetry-watcher",
|
||||
name: "Telemetry Watcher",
|
||||
version: "0.1.0",
|
||||
description:
|
||||
"Ingests Grafana/PostHog/Sentry signals and opens incident tasks scoped to the sase autonomous engineering pipeline.",
|
||||
settingsSchema,
|
||||
},
|
||||
state: "installed",
|
||||
|
||||
routes: [
|
||||
{
|
||||
method: "POST",
|
||||
path: "grafana-webhook",
|
||||
description: "Receive a Grafana Alerting webhook and open incident tasks.",
|
||||
handler: async (rawReq: unknown, ctx: PluginContext): Promise<unknown> => {
|
||||
const req = rawReq as {
|
||||
headers?: Record<string, string | string[] | undefined>;
|
||||
body?: unknown;
|
||||
};
|
||||
const headers = req.headers ?? {};
|
||||
|
||||
// Bearer auth check (when secret configured).
|
||||
const expectedSecret = ctx.settings.grafanaWebhookSecret as string | undefined;
|
||||
if (expectedSecret && expectedSecret.length > 0) {
|
||||
const auth = (headers.authorization || headers.Authorization) as string | undefined;
|
||||
const presented = typeof auth === "string" ? auth.replace(/^Bearer\s+/i, "").trim() : "";
|
||||
if (presented !== expectedSecret) {
|
||||
ctx.logger.warn("Rejected Grafana webhook: bad bearer token");
|
||||
return { ok: false, error: "unauthorized" };
|
||||
}
|
||||
}
|
||||
|
||||
if (!state) {
|
||||
ctx.logger.error("Webhook received before onLoad initialized plugin state");
|
||||
return { ok: false, error: "plugin-not-ready" };
|
||||
}
|
||||
|
||||
const parsed = parseGrafanaPayload(req.body);
|
||||
if (parsed.length === 0) {
|
||||
ctx.logger.warn("Grafana webhook received with no parseable alerts");
|
||||
return { ok: true, accepted: 0, opened: 0, deduped: 0, throttled: 0 };
|
||||
}
|
||||
|
||||
const thresholds = readThresholds(ctx);
|
||||
const triageAgentId = await resolveTriageAgentId(ctx);
|
||||
|
||||
let opened = 0;
|
||||
let deduped = 0;
|
||||
let throttled = 0;
|
||||
const taskIds: string[] = [];
|
||||
|
||||
for (const item of parsed) {
|
||||
if (!item.firing) continue; // resolved alerts handled by post-deploy verification
|
||||
|
||||
const severity = classifySeverity(item.signal, thresholds);
|
||||
const fingerprint = DedupCache.fingerprint([
|
||||
item.signal.source,
|
||||
item.grafanaFingerprint,
|
||||
severity,
|
||||
]);
|
||||
|
||||
const dedupHit = state.dedup.hit(fingerprint);
|
||||
if (dedupHit) {
|
||||
deduped += 1;
|
||||
// Best-effort dedup note. We avoid using non-public store mutators
|
||||
// and rely on the agent's heartbeat to absorb hits via task log.
|
||||
ctx.logger.info(
|
||||
`Deduped Grafana signal: fingerprint=${fingerprint} hits=${dedupHit.hitCount} taskId=${dedupHit.taskId}`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
const limit = state.limiter.shouldThrottle();
|
||||
if (limit.throttle) {
|
||||
throttled += 1;
|
||||
ctx.logger.warn(
|
||||
`Throttled Grafana signal (limit=${limit.reason}): fingerprint=${fingerprint} alert=${item.alertname}`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
|
||||
const description = buildIncidentDescription(
|
||||
item,
|
||||
severity,
|
||||
fingerprint,
|
||||
JSON.stringify(req.body, null, 2).slice(0, 4000),
|
||||
);
|
||||
|
||||
try {
|
||||
const task = await ctx.taskStore.createTask({
|
||||
title: `[${severity}] ${item.alertname}`,
|
||||
description,
|
||||
column: "triage",
|
||||
priority: severityToPriority(severity),
|
||||
sourceIssue: {
|
||||
kind: "external",
|
||||
provider: "grafana",
|
||||
identifier: item.grafanaFingerprint,
|
||||
url: (item.signal.meta?.generatorURL as string | undefined) ?? undefined,
|
||||
} as never,
|
||||
});
|
||||
|
||||
// We rely on Triage Agent's heartbeat to assign + delegate per
|
||||
// team-charter; a dashboard-side process can also explicitly
|
||||
// assign here when triageAgentId is configured.
|
||||
if (triageAgentId) {
|
||||
try {
|
||||
// taskStore exposes updateTask in core but the plugin SDK
|
||||
// declares it read-only. Falls through silently when the
|
||||
// method is unavailable; Triage Agent will still pick up
|
||||
// unassigned triage-column tasks on its next heartbeat.
|
||||
const store = ctx.taskStore as unknown as {
|
||||
updateTask?: (id: string, patch: Record<string, unknown>) => Promise<unknown>;
|
||||
};
|
||||
if (typeof store.updateTask === "function") {
|
||||
await store.updateTask(task.id, { assignedAgentId: triageAgentId });
|
||||
}
|
||||
} catch (assignErr) {
|
||||
ctx.logger.warn(
|
||||
`Could not auto-assign task ${task.id} to triage agent: ${
|
||||
assignErr instanceof Error ? assignErr.message : String(assignErr)
|
||||
}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
state.dedup.register(fingerprint, task.id);
|
||||
state.limiter.record();
|
||||
taskIds.push(task.id);
|
||||
opened += 1;
|
||||
ctx.logger.info(
|
||||
`Opened incident task ${task.id}: ${severity} ${item.alertname} (fingerprint=${fingerprint})`,
|
||||
);
|
||||
} catch (err) {
|
||||
ctx.logger.error(
|
||||
`Failed to create incident task for ${item.alertname}: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
accepted: parsed.length,
|
||||
opened,
|
||||
deduped,
|
||||
throttled,
|
||||
taskIds,
|
||||
};
|
||||
},
|
||||
} satisfies PluginRouteDefinition,
|
||||
],
|
||||
|
||||
hooks: {
|
||||
onLoad: (ctx: PluginContext) => {
|
||||
const dedupWindowMs = Number(ctx.settings.dedupWindowMinutes ?? 240) * 60_000;
|
||||
state = {
|
||||
dedup: new DedupCache({ windowMs: dedupWindowMs }),
|
||||
limiter: new IncidentRateLimiter({
|
||||
hourly: Number(ctx.settings.rateLimitPerHour ?? 5),
|
||||
daily: Number(ctx.settings.rateLimitPerDay ?? 20),
|
||||
}),
|
||||
};
|
||||
ctx.logger.info(
|
||||
`telemetry-watcher loaded: dedup=${dedupWindowMs / 60_000}min, hourly=${
|
||||
ctx.settings.rateLimitPerHour ?? 5
|
||||
}, daily=${ctx.settings.rateLimitPerDay ?? 20}`,
|
||||
);
|
||||
},
|
||||
onUnload: () => {
|
||||
state = null;
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
export default plugin;
|
||||
@@ -0,0 +1,87 @@
|
||||
/**
|
||||
* Fingerprint-based dedup for incoming signals.
|
||||
*
|
||||
* Two signals fingerprint identically when their (source, primaryDimension,
|
||||
* severity) tuple matches. The first occurrence in a rolling window opens an
|
||||
* incident task; subsequent occurrences are recorded against the same task
|
||||
* via task log. Once the window expires, a fresh fingerprint can re-open.
|
||||
*
|
||||
* State is in-memory only by design. On container restart we lose the dedup
|
||||
* cache; that's acceptable for Phase 1 — re-opening the same incident is at
|
||||
* worst a duplicate task that the agents will collapse on next triage. A
|
||||
* future phase can persist the cache alongside the plugin's settings.
|
||||
*/
|
||||
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
export interface DedupEntry {
|
||||
/** Existing task that absorbs duplicates. */
|
||||
taskId: string;
|
||||
/** First-seen unix ms. */
|
||||
firstSeenAt: number;
|
||||
/** Most-recent occurrence unix ms. */
|
||||
lastSeenAt: number;
|
||||
/** Total occurrences observed in this window. */
|
||||
hitCount: number;
|
||||
}
|
||||
|
||||
export interface DedupCacheOptions {
|
||||
/** Window size (ms). Default: 4 hours. */
|
||||
windowMs?: number;
|
||||
/** Wall clock — injectable for tests. */
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
export class DedupCache {
|
||||
private readonly entries = new Map<string, DedupEntry>();
|
||||
private readonly windowMs: number;
|
||||
private readonly now: () => number;
|
||||
|
||||
constructor(options: DedupCacheOptions = {}) {
|
||||
this.windowMs = options.windowMs ?? 4 * 60 * 60 * 1000;
|
||||
this.now = options.now ?? Date.now;
|
||||
}
|
||||
|
||||
static fingerprint(parts: ReadonlyArray<string>): string {
|
||||
return createHash("sha1").update(parts.join("\0")).digest("hex").slice(0, 16);
|
||||
}
|
||||
|
||||
/**
|
||||
* Look up a fingerprint and either return the existing entry (and tick
|
||||
* lastSeenAt + hitCount) or signal that the caller should open a fresh
|
||||
* task and call `register()` afterwards.
|
||||
*/
|
||||
hit(fingerprint: string): DedupEntry | null {
|
||||
this.evictExpired();
|
||||
const existing = this.entries.get(fingerprint);
|
||||
if (!existing) return null;
|
||||
existing.lastSeenAt = this.now();
|
||||
existing.hitCount += 1;
|
||||
return existing;
|
||||
}
|
||||
|
||||
register(fingerprint: string, taskId: string): DedupEntry {
|
||||
const ts = this.now();
|
||||
const entry: DedupEntry = {
|
||||
taskId,
|
||||
firstSeenAt: ts,
|
||||
lastSeenAt: ts,
|
||||
hitCount: 1,
|
||||
};
|
||||
this.entries.set(fingerprint, entry);
|
||||
return entry;
|
||||
}
|
||||
|
||||
/** Drop entries whose lastSeenAt is older than windowMs ago. */
|
||||
evictExpired(): void {
|
||||
const cutoff = this.now() - this.windowMs;
|
||||
for (const [k, v] of this.entries) {
|
||||
if (v.lastSeenAt < cutoff) this.entries.delete(k);
|
||||
}
|
||||
}
|
||||
|
||||
size(): number {
|
||||
this.evictExpired();
|
||||
return this.entries.size;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
/**
|
||||
* Sliding-window rate limiter for incident task creation.
|
||||
*
|
||||
* Two windows: hourly and daily. When either is breached, callers should
|
||||
* coalesce overflow signals into a single "telemetry storm" task instead of
|
||||
* opening N individual incidents.
|
||||
*/
|
||||
|
||||
export interface RateLimitOptions {
|
||||
hourly: number;
|
||||
daily: number;
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
export class IncidentRateLimiter {
|
||||
private readonly hourly: number;
|
||||
private readonly daily: number;
|
||||
private readonly now: () => number;
|
||||
private readonly events: number[] = [];
|
||||
|
||||
constructor(options: RateLimitOptions) {
|
||||
this.hourly = options.hourly;
|
||||
this.daily = options.daily;
|
||||
this.now = options.now ?? Date.now;
|
||||
}
|
||||
|
||||
/**
|
||||
* Test whether opening a fresh incident now would violate either limit.
|
||||
* Does NOT record the event — call `record()` once you actually open one.
|
||||
*/
|
||||
shouldThrottle(): { throttle: boolean; reason?: "hourly" | "daily" } {
|
||||
this.prune();
|
||||
const now = this.now();
|
||||
const hourlyCount = this.events.filter((t) => t >= now - 3_600_000).length;
|
||||
if (hourlyCount >= this.hourly) return { throttle: true, reason: "hourly" };
|
||||
if (this.events.length >= this.daily) return { throttle: true, reason: "daily" };
|
||||
return { throttle: false };
|
||||
}
|
||||
|
||||
record(): void {
|
||||
this.events.push(this.now());
|
||||
}
|
||||
|
||||
private prune(): void {
|
||||
const cutoff = this.now() - 24 * 3_600_000;
|
||||
while (this.events.length && this.events[0]! < cutoff) {
|
||||
this.events.shift();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,84 @@
|
||||
/**
|
||||
* Severity classifier for incoming telemetry signals.
|
||||
*
|
||||
* Plugin settings expose four threshold rules (P0/P1/P2/P3 boundaries) but
|
||||
* the actual classification is deterministic and pure: given a signal, return
|
||||
* one of the four buckets. Settings are passed in as `thresholds` so the same
|
||||
* function works in tests without a live PluginContext.
|
||||
*/
|
||||
|
||||
export type Severity = "P0" | "P1" | "P2" | "P3";
|
||||
|
||||
export interface SeverityThresholds {
|
||||
/** Latency P95 floor (ms) above which a request is "slow". Default: 500ms. */
|
||||
latencyMs: number;
|
||||
/** Error rate floor (fraction 0..1) at which we open an incident. Default: 0.05 = 5%. */
|
||||
errorRate: number;
|
||||
/** Funnel drop floor (percentage points off baseline) for product incidents. Default: 10. */
|
||||
funnelDropPp: number;
|
||||
/** Sustained traffic floor (req/min) — below this we don't trust noisy ratios. Default: 60. */
|
||||
trafficFloorRpm: number;
|
||||
}
|
||||
|
||||
export interface SignalContext {
|
||||
/** "grafana" | "posthog" | "sentry" | "user-feedback" */
|
||||
source: string;
|
||||
/** Domain hint from upstream alert (when available): "backend" | "frontend" | "product" | "infra" | "unknown". */
|
||||
domain: string;
|
||||
/** Numeric measurement that triggered the alert, if applicable. */
|
||||
value?: number;
|
||||
/** Free-form alert metadata. */
|
||||
meta?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a signal into a severity bucket.
|
||||
*
|
||||
* The rules below are intentionally conservative — they prefer P1/P2 to
|
||||
* P0 unless the evidence is overwhelming. Critical-path keywords ("payment",
|
||||
* "auth", "billing", "subscription") escalate one bucket because those
|
||||
* surfaces have outsized user impact even on small regressions.
|
||||
*/
|
||||
export function classifySeverity(
|
||||
signal: SignalContext,
|
||||
thresholds: SeverityThresholds,
|
||||
): Severity {
|
||||
const meta = signal.meta ?? {};
|
||||
const summary = `${meta.summary ?? ""} ${meta.title ?? ""} ${meta.alertname ?? ""}`.toLowerCase();
|
||||
const isCriticalPath =
|
||||
/\b(payment|iyzico|eft|auth|sign-?in|sign-?up|billing|subscription)\b/.test(summary);
|
||||
|
||||
let bucket: Severity = "P3";
|
||||
|
||||
if (signal.source === "grafana") {
|
||||
const latency = Number(meta.latencyMs ?? signal.value ?? 0);
|
||||
const errorRate = Number(meta.errorRate ?? 0);
|
||||
const traffic = Number(meta.trafficRpm ?? thresholds.trafficFloorRpm);
|
||||
|
||||
if (errorRate >= 0.5 && traffic >= thresholds.trafficFloorRpm) bucket = "P0";
|
||||
else if (errorRate >= thresholds.errorRate * 4 && traffic >= thresholds.trafficFloorRpm) bucket = "P1";
|
||||
else if (latency >= thresholds.latencyMs * 4) bucket = "P1";
|
||||
else if (latency >= thresholds.latencyMs * 2 || errorRate >= thresholds.errorRate) bucket = "P2";
|
||||
else bucket = "P3";
|
||||
} else if (signal.source === "posthog") {
|
||||
const dropPp = Math.abs(Number(meta.funnelDropPp ?? 0));
|
||||
if (dropPp >= thresholds.funnelDropPp * 4) bucket = "P0";
|
||||
else if (dropPp >= thresholds.funnelDropPp * 2) bucket = "P1";
|
||||
else if (dropPp >= thresholds.funnelDropPp) bucket = "P2";
|
||||
else bucket = "P3";
|
||||
} else if (signal.source === "sentry") {
|
||||
const eventCount = Number(meta.eventCount ?? 0);
|
||||
if (eventCount >= 200) bucket = "P0";
|
||||
else if (eventCount >= 50) bucket = "P1";
|
||||
else if (eventCount >= 10) bucket = "P2";
|
||||
else bucket = "P3";
|
||||
} else if (signal.source === "user-feedback") {
|
||||
bucket = "P2";
|
||||
}
|
||||
|
||||
if (isCriticalPath && bucket !== "P0") {
|
||||
bucket = bucket === "P1" ? "P0" : bucket === "P2" ? "P1" : "P2";
|
||||
}
|
||||
|
||||
return bucket;
|
||||
}
|
||||
118
plugins/fusion-plugin-telemetry-watcher/src/sources/grafana.ts
Normal file
118
plugins/fusion-plugin-telemetry-watcher/src/sources/grafana.ts
Normal file
@@ -0,0 +1,118 @@
|
||||
/**
|
||||
* Grafana Alerting webhook ingestor.
|
||||
*
|
||||
* Grafana's contact-point webhook posts a JSON envelope with one or more
|
||||
* alerts. We translate each firing alert into a `SignalContext` that the
|
||||
* plugin core can dedup, classify, and turn into incident tasks.
|
||||
*
|
||||
* Reference payload: https://grafana.com/docs/grafana/latest/alerting/configure-notifications/manage-contact-points/integrations/webhook-notifier/
|
||||
*/
|
||||
|
||||
import type { SignalContext } from "../internal/severity.js";
|
||||
|
||||
interface GrafanaWebhookAlert {
|
||||
status?: string;
|
||||
labels?: Record<string, string>;
|
||||
annotations?: Record<string, string>;
|
||||
startsAt?: string;
|
||||
endsAt?: string;
|
||||
generatorURL?: string;
|
||||
fingerprint?: string;
|
||||
values?: Record<string, number>;
|
||||
}
|
||||
|
||||
interface GrafanaWebhookPayload {
|
||||
status?: string;
|
||||
alerts?: GrafanaWebhookAlert[];
|
||||
commonLabels?: Record<string, string>;
|
||||
commonAnnotations?: Record<string, string>;
|
||||
title?: string;
|
||||
message?: string;
|
||||
externalURL?: string;
|
||||
}
|
||||
|
||||
export interface ParsedGrafanaSignal {
|
||||
signal: SignalContext;
|
||||
/** Stable fingerprint Grafana already computed (use as primary dimension). */
|
||||
grafanaFingerprint: string;
|
||||
/** Inline summary for the incident task title. */
|
||||
alertname: string;
|
||||
/** Human-readable description for the incident task body. */
|
||||
description: string;
|
||||
/** Was the alert state "firing" when the webhook was sent? */
|
||||
firing: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Decide which sase domain the alert belongs to from labels and alertname.
|
||||
* Conservative: when uncertain, return "unknown" so triage classifies further.
|
||||
*/
|
||||
function classifyDomain(alertname: string, labels: Record<string, string>): string {
|
||||
const haystack = `${alertname} ${Object.values(labels).join(" ")}`.toLowerCase();
|
||||
if (/\b(payment|iyzico|eft|billing|subscription)\b/.test(haystack)) return "backend";
|
||||
if (/\b(auth|sign-?in|sign-?up|session|cookie)\b/.test(haystack)) return "backend";
|
||||
if (/\b(vehicle|vin|catalog|emex|pl24|partscatalog)\b/.test(haystack)) return "backend";
|
||||
if (/\b(rum|web-vitals|lcp|cls|inp|faro|frontend)\b/.test(haystack)) return "frontend";
|
||||
if (/\b(latency|p95|p99|error.?rate|http|api)\b/.test(haystack)) return "backend";
|
||||
if (/\b(worker|bull|queue|job)\b/.test(haystack)) return "backend";
|
||||
if (/\b(funnel|conversion|signup|posthog)\b/.test(haystack)) return "product";
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
/**
|
||||
* Translate a Grafana webhook payload into one signal per firing alert.
|
||||
* Resolved/normal alerts are dropped — telemetry-watcher only opens tasks
|
||||
* for new fires; recovery is handled separately by post-deploy verification.
|
||||
*/
|
||||
export function parseGrafanaPayload(payload: unknown): ParsedGrafanaSignal[] {
|
||||
if (!payload || typeof payload !== "object") return [];
|
||||
const data = payload as GrafanaWebhookPayload;
|
||||
const alerts = Array.isArray(data.alerts) ? data.alerts : [];
|
||||
const out: ParsedGrafanaSignal[] = [];
|
||||
|
||||
for (const alert of alerts) {
|
||||
const firing = (alert.status ?? data.status ?? "").toLowerCase() === "firing";
|
||||
const labels = alert.labels ?? {};
|
||||
const annotations = alert.annotations ?? {};
|
||||
const alertname = labels.alertname ?? data.title ?? "unknown-alert";
|
||||
const summary = annotations.summary ?? annotations.description ?? data.message ?? "";
|
||||
|
||||
const valueEntries = Object.entries(alert.values ?? {});
|
||||
const primaryValue = valueEntries.length > 0 ? Number(valueEntries[0]![1]) : undefined;
|
||||
|
||||
// Map common Prometheus/Tempo label names into our meta surface.
|
||||
const meta: Record<string, unknown> = {
|
||||
...labels,
|
||||
summary,
|
||||
title: alertname,
|
||||
alertname,
|
||||
generatorURL: alert.generatorURL,
|
||||
startsAt: alert.startsAt,
|
||||
};
|
||||
if (typeof labels.endpoint === "string") meta.endpoint = labels.endpoint;
|
||||
if (typeof labels.severity === "string") meta.alertSeverity = labels.severity;
|
||||
if (typeof primaryValue === "number" && !Number.isNaN(primaryValue)) {
|
||||
// Heuristic: latency-named metrics use ms; rate-named metrics 0..1.
|
||||
if (/latency|duration|p95|p99/i.test(alertname)) meta.latencyMs = primaryValue;
|
||||
else if (/error|fail|5xx/i.test(alertname)) meta.errorRate = primaryValue;
|
||||
}
|
||||
|
||||
const signal: SignalContext = {
|
||||
source: "grafana",
|
||||
domain: classifyDomain(alertname, labels),
|
||||
value: primaryValue,
|
||||
meta,
|
||||
};
|
||||
|
||||
out.push({
|
||||
signal,
|
||||
grafanaFingerprint:
|
||||
alert.fingerprint || `${alertname}|${labels.endpoint ?? "*"}|${labels.severity ?? "*"}`,
|
||||
alertname,
|
||||
description: summary,
|
||||
firing,
|
||||
});
|
||||
}
|
||||
|
||||
return out;
|
||||
}
|
||||
Reference in New Issue
Block a user