de3fd88948
* 默认监听调整为 0.0.0.0:8080，并要求查询参数 token * 支持首次 run 自动生成含随机 token 的配置文件 * 补充 HTTP 鉴权、配置校验和相关测试 * 完善代码注释，覆盖缓存、上报、服务和平台采集路径 * 新增架构与开发流程文档，补充一键构建脚本
114 lines
3.6 KiB
Go
114 lines
3.6 KiB
Go
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 是按固定顺序执行的子采集器集合。
|
||
collectors []Collector
|
||
// version 和 startedAt 会写入每次快照的 agent 元数据。
|
||
version string
|
||
startedAt time.Time
|
||
// mu 保护 last 和 hasLast，允许 HTTP 与上报并发读取缓存。
|
||
mu sync.RWMutex
|
||
// last 保存最近一次完整或 partial 的 HostStatus。
|
||
last status.HostStatus
|
||
// hasLast 表示首次采集是否已经完成。
|
||
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 {
|
||
// 单个子采集器失败只让整份快照降级为 partial，其他已成功字段继续对外可用。
|
||
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) {
|
||
// 启动后先同步完成一轮采集，让 /health 和 /status 尽快进入就绪状态。
|
||
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) {
|
||
// 超时通过 context 传递给子采集器；无法被 context 打断的系统调用仍依赖平台返回。
|
||
collectCtx, cancel := context.WithTimeout(ctx, timeout)
|
||
defer cancel()
|
||
c.Collect(collectCtx)
|
||
}
|