diff --git a/.github/workflows/docker-image-alpha.yml b/.github/workflows/docker-image-alpha.yml deleted file mode 100644 index 2a7d43ad53ff..000000000000 --- a/.github/workflows/docker-image-alpha.yml +++ /dev/null @@ -1,151 +0,0 @@ -name: Publish Docker image (alpha) - -on: - push: - branches: - - alpha - workflow_dispatch: - inputs: - name: - description: "reason" - required: false - -jobs: - build_single_arch: - name: Build & push (${{ matrix.arch }}) [native] - strategy: - fail-fast: false - matrix: - include: - - arch: amd64 - platform: linux/amd64 - runner: ubuntu-latest - - arch: arm64 - platform: linux/arm64 - runner: ubuntu-24.04-arm - runs-on: ${{ matrix.runner }} - permissions: - packages: write - contents: read - steps: - - name: Check out (shallow) - uses: actions/checkout@v4 - with: - fetch-depth: 1 - - - name: Determine alpha version - id: version - run: | - VERSION="alpha-$(date +'%Y%m%d')-$(git rev-parse --short HEAD)" - echo "$VERSION" > VERSION - echo "value=$VERSION" >> $GITHUB_OUTPUT - echo "VERSION=$VERSION" >> $GITHUB_ENV - echo "Publishing version: $VERSION for ${{ matrix.arch }}" - - - name: Normalize GHCR repository - run: echo "GHCR_REPOSITORY=${GITHUB_REPOSITORY,,}" >> $GITHUB_ENV - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 - - - name: Log in to Docker Hub - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKERHUB_USERNAME }} - password: ${{ secrets.DOCKERHUB_TOKEN }} - - - name: Log in to GHCR - uses: docker/login-action@v3 - with: - registry: ghcr.io - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - - name: Extract metadata (labels) - id: meta - uses: docker/metadata-action@v5 - with: - images: | - calciumion/new-api - ghcr.io/${{ env.GHCR_REPOSITORY }} - - - name: Build & push single-arch (to both registries) - uses: docker/build-push-action@v6 - with: - context: . - platforms: ${{ matrix.platform }} - push: true - tags: | - calciumion/new-api:alpha-${{ matrix.arch }} - calciumion/new-api:${{ steps.version.outputs.value }}-${{ matrix.arch }} - ghcr.io/${{ env.GHCR_REPOSITORY }}:alpha-${{ matrix.arch }} - ghcr.io/${{ env.GHCR_REPOSITORY }}:${{ steps.version.outputs.value }}-${{ matrix.arch }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max - provenance: false - sbom: false - - create_manifests: - name: Create multi-arch manifests (Docker Hub + GHCR) - needs: [build_single_arch] - runs-on: ubuntu-latest - permissions: - packages: write - contents: read - steps: - - name: Check out (shallow) - uses: actions/checkout@v4 - with: - fetch-depth: 1 - - - name: Normalize GHCR repository - run: echo "GHCR_REPOSITORY=${GITHUB_REPOSITORY,,}" >> $GITHUB_ENV - - - name: Determine alpha version - id: version - run: | - VERSION="alpha-$(date +'%Y%m%d')-$(git rev-parse --short HEAD)" - echo "value=$VERSION" >> $GITHUB_OUTPUT - echo "VERSION=$VERSION" >> $GITHUB_ENV - - - name: Log in to Docker Hub - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKERHUB_USERNAME }} - password: ${{ secrets.DOCKERHUB_TOKEN }} - - - name: Create & push manifest (Docker Hub - alpha) - run: | - docker buildx imagetools create \ - -t calciumion/new-api:alpha \ - calciumion/new-api:alpha-amd64 \ - calciumion/new-api:alpha-arm64 - - - name: Create & push manifest (Docker Hub - versioned alpha) - run: | - docker buildx imagetools create \ - -t calciumion/new-api:${VERSION} \ - calciumion/new-api:${VERSION}-amd64 \ - calciumion/new-api:${VERSION}-arm64 - - - name: Log in to GHCR - uses: docker/login-action@v3 - with: - registry: ghcr.io - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - - name: Create & push manifest (GHCR - alpha) - run: | - docker buildx imagetools create \ - -t ghcr.io/${GHCR_REPOSITORY}:alpha \ - ghcr.io/${GHCR_REPOSITORY}:alpha-amd64 \ - ghcr.io/${GHCR_REPOSITORY}:alpha-arm64 - - - name: Create & push manifest (GHCR - versioned alpha) - run: | - docker buildx imagetools create \ - -t ghcr.io/${GHCR_REPOSITORY}:${VERSION} \ - ghcr.io/${GHCR_REPOSITORY}:${VERSION}-amd64 \ - ghcr.io/${GHCR_REPOSITORY}:${VERSION}-arm64 diff --git a/.github/workflows/docker-image-arm64.yml b/.github/workflows/docker-image-arm64.yml deleted file mode 100644 index 78517af0ee2d..000000000000 --- a/.github/workflows/docker-image-arm64.yml +++ /dev/null @@ -1,138 +0,0 @@ -name: Publish Docker image (Multi Registries, native amd64+arm64) - -on: - push: - tags: - - '*' - -jobs: - build_single_arch: - name: Build & push (${{ matrix.arch }}) [native] - strategy: - fail-fast: false - matrix: - include: - - arch: amd64 - platform: linux/amd64 - runner: ubuntu-latest - - arch: arm64 - platform: linux/arm64 - runner: ubuntu-24.04-arm - runs-on: ${{ matrix.runner }} - - permissions: - packages: write - contents: read - - steps: - - name: Check out (shallow) - uses: actions/checkout@v4 - with: - fetch-depth: 1 - - - name: Resolve tag & write VERSION - run: | - git fetch --tags --force --depth=1 - TAG=${GITHUB_REF#refs/tags/} - echo "TAG=$TAG" >> $GITHUB_ENV - echo "$TAG" > VERSION - echo "Building tag: $TAG for ${{ matrix.arch }}" - - -# - name: Normalize GHCR repository -# run: echo "GHCR_REPOSITORY=${GITHUB_REPOSITORY,,}" >> $GITHUB_ENV - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 - - - name: Log in to Docker Hub - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKERHUB_USERNAME }} - password: ${{ secrets.DOCKERHUB_TOKEN }} - -# - name: Log in to GHCR -# uses: docker/login-action@v3 -# with: -# registry: ghcr.io -# username: ${{ github.actor }} -# password: ${{ secrets.GITHUB_TOKEN }} - - - name: Extract metadata (labels) - id: meta - uses: docker/metadata-action@v5 - with: - images: | - calciumion/new-api -# ghcr.io/${{ env.GHCR_REPOSITORY }} - - - name: Build & push single-arch (to both registries) - uses: docker/build-push-action@v6 - with: - context: . - platforms: ${{ matrix.platform }} - push: true - tags: | - calciumion/new-api:${{ env.TAG }}-${{ matrix.arch }} - calciumion/new-api:latest-${{ matrix.arch }} -# ghcr.io/${{ env.GHCR_REPOSITORY }}:${{ env.TAG }}-${{ matrix.arch }} -# ghcr.io/${{ env.GHCR_REPOSITORY }}:latest-${{ matrix.arch }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max - provenance: false - sbom: false - - create_manifests: - name: Create multi-arch manifests (Docker Hub) - needs: [build_single_arch] - runs-on: ubuntu-latest - if: startsWith(github.ref, 'refs/tags/') - steps: - - name: Extract tag - run: echo "TAG=${GITHUB_REF#refs/tags/}" >> $GITHUB_ENV -# -# - name: Normalize GHCR repository -# run: echo "GHCR_REPOSITORY=${GITHUB_REPOSITORY,,}" >> $GITHUB_ENV - - - name: Log in to Docker Hub - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKERHUB_USERNAME }} - password: ${{ secrets.DOCKERHUB_TOKEN }} - - - name: Create & push manifest (Docker Hub - version) - run: | - docker buildx imagetools create \ - -t calciumion/new-api:${TAG} \ - calciumion/new-api:${TAG}-amd64 \ - calciumion/new-api:${TAG}-arm64 - - - name: Create & push manifest (Docker Hub - latest) - run: | - docker buildx imagetools create \ - -t calciumion/new-api:latest \ - calciumion/new-api:latest-amd64 \ - calciumion/new-api:latest-arm64 - - # ---- GHCR ---- -# - name: Log in to GHCR -# uses: docker/login-action@v3 -# with: -# registry: ghcr.io -# username: ${{ github.actor }} -# password: ${{ secrets.GITHUB_TOKEN }} - -# - name: Create & push manifest (GHCR - version) -# run: | -# docker buildx imagetools create \ -# -t ghcr.io/${GHCR_REPOSITORY}:${TAG} \ -# ghcr.io/${GHCR_REPOSITORY}:${TAG}-amd64 \ -# ghcr.io/${GHCR_REPOSITORY}:${TAG}-arm64 -# -# - name: Create & push manifest (GHCR - latest) -# run: | -# docker buildx imagetools create \ -# -t ghcr.io/${GHCR_REPOSITORY}:latest \ -# ghcr.io/${GHCR_REPOSITORY}:latest-amd64 \ -# ghcr.io/${GHCR_REPOSITORY}:latest-arm64 diff --git a/.github/workflows/ghcr-publish.yml b/.github/workflows/ghcr-publish.yml new file mode 100644 index 000000000000..dbc9ba1531c4 --- /dev/null +++ b/.github/workflows/ghcr-publish.yml @@ -0,0 +1,68 @@ +name: Publish Docker image to GHCR + +on: + push: + tags: + - "v*" + workflow_dispatch: + +permissions: + contents: read + packages: write + +jobs: + build-and-push: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Detect semver tag + shell: bash + run: | + TAG="${GITHUB_REF#refs/tags/}" + if [[ "$TAG" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then + echo "IS_SEMVER=true" >> "$GITHUB_ENV" + else + echo "IS_SEMVER=false" >> "$GITHUB_ENV" + fi + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Log in to GHCR + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Docker metadata (tags, labels) + id: meta + uses: docker/metadata-action@v5 + with: + images: ghcr.io/${{ github.repository }} + tags: | + type=ref,event=tag + type=raw,value=latest,enable=${{ env.IS_SEMVER == 'true' }} + + - name: Build and push (amd64 only) + uses: docker/build-push-action@v6 + with: + context: . + file: ./Dockerfile + platforms: linux/amd64 + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha + cache-to: type=gha,mode=max + + # ============================================ + # 以下是原来的 QEMU 多架构构建配置(已注释) + # 如需恢复 arm64 支持,取消注释 platforms 行 + # ============================================ + # - name: Set up QEMU + # uses: docker/setup-qemu-action@v3 + # + # platforms: linux/amd64,linux/arm64 \ No newline at end of file diff --git a/Difference.md b/Difference.md new file mode 100644 index 000000000000..4107788167f9 --- /dev/null +++ b/Difference.md @@ -0,0 +1,10 @@ +# 和Newapi上游的区别 + +1. 实现维护模型健康度(健康度是一个时间的比值,每5分钟一个单位,如果该单位内只有失败的请求,那么记为失败时间片,如果有一个或多个成功请求并且(返回的byte长度大于1k或完成token大于2或实际响应模型回复大于2char),记为成功时间片,可查看不同小时时间段的模型成功率),实现后端和对应前端,,设计数据结构和新表实现良好性能。实现对非管理员隐藏可自定义模型和时间的查询(在控制台),并实现在导航栏添加新的页面(新页面所有用户即使非登录也可查看),显示所有模型最近24小时每小时的健康度。 +2. 若Newapi无法实现在对所有用户限速的情况下,使用标签解除对应的限速(而不是其他限速),那么添加管理员豁免用户限速RPM【已实现】 +3. 实现缓存最近100次API调用的请求和返回信息到内存里(包括报错,记录客户端原始请求和上游原始响应(包括上游原始流式响应)),提供UI查阅,实现后端和对应前端,设计数据结构和新表实现良好性能【已实现】 +4. 实现按可选定的小时统计用户的总调用次数(可分别显示成功次数或报错次数)并降序显示,实现后端和对应前端,设计数据结构和新表实现良好性能 +5. 实现生成随机兑换码(输入最小值和最大值,以及其他普通兑换码具有的字段,并且支持设置生成的兑换码前缀,生成随机的兑换码并提供文件下载),实现后端和对应前端 +6. 实现模型自定义配置没有将特定role转换为另一种role的功能的话实现它。不是全局模型映射,而是每个渠道一个配置,加入渠道额外设置 +7. 实现强制在日志记录IP,即使用户关闭IP记录【已实现】 +8. web\public\oauth-redirect.html 多站点重定向登录【已实现】 \ No newline at end of file diff --git a/PROJECT_CONTEXT.md b/PROJECT_CONTEXT.md new file mode 100644 index 000000000000..8f8a497b097a --- /dev/null +++ b/PROJECT_CONTEXT.md @@ -0,0 +1,11 @@ +# PROJECT_CONTEXT + +本文档用于快速理解仓库的“特点与环境信息(开发/构建/CI)”,便于后续维护与交接。 + +## 约定 + +- 约定:`tag++` 的默认含义 + - 先提交所有工作区 + - 执行 `git tag --sort=-creatordate | head -n 10` 拉取最近 10 个 tag + - 从中解析出“最新的可自增 tag”(优先 semver,例如 `vX.Y.Z`),将其 `+1`(默认补丁号 `Z+1`) + - 创建并推送新的 tag 到仓库(用于触发发布流水线) \ No newline at end of file diff --git a/VERSION b/VERSION index e69de29bb2d1..2696fdf55be2 100644 --- a/VERSION +++ b/VERSION @@ -0,0 +1 @@ +v0.1.29 \ No newline at end of file diff --git a/common/constants.go b/common/constants.go index e33a64b221fc..b353950445ae 100644 --- a/common/constants.go +++ b/common/constants.go @@ -71,6 +71,10 @@ var EmailLoginAuthServerList = []string{ var DebugEnabled bool var MemoryCacheEnabled bool +// HourlyCallRankCountFailedEnabled 控制“用户小时调用排行”是否把失败请求计入 total_calls。 +// 需求口径是“总调用次数”,默认应计入失败(只要发生调用就算)。 +var HourlyCallRankCountFailedEnabled = true + var LogConsumeEnabled = true var SMTPServer = "" diff --git a/controller/debug_recent_calls.go b/controller/debug_recent_calls.go new file mode 100644 index 000000000000..d9dda0598dcf --- /dev/null +++ b/controller/debug_recent_calls.go @@ -0,0 +1,53 @@ +package controller + +import ( + "net/http" + "strconv" + + "github.com/QuantumNous/new-api/service" + "github.com/gin-gonic/gin" +) + +func GetRecentCalls(c *gin.Context) { + limit := service.DefaultRecentCallsCapacity + if v := c.Query("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + limit = n + } + } + + var beforeID uint64 + if v := c.Query("before_id"); v != "" { + if n, err := strconv.ParseUint(v, 10, 64); err == nil { + beforeID = n + } + } + + items := service.RecentCallsCache().List(limit, beforeID) + c.JSON(http.StatusOK, gin.H{ + "data": items, + "limit": limit, + }) +} + +func GetRecentCallByID(c *gin.Context) { + id, err := strconv.ParseUint(c.Param("id"), 10, 64) + if err != nil || id == 0 { + c.JSON(http.StatusBadRequest, gin.H{ + "error": "invalid id", + }) + return + } + + rec, ok := service.RecentCallsCache().Get(id) + if !ok { + c.JSON(http.StatusNotFound, gin.H{ + "error": "not found", + }) + return + } + + c.JSON(http.StatusOK, gin.H{ + "data": rec, + }) +} \ No newline at end of file diff --git a/controller/hour_utils.go b/controller/hour_utils.go new file mode 100644 index 000000000000..988c2ef188bf --- /dev/null +++ b/controller/hour_utils.go @@ -0,0 +1,36 @@ +package controller + +import ( + "sort" + "strconv" + "strings" +) + +func parseHourListParam(raw string) ([]int64, bool, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, false, nil + } + parts := strings.Split(raw, ",") + hours := make([]int64, 0, len(parts)) + for _, p := range parts { + p = strings.TrimSpace(p) + if p == "" { + continue + } + ts, err := strconv.ParseInt(p, 10, 64) + if err != nil { + return nil, true, err + } + hours = append(hours, ts) + } + if len(hours) == 0 { + return nil, false, nil + } + sort.Slice(hours, func(i, j int) bool { return hours[i] < hours[j] }) + return hours, true, nil +} + +func isAlignedHour(ts int64) bool { + return ts > 0 && ts%3600 == 0 +} \ No newline at end of file diff --git a/controller/model.go b/controller/model.go index aa6c6e2b9db7..a72f8a1a76bf 100644 --- a/controller/model.go +++ b/controller/model.go @@ -3,6 +3,7 @@ package controller import ( "fmt" "net/http" + "sort" "time" "github.com/QuantumNous/new-api/common" @@ -247,10 +248,31 @@ func ChannelListModels(c *gin.Context) { }) } +func flattenChannelModels(modelsByChannelType map[int][]string) []string { + if len(modelsByChannelType) == 0 { + return []string{} + } + flat := make([]string, 0) + for _, models := range modelsByChannelType { + for _, name := range models { + if name == "" { + continue + } + flat = append(flat, name) + } + } + flat = lo.Uniq(flat) + sort.Strings(flat) + return flat +} + func DashboardListModels(c *gin.Context) { + // Frontend expects an array, not a map keyed by channel type. + // Keep the grouped data as an extra field for backward/diagnostic usage. c.JSON(200, gin.H{ "success": true, - "data": channelId2Models, + "data": flattenChannelModels(channelId2Models), + "grouped": channelId2Models, }) } diff --git a/controller/model_health.go b/controller/model_health.go new file mode 100644 index 000000000000..201cfe1dc03b --- /dev/null +++ b/controller/model_health.go @@ -0,0 +1,257 @@ +package controller + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + + "github.com/gin-gonic/gin" +) + +// 公共模型健康度缓存配置 +const ( + publicModelHealthCacheKey = "public_model_health:hourly_last24h" + publicModelHealthCacheTTL = 5 * time.Minute // 缓存 5 分钟 +) + +// 内存缓存(当 Redis 不可用时使用) +var ( + publicModelHealthMemCache *publicModelHealthCacheData + publicModelHealthMemCacheLock sync.RWMutex +) + +type publicModelHealthCacheData struct { + Data interface{} + ExpireAt time.Time +} + +type modelHealthHourlyRespItem struct { + ModelName string `json:"model_name"` + HourStartTs int64 `json:"hour_start_ts"` + SuccessSlices int64 `json:"success_slices"` + TotalSlices int64 `json:"total_slices"` + SuccessRate float64 `json:"success_rate"` +} + +type publicModelsHealthHourlyLast24hRespItem struct { + ModelName string `json:"model_name"` + HourStartTs int64 `json:"hour_start_ts"` + SuccessSlices int64 `json:"success_slices"` + TotalSlices int64 `json:"total_slices"` + SuccessRate float64 `json:"success_rate"` +} + +// GetModelHealthHourlyStatsAPI 查询模型在小时 bucket 上的健康度(success_slices/total_slices/success_rate)。 +// 参数: +// - model_name: string (required) +// - start_hour: unix seconds, aligned to 3600 (optional when hours provided) +// - end_hour: unix seconds, aligned to 3600, exclusive (optional when hours provided) +// - hours: comma separated unix seconds list, aligned to 3600 (optional) +func GetModelHealthHourlyStatsAPI(c *gin.Context) { + modelName := strings.TrimSpace(c.Query("model_name")) + if modelName == "" { + c.JSON(http.StatusOK, gin.H{"success": false, "message": "model_name is required"}) + return + } + + hours, hasHours, err := parseHourListParam(c.Query("hours")) + if err != nil { + common.ApiError(c, err) + return + } + + var startHourTs int64 + var endHourTs int64 + if hasHours { + for _, h := range hours { + if !isAlignedHour(h) { + c.JSON(http.StatusOK, gin.H{"success": false, "message": "hours must be aligned to hour (ts % 3600 == 0)"}) + return + } + } + startHourTs = hours[0] + endHourTs = hours[len(hours)-1] + 3600 + } else { + startHourTs, _ = strconv.ParseInt(c.Query("start_hour"), 10, 64) + endHourTs, _ = strconv.ParseInt(c.Query("end_hour"), 10, 64) + if !isAlignedHour(startHourTs) || !isAlignedHour(endHourTs) || endHourTs <= startHourTs { + c.JSON(http.StatusOK, gin.H{"success": false, "message": "invalid hour range, require start_hour/end_hour aligned to hour and end_hour > start_hour"}) + return + } + // limit range to 31 days to avoid large scan (best-effort guardrail) + if endHourTs-startHourTs > 31*24*3600 { + c.JSON(http.StatusOK, gin.H{"success": false, "message": "hour range too large (max 31 days)"}) + return + } + } + + rows, err := model.GetModelHealthHourlyStats(model.DB, modelName, startHourTs, endHourTs) + if err != nil { + common.ApiError(c, err) + return + } + + rowMap := make(map[int64]model.ModelHealthHourlyStat, len(rows)) + for _, r := range rows { + rowMap[r.HourStartTs] = r + } + + var wantHours []int64 + if hasHours { + wantHours = hours + } else { + count := int((endHourTs - startHourTs) / 3600) + wantHours = make([]int64, 0, count) + for h := startHourTs; h < endHourTs; h += 3600 { + wantHours = append(wantHours, h) + } + } + + resp := make([]modelHealthHourlyRespItem, 0, len(wantHours)) + for _, h := range wantHours { + if stat, ok := rowMap[h]; ok { + resp = append(resp, modelHealthHourlyRespItem{ + ModelName: stat.ModelName, + HourStartTs: stat.HourStartTs, + SuccessSlices: stat.SuccessSlices, + TotalSlices: stat.TotalSlices, + SuccessRate: stat.SuccessRate, + }) + continue + } + resp = append(resp, modelHealthHourlyRespItem{ + ModelName: modelName, + HourStartTs: h, + SuccessSlices: 0, + TotalSlices: 0, + SuccessRate: 0, + }) + } + + common.ApiSuccess(c, resp) +} + +// GetPublicModelsHealthHourlyLast24hAPI 公共接口:查询所有模型最近 24 小时每小时健康度。 +// GET /api/public/model_health/hourly_last24h +// 支持 Redis 缓存和内存缓存 +func GetPublicModelsHealthHourlyLast24hAPI(c *gin.Context) { + // 尝试从缓存获取 + if cachedData, ok := getPublicModelHealthCache(); ok { + common.ApiSuccess(c, cachedData) + return + } + + // 缓存未命中,从数据库查询 + now := time.Now().Unix() + endHourTs := now - (now % 3600) + 3600 // exclusive, aligned to next hour + startHourTs := endHourTs - 24*3600 + + rows, err := model.GetAllModelsHealthHourlyStats(model.DB, startHourTs, endHourTs) + if err != nil { + common.ApiError(c, err) + return + } + + // Fill missing hours per model with zeros for stable UI rendering. + // Build desired hours list + wantHours := make([]int64, 0, 24) + for h := startHourTs; h < endHourTs; h += 3600 { + wantHours = append(wantHours, h) + } + + // Group by model_name + grouped := make(map[string]map[int64]model.ModelHealthHourlyStat) + modelOrder := make([]string, 0) + for _, r := range rows { + if _, ok := grouped[r.ModelName]; !ok { + grouped[r.ModelName] = make(map[int64]model.ModelHealthHourlyStat) + modelOrder = append(modelOrder, r.ModelName) + } + grouped[r.ModelName][r.HourStartTs] = r + } + + resp := make([]publicModelsHealthHourlyLast24hRespItem, 0, len(modelOrder)*len(wantHours)) + for _, modelName := range modelOrder { + hourMap := grouped[modelName] + for _, h := range wantHours { + if stat, ok := hourMap[h]; ok { + resp = append(resp, publicModelsHealthHourlyLast24hRespItem{ + ModelName: stat.ModelName, + HourStartTs: stat.HourStartTs, + SuccessSlices: stat.SuccessSlices, + TotalSlices: stat.TotalSlices, + SuccessRate: stat.SuccessRate, + }) + continue + } + resp = append(resp, publicModelsHealthHourlyLast24hRespItem{ + ModelName: modelName, + HourStartTs: h, + SuccessSlices: 0, + TotalSlices: 0, + SuccessRate: 0, + }) + } + } + + result := gin.H{ + "start_hour": startHourTs, + "end_hour": endHourTs, + "rows": resp, + } + + // 存入缓存 + setPublicModelHealthCache(result) + + common.ApiSuccess(c, result) +} + +// getPublicModelHealthCache 从缓存获取公共模型健康度数据 +func getPublicModelHealthCache() (interface{}, bool) { + // 优先使用 Redis 缓存 + if common.RedisEnabled { + cached, err := common.RedisGet(publicModelHealthCacheKey) + if err == nil && cached != "" { + var data map[string]interface{} + if err := json.Unmarshal([]byte(cached), &data); err == nil { + return data, true + } + } + } + + // 回退到内存缓存 + publicModelHealthMemCacheLock.RLock() + defer publicModelHealthMemCacheLock.RUnlock() + + if publicModelHealthMemCache != nil && time.Now().Before(publicModelHealthMemCache.ExpireAt) { + return publicModelHealthMemCache.Data, true + } + + return nil, false +} + +// setPublicModelHealthCache 将公共模型健康度数据存入缓存 +func setPublicModelHealthCache(data interface{}) { + // 存入 Redis 缓存 + if common.RedisEnabled { + jsonData, err := json.Marshal(data) + if err == nil { + _ = common.RedisSet(publicModelHealthCacheKey, string(jsonData), publicModelHealthCacheTTL) + } + } + + // 同时存入内存缓存(作为备份) + publicModelHealthMemCacheLock.Lock() + defer publicModelHealthMemCacheLock.Unlock() + + publicModelHealthMemCache = &publicModelHealthCacheData{ + Data: data, + ExpireAt: time.Now().Add(publicModelHealthCacheTTL), + } +} \ No newline at end of file diff --git a/controller/model_test.go b/controller/model_test.go new file mode 100644 index 000000000000..7a8a902e3e9c --- /dev/null +++ b/controller/model_test.go @@ -0,0 +1,46 @@ +package controller + +import ( + "encoding/json" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" +) + +func TestDashboardListModels_DataIsArray(t *testing.T) { + gin.SetMode(gin.TestMode) + + orig := channelId2Models + channelId2Models = map[int][]string{ + 1: {"gpt-4o", "gpt-4o-mini"}, + 4: {"llama3-7b"}, + } + t.Cleanup(func() { + channelId2Models = orig + }) + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + + DashboardListModels(c) + + var resp struct { + Success bool `json:"success"` + Data json.RawMessage `json:"data"` + } + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("unmarshal response: %v; body=%s", err, w.Body.String()) + } + if !resp.Success { + t.Fatalf("expected success=true; body=%s", w.Body.String()) + } + + var dataAny any + if err := json.Unmarshal(resp.Data, &dataAny); err != nil { + t.Fatalf("unmarshal data: %v; data=%s", err, string(resp.Data)) + } + if _, ok := dataAny.([]any); !ok { + t.Fatalf("expected data to be JSON array; got %T; data=%s", dataAny, string(resp.Data)) + } +} \ No newline at end of file diff --git a/controller/option.go b/controller/option.go index 89b2fc4d52a4..9c5b2d332344 100644 --- a/controller/option.go +++ b/controller/option.go @@ -173,6 +173,15 @@ func UpdateOption(c *gin.Context) { }) return } + case "ModelRequestRateLimitExemptUserIDs": + _, err = setting.ParseModelRequestRateLimitExemptUserIDs(option.Value.(string)) + if err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": err.Error(), + }) + return + } case "console_setting.api_info": err = console_setting.ValidateConsoleSettings(option.Value.(string), "ApiInfo") if err != nil { diff --git a/controller/redemption.go b/controller/redemption.go index 945cefa35358..c68fba331a5a 100644 --- a/controller/redemption.go +++ b/controller/redemption.go @@ -1,12 +1,17 @@ package controller import ( + crand "crypto/rand" "errors" + "fmt" + "math/big" "net/http" "strconv" + "strings" "unicode/utf8" "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/dto" "github.com/QuantumNous/new-api/model" "github.com/gin-gonic/gin" @@ -59,47 +64,108 @@ func GetRedemption(c *gin.Context) { } func AddRedemption(c *gin.Context) { - redemption := model.Redemption{} - err := c.ShouldBindJSON(&redemption) + req := dto.CreateRedemptionRequest{} + err := c.ShouldBindJSON(&req) if err != nil { common.ApiError(c, err) return } - if utf8.RuneCountInString(redemption.Name) == 0 || utf8.RuneCountInString(redemption.Name) > 20 { + + if utf8.RuneCountInString(req.Name) == 0 || utf8.RuneCountInString(req.Name) > 20 { c.JSON(http.StatusOK, gin.H{ "success": false, "message": "兑换码名称长度必须在1-20之间", }) return } - if redemption.Count <= 0 { + + count := req.EffectiveCount() + if count <= 0 { c.JSON(http.StatusOK, gin.H{ "success": false, "message": "兑换码个数必须大于0", }) return } - if redemption.Count > 100 { + + const maxCount = 100 + + if count > maxCount { c.JSON(http.StatusOK, gin.H{ "success": false, - "message": "一次兑换码批量生成的个数不能大于 100", + "message": fmt.Sprintf("一次兑换码批量生成的个数不能大于 %d", maxCount), }) return } - if err := validateExpiredTime(redemption.ExpiredTime); err != nil { + + // 随机额度模式校验 + if req.RandomQuotaMode() { + if req.QuotaMin == nil || req.QuotaMax == nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "启用随机额度需同时提供 quota_min 与 quota_max", + }) + return + } + if *req.QuotaMin <= 0 || *req.QuotaMax <= 0 { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "quota_min 和 quota_max 必须大于 0", + }) + return + } + if *req.QuotaMin > *req.QuotaMax { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "quota_min 不能大于 quota_max", + }) + return + } + } else { + if req.Quota <= 0 { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "额度必须大于 0", + }) + return + } + } + + if err := validateExpiredTime(req.ExpiredTime); err != nil { c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) return } - var keys []string - for i := 0; i < redemption.Count; i++ { - key := common.GetUUID() + + keys := make([]string, 0, count) + keyPrefix := strings.TrimSpace(req.KeyPrefix) + + for i := 0; i < count; i++ { + // 生成 Key:前缀 + UUID + key := keyPrefix + common.GetUUID() + + // 确定额度:随机模式或固定模式 + quota := req.Quota + if req.RandomQuotaMode() { + randomQuota, err := cryptoRandIntInclusive(*req.QuotaMin, *req.QuotaMax) + if err != nil { + c.JSON(http.StatusOK, gin.H{ + "success": false, + "message": "生成随机额度失败: " + err.Error(), + "data": keys, + "keys": keys, + }) + return + } + quota = randomQuota + } + cleanRedemption := model.Redemption{ UserId: c.GetInt("id"), - Name: redemption.Name, + Name: req.Name, Key: key, CreatedTime: common.GetTimestamp(), - Quota: redemption.Quota, - ExpiredTime: redemption.ExpiredTime, + Quota: quota, + ExpiredTime: req.ExpiredTime, } err = cleanRedemption.Insert() if err != nil { @@ -107,15 +173,18 @@ func AddRedemption(c *gin.Context) { "success": false, "message": err.Error(), "data": keys, + "keys": keys, }) return } keys = append(keys, key) } + c.JSON(http.StatusOK, gin.H{ "success": true, "message": "", "data": keys, + "keys": keys, }) return } @@ -193,3 +262,35 @@ func validateExpiredTime(expired int64) error { } return nil } + +func cryptoRandIntInclusive(min int, max int) (int, error) { + if min > max { + return 0, errors.New("invalid range: min > max") + } + rangeSize := new(big.Int).SetInt64(int64(max - min + 1)) + n, err := crand.Int(crand.Reader, rangeSize) + if err != nil { + return 0, err + } + return int(n.Int64()) + min, nil +} + +func isUniqueConstraintError(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + // MySQL: Error 1062 / Duplicate entry + if strings.Contains(msg, "error 1062") || strings.Contains(msg, "duplicate entry") { + return true + } + // PostgreSQL: SQLSTATE 23505 / duplicate key value violates unique constraint + if strings.Contains(msg, "sqlstate 23505") || strings.Contains(msg, "duplicate key value violates unique constraint") { + return true + } + // SQLite: UNIQUE constraint failed + if strings.Contains(msg, "unique constraint failed") { + return true + } + return false +} diff --git a/controller/relay.go b/controller/relay.go index 9759fa30ce5a..ea081d42bb3f 100644 --- a/controller/relay.go +++ b/controller/relay.go @@ -114,6 +114,9 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) { return } + // Keep an original snapshot so we can re-apply per-channel role mappings on retries. + roleSnapshot := service.SnapshotRequestRoles(request) + relayInfo, err := relaycommon.GenRelayInfo(c, relayFormat, request, ws) if err != nil { newAPIError = types.NewError(err, types.ErrorCodeGenRelayInfoFailed) @@ -186,6 +189,10 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) { break } + // Restore original roles then apply per-channel mappings (channel settings are available after selection). + service.RestoreRequestRoles(request, roleSnapshot) + service.ApplyModelRoleMappingsToRequest(c, request) + addUsedChannel(c, channel.Id) requestBody, bodyErr := common.GetRequestBody(c) if bodyErr != nil { @@ -197,6 +204,11 @@ func Relay(c *gin.Context, relayFormat types.RelayFormat) { } break } + + if _, ok := c.Get(service.RecentCallsContextKeyID); !ok { + service.RecentCallsCache().BeginFromContext(c, relayInfo, requestBody) + } + c.Request.Body = io.NopCloser(bytes.NewBuffer(requestBody)) switch relayFormat { @@ -344,6 +356,9 @@ func shouldRetry(c *gin.Context, openaiErr *types.NewAPIError, retryTimes int) b func processChannelError(c *gin.Context, channelError types.ChannelError, err *types.NewAPIError) { logger.LogError(c, fmt.Sprintf("channel error (channel #%d, status code: %d): %s", channelError.ChannelId, err.StatusCode, err.Error())) + + service.RecentCallsCache().UpsertErrorByContext(c, err.MaskSensitiveError(), fmt.Sprint(err.GetErrorType()), fmt.Sprint(err.GetErrorCode()), err.StatusCode) + // 不要使用context获取渠道信息,异步处理时可能会出现渠道信息不一致的情况 // do not use context to get channel info, there may be inconsistent channel info when processing asynchronously if service.ShouldDisableChannel(channelError.ChannelType, err) && channelError.AutoBan { diff --git a/controller/user_rank.go b/controller/user_rank.go new file mode 100644 index 000000000000..03139d4f1dfe --- /dev/null +++ b/controller/user_rank.go @@ -0,0 +1,71 @@ +package controller + +import ( + "net/http" + "strconv" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/model" + + "github.com/gin-gonic/gin" +) + +type userHourlyCallsRankRespItem struct { + UserId int `json:"user_id"` + Username string `json:"username,omitempty"` + TotalCalls int64 `json:"total_calls"` +} + +// GetUserHourlyCallsRankAPI +// GET /api/user_rank/hourly_calls +// 参数: +// - hours: unix 秒整点列表(逗号分隔) +// - start_hour/end_hour: unix 秒整点,end 开区间(当 hours 未提供时使用) +// - limit: 默认 50,最大 500 +func GetUserHourlyCallsRankAPI(c *gin.Context) { + limit, _ := strconv.Atoi(c.Query("limit")) + if limit <= 0 { + limit = 50 + } + + hours, hasHours, err := parseHourListParam(c.Query("hours")) + if err != nil { + common.ApiError(c, err) + return + } + + var startHourTs int64 + var endHourTs int64 + if hasHours { + normalized, err := model.NormalizeHourList(hours) + if err != nil { + c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) + return + } + hours = normalized + } else { + startHourTs, _ = strconv.ParseInt(c.Query("start_hour"), 10, 64) + endHourTs, _ = strconv.ParseInt(c.Query("end_hour"), 10, 64) + if !isAlignedHour(startHourTs) || !isAlignedHour(endHourTs) || endHourTs <= startHourTs { + c.JSON(http.StatusOK, gin.H{"success": false, "message": "invalid hour range, require start_hour/end_hour aligned to hour and end_hour > start_hour"}) + return + } + } + + rows, err := model.GetUserHourlyCallsRank(model.DB, hours, startHourTs, endHourTs, limit) + if err != nil { + common.ApiError(c, err) + return + } + + resp := make([]userHourlyCallsRankRespItem, 0, len(rows)) + for _, r := range rows { + resp = append(resp, userHourlyCallsRankRespItem{ + UserId: r.UserId, + Username: r.Username, + TotalCalls: r.TotalCalls, + }) + } + + common.ApiSuccess(c, resp) +} \ No newline at end of file diff --git a/dto/channel_settings.go b/dto/channel_settings.go index e88f2235e330..cf504a52cc7a 100644 --- a/dto/channel_settings.go +++ b/dto/channel_settings.go @@ -1,5 +1,63 @@ package dto +import ( + "bytes" + "encoding/json" +) + +// ModelRoleMappingsField supports both object form and "json string" form: +// +// 1) Object: { "gpt-4o": { "system": "developer" } } +// 2) String: "{\"gpt-4o\":{\"system\":\"developer\"}}" +// +// It also tolerates legacy object: { "system": "developer" } which will be treated as wildcard prefix "*". +type ModelRoleMappingsField map[string]map[string]string + +func (m *ModelRoleMappingsField) UnmarshalJSON(data []byte) error { + data = bytes.TrimSpace(data) + if len(data) == 0 || bytes.Equal(data, []byte("null")) { + *m = nil + return nil + } + + // If it's a JSON string, parse the inner JSON. + if len(data) > 0 && data[0] == '"' { + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + sBytes := bytes.TrimSpace([]byte(s)) + if len(sBytes) == 0 { + *m = nil + return nil + } + return m.UnmarshalJSON(sBytes) + } + + // First try the desired shape: map[string]map[string]string + var nested map[string]map[string]string + if err := json.Unmarshal(data, &nested); err == nil { + *m = ModelRoleMappingsField(nested) + return nil + } + + // Then try legacy shape: map[string]string (apply to all models via wildcard "*") + var flat map[string]string + if err := json.Unmarshal(data, &flat); err == nil { + *m = ModelRoleMappingsField(map[string]map[string]string{ + "*": flat, + }) + return nil + } + + // Return the original error for better diagnostics + return json.Unmarshal(data, &nested) +} + +func (m ModelRoleMappingsField) MarshalJSON() ([]byte, error) { + return json.Marshal(map[string]map[string]string(m)) +} + type ChannelSettings struct { ForceFormat bool `json:"force_format,omitempty"` ThinkingToContent bool `json:"thinking_to_content,omitempty"` @@ -7,6 +65,9 @@ type ChannelSettings struct { PassThroughBodyEnabled bool `json:"pass_through_body_enabled,omitempty"` SystemPrompt string `json:"system_prompt,omitempty"` SystemPromptOverride bool `json:"system_prompt_override,omitempty"` + + // per-channel role mapping: { [modelPrefix]: { [fromRole]: toRole } } + ModelRoleMappings ModelRoleMappingsField `json:"model_role_mappings,omitempty"` } type VertexKeyType string diff --git a/dto/redemption.go b/dto/redemption.go new file mode 100644 index 000000000000..2f153e01737a --- /dev/null +++ b/dto/redemption.go @@ -0,0 +1,32 @@ +package dto + +type CreateRedemptionRequest struct { + Name string `json:"name"` + Quota int `json:"quota"` + ExpiredTime int64 `json:"expired_time"` + + // Backward-compatible count. + Count int `json:"count"` + + // Key prefix for generated redemption codes (e.g., "VIP-"). + KeyPrefix string `json:"key_prefix"` + + // Random quota mode: generate redemption codes with random quota in [QuotaMin, QuotaMax]. + RandomQuotaEnabled *bool `json:"random_quota_enabled"` + QuotaMin *int `json:"quota_min"` + QuotaMax *int `json:"quota_max"` +} + +func (r CreateRedemptionRequest) EffectiveCount() int { + if r.Count <= 0 { + return 1 + } + return r.Count +} + +func (r CreateRedemptionRequest) RandomQuotaMode() bool { + if r.RandomQuotaEnabled != nil && *r.RandomQuotaEnabled { + return true + } + return r.QuotaMin != nil && r.QuotaMax != nil +} \ No newline at end of file diff --git a/go.mod b/go.mod index 4b5d63e49332..4c32b14585f2 100644 --- a/go.mod +++ b/go.mod @@ -1,7 +1,7 @@ module github.com/QuantumNous/new-api // +heroku goVersion go1.18 -go 1.25.1 +go 1.25.0 require ( github.com/Calcium-Ion/go-epay v0.0.4 diff --git a/middleware/logger.go b/middleware/logger.go index b4ed8c89d7ee..1255195acef1 100644 --- a/middleware/logger.go +++ b/middleware/logger.go @@ -2,25 +2,95 @@ package middleware import ( "fmt" + "net" + "net/http" + "strings" "github.com/QuantumNous/new-api/common" "github.com/gin-gonic/gin" ) +const requestLogClientIPKey = "client_ip" + func SetUpLogger(server *gin.Engine) { server.Use(gin.LoggerWithFormatter(func(param gin.LogFormatterParams) string { var requestID string if param.Keys != nil { requestID = param.Keys[common.RequestIdKey].(string) } + + clientIP := extractClientIP(param.Request) + if clientIP == "" { + clientIP = param.ClientIP + } + if clientIP == "" { + clientIP = "unknown" + } + return fmt.Sprintf("[GIN] %s | %s | %3d | %13v | %15s | %7s %s\n", param.TimeStamp.Format("2006/01/02 - 15:04:05"), requestID, param.StatusCode, param.Latency, - param.ClientIP, + requestLogClientIPKey+"="+clientIP, param.Method, param.Path, ) })) } + +func extractClientIP(r *http.Request) string { + if r == nil { + return "" + } + + if xff := strings.TrimSpace(r.Header.Get("X-Forwarded-For")); xff != "" { + first := strings.TrimSpace(strings.Split(xff, ",")[0]) + if ip := normalizeIP(first); ip != "" { + return ip + } + } + + if xrip := strings.TrimSpace(r.Header.Get("X-Real-IP")); xrip != "" { + if ip := normalizeIP(xrip); ip != "" { + return ip + } + } + + host, _, err := net.SplitHostPort(strings.TrimSpace(r.RemoteAddr)) + if err == nil { + if ip := normalizeIP(host); ip != "" { + return ip + } + } + + if ip := normalizeIP(strings.TrimSpace(r.RemoteAddr)); ip != "" { + return ip + } + + return "" +} + +func normalizeIP(v string) string { + v = strings.TrimSpace(v) + if v == "" { + return "" + } + + if strings.HasPrefix(v, "[") && strings.Contains(v, "]") { + v = strings.TrimPrefix(v, "[") + v = strings.SplitN(v, "]", 2)[0] + } + + if strings.Contains(v, ":") && !strings.Contains(v, ".") { + if ip := net.ParseIP(v); ip != nil { + return ip.String() + } + } + + if ip := net.ParseIP(v); ip != nil { + return ip.String() + } + + return "" +} diff --git a/middleware/model-rate-limit.go b/middleware/model-rate-limit.go index 80a3995df097..258659aa5e23 100644 --- a/middleware/model-rate-limit.go +++ b/middleware/model-rate-limit.go @@ -172,6 +172,13 @@ func ModelRequestRateLimit() func(c *gin.Context) { return } + userID := c.GetInt("id") + if setting.IsModelRequestRateLimitExemptUser(userID) { + c.Header("X-RateLimit-Bypass", "ModelRequestRateLimit") + c.Next() + return + } + // 计算限流参数 duration := int64(setting.ModelRequestRateLimitDurationMinutes * 60) totalMaxCount := setting.ModelRequestRateLimitCount diff --git a/model/log.go b/model/log.go index 7495d647d0aa..dbbcfb01d8fb 100644 --- a/model/log.go +++ b/model/log.go @@ -101,13 +101,6 @@ func RecordErrorLog(c *gin.Context, userId int, channelId int, modelName string, logger.LogInfo(c, fmt.Sprintf("record error log: userId=%d, channelId=%d, modelName=%s, tokenName=%s, content=%s", userId, channelId, modelName, tokenName, content)) username := c.GetString("username") otherStr := common.MapToJsonStr(other) - // 判断是否需要记录 IP - needRecordIp := false - if settingMap, err := GetUserSetting(userId, false); err == nil { - if settingMap.RecordIpLog { - needRecordIp = true - } - } log := &Log{ UserId: userId, Username: username, @@ -125,10 +118,10 @@ func RecordErrorLog(c *gin.Context, userId int, channelId int, modelName string, IsStream: isStream, Group: group, Ip: func() string { - if needRecordIp { - return c.ClientIP() + if c == nil { + return "" } - return "" + return c.ClientIP() }(), Other: otherStr, } @@ -136,6 +129,23 @@ func RecordErrorLog(c *gin.Context, userId int, channelId int, modelName string, if err != nil { logger.LogError(c, "failed to record log: "+err.Error()) } + + RecordModelHealthEventAsync(c, &ModelHealthEvent{ + ModelName: modelName, + CreatedAt: log.CreatedAt, + IsError: true, + }) + + // 需求口径:总调用次数默认包含失败(只要发生调用就算)。 + // 但并非所有失败都会落 error log(受 types.IsRecordErrorLog 等影响),因此这里是 best-effort。 + if common.HourlyCallRankCountFailedEnabled { + RecordUserCallHourlyEventAsync(c, &UserCallHourlyEvent{ + UserId: userId, + Username: username, + CreatedAt: log.CreatedAt, + IsError: true, + }) + } } type RecordConsumeLogParams struct { @@ -160,13 +170,6 @@ func RecordConsumeLog(c *gin.Context, userId int, params RecordConsumeLogParams) logger.LogInfo(c, fmt.Sprintf("record consume log: userId=%d, params=%s", userId, common.GetJsonString(params))) username := c.GetString("username") otherStr := common.MapToJsonStr(params.Other) - // 判断是否需要记录 IP - needRecordIp := false - if settingMap, err := GetUserSetting(userId, false); err == nil { - if settingMap.RecordIpLog { - needRecordIp = true - } - } log := &Log{ UserId: userId, Username: username, @@ -184,10 +187,10 @@ func RecordConsumeLog(c *gin.Context, userId int, params RecordConsumeLogParams) IsStream: params.IsStream, Group: params.Group, Ip: func() string { - if needRecordIp { - return c.ClientIP() + if c == nil { + return "" } - return "" + return c.ClientIP() }(), Other: otherStr, } @@ -200,6 +203,28 @@ func RecordConsumeLog(c *gin.Context, userId int, params RecordConsumeLogParams) LogQuotaData(userId, username, params.ModelName, params.Quota, common.GetTimestamp(), params.PromptTokens+params.CompletionTokens) }) } + + responseBytes := 0 + assistantChars := 0 + if c != nil { + responseBytes = c.GetInt("response_bytes") + assistantChars = c.GetInt("assistant_content_chars") + } + RecordModelHealthEventAsync(c, &ModelHealthEvent{ + ModelName: params.ModelName, + CreatedAt: log.CreatedAt, + IsError: false, + ResponseBytes: responseBytes, + CompletionTokens: params.CompletionTokens, + AssistantChars: assistantChars, + }) + + RecordUserCallHourlyEventAsync(c, &UserCallHourlyEvent{ + UserId: userId, + Username: username, + CreatedAt: log.CreatedAt, + IsError: false, + }) } func GetAllLogs(logType int, startTimestamp int64, endTimestamp int64, modelName string, username string, tokenName string, startIdx int, num int, channel int, group string) (logs []*Log, total int64, err error) { diff --git a/model/main.go b/model/main.go index 04842f13f5bc..fc97640d6fe2 100644 --- a/model/main.go +++ b/model/main.go @@ -267,6 +267,8 @@ func migrateDB() error { &Setup{}, &TwoFA{}, &TwoFABackupCode{}, + &ModelHealthSlice5m{}, + &UserCallHourly{}, ) if err != nil { return err diff --git a/model/model_health_query.go b/model/model_health_query.go new file mode 100644 index 000000000000..075833fd9766 --- /dev/null +++ b/model/model_health_query.go @@ -0,0 +1,93 @@ +package model + +import ( + "fmt" + + "gorm.io/gorm" +) + +type ModelHealthHourlyStat struct { + ModelName string `json:"model_name"` + HourStartTs int64 `json:"hour_start_ts"` + SuccessSlices int64 `json:"success_slices"` + TotalSlices int64 `json:"total_slices"` + SuccessRate float64 `json:"success_rate"` +} + +func hourStartExprSQL(db *gorm.DB) string { + // Align 5m slice timestamp (seconds) to hour start, and keep the result INTEGER. + // Notes: + // - MySQL: `/` is floating division; use `DIV` for integer division. + // - SQLite: `/` returns REAL; cast back to INTEGER. + // - Postgres: int/int is integer division. + if db != nil && db.Dialector != nil { + switch db.Dialector.Name() { + case "mysql": + return "((slice_start_ts DIV 3600) * 3600)" + case "sqlite": + return "(CAST((slice_start_ts / 3600) AS INTEGER) * 3600)" + } + } + return "((slice_start_ts / 3600) * 3600)" +} + +func successRateExprSQL() string { + // Force float division across DBs (Postgres int/int would otherwise truncate). + return "CASE WHEN COUNT(*) = 0 THEN 0 ELSE (1.0 * SUM(has_success_qualified)) / COUNT(*) END" +} + +func GetModelHealthHourlyStats(db *gorm.DB, modelName string, startHourTs int64, endHourTs int64) ([]ModelHealthHourlyStat, error) { + if db == nil { + return nil, fmt.Errorf("db is nil") + } + if modelName == "" { + return nil, fmt.Errorf("model_name is required") + } + if startHourTs <= 0 || endHourTs <= 0 || endHourTs <= startHourTs { + return nil, fmt.Errorf("invalid hour range") + } + + var rows []ModelHealthHourlyStat + err := db.Table((&ModelHealthSlice5m{}).TableName()). + Select(fmt.Sprintf(` +model_name as model_name, +%s as hour_start_ts, +SUM(has_success_qualified) as success_slices, +COUNT(*) as total_slices, +%s as success_rate`, hourStartExprSQL(db), successRateExprSQL())). + Where("model_name = ?", modelName). + Where("slice_start_ts >= ? AND slice_start_ts < ?", startHourTs, endHourTs). + Group("model_name, hour_start_ts"). + Order("hour_start_ts ASC"). + Scan(&rows).Error + if err != nil { + return nil, err + } + return rows, nil +} + +func GetAllModelsHealthHourlyStats(db *gorm.DB, startHourTs int64, endHourTs int64) ([]ModelHealthHourlyStat, error) { + if db == nil { + return nil, fmt.Errorf("db is nil") + } + if startHourTs <= 0 || endHourTs <= 0 || endHourTs <= startHourTs { + return nil, fmt.Errorf("invalid hour range") + } + + var rows []ModelHealthHourlyStat + err := db.Table((&ModelHealthSlice5m{}).TableName()). + Select(fmt.Sprintf(` +model_name as model_name, +%s as hour_start_ts, +SUM(has_success_qualified) as success_slices, +COUNT(*) as total_slices, +%s as success_rate`, hourStartExprSQL(db), successRateExprSQL())). + Where("slice_start_ts >= ? AND slice_start_ts < ?", startHourTs, endHourTs). + Group("model_name, hour_start_ts"). + Order("model_name ASC, hour_start_ts ASC"). + Scan(&rows).Error + if err != nil { + return nil, err + } + return rows, nil +} \ No newline at end of file diff --git a/model/model_health_slice.go b/model/model_health_slice.go new file mode 100644 index 000000000000..f4b923421b8a --- /dev/null +++ b/model/model_health_slice.go @@ -0,0 +1,126 @@ +package model + +import ( + "context" + "errors" + "time" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +const ( + modelHealthSliceSeconds = int64(300) +) + +type ModelHealthSlice5m struct { + SliceStartTs int64 `json:"slice_start_ts" gorm:"primaryKey;autoIncrement:false;index:idx_slice_start;index:idx_slice_model,priority:1;comment:slice start unix seconds, aligned to 300s"` + ModelName string `json:"model_name" gorm:"size:64;primaryKey;autoIncrement:false;default:'';index:idx_slice_model,priority:2;comment:origin model name (after mapping: log model)"` + TotalRequests int `json:"total_requests" gorm:"not null;default:0;comment:events observed in this slice for this model"` + ErrorRequests int `json:"error_requests" gorm:"not null;default:0;comment:events considered failure in this slice for this model"` + SuccessQualifiedRequests int `json:"success_qualified_requests" gorm:"not null;default:0;comment:successful requests meeting threshold"` + HasSuccessQualified bool `json:"has_success_qualified" gorm:"not null;default:false;comment:1 if any qualified success in slice"` + MaxResponseBytes int `json:"max_response_bytes" gorm:"not null;default:0;comment:max response bytes observed in slice (0 if unknown)"` + MaxCompletionTokens int `json:"max_completion_tokens" gorm:"not null;default:0;comment:max completion tokens observed in slice"` + MaxAssistantChars int `json:"max_assistant_chars" gorm:"not null;default:0;comment:max assistant content char length observed in slice (0 if unknown)"` + UpdatedAt time.Time +} + +func (ModelHealthSlice5m) TableName() string { + return "model_health_slice_5m" +} + +type ModelHealthEvent struct { + ModelName string + CreatedAt int64 + IsError bool + ResponseBytes int + CompletionTokens int + AssistantChars int + SuccessIsQualified bool + HasMetricsAvailable bool +} + +func AlignSliceStartTs(createdAt int64) int64 { + return createdAt - (createdAt % modelHealthSliceSeconds) +} + +func IsQualifiedSuccess(responseBytes, completionTokens, assistantChars int) bool { + return responseBytes > 1024 || completionTokens > 2 || assistantChars > 2 +} + +func (e *ModelHealthEvent) Normalize() error { + if e == nil { + return errors.New("event is nil") + } + if e.ModelName == "" { + return errors.New("model_name is required") + } + if e.CreatedAt <= 0 { + return errors.New("created_at must be positive") + } + if e.ResponseBytes < 0 || e.CompletionTokens < 0 || e.AssistantChars < 0 { + return errors.New("metrics must be non-negative") + } + e.SuccessIsQualified = !e.IsError && IsQualifiedSuccess(e.ResponseBytes, e.CompletionTokens, e.AssistantChars) + e.HasMetricsAvailable = e.ResponseBytes > 0 || e.CompletionTokens > 0 || e.AssistantChars > 0 + return nil +} + +func UpsertModelHealthSlice5m(ctx context.Context, db *gorm.DB, event *ModelHealthEvent) error { + if event == nil { + return errors.New("event is nil") + } + if err := event.Normalize(); err != nil { + return err + } + if db == nil { + return errors.New("db is nil") + } + + sliceStart := AlignSliceStartTs(event.CreatedAt) + + row := &ModelHealthSlice5m{ + SliceStartTs: sliceStart, + ModelName: event.ModelName, + TotalRequests: 1, + ErrorRequests: 0, + SuccessQualifiedRequests: 0, + HasSuccessQualified: event.SuccessIsQualified, + MaxResponseBytes: maxInt(0, event.ResponseBytes), + MaxCompletionTokens: maxInt(0, event.CompletionTokens), + MaxAssistantChars: maxInt(0, event.AssistantChars), + } + + if event.IsError { + row.ErrorRequests = 1 + } + if event.SuccessIsQualified { + row.SuccessQualifiedRequests = 1 + } + + updates := map[string]any{ + "total_requests": gorm.Expr("total_requests + VALUES(total_requests)"), + "error_requests": gorm.Expr("error_requests + VALUES(error_requests)"), + "success_qualified_requests": gorm.Expr("success_qualified_requests + VALUES(success_qualified_requests)"), + "has_success_qualified": gorm.Expr("has_success_qualified OR VALUES(has_success_qualified)"), + "max_response_bytes": gorm.Expr("GREATEST(max_response_bytes, VALUES(max_response_bytes))"), + "max_completion_tokens": gorm.Expr("GREATEST(max_completion_tokens, VALUES(max_completion_tokens))"), + "max_assistant_chars": gorm.Expr("GREATEST(max_assistant_chars, VALUES(max_assistant_chars))"), + } + + return db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{ + {Name: "model_name"}, + {Name: "slice_start_ts"}, + }, + DoUpdates: clause.Assignments(updates), + }).Create(row).Error +} + +func maxInt(a, b int) int { + if a > b { + return a + } + return b +} \ No newline at end of file diff --git a/model/model_health_writer.go b/model/model_health_writer.go new file mode 100644 index 000000000000..d9a1789c1355 --- /dev/null +++ b/model/model_health_writer.go @@ -0,0 +1,52 @@ +package model + +import ( + "context" + "sync" + + "github.com/bytedance/gopkg/util/gopool" +) + +const ( + modelHealthEventQueueSize = 8192 + modelHealthWorkerCount = 4 +) + +var ( + modelHealthOnce sync.Once + modelHealthQueue chan *ModelHealthEvent +) + +func initModelHealthWriter() { + modelHealthQueue = make(chan *ModelHealthEvent, modelHealthEventQueueSize) + for i := 0; i < modelHealthWorkerCount; i++ { + gopool.Go(func() { + for event := range modelHealthQueue { + func() { + defer func() { + _ = recover() + }() + _ = UpsertModelHealthSlice5m(context.Background(), DB, event) + }() + } + }) + } +} + +func RecordModelHealthEventAsync(_ any, event *ModelHealthEvent) { + if event == nil { + return + } + modelHealthOnce.Do(initModelHealthWriter) + + select { + case modelHealthQueue <- event: + default: + gopool.Go(func() { + defer func() { + _ = recover() + }() + _ = UpsertModelHealthSlice5m(context.Background(), DB, event) + }) + } +} \ No newline at end of file diff --git a/model/option.go b/model/option.go index e9fd50d7f357..be99474864a9 100644 --- a/model/option.go +++ b/model/option.go @@ -45,6 +45,7 @@ func InitOptionMap() { common.OptionMap["RegisterEnabled"] = strconv.FormatBool(common.RegisterEnabled) common.OptionMap["AutomaticDisableChannelEnabled"] = strconv.FormatBool(common.AutomaticDisableChannelEnabled) common.OptionMap["AutomaticEnableChannelEnabled"] = strconv.FormatBool(common.AutomaticEnableChannelEnabled) + common.OptionMap["HourlyCallRankCountFailedEnabled"] = strconv.FormatBool(common.HourlyCallRankCountFailedEnabled) common.OptionMap["LogConsumeEnabled"] = strconv.FormatBool(common.LogConsumeEnabled) common.OptionMap["DisplayInCurrencyEnabled"] = strconv.FormatBool(common.DisplayInCurrencyEnabled) common.OptionMap["DisplayTokenStatEnabled"] = strconv.FormatBool(common.DisplayTokenStatEnabled) @@ -111,6 +112,7 @@ func InitOptionMap() { common.OptionMap["ModelRequestRateLimitDurationMinutes"] = strconv.Itoa(setting.ModelRequestRateLimitDurationMinutes) common.OptionMap["ModelRequestRateLimitSuccessCount"] = strconv.Itoa(setting.ModelRequestRateLimitSuccessCount) common.OptionMap["ModelRequestRateLimitGroup"] = setting.ModelRequestRateLimitGroup2JSONString() + common.OptionMap["ModelRequestRateLimitExemptUserIDs"] = "" common.OptionMap["ModelRatio"] = ratio_setting.ModelRatio2JSONString() common.OptionMap["ModelPrice"] = ratio_setting.ModelPrice2JSONString() common.OptionMap["CacheRatio"] = ratio_setting.CacheRatio2JSONString() @@ -242,6 +244,8 @@ func updateOptionMap(key string, value string) (err error) { common.AutomaticDisableChannelEnabled = boolValue case "AutomaticEnableChannelEnabled": common.AutomaticEnableChannelEnabled = boolValue + case "HourlyCallRankCountFailedEnabled": + common.HourlyCallRankCountFailedEnabled = boolValue case "LogConsumeEnabled": common.LogConsumeEnabled = boolValue case "DisplayInCurrencyEnabled": @@ -404,6 +408,8 @@ func updateOptionMap(key string, value string) (err error) { setting.ModelRequestRateLimitSuccessCount, _ = strconv.Atoi(value) case "ModelRequestRateLimitGroup": err = setting.UpdateModelRequestRateLimitGroupByJSONString(value) + case "ModelRequestRateLimitExemptUserIDs": + err = setting.UpdateModelRequestRateLimitExemptUserIDs(value) case "RetryTimes": common.RetryTimes, _ = strconv.Atoi(value) case "DataExportInterval": diff --git a/model/user_call_hourly.go b/model/user_call_hourly.go new file mode 100644 index 000000000000..bedaae142e18 --- /dev/null +++ b/model/user_call_hourly.go @@ -0,0 +1,16 @@ +package model + +import "time" + +type UserCallHourly struct { + HourStartTs int64 `json:"hour_start_ts" gorm:"primaryKey;autoIncrement:false;index:idx_hour_calls,priority:1;index:idx_user_hour,priority:2;comment:hour start unix seconds, aligned to 3600s"` + UserId int `json:"user_id" gorm:"primaryKey;autoIncrement:false;index:idx_hour_calls,priority:3;index:idx_user_hour,priority:1;comment:user id"` + Username string `json:"username" gorm:"size:64;not null;default:'';comment:denormalized username for display"` + TotalCalls int `json:"total_calls" gorm:"not null;default:0;index:idx_hour_calls,priority:2;comment:total calls in this hour"` + SuccessCalls int `json:"success_calls" gorm:"not null;default:0;comment:successful calls in this hour (best-effort)"` + UpdatedAt time.Time `json:"updated_at"` +} + +func (UserCallHourly) TableName() string { + return "user_call_hourly" +} \ No newline at end of file diff --git a/model/user_call_hourly_query.go b/model/user_call_hourly_query.go new file mode 100644 index 000000000000..2cf5d74f8ddb --- /dev/null +++ b/model/user_call_hourly_query.go @@ -0,0 +1,91 @@ +package model + +import ( + "fmt" + "sort" + "strings" + + "gorm.io/gorm" +) + +type UserHourlyCallsRankItem struct { + UserId int `json:"user_id"` + Username string `json:"username"` + TotalCalls int64 `json:"total_calls"` +} + +func NormalizeHourList(hours []int64) ([]int64, error) { + if len(hours) == 0 { + return nil, nil + } + out := make([]int64, 0, len(hours)) + seen := make(map[int64]struct{}, len(hours)) + for _, h := range hours { + if h <= 0 || h%3600 != 0 { + return nil, fmt.Errorf("hours must be aligned to hour (ts %% 3600 == 0)") + } + if _, ok := seen[h]; ok { + continue + } + seen[h] = struct{}{} + out = append(out, h) + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out, nil +} + +func GetUserHourlyCallsRank(db *gorm.DB, hours []int64, startHourTs int64, endHourTs int64, limit int) ([]UserHourlyCallsRankItem, error) { + if db == nil { + return nil, fmt.Errorf("db is nil") + } + if limit <= 0 { + limit = 50 + } + if limit > 500 { + limit = 500 + } + + useHoursList := len(hours) > 0 + if useHoursList { + var err error + hours, err = NormalizeHourList(hours) + if err != nil { + return nil, err + } + if len(hours) == 0 { + return []UserHourlyCallsRankItem{}, nil + } + } else { + if startHourTs <= 0 || endHourTs <= 0 || endHourTs <= startHourTs { + return nil, fmt.Errorf("invalid hour range") + } + if startHourTs%3600 != 0 || endHourTs%3600 != 0 { + return nil, fmt.Errorf("start_hour/end_hour must be aligned to hour (ts %% 3600 == 0)") + } + // guardrail: 31 days + if endHourTs-startHourTs > 31*24*3600 { + return nil, fmt.Errorf("hour range too large (max 31 days)") + } + } + + base := db.Table((&UserCallHourly{}).TableName()).Select(strings.TrimSpace(` +user_id as user_id, +MAX(username) as username, +SUM(total_calls) as total_calls`)) + + if useHoursList { + base = base.Where("hour_start_ts IN ?", hours) + } else { + base = base.Where("hour_start_ts >= ? AND hour_start_ts < ?", startHourTs, endHourTs) + } + + var rows []UserHourlyCallsRankItem + err := base.Group("user_id"). + Order("total_calls DESC"). + Limit(limit). + Scan(&rows).Error + if err != nil { + return nil, err + } + return rows, nil +} \ No newline at end of file diff --git a/model/user_call_hourly_writer.go b/model/user_call_hourly_writer.go new file mode 100644 index 000000000000..486aa97dc82f --- /dev/null +++ b/model/user_call_hourly_writer.go @@ -0,0 +1,97 @@ +package model + +import ( + "context" + "sync" + + "github.com/bytedance/gopkg/util/gopool" + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +const ( + userCallHourlyEventQueueSize = 8192 + userCallHourlyWorkerCount = 4 +) + +type UserCallHourlyEvent struct { + UserId int + Username string + CreatedAt int64 + IsError bool +} + +func AlignHourStartTs(createdAt int64) int64 { + if createdAt <= 0 { + return 0 + } + return createdAt - (createdAt % 3600) +} + +func UpsertUserCallHourly(ctx context.Context, db *gorm.DB, event *UserCallHourlyEvent) error { + if event == nil || db == nil { + return nil + } + hourStart := AlignHourStartTs(event.CreatedAt) + if hourStart == 0 || event.UserId <= 0 { + return nil + } + + row := &UserCallHourly{ + HourStartTs: hourStart, + UserId: event.UserId, + Username: event.Username, + TotalCalls: 1, + SuccessCalls: 0, + } + if !event.IsError { + row.SuccessCalls = 1 + } + + return db.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{ + {Name: "hour_start_ts"}, + {Name: "user_id"}, + }, + DoUpdates: clause.Assignments(map[string]any{ + "username": row.Username, + "total_calls": gorm.Expr("total_calls + ?", row.TotalCalls), + "success_calls": gorm.Expr("success_calls + ?", row.SuccessCalls), + }), + }).Create(row).Error +} + +var ( + userCallHourlyOnce sync.Once + userCallHourlyQueue chan *UserCallHourlyEvent +) + +func initUserCallHourlyWriter() { + userCallHourlyQueue = make(chan *UserCallHourlyEvent, userCallHourlyEventQueueSize) + for i := 0; i < userCallHourlyWorkerCount; i++ { + gopool.Go(func() { + for event := range userCallHourlyQueue { + func() { + defer func() { _ = recover() }() + _ = UpsertUserCallHourly(context.Background(), DB, event) + }() + } + }) + } +} + +func RecordUserCallHourlyEventAsync(_ any, event *UserCallHourlyEvent) { + if event == nil { + return + } + userCallHourlyOnce.Do(initUserCallHourlyWriter) + + select { + case userCallHourlyQueue <- event: + default: + gopool.Go(func() { + defer func() { _ = recover() }() + _ = UpsertUserCallHourly(context.Background(), DB, event) + }) + } +} \ No newline at end of file diff --git a/plans/health_migration_design.md b/plans/health_migration_design.md new file mode 100644 index 000000000000..9cb6ec9ff41e --- /dev/null +++ b/plans/health_migration_design.md @@ -0,0 +1,332 @@ +# 新表与索引/迁移方案(健康度5分钟切片 + 小时成功率 + 可选小时用户调用排行) + +> 范围严格对齐需求:仅做两项的表/索引/迁移设计,不写实现代码。需求来源:[`Difference.md`](Difference.md:3)、[`Difference.md`](Difference.md:6) + +## 现状入口(用于落地与改造定位) + +- 日志表结构:[`model.Log`](model/log.go:20) +- 成功请求落日志入口:[`model.RecordConsumeLog()`](model/log.go:156),主要调用链:[`relay.postConsumeQuota()`](relay/compatible_handler.go:192) → [`model.RecordConsumeLog()`](model/log.go:156) +- 失败请求落日志入口:[`model.RecordErrorLog()`](model/log.go:99),主要调用链:[`controller.processChannelError()`](controller/relay.go:345) → [`model.RecordErrorLog()`](model/log.go:99),并受 [`types.IsRecordErrorLog()`](types/error.go:363) 控制 +- 现有小时聚合:[`model.QuotaData`](model/usedata.go:13) 与写入 [`model.LogQuotaData()`](model/usedata.go:58)(只精确到小时) +- 现有看板 API:[`controller.GetAllQuotaDates()`](controller/usedata.go:13) + +## 口径定义(固定,不在实现中再猜) + +### 1) 时间对齐 +- 5 分钟时间片:`slice_start_ts = created_at - (created_at % 300)`,以服务器时区的“时间语义”展示(存储仍建议用 unix seconds,展示端按服务器时区渲染)。 +- 小时:`hour_start_ts = created_at - (created_at % 3600)`(与现有 [`model.LogQuotaData()`](model/usedata.go:58) 对齐),展示同上按服务器时区。 + +### 2) 健康度 success slice 判定 +来自需求:[`Difference.md`](Difference.md:3) + +对某个 `model_name` 的某个 5 分钟 slice: +- `total_slice`:该 slice 内“有请求事件出现”则计 1(每个 slice 至多 1)。 +- `success_slice`:该 slice 内只要存在至少 1 个“成功请求且满足阈值”则计 1(混合成功/失败时按成功)。 +- “成功请求且满足阈值”的定义: + - 请求在业务意义上成功(没有走错误返回;或可用 `HTTP 2xx` + 非错误响应体 来判定),且 + - 满足三者之一: + - `response_bytes > 1024`(>1KB) + - `completion_tokens > 2` + - `assistant_content_chars > 2` + +说明:失败请求定义为进入 [`controller.processChannelError()`](controller/relay.go:345) 的 `newAPIError != nil` 分支并且 `types.IsRecordErrorLog()` 为 true 时能落库;此外还存在“未落 error log 的失败”(例如显式配置 `types.ErrOptionWithNoRecordErrorLog()`),它们在健康度分母内是否计入属于产品口径问题;本方案默认:**健康度基于可观测事件**,即以“写入聚合的事件”为准。 + +--- + +## 总体建模选择 + +为满足高性能查询(按 model + 小时范围计算 success_slice/total_slice)与可选小时用户排行(按小时集合聚合 user count 降序),采用 **2 张新表**: + +1) `model_health_slice_5m`:按 `model_name + slice_start_ts` 聚合 5 分钟切片结果(每行一个 model 的一个 5 分钟 slice)。 +2) `user_call_hourly`:按 `hour_start_ts + user_id` 聚合该用户在该小时的调用次数(可用于单小时 topN,也可用于小时集合求和排行)。 + +不扩展现有 `quota_data` 的原因: +- `quota_data` 的主键维度是 `(user_id, username, model_name, created_at hour)`,偏向额度/令牌/模型维度;本需求的“用户总调用次数排行”不需要模型维度,且需要高效 topN(单小时)/小时集合求和(多小时)。单独新表能更轻、更专用、索引更精准,避免 `quota_data` 额外索引膨胀和聚合成本。 + +--- + +## 表 A:模型健康度 5 分钟切片表 + +### DDL(MySQL InnoDB) + +```sql +CREATE TABLE IF NOT EXISTS model_health_slice_5m ( + slice_start_ts BIGINT NOT NULL COMMENT 'slice start unix seconds, aligned to 300s', + model_name VARCHAR(64) NOT NULL DEFAULT '' COMMENT 'origin model name (after mapping: log model)', + total_requests INT NOT NULL DEFAULT 0 COMMENT 'events observed in this slice for this model', + error_requests INT NOT NULL DEFAULT 0 COMMENT 'events considered failure in this slice for this model', + success_qualified_requests INT NOT NULL DEFAULT 0 COMMENT 'successful requests meeting threshold', + has_success_qualified TINYINT(1) NOT NULL DEFAULT 0 COMMENT '1 if any qualified success in slice', + max_response_bytes INT NOT NULL DEFAULT 0 COMMENT 'max response bytes observed in slice (0 if unknown)', + max_completion_tokens INT NOT NULL DEFAULT 0 COMMENT 'max completion tokens observed in slice', + max_assistant_chars INT NOT NULL DEFAULT 0 COMMENT 'max assistant content char length observed in slice (0 if unknown)', + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (model_name, slice_start_ts), + KEY idx_slice_start (slice_start_ts), + KEY idx_slice_model (slice_start_ts, model_name) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +``` + +### 索引设计理由 +- `PRIMARY KEY (model_name, slice_start_ts)`:写入与幂等更新以“模型+切片”为自然键,便于 `INSERT ... ON DUPLICATE KEY UPDATE`。 +- `KEY idx_slice_model (slice_start_ts, model_name)`:健康度查询通常是 `WHERE slice_start_ts BETWEEN ? AND ? AND model_name IN (...)` 或 `GROUP BY model_name`,该组合支持范围扫描 + 按模型聚合。 +- `KEY idx_slice_start (slice_start_ts)`:用于按时间清理/归档、以及按全模型时间窗统计时的范围扫描。 + +### 分区/归档建议 +- 若数据量大(模型多、QPS 高、长期保存):建议按月对 `slice_start_ts` 做 RANGE 分区(例如每月一个分区)。MySQL 分区 DDL 需结合上线月份生成,略。 +- 保留策略建议:保留 90 天或 180 天(由业务需要决定);过期分区可直接 `DROP PARTITION` 快速清理。 + +--- + +## 表 B:小时用户调用次数排行表 + +### DDL(MySQL InnoDB) + +```sql +CREATE TABLE IF NOT EXISTS user_call_hourly ( + hour_start_ts BIGINT NOT NULL COMMENT 'hour start unix seconds, aligned to 3600s', + user_id INT NOT NULL COMMENT 'user id', + username VARCHAR(64) NOT NULL DEFAULT '' COMMENT 'denormalized username for display', + total_calls INT NOT NULL DEFAULT 0 COMMENT 'total calls in this hour', + success_calls INT NOT NULL DEFAULT 0 COMMENT 'successful calls in this hour (best-effort)', + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (hour_start_ts, user_id), + KEY idx_hour_calls (hour_start_ts, total_calls, user_id), + KEY idx_user_hour (user_id, hour_start_ts) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +``` + +### 索引设计理由 +- `PRIMARY KEY (hour_start_ts, user_id)`:自然聚合键,小时级 UPSERT 非常直接。 +- `KEY idx_hour_calls (hour_start_ts, total_calls, user_id)`:单小时 topN 查询形态是 `WHERE hour_start_ts=? ORDER BY total_calls DESC LIMIT N`,该索引能在同一小时分组内更快定位高 calls(MySQL 对 DESC 索引的利用依版本/执行计划而定,但该索引仍利于过滤与回表减少)。 +- `KEY idx_user_hour (user_id, hour_start_ts)`:支持按用户查历史(可选),以及回填/核对数据时的快速定位。 + +### 分区/归档建议 +- 同样可按 `hour_start_ts` 做月分区;保留周期建议与健康度表一致或更短(例如 90 天)。 + +--- + +## 写入路径(事件来源、成功/失败判定、字段来源、需要补采样的点) + +### 1) 成功事件来源(consume log 路径) +主链路:[`relay.postConsumeQuota()`](relay/compatible_handler.go:192) → [`model.RecordConsumeLog()`](model/log.go:156) + +在 `RecordConsumeLog` 发生前,系统已计算并持有: +- `completion_tokens`:来自 `usage.CompletionTokens`(多渠道由 adaptor 解析或本地估算,见 [`relay/channel/openai.OaiStreamHandler()`](relay/channel/openai/relay-openai.go:106) 末尾 `ResponseText2Usage`)。 +- `prompt_tokens`:同上。 +- `model_name`:`relayInfo.OriginModelName`,写入时可能对 gizmo 做归一化([`relay.postConsumeQuota()`](relay/compatible_handler.go:405))。 +- `is_stream`:`relayInfo.IsStream`([`relay.TextHelper()`](relay/compatible_handler.go:28) 内对 Content-Type 的判断)。 + +需要补采样字段: +- `response_bytes`: + - 非流式:在 [`service.IOCopyBytesGracefully()`](service/http.go:25) 中已知 `len(data)`,建议在此处把 `len(data)` 写入 `gin.Context`(如 `c.Set("resp_bytes", len(data))`)或写入 `other` map 后再落日志。 + - 流式:目前没有统一的 “写入字节计数器”,建议在 stream handler 中统计写出的 bytes(例如累计 `len(lastStreamData)` 或底层 writer 计数),最终放到 `other`,供聚合使用。 +- `assistant_content_chars`: + - OpenAI 流式:已有 `responseTextBuilder`([`relay/channel/openai.OaiStreamHandler()`](relay/channel/openai/relay-openai.go:119)),可在结束时取 `len(responseTextBuilder.String())`(或更优为 builder.Len)。 + - OpenAI 非流式:`simpleResponse.Choices[].Message` 的 content 可解析([`relay/channel/openai.OpenaiHandler()`](relay/channel/openai/relay-openai.go:196)),可求和/取 max。 + - 其他渠道:若走 `ResponseText2Usage`([`relay/channel/openai.OaiStreamHandler()`](relay/channel/openai/relay-openai.go:184))通常也有 responseText,可同样计算长度;否则需要在各 adaptor 的 DoResponse 中补齐(以“尽量可用”为目标,无法获取则置 0)。 +- `success_qualified`(布尔): + - 成功判定:成功路径天然是 `RecordConsumeLog` 被调用(无 newAPIError),可视为“成功请求”候选。 + - 阈值判定:`resp_bytes>1024 OR completion_tokens>2 OR assistant_chars>2`。 + +写入动作(逻辑层,不实现): +- 每次 `RecordConsumeLog`: + - 计算 `slice_start_ts`、`hour_start_ts` + - `model_health_slice_5m`:对 `(model_name, slice_start_ts)` 做 UPSERT 增量: + - `total_requests += 1` + - `success_qualified_requests += (qualified?1:0)` + - `has_success_qualified = has_success_qualified OR qualified` + - `max_* = GREATEST(max_*, current_*)` + - `user_call_hourly`:对 `(hour_start_ts, user_id)` UPSERT: + - `total_calls += 1` + - `success_calls += 1`(成功路径) + +### 2) 失败事件来源(error log 路径) +主要入口:[`controller.processChannelError()`](controller/relay.go:345) → [`model.RecordErrorLog()`](model/log.go:99) + +当前 error log 具备: +- `userId/modelName/channelId/tokenId/group` 等维度([`controller.processChannelError()`](controller/relay.go:355))。 +- `other` 里有 `error_type/error_code/status_code` 等([`controller.processChannelError()`](controller/relay.go:363))。 + +缺失字段(用于阈值判定): +- `completion_tokens/resp_bytes/assistant_chars` 通常不可得(失败时可能没有有效响应体/usage)。 + +写入动作(逻辑层,不实现): +- 每次 `RecordErrorLog`: + - 计算 `slice_start_ts`、`hour_start_ts` + - `model_health_slice_5m` UPSERT: + - `total_requests += 1` + - `error_requests += 1` + - `has_success_qualified` 不变(失败不触发) + - `user_call_hourly` UPSERT: + - `total_calls += 1` + - `success_calls` 不变 + +注意:并非所有失败都会进入 `RecordErrorLog`(例如错误使用了 [`types.ErrOptionWithNoRecordErrorLog()`](types/error.go:348)),因此健康度与排行的“失败覆盖率”取决于该开关。若需要 100% 覆盖,应在更底层(请求生命周期结束处)补“统一失败事件”,但这超出本次范围;本方案仅声明风险。 + +--- + +## 查询模式(健康度 & 用户排行) + +### 1) 健康度:按模型 + 小时段(可选多个小时)返回 success_slice/total_slice 与成功率 + +输入: +- `model_names`(可多选) +- `hours`:一组 `hour_start_ts` 或一个时间范围 `[start_hour, end_hour)`(服务器时区语义) + +查询思路: +- 小时段内包含若干 5 分钟 slice:`slice_start_ts BETWEEN hour_start_ts AND hour_start_ts+3600-300` +- `total_slice = COUNT(*)`(每行代表一个 slice) +- `success_slice = SUM(has_success_qualified)` +- `success_rate = success_slice / total_slice` + +示例 SQL(单模型、多小时范围): + +```sql +SELECT + model_name, + FLOOR(slice_start_ts / 3600) AS hour_bucket, + SUM(has_success_qualified) AS success_slice, + COUNT(*) AS total_slice, + SUM(has_success_qualified) / COUNT(*) AS success_rate +FROM model_health_slice_5m +WHERE model_name = ? + AND slice_start_ts >= ? + AND slice_start_ts < ? +GROUP BY model_name, hour_bucket +ORDER BY hour_bucket ASC; +``` + +示例 SQL(多模型、指定小时集合): + +```sql +SELECT + model_name, + FLOOR(slice_start_ts / 3600) AS hour_bucket, + SUM(has_success_qualified) AS success_slice, + COUNT(*) AS total_slice, + SUM(has_success_qualified) / COUNT(*) AS success_rate +FROM model_health_slice_5m +WHERE model_name IN ( ... ) + AND FLOOR(slice_start_ts / 3600) IN ( ... ) +GROUP BY model_name, hour_bucket; +``` + +> 注:`FLOOR(slice_start_ts / 3600)` 用于把 5 分钟 bucket 归到小时 bucket(基于 unix 秒),展示层按服务器时区解释。 + +### 2) 用户排行:给定小时集合或单小时,按用户聚合 count 降序 + +单小时 topN(最快路径): + +```sql +SELECT user_id, username, total_calls +FROM user_call_hourly +WHERE hour_start_ts = ? +ORDER BY total_calls DESC +LIMIT ?; +``` + +多小时集合(求和排行): + +```sql +SELECT user_id, + MAX(username) AS username, + SUM(total_calls) AS total_calls +FROM user_call_hourly +WHERE hour_start_ts IN ( ... ) +GROUP BY user_id +ORDER BY total_calls DESC +LIMIT ?; +``` + +索引命中解释: +- 单小时 topN:`WHERE hour_start_ts=?` 走 `PRIMARY KEY` 前缀或 `idx_hour_calls`,排序字段 `total_calls` 与索引靠近能减少额外排序开销。 +- 多小时集合:`WHERE hour_start_ts IN (...)` 走 `PRIMARY KEY` 扫描对应小时分区/范围,聚合后排序(不可完全避免),但比从原始 logs 聚合小得多。 + +--- + +## 迁移/回填策略(从现有 logs/quota_data) + +### 目标 +- 让新表在上线后“尽快可用”,并在可行范围内补历史数据。 + +### 可回填的部分 +1) `user_call_hourly`: +- 从 [`model.Log`](model/log.go:20) 可回填(强可行): + - 成功:`logs.type = LogTypeConsume`([`model.LogTypeConsume`](model/log.go:46))视为成功调用事件 + - 失败:`logs.type = LogTypeError`([`model.LogTypeError`](model/log.go:49))视为失败调用事件(注意:受 `IsRecordErrorLog` 影响,历史 error 不一定全) +- 回填 SQL(示意,按小时聚合): + +```sql +INSERT INTO user_call_hourly (hour_start_ts, user_id, username, total_calls, success_calls) +SELECT + (created_at - (created_at % 3600)) AS hour_start_ts, + user_id, + MAX(username) AS username, + COUNT(*) AS total_calls, + SUM(type = 2) AS success_calls +FROM logs +WHERE created_at >= ? AND created_at < ? + AND type IN (2, 5) +GROUP BY hour_start_ts, user_id +ON DUPLICATE KEY UPDATE + username = VALUES(username), + total_calls = VALUES(total_calls), + success_calls = VALUES(success_calls); +``` + +风险与成本: +- 成本:按时间范围扫 `logs`,若 logs 很大需分批(按天/按小时)执行。 +- 风险:历史错误日志可能不全(跳过记录),`success_calls` 可靠、`failure` 可能偏低。 + +2) `model_health_slice_5m`: +- 从 logs 只能“部分回填”(强约束): + - `completion_tokens` 在 consume log 有([`model.Log.CompletionTokens`](model/log.go:31)),可用于阈值之一(`completion_tokens>2`)。 + - `response_bytes` 与 `assistant_content_chars` 历史上不在 logs 明确存储([`service.IOCopyBytesGracefully()`](service/http.go:25) 仅写 header,不持久化),因此无法严格按需求口径回填。 +- 可选回填策略(折中): + - 仅用 `completion_tokens>2` 作为“阈值满足”的代理条件回填历史 `has_success_qualified`。 + - 对历史数据,明确标注“健康度为近似口径”,避免误导(展示层可加说明,但超出本次范围;此处仅声明风险)。 + +示例回填 SQL(近似口径,仅基于 completion_tokens): + +```sql +INSERT INTO model_health_slice_5m + (slice_start_ts, model_name, total_requests, error_requests, success_qualified_requests, has_success_qualified, max_completion_tokens) +SELECT + (created_at - (created_at % 300)) AS slice_start_ts, + model_name, + COUNT(*) AS total_requests, + SUM(type = 5) AS error_requests, + SUM(type = 2 AND completion_tokens > 2) AS success_qualified_requests, + MAX(type = 2 AND completion_tokens > 2) AS has_success_qualified, + MAX(CASE WHEN type = 2 THEN completion_tokens ELSE 0 END) AS max_completion_tokens +FROM logs +WHERE created_at >= ? AND created_at < ? + AND type IN (2, 5) +GROUP BY slice_start_ts, model_name +ON DUPLICATE KEY UPDATE + total_requests = VALUES(total_requests), + error_requests = VALUES(error_requests), + success_qualified_requests = VALUES(success_qualified_requests), + has_success_qualified = VALUES(has_success_qualified), + max_completion_tokens = GREATEST(max_completion_tokens, VALUES(max_completion_tokens)); +``` + +风险与成本: +- 风险(核心):历史健康度不满足 “response bytes / assistant chars” 两个条件的严格口径;只用 `completion_tokens` 会低估某些“低 token 但有输出内容/大响应”的成功 slice。 +- 成本:同样需要扫 logs;建议只回填最近 N 天,并在上线后逐步补齐(真正口径需新增采样后才成立)。 + +--- + +## 方案小结(给实现方的最小指令集) + +- 建两张表:`model_health_slice_5m`、`user_call_hourly`(DDL 如上)。 +- 成功事件:在 [`model.RecordConsumeLog()`](model/log.go:156) 触发处(或其上游统一点)做两个表的 UPSERT 增量。 +- 失败事件:在 [`controller.processChannelError()`](controller/relay.go:345) / [`model.RecordErrorLog()`](model/log.go:99) 触发处做两个表的 UPSERT 增量。 +- 必须补采样字段: + - 非流式 `response_bytes`:可从 [`service.IOCopyBytesGracefully()`](service/http.go:25) 的 `len(data)` 得到并传递到日志/聚合。 + - 流式 `response_bytes`:需要在 stream handler 增加计数(无现成统一计数)。 + - `assistant_content_chars`:可从流式聚合文本 builder / 非流式 choice 内容解析得到。 +- 回填: + - `user_call_hourly`:可从 logs 回填(高可行)。 + - `model_health_slice_5m`:只能近似回填(仅 completion_tokens 口径),或不回填历史,待采样上线后自然积累。 diff --git a/relay/channel/baidu/relay-baidu.go b/relay/channel/baidu/relay-baidu.go index 691d418886b9..614ad9caf562 100644 --- a/relay/channel/baidu/relay-baidu.go +++ b/relay/channel/baidu/relay-baidu.go @@ -115,7 +115,14 @@ func embeddingResponseBaidu2OpenAI(response *BaiduEmbeddingResponse) *dto.OpenAI func baiduStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*types.NewAPIError, *dto.Usage) { usage := &dto.Usage{} + + service.RecentCallsCache().EnsureStreamByContext(c, resp) + helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + var baiduResponse BaiduChatStreamResponse err := common.Unmarshal([]byte(data), &baiduResponse) if err != nil { @@ -134,6 +141,9 @@ func baiduStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http. } return true }) + + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, "") + service.CloseResponseBodyGracefully(resp) return nil, usage } diff --git a/relay/channel/claude/relay-claude.go b/relay/channel/claude/relay-claude.go index d3986236aa9b..3b5755dde307 100644 --- a/relay/channel/claude/relay-claude.go +++ b/relay/channel/claude/relay-claude.go @@ -710,8 +710,15 @@ func ClaudeStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon. ResponseText: strings.Builder{}, Usage: &dto.Usage{}, } + + service.RecentCallsCache().EnsureStreamByContext(c, resp) + var err *types.NewAPIError helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + err = HandleStreamResponseData(c, info, claudeInfo, data, requestMode) if err != nil { return false @@ -722,6 +729,8 @@ func ClaudeStreamHandler(c *gin.Context, resp *http.Response, info *relaycommon. return nil, err } + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, claudeInfo.ResponseText.String()) + HandleStreamFinalResponse(c, info, claudeInfo, requestMode) return claudeInfo.Usage, nil } diff --git a/relay/channel/dify/relay-dify.go b/relay/channel/dify/relay-dify.go index 24f5218a41cf..48f17cfae4eb 100644 --- a/relay/channel/dify/relay-dify.go +++ b/relay/channel/dify/relay-dify.go @@ -216,7 +216,14 @@ func difyStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.R usage := &dto.Usage{} var nodeToken int helper.SetEventStreamHeaders(c) + + service.RecentCallsCache().EnsureStreamByContext(c, resp) + helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + var difyResponse DifyChunkChatCompletionResponse err := json.Unmarshal([]byte(data), &difyResponse) if err != nil { @@ -244,6 +251,9 @@ func difyStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.R } return true }) + + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, responseText) + helper.Done(c) if usage.TotalTokens == 0 { usage = service.ResponseText2Usage(c, responseText, info.UpstreamModelName, info.GetEstimatePromptTokens()) diff --git a/relay/channel/gemini/relay-gemini-native.go b/relay/channel/gemini/relay-gemini-native.go index 5f9ff7cdfe98..2697f44b36d7 100644 --- a/relay/channel/gemini/relay-gemini-native.go +++ b/relay/channel/gemini/relay-gemini-native.go @@ -24,6 +24,9 @@ func GeminiTextGenerationHandler(c *gin.Context, info *relaycommon.RelayInfo, re return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) } + rawUpstreamBody := append([]byte(nil), responseBody...) + service.RecentCallsCache().UpsertUpstreamResponseByContext(c, resp, rawUpstreamBody) + if common.DebugEnabled { println(string(responseBody)) } @@ -65,6 +68,9 @@ func NativeGeminiEmbeddingHandler(c *gin.Context, resp *http.Response, info *rel return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) } + rawUpstreamBody := append([]byte(nil), responseBody...) + service.RecentCallsCache().UpsertUpstreamResponseByContext(c, resp, rawUpstreamBody) + if common.DebugEnabled { println(string(responseBody)) } @@ -93,7 +99,13 @@ func NativeGeminiEmbeddingHandler(c *gin.Context, resp *http.Response, info *rel func GeminiTextGenerationStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Response) (*dto.Usage, *types.NewAPIError) { helper.SetEventStreamHeaders(c) + service.RecentCallsCache().EnsureStreamByContext(c, resp) + return geminiStreamHandler(c, info, resp, func(data string, geminiResponse *dto.GeminiChatResponse) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + err := helper.StringData(c, data) if err != nil { logger.LogError(c, "failed to write stream data: "+err.Error()) diff --git a/relay/channel/gemini/relay-gemini.go b/relay/channel/gemini/relay-gemini.go index f75a921404ae..96d47b89556a 100644 --- a/relay/channel/gemini/relay-gemini.go +++ b/relay/channel/gemini/relay-gemini.go @@ -1077,7 +1077,13 @@ func geminiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http var imageCount int responseText := strings.Builder{} + service.RecentCallsCache().EnsureStreamByContext(c, resp) + helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + var geminiResponse dto.GeminiChatResponse err := common.UnmarshalJsonStr(data, &geminiResponse) if err != nil { @@ -1115,6 +1121,8 @@ func geminiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http return callback(data, &geminiResponse) }) + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, responseText.String()) + if imageCount != 0 { if usage.CompletionTokens == 0 { usage.CompletionTokens = imageCount * 1400 @@ -1209,6 +1217,10 @@ func GeminiChatHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.R if err != nil { return nil, types.NewOpenAIError(err, types.ErrorCodeBadResponseBody, http.StatusInternalServerError) } + + rawUpstreamBody := append([]byte(nil), responseBody...) + service.RecentCallsCache().UpsertUpstreamResponseByContext(c, resp, rawUpstreamBody) + service.CloseResponseBodyGracefully(resp) if common.DebugEnabled { println(string(responseBody)) diff --git a/relay/channel/gemini/relay_gemini_stream_recent_calls_test.go b/relay/channel/gemini/relay_gemini_stream_recent_calls_test.go new file mode 100644 index 000000000000..aa8b7ec71223 --- /dev/null +++ b/relay/channel/gemini/relay_gemini_stream_recent_calls_test.go @@ -0,0 +1,69 @@ +package gemini + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/QuantumNous/new-api/dto" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/QuantumNous/new-api/service" + "github.com/gin-gonic/gin" +) + +func TestGeminiStreamWritesRecentCallsChunksAndAggregatedText(t *testing.T) { + gin.SetMode(gin.TestMode) + + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest(http.MethodPost, "/v1beta/models/gemini-1.5-pro:streamGenerateContent", strings.NewReader(`{"contents":[]}`)) + + id := service.RecentCallsCache().BeginFromContext(c, nil, []byte(`{"contents":[]}`)) + if id == 0 { + t.Fatalf("expected non-zero recent call id") + } + + body := strings.Join([]string{ + `data: {"candidates":[{"index":0,"content":{"parts":[{"text":"hello "}]}}],"usageMetadata":{"promptTokenCount":1,"candidatesTokenCount":1,"totalTokenCount":2}}`, + `data: {"candidates":[{"index":0,"content":{"parts":[{"text":"world"}]}}],"usageMetadata":{"promptTokenCount":1,"candidatesTokenCount":2,"totalTokenCount":3}}`, + `data: [DONE]`, + "", + }, "\n") + + resp := &http.Response{ + StatusCode: 200, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader(body)), + } + + info := &relaycommon.RelayInfo{ + DisablePing: true, + ChannelMeta: &relaycommon.ChannelMeta{ + UpstreamModelName: "gemini-1.5-pro", + ChannelSetting: dto.ChannelSettings{}, + }, + } + _, apiErr := GeminiChatStreamHandler(c, info, resp) + if apiErr != nil { + t.Fatalf("unexpected api error: %v", apiErr) + } + + rec, ok := service.RecentCallsCache().Get(id) + if !ok { + t.Fatalf("expected recent call record") + } + if rec.Stream == nil { + t.Fatalf("expected stream info") + } + if len(rec.Stream.Chunks) != 2 { + t.Fatalf("expected 2 chunks, got %d", len(rec.Stream.Chunks)) + } + if rec.Stream.Chunks[0] == "" || rec.Stream.Chunks[1] == "" { + t.Fatalf("expected non-empty chunks") + } + if rec.Stream.AggregatedText != "hello world" { + t.Fatalf("unexpected aggregated text: %q", rec.Stream.AggregatedText) + } +} \ No newline at end of file diff --git a/relay/channel/openai/audio.go b/relay/channel/openai/audio.go index 877f5bb1ccd4..9b126233e8a9 100644 --- a/relay/channel/openai/audio.go +++ b/relay/channel/openai/audio.go @@ -35,7 +35,13 @@ func OpenaiTTSHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel c.Writer.WriteHeader(resp.StatusCode) if info.IsStream { + service.RecentCallsCache().EnsureStreamByContext(c, resp) + helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + if service.SundaySearch(data, "usage") { var simpleResponse dto.SimpleResponse err := common.Unmarshal([]byte(data), &simpleResponse) @@ -51,6 +57,8 @@ func OpenaiTTSHandler(c *gin.Context, resp *http.Response, info *relaycommon.Rel _ = helper.StringData(c, data) return true }) + + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, "") } else { common.SetContextKey(c, constant.ContextKeyLocalCountTokens, true) // 读取响应体到缓冲区 diff --git a/relay/channel/openai/relay-openai.go b/relay/channel/openai/relay-openai.go index ac44312eb0d4..e75d454c9d8c 100644 --- a/relay/channel/openai/relay-openai.go +++ b/relay/channel/openai/relay-openai.go @@ -111,6 +111,8 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re defer service.CloseResponseBodyGracefully(resp) + service.RecentCallsCache().EnsureStreamByContext(c, resp) + model := info.UpstreamModelName var responseId string var createAt int64 = 0 @@ -122,6 +124,7 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re var streamItems []string // store stream items var lastStreamData string var secondLastStreamData string // 存储倒数第二个stream data,用于音频模型 + var responseBytes int // 检查是否为音频模型 isAudioModel := strings.Contains(strings.ToLower(model), "audio") @@ -132,8 +135,11 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re if err != nil { common.SysLog("error handling stream format: " + err.Error()) } + responseBytes += len(lastStreamData) } if len(data) > 0 { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + // 对音频模型,保存倒数第二个stream data if isAudioModel && lastStreamData != "" { secondLastStreamData = lastStreamData @@ -181,11 +187,18 @@ func OaiStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re logger.LogError(c, "error processing tokens: "+err.Error()) } + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, responseTextBuilder.String()) + if !containStreamUsage { usage = service.ResponseText2Usage(c, responseTextBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) usage.CompletionTokens += toolCount * 7 } + if c != nil { + c.Set("response_bytes", responseBytes) + c.Set("assistant_content_chars", responseTextBuilder.Len()) + } + applyUsagePostProcessing(info, usage, nil) HandleFinalResponse(c, info, lastStreamData, responseId, createAt, model, systemFingerprint, usage, containStreamUsage) @@ -201,6 +214,10 @@ func OpenaiHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Respo if err != nil { return nil, types.NewOpenAIError(err, types.ErrorCodeReadResponseBodyFailed, http.StatusInternalServerError) } + + rawUpstreamBody := append([]byte(nil), responseBody...) + service.RecentCallsCache().UpsertUpstreamResponseByContext(c, resp, rawUpstreamBody) + if common.DebugEnabled { println("upstream response body:", string(responseBody)) } @@ -288,6 +305,17 @@ func OpenaiHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Respo responseBody = geminiRespStr } + if c != nil { + maxAssistantChars := 0 + for _, choice := range simpleResponse.Choices { + content := choice.Message.StringContent() + if len(content) > maxAssistantChars { + maxAssistantChars = len(content) + } + } + c.Set("assistant_content_chars", maxAssistantChars) + } + service.IOCopyBytesGracefully(c, resp, responseBody) return &simpleResponse.Usage, nil diff --git a/relay/channel/openai/relay_responses.go b/relay/channel/openai/relay_responses.go index b92c8c7234cd..a214dcc1a183 100644 --- a/relay/channel/openai/relay_responses.go +++ b/relay/channel/openai/relay_responses.go @@ -79,7 +79,12 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp var usage = &dto.Usage{} var responseTextBuilder strings.Builder + service.RecentCallsCache().EnsureStreamByContext(c, resp) + helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } // 检查当前数据是否包含 completed 状态和 usage 信息 var streamResponse dto.ResponsesStreamResponse @@ -130,6 +135,8 @@ func OaiResponsesStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp return true }) + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, responseTextBuilder.String()) + if usage.CompletionTokens == 0 { // 计算输出文本的 token 数量 tempStr := responseTextBuilder.String() diff --git a/relay/channel/xai/text.go b/relay/channel/xai/text.go index aa4d329f3111..18576a919037 100644 --- a/relay/channel/xai/text.go +++ b/relay/channel/xai/text.go @@ -44,7 +44,13 @@ func xAIStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re helper.SetEventStreamHeaders(c) + service.RecentCallsCache().EnsureStreamByContext(c, resp) + helper.StreamScannerHandler(c, resp, info, func(data string) bool { + if data != "" { + service.RecentCallsCache().AppendStreamChunkByContext(c, data) + } + var xAIResp *dto.ChatCompletionsStreamResponse err := json.Unmarshal([]byte(data), &xAIResp) if err != nil { @@ -69,6 +75,8 @@ func xAIStreamHandler(c *gin.Context, info *relaycommon.RelayInfo, resp *http.Re return true }) + service.RecentCallsCache().FinalizeStreamAggregatedTextByContext(c, responseTextBuilder.String()) + if !containStreamUsage { usage = service.ResponseText2Usage(c, responseTextBuilder.String(), info.UpstreamModelName, info.GetEstimatePromptTokens()) usage.CompletionTokens += toolCount * 7 diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go index 13c32c6758f6..2da9c5263258 100644 --- a/relay/helper/stream_scanner.go +++ b/relay/helper/stream_scanner.go @@ -48,6 +48,9 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon }() streamingTimeout := time.Duration(constant.StreamingTimeout) * time.Second + if streamingTimeout <= 0 { + streamingTimeout = 60 * time.Second + } var ( stopChan = make(chan bool, 3) // 增加缓冲区避免阻塞 diff --git a/router/api-router.go b/router/api-router.go index fd204e7e6b64..f1d58027593c 100644 --- a/router/api-router.go +++ b/router/api-router.go @@ -48,6 +48,13 @@ func SetApiRouter(router *gin.Engine) { apiRouter.POST("/verify", middleware.UserAuth(), middleware.CriticalRateLimit(), controller.UniversalVerify) apiRouter.GET("/verify/status", middleware.UserAuth(), controller.GetVerificationStatus) + debugRoute := apiRouter.Group("/debug") + debugRoute.Use(middleware.AdminAuth()) + { + debugRoute.GET("/recent_calls", controller.GetRecentCalls) + debugRoute.GET("/recent_calls/:id", controller.GetRecentCallByID) + } + userRoute := apiRouter.Group("/user") { userRoute.POST("/register", middleware.CriticalRateLimit(), middleware.TurnstileCheck(), controller.Register) @@ -256,5 +263,23 @@ func SetApiRouter(router *gin.Engine) { modelsRoute.PUT("/", controller.UpdateModelMeta) modelsRoute.DELETE("/:id", controller.DeleteModelMeta) } + + userRankRoute := apiRouter.Group("/user_rank") + userRankRoute.Use(middleware.AdminAuth()) + { + userRankRoute.GET("/hourly_calls", controller.GetUserHourlyCallsRankAPI) + } + + modelHealthRoute := apiRouter.Group("/model_health") + modelHealthRoute.Use(middleware.AdminAuth()) + { + modelHealthRoute.GET("/hourly", controller.GetModelHealthHourlyStatsAPI) + } + + // Public model health view (no auth): last 24h hourly stats for all models + publicModelHealthRoute := apiRouter.Group("/public/model_health") + { + publicModelHealthRoute.GET("/hourly_last24h", controller.GetPublicModelsHealthHourlyLast24hAPI) + } } } diff --git a/service/http.go b/service/http.go index 7bd54c4acd00..2794057cf41a 100644 --- a/service/http.go +++ b/service/http.go @@ -29,6 +29,10 @@ func IOCopyBytesGracefully(c *gin.Context, src *http.Response, data []byte) { body := io.NopCloser(bytes.NewBuffer(data)) + if c != nil { + c.Set("response_bytes", len(data)) + } + // We shouldn't set the header before we parse the response body, because the parse part may fail. // And then we will have to send an error response, but in this case, the header has already been set. // So the httpClient will be confused by the response. diff --git a/service/model_role_mapping.go b/service/model_role_mapping.go new file mode 100644 index 000000000000..2724ff4dc697 --- /dev/null +++ b/service/model_role_mapping.go @@ -0,0 +1,344 @@ +package service + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "sync" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + "github.com/QuantumNous/new-api/dto" + "github.com/QuantumNous/new-api/logger" + + "github.com/gin-gonic/gin" +) + +type ModelRoleMappings map[string]map[string]string + +type RequestRoleSnapshot struct { + GeneralOpenAI *GeneralOpenAIRoleSnapshot + Responses *ResponsesRoleSnapshot +} + +type GeneralOpenAIRoleSnapshot struct { + MessagesRoles []string +} + +type ResponsesRoleSnapshot struct { + InputRoles []string +} + +var ( + allowedOpenAIRoles = map[string]struct{}{ + "system": {}, + "user": {}, + "assistant": {}, + "developer": {}, + "tool": {}, + } + + unknownRoleWarnOnce sync.Map // key: model + "|" + role +) + +func ValidateModelRoleMappingsJSON(jsonStr string) error { + _, err := ParseAndValidateModelRoleMappingsJSON(jsonStr) + return err +} + +func ParseAndValidateModelRoleMappingsJSON(jsonStr string) (ModelRoleMappings, error) { + if strings.TrimSpace(jsonStr) == "" { + return ModelRoleMappings{}, nil + } + + var raw any + if err := json.Unmarshal([]byte(jsonStr), &raw); err != nil { + return nil, fmt.Errorf("invalid json: %w", err) + } + + out, ok := raw.(map[string]any) + if !ok { + return nil, fmt.Errorf("expected object: map[modelPrefix]map[fromRole]toRole") + } + + mappings := ModelRoleMappings{} + for modelPrefix, v := range out { + if strings.TrimSpace(modelPrefix) == "" { + return nil, fmt.Errorf("model prefix cannot be empty") + } + roleMapAny, ok := v.(map[string]any) + if !ok { + return nil, fmt.Errorf("model %q mapping must be an object", modelPrefix) + } + roleMap := map[string]string{} + for fromRole, toAny := range roleMapAny { + toRole, ok := toAny.(string) + if !ok { + return nil, fmt.Errorf("model %q role %q target must be string", modelPrefix, fromRole) + } + fromRole = strings.TrimSpace(fromRole) + toRole = strings.TrimSpace(toRole) + if fromRole == "" || toRole == "" { + return nil, fmt.Errorf("model %q role mapping cannot have empty roles", modelPrefix) + } + if !IsAllowedOpenAIRole(fromRole) { + return nil, fmt.Errorf("model %q has invalid fromRole %q", modelPrefix, fromRole) + } + if !IsAllowedOpenAIRole(toRole) { + return nil, fmt.Errorf("model %q has invalid toRole %q", modelPrefix, toRole) + } + roleMap[fromRole] = toRole + } + mappings[modelPrefix] = roleMap + } + + return mappings, nil +} + +func IsAllowedOpenAIRole(role string) bool { + _, ok := allowedOpenAIRoles[role] + return ok +} + +func GetModelRoleMappingsFromChannelSettings(ctx context.Context) (ModelRoleMappings, bool) { + ginCtx, ok := ctx.(*gin.Context) + if !ok || ginCtx == nil { + return nil, false + } + + setting, ok := common.GetContextKeyType[dto.ChannelSettings](ginCtx, constant.ContextKeyChannelSetting) + if !ok { + return nil, false + } + if len(setting.ModelRoleMappings) == 0 { + return nil, false + } + + // Validate config defensively (it comes from persisted JSON) + b, err := common.Marshal(setting.ModelRoleMappings) + if err != nil { + logger.LogWarn(ctx, fmt.Sprintf("invalid channel model_role_mappings: %v", err)) + return nil, false + } + m, err := ParseAndValidateModelRoleMappingsJSON(string(b)) + if err != nil { + logger.LogWarn(ctx, fmt.Sprintf("invalid channel model_role_mappings: %v", err)) + return nil, false + } + if len(m) == 0 { + return nil, false + } + return m, true +} + +func ResolveRoleMappingForModel(model string, mappings ModelRoleMappings) (map[string]string, bool) { + if len(mappings) == 0 || model == "" { + return nil, false + } + + var ( + bestLen int = -1 + bestMap map[string]string + ) + for prefix, roleMap := range mappings { + matched := false + candidateLen := len(prefix) + + if prefix == "*" { + // Wildcard: matches any model but should have the lowest priority. + matched = true + candidateLen = 0 + } else if strings.HasPrefix(model, prefix) { + matched = true + } + + if matched && candidateLen > bestLen { + bestLen = candidateLen + bestMap = roleMap + } + } + if bestMap == nil { + return nil, false + } + return bestMap, true +} + +func SnapshotRequestRoles(request dto.Request) *RequestRoleSnapshot { + if request == nil { + return nil + } + + switch r := request.(type) { + case *dto.GeneralOpenAIRequest: + roles := make([]string, len(r.Messages)) + for i := range r.Messages { + roles[i] = r.Messages[i].Role + } + return &RequestRoleSnapshot{ + GeneralOpenAI: &GeneralOpenAIRoleSnapshot{MessagesRoles: roles}, + } + case *dto.OpenAIResponsesRequest: + if len(r.Input) == 0 || common.GetJsonType(r.Input) != "array" { + return &RequestRoleSnapshot{Responses: &ResponsesRoleSnapshot{InputRoles: nil}} + } + var inputs []dto.Input + if err := common.Unmarshal(r.Input, &inputs); err != nil { + return &RequestRoleSnapshot{Responses: &ResponsesRoleSnapshot{InputRoles: nil}} + } + roles := make([]string, len(inputs)) + for i := range inputs { + roles[i] = inputs[i].Role + } + return &RequestRoleSnapshot{ + Responses: &ResponsesRoleSnapshot{InputRoles: roles}, + } + default: + return nil + } +} + +func RestoreRequestRoles(request dto.Request, snapshot *RequestRoleSnapshot) { + if request == nil || snapshot == nil { + return + } + + switch r := request.(type) { + case *dto.GeneralOpenAIRequest: + if snapshot.GeneralOpenAI == nil { + return + } + if len(snapshot.GeneralOpenAI.MessagesRoles) != len(r.Messages) { + return + } + for i := range r.Messages { + r.Messages[i].Role = snapshot.GeneralOpenAI.MessagesRoles[i] + } + case *dto.OpenAIResponsesRequest: + if snapshot.Responses == nil { + return + } + if len(r.Input) == 0 || common.GetJsonType(r.Input) != "array" { + return + } + var inputs []dto.Input + if err := common.Unmarshal(r.Input, &inputs); err != nil { + return + } + if len(snapshot.Responses.InputRoles) != len(inputs) { + return + } + changed := false + for i := range inputs { + if inputs[i].Role != snapshot.Responses.InputRoles[i] { + inputs[i].Role = snapshot.Responses.InputRoles[i] + changed = true + } + } + if !changed { + return + } + b, err := common.Marshal(inputs) + if err != nil { + return + } + r.Input = b + default: + return + } +} + +func ApplyModelRoleMappingsToRequest(ctx context.Context, request dto.Request) { + if request == nil { + return + } + + mappings, ok := GetModelRoleMappingsFromChannelSettings(ctx) + if !ok { + return + } + + switch r := request.(type) { + case *dto.GeneralOpenAIRequest: + applyToGeneralOpenAIRequest(ctx, r, mappings) + case *dto.OpenAIResponsesRequest: + applyToOpenAIResponsesRequest(ctx, r, mappings) + default: + return + } +} + +func applyToGeneralOpenAIRequest(ctx context.Context, r *dto.GeneralOpenAIRequest, mappings ModelRoleMappings) { + roleMap, ok := ResolveRoleMappingForModel(r.Model, mappings) + if !ok || len(roleMap) == 0 { + return + } + + for i := range r.Messages { + orig := r.Messages[i].Role + if orig == "" { + continue + } + target, has := roleMap[orig] + if has { + r.Messages[i].Role = target + continue + } + if !IsAllowedOpenAIRole(orig) { + warnUnknownRoleOnce(ctx, r.Model, orig) + } + } +} + +func applyToOpenAIResponsesRequest(ctx context.Context, r *dto.OpenAIResponsesRequest, mappings ModelRoleMappings) { + roleMap, ok := ResolveRoleMappingForModel(r.Model, mappings) + if !ok || len(roleMap) == 0 { + return + } + if len(r.Input) == 0 { + return + } + + if common.GetJsonType(r.Input) != "array" { + return + } + + var inputs []dto.Input + if err := common.Unmarshal(r.Input, &inputs); err != nil { + return + } + + changed := false + for i := range inputs { + orig := strings.TrimSpace(inputs[i].Role) + if orig == "" { + continue + } + target, has := roleMap[orig] + if has { + inputs[i].Role = target + changed = true + continue + } + if !IsAllowedOpenAIRole(orig) { + warnUnknownRoleOnce(ctx, r.Model, orig) + } + } + + if !changed { + return + } + b, err := common.Marshal(inputs) + if err != nil { + return + } + r.Input = b +} + +func warnUnknownRoleOnce(ctx context.Context, model string, role string) { + key := model + "|" + role + if _, loaded := unknownRoleWarnOnce.LoadOrStore(key, struct{}{}); loaded { + return + } + logger.LogWarn(ctx, fmt.Sprintf("unknown role in request (model=%s): %s", model, role)) +} \ No newline at end of file diff --git a/service/recent_calls_cache.go b/service/recent_calls_cache.go new file mode 100644 index 000000000000..8c31f702542d --- /dev/null +++ b/service/recent_calls_cache.go @@ -0,0 +1,551 @@ +package service + +import ( + "encoding/base64" + "net/http" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/QuantumNous/new-api/common" + "github.com/QuantumNous/new-api/constant" + relaycommon "github.com/QuantumNous/new-api/relay/common" + "github.com/gin-gonic/gin" +) + +const ( + RecentCallsContextKeyID = "recent_calls_id" + + DefaultRecentCallsCapacity = 100 + + DefaultMaxRequestBodyBytes = 64 << 10 // 64KiB + DefaultMaxResponseBodyBytes = 256 << 10 // 256KiB + + DefaultMaxStreamChunkBytes = 8 << 10 // 8KiB + DefaultMaxStreamTotalBytes = 256 << 10 // 256KiB +) + +type RecentCallsCacheConfig struct { + Capacity int + + MaxRequestBodyBytes int + MaxResponseBodyBytes int + + MaxStreamChunkBytes int + MaxStreamTotalBytes int +} + +type RecentCallRequest struct { + Method string `json:"method"` + Path string `json:"path"` + Header map[string]string `json:"headers,omitempty"` + + BodyType string `json:"body_type,omitempty"` // json/text/binary/unknown/omitted + Body string `json:"body,omitempty"` // truncated string or base64 (when BodyType=binary) + Truncated bool `json:"truncated,omitempty"` // body truncated + Omitted bool `json:"omitted,omitempty"` // body not recorded + OmitReason string `json:"omit_reason,omitempty"` +} + +type RecentCallUpstreamResponse struct { + StatusCode int `json:"status_code"` + Header map[string]string `json:"headers,omitempty"` + + BodyType string `json:"body_type,omitempty"` // json/text/binary/unknown/omitted + Body string `json:"body,omitempty"` // raw upstream body (string or base64) + Truncated bool `json:"truncated,omitempty"` + Omitted bool `json:"omitted,omitempty"` + OmitReason string `json:"omit_reason,omitempty"` +} + +type RecentCallUpstreamStream struct { + Chunks []string `json:"chunks,omitempty"` // raw SSE data payload lines + ChunksTruncated bool `json:"chunks_truncated,omitempty"` // some chunks dropped/truncated due to limits + AggregatedText string `json:"aggregated_text,omitempty"` // best-effort aggregated assistant text + AggregatedTruncated bool `json:"aggregated_truncated,omitempty"` + + StreamBytes int `json:"-"` +} + +type RecentCallErrorInfo struct { + Message string `json:"message,omitempty"` + Type string `json:"type,omitempty"` + Code string `json:"code,omitempty"` + Status int `json:"status,omitempty"` +} + +type RecentCallRecord struct { + ID uint64 `json:"id"` + CreatedAt time.Time `json:"created_at"` + + UserID int `json:"user_id"` + ChannelID int `json:"channel_id,omitempty"` + ModelName string `json:"model_name,omitempty"` + + Method string `json:"method"` + Path string `json:"path"` + + Request RecentCallRequest `json:"request"` + Response *RecentCallUpstreamResponse `json:"response,omitempty"` + Stream *RecentCallUpstreamStream `json:"stream,omitempty"` + Error *RecentCallErrorInfo `json:"error,omitempty"` +} + +type recentCallsCache struct { + cfg RecentCallsCacheConfig + + nextID atomic.Uint64 + + mu sync.RWMutex + buffer []*RecentCallRecord +} + +var recentCallsSingleton = newRecentCallsCache(RecentCallsCacheConfig{ + Capacity: DefaultRecentCallsCapacity, + + MaxRequestBodyBytes: DefaultMaxRequestBodyBytes, + MaxResponseBodyBytes: DefaultMaxResponseBodyBytes, + + MaxStreamChunkBytes: DefaultMaxStreamChunkBytes, + MaxStreamTotalBytes: DefaultMaxStreamTotalBytes, +}) + +func RecentCallsCache() *recentCallsCache { + return recentCallsSingleton +} + +func newRecentCallsCache(cfg RecentCallsCacheConfig) *recentCallsCache { + if cfg.Capacity <= 0 { + cfg.Capacity = DefaultRecentCallsCapacity + } + if cfg.MaxRequestBodyBytes <= 0 { + cfg.MaxRequestBodyBytes = DefaultMaxRequestBodyBytes + } + if cfg.MaxResponseBodyBytes <= 0 { + cfg.MaxResponseBodyBytes = DefaultMaxResponseBodyBytes + } + if cfg.MaxStreamChunkBytes <= 0 { + cfg.MaxStreamChunkBytes = DefaultMaxStreamChunkBytes + } + if cfg.MaxStreamTotalBytes <= 0 { + cfg.MaxStreamTotalBytes = DefaultMaxStreamTotalBytes + } + + return &recentCallsCache{ + cfg: cfg, + buffer: make([]*RecentCallRecord, cfg.Capacity), + } +} + +func (cch *recentCallsCache) BeginFromContext(c *gin.Context, info *relaycommon.RelayInfo, rawRequestBody []byte) uint64 { + if cch == nil || c == nil { + return 0 + } + + id := cch.nextID.Add(1) + + path := "" + if c.Request != nil && c.Request.URL != nil { + path = c.Request.URL.Path + } + method := "" + if c.Request != nil { + method = c.Request.Method + } + + userID := common.GetContextKeyInt(c, constant.ContextKeyUserId) + channelID := common.GetContextKeyInt(c, constant.ContextKeyChannelId) + + modelName := "" + if info != nil { + modelName = info.OriginModelName + if modelName == "" { + modelName = info.UpstreamModelName + } + } + + rec := &RecentCallRecord{ + ID: id, + CreatedAt: time.Now().UTC(), + + UserID: userID, + ChannelID: channelID, + ModelName: modelName, + + Method: method, + Path: path, + + Request: RecentCallRequest{ + Method: method, + Path: path, + Header: sanitizeHeaders(c.Request.Header), + }, + } + + rec.Request.BodyType, rec.Request.Body, rec.Request.Truncated, rec.Request.Omitted, rec.Request.OmitReason = + encodeBodyForRecord(c.Request.Header.Get("Content-Type"), rawRequestBody, cch.cfg.MaxRequestBodyBytes) + + c.Set(RecentCallsContextKeyID, id) + cch.put(rec) + return id +} + +func (cch *recentCallsCache) UpsertErrorByContext(c *gin.Context, errMsg string, errType string, errCode string, status int) { + if cch == nil || c == nil { + return + } + id := getRecentCallID(c) + if id == 0 { + return + } + cch.mu.Lock() + defer cch.mu.Unlock() + rec := cch.getLocked(id) + if rec == nil { + return + } + rec.Error = &RecentCallErrorInfo{ + Message: errMsg, + Type: errType, + Code: errCode, + Status: status, + } +} + +func (cch *recentCallsCache) UpsertUpstreamResponseByContext(c *gin.Context, resp *http.Response, rawUpstreamBody []byte) { + if cch == nil || c == nil { + return + } + id := getRecentCallID(c) + if id == 0 { + return + } + + header := map[string]string(nil) + statusCode := 0 + contentType := "" + if resp != nil { + statusCode = resp.StatusCode + contentType = resp.Header.Get("Content-Type") + header = sanitizeHeaders(resp.Header) + } + + bodyType, body, truncated, omitted, omitReason := encodeBodyForRecord(contentType, rawUpstreamBody, cch.cfg.MaxResponseBodyBytes) + + cch.mu.Lock() + defer cch.mu.Unlock() + rec := cch.getLocked(id) + if rec == nil { + return + } + rec.Response = &RecentCallUpstreamResponse{ + StatusCode: statusCode, + Header: header, + BodyType: bodyType, + Body: body, + Truncated: truncated, + Omitted: omitted, + OmitReason: omitReason, + } +} + +func (cch *recentCallsCache) EnsureStreamByContext(c *gin.Context, resp *http.Response) { + if cch == nil || c == nil { + return + } + id := getRecentCallID(c) + if id == 0 { + return + } + + cch.mu.Lock() + defer cch.mu.Unlock() + rec := cch.getLocked(id) + if rec == nil { + return + } + if rec.Stream == nil { + rec.Stream = &RecentCallUpstreamStream{ + Chunks: make([]string, 0, 32), + } + } + if rec.Response == nil && resp != nil { + rec.Response = &RecentCallUpstreamResponse{ + StatusCode: resp.StatusCode, + Header: sanitizeHeaders(resp.Header), + } + } +} + +func (cch *recentCallsCache) AppendStreamChunkByContext(c *gin.Context, chunk string) { + if cch == nil || c == nil || chunk == "" { + return + } + id := getRecentCallID(c) + if id == 0 { + return + } + + chunkTruncated := false + if cch.cfg.MaxStreamChunkBytes > 0 && len(chunk) > cch.cfg.MaxStreamChunkBytes { + chunk = chunk[:cch.cfg.MaxStreamChunkBytes] + chunkTruncated = true + } + + cch.mu.Lock() + defer cch.mu.Unlock() + rec := cch.getLocked(id) + if rec == nil { + return + } + if rec.Stream == nil { + rec.Stream = &RecentCallUpstreamStream{ + Chunks: make([]string, 0, 32), + } + } + + if chunkTruncated { + rec.Stream.ChunksTruncated = true + } + + if cch.cfg.MaxStreamTotalBytes > 0 && rec.Stream.StreamBytes+len(chunk) > cch.cfg.MaxStreamTotalBytes { + rec.Stream.ChunksTruncated = true + return + } + + rec.Stream.Chunks = append(rec.Stream.Chunks, chunk) + rec.Stream.StreamBytes += len(chunk) +} + +func (cch *recentCallsCache) FinalizeStreamAggregatedTextByContext(c *gin.Context, aggregated string) { + if cch == nil || c == nil { + return + } + id := getRecentCallID(c) + if id == 0 { + return + } + + truncated := false + if cch.cfg.MaxResponseBodyBytes > 0 && len(aggregated) > cch.cfg.MaxResponseBodyBytes { + aggregated = aggregated[:cch.cfg.MaxResponseBodyBytes] + truncated = true + } + + cch.mu.Lock() + defer cch.mu.Unlock() + rec := cch.getLocked(id) + if rec == nil { + return + } + if rec.Stream == nil { + rec.Stream = &RecentCallUpstreamStream{ + Chunks: make([]string, 0, 32), + } + } + rec.Stream.AggregatedText = aggregated + rec.Stream.AggregatedTruncated = truncated +} + +func (cch *recentCallsCache) Get(id uint64) (*RecentCallRecord, bool) { + if cch == nil || id == 0 { + return nil, false + } + cch.mu.RLock() + defer cch.mu.RUnlock() + rec := cch.getLocked(id) + if rec == nil { + return nil, false + } + dup := *rec + if rec.Response != nil { + r := *rec.Response + dup.Response = &r + } + if rec.Stream != nil { + s := *rec.Stream + if rec.Stream.Chunks != nil { + s.Chunks = append([]string(nil), rec.Stream.Chunks...) + } + s.StreamBytes = 0 + dup.Stream = &s + } + if rec.Error != nil { + e := *rec.Error + dup.Error = &e + } + return &dup, true +} + +func (cch *recentCallsCache) List(limit int, beforeID uint64) []*RecentCallRecord { + if cch == nil { + return nil + } + if limit <= 0 { + limit = cch.cfg.Capacity + } + if limit > cch.cfg.Capacity { + limit = cch.cfg.Capacity + } + + cch.mu.RLock() + defer cch.mu.RUnlock() + + items := make([]*RecentCallRecord, 0, limit) + for _, rec := range cch.buffer { + if rec == nil { + continue + } + if beforeID != 0 && rec.ID >= beforeID { + continue + } + items = append(items, rec) + } + + sort.Slice(items, func(i, j int) bool { return items[i].ID > items[j].ID }) + if len(items) > limit { + items = items[:limit] + } + + out := make([]*RecentCallRecord, 0, len(items)) + for _, rec := range items { + dup := *rec + if rec.Response != nil { + r := *rec.Response + dup.Response = &r + } + if rec.Stream != nil { + s := *rec.Stream + if rec.Stream.Chunks != nil { + s.Chunks = append([]string(nil), rec.Stream.Chunks...) + } + s.StreamBytes = 0 + dup.Stream = &s + } + if rec.Error != nil { + e := *rec.Error + dup.Error = &e + } + out = append(out, &dup) + } + return out +} + +func (cch *recentCallsCache) put(rec *RecentCallRecord) { + if cch == nil || rec == nil { + return + } + idx := int(rec.ID % uint64(cch.cfg.Capacity)) + cch.mu.Lock() + cch.buffer[idx] = rec + cch.mu.Unlock() +} + +func (cch *recentCallsCache) getLocked(id uint64) *RecentCallRecord { + if cch == nil || id == 0 { + return nil + } + idx := int(id % uint64(cch.cfg.Capacity)) + rec := cch.buffer[idx] + if rec == nil || rec.ID != id { + return nil + } + return rec +} + +func getRecentCallID(c *gin.Context) uint64 { + if c == nil { + return 0 + } + v, ok := c.Get(RecentCallsContextKeyID) + if !ok || v == nil { + return 0 + } + switch t := v.(type) { + case uint64: + return t + case uint: + return uint64(t) + case int: + if t < 0 { + return 0 + } + return uint64(t) + case int64: + if t < 0 { + return 0 + } + return uint64(t) + case string: + parsed, _ := strconv.ParseUint(t, 10, 64) + return parsed + default: + return 0 + } +} + +func sanitizeHeaders(h http.Header) map[string]string { + if h == nil { + return nil + } + out := make(map[string]string, len(h)) + for k, vals := range h { + if len(vals) == 0 { + continue + } + v := strings.Join(vals, ",") + switch strings.ToLower(k) { + case "authorization", "x-api-key", "x-goog-api-key", "proxy-authorization": + out[k] = "***masked***" + default: + out[k] = v + } + } + return out +} + +func encodeBodyForRecord(contentType string, body []byte, limit int) (bodyType string, encoded string, truncated bool, omitted bool, omitReason string) { + if len(body) == 0 { + return "unknown", "", false, true, "empty" + } + + ct := strings.ToLower(strings.TrimSpace(contentType)) + if strings.HasPrefix(ct, "application/json") || strings.HasPrefix(ct, "text/") || strings.Contains(ct, "application/x-www-form-urlencoded") { + bodyType = "text" + if strings.HasPrefix(ct, "application/json") { + bodyType = "json" + } + s := string(body) + if limit > 0 && len(s) > limit { + s = s[:limit] + truncated = true + } + return bodyType, s, truncated, false, "" + } + + if strings.Contains(ct, "multipart/form-data") { + return "binary", "", false, true, "multipart_form_data" + } + + if strings.HasPrefix(ct, "application/octet-stream") { + // base64 with limit + b := body + if limit > 0 && len(b) > limit { + b = b[:limit] + truncated = true + } + return "binary", base64.StdEncoding.EncodeToString(b), truncated, false, "" + } + + // Unknown content-type: best-effort treat as text if printable-ish, otherwise omit + bodyType = "unknown" + s := string(body) + if limit > 0 && len(s) > limit { + s = s[:limit] + truncated = true + } + return bodyType, s, truncated, false, "" +} \ No newline at end of file diff --git a/setting/rate_limit.go b/setting/rate_limit.go index 413f3958d759..ca2fa5afdf56 100644 --- a/setting/rate_limit.go +++ b/setting/rate_limit.go @@ -4,7 +4,10 @@ import ( "encoding/json" "fmt" "math" + "strconv" + "strings" "sync" + "unicode" "github.com/QuantumNous/new-api/common" ) @@ -14,6 +17,7 @@ var ModelRequestRateLimitDurationMinutes = 1 var ModelRequestRateLimitCount = 0 var ModelRequestRateLimitSuccessCount = 1000 var ModelRequestRateLimitGroup = map[string][2]int{} +var ModelRequestRateLimitExemptUserIDs = map[int]struct{}{} var ModelRequestRateLimitMutex sync.RWMutex func ModelRequestRateLimitGroup2JSONString() string { @@ -27,6 +31,49 @@ func ModelRequestRateLimitGroup2JSONString() string { return string(jsonBytes) } +func ParseModelRequestRateLimitExemptUserIDs(raw string) (map[int]struct{}, error) { + ids := make(map[int]struct{}) + for _, token := range strings.FieldsFunc(raw, func(r rune) bool { + return r == ',' || r == '\n' || r == '\r' || r == '\t' || unicode.IsSpace(r) + }) { + token = strings.TrimSpace(token) + if token == "" { + continue + } + id, err := strconv.Atoi(token) + if err != nil { + return nil, fmt.Errorf("invalid userId: %s", token) + } + if id <= 0 { + continue + } + ids[id] = struct{}{} + } + return ids, nil +} + +func UpdateModelRequestRateLimitExemptUserIDs(raw string) error { + ModelRequestRateLimitMutex.Lock() + defer ModelRequestRateLimitMutex.Unlock() + + ids, err := ParseModelRequestRateLimitExemptUserIDs(raw) + if err != nil { + return err + } + ModelRequestRateLimitExemptUserIDs = ids + return nil +} + +func IsModelRequestRateLimitExemptUser(userID int) bool { + if userID <= 0 { + return false + } + ModelRequestRateLimitMutex.RLock() + defer ModelRequestRateLimitMutex.RUnlock() + _, ok := ModelRequestRateLimitExemptUserIDs[userID] + return ok +} + func UpdateModelRequestRateLimitGroupByJSONString(jsonStr string) error { ModelRequestRateLimitMutex.RLock() defer ModelRequestRateLimitMutex.RUnlock() diff --git a/setting/rate_limit_test.go b/setting/rate_limit_test.go new file mode 100644 index 000000000000..e8cc1c13f4b8 --- /dev/null +++ b/setting/rate_limit_test.go @@ -0,0 +1,57 @@ +package setting + +import "testing" + +func TestParseModelRequestRateLimitExemptUserIDs(t *testing.T) { + t.Run("empty", func(t *testing.T) { + ids, err := ParseModelRequestRateLimitExemptUserIDs("") + if err != nil { + t.Fatalf("expected nil error, got %v", err) + } + if len(ids) != 0 { + t.Fatalf("expected empty map, got %v", ids) + } + }) + + t.Run("comma and newline separated", func(t *testing.T) { + ids, err := ParseModelRequestRateLimitExemptUserIDs("1,2\n3\r\n4\t5") + if err != nil { + t.Fatalf("expected nil error, got %v", err) + } + for _, want := range []int{1, 2, 3, 4, 5} { + if _, ok := ids[want]; !ok { + t.Fatalf("expected id %d to exist, got %v", want, ids) + } + } + }) + + t.Run("ignores non-positive", func(t *testing.T) { + ids, err := ParseModelRequestRateLimitExemptUserIDs("0,-1,2") + if err != nil { + t.Fatalf("expected nil error, got %v", err) + } + if _, ok := ids[2]; !ok || len(ids) != 1 { + t.Fatalf("expected only id 2, got %v", ids) + } + }) + + t.Run("invalid token", func(t *testing.T) { + _, err := ParseModelRequestRateLimitExemptUserIDs("1,abc") + if err == nil { + t.Fatalf("expected error, got nil") + } + }) +} + +func TestIsModelRequestRateLimitExemptUser(t *testing.T) { + if err := UpdateModelRequestRateLimitExemptUserIDs("10,20"); err != nil { + t.Fatalf("UpdateModelRequestRateLimitExemptUserIDs error: %v", err) + } + + if !IsModelRequestRateLimitExemptUser(10) { + t.Fatalf("expected user 10 to be exempt") + } + if IsModelRequestRateLimitExemptUser(11) { + t.Fatalf("expected user 11 to not be exempt") + } +} \ No newline at end of file diff --git a/web/package.json b/web/package.json index 9ac8e266eb39..2f8c75720ce5 100644 --- a/web/package.json +++ b/web/package.json @@ -10,6 +10,7 @@ "@visactor/react-vchart": "~1.8.8", "@visactor/vchart": "~1.8.8", "@visactor/vchart-semi-theme": "~1.8.8", + "antd": "^5.29.3", "axios": "1.12.0", "clsx": "^2.1.1", "country-flag-icons": "^1.5.19", diff --git a/web/public/oauth-redirect.html b/web/public/oauth-redirect.html new file mode 100644 index 000000000000..df9c5a6eaa73 --- /dev/null +++ b/web/public/oauth-redirect.html @@ -0,0 +1,203 @@ + + +
+ + +请稍候,正在为您跳转回原站点...
+ ++ 按小时查看单个模型的健康度趋势 +
+