diff options
Diffstat (limited to '')
| -rw-r--r-- | internal/controlplane/usage_analytics.go | 61 |
1 files changed, 56 insertions, 5 deletions
diff --git a/internal/controlplane/usage_analytics.go b/internal/controlplane/usage_analytics.go index 7cc042f..1bc6228 100644 --- a/internal/controlplane/usage_analytics.go +++ b/internal/controlplane/usage_analytics.go @@ -22,7 +22,13 @@ func (s *Store) UsageAnalytics(ctx context.Context, query UsageQuery) (UsageAnal if !to.After(from) { return UsageAnalytics{}, fmt.Errorf("usage analytics range must be positive") } - result := UsageAnalytics{RangeStart: from, RangeEnd: to, Models: make([]UsageModelAnalytics, 0), Providers: make([]UsageProviderAnalytics, 0)} + result := UsageAnalytics{ + RangeStart: from, + RangeEnd: to, + Models: make([]UsageModelAnalytics, 0), + Providers: make([]UsageProviderAnalytics, 0), + Keys: make([]UsageKeyAnalytics, 0), + } modelPrevious, err := s.usageModelCharges(ctx, query, from.Add(-to.Sub(from)), from) if err != nil { @@ -49,8 +55,13 @@ func (s *Store) UsageAnalytics(ctx context.Context, query UsageQuery) (UsageAnal providers[index].PreviousChargedMicros = providerPrevious[providers[index].ProviderID] providers[index].ChargeChangePercent = chargeChange(providers[index].ChargedMicros, providers[index].PreviousChargedMicros) } + keys, err := s.usageKeyAnalytics(ctx, query, from, to) + if err != nil { + return UsageAnalytics{}, err + } result.Models = models result.Providers = providers + result.Keys = keys return result, nil } @@ -128,7 +139,11 @@ func (s *Store) usageModelAnalytics(ctx context.Context, query UsageQuery, from, count(DISTINCT NULLIF(e.provider_id,'')), COALESCE(sum(e.input_tokens),0), COALESCE(sum(e.output_tokens),0), COALESCE(sum(e.total_tokens),0), COALESCE(sum(e.cache_read_input_tokens),0), COALESCE(sum(e.cache_creation_input_tokens),0), COALESCE(sum(e.charged_micros),0), COALESCE(sum(e.uncollected_micros),0), count(*) FILTER (WHERE e.metering_status='missing'), - COALESCE(round(avg(e.duration_ms)),0)::bigint, COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint + COALESCE(round(avg(e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.50) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.50) WITHIN GROUP (ORDER BY e.ttft_ms) FILTER (WHERE e.ttft_ms > 0)),0)::bigint, + COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.ttft_ms) FILTER (WHERE e.ttft_ms > 0)),0)::bigint FROM usage_events e WHERE `+where+` GROUP BY e.public_model ORDER BY sum(e.charged_micros) DESC, e.public_model`, args...) if err != nil { return nil, fmt.Errorf("query usage model analytics: %w", err) @@ -139,7 +154,8 @@ func (s *Store) usageModelAnalytics(ctx context.Context, query UsageQuery, from, var item UsageModelAnalytics if err := rows.Scan(&item.PublicModel, &item.RequestCount, &item.SuccessfulRequests, &item.ErrorCount, &item.ProviderCount, &item.InputTokens, &item.OutputTokens, &item.TotalTokens, &item.CacheReadInputTokens, &item.CacheCreationInputTokens, - &item.ChargedMicros, &item.UncollectedMicros, &item.MissingUsageRequests, &item.AverageDurationMS, &item.P95DurationMS); err != nil { + &item.ChargedMicros, &item.UncollectedMicros, &item.MissingUsageRequests, &item.AverageDurationMS, + &item.P50DurationMS, &item.P95DurationMS, &item.P50TTFTMS, &item.P95TTFTMS); err != nil { return nil, fmt.Errorf("scan usage model analytics: %w", err) } result = append(result, item) @@ -174,7 +190,10 @@ func (s *Store) usageProviderAnalytics(ctx context.Context, query UsageQuery, fr COALESCE(sum(e.input_tokens),0), COALESCE(sum(e.output_tokens),0), COALESCE(sum(e.total_tokens),0), COALESCE(sum(e.cache_read_input_tokens),0), COALESCE(sum(e.cache_creation_input_tokens),0), COALESCE(sum(e.charged_micros),0), COALESCE(sum(e.uncollected_micros),0), count(*) FILTER (WHERE e.metering_status='missing'), COALESCE(round(avg(e.duration_ms)),0)::bigint, - COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint + COALESCE(round(percentile_cont(0.50) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.50) WITHIN GROUP (ORDER BY e.ttft_ms) FILTER (WHERE e.ttft_ms > 0)),0)::bigint, + COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.ttft_ms) FILTER (WHERE e.ttft_ms > 0)),0)::bigint FROM usage_events e LEFT JOIN providers p ON p.id::text=e.provider_id WHERE `+where+` GROUP BY e.provider_id, p.name, p.wire_api ORDER BY sum(e.charged_micros) DESC, COALESCE(NULLIF(p.name,''),'Unassigned')`, args...) if err != nil { @@ -186,7 +205,8 @@ func (s *Store) usageProviderAnalytics(ctx context.Context, query UsageQuery, fr var item UsageProviderAnalytics if err := rows.Scan(&item.ProviderID, &item.ProviderName, &item.WireAPI, &item.RequestCount, &item.SuccessfulRequests, &item.ErrorCount, &item.ModelCount, &item.InputTokens, &item.OutputTokens, &item.TotalTokens, &item.CacheReadInputTokens, &item.CacheCreationInputTokens, - &item.ChargedMicros, &item.UncollectedMicros, &item.MissingUsageRequests, &item.AverageDurationMS, &item.P95DurationMS); err != nil { + &item.ChargedMicros, &item.UncollectedMicros, &item.MissingUsageRequests, &item.AverageDurationMS, + &item.P50DurationMS, &item.P95DurationMS, &item.P50TTFTMS, &item.P95TTFTMS); err != nil { return nil, fmt.Errorf("scan usage provider analytics: %w", err) } result = append(result, item) @@ -194,6 +214,37 @@ func (s *Store) usageProviderAnalytics(ctx context.Context, query UsageQuery, fr return result, rows.Err() } +func (s *Store) usageKeyAnalytics(ctx context.Context, query UsageQuery, from, to time.Time) ([]UsageKeyAnalytics, error) { + where, args := analyticsUsageWhere(query, from, to) + rows, err := s.db.Query(ctx, `SELECT e.key_id::text, COALESCE(NULLIF(k.name,''),'Deleted key'), + count(*), count(*) FILTER (WHERE e.success), count(*) FILTER (WHERE NOT e.success), count(DISTINCT e.public_model), + COALESCE(sum(e.input_tokens),0), COALESCE(sum(e.output_tokens),0), COALESCE(sum(e.total_tokens),0), + COALESCE(sum(e.charged_micros),0), COALESCE(sum(e.uncollected_micros),0), count(*) FILTER (WHERE e.metering_status='missing'), + COALESCE(round(avg(e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.50) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.duration_ms)),0)::bigint, + COALESCE(round(percentile_cont(0.50) WITHIN GROUP (ORDER BY e.ttft_ms) FILTER (WHERE e.ttft_ms > 0)),0)::bigint, + COALESCE(round(percentile_cont(0.95) WITHIN GROUP (ORDER BY e.ttft_ms) FILTER (WHERE e.ttft_ms > 0)),0)::bigint + FROM usage_events e LEFT JOIN api_keys k ON k.id=e.key_id WHERE `+where+` + GROUP BY e.key_id, k.name ORDER BY sum(e.charged_micros) DESC, COALESCE(NULLIF(k.name,''),'Deleted key')`, args...) + if err != nil { + return nil, fmt.Errorf("query usage API key analytics: %w", err) + } + defer rows.Close() + result := make([]UsageKeyAnalytics, 0) + for rows.Next() { + var item UsageKeyAnalytics + if err := rows.Scan(&item.KeyID, &item.KeyName, &item.RequestCount, &item.SuccessfulRequests, &item.ErrorCount, + &item.ModelCount, &item.InputTokens, &item.OutputTokens, &item.TotalTokens, &item.ChargedMicros, + &item.UncollectedMicros, &item.MissingUsageRequests, &item.AverageDurationMS, &item.P50DurationMS, + &item.P95DurationMS, &item.P50TTFTMS, &item.P95TTFTMS); err != nil { + return nil, fmt.Errorf("scan usage API key analytics: %w", err) + } + result = append(result, item) + } + return result, rows.Err() +} + func (s *Store) usageProviderCharges(ctx context.Context, query UsageQuery, from, to time.Time) (map[string]int64, error) { where, args := analyticsUsageWhere(query, from, to) rows, err := s.db.Query(ctx, `SELECT COALESCE(e.provider_id,''), COALESCE(sum(e.charged_micros),0) |