diff --git a/backend/internal/repository/ops_repo_dashboard.go b/backend/internal/repository/ops_repo_dashboard.go index a22235a805..6fa037d370 100644 --- a/backend/internal/repository/ops_repo_dashboard.go +++ b/backend/internal/repository/ops_repo_dashboard.go @@ -33,18 +33,63 @@ func (r *opsRepository) GetDashboardOverview(ctx context.Context, filter *servic mode = service.OpsQueryModeRaw } + var overview *service.OpsDashboardOverview + var err error switch mode { case service.OpsQueryModePreagg: - return r.getDashboardOverviewPreaggregated(ctx, filter) + overview, err = r.getDashboardOverviewPreaggregated(ctx, filter) case service.OpsQueryModeAuto: - out, err := r.getDashboardOverviewPreaggregated(ctx, filter) - if err != nil && errors.Is(err, service.ErrOpsPreaggregatedNotPopulated) { - return r.getDashboardOverviewRaw(ctx, filter) + overview, err = r.getDashboardOverviewPreaggregated(ctx, filter) + if errors.Is(err, service.ErrOpsPreaggregatedNotPopulated) { + overview, err = r.getDashboardOverviewRaw(ctx, filter) } - return out, err default: - return r.getDashboardOverviewRaw(ctx, filter) + overview, err = r.getDashboardOverviewRaw(ctx, filter) } + if err != nil { + return nil, err + } + // Percentiles must use individual rates over the entire window, even in + // preagg mode: hourly percentiles cannot be averaged into window percentiles. + rateCtx, cancelRate := context.WithTimeout(ctx, opsRawLatencyQueryTimeout) + defer cancelRate() + overview.OutputTPS, err = r.queryOutputTPS(rateCtx, filter) + if err != nil && !isQueryTimeoutErr(err) { + return nil, err + } + return overview, nil +} + +func (r *opsRepository) queryOutputTPS(ctx context.Context, filter *service.OpsDashboardFilter) (*service.OpsOutputTPS, error) { + join, where, args, _ := buildUsageWhere(filter, filter.StartTime.UTC(), filter.EndTime.UTC(), 1) + query := ` +SELECT + percentile_cont(0.05) WITHIN GROUP (ORDER BY output_tps), + percentile_cont(0.10) WITHIN GROUP (ORDER BY output_tps), + percentile_cont(0.50) WITHIN GROUP (ORDER BY output_tps), + AVG(output_tps), + COUNT(*) +FROM ( + SELECT ul.output_tokens * 1000.0 / NULLIF(ul.duration_ms, 0) AS output_tps + FROM usage_logs ul + ` + join + ` + ` + where + ` + AND ul.output_tokens > 0 AND ul.duration_ms > 0 + AND ul.image_count = 0 AND ul.image_output_tokens = 0 + AND COALESCE(ul.billing_mode, '') <> 'image' + ` + fmt.Sprintf("AND ul.request_type IN (%d, %d, %d, %d)", + service.RequestTypeSync, service.RequestTypeStream, service.RequestTypeWSV2, service.RequestTypeCyberBlocked) + ` +) rates` + var stats service.OpsOutputTPS + if err := r.db.QueryRowContext(ctx, query, args...).Scan(&stats.P5, &stats.P10, &stats.P50, &stats.Avg, &stats.SampleCount); err != nil { + // lib/pq reports query cancellation without wrapping the context error. + // Preserve deadline vs caller cancellation for the overview's timeout policy. + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + return nil, err + } + return &stats, nil } func (r *opsRepository) getDashboardOverviewRaw(ctx context.Context, filter *service.OpsDashboardFilter) (*service.OpsDashboardOverview, error) { diff --git a/backend/internal/repository/ops_repo_dashboard_integration_test.go b/backend/internal/repository/ops_repo_dashboard_integration_test.go new file mode 100644 index 0000000000..5cc331f524 --- /dev/null +++ b/backend/internal/repository/ops_repo_dashboard_integration_test.go @@ -0,0 +1,124 @@ +//go:build integration + +package repository + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" +) + +func TestOpsOutputTPSDistribution(t *testing.T) { + ctx := context.Background() + client := testEntClient(t) + user := mustCreateUser(t, client, &service.User{}) + key := mustCreateApiKey(t, client, &service.APIKey{UserID: user.ID}) + account := mustCreateAccount(t, client, &service.Account{Name: "output-tps", Platform: service.PlatformAnthropic}) + group := mustCreateGroup(t, client, &service.Group{Name: "output-tps", Platform: service.PlatformOpenAI}) + otherGroup := mustCreateGroup(t, client, &service.Group{Name: "other-output-tps", Platform: service.PlatformOpenAI}) + start := time.Date(2021, 1, 17, 0, 30, 0, 0, time.UTC) + end := start.Add(6 * time.Hour) + t.Cleanup(func() { + _, _ = integrationDB.ExecContext(ctx, "DELETE FROM usage_logs WHERE user_id = $1", user.ID) + _, _ = integrationDB.ExecContext(ctx, "DELETE FROM api_keys WHERE id = $1", key.ID) + _, _ = integrationDB.ExecContext(ctx, "DELETE FROM accounts WHERE id = $1", account.ID) + _, _ = integrationDB.ExecContext(ctx, "DELETE FROM ops_metrics_hourly WHERE bucket_start >= $1 AND bucket_start < $2", start.Truncate(time.Hour), end) + _, _ = integrationDB.ExecContext(ctx, "DELETE FROM groups WHERE id IN ($1, $2)", group.ID, otherGroup.ID) + _, _ = integrationDB.ExecContext(ctx, "DELETE FROM users WHERE id = $1", user.ID) + }) + insert := func(at time.Time, groupID *int64, output int, duration *int, requestType service.RequestType, imageCount, imageTokens int, billingMode string) { + t.Helper() + _, err := integrationDB.ExecContext(ctx, `INSERT INTO usage_logs + (user_id, api_key_id, account_id, group_id, request_id, model, input_tokens, output_tokens, duration_ms, request_type, image_count, image_output_tokens, billing_mode, created_at, image_size) + VALUES ($1,$2,$3,$4,$5,'gpt-test',100000,$6,$7,$8,$9,$10,$11,$12,CASE WHEN $9 > 0 THEN '1K' ELSE NULL END)`, + user.ID, key.ID, account.ID, groupID, fmt.Sprintf("output-tps-%d-%d", user.ID, time.Now().UnixNano()), output, duration, requestType, imageCount, imageTokens, billingMode, at) + require.NoError(t, err) + } + // Rates 1, 10, 100, 1000: unequal durations catch ratio-of-sums mistakes. + // Put samples in the raw head, stable hourly buckets, and raw tail. + for _, sample := range []struct { + output, duration int + kind service.RequestType + at time.Time + }{ + {1, 1000, service.RequestTypeSync, start}, + {20, 2000, service.RequestTypeStream, start.Add(time.Hour)}, + {100, 1000, service.RequestTypeWSV2, start.Add(2 * time.Hour)}, + {500, 500, service.RequestTypeCyberBlocked, end.Add(-time.Minute)}, + } { + insert(sample.at, &group.ID, sample.output, &sample.duration, sample.kind, 0, 0, "token") + } + positive, zero, negative := 1000, 0, -1 + for _, d := range []*int{nil, &zero, &negative} { + insert(start, &group.ID, 9999, d, service.RequestTypeSync, 0, 0, "token") + } + for _, n := range []int{0, -1} { + insert(start, &group.ID, n, &positive, service.RequestTypeSync, 0, 0, "token") + } + for _, kind := range []service.RequestType{service.RequestTypeLive, service.RequestTypeUnknown} { + insert(start, &group.ID, 9999, &positive, kind, 0, 0, "token") + } + insert(start, &group.ID, 9999, &positive, service.RequestTypeSync, 1, 0, "token") + insert(start, &group.ID, 9999, &positive, service.RequestTypeSync, 0, 1, "token") + insert(start, &group.ID, 9999, &positive, service.RequestTypeSync, 0, 0, "image") + insert(start.Add(-time.Second), &group.ID, 9999, &positive, service.RequestTypeSync, 0, 0, "token") + insert(end, &group.ID, 9999, &positive, service.RequestTypeSync, 0, 0, "token") + insert(start, &otherGroup.ID, 9999, &positive, service.RequestTypeSync, 0, 0, "token") + insert(start, nil, 50, &positive, service.RequestTypeSync, 0, 0, "token") + repo := NewOpsRepository(integrationDB).(*opsRepository) + filter := &service.OpsDashboardFilter{StartTime: start, EndTime: end, Platform: " OpenAI ", GroupID: &group.ID} + assertStats := func(stats *service.OpsOutputTPS) { + t.Helper() + require.NotNil(t, stats) + require.EqualValues(t, 4, stats.SampleCount) + require.NotNil(t, stats.P5) + require.NotNil(t, stats.P10) + require.NotNil(t, stats.P50) + require.NotNil(t, stats.Avg) + require.InDelta(t, 2.35, *stats.P5, 1e-9) + require.InDelta(t, 3.7, *stats.P10, 1e-9) + require.InDelta(t, 55, *stats.P50, 1e-9) + require.InDelta(t, 277.75, *stats.Avg, 1e-9) + } + // Auto first exercises fallback before hourly rows exist. + for _, mode := range []service.OpsQueryMode{service.OpsQueryModeRaw, service.OpsQueryModeAuto} { + filter.QueryMode = mode + overview, err := repo.GetDashboardOverview(ctx, filter) + require.NoError(t, err) + assertStats(overview.OutputTPS) + } + require.NoError(t, repo.UpsertHourlyMetrics(ctx, start.Truncate(time.Hour), end)) + for _, mode := range []service.OpsQueryMode{service.OpsQueryModePreagg, service.OpsQueryModeAuto} { + filter.QueryMode = mode + overview, err := repo.GetDashboardOverview(ctx, filter) + require.NoError(t, err) + assertStats(overview.OutputTPS) + } + filter.Platform = service.PlatformAnthropic + empty, err := repo.queryOutputTPS(ctx, filter) + require.NoError(t, err) + require.Equal(t, &service.OpsOutputTPS{}, empty) + // Ungrouped logs use the account platform; group platform takes precedence. + filter.GroupID = nil + fallback, err := repo.queryOutputTPS(ctx, filter) + require.NoError(t, err) + require.EqualValues(t, 1, fallback.SampleCount) + require.Equal(t, 50.0, *fallback.P50) +} + +func TestOpsOutputTPSQueryDeadline(t *testing.T) { + tx := testTx(t) + _, err := tx.ExecContext(context.Background(), "LOCK TABLE usage_logs IN ACCESS EXCLUSIVE MODE") + require.NoError(t, err) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + repo := NewOpsRepository(integrationDB).(*opsRepository) + now := time.Now().UTC() + stats, err := repo.queryOutputTPS(ctx, &service.OpsDashboardFilter{StartTime: now.Add(-time.Hour), EndTime: now}) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Nil(t, stats) +} diff --git a/backend/internal/repository/ops_repo_dashboard_timeout_test.go b/backend/internal/repository/ops_repo_dashboard_timeout_test.go index 76332ca0a6..d0763b3f96 100644 --- a/backend/internal/repository/ops_repo_dashboard_timeout_test.go +++ b/backend/internal/repository/ops_repo_dashboard_timeout_test.go @@ -2,8 +2,15 @@ package repository import ( "context" + "database/sql/driver" + "errors" "fmt" "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/Wei-Shaw/sub2api/internal/service" + "github.com/stretchr/testify/require" ) func TestIsQueryTimeoutErr(t *testing.T) { @@ -20,3 +27,42 @@ func TestIsQueryTimeoutErr(t *testing.T) { t.Fatalf("wrapped context.Canceled should not be treated as query timeout") } } + +func TestOpsOutputTPSOverviewFailurePolicy(t *testing.T) { + for _, tc := range []struct { + name string + err error + wantError bool + }{ + {"timeout keeps overview", context.DeadlineExceeded, false}, + {"cancellation propagates", context.Canceled, true}, + {"database errors propagate", errors.New("database unavailable"), true}, + } { + t.Run(tc.name, func(t *testing.T) { + db, mock := newSQLMock(t) + repo := &opsRepository{db: db} + // The six existing raw overview queries run before the output-rate query. + for _, count := range []int{2, 13, 6, 2, 6, 2} { + columns := make([]string, count) + values := make([]driver.Value, count) + for i := range columns { + columns[i] = fmt.Sprint(i) + values[i] = int64(0) + } + mock.ExpectQuery("SELECT|WITH").WillReturnRows(sqlmock.NewRows(columns).AddRow(values...)) + } + mock.ExpectQuery(`percentile_cont\(0.05\)`).WillReturnError(tc.err) + start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + overview, err := repo.GetDashboardOverview(context.Background(), &service.OpsDashboardFilter{StartTime: start, EndTime: start.Add(time.Hour), QueryMode: service.OpsQueryModeRaw}) + if tc.wantError { + require.ErrorIs(t, err, tc.err) + require.Nil(t, overview) + } else { + require.NoError(t, err) + require.NotNil(t, overview) + require.Nil(t, overview.OutputTPS) + } + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +} diff --git a/backend/internal/service/ops_dashboard_models.go b/backend/internal/service/ops_dashboard_models.go index f189031bd1..7ab9e69bca 100644 --- a/backend/internal/service/ops_dashboard_models.go +++ b/backend/internal/service/ops_dashboard_models.go @@ -29,6 +29,15 @@ type OpsPercentiles struct { Max *int `json:"max_ms"` } +// OpsOutputTPS describes per-request output rates, not aggregate token throughput. +type OpsOutputTPS struct { + P5 *float64 `json:"p5"` + P10 *float64 `json:"p10"` + P50 *float64 `json:"p50"` + Avg *float64 `json:"avg"` + SampleCount int64 `json:"sample_count"` +} + type OpsDashboardOverview struct { StartTime time.Time `json:"start_time"` EndTime time.Time `json:"end_time"` @@ -62,8 +71,9 @@ type OpsDashboardOverview struct { Upstream429Count int64 `json:"upstream_429_count"` Upstream529Count int64 `json:"upstream_529_count"` - QPS OpsRateSummary `json:"qps"` - TPS OpsRateSummary `json:"tps"` + QPS OpsRateSummary `json:"qps"` + TPS OpsRateSummary `json:"tps"` + OutputTPS *OpsOutputTPS `json:"output_tps"` Duration OpsPercentiles `json:"duration"` TTFT OpsPercentiles `json:"ttft"` diff --git a/frontend/src/api/admin/ops.ts b/frontend/src/api/admin/ops.ts index 8754f19583..f4c1b3740d 100644 --- a/frontend/src/api/admin/ops.ts +++ b/frontend/src/api/admin/ops.ts @@ -63,6 +63,14 @@ export interface OpsDashboardOverview { avg: number } + output_tps?: { + p5: number | null + p10: number | null + p50: number | null + avg: number | null + sample_count: number + } | null + duration: OpsPercentiles ttft: OpsPercentiles } diff --git a/frontend/src/i18n/locales/en/admin/ops.ts b/frontend/src/i18n/locales/en/admin/ops.ts index 8576736b9c..2830b43250 100644 --- a/frontend/src/i18n/locales/en/admin/ops.ts +++ b/frontend/src/i18n/locales/en/admin/ops.ts @@ -4,6 +4,8 @@ export default { description: 'Operational monitoring and troubleshooting', // Dashboard systemHealth: 'System Health', + outputTps: 'Per-request output TPS', + outputTpsSamples: 'Valid samples: {count}', overview: 'Overview', noSystemMetrics: 'No system metrics collected yet.', collectedAt: 'Collected at:', @@ -803,6 +805,7 @@ export default { accountError: 'Error' }, tooltips: { + outputTps: 'Percentiles of each valid usage record’s output tokens / total duration, including first-token wait, within the selected time, platform and group. Output may include reasoning tokens; they are not added again. P50 is the median; P5/P10 show slower requests. Higher is faster. Excludes images, Live and records without positive output or duration. Samples come from retained usage logs; — means no samples or temporarily unavailable statistics.', totalRequests: 'Total number of requests (including both successful and failed requests) in the selected time window.', throughputTrend: 'Requests/QPS + Tokens/TPS in the selected window.', switchRateTrend: 'Trend of account switches / total requests over the last 5 hours (avg switches).', diff --git a/frontend/src/i18n/locales/zh/admin/ops.ts b/frontend/src/i18n/locales/zh/admin/ops.ts index ace6fff1e4..dcb55a1ab5 100644 --- a/frontend/src/i18n/locales/zh/admin/ops.ts +++ b/frontend/src/i18n/locales/zh/admin/ops.ts @@ -4,6 +4,8 @@ export default { description: '运维监控与排障', // Dashboard systemHealth: '系统健康', + outputTps: '单次输出 TPS', + outputTpsSamples: '有效样本：{count}', overview: '概览', noSystemMetrics: '尚未收集系统指标。', collectedAt: '采集时间：', @@ -804,6 +806,7 @@ export default { accountError: '异常' }, tooltips: { + outputTps: '每条有效用量记录的输出 Token ÷ 总耗时（含首字等待），再按当前时间、平台和分组计算分位数。输出可含推理 Token，不重复相加。P50 为中位数，P5/P10 反映较慢请求；数值越高越快。排除图片、Live、无有效输出或耗时的记录。样本来自仍保留的使用明细；无样本或统计暂不可用时显示 —。', totalRequests: '当前时间窗口内的总请求数和Token消耗量。', throughputTrend: '当前窗口内的请求/QPS 与 token/TPS 趋势。', switchRateTrend: '近5小时内账号切换次数 / 请求总数的趋势（平均切换次数）。', diff --git a/frontend/src/views/admin/ops/components/OpsDashboardHeader.vue b/frontend/src/views/admin/ops/components/OpsDashboardHeader.vue index 41040ed773..35bb9e7a30 100644 --- a/frontend/src/views/admin/ops/components/OpsDashboardHeader.vue +++ b/frontend/src/views/admin/ops/components/OpsDashboardHeader.vue @@ -1428,6 +1428,28 @@ function handleToolbarRefresh() { + + +
+
+ {{ t('admin.ops.outputTps') }} + +
+
+
{{ overview.output_tps?.p50?.toFixed(1) ?? '—' }}
+ tok/s (P50) +
+
+
+ {{ metric === 'avg' ? 'Avg' : metric.toUpperCase() }}: + {{ overview.output_tps?.[metric]?.toFixed(1) ?? '—' }} + tok/s +
+
+
+ {{ t('admin.ops.outputTpsSamples', { count: overview.output_tps == null ? '—' : formatNumber(overview.output_tps.sample_count) }) }} +
+
diff --git a/frontend/src/views/admin/ops/components/OpsDashboardSkeleton.vue b/frontend/src/views/admin/ops/components/OpsDashboardSkeleton.vue index 9b520e7572..030536096b 100644 --- a/frontend/src/views/admin/ops/components/OpsDashboardSkeleton.vue +++ b/frontend/src/views/admin/ops/components/OpsDashboardSkeleton.vue @@ -43,7 +43,7 @@ const props = withDefaults(defineProps(), {
-
+
diff --git a/frontend/src/views/admin/ops/components/__tests__/OpsDashboardHeader.spec.ts b/frontend/src/views/admin/ops/components/__tests__/OpsDashboardHeader.spec.ts new file mode 100644 index 0000000000..6388c35f7c --- /dev/null +++ b/frontend/src/views/admin/ops/components/__tests__/OpsDashboardHeader.spec.ts @@ -0,0 +1,51 @@ +import { describe, expect, it, vi } from 'vitest' +import { flushPromises, mount } from '@vue/test-utils' +import OpsDashboardHeader from '../OpsDashboardHeader.vue' +import type { OpsDashboardOverview } from '@/api/admin/ops' + +vi.mock('@/api', () => ({ adminAPI: { groups: { getAll: vi.fn().mockResolvedValue([]) } } })) +vi.mock('@/api/admin/ops', () => ({ opsAPI: { getRealtimeTrafficSummary: vi.fn().mockResolvedValue({ enabled: true, summary: null }) } })) +vi.mock('@/stores', () => ({ useAdminSettingsStore: () => ({ opsRealtimeMonitoringEnabled: true, setOpsRealtimeMonitoringEnabledLocal: vi.fn() }) })) +vi.mock('vue-i18n', async () => ({ + ...await vi.importActual('vue-i18n'), + useI18n: () => ({ t: (key: string, params?: Record) => params ? `${key}:${JSON.stringify(params)}` : key }) +})) + +const overview = { tps: { avg: 242.7 } } as OpsDashboardOverview +function mountHeader(value = overview) { + return mount(OpsDashboardHeader, { + props: { overview: value, platform: '', groupId: null, timeRange: '1h', queryMode: 'auto', loading: false, lastUpdated: null }, + global: { stubs: { Select: true, HelpTooltip: true, BaseDialog: true, Icon: true } } + }) +} + +describe('OpsDashboardHeader output TPS', () => { + + it('shows per-request P50, slower-tail percentiles and sample count separately from traffic TPS', async () => { + const wrapper = mountHeader({ ...overview, output_tps: { p5: 2.35, p10: 3.7, p50: 55, avg: 277.75, sample_count: 4 } }) + await flushPromises() + const card = wrapper.get('[data-testid="output-tps-card"]') + expect(card.get('[data-testid="output-tps-p50"]').text()).toBe('55.0') + expect(card.get('[data-testid="output-tps-p5"]').text()).toBe('2.4') + expect(card.get('[data-testid="output-tps-p10"]').text()).toBe('3.7') + expect(card.get('[data-testid="output-tps-avg"]').text()).toBe('277.8') + expect(card.get('[data-testid="output-tps-samples"]').text()).toContain('4') + expect(card.text()).toContain('tok/s (P50)') + expect(card.text()).not.toContain('242.7') + await wrapper.setProps({ fullscreen: true, overview: { ...overview, output_tps: { p5: 0.01, p10: 0.02, p50: 0.04, avg: 0.05, sample_count: 1 } } }) + expect(card.get('[data-testid="output-tps-p50"]').text()).toBe('0.0') + expect(card.get('[data-testid="output-tps-samples"]').text()).toContain('1') + wrapper.unmount() + }) + + it.each([undefined, null, { p5: null, p10: null, p50: null, avg: null, sample_count: 0 }])('shows missing output TPS as unavailable, not zero: %s', async (output_tps) => { + const wrapper = mountHeader({ ...overview, output_tps }) + await flushPromises() + for (const metric of ['p5', 'p10', 'p50', 'avg']) { + expect(wrapper.get(`[data-testid="output-tps-${metric}"]`).text()).toBe('—') + } + expect(wrapper.get('[data-testid="output-tps-samples"]').text()).toContain(output_tps == null ? '—' : '0') + wrapper.unmount() + }) + +})