mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-16 07:33:06 +08:00
feat(ops): show per-request output TPS percentiles
This commit is contained in:
lyy0709
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"`
|
||||
|
||||
Reference in New Issue
Block a user