新增实时监控模式,agent 1s 并行采集 + 前端滑动窗口

agent
- 移除 collector.interval 配置项;按指标类型内置节奏:cpu/memory/network/
  disk_io/gpu/network_connection 1s,temperature 5s,disk 30s,host 60s
- doCollectAndSend 改为并行执行本 tick 到期的采集器,避免串行累计耗时
  超过 1s 节奏;每个 goroutine 加 recover 兜底,单点 panic 不拖垮整 tick

server
- LatestMetrics 加 sync.RWMutex + Update/Snapshot 方法,HandleMetricData
  所有写入走 Update;GetLatestMetrics 返回 Snapshot,handler 在副本上
  sanitize,修复 marshal 时与 ws 写入的 data race(go test -race 验证)
- HandleMetricData defer Set 缓存项,每次上报刷新 TTL;host case 也走
  Update,不再 1 小时后失效
- LatestMetrics.Timestamp 用服务端 time.Now(),agent 时钟倒拨不会冻结
  前端实时图;Timestamp/DiskIOSummary 字段补到 LatestMetrics
- AutoStep 新增 r ≤ 5m → 1s 档;timeRangeMilliseconds 加 2m

frontend
- 详情页加"实时"档位(默认),1s 拉 latestMetrics + 1m 历史播种 + 滑动
  2 分钟窗口;快图表(CPU/Memory/Network/DiskIO/NetworkConnection)走
  useLiveBuffer 追加,慢图表(GPU/Temperature/Monitor)保留 refetch
- useLiveBuffer:seed 不再覆盖已追加的 livePoint;resetKey 处理切网卡 /
  切 agent 时的语义复位
- formatChartTime live 档位精确到秒;修复 1d/24h 仅显示 HH:mm 跨夜歧义
- 慢图表用 React.memo,阻断 1Hz 节奏的级联重渲染
- GpuChart 修复 utilization/temperature 时间戳错位时温度点丢失

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
dushixiangandClaude Opus 4.7 committed 2026-05-06 23:18:50 +08:00
1 parent e115d828ab
commit c431f300bc
25 files changed
+655 -217

No files matched your search

