From e687ca3e9d9c943bb7b1edb0f0622b783c671963 Mon Sep 17 00:00:00 2001 From: li Date: Fri, 7 Aug 2026 18:45:05 +0800 Subject: [PATCH] =?UTF-8?q?fix(ops):=20=E7=B3=BB=E7=BB=9F=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E8=90=BD=E5=BA=93=E5=A4=B1=E8=B4=A5=E5=90=8E=E9=80=80?= =?UTF-8?q?=E9=81=BF=E9=87=8D=E8=AF=95=EF=BC=8C=E9=81=BF=E5=85=8D=E6=8B=96?= =?UTF-8?q?=E5=9E=AE=E6=95=B0=E6=8D=AE=E5=BA=93=E8=BF=9E=E6=8E=A5=E6=B1=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OpsSystemLogSink 的 flush 没有任何失败退避：ticker 每 1 秒触发一次，每次都 带 5 秒超时打一批 COPY，失败后立刻在下一个 tick 再来一次。远程 PostgreSQL 上 COPY 被 context 取消会导致协议失步、连接被服务端 FATAL 终止，于是这条循环 变成持续的「占用连接 → 超时取消 → 销毁重建」；在 max_open 较小的连接池里， 日志通道会长期占满连接，业务侧最终报 Billing 503。 改为连续失败后指数退避（2s 起，翻倍，封顶 60s）： - 退避窗口内直接丢弃当前批次并计入 dropped_count，不再触达数据库； - 任意一次成功立即清空失败计数与抑制窗口； - 失败日志随退避天然收敛为每个窗口至多一条，并附带 failures/backoff。 Fixes #5265 --- .../internal/service/ops_system_log_sink.go | 71 +++++- .../ops_system_log_sink_backoff_test.go | 225 ++++++++++++++++++ 2 files changed, 287 insertions(+), 9 deletions(-) create mode 100644 backend/internal/service/ops_system_log_sink_backoff_test.go diff --git a/backend/internal/service/ops_system_log_sink.go b/backend/internal/service/ops_system_log_sink.go index 1cbac237c4..7a247a290d 100644 --- a/backend/internal/service/ops_system_log_sink.go +++ b/backend/internal/service/ops_system_log_sink.go @@ -34,6 +34,10 @@ type OpsSystemLogSink struct { batchSize int flushInterval time.Duration + // 连续写入失败后的退避参数。构造后只读，测试可在 Start 前覆盖。 + flushBackoff time.Duration + flushBackoffMax time.Duration + ctx context.Context cancel context.CancelFunc wg sync.WaitGroup @@ -48,22 +52,54 @@ type OpsSystemLogSink struct { const maxSystemLogHostLength = 255 +const ( + // 首次写入失败后暂停落库的时长，之后逐次翻倍到上限。 + defaultOpsSystemLogFlushBackoff = 2 * time.Second + // 退避上限。日志是尽力而为的观测数据，不值得为它无限期占用连接池。 + defaultOpsSystemLogFlushBackoffMax = 60 * time.Second +) + func NewOpsSystemLogSink(opsRepo OpsRepository) *OpsSystemLogSink { ctx, cancel := context.WithCancel(context.Background()) rawHost, err := os.Hostname() s := &OpsSystemLogSink{ - opsRepo: opsRepo, - host: normalizeSystemLogHost(rawHost, err), - queue: make(chan *logger.LogEvent, 5000), - batchSize: 200, - flushInterval: time.Second, - ctx: ctx, - cancel: cancel, + opsRepo: opsRepo, + host: normalizeSystemLogHost(rawHost, err), + queue: make(chan *logger.LogEvent, 5000), + batchSize: 200, + flushInterval: time.Second, + flushBackoff: defaultOpsSystemLogFlushBackoff, + flushBackoffMax: defaultOpsSystemLogFlushBackoffMax, + ctx: ctx, + cancel: cancel, } s.lastError.Store("") return s } +// flushBackoffFor 返回第 failures 次连续失败后的退避时长（指数退避，封顶）。 +func (s *OpsSystemLogSink) flushBackoffFor(failures int) time.Duration { + base := s.flushBackoff + if base <= 0 { + base = defaultOpsSystemLogFlushBackoff + } + maxBackoff := s.flushBackoffMax + if maxBackoff <= 0 { + maxBackoff = defaultOpsSystemLogFlushBackoffMax + } + if maxBackoff < base { + maxBackoff = base + } + backoff := base + for i := 1; i < failures && backoff < maxBackoff; i++ { + backoff *= 2 + } + if backoff > maxBackoff { + backoff = maxBackoff + } + return backoff +} + func normalizeSystemLogHost(host string, err error) string { host = strings.TrimSpace(host) if err != nil || host == "" { @@ -146,20 +182,37 @@ func (s *OpsSystemLogSink) run() { defer ticker.Stop() batch := make([]*logger.LogEvent, 0, s.batchSize) + // 仅在本 goroutine 内读写，无需加锁。 + failures := 0 + var suppressedUntil time.Time flush := func(baseCtx context.Context) { if len(batch) == 0 { return } + now := time.Now() + if now.Before(suppressedUntil) { + // 退避窗口内直接丢弃本批：日志是尽力而为的观测数据，继续攒批只会把 + // 压力转移到内存，而每次重试都会再占用并取消一条池内连接。 + atomic.AddUint64(&s.droppedCount, uint64(len(batch))) + batch = batch[:0] + return + } started := time.Now() inserted, err := s.flushBatch(baseCtx, batch) delay := time.Since(started) if err != nil { + failures++ + backoff := s.flushBackoffFor(failures) + suppressedUntil = time.Now().Add(backoff) atomic.AddUint64(&s.writeFailed, uint64(len(batch))) s.lastError.Store(err.Error()) - _, _ = fmt.Fprintf(os.Stderr, "time=%s level=WARN msg=\"ops system log sink flush failed\" err=%v batch=%d\n", - time.Now().Format(time.RFC3339Nano), err, len(batch), + // 每个退避窗口至多一条，避免数据库故障期间刷屏。 + _, _ = fmt.Fprintf(os.Stderr, "time=%s level=WARN msg=\"ops system log sink flush failed\" err=%v batch=%d failures=%d backoff=%s\n", + time.Now().Format(time.RFC3339Nano), err, len(batch), failures, backoff, ) } else { + failures = 0 + suppressedUntil = time.Time{} atomic.AddUint64(&s.writtenCount, uint64(inserted)) atomic.AddUint64(&s.totalDelayNs, uint64(delay.Nanoseconds())) s.lastError.Store("") diff --git a/backend/internal/service/ops_system_log_sink_backoff_test.go b/backend/internal/service/ops_system_log_sink_backoff_test.go new file mode 100644 index 0000000000..3a29cb6d0d --- /dev/null +++ b/backend/internal/service/ops_system_log_sink_backoff_test.go @@ -0,0 +1,225 @@ +package service + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/Wei-Shaw/sub2api/internal/pkg/logger" +) + +func opsSystemLogBackoffEvent() *logger.LogEvent { + return &logger.LogEvent{ + Time: time.Now().UTC(), + Level: "warn", + Component: "app", + Message: "boom", + Fields: map[string]any{}, + } +} + +// pumpOpsSystemLogEvents 持续投递日志事件，模拟故障期间业务侧不断产生 WARN/ERROR。 +func pumpOpsSystemLogEvents(t *testing.T, sink *OpsSystemLogSink) { + t.Helper() + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + for { + select { + case <-stop: + return + default: + } + sink.WriteLogEvent(opsSystemLogBackoffEvent()) + time.Sleep(2 * time.Millisecond) + } + }() + t.Cleanup(func() { + close(stop) + <-done + }) +} + +func waitForOpsSystemLogCondition(t *testing.T, timeout time.Duration, cond func() bool) bool { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return true + } + time.Sleep(5 * time.Millisecond) + } + return cond() +} + +func TestOpsSystemLogSinkFlushBackoffFor(t *testing.T) { + sink := &OpsSystemLogSink{flushBackoff: time.Second, flushBackoffMax: 8 * time.Second} + + cases := []struct { + name string + failures int + want time.Duration + }{ + {"first_failure", 1, time.Second}, + {"second_failure", 2, 2 * time.Second}, + {"third_failure", 3, 4 * time.Second}, + {"fourth_failure", 4, 8 * time.Second}, + {"capped", 9, 8 * time.Second}, + {"large_streak_does_not_overflow", 1000, 8 * time.Second}, + {"zero_streak_uses_base", 0, time.Second}, + {"negative_streak_uses_base", -1, time.Second}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := sink.flushBackoffFor(tc.failures); got != tc.want { + t.Fatalf("flushBackoffFor(%d) = %s, want %s", tc.failures, got, tc.want) + } + }) + } +} + +// 未显式配置时回落到默认值；max 小于 base 时以 base 为准，不产生比基准还短的退避。 +func TestOpsSystemLogSinkFlushBackoffForFallbacks(t *testing.T) { + zero := &OpsSystemLogSink{} + if got := zero.flushBackoffFor(1); got != defaultOpsSystemLogFlushBackoff { + t.Fatalf("zero-value base = %s, want %s", got, defaultOpsSystemLogFlushBackoff) + } + if got := zero.flushBackoffFor(100); got != defaultOpsSystemLogFlushBackoffMax { + t.Fatalf("zero-value cap = %s, want %s", got, defaultOpsSystemLogFlushBackoffMax) + } + + inverted := &OpsSystemLogSink{flushBackoff: 5 * time.Second, flushBackoffMax: time.Second} + if got := inverted.flushBackoffFor(3); got != 5*time.Second { + t.Fatalf("inverted bounds = %s, want 5s", got) + } +} + +// issue #5265：写入失败后如果按 flushInterval 继续每秒重试，每一轮都会占用并取消 +// 一条池内连接（远程 PG 上 COPY 取消会让连接协议失步而被销毁），小连接池会被日志 +// 通道长期占满，业务侧最终报 Billing 503。失败后必须退避。 +func TestOpsSystemLogSinkSuppressesRetriesDuringBackoff(t *testing.T) { + var calls int64 + repo := &opsRepoMock{ + BatchInsertSystemLogsFn: func(_ context.Context, _ []*OpsInsertSystemLogInput) (int64, error) { + atomic.AddInt64(&calls, 1) + return 0, errors.New("db unavailable") + }, + } + + sink := NewOpsSystemLogSink(repo) + sink.batchSize = 1 + sink.flushInterval = 5 * time.Millisecond + sink.flushBackoff = 800 * time.Millisecond + sink.flushBackoffMax = 800 * time.Millisecond + sink.Start() + defer sink.Stop() + pumpOpsSystemLogEvents(t, sink) + + if !waitForOpsSystemLogCondition(t, 2*time.Second, func() bool { return atomic.LoadInt64(&calls) >= 1 }) { + t.Fatalf("first flush never happened") + } + + // 退避窗口内不得再打上游：修复前 5ms 的 flushInterval 会在这 300ms 里打出上百次。 + time.Sleep(300 * time.Millisecond) + if got := atomic.LoadInt64(&calls); got != 1 { + t.Fatalf("upstream calls during backoff = %d, want 1", got) + } + + // 被抑制的批次记为 dropped，而不是 write_failed —— 没有尝试过就不算写失败。 + health := sink.Health() + if health.DroppedCount == 0 { + t.Fatalf("dropped_count should grow while flushing is suppressed") + } + if health.WriteFailed == 0 || health.LastError == "" { + t.Fatalf("failed flush should still surface in health: %+v", health) + } +} + +// 退避到期后必须自动恢复，不能变成永久停写。 +func TestOpsSystemLogSinkResumesAfterBackoffWindow(t *testing.T) { + var calls int64 + repo := &opsRepoMock{ + BatchInsertSystemLogsFn: func(_ context.Context, _ []*OpsInsertSystemLogInput) (int64, error) { + atomic.AddInt64(&calls, 1) + return 0, errors.New("db unavailable") + }, + } + + sink := NewOpsSystemLogSink(repo) + sink.batchSize = 1 + sink.flushInterval = 5 * time.Millisecond + sink.flushBackoff = 150 * time.Millisecond + sink.flushBackoffMax = 150 * time.Millisecond + sink.Start() + defer sink.Stop() + pumpOpsSystemLogEvents(t, sink) + + if !waitForOpsSystemLogCondition(t, 3*time.Second, func() bool { return atomic.LoadInt64(&calls) >= 3 }) { + t.Fatalf("sink did not resume flushing after backoff, calls=%d", atomic.LoadInt64(&calls)) + } +} + +// 一次成功必须清空失败计数与抑制窗口，否则短暂抖动后写入速率无法恢复。 +func TestOpsSystemLogSinkSuccessClearsSuppression(t *testing.T) { + var calls int64 + repo := &opsRepoMock{ + BatchInsertSystemLogsFn: func(_ context.Context, inputs []*OpsInsertSystemLogInput) (int64, error) { + if atomic.AddInt64(&calls, 1) == 1 { + return 0, errors.New("db unavailable") + } + return int64(len(inputs)), nil + }, + } + + sink := NewOpsSystemLogSink(repo) + sink.batchSize = 1 + sink.flushInterval = 5 * time.Millisecond + sink.flushBackoff = 150 * time.Millisecond + sink.flushBackoffMax = 150 * time.Millisecond + sink.Start() + defer sink.Stop() + pumpOpsSystemLogEvents(t, sink) + + // 第 2 次调用（恢复后的首次成功）之后不应再有抑制：调用次数需要快速爬升。 + if !waitForOpsSystemLogCondition(t, 3*time.Second, func() bool { return atomic.LoadInt64(&calls) >= 2 }) { + t.Fatalf("sink never retried after the first failure") + } + recovered := atomic.LoadInt64(&calls) + time.Sleep(200 * time.Millisecond) + if got := atomic.LoadInt64(&calls) - recovered; got < 3 { + t.Fatalf("calls after recovery = %d in 200ms, want >=3 (suppression not cleared)", got) + } + if sink.Health().WrittenCount == 0 { + t.Fatalf("written_count should grow after recovery") + } +} + +// 健康路径不受影响：一直成功就永远不进入退避。 +func TestOpsSystemLogSinkHealthyPathNeverSuppressed(t *testing.T) { + var calls int64 + repo := &opsRepoMock{ + BatchInsertSystemLogsFn: func(_ context.Context, inputs []*OpsInsertSystemLogInput) (int64, error) { + atomic.AddInt64(&calls, 1) + return int64(len(inputs)), nil + }, + } + + sink := NewOpsSystemLogSink(repo) + sink.batchSize = 1 + sink.flushInterval = 5 * time.Millisecond + sink.flushBackoff = time.Hour + sink.flushBackoffMax = time.Hour + sink.Start() + defer sink.Stop() + pumpOpsSystemLogEvents(t, sink) + + if !waitForOpsSystemLogCondition(t, 3*time.Second, func() bool { return atomic.LoadInt64(&calls) >= 10 }) { + t.Fatalf("healthy sink should keep flushing, calls=%d", atomic.LoadInt64(&calls)) + } + if got := sink.Health().DroppedCount; got != 0 { + t.Fatalf("healthy sink dropped_count = %d, want 0", got) + } +}