fix(monitor): 优化消耗分布切换与分组展示

This commit is contained in:
tbphp committed 2026-08-12 22:15:59 +08:00
1 parent 151f0c0b68
commit 1df1d02e2d
15 files changed
+833 -622

No files matched your search

+88 -72
View File
@@ -74,6 +74,11 @@ type usageDistributionResponse struct {
Other *usageDistributionAggregateResponse `json:"other"`
}
type usageDistributionViewsResponse struct {
Group map[requestlog.UsageDistributionMetric]usageDistributionResponse `json:"group,omitempty"`
Model map[requestlog.UsageDistributionMetric]usageDistributionResponse `json:"model"`
}
type usageDistributionAggregateResponse struct {
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
@@ -87,16 +92,16 @@ type usageCollectionHealthResponse struct {
}
type usageResponse struct {
Range string `json:"range"`
Granularity requestlog.UsageGranularity `json:"granularity"`
BucketWidthMS int64 `json:"bucket_width_ms"`
FromMS int64 `json:"from_ms"`
ToMS int64 `json:"to_ms"`
ObservedAtMS int64 `json:"observed_at_ms"`
Summary usageAggregateResponse `json:"summary"`
Series []usageSeriesResponse `json:"series"`
Distribution usageDistributionResponse `json:"distribution"`
CollectionHealth usageCollectionHealthResponse `json:"collection_health"`
Range string `json:"range"`
Granularity requestlog.UsageGranularity `json:"granularity"`
BucketWidthMS int64 `json:"bucket_width_ms"`
FromMS int64 `json:"from_ms"`
ToMS int64 `json:"to_ms"`
ObservedAtMS int64 `json:"observed_at_ms"`
Summary usageAggregateResponse `json:"summary"`
Series []usageSeriesResponse `json:"series"`
Distributions usageDistributionViewsResponse `json:"distributions"`
CollectionHealth usageCollectionHealthResponse `json:"collection_health"`
}
func (service *Service) QueryUsage(
@@ -130,7 +135,6 @@ func (server *Server) handleUsage(c *gin.Context) {
return
}
query.AccessKeyID = &accessKeyID
query.Distribution = requestlog.UsageDistributionDimensionModel
}
report, err := server.service.QueryUsage(c.Request.Context(), query)
if err != nil {
@@ -151,15 +155,13 @@ func parseUsageQuery(rawQuery string, observedAtMS int64) (requestlog.UsageQuery
return requestlog.UsageQuery{}, app_errors.ErrBadRequest
}
allowed := map[string]struct{}{
"range": {},
"from_ms": {},
"to_ms": {},
"group_id": {},
"channel_id": {},
"credential_id": {},
"upstream_model": {},
"distribution": {},
"distribution_metric": {},
"range": {},
"from_ms": {},
"to_ms": {},
"group_id": {},
"channel_id": {},
"credential_id": {},
"upstream_model": {},
}
for key, value := range values {
if _, ok := allowed[key]; !ok || len(value) != 1 {
@@ -170,10 +172,7 @@ func parseUsageQuery(rawQuery string, observedAtMS int64) (requestlog.UsageQuery
if err := validateSafeMilliseconds(observedAtMS); err != nil {
return requestlog.UsageQuery{}, app_errors.ErrInternalServer
}
query := requestlog.UsageQuery{
Distribution: requestlog.UsageDistributionDimensionGroup,
DistributionMetric: requestlog.UsageDistributionMetricRequests,
}
query := requestlog.UsageQuery{}
rangeValue := usageRange24Hours
if value, ok := singleQueryValue(values, "range"); ok {
rangeValue = value
@@ -262,26 +261,6 @@ func parseUsageQuery(rawQuery string, observedAtMS int64) (requestlog.UsageQuery
}
query.UpstreamModel = value
}
if value, ok := singleQueryValue(values, "distribution"); ok {
switch requestlog.UsageDistributionDimension(value) {
case requestlog.UsageDistributionDimensionGroup:
query.Distribution = requestlog.UsageDistributionDimensionGroup
case requestlog.UsageDistributionDimensionModel:
query.Distribution = requestlog.UsageDistributionDimensionModel
default:
return requestlog.UsageQuery{}, app_errors.ErrValidation
}
}
if value, ok := singleQueryValue(values, "distribution_metric"); ok {
switch requestlog.UsageDistributionMetric(value) {
case requestlog.UsageDistributionMetricRequests:
query.DistributionMetric = requestlog.UsageDistributionMetricRequests
case requestlog.UsageDistributionMetricCost:
query.DistributionMetric = requestlog.UsageDistributionMetricCost
default:
return requestlog.UsageQuery{}, app_errors.ErrValidation
}
}
return query, nil
}
@@ -346,13 +325,6 @@ func (service *Service) mapUsageResponse(
if !accessKeyScoped && service.requestLogStats == nil {
return usageResponse{}, app_errors.ErrInternalServer
}
if report.Distribution.Dimension != query.Distribution ||
report.Distribution.Metric != query.DistributionMetric {
return usageResponse{}, fmt.Errorf("map usage response: distribution metadata mismatch")
}
if len(report.Distribution.Items) > usageDistributionLimit {
return usageResponse{}, fmt.Errorf("map usage response: distribution exceeds limit")
}
summary, err := mapUsageAggregate(report.Summary)
if err != nil {
return usageResponse{}, err
@@ -401,10 +373,8 @@ func (service *Service) mapUsageResponse(
},
Summary: summary,
Series: make([]usageSeriesResponse, 0, len(report.Series)),
Distribution: usageDistributionResponse{
Dimension: report.Distribution.Dimension,
Metric: report.Distribution.Metric,
Items: make([]usageDistributionItemResponse, 0, len(report.Distribution.Items)),
Distributions: usageDistributionViewsResponse{
Model: make(map[requestlog.UsageDistributionMetric]usageDistributionResponse, 2),
},
}
if accessKeyScoped {
@@ -427,53 +397,99 @@ func (service *Service) mapUsageResponse(
usageAggregateResponse: aggregate,
})
}
seenDistributionIdentities := make(map[string]struct{}, len(report.Distribution.Items))
for _, row := range report.Distribution.Items {
if !accessKeyScoped {
result.Distributions.Group = make(map[requestlog.UsageDistributionMetric]usageDistributionResponse, 2)
}
for _, dimension := range []requestlog.UsageDistributionDimension{
requestlog.UsageDistributionDimensionGroup,
requestlog.UsageDistributionDimensionModel,
} {
if accessKeyScoped && dimension == requestlog.UsageDistributionDimensionGroup {
continue
}
for _, metric := range []requestlog.UsageDistributionMetric{
requestlog.UsageDistributionMetricRequests,
requestlog.UsageDistributionMetricCost,
} {
distribution, ok := report.Distributions.Get(dimension, metric)
if !ok {
return usageResponse{}, fmt.Errorf("map usage response: distribution view unavailable")
}
if distribution.Dimension != dimension || distribution.Metric != metric {
return usageResponse{}, fmt.Errorf("map usage response: distribution metadata mismatch")
}
mapped, err := mapUsageDistribution(distribution)
if err != nil {
return usageResponse{}, err
}
if err := validateMappedUsageDistribution(result.Summary, mapped); err != nil {
return usageResponse{}, err
}
if dimension == requestlog.UsageDistributionDimensionGroup {
result.Distributions.Group[metric] = mapped
} else {
result.Distributions.Model[metric] = mapped
}
}
}
return result, nil
}
func mapUsageDistribution(
distribution requestlog.UsageDistribution,
) (usageDistributionResponse, error) {
if len(distribution.Items) > usageDistributionLimit {
return usageDistributionResponse{}, fmt.Errorf("map usage response: distribution exceeds limit")
}
result := usageDistributionResponse{
Dimension: distribution.Dimension,
Metric: distribution.Metric,
Items: make([]usageDistributionItemResponse, 0, len(distribution.Items)),
}
seenDistributionIdentities := make(map[string]struct{}, len(distribution.Items))
for _, row := range distribution.Items {
if uint64(row.GroupID) > uint64(maxSafeInteger) {
return usageResponse{}, fmt.Errorf("map usage distribution: unsafe group")
return usageDistributionResponse{}, fmt.Errorf("map usage distribution: unsafe group")
}
identity := row.Model
switch report.Distribution.Dimension {
switch distribution.Dimension {
case requestlog.UsageDistributionDimensionGroup:
if row.GroupID == 0 || row.Model != "" {
return usageResponse{}, fmt.Errorf("map usage distribution: invalid group identity")
return usageDistributionResponse{}, fmt.Errorf("map usage distribution: invalid group identity")
}
identity = strconv.FormatUint(uint64(row.GroupID), 10)
case requestlog.UsageDistributionDimensionModel:
if row.GroupID != 0 || !validUsageDistributionModel(row.Model) {
return usageResponse{}, fmt.Errorf("map usage distribution: invalid model identity")
return usageDistributionResponse{}, fmt.Errorf("map usage distribution: invalid model identity")
}
}
if _, exists := seenDistributionIdentities[identity]; exists {
return usageResponse{}, fmt.Errorf("map usage distribution: duplicate identity")
return usageDistributionResponse{}, fmt.Errorf("map usage distribution: duplicate identity")
}
seenDistributionIdentities[identity] = struct{}{}
aggregate, err := mapUsageDistributionAggregate(row.UsageDistributionAggregate)
if err != nil {
return usageResponse{}, err
return usageDistributionResponse{}, err
}
item := usageDistributionItemResponse{
RequestCount: aggregate.RequestCount,
EstimatedCostNanoUSD: aggregate.EstimatedCostNanoUSD,
}
if report.Distribution.Dimension == requestlog.UsageDistributionDimensionGroup {
if distribution.Dimension == requestlog.UsageDistributionDimensionGroup {
groupID := row.GroupID
item.GroupID = &groupID
} else {
model := row.Model
item.Model = &model
}
result.Distribution.Items = append(result.Distribution.Items, item)
result.Items = append(result.Items, item)
}
if report.Distribution.Other != nil {
other, err := mapUsageDistributionAggregate(*report.Distribution.Other)
if distribution.Other != nil {
other, err := mapUsageDistributionAggregate(*distribution.Other)
if err != nil {
return usageResponse{}, err
return usageDistributionResponse{}, err
}
result.Distribution.Other = &other
}
if err := validateMappedUsageDistribution(result.Summary, result.Distribution); err != nil {
return usageResponse{}, err
result.Other = &other
}
return result, nil
}
+274 -147
View File
@@ -197,62 +197,71 @@ func TestUsageAPIReturnsDistributionWithoutCredentialIdentity(t *testing.T) {
RequestCount: 13, SuccessCount: 11, FailureCount: 2,
EstimatedCostNanoUSD: 2_300_000_000,
},
Distribution: requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricCost,
Items: []requestlog.UsageDistributionItem{{
GroupID: 7,
UsageDistributionAggregate: requestlog.UsageDistributionAggregate{
RequestCount: 10,
EstimatedCostNanoUSD: 2_000_000_000,
},
}},
Other: &other,
},
Distributions: usageTestDistributions(
requestlog.UsageAggregate{
RequestCount: 13, SuccessCount: 11, FailureCount: 2,
EstimatedCostNanoUSD: 2_300_000_000,
},
requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricCost,
Items: []requestlog.UsageDistributionItem{{
GroupID: 7,
UsageDistributionAggregate: requestlog.UsageDistributionAggregate{
RequestCount: 10,
EstimatedCostNanoUSD: 2_000_000_000,
},
}},
Other: &other,
},
),
}}
engine, _ := newUsageTestEngine(t, now, reader)
recorder := performUsageRequest(
engine,
"test-auth-key",
"distribution=group&distribution_metric=cost",
"",
)
if recorder.Code != http.StatusOK {
t.Fatalf("response = %d %s", recorder.Code, recorder.Body.String())
}
if len(reader.queries) != 1 ||
reader.queries[0].Distribution != requestlog.UsageDistributionDimensionGroup ||
reader.queries[0].DistributionMetric != requestlog.UsageDistributionMetricCost {
if len(reader.queries) != 1 {
t.Fatalf("usage query = %#v", reader.queries)
}
var envelope struct {
Data struct {
Distribution struct {
Dimension string `json:"dimension"`
Metric string `json:"metric"`
Items []struct {
GroupID uint `json:"group_id"`
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"items"`
Other *struct {
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"other"`
} `json:"distribution"`
Distributions struct {
Group struct {
Cost struct {
Dimension string `json:"dimension"`
Metric string `json:"metric"`
Items []struct {
GroupID uint `json:"group_id"`
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"items"`
Other *struct {
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"other"`
} `json:"cost"`
} `json:"group"`
} `json:"distributions"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatalf("decode response: %v", err)
}
if envelope.Data.Distribution.Dimension != "group" ||
envelope.Data.Distribution.Metric != "cost" ||
len(envelope.Data.Distribution.Items) != 1 ||
envelope.Data.Distribution.Items[0].GroupID != 7 ||
envelope.Data.Distribution.Items[0].RequestCount != 10 ||
envelope.Data.Distribution.Items[0].EstimatedCostNanoUSD != "2000000000" ||
envelope.Data.Distribution.Other == nil ||
envelope.Data.Distribution.Other.RequestCount != 3 {
t.Fatalf("distribution response = %#v", envelope.Data.Distribution)
distribution := envelope.Data.Distributions.Group.Cost
if distribution.Dimension != "group" ||
distribution.Metric != "cost" ||
len(distribution.Items) != 1 ||
distribution.Items[0].GroupID != 7 ||
distribution.Items[0].RequestCount != 10 ||
distribution.Items[0].EstimatedCostNanoUSD != "2000000000" ||
distribution.Other == nil ||
distribution.Other.RequestCount != 3 {
t.Fatalf("distribution response = %#v", distribution)
}
for _, forbidden := range []string{`"credential_id"`, `"channel_id"`} {
if strings.Contains(recorder.Body.String(), forbidden) {
@@ -282,26 +291,26 @@ func TestUsageAPIDefaultsToFixedUTCAligned24HoursAndReturnsZeroArrays(t *testing
if query.Granularity != requestlog.UsageGranularityHour ||
query.FromMS != time.Date(2026, time.July, 26, 5, 0, 0, 0, time.UTC).UnixMilli() ||
query.ToMS != time.Date(2026, time.July, 27, 5, 0, 0, 0, time.UTC).UnixMilli() ||
query.GroupID != nil || query.UpstreamModel != "" ||
query.Distribution != requestlog.UsageDistributionDimensionGroup ||
query.DistributionMetric != requestlog.UsageDistributionMetricRequests {
query.GroupID != nil || query.UpstreamModel != "" {
t.Fatalf("default UsageQuery = %#v", query)
}
var envelope struct {
Code int `json:"code"`
Data struct {
Range string `json:"range"`
Granularity requestlog.UsageGranularity `json:"granularity"`
FromMS int64 `json:"from_ms"`
ToMS int64 `json:"to_ms"`
ObservedAtMS int64 `json:"observed_at_ms"`
Series []json.RawMessage `json:"series"`
Distribution struct {
Dimension string `json:"dimension"`
Metric string `json:"metric"`
Items []json.RawMessage `json:"items"`
Other json.RawMessage `json:"other"`
} `json:"distribution"`
Range string `json:"range"`
Granularity requestlog.UsageGranularity `json:"granularity"`
FromMS int64 `json:"from_ms"`
ToMS int64 `json:"to_ms"`
ObservedAtMS int64 `json:"observed_at_ms"`
Series []json.RawMessage `json:"series"`
Distributions struct {
Model struct {
Cost struct {
Items []json.RawMessage `json:"items"`
Other json.RawMessage `json:"other"`
} `json:"cost"`
} `json:"model"`
} `json:"distributions"`
Health struct {
Scope string `json:"scope"`
DroppedTotal uint64 `json:"dropped_total"`
@@ -319,10 +328,9 @@ func TestUsageAPIDefaultsToFixedUTCAligned24HoursAndReturnsZeroArrays(t *testing
envelope.Data.ToMS != time.Date(2026, time.July, 27, 5, 0, 0, 0, time.UTC).UnixMilli() ||
envelope.Data.ObservedAtMS != now.UnixMilli() ||
envelope.Data.Series == nil || len(envelope.Data.Series) != 0 ||
envelope.Data.Distribution.Dimension != "group" ||
envelope.Data.Distribution.Metric != "requests" ||
envelope.Data.Distribution.Items == nil || len(envelope.Data.Distribution.Items) != 0 ||
string(envelope.Data.Distribution.Other) != "null" ||
envelope.Data.Distributions.Model.Cost.Items == nil ||
len(envelope.Data.Distributions.Model.Cost.Items) != 0 ||
string(envelope.Data.Distributions.Model.Cost.Other) != "null" ||
envelope.Data.Health.Scope != "current_process" ||
envelope.Data.Health.DroppedTotal != 2 ||
envelope.Data.Health.WriteFailureTotal != 1 ||
@@ -346,6 +354,46 @@ func TestUsageAPIDefaultsToFixedUTCAligned24HoursAndReturnsZeroArrays(t *testing
}
}
func TestUsageAPIReturnsAllDistributionViewsInOneResponse(t *testing.T) {
now := time.Date(2026, time.August, 12, 8, 0, 0, 0, time.UTC)
reader := &recordingUsageStatReader{}
engine, _ := newUsageTestEngine(t, now, reader)
recorder := performUsageRequest(engine, "test-auth-key", "range=24h")
if recorder.Code != http.StatusOK {
t.Fatalf("response = %d %s", recorder.Code, recorder.Body.String())
}
if len(reader.queries) != 1 {
t.Fatalf("QueryUsage calls = %d, want one", len(reader.queries))
}
var envelope struct {
Data struct {
Distribution json.RawMessage `json:"distribution"`
Distributions struct {
Group struct {
Requests json.RawMessage `json:"requests"`
Cost json.RawMessage `json:"cost"`
} `json:"group"`
Model struct {
Requests json.RawMessage `json:"requests"`
Cost json.RawMessage `json:"cost"`
} `json:"model"`
} `json:"distributions"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatalf("decode response: %v", err)
}
if len(envelope.Data.Distribution) != 0 ||
len(envelope.Data.Distributions.Group.Requests) == 0 ||
len(envelope.Data.Distributions.Group.Cost) == 0 ||
len(envelope.Data.Distributions.Model.Requests) == 0 ||
len(envelope.Data.Distributions.Model.Cost) == 0 {
t.Fatalf("usage distributions = %#v; body=%s", envelope.Data.Distributions, recorder.Body.String())
}
}
func TestUsageAPISelectsThirtyUTCDaysAndAppliesFilters(t *testing.T) {
now := time.Date(2026, time.July, 27, 12, 0, 0, 0, time.UTC)
reader := &recordingUsageStatReader{report: requestlog.UsageReport{
@@ -363,23 +411,30 @@ func TestUsageAPISelectsThirtyUTCDaysAndAppliesFilters(t *testing.T) {
EstimatedCostNanoUSD: 1_123_456_789_012,
},
}},
Distribution: requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionModel,
Metric: requestlog.UsageDistributionMetricCost,
Items: []requestlog.UsageDistributionItem{{
Model: "upstream-model",
UsageDistributionAggregate: requestlog.UsageDistributionAggregate{
RequestCount: 1,
EstimatedCostNanoUSD: 250_000_000,
},
}},
},
Distributions: usageTestDistributions(
requestlog.UsageAggregate{
RequestCount: 1, UncachedInputTokens: 2, CacheWriteUnknownTokens: 4,
OutputTokens: 3, PricingPartialCount: 1,
EstimatedCostNanoUSD: 250_000_000,
},
requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionModel,
Metric: requestlog.UsageDistributionMetricCost,
Items: []requestlog.UsageDistributionItem{{
Model: "upstream-model",
UsageDistributionAggregate: requestlog.UsageDistributionAggregate{
RequestCount: 1,
EstimatedCostNanoUSD: 250_000_000,
},
}},
},
),
}}
engine, _ := newUsageTestEngine(t, now, reader)
recorder := performUsageRequest(
engine,
"test-auth-key",
"range=30d&group_id=9&channel_id=openai&credential_id=13&upstream_model=upstream-model&distribution=model&distribution_metric=cost",
"range=30d&group_id=9&channel_id=openai&credential_id=13&upstream_model=upstream-model",
)
if recorder.Code != http.StatusOK {
t.Fatalf("response = %d %s", recorder.Code, recorder.Body.String())
@@ -394,9 +449,7 @@ func TestUsageAPISelectsThirtyUTCDaysAndAppliesFilters(t *testing.T) {
query.GroupID == nil || *query.GroupID != 9 ||
query.ChannelID != channel.OpenAI ||
query.CredentialID == nil || *query.CredentialID != 13 ||
query.UpstreamModel != "upstream-model" ||
query.Distribution != requestlog.UsageDistributionDimensionModel ||
query.DistributionMetric != requestlog.UsageDistributionMetricCost {
query.UpstreamModel != "upstream-model" {
t.Fatalf("30 day UsageQuery = %#v", query)
}
var envelope struct {
@@ -418,15 +471,19 @@ func TestUsageAPISelectsThirtyUTCDaysAndAppliesFilters(t *testing.T) {
PricingPartialCount int64 `json:"pricing_partial_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"series"`
Distribution struct {
Dimension string `json:"dimension"`
Metric string `json:"metric"`
Items []struct {
Model string `json:"model"`
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"items"`
} `json:"distribution"`
Distributions struct {
Model struct {
Cost struct {
Dimension string `json:"dimension"`
Metric string `json:"metric"`
Items []struct {
Model string `json:"model"`
RequestCount int64 `json:"request_count"`
EstimatedCostNanoUSD string `json:"estimated_cost_nano_usd"`
} `json:"items"`
} `json:"cost"`
} `json:"model"`
} `json:"distributions"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
@@ -446,12 +503,12 @@ func TestUsageAPISelectsThirtyUTCDaysAndAppliesFilters(t *testing.T) {
envelope.Data.Series[0].CacheWriteUnknownTokens != 4 ||
envelope.Data.Series[0].PricingPartialCount != 1 ||
envelope.Data.Series[0].EstimatedCostNanoUSD != "1123456789012" ||
envelope.Data.Distribution.Dimension != "model" ||
envelope.Data.Distribution.Metric != "cost" ||
len(envelope.Data.Distribution.Items) != 1 ||
envelope.Data.Distribution.Items[0].Model != "upstream-model" ||
envelope.Data.Distribution.Items[0].RequestCount != 1 ||
envelope.Data.Distribution.Items[0].EstimatedCostNanoUSD != "250000000" {
envelope.Data.Distributions.Model.Cost.Dimension != "model" ||
envelope.Data.Distributions.Model.Cost.Metric != "cost" ||
len(envelope.Data.Distributions.Model.Cost.Items) != 1 ||
envelope.Data.Distributions.Model.Cost.Items[0].Model != "upstream-model" ||
envelope.Data.Distributions.Model.Cost.Items[0].RequestCount != 1 ||
envelope.Data.Distributions.Model.Cost.Items[0].EstimatedCostNanoUSD != "250000000" {
t.Fatalf("usage response = %#v", envelope.Data)
}
for _, legacyField := range []string{`"key_id"`, `"upstream_key_id"`} {
@@ -542,10 +599,10 @@ func TestUsageAPIRejectsStrictInvalidQueriesWithoutCallingReader(t *testing.T) {
{query: "upstream_key_id=1", code: "BAD_REQUEST"},
{query: "model=legacy", code: "BAD_REQUEST"},
{query: "upstream_model=", code: "VALIDATION_FAILED"},
{query: "distribution=", code: "VALIDATION_FAILED"},
{query: "distribution=credential", code: "VALIDATION_FAILED"},
{query: "distribution_metric=", code: "VALIDATION_FAILED"},
{query: "distribution_metric=tokens", code: "VALIDATION_FAILED"},
{query: "distribution=", code: "BAD_REQUEST"},
{query: "distribution=credential", code: "BAD_REQUEST"},
{query: "distribution_metric=", code: "BAD_REQUEST"},
{query: "distribution_metric=tokens", code: "BAD_REQUEST"},
{
query: "distribution_metric=cost&distribution_metric=requests",
code: "BAD_REQUEST",
@@ -565,11 +622,9 @@ func TestUsageAPIRejectsStrictInvalidQueriesWithoutCallingReader(t *testing.T) {
func TestMapUsageRejectsUnsafeOrMismatchedDistribution(t *testing.T) {
fixture := newServiceFixture(t)
query := requestlog.UsageQuery{
FromMS: time.Date(2026, time.July, 27, 0, 0, 0, 0, time.UTC).UnixMilli(),
ToMS: time.Date(2026, time.July, 28, 0, 0, 0, 0, time.UTC).UnixMilli(),
Granularity: requestlog.UsageGranularityHour,
Distribution: requestlog.UsageDistributionDimensionGroup,
DistributionMetric: requestlog.UsageDistributionMetricRequests,
FromMS: time.Date(2026, time.July, 27, 0, 0, 0, 0, time.UTC).UnixMilli(),
ToMS: time.Date(2026, time.July, 28, 0, 0, 0, 0, time.UTC).UnixMilli(),
Granularity: requestlog.UsageGranularityHour,
}
tests := []struct {
name string
@@ -578,70 +633,83 @@ func TestMapUsageRejectsUnsafeOrMismatchedDistribution(t *testing.T) {
{
name: "response dimension mismatch",
report: requestlog.UsageReport{
Distribution: requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionModel,
Metric: requestlog.UsageDistributionMetricRequests,
},
Distributions: usageTestDistributionsWithOverride(
requestlog.UsageAggregate{},
requestlog.UsageDistributionDimensionGroup,
requestlog.UsageDistributionMetricRequests,
requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionModel,
Metric: requestlog.UsageDistributionMetricRequests,
},
),
},
},
{
name: "response metric mismatch",
report: requestlog.UsageReport{
Distribution: requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricCost,
},
Distributions: usageTestDistributionsWithOverride(
requestlog.UsageAggregate{},
requestlog.UsageDistributionDimensionGroup,
requestlog.UsageDistributionMetricRequests,
requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricCost,
},
),
},
},
{
name: "unsafe group identity",
report: requestlog.UsageReport{
Distribution: requestlog.UsageDistribution{
Distributions: usageTestDistributions(requestlog.UsageAggregate{}, requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricRequests,
Items: []requestlog.UsageDistributionItem{{
GroupID: uint(maxSafeInteger) + 1,
}},
},
}),
},
},
{
name: "more than five items",
report: requestlog.UsageReport{
Distribution: requestlog.UsageDistribution{
Distributions: usageTestDistributions(requestlog.UsageAggregate{}, requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricRequests,
Items: []requestlog.UsageDistributionItem{
{GroupID: 1}, {GroupID: 2}, {GroupID: 3},
{GroupID: 4}, {GroupID: 5}, {GroupID: 6},
},
},
}),
},
},
{
name: "duplicate group identity",
report: requestlog.UsageReport{
Distribution: requestlog.UsageDistribution{
Distributions: usageTestDistributions(requestlog.UsageAggregate{}, requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricRequests,
Items: []requestlog.UsageDistributionItem{{GroupID: 1}, {GroupID: 1}},
},
}),
},
},
{
name: "distribution total mismatch",
report: requestlog.UsageReport{
Summary: requestlog.UsageAggregate{RequestCount: 2, SuccessCount: 2},
Distribution: requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricRequests,
Items: []requestlog.UsageDistributionItem{{
GroupID: 1,
UsageDistributionAggregate: requestlog.UsageDistributionAggregate{
RequestCount: 1,
},
}},
},
Distributions: usageTestDistributions(
requestlog.UsageAggregate{RequestCount: 2, SuccessCount: 2},
requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionGroup,
Metric: requestlog.UsageDistributionMetricRequests,
Items: []requestlog.UsageDistributionItem{{
GroupID: 1,
UsageDistributionAggregate: requestlog.UsageDistributionAggregate{
RequestCount: 1,
},
}},
},
),
},
},
}
@@ -734,18 +802,22 @@ func TestUsageAPIExcludesLegacyZeroAttemptAggregateFromSQLite(t *testing.T) {
RequestCount int64 `json:"request_count"`
FailureCount int64 `json:"failure_count"`
} `json:"summary"`
Distribution struct {
Items []json.RawMessage `json:"items"`
Other json.RawMessage `json:"other"`
} `json:"distribution"`
Distributions struct {
Model struct {
Cost struct {
Items []json.RawMessage `json:"items"`
Other json.RawMessage `json:"other"`
} `json:"cost"`
} `json:"model"`
} `json:"distributions"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatalf("decode response: %v", err)
}
if envelope.Data.Summary.RequestCount != 0 || envelope.Data.Summary.FailureCount != 0 ||
len(envelope.Data.Distribution.Items) != 0 ||
string(envelope.Data.Distribution.Other) != "null" {
len(envelope.Data.Distributions.Model.Cost.Items) != 0 ||
string(envelope.Data.Distributions.Model.Cost.Other) != "null" {
t.Fatalf("usage data = %#v", envelope.Data)
}
}
@@ -789,13 +861,11 @@ func TestUsageAPIBindsAccessKeyScopeAndRedactsProcessHealth(t *testing.T) {
}
now := time.Date(2026, time.August, 8, 17, 30, 0, 0, time.UTC)
reader := &recordingUsageStatReader{report: requestlog.UsageReport{
Distribution: requestlog.UsageDistribution{
Distributions: usageTestDistributions(requestlog.UsageAggregate{}, requestlog.UsageDistribution{
Dimension: requestlog.UsageDistributionDimensionModel,
Metric: requestlog.UsageDistributionMetricRequests,
Items: []requestlog.UsageDistributionItem{{
Model: "allowed-model",
}},
},
Items: []requestlog.UsageDistributionItem{{Model: "allowed-model"}},
}),
}}
fixture.service.now = func() time.Time { return now }
fixture.service.usageStats = reader
@@ -809,27 +879,31 @@ func TestUsageAPIBindsAccessKeyScopeAndRedactsProcessHealth(t *testing.T) {
t.Fatalf("AccessKey usage response = %d %s, want 200", recorder.Code, recorder.Body.String())
}
if len(reader.queries) != 1 || reader.queries[0].AccessKeyID == nil ||
*reader.queries[0].AccessKeyID != created.ID || reader.queries[0].GroupID != nil ||
reader.queries[0].Distribution != requestlog.UsageDistributionDimensionModel {
*reader.queries[0].AccessKeyID != created.ID || reader.queries[0].GroupID != nil {
t.Fatalf("AccessKey UsageQuery = %#v", reader.queries)
}
var envelope struct {
Data struct {
Distribution struct {
Dimension string `json:"dimension"`
Items []struct {
Model string `json:"model"`
} `json:"items"`
} `json:"distribution"`
Distributions struct {
Model struct {
Cost struct {
Dimension string `json:"dimension"`
Items []struct {
Model string `json:"model"`
} `json:"items"`
} `json:"cost"`
} `json:"model"`
} `json:"distributions"`
CollectionHealth usageCollectionHealthResponse `json:"collection_health"`
} `json:"data"`
}
if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil {
t.Fatalf("decode AccessKey usage response: %v", err)
}
if envelope.Data.Distribution.Dimension != "model" ||
len(envelope.Data.Distribution.Items) != 1 ||
envelope.Data.Distribution.Items[0].Model != "allowed-model" ||
distribution := envelope.Data.Distributions.Model.Cost
if distribution.Dimension != "model" ||
len(distribution.Items) != 1 ||
distribution.Items[0].Model != "allowed-model" ||
envelope.Data.CollectionHealth.Scope != "access_key" ||
envelope.Data.CollectionHealth.DroppedTotal != 0 ||
envelope.Data.CollectionHealth.WriteFailureTotal != 0 ||
@@ -857,6 +931,62 @@ type recordingUsageStatReader struct {
err error
}
func usageTestDistributions(
summary requestlog.UsageAggregate,
source requestlog.UsageDistribution,
) requestlog.UsageDistributions {
result := requestlog.UsageDistributions{
Group: make(map[requestlog.UsageDistributionMetric]requestlog.UsageDistribution, 2),
Model: make(map[requestlog.UsageDistributionMetric]requestlog.UsageDistribution, 2),
}
for _, dimension := range []requestlog.UsageDistributionDimension{
requestlog.UsageDistributionDimensionGroup,
requestlog.UsageDistributionDimensionModel,
} {
for _, metric := range []requestlog.UsageDistributionMetric{
requestlog.UsageDistributionMetricRequests,
requestlog.UsageDistributionMetricCost,
} {
distribution := requestlog.UsageDistribution{
Dimension: dimension,
Metric: metric,
}
if source.Dimension == dimension {
distribution = source
distribution.Dimension = dimension
distribution.Metric = metric
} else if summary.RequestCount > 0 || summary.EstimatedCostNanoUSD > 0 {
other := requestlog.UsageDistributionAggregate{
RequestCount: summary.RequestCount,
EstimatedCostNanoUSD: summary.EstimatedCostNanoUSD,
}
distribution.Other = &other
}
if dimension == requestlog.UsageDistributionDimensionGroup {
result.Group[metric] = distribution
} else {
result.Model[metric] = distribution
}
}
}
return result
}
func usageTestDistributionsWithOverride(
summary requestlog.UsageAggregate,
dimension requestlog.UsageDistributionDimension,
metric requestlog.UsageDistributionMetric,
distribution requestlog.UsageDistribution,
) requestlog.UsageDistributions {
result := usageTestDistributions(summary, requestlog.UsageDistribution{})
if dimension == requestlog.UsageDistributionDimensionGroup {
result.Group[metric] = distribution
} else {
result.Model[metric] = distribution
}
return result
}
func (reader *recordingUsageStatReader) QueryUsage(
_ context.Context,
query requestlog.UsageQuery,
@@ -866,11 +996,8 @@ func (reader *recordingUsageStatReader) QueryUsage(
return requestlog.UsageReport{}, reader.err
}
report := reader.report
if report.Distribution.Dimension == "" {
report.Distribution.Dimension = query.Distribution
}
if report.Distribution.Metric == "" {
report.Distribution.Metric = query.DistributionMetric
if len(report.Distributions.Group) == 0 && len(report.Distributions.Model) == 0 {
report.Distributions = usageTestDistributions(report.Summary, requestlog.UsageDistribution{})
}
return report, nil
}
+28 -30
View File
@@ -187,20 +187,19 @@ func TestQueryUsageFiltersAndProjectsChannelCredential(t *testing.T) {
credentialID := uint(11)
report, err := newRequestLogTestService(db).QueryUsage(t.Context(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
ChannelID: channel.OpenAI,
CredentialID: &credentialID,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
ChannelID: channel.OpenAI,
CredentialID: &credentialID,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if report.Summary.RequestCount != 2 || len(report.Distribution.Items) != 1 ||
report.Distribution.Items[0].GroupID != 7 ||
report.Distribution.Items[0].RequestCount != 2 {
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, UsageDistributionMetricRequests)
if report.Summary.RequestCount != 2 || len(distribution.Items) != 1 ||
distribution.Items[0].GroupID != 7 ||
distribution.Items[0].RequestCount != 2 {
t.Fatalf("scoped usage report = %#v", report)
}
}
@@ -223,27 +222,26 @@ func TestAccessScopedUsageDistributionCollapsesHiddenRoutesByModel(t *testing.T)
accessKeyID := uint(41)
report, err := newRequestLogTestService(db).QueryUsage(t.Context(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &accessKeyID,
Distribution: UsageDistributionDimensionModel,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &accessKeyID,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if len(report.Distribution.Items) != 1 {
modelDistribution := usageDistribution(t, report, UsageDistributionDimensionModel, UsageDistributionMetricRequests)
if len(modelDistribution.Items) != 1 {
t.Fatalf("access-scoped distribution = %#v", report)
}
distribution := report.Distribution.Items[0]
distribution := modelDistribution.Items[0]
if distribution.GroupID != 0 || distribution.Model != "shared-model" ||
distribution.RequestCount != 10 {
t.Fatalf("access-scoped distribution exposes or fragments routes: %#v", distribution)
}
}
func TestAccessScopedUsageForcesModelDistribution(t *testing.T) {
func TestAccessScopedUsageReturnsOnlyModelDistributions(t *testing.T) {
db := openRequestLogQueryDB(t)
start := time.Date(2026, time.August, 9, 2, 45, 0, 0, time.UTC).Truncate(time.Hour)
first := usageStat(start, 7, "shared-model", 2)
@@ -254,21 +252,21 @@ func TestAccessScopedUsageForcesModelDistribution(t *testing.T) {
accessKeyID := uint(41)
report, err := newRequestLogTestService(db).QueryUsage(t.Context(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &accessKeyID,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &accessKeyID,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if report.Distribution.Dimension != UsageDistributionDimensionModel ||
len(report.Distribution.Items) != 1 ||
report.Distribution.Items[0].Model != "shared-model" ||
report.Distribution.Items[0].RequestCount != 5 {
t.Fatalf("access-scoped distribution = %#v", report.Distribution)
distribution := usageDistribution(t, report, UsageDistributionDimensionModel, UsageDistributionMetricRequests)
if len(report.Distributions.Group) != 0 ||
distribution.Dimension != UsageDistributionDimensionModel ||
len(distribution.Items) != 1 ||
distribution.Items[0].Model != "shared-model" ||
distribution.Items[0].RequestCount != 5 {
t.Fatalf("access-scoped distributions = %#v", report.Distributions)
}
}
+34 -14
View File
@@ -173,17 +173,15 @@ const (
)
type UsageQuery struct {
FromMS int64
ToMS int64
Granularity UsageGranularity
BucketWidthMS int64
AccessKeyID *uint
GroupID *uint
ChannelID channel.ID
CredentialID *uint
UpstreamModel string
Distribution UsageDistributionDimension
DistributionMetric UsageDistributionMetric
FromMS int64
ToMS int64
Granularity UsageGranularity
BucketWidthMS int64
AccessKeyID *uint
GroupID *uint
ChannelID channel.ID
CredentialID *uint
UpstreamModel string
}
type UsageAggregate struct {
@@ -228,9 +226,31 @@ type UsageDistribution struct {
}
type UsageReport struct {
Summary UsageAggregate
Series []UsageSeriesPoint
Distribution UsageDistribution
Summary UsageAggregate
Series []UsageSeriesPoint
Distributions UsageDistributions
}
type UsageDistributions struct {
Group map[UsageDistributionMetric]UsageDistribution
Model map[UsageDistributionMetric]UsageDistribution
}
func (distributions UsageDistributions) Get(
dimension UsageDistributionDimension,
metric UsageDistributionMetric,
) (UsageDistribution, bool) {
var values map[UsageDistributionMetric]UsageDistribution
switch dimension {
case UsageDistributionDimensionGroup:
values = distributions.Group
case UsageDistributionDimensionModel:
values = distributions.Model
default:
return UsageDistribution{}, false
}
distribution, ok := values[metric]
return distribution, ok
}
type Stats struct {
+42 -33
View File
@@ -27,15 +27,6 @@ func (service *Service) QueryUsage(ctx context.Context, input UsageQuery) (Usage
if ctx == nil {
ctx = context.Background()
}
if input.Distribution == "" {
input.Distribution = UsageDistributionDimensionGroup
}
if input.DistributionMetric == "" {
input.DistributionMetric = UsageDistributionMetricRequests
}
if input.AccessKeyID != nil {
input.Distribution = UsageDistributionDimensionModel
}
bucketWidthMS, err := validateUsageQuery(input)
if err != nil {
return UsageReport{}, err
@@ -58,17 +49,14 @@ func (service *Service) QueryUsage(ctx context.Context, input UsageQuery) (Usage
if err != nil {
return err
}
distribution, err := queryUsageDistribution(
usageStatScope(connection, input),
summary,
input.Distribution,
input.DistributionMetric,
distributions, err := queryUsageDistributions(
usageStatScope(connection, input), summary, input.AccessKeyID != nil,
)
if err != nil {
return err
}
report = UsageReport{
Summary: summary, Series: series, Distribution: distribution,
Summary: summary, Series: series, Distributions: distributions,
}
return nil
})
@@ -125,25 +113,44 @@ func validateUsageQuery(input UsageQuery) (int64, error) {
(input.ToMS-input.FromMS)%bucketWidthMS != 0 {
return 0, fmt.Errorf("query usage: time range is not bucket aligned")
}
switch input.Distribution {
case UsageDistributionDimensionGroup, UsageDistributionDimensionModel:
default:
return 0, fmt.Errorf(
"query usage: unsupported distribution dimension %q",
input.Distribution,
)
}
switch input.DistributionMetric {
case UsageDistributionMetricRequests, UsageDistributionMetricCost:
default:
return 0, fmt.Errorf(
"query usage: unsupported distribution metric %q",
input.DistributionMetric,
)
}
return bucketWidthMS, nil
}
func queryUsageDistributions(
scope *gorm.DB,
summary UsageAggregate,
accessKeyScoped bool,
) (UsageDistributions, error) {
result := UsageDistributions{
Group: make(map[UsageDistributionMetric]UsageDistribution, 2),
Model: make(map[UsageDistributionMetric]UsageDistribution, 2),
}
dimensions := []UsageDistributionDimension{
UsageDistributionDimensionGroup,
UsageDistributionDimensionModel,
}
if accessKeyScoped {
dimensions = []UsageDistributionDimension{UsageDistributionDimensionModel}
}
for _, dimension := range dimensions {
for _, metric := range []UsageDistributionMetric{
UsageDistributionMetricRequests,
UsageDistributionMetricCost,
} {
distribution, err := queryUsageDistribution(scope, summary, dimension, metric)
if err != nil {
return UsageDistributions{}, err
}
if dimension == UsageDistributionDimensionGroup {
result.Group[metric] = distribution
} else {
result.Model[metric] = distribution
}
}
}
return result, nil
}
func usageStatScope(db *gorm.DB, input UsageQuery) *gorm.DB {
scope := db.Session(&gorm.Session{NewDB: true}).Model(&models.UsageStat{}).
Where("bucket_start_ms >= ? AND bucket_start_ms < ?", input.FromMS, input.ToMS).
@@ -255,10 +262,12 @@ func queryUsageDistribution(
metric UsageDistributionMetric,
) (UsageDistribution, error) {
var rows []usageDistributionRow
query := scope
query := scope.Session(&gorm.Session{})
switch dimension {
case UsageDistributionDimensionGroup:
query = query.Select("group_id, '' AS model, " + usageDistributionAggregateSelect).
query = query.Select("group_id, '' AS model, "+usageDistributionAggregateSelect).
Where("group_id IN (?)", scope.Session(&gorm.Session{NewDB: true}).
Model(&models.Group{}).Select("id")).
Group("group_id")
case UsageDistributionDimensionModel:
query = query.Select("0 AS group_id, model, " + usageDistributionAggregateSelect).
+180 -109
View File
@@ -36,13 +36,11 @@ func TestQueryUsageHourAggregatesFiltersAndLeavesSparseBucketsAbsent(t *testing.
groupID := uint(7)
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(24 * time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
GroupID: &groupID,
UpstreamModel: "target",
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(24 * time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
GroupID: &groupID,
UpstreamModel: "target",
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
@@ -70,9 +68,10 @@ func TestQueryUsageHourAggregatesFiltersAndLeavesSparseBucketsAbsent(t *testing.
)
}
}
if len(report.Distribution.Items) != 1 || report.Distribution.Items[0].GroupID != 7 ||
report.Distribution.Items[0].RequestCount != 292 || report.Distribution.Other != nil {
t.Fatalf("filtered distribution = %#v", report.Distribution)
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, UsageDistributionMetricRequests)
if len(distribution.Items) != 1 || distribution.Items[0].GroupID != 7 ||
distribution.Items[0].RequestCount != 292 || distribution.Other != nil {
t.Fatalf("filtered distribution = %#v", distribution)
}
}
@@ -93,11 +92,9 @@ func TestQueryUsageExcludesLegacyZeroAttemptAggregates(t *testing.T) {
createUsageStats(t, db, attributed, legacyZeroAttempt)
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
@@ -110,9 +107,10 @@ func TestQueryUsageExcludesLegacyZeroAttemptAggregates(t *testing.T) {
report.Series[0].FailureCount != 0 {
t.Fatalf("series = %#v, want only attributed requests", report.Series)
}
if len(report.Distribution.Items) != 1 || report.Distribution.Items[0].GroupID != 7 ||
report.Distribution.Items[0].RequestCount != 3 || report.Distribution.Other != nil {
t.Fatalf("distribution = %#v", report.Distribution)
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, UsageDistributionMetricRequests)
if len(distribution.Items) != 1 || distribution.Items[0].GroupID != 7 ||
distribution.Items[0].RequestCount != 3 || distribution.Other != nil {
t.Fatalf("distribution = %#v", distribution)
}
}
@@ -133,28 +131,27 @@ func TestQueryUsageScopesAccessKeyAndDistributesByModel(t *testing.T) {
accessKeyID := uint(41)
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &accessKeyID,
Distribution: UsageDistributionDimensionModel,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &accessKeyID,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if report.Summary.RequestCount != 6 || len(report.Distribution.Items) != 2 {
distribution := usageDistribution(t, report, UsageDistributionDimensionModel, UsageDistributionMetricRequests)
if report.Summary.RequestCount != 6 || len(distribution.Items) != 2 {
t.Fatalf("scoped report = %#v", report)
}
if report.Distribution.Items[0].GroupID != 0 ||
report.Distribution.Items[0].Model != "shared-model" ||
report.Distribution.Items[0].RequestCount != 5 {
t.Fatalf("model distribution[0] = %#v", report.Distribution.Items[0])
if distribution.Items[0].GroupID != 0 ||
distribution.Items[0].Model != "shared-model" ||
distribution.Items[0].RequestCount != 5 {
t.Fatalf("model distribution[0] = %#v", distribution.Items[0])
}
if report.Distribution.Items[1].GroupID != 0 ||
report.Distribution.Items[1].Model != "other-model" ||
report.Distribution.Items[1].RequestCount != 1 {
t.Fatalf("model distribution[1] = %#v", report.Distribution.Items[1])
if distribution.Items[1].GroupID != 0 ||
distribution.Items[1].Model != "other-model" ||
distribution.Items[1].RequestCount != 1 {
t.Fatalf("model distribution[1] = %#v", distribution.Items[1])
}
}
@@ -165,12 +162,10 @@ func TestQueryUsageRejectsZeroAccessKeyScope(t *testing.T) {
zero := uint(0)
_, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &zero,
Distribution: UsageDistributionDimensionModel,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
AccessKeyID: &zero,
})
if err == nil {
t.Fatal("QueryUsage() error = nil, want zero AccessKey scope rejection")
@@ -196,27 +191,80 @@ func TestQueryUsageDistributionAggregatesCredentialsAndFoldsRemainder(t *testing
createUsageStats(t, db, rows...)
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if len(report.Distribution.Items) != 5 || report.Distribution.Items[0].GroupID != 1 ||
report.Distribution.Items[0].RequestCount != 27 {
t.Fatalf("group distribution = %#v, want credentials aggregated into top five", report.Distribution)
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, UsageDistributionMetricRequests)
if len(distribution.Items) != 5 || distribution.Items[0].GroupID != 1 ||
distribution.Items[0].RequestCount != 27 {
t.Fatalf("group distribution = %#v, want credentials aggregated into top five", distribution)
}
if report.Distribution.Other == nil || report.Distribution.Other.RequestCount != 7 {
t.Fatalf("other distribution = %#v, want remaining two groups aggregated", report.Distribution.Other)
if distribution.Other == nil || distribution.Other.RequestCount != 7 {
t.Fatalf("other distribution = %#v, want remaining two groups aggregated", distribution.Other)
}
if report.Summary.RequestCount != 60 {
t.Fatalf("summary request count = %d, want 60", report.Summary.RequestCount)
}
}
func TestQueryUsageGroupDistributionKeepsOnlyPersistedGroupsInTopFive(t *testing.T) {
db := openRequestLogQueryDB(t)
service := newRequestLogTestService(db)
start := time.Date(2026, time.July, 2, 0, 30, 0, 0, time.UTC).Truncate(time.Hour)
for _, groupID := range []uint{1, 2} {
if err := db.Create(&models.Group{
ID: groupID,
Name: fmt.Sprintf("known-group-%d", groupID),
ChannelID: "openai",
Params: models.JSON(`{}`),
Models: models.JSON(`[]`),
Enabled: true,
}).Error; err != nil {
t.Fatalf("create known group %d: %v", groupID, err)
}
}
knownFirst := usageStat(start, 1, "known-first", 30)
knownFirst.EstimatedCostNanoUSD = 3_000_000_000
knownSecond := usageStat(start, 2, "known-second", 20)
knownSecond.EstimatedCostNanoUSD = 2_000_000_000
deleted := usageStat(start, 90, "deleted", 1_000)
deleted.EstimatedCostNanoUSD = 100_000_000_000
unknown := usageStat(start, 91, "unknown", 900)
unknown.EstimatedCostNanoUSD = 90_000_000_000
createUsageStatsWithoutGroups(t, db, knownFirst, knownSecond, deleted, unknown)
for _, metric := range []UsageDistributionMetric{
UsageDistributionMetricRequests,
UsageDistributionMetricCost,
} {
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
})
if err != nil {
t.Fatalf("QueryUsage(%s) error = %v", metric, err)
}
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, metric)
if len(distribution.Items) != 2 ||
distribution.Items[0].GroupID != 1 ||
distribution.Items[1].GroupID != 2 {
t.Fatalf("%s distribution items = %#v, want only persisted groups", metric, distribution.Items)
}
if distribution.Other == nil ||
distribution.Other.RequestCount != 1_900 ||
distribution.Other.EstimatedCostNanoUSD != 190_000_000_000 {
t.Fatalf("%s distribution other = %#v, want deleted and unknown totals", metric, distribution.Other)
}
}
}
func TestQueryUsageModelDistributionAggregatesGroupsAndCredentials(t *testing.T) {
db := openRequestLogQueryDB(t)
service := newRequestLogTestService(db)
@@ -233,33 +281,17 @@ func TestQueryUsageModelDistributionAggregatesGroupsAndCredentials(t *testing.T)
createUsageStats(t, db, rows...)
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimensionModel,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if len(report.Distribution.Items) != 2 || report.Distribution.Items[0].Model != "shared-model" ||
report.Distribution.Items[0].RequestCount != 12 || report.Distribution.Other != nil {
t.Fatalf("model distribution = %#v", report.Distribution)
}
}
func TestQueryUsageRejectsInvalidDistributionDimension(t *testing.T) {
db := openRequestLogQueryDB(t)
start := time.Date(2026, time.July, 2, 0, 0, 0, 0, time.UTC)
_, err := newRequestLogTestService(db).QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimension("credential"),
DistributionMetric: UsageDistributionMetricRequests,
})
if err == nil {
t.Fatal("QueryUsage() error = nil, want invalid distribution dimension rejection")
distribution := usageDistribution(t, report, UsageDistributionDimensionModel, UsageDistributionMetricRequests)
if len(distribution.Items) != 2 || distribution.Items[0].Model != "shared-model" ||
distribution.Items[0].RequestCount != 12 || distribution.Other != nil {
t.Fatalf("model distribution = %#v", distribution)
}
}
@@ -408,30 +440,29 @@ func TestQueryUsageOrdersDistributionByRequestsAndCost(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
report, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: test.metric,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
})
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
if report.Distribution.Dimension != UsageDistributionDimensionGroup ||
report.Distribution.Metric != test.metric {
t.Fatalf("distribution metadata = %#v", report.Distribution)
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, test.metric)
if distribution.Dimension != UsageDistributionDimensionGroup ||
distribution.Metric != test.metric {
t.Fatalf("distribution metadata = %#v", distribution)
}
if len(report.Distribution.Items) != len(test.want) {
t.Fatalf("distribution = %#v, want %d rows", report.Distribution, len(test.want))
if len(distribution.Items) != len(test.want) {
t.Fatalf("distribution = %#v, want %d rows", distribution, len(test.want))
}
for index, wantGroupID := range test.want {
if report.Distribution.Items[index].GroupID != wantGroupID {
if distribution.Items[index].GroupID != wantGroupID {
t.Fatalf(
"distribution[%d].GroupID = %d, want %d; report=%#v",
index,
report.Distribution.Items[index].GroupID,
distribution.Items[index].GroupID,
wantGroupID,
report.Distribution,
distribution,
)
}
}
@@ -439,22 +470,6 @@ func TestQueryUsageOrdersDistributionByRequestsAndCost(t *testing.T) {
}
}
func TestQueryUsageRejectsInvalidDistributionMetric(t *testing.T) {
db := openRequestLogQueryDB(t)
service := newRequestLogTestService(db)
start := time.Date(2026, time.July, 2, 3, 0, 0, 0, time.UTC)
_, err := service.QueryUsage(context.Background(), UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetric("tokens"),
})
if err == nil {
t.Fatal("QueryUsage() error = nil, want invalid distribution metric rejection")
}
}
func TestQueryUsageUsesOneReadSnapshot(t *testing.T) {
db, dsn := openRequestLogFileDB(t)
service := newRequestLogTestService(db)
@@ -496,8 +511,9 @@ func TestQueryUsageUsesOneReadSnapshot(t *testing.T) {
if err != nil {
t.Fatalf("QueryUsage() error = %v", err)
}
distribution := usageDistribution(t, report, UsageDistributionDimensionGroup, UsageDistributionMetricRequests)
if !inserted || report.Summary.RequestCount != 1 || len(report.Series) != 1 ||
len(report.Distribution.Items) != 1 || report.Distribution.Items[0].GroupID != 1 {
len(distribution.Items) != 1 || distribution.Items[0].GroupID != 1 {
t.Fatalf("report did not retain one snapshot: inserted=%t report=%#v", inserted, report)
}
}
@@ -569,14 +585,24 @@ func TestQueryUsageRejectsCorruptRowsOutsideTopDistribution(t *testing.T) {
corrupt.FailureCount = 0
corrupt.UncachedInputTokens = -1
rows = append(rows, compensating, corrupt)
for groupID := uint(1); groupID <= 102; groupID++ {
if err := db.Create(&models.Group{
ID: groupID,
Name: fmt.Sprintf("integrity-group-%d", groupID),
ChannelID: "openai",
Params: models.JSON(`{}`),
Models: models.JSON(`[]`),
Enabled: true,
}).Error; err != nil {
t.Fatalf("create integrity group %d: %v", groupID, err)
}
}
createCorruptUsageStats(t, db, rows...)
input := UsageQuery{
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
Distribution: UsageDistributionDimensionGroup,
DistributionMetric: UsageDistributionMetricRequests,
FromMS: start.UnixMilli(),
ToMS: start.Add(time.Hour).UnixMilli(),
Granularity: UsageGranularityHour,
}
summary, err := queryUsageSummary(usageStatScope(db, input))
if err != nil || summary.RequestCount != 303 {
@@ -588,8 +614,8 @@ func TestQueryUsageRejectsCorruptRowsOutsideTopDistribution(t *testing.T) {
if distribution, err := queryUsageDistribution(
usageStatScope(db, input),
summary,
input.Distribution,
input.DistributionMetric,
UsageDistributionDimensionGroup,
UsageDistributionMetricRequests,
); err != nil || len(distribution.Items) != 5 || distribution.Items[4].GroupID != 5 {
t.Fatalf("pre-integrity distribution = %#v/%v, want valid top five", distribution, err)
}
@@ -600,6 +626,20 @@ func TestQueryUsageRejectsCorruptRowsOutsideTopDistribution(t *testing.T) {
}
}
func usageDistribution(
t *testing.T,
report UsageReport,
dimension UsageDistributionDimension,
metric UsageDistributionMetric,
) UsageDistribution {
t.Helper()
distribution, ok := report.Distributions.Get(dimension, metric)
if !ok {
t.Fatalf("distribution %s/%s unavailable: %#v", dimension, metric, report.Distributions)
}
return distribution
}
func TestQueryUsageRejectsCorruptAggregates(t *testing.T) {
start := time.Date(2026, time.July, 4, 0, 0, 0, 0, time.UTC)
tests := []struct {
@@ -717,6 +757,37 @@ func usageStat(hour time.Time, groupID uint, model string, requestCount int64) m
}
func createUsageStats(t *testing.T, db *gorm.DB, rows ...models.UsageStat) {
t.Helper()
groupIDs := make(map[uint]struct{})
for _, row := range rows {
if row.GroupID == 0 {
continue
}
groupIDs[row.GroupID] = struct{}{}
}
for groupID := range groupIDs {
var count int64
if err := db.Model(&models.Group{}).Where("id = ?", groupID).Count(&count).Error; err != nil {
t.Fatalf("count group %d: %v", groupID, err)
}
if count > 0 {
continue
}
if err := db.Create(&models.Group{
ID: groupID,
Name: fmt.Sprintf("usage-group-%d", groupID),
ChannelID: "openai",
Params: models.JSON(`{}`),
Models: models.JSON(`[]`),
Enabled: true,
}).Error; err != nil {
t.Fatalf("create usage group %d: %v", groupID, err)
}
}
createUsageStatsWithoutGroups(t, db, rows...)
}
func createUsageStatsWithoutGroups(t *testing.T, db *gorm.DB, rows ...models.UsageStat) {
t.Helper()
for _, row := range rows {
if err := db.Create(&row).Error; err != nil {
-2
View File
@@ -53,8 +53,6 @@ export function normalizeAccessKeyCollectionFilters(
function normalizeUsageFilters(filters: UsageFilters): UsageFilters {
const result: UsageFilters = {
range: filters.range,
distribution: filters.distribution ?? 'group',
distribution_metric: filters.distribution_metric ?? 'requests',
}
if (filters.group_id !== undefined) result.group_id = filters.group_id
if (filters.channel_id !== undefined) result.channel_id = filters.channel_id
+108 -84
View File
@@ -25,19 +25,12 @@ export type UsageRange = TimeRange
export interface UsageFilters {
range: UsageRange
distribution?: UsageDistributionDimension
distribution_metric?: UsageDistributionMetric
group_id?: number
channel_id?: string
credential_id?: number
upstream_model?: string
}
type NormalizedUsageFilters = UsageFilters & {
distribution: UsageDistributionDimension
distribution_metric: UsageDistributionMetric
}
export interface UsageAggregateDto {
request_count: number
success_count: number
@@ -70,11 +63,9 @@ export interface UsageReportDto {
observed_at_ms: number
summary: UsageAggregateDto
series: Array<UsageAggregateDto & { bucket_start_ms: number; bucket_end_ms: number }>
distribution: {
dimension: UsageDistributionDimension
metric: UsageDistributionMetric
items: Array<UsageDistributionAggregateDto & { group_id?: number; model?: string }>
other: UsageDistributionAggregateDto | null
distributions: {
group?: Record<UsageDistributionMetric, UsageDistributionDto>
model: Record<UsageDistributionMetric, UsageDistributionDto>
}
collection_health: {
scope: 'current_process' | 'access_key'
@@ -84,6 +75,13 @@ export interface UsageReportDto {
}
}
export interface UsageDistributionDto {
dimension: UsageDistributionDimension
metric: UsageDistributionMetric
items: Array<UsageDistributionAggregateDto & { group_id?: number; model?: string }>
other: UsageDistributionAggregateDto | null
}
const aggregateKeys = [
'request_count',
'success_count',
@@ -111,7 +109,7 @@ const reportFields = [
'observed_at_ms',
'summary',
'series',
'distribution',
'distributions',
'collection_health',
] as const
const hourMs = 60 * 60 * 1000
@@ -255,60 +253,100 @@ export function projectUsageReport(value: unknown): UsageReportDto {
bucket_end_ms: bucketEndMS,
}
})
const distributionRecord = projectRecord(record.distribution)
assertNoSecretLikeFields(distributionRecord, ['dimension', 'metric', 'items', 'other'])
const distributionDimension = projectEnum(distributionRecord.dimension, [
'group',
'model',
] as const)
const distributionMetric = projectEnum(distributionRecord.metric, ['requests', 'cost'] as const)
const identities = new Set<string>()
const distributionItems = projectArray(distributionRecord.items, (value) => {
const item = projectRecord(value)
const identityField = distributionDimension === 'group' ? 'group_id' : 'model'
assertNoSecretLikeFields(item, [...distributionAggregateFields, identityField])
const aggregate = projectUsageDistributionAggregate(
Object.fromEntries(distributionAggregateFields.map((field) => [field, item[field]])),
)
if (distributionDimension === 'group') {
const groupID = projectSafeInteger(item.group_id, { minimum: 1 })
if (identities.has(String(groupID))) invalidResponse()
identities.add(String(groupID))
return { ...aggregate, group_id: groupID }
const distributionsRecord = projectRecord(record.distributions)
assertNoSecretLikeFields(distributionsRecord, ['group', 'model'])
function projectDistribution(
value: unknown,
expectedDimension: UsageDistributionDimension,
expectedMetric: UsageDistributionMetric,
): UsageDistributionDto {
const distributionRecord = projectRecord(value)
assertNoSecretLikeFields(distributionRecord, ['dimension', 'metric', 'items', 'other'])
const distributionDimension = projectEnum(distributionRecord.dimension, [
'group',
'model',
] as const)
const distributionMetric = projectEnum(distributionRecord.metric, ['requests', 'cost'] as const)
if (distributionDimension !== expectedDimension || distributionMetric !== expectedMetric) {
invalidResponse()
}
const model = projectString(item.model, { allowEmpty: true })
const identities = new Set<string>()
const distributionItems = projectArray(distributionRecord.items, (value) => {
const item = projectRecord(value)
const identityField = distributionDimension === 'group' ? 'group_id' : 'model'
assertNoSecretLikeFields(item, [...distributionAggregateFields, identityField])
const aggregate = projectUsageDistributionAggregate(
Object.fromEntries(distributionAggregateFields.map((field) => [field, item[field]])),
)
if (distributionDimension === 'group') {
const groupID = projectSafeInteger(item.group_id, { minimum: 1 })
if (identities.has(String(groupID))) invalidResponse()
identities.add(String(groupID))
return { ...aggregate, group_id: groupID }
}
const model = projectString(item.model, { allowEmpty: true })
if (
new TextEncoder().encode(model).length > 255 ||
model !== model.trim() ||
/[\p{Cc}]/u.test(model) ||
identities.has(model)
) {
invalidResponse()
}
identities.add(model)
return { ...aggregate, model }
})
if (distributionItems.length > 5) invalidResponse()
const distributionOther =
distributionRecord.other === null
? null
: projectUsageDistributionAggregate(distributionRecord.other)
const summary = projectUsageAggregate(record.summary)
const visibleAndOther = [
...distributionItems,
...(distributionOther === null ? [] : [distributionOther]),
]
const distributedRequests = visibleAndOther.reduce(
(total, item) => total + item.request_count,
0,
)
const distributedCost = visibleAndOther.reduce(
(total, item) => total + BigInt(item.estimated_cost_nano_usd),
0n,
)
if (
new TextEncoder().encode(model).length > 255 ||
model !== model.trim() ||
/[\p{Cc}]/u.test(model) ||
identities.has(model)
distributedRequests !== summary.request_count ||
distributedCost !== BigInt(summary.estimated_cost_nano_usd)
) {
invalidResponse()
}
identities.add(model)
return { ...aggregate, model }
})
if (distributionItems.length > 5) invalidResponse()
const distributionOther =
distributionRecord.other === null
? null
: projectUsageDistributionAggregate(distributionRecord.other)
const summary = projectUsageAggregate(record.summary)
const visibleAndOther = [
...distributionItems,
...(distributionOther === null ? [] : [distributionOther]),
]
const distributedRequests = visibleAndOther.reduce((total, item) => total + item.request_count, 0)
const distributedCost = visibleAndOther.reduce(
(total, item) => total + BigInt(item.estimated_cost_nano_usd),
0n,
)
if (
distributedRequests !== summary.request_count ||
distributedCost !== BigInt(summary.estimated_cost_nano_usd)
) {
invalidResponse()
return {
dimension: distributionDimension,
metric: distributionMetric,
items: distributionItems,
other: distributionOther,
}
}
function projectDistributionMetricRecord(
value: unknown,
dimension: UsageDistributionDimension,
): Record<UsageDistributionMetric, UsageDistributionDto> {
const metricRecord = projectRecord(value)
assertNoSecretLikeFields(metricRecord, ['requests', 'cost'])
return {
requests: projectDistribution(metricRecord.requests, dimension, 'requests'),
cost: projectDistribution(metricRecord.cost, dimension, 'cost'),
}
}
const modelDistributions = projectDistributionMetricRecord(distributionsRecord.model, 'model')
const groupDistributions = Object.prototype.hasOwnProperty.call(distributionsRecord, 'group')
? projectDistributionMetricRecord(distributionsRecord.group, 'group')
: undefined
const summary = projectUsageAggregate(record.summary)
return {
range,
granularity,
@@ -318,26 +356,16 @@ export function projectUsageReport(value: unknown): UsageReportDto {
observed_at_ms: observedAtMS,
summary,
series,
distribution: {
dimension: distributionDimension,
metric: distributionMetric,
items: distributionItems,
other: distributionOther,
distributions: {
...(groupDistributions === undefined ? {} : { group: groupDistributions }),
model: modelDistributions,
},
collection_health: projectCollectionHealth(record.collection_health),
}
}
export function normalizeUsageFilters(filters: UsageFilters): NormalizedUsageFilters {
const distribution = filters.distribution ?? 'group'
const distributionMetric = filters.distribution_metric ?? 'requests'
if (distribution !== 'group' && distribution !== 'model') invalidResponse()
if (distributionMetric !== 'requests' && distributionMetric !== 'cost') invalidResponse()
const result: NormalizedUsageFilters = {
range: filters.range,
distribution,
distribution_metric: distributionMetric,
}
export function normalizeUsageFilters(filters: UsageFilters): UsageFilters {
const result: UsageFilters = { range: filters.range }
if (filters.group_id !== undefined) result.group_id = filters.group_id
if (filters.channel_id !== undefined) result.channel_id = filters.channel_id
if (filters.credential_id !== undefined) result.credential_id = filters.credential_id
@@ -356,8 +384,6 @@ export async function getUsageReport(
): Promise<UsageReportDto> {
const normalized = normalizeUsageFilters(filters)
const params = new URLSearchParams([['range', normalized.range]])
params.append('distribution', normalized.distribution)
params.append('distribution_metric', normalized.distribution_metric)
if (normalized.group_id !== undefined) params.append('group_id', String(normalized.group_id))
if (normalized.channel_id !== undefined) params.append('channel_id', normalized.channel_id)
if (normalized.credential_id !== undefined) {
@@ -369,12 +395,6 @@ export async function getUsageReport(
const report = projectUsageReport(
await client.request(`/api/usage?${params.toString()}`, { method: 'GET', signal }),
)
if (
report.distribution.dimension !== normalized.distribution ||
report.distribution.metric !== normalized.distribution_metric
) {
invalidResponse()
}
return report
}
@@ -387,6 +407,10 @@ export function usageQueryOptions(
queryKey: computed(() => usageQueryIdentity(toValue(filters))),
queryFn: ({ signal }) => getUsageReport(client, toValue(filters), signal),
placeholderData: keepPreviousData,
staleTime: Number.POSITIVE_INFINITY,
refetchOnMount: false,
refetchOnWindowFocus: false,
refetchOnReconnect: false,
...(intervalMs !== undefined
? {
refetchInterval: intervalMs,
+19 -57
View File
@@ -1,5 +1,5 @@
<script setup lang="ts">
import { Boxes, ChevronRight, Layers3 } from '@lucide/vue'
import { Boxes, Layers3 } from '@lucide/vue'
import { computed } from 'vue'
import { useI18n } from 'vue-i18n'
@@ -8,29 +8,27 @@ import type { ChannelDto } from '@/app/resources/channels'
import type {
UsageAggregateDto,
UsageDistributionAggregateDto,
UsageReportDto,
UsageDistributionDto,
} from '@/app/resources/usage'
import ChannelIcon from '@/components/brand/ChannelIcon.vue'
import OverflowTooltip from '@/components/ui/OverflowTooltip.vue'
import { formatEstimatedCost, formatInteger, formatPercent } from '@/lib/format'
type DistributionItem = UsageReportDto['distribution']['items'][number]
type DistributionItem = UsageDistributionDto['items'][number]
type DistributionRow = {
key: string
item: UsageDistributionAggregateDto
identity?: DistributionItem
rank?: number
selectable: boolean
}
const props = defineProps<{
distribution: UsageReportDto['distribution']
distribution: UsageDistributionDto
summary: UsageAggregateDto
groups: GroupOptionDto[]
channels: ChannelDto[]
}>()
const emit = defineEmits<{ select: [item: DistributionItem] }>()
const { locale, t } = useI18n()
const rows = computed<DistributionRow[]>(() => {
@@ -42,10 +40,9 @@ const rows = computed<DistributionRow[]>(() => {
item,
identity: item,
rank: index + 1,
selectable: props.distribution.dimension === 'group' || item.model !== '',
}))
if (props.distribution.other !== null) {
items.push({ key: 'other', item: props.distribution.other, selectable: false })
items.push({ key: 'other', item: props.distribution.other })
}
return items
})
@@ -69,13 +66,18 @@ function identityLabel(row: DistributionRow): string {
}
function identityMeta(row: DistributionRow): string {
if (row.identity === undefined) return t('monitor.usage.distribution.otherHint')
if (row.identity === undefined) {
return props.distribution.dimension === 'group'
? t('monitor.usage.distribution.otherGroupHint')
: t('monitor.usage.distribution.otherHint')
}
if (props.distribution.dimension === 'model') {
return t('monitor.usage.distribution.modelHint')
}
const groupID = row.identity.group_id ?? 0
const groupChannel = channel(row.identity)
return groupChannel ? `${groupChannel.name} · Group #${groupID}` : `Group #${groupID}`
const groupLabel = t('monitor.usage.distribution.groupValue', { id: groupID })
return groupChannel ? `${groupChannel.name} · ${groupLabel}` : groupLabel
}
function metricValue(item: UsageDistributionAggregateDto): number | bigint {
@@ -125,22 +127,15 @@ function secondaryValue(item: UsageDistributionAggregateDto): string {
})
: formatEstimatedCost(item.estimated_cost_nano_usd, locale.value)
}
function select(row: DistributionRow): void {
if (row.selectable && row.identity !== undefined) emit('select', row.identity)
}
</script>
<template>
<ol class="usage-distribution" :aria-label="t('monitor.usage.distribution.caption')">
<li v-for="row in rows" :key="row.key" class="usage-distribution__item">
<component
:is="row.selectable ? 'button' : 'div'"
<div
class="usage-distribution__row"
:class="{ 'usage-distribution__row--other': row.identity === undefined }"
:type="row.selectable ? 'button' : undefined"
:title="row.selectable ? t('monitor.usage.distribution.filterHint') : identityLabel(row)"
@click="select(row)"
:title="identityLabel(row)"
>
<span class="usage-distribution__rank" aria-hidden="true">
{{ row.rank === undefined ? '∑' : String(row.rank).padStart(2, '0') }}
@@ -178,13 +173,7 @@ function select(row: DistributionRow): void {
</span>
<span class="usage-distribution__share">{{ shareLabel(row.item) }}</span>
<ChevronRight
v-if="row.selectable"
class="usage-distribution__arrow"
:size="16"
aria-hidden="true"
/>
</component>
</div>
</li>
</ol>
</template>
@@ -211,7 +200,7 @@ function select(row: DistributionRow): void {
display: grid;
min-width: 0;
min-height: 62px;
grid-template-columns: 34px minmax(180px, 0.85fr) minmax(160px, 1.25fr) 116px 64px 22px;
grid-template-columns: 34px minmax(180px, 0.85fr) minmax(160px, 1.25fr) 116px 64px;
align-items: center;
gap: 14px;
border: 0;
@@ -222,22 +211,6 @@ function select(row: DistributionRow): void {
font: inherit;
}
button.usage-distribution__row {
cursor: pointer;
transition: background-color var(--duration-fast) var(--easing-standard);
}
button.usage-distribution__row:hover {
background: var(--color-surface-sunken);
}
button.usage-distribution__row:focus-visible {
position: relative;
z-index: 1;
outline: 2px solid var(--color-focus);
outline-offset: -2px;
}
.usage-distribution__item:last-child {
border-bottom: 0;
}
@@ -348,19 +321,9 @@ button.usage-distribution__row:focus-visible {
font-size: var(--text-sm);
}
.usage-distribution__arrow {
color: var(--color-text-faint);
transition: transform var(--duration-fast) var(--easing-standard);
}
button.usage-distribution__row:hover .usage-distribution__arrow {
transform: translateX(2px);
color: var(--color-text-muted);
}
@media (max-width: 820px) {
.usage-distribution__row {
grid-template-columns: 30px minmax(0, 1fr) 104px 58px 18px;
grid-template-columns: 30px minmax(0, 1fr) 104px 58px;
gap: 10px;
padding-inline: 12px;
}
@@ -373,7 +336,7 @@ button.usage-distribution__row:hover .usage-distribution__arrow {
@media (max-width: 560px) {
.usage-distribution__row {
min-height: 68px;
grid-template-columns: 26px minmax(0, 1fr) 86px 18px;
grid-template-columns: 26px minmax(0, 1fr) 86px;
}
.usage-distribution__share {
@@ -387,8 +350,7 @@ button.usage-distribution__row:hover .usage-distribution__arrow {
}
@media (prefers-reduced-motion: reduce) {
.usage-distribution__fill,
.usage-distribution__arrow {
.usage-distribution__fill {
transition: none;
}
}
+40 -42
View File
@@ -69,15 +69,33 @@ const filterOpen = computed(() => routeState.value.filtersOpen)
const draft = ref<UsageFilterDraft>(createUsageFilterDraft(appliedFilters.value))
const filterErrors = ref<UsageFilterErrors>({})
const groupsQuery = useQuery(groupOptionsQueryOptions(client, () => !isAccessKey.value))
const groupsQuery = useQuery({
...groupOptionsQueryOptions(client, () => !isAccessKey.value),
refetchOnMount: false,
})
const channelsQuery = useQuery({
queryKey: controlQueryKeys.channels.list(''),
queryFn: ({ signal }) => listChannels(client, '', signal),
enabled: computed(() => !isAccessKey.value),
staleTime: 5 * 60 * 1_000,
staleTime: Number.POSITIVE_INFINITY,
refetchOnMount: false,
refetchOnWindowFocus: false,
refetchOnReconnect: false,
})
const usageQuery = useQuery(usageQueryOptions(client, appliedFilters))
const report = computed(() => usageQuery.data.value)
const distributionDimension = ref<UsageDistributionDimension>('model')
const distributionMetric = ref<UsageDistributionMetric>('cost')
const distribution = computed(() => {
const distributions = report.value?.distributions
if (distributions === undefined) return undefined
if (isAccessKey.value || distributionDimension.value === 'model') {
return distributions.model[distributionMetric.value]
}
return (
distributions.group?.[distributionMetric.value] ?? distributions.model[distributionMetric.value]
)
})
const {
initial: initialLoading,
transition: reportTransition,
@@ -89,8 +107,7 @@ const {
fetching: () => usageQuery.isFetching.value,
hasData: () => report.value !== undefined,
itemCount: () =>
(report.value?.distribution.items.length ?? 0) +
(report.value?.distribution.other === null ? 0 : 1),
(distribution.value?.items.length ?? 0) + (distribution.value?.other === null ? 0 : 1),
},
{ fallbackRows: 5 },
)
@@ -104,12 +121,12 @@ const usageRefreshing = computed(
)
const hasData = computed(() => (report.value?.summary.request_count ?? 0) > 0)
const distributionDimensionOptions = computed(() => [
{ value: 'group', label: t('monitor.usage.distribution.dimensions.group') },
{ value: 'model', label: t('monitor.usage.distribution.dimensions.model') },
{ value: 'group', label: t('monitor.usage.distribution.dimensions.group') },
])
const distributionMetricOptions = computed(() => [
{ value: 'requests', label: t('monitor.usage.distribution.metrics.requests') },
{ value: 'cost', label: t('monitor.usage.distribution.metrics.cost') },
{ value: 'requests', label: t('monitor.usage.distribution.metrics.requests') },
])
const costChartResolution = 1_000_000_000n
const trendMetricOptions = computed(() => [
@@ -211,8 +228,6 @@ watch(
() => appliedFilters.value.channel_id,
() => appliedFilters.value.credential_id,
() => appliedFilters.value.upstream_model,
() => appliedFilters.value.distribution,
() => appliedFilters.value.distribution_metric,
],
() => {
draft.value = createUsageFilterDraft(appliedFilters.value)
@@ -281,39 +296,19 @@ async function applyFilters(): Promise<void> {
}
async function resetFilters(): Promise<void> {
await navigate({
range: appliedFilters.value.range,
distribution: appliedFilters.value.distribution,
distribution_metric: appliedFilters.value.distribution_metric,
})
await navigate({ range: appliedFilters.value.range })
}
async function updateDistributionDimension(value: string): Promise<void> {
function updateDistributionDimension(value: string): void {
if (value !== 'group' && value !== 'model') return
await navigate(
{ ...appliedFilters.value, distribution: value as UsageDistributionDimension },
routeState.value,
)
}
async function updateDistributionMetric(value: string): Promise<void> {
if (value !== 'requests' && value !== 'cost') return
await navigate(
{ ...appliedFilters.value, distribution_metric: value as UsageDistributionMetric },
routeState.value,
)
}
async function selectDistributionItem(
item: NonNullable<typeof report.value>['distribution']['items'][number],
): Promise<void> {
if (report.value?.distribution.dimension === 'group' && item.group_id !== undefined) {
await navigate({ ...appliedFilters.value, group_id: item.group_id })
if (isAccessKey.value || (value === 'group' && report.value?.distributions.group === undefined))
return
}
if (report.value?.distribution.dimension === 'model' && item.model) {
await navigate({ ...appliedFilters.value, upstream_model: item.model })
}
distributionDimension.value = value as UsageDistributionDimension
}
function updateDistributionMetric(value: string): void {
if (value !== 'requests' && value !== 'cost') return
distributionMetric.value = value as UsageDistributionMetric
}
async function updateTrendMetric(value: string): Promise<void> {
@@ -343,7 +338,10 @@ function setSeriesExpanded(event: Event): void {
}
async function refresh(): Promise<void> {
await usageQuery.refetch()
await Promise.all([
usageQuery.refetch(),
...(!isAccessKey.value ? [groupsQuery.refetch(), channelsQuery.refetch()] : []),
])
}
defineExpose({ openFilters, refresh })
@@ -538,14 +536,14 @@ defineExpose({ openFilters, refresh })
<template #actions>
<SegmentedControl
v-if="!isAccessKey"
:model-value="report.distribution.dimension"
:model-value="distributionDimension"
:label="t('monitor.usage.distribution.dimensionLabel')"
:options="distributionDimensionOptions"
size="compact"
@update:model-value="updateDistributionDimension"
/>
<SegmentedControl
:model-value="report.distribution.metric"
:model-value="distributionMetric"
:label="t('monitor.usage.distribution.metricLabel')"
:options="distributionMetricOptions"
size="compact"
@@ -555,11 +553,11 @@ defineExpose({ openFilters, refresh })
</MonitorSectionHeading>
<UsageDistribution
:distribution="report.distribution"
v-if="distribution"
:distribution="distribution"
:summary="report.summary"
:groups="groupsQuery.data.value ?? []"
:channels="channelsQuery.data.value?.items ?? []"
@select="selectDistributionItem"
/>
</section>
@@ -87,7 +87,6 @@ export function scopeAccessKeyUsageFilters(filters: UsageFilters): UsageFilters
delete scoped.group_id
delete scoped.channel_id
delete scoped.credential_id
scoped.distribution = 'model'
return scoped
}
@@ -140,8 +139,6 @@ export function usageMonitorQuery(
if (channelID !== undefined) normalized.channel_id = channelID
if (credentialID !== undefined) normalized.credential_id = String(credentialID)
if (upstreamModel !== undefined) normalized.upstream_model = upstreamModel
if (filters.distribution === 'model') normalized.distribution = 'model'
if (filters.distribution_metric === 'cost') normalized.distribution_metric = 'cost'
if (state.filtersOpen) normalized.panel = 'filters'
if (state.seriesExpanded) normalized.series = 'expanded'
return normalized
+1 -13
View File
@@ -5,22 +5,16 @@ import { normalizeMonitorText } from './filter-validation'
export interface UsageFilterDraft {
range: UsageFilters['range']
distribution: NonNullable<UsageFilters['distribution']>
distribution_metric: NonNullable<UsageFilters['distribution_metric']>
group_id: string
channel_id: string
credential_id: string
upstream_model: string
}
export type UsageFilterErrors = Partial<
Record<Exclude<keyof UsageFilterDraft, 'range' | 'distribution' | 'distribution_metric'>, string>
>
export type UsageFilterErrors = Partial<Record<Exclude<keyof UsageFilterDraft, 'range'>, string>>
const emptyDraft = (): UsageFilterDraft => ({
range: defaultTimeRange,
distribution: 'group',
distribution_metric: 'requests',
group_id: '',
channel_id: '',
credential_id: '',
@@ -60,8 +54,6 @@ export function parseAppliedUsageFilters(query: Record<string, unknown>): UsageF
if (channelID !== undefined) filters.channel_id = channelID
if (credentialID !== undefined) filters.credential_id = credentialID
if (upstreamModel !== undefined) filters.upstream_model = upstreamModel
if (query.distribution === 'model') filters.distribution = 'model'
if (query.distribution_metric === 'cost') filters.distribution_metric = 'cost'
return filters
}
@@ -69,8 +61,6 @@ export function createUsageFilterDraft(filters: UsageFilters): UsageFilterDraft
return {
...emptyDraft(),
range: filters.range,
distribution: filters.distribution ?? 'group',
distribution_metric: filters.distribution_metric ?? 'requests',
group_id: filters.group_id === undefined ? '' : String(filters.group_id),
channel_id: filters.channel_id ?? '',
credential_id: filters.credential_id === undefined ? '' : String(filters.credential_id),
@@ -81,8 +71,6 @@ export function createUsageFilterDraft(filters: UsageFilters): UsageFilterDraft
export function applyUsageFilterDraft(draft: UsageFilterDraft): UsageFilters {
const filters: UsageFilters = {
range: normalizeUsageRange(draft.range),
distribution: draft.distribution,
distribution_metric: draft.distribution_metric,
}
const groupID = normalizeUsageGroupID(draft.group_id)
const channelID = normalizeUsageChannelID(draft.channel_id)
+2 -1
View File
@@ -340,10 +340,11 @@ export default {
metrics: { requests: 'Requests', cost: 'Estimated cost' },
other: 'Other',
otherHint: 'Combined remainder outside the Top 5',
otherGroupHint: 'Combined remainder, including deleted or unknown Groups',
groupValue: 'Group #{id}',
modelHint: 'Upstream model',
unknownModel: 'Unknown model',
requestsValue: '{value} requests',
filterHint: 'Click to apply this item as a filter',
},
columns: {
window: 'UTC window',
+4 -3
View File
@@ -330,19 +330,20 @@ export default {
},
distribution: {
title: '消費分布',
description: '上流の認証情報を個別展開せず、主要な Group またはモデルを集約表示します。',
description: '上流の認証情報を個別展開せず、主要なグループまたはモデルを集約表示します。',
caption: '現在の絞り込み範囲における消費分布',
limit: '上位 5 件 + その他',
dimensionLabel: '分布軸',
metricLabel: '順位指標',
dimensions: { group: 'Group', model: 'モデル' },
dimensions: { group: 'グループ', model: 'モデル' },
metrics: { requests: 'リクエスト', cost: '推定コスト' },
other: 'その他',
otherHint: '上位 5 件以外の合計',
otherGroupHint: '上位 5 件以外(削除済みまたは不明なグループを含む)の合計',
groupValue: 'グループ #{id}',
modelHint: 'アップストリームモデル',
unknownModel: '不明なモデル',
requestsValue: '{value} リクエスト',
filterHint: 'クリックしてこの項目を絞り込み条件に適用',
},
columns: {
window: 'UTC 時間枠',
+13 -12
View File
@@ -178,13 +178,13 @@ export default {
loadFailed: '无法加载用量报告。',
stale: '后台刷新失败,用量数据可能已过期。',
options: {
groupsFailed: '无法加载 Group 名称。可改为输入正整数 Group ID;用量报告不受影响。',
partialFailed: '部分 Group 或渠道选项暂时无法加载;用量查询仍可正常使用。',
groupsFailed: '无法加载分组名称。可改为输入正整数分组 ID;用量报告不受影响。',
partialFailed: '部分分组或渠道选项暂时无法加载;用量查询仍可正常使用。',
},
filters: {
label: '用量报告筛选',
title: '筛选用量与成本',
description: '按 Group、渠道、凭据与上游模型缩小当前统计范围。',
description: '按分组、渠道、凭据与上游模型缩小当前统计范围。',
close: '关闭用量与成本筛选',
button: '筛选',
range: '时间范围',
@@ -198,10 +198,10 @@ export default {
'15d': '15 天',
'30d': '30 天',
},
group: 'Group',
groupHelp: '可选,精确限定 Group。',
groupIdHelp: 'Group 名称加载失败;可输入正整数 Group ID。',
anyGroup: '全部 Group',
group: '分组',
groupHelp: '可选,精确限定分组。',
groupIdHelp: '分组名称加载失败;可输入正整数分组 ID。',
anyGroup: '全部分组',
channel: '渠道',
anyChannel: '全部渠道',
credential: '凭据 ID',
@@ -219,7 +219,7 @@ export default {
refreshedAt: '最近刷新时间',
},
errors: {
positiveId: '请输入有效的正整数 Group ID。',
positiveId: '请输入有效的正整数分组 ID。',
channelId: '请选择或输入有效的渠道 ID。',
credentialId: '请输入有效的正整数凭据 ID。',
model: '上游模型不能包含首尾空白或控制字符。',
@@ -231,7 +231,7 @@ export default {
},
empty: {
title: '没有匹配的用量数据',
description: '请调整时间范围,或重置 Group、渠道、凭据与上游模型筛选。',
description: '请调整时间范围,或重置分组、渠道、凭据与上游模型筛选。',
},
kpi: {
title: '用量摘要',
@@ -315,19 +315,20 @@ export default {
},
distribution: {
title: '消耗分布',
description: '聚合展示主要 Group 或模型,不再按上游凭据逐条展开。',
description: '聚合展示主要分组或模型,不再按上游凭据逐条展开。',
caption: '当前筛选范围内的消耗分布',
limit: 'Top 5 + 其他',
dimensionLabel: '分布维度',
metricLabel: '排序指标',
dimensions: { group: 'Group', model: '模型' },
dimensions: { group: '分组', model: '模型' },
metrics: { requests: '请求数', cost: '预估成本' },
other: '其他',
otherHint: '未进入 Top 5 的合计',
otherGroupHint: '未进入 Top 5、已删除或未知分组的合计',
groupValue: '分组 #{id}',
modelHint: '上游模型',
unknownModel: '未知模型',
requestsValue: '{value} 次请求',
filterHint: '点击应用此项为筛选条件',
},
columns: {
window: 'UTC 时间窗',