+1 -15
View File
@@ -69,9 +69,6 @@ type AgentConfig struct {
// CollectorConfig 采集器配置
type CollectorConfig struct {
// 数据采集间隔(秒)
Interval int `yaml:"interval"`
// 网络采集包含的网卡列表(白名单,支持正则表达式)
// 如果配置了此项,则只采集匹配的网卡,忽略 NetworkExclude
// 例如: ["^eth0$", "^en0$", "^ens.*"]
@@ -116,9 +113,7 @@ func DefaultConfig() *Config {
LogMaxAge: 28,
LogCompress: true,
},
Collector: CollectorConfig{
Interval: 5,
},
Collector: CollectorConfig{},
AutoUpdate: AutoUpdateConfig{
Enabled: true,
CheckInterval: "10m",
@@ -197,10 +192,6 @@ func (c *Config) Save(path string) error {
// Validate 验证配置
func (c *Config) Validate() error {
if c.Collector.Interval <= 0 {
c.Collector.Interval = 5
}
if c.AutoUpdate.Enabled {
if _, err := time.ParseDuration(c.AutoUpdate.CheckInterval); err != nil {
return fmt.Errorf("更新检查间隔格式错误: %w", err)
@@ -224,11 +215,6 @@ func (c *Config) Validate() error {
return nil
}
// GetCollectorInterval 获取采集间隔时长
func (c *Config) GetCollectorInterval() time.Duration {
return time.Duration(c.Collector.Interval) * time.Second
}
// GetUpdateCheckInterval 获取更新检查间隔时长
func (c *Config) GetUpdateCheckInterval() time.Duration {
duration, _ := time.ParseDuration(c.AutoUpdate.CheckInterval)
+75 -36
View File
@@ -486,6 +486,9 @@ func (a *Agent) getCollectorManager() *collector.Manager {
return a.collectorManager
}
// collectorBaseInterval 采集循环的基础节奏,所有采集器的采集间隔必须是它的整数倍
const collectorBaseInterval = 1 * time.Second
// metricsLoop 指标采集循环
func (a *Agent) metricsLoop(ctx context.Context) {
manager := a.getCollectorManager()
@@ -494,20 +497,20 @@ func (a *Agent) metricsLoop(ctx context.Context) {
a.setCollectorManager(manager)
}
// 立即采集一次动态数据
if err := a.collectAndSendAllMetrics(manager); err != nil {
// 立即采集一次(tick=0 时所有采集器都到期,等于全量采集)
var tickCount uint64
if err := a.collectAndSendAllMetrics(manager, tickCount); err != nil {
slog.Warn("初始数据采集失败", "error", err)
}
// 定时采集动态指标
ticker := time.NewTicker(a.cfg.GetCollectorInterval())
ticker := time.NewTicker(collectorBaseInterval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
// 采集并发送各种动态指标
if err := a.collectAndSendAllMetrics(manager); err != nil {
tickCount++
if err := a.collectAndSendAllMetrics(manager, tickCount); err != nil {
slog.Warn("数据采集失败", "error", err)
}
case <-ctx.Done():
@@ -517,7 +520,7 @@ func (a *Agent) metricsLoop(ctx context.Context) {
}
// collectAndSendAllMetrics 采集并发送所有动态指标
func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager) error {
func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager, tickCount uint64) error {
if manager == nil {
return fmt.Errorf("采集器未初始化")
}
@@ -528,7 +531,7 @@ func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager) error {
done := make(chan result, 1)
go func() {
done <- result{err: a.doCollectAndSend(manager)}
done <- result{err: a.doCollectAndSend(manager, tickCount)}
}()
select {
@@ -560,7 +563,8 @@ func (t *collectTimer) done() {
type collectFn func() (protocol.MetricSample, error)
// doCollectAndSend 采集所有动态指标,打包成一个 batch 上报
func (a *Agent) doCollectAndSend(manager *collector.Manager) error {
// 各采集器按自身 interval 节流;本 tick 到期的采集器并行执行,避免串行累计耗时压垮 1s 节奏
func (a *Agent) doCollectAndSend(manager *collector.Manager, tickCount uint64) error {
conn := a.getActiveConn()
if conn != nil {
if sent, err := a.outboundBuffer.Flush(conn); err != nil {
@@ -571,45 +575,80 @@ func (a *Agent) doCollectAndSend(manager *collector.Manager) error {
}
// required=true 时采集失败计入错误;required=false 适用于 GPU/温度等可选项
// interval 必须为 collectorBaseInterval 的整数倍
collectors := []struct {
name string
required bool
interval time.Duration
fn collectFn
}{
{"cpu", true, manager.CollectCPU},
{"memory", true, manager.CollectMemory},
{"disk", true, manager.CollectDisk},
{"disk_io", true, manager.CollectDiskIO},
{"network", true, manager.CollectNetwork},
{"network_connection", true, manager.CollectNetworkConnection},
{"host", true, manager.CollectHost},
{"gpu", false, manager.CollectGPU},
{"temperature", false, manager.CollectTemperature},
{"cpu", true, 1 * time.Second, manager.CollectCPU},
{"memory", true, 1 * time.Second, manager.CollectMemory},
{"disk_io", true, 1 * time.Second, manager.CollectDiskIO},
{"network", true, 1 * time.Second, manager.CollectNetwork},
{"gpu", false, 1 * time.Second, manager.CollectGPU},
{"network_connection", true, 1 * time.Second, manager.CollectNetworkConnection},
{"temperature", false, 5 * time.Second, manager.CollectTemperature},
{"disk", true, 30 * time.Second, manager.CollectDisk},
{"host", true, 60 * time.Second, manager.CollectHost},
}
samples := make([]protocol.MetricSample, 0, len(collectors))
var hasError bool
type collectResult struct {
name string
required bool
sample protocol.MetricSample
err error
}
for _, c := range collectors {
func() {
// 收集本 tick 到期的采集器
due := make([]int, 0, len(collectors))
for i, c := range collectors {
every := uint64(c.interval / collectorBaseInterval)
if every == 0 || tickCount%every != 0 {
continue
}
due = append(due, i)
}
results := make(chan collectResult, len(due))
var wg sync.WaitGroup
for _, idx := range due {
c := collectors[idx]
wg.Add(1)
go func() {
defer wg.Done()
defer func() {
if r := recover(); r != nil {
slog.Error("采集器 panic", "collector", c.name, "panic", r)
results <- collectResult{name: c.name, required: c.required, err: fmt.Errorf("panic: %v", r)}
}
}()
t := newCollectTimer(c.name)
defer t.done()
sample, err := c.fn()
if err != nil {
if errors.Is(err, collector.ErrNoData) {
return
}
if c.required {
slog.Warn("采集指标失败", "collector", c.name, "error", err)
hasError = true
} else {
slog.Info("采集可选指标失败", "collector", c.name, "error", err)
}
return
}
samples = append(samples, sample)
t.done()
results <- collectResult{name: c.name, required: c.required, sample: sample, err: err}
}()
}
wg.Wait()
close(results)
samples := make([]protocol.MetricSample, 0, len(due))
var hasError bool
for r := range results {
if r.err != nil {
if errors.Is(r.err, collector.ErrNoData) {
continue
}
if r.required {
slog.Warn("采集指标失败", "collector", r.name, "error", r.err)
hasError = true
} else {
slog.Info("采集可选指标失败", "collector", r.name, "error", r.err)
}
continue
}
samples = append(samples, r.sample)
}
writer := newOutboundWriter(conn, a.outboundBuffer)
+1 -2
View File
@@ -290,8 +290,7 @@ func (m *ServiceManager) Run() error {
})
slog.Info("配置加载成功",
"server_endpoint", m.cfg.Server.Endpoint,
"collector_interval", m.cfg.GetCollectorInterval())
"server_endpoint", m.cfg.Server.Endpoint)
// 初始化系统配置(Linux ICMP 权限等)
configureICMP()