diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 98f8e1f..7dc9d11 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -102,3 +102,36 @@ jobs: flags: unittests token: ${{ secrets.CODECOV_TOKEN }} slug: platformlabs-co/fetchr.sh + + alerting: + name: Alert Rules and Lifecycle + runs-on: ubuntu-latest + timeout-minutes: 10 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v5 + with: + go-version: '1.24' + - name: Download verified monitoring tools + run: | + mkdir -p "$RUNNER_TEMP/monitoring" + cd "$RUNNER_TEMP/monitoring" + for spec in prometheus:3.14.0 alertmanager:0.34.1; do + tool="${spec%:*}" + version="${spec#*:}" + archive="$tool-$version.linux-amd64.tar.gz" + curl -fsSL "https://github.com/prometheus/$tool/releases/download/v$version/$archive" -o "$archive" + curl -fsSL "https://github.com/prometheus/$tool/releases/download/v$version/sha256sums.txt" -o "$tool.sums" + sha256sum --check --ignore-missing "$tool.sums" + tar -xzf "$archive" + done + - name: Validate rules and routing + run: | + "$RUNNER_TEMP/monitoring/prometheus-3.14.0.linux-amd64/promtool" check config examples/alerting/prometheus.yml + "$RUNNER_TEMP/monitoring/prometheus-3.14.0.linux-amd64/promtool" test rules examples/alerting/rules.test.yml + "$RUNNER_TEMP/monitoring/alertmanager-0.34.1.linux-amd64/amtool" check-config examples/alerting/alertmanager.yml + - name: Validate real notification lifecycle + run: | + go test -race ./internal/alerting ./internal/proxy + go build -o "$RUNNER_TEMP/netkit-alert-test" ./cmd/netkit + python3 scripts/test-alerting-integration.py --netkit "$RUNNER_TEMP/netkit-alert-test" --alertmanager "$RUNNER_TEMP/monitoring/alertmanager-0.34.1.linux-amd64/alertmanager" diff --git a/.gitignore b/.gitignore index 8b33eae..7b9f377 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,10 @@ bin/ # Embedded dashboard files (generated during build) internal/dashboard/out/ + +# Local alerting examples: credentials and retained evidence +examples/alerting/webhook-token +examples/alerting/notification-token +examples/alerting/incidents/ +examples/alerting/alertmanager-data/ +examples/alerting/prometheus-data/ diff --git a/README.md b/README.md index 8d85871..3776314 100644 --- a/README.md +++ b/README.md @@ -372,3 +372,7 @@ Captured history is sanitized before retention. The dashboard separates HTTP sta ### Investigation workspace Traffic is the landing page, with preserved filters, adjacent request inspection, buffered Live/Pause, and an Overview that drills into matching retained evidence. Compose is available at `/compose/`. See [investigation workspace semantics](docs/investigation-workspace.md) for freshness, URL scope, latency populations, and retention limits. + +### Focused alerting + +Use the [Prometheus and Alertmanager example](examples/alerting/README.md) for sustained upstream/transport failure alerts, opt-in ordinary-workload header latency, external availability and insufficient-sample signals. The separate `netkit alerts` process retains bounded metadata-only incident snapshots with authenticated webhook writes. Alert evaluation and notification ownership remain external; no native monitor editor or production notification destination is enabled automatically. diff --git a/cmd/netkit/alerts.go b/cmd/netkit/alerts.go new file mode 100644 index 0000000..694ee07 --- /dev/null +++ b/cmd/netkit/alerts.go @@ -0,0 +1,63 @@ +package main + +import ( + "context" + "flag" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/biancarosa/netkit/internal/alerting" +) + +func runAlerts(args []string) error { + flags := flag.NewFlagSet("alerts", flag.ContinueOnError) + listen := flags.String("listen", "127.0.0.1:8090", "Incident receiver address; protect with authenticated ingress before exposing") + dir := flags.String("dir", "", "Private incident directory (required)") + admin := flags.String("admin-url", "http://127.0.0.1:8081", "Fixed Netkit admin URL for metadata evidence") + dashboard := flags.String("dashboard-url", "http://127.0.0.1:3000", "Public Netkit dashboard URL including any prefix") + instance := flags.String("instance", "netkit-local", "Expected Prometheus instance alias") + capacity := flags.Int("capacity", 100, "Maximum stored incidents (1–1000)") + ttl := flags.Duration("retention", 7*24*time.Hour, "Incident retention (1 minute–30 days)") + if err := flags.Parse(args); err != nil { + return err + } + if *dir == "" { + return fmt.Errorf("--dir is required") + } + receiver, err := alerting.New(alerting.Config{Dir: *dir, Token: os.Getenv("NETKIT_ALERT_TOKEN"), Instance: *instance, AdminURL: *admin, DashboardURL: *dashboard, Capacity: *capacity, TTL: *ttl}) + if err != nil { + return err + } + if err := receiver.Prune(); err != nil { + return err + } + server := &http.Server{Addr: *listen, Handler: receiver.Handler(), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 45 * time.Second, IdleTimeout: 30 * time.Second} + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + go func() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second) + _ = server.Shutdown(shutdown) + cancel() + return + case <-ticker.C: + if err := receiver.Prune(); err != nil { + log.Print("incident retention maintenance failed; check private storage") + } + } + } + }() + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + return err + } + return nil +} diff --git a/cmd/netkit/main.go b/cmd/netkit/main.go index b1e3e7a..6b88d14 100644 --- a/cmd/netkit/main.go +++ b/cmd/netkit/main.go @@ -14,13 +14,17 @@ import ( func main() { // Parse command if len(os.Args) < 2 { - log.Fatal("Please specify a command: serve, request, or ca") + log.Fatal("Please specify a command: serve, request, ca, or alerts") } command := os.Args[1] os.Args = os.Args[1:] // Remove command from args for flag parsing switch command { + case "alerts": + if err := runAlerts(os.Args[1:]); err != nil { + log.Fatal(err) + } case "ca": if err := runCA(os.Args[1:]); err != nil { log.Fatal(err) @@ -32,7 +36,7 @@ func main() { log.Fatal(err) } default: - log.Fatalf("Unknown command: %s. Use 'serve', 'request', or 'ca'", command) + log.Fatalf("Unknown command: %s. Use 'serve', 'request', 'ca', or 'alerts'", command) } } diff --git a/dashboard/src/components/InvestigationWorkspace.tsx b/dashboard/src/components/InvestigationWorkspace.tsx index 0f21e7d..ceae061 100644 --- a/dashboard/src/components/InvestigationWorkspace.tsx +++ b/dashboard/src/components/InvestigationWorkspace.tsx @@ -302,8 +302,12 @@ export function InvestigationWorkspace({ > Clear filters - {(state.scope.route || state.scope.from || state.scope.to) && ( + {(state.scope.signal || + state.scope.route || + state.scope.from || + state.scope.to) && (

