Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7caafbaf80 | |||
| bc285799d1 | |||
| dcb2a86f9a | |||
| be739c319f | |||
| 28cf9faf77 |
@@ -0,0 +1,34 @@
|
|||||||
|
name: CI
|
||||||
|
|
||||||
|
# The release workflow gates a tag, which is late: a broken commit sits green
|
||||||
|
# until somebody decides to publish. This runs the same checks on the way in.
|
||||||
|
#
|
||||||
|
# push is scoped to main rather than all branches for two reasons: a branch
|
||||||
|
# pushed as part of a pull request would otherwise be checked twice, and
|
||||||
|
# gh-pages holds the published Helm chart index with no Go code in it, so
|
||||||
|
# `go vet ./...` there would fail on a missing go.mod.
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
branches: [main]
|
||||||
|
pull_request:
|
||||||
|
|
||||||
|
# A rapid series of pushes only needs the last one checked.
|
||||||
|
concurrency:
|
||||||
|
group: ci-${{ github.ref }}
|
||||||
|
cancel-in-progress: true
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
test:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- uses: actions/setup-go@v5
|
||||||
|
with:
|
||||||
|
go-version-file: go.mod
|
||||||
|
|
||||||
|
- name: Vet
|
||||||
|
run: go vet ./...
|
||||||
|
|
||||||
|
- name: Test
|
||||||
|
run: go test ./...
|
||||||
@@ -7,7 +7,25 @@ on:
|
|||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
|
# Gates every publishing job below. A tag that fails here publishes nothing:
|
||||||
|
# the binaries, the image and the chart are all downstream of it.
|
||||||
|
test:
|
||||||
|
runs-on: ubuntu-latest
|
||||||
|
steps:
|
||||||
|
- uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- uses: actions/setup-go@v5
|
||||||
|
with:
|
||||||
|
go-version-file: go.mod
|
||||||
|
|
||||||
|
- name: Vet
|
||||||
|
run: go vet ./...
|
||||||
|
|
||||||
|
- name: Test
|
||||||
|
run: go test ./...
|
||||||
|
|
||||||
build:
|
build:
|
||||||
|
needs: test
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
strategy:
|
strategy:
|
||||||
matrix:
|
matrix:
|
||||||
@@ -44,6 +62,7 @@ jobs:
|
|||||||
path: terdut-${{ github.ref_name }}-${{ matrix.goos }}-${{ matrix.goarch }}
|
path: terdut-${{ github.ref_name }}-${{ matrix.goos }}-${{ matrix.goarch }}
|
||||||
|
|
||||||
docker:
|
docker:
|
||||||
|
needs: test
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
permissions:
|
permissions:
|
||||||
contents: read
|
contents: read
|
||||||
@@ -76,6 +95,7 @@ jobs:
|
|||||||
ghcr.io/yeniklas/terdut-server:${{ github.ref_name }}
|
ghcr.io/yeniklas/terdut-server:${{ github.ref_name }}
|
||||||
|
|
||||||
chart:
|
chart:
|
||||||
|
needs: test
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
permissions:
|
permissions:
|
||||||
contents: write
|
contents: write
|
||||||
|
|||||||
@@ -50,6 +50,46 @@ docker run -p 8080:8080 -v $(pwd)/data:/data \
|
|||||||
terdut-server
|
terdut-server
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### Kubernetes
|
||||||
|
|
||||||
|
A Helm chart is published from this repository:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
helm repo add terdut-server https://yeniklas.github.io/terdut-server
|
||||||
|
helm upgrade --install terdut-server terdut-server/terdut-server \
|
||||||
|
--namespace terdut-server --create-namespace \
|
||||||
|
--set networking.hostname=terdut.example.com
|
||||||
|
```
|
||||||
|
|
||||||
|
The chart expects a [Gateway API](https://gateway-api.sigs.k8s.io/) Gateway named `envoy-main` in
|
||||||
|
the `envoy-gateway-system` namespace to already exist — it renders an `HTTPRoute` against it rather
|
||||||
|
than an `Ingress`. TLS is terminated at the gateway, so the server itself never sees a certificate.
|
||||||
|
|
||||||
|
| Value | Default | Description |
|
||||||
|
|---|---|---|
|
||||||
|
| `networking.hostname` | `terdut.example.com` | Hostname the `HTTPRoute` serves |
|
||||||
|
| `networking.listener` | `""` | Gateway listener (`sectionName`) to bind to. Empty attaches to every matching listener, **including plaintext HTTP** — set it to the HTTPS listener's name to serve TLS only |
|
||||||
|
| `networking.servicePort` | `8080` | Port the route forwards to; keep in sync with `service.port` |
|
||||||
|
| `bootstrap.enabled` | `true` | Runs a post-install hook that creates the first user and stores its API key in the `<release>-admin-key` Secret. Already-bootstrapped servers are left alone |
|
||||||
|
| `backupSidecar.enabled` | `true` | Adds an idle `python` sidecar and the [k8up](https://k8up.io/) annotations that dump the database through it |
|
||||||
|
|
||||||
|
The API key travels in an `Authorization: Bearer` header, so set `networking.listener` whenever the
|
||||||
|
hostname is reachable outside a trusted network.
|
||||||
|
|
||||||
|
#### Backups
|
||||||
|
|
||||||
|
The server image is `FROM scratch` — the binary and nothing else — so there is no interpreter to
|
||||||
|
run a database dump in, and the database runs in WAL mode, where a file-level copy of the volume is
|
||||||
|
not crash-consistent. The chart therefore ships an idle `python:*-alpine` sidecar that shares the
|
||||||
|
data volume, and points k8up's `backupcommand` at it with `k8up.io/backupcommand-container`. Without
|
||||||
|
that annotation k8up execs into `.spec.containers[0]` and the dump fails.
|
||||||
|
|
||||||
|
The dump is buffered and sanity-checked before its first byte reaches stdout, because k8up streams
|
||||||
|
stdout straight into Restic: a dump that dies partway is otherwise stored as a silently truncated
|
||||||
|
snapshot that k8up still reports as successful.
|
||||||
|
|
||||||
|
Set `backupSidecar.enabled=false` if you back the volume up some other way.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## Configuration
|
## Configuration
|
||||||
@@ -60,10 +100,15 @@ docker run -p 8080:8080 -v $(pwd)/data:/data \
|
|||||||
| `TERDUT_DB_PATH` | `terdut.db` | Path to the SQLite database file |
|
| `TERDUT_DB_PATH` | `terdut.db` | Path to the SQLite database file |
|
||||||
| `TERDUT_ARCHIVE_AFTER` | `168h` (7d) | How long a resolved alert or incident stays in the default list before being auto-archived |
|
| `TERDUT_ARCHIVE_AFTER` | `168h` (7d) | How long a resolved alert or incident stays in the default list before being auto-archived |
|
||||||
| `TERDUT_STALE_AFTER` | `6h` | How long a firing alert may go without a refreshing webhook before it is treated as resolved — **must exceed your Alertmanager `repeat_interval`** |
|
| `TERDUT_STALE_AFTER` | `6h` | How long a firing alert may go without a refreshing webhook before it is treated as resolved — **must exceed your Alertmanager `repeat_interval`** |
|
||||||
|
| `TERDUT_NTFY_URL` | — | ntfy server to publish push notifications to. Empty disables notifications entirely |
|
||||||
|
| `TERDUT_NTFY_TOKEN` | — | Bearer token for an access-controlled ntfy |
|
||||||
|
| `TERDUT_NTFY_FALLBACK_TOPIC` | — | Topic used when nobody is on call |
|
||||||
|
| `TERDUT_PUBLIC_URL` | — | Base URL a phone uses to reach this server, for the link and Acknowledge button inside a notification |
|
||||||
|
| `TERDUT_NOTIFY_REPEAT` | `15m` | How long an incident may sit unacknowledged before it is paged again. `0` notifies once and never repeats |
|
||||||
|
|
||||||
Durations use Go syntax (`30m`, `12h`, `168h`). An unparseable value falls back to the default.
|
Durations use Go syntax (`30m`, `12h`, `168h`). An unparseable value falls back to the default.
|
||||||
|
|
||||||
In the Helm chart the two sweeper durations are set via `sweeper.staleAfter` and `sweeper.archiveAfter`.
|
In the Helm chart the two sweeper durations are set via `sweeper.staleAfter` and `sweeper.archiveAfter`, and notifications via the `notify.*` values.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
@@ -145,6 +190,44 @@ A new incident is assigned to whoever holds today's schedule entry at the moment
|
|||||||
it opens (`GET /api/schedule/current`). If nobody is scheduled it opens
|
it opens (`GET /api/schedule/current`). If nobody is scheduled it opens
|
||||||
unassigned. Reassign with `POST /api/incidents/{id}/assign`.
|
unassigned. Reassign with `POST /api/incidents/{id}/assign`.
|
||||||
|
|
||||||
|
### Push notifications
|
||||||
|
|
||||||
|
With `TERDUT_NTFY_URL` set, an incident that opens is pushed to the on-call
|
||||||
|
person's phone through [ntfy](https://ntfy.sh). Set each user's topic with
|
||||||
|
`PUT /api/users/{id}/notify`; a user with no topic falls back to
|
||||||
|
`TERDUT_NTFY_FALLBACK_TOPIC`, as does an incident that opens with nobody on call.
|
||||||
|
If neither yields a topic, nothing is queued.
|
||||||
|
|
||||||
|
Three things get pushed:
|
||||||
|
|
||||||
|
- **triggered** — an incident opened. Priority follows severity (`critical` maps
|
||||||
|
to ntfy's max priority, the one that overrides the phone's quiet settings).
|
||||||
|
- **reminder** — the incident is still `triggered` after `TERDUT_NOTIFY_REPEAT`.
|
||||||
|
Repeats until somebody acts. Acknowledging, snoozing, resolving or archiving
|
||||||
|
all stop it — snooze is the mute button.
|
||||||
|
- **resolved** — every alert under the incident stopped firing. Only sent to
|
||||||
|
whoever was paged in the first place, and only for the automatic cascade:
|
||||||
|
resolving by hand pushes nothing, since the person who did it already knows.
|
||||||
|
|
||||||
|
Notifications carry an **Acknowledge** button that acknowledges the incident
|
||||||
|
without opening anything. It POSTs to `/api/notify/ack/{token}`, an
|
||||||
|
unauthenticated route authorised by the 256-bit single-use token in its path —
|
||||||
|
minted fresh per notification, scoped to one incident and one action, and valid
|
||||||
|
for 24 hours. A real API key is never put in a notification, because the message
|
||||||
|
is stored on the ntfy server and cached on the device.
|
||||||
|
|
||||||
|
Two consequences worth planning for:
|
||||||
|
|
||||||
|
- `/api/notify/ack/{token}` **must stay publicly reachable**, or the button will
|
||||||
|
not work when the responder is off your network.
|
||||||
|
- Notifications sent to the fallback topic carry **no** Acknowledge button. The
|
||||||
|
topic is shared, and a button on it would let any subscriber acknowledge as
|
||||||
|
somebody else.
|
||||||
|
|
||||||
|
Delivery is a queue, not an inline call: the webhook writes a row and a
|
||||||
|
background notifier sends it within 30 seconds, retrying with exponential
|
||||||
|
backoff up to 8 attempts. Nothing about ingestion blocks on ntfy being reachable.
|
||||||
|
|
||||||
### Stale alert expiry
|
### Stale alert expiry
|
||||||
|
|
||||||
A resolved webhook is the only signal that an alert has stopped firing, so a
|
A resolved webhook is the only signal that an alert has stopped firing, so a
|
||||||
@@ -184,6 +267,7 @@ Authorization: Bearer <api-key>
|
|||||||
| `GET` | `/api/users` | List users |
|
| `GET` | `/api/users` | List users |
|
||||||
| `POST` | `/api/users` | Create user `{"username","email"}` |
|
| `POST` | `/api/users` | Create user `{"username","email"}` |
|
||||||
| `DELETE` | `/api/users/{id}` | Delete user (cascades to keys) |
|
| `DELETE` | `/api/users/{id}` | Delete user (cascades to keys) |
|
||||||
|
| `PUT` | `/api/users/{id}/notify` | Set push notification target `{"ntfy_topic"}` — empty string clears it |
|
||||||
| `POST` | `/api/users/{id}/api-keys` | Issue API key `{"name"}` — key shown once |
|
| `POST` | `/api/users/{id}/api-keys` | Issue API key `{"name"}` — key shown once |
|
||||||
| `DELETE` | `/api/users/{id}/api-keys/{keyID}` | Revoke API key |
|
| `DELETE` | `/api/users/{id}/api-keys/{keyID}` | Revoke API key |
|
||||||
|
|
||||||
@@ -193,6 +277,12 @@ Authorization: Bearer <api-key>
|
|||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `POST` | `/api/alertmanager/webhook` | Alertmanager v4 webhook receiver (no auth) |
|
| `POST` | `/api/alertmanager/webhook` | Alertmanager v4 webhook receiver (no auth) |
|
||||||
|
|
||||||
|
### Notifications
|
||||||
|
|
||||||
|
| Method | Path | Description |
|
||||||
|
|---|---|---|
|
||||||
|
| `POST` | `/api/notify/ack/{token}` | Acknowledge an incident from a push notification's Acknowledge button. No auth: the single-use token in the path is the credential. Must stay publicly reachable |
|
||||||
|
|
||||||
### Incidents
|
### Incidents
|
||||||
|
|
||||||
| Method | Path | Description |
|
| Method | Path | Description |
|
||||||
|
|||||||
@@ -2,5 +2,5 @@ apiVersion: v2
|
|||||||
name: terdut-server
|
name: terdut-server
|
||||||
description: A Helm chart for Terminal Duty — on-call alert management server
|
description: A Helm chart for Terminal Duty — on-call alert management server
|
||||||
type: application
|
type: application
|
||||||
version: 0.2.0
|
version: 0.6.0
|
||||||
appVersion: "latest"
|
appVersion: "latest"
|
||||||
|
|||||||
@@ -10,10 +10,50 @@ spec:
|
|||||||
selector:
|
selector:
|
||||||
matchLabels:
|
matchLabels:
|
||||||
{{- include "terdut-server.selectorLabels" . | nindent 6 }}
|
{{- include "terdut-server.selectorLabels" . | nindent 6 }}
|
||||||
|
# The data PVC is ReadWriteOnce, so a RollingUpdate deadlocks: the new pod
|
||||||
|
# cannot attach the volume until the old one releases it, and the old one is
|
||||||
|
# not torn down until the new one is ready.
|
||||||
|
strategy:
|
||||||
|
type: Recreate
|
||||||
template:
|
template:
|
||||||
metadata:
|
metadata:
|
||||||
labels:
|
labels:
|
||||||
{{- include "terdut-server.selectorLabels" . | nindent 8 }}
|
{{- include "terdut-server.selectorLabels" . | nindent 8 }}
|
||||||
|
{{- if .Values.backupSidecar.enabled }}
|
||||||
|
annotations:
|
||||||
|
# Dumps the whole database: incidents, alerts, users, API key hashes,
|
||||||
|
# the schedule and the notification outbox.
|
||||||
|
#
|
||||||
|
# Runs in the `backup` sidecar, NOT in the app container: the server
|
||||||
|
# image is FROM scratch and has no interpreter at all. k8up execs into
|
||||||
|
# .spec.containers[0] unless told otherwise, hence the explicit
|
||||||
|
# k8up.io/backupcommand-container.
|
||||||
|
#
|
||||||
|
# Buffered and sanity-checked before the first byte reaches stdout: k8up
|
||||||
|
# streams stdout straight into restic, so a dump that dies partway is
|
||||||
|
# stored as a silently-truncated snapshot that k8up still reports as
|
||||||
|
# Succeeded. The check counts users rather than incidents -- incidents
|
||||||
|
# are swept and archived, so an empty incidents table is a legitimate
|
||||||
|
# state, whereas a database with no users never is.
|
||||||
|
#
|
||||||
|
# The connection is read-only but the mount is not: the database runs in
|
||||||
|
# WAL mode, and opening it mode=ro still needs write access to the -shm
|
||||||
|
# wal-index.
|
||||||
|
#
|
||||||
|
# chr(10), not '\n': k8up parses this annotation with go-shellquote.
|
||||||
|
k8up.io/backupcommand-container: backup
|
||||||
|
k8up.io/backupcommand: >-
|
||||||
|
python3 -c "import sqlite3, sys;
|
||||||
|
con = sqlite3.connect('file:/data/terdut.db?mode=ro', uri=True);
|
||||||
|
con.execute('BEGIN');
|
||||||
|
users = con.execute('SELECT count(*) FROM users').fetchone()[0];
|
||||||
|
out = chr(10).join(con.iterdump()) + chr(10);
|
||||||
|
(users > 0 and out.rstrip().endswith('COMMIT;'))
|
||||||
|
or sys.exit('terdut: db dump failed sanity checks');
|
||||||
|
sys.stdout.write(out)"
|
||||||
|
k8up.io/file-extension: ".sql"
|
||||||
|
k8up.io/backup: "true"
|
||||||
|
{{- end }}
|
||||||
spec:
|
spec:
|
||||||
enableServiceLinks: false
|
enableServiceLinks: false
|
||||||
containers:
|
containers:
|
||||||
@@ -33,6 +73,23 @@ spec:
|
|||||||
value: "{{ .Values.sweeper.staleAfter }}"
|
value: "{{ .Values.sweeper.staleAfter }}"
|
||||||
- name: TERDUT_ARCHIVE_AFTER
|
- name: TERDUT_ARCHIVE_AFTER
|
||||||
value: "{{ .Values.sweeper.archiveAfter }}"
|
value: "{{ .Values.sweeper.archiveAfter }}"
|
||||||
|
{{- if .Values.notify.ntfyUrl }}
|
||||||
|
- name: TERDUT_NTFY_URL
|
||||||
|
value: "{{ .Values.notify.ntfyUrl }}"
|
||||||
|
- name: TERDUT_NTFY_FALLBACK_TOPIC
|
||||||
|
value: "{{ .Values.notify.fallbackTopic }}"
|
||||||
|
- name: TERDUT_NOTIFY_REPEAT
|
||||||
|
value: "{{ .Values.notify.repeatEvery }}"
|
||||||
|
- name: TERDUT_PUBLIC_URL
|
||||||
|
value: "{{ .Values.notify.publicUrl | default (printf "https://%s" .Values.networking.hostname) }}"
|
||||||
|
{{- if .Values.notify.tokenSecret.name }}
|
||||||
|
- name: TERDUT_NTFY_TOKEN
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: {{ .Values.notify.tokenSecret.name }}
|
||||||
|
key: {{ .Values.notify.tokenSecret.key }}
|
||||||
|
{{- end }}
|
||||||
|
{{- end }}
|
||||||
volumeMounts:
|
volumeMounts:
|
||||||
- name: data
|
- name: data
|
||||||
mountPath: /data
|
mountPath: /data
|
||||||
@@ -46,6 +103,25 @@ spec:
|
|||||||
path: /healthz
|
path: /healthz
|
||||||
port: http
|
port: http
|
||||||
initialDelaySeconds: 5
|
initialDelaySeconds: 5
|
||||||
|
|
||||||
|
{{- if .Values.backupSidecar.enabled }}
|
||||||
|
# Idle sidecar. It exists only so k8up has a container with a sqlite3
|
||||||
|
# module to exec the backupcommand in. Mounted read-write on purpose:
|
||||||
|
# see the note on the backupcommand annotation above.
|
||||||
|
- name: backup
|
||||||
|
image: "{{ .Values.backupSidecar.image.repository }}:{{ .Values.backupSidecar.image.tag }}"
|
||||||
|
imagePullPolicy: {{ .Values.backupSidecar.image.pullPolicy }}
|
||||||
|
command: ["sleep", "infinity"]
|
||||||
|
volumeMounts:
|
||||||
|
- name: data
|
||||||
|
mountPath: /data
|
||||||
|
resources:
|
||||||
|
requests:
|
||||||
|
memory: "16Mi"
|
||||||
|
cpu: "10m"
|
||||||
|
limits:
|
||||||
|
memory: "64Mi"
|
||||||
|
{{- end }}
|
||||||
volumes:
|
volumes:
|
||||||
- name: data
|
- name: data
|
||||||
persistentVolumeClaim:
|
persistentVolumeClaim:
|
||||||
|
|||||||
@@ -9,6 +9,9 @@ spec:
|
|||||||
parentRefs:
|
parentRefs:
|
||||||
- name: envoy-main
|
- name: envoy-main
|
||||||
namespace: envoy-gateway-system
|
namespace: envoy-gateway-system
|
||||||
|
{{- with .Values.networking.listener }}
|
||||||
|
sectionName: {{ . | quote }}
|
||||||
|
{{- end }}
|
||||||
rules:
|
rules:
|
||||||
- backendRefs:
|
- backendRefs:
|
||||||
- name: {{ include "terdut-server.fullname" . }}
|
- name: {{ include "terdut-server.fullname" . }}
|
||||||
|
|||||||
@@ -1,6 +1,10 @@
|
|||||||
networking:
|
networking:
|
||||||
hostname: "terdut.example.com"
|
hostname: "terdut.example.com"
|
||||||
servicePort: 8080
|
servicePort: 8080
|
||||||
|
# Gateway listener to bind the HTTPRoute to. Empty attaches to every matching
|
||||||
|
# listener, including plaintext HTTP. Set this to the name of the HTTPS
|
||||||
|
# listener to serve the API over TLS only.
|
||||||
|
listener: ""
|
||||||
|
|
||||||
image:
|
image:
|
||||||
repository: ghcr.io/yeniklas/terdut-server
|
repository: ghcr.io/yeniklas/terdut-server
|
||||||
@@ -22,6 +26,42 @@ sweeper:
|
|||||||
# How long a resolved alert stays in the default list before auto-archiving.
|
# How long a resolved alert stays in the default list before auto-archiving.
|
||||||
archiveAfter: 168h
|
archiveAfter: 168h
|
||||||
|
|
||||||
|
notify:
|
||||||
|
# ntfy server that push notifications are published to, e.g.
|
||||||
|
# http://ntfy.ntfy.svc.cluster.local. Empty disables notifications entirely.
|
||||||
|
ntfyUrl: ""
|
||||||
|
# Topic used when nobody is on call today. Notifications sent here carry no
|
||||||
|
# Acknowledge button: the topic is shared, so there is no user to attribute an
|
||||||
|
# acknowledgement to. Leave empty to send nothing when the schedule is unset.
|
||||||
|
fallbackTopic: ""
|
||||||
|
# How long an incident may sit unacknowledged before it is paged again.
|
||||||
|
# Set to 0 to notify once and never repeat.
|
||||||
|
repeatEvery: 15m
|
||||||
|
# Base URL a phone uses to reach this server, for the link and the Acknowledge
|
||||||
|
# button inside a notification. Defaults to https://<networking.hostname>.
|
||||||
|
#
|
||||||
|
# The Acknowledge button is a POST to /api/notify/ack/{token} from the
|
||||||
|
# responder's phone, so that path has to stay publicly reachable — it is
|
||||||
|
# authorised by the single-use token in the URL, not by network placement.
|
||||||
|
publicUrl: ""
|
||||||
|
# Optional bearer token for an access-controlled ntfy, read from an existing
|
||||||
|
# Secret. Leave name empty for an open ntfy.
|
||||||
|
tokenSecret:
|
||||||
|
name: ""
|
||||||
|
key: token
|
||||||
|
|
||||||
|
# The server image is FROM scratch — just the binary, with no shell, no sqlite3
|
||||||
|
# and no python — so a k8up backupcommand cannot run in the app container. This
|
||||||
|
# idle sidecar shares the data volume and is selected with
|
||||||
|
# k8up.io/backupcommand-container. Only the stdlib sqlite3 module is used, so any
|
||||||
|
# python image works.
|
||||||
|
backupSidecar:
|
||||||
|
enabled: true
|
||||||
|
image:
|
||||||
|
repository: python
|
||||||
|
tag: "3.13-alpine"
|
||||||
|
pullPolicy: IfNotPresent
|
||||||
|
|
||||||
bootstrap:
|
bootstrap:
|
||||||
enabled: true
|
enabled: true
|
||||||
username: admin
|
username: admin
|
||||||
|
|||||||
+10
-1
@@ -28,7 +28,15 @@ func main() {
|
|||||||
log.Fatalf("migrate: %v", err)
|
log.Fatalf("migrate: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
router := api.NewRouter(database)
|
notify := api.NotifyConfig{
|
||||||
|
BaseURL: cfg.NtfyURL,
|
||||||
|
Token: cfg.NtfyToken,
|
||||||
|
FallbackTopic: cfg.NtfyFallbackTopic,
|
||||||
|
PublicURL: cfg.PublicURL,
|
||||||
|
RepeatEvery: cfg.NotifyRepeat,
|
||||||
|
}
|
||||||
|
|
||||||
|
router := api.NewRouter(database, notify)
|
||||||
|
|
||||||
srv := &http.Server{
|
srv := &http.Server{
|
||||||
Addr: cfg.Addr,
|
Addr: cfg.Addr,
|
||||||
@@ -42,6 +50,7 @@ func main() {
|
|||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
go api.StartArchiver(ctx, database, cfg.ArchiveAfter, cfg.StaleAfter)
|
go api.StartArchiver(ctx, database, cfg.ArchiveAfter, cfg.StaleAfter)
|
||||||
|
go api.StartNotifier(ctx, database, notify)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
log.Printf("terdut-server %s listening on %s", version, cfg.Addr)
|
log.Printf("terdut-server %s listening on %s", version, cfg.Addr)
|
||||||
|
|||||||
@@ -58,7 +58,7 @@ type ingested struct {
|
|||||||
justResolved bool
|
justResolved bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc {
|
func handleAlertmanagerWebhook(db *sql.DB, notify NotifyConfig) http.HandlerFunc {
|
||||||
return func(w http.ResponseWriter, r *http.Request) {
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
var payload amPayload
|
var payload amPayload
|
||||||
if err := decodeJSON(r, &payload); err != nil {
|
if err := decodeJSON(r, &payload); err != nil {
|
||||||
@@ -69,7 +69,7 @@ func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc {
|
|||||||
// Alertmanager retries anything that is not 2xx, and a retry of a payload
|
// Alertmanager retries anything that is not 2xx, and a retry of a payload
|
||||||
// we failed to store is more useful than an error it cannot act on — so
|
// we failed to store is more useful than an error it cannot act on — so
|
||||||
// failures are logged, not surfaced.
|
// failures are logged, not surfaced.
|
||||||
if err := ingest(r.Context(), db, payload); err != nil {
|
if err := ingest(r.Context(), db, notify, payload); err != nil {
|
||||||
log.Printf("webhook ingest (group %q): %v", payload.GroupKey, err)
|
log.Printf("webhook ingest (group %q): %v", payload.GroupKey, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -80,7 +80,7 @@ func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc {
|
|||||||
// ingest stores a payload's alerts and reconciles the incident for its group.
|
// ingest stores a payload's alerts and reconciles the incident for its group.
|
||||||
// The whole payload is one transaction: an incident that opened but whose alerts
|
// The whole payload is one transaction: an incident that opened but whose alerts
|
||||||
// failed to link would be a work item nobody could act on.
|
// failed to link would be a work item nobody could act on.
|
||||||
func ingest(ctx context.Context, db *sql.DB, payload amPayload) error {
|
func ingest(ctx context.Context, db *sql.DB, notify NotifyConfig, payload amPayload) error {
|
||||||
tx, err := db.BeginTx(ctx, nil)
|
tx, err := db.BeginTx(ctx, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -96,7 +96,7 @@ func ingest(ctx context.Context, db *sql.DB, payload amPayload) error {
|
|||||||
// resolution cascade are recomputed once per incident at the end.
|
// resolution cascade are recomputed once per incident at the end.
|
||||||
touched := map[int64]bool{}
|
touched := map[int64]bool{}
|
||||||
|
|
||||||
incidentID, err := incidentForGroup(ctx, tx, payload, accepted)
|
incidentID, err := incidentForGroup(ctx, tx, notify, payload, accepted)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -258,7 +258,7 @@ func upsertAlerts(ctx context.Context, tx *sql.Tx, alerts []amAlert) ([]ingested
|
|||||||
// something actually started firing. Without that, a manually resolved incident
|
// something actually started firing. Without that, a manually resolved incident
|
||||||
// would reappear on the next repeat_interval re-send of an alert that never
|
// would reappear on the next repeat_interval re-send of an alert that never
|
||||||
// stopped, and manual resolution would be meaningless.
|
// stopped, and manual resolution would be meaningless.
|
||||||
func incidentForGroup(ctx context.Context, tx *sql.Tx, payload amPayload, accepted []ingested) (int64, error) {
|
func incidentForGroup(ctx context.Context, tx *sql.Tx, notify NotifyConfig, payload amPayload, accepted []ingested) (int64, error) {
|
||||||
var firstName string
|
var firstName string
|
||||||
anyFiring, anyNew := false, false
|
anyFiring, anyNew := false, false
|
||||||
for _, a := range accepted {
|
for _, a := range accepted {
|
||||||
@@ -297,12 +297,12 @@ func incidentForGroup(ctx context.Context, tx *sql.Tx, payload amPayload, accept
|
|||||||
if !anyNew {
|
if !anyNew {
|
||||||
return 0, nil
|
return 0, nil
|
||||||
}
|
}
|
||||||
return openIncident(ctx, tx, groupKey, payload.GroupLabels, firstName)
|
return openIncident(ctx, tx, notify, groupKey, payload.GroupLabels, firstName)
|
||||||
}
|
}
|
||||||
|
|
||||||
// openIncident creates an incident for a group and assigns it to whoever is on
|
// openIncident creates an incident for a group and assigns it to whoever is on
|
||||||
// call today, which is the point at which the schedule stops being decorative.
|
// call today, which is the point at which the schedule stops being decorative.
|
||||||
func openIncident(ctx context.Context, tx *sql.Tx, groupKey string, groupLabels map[string]string, fallbackName string) (int64, error) {
|
func openIncident(ctx context.Context, tx *sql.Tx, notify NotifyConfig, groupKey string, groupLabels map[string]string, fallbackName string) (int64, error) {
|
||||||
onCall, err := currentOnCall(ctx, tx)
|
onCall, err := currentOnCall(ctx, tx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
@@ -335,6 +335,13 @@ func openIncident(ctx context.Context, tx *sql.Tx, groupKey string, groupLabels
|
|||||||
return 0, err
|
return 0, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Queue the page, but do not send it here: this runs inside the webhook's
|
||||||
|
// transaction on a single-connection pool, so an HTTP call would hold up
|
||||||
|
// every other request. The notifier picks the row up within a tick.
|
||||||
|
if err := enqueueOpened(ctx, tx, notify, id, onCall); err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
return id, nil
|
return id, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -22,10 +22,18 @@ type ts struct {
|
|||||||
*httptest.Server
|
*httptest.Server
|
||||||
key string
|
key string
|
||||||
db *sql.DB
|
db *sql.DB
|
||||||
|
notify api.NotifyConfig
|
||||||
}
|
}
|
||||||
|
|
||||||
func newTS(t *testing.T) *ts {
|
// newTS builds a server over a fresh in-memory database. Notifications are off
|
||||||
|
// unless a NotifyConfig is passed, so tests that predate them are unaffected.
|
||||||
|
func newTS(t *testing.T, notify ...api.NotifyConfig) *ts {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
var cfg api.NotifyConfig
|
||||||
|
if len(notify) > 0 {
|
||||||
|
cfg = notify[0]
|
||||||
|
}
|
||||||
|
|
||||||
database, err := db.Open(":memory:")
|
database, err := db.Open(":memory:")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("open db: %v", err)
|
t.Fatalf("open db: %v", err)
|
||||||
@@ -33,7 +41,7 @@ func newTS(t *testing.T) *ts {
|
|||||||
if err := db.Migrate(database); err != nil {
|
if err := db.Migrate(database); err != nil {
|
||||||
t.Fatalf("migrate: %v", err)
|
t.Fatalf("migrate: %v", err)
|
||||||
}
|
}
|
||||||
srv := httptest.NewServer(api.NewRouter(database))
|
srv := httptest.NewServer(api.NewRouter(database, cfg))
|
||||||
t.Cleanup(func() { srv.Close(); database.Close() })
|
t.Cleanup(func() { srv.Close(); database.Close() })
|
||||||
|
|
||||||
body, _ := json.Marshal(map[string]string{"username": "admin", "email": "admin@test.com"})
|
body, _ := json.Marshal(map[string]string{"username": "admin", "email": "admin@test.com"})
|
||||||
@@ -49,7 +57,7 @@ func newTS(t *testing.T) *ts {
|
|||||||
json.NewDecoder(resp.Body).Decode(&result)
|
json.NewDecoder(resp.Body).Decode(&result)
|
||||||
key := result["api_key"].(map[string]any)["key"].(string)
|
key := result["api_key"].(map[string]any)["key"].(string)
|
||||||
|
|
||||||
return &ts{Server: srv, key: key, db: database}
|
return &ts{Server: srv, key: key, db: database, notify: cfg}
|
||||||
}
|
}
|
||||||
|
|
||||||
// exec runs a statement against the test database.
|
// exec runs a statement against the test database.
|
||||||
|
|||||||
@@ -44,6 +44,7 @@ func Sweep(ctx context.Context, db *sql.DB, archiveAfter, staleAfter time.Durati
|
|||||||
resolveSettledIncidents(ctx, db)
|
resolveSettledIncidents(ctx, db)
|
||||||
archiveResolved(ctx, db, archiveAfter)
|
archiveResolved(ctx, db, archiveAfter)
|
||||||
archiveResolvedIncidents(ctx, db, archiveAfter)
|
archiveResolvedIncidents(ctx, db, archiveAfter)
|
||||||
|
purgeAckTokens(ctx, db)
|
||||||
}
|
}
|
||||||
|
|
||||||
// expireStale resolves firing alerts that Alertmanager has stopped refreshing.
|
// expireStale resolves firing alerts that Alertmanager has stopped refreshing.
|
||||||
|
|||||||
@@ -223,7 +223,32 @@ func resolveIfSettled(ctx context.Context, q querier, incidentID int64) (bool, e
|
|||||||
if n == 0 {
|
if n == 0 {
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
return true, logEvent(ctx, q, incidentID, evResolved, nil, nil, nil)
|
if err := logEvent(ctx, q, incidentID, evResolved, nil, nil, nil); err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
// The all-clear goes only to whoever was paged in the first place, which
|
||||||
|
// enqueueResolved works out from the incident's own notification history.
|
||||||
|
// Manual resolution sends nothing: the person who closed it already knows.
|
||||||
|
return true, enqueueResolved(ctx, q, incidentID)
|
||||||
|
}
|
||||||
|
|
||||||
|
// acknowledgeIncident records that userID has picked an incident up, and reports
|
||||||
|
// whether it changed anything — an already-resolved incident is left alone.
|
||||||
|
// Shared by the authenticated handler and the Acknowledge button in a push
|
||||||
|
// notification, so both write the same state and the same timeline entry.
|
||||||
|
func acknowledgeIncident(ctx context.Context, q querier, incidentID, userID int64) (bool, error) {
|
||||||
|
res, err := q.ExecContext(ctx, `
|
||||||
|
UPDATE incidents
|
||||||
|
SET status = 'acknowledged', acknowledged_by = ?, acknowledged_at = ?
|
||||||
|
WHERE id = ? AND resolved_at IS NULL`,
|
||||||
|
userID, time.Now().Unix(), incidentID)
|
||||||
|
if err != nil {
|
||||||
|
return false, err
|
||||||
|
}
|
||||||
|
if n, _ := res.RowsAffected(); n == 0 {
|
||||||
|
return false, nil
|
||||||
|
}
|
||||||
|
return true, logEvent(ctx, q, incidentID, evAcknowledged, &userID, nil, nil)
|
||||||
}
|
}
|
||||||
|
|
||||||
// openIncidentForAlert returns the open incident an alert currently belongs to,
|
// openIncidentForAlert returns the open incident an alert currently belongs to,
|
||||||
|
|||||||
@@ -189,13 +189,16 @@ func handleIncidentAcknowledge(db *sql.DB) http.HandlerFunc {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
user, _ := userFromContext(r.Context())
|
user, _ := userFromContext(r.Context())
|
||||||
if !updateOpenIncident(w, r, db, id,
|
acked, err := acknowledgeIncident(r.Context(), db, id, user.ID)
|
||||||
`UPDATE incidents SET status = 'acknowledged', acknowledged_by = ?, acknowledged_at = ?
|
if err != nil {
|
||||||
WHERE id = ? AND resolved_at IS NULL`, user.ID, time.Now().Unix(), id) {
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if err := logEvent(r.Context(), db, id, evAcknowledged, &user.ID, nil, nil); err != nil {
|
if !acked {
|
||||||
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
if !incidentExists(w, r, db, id) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
respond(w, http.StatusConflict, errResp("incident is resolved"))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
respondIncident(w, r, db, id)
|
respondIncident(w, r, db, id)
|
||||||
|
|||||||
@@ -0,0 +1,532 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
"net/http"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/yeniklas/terdut-server/internal/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
// notifyInterval is how often the notifier looks for work. The archiver's
|
||||||
|
// 15 minute tick is far too coarse for something that has to wake a person.
|
||||||
|
notifyInterval = 30 * time.Second
|
||||||
|
|
||||||
|
// notifyRetryBase and notifyRetryMax bound the delivery backoff. ntfy being
|
||||||
|
// briefly unreachable should not lose the page.
|
||||||
|
notifyRetryBase = 30 * time.Second
|
||||||
|
notifyRetryMax = 15 * time.Minute
|
||||||
|
|
||||||
|
// notifyMaxAttempts stops a permanently undeliverable row from being retried
|
||||||
|
// forever. It keeps last_error so the reason survives.
|
||||||
|
notifyMaxAttempts = 8
|
||||||
|
|
||||||
|
// notifyBatch caps one delivery pass, so a large backlog cannot hold the
|
||||||
|
// single database connection for an unbounded stretch.
|
||||||
|
notifyBatch = 100
|
||||||
|
|
||||||
|
// ackTokenTTL is how long the Acknowledge button in a notification keeps
|
||||||
|
// working. Past this the notification is stale enough that the responder
|
||||||
|
// should look at the incident rather than blind-acknowledge it.
|
||||||
|
ackTokenTTL = 24 * time.Hour
|
||||||
|
)
|
||||||
|
|
||||||
|
// Notification kinds, recording why a push was sent.
|
||||||
|
const (
|
||||||
|
notifyTriggered = "triggered"
|
||||||
|
notifyReminder = "reminder"
|
||||||
|
notifyResolved = "resolved"
|
||||||
|
)
|
||||||
|
|
||||||
|
// NotifyConfig is everything the notifier needs to reach ntfy and to build URLs
|
||||||
|
// a phone can follow back to this server.
|
||||||
|
type NotifyConfig struct {
|
||||||
|
// BaseURL is the ntfy server. Empty disables notifications entirely: no
|
||||||
|
// goroutine, and nothing is ever enqueued.
|
||||||
|
BaseURL string
|
||||||
|
|
||||||
|
// Token is an optional bearer token for an access-controlled ntfy.
|
||||||
|
Token string
|
||||||
|
|
||||||
|
// FallbackTopic receives incidents that open with nobody on call. Those
|
||||||
|
// notifications carry no Acknowledge button — there is no user to attribute
|
||||||
|
// the acknowledgement to, and putting one on a shared topic would let any
|
||||||
|
// subscriber acknowledge as somebody else.
|
||||||
|
FallbackTopic string
|
||||||
|
|
||||||
|
// PublicURL is the base URL a phone uses to reach this server, for the
|
||||||
|
// notification's click target and its Acknowledge action. Without it a
|
||||||
|
// notification is informational only.
|
||||||
|
PublicURL string
|
||||||
|
|
||||||
|
// RepeatEvery is how long an incident may sit unacknowledged before it is
|
||||||
|
// notified again. Zero disables reminders.
|
||||||
|
RepeatEvery time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// enabled reports whether notifications are configured at all.
|
||||||
|
func (c NotifyConfig) enabled() bool { return c.BaseURL != "" }
|
||||||
|
|
||||||
|
// notifyClient is shared: a page is small and infrequent, and the timeout is
|
||||||
|
// what keeps a hung ntfy from stalling the delivery pass.
|
||||||
|
var notifyClient = &http.Client{Timeout: 10 * time.Second}
|
||||||
|
|
||||||
|
// StartNotifier delivers queued notifications until ctx is cancelled, starting
|
||||||
|
// with an immediate pass so a restart flushes whatever the last one left behind.
|
||||||
|
func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||||
|
if !cfg.enabled() {
|
||||||
|
log.Print("notifier: disabled (no ntfy URL configured)")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
log.Printf("notifier: publishing to %s", cfg.BaseURL)
|
||||||
|
|
||||||
|
ticker := time.NewTicker(notifyInterval)
|
||||||
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
NotifySweep(ctx, db, cfg)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ticker.C:
|
||||||
|
NotifySweep(ctx, db, cfg)
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// NotifySweep runs a single pass: queue reminders for incidents nobody has
|
||||||
|
// picked up, then deliver everything that is due. Reminders are queued first so
|
||||||
|
// a freshly due one goes out in the same pass rather than a tick later.
|
||||||
|
// Exported so tests can drive a pass without waiting on the ticker.
|
||||||
|
func NotifySweep(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||||
|
enqueueReminders(ctx, db, cfg)
|
||||||
|
deliverPending(ctx, db, cfg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// enqueueReminders re-notifies incidents that are still sitting untouched.
|
||||||
|
//
|
||||||
|
// The stop conditions are the incident states that already mean "somebody has
|
||||||
|
// this": acknowledged, snoozed, resolved, archived. Snooze in particular is the
|
||||||
|
// mute button — a deliberate "not now" that should not keep buzzing — which is
|
||||||
|
// why there is no separate reminder cap.
|
||||||
|
//
|
||||||
|
// The previous notification must have actually been sent before another is
|
||||||
|
// queued, so an ntfy outage produces a retry backlog rather than a reminder
|
||||||
|
// backlog that all lands at once when it comes back.
|
||||||
|
func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||||
|
if cfg.RepeatEvery <= 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
now := time.Now()
|
||||||
|
|
||||||
|
type due struct {
|
||||||
|
incidentID int64
|
||||||
|
userID *int64
|
||||||
|
topic string
|
||||||
|
}
|
||||||
|
|
||||||
|
rows, err := db.QueryContext(ctx, `
|
||||||
|
SELECT n.incident_id, n.user_id, n.topic
|
||||||
|
FROM notifications n
|
||||||
|
JOIN incidents i ON i.id = n.incident_id
|
||||||
|
WHERE n.id = (SELECT MAX(id) FROM notifications WHERE incident_id = n.incident_id)
|
||||||
|
AND n.sent_at IS NOT NULL
|
||||||
|
AND n.created_at <= ?
|
||||||
|
AND i.resolved_at IS NULL
|
||||||
|
AND i.archived_at IS NULL
|
||||||
|
AND i.status = 'triggered'
|
||||||
|
AND (i.snoozed_until IS NULL OR i.snoozed_until <= ?)`,
|
||||||
|
now.Add(-cfg.RepeatEvery).Unix(), now.Unix())
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("notifier: find reminders: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Collected before inserting: the pool is limited to a single connection, so
|
||||||
|
// an open cursor would block the writes behind it.
|
||||||
|
var pending []due
|
||||||
|
for rows.Next() {
|
||||||
|
var d due
|
||||||
|
if err := rows.Scan(&d.incidentID, &d.userID, &d.topic); err != nil {
|
||||||
|
rows.Close()
|
||||||
|
log.Printf("notifier: scan reminder: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
pending = append(pending, d)
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
rows.Close()
|
||||||
|
log.Printf("notifier: find reminders: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
rows.Close()
|
||||||
|
|
||||||
|
for _, d := range pending {
|
||||||
|
if err := enqueueNotification(ctx, db, d.incidentID, d.userID, d.topic, notifyReminder); err != nil {
|
||||||
|
log.Printf("notifier: queue reminder for incident %d: %v", d.incidentID, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(pending) > 0 {
|
||||||
|
log.Printf("notifier: queued %d reminder(s)", len(pending))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// outboxRow is one queued notification, read before any HTTP happens.
|
||||||
|
type outboxRow struct {
|
||||||
|
id int64
|
||||||
|
incidentID int64
|
||||||
|
userID *int64
|
||||||
|
topic string
|
||||||
|
kind string
|
||||||
|
attempts int
|
||||||
|
}
|
||||||
|
|
||||||
|
// deliverPending sends everything that is due and records the outcome.
|
||||||
|
func deliverPending(ctx context.Context, db *sql.DB, cfg NotifyConfig) {
|
||||||
|
batch, err := pendingNotifications(ctx, db)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("notifier: find pending: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
sent := 0
|
||||||
|
for _, n := range batch {
|
||||||
|
if err := deliver(ctx, db, cfg, n); err != nil {
|
||||||
|
log.Printf("notifier: deliver %d (incident %d): %v", n.id, n.incidentID, err)
|
||||||
|
markFailed(ctx, db, n, err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if _, err := db.ExecContext(ctx,
|
||||||
|
"UPDATE notifications SET sent_at = ?, attempts = attempts + 1, last_error = NULL WHERE id = ?",
|
||||||
|
time.Now().Unix(), n.id); err != nil {
|
||||||
|
log.Printf("notifier: mark sent %d: %v", n.id, err)
|
||||||
|
}
|
||||||
|
sent++
|
||||||
|
}
|
||||||
|
if sent > 0 {
|
||||||
|
log.Printf("notifier: delivered %d notification(s)", sent)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// pendingNotifications reads the due rows and closes the cursor before the
|
||||||
|
// caller writes, for the same single-connection reason as staleAlertIDs.
|
||||||
|
func pendingNotifications(ctx context.Context, db *sql.DB) ([]outboxRow, error) {
|
||||||
|
rows, err := db.QueryContext(ctx, `
|
||||||
|
SELECT id, incident_id, user_id, topic, kind, attempts
|
||||||
|
FROM notifications
|
||||||
|
WHERE sent_at IS NULL
|
||||||
|
AND send_after <= ?
|
||||||
|
AND attempts < ?
|
||||||
|
ORDER BY id
|
||||||
|
LIMIT ?`, time.Now().Unix(), notifyMaxAttempts, notifyBatch)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
var batch []outboxRow
|
||||||
|
for rows.Next() {
|
||||||
|
var n outboxRow
|
||||||
|
if err := rows.Scan(&n.id, &n.incidentID, &n.userID, &n.topic, &n.kind, &n.attempts); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
batch = append(batch, n)
|
||||||
|
}
|
||||||
|
return batch, rows.Err()
|
||||||
|
}
|
||||||
|
|
||||||
|
// markFailed bumps the attempt count and pushes the row out to its next retry.
|
||||||
|
func markFailed(ctx context.Context, db *sql.DB, n outboxRow, cause error) {
|
||||||
|
next := time.Now().Add(retryDelay(n.attempts)).Unix()
|
||||||
|
if _, err := db.ExecContext(ctx,
|
||||||
|
"UPDATE notifications SET attempts = attempts + 1, send_after = ?, last_error = ? WHERE id = ?",
|
||||||
|
next, cause.Error(), n.id); err != nil {
|
||||||
|
log.Printf("notifier: mark failed %d: %v", n.id, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// retryDelay doubles the wait per attempt, up to notifyRetryMax.
|
||||||
|
func retryDelay(attempts int) time.Duration {
|
||||||
|
d := notifyRetryBase << attempts
|
||||||
|
if d > notifyRetryMax || d <= 0 {
|
||||||
|
return notifyRetryMax
|
||||||
|
}
|
||||||
|
return d
|
||||||
|
}
|
||||||
|
|
||||||
|
// deliver renders one notification against the incident's *current* state and
|
||||||
|
// publishes it. Rendering happens here rather than at enqueue time so a message
|
||||||
|
// that waited in the queue while its incident escalated goes out at the
|
||||||
|
// severity the incident has now.
|
||||||
|
func deliver(ctx context.Context, db *sql.DB, cfg NotifyConfig, n outboxRow) error {
|
||||||
|
inc, err := fetchIncident(ctx, db, n.incidentID)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("load incident: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var firing int
|
||||||
|
if err := db.QueryRowContext(ctx, `
|
||||||
|
SELECT COUNT(*)
|
||||||
|
FROM incident_alerts ia
|
||||||
|
JOIN alerts a ON a.id = ia.alert_id
|
||||||
|
WHERE ia.incident_id = ? AND a.status = 'firing'`, n.incidentID).Scan(&firing); err != nil {
|
||||||
|
return fmt.Errorf("count firing: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
msg := renderNotification(inc, n, firing, cfg)
|
||||||
|
|
||||||
|
// An Acknowledge button needs both a user to attribute the acknowledgement
|
||||||
|
// to and a URL the phone can reach. Minted per delivery, so every push
|
||||||
|
// carries its own short-lived token rather than reusing one.
|
||||||
|
if n.kind != notifyResolved && n.userID != nil && cfg.PublicURL != "" {
|
||||||
|
raw, err := issueAckToken(ctx, db, n.incidentID, *n.userID)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("issue ack token: %w", err)
|
||||||
|
}
|
||||||
|
msg.Actions = append(msg.Actions, ntfyAction{
|
||||||
|
Action: "http",
|
||||||
|
Label: "Acknowledge",
|
||||||
|
URL: strings.TrimSuffix(cfg.PublicURL, "/") + "/api/notify/ack/" + raw,
|
||||||
|
Method: "POST",
|
||||||
|
Clear: true,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
return publish(ctx, cfg, msg)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ntfyMessage is ntfy's JSON publish format. Using it rather than the X-Actions
|
||||||
|
// header avoids that header's comma and quote escaping rules, which are easy to
|
||||||
|
// break with a title that happens to contain a comma.
|
||||||
|
type ntfyMessage struct {
|
||||||
|
Topic string `json:"topic"`
|
||||||
|
Title string `json:"title,omitempty"`
|
||||||
|
Message string `json:"message"`
|
||||||
|
Priority int `json:"priority,omitempty"`
|
||||||
|
Tags []string `json:"tags,omitempty"`
|
||||||
|
Click string `json:"click,omitempty"`
|
||||||
|
Actions []ntfyAction `json:"actions,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
type ntfyAction struct {
|
||||||
|
Action string `json:"action"`
|
||||||
|
Label string `json:"label"`
|
||||||
|
URL string `json:"url"`
|
||||||
|
Method string `json:"method,omitempty"`
|
||||||
|
Clear bool `json:"clear,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// renderNotification builds the message body for one queued notification.
|
||||||
|
func renderNotification(inc models.Incident, n outboxRow, firing int, cfg NotifyConfig) ntfyMessage {
|
||||||
|
msg := ntfyMessage{Topic: n.topic}
|
||||||
|
|
||||||
|
if cfg.PublicURL != "" {
|
||||||
|
msg.Click = fmt.Sprintf("%s/api/incidents/%d",
|
||||||
|
strings.TrimSuffix(cfg.PublicURL, "/"), inc.ID)
|
||||||
|
}
|
||||||
|
|
||||||
|
switch n.kind {
|
||||||
|
case notifyResolved:
|
||||||
|
msg.Title = "Resolved: " + inc.Title
|
||||||
|
msg.Message = "All alerts stopped firing after " +
|
||||||
|
humanDuration(time.Since(inc.TriggeredAt))
|
||||||
|
msg.Priority = ntfyPriorityLow
|
||||||
|
msg.Tags = []string{"white_check_mark"}
|
||||||
|
return msg
|
||||||
|
|
||||||
|
case notifyReminder:
|
||||||
|
msg.Title = "Still unacknowledged: " + inc.Title
|
||||||
|
default:
|
||||||
|
msg.Title = inc.Title
|
||||||
|
}
|
||||||
|
|
||||||
|
severity := derefString(inc.Severity)
|
||||||
|
|
||||||
|
parts := []string{fmt.Sprintf("%d alert%s firing", firing, plural(firing))}
|
||||||
|
if severity != "" {
|
||||||
|
parts = append(parts, "severity "+severity)
|
||||||
|
}
|
||||||
|
if assignee := derefString(inc.AssignedToUser); assignee != "" {
|
||||||
|
parts = append(parts, "on call: "+assignee)
|
||||||
|
}
|
||||||
|
if n.kind == notifyReminder {
|
||||||
|
parts = append(parts, "open "+humanDuration(time.Since(inc.TriggeredAt)))
|
||||||
|
}
|
||||||
|
|
||||||
|
msg.Message = strings.Join(parts, " · ")
|
||||||
|
msg.Priority = ntfyPriority(severity)
|
||||||
|
msg.Tags = []string{severityTag(severity)}
|
||||||
|
return msg
|
||||||
|
}
|
||||||
|
|
||||||
|
// ntfy's priority scale. Max is the one that overrides the phone's quiet
|
||||||
|
// settings, which is the whole point of paging on critical.
|
||||||
|
const (
|
||||||
|
ntfyPriorityLow = 2
|
||||||
|
ntfyPriorityDefault = 3
|
||||||
|
ntfyPriorityHigh = 4
|
||||||
|
ntfyPriorityMax = 5
|
||||||
|
)
|
||||||
|
|
||||||
|
// ntfyPriority maps an incident's severity onto ntfy's scale, following the
|
||||||
|
// same ordering severityRank uses. An unrecognised severity gets the default
|
||||||
|
// rather than being silenced.
|
||||||
|
func ntfyPriority(severity string) int {
|
||||||
|
switch severityRank(severity) {
|
||||||
|
case 4:
|
||||||
|
return ntfyPriorityMax
|
||||||
|
case 3:
|
||||||
|
return ntfyPriorityHigh
|
||||||
|
case 2:
|
||||||
|
return ntfyPriorityDefault
|
||||||
|
case 1:
|
||||||
|
return ntfyPriorityLow
|
||||||
|
default:
|
||||||
|
return ntfyPriorityDefault
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func severityTag(severity string) string {
|
||||||
|
switch severityRank(severity) {
|
||||||
|
case 4:
|
||||||
|
return "rotating_light"
|
||||||
|
case 3:
|
||||||
|
return "red_circle"
|
||||||
|
case 2:
|
||||||
|
return "warning"
|
||||||
|
case 1:
|
||||||
|
return "information_source"
|
||||||
|
default:
|
||||||
|
return "bell"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// publish POSTs one message to ntfy.
|
||||||
|
func publish(ctx context.Context, cfg NotifyConfig, msg ntfyMessage) error {
|
||||||
|
body, err := json.Marshal(msg)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
|
||||||
|
strings.TrimSuffix(cfg.BaseURL, "/"), bytes.NewReader(body))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
if cfg.Token != "" {
|
||||||
|
req.Header.Set("Authorization", "Bearer "+cfg.Token)
|
||||||
|
}
|
||||||
|
|
||||||
|
resp, err := notifyClient.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||||
|
return fmt.Errorf("ntfy returned %s", resp.Status)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// enqueueNotification adds one row to the outbox, due immediately.
|
||||||
|
func enqueueNotification(ctx context.Context, q querier, incidentID int64, userID *int64, topic, kind string) error {
|
||||||
|
now := time.Now().Unix()
|
||||||
|
_, err := q.ExecContext(ctx, `
|
||||||
|
INSERT INTO notifications (incident_id, user_id, topic, kind, created_at, send_after)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?)`, incidentID, userID, topic, kind, now, now)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// notifyTarget decides where a newly opened incident's notification goes.
|
||||||
|
//
|
||||||
|
// The on-call user's own topic when they have one, otherwise the fallback
|
||||||
|
// topic with no user attached. Deliberately not "the fallback topic, attributed
|
||||||
|
// to the on-call user": the fallback is shared, and an Acknowledge button on a
|
||||||
|
// shared topic would let any subscriber acknowledge as somebody else.
|
||||||
|
func notifyTarget(ctx context.Context, q querier, cfg NotifyConfig, onCall *int64) (topic string, userID *int64) {
|
||||||
|
if onCall != nil {
|
||||||
|
var t *string
|
||||||
|
err := q.QueryRowContext(ctx,
|
||||||
|
"SELECT ntfy_topic FROM users WHERE id = ?", *onCall).Scan(&t)
|
||||||
|
if err == nil && t != nil && *t != "" {
|
||||||
|
return *t, onCall
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return cfg.FallbackTopic, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// enqueueOpened queues the notification for a freshly opened incident. It is the
|
||||||
|
// only enqueue point that has to resolve a topic from scratch; every later
|
||||||
|
// notification for the incident reuses what this one chose.
|
||||||
|
func enqueueOpened(ctx context.Context, q querier, cfg NotifyConfig, incidentID int64, onCall *int64) error {
|
||||||
|
if !cfg.enabled() {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
topic, userID := notifyTarget(ctx, q, cfg, onCall)
|
||||||
|
if topic == "" {
|
||||||
|
// Nobody on call has a topic and there is no fallback: there is nowhere
|
||||||
|
// to send this, and queueing it would only accumulate undeliverable rows.
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return enqueueNotification(ctx, q, incidentID, userID, topic, notifyTriggered)
|
||||||
|
}
|
||||||
|
|
||||||
|
// enqueueResolved queues the all-clear, reusing the topic the incident's last
|
||||||
|
// notification went to. That needs no configuration to reach this function, and
|
||||||
|
// it gives the right rule for free: you only hear that something resolved if you
|
||||||
|
// were told it started.
|
||||||
|
func enqueueResolved(ctx context.Context, q querier, incidentID int64) error {
|
||||||
|
var topic string
|
||||||
|
var userID *int64
|
||||||
|
err := q.QueryRowContext(ctx, `
|
||||||
|
SELECT topic, user_id FROM notifications
|
||||||
|
WHERE incident_id = ? ORDER BY id DESC LIMIT 1`, incidentID).Scan(&topic, &userID)
|
||||||
|
if err == sql.ErrNoRows {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return enqueueNotification(ctx, q, incidentID, userID, topic, notifyResolved)
|
||||||
|
}
|
||||||
|
|
||||||
|
// humanDuration renders an age the way a person reads it at 3am: coarse, and
|
||||||
|
// never more than two units.
|
||||||
|
func humanDuration(d time.Duration) string {
|
||||||
|
if d < time.Minute {
|
||||||
|
return "less than a minute"
|
||||||
|
}
|
||||||
|
if d < time.Hour {
|
||||||
|
return fmt.Sprintf("%dm", int(d.Minutes()))
|
||||||
|
}
|
||||||
|
h := int(d.Hours())
|
||||||
|
m := int(d.Minutes()) - h*60
|
||||||
|
if m == 0 {
|
||||||
|
return fmt.Sprintf("%dh", h)
|
||||||
|
}
|
||||||
|
return fmt.Sprintf("%dh%dm", h, m)
|
||||||
|
}
|
||||||
|
|
||||||
|
func plural(n int) string {
|
||||||
|
if n == 1 {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return "s"
|
||||||
|
}
|
||||||
|
|
||||||
|
// derefString reads a nullable text column as a plain string.
|
||||||
|
func derefString(s *string) string {
|
||||||
|
if s == nil {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
return *s
|
||||||
|
}
|
||||||
@@ -0,0 +1,98 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"crypto/sha256"
|
||||||
|
"database/sql"
|
||||||
|
"encoding/hex"
|
||||||
|
"log"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-chi/chi/v5"
|
||||||
|
)
|
||||||
|
|
||||||
|
// issueAckToken mints the secret behind one notification's Acknowledge button
|
||||||
|
// and returns the raw value to embed in its URL. Only the hash is stored, the
|
||||||
|
// same way api_keys works.
|
||||||
|
//
|
||||||
|
// A fresh token per delivery rather than one per incident: the raw value only
|
||||||
|
// exists for as long as it takes to build the message, so there is nothing to
|
||||||
|
// look up and reuse later, and a reminder that supersedes an earlier page
|
||||||
|
// carries its own credential.
|
||||||
|
func issueAckToken(ctx context.Context, q querier, incidentID, userID int64) (string, error) {
|
||||||
|
raw, hash, err := randomToken()
|
||||||
|
if err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
now := time.Now()
|
||||||
|
if _, err := q.ExecContext(ctx, `
|
||||||
|
INSERT INTO incident_ack_tokens (token_hash, incident_id, user_id, created_at, expires_at)
|
||||||
|
VALUES (?, ?, ?, ?, ?)`,
|
||||||
|
hash, incidentID, userID, now.Unix(), now.Add(ackTokenTTL).Unix()); err != nil {
|
||||||
|
return "", err
|
||||||
|
}
|
||||||
|
return raw, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// handleNotifyAck acknowledges an incident from the Acknowledge button in a
|
||||||
|
// push notification.
|
||||||
|
//
|
||||||
|
// It is deliberately outside AuthMiddleware: the caller is a phone acting on a
|
||||||
|
// notification, not a client holding an API key. What stands in for the key is
|
||||||
|
// the token in the path — 256 bits of entropy, valid for one incident, one
|
||||||
|
// action, and one day. It must stay publicly reachable for the button to work
|
||||||
|
// when the responder is off the cluster network.
|
||||||
|
func handleNotifyAck(db *sql.DB) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
h := sha256.Sum256([]byte(chi.URLParam(r, "token")))
|
||||||
|
hash := hex.EncodeToString(h[:])
|
||||||
|
|
||||||
|
var incidentID, userID int64
|
||||||
|
err := db.QueryRowContext(r.Context(), `
|
||||||
|
SELECT incident_id, user_id FROM incident_ack_tokens
|
||||||
|
WHERE token_hash = ? AND expires_at > ?`,
|
||||||
|
hash, time.Now().Unix()).Scan(&incidentID, &userID)
|
||||||
|
if err != nil {
|
||||||
|
// Unknown and expired get the same answer, so the endpoint cannot be
|
||||||
|
// used to probe which tokens once existed.
|
||||||
|
respond(w, http.StatusNotFound, errResp("invalid or expired token"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
acked, err := acknowledgeIncident(r.Context(), db, incidentID, userID)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !acked {
|
||||||
|
// The incident closed between the page and the tap. Nothing to do,
|
||||||
|
// and nothing the responder did wrong — report the state, not an error,
|
||||||
|
// so ntfy shows a success toast rather than a failure.
|
||||||
|
respond(w, http.StatusOK, map[string]any{
|
||||||
|
"incident_id": incidentID,
|
||||||
|
"status": "resolved",
|
||||||
|
})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
respond(w, http.StatusOK, map[string]any{
|
||||||
|
"incident_id": incidentID,
|
||||||
|
"status": "acknowledged",
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// purgeAckTokens drops tokens whose notifications are long past. Nothing else
|
||||||
|
// deletes them: incidents are archived rather than removed, so the cascade never
|
||||||
|
// fires in practice.
|
||||||
|
func purgeAckTokens(ctx context.Context, db *sql.DB) {
|
||||||
|
res, err := db.ExecContext(ctx,
|
||||||
|
"DELETE FROM incident_ack_tokens WHERE expires_at < ?", time.Now().Unix())
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("sweeper: purge ack tokens: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if n, _ := res.RowsAffected(); n > 0 {
|
||||||
|
log.Printf("sweeper: purged %d expired ack token(s)", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,580 @@
|
|||||||
|
package api_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/yeniklas/terdut-server/internal/api"
|
||||||
|
)
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Fake ntfy
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
// pushed is one message the fake ntfy received, in ntfy's JSON publish shape.
|
||||||
|
type pushed struct {
|
||||||
|
Topic string `json:"topic"`
|
||||||
|
Title string `json:"title"`
|
||||||
|
Message string `json:"message"`
|
||||||
|
Priority int `json:"priority"`
|
||||||
|
Tags []string `json:"tags"`
|
||||||
|
Click string `json:"click"`
|
||||||
|
Actions []struct {
|
||||||
|
Action string `json:"action"`
|
||||||
|
Label string `json:"label"`
|
||||||
|
URL string `json:"url"`
|
||||||
|
Method string `json:"method"`
|
||||||
|
Clear bool `json:"clear"`
|
||||||
|
} `json:"actions"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// fakeNtfy records what the notifier published. status controls the reply, so a
|
||||||
|
// test can make delivery fail.
|
||||||
|
type fakeNtfy struct {
|
||||||
|
*httptest.Server
|
||||||
|
mu sync.Mutex
|
||||||
|
got []pushed
|
||||||
|
status int
|
||||||
|
}
|
||||||
|
|
||||||
|
func newFakeNtfy(t *testing.T) *fakeNtfy {
|
||||||
|
t.Helper()
|
||||||
|
f := &fakeNtfy{status: http.StatusOK}
|
||||||
|
f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
var msg pushed
|
||||||
|
if err := json.NewDecoder(r.Body).Decode(&msg); err != nil {
|
||||||
|
http.Error(w, "bad json", http.StatusBadRequest)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
f.mu.Lock()
|
||||||
|
f.got = append(f.got, msg)
|
||||||
|
status := f.status
|
||||||
|
f.mu.Unlock()
|
||||||
|
w.WriteHeader(status)
|
||||||
|
}))
|
||||||
|
t.Cleanup(f.Close)
|
||||||
|
return f
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *fakeNtfy) messages() []pushed {
|
||||||
|
f.mu.Lock()
|
||||||
|
defer f.mu.Unlock()
|
||||||
|
return append([]pushed(nil), f.got...)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (f *fakeNtfy) failWith(status int) {
|
||||||
|
f.mu.Lock()
|
||||||
|
defer f.mu.Unlock()
|
||||||
|
f.status = status
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Harness
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
// notifyTS builds a server with notifications enabled, the admin on call today,
|
||||||
|
// and a topic on the admin — the setup every delivery test needs.
|
||||||
|
func notifyTS(t *testing.T, cfg api.NotifyConfig) (*ts, *fakeNtfy) {
|
||||||
|
t.Helper()
|
||||||
|
f := newFakeNtfy(t)
|
||||||
|
cfg.BaseURL = f.URL
|
||||||
|
s := newTS(t, cfg)
|
||||||
|
|
||||||
|
putOnCall(t, s, 1)
|
||||||
|
setTopic(t, s, 1, "terdut-admin")
|
||||||
|
return s, f
|
||||||
|
}
|
||||||
|
|
||||||
|
func putOnCall(t *testing.T, s *ts, userID int) {
|
||||||
|
t.Helper()
|
||||||
|
today := time.Now().UTC().Format("2006-01-02")
|
||||||
|
resp := s.req(t, http.MethodPost, "/api/schedule",
|
||||||
|
map[string]any{"user_id": userID, "dates": []string{today}})
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode != http.StatusCreated {
|
||||||
|
t.Fatalf("schedule assignment returned %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func setTopic(t *testing.T, s *ts, userID int, topic string) {
|
||||||
|
t.Helper()
|
||||||
|
resp := s.req(t, http.MethodPut,
|
||||||
|
fmt.Sprintf("/api/users/%d/notify", userID), map[string]any{"ntfy_topic": topic})
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("set notify topic returned %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *ts) sweepNotify(t *testing.T) {
|
||||||
|
t.Helper()
|
||||||
|
api.NotifySweep(context.Background(), s.db, s.notify)
|
||||||
|
}
|
||||||
|
|
||||||
|
// countNotifications reports how many outbox rows exist, optionally of one kind.
|
||||||
|
func (s *ts) countNotifications(t *testing.T, kind string) int {
|
||||||
|
t.Helper()
|
||||||
|
var n int
|
||||||
|
query := "SELECT COUNT(*) FROM notifications"
|
||||||
|
args := []any{}
|
||||||
|
if kind != "" {
|
||||||
|
query += " WHERE kind = ?"
|
||||||
|
args = append(args, kind)
|
||||||
|
}
|
||||||
|
if err := s.db.QueryRow(query, args...).Scan(&n); err != nil {
|
||||||
|
t.Fatalf("count notifications: %v", err)
|
||||||
|
}
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
|
||||||
|
// fireCritical posts a single critical alert, which opens one incident.
|
||||||
|
func fireCritical(t *testing.T, s *ts) {
|
||||||
|
t.Helper()
|
||||||
|
postWebhook(t, s, []map[string]any{
|
||||||
|
amAlert("fp-notify", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime,
|
||||||
|
map[string]string{"severity": "critical"}),
|
||||||
|
}, "{}:{alertname=\"DiskFull\"}")
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Delivery
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
func TestNotify_TriggeredIncidentPagesOnCall(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
|
||||||
|
if got := s.countNotifications(t, "triggered"); got != 1 {
|
||||||
|
t.Fatalf("expected 1 queued notification, got %d", got)
|
||||||
|
}
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
msgs := f.messages()
|
||||||
|
if len(msgs) != 1 {
|
||||||
|
t.Fatalf("expected 1 push, got %d", len(msgs))
|
||||||
|
}
|
||||||
|
m := msgs[0]
|
||||||
|
|
||||||
|
if m.Topic != "terdut-admin" {
|
||||||
|
t.Errorf("expected the on-call user's topic, got %q", m.Topic)
|
||||||
|
}
|
||||||
|
if m.Priority != 5 {
|
||||||
|
t.Errorf("expected max priority for a critical incident, got %d", m.Priority)
|
||||||
|
}
|
||||||
|
if !strings.Contains(m.Title, "DiskFull") {
|
||||||
|
t.Errorf("expected the incident title in %q", m.Title)
|
||||||
|
}
|
||||||
|
if !strings.Contains(m.Message, "severity critical") {
|
||||||
|
t.Errorf("expected the severity in %q", m.Message)
|
||||||
|
}
|
||||||
|
if m.Click != "https://terdut.example.com/api/incidents/1" {
|
||||||
|
t.Errorf("unexpected click target %q", m.Click)
|
||||||
|
}
|
||||||
|
if len(m.Actions) != 1 || m.Actions[0].Label != "Acknowledge" {
|
||||||
|
t.Fatalf("expected an Acknowledge action, got %+v", m.Actions)
|
||||||
|
}
|
||||||
|
if m.Actions[0].Method != http.MethodPost {
|
||||||
|
t.Errorf("expected the action to POST, got %q", m.Actions[0].Method)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A delivered row must not be delivered again on the next pass.
|
||||||
|
func TestNotify_DeliveredOnlyOnce(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if got := len(f.messages()); got != 1 {
|
||||||
|
t.Errorf("expected 1 push across two passes, got %d", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Acknowledging from the notification
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
func TestNotify_AckButtonAcknowledgesIncident(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
ackURL := f.messages()[0].Actions[0].URL
|
||||||
|
// The action URL is built for the public hostname; point it at the test
|
||||||
|
// server, which is the same handler.
|
||||||
|
path := ackURL[strings.Index(ackURL, "/api/notify/ack/"):]
|
||||||
|
|
||||||
|
resp, err := http.Post(s.URL+path, "application/json", nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ack: %v", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("expected 200 from the ack button, got %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
|
||||||
|
inc := getIncident(t, s, 1)
|
||||||
|
if inc["status"] != "acknowledged" {
|
||||||
|
t.Errorf("expected the incident acknowledged, got %v", inc["status"])
|
||||||
|
}
|
||||||
|
if inc["acknowledged_by"] != "admin" {
|
||||||
|
t.Errorf("expected the ack attributed to the token's user, got %v", inc["acknowledged_by"])
|
||||||
|
}
|
||||||
|
|
||||||
|
// The timeline must record it the same way the authenticated route would.
|
||||||
|
events := timeline(t, s, 1)
|
||||||
|
if !contains(eventTypes(events), "acknowledged") {
|
||||||
|
t.Errorf("expected an acknowledged event, got %v", eventTypes(events))
|
||||||
|
}
|
||||||
|
for _, e := range events {
|
||||||
|
if e["type"] == "acknowledged" && e["username"] != "admin" {
|
||||||
|
t.Errorf("expected the acknowledged event attributed to admin, got %v", e["username"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNotify_AckRejectsUnknownToken(t *testing.T) {
|
||||||
|
s, _ := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
fireCritical(t, s)
|
||||||
|
|
||||||
|
resp, err := http.Post(s.URL+"/api/notify/ack/deadbeef", "application/json", nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ack: %v", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode != http.StatusNotFound {
|
||||||
|
t.Errorf("expected 404 for an unknown token, got %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
if inc := getIncident(t, s, 1); inc["status"] != "triggered" {
|
||||||
|
t.Errorf("expected the incident untouched, got %v", inc["status"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNotify_AckRejectsExpiredToken(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
ackURL := f.messages()[0].Actions[0].URL
|
||||||
|
path := ackURL[strings.Index(ackURL, "/api/notify/ack/"):]
|
||||||
|
|
||||||
|
// Age the token past its TTL. The token's inputs are wall-clock timestamps,
|
||||||
|
// so this is the same trick the sweeper tests use.
|
||||||
|
s.exec(t, "UPDATE incident_ack_tokens SET expires_at = ?", time.Now().Add(-time.Minute).Unix())
|
||||||
|
|
||||||
|
resp, err := http.Post(s.URL+path, "application/json", nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ack: %v", err)
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
if resp.StatusCode != http.StatusNotFound {
|
||||||
|
t.Errorf("expected 404 for an expired token, got %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
if inc := getIncident(t, s, 1); inc["status"] != "triggered" {
|
||||||
|
t.Errorf("expected the incident untouched, got %v", inc["status"])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The sweeper is what stops expired tokens accumulating forever.
|
||||||
|
func TestNotify_SweepPurgesExpiredAckTokens(t *testing.T) {
|
||||||
|
s, _ := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
s.exec(t, "UPDATE incident_ack_tokens SET expires_at = ?", time.Now().Add(-time.Minute).Unix())
|
||||||
|
|
||||||
|
api.Sweep(context.Background(), s.db, 168*time.Hour, 6*time.Hour)
|
||||||
|
|
||||||
|
var n int
|
||||||
|
if err := s.db.QueryRow("SELECT COUNT(*) FROM incident_ack_tokens").Scan(&n); err != nil {
|
||||||
|
t.Fatalf("count tokens: %v", err)
|
||||||
|
}
|
||||||
|
if n != 0 {
|
||||||
|
t.Errorf("expected expired tokens purged, %d left", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Reminders
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
// ageNotifications backdates every sent notification so the next pass sees the
|
||||||
|
// reminder as due.
|
||||||
|
func (s *ts) ageNotifications(t *testing.T, by time.Duration) {
|
||||||
|
t.Helper()
|
||||||
|
s.exec(t, "UPDATE notifications SET created_at = ? WHERE sent_at IS NOT NULL",
|
||||||
|
time.Now().Add(-by).Unix())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNotify_UnacknowledgedIncidentIsRenotified(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{
|
||||||
|
PublicURL: "https://terdut.example.com",
|
||||||
|
RepeatEvery: 15 * time.Minute,
|
||||||
|
})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
s.ageNotifications(t, 20*time.Minute)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
msgs := f.messages()
|
||||||
|
if len(msgs) != 2 {
|
||||||
|
t.Fatalf("expected a reminder push, got %d message(s)", len(msgs))
|
||||||
|
}
|
||||||
|
if !strings.Contains(msgs[1].Title, "Still unacknowledged") {
|
||||||
|
t.Errorf("expected the reminder to say so, got %q", msgs[1].Title)
|
||||||
|
}
|
||||||
|
if msgs[1].Topic != "terdut-admin" {
|
||||||
|
t.Errorf("expected the reminder on the same topic, got %q", msgs[1].Topic)
|
||||||
|
}
|
||||||
|
// Each page carries its own credential.
|
||||||
|
if len(msgs[1].Actions) != 1 || msgs[1].Actions[0].URL == msgs[0].Actions[0].URL {
|
||||||
|
t.Errorf("expected the reminder to carry a fresh ack token")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNotify_AcknowledgedIncidentStopsReminders(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{
|
||||||
|
PublicURL: "https://terdut.example.com",
|
||||||
|
RepeatEvery: 15 * time.Minute,
|
||||||
|
})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
resp := s.req(t, http.MethodPost, "/api/incidents/1/acknowledge", nil)
|
||||||
|
resp.Body.Close()
|
||||||
|
|
||||||
|
s.ageNotifications(t, 20*time.Minute)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if got := len(f.messages()); got != 1 {
|
||||||
|
t.Errorf("expected no reminder once acknowledged, got %d message(s)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Snooze is the deliberate "not now", and it is what mutes the pager.
|
||||||
|
func TestNotify_SnoozedIncidentStopsReminders(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{
|
||||||
|
PublicURL: "https://terdut.example.com",
|
||||||
|
RepeatEvery: 15 * time.Minute,
|
||||||
|
})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
resp := s.req(t, http.MethodPost, "/api/incidents/1/snooze", map[string]any{"duration": "1h"})
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("snooze returned %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
resp.Body.Close()
|
||||||
|
|
||||||
|
s.ageNotifications(t, 20*time.Minute)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if got := len(f.messages()); got != 1 {
|
||||||
|
t.Errorf("expected no reminder while snoozed, got %d message(s)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNotify_ZeroRepeatDisablesReminders(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
s.ageNotifications(t, 24*time.Hour)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if got := len(f.messages()); got != 1 {
|
||||||
|
t.Errorf("expected reminders off, got %d message(s)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Resolution
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
func TestNotify_ResolvedIncidentSendsAllClear(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
postWebhook(t, s, []map[string]any{
|
||||||
|
amAlert("fp-notify", "DiskFull", "resolved", "2026-05-20T10:00:00Z",
|
||||||
|
"2026-05-20T11:00:00Z", map[string]string{"severity": "critical"}),
|
||||||
|
}, "{}:{alertname=\"DiskFull\"}")
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
msgs := f.messages()
|
||||||
|
if len(msgs) != 2 {
|
||||||
|
t.Fatalf("expected an all-clear push, got %d message(s)", len(msgs))
|
||||||
|
}
|
||||||
|
if !strings.HasPrefix(msgs[1].Title, "Resolved:") {
|
||||||
|
t.Errorf("expected a resolved title, got %q", msgs[1].Title)
|
||||||
|
}
|
||||||
|
if msgs[1].Priority != 2 {
|
||||||
|
t.Errorf("expected the all-clear at low priority, got %d", msgs[1].Priority)
|
||||||
|
}
|
||||||
|
// Nothing to acknowledge on a closed incident.
|
||||||
|
if len(msgs[1].Actions) != 0 {
|
||||||
|
t.Errorf("expected no actions on the all-clear, got %+v", msgs[1].Actions)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Closing an incident by hand sends nothing: the person who did it knows.
|
||||||
|
func TestNotify_ManualResolveSendsNothing(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
resp := s.req(t, http.MethodPost, "/api/incidents/1/resolve", nil)
|
||||||
|
resp.Body.Close()
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if got := len(f.messages()); got != 1 {
|
||||||
|
t.Errorf("expected no push for a manual resolve, got %d message(s)", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Routing and configuration
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
// With nobody on call the page goes to the shared fallback, and carries no
|
||||||
|
// Acknowledge button — there is no user to attribute the acknowledgement to.
|
||||||
|
func TestNotify_FallbackTopicHasNoAckButton(t *testing.T) {
|
||||||
|
f := newFakeNtfy(t)
|
||||||
|
s := newTS(t, api.NotifyConfig{
|
||||||
|
BaseURL: f.URL,
|
||||||
|
FallbackTopic: "terdut-oncall",
|
||||||
|
PublicURL: "https://terdut.example.com",
|
||||||
|
})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
msgs := f.messages()
|
||||||
|
if len(msgs) != 1 {
|
||||||
|
t.Fatalf("expected 1 push, got %d", len(msgs))
|
||||||
|
}
|
||||||
|
if msgs[0].Topic != "terdut-oncall" {
|
||||||
|
t.Errorf("expected the fallback topic, got %q", msgs[0].Topic)
|
||||||
|
}
|
||||||
|
if len(msgs[0].Actions) != 0 {
|
||||||
|
t.Errorf("expected no ack button on a shared topic, got %+v", msgs[0].Actions)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Nobody on call and no fallback means there is nowhere to send: queueing would
|
||||||
|
// only pile up rows that can never be delivered.
|
||||||
|
func TestNotify_NoTargetQueuesNothing(t *testing.T) {
|
||||||
|
f := newFakeNtfy(t)
|
||||||
|
s := newTS(t, api.NotifyConfig{BaseURL: f.URL})
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
|
||||||
|
if got := s.countNotifications(t, ""); got != 0 {
|
||||||
|
t.Errorf("expected nothing queued without a target, got %d", got)
|
||||||
|
}
|
||||||
|
s.sweepNotify(t)
|
||||||
|
if got := len(f.messages()); got != 0 {
|
||||||
|
t.Errorf("expected no push, got %d", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The zero NotifyConfig is what every pre-existing test runs under.
|
||||||
|
func TestNotify_DisabledQueuesNothing(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
putOnCall(t, s, 1)
|
||||||
|
setTopic(t, s, 1, "terdut-admin")
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
|
||||||
|
if got := s.countNotifications(t, ""); got != 0 {
|
||||||
|
t.Errorf("expected nothing queued with notifications off, got %d", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Retries
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
func TestNotify_FailedDeliveryRetriesWithBackoff(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com"})
|
||||||
|
f.failWith(http.StatusInternalServerError)
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
var attempts int
|
||||||
|
var sentAt *int64
|
||||||
|
var sendAfter int64
|
||||||
|
var lastError *string
|
||||||
|
if err := s.db.QueryRow(
|
||||||
|
"SELECT attempts, sent_at, send_after, last_error FROM notifications WHERE id = 1").
|
||||||
|
Scan(&attempts, &sentAt, &sendAfter, &lastError); err != nil {
|
||||||
|
t.Fatalf("read notification: %v", err)
|
||||||
|
}
|
||||||
|
if attempts != 1 {
|
||||||
|
t.Errorf("expected 1 attempt recorded, got %d", attempts)
|
||||||
|
}
|
||||||
|
if sentAt != nil {
|
||||||
|
t.Errorf("expected the row unsent, got sent_at %v", *sentAt)
|
||||||
|
}
|
||||||
|
if sendAfter <= time.Now().Unix() {
|
||||||
|
t.Errorf("expected the retry pushed into the future, got %d", sendAfter)
|
||||||
|
}
|
||||||
|
if lastError == nil || !strings.Contains(*lastError, "500") {
|
||||||
|
t.Errorf("expected the failure recorded, got %v", lastError)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Backing off means the next pass leaves it alone until it is due.
|
||||||
|
s.sweepNotify(t)
|
||||||
|
if got := len(f.messages()); got != 1 {
|
||||||
|
t.Errorf("expected no immediate retry, got %d attempt(s)", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Once due and once ntfy recovers, it goes out.
|
||||||
|
f.failWith(http.StatusOK)
|
||||||
|
s.exec(t, "UPDATE notifications SET send_after = ? WHERE id = 1", time.Now().Add(-time.Second).Unix())
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if err := s.db.QueryRow("SELECT sent_at FROM notifications WHERE id = 1").Scan(&sentAt); err != nil {
|
||||||
|
t.Fatalf("read notification: %v", err)
|
||||||
|
}
|
||||||
|
if sentAt == nil {
|
||||||
|
t.Error("expected the retry to succeed once ntfy recovered")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// An ntfy outage must not produce a reminder backlog that all lands at once
|
||||||
|
// when it comes back: the previous page has to have been sent first.
|
||||||
|
func TestNotify_UnsentNotificationBlocksReminders(t *testing.T) {
|
||||||
|
s, f := notifyTS(t, api.NotifyConfig{
|
||||||
|
PublicURL: "https://terdut.example.com",
|
||||||
|
RepeatEvery: 15 * time.Minute,
|
||||||
|
})
|
||||||
|
f.failWith(http.StatusInternalServerError)
|
||||||
|
|
||||||
|
fireCritical(t, s)
|
||||||
|
s.sweepNotify(t)
|
||||||
|
s.exec(t, "UPDATE notifications SET created_at = ?", time.Now().Add(-time.Hour).Unix())
|
||||||
|
s.sweepNotify(t)
|
||||||
|
|
||||||
|
if got := s.countNotifications(t, "reminder"); got != 0 {
|
||||||
|
t.Errorf("expected no reminders queued behind an undelivered page, got %d", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
+11
-3
@@ -8,7 +8,10 @@ import (
|
|||||||
"github.com/go-chi/chi/v5/middleware"
|
"github.com/go-chi/chi/v5/middleware"
|
||||||
)
|
)
|
||||||
|
|
||||||
func NewRouter(db *sql.DB) http.Handler {
|
// NewRouter builds the HTTP surface. notify is passed through to the webhook,
|
||||||
|
// the only handler that has to decide where a new incident's page goes; a zero
|
||||||
|
// value disables notifications.
|
||||||
|
func NewRouter(db *sql.DB, notify NotifyConfig) http.Handler {
|
||||||
r := chi.NewRouter()
|
r := chi.NewRouter()
|
||||||
r.Use(middleware.Logger)
|
r.Use(middleware.Logger)
|
||||||
r.Use(middleware.Recoverer)
|
r.Use(middleware.Recoverer)
|
||||||
@@ -17,9 +20,13 @@ func NewRouter(db *sql.DB) http.Handler {
|
|||||||
respond(w, http.StatusOK, map[string]string{"status": "ok"})
|
respond(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||||
})
|
})
|
||||||
|
|
||||||
// Unauthenticated: bootstrap and Alertmanager webhook receiver.
|
// Unauthenticated: bootstrap, the Alertmanager webhook receiver, and the
|
||||||
|
// Acknowledge button in a push notification. The last one is authorised by
|
||||||
|
// the single-use token in its path rather than an API key, and has to stay
|
||||||
|
// reachable from outside the cluster for the button to work.
|
||||||
r.Post("/api/bootstrap", handleBootstrap(db))
|
r.Post("/api/bootstrap", handleBootstrap(db))
|
||||||
r.Post("/api/alertmanager/webhook", handleAlertmanagerWebhook(db))
|
r.Post("/api/alertmanager/webhook", handleAlertmanagerWebhook(db, notify))
|
||||||
|
r.Post("/api/notify/ack/{token}", handleNotifyAck(db))
|
||||||
|
|
||||||
// All other /api routes require a valid API key.
|
// All other /api routes require a valid API key.
|
||||||
r.Group(func(r chi.Router) {
|
r.Group(func(r chi.Router) {
|
||||||
@@ -28,6 +35,7 @@ func NewRouter(db *sql.DB) http.Handler {
|
|||||||
r.Get("/api/users", handleListUsers(db))
|
r.Get("/api/users", handleListUsers(db))
|
||||||
r.Post("/api/users", handleCreateUser(db))
|
r.Post("/api/users", handleCreateUser(db))
|
||||||
r.Delete("/api/users/{id}", handleDeleteUser(db))
|
r.Delete("/api/users/{id}", handleDeleteUser(db))
|
||||||
|
r.Put("/api/users/{id}/notify", handleSetNotifyTarget(db))
|
||||||
r.Post("/api/users/{id}/api-keys", handleCreateAPIKey(db))
|
r.Post("/api/users/{id}/api-keys", handleCreateAPIKey(db))
|
||||||
r.Delete("/api/users/{id}/api-keys/{keyID}", handleDeleteAPIKey(db))
|
r.Delete("/api/users/{id}/api-keys/{keyID}", handleDeleteAPIKey(db))
|
||||||
|
|
||||||
|
|||||||
+54
-8
@@ -48,7 +48,7 @@ func handleBootstrap(db *sql.DB) http.HandlerFunc {
|
|||||||
}
|
}
|
||||||
userID, _ := res.LastInsertId()
|
userID, _ := res.LastInsertId()
|
||||||
|
|
||||||
raw, hash, err := newAPIKey()
|
raw, hash, err := randomToken()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
return
|
return
|
||||||
@@ -70,7 +70,7 @@ func handleBootstrap(db *sql.DB) http.HandlerFunc {
|
|||||||
func handleListUsers(db *sql.DB) http.HandlerFunc {
|
func handleListUsers(db *sql.DB) http.HandlerFunc {
|
||||||
return func(w http.ResponseWriter, r *http.Request) {
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
rows, err := db.QueryContext(r.Context(),
|
rows, err := db.QueryContext(r.Context(),
|
||||||
"SELECT id, username, email, created_at FROM users ORDER BY id")
|
"SELECT id, username, email, created_at, ntfy_topic FROM users ORDER BY id")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
return
|
return
|
||||||
@@ -81,7 +81,7 @@ func handleListUsers(db *sql.DB) http.HandlerFunc {
|
|||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var u models.User
|
var u models.User
|
||||||
var ts int64
|
var ts int64
|
||||||
if err := rows.Scan(&u.ID, &u.Username, &u.Email, &ts); err != nil {
|
if err := rows.Scan(&u.ID, &u.Username, &u.Email, &ts, &u.NtfyTopic); err != nil {
|
||||||
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -123,6 +123,50 @@ func handleCreateUser(db *sql.DB) http.HandlerFunc {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// handleSetNotifyTarget points a user's push notifications at an ntfy topic, or
|
||||||
|
// clears it with an empty string. The topic is a shared secret with the ntfy
|
||||||
|
// server — anyone who knows it can publish to it — so pick an unguessable one
|
||||||
|
// unless your ntfy enforces access control.
|
||||||
|
func handleSetNotifyTarget(db *sql.DB) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusBadRequest, errResp("invalid user id"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var req struct {
|
||||||
|
NtfyTopic string `json:"ntfy_topic"`
|
||||||
|
}
|
||||||
|
if err := decodeJSON(r, &req); err != nil {
|
||||||
|
respond(w, http.StatusBadRequest, errResp("invalid request body"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
var topic *string
|
||||||
|
if t := strings.TrimSpace(req.NtfyTopic); t != "" {
|
||||||
|
topic = &t
|
||||||
|
}
|
||||||
|
|
||||||
|
res, err := db.ExecContext(r.Context(),
|
||||||
|
"UPDATE users SET ntfy_topic = ? WHERE id = ?", topic, id)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if n, _ := res.RowsAffected(); n == 0 {
|
||||||
|
respond(w, http.StatusNotFound, errResp("user not found"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
user, err := fetchUser(r.Context(), db, id)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
respond(w, http.StatusOK, user)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func handleDeleteUser(db *sql.DB) http.HandlerFunc {
|
func handleDeleteUser(db *sql.DB) http.HandlerFunc {
|
||||||
return func(w http.ResponseWriter, r *http.Request) {
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
|
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
|
||||||
@@ -170,7 +214,7 @@ func handleCreateAPIKey(db *sql.DB) http.HandlerFunc {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
raw, hash, err := newAPIKey()
|
raw, hash, err := randomToken()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
return
|
return
|
||||||
@@ -215,8 +259,9 @@ func handleDeleteAPIKey(db *sql.DB) http.HandlerFunc {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// newAPIKey generates a random 32-byte key encoded as hex, plus its SHA-256 hash for storage.
|
// randomToken generates a random 32-byte secret encoded as hex, plus its SHA-256
|
||||||
func newAPIKey() (raw, hash string, err error) {
|
// hash for storage. Used for API keys and for notification acknowledge tokens.
|
||||||
|
func randomToken() (raw, hash string, err error) {
|
||||||
b := make([]byte, 32)
|
b := make([]byte, 32)
|
||||||
if _, err = rand.Read(b); err != nil {
|
if _, err = rand.Read(b); err != nil {
|
||||||
return
|
return
|
||||||
@@ -230,8 +275,9 @@ func newAPIKey() (raw, hash string, err error) {
|
|||||||
func fetchUser(ctx context.Context, db *sql.DB, id int64) (models.User, error) {
|
func fetchUser(ctx context.Context, db *sql.DB, id int64) (models.User, error) {
|
||||||
var u models.User
|
var u models.User
|
||||||
var ts int64
|
var ts int64
|
||||||
err := db.QueryRowContext(ctx, "SELECT id, username, email, created_at FROM users WHERE id = ?", id).
|
err := db.QueryRowContext(ctx,
|
||||||
Scan(&u.ID, &u.Username, &u.Email, &ts)
|
"SELECT id, username, email, created_at, ntfy_topic FROM users WHERE id = ?", id).
|
||||||
|
Scan(&u.ID, &u.Username, &u.Email, &ts, &u.NtfyTopic)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return u, err
|
return u, err
|
||||||
}
|
}
|
||||||
|
|||||||
+42
-12
@@ -14,6 +14,25 @@ type Config struct {
|
|||||||
// before the sweeper treats it as resolved. It must exceed Alertmanager's
|
// before the sweeper treats it as resolved. It must exceed Alertmanager's
|
||||||
// repeat_interval (default 4h), which is what refreshes the alert.
|
// repeat_interval (default 4h), which is what refreshes the alert.
|
||||||
StaleAfter time.Duration
|
StaleAfter time.Duration
|
||||||
|
|
||||||
|
// NtfyURL is the ntfy server push notifications are published to. Empty
|
||||||
|
// disables notifications entirely.
|
||||||
|
NtfyURL string
|
||||||
|
|
||||||
|
// NtfyToken is an optional bearer token for an access-controlled ntfy.
|
||||||
|
NtfyToken string
|
||||||
|
|
||||||
|
// NtfyFallbackTopic receives incidents that open with nobody on call.
|
||||||
|
NtfyFallbackTopic string
|
||||||
|
|
||||||
|
// PublicURL is the base URL a phone uses to reach this server, used for the
|
||||||
|
// link and the Acknowledge button inside a notification. Without it
|
||||||
|
// notifications carry neither.
|
||||||
|
PublicURL string
|
||||||
|
|
||||||
|
// NotifyRepeat is how long an incident may sit unacknowledged before it is
|
||||||
|
// notified again. Zero disables reminders.
|
||||||
|
NotifyRepeat time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
func Load() Config {
|
func Load() Config {
|
||||||
@@ -25,17 +44,28 @@ func Load() Config {
|
|||||||
if dbPath == "" {
|
if dbPath == "" {
|
||||||
dbPath = "terdut.db"
|
dbPath = "terdut.db"
|
||||||
}
|
}
|
||||||
archiveAfter := 7 * 24 * time.Hour
|
return Config{
|
||||||
if s := os.Getenv("TERDUT_ARCHIVE_AFTER"); s != "" {
|
Addr: addr,
|
||||||
if d, err := time.ParseDuration(s); err == nil {
|
DBPath: dbPath,
|
||||||
archiveAfter = d
|
ArchiveAfter: duration("TERDUT_ARCHIVE_AFTER", 7*24*time.Hour),
|
||||||
|
StaleAfter: duration("TERDUT_STALE_AFTER", 6*time.Hour),
|
||||||
|
|
||||||
|
NtfyURL: os.Getenv("TERDUT_NTFY_URL"),
|
||||||
|
NtfyToken: os.Getenv("TERDUT_NTFY_TOKEN"),
|
||||||
|
NtfyFallbackTopic: os.Getenv("TERDUT_NTFY_FALLBACK_TOPIC"),
|
||||||
|
PublicURL: os.Getenv("TERDUT_PUBLIC_URL"),
|
||||||
|
NotifyRepeat: duration("TERDUT_NOTIFY_REPEAT", 15*time.Minute),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
staleAfter := 6 * time.Hour
|
|
||||||
if s := os.Getenv("TERDUT_STALE_AFTER"); s != "" {
|
// duration reads a time.ParseDuration-formatted env var. An unset or
|
||||||
if d, err := time.ParseDuration(s); err == nil {
|
// unparseable value falls back to def rather than failing startup: a typo in one
|
||||||
staleAfter = d
|
// tuning knob should not take the server down.
|
||||||
}
|
func duration(env string, def time.Duration) time.Duration {
|
||||||
}
|
if s := os.Getenv(env); s != "" {
|
||||||
return Config{Addr: addr, DBPath: dbPath, ArchiveAfter: archiveAfter, StaleAfter: staleAfter}
|
if d, err := time.ParseDuration(s); err == nil {
|
||||||
|
return d
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return def
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,44 @@
|
|||||||
|
-- Adds push notification delivery, so an incident reaches the person on call
|
||||||
|
-- instead of waiting to be discovered.
|
||||||
|
--
|
||||||
|
-- Delivery is an outbox rather than an inline HTTP call: the pool is limited to
|
||||||
|
-- a single connection, so a POST made while holding the webhook's transaction
|
||||||
|
-- would stall every other request behind it. The webhook inserts a row; the
|
||||||
|
-- notifier goroutine delivers it.
|
||||||
|
|
||||||
|
CREATE TABLE notifications (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE,
|
||||||
|
-- Nullable: a notification sent to the fallback topic belongs to nobody,
|
||||||
|
-- because nobody was on call when the incident opened.
|
||||||
|
user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
|
||||||
|
topic TEXT NOT NULL, -- resolved at enqueue: who was on call then
|
||||||
|
kind TEXT NOT NULL CHECK(kind IN ('triggered', 'reminder', 'resolved')),
|
||||||
|
created_at INTEGER NOT NULL,
|
||||||
|
send_after INTEGER NOT NULL, -- retry backoff watermark
|
||||||
|
attempts INTEGER NOT NULL DEFAULT 0,
|
||||||
|
sent_at INTEGER,
|
||||||
|
last_error TEXT -- kept after the last attempt, for debugging
|
||||||
|
);
|
||||||
|
|
||||||
|
-- The delivery loop's only query: what is due and still unsent.
|
||||||
|
CREATE INDEX notifications_pending_idx ON notifications(send_after) WHERE sent_at IS NULL;
|
||||||
|
-- Reminders and resolved notices both look up an incident's newest row.
|
||||||
|
CREATE INDEX notifications_incident_idx ON notifications(incident_id, id DESC);
|
||||||
|
|
||||||
|
-- A notification body is stored on the ntfy server and cached on the device, so
|
||||||
|
-- a real API key must never appear in one. Each delivery mints its own token
|
||||||
|
-- instead: one incident, one action, one day.
|
||||||
|
CREATE TABLE incident_ack_tokens (
|
||||||
|
token_hash TEXT PRIMARY KEY, -- SHA-256 of the raw token, as with api_keys
|
||||||
|
incident_id INTEGER NOT NULL REFERENCES incidents(id) ON DELETE CASCADE,
|
||||||
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||||||
|
created_at INTEGER NOT NULL,
|
||||||
|
expires_at INTEGER NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX incident_ack_tokens_expires_idx ON incident_ack_tokens(expires_at);
|
||||||
|
|
||||||
|
-- Where this user's notifications go. NULL means they get none; incidents
|
||||||
|
-- assigned to them fall back to the configured fallback topic.
|
||||||
|
ALTER TABLE users ADD COLUMN ntfy_topic TEXT;
|
||||||
@@ -7,6 +7,11 @@ type User struct {
|
|||||||
Username string `json:"username"`
|
Username string `json:"username"`
|
||||||
Email string `json:"email"`
|
Email string `json:"email"`
|
||||||
CreatedAt time.Time `json:"created_at"`
|
CreatedAt time.Time `json:"created_at"`
|
||||||
|
|
||||||
|
// NtfyTopic is where this user's push notifications go. Nil means they get
|
||||||
|
// none of their own; incidents assigned to them fall back to the configured
|
||||||
|
// fallback topic instead.
|
||||||
|
NtfyTopic *string `json:"ntfy_topic,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type APIKey struct {
|
type APIKey struct {
|
||||||
|
|||||||
Reference in New Issue
Block a user