diff --git a/.changeset/resend-failed-alert-resolves.md b/.changeset/resend-failed-alert-resolves.md new file mode 100644 index 0000000000..cc40f56d74 --- /dev/null +++ b/.changeset/resend-failed-alert-resolves.md @@ -0,0 +1,8 @@ +--- +'@hyperdx/api': patch +--- + +fix: a resolve notification whose webhook delivery fails is now sent again on +later checks, backing off up to every 30 minutes, for up to 24 hours. A resolve +that is never delivered is recorded on the alert's evaluation history. Webhook +delivery failures now log the underlying cause instead of only "fetch failed". diff --git a/packages/api/src/models/pendingNotification.ts b/packages/api/src/models/pendingNotification.ts new file mode 100644 index 0000000000..c23a4b24df --- /dev/null +++ b/packages/api/src/models/pendingNotification.ts @@ -0,0 +1,57 @@ +import mongoose, { Schema } from 'mongoose'; + +import type { ObjectId } from '.'; + +/** + * An alert notification whose delivery failed and that the check-alerts task + * sends again on later ticks. Only resolves are stored: a firing alert sends + * ALERT again on every evaluation, but a resolve is sent once. + */ +export interface IPendingNotification { + /** Idempotency key of the notification: hash of alert, channel and group. */ + eventId: string; + team: ObjectId; + alert: ObjectId; + /** Group key of a group-by alert; '' for an alert without groups. */ + group: string; + webhook: ObjectId; + /** + * The rendered message as JSON. A string because group-by column names can + * contain dots, which Mongo would read as nested paths in a key. + */ + message: string; + /** Evaluation window whose ERROR history row records the failure. */ + windowStart: Date; + firstFailedAt: Date; + /** Failed deliveries so far, including the original one. */ + attempts: number; + nextAttemptAt: Date; + /** Removes records nothing ever cleans up, e.g. for a deleted alert. */ + expireAt: Date; +} + +const PendingNotificationSchema = new Schema( + { + eventId: { type: String, required: true }, + team: { type: Schema.Types.ObjectId, ref: 'Team', required: true }, + alert: { type: Schema.Types.ObjectId, ref: 'Alert', required: true }, + group: { type: String, default: '' }, + webhook: { type: Schema.Types.ObjectId, ref: 'Webhook', required: true }, + message: { type: String, required: true }, + windowStart: { type: Date, required: true }, + firstFailedAt: { type: Date, required: true }, + attempts: { type: Number, required: true }, + nextAttemptAt: { type: Date, required: true }, + expireAt: { type: Date, required: true }, + }, + { timestamps: true }, +); + +PendingNotificationSchema.index({ eventId: 1 }, { unique: true }); +PendingNotificationSchema.index({ alert: 1 }); +PendingNotificationSchema.index({ expireAt: 1 }, { expireAfterSeconds: 0 }); + +export default mongoose.model( + 'PendingNotification', + PendingNotificationSchema, +); diff --git a/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts index d3d0bbdc3b..1105ca65f3 100644 --- a/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts +++ b/packages/api/src/tasks/checkAlerts/__tests__/checkAlerts.int.test.ts @@ -31,6 +31,7 @@ import Alert, { AlertSource, IAlert } from '@/models/alert'; import AlertHistory from '@/models/alertHistory'; import Connection, { IConnection } from '@/models/connection'; import Dashboard, { IDashboard } from '@/models/dashboard'; +import PendingNotification from '@/models/pendingNotification'; import { ISavedSearch, SavedSearch } from '@/models/savedSearch'; import { ISource, LogSource, Source } from '@/models/source'; import { ITeam } from '@/models/team'; @@ -4483,6 +4484,10 @@ describe('checkAlerts', () => { expect(updated!.executionErrors![0].type).toBe( AlertErrorType.WEBHOOK_ERROR, ); + // Stored so a later tick sends the resolve again. + expect( + await PendingNotification.countDocuments({ alert: details.alert.id }), + ).toBe(1); }); it('clears errors after a successful execution', async () => { diff --git a/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts index 5b5d348898..46eb4793da 100644 --- a/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts +++ b/packages/api/src/tasks/checkAlerts/__tests__/checkAlertsTask.test.ts @@ -4,10 +4,12 @@ import mongoose from 'mongoose'; import * as config from '@/config'; import { ObjectId } from '@/models'; import { AlertSource, AlertThresholdType, IAlert } from '@/models/alert'; +import { IPendingNotification } from '@/models/pendingNotification'; import { ISource } from '@/models/source'; import { IWebhook } from '@/models/webhook'; import * as checkAlerts from '@/tasks/checkAlerts'; import CheckAlertTask from '@/tasks/checkAlerts'; +import * as pendingNotifications from '@/tasks/checkAlerts/pendingNotifications'; import { AlertDetails, AlertProvider, @@ -38,6 +40,11 @@ describe('CheckAlertTask', () => { getWebhooks: jest.fn(), updateAlertState: jest.fn(), recordAlertErrors: jest.fn(), + appendAlertErrorHistory: jest.fn(), + getPendingNotifications: jest.fn().mockResolvedValue([]), + savePendingNotifications: jest.fn(), + updatePendingNotification: jest.fn(), + deletePendingNotification: jest.fn(), asyncDispose: jest.fn(), buildChartLink: jest.fn(), buildChartExplorerLink: jest.fn(), @@ -173,6 +180,88 @@ describe('CheckAlertTask', () => { mockProcessAlert.mockRestore(); }); + const makeTaskForAlert = (alertId: string): AlertTask => ({ + alerts: [ + { + alert: { id: alertId } as IAlert, + taskType: AlertTaskType.SAVED_SEARCH, + previousMap: new Map(), + } as AlertDetails, + ], + conn: { + id: 'conn-123', + _id: new mongoose.Types.ObjectId(), + host: config.CLICKHOUSE_HOST, + username: config.CLICKHOUSE_USER, + password: config.CLICKHOUSE_PASSWORD, + name: '', + team: new mongoose.Types.ObjectId(), + }, + now: new Date(), + }); + + it("sends an alert's pending notifications again before evaluating it", async () => { + const task = new CheckAlertTask({ taskName: TaskName.CHECK_ALERTS }); + const alertId = new mongoose.Types.ObjectId().toString(); + const alertTask = makeTaskForAlert(alertId); + const record: IPendingNotification = { + eventId: 'event-1', + team: alertTask.conn.team, + alert: new mongoose.Types.ObjectId(alertId), + group: '', + webhook: new mongoose.Types.ObjectId(), + message: '{}', + windowStart: alertTask.now, + firstFailedAt: alertTask.now, + attempts: 1, + nextAttemptAt: alertTask.now, + expireAt: alertTask.now, + }; + mockAlertProvider.getAlertTasks.mockResolvedValue([alertTask]); + mockAlertProvider.getWebhooks.mockResolvedValue(new Map()); + mockAlertProvider.getPendingNotifications.mockResolvedValue([record]); + + const calls: string[] = []; + const resendSpy = jest + .spyOn(pendingNotifications, 'resendPendingNotifications') + .mockImplementation(async () => { + calls.push('resend'); + }); + mockProcessAlert.mockImplementation(async () => { + calls.push('process'); + }); + + await task.execute(); + + expect(mockAlertProvider.getPendingNotifications).toHaveBeenCalledWith([ + alertId, + ]); + expect(resendSpy).toHaveBeenCalledWith( + expect.objectContaining({ + alert: alertTask.alerts[0].alert, + pending: [record], + now: alertTask.now, + }), + ); + expect(calls).toEqual(['resend', 'process']); + resendSpy.mockRestore(); + }); + + it('still evaluates alerts when pending notifications cannot be loaded', async () => { + const task = new CheckAlertTask({ taskName: TaskName.CHECK_ALERTS }); + mockAlertProvider.getAlertTasks.mockResolvedValue([ + makeTaskForAlert('alert-123'), + ]); + mockAlertProvider.getWebhooks.mockResolvedValue(new Map()); + mockAlertProvider.getPendingNotifications.mockRejectedValue( + new Error('mongo down'), + ); + + await task.execute(); + + expect(mockProcessAlert).toHaveBeenCalledTimes(1); + }); + it("should ensure that the correct team's webhooks are passed to processAlert", async () => { const args: CheckAlertsTaskArgs = { taskName: TaskName.CHECK_ALERTS }; const task = new CheckAlertTask(args); diff --git a/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts new file mode 100644 index 0000000000..e5cee24d9a --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.int.test.ts @@ -0,0 +1,327 @@ +import { ClickhouseClient } from '@hyperdx/common-utils/dist/clickhouse/node'; +import { AlertErrorType } from '@hyperdx/common-utils/dist/types'; +import mongoose from 'mongoose'; +import ms from 'ms'; + +import { createAlert } from '@/controllers/alerts'; +import { createTeam } from '@/controllers/team'; +import { bulkInsertLogs, getServer } from '@/fixtures'; +import Alert, { + AlertSource, + AlertState, + AlertThresholdType, +} from '@/models/alert'; +import AlertHistory from '@/models/alertHistory'; +import PendingNotification from '@/models/pendingNotification'; +import Webhook, { IWebhook } from '@/models/webhook'; +import { getPreviousAlertHistories, processAlert } from '@/tasks/checkAlerts'; +import { resendPendingNotifications } from '@/tasks/checkAlerts/pendingNotifications'; +import { + AlertProvider, + AlertTaskType, + loadProvider, +} from '@/tasks/checkAlerts/providers'; + +import { + HOOK_BAD, + HOOK_OK, + makeGenericWebhook, + setupAlertFixtures, +} from './multiChannelAlerts.harness'; + +const FIRING_AT = new Date('2023-11-16T22:12:00.000Z'); +const RESOLVING_AT = new Date('2023-11-16T22:17:00.000Z'); +// The 5m window RESOLVING_AT evaluates, where its history rows are written. +const RESOLVED_WINDOW = new Date('2023-11-16T22:15:00.000Z'); +const minutesAfterResolve = (minutes: number) => + new Date(RESOLVING_AT.getTime() + ms(`${minutes}m`)); + +type Request = { url: string; idempotencyKey?: string; body: string }; + +describe('sending failed resolves again', () => { + let alertProvider: AlertProvider; + let server: ReturnType; + let requests: Request[]; + // Endpoints that currently fail every request. + let down: Set; + + beforeAll(async () => { + alertProvider = await loadProvider(); + server = getServer(); + await server.start(); + }); + + beforeEach(() => { + requests = []; + down = new Set(); + jest.mocked(global.fetch).mockImplementation(async (input, init) => { + const url = String(input); + requests.push({ + url, + idempotencyKey: + new Headers(init?.headers).get('Idempotency-Key') ?? undefined, + body: String(init?.body), + }); + if (down.has(url)) { + // 400 so withRetry does not retry within one delivery. + return new Response('unavailable', { status: 400 }); + } + return new Response('ok', { status: 200 }); + }); + }); + + afterEach(async () => { + await server.clearDBs(); + jest.clearAllMocks(); + }); + + afterAll(async () => { + await server.stop(); + }); + + const setup = async ( + webhookUrls: string[] = [HOOK_OK], + ): Promise<{ + alertId: string; + webhooks: IWebhook[]; + evaluate: (now: Date) => Promise; + resend: (now: Date, webhooks?: IWebhook[]) => Promise; + }> => { + const { team, connection, source, savedSearch } = + await setupAlertFixtures(createTeam); + const webhooks = await Promise.all( + webhookUrls.map((url, i) => + makeGenericWebhook(team._id, `Hook ${i}`, url), + ), + ); + const alert = await createAlert( + team._id, + { + source: AlertSource.SAVED_SEARCH, + channels: webhooks.map(w => ({ + type: 'webhook' as const, + webhookId: w._id.toString(), + })), + interval: '5m', + thresholdType: AlertThresholdType.ABOVE, + threshold: 1, + savedSearchId: savedSearch.id, + name: 'Resend Alert', + }, + new mongoose.Types.ObjectId(), + ); + const clickhouseClient = new ClickhouseClient({ + host: connection.host, + username: connection.username, + password: connection.password, + }); + const webhooksById = (list: IWebhook[]) => + new Map(list.map(w => [w._id.toString(), w])); + + const evaluate = async (now: Date) => { + const enhancedAlert: any = await Alert.findById(alert.id).populate([ + 'team', + 'savedSearch', + ]); + await processAlert( + now, + { + alert: enhancedAlert, + source, + taskType: AlertTaskType.SAVED_SEARCH, + savedSearch, + previousMap: await getPreviousAlertHistories([alert.id], now), + }, + clickhouseClient, + connection.id, + alertProvider, + webhooksById(webhooks), + ); + }; + + const resend = async (now: Date, available: IWebhook[] = webhooks) => { + const previousMap = await getPreviousAlertHistories([alert.id], now); + await resendPendingNotifications({ + alert: (await Alert.findById(alert.id))!, + // Non-grouped alerts key their history by the bare alert id. + getPreviousState: () => previousMap.get(alert.id)?.state, + now, + pending: await alertProvider.getPendingNotifications([alert.id]), + provider: alertProvider, + teamWebhooksById: webhooksById(available), + }); + }; + + return { alertId: alert.id, webhooks, evaluate, resend }; + }; + + const seedErrorLogs = (at: Date) => + bulkInsertLogs( + [0, 1].map(() => ({ + ServiceName: 'api', + Timestamp: at, + SeverityText: 'error', + Body: 'Test error message', + })), + ); + + // Fires at FIRING_AT, then resolves at RESOLVING_AT while `failing` is down. + const fireThenFailResolve = async ( + ctx: Awaited>, + failing: string[] = [HOOK_OK], + ) => { + await seedErrorLogs(new Date(FIRING_AT.getTime() - ms('3m'))); + await ctx.evaluate(FIRING_AT); + failing.forEach(url => down.add(url)); + await ctx.evaluate(RESOLVING_AT); + requests = []; + }; + + it('stores a resolve whose delivery failed', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + + const alert = await Alert.findById(ctx.alertId); + expect(alert!.state).toBe(AlertState.OK); + expect(alert!.executionErrors?.[0]?.type).toBe( + AlertErrorType.WEBHOOK_ERROR, + ); + const records = await PendingNotification.find({ alert: ctx.alertId }); + expect(records).toHaveLength(1); + expect(records[0]).toMatchObject({ + group: '', + attempts: 1, + windowStart: RESOLVED_WINDOW, + firstFailedAt: RESOLVING_AT, + nextAttemptAt: minutesAfterResolve(1), + }); + expect(records[0].webhook.toString()).toBe(ctx.webhooks[0]._id.toString()); + expect(JSON.parse(records[0].message).state).toBe(AlertState.OK); + }); + + it('delivers the resolve on a later tick with the same idempotency key', async () => { + const ctx = await setup(); + await seedErrorLogs(new Date(FIRING_AT.getTime() - ms('3m'))); + await ctx.evaluate(FIRING_AT); + down.add(HOOK_OK); + await ctx.evaluate(RESOLVING_AT); + const failedResolve = requests[requests.length - 1]; + + requests = []; + down.clear(); + await ctx.resend(minutesAfterResolve(1)); + + expect(requests).toHaveLength(1); + expect(requests[0].url).toBe(HOOK_OK); + expect(requests[0].idempotencyKey).toBe(failedResolve.idempotencyKey); + expect(requests[0].body).toBe(failedResolve.body); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('backs off after another failure', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + + await ctx.resend(minutesAfterResolve(1)); + expect(requests).toHaveLength(1); + const record = await PendingNotification.findOne({ alert: ctx.alertId }); + expect(record).toMatchObject({ + attempts: 2, + nextAttemptAt: minutesAfterResolve(3), + }); + + requests = []; + await ctx.resend(minutesAfterResolve(2)); + expect(requests).toHaveLength(0); + }); + + it('only stores the channels that failed', async () => { + const ctx = await setup([HOOK_OK, HOOK_BAD]); + await fireThenFailResolve(ctx, [HOOK_BAD]); + + const records = await PendingNotification.find({ alert: ctx.alertId }); + expect(records.map(r => r.webhook.toString())).toEqual([ + ctx.webhooks[1]._id.toString(), + ]); + + await ctx.resend(minutesAfterResolve(1)); + expect(requests.map(r => r.url)).toEqual([HOOK_BAD]); + }); + + it('does not store a failed ALERT, which the next evaluation sends again', async () => { + const ctx = await setup(); + await seedErrorLogs(new Date(FIRING_AT.getTime() - ms('3m'))); + down.add(HOOK_OK); + await ctx.evaluate(FIRING_AT); + + expect((await Alert.findById(ctx.alertId))!.state).toBe(AlertState.ALERT); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('drops the resolve once the group fires again', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + down.clear(); + await seedErrorLogs(minutesAfterResolve(1)); + await ctx.evaluate(minutesAfterResolve(5)); + requests = []; + + await ctx.resend(minutesAfterResolve(6)); + + expect(requests).toHaveLength(0); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('records an error on the resolved window and stops after 24 hours', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + + await ctx.resend(new Date(RESOLVING_AT.getTime() + ms('24h'))); + + expect(requests).toHaveLength(0); + expect(await PendingNotification.countDocuments()).toBe(0); + const errorRow = await AlertHistory.findOne({ + alert: ctx.alertId, + createdAt: RESOLVED_WINDOW, + state: AlertState.ERROR, + }); + expect(errorRow!.errors).toHaveLength(2); + expect(errorRow!.errors![1]).toMatchObject({ + type: AlertErrorType.WEBHOOK_ERROR, + message: expect.stringContaining( + 'Resolve notification to webhook "Hook 0" was not delivered within 1 day (1 attempt)', + ), + }); + }); + + it('drops the resolve when its webhook was deleted', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + await Webhook.deleteOne({ _id: ctx.webhooks[0]._id }); + + await ctx.resend(minutesAfterResolve(1), []); + + expect(requests).toHaveLength(0); + expect(await PendingNotification.countDocuments()).toBe(0); + }); + + it('holds the resolve while the alert is silenced', async () => { + const ctx = await setup(); + await fireThenFailResolve(ctx); + await Alert.updateOne( + { _id: ctx.alertId }, + { + $set: { + silenced: { at: RESOLVING_AT, until: minutesAfterResolve(60) }, + }, + }, + ); + + await ctx.resend(minutesAfterResolve(1)); + + expect(requests).toHaveLength(0); + expect( + await PendingNotification.findOne({ alert: ctx.alertId }), + ).toMatchObject({ attempts: 1 }); + }); +}); diff --git a/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts new file mode 100644 index 0000000000..245ce426c3 --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/__tests__/pendingNotifications.test.ts @@ -0,0 +1,16 @@ +import { getNextAttemptAt } from '@/tasks/checkAlerts/pendingNotifications'; + +describe('getNextAttemptAt', () => { + const now = new Date('2026-11-13T12:00:00.000Z'); + const minutesAfterNow = (attempts: number) => + (getNextAttemptAt(now, attempts).getTime() - now.getTime()) / 60_000; + + it('waits 1 minute after the first failure and doubles after each one', () => { + expect([1, 2, 3, 4, 5].map(minutesAfterNow)).toEqual([1, 2, 4, 8, 16]); + }); + + it('waits at most 30 minutes', () => { + expect(minutesAfterNow(6)).toBe(30); + expect(minutesAfterNow(50)).toBe(30); + }); +}); diff --git a/packages/api/src/tasks/checkAlerts/index.ts b/packages/api/src/tasks/checkAlerts/index.ts index d34af3dd85..0633918b8a 100644 --- a/packages/api/src/tasks/checkAlerts/index.ts +++ b/packages/api/src/tasks/checkAlerts/index.ts @@ -67,6 +67,7 @@ import AlertHistory, { IAlertHistoryAnalytics, } from '@/models/alertHistory'; import { IDashboard } from '@/models/dashboard'; +import { IPendingNotification } from '@/models/pendingNotification'; import { ISavedSearch } from '@/models/savedSearch'; import { ISource } from '@/models/source'; import { IWebhook } from '@/models/webhook'; @@ -85,6 +86,10 @@ import { findAlertEvaluator, THRESHOLD_EVALUATOR_NAME, } from '@/tasks/checkAlerts/evaluators'; +import { + resendPendingNotifications, + toPendingNotification, +} from '@/tasks/checkAlerts/pendingNotifications'; import { AlertDetails, AlertProvider, @@ -110,6 +115,7 @@ import { unflattenObject, } from '@/tasks/util'; import { alertConfigHasGroupBy, isPopulatedRef } from '@/utils/alerts'; +import { serializeErrorWithCause } from '@/utils/errors'; import { getActiveTraceId, getCounter, @@ -1286,6 +1292,35 @@ export const processAlert = async ( } : undefined; + // A firing alert sends ALERT again on every evaluation, but a resolve is + // sent once, so a resolve that failed to deliver is stored and sent again + // on later ticks (see resendPendingNotifications). Failures before + // dispatch (deleted webhook, notification cap) carry no job: sending them + // again cannot succeed. + const savePendingResolves = async (failures: NotificationFailure[]) => { + const records = failures.flatMap(failure => { + const record = + failure.job && + toPendingNotification(failure.job, nowInMinsRoundDown, wallClockNow); + return record ? [record] : []; + }); + if (records.length === 0) { + return; + } + try { + await alertProvider.savePendingNotifications(records); + } catch (e) { + logger.error( + { + alertId: alert.id, + eventIds: records.map(r => r.eventId), + error: serializeError(e), + }, + 'Failed to store undelivered alert resolve', + ); + } + }; + // Helper to send a notification, catching and logging any errors. // Resolves true when a dispatch was attempted (even if delivery later // fails — failures surface via executionErrors, mirroring the threshold @@ -1375,12 +1410,15 @@ export const processAlert = async ( alertId: alert.id, group, target: failure.target, - error: serializeError(failure.error), + error: serializeErrorWithCause(failure.error), }, 'Notification target could not be dispatched', ); executionErrors.push(makeNotificationAlertError(failure)); } + if (state === AlertState.OK) { + await savePendingResolves(failures); + } } catch (e) { // Render-level failures (title/link building, template compile) — // nothing was dispatched. @@ -2123,6 +2161,7 @@ export default class CheckAlertTask implements HdxTask { async processAlertTask( alertTask: AlertTask, teamWebhooksById: Map, + pendingByAlert: Map = new Map(), ): Promise { return tasksTracer.startActiveSpan('processAlertTask', async span => { setBusinessContext({ teamId: alertTask.conn.team.toString() }); @@ -2155,6 +2194,20 @@ export default class CheckAlertTask implements HdxTask { 'processAlert', async () => { setBusinessContext({ teamId: conn.team.toString() }); + const pending = pendingByAlert.get(alert.alert.id); + if (pending != null && pending.length > 0) { + await resendPendingNotifications({ + alert: alert.alert, + getPreviousState: group => + alert.previousMap.get( + computeHistoryMapKey(alert.alert.id, group), + )?.state, + now: alertTask.now, + pending, + provider: this.provider, + teamWebhooksById, + }); + } // Set once here, so the alert's own query, the EXPLAIN // probes and the notification's sample rows all get tagged. await withQueryAttribution( @@ -2204,6 +2257,34 @@ export default class CheckAlertTask implements HdxTask { }); } + /** + * Failed resolves to send again this tick, by alert id. A failed lookup only + * delays them to the next tick, so it must not stop evaluation. + */ + private async getPendingNotificationsByAlert( + alertIds: string[], + ): Promise> { + const byAlert = new Map(); + try { + const records = await this.provider.getPendingNotifications(alertIds); + for (const record of records) { + const alertId = record.alert.toString(); + const existing = byAlert.get(alertId); + if (existing) { + existing.push(record); + } else { + byAlert.set(alertId, [record]); + } + } + } catch (e) { + logger.error( + { error: serializeError(e) }, + 'Failed to load pending alert notifications', + ); + } + return byAlert; + } + async execute(): Promise { if (this.args.taskName !== 'check-alerts') { throw new Error( @@ -2246,12 +2327,20 @@ export default class CheckAlertTask implements HdxTask { `Obtained teams and webhooks for all alertTasks`, ); + const pendingByAlert = await this.getPendingNotificationsByAlert( + alertTasks.flatMap(task => task.alerts.map(details => details.alert.id)), + ); + let failedBatchCount = 0; for (const task of alertTasks) { const teamWebhooksById = teamToWebhooks.get(task.conn.team.toString()) ?? new Map(); this.task_queue.add(async () => { - const success = await this.processAlertTask(task, teamWebhooksById); + const success = await this.processAlertTask( + task, + teamWebhooksById, + pendingByAlert, + ); if (!success) { failedBatchCount++; } diff --git a/packages/api/src/tasks/checkAlerts/pendingNotifications.ts b/packages/api/src/tasks/checkAlerts/pendingNotifications.ts new file mode 100644 index 0000000000..6ac370a3f9 --- /dev/null +++ b/packages/api/src/tasks/checkAlerts/pendingNotifications.ts @@ -0,0 +1,229 @@ +import { AlertErrorType } from '@hyperdx/common-utils/dist/types'; +import mongoose from 'mongoose'; +import ms from 'ms'; + +import { AlertState, IAlert } from '@/models/alert'; +import { IPendingNotification } from '@/models/pendingNotification'; +import { IWebhook } from '@/models/webhook'; +import { + deliverNotification, + NotificationJob, +} from '@/tasks/checkAlerts/notifications'; +import type { AlertProvider } from '@/tasks/checkAlerts/providers'; +import type { Message } from '@/tasks/checkAlerts/transports'; +import { serializeErrorWithCause } from '@/utils/errors'; +import { getCounter } from '@/utils/instrumentation'; +import logger from '@/utils/logger'; + +/** How long a failed resolve keeps being sent again. */ +export const RESEND_WINDOW_MS = ms('24h'); +const BASE_BACKOFF_MS = ms('1m'); +const MAX_BACKOFF_MS = ms('30m'); +// Longer than the re-send window, so the TTL only removes records the task +// never gets to, such as those of a deleted alert. +const RECORD_TTL_MS = RESEND_WINDOW_MS + ms('7d'); + +export type ResendOutcome = + | 'delivered' + | 'failed' + | 'expired' + | 'superseded' + | 'webhook_missing' + | 'skipped_silenced'; + +const notificationResendsCounter = getCounter( + 'hyperdx.alerts.notification_resends', + { + description: + 'Count of attempts to send a failed alert resolve again, labeled by outcome (delivered, failed, expired, superseded, webhook_missing, skipped_silenced).', + }, +); + +/** 1m after the first failure, doubling up to 30m. */ +export const getNextAttemptAt = (now: Date, attempts: number): Date => + new Date( + now.getTime() + + Math.min( + BASE_BACKOFF_MS * 2 ** Math.max(0, attempts - 1), + MAX_BACKOFF_MS, + ), + ); + +/** + * The record to store for a resolve whose delivery failed, or undefined when + * the job cannot be sent again from a record (it is not a webhook delivery, or + * it has no alert or team). + */ +export const toPendingNotification = ( + job: NotificationJob, + windowStart: Date, + now: Date, +): IPendingNotification | undefined => { + if ( + job.populatedChannel.type !== 'webhook' || + job.alertId == null || + job.teamId == null + ) { + return undefined; + } + return { + eventId: job.eventId, + team: new mongoose.Types.ObjectId(job.teamId), + alert: new mongoose.Types.ObjectId(job.alertId), + group: job.group ?? '', + webhook: new mongoose.Types.ObjectId( + job.populatedChannel.channel._id.toString(), + ), + message: JSON.stringify(job.message), + windowStart, + firstFailedAt: now, + attempts: 1, + nextAttemptAt: getNextAttemptAt(now, 1), + expireAt: new Date(now.getTime() + RECORD_TTL_MS), + }; +}; + +const resendOne = async ({ + alert, + getPreviousState, + now, + provider, + record, + teamWebhooksById, +}: { + alert: IAlert; + getPreviousState: (group: string) => AlertState | undefined; + now: Date; + provider: AlertProvider; + record: IPendingNotification; + teamWebhooksById: Map; +}): Promise => { + // The group fired again after the failed resolve, and that ALERT has gone + // out. A late resolve would close the new incident. + if (getPreviousState(record.group) === AlertState.ALERT) { + await provider.deletePendingNotification(record.eventId); + return 'superseded'; + } + + const webhook = teamWebhooksById.get(record.webhook.toString()); + if (webhook == null) { + await provider.deletePendingNotification(record.eventId); + return 'webhook_missing'; + } + + if (now.getTime() - record.firstFailedAt.getTime() >= RESEND_WINDOW_MS) { + await provider.appendAlertErrorHistory(alert.id, record.windowStart, { + timestamp: now, + type: AlertErrorType.WEBHOOK_ERROR, + message: `Resolve notification to webhook "${webhook.name}" was not delivered within ${ms(RESEND_WINDOW_MS, { long: true })} (${record.attempts} ${record.attempts === 1 ? 'attempt' : 'attempts'}). The incident may still be open in the receiving tool.`, + }); + await provider.deletePendingNotification(record.eventId); + return 'expired'; + } + + if (record.nextAttemptAt.getTime() > now.getTime()) { + return undefined; + } + + if ((alert.silenced?.until?.getTime() ?? 0) > now.getTime()) { + return 'skipped_silenced'; + } + + const message: Message = JSON.parse(record.message); + try { + await deliverNotification({ + eventId: record.eventId, + alertId: alert.id, + teamId: record.team.toString(), + group: record.group, + populatedChannel: { type: 'webhook', channel: webhook }, + message, + }); + } catch (e) { + const attempts = record.attempts + 1; + logger.warn( + { + alertId: alert.id, + eventId: record.eventId, + webhookId: webhook._id.toString(), + attempts, + error: serializeErrorWithCause(e), + }, + 'Failed to send alert resolve again', + ); + await provider.updatePendingNotification(record.eventId, { + attempts, + nextAttemptAt: getNextAttemptAt(now, attempts), + }); + return 'failed'; + } + await provider.deletePendingNotification(record.eventId); + logger.info( + { + alertId: alert.id, + eventId: record.eventId, + webhookId: webhook._id.toString(), + attempts: record.attempts + 1, + }, + 'Sent alert resolve again after an earlier delivery failed', + ); + return 'delivered'; +}; + +/** + * Send the alert's failed resolves again, ahead of its evaluation so a resolve + * still reaches the receiver before any ALERT this evaluation sends. Never + * throws: a failure here must not stop the evaluation. + */ +export const resendPendingNotifications = async ({ + alert, + getPreviousState, + now, + pending, + provider, + teamWebhooksById, +}: { + alert: IAlert; + /** The state of the group's latest history row. */ + getPreviousState: (group: string) => AlertState | undefined; + now: Date; + pending: IPendingNotification[]; + provider: AlertProvider; + teamWebhooksById: Map; +}): Promise => { + for (const record of pending) { + try { + const outcome = await resendOne({ + alert, + getPreviousState, + now, + provider, + record, + teamWebhooksById, + }); + if (outcome != null) { + notificationResendsCounter.add(1, { outcome }); + } + if (outcome === 'expired') { + logger.error( + { + alertId: alert.id, + eventId: record.eventId, + webhookId: record.webhook.toString(), + attempts: record.attempts, + }, + 'Gave up sending alert resolve again', + ); + } + } catch (e) { + logger.error( + { + alertId: alert.id, + eventId: record.eventId, + error: serializeErrorWithCause(e), + }, + 'Failed to process pending alert notification', + ); + } + } +}; diff --git a/packages/api/src/tasks/checkAlerts/providers/default.ts b/packages/api/src/tasks/checkAlerts/providers/default.ts index b409fafe98..ebcc47c3b6 100644 --- a/packages/api/src/tasks/checkAlerts/providers/default.ts +++ b/packages/api/src/tasks/checkAlerts/providers/default.ts @@ -27,6 +27,9 @@ import AlertHistory, { } from '@/models/alertHistory'; import Connection, { IConnection } from '@/models/connection'; import Dashboard from '@/models/dashboard'; +import PendingNotification, { + IPendingNotification, +} from '@/models/pendingNotification'; import { type ISavedSearch, SavedSearch } from '@/models/savedSearch'; import { type ISource, Source } from '@/models/source'; import Webhook, { IWebhook } from '@/models/webhook'; @@ -672,6 +675,61 @@ export default class DefaultAlertProvider implements AlertProvider { ); } + async appendAlertErrorHistory( + alertId: string, + evaluationWindowStart: Date, + error: IAlertError, + ) { + await AlertHistory.updateOne( + { + alert: new mongoose.Types.ObjectId(alertId), + createdAt: evaluationWindowStart, + state: AlertState.ERROR, + }, + { + $push: { errors: error }, + $setOnInsert: { counts: 0, lastValues: [] }, + }, + { upsert: true }, + ); + } + + async getPendingNotifications(alertIds: string[]) { + if (alertIds.length === 0) { + return []; + } + return PendingNotification.find({ + alert: { $in: alertIds.map(id => new mongoose.Types.ObjectId(id)) }, + }).lean(); + } + + async savePendingNotifications(records: IPendingNotification[]) { + if (records.length === 0) { + return; + } + await PendingNotification.bulkWrite( + records.map(record => ({ + replaceOne: { + filter: { eventId: record.eventId }, + replacement: record, + upsert: true, + }, + })), + { ordered: false }, + ); + } + + async updatePendingNotification( + eventId: string, + update: Pick, + ) { + await PendingNotification.updateOne({ eventId }, { $set: update }); + } + + async deletePendingNotification(eventId: string) { + await PendingNotification.deleteOne({ eventId }); + } + async getWebhooks(teamId: string | ObjectId) { const webhooks = await Webhook.find({ team: new mongoose.Types.ObjectId(teamId), diff --git a/packages/api/src/tasks/checkAlerts/providers/index.ts b/packages/api/src/tasks/checkAlerts/providers/index.ts index 96071696ed..151d472e80 100644 --- a/packages/api/src/tasks/checkAlerts/providers/index.ts +++ b/packages/api/src/tasks/checkAlerts/providers/index.ts @@ -7,6 +7,7 @@ import { IAlert, IAlertError } from '@/models/alert'; import { IAlertHistory, IAlertHistoryAnalytics } from '@/models/alertHistory'; import { IConnection } from '@/models/connection'; import { IDashboard } from '@/models/dashboard'; +import { IPendingNotification } from '@/models/pendingNotification'; import { ISavedSearch } from '@/models/savedSearch'; import { ISource } from '@/models/source'; import { IWebhook } from '@/models/webhook'; @@ -142,6 +143,31 @@ export interface AlertProvider { analytics?: IAlertHistoryAnalytics, ): Promise; + /** + * Append one error to the ERROR-state history row of the given window, + * keeping the errors already on it. Used for failures noticed after the + * window's evaluation finished, such as a resolve that was never delivered. + */ + appendAlertErrorHistory( + alertId: string, + evaluationWindowStart: Date, + error: IAlertError, + ): Promise; + + /** Notifications waiting to be sent again for the given alerts. */ + getPendingNotifications(alertIds: string[]): Promise; + + /** Store failed notifications, replacing any stored with the same eventId. */ + savePendingNotifications(records: IPendingNotification[]): Promise; + + /** Record another failed attempt for a stored notification. */ + updatePendingNotification( + eventId: string, + update: Pick, + ): Promise; + + deletePendingNotification(eventId: string): Promise; + /** Fetch all webhooks for the given team, returning a map of webhook ID to webhook */ getWebhooks(teamId: string | ObjectId): Promise>; diff --git a/packages/api/src/tasks/checkAlerts/template.ts b/packages/api/src/tasks/checkAlerts/template.ts index 94e96d1dd3..9152df4055 100644 --- a/packages/api/src/tasks/checkAlerts/template.ts +++ b/packages/api/src/tasks/checkAlerts/template.ts @@ -62,6 +62,7 @@ import { createHandlebarsWithHelpers } from '@/tasks/checkAlerts/transports'; import { unflattenObject } from '@/tasks/util'; import { resolveAlertDisplayFields } from '@/utils/alerts'; import { truncateString } from '@/utils/common'; +import { serializeErrorWithCause } from '@/utils/errors'; import { getCounter } from '@/utils/instrumentation'; import logger from '@/utils/logger'; @@ -479,6 +480,8 @@ export type NotificationFailure = { /** The channel type the target belongs to, or 'unknown' when it couldn't be determined (e.g. an unparseable @mention). */ type: AlertChannelType | 'unknown'; error: unknown; + /** Set when `dispatch()` rejected, so the caller can send the job again. */ + job?: NotificationJob; }; // PopulatedAlertChannel only has a `webhook` variant in this repo, but a @@ -995,7 +998,7 @@ ${targetTemplate}`; { alertId: alert.id, webhookId: channelKey(job.populatedChannel), - error: serializeError(e), + error: serializeErrorWithCause(e), }, 'Failed to deliver alert notification', ); @@ -1003,6 +1006,7 @@ ${targetTemplate}`; target: channelLabel(job.populatedChannel), type: job.populatedChannel.type, error: e, + job, }); } finally { timings.push({ diff --git a/packages/api/src/tasks/checkAlerts/transports/generic.ts b/packages/api/src/tasks/checkAlerts/transports/generic.ts index b4b2f04769..f5bb280300 100644 --- a/packages/api/src/tasks/checkAlerts/transports/generic.ts +++ b/packages/api/src/tasks/checkAlerts/transports/generic.ts @@ -20,6 +20,7 @@ import type { WebhookChannel, } from '@/tasks/checkAlerts/transports/types'; import { escapeJsonString } from '@/tasks/util'; +import { serializeErrorWithCause } from '@/utils/errors'; import { getCounter, getHistogram } from '@/utils/instrumentation'; import logger from '@/utils/logger'; import { withRetry } from '@/utils/retry'; @@ -293,7 +294,8 @@ const sendGenericWebhook = async ( } catch (e) { logger.error( { - error: serializeError(e), + error: serializeErrorWithCause(e), + webhookId: webhook._id.toString(), }, 'Failed to send generic webhook message', ); diff --git a/packages/api/src/utils/__tests__/errors.test.ts b/packages/api/src/utils/__tests__/errors.test.ts index 75999d1bd7..fada1dd114 100644 --- a/packages/api/src/utils/__tests__/errors.test.ts +++ b/packages/api/src/utils/__tests__/errors.test.ts @@ -1,4 +1,9 @@ -import { Api500Error, BaseError, isOperationalError } from '@/utils/errors'; +import { + Api500Error, + BaseError, + isOperationalError, + serializeErrorWithCause, +} from '@/utils/errors'; describe('Errors utils', () => { test('BaseError class', () => { @@ -18,4 +23,53 @@ describe('Errors utils', () => { ).toBeTruthy(); expect(isOperationalError(new Api500Error('BANG'))).toBeTruthy(); }); + + describe('serializeErrorWithCause', () => { + it('keeps the cause chain that serializeError drops', () => { + const connectError = Object.assign( + new Error('connect ECONNREFUSED 10.0.0.1:443'), + { code: 'ECONNREFUSED' }, + ); + const error = new TypeError('fetch failed', { cause: connectError }); + + expect(serializeErrorWithCause(error)).toMatchObject({ + name: 'TypeError', + message: 'fetch failed', + cause: { + message: 'connect ECONNREFUSED 10.0.0.1:443', + code: 'ECONNREFUSED', + }, + }); + }); + + it("keeps an AggregateError's errors", () => { + const error = new TypeError('fetch failed', { + cause: new AggregateError( + [new Error('connect ETIMEDOUT a'), new Error('connect ETIMEDOUT b')], + 'all attempts failed', + ), + }); + + expect(serializeErrorWithCause(error)).toMatchObject({ + cause: { + message: 'all attempts failed', + errors: [ + { message: 'connect ETIMEDOUT a' }, + { message: 'connect ETIMEDOUT b' }, + ], + }, + }); + }); + + it('stops at a self-referential cause', () => { + const error = new Error('loop'); + error.cause = error; + + expect(() => serializeErrorWithCause(error)).not.toThrow(); + }); + + it('passes non-errors through serializeError', () => { + expect(serializeErrorWithCause('boom')).toBe('boom'); + }); + }); }); diff --git a/packages/api/src/utils/errors.ts b/packages/api/src/utils/errors.ts index b50c55f99f..0b99319390 100644 --- a/packages/api/src/utils/errors.ts +++ b/packages/api/src/utils/errors.ts @@ -1,3 +1,5 @@ +import { serializeError } from 'serialize-error'; + export enum StatusCode { BAD_REQUEST = 400, CONFLICT = 409, @@ -63,3 +65,30 @@ export const isDuplicateKeyError = (e: unknown): boolean => e != null && typeof e === 'object' && (e as { code?: unknown }).code === 11000; + +// Guards against self-referential cause chains. +const MAX_CAUSE_DEPTH = 5; + +/** + * serializeError, keeping `cause` and AggregateError's `errors`, which it + * drops. undici's fetch rejects with a bare "fetch failed" and puts the + * reason (DNS, connect, TLS) only on `cause`. + */ +export const serializeErrorWithCause = (error: unknown, depth = 0): unknown => { + if (!(error instanceof Error)) { + return serializeError(error); + } + const serialized: Record = { ...serializeError(error) }; + if (depth >= MAX_CAUSE_DEPTH) { + return serialized; + } + if (error.cause != null) { + serialized.cause = serializeErrorWithCause(error.cause, depth + 1); + } + if (error instanceof AggregateError) { + serialized.errors = error.errors.map(e => + serializeErrorWithCause(e, depth + 1), + ); + } + return serialized; +}; diff --git a/packages/api/src/utils/retry.ts b/packages/api/src/utils/retry.ts index fd14bfcdff..1c5c729c84 100644 --- a/packages/api/src/utils/retry.ts +++ b/packages/api/src/utils/retry.ts @@ -1,5 +1,4 @@ -import { serializeError } from 'serialize-error'; - +import { serializeErrorWithCause } from './errors'; import logger from './logger'; interface RetryOptions { @@ -82,7 +81,7 @@ export const withRetry = async ( logger.warn( { - error: serializeError(error), + error: serializeErrorWithCause(error), attempt, maxRetries, delayMs: delay,