+ {state.scope.signal && `Alert signal: ${state.scope.signal} · `} {state.scope.route && `Route: ${state.scope.route} · `} {state.scope.from && `From ${new Date(state.scope.from).toLocaleString()} `} @@ -311,9 +315,11 @@ export function InvestigationWorkspace({ `to ${new Date(state.scope.to).toLocaleString()}`}{" "}

)} diff --git a/dashboard/src/lib/investigation.test.ts b/dashboard/src/lib/investigation.test.ts index c580678..c5dc371 100644 --- a/dashboard/src/lib/investigation.test.ts +++ b/dashboard/src/lib/investigation.test.ts @@ -22,6 +22,8 @@ const record = ( outcome: "complete", success: true, response_source: "upstream", + upstream_start_time: "2026-09-22T12:00:00Z", + upstream_end_time: "2026-09-22T12:00:00.001Z", upstream_latency_us: 1000, total_duration_us: 30000000, ...overrides, @@ -101,3 +103,43 @@ test("trend bucket boundaries partition records exactly and produce matching dri assert.equal(buckets([record("a"), record("b")]).length, 1); assert.equal(summary([]).p95, null); }); + +test("alert links use exact signal populations and exclude unrelated failures", () => { + const records = [ + record("http5xx", { response_status: 503 }), + record("401", { response_status: 401 }), + record("transport", { + outcome: "upstream_error", + response_status: 502, + response_source: "proxy", + }), + record("canceled", { + outcome: "client_canceled", + response_status: 0, + response_source: "proxy", + }), + ]; + assert.deepEqual( + records + .filter((r) => matches(r, { ...emptyScope, signal: "http5xx" })) + .map((r) => r.id), + ["http5xx"], + ); + assert.deepEqual( + records + .filter((r) => matches(r, { ...emptyScope, signal: "transport" })) + .map((r) => r.id), + ["transport"], + ); + assert.deepEqual( + records + .filter((r) => matches(r, { ...emptyScope, signal: "headers" })) + .map((r) => r.id), + ["http5xx", "401"], + ); + assert.equal( + records.filter((r) => matches(r, { ...emptyScope, signal: "unknown" })) + .length, + 0, + ); +}); diff --git a/dashboard/src/lib/investigation.ts b/dashboard/src/lib/investigation.ts index d37c46d..df5b424 100644 --- a/dashboard/src/lib/investigation.ts +++ b/dashboard/src/lib/investigation.ts @@ -1,5 +1,6 @@ import type { BackendRequestRecord as Record } from "../services/api"; export type Scope = { + signal: string; host: string; route: string; method: string; @@ -8,6 +9,7 @@ export type Scope = { to: string; }; export const emptyScope: Scope = { + signal: "", host: "", route: "", method: "", @@ -42,6 +44,16 @@ export function matches(record: Record, scope: Scope) { const target = endpoint(record); const time = Date.parse(record.timestamp); return ( + (!scope.signal || + (scope.signal === "http5xx" + ? record.response_source === "upstream" && + record.response_status >= 500 && + record.response_status <= 599 + : scope.signal === "transport" + ? ["upstream_error", "proxy_error"].includes(record.outcome) + : scope.signal === "headers" + ? headersEligible(record) + : false)) && (!scope.host || scope.host === target.host) && (!scope.route || scope.route === target.route) && (!scope.method || scope.method === record.method) && @@ -58,6 +70,10 @@ export function headersEligible(record: Record) { record.response_source === "upstream" && record.response_status > 0 && record.method !== "CONNECT" && + Number.isFinite(Date.parse(record.upstream_start_time)) && + Date.parse(record.upstream_start_time) > 0 && + Date.parse(record.upstream_end_time) >= + Date.parse(record.upstream_start_time) && record.upstream_latency_us >= 0 ); } diff --git a/examples/alerting/README.md b/examples/alerting/README.md new file mode 100644 index 0000000..10806ed --- /dev/null +++ b/examples/alerting/README.md @@ -0,0 +1,94 @@ +# Focused alerting + +Prometheus evaluates rules; Alertmanager groups, silences and delivers notifications. Netkit exports HTTP signals and provides investigation and bounded incident metadata. This example does not configure a real email/chat destination or deploy any service. + +## Rules and populations + +| Rule | Condition | Gate / hold | Interpretation | +| --- | --- | --- | --- | +| NetkitUpstreamFailureRate | Upstream HTTP 5xx / non-tunnel HTTP observations > 10% over 5m | At least 20 observations; 3m sustained | Does not call 401/429 or inspection rejection an upstream outage. | +| NetkitTransportFailureRate | Proxy errors or upstream connection failures / HTTP observations > 10% over 5m | At least 20 observations; 3m sustained | Client cancellation and inspection rejection are excluded from the numerator. | +| NetkitHeaderLatencyHigh | p95 upstream response-header latency > 1 second over 5m | At least 20 header samples; 5m sustained; **ordinary profile only** | Excludes response body lifetime, tunnels and failed attempts without headers. | +| NetkitUnavailable | External Prometheus scrape `up == 0` | 2m sustained | Reachability, not proof of capture completeness. | +| NetkitInsufficientSamples | Fewer than 20 HTTP observations over 5m | 5m sustained; informational | No data/low volume is unknown, not healthy or a missing-traffic failure. | + +Thresholds are examples, not universal defaults. The sample `netkit_latency_profile: unclassified` deliberately disables the latency rule. Enable `ordinary` only on an isolated proxy serving a comparable ordinary HTTP workload. Netkit does not infer request types, dependencies or route templates. Never enable a blanket latency rule for mixed long-polling/streaming traffic. Existing metrics support instance-level scope, not host/route alerts; the UI must not imply otherwise. + +Each rule includes a value, threshold, window and investigation link. Static `netkit_dashboard` and `netkit_incidents` external labels must contain public URLs without credentials or query secrets. Set the dashboard URL to include any deployment prefix, without a trailing slash. Metric/alert labels must never contain request URLs, credentials or arbitrary payload fields. + +## Local setup + +Use Prometheus 3.14.0 and Alertmanager 0.34.1 (the versions validated in CI), plus a current Netkit build. Start Netkit normally with admin port 8081 and dashboard port 3000. Run the incident receiver as a **separate process**, ideally on a different host or failure domain from the monitored proxy in a real deployment. + +From this directory: + +```sh +umask 077 +openssl rand -hex 32 > webhook-token +mkdir -m 700 incidents +NETKIT_ALERT_TOKEN="$(cat webhook-token)" netkit alerts \ + --dir "$PWD/incidents" \ + --listen 127.0.0.1:8090 \ + --admin-url http://127.0.0.1:8081 \ + --dashboard-url http://localhost:3000 \ + --instance netkit-local +``` + +In separate terminals, from this same directory: + +```sh +alertmanager --config.file=alertmanager.yml --storage.path=./alertmanager-data +prometheus --config.file=prometheus.yml --storage.tsdb.path=./prometheus-data +``` + +The webhook token is mandatory (at least 16 characters), provided only through the receiver environment and Alertmanager's credentials file. Keep it out of Git. The receiver binds loopback by default. Its read-only HTML interface has **no built-in authentication**: use authenticated private ingress before exposing it. The fixed admin URL is trusted configuration; redirects and alert-supplied target URLs are not followed. Keep the existing Netkit admin endpoint private as well. + +The example expects one proxy with the safe Prometheus instance alias `netkit-local`. For multiple instances, use one receiver and directory per instance and route Alertmanager by instance; mismatched instances are rejected rather than snapshotting unrelated traffic. + +Open Prometheus's Alerts page to see pending/firing evaluation, Alertmanager to manage silences, and `http://localhost:8090/` for received incident evidence. Absence of an incident is not a health signal: pending or silenced alerts may never notify this receiver. + +## Notification routing + +The supplied receiver writes local evidence only. Add your operator-owned notification integration to the same Alertmanager receiver, using its documented email/chat/webhook schema. For a generic endpoint that accepts Alertmanager's webhook JSON, add another entry under `webhook_configs`: + +```yaml +- url: https://notifications.example.invalid/alertmanager + send_resolved: true + http_config: + authorization: + credentials_file: ./notification-token +``` + +Replace the `.invalid` placeholder and create the private token file before enabling it. Keep grouping by alertname/job/instance, 30s initial group delay, 5m group interval and 4h repeat interval unless your response policy calls for different values. A NetkitUnavailable alert inhibits the traffic rules for the same job/instance. Alertmanager silences affect both evidence delivery and other integrations; snapshots therefore reflect **delivered notifications**, not an exhaustive incident history. Monitor Prometheus, Alertmanager notification failures, and the receiver independently; Netkit cannot report its own total monitoring-stack failure. + +## Evidence contract and retention + +`GET /requests/evidence?signal=http5xx|transport|headers&from=&to=` is an admin-only metadata boundary. It returns at most 20 matching retained records and the total matching retained count. It excludes URLs, host/path names, headers, bodies and error text entirely. The receiver validates a fixed schema and discards all arbitrary alert annotations/labels, storing only known rule identity, lifecycle timestamps, finite numeric value/threshold and typed evidence. + +The first delivered notification snapshots the available evidence from five minutes before condition onset through delivery. This contextual window includes evaluation and pending time; it is not an immutable copy of every metric sample. Later retries and recovery update lifecycle metadata without replacing the initial evidence. An out-of-order firing retry cannot reopen a resolved incident with the same start time. A resolved notification received without an earlier firing delivery is recorded as such, with whatever evidence is available at delivery. + +Snapshots are atomic mode-0600 JSON files in a required private mode-0700 directory. Defaults are 100 incidents and seven days, configurable with `--capacity` (1–1000) and `--retention` (1m–30d). TTL is measured from snapshot creation, not renewed on each retry. Pruning runs on reads/writes and once per minute while the process is running; offline expiry is enforced on restart. The receiver never deletes files outside its incident naming convention. Disk errors return non-success so Alertmanager can retry. + +Evidence states distinguish captured, no retained matches, unavailable (including failed fetch or invalid schema), and not applicable (availability/insufficient-sample alerts). A snapshot can survive request-history clearing, eviction and proxy/receiver restarts. Incident links return explicit 404/expired state after retention. Investigation links apply the signal and snapshot time bounds to the **current** Traffic buffer; that buffer may no longer contain the saved evidence. Resolved means the alert expression stopped firing, not proof of health—scrape failure and insufficient data have separate signals. + +The receiver preserves metadata only. Payload reproduction, durable full traffic storage, per-route metrics, rate-limiting/stream-interruption rules and native monitor editing remain later work. + +## Validation + +```sh +promtool check config prometheus.yml +promtool check rules rules.yml +promtool test rules rules.test.yml +amtool check-config alertmanager.yml +# From repository root, with real binaries available: +go test -race ./internal/alerting ./internal/proxy +go build -o /tmp/netkit-alert-test ./cmd/netkit +python3 scripts/test-alerting-integration.py \ + --netkit /tmp/netkit-alert-test --alertmanager /path/to/alertmanager +``` + +Rule tests use synthetic clocks/series to verify brief spikes, sustained failures, sample gates, low/no-data, external downtime/recovery, cancellation/inspection exclusions and latency profile gating. The integration test runs a real Alertmanager and real receiver on disposable loopback ports, verifying grouping, repeat delivery, silence suppression, recovery, deduplication and safe evidence persistence. It sends no external notifications. + +A scrape target removed from configuration produces no `up` series; `NetkitUnavailable` monitors configured targets, not inventory deletion. Maintain target inventory and independently monitor the monitoring stack. Missing expected traffic requires an explicit cadence policy and is not inferred from ordinary silence. + +References: [Prometheus alerting rules](https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/), [rule unit tests](https://prometheus.io/docs/prometheus/latest/configuration/unit_testing_rules/), [Alertmanager configuration](https://prometheus.io/docs/alerting/latest/configuration/). diff --git a/examples/alerting/alertmanager.yml b/examples/alerting/alertmanager.yml new file mode 100644 index 0000000..e6c75b1 --- /dev/null +++ b/examples/alerting/alertmanager.yml @@ -0,0 +1,21 @@ +route: + receiver: incident-evidence + group_by: [alertname, job, instance] + group_wait: 30s + group_interval: 5m + repeat_interval: 4h +receivers: + - name: incident-evidence + webhook_configs: + - url: http://127.0.0.1:8090/webhook + send_resolved: true + max_alerts: 8 + http_config: + authorization: + credentials_file: ./webhook-token +# Add an operator-owned notification receiver to this route when ready. +# The example deliberately sends only to the local evidence receiver. +inhibit_rules: + - source_matchers: ['alertname="NetkitUnavailable"'] + target_matchers: ['alertname=~"NetkitUpstreamFailureRate|NetkitTransportFailureRate|NetkitHeaderLatencyHigh|NetkitInsufficientSamples"'] + equal: [job, instance] diff --git a/examples/alerting/prometheus.yml b/examples/alerting/prometheus.yml new file mode 100644 index 0000000..3d7d106 --- /dev/null +++ b/examples/alerting/prometheus.yml @@ -0,0 +1,21 @@ +global: + scrape_interval: 15s + evaluation_interval: 15s + external_labels: + netkit_dashboard: http://localhost:3000 + netkit_incidents: http://localhost:8090 +rule_files: + - rules.yml +alerting: + alertmanagers: + - static_configs: + - targets: ['127.0.0.1:9093'] +scrape_configs: + - job_name: netkit + static_configs: + - targets: ['127.0.0.1:8081'] + labels: + instance: netkit-local + # Enable ONLY for an isolated, comparable ordinary HTTP workload. + # Do not enable for a mixed streaming/long-polling proxy. + netkit_latency_profile: unclassified diff --git a/examples/alerting/rules.test.yml b/examples/alerting/rules.test.yml new file mode 100644 index 0000000..470dd0f --- /dev/null +++ b/examples/alerting/rules.test.yml @@ -0,0 +1,314 @@ +{ + "rule_files": [ + "rules.yml" + ], + "evaluation_interval": "1m", + "tests": [ + { + "name": "sustained HTTP failure fires after hold and recovers", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"complete\",http_status_class=\"2xx\"}", + "values": "0+80x30" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"complete\",http_status_class=\"5xx\"}", + "values": "0+20x8 160+0x21" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "2m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "5m", + "exp_samples": [ + { + "labels": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\",instance=\"netkit-local\",job=\"netkit\",severity=\"warning\"}", + "value": 1 + } + ] + }, + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "15m", + "exp_samples": [] + } + ] + }, + { + "name": "short failure spike stays quiet", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"complete\",http_status_class=\"2xx\"}", + "values": "0+80x20" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"complete\",http_status_class=\"5xx\"}", + "values": "0 20+0x19" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "3m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "8m", + "exp_samples": [] + } + ] + }, + { + "name": "low sample counts do not page", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"complete\",http_status_class=\"5xx\"}", + "values": "0+1x20" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "10m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitInsufficientSamples\",alertstate=\"firing\"}", + "eval_time": "10m", + "exp_samples": [ + { + "labels": "ALERTS{alertname=\"NetkitInsufficientSamples\",alertstate=\"firing\",instance=\"netkit-local\",job=\"netkit\",severity=\"info\"}", + "value": 1 + } + ] + } + ] + }, + { + "name": "fresh reachable proxy has no data, not healthy", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitInsufficientSamples\",alertstate=\"firing\"}", + "eval_time": "6m", + "exp_samples": [ + { + "labels": "ALERTS{alertname=\"NetkitInsufficientSamples\",alertstate=\"firing\",instance=\"netkit-local\",job=\"netkit\",severity=\"info\"}", + "value": 1 + } + ] + }, + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "6m", + "exp_samples": [] + } + ] + }, + { + "name": "external downtime and recovery", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1 0+0x5 1+0x10" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitUnavailable\",alertstate=\"firing\"}", + "eval_time": "2m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitUnavailable\",alertstate=\"firing\"}", + "eval_time": "3m", + "exp_samples": [ + { + "labels": "ALERTS{alertname=\"NetkitUnavailable\",alertstate=\"firing\",instance=\"netkit-local\",job=\"netkit\",severity=\"critical\"}", + "value": 1 + } + ] + }, + { + "expr": "ALERTS{alertname=\"NetkitUnavailable\",alertstate=\"firing\"}", + "eval_time": "7m", + "exp_samples": [] + } + ] + }, + { + "name": "transport errors fire independently", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"complete\",http_status_class=\"2xx\"}", + "values": "0+80x20" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"upstream_error\",http_status_class=\"none\"}", + "values": "0+20x20" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitTransportFailureRate\",alertstate=\"firing\"}", + "eval_time": "5m", + "exp_samples": [ + { + "labels": "ALERTS{alertname=\"NetkitTransportFailureRate\",alertstate=\"firing\",instance=\"netkit-local\",job=\"netkit\",severity=\"warning\"}", + "value": 1 + } + ] + }, + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "5m", + "exp_samples": [] + } + ] + }, + { + "name": "latency profile ordinary", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_response_headers_seconds_bucket{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"ordinary\",method=\"GET\",le=\"1\"}", + "values": "0+0x20" + }, + { + "series": "netkit_response_headers_seconds_bucket{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"ordinary\",method=\"GET\",le=\"2.5\"}", + "values": "0+100x20" + }, + { + "series": "netkit_response_headers_seconds_bucket{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"ordinary\",method=\"GET\",le=\"+Inf\"}", + "values": "0+100x20" + }, + { + "series": "netkit_response_headers_seconds_count{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"ordinary\",method=\"GET\"}", + "values": "0+100x20" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitHeaderLatencyHigh\",alertstate=\"firing\"}", + "eval_time": "3m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitHeaderLatencyHigh\",alertstate=\"firing\"}", + "eval_time": "7m", + "exp_samples": [ + { + "labels": "ALERTS{alertname=\"NetkitHeaderLatencyHigh\",alertstate=\"firing\",instance=\"netkit-local\",job=\"netkit\",severity=\"warning\"}", + "value": 1 + } + ] + } + ] + }, + { + "name": "latency profile unclassified", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_response_headers_seconds_bucket{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"unclassified\",method=\"GET\",le=\"1\"}", + "values": "0+0x20" + }, + { + "series": "netkit_response_headers_seconds_bucket{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"unclassified\",method=\"GET\",le=\"2.5\"}", + "values": "0+100x20" + }, + { + "series": "netkit_response_headers_seconds_bucket{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"unclassified\",method=\"GET\",le=\"+Inf\"}", + "values": "0+100x20" + }, + { + "series": "netkit_response_headers_seconds_count{job=\"netkit\",instance=\"netkit-local\",netkit_latency_profile=\"unclassified\",method=\"GET\"}", + "values": "0+100x20" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitHeaderLatencyHigh\",alertstate=\"firing\"}", + "eval_time": "3m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitHeaderLatencyHigh\",alertstate=\"firing\"}", + "eval_time": "7m", + "exp_samples": [] + } + ] + }, + { + "name": "cancellations and inspection rejection do not become transport failures", + "interval": "1m", + "input_series": [ + { + "series": "up{job=\"netkit\",instance=\"netkit-local\"}", + "values": "1+0x40" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"client_canceled\",http_status_class=\"none\"}", + "values": "0+100x20" + }, + { + "series": "netkit_exchanges_total{job=\"netkit\",instance=\"netkit-local\",method=\"GET\",outcome=\"blocked_by_inspection\",http_status_class=\"none\"}", + "values": "0+100x20" + } + ], + "promql_expr_test": [ + { + "expr": "ALERTS{alertname=\"NetkitTransportFailureRate\",alertstate=\"firing\"}", + "eval_time": "10m", + "exp_samples": [] + }, + { + "expr": "ALERTS{alertname=\"NetkitUpstreamFailureRate\",alertstate=\"firing\"}", + "eval_time": "10m", + "exp_samples": [] + } + ] + } + ] +} diff --git a/examples/alerting/rules.yml b/examples/alerting/rules.yml new file mode 100644 index 0000000..66492d9 --- /dev/null +++ b/examples/alerting/rules.yml @@ -0,0 +1,68 @@ +groups: + - name: netkit.focused + rules: + - record: netkit:http_observations:increase5m + expr: sum by (job, instance) (increase(netkit_exchanges_total{job="netkit",method!="CONNECT",outcome!="tunnel_only"}[5m])) or on (job, instance) (sum by (job, instance) (0 * up{job="netkit"})) + - record: netkit:upstream_failures:ratio5m + expr: (sum by (job, instance) (increase(netkit_exchanges_total{job="netkit",method!="CONNECT",http_status_class="5xx"}[5m])) or on (job, instance) (sum by (job, instance) (0 * up{job="netkit"}))) / netkit:http_observations:increase5m + - record: netkit:transport_failures:ratio5m + expr: (sum by (job, instance) (increase(netkit_exchanges_total{job="netkit",method!="CONNECT",outcome=~"upstream_error|proxy_error"}[5m])) or on (job, instance) (sum by (job, instance) (0 * up{job="netkit"}))) / netkit:http_observations:increase5m + - record: netkit:headers:p95_5m + expr: histogram_quantile(0.95, sum by (job, instance, le) (rate(netkit_response_headers_seconds_bucket{job="netkit",netkit_latency_profile="ordinary",method!="CONNECT"}[5m]))) + - record: netkit:headers:count5m + expr: sum by (job, instance) (increase(netkit_response_headers_seconds_count{job="netkit",netkit_latency_profile="ordinary",method!="CONNECT"}[5m])) + - alert: NetkitUpstreamFailureRate + expr: (netkit:upstream_failures:ratio5m > 0.10) and on (job, instance) (netkit:http_observations:increase5m >= 20) and on (job, instance) (up{job="netkit"} == 1) + for: 3m + labels: {severity: warning} + annotations: + summary: Sustained upstream HTTP 5xx + value: '{{ $value }}' + threshold: '0.10' + window: '5m; at least 20 HTTP observations; sustained 3m' + investigate_url: '{{ $externalLabels.netkit_dashboard }}/?signal=http5xx' + incident_url: '{{ $externalLabels.netkit_incidents }}/' + - alert: NetkitTransportFailureRate + expr: (netkit:transport_failures:ratio5m > 0.10) and on (job, instance) (netkit:http_observations:increase5m >= 20) and on (job, instance) (up{job="netkit"} == 1) + for: 3m + labels: {severity: warning} + annotations: + summary: Sustained proxy or upstream connection failures + value: '{{ $value }}' + threshold: '0.10' + window: '5m; at least 20 HTTP observations; sustained 3m; excludes client cancellations and inspection rejection' + investigate_url: '{{ $externalLabels.netkit_dashboard }}/?signal=transport' + incident_url: '{{ $externalLabels.netkit_incidents }}/' + - alert: NetkitHeaderLatencyHigh + expr: (netkit:headers:p95_5m > 1) and on (job, instance) (netkit:headers:count5m >= 20) and on (job, instance) (up{job="netkit"} == 1) + for: 5m + labels: {severity: warning} + annotations: + summary: High time to response headers for ordinary HTTP profile + value: '{{ $value }}' + threshold: '1' + window: '5m; at least 20 header samples; sustained 5m; ordinary profile only' + investigate_url: '{{ $externalLabels.netkit_dashboard }}/?signal=headers' + incident_url: '{{ $externalLabels.netkit_incidents }}/' + - alert: NetkitUnavailable + expr: min by (job, instance) (up{job="netkit"}) == 0 + for: 2m + labels: {severity: critical} + annotations: + summary: Netkit scrape unavailable + value: '{{ $value }}' + threshold: '1' + window: 'External scrape failed for 2m; checks reachability, not capture completeness' + investigate_url: '{{ $externalLabels.netkit_dashboard }}/' + incident_url: '{{ $externalLabels.netkit_incidents }}/' + - alert: NetkitInsufficientSamples + expr: (netkit:http_observations:increase5m < 20) and on (job, instance) (up{job="netkit"} == 1) + for: 5m + labels: {severity: info} + annotations: + summary: Insufficient samples; traffic health is unknown + value: '{{ $value }}' + threshold: '20' + window: 'Fewer than 20 HTTP observations in 5m; not a missing-traffic failure' + investigate_url: '{{ $externalLabels.netkit_dashboard }}/' + incident_url: '{{ $externalLabels.netkit_incidents }}/' diff --git a/internal/alerting/evidence.go b/internal/alerting/evidence.go new file mode 100644 index 0000000..d137b95 --- /dev/null +++ b/internal/alerting/evidence.go @@ -0,0 +1,50 @@ +// Package alerting provides an opt-in, separately operated incident receiver. +// Alert evaluation and notification ownership remain in Prometheus/Alertmanager. +package alerting + +import ( + "regexp" + "time" +) + +type Evidence struct { + ID string `json:"id"` + Timestamp time.Time `json:"timestamp"` + Method string `json:"method"` + Outcome string `json:"outcome"` + Status int `json:"status"` + Source string `json:"source"` + HeadersUS int64 `json:"headers_us"` + DurationUS int64 `json:"duration_us"` +} + +var recordID = regexp.MustCompile(`^[a-f0-9]{32}$`) + +func (e Evidence) Valid() bool { + if !recordID.MatchString(e.ID) || e.Timestamp.IsZero() || e.HeadersUS < 0 || e.DurationUS < 0 || e.Status < 0 || e.Status > 599 { + return false + } + switch e.Method { + case "GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS", "CONNECT", "OTHER": + default: + return false + } + switch e.Outcome { + case "complete", "interrupted", "client_canceled", "blocked_by_inspection", "upstream_error", "proxy_error", "tunnel_only", "unknown": + default: + return false + } + return e.Source == "upstream" || e.Source == "proxy" || e.Source == "unknown" +} +func (e Evidence) Matches(signal string) bool { + switch signal { + case "http5xx": + return e.Source == "upstream" && e.Status >= 500 && e.Status <= 599 + case "transport": + return e.Outcome == "upstream_error" || e.Outcome == "proxy_error" + case "headers": + return e.Source == "upstream" && e.Status > 0 && e.Method != "CONNECT" + default: + return false + } +} diff --git a/internal/alerting/incident.html b/internal/alerting/incident.html new file mode 100644 index 0000000..3a5307a --- /dev/null +++ b/internal/alerting/incident.html @@ -0,0 +1,121 @@ + + + Netkit incident evidence +

