sync: 同步飞书机器人 v1.0.4 并脱敏
This commit is contained in:
MiFaZhan
@@ -25,6 +25,10 @@ type Service struct {
|
||||
// warned 记录已经报过错的归档，避免坏文件每轮扫描都刷一条日志。
|
||||
// 键为「路径 + mtime + 大小」：文件一旦被重写就会重新告警。
|
||||
warned map[string]struct{}
|
||||
// transient 记录临时错误的重报节奏与累计次数。
|
||||
// 临时错误必须持续可见（一次网络抖动不能被吞掉），但 bot 被移出群这类
|
||||
// 长期故障不能每秒刷一条，否则日志被淹、真告警失联。
|
||||
transient map[string]*transientWarn
|
||||
}
|
||||
|
||||
// Sender 是发送端抽象：*feishu.Client 满足它，测试可注入替身。
|
||||
@@ -37,7 +41,7 @@ func New(archiveDir, environment string, pollInterval time.Duration, store state
|
||||
if logger == nil {
|
||||
logger = DefaultLogger()
|
||||
}
|
||||
return Service{archiveDir: archiveDir, environment: environment, stateDir: store.Dir(), pollInterval: pollInterval, store: store, client: client, logger: logger, warned: map[string]struct{}{}}
|
||||
return Service{archiveDir: archiveDir, environment: environment, stateDir: store.Dir(), pollInterval: pollInterval, store: store, client: client, logger: logger, warned: map[string]struct{}{}, transient: map[string]*transientWarn{}}
|
||||
}
|
||||
func (s Service) Render(path string) (string, error) {
|
||||
a, err := archive.Load(path)
|
||||
@@ -94,11 +98,19 @@ func isPermanent(err error) bool {
|
||||
return errors.As(err, &ae) && ae.permanent
|
||||
}
|
||||
|
||||
// warnOnce 对同一个归档文件只记一次「跳过」日志。
|
||||
// 只有永久性错误才去重；临时错误原样返回给调用方，保持每轮可见。
|
||||
// warnOnce 对同一个归档文件的重报做节流。
|
||||
//
|
||||
// 永久错误（归档格式坏、旧版本）只记一次：重试也不会好。
|
||||
// 临时错误（飞书超时、bot 不在群）按退避节奏重报：
|
||||
// - 前 transientWarnFastPhase 次每次都报，保证瞬时抖动可见；
|
||||
// - 之后按指数退避（1min→2min→4min…上限 transientWarnMaxInterval）；
|
||||
// - 超过 transientWarnMaxCount 次后只记一条汇总并停止重报，避免日志风暴。
|
||||
//
|
||||
// 历史上 230002（bot 不在群）持续 10 小时刷了 18,266 条 ERROR，占日志 99.9%，
|
||||
// 把真正的告警完全淹没——这正是本函数要防的情况。
|
||||
func (s Service) warnOnce(path string, err error) {
|
||||
if !isPermanent(err) {
|
||||
s.logger.Error("process archive failed", "path", path, "error", err)
|
||||
s.warnTransient(path, err)
|
||||
return
|
||||
}
|
||||
key := skipKey(path)
|
||||
@@ -111,6 +123,69 @@ func (s Service) warnOnce(path string, err error) {
|
||||
s.logger.Error("skip archive", "path", path, "error", err)
|
||||
}
|
||||
|
||||
const (
|
||||
// transientWarnFastPhase 是「每次都报」的次数，覆盖瞬时抖动。
|
||||
transientWarnFastPhase = 5
|
||||
// transientWarnMaxInterval 是退避重报的间隔上限。
|
||||
transientWarnMaxInterval = 10 * time.Minute
|
||||
// transientWarnMaxCount 是同一归档同一错误的最大重报次数，超过后静默。
|
||||
transientWarnMaxCount = 30
|
||||
)
|
||||
|
||||
// transientWarn 记录一个归档的临时错误重报状态。
|
||||
type transientWarn struct {
|
||||
count int
|
||||
lastLog time.Time
|
||||
suppressed bool
|
||||
}
|
||||
|
||||
// warnTransient 按退避节奏记录临时错误。
|
||||
func (s Service) warnTransient(path string, err error) {
|
||||
key := skipKey(path)
|
||||
now := time.Now()
|
||||
|
||||
if s.transient == nil {
|
||||
s.transient = map[string]*transientWarn{}
|
||||
}
|
||||
st, ok := s.transient[key]
|
||||
if !ok {
|
||||
st = &transientWarn{}
|
||||
s.transient[key] = st
|
||||
}
|
||||
|
||||
if st.suppressed {
|
||||
return
|
||||
}
|
||||
|
||||
// 前 fastPhase 次每次都报。
|
||||
if st.count < transientWarnFastPhase {
|
||||
st.count++
|
||||
st.lastLog = now
|
||||
s.logger.Error("process archive failed", "path", path, "error", err, "attempt", st.count)
|
||||
return
|
||||
}
|
||||
|
||||
// 之后按指数退避。
|
||||
interval := time.Minute << uint(st.count-transientWarnFastPhase)
|
||||
if interval > transientWarnMaxInterval {
|
||||
interval = transientWarnMaxInterval
|
||||
}
|
||||
if now.Sub(st.lastLog) < interval {
|
||||
return
|
||||
}
|
||||
|
||||
st.count++
|
||||
st.lastLog = now
|
||||
if st.count > transientWarnMaxCount {
|
||||
st.suppressed = true
|
||||
s.logger.Error("process archive failed, suppressing further identical alerts",
|
||||
"path", path, "error", err, "attempts", st.count-1,
|
||||
"hint", "this is likely a persistent configuration problem (e.g. bot removed from chat), fix the cause to resume notifications")
|
||||
return
|
||||
}
|
||||
s.logger.Error("process archive failed", "path", path, "error", err, "attempt", st.count)
|
||||
}
|
||||
|
||||
// Scan 扫描归档目录并发送尚未通知过的 Run。
|
||||
//
|
||||
// 单个归档出错（旧格式、写了一半、权限不足）只跳过并记日志：
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-bot/internal/state"
|
||||
)
|
||||
|
||||
// captureHandler 收集所有日志记录，用于断言重报次数。
|
||||
type captureHandler struct {
|
||||
mu sync.Mutex
|
||||
records []slog.Record
|
||||
}
|
||||
|
||||
func (h *captureHandler) Enabled(_ context.Context, _ slog.Level) bool { return true }
|
||||
|
||||
func (h *captureHandler) Handle(_ context.Context, r slog.Record) error {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
h.records = append(h.records, r)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *captureHandler) WithAttrs(_ []slog.Attr) slog.Handler { return h }
|
||||
func (h *captureHandler) WithGroup(_ string) slog.Handler { return h }
|
||||
|
||||
func (h *captureHandler) count() int {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return len(h.records)
|
||||
}
|
||||
|
||||
func (h *captureHandler) lastMessage() string {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if len(h.records) == 0 {
|
||||
return ""
|
||||
}
|
||||
return h.records[len(h.records)-1].Message
|
||||
}
|
||||
|
||||
func newCaptureLogger() (*slog.Logger, *captureHandler) {
|
||||
h := &captureHandler{}
|
||||
return slog.New(h), h
|
||||
}
|
||||
|
||||
// TestTransientWarnThrottlesRepeatedErrors 验证同一个归档的临时错误
|
||||
// 不会每轮都刷一条 ERROR：前几次可见，之后进入退避，超过上限后静默。
|
||||
//
|
||||
// 背景：2026-10-21/22 bot 被移出飞书群 10 小时，230002 错误刷了 18,266 条，
|
||||
// 占日志 99.9%，真正的同步结果通知被完全淹没。
|
||||
func TestTransientWarnThrottlesRepeatedErrors(t *testing.T) {
|
||||
logger, h := newCaptureLogger()
|
||||
svc := New(t.TempDir(), "test", time.Second, state.New(t.TempDir()), nil, logger)
|
||||
|
||||
transient := errors.New("feishu send failed: http=400 code=230002 msg=Bot/User can NOT be out of the chat.")
|
||||
|
||||
// 模拟 500 轮扫描都失败（约等于 8 分钟的 1 秒轮询）。
|
||||
for i := 0; i < 500; i++ {
|
||||
svc.warnOnce("/data/full-sync/run-abc.json", transient)
|
||||
}
|
||||
|
||||
got := h.count()
|
||||
// 前 5 次必报，之后受退避限制。500 轮里退避间隔从 1 分钟起翻倍，
|
||||
// 最多也就多出个位数。这里给一个宽松上界，重点是不能是 500。
|
||||
if got > transientWarnMaxCount+2 {
|
||||
t.Fatalf("logged %d times for 500 failures, want <= %d (throttle is not working)", got, transientWarnMaxCount+2)
|
||||
}
|
||||
if got < transientWarnFastPhase {
|
||||
t.Fatalf("logged %d times, want at least %d (fast phase should always report)", got, transientWarnFastPhase)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTransientWarnDifferentArchivesAreIndependent 验证节流按归档文件隔离，
|
||||
// 一个坏文件不能掩盖另一个文件的错误。
|
||||
func TestTransientWarnDifferentArchivesAreIndependent(t *testing.T) {
|
||||
logger, h := newCaptureLogger()
|
||||
svc := New(t.TempDir(), "test", time.Second, state.New(t.TempDir()), nil, logger)
|
||||
|
||||
errA := errors.New("boom-a")
|
||||
for i := 0; i < 20; i++ {
|
||||
svc.warnOnce("/data/full-sync/run-a.json", errA)
|
||||
}
|
||||
afterA := h.count()
|
||||
|
||||
// 另一个文件应该独立计数，立刻能看到自己的错误。
|
||||
svc.warnOnce("/data/full-sync/run-b.json", errors.New("boom-b"))
|
||||
if h.count() != afterA+1 {
|
||||
t.Fatalf("second archive was throttled by the first: before=%d after=%d", afterA, h.count())
|
||||
}
|
||||
}
|
||||
|
||||
// TestPermanentWarnOnlyOnce 验证永久错误仍然只报一次。
|
||||
func TestPermanentWarnOnlyOnce(t *testing.T) {
|
||||
logger, h := newCaptureLogger()
|
||||
svc := New(t.TempDir(), "test", time.Second, state.New(t.TempDir()), nil, logger)
|
||||
|
||||
permanent := permanentErr(errors.New("archive rounds is empty"))
|
||||
for i := 0; i < 50; i++ {
|
||||
svc.warnOnce("/data/full-sync/run-bad.json", permanent)
|
||||
}
|
||||
|
||||
if got := h.count(); got != 1 {
|
||||
t.Fatalf("permanent error logged %d times, want exactly 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTransientWarnSuppressionMessage 验证超过上限时给出的汇总日志包含排查提示，
|
||||
// 让运维一眼看出这是配置问题而不是代码 bug。
|
||||
func TestTransientWarnSuppressionMessage(t *testing.T) {
|
||||
logger, h := newCaptureLogger()
|
||||
svc := New(t.TempDir(), "test", time.Second, state.New(t.TempDir()), nil, logger)
|
||||
|
||||
transient := errors.New("code=230002 Bot/User can NOT be out of the chat.")
|
||||
// 直接构造已进入退避阶段的状态，避免测试里真的等几分钟。
|
||||
key := skipKey("/data/full-sync/run-abc.json")
|
||||
svc.transient[key] = &transientWarn{
|
||||
count: transientWarnMaxCount,
|
||||
lastLog: time.Now().Add(-time.Hour),
|
||||
}
|
||||
|
||||
svc.warnOnce("/data/full-sync/run-abc.json", transient)
|
||||
|
||||
msg := h.lastMessage()
|
||||
if !strings.Contains(msg, "suppressing") {
|
||||
t.Fatalf("suppression message = %q, want it to mention suppressing", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTransientWarnZeroValueService 验证未初始化的 Service（transient 为 nil）
|
||||
// 不会 panic——生产上 Service 由 New 构造，但零值也应安全。
|
||||
func TestTransientWarnZeroValueService(t *testing.T) {
|
||||
svc := Service{logger: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError + 10}))}
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
t.Fatalf("zero-value Service panicked: %v", r)
|
||||
}
|
||||
}()
|
||||
svc.warnOnce("/tmp/run-x.json", errors.New("transient"))
|
||||
}
|
||||
Reference in New Issue
Block a user