From 40e91d9450f58e83118cc888d58e4d5dd2be3660 Mon Sep 17 00:00:00 2001 From: qian Date: Thu, 20 Aug 2026 08:01:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=E6=96=B0=E5=A2=9E=E6=B8=A0?= =?UTF-8?q?=E9=81=93=E8=B0=83=E8=AF=95=E6=97=A5=E5=BF=97=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/admin/admin.go | 6 + api/admin/v1/channel.go | 8 + api/admin/v1/channel_debug_log.go | 75 +++ api/admin/v1/tenant.go | 24 + docs/reference/goframe-conventions.md | 28 + internal/cmd/cmd.go | 4 + .../admin/admin_v1_channel_debug_log.go | 24 + internal/controller/admin/admin_v1_tenant.go | 3 + internal/dao/chn_debug_logs.go | 22 + internal/dao/internal/chn_debug_logs.go | 157 +++++ internal/handler/relay/relay.go | 30 +- internal/logic/admin/channel.go | 34 +- internal/logic/admin/channel_debug_log.go | 181 +++++ internal/logic/admin/tenant.go | 41 ++ internal/logic/common/audit.go | 13 +- .../logic/common/channel_debug_log_writer.go | 30 + internal/logic/relay/debug_log.go | 93 +++ internal/model/do/chn_debug_logs.go | 55 ++ internal/model/entity/chn_debug_logs.go | 53 ++ internal/service/admin.go | 17 +- internal/service/tenant.go | 1 + migrations/000017_channel_debug_logs.sql | 94 +++ .../000018_channel_debug_logs_conversion.sql | 11 + relay/common/debug_capture.go | 340 ++++++++++ relay/common/debug_capture_test.go | 241 +++++++ relay/common/debug_log.go | 57 ++ relay/common/debug_session.go | 229 +++++++ relay/common/debug_session_test.go | 199 ++++++ relay/common/http_client.go | 14 +- relay/common/http_client_test.go | 23 +- relay/common/relay_info.go | 34 + relay/handler/relay_handler.go | 38 +- web/admin/src/views/ai/ChannelDetailPage.vue | 616 +++++++++++++++++- 33 files changed, 2754 insertions(+), 41 deletions(-) create mode 100644 api/admin/v1/channel_debug_log.go create mode 100644 internal/controller/admin/admin_v1_channel_debug_log.go create mode 100644 internal/dao/chn_debug_logs.go create mode 100644 internal/dao/internal/chn_debug_logs.go create mode 100644 internal/logic/admin/channel_debug_log.go create mode 100644 internal/logic/common/channel_debug_log_writer.go create mode 100644 internal/logic/relay/debug_log.go create mode 100644 internal/model/do/chn_debug_logs.go create mode 100644 internal/model/entity/chn_debug_logs.go create mode 100644 migrations/000017_channel_debug_logs.sql create mode 100644 migrations/000018_channel_debug_logs_conversion.sql create mode 100644 relay/common/debug_capture.go create mode 100644 relay/common/debug_capture_test.go create mode 100644 relay/common/debug_log.go create mode 100644 relay/common/debug_session.go create mode 100644 relay/common/debug_session_test.go diff --git a/api/admin/admin.go b/api/admin/admin.go index 9695efd..02a1b89 100644 --- a/api/admin/admin.go +++ b/api/admin/admin.go @@ -64,6 +64,11 @@ type IAdminV1 interface { ChannelExport(ctx context.Context, req *v1.ChannelExportReq) (res *v1.ChannelExportRes, err error) ChannelClone(ctx context.Context, req *v1.ChannelCloneReq) (res *v1.ChannelCloneRes, err error) ChannelResetHealth(ctx context.Context, req *v1.ChannelResetHealthReq) (res *v1.ChannelResetHealthRes, err error) + ChannelDebugLogList(ctx context.Context, req *v1.ChannelDebugLogListReq) (res *v1.ChannelDebugLogListRes, err error) + ChannelDebugLogStats(ctx context.Context, req *v1.ChannelDebugLogStatsReq) (res *v1.ChannelDebugLogStatsRes, err error) + ChannelDebugLogDetail(ctx context.Context, req *v1.ChannelDebugLogDetailReq) (res *v1.ChannelDebugLogDetailRes, err error) + ChannelDebugLogDelete(ctx context.Context, req *v1.ChannelDebugLogDeleteReq) (res *v1.ChannelDebugLogDeleteRes, err error) + ChannelDebugLogClear(ctx context.Context, req *v1.ChannelDebugLogClearReq) (res *v1.ChannelDebugLogClearRes, err error) ChannelErrorEventList(ctx context.Context, req *v1.ChannelErrorEventListReq) (res *v1.ChannelErrorEventListRes, err error) ChannelErrorStats(ctx context.Context, req *v1.ChannelErrorStatsReq) (res *v1.ChannelErrorStatsRes, err error) ChannelErrorTrend(ctx context.Context, req *v1.ChannelErrorTrendReq) (res *v1.ChannelErrorTrendRes, err error) @@ -253,6 +258,7 @@ type IAdminV1 interface { TenantExport(ctx context.Context, req *v1.TenantExportReq) (res *v1.TenantExportRes, err error) AdminMemberExport(ctx context.Context, req *v1.AdminMemberExportReq) (res *v1.AdminMemberExportRes, err error) TenantSelect(ctx context.Context, req *v1.TenantSelectReq) (res *v1.TenantSelectRes, err error) + TenantApiKeySelect(ctx context.Context, req *v1.TenantApiKeySelectReq) (res *v1.TenantApiKeySelectRes, err error) TenantLevelConfigList(ctx context.Context, req *v1.TenantLevelConfigListReq) (res *v1.TenantLevelConfigListRes, err error) TenantLevelConfigCreate(ctx context.Context, req *v1.TenantLevelConfigCreateReq) (res *v1.TenantLevelConfigCreateRes, err error) TenantLevelConfigUpdate(ctx context.Context, req *v1.TenantLevelConfigUpdateReq) (res *v1.TenantLevelConfigUpdateRes, err error) diff --git a/api/admin/v1/channel.go b/api/admin/v1/channel.go index d5a1e41..796a674 100644 --- a/api/admin/v1/channel.go +++ b/api/admin/v1/channel.go @@ -93,6 +93,10 @@ type ChannelUpdateReq struct { StrictCapacity *bool `json:"strict_capacity" dc:"严格容量（fail-closed）"` IsVIP *bool `json:"is_vip" dc:"是否VIP专属渠道"` UseProxy *bool `json:"use_proxy" dc:"启用代理"` + DebugLogEnabled *bool `json:"debug_log_enabled" dc:"启用渠道调试日志（记录四段完整报文，排障用，用完及时关闭）"` + DebugLogTenantID *int64 `json:"debug_log_tenant_id" dc:"调试目标租户ID过滤（0=不限）"` + DebugLogUserID *int64 `json:"debug_log_user_id" dc:"调试目标成员ID过滤（0=不限）"` + DebugLogApiKeyID *int64 `json:"debug_log_api_key_id" dc:"调试目标密钥ID过滤（0=不限）"` SharingThreshold *float64 `json:"sharing_threshold" dc:"普通租户借用阈值"` PreemptionThreshold *float64 `json:"preemption_threshold" dc:"VIP抢占阈值"` BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds" dc:"被抢占后冷却时间(秒)"` @@ -127,6 +131,10 @@ type ChannelDetailRes struct { Remark string `json:"remark"` IsVIP bool `json:"is_vip"` UseProxy bool `json:"use_proxy"` + DebugLogEnabled bool `json:"debug_log_enabled"` + DebugLogTenantID int64 `json:"debug_log_tenant_id"` + DebugLogUserID int64 `json:"debug_log_user_id"` + DebugLogApiKeyID int64 `json:"debug_log_api_key_id"` SharingThreshold *float64 `json:"sharing_threshold"` PreemptionThreshold *float64 `json:"preemption_threshold"` BorrowingCooldownSeconds *int `json:"borrowing_cooldown_seconds"` diff --git a/api/admin/v1/channel_debug_log.go b/api/admin/v1/channel_debug_log.go new file mode 100644 index 0000000..915e890 --- /dev/null +++ b/api/admin/v1/channel_debug_log.go @@ -0,0 +1,75 @@ +package v1 + +import "github.com/gogf/gf/v2/frame/g" + +// 渠道调试日志：调试开关开启的渠道，per-attempt 记录客户端↔系统↔上游四段完整报文。 +// 端点全部为 GET/DELETE，路径在 /channels/ 下，自动命中 rbac 前缀规则 +//（GET → channel:view，DELETE → channel:delete），无需新增权限点。 + +// ChannelDebugLogListReq 渠道调试日志列表请求（只返回元数据与四段体积，不返回 body） +type ChannelDebugLogListReq struct { + g.Meta `path:"/channels/{channel_id}/debug-logs" method:"get" mime:"json" tags:"管理后台-渠道" summary:"渠道调试日志列表"` + ChannelID int64 `json:"channel_id" in:"path" v:"required" dc:"渠道ID"` + Page int `json:"page" in:"query" d:"1" v:"min:1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"20" v:"min:1|max:100" dc:"每页数量"` + RequestID string `json:"request_id" in:"query" dc:"请求ID精确筛选"` + ModelName string `json:"model_name" in:"query" dc:"模型名模糊筛选"` + UpstreamStatus *int `json:"upstream_status" in:"query" dc:"上游状态码筛选"` + ClientStatus *int `json:"client_status" in:"query" dc:"客户端状态码筛选"` + IsStream *bool `json:"is_stream" in:"query" dc:"是否流式"` + OnlyError *bool `json:"only_error" in:"query" dc:"仅看有错误的记录"` + StartDate string `json:"start_date" in:"query" dc:"开始日期 YYYY-MM-DD"` + EndDate string `json:"end_date" in:"query" dc:"结束日期 YYYY-MM-DD"` +} + +// ChannelDebugLogListRes 渠道调试日志列表响应 +type ChannelDebugLogListRes struct { + List []map[string]any `json:"list"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` +} + +// ChannelDebugLogStatsReq 渠道调试日志统计请求 +type ChannelDebugLogStatsReq struct { + g.Meta `path:"/channels/{channel_id}/debug-logs/stats" method:"get" mime:"json" tags:"管理后台-渠道" summary:"渠道调试日志统计"` + ChannelID int64 `json:"channel_id" in:"path" v:"required" dc:"渠道ID"` +} + +// ChannelDebugLogStatsRes 渠道调试日志统计响应 +type ChannelDebugLogStatsRes struct { + Total int64 `json:"total" dc:"记录条数"` + TotalBytes int64 `json:"total_bytes" dc:"四段 body 落库总字节数"` + OldestAt string `json:"oldest_at" dc:"最早记录时间"` +} + +// ChannelDebugLogDetailReq 渠道调试日志详情请求（返回四段完整报文） +type ChannelDebugLogDetailReq struct { + g.Meta `path:"/channels/{channel_id}/debug-logs/{id}" method:"get" mime:"json" tags:"管理后台-渠道" summary:"渠道调试日志详情"` + ChannelID int64 `json:"channel_id" in:"path" v:"required" dc:"渠道ID"` + ID int64 `json:"id" in:"path" v:"required" dc:"记录ID"` +} + +// ChannelDebugLogDetailRes 渠道调试日志详情响应 +type ChannelDebugLogDetailRes struct { + Data map[string]any `json:"data"` +} + +// ChannelDebugLogDeleteReq 删除单条调试日志请求（硬删除） +type ChannelDebugLogDeleteReq struct { + g.Meta `path:"/channels/{channel_id}/debug-logs/{id}" method:"delete" mime:"json" tags:"管理后台-渠道" summary:"删除调试日志"` + ChannelID int64 `json:"channel_id" in:"path" v:"required" dc:"渠道ID"` + ID int64 `json:"id" in:"path" v:"required" dc:"记录ID"` +} + +// ChannelDebugLogDeleteRes 删除调试日志响应 +type ChannelDebugLogDeleteRes struct{} + +// ChannelDebugLogClearReq 清空渠道调试日志请求（按渠道硬删除全部） +type ChannelDebugLogClearReq struct { + g.Meta `path:"/channels/{channel_id}/debug-logs" method:"delete" mime:"json" tags:"管理后台-渠道" summary:"清空渠道调试日志"` + ChannelID int64 `json:"channel_id" in:"path" v:"required" dc:"渠道ID"` +} + +// ChannelDebugLogClearRes 清空渠道调试日志响应 +type ChannelDebugLogClearRes struct{} diff --git a/api/admin/v1/tenant.go b/api/admin/v1/tenant.go index 24cf348..57c40a9 100644 --- a/api/admin/v1/tenant.go +++ b/api/admin/v1/tenant.go @@ -174,3 +174,27 @@ type TenantSelectItem struct { Name string `json:"name"` Code string `json:"code"` } + +// TenantApiKeySelectReq 租户 API Key 下拉选择列表（轻量，用于调试目标过滤等选择器） +type TenantApiKeySelectReq struct { + g.Meta `path:"/tenants/{tenant_id}/api-keys/select" method:"get" mime:"json" tags:"管理后台-租户管理" summary:"租户API Key下拉选择列表"` + TenantID int64 `json:"tenant_id" in:"path" v:"required" dc:"租户ID"` + UserID int64 `json:"user_id" in:"query" dc:"按创建者成员ID筛选（0=不限）"` + Page int `json:"page" in:"query" d:"1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"100" v:"min:1|max:200" dc:"每页数量"` +} + +type TenantApiKeySelectRes struct { + List []TenantApiKeySelectItem `json:"list"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` +} + +type TenantApiKeySelectItem struct { + ID int64 `json:"id"` + Name string `json:"name"` + KeyPrefix string `json:"key_prefix"` + UserID int64 `json:"user_id"` + Status string `json:"status"` +} diff --git a/docs/reference/goframe-conventions.md b/docs/reference/goframe-conventions.md index 1db0c80..ad86313 100644 --- a/docs/reference/goframe-conventions.md +++ b/docs/reference/goframe-conventions.md @@ -751,3 +751,31 @@ err := g.DB().Transaction(dbCtx, func(txCtx context.Context, tx gdb.TX) error { - **前端编辑表单慎用 `x || 默认值` 兜底**：对数值字段，0 会被 falsy 规则吞掉，既掩盖真实数据，又可能在保存时把合法的 0 改写为默认值。 排查信号：某字段在库里莫名变成 0/空串，而创建时明明写过合法值——查该表所有 UPDATE 路径中是否存在非指针字段的无条件赋值。 + +### 2026-08-20（三）：Fields 挂在共用模型上导致 COUNT(col1, col2, ...) 报错 + +**问题**：`TenantApiKeySelect` 中把 `.Fields("id, name, ...")` 挂在模型链上后调用 `m.Count()`，PostgreSQL 报 `function count(bigint, character varying, ...) does not exist`——gdb 生成了 `SELECT COUNT("id","name",...)` 而不是 `COUNT(*)`。 + +**原因**：gdb 的 `Count()` 会复用模型上已设置的 `Fields` 作为 COUNT 的参数列表（多字段时逐个传入 count 聚合函数），而 PostgreSQL 的 `count` 只接受单个参数（MySQL 语义下多参 count 也不等于总行数）。这是"链式复用同一 model 做计数 + 查询"模式下的隐蔽陷阱：查列表正常、计数 SQL 被悄悄改写。 + +**修复**：`.Fields(...)` 只挂在分页查询链上（`m.Fields(...).Page(...).Scan(...)`），`Count()` 用不带 Fields 的基础模型（生成 `COUNT(*)`）。 + +**正确做法**：凡"先 Count 再分页 Scan"的写法，构建条件模型（Where/Order）后**先 Count**，再把 `Fields/Page/Scan` 链在同一个基础模型上；不要在 Count 之前设置 Fields。若必须提前设置 Fields，用 `m.Clone()` 或为 Count 单独构建条件链。 + +排查信号：日志出现 `function count(... N 个参数 ...) does not exist` 或 `COUNT("a","b")` 形态的 SQL，且调用栈落在 `*.Count()`。 + +### 2026-08-20（三）：Model.Scan(&string) 静默吞错，局部更新把未提交的 JSONB 字段重置为默认值 + +**问题**：渠道调试日志「保存捕捉条件后开关自动关闭」。PUT 只带 `debug_log_tenant_id` 等过滤字段，settings JSONB 里的 `debug_log_enabled:true` 被静默丢弃。查库发现过滤字段写入成功、唯独开关标志消失。 + +**原因**：读旧 settings 用的是 `_ = dao.ChnChannels...Fields("settings").Scan(¤tSettings)`——gdb 的 `Model.Scan` **只接受 struct/*struct/[]struct/[]*struct**，传入 `*string` 直接返回错误 `element of parameter "pointer" for function Scan should type of struct/...`；`_ =` 把错误吞掉后 `currentSettings` 永远是空串，`ParseChannelSettings("")` 返回默认值，套上本次请求的字段 marshal 回写——**所有未在本次请求中显式提交的 settings 字段全部被重置为默认**。开关单独提交时显式带上了 enabled 所以正常；过滤条件单独提交时没带 enabled，于是被重置为 false。这是个从原 UseProxy 更新块继承来的潜伏 bug：此前每次切换 use_proxy 同样会把 timeout_seconds 覆盖、header_override 等自定义配置静默洗掉，只是显式字段每次都带、无人察觉。同模式的 `GetTenantAuditLevel`（audit.go，只读路径）也中招——租户级审计设置因此**从未生效过**。 + +**修复**：读单列字符串一律用 `Model.Value()`（返回 `*gvar.Var`，`v.String()` 取值），两处（channel.go 更新块、audit.go）均已改为 Value()。 + +**正确做法（通用规则）**： + +- **读单个标量列用 `.Value()` 或 `.Array()`/`.All()` 后取字段，禁止 `Scan(&string)`/`Scan(&int)`**——Scan 只认 struct 系列，对基础类型指针直接报错。 +- **禁止 `_ =` 吞掉 DB 读取错误**，尤其是「读旧值 → 改字段 → 写回」的合并更新模式：读失败意味着回写会把整个 JSONB 洗成默认值，属于数据破坏而非可忽略的小错。 +- 「读-改-写」JSONB 配置的合并更新块，读失败时应中止更新并返回错误，而不是带着默认值继续。 + +排查信号：JSONB 配置列中「本次没提交的字段莫名回到默认值」；`grep 'Scan(&' `命中基础类型指针。 diff --git a/internal/cmd/cmd.go b/internal/cmd/cmd.go index dd0dffc..337e1df 100644 --- a/internal/cmd/cmd.go +++ b/internal/cmd/cmd.go @@ -119,6 +119,10 @@ var ( // Initialize async audit log writer common.InitAuditLogWriter() + // Initialize async channel debug log writer + 注入 relay 层提交钩子 + common.InitChannelDebugLogWriter() + relayLogic.InitDebugLogRecorder() + // Initialize async error log writer response.InitErrorLogWriter() diff --git a/internal/controller/admin/admin_v1_channel_debug_log.go b/internal/controller/admin/admin_v1_channel_debug_log.go new file mode 100644 index 0000000..31c50da --- /dev/null +++ b/internal/controller/admin/admin_v1_channel_debug_log.go @@ -0,0 +1,24 @@ +package admin + +import ( + "context" + + "github.com/qianfree/team-api/api/admin/v1" + "github.com/qianfree/team-api/internal/service" +) + +func (c *ControllerV1) ChannelDebugLogList(ctx context.Context, req *v1.ChannelDebugLogListReq) (res *v1.ChannelDebugLogListRes, err error) { + return service.Admin().ChannelDebugLogList(ctx, req) +} +func (c *ControllerV1) ChannelDebugLogStats(ctx context.Context, req *v1.ChannelDebugLogStatsReq) (res *v1.ChannelDebugLogStatsRes, err error) { + return service.Admin().ChannelDebugLogStats(ctx, req) +} +func (c *ControllerV1) ChannelDebugLogDetail(ctx context.Context, req *v1.ChannelDebugLogDetailReq) (res *v1.ChannelDebugLogDetailRes, err error) { + return service.Admin().ChannelDebugLogDetail(ctx, req) +} +func (c *ControllerV1) ChannelDebugLogDelete(ctx context.Context, req *v1.ChannelDebugLogDeleteReq) (res *v1.ChannelDebugLogDeleteRes, err error) { + return service.Admin().ChannelDebugLogDelete(ctx, req) +} +func (c *ControllerV1) ChannelDebugLogClear(ctx context.Context, req *v1.ChannelDebugLogClearReq) (res *v1.ChannelDebugLogClearRes, err error) { + return service.Admin().ChannelDebugLogClear(ctx, req) +} diff --git a/internal/controller/admin/admin_v1_tenant.go b/internal/controller/admin/admin_v1_tenant.go index 6c836be..8cf84fb 100644 --- a/internal/controller/admin/admin_v1_tenant.go +++ b/internal/controller/admin/admin_v1_tenant.go @@ -37,3 +37,6 @@ func (c *ControllerV1) AdminMemberExport(ctx context.Context, req *v1.AdminMembe func (c *ControllerV1) TenantSelect(ctx context.Context, req *v1.TenantSelectReq) (res *v1.TenantSelectRes, err error) { return service.Admin().TenantSelect(ctx, req) } +func (c *ControllerV1) TenantApiKeySelect(ctx context.Context, req *v1.TenantApiKeySelectReq) (res *v1.TenantApiKeySelectRes, err error) { + return service.Admin().TenantApiKeySelect(ctx, req) +} diff --git a/internal/dao/chn_debug_logs.go b/internal/dao/chn_debug_logs.go new file mode 100644 index 0000000..c28a047 --- /dev/null +++ b/internal/dao/chn_debug_logs.go @@ -0,0 +1,22 @@ +// ================================================================================= +// This file is auto-generated by the GoFrame CLI tool. You may modify it as needed. +// ================================================================================= + +package dao + +import ( + "github.com/qianfree/team-api/internal/dao/internal" +) + +// chnDebugLogsDao is the data access object for the table chn_debug_logs. +// You can define custom methods on it to extend its functionality as needed. +type chnDebugLogsDao struct { + *internal.ChnDebugLogsDao +} + +var ( + // ChnDebugLogs is a globally accessible object for table chn_debug_logs operations. + ChnDebugLogs = chnDebugLogsDao{internal.NewChnDebugLogsDao()} +) + +// Add your custom methods and functionality below. diff --git a/internal/dao/internal/chn_debug_logs.go b/internal/dao/internal/chn_debug_logs.go new file mode 100644 index 0000000..ec1160e --- /dev/null +++ b/internal/dao/internal/chn_debug_logs.go @@ -0,0 +1,157 @@ +// ========================================================================== +// Code generated and maintained by GoFrame CLI tool. DO NOT EDIT. +// ========================================================================== + +package internal + +import ( + "context" + + "github.com/gogf/gf/v2/database/gdb" + "github.com/gogf/gf/v2/frame/g" +) + +// ChnDebugLogsDao is the data access object for the table chn_debug_logs. +type ChnDebugLogsDao struct { + table string // table is the underlying table name of the DAO. + group string // group is the database configuration group name of the current DAO. + columns ChnDebugLogsColumns // columns contains all the column names of Table for convenient usage. + handlers []gdb.ModelHandler // handlers for customized model modification. +} + +// ChnDebugLogsColumns defines and stores column names for the table chn_debug_logs. +type ChnDebugLogsColumns struct { + Id string // 主键ID + ChannelId string // 渠道ID（逻辑关联 chn_channels.id，无外键） + ChannelName string // 渠道名称（冗余存储，渠道删除后仍可辨识） + ChannelType string // 渠道类型（ProviderType 枚举值） + RequestId string // 请求唯一ID（同一请求多次重试共享） + TenantId string // 租户ID + UserId string // 用户ID + ApiKeyId string // API Key ID + ModelName string // 用户请求的模型名 + UpstreamModel string // 上游实际使用的模型名（模型映射后） + RelayMode string // 转发模式（chat_completions/claude_messages/embeddings 等） + InboundPath string // 入站端点路径（如 /v1/chat/completions） + UpstreamUrl string // 上游请求 URL（query 中的凭证参数已脱敏） + IsStream string // 是否流式请求 + RetryIndex string // 重试轮次（0=首次尝试） + IsFinal string // 是否为产生客户端响应的最终尝试（成功/终止/流中断） + UpstreamStatusCode string // 上游 HTTP 状态码（未发起请求或连接失败为 NULL） + ClientStatusCode string // 返回客户端的状态码 + Error string // 本尝试的错误信息（成功为空） + ClientReqHeaders string // 段1 客户端请求头（凭证类头脱敏：前6后4）JSON + ClientReqBody string // 段1 客户端请求体（完整不截断；二进制为 base64，见 encoding 列） + ClientReqEncoding string // 段1 编码：plain / base64 + UpstreamReqHeaders string // 段2 发往上游的最终请求头（协议转换+override 后，凭证类脱敏）JSON + UpstreamReqBody string // 段2 发往上游的请求体（实际发送字节，含协议转换；二进制为 base64） + UpstreamReqEncoding string // 段2 编码：plain / base64 + UpstreamRespHeaders string // 段3 上游响应头 JSON（Content-Encoding 可能已被 net/http 透明解压移除） + UpstreamRespBody string // 段3 上游响应体（Go 透明解压后的字节；流式为 SSE 原文；二进制为 base64） + UpstreamRespEncoding string // 段3 编码：plain / base64 + ClientRespHeaders string // 段4 返回客户端的响应头 JSON + ClientRespBody string // 段4 返回客户端的响应体（协议转换后，完整不截断；二进制为 base64） + ClientRespEncoding string // 段4 编码：plain / base64 + UpstreamLatencyMs string // 上游往返耗时（RoundTrip 到响应头）毫秒 + TotalLatencyMs string // 请求总耗时毫秒 + FirstTokenMs string // 首字节延迟毫秒 + ClientReqBytes string // 段1 请求体原始字节数（base64 落库膨胀前的真实大小） + UpstreamReqBytes string // 段2 请求体原始字节数 + UpstreamRespBytes string // 段3 响应体原始字节数 + ClientRespBytes string // 段4 响应体原始字节数 + CreatedAt string // 创建时间 + Conversion string // 协议转换信息 JSON：client_format 客户端协议 / upstream_format 上游协议 / chain 请求侧转换链 / bridge 桥接方式（responses_api|responses_direct|pass_through，空=常规转换） +} + +// chnDebugLogsColumns holds the columns for the table chn_debug_logs. +var chnDebugLogsColumns = ChnDebugLogsColumns{ + Id: "id", + ChannelId: "channel_id", + ChannelName: "channel_name", + ChannelType: "channel_type", + RequestId: "request_id", + TenantId: "tenant_id", + UserId: "user_id", + ApiKeyId: "api_key_id", + ModelName: "model_name", + UpstreamModel: "upstream_model", + RelayMode: "relay_mode", + InboundPath: "inbound_path", + UpstreamUrl: "upstream_url", + IsStream: "is_stream", + RetryIndex: "retry_index", + IsFinal: "is_final", + UpstreamStatusCode: "upstream_status_code", + ClientStatusCode: "client_status_code", + Error: "error", + ClientReqHeaders: "client_req_headers", + ClientReqBody: "client_req_body", + ClientReqEncoding: "client_req_encoding", + UpstreamReqHeaders: "upstream_req_headers", + UpstreamReqBody: "upstream_req_body", + UpstreamReqEncoding: "upstream_req_encoding", + UpstreamRespHeaders: "upstream_resp_headers", + UpstreamRespBody: "upstream_resp_body", + UpstreamRespEncoding: "upstream_resp_encoding", + ClientRespHeaders: "client_resp_headers", + ClientRespBody: "client_resp_body", + ClientRespEncoding: "client_resp_encoding", + UpstreamLatencyMs: "upstream_latency_ms", + TotalLatencyMs: "total_latency_ms", + FirstTokenMs: "first_token_ms", + ClientReqBytes: "client_req_bytes", + UpstreamReqBytes: "upstream_req_bytes", + UpstreamRespBytes: "upstream_resp_bytes", + ClientRespBytes: "client_resp_bytes", + CreatedAt: "created_at", + Conversion: "conversion", +} + +// NewChnDebugLogsDao creates and returns a new DAO object for table data access. +func NewChnDebugLogsDao(handlers ...gdb.ModelHandler) *ChnDebugLogsDao { + return &ChnDebugLogsDao{ + group: "default", + table: "chn_debug_logs", + columns: chnDebugLogsColumns, + handlers: handlers, + } +} + +// DB retrieves and returns the underlying raw database management object of the current DAO. +func (dao *ChnDebugLogsDao) DB() gdb.DB { + return g.DB(dao.group) +} + +// Table returns the table name of the current DAO. +func (dao *ChnDebugLogsDao) Table() string { + return dao.table +} + +// Columns returns all column names of the current DAO. +func (dao *ChnDebugLogsDao) Columns() ChnDebugLogsColumns { + return dao.columns +} + +// Group returns the database configuration group name of the current DAO. +func (dao *ChnDebugLogsDao) Group() string { + return dao.group +} + +// Ctx creates and returns a Model for the current DAO. It automatically sets the context for the current operation. +func (dao *ChnDebugLogsDao) Ctx(ctx context.Context) *gdb.Model { + model := dao.DB().Model(dao.table) + for _, handler := range dao.handlers { + model = handler(model) + } + return model.Safe().Ctx(ctx) +} + +// Transaction wraps the transaction logic using function f. +// It rolls back the transaction and returns the error if function f returns a non-nil error. +// It commits the transaction and returns nil if function f returns nil. +// +// Note: Do not commit or roll back the transaction in function f, +// as it is automatically handled by this function. +func (dao *ChnDebugLogsDao) Transaction(ctx context.Context, f func(ctx context.Context, tx gdb.TX) error) (err error) { + return dao.Ctx(ctx).Transaction(ctx, f) +} diff --git a/internal/handler/relay/relay.go b/internal/handler/relay/relay.go index f42da2e..e71e18f 100644 --- a/internal/handler/relay/relay.go +++ b/internal/handler/relay/relay.go @@ -56,7 +56,7 @@ func HandleChatCompletions(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/chat/completions", latencyMs, firstTokenMs(billingResult)) return } @@ -90,7 +90,7 @@ func HandleEmbeddings(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/embeddings", latencyMs, firstTokenMs(billingResult)) return } @@ -124,7 +124,7 @@ func HandleImagesGenerations(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/images/generations", latencyMs, firstTokenMs(billingResult)) return } @@ -158,7 +158,7 @@ func HandleCompletions(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/completions", latencyMs, firstTokenMs(billingResult)) return } @@ -196,7 +196,7 @@ func HandleResponses(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/responses", latencyMs, firstTokenMs(billingResult)) return } @@ -228,7 +228,7 @@ func HandleMessages(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteClaudeRelayError(capture, err) + handler.WriteClaudeRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/messages", latencyMs, firstTokenMs(billingResult)) return } @@ -314,7 +314,7 @@ func HandleGeminiGenerateContent(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteGeminiRelayError(capture, err) + handler.WriteGeminiRelayError(rc.Writer, err) recordAudit(r, rc, capture, modifiedBody, path, latencyMs, firstTokenMs(billingResult)) return } @@ -568,6 +568,10 @@ func recordAudit(r *ghttp.Request, rc *handler.RelayContext, capture *ResponseCa ResponseHeaders: capture.ResponseHeaders(), ForwardingTrace: rc.ForwardingTrace, }) + + // 渠道调试日志：补段4（客户端响应，此时错误分支的 WriteRelayError 已写入 rc.Writer） + // 并提交最终尝试记录；无调试会话（开关未开启）时 no-op + rc.Debug.FinalizeAndSubmit(int64(latencyMs), int64(ttft)) } // isBinaryContentType 判断请求是否为 multipart 或其它二进制类型， @@ -641,7 +645,7 @@ func HandleAudioSpeech(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/audio/speech", latencyMs, firstTokenMs(billingResult)) return } @@ -672,7 +676,7 @@ func HandleAudioTranscription(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/audio/transcriptions", latencyMs, firstTokenMs(billingResult)) return } @@ -703,7 +707,7 @@ func HandleAudioTranslation(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/audio/translations", latencyMs, firstTokenMs(billingResult)) return } @@ -734,7 +738,7 @@ func HandleRerank(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/rerank", latencyMs, firstTokenMs(billingResult)) return } @@ -819,7 +823,7 @@ func HandleModerations(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/moderations", latencyMs, firstTokenMs(billingResult)) return } @@ -850,7 +854,7 @@ func HandleImagesEdits(r *ghttp.Request) { setDeprecationHeaders(r, billingResult) if err != nil { - handler.WriteRelayError(capture, err) + handler.WriteRelayError(rc.Writer, err) recordAudit(r, rc, capture, body, "/images/edits", latencyMs, firstTokenMs(billingResult)) return } diff --git a/internal/logic/admin/channel.go b/internal/logic/admin/channel.go index ac7c61a..c914d7e 100644 --- a/internal/logic/admin/channel.go +++ b/internal/logic/admin/channel.go @@ -405,12 +405,34 @@ func (s *sAdmin) UpdateChannel(ctx context.Context, req *v1.ChannelUpdateReq) (* data.BorrowingCooldownSeconds = *req.BorrowingCooldownSeconds } - // Update use_proxy in settings JSONB - if req.UseProxy != nil { + // Update use_proxy / debug_log_* in settings JSONB。 + // 必须单块合并处理（读一次旧值 → 同时应用全部字段 → 回写一次）： + // 若写成多个同构块，各自从 DB 读旧值再各自 marshal，同请求多字段时会互相覆盖。 + if req.UseProxy != nil || req.DebugLogEnabled != nil || req.DebugLogTenantID != nil || + req.DebugLogUserID != nil || req.DebugLogApiKeyID != nil { + // 注意：读单列字符串必须用 Value()——gdb 的 Model.Scan 只接受 struct 系列， + // 传入 *string 会直接报错；此前用 _ = Scan(&string) 吞错导致永远读到空串， + // 每次局部更新都会把未提交的 settings 字段静默重置为默认值 var currentSettings string - _ = dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Fields("settings").Scan(¤tSettings) + if v, err := dao.ChnChannels.Ctx(ctx).Where("id", req.ID).Fields("settings").Value(); err == nil { + currentSettings = v.String() + } settings := relay.ParseChannelSettings(currentSettings) - settings.UseProxy = *req.UseProxy + if req.UseProxy != nil { + settings.UseProxy = *req.UseProxy + } + if req.DebugLogEnabled != nil { + settings.DebugLogEnabled = *req.DebugLogEnabled + } + if req.DebugLogTenantID != nil { + settings.DebugLogTenantID = *req.DebugLogTenantID + } + if req.DebugLogUserID != nil { + settings.DebugLogUserID = *req.DebugLogUserID + } + if req.DebugLogApiKeyID != nil { + settings.DebugLogApiKeyID = *req.DebugLogApiKeyID + } if settingsJSON, err := json.Marshal(settings); err == nil { data.Settings = string(settingsJSON) } @@ -554,6 +576,10 @@ func (s *sAdmin) GetChannelDetail(ctx context.Context, req *v1.ChannelDetailReq) Remark: ch.Remark, IsVIP: ch.IsVIP, UseProxy: settings.UseProxy, + DebugLogEnabled: settings.DebugLogEnabled, + DebugLogTenantID: settings.DebugLogTenantID, + DebugLogUserID: settings.DebugLogUserID, + DebugLogApiKeyID: settings.DebugLogApiKeyID, SharingThreshold: ch.SharingThreshold, PreemptionThreshold: ch.PreemptionThreshold, BorrowingCooldownSeconds: ch.BorrowingCooldownSeconds, diff --git a/internal/logic/admin/channel_debug_log.go b/internal/logic/admin/channel_debug_log.go new file mode 100644 index 0000000..76a0f8d --- /dev/null +++ b/internal/logic/admin/channel_debug_log.go @@ -0,0 +1,181 @@ +package admin + +import ( + "context" + "encoding/json" + "strings" + "time" + + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" + + v1 "github.com/qianfree/team-api/api/admin/v1" + "github.com/qianfree/team-api/internal/dao" + "github.com/qianfree/team-api/internal/logic/common" +) + +// debugLogListFields 列表查询字段：元数据 + 四段体积（octet_length），绝不 SELECT body 列 +const debugLogListFields = "id, channel_id, channel_name, channel_type, request_id, tenant_id, user_id, api_key_id, " + + "model_name, upstream_model, relay_mode, inbound_path, upstream_url, is_stream, retry_index, is_final, " + + "upstream_status_code, client_status_code, error, upstream_latency_ms, total_latency_ms, first_token_ms, conversion, " + + "octet_length(client_req_body) AS client_req_body_size, " + + "octet_length(upstream_req_body) AS upstream_req_body_size, " + + "octet_length(upstream_resp_body) AS upstream_resp_body_size, " + + "octet_length(client_resp_body) AS client_resp_body_size, " + + "created_at" + +// ChannelDebugLogList 渠道调试日志列表（原生 SQL：octet_length 表达式经 gdb Fields 有引号化风险） +func (s *sAdmin) ChannelDebugLogList(ctx context.Context, req *v1.ChannelDebugLogListReq) (*v1.ChannelDebugLogListRes, error) { + if err := common.ValidateDateParam(req.StartDate, "开始日期"); err != nil { + return nil, err + } + if err := common.ValidateDateParam(req.EndDate, "结束日期"); err != nil { + return nil, err + } + + conditions := []string{"channel_id = ?"} + args := []any{req.ChannelID} + if req.RequestID != "" { + conditions = append(conditions, "request_id = ?") + args = append(args, req.RequestID) + } + if req.ModelName != "" { + conditions = append(conditions, "model_name ILIKE ?") + args = append(args, "%"+req.ModelName+"%") + } + if req.UpstreamStatus != nil { + conditions = append(conditions, "upstream_status_code = ?") + args = append(args, *req.UpstreamStatus) + } + if req.ClientStatus != nil { + conditions = append(conditions, "client_status_code = ?") + args = append(args, *req.ClientStatus) + } + if req.IsStream != nil { + conditions = append(conditions, "is_stream = ?") + args = append(args, *req.IsStream) + } + if req.OnlyError != nil && *req.OnlyError { + conditions = append(conditions, "COALESCE(error, '') <> ''") + } + if req.StartDate != "" { + conditions = append(conditions, "created_at >= ?") + args = append(args, req.StartDate+" 00:00:00") + } + if req.EndDate != "" { + conditions = append(conditions, "created_at <= ?") + args = append(args, req.EndDate+" 23:59:59") + } + where := strings.Join(conditions, " AND ") + + countRows, err := g.DB().Ctx(ctx).Query(ctx, + "SELECT COUNT(*) AS total FROM chn_debug_logs WHERE "+where, args...) + if err != nil { + return nil, err + } + total := 0 + if len(countRows) > 0 { + total = countRows[0]["total"].Int() + } + if total == 0 { + return &v1.ChannelDebugLogListRes{List: []map[string]any{}, Total: 0, Page: req.Page, PageSize: req.PageSize}, nil + } + + rows, err := g.DB().Ctx(ctx).Query(ctx, + "SELECT "+debugLogListFields+" FROM chn_debug_logs WHERE "+where+ + " ORDER BY id DESC LIMIT ? OFFSET ?", + append(args, req.PageSize, (req.Page-1)*req.PageSize)...) + if err != nil { + return nil, err + } + + return &v1.ChannelDebugLogListRes{ + List: toMapList(rows), + Total: total, + Page: req.Page, + PageSize: req.PageSize, + }, nil +} + +// ChannelDebugLogStats 渠道调试日志统计（条数/落库体积/最早时间，供清理提示） +func (s *sAdmin) ChannelDebugLogStats(ctx context.Context, req *v1.ChannelDebugLogStatsReq) (*v1.ChannelDebugLogStatsRes, error) { + rows, err := g.DB().Ctx(ctx).Query(ctx, + "SELECT COUNT(*) AS total, "+ + "COALESCE(SUM(octet_length(client_req_body) + octet_length(upstream_req_body) + "+ + "octet_length(upstream_resp_body) + octet_length(client_resp_body)), 0) AS total_bytes, "+ + "MIN(created_at) AS oldest_at "+ + "FROM chn_debug_logs WHERE channel_id = ?", req.ChannelID) + if err != nil { + return nil, err + } + res := &v1.ChannelDebugLogStatsRes{} + if len(rows) > 0 { + res.Total = rows[0]["total"].Int64() + res.TotalBytes = rows[0]["total_bytes"].Int64() + if v := rows[0]["oldest_at"]; v != nil { + if t, ok := v.Interface().(time.Time); ok && !t.IsZero() { + res.OldestAt = gtime.New(t).String() + } + } + } + return res, nil +} + +// ChannelDebugLogDetail 渠道调试日志详情（四段完整报文；headers JSONB 解析为对象便于前端渲染） +func (s *sAdmin) ChannelDebugLogDetail(ctx context.Context, req *v1.ChannelDebugLogDetailReq) (*v1.ChannelDebugLogDetailRes, error) { + record, err := dao.ChnDebugLogs.Ctx(ctx). + Where("channel_id", req.ChannelID). + Where("id", req.ID). + One() + if err != nil { + return nil, err + } + if record.IsEmpty() { + return nil, common.NewNotFoundError("调试日志") + } + + data := record.Map() + // headers JSONB 列可能以字符串返回，解析为对象；无效/空值保持原样 + for _, field := range []string{ + "client_req_headers", "upstream_req_headers", "upstream_resp_headers", "client_resp_headers", + } { + data[field] = parseJSONField(data[field]) + } + return &v1.ChannelDebugLogDetailRes{Data: data}, nil +} + +// ChannelDebugLogDelete 删除单条调试日志（硬删除） +func (s *sAdmin) ChannelDebugLogDelete(ctx context.Context, req *v1.ChannelDebugLogDeleteReq) (*v1.ChannelDebugLogDeleteRes, error) { + _, err := dao.ChnDebugLogs.Ctx(ctx). + Where("channel_id", req.ChannelID). + Where("id", req.ID). + Delete() + if err != nil { + return nil, err + } + return nil, nil +} + +// ChannelDebugLogClear 清空渠道调试日志（按渠道硬删除全部） +func (s *sAdmin) ChannelDebugLogClear(ctx context.Context, req *v1.ChannelDebugLogClearReq) (*v1.ChannelDebugLogClearRes, error) { + _, err := dao.ChnDebugLogs.Ctx(ctx). + Where("channel_id", req.ChannelID). + Delete() + if err != nil { + return nil, err + } + return nil, nil +} + +// parseJSONField 尝试将字符串形式的 JSON 解析为对象；解析失败或非字符串时原样返回 +func parseJSONField(v any) any { + s, ok := v.(string) + if !ok || s == "" || s == "null" { + return v + } + var parsed any + if err := json.Unmarshal([]byte(s), &parsed); err != nil { + return v + } + return parsed +} diff --git a/internal/logic/admin/tenant.go b/internal/logic/admin/tenant.go index 693328a..92d1a88 100644 --- a/internal/logic/admin/tenant.go +++ b/internal/logic/admin/tenant.go @@ -654,3 +654,44 @@ func (s *sAdmin) ExportTenants(ctx context.Context, req *v1.TenantExportReq) (*v } }) } + +// TenantApiKeySelect 租户 API Key 下拉选择列表（轻量，不返回密钥原值）。 +// 供渠道调试目标过滤等管理端选择器使用；user_id > 0 时按创建者过滤。 +// 注意：Fields 不能挂在 Count 之前的模型上——gdb 会生成 COUNT(col1, col2, ...)， +// PostgreSQL 的 count 只接受单参数，会直接报错。 +func (s *sAdmin) TenantApiKeySelect(ctx context.Context, req *v1.TenantApiKeySelectReq) (*v1.TenantApiKeySelectRes, error) { + m := dao.ApiKeys.Ctx(ctx). + Where("tenant_id", req.TenantID). + OrderDesc("id") + if req.UserID > 0 { + m = m.Where("user_id", req.UserID) + } + + total, err := m.Count() + if err != nil { + return nil, err + } + + var keys []struct { + ID int64 `json:"id"` + Name string `json:"name"` + KeyPrefix string `json:"key_prefix"` + UserID int64 `json:"user_id"` + Status string `json:"status"` + } + if err := m.Fields("id, name, key_prefix, user_id, status").Page(req.Page, req.PageSize).Scan(&keys); err != nil { + return nil, err + } + + list := make([]v1.TenantApiKeySelectItem, len(keys)) + for i, k := range keys { + list[i] = v1.TenantApiKeySelectItem{ + ID: k.ID, + Name: k.Name, + KeyPrefix: k.KeyPrefix, + UserID: k.UserID, + Status: k.Status, + } + } + return &v1.TenantApiKeySelectRes{List: list, Total: total, Page: req.Page, PageSize: req.PageSize}, nil +} diff --git a/internal/logic/common/audit.go b/internal/logic/common/audit.go index 7f43873..96adcc0 100644 --- a/internal/logic/common/audit.go +++ b/internal/logic/common/audit.go @@ -50,12 +50,17 @@ func GetAuditLevels(ctx context.Context, tenantID int64) (globalLevel, tenantLev // GetTenantAuditLevel 从 tnt_tenants.settings JSONB 中读取租户自身设置的审计级别。 // 未设置时返回空字符串，不回退到全局级别。 func GetTenantAuditLevel(ctx context.Context, tenantID int64) string { - var settingsJSON string - err := dao.TntTenants.Ctx(ctx). + // 读单列字符串必须用 Value()——gdb 的 Model.Scan 只接受 struct 系列， + // 传入 *string 直接报错，会导致租户审计级别永远读不到 + v, err := dao.TntTenants.Ctx(ctx). Where("id", tenantID). Fields("settings"). - Scan(&settingsJSON) - if err != nil || settingsJSON == "" { + Value() + settingsJSON := "" + if err == nil { + settingsJSON = v.String() + } + if settingsJSON == "" { return "" } var settings struct { diff --git a/internal/logic/common/channel_debug_log_writer.go b/internal/logic/common/channel_debug_log_writer.go new file mode 100644 index 0000000..ddec642 --- /dev/null +++ b/internal/logic/common/channel_debug_log_writer.go @@ -0,0 +1,30 @@ +package common + +import ( + "time" +) + +// DefaultChannelDebugLogWriter 渠道调试日志全局写入器 +var DefaultChannelDebugLogWriter *UsageLogWriter + +// InitChannelDebugLogWriter 初始化渠道调试日志写入器。 +// 单条记录可能达数 MB（完整四段报文）：小批量 + 短间隔切批，队列满时丢弃并计数 +// （调试日志允许丢，不反压请求；管理员可重发请求补录）。 +func InitChannelDebugLogWriter() { + DefaultChannelDebugLogWriter = NewUsageLogWriter(UsageLogWriterConfig{ + Table: "chn_debug_logs", + QueueSize: 256, + BatchSize: 4, + FlushInterval: 2 * time.Second, + Workers: 2, + Overflow: OverflowDrop, + }) + DefaultChannelDebugLogWriter.Start() +} + +// CloseChannelDebugLogWriter 关闭渠道调试日志写入器 +func CloseChannelDebugLogWriter() { + if DefaultChannelDebugLogWriter != nil { + DefaultChannelDebugLogWriter.Close() + } +} diff --git a/internal/logic/relay/debug_log.go b/internal/logic/relay/debug_log.go new file mode 100644 index 0000000..a0b7d8f --- /dev/null +++ b/internal/logic/relay/debug_log.go @@ -0,0 +1,93 @@ +package relay + +import ( + "context" + "encoding/json" + + relaycommon "github.com/qianfree/team-api/relay/common" + + "github.com/qianfree/team-api/internal/logic/common" + "github.com/qianfree/team-api/internal/model/do" +) + +// InitDebugLogRecorder 注入渠道调试日志提交钩子（进程启动时调用一次）。 +// relay 层通过包级钩子解耦对 internal/logic 的依赖（仿 billing webhook 钩子模式）。 +func InitDebugLogRecorder() { + relaycommon.SubmitDebugLog = submitDebugLog +} + +// submitDebugLog 将四段报文记录映射为 chn_debug_logs 行并投递批量写入器。 +// body 经 EncodeBody 处理二进制（base64），headers 序列化为 JSON 串写入 JSONB 列。 +func submitDebugLog(_ context.Context, record *relaycommon.DebugLogRecord) { + if record == nil || common.DefaultChannelDebugLogWriter == nil { + return + } + + clientReqBody, clientReqEnc := relaycommon.EncodeBody(record.ClientReqBody) + upstreamReqBody, upstreamReqEnc := relaycommon.EncodeBody(record.UpstreamReqBody) + upstreamRespBody, upstreamRespEnc := relaycommon.EncodeBody(record.UpstreamRespBody) + clientRespBody, clientRespEnc := relaycommon.EncodeBody(record.ClientRespBody) + + insert := do.ChnDebugLogs{ + ChannelId: record.ChannelID, + ChannelName: record.ChannelName, + ChannelType: record.ChannelType, + RequestId: record.RequestID, + TenantId: record.TenantID, + UserId: record.UserID, + ApiKeyId: record.ApiKeyID, + ModelName: record.ModelName, + UpstreamModel: record.UpstreamModel, + RelayMode: record.RelayMode, + InboundPath: record.InboundPath, + UpstreamUrl: record.UpstreamURL, + IsStream: record.IsStream, + RetryIndex: record.RetryIndex, + IsFinal: record.IsFinal, + ClientStatusCode: record.ClientStatusCode, + Error: record.Error, + ClientReqHeaders: headersJSON(record.ClientReqHeaders), + ClientReqBody: clientReqBody, + ClientReqEncoding: clientReqEnc, + UpstreamReqHeaders: headersJSON(record.UpstreamReqHeaders), + UpstreamReqBody: upstreamReqBody, + UpstreamReqEncoding: upstreamReqEnc, + UpstreamRespHeaders: headersJSON(record.UpstreamRespHeaders), + UpstreamRespBody: upstreamRespBody, + UpstreamRespEncoding: upstreamRespEnc, + ClientRespHeaders: headersJSON(record.ClientRespHeaders), + ClientRespBody: clientRespBody, + ClientRespEncoding: clientRespEnc, + UpstreamLatencyMs: record.UpstreamLatencyMs, + TotalLatencyMs: record.TotalLatencyMs, + FirstTokenMs: record.FirstTokenMs, + ClientReqBytes: int64(len(record.ClientReqBody)), + UpstreamReqBytes: int64(len(record.UpstreamReqBody)), + UpstreamRespBytes: int64(len(record.UpstreamRespBody)), + ClientRespBytes: int64(len(record.ClientRespBody)), + } + // 0 表示未发起请求/连接失败，落库为 NULL 以区分真实状态码 0 之外的语义 + if record.UpstreamStatusCode > 0 { + insert.UpstreamStatusCode = record.UpstreamStatusCode + } + // 协议转换信息 → JSONB + if record.Conversion != nil { + if b, err := json.Marshal(record.Conversion); err == nil { + insert.Conversion = string(b) + } + } + + common.DefaultChannelDebugLogWriter.Submit(insert) +} + +// headersJSON headers map 序列化为 JSON 串（JSONB 列）；nil 时写 "null"（与审计日志同口径） +func headersJSON(h map[string]string) any { + if h == nil { + return "null" + } + b, err := json.Marshal(h) + if err != nil { + return "null" + } + return string(b) +} diff --git a/internal/model/do/chn_debug_logs.go b/internal/model/do/chn_debug_logs.go new file mode 100644 index 0000000..4e8cc06 --- /dev/null +++ b/internal/model/do/chn_debug_logs.go @@ -0,0 +1,55 @@ +// ================================================================================= +// Code generated and maintained by GoFrame CLI tool. DO NOT EDIT. +// ================================================================================= + +package do + +import ( + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" +) + +// ChnDebugLogs is the golang structure of table chn_debug_logs for DAO operations like Where/Data. +type ChnDebugLogs struct { + g.Meta `orm:"table:chn_debug_logs, do:true"` + Id any // 主键ID + ChannelId any // 渠道ID（逻辑关联 chn_channels.id，无外键） + ChannelName any // 渠道名称（冗余存储，渠道删除后仍可辨识） + ChannelType any // 渠道类型（ProviderType 枚举值） + RequestId any // 请求唯一ID（同一请求多次重试共享） + TenantId any // 租户ID + UserId any // 用户ID + ApiKeyId any // API Key ID + ModelName any // 用户请求的模型名 + UpstreamModel any // 上游实际使用的模型名（模型映射后） + RelayMode any // 转发模式（chat_completions/claude_messages/embeddings 等） + InboundPath any // 入站端点路径（如 /v1/chat/completions） + UpstreamUrl any // 上游请求 URL（query 中的凭证参数已脱敏） + IsStream any // 是否流式请求 + RetryIndex any // 重试轮次（0=首次尝试） + IsFinal any // 是否为产生客户端响应的最终尝试（成功/终止/流中断） + UpstreamStatusCode any // 上游 HTTP 状态码（未发起请求或连接失败为 NULL） + ClientStatusCode any // 返回客户端的状态码 + Error any // 本尝试的错误信息（成功为空） + ClientReqHeaders any // 段1 客户端请求头（凭证类头脱敏：前6后4）JSON + ClientReqBody any // 段1 客户端请求体（完整不截断；二进制为 base64，见 encoding 列） + ClientReqEncoding any // 段1 编码：plain / base64 + UpstreamReqHeaders any // 段2 发往上游的最终请求头（协议转换+override 后，凭证类脱敏）JSON + UpstreamReqBody any // 段2 发往上游的请求体（实际发送字节，含协议转换；二进制为 base64） + UpstreamReqEncoding any // 段2 编码：plain / base64 + UpstreamRespHeaders any // 段3 上游响应头 JSON（Content-Encoding 可能已被 net/http 透明解压移除） + UpstreamRespBody any // 段3 上游响应体（Go 透明解压后的字节；流式为 SSE 原文；二进制为 base64） + UpstreamRespEncoding any // 段3 编码：plain / base64 + ClientRespHeaders any // 段4 返回客户端的响应头 JSON + ClientRespBody any // 段4 返回客户端的响应体（协议转换后，完整不截断；二进制为 base64） + ClientRespEncoding any // 段4 编码：plain / base64 + UpstreamLatencyMs any // 上游往返耗时（RoundTrip 到响应头）毫秒 + TotalLatencyMs any // 请求总耗时毫秒 + FirstTokenMs any // 首字节延迟毫秒 + ClientReqBytes any // 段1 请求体原始字节数（base64 落库膨胀前的真实大小） + UpstreamReqBytes any // 段2 请求体原始字节数 + UpstreamRespBytes any // 段3 响应体原始字节数 + ClientRespBytes any // 段4 响应体原始字节数 + CreatedAt *gtime.Time // 创建时间 + Conversion any // 协议转换信息 JSON：client_format 客户端协议 / upstream_format 上游协议 / chain 请求侧转换链 / bridge 桥接方式（responses_api|responses_direct|pass_through，空=常规转换） +} diff --git a/internal/model/entity/chn_debug_logs.go b/internal/model/entity/chn_debug_logs.go new file mode 100644 index 0000000..1f22193 --- /dev/null +++ b/internal/model/entity/chn_debug_logs.go @@ -0,0 +1,53 @@ +// ================================================================================= +// Code generated and maintained by GoFrame CLI tool. DO NOT EDIT. +// ================================================================================= + +package entity + +import ( + "github.com/gogf/gf/v2/os/gtime" +) + +// ChnDebugLogs is the golang structure for table chn_debug_logs. +type ChnDebugLogs struct { + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + ChannelId int64 `json:"channel_id" orm:"channel_id" description:"渠道ID（逻辑关联 chn_channels.id，无外键）"` // 渠道ID（逻辑关联 chn_channels.id，无外键） + ChannelName string `json:"channel_name" orm:"channel_name" description:"渠道名称（冗余存储，渠道删除后仍可辨识）"` // 渠道名称（冗余存储，渠道删除后仍可辨识） + ChannelType int `json:"channel_type" orm:"channel_type" description:"渠道类型（ProviderType 枚举值）"` // 渠道类型（ProviderType 枚举值） + RequestId string `json:"request_id" orm:"request_id" description:"请求唯一ID（同一请求多次重试共享）"` // 请求唯一ID（同一请求多次重试共享） + TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID + UserId int64 `json:"user_id" orm:"user_id" description:"用户ID"` // 用户ID + ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"API Key ID"` // API Key ID + ModelName string `json:"model_name" orm:"model_name" description:"用户请求的模型名"` // 用户请求的模型名 + UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:"上游实际使用的模型名（模型映射后）"` // 上游实际使用的模型名（模型映射后） + RelayMode string `json:"relay_mode" orm:"relay_mode" description:"转发模式（chat_completions/claude_messages/embeddings 等）"` // 转发模式（chat_completions/claude_messages/embeddings 等） + InboundPath string `json:"inbound_path" orm:"inbound_path" description:"入站端点路径（如 /v1/chat/completions）"` // 入站端点路径（如 /v1/chat/completions） + UpstreamUrl string `json:"upstream_url" orm:"upstream_url" description:"上游请求 URL（query 中的凭证参数已脱敏）"` // 上游请求 URL（query 中的凭证参数已脱敏） + IsStream bool `json:"is_stream" orm:"is_stream" description:"是否流式请求"` // 是否流式请求 + RetryIndex int `json:"retry_index" orm:"retry_index" description:"重试轮次（0=首次尝试）"` // 重试轮次（0=首次尝试） + IsFinal bool `json:"is_final" orm:"is_final" description:"是否为产生客户端响应的最终尝试（成功/终止/流中断）"` // 是否为产生客户端响应的最终尝试（成功/终止/流中断） + UpstreamStatusCode int `json:"upstream_status_code" orm:"upstream_status_code" description:"上游 HTTP 状态码（未发起请求或连接失败为 NULL）"` // 上游 HTTP 状态码（未发起请求或连接失败为 NULL） + ClientStatusCode int `json:"client_status_code" orm:"client_status_code" description:"返回客户端的状态码"` // 返回客户端的状态码 + Error string `json:"error" orm:"error" description:"本尝试的错误信息（成功为空）"` // 本尝试的错误信息（成功为空） + ClientReqHeaders string `json:"client_req_headers" orm:"client_req_headers" description:"段1 客户端请求头（凭证类头脱敏：前6后4）JSON"` // 段1 客户端请求头（凭证类头脱敏：前6后4）JSON + ClientReqBody string `json:"client_req_body" orm:"client_req_body" description:"段1 客户端请求体（完整不截断；二进制为 base64，见 encoding 列）"` // 段1 客户端请求体（完整不截断；二进制为 base64，见 encoding 列） + ClientReqEncoding string `json:"client_req_encoding" orm:"client_req_encoding" description:"段1 编码：plain / base64"` // 段1 编码：plain / base64 + UpstreamReqHeaders string `json:"upstream_req_headers" orm:"upstream_req_headers" description:"段2 发往上游的最终请求头（协议转换+override 后，凭证类脱敏）JSON"` // 段2 发往上游的最终请求头（协议转换+override 后，凭证类脱敏）JSON + UpstreamReqBody string `json:"upstream_req_body" orm:"upstream_req_body" description:"段2 发往上游的请求体（实际发送字节，含协议转换；二进制为 base64）"` // 段2 发往上游的请求体（实际发送字节，含协议转换；二进制为 base64） + UpstreamReqEncoding string `json:"upstream_req_encoding" orm:"upstream_req_encoding" description:"段2 编码：plain / base64"` // 段2 编码：plain / base64 + UpstreamRespHeaders string `json:"upstream_resp_headers" orm:"upstream_resp_headers" description:"段3 上游响应头 JSON（Content-Encoding 可能已被 net/http 透明解压移除）"` // 段3 上游响应头 JSON（Content-Encoding 可能已被 net/http 透明解压移除） + UpstreamRespBody string `json:"upstream_resp_body" orm:"upstream_resp_body" description:"段3 上游响应体（Go 透明解压后的字节；流式为 SSE 原文；二进制为 base64）"` // 段3 上游响应体（Go 透明解压后的字节；流式为 SSE 原文；二进制为 base64） + UpstreamRespEncoding string `json:"upstream_resp_encoding" orm:"upstream_resp_encoding" description:"段3 编码：plain / base64"` // 段3 编码：plain / base64 + ClientRespHeaders string `json:"client_resp_headers" orm:"client_resp_headers" description:"段4 返回客户端的响应头 JSON"` // 段4 返回客户端的响应头 JSON + ClientRespBody string `json:"client_resp_body" orm:"client_resp_body" description:"段4 返回客户端的响应体（协议转换后，完整不截断；二进制为 base64）"` // 段4 返回客户端的响应体（协议转换后，完整不截断；二进制为 base64） + ClientRespEncoding string `json:"client_resp_encoding" orm:"client_resp_encoding" description:"段4 编码：plain / base64"` // 段4 编码：plain / base64 + UpstreamLatencyMs int `json:"upstream_latency_ms" orm:"upstream_latency_ms" description:"上游往返耗时（RoundTrip 到响应头）毫秒"` // 上游往返耗时（RoundTrip 到响应头）毫秒 + TotalLatencyMs int `json:"total_latency_ms" orm:"total_latency_ms" description:"请求总耗时毫秒"` // 请求总耗时毫秒 + FirstTokenMs int `json:"first_token_ms" orm:"first_token_ms" description:"首字节延迟毫秒"` // 首字节延迟毫秒 + ClientReqBytes int64 `json:"client_req_bytes" orm:"client_req_bytes" description:"段1 请求体原始字节数（base64 落库膨胀前的真实大小）"` // 段1 请求体原始字节数（base64 落库膨胀前的真实大小） + UpstreamReqBytes int64 `json:"upstream_req_bytes" orm:"upstream_req_bytes" description:"段2 请求体原始字节数"` // 段2 请求体原始字节数 + UpstreamRespBytes int64 `json:"upstream_resp_bytes" orm:"upstream_resp_bytes" description:"段3 响应体原始字节数"` // 段3 响应体原始字节数 + ClientRespBytes int64 `json:"client_resp_bytes" orm:"client_resp_bytes" description:"段4 响应体原始字节数"` // 段4 响应体原始字节数 + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + Conversion string `json:"conversion" orm:"conversion" description:"协议转换信息 JSON：client_format 客户端协议 / upstream_format 上游协议 / chain 请求侧转换链 / bridge 桥接方式（responses_api|responses_direct|pass_through，空=常规转换）"` // 协议转换信息 JSON：client_format 客户端协议 / upstream_format 上游协议 / chain 请求侧转换链 / bridge 桥接方式（responses_api|responses_direct|pass_through，空=常规转换） +} diff --git a/internal/service/admin.go b/internal/service/admin.go index 0b9246b..af2d0b2 100644 --- a/internal/service/admin.go +++ b/internal/service/admin.go @@ -124,6 +124,16 @@ type ( ResetChannelHealth(ctx context.Context, req *v1.ChannelResetHealthReq) (*v1.ChannelResetHealthRes, error) // ExportChannels exports channel list to CSV or Excel. ExportChannels(ctx context.Context, req *v1.ChannelExportReq) (*v1.ChannelExportRes, error) + // ChannelDebugLogList 渠道调试日志列表（原生 SQL：octet_length 表达式经 gdb Fields 有引号化风险） + ChannelDebugLogList(ctx context.Context, req *v1.ChannelDebugLogListReq) (*v1.ChannelDebugLogListRes, error) + // ChannelDebugLogStats 渠道调试日志统计（条数/落库体积/最早时间，供清理提示） + ChannelDebugLogStats(ctx context.Context, req *v1.ChannelDebugLogStatsReq) (*v1.ChannelDebugLogStatsRes, error) + // ChannelDebugLogDetail 渠道调试日志详情（四段完整报文；headers JSONB 解析为对象便于前端渲染） + ChannelDebugLogDetail(ctx context.Context, req *v1.ChannelDebugLogDetailReq) (*v1.ChannelDebugLogDetailRes, error) + // ChannelDebugLogDelete 删除单条调试日志（硬删除） + ChannelDebugLogDelete(ctx context.Context, req *v1.ChannelDebugLogDeleteReq) (*v1.ChannelDebugLogDeleteRes, error) + // ChannelDebugLogClear 清空渠道调试日志（按渠道硬删除全部） + ChannelDebugLogClear(ctx context.Context, req *v1.ChannelDebugLogClearReq) (*v1.ChannelDebugLogClearRes, error) // ChannelErrorEventList 渠道错误事件列表 ChannelErrorEventList(ctx context.Context, req *v1.ChannelErrorEventListReq) (*v1.ChannelErrorEventListRes, error) // ChannelErrorStats 渠道错误统计 @@ -140,7 +150,9 @@ type ( ChannelOAuthExchange(ctx context.Context, req *v1.ChannelOAuthExchangeReq) (*v1.ChannelOAuthExchangeRes, error) // ChannelOAuthRefresh 手动刷新 OAuth 令牌 ChannelOAuthRefresh(ctx context.Context, req *v1.ChannelOAuthRefreshReq) (*v1.ChannelOAuthRefreshRes, error) - // TestChannel 测试渠道可用性（发送最小请求验证） + // TestChannel 测试渠道可用性（发送最小请求验证）。 + // 不限制渠道状态：禁用中的渠道也允许手动测试，便于管理员在启用放流量前确认已恢复； + // 自动探测 cron 只查 active 渠道，禁用渠道不会被自动探测。 TestChannel(ctx context.Context, req *v1.ChannelTestReq) (*v1.ChannelTestRes, error) // CronJobList returns all registered cron jobs with their last execution status. CronJobList(ctx context.Context, _ *v1.CronJobListReq) (*v1.CronJobListRes, error) @@ -481,6 +493,9 @@ type ( UpdateTenantChannelScope(ctx context.Context, req *v1.TenantChannelScopeUpdateReq) (*v1.TenantChannelScopeUpdateRes, error) // ExportTenants exports tenant list to CSV or Excel. ExportTenants(ctx context.Context, req *v1.TenantExportReq) (*v1.TenantExportRes, error) + // TenantApiKeySelect 租户 API Key 下拉选择列表（轻量，不返回密钥原值）。 + // 供渠道调试目标过滤等管理端选择器使用；user_id > 0 时按创建者过滤。 + TenantApiKeySelect(ctx context.Context, req *v1.TenantApiKeySelectReq) (*v1.TenantApiKeySelectRes, error) ListTenantLevelConfigs(ctx context.Context, _ *v1.TenantLevelConfigListReq) (*v1.TenantLevelConfigListRes, error) CreateTenantLevelConfig(ctx context.Context, req *v1.TenantLevelConfigCreateReq) (*v1.TenantLevelConfigCreateRes, error) UpdateTenantLevelConfig(ctx context.Context, req *v1.TenantLevelConfigUpdateReq) (*v1.TenantLevelConfigUpdateRes, error) diff --git a/internal/service/tenant.go b/internal/service/tenant.go index 229d1ec..12a5ad1 100644 --- a/internal/service/tenant.go +++ b/internal/service/tenant.go @@ -155,6 +155,7 @@ type ( // ListMemberApiKeys returns a paginated list of API keys belonging to a specific member. ListMemberApiKeys(ctx context.Context, req *v1.TenantMemberApiKeysReq) (*v1.TenantMemberApiKeysRes, error) // ExportMembers exports the tenant member list as CSV or Excel. + // 导出列与成员列表页表格保持一致（用户/角色/状态/额度限制/可用模型/本月消费/加入时间/最后更新）。 ExportMembers(ctx context.Context, req *v1.TenantMemberExportReq) (*v1.TenantMemberExportRes, error) // MemberImport parses CSV content, validates, creates an import record. MemberImport(ctx context.Context, req *v1.TenantMemberImportReq) (*v1.TenantMemberImportRes, error) diff --git a/migrations/000017_channel_debug_logs.sql b/migrations/000017_channel_debug_logs.sql new file mode 100644 index 0000000..37a9693 --- /dev/null +++ b/migrations/000017_channel_debug_logs.sql @@ -0,0 +1,94 @@ +-- +goose Up +-- 渠道调试日志：渠道调试开关开启时，记录经该渠道每次请求尝试（per-attempt，failover 各自成一条） +-- 的四段完整报文（客户端↔系统↔上游）。调试用途：body 不截断、无自动过期，由管理员手动删除/清空。 +-- 不按月分区（需按渠道硬删，分区反而碍事）；BRIN 应对按时间扫描。 + +CREATE TABLE IF NOT EXISTS chn_debug_logs ( + id BIGSERIAL PRIMARY KEY, + channel_id BIGINT NOT NULL, + channel_name VARCHAR(128), + channel_type INT, + request_id VARCHAR(64) NOT NULL DEFAULT '', + tenant_id BIGINT, + user_id BIGINT, + api_key_id BIGINT, + model_name VARCHAR(128), + upstream_model VARCHAR(128), + relay_mode VARCHAR(32), + inbound_path VARCHAR(128), + upstream_url TEXT, + is_stream BOOLEAN NOT NULL DEFAULT FALSE, + retry_index INT NOT NULL DEFAULT 0, + is_final BOOLEAN NOT NULL DEFAULT TRUE, + upstream_status_code INT, + client_status_code INT, + error TEXT, + client_req_headers JSONB, + client_req_body TEXT, + client_req_encoding VARCHAR(8) NOT NULL DEFAULT 'plain', + upstream_req_headers JSONB, + upstream_req_body TEXT, + upstream_req_encoding VARCHAR(8) NOT NULL DEFAULT 'plain', + upstream_resp_headers JSONB, + upstream_resp_body TEXT, + upstream_resp_encoding VARCHAR(8) NOT NULL DEFAULT 'plain', + client_resp_headers JSONB, + client_resp_body TEXT, + client_resp_encoding VARCHAR(8) NOT NULL DEFAULT 'plain', + upstream_latency_ms INT, + total_latency_ms INT, + first_token_ms INT, + client_req_bytes BIGINT, + upstream_req_bytes BIGINT, + upstream_resp_bytes BIGINT, + client_resp_bytes BIGINT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_chn_debug_logs_channel_id ON chn_debug_logs (channel_id, id DESC); +CREATE INDEX IF NOT EXISTS idx_chn_debug_logs_request_id ON chn_debug_logs (request_id); +CREATE INDEX IF NOT EXISTS idx_chn_debug_logs_created_brin ON chn_debug_logs USING BRIN (created_at); + +COMMENT ON TABLE chn_debug_logs IS '渠道调试日志：调试开关开启的渠道，per-attempt 记录客户端↔系统↔上游四段完整报文（头/体/编码），敏感凭证脱敏，二进制体 base64；body 不截断、无自动过期，按渠道手动清理'; +COMMENT ON COLUMN chn_debug_logs.id IS '主键ID'; +COMMENT ON COLUMN chn_debug_logs.channel_id IS '渠道ID（逻辑关联 chn_channels.id，无外键）'; +COMMENT ON COLUMN chn_debug_logs.channel_name IS '渠道名称（冗余存储，渠道删除后仍可辨识）'; +COMMENT ON COLUMN chn_debug_logs.channel_type IS '渠道类型（ProviderType 枚举值）'; +COMMENT ON COLUMN chn_debug_logs.request_id IS '请求唯一ID（同一请求多次重试共享）'; +COMMENT ON COLUMN chn_debug_logs.tenant_id IS '租户ID'; +COMMENT ON COLUMN chn_debug_logs.user_id IS '用户ID'; +COMMENT ON COLUMN chn_debug_logs.api_key_id IS 'API Key ID'; +COMMENT ON COLUMN chn_debug_logs.model_name IS '用户请求的模型名'; +COMMENT ON COLUMN chn_debug_logs.upstream_model IS '上游实际使用的模型名（模型映射后）'; +COMMENT ON COLUMN chn_debug_logs.relay_mode IS '转发模式（chat_completions/claude_messages/embeddings 等）'; +COMMENT ON COLUMN chn_debug_logs.inbound_path IS '入站端点路径（如 /v1/chat/completions）'; +COMMENT ON COLUMN chn_debug_logs.upstream_url IS '上游请求 URL（query 中的凭证参数已脱敏）'; +COMMENT ON COLUMN chn_debug_logs.is_stream IS '是否流式请求'; +COMMENT ON COLUMN chn_debug_logs.retry_index IS '重试轮次（0=首次尝试）'; +COMMENT ON COLUMN chn_debug_logs.is_final IS '是否为产生客户端响应的最终尝试（成功/终止/流中断）'; +COMMENT ON COLUMN chn_debug_logs.upstream_status_code IS '上游 HTTP 状态码（未发起请求或连接失败为 NULL）'; +COMMENT ON COLUMN chn_debug_logs.client_status_code IS '返回客户端的状态码'; +COMMENT ON COLUMN chn_debug_logs.error IS '本尝试的错误信息（成功为空）'; +COMMENT ON COLUMN chn_debug_logs.client_req_headers IS '段1 客户端请求头（凭证类头脱敏：前6后4）JSON'; +COMMENT ON COLUMN chn_debug_logs.client_req_body IS '段1 客户端请求体（完整不截断；二进制为 base64，见 encoding 列）'; +COMMENT ON COLUMN chn_debug_logs.client_req_encoding IS '段1 编码：plain / base64'; +COMMENT ON COLUMN chn_debug_logs.upstream_req_headers IS '段2 发往上游的最终请求头（协议转换+override 后，凭证类脱敏）JSON'; +COMMENT ON COLUMN chn_debug_logs.upstream_req_body IS '段2 发往上游的请求体（实际发送字节，含协议转换；二进制为 base64）'; +COMMENT ON COLUMN chn_debug_logs.upstream_req_encoding IS '段2 编码：plain / base64'; +COMMENT ON COLUMN chn_debug_logs.upstream_resp_headers IS '段3 上游响应头 JSON（Content-Encoding 可能已被 net/http 透明解压移除）'; +COMMENT ON COLUMN chn_debug_logs.upstream_resp_body IS '段3 上游响应体（Go 透明解压后的字节；流式为 SSE 原文；二进制为 base64）'; +COMMENT ON COLUMN chn_debug_logs.upstream_resp_encoding IS '段3 编码：plain / base64'; +COMMENT ON COLUMN chn_debug_logs.client_resp_headers IS '段4 返回客户端的响应头 JSON'; +COMMENT ON COLUMN chn_debug_logs.client_resp_body IS '段4 返回客户端的响应体（协议转换后，完整不截断；二进制为 base64）'; +COMMENT ON COLUMN chn_debug_logs.client_resp_encoding IS '段4 编码：plain / base64'; +COMMENT ON COLUMN chn_debug_logs.upstream_latency_ms IS '上游往返耗时（RoundTrip 到响应头）毫秒'; +COMMENT ON COLUMN chn_debug_logs.total_latency_ms IS '请求总耗时毫秒'; +COMMENT ON COLUMN chn_debug_logs.first_token_ms IS '首字节延迟毫秒'; +COMMENT ON COLUMN chn_debug_logs.client_req_bytes IS '段1 请求体原始字节数（base64 落库膨胀前的真实大小）'; +COMMENT ON COLUMN chn_debug_logs.upstream_req_bytes IS '段2 请求体原始字节数'; +COMMENT ON COLUMN chn_debug_logs.upstream_resp_bytes IS '段3 响应体原始字节数'; +COMMENT ON COLUMN chn_debug_logs.client_resp_bytes IS '段4 响应体原始字节数'; +COMMENT ON COLUMN chn_debug_logs.created_at IS '创建时间'; + +-- +goose Down +DROP TABLE IF EXISTS chn_debug_logs; diff --git a/migrations/000018_channel_debug_logs_conversion.sql b/migrations/000018_channel_debug_logs_conversion.sql new file mode 100644 index 0000000..ffbcd1a --- /dev/null +++ b/migrations/000018_channel_debug_logs_conversion.sql @@ -0,0 +1,11 @@ +-- +goose Up +-- 渠道调试日志：新增协议转换信息列。 +-- 记录客户端协议 →（relaykit 转换链）→ 上游协议及桥接方式（responses_api 桥接 / responses 直连 / 直传）， +-- 供排查协议转换类问题（字段丢失/格式不符）时与四段报文对照。 + +ALTER TABLE chn_debug_logs ADD COLUMN IF NOT EXISTS conversion JSONB; + +COMMENT ON COLUMN chn_debug_logs.conversion IS '协议转换信息 JSON：client_format 客户端协议 / upstream_format 上游协议 / chain 请求侧转换链 / bridge 桥接方式（responses_api|responses_direct|pass_through，空=常规转换）'; + +-- +goose Down +ALTER TABLE chn_debug_logs DROP COLUMN IF EXISTS conversion; diff --git a/relay/common/debug_capture.go b/relay/common/debug_capture.go new file mode 100644 index 0000000..fdfccaf --- /dev/null +++ b/relay/common/debug_capture.go @@ -0,0 +1,340 @@ +package common + +import ( + "bytes" + "context" + "encoding/base64" + "io" + "net/http" + "net/url" + "strings" + "sync" + "time" + "unicode/utf8" +) + +// 渠道调试日志传输层捕获：仅在渠道调试开关开启时，经 ctx 携带捕获器镜像 +// 「系统→远程」「远程→系统」两段报文（最终形态，含协议转换与 header override 后的结果）。 +// 开关关闭的请求 ctx 中无捕获器，DebugRoundTripper 直接透传，热路径零开销。 + +type debugAttemptCtxKey struct{} + +// WithDebugAttempt 将本次尝试的捕获器挂入 ctx（作用域仅限当前 attempt，勿跨重试轮次复用） +func WithDebugAttempt(ctx context.Context, a *DebugAttemptCapture) context.Context { + return context.WithValue(ctx, debugAttemptCtxKey{}, a) +} + +// DebugAttemptFromContext 取出当前 ctx 的捕获器；未开启调试时返回 nil +func DebugAttemptFromContext(ctx context.Context) *DebugAttemptCapture { + if ctx == nil { + return nil + } + a, _ := ctx.Value(debugAttemptCtxKey{}).(*DebugAttemptCapture) + return a +} + +// DebugAttemptCapture 单次上游尝试的传输层捕获器。tee 完整累积请求/响应字节，不截断 +// （调试用途，可能含 base64 图片；靠管理端提示及时清理控制数据量）。 +type DebugAttemptCapture struct { + mu sync.Mutex + + upstreamMethod string + upstreamURL string // query 中凭证参数已脱敏 + upstreamReqHeaders map[string]string + upstreamRespHeaders map[string]string + + upstreamReqBody []byte + upstreamRespBody []byte + + upstreamStatus int + upstreamError string + upstreamLatencyMs int64 +} + +// captureUpstreamRequest 快照最终上游请求并替换 Body 为 tee（在 RoundTrip 内调用） +func (a *DebugAttemptCapture) captureUpstreamRequest(req *http.Request) { + a.mu.Lock() + a.upstreamMethod = req.Method + a.upstreamURL = MaskURLSecret(req.URL.String()) + a.upstreamReqHeaders = MaskHeaders(req.Header) + a.mu.Unlock() + if req.Body != nil { + req.Body = newDebugTee(req.Body, a, &a.upstreamReqBody) + } +} + +// captureUpstreamResponse 快照上游响应并替换 Body 为 tee +func (a *DebugAttemptCapture) captureUpstreamResponse(resp *http.Response) { + a.mu.Lock() + a.upstreamStatus = resp.StatusCode + a.upstreamRespHeaders = MaskHeaders(resp.Header) + a.mu.Unlock() + if resp.Body != nil { + resp.Body = newDebugTee(resp.Body, a, &a.upstreamRespBody) + } +} + +func (a *DebugAttemptCapture) captureError(err error) { + a.mu.Lock() + a.upstreamError = err.Error() + a.mu.Unlock() +} + +func (a *DebugAttemptCapture) setLatency(d time.Duration) { + a.mu.Lock() + a.upstreamLatencyMs = d.Milliseconds() + a.mu.Unlock() +} + +// upstreamSnapshot 捕获数据快照（构建落库记录时调用一次；body 切片引用捕获缓冲，此后不再写入） +type upstreamSnapshot struct { + Method string + URL string + ReqHeaders map[string]string + RespHeaders map[string]string + ReqBody []byte + RespBody []byte + Status int + Err string + LatencyMs int64 +} + +func (a *DebugAttemptCapture) snapshot() upstreamSnapshot { + a.mu.Lock() + defer a.mu.Unlock() + return upstreamSnapshot{ + Method: a.upstreamMethod, + URL: a.upstreamURL, + ReqHeaders: a.upstreamReqHeaders, + RespHeaders: a.upstreamRespHeaders, + ReqBody: a.upstreamReqBody, + RespBody: a.upstreamRespBody, + Status: a.upstreamStatus, + Err: a.upstreamError, + LatencyMs: a.upstreamLatencyMs, + } +} + +// debugTeeReadCloser 透传读写并在读取时镜像字节到捕获缓冲 +type debugTeeReadCloser struct { + src io.ReadCloser + a *DebugAttemptCapture + field *[]byte +} + +func newDebugTee(src io.ReadCloser, a *DebugAttemptCapture, field *[]byte) *debugTeeReadCloser { + return &debugTeeReadCloser{src: src, a: a, field: field} +} + +func (t *debugTeeReadCloser) Read(p []byte) (int, error) { + n, err := t.src.Read(p) + if n > 0 { + t.a.mu.Lock() + *t.field = append(*t.field, p[:n]...) + t.a.mu.Unlock() + } + return n, err +} + +func (t *debugTeeReadCloser) Close() error { + return t.src.Close() +} + +// DebugRoundTripper 渠道调试日志传输层包装。ctx 无捕获器时直接透传（一次 ctx 查找）； +// 有捕获器时镜像最终上游请求/响应。连接池与 HTTP/2 仍由底层 *http.Transport 管理。 +type DebugRoundTripper struct { + Base http.RoundTripper +} + +// WrapDebugTransport 包装传输层供 http.Client 的 Transport 字段使用 +func WrapDebugTransport(base *http.Transport) http.RoundTripper { + return &DebugRoundTripper{Base: base} +} + +func (t *DebugRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + a := DebugAttemptFromContext(req.Context()) + if a == nil { + return t.Base.RoundTrip(req) + } + // Clone 后再替换 Body：不污染适配器持有的原始 req（适配器在 client.Do 返回后会 + // Close httpReq.Body，tee 的 Close 透传保持原有语义）。 + // 已知边角：底层 Transport 建连失败内部重放走 GetBody（重读原始 body），会绕过 tee， + // 该次捕获的段2 可能不完整——罕见且不影响请求本身，可接受。 + clone := req.Clone(req.Context()) + a.captureUpstreamRequest(clone) + start := time.Now() + resp, err := t.Base.RoundTrip(clone) + a.setLatency(time.Since(start)) + if err != nil { + a.captureError(err) + return resp, err + } + a.captureUpstreamResponse(resp) + return resp, nil +} + +// DebugClientWriter 渠道调试日志的客户端响应捕获 writer：完整累积、不截断 +// （区别于审计的 ResponseCaptureWriter，后者有 256KB head+tail 截断）。 +// 仅在调试开关开启时由 RelayHandler 包在 rc.Writer 外层，请求结束后由 +// DebugSession.FinalizeAndSubmit 取走数据。 +type DebugClientWriter struct { + http.ResponseWriter + mu sync.Mutex + buf []byte + status int +} + +// NewDebugClientWriter 创建客户端响应捕获 writer +func NewDebugClientWriter(w http.ResponseWriter) *DebugClientWriter { + return &DebugClientWriter{ResponseWriter: w} +} + +// WriteHeader 捕获状态码并委托底层 writer（仅首次生效） +func (w *DebugClientWriter) WriteHeader(code int) { + w.mu.Lock() + if w.status != 0 { + w.mu.Unlock() + return + } + w.status = code + w.mu.Unlock() + w.ResponseWriter.WriteHeader(code) +} + +// Write 委托底层 writer 并完整镜像实际写出的字节 +func (w *DebugClientWriter) Write(b []byte) (int, error) { + w.mu.Lock() + if w.status == 0 { + w.status = http.StatusOK + } + w.mu.Unlock() + + n, err := w.ResponseWriter.Write(b) + + w.mu.Lock() + w.buf = append(w.buf, b[:n]...) + w.mu.Unlock() + return n, err +} + +// Flush 实现 http.Flusher，转发到底层 writer（流式响应依赖） +func (w *DebugClientWriter) Flush() { + if f, ok := w.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +// Bytes 返回已捕获的完整响应体 +func (w *DebugClientWriter) Bytes() []byte { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf +} + +// StatusCode 返回捕获的状态码（未显式设置时按 200） +func (w *DebugClientWriter) StatusCode() int { + w.mu.Lock() + defer w.mu.Unlock() + if w.status == 0 { + return http.StatusOK + } + return w.status +} + +// HeaderSnapshot 从底层 writer 提取当前响应头快照（响应已写出后调用，取最终形态） +func (w *DebugClientWriter) HeaderSnapshot() map[string]string { + headers := make(map[string]string) + for k, vals := range w.ResponseWriter.Header() { + if len(vals) > 0 { + headers[k] = vals[0] + } + } + return headers +} + +// ===== 脱敏与编码 ===== + +// sensitiveHeaderNames 凭证类请求/响应头，落库前脱敏（保留前6后4，足以区分 Key 又不泄露完整值） +var sensitiveHeaderNames = map[string]bool{ + "authorization": true, + "proxy-authorization": true, + "cookie": true, + "set-cookie": true, + "x-api-key": true, + "api-key": true, + "x-goog-api-key": true, + "x-auth-token": true, +} + +// urlSecretParams URL query 中携带凭证的参数名（如 Gemini 的 ?key=） +var urlSecretParams = map[string]bool{ + "key": true, + "api_key": true, + "apikey": true, + "token": true, + "access_token": true, +} + +// MaskSecret 脱敏单个凭证值：长度 < 12 全遮蔽；否则保留前 6 后 4 字符 +func MaskSecret(v string) string { + if len(v) < 12 { + return "****" + } + return v[:6] + "****" + v[len(v)-4:] +} + +// MaskHeaders 头快照并脱敏凭证类头；多值头用 ", " 连接 +func MaskHeaders(h http.Header) map[string]string { + if h == nil { + return nil + } + out := make(map[string]string, len(h)) + for k, vals := range h { + v := strings.Join(vals, ", ") + if sensitiveHeaderNames[strings.ToLower(k)] { + v = MaskSecret(v) + } + out[k] = v + } + return out +} + +// MaskURLSecret 遮蔽 URL query 中的凭证参数值 +func MaskURLSecret(rawURL string) string { + u, err := url.Parse(rawURL) + if err != nil || u.RawQuery == "" { + return rawURL + } + q, err := url.ParseQuery(u.RawQuery) + if err != nil { + return rawURL + } + changed := false + for name, vals := range q { + if !urlSecretParams[strings.ToLower(name)] { + continue + } + for i := range vals { + vals[i] = MaskSecret(vals[i]) + } + changed = true + } + if !changed { + return rawURL + } + // Encode 会把脱敏占位符 * 百分号转义成 %2A（可读性差），还原为字面 * + u.RawQuery = strings.ReplaceAll(q.Encode(), "%2A", "*") + return u.String() +} + +// EncodeBody body 落库编码：有效 UTF-8 且不含 NUL 字节 → 原文 plain； +// 否则（multipart/图片/音频等二进制）→ base64（PostgreSQL TEXT 列不接受 NUL 与无效 UTF-8） +func EncodeBody(b []byte) (string, string) { + if len(b) == 0 { + return "", "plain" + } + if utf8.Valid(b) && !bytes.ContainsRune(b, 0) { + return string(b), "plain" + } + return base64.StdEncoding.EncodeToString(b), "base64" +} diff --git a/relay/common/debug_capture_test.go b/relay/common/debug_capture_test.go new file mode 100644 index 0000000..f521431 --- /dev/null +++ b/relay/common/debug_capture_test.go @@ -0,0 +1,241 @@ +package common + +import ( + "bytes" + "context" + "encoding/base64" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// ===== 脱敏 ===== + +func TestMaskSecret(t *testing.T) { + cases := []struct { + in, want string + }{ + {"short", "****"}, // < 12 全遮蔽 + {"12345678901", "****"}, // 11 位仍全遮蔽 + {"sk-abcdef1234wxyz", "sk-abc****wxyz"}, // 前 6 后 4 + {"", "****"}, // 空值 + {"Bearer sk-abcdefghijk", "Bearer****hijk"}, // 带前缀的完整头值 + } + for _, c := range cases { + if got := MaskSecret(c.in); got != c.want { + t.Errorf("MaskSecret(%q) = %q, want %q", c.in, got, c.want) + } + } +} + +func TestMaskSensitiveHeaders(t *testing.T) { + h := http.Header{} + h.Set("Authorization", "Bearer sk-abcdefghijklmnop") + h.Set("X-Api-Key", "short") + h.Set("Content-Type", "application/json") + h.Add("X-Multi", "a") + h.Add("X-Multi", "b") + + out := MaskHeaders(h) + if got := out["Authorization"]; got != "Bearer****mnop" { + t.Errorf("Authorization 未按前6后4脱敏: %q", got) + } + if got := out["X-Api-Key"]; got != "****" { + t.Errorf("短凭证值应全遮蔽: %q", got) + } + if got := out["Content-Type"]; got != "application/json" { + t.Errorf("非敏感头不应改动: %q", got) + } + if got := out["X-Multi"]; got != "a, b" { + t.Errorf("多值头应逗号连接: %q", got) + } + if MaskHeaders(nil) != nil { + t.Error("nil 头应返回 nil") + } +} + +func TestMaskURLSecret(t *testing.T) { + raw := "https://example.com/v1/models?key=abcdefghijk1234&x=1" + got := MaskURLSecret(raw) + if strings.Contains(got, "abcdefghijk1234") { + t.Errorf("URL query 凭证未脱敏: %q", got) + } + if !strings.Contains(got, "abcdef****1234") { + t.Errorf("脱敏格式不符合前6后4: %q", got) + } + if !strings.Contains(got, "x=1") { + t.Errorf("非凭证参数不应改动: %q", got) + } + // 无凭证参数的 URL 原样返回 + plain := "https://example.com/v1/chat" + if got := MaskURLSecret(plain); got != plain { + t.Errorf("无凭证 URL 被改动: %q", got) + } +} + +// ===== body 编码 ===== + +func TestEncodeBody(t *testing.T) { + // 有效 UTF-8 → plain + data, enc := EncodeBody([]byte(`{"model":"gpt"}`)) + if enc != "plain" || data != `{"model":"gpt"}` { + t.Errorf("UTF-8 body 应原样 plain: %q %q", data, enc) + } + // 含 NUL → base64（PG TEXT 不接受 NUL） + nul := []byte{'a', 0x00, 'b'} + data, enc = EncodeBody(nul) + if enc != "base64" { + t.Errorf("含 NUL body 应 base64: %q", enc) + } + if decoded, err := base64Decode(data); err != nil || !bytes.Equal(decoded, nul) { + t.Errorf("base64 往返不一致: %v %v", decoded, err) + } + // 无效 UTF-8 → base64 + invalid := []byte{0xff, 0xfe, 0xfd} + if _, enc = EncodeBody(invalid); enc != "base64" { + t.Errorf("无效 UTF-8 body 应 base64: %q", enc) + } + // 空 body + if data, enc = EncodeBody(nil); enc != "plain" || data != "" { + t.Errorf("空 body 应 plain 空串: %q %q", data, enc) + } +} + +// ===== DebugClientWriter：不截断 ===== + +func TestDebugClientWriter(t *testing.T) { + rec := httptest.NewRecorder() + w := NewDebugClientWriter(rec) + + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(200) + + // 写 2MB，验证完整累积不截断 + chunk := bytes.Repeat([]byte("x"), 64*1024) + total := 0 + for i := 0; i < 32; i++ { + n, err := w.Write(chunk) + if err != nil || n != len(chunk) { + t.Fatalf("write 失败: n=%d err=%v", n, err) + } + total += n + } + if got := len(w.Bytes()); got != total { + t.Errorf("捕获 %d 字节，期望 %d（不允许截断）", got, total) + } + if w.StatusCode() != 200 { + t.Errorf("状态码 = %d, want 200", w.StatusCode()) + } + if ct := w.HeaderSnapshot()["Content-Type"]; ct != "text/event-stream" { + t.Errorf("响应头快照错误: %q", ct) + } + // 未显式 WriteHeader 时默认 200 + if code := NewDebugClientWriter(httptest.NewRecorder()).StatusCode(); code != 200 { + t.Errorf("默认状态码 = %d, want 200", code) + } +} + +// ===== DebugRoundTripper ===== + +func TestDebugRoundTripperPassthrough(t *testing.T) { + // ctx 无捕获器：直接透传，行为与裸 transport 一致 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("ok")) + })) + defer srv.Close() + + client := &http.Client{Transport: WrapDebugTransport(http.DefaultTransport.(*http.Transport).Clone())} + resp, err := client.Get(srv.URL) + if err != nil { + t.Fatalf("passthrough 请求失败: %v", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if string(body) != "ok" { + t.Errorf("透传响应被篡改: %q", body) + } +} + +func TestDebugRoundTripperCapture(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // 分块写入模拟流式响应，验证 tee 完整镜像 + _, _ = w.Write([]byte(`{"chunk":1}`)) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + _, _ = w.Write([]byte(`{"chunk":2}`)) + })) + defer srv.Close() + + a := &DebugAttemptCapture{} + ctx := WithDebugAttempt(context.Background(), a) + client := &http.Client{Transport: WrapDebugTransport(http.DefaultTransport.(*http.Transport).Clone())} + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+"/v1/chat?key=topsecret12345", + strings.NewReader(`{"model":"gpt","messages":[]}`)) + if err != nil { + t.Fatalf("构建请求失败: %v", err) + } + req.Header.Set("Authorization", "Bearer sk-abcdefghijklmnop") + req.Header.Set("Content-Type", "application/json") + + resp, err := client.Do(req) + if err != nil { + t.Fatalf("请求失败: %v", err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + + snap := a.snapshot() + + // 段2：上游请求捕获（headers 脱敏 + URL 凭证脱敏 + body 完整） + if snap.Method != http.MethodPost { + t.Errorf("method = %q", snap.Method) + } + if !strings.Contains(snap.URL, "topsec****2345") { + t.Errorf("URL 凭证未脱敏: %q", snap.URL) + } + if got := snap.ReqHeaders["Authorization"]; got != "Bearer****mnop" { + t.Errorf("Authorization 未脱敏: %q", got) + } + if string(snap.ReqBody) != `{"model":"gpt","messages":[]}` { + t.Errorf("上游请求体捕获不完整: %q", snap.ReqBody) + } + + // 段3：上游响应捕获（分块写入也应完整） + if snap.Status != 200 { + t.Errorf("status = %d", snap.Status) + } + if !strings.Contains(string(body), `{"chunk":1}`) { + t.Errorf("客户端读取的响应不完整: %q", body) + } + if string(snap.RespBody) != `{"chunk":1}{"chunk":2}` { + t.Errorf("上游响应体镜像不完整: %q", snap.RespBody) + } + if snap.LatencyMs < 0 { + t.Errorf("latency = %d", snap.LatencyMs) + } +} + +func TestDebugRoundTripperError(t *testing.T) { + // 连接失败：err 路径记录 upstreamError，不 panic + a := &DebugAttemptCapture{} + ctx := WithDebugAttempt(context.Background(), a) + client := &http.Client{Transport: WrapDebugTransport(http.DefaultTransport.(*http.Transport).Clone())} + + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "http://127.0.0.1:1/nope", nil) + _, err := client.Do(req) + if err == nil { + t.Fatal("期望连接失败") + } + snap := a.snapshot() + if snap.Err == "" { + t.Error("传输层错误未被记录") + } +} + +func base64Decode(s string) ([]byte, error) { + return base64.StdEncoding.DecodeString(s) +} diff --git a/relay/common/debug_log.go b/relay/common/debug_log.go new file mode 100644 index 0000000..ffb493c --- /dev/null +++ b/relay/common/debug_log.go @@ -0,0 +1,57 @@ +package common + +import "context" + +// DebugLogRecord 渠道调试日志记录（四段报文 + 元数据），经 SubmitDebugLog 钩子异步落库。 +// headers 均已脱敏；body 为原始字节（落库时经 EncodeBody 处理二进制）。 +type DebugLogRecord struct { + ChannelID int64 + ChannelName string + ChannelType int + RequestID string + TenantID int64 + UserID int64 + ApiKeyID int64 + ModelName string + UpstreamModel string + RelayMode string + InboundPath string + UpstreamURL string + IsStream bool + RetryIndex int + IsFinal bool + + UpstreamStatusCode int // 0 = 未发起请求或连接失败（落库为 NULL） + ClientStatusCode int + Error string + + // 四段报文：段1 客户端请求 / 段2 发往上游最终请求 / 段3 上游响应 / 段4 返回客户端响应 + ClientReqHeaders map[string]string + ClientReqBody []byte + UpstreamReqHeaders map[string]string + UpstreamReqBody []byte + UpstreamRespHeaders map[string]string + UpstreamRespBody []byte + ClientRespHeaders map[string]string + ClientRespBody []byte + + UpstreamLatencyMs int64 + TotalLatencyMs int64 + FirstTokenMs int64 + + // Conversion 协议转换信息（客户端协议 → 上游协议及转换链），buildRelayInfo 后由 CaptureProtocol 快照 + Conversion *DebugConversion +} + +// DebugConversion 协议转换信息：追踪客户端协议经哪些格式转换到上游协议。 +// 段3/段4 的协议无需单独记录——段3 = upstream_format，段4 = client_format（推导关系）。 +type DebugConversion struct { + ClientFormat string `json:"client_format"` // 客户端原始协议（openai/claude/gemini/responses） + UpstreamFormat string `json:"upstream_format"` // 发往上游的协议 + Chain []string `json:"chain,omitempty"` // 请求侧转换链（relaykit 填充；为空且两端不同时兜底两端） + Bridge string `json:"bridge,omitempty"` // 桥接方式：responses_api / responses_direct / pass_through，空=常规转换 +} + +// SubmitDebugLog 调试日志提交钩子。relay 层不直接依赖 internal/logic（避免反向导入）， +// 由 internal/logic/relay 在进程启动时注入实现；nil（未初始化/单测）时调用方静默丢弃。 +var SubmitDebugLog func(ctx context.Context, record *DebugLogRecord) diff --git a/relay/common/debug_session.go b/relay/common/debug_session.go new file mode 100644 index 0000000..14efa63 --- /dev/null +++ b/relay/common/debug_session.go @@ -0,0 +1,229 @@ +package common + +import ( + "context" + "net/http" + + "github.com/qianfree/team-api/relay/constant" +) + +// DebugSession 渠道调试日志会话：挂载于 RelayContext.Debug，整请求存活。 +// 重试循环中首个「调试开关开启」的渠道尝试触发创建：捕获段1（客户端请求）并包装段4 +// 的客户端响应 writer（保持到请求结束，中间失败尝试不向客户端写字节，累积字节即最终响应）。 +// 中间失败尝试经 DebugAttempt.Submit 立即落库（is_final=false）；最终尝试经 MarkFinal +// 挂起，由外层 handler 在响应写完后调用 FinalizeAndSubmit 补段4并统一提交（幂等）。 +type DebugSession struct { + requestID string + tenantID int64 + userID int64 + apiKeyID int64 + path string + + clientReqHeaders map[string]string // 段1（脱敏后） + clientReqBody []byte // 段1 完整请求体 + + clientWriter *DebugClientWriter // 段4 捕获 writer + + pending *DebugAttempt // MarkFinal 挂起的最终尝试 + finalized bool +} + +// NewDebugSession 创建调试会话 +func NewDebugSession(requestID string, tenantID, userID, apiKeyID int64, path string) *DebugSession { + return &DebugSession{ + requestID: requestID, + tenantID: tenantID, + userID: userID, + apiKeyID: apiKeyID, + path: path, + } +} + +// CaptureClientRequest 捕获段1：客户端请求头（脱敏）+ 完整请求体 +func (s *DebugSession) CaptureClientRequest(headers http.Header, body []byte) { + s.clientReqHeaders = MaskHeaders(headers) + s.clientReqBody = body +} + +// SetClientWriter 回填段4捕获 writer（由 RelayHandler 包装 rc.Writer 后调用） +func (s *DebugSession) SetClientWriter(w *DebugClientWriter) { + s.clientWriter = w +} + +// BeginAttempt 为当前渠道尝试创建捕获上下文（每轮重试一次） +func (s *DebugSession) BeginAttempt(selection *ChannelSelection, modelName, relayModeStr string, isStream bool, retryIndex int) *DebugAttempt { + return &DebugAttempt{ + session: s, + Capture: &DebugAttemptCapture{}, + channelID: selection.ChannelID, + channelName: selection.ChannelName, + channelType: selection.ChannelType, + modelName: modelName, + upstreamModel: selection.UpstreamModelName, + relayMode: relayModeStr, + isStream: isStream, + retryIndex: retryIndex, + } +} + +// DebugAttempt 单次尝试的调试上下文：会话引用 + 渠道元数据 + 传输层捕获器。 +// Capture 经 WithDebugAttempt 挂入 ctx，由 DebugRoundTripper 填充段2/段3。 +type DebugAttempt struct { + session *DebugSession + Capture *DebugAttemptCapture + + channelID int64 + channelName string + channelType int + modelName string + upstreamModel string + relayMode string + isStream bool + retryIndex int + errText string + conversion *DebugConversion // 协议转换信息（CaptureProtocol 快照） +} + +// CaptureProtocol 快照协议转换信息（RelayHandler 在 buildRelayInfo 之后调用一次）。 +// 上游协议按桥接标志与渠道类型推导：直传 > Responses 桥接 > Responses 直连 > 渠道原生协议。 +func (a *DebugAttempt) CaptureProtocol(info *RelayInfo) { + if a == nil || info == nil { + return + } + conv := &DebugConversion{ + ClientFormat: string(info.GetOriginalClientFormat()), + } + switch { + case info.ChannelMeta != nil && info.ChannelMeta.Settings.PassThroughBodyEnabled: + conv.UpstreamFormat = conv.ClientFormat + conv.Bridge = "pass_through" + case info.UseResponsesAPI: + conv.UpstreamFormat = string(constant.RelayFormatResponses) + conv.Bridge = "responses_api" + case info.ChannelMeta != nil && info.ChannelMeta.UpstreamSpeaksResponses(): + conv.UpstreamFormat = string(constant.RelayFormatResponses) + conv.Bridge = "responses_direct" + default: + conv.UpstreamFormat = nativeFormatOfProvider(a.channelType) + } + for _, f := range info.ConversionChain() { + conv.Chain = append(conv.Chain, string(f)) + } + // relaykit 链未覆盖（直传/同构/adaptor 手写转换）且两端协议不同时，兜底记录两端 + if len(conv.Chain) == 0 && conv.UpstreamFormat != conv.ClientFormat { + conv.Chain = []string{conv.ClientFormat, conv.UpstreamFormat} + } + a.conversion = conv +} + +// nativeFormatOfProvider 渠道类型 → 上游原生协议（展示用途）。 +// openai 兼容系渠道占绝大多数，默认 openai；anthropic / gemini 系例外。 +func nativeFormatOfProvider(providerType int) string { + switch constant.ProviderType(providerType) { + case constant.ProviderClaude: + return string(constant.RelayFormatClaude) + case constant.ProviderGemini, constant.ProviderVertex: + return string(constant.RelayFormatGemini) + default: + return string(constant.RelayFormatOpenAI) + } +} + +// Submit 中间失败尝试：立即经钩子异步落库（is_final=false，段4为空——该尝试未向客户端写出字节） +func (a *DebugAttempt) Submit(err error) { + if a == nil { + return + } + a.errText = debugErrText(err) + if hook := SubmitDebugLog; hook != nil { + hook(context.Background(), a.buildRecord(false, 0, 0)) + } +} + +// MarkFinal 标记为最终尝试（成功/终止/流中断）：挂起等待外层 FinalizeAndSubmit 补段4后提交 +func (a *DebugAttempt) MarkFinal(err error) { + if a == nil { + return + } + a.errText = debugErrText(err) + a.session.pending = a +} + +// FinalizeAndSubmit 补段4并提交最终尝试记录。由外层 handler 在客户端响应写完后调用 +// （recordAudit 收尾处），幂等：重复调用或无挂起尝试时为 no-op。 +func (s *DebugSession) FinalizeAndSubmit(totalLatencyMs, firstTokenMs int64) { + if s == nil || s.finalized { + return + } + s.finalized = true + a := s.pending + if a == nil { + return + } + if hook := SubmitDebugLog; hook != nil { + hook(context.Background(), a.buildRecord(true, totalLatencyMs, firstTokenMs)) + } +} + +// buildRecord 汇总四段与元数据构建落库记录。isFinal=true 时从 clientWriter 补段4。 +func (a *DebugAttempt) buildRecord(isFinal bool, totalLatencyMs, firstTokenMs int64) *DebugLogRecord { + s := a.session + snap := a.Capture.snapshot() + + rec := &DebugLogRecord{ + ChannelID: a.channelID, + ChannelName: a.channelName, + ChannelType: a.channelType, + RequestID: s.requestID, + TenantID: s.tenantID, + UserID: s.userID, + ApiKeyID: s.apiKeyID, + ModelName: a.modelName, + UpstreamModel: a.upstreamModel, + RelayMode: a.relayMode, + InboundPath: s.path, + UpstreamURL: snap.URL, + IsStream: a.isStream, + RetryIndex: a.retryIndex, + IsFinal: isFinal, + + UpstreamStatusCode: snap.Status, + Error: a.errText, + + // 段1（会话级，全部尝试共享） + ClientReqHeaders: s.clientReqHeaders, + ClientReqBody: s.clientReqBody, + + // 段2/段3（传输层捕获） + UpstreamReqHeaders: snap.ReqHeaders, + UpstreamReqBody: snap.ReqBody, + UpstreamRespHeaders: snap.RespHeaders, + UpstreamRespBody: snap.RespBody, + + UpstreamLatencyMs: snap.LatencyMs, + + Conversion: a.conversion, + } + + if snap.Err != "" && rec.Error == "" { + // 传输层错误（连接失败等）未被上层 err 覆盖时兜底保留 + rec.Error = snap.Err + } + + if isFinal && s.clientWriter != nil { + // 段4：客户端响应 + rec.ClientRespHeaders = s.clientWriter.HeaderSnapshot() + rec.ClientRespBody = s.clientWriter.Bytes() + rec.ClientStatusCode = s.clientWriter.StatusCode() + rec.TotalLatencyMs = totalLatencyMs + rec.FirstTokenMs = firstTokenMs + } + return rec +} + +func debugErrText(err error) string { + if err == nil { + return "" + } + return err.Error() +} diff --git a/relay/common/debug_session_test.go b/relay/common/debug_session_test.go new file mode 100644 index 0000000..34dd584 --- /dev/null +++ b/relay/common/debug_session_test.go @@ -0,0 +1,199 @@ +package common + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/qianfree/team-api/relay/constant" + "github.com/qianfree/team-api/relaykit/types" +) + +// 验证调试会话完整生命周期： +// 中间失败尝试 Submit 立即提交（is_final=false、段4为空）， +// 最终尝试 MarkFinal 挂起、FinalizeAndSubmit 补段4后提交，且幂等。 +func TestDebugSessionLifecycle(t *testing.T) { + var records []*DebugLogRecord + origHook := SubmitDebugLog + SubmitDebugLog = func(ctx context.Context, r *DebugLogRecord) { records = append(records, r) } + defer func() { SubmitDebugLog = origHook }() + + s := NewDebugSession("req-1", 1, 2, 3, "/v1/chat/completions") + // 段1：客户端请求（含凭证头，应脱敏） + h := http.Header{} + h.Set("Authorization", "Bearer sk-test-key-12345678") + h.Set("Content-Type", "application/json") + s.CaptureClientRequest(h, []byte(`{"model":"m"}`)) + + selection := &ChannelSelection{ChannelID: 9, ChannelName: "ch9", ChannelType: 1, UpstreamModelName: "up-m"} + + // 第一次尝试失败（中间）：Submit 立即提交 + a1 := s.BeginAttempt(selection, "m", "chat_completions", true, 0) + a1.Submit(errors.New("boom")) + if len(records) != 1 { + t.Fatalf("中间尝试应提交 1 条，实际 %d", len(records)) + } + r1 := records[0] + if r1.IsFinal { + t.Error("中间尝试 is_final 应为 false") + } + if r1.Error != "boom" || r1.ChannelID != 9 || r1.RetryIndex != 0 || !r1.IsStream { + t.Errorf("中间尝试元数据错误: %+v", r1) + } + if string(r1.ClientReqBody) != `{"model":"m"}` { + t.Errorf("段1 请求体错误: %q", r1.ClientReqBody) + } + if r1.ClientReqHeaders["Authorization"] != "Bearer****5678" { + t.Errorf("段1 凭证头未脱敏: %q", r1.ClientReqHeaders["Authorization"]) + } + if r1.ClientRespBody != nil { + t.Errorf("中间尝试段4应为空: %q", r1.ClientRespBody) + } + + // 第二次尝试成功：MarkFinal → 模拟段2捕获 → FinalizeAndSubmit 补段4 + a2 := s.BeginAttempt(selection, "m", "chat_completions", true, 1) + req, _ := http.NewRequest(http.MethodPost, "http://upstream/v1/chat", strings.NewReader("converted-body")) + a2.Capture.captureUpstreamRequest(req) + // 模拟传输层读取请求体 + 上游响应 + _, _ = io.ReadAll(req.Body) + _ = req.Body.Close() + resp := &http.Response{StatusCode: 200, Header: http.Header{"X-Up": []string{"v"}}, Body: io.NopCloser(strings.NewReader("upstream-resp"))} + a2.Capture.captureUpstreamResponse(resp) + _, _ = io.ReadAll(resp.Body) + _ = resp.Body.Close() + + // 段4 writer + rec := httptest.NewRecorder() + dw := NewDebugClientWriter(rec) + s.SetClientWriter(dw) + dw.WriteHeader(200) + _, _ = dw.Write([]byte("client-resp")) + + a2.MarkFinal(nil) + s.FinalizeAndSubmit(123, 45) + + if len(records) != 2 { + t.Fatalf("最终尝试应提交第 2 条，实际 %d", len(records)) + } + r2 := records[1] + if !r2.IsFinal || r2.RetryIndex != 1 || r2.Error != "" { + t.Errorf("最终尝试元数据错误: %+v", r2) + } + if string(r2.UpstreamReqBody) != "converted-body" { + t.Errorf("段2 请求体错误: %q", r2.UpstreamReqBody) + } + if r2.UpstreamStatusCode != 200 || r2.UpstreamRespHeaders["X-Up"] != "v" { + t.Errorf("段3 状态/响应头错误: %d %v", r2.UpstreamStatusCode, r2.UpstreamRespHeaders) + } + if string(r2.UpstreamRespBody) != "upstream-resp" { + t.Errorf("段3 响应体错误: %q", r2.UpstreamRespBody) + } + if string(r2.ClientRespBody) != "client-resp" || r2.ClientStatusCode != 200 { + t.Errorf("段4 错误: %q %d", r2.ClientRespBody, r2.ClientStatusCode) + } + if r2.TotalLatencyMs != 123 || r2.FirstTokenMs != 45 { + t.Errorf("耗时元数据错误: %d %d", r2.TotalLatencyMs, r2.FirstTokenMs) + } + + // 幂等：重复 Finalize 不再提交 + s.FinalizeAndSubmit(1, 1) + if len(records) != 2 { + t.Errorf("FinalizeAndSubmit 应幂等，实际提交 %d 条", len(records)) + } +} + +// nil 会话（开关未开启）调用 FinalizeAndSubmit 应安全 no-op +func TestDebugSessionNilSafe(t *testing.T) { + var s *DebugSession + s.FinalizeAndSubmit(0, 0) // 不应 panic + + var a *DebugAttempt + a.Submit(errors.New("x")) // 不应 panic + a.MarkFinal(nil) // 不应 panic +} + +// 调试目标过滤：AND 组合，0 = 不限 +func TestDebugTargetMatch(t *testing.T) { + s := ChannelSettings{DebugLogEnabled: true} + if !s.DebugTargetMatch(1, 2, 3) { + t.Error("未设过滤器应全部匹配") + } + + s.DebugLogTenantID = 10 + if !s.DebugTargetMatch(10, 99, 99) { + t.Error("匹配租户时应通过") + } + if s.DebugTargetMatch(11, 2, 3) { + t.Error("租户不匹配应拒绝") + } + + s.DebugLogUserID = 20 + if !s.DebugTargetMatch(10, 20, 99) { + t.Error("租户+成员均匹配应通过") + } + if s.DebugTargetMatch(10, 21, 99) { + t.Error("成员不匹配应拒绝") + } + + s.DebugLogApiKeyID = 30 + if !s.DebugTargetMatch(10, 20, 30) { + t.Error("三者均匹配应通过") + } + if s.DebugTargetMatch(10, 20, 31) { + t.Error("密钥不匹配应拒绝") + } +} + +// 协议转换快照：常规转换按渠道类型推导上游协议；桥接标志优先；链路兜底两端 +func TestCaptureProtocol(t *testing.T) { + s := NewDebugSession("req", 1, 2, 3, "/v1/chat/completions") + selection := &ChannelSelection{ChannelID: 9, ChannelType: int(constant.ProviderClaude)} + a := s.BeginAttempt(selection, "m", "chat_completions", false, 0) + + // 客户端 openai → claude 渠道：上游协议 claude，链路兜底两端 + info := &RelayInfo{ + InboundFormat: constant.RelayFormatOpenAI, + ChannelMeta: &ChannelMeta{ChannelType: int(constant.ProviderClaude)}, + } + info.AppendRequestConversion(types.RelayFormat(constant.RelayFormatOpenAI)) + a.CaptureProtocol(info) + if a.conversion == nil { + t.Fatal("conversion 未快照") + } + if a.conversion.ClientFormat != "openai" || a.conversion.UpstreamFormat != "claude" { + t.Errorf("协议推导错误: %+v", a.conversion) + } + if a.conversion.Bridge != "" { + t.Errorf("常规转换不应有桥接标志: %q", a.conversion.Bridge) + } + + // 直传：上游协议 = 客户端协议 + info2 := &RelayInfo{ + InboundFormat: constant.RelayFormatOpenAI, + ChannelMeta: &ChannelMeta{ + ChannelType: int(constant.ProviderOpenAI), + Settings: ChannelSettings{PassThroughBodyEnabled: true}, + }, + } + a2 := s.BeginAttempt(selection, "m", "chat_completions", false, 1) + a2.CaptureProtocol(info2) + if a2.conversion.UpstreamFormat != "openai" || a2.conversion.Bridge != "pass_through" { + t.Errorf("直传推导错误: %+v", a2.conversion) + } + + // Responses 桥接 + info3 := &RelayInfo{ + InboundFormat: constant.RelayFormatOpenAI, + UseResponsesAPI: true, + ChannelMeta: &ChannelMeta{ChannelType: int(constant.ProviderOpenAI)}, + } + a3 := s.BeginAttempt(selection, "m", "chat_completions", false, 2) + a3.CaptureProtocol(info3) + if a3.conversion.UpstreamFormat != "responses" || a3.conversion.Bridge != "responses_api" { + t.Errorf("Responses 桥接推导错误: %+v", a3.conversion) + } +} diff --git a/relay/common/http_client.go b/relay/common/http_client.go index b3d1b62..da5091f 100644 --- a/relay/common/http_client.go +++ b/relay/common/http_client.go @@ -48,13 +48,15 @@ var longRunSharedTransport = &http.Transport{ } // nonStreamClient / streamClient 预创建的单例 Client（无代理）。 +// Transport 统一经 WrapDebugTransport 包装：渠道调试开关开启的请求经 ctx 捕获器 +// 镜像上游报文，其余请求一次 ctx 查找后直接透传（连接池/H2 仍由底层 Transport 管理）。 var ( nonStreamClient = &http.Client{ - Transport: sharedTransport, + Transport: WrapDebugTransport(sharedTransport), Timeout: 300 * time.Second, } streamClient = &http.Client{ - Transport: sharedTransport, + Transport: WrapDebugTransport(sharedTransport), } ) @@ -140,11 +142,11 @@ func getProxiedClients() (*http.Client, *http.Client) { proxiedState.longRun = buildProxiedTransport(proxyURL, 0) proxiedState.nonStream = &http.Client{ - Transport: proxiedState.transport, + Transport: WrapDebugTransport(proxiedState.transport), Timeout: 300 * time.Second, } proxiedState.stream = &http.Client{ - Transport: proxiedState.transport, + Transport: WrapDebugTransport(proxiedState.transport), } proxiedState.proxyURL = proxyURL @@ -179,7 +181,7 @@ func NewPooledClient(timeoutSeconds int, useProxy bool, isStream ...bool) *http. } proxiedState.mu.RUnlock() return &http.Client{ - Transport: transport, + Transport: WrapDebugTransport(transport), Timeout: time.Duration(timeoutSeconds) * time.Second, } } @@ -195,7 +197,7 @@ func NewPooledClient(timeoutSeconds int, useProxy bool, isStream ...bool) *http. transport = longRunSharedTransport } return &http.Client{ - Transport: transport, + Transport: WrapDebugTransport(transport), Timeout: time.Duration(timeoutSeconds) * time.Second, } } diff --git a/relay/common/http_client_test.go b/relay/common/http_client_test.go index 1cdd0a5..c6b71b9 100644 --- a/relay/common/http_client_test.go +++ b/relay/common/http_client_test.go @@ -11,13 +11,19 @@ import ( // 避免图片/音频等长耗时同步请求被 180s 头超时误杀。 // - timeout ≤ 180s → 普通传输层（ResponseHeaderTimeout=180s），保留对短超时请求的假死保护。 // 非代理路径直接断言底层 *http.Transport 的字段，是最稳定的判定方式。 +// Transport 统一经 WrapDebugTransport 包装（渠道调试日志捕获），测试中先解包再断言底层传输层。 + +// unwrapDebugTransport 解开调试捕获包装，返回底层 *http.Transport +func unwrapDebugTransport(rt http.RoundTripper) *http.Transport { + if d, ok := rt.(*DebugRoundTripper); ok { + return d.Base.(*http.Transport) + } + return rt.(*http.Transport) +} func TestNewPooledClient_LongRunDisablesResponseHeaderTimeout(t *testing.T) { c := NewPooledClient(600, false) // 图片生成的典型超时（GetTimeoutSeconds 下限） - tr, ok := c.Transport.(*http.Transport) - if !ok { - t.Fatalf("expected *http.Transport, got %T", c.Transport) - } + tr := unwrapDebugTransport(c.Transport) if tr.ResponseHeaderTimeout != 0 { t.Errorf("long-run transport: ResponseHeaderTimeout=%v, want 0 (disabled)", tr.ResponseHeaderTimeout) } @@ -28,20 +34,17 @@ func TestNewPooledClient_LongRunDisablesResponseHeaderTimeout(t *testing.T) { func TestNewPooledClient_LongRunAtBoundary(t *testing.T) { // 恰好 181s 触发 longRun；180s 仍走普通传输层 - if c := NewPooledClient(181, false); c.Transport != longRunSharedTransport { + if tr := unwrapDebugTransport(NewPooledClient(181, false).Transport); tr != longRunSharedTransport { t.Error("timeout=181s should select longRunSharedTransport") } - if c := NewPooledClient(180, false); c.Transport != sharedTransport { + if tr := unwrapDebugTransport(NewPooledClient(180, false).Transport); tr != sharedTransport { t.Error("timeout=180s should select sharedTransport") } } func TestNewPooledClient_NormalKeepsResponseHeaderTimeout(t *testing.T) { c := NewPooledClient(60, false) // 普通对话默认超时 - tr, ok := c.Transport.(*http.Transport) - if !ok { - t.Fatalf("expected *http.Transport, got %T", c.Transport) - } + tr := unwrapDebugTransport(c.Transport) if tr.ResponseHeaderTimeout != defaultResponseHeaderTimeout { t.Errorf("normal transport: ResponseHeaderTimeout=%v, want %v", tr.ResponseHeaderTimeout, defaultResponseHeaderTimeout) } diff --git a/relay/common/relay_info.go b/relay/common/relay_info.go index 78c5c61..36ba205 100644 --- a/relay/common/relay_info.go +++ b/relay/common/relay_info.go @@ -84,6 +84,31 @@ type ChannelSettings struct { // UseProxy 启用代理，使用系统配置的代理地址转发请求 UseProxy bool `json:"use_proxy,omitempty"` + + // DebugLogEnabled 启用渠道调试日志：记录经该渠道每次请求尝试的四段完整报文 + //（客户端↔系统↔上游，body 不截断、凭证脱敏）。数据量大，仅排查问题时开启 + DebugLogEnabled bool `json:"debug_log_enabled,omitempty"` + + // 调试目标过滤（AND 组合，0 = 不限）：只捕捉匹配租户/成员/密钥的请求， + // 避免调试期间把全部租户的报文都录下来 + DebugLogTenantID int64 `json:"debug_log_tenant_id,omitempty"` + DebugLogUserID int64 `json:"debug_log_user_id,omitempty"` + DebugLogApiKeyID int64 `json:"debug_log_api_key_id,omitempty"` +} + +// DebugTargetMatch 调试目标过滤：已设置的过滤器全部匹配（AND），未设置（0）= 不限。 +// 在渠道调试开关开启的前提下由 relay 接线处调用，不匹配则整条捕获链跳过。 +func (s ChannelSettings) DebugTargetMatch(tenantID, userID, apiKeyID int64) bool { + if s.DebugLogTenantID > 0 && tenantID != s.DebugLogTenantID { + return false + } + if s.DebugLogUserID > 0 && userID != s.DebugLogUserID { + return false + } + if s.DebugLogApiKeyID > 0 && apiKeyID != s.DebugLogApiKeyID { + return false + } + return true } // RelayInfo 代理请求上下文，贯穿整个 relay 请求链路 @@ -290,6 +315,15 @@ func (info *RelayInfo) AppendRequestConversion(format types.RelayFormat) { info.conversionChain = append(info.conversionChain, format) } +// ConversionChain 返回请求侧协议转换链（relaykit 转换器追加；直传/同构路径可能为空）。 +// 供渠道调试日志记录协议转换轨迹。 +func (info *RelayInfo) ConversionChain() []types.RelayFormat { + if info == nil { + return nil + } + return info.conversionChain +} + func (info *RelayInfo) ConvOptions() *convmeta.Options { if info == nil { return &convmeta.Options{} diff --git a/relay/handler/relay_handler.go b/relay/handler/relay_handler.go index 8a287e1..1e7cd0f 100644 --- a/relay/handler/relay_handler.go +++ b/relay/handler/relay_handler.go @@ -41,6 +41,7 @@ type RelayContext struct { KeyTotalQuota float64 // API Key 总额度，0 表示不限制 KeyUsedQuota float64 // API Key 已用额度（仅展示/审计，额度检查会鲜读 DB） ForwardingTrace *common.ForwardingTrace // 转发路径追踪（仅管理员可见） + Debug *common.DebugSession // 渠道调试日志会话（仅调试开关开启的请求非 nil） } // BillingResult 计费结果（返回给调用方用于设置响应头） @@ -569,7 +570,30 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He "this image model uses asynchronous generation on Alibaba DashScope; submit via POST /v1/images/generations/async and poll for the result", nil) } - info := buildRelayInfo(ctx, rc, v, selection, path, headers, attempt) + // 渠道调试日志：首个开启调试且匹配目标过滤（租户/成员/密钥，AND 组合，0=不限）的尝试 + // 创建会话（捕获段1客户端请求 + 包装段4 writer，writer 保持到请求结束——中间失败尝试 + // 不向客户端写字节，累积字节即最终响应）；每轮尝试独立捕获器经 attemptCtx 传递给传输层。 + // 必须用独立变量 attemptCtx，不可重赋值 ctx：否则陈旧捕获器会泄漏进下一轮 sess.Next + // 选中无调试渠道的请求。 + var dbgAttempt *common.DebugAttempt + if selection.Settings.DebugLogEnabled && + selection.Settings.DebugTargetMatch(rc.TenantID, rc.UserID, rc.ApiKeyID) { + if rc.Debug == nil { + rc.Debug = common.NewDebugSession(rc.RequestID, rc.TenantID, rc.UserID, rc.ApiKeyID, path) + rc.Debug.CaptureClientRequest(headers, body) + dw := common.NewDebugClientWriter(rc.Writer) + rc.Debug.SetClientWriter(dw) + rc.Writer = dw + } + dbgAttempt = rc.Debug.BeginAttempt(selection, v.modelName, v.relayModeStr, v.isStream, attempt) + } + attemptCtx := ctx + if dbgAttempt != nil { + attemptCtx = common.WithDebugAttempt(ctx, dbgAttempt.Capture) + } + + info := buildRelayInfo(attemptCtx, rc, v, selection, path, headers, attempt) + dbgAttempt.CaptureProtocol(info) if tr := monitor.GetTrackedRequest(rc.RequestID); tr != nil { tr.ChannelID = selection.ChannelID @@ -582,6 +606,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He if billing != nil && preDeductAmount > 0 { _ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount) } + dbgAttempt.MarkFinal(fmt.Errorf("unsupported channel type: %d", selection.ChannelType)) return nil, v.billingResult, fmt.Errorf("unsupported channel type: %d", selection.ChannelType) } adaptor.Init(info) @@ -607,21 +632,24 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He if billing != nil && preDeductAmount > 0 { _ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount) } + dbgAttempt.MarkFinal(err) return nil, v.billingResult, constant.NewRequestError( "previous_response_id is only supported on responses-native channels; no eligible channel remains for this model. Resend the full conversation input without previous_response_id.", nil) } + dbgAttempt.Submit(err) continue } if billing != nil && preDeductAmount > 0 { _ = billing.SettleFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount) } + dbgAttempt.MarkFinal(err) return nil, v.billingResult, err } // 容量租约已由调度器在 Next 内获取；长流式请求由续期器保活 leaseRefresh := startDispatchLeaseRefresher(d.Channel.ID, rc.RequestID) - upstreamCtx := context.WithoutCancel(ctx) + upstreamCtx := context.WithoutCancel(attemptCtx) settleCtx, settleCancel := context.WithTimeout(context.Background(), 30*time.Second) defer settleCancel() @@ -645,6 +673,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He appendHop(trace, hop, false, err.Error(), info.LatencyMs()) settleCancel() sleepBackoff(ctx, backoff) + dbgAttempt.Submit(err) continue } @@ -654,6 +683,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err) recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs()) finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs()) + dbgAttempt.MarkFinal(err) return nil, v.billingResult, helper.RemapStatusCode(constant.NewUpstreamError(502, "upstream request failed", err), info.ChannelMeta.Settings.StatusCodeMapping) } @@ -695,6 +725,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He } recordFailedUsageWithTokens(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, streamUsage) finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs()) + dbgAttempt.MarkFinal(err) return usage, v.billingResult, err } @@ -733,6 +764,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He appendHop(trace, hop, false, err.Error(), info.LatencyMs()) settleCancel() sleepBackoff(ctx, backoff) + dbgAttempt.Submit(err) continue } @@ -742,10 +774,12 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err) recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs()) finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs()) + dbgAttempt.MarkFinal(err) return nil, v.billingResult, err } // 成功路径：绑定续期 + 健康上报由 Finish 完成 + dbgAttempt.MarkFinal(nil) sess.Finish(settleCtx, true, info.LatencyMs()) appendHop(trace, hop, true, "", info.LatencyMs()) trace.TotalAttempts = attempt + 1 diff --git a/web/admin/src/views/ai/ChannelDetailPage.vue b/web/admin/src/views/ai/ChannelDetailPage.vue index 6ff9a41..988431d 100644 --- a/web/admin/src/views/ai/ChannelDetailPage.vue +++ b/web/admin/src/views/ai/ChannelDetailPage.vue @@ -1,7 +1,7 @@