Incident evidence

+

+ Prometheus evaluates rules; Alertmanager owns grouping, silences and + delivery. This receiver stores bounded metadata snapshots. “Resolved” means + the rule stopped firing, not proof of health. +

+

+ Instance: {{.Instance}} · up to {{.Capacity}} incidents · retention + {{retention .TTL}}. Evidence excludes URLs, headers, bodies and arbitrary alert text. +

+ {{range .Items}} +
+

{{title .Incident.Rule}}

+

{{.Incident.Status}} · {{.Incident.Rule}}

+

+ First reported value: {{number .Incident.Observed}} · threshold: {{number + .Incident.Threshold}} ({{unit .Incident.Rule}})
Started: + {{when .Incident.Started}}
Evidence window: {{when .Incident.From}} – + {{when .Incident.To}} + {{if eq .Incident.Status "resolved"}}
Resolved: {{when .Incident.EndsAt}}{{end}} +

+

+ Evidence: {{.Incident.EvidenceState}} · {{len .Incident.Evidence}} of + {{.Incident.Matching}} matching retained records at notification time. +

+

+ Investigate matching traffic · + Permanent incident link (until retention expires) +

+
+ + + + + + + + + + {{range .Incident.Evidence}} + + + + + + + + + {{end}} +
TimeMethodHTTP / sourceOutcomeHeaders μsTotal μs
{{when .Timestamp}}{{.Method}}{{.Status}} / {{.Source}}{{.Outcome}}{{.HeadersUS}}{{.DurationUS}}
+
+

