From 5732bf3384439c852472b3259066ab5c062beba0 Mon Sep 17 00:00:00 2001 From: MiFaZhan Date: Thu, 22 Oct 2026 16:14:24 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=88=9D=E5=A7=8B=E5=BC=80=E6=BA=90?= =?UTF-8?q?=E5=8F=91=E5=B8=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit iotlcm-bot：独立飞书应用机器人（读取 Worker 终态 Run JSON，渲染并发送全量同步结果卡片）。 - Go module：git.5iot.com/iotlcm/iotlcm-bot → git.5iot.com/iotlcm-public/iotlcm-bot - 飞书凭据占位：appId/appSecret/chatId → ${FEISHU_APP_ID}/${FEISHU_APP_SECRET}/${FEISHU_CHAT_ID} - 移除内部资产：CI 触发器 - Dockerfile 改用公共基础镜像（golang:1.25-alpine → scratch） - README 重写（命令/配置/挂载契约/常见问题），去掉内部环境路径 - 补 Apache-2.0 LICENSE，LF 行尾 + .gitattributes --- .gitattributes | 13 ++ .gitignore | 8 + Dockerfile | 22 +++ LICENSE | 201 +++++++++++++++++++++ README.md | 127 +++++++++++++ cmd/iotlcm-bot/main.go | 137 ++++++++++++++ config.yaml | 8 + go.mod | 5 + go.sum | 4 + internal/archive/model.go | 159 +++++++++++++++++ internal/archive/model_test.go | 48 +++++ internal/archive/supply.go | 47 +++++ internal/config/config.go | 63 +++++++ internal/config/config_test.go | 32 ++++ internal/feishu/client.go | 155 ++++++++++++++++ internal/feishu/client_test.go | 62 +++++++ internal/notify/alertstate.go | 70 ++++++++ internal/notify/alertstate_test.go | 107 +++++++++++ internal/notify/health.go | 118 +++++++++++++ internal/notify/health_linux.go | 19 ++ internal/notify/health_other.go | 9 + internal/notify/health_test.go | 198 +++++++++++++++++++++ internal/notify/service.go | 275 +++++++++++++++++++++++++++++ internal/notify/service_test.go | 40 +++++ internal/render/alert.go | 90 ++++++++++ internal/render/alert_test.go | 86 +++++++++ internal/render/card.go | 187 ++++++++++++++++++++ internal/render/card_test.go | 110 ++++++++++++ internal/render/supply.go | 68 +++++++ internal/render/supply_test.go | 49 +++++ internal/state/state.go | 64 +++++++ internal/state/state_test.go | 31 ++++ testdata/current-run.json | 64 +++++++ 33 files changed, 2676 insertions(+) create mode 100644 .gitattributes create mode 100644 .gitignore create mode 100644 Dockerfile create mode 100644 LICENSE create mode 100644 README.md create mode 100644 cmd/iotlcm-bot/main.go create mode 100644 config.yaml create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/archive/model.go create mode 100644 internal/archive/model_test.go create mode 100644 internal/archive/supply.go create mode 100644 internal/config/config.go create mode 100644 internal/config/config_test.go create mode 100644 internal/feishu/client.go create mode 100644 internal/feishu/client_test.go create mode 100644 internal/notify/alertstate.go create mode 100644 internal/notify/alertstate_test.go create mode 100644 internal/notify/health.go create mode 100644 internal/notify/health_linux.go create mode 100644 internal/notify/health_other.go create mode 100644 internal/notify/health_test.go create mode 100644 internal/notify/service.go create mode 100644 internal/notify/service_test.go create mode 100644 internal/render/alert.go create mode 100644 internal/render/alert_test.go create mode 100644 internal/render/card.go create mode 100644 internal/render/card_test.go create mode 100644 internal/render/supply.go create mode 100644 internal/render/supply_test.go create mode 100644 internal/state/state.go create mode 100644 internal/state/state_test.go create mode 100644 testdata/current-run.json diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..5837541 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,13 @@ +# 统一使用 LF，避免跨平台协作时产生整文件 diff +* text=auto eol=lf + +# 二进制文件不做行尾转换 +*.png binary +*.jpg binary +*.gif binary +*.ico binary +*.woff binary +*.woff2 binary +*.ttf binary +*.pdf binary +*.exe binary diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0167efe --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.env +.env.* +!.env.example +bin/ +*.exe +coverage.out +/state/ +*.log \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..822629b --- /dev/null +++ b/Dockerfile @@ -0,0 +1,22 @@ +FROM 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 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/LICENSE b/LICENSE new file mode 100644 index 0000000..261eeb9 --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md new file mode 100644 index 0000000..12da0b4 --- /dev/null +++ b/README.md @@ -0,0 +1,127 @@ +# 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 范围 + +- 只接受 `version=1` 且包含 `run` 和非空 `rounds` 的归档 JSON +- 不兼容旧版 `rounds=[]` 文件 +- 不读取 Redis、Worker HTTP 接口或日志 +- 不实现 Replay、事件回调、卡片按钮和错误聚合 + +## 快速开始 + +```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 +``` + +## 配置 + +配置文件为 `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 | + +```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 +go run ./cmd/iotlcm-bot send --config /path/to/iotlcm-bot.yaml --file /path/to/run.json --force +``` + +`--force` 用于确需重发时；正常运行不要使用。发送成功会写入 `state/sent/.json`，同一 Run 再次执行会跳过。 + +## 容器化部署 + +```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 +``` + +容器内路径约定： + +| 容器内路径 | 用途 | 挂载方式 | +| --- | --- | --- | +| `/data/full-sync` | Worker 归档目录 | **只读** | +| `/data/state` | 已发送标记 | 可写 | +| `/app/config/config.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) diff --git a/cmd/iotlcm-bot/main.go b/cmd/iotlcm-bot/main.go new file mode 100644 index 0000000..a25d85b --- /dev/null +++ b/cmd/iotlcm-bot/main.go @@ -0,0 +1,137 @@ +package main + +import ( + "context" + "flag" + "fmt" + "os" + "os/signal" + "syscall" + + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/config" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/feishu" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/notify" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/render" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/state" +) + +func main() { + if len(os.Args) < 2 { + usage() + } + switch os.Args[1] { + case "render": + renderCommand(os.Args[2:]) + case "send": + sendCommand(os.Args[2:]) + case "send-text": + sendTextCommand(os.Args[2:]) + case "watch": + watchCommand(os.Args[2:]) + default: + usage() + } +} + +func renderCommand(args []string) { + fs := flag.NewFlagSet("render", flag.ExitOnError) + file := fs.String("file", "", "path to a Worker Run JSON") + environment := fs.String("environment", "preview", "environment label") + jsonOutput := fs.Bool("json", false, "print Feishu interactive-card JSON") + fs.Parse(args) + if *file == "" { + fmt.Fprintln(os.Stderr, "render requires --file") + os.Exit(2) + } + a, err := archive.Load(*file) + if err != nil { + fail(err) + } + if *jsonOutput { + body, err := render.Card(a, *environment) + if err != nil { + fail(err) + } + fmt.Println(string(body)) + return + } + fmt.Println(render.Text(a, *environment)) +} + +func sendCommand(args []string) { + fs := flag.NewFlagSet("send", flag.ExitOnError) + file := fs.String("file", "", "path to a Worker Run JSON") + configPath := fs.String("config", "", "path to YAML config") + force := fs.Bool("force", false, "send even when the Run was already marked sent") + fs.Parse(args) + if *file == "" || *configPath == "" { + fmt.Fprintln(os.Stderr, "send requires --file and --config") + os.Exit(2) + } + cfg, err := config.Load(*configPath) + if err != nil { + fail(err) + } + if err := cfg.ValidateSend(); err != nil { + fail(err) + } + client := feishu.NewClient(cfg.Feishu.AppID, cfg.Feishu.AppSecret, cfg.Feishu.ChatID) + svc := notify.New(cfg.ArchiveDir, cfg.Environment, cfg.PollDuration, state.New(cfg.StateDir), client, notify.DefaultLogger()) + if err := svc.SendFile(context.Background(), *file, *force); err != nil { + fail(err) + } +} + +func sendTextCommand(args []string) { + fs := flag.NewFlagSet("send-text", flag.ExitOnError) + configPath := fs.String("config", "", "path to YAML config") + text := fs.String("text", "", "text to send") + fs.Parse(args) + if *configPath == "" || *text == "" { + fmt.Fprintln(os.Stderr, "send-text requires --config and --text") + os.Exit(2) + } + cfg, err := config.Load(*configPath) + if err != nil { + fail(err) + } + if err := cfg.ValidateSend(); err != nil { + fail(err) + } + client := feishu.NewClient(cfg.Feishu.AppID, cfg.Feishu.AppSecret, cfg.Feishu.ChatID) + if err := client.SendText(context.Background(), *text); err != nil { + fail(err) + } + fmt.Println("Feishu text message sent") +} + +func watchCommand(args []string) { + fs := flag.NewFlagSet("watch", flag.ExitOnError) + configPath := fs.String("config", "", "path to YAML config") + fs.Parse(args) + if *configPath == "" { + fmt.Fprintln(os.Stderr, "watch requires --config") + os.Exit(2) + } + cfg, err := config.Load(*configPath) + if err != nil { + fail(err) + } + if err := cfg.ValidateSend(); err != nil { + fail(err) + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + client := feishu.NewClient(cfg.Feishu.AppID, cfg.Feishu.AppSecret, cfg.Feishu.ChatID) + svc := notify.New(cfg.ArchiveDir, cfg.Environment, cfg.PollDuration, state.New(cfg.StateDir), client, notify.DefaultLogger()) + if err := svc.Watch(ctx); err != nil && err != context.Canceled { + fail(err) + } +} + +func usage() { + fmt.Fprintln(os.Stderr, "usage: iotlcm-bot [options]") + os.Exit(2) +} +func fail(err error) { fmt.Fprintln(os.Stderr, "iotlcm-bot:", err); os.Exit(1) } diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..0206d91 --- /dev/null +++ b/config.yaml @@ -0,0 +1,8 @@ +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/go.mod b/go.mod new file mode 100644 index 0000000..8235e28 --- /dev/null +++ b/go.mod @@ -0,0 +1,5 @@ +module git.5iot.com/iotlcm-public/iotlcm-bot + +go 1.23 + +require gopkg.in/yaml.v3 v3.0.1 diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..a62c313 --- /dev/null +++ b/go.sum @@ -0,0 +1,4 @@ +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/archive/model.go b/internal/archive/model.go new file mode 100644 index 0000000..a4f6020 --- /dev/null +++ b/internal/archive/model.go @@ -0,0 +1,159 @@ +package archive + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "time" +) + +type RunArchive struct { + Version int `json:"version"` + Run RunSummary `json:"run"` + Rounds []RoundStats `json:"rounds"` +} + +type RunSummary struct { + SupplyObservation json.RawMessage `json:"supplyObservation,omitempty"` + SyncType string `json:"syncType,omitempty"` + RunID string `json:"runId"` + Operation string `json:"operation"` + Status string `json:"status"` + SuccessCount int64 `json:"successCount"` + FailedCount int64 `json:"failedCount"` + CreatedAt string `json:"createdAt"` + CompletedAt string `json:"completedAt"` +} + +type RoundStats struct { + RunID string `json:"runId"` + Round int `json:"round"` + EnqueuedCount int64 `json:"enqueuedCount"` + ActiveCount int64 `json:"activeCount"` + NextRoundCount int64 `json:"nextRoundCount"` + SuccessCount int64 `json:"successCount"` + FailedCount int64 `json:"failedCount"` + UpstreamAttemptCount int64 `json:"upstreamAttemptCount"` + BatchCount int64 `json:"batchCount"` + StartedAt string `json:"startedAt"` + CompletedAt string `json:"completedAt"` + ErrorCounts map[string]int `json:"errorCounts"` +} + +func Load(path string) (RunArchive, error) { + body, err := os.ReadFile(path) + if err != nil { + return RunArchive{}, err + } + var a RunArchive + if err := json.Unmarshal(body, &a); err != nil { + return RunArchive{}, fmt.Errorf("parse %s: %w", path, err) + } + if err := a.Validate(); err != nil { + return RunArchive{}, fmt.Errorf("validate %s: %w", path, err) + } + if err := ValidateFilename(path, a.Run.RunID); err != nil { + return RunArchive{}, err + } + return a, nil +} + +func ValidateFilename(path, runID string) error { + want := "run-" + runID + ".json" + if filepath.Base(path) != want { + return fmt.Errorf("archive filename %q does not match RunID; expected %q", filepath.Base(path), want) + } + return nil +} + +func (a *RunArchive) Validate() error { + if a.Version != 1 { + return fmt.Errorf("unsupported version %d", a.Version) + } + if a.Run.RunID == "" || a.Run.Operation == "" { + return fmt.Errorf("runId and operation are required") + } + if a.Run.Status != "completed" && a.Run.Status != "completed_with_failures" { + return fmt.Errorf("run is not terminal: %s", a.Run.Status) + } + if len(a.Rounds) == 0 { + return fmt.Errorf("rounds must be non-empty") + } + copyRounds := append([]RoundStats(nil), a.Rounds...) + sort.Slice(copyRounds, func(i, j int) bool { return copyRounds[i].Round < copyRounds[j].Round }) + var success, failed int64 + for i, r := range copyRounds { + if r.Round != i+1 { + return fmt.Errorf("round sequence is not contiguous at %d", r.Round) + } + if r.RunID != "" && r.RunID != a.Run.RunID { + return fmt.Errorf("round %d runId does not match root runId", r.Round) + } + if r.EnqueuedCount < 0 || r.ActiveCount < 0 || r.NextRoundCount < 0 || r.SuccessCount < 0 || r.FailedCount < 0 || r.UpstreamAttemptCount < 0 || r.BatchCount < 0 { + return fmt.Errorf("round %d contains negative count", r.Round) + } + if r.EnqueuedCount != r.SuccessCount+r.FailedCount+r.NextRoundCount { + return fmt.Errorf("round %d count does not close", r.Round) + } + if _, err := parseTime(r.StartedAt); err != nil { + return fmt.Errorf("round %d invalid startedAt: %w", r.Round, err) + } + if _, err := parseTime(r.CompletedAt); err != nil { + return fmt.Errorf("round %d invalid completedAt: %w", r.Round, err) + } + success += r.SuccessCount + failed += r.FailedCount + } + if a.Run.SuccessCount != success || a.Run.FailedCount != failed { + return fmt.Errorf("run totals do not match rounds: run success=%d failed=%d, rounds success=%d failed=%d", a.Run.SuccessCount, a.Run.FailedCount, success, failed) + } + if _, err := parseTime(a.Run.CreatedAt); err != nil { + return fmt.Errorf("invalid createdAt: %w", err) + } + if _, err := parseTime(a.Run.CompletedAt); err != nil { + return fmt.Errorf("invalid completedAt: %w", err) + } + return nil +} + +func (a RunArchive) TotalCount() int64 { return a.Run.SuccessCount + a.Run.FailedCount } +func (a RunArchive) SuccessRate() float64 { + if a.TotalCount() == 0 { + return 0 + } + return float64(a.Run.SuccessCount) / float64(a.TotalCount()) * 100 +} +func (a RunArchive) Duration() time.Duration { + start, err1 := parseTime(a.Run.CreatedAt) + end, err2 := parseTime(a.Run.CompletedAt) + if err1 != nil || err2 != nil { + return 0 + } + return end.Sub(start) +} +func (r RoundStats) SuccessRate() float64 { + if r.EnqueuedCount == 0 { + return 0 + } + return float64(r.SuccessCount) / float64(r.EnqueuedCount) * 100 +} + +func (r RoundStats) Duration() time.Duration { + start, err1 := parseTime(r.StartedAt) + end, err2 := parseTime(r.CompletedAt) + if err1 != nil || err2 != nil { + return 0 + } + return end.Sub(start) +} + +func parseTime(value string) (time.Time, error) { + for _, layout := range []string{time.RFC3339Nano, "2006-01-02 15:04:05.999", "2006-01-02 15:04:05"} { + if t, err := time.Parse(layout, value); err == nil { + return t, nil + } + } + return time.Time{}, fmt.Errorf("unsupported time %q", value) +} diff --git a/internal/archive/model_test.go b/internal/archive/model_test.go new file mode 100644 index 0000000..08b3e53 --- /dev/null +++ b/internal/archive/model_test.go @@ -0,0 +1,48 @@ +package archive + +import "testing" + +func validArchive() RunArchive { + return RunArchive{Version: 1, Run: RunSummary{RunID: "run-1", Operation: "sim_status_queue_full_sync", Status: "completed_with_failures", SuccessCount: 175893, FailedCount: 10504, CreatedAt: "2026-09-22T11:42:33Z", CompletedAt: "2026-09-22T12:32:01Z"}, Rounds: []RoundStats{ + {RunID: "run-1", Round: 1, EnqueuedCount: 186397, ActiveCount: 0, NextRoundCount: 9023, SuccessCount: 175870, FailedCount: 1504, UpstreamAttemptCount: 0, BatchCount: 1872, StartedAt: "2026-09-22T11:42:34Z", CompletedAt: "2026-09-22T12:21:49Z"}, + {RunID: "run-1", Round: 2, EnqueuedCount: 9023, ActiveCount: 0, NextRoundCount: 9000, SuccessCount: 23, FailedCount: 0, UpstreamAttemptCount: 0, BatchCount: 92, StartedAt: "2026-09-22T12:21:51Z", CompletedAt: "2026-09-22T12:26:55Z"}, + {RunID: "run-1", Round: 3, EnqueuedCount: 9000, ActiveCount: 0, NextRoundCount: 0, SuccessCount: 0, FailedCount: 9000, UpstreamAttemptCount: 0, BatchCount: 90, StartedAt: "2026-09-22T12:26:56Z", CompletedAt: "2026-09-22T12:32:01Z"}, + }} +} + +func TestValidateThreeRounds(t *testing.T) { + a := validArchive() + if err := a.Validate(); err != nil { + t.Fatal(err) + } + if len(a.Rounds) != 3 || a.Run.FailedCount != 10504 { + t.Fatalf("unexpected archive: %+v", a) + } +} + +func TestValidateRejectsEmptyRounds(t *testing.T) { + a := RunArchive{Version: 1, Run: RunSummary{RunID: "run-1", Operation: "sim_status_queue_full_sync", Status: "completed"}} + if err := a.Validate(); err == nil { + t.Fatal("expected empty rounds to be rejected") + } +} + +func TestValidateRejectsUnclosedRound(t *testing.T) { + a := validArchive() + a.Rounds[0].NextRoundCount++ + if err := a.Validate(); err == nil { + t.Fatal("expected count closure error") + } +} + +func TestValidateFilenameRequiresExactRunID(t *testing.T) { + if err := ValidateFilename(`/archive/run-full-run-id.json`, "full-run-id"); err != nil { + t.Fatal(err) + } + if err := ValidateFilename(`/archive/run-other-id.json`, "full-run-id"); err == nil { + t.Fatal("expected mismatched filename to be rejected") + } + if err := ValidateFilename(`/archive/action-full-run-id.json`, "full-run-id"); err == nil { + t.Fatal("expected non-run filename to be rejected") + } +} diff --git a/internal/archive/supply.go b/internal/archive/supply.go new file mode 100644 index 0000000..c7cf5c1 --- /dev/null +++ b/internal/archive/supply.go @@ -0,0 +1,47 @@ +package archive + +import "encoding/json" + +type SupplyWait struct { + Count int64 `json:"count"` + TotalMS int64 `json:"totalMs"` + LongestMS int64 `json:"longestMs"` +} +type SupplyChannel struct { + Key string `json:"chanKey"` + Startup SupplyWait `json:"startupWait"` + Confirmed SupplyWait `json:"confirmedWait"` + Unconfirmed SupplyWait `json:"unconfirmedIdle"` + UnknownMS int64 `json:"unknownMs"` + Stages map[string]int64 `json:"producerStageMs"` +} +type SupplySummary struct { + UnknownMS int64 `json:"unknownChannelMs"` + UnconfirmedMS int64 `json:"unconfirmedChannelMs"` + ConsumerMS int64 `json:"consumerStartupChannelMs"` + Targets int `json:"targetChannels"` + Valid int `json:"validChannels"` + Confirmed int `json:"confirmedChannels"` + Unknown int `json:"unknownChannels"` + TotalMS int64 `json:"confirmedChannelMs"` + LongestMS int64 `json:"longestMs"` +} +type Supply struct { + SchemaVersion int `json:"schemaVersion"` + ResolutionMS int `json:"resolutionMs"` + Partial bool `json:"partial"` + Truncated bool `json:"truncated"` + Summary SupplySummary `json:"summary"` + Channels []SupplyChannel `json:"channels"` +} + +func (r RunSummary) Supply() (*Supply, bool) { + if len(r.SupplyObservation) == 0 || string(r.SupplyObservation) == "null" { + return nil, false + } + var s Supply + if json.Unmarshal(r.SupplyObservation, &s) != nil || s.SchemaVersion != 1 || s.ResolutionMS != 1000 || s.Summary.Targets < 0 || s.Summary.Valid < 0 || s.Summary.Valid > s.Summary.Targets || s.Summary.TotalMS < 0 || s.Summary.LongestMS < 0 { + return nil, true + } + return &s, true +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..7a97c63 --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,63 @@ +package config + +import ( + "fmt" + "os" + "time" + + "gopkg.in/yaml.v3" +) + +type Config struct { + Environment string `yaml:"environment"` + ArchiveDir string `yaml:"archiveDir"` + StateDir string `yaml:"stateDir"` + PollInterval string `yaml:"pollInterval"` + Feishu FeishuConfig `yaml:"feishu"` + + PollDuration time.Duration `yaml:"-"` +} + +type FeishuConfig struct { + AppID string `yaml:"appId"` + AppSecret string `yaml:"appSecret"` + ChatID string `yaml:"chatId"` +} + +func Load(path string) (Config, error) { + body, err := os.ReadFile(path) + if err != nil { + return Config{}, fmt.Errorf("read config %s: %w", path, err) + } + var cfg Config + if err := yaml.Unmarshal(body, &cfg); err != nil { + return Config{}, fmt.Errorf("parse config %s: %w", path, err) + } + if cfg.Environment == "" { + cfg.Environment = "preview" + } + if cfg.ArchiveDir == "" { + cfg.ArchiveDir = "/data/full-sync" + } + if cfg.StateDir == "" { + cfg.StateDir = "/data/state" + } + if cfg.PollInterval == "" { + cfg.PollInterval = "10s" + } + cfg.PollDuration, err = time.ParseDuration(cfg.PollInterval) + if err != nil || cfg.PollDuration <= 0 { + return Config{}, fmt.Errorf("invalid pollInterval %q", cfg.PollInterval) + } + return cfg, nil +} + +func (c Config) ValidateSend() error { + if c.Feishu.AppID == "" || c.Feishu.AppSecret == "" { + return fmt.Errorf("feishu.appId and feishu.appSecret are required") + } + if c.Feishu.ChatID == "" { + return fmt.Errorf("feishu.chatId is required") + } + return nil +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..ced3703 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,32 @@ +package config + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestLoadYAML(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yaml") + body := []byte("environment: preview\narchiveDir: /data/full-sync\nstateDir: /data/state\npollInterval: 5s\nfeishu:\n appId: app-id\n appSecret: secret\n chatId: chat-id\n") + if err := os.WriteFile(path, body, 0o600); err != nil { + t.Fatal(err) + } + cfg, err := Load(path) + if err != nil { + t.Fatal(err) + } + if cfg.PollDuration != 5*time.Second || cfg.Feishu.AppID != "app-id" || cfg.ArchiveDir != "/data/full-sync" { + t.Fatalf("unexpected config: %+v", cfg) + } + if err := cfg.ValidateSend(); err != nil { + t.Fatal(err) + } +} + +func TestValidateSendRejectsMissingCredentials(t *testing.T) { + if err := (Config{}).ValidateSend(); err == nil { + t.Fatal("expected validation error") + } +} diff --git a/internal/feishu/client.go b/internal/feishu/client.go new file mode 100644 index 0000000..531da90 --- /dev/null +++ b/internal/feishu/client.go @@ -0,0 +1,155 @@ +package feishu + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "sync" + "time" +) + +type Client struct { + appID string + appSecret string + chatID string + http *http.Client + baseURL string + mu sync.Mutex + token string + tokenExp time.Time +} + +func NewClient(appID, appSecret, chatID string) *Client { + return &Client{appID: appID, appSecret: appSecret, chatID: chatID, http: &http.Client{Timeout: 10 * time.Second}, baseURL: "https://open.feishu.cn"} +} + +func (c *Client) SendText(ctx context.Context, text string) error { + if c.chatID == "" { + return fmt.Errorf("feishu chatId is required") + } + if text == "" { + return fmt.Errorf("text is required") + } + token, err := c.accessToken(ctx, false) + if err != nil { + return err + } + if err := c.sendTextWithToken(ctx, token, text); err != nil && err.Error() == "feishu authentication failed" { + token, err = c.accessToken(ctx, true) + if err != nil { + return err + } + return c.sendTextWithToken(ctx, token, text) + } + return err +} + +func (c *Client) Send(ctx context.Context, card []byte) error { + if c.chatID == "" { + return fmt.Errorf("feishu chatId is required") + } + token, err := c.accessToken(ctx, false) + if err != nil { + return err + } + if err := c.sendWithToken(ctx, token, card); err != nil && err.Error() == "feishu authentication failed" { + token, err = c.accessToken(ctx, true) + if err != nil { + return err + } + return c.sendWithToken(ctx, token, card) + } else { + return err + } +} + +func (c *Client) accessToken(ctx context.Context, force bool) (string, error) { + c.mu.Lock() + defer c.mu.Unlock() + if !force && c.token != "" && time.Until(c.tokenExp) > time.Minute { + return c.token, nil + } + body, _ := json.Marshal(map[string]string{"app_id": c.appID, "app_secret": c.appSecret}) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/open-apis/auth/v3/tenant_access_token/internal", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json; charset=utf-8") + resp, err := c.http.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + TenantAccessToken string `json:"tenant_access_token"` + Expire int `json:"expire"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 || result.Code != 0 || result.TenantAccessToken == "" { + return "", fmt.Errorf("feishu token request failed: http=%d code=%d msg=%s", resp.StatusCode, result.Code, result.Msg) + } + c.token, c.tokenExp = result.TenantAccessToken, time.Now().Add(time.Duration(result.Expire)*time.Second) + return c.token, nil +} + +func (c *Client) sendTextWithToken(ctx context.Context, token, text string) error { + payload, _ := json.Marshal(map[string]string{"receive_id": c.chatID, "msg_type": "text", "content": fmt.Sprintf(`{"text":%q}`, text)}) + return c.postMessage(ctx, token, payload) +} + +func (c *Client) sendWithToken(ctx context.Context, token string, card []byte) error { + var content json.RawMessage + if err := json.Unmarshal(card, &content); err != nil { + return fmt.Errorf("invalid card: %w", err) + } + var outer map[string]any + if err := json.Unmarshal(content, &outer); err != nil { + return err + } + msgType, ok := outer["msg_type"].(string) + if !ok { + return fmt.Errorf("card msg_type is required") + } + cardBody, ok := outer["card"] + if !ok { + return fmt.Errorf("card body is required") + } + cardJSON, _ := json.Marshal(cardBody) + payload, _ := json.Marshal(map[string]any{"receive_id": c.chatID, "msg_type": msgType, "content": string(cardJSON)}) + return c.postMessage(ctx, token, payload) +} + +func (c *Client) postMessage(ctx context.Context, token string, payload []byte) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/open-apis/im/v1/messages?receive_id_type="+url.QueryEscape("chat_id"), bytes.NewReader(payload)) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json; charset=utf-8") + resp, err := c.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + } + _ = json.Unmarshal(data, &result) + if resp.StatusCode == http.StatusUnauthorized { + return fmt.Errorf("feishu authentication failed") + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 || result.Code != 0 { + return fmt.Errorf("feishu send failed: http=%d code=%d msg=%s", resp.StatusCode, result.Code, result.Msg) + } + return nil +} diff --git a/internal/feishu/client_test.go b/internal/feishu/client_test.go new file mode 100644 index 0000000..3fbf49d --- /dev/null +++ b/internal/feishu/client_test.go @@ -0,0 +1,62 @@ +package feishu + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestSendText(t *testing.T) { + var gotText string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/open-apis/auth/v3/tenant_access_token/internal": + _, _ = w.Write([]byte(`{"code":0,"tenant_access_token":"test-token","expire":7200}`)) + case "/open-apis/im/v1/messages": + if r.URL.Query().Get("receive_id_type") != "chat_id" || r.Header.Get("Authorization") != "Bearer test-token" { + t.Errorf("unexpected request: %s %s", r.URL.String(), r.Header.Get("Authorization")) + } + var payload struct { + ReceiveID string `json:"receive_id"` + MsgType string `json:"msg_type"` + Content string `json:"content"` + } + if err := json.NewDecoder(r.Body).Decode(&payload); err != nil { + t.Error(err) + } + if payload.ReceiveID != "test-chat" || payload.MsgType != "text" { + t.Errorf("unexpected payload: %+v", payload) + } + var content struct { + Text string `json:"text"` + } + if err := json.Unmarshal([]byte(payload.Content), &content); err != nil { + t.Error(err) + } + gotText = content.Text + _, _ = w.Write([]byte(`{"code":0,"msg":"success"}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client := NewClient("test-app", "test-secret", "test-chat") + client.baseURL = server.URL + message := "连接测试：含引号 \" 和换行\n第二行" + if err := client.SendText(context.Background(), message); err != nil { + t.Fatal(err) + } + if gotText != message { + t.Fatalf("text mismatch: %q", gotText) + } +} + +func TestSendTextRejectsEmpty(t *testing.T) { + client := NewClient("test-app", "test-secret", "test-chat") + if err := client.SendText(context.Background(), ""); err == nil || !strings.Contains(err.Error(), "text is required") { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/internal/notify/alertstate.go b/internal/notify/alertstate.go new file mode 100644 index 0000000..c35fc44 --- /dev/null +++ b/internal/notify/alertstate.go @@ -0,0 +1,70 @@ +package notify + +import ( + "encoding/json" + "os" + "path/filepath" + "time" +) + +// alertState 记录告警节流信息。 +// +// 为什么必须落盘：进程检测到挂载失效后会主动退出、由 restart 策略拉起。 +// 若节流时间只存在内存里，重启后 lastAlert 归零，会立刻再次告警 —— +// 持续故障就变成「每轮重启都发一条飞书」，把真正的告警淹成骚扰。 +type alertState struct { + // LastDegradedAlert 是上次发送「归档目录不可用」告警的时间。 + LastDegradedAlert time.Time `json:"lastDegradedAlert,omitempty"` + // LastDegradedReason 用于在故障原因变化时立即重新告警。 + LastDegradedReason string `json:"lastDegradedReason,omitempty"` + // ExitCount 统计因挂载失效而退出的累计次数，用于识别「反复重启」。 + ExitCount int `json:"exitCount,omitempty"` +} + +func alertStatePath(stateDir string) string { + return filepath.Join(stateDir, "alert-state.json") +} + +func loadAlertState(stateDir string) alertState { + var st alertState + if stateDir == "" { + return st + } + body, err := os.ReadFile(alertStatePath(stateDir)) + if err != nil { + return st + } + _ = json.Unmarshal(body, &st) + return st +} + +func saveAlertState(stateDir string, st alertState) error { + if stateDir == "" { + return nil + } + path := alertStatePath(stateDir) + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + body, err := json.MarshalIndent(st, "", " ") + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, body, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} + +// shouldAlert 判断是否该发告警：间隔到了，或故障原因发生变化。 +func (st alertState) shouldAlert(reason string, now time.Time, interval time.Duration) bool { + if st.LastDegradedAlert.IsZero() { + return true + } + if reason != st.LastDegradedReason { + // 原因变了（例如从「目录不存在」变成「不可读」）值得立刻提醒。 + return true + } + return now.Sub(st.LastDegradedAlert) >= interval +} diff --git a/internal/notify/alertstate_test.go b/internal/notify/alertstate_test.go new file mode 100644 index 0000000..0146fb7 --- /dev/null +++ b/internal/notify/alertstate_test.go @@ -0,0 +1,107 @@ +package notify + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" +) + +func TestAlertStateSurvivesRestart(t *testing.T) { + dir := t.TempDir() + now := time.Now() + + // 首次：没有任何记录 → 应该告警 + first := loadAlertState(dir) + if !first.shouldAlert("dir gone", now, DegradedAlertInterval) { + t.Fatal("first alert must be sent") + } + first.LastDegradedAlert = now + first.LastDegradedReason = "dir gone" + if err := saveAlertState(dir, first); err != nil { + t.Fatal(err) + } + + // 模拟进程退出后重启：重新从磁盘加载 + reloaded := loadAlertState(dir) + if reloaded.shouldAlert("dir gone", now.Add(time.Second), DegradedAlertInterval) { + t.Fatal("after restart the same failure must NOT alert again immediately (this is the alert-flood bug)") + } + // 间隔到了才允许再报 + if !reloaded.shouldAlert("dir gone", now.Add(DegradedAlertInterval+time.Second), DegradedAlertInterval) { + t.Fatal("alert must be sent again after the interval elapses") + } + // 故障原因变化时立即告警 + if !reloaded.shouldAlert("permission denied", now.Add(time.Second), DegradedAlertInterval) { + t.Fatal("a changed failure reason must alert immediately") + } +} + +func TestExitCountIsPersistedAndIncremented(t *testing.T) { + dir := t.TempDir() + st := loadAlertState(dir) + if st.ExitCount != 0 { + t.Fatalf("fresh state exitCount=%d want 0", st.ExitCount) + } + st.ExitCount++ + if err := saveAlertState(dir, st); err != nil { + t.Fatal(err) + } + // 重启后计数必须延续，否则无法识别「反复重启」 + if got := loadAlertState(dir).ExitCount; got != 1 { + t.Fatalf("exitCount after reload=%d want 1", got) + } +} + +// 关键回归：挂载失效 → 退出 → 重启后，同一次故障只能推一条飞书。 +func TestWatchAlertsOnceAcrossRestarts(t *testing.T) { + shortenExitDelay(t) + if !ExitOnDegraded { + t.Skip("ExitOnDegraded disabled") + } + stateDir := t.TempDir() + archiveDir := filepath.Join(t.TempDir(), "gone") + fake := &fakeFeishu{} + newSvc := func() Service { + return Service{ + archiveDir: archiveDir, + environment: "test", + stateDir: stateDir, + pollInterval: 10 * time.Millisecond, + store: newTestStore(stateDir), + client: fake, + logger: discardLogger(), + } + } + // 第一次运行：应告警并退出 + if err := newSvc().Watch(context.Background()); err == nil { + t.Fatal("Watch must exit when the mount is degraded") + } + if fake.sent != 1 { + t.Fatalf("first run must send exactly one alert, sent=%d", fake.sent) + } + // 模拟被 restart 策略拉起：新进程、新内存，但 state 目录相同 + if err := newSvc().Watch(context.Background()); err == nil { + t.Fatal("Watch must exit again") + } + if fake.sent != 1 { + t.Fatalf("restart must NOT resend the same alert (alert flood), sent=%d", fake.sent) + } + // 退出计数应累计到 2 + if got := loadAlertState(stateDir).ExitCount; got != 2 { + t.Fatalf("exitCount=%d want 2", got) + } +} + +func TestAlertStateMissingDirIsNotFatal(t *testing.T) { + // 状态目录不可写时不应影响告警逻辑（退化为每次都报，但至少不崩） + st := loadAlertState("") + if !st.shouldAlert("x", time.Now(), DegradedAlertInterval) { + t.Fatal("empty state must allow alerting") + } + if err := saveAlertState("", st); err != nil { + t.Fatalf("saving to empty state dir must be a no-op, got %v", err) + } + _ = os.Remove("nonexistent") +} diff --git a/internal/notify/health.go b/internal/notify/health.go new file mode 100644 index 0000000..c34d956 --- /dev/null +++ b/internal/notify/health.go @@ -0,0 +1,118 @@ +package notify + +import ( + "fmt" + "os" + "time" +) + +// Health 描述归档目录的可用性。 +// +// 背景：归档目录通过 bind mount 进容器后，宿主机一旦把它删除，容器里的路径 +// 会变成 mountinfo 中带 "//deleted" 的空壳目录（硬链接数 nlink=0），读起来 +// "存在但为空"，bot 会静默地再也发不出任何通知 —— 2026-10-07 预演环境实际 +// 发生过（宿主日志清理脚本删空归档目录，bot 从 10-07 静默到 10-12）。 +// 这里把该状态显式暴露出来，用于日志与飞书告警。 +type Health struct { + Mounted bool + Nlink uint64 + Entries int + Degraded bool + Reason string +} + +// StaleAfter 是「目录持续为空多久后开始高频复查」的阈值。 +// 归档只在全量同步结束后写入，中间的空窗是正常的，因此不能一空就缩短周期。 +const StaleAfter = 10 * time.Minute + +// DegradedAlertInterval 是同一个失效状态重复告警的最小间隔。 +const DegradedAlertInterval = 6 * time.Hour + +// DegradedExitDelay 是检测到挂载失效后，退出前的等待时间。 +// +// 为什么不立即退出：compose 的 restart 策略会立即拉起容器，若进程马上又检测到 +// 失效并退出，就变成几秒一次的崩溃循环（实测 8 次重启只花了约 9 秒），既刷日志 +// 又毫无意义。等一会再退，既给宿主一个修复窗口，也让重启频率可控。 +// 实测：目录真的被删掉时，等几十秒后再重启依然能自愈（重启会重新解析挂载）。 +// +// 声明为变量便于测试缩短（生产用默认值）。 +var DegradedExitDelay = 30 * time.Second + +// DegradedMaxExits 是同一进程内连续退出的上限，防止极端情况下无限退出。 +// 超过后不再退出，只保留告警与日志，避免变成无意义的重启风暴。 +const DegradedMaxExits = 3 + +// ExitOnDegraded 为 true 时，检测到挂载失效（目录被宿主删除，nlink=0） +// 就告警并退出进程，交给 restart policy 拉起。 +// +// 为什么退出就能修好：bind mount 的源路径是**容器启动时**解析的。宿主把目录删掉 +// 再重建后，容器仍指向旧 inode（mountinfo 里带 //deleted、nlink=0），看不到任何 +// 新文件；但重新启动容器会重新解析路径，拿到新 inode 并立即恢复正常。 +// 实测（2026-10-13，dev 100.29）：手工 docker restart 后 inode 从旧变新、文件立即可见； +// 进程自行退出后由 restart=unless-stopped 拉起，同样自愈（restarts=1）。 +// +// 这样就不再需要「重建 worker 必须同批重建 bot」这种人工纪律： +// worker 重建把目录换掉后，bot 自己会在下一轮扫描退出并被拉起，自动重新挂载。 +var ExitOnDegraded = true + +// Health 检查归档目录当前是否可用。 +func (s Service) Health() Health { + h := Health{Mounted: true} + info, err := os.Stat(s.archiveDir) + if err != nil { + h.Mounted = false + h.Degraded = true + h.Reason = fmt.Sprintf("archive directory is not readable: %v", err) + return h + } + if !info.IsDir() { + h.Mounted = false + h.Degraded = true + h.Reason = fmt.Sprintf("archive path is not a directory: %s", s.archiveDir) + return h + } + if nlink, ok := linkCount(info); ok { + h.Nlink = nlink + if nlink == 0 { + // 宿主机目录已被删除，容器仍挂着它的空壳。 + h.Mounted = false + h.Degraded = true + h.Reason = fmt.Sprintf("archive directory was deleted on the host (nlink=0); recreate this container to rebind %s", s.archiveDir) + return h + } + } + entries, err := os.ReadDir(s.archiveDir) + if err != nil { + h.Mounted = false + h.Degraded = true + h.Reason = fmt.Sprintf("archive directory is not listable: %v", err) + return h + } + h.Entries = len(entries) + if h.Entries == 0 { + h.Reason = "archive directory is empty" + } + return h +} + +// Interval 返回下一次扫描前的等待时长。 +// +// 正常情况下保持配置的轮询周期；只有在目录长期为空（可能正被写入，也可能已失效） +// 时才缩短到 10s，避免把通知延迟留给下一次发布。 +func (s Service) Interval(emptySince time.Time) time.Duration { + base := s.pollInterval + if base <= 0 { + base = 10 * time.Second + } + if base <= 10*time.Second { + return base + } + h := s.Health() + if h.Entries > 0 { + return base + } + if emptySince.IsZero() || time.Since(emptySince) < StaleAfter { + return base + } + return 10 * time.Second +} diff --git a/internal/notify/health_linux.go b/internal/notify/health_linux.go new file mode 100644 index 0000000..6483c7a --- /dev/null +++ b/internal/notify/health_linux.go @@ -0,0 +1,19 @@ +//go:build linux + +package notify + +import ( + "os" + "syscall" +) + +// linkCount 返回目录的硬链接数。 +// 宿主机把 bind mount 的源目录删掉后，容器内该目录的 nlink 会变成 0 +// （mountinfo 里也能看到 "//deleted"），这是唯一可靠的失效信号。 +func linkCount(info os.FileInfo) (uint64, bool) { + st, ok := info.Sys().(*syscall.Stat_t) + if !ok { + return 0, false + } + return uint64(st.Nlink), true +} diff --git a/internal/notify/health_other.go b/internal/notify/health_other.go new file mode 100644 index 0000000..51f7c0c --- /dev/null +++ b/internal/notify/health_other.go @@ -0,0 +1,9 @@ +//go:build !linux + +package notify + +import "os" + +// linkCount 在非 Linux 平台不可用（syscall.Stat_t 没有 Nlink 字段）。 +// bot 实际只跑在 Linux 容器里，这里让本机测试仍可编译运行。 +func linkCount(info os.FileInfo) (uint64, bool) { return 0, false } diff --git a/internal/notify/health_test.go b/internal/notify/health_test.go new file mode 100644 index 0000000..a3a5ecc --- /dev/null +++ b/internal/notify/health_test.go @@ -0,0 +1,198 @@ +package notify + +import ( + "context" + "errors" + "log/slog" + "os" + "path/filepath" + "testing" + "time" + + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/state" +) + +func writeArchive(t *testing.T, dir, name, body string) { + t.Helper() + if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0o600); err != nil { + t.Fatal(err) + } +} + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError + 1})) +} + +// countingLogger 统计 ERROR 级日志条数，用于验证去重告警。 +func countingLogger(count *int) *slog.Logger { + return slog.New(slog.NewTextHandler(countingWriter{count}, &slog.HandlerOptions{Level: slog.LevelError})) +} + +type countingWriter struct{ count *int } + +func (w countingWriter) Write(p []byte) (int, error) { + *w.count++ + return len(p), nil +} + +const validArchiveBody = `{"version":1,"run":{"runId":"z-good","operation":"op","status":"completed","successCount":1,"failedCount":0,"createdAt":"2026-10-12T00:00:00Z","completedAt":"2026-10-12T00:01:00Z"},"rounds":[{"round":1,"enqueuedCount":1,"activeCount":0,"nextRoundCount":0,"successCount":1,"failedCount":0,"startedAt":"2026-10-12T00:00:00Z","completedAt":"2026-10-12T00:01:00Z"}]}` + +// 一个坏归档不能挡住后面的归档，但也不能每轮扫描都刷一条日志。 +func TestScanSkipsBrokenArchiveAndLogsOnlyOnce(t *testing.T) { + root := t.TempDir() + // 字典序在前：旧格式（rounds 为空）会校验失败。 + writeArchive(t, root, "run-broken.json", `{"version":1,"run":{"runId":"broken","operation":"op","status":"completed"},"rounds":[]}`) + writeArchive(t, root, "run-z-good.json", validArchiveBody) + + fake := &fakeFeishu{} + var logged int + svc := Service{archiveDir: root, environment: "preview", store: state.New(t.TempDir()), client: fake, logger: countingLogger(&logged), warned: map[string]struct{}{}} + + for i := 0; i < 3; i++ { + if err := svc.Scan(context.Background()); err != nil { + t.Fatalf("Scan() returned error for a broken archive: %v", err) + } + } + if fake.sent != 1 { + t.Fatalf("expected the valid archive to be sent once, sent=%d", fake.sent) + } + if logged != 1 { + t.Fatalf("broken archive should be reported once across 3 scans, logged=%d", logged) + } +} + +// 文件被重写（内容变了）后应该重新告警。 +func TestScanWarnsAgainAfterArchiveIsRewritten(t *testing.T) { + root := t.TempDir() + writeArchive(t, root, "run-broken.json", `{"version":1,"run":{"runId":"broken","operation":"op","status":"completed"},"rounds":[]}`) + + var logged int + svc := Service{archiveDir: root, environment: "preview", store: state.New(t.TempDir()), client: &fakeFeishu{}, logger: countingLogger(&logged), warned: map[string]struct{}{}} + + if err := svc.Scan(context.Background()); err != nil { + t.Fatal(err) + } + // 重写为另一个仍然无效的内容，mtime/大小变化 + time.Sleep(10 * time.Millisecond) + writeArchive(t, root, "run-broken.json", `{"version":1,"run":{"runId":"broken","operation":"op","status":"completed","queryMonth":"202610"},"rounds":[]}`) + if err := svc.Scan(context.Background()); err != nil { + t.Fatal(err) + } + if logged != 2 { + t.Fatalf("rewritten broken archive should be reported again, logged=%d", logged) + } +} + +// 发送失败（飞书临时故障）必须每轮都报，不能被去重掉： +// 否则一次网络抖动就会让通知静默丢失。 +func TestScanSendFailureIsNotDeduped(t *testing.T) { + root := t.TempDir() + writeArchive(t, root, "run-z-good.json", validArchiveBody) + + fail := &fakeFeishu{err: errors.New("feishu timeout")} + var logged int + svc := Service{archiveDir: root, environment: "preview", store: state.New(t.TempDir()), client: fail, logger: countingLogger(&logged), warned: map[string]struct{}{}} + + for i := 0; i < 3; i++ { + if err := svc.Scan(context.Background()); err != nil { + t.Fatal(err) + } + } + if logged != 3 { + t.Fatalf("transient send failures must be reported every scan, logged=%d want=3", logged) + } +} + +func TestHealthReportsEmptyDirectory(t *testing.T) { + svc := Service{archiveDir: t.TempDir()} + h := svc.Health() + if !h.Mounted || h.Degraded { + t.Fatalf("empty but mounted directory should not be degraded: %+v", h) + } + if h.Entries != 0 || h.Reason == "" { + t.Fatalf("unexpected health: %+v", h) + } +} + +func TestHealthReportsMissingDirectory(t *testing.T) { + svc := Service{archiveDir: filepath.Join(t.TempDir(), "gone")} + h := svc.Health() + if h.Mounted || !h.Degraded { + t.Fatalf("missing directory must be degraded: %+v", h) + } +} + +// 挂载失效时必须退出进程，交给 restart policy 重新拉起以重新解析 bind mount。 +// 这是「不再需要人工同批重建 bot」的关键：重启后挂载指向宿主当前目录。 +func TestWatchExitsWhenArchiveMountIsDegraded(t *testing.T) { + shortenExitDelay(t) + if !ExitOnDegraded { + t.Skip("ExitOnDegraded disabled") + } + stateDir := t.TempDir() + svc := Service{ + archiveDir: filepath.Join(t.TempDir(), "gone"), + environment: "test", + stateDir: stateDir, + pollInterval: 10 * time.Millisecond, + store: state.New(stateDir), + logger: discardLogger(), + } + done := make(chan error, 1) + go func() { done <- svc.Watch(context.Background()) }() + select { + case err := <-done: + if err == nil { + t.Fatal("Watch must return an error so the process exits non-zero and restarts") + } + case <-time.After(3 * time.Second): + t.Fatal("Watch did not exit after detecting a degraded archive mount") + } +} + +// 目录长期为空时才缩短轮询；有内容或空窗很短时保持配置周期。 +func TestIntervalShortensOnlyAfterLongEmptyPeriod(t *testing.T) { + root := t.TempDir() + svc := Service{archiveDir: root, pollInterval: time.Minute} + if got := svc.Interval(time.Time{}); got != time.Minute { + t.Fatalf("fresh empty dir should keep poll interval, got %s", got) + } + if got := svc.Interval(time.Now().Add(-StaleAfter - time.Second)); got != 10*time.Second { + t.Fatalf("long-empty dir should poll every 10s, got %s", got) + } + writeArchive(t, root, "run-anything.json", "{}") + if got := svc.Interval(time.Now().Add(-time.Hour)); got != time.Minute { + t.Fatalf("non-empty dir should keep poll interval, got %s", got) + } +} + +type fakeFeishu struct { + sent int + err error + textCount int + texts []string +} + +func (f *fakeFeishu) Send(ctx context.Context, card []byte) error { + if f.err != nil { + return f.err + } + f.sent++ + return nil +} +func (f *fakeFeishu) SendText(ctx context.Context, text string) error { + f.textCount++ + f.texts = append(f.texts, text) + return nil +} + +// newTestStore 构造一个 state.Store 并返回其 Dir，供 alertstate 测试使用。 +func newTestStore(dir string) state.Store { return state.New(dir) } + +// shortenExitDelay 把退出等待改短，避免测试白等 30 秒（仅测试内使用）。 +func shortenExitDelay(t *testing.T) { + t.Helper() + original := DegradedExitDelay + DegradedExitDelay = 10 * time.Millisecond + t.Cleanup(func() { DegradedExitDelay = original }) +} diff --git a/internal/notify/service.go b/internal/notify/service.go new file mode 100644 index 0000000..d39712a --- /dev/null +++ b/internal/notify/service.go @@ -0,0 +1,275 @@ +package notify + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "sort" + "time" + + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/render" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/state" +) + +type Service struct { + archiveDir, environment string + stateDir string + pollInterval time.Duration + store state.Store + client Sender + logger *slog.Logger + // warned 记录已经报过错的归档，避免坏文件每轮扫描都刷一条日志。 + // 键为「路径 + mtime + 大小」：文件一旦被重写就会重新告警。 + warned map[string]struct{} +} + +// Sender 是发送端抽象：*feishu.Client 满足它，测试可注入替身。 +type Sender interface { + Send(ctx context.Context, card []byte) error + SendText(ctx context.Context, text string) error +} + +func New(archiveDir, environment string, pollInterval time.Duration, store state.Store, client Sender, logger *slog.Logger) Service { + 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{}{}} +} +func (s Service) Render(path string) (string, error) { + a, err := archive.Load(path) + if err != nil { + return "", err + } + return render.Text(a, s.environment), nil +} + +func (s Service) SendFile(ctx context.Context, path string, force bool) error { + if err := s.store.Prepare(); err != nil { + return err + } + return s.process(ctx, path, force) +} + +func (s Service) runFiles() ([]string, error) { + files, err := filepath.Glob(filepath.Join(s.archiveDir, "run-*.json")) + if err != nil { + return nil, err + } + sort.Strings(files) + return files, nil +} + +// skipKey 生成归档文件的「内容版本」标识，用于去重告警。 +// 读不到元数据时退化为路径本身，宁可少报警也不要刷屏。 +func skipKey(path string) string { + info, err := os.Stat(path) + if err != nil { + return path + } + return fmt.Sprintf("%s|%d|%d", path, info.ModTime().UnixNano(), info.Size()) +} + +// archiveError 区分「这个归档永远不会成功」与「这次发送碰上了临时故障」。 +// 前者（格式不对、旧版本、rounds 为空）是常态，只该报一次； +// 后者（飞书超时、鉴权失败、写状态失败）必须每轮重试并持续可见， +// 否则一次网络抖动就会被去重逻辑吞掉，通知静默丢失。 +type archiveError struct { + err error + permanent bool +} + +func (e *archiveError) Error() string { return e.err.Error() } +func (e *archiveError) Unwrap() error { return e.err } + +// permanentErr 标记「重试也不会好」的错误。 +func permanentErr(err error) error { return &archiveError{err: err, permanent: true} } + +// isPermanent 判断错误是否属于「归档本身无效」。 +func isPermanent(err error) bool { + var ae *archiveError + return errors.As(err, &ae) && ae.permanent +} + +// warnOnce 对同一个归档文件只记一次「跳过」日志。 +// 只有永久性错误才去重；临时错误原样返回给调用方，保持每轮可见。 +func (s Service) warnOnce(path string, err error) { + if !isPermanent(err) { + s.logger.Error("process archive failed", "path", path, "error", err) + return + } + key := skipKey(path) + if s.warned != nil { + if _, seen := s.warned[key]; seen { + return + } + s.warned[key] = struct{}{} + } + s.logger.Error("skip archive", "path", path, "error", err) +} + +// Scan 扫描归档目录并发送尚未通知过的 Run。 +// +// 单个归档出错（旧格式、写了一半、权限不足）只跳过并记日志： +// 过去这里直接 return，一个坏文件会挡住后面所有归档，通知会静默丢失。 +// 坏文件是常态（例如 manual 部分同步的 Run 天然没有 rounds），所以只记一次。 +func (s Service) Scan(ctx context.Context) error { + if err := s.store.Prepare(); err != nil { + return err + } + files, err := s.runFiles() + if err != nil { + return err + } + for _, path := range files { + if err := s.process(ctx, path, false); err != nil { + s.warnOnce(path, err) + } + } + return nil +} +func (s Service) Watch(ctx context.Context) error { + if s.pollInterval <= 0 { + s.pollInterval = 10 * time.Second + } + var emptySince time.Time + for { + h := s.Health() + if h.Degraded { + // 目录已被宿主删除等硬故障：告警节流状态落盘，避免退出重启后 + // 把同一次故障反复推送成飞书骚扰。 + now := time.Now() + st := loadAlertState(s.stateDir) + if st.shouldAlert(h.Reason, now, DegradedAlertInterval) { + s.logger.Error("archive directory is not usable", "path", s.archiveDir, "reason", h.Reason, "entries", h.Entries) + s.alertMount(ctx, h.Reason, st.ExitCount+1) + st.LastDegradedAlert = now + st.LastDegradedReason = h.Reason + if err := saveAlertState(s.stateDir, st); err != nil { + s.logger.Error("save alert state failed", "error", err) + } + } + if ExitOnDegraded { + // 退出并交给 restart policy 重新拉起：重启会重新解析 bind mount， + // 从而重新挂载到宿主当前的目录（详见 ExitOnDegraded 注释）。 + st.ExitCount++ + if err := saveAlertState(s.stateDir, st); err != nil { + s.logger.Error("save alert state failed", "error", err) + } + if st.ExitCount > DegradedMaxExits { + // 已经反复重启仍未恢复（例如宿主把目录删了且不再重建）： + // 继续退出只会变成重启风暴，改为留在原地持续观测+告警。 + s.logger.Error("archive mount still unusable after repeated restarts, staying up to keep reporting", + "path", s.archiveDir, "reason", h.Reason, "exitCount", st.ExitCount) + } else { + // 等一会再退：restart 策略是立即拉起的，立刻退出会造成几秒一次的崩溃循环。 + s.logger.Error("exit to rebind archive mount", "path", s.archiveDir, "reason", h.Reason, "exitCount", st.ExitCount, "delay", DegradedExitDelay) + timer := time.NewTimer(DegradedExitDelay) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + return fmt.Errorf("archive directory unusable, restarting to rebind mount: %s", h.Reason) + } + } + } else if h.Entries == 0 { + if emptySince.IsZero() { + emptySince = time.Now() + } + } else { + emptySince = time.Time{} + } + + if err := s.Scan(ctx); err != nil { + s.logger.Error("scan failed", "error", err) + } + + timer := time.NewTimer(s.Interval(emptySince)) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-timer.C: + } + } +} + +// alert 在飞书客户端可用时发一条文本告警；失败只记日志，不影响主循环。 +func (s Service) alert(ctx context.Context, title, reason string) { + if s.client == nil { + return + } + text := fmt.Sprintf("[iotlcm-bot][%s] %s: %s", s.environment, title, reason) + if err := s.client.SendText(ctx, text); err != nil { + s.logger.Error("send alert failed", "error", err) + } +} + +// alertMount 发送「归档目录不可用」告警卡片。 +// +// 用卡片而非纯文本：text 消息在飞书里是灰底等宽块，长错误串会把气泡撑得很宽； +// 卡片带红色 header 与分区字段，一眼能看出是故障，且与同步结果卡片风格一致。 +func (s Service) alertMount(ctx context.Context, reason string, exitCount int) { + if s.client == nil { + return + } + summary := "正在重启以重新挂载" + if exitCount > DegradedMaxExits { + summary = "反复重启仍未恢复，已改为持续监控" + } + card, err := render.AlertCard(render.Alert{ + Environment: s.environment, + Title: "归档目录不可用", + Summary: summary, + Reason: render.FriendlyReason(reason), + ExitCount: exitCount, + Path: s.archiveDir, + }) + if err != nil { + s.logger.Error("render alert card failed", "error", err) + return + } + if err := s.client.Send(ctx, card); err != nil { + s.logger.Error("send alert failed", "error", err) + } +} +func (s Service) process(ctx context.Context, path string, force bool) error { + a, err := archive.Load(path) + if err != nil { + // 归档本身无效（格式/版本/rounds 为空）：重试也不会好。 + return permanentErr(err) + } + if !force { + sent, err := s.store.Has(a.Run.RunID) + if err != nil { + return err + } + if sent { + return nil + } + } + card, err := render.Card(a, s.environment) + if err != nil { + return permanentErr(err) + } + if s.client == nil { + return fmt.Errorf("feishu client is not configured") + } + if err := s.client.Send(ctx, card); err != nil { + return err + } + if err := s.store.Mark(a.Run.RunID); err != nil { + return err + } + s.logger.Info("run notification sent", "runId", a.Run.RunID, "path", path) + return nil +} + +func DefaultLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo})) +} diff --git a/internal/notify/service_test.go b/internal/notify/service_test.go new file mode 100644 index 0000000..0ad791e --- /dev/null +++ b/internal/notify/service_test.go @@ -0,0 +1,40 @@ +package notify + +import ( + "os" + "path/filepath" + "reflect" + "testing" +) + +func TestRunFilesSelectsOnlyTopLevelRunJSON(t *testing.T) { + root := t.TempDir() + for _, name := range []string{ + "run-abc.json", + "run-full-id.json", + "action-abc.json", + "sim-status-legacy.json", + "run-temp.json.tmp", + } { + if err := os.WriteFile(filepath.Join(root, name), []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + } + nested := filepath.Join(root, "2026", "09") + if err := os.MkdirAll(nested, 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(nested, "run-nested.json"), []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + + svc := Service{archiveDir: root} + got, err := svc.runFiles() + if err != nil { + t.Fatal(err) + } + want := []string{filepath.Join(root, "run-abc.json"), filepath.Join(root, "run-full-id.json")} + if !reflect.DeepEqual(got, want) { + t.Fatalf("runFiles() = %#v, want %#v", got, want) + } +} diff --git a/internal/render/alert.go b/internal/render/alert.go new file mode 100644 index 0000000..eaa6d81 --- /dev/null +++ b/internal/render/alert.go @@ -0,0 +1,90 @@ +package render + +import ( + "encoding/json" + "fmt" + "strings" +) + +// Alert 描述一条运维告警。 +type Alert struct { + Environment string + // Title 是卡片标题，如「归档目录不可用」。 + Title string + // Summary 是一句人话结论，如「正在重启以重新挂载」。 + Summary string + // Reason 是技术细节（英文错误原样保留，便于排查）。 + Reason string + // ExitCount 为累计退出次数，>1 时提示可能反复出现。 + ExitCount int + // Path 是涉及的路径。 + Path string +} + +// AlertCard 把告警渲染成飞书交互卡片。 +// +// 为什么不用纯文本：text 消息在飞书里渲染成灰底等宽块，长错误串会把气泡撑得很宽， +// 而且和同步结果卡片（interactive）风格不一致。卡片可以带红色 header、分区字段， +// 一眼能看出是「出事了」而不是一条普通消息。 +func AlertCard(a Alert) ([]byte, error) { + title := fmt.Sprintf("%s · %s", displayEnvironment(a.Environment), a.Title) + + fields := []any{} + if a.Summary != "" { + fields = append(fields, map[string]any{ + "tag": "div", + "text": map[string]any{"tag": "lark_md", "content": "**" + a.Summary + "**"}, + }) + } + if a.Path != "" { + fields = append(fields, map[string]any{ + "tag": "div", + "text": map[string]any{"tag": "lark_md", "content": "**路径**\n`" + a.Path + "`"}, + }) + } + if a.Reason != "" { + fields = append(fields, map[string]any{ + "tag": "div", + "text": map[string]any{"tag": "lark_md", "content": "**原因**\n" + a.Reason}, + }) + } + if a.ExitCount > 1 { + fields = append(fields, map[string]any{ + "tag": "div", + "text": map[string]any{"tag": "lark_md", "content": fmt.Sprintf("**已重复 %d 次**，若持续出现请检查宿主机上该目录是否被反复删除。", a.ExitCount)}, + }) + } + fields = append(fields, map[string]any{"tag": "hr"}) + fields = append(fields, map[string]any{ + "tag": "note", + "elements": []any{ + map[string]any{"tag": "plain_text", "content": "iotlcm-bot 自动检测 · 恢复后不再重复提醒"}, + }, + }) + + card := map[string]any{ + "msg_type": "interactive", + "card": map[string]any{ + // red 让告警在消息流里一眼可辨，与结果卡片的 blue 区分开。 + "header": map[string]any{"template": "red", "title": map[string]any{"tag": "plain_text", "content": title}}, + "elements": fields, + }, + } + return json.Marshal(card) +} + +// FriendlyReason 把常见的技术错误翻成人话，保留原文以便排查。 +func FriendlyReason(reason string) string { + switch { + case strings.Contains(reason, "deleted on the host"): + return "宿主机上的归档目录已被删除，容器仍指向旧目录（挂载失效）。" + case strings.Contains(reason, "no such file or directory"): + return "目录不存在。" + case strings.Contains(reason, "permission denied"): + return "没有读取权限。" + case strings.Contains(reason, "not a directory"): + return "该路径不是目录。" + default: + return reason + } +} diff --git a/internal/render/alert_test.go b/internal/render/alert_test.go new file mode 100644 index 0000000..d602ece --- /dev/null +++ b/internal/render/alert_test.go @@ -0,0 +1,86 @@ +package render + +import ( + "encoding/json" + "strings" + "testing" +) + +func TestAlertCardIsInteractiveWithRedHeader(t *testing.T) { + body, err := AlertCard(Alert{ + Environment: "preview", + Title: "归档目录不可用", + Summary: "正在重启以重新挂载", + Reason: "宿主机上的归档目录已被删除，容器仍指向旧目录（挂载失效）。", + Path: "/data/full-sync", + }) + if err != nil { + t.Fatal(err) + } + var decoded map[string]any + if err := json.Unmarshal(body, &decoded); err != nil { + t.Fatalf("alert card must be valid JSON: %v", err) + } + if decoded["msg_type"] != "interactive" { + t.Fatalf("alert must be an interactive card, got %v", decoded["msg_type"]) + } + card := decoded["card"].(map[string]any) + header := card["header"].(map[string]any) + if header["template"] != "red" { + t.Fatalf("alert header should be red so it stands out, got %v", header["template"]) + } + title := header["title"].(map[string]any)["content"].(string) + if !strings.Contains(title, "预演") || !strings.Contains(title, "归档目录不可用") { + t.Fatalf("unexpected title: %q", title) + } +} + +func TestAlertCardShowsPathAndRepeatCount(t *testing.T) { + body, err := AlertCard(Alert{ + Environment: "dev", + Title: "归档目录不可用", + Summary: "正在重启以重新挂载", + Reason: "目录不存在。", + Path: "/data/full-sync", + ExitCount: 3, + }) + if err != nil { + t.Fatal(err) + } + text := string(body) + if !strings.Contains(text, "/data/full-sync") { + t.Fatal("card should show the path") + } + if !strings.Contains(text, "已重复 3 次") { + t.Fatal("card should warn about repeated exits") + } +} + +// 单次故障时不该出现「已重复」这种话，避免误导。 +func TestAlertCardOmitsRepeatNoticeOnFirstOccurrence(t *testing.T) { + body, err := AlertCard(Alert{Environment: "dev", Title: "归档目录不可用", ExitCount: 1}) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(body), "已重复") { + t.Fatal("first occurrence must not claim repeats") + } +} + +func TestFriendlyReasonTranslatesKnownErrors(t *testing.T) { + cases := map[string]string{ + "archive directory was deleted on the host (nlink=0); recreate this container": "宿主机上的归档目录已被删除", + "stat /data/x: no such file or directory": "目录不存在。", + "open /data/x: permission denied": "没有读取权限。", + } + for input, want := range cases { + if got := FriendlyReason(input); !strings.Contains(got, want) { + t.Fatalf("FriendlyReason(%q) = %q, want it to contain %q", input, got, want) + } + } + // 未知错误原样保留，便于排查 + unknown := "some unexpected failure" + if got := FriendlyReason(unknown); got != unknown { + t.Fatalf("unknown reason must be preserved, got %q", got) + } +} diff --git a/internal/render/card.go b/internal/render/card.go new file mode 100644 index 0000000..d8cc9e9 --- /dev/null +++ b/internal/render/card.go @@ -0,0 +1,187 @@ +package render + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive" +) + +// Kind 返回卡片标题用的中文名称。流量同步有日、月历史、本月快照三条链路， +// 标题必须能区分，否则运营无法从卡片判断是哪一类同步。 +func Kind(operation string) string { + switch operation { + case "previous_day_usage_full_sync": + return "日流量同步" + case "previous_month_usage_full_sync": + return "月历史同步" + case "current_month_usage_full_sync": + return "本月流量快照同步" + } + if strings.Contains(operation, "sim_status") { + return "卡状态全量同步" + } + if isTrafficUsage(operation) { + return "流量全量同步" + } + return operation +} + +// isTrafficUsage 判定 operation 是否属于流量族（日、月历史、本月快照及兼容旧操作名）。 +// 供料观测等按链路区分的行为用它判断，避免依赖 Kind 返回的中文标题。 +func isTrafficUsage(operation string) bool { + return strings.Contains(operation, "traffic") || strings.Contains(operation, "usage") +} + +func Text(a archive.RunArchive, environment string) string { + title := fmt.Sprintf("【%s｜%s】", environment, Kind(a.Run.Operation)) + startAt, endAt := executionTimes(a) + lines := []string{ + title, + "", + fmt.Sprintf("总数：%s", number(a.TotalCount())), + fmt.Sprintf("成功：%s", number(a.Run.SuccessCount)), + fmt.Sprintf("失败：%s", number(a.Run.FailedCount)), + fmt.Sprintf("成功率：%.2f%%", a.SuccessRate()), + fmt.Sprintf("开始时间：%s", timestamp(startAt)), + fmt.Sprintf("结束时间：%s", timestamp(endAt)), + fmt.Sprintf("总耗时：%s", duration(a.Duration())), + fmt.Sprintf("实际执行：%d 轮", len(a.Rounds)), + "", + "轮次结果：", + } + for _, r := range a.Rounds { + lines = append(lines, roundLine(r)) + } + if block := supplyBlock(a); block != "" { + lines = append(lines, "", block) + } + lines = append(lines, "", "Run："+a.Run.RunID) + return strings.Join(lines, "\n") +} + +func Card(a archive.RunArchive, environment string) ([]byte, error) { + title := fmt.Sprintf("%s · %s", displayEnvironment(environment), Kind(a.Run.Operation)) + startAt, endAt := executionTimes(a) + // Full sync runs commonly have final failures; use one neutral header color. + template := "blue" + elements := []any{ + fields( + field("总数", number(a.TotalCount())), + field("成功", number(a.Run.SuccessCount)), + field("失败", number(a.Run.FailedCount)), + field("成功率", fmt.Sprintf("%.2f%%", a.SuccessRate())), + field("开始时间", timestamp(startAt)), + field("结束时间", timestamp(endAt)), + field("总耗时", duration(a.Duration())), + field("实际轮次", fmt.Sprintf("%d轮", len(a.Rounds))), + ), + map[string]any{"tag": "hr"}, + } + for _, r := range a.Rounds { + elements = append(elements, map[string]any{ + "tag": "div", + "text": map[string]any{"tag": "lark_md", "content": roundBlock(r)}, + }) + } + if block := supplyBlock(a); block != "" { + elements = append(elements, map[string]any{"tag": "div", "text": map[string]any{"tag": "lark_md", "content": block}}) + } + elements = append(elements, + map[string]any{"tag": "hr"}, + map[string]any{"tag": "note", "elements": []any{map[string]any{"tag": "plain_text", "content": "Run: " + a.Run.RunID}}}, + ) + card := map[string]any{ + "msg_type": "interactive", + "card": map[string]any{ + "header": map[string]any{"template": template, "title": map[string]any{"tag": "plain_text", "content": title}}, + "elements": elements, + }, + } + return json.Marshal(card) +} + +func fields(values ...map[string]any) map[string]any { + entries := make([]any, len(values)) + for i, value := range values { + entries[i] = value + } + return map[string]any{"tag": "div", "fields": entries} +} + +func field(label, value string) map[string]any { + return map[string]any{ + "is_short": true, + "text": map[string]any{"tag": "lark_md", "content": fmt.Sprintf("**%s**\n%s", label, value)}, + } +} + +func roundBlock(r archive.RoundStats) string { + return fmt.Sprintf("**第 %d 轮 · %s 张 · %s**\n成功 %s（%.2f%%）\n进入下轮 %s · 终态失败 %s", r.Round, number(r.EnqueuedCount), duration(r.Duration()), number(r.SuccessCount), r.SuccessRate(), number(r.NextRoundCount), number(r.FailedCount)) +} + +func displayEnvironment(environment string) string { + switch environment { + case "preview": + return "预演" + case "prod", "production": + return "生产" + default: + return environment + } +} + +func roundLine(r archive.RoundStats) string { + return fmt.Sprintf("第 %d 轮｜%s 张｜成功 %s（%.2f%%）｜进入下轮 %s｜终态失败 %s｜%s", r.Round, number(r.EnqueuedCount), number(r.SuccessCount), r.SuccessRate(), number(r.NextRoundCount), number(r.FailedCount), duration(r.Duration())) +} + +func executionTimes(a archive.RunArchive) (string, string) { + if len(a.Rounds) == 0 { + return a.Run.CreatedAt, a.Run.CompletedAt + } + return a.Rounds[0].StartedAt, a.Rounds[len(a.Rounds)-1].CompletedAt +} + +func timestamp(value string) string { + t, err := time.Parse(time.RFC3339Nano, value) + if err != nil { + return value + } + return t.In(time.FixedZone("CST", 8*60*60)).Format("2006-01-02 15:04:05") +} + +func number(n int64) string { + negative := n < 0 + if negative { + n = -n + } + s := fmt.Sprintf("%d", n) + for i := len(s) - 3; i > 0; i -= 3 { + s = s[:i] + "," + s[i:] + } + if negative { + return "-" + s + } + return s +} + +func duration(d time.Duration) string { + seconds := int64(d / time.Second) + if seconds < 0 { + return "未记录" + } + h, m, s := seconds/3600, (seconds%3600)/60, seconds%60 + parts := make([]string, 0, 3) + if h > 0 { + parts = append(parts, fmt.Sprintf("%d小时", h)) + } + if m > 0 { + parts = append(parts, fmt.Sprintf("%d分", m)) + } + if s > 0 || len(parts) == 0 { + parts = append(parts, fmt.Sprintf("%d秒", s)) + } + return strings.Join(parts, "") +} diff --git a/internal/render/card_test.go b/internal/render/card_test.go new file mode 100644 index 0000000..4f1189e --- /dev/null +++ b/internal/render/card_test.go @@ -0,0 +1,110 @@ +package render + +import ( + "strings" + "testing" + + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive" +) + +func sample() archive.RunArchive { + return archive.RunArchive{Version: 1, Run: archive.RunSummary{RunID: "run-1", Operation: "sim_status_queue_full_sync", Status: "completed_with_failures", SuccessCount: 175893, FailedCount: 10504, CreatedAt: "2026-09-22T11:42:33Z", CompletedAt: "2026-09-22T12:32:01Z"}, Rounds: []archive.RoundStats{ + {Round: 1, EnqueuedCount: 186397, NextRoundCount: 9023, SuccessCount: 175870, FailedCount: 1504, StartedAt: "2026-09-22T11:42:34Z", CompletedAt: "2026-09-22T12:21:49Z"}, + {Round: 2, EnqueuedCount: 9023, NextRoundCount: 9000, SuccessCount: 23, FailedCount: 0, StartedAt: "2026-09-22T12:21:51Z", CompletedAt: "2026-09-22T12:26:55Z"}, + {Round: 3, EnqueuedCount: 9000, NextRoundCount: 0, SuccessCount: 0, FailedCount: 9000, StartedAt: "2026-09-22T12:26:56Z", CompletedAt: "2026-09-22T12:32:01Z"}, + }} +} + +func TestKindNamesEachSyncType(t *testing.T) { + cases := []struct{ operation, want string }{ + {"previous_day_usage_full_sync", "日流量同步"}, + {"previous_month_usage_full_sync", "月历史同步"}, + {"current_month_usage_full_sync", "本月流量快照同步"}, + {"sim_status_queue_full_sync", "卡状态全量同步"}, + {"traffic_usage_full_sync", "流量全量同步"}, + {"usage_full_sync", "流量全量同步"}, + {"imei_full_sync", "imei_full_sync"}, + } + for _, c := range cases { + if got := Kind(c.operation); got != c.want { + t.Errorf("Kind(%q) = %q, want %q", c.operation, got, c.want) + } + } +} + +func TestKindKeepsTrafficSyncTitlesDistinct(t *testing.T) { + seen := map[string]string{} + for _, operation := range []string{"previous_day_usage_full_sync", "previous_month_usage_full_sync", "current_month_usage_full_sync"} { + title := Kind(operation) + if other, exists := seen[title]; exists { + t.Fatalf("%q and %q share title %q", other, operation, title) + } + seen[title] = operation + } +} + +func TestCardTitleNamesTrafficSyncType(t *testing.T) { + cases := []struct{ operation, want string }{ + {"previous_day_usage_full_sync", "生产 · 日流量同步"}, + {"previous_month_usage_full_sync", "生产 · 月历史同步"}, + {"current_month_usage_full_sync", "生产 · 本月流量快照同步"}, + } + for _, c := range cases { + a := archive.RunArchive{Version: 1, Run: archive.RunSummary{RunID: "run-title", Operation: c.operation}} + body, err := Card(a, "prod") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(body), `"content":"`+c.want+`"`) { + t.Fatalf("card for %q missing title %q: %s", c.operation, c.want, body) + } + } +} + +func TestTextContainsRoundThreeFailure(t *testing.T) { + text := Text(sample(), "preview") + for _, want := range []string{"开始时间：2026-09-22 19:42:34", "结束时间：2026-09-22 20:32:01", "实际执行：3 轮", "第 3 轮", "终态失败 9,000", "5分5秒", "0.00%"} { + if !strings.Contains(text, want) { + t.Fatalf("text missing %q:\n%s", want, text) + } + } + if strings.Contains(text, "2026-09-22 20:26:56") { + t.Fatalf("text exposes per-round start time: %s", text) + } +} + +func TestCardIsStructuredAndShowsFullRunID(t *testing.T) { + body, err := Card(sample(), "preview") + if err != nil { + t.Fatal(err) + } + value := string(body) + for _, want := range []string{ + `"msg_type":"interactive"`, + `"content":"预演 · 卡状态全量同步"`, + `"fields"`, + `**开始时间**\n2026-09-22 19:42:34`, + `**结束时间**\n2026-09-22 20:32:01`, + `第 3 轮 · 9,000 张 · 5分5秒`, + `进入下轮 0 · 终态失败 9,000`, + `Run: run-1`, + } { + if !strings.Contains(value, want) { + t.Fatalf("card missing %q: %s", want, body) + } + } + if strings.Contains(value, "2026-09-22 20:26:56") { + t.Fatalf("card exposes per-round start time: %s", body) + } +} + +func TestCardShowsFullRunID(t *testing.T) { + runID := "sim-status-20260922194233-c96b21ec7d3746ae9caef457e89faa41" + card, err := Card(archive.RunArchive{Version: 1, Run: archive.RunSummary{RunID: runID}}, "preview") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(card), "Run: "+runID) { + t.Fatalf("card truncated Run ID: %s", card) + } +} diff --git a/internal/render/supply.go b/internal/render/supply.go new file mode 100644 index 0000000..f5aa668 --- /dev/null +++ b/internal/render/supply.go @@ -0,0 +1,68 @@ +package render + +import ( + "fmt" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive" + "sort" + "strings" +) + +func supplyBlock(a archive.RunArchive) string { + if !isTrafficUsage(a.Run.Operation) || a.Run.SyncType == "replay" { + return "" + } + s, present := a.Run.Supply() + if !present { + return "" + } + if s == nil { + return "供料观测：不可用（遥测格式非法）；业务结果不受影响" + } + status := []string{} + if s.Partial { + status = append(status, "partial") + } + if s.Truncated { + status = append(status, "truncated") + } + if len(status) == 0 { + status = append(status, "完整") + } + unknown, unconfirmed := s.Summary.UnknownMS, s.Summary.UnconfirmedMS + lines := []string{"供料观测（Run 首轮扫描窗口，约1秒分辨率，时间为采样估计）", fmt.Sprintf("目标 %d · 有效 %d · 确认缺料 %d · unknown 通道 %d", s.Summary.Targets, s.Summary.Valid, s.Summary.Confirmed, s.Summary.Unknown), fmt.Sprintf("最长中断 %.1f秒 · 累计 %.1f通道秒（允许通道重叠）", float64(s.Summary.LongestMS)/1000, float64(s.Summary.TotalMS)/1000), fmt.Sprintf("待确认空闲 %.1f通道秒 · unknown %.1f通道秒 · %s", float64(unconfirmed)/1000, float64(unknown)/1000, strings.Join(status, " / "))} + if s.Truncated { + lines = append(lines, "明细已截断；Run摘要覆盖全部目标，归档保留最长等待前100通道") + } + if s.Summary.Confirmed == 0 && !s.Partial && !s.Truncated && s.Summary.Unknown == 0 && unconfirmed == 0 && s.Summary.Valid == s.Summary.Targets { + lines = append(lines, "约1秒分辨率下未发现确认缺料") + } + channels := append([]archive.SupplyChannel(nil), s.Channels...) + sort.Slice(channels, func(i, j int) bool { + a, b := channels[i], channels[j] + av, bv := max(a.Startup.LongestMS, a.Confirmed.LongestMS), max(b.Startup.LongestMS, b.Confirmed.LongestMS) + if av == bv { + return a.Key < b.Key + } + return av > bv + }) + for _, c := range channels[:min(5, len(channels))] { + longest := max(c.Startup.LongestMS, c.Confirmed.LongestMS) + if longest == 0 { + continue + } + phase := "未记录" + var phaseMS int64 + for k, v := range c.Stages { + if v > phaseMS || (v == phaseMS && k < phase) { + phase, phaseMS = k, v + } + } + signal := "" + if longest >= 5000 { + signal = "（≥5秒排查信号）" + } + lines = append(lines, fmt.Sprintf("%s：最长 %.1f秒%s；启动 %.1f秒 / 后续 %.1f秒；同期主要生产者阶段 %s %.1f秒", c.Key, float64(longest)/1000, signal, float64(c.Startup.TotalMS)/1000, float64(c.Confirmed.TotalMS)/1000, phase, float64(phaseMS)/1000)) + } + lines = append(lines, "小于采样间隔的中断可能漏检；同期阶段不证明因果，5秒提示不构成SLA。完整明细见归档。") + return strings.Join(lines, "\n") +} diff --git a/internal/render/supply_test.go b/internal/render/supply_test.go new file mode 100644 index 0000000..dba2246 --- /dev/null +++ b/internal/render/supply_test.go @@ -0,0 +1,49 @@ +package render + +import ( + "encoding/json" + "git.5iot.com/iotlcm-public/iotlcm-bot/internal/archive" + "strings" + "testing" +) + +func TestSupplyBlockCoversEveryTrafficOperation(t *testing.T) { + observation := json.RawMessage(`{"schemaVersion":1,"resolutionMs":1000,"summary":{"targetChannels":1,"validChannels":1,"confirmedChannels":1,"confirmedChannelMs":6000,"longestMs":6000}}`) + for _, operation := range []string{"previous_day_usage_full_sync", "previous_month_usage_full_sync", "current_month_usage_full_sync", "traffic_usage_full_sync"} { + a := archive.RunArchive{Run: archive.RunSummary{Operation: operation, SupplyObservation: observation}} + if supplyBlock(a) == "" { + t.Fatalf("traffic operation %q must show supply block", operation) + } + } + for _, operation := range []string{"sim_status_queue_full_sync", "imei_full_sync"} { + a := archive.RunArchive{Run: archive.RunSummary{Operation: operation, SupplyObservation: observation}} + if supplyBlock(a) != "" { + t.Fatalf("non-traffic operation %q must omit supply block", operation) + } + } +} + +func TestSupplyCompatibilityAndPartial(t *testing.T) { + a := archive.RunArchive{Run: archive.RunSummary{Operation: "current_month_usage_full_sync"}} + if supplyBlock(a) != "" { + t.Fatal("legacy archive must omit telemetry") + } + a.Run.SupplyObservation = json.RawMessage(`{"schemaVersion":1,"resolutionMs":1000,"partial":true,"summary":{"targetChannels":2,"validChannels":2,"confirmedChannels":1,"confirmedChannelMs":6000,"longestMs":6000},"channels":[{"chanKey":"a","confirmedWait":{"totalMs":6000,"longestMs":6000},"producerStageMs":{"global_capacity":6000}}]}`) + text := supplyBlock(a) + for _, want := range []string{"partial", "6.0通道秒", "global_capacity", "约1秒分辨率", "≥5秒"} { + if !strings.Contains(text, want) { + t.Fatalf("missing %s in %s", want, text) + } + } + if strings.Contains(text, "全程供料充足") { + t.Fatal("unsupported conclusion") + } + a.Run.SupplyObservation = json.RawMessage(`"bad"`) + if !strings.Contains(supplyBlock(a), "不可用") { + t.Fatal("bad optional telemetry must degrade") + } + a.Run.SyncType = "replay" + if supplyBlock(a) != "" { + t.Fatal("replay changed") + } +} diff --git a/internal/state/state.go b/internal/state/state.go new file mode 100644 index 0000000..daa20d3 --- /dev/null +++ b/internal/state/state.go @@ -0,0 +1,64 @@ +package state + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" +) + +type State struct { + RunID string `json:"runId"` + SentAt time.Time `json:"sentAt"` +} + +type Store struct{ dir string } + +func New(dir string) Store { return Store{dir: dir} } +func (s Store) Prepare() error { + dir := filepath.Join(s.dir, "sent") + if err := os.MkdirAll(dir, 0o700); err != nil { + return fmt.Errorf("create sent marker directory: %w", err) + } + file, err := os.CreateTemp(dir, ".write-check-*") + if err != nil { + return fmt.Errorf("check sent marker directory: %w", err) + } + name := file.Name() + if err := file.Close(); err != nil { + _ = os.Remove(name) + return fmt.Errorf("close sent marker write check: %w", err) + } + if err := os.Remove(name); err != nil { + return fmt.Errorf("remove sent marker write check: %w", err) + } + return nil +} + +func (s Store) Has(runID string) (bool, error) { + _, err := os.Stat(filepath.Join(s.dir, "sent", runID+".json")) + if err == nil { + return true, nil + } + if os.IsNotExist(err) { + return false, nil + } + return false, fmt.Errorf("check sent marker: %w", err) +} +func (s Store) Mark(runID string) error { + if err := s.Prepare(); err != nil { + return err + } + dir := filepath.Join(s.dir, "sent") + body, _ := json.Marshal(State{RunID: runID, SentAt: time.Now()}) + tmp := filepath.Join(dir, fmt.Sprintf(".%s.tmp", runID)) + final := filepath.Join(dir, runID+".json") + if err := os.WriteFile(tmp, body, 0o600); err != nil { + return err + } + return os.Rename(tmp, final) +} + +// Dir 返回状态目录，供调用方复用（例如告警节流状态落盘）。 +func (s Store) Dir() string { return s.dir } diff --git a/internal/state/state_test.go b/internal/state/state_test.go new file mode 100644 index 0000000..31b37be --- /dev/null +++ b/internal/state/state_test.go @@ -0,0 +1,31 @@ +package state + +import "testing" + +func TestPrepareCreatesWritableSentDirectory(t *testing.T) { + s := New(t.TempDir()) + if err := s.Prepare(); err != nil { + t.Fatal(err) + } +} + +func TestMarkIsIdempotentForLookup(t *testing.T) { + s := New(t.TempDir()) + sent, err := s.Has("run-1") + if err != nil { + t.Fatal(err) + } + if sent { + t.Fatal("run should not be marked before send") + } + if err := s.Mark("run-1"); err != nil { + t.Fatal(err) + } + sent, err = s.Has("run-1") + if err != nil { + t.Fatal(err) + } + if !sent { + t.Fatal("run should be marked") + } +} diff --git a/testdata/current-run.json b/testdata/current-run.json new file mode 100644 index 0000000..5c89e59 --- /dev/null +++ b/testdata/current-run.json @@ -0,0 +1,64 @@ +{ + "version": 1, + "run": { + "runId": "sim-status-20260922194233-c96b21ec7d3746ae9caef457e89faa41", + "operation": "sim_status_queue_full_sync", + "queryMonth": "", + "status": "completed_with_failures", + "producerDone": true, + "currentRound": 3, + "activeCount": 0, + "inflightCount": 0, + "nextRoundCount": 0, + "successCount": 175893, + "failedCount": 10504, + "createdAt": "2026-09-22T11:42:33.562765565Z", + "producerCompletedAt": "2026-09-22T11:43:36.656900096Z", + "completedAt": "2026-09-22T12:32:01.555760358Z", + "syncType": "normal" + }, + "rounds": [ + { + "runId": "sim-status-20260922194233-c96b21ec7d3746ae9caef457e89faa41", + "round": 1, + "enqueuedCount": 186397, + "activeCount": 0, + "nextRoundCount": 9023, + "successCount": 175870, + "failedCount": 1504, + "upstreamAttemptCount": 0, + "batchCount": 1872, + "startedAt": "2026-09-22T11:42:34.483546236Z", + "completedAt": "2026-09-22T12:21:49.958362058Z", + "errorCounts": {} + }, + { + "runId": "sim-status-20260922194233-c96b21ec7d3746ae9caef457e89faa41", + "round": 2, + "enqueuedCount": 9023, + "activeCount": 0, + "nextRoundCount": 9000, + "successCount": 23, + "failedCount": 0, + "upstreamAttemptCount": 0, + "batchCount": 92, + "startedAt": "2026-09-22T12:21:51.441846314Z", + "completedAt": "2026-09-22T12:26:55.781082913Z", + "errorCounts": {} + }, + { + "runId": "sim-status-20260922194233-c96b21ec7d3746ae9caef457e89faa41", + "round": 3, + "enqueuedCount": 9000, + "activeCount": 0, + "nextRoundCount": 0, + "successCount": 0, + "failedCount": 9000, + "upstreamAttemptCount": 0, + "batchCount": 90, + "startedAt": "2026-09-22T12:26:56.677525864Z", + "completedAt": "2026-09-22T12:32:01.555013325Z", + "errorCounts": {} + } + ] +}