优化数据指标的存储和查询
This commit is contained in:
dushixiang committed
1 parent
91530b1282
commit
d23ebbce22
8 files changed
+902
-744
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
+15
-2
@@ -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<GetAgentMetricsResponse>(`/agents/${agentId}/metrics?${query.toString()}`);
|
||||
};
|
||||
|
||||
@@ -72,6 +76,15 @@ export const getAgentLatestMetrics = (agentId: string) => {
|
||||
return get<LatestMetrics>(`/agents/${agentId}/metrics/latest`);
|
||||
};
|
||||
|
||||
// 获取探针的可用网卡列表
|
||||
export interface GetNetworkInterfacesResponse {
|
||||
interfaces: string[];
|
||||
}
|
||||
|
||||
export const getAvailableNetworkInterfaces = (agentId: string) => {
|
||||
return get<GetNetworkInterfacesResponse>(`/agents/${agentId}/network-interfaces`);
|
||||
};
|
||||
|
||||
export interface GetNetworkMetricsByInterfaceRequest {
|
||||
agentId: string;
|
||||
range?: '1m' | '5m' | '15m' | '30m' | '1h' | '3h' | '6h' | '12h' | '1d' | '24h' | '3d' | '7d' | '30d';
|
||||
|
||||
Reference in new issue
Block a user