From 24ae53bc0be326ba7778b0f40eef422e2828e0ee Mon Sep 17 00:00:00 2001 From: MiFaZhan Date: Thu, 29 Oct 2026 14:19:44 +0800 Subject: [PATCH] =?UTF-8?q?sync:=20=E5=90=8C=E6=AD=A5=E9=A3=9E=E4=B9=A6?= =?UTF-8?q?=E6=9C=BA=E5=99=A8=E4=BA=BA=20v1.0.4=20=E5=B9=B6=E8=84=B1?= =?UTF-8?q?=E6=95=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 6 +- Dockerfile | 13 +-- README.md | 157 ++++++++++---------------- config.example.yaml | 10 ++ config.yaml | 8 -- internal/notify/service.go | 83 +++++++++++++- internal/notify/warn_throttle_test.go | 147 ++++++++++++++++++++++++ 7 files changed, 305 insertions(+), 119 deletions(-) create mode 100644 config.example.yaml delete mode 100644 config.yaml create mode 100644 internal/notify/warn_throttle_test.go diff --git a/.gitignore b/.gitignore index 0167efe..9df9e0f 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,8 @@ bin/ *.exe coverage.out /state/ -*.log \ No newline at end of file +*.log +# Deployment secrets must never be checked into the public repository. +config.yaml +config.*.yaml +!config.example.yaml diff --git a/Dockerfile b/Dockerfile index 822629b..7170d9e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,22 +1,17 @@ -FROM golang:1.25-alpine AS build - -ARG GOPROXY=https://goproxy.cn,https://goproxy.io,https://proxy.golang.org,direct +ARG BASE_REGISTRY=docker.io/library +FROM ${BASE_REGISTRY}/golang:1.25-alpine AS build +ARG GOPROXY=https://goproxy.cn|https://goproxy.io|https://proxy.golang.org|direct ENV GOPROXY=${GOPROXY} - WORKDIR /src - COPY go.mod go.sum ./ +RUN apk add --no-cache ca-certificates RUN go mod download - COPY . ./ - RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags='-s -w' -o /out/iotlcm-bot ./cmd/iotlcm-bot FROM scratch COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt COPY --from=build /out/iotlcm-bot /usr/local/bin/iotlcm-bot - USER 65532:65532 - ENTRYPOINT ["/usr/local/bin/iotlcm-bot"] diff --git a/README.md b/README.md index 12da0b4..64077d8 100644 --- a/README.md +++ b/README.md @@ -1,127 +1,90 @@ # iotlcm-bot -独立飞书应用机器人：读取 Worker 终态 Run JSON，渲染并发送全量同步结果卡片。 - -本服务是 [iotlcm](https://git.5iot.com/iotlcm-public) 物联网卡全生命周期管理平台的配套组件，用于把异步全量同步（卡状态、流量、IMEI 等）的执行结果推送到飞书群。 - -## 特性 - -- **零外部依赖**：只依赖 `gopkg.in/yaml.v3`，不需要 Redis、数据库或 Worker HTTP 接口 -- **只读归档**：只读挂载 Worker 的归档目录，按 `runId` 去重后发送 -- **多类卡片**：全量同步汇总卡、告警卡、供料提醒卡 -- **健康自检**：检测归档目录被宿主删除（inode 丢失）这类静默故障并告警 -- **单镜像**：`scratch` 基础镜像，非 root 运行 +V1 独立飞书应用机器人：读取 Worker 终态 Run JSON，渲染并发送全量同步结果卡片。 ## V1 范围 -- 只接受 `version=1` 且包含 `run` 和非空 `rounds` 的归档 JSON -- 不兼容旧版 `rounds=[]` 文件 -- 不读取 Redis、Worker HTTP 接口或日志 -- 不实现 Replay、事件回调、卡片按钮和错误聚合 +- 只接受当前版本 `version=1` 且包含 `run` 和非空 `rounds` 的归档 JSON。 +- 不兼容旧版 `rounds=[]` 文件。 +- 不读取 Redis、Worker HTTP 接口或日志。 +- 不实现 Replay、事件回调、卡片按钮和错误聚合。 +- 按 `runId` 去重，避免重复发送。 -## 快速开始 +## 本地验证 ```bash -go mod tidy go test ./... -``` - -本地渲染一张卡片（不调用飞书 API）： - -```bash go run ./cmd/iotlcm-bot render --file testdata/current-run.json go run ./cmd/iotlcm-bot render --file testdata/current-run.json --json ``` -## 配置 +`render` 只渲染通知内容，不调用飞书 API。 -配置文件为 `config.yaml`。**公开版里的飞书凭据是占位符**，真实值请用环境变量或本地覆盖文件注入： +## 本地单次发送 -```yaml -environment: preview -archiveDir: /data/full-sync -stateDir: ./state -pollInterval: 10s -feishu: - appId: "${FEISHU_APP_ID}" - appSecret: "${FEISHU_APP_SECRET}" - chatId: "${FEISHU_CHAT_ID}" -``` - -| 配置项 | 说明 | -| --- | --- | -| `archiveDir` | Worker 归档目录（只读挂载） | -| `stateDir` | 已发送 Run 的标记目录（可写） | -| `pollInterval` | 扫描间隔，默认 `10s` | -| `feishu.appId` / `appSecret` | 飞书应用凭据 | -| `feishu.chatId` | 目标群 ID | - -## 命令 - -| 命令 | 作用 | -| --- | --- | -| `watch` | 常驻扫描归档目录，自动发送新 Run（默认模式） | -| `send` | 发送指定 Run JSON | -| `send-text` | 向配置里的群发一条纯文本（用于连通性验证） | -| `render` | 只渲染卡片内容，不调用飞书 API | +机器人加入测试群，且 YAML 配置已填写 `feishu.appSecret` 和 `feishu.chatId` 后，可以发送指定 Run： ```bash -go run ./cmd/iotlcm-bot watch --config /path/to/iotlcm-bot.yaml go run ./cmd/iotlcm-bot send --config /path/to/iotlcm-bot.yaml --file /path/to/run.json +``` + +发送成功后会写入 `state/sent/.json`；同一 Run 再次执行会跳过。确需重发时使用： + +```bash go run ./cmd/iotlcm-bot send --config /path/to/iotlcm-bot.yaml --file /path/to/run.json --force ``` -`--force` 用于确需重发时；正常运行不要使用。发送成功会写入 `state/sent/.json`，同一 Run 再次执行会跳过。 +正常运行时不要使用 `--force`。 -## 容器化部署 +## 已部署容器的发送测试 + +只验证飞书连接时，无需提供 Run JSON。Bot 容器运行后，发送一条自定义文本： ```bash -docker build -t iotlcm-bot:latest . -docker run -d \ - -v /host/worker/full-sync:/data/full-sync:ro \ - -v /host/bot/state:/data/state \ - -v /host/bot/config.yaml:/app/config/config.yaml:ro \ - iotlcm-bot:latest +docker exec iotlcm-bot /usr/local/bin/iotlcm-bot send-text --config /app/config/config.yaml --text "iotlcm-bot 飞书连接测试" ``` -容器内路径约定： +`send-text` 不扫描归档、不写 Run 发送标记，可以重复执行。它直接通过飞书消息 API 向配置文件中的群发送文本，不是 HTTP 测试接口。 -| 容器内路径 | 用途 | 挂载方式 | -| --- | --- | --- | -| `/data/full-sync` | Worker 归档目录 | **只读** | -| `/data/state` | 已发送标记 | 可写 | -| `/app/config/config.yaml` | 配置文件 | 只读 | +需要测试完整的 Run 卡片链路时，才使用下面的单次发送命令： -## 目录结构 +```bash +docker exec iotlcm-bot /usr/local/bin/iotlcm-bot send --config /app/config/config.yaml --file /data/full-sync/run-.json +``` + +如该 Run 已发送，明确需要再次发送时加 `--force`： + +```bash +docker exec iotlcm-bot /usr/local/bin/iotlcm-bot send --config /app/config/config.yaml --file /data/full-sync/run-.json --force +``` + +发送成功会写入 `/data/state/sent/.json`。检查容器和日志： + +```bash +docker ps --filter name=iotlcm-bot +docker logs --tail 100 iotlcm-bot +``` + +## 配置文件 + +复制 `config.example.yaml` 为未跟踪的 `config.yaml`，填入实际飞书应用凭据和群 ID。 +配置加载器直接读取 YAML，不展开环境变量；请勿把真实凭据提交到仓库。 +部署时通过只读挂载将配置注入容器，容器内路径可为 `/app/config/config.yaml`。 + +## 自动通知 + +Compose 默认以 `watch` 模式启动 Bot。Bot 每 10 秒扫描挂载的终态 Run JSON，校验通过且未发送过的 Run 会自动发到配置的飞书群。 + +本地启动方式： + +```bash +go run ./cmd/iotlcm-bot watch --config /path/to/iotlcm-bot.yaml +``` + +## 目录挂载 ```text -iotlcm-bot/ -├── cmd/iotlcm-bot/main.go # CLI 入口 -├── internal/ -│ ├── archive/ # 归档 JSON 解析 -│ ├── config/ # 配置加载 -│ ├── feishu/ # 飞书 API 客户端 -│ ├── notify/ # 通知服务与健康自检 -│ ├── render/ # 卡片渲染 -│ └── state/ # 已发送标记 -├── config.yaml # 配置（公开版为占位值） -├── testdata/ # 测试用归档样例 -└── Dockerfile -``` - -## 常见问题 - -**卡片不发送？** 检查归档目录是否可读、`runId` 是否已发送过（`state/sent/` 下有标记则跳过）、群机器人是否被移出。 - -**日志刷满？** `watch` 模式对同一文件只记一次错误，文件被重写后会重新记录。 - -**归档目录被宿主删了？** Bot 会检测到（inode 丢失）并降级告警——此时应检查宿主的日志清理脚本是否误删了挂载源目录。 - -## 相关项目 - -- [iotlcm-worker-svr](https://git.5iot.com/iotlcm-public/iotlcm-worker-svr) —— 异步执行中枢，产出归档 JSON -- [iotlcm-openapi](https://git.5iot.com/iotlcm-public/iotlcm-openapi) —— 对外 API 网关 - -## License - -[Apache License 2.0](./LICENSE) +Worker 宿主归档目录： +Bot 只读挂载目录：/data/full-sync +Bot 可写状态目录： -> /data/state +``` \ No newline at end of file diff --git a/config.example.yaml b/config.example.yaml new file mode 100644 index 0000000..788aa1b --- /dev/null +++ b/config.example.yaml @@ -0,0 +1,10 @@ +# Copy to config.yaml (untracked) and replace these placeholders before sending. +# The bot reads YAML literally; it does not expand environment variables. +environment: preview +archiveDir: /data/full-sync +stateDir: ./state +pollInterval: 10s +feishu: + appId: "YOUR_FEISHU_APP_ID" + appSecret: "YOUR_FEISHU_APP_SECRET" + chatId: "YOUR_FEISHU_CHAT_ID" diff --git a/config.yaml b/config.yaml deleted file mode 100644 index 0206d91..0000000 --- a/config.yaml +++ /dev/null @@ -1,8 +0,0 @@ -environment: preview -archiveDir: /data/full-sync -stateDir: ./state -pollInterval: 10s -feishu: - appId: "${FEISHU_APP_ID}" - appSecret: "${FEISHU_APP_SECRET}" - chatId: "${FEISHU_CHAT_ID}" \ No newline at end of file diff --git a/internal/notify/service.go b/internal/notify/service.go index d39712a..eccf3df 100644 --- a/internal/notify/service.go +++ b/internal/notify/service.go @@ -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。 // // 单个归档出错（旧格式、写了一半、权限不足）只跳过并记日志： diff --git a/internal/notify/warn_throttle_test.go b/internal/notify/warn_throttle_test.go new file mode 100644 index 0000000..2e5fb66 --- /dev/null +++ b/internal/notify/warn_throttle_test.go @@ -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")) +}