Skip to content

Commit 68e906f

Browse files
kathiekiwiTrigger.dev RepoOps
authored andcommitted
feat(alerts): alerts foundation — contracts, storage, evaluators, engine, delivery
Mono-RevId: 0e4b1babcfa22894459865e3d0076462a0711db9
1 parent dfa3a32 commit 68e906f

122 files changed

Lines changed: 19140 additions & 312 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.env.schema‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,76 @@ POSTHOG_PROJECT_KEY=
198198
# INTERNAL_OTEL_TRACE_LOGGING_ENABLED=1
199199
# INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED=0
200200

201+
# Alerts engine
202+
# @sensitive=false
203+
ALERTS_ENABLED=0
204+
# @sensitive=false
205+
ALERT_QUERY_CONCURRENCY=1
206+
# @sensitive=false
207+
ALERT_FAILURE_RATE_MIN_VOLUME=10
208+
# @sensitive=false
209+
ALERT_SNAPSHOT_RANGE_SEC=7776000
210+
# Alert delivery and engine tuning. Defaults shown; uncomment to override
211+
# ALERT_EMAIL_TRANSPORT=resend # resend | smtp | aws-ses
212+
# ALERT_FROM_EMAIL=
213+
# ALERT_REPLY_TO_EMAIL=
214+
# ALERT_RESEND_API_KEY=
215+
# ALERT_SMTP_HOST=
216+
# ALERT_SMTP_PORT=
217+
# ALERT_SMTP_SECURE=
218+
# ALERT_SMTP_USER=
219+
# ALERT_SMTP_PASSWORD=
220+
# ALERT_CHECKPOINT_LOOKBACK_SEC=3600
221+
# ALERT_SEEN_RETENTION_SEC=604800
222+
# ALERT_DELIVERY_MAX_ATTEMPTS=8
223+
# ALERT_DELIVERY_MAX_AGE_HOURS=24
224+
# Testing only: caps the alert evaluation cadence (seconds)
225+
# ALERT_EVALUATION_CADENCE_CAP_SEC=
226+
# ALERT_RATE_LIMITER_EMISSION_INTERVAL=2500
227+
# ALERT_RATE_LIMITER_BURST_TOLERANCE=10000
228+
# ALERT_RATE_LIMITER_REDIS_HOST= # Defaults to REDIS_HOST (same for the other ALERT_RATE_LIMITER_REDIS_* vars)
229+
# ALERT_RATE_LIMITER_REDIS_READER_HOST=
230+
# ALERT_RATE_LIMITER_REDIS_READER_PORT=
231+
# ALERT_RATE_LIMITER_REDIS_PORT=
232+
# ALERT_RATE_LIMITER_REDIS_USERNAME=
233+
# ALERT_RATE_LIMITER_REDIS_PASSWORD=
234+
# ALERT_RATE_LIMITER_REDIS_TLS_DISABLED=false
235+
# ALERT_RATE_LIMITER_REDIS_CLUSTER_MODE_ENABLED=0
236+
# ALERTS_WORKER_ENABLED=true # Defaults to WORKER_ENABLED
237+
# ALERTS_WORKER_CONCURRENCY_WORKERS=2
238+
# ALERTS_WORKER_CONCURRENCY_TASKS_PER_WORKER=10
239+
# ALERTS_WORKER_POLL_INTERVAL=1000
240+
# ALERTS_WORKER_IMMEDIATE_POLL_INTERVAL=100
241+
# ALERTS_WORKER_CONCURRENCY_LIMIT=50
242+
# ALERTS_WORKER_SHUTDOWN_TIMEOUT_MS=60000
243+
# ALERTS_WORKER_LOG_LEVEL=info
244+
# ALERTS_WORKER_REDIS_HOST= # Defaults to REDIS_HOST (same for the other ALERTS_WORKER_REDIS_* vars)
245+
# ALERTS_WORKER_REDIS_READER_HOST=
246+
# ALERTS_WORKER_REDIS_READER_PORT=
247+
# ALERTS_WORKER_REDIS_PORT=
248+
# ALERTS_WORKER_REDIS_USERNAME=
249+
# ALERTS_WORKER_REDIS_PASSWORD=
250+
# ALERTS_WORKER_REDIS_TLS_DISABLED=false
251+
# ALERTS_WORKER_REDIS_CLUSTER_MODE_ENABLED=0
252+
# ALERT_ENGINE_WORKER_ENABLED=true # Defaults to WORKER_ENABLED
253+
# ALERT_ENGINE_WORKER_CONCURRENCY_WORKERS=1
254+
# ALERT_ENGINE_WORKER_CONCURRENCY_TASKS_PER_WORKER=10
255+
# ALERT_ENGINE_WORKER_CONCURRENCY_LIMIT=10
256+
# ALERT_ENGINE_WORKER_POLL_INTERVAL_MS=1000
257+
# ALERT_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL_MS=100
258+
# ALERT_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS=60000
259+
# ALERT_ENGINE_WORKER_LOG_LEVEL=info
260+
# ALERT_ENGINE_WORKER_DISTRIBUTION_WINDOW_SECONDS=30
261+
# ALERT_ENGINE_WORKER_EVAL_VISIBILITY_TIMEOUT_MS=120000
262+
# ALERT_ENGINE_WORKER_REDIS_HOST= # Defaults to REDIS_HOST (same for the other ALERT_ENGINE_WORKER_REDIS_* vars)
263+
# ALERT_ENGINE_WORKER_REDIS_READER_HOST=
264+
# ALERT_ENGINE_WORKER_REDIS_READER_PORT=
265+
# ALERT_ENGINE_WORKER_REDIS_PORT=
266+
# ALERT_ENGINE_WORKER_REDIS_USERNAME=
267+
# ALERT_ENGINE_WORKER_REDIS_PASSWORD=
268+
# ALERT_ENGINE_WORKER_REDIS_TLS_DISABLED=false
269+
# ALERT_ENGINE_WORKER_REDIS_CLUSTER_MODE_ENABLED=0
270+
201271
# Enable local observability stack (requires `pnpm run docker:full` to bring up otel-collector + prometheus + grafana)
202272
# Uncomment these to send metrics to the local Prometheus via OTEL Collector:
203273
# INTERNAL_OTEL_METRIC_EXPORTER_ENABLED=1
Lines changed: 233 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,233 @@
1+
export type MetricSeriesPoint = { bucket: number; value: number | null; key?: string };
2+
3+
export type LifecycleTimelineEvent = {
4+
type: "raised" | "recovered" | "resolved" | "stopped" | "expired";
5+
at: Date;
6+
};
7+
8+
export type ScopeType = "task" | "queue" | "fingerprint" | "queuePattern" | "all" | "errorGroup";
9+
10+
export function pickThresholdSeries(
11+
seriesMap: Map<string, MetricSeriesPoint[]>,
12+
scopeType: ScopeType
13+
):
14+
| { series: MetricSeriesPoint[]; multi: false }
15+
| { seriesMap: Map<string, MetricSeriesPoint[]>; multi: true } {
16+
const isSingle =
17+
scopeType === "task" ||
18+
scopeType === "queue" ||
19+
scopeType === "fingerprint" ||
20+
scopeType === "errorGroup" ||
21+
scopeType === "all";
22+
if (isSingle) {
23+
return { series: seriesMap.get("__agg__") ?? [], multi: false };
24+
}
25+
return { seriesMap, multi: true };
26+
}
27+
28+
export type QuietWindowBounds = { x1: number; x2: number } | null;
29+
30+
export function quietWindowBounds(
31+
series: MetricSeriesPoint[],
32+
quietWindowSec: number | undefined
33+
): QuietWindowBounds {
34+
if (!quietWindowSec || series.length === 0) return null;
35+
const last = series[series.length - 1]!;
36+
const x2 = last.bucket;
37+
const x1 = Math.max(x2 - quietWindowSec * 1000, series[0]!.bucket);
38+
return { x1, x2 };
39+
}
40+
41+
export type HealthBandRow = {
42+
bucket: number;
43+
degraded: 1 | null;
44+
healthy: 1 | null;
45+
unknown: 1 | null;
46+
};
47+
48+
type HealthFinding = { type: string; severity: string };
49+
50+
export const HEALTH_BUCKET_MS = 4 * 60 * 60 * 1000;
51+
const HEALTH_BAND_DAYS = 7;
52+
53+
function statusFromFindings(facts: Record<string, unknown>): "degraded" | "healthy" {
54+
const findings = Array.isArray(facts?.findings) ? (facts.findings as HealthFinding[]) : [];
55+
return findings.some((f) => f.severity === "warn" || f.severity === "crit")
56+
? "degraded"
57+
: "healthy";
58+
}
59+
60+
export function buildHealthBands(
61+
events: Array<{ at: Date; facts: Record<string, unknown> }>,
62+
now = Date.now(),
63+
startMs = now - HEALTH_BAND_DAYS * 24 * 60 * 60 * 1000,
64+
minBucketMs = HEALTH_BUCKET_MS,
65+
healthySince?: number
66+
): HealthBandRow[] {
67+
const windowMs = now - startMs;
68+
const maxBuckets = (HEALTH_BAND_DAYS * 24 * 60 * 60 * 1000) / HEALTH_BUCKET_MS;
69+
const bucketMs = Math.max(minBucketMs, Math.ceil(windowMs / maxBuckets));
70+
const totalBuckets = Math.ceil(windowMs / bucketMs);
71+
72+
const allSorted = [...events].sort((a, b) => a.at.getTime() - b.at.getTime());
73+
74+
let seedStatus: "degraded" | "healthy" | "unknown" = "unknown";
75+
for (const ev of allSorted) {
76+
if (ev.at.getTime() >= startMs) break;
77+
seedStatus = statusFromFindings(ev.facts);
78+
}
79+
80+
const sorted = allSorted.filter((e) => e.at.getTime() >= startMs);
81+
82+
function statusAt(atMs: number): "degraded" | "healthy" | "unknown" {
83+
let last: "degraded" | "healthy" | "unknown" =
84+
seedStatus === "unknown" && healthySince !== undefined && atMs > healthySince
85+
? "healthy"
86+
: seedStatus;
87+
for (const ev of sorted) {
88+
if (ev.at.getTime() > atMs) break;
89+
last = statusFromFindings(ev.facts);
90+
}
91+
return last;
92+
}
93+
94+
const rows: HealthBandRow[] = [];
95+
for (let i = 0; i < totalBuckets; i++) {
96+
const bucketStart = startMs + i * bucketMs;
97+
const endStatus = statusAt(bucketStart + bucketMs);
98+
// A degraded stretch shorter than a bucket still marks the bucket.
99+
const status = sorted.some(
100+
(ev) =>
101+
ev.at.getTime() > bucketStart &&
102+
ev.at.getTime() <= bucketStart + bucketMs &&
103+
statusFromFindings(ev.facts) === "degraded"
104+
)
105+
? "degraded"
106+
: endStatus;
107+
rows.push({
108+
bucket: bucketStart,
109+
degraded: status === "degraded" ? 1 : null,
110+
healthy: status === "healthy" ? 1 : null,
111+
unknown: status === "unknown" ? 1 : null,
112+
});
113+
}
114+
return rows;
115+
}
116+
117+
export const CLEARING_EVENTS = new Set(["recovered", "resolved", "stopped", "expired"]);
118+
119+
export function healthEventsForBands<T extends { type: string; at: Date; facts: unknown }>(
120+
events: T[]
121+
): Array<{ at: Date; facts: Record<string, unknown> }> {
122+
return events.flatMap((event) => {
123+
const facts = (event.facts ?? {}) as Record<string, unknown>;
124+
if (Array.isArray(facts.findings)) return [{ at: event.at, facts }];
125+
if (event.type === "raised") {
126+
return [{ at: event.at, facts: { findings: [{ type: "raised", severity: "warn" }] } }];
127+
}
128+
if (CLEARING_EVENTS.has(event.type)) return [{ at: event.at, facts: { findings: [] } }];
129+
return [];
130+
});
131+
}
132+
133+
const LIFECYCLE_TYPES = new Set(["raised", "recovered", "resolved", "stopped", "expired"]);
134+
135+
export function mapEventsToTimeline(
136+
events: Array<{ type: string; at: Date }>
137+
): LifecycleTimelineEvent[] {
138+
return events
139+
.filter((e) => LIFECYCLE_TYPES.has(e.type))
140+
.map((e) => ({ type: e.type as LifecycleTimelineEvent["type"], at: e.at }))
141+
.sort((a, b) => a.at.getTime() - b.at.getTime());
142+
}
143+
144+
export type HealthQuerySeries = {
145+
key: string;
146+
label: string;
147+
threshold: number;
148+
points: MetricSeriesPoint[];
149+
};
150+
151+
function splitTopLevelOr(expression: string): string[] {
152+
const parts: string[] = [];
153+
let depth = 0;
154+
let start = 0;
155+
for (let i = 0; i < expression.length; i++) {
156+
const ch = expression[i];
157+
if (ch === "(") depth++;
158+
else if (ch === ")") depth--;
159+
else if (depth === 0 && /^\sOR\s/i.test(expression.slice(i, i + 4))) {
160+
parts.push(expression.slice(start, i));
161+
start = i + 4;
162+
i += 3;
163+
}
164+
}
165+
parts.push(expression.slice(start));
166+
return parts.map((p) => p.trim());
167+
}
168+
169+
function topLevel(expression: string): string {
170+
let depth = 0;
171+
let out = "";
172+
for (const ch of expression) {
173+
if (ch === "(") depth++;
174+
else if (ch === ")") depth--;
175+
else if (depth === 0) out += ch;
176+
}
177+
return out;
178+
}
179+
180+
export type HealthQueryTerm = {
181+
template: string;
182+
paramKey: string;
183+
groupBy: string | null;
184+
lastRun?: boolean;
185+
};
186+
187+
export function healthQueryTerms(template: string): { lines: HealthQueryTerm[] } {
188+
const match = template.match(/^SELECT\s+(.*?)\s+FROM\s+(.*)$/is);
189+
if (!match) return { lines: [] };
190+
const rest = match[2]!;
191+
const groupBy = rest.match(/\s+GROUP\s+BY\s+(\w+)\s*$/i);
192+
const from = groupBy ? rest.slice(0, groupBy.index) : rest;
193+
const lines: HealthQueryTerm[] = [];
194+
for (const term of splitTopLevelOr(match[1]!)) {
195+
const cmp = term.match(/^(.+?)\s*>\s*\{\{p\.(\w+)\}\}$/s);
196+
const lhs = cmp?.[1]?.trim();
197+
const lastRun = lhs?.match(/^dateDiff\('second',\s*max\((\w+)\),\s*now\(\)\)$/i);
198+
if (lastRun) {
199+
lines.push({
200+
template: `SELECT toUnixTimestamp(max(${lastRun[1]})) FROM ${from}`,
201+
paramKey: cmp![2]!,
202+
groupBy: null,
203+
lastRun: true,
204+
});
205+
continue;
206+
}
207+
// Sums grow with bucket width and now()-relative ages drift, so neither plots per bucket.
208+
if (
209+
!lhs ||
210+
/[<>=]|\bAND\b|\bOR\b/i.test(topLevel(lhs)) ||
211+
/^sum\(/i.test(lhs) ||
212+
/\bnow\(\)/i.test(lhs)
213+
) {
214+
continue;
215+
}
216+
lines.push({
217+
template: `SELECT ${lhs} FROM ${from}`,
218+
paramKey: cmp![2]!,
219+
groupBy: groupBy?.[1] ?? null,
220+
});
221+
}
222+
return { lines };
223+
}
224+
225+
export function percentOfThreshold(
226+
points: MetricSeriesPoint[],
227+
threshold: number
228+
): MetricSeriesPoint[] {
229+
return points.map((p) => ({
230+
...p,
231+
value: p.value === null ? null : (p.value / threshold) * 100,
232+
}));
233+
}

‎apps/webapp/app/db.server.ts‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -173,7 +173,8 @@ type DatasourceLabel =
173173
| "run-ops-replica"
174174
| "webhook-writer"
175175
| "webhook-partitions"
176-
| "webhook-replica";
176+
| "webhook-replica"
177+
| "alert-partitions";
177178

178179
function tagDatasource<T extends PrismaClient>(datasource: DatasourceLabel, client: T): T {
179180
return client.$extends({
@@ -279,6 +280,20 @@ export const webhookPartitionPrisma: WebhookDatabase = singleton("webhookPartiti
279280
);
280281
});
281282

283+
export function createAlertPartitionPrisma(): PrismaClient {
284+
return captureInfrastructureErrors(
285+
tagDatasource(
286+
"alert-partitions",
287+
buildWriterClient({
288+
url: env.DATABASE_URL,
289+
clientType: "alert-partitions",
290+
connectionLimit: 2,
291+
eagerConnect: false,
292+
})
293+
)
294+
);
295+
}
296+
282297
/**
283298
* Webhook reader chain: an explicit webhook replica, else the webhook writer once split (no
284299
* separate replica yet), else the main $replica when the feature is not split.

‎apps/webapp/app/entry.server.tsx‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
1111
import { initSnapshotRecoveryWorker } from "~/v3/snapshotRecoveryWorker.server";
1212
import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.server";
1313
import { initBillingLimitWorker } from "~/v3/billingLimitWorker.server";
14+
import { initAlertEngineWorker } from "~/v3/alertEngineWorker.server";
1415
import { initLogsSearchProjectorWorker } from "~/v3/logsSearchProjectorWorker.server";
1516
import { initQueueMetricsConsumer, initQueueMetricsEmitter } from "~/v3/queueMetrics.server";
1617
import { initApiRateLimitMetrics } from "~/services/apiRateLimitMetrics.server";
@@ -281,6 +282,7 @@ initMollifierDrainerWorker();
281282
initSnapshotRecoveryWorker();
282283
initMollifierStaleSweepWorker();
283284
initBillingLimitWorker();
285+
initAlertEngineWorker();
284286
initLogsSearchProjectorWorker();
285287
initQueueMetricsEmitter();
286288
initQueueMetricsConsumer();

0 commit comments

Comments
 (0)