+ Snapshot may cover only part of the evaluation window. Current traffic + can expire independently. No matching evidence does not disprove the + alert. +

+
+ {{else}} +

{{.Empty}}

+ {{end}} + diff --git a/internal/alerting/receiver.go b/internal/alerting/receiver.go new file mode 100644 index 0000000..547f3b5 --- /dev/null +++ b/internal/alerting/receiver.go @@ -0,0 +1,364 @@ +package alerting + +import ( + "crypto/sha256" + "crypto/subtle" + _ "embed" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "html/template" + "io" + "math" + "net/http" + "net/url" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +type Rule struct{ Title, Signal, Unit string } + +var Rules = map[string]Rule{ + "NetkitUpstreamFailureRate": {"Sustained upstream HTTP 5xx", "http5xx", "fraction of HTTP observations"}, + "NetkitTransportFailureRate": {"Sustained proxy or upstream connection failures", "transport", "fraction of HTTP observations"}, + "NetkitHeaderLatencyHigh": {"High time to response headers (ordinary profile)", "headers", "seconds"}, + "NetkitUnavailable": {"Netkit scrape unavailable", "", "scrape up"}, + "NetkitInsufficientSamples": {"Insufficient samples; health is unknown", "", "HTTP observations"}, +} + +type Config struct { + Dir, Token, Instance, AdminURL, DashboardURL string + Capacity int + TTL time.Duration +} +type Incident struct { + ID string `json:"id"` + Rule string `json:"rule"` + Status string `json:"status"` + Started time.Time `json:"started"` + Created time.Time `json:"created"` + Updated time.Time `json:"updated"` + EndsAt time.Time `json:"ends_at,omitempty"` + From time.Time `json:"from"` + To time.Time `json:"to"` + Observed *float64 `json:"observed,omitempty"` + Threshold *float64 `json:"threshold,omitempty"` + EvidenceState string `json:"evidence_state"` + Matching int `json:"matching_retained"` + Evidence []Evidence `json:"evidence"` +} +type Receiver struct { + mu sync.Mutex + config Config + client *http.Client + now func() time.Time +} + +func New(config Config) (*Receiver, error) { + if len(config.Token) < 16 { + return nil, errors.New("incident webhook token must contain at least 16 characters") + } + if config.Instance == "" || len(config.Instance) > 128 { + return nil, errors.New("instance alias required (up to 128 characters)") + } + if config.Capacity < 1 || config.Capacity > 1000 || config.TTL < time.Minute || config.TTL > 30*24*time.Hour { + return nil, errors.New("incident capacity must be 1–1000 and TTL 1 minute–30 days") + } + for _, raw := range []string{config.AdminURL, config.DashboardURL} { + u, err := url.Parse(raw) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" { + return nil, errors.New("admin/dashboard URLs must be HTTP(S), without credentials, query or fragment") + } + } + if err := os.MkdirAll(config.Dir, 0700); err != nil { + return nil, err + } + info, err := os.Stat(config.Dir) + if err != nil { + return nil, err + } + if !info.IsDir() || info.Mode().Perm()&0077 != 0 { + return nil, errors.New("incident directory must be private (mode 0700)") + } + return &Receiver{config: config, now: time.Now, client: &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }}}, nil +} +func (r *Receiver) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/webhook", r.webhook) + mux.HandleFunc("/", r.view) + return mux +} +func (r *Receiver) Prune() error { r.mu.Lock(); defer r.mu.Unlock(); _, err := r.load(); return err } + +// load enforces both bounds. No labels or request content can become filesystem paths. +func (r *Receiver) load() ([]Incident, error) { + entries, err := os.ReadDir(r.config.Dir) + if err != nil { + return nil, err + } + incidents := make([]Incident, 0) + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") || len(entry.Name()) != 69 { + continue + } + if _, err := hex.DecodeString(strings.TrimSuffix(entry.Name(), ".json")); err != nil { + continue + } + info, err := entry.Info() + if err != nil { + return nil, err + } + if !info.Mode().IsRegular() || info.Size() > 128*1024 { + continue + } + path := filepath.Join(r.config.Dir, entry.Name()) + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var incident Incident + if err = json.Unmarshal(data, &incident); err != nil { + return nil, errors.New("invalid incident file") + } + if incident.ID != strings.TrimSuffix(entry.Name(), ".json") { + return nil, errors.New("incident filename mismatch") + } + if r.now().Sub(incident.Created) >= r.config.TTL { + if err := os.Remove(path); err != nil { + return nil, err + } + continue + } + incidents = append(incidents, incident) + } + sort.Slice(incidents, func(i, j int) bool { return incidents[i].Created.After(incidents[j].Created) }) + for len(incidents) > r.config.Capacity { + last := incidents[len(incidents)-1] + if err := os.Remove(filepath.Join(r.config.Dir, last.ID+".json")); err != nil { + return nil, err + } + incidents = incidents[:len(incidents)-1] + } + return incidents, nil +} +func (r *Receiver) save(incident Incident) error { + data, err := json.Marshal(incident) + if err != nil { + return err + } + file, err := os.CreateTemp(r.config.Dir, ".incident-") + if err != nil { + return err + } + name := file.Name() + defer func() { _ = os.Remove(name) }() + if _, err = file.Write(data); err != nil { + _ = file.Close() + return err + } + if err = file.Sync(); err != nil { + _ = file.Close() + return err + } + if err = file.Close(); err != nil { + return err + } + return os.Rename(name, filepath.Join(r.config.Dir, incident.ID+".json")) +} +func number(value string) *float64 { + n, err := strconv.ParseFloat(value, 64) + if err != nil || math.IsNaN(n) || math.IsInf(n, 0) { + return nil + } + return &n +} + +type webhookAlert struct { + Status string `json:"status"` + Labels map[string]string `json:"labels"` + Annotations map[string]string `json:"annotations"` + StartsAt time.Time `json:"startsAt"` + EndsAt time.Time `json:"endsAt"` +} + +func (r *Receiver) webhook(w http.ResponseWriter, req *http.Request) { + if req.Method != "POST" { + w.Header().Set("Allow", "POST") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if subtle.ConstantTimeCompare([]byte(req.Header.Get("Authorization")), []byte("Bearer "+r.config.Token)) != 1 { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + var payload struct { + Alerts []webhookAlert `json:"alerts"` + } + decoder := json.NewDecoder(http.MaxBytesReader(w, req.Body, 1<<20)) + if err := decoder.Decode(&payload); err != nil || len(payload.Alerts) == 0 || len(payload.Alerts) > 8 { + http.Error(w, "invalid webhook", http.StatusBadRequest) + return + } + if err := decoder.Decode(new(any)); err != io.EOF { + http.Error(w, "invalid webhook", http.StatusBadRequest) + return + } + for _, alert := range payload.Alerts { + if _, ok := Rules[alert.Labels["alertname"]]; !ok || alert.Labels["job"] != "netkit" || alert.Labels["instance"] != r.config.Instance || alert.StartsAt.IsZero() || alert.StartsAt.After(r.now().Add(time.Minute)) || (alert.Status != "firing" && alert.Status != "resolved") || (alert.Status == "resolved" && (alert.EndsAt.IsZero() || alert.EndsAt.Before(alert.StartsAt))) { + http.Error(w, "unsupported alert scope or lifecycle", http.StatusBadRequest) + return + } + } + r.mu.Lock() + defer r.mu.Unlock() + incidents, err := r.load() + if err != nil { + http.Error(w, "incident storage unavailable", http.StatusServiceUnavailable) + return + } + existing := make(map[string]Incident) + for _, i := range incidents { + existing[i.ID] = i + } + for _, alert := range payload.Alerts { + name := alert.Labels["alertname"] + hash := sha256.Sum256([]byte(name + "\n" + r.config.Instance + "\n" + alert.StartsAt.UTC().Format(time.RFC3339Nano))) + id := hex.EncodeToString(hash[:]) + incident, exists := existing[id] + if !exists { + incident = Incident{ID: id, Rule: name, Status: alert.Status, Started: alert.StartsAt, Created: r.now(), From: alert.StartsAt.Add(-5 * time.Minute), To: r.now(), Evidence: []Evidence{}, EvidenceState: "not_applicable"} + // Capture the first notification's retained evidence once, before eviction can change it. + if Rules[name].Signal != "" { + r.capture(&incident, Rules[name].Signal) + } + } + // An out-of-order firing retry must not reopen an already resolved incident. + if incident.Status != "resolved" || alert.Status == "resolved" { + incident.Status = alert.Status + incident.Updated = r.now() + if alert.Status == "resolved" { + incident.EndsAt = alert.EndsAt + } + if incident.Observed == nil { + incident.Observed = number(alert.Annotations["value"]) + } + if incident.Threshold == nil { + incident.Threshold = number(alert.Annotations["threshold"]) + } + } + if err := r.save(incident); err != nil { + http.Error(w, "incident storage unavailable", http.StatusServiceUnavailable) + return + } + existing[id] = incident + } + if _, err := r.load(); err != nil { + http.Error(w, "incident retention unavailable", http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusNoContent) +} +func (r *Receiver) capture(incident *Incident, signal string) { + incident.EvidenceState = "unavailable" + endpoint, err := url.Parse(strings.TrimRight(r.config.AdminURL, "/") + "/requests/evidence") + if err != nil { + return + } + q := url.Values{"from": {incident.From.Format(time.RFC3339Nano)}, "to": {incident.To.Format(time.RFC3339Nano)}, "signal": {signal}} + endpoint.RawQuery = q.Encode() + response, err := r.client.Get(endpoint.String()) + if err != nil { + return + } + defer func() { _ = response.Body.Close() }() + if response.StatusCode != http.StatusOK { + return + } + var data struct { + Records []Evidence `json:"records"` + Matching int `json:"matching_retained"` + } + if err := json.NewDecoder(io.LimitReader(response.Body, 128*1024)).Decode(&data); err != nil || len(data.Records) > 20 || data.Matching < len(data.Records) { + return + } + for _, e := range data.Records { + if !e.Valid() || !e.Matches(signal) || e.Timestamp.Before(incident.From) || e.Timestamp.After(incident.To) { + return + } + } + incident.Evidence = data.Records + incident.Matching = data.Matching + incident.EvidenceState = "captured" + if len(data.Records) == 0 { + incident.EvidenceState = "no_retained_matches" + } +} +func (r *Receiver) investigate(incident Incident) string { + u, _ := url.Parse(r.config.DashboardURL) + u.Path = strings.TrimRight(u.Path, "/") + "/" + u.RawQuery = url.Values{"from": {incident.From.Format(time.RFC3339Nano)}, "to": {incident.To.Format(time.RFC3339Nano)}, "signal": {Rules[incident.Rule].Signal}}.Encode() + return u.String() +} + +//go:embed incident.html +var incidentHTML string + +var page = template.Must(template.New("incidents").Funcs(template.FuncMap{"when": func(value time.Time) string { return value.UTC().Format("2006-01-02 15:04:05.000 UTC") }, "retention": func(value time.Duration) string { + if value%(24*time.Hour) == 0 { + return fmt.Sprintf("%d days", int(value.Hours()/24)) + } + return value.String() +}, "unit": func(name string) string { return Rules[name].Unit }, "title": func(name string) string { return Rules[name].Title }, "number": func(value *float64) string { + if value == nil { + return "not supplied" + } + return fmt.Sprintf("%.4g", *value) +}}).Parse(incidentHTML)) + +func (r *Receiver) view(w http.ResponseWriter, req *http.Request) { + if req.Method != "GET" || req.URL.Path != "/" { + http.NotFound(w, req) + return + } + r.mu.Lock() + incidents, err := r.load() + r.mu.Unlock() + if err != nil { + http.Error(w, "incident storage unavailable", http.StatusServiceUnavailable) + return + } + type item struct { + Incident Incident + Link string + } + items := []item{} + id := req.URL.Query().Get("id") + for _, incident := range incidents { + if id == "" || id == incident.ID { + items = append(items, item{incident, r.investigate(incident)}) + } + } + empty := "No incident notifications received. This does not imply healthy monitoring." + if id != "" && len(items) == 0 { + empty = "Incident unavailable or expired. Check the receiver retention policy; this is not a healthy status." + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; base-uri 'none'; frame-ancestors 'none'") + if id != "" && len(items) == 0 { + w.WriteHeader(http.StatusNotFound) + } + _ = page.Execute(w, struct { + Instance string + Capacity int + TTL time.Duration + Items []item + Empty string + }{r.config.Instance, r.config.Capacity, r.config.TTL, items, empty}) +} diff --git a/internal/alerting/receiver_test.go b/internal/alerting/receiver_test.go new file mode 100644 index 0000000..66c65eb --- /dev/null +++ b/internal/alerting/receiver_test.go @@ -0,0 +1,209 @@ +package alerting + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +const testToken = "synthetic-webhook-token-only" + +func newTestReceiver(t *testing.T) (*Receiver, *int) { + t.Helper() + now := time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC) + calls := new(int) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + *calls++ + if req.URL.Path != "/requests/evidence" { + t.Errorf("unexpected evidence path: %s", req.URL.Path) + } + _ = json.NewEncoder(w).Encode(map[string]any{"matching_retained": 1, "records": []map[string]any{{"id": strings.Repeat("a", 32), "timestamp": now.Add(-time.Minute), "method": "GET", "outcome": "complete", "status": 503, "source": "upstream", "headers_us": 100, "duration_us": 200, "url": "https://secret.invalid/token/fixture-secret", "request_body": "fixture-secret"}}}) + })) + t.Cleanup(upstream.Close) + r, err := New(Config{Dir: filepath.Join(t.TempDir(), "incidents"), Token: testToken, Instance: "netkit-local", AdminURL: upstream.URL, DashboardURL: "https://dashboard.example.test/tools/netkit", Capacity: 2, TTL: time.Hour}) + if err != nil { + t.Fatal(err) + } + r.now = func() time.Time { return now } + return r, calls +} +func deliver(r *Receiver, status string, start time.Time, annotations map[string]string) *httptest.ResponseRecorder { + alert := webhookAlert{Status: status, Labels: map[string]string{"alertname": "NetkitUpstreamFailureRate", "job": "netkit", "instance": "netkit-local", "secret": "fixture-secret"}, Annotations: annotations, StartsAt: start} + if status == "resolved" { + alert.EndsAt = r.now() + } + data, _ := json.Marshal(map[string]any{"alerts": []webhookAlert{alert}, "externalURL": "http://fixture-secret.invalid"}) + req := httptest.NewRequest("POST", "/webhook", bytes.NewReader(data)) + req.Header.Set("Authorization", "Bearer "+testToken) + w := httptest.NewRecorder() + r.Handler().ServeHTTP(w, req) + return w +} +func TestIncidentEvidenceLifecycleAndSanitization(t *testing.T) { + r, calls := newTestReceiver(t) + start := r.now().Add(-3 * time.Minute) + for i := 0; i < 2; i++ { + w := deliver(r, "firing", start, map[string]string{"value": "0.5", "threshold": "0.1", "description": "fixture-secret"}) + if w.Code != 204 { + t.Fatal(w.Code, w.Body.String()) + } + } + incidents, err := r.load() + if err != nil || len(incidents) != 1 { + t.Fatal(incidents, err) + } + original := incidents[0] + if *calls != 1 || len(original.Evidence) != 1 || original.EvidenceState != "captured" { + t.Fatal(original, *calls) + } + path := filepath.Join(r.config.Dir, original.ID+".json") + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if bytes.Contains(data, []byte("fixture-secret")) || bytes.Contains(data, []byte("request_body")) { + t.Fatalf("snapshot retained arbitrary content: %s", data) + } + info, _ := os.Stat(path) + if info.Mode().Perm() != 0600 { + t.Fatal(info.Mode()) + } + if w := deliver(r, "resolved", start, nil); w.Code != 204 { + t.Fatal(w.Code) + } + if w := deliver(r, "firing", start, nil); w.Code != 204 { + t.Fatal(w.Code) + } + incidents, _ = r.load() + if incidents[0].Status != "resolved" || *calls != 1 || incidents[0].Observed == nil || *incidents[0].Observed != .5 { + t.Fatal(incidents, *calls) + } + // Evidence survives a receiver restart, independently of proxy/history lifetime. + restored, err := New(r.config) + if err != nil { + t.Fatal(err) + } + restored.now = r.now + records, err := restored.load() + if err != nil || len(records) != 1 || len(records[0].Evidence) != 1 { + t.Fatal(records, err) + } + if !strings.Contains(r.investigate(original), "/tools/netkit/?") || !strings.Contains(r.investigate(original), "signal=http5xx") { + t.Fatal(r.investigate(original)) + } + w := httptest.NewRecorder() + r.Handler().ServeHTTP(w, httptest.NewRequest("GET", "/?id="+original.ID, nil)) + if w.Code != 200 || strings.Contains(w.Body.String(), "fixture-secret") || !strings.Contains(w.Body.String(), "resolved") { + t.Fatal(w.Code, w.Body.String()) + } +} +func TestIncidentBoundsAndExpiry(t *testing.T) { + r, _ := newTestReceiver(t) + base := r.now() + for i := 0; i < 3; i++ { + now := base.Add(time.Duration(i) * time.Minute) + r.now = func() time.Time { return now } + if w := deliver(r, "firing", base.Add(time.Duration(i)*time.Second), nil); w.Code != 204 { + t.Fatal(w.Code, w.Body.String()) + } + } + records, _ := r.load() + if len(records) != 2 { + t.Fatal(len(records)) + } + id := records[0].ID + r.now = func() time.Time { return base.Add(2 * time.Hour) } + if err := r.Prune(); err != nil { + t.Fatal(err) + } + records, _ = r.load() + if len(records) != 0 { + t.Fatal(records) + } + w := httptest.NewRecorder() + r.Handler().ServeHTTP(w, httptest.NewRequest("GET", "/?id="+id, nil)) + if w.Code != 404 || !strings.Contains(w.Body.String(), "expired") { + t.Fatal(w.Code, w.Body.String()) + } +} +func TestRejectUnauthorizedUnboundedOrForeignInput(t *testing.T) { + r, _ := newTestReceiver(t) + for _, tc := range []struct { + body, token string + code int + }{{`{"alerts":[]}`, "", 401}, {strings.Repeat("x", (1<<20)+1), testToken, 400}, {`{"alerts":[{"status":"firing","startsAt":"2026-09-22T11:59:00Z","labels":{"alertname":"NetkitUpstreamFailureRate","job":"netkit","instance":"another-instance"}}]}`, testToken, 400}} { + req := httptest.NewRequest("POST", "/webhook", strings.NewReader(tc.body)) + req.Header.Set("Authorization", "Bearer "+tc.token) + w := httptest.NewRecorder() + r.Handler().ServeHTTP(w, req) + if w.Code != tc.code { + t.Fatal(w.Code, tc.code) + } + } + records, _ := r.load() + if len(records) != 0 { + t.Fatal(records) + } +} +func TestUnavailableEvidenceDoesNotBecomeHealthy(t *testing.T) { + r, _ := newTestReceiver(t) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + })) + defer upstream.Close() + r.config.AdminURL = upstream.URL + if w := deliver(r, "firing", r.now().Add(-time.Minute), nil); w.Code != 204 { + t.Fatal(w.Code) + } + records, _ := r.load() + if records[0].EvidenceState != "unavailable" || records[0].Status != "firing" { + t.Fatal(records) + } +} +func TestConcurrentRetriesDeduplicate(t *testing.T) { + r, calls := newTestReceiver(t) + start := r.now().Add(-time.Minute) + var group sync.WaitGroup + for i := 0; i < 8; i++ { + group.Add(1) + go func() { + defer group.Done() + if w := deliver(r, "firing", start, nil); w.Code != 204 { + t.Error(w.Code) + } + }() + } + group.Wait() + records, _ := r.load() + if len(records) != 1 || *calls != 1 { + t.Fatal(len(records), *calls) + } +} +func TestEvidenceValidationAndFixedURL(t *testing.T) { + r, _ := newTestReceiver(t) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, `{"matching_retained":1,"records":[{"id":"fixture-secret","method":"fixture-secret"}]}`) + })) + defer upstream.Close() + r.config.AdminURL = upstream.URL + if w := deliver(r, "firing", r.now().Add(-time.Minute), map[string]string{"generatorURL": "http://fixture-secret.invalid", "value": "NaN", "threshold": "Inf"}); w.Code != 204 { + t.Fatal(w.Code) + } + records, _ := r.load() + if records[0].EvidenceState != "unavailable" || records[0].Observed != nil || records[0].Threshold != nil { + t.Fatal(records) + } + bad := r.config + bad.DashboardURL = "https://user:password@example.test/" + if _, err := New(bad); err == nil { + t.Fatal("accepted credentials in dashboard URL") + } +} diff --git a/internal/proxy/evidence.go b/internal/proxy/evidence.go new file mode 100644 index 0000000..519c873 --- /dev/null +++ b/internal/proxy/evidence.go @@ -0,0 +1,46 @@ +package proxy + +import ( + "encoding/json" + "net/http" + "time" + + "github.com/biancarosa/netkit/internal/alerting" +) + +// The incident boundary intentionally excludes URLs, headers, bodies and error text. +func (p *Proxy) handleEvidence(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + w.Header().Set("Allow", "GET") + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + from, err1 := time.Parse(time.RFC3339Nano, r.URL.Query().Get("from")) + to, err2 := time.Parse(time.RFC3339Nano, r.URL.Query().Get("to")) + signal := r.URL.Query().Get("signal") + if err1 != nil || err2 != nil || to.Before(from) || (signal != "http5xx" && signal != "transport" && signal != "headers") { + http.Error(w, "invalid evidence scope", http.StatusBadRequest) + return + } + records := p.history.GetRecords() + evidence := make([]alerting.Evidence, 0, 20) + matching := 0 + for _, record := range records { + if signal == "headers" && (record.UpstreamStartTime.IsZero() || record.UpstreamEndTime.IsZero() || record.UpstreamEndTime.Before(record.UpstreamStartTime)) { + continue + } + e := alerting.Evidence{ID: record.ID, Timestamp: record.Timestamp, Method: metricMethod(record.Method), Outcome: record.Outcome, Status: record.ResponseStatus, Source: record.ResponseSource, HeadersUS: record.UpstreamLatencyUs, DurationUS: record.TotalDurationUs} + if e.Valid() && e.Matches(signal) && !e.Timestamp.Before(from) && !e.Timestamp.After(to) { + matching++ + if len(evidence) < 20 { + evidence = append(evidence, e) + } + } + } + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "no-store") + _ = json.NewEncoder(w).Encode(struct { + Records []alerting.Evidence `json:"records"` + Matching int `json:"matching_retained"` + }{evidence, matching}) +} diff --git a/internal/proxy/evidence_test.go b/internal/proxy/evidence_test.go new file mode 100644 index 0000000..14800a3 --- /dev/null +++ b/internal/proxy/evidence_test.go @@ -0,0 +1,73 @@ +package proxy + +import ( + "encoding/json" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/biancarosa/netkit/internal/alerting" +) + +func TestEvidenceIsBoundedMetadataWithExactScope(t *testing.T) { + p := New(&Config{HistorySize: 100}) + now := time.Now().UTC() + for i := 0; i < 25; i++ { + p.history.AddRecord(RequestRecord{ID: strings.Repeat("a", 32), Timestamp: now, Method: "GET", URL: "http://example.test/token/fixture-secret", RequestBody: "fixture-secret", ResponseBody: "fixture-secret", RequestHeaders: map[string]string{"Authorization": "fixture-secret"}, ResponseStatus: 503, ResponseSource: "upstream", Outcome: "complete", ProxyStartTime: now, ProxyEndTime: now.Add(time.Millisecond)}) + } + p.history.AddRecord(RequestRecord{ID: strings.Repeat("b", 32), Timestamp: now, Method: "GET", ResponseStatus: 200, ResponseSource: "upstream", Outcome: "complete"}) + req := httptest.NewRequest("GET", "/requests/evidence?signal=http5xx&from="+now.Add(-time.Minute).Format(time.RFC3339Nano)+"&to="+now.Add(time.Minute).Format(time.RFC3339Nano), nil) + w := httptest.NewRecorder() + p.handleEvidence(w, req) + if w.Code != 200 || strings.Contains(w.Body.String(), "fixture-secret") || strings.Contains(w.Body.String(), "request_body") { + t.Fatal(w.Code, w.Body.String()) + } + var data struct { + Records []alerting.Evidence `json:"records"` + Matching int `json:"matching_retained"` + } + if err := json.Unmarshal(w.Body.Bytes(), &data); err != nil { + t.Fatal(err) + } + if len(data.Records) != 20 || data.Matching != 25 { + t.Fatal(len(data.Records), data.Matching) + } + for _, e := range data.Records { + if !e.Valid() || !e.Matches("http5xx") { + t.Fatal(e) + } + } + for _, method := range []string{"GET", "POST"} { + w = httptest.NewRecorder() + p.handleEvidence(w, httptest.NewRequest(method, "/requests/evidence", nil)) + if w.Code == 200 { + t.Fatal("invalid request accepted") + } + } +} + +func TestHeaderEvidenceRequiresObservedPhase(t *testing.T) { + p := New(&Config{HistorySize: 10}) + now := time.Now().UTC() + for _, observed := range []bool{false, true} { + record := RequestRecord{ID: strings.Repeat("c", 32), Timestamp: now, Method: "GET", ResponseStatus: 200, ResponseSource: "upstream", Outcome: "complete"} + if observed { + record.UpstreamStartTime = now + record.UpstreamEndTime = now.Add(time.Millisecond) + } + p.history.AddRecord(record) + } + w := httptest.NewRecorder() + p.handleEvidence(w, httptest.NewRequest("GET", "/requests/evidence?signal=headers&from="+now.Add(-time.Minute).Format(time.RFC3339Nano)+"&to="+now.Add(time.Minute).Format(time.RFC3339Nano), nil)) + var data struct { + Records []alerting.Evidence `json:"records"` + Matching int `json:"matching_retained"` + } + if err := json.Unmarshal(w.Body.Bytes(), &data); err != nil { + t.Fatal(err) + } + if w.Code != 200 || data.Matching != 1 || len(data.Records) != 1 { + t.Fatal(w.Code, w.Body.String()) + } +} diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 91ec165..06e968e 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -86,6 +86,7 @@ func New(config *Config) *Proxy { // Add request history endpoints adminMux.HandleFunc("/requests", proxy.handleRequestHistory) + adminMux.HandleFunc("/requests/evidence", proxy.handleEvidence) adminMux.HandleFunc("/requests/stats", proxy.handleRequestStats) adminMux.HandleFunc("/requests/clear", proxy.handleClearHistory) diff --git a/scripts/test-alerting-integration.py b/scripts/test-alerting-integration.py new file mode 100644 index 0000000..0b449d7 --- /dev/null +++ b/scripts/test-alerting-integration.py @@ -0,0 +1,168 @@ +#!/usr/bin/env python3 +"""Exercise a real Alertmanager and netkit alerts receiver with disposable local data.""" +import argparse +import datetime as dt +import json +import os +from pathlib import Path +import socket +import subprocess +import tempfile +import threading +import time +import urllib.request +import urllib.error +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + + +def port(): + with socket.socket() as listener: + listener.bind(('127.0.0.1', 0)) + return listener.getsockname()[1] + + +def stamp(value=None): + return (value or dt.datetime.now(dt.timezone.utc)).isoformat().replace('+00:00', 'Z') + + +def request(url, data=None, method=None, token=None): + headers = {'Content-Type': 'application/json'} + if token: + headers['Authorization'] = 'Bearer ' + token + payload = None if data is None else json.dumps(data).encode() + with urllib.request.urlopen(urllib.request.Request(url, payload, headers, method=method), timeout=5) as response: + return response.read() + + +def until(predicate, message, timeout=15): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + if predicate(): + return + except (OSError, urllib.error.URLError): + pass + time.sleep(.1) + raise AssertionError(message) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--netkit', required=True) + parser.add_argument('--alertmanager', required=True) + args = parser.parse_args() + receiver_port, am_port = port(), port() + receiver_url = f'http://127.0.0.1:{receiver_port}' + am_url = f'http://127.0.0.1:{am_port}' + token = 'disposable-integration-token' + delivered = [] + lock = threading.Lock() + evidence_time = stamp() + + class Fixture(BaseHTTPRequestHandler): + def do_GET(self): + body = json.dumps({'matching_retained': 1, 'records': [{ + 'id': 'a' * 32, 'timestamp': evidence_time, 'method': 'GET', + 'outcome': 'complete', 'status': 503, 'source': 'upstream', + 'headers_us': 2000, 'duration_us': 3000, + 'url': 'https://fixture.invalid/secret/never-store-me', + }]}).encode() + self.send_response(200) + self.end_headers() + self.wfile.write(body) + + def do_POST(self): + payload = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + try: + request(receiver_url + '/webhook', payload, token=token) + except urllib.error.URLError: + self.send_response(503) + self.end_headers() + return + with lock: + delivered.append(payload) + self.send_response(204) + self.end_headers() + + def log_message(self, *_): + pass + + fixture = ThreadingHTTPServer(('127.0.0.1', 0), Fixture) + threading.Thread(target=fixture.serve_forever, daemon=True).start() + fixture_url = f'http://127.0.0.1:{fixture.server_port}' + children = [] + try: + with tempfile.TemporaryDirectory(prefix='netkit-alert-integration-') as work: + root = Path(work) + config = root / 'alertmanager.yml' + config.write_text(f'''route: + receiver: test + group_by: [alertname, job, instance] + group_wait: 1s + group_interval: 1s + repeat_interval: 3s +receivers: + - name: test + webhook_configs: + - url: {fixture_url}/webhook + send_resolved: true +''') + with (root / 'process.log').open('w') as output: + children.append(subprocess.Popen([ + args.netkit, 'alerts', '--listen', f'127.0.0.1:{receiver_port}', + '--dir', str(root / 'incidents'), '--admin-url', fixture_url, + '--dashboard-url', 'http://127.0.0.1:18080', + ], env={**os.environ, 'NETKIT_ALERT_TOKEN': token}, stdout=output, stderr=output)) + children.append(subprocess.Popen([ + args.alertmanager, '--config.file', str(config), + '--storage.path', str(root / 'am'), '--cluster.listen-address=', + '--web.listen-address', f'127.0.0.1:{am_port}', + ], stdout=output, stderr=output)) + until(lambda: request(am_url + '/-/ready') and request(receiver_url + '/'), 'services not ready') + start = dt.datetime.now(dt.timezone.utc) + alerts = [{ + 'labels': {'alertname': 'NetkitUpstreamFailureRate', 'job': 'netkit', + 'instance': 'netkit-local', 'test_case': case}, + 'annotations': {'value': '0.5', 'threshold': '0.1', 'description': 'never-store-me'}, + 'startsAt': stamp(start), 'endsAt': stamp(start + dt.timedelta(minutes=5)), + } for case in ('a', 'b')] + request(am_url + '/api/v2/alerts', alerts) + until(lambda: len(delivered) >= 1, 'initial grouped notification missing') + assert len(delivered[0]['alerts']) == 2, 'alerts were not grouped' + until(lambda: len(delivered) >= 2, 'repeat notification missing') + files = list((root / 'incidents').glob('*.json')) + assert len(files) == 1, 'retries duplicated the logical incident' + incident = json.loads(files[0].read_text()) + assert incident['status'] == 'firing' and len(incident['evidence']) == 1 + assert 'never-store-me' not in files[0].read_text() + silence = json.loads(request(am_url + '/api/v2/silences', { + 'matchers': [{'name': 'instance', 'value': 'netkit-local', 'isRegex': False, 'isEqual': True}], + 'startsAt': stamp(), 'endsAt': stamp(start + dt.timedelta(minutes=2)), + 'createdBy': 'local-test', 'comment': 'disposable integration silence', + })) + time.sleep(1.5) # Drain any notification already in flight before silence creation. + count = len(delivered) + time.sleep(4) + assert len(delivered) == count, 'silence did not suppress repeated notifications' + request(am_url + '/api/v2/silence/' + silence['silenceID'], method='DELETE') + for alert in alerts: + alert['endsAt'] = stamp() + request(am_url + '/api/v2/alerts', alerts) + until(lambda: any(p['status'] == 'resolved' for p in delivered), 'recovery notification missing') + until(lambda: json.loads(files[0].read_text())['status'] == 'resolved', 'incident did not resolve') + assert json.loads(files[0].read_text())['evidence'] == incident['evidence'], 'recovery replaced original evidence' + print('PASS: grouping, repeat delivery, silence, recovery, durable deduplication and metadata-only evidence') + for child in children: + child.terminate() + for child in children: + child.wait(timeout=10) + finally: + for child in children: + if child.poll() is None: + child.kill() + child.wait() + fixture.shutdown() + + +if __name__ == '__main__': + main()