From 15ec726f689425013d22c85cf0526eaa48bfd03d Mon Sep 17 00:00:00 2001 From: qian Date: Tue, 1 Sep 2026 00:11:32 +0800 Subject: [PATCH] =?UTF-8?q?improve(log):=E4=BC=98=E5=8C=96=E6=B8=A0?= =?UTF-8?q?=E9=81=93=E6=97=A5=E5=BF=97=E6=8F=AD=E9=9C=B2=EF=BC=8C=E6=96=B9?= =?UTF-8?q?=E4=BE=BF=E6=8E=92=E6=9F=A5=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/logic/task/async_polling.go | 32 ++++--- relay/handler/relay_handler.go | 57 ++++++++++-- relay/handler/stream_diag_test.go | 131 +++++++++++++++++++++++++++ 3 files changed, 195 insertions(+), 25 deletions(-) create mode 100644 relay/handler/stream_diag_test.go diff --git a/internal/logic/task/async_polling.go b/internal/logic/task/async_polling.go index 7fe06ab..c2a0e53 100644 --- a/internal/logic/task/async_polling.go +++ b/internal/logic/task/async_polling.go @@ -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, diff --git a/relay/handler/relay_handler.go b/relay/handler/relay_handler.go index cdccf07..e48167a 100644 --- a/relay/handler/relay_handler.go +++ b/relay/handler/relay_handler.go @@ -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 diff --git a/relay/handler/stream_diag_test.go b/relay/handler/stream_diag_test.go new file mode 100644 index 0000000..51d20f5 --- /dev/null +++ b/relay/handler/stream_diag_test.go @@ -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) + } +}