From 3c4f9783959187fe0f4666f78bfd49b664c98cf1 Mon Sep 17 00:00:00 2001 From: dushixiang Date: Sun, 12 Jul 2026 21:00:04 +0800 Subject: [PATCH] Refactor agent metrics collection and delivery --- pkg/agent/service/agent.go | 193 +++++++----------------- pkg/agent/service/metrics_store.go | 97 ++++++++++++ pkg/agent/service/metrics_store_test.go | 95 ++++++++++++ pkg/agent/service/scheduler.go | 114 ++++++++++++++ 4 files changed, 361 insertions(+), 138 deletions(-) create mode 100644 pkg/agent/service/metrics_store.go create mode 100644 pkg/agent/service/metrics_store_test.go create mode 100644 pkg/agent/service/scheduler.go diff --git a/pkg/agent/service/agent.go b/pkg/agent/service/agent.go index cd085b8..b302e7a 100644 --- a/pkg/agent/service/agent.go +++ b/pkg/agent/service/agent.go @@ -4,7 +4,6 @@ import ( "context" "crypto/tls" "encoding/json" - "errors" "fmt" "log/slog" "net/http" @@ -81,6 +80,7 @@ type Agent struct { activeConn *safeConn collectorMu sync.RWMutex collectorManager *collector.Manager + metricsStore *metricsStore tamperProtector *tamper.Protector sshMonitor *sshmonitor.Monitor } @@ -91,6 +91,7 @@ func New(cfg *config.Config) *Agent { cfg: cfg, idMgr: id.NewManager(), collectorManager: collector.NewManager(cfg), + metricsStore: newMetricsStore(), tamperProtector: tamper.NewProtector(), sshMonitor: sshmonitor.NewMonitor(), } @@ -102,7 +103,9 @@ func (a *Agent) Start(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) a.cancel = cancel - go a.metricsLoop(ctx) + // 采集与发送解耦：采集永远运行（断线也写快照），发送独立读快照上报 + go a.collectLoop(ctx) + go a.sendLoop(ctx) // 启动探针主循环 b := &backoff.Backoff{ @@ -206,6 +209,9 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { slog.Info("探针注册成功，开始监控...") a.setActiveConn(conn) + // 重置发送游标，使下一个 sendLoop tick 立即发送当前全量快照， + // 保证重连瞬间服务端就能拿到最新数据 + a.metricsStore.reset() defer func() { a.setActiveConn(nil) }() @@ -483,16 +489,19 @@ func (a *Agent) getCollectorManager() *collector.Manager { return a.collectorManager } -// collectorBaseInterval 采集循环的基础节奏，所有采集器的采集间隔必须是它的整数倍 -const collectorBaseInterval = 1 * time.Second +// metricsSendInterval 指标发送循环的节奏，独立于采集节奏 +const metricsSendInterval = 1 * time.Second -// metricsLoop 指标采集循环 -func (a *Agent) metricsLoop(ctx context.Context) { +// collectLoop 指标采集循环：只采集写快照，不关心连接状态。 +// 断线期间采集持续运行，避免重连后数据出现空洞。 +func (a *Agent) collectLoop(ctx context.Context) { manager := a.getCollectorManager() if manager == nil { manager = collector.NewManager(a.cfg) a.setCollectorManager(manager) } + // 采集器表只构造一次，后续每个 tick 复用 + scheduler := newMetricsScheduler(manager) var tickCount uint64 ticker := time.NewTicker(collectorBaseInterval) @@ -501,13 +510,7 @@ func (a *Agent) metricsLoop(ctx context.Context) { for { select { case <-ticker.C: - if a.getActiveConn() == nil { - continue - } - - if err := a.collectAndSendAllMetrics(manager, tickCount); err != nil { - slog.Warn("数据采集失败", "error", err) - } + a.collectOnce(scheduler, tickCount) tickCount++ case <-ctx.Done(): return @@ -515,150 +518,64 @@ func (a *Agent) metricsLoop(ctx context.Context) { } } -// collectAndSendAllMetrics 采集并发送所有动态指标 -func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager, tickCount uint64) error { - if manager == nil { - return fmt.Errorf("collector is not initialized") - } - - type result struct { - err error - } - done := make(chan result, 1) - +// collectOnce 在超时保护下采集本 tick 到期的指标并写入快照存储。 +func (a *Agent) collectOnce(scheduler *metricsScheduler, tickCount uint64) { + done := make(chan struct{}) go func() { - done <- result{err: a.doCollectAndSend(manager, tickCount)} + defer close(done) + samples, hasError := scheduler.collect(tickCount) + if len(samples) > 0 { + a.metricsStore.put(samples) + } + if hasError { + slog.Warn("部分指标采集失败") + } }() select { - case r := <-done: - return r.err + case <-done: case <-time.After(agentCollectTimeout): - return fmt.Errorf("metrics collection timed out after %v", agentCollectTimeout) + slog.Warn("数据采集超时", "timeout", agentCollectTimeout) } } -// collectTimer 采集耗时计时器，记录每个采集器的耗时 -type collectTimer struct { - name string - start time.Time -} +// sendLoop 指标发送循环：独立于采集，定时读取快照发送"新采集"的样本。 +func (a *Agent) sendLoop(ctx context.Context) { + ticker := time.NewTicker(metricsSendInterval) + defer ticker.Stop() -func newCollectTimer(name string) *collectTimer { - return &collectTimer{name: name, start: time.Now()} -} - -func (t *collectTimer) done() { - d := time.Since(t.start) - if d > 1500*time.Millisecond { - slog.Info("采集耗时", "collector", t.name, "duration", d) + for { + select { + case <-ticker.C: + a.sendMetricsOnce() + case <-ctx.Done(): + return + } } } -// collectFn 单个采集器的采集函数签名 -type collectFn func() (protocol.MetricSample, error) - -// doCollectAndSend 采集所有动态指标，打包成一个 batch 上报 -// 各采集器按自身 interval 节流；本 tick 到期的采集器并行执行，避免串行累计耗时压垮 1s 节奏 -func (a *Agent) doCollectAndSend(manager *collector.Manager, tickCount uint64) error { +// sendMetricsOnce 发送快照中比上次更新的样本；无连接或无新数据时跳过。 +func (a *Agent) sendMetricsOnce() { conn := a.getActiveConn() - - // required=true 时采集失败计入错误；required=false 适用于 GPU/温度等可选项 - // interval 必须为 collectorBaseInterval 的整数倍 - collectors := []struct { - name string - required bool - interval time.Duration - fn collectFn - }{ - {"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}, + if conn == nil { + return } - type collectResult struct { - name string - required bool - sample protocol.MetricSample - err error + samples, cursor := a.metricsStore.pending() + if len(samples) == 0 { + return } - // 收集本 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) + if err := conn.WriteJSON(protocol.OutboundMessage{ + Type: protocol.MessageTypeMetrics, + Data: protocol.MetricsBatch{Samples: samples}, + }); err != nil { + slog.Warn("发送指标 batch 失败", "error", err) + // 连接写失败仅用于日志，重连由 readLoop/pingLoop 驱动 + return } - 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) - sample, err := c.fn() - 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) - } - - if len(samples) > 0 { - if conn == nil { - return fmt.Errorf("connection is unavailable") - } - - if err := conn.WriteJSON(protocol.OutboundMessage{ - Type: protocol.MessageTypeMetrics, - Data: protocol.MetricsBatch{Samples: samples}, - }); err != nil { - slog.Warn("发送指标 batch 失败", "error", err) - // 连接写失败视为连接异常，返回错误触发重连 - return fmt.Errorf("connection send failed: %w", err) - } - } - - if hasError { - return fmt.Errorf("some metrics collection failed") - } - - return nil + a.metricsStore.ack(cursor) } // handleCommand 处理服务端下发的指令 diff --git a/pkg/agent/service/metrics_store.go b/pkg/agent/service/metrics_store.go new file mode 100644 index 0000000..a33d012 --- /dev/null +++ b/pkg/agent/service/metrics_store.go @@ -0,0 +1,97 @@ +package service + +import ( + "sort" + "sync" + + "github.com/dushixiang/pika/internal/protocol" +) + +// metricsStore is a constant-size, latest-wins snapshot store. Delivery state +// uses an internal sequence instead of sample timestamps, which may collide or +// move backwards when the system clock changes. +type metricsStore struct { + mu sync.RWMutex + latest map[protocol.MetricType]versionedMetric + nextSeq uint64 + ackedSeq uint64 + generation uint64 +} + +type versionedMetric struct { + sample protocol.MetricSample + seq uint64 +} + +// metricsCursor identifies exactly the snapshot observed by pending. The +// generation prevents a delayed acknowledgement from an old connection from +// advancing the cursor after reset. +type metricsCursor struct { + seq uint64 + generation uint64 +} + +func newMetricsStore() *metricsStore { + return &metricsStore{latest: make(map[protocol.MetricType]versionedMetric)} +} + +// put stores samples in collection order. The last inserted sample for a +// metric type wins, regardless of its wall-clock timestamp. +func (s *metricsStore) put(samples []protocol.MetricSample) { + s.mu.Lock() + defer s.mu.Unlock() + + for _, sample := range samples { + s.nextSeq++ + s.latest[sample.Type] = versionedMetric{sample: sample, seq: s.nextSeq} + } +} + +// pending returns samples not acknowledged on the current connection. It does +// not mutate delivery state; the caller must acknowledge the cursor only after +// the complete network write succeeds. +func (s *metricsStore) pending() ([]protocol.MetricSample, metricsCursor) { + s.mu.RLock() + defer s.mu.RUnlock() + + cursor := metricsCursor{seq: s.ackedSeq, generation: s.generation} + samples := make([]protocol.MetricSample, 0, len(s.latest)) + for _, metric := range s.latest { + if metric.seq <= s.ackedSeq { + continue + } + samples = append(samples, metric.sample) + if metric.seq > cursor.seq { + cursor.seq = metric.seq + } + } + + sort.Slice(samples, func(i, j int) bool { + if samples[i].Timestamp == samples[j].Timestamp { + return samples[i].Type < samples[j].Type + } + return samples[i].Timestamp < samples[j].Timestamp + }) + return samples, cursor +} + +// ack commits a successful write. Stale and old-connection acknowledgements +// are harmless. +func (s *metricsStore) ack(cursor metricsCursor) { + s.mu.Lock() + defer s.mu.Unlock() + + if cursor.generation == s.generation && cursor.seq > s.ackedSeq { + s.ackedSeq = cursor.seq + } +} + +// reset starts a new delivery generation and makes the current full snapshot +// pending for the new connection. +func (s *metricsStore) reset() { + s.mu.Lock() + defer s.mu.Unlock() + + s.generation++ + s.ackedSeq = 0 +} diff --git a/pkg/agent/service/metrics_store_test.go b/pkg/agent/service/metrics_store_test.go new file mode 100644 index 0000000..317f189 --- /dev/null +++ b/pkg/agent/service/metrics_store_test.go @@ -0,0 +1,95 @@ +package service + +import ( + "testing" + + "github.com/dushixiang/pika/internal/protocol" +) + +func metricSample(metricType protocol.MetricType, timestamp int64) protocol.MetricSample { + return protocol.MetricSample{Type: metricType, Timestamp: timestamp} +} + +func TestMetricsStorePendingRequiresAck(t *testing.T) { + store := newMetricsStore() + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeCPU, 100)}) + + first, cursor := store.pending() + second, _ := store.pending() + if len(first) != 1 || len(second) != 1 { + t.Fatalf("unacknowledged sample must remain pending: first=%d second=%d", len(first), len(second)) + } + + store.ack(cursor) + afterAck, _ := store.pending() + if len(afterAck) != 0 { + t.Fatalf("acknowledged sample returned again: %v", afterAck) + } +} + +func TestMetricsStoreAckDoesNotConsumeConcurrentUpdate(t *testing.T) { + store := newMetricsStore() + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeCPU, 100)}) + _, cursor := store.pending() + + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeMemory, 100)}) + store.ack(cursor) + + pending, _ := store.pending() + if len(pending) != 1 || pending[0].Type != protocol.MetricTypeMemory { + t.Fatalf("concurrent update was consumed by an older ack: %v", pending) + } +} + +func TestMetricsStoreUsesSequenceInsteadOfTimestampCursor(t *testing.T) { + store := newMetricsStore() + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeCPU, 100)}) + _, cursor := store.pending() + store.ack(cursor) + + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeMemory, 100)}) + pending, _ := store.pending() + if len(pending) != 1 || pending[0].Type != protocol.MetricTypeMemory { + t.Fatalf("sample sharing an acknowledged timestamp was lost: %v", pending) + } +} + +func TestMetricsStoreIgnoresAckFromPreviousGeneration(t *testing.T) { + store := newMetricsStore() + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeCPU, 100)}) + _, oldCursor := store.pending() + + store.reset() + store.ack(oldCursor) + + pending, _ := store.pending() + if len(pending) != 1 { + t.Fatalf("old connection ack consumed reset snapshot: %v", pending) + } +} + +func TestMetricsStoreLastInsertedSampleWinsWithinBatch(t *testing.T) { + store := newMetricsStore() + store.put([]protocol.MetricSample{ + metricSample(protocol.MetricTypeCPU, 200), + metricSample(protocol.MetricTypeCPU, 100), + }) + + pending, _ := store.pending() + if len(pending) != 1 || pending[0].Timestamp != 100 { + t.Fatalf("last inserted sample did not win: %v", pending) + } +} + +func TestMetricsStoreAcceptsUpdateAfterClockMovesBackward(t *testing.T) { + store := newMetricsStore() + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeCPU, 200)}) + _, cursor := store.pending() + store.ack(cursor) + + store.put([]protocol.MetricSample{metricSample(protocol.MetricTypeCPU, 100)}) + pending, _ := store.pending() + if len(pending) != 1 || pending[0].Timestamp != 100 { + t.Fatalf("sample collected after clock moved backward was lost: %v", pending) + } +} diff --git a/pkg/agent/service/scheduler.go b/pkg/agent/service/scheduler.go new file mode 100644 index 0000000..1885c22 --- /dev/null +++ b/pkg/agent/service/scheduler.go @@ -0,0 +1,114 @@ +package service + +import ( + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/dushixiang/pika/internal/protocol" + "github.com/dushixiang/pika/pkg/agent/collector" +) + +// collectorBaseInterval 采集循环的基础节奏，所有采集器的采集间隔必须是它的整数倍 +const collectorBaseInterval = 1 * time.Second + +// slowCollectorThreshold 单个采集器耗时超过此阈值时打印慢日志 +const slowCollectorThreshold = 1500 * time.Millisecond + +// collectFn 单个采集器的采集函数签名 +type collectFn func() (protocol.MetricSample, error) + +// collectorSpec 描述一个周期性采集器 +type collectorSpec struct { + name string + required bool // required=true 时采集失败计入错误；GPU/温度等可选项为 false + interval time.Duration // 必须为 collectorBaseInterval 的整数倍 + fn collectFn +} + +// metricsScheduler 按 tick 调度并行采集，封装到期判定、并发执行、错误分类 +type metricsScheduler struct { + collectors []collectorSpec +} + +// newMetricsScheduler 用 manager 构造一次性的采集器表（不再每 tick 重建） +func newMetricsScheduler(m *collector.Manager) *metricsScheduler { + return &metricsScheduler{ + collectors: []collectorSpec{ + {"cpu", true, 1 * time.Second, m.CollectCPU}, + {"memory", true, 1 * time.Second, m.CollectMemory}, + {"disk_io", true, 1 * time.Second, m.CollectDiskIO}, + {"network", true, 1 * time.Second, m.CollectNetwork}, + {"gpu", false, 1 * time.Second, m.CollectGPU}, + {"network_connection", true, 1 * time.Second, m.CollectNetworkConnection}, + {"temperature", false, 5 * time.Second, m.CollectTemperature}, + {"disk", true, 30 * time.Second, m.CollectDisk}, + {"host", true, 60 * time.Second, m.CollectHost}, + }, + } +} + +// collectResult 单个采集器的执行结果 +type collectResult struct { + name string + required bool + sample protocol.MetricSample + err error +} + +// collect 执行本次 tick 到期的采集器，返回样本与是否有错误。 +// 到期采集器并行执行，每个有 panic 恢复；错误按 ErrNoData/required/optional 三态分类。 +func (s *metricsScheduler) collect(tickCount uint64) (samples []protocol.MetricSample, hasError bool) { + results := make(chan collectResult, len(s.collectors)) + var wg sync.WaitGroup + + for i := range s.collectors { + c := &s.collectors[i] + every := uint64(c.interval / collectorBaseInterval) + if every == 0 || tickCount%every != 0 { + continue + } + + 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)} + } + }() + + start := time.Now() + sample, err := c.fn() + if d := time.Since(start); d > slowCollectorThreshold { + slog.Info("采集耗时", "collector", c.name, "duration", d) + } + results <- collectResult{name: c.name, required: c.required, sample: sample, err: err} + }() + } + + wg.Wait() + close(results) + + samples = make([]protocol.MetricSample, 0, len(s.collectors)) + 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) + } + + return samples, hasError +}