mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-29 08:11:32 +00:00
feat(usage): 用量记录落库上游请求标识并支持按账户配置响应头名
- usage_logs 新增 upstream_request_id 列(迁移 232)与仅索引非空行的 CONCURRENTLY 部分索引(迁移 233,runner 登记失效索引清理) - 账户 extra.upstream_request_id_header 指定要读取的上游响应头,只读该头;留空不记录;创建/更新/批量入口校验头名合法且不超过 64 字节 - ForwardResult/OpenAIForwardResult 携带 UpstreamHeaders,中央计费按账户解析落库,WS 轮次保持 NULL - 管理端用量明细新增默认隐藏的"上游ID"列,带与请求ID列一致的复制按钮,并同步导出 - 创建/编辑账户弹窗新增"上游ID"输入(UpstreamRequestIdHeaderField),提示按接入类型与平台分条列出常见头名 - HelpTooltip 悬停模式下指针移入提示框时保持打开,便于选中复制提示内容
This commit is contained in:
@@ -837,6 +837,10 @@ func (h *AccountHandler) Create(c *gin.Context) {
|
||||
}
|
||||
// base_rpm 输入校验:负值归零,超过 10000 截断
|
||||
sanitizeExtraBaseRPM(req.Extra)
|
||||
if err := service.ValidateUpstreamRequestIDHeaderExtra(req.Extra); err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 确定是否跳过混合渠道检查
|
||||
skipCheck := req.ConfirmMixedChannelRisk != nil && *req.ConfirmMixedChannelRisk
|
||||
@@ -970,6 +974,10 @@ func (h *AccountHandler) Update(c *gin.Context) {
|
||||
}
|
||||
// base_rpm 输入校验:负值归零,超过 10000 截断
|
||||
sanitizeExtraBaseRPM(req.Extra)
|
||||
if err := service.ValidateUpstreamRequestIDHeaderExtra(req.Extra); err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 确定是否跳过混合渠道检查
|
||||
skipCheck := req.ConfirmMixedChannelRisk != nil && *req.ConfirmMixedChannelRisk
|
||||
@@ -1424,6 +1432,10 @@ func (h *AccountHandler) ApplyOAuthCredentials(c *gin.Context) {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
if err := service.ValidateUpstreamRequestIDHeaderExtra(req.Extra); err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
// Drop SSO/password residue; re-auth must leave only OAuth tokens on disk.
|
||||
req.Credentials = service.SanitizeStoredCredentials(existing.Platform, req.Credentials)
|
||||
@@ -1897,6 +1909,15 @@ func (h *AccountHandler) BatchCreate(c *gin.Context) {
|
||||
|
||||
// base_rpm 输入校验:负值归零,超过 10000 截断
|
||||
sanitizeExtraBaseRPM(item.Extra)
|
||||
if err := service.ValidateUpstreamRequestIDHeaderExtra(item.Extra); err != nil {
|
||||
failed++
|
||||
results = append(results, gin.H{
|
||||
"name": item.Name,
|
||||
"success": false,
|
||||
"error": err.Error(),
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
skipCheck := item.ConfirmMixedChannelRisk != nil && *item.ConfirmMixedChannelRisk
|
||||
|
||||
@@ -2091,6 +2112,10 @@ func (h *AccountHandler) BulkUpdate(c *gin.Context) {
|
||||
}
|
||||
// base_rpm 输入校验:负值归零,超过 10000 截断
|
||||
sanitizeExtraBaseRPM(req.Extra)
|
||||
if err := service.ValidateUpstreamRequestIDHeaderExtra(req.Extra); err != nil {
|
||||
response.ErrorFrom(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 确定是否跳过混合渠道检查
|
||||
skipCheck := req.ConfirmMixedChannelRisk != nil && *req.ConfirmMixedChannelRisk
|
||||
|
||||
@@ -723,6 +723,7 @@ func UsageLogFromServiceAdmin(l *service.UsageLog) *AdminUsageLog {
|
||||
UpstreamModelMismatch: l.UpstreamModelMismatch,
|
||||
ChannelID: l.ChannelID,
|
||||
ModelMappingChain: l.ModelMappingChain,
|
||||
UpstreamRequestID: l.UpstreamRequestID,
|
||||
BillingTier: l.BillingTier,
|
||||
AccountRateMultiplier: l.AccountRateMultiplier,
|
||||
AccountStatsCost: l.AccountStatsCost,
|
||||
|
||||
@@ -588,6 +588,8 @@ type AdminUsageLog struct {
|
||||
ChannelID *int64 `json:"channel_id,omitempty"`
|
||||
// ModelMappingChain 模型映射链,如 "a→b→c"
|
||||
ModelMappingChain *string `json:"model_mapping_chain,omitempty"`
|
||||
// UpstreamRequestID 是直接上游声明的请求标识,仅管理端可见。
|
||||
UpstreamRequestID *string `json:"upstream_request_id,omitempty"`
|
||||
// BillingTier 计费层级标签(per_request/image 模式)
|
||||
BillingTier *string `json:"billing_tier,omitempty"`
|
||||
|
||||
|
||||
@@ -62,6 +62,8 @@ const usageLogsUpstreamModelMismatchIndex = "idx_usage_logs_upstream_model_misma
|
||||
const usageLogsEffectiveModelIndexesMigration = "226_add_usage_log_effective_model_indexes_notx.sql"
|
||||
const usageLogsEffectiveRequestedModelIndex = "idx_usage_logs_effective_requested_model_created"
|
||||
const usageLogsEffectiveUpstreamModelIndex = "idx_usage_logs_effective_upstream_model_created"
|
||||
const usageLogsUpstreamRequestIDIndexMigration = "233_add_usage_log_upstream_request_id_index_notx.sql"
|
||||
const usageLogsUpstreamRequestIDIndex = "idx_usage_logs_upstream_request_id"
|
||||
|
||||
type migrationChecksumCompatibilityRule struct {
|
||||
fileChecksum string
|
||||
@@ -305,6 +307,8 @@ func prepareNonTransactionalMigration(ctx context.Context, db migrationConnectio
|
||||
}
|
||||
}
|
||||
return nil
|
||||
case usageLogsUpstreamRequestIDIndexMigration:
|
||||
return dropInvalidIndexIfPresent(ctx, db, usageLogsUpstreamRequestIDIndex)
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -82,6 +82,7 @@ var usageLogInsertArgTypes = [...]string{
|
||||
"text", // billing_tier
|
||||
"text", // billing_mode
|
||||
"numeric", // account_stats_cost
|
||||
"text", // upstream_request_id
|
||||
"text", // session_id
|
||||
"boolean", // native_compaction_v2
|
||||
"timestamptz", // created_at
|
||||
@@ -282,6 +283,7 @@ func (r *usageLogRepository) createSingle(ctx context.Context, sqlq sqlExecutor,
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -291,7 +293,7 @@ func (r *usageLogRepository) createSingle(ctx context.Context, sqlq sqlExecutor,
|
||||
$12, $13, $14, $15,
|
||||
$16, $17, $18, $19,
|
||||
$20, $21, $22, $23, $24, $25,
|
||||
$26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60, $61
|
||||
$26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60, $61, $62
|
||||
)
|
||||
ON CONFLICT (request_id, api_key_id) DO NOTHING
|
||||
RETURNING id, created_at
|
||||
@@ -741,6 +743,7 @@ func buildUsageLogBatchInsertQuery(keys []string, preparedByKey map[string]usage
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -835,6 +838,7 @@ func buildUsageLogBatchInsertQuery(keys []string, preparedByKey map[string]usage
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -898,6 +902,7 @@ func buildUsageLogBatchInsertQuery(keys []string, preparedByKey map[string]usage
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -1001,6 +1006,7 @@ func buildUsageLogBestEffortInsertQuery(preparedList []usageLogInsertPrepared) (
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -1090,6 +1096,7 @@ func buildUsageLogBestEffortInsertQuery(preparedList []usageLogInsertPrepared) (
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -1153,6 +1160,7 @@ func buildUsageLogBestEffortInsertQuery(preparedList []usageLogInsertPrepared) (
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -1224,6 +1232,7 @@ func execUsageLogInsertNoResult(ctx context.Context, sqlq sqlExecutor, prepared
|
||||
billing_tier,
|
||||
billing_mode,
|
||||
account_stats_cost,
|
||||
upstream_request_id,
|
||||
session_id,
|
||||
native_compaction_v2,
|
||||
created_at
|
||||
@@ -1233,7 +1242,7 @@ func execUsageLogInsertNoResult(ctx context.Context, sqlq sqlExecutor, prepared
|
||||
$12, $13, $14, $15,
|
||||
$16, $17, $18, $19,
|
||||
$20, $21, $22, $23, $24, $25,
|
||||
$26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60, $61
|
||||
$26, $27, $28, $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42, $43, $44, $45, $46, $47, $48, $49, $50, $51, $52, $53, $54, $55, $56, $57, $58, $59, $60, $61, $62
|
||||
)
|
||||
ON CONFLICT (request_id, api_key_id) DO NOTHING
|
||||
`, prepared.args...)
|
||||
@@ -1275,6 +1284,7 @@ func prepareUsageLogInsert(log *service.UsageLog) usageLogInsertPrepared {
|
||||
modelMappingChain := nullString(log.ModelMappingChain)
|
||||
billingTier := nullString(log.BillingTier)
|
||||
billingMode := nullString(log.BillingMode)
|
||||
upstreamRequestID := nullString(log.UpstreamRequestID)
|
||||
sessionID := nullString(log.SessionID)
|
||||
requestedModel := strings.TrimSpace(log.RequestedModel)
|
||||
if requestedModel == "" {
|
||||
@@ -1353,6 +1363,7 @@ func prepareUsageLogInsert(log *service.UsageLog) usageLogInsertPrepared {
|
||||
billingTier,
|
||||
billingMode,
|
||||
log.AccountStatsCost, // account_stats_cost
|
||||
upstreamRequestID, // upstream_request_id
|
||||
sessionID, // session_id
|
||||
log.NativeCompactionV2,
|
||||
createdAt,
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
//go:build unit
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
var (
|
||||
usageLogStaticInsertShapeRe = regexp.MustCompile(`(?s)INSERT INTO usage_logs \((.*?)\) VALUES \((.*?)\)`)
|
||||
usageLogPlaceholderRe = regexp.MustCompile(`\$(\d+)`)
|
||||
)
|
||||
|
||||
// newSQLCapturingMock 返回把实际下发 SQL 记录到 captured 的 sqlmock;语句一律视为匹配,
|
||||
// 参数仍由 WithArgs 校验。
|
||||
func newSQLCapturingMock(t *testing.T, captured *[]string) (*sql.DB, sqlmock.Sqlmock) {
|
||||
t.Helper()
|
||||
matcher := sqlmock.QueryMatcherFunc(func(_, actualSQL string) error {
|
||||
*captured = append(*captured, actualSQL)
|
||||
return nil
|
||||
})
|
||||
db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
return db, mock
|
||||
}
|
||||
|
||||
// requireStaticInsertMatchesArgTypes 断言手写的 INSERT:列清单长度与 VALUES 占位符数量
|
||||
// 都等于 usageLogInsertArgTypes,且占位符恰为 $1..$N 各出现一次。
|
||||
func requireStaticInsertMatchesArgTypes(t *testing.T, query string) {
|
||||
t.Helper()
|
||||
m := usageLogStaticInsertShapeRe.FindStringSubmatch(query)
|
||||
require.Len(t, m, 3, "unrecognised INSERT shape:\n%s", query)
|
||||
|
||||
want := len(usageLogInsertArgTypes)
|
||||
columns := 0
|
||||
for _, col := range strings.Split(m[1], ",") {
|
||||
if strings.TrimSpace(col) != "" {
|
||||
columns++
|
||||
}
|
||||
}
|
||||
require.Equal(t, want, columns, "INSERT column list must match usageLogInsertArgTypes")
|
||||
|
||||
seen := make(map[int]struct{}, want)
|
||||
for _, ph := range usageLogPlaceholderRe.FindAllStringSubmatch(m[2], -1) {
|
||||
n, err := strconv.Atoi(ph[1])
|
||||
require.NoError(t, err)
|
||||
_, dup := seen[n]
|
||||
require.False(t, dup, "duplicate placeholder $%d", n)
|
||||
seen[n] = struct{}{}
|
||||
}
|
||||
require.Len(t, seen, want, "VALUES placeholder count must match usageLogInsertArgTypes")
|
||||
for i := 1; i <= want; i++ {
|
||||
_, ok := seen[i]
|
||||
require.True(t, ok, "missing placeholder $%d", i)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUsageLogStaticInsertShape_PlaceholdersMatchArgTypes 覆盖两条不经占位符生成器、
|
||||
// 直接手写 $1..$N 的 INSERT 路径,防止加列后漏补占位符只在集成测试才暴露。
|
||||
func TestUsageLogStaticInsertShape_PlaceholdersMatchArgTypes(t *testing.T) {
|
||||
upstreamRequestID := "20260902080329-oneapi"
|
||||
log := &service.UsageLog{
|
||||
UserID: 1,
|
||||
APIKeyID: 2,
|
||||
AccountID: 3,
|
||||
RequestID: "client:insert-shape",
|
||||
UpstreamRequestID: &upstreamRequestID,
|
||||
Model: "claude-3",
|
||||
InputTokens: 10,
|
||||
CreatedAt: time.Date(2026, 9, 2, 12, 0, 0, 0, time.UTC),
|
||||
}
|
||||
prepared := prepareUsageLogInsert(log)
|
||||
args := anySliceToDriverValues(prepared.args)
|
||||
|
||||
t.Run("createSingle", func(t *testing.T) {
|
||||
var captured []string
|
||||
db, mock := newSQLCapturingMock(t, &captured)
|
||||
repo := &usageLogRepository{sql: db}
|
||||
|
||||
mock.ExpectQuery("INSERT INTO usage_logs").
|
||||
WithArgs(args...).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "created_at"}).AddRow(int64(1), log.CreatedAt))
|
||||
|
||||
inserted, err := repo.Create(context.Background(), log)
|
||||
require.NoError(t, err)
|
||||
require.True(t, inserted)
|
||||
require.NoError(t, mock.ExpectationsWereMet())
|
||||
require.Len(t, captured, 1)
|
||||
requireStaticInsertMatchesArgTypes(t, captured[0])
|
||||
})
|
||||
|
||||
t.Run("execUsageLogInsertNoResult", func(t *testing.T) {
|
||||
var captured []string
|
||||
db, mock := newSQLCapturingMock(t, &captured)
|
||||
|
||||
mock.ExpectExec("INSERT INTO usage_logs").
|
||||
WithArgs(args...).
|
||||
WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
|
||||
require.NoError(t, execUsageLogInsertNoResult(context.Background(), db, prepared))
|
||||
require.NoError(t, mock.ExpectationsWereMet())
|
||||
require.Len(t, captured, 1)
|
||||
requireStaticInsertMatchesArgTypes(t, captured[0])
|
||||
})
|
||||
}
|
||||
|
||||
// TestPrepareUsageLogInsert_UpstreamRequestIDArgWiring 把 upstream_request_id 钉在
|
||||
// session_id 之前,与参数类型表保持同位;缺失时落 NULL 而不是空串。
|
||||
func TestPrepareUsageLogInsert_UpstreamRequestIDArgWiring(t *testing.T) {
|
||||
upstreamRequestID := "req_upstream_123"
|
||||
prepared := prepareUsageLogInsert(&service.UsageLog{
|
||||
UserID: 1,
|
||||
APIKeyID: 2,
|
||||
RequestID: "client:wiring",
|
||||
Model: "gpt-5",
|
||||
UpstreamRequestID: &upstreamRequestID,
|
||||
CreatedAt: time.Now().UTC(),
|
||||
})
|
||||
require.Len(t, prepared.args, len(usageLogInsertArgTypes))
|
||||
|
||||
idx := len(prepared.args) - 4
|
||||
arg, ok := prepared.args[idx].(sql.NullString)
|
||||
require.True(t, ok, "upstream_request_id arg should be sql.NullString, got %T", prepared.args[idx])
|
||||
require.True(t, arg.Valid)
|
||||
require.Equal(t, upstreamRequestID, arg.String)
|
||||
require.Equal(t, "text", usageLogInsertArgTypes[idx])
|
||||
|
||||
absent := prepareUsageLogInsert(&service.UsageLog{UserID: 1, APIKeyID: 2, RequestID: "client:absent", Model: "gpt-5", CreatedAt: time.Now().UTC()})
|
||||
nullArg, ok := absent.args[idx].(sql.NullString)
|
||||
require.True(t, ok)
|
||||
require.False(t, nullArg.Valid, "absent upstream request id must be NULL")
|
||||
|
||||
require.Contains(t, usageLogSelectColumns, "upstream_request_id")
|
||||
}
|
||||
@@ -19,7 +19,7 @@ import (
|
||||
"github.com/Wei-Shaw/sub2api/internal/service"
|
||||
)
|
||||
|
||||
const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, upstream_response_model, upstream_model_mismatch, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, image_input_tokens, image_input_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, video_count, video_resolution, video_duration_seconds, service_tier, reasoning_effort, requested_reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, long_context_billing_applied, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, session_id, native_compaction_v2, created_at"
|
||||
const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, upstream_response_model, upstream_model_mismatch, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, image_input_tokens, image_input_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, video_count, video_resolution, video_duration_seconds, service_tier, reasoning_effort, requested_reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, long_context_billing_applied, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, upstream_request_id, session_id, native_compaction_v2, created_at"
|
||||
|
||||
func (r *usageLogRepository) GetByID(ctx context.Context, id int64) (log *service.UsageLog, err error) {
|
||||
query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE id = $1"
|
||||
@@ -499,6 +499,7 @@ func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, e
|
||||
billingTier sql.NullString
|
||||
billingMode sql.NullString
|
||||
accountStatsCost sql.NullFloat64
|
||||
upstreamRequestID sql.NullString
|
||||
sessionID sql.NullString
|
||||
nativeCompactionV2 bool
|
||||
createdAt time.Time
|
||||
@@ -564,6 +565,7 @@ func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, e
|
||||
&billingTier,
|
||||
&billingMode,
|
||||
&accountStatsCost,
|
||||
&upstreamRequestID,
|
||||
&sessionID,
|
||||
&nativeCompactionV2,
|
||||
&createdAt,
|
||||
@@ -700,6 +702,9 @@ func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, e
|
||||
if sessionID.Valid {
|
||||
log.SessionID = &sessionID.String
|
||||
}
|
||||
if upstreamRequestID.Valid {
|
||||
log.UpstreamRequestID = &upstreamRequestID.String
|
||||
}
|
||||
|
||||
return log, nil
|
||||
}
|
||||
|
||||
@@ -99,6 +99,7 @@ func TestUsageLogRepositoryCreateSyncRequestTypeAndLegacyFields(t *testing.T) {
|
||||
sqlmock.AnyArg(), // billing_tier
|
||||
sqlmock.AnyArg(), // billing_mode
|
||||
sqlmock.AnyArg(), // account_stats_cost
|
||||
sqlmock.AnyArg(), // upstream_request_id
|
||||
sqlmock.AnyArg(), // session_id
|
||||
log.NativeCompactionV2,
|
||||
createdAt,
|
||||
@@ -193,6 +194,7 @@ func TestUsageLogRepositoryCreate_PersistsServiceTier(t *testing.T) {
|
||||
sqlmock.AnyArg(), // billing_tier
|
||||
sqlmock.AnyArg(), // billing_mode
|
||||
sqlmock.AnyArg(), // account_stats_cost
|
||||
sqlmock.AnyArg(), // upstream_request_id
|
||||
sqlmock.AnyArg(), // session_id
|
||||
log.NativeCompactionV2,
|
||||
createdAt,
|
||||
@@ -955,6 +957,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
|
||||
sql.NullString{},
|
||||
sql.NullString{},
|
||||
sql.NullFloat64{},
|
||||
sql.NullString{}, // upstream_request_id
|
||||
sql.NullString{},
|
||||
false, // native_compaction_v2
|
||||
now,
|
||||
@@ -1034,6 +1037,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
|
||||
sql.NullString{}, // billing_tier
|
||||
sql.NullString{}, // billing_mode
|
||||
sql.NullFloat64{}, // account_stats_cost
|
||||
sql.NullString{}, // upstream_request_id
|
||||
sql.NullString{}, // session_id
|
||||
false, // native_compaction_v2
|
||||
now,
|
||||
@@ -1096,6 +1100,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
|
||||
sql.NullString{}, // billing_tier
|
||||
sql.NullString{}, // billing_mode
|
||||
sql.NullFloat64{}, // account_stats_cost
|
||||
sql.NullString{}, // upstream_request_id
|
||||
sql.NullString{}, // session_id
|
||||
true, // native_compaction_v2
|
||||
now,
|
||||
@@ -1159,6 +1164,7 @@ func TestScanUsageLogRequestTypeAndLegacyFallback(t *testing.T) {
|
||||
sql.NullString{}, // billing_tier
|
||||
sql.NullString{}, // billing_mode
|
||||
sql.NullFloat64{}, // account_stats_cost
|
||||
sql.NullString{}, // upstream_request_id
|
||||
sql.NullString{}, // session_id
|
||||
false, // native_compaction_v2
|
||||
now,
|
||||
|
||||
@@ -450,6 +450,7 @@ func (s *AntigravityGatewayService) Forward(ctx context.Context, c *gin.Context,
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: billingModel,
|
||||
|
||||
@@ -364,6 +364,7 @@ func (s *AntigravityGatewayService) consumeAntigravityCompatResponse(
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *streamResult.usage,
|
||||
Model: call.request.originalModel,
|
||||
UpstreamModel: call.billingModel,
|
||||
|
||||
@@ -434,6 +434,7 @@ handleSuccess:
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: forwardedModel,
|
||||
|
||||
@@ -272,6 +272,7 @@ func (s *GatewayService) forwardAnthropicAPIKeyPassthroughWithInput(
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: input.OriginalModel,
|
||||
UpstreamModel: input.RequestModel,
|
||||
|
||||
@@ -159,6 +159,7 @@ func (s *GatewayService) forwardBedrock(
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: resp.Header.Get("x-amzn-requestid"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: reqModel,
|
||||
UpstreamModel: mappedModel,
|
||||
|
||||
@@ -862,6 +862,7 @@ func (s *GatewayService) Forward(ctx context.Context, c *gin.Context, account *A
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: originalModel, // 使用原始模型用于计费和日志
|
||||
UpstreamModel: mappedModel,
|
||||
|
||||
@@ -337,6 +337,7 @@ func (s *GatewayService) handleCCBufferedFromAnthropic(
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
@@ -389,6 +390,7 @@ func (s *GatewayService) handleCCStreamingFromAnthropic(
|
||||
resultWithUsage := func() *ForwardResult {
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
|
||||
@@ -466,6 +466,7 @@ func (s *GatewayService) handleResponsesBufferedStreamingResponse(
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
@@ -514,6 +515,7 @@ func (s *GatewayService) handleResponsesStreamingResponse(
|
||||
resultWithUsage := func() *ForwardResult {
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
|
||||
@@ -610,8 +610,10 @@ type AudioUsage struct {
|
||||
|
||||
type ForwardResult struct {
|
||||
RequestID string
|
||||
Usage ClaudeUsage
|
||||
Model string
|
||||
// UpstreamHeaders 是直接上游的响应头,用于按账户配置解析上游请求标识。
|
||||
UpstreamHeaders http.Header
|
||||
Usage ClaudeUsage
|
||||
Model string
|
||||
// UpstreamModel is the actual upstream model after mapping.
|
||||
// Prefer empty when it is identical to Model; persistence normalizes equal values away as no-op mappings.
|
||||
UpstreamModel string
|
||||
|
||||
@@ -675,6 +675,7 @@ func partialStreamUsageResult(c *gin.Context, resp *http.Response, streamResult
|
||||
}
|
||||
return &ForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *streamResult.usage,
|
||||
Model: model,
|
||||
UpstreamModel: upstreamModel,
|
||||
|
||||
@@ -1143,6 +1143,7 @@ func (s *GatewayService) buildRecordUsageLog(
|
||||
APIKeyID: apiKey.ID,
|
||||
AccountID: account.ID,
|
||||
RequestID: requestID,
|
||||
UpstreamRequestID: usageUpstreamRequestIDPtr(account, result.UpstreamHeaders, false),
|
||||
Model: result.Model,
|
||||
RequestedModel: requestedModel,
|
||||
UpstreamModel: optionalTrimmedStringPtr(result.UpstreamModel),
|
||||
|
||||
@@ -290,6 +290,7 @@ func (s *GeminiMessagesCompatService) forwardClaudeBodyAsChatCompletions(
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
|
||||
@@ -1103,6 +1103,7 @@ func (s *GeminiMessagesCompatService) Forward(ctx context.Context, c *gin.Contex
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
@@ -1451,13 +1452,14 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
|
||||
estimated := estimateGeminiCountTokens(body)
|
||||
c.JSON(http.StatusOK, map[string]any{"totalTokens": estimated})
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
Usage: ClaudeUsage{},
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
Stream: false,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: nil,
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: ClaudeUsage{},
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
Stream: false,
|
||||
Duration: time.Since(startTime),
|
||||
FirstTokenMs: nil,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -1604,6 +1606,7 @@ func (s *GeminiMessagesCompatService) ForwardNative(ctx context.Context, c *gin.
|
||||
|
||||
return &ForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
UpstreamModel: mappedModel,
|
||||
|
||||
@@ -111,11 +111,12 @@ func (s *OpenAIGatewayService) ForwardGrokVoice(ctx context.Context, c *gin.Cont
|
||||
upstreamID := firstNonEmpty(resp.Header.Get("x-request-id"), resp.Header.Get("xai-request-id"))
|
||||
return &OpenAIForwardResult{
|
||||
// Forced durable money-event id so usage_billing_dedup cannot collapse under a reused client id.
|
||||
RequestID: StableGrokAudioBillingRequestID(upstreamID),
|
||||
Model: baseEndpoint,
|
||||
UpstreamModel: baseEndpoint,
|
||||
Duration: time.Since(started),
|
||||
AudioUsage: audioUsage,
|
||||
RequestID: StableGrokAudioBillingRequestID(upstreamID),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Model: baseEndpoint,
|
||||
UpstreamModel: baseEndpoint,
|
||||
Duration: time.Since(started),
|
||||
AudioUsage: audioUsage,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -739,6 +739,7 @@ func (s *OpenAIGatewayService) ForwardGrokMedia(
|
||||
}
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestIDHeader,
|
||||
UpstreamHeaders: resp.Header,
|
||||
ResponseID: usage.ResponseID,
|
||||
Usage: usage.Usage,
|
||||
Model: resultModel,
|
||||
@@ -873,6 +874,7 @@ func (s *OpenAIGatewayService) forwardGrokMediaVideoContent(
|
||||
// (same path as status polling). Pending snapshot is merged in the handler.
|
||||
result := &OpenAIForwardResult{
|
||||
RequestID: contentRequestID,
|
||||
UpstreamHeaders: contentResp.Header,
|
||||
ResponseHeaders: contentResp.Header.Clone(),
|
||||
Duration: time.Since(startTime),
|
||||
}
|
||||
|
||||
@@ -128,11 +128,12 @@ func (s *OpenAIGatewayService) ForwardAlphaSearch(ctx context.Context, c *gin.Co
|
||||
return nil, nil
|
||||
}
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: strings.TrimSpace(resp.Header.Get("x-request-id")),
|
||||
Model: requestedModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
Duration: time.Since(upstreamStart),
|
||||
WebSearchCalls: 1,
|
||||
RequestID: strings.TrimSpace(resp.Header.Get("x-request-id")),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Model: requestedModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
Duration: time.Since(upstreamStart),
|
||||
WebSearchCalls: 1,
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -212,6 +213,7 @@ func (s *OpenAIGatewayService) forwardAlphaSearchViaResponsesWebSearch(
|
||||
c.Data(http.StatusOK, "application/json", alphaRespBody)
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: strings.TrimSpace(resp.Header.Get("x-request-id")),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Model: requestedModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
UpstreamEndpoint: "/v1/responses",
|
||||
|
||||
@@ -160,13 +160,14 @@ func (s *OpenAIGatewayService) ForwardEmbeddings(
|
||||
writeOpenAIEmbeddingsUpstreamResponse(c, resp, respBody, s.responseHeaderFilter)
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: firstNonEmptyString(resp.Header.Get("x-request-id"), resp.Header.Get("request-id")),
|
||||
Usage: extractOpenAIEmbeddingsUsage(respBody),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
Stream: false,
|
||||
Duration: time.Since(startTime),
|
||||
RequestID: firstNonEmptyString(resp.Header.Get("x-request-id"), resp.Header.Get("request-id")),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: extractOpenAIEmbeddingsUsage(respBody),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
Stream: false,
|
||||
Duration: time.Since(startTime),
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -577,6 +577,7 @@ func (s *OpenAIGatewayService) handleChatBufferedStreamingResponse(
|
||||
|
||||
result := &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -698,6 +699,7 @@ func (s *OpenAIGatewayService) handleChatStreamingResponse(
|
||||
resultWithUsage := func() *OpenAIForwardResult {
|
||||
out := &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -281,6 +281,7 @@ func (s *OpenAIGatewayService) handleCCBufferedFromNativeAnthropic(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: claudeUsageToOpenAIUsage(&usage),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -336,6 +337,7 @@ func (s *OpenAIGatewayService) handleCCStreamingFromNativeAnthropic(
|
||||
resultWithUsage := func() *OpenAIForwardResult {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: claudeUsageToOpenAIUsage(&usage),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -356,6 +356,7 @@ func (s *OpenAIGatewayService) streamRawChatCompletions(
|
||||
resultWithUsage := func() *OpenAIForwardResult {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -524,6 +525,7 @@ func (s *OpenAIGatewayService) bufferRawChatCompletions(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -1238,6 +1238,7 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
|
||||
|
||||
forwardResult := &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
ResponseID: responseID,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
|
||||
@@ -248,6 +248,7 @@ func (s *OpenAIGatewayService) forwardGrokResponses(
|
||||
reasoningEffort := extractOpenAIReasoningEffortFromBody(patchedBody, originalModel)
|
||||
result := &OpenAIForwardResult{
|
||||
RequestID: firstNonEmpty(resp.Header.Get("x-request-id"), resp.Header.Get("xai-request-id")),
|
||||
UpstreamHeaders: resp.Header,
|
||||
ResponseID: responseID,
|
||||
Usage: *usage,
|
||||
Model: originalModel,
|
||||
|
||||
@@ -651,6 +651,7 @@ func (s *OpenAIGatewayService) handleAnthropicBufferedStreamingResponse(
|
||||
|
||||
result := &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
ResponseID: finalResponse.ID,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
@@ -955,6 +956,7 @@ func (s *OpenAIGatewayService) handleAnthropicStreamingResponse(
|
||||
resultWithUsage := func() *OpenAIForwardResult {
|
||||
out := &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
ResponseID: responseID,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
|
||||
@@ -244,6 +244,7 @@ func (s *OpenAIGatewayService) handleNativeAnthropicBufferedResponse(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: claudeUsageToOpenAIUsage(usage),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -517,6 +518,7 @@ func (s *OpenAIGatewayService) nativeAnthropicStreamResult(
|
||||
}
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: claudeUsageToOpenAIUsage(usage),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -163,6 +163,7 @@ func (s *OpenAIGatewayService) bufferChatCompletionsAsAnthropic(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -224,6 +225,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
// (mirrors forwardResponsesViaRawChatCompletions).
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -260,6 +262,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsAnthropic(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -519,6 +519,7 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough(
|
||||
|
||||
forwardResult := &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
ResponseID: responseID,
|
||||
Usage: *usage,
|
||||
Model: reqModel,
|
||||
|
||||
@@ -287,6 +287,7 @@ func (s *OpenAIGatewayService) handleResponsesBufferedFromNativeAnthropic(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: claudeUsageToOpenAIUsage(&usage),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -340,6 +341,7 @@ func (s *OpenAIGatewayService) handleResponsesStreamingFromNativeAnthropic(
|
||||
resultWithUsage := func() *OpenAIForwardResult {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: claudeUsageToOpenAIUsage(&usage),
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -154,6 +154,7 @@ func (s *OpenAIGatewayService) bufferChatCompletionsAsResponses(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -225,6 +226,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
if scan.Err != nil {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: scan.Usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -240,6 +242,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
if err := state.ValidateToolCallArguments(); err != nil {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: scan.Usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
@@ -271,6 +274,7 @@ func (s *OpenAIGatewayService) streamChatCompletionsAsResponses(
|
||||
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: requestID,
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: scan.Usage,
|
||||
Model: originalModel,
|
||||
BillingModel: billingModel,
|
||||
|
||||
@@ -233,8 +233,10 @@ type OpenAIUsage struct {
|
||||
type OpenAIForwardResult struct {
|
||||
RequestID string
|
||||
ResponseID string
|
||||
Usage OpenAIUsage
|
||||
Model string // 原始模型(用于响应和日志显示)
|
||||
// UpstreamHeaders 是直接上游的响应头,用于按账户配置解析上游请求标识。
|
||||
UpstreamHeaders http.Header
|
||||
Usage OpenAIUsage
|
||||
Model string // 原始模型(用于响应和日志显示)
|
||||
// BillingModel is the model used for cost calculation.
|
||||
// When non-empty, CalculateCost uses this instead of Model.
|
||||
// This is set by the Anthropic Messages conversion path where
|
||||
|
||||
@@ -371,6 +371,7 @@ func (s *OpenAIGatewayService) RecordUsage(ctx context.Context, input *OpenAIRec
|
||||
APIKeyID: apiKey.ID,
|
||||
AccountID: account.ID,
|
||||
RequestID: requestID,
|
||||
UpstreamRequestID: usageUpstreamRequestIDPtr(account, result.UpstreamHeaders, result.OpenAIWSMode),
|
||||
Model: result.Model,
|
||||
RequestedModel: requestedModel,
|
||||
UpstreamModel: optionalTrimmedStringPtr(result.UpstreamModel),
|
||||
|
||||
@@ -682,6 +682,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
|
||||
if streamCount > 0 {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: streamUsage,
|
||||
Model: requestModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
@@ -703,6 +704,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
|
||||
firstTokenMs = ttft
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: requestModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
@@ -726,6 +728,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesAPIKey(
|
||||
}
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: requestModel,
|
||||
UpstreamModel: upstreamModel,
|
||||
|
||||
@@ -1891,6 +1891,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth(
|
||||
if imageCount > 0 {
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: requestModel,
|
||||
UpstreamModel: requestModel,
|
||||
@@ -1935,6 +1936,7 @@ func (s *OpenAIGatewayService) forwardOpenAIImagesOAuth(
|
||||
}
|
||||
return &OpenAIForwardResult{
|
||||
RequestID: resp.Header.Get("x-request-id"),
|
||||
UpstreamHeaders: resp.Header,
|
||||
Usage: usage,
|
||||
Model: requestModel,
|
||||
UpstreamModel: requestModel,
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
|
||||
"golang.org/x/net/http/httpguts"
|
||||
)
|
||||
|
||||
// AccountExtraUpstreamRequestIDHeader 是账户 extra 中的键,值为直接上游声明请求标识的响应头名。
|
||||
// 未指定时不记录上游请求标识。
|
||||
const AccountExtraUpstreamRequestIDHeader = "upstream_request_id_header"
|
||||
|
||||
const (
|
||||
maxUpstreamRequestIDHeaderNameLen = 64
|
||||
// maxUsageUpstreamRequestIDLen 与 usage_logs.upstream_request_id VARCHAR(128) 对齐。
|
||||
maxUsageUpstreamRequestIDLen = 128
|
||||
)
|
||||
|
||||
// UpstreamRequestIDHeaderName 返回账户指定的上游请求标识头名,未指定时为空串。
|
||||
func UpstreamRequestIDHeaderName(account *Account) string {
|
||||
if account == nil {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(account.GetExtraString(AccountExtraUpstreamRequestIDHeader))
|
||||
}
|
||||
|
||||
// UpstreamRequestIDFromHeaders 从直接上游的响应头解析请求标识。
|
||||
// 只读账户指定的头;账户未指定头名时恒为空串。
|
||||
func UpstreamRequestIDFromHeaders(account *Account, h http.Header) string {
|
||||
if len(h) == 0 {
|
||||
return ""
|
||||
}
|
||||
name := UpstreamRequestIDHeaderName(account)
|
||||
if name == "" {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(h.Get(name))
|
||||
}
|
||||
|
||||
// usageUpstreamRequestIDPtr 生成落库到 usage_logs.upstream_request_id 的值。
|
||||
// WS 轮次没有 HTTP 响应头,保持 nil;超长时截断到列宽而不是让整条用量行失败。
|
||||
func usageUpstreamRequestIDPtr(account *Account, h http.Header, wsMode bool) *string {
|
||||
if wsMode {
|
||||
return nil
|
||||
}
|
||||
id := UpstreamRequestIDFromHeaders(account, h)
|
||||
if id == "" {
|
||||
return nil
|
||||
}
|
||||
if len(id) > maxUsageUpstreamRequestIDLen {
|
||||
id = id[:maxUsageUpstreamRequestIDLen]
|
||||
}
|
||||
return &id
|
||||
}
|
||||
|
||||
// ValidateUpstreamRequestIDHeaderExtra 校验并规范化 extra 中的上游请求标识头名:
|
||||
// 必须是合法的 HTTP 头字段名且不超过 64 字节;空白值视为未指定并从 extra 中移除。
|
||||
func ValidateUpstreamRequestIDHeaderExtra(extra map[string]any) error {
|
||||
if extra == nil {
|
||||
return nil
|
||||
}
|
||||
raw, ok := extra[AccountExtraUpstreamRequestIDHeader]
|
||||
if !ok || raw == nil {
|
||||
return nil
|
||||
}
|
||||
name, ok := raw.(string)
|
||||
if !ok {
|
||||
return infraerrors.BadRequest("INVALID_UPSTREAM_REQUEST_ID_HEADER",
|
||||
"upstream_request_id_header must be a string")
|
||||
}
|
||||
name = strings.TrimSpace(name)
|
||||
if name == "" {
|
||||
delete(extra, AccountExtraUpstreamRequestIDHeader)
|
||||
return nil
|
||||
}
|
||||
if len(name) > maxUpstreamRequestIDHeaderNameLen || !httpguts.ValidHeaderFieldName(name) {
|
||||
return infraerrors.BadRequest("INVALID_UPSTREAM_REQUEST_ID_HEADER",
|
||||
"upstream_request_id_header must be a valid HTTP header name of at most 64 bytes")
|
||||
}
|
||||
extra[AccountExtraUpstreamRequestIDHeader] = name
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestUpstreamRequestIDFromHeaders_UnconfiguredAccountRecordsNothing(t *testing.T) {
|
||||
h := http.Header{}
|
||||
h.Set("X-Client-Request-ID", "sub2api-client")
|
||||
h.Set("X-Request-ID", "sub2api-local")
|
||||
h.Set("X-Oneapi-Request-Id", "oneapi-1")
|
||||
h.Set("Request-Id", "req_official")
|
||||
h.Set("xai-request-id", "xai-1")
|
||||
h.Set("x-goog-request-id", "goog-1")
|
||||
|
||||
require.Equal(t, "", UpstreamRequestIDFromHeaders(nil, h))
|
||||
for _, platform := range []string{PlatformAnthropic, PlatformOpenAI, PlatformGemini, PlatformAntigravity, PlatformGrok} {
|
||||
require.Equal(t, "", UpstreamRequestIDFromHeaders(&Account{Platform: platform}, h), platform)
|
||||
}
|
||||
blank := &Account{Platform: PlatformOpenAI, Extra: map[string]any{AccountExtraUpstreamRequestIDHeader: " "}}
|
||||
require.Equal(t, "", UpstreamRequestIDFromHeaders(blank, h))
|
||||
}
|
||||
|
||||
func TestUpstreamRequestIDFromHeaders_ReadsOnlyConfiguredHeader(t *testing.T) {
|
||||
account := &Account{
|
||||
Platform: PlatformOpenAI,
|
||||
Extra: map[string]any{AccountExtraUpstreamRequestIDHeader: " x-oneapi-request-id "},
|
||||
}
|
||||
h := http.Header{}
|
||||
h.Set("X-Request-ID", "passthrough-from-real-upstream")
|
||||
require.Equal(t, "", UpstreamRequestIDFromHeaders(account, h))
|
||||
|
||||
h.Set("X-Oneapi-Request-Id", " oneapi-2 ")
|
||||
require.Equal(t, "oneapi-2", UpstreamRequestIDFromHeaders(account, h))
|
||||
require.Equal(t, "", UpstreamRequestIDFromHeaders(account, nil))
|
||||
|
||||
official := &Account{Platform: PlatformAnthropic, Extra: map[string]any{AccountExtraUpstreamRequestIDHeader: "request-id"}}
|
||||
only := http.Header{}
|
||||
only.Set("Request-Id", "req_official")
|
||||
require.Equal(t, "req_official", UpstreamRequestIDFromHeaders(official, only))
|
||||
}
|
||||
|
||||
func TestUsageUpstreamRequestIDPtr(t *testing.T) {
|
||||
account := &Account{Extra: map[string]any{AccountExtraUpstreamRequestIDHeader: "X-Request-ID"}}
|
||||
h := http.Header{}
|
||||
h.Set("X-Request-ID", strings.Repeat("a", 200))
|
||||
require.Nil(t, usageUpstreamRequestIDPtr(account, h, true))
|
||||
require.Nil(t, usageUpstreamRequestIDPtr(account, http.Header{}, false))
|
||||
require.Nil(t, usageUpstreamRequestIDPtr(nil, h, false))
|
||||
require.Nil(t, usageUpstreamRequestIDPtr(&Account{}, h, false))
|
||||
|
||||
got := usageUpstreamRequestIDPtr(account, h, false)
|
||||
require.NotNil(t, got)
|
||||
require.Len(t, *got, maxUsageUpstreamRequestIDLen)
|
||||
}
|
||||
|
||||
func TestValidateUpstreamRequestIDHeaderExtra(t *testing.T) {
|
||||
require.NoError(t, ValidateUpstreamRequestIDHeaderExtra(nil))
|
||||
require.NoError(t, ValidateUpstreamRequestIDHeaderExtra(map[string]any{}))
|
||||
|
||||
blank := map[string]any{AccountExtraUpstreamRequestIDHeader: " "}
|
||||
require.NoError(t, ValidateUpstreamRequestIDHeaderExtra(blank))
|
||||
_, present := blank[AccountExtraUpstreamRequestIDHeader]
|
||||
require.False(t, present, "blank header name must be removed")
|
||||
|
||||
valid := map[string]any{AccountExtraUpstreamRequestIDHeader: " X-Oneapi-Request-Id "}
|
||||
require.NoError(t, ValidateUpstreamRequestIDHeaderExtra(valid))
|
||||
require.Equal(t, "X-Oneapi-Request-Id", valid[AccountExtraUpstreamRequestIDHeader])
|
||||
|
||||
require.Error(t, ValidateUpstreamRequestIDHeaderExtra(map[string]any{AccountExtraUpstreamRequestIDHeader: 1}))
|
||||
require.Error(t, ValidateUpstreamRequestIDHeaderExtra(map[string]any{AccountExtraUpstreamRequestIDHeader: "X Request Id"}))
|
||||
require.Error(t, ValidateUpstreamRequestIDHeaderExtra(map[string]any{AccountExtraUpstreamRequestIDHeader: "X-Request-Id:"}))
|
||||
require.Error(t, ValidateUpstreamRequestIDHeaderExtra(map[string]any{AccountExtraUpstreamRequestIDHeader: strings.Repeat("x", 65)}))
|
||||
}
|
||||
@@ -186,6 +186,10 @@ type UsageLog struct {
|
||||
// (e.g. the session_id / X-Session-Id headers). Nil when the client sent no
|
||||
// valid session header. It is never derived from prompt_cache_key or content.
|
||||
SessionID *string
|
||||
// UpstreamRequestID 是直接上游在响应头中声明的请求标识,只读账户
|
||||
// extra.upstream_request_id_header 指定的头;账户未指定头名、WS 轮次
|
||||
// 与上游没有该头的路径为 nil。
|
||||
UpstreamRequestID *string
|
||||
|
||||
// Cache TTL Override 标记(管理员强制替换了缓存 TTL 计费)
|
||||
CacheTTLOverridden bool
|
||||
|
||||
Reference in New Issue
Block a user