diff --git a/BACKLOG.md b/BACKLOG.md index c61dfe03..4d196713 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -412,7 +412,12 @@ portal is not done — the page shows; the console and ctl change. templates driven by `server init` or the console; llmgw-native webhooks for model parked, budget spent, update available (from the reconciler and `modelserve`). -6. **Request metadata and tags (S).** Accept `metadata`/`user` from the +6. **Done (branch `features-c2`):** `metadata`, `x-llmgw-tags` and `user` + on the usage record (migration 0023), `ctl request` shows them, + `router.tag_labels` metrics with 100 values per tag. Left: export and + `ctl usage` by tag or end user (the hourly and daily sums do not carry + tags; it would read the request records, within their retention). + **Request metadata and tags (S).** Accept `metadata`/`user` from the body or `x-llmgw-tags`; store on the usage row; tenant metric labels behind an allowlist. `router/handlers.go` parse, `accounting/recorder.go`. 7. **Canary and weighted A/B between model revisions (M).** diff --git a/CHANGELOG.md b/CHANGELOG.md index cf7b0f41..2a224821 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -71,6 +71,17 @@ 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. +### Requests can be tagged + +- **Tags and end users on every usage record.** OpenAI's `metadata` and the + `x-llmgw-tags` header (`feature=search, env=prod`) tag a request; OpenAI's + `user` field names the client's own end user. Both are kept with the + request's usage and shown by `llmgw ctl request`. Tags are not sent to the + engine. A malformed one is refused with 400. +- **Counted by tag, for the tags named.** `router.tag_labels` lists the tags + counted in `llmgw_tagged_requests_total` and `llmgw_tagged_tokens_total`, + the first 100 values of each apart and the rest as `other`. + ### A hardened install by default - **`server init` and the setup screens write a hardened configuration.** diff --git a/docs/data.md b/docs/data.md index 77ed63a4..c103c23b 100644 --- a/docs/data.md +++ b/docs/data.md @@ -32,7 +32,12 @@ the text, the value that matched, or the placeholder map of a redaction. See **Usage records**, one per request: the key, the project, the model asked for, the model that answered, prompt and completion token counts, cost, HTTP status and latency. No message content. The row is in `internal/store/store.go` as -`Usage` and there is no field for text in it. +`Usage` and there is no field for text in it. What the client chose to say +about the request is kept with it: OpenAI's `user` field and its `metadata` +or `x-llmgw-tags` — short strings, set by the client, for attributing spend. +A client that puts a person's name or address there puts it in the usage +records; the field is the client's, and so is that choice. Tags are not sent +on to the engine. **API keys**, as SHA-256 hashes. The key itself is shown once, at creation, and never stored — so a leaked database hands over no working credential, and diff --git a/docs/install.md b/docs/install.md index 9059a9ba..4eb4b71e 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1062,6 +1062,41 @@ outcome than a smaller cache, so cap it: `shared_buffers = 96MB` and `max_connections = 40` are enough for a store whose working set is measured in megabytes. +## Tagging requests + +A client can say what a request was for — which feature, which customer, +which environment — and the spend can then be found by it inside a project: + +``` +curl http://gateway:8080/v1/chat/completions \ + -H "Authorization: Bearer $LLMGW_API_KEY" \ + -H "x-llmgw-tags: feature=search, env=prod" \ + -d '{"model": "qwen-1.5b", "user": "customer-42", + "metadata": {"feature": "search"}, + "messages": [{"role": "user", "content": "hello"}]}' +``` + +- **`metadata`** is OpenAI's: an object of strings, up to 16 keys. **`x-llmgw-tags`** + is the same as a header, for clients that cannot change the body; for a + key both give, the header wins. A key is 1 to 64 letters, digits, `_`, + `.` or `-`, a value at most 512 bytes; anything else is refused with 400 + rather than dropped, since a tag lost silently is spend nobody finds. +- **`user`** is OpenAI's end-user field: the client's own name for whoever + it serves, not an account here. It is passed on to the engine, as OpenAI + does; tags are not. +- **Where they show:** every usage record keeps them, and `llmgw ctl request + ` prints them. For Prometheus, name the tags worth a series: + + ```yaml + router: + tag_labels: [feature, env] + ``` + + `llmgw_tagged_requests_total{tag,value,outcome}` and + `llmgw_tagged_tokens_total{tag,value,direction}` count those, the first + 100 values of each apart and the rest as `other`: a tag is the client's + to set, and every value is a series. + ## Conversations across models Off everywhere until a project asks. llmgw otherwise stores no message content diff --git a/docs/reference.md b/docs/reference.md index df96ec45..ceb5a022 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -190,6 +190,7 @@ rather than repeated. | `session_header` | `string` | [Router](#router) | | `inject_usage` | `bool` | [Router](#router) | | `tenant_label` | `string` | [Router](#router) | +| `tag_labels` | `[]string` | [Router](#router) | | `max_body_bytes` | `int64` | [Router](#router) | | `trusted_proxies` | `[]string` | [Router](#router) | | `timeouts` | `Timeouts` | [Router](#router) | @@ -1557,6 +1558,16 @@ labelled by — llmgw_tenant_requests_total and its siblings: "org", on a cluster of thousands is more series than Prometheus wants, and a team is what an alert is usually about; "team" is the default. +### `tag_labels` + +`[]string` + +TagLabels are the request tags (metadata, x-llmgw-tags) counted in +llmgw_tagged_requests_total and llmgw_tagged_tokens_total, by value. +Only these: a tag is the client's to set, and every value of one is +a series. The first 100 values of each are counted apart, the rest as +"other". Empty counts none; the usage records keep every tag anyway. + ### `max_body_bytes` `int64` diff --git a/internal/cli/ctl_request.go b/internal/cli/ctl_request.go index 036b888e..d0c8e5fc 100644 --- a/internal/cli/ctl_request.go +++ b/internal/cli/ctl_request.go @@ -4,6 +4,8 @@ import ( "encoding/json" "fmt" "net/url" + "slices" + "strings" "github.com/spf13/cobra" @@ -39,6 +41,21 @@ func newRequestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { fmt.Fprintf(out, "time %d ms total: %d queued, %d to the first token\n", v.LatencyMs, v.QueueMs, v.TTFTMs) fmt.Fprintf(out, "tokens %d in, %d out\n", v.PromptTokens, v.CompletionTokens) fmt.Fprintf(out, "who key %s, project %s%s\n", v.KeyID, v.ProjectID, prefixed(", user ", v.UserID)) + if v.EndUser != "" { + fmt.Fprintf(out, "end user %s (the client's own name for them)\n", v.EndUser) + } + if len(v.Tags) > 0 { + keys := make([]string, 0, len(v.Tags)) + for k := range v.Tags { + keys = append(keys, k) + } + slices.Sort(keys) + var kv []string + for _, k := range keys { + kv = append(kv, k+"="+v.Tags[k]) + } + fmt.Fprintf(out, "tags %s\n", strings.Join(kv, ", ")) + } if v.GuardAction != "" { fmt.Fprintf(out, "guard %s by policy %s (%s), %d ms\n", v.GuardAction, v.GuardPolicy, or(v.GuardCategory, "-"), v.GuardMs) diff --git a/internal/config/config.go b/internal/config/config.go index 5d7050cf..d19a2e20 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -443,6 +443,13 @@ type Router struct { // team is what an alert is usually about; "team" is the default. TenantLabel string `koanf:"tenant_label"` + // TagLabels are the request tags (metadata, x-llmgw-tags) counted in + // llmgw_tagged_requests_total and llmgw_tagged_tokens_total, by value. + // Only these: a tag is the client's to set, and every value of one is + // a series. The first 100 values of each are counted apart, the rest as + // "other". Empty counts none; the usage records keep every tag anyway. + TagLabels []string `koanf:"tag_labels"` + // MaxBodyBytes bounds the request body the router will read. MaxBodyBytes int64 `koanf:"max_body_bytes"` diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 12a217e4..19aecd00 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -1406,6 +1406,9 @@ type RequestView struct { GuardPolicy string `json:"guard_policy,omitempty"` GuardCategory string `json:"guard_category,omitempty"` GuardMs int64 `json:"guard_ms,omitempty"` + // What the client said about it: OpenAI's user field, and its tags. + EndUser string `json:"end_user,omitempty"` + Tags map[string]string `json:"tags,omitempty"` } func (a *API) request(w http.ResponseWriter, r *http.Request) { @@ -1429,6 +1432,7 @@ func (a *API) request(w http.ResponseWriter, r *http.Request) { LatencyMs: u.LatencyMs, QueueMs: u.QueueMs, TTFTMs: u.TTFTMs, PromptTokens: u.PromptTokens, CompletionTokens: u.CompletionTokens, CostMicros: u.CostMicros, GuardAction: u.GuardAction, GuardPolicy: u.GuardPolicy, GuardCategory: u.GuardCategory, GuardMs: u.GuardMs, + EndUser: u.EndUser, Tags: u.Tags, }) } diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index 839dea7e..39f81dc0 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -230,6 +230,7 @@ var Reference = []Block{ {Key: "session_header", Type: "string", Doc: "SessionHeader carries the session id for affinity (plan §7.5 tier 0).", Nested: ""}, {Key: "inject_usage", Type: "bool", Doc: "InjectUsage asks engines for usage even when the client did not, so token\naccounting works regardless of client behaviour.", Nested: ""}, {Key: "tenant_label", Type: "string", Doc: "TenantLabel is the level of the tree the per-tenant request series are\nlabelled by — llmgw_tenant_requests_total and its siblings: \"org\",\n\"team\" or \"project\", or \"none\" for no such series. A label per project\non a cluster of thousands is more series than Prometheus wants, and a\nteam is what an alert is usually about; \"team\" is the default.", Nested: ""}, + {Key: "tag_labels", Type: "[]string", Doc: "TagLabels are the request tags (metadata, x-llmgw-tags) counted in\nllmgw_tagged_requests_total and llmgw_tagged_tokens_total, by value.\nOnly these: a tag is the client's to set, and every value of one is\na series. The first 100 values of each are counted apart, the rest as\n\"other\". Empty counts none; the usage records keep every tag anyway.", Nested: ""}, {Key: "max_body_bytes", Type: "int64", Doc: "MaxBodyBytes bounds the request body the router will read.", Nested: ""}, {Key: "trusted_proxies", Type: "[]string", Doc: "TrustedProxies are the reverse proxies in front of the client port,\nas addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host\nare believed; anyone else's are not, since they decide the address\nwritten into the setup scripts and whose attempts are counted where.", Nested: ""}, {Key: "timeouts", Type: "Timeouts", Doc: "", Nested: "config.Timeouts"}, diff --git a/internal/router/guard.go b/internal/router/guard.go index 114e467b..6df9d144 100644 --- a/internal/router/guard.go +++ b/internal/router/guard.go @@ -135,6 +135,7 @@ func (r *Router) refuseGuarded(w http.ResponseWriter, req *parsedRequest, v guar At: time.Now(), KeyID: req.KeyID, UserID: req.UserID, ProjectID: req.ProjectID, Model: req.Model, ServedModel: req.Model, Status: status, LatencyMs: time.Since(start).Milliseconds(), RequestID: req.RequestID, QueueMs: req.QueueMs, + EndUser: req.EndUser, Tags: req.Tags, } guardFields(&u, v, guard.Verdict{}) // Settled at nothing: the estimate the limit check charged is diff --git a/internal/router/handlers.go b/internal/router/handlers.go index 5e6a5e43..f18a05c0 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -280,6 +280,10 @@ func (r *Router) Handler() http.Handler { // business interpreting. type parsedRequest struct { Path string + // Tags are the client's metadata and x-llmgw-tags; EndUser is OpenAI's + // user field. Both are recorded with the usage, never sent on. + Tags map[string]string + EndUser string // Model is what will be dispatched, which is not always what was asked // for: a caller out of budget may be moved to a model they can afford. Model string @@ -338,9 +342,10 @@ type parsedRequest struct { // inferenceBody covers the fields the router reads across all inference // endpoints. Unknown fields are ignored here and preserved in UpstreamBody. type inferenceBody struct { - Model string `json:"model"` - Stream bool `json:"stream"` - User string `json:"user"` + Model string `json:"model"` + Stream bool `json:"stream"` + User string `json:"user"` + Metadata json.RawMessage `json:"metadata"` Messages []struct { Role string `json:"role"` Content json.RawMessage `json:"content"` @@ -587,6 +592,7 @@ func (r *Router) account(req *parsedRequest, res dispatchResult, status int, too LatencyMs: took.Milliseconds(), RequestID: req.RequestID, Replica: res.endpointID, QueueMs: req.QueueMs, TTFTMs: res.outcome.TTFT.Milliseconds(), + EndUser: req.EndUser, Tags: req.Tags, } guardFields(&u, req.guardIn, res.guard) if u.ServedModel == "" { @@ -625,6 +631,7 @@ func (r *Router) rejectLimit(w http.ResponseWriter, req *parsedRequest, err erro At: time.Now(), KeyID: req.KeyID, UserID: req.UserID, ProjectID: req.ProjectID, Model: req.Model, ServedModel: req.Model, Status: http.StatusTooManyRequests, LatencyMs: time.Since(start).Milliseconds(), RequestID: req.RequestID, + EndUser: req.EndUser, Tags: req.Tags, }) r.accessLog(req, dispatchResult{err: err}, http.StatusTooManyRequests, time.Since(start)) } @@ -644,7 +651,14 @@ func (r *Router) parse(req *http.Request, path string) (*parsedRequest, error) { return nil, errors.New("model is required") } + tags, err := requestTags(b.Metadata, req.Header.Get(tagsHeader)) + if err != nil { + return nil, err + } + p := &parsedRequest{ + Tags: tags, + EndUser: endUser(b.User), Path: path, Model: b.Model, RequestedModel: b.Model, @@ -680,10 +694,16 @@ func (r *Router) parse(req *http.Request, path string) (*parsedRequest, error) { // see promptBytes. p.PromptTokens = promptBytes(&b) / 4 + // Tags are the gateway's: an engine has no use for them, and its logs + // are no place for what a client wrote there. + if len(b.Metadata) > 0 { + p.UpstreamBody = withoutField(p.UpstreamBody, "metadata") + } + // Ask the engine for usage the client did not request, so accounting works // regardless of client behaviour; the extra chunk is dropped on the way back. if r.cfg.InjectUsage && b.Stream && (b.StreamOptions == nil || !b.StreamOptions.IncludeUsage) { - if injected, ok := injectIncludeUsage(body); ok { + if injected, ok := injectIncludeUsage(p.UpstreamBody); ok { p.UpstreamBody = injected p.SuppressUsageChunk = true } diff --git a/internal/router/router.go b/internal/router/router.go index b94ede70..52ed428d 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -209,6 +209,7 @@ func New(cfg config.Server, log *slog.Logger) (*Router, error) { // guessing at it. metrics := newMetrics() metrics.tenant.level = cfg.Router.TenantLabel + metrics.tenant.setTagLabels(cfg.Router.TagLabels) var prefixScorer *scheduler.PrefixScorer if cfg.Router.Prefix.Enabled { prefixScorer = scheduler.NewPrefixScorer( diff --git a/internal/router/tags.go b/internal/router/tags.go new file mode 100644 index 00000000..481698e3 --- /dev/null +++ b/internal/router/tags.go @@ -0,0 +1,91 @@ +package router + +import ( + "encoding/json" + "fmt" + "regexp" + "strings" +) + +// What a client may say about a request, for attributing spend inside a +// project: OpenAI's metadata (at most 16 keys, as OpenAI allows) and the +// x-llmgw-tags header, which wins for a key both give. +const ( + tagsHeader = "x-llmgw-tags" + maxTags = 16 + maxTagValue = 512 + maxEndUser = 256 +) + +var tagKey = regexp.MustCompile(`^[A-Za-z0-9_.-]{1,64}$`) + +// requestTags reads metadata from the body and x-llmgw-tags ("team=search, +// feature=autocomplete") from the header. An invalid one is the client's +// mistake and is said, not dropped: a tag silently lost is spend that +// cannot be found again. +func requestTags(metadata json.RawMessage, header string) (map[string]string, error) { + tags := map[string]string{} + if len(metadata) > 0 && string(metadata) != "null" { + var m map[string]any + if err := json.Unmarshal(metadata, &m); err != nil { + return nil, fmt.Errorf("metadata is an object of strings") + } + for k, v := range m { + s, ok := v.(string) + if !ok { + return nil, fmt.Errorf("metadata.%s is not a string", k) + } + tags[k] = s + } + } + if header = strings.TrimSpace(header); header != "" { + for _, part := range strings.Split(header, ",") { + k, v, ok := strings.Cut(strings.TrimSpace(part), "=") + if !ok { + return nil, fmt.Errorf("%s: %q is not key=value", tagsHeader, part) + } + tags[strings.TrimSpace(k)] = strings.TrimSpace(v) + } + } + if len(tags) > maxTags { + return nil, fmt.Errorf("%d tags; at most %d", len(tags), maxTags) + } + for k, v := range tags { + if !tagKey.MatchString(k) { + return nil, fmt.Errorf("tag %q: a key is 1 to 64 letters, digits, '_', '.' or '-'", k) + } + if len(v) > maxTagValue { + return nil, fmt.Errorf("tag %s: the value is longer than %d bytes", k, maxTagValue) + } + } + if len(tags) == 0 { + return nil, nil + } + return tags, nil +} + +// endUser is OpenAI's user field as kept: bounded, since it is the client's. +func endUser(u string) string { + if len(u) > maxEndUser { + return u[:maxEndUser] + } + return u +} + +// withoutField removes a top-level field from a JSON body, or returns it +// unchanged when the field is not there or the body is not an object. +func withoutField(body []byte, field string) []byte { + var m map[string]json.RawMessage + if err := json.Unmarshal(body, &m); err != nil { + return body + } + if _, ok := m[field]; !ok { + return body + } + delete(m, field) + out, err := json.Marshal(m) + if err != nil { + return body + } + return out +} diff --git a/internal/router/tags_test.go b/internal/router/tags_test.go new file mode 100644 index 00000000..b551086e --- /dev/null +++ b/internal/router/tags_test.go @@ -0,0 +1,101 @@ +package router + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" +) + +func TestTagsComeFromMetadataAndTheHeader(t *testing.T) { + got, err := requestTags(json.RawMessage(`{"feature":"search","team":"a"}`), "team=b, env=prod") + if err != nil { + t.Fatal(err) + } + if len(got) != 3 || got["team"] != "b" || got["feature"] != "search" || got["env"] != "prod" { + t.Errorf("%v", got) + } + if got, err := requestTags(nil, ""); got != nil || err != nil { + t.Errorf("none: %v %v", got, err) + } + many := map[string]string{} + for i := range 17 { + many[strings.Repeat("k", i+1)] = "v" + } + b, _ := json.Marshal(many) + for name, c := range map[string]struct { + meta string + header string + }{ + "not strings": {`{"n": 1}`, ""}, + "not an object": {`["a"]`, ""}, + "no equals": {"", "feature"}, + "bad key": {`{"a b": "x"}`, ""}, + "long value": {`{"a": "` + strings.Repeat("x", 513) + `"}`, ""}, + "too many": {string(b), ""}, + } { + if _, err := requestTags(json.RawMessage(c.meta), c.header); err == nil { + t.Errorf("%s: accepted", name) + } + } +} + +func TestTagsAreRecordedAndNotSentToTheEngine(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := pricedRouter(t, e.srv.URL) + rt.m.tenant.setTagLabels([]string{"feature"}) + + rec := post(t, rt.Handler(), + map[string]any{"model": "q14", "user": "customer-42", "metadata": map[string]string{"feature": "search"}, + "messages": []any{map[string]any{"role": "user", "content": "hello"}}}, + map[string]string{"Authorization": "Bearer " + secret, "X-Request-Id": "tagged-1", "X-Llmgw-Tags": "env=prod"}) + if rec.Code != 200 { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + sent, _ := e.last.Load().([]byte) + if strings.Contains(string(sent), "metadata") || strings.Contains(string(sent), "search") { + t.Errorf("the engine was sent the tags: %s", sent) + } + if !strings.Contains(string(sent), "customer-42") { + t.Errorf("the user field, which engines know, was not passed on: %s", sent) + } + + deadline := time.Now().Add(5 * time.Second) + for { + u, ok, err := rt.store.UsageByRequest(t.Context(), "tagged-1") + if err != nil { + t.Fatal(err) + } + if ok { + if u.EndUser != "customer-42" || u.Tags["feature"] != "search" || u.Tags["env"] != "prod" { + t.Errorf("recorded end user %q tags %v", u.EndUser, u.Tags) + } + break + } + if time.Now().After(deadline) { + t.Fatal("no usage recorded") + } + time.Sleep(20 * time.Millisecond) + } + + bad := post(t, rt.Handler(), + map[string]any{"model": "q14", "metadata": map[string]any{"n": 1}, + "messages": []any{map[string]any{"role": "user", "content": "hello"}}}, + map[string]string{"Authorization": "Bearer " + secret}) + if bad.Code != 400 || !strings.Contains(bad.Body.String(), "metadata.n") { + t.Errorf("a bad tag: %d %s", bad.Code, bad.Body.String()) + } +} + +func TestOnlyNamedTagsAreCountedAndTheirValuesBounded(t *testing.T) { + m := newTenantMetrics() + m.setTagLabels([]string{"feature"}) + for i := range maxTagValues + 20 { + m.observe(&parsedRequest{Tags: map[string]string{"feature": strings.Repeat("f", i+1), "secret": "x"}}, 200, 1, 1, 0) + } + if n := testutil.CollectAndCount(m.tagRequests); n != maxTagValues+1 { + t.Errorf("%d series; want %d values and other", testutil.CollectAndCount(m.tagRequests), maxTagValues+1) + } +} diff --git a/internal/router/tenantmetrics.go b/internal/router/tenantmetrics.go index 7202b7b2..8f855a28 100644 --- a/internal/router/tenantmetrics.go +++ b/internal/router/tenantmetrics.go @@ -28,6 +28,12 @@ type tenantMetrics struct { ttft *prometheus.HistogramVec budget *prometheus.GaugeVec + // Request tags, for the keys router.tag_labels names. + tagKeys map[string]bool + tagRequests *prometheus.CounterVec + tagTokens *prometheus.CounterVec + tagValues map[string]map[string]bool // tag -> values counted apart + mu sync.RWMutex names map[string]string // "team:" -> "acme/platform" } @@ -52,12 +58,68 @@ func newTenantMetrics() *tenantMetrics { Name: "llmgw_budget_spent_ratio", Help: "Spend this period over the budget, for every organization, team, project and person given one. 1 is spent.", }, []string{"scope", "id", "name", "period"}), - names: map[string]string{}, + tagRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "llmgw_tagged_requests_total", + Help: "Requests by a request tag router.tag_labels names, its value, and outcome: ok, error, or limited.", + }, []string{"tag", "value", "outcome"}), + tagTokens: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "llmgw_tagged_tokens_total", + Help: "Tokens by a request tag router.tag_labels names, its value, and direction.", + }, []string{"tag", "value", "direction"}), + tagValues: map[string]map[string]bool{}, + names: map[string]string{}, + } +} + +// maxTagValues is how many values of one tag are counted apart. +const maxTagValues = 100 + +// setTagLabels names the tags counted. +func (t *tenantMetrics) setTagLabels(keys []string) { + t.tagKeys = map[string]bool{} + for _, k := range keys { + t.tagKeys[k] = true + } +} + +// tagValue is the label a tag's value is counted under: itself while there +// is room, "other" after. +func (t *tenantMetrics) tagValue(tag, value string) string { + t.mu.Lock() + defer t.mu.Unlock() + seen := t.tagValues[tag] + if seen == nil { + seen = map[string]bool{} + t.tagValues[tag] = seen + } + if seen[value] { + return value + } + if len(seen) >= maxTagValues { + return "other" + } + seen[value] = true + return value +} + +func (t *tenantMetrics) observeTags(req *parsedRequest, outcome string, prompt, completion int64) { + for k, v := range req.Tags { + if !t.tagKeys[k] { + continue + } + v = t.tagValue(k, v) + t.tagRequests.WithLabelValues(k, v, outcome).Inc() + if prompt > 0 { + t.tagTokens.WithLabelValues(k, v, "prompt").Add(float64(prompt)) + } + if completion > 0 { + t.tagTokens.WithLabelValues(k, v, "completion").Add(float64(completion)) + } } } func (t *tenantMetrics) collectors() []prometheus.Collector { - return []prometheus.Collector{t.requests, t.tokens, t.ttft, t.budget} + return []prometheus.Collector{t.requests, t.tokens, t.ttft, t.budget, t.tagRequests, t.tagTokens} } // tenantOf is the label a request is counted under: the name of its @@ -91,10 +153,6 @@ func (t *tenantMetrics) tenantOf(req *parsedRequest) string { // observe counts one finished or refused request. func (t *tenantMetrics) observe(req *parsedRequest, status int, prompt, completion int64, ttft time.Duration) { - tenant := t.tenantOf(req) - if tenant == "" { - return - } outcome := "ok" switch { case status == 429: @@ -102,6 +160,13 @@ func (t *tenantMetrics) observe(req *parsedRequest, status int, prompt, completi case status >= 400: outcome = "error" } + if len(t.tagKeys) > 0 { + t.observeTags(req, outcome, prompt, completion) + } + tenant := t.tenantOf(req) + if tenant == "" { + return + } t.requests.WithLabelValues(tenant, outcome).Inc() if prompt > 0 { t.tokens.WithLabelValues(tenant, "prompt").Add(float64(prompt)) diff --git a/internal/store/migrations/postgres/0023_usage_tags.sql b/internal/store/migrations/postgres/0023_usage_tags.sql new file mode 100644 index 00000000..433ca5ad --- /dev/null +++ b/internal/store/migrations/postgres/0023_usage_tags.sql @@ -0,0 +1,7 @@ +-- Ported from the SQLite migration of the same number. +-- What a client said about a request: OpenAI's user field (end_user, the +-- caller's own name for whoever it serves) and its metadata or +-- x-llmgw-tags (tags, a JSON object of strings, keys sorted). For +-- attributing spend inside a project; never the text. +ALTER TABLE usage ADD COLUMN end_user TEXT NOT NULL DEFAULT ''; +ALTER TABLE usage ADD COLUMN tags TEXT NOT NULL DEFAULT ''; diff --git a/internal/store/migrations/sqlite/0023_usage_tags.sql b/internal/store/migrations/sqlite/0023_usage_tags.sql new file mode 100644 index 00000000..b7a822ad --- /dev/null +++ b/internal/store/migrations/sqlite/0023_usage_tags.sql @@ -0,0 +1,6 @@ +-- What a client said about a request: OpenAI's user field (end_user, the +-- caller's own name for whoever it serves) and its metadata or +-- x-llmgw-tags (tags, a JSON object of strings, keys sorted). For +-- attributing spend inside a project; never the text. +ALTER TABLE usage ADD COLUMN end_user TEXT NOT NULL DEFAULT ''; +ALTER TABLE usage ADD COLUMN tags TEXT NOT NULL DEFAULT ''; diff --git a/internal/store/sqlite.go b/internal/store/sqlite.go index 4da30c75..85b2ed55 100644 --- a/internal/store/sqlite.go +++ b/internal/store/sqlite.go @@ -430,8 +430,8 @@ func (s *DB) WriteUsage(ctx context.Context, records []Usage) error { `INSERT INTO usage (ts, key_id, project_id, model, served_model, prompt_tokens, completion_tokens, cost_micros, status, latency_ms, user_id, request_id, replica, queue_ms, ttft_ms, - guard_action, guard_policy, guard_category, guard_ms) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) + guard_action, guard_policy, guard_category, guard_ms, end_user, tags) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) if err != nil { return err } @@ -441,7 +441,7 @@ func (s *DB) WriteUsage(ctx context.Context, records []Usage) error { r.Model, r.ServedModel, r.PromptTokens, r.CompletionTokens, r.CostMicros, r.Status, r.LatencyMs, r.UserID, r.RequestID, r.Replica, r.QueueMs, r.TTFTMs, - r.GuardAction, r.GuardPolicy, r.GuardCategory, r.GuardMs); err != nil { + r.GuardAction, r.GuardPolicy, r.GuardCategory, r.GuardMs, r.EndUser, tagsJSON(r.Tags)); err != nil { return err } } @@ -850,14 +850,16 @@ func (s *DB) SnapshotTo(ctx context.Context, path string) error { func (s *DB) UsageByRequest(ctx context.Context, requestID string) (Usage, bool, error) { var u Usage var ts int64 + var tags string err := s.db.QueryRowContext(ctx, `SELECT ts, key_id, project_id, COALESCE(user_id, ''), model, served_model, prompt_tokens, completion_tokens, cost_micros, status, latency_ms, request_id, replica, queue_ms, ttft_ms, - guard_action, guard_policy, guard_category, guard_ms + guard_action, guard_policy, guard_category, guard_ms, end_user, tags FROM usage WHERE request_id = ? ORDER BY ts DESC LIMIT 1`, requestID). Scan(&ts, &u.KeyID, &u.ProjectID, &u.UserID, &u.Model, &u.ServedModel, &u.PromptTokens, &u.CompletionTokens, &u.CostMicros, &u.Status, &u.LatencyMs, &u.RequestID, &u.Replica, - &u.QueueMs, &u.TTFTMs, &u.GuardAction, &u.GuardPolicy, &u.GuardCategory, &u.GuardMs) + &u.QueueMs, &u.TTFTMs, &u.GuardAction, &u.GuardPolicy, &u.GuardCategory, &u.GuardMs, + &u.EndUser, &tags) if errors.Is(err, sql.ErrNoRows) { return Usage{}, false, nil } @@ -865,5 +867,6 @@ func (s *DB) UsageByRequest(ctx context.Context, requestID string) (Usage, bool, return Usage{}, false, err } u.At = time.UnixMilli(ts) + u.Tags = parseTags(tags) return u, true, nil } diff --git a/internal/store/store.go b/internal/store/store.go index 5183a25a..d1d338db 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -16,6 +16,7 @@ package store import ( "context" + "encoding/json" "errors" "strings" "time" @@ -262,6 +263,33 @@ type Usage struct { GuardPolicy string GuardCategory string GuardMs int64 + + // EndUser is OpenAI's user field as the client sent it: its own name + // for whoever it serves, not an account here. Tags are its metadata + // and x-llmgw-tags. Both for attributing spend inside a project. + EndUser string + Tags map[string]string +} + +// tagsJSON is how tags are kept: a JSON object, keys sorted, or empty. +func tagsJSON(t map[string]string) string { + if len(t) == 0 { + return "" + } + b, _ := json.Marshal(t) // encoding/json sorts map keys + return string(b) +} + +// parseTags reads tagsJSON's form back. +func parseTags(s string) map[string]string { + if s == "" { + return nil + } + var t map[string]string + if json.Unmarshal([]byte(s), &t) != nil { + return nil + } + return t } // Store is the persistent half of tenancy and accounting.