package snapshot import ( "context" "sync" "time" "mfd-monitor-agent/internal/status" ) // Collector 将 HostStatus 的一个子区块采集并写入提供的快照。 type Collector interface { // Name 返回用于 partial 快照错误记录的稳定采集器名称。 Name() string // Collect 填充 out 中对应子区块;该子区块不完整时返回错误。 Collect(context.Context, *status.HostStatus) error } // CollectorFunc 将函数适配为 Collector,供测试与小型采集器使用。 type CollectorFunc func(context.Context, *status.HostStatus) error // Name 为基于函数的采集器返回通用名称。 func (fn CollectorFunc) Name() string { return "collector" } // Collect 调用包装的采集函数。 func (fn CollectorFunc) Collect(ctx context.Context, out *status.HostStatus) error { return fn(ctx, out) } // Cache 在并发安全接口后保存最近一次 HostStatus 快照。 type Cache struct { collectors []Collector version string startedAt time.Time mu sync.RWMutex last status.HostStatus hasLast bool } // NewCache 使用给定采集器与 agent 元数据创建空的快照缓存。 func NewCache(collectors []Collector, version string, startedAt time.Time) *Cache { return &Cache{collectors: collectors, version: version, startedAt: startedAt} } // Collect 运行全部采集器一次并存储结果;部分采集器失败时仍保存快照。 func (c *Cache) Collect(ctx context.Context) status.HostStatus { out := status.HostStatus{ Agent: status.AgentInfo{ Version: c.version, StartedAt: c.startedAt, SnapshotAt: time.Now().UTC(), SnapshotStatus: "ok", }, Processes: status.ProcessResult{Status: "ok"}, TCPListeners: status.TCPResult{Status: "ok"}, } for _, collector := range c.collectors { if err := collector.Collect(ctx, &out); err != nil { out.Agent.SnapshotStatus = "partial" out.Errors = append(out.Errors, status.StatusError{ Collector: collector.Name(), Message: err.Error(), }) } } c.mu.Lock() c.last = out c.hasLast = true c.mu.Unlock() return out } // Get 返回最新缓存快照,以及是否至少完成过一次采集。 func (c *Cache) Get() (status.HostStatus, bool) { c.mu.RLock() defer c.mu.RUnlock() return c.last, c.hasLast } // Run 立即采集一次,随后在 ctx 取消前按间隔刷新缓存。 func (c *Cache) Run(ctx context.Context, interval time.Duration, timeout time.Duration) { c.collectWithTimeout(ctx, timeout) ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: c.collectWithTimeout(ctx, timeout) } } } // collectWithTimeout 在整体截止时间下执行一轮采集。 func (c *Cache) collectWithTimeout(ctx context.Context, timeout time.Duration) { collectCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() c.Collect(collectCtx) }