Requests kept and replayed: keys requests per project with retention and a share, ctl request --body and --replay

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ilker ManapandClaude Opus 5.5 committed 2026-10-04 22:51:21 +02:00
1 parent 841ee21c12
commit 32ff1dbd6f
22 files changed
+757 -10

No files matched your search

+5 -1
View File
@@ -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 <id> --body`; replay against another model.
10. **OpenAI compatibility gaps (S–M each).** Responses: `text.format` →
+9
View File
@@ -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),
+2 -1
View File
@@ -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.
+6 -1
View File
@@ -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,
+23
View File
@@ -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
+85 -3
View File
@@ -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)
}
+1
View File
@@ -42,6 +42,7 @@ func newTenancyCmd() *cobra.Command {
newKeyTTLCmd(open),
newLimitsCmd(open),
newThreadsCmd(open),
newRequestLogCmd(open),
newMemoryCmd(open),
newMoveStoreCmd(),
)
+74
View File
@@ -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)
}
+1
View File
@@ -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})
}
+33 -2
View File
@@ -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,
+4
View File
@@ -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"))
+86
View File
@@ -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)
}
+7
View File
@@ -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),
+152
View File
@@ -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
}
+96
View File
@@ -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")
}
}
+3
View File
@@ -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)
}
+1 -1
View File
@@ -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
+1
View File
@@ -61,6 +61,7 @@ var copyOrder = []string{
"files",
"batches",
"batch_results",
"request_bodies",
}
// notCopied are tables deliberately left behind.
@@ -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);
@@ -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);
+1 -1
View File
@@ -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)
}
+120
View File
@@ -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()
}