From 33fb54bfe67d2990539f2f240ed0bb59495be259 Mon Sep 17 00:00:00 2001 From: tbphp Date: Mon, 7 Sep 2026 20:20:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(monitor):=20=E4=BF=AE=E6=AD=A3=E6=9C=80?= =?UTF-8?q?=E8=BF=91=E4=B8=80=E5=B0=8F=E6=97=B6=E7=94=A8=E9=87=8F=E4=B8=8E?= =?UTF-8?q?=E6=88=90=E6=9C=AC=E7=BB=9F=E8=AE=A1=20(#588)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/control/usage.go | 43 ++- internal/control/usage_rolling_test.go | 118 +++++++ internal/control/usage_test.go | 63 +++- .../external_database_integration_test.go | 30 ++ internal/requestlog/types.go | 6 +- internal/requestlog/usage_query.go | 35 +- internal/requestlog/usage_query_minute.go | 79 +++++ .../requestlog/usage_query_minute_test.go | 318 ++++++++++++++++++ web/src/app/resources/usage.ts | 33 +- web/src/components/charts/TrendChart.vue | 3 + web/src/components/charts/trend-chart.ts | 9 +- web/src/features/monitor/UsageTab.vue | 51 ++- web/src/features/monitor/usage-bar-chart.ts | 4 +- web/src/i18n/locales/en-US/monitor.ts | 3 +- web/src/i18n/locales/ja-JP/monitor.ts | 3 +- web/src/i18n/locales/zh-CN/monitor.ts | 3 +- 16 files changed, 755 insertions(+), 46 deletions(-) create mode 100644 internal/control/usage_rolling_test.go create mode 100644 internal/requestlog/usage_query_minute.go create mode 100644 internal/requestlog/usage_query_minute_test.go diff --git a/internal/control/usage.go b/internal/control/usage.go index c8a62a6c..790cf07d 100644 --- a/internal/control/usage.go +++ b/internal/control/usage.go @@ -225,16 +225,24 @@ func parseUsageQuery(rawQuery string, observedAtMS int64) (requestlog.UsageQuery if !ok { return requestlog.UsageQuery{}, app_errors.ErrValidation } - fromMS, toMS, err := epochms.WindowEndingAt( - observedAtMS, - preset.bucketWidthMS, - preset.bucketCount, - ) - if err != nil { - return requestlog.UsageQuery{}, app_errors.ErrInternalServer + if rangeValue == usageRange1Hour { + if observedAtMS < epochms.MillisecondsPerHour { + return requestlog.UsageQuery{}, app_errors.ErrInternalServer + } + query.FromMS = observedAtMS - epochms.MillisecondsPerHour + query.ToMS = observedAtMS + } else { + fromMS, toMS, err := epochms.WindowEndingAt( + observedAtMS, + preset.bucketWidthMS, + preset.bucketCount, + ) + if err != nil { + return requestlog.UsageQuery{}, app_errors.ErrInternalServer + } + query.FromMS = fromMS + query.ToMS = toMS } - query.FromMS = fromMS - query.ToMS = toMS query.Granularity = preset.granularity query.BucketWidthMS = preset.bucketWidthMS } @@ -277,7 +285,7 @@ type usagePreset struct { func usageRangePreset(value string) (usagePreset, bool) { switch value { case usageRange1Hour: - return usagePreset{epochms.MillisecondsPerHour, 1, requestlog.UsageGranularityHour}, true + return usagePreset{requestlog.UsageFiveMinuteBucketMS, 12, requestlog.UsageGranularityMinute}, true case usageRange24Hours: return usagePreset{epochms.MillisecondsPerHour, 24, requestlog.UsageGranularityHour}, true case usageRange3Days: @@ -592,6 +600,13 @@ func validUsageDistributionModel(value string) bool { func usageResponseBucketWidth(query requestlog.UsageQuery) (int64, error) { bucketWidthMS := query.BucketWidthMS + if query.Granularity == requestlog.UsageGranularityMinute { + if bucketWidthMS != requestlog.UsageFiveMinuteBucketMS || + query.ToMS-query.FromMS != epochms.MillisecondsPerHour { + return 0, fmt.Errorf("map usage response: invalid minute window") + } + return bucketWidthMS, nil + } if bucketWidthMS == 0 { switch query.Granularity { case requestlog.UsageGranularityHour: @@ -620,10 +635,16 @@ func usageResponseBucketWidth(query requestlog.UsageQuery) (int64, error) { func usageResponseRange(query requestlog.UsageQuery, bucketWidthMS int64) (string, error) { duration := query.ToMS - query.FromMS - if query.Granularity != requestlog.UsageGranularityHour && + if query.Granularity != requestlog.UsageGranularityMinute && + query.Granularity != requestlog.UsageGranularityHour && query.Granularity != requestlog.UsageGranularityDay { return "", fmt.Errorf("map usage response: invalid granularity") } + // 显式指定一个完整小时的自定义查询继续返回原有范围标签。 + if query.Granularity == requestlog.UsageGranularityHour && + bucketWidthMS == epochms.MillisecondsPerHour && duration == epochms.MillisecondsPerHour { + return usageRange1Hour, nil + } for _, rangeValue := range []string{ usageRange1Hour, usageRange24Hours, diff --git a/internal/control/usage_rolling_test.go b/internal/control/usage_rolling_test.go new file mode 100644 index 00000000..558ac962 --- /dev/null +++ b/internal/control/usage_rolling_test.go @@ -0,0 +1,118 @@ +package control + +import ( + "encoding/json" + "fmt" + "net/http" + "testing" + "time" + + "github.com/gin-gonic/gin" + + "gpt-load/internal/platform/config" + "gpt-load/internal/requestlog" + "gpt-load/internal/storage/models" +) + +func TestUsageAPIRollingHourReadsLogsAndScopesAccessKey(t *testing.T) { + t.Parallel() + initControlI18n(t) + fixture := newServiceFixture(t) + key, err := fixture.service.CreateAccessKey(t.Context(), AccessKeyCreateRequest{Name: "rolling usage viewer"}) + if err != nil { + t.Fatalf("create usage access key: %v", err) + } + now := time.Date(2026, time.September, 7, 14, 5, 23, 123_000_000, time.UTC) + from := now.Add(-time.Hour) + for index, item := range []struct { + completedAt time.Time + accessKeyID uint + attempts int + }{ + {from.Add(-time.Millisecond), key.ID, 1}, + {from, key.ID, 1}, + {now.Add(-time.Millisecond), key.ID, 1}, + {now, key.ID, 1}, + {now.Add(-time.Minute), key.ID, 0}, + {now.Add(-time.Second), key.ID + 1, 1}, + } { + row := models.RequestLog{ + ID: fmt.Sprintf("00000000-0000-4000-8000-%012d", index+1), + CompletedAtMS: item.completedAt.UnixMilli(), AccessKeyID: item.accessKeyID, + GroupID: 7, ChannelID: "openai", CredentialID: 11, + Protocol: "openai-completions", ClientModel: "usage-model", UpstreamModel: "usage-model", + Status: "success", StatusCode: 200, AttemptCount: item.attempts, + UsageState: "complete", CostState: "priced", PricingCompleteness: "complete", + UncachedInputTokens: 3, OutputTokens: 5, EstimatedCostNanoUSD: 700, + } + if err := fixture.db.Create(&row).Error; err != nil { + t.Fatalf("create usage request log: %v", err) + } + } + if err := fixture.db.Create(&models.UsageStat{ + BucketStartMS: from.Truncate(time.Hour).UnixMilli(), AccessKeyID: key.ID, + GroupID: 7, ChannelID: "openai", CredentialID: 11, Model: "usage-model", + RequestCount: 99, SuccessCount: 99, + }).Error; err != nil { + t.Fatalf("create independent hourly usage stat: %v", err) + } + fixture.service.now = func() time.Time { return now } + fixture.service.usageStats = requestlog.NewService(fixture.db, nil, nil) + engine := gin.New() + NewServer(&config.Config{AuthKey: "test-auth-key"}, fixture.service).RegisterRoutes(engine) + for _, test := range []struct { + name string + auth string + query string + wantCount int64 + wantCost string + wantRange string + wantGrain requestlog.UsageGranularity + }{ + {"management rolling hour", "test-auth-key", "range=1h", 3, "2100", "1h", requestlog.UsageGranularityMinute}, + {"access key rolling hour", key.Key, "range=1h", 2, "1400", "1h", requestlog.UsageGranularityMinute}, + {"daily range retains hourly source", "test-auth-key", "range=24h", 99, "0", "24h", requestlog.UsageGranularityHour}, + {"custom hour retains hourly source and metadata", "test-auth-key", + fmt.Sprintf("from_ms=%d&to_ms=%d", from.Truncate(time.Hour).UnixMilli(), from.Truncate(time.Hour).Add(time.Hour).UnixMilli()), + 99, "0", "1h", requestlog.UsageGranularityHour}, + } { + t.Run(test.name, func(t *testing.T) { + recorder := performUsageRequest(engine, test.auth, test.query) + if recorder.Code != http.StatusOK { + t.Fatalf("usage response = %d %s", recorder.Code, recorder.Body.String()) + } + var envelope struct { + Data usageResponse `json:"data"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode usage response: %v", err) + } + data := envelope.Data + if data.Range != test.wantRange || data.Granularity != test.wantGrain { + t.Fatalf("usage range/granularity = %s/%s, want %s/%s", data.Range, data.Granularity, test.wantRange, test.wantGrain) + } + if data.Summary.RequestCount != test.wantCount || data.Summary.EstimatedCostNanoUSD != test.wantCost { + t.Fatalf("usage summary = %+v", data.Summary) + } + if test.query == "range=1h" { + if len(data.Series) != 2 || data.Series[0].BucketStartMS != from.UnixMilli() || + data.Series[1].BucketEndMS != now.UnixMilli() || data.Granularity != "minute" { + t.Fatalf("rolling usage series = %+v", data.Series) + } + if data.Series[0].RequestCount+data.Series[1].RequestCount != test.wantCount { + t.Fatalf("series differs from summary: %+v", data.Series) + } + for _, distribution := range data.Distributions.Model { + if len(distribution.Items) != 1 || distribution.Items[0].RequestCount != test.wantCount || + distribution.Items[0].EstimatedCostNanoUSD != test.wantCost { + t.Fatalf("model distribution differs from summary: %+v", distribution) + } + } + } + if test.auth == key.Key && (data.Distributions.Group != nil || data.Distributions.AccessKey != nil || + data.CollectionHealth.Scope != "access_key") { + t.Fatalf("access key response exposes management scope: %+v", data) + } + }) + } +} diff --git a/internal/control/usage_test.go b/internal/control/usage_test.go index 3f8628c1..fe6cfbae 100644 --- a/internal/control/usage_test.go +++ b/internal/control/usage_test.go @@ -40,7 +40,7 @@ func TestUsageAPIRouteUsesManagementAuthentication(t *testing.T) { } } -func TestParseUsageQueryUsesFixedUTCAlignedWindows(t *testing.T) { +func TestParseUsageQueryUsesRangeSpecificWindows(t *testing.T) { t.Parallel() tests := []struct { name string @@ -52,11 +52,29 @@ func TestParseUsageQueryUsesFixedUTCAlignedWindows(t *testing.T) { bucketWidth int64 }{ { - name: "1 hour uses one complete UTC hour", + name: "1 hour uses the last sixty minutes across the hour boundary", rawQuery: "range=1h", + observedAt: time.Date(2026, time.July, 27, 12, 34, 56, 789_000_000, time.UTC), + wantFrom: time.Date(2026, time.July, 27, 11, 34, 56, 789_000_000, time.UTC), + wantTo: time.Date(2026, time.July, 27, 12, 34, 56, 789_000_000, time.UTC), + granularity: requestlog.UsageGranularityMinute, + bucketWidth: int64(5 * time.Minute / time.Millisecond), + }, + { + name: "1 hour at an exact hour includes the preceding hour", + rawQuery: "range=1h", + observedAt: time.Date(2026, time.July, 27, 12, 0, 0, 0, time.UTC), + wantFrom: time.Date(2026, time.July, 27, 11, 0, 0, 0, time.UTC), + wantTo: time.Date(2026, time.July, 27, 12, 0, 0, 0, time.UTC), + granularity: requestlog.UsageGranularityMinute, + bucketWidth: int64(5 * time.Minute / time.Millisecond), + }, + { + name: "custom one hour retains hourly aggregation", + rawQuery: "from_ms=1785146400000&to_ms=1785150000000", observedAt: time.Date(2026, time.July, 27, 12, 34, 56, 789, time.UTC), - wantFrom: time.Date(2026, time.July, 27, 12, 0, 0, 0, time.UTC), - wantTo: time.Date(2026, time.July, 27, 13, 0, 0, 0, time.UTC), + wantFrom: time.UnixMilli(1785146400000), + wantTo: time.UnixMilli(1785150000000), granularity: requestlog.UsageGranularityHour, bucketWidth: int64(time.Hour / time.Millisecond), }, @@ -158,7 +176,7 @@ func TestUsageAPIReturnsExactPresetRange(t *testing.T) { rangeValue string bucketWidthMS int64 }{ - {rangeValue: "1h", bucketWidthMS: int64(time.Hour / time.Millisecond)}, + {rangeValue: "1h", bucketWidthMS: int64(5 * time.Minute / time.Millisecond)}, {rangeValue: "24h", bucketWidthMS: int64(time.Hour / time.Millisecond)}, {rangeValue: "3d", bucketWidthMS: int64(3 * time.Hour / time.Millisecond)}, {rangeValue: "7d", bucketWidthMS: int64(6 * time.Hour / time.Millisecond)}, @@ -189,6 +207,41 @@ func TestUsageAPIReturnsExactPresetRange(t *testing.T) { } } +func TestUsageAPIRollingHourRefreshUsesNewObservationAndPreservesFilters(t *testing.T) { + t.Parallel() + now := time.Date(2026, time.September, 7, 14, 5, 23, 123_000_000, time.UTC) + reader := &recordingUsageStatReader{} + engine, fixture := newUsageTestEngine(t, now, reader) + fixture.service.now = func() time.Time { return now } + for _, observation := range []time.Time{now, now.Add(2 * time.Minute)} { + now = observation + recorder := performUsageRequest(engine, "test-auth-key", + "range=1h&group_id=7&channel_id=openai&credential_id=11&upstream_model=usage-model") + if recorder.Code != http.StatusOK { + t.Fatalf("rolling usage response = %d %s", recorder.Code, recorder.Body.String()) + } + var envelope struct { + Data usageResponse `json:"data"` + } + if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode rolling usage response: %v", err) + } + data := envelope.Data + if data.Range != "1h" || data.Granularity != "minute" || + data.BucketWidthMS != int64(5*time.Minute/time.Millisecond) || + data.FromMS != now.Add(-time.Hour).UnixMilli() || data.ToMS != now.UnixMilli() || + data.ObservedAtMS != now.UnixMilli() { + t.Fatalf("rolling usage response window = %+v", data) + } + query := reader.queries[len(reader.queries)-1] + if query.GroupID == nil || *query.GroupID != 7 || query.ChannelID != "openai" || + query.CredentialID == nil || *query.CredentialID != 11 || + query.UpstreamModel != "usage-model" { + t.Fatalf("rolling usage filters = %+v", query) + } + } +} + func TestUsageAPIReturnsDistributionWithoutCredentialIdentity(t *testing.T) { t.Parallel() now := time.Date(2026, time.August, 12, 8, 0, 0, 0, time.UTC) diff --git a/internal/requestlog/external_database_integration_test.go b/internal/requestlog/external_database_integration_test.go index 0da9991d..01ee025d 100644 --- a/internal/requestlog/external_database_integration_test.go +++ b/internal/requestlog/external_database_integration_test.go @@ -138,6 +138,36 @@ func TestExternalDatabaseRequestLogLifecycle(t *testing.T) { *credentialActivity.LastUsedAtMS != row.CompletedAtMS { t.Fatalf("credential activity = %+v, want one exact successful attempt", credentialActivity) } + usageQuery := UsageQuery{ + FromMS: row.CompletedAtMS + 1234 - time.Hour.Milliseconds(), + ToMS: row.CompletedAtMS + 1234, + Granularity: UsageGranularityMinute, + BucketWidthMS: UsageFiveMinuteBucketMS, + GroupID: &groupID, + ChannelID: "openai", + CredentialID: &credentialID, + UpstreamModel: modelID, + } + usageReport, err := service.QueryUsage(t.Context(), usageQuery) + if err != nil { + t.Fatalf("minute QueryUsage() error = %v", err) + } + if usageReport.Summary.RequestCount != 1 || + usageReport.Summary.EstimatedCostNanoUSD != row.EstimatedCostNanoUSD || + len(usageReport.Series) != 1 || usageReport.Series[0].BucketStartMS != row.CompletedAtMS || + usageReport.Series[0].BucketEndMS != usageQuery.ToMS { + t.Fatalf("minute usage report = %#v, want one frozen request in its five-minute bucket", usageReport) + } + assertMinuteUsageReportTotals(t, usageReport) + if len(usageReport.Distributions.Group) != 3 || len(usageReport.Distributions.AccessKey) != 3 { + t.Fatalf("minute admin distributions = %#v", usageReport.Distributions) + } + usageQuery.FromMS -= 1234 + usageQuery.ToMS -= 1234 + excludedUsage, err := service.QueryUsage(t.Context(), usageQuery) + if err != nil || excludedUsage.Summary.RequestCount != 0 || len(excludedUsage.Series) != 0 { + t.Fatalf("minute upper-bound exclusion = %#v/%v, want no request at ToMS", excludedUsage, err) + } service.Sweep(context.Background(), now) assertExternalRequestLogCount(t, db, "id = ?", []any{requestID}, 0) assertExternalUsageStatCount( diff --git a/internal/requestlog/types.go b/internal/requestlog/types.go index 82c687ff..0f5b39c0 100644 --- a/internal/requestlog/types.go +++ b/internal/requestlog/types.go @@ -161,8 +161,10 @@ type Page struct { type UsageGranularity string const ( - UsageGranularityHour UsageGranularity = "hour" - UsageGranularityDay UsageGranularity = "day" + UsageGranularityMinute UsageGranularity = "minute" + UsageGranularityHour UsageGranularity = "hour" + UsageGranularityDay UsageGranularity = "day" + UsageFiveMinuteBucketMS int64 = 5 * 60 * 1000 ) type UsageDistributionDimension string diff --git a/internal/requestlog/usage_query.go b/internal/requestlog/usage_query.go index 4aa9a7b1..a858ffa4 100644 --- a/internal/requestlog/usage_query.go +++ b/internal/requestlog/usage_query.go @@ -38,19 +38,30 @@ func (service *Service) QueryUsage(ctx context.Context, input UsageQuery) (Usage CleanupTimeout: usageRollbackTimeout, Operation: "usage read transaction", }, func(connection *gorm.DB) error { - if err := validateUsageStatIntegrity(usageStatScope(connection, input)); err != nil { + scope := usageStatScope + alignmentMS := epochms.MillisecondsPerHour + if input.Granularity == UsageGranularityMinute { + scope = usageRequestLogScope + alignmentMS = UsageFiveMinuteBucketMS + } + if err := validateUsageIntegrity(scope(connection, input), alignmentMS); err != nil { return err } - summary, err := queryUsageSummary(usageStatScope(connection, input)) + summary, err := queryUsageSummary(scope(connection, input)) if err != nil { return err } - series, err := queryUsageSeries(usageStatScope(connection, input), bucketWidthMS) + var series []UsageSeriesPoint + if input.Granularity == UsageGranularityMinute { + series, err = queryUsageMinuteSeries(scope(connection, input), input) + } else { + series, err = queryUsageSeries(scope(connection, input), bucketWidthMS) + } if err != nil { return err } distributions, err := queryUsageDistributions( - usageStatScope(connection, input), summary, input.AccessKeyID != nil, + scope(connection, input), summary, input.AccessKeyID != nil, ) if err != nil { return err @@ -85,6 +96,14 @@ func validateUsageQuery(input UsageQuery) (int64, error) { } bucketWidthMS := input.BucketWidthMS switch input.Granularity { + case UsageGranularityMinute: + if input.ToMS-input.FromMS != epochms.MillisecondsPerHour { + return 0, fmt.Errorf("query usage: minute granularity requires a one-hour range") + } + if bucketWidthMS != 0 && bucketWidthMS != UsageFiveMinuteBucketMS { + return 0, fmt.Errorf("query usage: minute granularity requires a five-minute bucket") + } + return UsageFiveMinuteBucketMS, nil case UsageGranularityHour: if bucketWidthMS == 0 { bucketWidthMS = epochms.MillisecondsPerHour @@ -183,11 +202,15 @@ func usageStatScope(db *gorm.DB, input UsageQuery) *gorm.DB { } func validateUsageStatIntegrity(scope *gorm.DB) error { + return validateUsageIntegrity(scope, epochms.MillisecondsPerHour) +} + +func validateUsageIntegrity(scope *gorm.DB, bucketAlignmentMS int64) error { var integrity usageStatIntegrity if err := scope.Select(` COALESCE(MAX(CASE WHEN bucket_start_ms < 0 - OR bucket_start_ms % 3600000 != 0 + OR bucket_start_ms % ? != 0 THEN 1 ELSE 0 END), 0) AS invalid_bucket, COALESCE(MAX(CASE WHEN request_count < 0 OR success_count < 0 OR failure_count < 0 @@ -207,7 +230,7 @@ func validateUsageStatIntegrity(scope *gorm.DB) error { COALESCE(MAX(CASE WHEN estimated_cost_nano_usd < 0 THEN 1 ELSE 0 END), 0) AS invalid_cost - `).Find(&integrity).Error; err != nil { + `, bucketAlignmentMS).Find(&integrity).Error; err != nil { return fmt.Errorf("check usage stat integrity: %w", err) } if integrity.InvalidBucket != 0 || integrity.InvalidCount != 0 || diff --git a/internal/requestlog/usage_query_minute.go b/internal/requestlog/usage_query_minute.go new file mode 100644 index 00000000..a5e54a4c --- /dev/null +++ b/internal/requestlog/usage_query_minute.go @@ -0,0 +1,79 @@ +package requestlog + +import ( + "fmt" + + "gorm.io/gorm" + + "gpt-load/internal/storage/models" +) + +// 将请求级已保存数据投影为小时聚合相同的统计字段，在数据库内复用总览和分布查询。 +func usageRequestLogScope(db *gorm.DB, input UsageQuery) *gorm.DB { + logs := db.Session(&gorm.Session{NewDB: true}).Model(&models.RequestLog{}). + Where("completed_at_ms >= ? AND completed_at_ms < ?", input.FromMS, input.ToMS). + Where("attempt_count > 0") + if input.GroupID != nil { + logs = logs.Where("group_id = ?", *input.GroupID) + } + if input.ChannelID != "" { + logs = logs.Where("channel_id = ?", input.ChannelID) + } + if input.CredentialID != nil { + logs = logs.Where("credential_id = ?", *input.CredentialID) + } + if input.AccessKeyID != nil { + logs = logs.Where("access_key_id = ?", *input.AccessKeyID) + } + if input.UpstreamModel != "" { + logs = logs.Where("upstream_model = ?", input.UpstreamModel) + } + return db.Session(&gorm.Session{NewDB: true}). + Table("(?) AS usage_rows", logs.Select(usageRequestLogProjection, UsageFiveMinuteBucketMS)) +} + +func queryUsageMinuteSeries(scope *gorm.DB, input UsageQuery) ([]UsageSeriesPoint, error) { + var source []usageHourPoint + if err := scope.Select("bucket_start_ms, " + usageAggregateSelect). + Group("bucket_start_ms").Order("bucket_start_ms ASC").Find(&source).Error; err != nil { + return nil, fmt.Errorf("query minute usage series: %w", err) + } + series := make([]UsageSeriesPoint, 0, len(source)) + for _, point := range source { + if err := validateUsageAggregate(point.UsageAggregate); err != nil { + return nil, fmt.Errorf("validate minute usage series: %w", err) + } + // 首尾自然桶截取到查询范围；先比较差值，避免接近 int64 上界时相加溢出。 + endMS := input.ToMS + if input.ToMS-point.BucketStartMS > UsageFiveMinuteBucketMS { + endMS = point.BucketStartMS + UsageFiveMinuteBucketMS + } + series = append(series, UsageSeriesPoint{ + BucketStartMS: max(point.BucketStartMS, input.FromMS), + BucketEndMS: endMS, + UsageAggregate: point.UsageAggregate, + }) + } + return series, nil +} + +// 与 usageStatDelta.addRow 保持一致：只统计已完成的最终归属，每个请求仅计一次； +// missing/not_applicable 不累加 Token，missing 也不计入可报价用量的 unpriced 数量。 +const usageRequestLogProjection = ` + completed_at_ms - completed_at_ms % ? AS bucket_start_ms, + group_id, access_key_id, upstream_model AS model, + 1 AS request_count, + CASE WHEN status = 'success' THEN 1 ELSE 0 END AS success_count, + CASE WHEN status IN ('error', 'incomplete', 'canceled') THEN 1 ELSE 0 END AS failure_count, + CASE WHEN usage_state IN ('complete', 'partial') THEN uncached_input_tokens ELSE 0 END AS uncached_input_tokens, + CASE WHEN usage_state IN ('complete', 'partial') THEN cache_read_tokens ELSE 0 END AS cache_read_tokens, + CASE WHEN usage_state IN ('complete', 'partial') THEN cache_write_5m_tokens ELSE 0 END AS cache_write_5m_tokens, + CASE WHEN usage_state IN ('complete', 'partial') THEN cache_write_1h_tokens ELSE 0 END AS cache_write_1h_tokens, + CASE WHEN usage_state IN ('complete', 'partial') THEN cache_write_unknown_tokens ELSE 0 END AS cache_write_unknown_tokens, + CASE WHEN usage_state IN ('complete', 'partial') THEN output_tokens ELSE 0 END AS output_tokens, + estimated_cost_nano_usd, + CASE WHEN usage_state = 'missing' THEN 1 ELSE 0 END AS usage_missing_count, + CASE WHEN usage_state = 'partial' THEN 1 ELSE 0 END AS partial_count, + CASE WHEN usage_state IN ('complete', 'partial') AND cost_state = 'unpriced' THEN 1 ELSE 0 END AS unpriced_request_count, + CASE WHEN cost_state = 'priced' AND pricing_completeness = 'partial' THEN 1 ELSE 0 END AS pricing_partial_count +` diff --git a/internal/requestlog/usage_query_minute_test.go b/internal/requestlog/usage_query_minute_test.go new file mode 100644 index 00000000..cfbb02ad --- /dev/null +++ b/internal/requestlog/usage_query_minute_test.go @@ -0,0 +1,318 @@ +package requestlog + +import ( + "context" + "math" + "reflect" + "testing" + "time" + + "gorm.io/gorm" + + "gpt-load/internal/channel" + "gpt-load/internal/storage/models" +) + +func TestQueryUsageMinuteReadsRollingLogsAndClampsBuckets(t *testing.T) { + db := openRequestLogQueryDB(t) + start := time.Date(2026, time.September, 7, 13, 2, 17, 123_000_000, time.UTC) + end := start.Add(time.Hour) + rows := []models.RequestLog{ + aggregationRow(aggregationRequestID(100), start.Add(-time.Millisecond), 7, "model"), + aggregationRow(aggregationRequestID(101), start, 7, "model"), + aggregationRow(aggregationRequestID(102), start.Truncate(time.Hour).Add(time.Hour), 7, "model"), + aggregationRow(aggregationRequestID(103), end.Add(-time.Millisecond), 7, "model"), + aggregationRow(aggregationRequestID(104), end, 7, "model"), + aggregationRow(aggregationRequestID(105), start.Add(time.Minute), 7, "model"), + } + rows[5].AttemptCount = 0 + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + createUsageStats(t, db, usageStat(start.Truncate(time.Hour), 7, "model", 999)) + report, err := newRequestLogTestService(db).QueryUsage(context.Background(), minuteUsageQuery(start)) + if err != nil { + t.Fatalf("QueryUsage() error = %v", err) + } + if report.Summary.RequestCount != 3 || report.Summary.EstimatedCostNanoUSD != 750_000_000 { + t.Fatalf("summary = %#v, want three in-range attempted requests", report.Summary) + } + if len(report.Series) != 2 || report.Series[0].BucketStartMS != start.UnixMilli() || + report.Series[0].BucketEndMS != start.Truncate(time.Hour).Add(5*time.Minute).UnixMilli() || + report.Series[0].RequestCount != 1 || + report.Series[1].BucketStartMS != end.Truncate(time.Hour).UnixMilli() || + report.Series[1].BucketEndMS != end.UnixMilli() || report.Series[1].RequestCount != 2 { + t.Fatalf("series = %#v, want sparse five-minute buckets clipped to rolling range", report.Series) + } + assertMinuteUsageReportTotals(t, report) + hourQuery := UsageQuery{ + FromMS: start.Truncate(time.Hour).UnixMilli(), + ToMS: start.Truncate(time.Hour).Add(24 * time.Hour).UnixMilli(), + Granularity: UsageGranularityHour, + } + hour, err := newRequestLogTestService(db).QueryUsage(context.Background(), hourQuery) + if err != nil || hour.Summary.RequestCount != 999 { + t.Fatalf("hourly report = %#v/%v, want unchanged hourly aggregate source", hour, err) + } +} + +func TestQueryUsageMinuteCanIncludeThirteenPartialAndFullBuckets(t *testing.T) { + db := openRequestLogQueryDB(t) + start := time.Date(2026, time.September, 7, 13, 2, 0, 0, time.UTC) + rows := []models.RequestLog{aggregationRow(aggregationRequestID(600), start, 7, "model")} + for index := 1; index <= 12; index++ { + rows = append(rows, aggregationRow(aggregationRequestID(600+index), start.Truncate(time.Hour).Add(time.Duration(index)*5*time.Minute), 7, "model")) + } + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + query := minuteUsageQuery(start) + query.BucketWidthMS = UsageFiveMinuteBucketMS + report, err := newRequestLogTestService(db).QueryUsage(context.Background(), query) + if err != nil || len(report.Series) != 13 { + t.Fatalf("QueryUsage() series/error = %#v/%v, want thirteen natural buckets", report.Series, err) + } + for index, point := range report.Series { + if point.BucketStartMS < query.FromMS || point.BucketEndMS > query.ToMS || + point.BucketStartMS >= point.BucketEndMS || point.RequestCount != 1 { + t.Fatalf("invalid bucket %d: %#v", index, point) + } + if index > 0 && report.Series[index-1].BucketEndMS != point.BucketStartMS { + t.Fatalf("noncontiguous buckets %d and %d", index-1, index) + } + } + assertMinuteUsageReportTotals(t, report) +} + +func TestQueryUsageMinuteClipsBucketNearTimestampLimit(t *testing.T) { + db := openRequestLogQueryDB(t) + start := time.UnixMilli(math.MaxInt64 - time.Hour.Milliseconds()) + row := aggregationRow(aggregationRequestID(700), time.UnixMilli(math.MaxInt64-1), 7, "model") + if err := db.Create(&row).Error; err != nil { + t.Fatal(err) + } + report, err := newRequestLogTestService(db).QueryUsage(context.Background(), minuteUsageQuery(start)) + if err != nil || len(report.Series) != 1 || report.Series[0].BucketEndMS != math.MaxInt64 || + report.Series[0].BucketStartMS < start.UnixMilli() { + t.Fatalf("QueryUsage() series/error = %#v/%v, want clipped bucket without overflow", report.Series, err) + } +} + +func TestQueryUsageMinuteMatchesFrozenHourlyAccounting(t *testing.T) { + db := openRequestLogQueryDB(t) + start := time.Date(2026, time.September, 7, 13, 0, 0, 0, time.UTC) + rows := make([]models.RequestLog, 7) + for index := range rows { + rows[index] = aggregationRow(aggregationRequestID(200+index), start.Add(time.Duration(index)*5*time.Minute), 7, "model") + rows[index].CacheReadTokens = 3 + rows[index].CacheWrite5MTokens = 4 + rows[index].CacheWrite1HTokens = 5 + rows[index].CacheWriteUnknownTokens = 6 + } + rows[1].Status = "error" + rows[1].UsageState, rows[1].CostState, rows[1].PricingCompleteness = "missing", "unpriced", "unavailable" + rows[1].EstimatedCostNanoUSD = 0 + rows[2].Status = "incomplete" + rows[2].UsageState, rows[2].PricingCompleteness = "partial", "partial" + rows[3].Status = "canceled" + rows[3].UsageState, rows[3].CostState, rows[3].PricingCompleteness = "partial", "unpriced", "unavailable" + rows[3].EstimatedCostNanoUSD = 0 + rows[4].CostState, rows[4].PricingCompleteness = "unpriced", "unavailable" + rows[4].EstimatedCostNanoUSD = 0 + rows[5].UsageState, rows[5].CostState, rows[5].PricingCompleteness = "not_applicable", "not_applicable", "not_applicable" + rows[5].EstimatedCostNanoUSD = 0 + rows[6].AttemptCount = 0 + if err := (&gormBatchWriter{db: db}).WriteBatch(context.Background(), rows); err != nil { + t.Fatal(err) + } + service := newRequestLogTestService(db) + minute, err := service.QueryUsage(context.Background(), minuteUsageQuery(start)) + if err != nil { + t.Fatalf("minute QueryUsage() error = %v", err) + } + hourQuery := minuteUsageQuery(start) + hourQuery.Granularity = UsageGranularityHour + hour, err := service.QueryUsage(context.Background(), hourQuery) + if err != nil { + t.Fatalf("hour QueryUsage() error = %v", err) + } + if minute.Summary != hour.Summary || !reflect.DeepEqual(minute.Distributions, hour.Distributions) { + t.Fatalf("frozen accounting differs: minute=%#v hour=%#v", minute, hour) + } + if minute.Summary.RequestCount != 6 || minute.Summary.SuccessCount != 3 || minute.Summary.FailureCount != 3 || + minute.Summary.UsageMissingCount != 1 || minute.Summary.PartialCount != 2 || + minute.Summary.UnpricedRequestCount != 2 || minute.Summary.PricingPartialCount != 1 || + minute.Summary.EstimatedCostNanoUSD != 500_000_000 || minute.Summary.UncachedInputTokens != 4 { + t.Fatalf("summary = %#v", minute.Summary) + } + assertMinuteUsageReportTotals(t, minute) +} + +func TestQueryUsageMinuteFiltersFinalAttributionAndAccessKey(t *testing.T) { + db := openRequestLogQueryDB(t) + start := time.Date(2026, time.September, 7, 13, 0, 0, 0, time.UTC) + rows := make([]models.RequestLog, 6) + for index := range rows { + rows[index] = aggregationRow(aggregationRequestID(300+index), start.Add(time.Minute), 7, "final-model") + rows[index].ChannelID, rows[index].CredentialID, rows[index].AccessKeyID = "openai", 11, 41 + rows[index].AttemptCount = 2 + } + rows[1].GroupID = 8 + rows[2].ChannelID = "anthropic" + rows[3].CredentialID = 12 + rows[4].AccessKeyID = 42 + rows[5].UpstreamModel = "other-model" + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + // 先前尝试既不能匹配最终归属筛选，也不能让同一请求被重复计数。 + for index := range rows { + attempts := []models.RequestLogAttempt{ + {RequestID: rows[index].ID, Sequence: 1, CompletedAtMS: rows[index].CompletedAtMS, GroupID: 99, CredentialID: 99, ChannelID: "gemini", UpstreamModel: "earlier-model", FailureCategory: "rate_limited", Action: "retry"}, + {RequestID: rows[index].ID, Sequence: 2, CompletedAtMS: rows[index].CompletedAtMS, GroupID: rows[index].GroupID, CredentialID: rows[index].CredentialID, ChannelID: rows[index].ChannelID, UpstreamModel: rows[index].UpstreamModel, FailureCategory: "ok", Action: "terminate"}, + } + if err := db.Create(&attempts).Error; err != nil { + t.Fatal(err) + } + } + groupID, credentialID, accessKeyID := uint(7), uint(11), uint(41) + query := minuteUsageQuery(start) + query.GroupID, query.ChannelID, query.CredentialID = &groupID, channel.OpenAI, &credentialID + query.AccessKeyID, query.UpstreamModel = &accessKeyID, "final-model" + report, err := newRequestLogTestService(db).QueryUsage(context.Background(), query) + if err != nil { + t.Fatalf("QueryUsage() error = %v", err) + } + if report.Summary.RequestCount != 1 || len(report.Distributions.Group) != 0 || len(report.Distributions.AccessKey) != 0 { + t.Fatalf("scoped report = %#v", report) + } + assertMinuteUsageReportTotals(t, report) + query.GroupID, query.ChannelID, query.CredentialID = nil, "", nil + query.UpstreamModel = "earlier-model" + report, err = newRequestLogTestService(db).QueryUsage(context.Background(), query) + if err != nil || report.Summary.RequestCount != 0 { + t.Fatalf("earlier attempt filter = %#v/%v, want empty report", report, err) + } +} + +func TestQueryUsageMinuteUsesOneReadSnapshot(t *testing.T) { + db, dsn := openRequestLogFileDB(t) + start := time.Date(2026, time.September, 7, 13, 2, 0, 0, time.UTC) + row := aggregationRow(aggregationRequestID(400), start, 7, "before") + if err := db.Create(&row).Error; err != nil { + t.Fatal(err) + } + writerDB, closeWriter := openUsageQueryWriterDB(t, dsn) + defer closeWriter() + inserted := false + const callbackName = "test:minute_usage_snapshot_insert" + if err := db.Callback().Query().After("gorm:query").Register(callbackName, func(tx *gorm.DB) { + if inserted || tx.DryRun { + return + } + inserted = true + after := aggregationRow(aggregationRequestID(401), start.Add(time.Minute), 8, "after") + if err := writerDB.Create(&after).Error; err != nil { + t.Errorf("insert concurrent request log: %v", err) + } + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := db.Callback().Query().Remove(callbackName); err != nil { + t.Error(err) + } + }) + report, err := newRequestLogTestService(db).QueryUsage(context.Background(), minuteUsageQuery(start)) + if err != nil || !inserted || report.Summary.RequestCount != 1 { + t.Fatalf("snapshot report/inserted/error = %#v/%t/%v", report, inserted, err) + } + assertMinuteUsageReportTotals(t, report) +} + +func TestQueryUsageMinuteRejectsInvalidRanges(t *testing.T) { + start := time.Date(2026, time.September, 7, 13, 2, 0, 0, time.UTC) + for _, test := range []struct { + name string + duration time.Duration + width time.Duration + }{ + {"short range", 59 * time.Minute, 5 * time.Minute}, + {"long range", 61 * time.Minute, 5 * time.Minute}, + {"wrong bucket", time.Hour, time.Minute}, + } { + t.Run(test.name, func(t *testing.T) { + query := minuteUsageQuery(start) + query.ToMS, query.BucketWidthMS = start.Add(test.duration).UnixMilli(), test.width.Milliseconds() + if _, err := validateUsageQuery(query); err == nil { + t.Fatal("validateUsageQuery() error = nil, want invalid minute range rejection") + } + }) + } +} + +func TestQueryUsageMinuteRejectsTokenAndCostOverflow(t *testing.T) { + for _, field := range []string{"uncached_input_tokens", "estimated_cost_nano_usd"} { + t.Run(field, func(t *testing.T) { + db := openRequestLogQueryDB(t) + start := time.Date(2026, time.September, 7, 13, 2, 0, 0, time.UTC) + rows := []models.RequestLog{ + aggregationRow(aggregationRequestID(500), start, 7, "model"), + aggregationRow(aggregationRequestID(501), start.Add(time.Minute), 8, "model"), + } + if err := db.Create(&rows).Error; err != nil { + t.Fatal(err) + } + if err := db.Model(&models.RequestLog{}).Where("id = ?", rows[0].ID).Update(field, int64(math.MaxInt64)).Error; err != nil { + t.Fatal(err) + } + if _, err := newRequestLogTestService(db).QueryUsage(context.Background(), minuteUsageQuery(start)); err == nil { + t.Fatal("QueryUsage() error = nil, want numeric overflow rejection") + } + }) + } +} + +func minuteUsageQuery(start time.Time) UsageQuery { + return UsageQuery{FromMS: start.UnixMilli(), ToMS: start.Add(time.Hour).UnixMilli(), Granularity: UsageGranularityMinute} +} + +func assertMinuteUsageReportTotals(t *testing.T, report UsageReport) { + t.Helper() + var total UsageAggregate + for _, point := range report.Series { + var err error + total, err = addUsageAggregates(total, point.UsageAggregate) + if err != nil { + t.Fatal(err) + } + } + if total != report.Summary { + t.Fatalf("series sum = %#v, want summary %#v", total, report.Summary) + } + tokens, err := usageAggregateTotalTokens(report.Summary) + if err != nil { + t.Fatal(err) + } + want := UsageDistributionAggregate{RequestCount: total.RequestCount, TotalTokens: tokens, EstimatedCostNanoUSD: total.EstimatedCostNanoUSD} + for _, distributions := range []map[UsageDistributionMetric]UsageDistribution{report.Distributions.Group, report.Distributions.Model, report.Distributions.AccessKey} { + for _, distribution := range distributions { + var got UsageDistributionAggregate + for _, item := range distribution.Items { + got, err = addUsageDistributionAggregates(got, item.UsageDistributionAggregate) + if err != nil { + t.Fatal(err) + } + } + if distribution.Other != nil { + got, err = addUsageDistributionAggregates(got, *distribution.Other) + if err != nil { + t.Fatal(err) + } + } + if got != want { + t.Fatalf("%s/%s total = %#v, want %#v", distribution.Dimension, distribution.Metric, got, want) + } + } + } +} diff --git a/web/src/app/resources/usage.ts b/web/src/app/resources/usage.ts index 78296cf8..d5f82ce1 100644 --- a/web/src/app/resources/usage.ts +++ b/web/src/app/resources/usage.ts @@ -57,7 +57,7 @@ export interface UsageDistributionAggregateDto { export interface UsageReportDto { range: UsageRange - granularity: 'hour' | 'day' + granularity: 'minute' | 'hour' | 'day' bucket_width_ms: number from_ms: number to_ms: number @@ -126,11 +126,12 @@ const reportFields = [ ] as const const hourMs = 60 * 60 * 1000 const dayMs = 24 * hourMs +const fiveMinutesMs = 5 * 60 * 1000 const usageRangeContract: Record< UsageRange, - { granularity: 'hour' | 'day'; bucketWidthMs: number; buckets: number } + { granularity: UsageReportDto['granularity']; bucketWidthMs: number; buckets: number } > = { - '1h': { granularity: 'hour', bucketWidthMs: hourMs, buckets: 1 }, + '1h': { granularity: 'minute', bucketWidthMs: fiveMinutesMs, buckets: 12 }, '24h': { granularity: 'hour', bucketWidthMs: hourMs, buckets: 24 }, '3d': { granularity: 'hour', bucketWidthMs: 3 * hourMs, buckets: 24 }, '7d': { granularity: 'hour', bucketWidthMs: 6 * hourMs, buckets: 28 }, @@ -217,21 +218,24 @@ export function projectUsageReport(value: unknown): UsageReportDto { const record = projectRecord(value) assertNoSecretLikeFields(record, reportFields) const range = projectEnum(record.range, usageRanges) - const granularity = projectEnum(record.granularity, ['hour', 'day'] as const) + const granularity = projectEnum(record.granularity, ['minute', 'hour', 'day'] as const) const rangeContract = usageRangeContract[range] if (granularity !== rangeContract.granularity) invalidResponse() - const bucketWidthMs = projectSafeInteger(record.bucket_width_ms, { minimum: hourMs }) + const bucketWidthMs = projectSafeInteger(record.bucket_width_ms, { minimum: fiveMinutesMs }) if (bucketWidthMs !== rangeContract.bucketWidthMs) invalidResponse() const observedAtMS = projectEpochMilliseconds(record.observed_at_ms) const rangeFromMS = projectEpochMilliseconds(record.from_ms) const rangeToMS = projectEpochMilliseconds(record.to_ms) const bucketCount = rangeContract.buckets + const rollingHour = range === '1h' if ( - !isUTCAligned(rangeFromMS, bucketWidthMs) || - !isUTCAligned(rangeToMS, bucketWidthMs) || rangeToMS - rangeFromMS !== bucketWidthMs * bucketCount || - observedAtMS < rangeToMS - bucketWidthMs || - observedAtMS >= rangeToMS + (rollingHour + ? rangeToMS !== observedAtMS + : !isUTCAligned(rangeFromMS, bucketWidthMs) || + !isUTCAligned(rangeToMS, bucketWidthMs) || + observedAtMS < rangeToMS - bucketWidthMs || + observedAtMS >= rangeToMS) ) { invalidResponse() } @@ -247,10 +251,15 @@ export function projectUsageReport(value: unknown): UsageReportDto { ]) const bucketStartMS = projectEpochMilliseconds(item.bucket_start_ms) const bucketEndMS = projectEpochMilliseconds(item.bucket_end_ms) + const alignedBucketStartMS = Math.floor(bucketStartMS / bucketWidthMs) * bucketWidthMs if ( - !isUTCAligned(bucketStartMS, bucketWidthMs) || - !isUTCAligned(bucketEndMS, bucketWidthMs) || - bucketEndMS - bucketStartMS !== bucketWidthMs || + (rollingHour + ? bucketStartMS !== Math.max(alignedBucketStartMS, rangeFromMS) || + bucketEndMS !== Math.min(alignedBucketStartMS + bucketWidthMs, rangeToMS) + : !isUTCAligned(bucketStartMS, bucketWidthMs) || + !isUTCAligned(bucketEndMS, bucketWidthMs) || + bucketEndMS - bucketStartMS !== bucketWidthMs) || + bucketEndMS <= bucketStartMS || bucketStartMS < rangeFromMS || bucketEndMS > rangeToMS || bucketStartMS < previousBucketEndMS diff --git a/web/src/components/charts/TrendChart.vue b/web/src/components/charts/TrendChart.vue index 8b71da4b..a727a3bc 100644 --- a/web/src/components/charts/TrendChart.vue +++ b/web/src/components/charts/TrendChart.vue @@ -26,6 +26,7 @@ const props = withDefaults( failureLabel: string rangeStart: number rangeEnd: number + centerBuckets?: boolean locale?: string showBucketRange?: boolean showSinglePoint?: boolean @@ -33,6 +34,7 @@ const props = withDefaults( }>(), { locale: 'en-US', + centerBuckets: false, showBucketRange: false, showSinglePoint: false, failureRateLabel: undefined, @@ -70,6 +72,7 @@ const geometry = computed(() => maximumFailureHeight, props.rangeStart, props.rangeEnd, + props.centerBuckets, ), ) const singlePoint = computed(() => diff --git a/web/src/components/charts/trend-chart.ts b/web/src/components/charts/trend-chart.ts index 31f54145..bf59b7fa 100644 --- a/web/src/components/charts/trend-chart.ts +++ b/web/src/components/charts/trend-chart.ts @@ -149,6 +149,7 @@ export function buildTrendGeometry( maximumFailureHeight: number, rangeStart: number, rangeEnd: number, + centerBuckets = false, ): TrendGeometry { const plotTop = validDimension(chartHeight) ? Math.min(14, chartHeight * 0.1) : 0 const plotBottomInset = validDimension(chartHeight) ? Math.min(8, chartHeight * 0.1) : 0 @@ -181,9 +182,11 @@ export function buildTrendGeometry( const points = parsed.map((datum) => { return { x: round( - pointRangeDuration === 0 - ? width / 2 - : ((datum.start - rangeStart) / pointRangeDuration) * width, + centerBuckets + ? ((datum.start - rangeStart + (datum.end - datum.start) / 2) / rangeDuration) * width + : pointRangeDuration === 0 + ? width / 2 + : ((datum.start - rangeStart) / pointRangeDuration) * width, ), y: round( maximumRequests === 0 diff --git a/web/src/features/monitor/UsageTab.vue b/web/src/features/monitor/UsageTab.vue index 74cac7ce..3d5337ce 100644 --- a/web/src/features/monitor/UsageTab.vue +++ b/web/src/features/monitor/UsageTab.vue @@ -18,6 +18,7 @@ import { type UsageDistributionMetric, type UsageFilters, type UsageRange, + type UsageReportDto, } from '@/app/resources/usage' import { monitorLocation } from '@/app/route-locations' import TrendChart from '@/components/charts/TrendChart.vue' @@ -150,6 +151,23 @@ const distributionMetricOptions = computed(() => [ { value: 'cost', label: t('monitor.usage.distribution.metrics.cost') }, ]) const costChartResolution = 1_000_000_000n +const emptyUsageAggregate: UsageAggregateDto = { + request_count: 0, + success_count: 0, + failure_count: 0, + uncached_input_tokens: 0, + cache_read_tokens: 0, + cache_write_5m_tokens: 0, + cache_write_1h_tokens: 0, + cache_write_unknown_tokens: 0, + output_tokens: 0, + total_tokens: 0, + estimated_cost_nano_usd: '0', + usage_missing_count: 0, + partial_count: 0, + unpriced_request_count: 0, + pricing_partial_count: 0, +} const trendMetricOptions = computed(() => [ { value: 'requests', label: t('monitor.usage.trend.metrics.requests') }, { value: 'tokens', label: t('monitor.usage.trend.metrics.tokens') }, @@ -197,8 +215,33 @@ function normalizeTrendCost(value: bigint, maximum: bigint): number { return Number((value * costChartResolution + maximum / 2n) / maximum) } +const trendBuckets = computed(() => { + const current = report.value + if (current?.range !== '1h') return current?.series ?? [] + + const byStart = new Map(current.series.map((bucket) => [bucket.bucket_start_ms, bucket])) + const buckets: UsageReportDto['series'] = [] + const width = current.bucket_width_ms + // 补齐自然五分钟桶，首尾只覆盖最近一小时内的部分。 + for ( + let alignedStart = Math.floor(current.from_ms / width) * width; + alignedStart < current.to_ms; + alignedStart += width + ) { + const start = Math.max(alignedStart, current.from_ms) + buckets.push( + byStart.get(start) ?? { + ...emptyUsageAggregate, + bucket_start_ms: start, + bucket_end_ms: Math.min(alignedStart + width, current.to_ms), + }, + ) + } + return buckets +}) + const requestTrendSeries = computed(() => - (report.value?.series ?? []).map((bucket) => ({ + trendBuckets.value.map((bucket) => ({ bucket_start_ms: bucket.bucket_start_ms, bucket_end_ms: bucket.bucket_end_ms, request_count: bucket.success_count, @@ -207,7 +250,7 @@ const requestTrendSeries = computed(() => ) const barTrendSeries = computed(() => { - const buckets = report.value?.series ?? [] + const buckets = trendBuckets.value if (routeState.value.metric === 'tokens') { return buckets.map((bucket) => ({ bucket_start_ms: bucket.bucket_start_ms, @@ -270,6 +313,9 @@ function rangeLabel(range: UsageRange): string { function granularityLabel(): string { const bucketWidthMS = report.value?.bucket_width_ms + if (report.value?.granularity === 'minute') { + return t('monitor.usage.trend.everyMinutes', { count: (bucketWidthMS ?? 0) / (60 * 1000) }) + } if (bucketWidthMS === 60 * 60 * 1000) return t('monitor.usage.trend.hourly') if (bucketWidthMS === 24 * 60 * 60 * 1000) return t('monitor.usage.trend.daily') return t('monitor.usage.trend.everyHours', { @@ -470,6 +516,7 @@ defineExpose({ openFilters, refresh }) :failure-label="trendPresentation.secondaryLabel ?? ''" :range-start="report.from_ms" :range-end="report.to_ms" + :center-buckets="report.range === '1h'" :locale="locale" show-bucket-range show-single-point diff --git a/web/src/features/monitor/usage-bar-chart.ts b/web/src/features/monitor/usage-bar-chart.ts index 0a272e3d..318e1ebe 100644 --- a/web/src/features/monitor/usage-bar-chart.ts +++ b/web/src/features/monitor/usage-bar-chart.ts @@ -194,7 +194,7 @@ export function buildUsageBarGeometry( const primaryBars = parsed.map((datum, index) => { const groupWidth = groupWidths[index]! - const gap = grouped ? Math.min(0.8, Math.max(0.4, groupWidth * 0.035)) : 0 + const gap = grouped ? Math.min(groupWidth / 2, 0.8, Math.max(0.4, groupWidth * 0.035)) : 0 const barWidth = grouped ? (groupWidth - gap) / 2 : groupWidth const point = points[index]! const height = scaledHeight(datum.primaryValue) @@ -209,7 +209,7 @@ export function buildUsageBarGeometry( const secondaryBars = grouped ? parsed.map((datum, index) => { const groupWidth = groupWidths[index]! - const gap = Math.min(0.8, Math.max(0.4, groupWidth * 0.035)) + const gap = Math.min(groupWidth / 2, 0.8, Math.max(0.4, groupWidth * 0.035)) const barWidth = (groupWidth - gap) / 2 const point = points[index]! const height = scaledHeight(datum.secondaryValue) diff --git a/web/src/i18n/locales/en-US/monitor.ts b/web/src/i18n/locales/en-US/monitor.ts index 95f89cd6..dda9f744 100644 --- a/web/src/i18n/locales/en-US/monitor.ts +++ b/web/src/i18n/locales/en-US/monitor.ts @@ -339,6 +339,7 @@ export default { }, empty: 'No request buckets were returned for this range.', hourly: 'Hourly', + everyMinutes: 'Every {count} minutes', everyHours: 'Every {count} hours', daily: 'Daily', failureRate: 'Failure rate', @@ -347,7 +348,7 @@ export default { }, series: { title: 'UTC buckets', - description: 'Backend aggregates in the selected hourly or daily granularity.', + description: 'Usage and cost aggregated at the selected time granularity.', caption: 'Usage aggregates by UTC time bucket', disclosure: 'View time-bucket details', }, diff --git a/web/src/i18n/locales/ja-JP/monitor.ts b/web/src/i18n/locales/ja-JP/monitor.ts index 0c354d9c..fe5edb4a 100644 --- a/web/src/i18n/locales/ja-JP/monitor.ts +++ b/web/src/i18n/locales/ja-JP/monitor.ts @@ -339,6 +339,7 @@ export default { }, empty: 'この期間のリクエストバケットはありません。', hourly: '時間別', + everyMinutes: '{count}分ごと', everyHours: '{count}時間ごと', daily: '日別', failureRate: '失敗率', @@ -347,7 +348,7 @@ export default { }, series: { title: 'UTC バケット', - description: '選択した時間または日単位でバックエンドが集計した値です。', + description: '選択した時間粒度で集計した使用量とコストです。', caption: 'UTC 時間バケット別の使用量集計', disclosure: '時間バケットの詳細を表示', }, diff --git a/web/src/i18n/locales/zh-CN/monitor.ts b/web/src/i18n/locales/zh-CN/monitor.ts index fba9f7f4..a2e26d82 100644 --- a/web/src/i18n/locales/zh-CN/monitor.ts +++ b/web/src/i18n/locales/zh-CN/monitor.ts @@ -323,6 +323,7 @@ export default { }, empty: '该范围内暂无请求时间桶。', hourly: '按小时', + everyMinutes: '每 {count} 分钟', everyHours: '每 {count} 小时', daily: '按天', failureRate: '失败率', @@ -331,7 +332,7 @@ export default { }, series: { title: 'UTC 时间桶', - description: '后端按所选小时或天粒度返回的聚合。', + description: '按所选时间粒度统计的用量与成本。', caption: '按 UTC 时间桶统计的用量聚合', disclosure: '查看时间桶明细', },