From 8547528fb7d6c9b1abbdde6a8fc09506e2bfdc17 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 15 Nov 2026 06:54:55 +0100 Subject: [PATCH] Prices from what the cluster costs: ctl prices suggest auth.gpu_hour_cost (and core_hour_cost) is what an hour of a GPU (a core) costs the operator. The leader keeps, by the hour, the GPU- and core-hours each model's replicas hold (held_1h); llmgw ctl prices suggest spreads a model's over the tokens it served, an output token counted as four input ones and a cached prompt token as a tenth of one, and --apply writes the prices into auth.pricing. Idle hours are in the cost, so a model held and little used comes out dear; one held and billed nothing is shown with what it held. Co-Authored-By: Claude Opus 5.5 --- BACKLOG.md | 3 +- CHANGELOG.md | 5 + docs/admin-api.openapi.json | 127 ++++++++++++++++++ docs/install.md | 13 ++ docs/reference.md | 17 +++ internal/accounting/suggest.go | 91 +++++++++++++ internal/accounting/suggest_test.go | 45 +++++++ internal/cli/ctl.go | 1 + internal/cli/ctl_prices.go | 121 +++++++++++++++++ internal/cli/server.go | 29 ++++ internal/config/config.go | 9 ++ internal/control/adminapi/api.go | 5 + internal/control/adminapi/prices.go | 108 +++++++++++++++ internal/control/adminapi/routes.go | 7 + internal/control/held/held.go | 93 +++++++++++++ internal/control/held/held_test.go | 40 ++++++ internal/docgen/reference_gen.go | 2 + internal/store/copy.go | 1 + internal/store/held.go | 62 +++++++++ internal/store/held_test.go | 35 +++++ .../store/migrations/postgres/0053_held.sql | 11 ++ .../store/migrations/sqlite/0053_held.sql | 10 ++ 22 files changed, 834 insertions(+), 1 deletion(-) create mode 100644 internal/accounting/suggest.go create mode 100644 internal/accounting/suggest_test.go create mode 100644 internal/cli/ctl_prices.go create mode 100644 internal/control/adminapi/prices.go create mode 100644 internal/control/held/held.go create mode 100644 internal/control/held/held_test.go create mode 100644 internal/store/held.go create mode 100644 internal/store/held_test.go create mode 100644 internal/store/migrations/postgres/0053_held.sql create mode 100644 internal/store/migrations/sqlite/0053_held.sql diff --git a/BACKLOG.md b/BACKLOG.md index ab8eff1..c7502b1 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -2112,7 +2112,8 @@ lent, partition`); the leader tells each named node's agent (root) what to be, t of it, kept by the leader from the store; `sbatch --account=PROJECT`. The records back: the leader reads `sacct --json` for jobs ended since the last read (through the controller's agent) into a `batch_usage` table (job, user, project, start, end, GPU-seconds by class, CPU-seconds, memory), priced - per GPU-hour and core-hour into cost_micros, summed with the requests' usage on the usage pages and in + per GPU-hour and core-hour (auth.gpu_hour_cost, core_hour_cost: the same basis `ctl prices suggest` + prices tokens from, 2026-11-15) into cost_micros, summed with the requests' usage on the usage pages and in budgets. Budgets enforced: a project's remaining budget as GPU-minutes in its account's GrpTRESMins, so Slurm holds its new jobs once spent. Needs the same users on every node (open: LDAP/sssd or llmgw-made). Also found: slurm.conf diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b73139..1971ba7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,11 @@ APIs change between minors without a deprecation period. there, their session put in the job (pam_slurm_adopt), so nobody uses a node's cards outside Slurm, nor on a node llmgw serves from. Root and the machines' own users are not held to it. +- **Prices from what the cluster costs.** `auth.gpu_hour_cost` (and + `core_hour_cost`) is what an hour of a GPU (a core) costs; the leader keeps + the hours each model's replicas hold, and `llmgw ctl prices suggest` prices + each model from them over the tokens it served, an output token four input + ones and a cached one a tenth. `--apply` writes the prices. - **Slurm's jobs in containers.** `llmgw ctl slurm containers on` (`slurm.containers`) installs enroot and pyxis on every Slurm machine, at pinned versions checked against their digests, and nvidia-container-cli on diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 0226ea5..5aee1aa 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -95,6 +95,9 @@ { "name": "presets" }, + { + "name": "prices" + }, { "name": "projects" }, @@ -4396,6 +4399,34 @@ } } }, + "/api/v1/prices/suggest": { + "get": { + "operationId": "getPricesSuggest", + "summary": "Each model's price, from what it held of the cluster", + "description": "days (30 if not given) is the window. A model's cost is the GPU- and core-hours its replicas held, idle ones too, at auth.gpu_hour_cost and auth.core_hour_cost; spread over its tokens, an output token four input ones and a cached prompt token a tenth of one, it is price, in micro-units a thousand tokens as auth.pricing takes them. current is the price in force. 409 with no cost set.", + "tags": [ + "prices" + ], + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/PriceSuggestions" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/projects": { "post": { "operationId": "postProjects", @@ -8846,6 +8877,73 @@ } } }, + "PriceSuggestion": { + "type": "object", + "description": "PriceSuggestion is one model's.", + "properties": { + "model": { + "type": "string" + }, + "gpu_hours": { + "type": "number" + }, + "core_hours": { + "type": "number" + }, + "cost": { + "type": "number", + "description": "Cost is what it held cost, in budget units." + }, + "prompt_tokens": { + "type": "integer", + "format": "int64" + }, + "completion_tokens": { + "type": "integer", + "format": "int64" + }, + "price": { + "$ref": "#/components/schemas/accounting.Price", + "description": "Price is the suggestion; zero with a Note saying why there is none." + }, + "note": { + "type": "string" + }, + "current": { + "$ref": "#/components/schemas/accounting.Price" + } + } + }, + "PriceSuggestions": { + "type": "object", + "description": "PriceSuggestions is a model's price from its cost, for each model.", + "properties": { + "days": { + "type": "integer", + "description": "Days is the window; RecordedFrom the first hour held is known for, later than the window's start on a cluster that began keeping it since, which makes the window shorter than asked." + }, + "since": { + "type": "string", + "format": "date-time" + }, + "recorded_from": { + "type": "string", + "format": "date-time" + }, + "gpu_hour_cost": { + "type": "number" + }, + "core_hour_cost": { + "type": "number" + }, + "models": { + "type": "array", + "items": { + "$ref": "#/components/schemas/PriceSuggestion" + } + } + } + }, "ProfileReplicasRequest": { "type": "object", "description": "ProfileReplicasRequest counts a model's replicas per profile: profile name to replicas, and CPU, for a llama.cpp model with no profile on CPUs, the one to add.", @@ -10007,6 +10105,35 @@ } } }, + "accounting.Price": { + "type": "object", + "description": "Price is what one model costs, in micro-units per thousand tokens.\n\nMicro-units rather than a float because spend is accumulated by repeated addition, and repeated addition of floats drifts. What a unit means is the operator's: self-hosted models have no per-token price, so it may be GPU-seconds, a notional rate, or 1000 everywhere — which turns a budget into a token quota.", + "properties": { + "PromptPer1k": { + "type": "integer", + "format": "int64" + }, + "CompletionPer1k": { + "type": "integer", + "format": "int64" + }, + "CachedPer1k": { + "type": "integer", + "format": "int64", + "description": "CachedPer1k is what a prompt token costs that the engine or provider read from its cache instead of computing: the part of prompt_tokens the answer counts as cached_tokens. Zero charges those at PromptPer1k, as before there was a rate for them." + }, + "PerImage": { + "type": "integer", + "format": "int64", + "description": "PerImage is what one picture costs, for a model that makes them: a picture's cost is the card's time, which its prompt's few tokens do not measure." + }, + "PerMesh": { + "type": "integer", + "format": "int64", + "description": "PerMesh is what one 3D model costs: minutes of a card, which no token count measures." + } + } + }, "accounting.Rates": { "type": "object", "description": "Rates are an accountant's two limiters' counts.", diff --git a/docs/install.md b/docs/install.md index 0a0f636..58cc941 100644 --- a/docs/install.md +++ b/docs/install.md @@ -457,6 +457,19 @@ by itself. Three things are worth knowing before you edit it: cache (the answer's `cached_tokens`) are charged at it, the rest at the prompt rate. A provider that bills cached reads at a tenth is `{prompt_per_1k: 2500, cached_per_1k: 250, completion_per_1k: 10000}`. +- What to price a self-hosted model at follows from what the cluster costs. + Set `auth.gpu_hour_cost`, what an hour of one GPU costs in the units budgets + are in: the hardware written off over its years, its power and room, and the + people who run it, over the GPUs and the 730 hours of a month (an 8-GPU + server at 250,000 over four years, 10 kW, and some upkeep is about 1.35); + `auth.core_hour_cost` likewise for models on CPUs. The leader keeps, by the + hour, the GPU- and core-hours each model's replicas hold, and `llmgw ctl + prices suggest` spreads them over the tokens the model served: an output + token counted as four input ones (making one is that much dearer), a cached + prompt token as a tenth of one. Idle hours are in the cost, so a model held + all month and seldom used comes out dear, which is the point; one held and + billed nothing (a guard's judge) is shown with what it held. `--apply` + writes the prices into `auth.pricing`. - `control.enabled: true` is what makes this a control plane while its catalog is empty. Without it, a file with no models is refused as having nothing to serve. diff --git a/docs/reference.md b/docs/reference.md index 2b8a320..e76666a 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -43,6 +43,8 @@ rather than repeated. | `personal` | `Personal` | [Auth](#auth) | | `pricing` | `map[string]Price` | [Auth](#auth) | | `default_price` | `Price` | [Auth](#auth) | +| `gpu_hour_cost` | `float64` | [Auth](#auth) | +| `core_hour_cost` | `float64` | [Auth](#auth) | | `flush_interval` | `time.Duration` | [Auth](#auth) | | `batch_size` | `int` | [Auth](#auth) | | `buffer_size` | `int` | [Auth](#auth) | @@ -722,6 +724,21 @@ table costs whatever DefaultPrice says, which is nothing unless set. Takes a [config.Price](#configprice) block. +### `gpu_hour_cost` + +`float64` + +GPUHourCost is what an hour of one GPU costs the operator, in the +units budgets are in: the hardware written off over its years, its +power and room, and the people who run it, over the GPUs and the +hours of a month. With CoreHourCost for a CPU core, it is what +`llmgw ctl prices suggest` prices each model from: the hours its +replicas held, idle ones too, over the tokens it served. + +### `core_hour_cost` + +`float64` + ### `flush_interval` `time.Duration` diff --git a/internal/accounting/suggest.go b/internal/accounting/suggest.go new file mode 100644 index 0000000..cbe855c --- /dev/null +++ b/internal/accounting/suggest.go @@ -0,0 +1,91 @@ +package accounting + +import ( + "math" + "slices" +) + +// Prices from what the cluster costs: what each model held of it (its +// replicas' GPU- and core-hours) at what an hour of a GPU or a core costs +// the operator, spread over the tokens it served. Idle time is in it: a +// model held all month and used an hour costs what it held, not the hour. +// +// An output token is priced OutputPerInput input tokens, as producing one +// is that much dearer (one step a token, against a prompt read in one +// pass); a cached prompt token a tenth of an input one. +const ( + OutputPerInput = 4 + CachedPerInput = 0.1 +) + +// Basis is what an hour costs, in the units budgets are in. +type Basis struct { + GPUHour float64 + CoreHour float64 +} + +// Held is what a model held, and Served what it answered with. +type Held struct { + GPUSeconds, CoreSeconds int64 +} + +// Served is a model's tokens. +type Served struct { + Prompt, Completion int64 +} + +// Suggestion is one model's price from its cost. +type Suggestion struct { + Model string `json:"model"` + GPUHours float64 `json:"gpu_hours"` + CoreHours float64 `json:"core_hours"` + // Cost is what it held cost, in budget units. + Cost float64 `json:"cost"` + Prompt int64 `json:"prompt_tokens"` + Completion int64 `json:"completion_tokens"` + // Price is the suggestion; zero with a Note saying why there is none. + Price Price `json:"price"` + Note string `json:"note,omitempty"` +} + +// Suggest prices every model that held something or served something. +func Suggest(held map[string]Held, served map[string]Served, b Basis) []Suggestion { + var names []string + for m := range held { + names = append(names, m) + } + for m := range served { + if _, ok := held[m]; !ok { + names = append(names, m) + } + } + slices.Sort(names) + out := make([]Suggestion, 0, len(names)) + for _, m := range names { + h, s := held[m], served[m] + g := Suggestion{Model: m, GPUHours: float64(h.GPUSeconds) / 3600, CoreHours: float64(h.CoreSeconds) / 3600, + Prompt: s.Prompt, Completion: s.Completion} + g.Cost = g.GPUHours*b.GPUHour + g.CoreHours*b.CoreHour + // Tokens counted as output ones: an input token is a fraction of one. + units := float64(s.Completion) + float64(s.Prompt)/OutputPerInput + switch { + case g.Cost == 0 && units > 0: + g.Note = "held nothing recorded here: an outside provider's, or served before the hours were kept" + case g.Cost > 0 && units == 0: + g.Note = "held the cluster and was billed no tokens (a guard's judge, or a model priced by the picture or 3D model): no one pays what it held" + case g.Cost > 0: + // Micro-units per thousand output tokens, rounded up so a + // cheap model is never free. + out1k := int64(math.Ceil(g.Cost / units * 1000 * 1e6)) + g.Price = Price{ + CompletionPer1k: out1k, + PromptPer1k: ceilDiv(out1k, OutputPerInput), + CachedPer1k: max(int64(math.Ceil(float64(ceilDiv(out1k, OutputPerInput))*CachedPerInput)), 1), + } + } + out = append(out, g) + } + return out +} + +func ceilDiv(a, b int64) int64 { return (a + b - 1) / b } diff --git a/internal/accounting/suggest_test.go b/internal/accounting/suggest_test.go new file mode 100644 index 0000000..be73d0b --- /dev/null +++ b/internal/accounting/suggest_test.go @@ -0,0 +1,45 @@ +package accounting + +import ( + "strings" + "testing" +) + +func TestSuggestSpreadsWhatAModelHeldOverItsTokens(t *testing.T) { + // The worked example: two cards for 1000 hours at 1.35 an hour is 2700; + // 3.0 billion output tokens and 4.0 billion input ones, counted as 1.0 + // billion output ones, are 4.0 billion: 0.675 a million output tokens. + held := map[string]Held{ + "llama-70b": {GPUSeconds: 2 * 1000 * 3600}, + "idle": {GPUSeconds: 3600}, + "small-cpu": {CoreSeconds: 8 * 100 * 3600}, + } + served := map[string]Served{ + "llama-70b": {Prompt: 4_000_000_000, Completion: 3_000_000_000}, + "small-cpu": {Prompt: 1_000_000, Completion: 1_000_000}, + "outside": {Prompt: 10, Completion: 10}, + } + got := Suggest(held, served, Basis{GPUHour: 1.35, CoreHour: 0.01}) + by := map[string]Suggestion{} + for _, s := range got { + by[s.Model] = s + } + l := by["llama-70b"] + if l.Cost != 2700 || l.Price.CompletionPer1k != 675 || l.Price.PromptPer1k != 169 || l.Price.CachedPer1k != 17 { + t.Errorf("llama: %+v", l) + } + // What it costs back at that price is what it held, within the rounding. + back := float64(l.Price.Cost(l.Prompt, 0, l.Completion, 0, 0)) / 1e6 + if back < l.Cost || back > l.Cost*1.01 { + t.Errorf("charged %.2f for a cost of %.2f", back, l.Cost) + } + if s := by["small-cpu"]; s.Cost != 8 || s.Price.CompletionPer1k == 0 { + t.Errorf("cpu: %+v", s) + } + if s := by["idle"]; !s.Price.Zero() || !strings.Contains(s.Note, "billed no tokens") { + t.Errorf("idle: %+v", s) + } + if s := by["outside"]; !s.Price.Zero() || !strings.Contains(s.Note, "held nothing") { + t.Errorf("outside: %+v", s) + } +} diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index feba2c1..56b3dbe 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -190,6 +190,7 @@ func newCtlCmd() *cobra.Command { newKBCmd(client, &asJSON), newTrainCmd(client, &asJSON), newSlurmCmd(client, &asJSON), + newPricesCmd(client, &asJSON), newConfigCmd(client, &asJSON), newAuditCmd(client, &asJSON), newNotifyCmd(client, &asJSON), diff --git a/internal/cli/ctl_prices.go b/internal/cli/ctl_prices.go new file mode 100644 index 0000000..90262d4 --- /dev/null +++ b/internal/cli/ctl_prices.go @@ -0,0 +1,121 @@ +package cli + +import ( + "encoding/json" + "fmt" + "net/url" + "strconv" + "text/tabwriter" + + "github.com/spf13/cobra" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/settings" +) + +func newPricesCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + cmd := &cobra.Command{ + Use: "prices", + Short: "What each model's tokens cost, from what the cluster costs", + } + var days int + var apply bool + var note string + suggest := &cobra.Command{ + Use: "suggest", + Short: "Each model's price from the GPU- and core-hours it held, over the tokens it served", + Long: "Set auth.gpu_hour_cost first: what an hour of one GPU costs, in the units budgets are in (the\n" + + "hardware written off over its years, power and room, the people who run it, over the GPUs and\n" + + "the hours of a month); auth.core_hour_cost likewise for models on CPUs. Each model's cost is\n" + + "the hours its replicas held, idle ones too, kept by the hour on the leader; spread over its\n" + + "tokens, an output token counted as four input ones and a cached prompt token as a tenth of one,\n" + + "it is its price. A model held and not used shows what that cost. --apply writes the prices\n" + + "into auth.pricing, one settings version, as `ctl config set` does.", + Example: " llmgw ctl config set auth.gpu_hour_cost 1.35\n" + + " llmgw ctl prices suggest --days 30\n" + + " llmgw ctl prices suggest --apply --note \"November's costs\"", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + var s adminapi.PriceSuggestions + if err := client().get(cmd.Context(), "/api/v1/prices/suggest?days="+url.QueryEscape(strconv.Itoa(days)), &s); err != nil { + return err + } + w := cmd.OutOrStdout() + if *asJSON { + return json.NewEncoder(w).Encode(s) + } + fmt.Fprintf(w, "cost: %g a GPU-hour, %g a core-hour; the last %d days", s.GPUHourCost, s.CoreHourCost, s.Days) + if !s.RecordedFrom.IsZero() && s.RecordedFrom.After(s.Since) { + fmt.Fprintf(w, " (held kept only since %s)", s.RecordedFrom.Local().Format("2006-01-02 15:04")) + } + fmt.Fprintln(w) + fmt.Fprintln(w, "prices a million tokens: input / output / cached input, in budget units") + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + fmt.Fprintln(tw, "MODEL\tGPU-H\tCORE-H\tCOST\tTOKENS IN/OUT\tSUGGESTED\tNOW\tNOTE") + var changed []adminapi.PriceSuggestion + for _, m := range s.Models { + sug := "-" + if !m.Price.Zero() { + sug = perMillion(m.Price.PromptPer1k, m.Price.CompletionPer1k, m.Price.CachedPer1k) + if m.Price.PromptPer1k != m.Current.PromptPer1k || m.Price.CompletionPer1k != m.Current.CompletionPer1k || + m.Price.CachedPer1k != m.Current.CachedPer1k { + changed = append(changed, m) + } + } + fmt.Fprintf(tw, "%s\t%.1f\t%.1f\t%.2f\t%s/%s\t%s\t%s\t%s\n", m.Model, m.GPUHours, m.CoreHours, m.Cost, + count(m.Prompt), count(m.Completion), sug, + perMillion(m.Current.PromptPer1k, m.Current.CompletionPer1k, m.Current.CachedPer1k), m.Note) + } + if err := tw.Flush(); err != nil { + return err + } + if !apply { + if len(changed) > 0 { + fmt.Fprintf(w, "%d model(s) priced otherwise now; --apply writes the suggestions\n", len(changed)) + } + return nil + } + if len(changed) == 0 { + fmt.Fprintln(w, "nothing to change") + return nil + } + return changeSettings(cmd, client, note, func(doc *settings.Doc) error { + for _, m := range changed { + for k, v := range map[string]int64{"prompt_per_1k": m.Price.PromptPer1k, + "completion_per_1k": m.Price.CompletionPer1k, "cached_per_1k": m.Price.CachedPer1k} { + if err := doc.SetYAML([]string{"auth", "pricing", m.Model, k}, strconv.FormatInt(v, 10)); err != nil { + return err + } + } + } + return nil + }) + }, + } + suggest.Flags().IntVar(&days, "days", 30, "the window, in days") + suggest.Flags().BoolVar(&apply, "apply", false, "write the suggested prices into auth.pricing") + suggest.Flags().StringVar(¬e, "note", "", "why, kept with the settings' version") + cmd.AddCommand(suggest) + return cmd +} + +// perMillion is micro-units a thousand tokens as budget units a million. +func perMillion(in, out, cached int64) string { + if in == 0 && out == 0 && cached == 0 { + return "-" + } + f := func(v int64) string { return strconv.FormatFloat(float64(v)/1000, 'f', -1, 64) } + return f(in) + " / " + f(out) + " / " + f(cached) +} + +func count(n int64) string { + switch { + case n >= 1e9: + return fmt.Sprintf("%.1fB", float64(n)/1e9) + case n >= 1e6: + return fmt.Sprintf("%.1fM", float64(n)/1e6) + case n >= 1e3: + return fmt.Sprintf("%.1fk", float64(n)/1e3) + } + return strconv.FormatInt(n, 10) +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 99e60fd..5be1fb8 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -36,6 +36,7 @@ import ( "golang.org/x/sync/errgroup" llmgwv1 "gitea.mynodes.xyz/ilker/llmgw/api/gen/llmgw/v1" + "gitea.mynodes.xyz/ilker/llmgw/internal/accounting" "gitea.mynodes.xyz/ilker/llmgw/internal/audit" "gitea.mynodes.xyz/ilker/llmgw/internal/bench/eval" "gitea.mynodes.xyz/ilker/llmgw/internal/bundle" @@ -47,6 +48,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/fabrictest" "gitea.mynodes.xyz/ilker/llmgw/internal/control/federation" "gitea.mynodes.xyz/ilker/llmgw/internal/control/fleetmetrics" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/held" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/control/slurmsvc" @@ -433,6 +435,27 @@ func newServerCmd() *cobra.Command { api.SetAgentRestarter(agentSvc.RestartAgents) } api.SetAudit(st) + // Prices from what the cluster costs: what each model + // holds, kept by the hour on the leader. + api.SetPrices(st, func() adminapi.PriceConfig { + au := current.Load().Auth + cur := make(map[string]accounting.Price, len(au.Pricing)) + for m, p := range au.Pricing { + cur[m] = accountingPrice(p) + } + return adminapi.PriceConfig{Basis: accounting.Basis{GPUHour: au.GPUHourCost, CoreHour: au.CoreHourCost}, + Current: cur, Default: accountingPrice(au.DefaultPrice)} + }) + go (&held.Recorder{Inv: admin.Inventory(), Store: st, Log: log, + Models: func() map[string]string { + m := map[string]string{} + for _, rep := range admin.Reconciler().Replicas() { + m[rep.ID] = rep.Model + } + return m + }, + Leader: func() bool { return admin.Leadership() == nil || admin.Leadership().IsLeader() }, + }).Run(ctx) // Knowledge bases, with the cluster's own models to embed a // probe (a collection's dimension) and an operator's search. api.SetKB(&adminapi.KBAdmin{Store: st, Embed: embedder(rt.Replay)}) @@ -1418,3 +1441,9 @@ func runPresetsPass(ctx context.Context, api *adminapi.API, lead *control.Leader } } } + +// accountingPrice is a configured price as accounting has it. +func accountingPrice(p config.Price) accounting.Price { + return accounting.Price{PromptPer1k: p.PromptPer1k, CompletionPer1k: p.CompletionPer1k, CachedPer1k: p.CachedPer1k, + PerImage: p.PerImage, PerMesh: p.PerMesh} +} diff --git a/internal/config/config.go b/internal/config/config.go index b23d258..9912d20 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -470,6 +470,15 @@ type Auth struct { Pricing map[string]Price `koanf:"pricing"` DefaultPrice Price `koanf:"default_price"` + // GPUHourCost is what an hour of one GPU costs the operator, in the + // units budgets are in: the hardware written off over its years, its + // power and room, and the people who run it, over the GPUs and the + // hours of a month. With CoreHourCost for a CPU core, it is what + // `llmgw ctl prices suggest` prices each model from: the hours its + // replicas held, idle ones too, over the tokens it served. + GPUHourCost float64 `koanf:"gpu_hour_cost"` + CoreHourCost float64 `koanf:"core_hour_cost"` + // FlushInterval and BatchSize decide when usage records are written. FlushInterval time.Duration `koanf:"flush_interval"` BatchSize int `koanf:"batch_size"` diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 3af429c..8ecb420 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -100,6 +100,10 @@ type API struct { ctl *control.Controller // slurmUsers are the Linux users made from the accounts; nil without a store. slurmUsers SlurmUsers + // prices are suggested from what each model held and served; nil + // without a store. + prices PriceStore + priceCfg func() PriceConfig // webhooks counts the settings' notify.webhooks; nil does not say. webhooks func() int } @@ -143,6 +147,7 @@ func (a *API) Register(mux Mux) { a.registerTenancy(mux) a.registerTrain(mux) a.registerSlurm(mux) + a.registerPrices(mux) a.registerGuard(mux) a.registerSettings(mux) a.registerAuth(mux) diff --git a/internal/control/adminapi/prices.go b/internal/control/adminapi/prices.go new file mode 100644 index 0000000..bf0ce72 --- /dev/null +++ b/internal/control/adminapi/prices.go @@ -0,0 +1,108 @@ +package adminapi + +import ( + "context" + "net/http" + "strconv" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/accounting" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// PriceStore is what a price is suggested from: what each model held of +// the cluster, and the tokens it served. +type PriceStore interface { + HeldSince(ctx context.Context, since time.Time) ([]store.Held, time.Time, error) + UsageByModel(ctx context.Context, since time.Time) ([]store.ModelUsage, error) +} + +// PriceConfig is the cost basis and the prices in force. +type PriceConfig struct { + Basis accounting.Basis + Current map[string]accounting.Price + Default accounting.Price +} + +// PriceSuggestions is a model's price from its cost, for each model. +type PriceSuggestions struct { + // Days is the window; RecordedFrom the first hour held is known for, + // later than the window's start on a cluster that began keeping it + // since, which makes the window shorter than asked. + Days int `json:"days"` + Since time.Time `json:"since"` + RecordedFrom time.Time `json:"recorded_from,omitempty"` + GPUHourCost float64 `json:"gpu_hour_cost"` + CoreHourCost float64 `json:"core_hour_cost"` + Models []PriceSuggestion `json:"models"` +} + +// PriceSuggestion is one model's. +type PriceSuggestion struct { + accounting.Suggestion + Current accounting.Price `json:"current"` +} + +// SetPrices gives the API what prices are suggested from. +func (a *API) SetPrices(st PriceStore, cfg func() PriceConfig) { a.prices, a.priceCfg = st, cfg } + +func (a *API) registerPrices(mux Mux) { + mux.HandleFunc("GET /api/v1/prices/suggest", a.suggestPrices) +} + +func (a *API) suggestPrices(w http.ResponseWriter, r *http.Request) { + if a.prices == nil || a.priceCfg == nil { + writeError(w, http.StatusNotImplemented, "prices are suggested from the store, which this control plane runs without") + return + } + days := 30 + if v := r.URL.Query().Get("days"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n < 1 || n > 366 { + writeError(w, http.StatusBadRequest, "days is a whole number of days, 1 to 366") + return + } + days = n + } + cfg := a.priceCfg() + if cfg.Basis.GPUHour <= 0 && cfg.Basis.CoreHour <= 0 { + writeError(w, http.StatusConflict, "no cost to price from: set auth.gpu_hour_cost (and auth.core_hour_cost for "+ + "models on CPUs), what an hour of one costs, in the units budgets are in") + return + } + since := time.Now().Add(-time.Duration(days) * 24 * time.Hour) + rows, first, err := a.prices.HeldSince(r.Context(), since) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + // Tokens from when holding was first recorded, so both sides cover the + // same hours. + from := since + if first.After(from) { + from = first + } + usage, err := a.prices.UsageByModel(r.Context(), from) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + held := map[string]accounting.Held{} + for _, h := range rows { + held[h.Model] = accounting.Held{GPUSeconds: h.GPUSeconds, CoreSeconds: h.CoreSeconds} + } + served := map[string]accounting.Served{} + for _, u := range usage { + served[u.Model] = accounting.Served{Prompt: u.Prompt, Completion: u.Completion} + } + out := PriceSuggestions{Days: days, Since: since.UTC(), RecordedFrom: first, + GPUHourCost: cfg.Basis.GPUHour, CoreHourCost: cfg.Basis.CoreHour, Models: []PriceSuggestion{}} + for _, s := range accounting.Suggest(held, served, cfg.Basis) { + cur, ok := cfg.Current[s.Model] + if !ok { + cur = cfg.Default + } + out.Models = append(out.Models, PriceSuggestion{Suggestion: s, Current: cur}) + } + writeJSON(w, http.StatusOK, out) +} diff --git a/internal/control/adminapi/routes.go b/internal/control/adminapi/routes.go index 2af4ec5..a55cdb2 100644 --- a/internal/control/adminapi/routes.go +++ b/internal/control/adminapi/routes.go @@ -408,6 +408,13 @@ var Routes = []Route{ "all away. The machines have them within a minute.", Request: SlurmKeysRequest{}, Response: slurmsvc.User{}}, + {Method: "GET", Path: "/api/v1/prices/suggest", Summary: "Each model's price, from what it held of the cluster", + Description: "days (30 if not given) is the window. A model's cost is the GPU- and core-hours its replicas held, idle " + + "ones too, at auth.gpu_hour_cost and auth.core_hour_cost; spread over its tokens, an output token four input " + + "ones and a cached prompt token a tenth of one, it is price, in micro-units a thousand tokens as " + + "auth.pricing takes them. current is the price in force. 409 with no cost set.", + Response: PriceSuggestions{}}, + {Method: "POST", Path: "/api/v1/train/jobs", Summary: "Queue a LoRA adapter's training on a node", Description: "The node trains when its agent is there, nothing else trains on it, and a card has gpu_memory_mib free; " + "the pairs are NDJSON of {\"q\", \"a\"}. An adapter learns how answers sound far better than what they say.", diff --git a/internal/control/held/held.go b/internal/control/held/held.go new file mode 100644 index 0000000..6b79835 --- /dev/null +++ b/internal/control/held/held.go @@ -0,0 +1,93 @@ +// Package held records, on the leader, what each model holds of the +// cluster: its replicas' GPUs and, for one on CPUs, its cores. Kept by the +// hour, it is the cost side of a token's price (llmgw ctl prices suggest). +package held + +import ( + "context" + "log/slog" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// Inventory is where the allocations are. +type Inventory interface { + Nodes() []inventory.Node + Allocations(nodeID string) []inventory.Allocation +} + +// Store keeps what was held. +type Store interface { + AddHeld(ctx context.Context, at time.Time, held []store.Held) error +} + +// Recorder adds, every Every, what each model holds for that long. +type Recorder struct { + Inv Inventory + // Models is each replica's model, by replica id. + Models func() map[string]string + Store Store + Leader func() bool + Every time.Duration + Log *slog.Logger +} + +// Run records until ctx ends. +func (r *Recorder) Run(ctx context.Context) { + every := r.Every + if every <= 0 { + every = 5 * time.Minute + } + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case now := <-t.C: + if r.Leader != nil && !r.Leader() { + continue + } + held := Sample(r.Inv, r.Models(), every) + if err := r.Store.AddHeld(ctx, now, held); err != nil && r.Log != nil { + r.Log.Warn("could not record what the models held, for their prices", "err", err) + } + } + } +} + +// Sample is what each model holds now, for d: a GPU for each card a shard +// was given, and the cores of a shard given cores and no GPU. A replica +// whose model is not known (headroom kept for another, one just reaped) is +// left out. +func Sample(inv Inventory, models map[string]string, d time.Duration) []store.Held { + by := map[string]*store.Held{} + var order []string + secs := int64(d.Seconds()) + for _, n := range inv.Nodes() { + for _, a := range inv.Allocations(n.ID) { + m, ok := models[a.ReplicaID] + if !ok { + continue + } + h := by[m] + if h == nil { + h = &store.Held{Model: m} + by[m] = h + order = append(order, m) + } + if g := len(a.GPUIndices); g > 0 { + h.GPUSeconds += int64(g) * secs + } else { + h.CoreSeconds += int64(max(a.CPUCores, 1)) * secs + } + } + } + out := make([]store.Held, 0, len(order)) + for _, m := range order { + out = append(out, *by[m]) + } + return out +} diff --git a/internal/control/held/held_test.go b/internal/control/held/held_test.go new file mode 100644 index 0000000..f44d409 --- /dev/null +++ b/internal/control/held/held_test.go @@ -0,0 +1,40 @@ +package held + +import ( + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +type inv map[string][]inventory.Allocation + +func (i inv) Nodes() []inventory.Node { + var out []inventory.Node + for id := range i { + out = append(out, inventory.Node{ID: id}) + } + return out +} + +func (i inv) Allocations(id string) []inventory.Allocation { return i[id] } + +func TestSampleCountsCardsAndCores(t *testing.T) { + in := inv{ + "gpu1": {{ReplicaID: "big-1", GPUIndices: []int{0, 1}}, {ReplicaID: "headroom", GPUIndices: []int{2}}}, + "gpu2": {{ReplicaID: "big-2", GPUIndices: []int{0, 1}}}, + "cpu1": {{ReplicaID: "small-1", CPUCores: 4}, {ReplicaID: "tiny-1"}}, + } + models := map[string]string{"big-1": "big", "big-2": "big", "small-1": "small", "tiny-1": "tiny"} + got := map[string]store.Held{} + for _, h := range Sample(in, models, 5*time.Minute) { + got[h.Model] = h + } + if got["big"].GPUSeconds != 4*300 || got["small"].CoreSeconds != 4*300 || got["tiny"].CoreSeconds != 300 { + t.Errorf("held %+v", got) + } + if len(got) != 3 { + t.Errorf("headroom counted: %+v", got) + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index ec5ff9a..b7e1d68 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -39,6 +39,8 @@ var Reference = []Block{ {Key: "personal", Type: "Personal", Doc: "Personal is what a person may do on their own, outside any project: make\npersonal keys and use the chat page personally, within these limits.", Nested: "config.Personal"}, {Key: "pricing", Type: "map[string]Price", Doc: "Pricing is what each model costs, for budgets. A model absent from the\ntable costs whatever DefaultPrice says, which is nothing unless set.", Nested: ""}, {Key: "default_price", Type: "Price", Doc: "", Nested: "config.Price"}, + {Key: "gpu_hour_cost", Type: "float64", Doc: "GPUHourCost is what an hour of one GPU costs the operator, in the\nunits budgets are in: the hardware written off over its years, its\npower and room, and the people who run it, over the GPUs and the\nhours of a month. With CoreHourCost for a CPU core, it is what\n`llmgw ctl prices suggest` prices each model from: the hours its\nreplicas held, idle ones too, over the tokens it served.", Nested: ""}, + {Key: "core_hour_cost", Type: "float64", Doc: "", Nested: ""}, {Key: "flush_interval", Type: "time.Duration", Doc: "FlushInterval and BatchSize decide when usage records are written.", Nested: ""}, {Key: "batch_size", Type: "int", Doc: "", Nested: ""}, {Key: "buffer_size", Type: "int", Doc: "BufferSize bounds the queue. Beyond it records are dropped rather than\nblocking a request: losing accounting is bad, turning a stalled database\ninto an outage is worse.", Nested: ""}, diff --git a/internal/store/copy.go b/internal/store/copy.go index fc96479..47d9240 100644 --- a/internal/store/copy.go +++ b/internal/store/copy.go @@ -44,6 +44,7 @@ var copyOrder = []string{ "usage_5m", "usage_1h", "usage_1d", + "held_1h", "models", // No foreign key to models, deliberately: a measurement outlives the // catalog entry it was taken for, and losing it when somebody removes a diff --git a/internal/store/held.go b/internal/store/held.go new file mode 100644 index 0000000..f596d4f --- /dev/null +++ b/internal/store/held.go @@ -0,0 +1,62 @@ +package store + +import ( + "context" + "time" +) + +// Held is what one model held of the cluster. +type Held struct { + Model string + GPUSeconds int64 + CoreSeconds int64 +} + +// AddHeld adds what models held, during the hour at holds. +func (s *DB) AddHeld(ctx context.Context, at time.Time, held []Held) error { + bucket := at.UTC().Truncate(time.Hour).UnixMilli() + for _, h := range held { + if h.Model == "" || (h.GPUSeconds <= 0 && h.CoreSeconds <= 0) { + continue + } + if _, err := s.db.ExecContext(ctx, + `INSERT INTO held_1h (bucket, model, gpu_seconds, core_seconds) VALUES (?, ?, ?, ?) + ON CONFLICT(bucket, model) DO UPDATE SET + gpu_seconds = held_1h.gpu_seconds + excluded.gpu_seconds, + core_seconds = held_1h.core_seconds + excluded.core_seconds`, + bucket, h.Model, h.GPUSeconds, h.CoreSeconds); err != nil { + return err + } + } + return nil +} + +// HeldSince is what each model held since the hour since falls in, and +// the first hour anything was recorded in (zero with nothing). +func (s *DB) HeldSince(ctx context.Context, since time.Time) ([]Held, time.Time, error) { + from := since.UTC().Truncate(time.Hour).UnixMilli() + rows, err := s.db.QueryContext(ctx, + `SELECT model, SUM(gpu_seconds), SUM(core_seconds), MIN(bucket) FROM held_1h + WHERE bucket >= ? GROUP BY model ORDER BY model`, from) + if err != nil { + return nil, time.Time{}, err + } + defer rows.Close() + var out []Held + var first int64 + for rows.Next() { + var h Held + var b int64 + if err := rows.Scan(&h.Model, &h.GPUSeconds, &h.CoreSeconds, &b); err != nil { + return nil, time.Time{}, err + } + if first == 0 || b < first { + first = b + } + out = append(out, h) + } + if first == 0 { + return out, time.Time{}, rows.Err() + } + return out, time.UnixMilli(first).UTC(), rows.Err() +} diff --git a/internal/store/held_test.go b/internal/store/held_test.go new file mode 100644 index 0000000..74a4140 --- /dev/null +++ b/internal/store/held_test.go @@ -0,0 +1,35 @@ +package store + +import ( + "context" + "testing" + "time" +) + +func TestHeldAddsUpByHour(t *testing.T) { + s := capacityStore(t) + ctx := context.Background() + at := time.Date(2026, 11, 14, 10, 5, 0, 0, time.UTC) + for i := range 3 { + if err := s.AddHeld(ctx, at.Add(time.Duration(i)*5*time.Minute), + []Held{{Model: "qwen", GPUSeconds: 300}, {Model: "small", CoreSeconds: 1200}, {Model: "idle"}}); err != nil { + t.Fatal(err) + } + } + if err := s.AddHeld(ctx, at.Add(time.Hour), []Held{{Model: "qwen", GPUSeconds: 600}}); err != nil { + t.Fatal(err) + } + got, first, err := s.HeldSince(ctx, at.Add(-24*time.Hour)) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 || got[0] != (Held{Model: "qwen", GPUSeconds: 1500}) || got[1] != (Held{Model: "small", CoreSeconds: 3600}) { + t.Errorf("held %+v", got) + } + if !first.Equal(at.Truncate(time.Hour)) { + t.Errorf("first %v", first) + } + if got, _, _ := s.HeldSince(ctx, at.Add(time.Hour)); len(got) != 1 || got[0].GPUSeconds != 600 { + t.Errorf("since the next hour: %+v", got) + } +} diff --git a/internal/store/migrations/postgres/0053_held.sql b/internal/store/migrations/postgres/0053_held.sql new file mode 100644 index 0000000..9cbc813 --- /dev/null +++ b/internal/store/migrations/postgres/0053_held.sql @@ -0,0 +1,11 @@ +-- Ported from the SQLite migration of the same number. +-- What each model held of the cluster, an hour at a time: its replicas' +-- GPUs and, for one on CPUs, its cores, in seconds. With the tokens it +-- served, it is what a token of it cost (llmgw ctl prices suggest). +CREATE TABLE held_1h ( + bucket BIGINT NOT NULL, + model TEXT NOT NULL, + gpu_seconds BIGINT NOT NULL DEFAULT 0, + core_seconds BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (bucket, model) +); diff --git a/internal/store/migrations/sqlite/0053_held.sql b/internal/store/migrations/sqlite/0053_held.sql new file mode 100644 index 0000000..cfd7e85 --- /dev/null +++ b/internal/store/migrations/sqlite/0053_held.sql @@ -0,0 +1,10 @@ +-- What each model held of the cluster, an hour at a time: its replicas' +-- GPUs and, for one on CPUs, its cores, in seconds. With the tokens it +-- served, it is what a token of it cost (llmgw ctl prices suggest). +CREATE TABLE held_1h ( + bucket BIGINT NOT NULL, + model TEXT NOT NULL, + gpu_seconds BIGINT NOT NULL DEFAULT 0, + core_seconds BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (bucket, model) +);