fujin/anthropic-proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(ratelimit): remove per-window token tracking from proxy

Browse Source 
Window token counts are now computed in Grafana using the @ modifier
with dashboard variables derived from proxy_usage_resets_at. This
eliminates in-memory state, file persistence, and restart sensitivity.

Removes: TokensIn/Out, RecordTokens, setResetTime, persist.go,
window_tokens observable gauges. -171 lines.
This commit is contained in:
Alexander
2026-04-14 14:25:31 +02:00
parent 273213cbed
commit 27b647e9b4
4 changed files with 13 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ratelimit/tracker.go
+13 -63
View File
@@ -5,7 +5,6 @@ import (
"net/http"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/rs/zerolog/log"
@@ -15,21 +14,17 @@ import (
type Window struct {
Utilization float64 // 0-100 from API
ResetsAt time.Time // when window resets
TokensIn atomic.Int64
TokensOut atomic.Int64
}
// Snapshot is a read-only copy of a Window's state.
type Snapshot struct {
Utilization float64
ResetsAt time.Time
TokensIn int64
TokensOut int64
}
// Tracker polls /api/oauth/usage and tracks per-window token usage.
// Tracker polls /api/oauth/usage and tracks per-window utilization.
type Tracker struct {
tokenFn func() string // returns current OAuth access token
tokenFn func() string
mu sync.RWMutex
fiveHour Window
sevenDay Window
@@ -46,34 +41,30 @@ func NewTracker(tokenFn func() string) *Tracker {
func (t *Tracker) Start(ctx context.Context) {
go func() {
t.poll(ctx)
t.Restore()
for {
select {
case <-ctx.Done():
t.Save()
return
case <-time.After(5 * time.Minute):
t.poll(ctx)
t.Save()
}
}
}()
}
// UpdateFromHeaders extracts rate limit data from /v1/messages response headers.
// This provides real-time utilization updates without polling the usage API.
func (t *Tracker) UpdateFromHeaders(h http.Header) {
t.mu.Lock()
defer t.mu.Unlock()
if v := h.Get("Anthropic-Ratelimit-Unified-5h-Utilization"); v != "" {
if f, err := strconv.ParseFloat(v, 64); err == nil {
t.fiveHour.Utilization = f * 100 // header is 0-1, we store 0-100
t.fiveHour.Utilization = f * 100
}
}
if v := h.Get("Anthropic-Ratelimit-Unified-5h-Reset"); v != "" {
if ts, err := strconv.ParseInt(v, 10, 64); err == nil {
t.setResetTime(&t.fiveHour, time.Unix(ts, 0).UTC().Truncate(time.Minute), "5h")
t.fiveHour.ResetsAt = time.Unix(ts, 0).UTC().Truncate(time.Minute)
}
}
if v := h.Get("Anthropic-Ratelimit-Unified-7d-Utilization"); v != "" {
@@ -83,51 +74,30 @@ func (t *Tracker) UpdateFromHeaders(h http.Header) {
}
if v := h.Get("Anthropic-Ratelimit-Unified-7d-Reset"); v != "" {
if ts, err := strconv.ParseInt(v, 10, 64); err == nil {
t.setResetTime(&t.sevenDay, time.Unix(ts, 0).UTC().Truncate(time.Minute), "7d")
t.sevenDay.ResetsAt = time.Unix(ts, 0).UTC().Truncate(time.Minute)
}
}
}
// RecordTokens adds tokens to all active windows.
func (t *Tracker) RecordTokens(inputTokens, outputTokens int64) {
t.fiveHour.TokensIn.Add(inputTokens)
t.fiveHour.TokensOut.Add(outputTokens)
t.sevenDay.TokensIn.Add(inputTokens)
t.sevenDay.TokensOut.Add(outputTokens)
}
// FiveHour returns a snapshot of the 5-hour window.
func (t *Tracker) FiveHour() Snapshot {
t.mu.RLock()
defer t.mu.RUnlock()
return Snapshot{
Utilization: t.fiveHour.Utilization,
ResetsAt: t.fiveHour.ResetsAt,
TokensIn: t.fiveHour.TokensIn.Load(),
TokensOut: t.fiveHour.TokensOut.Load(),
}
return Snapshot{Utilization: t.fiveHour.Utilization, ResetsAt: t.fiveHour.ResetsAt}
}
// SevenDay returns a snapshot of the 7-day window.
func (t *Tracker) SevenDay() Snapshot {
t.mu.RLock()
defer t.mu.RUnlock()
return Snapshot{
Utilization: t.sevenDay.Utilization,
ResetsAt: t.sevenDay.ResetsAt,
TokensIn: t.sevenDay.TokensIn.Load(),
TokensOut: t.sevenDay.TokensOut.Load(),
}
return Snapshot{Utilization: t.sevenDay.Utilization, ResetsAt: t.sevenDay.ResetsAt}
}
// Sonnet returns a snapshot of the 7-day sonnet window.
func (t *Tracker) Sonnet() Snapshot {
t.mu.RLock()
defer t.mu.RUnlock()
return Snapshot{
Utilization: t.sonnet.Utilization,
ResetsAt: t.sonnet.ResetsAt,
}
return Snapshot{Utilization: t.sonnet.Utilization, ResetsAt: t.sonnet.ResetsAt}
}
// Extra returns the current extra usage state.
@@ -153,13 +123,13 @@ func (t *Tracker) poll(ctx context.Context) {
defer t.mu.Unlock()
if usage.FiveHour != nil {
t.updateWindow(&t.fiveHour, usage.FiveHour, "5h")
t.updateWindow(&t.fiveHour, usage.FiveHour)
}
if usage.SevenDay != nil {
t.updateWindow(&t.sevenDay, usage.SevenDay, "7d")
t.updateWindow(&t.sevenDay, usage.SevenDay)
}
if usage.SevenDaySonnet != nil {
t.updateWindow(&t.sonnet, usage.SevenDaySonnet, "7d_sonnet")
t.updateWindow(&t.sonnet, usage.SevenDaySonnet)
}
if usage.ExtraUsage != nil {
t.extra = *usage.ExtraUsage
@@ -168,15 +138,10 @@ func (t *Tracker) poll(ctx context.Context) {
log.Debug().
Float64("5h_util", t.fiveHour.Utilization).
Time("5h_resets", t.fiveHour.ResetsAt).
Int64("5h_tokens_in", t.fiveHour.TokensIn.Load()).
Int64("5h_tokens_out", t.fiveHour.TokensOut.Load()).
Float64("7d_util", t.sevenDay.Utilization).
Time("7d_resets", t.sevenDay.ResetsAt).
Int64("7d_tokens_in", t.sevenDay.TokensIn.Load()).
Int64("7d_tokens_out", t.sevenDay.TokensOut.Load()).
Msg("usage poll")
// Warn on high utilization
if t.fiveHour.Utilization > 80 {
log.Warn().Float64("utilization", t.fiveHour.Utilization).Time("resets_at", t.fiveHour.ResetsAt).Msg("5h window utilization high")
}
@@ -185,7 +150,7 @@ func (t *Tracker) poll(ctx context.Context) {
}
}
func (t *Tracker) updateWindow(w *Window, rl *RateLimit, name string) {
func (t *Tracker) updateWindow(w *Window, rl *RateLimit) {
if rl.Utilization != nil {
w.Utilization = *rl.Utilization
}
@@ -195,22 +160,7 @@ func (t *Tracker) updateWindow(w *Window, rl *RateLimit, name string) {
parsed, err = time.Parse(time.RFC3339, *rl.ResetsAt)
}
if err == nil {
t.setResetTime(w, parsed.UTC().Truncate(time.Minute), name)
w.ResetsAt = parsed.UTC().Truncate(time.Minute)
}
}
}
func (t *Tracker) setResetTime(w *Window, newReset time.Time, name string) {
if !w.ResetsAt.IsZero() && newReset != w.ResetsAt {
log.Info().
Str("window", name).
Int64("prev_tokens_in", w.TokensIn.Load()).
Int64("prev_tokens_out", w.TokensOut.Load()).
Time("old_reset", w.ResetsAt).
Time("new_reset", newReset).
Msg("window reset detected")
w.TokensIn.Store(0)
w.TokensOut.Store(0)
}
w.ResetsAt = newReset
}