Notifications: webhooks for a parked model, a lost node, a spent budget and new weights; Alertmanager receiver templates

notify.webhooks sends signed JSON or Slack-compatible messages, queued,
retried, de-duplicated hourly; ctl notify test checks each webhook.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-10-04 16:58:11 +02:00
1 parent 646ff996c7
commit 1b71c38f89
24 files changed
+907 -4

No files matched your search

+7 -1
View File
@@ -401,7 +401,13 @@ portal is not done — the page shows; the console and ctl change.
console). Left: Parquet, and a scheduled push to S3/SFTP. **Usage export (S).** CSV/Parquet by org/team/project/model/period from
the rollups (`internal/store/usageseries.go`), `GET
/api/v1/usage/export`, optional scheduled push to S3/SFTP.
5. **Alerting out of the box (S).** `deploy/monitoring/alertmanager.yml`
5. **Done (branch `features-c`):** `notify.webhooks` (model parked, node
lost, budget spent, update available; signed JSON or Slack; retries,
hourly de-duplication), `ctl notify test`, Alertmanager receiver
templates. Left: `server init` and the console do not yet ask for a
receiver — the templates are filled in by hand; each control plane sends
budget events on its own, so two behind a balancer may both send one.
**Alerting out of the box (S).** `deploy/monitoring/alertmanager.yml`
ships `receiver: none`; add Slack/email/webhook/PagerDuty receiver
templates driven by `server init` or the console; llmgw-native webhooks
for model parked, budget spent, update available (from the reconciler
+15
View File
@@ -71,6 +71,21 @@ APIs change between minors without a deprecation period.
- **Other ways in:** `GET /api/v1/usage/export` (CSV, or `format=json`)
and **E** on the console's Usage view.
### Notifications
- **Webhooks for what needs a person** (`notify.webhooks`): a model parked,
a node lost, a budget spent, new weights in a model's repository — sent by
the gateway itself, with no Prometheus in between.
- **Formats:** JSON signed with HMAC-SHA256 (`X-Llmgw-Signature`) when
`secret_file` is set, or a Slack-compatible `{"text": …}`.
- **Delivery:** queued, never holding up what raised it; tried three
times; the same event about the same thing at most once an hour, and new
weights once per revision.
- **Checking:** `llmgw ctl notify test`.
- **Alertmanager receiver templates** for Slack, e-mail, PagerDuty and a
webhook in `deploy/monitoring/alertmanager.yml`, commented out until
chosen.
### Guardrails
- **Hooks for guardrails on the request path** (nothing judges yet; B2 adds
+50 -2
View File
@@ -1,10 +1,11 @@
# Alertmanager for the llmgw cluster.
#
# Nothing is sent anywhere yet, on purpose. Where an alert should go — an
# Nothing is sent anywhere until a receiver is chosen, on purpose. Where an alert should go — an
# address, a Slack workspace, a webhook — is a decision with credentials
# attached, and a monitoring system that quietly starts mailing people is worse
# than one that is visibly not wired up. Alerts group and fire here and are
# visible at :9093 and in Prometheus; add a receiver below to make them travel.
# visible at :9093 and in Prometheus; uncomment a receiver template below and
# name it on the route to make them travel.
global:
resolve_timeout: 5m
@@ -31,6 +32,53 @@ route:
receivers:
- name: none
# Receiver templates. Uncomment one, fill in where it goes, and name it in
# `receiver:` above — on the route, the critical branch, or both. Secrets go
# in files Alertmanager reads (the *_file settings), mode 0640 owned by
# root:prometheus, not in this file. `amtool check-config` before a restart.
#
# Slack (an incoming webhook):
# - name: slack
# slack_configs:
# - api_url_file: /etc/prometheus/slack-webhook-url
# channel: '#llmgw'
# send_resolved: true
# title: '{{ .Status | toUpper }}: {{ .CommonLabels.alertname }}'
# text: >-
# {{ range .Alerts }}{{ .Annotations.summary }}
# {{ .Annotations.description }}
# {{ end }}
#
# E-mail (through a relay that takes authenticated submission):
# - name: email
# email_configs:
# - to: ops@example.com
# from: llmgw-alerts@example.com
# smarthost: smtp.example.com:587
# auth_username: llmgw-alerts@example.com
# auth_password_file: /etc/prometheus/smtp-password
# send_resolved: true
#
# PagerDuty (Events API v2), for the critical branch — what should wake
# someone:
# - name: pagerduty
# pagerduty_configs:
# - routing_key_file: /etc/prometheus/pagerduty-routing-key
# severity: '{{ .CommonLabels.severity }}'
# description: '{{ .CommonAnnotations.summary }}'
#
# A webhook of your own (Alertmanager's JSON, version 4):
# - name: webhook
# webhook_configs:
# - url_file: /etc/prometheus/alert-webhook-url
# send_resolved: true
# max_alerts: 20
#
# The gateway itself also sends a few events straight to webhooks, with no
# Prometheus in between — a model parked, a node lost, a budget spent, new
# weights published: `notify.webhooks` in server.yaml (docs/install.md,
# "Being told").
inhibit_rules:
# A node that is down explains every replica on it. Reporting both says the
# same thing twice and buries whichever is the actual cause.
+35
View File
@@ -927,6 +927,41 @@ fleet means the measurement is missing, not that the fleet is idle. And
`llmgw_gpu_nvlink_peers` at zero everywhere means the topology is unknown,
which is what a binary built without the nvml tag reports.
### Being told
Prometheus alerts travel once `alertmanager.yml` names a receiver; it ships
templates for Slack, e-mail, PagerDuty and a webhook of your own, commented
out until you choose.
A few things are worth a message even without a monitoring stack, and the
control plane sends those itself:
| Event | When |
|---|---|
| `model_parked` | A model failed every start in a row and is no longer retried (`ctl retry` tries again). |
| `node_lost` | A node was silent past `control.lost_after`; its replicas are rebuilt elsewhere. |
| `budget_spent` | A key, project, team or organization reached its budget; priced models are refused. |
| `update_available` | A model's registry repository published different weights. Once per revision. |
```yaml
notify:
webhooks:
- url: https://hooks.slack.com/services/… # Slack, Mattermost, Rocket.Chat
format: slack
events: [model_parked, node_lost]
- url: https://ops.example.com/llmgw # your own receiver
secret_file: /etc/llmgw/notify-secret # signs the body
```
A `json` webhook (the default) gets the event as JSON — `kind`, `at`,
`subject`, `message`, `detail` — with `X-Llmgw-Event` naming the kind and,
when `secret_file` is set, `X-Llmgw-Signature: sha256=<hex>`, the HMAC-SHA256
of the body under the file's contents; check it before believing the body. A
webhook that does not answer is tried three times and then given up on, and the
same event about the same thing is sent at most once an hour, so a budget that
refuses a thousand requests is one message. `llmgw ctl notify test` sends a
test event to each webhook and says how each answered.
## PostgreSQL, when SQLite is no longer the right shape
SQLite is the default and stays the right answer for most installations: one
+56
View File
@@ -230,6 +230,7 @@ rather than repeated.
| `engine_overrides` | `[]EngineOverride` | [Server](#server) |
| `backup_dir` | `string` | [Server](#server) |
| `guard` | `guard.Config` | [Server](#server) |
| `notify` | `notify.Config` | [Server](#server) |
| `firewall_networks` | `[]string` | [Server](#server) |
| `backup_recipients` | `[]string` | [Server](#server) |
| `releases` | `string` | [Server](#server) |
@@ -256,6 +257,10 @@ rather than repeated.
| `bin` | `string` | [VLLMConfig](#vllmconfig) |
| `launch_mode` | `string` | [VLLMConfig](#vllmconfig) |
| `ray_bin` | `string` | [VLLMConfig](#vllmconfig) |
| `url` | `string` | [Webhook](#webhook) |
| `format` | `string` | [Webhook](#webhook) |
| `events` | `[]string` | [Webhook](#webhook) |
| `secret_file` | `string` | [Webhook](#webhook) |
| `prompt_per_1k` | `int64` | [accounting.Price](#accountingprice) |
| `completion_per_1k` | `int64` | [accounting.Price](#accountingprice) |
| `name` | `string` | [admission.Band](#admissionband) |
@@ -298,6 +303,8 @@ rather than repeated.
| `sim` | `SimConfig` | [engines.Config](#enginesconfig) |
| `default` | `[]string` | [guard.Config](#guardconfig) |
| `policies` | `map[string]Policy` | [guard.Config](#guardconfig) |
| `webhooks` | `[]Webhook` | [notify.Config](#notifyconfig) |
| `host` | `string` | [notify.Config](#notifyconfig) |
| `work_dir` | `string` | [supervisor.Config](#supervisorconfig) |
| `log_lines` | `int` | [supervisor.Config](#supervisorconfig) |
| `stop_timeout` | `time.Duration` | [supervisor.Config](#supervisorconfig) |
@@ -1868,6 +1875,16 @@ guard.default, and to parts of the tenancy tree by attaching them
Takes a [guard.Config](#guardconfig) block.
### `notify`
`notify.Config`
Notify sends what needs a person — a model parked, a node lost, a
budget spent, new weights published — to webhooks, signed, without a
Prometheus and Alertmanager in between. Empty sends nothing.
Takes a [notify.Config](#notifyconfig) block.
### `firewall_networks`
`[]string`
@@ -2056,6 +2073,33 @@ node and has not been checked against a real cluster.
RayBin is used when LaunchMode is "ray".
## Webhook
### `url`
`string`
### `format`
`string`
Format is json (the event, with X-Llmgw-Signature when SecretFile is
set) or slack ({"text": …}, which Slack, Mattermost and Rocket.Chat
incoming webhooks take).
### `events`
`[]string`
Events are the kinds sent here; empty is all.
### `secret_file`
`string`
SecretFile holds the key the body is signed with: X-Llmgw-Signature
is sha256=<hex HMAC of the body>.
## accounting.Price
### `prompt_per_1k`
@@ -2340,6 +2384,18 @@ Takes a [SimConfig](#simconfig) block.
`map[string]Policy`
## notify.Config
### `webhooks`
`[]Webhook`
### `host`
`string`
Host names the gateway in messages; empty is this machine's name.
## supervisor.Config
### `work_dir`
+11
View File
@@ -6,9 +6,11 @@ import (
"fmt"
"log/slog"
"strings"
"sync/atomic"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/auth"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
@@ -45,8 +47,13 @@ type Accountant struct {
tokens *Limiter
recorder *Recorder
log *slog.Logger
// notify is told when a budget runs out; nil tells no one.
notify atomic.Pointer[notify.Notifier]
}
// SetNotifier gives the accountant somewhere to say a budget is spent.
func (a *Accountant) SetNotifier(n *notify.Notifier) { a.notify.Store(n) }
// New builds an accountant.
func New(pricing *Pricing, recorder *Recorder, log *slog.Logger) *Accountant {
if log == nil {
@@ -182,6 +189,10 @@ func (a *Accountant) Check(p auth.Principal, model string, estimatedTokens int64
if !price.Zero() {
for _, l := range chain {
if l.Budget != nil && a.budgets.Exhausted(l.Counter, l.Period, *l.Budget) {
a.notify.Load().Notify(notify.BudgetSpent, l.Counter,
fmt.Sprintf("the %s budget of %s is spent: requests to priced models are refused until it renews",
or(string(l.Period), "total"), l.Counter),
map[string]any{"counter": l.Counter, "period": string(l.Period), "budget": *l.Budget})
return res, &LimitError{Err: ErrBudgetSpent}
}
}
+27
View File
@@ -5,11 +5,14 @@ import (
"errors"
"log/slog"
"path/filepath"
"strings"
"sync"
"testing"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/auth"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify/notifytest"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
@@ -677,3 +680,27 @@ func TestAFallbackIsChargedWhatTheModelThatAnsweredCosts(t *testing.T) {
t.Error("the paid model answering cost nothing")
}
}
func TestASpentBudgetTellsSomeoneOnce(t *testing.T) {
a := New(pricing(t), nil, slog.New(slog.DiscardHandler))
n, sent := notifytest.Catch(t)
a.SetNotifier(n)
p := principal("k1", func(e *store.Effective) {
e.Budget = i64p(100)
e.BudgetPeriod = store.PeriodMonth
})
res, err := a.Check(p, "paid", 10)
if err != nil {
t.Fatal(err)
}
a.Record(res, store.Usage{KeyID: "k1", PromptTokens: 1000})
for range 3 {
if _, err := a.Check(p, "paid", 10); !errors.Is(err, ErrBudgetSpent) {
t.Fatalf("not refused: %v", err)
}
}
got := sent()
if len(got) != 1 || got[0].Kind != notify.BudgetSpent || !strings.Contains(got[0].Message, "month") {
t.Fatalf("sent %+v, want one budget_spent", got)
}
}
+1
View File
@@ -162,6 +162,7 @@ func newCtlCmd() *cobra.Command {
newQuotaCmd(client, &asJSON),
newGuardCmd(client, &asJSON),
newAuditCmd(client, &asJSON),
newNotifyCmd(client, &asJSON),
newUsageCmd(client, &asJSON),
newCertCmd(),
)
+48
View File
@@ -0,0 +1,48 @@
package cli
import (
"encoding/json"
"fmt"
"github.com/spf13/cobra"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
)
func newNotifyCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
c := &cobra.Command{
Use: "notify",
Short: "The webhooks told about parked models, lost nodes, spent budgets and new weights",
}
c.AddCommand(&cobra.Command{
Use: "test",
Short: "Send a test notification to every webhook in server.yaml's notify section",
Long: "Sends one event of kind \"test\" to each webhook, once, and says how each\n" +
"answered: the way to check an address, a secret and a firewall before\n" +
"something real has to travel.",
Args: cobra.NoArgs,
RunE: func(cmd *cobra.Command, _ []string) error {
var res []notify.Result
if err := client().post(cmd.Context(), "/api/v1/notify/test", nil, &res); err != nil {
return err
}
if *asJSON {
return json.NewEncoder(cmd.OutOrStdout()).Encode(res)
}
failed := 0
for _, r := range res {
if r.Err != "" {
failed++
fmt.Fprintf(cmd.OutOrStdout(), "%s failed: %s\n", r.URL, r.Err)
} else {
fmt.Fprintf(cmd.OutOrStdout(), "%s delivered\n", r.URL)
}
}
if failed > 0 {
return fmt.Errorf("%d of %d webhooks were not reached", failed, len(res))
}
return nil
},
})
return c
}
+14 -1
View File
@@ -44,6 +44,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/obs"
"gitea.mynodes.xyz/ilker/llmgw/internal/promq"
"gitea.mynodes.xyz/ilker/llmgw/internal/router"
@@ -153,6 +154,16 @@ func newServerCmd() *cobra.Command {
rt.Start(ctx)
defer rt.Stop()
// Webhooks for what needs a person, when server.yaml names any.
notifier, err := notify.New(ctx, cfg.Notify, log)
if err != nil {
return err
}
rt.SetNotifier(notifier)
if notifier != nil {
log.Info("notifications on", "webhooks", len(cfg.Notify.Webhooks))
}
// The control plane replaces the static endpoint list: agents
// register, the reconciler places replicas, and Ready ones are
// published to the router (plan §6.4).
@@ -180,6 +191,7 @@ func newServerCmd() *cobra.Command {
return err
}
defer ctl.Close()
ctl.Reconciler().SetNotifier(notifier)
admin = ctl
agentPath, agentHandler = svc.Handler()
@@ -271,6 +283,7 @@ func newServerCmd() *cobra.Command {
}
adminAPI = api
api.SetReleases(cfg.Releases)
api.SetNotifier(notifier)
if st := rt.Store(); st != nil && gate != nil {
api.SetGuard(&adminapi.GuardAdmin{Library: cfg.Guard, Store: st, Invalidate: gate.Invalidate})
}
@@ -383,7 +396,7 @@ func newServerCmd() *cobra.Command {
// changes it, and what replaces replicas.
Current: func() []catalog.Model { return admin.Catalog().Models() },
Models: catalogSvc, Restart: admin.Reconciler().Restart,
Pool: adminAPI.PoolNodes,
Pool: adminAPI.PoolNodes, Notify: notifier,
})
adminAPI.SetServe(serve)
updateMetrics.Set(serve)
+9
View File
@@ -19,6 +19,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
)
@@ -128,6 +129,11 @@ type Server struct {
// (llmgw ctl guard attach). Empty checks nothing.
Guard guard.Config `koanf:"guard"`
// Notify sends what needs a person — a model parked, a node lost, a
// budget spent, new weights published — to webhooks, signed, without a
// Prometheus and Alertmanager in between. Empty sends nothing.
Notify notify.Config `koanf:"notify"`
// FirewallNetworks are networks, besides those each machine reaches the
// cluster on, that llmgw's internal ports answer: a monitoring host on
// another subnet, nodes behind a router. Written into every machine's
@@ -656,6 +662,9 @@ func (s Server) Validate() error {
if _, err := s.FirewallAllow(); err != nil {
return err
}
if err := s.Notify.Validate(); err != nil {
return err
}
if err := s.Guard.Validate(); err != nil {
return err
}
+4
View File
@@ -35,11 +35,14 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/nvlink"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/topology"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/store"
)
// API serves the management endpoints.
type API struct {
// notifier is the webhooks `ctl notify test` reaches; nil without any.
notifier *notify.Notifier
// releases is where release binaries are fetched from, for upgrades.
releases string
// storeStatus reads a replicated store's members; nil for one database.
@@ -84,6 +87,7 @@ func (a *API) Register(mux *http.ServeMux) {
a.registerGuard(mux)
mux.HandleFunc("GET /api/v1/audit", a.auditEvents)
mux.HandleFunc("GET /api/v1/usage/export", a.usageExport)
mux.HandleFunc("POST /api/v1/notify/test", a.notifyTest)
mux.HandleFunc("POST /api/v1/models/{name}/adapters", a.setAdapters)
mux.HandleFunc("GET /api/v1/models/{name}/target", a.target)
mux.HandleFunc("POST /api/v1/models/{name}/class", func(w http.ResponseWriter, r *http.Request) {
+19
View File
@@ -0,0 +1,19 @@
package adminapi
import (
"net/http"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
)
// SetNotifier gives the API the webhooks `ctl notify test` reaches.
func (a *API) SetNotifier(n *notify.Notifier) { a.notifier = n }
// notifyTest sends a test event to every webhook and says how each answered.
func (a *API) notifyTest(w http.ResponseWriter, r *http.Request) {
if a.notifier == nil {
writeError(w, http.StatusNotFound, "no webhooks: server.yaml has no notify.webhooks")
return
}
writeJSON(w, http.StatusOK, a.notifier.Test(r.Context()))
}
@@ -31,6 +31,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/modelplan"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
)
// Registry reads a repository.
@@ -85,6 +86,9 @@ type Deps struct {
// UnplacedGrace is how long a model may go unplaced before the job
// stops waiting for it: a minute unless set.
UnplacedGrace time.Duration
// Notify is told when a model's repository publishes new weights; nil
// tells no one.
Notify *notify.Notifier
}
// errBusy is a model the cluster has the hardware for, all of it in use now.
+9
View File
@@ -16,6 +16,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/gguf"
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"github.com/prometheus/client_golang/prometheus"
)
@@ -343,8 +344,16 @@ func (s *Server) checkOne(ctx context.Context, m catalog.Model, o *catalog.Origi
}
}
s.mu.Lock()
prev, had := s.updates[m.Name]
s.updates[m.Name] = u
s.mu.Unlock()
// Told once per new revision, not at every check that finds it still there.
if u.Available && (!had || !prev.Available || prev.Latest != u.Latest) {
s.deps.Notify.Notify(notify.UpdateAvailable, m.Name+"@"+u.Latest,
fmt.Sprintf("%s has new weights at %s (%s); `llmgw ctl model update %s` moves to them",
m.Name, o.Repo, catalog.ShortRevision(u.Latest), m.Name),
map[string]any{"model": m.Name, "repo": o.Repo, "current": o.Revision, "latest": u.Latest})
}
return u
}
@@ -14,6 +14,8 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler"
"gitea.mynodes.xyz/ilker/llmgw/internal/hf"
"gitea.mynodes.xyz/ilker/llmgw/internal/mirror"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify/notifytest"
)
const (
@@ -338,3 +340,15 @@ func TestAnOriginGivenByHandIsChecked(t *testing.T) {
t.Fatalf("%+v", o)
}
}
func TestNewWeightsAreToldOncePerRevision(t *testing.T) {
s, _, _, _, _ := updSetup(t)
n, sent := notifytest.Catch(t)
s.deps.Notify = n
s.CheckUpdates(context.Background())
s.CheckUpdates(context.Background()) // still there: not told again
got := sent()
if len(got) != 1 || got[0].Kind != notify.UpdateAvailable || got[0].Detail["model"] != "qwen" {
t.Fatalf("sent %+v, want one update_available for qwen", got)
}
}
@@ -0,0 +1,46 @@
package reconciler
import (
"testing"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify/notifytest"
)
func TestParkingAModelTellsSomeone(t *testing.T) {
h := newHarness(t, testModel(1), gpuNode("1", 8))
n, sent := notifytest.Catch(t)
h.r.SetNotifier(n)
failRepeatedly(t, h, DefaultConfig().ParkAfter, "unknown model architecture: 'ggmlc'")
h.advance(DefaultConfig().MaxBackoff + time.Second)
h.r.Reconcile(t.Context())
got := sent()
if len(got) != 1 || got[0].Kind != notify.ModelParked || got[0].Subject != "llama" {
t.Fatalf("sent %+v, want one model_parked for llama", got)
}
}
func TestLosingANodeTellsSomeone(t *testing.T) {
h := newHarness(t, testModel(1), gpuNode("1", 8), gpuNode("2", 8))
n, sent := notifytest.Catch(t)
h.r.SetNotifier(n)
h.r.Reconcile(t.Context())
h.markReady()
h.r.Reconcile(t.Context())
for _, d := range []time.Duration{20 * time.Second, 90 * time.Second} {
h.clock.advance(d)
touchAt(h.inv, "2", h.clock.now())
h.r.Reconcile(t.Context())
}
if n, _ := h.inv.Get("1"); n.State != inventory.Lost {
t.Fatalf("node 1 is %s", n.State)
}
got := sent()
if len(got) != 1 || got[0].Kind != notify.NodeLost || got[0].Subject != "1" {
t.Fatalf("sent %+v, want one node_lost for 1", got)
}
}
+18
View File
@@ -8,6 +8,7 @@ import (
"math/rand/v2"
"slices"
"sync"
"sync/atomic"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/engines"
@@ -16,6 +17,7 @@ import (
"gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/placement"
"gitea.mynodes.xyz/ilker/llmgw/internal/control/quota"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
)
// ShardCommand is everything an agent needs to launch one shard.
@@ -231,6 +233,9 @@ type Reconciler struct {
disp Dispatcher
log *slog.Logger
now func() time.Time
// notify tells people about parked models and lost nodes; nil tells
// no one.
notify atomic.Pointer[notify.Notifier]
mu sync.Mutex
replicas map[string]*Replica
@@ -559,6 +564,9 @@ func (r *Reconciler) expireNodes() {
r.inv.SetState(n.ID, inventory.Lost)
r.log.Error("node lost; its replicas will be rebuilt elsewhere",
"node", n.ID, "silent_for", silent.Round(time.Second))
r.notify.Load().Notify(notify.NodeLost, n.ID,
fmt.Sprintf("node %s was lost: silent for %s; its replicas are rebuilt elsewhere", n.ID, silent.Round(time.Second)),
map[string]any{"node": n.ID, "silent_seconds": int(silent.Seconds())})
r.failReplicasOn(n.ID, "node "+n.ID+" was lost")
}
}
@@ -644,6 +652,10 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
r.log.Error("model parked: it failed every time it started, and is not retried "+
"until its definition changes or `llmgw ctl retry` is run",
"model", rep.Model, "failures", n, "last_error", rep.Err)
r.notify.Load().Notify(notify.ModelParked, rep.Model,
fmt.Sprintf("model %s is parked: it failed %d times in a row (%s); `llmgw ctl retry %s` tries again",
rep.Model, n, rep.Err, rep.Model),
map[string]any{"model": rep.Model, "failures": n, "last_error": rep.Err})
}
}
}
@@ -1216,3 +1228,9 @@ func (r *Reconciler) engineKey(replicaID string) string {
}
return r.cfg.EngineKey(replicaID)
}
// SetNotifier gives the reconciler somewhere to say a model was parked or a
// node lost.
func (r *Reconciler) SetNotifier(n *notify.Notifier) {
r.notify.Store(n)
}
+11
View File
@@ -284,6 +284,7 @@ var Reference = []Block{
{Key: "engine_overrides", Type: "[]EngineOverride", Doc: "EngineOverrides give some nodes different engines from the rest: a\nvLLM built for CUDA 12 on the Volta cards CUDA 13 dropped, or none at\nall on nodes meant only for llama.cpp. Each picks nodes and lays its\nengines over the cluster's, in order, later ones last; an empty version\nremoves that engine from the nodes it picks.", Nested: ""},
{Key: "backup_dir", Type: "string", Doc: "BackupDir is where `llmgw server backup` writes, and where the newest\narchive's age is read for llmgw_backup_last_success_timestamp_seconds.\nEmpty means /var/backups/llmgw-server.", Nested: ""},
{Key: "guard", Type: "guard.Config", Doc: "Guard is the library of guardrail policies: what requests and answers\nare checked for, and what is done. Applied to every request through\nguard.default, and to parts of the tenancy tree by attaching them\n(llmgw ctl guard attach). Empty checks nothing.", Nested: "guard.Config"},
{Key: "notify", Type: "notify.Config", Doc: "Notify sends what needs a person — a model parked, a node lost, a\nbudget spent, new weights published — to webhooks, signed, without a\nPrometheus and Alertmanager in between. Empty sends nothing.", Nested: "notify.Config"},
{Key: "firewall_networks", Type: "[]string", Doc: "FirewallNetworks are networks, besides those each machine reaches the\ncluster on, that llmgw's internal ports answer: a monitoring host on\nanother subnet, nodes behind a router. Written into every machine's\nfirewall by `server init`, `node add` and `server add`.", Nested: ""},
{Key: "backup_recipients", Type: "[]string", Doc: "BackupRecipients encrypt every archive `llmgw server backup` writes, to\nwhoever holds one of their keys: age recipients (age1…) or SSH public\nkeys (ssh-ed25519 …, ssh-rsa …), as `age` takes them. An archive holds\nthe certificate authority's key and every secret in the store, and is\ncopied off the machine; encrypted, the copy is safe wherever it lands.\nEmpty writes archives unencrypted, readable by root only.", Nested: ""},
{Key: "releases", Type: "string", Doc: "Releases is where `llmgw ctl upgrade` fetches release binaries from,\nfor this machine and — through their agents — every node. Empty means\nthe project's own releases.", Nested: ""},
@@ -323,6 +324,12 @@ var Reference = []Block{
{Key: "launch_mode", Type: "string", Doc: "LaunchMode selects the multi-node mechanism: \"mp\", the default, or\n\"ray\".\n\nmp is what vLLM documents today — every node runs serve with its rank,\nthe workers headless — and needs nothing but vLLM. ray is the older way,\nkept for a vLLM too old for mp's multi-node flags; it needs Ray on every\nnode and has not been checked against a real cluster.", Nested: ""},
{Key: "ray_bin", Type: "string", Doc: "RayBin is used when LaunchMode is \"ray\".", Nested: ""},
}},
{Package: "notify", Name: "Webhook", Title: "Webhook", Doc: "", Fields: []Field{
{Key: "url", Type: "string", Doc: "", Nested: ""},
{Key: "format", Type: "string", Doc: "Format is json (the event, with X-Llmgw-Signature when SecretFile is\nset) or slack ({\"text\": …}, which Slack, Mattermost and Rocket.Chat\nincoming webhooks take).", Nested: ""},
{Key: "events", Type: "[]string", Doc: "Events are the kinds sent here; empty is all.", Nested: ""},
{Key: "secret_file", Type: "string", Doc: "SecretFile holds the key the body is signed with: X-Llmgw-Signature\nis sha256=<hex HMAC of the body>.", Nested: ""},
}},
{Package: "accounting", Name: "Price", Title: "accounting.Price", Doc: "", Fields: []Field{
{Key: "prompt_per_1k", Type: "int64", Doc: "", Nested: ""},
{Key: "completion_per_1k", Type: "int64", Doc: "", Nested: ""},
@@ -381,6 +388,10 @@ var Reference = []Block{
{Key: "default", Type: "[]string", Doc: "", Nested: ""},
{Key: "policies", Type: "map[string]Policy", Doc: "", Nested: ""},
}},
{Package: "notify", Name: "Config", Title: "notify.Config", Doc: "", Fields: []Field{
{Key: "webhooks", Type: "[]Webhook", Doc: "", Nested: ""},
{Key: "host", Type: "string", Doc: "Host names the gateway in messages; empty is this machine's name.", Nested: ""},
}},
{Package: "supervisor", Name: "Config", Title: "supervisor.Config", Doc: "", Fields: []Field{
{Key: "work_dir", Type: "string", Doc: "WorkDir holds pid files and per-shard scratch space.", Nested: ""},
{Key: "log_lines", Type: "int", Doc: "LogLines is how many lines of engine output to retain per shard.", Nested: ""},
+1
View File
@@ -29,6 +29,7 @@ var Sources = []string{
"internal/guard/policy.go",
"internal/guard/judge.go",
"internal/config/oidc.go",
"internal/notify/notify.go",
}
// notReference are files with koanf tags that an operator never writes.
+288
View File
@@ -0,0 +1,288 @@
// Package notify tells the people running the gateway about what needs them,
// without a Prometheus in between: a model parked after failing every start,
// a node lost, a budget spent, a model's repository with new weights. Each
// event goes to the webhooks configured for it — JSON signed with a shared
// secret, or a Slack-compatible message.
//
// Sending never holds up what raised the event: events are queued, and a
// webhook that does not answer is retried a few times and then given up on.
// The same event about the same thing is sent once an hour at most.
package notify
import (
"bytes"
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"net/url"
"os"
"slices"
"strings"
"sync"
"time"
)
// Event kinds.
const (
ModelParked = "model_parked"
NodeLost = "node_lost"
BudgetSpent = "budget_spent"
UpdateAvailable = "update_available"
)
// Kinds are every event there is.
var Kinds = []string{ModelParked, NodeLost, BudgetSpent, UpdateAvailable}
// Event is one thing that happened.
type Event struct {
Kind string `json:"kind"`
At time.Time `json:"at"`
Subject string `json:"subject"`
Message string `json:"message"`
Detail map[string]any `json:"detail,omitempty"`
}
// Webhook is where events go.
type Webhook struct {
URL string `koanf:"url"`
// Format is json (the event, with X-Llmgw-Signature when SecretFile is
// set) or slack ({"text": …}, which Slack, Mattermost and Rocket.Chat
// incoming webhooks take).
Format string `koanf:"format"`
// Events are the kinds sent here; empty is all.
Events []string `koanf:"events"`
// SecretFile holds the key the body is signed with: X-Llmgw-Signature
// is sha256=<hex HMAC of the body>.
SecretFile string `koanf:"secret_file"`
}
// Config is server.yaml's notify: section.
type Config struct {
Webhooks []Webhook `koanf:"webhooks"`
// Host names the gateway in messages; empty is this machine's name.
Host string `koanf:"host"`
}
// Validate checks the settings.
func (c Config) Validate() error {
for i, w := range c.Webhooks {
if !strings.HasPrefix(w.URL, "https://") && !strings.HasPrefix(w.URL, "http://") {
return fmt.Errorf("notify.webhooks[%d].url %q: an http(s) address", i, w.URL)
}
switch w.Format {
case "", "json", "slack":
default:
return fmt.Errorf("notify.webhooks[%d].format %q: json or slack", i, w.Format)
}
for _, e := range w.Events {
if !slices.Contains(Kinds, e) {
return fmt.Errorf("notify.webhooks[%d].events: %q is not one of %s", i, e, strings.Join(Kinds, ", "))
}
}
}
return nil
}
// Notifier sends events. A nil one sends nothing.
type Notifier struct {
hooks []hook
host string
log *slog.Logger
client *http.Client
queue chan Event
retry time.Duration
mu sync.Mutex
sent map[string]time.Time
wg sync.WaitGroup
}
type hook struct {
Webhook
secret []byte
}
// dedupe is how long the same event about the same thing is not sent again.
const dedupe = time.Hour
// New builds a notifier and starts sending; it stops when ctx ends. Nil when
// nothing is configured.
func New(ctx context.Context, c Config, log *slog.Logger) (*Notifier, error) {
if len(c.Webhooks) == 0 {
return nil, nil
}
if err := c.Validate(); err != nil {
return nil, err
}
if log == nil {
log = slog.Default()
}
n := &Notifier{host: c.Host, log: log, client: &http.Client{Timeout: 10 * time.Second},
queue: make(chan Event, 256), sent: map[string]time.Time{}, retry: 5 * time.Second}
if n.host == "" {
n.host, _ = os.Hostname()
}
for _, w := range c.Webhooks {
h := hook{Webhook: w}
if w.SecretFile != "" {
b, err := os.ReadFile(w.SecretFile)
if err != nil {
return nil, fmt.Errorf("notify: %w", err)
}
h.secret = bytes.TrimSpace(b)
}
n.hooks = append(n.hooks, h)
}
n.wg.Add(1)
go n.run(ctx)
return n, nil
}
// Notify queues an event. It never blocks: with the queue full the event is
// dropped, and said so in the log.
func (n *Notifier) Notify(kind, subject, message string, detail map[string]any) {
if n == nil {
return
}
k := kind + "\x00" + subject
now := time.Now()
n.mu.Lock()
if last, ok := n.sent[k]; ok && now.Sub(last) < dedupe {
n.mu.Unlock()
return
}
n.sent[k] = now
for key, at := range n.sent {
if now.Sub(at) > dedupe {
delete(n.sent, key)
}
}
n.mu.Unlock()
select {
case n.queue <- Event{Kind: kind, At: now.UTC(), Subject: subject, Message: message, Detail: detail}:
default:
n.log.Warn("a notification was dropped: too many waiting", "kind", kind, "subject", subject)
}
}
// Wait waits for what is queued to be sent, for tests and shutdown.
func (n *Notifier) Wait() {
if n == nil {
return
}
close(n.queue)
n.wg.Wait()
}
func (n *Notifier) run(ctx context.Context) {
defer n.wg.Done()
for {
select {
case <-ctx.Done():
return
case e, ok := <-n.queue:
if !ok {
return
}
for _, h := range n.hooks {
if len(h.Events) > 0 && !slices.Contains(h.Events, e.Kind) {
continue
}
if err := n.send(ctx, h, e, 3); err != nil {
n.log.Warn("a notification could not be delivered", "kind", e.Kind, "subject", e.Subject, "err", err)
}
}
}
}
}
// Test sends a test event to every webhook now, once each, and says how each
// answered: what `llmgw ctl notify test` shows.
func (n *Notifier) Test(ctx context.Context) []Result {
if n == nil {
return nil
}
e := Event{Kind: "test", At: time.Now().UTC(), Subject: "test",
Message: "a test notification: this webhook is reached"}
var out []Result
for _, h := range n.hooks {
r := Result{URL: redact(h.URL)}
if err := n.send(ctx, h, e, 1); err != nil {
r.Err = err.Error()
}
out = append(out, r)
}
return out
}
// Result is how one webhook answered a test.
type Result struct {
URL string `json:"url"`
Err string `json:"error,omitempty"`
}
// redact keeps a webhook's host and drops its path, where Slack and others
// put the secret.
func redact(u string) string {
if i := strings.Index(u, "://"); i >= 0 {
if j := strings.IndexByte(u[i+3:], '/'); j >= 0 {
return u[:i+3+j] + "/…"
}
}
return u
}
func (n *Notifier) send(ctx context.Context, h hook, e Event, attempts int) error {
var body []byte
if h.Format == "slack" {
body, _ = json.Marshal(map[string]string{"text": fmt.Sprintf("*llmgw on %s* — %s", n.host, e.Message)})
} else {
ev := e
if ev.Detail == nil {
ev.Detail = map[string]any{}
}
ev.Detail["gateway"] = n.host
body, _ = json.Marshal(ev)
}
for attempt := 1; ; attempt++ {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, h.URL, bytes.NewReader(body))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Llmgw-Event", e.Kind)
if h.secret != nil {
m := hmac.New(sha256.New, h.secret)
m.Write(body)
req.Header.Set("X-Llmgw-Signature", "sha256="+hex.EncodeToString(m.Sum(nil)))
}
resp, err := n.client.Do(req)
// Not the error as it comes: it names the whole address, and the
// path of a Slack or PagerDuty webhook is its secret.
var ue *url.Error
if errors.As(err, &ue) {
err = fmt.Errorf("%s %s: %w", ue.Op, redact(h.URL), ue.Err)
}
if err == nil {
resp.Body.Close()
if resp.StatusCode < 300 {
return nil
}
err = fmt.Errorf("answered %s", resp.Status)
}
if attempt >= attempts {
return err
}
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(n.retry * time.Duration(attempt)):
}
}
}
+169
View File
@@ -0,0 +1,169 @@
package notify
import (
"context"
"crypto/hmac"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"sync"
"testing"
"time"
)
type got struct {
mu sync.Mutex
calls []*http.Request
body [][]byte
}
func (g *got) handler(fail int) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
b, _ := io.ReadAll(r.Body)
g.mu.Lock()
defer g.mu.Unlock()
g.calls = append(g.calls, r)
g.body = append(g.body, b)
if len(g.calls) <= fail {
w.WriteHeader(http.StatusBadGateway)
}
}
}
func newNotifier(t *testing.T, c Config) *Notifier {
t.Helper()
n, err := New(context.Background(), c, nil)
if err != nil {
t.Fatal(err)
}
n.retry = time.Millisecond
return n
}
func TestSignedJSON(t *testing.T) {
var g got
srv := httptest.NewServer(g.handler(0))
defer srv.Close()
secret := filepath.Join(t.TempDir(), "s")
os.WriteFile(secret, []byte("k3y\n"), 0o600)
n := newNotifier(t, Config{Host: "gw1", Webhooks: []Webhook{{URL: srv.URL, SecretFile: secret}}})
n.Notify(ModelParked, "qwen", "model qwen is parked", map[string]any{"failures": 3})
n.Wait()
if len(g.calls) != 1 {
t.Fatalf("%d calls, want 1", len(g.calls))
}
m := hmac.New(sha256.New, []byte("k3y"))
m.Write(g.body[0])
if want := "sha256=" + hex.EncodeToString(m.Sum(nil)); g.calls[0].Header.Get("X-Llmgw-Signature") != want {
t.Errorf("signature %q, want %q", g.calls[0].Header.Get("X-Llmgw-Signature"), want)
}
if g.calls[0].Header.Get("X-Llmgw-Event") != ModelParked {
t.Errorf("event header %q", g.calls[0].Header.Get("X-Llmgw-Event"))
}
var e Event
if err := json.Unmarshal(g.body[0], &e); err != nil {
t.Fatal(err)
}
if e.Kind != ModelParked || e.Subject != "qwen" || e.Detail["gateway"] != "gw1" || e.Detail["failures"] != float64(3) {
t.Errorf("event %+v", e)
}
}
func TestSlackFiltersAndDedupes(t *testing.T) {
var g got
srv := httptest.NewServer(g.handler(0))
defer srv.Close()
n := newNotifier(t, Config{Host: "gw1", Webhooks: []Webhook{{URL: srv.URL, Format: "slack", Events: []string{NodeLost}}}})
n.Notify(ModelParked, "qwen", "not for this hook", nil)
n.Notify(NodeLost, "n1", "node n1 was lost", nil)
n.Notify(NodeLost, "n1", "node n1 was lost again", nil) // within the hour: not sent
n.Notify(NodeLost, "n2", "node n2 was lost", nil)
n.Wait()
if len(g.calls) != 2 {
t.Fatalf("%d calls, want 2: %q", len(g.calls), g.body)
}
var m map[string]string
json.Unmarshal(g.body[0], &m)
if !strings.Contains(m["text"], "gw1") || !strings.Contains(m["text"], "node n1 was lost") {
t.Errorf("text %q", m["text"])
}
if g.calls[0].Header.Get("X-Llmgw-Signature") != "" {
t.Error("signed without a secret")
}
}
func TestRetries(t *testing.T) {
var g got
srv := httptest.NewServer(g.handler(2))
defer srv.Close()
n := newNotifier(t, Config{Webhooks: []Webhook{{URL: srv.URL}}})
n.Notify(BudgetSpent, "org:1", "spent", nil)
n.Wait()
if len(g.calls) != 3 {
t.Fatalf("%d attempts, want 3 (two failures, then delivered)", len(g.calls))
}
var h got
srv2 := httptest.NewServer(h.handler(10))
defer srv2.Close()
n = newNotifier(t, Config{Webhooks: []Webhook{{URL: srv2.URL}}})
n.Notify(BudgetSpent, "org:1", "spent", nil)
n.Wait()
if len(h.calls) != 3 {
t.Fatalf("%d attempts at a failing hook, want 3 then given up", len(h.calls))
}
}
func TestNilAndValidate(t *testing.T) {
var n *Notifier
n.Notify(NodeLost, "x", "y", nil) // must not panic
n.Wait()
if n, err := New(context.Background(), Config{}, nil); n != nil || err != nil {
t.Fatalf("nothing configured: %v %v", n, err)
}
for _, c := range []Config{
{Webhooks: []Webhook{{URL: "ftp://x"}}},
{Webhooks: []Webhook{{URL: "https://x", Format: "teams"}}},
{Webhooks: []Webhook{{URL: "https://x", Events: []string{"model_gone"}}}},
} {
if c.Validate() == nil {
t.Errorf("%+v accepted", c)
}
}
if _, err := New(context.Background(), Config{Webhooks: []Webhook{{URL: "https://x", SecretFile: "/nonexistent"}}}, nil); err == nil {
t.Error("a missing secret file accepted")
}
}
func TestTestSaysHowEachAnsweredWithoutTheSecretPath(t *testing.T) {
var g got
ok := httptest.NewServer(g.handler(0))
defer ok.Close()
bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusForbidden) }))
defer bad.Close()
n := newNotifier(t, Config{Webhooks: []Webhook{
{URL: ok.URL + "/services/T0/B0/s3cret"},
{URL: bad.URL + "/hook"},
{URL: "http://127.0.0.1:1/services/T0/B0/s3cret"},
}})
defer n.Wait()
res := n.Test(context.Background())
if len(res) != 3 || res[0].Err != "" || !strings.Contains(res[1].Err, "403") || res[2].Err == "" {
t.Fatalf("%+v", res)
}
for _, r := range res {
if strings.Contains(r.URL+r.Err, "s3cret") {
t.Errorf("the secret path shown: %+v", r)
}
}
}
+41
View File
@@ -0,0 +1,41 @@
// Package notifytest catches notifications in tests.
package notifytest
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"sync"
"testing"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
)
// Catch returns a notifier whose webhook records what it is sent, and a func
// that waits for the sending and returns the events. The notifier takes no
// more events once that func is called.
func Catch(t *testing.T) (*notify.Notifier, func() []notify.Event) {
t.Helper()
var mu sync.Mutex
var got []notify.Event
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
b, _ := io.ReadAll(req.Body)
var e notify.Event
_ = json.Unmarshal(b, &e)
mu.Lock()
got = append(got, e)
mu.Unlock()
}))
t.Cleanup(srv.Close)
n, err := notify.New(t.Context(), notify.Config{Webhooks: []notify.Webhook{{URL: srv.URL}}}, nil)
if err != nil {
t.Fatal(err)
}
return n, func() []notify.Event {
n.Wait()
mu.Lock()
defer mu.Unlock()
return got
}
}
+10
View File
@@ -3,6 +3,8 @@ package router
import (
"context"
"time"
"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
)
// Activator loads a model that is configured but not running.
@@ -90,3 +92,11 @@ func (r *Router) activate(ctx context.Context, model string) bool {
// own timeout is not what ends the request — being told to retry is a better
// outcome than a socket closing with nothing on it.
const defaultActivationWait = 45 * time.Second
// SetNotifier tells the accountant where to say a budget is spent. Nothing
// without a store, where there are no budgets.
func (r *Router) SetNotifier(n *notify.Notifier) {
if r.acct != nil {
r.acct.SetNotifier(n)
}
}