From e115d828ab346e5dbede70009b87387432c49e4b Mon Sep 17 00:00:00 2001 From: dushixiang Date: Tue, 5 May 2026 17:46:32 +0800 Subject: [PATCH] =?UTF-8?q?=E9=87=8D=E6=9E=84=E6=95=B0=E6=8D=AE=E9=87=87?= =?UTF-8?q?=E9=9B=86=E6=89=93=E5=8C=85=EF=BC=8C=E4=BF=AE=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E6=89=93=E5=8C=85=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/handler/agent_ws_handler.go | 29 +--- internal/protocol/message.go | 14 +- pkg/agent/collector/manager.go | 212 +++++++++++---------------- pkg/agent/service/agent.go | 189 ++++++++---------------- pkg/agent/service/metrics_buffer.go | 8 +- 5 files changed, 169 insertions(+), 283 deletions(-) diff --git a/internal/handler/agent_ws_handler.go b/internal/handler/agent_ws_handler.go index 9fa46fe..8833ec1 100644 --- a/internal/handler/agent_ws_handler.go +++ b/internal/handler/agent_ws_handler.go @@ -80,9 +80,6 @@ func (h *AgentHandler) handleWebSocketMessage(ctx context.Context, agentID strin case protocol.MessageTypeMetrics: return h.handleMetricsMessage(ctx, agentID, data) - case protocol.MessageTypeBundle: - return h.handleBundleMessage(ctx, agentID, data) - case protocol.MessageTypeCommandResp: return h.handleCommandResponseMessage(ctx, agentID, data) @@ -164,31 +161,19 @@ func (h *AgentHandler) handleWebSocketPong(agentID string) { } func (h *AgentHandler) handleMetricsMessage(ctx context.Context, agentID string, data json.RawMessage) error { - var metricsWrapper protocol.MetricsPayload - if err := json.Unmarshal(data, &metricsWrapper); err != nil { - return err - } - metricsData, err := json.Marshal(metricsWrapper.Data) - if err != nil { - return err - } - return h.metricService.HandleMetricData(ctx, agentID, string(metricsWrapper.Type), metricsData, metricsWrapper.Timestamp) -} - -func (h *AgentHandler) handleBundleMessage(ctx context.Context, agentID string, data json.RawMessage) error { - var bundle protocol.BundlePayload - if err := json.Unmarshal(data, &bundle); err != nil { + var batch protocol.MetricsBatch + if err := json.Unmarshal(data, &batch); err != nil { return err } - for _, item := range bundle.Items { - metricsData, err := json.Marshal(item.Data) + for _, sample := range batch.Samples { + metricsData, err := json.Marshal(sample.Data) if err != nil { - h.logger.Warn("failed to marshal bundled metric item", zap.Error(err)) + h.logger.Warn("failed to marshal metric sample", zap.Error(err)) continue } - if err := h.metricService.HandleMetricData(ctx, agentID, string(item.Type), metricsData, item.Timestamp); err != nil { - h.logger.Warn("failed to handle bundled metric item", zap.Error(err), zap.String("type", string(item.Type))) + if err := h.metricService.HandleMetricData(ctx, agentID, string(sample.Type), metricsData, sample.Timestamp); err != nil { + h.logger.Warn("failed to handle metric sample", zap.Error(err), zap.String("type", string(sample.Type))) continue } } diff --git a/internal/protocol/message.go b/internal/protocol/message.go index 6d32a52..b4300d8 100644 --- a/internal/protocol/message.go +++ b/internal/protocol/message.go @@ -37,17 +37,18 @@ type AgentInfo struct { Version string `json:"version"` // 版本号 } -// MetricsPayload 指标数据包装，发送端/接收端统一使用 -type MetricsPayload struct { +// MetricSample 单条指标采样 +type MetricSample struct { Type MetricType `json:"type"` Data interface{} `json:"data"` Timestamp int64 `json:"timestamp,omitempty"` // 客户端采集时间(毫秒) } -// BundlePayload 批量指标数据包装 -type BundlePayload struct { - Items []MetricsPayload `json:"items"` +// MetricsBatch 一次采集周期内的批量指标，是 metrics 上报的唯一载体 +type MetricsBatch struct { + Samples []MetricSample `json:"samples"` } + type MessageType string // 控制消息 @@ -58,9 +59,8 @@ const ( MessageTypeCommand MessageType = "command" MessageTypeCommandResp MessageType = "command_response" MessageTypeUninstall MessageType = "uninstall" - // 指标消息 + // 指标消息（批量） MessageTypeMetrics MessageType = "metrics" - MessageTypeBundle MessageType = "bundle" MessageTypeMonitorConfig MessageType = "monitor_config" // 防篡改消息 MessageTypeTamperProtect MessageType = "tamper_protect" diff --git a/pkg/agent/collector/manager.go b/pkg/agent/collector/manager.go index 7f13b8c..59cb6b8 100644 --- a/pkg/agent/collector/manager.go +++ b/pkg/agent/collector/manager.go @@ -1,18 +1,18 @@ package collector import ( + "errors" "time" "github.com/dushixiang/pika/internal/protocol" "github.com/dushixiang/pika/pkg/agent/config" ) -// WebSocketWriter 定义 WebSocket 写入接口 -type WebSocketWriter interface { - WriteJSON(v interface{}) error -} +// ErrNoData 表示采集成功但当前周期无有效数据可上报（如 GPU 不存在、磁盘列表为空）。 +// 调用方应静默跳过，不视为错误。 +var ErrNoData = errors.New("no data") -// Manager 采集器管理器 +// Manager 采集器管理器：纯采集，不做 I/O type Manager struct { cpuCollector *CPUCollector memoryCollector *MemoryCollector @@ -24,7 +24,6 @@ type Manager struct { temperatureCollector *TemperatureCollector gpuCollector *GPUCollector monitorCollector *MonitorCollector - ddnsCollector *DDNSCollector } // NewManager 创建采集器管理器 @@ -40,152 +39,115 @@ func NewManager(cfg *config.Config) *Manager { temperatureCollector: NewTemperatureCollector(), gpuCollector: NewGPUCollector(), monitorCollector: NewMonitorCollector(), - ddnsCollector: nil, // DDNS 采集器需要配置后才能初始化 } } -// CollectAndSendCPU 采集并发送 CPU 指标 -func (m *Manager) CollectAndSendCPU(conn WebSocketWriter) error { - cpuData, err := m.cpuCollector.Collect() +// makeSample 构造一条 MetricSample +func makeSample(t protocol.MetricType, data interface{}) protocol.MetricSample { + return protocol.MetricSample{ + Type: t, + Data: data, + Timestamp: time.Now().UnixMilli(), + } +} + +// CollectCPU 采集 CPU 指标 +func (m *Manager) CollectCPU() (protocol.MetricSample, error) { + data, err := m.cpuCollector.Collect() if err != nil { - return err + return protocol.MetricSample{}, err } - - return m.sendMetrics(conn, protocol.MetricTypeCPU, cpuData) + return makeSample(protocol.MetricTypeCPU, data), nil } -// CollectAndSendMemory 采集并发送内存指标 -func (m *Manager) CollectAndSendMemory(conn WebSocketWriter) error { - memData, err := m.memoryCollector.Collect() +// CollectMemory 采集内存指标 +func (m *Manager) CollectMemory() (protocol.MetricSample, error) { + data, err := m.memoryCollector.Collect() if err != nil { - return err + return protocol.MetricSample{}, err } - - return m.sendMetrics(conn, protocol.MetricTypeMemory, memData) + return makeSample(protocol.MetricTypeMemory, data), nil } -// CollectAndSendDisk 采集并发送磁盘指标 -func (m *Manager) CollectAndSendDisk(conn WebSocketWriter) error { - diskDataList, err := m.diskCollector.Collect() - if err != nil || len(diskDataList) == 0 { - // 采集失败或无有效数据，不发送空列表 - return nil - } - return m.sendMetrics(conn, protocol.MetricTypeDisk, diskDataList) -} - -// CollectAndSendDiskIO 采集并发送磁盘 IO 指标 -func (m *Manager) CollectAndSendDiskIO(conn WebSocketWriter) error { - diskIODataList, err := m.diskIOCollector.Collect() +// CollectDisk 采集磁盘指标。无有效数据时返回 ErrNoData。 +func (m *Manager) CollectDisk() (protocol.MetricSample, error) { + data, err := m.diskCollector.Collect() if err != nil { - return err + return protocol.MetricSample{}, err } - return m.sendMetrics(conn, protocol.MetricTypeDiskIO, diskIODataList) + if len(data) == 0 { + return protocol.MetricSample{}, ErrNoData + } + return makeSample(protocol.MetricTypeDisk, data), nil } -// CollectAndSendNetwork 采集并发送网络指标 -func (m *Manager) CollectAndSendNetwork(conn WebSocketWriter) error { - networkDataList, err := m.networkCollector.Collect() - if err != nil || len(networkDataList) == 0 { - // 采集失败或无有效数据，不发送空列表 - return nil - } - return m.sendMetrics(conn, protocol.MetricTypeNetwork, networkDataList) -} - -// CollectAndSendNetworkConnection 采集并发送网络连接统计 -func (m *Manager) CollectAndSendNetworkConnection(conn WebSocketWriter) error { - connectionData, err := m.networkConnectionCollector.Collect() +// CollectDiskIO 采集磁盘 IO 指标 +func (m *Manager) CollectDiskIO() (protocol.MetricSample, error) { + data, err := m.diskIOCollector.Collect() if err != nil { - return err + return protocol.MetricSample{}, err } - return m.sendMetrics(conn, protocol.MetricTypeNetworkConnection, connectionData) + return makeSample(protocol.MetricTypeDiskIO, data), nil } -// CollectAndSendHost 采集并发送主机信息 -func (m *Manager) CollectAndSendHost(conn WebSocketWriter) error { - hostData, err := m.hostCollector.Collect() +// CollectNetwork 采集网络指标。无有效数据时返回 ErrNoData。 +func (m *Manager) CollectNetwork() (protocol.MetricSample, error) { + data, err := m.networkCollector.Collect() if err != nil { - return err + return protocol.MetricSample{}, err } - - return m.sendMetrics(conn, protocol.MetricTypeHost, hostData) + if len(data) == 0 { + return protocol.MetricSample{}, ErrNoData + } + return makeSample(protocol.MetricTypeNetwork, data), nil } -// CollectAndSendGPU 采集并发送 GPU 指标 -func (m *Manager) CollectAndSendGPU(conn WebSocketWriter) error { - gpuDataList, err := m.gpuCollector.Collect() - if err != nil || len(gpuDataList) == 0 { - // GPU 监控不是必须的,失败或无数据时直接返回 - return nil - } - - return m.sendMetrics(conn, protocol.MetricTypeGPU, gpuDataList) -} - -// CollectAndSendTemperature 采集并发送温度信息 -func (m *Manager) CollectAndSendTemperature(conn WebSocketWriter) error { - tempDataList, err := m.temperatureCollector.Collect() - if err != nil || len(tempDataList) == 0 { - // 温度监控不是必须的,失败或无数据时直接返回 - return nil - } - - return m.sendMetrics(conn, protocol.MetricTypeTemperature, tempDataList) -} - -// CollectAndSendMonitor 采集并发送监控数据 -func (m *Manager) CollectAndSendMonitor(conn WebSocketWriter, items []protocol.MonitorItem) error { - monitorDataList := m.monitorCollector.Collect(items) - return m.sendMetrics(conn, protocol.MetricTypeMonitor, monitorDataList) -} - -// UpdateDDNSConfig 更新 DDNS 配置 -func (m *Manager) UpdateDDNSConfig(config *protocol.DDNSConfigData) { - if config == nil || !config.Enabled { - m.ddnsCollector = nil - return - } - - if m.ddnsCollector == nil { - m.ddnsCollector = NewDDNSCollector(config) - } else { - m.ddnsCollector.UpdateConfig(config) - } -} - -// CollectAndSendDDNSIP 采集并发送 DDNS IP 地址 -func (m *Manager) CollectAndSendDDNSIP(conn WebSocketWriter) error { - if m.ddnsCollector == nil { - return nil // DDNS 未启用，静默返回 - } - - ipData, err := m.ddnsCollector.Collect() +// CollectNetworkConnection 采集网络连接统计 +func (m *Manager) CollectNetworkConnection() (protocol.MetricSample, error) { + data, err := m.networkConnectionCollector.Collect() if err != nil { - return err + return protocol.MetricSample{}, err } - - // 只有当至少有一个 IP 地址时才发送 - if ipData.IPv4 == "" && ipData.IPv6 == "" { - return nil - } - - return conn.WriteJSON(protocol.OutboundMessage{ - Type: protocol.MessageTypeDDNSIPReport, - Data: ipData, - }) + return makeSample(protocol.MetricTypeNetworkConnection, data), nil } -// sendMetrics 发送指标数据 -func (m *Manager) sendMetrics(conn WebSocketWriter, metricType protocol.MetricType, data interface{}) error { - return conn.WriteJSON(protocol.OutboundMessage{ - Type: protocol.MessageTypeMetrics, - Data: protocol.MetricsPayload{ - Type: metricType, - Data: data, - Timestamp: time.Now().UnixMilli(), - }, - }) +// CollectHost 采集主机信息（含 Load） +func (m *Manager) CollectHost() (protocol.MetricSample, error) { + data, err := m.hostCollector.Collect() + if err != nil { + return protocol.MetricSample{}, err + } + return makeSample(protocol.MetricTypeHost, data), nil +} + +// CollectGPU 采集 GPU 指标。无 GPU 时返回 ErrNoData（GPU 不是必备） +func (m *Manager) CollectGPU() (protocol.MetricSample, error) { + data, err := m.gpuCollector.Collect() + if err != nil { + return protocol.MetricSample{}, err + } + if len(data) == 0 { + return protocol.MetricSample{}, ErrNoData + } + return makeSample(protocol.MetricTypeGPU, data), nil +} + +// CollectTemperature 采集温度信息。无传感器时返回 ErrNoData（温度不是必备） +func (m *Manager) CollectTemperature() (protocol.MetricSample, error) { + data, err := m.temperatureCollector.Collect() + if err != nil { + return protocol.MetricSample{}, err + } + if len(data) == 0 { + return protocol.MetricSample{}, ErrNoData + } + return makeSample(protocol.MetricTypeTemperature, data), nil +} + +// CollectMonitor 采集服务监控数据（HTTP/TCP 探活） +func (m *Manager) CollectMonitor(items []protocol.MonitorItem) protocol.MetricSample { + data := m.monitorCollector.Collect(items) + return makeSample(protocol.MetricTypeMonitor, data) } // GetPublicIP 通过 API 获取公网 IP 地址 diff --git a/pkg/agent/service/agent.go b/pkg/agent/service/agent.go index 71028ab..84d49a3 100644 --- a/pkg/agent/service/agent.go +++ b/pkg/agent/service/agent.go @@ -4,6 +4,7 @@ import ( "context" "crypto/tls" "encoding/json" + "errors" "fmt" "log/slog" "net/http" @@ -28,10 +29,10 @@ import ( ) const ( - agentPingInterval = 10 * time.Second - agentPongWait = 30 * time.Second - agentWriteWait = 5 * time.Second - agentCollectTimeout = 30 * time.Second + agentPingInterval = 10 * time.Second + agentPongWait = 30 * time.Second + agentWriteWait = 5 * time.Second + agentCollectTimeout = 30 * time.Second ) // safeConn 线程安全的 WebSocket 连接包装器 @@ -162,8 +163,8 @@ func (a *Agent) runOnce(ctx context.Context, onRegistered func()) error { // 创建自定义的 Dialer dialer := &websocket.Dialer{ - Proxy: http.ProxyFromEnvironment, - HandshakeTimeout: 45 * time.Second, + Proxy: http.ProxyFromEnvironment, + HandshakeTimeout: 45 * time.Second, EnableCompression: true, } if a.cfg.Server.InsecureSkipVerify { @@ -429,10 +430,14 @@ func (a *Agent) handleMonitorConfig(data json.RawMessage) { slog.Info("收到服务监控配置，立即执行检测", "count", len(payload.Items)) - // 立即执行一次监控检测 + // 立即执行一次监控检测，结果以单元素 batch 上报 + sample := manager.CollectMonitor(payload.Items) writer := newOutboundWriter(a.getActiveConn(), a.outboundBuffer) - if err := manager.CollectAndSendMonitor(writer, payload.Items); err != nil { - slog.Warn("监控检测失败", "error", err) + if err := writer.WriteJSON(protocol.OutboundMessage{ + Type: protocol.MessageTypeMetrics, + Data: protocol.MetricsBatch{Samples: []protocol.MetricSample{sample}}, + }); err != nil { + slog.Warn("发送监控结果失败", "error", err) } else { slog.Info("服务监控检测完成，已上报或缓存监控项结果", "count", len(payload.Items)) } @@ -511,26 +516,6 @@ func (a *Agent) metricsLoop(ctx context.Context) { } } -type bundleWriter struct { - items []protocol.MetricsPayload -} - -func (w *bundleWriter) WriteJSON(v interface{}) error { - msg, ok := v.(protocol.OutboundMessage) - if !ok { - return fmt.Errorf("invalid message type for bundleWriter") - } - - if msg.Type == protocol.MessageTypeMetrics { - payload, ok := msg.Data.(protocol.MetricsPayload) - if ok { - w.items = append(w.items, payload) - return nil - } - } - return fmt.Errorf("only metrics messages can be bundled") -} - // collectAndSendAllMetrics 采集并发送所有动态指标 func (a *Agent) collectAndSendAllMetrics(manager *collector.Manager) error { if manager == nil { @@ -566,12 +551,15 @@ func newCollectTimer(name string) *collectTimer { func (t *collectTimer) done() { d := time.Since(t.start) - if d > 500*time.Millisecond { + if d > 1500*time.Millisecond { slog.Info("采集耗时", "collector", t.name, "duration", d) } } -// doCollectAndSend 执行实际的采集和发送 +// collectFn 单个采集器的采集函数签名 +type collectFn func() (protocol.MetricSample, error) + +// doCollectAndSend 采集所有动态指标，打包成一个 batch 上报 func (a *Agent) doCollectAndSend(manager *collector.Manager) error { conn := a.getActiveConn() if conn != nil { @@ -582,108 +570,55 @@ func (a *Agent) doCollectAndSend(manager *collector.Manager) error { } } - writer := newOutboundWriter(conn, a.outboundBuffer) - bw := &bundleWriter{} + // required=true 时采集失败计入错误；required=false 适用于 GPU/温度等可选项 + collectors := []struct { + name string + required bool + 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}, + } + + samples := make([]protocol.MetricSample, 0, len(collectors)) var hasError bool - // CPU 动态指标 - func() { - t := newCollectTimer("cpu") - defer t.done() - if err := manager.CollectAndSendCPU(bw); err != nil { - slog.Warn("采集CPU指标失败", "error", err) - hasError = true - } - }() + for _, c := range collectors { + func() { + 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) + }() + } - // 内存动态指标 - func() { - t := newCollectTimer("memory") - defer t.done() - if err := manager.CollectAndSendMemory(bw); err != nil { - slog.Warn("采集内存指标失败", "error", err) - hasError = true - } - }() + writer := newOutboundWriter(conn, a.outboundBuffer) - // 磁盘指标 - func() { - t := newCollectTimer("disk") - defer t.done() - if err := manager.CollectAndSendDisk(bw); err != nil { - slog.Warn("采集磁盘指标失败", "error", err) - hasError = true - } - }() - - // 磁盘 IO 指标 - func() { - t := newCollectTimer("disk_io") - defer t.done() - if err := manager.CollectAndSendDiskIO(bw); err != nil { - slog.Warn("采集磁盘IO指标失败", "error", err) - hasError = true - } - }() - - // 网络指标 - func() { - t := newCollectTimer("network") - defer t.done() - if err := manager.CollectAndSendNetwork(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） - func() { - t := newCollectTimer("host") - defer t.done() - if err := manager.CollectAndSendHost(bw); err != nil { - slog.Warn("采集主机信息失败", "error", err) - hasError = true - } - }() - - // GPU 信息（可选） - func() { - t := newCollectTimer("gpu") - defer t.done() - if err := manager.CollectAndSendGPU(bw); err != nil { - slog.Info("采集GPU信息失败", "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 { - bundle := protocol.BundlePayload{ - Items: bw.items, - } + if len(samples) > 0 { if err := writer.WriteJSON(protocol.OutboundMessage{ - Type: protocol.MessageTypeBundle, - Data: bundle, + Type: protocol.MessageTypeMetrics, + Data: protocol.MetricsBatch{Samples: samples}, }); err != nil { - slog.Warn("发送打包指标失败", "error", err) + slog.Warn("发送指标 batch 失败", "error", err) hasError = true } } diff --git a/pkg/agent/service/metrics_buffer.go b/pkg/agent/service/metrics_buffer.go index cc52008..631a12d 100644 --- a/pkg/agent/service/metrics_buffer.go +++ b/pkg/agent/service/metrics_buffer.go @@ -11,7 +11,6 @@ import ( "time" "github.com/dushixiang/pika/internal/protocol" - "github.com/dushixiang/pika/pkg/agent/collector" "github.com/dushixiang/pika/pkg/agent/utils" bolt "go.etcd.io/bbolt" ) @@ -23,6 +22,11 @@ const ( outboundBufferRetention = 24 * time.Hour ) +// WebSocketWriter 抽象 JSON 写入操作，由 outboundBuffer 和 outboundWriter 共用 +type WebSocketWriter interface { + WriteJSON(v interface{}) error +} + type outboundBuffer struct { path string mu sync.Mutex @@ -101,7 +105,7 @@ func (b *outboundBuffer) Append(v interface{}) error { return nil } -func (b *outboundBuffer) Flush(writer collector.WebSocketWriter) (int, error) { +func (b *outboundBuffer) Flush(writer WebSocketWriter) (int, error) { b.mu.Lock() defer b.mu.Unlock()