From 141eac106a07744d15fd101ba583fe76887a6696 Mon Sep 17 00:00:00 2001 From: chengqian Date: Mon, 3 Aug 2026 15:29:38 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E6=B7=BB=E5=8A=A0rpm/tpm=E6=8C=87?= =?UTF-8?q?=E6=A0=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/tenant/v1/dashboard.go | 2 + internal/logic/common/realtime_metrics.go | 151 ++++++++++++++++++ .../logic/common/realtime_metrics_test.go | 100 ++++++++++++ internal/logic/monitor/monitor.go | 6 +- internal/logic/monitor/request_tracker.go | 2 + internal/logic/relay/provider.go | 13 +- internal/logic/tenant/dashboard.go | 5 + .../src/views/monitor/RealtimeMonitorPage.vue | 36 +++++ .../src/views/dashboard/DashboardPage.vue | 75 ++++++++- 9 files changed, 380 insertions(+), 10 deletions(-) create mode 100644 internal/logic/common/realtime_metrics.go create mode 100644 internal/logic/common/realtime_metrics_test.go diff --git a/api/tenant/v1/dashboard.go b/api/tenant/v1/dashboard.go index 0cb15a0..a1083a7 100644 --- a/api/tenant/v1/dashboard.go +++ b/api/tenant/v1/dashboard.go @@ -12,6 +12,8 @@ type TenantDashboardRes struct { Today map[string]any `json:"today"` Month map[string]any `json:"month"` Wallet map[string]any `json:"wallet"` + Rpm int64 `json:"rpm" dc:"最近60秒滑动窗口请求数（本租户）"` + Tpm int64 `json:"tpm" dc:"最近60秒滑动窗口token数（本租户）"` ActiveKeys int `json:"active_keys"` MemberCount int `json:"member_count"` } diff --git a/internal/logic/common/realtime_metrics.go b/internal/logic/common/realtime_metrics.go new file mode 100644 index 0000000..31f2aa0 --- /dev/null +++ b/internal/logic/common/realtime_metrics.go @@ -0,0 +1,151 @@ +package common + +import ( + "context" + "fmt" + "math" + "time" + + "github.com/gogf/gf/v2/frame/g" +) + +// 实时 RPM/TPM 统计（参考 new-api 的"最近 60 秒"口径，但不查日志表）： +// 写入侧在每条用量记录落地时向 Redis 的 10 秒桶原子累加请求数与 token 数（全局 + 租户两个维度）， +// 读取侧对最近 60 秒做滑动窗口聚合（6 个整桶 + 最旧桶按窗口覆盖比例加权）。 +// 相比 new-api 每次实时 COUNT 日志表，写入 O(1)、读取 O(桶数)，与 bil_usage_logs 的规模解耦。 +// 指标为 best-effort：Redis 不可用时计数丢弃、查询返回 0，不影响请求主链路。 +const ( + rtMetricsKeyPrefix = "rt_metrics:" // rt_metrics:{scope}:{bucketStartUnix} + rtMetricsBucketSeconds = 10 // 桶粒度（秒） + rtMetricsWindowSeconds = 60 // 统计窗口（秒），与 new-api 的 RPM/TPM 口径一致 + rtMetricsKeyTTL = 180 // 桶 TTL（秒），覆盖读取窗口即可 +) + +// rtMetricsWindowBuckets 窗口内整桶数 +const rtMetricsWindowBuckets = rtMetricsWindowSeconds / rtMetricsBucketSeconds + +// rtMetricsIncrLua 一次 RTT 原子累加所有维度的桶计数并续期。 +// KEYS = 各维度的当前桶 key；ARGV[1] = token 数，ARGV[2] = TTL 秒。 +const rtMetricsIncrLua = ` +for i = 1, #KEYS do + redis.call("HINCRBY", KEYS[i], "req", 1) + redis.call("HINCRBY", KEYS[i], "tok", tonumber(ARGV[1])) + redis.call("EXPIRE", KEYS[i], tonumber(ARGV[2])) +end +return 1 +` + +// rtMetricsReadLua 一次 RTT 读取一组桶的 req/tok 计数。 +// KEYS = 桶 key（从旧到新）；返回扁平数组 [req1, tok1, req2, tok2, ...]，缺失桶为 0。 +const rtMetricsReadLua = ` +local res = {} +for i = 1, #KEYS do + res[2*i-1] = tonumber(redis.call("HGET", KEYS[i], "req") or "0") + res[2*i] = tonumber(redis.call("HGET", KEYS[i], "tok") or "0") +end +return res +` + +// rtMetricsBucketKey 桶 key：scope 为 "g"（全局）或 "t:{tenantID}"（租户） +func rtMetricsBucketKey(scope string, bucketStart int64) string { + return fmt.Sprintf("%s%s:%d", rtMetricsKeyPrefix, scope, bucketStart) +} + +func rtMetricsTenantScope(tenantID int64) string { + return fmt.Sprintf("t:%d", tenantID) +} + +// RecordRealtimeMetrics 累加实时 RPM/TPM 计数（全局 + 租户维度）。 +// 在用量记录入口（DataProvider.RecordUsage）调用，每条用量记录计 1 次请求、tokens 个 token。 +// 内部异步执行且脱离请求 ctx 的取消，失败仅记 debug 日志，不影响主链路。 +func RecordRealtimeMetrics(ctx context.Context, tenantID int64, tokens int64) { + if tokens < 0 { + tokens = 0 + } + go func() { + defer func() { + if r := recover(); r != nil { + g.Log().Warningf(context.Background(), "realtime metrics incr panic: tenant=%d panic=%v", tenantID, r) + } + }() + bgCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second) + defer cancel() + + now := time.Now().Unix() + bucketStart := now - now%rtMetricsBucketSeconds + + keys := []any{rtMetricsBucketKey("g", bucketStart)} + if tenantID > 0 { + keys = append(keys, rtMetricsBucketKey(rtMetricsTenantScope(tenantID), bucketStart)) + } + + args := make([]any, 0, len(keys)+4) + args = append(args, rtMetricsIncrLua, len(keys)) + args = append(args, keys...) + args = append(args, tokens, rtMetricsKeyTTL) + if _, err := g.Redis().Do(bgCtx, "EVAL", args...); err != nil { + // 指标是 best-effort，Redis 抖动只记 debug，不刷告警 + g.Log().Debugf(bgCtx, "realtime metrics incr failed: tenant=%d err=%v", tenantID, err) + } + }() +} + +// GetRealtimeMetrics 返回最近 60 秒滑动窗口的 RPM（请求数）与 TPM（token 数）。 +// tenantID <= 0 时返回全局维度，否则返回指定租户维度。 +// Redis 不可用时返回 0, 0（看板容忍指标短暂缺失）。 +func GetRealtimeMetrics(ctx context.Context, tenantID int64) (rpm int64, tpm int64) { + scope := "g" + if tenantID > 0 { + scope = rtMetricsTenantScope(tenantID) + } + + now := time.Now().Unix() + bucketStart := now - now%rtMetricsBucketSeconds + elapsed := now - bucketStart + + // 桶 key 从旧到新：窗口涉及 windowBuckets+1 个桶（最旧桶只有部分时间落在窗口内） + keys := make([]any, 0, rtMetricsWindowBuckets+1) + for i := rtMetricsWindowBuckets; i >= 0; i-- { + keys = append(keys, rtMetricsBucketKey(scope, bucketStart-int64(i)*rtMetricsBucketSeconds)) + } + + args := make([]any, 0, len(keys)+2) + args = append(args, rtMetricsReadLua, len(keys)) + args = append(args, keys...) + result, err := g.Redis().Do(ctx, "EVAL", args...) + if err != nil { + g.Log().Debugf(ctx, "realtime metrics read failed: scope=%s err=%v", scope, err) + return 0, 0 + } + + vals := result.Vars() + reqBuckets := make([]int64, 0, rtMetricsWindowBuckets+1) + tokBuckets := make([]int64, 0, rtMetricsWindowBuckets+1) + for i := 0; i+1 < len(vals); i += 2 { + reqBuckets = append(reqBuckets, vals[i].Int64()) + tokBuckets = append(tokBuckets, vals[i+1].Int64()) + } + + return slidingWindowSum(reqBuckets, elapsed), slidingWindowSum(tokBuckets, elapsed) +} + +// slidingWindowSum 对从旧到新排列的桶计数做 60 秒滑动窗口聚合。 +// 最旧桶只有 (bucketSeconds - elapsed) 秒落在窗口内，按覆盖比例加权，其余桶全额计入； +// elapsed 为当前时间在当前桶内经过的秒数（0 ~ bucketSeconds-1）。 +func slidingWindowSum(buckets []int64, elapsed int64) int64 { + if len(buckets) == 0 { + return 0 + } + oldestWeight := float64(rtMetricsBucketSeconds-elapsed) / float64(rtMetricsBucketSeconds) + if oldestWeight < 0 { + oldestWeight = 0 + } + if oldestWeight > 1 { + oldestWeight = 1 + } + sum := float64(buckets[0]) * oldestWeight + for _, v := range buckets[1:] { + sum += float64(v) + } + return int64(math.Round(sum)) +} diff --git a/internal/logic/common/realtime_metrics_test.go b/internal/logic/common/realtime_metrics_test.go new file mode 100644 index 0000000..326cdcb --- /dev/null +++ b/internal/logic/common/realtime_metrics_test.go @@ -0,0 +1,100 @@ +package common + +import "testing" + +// 滑动窗口聚合的纯函数单测：桶从旧到新排列，最旧桶按 (bucketSeconds-elapsed)/bucketSeconds 加权。 +func TestSlidingWindowSum(t *testing.T) { + tests := []struct { + name string + buckets []int64 + elapsed int64 + want int64 + }{ + { + name: "空桶返回0", + buckets: nil, + elapsed: 5, + want: 0, + }, + { + name: "全零桶", + buckets: []int64{0, 0, 0, 0, 0, 0, 0}, + elapsed: 3, + want: 0, + }, + { + // elapsed=0：当前桶刚开始，最旧桶完整落在窗口内（权重 1），7 桶全额相加 + name: "桶起点最旧桶全额计入", + buckets: []int64{10, 10, 10, 10, 10, 10, 10}, + elapsed: 0, + want: 70, + }, + { + // elapsed=5：最旧桶权重 (10-5)/10 = 0.5 → 10*0.5 + 60 = 65 + name: "桶中点最旧桶半权重", + buckets: []int64{10, 10, 10, 10, 10, 10, 10}, + elapsed: 5, + want: 65, + }, + { + // elapsed=9：最旧桶权重 0.1 → 10*0.1 + 60 = 61 + name: "桶末尾最旧桶接近零权重", + buckets: []int64{10, 10, 10, 10, 10, 10, 10}, + elapsed: 9, + want: 61, + }, + { + // 只有最旧桶有值：elapsed=5 → 100*0.5 = 50 + name: "仅最旧桶有计数", + buckets: []int64{100, 0, 0, 0, 0, 0, 0}, + elapsed: 5, + want: 50, + }, + { + // 只有当前桶有值：不加权，全额计入 + name: "仅当前桶有计数", + buckets: []int64{0, 0, 0, 0, 0, 0, 100}, + elapsed: 5, + want: 100, + }, + { + // 加权结果四舍五入：7*0.3 = 2.1 → 2 + 5 = 7 + name: "加权结果四舍五入", + buckets: []int64{7, 5}, + elapsed: 7, + want: 7, + }, + { + // 异常 elapsed 超过桶宽：权重钳制为 0，最旧桶不计入 + name: "elapsed超界权重钳制为零", + buckets: []int64{100, 10}, + elapsed: 15, + want: 10, + }, + { + // 异常负 elapsed：权重钳制为 1，最旧桶全额计入 + name: "负elapsed权重钳制为一", + buckets: []int64{100, 10}, + elapsed: -3, + want: 110, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := slidingWindowSum(tt.buckets, tt.elapsed); got != tt.want { + t.Errorf("slidingWindowSum(%v, %d) = %d, want %d", tt.buckets, tt.elapsed, got, tt.want) + } + }) + } +} + +// 桶 key 生成格式校验 +func TestRtMetricsBucketKey(t *testing.T) { + if got := rtMetricsBucketKey("g", 1700000000); got != "rt_metrics:g:1700000000" { + t.Errorf("global bucket key = %q", got) + } + if got := rtMetricsBucketKey(rtMetricsTenantScope(42), 1700000000); got != "rt_metrics:t:42:1700000000" { + t.Errorf("tenant bucket key = %q", got) + } +} diff --git a/internal/logic/monitor/monitor.go b/internal/logic/monitor/monitor.go index 36d09c4..788faf9 100644 --- a/internal/logic/monitor/monitor.go +++ b/internal/logic/monitor/monitor.go @@ -6,6 +6,7 @@ import ( "github.com/gogf/gf/v2/frame/g" v1 "github.com/qianfree/team-api/api/admin/v1" + lcommon "github.com/qianfree/team-api/internal/logic/common" "github.com/qianfree/team-api/internal/middleware" do "github.com/qianfree/team-api/internal/model/do" "github.com/qianfree/team-api/internal/service" @@ -172,7 +173,10 @@ func (s *sMonitor) RedisPool(ctx context.Context, _ *v1.MonitorRedisPoolReq) (*v } func (s *sMonitor) Realtime(ctx context.Context, _ *v1.MonitorRealtimeReq) (*v1.MonitorRealtimeRes, error) { - return &v1.MonitorRealtimeRes{Data: GetRealtimeData()}, nil + data := GetRealtimeData() + // 实时 RPM/TPM（Redis 60 秒滑动窗口，全平台维度；Redis 不可用时为 0） + data.Rpm, data.Tpm = lcommon.GetRealtimeMetrics(ctx, 0) + return &v1.MonitorRealtimeRes{Data: data}, nil } // ===================== Alert Rules ===================== diff --git a/internal/logic/monitor/request_tracker.go b/internal/logic/monitor/request_tracker.go index 06bfeaa..512bc91 100644 --- a/internal/logic/monitor/request_tracker.go +++ b/internal/logic/monitor/request_tracker.go @@ -55,6 +55,8 @@ type RealtimeData struct { ByChannel map[string]int `json:"by_channel"` ByTenant map[int64]int `json:"by_tenant"` SyncImagePool SyncImagePoolSnapshot `json:"sync_image_pool"` + Rpm int64 `json:"rpm"` // 最近60秒滑动窗口请求数（全平台，由 Realtime 方法填充） + Tpm int64 `json:"tpm"` // 最近60秒滑动窗口token数（全平台，由 Realtime 方法填充） } // snapshotRing is a fixed-size circular buffer for arbitrary snapshot types. diff --git a/internal/logic/relay/provider.go b/internal/logic/relay/provider.go index 1be0617..7e23bd5 100644 --- a/internal/logic/relay/provider.go +++ b/internal/logic/relay/provider.go @@ -106,9 +106,9 @@ func (p *DataProviderImpl) CheckTenantModelAccess(ctx context.Context, tenantID // 缓存结构体（与 accessRow 一致，用于缓存序列化） type cachedAccess struct { - Enabled bool `json:"enabled"` - ChannelScope string `json:"channel_scope"` - IsNil bool `json:"is_nil"` // 标记数据库中无记录（需走分组权限） + Enabled bool `json:"enabled"` + ChannelScope string `json:"channel_scope"` + IsNil bool `json:"is_nil"` // 标记数据库中无记录（需走分组权限） } cacheKey := fmt.Sprintf("%d:%s", tenantID, modelName) @@ -325,6 +325,13 @@ func loadModelCached(ctx context.Context, modelName string) (*modelInfoCached, e // RecordUsage 实现 DataProvider.RecordUsage func (p *DataProviderImpl) RecordUsage(ctx context.Context, record *common.UsageRecord) { lcommon.DefaultUsageLogWriter.Submit(buildUsageLogDO(record)) + + // 实时 RPM/TPM 计数（Redis 滑动窗口桶，内部异步、best-effort） + tokens := record.TotalTokens + if tokens == 0 { + tokens = record.PromptTokens + record.CompletionTokens + } + lcommon.RecordRealtimeMetrics(ctx, record.TenantID, int64(tokens)) } // buildUsageLogDO 将 UsageRecord 转换为 DO 对象 diff --git a/internal/logic/tenant/dashboard.go b/internal/logic/tenant/dashboard.go index 3366938..576310b 100644 --- a/internal/logic/tenant/dashboard.go +++ b/internal/logic/tenant/dashboard.go @@ -98,6 +98,9 @@ func (s *sTenant) Dashboard(ctx context.Context, req *v1.TenantDashboardReq) (*v return nil, err } + // 实时 RPM/TPM（Redis 滑动窗口，本租户维度；Redis 不可用时为 0） + rpm, tpm := common.GetRealtimeMetrics(ctx, tenantID) + return &v1.TenantDashboardRes{ Today: map[string]any{ "requests": todayRow.Requests, @@ -117,6 +120,8 @@ func (s *sTenant) Dashboard(ctx context.Context, req *v1.TenantDashboardReq) (*v "available": roundUSD(wallet.Balance - wallet.FrozenBalance), "warning_threshold": roundUSD(wallet.WarningThreshold), }, + Rpm: rpm, + Tpm: tpm, ActiveKeys: activeKeys, MemberCount: memberCount, }, nil diff --git a/web/admin/src/views/monitor/RealtimeMonitorPage.vue b/web/admin/src/views/monitor/RealtimeMonitorPage.vue index 578f758..81af7b6 100644 --- a/web/admin/src/views/monitor/RealtimeMonitorPage.vue +++ b/web/admin/src/views/monitor/RealtimeMonitorPage.vue @@ -291,6 +291,13 @@ onUnmounted(() => {
+ @@ -721,6 +728,35 @@ onUnmounted(() => { font-weight: 500; } +/* 并发趋势卡片 header 右侧的实时 RPM/TPM */ +.rt-rate-group { + display: inline-flex; + align-items: center; + gap: 10px; +} +.rt-rate { + display: inline-flex; + align-items: baseline; + gap: 5px; +} +.rt-rate-label { + font-size: 11px; + color: var(--color-text-3); + font-weight: 500; + letter-spacing: 0.5px; +} +.rt-rate-value { + font-size: 15px; + font-weight: 600; + font-family: 'Menlo', 'Monaco', 'Courier New', monospace; + line-height: 1; +} +.rt-rate-sep { + width: 1px; + height: 14px; + background: var(--color-border); +} + /* Sync-image worker pool */ .pool-grid { display: grid; diff --git a/web/tenant/src/views/dashboard/DashboardPage.vue b/web/tenant/src/views/dashboard/DashboardPage.vue index 79f15c8..04869b9 100644 --- a/web/tenant/src/views/dashboard/DashboardPage.vue +++ b/web/tenant/src/views/dashboard/DashboardPage.vue @@ -1,5 +1,5 @@