From d23ebbce22d42241ab369a3bc433deff027df5ab Mon Sep 17 00:00:00 2001 From: dushixiang Date: Sun, 30 Nov 2025 18:50:00 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=95=B0=E6=8D=AE=E6=8C=87?= =?UTF-8?q?=E6=A0=87=E7=9A=84=E5=AD=98=E5=82=A8=E5=92=8C=E6=9F=A5=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/app.go | 1 + internal/handler/agent_handler.go | 41 +- internal/repo/metric_repo.go | 65 ++- internal/service/agent_service.go | 720 +------------------------- internal/service/metric_service.go | 798 +++++++++++++++++++++++++++++ internal/wire.go | 1 + internal/wire_gen.go | 3 +- web/src/api/agent.ts | 17 +- 8 files changed, 902 insertions(+), 744 deletions(-) create mode 100644 internal/service/metric_service.go diff --git a/internal/app.go b/internal/app.go index e92fca0..6bd1d34 100644 --- a/internal/app.go +++ b/internal/app.go @@ -186,6 +186,7 @@ func setupApi(app *orz.App, components *AppComponents) { publicApiWithOptionalAuth.GET("/agents/:id", components.AgentHandler.Get) publicApiWithOptionalAuth.GET("/agents/:id/metrics", components.AgentHandler.GetMetrics) publicApiWithOptionalAuth.GET("/agents/:id/metrics/latest", components.AgentHandler.GetLatestMetrics) + publicApiWithOptionalAuth.GET("/agents/:id/network-interfaces", components.AgentHandler.GetAvailableNetworkInterfaces) // 监控统计数据（公开访问，支持可选认证）- 用于公共展示页面 publicApiWithOptionalAuth.GET("/monitors", components.MonitorHandler.GetMonitors) diff --git a/internal/handler/agent_handler.go b/internal/handler/agent_handler.go index 6926330..f07477a 100644 --- a/internal/handler/agent_handler.go +++ b/internal/handler/agent_handler.go @@ -406,6 +406,7 @@ func (h *AgentHandler) GetMetrics(c echo.Context) error { metricType := c.QueryParam("type") rangeParam := c.QueryParam("range") + interfaceName := c.QueryParam("interface") // 网卡过滤参数（仅对 network 类型有效） // 验证指标类型 validTypes := map[string]bool{ @@ -425,22 +426,20 @@ func (h *AgentHandler) GetMetrics(c echo.Context) error { return orz.NewError(400, err.Error()) } - // 服务端自动计算最优聚合间隔 - interval := service.CalculateInterval(start, end) - - metrics, err := h.agentService.GetMetrics(ctx, agentID, metricType, start, end, interval) + // GetMetrics 内部会自动计算最优聚合间隔 + metrics, err := h.agentService.GetMetrics(ctx, agentID, metricType, start, end, 0, interfaceName) if err != nil { return err } return orz.Ok(c, orz.Map{ - "agentId": agentID, - "type": metricType, - "range": rangeParam, - "start": start, - "end": end, - "interval": interval, - "metrics": metrics, + "agentId": agentID, + "type": metricType, + "range": rangeParam, + "start": start, + "end": end, + "interface": interfaceName, + "metrics": metrics, }) } @@ -462,6 +461,26 @@ func (h *AgentHandler) GetLatestMetrics(c echo.Context) error { return orz.Ok(c, metrics) } +// GetAvailableNetworkInterfaces 获取探针的可用网卡列表（公开接口，已登录返回全部，未登录返回公开可见） +func (h *AgentHandler) GetAvailableNetworkInterfaces(c echo.Context) error { + id := c.Param("id") + ctx := c.Request().Context() + + // 验证探针访问权限 + if _, err := h.agentService.GetAgentByAuth(ctx, id, utils.IsAuthenticated(c)); err != nil { + return err + } + + interfaces, err := h.agentService.GetAvailableNetworkInterfaces(ctx, id) + if err != nil { + return err + } + + return orz.Ok(c, orz.Map{ + "interfaces": interfaces, + }) +} + // GetAgents 获取探针列表（公开接口，已登录返回全部，未登录返回公开可见） func (h *AgentHandler) GetAgents(c echo.Context) error { ctx := c.Request().Context() diff --git a/internal/repo/metric_repo.go b/internal/repo/metric_repo.go index 9efe9d3..8660ca2 100644 --- a/internal/repo/metric_repo.go +++ b/internal/repo/metric_repo.go @@ -335,10 +335,15 @@ type AggregatedNetworkMetricByInterface struct { MaxRecvRate float64 `json:"maxRecvRate"` } -// GetNetworkMetrics 获取聚合后的网络指标（按网卡接口分组） -func (r *MetricRepo) GetNetworkMetrics(ctx context.Context, agentID string, start, end int64, interval int) ([]AggregatedNetworkMetric, error) { +// GetNetworkMetrics 获取聚合后的网络指标（可选按网卡接口过滤） +// 不指定网卡时查询 interface=” 的预聚合数据，指定网卡时只返回该网卡的数据 +func (r *MetricRepo) GetNetworkMetrics(ctx context.Context, agentID string, start, end int64, interval int, interfaceName string) ([]AggregatedNetworkMetric, error) { var metrics []AggregatedNetworkMetric + intervalMs := int64(interval * 1000) + + // 不管是否指定网卡，查询逻辑都一样：直接查询对应 interface 的数据 + // interfaceName 为空字符串时，会查询到预先保存的总和数据 query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, @@ -346,14 +351,13 @@ func (r *MetricRepo) GetNetworkMetrics(ctx context.Context, agentID string, star MAX(bytes_sent_rate) as max_sent_rate, MAX(bytes_recv_rate) as max_recv_rate FROM network_metrics - WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? + WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? AND interface = ? GROUP BY 1, interface - ORDER BY timestamp ASC, interface ASC + ORDER BY timestamp ASC ` - intervalMs := int64(interval * 1000) err := r.db.WithContext(ctx). - Raw(query, intervalMs, intervalMs, agentID, start, end). + Raw(query, intervalMs, intervalMs, agentID, start, end, interfaceName). Scan(&metrics).Error return metrics, err @@ -557,25 +561,17 @@ type AggregatedDiskIOMetric struct { MaxIopsInProgress uint64 `json:"maxIopsInProgress"` // 最大进行中的IO操作数 } -// GetDiskIOMetrics 获取聚合后的磁盘IO指标（汇总所有磁盘） +// GetDiskIOMetrics 获取聚合后的磁盘IO指标（已在存储时合并所有磁盘） func (r *MetricRepo) GetDiskIOMetrics(ctx context.Context, agentID string, start, end int64, interval int) ([]AggregatedDiskIOMetric, error) { var metrics []AggregatedDiskIOMetric query := ` SELECT CAST(FLOOR(timestamp / ?) * ? AS BIGINT) as timestamp, - SUM(read_bytes_rate) as max_read_rate, - SUM(write_bytes_rate) as max_write_rate, - SUM(CASE - WHEN read_bytes >= LAG(read_bytes) OVER (PARTITION BY device ORDER BY timestamp) - THEN read_bytes - LAG(read_bytes) OVER (PARTITION BY device ORDER BY timestamp) - ELSE 0 - END) as total_read_bytes, - SUM(CASE - WHEN write_bytes >= LAG(write_bytes) OVER (PARTITION BY device ORDER BY timestamp) - THEN write_bytes - LAG(write_bytes) OVER (PARTITION BY device ORDER BY timestamp) - ELSE 0 - END) as total_write_bytes, + MAX(read_bytes_rate) as max_read_rate, + MAX(write_bytes_rate) as max_write_rate, + MAX(read_bytes) as total_read_bytes, + MAX(write_bytes) as total_write_bytes, MAX(iops_in_progress) as max_iops_in_progress FROM disk_io_metrics WHERE agent_id = ? AND timestamp >= ? AND timestamp <= ? @@ -858,7 +854,7 @@ func (r *MetricRepo) AggregateNetworkConnectionToAgg(ctx context.Context, bucket `, bucketSeconds, bucketMs, bucketMs, start, end).Error } -// AggregateDiskIOToAgg 将原始磁盘IO数据聚合到聚合表（汇总所有磁盘） +// AggregateDiskIOToAgg 将原始磁盘IO数据聚合到聚合表（已在存储时合并所有磁盘） func (r *MetricRepo) AggregateDiskIOToAgg(ctx context.Context, bucketSeconds int, start, end int64) error { bucketMs := int64(bucketSeconds * 1000) return r.db.WithContext(ctx).Exec(` @@ -870,8 +866,8 @@ func (r *MetricRepo) AggregateDiskIOToAgg(ctx context.Context, bucketSeconds int agent_id, ? as bucket_seconds, (timestamp / ?) * ? as bucket_start, - SUM(read_bytes_rate) as max_read_bytes_rate, - SUM(write_bytes_rate) as max_write_bytes_rate, + MAX(read_bytes_rate) as max_read_bytes_rate, + MAX(write_bytes_rate) as max_write_bytes_rate, MAX(iops_in_progress) as max_iops_in_progress FROM disk_io_metrics WHERE timestamp >= ? AND timestamp < ? @@ -974,18 +970,35 @@ func (r *MetricRepo) GetDiskMetricsAgg(ctx context.Context, agentID string, star return metrics, err } -// GetNetworkMetricsAgg 从聚合表获取网络指标 -func (r *MetricRepo) GetNetworkMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedNetworkMetric, error) { +// GetNetworkMetricsAgg 从聚合表获取网络指标（可选按网卡接口过滤） +// 不指定网卡时查询 interface=” 的预聚合数据，指定网卡时只返回该网卡的数据 +func (r *MetricRepo) GetNetworkMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int, interfaceName string) ([]AggregatedNetworkMetric, error) { var metrics []AggregatedNetworkMetric + + // 不管是否指定网卡，查询逻辑都一样：直接查询对应 interface 的数据 + // interfaceName 为空字符串时，会查询到预先保存的总和数据 err := r.db.WithContext(ctx). Table("network_metrics_aggs"). Select("bucket_start as timestamp, interface, max_sent_rate, max_recv_rate"). - Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ?", agentID, bucketSeconds, start, end). - Order("bucket_start, interface"). + Where("agent_id = ? AND bucket_seconds = ? AND bucket_start >= ? AND bucket_start < ? AND interface = ?", + agentID, bucketSeconds, start, end, interfaceName). + Order("bucket_start"). Scan(&metrics).Error return metrics, err } +// GetAvailableNetworkInterfaces 获取探针的可用网卡列表（不包括空白的总和记录） +func (r *MetricRepo) GetAvailableNetworkInterfaces(ctx context.Context, agentID string) ([]string, error) { + var interfaces []string + err := r.db.WithContext(ctx). + Table("network_metrics"). + Select("DISTINCT interface"). + Where("agent_id = ? AND interface != ?", agentID, ""). // 排除空字符串（总和记录） + Order("interface"). + Pluck("interface", &interfaces).Error + return interfaces, err +} + // GetNetworkConnectionMetricsAgg 从聚合表获取网络连接指标 func (r *MetricRepo) GetNetworkConnectionMetricsAgg(ctx context.Context, agentID string, start, end int64, bucketSeconds int) ([]AggregatedNetworkConnectionMetric, error) { var metrics []AggregatedNetworkConnectionMetric diff --git a/internal/service/agent_service.go b/internal/service/agent_service.go index 3bf873d..af25efc 100644 --- a/internal/service/agent_service.go +++ b/internal/service/agent_service.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "math" "time" "github.com/dushixiang/pika/internal/models" @@ -16,38 +15,23 @@ import ( "gorm.io/gorm" ) -const ( - defaultMetricsRetentionHours = 24 * 7 // 默认保留 7 天 - defaultMaxQueryPoints = 720 // 默认最多返回 720 个点 -) - -var allowedIntervals = []int{ - 1, 2, 5, - 10, 15, 30, - 60, 120, 300, - 600, 900, 1800, - 3600, 7200, 14400, -} - type AgentService struct { logger *zap.Logger *orz.Service AgentRepo *repo.AgentRepo - metricRepo *repo.MetricRepo monitorStatsRepo *repo.MonitorStatsRepo apiKeyService *ApiKeyService - propertyService *PropertyService + metricService *MetricService } -func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyService, propertyService *PropertyService) *AgentService { +func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyService, metricService *MetricService) *AgentService { return &AgentService{ logger: logger, Service: orz.NewService(db), AgentRepo: repo.NewAgentRepo(db), - metricRepo: repo.NewMetricRepo(db), monitorStatsRepo: repo.NewMonitorStatsRepo(db), apiKeyService: apiKeyService, - propertyService: propertyService, + metricService: metricService, } } @@ -149,704 +133,32 @@ func (s *AgentService) ListOnlineAgents(ctx context.Context) ([]models.Agent, er // HandleMetricData 处理指标数据 func (s *AgentService) HandleMetricData(ctx context.Context, agentID string, metricType string, data json.RawMessage) error { - now := time.Now().UnixMilli() - - switch protocol.MetricType(metricType) { - case protocol.MetricTypeCPU: - // CPU数据现在包含静态和动态信息 - var cpuData protocol.CPUData - if err := json.Unmarshal(data, &cpuData); err != nil { - return err - } - metric := &models.CPUMetric{ - AgentID: agentID, - UsagePercent: cpuData.UsagePercent, - LogicalCores: cpuData.LogicalCores, - PhysicalCores: cpuData.PhysicalCores, - ModelName: cpuData.ModelName, - Timestamp: now, - } - return s.metricRepo.SaveCPUMetric(ctx, metric) - - case protocol.MetricTypeMemory: - // Memory数据现在包含静态和动态信息 - var memData protocol.MemoryData - if err := json.Unmarshal(data, &memData); err != nil { - return err - } - metric := &models.MemoryMetric{ - AgentID: agentID, - Total: memData.Total, // 现在可以从合并后的数据获取 - Used: memData.Used, - Free: memData.Free, - UsagePercent: memData.UsagePercent, - SwapTotal: memData.SwapTotal, - SwapUsed: memData.SwapUsed, - SwapFree: memData.SwapFree, - Timestamp: now, - } - return s.metricRepo.SaveMemoryMetric(ctx, metric) - - case protocol.MetricTypeDisk: - // Disk现在是数组,需要批量处理 - var diskDataList []protocol.DiskData - if err := json.Unmarshal(data, &diskDataList); err != nil { - return err - } - // 保存每个磁盘的数据 - for _, diskData := range diskDataList { - metric := &models.DiskMetric{ - AgentID: agentID, - MountPoint: diskData.MountPoint, - Total: diskData.Total, - Used: diskData.Used, - Free: diskData.Free, - UsagePercent: diskData.UsagePercent, - Timestamp: now, - } - if err := s.metricRepo.SaveDiskMetric(ctx, metric); err != nil { - s.logger.Error("failed to save disk metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.String("mountPoint", diskData.MountPoint)) - } - } - return nil - - case protocol.MetricTypeNetwork: - // Network现在是数组,需要批量处理 - var networkDataList []protocol.NetworkData - if err := json.Unmarshal(data, &networkDataList); err != nil { - return err - } - // 保存每个网卡的数据 - for _, netData := range networkDataList { - metric := &models.NetworkMetric{ - AgentID: agentID, - Interface: netData.Interface, - BytesSentRate: netData.BytesSentRate, - BytesRecvRate: netData.BytesRecvRate, - BytesSentTotal: netData.BytesSentTotal, - BytesRecvTotal: netData.BytesRecvTotal, - Timestamp: now, - } - if err := s.metricRepo.SaveNetworkMetric(ctx, metric); err != nil { - s.logger.Error("failed to save network metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.String("interface", netData.Interface)) - } - } - return nil - - case protocol.MetricTypeNetworkConnection: - var connData protocol.NetworkConnectionData - if err := json.Unmarshal(data, &connData); err != nil { - return err - } - metric := &models.NetworkConnectionMetric{ - AgentID: agentID, - Established: connData.Established, - SynSent: connData.SynSent, - SynRecv: connData.SynRecv, - FinWait1: connData.FinWait1, - FinWait2: connData.FinWait2, - TimeWait: connData.TimeWait, - Close: connData.Close, - CloseWait: connData.CloseWait, - LastAck: connData.LastAck, - Listen: connData.Listen, - Closing: connData.Closing, - Total: connData.Total, - Timestamp: now, - } - return s.metricRepo.SaveNetworkConnectionMetric(ctx, metric) - - case protocol.MetricTypeDiskIO: - // DiskIO现在是数组,需要批量处理 - var diskIODataList []*protocol.DiskIOData - if err := json.Unmarshal(data, &diskIODataList); err != nil { - return err - } - // 保存每个磁盘的IO数据 - for _, diskIOData := range diskIODataList { - metric := &models.DiskIOMetric{ - AgentID: agentID, - Device: diskIOData.Device, - ReadCount: diskIOData.ReadCount, - WriteCount: diskIOData.WriteCount, - ReadBytes: diskIOData.ReadBytes, - WriteBytes: diskIOData.WriteBytes, - ReadBytesRate: diskIOData.ReadBytesRate, - WriteBytesRate: diskIOData.WriteBytesRate, - ReadTime: diskIOData.ReadTime, - WriteTime: diskIOData.WriteTime, - IoTime: diskIOData.IoTime, - IopsInProgress: diskIOData.IopsInProgress, - Timestamp: now, - } - if err := s.metricRepo.SaveDiskIOMetric(ctx, metric); err != nil { - s.logger.Error("failed to save disk io metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.String("device", diskIOData.Device)) - } - } - return nil - - case protocol.MetricTypeHost: - var hostData protocol.HostInfoData - if err := json.Unmarshal(data, &hostData); err != nil { - return err - } - // 保存主机信息 - metric := &models.HostMetric{ - AgentID: agentID, - OS: hostData.OS, - Platform: hostData.Platform, - PlatformVersion: hostData.PlatformVersion, - KernelVersion: hostData.KernelVersion, - KernelArch: hostData.KernelArch, - Uptime: hostData.Uptime, - BootTime: hostData.BootTime, - Procs: hostData.Procs, - Timestamp: now, - } - return s.metricRepo.SaveHostMetric(ctx, metric) - - case protocol.MetricTypeGPU: - // GPU现在是数组,需要批量处理 - var gpuDataList []protocol.GPUData - if err := json.Unmarshal(data, &gpuDataList); err != nil { - return err - } - // 保存每个GPU的数据 - for _, gpuData := range gpuDataList { - metric := &models.GPUMetric{ - AgentID: agentID, - Index: gpuData.Index, - Name: gpuData.Name, - Utilization: gpuData.Utilization, - MemoryTotal: gpuData.MemoryTotal, - MemoryUsed: gpuData.MemoryUsed, - MemoryFree: gpuData.MemoryFree, - Temperature: gpuData.Temperature, - PowerDraw: gpuData.PowerUsage, // protocol 中是 PowerUsage - FanSpeed: gpuData.FanSpeed, - PerformanceState: "", // protocol 中没有这个字段，留空 - Timestamp: now, - } - if err := s.metricRepo.SaveGPUMetric(ctx, metric); err != nil { - s.logger.Error("failed to save gpu metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.Int("index", gpuData.Index)) - } - } - return nil - - case protocol.MetricTypeTemperature: - // Temperature现在是数组,需要批量处理 - var tempDataList []protocol.TemperatureData - if err := json.Unmarshal(data, &tempDataList); err != nil { - return err - } - // 保存每个温度传感器的数据 - for _, tempData := range tempDataList { - metric := &models.TemperatureMetric{ - AgentID: agentID, - SensorKey: tempData.SensorKey, - SensorLabel: tempData.SensorKey, // protocol 中没有 SensorLabel，使用 SensorKey - Temperature: tempData.Temperature, - Timestamp: now, - } - if err := s.metricRepo.SaveTemperatureMetric(ctx, metric); err != nil { - s.logger.Error("failed to save temperature metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.String("sensor", tempData.SensorKey)) - } - } - return nil - - case protocol.MetricTypeMonitor: - // 监控数据也是数组,需要批量处理 - var monitorDataList []protocol.MonitorData - if err := json.Unmarshal(data, &monitorDataList); err != nil { - return err - } - // 保存每个监控项的数据 - for _, monitorData := range monitorDataList { - metric := &models.MonitorMetric{ - AgentId: agentID, - MonitorId: monitorData.ID, - Type: monitorData.Type, - Target: monitorData.Target, - Status: monitorData.Status, - StatusCode: monitorData.StatusCode, - ResponseTime: monitorData.ResponseTime, - Error: monitorData.Error, - Message: monitorData.Message, - ContentMatch: monitorData.ContentMatch, - CertExpiryTime: monitorData.CertExpiryTime, - CertDaysLeft: monitorData.CertDaysLeft, - Timestamp: monitorData.CheckedAt, // 使用检测时间 - } - if err := s.metricRepo.SaveMonitorMetric(ctx, metric); err != nil { - s.logger.Error("failed to save monitor metric", - zap.Error(err), - zap.String("agentID", agentID), - zap.String("MonitorId", monitorData.ID)) - } - } - return nil - - default: - s.logger.Warn("unknown metric type", zap.String("type", metricType)) - return nil - } -} - -// CalculateInterval 根据时间范围计算合适的聚合间隔（秒） -// 目标是返回尽量平滑的曲线，同时控制数据点数量 -func CalculateInterval(start, end int64) int { - duration := (end - start) / 1000 // 转换为秒 - - switch { - case duration <= 60: // 1分钟内: 2秒 - return 2 - case duration <= 300: // 5分钟内: 5秒 - return 5 - case duration <= 900: // 15分钟内: 15秒 - return 15 - case duration <= 1800: // 30分钟内: 30秒 - return 30 - default: // 1小时内: 60秒 - return 60 - } + return s.metricService.HandleMetricData(ctx, agentID, metricType, data) } // GetMetrics 获取聚合指标数据（自动路由到聚合表或原始表） -func (s *AgentService) GetMetrics(ctx context.Context, agentID, metricType string, start, end int64, interval int) (interface{}, error) { - start, end = s.normalizeTimeRange(ctx, start, end) - interval = s.DetermineInterval(ctx, start, end, interval) - - // 判断是否可以使用聚合表（仅支持部分指标类型） - aggCapable := map[string]bool{ - "cpu": true, - "memory": true, - "disk": true, - "network": true, - "network_connection": true, - "disk_io": true, - "gpu": true, - "temperature": true, - } - - // 智能选择聚合粒度：根据查询间隔选择最合适的bucket - // 例如：查询90秒数据时使用60秒bucket，查询600秒数据时使用300秒bucket - var bucketSeconds int - useAgg := false - if aggCapable[metricType] { - if interval >= 3600 { - bucketSeconds = 3600 - useAgg = true - } else if interval >= 300 { - bucketSeconds = 300 - useAgg = true - } else if interval >= 60 { - bucketSeconds = 60 - useAgg = true - } - } - - switch metricType { - case "cpu": - if useAgg { - if metrics, err := s.metricRepo.GetCPUMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetCPUMetrics(ctx, agentID, start, end, interval) - case "memory": - if useAgg { - if metrics, err := s.metricRepo.GetMemoryMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetMemoryMetrics(ctx, agentID, start, end, interval) - case "disk": - if useAgg { - if metrics, err := s.metricRepo.GetDiskMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetDiskMetrics(ctx, agentID, start, end, interval) - case "network": - if useAgg { - if metrics, err := s.metricRepo.GetNetworkMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetNetworkMetrics(ctx, agentID, start, end, interval) - case "network_connection": - if useAgg { - if metrics, err := s.metricRepo.GetNetworkConnectionMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetNetworkConnectionMetrics(ctx, agentID, start, end, interval) - case "disk_io": - if useAgg { - if metrics, err := s.metricRepo.GetDiskIOMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetDiskIOMetrics(ctx, agentID, start, end, interval) - case "gpu": - if useAgg { - if metrics, err := s.metricRepo.GetGPUMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetGPUMetrics(ctx, agentID, start, end, interval) - case "temperature": - if useAgg { - if metrics, err := s.metricRepo.GetTemperatureMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { - return metrics, nil - } - } - return s.metricRepo.GetTemperatureMetrics(ctx, agentID, start, end, interval) - default: - return nil, nil - } +func (s *AgentService) GetMetrics(ctx context.Context, agentID, metricType string, start, end int64, interval int, interfaceName string) (interface{}, error) { + return s.metricService.GetMetrics(ctx, agentID, metricType, start, end, interval, interfaceName) } -// DetermineInterval 根据配置、用户请求和时间范围决定聚合粒度 -func (s *AgentService) DetermineInterval(ctx context.Context, start, end int64, requested int) int { - cfg := s.getMetricsConfig(ctx) - - interval := requested - if interval <= 0 { - interval = calculateBaseInterval(start, end) - } - - interval = adjustIntervalForMaxPoints(start, end, interval, cfg.MaxQueryPoints) - return interval -} - -// normalizeTimeRange 将时间范围限制在保留周期内，避免无意义的全表扫描 -func (s *AgentService) normalizeTimeRange(ctx context.Context, start, end int64) (int64, int64) { - cfg := s.getMetricsConfig(ctx) - retentionDuration := time.Duration(cfg.RetentionHours) * time.Hour - - retentionBoundary := time.Now().Add(-retentionDuration).UnixMilli() - if start < retentionBoundary { - start = retentionBoundary - } - if end <= start { - end = start + 1000 - } - return start, end -} - -func calculateBaseInterval(start, end int64) int { - duration := (end - start) / 1000 // 秒 - - switch { - case duration <= 60: // 1 分钟内 - return 2 - case duration <= 5*60: // 5 分钟内 - return 5 - case duration <= 15*60: // 15 分钟内 - return 15 - case duration <= 30*60: // 30 分钟内 - return 30 - case duration <= 60*60: // 1 小时内 - return 60 - case duration <= 3*60*60: // 3 小时内 - return 180 - case duration <= 6*60*60: // 6 小时内 - return 300 - case duration <= 12*60*60: // 12 小时内 - return 600 - case duration <= 24*60*60: // 1 天内 - return 900 - case duration <= 3*24*60*60: // 3 天内 - return 1800 - case duration <= 7*24*60*60: // 7 天内 - return 3600 - case duration <= 14*24*60*60: // 14 天内 - return 7200 - default: - return 14400 // 更长时间：4 小时粒度 - } -} - -// adjustIntervalForMaxPoints 根据最大数据点限制提升聚合粒度 -func adjustIntervalForMaxPoints(start, end int64, interval int, maxPoints int) int { - if interval <= 0 { - interval = 1 - } - if maxPoints <= 0 { - return alignInterval(interval) - } - - durationSeconds := float64(end-start) / 1000 - if durationSeconds <= 0 { - return alignInterval(interval) - } - - required := int(math.Ceil(durationSeconds / float64(maxPoints))) - interval = maxInt(interval, required) - return alignInterval(interval) -} - -func alignInterval(interval int) int { - for _, candidate := range allowedIntervals { - if interval <= candidate { - return candidate - } - } - return allowedIntervals[len(allowedIntervals)-1] -} - -func maxInt(a, b int) int { - if a > b { - return a - } - return b -} - -func (s *AgentService) getMetricsConfig(ctx context.Context) models.MetricsConfig { - cfg := models.MetricsConfig{ - RetentionHours: defaultMetricsRetentionHours, - MaxQueryPoints: defaultMaxQueryPoints, - } - - if s.propertyService == nil { - return cfg - } - - loaded := s.propertyService.GetMetricsConfig(ctx) - if loaded.RetentionHours > 0 { - cfg.RetentionHours = loaded.RetentionHours - } - if loaded.MaxQueryPoints > 0 { - cfg.MaxQueryPoints = loaded.MaxQueryPoints - } - return cfg -} - -// runAggregation 按固定 bucket 下采样存储 -func (s *AgentService) runAggregation(ctx context.Context) { - cfg := s.getMetricsConfig(ctx) - retention := time.Duration(cfg.RetentionHours) * time.Hour - // 标准 bucket：1m、5m、1h - buckets := []int{60, 300, 3600} - - for _, bucket := range buckets { - s.aggregateMetric(ctx, "cpu", bucket, retention, s.metricRepo.AggregateCPUToAgg) - s.aggregateMetric(ctx, "memory", bucket, retention, s.metricRepo.AggregateMemoryToAgg) - s.aggregateMetric(ctx, "disk", bucket, retention, s.metricRepo.AggregateDiskToAgg) - s.aggregateMetric(ctx, "network", bucket, retention, s.metricRepo.AggregateNetworkToAgg) - s.aggregateMetric(ctx, "network_connection", bucket, retention, s.metricRepo.AggregateNetworkConnectionToAgg) - s.aggregateMetric(ctx, "disk_io", bucket, retention, s.metricRepo.AggregateDiskIOToAgg) - s.aggregateMetric(ctx, "gpu", bucket, retention, s.metricRepo.AggregateGPUToAgg) - s.aggregateMetric(ctx, "temperature", bucket, retention, s.metricRepo.AggregateTemperatureToAgg) - } -} - -type aggregateFn func(ctx context.Context, bucketSeconds int, start, end int64) error - -func (s *AgentService) aggregateMetric(ctx context.Context, metricType string, bucketSeconds int, retention time.Duration, fn aggregateFn) { - bucketMs := int64(bucketSeconds * 1000) - - start := s.getAggregationStart(ctx, metricType, bucketSeconds, retention, bucketMs) - endBucket := (time.Now().Add(-time.Duration(bucketSeconds)*time.Second).UnixMilli() / bucketMs) * bucketMs - - if endBucket <= start { - return - } - - end := endBucket + bucketMs - 1 - - if err := fn(ctx, bucketSeconds, start, end); err != nil { - s.logger.Error("aggregate metric failed", zap.String("metricType", metricType), zap.Int("bucketSeconds", bucketSeconds), zap.Error(err)) - return - } - - if err := s.metricRepo.UpsertAggregationProgress(ctx, metricType, bucketSeconds, endBucket); err != nil { - s.logger.Error("update aggregation progress failed", zap.String("metricType", metricType), zap.Int("bucketSeconds", bucketSeconds), zap.Error(err)) - } -} - -func (s *AgentService) getAggregationStart(ctx context.Context, metricType string, bucketSeconds int, retention time.Duration, bucketMs int64) int64 { - progress, err := s.metricRepo.GetAggregationProgress(ctx, metricType, bucketSeconds) - if err == nil && progress != nil && progress.LastBucket > 0 { - return progress.LastBucket + bucketMs - } - - // 默认从保留窗口开始 - start := time.Now().Add(-retention).UnixMilli() - return (start / bucketMs) * bucketMs +// GetAvailableNetworkInterfaces 获取探针的可用网卡列表 +func (s *AgentService) GetAvailableNetworkInterfaces(ctx context.Context, agentID string) ([]string, error) { + return s.metricService.GetAvailableNetworkInterfaces(ctx, agentID) } // GetLatestMetrics 获取最新指标 func (s *AgentService) GetLatestMetrics(ctx context.Context, agentID string) (*LatestMetrics, error) { - result := &LatestMetrics{} - - // 获取最新CPU指标 - if cpu, err := s.metricRepo.GetLatestCPUMetric(ctx, agentID); err == nil { - result.CPU = cpu - } - - // 获取最新内存指标 - if memory, err := s.metricRepo.GetLatestMemoryMetric(ctx, agentID); err == nil { - result.Memory = memory - } - - // 获取最新磁盘指标并计算平均使用率和总容量 - if disks, err := s.metricRepo.GetLatestDiskMetrics(ctx, agentID); err == nil && len(disks) > 0 { - var totalUsage float64 - var totalSpace, usedSpace, freeSpace uint64 - for _, disk := range disks { - totalUsage += disk.UsagePercent - totalSpace += disk.Total - usedSpace += disk.Used - freeSpace += disk.Free - } - result.Disk = &DiskSummary{ - AvgUsagePercent: totalUsage / float64(len(disks)), - TotalDisks: len(disks), - Total: totalSpace, - Used: usedSpace, - Free: freeSpace, - } - } - - // 获取最新网络指标并汇总速率和累计流量 - // 注意: 采集器已经计算好了每秒速率,这里直接汇总所有网卡的速率和累计流量 - if networks, err := s.metricRepo.GetLatestNetworkMetrics(ctx, agentID); err == nil && len(networks) > 0 { - var totalSentRate, totalRecvRate, totalSentTotal, totalRecvTotal uint64 - for _, net := range networks { - totalSentRate += net.BytesSentRate // 累加每个网卡的发送速率 - totalRecvRate += net.BytesRecvRate // 累加每个网卡的接收速率 - totalSentTotal += net.BytesSentTotal // 累加每个网卡的累计发送流量 - totalRecvTotal += net.BytesRecvTotal // 累加每个网卡的累计接收流量 - } - result.Network = &NetworkSummary{ - TotalBytesSentRate: totalSentRate, // 所有网卡的总发送速率(字节/秒) - TotalBytesRecvRate: totalRecvRate, // 所有网卡的总接收速率(字节/秒) - TotalBytesSentTotal: totalSentTotal, // 所有网卡的累计发送流量 - TotalBytesRecvTotal: totalRecvTotal, // 所有网卡的累计接收流量 - TotalInterfaces: len(networks), - } - } - - // 获取最新主机信息 - if host, err := s.metricRepo.GetLatestHostMetric(ctx, agentID); err == nil { - result.Host = host - } - - // 获取最新GPU信息 - if gpu, err := s.metricRepo.GetLatestGPUMetrics(ctx, agentID); err == nil && len(gpu) > 0 { - result.GPU = gpu - } - - // 获取最新温度信息 - if temp, err := s.metricRepo.GetLatestTemperatureMetrics(ctx, agentID); err == nil && len(temp) > 0 { - result.Temp = temp - } - - // 获取最新网络连接统计 - if netConn, err := s.metricRepo.GetLatestNetworkConnectionMetric(ctx, agentID); err == nil { - result.NetworkConnection = netConn - } - - return result, nil + return s.metricService.GetLatestMetrics(ctx, agentID) } // StartCleanupTask 启动数据清理任务 func (s *AgentService) StartCleanupTask(ctx context.Context) { - ticker := time.NewTicker(1 * time.Minute) - defer ticker.Stop() - - s.logger.Info("cleanup task started") - - for { - select { - case <-ctx.Done(): - s.logger.Info("cleanup task stopped") - return - case <-ticker.C: - s.cleanupOldMetrics(ctx) - } - } -} - -// cleanupOldMetrics 清理旧数据 -func (s *AgentService) cleanupOldMetrics(ctx context.Context) { - cfg := s.getMetricsConfig(ctx) - retentionDuration := time.Duration(cfg.RetentionHours) * time.Hour - before := time.Now().Add(-retentionDuration).UnixMilli() - - s.logger.Info("starting to clean old metrics", zap.Int64("beforeTimestamp", before), zap.Int("retentionHours", cfg.RetentionHours)) - - if err := s.metricRepo.DeleteOldMetrics(ctx, before); err != nil { - s.logger.Error("failed to clean old metrics", zap.Error(err)) - return - } - - s.logger.Info("old metrics cleaned successfully") + s.metricService.StartCleanupTask(ctx) } // StartAggregationTask 启动聚合下采样任务 func (s *AgentService) StartAggregationTask(ctx context.Context) { - ticker := time.NewTicker(1 * time.Minute) - defer ticker.Stop() - - s.logger.Info("aggregation task started") - - for { - select { - case <-ctx.Done(): - s.logger.Info("aggregation task stopped") - return - case <-ticker.C: - s.runAggregation(ctx) - } - } -} - -// DiskSummary 磁盘汇总数据 -type DiskSummary struct { - AvgUsagePercent float64 `json:"avgUsagePercent"` // 平均使用率 - TotalDisks int `json:"totalDisks"` // 磁盘数量 - Total uint64 `json:"total"` // 总容量(字节) - Used uint64 `json:"used"` // 已使用(字节) - Free uint64 `json:"free"` // 空闲(字节) -} - -// NetworkSummary 网络汇总数据 -type NetworkSummary struct { - TotalBytesSentRate uint64 `json:"totalBytesSentRate"` // 总发送速率(字节/秒) - TotalBytesRecvRate uint64 `json:"totalBytesRecvRate"` // 总接收速率(字节/秒) - TotalBytesSentTotal uint64 `json:"totalBytesSentTotal"` // 累计总发送流量 - TotalBytesRecvTotal uint64 `json:"totalBytesRecvTotal"` // 累计总接收流量 - TotalInterfaces int `json:"totalInterfaces"` // 网卡数量 -} - -// LatestMetrics 最新指标数据（用于API响应） -type LatestMetrics struct { - CPU *models.CPUMetric `json:"cpu,omitempty"` - Memory *models.MemoryMetric `json:"memory,omitempty"` - Disk *DiskSummary `json:"disk,omitempty"` - Network *NetworkSummary `json:"network,omitempty"` - NetworkConnection *models.NetworkConnectionMetric `json:"networkConnection,omitempty"` - Host *models.HostMetric `json:"host,omitempty"` - GPU []models.GPUMetric `json:"gpu,omitempty"` - Temp []models.TemperatureMetric `json:"temperature,omitempty"` + s.metricService.StartAggregationTask(ctx) } // HandleCommandResponse 处理指令响应 @@ -1001,17 +313,17 @@ func (s *AgentService) GetStatistics(ctx context.Context) (map[string]interface{ // GetLatestMonitorMetrics 获取最新的监控指标 func (s *AgentService) GetLatestMonitorMetrics(ctx context.Context, agentID string) ([]models.MonitorMetric, error) { - return s.metricRepo.GetLatestMonitorMetrics(ctx, agentID) + return s.metricService.GetLatestMonitorMetrics(ctx, agentID) } // GetMonitorMetrics 获取监控指标历史数据 func (s *AgentService) GetMonitorMetrics(ctx context.Context, agentID, monitorName string, start, end int64) ([]models.MonitorMetric, error) { - return s.metricRepo.GetMonitorMetrics(ctx, agentID, monitorName, start, end) + return s.metricService.GetMonitorMetrics(ctx, agentID, monitorName, start, end) } // GetMonitorMetricsByName 获取指定监控项的历史数据 func (s *AgentService) GetMonitorMetricsByName(ctx context.Context, agentID, monitorName string, start, end int64, limit int) ([]models.MonitorMetric, error) { - return s.metricRepo.GetMonitorMetricsByName(ctx, agentID, monitorName, start, end, limit) + return s.metricService.GetMonitorMetricsByName(ctx, agentID, monitorName, start, end, limit) } // DeleteAgent 删除探针及其所有相关数据 @@ -1019,7 +331,7 @@ func (s *AgentService) DeleteAgent(ctx context.Context, agentID string) error { // 在事务中执行所有删除操作 return s.Transaction(ctx, func(ctx context.Context) error { // 1. 删除探针的所有指标数据 - if err := s.metricRepo.DeleteAgentMetrics(ctx, agentID); err != nil { + if err := s.metricService.DeleteAgentMetrics(ctx, agentID); err != nil { s.logger.Error("删除探针指标数据失败", zap.String("agentId", agentID), zap.Error(err)) return err } diff --git a/internal/service/metric_service.go b/internal/service/metric_service.go new file mode 100644 index 0000000..d46387b --- /dev/null +++ b/internal/service/metric_service.go @@ -0,0 +1,798 @@ +package service + +import ( + "context" + "encoding/json" + "math" + "time" + + "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/protocol" + "github.com/dushixiang/pika/internal/repo" + "go.uber.org/zap" + "gorm.io/gorm" +) + +const ( + defaultMetricsRetentionHours = 24 * 7 // 默认保留 7 天 + defaultMaxQueryPoints = 720 // 默认最多返回 720 个点 +) + +var allowedIntervals = []int{ + 1, 2, 5, + 10, 15, 30, + 60, 120, 300, + 600, 900, 1800, + 3600, 7200, 14400, +} + +// MetricService 指标服务 +type MetricService struct { + logger *zap.Logger + metricRepo *repo.MetricRepo + monitorStatsRepo *repo.MonitorStatsRepo + propertyService *PropertyService +} + +// NewMetricService 创建指标服务 +func NewMetricService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService) *MetricService { + return &MetricService{ + logger: logger, + metricRepo: repo.NewMetricRepo(db), + monitorStatsRepo: repo.NewMonitorStatsRepo(db), + propertyService: propertyService, + } +} + +// HandleMetricData 处理指标数据 +func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, metricType string, data json.RawMessage) error { + now := time.Now().UnixMilli() + + switch protocol.MetricType(metricType) { + case protocol.MetricTypeCPU: + // CPU数据现在包含静态和动态信息 + var cpuData protocol.CPUData + if err := json.Unmarshal(data, &cpuData); err != nil { + return err + } + metric := &models.CPUMetric{ + AgentID: agentID, + UsagePercent: cpuData.UsagePercent, + LogicalCores: cpuData.LogicalCores, + PhysicalCores: cpuData.PhysicalCores, + ModelName: cpuData.ModelName, + Timestamp: now, + } + return s.metricRepo.SaveCPUMetric(ctx, metric) + + case protocol.MetricTypeMemory: + // Memory数据现在包含静态和动态信息 + var memData protocol.MemoryData + if err := json.Unmarshal(data, &memData); err != nil { + return err + } + metric := &models.MemoryMetric{ + AgentID: agentID, + Total: memData.Total, // 现在可以从合并后的数据获取 + Used: memData.Used, + Free: memData.Free, + UsagePercent: memData.UsagePercent, + SwapTotal: memData.SwapTotal, + SwapUsed: memData.SwapUsed, + SwapFree: memData.SwapFree, + Timestamp: now, + } + return s.metricRepo.SaveMemoryMetric(ctx, metric) + + case protocol.MetricTypeDisk: + // Disk现在是数组,需要批量处理 + var diskDataList []protocol.DiskData + if err := json.Unmarshal(data, &diskDataList); err != nil { + return err + } + // 保存每个磁盘的数据 + for _, diskData := range diskDataList { + metric := &models.DiskMetric{ + AgentID: agentID, + MountPoint: diskData.MountPoint, + Total: diskData.Total, + Used: diskData.Used, + Free: diskData.Free, + UsagePercent: diskData.UsagePercent, + Timestamp: now, + } + if err := s.metricRepo.SaveDiskMetric(ctx, metric); err != nil { + s.logger.Error("failed to save disk metric", + zap.Error(err), + zap.String("agentID", agentID), + zap.String("mountPoint", diskData.MountPoint)) + } + } + return nil + + case protocol.MetricTypeNetwork: + // Network现在是数组,需要批量处理 + var networkDataList []protocol.NetworkData + if err := json.Unmarshal(data, &networkDataList); err != nil { + return err + } + + // 合并所有网卡的数据用于保存总和 + var totalSentRate, totalRecvRate uint64 + var totalSentTotal, totalRecvTotal uint64 + + // 保存每个网卡的数据，同时累加总和 + for _, netData := range networkDataList { + // 保存单个网卡数据 + metric := &models.NetworkMetric{ + AgentID: agentID, + Interface: netData.Interface, + BytesSentRate: netData.BytesSentRate, + BytesRecvRate: netData.BytesRecvRate, + BytesSentTotal: netData.BytesSentTotal, + BytesRecvTotal: netData.BytesRecvTotal, + Timestamp: now, + } + if err := s.metricRepo.SaveNetworkMetric(ctx, metric); err != nil { + s.logger.Error("failed to save network metric", + zap.Error(err), + zap.String("agentID", agentID), + zap.String("interface", netData.Interface)) + } + + // 累加所有网卡的数据 + totalSentRate += netData.BytesSentRate + totalRecvRate += netData.BytesRecvRate + totalSentTotal += netData.BytesSentTotal + totalRecvTotal += netData.BytesRecvTotal + } + + // 保存合并后的总和数据（interface 字段设置为空字符串） + totalMetric := &models.NetworkMetric{ + AgentID: agentID, + Interface: "", // 空字符串表示所有网卡的合并数据 + BytesSentRate: totalSentRate, + BytesRecvRate: totalRecvRate, + BytesSentTotal: totalSentTotal, + BytesRecvTotal: totalRecvTotal, + Timestamp: now, + } + return s.metricRepo.SaveNetworkMetric(ctx, totalMetric) + + case protocol.MetricTypeNetworkConnection: + var connData protocol.NetworkConnectionData + if err := json.Unmarshal(data, &connData); err != nil { + return err + } + metric := &models.NetworkConnectionMetric{ + AgentID: agentID, + Established: connData.Established, + SynSent: connData.SynSent, + SynRecv: connData.SynRecv, + FinWait1: connData.FinWait1, + FinWait2: connData.FinWait2, + TimeWait: connData.TimeWait, + Close: connData.Close, + CloseWait: connData.CloseWait, + LastAck: connData.LastAck, + Listen: connData.Listen, + Closing: connData.Closing, + Total: connData.Total, + Timestamp: now, + } + return s.metricRepo.SaveNetworkConnectionMetric(ctx, metric) + + case protocol.MetricTypeDiskIO: + // DiskIO现在是数组，直接合并所有磁盘的数据存储为一条记录 + var diskIODataList []*protocol.DiskIOData + if err := json.Unmarshal(data, &diskIODataList); err != nil { + return err + } + + // 合并所有磁盘的数据 + var totalReadCount, totalWriteCount uint64 + var totalReadBytes, totalWriteBytes uint64 + var totalReadBytesRate, totalWriteBytesRate uint64 + var totalReadTime, totalWriteTime, totalIoTime uint64 + var maxIopsInProgress uint64 + + for _, diskIOData := range diskIODataList { + totalReadCount += diskIOData.ReadCount + totalWriteCount += diskIOData.WriteCount + totalReadBytes += diskIOData.ReadBytes + totalWriteBytes += diskIOData.WriteBytes + totalReadBytesRate += diskIOData.ReadBytesRate + totalWriteBytesRate += diskIOData.WriteBytesRate + totalReadTime += diskIOData.ReadTime + totalWriteTime += diskIOData.WriteTime + totalIoTime += diskIOData.IoTime + if diskIOData.IopsInProgress > maxIopsInProgress { + maxIopsInProgress = diskIOData.IopsInProgress + } + } + + // 保存合并后的数据（device 字段设置为空或 "all"） + metric := &models.DiskIOMetric{ + AgentID: agentID, + Device: "", // 空字符串表示所有磁盘的合并数据 + ReadCount: totalReadCount, + WriteCount: totalWriteCount, + ReadBytes: totalReadBytes, + WriteBytes: totalWriteBytes, + ReadBytesRate: totalReadBytesRate, + WriteBytesRate: totalWriteBytesRate, + ReadTime: totalReadTime, + WriteTime: totalWriteTime, + IoTime: totalIoTime, + IopsInProgress: maxIopsInProgress, + Timestamp: now, + } + return s.metricRepo.SaveDiskIOMetric(ctx, metric) + + case protocol.MetricTypeHost: + var hostData protocol.HostInfoData + if err := json.Unmarshal(data, &hostData); err != nil { + return err + } + // 保存主机信息 + metric := &models.HostMetric{ + AgentID: agentID, + OS: hostData.OS, + Platform: hostData.Platform, + PlatformVersion: hostData.PlatformVersion, + KernelVersion: hostData.KernelVersion, + KernelArch: hostData.KernelArch, + Uptime: hostData.Uptime, + BootTime: hostData.BootTime, + Procs: hostData.Procs, + Timestamp: now, + } + return s.metricRepo.SaveHostMetric(ctx, metric) + + case protocol.MetricTypeGPU: + // GPU现在是数组,需要批量处理 + var gpuDataList []protocol.GPUData + if err := json.Unmarshal(data, &gpuDataList); err != nil { + return err + } + // 保存每个GPU的数据 + for _, gpuData := range gpuDataList { + metric := &models.GPUMetric{ + AgentID: agentID, + Index: gpuData.Index, + Name: gpuData.Name, + Utilization: gpuData.Utilization, + MemoryTotal: gpuData.MemoryTotal, + MemoryUsed: gpuData.MemoryUsed, + MemoryFree: gpuData.MemoryFree, + Temperature: gpuData.Temperature, + PowerDraw: gpuData.PowerUsage, // protocol 中是 PowerUsage + FanSpeed: gpuData.FanSpeed, + PerformanceState: "", // protocol 中没有这个字段，留空 + Timestamp: now, + } + if err := s.metricRepo.SaveGPUMetric(ctx, metric); err != nil { + s.logger.Error("failed to save gpu metric", + zap.Error(err), + zap.String("agentID", agentID), + zap.Int("index", gpuData.Index)) + } + } + return nil + + case protocol.MetricTypeTemperature: + // Temperature现在是数组,需要批量处理 + var tempDataList []protocol.TemperatureData + if err := json.Unmarshal(data, &tempDataList); err != nil { + return err + } + // 保存每个温度传感器的数据 + for _, tempData := range tempDataList { + metric := &models.TemperatureMetric{ + AgentID: agentID, + SensorKey: tempData.SensorKey, + SensorLabel: tempData.SensorKey, // protocol 中没有 SensorLabel，使用 SensorKey + Temperature: tempData.Temperature, + Timestamp: now, + } + if err := s.metricRepo.SaveTemperatureMetric(ctx, metric); err != nil { + s.logger.Error("failed to save temperature metric", + zap.Error(err), + zap.String("agentID", agentID), + zap.String("sensor", tempData.SensorKey)) + } + } + return nil + + case protocol.MetricTypeMonitor: + // 监控数据也是数组,需要批量处理 + var monitorDataList []protocol.MonitorData + if err := json.Unmarshal(data, &monitorDataList); err != nil { + return err + } + // 保存每个监控项的数据 + for _, monitorData := range monitorDataList { + metric := &models.MonitorMetric{ + AgentId: agentID, + MonitorId: monitorData.ID, + Type: monitorData.Type, + Target: monitorData.Target, + Status: monitorData.Status, + StatusCode: monitorData.StatusCode, + ResponseTime: monitorData.ResponseTime, + Error: monitorData.Error, + Message: monitorData.Message, + ContentMatch: monitorData.ContentMatch, + CertExpiryTime: monitorData.CertExpiryTime, + CertDaysLeft: monitorData.CertDaysLeft, + Timestamp: monitorData.CheckedAt, // 使用检测时间 + } + if err := s.metricRepo.SaveMonitorMetric(ctx, metric); err != nil { + s.logger.Error("failed to save monitor metric", + zap.Error(err), + zap.String("agentID", agentID), + zap.String("MonitorId", monitorData.ID)) + } + } + return nil + + default: + s.logger.Warn("unknown metric type", zap.String("type", metricType)) + return nil + } +} + +// GetMetrics 获取聚合指标数据（自动路由到聚合表或原始表） +// interfaceName: 网卡过滤参数（仅对 network 类型有效） +func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType string, start, end int64, interval int, interfaceName string) (interface{}, error) { + start, end = s.normalizeTimeRange(ctx, start, end) + interval = s.DetermineInterval(ctx, start, end, interval) + + // 判断是否可以使用聚合表（仅支持部分指标类型） + aggCapable := map[string]bool{ + "cpu": true, + "memory": true, + "disk": true, + "network": true, + "network_connection": true, + "disk_io": true, + "gpu": true, + "temperature": true, + } + + // 智能选择聚合粒度：根据查询间隔选择最合适的bucket + // 例如：查询90秒数据时使用60秒bucket，查询600秒数据时使用300秒bucket + var bucketSeconds int + useAgg := false + if aggCapable[metricType] { + if interval >= 3600 { + bucketSeconds = 3600 + useAgg = true + } else if interval >= 300 { + bucketSeconds = 300 + useAgg = true + } else if interval >= 60 { + bucketSeconds = 60 + useAgg = true + } + } + + switch metricType { + case "cpu": + if useAgg { + if metrics, err := s.metricRepo.GetCPUMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetCPUMetrics(ctx, agentID, start, end, interval) + case "memory": + if useAgg { + if metrics, err := s.metricRepo.GetMemoryMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetMemoryMetrics(ctx, agentID, start, end, interval) + case "disk": + if useAgg { + if metrics, err := s.metricRepo.GetDiskMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetDiskMetrics(ctx, agentID, start, end, interval) + case "network": + if useAgg { + if metrics, err := s.metricRepo.GetNetworkMetricsAgg(ctx, agentID, start, end, bucketSeconds, interfaceName); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetNetworkMetrics(ctx, agentID, start, end, interval, interfaceName) + case "network_connection": + if useAgg { + if metrics, err := s.metricRepo.GetNetworkConnectionMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetNetworkConnectionMetrics(ctx, agentID, start, end, interval) + case "disk_io": + if useAgg { + if metrics, err := s.metricRepo.GetDiskIOMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetDiskIOMetrics(ctx, agentID, start, end, interval) + case "gpu": + if useAgg { + if metrics, err := s.metricRepo.GetGPUMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetGPUMetrics(ctx, agentID, start, end, interval) + case "temperature": + if useAgg { + if metrics, err := s.metricRepo.GetTemperatureMetricsAgg(ctx, agentID, start, end, bucketSeconds); err == nil && len(metrics) > 0 { + return metrics, nil + } + } + return s.metricRepo.GetTemperatureMetrics(ctx, agentID, start, end, interval) + default: + return nil, nil + } +} + +// DetermineInterval 根据配置、用户请求和时间范围决定聚合粒度 +func (s *MetricService) DetermineInterval(ctx context.Context, start, end int64, requested int) int { + cfg := s.getMetricsConfig(ctx) + + interval := requested + if interval <= 0 { + interval = calculateBaseInterval(start, end) + } + + interval = adjustIntervalForMaxPoints(start, end, interval, cfg.MaxQueryPoints) + return interval +} + +// normalizeTimeRange 将时间范围限制在保留周期内，避免无意义的全表扫描 +func (s *MetricService) normalizeTimeRange(ctx context.Context, start, end int64) (int64, int64) { + cfg := s.getMetricsConfig(ctx) + retentionDuration := time.Duration(cfg.RetentionHours) * time.Hour + + retentionBoundary := time.Now().Add(-retentionDuration).UnixMilli() + if start < retentionBoundary { + start = retentionBoundary + } + if end <= start { + end = start + 1000 + } + return start, end +} + +// calculateBaseInterval 根据时间范围计算基础间隔 +func calculateBaseInterval(start, end int64) int { + duration := (end - start) / 1000 // 秒 + + switch { + case duration <= 60: // 1 分钟内 + return 2 + case duration <= 5*60: // 5 分钟内 + return 5 + case duration <= 15*60: // 15 分钟内 + return 15 + case duration <= 30*60: // 30 分钟内 + return 30 + case duration <= 60*60: // 1 小时内 + return 60 + case duration <= 3*60*60: // 3 小时内 + return 180 + case duration <= 6*60*60: // 6 小时内 + return 300 + case duration <= 12*60*60: // 12 小时内 + return 600 + case duration <= 24*60*60: // 1 天内 + return 900 + case duration <= 3*24*60*60: // 3 天内 + return 1800 + case duration <= 7*24*60*60: // 7 天内 + return 3600 + case duration <= 14*24*60*60: // 14 天内 + return 7200 + default: + return 14400 // 更长时间：4 小时粒度 + } +} + +// adjustIntervalForMaxPoints 根据最大数据点限制提升聚合粒度 +func adjustIntervalForMaxPoints(start, end int64, interval int, maxPoints int) int { + if interval <= 0 { + interval = 1 + } + if maxPoints <= 0 { + return alignInterval(interval) + } + + durationSeconds := float64(end-start) / 1000 + if durationSeconds <= 0 { + return alignInterval(interval) + } + + required := int(math.Ceil(durationSeconds / float64(maxPoints))) + interval = maxInt(interval, required) + return alignInterval(interval) +} + +// alignInterval 将间隔对齐到允许的值 +func alignInterval(interval int) int { + for _, candidate := range allowedIntervals { + if interval <= candidate { + return candidate + } + } + return allowedIntervals[len(allowedIntervals)-1] +} + +// maxInt 返回两个整数的最大值 +func maxInt(a, b int) int { + if a > b { + return a + } + return b +} + +// getMetricsConfig 获取指标配置 +func (s *MetricService) getMetricsConfig(ctx context.Context) models.MetricsConfig { + cfg := models.MetricsConfig{ + RetentionHours: defaultMetricsRetentionHours, + MaxQueryPoints: defaultMaxQueryPoints, + } + + if s.propertyService == nil { + return cfg + } + + loaded := s.propertyService.GetMetricsConfig(ctx) + if loaded.RetentionHours > 0 { + cfg.RetentionHours = loaded.RetentionHours + } + if loaded.MaxQueryPoints > 0 { + cfg.MaxQueryPoints = loaded.MaxQueryPoints + } + return cfg +} + +// StartAggregationTask 启动聚合下采样任务 +func (s *MetricService) StartAggregationTask(ctx context.Context) { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + + s.logger.Info("aggregation task started") + + for { + select { + case <-ctx.Done(): + s.logger.Info("aggregation task stopped") + return + case <-ticker.C: + s.runAggregation(ctx) + } + } +} + +// runAggregation 按固定 bucket 下采样存储 +func (s *MetricService) runAggregation(ctx context.Context) { + cfg := s.getMetricsConfig(ctx) + retention := time.Duration(cfg.RetentionHours) * time.Hour + // 标准 bucket：1m、5m、1h + buckets := []int{60, 300, 3600} + + for _, bucket := range buckets { + s.aggregateMetric(ctx, "cpu", bucket, retention, s.metricRepo.AggregateCPUToAgg) + s.aggregateMetric(ctx, "memory", bucket, retention, s.metricRepo.AggregateMemoryToAgg) + s.aggregateMetric(ctx, "disk", bucket, retention, s.metricRepo.AggregateDiskToAgg) + s.aggregateMetric(ctx, "network", bucket, retention, s.metricRepo.AggregateNetworkToAgg) + s.aggregateMetric(ctx, "network_connection", bucket, retention, s.metricRepo.AggregateNetworkConnectionToAgg) + s.aggregateMetric(ctx, "disk_io", bucket, retention, s.metricRepo.AggregateDiskIOToAgg) + s.aggregateMetric(ctx, "gpu", bucket, retention, s.metricRepo.AggregateGPUToAgg) + s.aggregateMetric(ctx, "temperature", bucket, retention, s.metricRepo.AggregateTemperatureToAgg) + } +} + +type aggregateFn func(ctx context.Context, bucketSeconds int, start, end int64) error + +// aggregateMetric 聚合指定类型的指标 +func (s *MetricService) aggregateMetric(ctx context.Context, metricType string, bucketSeconds int, retention time.Duration, fn aggregateFn) { + bucketMs := int64(bucketSeconds * 1000) + + start := s.getAggregationStart(ctx, metricType, bucketSeconds, retention, bucketMs) + endBucket := (time.Now().Add(-time.Duration(bucketSeconds)*time.Second).UnixMilli() / bucketMs) * bucketMs + + if endBucket <= start { + return + } + + end := endBucket + bucketMs - 1 + + if err := fn(ctx, bucketSeconds, start, end); err != nil { + s.logger.Error("aggregate metric failed", zap.String("metricType", metricType), zap.Int("bucketSeconds", bucketSeconds), zap.Error(err)) + return + } + + if err := s.metricRepo.UpsertAggregationProgress(ctx, metricType, bucketSeconds, endBucket); err != nil { + s.logger.Error("update aggregation progress failed", zap.String("metricType", metricType), zap.Int("bucketSeconds", bucketSeconds), zap.Error(err)) + } +} + +// getAggregationStart 获取聚合开始时间 +func (s *MetricService) getAggregationStart(ctx context.Context, metricType string, bucketSeconds int, retention time.Duration, bucketMs int64) int64 { + progress, err := s.metricRepo.GetAggregationProgress(ctx, metricType, bucketSeconds) + if err == nil && progress != nil && progress.LastBucket > 0 { + return progress.LastBucket + bucketMs + } + + // 默认从保留窗口开始 + start := time.Now().Add(-retention).UnixMilli() + return (start / bucketMs) * bucketMs +} + +// StartCleanupTask 启动数据清理任务 +func (s *MetricService) StartCleanupTask(ctx context.Context) { + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + + s.logger.Info("cleanup task started") + + for { + select { + case <-ctx.Done(): + s.logger.Info("cleanup task stopped") + return + case <-ticker.C: + s.cleanupOldMetrics(ctx) + } + } +} + +// cleanupOldMetrics 清理旧数据 +func (s *MetricService) cleanupOldMetrics(ctx context.Context) { + cfg := s.getMetricsConfig(ctx) + retentionDuration := time.Duration(cfg.RetentionHours) * time.Hour + before := time.Now().Add(-retentionDuration).UnixMilli() + + s.logger.Info("starting to clean old metrics", zap.Int64("beforeTimestamp", before), zap.Int("retentionHours", cfg.RetentionHours)) + + if err := s.metricRepo.DeleteOldMetrics(ctx, before); err != nil { + s.logger.Error("failed to clean old metrics", zap.Error(err)) + return + } + + s.logger.Info("old metrics cleaned successfully") +} + +// GetLatestMetrics 获取最新指标 +func (s *MetricService) GetLatestMetrics(ctx context.Context, agentID string) (*LatestMetrics, error) { + result := &LatestMetrics{} + + // 获取最新CPU指标 + if cpu, err := s.metricRepo.GetLatestCPUMetric(ctx, agentID); err == nil { + result.CPU = cpu + } + + // 获取最新内存指标 + if memory, err := s.metricRepo.GetLatestMemoryMetric(ctx, agentID); err == nil { + result.Memory = memory + } + + // 获取最新磁盘指标并计算平均使用率和总容量 + if disks, err := s.metricRepo.GetLatestDiskMetrics(ctx, agentID); err == nil && len(disks) > 0 { + var totalUsage float64 + var totalSpace, usedSpace, freeSpace uint64 + for _, disk := range disks { + totalUsage += disk.UsagePercent + totalSpace += disk.Total + usedSpace += disk.Used + freeSpace += disk.Free + } + result.Disk = &DiskSummary{ + AvgUsagePercent: totalUsage / float64(len(disks)), + TotalDisks: len(disks), + Total: totalSpace, + Used: usedSpace, + Free: freeSpace, + } + } + + // 获取最新网络指标并汇总速率和累计流量 + // 注意: 采集器已经计算好了每秒速率,这里直接汇总所有网卡的速率和累计流量 + if networks, err := s.metricRepo.GetLatestNetworkMetrics(ctx, agentID); err == nil && len(networks) > 0 { + var totalSentRate, totalRecvRate, totalSentTotal, totalRecvTotal uint64 + for _, net := range networks { + totalSentRate += net.BytesSentRate // 累加每个网卡的发送速率 + totalRecvRate += net.BytesRecvRate // 累加每个网卡的接收速率 + totalSentTotal += net.BytesSentTotal // 累加每个网卡的累计发送流量 + totalRecvTotal += net.BytesRecvTotal // 累加每个网卡的累计接收流量 + } + result.Network = &NetworkSummary{ + TotalBytesSentRate: totalSentRate, // 所有网卡的总发送速率(字节/秒) + TotalBytesRecvRate: totalRecvRate, // 所有网卡的总接收速率(字节/秒) + TotalBytesSentTotal: totalSentTotal, // 所有网卡的累计发送流量 + TotalBytesRecvTotal: totalRecvTotal, // 所有网卡的累计接收流量 + TotalInterfaces: len(networks), + } + } + + // 获取最新主机信息 + if host, err := s.metricRepo.GetLatestHostMetric(ctx, agentID); err == nil { + result.Host = host + } + + // 获取最新GPU信息 + if gpu, err := s.metricRepo.GetLatestGPUMetrics(ctx, agentID); err == nil && len(gpu) > 0 { + result.GPU = gpu + } + + // 获取最新温度信息 + if temp, err := s.metricRepo.GetLatestTemperatureMetrics(ctx, agentID); err == nil && len(temp) > 0 { + result.Temp = temp + } + + // 获取最新网络连接统计 + if netConn, err := s.metricRepo.GetLatestNetworkConnectionMetric(ctx, agentID); err == nil { + result.NetworkConnection = netConn + } + + return result, nil +} + +// GetLatestMonitorMetrics 获取最新的监控指标 +func (s *MetricService) GetLatestMonitorMetrics(ctx context.Context, agentID string) ([]models.MonitorMetric, error) { + return s.metricRepo.GetLatestMonitorMetrics(ctx, agentID) +} + +// GetMonitorMetrics 获取监控指标历史数据 +func (s *MetricService) GetMonitorMetrics(ctx context.Context, agentID, monitorName string, start, end int64) ([]models.MonitorMetric, error) { + return s.metricRepo.GetMonitorMetrics(ctx, agentID, monitorName, start, end) +} + +// GetMonitorMetricsByName 获取指定监控项的历史数据 +func (s *MetricService) GetMonitorMetricsByName(ctx context.Context, agentID, monitorName string, start, end int64, limit int) ([]models.MonitorMetric, error) { + return s.metricRepo.GetMonitorMetricsByName(ctx, agentID, monitorName, start, end, limit) +} + +// DeleteAgentMetrics 删除探针的所有指标数据 +func (s *MetricService) DeleteAgentMetrics(ctx context.Context, agentID string) error { + return s.metricRepo.DeleteAgentMetrics(ctx, agentID) +} + +// GetAvailableNetworkInterfaces 获取探针的可用网卡列表 +func (s *MetricService) GetAvailableNetworkInterfaces(ctx context.Context, agentID string) ([]string, error) { + return s.metricRepo.GetAvailableNetworkInterfaces(ctx, agentID) +} + +// DiskSummary 磁盘汇总数据 +type DiskSummary struct { + AvgUsagePercent float64 `json:"avgUsagePercent"` // 平均使用率 + TotalDisks int `json:"totalDisks"` // 磁盘数量 + Total uint64 `json:"total"` // 总容量(字节) + Used uint64 `json:"used"` // 已使用(字节) + Free uint64 `json:"free"` // 空闲(字节) +} + +// NetworkSummary 网络汇总数据 +type NetworkSummary struct { + TotalBytesSentRate uint64 `json:"totalBytesSentRate"` // 总发送速率(字节/秒) + TotalBytesRecvRate uint64 `json:"totalBytesRecvRate"` // 总接收速率(字节/秒) + TotalBytesSentTotal uint64 `json:"totalBytesSentTotal"` // 累计总发送流量 + TotalBytesRecvTotal uint64 `json:"totalBytesRecvTotal"` // 累计总接收流量 + TotalInterfaces int `json:"totalInterfaces"` // 网卡数量 +} + +// LatestMetrics 最新指标数据（用于API响应） +type LatestMetrics struct { + CPU *models.CPUMetric `json:"cpu,omitempty"` + Memory *models.MemoryMetric `json:"memory,omitempty"` + Disk *DiskSummary `json:"disk,omitempty"` + Network *NetworkSummary `json:"network,omitempty"` + NetworkConnection *models.NetworkConnectionMetric `json:"networkConnection,omitempty"` + Host *models.HostMetric `json:"host,omitempty"` + GPU []models.GPUMetric `json:"gpu,omitempty"` + Temp []models.TemperatureMetric `json:"temperature,omitempty"` +} diff --git a/internal/wire.go b/internal/wire.go index e3bae89..2ae39b3 100644 --- a/internal/wire.go +++ b/internal/wire.go @@ -27,6 +27,7 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App service.NewPropertyService, service.NewMonitorService, service.NewTamperService, + service.NewMetricService, service.NewNotifier, // WebSocket Manager diff --git a/internal/wire_gen.go b/internal/wire_gen.go index 0d6cea6..aee7d0f 100644 --- a/internal/wire_gen.go +++ b/internal/wire_gen.go @@ -27,7 +27,8 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App accountHandler := handler.NewAccountHandler(accountService) apiKeyService := service.NewApiKeyService(logger, db) propertyService := service.NewPropertyService(logger, db) - agentService := service.NewAgentService(logger, db, apiKeyService, propertyService) + metricService := service.NewMetricService(logger, db, propertyService) + agentService := service.NewAgentService(logger, db, apiKeyService, metricService) manager := websocket.NewManager(logger) monitorService := service.NewMonitorService(logger, db, manager) tamperRepo := repo.NewTamperRepo(db) diff --git a/web/src/api/agent.ts b/web/src/api/agent.ts index 195bb79..ec8c893 100644 --- a/web/src/api/agent.ts +++ b/web/src/api/agent.ts @@ -10,6 +10,7 @@ export interface GetAgentMetricsRequest { agentId: string; type: 'cpu' | 'memory' | 'disk' | 'network' | 'network_connection' | 'disk_io' | 'gpu' | 'temperature'; range?: '1m' | '5m' | '15m' | '30m' | '1h' | '3h' | '6h' | '12h' | '1d' | '24h' | '3d' | '7d' | '30d'; + interface?: string; // 网卡过滤参数（仅对 network 类型有效） } export interface GetAgentMetricsResponse { @@ -18,7 +19,7 @@ export interface GetAgentMetricsResponse { range: string; start: number; end: number; - interval: number; + interface?: string; metrics: any[]; } @@ -61,10 +62,13 @@ export const getAgentForAdmin = (id: string) => { }; export const getAgentMetrics = (params: GetAgentMetricsRequest) => { - const {agentId, type, range = '1h'} = params; + const {agentId, type, range = '1h', interface: interfaceName} = params; const query = new URLSearchParams(); query.append('type', type); query.append('range', range); + if (interfaceName) { + query.append('interface', interfaceName); + } return get(`/agents/${agentId}/metrics?${query.toString()}`); }; @@ -72,6 +76,15 @@ export const getAgentLatestMetrics = (agentId: string) => { return get(`/agents/${agentId}/metrics/latest`); }; +// 获取探针的可用网卡列表 +export interface GetNetworkInterfacesResponse { + interfaces: string[]; +} + +export const getAvailableNetworkInterfaces = (agentId: string) => { + return get(`/agents/${agentId}/network-interfaces`); +}; + export interface GetNetworkMetricsByInterfaceRequest { agentId: string; range?: '1m' | '5m' | '15m' | '30m' | '1h' | '3h' | '6h' | '12h' | '1d' | '24h' | '3d' | '7d' | '30d';