chore: 初始开源发布
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
This commit is contained in:
MiFaZhan
@@ -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
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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 }
|
||||
@@ -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 })
|
||||
}
|
||||
@@ -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}))
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user