diff --git a/go.mod b/go.mod index fd3aa5d..7c723ee 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/coreos/go-oidc/v3 v3.17.0 github.com/fsnotify/fsnotify v1.9.0 github.com/go-errors/errors v1.5.1 + github.com/go-orz/cache v0.0.4 github.com/go-orz/orz v0.2.9 github.com/go-orz/toolkit v0.1.0 github.com/go-playground/locales v0.14.1 diff --git a/go.sum b/go.sum index dc2cde3..72e6780 100644 --- a/go.sum +++ b/go.sum @@ -28,6 +28,8 @@ github.com/go-jose/go-jose/v4 v4.1.3 h1:CVLmWDhDVRa6Mi/IgCgaopNosCaHz7zrMeF9MlZR github.com/go-jose/go-jose/v4 v4.1.3/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/go-orz/cache v0.0.4 h1:A8EwJQPiuctmnukFqkWFv4yoOKVen7DEpCVjSJAkAtw= +github.com/go-orz/cache v0.0.4/go.mod h1:qY5/YWUiMMFDHnWMCUJQakXILwJ/sIvbNCR04k08fBs= github.com/go-orz/orz v0.2.9 h1:9n3yl+N450nDxEfbfE6OX8muQmWorNp9E7YIpvx0SYQ= github.com/go-orz/orz v0.2.9/go.mod h1:KMrLiZ5R2ZyTho4VHWCJ7z5ff2dGYXCFOu6+lCNm++4= github.com/go-orz/toolkit v0.1.0 h1:u7jx6wuwgzTSMudiyqdFedgP1VmhZzUO+1r5F1Yw6mc= diff --git a/internal/handler/agent_handler.go b/internal/handler/agent_handler.go index cf3798a..2bd78fc 100644 --- a/internal/handler/agent_handler.go +++ b/internal/handler/agent_handler.go @@ -713,22 +713,6 @@ func (h *AgentHandler) GetMonitorMetrics(c echo.Context) error { }) } -// GetLatestMonitorMetrics 获取最新的监控指标 -func (h *AgentHandler) GetLatestMonitorMetrics(c echo.Context) error { - agentID := c.Param("id") - ctx := c.Request().Context() - - metrics, err := h.agentService.GetLatestMonitorMetrics(ctx, agentID) - if err != nil { - return err - } - - return orz.Ok(c, orz.Map{ - "agentId": agentID, - "metrics": metrics, - }) -} - // Delete 删除探针 func (h *AgentHandler) Delete(c echo.Context) error { agentID := c.Param("id") diff --git a/internal/models/metric.go b/internal/models/metric.go index e528418..2326517 100644 --- a/internal/models/metric.go +++ b/internal/models/metric.go @@ -43,7 +43,6 @@ type DiskMetric struct { Free uint64 `json:"free"` // 空闲(字节) UsagePercent float64 `json:"usagePercent"` // 使用率 Timestamp int64 `gorm:"index:idx_disk_agent_ts,priority:2;index:idx_disk_ts" json:"timestamp"` // 时间戳（毫秒） - TotalDisks int `json:"totalDisks"` // 磁盘总数量 } func (DiskMetric) TableName() string { @@ -52,15 +51,14 @@ func (DiskMetric) TableName() string { // NetworkMetric 网络指标 type NetworkMetric struct { - ID uint `gorm:"primaryKey;autoIncrement" json:"id"` - AgentID string `gorm:"index:idx_net_agent_ts,priority:1" json:"agentId"` // 探针ID - Interface string `json:"interface"` // 网卡名称 - BytesSentRate uint64 `json:"bytesSentRate"` // 发送速率(字节/秒) - BytesRecvRate uint64 `json:"bytesRecvRate"` // 接收速率(字节/秒) - BytesSentTotal uint64 `json:"bytesSentTotal"` // 累计发送字节数 - BytesRecvTotal uint64 `json:"bytesRecvTotal"` // 累计接收字节数 - Timestamp int64 `gorm:"index:idx_net_agent_ts,priority:2;index:idx_net_ts" json:"timestamp"` // 时间戳（毫秒） - TotalInterfaces int `json:"totalInterfaces"` // 网卡总数量 + ID uint `gorm:"primaryKey;autoIncrement" json:"id"` + AgentID string `gorm:"index:idx_net_agent_ts,priority:1" json:"agentId"` // 探针ID + Interface string `json:"interface"` // 网卡名称 + BytesSentRate uint64 `json:"bytesSentRate"` // 发送速率(字节/秒) + BytesRecvRate uint64 `json:"bytesRecvRate"` // 接收速率(字节/秒) + BytesSentTotal uint64 `json:"bytesSentTotal"` // 累计发送字节数 + BytesRecvTotal uint64 `json:"bytesRecvTotal"` // 累计接收字节数 + Timestamp int64 `gorm:"index:idx_net_agent_ts,priority:2;index:idx_net_ts" json:"timestamp"` // 时间戳（毫秒） } func (NetworkMetric) TableName() string { diff --git a/internal/repo/metric_repo.go b/internal/repo/metric_repo.go index 11a076e..961641d 100644 --- a/internal/repo/metric_repo.go +++ b/internal/repo/metric_repo.go @@ -2,8 +2,6 @@ package repo import ( "context" - "database/sql" - "fmt" "github.com/dushixiang/pika/internal/models" "gorm.io/gorm" @@ -55,46 +53,6 @@ func (r *MetricRepo) SaveTemperatureMetric(ctx context.Context, metric *models.T return r.db.WithContext(ctx).Create(metric).Error } -// GetLatestGPUMetrics 获取最新的GPU指标列表 -func (r *MetricRepo) GetLatestGPUMetrics(ctx context.Context, agentID string) ([]models.GPUMetric, error) { - var metrics []models.GPUMetric - // 获取每个GPU的最新记录 - err := r.db.WithContext(ctx). - Raw(` - SELECT g1.* FROM gpu_metrics g1 - INNER JOIN ( - SELECT index, MAX(timestamp) as max_timestamp - FROM gpu_metrics - WHERE agent_id = ? - GROUP BY index - ) g2 ON g1.index = g2.index AND g1.timestamp = g2.max_timestamp - WHERE g1.agent_id = ? - ORDER BY g1.index - `, agentID, agentID). - Scan(&metrics).Error - return metrics, err -} - -// GetLatestTemperatureMetrics 获取最新的温度指标列表 -func (r *MetricRepo) GetLatestTemperatureMetrics(ctx context.Context, agentID string) ([]models.TemperatureMetric, error) { - var metrics []models.TemperatureMetric - // 获取每个传感器的最新记录 - err := r.db.WithContext(ctx). - Raw(` - SELECT t1.* FROM temperature_metrics t1 - INNER JOIN ( - SELECT sensor_key, MAX(timestamp) as max_timestamp - FROM temperature_metrics - WHERE agent_id = ? - GROUP BY sensor_key - ) t2 ON t1.sensor_key = t2.sensor_key AND t1.timestamp = t2.max_timestamp - WHERE t1.agent_id = ? - ORDER BY t1.sensor_key - `, agentID, agentID). - Scan(&metrics).Error - return metrics, err -} - // SaveHostMetric 保存主机信息指标（按 agent 覆盖，避免先删后插的空窗） func (r *MetricRepo) SaveHostMetric(ctx context.Context, metric *models.HostMetric) error { return r.db.WithContext(ctx). @@ -105,37 +63,11 @@ func (r *MetricRepo) SaveHostMetric(ctx context.Context, metric *models.HostMetr Create(metric).Error } -// GetLatestHostMetric 获取最新的主机信息 -func (r *MetricRepo) GetLatestHostMetric(ctx context.Context, agentID string) (*models.HostMetric, error) { - var metric models.HostMetric - err := r.db.WithContext(ctx). - Where("agent_id = ?", agentID). - Order("timestamp DESC"). - First(&metric).Error - if err != nil { - return nil, err - } - return &metric, nil -} - // SaveNetworkConnectionMetric 保存网络连接统计指标 func (r *MetricRepo) SaveNetworkConnectionMetric(ctx context.Context, metric *models.NetworkConnectionMetric) error { return r.db.WithContext(ctx).Create(metric).Error } -// GetLatestNetworkConnectionMetric 获取最新的网络连接统计 -func (r *MetricRepo) GetLatestNetworkConnectionMetric(ctx context.Context, agentID string) (*models.NetworkConnectionMetric, error) { - var metric models.NetworkConnectionMetric - err := r.db.WithContext(ctx). - Where("agent_id = ?", agentID). - Order("timestamp DESC"). - First(&metric).Error - if err != nil { - return nil, err - } - return &metric, nil -} - // AggregatedNetworkConnectionMetric 网络连接统计聚合指标 type AggregatedNetworkConnectionMetric struct { Timestamp int64 `json:"timestamp"` @@ -365,86 +297,6 @@ func (r *MetricRepo) GetNetworkMetrics(ctx context.Context, agentID string, star return metrics, err } -// GetLatestCPUMetric 获取最新的CPU指标 -func (r *MetricRepo) GetLatestCPUMetric(ctx context.Context, agentID string) (*models.CPUMetric, error) { - var metric models.CPUMetric - err := r.db.WithContext(ctx). - Where("agent_id = ?", agentID). - Order("timestamp DESC"). - First(&metric).Error - if err != nil { - return nil, err - } - return &metric, nil -} - -// GetLatestMemoryMetric 获取最新的内存指标 -func (r *MetricRepo) GetLatestMemoryMetric(ctx context.Context, agentID string) (*models.MemoryMetric, error) { - var metric models.MemoryMetric - err := r.db.WithContext(ctx). - Where("agent_id = ?", agentID). - Order("timestamp DESC"). - First(&metric).Error - if err != nil { - return nil, err - } - return &metric, nil -} - -// GetLatestDiskMetrics 获取最新的磁盘指标（所有挂载点） -func (r *MetricRepo) GetLatestDiskMetrics(ctx context.Context, agentID string) (*models.DiskMetric, error) { - // 先获取最新时间戳 - var latestTimestamp sql.NullInt64 - err := r.db.WithContext(ctx). - Model(&models.DiskMetric{}). - Where("agent_id = ?", agentID). - Select("MAX(timestamp)"). - Scan(&latestTimestamp).Error - - if err != nil { - return nil, err - } - - if !latestTimestamp.Valid { - return nil, fmt.Errorf("no disk metrics found for agent %s", agentID) - } - - // 获取该时间戳的所有磁盘数据 - var metric models.DiskMetric - err = r.db.WithContext(ctx). - Where("mount_point = ? AND agent_id = ? AND timestamp = ?", "all", agentID, latestTimestamp.Int64). - First(&metric).Error - - return &metric, err -} - -// GetLatestNetworkMetrics 获取最新的网络指标（所有网卡） -func (r *MetricRepo) GetLatestNetworkMetrics(ctx context.Context, agentID string) (*models.NetworkMetric, error) { - // 先获取最新时间戳 - var latestTimestamp sql.NullInt64 - err := r.db.WithContext(ctx). - Model(&models.NetworkMetric{}). - Where("agent_id = ?", agentID). - Select("MAX(timestamp)"). - Scan(&latestTimestamp).Error - - if err != nil { - return nil, err - } - - if !latestTimestamp.Valid { - return nil, fmt.Errorf("no network metrics found for agent %s", agentID) - } - - // 获取该时间戳的所有网络数据 - var metric models.NetworkMetric - err = r.db.WithContext(ctx). - Where("agent_id = ? AND timestamp = ?", agentID, latestTimestamp.Int64). - First(&metric).Error - - return &metric, err -} - // SaveMonitorMetric 保存监控指标 func (r *MetricRepo) SaveMonitorMetric(ctx context.Context, metric *models.MonitorMetric) error { return r.db.WithContext(ctx).Create(metric).Error @@ -465,24 +317,6 @@ func (r *MetricRepo) GetMonitorMetrics(ctx context.Context, agentID, monitorID s return metrics, err } -// GetLatestMonitorMetrics 获取最新的监控指标（每个监控项的最新一条） -func (r *MetricRepo) GetLatestMonitorMetrics(ctx context.Context, agentID string) ([]models.MonitorMetric, error) { - var metrics []models.MonitorMetric - err := r.db.WithContext(ctx).Raw(` - SELECT m.* - FROM monitor_metrics m - INNER JOIN ( - SELECT monitor_id, MAX(timestamp) AS ts - FROM monitor_metrics - WHERE agent_id = ? - GROUP BY monitor_id - ) latest ON m.monitor_id = latest.monitor_id AND m.timestamp = latest.ts - WHERE m.agent_id = ? - ORDER BY m.monitor_id - `, agentID, agentID).Scan(&metrics).Error - return metrics, err -} - // GetMonitorMetricsByName 获取指定监控项的历史数据 func (r *MetricRepo) GetMonitorMetricsByName(ctx context.Context, agentID, monitorID string, start, end int64, limit int) ([]models.MonitorMetric, error) { var metrics []models.MonitorMetric diff --git a/internal/service/agent_service.go b/internal/service/agent_service.go index af25efc..88ccbca 100644 --- a/internal/service/agent_service.go +++ b/internal/service/agent_service.go @@ -311,11 +311,6 @@ func (s *AgentService) GetStatistics(ctx context.Context) (map[string]interface{ }, nil } -// GetLatestMonitorMetrics 获取最新的监控指标 -func (s *AgentService) GetLatestMonitorMetrics(ctx context.Context, agentID string) ([]models.MonitorMetric, error) { - 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.metricService.GetMonitorMetrics(ctx, agentID, monitorName, start, end) diff --git a/internal/service/metric_service.go b/internal/service/metric_service.go index 47155cf..1fa755f 100644 --- a/internal/service/metric_service.go +++ b/internal/service/metric_service.go @@ -9,6 +9,8 @@ import ( "github.com/dushixiang/pika/internal/models" "github.com/dushixiang/pika/internal/protocol" "github.com/dushixiang/pika/internal/repo" + + "github.com/go-orz/cache" "go.uber.org/zap" "gorm.io/gorm" ) @@ -43,6 +45,8 @@ type MetricService struct { metricRepo *repo.MetricRepo monitorStatsRepo *repo.MonitorStatsRepo propertyService *PropertyService + + latestCache cache.Cache[string, *LatestMetrics] } // NewMetricService 创建指标服务 @@ -52,6 +56,7 @@ func NewMetricService(logger *zap.Logger, db *gorm.DB, propertyService *Property metricRepo: repo.NewMetricRepo(db), monitorStatsRepo: repo.NewMonitorStatsRepo(db), propertyService: propertyService, + latestCache: cache.New[string, *LatestMetrics](time.Minute), } } @@ -59,6 +64,12 @@ func NewMetricService(logger *zap.Logger, db *gorm.DB, propertyService *Property func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, metricType string, data json.RawMessage) error { now := time.Now().UnixMilli() + latestMetrics, ok := s.latestCache.Get(agentID) + if !ok { + latestMetrics = &LatestMetrics{} + s.latestCache.Set(agentID, latestMetrics, time.Hour) + } + switch protocol.MetricType(metricType) { case protocol.MetricTypeCPU: // CPU数据现在包含静态和动态信息 @@ -74,6 +85,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me ModelName: cpuData.ModelName, Timestamp: now, } + latestMetrics.CPU = metric return s.metricRepo.SaveCPUMetric(ctx, metric) case protocol.MetricTypeMemory: @@ -93,6 +105,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me SwapFree: memData.SwapFree, Timestamp: now, } + latestMetrics.Memory = metric return s.metricRepo.SaveMemoryMetric(ctx, metric) case protocol.MetricTypeDisk: @@ -143,7 +156,13 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me Free: totalFree, UsagePercent: maxUsagePercent, // 使用率取最大值 Timestamp: now, - TotalDisks: len(diskDataList), + } + latestMetrics.Disk = &DiskSummary{ + AvgUsagePercent: totalMetric.UsagePercent, + TotalDisks: len(diskDataList), + Total: totalMetric.Total, + Used: totalMetric.Used, + Free: totalMetric.Free, } return s.metricRepo.SaveDiskMetric(ctx, totalMetric) @@ -186,14 +205,20 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me // 保存合并后的总和数据（interface 字段设置为空字符串） totalMetric := &models.NetworkMetric{ - AgentID: agentID, - Interface: "all", // 空字符串表示所有网卡的合并数据 - BytesSentRate: totalSentRate, - BytesRecvRate: totalRecvRate, - BytesSentTotal: totalSentTotal, - BytesRecvTotal: totalRecvTotal, - Timestamp: now, - TotalInterfaces: len(networkDataList), + AgentID: agentID, + Interface: "all", // 空字符串表示所有网卡的合并数据 + BytesSentRate: totalSentRate, + BytesRecvRate: totalRecvRate, + BytesSentTotal: totalSentTotal, + BytesRecvTotal: totalRecvTotal, + Timestamp: now, + } + latestMetrics.Network = &NetworkSummary{ + TotalBytesSentRate: totalSentRate, + TotalBytesRecvRate: totalRecvRate, + TotalBytesSentTotal: totalSentTotal, + TotalBytesRecvTotal: totalRecvTotal, + TotalInterfaces: len(networkDataList), } return s.metricRepo.SaveNetworkMetric(ctx, totalMetric) @@ -218,6 +243,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me Total: connData.Total, Timestamp: now, } + latestMetrics.NetworkConnection = metric return s.metricRepo.SaveNetworkConnectionMetric(ctx, metric) case protocol.MetricTypeDiskIO: @@ -285,6 +311,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me Procs: hostData.Procs, Timestamp: now, } + latestMetrics.Host = metric return s.metricRepo.SaveHostMetric(ctx, metric) case protocol.MetricTypeGPU: @@ -725,67 +752,8 @@ func (s *MetricService) cleanupOldMetrics(ctx context.Context) { // 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 disk, err := s.metricRepo.GetLatestDiskMetrics(ctx, agentID); err == nil { - result.Disk = &DiskSummary{ - AvgUsagePercent: disk.UsagePercent, - TotalDisks: disk.TotalDisks, - Total: disk.Total, - Used: disk.Used, - Free: disk.Free, - } - } - - // 获取最新网络指标并汇总速率和累计流量 - // 注意: 采集器已经计算好了每秒速率,这里直接汇总所有网卡的速率和累计流量 - if network, err := s.metricRepo.GetLatestNetworkMetrics(ctx, agentID); err == nil { - result.Network = &NetworkSummary{ - TotalBytesSentRate: network.BytesSentRate, // 所有网卡的总发送速率(字节/秒) - TotalBytesRecvRate: network.BytesRecvRate, // 所有网卡的总接收速率(字节/秒) - TotalBytesSentTotal: network.BytesSentTotal, // 所有网卡的累计发送流量 - TotalBytesRecvTotal: network.BytesRecvTotal, // 所有网卡的累计接收流量 - TotalInterfaces: network.TotalInterfaces, - } - } - - // 获取最新主机信息 - 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) + metrics, _ := s.latestCache.Get(agentID) + return metrics, nil } // GetMonitorMetrics 获取监控指标历史数据 @@ -834,6 +802,4 @@ type LatestMetrics struct { 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"` }