opensource/sub2api
3
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-09-21 14:19:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(ops): weight TTFT percentiles by streaming sample count

Browse Source 
TTFT (first_token_ms) is only recorded for streaming requests, but the
ops dashboard weighted merged TTFT percentiles by success_count (all
successful requests, streaming + non-streaming). When non-streaming
traffic was present this diluted/skewed the merged TTFT figures shown
for longer (pre-aggregated) time ranges; the realtime path was exact.

Add a per-bucket ttft_sample_count (rows that actually recorded
first_token_ms) to ops_metrics_hourly / ops_metrics_daily and weight all
TTFT percentile merges by it instead of success_count:

- hourly/daily pre-agg upserts populate and propagate ttft_sample_count;
  daily TTFT p50/p90/avg now weighted by ttft_sample_count.
- dashboard hourly-row merge and cross-segment combine weight TTFT by
  the streaming sample count; queryUsageLatency returns it for raw
  head/tail fragments.

duration stays weighted by success_count (recorded for every request);
p95/p99/max keep the conservative MAX merge (weight-independent).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
bwlc
2026-06-05 10:58:44 +08:00
co-authored by Claude Opus 4.8
parent aa69e3947d
commit 69b4654510
3 changed files with 71 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/repository/ops_repo_dashboard.go
+34 -17
View File
@@ -58,7 +58,7 @@ func (r *opsRepository) getDashboardOverviewRaw(ctx context.Context, filter *ser
}
latencyCtx, cancelLatency := context.WithTimeout(ctx, opsRawLatencyQueryTimeout)
duration, ttft, err := r.queryUsageLatency(latencyCtx, filter, start, end)
duration, ttft, _, err := r.queryUsageLatency(latencyCtx, filter, start, end)
cancelLatency()
if err != nil {
if isQueryTimeoutErr(err) {
@@ -163,6 +163,7 @@ func (r *opsRepository) getDashboardOverviewRaw(ctx context.Context, filter *ser
type opsDashboardPartial struct {
successCount int64
ttftSampleCount int64
errorCountTotal int64
businessLimitedCount int64
errorCountSLA int64
@@ -247,10 +248,12 @@ func (r *opsRepository) getDashboardOverviewPreaggregated(ctx context.Context, f
{weight: head.successCount, p: head.duration},
{weight: tail.successCount, p: tail.duration},
})
// TTFT segments are weighted by the streaming sample count (rows that
// actually recorded first_token_ms), not the total success count.
ttft := combineApproxPercentiles([]opsPercentileSegment{
{weight: preagg.successCount, p: preagg.ttft},
{weight: head.successCount, p: head.ttft},
{weight: tail.successCount, p: tail.ttft},
{weight: preagg.ttftSampleCount, p: preagg.ttft},
{weight: head.ttftSampleCount, p: head.ttft},
{weight: tail.ttftSampleCount, p: tail.ttft},
})
windowSeconds := end.Sub(start).Seconds()
@@ -345,6 +348,7 @@ type opsHourlyMetricsRow struct {
bucketStart time.Time
successCount int64
ttftSampleCount int64
errorCountTotal int64
businessLimitedCount int64
errorCountSLA int64
@@ -429,7 +433,8 @@ SELECT
ttft_p95_ms,
ttft_p99_ms,
ttft_avg_ms,
ttft_max_ms
ttft_max_ms,
ttft_sample_count
FROM ops_metrics_hourly
WHERE ` + where + `
ORDER BY bucket_start ASC`
@@ -465,6 +470,7 @@ ORDER BY bucket_start ASC`
&row.ttftP99,
&row.ttftAvg,
&row.ttftMax,
&row.ttftSampleCount,
); err != nil {
return nil, err
}
@@ -511,6 +517,7 @@ func aggregateHourlyRows(rows []opsHourlyMetricsRow) opsDashboardPartial {
for _, row := range rows {
out.successCount += row.successCount
out.ttftSampleCount += row.ttftSampleCount
out.errorCountTotal += row.errorCountTotal
out.businessLimitedCount += row.businessLimitedCount
out.errorCountSLA += row.errorCountSLA
@@ -534,17 +541,23 @@ func aggregateHourlyRows(rows []opsHourlyMetricsRow) opsDashboardPartial {
avgSum += row.durationAvg.Float64 * float64(row.successCount)
avgW += row.successCount
}
}
// TTFT is weighted by ttftSampleCount (streaming rows only), NOT
// successCount: first_token_ms is recorded only for streaming requests,
// so weighting by total successes dilutes the merged TTFT figures.
if row.ttftSampleCount > 0 {
if row.ttftP50.Valid {
ttftP50Sum += float64(row.ttftP50.Int64) * float64(row.successCount)
ttftP50W += row.successCount
ttftP50Sum += float64(row.ttftP50.Int64) * float64(row.ttftSampleCount)
ttftP50W += row.ttftSampleCount
}
if row.ttftP90.Valid {
ttftP90Sum += float64(row.ttftP90.Int64) * float64(row.successCount)
ttftP90W += row.successCount
ttftP90Sum += float64(row.ttftP90.Int64) * float64(row.ttftSampleCount)
ttftP90W += row.ttftSampleCount
}
if row.ttftAvg.Valid {
ttftAvgSum += row.ttftAvg.Float64 * float64(row.successCount)
ttftAvgW += row.successCount
ttftAvgSum += row.ttftAvg.Float64 * float64(row.ttftSampleCount)
ttftAvgW += row.ttftSampleCount
}
}
@@ -631,12 +644,13 @@ func (r *opsRepository) queryRawPartial(ctx context.Context, filter *service.Ops
}
latencyCtx, cancelLatency := context.WithTimeout(ctx, opsRawLatencyQueryTimeout)
duration, ttft, err := r.queryUsageLatency(latencyCtx, filter, start, end)
duration, ttft, ttftSampleCount, err := r.queryUsageLatency(latencyCtx, filter, start, end)
cancelLatency()
if err != nil {
if isQueryTimeoutErr(err) {
duration = service.OpsPercentiles{}
ttft = service.OpsPercentiles{}
ttftSampleCount = 0
} else {
return nil, err
}
@@ -649,6 +663,7 @@ func (r *opsRepository) queryRawPartial(ctx context.Context, filter *service.Ops
return &opsDashboardPartial{
successCount: successCount,
ttftSampleCount: ttftSampleCount,
errorCountTotal: errorTotal,
businessLimitedCount: businessLimited,
errorCountSLA: errorCountSLA,
@@ -794,7 +809,7 @@ FROM usage_logs ul
return successCount, tokenConsumed, nil
}
func (r *opsRepository) queryUsageLatency(ctx context.Context, filter *service.OpsDashboardFilter, start, end time.Time) (duration service.OpsPercentiles, ttft service.OpsPercentiles, err error) {
func (r *opsRepository) queryUsageLatency(ctx context.Context, filter *service.OpsDashboardFilter, start, end time.Time) (duration service.OpsPercentiles, ttft service.OpsPercentiles, ttftSampleCount int64, err error) {
join, where, args, _ := buildUsageWhere(filter, start, end, 1)
q := `
SELECT
@@ -809,7 +824,8 @@ SELECT
percentile_cont(0.95) WITHIN GROUP (ORDER BY first_token_ms) FILTER (WHERE first_token_ms IS NOT NULL) AS ttft_p95,
percentile_cont(0.99) WITHIN GROUP (ORDER BY first_token_ms) FILTER (WHERE first_token_ms IS NOT NULL) AS ttft_p99,
AVG(first_token_ms) FILTER (WHERE first_token_ms IS NOT NULL) AS ttft_avg,
MAX(first_token_ms) AS ttft_max
MAX(first_token_ms) AS ttft_max,
COUNT(first_token_ms) AS ttft_sample_count
FROM usage_logs ul
` + join + `
` + where
@@ -820,11 +836,12 @@ FROM usage_logs ul
var tP50, tP90, tP95, tP99 sql.NullFloat64
var tAvg sql.NullFloat64
var tMax sql.NullInt64
var tCount int64
if err := r.db.QueryRowContext(ctx, q, args...).Scan(
&dP50, &dP90, &dP95, &dP99, &dAvg, &dMax,
&tP50, &tP90, &tP95, &tP99, &tAvg, &tMax,
&tP50, &tP90, &tP95, &tP99, &tAvg, &tMax, &tCount,
); err != nil {
return service.OpsPercentiles{}, service.OpsPercentiles{}, err
return service.OpsPercentiles{}, service.OpsPercentiles{}, 0, err
}
duration.P50 = floatToIntPtr(dP50)
@@ -847,7 +864,7 @@ FROM usage_logs ul
ttft.Max = &v
}
return duration, ttft, nil
return duration, ttft, tCount, nil
}
func (r *opsRepository) queryErrorCounts(ctx context.Context, filter *service.OpsDashboardFilter, start, end time.Time) (
backend/internal/repository/ops_repo_preagg.go
+16 -6
View File
@@ -46,6 +46,7 @@ usage_agg AS (
CASE WHEN GROUPING(platform) = 1 THEN NULL ELSE platform END AS platform,
CASE WHEN GROUPING(group_id) = 1 THEN NULL ELSE group_id END AS group_id,
COUNT(*) AS success_count,
COUNT(*) FILTER (WHERE first_token_ms IS NOT NULL) AS ttft_sample_count,
COALESCE(SUM(tokens), 0) AS token_consumed,
percentile_cont(0.50) WITHIN GROUP (ORDER BY duration_ms) FILTER (WHERE duration_ms IS NOT NULL) AS duration_p50_ms,
@@ -110,6 +111,7 @@ combined AS (
COALESCE(u.group_id, e.group_id) AS group_id,
COALESCE(u.success_count, 0) AS success_count,
COALESCE(u.ttft_sample_count, 0) AS ttft_sample_count,
COALESCE(e.error_count_total, 0) AS error_count_total,
COALESCE(e.business_limited_count, 0) AS business_limited_count,
COALESCE(e.error_count_sla, 0) AS error_count_sla,
@@ -143,6 +145,7 @@ INSERT INTO ops_metrics_hourly (
platform,
group_id,
success_count,
ttft_sample_count,
error_count_total,
business_limited_count,
error_count_sla,
@@ -169,6 +172,7 @@ SELECT
NULLIF(platform, '') AS platform,
group_id,
success_count,
ttft_sample_count,
error_count_total,
business_limited_count,
error_count_sla,
@@ -194,6 +198,7 @@ WHERE bucket_start IS NOT NULL
AND (platform IS NULL OR platform <> '')
ON CONFLICT (bucket_start, COALESCE(platform, ''), COALESCE(group_id, 0)) DO UPDATE SET
success_count = EXCLUDED.success_count,
ttft_sample_count = EXCLUDED.ttft_sample_count,
error_count_total = EXCLUDED.error_count_total,
business_limited_count = EXCLUDED.business_limited_count,
error_count_sla = EXCLUDED.error_count_sla,
@@ -240,6 +245,7 @@ INSERT INTO ops_metrics_daily (
platform,
group_id,
success_count,
ttft_sample_count,
error_count_total,
business_limited_count,
error_count_sla,
@@ -267,6 +273,7 @@ SELECT
group_id,
COALESCE(SUM(success_count), 0) AS success_count,
COALESCE(SUM(ttft_sample_count), 0) AS ttft_sample_count,
COALESCE(SUM(error_count_total), 0) AS error_count_total,
COALESCE(SUM(business_limited_count), 0) AS business_limited_count,
COALESCE(SUM(error_count_sla), 0) AS error_count_sla,
@@ -286,14 +293,16 @@ SELECT
/ NULLIF(SUM(success_count) FILTER (WHERE duration_avg_ms IS NOT NULL), 0) AS duration_avg_ms,
MAX(duration_max_ms) AS duration_max_ms,
ROUND(SUM(ttft_p50_ms::double precision * success_count) FILTER (WHERE ttft_p50_ms IS NOT NULL)
/ NULLIF(SUM(success_count) FILTER (WHERE ttft_p50_ms IS NOT NULL), 0))::int AS ttft_p50_ms,
ROUND(SUM(ttft_p90_ms::double precision * success_count) FILTER (WHERE ttft_p90_ms IS NOT NULL)
/ NULLIF(SUM(success_count) FILTER (WHERE ttft_p90_ms IS NOT NULL), 0))::int AS ttft_p90_ms,
-- TTFT is weighted by ttft_sample_count (streaming rows only), NOT success_count,
-- because first_token_ms is recorded only for streaming requests.
ROUND(SUM(ttft_p50_ms::double precision * ttft_sample_count) FILTER (WHERE ttft_p50_ms IS NOT NULL)
/ NULLIF(SUM(ttft_sample_count) FILTER (WHERE ttft_p50_ms IS NOT NULL), 0))::int AS ttft_p50_ms,
ROUND(SUM(ttft_p90_ms::double precision * ttft_sample_count) FILTER (WHERE ttft_p90_ms IS NOT NULL)
/ NULLIF(SUM(ttft_sample_count) FILTER (WHERE ttft_p90_ms IS NOT NULL), 0))::int AS ttft_p90_ms,
MAX(ttft_p95_ms) AS ttft_p95_ms,
MAX(ttft_p99_ms) AS ttft_p99_ms,
SUM(ttft_avg_ms * success_count) FILTER (WHERE ttft_avg_ms IS NOT NULL)
/ NULLIF(SUM(success_count) FILTER (WHERE ttft_avg_ms IS NOT NULL), 0) AS ttft_avg_ms,
SUM(ttft_avg_ms * ttft_sample_count) FILTER (WHERE ttft_avg_ms IS NOT NULL)
/ NULLIF(SUM(ttft_sample_count) FILTER (WHERE ttft_avg_ms IS NOT NULL), 0) AS ttft_avg_ms,
MAX(ttft_max_ms) AS ttft_max_ms,
NOW()
@@ -302,6 +311,7 @@ WHERE bucket_start >= $1 AND bucket_start < $2
GROUP BY 1, 2, 3
ON CONFLICT (bucket_date, COALESCE(platform, ''), COALESCE(group_id, 0)) DO UPDATE SET
success_count = EXCLUDED.success_count,
ttft_sample_count = EXCLUDED.ttft_sample_count,
error_count_total = EXCLUDED.error_count_total,
business_limited_count = EXCLUDED.business_limited_count,
error_count_sla = EXCLUDED.error_count_sla,
backend/migrations/145_ops_metrics_ttft_sample_count.sql
+21
View File
@@ -0,0 +1,21 @@
-- Add ttft_sample_count to ops_metrics_hourly / ops_metrics_daily.
--
-- first_token_ms (TTFT) is only recorded for streaming requests, but the
-- dashboard previously weighted merged TTFT percentiles by success_count
-- (all successful requests, streaming + non-streaming). That diluted/skewed
-- the merged TTFT figures whenever non-streaming traffic was present.
--
-- Store the real streaming sample count per bucket so TTFT percentiles can be
-- weighted by the number of rows that actually contributed a first_token_ms.
--
-- Existing rows default to 0; they will be repopulated on the next hourly
-- re-aggregation pass. Migration is idempotent and safe to re-run.
SET LOCAL lock_timeout = '5s';
SET LOCAL statement_timeout = '10min';
ALTER TABLE ops_metrics_hourly
ADD COLUMN IF NOT EXISTS ttft_sample_count BIGINT NOT NULL DEFAULT 0;
ALTER TABLE ops_metrics_daily
ADD COLUMN IF NOT EXISTS ttft_sample_count BIGINT NOT NULL DEFAULT 0;