fix(observability): 治理SSE慢请求虚报、客户端取消日志污染、退款重试刷屏与会话计数慢查询

This commit is contained in:
yml2213
2026-09-10 20:02:54 +08:00
parent ee0e70ccad
commit 784ffb5f72
7 changed files with 24 additions and 8 deletions
+4 -1
View File
@@ -72,9 +72,12 @@ func (l *structuredGormLogger) Trace(ctx context.Context, begin time.Time, query
} }
elapsed := time.Since(begin) elapsed := time.Since(begin)
switch { switch {
case err != nil && !errors.Is(err, gorm.ErrRecordNotFound) && l.level >= gormLogger.Error: case err != nil && !errors.Is(err, gorm.ErrRecordNotFound) && !errors.Is(err, context.Canceled) && l.level >= gormLogger.Error:
sql, rows := query() sql, rows := query()
l.withContext(ctx).Error("数据库查询失败", queryFields(sql, rows, elapsed, err)...) l.withContext(ctx).Error("数据库查询失败", queryFields(sql, rows, elapsed, err)...)
case errors.Is(err, context.Canceled) && l.level >= gormLogger.Info:
sql, rows := query()
l.withContext(ctx).Debug("数据库查询被客户端取消", queryFields(sql, rows, elapsed, err)...)
case elapsed >= l.slowThreshold && l.level >= gormLogger.Warn: case elapsed >= l.slowThreshold && l.level >= gormLogger.Warn:
sql, rows := query() sql, rows := query()
l.withContext(ctx).Warn("数据库慢查询", queryFields(sql, rows, elapsed, nil)...) l.withContext(ctx).Warn("数据库慢查询", queryFields(sql, rows, elapsed, nil)...)
+1 -1
View File
@@ -123,7 +123,7 @@ func (j *Job) run(ctx context.Context) {
if err != nil { if err != nil {
j.logger.Warn("退款重试任务扫描缺失退款单失败", zap.Error(err)) j.logger.Warn("退款重试任务扫描缺失退款单失败", zap.Error(err))
} }
if processed > 0 || missing > 0 { if processed > 0 || rebuilt > 0 {
j.logger.Info("退款重试任务处理完成", j.logger.Info("退款重试任务处理完成",
zap.Int("processed", processed), zap.Int("processed", processed),
zap.Int("missing_refund_orders", missing), zap.Int("missing_refund_orders", missing),
+11 -4
View File
@@ -1,6 +1,8 @@
package middleware package middleware
import ( import (
"context"
"errors"
"net/http" "net/http"
"strings" "strings"
"time" "time"
@@ -72,6 +74,10 @@ func RequestLogger(logger *zap.Logger) gin.HandlerFunc {
} }
switch { switch {
case errors.Is(c.Request.Context().Err(), context.Canceled):
if logger.Core().Enabled(zap.DebugLevel) {
logger.Debug("HTTP 客户端中断连接", fields...)
}
case status >= 500: case status >= 500:
logger.Error("HTTP 请求失败", fields...) logger.Error("HTTP 请求失败", fields...)
case status == http.StatusTooManyRequests: case status == http.StatusTooManyRequests:
@@ -113,7 +119,7 @@ func shouldSkipHTTPLog(path, route string, status int, latencyMs float64) bool {
return false return false
} }
// SSE 的请求耗时就是连接存活时间,由 chathub 单独记录连接生命周期。 // SSE 的请求耗时就是连接存活时间,由 chathub 单独记录连接生命周期。
if isChatSSEPath(path, route) { if isSSEPath(path, route) {
return true return true
} }
if latencyMs >= slowRequestThresholdMs { if latencyMs >= slowRequestThresholdMs {
@@ -122,9 +128,10 @@ func shouldSkipHTTPLog(path, route string, status int, latencyMs float64) bool {
return true return true
} }
func isChatSSEPath(path, route string) bool { func isSSEPath(path, route string) bool {
return path == "/api/chats/events" || path == "/api/admin/chats/events" || return path == "/api/auth/events" || route == "/api/auth/events" ||
route == "/api/chats/events" || route == "/api/admin/chats/events" path == "/api/chats/events" || route == "/api/chats/events" ||
path == "/api/admin/chats/events" || route == "/api/admin/chats/events"
} }
func meaningfulAuthFailure(c *gin.Context) bool { func meaningfulAuthFailure(c *gin.Context) bool {
@@ -19,6 +19,7 @@ func TestShouldSkipHTTPLog(t *testing.T) {
{name: "轮询 500 不跳过", path: "/api/wallet/balance", status: 500, latencyMs: 1, wantSkip: false}, {name: "轮询 500 不跳过", path: "/api/wallet/balance", status: 500, latencyMs: 1, wantSkip: false},
{name: "限流请求不跳过", path: "/api/auth/sms", status: 429, latencyMs: 1, wantSkip: false}, {name: "限流请求不跳过", path: "/api/auth/sms", status: 429, latencyMs: 1, wantSkip: false},
{name: "慢请求不跳过", path: "/api/orders", status: 200, latencyMs: 500, wantSkip: false}, {name: "慢请求不跳过", path: "/api/orders", status: 200, latencyMs: 500, wantSkip: false},
{name: "用户全局通知 SSE 长连接跳过", path: "/api/auth/events", route: "/api/auth/events", status: 200, latencyMs: 60_000, wantSkip: true},
{name: "用户 SSE 长连接跳过", path: "/api/chats/events", route: "/api/chats/events", status: 200, latencyMs: 60_000, wantSkip: true}, {name: "用户 SSE 长连接跳过", path: "/api/chats/events", route: "/api/chats/events", status: 200, latencyMs: 60_000, wantSkip: true},
{name: "后台 SSE 长连接跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 200, latencyMs: 600_000, wantSkip: true}, {name: "后台 SSE 长连接跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 200, latencyMs: 600_000, wantSkip: true},
{name: "SSE 服务端错误不跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 500, latencyMs: 600_000, wantSkip: false}, {name: "SSE 服务端错误不跳过", path: "/api/admin/chats/events", route: "/api/admin/chats/events", status: 500, latencyMs: 600_000, wantSkip: false},
@@ -1,6 +1,7 @@
package chat package chat
import ( import (
"context"
"errors" "errors"
"net/http" "net/http"
@@ -10,6 +11,9 @@ import (
) )
func writeChatError(c *gin.Context, err error) { func writeChatError(c *gin.Context, err error) {
if errors.Is(err, context.Canceled) || errors.Is(c.Request.Context().Err(), context.Canceled) {
return
}
response.RecordError(c, err) response.RecordError(c, err)
switch { switch {
case errors.Is(err, ErrDependencyUnavailable): case errors.Is(err, ErrDependencyUnavailable):
+1 -1
View File
@@ -229,6 +229,7 @@ func (r *Repository) listAdminConversations(ctx context.Context, principal Princ
var rows []conversationRow var rows []conversationRow
offset := (page - 1) * pageSize offset := (page - 1) * pageSize
queryDB := r.adminConversationBase(ctx, principal, keyword). queryDB := r.adminConversationBase(ctx, principal, keyword).
Joins("LEFT JOIN chat_messages AS lm ON lm.id = c.last_message_id").
Select(`c.id, c.order_id, c.listing_id, c.type, c.support_scene, c.title, c.status, c.last_message_id, Select(`c.id, c.order_id, c.listing_id, c.type, c.support_scene, c.title, c.status, c.last_message_id,
c.last_message_preview, c.last_message_at, c.created_at, c.updated_at, c.last_message_preview, c.last_message_at, c.created_at, c.updated_at,
COALESCE(cp_me.role, 'admin') AS role, COALESCE(cp_me.role, 'admin') AS role,
@@ -302,7 +303,6 @@ func (r *Repository) adminConversationBase(ctx context.Context, principal Princi
db := r.db.WithContext(ctx).Table("chat_conversations AS c"). db := r.db.WithContext(ctx).Table("chat_conversations AS c").
Joins("LEFT JOIN chat_participants AS cp_me ON cp_me.conversation_id = c.id AND cp_me.participant_type = ? AND cp_me.participant_id = ?", principal.Type, principal.ID). Joins("LEFT JOIN chat_participants AS cp_me ON cp_me.conversation_id = c.id AND cp_me.participant_type = ? AND cp_me.participant_id = ?", principal.Type, principal.ID).
Joins("LEFT JOIN chat_admin_conversation_states AS cas ON cas.conversation_id = c.id AND cas.admin_user_id = ?", principal.ID). Joins("LEFT JOIN chat_admin_conversation_states AS cas ON cas.conversation_id = c.id AND cas.admin_user_id = ?", principal.ID).
Joins("LEFT JOIN chat_messages AS lm ON lm.id = c.last_message_id").
Joins("LEFT JOIN rental_orders AS explicit_lo ON explicit_lo.id = c.order_id"). Joins("LEFT JOIN rental_orders AS explicit_lo ON explicit_lo.id = c.order_id").
Joins("LEFT JOIN rental_listings AS l ON l.id = COALESCE(explicit_lo.listing_id, c.listing_id)") Joins("LEFT JOIN rental_listings AS l ON l.id = COALESCE(explicit_lo.listing_id, c.listing_id)")
if keyword != "" { if keyword != "" {
+2 -1
View File
@@ -1,6 +1,7 @@
package response package response
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"net/http" "net/http"
@@ -45,7 +46,7 @@ func Error(c *gin.Context, status int, code, message string) {
// RecordError 保存原始错误,供统一请求日志记录服务端故障原因。 // RecordError 保存原始错误,供统一请求日志记录服务端故障原因。
func RecordError(c *gin.Context, err error) { func RecordError(c *gin.Context, err error) {
if c == nil || err == nil { if c == nil || err == nil || errors.Is(err, context.Canceled) {
return return
} }
for _, item := range c.Errors { for _, item := range c.Errors {