diff --git a/.env.example b/.env.example index 7a5d584..e210428 100644 --- a/.env.example +++ b/.env.example @@ -45,6 +45,16 @@ OT_PROXY_FAILOVER_CAPS_NET=4 OT_PROXY_FAILOVER_CAPS_SERVER=3 OT_PROXY_FAILOVER_CAPS_RATE=2 OT_PROXY_FAILOVER_CAPS_QUOTA=3 +# 全部候选渠道都在运行期冷却时，最早恢复的渠道剩余冷却 ≤ 此值就兜底重试一次， +# 而不是立刻 503（429 的 Retry-After 语义本就是"到点再试"；0=关闭） +OT_PROXY_COOLDOWN_GRACE=30s +# 预扣（reserve/capture）估算的最大输出 token（0=关闭）：每请求按 body 粗估成本上界 +# 计入冻结额，余额检查变为「余额 - 在途冻结 ≤ 0」，并发的长流式请求不再互相透支 +OT_PROXY_RESERVE_MAX_TOKENS=4096 + +# 暴露 Prometheus /metrics 端点（默认关）：会泄露模型/渠道名与流量规模， +# 公网部署开启时建议在反代层做访问控制 +OT_METRICS_ENABLED=false # 流式空闲多久补一行 SSE 注释心跳（0=关闭）。挂在 nginx/Cloudflare 后面时， # 上游长时间不出字（长思考）会被对端的空闲超时掐掉，心跳用来续住这条流。 OT_PROXY_STREAM_HEARTBEAT=15s diff --git a/server/go.mod b/server/go.mod index 3ddfe38..3d4bf86 100644 --- a/server/go.mod +++ b/server/go.mod @@ -1,26 +1,29 @@ module github.com/openteam/server -go 1.26 +go 1.26.0 require ( github.com/gin-gonic/gin v1.12.0 github.com/glebarez/sqlite v1.11.0 github.com/go-webauthn/webauthn v0.17.4 github.com/golang-jwt/jwt/v5 v5.3.1 + github.com/prometheus/client_golang v1.25.0 github.com/shopspring/decimal v1.6.0 github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 github.com/tiktoken-go/tokenizer v0.8.1 - golang.org/x/crypto v0.55.0 - golang.org/x/term v0.45.0 + golang.org/x/crypto v0.57.0 + golang.org/x/term v0.46.0 gorm.io/driver/postgres v1.6.2 gorm.io/gorm v1.31.2 ) require ( + github.com/beorn7/perks v1.0.1 // indirect github.com/bytedance/gopkg v0.1.3 // indirect github.com/bytedance/sonic v1.15.0 // indirect github.com/bytedance/sonic/loader v0.5.0 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.6 // indirect github.com/dlclark/regexp2/v2 v2.5.1 // indirect github.com/dustin/go-humanize v1.0.1 // indirect @@ -51,8 +54,12 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/philhofer/fwd v1.2.0 // indirect + github.com/prometheus/client_model v0.6.3 // indirect + github.com/prometheus/common v0.72.0 // indirect + github.com/prometheus/procfs v0.22.0 // indirect github.com/quic-go/qpack v0.6.0 // indirect github.com/quic-go/quic-go v0.59.0 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect @@ -67,13 +74,13 @@ require ( github.com/ugorji/go/codec v1.3.1 // indirect github.com/x448/float16 v0.8.4 // indirect go.mongodb.org/mongo-driver/v2 v2.5.0 // indirect - go.yaml.in/yaml/v3 v3.0.4 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect golang.org/x/arch v0.22.0 // indirect - golang.org/x/net v0.57.0 // indirect - golang.org/x/sync v0.22.0 // indirect - golang.org/x/sys v0.47.0 // indirect - golang.org/x/text v0.41.0 // indirect - google.golang.org/protobuf v1.36.10 // indirect + golang.org/x/net v0.59.0 // indirect + golang.org/x/sync v0.23.0 // indirect + golang.org/x/sys v0.48.0 // indirect + golang.org/x/text v0.42.0 // indirect + google.golang.org/protobuf v1.36.12 // indirect modernc.org/libc v1.22.5 // indirect modernc.org/mathutil v1.5.0 // indirect modernc.org/memory v1.5.0 // indirect diff --git a/server/go.sum b/server/go.sum index 2606b87..58d4778 100644 --- a/server/go.sum +++ b/server/go.sum @@ -1,9 +1,13 @@ +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M= github.com/bytedance/gopkg v0.1.3/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM= github.com/bytedance/sonic v1.15.0 h1:/PXeWFaR5ElNcVE84U0dOHjiMHQOwNIx3K4ymzh/uSE= github.com/bytedance/sonic v1.15.0/go.mod h1:tFkWrPz0/CUCLEF4ri4UkHekCIcdnkqXw9VduqpJh0k= github.com/bytedance/sonic/loader v0.5.0 h1:gXH3KVnatgY7loH5/TkeVyXPfESoqSBSBEiDd5VjlgE= github.com/bytedance/sonic/loader v0.5.0/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudwego/base64x v0.1.6 h1:t11wG9AECkCDk5fMSoxmufanudBtJ+/HemLstXDLI2M= github.com/cloudwego/base64x v0.1.6/go.mod h1:OFcloc187FXDaYHvrNIjxSe8ncn0OOM8gEHfghB2IPU= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= @@ -77,12 +81,16 @@ github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y= github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= +github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= @@ -94,12 +102,21 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.25.0 h1:cDtW1TcFsMi1RJc8khwCKB4VhzwV04aKqbYtNOXuiK0= +github.com/prometheus/client_golang v1.25.0/go.mod h1:j2X/4/kaQHLPnmyEU8zjmkndo1T317VpCIQEml8LTKk= +github.com/prometheus/client_model v0.6.3 h1:O0jaTVAYNxTHYInEPFJt5I3+sN8zqBtVMPTB1qyxiEo= +github.com/prometheus/client_model v0.6.3/go.mod h1:gpN5P9S7Rr6Yr92PiQ+Ixvhf6JZEkF1dnxsYL2aPBEM= +github.com/prometheus/common v0.72.0 h1:tAYsE+sPJxIncDAobm4H5aQjmox9ZxEIIqPbiffa8G4= +github.com/prometheus/common v0.72.0/go.mod h1:77NWqAQ2tXT7BIK40qjJdw5Acrsrg1TlHAnsQi3i6mk= +github.com/prometheus/procfs v0.22.0 h1:6q9+/JL9IKAPbCmBrv9n5O5Ty3NKnciV5X7YGw0oics= +github.com/prometheus/procfs v0.22.0/go.mod h1:CvmFr/GVhIjIvWJZW3tgkODBQMRIf0EyWMQLHCHab58= github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= @@ -137,8 +154,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= -github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= -github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= +github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/tiktoken-go/tokenizer v0.8.1 h1:4obDoB6/dhdBt9xMweX4nww5cjdOq/nYF4ecwPq2+mg= @@ -153,32 +170,34 @@ github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE= go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y= go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU= -go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI= golang.org/x/arch v0.22.0/go.mod h1:dNHoOeKiyja7GTvF9NJS1l3Z2yntpQNzgrjh1cU103A= -golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= -golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= -golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= -golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= +golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= -golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= -golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= -golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= -google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= -google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= +golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= +golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gorm.io/driver/postgres v1.6.2 h1:BvXQ/cNUg63q5TFNg672DmDcowZSFrNLkkA3Xe6GXq4= gorm.io/driver/postgres v1.6.2/go.mod h1:0c4fQA44XhOklXDkgtuKqysHCycTa5i9e3EIpDGCwXk= diff --git a/server/internal/api/router.go b/server/internal/api/router.go index edb43e8..4f58104 100644 --- a/server/internal/api/router.go +++ b/server/internal/api/router.go @@ -8,6 +8,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/api/middleware" "github.com/openteam/server/internal/app" + "github.com/openteam/server/internal/pkg/metrics" "github.com/openteam/server/internal/proxy" ) @@ -149,5 +150,10 @@ func NewRouter(a *app.App, gw *proxy.Gateway) *gin.Engine { r.GET("/healthz", func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"status": "ok"}) }) + // Prometheus 指标端点：默认关闭——模型/渠道名与流量规模本身就是信息， + // 公网部署要开时建议在反代层做访问控制。 + if a.Cfg.MetricsEnabled { + r.GET("/metrics", gin.WrapH(metrics.Handler())) + } return r } diff --git a/server/internal/channel/channel.go b/server/internal/channel/channel.go index 31ee145..988785d 100644 --- a/server/internal/channel/channel.go +++ b/server/internal/channel/channel.go @@ -11,6 +11,7 @@ import ( "time" "github.com/openteam/server/internal/pkg/crypto" + "github.com/openteam/server/internal/pkg/metrics" "github.com/openteam/server/internal/store" "gorm.io/gorm" ) @@ -142,6 +143,43 @@ func (s *Service) Candidates(model string) []Candidate { return out } +// EarliestRecovery 在该模型的候选渠道里找运行期冷却最早结束的一个。 +// 全部候选都在冷却时的兜底（配合调用方的冷却宽限配置）：与其立刻 503， +// 不如试试最先恢复的那条。DB 层不可用（停用/健康探测失败）的渠道不参与—— +// 运行期冷却到期即可恢复，健康状态是另一回事，不要在这里越权放行。 +func (s *Service) EarliestRecovery(model string) (Candidate, time.Duration, bool) { + rows, bound := s.catalog(model) + var best Candidate + var bestRem time.Duration + found := false + for i := range rows { + ch := rows[i].ch + if !ch.Enabled || ch.HealthStatus != store.ChannelHealthHealthy { + continue + } + s.mu.Lock() + st := s.runt[ch.ID] + cooling := st != nil && !st.until.IsZero() && time.Now().Before(st.until) + var rem time.Duration + if cooling { + rem = time.Until(st.until) + } + s.mu.Unlock() + if !cooling { + continue + } + if !found || rem < bestRem { + cand := Candidate{Channel: &ch} + if bound { + cand.UpstreamModel = rows[i].upstreamModel + cand.Weight = rows[i].weight + } + best, bestRem, found = cand, rem, true + } + } + return best, bestRem, found +} + // catalog 取该模型的目录行（缓存命中返回副本，避免调用方改写缓存内容）。 // bound 表示这些行来自「该模型的渠道绑定」而不是全局渠道池。 func (s *Service) catalog(model string) (rows []catRow, bound bool) { @@ -370,6 +408,7 @@ func (s *Service) UpstreamKey(ch *store.Channel) (string, error) { // maxCooldown——不能让一个写错的重试头把渠道钉死几小时；比退避值短的 Retry-After // 不缩短冷却，否则 1s 就把坏渠道放回轮转。 func (s *Service) ReportFailure(ch *store.Channel, kind FailKind, retryAfter time.Duration) { + metrics.ChannelFailures.WithLabelValues(ch.Name, kind.String()).Inc() s.mu.Lock() st := s.runt[ch.ID] if st == nil { diff --git a/server/internal/cli/serve.go b/server/internal/cli/serve.go index 7f37ed5..296ef62 100644 --- a/server/internal/cli/serve.go +++ b/server/internal/cli/serve.go @@ -41,6 +41,8 @@ func serve() error { MaxBodyMB: cfg.Proxy.MaxBodyMB, FailoverBudget: cfg.Proxy.FailoverBudget, FailoverSwitchCaps: cfg.Proxy.FailoverSwitchCaps, + CooldownGrace: cfg.Proxy.CooldownGrace, + ReserveMaxTokens: cfg.Proxy.ReserveMaxTokens, StreamHeartbeat: cfg.Proxy.StreamHeartbeat, FirstTokenWait: cfg.Proxy.FirstTokenWait, AffinityTTL: cfg.Proxy.AffinityTTL, diff --git a/server/internal/config/config.go b/server/internal/config/config.go index d34ffe5..10599bd 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -31,6 +31,9 @@ type Config struct { // 反向代理列表（CIDR 或 IP）。只有配置它，gin 才信任 X-Forwarded-For 取客户端 IP， // 认证限速依赖这个 IP；直连部署留空即可。 TrustedProxies []string + + // MetricsEnabled 是否暴露 Prometheus /metrics 端点（会泄露模型/渠道名与流量，默认关）。 + MetricsEnabled bool } // UsageConfig 用量数据的保留策略。 @@ -90,7 +93,13 @@ type ProxyConfig struct { FailoverBudget time.Duration // 故障转移给"第二个及以后的渠道"留的时间预算，0=不限 // FailoverSwitchCaps 按失败类型（net|server|rate|quota）限制换渠道次数，0=不限。 // 429/配额类失败瞬时返回，时间预算对它们形同虚设，按次数封顶才有意义。 - FailoverSwitchCaps map[string]int + FailoverSwitchCaps map[string]int + // CooldownGrace 全部候选渠道都在运行期冷却时，最早恢复者剩余冷却 ≤ 此值就兜底 + // 重试一次而不是 503（429 的 Retry-After 语义）。0=关闭。 + CooldownGrace time.Duration + // ReserveMaxTokens 预扣（reserve/capture）估算的最大输出 token，0=关闭。 + // 每请求按 body 估算成本上界计入冻结额，402 判定改为「余额 - 在途冻结 ≤ 0」。 + ReserveMaxTokens int StreamHeartbeat time.Duration // 流式空闲多久发一次 SSE 注释心跳，0=关闭 FirstTokenWait time.Duration // 流式放行前等内容的最长时间，0=不限；超时未出内容换下一个渠道 AffinityTTL time.Duration // 会话亲和 TTL，0=关闭；同一会话粘住上次应答的渠道以命中上游缓存 @@ -166,6 +175,9 @@ func Load() (*Config, error) { v.SetDefault("proxy.failover_caps.server", 3) v.SetDefault("proxy.failover_caps.rate", 2) v.SetDefault("proxy.failover_caps.quota", 3) + v.SetDefault("proxy.cooldown_grace", "30s") + v.SetDefault("proxy.reserve_max_tokens", 4096) + v.SetDefault("metrics.enabled", false) v.SetDefault("proxy.stream_heartbeat", "15s") v.SetDefault("proxy.first_token_wait", "0s") v.SetDefault("proxy.affinity_ttl", "5m") @@ -226,6 +238,8 @@ func Load() (*Config, error) { "rate": v.GetInt("proxy.failover_caps.rate"), "quota": v.GetInt("proxy.failover_caps.quota"), }, + CooldownGrace: v.GetDuration("proxy.cooldown_grace"), + ReserveMaxTokens: v.GetInt("proxy.reserve_max_tokens"), StreamHeartbeat: v.GetDuration("proxy.stream_heartbeat"), FirstTokenWait: v.GetDuration("proxy.first_token_wait"), AffinityTTL: v.GetDuration("proxy.affinity_ttl"), @@ -253,6 +267,7 @@ func Load() (*Config, error) { AdminPassword: v.GetString("admin_password"), CORSAllowOrigins: splitList(v.GetString("cors.allow_origins")), TrustedProxies: splitList(v.GetString("server.trusted_proxies")), + MetricsEnabled: v.GetBool("metrics.enabled"), } if err := cfg.validate(); err != nil { return nil, err diff --git a/server/internal/pkg/metrics/metrics.go b/server/internal/pkg/metrics/metrics.go new file mode 100644 index 0000000..adf8637 --- /dev/null +++ b/server/internal/pkg/metrics/metrics.go @@ -0,0 +1,62 @@ +// Package metrics Prometheus 指标：进程内 registry + 文本端点（api 层按开关挂载 /metrics）。 +// +// label 基数约束：channel/model 来自受管目录、status/kind 为有限枚举，可安全做 label； +// 请求级随机值（trace id、用户 id、错误详情）禁止入 label，否则基数爆炸拖垮采集端。 +package metrics + +import ( + "net/http" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +var reg = prometheus.NewRegistry() + +var ( + // RequestsTotal 代理请求终结计数。channel 为 "none" 表示未打到任何渠道 + // （无候选/满载等），model 为请求的模型名。 + RequestsTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "openteam_proxy_requests_total", + Help: "Proxied requests terminated, by channel/model/status.", + }, []string{"channel", "model", "status"}) + + // RequestDuration 请求从进入 handler 到终结的耗时（按渠道）。 + RequestDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "openteam_proxy_request_duration_seconds", + Help: "Proxy request duration, by channel.", + Buckets: []float64{.05, .1, .25, .5, 1, 2.5, 5, 10, 30, 60, 120, 300}, + }, []string{"channel"}) + + // ChannelFailures 上游失败计数（按渠道与失败类型），与运行期冷却同源。 + ChannelFailures = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "openteam_channel_failures_total", + Help: "Upstream failures, by channel/fail kind.", + }, []string{"channel", "kind"}) + + // RecorderQueueDepth 记账队列中待落库的记录数。持续增长说明 DB 写入跟不上。 + RecorderQueueDepth = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "openteam_usage_queue_depth", + Help: "Usage records buffered, waiting to be flushed.", + }) + + // FrozenBalance 在途请求预扣（冻结）的金额合计（USD）。见 proxy 的 reserve/capture。 + FrozenBalance = prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "openteam_frozen_balance", + Help: "Balance reserved by in-flight requests (USD).", + }) +) + +func init() { + reg.MustRegister( + RequestsTotal, RequestDuration, ChannelFailures, RecorderQueueDepth, FrozenBalance, + collectors.NewGoCollector(), + collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), + ) +} + +// Handler /metrics 的 http.Handler（独立 registry，避免全局默认注册器的耦合）。 +func Handler() http.Handler { + return promhttp.HandlerFor(reg, promhttp.HandlerOpts{}) +} diff --git a/server/internal/pkg/metrics/metrics_test.go b/server/internal/pkg/metrics/metrics_test.go new file mode 100644 index 0000000..9e41aea --- /dev/null +++ b/server/internal/pkg/metrics/metrics_test.go @@ -0,0 +1,58 @@ +package metrics + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// /metrics 文本端点：注入计数/观测后可被抓取，指标名与 label 齐全。 +func TestHandlerExposesInstruments(t *testing.T) { + RequestsTotal.WithLabelValues("c1", "m1", "success").Inc() + RequestsTotal.WithLabelValues("none", "m1", "error").Inc() + RequestDuration.WithLabelValues("c1").Observe(0.42) + ChannelFailures.WithLabelValues("c1", "rate_limited").Inc() + RecorderQueueDepth.Set(7) + FrozenBalance.Set(0.00246) + + srv := httptest.NewServer(Handler()) + defer srv.Close() + resp, err := http.Get(srv.URL) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d", resp.StatusCode) + } + buf := make([]byte, 1<<16) + n, _ := resp.Body.Read(buf) + body := string(buf[:n]) + for _, want := range []string{ + "openteam_proxy_requests_total{", + `channel="c1"`, + `status="success"`, + "openteam_proxy_request_duration_seconds", + "openteam_channel_failures_total", + "openteam_usage_queue_depth 7", + "openteam_frozen_balance 0.00246", + "go_goroutines", // Go 运行时指标一并暴露 + } { + if !strings.Contains(body, want) { + t.Errorf("metrics 输出缺少 %q", want) + } + } + // 直方图观测应产生 bucket 样本 + if !strings.Contains(body, `openteam_proxy_request_duration_seconds_bucket`) && n < len(buf) { + // 一次 Read 可能读不完，重读剩余部分再断言 + rest := make([]byte, 1<<16) + n2, _ := resp.Body.Read(rest) + body += string(rest[:n2]) + if !strings.Contains(body, "openteam_proxy_request_duration_seconds_bucket") { + t.Errorf("metrics 输出缺少直方图 bucket 样本") + } + } + _ = time.Second +} diff --git a/server/internal/proxy/gateway.go b/server/internal/proxy/gateway.go index b8832d7..c25ef29 100644 --- a/server/internal/proxy/gateway.go +++ b/server/internal/proxy/gateway.go @@ -16,11 +16,14 @@ import ( "github.com/openteam/server/internal/channel" "github.com/openteam/server/internal/pkg/apikey" "github.com/openteam/server/internal/pkg/crypto" + "github.com/openteam/server/internal/pkg/metrics" "github.com/openteam/server/internal/pkg/ratelimit" "github.com/openteam/server/internal/proxy/convert" "github.com/openteam/server/internal/store" "github.com/openteam/server/internal/usage" + "github.com/shopspring/decimal" "gorm.io/gorm" + "sync/atomic" ) const ( @@ -29,6 +32,11 @@ const ( CtxKeyID = "proxy_key_id" CtxKeyModel = "proxy_key_models" CtxTrace = "proxy_trace_id" + + // CtxReserve 本请求的预扣（冻结）金额，单位 1e-8 USD（与 numeric(20,8) 对齐）。 + CtxReserve = "proxy_reserve_1e8" + // CtxForceCooling 候选集来自「全冷却时最早恢复」的兜底，选路时放行冷却中的渠道。 + CtxForceCooling = "proxy_force_cooling" ) type Gateway struct { @@ -49,6 +57,16 @@ type Gateway struct { // 按失败类型限制换渠道次数（0=不限）：429/配额类失败是瞬时返回， // 时间预算对它们形同虚设，按次数封顶才有意义。 failoverCaps map[channel.FailKind]int + // 冷却兜底宽限：全部候选都在运行期冷却时，最早恢复的渠道剩余冷却 ≤ 此值 + // 就带上它兜底重试一次（429 的 Retry-After 语义本就是"到点再试"）；0=关闭。 + cooldownGrace time.Duration + // 预扣（reserve/capture）：每请求按 body 估算成本上界计入冻结额， + // 402 判定改为「余额 - 在途冻结 ≤ 0」，并发的长流式请求不再能把余额拖到深度负值。 + // reserveMaxTokens 是输出 token 估算基数（0=关闭预扣）；冻结额按用户分桶存 + // 1e-8 USD 整数（每用户一个 atomic.Int64），冻结/释放 O(1) 无锁竞争。 + reserveMaxTokens int + frozen sync.Map // uint64(用户ID) → *atomic.Int64（1e-8 USD） + frozenTotal atomic.Int64 // 全局冻结合计（指标用） // 流式空闲多久补一行 SSE 注释心跳（0=关闭）。反向代理的空闲超时会掐掉长思考的流。 streamHeartbeat time.Duration // 流式放行前等多久内容（0=不限制）。超时未出内容取消本次尝试换下一个渠道。 @@ -226,6 +244,8 @@ type Options struct { MaxBodyMB int // /v1 请求体上限（MB） FailoverBudget time.Duration // 故障转移重试的时间预算 FailoverSwitchCaps map[string]int // 按失败类型（net|server|rate|quota）限制换渠道次数，0=不限 + CooldownGrace time.Duration // 全冷却时最早恢复渠道的兜底宽限（0=关闭） + ReserveMaxTokens int // 预扣估算的最大输出 token（0=关闭预扣） StreamHeartbeat time.Duration // 流式空闲心跳间隔 FirstTokenWait time.Duration // 流式放行前等内容的最长时间（0=不限） AffinityTTL time.Duration // 会话亲和 TTL（0=关闭） @@ -267,6 +287,8 @@ func NewGateway(db *gorm.DB, enc *crypto.Encryptor, rec *usage.Recorder, lim *ra maxBodyMB: o.MaxBodyMB, failoverBudget: o.FailoverBudget, failoverCaps: failoverCapsFrom(o.FailoverSwitchCaps), + cooldownGrace: o.CooldownGrace, + reserveMaxTokens: o.ReserveMaxTokens, streamHeartbeat: o.StreamHeartbeat, firstTokenWait: o.FirstTokenWait, affinityTTL: o.AffinityTTL, @@ -500,6 +522,23 @@ func (g *Gateway) candidateChannels(model string) []channel.Candidate { return g.ch.Candidates(model) } +// candidatesFor 候选获取的统一入口：健康候选为空且配置了冷却宽限时， +// 找冷却最早结束的候选兜底重试一次（429 的 Retry-After 语义本就是"到点再试"， +// 与其立刻 503 不如试试最先恢复的那条）。命中兜底时在 context 打标， +// doProxy 选路会放行这个冷却中的渠道。 +func (g *Gateway) candidatesFor(c *gin.Context, model string) []channel.Candidate { + cands := g.candidateChannels(model) + if len(cands) > 0 || g.cooldownGrace <= 0 { + return cands + } + cand, remaining, ok := g.ch.EarliestRecovery(model) + if !ok || remaining > g.cooldownGrace { + return nil + } + c.Set(CtxForceCooling, true) + return []channel.Candidate{cand} +} + // resolveUser 取当前用户（含余额）。Auth 中间件已查过一次并放入上下文， // handler 里直接复用，省掉每请求一次重复查询。 func (g *Gateway) resolveUser(c *gin.Context) (*store.User, bool) { @@ -518,14 +557,86 @@ func (g *Gateway) resolveUser(c *gin.Context) (*store.User, bool) { } // checkBalance 余额不足返回 402（PLANNING §4.4.3）。 +// 可用余额 = 余额 - 在途预扣（冻结）额：并发流式请求各持一份冻结， +// 不会互相透支把余额拖到深度负值（冻结语义见 freezeFor）。 func (g *Gateway) checkBalance(c *gin.Context, u *store.User) bool { - if !u.Balance.IsPositive() { + avail := u.Balance + if a := g.frozenOf(u.ID); a != nil && a.Load() > 0 { + avail = avail.Sub(decimal.NewFromInt(a.Load()).Shift(-8)) + } + if !avail.IsPositive() { apiError(c, http.StatusPaymentRequired, "insufficient_balance", "Insufficient balance. Please recharge or contact admin.") return false } return true } +// frozenOf 取用户的在途冻结额桶；无冻结返回 nil（零分配快路径）。 +func (g *Gateway) frozenOf(uid uint64) *atomic.Int64 { + if v, ok := g.frozen.Load(uid); ok { + return v.(*atomic.Int64) + } + return nil +} + +// freezeFor 预扣（reserve）：按请求体估算本请求的成本上界并计入冻结额，随 gin context +// 携带，请求终结（含失败/中断/panic，由 handler 的 defer 释放）时按同一数值 capture。 +// 估算只求上界不求准：输入按 ~4 字节/token 粗估并用缓存写价（1.25×/2× 输入价）封顶， +// 输出取 max_tokens（缺省用 reserveMaxTokens 基数）。 +// 冻结只在本进程内存里，单实例语义；多实例部署时各实例独立冻结（余额本身允许为负兜底）。 +// 检查（checkBalance）与冻结之间存在乐观竞窗：最坏超额 = 同时穿过竞窗的请求数， +// 有界且被「余额允许为负」兜底；预扣把无界的流式透支压成了这个常数。 +func (g *Gateway) freezeFor(c *gin.Context, uid uint64, model store.Model, br *bodyReq, body []byte) { + if g.reserveMaxTokens <= 0 || model.ID == 0 { + return + } + outTok := int64(g.reserveMaxTokens) + if br != nil { + if br.MaxTokens != nil && *br.MaxTokens > 0 && int64(*br.MaxTokens) < outTok { + outTok = int64(*br.MaxTokens) + } + if br.MaxOutputTokens != nil && *br.MaxOutputTokens > 0 && int64(*br.MaxOutputTokens) < outTok { + outTok = int64(*br.MaxOutputTokens) + } + } + inPrice := model.InputPrice + if cwp := model.CacheWritePriceOr(); cwp.GreaterThan(inPrice) { + inPrice = cwp + } + million := decimal.NewFromInt(1e6) + est := decimal.NewFromInt(int64(len(body) / 4)).Mul(inPrice). + Add(decimal.NewFromInt(outTok).Mul(model.OutputPrice)). + Div(million).Round(8) + n := est.Shift(8).Ceil().BigInt().Int64() // 1e-8 USD 整数，向上取整保上界 + if n <= 0 { + return + } + v, _ := g.frozen.LoadOrStore(uid, &atomic.Int64{}) + v.(*atomic.Int64).Add(n) + g.frozenTotal.Add(n) + metrics.FrozenBalance.Set(float64(g.frozenTotal.Load()) / 1e8) + c.Set(CtxReserve, n) +} + +// unfreeze 释放冻结额（capture）。幂等：CtxReserve 置零后重复调用为 no-op， +// handler 的 defer 与任何终结路径的双释放只生效一次。 +func (g *Gateway) unfreeze(c *gin.Context, uid uint64) { + v, ok := c.Get(CtxReserve) + if !ok { + return + } + n, _ := v.(int64) + c.Set(CtxReserve, int64(0)) + if n == 0 { + return + } + if a := g.frozenOf(uid); a != nil { + a.Add(-n) + } + total := g.frozenTotal.Add(-n) + metrics.FrozenBalance.Set(float64(total) / 1e8) +} + // --------------------------------------------------------------------------- // 协议分派 diff --git a/server/internal/proxy/handlers.go b/server/internal/proxy/handlers.go index 71bf2ec..f56c535 100644 --- a/server/internal/proxy/handlers.go +++ b/server/internal/proxy/handlers.go @@ -46,7 +46,13 @@ func (g *Gateway) chatCompletions(c *gin.Context) { return } - cands := g.candidateChannels(br.Model) + // 预扣（reserve）：按 body 估算成本上界计入冻结额；defer 释放（capture）， + // 失败/中断/panic 都会走到，finishUsage 的扣款在冻结释放前完成。 + model, _ := g.modelByName(br.Model) + g.freezeFor(c, u.ID, model, br, body) + defer g.unfreeze(c, u.ID) + + cands := g.candidatesFor(c, br.Model) if len(cands) == 0 { apiError(c, http.StatusServiceUnavailable, "no_channel", "No available upstream channel") g.recordError(c, nil, nil, now(), "no_channel") @@ -82,7 +88,13 @@ func (g *Gateway) responses(c *gin.Context) { return } - cands := g.candidateChannels(br.Model) + // 预扣（reserve）：按 body 估算成本上界计入冻结额；defer 释放（capture）， + // 失败/中断/panic 都会走到，finishUsage 的扣款在冻结释放前完成。 + model, _ := g.modelByName(br.Model) + g.freezeFor(c, u.ID, model, br, body) + defer g.unfreeze(c, u.ID) + + cands := g.candidatesFor(c, br.Model) if len(cands) == 0 { apiError(c, http.StatusServiceUnavailable, "no_channel", "No available upstream channel") g.recordError(c, nil, nil, now(), "no_channel") @@ -118,7 +130,13 @@ func (g *Gateway) messages(c *gin.Context) { return } - cands := g.candidateChannels(br.Model) + // 预扣（reserve）：按 body 估算成本上界计入冻结额；defer 释放（capture）， + // 失败/中断/panic 都会走到，finishUsage 的扣款在冻结释放前完成。 + model, _ := g.modelByName(br.Model) + g.freezeFor(c, u.ID, model, br, body) + defer g.unfreeze(c, u.ID) + + cands := g.candidatesFor(c, br.Model) if len(cands) == 0 { apiError(c, http.StatusServiceUnavailable, "no_channel", "No available upstream channel") g.recordError(c, nil, nil, now(), "no_channel") diff --git a/server/internal/proxy/passthrough.go b/server/internal/proxy/passthrough.go index 333c930..811a46d 100644 --- a/server/internal/proxy/passthrough.go +++ b/server/internal/proxy/passthrough.go @@ -17,6 +17,7 @@ import ( "github.com/gin-gonic/gin" "github.com/openteam/server/internal/channel" + "github.com/openteam/server/internal/pkg/metrics" "github.com/openteam/server/internal/pkg/tokenizer" "github.com/openteam/server/internal/proxy/convert" "github.com/openteam/server/internal/store" @@ -34,6 +35,9 @@ func newTraceID() string { type bodyReq struct { Model string `json:"model"` Stream bool `json:"stream"` + // 预扣估算用（freezeFor）：chat/messages 的 max_tokens、responses 的 max_output_tokens + MaxTokens *int `json:"max_tokens"` + MaxOutputTokens *int `json:"max_output_tokens"` } // parseBody 读取并回填请求体，解析 model/stream。 @@ -283,6 +287,8 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto // 那不是「上游坏了」（502），而是「我方满了」（503 + Retry-After）。 saturated := false startAll := time.Now() + // 全冷却兜底：候选来自 EarliestRecovery 时放行冷却中的渠道（只此一次尝试机会） + _, forceCooling := c.Get(CtxForceCooling) // 换渠道次数按失败类型计数：cap 到了就不再为同类失败换渠道（配置见 failover_caps）。 switches := map[channel.FailKind]int{} var lastKind channel.FailKind @@ -311,7 +317,7 @@ func (g *Gateway) doProxy(c *gin.Context, cands []channel.Candidate, clientProto g.finishUsage(c, ch, now(), store.UsageStatusCanceled, "client_disconnect") return } - if g.ch.InCooldown(ch.ID) { + if !forceCooling && g.ch.InCooldown(ch.ID) { continue // 本次请求期间已被并发请求判入运行期冷却 } plan, err := prepareUpstream(ch, clientProto, body, cand.UpstreamModel) @@ -1071,6 +1077,14 @@ func (g *Gateway) finishUsage(c *gin.Context, ch *store.Channel, start time.Time traceStr, _ := trace.(string) latency := int(time.Since(start).Milliseconds()) + // 指标：终结计数与耗时。channel 为 "none" 表示没打到任何渠道（无候选/满载等） + chName := "none" + if ch != nil { + chName = ch.Name + } + metrics.RequestsTotal.WithLabelValues(chName, mn, status).Inc() + metrics.RequestDuration.WithLabelValues(chName).Observe(time.Since(start).Seconds()) + if status == store.UsageStatusSuccess && c.Writer.Status() >= 400 { status = store.UsageStatusError } diff --git a/server/internal/proxy/proxy_test.go b/server/internal/proxy/proxy_test.go index 9693f5f..3301dd8 100644 --- a/server/internal/proxy/proxy_test.go +++ b/server/internal/proxy/proxy_test.go @@ -509,6 +509,91 @@ func TestFailoverSwitchCaps(t *testing.T) { } } +// 预扣（reserve/capture）：在途请求的冻结额让「余额 - 冻结 ≤ 0」触发 402， +// 释放后恢复放行。并发的长流式请求不再互相透支。 +func TestReserveFreezesBalance(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, `{"id":"x","choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`) + })) + defer srv.Close() + + h := newHarness(t, srv.URL, 0, true, nil) + h.gw.reserveMaxTokens = 4096 + // 余额压到 0.001：4096 out × 0.6/1e6 = 0.00246 的冻结上界必然覆盖它 + if err := h.db.Model(&store.User{}).Where("username = ?", "u1"). + Update("balance", 0.001).Error; err != nil { + t.Fatal(err) + } + var u store.User + h.db.Where("username = ?", "u1").First(&u) + var model store.Model + h.db.Where("name = ?", "m1").First(&model) + + // 模拟一条在途请求的冻结（不释放） + fr, _ := gin.CreateTestContext(httptest.NewRecorder()) + fr.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) + h.gw.freezeFor(fr, u.ID, model, &bodyReq{MaxTokens: nil}, make([]byte, 64)) + if w := h.run(t, nil, `{"model":"m1","messages":[{"role":"user","content":"hi"}]}`); w.Code != http.StatusPaymentRequired { + t.Fatalf("冻结额覆盖余额时应 402，实际 %d: %s", w.Code, w.Body.String()) + } + + // capture（释放）后恢复放行 + h.gw.unfreeze(fr, u.ID) + if a := h.gw.frozenOf(u.ID); a != nil && a.Load() != 0 { + t.Fatalf("释放后冻结额应为 0，实际 %d", a.Load()) + } + w := h.run(t, nil, `{"model":"m1","messages":[{"role":"user","content":"hi"}]}`) + if w.Code != http.StatusOK { + t.Fatalf("释放后应正常代理，实际 %d: %s", w.Code, w.Body.String()) + } + + // 幂等：重复 unfreeze 不产生负冻结 + h.gw.unfreeze(fr, u.ID) + if a := h.gw.frozenOf(u.ID); a != nil && a.Load() < 0 { + t.Fatalf("重复释放不应产生负冻结，实际 %d", a.Load()) + } +} + +// 冷却兜底宽限：唯一渠道进运行期冷却后，请求 503；配置 grace 且上游已恢复时， +// 带上冷却中的渠道兜底重试一次即成功。 +func TestCooldownGraceFallback(t *testing.T) { + var healthy atomic.Bool + healthy.Store(false) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !healthy.Load() { + w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, `{"error":{"message":"boom"}}`) + return + } + _, _ = io.WriteString(w, `{"id":"x","choices":[{"message":{"role":"assistant","content":"hi"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`) + })) + defer srv.Close() + + h := newHarness(t, srv.URL, 0, true, nil) + body := `{"model":"m1","messages":[{"role":"user","content":"hi"}]}` + + // 连续 3 次 5xx：触发运行期冷却（failThreshold=3） + for i := 0; i < 3; i++ { + h.run(t, nil, body) + } + // 默认 grace=0（关闭）：冷却中 → 无候选 → 503 + if w := h.run(t, nil, body); w.Code != http.StatusServiceUnavailable { + t.Fatalf("冷却中且未配置宽限时应 503，实际 %d: %s", w.Code, w.Body.String()) + } + + // 上游恢复 + 配置宽限：兜底重试冷却中的渠道 → 200 + healthy.Store(true) + h.gw.cooldownGrace = time.Minute + if w := h.run(t, nil, body); w.Code != http.StatusOK { + t.Fatalf("宽限期内应兜底重试恢复的渠道，实际 %d: %s", w.Code, w.Body.String()) + } + // 冷却截止时间未被成功清除（由 InCooldown 到期惰性恢复，语义不变）， + // 但宽限期内每次请求都能经兜底打到这条渠道 + if w := h.run(t, nil, body); w.Code != http.StatusOK { + t.Fatalf("宽限期内后续请求也应走兜底，实际 %d: %s", w.Code, w.Body.String()) + } +} + // 流式空闲时补 SSE 注释心跳：上游长时间不出字（长思考）不能被反代空闲超时掐掉。 func TestStreamHeartbeatWhileIdle(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/server/internal/usage/usage.go b/server/internal/usage/usage.go index 5880779..6b3a300 100644 --- a/server/internal/usage/usage.go +++ b/server/internal/usage/usage.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "github.com/openteam/server/internal/pkg/metrics" "github.com/openteam/server/internal/store" "github.com/shopspring/decimal" "gorm.io/gorm" @@ -115,6 +116,16 @@ func (r *Recorder) Record(l *store.UsageLog) { r.mu.RUnlock() r.flushRetry([]*store.UsageLog{l}) } + r.updateQueueDepth() +} + +// updateQueueDepth 刷新记账队列深度指标（多 recorder 并存时后写者胜，无碍）。 +func (r *Recorder) updateQueueDepth() { + var depth int + for _, ch := range r.shards { + depth += len(ch) + } + metrics.RecorderQueueDepth.Set(float64(depth)) } // Close 停止接收新记账并等待后台 goroutine 排空队列后返回。幂等。