diff --git a/pkg/perf_metrics/metrics.go b/pkg/perf_metrics/metrics.go index cf6af6d7d..7dbb3f813 100644 --- a/pkg/perf_metrics/metrics.go +++ b/pkg/perf_metrics/metrics.go @@ -199,11 +199,7 @@ func QuerySummaryAll(hours int, groups []string, recentModels ...string) (Summar startTs, endTs := queryWindow(now, hours) recentStartTs, _ := queryWindow(now, recentSuccessHistoryHours) allowedGroups := allowedGroupSet(groups) - - rows, err := model.GetPerfMetricsSummaryBucketsAll(startTs, endTs, groups) - if err != nil { - return SummaryAllResult{}, err - } + queryStartTs := startTs recentBuckets := make(map[string]map[int64]counters, len(recentModels)) for _, name := range recentModels { @@ -219,6 +215,7 @@ func QuerySummaryAll(hours int, groups []string, recentModels ...string) (Summar return SummaryAllResult{}, err } for _, bucket := range buckets { + queryStartTs = min(queryStartTs, bucket.BucketTs) mergeModelBucket(recentBuckets, name, bucket.BucketTs, counters{ requestCount: bucket.RequestCount, successCount: bucket.SuccessCount, @@ -226,7 +223,11 @@ func QuerySummaryAll(hours int, groups []string, recentModels ...string) (Summar } } - totals := map[string]counters{} + rows, err := model.GetPerfMetricsSummaryBucketsAll(queryStartTs, endTs, groups) + if err != nil { + return SummaryAllResult{}, err + } + modelBuckets := map[string]map[int64]counters{} for _, row := range rows { value := counters{ @@ -236,7 +237,6 @@ func QuerySummaryAll(hours int, groups []string, recentModels ...string) (Summar outputTokens: row.OutputTokens, generationMs: row.GenerationMs, } - mergeModelTotals(totals, row.ModelName, value) mergeModelBucket(modelBuckets, row.ModelName, row.BucketTs, value) } @@ -257,14 +257,38 @@ func QuerySummaryAll(hours int, groups []string, recentModels ...string) (Summar if _, requested := recentBuckets[k.model]; requested && k.bucketTs >= recentStartTs { mergeModelBucket(recentBuckets, k.model, k.bucketTs, snap) } - if k.bucketTs < startTs { + if k.bucketTs < queryStartTs { return true } - mergeModelTotals(totals, k.model, snap) mergeModelBucket(modelBuckets, k.model, k.bucketTs, snap) return true }) + recentSeries := make(map[string][]SuccessRatePoint, len(recentBuckets)) + for name, buckets := range recentBuckets { + points := recentSuccessSeries(buckets) + if len(points) == 0 { + continue + } + recentSeries[name] = points[max(0, len(points)-recentSuccessSampleLimit):] + } + + // Requested models are summarized over the same hours as their status samples. + totals := map[string]counters{} + for name, buckets := range modelBuckets { + fromTs := startTs + if points, ok := recentSeries[name]; ok { + fromTs = points[0].Ts + } + for bucketTs, value := range buckets { + if bucketTs < fromTs { + delete(buckets, bucketTs) + continue + } + mergeModelTotals(totals, name, value) + } + } + all := counters{} models := make([]ModelSummary, 0, len(totals)) for name, total := range totals { @@ -295,14 +319,6 @@ func QuerySummaryAll(hours int, groups []string, recentModels ...string) (Summar return models[i].RequestCount > models[j].RequestCount }) - recentSeries := make(map[string][]SuccessRatePoint, len(recentBuckets)) - for name, buckets := range recentBuckets { - points := recentSuccessSeries(buckets) - if len(points) == 0 { - continue - } - recentSeries[name] = points[max(0, len(points)-recentSuccessSampleLimit):] - } return SummaryAllResult{ Summary: summarize(all), WindowStart: startTs, WindowEnd: endTs, Models: models, RecentSuccessSeries: recentSeries, diff --git a/pkg/perf_metrics/metrics_test.go b/pkg/perf_metrics/metrics_test.go index b4ac25a4e..8ed2a2c55 100644 --- a/pkg/perf_metrics/metrics_test.go +++ b/pkg/perf_metrics/metrics_test.go @@ -12,6 +12,7 @@ import ( "github.com/QuantumNous/new-api/model" relaycommon "github.com/QuantumNous/new-api/relay/common" "github.com/QuantumNous/new-api/relaykit/types" + "github.com/samber/lo" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -262,7 +263,7 @@ func TestPerformanceAggregationAndFlush(t *testing.T) { assert.Equal(t, 98.04, combined.Summary.SuccessRate) assert.Equal(t, 99.01, combined.Models[0].SuccessRate) - t.Run("recent history does not change window summaries", func(t *testing.T) { + t.Run("recent models are summarized over their status sample hours", func(t *testing.T) { for i := range 26 { require.NoError(t, model.UpsertPerfMetric(&model.PerfMetric{ ModelName: "history-model", Group: "a", BucketTs: hour - int64(i+1)*6*3600, @@ -293,9 +294,13 @@ func TestPerformanceAggregationAndFlush(t *testing.T) { require.NoError(t, err) history, err := QuerySummaryAll(24, groups, "history-model", "history-idle", "history-hidden", "history-hot", "missing") require.NoError(t, err) - assert.Equal(t, window.Summary, history.Summary) - assert.ElementsMatch(t, window.Models, history.Models) assert.Equal(t, window.WindowStart, history.WindowStart) + summaries := lo.KeyBy(history.Models, func(m ModelSummary) string { return m.ModelName }) + assert.Equal(t, int64(124), summaries["history-model"].RequestCount) + assert.Equal(t, 99.19, summaries["history-model"].SuccessRate) + assert.Equal(t, int64(1), summaries["history-idle"].RequestCount) + assert.Equal(t, 0.0, summaries["history-idle"].SuccessRate) + assert.NotContains(t, summaries, "history-hidden") points := history.RecentSuccessSeries["history-model"] require.Len(t, points, 24) expected := make([]SuccessRatePoint, 0, 24)