Requests can be tagged: metadata, x-llmgw-tags and the user field on the usage record, and counted by the tags router.tag_labels names
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
e59ff52932
commit
29a7347df2
19 files changed
+436
-17
No files matched your search
+6
-1
@@ -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).**
|
||||
|
||||
@@ -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.**
|
||||
|
||||
+6
-1
@@ -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
|
||||
|
||||
@@ -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
|
||||
<id>` 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
|
||||
|
||||
@@ -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`
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"`
|
||||
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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"},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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:<id>" -> "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))
|
||||
|
||||
@@ -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 '';
|
||||
@@ -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 '';
|
||||
@@ -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
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user