fix: 修复 Agent 卡死不触发重连的问题

Agent 在连接断开后完全卡死 17 小时，根因分析如下：
1. 并发写竞争：pingLoop 直接使用 rawConn.WriteControl 与 metricsLoop
   的 safeConn.WriteJSON 竞争同一个 gorilla websocket 内部锁
2. 连接写失败被吞掉：outboundWriter.WriteJSON 写失败时返回 nil，
   metricsLoop 感知不到连接错误，不会触发重连
3. 采集器无超时保护：任何采集器阻塞都会冻结整个 metricsLoop

修复：
- safeConn 新增 WriteControl 方法，pingLoop 改用 safeConn 写入
- outboundWriter.WriteJSON 写失败时返回真实错误
- collectAndSendAllMetrics 增加 30 秒全局超时保护
- 连接写失败视为连接异常，返回错误触发重连
- 新增采集耗时监控日志（耗时超过 500ms 时记录）

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

pkg/agent/service/agent.go

@@ -28,9 +28,10 @@ import (
 )
 
 const (
-	agentPingInterval = 10 * time.Second
-	agentPongWait     = 30 * time.Second
-	agentWriteWait    = 5 * time.Second
+	agentPingInterval       = 10 * time.Second
+	agentPongWait           = 30 * time.Second
+	agentWriteWait          = 5 * time.Second
+	agentCollectTimeout     = 30 * time.Second
 )
 
 // safeConn 线程安全的 WebSocket 连接包装器
@@ -63,6 +64,13 @@ func (sc *safeConn) Close() error {
 	return sc.conn.Close()
 }
 
