yml/hfb_sys
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化生产日志体积：访问降噪、压缩清理与退款告警节流

Browse Source 
跳过轮询/封面等热路径成功请求，按日 gzip 并保留可配天数，退款 max-retry 汇总告警。
This commit is contained in:
yml2213
2026-07-13 22:02:20 +08:00
parent f51aed1f0f
commit 2efcede800
8 changed files with 361 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/config/config.go
+3
View File
@@ -57,6 +57,8 @@ type LogConfig struct {
Dir string
EnableConsole bool
EnableFile bool
// RetainDays 日志保留天数（含当天）；过期的 .log / .log.gz 会被清理。0 表示不自动清理。
RetainDays int
}
type RateLimitConfig struct {
@@ -108,6 +110,7 @@ func Load() Config {
Dir: getEnv("LOG_DIR", "logs"),
EnableConsole: getEnvBool("LOG_ENABLE_CONSOLE", true),
EnableFile: getEnvBool("LOG_ENABLE_FILE", true),
RetainDays: getEnvInt("LOG_RETAIN_DAYS", 14),
},
RateLimit: RateLimitConfig{
Enabled: getEnvBool("RATE_LIMIT_ENABLED", true),
backend/internal/jobs/refundretry/job.go
+50 -21
View File
@@ -225,14 +225,25 @@ func (j *Job) warnMaxRetryRefunds(ctx context.Context, now time.Time) {
j.logger.Warn("refund retry max count scan failed", zap.Error(err))
return
}
if len(rows) == 0 {
return
}
// 汇总一条告警，避免每 2 分钟对同一批单刷屏；更新 last_retry_at 实现 24h 节流
ids := make([]uint64, 0, len(rows))
orderIDs := make([]uint64, 0, len(rows))
for _, row := range rows {
j.logger.Warn("refund retry reached max count, need manual check",
zap.Uint64("payment_id", row.ID),
zap.Uint64("order_id", row.OrderID),
zap.String("biz_type", row.BizType),
zap.String("status", row.Status),
zap.Int("retry_count", row.RetryCount),
)
ids = append(ids, row.ID)
orderIDs = append(orderIDs, row.OrderID)
}
j.logger.Warn("refund retry reached max count, need manual check",
zap.Int("count", len(rows)),
zap.Uint64s("payment_ids", ids),
zap.Uint64s("order_ids", orderIDs),
)
if err := j.db.WithContext(ctx).Model(&model.PaymentOrder{}).
Where("id IN ?", ids).
Update("last_retry_at", now).Error; err != nil {
j.logger.Warn("refund retry bump last_retry_at failed", zap.Error(err))
}
}
@@ -248,6 +259,8 @@ func (j *Job) warnMissingRefundOrders(ctx context.Context, now time.Time) (int,
}
missing := 0
rebuilt := 0
var missingOrderIDs []uint64
var rebuildFailOrderIDs []uint64
for _, row := range rows {
var count int64
if err := j.db.WithContext(ctx).Model(&model.PaymentOrder{}).
@@ -264,18 +277,11 @@ func (j *Job) warnMissingRefundOrders(ctx context.Context, now time.Time) (int,
// 非仲裁孤儿 biz_type 无法从订单状态可靠区分，维持只告警。
if row.SettlementStatus == "arbitrated" {
if j.payments == nil {
j.logger.Warn("auto rebuild arbitration refund skipped, payment repository unavailable",
zap.Uint64("order_id", row.ID),
zap.Int64("refund_amount_cent", row.RefundAmountCent),
)
rebuildFailOrderIDs = append(rebuildFailOrderIDs, row.ID)
continue
}
if _, err := j.payments.StartRefund(ctx, row.ID, row.RefundAmountCent, "arbitration_refund", "仲裁退款补偿补建"); err != nil {
j.logger.Warn("auto rebuild arbitration refund failed, need manual check",
zap.Uint64("order_id", row.ID),
zap.Int64("refund_amount_cent", row.RefundAmountCent),
zap.Error(err),
)
rebuildFailOrderIDs = append(rebuildFailOrderIDs, row.ID)
} else {
rebuilt++
j.logger.Info("auto rebuild arbitration refund submitted",
@@ -285,13 +291,36 @@ func (j *Job) warnMissingRefundOrders(ctx context.Context, now time.Time) (int,
}
continue
}
missingOrderIDs = append(missingOrderIDs, row.ID)
}
// 汇总告警 + Redis 24h 去重，避免每轮扫描刷屏
if len(missingOrderIDs) > 0 && j.shouldWarn(ctx, "missing_payment", 0) {
j.logger.Warn("refund order missing payment record, need manual check",
zap.Uint64("order_id", row.ID),
zap.String("order_no", row.OrderNo),
zap.String("refund_status", row.RefundStatus),
zap.String("settlement_status", row.SettlementStatus),
zap.Int64("refund_amount_cent", row.RefundAmountCent),
zap.Int("count", len(missingOrderIDs)),
zap.Uint64s("order_ids", missingOrderIDs),
)
}
if len(rebuildFailOrderIDs) > 0 && j.shouldWarn(ctx, "rebuild_fail", 0) {
j.logger.Warn("auto rebuild arbitration refund failed, need manual check",
zap.Int("count", len(rebuildFailOrderIDs)),
zap.Uint64s("order_ids", rebuildFailOrderIDs),
)
}
return missing, rebuilt, nil
}
// shouldWarn 用 Redis 做 24h 节流；无 Redis 时每个 job 周期最多打一次同类汇总（由调用方聚合）。
func (j *Job) shouldWarn(ctx context.Context, kind string, id uint64) bool {
if j.redis == nil {
return true
}
key := fmt.Sprintf("hfb:job:refundretry:warn:%s", kind)
if id > 0 {
key = fmt.Sprintf("%s:%d", key, id)
}
ok, err := j.redis.SetNX(ctx, key, "1", manualWarnThreshold).Result()
if err != nil {
return true
}
return ok
}
backend/internal/jobs/refundretry/job_test.go
+44
View File
@@ -46,6 +46,50 @@ func TestRetryBackoffCapsAtOneHour(t *testing.T) {
}
}
func TestWarnMaxRetryRefundsThrottlesByLastRetryAt(t *testing.T) {
db := setupRefundRetryTestDB(t)
job := New(db, nil, zap.NewNop(), nil)
now := time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC)
old := now.Add(-25 * time.Hour)
row := model.PaymentOrder{
PaymentNo: "PAY202607110001",
OrderID: 9,
OrderNo: "ORD202607110001",
UserID: 2,
Provider: "mock",
ThirdOrderID: "REF202607110001",
BizType: "admin_refund",
Status: "failed",
RetryCount: maxRetryCount,
LastRetryAt: &old,
}
if err := db.Create(&row).Error; err != nil {
t.Fatalf("创建退款单失败: %v", err)
}
job.warnMaxRetryRefunds(t.Context(), now)
var latest model.PaymentOrder
if err := db.First(&latest, row.ID).Error; err != nil {
t.Fatalf("查询退款单失败: %v", err)
}
if latest.LastRetryAt == nil || !latest.LastRetryAt.Equal(now) {
t.Fatalf("LastRetryAt 应更新为 now 以节流，got %v", latest.LastRetryAt)
}
// 立即再扫不应匹配（last_retry_at 未超过 24h）
var count int64
if err := db.Model(&model.PaymentOrder{}).
Where("retry_count >= ? AND (last_retry_at IS NULL OR last_retry_at <= ?)",
maxRetryCount, now.Add(-manualWarnThreshold)).
Count(&count).Error; err != nil {
t.Fatalf("count: %v", err)
}
if count != 0 {
t.Fatalf("throttled rows count = %d, want 0", count)
}
}
func TestMarkRetryFailedStopsAtMaxCount(t *testing.T) {
db := setupRefundRetryTestDB(t)
job := New(db, nil, zap.NewNop(), nil)
backend/internal/logging/daily_writer.go
+108 -10
View File
@@ -1,27 +1,35 @@
package logging
import (
"compress/gzip"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
type dailyWriter struct {
mu sync.Mutex
dir string
prefix string
location *time.Location
day string
file *os.File
mu sync.Mutex
dir string
prefix string
location *time.Location
retainDays int
day string
file *os.File
}
func newDailyWriter(dir string, prefix string, location *time.Location) *dailyWriter {
func newDailyWriter(dir string, prefix string, location *time.Location, retainDays int) *dailyWriter {
if retainDays < 0 {
retainDays = 0
}
return &dailyWriter{
dir: dir,
prefix: prefix,
location: location,
dir: dir,
prefix: prefix,
location: location,
retainDays: retainDays,
}
}
@@ -55,6 +63,7 @@ func (w *dailyWriter) rotateIfNeeded(now time.Time) error {
return err
}
prevDay := w.day
if w.file != nil {
_ = w.file.Close()
w.file = nil
@@ -67,5 +76,94 @@ func (w *dailyWriter) rotateIfNeeded(now time.Time) error {
}
w.file = file
w.day = day
// 异步压缩昨日日志并清理过期文件，避免阻塞业务写路径
if prevDay != "" && prevDay != day {
go w.maintain(prevDay, now)
} else {
// 进程启动或首写：尝试压缩昨天 + 清理过期
yesterday := now.AddDate(0, 0, -1).Format("2006-01-02")
go w.maintain(yesterday, now)
}
return nil
}
func (w *dailyWriter) maintain(prevDay string, now time.Time) {
if prevDay != "" {
_ = w.compressDay(prevDay)
}
if w.retainDays > 0 {
_ = w.purgeOlderThan(now)
}
}
func (w *dailyWriter) compressDay(day string) error {
srcPath := filepath.Join(w.dir, fmt.Sprintf("%s-%s.log", w.prefix, day))
dstPath := srcPath + ".gz"
if _, err := os.Stat(dstPath); err == nil {
// 已压缩则删除明文（若仍存在）
_ = os.Remove(srcPath)
return nil
}
src, err := os.Open(srcPath)
if err != nil {
if os.IsNotExist(err) {
return nil
}
return err
}
defer src.Close()
dst, err := os.OpenFile(dstPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644)
if err != nil {
return err
}
gz := gzip.NewWriter(dst)
if _, err := io.Copy(gz, src); err != nil {
_ = gz.Close()
_ = dst.Close()
_ = os.Remove(dstPath)
return err
}
if err := gz.Close(); err != nil {
_ = dst.Close()
_ = os.Remove(dstPath)
return err
}
if err := dst.Close(); err != nil {
return err
}
return os.Remove(srcPath)
}
func (w *dailyWriter) purgeOlderThan(now time.Time) error {
// 按「日历天」比较：保留最近 retainDays 天（含当天），更早的 .log / .log.gz 删除。
today := time.Date(now.In(w.location).Year(), now.In(w.location).Month(), now.In(w.location).Day(), 0, 0, 0, 0, w.location)
cutoff := today.AddDate(0, 0, -w.retainDays) // day < cutoff 才删除；retainDays=2 且今天 12 号 → 删除 10 号之前
entries, err := os.ReadDir(w.dir)
if err != nil {
return err
}
prefix := w.prefix + "-"
for _, entry := range entries {
if entry.IsDir() {
continue
}
name := entry.Name()
if !strings.HasPrefix(name, prefix) {
continue
}
// app-2026-07-11.log 或 app-2026-07-11.log.gz
rest := strings.TrimPrefix(name, prefix)
rest = strings.TrimSuffix(rest, ".gz")
rest = strings.TrimSuffix(rest, ".log")
day, err := time.ParseInLocation("2006-01-02", rest, w.location)
if err != nil {
continue
}
if day.Before(cutoff) {
_ = os.Remove(filepath.Join(w.dir, name))
}
}
return nil
}
backend/internal/logging/daily_writer_test.go
+66
View File
@@ -0,0 +1,66 @@
package logging
import (
"os"
"path/filepath"
"testing"
"time"
)
func TestCompressDayAndPurge(t *testing.T) {
dir := t.TempDir()
loc := time.FixedZone("Asia/Shanghai", 8*60*60)
w := newDailyWriter(dir, "app", loc, 2)
// 准备三天明文日志
days := []string{"2026-07-01", "2026-07-10", "2026-07-11"}
for _, day := range days {
path := filepath.Join(dir, "app-"+day+".log")
if err := os.WriteFile(path, []byte(`{"msg":"hello `+day+`"}`+"\n"), 0o644); err != nil {
t.Fatalf("write log: %v", err)
}
}
// 压缩 7-10
if err := w.compressDay("2026-07-10"); err != nil {
t.Fatalf("compressDay: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, "app-2026-07-10.log")); !os.IsNotExist(err) {
t.Fatalf("plain log should be removed after gzip, err=%v", err)
}
if _, err := os.Stat(filepath.Join(dir, "app-2026-07-10.log.gz")); err != nil {
t.Fatalf("gzip missing: %v", err)
}
// 以 7-12 为「现在」，保留 2 天 → cutoff=7-10，删除 7-10 之前（仅 7-01）
now := time.Date(2026, 7, 12, 12, 0, 0, 0, loc)
if err := w.purgeOlderThan(now); err != nil {
t.Fatalf("purgeOlderThan: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, "app-2026-07-01.log")); !os.IsNotExist(err) {
t.Fatal("expired log should be purged")
}
// 7-10 / 7-11 仍应存在（cutoff 当天起保留）
if _, err := os.Stat(filepath.Join(dir, "app-2026-07-10.log.gz")); err != nil {
t.Fatalf("recent gzip should remain: %v", err)
}
if _, err := os.Stat(filepath.Join(dir, "app-2026-07-11.log")); err != nil {
t.Fatalf("recent plain should remain: %v", err)
}
}
func TestCompressDayIdempotent(t *testing.T) {
dir := t.TempDir()
loc := time.UTC
w := newDailyWriter(dir, "app", loc, 0)
path := filepath.Join(dir, "app-2026-07-11.log")
if err := os.WriteFile(path, []byte("line\n"), 0o644); err != nil {
t.Fatal(err)
}
if err := w.compressDay("2026-07-11"); err != nil {
t.Fatal(err)
}
if err := w.compressDay("2026-07-11"); err != nil {
t.Fatalf("second compress should be no-op: %v", err)
}
}
backend/internal/logging/logger.go
+1 -1
View File
@@ -39,7 +39,7 @@ func New(cfg config.LogConfig) (*zap.Logger, error) {
))
}
if cfg.EnableFile {
writer := newDailyWriter(cfg.Dir, "app", location)
writer := newDailyWriter(cfg.Dir, "app", location, cfg.RetainDays)
cores = append(cores, zapcore.NewCore(
zapcore.NewJSONEncoder(encoderConfig),
writer,
backend/internal/middleware/request_logger.go
+56 -9
View File
@@ -8,24 +8,40 @@ import (
"go.uber.org/zap"
)
// 成功且延迟低于该阈值的热路径请求不再写访问日志（错误/慢请求仍全量记录）。
const slowRequestThresholdMs = 200
func RequestLogger(logger *zap.Logger) gin.HandlerFunc {
return func(c *gin.Context) {
start := time.Now()
c.Next()
latency := time.Since(start)
latencyMs := float64(time.Since(start).Microseconds()) / 1000
status := c.Writer.Status()
path := c.Request.URL.Path
route := c.FullPath()
if shouldSkipHTTPLog(path, route, status, latencyMs) {
return
}
fields := []zap.Field{
zap.String("request_id", GetRequestID(c)),
zap.String("method", c.Request.Method),
zap.String("path", c.Request.URL.Path),
zap.String("route", c.FullPath()),
zap.Int("status", c.Writer.Status()),
zap.Float64("latency_ms", float64(latency.Microseconds())/1000),
zap.String("path", path),
zap.String("route", route),
zap.Int("status", status),
zap.Float64("latency_ms", latencyMs),
zap.String("client_ip", c.ClientIP()),
zap.String("user_agent", c.GetHeader("User-Agent")),
zap.String("referer", c.GetHeader("Referer")),
zap.Int("response_size", c.Writer.Size()),
}
// 成功响应省略 UA/referer，错误与 4xx/5xx 保留完整现场
if status >= 400 {
fields = append(fields,
zap.String("user_agent", c.GetHeader("User-Agent")),
zap.String("referer", c.GetHeader("Referer")),
)
}
if userID, ok := c.Get(ContextUserID); ok {
fields = append(fields, zap.Any("user_id", userID))
}
@@ -37,12 +53,43 @@ func RequestLogger(logger *zap.Logger) gin.HandlerFunc {
}
switch {
case c.Writer.Status() >= 500:
case status >= 500:
logger.Error("http request", fields...)
case c.Writer.Status() >= 400:
case status >= 400:
logger.Warn("http request", fields...)
default:
logger.Info("http request", fields...)
}
}
}
// shouldSkipHTTPLog 判断是否跳过写入：仅跳过「成功 + 非慢请求」的高频热路径。
func shouldSkipHTTPLog(path, route string, status int, latencyMs float64) bool {
if status >= 400 {
return false
}
if latencyMs >= slowRequestThresholdMs {
return false
}
return isHotPath(path, route)
}
func isHotPath(path, route string) bool {
if strings.HasSuffix(path, "/unread-count") {
return true
}
if path == "/api/wallet/balance" {
return true
}
if path == "/api/mobile-home-config" {
return true
}
if strings.HasSuffix(path, "/cover") || route == "/api/listings/:id/cover" {
return true
}
if path == "/api/files/object" || path == "/api/admin/files/object" ||
route == "/api/files/object" || route == "/api/admin/files/object" {
return true
}
return false
}
backend/internal/middleware/request_logger_test.go
+33
View File
@@ -0,0 +1,33 @@
package middleware
import "testing"
func TestShouldSkipHTTPLog(t *testing.T) {
tests := []struct {
name string
path string
route string
status int
latencyMs float64
wantSkip bool
}{
{name: "轮询未读成功跳过", path: "/api/chats/unread-count", status: 200, latencyMs: 1, wantSkip: true},
{name: "钱包余额成功跳过", path: "/api/wallet/balance", status: 200, latencyMs: 2, wantSkip: true},
{name: "封面成功跳过", path: "/api/listings/12/cover", route: "/api/listings/:id/cover", status: 200, latencyMs: 3, wantSkip: true},
{name: "文件对象成功跳过", path: "/api/files/object", status: 200, latencyMs: 5, wantSkip: true},
{name: "首页配置成功跳过", path: "/api/mobile-home-config", status: 200, latencyMs: 4, wantSkip: true},
{name: "轮询 401 不跳过", path: "/api/chats/unread-count", status: 401, latencyMs: 1, wantSkip: false},
{name: "轮询 500 不跳过", path: "/api/wallet/balance", status: 500, latencyMs: 1, wantSkip: false},
{name: "轮询慢请求不跳过", path: "/api/chats/unread-count", status: 200, latencyMs: 250, wantSkip: false},
{name: "普通业务成功不跳过", path: "/api/orders", status: 200, latencyMs: 20, wantSkip: false},
{name: "下单成功不跳过", path: "/api/orders", status: 201, latencyMs: 30, wantSkip: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := shouldSkipHTTPLog(tt.path, tt.route, tt.status, tt.latencyMs)
if got != tt.wantSkip {
t.Fatalf("shouldSkipHTTPLog() = %v, want %v", got, tt.wantSkip)
}
})
}
}