okooooooo/pika
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改最新数据为内存缓存，减少数据库查询

Browse Source
This commit is contained in:
dushixiang committed 2025-12-01 21:24:30 +08:00
1 parent 34a955bc65
commit f4b4975abd
7 files changed
+49 -269
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

go.mod
+1
View File
@@ -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
go.sum
+2
View File
@@ -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=
internal/handler/agent_handler.go
-16
View File
@@ -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")
internal/models/metric.go
+8 -10
View File
@@ -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 {
internal/repo/metric_repo.go
-166
View File
@@ -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
internal/service/agent_service.go
-5
View File
@@ -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)
internal/service/metric_service.go
+38 -72
View File
@@ -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"`
}