diff --git a/BACKLOG.md b/BACKLOG.md index 95631cb7..e30e132b 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -441,7 +441,11 @@ portal is not done — the page shows; the console and ctl change. **Batch API and Files (M–L).** `/v1/files`, `/v1/batches`, JSONL storage, a worker feeding the bronze band, accounting. New `internal/router/batch`. -9. **Request/response logging and replay (M).** Opt-in per project with +9. **Done (branch `features-c2`):** `keys requests` (retention, sample), + kept after redaction, `ctl request --body/--replay`, console, pruning. + Left: a streamed answer is kept as its text, not its chunks; replay is + one request at a time (a set of them against a model is item 21's eval). + **Request/response logging and replay (M).** Opt-in per project with retention like threads; sampled or full; redaction hooks from B; `ctl request --body`; replay against another model. 10. **OpenAI compatibility gaps (S–M each).** Responses: `text.format` → diff --git a/CHANGELOG.md b/CHANGELOG.md index b8c411b9..a80c6d13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -71,6 +71,15 @@ 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 kept, and replayed + +- **A project can keep its requests and answers** (`llmgw keys requests + --project P --retain-hours N [--sample 0.05]`, or **Keep requests…** in the + console): what the model was sent, after redaction, and what the client + got. Off until asked; never under `no_store`; pruned hourly. +- **`llmgw ctl request ID --body`** shows one; **`--replay MODEL`** sends it + to another model and shows both answers. + ### Batches - **OpenAI's Batch and Files APIs.** `POST /v1/files` (purpose batch), diff --git a/docs/console.md b/docs/console.md index 44fcbdc7..cfab53f4 100644 --- a/docs/console.md +++ b/docs/console.md @@ -192,7 +192,8 @@ with anything secret taken out. `llmgw ctl audit` is the same, with filters. projects, with members, keys and limits. **A** adds an organization. On a row: add a team or project inside it, rename, move, limits for the level and for each of its members, its capacity quota, and for a project how long -conversations and agent notes are kept. **Longest key lifetime…** caps how +conversations and agent notes are kept, and whether its requests are +(**Keep requests…**: for how long, and what share). **Longest key lifetime…** caps how long a key made there may live, and is the expiry of one made without any (`llmgw keys ttl`; `auth.max_key_ttl` in server.yaml caps every key). **Remove** takes away only something empty. diff --git a/docs/data.md b/docs/data.md index 6d3cf02e..1b6d4271 100644 --- a/docs/data.md +++ b/docs/data.md @@ -51,6 +51,11 @@ limits attached to it. Passwords are hashed. **Conversations and agent notes** — only for a project that asked. See below. +**Kept requests** — only for a project that asked (`llmgw keys requests`): +what each request sent the model, after the guardrails' redaction and with +any stored conversation in it, and the answer the client got, for the hours +it said, all of them or a share. Never under a `no_store` policy. + **Batch files** — when a client uses the Batch API: the file of requests it uploaded (prompts) and the files the batch wrote (answers), in the store, until deleted (`DELETE /v1/files/{id}`) or, for what a batch wrote, until @@ -61,7 +66,7 @@ them. ## What is not stored Prompts. Completions. Anything the model was shown or produced, unless the -project turned on conversations or notes, or sent a batch. +project turned on conversations, notes or kept requests, or sent a batch. The chat page in llmgw's own web interface keeps the conversation in a JavaScript variable in the browser. Refresh it and the conversation is gone, diff --git a/docs/install.md b/docs/install.md index ae7b9f69..984840b7 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1130,6 +1130,29 @@ curl http://gateway:8080/v1/chat/completions \ 100 values of each apart and the rest as `other`: a tag is the client's to set, and every value is a series. +## Keeping requests, and replaying them + +To follow a request up — what exactly was sent, what came back — or to try +real traffic on another model, a project can keep its requests: + +``` +llmgw keys requests --project Acme/ml/train --retain-hours 72 # every one +llmgw keys requests --project Acme/ml/train --retain-hours 168 --sample 0.05 +llmgw ctl request req_3f2a… --body # what was sent and answered +llmgw ctl request req_3f2a… --replay qwen2.5-14b-instruct # the same, to another model +llmgw keys requests --project Acme/ml/train --off +``` + +- **What is kept:** the request as the model was sent it — after the + guardrails' redaction, with any stored conversation spliced in — and the + answer as the client got it, up to 128 KB of it. A share (`--sample`) is + the same requests every time for the same ids. +- **Replay** sends it as the gateway itself: past the limits, the budget and + the guardrails, which judged it the first time, and charged to nobody. +- **Never** for a project under a guardrails policy with `no_store`. Kept + requests past their hours, or of a project turned off, are removed hourly. + The console has it as **Keep requests…** on a project. + ## Batches Work that can wait — a night's classification, embeddings for a document diff --git a/internal/cli/ctl_request.go b/internal/cli/ctl_request.go index d0c8e5fc..74cba2ee 100644 --- a/internal/cli/ctl_request.go +++ b/internal/cli/ctl_request.go @@ -10,18 +10,34 @@ import ( "github.com/spf13/cobra" "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" + "gitea.mynodes.xyz/ilker/llmgw/internal/store" ) func newRequestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { - return &cobra.Command{ + var ( + body bool + replay string + ) + c := &cobra.Command{ Use: "request ID", Short: "What one request went through: replica, queue, first token, tokens, guardrails", Long: "Every response carries its id in X-Request-Id, and a client may send its\n" + "own. Given one, this says which replica answered, how long admission held\n" + "it, how long the first token took and what it cost — the answer to \"why\n" + - "was this slow\". The access log has the same fields under \"request\".", - Args: cobra.ExactArgs(1), + "was this slow\". The access log has the same fields under \"request\".\n\n" + + "For a project that keeps its requests (`llmgw keys requests`), --body\n" + + "shows what the model was sent and what came back, and --replay MODEL\n" + + "sends the same request to another model, past the limits and the\n" + + "guardrails, and shows its answer beside the first.", + Example: " llmgw ctl request req_3f2a…\n llmgw ctl request req_3f2a… --body\n llmgw ctl request req_3f2a… --replay qwen2.5-14b-instruct", + Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { + if replay != "" { + return replayRequest(cmd, client, args[0], replay, *asJSON) + } + if body { + return showRequestBody(cmd, client, args[0], *asJSON) + } var v adminapi.RequestView if err := client().get(cmd.Context(), "/api/v1/requests/"+url.PathEscape(args[0]), &v); err != nil { return err @@ -65,4 +81,70 @@ func newRequestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { return nil }, } + c.Flags().BoolVar(&body, "body", false, "show the request and its answer, for a project that keeps them") + c.Flags().StringVar(&replay, "replay", "", "send the kept request to this model and show both answers") + return c +} + +func showRequestBody(cmd *cobra.Command, client func() *ctlClient, id string, asJSON bool) error { + var b store.RequestBody + if err := client().get(cmd.Context(), "/api/v1/requests/"+url.PathEscape(id)+"/body", &b); err != nil { + return err + } + if asJSON { + return json.NewEncoder(cmd.OutOrStdout()).Encode(b) + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "request %s at %s, %s to %s (served by %s), status %d\n", b.RequestID, + b.At.UTC().Format("2006-01-02 15:04:05Z"), b.Path, b.Model, b.ServedModel, b.Status) + fmt.Fprintf(out, "kept until %s\n\n", b.ExpiresAt.UTC().Format("2006-01-02 15:04Z")) + fmt.Fprintln(out, "--- sent") + fmt.Fprintln(out, indentJSON(b.Request)) + fmt.Fprintln(out, "--- answered") + fmt.Fprintln(out, or(b.Response, "(no text: an error, or not a completion)")) + if b.Truncated { + fmt.Fprintln(out, "(longer than what is kept; cut here)") + } + return nil +} + +func replayRequest(cmd *cobra.Command, client func() *ctlClient, id, model string, asJSON bool) error { + var v adminapi.ReplayView + body, _ := json.Marshal(map[string]string{"model": model}) + if err := client().post(cmd.Context(), "/api/v1/requests/"+url.PathEscape(id)+"/replay", body, &v); err != nil { + return err + } + if asJSON { + return json.NewEncoder(cmd.OutOrStdout()).Encode(v) + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "--- %s answered (status %d)\n%s\n\n", v.Original.ServedModel, v.Original.Status, + or(v.Original.Response, "(no text)")) + fmt.Fprintf(out, "--- %s answers (status %d, %d ms)\n%s\n", v.Model, v.Status, v.TookMs, or(replyText(v.Body), string(v.Body))) + return nil +} + +// replyText is a completion's text, or "" when it is not one. +func replyText(body json.RawMessage) string { + var c struct { + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + Text string `json:"text"` + } `json:"choices"` + } + if json.Unmarshal(body, &c) != nil || len(c.Choices) == 0 { + return "" + } + return c.Choices[0].Message.Content + c.Choices[0].Text +} + +func indentJSON(s string) string { + var v any + if json.Unmarshal([]byte(s), &v) != nil { + return s + } + b, _ := json.MarshalIndent(v, "", " ") + return string(b) } diff --git a/internal/cli/keys.go b/internal/cli/keys.go index 717079f0..24c3d585 100644 --- a/internal/cli/keys.go +++ b/internal/cli/keys.go @@ -42,6 +42,7 @@ func newTenancyCmd() *cobra.Command { newKeyTTLCmd(open), newLimitsCmd(open), newThreadsCmd(open), + newRequestLogCmd(open), newMemoryCmd(open), newMoveStoreCmd(), ) diff --git a/internal/cli/requestlog.go b/internal/cli/requestlog.go new file mode 100644 index 00000000..fa8d0fb2 --- /dev/null +++ b/internal/cli/requestlog.go @@ -0,0 +1,74 @@ +package cli + +import ( + "fmt" + + "github.com/spf13/cobra" +) + +func newRequestLogCmd(open opener) *cobra.Command { + var ( + project string + hours int + sample float64 + off bool + ) + cmd := &cobra.Command{ + Use: "requests", + Short: "Keep a project's requests and answers, to follow one up or replay it", + Long: "With this on, what each request of the project sent the model — after the\n" + + "guardrails' redaction, with any stored conversation in it — and the answer\n" + + "the client got are kept for --retain-hours. `llmgw ctl request ID --body`\n" + + "shows one, and `--replay MODEL` sends it to another model.\n\n" + + "Off for every project until this is run, like conversations: llmgw\n" + + "otherwise keeps no text. --sample keeps a share of the requests rather\n" + + "than all of them. A project under a guardrails policy with no_store keeps\n" + + "nothing whatever this says.", + Example: " llmgw keys requests --project Acme/ml/train --retain-hours 72\n" + + " llmgw keys requests --project Acme/ml/train --retain-hours 168 --sample 0.05\n" + + " llmgw keys requests --project Acme/ml/train --off", + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if !off && hours <= 0 { + return fmt.Errorf("--retain-hours is required: say how long requests are kept, or pass --off") + } + if sample <= 0 || sample > 1 { + return fmt.Errorf("--sample is a share: above 0, at most 1") + } + s, err := open(cmd.Context()) + if err != nil { + return err + } + defer s.Close() + p, err := resolveProject(cmd.Context(), s, project) + if err != nil { + return err + } + if off { + hours = 0 + } + if err := s.SetRequestLog(cmd.Context(), p.ID, hours, sample); err != nil { + return err + } + if hours == 0 { + fmt.Fprintf(cmd.OutOrStdout(), "%s keeps no requests. Those kept are removed at the next prune.\n", p.Path) + return nil + } + fmt.Fprintf(cmd.OutOrStdout(), "%s keeps %s of its requests and answers for %d hours.\n", + p.Path, share(sample), hours) + return nil + }, + } + cmd.Flags().StringVar(&project, "project", "", "project: id, path or unique name") + cmd.Flags().IntVar(&hours, "retain-hours", 0, "how long a request and its answer are kept") + cmd.Flags().Float64Var(&sample, "sample", 1, "the share of requests kept, above 0 and at most 1") + cmd.Flags().BoolVar(&off, "off", false, "stop keeping requests for this project") + return cmd +} + +func share(f float64) string { + if f >= 1 { + return "every one" + } + return fmt.Sprintf("%g%%", f*100) +} diff --git a/internal/cli/server.go b/internal/cli/server.go index e9aadfea..3555ccbf 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -292,6 +292,7 @@ func newServerCmd() *cobra.Command { adminAPI = api api.SetReleases(cfg.Releases) api.SetNotifier(notifier) + api.SetReplayer(rt.Replay) if st := rt.Store(); st != nil && gate != nil { api.SetGuard(&adminapi.GuardAdmin{Library: cfg.Guard, Store: st, Invalidate: gate.Invalidate}) } diff --git a/internal/cli/tui_people.go b/internal/cli/tui_people.go index d42a981f..66ccb914 100644 --- a/internal/cli/tui_people.go +++ b/internal/cli/tui_people.go @@ -70,6 +70,8 @@ type peopleData struct { limits map[string]store.Limits threads map[string]int // project id → hours kept, conversations notes map[string]int // project id → hours kept, agent notes + // requests is how each project keeps its requests and answers. + requests map[string]store.RequestLog // guardPolicies are server.yaml's guardrail policies, guardDefault those // applied everywhere, and guardAt which are attached where. guardPolicies []string @@ -122,7 +124,7 @@ func (m consoleModel) fetchPeople() tea.Cmd { d.guardDefault = cfg.Guard.Default d.guardAt, _ = s.GuardAttachments(ctx) } - d.threads, d.notes = map[string]int{}, map[string]int{} + d.threads, d.notes, d.requests = map[string]int{}, map[string]int{}, map[string]store.RequestLog{} for _, r := range d.orgs { if r.level != "project" { continue @@ -133,6 +135,9 @@ func (m consoleModel) fetchPeople() tea.Cmd { if t, err := s.MemoryRetention(ctx, r.id); err == nil { d.notes[r.id] = int(t.Hours()) } + if l, err := s.RequestLogOf(ctx, r.id); err == nil { + d.requests[r.id] = l + } } return consolePeopleMsg(d) } @@ -502,7 +507,33 @@ func (m consoleModel) orgMenu(r orgRow) *menuDialog { } items = append(items, retention("Keep conversations", "threads", "conversations", m.people.threads[r.id]), - retention("Keep agent notes", "memory", "notes", m.people.notes[r.id])) + retention("Keep agent notes", "memory", "notes", m.people.notes[r.id]), + menuItem{"Keep requests…", func(m *consoleModel) tea.Cmd { + l := m.people.requests[r.id] + sample := "1" + if l.Sample > 0 { + sample = strconv.FormatFloat(l.Sample, 'f', -1, 64) + } + m.openForm(&formDialog{title: "Keep requests · " + r.path, + note: "Off by default: llmgw keeps no requests unless a project asks for it.\n" + + "Kept, `llmgw ctl request ID --body` shows one and --replay sends it to another model.", + fields: []formField{ + textF("hours", strconv.Itoa(int(l.Retain.Hours())), "how long they are kept; 0 keeps none"), + textF("share", sample, "of requests kept, above 0 and at most 1"), + }, + submit: func(m *consoleModel, v []string) (tea.Cmd, error) { + h, err := strconv.Atoi(v[0]) + if err != nil || h < 0 { + return nil, fmt.Errorf("hours is a whole number") + } + args := []string{"keys", "requests", "--project=" + r.id} + if h == 0 { + return m.runCLI("Keep requests", append(args, "--off")...), nil + } + return m.runCLI("Keep requests", append(args, "--retain-hours="+v[0], "--sample="+v[1])...), nil + }}) + return nil + }}) } items = append(items, menuItem{"Remove…", func(m *consoleModel) tea.Cmd { m.confirmD = &confirmDialog{title: "Remove " + r.path, diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index c880ad2d..920b88b1 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -41,6 +41,8 @@ import ( // API serves the management endpoints. type API struct { + // replay sends a kept request again; nil without a router here. + replay Replayer // notifier is the webhooks `ctl notify test` reaches; nil without any. notifier *notify.Notifier // releases is where release binaries are fetched from, for upgrades. @@ -159,6 +161,8 @@ func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("GET /api/v1/plan", a.planModel) mux.HandleFunc("POST /api/v1/metrics/range", a.metricsBatch) mux.HandleFunc("GET /api/v1/requests/{id}", a.request) + mux.HandleFunc("GET /api/v1/requests/{id}/body", a.requestBody) + mux.HandleFunc("POST /api/v1/requests/{id}/replay", a.replayRequest) mux.HandleFunc("POST /api/v1/models/{name}/retry", a.retryModel) mux.HandleFunc("GET /api/v1/models/{name}/headroom", func(w http.ResponseWriter, r *http.Request) { limit, _ := strconv.Atoi(r.URL.Query().Get("limit")) diff --git a/internal/control/adminapi/requestbody.go b/internal/control/adminapi/requestbody.go new file mode 100644 index 00000000..edfbe2cf --- /dev/null +++ b/internal/control/adminapi/requestbody.go @@ -0,0 +1,86 @@ +package adminapi + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +type bodyFinder interface { + RequestBodyByID(ctx context.Context, id string) (store.RequestBody, error) +} + +// Replayer sends a kept request again to a model: the router's Replay. +type Replayer func(ctx context.Context, path, model string, body []byte) (int, []byte, time.Duration, error) + +// SetReplayer gives the API the router to replay kept requests through. +func (a *API) SetReplayer(f Replayer) { a.replay = f } + +// ReplayView is a kept request sent again, beside what it got the first time. +type ReplayView struct { + Original store.RequestBody `json:"original"` + Model string `json:"model"` + Status int `json:"status"` + Body json.RawMessage `json:"body"` + TookMs int64 `json:"took_ms"` +} + +func (a *API) keptBody(w http.ResponseWriter, r *http.Request) (store.RequestBody, bool) { + f, ok := a.usage.(bodyFinder) + if !ok { + writeError(w, http.StatusNotImplemented, "this control plane keeps no requests") + return store.RequestBody{}, false + } + b, err := f.RequestBodyByID(r.Context(), r.PathValue("id")) + if errors.Is(err, store.ErrNotFound) { + writeError(w, http.StatusNotFound, "no request with that id is kept: its project keeps none "+ + "(llmgw keys requests --project P --retain-hours N), it fell outside the share kept, or it has expired") + return store.RequestBody{}, false + } + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return store.RequestBody{}, false + } + return b, true +} + +func (a *API) requestBody(w http.ResponseWriter, r *http.Request) { + if b, ok := a.keptBody(w, r); ok { + writeJSON(w, http.StatusOK, b) + } +} + +func (a *API) replayRequest(w http.ResponseWriter, r *http.Request) { + if a.replay == nil { + writeError(w, http.StatusNotImplemented, "this control plane does not route requests") + return + } + var req struct { + Model string `json:"model"` + } + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<12)).Decode(&req); err != nil || req.Model == "" { + writeError(w, http.StatusBadRequest, "say which model: {\"model\": \"...\"}") + return + } + b, ok := a.keptBody(w, r) + if !ok { + return + } + status, body, took, err := a.replay(r.Context(), b.Path, req.Model, []byte(b.Request)) + if err != nil { + writeError(w, http.StatusBadGateway, err.Error()) + return + } + v := ReplayView{Original: b, Model: req.Model, Status: status, TookMs: took.Milliseconds()} + if json.Valid(body) { + v.Body = body + } else { + v.Body, _ = json.Marshal(string(body)) + } + writeJSON(w, http.StatusOK, v) +} diff --git a/internal/router/handlers.go b/internal/router/handlers.go index 790a2bd6..e5c32274 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -294,6 +294,9 @@ type parsedRequest struct { // user field. Both are recorded with the usage, never sent on. Tags map[string]string EndUser string + // logRetain is how long the request and its answer are kept, when its + // project keeps them and it fell in the share kept. + logRetain time.Duration // 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 @@ -499,9 +502,13 @@ func (r *Router) handleInference(path string) http.HandlerFunc { return } + r.decideRequestLog(req.Context(), parsed) res := r.dispatch(req.Context(), w, parsed) status := r.finish(w, parsed, res, start) r.observeCanary(parsed, res, status, time.Since(start)) + if parsed.logRetain > 0 { + go r.keepRequest(context.WithoutCancel(req.Context()), parsed, res, status) + } r.observeOutput(parsed, res, res.judged) span.SetAttributes(attribute.Int("http.response.status_code", status), attribute.String("llmgw.served_model", res.model), diff --git a/internal/router/requestlog.go b/internal/router/requestlog.go new file mode 100644 index 00000000..105dda47 --- /dev/null +++ b/internal/router/requestlog.go @@ -0,0 +1,152 @@ +package router + +import ( + "bytes" + "context" + "fmt" + "hash/fnv" + "net/http" + "sync" + "time" + + storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +// Keeping requests and answers, for a project that asks: what was sent to +// the model and what came back, for following one request up and for +// sending it again to another model. + +// requestLogTTL is how long a project's setting is believed before it is +// read again. +const requestLogTTL = 30 * time.Second + +type requestLogs struct { + mu sync.Mutex + by map[string]cachedLog +} + +type cachedLog struct { + l storepkg.RequestLog + at time.Time +} + +// requestLogOf is a project's setting, read at most every requestLogTTL. +func (r *Router) requestLogOf(ctx context.Context, project string) storepkg.RequestLog { + if r.store == nil || project == "" { + return storepkg.RequestLog{} + } + r.reqLogs.mu.Lock() + c, ok := r.reqLogs.by[project] + r.reqLogs.mu.Unlock() + if ok && time.Since(c.at) < requestLogTTL { + return c.l + } + l, err := r.store.RequestLogOf(ctx, project) + if err != nil { + r.log.Warn("could not read whether a project keeps its requests", "project", project, "err", err) + return c.l + } + r.reqLogs.mu.Lock() + if r.reqLogs.by == nil { + r.reqLogs.by = map[string]cachedLog{} + } + r.reqLogs.by[project] = cachedLog{l: l, at: time.Now()} + r.reqLogs.mu.Unlock() + return l +} + +// sampled says whether a request falls in a share of them, by its id, so +// the same id always says the same. +func sampled(id string, share float64) bool { + if share >= 1 { + return true + } + h := fnv.New32a() + h.Write([]byte(id)) + return float64(h.Sum32()%10000) < share*10000 +} + +// decideRequestLog marks a request to be kept, once it is known who sent +// it and that no policy forbids keeping it. +func (r *Router) decideRequestLog(ctx context.Context, req *parsedRequest) { + if req.noStore || req.internal { + return + } + l := r.requestLogOf(ctx, req.ProjectID) + if l.Retain > 0 && sampled(req.RequestID, l.Sample) { + req.logRetain = l.Retain + } +} + +// keepRequest writes a finished request and its answer, for one marked. +func (r *Router) keepRequest(ctx context.Context, req *parsedRequest, res dispatchResult, status int) { + if req.logRetain <= 0 { + return + } + served := res.model + if served == "" { + served = req.Model + } + now := time.Now() + b := storepkg.RequestBody{RequestID: req.RequestID, ProjectID: req.ProjectID, At: now, + ExpiresAt: now.Add(req.logRetain), Path: req.Path, Model: req.Model, ServedModel: served, + Status: status, Request: string(req.UpstreamBody), Response: res.outcome.Reply, + Truncated: res.outcome.ReplyTruncated} + if err := r.store.PutRequestBody(ctx, b); err != nil { + r.log.Warn("could not keep a request", "request_id", req.RequestID, "err", err) + } +} + +// pruneRequestBodies removes kept requests past their retention, hourly. +func (r *Router) pruneRequestBodies(ctx context.Context) { + t := time.NewTicker(time.Hour) + defer t.Stop() + for { + if n, err := r.store.PruneRequestBodies(ctx, time.Now()); err != nil { + if ctx.Err() == nil { + r.log.Warn("could not prune kept requests", "err", err) + } + } else if n > 0 { + r.log.Info("removed kept requests past their retention", "requests", n) + } + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +// Replay sends a kept request again to a model, as the gateway itself: past +// admission, accounting and the guardrails, which judged it the first time. +// For comparing models on real traffic. It answers the status, the body, +// and how long it took. +func (r *Router) Replay(ctx context.Context, path, model string, body []byte) (int, []byte, time.Duration, error) { + if !batchEndpoints[path] { + return 0, nil, 0, fmt.Errorf("a request to %s cannot be replayed", path) + } + body = withoutField(withModel(body, model), "stream") + hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, path, bytes.NewReader(body)) + if err != nil { + return 0, nil, 0, err + } + req, err := r.parse(hreq, path) + if err != nil { + return 0, nil, 0, err + } + req.internal, req.Stream = true, false + req.RequestID = requestID("") + req.Header = http.Header{"X-Request-Id": []string{req.RequestID}} + w := &bufferWriter{header: http.Header{}} + start := time.Now() + res := r.dispatch(ctx, w, req) + took := time.Since(start) + if res.err != nil { + return 0, nil, took, res.err + } + status := w.status + if status == 0 { + status = http.StatusOK + } + return status, w.buf.Bytes(), took, nil +} diff --git a/internal/router/requestlog_test.go b/internal/router/requestlog_test.go new file mode 100644 index 00000000..fa0f7f04 --- /dev/null +++ b/internal/router/requestlog_test.go @@ -0,0 +1,96 @@ +package router + +import ( + "fmt" + "strings" + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/store" +) + +func keptBody(t *testing.T, rt *Router, id string) (store.RequestBody, bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for { + b, err := rt.store.RequestBodyByID(t.Context(), id) + if err == nil { + return b, true + } + if time.Now().After(deadline) { + return store.RequestBody{}, false + } + time.Sleep(20 * time.Millisecond) + } +} + +func TestAProjectThatAsksKeepsItsRequestsAndTheyReplay(t *testing.T) { + e := newFakeEngine(t, "qwen2.5-14b-instruct") + rt, secret := pricedRouter(t, e.srv.URL) + h := rt.Handler() + project := mustProjectOf(t, rt, "key-1") + + send := func(id string) { + rec := post(t, h, map[string]any{"model": "q14", "messages": []any{ + map[string]any{"role": "user", "content": "what is 2+2"}}}, + map[string]string{"Authorization": "Bearer " + secret, "X-Request-Id": id}) + if rec.Code != 200 { + t.Fatalf("%d %s", rec.Code, rec.Body.String()) + } + } + send("before") + if _, ok := keptBody(t, rt, "before"); ok { + t.Fatal("a request was kept by a project that keeps none") + } + + if err := rt.store.SetRequestLog(t.Context(), project, 24, 1); err != nil { + t.Fatal(err) + } + rt.reqLogs.by = nil + send("kept-1") + b, ok := keptBody(t, rt, "kept-1") + if !ok { + t.Fatal("not kept") + } + if !strings.Contains(b.Request, "what is 2+2") || b.Response != "hi" || b.Status != 200 || + b.ServedModel != "qwen2.5-14b-instruct" || b.Path != "/v1/chat/completions" { + t.Errorf("kept %+v", b) + } + if b.ExpiresAt.Sub(b.At) != 24*time.Hour { + t.Errorf("kept for %s", b.ExpiresAt.Sub(b.At)) + } + + status, body, _, err := rt.Replay(t.Context(), b.Path, "q14", []byte(b.Request)) + if err != nil || status != 200 || !strings.Contains(string(body), "hi") { + t.Errorf("replay: %d %s %v", status, body, err) + } + if _, _, _, err := rt.Replay(t.Context(), "/v1/files", "q14", nil); err == nil { + t.Error("a non-inference path was replayed") + } + + // A small share keeps few. + _ = rt.store.SetRequestLog(t.Context(), project, 24, 0.05) + rt.reqLogs.by = nil + for i := range 100 { + send(fmt.Sprintf("s-%d", i)) + } + time.Sleep(200 * time.Millisecond) + n := 0 + for i := range 100 { + if _, err := rt.store.RequestBodyByID(t.Context(), fmt.Sprintf("s-%d", i)); err == nil { + n++ + } + } + if n == 0 || n > 20 { + t.Errorf("kept %d of 100 at 5%%", n) + } + + // Off: what was kept goes at the next prune. + _ = rt.store.SetRequestLog(t.Context(), project, 0, 0) + if _, err := rt.store.PruneRequestBodies(t.Context(), time.Now()); err != nil { + t.Fatal(err) + } + if _, err := rt.store.RequestBodyByID(t.Context(), "kept-1"); err == nil { + t.Error("a project that stopped keeping requests still has them after a prune") + } +} diff --git a/internal/router/router.go b/internal/router/router.go index f6739853..e5e7f03b 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -70,6 +70,8 @@ type Router struct { canaries canaryStats // batchWake has the batch worker look for work at once. batchWake chan struct{} + // reqLogs is each project's keeping of requests, as last read. + reqLogs requestLogs // extraMetrics are collectors other components publish on /metrics. extraMetrics []prometheus.Collector @@ -448,6 +450,7 @@ func (r *Router) Start(ctx context.Context) { } if r.store != nil { go r.runBatches(ctx) + go r.pruneRequestBodies(ctx) go r.refreshTenants(ctx) go r.purgeSessions(ctx) } diff --git a/internal/router/threads.go b/internal/router/threads.go index 04c12a2a..c7d5c374 100644 --- a/internal/router/threads.go +++ b/internal/router/threads.go @@ -221,7 +221,7 @@ const maxStoredReply = 128 << 10 // Zero for every request not continuing a conversation, which is what keeps // the cost of this feature off the path of requests that do not use it. func (p *parsedRequest) captureBytes() int { - if p.ThreadID == "" || p.noStore { + if p.ThreadID == "" && p.logRetain <= 0 || p.noStore { return 0 } return maxStoredReply diff --git a/internal/store/copy.go b/internal/store/copy.go index a7a630c5..535ca893 100644 --- a/internal/store/copy.go +++ b/internal/store/copy.go @@ -61,6 +61,7 @@ var copyOrder = []string{ "files", "batches", "batch_results", + "request_bodies", } // notCopied are tables deliberately left behind. diff --git a/internal/store/migrations/postgres/0025_request_log.sql b/internal/store/migrations/postgres/0025_request_log.sql new file mode 100644 index 00000000..7f38b57f --- /dev/null +++ b/internal/store/migrations/postgres/0025_request_log.sql @@ -0,0 +1,24 @@ +-- Ported from the SQLite migration of the same number. +-- Keeping requests and answers, for a project that asks: how long, and +-- what share of its requests. Off for every project until set, like +-- conversations: a gateway that keeps no text is the default. +ALTER TABLE projects ADD COLUMN request_log_hours BIGINT; +ALTER TABLE projects ADD COLUMN request_log_sample DOUBLE PRECISION; + +-- The request as the model was sent it — after the guardrails' redaction +-- and with any stored conversation spliced in — and the answer as the +-- client was sent it. +CREATE TABLE request_bodies ( + request_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + ts BIGINT NOT NULL, + expires_at BIGINT NOT NULL, + path TEXT NOT NULL, + model TEXT NOT NULL, + served_model TEXT NOT NULL, + status BIGINT NOT NULL, + request TEXT NOT NULL, + response TEXT NOT NULL, + truncated BIGINT NOT NULL DEFAULT 0 +); +CREATE INDEX request_bodies_expiry ON request_bodies (expires_at); diff --git a/internal/store/migrations/sqlite/0025_request_log.sql b/internal/store/migrations/sqlite/0025_request_log.sql new file mode 100644 index 00000000..658ef214 --- /dev/null +++ b/internal/store/migrations/sqlite/0025_request_log.sql @@ -0,0 +1,23 @@ +-- Keeping requests and answers, for a project that asks: how long, and +-- what share of its requests. Off for every project until set, like +-- conversations: a gateway that keeps no text is the default. +ALTER TABLE projects ADD COLUMN request_log_hours INTEGER; +ALTER TABLE projects ADD COLUMN request_log_sample REAL; + +-- The request as the model was sent it — after the guardrails' redaction +-- and with any stored conversation spliced in — and the answer as the +-- client was sent it. +CREATE TABLE request_bodies ( + request_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + ts INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + path TEXT NOT NULL, + model TEXT NOT NULL, + served_model TEXT NOT NULL, + status INTEGER NOT NULL, + request TEXT NOT NULL, + response TEXT NOT NULL, + truncated INTEGER NOT NULL DEFAULT 0 +); +CREATE INDEX request_bodies_expiry ON request_bodies (expires_at); diff --git a/internal/store/postgres_test.go b/internal/store/postgres_test.go index 13e733c0..010df303 100644 --- a/internal/store/postgres_test.go +++ b/internal/store/postgres_test.go @@ -43,7 +43,7 @@ func openPG(t *testing.T) *DB { // every subtest gets a fresh file, and failed only here. `TRUNCATE thread_messages, threads, memory_notes, usage, usage_5m, usage_1h, usage_1d, api_keys, memberships, key_claims, limits, models, sessions, users, projects, teams, - organizations, leadership, control_planes, files, batches, batch_results RESTART IDENTITY CASCADE`) + organizations, leadership, control_planes, files, batches, batch_results, request_bodies RESTART IDENTITY CASCADE`) if err != nil { t.Fatalf("clearing the test database: %v", err) } diff --git a/internal/store/requestlog.go b/internal/store/requestlog.go new file mode 100644 index 00000000..3d19ffd9 --- /dev/null +++ b/internal/store/requestlog.go @@ -0,0 +1,120 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +// RequestLog is how a project keeps its requests and answers: for how long, +// and what share of them. Zero Retain keeps none. +type RequestLog struct { + Retain time.Duration + // Sample is the share kept, above 0 and at most 1. + Sample float64 +} + +// SetRequestLog turns keeping requests and answers on for a project, or off +// with hours 0. A sample outside (0, 1] keeps every request. +func (s *DB) SetRequestLog(ctx context.Context, projectID string, hours int, sample float64) error { + var h, sm any + if hours > 0 { + h = hours + if sample <= 0 || sample > 1 { + sample = 1 + } + sm = sample + } + res, err := s.db.ExecContext(ctx, + `UPDATE projects SET request_log_hours = ?, request_log_sample = ? WHERE id = ?`, h, sm, projectID) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return fmt.Errorf("store: no project %q", projectID) + } + return nil +} + +// RequestLogOf is a project's setting; zero when it keeps none. +func (s *DB) RequestLogOf(ctx context.Context, projectID string) (RequestLog, error) { + var hours sql.NullInt64 + var sample sql.NullFloat64 + err := s.db.QueryRowContext(ctx, + `SELECT request_log_hours, request_log_sample FROM projects WHERE id = ?`, projectID).Scan(&hours, &sample) + if errors.Is(err, sql.ErrNoRows) || err == nil && (!hours.Valid || hours.Int64 <= 0) { + return RequestLog{}, nil + } + if err != nil { + return RequestLog{}, err + } + l := RequestLog{Retain: time.Duration(hours.Int64) * time.Hour, Sample: 1} + if sample.Valid && sample.Float64 > 0 && sample.Float64 <= 1 { + l.Sample = sample.Float64 + } + return l, nil +} + +// RequestBody is one kept request and its answer. +type RequestBody struct { + RequestID string `json:"request_id"` + ProjectID string `json:"project_id"` + At time.Time `json:"at"` + ExpiresAt time.Time `json:"expires_at"` + Path string `json:"path"` + Model string `json:"model"` + ServedModel string `json:"served_model"` + Status int `json:"status"` + Request string `json:"request"` + // Response is the answer's text as the client was sent it; Truncated + // says it was longer than what is kept. + Response string `json:"response"` + Truncated bool `json:"truncated,omitempty"` +} + +// PutRequestBody keeps one. +func (s *DB) PutRequestBody(ctx context.Context, b RequestBody) error { + tr := 0 + if b.Truncated { + tr = 1 + } + _, err := s.db.ExecContext(ctx, `INSERT INTO request_bodies (request_id, project_id, ts, expires_at, path, + model, served_model, status, request, response, truncated) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (request_id) DO NOTHING`, + b.RequestID, b.ProjectID, millis(b.At), millis(b.ExpiresAt), b.Path, b.Model, b.ServedModel, b.Status, + b.Request, b.Response, tr) + return err +} + +// RequestBodyByID finds one, ErrNotFound when none is kept. +func (s *DB) RequestBodyByID(ctx context.Context, id string) (RequestBody, error) { + var b RequestBody + var ts, exp int64 + var tr int + err := s.db.QueryRowContext(ctx, `SELECT request_id, project_id, ts, expires_at, path, model, served_model, + status, request, response, truncated FROM request_bodies WHERE request_id = ?`, id). + Scan(&b.RequestID, &b.ProjectID, &ts, &exp, &b.Path, &b.Model, &b.ServedModel, &b.Status, + &b.Request, &b.Response, &tr) + if errors.Is(err, sql.ErrNoRows) { + return RequestBody{}, ErrNotFound + } + if err != nil { + return RequestBody{}, err + } + b.At, b.ExpiresAt, b.Truncated = time.UnixMilli(ts), time.UnixMilli(exp), tr == 1 + return b, nil +} + +// PruneRequestBodies removes those past their retention, and those of +// projects that keep none any more. +func (s *DB) PruneRequestBodies(ctx context.Context, now time.Time) (int64, error) { + res, err := s.db.ExecContext(ctx, `DELETE FROM request_bodies WHERE expires_at < ? + OR project_id IN (SELECT id FROM projects WHERE request_log_hours IS NULL OR request_log_hours <= 0)`, + millis(now)) + if err != nil { + return 0, err + } + return res.RowsAffected() +}