fix(monitor): 修正最近一小时用量与成本统计 (#588)
This commit is contained in:
1 parent
0ddc41d8b7
commit
33fb54bfe6
16 files changed
+755
-46
No files matched your search
+32
-11
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 ||
|
||||
|
||||
@@ -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
|
||||
`
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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(() =>
|
||||
|
||||
@@ -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<TrendPoint>((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
|
||||
|
||||
@@ -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<UsageReportDto['series']>(() => {
|
||||
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<TrendDatum[]>(() =>
|
||||
(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<TrendDatum[]>(() =>
|
||||
)
|
||||
|
||||
const barTrendSeries = computed<UsageBarDatum[]>(() => {
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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',
|
||||
},
|
||||
|
||||
@@ -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: '時間バケットの詳細を表示',
|
||||
},
|
||||
|
||||
@@ -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: '查看时间桶明细',
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user