diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 4826e3fcc..7af92cd5c 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -342,16 +342,16 @@ receiver hears about them without anyone watching the console. Set | Field | Rule | |---|---| | `name` | Lowercase letters, digits and dashes, starting with a letter, at most 32 characters; no two destinations share one. | -| `format` | `generic`, the only format so far. | +| `format` | `generic` (the signed JSON envelope) or `slack` (a written message for Slack). | | `url` | `https://`, or `http://` to `localhost`, `127.0.0.1` or `[::1]`; no user name or password in it. | | `events` | A non-empty list, each event once, from: `worker.connected`, `worker.disconnected`, `worker.rejected`, `worker.drain-started`, `worker.drain-ended`, `device.quarantine-abandoned`, `device.quarantine-stranded`, `device.recovery-failed`, `device.stalled-transition-detected`, `disk.pressure-detected`, `component.install-failed`. | -| `secret` | A non-empty string the gateway signs each message with. | +| `secret` | A `generic` destination needs a non-empty string the gateway signs each message with. A `slack` destination takes none; a `secret` key on it is refused. | An event a worker reports reaches the destination too, with `workerId` in its payload. A key not listed above is warned about and ignored, and `simlock config set` refuses it. -**What the receiver gets.** One `POST` per event, with the event as the body: the +**What a `generic` receiver gets.** One `POST` per event, with the event as the body: the same JSON `simlock events --json` prints (`id`, `seq`, `timestamp`, `event`, `payload`, `module`). Headers: @@ -367,6 +367,36 @@ To check a message, compute the HMAC-SHA256 of the bytes you received with the secret and compare it to the header value in constant time. Use the raw body, not a re-serialised copy. +**Slack.** A `slack` destination gets `{"text": ""}` for each event it +lists, written for a person, through the same queue, tries and limits as +`generic`, with only the `Content-Type` and `User-Agent` headers (no `X-Simlock-Event`, `X-Simlock-Delivery` or `X-Simlock-Signature`, and no body but the `text`). Slack is the +only service promised; other services that take Slack's incoming-webhook body +may work. `` is the worker's label as the gateway knows it when the +message is sent, or its id when it has no label. Every variable part is cut to +300 characters (`โ€ฆ` added) and has `&`, `<` and `>` written as `&`, `<` +and `>`, so a label like `` never pings a channel. A field a worker +left out prints as `undefined`, and a size or age that is not a number prints as +`?`. + +| Event | Message | +|---|---| +| `worker.connected` | `๐ŸŸข Worker connected (simlock ).`, or `๐ŸŸข Worker connected.` without a version | +| `worker.disconnected` | `๐Ÿ”ด Worker disconnected.`, or `โ€ฆ disconnected while holding 1 lease.`, or `โ€ฆ while holding leases.` | +| `worker.rejected` | `โ›” A worker was refused (): .` and, when `count` is above 1, ` more refusals like it came before this one.` `` is the label the connection claimed, else its worker id, else `an unknown worker`. | +| `worker.drain-started` | `โธ Worker is drained: it takes no new leases.` | +| `worker.drain-ended` | `โ–ถ๏ธ Worker takes new leases again.` | +| `device.quarantine-abandoned` | `โš ๏ธ Device on was retired after failed clean-ups.` | +| `device.quarantine-stranded` | `๐Ÿ›‘ Device on could not be cleaned up or deleted after tries: . It needs a person.` | +| `device.recovery-failed` | `โš ๏ธ Device on crashed during lease and could not be brought back: .` | +| `device.stalled-transition-detected` | `โณ Device on has been for .` | +| `disk.pressure-detected` | `๐Ÿ’พ Worker is low on disk: free (threshold ).` | +| `component.install-failed` | `โŒ Installing on failed: .` | + +`` reads `45 s`, `14 min` or `1 h 5 min` (`2 h` when no minutes are left), +rounded down. `` and `` are in GiB (1024ยณ bytes), or MiB +under 1 GiB, with one decimal rounded half up and `.0` dropped: `8.2 GiB`, +`10 GiB`, `512 MiB`. + **Delivery.** Each destination sends one message at a time, in the order the gateway saw the events, and a slow destination never delays another. A try ends after 10 seconds. A network error, a timeout, `429` or a `5xx` is tried diff --git a/src/alerts/AGENTS.md b/src/alerts/AGENTS.md index 9ee7c9c5b..7c5c97a97 100644 --- a/src/alerts/AGENTS.md +++ b/src/alerts/AGENTS.md @@ -9,13 +9,14 @@ Imported by: `../daemon/main.ts` only. ## Files -| File | Role | -| ------------------ | ------------------------------------------------------------------------ | -| `index.ts` | `createAlerts`: one subscription per listed event, `stop()` | -| `queue.ts` | `DestinationQueue`: one message at a time, tries, timeout, bounded wait | -| `formats/types.ts` | `AlertFormatDefinition`: `build` turns an envelope into headers and body | -| `formats/index.ts` | `ALERT_FORMAT_REGISTRY`, typed over core's `ALERT_FORMATS` | -| `formats/generic/` | The envelope as JSON, signed with HMAC-SHA256 | +| File | Role | +| ------------------ | ---------------------------------------------------------------------------------------------------------- | +| `index.ts` | `createAlerts`: one subscription per listed event, `stop()` | +| `queue.ts` | `DestinationQueue`: one message at a time, tries, timeout, bounded wait | +| `formats/types.ts` | `AlertFormatDefinition`: `build` turns an envelope into headers and body | +| `formats/index.ts` | `ALERT_FORMAT_REGISTRY`, typed over core's `ALERT_FORMATS` | +| `formats/generic/` | The envelope as JSON, signed with HMAC-SHA256 | +| `formats/slack/` | `{text}` from a message table (`formats/slack/messages.ts`), `formats/slack/units.ts` cuts, escapes, sizes | ## How it works @@ -28,3 +29,18 @@ Imported by: `../daemon/main.ts` only. Logs name the destination, never its URL or secret; a failed try is logged by code, never by an error's message. A format is added by a directory under `formats/`, an entry in the registry and in core's `ALERT_FORMATS`; the queue never names a format. + +## Slack messages + +`formats/slack/messages.ts` is a `Record`: an event added to `ALERT_EVENTS` does not +compile until it has a message. Every variable part goes through `part()` (cut to 300 code points, then +`&`, `<`, `>` escaped), and `worker.rejected` names what the connection claimed, never the registry. +`createAlerts` takes `workerLabel(workerId)` (the gateway's view of the worker, from `../daemon/main.ts`) +and falls back to the id; the message is built when it leaves the queue, so it names the worker as known +then. + +## Known gaps + +- A worker's relayed payload is not checked (the gateway replaces a non-object with `{workerId}`): a Slack + message built from a missing field prints `undefined`, and an age or size that is not a finite number + prints `?`. diff --git a/src/alerts/alerts.test.ts b/src/alerts/alerts.test.ts index 149e092b8..b83bb766a 100644 --- a/src/alerts/alerts.test.ts +++ b/src/alerts/alerts.test.ts @@ -23,6 +23,11 @@ function destination(overrides: Partial = {}): AlertDestinatio }; } +function slackDestination(overrides: Partial = {}): AlertDestination { + const { secret: _secret, ...rest } = destination({ format: "slack", ...overrides }); + return rest; +} + function settle(): Promise { return new Promise((resolve) => setImmediate(resolve)); } @@ -34,6 +39,7 @@ class Harness { readonly eventBus = new EventBus(this.clock); readonly poster = new FakeWebhookPoster(); readonly sink = new MemoryLogSink(); + readonly labels = new Map(); alerts: Alerts; constructor(destinations: AlertDestination[] = [destination()]) { @@ -46,6 +52,7 @@ class Harness { "alerts", ), version: "9.9.9", + workerLabel: (id) => this.labels.get(id), }); } @@ -136,6 +143,388 @@ describe("alerts", () => { }, ); + it('a slack destination posts {"text": message} with only the Content-Type and User-Agent headers', async () => { + const h = harness([slackDestination()]); + h.emit("worker.connected", connected("wrk_1")); + await settle(); + + expect(h.poster.requests).toHaveLength(1); + expect(h.poster.requests[0]?.headers).toEqual({ + "Content-Type": "application/json", + "User-Agent": "simlock/9.9.9", + }); + expect(JSON.parse(h.poster.requests[0]?.body ?? "")).toEqual({ + text: "๐ŸŸข Worker wrk_1 connected.", + }); + }); + + it("each of the eleven events posts its message word for word", async () => { + const rows: [EventName, Record, string][] = [ + [ + "worker.connected", + { workerId: "wrk_1", version: "1.2.3" }, + "๐ŸŸข Worker wrk_1 connected (simlock 1.2.3).", + ], + ["worker.connected", { workerId: "wrk_1" }, "๐ŸŸข Worker wrk_1 connected."], + [ + "worker.disconnected", + { workerId: "wrk_1", leaseCount: 0 }, + "๐Ÿ”ด Worker wrk_1 disconnected.", + ], + [ + "worker.disconnected", + { workerId: "wrk_1", leaseCount: 1 }, + "๐Ÿ”ด Worker wrk_1 disconnected while holding 1 lease.", + ], + [ + "worker.disconnected", + { workerId: "wrk_1", leaseCount: 3 }, + "๐Ÿ”ด Worker wrk_1 disconnected while holding 3 leases.", + ], + [ + "worker.rejected", + { reason: "forbidden", workerId: "wrk_x" }, + "โ›” A worker was refused (forbidden): wrk_x.", + ], + [ + "worker.drain-started", + { workerId: "wrk_1" }, + "โธ Worker wrk_1 is drained: it takes no new leases.", + ], + ["worker.drain-ended", { workerId: "wrk_1" }, "โ–ถ๏ธ Worker wrk_1 takes new leases again."], + [ + "device.quarantine-abandoned", + { workerId: "wrk_1", deviceId: "dev_1", attempts: 3 }, + "โš ๏ธ Device dev_1 on wrk_1 was retired after 3 failed clean-ups.", + ], + [ + "device.quarantine-stranded", + { workerId: "wrk_1", deviceId: "dev_1", attempts: 4, error: "disk full" }, + "๐Ÿ›‘ Device dev_1 on wrk_1 could not be cleaned up or deleted after 4 tries: disk full. It needs a person.", + ], + [ + "device.recovery-failed", + { workerId: "wrk_1", deviceId: "dev_1", leaseId: "lse_1", reason: "boot-timeout" }, + "โš ๏ธ Device dev_1 on wrk_1 crashed during lease lse_1 and could not be brought back: boot-timeout.", + ], + [ + "device.stalled-transition-detected", + { workerId: "wrk_1", deviceId: "dev_1", state: "booting", ageMs: 840_000 }, + "โณ Device dev_1 on wrk_1 has been booting for 14 min.", + ], + [ + "disk.pressure-detected", + { workerId: "wrk_1", freeBytes: 8_804_682_956, threshold: 10_737_418_240 }, + "๐Ÿ’พ Worker wrk_1 is low on disk: 8.2 GiB free (threshold 10 GiB).", + ], + [ + "component.install-failed", + { workerId: "wrk_1", platform: "ios", componentId: "runtime-18", error: "no space" }, + "โŒ Installing ios runtime-18 on wrk_1 failed: no space.", + ], + ]; + const events = [...new Set(rows.map(([event]) => event))] as AlertDestination["events"]; + const h = harness([slackDestination({ events })]); + for (const [event, payload] of rows) { + h.emit(event, payload as never); + await settle(); + } + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual( + rows.map(([, , text]) => text), + ); + }); + + it("names the worker by the label workerLabel answers, and by the id when it answers undefined", async () => { + const h = harness([slackDestination({ events: ["worker.drain-started"] })]); + h.labels.set("wrk_1", "build-mac-3"); + h.emit("worker.drain-started", { workerId: "wrk_1" }); + h.emit("worker.drain-started", { workerId: "wrk_2" }); + await settle(); + await h.advance(1_000); + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual([ + "โธ Worker build-mac-3 is drained: it takes no new leases.", + "โธ Worker wrk_2 is drained: it takes no new leases.", + ]); + }); + + it("builds a message when it leaves the queue and resends the same body on later tries", async () => { + const h = harness([slackDestination({ events: ["worker.drain-started"] })]); + const first = deferred(); + h.poster.respond = (_request, index) => + index === 0 ? first.promise : Promise.resolve({ status: 503 }); + h.labels.set("wrk_1", "old-name"); + h.emit("worker.drain-started", { workerId: "wrk_0" }); + await settle(); + h.emit("worker.drain-started", { workerId: "wrk_1" }); + h.labels.set("wrk_1", "new-name"); + first.resolve({ status: 200 }); + await settle(); + h.labels.set("wrk_1", "newer-name"); + await h.advance(1_000); + + const texts = h.poster.requests.slice(1).map((request) => JSON.parse(request.body).text); + expect(texts).toEqual([ + "โธ Worker new-name is drained: it takes no new leases.", + "โธ Worker new-name is drained: it takes no new leases.", + ]); + }); + + it("worker.rejected names the claimed label, else the claimed id, else an unknown worker, and adds the count only above one", async () => { + const h = harness([slackDestination({ events: ["worker.rejected"] })]); + h.labels.set("wrk_9", "registry-name"); + const cases: [Record, string][] = [ + [{ reason: "forbidden", workerId: "wrk_9", label: "claimed-label" }, "claimed-label."], + [{ reason: "forbidden", workerId: "wrk_9" }, "wrk_9."], + [{ reason: "unauthenticated" }, "an unknown worker."], + [{ reason: "forbidden", workerId: "wrk_9", count: 1 }, "wrk_9."], + [ + { reason: "forbidden", workerId: "wrk_9", count: 4 }, + "wrk_9. 4 more refusals like it came before this one.", + ], + ]; + for (const [payload] of cases) { + h.emit("worker.rejected", payload as never); + await settle(); + } + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual( + cases.map(([payload, tail]) => `โ›” A worker was refused (${payload.reason}): ${tail}`), + ); + }); + + it("formats an age in seconds, minutes, and hours with minutes", async () => { + const h = harness([slackDestination({ events: ["device.stalled-transition-detected"] })]); + const cases: [number, string][] = [ + [45_000, "45 s"], + [840_000, "14 min"], + [3_900_000, "1 h 5 min"], + [7_200_000, "2 h"], + [59_999, "59 s"], + [60_000, "1 min"], + [3_600_000, "1 h"], + ]; + for (const [ageMs] of cases) { + h.emit("device.stalled-transition-detected", { + workerId: "wrk_1", + deviceId: "dev_1", + state: "booting", + ageMs, + } as never); + await settle(); + } + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual( + cases.map(([, age]) => `โณ Device dev_1 on wrk_1 has been booting for ${age}.`), + ); + }); + + it("formats bytes in GiB and MiB with one decimal rounded half up, dropping .0", async () => { + const h = harness([slackDestination({ events: ["disk.pressure-detected"] })]); + const cases: [number, string][] = [ + [8_804_682_956, "8.2 GiB"], + [10_737_418_240, "10 GiB"], + [536_870_912, "512 MiB"], + [1_020_054_733, "972.8 MiB"], + [1_073_741_824, "1 GiB"], + ]; + for (const [freeBytes] of cases) { + h.emit("disk.pressure-detected", { + workerId: "wrk_1", + freeBytes, + threshold: 10_737_418_240, + } as never); + await settle(); + } + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual( + cases.map(([, size]) => `๐Ÿ’พ Worker wrk_1 is low on disk: ${size} free (threshold 10 GiB).`), + ); + }); + + it("cuts every variable part to 300 code points, then escapes &, < and >", async () => { + const h = harness([ + slackDestination({ + events: [ + "component.install-failed", + "worker.drain-started", + "worker.rejected", + "device.quarantine-abandoned", + "device.stalled-transition-detected", + ], + }), + ]); + h.labels.set("wrk_1", ""); + h.labels.set("wrk_2", `${"a".repeat(299)}๐Ÿ˜€b`); + const install = (error: string) => + [ + "component.install-failed", + { workerId: "wrk_0", platform: "ios", componentId: "c", error }, + ] as const; + const cases: [readonly [EventName, Record], string][] = [ + [install("&"), "โŒ Installing ios c on wrk_0 failed: <b>&."], + [ + ["worker.drain-started", { workerId: "wrk_1" }], + "โธ Worker <!channel> is drained: it takes no new leases.", + ], + [ + ["worker.rejected", { reason: "forbidden", label: "" }], + "โ›” A worker was refused (forbidden): <!here>.", + ], + [ + ["worker.rejected", { reason: "", label: "x" }], + "โ›” A worker was refused (<r>): x.", + ], + [ + [ + "device.stalled-transition-detected", + { workerId: "wrk_0", deviceId: "d", state: "", ageMs: 1000 }, + ], + "โณ Device d on wrk_0 has been <s> for 1 s.", + ], + [ + ["device.quarantine-abandoned", { workerId: "wrk_0", deviceId: "d", attempts: "" }], + "โš ๏ธ Device d on wrk_0 was retired after <a> failed clean-ups.", + ], + [ + install(`${"e".repeat(299)}&b`), + `โŒ Installing ios c on wrk_0 failed: ${"e".repeat(299)}&โ€ฆ.`, + ], + [ + install(`${"e".repeat(299)} JSON.parse(request.body).text)).toEqual( + cases.map(([, text]) => text), + ); + }); + + it("escapes every variable part of every message that carries one", async () => { + const x = ""; + const h = harness([ + slackDestination({ + events: [ + "worker.connected", + "worker.disconnected", + "worker.rejected", + "device.quarantine-abandoned", + "device.quarantine-stranded", + "device.recovery-failed", + "device.stalled-transition-detected", + "component.install-failed", + ], + }), + ]); + const w = { workerId: "wrk_0" }; + const cases: [EventName, Record, string][] = [ + ["worker.connected", { ...w, version: x }, "๐ŸŸข Worker wrk_0 connected (simlock <x>)."], + [ + "worker.disconnected", + { ...w, leaseCount: x }, + "๐Ÿ”ด Worker wrk_0 disconnected while holding <x> leases.", + ], + [ + "worker.rejected", + { reason: x, workerId: x }, + "โ›” A worker was refused (<x>): <x>.", + ], + [ + "device.quarantine-abandoned", + { ...w, deviceId: x, attempts: x }, + "โš ๏ธ Device <x> on wrk_0 was retired after <x> failed clean-ups.", + ], + [ + "device.quarantine-stranded", + { ...w, deviceId: x, attempts: x, error: x }, + "๐Ÿ›‘ Device <x> on wrk_0 could not be cleaned up or deleted after <x> tries: <x>. It needs a person.", + ], + [ + "device.recovery-failed", + { ...w, deviceId: x, leaseId: x, reason: x }, + "โš ๏ธ Device <x> on wrk_0 crashed during lease <x> and could not be brought back: <x>.", + ], + [ + "device.stalled-transition-detected", + { ...w, deviceId: x, state: x, ageMs: 1000 }, + "โณ Device <x> on wrk_0 has been <x> for 1 s.", + ], + [ + "component.install-failed", + { ...w, platform: x, componentId: x, error: x }, + "โŒ Installing <x> <x> on wrk_0 failed: <x>.", + ], + ]; + for (const [event, payload] of cases) { + h.emit(event, payload as never); + await settle(); + } + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual( + cases.map(([, , text]) => text), + ); + }); + + it("a relayed payload with missing fields posts the text the Known gap names", async () => { + const h = harness([ + slackDestination({ + events: [ + "device.quarantine-abandoned", + "disk.pressure-detected", + "device.stalled-transition-detected", + ], + }), + ]); + const cases: [EventName, Record, string][] = [ + [ + "device.quarantine-abandoned", + { workerId: "wrk_1", deviceId: "dev_1" }, + "โš ๏ธ Device dev_1 on wrk_1 was retired after undefined failed clean-ups.", + ], + [ + "disk.pressure-detected", + { workerId: "wrk_1", threshold: 10_737_418_240 }, + "๐Ÿ’พ Worker wrk_1 is low on disk: ? free (threshold 10 GiB).", + ], + [ + "device.stalled-transition-detected", + { workerId: "wrk_1" }, + "โณ Device undefined on wrk_1 has been undefined for ?.", + ], + ]; + for (const [event, payload] of cases) { + h.emit(event, payload as never); + await settle(); + } + + expect(h.poster.requests.map((request) => JSON.parse(request.body).text)).toEqual( + cases.map(([, , text]) => text), + ); + }); + + it("a slack destination is tried again on 5xx like a generic one", async () => { + const h = harness([slackDestination()]); + h.failWith(503); + h.emit("worker.connected", connected()); + await settle(); + await h.advance(1_000); + + expect(h.poster.requests).toHaveLength(2); + expect(h.poster.requests[1]?.body).toBe(h.poster.requests[0]?.body); + }); + it("posts nothing for an event the destination did not list", async () => { const h = harness(); h.emit("worker.connected", connected()); @@ -499,6 +888,7 @@ describe("alerts", () => { clock: h.clock, logger: new JsonLinesLogger({ clock: h.clock, level: "debug", sink: h.sink }), version: "9.9.9", + workerLabel: () => undefined, }); h.emit("worker.connected", connected()); diff --git a/src/alerts/formats/index.ts b/src/alerts/formats/index.ts index d38911a6e..9d0a7c261 100644 --- a/src/alerts/formats/index.ts +++ b/src/alerts/formats/index.ts @@ -1,7 +1,9 @@ +import { slackFormat } from "./slack/index.js"; import type { AlertFormatName } from "../../core/index.js"; import { genericFormat } from "./generic/index.js"; import type { AlertFormatDefinition } from "./types.js"; export const ALERT_FORMAT_REGISTRY: { readonly [N in AlertFormatName]: AlertFormatDefinition } = { generic: genericFormat, + slack: slackFormat, }; diff --git a/src/alerts/formats/slack/index.ts b/src/alerts/formats/slack/index.ts new file mode 100644 index 000000000..dfed1b559 --- /dev/null +++ b/src/alerts/formats/slack/index.ts @@ -0,0 +1,16 @@ +import type { AlertEventName } from "../../../core/index.js"; +import type { AlertFormatDefinition } from "../types.js"; +import { SLACK_MESSAGES } from "./messages.js"; +import { part } from "./units.js"; + +export const slackFormat: AlertFormatDefinition = { + name: "slack", + build({ envelope, version, workerLabel }) { + const payload = envelope.payload as Record; + const message = SLACK_MESSAGES[envelope.event as AlertEventName]; + return { + body: JSON.stringify({ text: message(payload, part(workerLabel(String(payload.workerId)))) }), + headers: { "Content-Type": "application/json", "User-Agent": `simlock/${version}` }, + }; + }, +}; diff --git a/src/alerts/formats/slack/messages.ts b/src/alerts/formats/slack/messages.ts new file mode 100644 index 000000000..539ca87b1 --- /dev/null +++ b/src/alerts/formats/slack/messages.ts @@ -0,0 +1,40 @@ +import type { AlertEventName } from "../../../core/index.js"; +import { formatAge, formatBytes, part } from "./units.js"; + +type Payload = Record; +type Message = (payload: Payload, worker: string) => string; + +function leases(count: unknown): string { + if (count === 0) return ""; + return count === 1 ? " while holding 1 lease" : ` while holding ${part(count)} leases`; +} + +function more(count: unknown): string { + return typeof count === "number" && count > 1 + ? ` ${count} more refusals like it came before this one.` + : ""; +} + +export const SLACK_MESSAGES: Record = { + "worker.connected": (p, worker) => + p.version === undefined + ? `๐ŸŸข Worker ${worker} connected.` + : `๐ŸŸข Worker ${worker} connected (simlock ${part(p.version)}).`, + "worker.disconnected": (p, worker) => `๐Ÿ”ด Worker ${worker} disconnected${leases(p.leaseCount)}.`, + "worker.rejected": (p) => + `โ›” A worker was refused (${part(p.reason)}): ${part(p.label ?? p.workerId ?? "an unknown worker")}.${more(p.count)}`, + "worker.drain-started": (_p, worker) => `โธ Worker ${worker} is drained: it takes no new leases.`, + "worker.drain-ended": (_p, worker) => `โ–ถ๏ธ Worker ${worker} takes new leases again.`, + "device.quarantine-abandoned": (p, worker) => + `โš ๏ธ Device ${part(p.deviceId)} on ${worker} was retired after ${part(p.attempts)} failed clean-ups.`, + "device.quarantine-stranded": (p, worker) => + `๐Ÿ›‘ Device ${part(p.deviceId)} on ${worker} could not be cleaned up or deleted after ${part(p.attempts)} tries: ${part(p.error)}. It needs a person.`, + "device.recovery-failed": (p, worker) => + `โš ๏ธ Device ${part(p.deviceId)} on ${worker} crashed during lease ${part(p.leaseId)} and could not be brought back: ${part(p.reason)}.`, + "device.stalled-transition-detected": (p, worker) => + `โณ Device ${part(p.deviceId)} on ${worker} has been ${part(p.state)} for ${formatAge(p.ageMs)}.`, + "disk.pressure-detected": (p, worker) => + `๐Ÿ’พ Worker ${worker} is low on disk: ${formatBytes(p.freeBytes)} free (threshold ${formatBytes(p.threshold)}).`, + "component.install-failed": (p, worker) => + `โŒ Installing ${part(p.platform)} ${part(p.componentId)} on ${worker} failed: ${part(p.error)}.`, +}; diff --git a/src/alerts/formats/slack/units.ts b/src/alerts/formats/slack/units.ts new file mode 100644 index 000000000..df0ddfa2d --- /dev/null +++ b/src/alerts/formats/slack/units.ts @@ -0,0 +1,30 @@ +const GIB = 1024 ** 3; +const MIB = 1024 ** 2; + +function oneDecimal(value: number): string { + const rounded = Math.floor(value * 10 + 0.5) / 10; + return Number.isInteger(rounded) ? String(rounded) : rounded.toFixed(1); +} + +export function formatBytes(value: unknown): string { + if (typeof value !== "number" || !Number.isFinite(value)) return "?"; + return value < GIB ? `${oneDecimal(value / MIB)} MiB` : `${oneDecimal(value / GIB)} GiB`; +} + +export function formatAge(value: unknown): string { + if (typeof value !== "number" || !Number.isFinite(value)) return "?"; + if (value < 60_000) return `${Math.floor(value / 1_000)} s`; + if (value < 3_600_000) return `${Math.floor(value / 60_000)} min`; + const hours = Math.floor(value / 3_600_000); + const minutes = Math.floor((value % 3_600_000) / 60_000); + return minutes === 0 ? `${hours} h` : `${hours} h ${minutes} min`; +} + +const MAX_PART = 300; + +/** why: Slack reads `&`, `<` and `>` in text as markup, and `` pings a channel. */ +export function part(value: unknown): string { + const points = Array.from(String(value)); + const cut = points.length > MAX_PART ? `${points.slice(0, MAX_PART).join("")}โ€ฆ` : points.join(""); + return cut.replaceAll("&", "&").replaceAll("<", "<").replaceAll(">", ">"); +} diff --git a/src/alerts/index.ts b/src/alerts/index.ts index a3e53fceb..cfa61daf0 100644 --- a/src/alerts/index.ts +++ b/src/alerts/index.ts @@ -15,10 +15,13 @@ export function createAlerts(options: { clock: Clock; logger: Logger; version: string; + workerLabel(workerId: string): string | undefined; }): Alerts { const { destinations, eventBus, poster, clock, logger, version } = options; + const workerLabel = (workerId: string): string => options.workerLabel(workerId) ?? workerId; const queues = destinations.map( - (destination) => new DestinationQueue({ destination, poster, clock, logger, version }), + (destination) => + new DestinationQueue({ destination, poster, clock, logger, version, workerLabel }), ); const unsubscribes = ALERT_EVENTS.filter((name) => diff --git a/src/alerts/queue.ts b/src/alerts/queue.ts index c2e3714a6..a72d922d9 100644 --- a/src/alerts/queue.ts +++ b/src/alerts/queue.ts @@ -21,6 +21,7 @@ interface QueueOptions { readonly clock: Clock; readonly logger: Logger; readonly version: string; + workerLabel(workerId: string): string; } function errorCode(error: unknown): string { @@ -95,7 +96,7 @@ export class DestinationQueue { envelope, destination: this.#options.destination, version: this.#options.version, - workerLabel: (id) => id, + workerLabel: this.#options.workerLabel, }); } catch { this.#fields("warn", "Alert send gave up", envelope, { attempt: 1, error: "build-failed" }); diff --git a/src/contract/AGENTS.md b/src/contract/AGENTS.md index 357fbac7d..09b4406da 100644 --- a/src/contract/AGENTS.md +++ b/src/contract/AGENTS.md @@ -55,7 +55,7 @@ Three identities: ## Protocol version -- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 49, max: 49 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`. +- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 50, max: 50 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`. - Every change to a contract shape bumps it by one in the same PR. Move `min` too, unless a compatibility path stays side by side. `.agents/scripts/protocol-bump-check.mjs` checks it on push and in CI; it counts every file here as a shape but tests, `index.ts` and the diff --git a/src/contract/protocol.test.ts b/src/contract/protocol.test.ts index b827f59f0..be9ee39fa 100644 --- a/src/contract/protocol.test.ts +++ b/src/contract/protocol.test.ts @@ -48,6 +48,6 @@ describe("helloRequestSchema", () => { describe("PROTOCOL_VERSION_RANGE", () => { it("is one above the base branch's, for alerts destinations in config.get", () => { - expect(PROTOCOL_VERSION_RANGE).toEqual({ max: 49, min: 49 }); + expect(PROTOCOL_VERSION_RANGE).toEqual({ max: 50, min: 50 }); }); }); diff --git a/src/contract/protocol.ts b/src/contract/protocol.ts index 09ce670dd..95b67b4d7 100644 --- a/src/contract/protocol.ts +++ b/src/contract/protocol.ts @@ -10,7 +10,7 @@ export interface ProtocolRange { const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() }); /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */ -export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 49, max: 49 }; +export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 50, max: 50 }; const LEGACY_DAEMON_PROTOCOL_VERSION = 2; diff --git a/src/contract/schemas.ts b/src/contract/schemas.ts index decd7ebc3..d1b5bbcc6 100644 --- a/src/contract/schemas.ts +++ b/src/contract/schemas.ts @@ -1117,7 +1117,7 @@ export const configSchema = z.object({ destinations: z.array( z.object({ name: z.string(), - format: z.enum(["generic"]), + format: z.enum(["generic", "slack"]), url: z.string(), events: z.array(z.string()), secret: z.string().optional(), diff --git a/src/core/config.test.ts b/src/core/config.test.ts index e46ff9532..c8f1f0d1a 100644 --- a/src/core/config.test.ts +++ b/src/core/config.test.ts @@ -1939,6 +1939,7 @@ describe("gateway.alerts.destinations", () => { "expected a name of lowercase letters, digits and dashes that starts with a letter, at most 32 characters"; const urlExpected = "expected an https:// URL, or an http:// URL to localhost, 127.0.0.1 or [::1], with no user name or password"; + const formatExpected = 'expected one of "generic", "slack"'; const eventsExpected = `expected a non-empty list of alert events, each once: ${ALERT_EVENTS.join(", ")}`; async function load( @@ -1987,9 +1988,9 @@ describe("gateway.alerts.destinations", () => { ["name", bad("name", ""), nameExpected], ["name", bad("name", "a".repeat(33)), nameExpected], ["name", without("name"), nameExpected], - ["format", bad("format", "slack"), 'expected one of "generic"'], - ["format", bad("format", "json"), 'expected one of "generic"'], - ["format", without("format"), 'expected one of "generic"'], + ["format", bad("format", "teams"), formatExpected], + ["format", bad("format", "json"), formatExpected], + ["format", without("format"), formatExpected], ["url", bad("url", "http://n8n.internal/x"), urlExpected], ["url", bad("url", "ftp://example.com"), urlExpected], ["url", bad("url", "n8n.example.com/x"), urlExpected], @@ -2011,6 +2012,26 @@ describe("gateway.alerts.destinations", () => { ); }); + it("loadConfig accepts a slack destination, refuses a secret on it, and lists both formats in the format message", async () => { + const { secret: _secret, ...slack } = { ...validDestination, format: "slack" }; + const config = await loadDestinations([slack]); + expect(config.gateway.alerts.destinations).toEqual([slack]); + + for (const secret of ["s", "", null]) { + await expect(loadDestinations([{ ...slack, secret }])).rejects.toThrow( + 'Invalid config value for "gateway.alerts.destinations[0].secret": expected no secret: a slack destination is not signed', + ); + } + await expect( + loadDestinations([{ ...slack, url: "https://user:pw@hooks.slack.com/x" }]), + ).rejects.toThrow( + `Invalid config value for "gateway.alerts.destinations[0].url": ${urlExpected}`, + ); + await expect(loadDestinations([bad("format", "teams")])).rejects.toThrow( + `Invalid config value for "gateway.alerts.destinations[0].format": ${formatExpected}`, + ); + }); + it("loadConfig refuses a repeated name, a bad list and a bad entry, on a gateway and on a worker", async () => { await expect(loadDestinations([validDestination, validDestination])).rejects.toThrow( 'Invalid config value for "gateway.alerts.destinations[1].name": expected a name no other destination uses', @@ -2060,8 +2081,10 @@ describe("gateway.alerts.destinations", () => { expect(destination.shape.format.options).toEqual([...ALERT_FORMATS]); }); - it("config.get's schema keeps every field of a loaded destination", async () => { - const config = await loadDestinations([validDestination]); + it("config.get's schema keeps every field of a loaded destination, a slack one without a secret", async () => { + const { secret: _secret, ...slack } = { ...validDestination, name: "chat", format: "slack" }; + const config = await loadDestinations([validDestination, slack]); + expect(config.gateway.alerts.destinations[1]).toEqual(slack); const parsed = configSchema.parse(config); expect(parsed.gateway.alerts).toEqual(config.gateway.alerts); }); @@ -2089,7 +2112,7 @@ describe("redactConfig", () => { expect(config.gateway.token).toBe("secret-join-token"); }); - it("masks every destination's url and secret, keeps name, format and events, and leaves its input untouched", async () => { + it("masks every destination's url and secret, keeps name, format and events, and leaves its input untouched (a slack destination gets no secret key)", async () => { const destinations = [ { name: "audit", @@ -2105,6 +2128,12 @@ describe("redactConfig", () => { events: ["worker.disconnected"], secret: "other", }, + { + name: "chat", + format: "slack", + url: "https://hooks.slack.com/services/T0/B0/token", + events: ["worker.connected"], + }, ]; for (const token of ["join-token", undefined]) { const config = await load({ @@ -2119,7 +2148,7 @@ describe("redactConfig", () => { destinations.map((destination) => ({ ...destination, url: REDACTED_VALUE, - secret: REDACTED_VALUE, + ...("secret" in destination ? { secret: REDACTED_VALUE } : {}), })), ); expect(redacted.gateway.token).toBe(token === undefined ? undefined : REDACTED_VALUE); diff --git a/src/core/config.ts b/src/core/config.ts index 9140b0f01..1efdb87e2 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -112,12 +112,13 @@ export type AlertEventName = (typeof ALERT_EVENTS)[number]; /** * why: re-declared in the contract's `configSchema`, which imports nothing from core; a test checks the two agree. */ -export const ALERT_FORMATS = ["generic"] as const; +export const ALERT_FORMATS = ["generic", "slack"] as const; export type AlertFormatName = (typeof ALERT_FORMATS)[number]; /** why: checked at config load, which cannot import the alerts module. */ const ALERT_FORMAT_SECRET: Record = { generic: "required", + slack: "refused", }; export interface AlertDestination { @@ -175,6 +176,24 @@ function checkAlertEvents(events: unknown, path: string): AlertEventName[] { return events as AlertEventName[]; } +function checkAlertSecret( + format: AlertFormatName, + entry: Record, + path: string, +): string | undefined { + const { secret } = entry; + if (ALERT_FORMAT_SECRET[format] === "refused") { + if ("secret" in entry) { + throw invalidValue(`${path}.secret`, `no secret: a ${format} destination is not signed`); + } + return undefined; + } + if (typeof secret !== "string" || secret.length === 0) { + throw invalidValue(`${path}.secret`, `a non-empty secret for a ${format} destination`); + } + return secret; +} + function alertDestinationValidator( names: Set, ): (value: unknown, path: string, warn: Warn) => AlertDestination { @@ -188,20 +207,14 @@ function alertDestinationValidator( const format = stringUnion(ALERT_FORMATS)(entry.format, `${path}.format`); if (!isAlertUrl(entry.url)) throw invalidValue(`${path}.url`, ALERT_URL_EXPECTED); const events = checkAlertEvents(entry.events, path); - const { secret } = entry; - if ( - ALERT_FORMAT_SECRET[format] === "required" && - (typeof secret !== "string" || secret.length === 0) - ) { - throw invalidValue(`${path}.secret`, `a non-empty secret for a ${format} destination`); - } + const secret = checkAlertSecret(format, entry, path); names.add(name); return { name, format, url: entry.url, events, - ...(typeof secret === "string" ? { secret } : {}), + ...(secret === undefined ? {} : { secret }), }; }; } diff --git a/src/core/index.ts b/src/core/index.ts index a72177c05..276df58b5 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -2,6 +2,7 @@ export { ALERT_EVENTS, ALERT_FORMATS, type AlertDestination, + type AlertEventName, type AlertFormatName, type Config, type ConfigOverrides, diff --git a/src/daemon-protocol/index.test.ts b/src/daemon-protocol/index.test.ts index edd7d6c36..d1ba19b00 100644 --- a/src/daemon-protocol/index.test.ts +++ b/src/daemon-protocol/index.test.ts @@ -9,7 +9,7 @@ import { describe("daemon protocol", () => { it("keeps the current protocol version and newline framing", () => { - expect(DAEMON_PROTOCOL_VERSION).toBe(49); + expect(DAEMON_PROTOCOL_VERSION).toBe(50); expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n'); }); diff --git a/src/daemon/AGENTS.md b/src/daemon/AGENTS.md index 02992ca77..e005917b7 100644 --- a/src/daemon/AGENTS.md +++ b/src/daemon/AGENTS.md @@ -52,7 +52,7 @@ An HTTP bind failure after a good start stops the daemon (`http-start-failed`). Principal: `hello.principal`, else `SIMLOCK_AGENT_ID`, else the daemon's pid. Both are fixed for the connection's life. 3. **Version** โ€” ranges overlap โ†’ highest common version. `PROTOCOL_VERSION_RANGE` - is `{ min: 49, max: 49 }` (`src/contract/protocol.ts`). No overlap โ†’ the + is `{ min: 50, max: 50 }` (`src/contract/protocol.ts`). No overlap โ†’ the connection stays open, and every request but `daemon.stop` gets `PROTOCOL_VERSION_UNSUPPORTED`. 4. **Request** โ€” one frame โ†’ one `dispatch()` call. While stopping: `DAEMON_STOPPING`. diff --git a/src/daemon/gateway-fleet.e2e.test.ts b/src/daemon/gateway-fleet.e2e.test.ts index a799b45f6..bae4e2038 100644 --- a/src/daemon/gateway-fleet.e2e.test.ts +++ b/src/daemon/gateway-fleet.e2e.test.ts @@ -679,13 +679,17 @@ describe("gateway fleet smoke (ADR 0005 ยง35)", () => { expect(poster.requests.filter((request) => request.url.includes("quick"))).toEqual([]); }, 30_000); - it("a worker joining a gateway with a generic destination makes a loopback receiver get one signed worker.connected naming that worker", async () => { - const received: Array<{ headers: Record; body: string }> = []; + it("a worker joining a gateway with a generic destination makes a loopback receiver get one signed worker.connected naming that worker, and with a slack destination one message naming the worker's label", async () => { + const received: Array<{ url: string; headers: Record; body: string }> = []; const receiver: Server = createHttpServer((request, response) => { const chunks: Buffer[] = []; request.on("data", (chunk: Buffer) => chunks.push(chunk)); request.on("end", () => { - received.push({ headers: request.headers, body: Buffer.concat(chunks).toString("utf8") }); + received.push({ + url: request.url ?? "", + headers: request.headers, + body: Buffer.concat(chunks).toString("utf8"), + }); response.writeHead(200).end(); }); }); @@ -695,6 +699,12 @@ describe("gateway fleet smoke (ADR 0005 ยง35)", () => { const gateway = await startGateway({ destinations: [ destinationFor({ url: `http://127.0.0.1:${port}/hook`, events: ["worker.connected"] }), + { + name: "chat", + format: "slack", + url: `http://127.0.0.1:${port}/slack`, + events: ["worker.connected"], + }, ], }); await startWorker({ @@ -704,8 +714,9 @@ describe("gateway fleet smoke (ADR 0005 ยง35)", () => { stdout: "a", }); - await vi.waitFor(() => expect(received).toHaveLength(1)); - const [message] = received; + await vi.waitFor(() => expect(received).toHaveLength(2)); + const message = received.find((each) => each.url === "/hook"); + const chat = received.find((each) => each.url === "/slack"); expect(message?.headers["x-simlock-signature"]).toBe( `sha256=${createHmac("sha256", "e2e-secret") .update(message?.body ?? "") @@ -716,6 +727,9 @@ describe("gateway fleet smoke (ADR 0005 ยง35)", () => { const { workers } = await gateway.dispatch("worker.list", {}, adminSession()); expect(body.payload.workerId).toBe(workers[0]?.id); expect(body.payload.label).toBe("worker-a"); + expect(JSON.parse(chat?.body ?? "")).toEqual({ + text: "๐ŸŸข Worker worker-a connected (simlock 1.0.0-e2e).", + }); } finally { receiver.closeAllConnections(); await new Promise((resolve) => receiver.close(() => resolve())); diff --git a/src/daemon/main.ts b/src/daemon/main.ts index 40abfc809..700158475 100644 --- a/src/daemon/main.ts +++ b/src/daemon/main.ts @@ -970,6 +970,7 @@ async function startGatewayDaemon(options: GatewayDaemonOptions): Promise gatewayService.workers.view(workerId)?.label, }); } diff --git a/src/daemon/server.test.ts b/src/daemon/server.test.ts index ace071f3d..e059e7140 100644 --- a/src/daemon/server.test.ts +++ b/src/daemon/server.test.ts @@ -221,7 +221,7 @@ describe("DaemonServer", () => { [47, "cannot read config.get's http.proxyAuth"], [48, "cannot read config.get's gateway.alerts"], ])( - "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 49, to a client on protocol %i, which %s", + "answers PROTOCOL_VERSION_UNSUPPORTED, naming protocol 50, to a client on protocol %i, which %s", async (version) => { const harness = await createHarness(); const previous = await createClient(harness.socketPath); @@ -231,7 +231,7 @@ describe("DaemonServer", () => { ).resolves.toMatchObject({ error: { code: "PROTOCOL_VERSION_UNSUPPORTED", - details: { client: { min: version, max: version }, daemon: { min: 49, max: 49 } }, + details: { client: { min: version, max: version }, daemon: { min: 50, max: 50 } }, }, ok: false, }); diff --git a/src/gateway/service.test.ts b/src/gateway/service.test.ts index 4b5d27143..537b7377f 100644 --- a/src/gateway/service.test.ts +++ b/src/gateway/service.test.ts @@ -1265,7 +1265,7 @@ describe("GatewayService", () => { min: PROTOCOL_VERSION_RANGE.max - 1, max: PROTOCOL_VERSION_RANGE.max - 1, }; - expect(PROTOCOL_VERSION_RANGE.max).toBe(49); + expect(PROTOCOL_VERSION_RANGE.max).toBe(50); expect(negotiateProtocolVersion(PROTOCOL_VERSION_RANGE, previous)).toBeUndefined(); const harness = fleet(); await harness.service.start();