okooooooo/team-api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve(log):优化渠道日志揭露，方便排查问题

Browse Source
This commit is contained in:
qian committed 2026-09-01 00:11:32 +08:00
1 parent bf5dc7684d
commit 15ec726f68
3 files changed
+195 -25
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/logic/task/async_polling.go
+17 -15
View File
@@ -15,6 +15,7 @@ import (
"github.com/qianfree/team-api/internal/logic/relay"
"github.com/qianfree/team-api/relay/common"
"github.com/qianfree/team-api/relay/constant"
"github.com/qianfree/team-api/relay/helper"
"github.com/qianfree/team-api/relay/taskchannel"
)
@@ -492,21 +493,22 @@ func recordTaskUsage(task *common.AsyncTask, channel *common.ChannelBasicInfo, s
}
record := &common.UsageRecord{
TenantID: task.TenantID,
UserID: task.UserID,
ApiKeyID: task.ApiKeyID,
ChannelID: task.ChannelID,
ChannelName: channelName,
ChannelType: channelType,
ModelName: task.ModelName,
RelayMode: int(constant.RelayModeVideoGenerations),
RequestType: 3, // async
LatencyMs: float64(latencyMs),
IsStream: false,
Success: success,
RequestID: task.RequestID,
Status: status,
ErrorMessage: errMsg,
TenantID: task.TenantID,
UserID: task.UserID,
ApiKeyID: task.ApiKeyID,
ChannelID: task.ChannelID,
ChannelName: channelName,
ChannelType: channelType,
ModelName: task.ModelName,
RelayMode: int(constant.RelayModeVideoGenerations),
RequestType: 3, // async
LatencyMs: float64(latencyMs),
IsStream: false,
Success: success,
RequestID: task.RequestID,
Status: status,
// errMsg 可能来自上游 FailReason 原文，租户端用量日志可见，抹除其中的 URL/IP
ErrorMessage: helper.RedactMessage(errMsg),
PromptTokens: task.PromptTokens,
CompletionTokens: task.CompletionTokens,
TotalTokens: task.TotalTokens,
relay/handler/relay_handler.go
+47 -10
View File
@@ -374,6 +374,12 @@ func settleSuccessfulRequest(
RetryIndex: info.RetryIndex,
}
// 流式追踪：正常结束也落库（done / eof）。eof（上游未发 [DONE] 即结束）按正常结束
// 结算但属上游侧异常信号，记录后可在用量日志直接区分，无需翻应用日志。
if info.StreamStatus != nil {
usageRecord.StreamEndReason = string(info.StreamStatus.GetEndReason())
}
// 填充结算费用数据
if settleResult != nil {
usageRecord.TotalCost = settleResult.BaseCost
@@ -692,7 +698,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
if billing != nil && preDeductAmount > 0 {
_ = billing.SettleFailed(settleCtx, rc.TenantID, rc.RequestID, preDeductAmount)
}
recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err)
recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, info.StreamStatus)
recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs())
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
dbgAttempt.MarkFinal(err)
@@ -710,8 +716,9 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
err = helper.RemapStatusCode(err, info.ChannelMeta.Settings.StatusCodeMapping)
if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() {
g.Log().Warningf(ctx, "[RelayHandler] Stream interrupted: adaptor=%s, model=%s, reason=%s",
adaptor.GetChannelName(), v.modelName, info.StreamStatus.Summary())
g.Log().Warningf(ctx, "[RelayHandler] Stream interrupted: request=%s tenant=%d channel=%d(%s) adaptor=%s model=%s attempt=%d reason=%s ctx_err=%v usage=%v pre_deduct=%.6f",
rc.RequestID, rc.TenantID, selection.ChannelID, selection.ChannelName, adaptor.GetChannelName(), v.modelName,
attempt, info.StreamStatus.Summary(), ctx.Err(), usage, preDeductAmount)
// 客户端已收到部分流：不可重试（响应已污染），上报健康后按流中断结算
_, _ = sess.Report(settleCtx, dispatchStatusCode(err), err, dispatch.DeliveryResponseStarted, info.LatencyMs(), 0)
streamUsage := usage
@@ -737,7 +744,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
}
interruptSettle = settleResult
}
recordFailedUsageWithTokens(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, streamUsage, interruptSettle)
recordFailedUsageWithTokens(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, streamUsage, interruptSettle, info.StreamStatus)
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
dbgAttempt.MarkFinal(err)
return usage, v.billingResult, err
@@ -785,7 +792,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
if billing != nil && preDeductAmount > 0 {
_ = billing.SettleFailed(settleCtx, rc.TenantID, rc.RequestID, preDeductAmount)
}
recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err)
recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, info.StreamStatus)
recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs())
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
dbgAttempt.MarkFinal(err)
@@ -954,7 +961,7 @@ func handleChannelUnavailable(
"当前模型暂时不可用",
constant.ErrAllChannelsFailed,
)
recordFailedUsage(provider, rc, nil, v.modelName, v.relayMode, v.isStream, allFailedErr)
recordFailedUsage(provider, rc, nil, v.modelName, v.relayMode, v.isStream, allFailedErr, nil)
return &channelUnavailableResult{nil, v.billingResult, allFailedErr}
}
@@ -1029,10 +1036,34 @@ func (e *RelayErrorWithRateLimit) Error() string {
return e.Message
}
// applyStreamEndDiag 把流式结束原因落进失败用量记录，替代无归因价值的哨兵文案。
// stream_end_reason 区分 client_gone / handler_stop / scanner_err / timeout / ping_fail / eof，
// error_message 保留真实底层错误（ctx 取消原因、写客户端失败的 TCP 错误等）；
// 哨兵错误（ErrStreamInterrupted）或状态里已带底层错误时直接用摘要，其余追加在原始错误后。
// error_message 对租户端可见，全部经 RedactMessage/SafeUpstreamErrorMessage 脱敏：
// 摘要中的本地连接地址抹 IP，外层错误按传输层归一化，禁止上游域名/URL/IP 落入用量日志。
func applyStreamEndDiag(record *common.UsageRecord, streamStatus *common.StreamStatus, err error) {
if streamStatus == nil {
return
}
reason := streamStatus.GetEndReason()
if reason == "" {
return
}
record.StreamEndReason = string(reason)
summary := helper.RedactMessage(streamStatus.Summary())
if err == nil || errors.Is(err, common.ErrStreamInterrupted) || streamStatus.Error() != nil {
record.ErrorMessage = summary
return
}
record.ErrorMessage = helper.SafeUpstreamErrorMessage(err) + " [stream: " + summary + "]"
}
// recordFailedUsage 记录失败用量。
// selection 非 nil 时记录具体失败渠道（ID/名称/类型/上游模型），便于在用量日志定位失败渠道；
// nil 表示无单一渠道（全部渠道失败），渠道字段留空，失败详情见 error_message。
func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error) {
// streamStatus 非 nil 且已有结束原因时，同步落 stream_end_reason 并修正 error_message。
func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, streamStatus *common.StreamStatus) {
record := &common.UsageRecord{
TenantID: rc.TenantID,
UserID: rc.UserID,
@@ -1045,8 +1076,11 @@ func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection
Success: false,
RequestID: rc.RequestID,
Status: "error",
ErrorMessage: err.Error(),
// 租户端用量日志可见 error_message，须经脱敏（上游域名/URL/IP 禁止落库）；
// 完整原始错误保留在应用日志与审计 ForwardingTrace（仅管理员可见）
ErrorMessage: helper.SafeUpstreamErrorMessage(err),
}
applyStreamEndDiag(record, streamStatus, err)
if selection != nil {
record.ChannelID = selection.ChannelID
record.ChannelName = selection.ChannelName
@@ -1058,7 +1092,7 @@ func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection
// recordFailedUsageWithTokens 记录失败用量（含 token 明细，用于流中断等已有部分 usage 的场景）。
// 与 recordFailedUsage 的区别：此函数会填充 token 字段，避免报表中流中断记录的 token 全为 0。
func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, usage *common.Usage, settleResult *common.SettlementResult) {
func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, usage *common.Usage, settleResult *common.SettlementResult, streamStatus *common.StreamStatus) {
record := &common.UsageRecord{
TenantID: rc.TenantID,
UserID: rc.UserID,
@@ -1071,8 +1105,11 @@ func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext,
Success: false,
RequestID: rc.RequestID,
Status: "error",
ErrorMessage: err.Error(),
// 租户端用量日志可见 error_message，须经脱敏（上游域名/URL/IP 禁止落库）；
// 完整原始错误保留在应用日志与审计 ForwardingTrace（仅管理员可见）
ErrorMessage: helper.SafeUpstreamErrorMessage(err),
}
applyStreamEndDiag(record, streamStatus, err)
if selection != nil {
record.ChannelID = selection.ChannelID
record.ChannelName = selection.ChannelName
relay/handler/stream_diag_test.go
+131
View File
@@ -0,0 +1,131 @@
package handler
import (
"context"
"errors"
"net"
"net/url"
"strings"
"testing"
"github.com/qianfree/team-api/relay/common"
"github.com/qianfree/team-api/relay/constant"
)
// TestApplyStreamEndDiag_SentinelReplacedBySummary 哨兵错误（ErrStreamInterrupted）本身无归因价值，
// error_message 必须换成 StreamStatus 摘要，让用量日志能区分 context canceled / TCP 写失败等真实原因。
func TestApplyStreamEndDiag_SentinelReplacedBySummary(t *testing.T) {
status := common.NewStreamStatus()
status.SetEndReason(common.StreamEndReasonClientGone, context.Canceled)
record := &common.UsageRecord{ErrorMessage: common.ErrStreamInterrupted.Error()}
applyStreamEndDiag(record, status, common.ErrStreamInterrupted)
if record.StreamEndReason != string(common.StreamEndReasonClientGone) {
t.Errorf("StreamEndReason = %q, want %q", record.StreamEndReason, common.StreamEndReasonClientGone)
}
want := "client_gone: context canceled"
if record.ErrorMessage != want {
t.Errorf("ErrorMessage = %q, want %q", record.ErrorMessage, want)
}
}
// TestApplyStreamEndDiag_StatusCarriesErrorSummaryWins 状态本身带底层错误（如 handler_stop 的
// 写客户端失败）时摘要信息量高于外层错误，直接以摘要作为 error_message。
func TestApplyStreamEndDiag_StatusCarriesErrorSummaryWins(t *testing.T) {
status := common.NewStreamStatus()
writeErr := errors.New("write tcp 127.0.0.1:18888->127.0.0.1:6396: wsasend: connection forcibly closed")
status.SetEndReason(common.StreamEndReasonHandlerStop, writeErr)
record := &common.UsageRecord{ErrorMessage: "upstream error"}
applyStreamEndDiag(record, status, errors.New("upstream error"))
if !strings.Contains(record.ErrorMessage, "write tcp") {
t.Errorf("ErrorMessage should keep underlying write error, got %q", record.ErrorMessage)
}
if record.StreamEndReason != string(common.StreamEndReasonHandlerStop) {
t.Errorf("StreamEndReason = %q, want %q", record.StreamEndReason, common.StreamEndReasonHandlerStop)
}
}
// TestApplyStreamEndDiag_RealErrorKeptWithSummary 非哨兵外层错误（如上游 5xx）且状态无底层错误时
// （如 eof），保留原始错误并把流结束原因追加在后，两条线索都不丢。
func TestApplyStreamEndDiag_RealErrorKeptWithSummary(t *testing.T) {
status := common.NewStreamStatus()
status.SetEndReason(common.StreamEndReasonEOF, nil)
upstreamErr := errors.New("upstream returned HTTP 502")
record := &common.UsageRecord{ErrorMessage: upstreamErr.Error()}
applyStreamEndDiag(record, status, upstreamErr)
if !strings.HasPrefix(record.ErrorMessage, upstreamErr.Error()) {
t.Errorf("ErrorMessage should keep original error, got %q", record.ErrorMessage)
}
if !strings.Contains(record.ErrorMessage, string(common.StreamEndReasonEOF)) {
t.Errorf("ErrorMessage should append stream summary, got %q", record.ErrorMessage)
}
if record.StreamEndReason != string(common.StreamEndReasonEOF) {
t.Errorf("StreamEndReason = %q, want %q", record.StreamEndReason, common.StreamEndReasonEOF)
}
}
// TestApplyStreamEndDiag_NoopCases nil 状态与未设置结束原因的状态不得改动记录。
func TestApplyStreamEndDiag_NoopCases(t *testing.T) {
record := &common.UsageRecord{ErrorMessage: "raw"}
applyStreamEndDiag(record, nil, errors.New("x"))
if record.StreamEndReason != "" || record.ErrorMessage != "raw" {
t.Errorf("nil status should be no-op, got reason=%q message=%q", record.StreamEndReason, record.ErrorMessage)
}
empty := common.NewStreamStatus()
applyStreamEndDiag(record, empty, errors.New("x"))
if record.StreamEndReason != "" || record.ErrorMessage != "raw" {
t.Errorf("empty reason should be no-op, got reason=%q message=%q", record.StreamEndReason, record.ErrorMessage)
}
}
// TestApplyStreamEndDiag_UpstreamURLNeverLeaks 用量日志 error_message 对租户可见，
// 编排层包装的上游传输错误（*url.Error 含完整上游域名/IP）必须归一化，禁止原样落库。
func TestApplyStreamEndDiag_UpstreamURLNeverLeaks(t *testing.T) {
// 还原线上泄露形态：NewUpstreamError(502, "请求处理失败", Post "https://api.deepseek.com/...": dial tcp ...)
dialErr := &net.OpError{
Op: "dial",
Net: "tcp",
Err: errors.New("connectex: A connection attempt failed because the connected party did not properly respond"),
}
upstreamErr := &url.Error{Op: "Post", URL: "https://api.deepseek.com/anthropic/v1/messages", Err: dialErr}
relayErr := constant.NewUpstreamError(502, "请求处理失败", upstreamErr)
status := common.NewStreamStatus()
status.SetEndReason(common.StreamEndReasonEOF, nil)
record := &common.UsageRecord{ErrorMessage: relayErr.Error()}
applyStreamEndDiag(record, status, relayErr)
for _, leak := range []string{"api.deepseek.com", "171.105.220.186", "https://", "/anthropic/"} {
if strings.Contains(record.ErrorMessage, leak) {
t.Errorf("error_message 泄露上游信息 %q: %q", leak, record.ErrorMessage)
}
}
}
// TestApplyStreamEndDiag_SummaryLocalAddrRedacted 摘要中的写客户端 TCP 错误含本地地址，
// IP 必须抹除（保留 wsasend 等错误类别供排查）。
func TestApplyStreamEndDiag_SummaryLocalAddrRedacted(t *testing.T) {
status := common.NewStreamStatus()
status.SetEndReason(common.StreamEndReasonHandlerStop,
errors.New("write tcp 127.0.0.1:18888->127.0.0.1:6396: wsasend: An existing connection was forcibly closed"))
record := &common.UsageRecord{ErrorMessage: "x"}
applyStreamEndDiag(record, status, common.ErrStreamInterrupted)
if strings.Contains(record.ErrorMessage, "127.0.0.1") {
t.Errorf("摘要未抹除本地 IP: %q", record.ErrorMessage)
}
if !strings.Contains(record.ErrorMessage, "wsasend") {
t.Errorf("应保留错误类别 wsasend: %q", record.ErrorMessage)
}
if !strings.Contains(record.ErrorMessage, string(common.StreamEndReasonHandlerStop)) {
t.Errorf("应保留结束原因 handler_stop: %q", record.ErrorMessage)
}
}