+// WriteControl 线程安全地写入控制消息
+func (sc *safeConn) WriteControl(messageType int, data []byte, deadline time.Time) error {
+	sc.mu.Lock()
+	defer sc.mu.Unlock()
+	return sc.conn.WriteControl(messageType, data, deadline)
+}
+
 // Agent 探针服务
 type Agent struct {
 	cfg              *config.Config
@@ -222,7 +230,7 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error {
 
 	// 主动发送 Ping，快速检测断线
 	wg.Go(func() {
-		if err := a.pingLoop(ctx, rawConn, done); err != nil {
+		if err := a.pingLoop(ctx, conn, done); err != nil {
 			select {
 			case errChan <- fmt.Errorf("ping失败: %w", err):
 			default:
@@ -314,7 +322,7 @@ func (a *Agent) readLoop(conn *websocket.Conn, done chan struct{}) error {
 	}
 }
 
-func (a *Agent) pingLoop(ctx context.Context, conn *websocket.Conn, done chan struct{}) error {
+func (a *Agent) pingLoop(ctx context.Context, conn *safeConn, done chan struct{}) error {
 	ticker := time.NewTicker(agentPingInterval)
 	defer ticker.Stop()
 
@@ -529,6 +537,42 @@ func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager) error {
 		return fmt.Errorf("采集器未初始化")
 	}
 
+	type result struct {
+		err error
+	}
+	done := make(chan result, 1)
+
+	go func() {
+		done <- result{err: a.doCollectAndSend(manager)}
+	}()
+
+	select {
+	case r := <-done:
+		return r.err
+	case <-time.After(agentCollectTimeout):
+		return fmt.Errorf("指标采集超时 (超过 %v)", agentCollectTimeout)
+	}
+}
+
+// collectTimer 采集耗时计时器，记录每个采集器的耗时
+type collectTimer struct {
+	name  string
+	start time.Time
+}
+
+func newCollectTimer(name string) *collectTimer {
+	return &collectTimer{name: name, start: time.Now()}
+}
+
+func (t *collectTimer) done() {
+	d := time.Since(t.start)
+	if d > 500*time.Millisecond {
+		slog.Info("采集耗时", "collector", t.name, "duration", d)
+	}
+}
+
+// doCollectAndSend 执行实际的采集和发送
+func (a *Agent) doCollectAndSend(manager *collector.Manager) error {
 	conn := a.getActiveConn()
 	if conn != nil {
 		if sent, err := a.outboundBuffer.Flush(conn); err != nil {
@@ -543,56 +587,92 @@ func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager) error {
 	var hasError bool
 
 	// CPU 动态指标
-	if err := manager.CollectAndSendCPU(bw); err != nil {
-		slog.Warn("采集CPU指标失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("cpu")
+		defer t.done()
+		if err := manager.CollectAndSendCPU(bw); err != nil {
+			slog.Warn("采集CPU指标失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// 内存动态指标
-	if err := manager.CollectAndSendMemory(bw); err != nil {
-		slog.Warn("采集内存指标失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("memory")
+		defer t.done()
+		if err := manager.CollectAndSendMemory(bw); err != nil {
+			slog.Warn("采集内存指标失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// 磁盘指标
-	if err := manager.CollectAndSendDisk(bw); err != nil {
-		slog.Warn("采集磁盘指标失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("disk")
+		defer t.done()
+		if err := manager.CollectAndSendDisk(bw); err != nil {
+			slog.Warn("采集磁盘指标失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// 磁盘 IO 指标
-	if err := manager.CollectAndSendDiskIO(bw); err != nil {
-		slog.Warn("采集磁盘IO指标失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("disk_io")
+		defer t.done()
+		if err := manager.CollectAndSendDiskIO(bw); err != nil {
+			slog.Warn("采集磁盘IO指标失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// 网络指标
-	if err := manager.CollectAndSendNetwork(bw); err != nil {
-		slog.Warn("采集网络指标失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("network")
+		defer t.done()
+		if err := manager.CollectAndSendNetwork(bw); err != nil {
+			slog.Warn("采集网络指标失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// 网络连接统计
-	if err := manager.CollectAndSendNetworkConnection(bw); err != nil {
-		slog.Warn("采集网络连接统计失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("network_connection")
+		defer t.done()
+		if err := manager.CollectAndSendNetworkConnection(bw); err != nil {
+			slog.Warn("采集网络连接统计失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// 主机信息（包含 Load）
-	if err := manager.CollectAndSendHost(bw); err != nil {
-		slog.Warn("采集主机信息失败", "error", err)
-		hasError = true
-	}
+	func() {
+		t := newCollectTimer("host")
+		defer t.done()
+		if err := manager.CollectAndSendHost(bw); err != nil {
+			slog.Warn("采集主机信息失败", "error", err)
+			hasError = true
+		}
+	}()
 
 	// GPU 信息（可选）
-	if err := manager.CollectAndSendGPU(bw); err != nil {
-		slog.Info("采集GPU信息失败", "error", err)
-	}
+	func() {
+		t := newCollectTimer("gpu")
+		defer t.done()
+		if err := manager.CollectAndSendGPU(bw); err != nil {
+			slog.Info("采集GPU信息失败", "error", err)
+		}
+	}()
 
 	// 温度信息（可选）
-	if err := manager.CollectAndSendTemperature(bw); err != nil {
-		slog.Info("采集温度信息失败", "error", err)
-	}
+	func() {
+		t := newCollectTimer("temperature")
+		defer t.done()
+		if err := manager.CollectAndSendTemperature(bw); err != nil {
+			slog.Info("采集温度信息失败", "error", err)
+		}
+	}()
 
 	// 发送打包后的指标
 	if len(bw.items) > 0 {
@@ -613,6 +693,8 @@ func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager) error {
 			slog.Info("当前连接不可用，消息已写入缓存")
 		} else if writer.sendErr != nil {
 			slog.Warn("发送消息失败，已写入缓存", "error", writer.sendErr)
+			// 连接写失败视为连接异常，返回错误触发重连
+			return fmt.Errorf("连接发送失败: %w", writer.sendErr)
 		}
 	}
 

pkg/agent/service/metrics_buffer.go

@@ -236,7 +236,7 @@ func (w *outboundWriter) WriteJSON(v interface{}) error {
 		}
 		w.buffered = true
 		w.sendErr = err
-		return nil
+		return err
 	}
 
 	return nil