Refactor agent metrics collection and delivery

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 处理服务端下发的指令

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
+}

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)
+	}
+}

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
+}