Documentation
¶
Index ¶
- Constants
- Variables
- func ContextWithPresentedApiKey(ctx context.Context, key string) context.Context
- func ContextWithRejectionReason(ctx context.Context, slot *rejectionSlot) context.Context
- func ContextWithSessionID(ctx context.Context, sessionID string) context.Context
- func ContextWithUpstreamIndex(ctx context.Context, slot *upstreamIndexSlot) context.Context
- func DefaultProxyTransport() *http.Transport
- func IsLoopbackRemoteAddr(addr string) bool
- func NewAdminLoopbackGuard(next http.Handler, isLoopback func(string) bool) http.Handler
- func NewAnthropicProxyHandler(upstream *url.URL, transport http.RoundTripper) http.Handler
- func NewAuthMiddleware(next http.Handler, allowedKeys map[string]struct{}) http.Handler
- func NewAuthSwapTransport(next http.RoundTripper, token string) http.RoundTripper
- func NewCircuitBreaker(next http.Handler, provider string, failureThreshold int, ...) http.Handler
- func NewColdStartGate(next http.Handler, provider string, budgetTokens int, ...) http.Handler
- func NewColdStartGateWithBackpressure(next http.Handler, provider string, budgetTokens int, ...) http.Handler
- func NewConcurrencyLimiter(next http.Handler, provider string, maxConcurrentRequests int, ...) http.Handler
- func NewDisableTraceHandler() http.Handler
- func NewEnableTraceHandler() http.Handler
- func NewHealthzHandler() http.Handler
- func NewHelloHandler() http.Handler
- func NewLoggingRoundTripper(inner http.RoundTripper, bodySampler liblog.Sampler, ...) http.RoundTripper
- func NewModelRouter(routes []ModelRoute, defaultProviderName string, defaultHandler http.Handler, ...) http.Handler
- func NewModelRouterWithPools(routes []ModelRoute, defaultProviderName string, defaultHandler http.Handler, ...) http.Handler
- func NewNotFoundHandler() http.Handler
- func NewProviderInFlightCollector(dispatched map[string][]func() int, queued map[string][]func() int) prometheus.Collector
- func NewRootLivenessHandler() http.Handler
- func NewSessionMiddleware(next http.Handler) http.Handler
- func NewSetLoglevelHandler() http.Handler
- func NewSetLoglevelHandlerWithRevert(autoRevert time.Duration) http.Handler
- func NewThrottleGate(next http.Handler, provider string, threshold int, maxDelay stdtime.Duration, ...) http.Handler
- func NewTraceMiddleware(next http.Handler, traceDir string, traceState *TraceState, ...) http.Handler
- func NewUpstreamInFlightCollector(hostLimiters map[string]*HostLimiter) prometheus.Collector
- func NewUpstreamPoolHandler(ctx context.Context, members []UpstreamMember) http.Handler
- func PresentedApiKeyFromContext(ctx context.Context) string
- func RedactBearerTokensInBody(b []byte) []byte
- func RedactHeadersForLog(h http.Header) map[string]string
- func RejectionSlotFromContext(ctx context.Context) *rejectionSlot
- func SessionIDFromContext(ctx context.Context) string
- func UpstreamIndexSlotFromContext(ctx context.Context) *upstreamIndexSlot
- type ColdGateMetrics
- type HostLimiter
- type InFlight
- type Metrics
- func (m *Metrics) ObserveAliasResolution(alias, resolved string)
- func (m *Metrics) ObserveCacheTokens(provider, model, direction string, count int)
- func (m *Metrics) ObserveRejection(provider, reason string)
- func (m *Metrics) ObserveRequest(provider, model string, status int, latencySeconds float64, isRouterError bool)
- func (m *Metrics) ObserveTokens(provider, model, direction string, count int)
- func (m *Metrics) PreInitRejection(provider string)
- func (m *Metrics) Register(reg prometheus.Registerer) error
- type ModelPool
- type ModelPoolMember
- type ModelRoute
- type TTFTBackpressure
- type TokenUsage
- type TraceState
- type UpstreamMember
- type WindowEligible
Constants ¶
const ( SetLoglevelDefault = 1 SetLoglevelAutoRevert = 5 * time.Minute )
SetLoglevelDefault is the verbosity we revert to after autoRevertAfter. V(1) keeps the structured per-request line; the operator bumps to 2/3 for alias/route detail or to higher tiers for deep debug.
const BodySampleMaxBytes = 4 << 10 // 4 KB
BodySampleMaxBytes is the maximum number of bytes captured from a request or response body for V(4) body-sample logging. 4 KB is large enough to see the full JSON envelope of any ordinary Anthropic /messages request while staying small enough to avoid log-line bloat for large streaming bodies.
const InFlightPeakWindowSeconds = 60
InFlightPeakWindowSeconds is the width of the sliding window over which ccrouter_inflight_requests_peak reports the highest concurrent request count per provider. A 60 s window at the default 15 s scrape interval means a burst that starts and ends between two scrapes is still visible for up to four scrapes.
const MaxRequestBodyBytes = 32 << 20 // 32 MB
MaxRequestBodyBytes caps inbound /v1/* request bodies at 32 MB to match the Anthropic API ceiling. Long Claude Code sessions (full conversation history + tool definitions + sub-agent results) routinely exceed 1 MB, so a tighter cap surfaces as confusing 413s that read as upstream errors. 32 MB still bounds memory exhaustion from an accidental multi-GB upload via io.ReadAll. On overflow, the wrapped body returns *http.MaxBytesError; the router responds with HTTP 413 Request Entity Too Large + a generic body (no internal state leaked).
const TailBufferBytes = 2 << 20 // 2 MiB
TailBufferBytes caps the number of response-body bytes retained for post-request usage extraction. The bound must hold the FULL response body when the upstream sends `Content-Encoding: gzip` because gzip is not self-synchronizing — decompression requires the start-of-stream bytes, so a mid-stream tail is unrecoverable. Real Anthropic 200 responses have been observed at ~500 KB gzipped through Cloudflare (spec 006 Reproduction). 2 MiB is a frozen constant chosen to cover the observed p99 with headroom; it is NOT a YAML field (spec 006 Non-goals). A full 5 MB streaming response therefore occupies at most TailBufferBytes of additional memory.
const TraceTTLDefault = 5 * time.Minute
TraceTTLDefault is the production TTL window for trace enable. The TTL timer turns tracing off automatically after this duration. The 5-minute constant is a frozen invariant; TRACE_TTL is a test-only override and is never read by production code.
const UnknownModelLabel = "_unknown_"
UnknownModelLabel is the sentinel used in place of an empty model string when the request body has no `model` field (probe traffic, misshapen bodies) or the router-side early-return paths reject before body parse. The leading + trailing underscore avoids collision with real Anthropic + OpenRouter model names, which never start with `_`. This constant is a MODEL-label sentinel; the same string is also used as the provider label value from the three router-side early-return paths in NewModelRouter (see spec 007 Desired Behavior 6) — that reuse is deliberate and scoped, NOT a general fallback for other labels.
Variables ¶
var LatencyBucketsSeconds = []float64{0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}
LatencyBucketsSeconds covers expected LLM end-to-end latency distribution: sub-second for cached / token-counter calls, multi- second for streaming completions, up to ~60s for long contexts. Bucket boundaries match what a Grafana p95/p99 panel on `histogram_quantile(0.95, ...ccrouter_request_duration_seconds...)` can meaningfully resolve.
var QueueWaitBucketsSeconds = []float64{0.1, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30, 45, 60}
QueueWaitBucketsSeconds covers how long a request waits for a concurrency slot before it either acquires one or times out at maxConcurrentWait (30s default). The ladder is denser than LatencyBucketsSeconds above 10s on purpose: a queue-timeout observation lands at exactly maxConcurrentWait, and a coarse top bucket makes histogram_quantile interpolate across the whole band — at a [10,30] boundary a saturated limiter reports a p95 well below the 27s the saturation criterion reads, which would make the criterion unable to distinguish a saturated queue from a lightly loaded one.
Functions ¶
func ContextWithPresentedApiKey ¶ added in v0.28.0
ContextWithPresentedApiKey returns a copy of ctx carrying the x-api-key value the auth middleware accepted. It is also a test seam: routing specs inject a key directly without running the middleware.
func ContextWithRejectionReason ¶ added in v0.54.0
ContextWithRejectionReason returns a copy of ctx carrying the shared rejection-reason slot. The model router injects one slot per request before dispatch.
func ContextWithSessionID ¶ added in v0.34.0
ContextWithSessionID returns a copy of ctx carrying the x-session-id value the session middleware stripped from the request. It is also a test seam: upstream-pool specs inject a session id directly without running the middleware.
func ContextWithUpstreamIndex ¶ added in v0.43.0
ContextWithUpstreamIndex returns a copy of ctx carrying the shared upstream-member-index slot. The model router injects one slot per request before dispatch.
func DefaultProxyTransport ¶ added in v0.3.0
DefaultProxyTransport returns an http.Transport with explicit timeouts suitable for upstream LLM API calls. Generous ResponseHeaderTimeout because long-generation requests (`/compact` on a large session, big code-gen prompts) can delay 60-300s before Anthropic sends the first byte of headers; short Dial because connections are quick HTTPS to api.anthropic.com.
Observed 2026-06-28: previous 60s ResponseHeaderTimeout produced `net/http: timeout awaiting response headers` 502s on `/compact` that took several minutes total. 300s (5 min) is generous enough for the worst observed case while still bounding a genuinely-wedged connection.
func IsLoopbackRemoteAddr ¶ added in v0.21.0
IsLoopbackRemoteAddr reports whether the given remote address is a loopback address. It accepts the address in the format Go's http server produces in http.Request.RemoteAddr: "127.0.0.1:54321" for IPv4 and "[::1]:54321" for IPv6. Both IPv4 loopback (127.0.0.0/8) and IPv6 loopback (::1) count as local; anything that is not host:port is not loopback.
The remote address MUST come from the connection only — never from X-Forwarded-For or any other client-supplied header. There is no trusted proxy in front of this router; honouring a forwarded header would let any remote caller claim to be loopback and bypass both the auth check and the admin guard.
func NewAdminLoopbackGuard ¶ added in v0.23.0
NewAdminLoopbackGuard wraps next so that only requests arriving from a loopback remote address reach the wrapped handler. Any other request is refused with HTTP 403 and never reaches next, so a refused request can never change router state. The guard is unconditional — there is no config knob to disable it — and stateless, so SIGHUP reloads need no plumbing.
isLoopback is injected (production passes IsLoopbackRemoteAddr) so tests can drive the guard with a fake predicate and a counting inner handler. The remote address comes from r.RemoteAddr (the connection) only — never from X-Forwarded-For or any other client-supplied header, since no trusted proxy sits in front of this router.
func NewAnthropicProxyHandler ¶ added in v0.3.0
NewAnthropicProxyHandler returns an HTTP handler that reverse-proxies every incoming request to upstream (typically https://api.anthropic.com).
The Authorization header passes through unchanged — this is what lets Claude Code's subscription OAuth bearer travel through the router to Anthropic without the router ever holding it. No body parsing, no model-based routing: that's task 3. v1 of task 2 = single upstream, verbatim forward.
Upstream errors (connection refused, 5xx before body, etc.) are logged server-side with the full error string for debugging, but the client sees only a generic "502 Bad Gateway / upstream unavailable" — the internal error details (IPs, TLS handshake failures, connection strings) are not leaked.
If transport is nil, DefaultProxyTransport is used.
func NewAuthMiddleware ¶ added in v0.22.0
NewAuthMiddleware wraps next so every non-loopback /v1/* request must present an x-api-key header whose value is in allowedKeys. A missing or non-matching key is rejected with 401 and never reaches next. Loopback requests bypass the key check entirely but still have the header stripped and still record the presented key in context, so an upstream never observes the caller's API key and the router can route by it. If allowedKeys is empty, the wrapper is a no-op and returns next unchanged — the request path is byte-for-byte identical to a release without key routing (feature-off default, AC 1).
func NewAuthSwapTransport ¶ added in v0.4.0
func NewAuthSwapTransport(next http.RoundTripper, token string) http.RoundTripper
NewAuthSwapTransport wraps next so each outbound request has its Authorization header replaced with `Bearer <token>`. Used by the model router to swap the client's subscription OAuth bearer for a per-provider API token (MiniMax, Ollama, vLLM) before forwarding.
If token is empty, the wrapper is a no-op and returns next.
func NewCircuitBreaker ¶ added in v0.53.0
func NewCircuitBreaker( next http.Handler, provider string, failureThreshold int, now func() libtime.DateTime, transitions *prometheus.CounterVec, ) http.Handler
NewCircuitBreaker returns a handler that, when failureThreshold > 0, counts consecutive upstream failures for the wrapped provider and opens the breaker once the count reaches the threshold (spec 020): while open, every request is answered HTTP 429 with the existing static Anthropic-shaped rate_limit_error body and an integer Retry-After header equal to the fixed open period, and the upstream is never invoked. The router never answers a client with a 5xx of its own making. When the open period elapses the breaker admits exactly one probe request while concurrent requests are answered 429; a probe whose response is not a failure closes the breaker and resets the count, and a failing probe reopens it for a further open period. A non-5xx response resets the consecutive-failure count. When failureThreshold <= 0 the wrapper is a no-op and returns next unchanged — the request path is byte-for-byte identical to a release without the breaker (feature-off default).
now is the router's injected clock and falls back to the real clock when nil. transitions, when non-nil, counts state transitions labeled by provider and state, with its three series pre-initialized to 0.
func NewColdStartGate ¶ added in v0.51.0
func NewColdStartGate( next http.Handler, provider string, budgetTokens int, sessionWindow stdtime.Duration, newSessionsPerMinute int, maxWait stdtime.Duration, now func() libtime.DateTime, metrics ColdGateMetrics, ) http.Handler
NewColdStartGate returns a handler that admits a request whose session id was seen inside sessionWindow straight through — no wait, no reservation, no new behaviour for an established session. A request whose session id is absent, unknown, or last seen outside the window is "cold": it is charged a prefill cost estimate derived from its body size and admitted only while the provider's in-flight cold tokens plus its estimate stay within budgetTokens and while the provider's per-minute new-session rate allows it. A cold request that cannot be admitted waits in a bounded queue (at most coldGateQueueCapacity waiters) for up to maxWait; a request that cannot be admitted within that wait, or that arrives when the queue is full, is answered HTTP 429 with the static Anthropic-shaped rate_limit_error body (limiter429Body) and an integer Retry-After header clamped to [1, 60]. No admission outcome produces a 5xx and no request is dropped without an answer. The reservation is released the moment the response stream emits its first content delta — the point at which prefill has finished — and also whenever the wrapped handler returns, so a stream end or an upstream error frees it too; a client that disconnects while waiting never holds a queue slot or a reservation.
When budgetTokens and newSessionsPerMinute are both <= 0 the wrapper is a no-op and returns next unchanged — the request path is byte-for-byte identical to a release without the gate (feature-off default). Either check alone enables the gate. sessionWindow <= 0 resolves to coldGateDefaultSessionWindow; maxWait <= 0 resolves to coldGateMaxWait. now is the router's injected clock and falls back to the real clock when nil.
The gate is per provider: one instance is constructed per provider, immediately after the upstream pool handler and before the throttle gate, so the existing 429 delay gate stays outermost.
func NewColdStartGateWithBackpressure ¶ added in v0.53.0
func NewColdStartGateWithBackpressure( next http.Handler, provider string, budgetTokens int, backpressure *TTFTBackpressure, sessionWindow stdtime.Duration, newSessionsPerMinute int, maxWait stdtime.Duration, now func() libtime.DateTime, metrics ColdGateMetrics, ) http.Handler
NewColdStartGateWithBackpressure is NewColdStartGate with a TTFT backpressure controller attached (spec 020): the gate admits against backpressure.Budget() (the live effective budget) instead of the value fixed at construction, and feeds the controller each forwarded request's response status and each cold request's time-to-first-token. A nil backpressure makes this identical to NewColdStartGate.
func NewConcurrencyLimiter ¶ added in v0.31.0
func NewConcurrencyLimiter( next http.Handler, provider string, maxConcurrentRequests int, maxConcurrentWait time.Duration, waiter libtime.WaiterDuration, now func() libtime.DateTime, queueWait *prometheus.HistogramVec, ) http.Handler
NewConcurrencyLimiter returns a handler that caps how many /v1/* requests reach next at the same time. When maxConcurrentRequests is <= 0 the wrapper is a no-op and returns next unchanged — the request path is byte-for-byte identical to a release without concurrency limiting (feature-off default). Excess requests queue in a buffered-channel semaphore; a request that cannot acquire a slot within maxConcurrentWait is answered HTTP 429 with an Anthropic-shaped rate_limit_error JSON body so the client's backoff retries cleanly, and a client that disconnects while queued never acquires a slot. The slot is held for the full duration of next.ServeHTTP — including streaming SSE responses — and released only when it returns. InFlight reports the semaphore occupancy — the per-upstream in-flight count the upstream pool handler's least-loaded selection reads — and Queued reports how many requests are blocked waiting for one of those slots. Together they split the dispatch path into what is running and what is waiting. maxConcurrentWait must be > 0; the factory resolves the 30s default (spec DB 5) before constructing.
provider labels the queue-wait histogram; queueWait, when non-nil, is observed with how long a request waited for a slot — on the acquire path and on the timeout path, but never for a request that took a free slot on its first attempt, because that request never queued and observing it would fold service-free requests into a distribution meant to measure contention. A nil queueWait disables the observation without changing the request path, so a caller with no metrics wiring (tests) behaves exactly as before.
now is the router's injected clock and falls back to realNow when nil, the same contract NewThrottleGate and NewCircuitBreaker use. It is injected rather than read directly because production code outside main.go must not call time.Now() — a queue wait measured from an uninjected clock cannot be driven from a test. The wait is still an elapsed-time measurement, so the injected getter is read twice and subtracted rather than compared to any absolute instant.
waiter times the queue wait: a queued request is refused once waiter.Wait(ctx, maxConcurrentWait) returns nil. It is required — the factory passes libtime.NewWaiterDuration() — so a test can substitute the counterfeiter WaiterDuration mock and decide exactly when a queued request times out, instead of sleeping through a real timer.
func NewDisableTraceHandler ¶ added in v0.16.0
NewDisableTraceHandler returns a handler that turns per-request trace logging off immediately and cancels any in-flight TTL timer so no late reset can flip tracing back on. Mirrors the /setloglevel pattern.
func NewEnableTraceHandler ¶ added in v0.16.0
NewEnableTraceHandler returns a handler that turns per-request trace logging on for a bounded 5-minute window (TraceTTLDefault). The window auto-disables on expiry; repeated calls reset the window. Mirrors the /setloglevel pattern: operator-local, no auth, short plaintext body.
func NewHealthzHandler ¶
NewHealthzHandler returns a handler that responds 200 with body "OK".
func NewHelloHandler ¶ added in v0.30.2
NewHelloHandler returns a 200 OK handler for `/api/hello`.
Claude Code's HTTP client HEAD-probes `{ANTHROPIC_BASE_URL}/api/hello` as a connectivity check. Without this handler the probe falls through to NewNotFoundHandler and logs a `[404] HEAD /api/hello` line roughly once per second per running session, burying the real unknown-path signals (misconfigured base URL, `/messages` typo without `/v1`) the logger exists to surface.
func NewLoggingRoundTripper ¶ added in v0.10.0
func NewLoggingRoundTripper( inner http.RoundTripper, bodySampler liblog.Sampler, currentDateTime libtime.CurrentDateTimeGetter, ) http.RoundTripper
NewLoggingRoundTripper wraps inner with upstream-call logging at three verbosity tiers:
V(3) [upstream.headers]: emitted before the inner RoundTrip call; dumps the outbound request headers (after the auth-swap transport has applied its Authorization rewrite) as a JSON object with credential-shaped values redacted via RedactHeadersForLog. Useful for confirming exactly what token / headers reached the provider. Enable via `curl http://127.0.0.1:8788/setloglevel/3`.
V(4) [upstream.start] / [upstream.end]: method+path on start; on end, adds TTFB (time-to-first-byte from when inner.RoundTrip was invoked until it returned with response headers) + status code (or error). Useful for debugging slow upstream behavior — distinguishes "Anthropic took 90s to send first byte" (high TTFB) from "body streaming was slow" (low TTFB, high total latency in the surrounding [req] line). Enable via `curl http://127.0.0.1:8788/setloglevel/4`.
V(4) [upstream.req.body] / [upstream.resp.body]: body-sample lines gated by bodySampler (typically SamplerList{NewSampleTime(1s), NewSamplerGlogLevel(5)}). Captures up to BodySampleMaxBytes (4 KB) of the request body before forwarding it, and wraps the response body so the same size prefix is logged on Close. Bearer tokens are redacted via RedactBearerTokensInBody before the line is emitted.
If inner is nil, http.DefaultTransport is used (matches the nil-default pattern in NewAnthropicProxyHandler).
Silent at default V(1)-V(2).
func NewModelRouter ¶ added in v0.4.0
func NewModelRouter( routes []ModelRoute, defaultProviderName string, defaultHandler http.Handler, aliases map[string]string, sampler liblog.Sampler, metrics *Metrics, currentDateTime libtime.CurrentDateTimeGetter, ) http.Handler
NewModelRouter returns an HTTP handler that body-parses each request's JSON `model` field, resolves it through the aliases map (single-hop, case-sensitive exact match), then dispatches to the first matching ModelRoute. Unmatched models (and non-JSON / no-model requests) fall through to defaultHandler (logged as provider=defaultProviderName). The body is fully read and replayed for the downstream handler — fine for /v1/messages JSON payloads (typically <100 KB); not suitable for unbounded upload bodies.
aliases may be nil or empty — both mean "no alias rewriting". On a hit, the body's top-level .model field is re-marshaled to the resolved value before route dispatch, so the upstream sees the full model name.
One structured `[req]` log line per request at V(1):
[req] POST /v1/messages model=m3 alias=MiniMax-M3-highspeed provider=minimax/0 status=200 latency=842ms
A 429 line additionally ends in ` reason=<r>` — the same cause the ccrouter_rejected_total counter records (queue_timeout, host_queue_timeout, throttle, cold_start, circuit_open, or upstream for a 429 no router gate claimed), so a single log line tells a router-side refusal from the upstream's own. Other statuses carry no reason field.
Non-200 responses are ALWAYS logged; 200 responses are gated by the sampler. `log.DefaultSamplerFactory` gives the canonical OR-combo: at most once per 10s, OR unconditionally when glog `-v` ≥ 4. This keeps the steady-state log readable while preserving every error event and giving full visibility once the operator bumps verbosity via `/setloglevel/4`.
At V(2), alias resolution and route match get their own `[alias]` / `[route]` detail lines (independent of the sampler — V(2) detail is already operator-opt-in, additional gating buys nothing).
match route → dispatch → observe metrics → emit logs. Each step's branching is local and reads sequentially; extracting any of it into a helper buys nothing (the prior `logReq` extraction was a naive line-count fix per architecture audit 2026-06-28 — inlined back). If a second log-event shape ever needs the same data, introduce a `requestLogger` struct holding `sampler` + `metrics` then.
func NewModelRouterWithPools ¶ added in v0.38.0
func NewModelRouterWithPools( routes []ModelRoute, defaultProviderName string, defaultHandler http.Handler, aliases map[string]string, modelPools map[string]*ModelPool, sampler liblog.Sampler, metrics *Metrics, currentDateTime libtime.CurrentDateTimeGetter, ) http.Handler
NewModelRouterWithPools is NewModelRouter with an additional model-pool pre-step: a request whose `model` field names a configured pool resolves to one member before alias/key/glob routing, rewrites the body's model field to that member's concrete model, and dispatches through the member's handler — the glob walk is never reached for a pool name. modelPools may be nil or empty — both mean "no pool resolution", identical to NewModelRouter.
func NewNotFoundHandler ¶ added in v0.7.0
NewNotFoundHandler returns a 404 handler that logs the unknown path before responding. Registered at `/` in the factory's mux so it catches everything not matched by a more specific route (`/v1/`, `/healthz`, `/readiness`, `/metrics`, `/setloglevel/`, `/gc`).
Logged at glog V(1) — same level as `[req]` — so unknown-path probes surface in the operator's default log alongside real traffic. Useful for catching misconfigured clients (wrong base URL, typo in `/v1/messages`) and any probing of the listener.
func NewProviderInFlightCollector ¶ added in v0.52.0
func NewProviderInFlightCollector( dispatched map[string][]func() int, queued map[string][]func() int, ) prometheus.Collector
NewProviderInFlightCollector returns a Prometheus collector exporting ccrouter_provider_dispatched{provider} and ccrouter_provider_queued{provider}: one pair of gauge series per provider, valued at scrape time by summing that provider's pool members' limiter counts. dispatched sums each member concurrencyLimiter's InFlight() — the slots it currently holds — and queued sums their Queued() — the requests blocked waiting for one. The two are disjoint by construction: a request holding a slot is dispatched, a request that found every slot held is queued, and no request is both.
A provider's members are summed rather than sampled because a provider may have several simultaneously-eligible members (a pool with parallel members, or a model pool), so one limiter's occupancy would under-report the provider's real load. The provider set is snapshotted at construction (sorted, for deterministic output) and never mutated.
The collector reads only channel lengths and atomic loads, so Collect takes no lock and cannot block the request path. It is registered once per handler tree, on the same registerer as the tree's other ccrouter_* series.
func NewRootLivenessHandler ¶ added in v0.13.0
NewRootLivenessHandler returns a 200 OK handler for `HEAD /`.
Claude Code's HTTP client probes the base URL with `HEAD /` before dispatching its first /v1/messages request on a fresh connection. Without this handler the probe falls through to NewNotFoundHandler and logs a `[404] HEAD /` line ahead of every real request, which is noisy and obscures the actual `[req]` traffic.
func NewSessionMiddleware ¶ added in v0.34.0
NewSessionMiddleware wraps next so every /v1/* request's x-session-id header is stripped before it reaches next and — when present — carried on the request context for the upstream pool handler to pin on. The header is read ONLY as hash input for pinning: never for auth, never logged. An absent or empty header leaves the context value unset (SessionIDFromContext returns ""), which dispatches the request keyless.
When x-session-id is absent, the id Claude Code sends on every request (x-claude-code-session-id) is used instead, so Claude Code sessions are recognised as warm by the cold-start gate. That header is left on the request: it is Claude Code's own, not router-private.
func NewSetLoglevelHandler ¶
NewSetLoglevelHandler returns a handler that flips glog's -v verbosity at runtime. Convenience wrapper for the production default (SetLoglevelAutoRevert = 5 min); tests use NewSetLoglevelHandlerWithRevert with a short window.
func NewSetLoglevelHandlerWithRevert ¶ added in v0.6.0
NewSetLoglevelHandlerWithRevert returns a handler that flips glog's -v verbosity at runtime. URL shape is `/setloglevel/<level>`; the integer suffix is parsed and passed to `log.LogLevelSetter.Set`, which auto- reverts to SetLoglevelDefault after autoRevert so a forgotten bump can't leave the router in verbose mode indefinitely.
The LogLevelSetter is created once at handler construction (single instance, shared across requests); `Set()` itself spawns the auto- revert goroutine per call — that's upstream bborbe/log behavior, and `resetLogLevel` is idempotent so overlapping timers are harmless.
Level validation: negative values are rejected with 400; glog itself accepts any int32, but a negative verbosity is meaningless and almost always a typo (`-1` instead of `1`).
Example:
$ curl http://127.0.0.1:8788/setloglevel/3 set loglevel to 3 completed
Stdlib-mux compatible: parses the level from URL.Path directly rather than relying on gorilla/mux path vars.
func NewThrottleGate ¶ added in v0.45.0
func NewThrottleGate( next http.Handler, provider string, threshold int, maxDelay stdtime.Duration, now func() libtime.DateTime, throttled *prometheus.CounterVec, ) http.Handler
NewThrottleGate returns a handler that, when threshold is > 0, delays /v1/* requests destined for the wrapped provider once a windowed count of upstream 429 responses reaches the threshold (spec 018): while throttled, each request waits the current pacing delay before forwarding. The delay follows AIMD — it starts at 1s (or maxDelay, whichever is smaller), doubles on each observed 429, and is capped at maxDelay; each clean 60s window (no 429) halves it, and below the 1s recovery floor the provider exits throttle and requests forward undelayed. The 429'd request is never retried — the response passes through unchanged and the status is observed only to adjust future pacing. When threshold is <= 0 the wrapper is a no-op and returns next unchanged — the request path is byte-for-byte identical to a release without the gate (feature-off default).
The pacing queue is bounded: while throttled, at most throttleMaxPacedRequests requests wait their pacing turn; a request that cannot acquire a pacing slot within maxDelay is answered HTTP 429 with the Anthropic-shaped rate_limit_error body (limiter429Body), and a client that disconnects while waiting neither holds a pacing slot nor is forwarded.
maxDelay must be > 0; the factory resolves the 30s default (spec DB 4) before constructing. throttled is the ccrouter_throttled_total CounterVec labeled by provider, incremented once per paced (delayed) request; it must be non-nil. now must be non-nil (the router's injected clock; the factory passes o.currentDateTime.Now) and falls back to the real clock when nil.
func NewTraceMiddleware ¶ added in v0.14.0
func NewTraceMiddleware( next http.Handler, traceDir string, traceState *TraceState, configAlwaysOn bool, ) http.Handler
NewTraceMiddleware wraps next in a handler that, for each request, captures the full request (method, path, headers, body) and full response (status, headers, body) and writes one JSON file to traceDir — BUT only when trace is enabled. Trace is enabled when either the injected traceState flag is true (toggled via /enabletrace, bounded by a 5-min TTL) OR configAlwaysOn is true (the legacy trace: config flag, deprecated). Authorization and x-api-key request headers are redacted to "***" (case-insensitive); all other headers and bodies are logged verbatim. Trace file writes are best-effort: a write failure logs glog.Warningf and the request still succeeds. The trace directory is created on demand (MkdirAll, 0o700).
func NewUpstreamInFlightCollector ¶ added in v0.49.0
func NewUpstreamInFlightCollector(hostLimiters map[string]*HostLimiter) prometheus.Collector
NewUpstreamInFlightCollector returns a Prometheus collector exporting ccrouter_upstream_inflight{host} (spec 019): one gauge series per entry of hostLimiters, valued by that HostLimiter's InFlight() read at scrape time — the shared semaphore occupancy for a capped host, the live in-flight count for an uncapped one. hostLimiters is the factory's one-limiter-per-host map; the collector snapshots its keys at construction (sorted, for deterministic output) and never mutates it. The collector is registered once per handler tree, on the same registerer as the tree's other ccrouter_* series.
func NewUpstreamPoolHandler ¶ added in v0.34.0
func NewUpstreamPoolHandler(ctx context.Context, members []UpstreamMember) http.Handler
NewUpstreamPoolHandler returns a handler that dispatches each request to exactly one of members: a request carrying a non-empty session id (from the session middleware's context) is pinned to the same member every time via a weighted ring hash of the session id — deterministic and stateless, recomputable from the id, no session→member map — so the session's upstream prompt cache stays warm on that server. A request with an empty session id is sent to the least-loaded member (fewest in-flight requests by the per-upstream semaphore), with round-robin tie-breaking among equally-loaded members so keyless floods spread instead of stacking on the first-declared member. A member whose time window does not contain "now", or whose weekday is not in its days set, is excluded from both selection paths — the ring and the least-loaded scan are computed over the eligible subset per request, so a member whose window closes mid-session stops receiving that session's requests immediately (spec 014 / 017). Each chosen dispatch emits a [route] session=<id> upstream=<url> glog V(2) detail line.
func PresentedApiKeyFromContext ¶ added in v0.28.0
PresentedApiKeyFromContext returns the x-api-key value stored by ContextWithPresentedApiKey, or "" when absent.
func RedactBearerTokensInBody ¶ added in v0.11.0
RedactBearerTokensInBody returns a copy of b with every "Bearer <token>" substring replaced by "Bearer <redacted>". The replacement is case-insensitive on the "Bearer" keyword. Returns b unchanged (same slice, no allocation) when no Bearer token is found.
func RedactHeadersForLog ¶ added in v0.10.0
RedactHeadersForLog returns a flat map of header name → value suitable for JSON marshaling. Multi-value headers are joined with ", ". Credential-shaped headers (see isCredentialHeader) have their joined value replaced with "<redacted len=N>" where N is the byte-length of the original joined string.
This helper is designed for V(3) upstream-header logging so operators can see exactly what went on the wire without leaking tokens into log files.
func RejectionSlotFromContext ¶ added in v0.54.0
RejectionSlotFromContext returns the shared slot carried by ctx, or nil when absent (a dispatch path outside the model router, such as a gate exercised directly in a test).
func SessionIDFromContext ¶ added in v0.34.0
SessionIDFromContext returns the x-session-id value stored by ContextWithSessionID, or "" when absent (the empty session id state that dispatches a request keyless).
func UpstreamIndexSlotFromContext ¶ added in v0.43.0
UpstreamIndexSlotFromContext returns the shared slot carried by ctx, or nil when absent (a dispatch path without a wired pool handler).
Types ¶
type ColdGateMetrics ¶ added in v0.51.0
type ColdGateMetrics struct {
Delayed *prometheus.CounterVec
Refused *prometheus.CounterVec
TokensInFlight *prometheus.GaugeVec
TTFT *prometheus.HistogramVec
}
ColdGateMetrics groups the four additive collectors the cold-start admission gate emits, each labeled by provider. A nil field disables that observation; the gate never dereferences a nil field.
type HostLimiter ¶ added in v0.49.0
type HostLimiter struct {
// contains filtered or unexported fields
}
HostLimiter is the one shared concurrency budget for an upstream host (spec 019). The factory builds exactly one per distinct host and wraps every member handler resolving to that host via Wrap, so all of them draw from the same budget. A capped HostLimiter (maxConcurrentRequests > 0) shares one buffered-channel semaphore across every wrapped handler and reuses the concurrency limiter's queue/wait/429 semantics unchanged: excess requests queue for up to maxConcurrentWait, a request still waiting is answered HTTP 429 with the static Anthropic-shaped rate_limit_error body (limiter429Body — no host name, queue depth, or provider name), and a client that disconnects while queued never acquires a slot. The slot is held for the full request, including streaming SSE responses. An unlimited HostLimiter (maxConcurrentRequests <= 0) never queues and never answers 429; its wrapped handlers only count the requests in flight, so InFlight is meaningful for every host.
func NewHostLimiter ¶ added in v0.49.0
func NewHostLimiter( maxConcurrentRequests int, maxConcurrentWait time.Duration, waiter libtime.WaiterDuration, ) *HostLimiter
NewHostLimiter returns the shared budget for one upstream host. maxConcurrentRequests <= 0 means unlimited. maxConcurrentWait must be > 0 on a capped limiter; the factory resolves the 30s default before constructing. waiter times the host-slot queue wait exactly as it does for NewConcurrencyLimiter; it is required.
func (*HostLimiter) InFlight ¶ added in v0.49.0
func (h *HostLimiter) InFlight() int
InFlight returns the host's current in-flight request count: the shared semaphore's occupancy when capped, the live in-flight count when unlimited.
func (*HostLimiter) Wrap ¶ added in v0.49.0
func (h *HostLimiter) Wrap( next http.Handler, provider string, queueWait *prometheus.HistogramVec, ) http.Handler
Wrap returns next guarded by this host's shared budget. Every handler returned by Wrap on the same HostLimiter shares one budget.
The returned limiter reuses the concurrencyLimiter type but is built as a raw literal rather than through NewConcurrencyLimiter. Wrap is called once per upstream member, so each returned wrapper carries its own provider label while the semaphore it draws from is shared across every provider resolving to that host. queueWait is the same ccrouter_queue_wait_seconds histogram the member limiter observes, so a request refused after waiting for a host slot is attributed to the provider it routed to — the same label its ccrouter_rejected_total{reason="host_queue_timeout"} series carries. A host-slot timeout records host_queue_timeout rather than the member limiter's queue_timeout, so a saturated host budget is distinguishable from a saturated member cap. A nil queueWait disables the observation without changing the request path. now is set explicitly — the type's clock is read on the waiting path, and a literal that omitted it would nil-panic on the first host-slot timeout.
type InFlight ¶ added in v0.49.0
type InFlight struct {
// contains filtered or unexported fields
}
InFlight is a prometheus.Collector that tracks, per provider, how many requests have entered the provider's dispatch path and not yet returned. The count is taken immediately before the provider handler runs, so it includes requests queued for a concurrency slot and requests held by the 429 pacing gate — it is NOT the number of requests with an open upstream socket. ccrouter_provider_dispatched{provider} reports the dispatched half of this count and ccrouter_provider_queued{provider} the queued half. It exports two gauges:
- ccrouter_inflight_requests{provider} — the current count. Incremented by Start, decremented by the end function Start returns.
- ccrouter_inflight_requests_peak{provider} — the maximum current count observed within the last InFlightPeakWindowSeconds, computed at collect time from per-second maxima. It is never reset on read (several scrapers or pushers see the same value) and is always at least the current count, so a request still open after the window has passed keeps peak >= current.
A provider appears in the output only after its first Start call — there is no pre-initialization; the provider set is bounded by the YAML config.
InFlight is safe for concurrent use: Start and the returned end function may be called from many goroutines at once.
func NewInFlight ¶ added in v0.49.0
func NewInFlight(currentDateTime libtime.CurrentDateTimeGetter) *InFlight
NewInFlight constructs an InFlight collector that reads the wall clock from currentDateTime. It does NOT register itself; register the returned collector on the registry /metrics scrapes (Metrics.Register does this).
func (*InFlight) Collect ¶ added in v0.49.0
func (i *InFlight) Collect(ch chan<- prometheus.Metric)
Collect implements prometheus.Collector. The peak is derived here (not stored as a reset-on-read value) so repeated collections return the same result until the window advances.
func (*InFlight) Describe ¶ added in v0.49.0
func (i *InFlight) Describe(ch chan<- *prometheus.Desc)
Describe implements prometheus.Collector.
func (*InFlight) Start ¶ added in v0.49.0
Start records that a request has been dispatched to provider and returns an end function that records its return. The end function is idempotent — calling it more than once decrements the count exactly once — and never lets the count drop below zero. It must be deferred so the decrement runs on every exit from the dispatch, including a panic in the target handler.
type Metrics ¶ added in v0.9.0
type Metrics struct {
RequestsTotal *prometheus.CounterVec
RequestDuration *prometheus.HistogramVec
AliasResolutions *prometheus.CounterVec
TokensTotal *prometheus.CounterVec
CacheTokensTotal *prometheus.CounterVec
ThrottledTotal *prometheus.CounterVec
// The four cold-start admission collectors (spec 019) are additive:
// they observe the per-provider cold-start gate and never change the
// existing request/latency/token series or the status_class enum.
ColdAdmissionDelayedTotal *prometheus.CounterVec
ColdAdmissionRefusedTotal *prometheus.CounterVec
ColdTokensInFlight *prometheus.GaugeVec
ColdTTFTSeconds *prometheus.HistogramVec
// The two spec-020 collectors are additive: the gauge carries the live
// effective cold-prefill budget the backpressure loop adjusts, and the
// counter carries the circuit breaker's state transitions. Neither
// changes the existing series or the status_class enum.
ColdPrefillBudgetTokens *prometheus.GaugeVec // ccrouter_cold_prefill_budget_tokens{provider}
BreakerStateTransitionsTotal *prometheus.CounterVec // ccrouter_breaker_state_transitions_total{provider,state}
// The two rejection collectors are additive: the counter carries why a
// request was rejected, and the histogram carries how long a request
// waited for a concurrency slot before it was admitted or timed out.
// Neither changes the existing series or the status_class enum —
// ccrouter_requests_total{status_class="4xx_rate_limited"} still counts
// every 429, and these split that one value by cause.
RejectedTotal *prometheus.CounterVec // ccrouter_rejected_total{provider,reason}
QueueWaitSeconds *prometheus.HistogramVec // ccrouter_queue_wait_seconds{provider}
InFlight *InFlight
}
Metrics groups the Prometheus collectors emitted from the model router: one CounterVec for request totals labeled by provider + model + status_class (`2xx`/`3xx`/`4xx_auth`/`4xx_rate_limited`/ `4xx_bad_request`/`5xx_upstream`/`5xx_router`), one HistogramVec for request latency labeled by provider + model with LLM-shaped buckets, one CounterVec for alias resolutions (operator-side observability for `/model qwen`-style short names), one CounterVec for token counts labeled by provider + model + direction (`input`/`output`), one CounterVec for requests delayed (paced) by the adaptive 429 delay gate, labeled by provider (spec 018), one InFlight collector exporting the current + 60 s-peak in-flight request count per provider, four further additive collectors observing the per-provider cold-start admission gate (spec 019): delayed and refused cold-request counters, an in-flight cold-token gauge, and a cold time-to-first-token histogram, and two more additive collectors for the spec-020 behaviours: the live effective cold-prefill budget gauge and the circuit-breaker state-transition counter.
Cardinality budget: 5 providers × ~15 active models × 7 status_classes = 525 series ceiling for the requests counter (~450 in practice because some tuples never fire); 5 × 15 × 2 = 150 series for the tokens counter; histogram adds 5×15×len(buckets) = 750 series; aliases counter bounded by the YAML config (≤10); the in-flight gauges add 2 series per provider (10). Total ~1.5k series — fine for a local Prometheus scrape.
func NewMetrics ¶ added in v0.9.0
func NewMetrics(aliases map[string]string, currentDateTime libtime.CurrentDateTimeGetter) *Metrics
NewMetrics constructs the collectors but does NOT register them. Call Register on a *prometheus.Registry to expose them; that split lets tests verify behavior against a fresh registry per spec without colliding on the global default registry.
aliases is used to pre-initialize the alias_resolutions counter so that `rate(...) > X` alerts evaluate to 0 (not no-data) for aliases that haven't been hit yet. A nil aliases map is safe — ranging a nil map yields zero iterations. TokensTotal is NOT pre-initialized: there is no closed set of (provider, model, direction) tuples known at boot; cardinality is bounded by real traffic.
currentDateTime is the clock the InFlight collector uses to bucket its 60 s peak window; the caller supplies it (never a fresh libtime.NewCurrentDateTime() built here) so the peak window and the router's request timing share one clock.
func (*Metrics) ObserveAliasResolution ¶ added in v0.9.0
ObserveAliasResolution increments the alias counter on each hit; labels are bounded by the YAML config's `aliases:` map size.
func (*Metrics) ObserveCacheTokens ¶ added in v0.46.0
ObserveCacheTokens increments the ccrouter_cache_tokens_total counter by count for the given (provider, model, direction) tuple. Drop rules mirror ObserveTokens:
- count <= 0 (zero-drop rule)
- direction is not "read" or "creation" (bounded-enum rule)
Kept as a separate observer (not an extension of ObserveTokens's input|output enum) so the existing ccrouter_tokens_total contract is unchanged — cache is a distinct metric, never merged into input.
func (*Metrics) ObserveRejection ¶ added in v0.54.0
ObserveRejection increments ccrouter_rejected_total for a 429 the router answered, labeled by the cause the refusing gate recorded. It is called once per 429 from the router's single post-dispatch observation point — the gates record their reason on the request's shared rejection slot instead of incrementing the counter themselves, because the gates are nested (a host limiter outside a member limiter, a throttle gate outside a cold-start gate) and a per-gate increment would double-count a request an outer gate refused after an inner one had already recorded.
reason must be one of the rejectionReason* constants; the caller resolves an unclaimed 429 to rejectionReasonUpstream. No bounded-enum guard is applied here because every caller passes a constant, and a guard would silently drop a future reason rather than fail loudly at the call site.
func (*Metrics) ObserveRequest ¶ added in v0.9.0
func (m *Metrics) ObserveRequest( provider, model string, status int, latencySeconds float64, isRouterError bool, )
ObserveRequest is the call-site shorthand used by NewModelRouter after each /v1/* dispatch: increments the request counter with the 7-value status_class enum (see statusClass) and observes the latency on the histogram. isRouterError distinguishes 5xx router-side rejections (body-too-large, alias-rewrite-fail) from 5xx upstream faults — see spec 007 Desired Behavior 2 and 6. Callers on the happy path pass isRouterError=false.
func (*Metrics) ObserveTokens ¶ added in v0.18.0
ObserveTokens increments the ccrouter_tokens_total counter by count for the given (provider, model, direction) tuple. Drop rules — the call is a no-op (no series created) when:
- count <= 0 (zero-drop rule; zero is not a data point, negative is a schema violation)
- direction is not "input" or "output" (bounded-enum rule; keeps cardinality contract)
These drops are silent: token counting is best-effort observability, never on the request-serving critical path. See spec 007 Failure Modes for the sentinel/negative/zero rows this method absorbs.
func (*Metrics) PreInitRejection ¶ added in v0.54.0
PreInitRejection creates the rejection series for provider at zero: every (provider, reason) counter and the provider's queue-wait histogram. Call it once per configured provider at boot, before the router serves.
The counter pre-initialization is load-bearing, not cosmetic: `reason` is a bounded enum and a reason that has not fired yet has no series, so `rate(ccrouter_rejected_total{reason="cold_start"}[5m])` returns no data rather than 0 and an alert on it never evaluates. The histogram is seeded for the same reason — an unobserved HistogramVec exports nothing, so a panel for a provider that has never queued would read "no data" instead of an empty-but-present distribution.
func (*Metrics) Register ¶ added in v0.9.0
func (m *Metrics) Register(reg prometheus.Registerer) error
Register registers all collectors against reg. Pass a fresh registry in tests; pass the registry the /metrics endpoint scrapes in production. Returns the first registration error (if any) so caller can decide whether to abort startup.
type ModelPool ¶ added in v0.38.0
type ModelPool struct {
// contains filtered or unexported fields
}
ModelPool is a runtime model pool: an ordered list of members. The only mutable state is the round-robin tie-break counter for idless dispatch; session pinning itself is stateless. A member whose provider pool has no eligible upstream member right now (spec 014) is excluded from pinning, least-loaded, and overflow selection — the ring and the load scans are recomputed over the eligible subset per request.
func NewModelPool ¶ added in v0.38.0
func NewModelPool(ctx context.Context, members []ModelPoolMember) *ModelPool
NewModelPool returns a pool over members. Config validation (spec 013 prompt 1) guarantees members is non-empty and every weight is positive; like the upstream pool handler (spec 012), no defensive validation is repeated here. The concrete *ModelPool return is deliberate: selection is pure logic over config-validated state, exercised through the real type in handler tests — no mock seam is needed, so no ModelPooler interface is introduced.
func (*ModelPool) Resolve ¶ added in v0.38.0
func (p *ModelPool) Resolve(ctx context.Context, sessionID string) ModelPoolMember
Resolve selects the member that serves one request for this pool. A request carrying a non-empty session id (from the session middleware's context) is pinned to the same member every time via a weighted ring hash of the id — deterministic and stateless, recomputable from the id, no session->member map — so the session's prompt cache stays warm on that member. A request with an empty session id goes to the least-loaded member, with round-robin tie-breaking among equally-loaded members so an idless burst spreads instead of stacking on the first member. When the pinned member's provider is saturated (Saturated returns true) and the member declares Overflow, the request falls over to the least-loaded sibling member (declaration-order tie-break) so availability wins over cache warmth; otherwise the pinned member is served and its provider's own concurrency semantics apply (spec 013 DB 4/5). A member whose provider pool has no eligible upstream member is skipped by pinning, least-loaded, and overflow (spec 014).
type ModelPoolMember ¶ added in v0.38.0
type ModelPoolMember struct {
Provider string
Model string
Weight int
Overflow bool
Handler http.Handler
InFlight func() int
Saturated func() bool
}
ModelPoolMember is one candidate of a model pool at runtime: the provider to route through, the fixed concrete model string that provider sees, the weight for session-pinned member selection, and whether the member may overflow to a sibling when its provider is saturated. Handler is the member provider's request handler (its upstream pool handler). InFlight reports the provider's current in-flight load; may be nil, meaning "always 0". Saturated reports whether the provider is at capacity; may be nil, meaning "never saturated". The factory fills all of these from config (spec 013).
type ModelRoute ¶ added in v0.4.0
type ModelRoute struct {
Pattern string
ProviderName string
Handler http.Handler
// RequiresLeadingSystem carries this route's provider's
// `requiresLeadingSystem` config globs. When the resolved model
// name matches one of them, the router lifts non-leading
// system-role messages into the top-level system block before
// dispatching. Nil or empty (the default for every route built
// from a config that omits the field) means: never transform.
RequiresLeadingSystem []string
// AllowedApiKeys carries this route's provider's `allowedApiKeys`
// config list. When the request's presented x-api-key (from the auth
// middleware's context) is in one of these lists, the router dispatches
// to that provider directly — the key is an explicit override that wins
// over model-glob matching. Because Config.Validate rejects a key
// claimed by more than one provider, at most one route ever matches a
// presented key. Nil or empty (the default for every route built from a
// config that omits the field) means: this provider claims no keys, so
// it is only reachable via glob routing.
AllowedApiKeys []string `display:"length"`
}
ModelRoute pairs a glob pattern (filepath.Match syntax) with the provider name + handler to invoke when an incoming request's `model` field matches. ProviderName is what appears in the structured log (`provider=minimax`) and is the same key as in the YAML config's `providers:` map.
type TTFTBackpressure ¶ added in v0.53.0
type TTFTBackpressure struct {
// contains filtered or unexported fields
}
TTFTBackpressure is the per-provider backpressure controller (spec 020). It watches the provider's own measured cold time-to-first-token over fixed 60-second windows, together with the upstream failures observed in the same window, and derives the live effective cold-prefill budget the cold-start gate admits against.
At most one adjustment happens per window, evaluated lazily when the first observation of a new window arrives:
- a window in which any upstream failure was observed, or whose median cold TTFT rose above the configured threshold, halves the budget — never below ceiling/backpressureFloorDivisor (at least 1);
- a failure-free window whose median cold TTFT sits below the fixed backpressureCleanMedianSeconds grows the budget by backpressureGrowthStep — never above the configured ceiling;
- a window whose median sits between the clean line and the threshold, with no failure, holds the budget unchanged, so the loop does not oscillate around the threshold.
The cold-start gate reads Budget() on every admission, so a halving takes effect on the next cold request rather than only after a restart. Each change of the effective budget emits exactly one INFO line and publishes the optional gauge.
A nil *TTFTBackpressure is valid and inert: every method is nil-safe, so a disabled provider needs no branch at the call site. All mutable state — the budget, the window samples, and the failure flag — lives under mu and is mutated from concurrent request goroutines.
func NewTTFTBackpressure ¶ added in v0.53.0
func NewTTFTBackpressure( provider string, ceilingTokens int, thresholdSeconds int, now func() libtime.DateTime, budgetGauge *prometheus.GaugeVec, ) *TTFTBackpressure
NewTTFTBackpressure returns the per-provider backpressure controller for a provider whose cold-prefill budget is ceilingTokens and whose cold-TTFT threshold is thresholdSeconds. It returns nil — an inert controller — when thresholdSeconds <= 0 or ceilingTokens <= 0, so the loop has an effect only while the cold-start gate is enabled and a threshold is configured.
now is the router's injected clock and falls back to the real clock when nil. The budget starts at the configured ceiling and is published to budgetGauge (when non-nil) immediately, so the gauge reads the configured ceiling at construction. The floor is ceilingTokens/16, never below 1 — a zero budget would silently disable the gate's admission check.
func (*TTFTBackpressure) Budget ¶ added in v0.53.0
func (b *TTFTBackpressure) Budget() int
Budget returns the live effective cold-prefill budget this provider admits against — the configured ceiling until the first adjustment, then the halved or grown value. A nil controller returns 0.
func (*TTFTBackpressure) ObserveColdTTFT ¶ added in v0.53.0
func (b *TTFTBackpressure) ObserveColdTTFT(seconds float64)
ObserveColdTTFT records one cold request's measured time-to-first-token in seconds into the current window. A nil controller is inert.
func (*TTFTBackpressure) ObserveStatus ¶ added in v0.53.0
func (b *TTFTBackpressure) ObserveStatus(status int)
ObserveStatus records one upstream response status into the current window. A status of 500 or above marks the window failed; anything below 500 does not, so a client disconnect — which never arrives as a 5xx — is never counted. A nil controller is inert.
type TokenUsage ¶ added in v0.17.0
TokenUsage holds the input/output token counts extracted from an upstream response body. When extraction fails or no usage is present, Input and Output are the empty string and the caller logs the sentinel "-" for each (see logLineValue below). The empty string is the "no data" signal — a real zero-token count from the upstream is reported as "0" (the extractor reports what it parsed).
func ExtractUsage ¶ added in v0.17.0
func ExtractUsage(tail []byte, contentType, contentEncoding string) (usage TokenUsage)
extractUsage pulls input/output token counts out of a response-body tail buffer. SSE responses (Content-Type: text/event-stream OR bytes containing "event: message_") are scanned for the LAST `event: message_start` event (which carries input_tokens) and the LAST `event: message_delta` event (which carries output_tokens) and the two are combined into a single TokenUsage. Anthropic splits the two counts across these two events; the terminal message_delta is always the last chunk of a stream, so it lives in the tail. JSON responses are parsed for a top-level `usage` object.
SSE detection is by Content-Type OR content scan: the primary signal is strings.Contains(contentType, "text/event-stream"), but reverse-proxied Anthropic responses have been observed with an empty or wrong Content-Type on the sniffed rec.Header() (see spec 005). The fallback bytes.Contains(tail, "event: message_") is cheap (single scan over ≤ 64 KB) and specific enough to Anthropic's protocol to keep the false-positive risk low. JSON parsing is the fallback for any other content type.
Extraction is best-effort: truncated tails, malformed JSON/SSE, missing usage, or zero-token usage all yield the noUsage sentinel ("-" / "-") and never an error — the caller's [req] log line must never be aborted by a parse failure.
Presence detection: the extractor first unmarshals the `usage` JSON object into json.RawMessage; if that RawMessage is nil or "null", the field is treated as absent and noUsage is returned. Otherwise the inner input_tokens/output_tokens integers are parsed. This means a present `{"input_tokens":0,"output_tokens":0}` is correctly reported as "0"/"0", and a missing usage block returns the sentinel.
Partial-data policy:
- Both message_start and message_delta parseable -> TokenUsage{Input:"<N>", Output:"<M>"}
- Only message_start parseable -> TokenUsage{Input:"<N>", Output:""} (logs "in=<N> out=-")
- Only message_delta parseable -> TokenUsage{Input:"", Output:"<M>"} (logs "in=- out=<M>")
- Neither parseable -> noUsage (logs "in=- out=-")
type TraceState ¶ added in v0.16.0
type TraceState struct {
// contains filtered or unexported fields
}
TraceState owns the process-global trace-enabled flag and its TTL timer. It guarantees exactly one live timer at any time: Enable() cancels the prior timer before starting a fresh window, and Disable() cancels the in-flight timer so no concurrent expiry can re-enable tracing. The flag is atomic for lock-free IsEnabled() reads on the request hot path.
func DefaultTraceState ¶ added in v0.16.0
func DefaultTraceState() *TraceState
DefaultTraceState returns the process-global trace-state instance.
func NewTraceState ¶ added in v0.16.0
func NewTraceState() *TraceState
NewTraceState returns a TraceState whose TTL window is the production constant TraceTTLDefault (5 minutes). Tests use NewTraceStateWithTTL with a shorter window.
func NewTraceStateWithTTL ¶ added in v0.16.0
func NewTraceStateWithTTL(ttl time.Duration) *TraceState
NewTraceStateWithTTL returns a TraceState whose TTL window is ttl. Tests use a short ttl for fast expiry assertions; production uses NewTraceState (TraceTTLDefault = 5 minutes).
func (*TraceState) Disable ¶ added in v0.16.0
func (ts *TraceState) Disable()
Disable sets the trace-enabled flag to false and cancels any in-flight TTL timer so no later expiry can flip tracing back on.
func (*TraceState) Enable ¶ added in v0.16.0
func (ts *TraceState) Enable()
Enable sets the trace-enabled flag to true, cancels any in-flight TTL timer, and starts a fresh TTL window. Repeated calls are idempotent on the flag but always reset the window to a fresh full ttl.
func (*TraceState) IsEnabled ¶ added in v0.16.0
func (ts *TraceState) IsEnabled() bool
IsEnabled returns the current trace-enabled flag value. This is the hot path consulted per request by the trace middleware; it performs an atomic load without locking.
type UpstreamMember ¶ added in v0.34.0
type UpstreamMember struct {
Upstream string
Handler http.Handler
Weight int
InFlight func() int
// Window, when non-nil, is the member's time-of-day eligibility
// window: the member is eligible only while Window.Contains(now)
// holds (spec 014). Nil = always eligible.
Window *pkg.Window
// Days, when non-nil, is the member's weekday eligibility set: the
// member is eligible only while Days.Contains(now, Window) holds
// (spec 017). Nil = every day.
Days *pkg.Days
// Now returns the router's current time (the injected
// libtime.CurrentDateTimeGetter). Consulted only when Window or Days
// is non-nil; nil Now falls back to the real clock.
Now func() libtime.DateTime
}
UpstreamMember is one server in a provider's upstream pool: the handler that serves it (per-upstream proxy wrapped in its concurrency limiter), the upstream URL used only for the [route] log line, and the selection inputs — Weight for the weighted ring hash of a pinned session, InFlight for least-loaded selection of keyless requests. InFlight may be nil, meaning "always 0" (an uncapped member). Window, when non-nil, restricts eligibility to the times its Contains holds; Days, when non-nil, restricts eligibility to the weekdays its set contains. A member whose weekday is not in its days set is excluded from session pinning and least-loaded selection alike (spec 017).
type WindowEligible ¶ added in v0.40.0
WindowEligible is implemented by provider handlers that can be ineligible for a dispatch because none of their pool members are eligible — their time windows and/or weekday sets exclude "now" (spec 014 / 017). A handler that does not implement the interface is always eligible.
Source Files
¶
- admin_loopback_guard.go
- anthropic-proxy.go
- auth-middleware.go
- auth-swap-transport.go
- body-sample.go
- circuit-breaker.go
- cold-start-gate.go
- concurrency-limiter.go
- disabletrace.go
- enabletrace.go
- healthz.go
- hello.go
- host-limiter.go
- inflight.go
- logging-roundtripper.go
- loopback.go
- metrics.go
- model-pool.go
- model-router.go
- not-found.go
- presented-api-key.go
- provider-inflight-collector.go
- redact.go
- rejection-reason.go
- root-liveness.go
- session-id.go
- session-middleware.go
- setloglevel.go
- status-recorder.go
- system-lift.go
- throttle-gate.go
- trace.go
- trace_state.go
- ttft-backpressure.go
- upstream-index.go
- upstream-inflight-collector.go
- upstream-pool-handler.go
- usage-recorder.go