diff --git a/backend/internal/repository/ops_repo_dashboard.go b/backend/internal/repository/ops_repo_dashboard.go index b43d6706f3..a22235a805 100644 --- a/backend/internal/repository/ops_repo_dashboard.go +++ b/backend/internal/repository/ops_repo_dashboard.go @@ -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) ( diff --git a/backend/internal/repository/ops_repo_preagg.go b/backend/internal/repository/ops_repo_preagg.go index ad94e13f7e..40b99fe3a4 100644 --- a/backend/internal/repository/ops_repo_preagg.go +++ b/backend/internal/repository/ops_repo_preagg.go @@ -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, diff --git a/backend/migrations/145_ops_metrics_ttft_sample_count.sql b/backend/migrations/145_ops_metrics_ttft_sample_count.sql new file mode 100644 index 0000000000..11859f7397 --- /dev/null +++ b/backend/migrations/145_ops_metrics_ttft_sample_count.sql @@ -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;