From 1b71c38f891f8790b87ae4ab2c81e8ef005ae1cb Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 4 Oct 2026 16:58:11 +0200 Subject: [PATCH] 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) --- BACKLOG.md | 8 +- CHANGELOG.md | 15 + deploy/monitoring/alertmanager.yml | 52 +++- docs/install.md | 35 +++ docs/reference.md | 56 ++++ internal/accounting/accounting.go | 11 + internal/accounting/accounting_test.go | 27 ++ internal/cli/ctl.go | 1 + internal/cli/notify.go | 48 ++++ internal/cli/server.go | 15 +- internal/config/config.go | 9 + internal/control/adminapi/api.go | 4 + internal/control/adminapi/notify.go | 19 ++ internal/control/modelserve/modelserve.go | 4 + internal/control/modelserve/updates.go | 9 + internal/control/modelserve/updates_test.go | 14 + internal/control/reconciler/notify_test.go | 46 ++++ internal/control/reconciler/reconciler.go | 18 ++ internal/docgen/reference_gen.go | 11 + internal/docgen/render.go | 1 + internal/notify/notify.go | 288 ++++++++++++++++++++ internal/notify/notify_test.go | 169 ++++++++++++ internal/notify/notifytest/notifytest.go | 41 +++ internal/router/activate.go | 10 + 24 files changed, 907 insertions(+), 4 deletions(-) create mode 100644 internal/cli/notify.go create mode 100644 internal/control/adminapi/notify.go create mode 100644 internal/control/reconciler/notify_test.go create mode 100644 internal/notify/notify.go create mode 100644 internal/notify/notify_test.go create mode 100644 internal/notify/notifytest/notifytest.go diff --git a/BACKLOG.md b/BACKLOG.md index efeb600b..c61dfe03 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 317a5f3a..94aab8e1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/deploy/monitoring/alertmanager.yml b/deploy/monitoring/alertmanager.yml index c4ecb025..fbc03604 100644 --- a/deploy/monitoring/alertmanager.yml +++ b/deploy/monitoring/alertmanager.yml @@ -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. diff --git a/docs/install.md b/docs/install.md index 1aeed9bd..c91daa02 100644 --- a/docs/install.md +++ b/docs/install.md @@ -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=`, 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 diff --git a/docs/reference.md b/docs/reference.md index a40a4b76..df96ec45 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -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=. + ## 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` diff --git a/internal/accounting/accounting.go b/internal/accounting/accounting.go index 7230d938..e9b2c4da 100644 --- a/internal/accounting/accounting.go +++ b/internal/accounting/accounting.go @@ -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} } } diff --git a/internal/accounting/accounting_test.go b/internal/accounting/accounting_test.go index ef6a0b67..4372b169 100644 --- a/internal/accounting/accounting_test.go +++ b/internal/accounting/accounting_test.go @@ -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) + } +} diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index 4986a3fd..def0aed2 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -162,6 +162,7 @@ func newCtlCmd() *cobra.Command { newQuotaCmd(client, &asJSON), newGuardCmd(client, &asJSON), newAuditCmd(client, &asJSON), + newNotifyCmd(client, &asJSON), newUsageCmd(client, &asJSON), newCertCmd(), ) diff --git a/internal/cli/notify.go b/internal/cli/notify.go new file mode 100644 index 00000000..30b4d63f --- /dev/null +++ b/internal/cli/notify.go @@ -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 +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 3137716f..db7a35ac 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -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) diff --git a/internal/config/config.go b/internal/config/config.go index 105a6ef6..5d7050cf 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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 } diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 34e08025..12a217e4 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -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) { diff --git a/internal/control/adminapi/notify.go b/internal/control/adminapi/notify.go new file mode 100644 index 00000000..0b4f163a --- /dev/null +++ b/internal/control/adminapi/notify.go @@ -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())) +} diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index b00c9f91..294d14ec 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -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. diff --git a/internal/control/modelserve/updates.go b/internal/control/modelserve/updates.go index f82ef83f..d3a6a28c 100644 --- a/internal/control/modelserve/updates.go +++ b/internal/control/modelserve/updates.go @@ -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 } diff --git a/internal/control/modelserve/updates_test.go b/internal/control/modelserve/updates_test.go index 4c586d8c..ee3b22ff 100644 --- a/internal/control/modelserve/updates_test.go +++ b/internal/control/modelserve/updates_test.go @@ -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) + } +} diff --git a/internal/control/reconciler/notify_test.go b/internal/control/reconciler/notify_test.go new file mode 100644 index 00000000..4926e6bf --- /dev/null +++ b/internal/control/reconciler/notify_test.go @@ -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) + } +} diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index d392d70c..2c822ec6 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -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) +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index ebee9e37..839dea7e 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -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=.", 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: ""}, diff --git a/internal/docgen/render.go b/internal/docgen/render.go index ef9a1c75..2f013589 100644 --- a/internal/docgen/render.go +++ b/internal/docgen/render.go @@ -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. diff --git a/internal/notify/notify.go b/internal/notify/notify.go new file mode 100644 index 00000000..73ccf37e --- /dev/null +++ b/internal/notify/notify.go @@ -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=. + 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)): + } + } +} diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go new file mode 100644 index 00000000..d87d2c6c --- /dev/null +++ b/internal/notify/notify_test.go @@ -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) + } + } +} diff --git a/internal/notify/notifytest/notifytest.go b/internal/notify/notifytest/notifytest.go new file mode 100644 index 00000000..137b1429 --- /dev/null +++ b/internal/notify/notifytest/notifytest.go @@ -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 + } +} diff --git a/internal/router/activate.go b/internal/router/activate.go index be433176..a738e3ca 100644 --- a/internal/router/activate.go +++ b/internal/router/activate.go @@ -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) + } +}