From 21d5c7a7177354111d4f75553de1a08cd93a2d81 Mon Sep 17 00:00:00 2001 From: dushixiang Date: Sat, 13 Dec 2025 11:55:31 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E6=9C=8D=E5=8A=A1=E7=9B=91?= =?UTF-8?q?=E6=8E=A7=E5=90=8E=E7=AB=AF=E9=87=8D=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/app.go | 6 +- internal/handler/agent_handler.go | 10 +- internal/handler/monitor_handler.go | 6 +- internal/metric/agent.go | 37 ++ internal/metric/metric.go | 29 + internal/metric/monitor.go | 20 + internal/protocol/message.go | 4 +- internal/repo/metric_repo.go | 124 ---- internal/repo/monitor_repo.go | 11 + internal/service/alert_service.go | 19 +- internal/service/metric_converter.go | 26 +- internal/service/metric_service.go | 855 +++++---------------------- internal/service/monitor_service.go | 50 +- internal/wire_gen.go | 2 +- pkg/agent/collector/monitor.go | 8 +- 15 files changed, 325 insertions(+), 882 deletions(-) create mode 100644 internal/metric/agent.go create mode 100644 internal/metric/metric.go create mode 100644 internal/metric/monitor.go diff --git a/internal/app.go b/internal/app.go index b0b16d0..d48a8be 100644 --- a/internal/app.go +++ b/internal/app.go @@ -371,9 +371,9 @@ func startMetricsMonitoring(ctx context.Context, components *AppComponents, logg for _, agent := range agents { // 获取最新指标 - latest, err := components.MetricService.GetLatestMetrics(ctx, agent.ID) - if err != nil { - logger.Debug("获取探针最新指标失败", zap.String("agentId", agent.ID), zap.Error(err)) + latest, ok := components.MetricService.GetLatestMetrics(agent.ID) + if !ok { + logger.Debug("获取探针最新指标失败", zap.String("agentId", agent.ID)) continue } if latest == nil { diff --git a/internal/handler/agent_handler.go b/internal/handler/agent_handler.go index 744add1..333df71 100644 --- a/internal/handler/agent_handler.go +++ b/internal/handler/agent_handler.go @@ -443,9 +443,9 @@ func (h *AgentHandler) GetLatestMetrics(c echo.Context) error { return err } - metrics, err := h.metricService.GetLatestMetrics(ctx, id) - if err != nil { - return err + metrics, ok := h.metricService.GetLatestMetrics(id) + if !ok { + return orz.NewError(404, "探针最新指标不存在") } return orz.Ok(c, metrics) @@ -514,8 +514,8 @@ func (h *AgentHandler) GetAgents(c echo.Context) error { } // 获取最新指标数据 - metrics, err := h.metricService.GetLatestMetrics(ctx, agent.ID) - if err == nil && metrics != nil { + metrics, ok := h.metricService.GetLatestMetrics(agent.ID) + if ok { item["metrics"] = metrics } diff --git a/internal/handler/monitor_handler.go b/internal/handler/monitor_handler.go index cb3624a..4bd5c3a 100644 --- a/internal/handler/monitor_handler.go +++ b/internal/handler/monitor_handler.go @@ -168,9 +168,9 @@ func (h *MonitorHandler) GetAgentStatsByID(c echo.Context) error { return err } - stats, err := h.monitorService.GetMonitorAgentStats(ctx, id) - if err != nil { - return err + stats, ok := h.monitorService.GetMonitorAgentStats(ctx, id) + if !ok { + return orz.NewError(404, "监控任务不存在") } return orz.Ok(c, stats) diff --git a/internal/metric/agent.go b/internal/metric/agent.go new file mode 100644 index 0000000..cc52d1c --- /dev/null +++ b/internal/metric/agent.go @@ -0,0 +1,37 @@ +package metric + +import ( + "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/protocol" +) + +// DiskSummary 磁盘汇总数据 +type DiskSummary struct { + UsagePercent float64 `json:"usagePercent"` // 平均使用率 + 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 *protocol.CPUData `json:"cpu,omitempty"` + Memory *protocol.MemoryData `json:"memory,omitempty"` + Disk *DiskSummary `json:"disk,omitempty"` + Network *NetworkSummary `json:"network,omitempty"` + NetworkConnection *protocol.NetworkConnectionData `json:"networkConnection,omitempty"` + Host *models.HostMetric `json:"host,omitempty"` + GPU []protocol.GPUData `json:"gpu,omitempty"` + Temp []protocol.TemperatureData `json:"temperature,omitempty"` + Monitors []protocol.MonitorData `json:"monitors,omitempty"` +} diff --git a/internal/metric/metric.go b/internal/metric/metric.go new file mode 100644 index 0000000..2da79d8 --- /dev/null +++ b/internal/metric/metric.go @@ -0,0 +1,29 @@ +package metric + +// DataPoint 统一的指标数据点结构 +type DataPoint struct { + Timestamp int64 `json:"timestamp"` // 毫秒时间戳 + Value float64 `json:"value"` +} + +// Series 指标系列（支持多系列，如多网卡、多传感器） +type Series struct { + Name string `json:"name"` // 系列名称 + Labels map[string]string `json:"labels,omitempty"` // 额外标签 + Data []DataPoint `json:"data"` // 数据点列表 +} + +// GetMetricsResponse 统一的查询响应格式 +type GetMetricsResponse struct { + AgentID string `json:"agentId"` + Type string `json:"type"` + Range string `json:"range"` + Series []Series `json:"series"` +} + +// QueryDefinition 查询定义（用于构建多个查询） +type QueryDefinition struct { + Name string // 系列名称 + Query string // PromQL 查询语句 + Labels map[string]string // 额外标签 +} diff --git a/internal/metric/monitor.go b/internal/metric/monitor.go new file mode 100644 index 0000000..59c9003 --- /dev/null +++ b/internal/metric/monitor.go @@ -0,0 +1,20 @@ +package metric + +import "github.com/dushixiang/pika/internal/protocol" + +// LatestMonitorMetrics 监控任务的最新指标（按 agent 分组） +type LatestMonitorMetrics struct { + MonitorID string `json:"monitorId"` + Agents map[string]*protocol.MonitorData `json:"agents"` // key: agentID + UpdatedAt int64 `json:"updatedAt"` // 最后更新时间 +} + +// MonitorStatsResult 监控统计结果（所有探针的聚合数据） +type MonitorStatsResult struct { + Status string `json:"status"` // 聚合状态（up/down/unknown） + ResponseTime int64 `json:"responseTime"` // 当前平均响应时间(ms) + CertExpiryDate int64 `json:"certExpiryDate,omitempty"` // 最早过期的证书时间(毫秒时间戳) + CertExpiryDays int `json:"certExpiryDays,omitempty"` // 证书剩余天数 + AgentCount int `json:"agentCount"` // 探针数量 + LastCheckTime int64 `json:"lastCheckTime"` // 最后检测时间(毫秒时间戳) +} diff --git a/internal/protocol/message.go b/internal/protocol/message.go index 0cd6e3d..e105d62 100644 --- a/internal/protocol/message.go +++ b/internal/protocol/message.go @@ -288,7 +288,8 @@ type Evidence struct { // MonitorData 监控数据 type MonitorData struct { - ID string `json:"id"` // 监控项ID + AgentId string `json:"agentId"` // 探针 ID + MonitorId string `json:"monitorId"` // 监控项ID Type string `json:"type"` // 监控类型: http, tcp Target string `json:"target"` // 监控目标 Status string `json:"status"` // 状态: up, down @@ -301,6 +302,7 @@ type MonitorData struct { // TLS 证书信息（仅用于 HTTPS） CertExpiryTime int64 `json:"certExpiryTime,omitempty"` // 证书过期时间(毫秒时间戳) CertDaysLeft int `json:"certDaysLeft,omitempty"` // 证书剩余天数 + Timestamp int64 `json:"timestamp"` // 检测时间(毫秒时间戳) } // TamperProtectConfig 防篡改保护配置（增量更新） diff --git a/internal/repo/metric_repo.go b/internal/repo/metric_repo.go index 13e5054..f470ec8 100644 --- a/internal/repo/metric_repo.go +++ b/internal/repo/metric_repo.go @@ -29,133 +29,9 @@ func (r *MetricRepo) SaveHostMetric(ctx context.Context, metric *models.HostMetr } // DeleteAgentMetrics 删除指定探针的所有指标数据 -// TODO: 需要改为调用 VictoriaMetrics 的 DELETE API func (r *MetricRepo) DeleteAgentMetrics(ctx context.Context, agentID string) error { - // 1. 删除 PostgreSQL 中的主机信息 if err := r.db.WithContext(ctx).Where("agent_id = ?", agentID).Delete(&models.HostMetric{}).Error; err != nil { return err } - - // 2. TODO: 调用 VictoriaMetrics 删除时间序列数据 - // DELETE /api/v1/admin/tsdb/delete_series?match[]=pika_*{agent_id="xxx"} - return nil } - -// 以下是兼容性数据结构定义，用于返回查询结果 -// TODO: 后续从 VictoriaMetrics 查询时使用这些结构 - -// AggregatedCPUMetric CPU聚合指标 -type AggregatedCPUMetric struct { - Timestamp int64 `json:"timestamp"` - MaxUsage float64 `json:"maxUsage"` - LogicalCores int `json:"logicalCores"` -} - -// AggregatedMemoryMetric 内存聚合指标 -type AggregatedMemoryMetric struct { - Timestamp int64 `json:"timestamp"` - MaxUsage float64 `json:"maxUsage"` - Total uint64 `json:"total"` -} - -// AggregatedDiskMetric 磁盘聚合指标 -type AggregatedDiskMetric struct { - Timestamp int64 `json:"timestamp"` - MountPoint string `json:"mountPoint"` - MaxUsage float64 `json:"maxUsage"` - Total uint64 `json:"total"` -} - -// AggregatedNetworkMetric 网络聚合指标 -type AggregatedNetworkMetric struct { - Timestamp int64 `json:"timestamp"` - Interface string `json:"interface"` - MaxSentRate float64 `json:"maxSentRate"` - MaxRecvRate float64 `json:"maxRecvRate"` -} - -// AggregatedNetworkConnectionMetric 网络连接统计聚合指标 -type AggregatedNetworkConnectionMetric struct { - Timestamp int64 `json:"timestamp"` - MaxEstablished uint32 `json:"maxEstablished"` - MaxSynSent uint32 `json:"maxSynSent"` - MaxSynRecv uint32 `json:"maxSynRecv"` - MaxFinWait1 uint32 `json:"maxFinWait1"` - MaxFinWait2 uint32 `json:"maxFinWait2"` - MaxTimeWait uint32 `json:"maxTimeWait"` - MaxClose uint32 `json:"maxClose"` - MaxCloseWait uint32 `json:"maxCloseWait"` - MaxLastAck uint32 `json:"maxLastAck"` - MaxListen uint32 `json:"maxListen"` - MaxClosing uint32 `json:"maxClosing"` - MaxTotal uint32 `json:"maxTotal"` -} - -// AggregatedDiskIOMetric 磁盘IO聚合指标 -type AggregatedDiskIOMetric struct { - Timestamp int64 `json:"timestamp"` - MaxReadRate float64 `json:"maxReadRate"` - MaxWriteRate float64 `json:"maxWriteRate"` - TotalReadBytes uint64 `json:"totalReadBytes"` - TotalWriteBytes uint64 `json:"totalWriteBytes"` - MaxIopsInProgress uint64 `json:"maxIopsInProgress"` -} - -// AggregatedGPUMetric GPU聚合指标 -type AggregatedGPUMetric struct { - Timestamp int64 `json:"timestamp"` - Index int `json:"index"` - Name string `json:"name"` - MaxUtilization float64 `json:"maxUtilization"` - MaxMemoryUsed uint64 `json:"maxMemoryUsed"` - MaxTemperature float64 `json:"maxTemperature"` - MaxPowerDraw float64 `json:"maxPowerDraw"` - MemoryTotal uint64 `json:"memoryTotal"` -} - -// AggregatedTemperatureMetric 温度聚合指标 -type AggregatedTemperatureMetric struct { - Timestamp int64 `json:"timestamp"` - SensorKey string `json:"sensorKey"` - SensorLabel string `json:"sensorLabel"` - MaxTemperature float64 `json:"maxTemperature"` -} - -// AggregatedMonitorMetric 聚合的监控指标 -type AggregatedMonitorMetric struct { - Timestamp int64 `json:"timestamp"` - AgentID string `json:"agentId"` - AvgResponse float64 `json:"avgResponse"` - MaxResponse int64 `json:"maxResponse"` - MinResponse int64 `json:"minResponse"` - SuccessCount int64 `json:"successCount"` - TotalCount int64 `json:"totalCount"` - SuccessRate float64 `json:"successRate"` - LastStatus string `json:"lastStatus"` - LastErrorMsg string `json:"lastErrorMsg"` -} - -// 注意：监控相关的查询方法已迁移到 MetricService -// DeleteMonitorMetrics 方法已在 MetricService 中实现 - -// MonitorMetric 监控指标（与service层定义保持一致） -type MonitorMetric struct { - ID uint `json:"id"` - AgentId string `json:"agentId"` - MonitorId string `json:"monitorId"` - Type string `json:"type"` - Target string `json:"target"` - Status string `json:"status"` - StatusCode int `json:"statusCode"` - ResponseTime int64 `json:"responseTime"` - Error string `json:"error"` - Message string `json:"message"` - ContentMatch bool `json:"contentMatch"` - CertExpiryTime int64 `json:"certExpiryTime"` - CertDaysLeft int `json:"certDaysLeft"` - Timestamp int64 `json:"timestamp"` -} - -// 注意：所有指标数据已迁移到 VictoriaMetrics -// 查询方法已移至 MetricService，通过 VMClient 访问 diff --git a/internal/repo/monitor_repo.go b/internal/repo/monitor_repo.go index ab3db11..b6193b0 100644 --- a/internal/repo/monitor_repo.go +++ b/internal/repo/monitor_repo.go @@ -68,3 +68,14 @@ func (r *MonitorRepo) FindByEnabled(ctx context.Context, enabled bool) ([]models } return monitors, nil } + +// FindByEnabledAndType 查找所有启用的监控任务 +func (r *MonitorRepo) FindByEnabledAndType(ctx context.Context, enabled bool, typ string) ([]models.MonitorTask, error) { + var monitors []models.MonitorTask + if err := r.GetDB(ctx). + Where("enabled = ? and type = ?", enabled, typ). + Find(&monitors).Error; err != nil { + return nil, err + } + return monitors, nil +} diff --git a/internal/service/alert_service.go b/internal/service/alert_service.go index 2c55e5a..30e69fe 100644 --- a/internal/service/alert_service.go +++ b/internal/service/alert_service.go @@ -6,6 +6,7 @@ import ( "time" "github.com/dushixiang/pika/internal/models" + "github.com/dushixiang/pika/internal/protocol" "github.com/dushixiang/pika/internal/repo" "github.com/go-orz/orz" "go.uber.org/zap" @@ -18,19 +19,19 @@ type AlertService struct { AlertRecordRepo *repo.AlertRecordRepo AlertStateRepo *repo.AlertStateRepo agentRepo *repo.AgentRepo - metricService *MetricService + monitorService *MonitorService propertyService *PropertyService notifier *Notifier logger *zap.Logger } -func NewAlertService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, metricService *MetricService, notifier *Notifier) *AlertService { +func NewAlertService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, monitorService *MonitorService, notifier *Notifier) *AlertService { return &AlertService{ Service: orz.NewService(db), AlertRecordRepo: repo.NewAlertRecordRepo(db), AlertStateRepo: repo.NewAlertStateRepo(db), agentRepo: repo.NewAgentRepo(db), - metricService: metricService, + monitorService: monitorService, propertyService: propertyService, notifier: notifier, logger: logger, @@ -373,7 +374,7 @@ func (s *AlertService) CheckMonitorAlerts(ctx context.Context) error { func (s *AlertService) checkCertificateAlerts(ctx context.Context, config *models.AlertConfig, now int64) error { // 获取所有最新的监控指标（仅HTTPS类型） // 这里需要查询最新的 monitor_metrics 记录，获取证书剩余天数 - monitors, err := s.metricService.GetLatestMonitorMetricsByType(ctx, "http") + monitors, err := s.monitorService.GetLatestMonitorMetricsByType(ctx, "http") if err != nil { return err } @@ -407,7 +408,7 @@ func (s *AlertService) checkCertificateAlerts(ctx context.Context, config *model } // checkCertAlert 检查并触发证书告警 -func (s *AlertService) checkCertAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *repo.MonitorMetric, certDaysLeft float64, now int64) { +func (s *AlertService) checkCertAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *protocol.MonitorData, certDaysLeft float64, now int64) { stateKey := fmt.Sprintf("%s:global:cert:%s", agent.ID, monitor.MonitorId) // 从数据库加载状态 @@ -479,7 +480,7 @@ func (s *AlertService) checkCertAlert(ctx context.Context, config *models.AlertC } // resolveCertAlert 恢复证书告警 -func (s *AlertService) resolveCertAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *repo.MonitorMetric, certDaysLeft float64) { +func (s *AlertService) resolveCertAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *protocol.MonitorData, certDaysLeft float64) { stateKey := fmt.Sprintf("%s:global:cert:%s", agent.ID, monitor.MonitorId) state, err := s.AlertStateRepo.GetAlertState(ctx, stateKey) @@ -536,7 +537,7 @@ func (s *AlertService) calculateCertLevel(daysLeft float64) string { // checkServiceDownAlerts 检查服务下线告警 func (s *AlertService) checkServiceDownAlerts(ctx context.Context, config *models.AlertConfig, now int64) error { // 获取所有最新的监控指标 - monitors, err := s.metricService.GetAllLatestMonitorMetrics(ctx) + monitors, err := s.monitorService.GetAllLatestMonitorMetrics(ctx) if err != nil { return err } @@ -603,7 +604,7 @@ func (s *AlertService) checkServiceDownAlerts(ctx context.Context, config *model } // fireServiceDownAlert 触发服务下线告警 -func (s *AlertService) fireServiceDownAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *repo.MonitorMetric, state *models.AlertState, now int64) { +func (s *AlertService) fireServiceDownAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *protocol.MonitorData, state *models.AlertState, now int64) { s.logger.Info("触发服务下线告警", zap.String("agentId", agent.ID), zap.String("monitorId", monitor.MonitorId), @@ -641,7 +642,7 @@ func (s *AlertService) fireServiceDownAlert(ctx context.Context, config *models. } // resolveServiceDownAlert 恢复服务下线告警 -func (s *AlertService) resolveServiceDownAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *repo.MonitorMetric, state *models.AlertState) { +func (s *AlertService) resolveServiceDownAlert(ctx context.Context, config *models.AlertConfig, agent *models.Agent, monitor *protocol.MonitorData, state *models.AlertState) { s.logger.Info("服务下线告警恢复", zap.String("agentId", agent.ID), zap.String("monitorId", monitor.MonitorId), diff --git a/internal/service/metric_converter.go b/internal/service/metric_converter.go index 66b94df..b5d9630 100644 --- a/internal/service/metric_converter.go +++ b/internal/service/metric_converter.go @@ -96,35 +96,11 @@ func (s *MetricService) convertToMetrics(agentID string, metricType string, data monitorDataList := data.([]protocol.MonitorData) for _, monitorData := range monitorDataList { labels := map[string]string{ - "monitor_id": monitorData.ID, + "monitor_id": monitorData.MonitorId, "monitor_type": monitorData.Type, "target": monitorData.Target, } metrics = append(metrics, createMetric("pika_monitor_response_time_ms", agentID, labels, float64(monitorData.ResponseTime), timestamp)) - - // 状态: up=1, down=0 - statusValue := 0.0 - if monitorData.Status == "up" { - statusValue = 1.0 - } - statusLabels := map[string]string{ - "monitor_id": monitorData.ID, - "monitor_type": monitorData.Type, - "target": monitorData.Target, - "status": monitorData.Status, - } - metrics = append(metrics, createMetric("pika_monitor_status", agentID, statusLabels, statusValue, timestamp)) - - // 状态码 - if monitorData.StatusCode > 0 { - metrics = append(metrics, createMetric("pika_monitor_status_code", agentID, labels, float64(monitorData.StatusCode), timestamp)) - } - - // 证书信息 - if monitorData.CertExpiryTime > 0 { - metrics = append(metrics, createMetric("pika_monitor_cert_expiry_timestamp_ms", agentID, labels, float64(monitorData.CertExpiryTime), timestamp)) - metrics = append(metrics, createMetric("pika_monitor_cert_days_left", agentID, labels, float64(monitorData.CertDaysLeft), timestamp)) - } } } diff --git a/internal/service/metric_service.go b/internal/service/metric_service.go index 21e7222..3668c88 100644 --- a/internal/service/metric_service.go +++ b/internal/service/metric_service.go @@ -7,6 +7,7 @@ import ( "strconv" "time" + "github.com/dushixiang/pika/internal/metric" "github.com/dushixiang/pika/internal/models" "github.com/dushixiang/pika/internal/protocol" "github.com/dushixiang/pika/internal/repo" @@ -17,34 +18,6 @@ import ( "gorm.io/gorm" ) -// MetricDataPoint 统一的指标数据点结构 -type MetricDataPoint struct { - Timestamp int64 `json:"timestamp"` // 毫秒时间戳 - Value float64 `json:"value"` -} - -// MetricSeries 指标系列（支持多系列，如多网卡、多传感器） -type MetricSeries struct { - Name string `json:"name"` // 系列名称 - Labels map[string]string `json:"labels,omitempty"` // 额外标签 - Data []MetricDataPoint `json:"data"` // 数据点列表 -} - -// GetMetricsResponse 统一的查询响应格式 -type GetMetricsResponse struct { - AgentID string `json:"agentId"` - Type string `json:"type"` - Range string `json:"range"` - Series []MetricSeries `json:"series"` -} - -// QueryDefinition 查询定义（用于构建多个查询） -type QueryDefinition struct { - Name string // 系列名称 - Query string // PromQL 查询语句 - Labels map[string]string // 额外标签 -} - // MetricService 指标服务 type MetricService struct { logger *zap.Logger @@ -53,18 +26,21 @@ type MetricService struct { trafficService *TrafficService // 流量统计服务 vmClient *vmclient.VMClient - latestCache cache.Cache[string, *LatestMetrics] + latestCache cache.Cache[string, *metric.LatestMetrics] // Agent 最新指标缓存 + + monitorLatestCache cache.Cache[string, *metric.LatestMonitorMetrics] // 监控最新指标缓存 } // NewMetricService 创建指标服务 func NewMetricService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, trafficService *TrafficService, vmClient *vmclient.VMClient) *MetricService { return &MetricService{ - logger: logger, - metricRepo: repo.NewMetricRepo(db), - propertyService: propertyService, - trafficService: trafficService, - vmClient: vmClient, - latestCache: cache.New[string, *LatestMetrics](time.Minute), + logger: logger, + metricRepo: repo.NewMetricRepo(db), + propertyService: propertyService, + trafficService: trafficService, + vmClient: vmClient, + latestCache: cache.New[string, *metric.LatestMetrics](time.Minute), + monitorLatestCache: cache.New[string, *metric.LatestMonitorMetrics](5 * time.Minute), // 监控数据缓存 5 分钟 } } @@ -75,7 +51,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me // 更新内存缓存 latestMetrics, ok := s.latestCache.Get(agentID) if !ok { - latestMetrics = &LatestMetrics{} + latestMetrics = &metric.LatestMetrics{} s.latestCache.Set(agentID, latestMetrics, time.Hour) } @@ -86,15 +62,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me if err := json.Unmarshal(data, &cpuData); err != nil { return err } - metric := &CPUMetric{ - AgentID: agentID, - UsagePercent: cpuData.UsagePercent, - LogicalCores: cpuData.LogicalCores, - PhysicalCores: cpuData.PhysicalCores, - ModelName: cpuData.ModelName, - Timestamp: now, - } - latestMetrics.CPU = metric + latestMetrics.CPU = &cpuData metrics := s.convertToMetrics(agentID, metricType, &cpuData, now) return s.vmClient.Write(ctx, metrics) @@ -103,19 +71,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me if err := json.Unmarshal(data, &memData); err != nil { return err } - metric := &MemoryMetric{ - AgentID: agentID, - Total: memData.Total, - Used: memData.Used, - Free: memData.Free, - Available: memData.Available, - UsagePercent: memData.UsagePercent, - SwapTotal: memData.SwapTotal, - SwapUsed: memData.SwapUsed, - SwapFree: memData.SwapFree, - Timestamp: now, - } - latestMetrics.Memory = metric + latestMetrics.Memory = &memData metrics := s.convertToMetrics(agentID, metricType, &memData, now) return s.vmClient.Write(ctx, metrics) @@ -135,7 +91,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me if totalTotal > 0 { usagePercent = float64(totalUsed) / float64(totalTotal) * 100 } - latestMetrics.Disk = &DiskSummary{ + latestMetrics.Disk = &metric.DiskSummary{ UsagePercent: usagePercent, TotalDisks: len(diskDataList), Total: totalTotal, @@ -159,7 +115,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me totalSentTotal += netData.BytesSentTotal totalRecvTotal += netData.BytesRecvTotal } - latestMetrics.Network = &NetworkSummary{ + latestMetrics.Network = &metric.NetworkSummary{ TotalBytesSentRate: totalSentRate, TotalBytesRecvRate: totalRecvRate, TotalBytesSentTotal: totalSentTotal, @@ -180,23 +136,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me if err := json.Unmarshal(data, &connData); err != nil { return err } - metric := &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, - } - latestMetrics.NetworkConnection = metric + latestMetrics.NetworkConnection = &connData metrics := s.convertToMetrics(agentID, metricType, &connData, now) return s.vmClient.Write(ctx, metrics) @@ -214,7 +154,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me return err } // Host 信息仍然保存到 PostgreSQL（静态信息，不频繁变化） - metric := &models.HostMetric{ + hostMetric := &models.HostMetric{ AgentID: agentID, OS: hostData.OS, Platform: hostData.Platform, @@ -226,8 +166,8 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me Procs: hostData.Procs, Timestamp: now, } - latestMetrics.Host = metric - return s.metricRepo.SaveHostMetric(ctx, metric) + latestMetrics.Host = hostMetric + return s.metricRepo.SaveHostMetric(ctx, hostMetric) case protocol.MetricTypeGPU: var gpuDataList []protocol.GPUData @@ -235,24 +175,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me return err } // 更新缓存 - var gpuMetrics []GPUMetric - for _, gpuData := range gpuDataList { - gpuMetrics = append(gpuMetrics, 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, - FanSpeed: gpuData.FanSpeed, - PerformanceState: "", - Timestamp: now, - }) - } - latestMetrics.GPU = gpuMetrics + latestMetrics.GPU = gpuDataList metrics := s.convertToMetrics(agentID, metricType, gpuDataList, now) return s.vmClient.Write(ctx, metrics) @@ -262,21 +185,7 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me return err } // 更新缓存 - var tempMetrics []TemperatureMetric - for _, tempData := range tempDataList { - sensorLabel := tempData.Type - if sensorLabel == "" { - sensorLabel = tempData.SensorKey - } - tempMetrics = append(tempMetrics, TemperatureMetric{ - AgentID: agentID, - SensorKey: tempData.SensorKey, - SensorLabel: sensorLabel, - Temperature: tempData.Temperature, - Timestamp: now, - }) - } - latestMetrics.Temp = tempMetrics + latestMetrics.Temp = tempDataList metrics := s.convertToMetrics(agentID, metricType, tempDataList, now) return s.vmClient.Write(ctx, metrics) @@ -285,18 +194,27 @@ func (s *MetricService) HandleMetricData(ctx context.Context, agentID string, me if err := json.Unmarshal(data, &monitorDataList); err != nil { return err } + for i := range monitorDataList { + monitorDataList[i].AgentId = agentID // 关联探针ID + } + // 更新缓存 + latestMetrics.Monitors = monitorDataList + for _, monitorData := range monitorDataList { + s.updateMonitorCache(agentID, &monitorData, now) + } + metrics := s.convertToMetrics(agentID, metricType, monitorDataList, now) return s.vmClient.Write(ctx, metrics) default: - s.logger.Warn("unknown metric type", zap.String("type", metricType)) + s.logger.Warn("unknown cpiMetric type", zap.String("type", metricType)) return nil } } // GetMetrics 获取聚合指标数据（从 VictoriaMetrics 查询） // 返回统一的 GetMetricsResponse 格式 -func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType string, start, end int64, interfaceName string) (*GetMetricsResponse, error) { +func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType string, start, end int64, interfaceName string) (*metric.GetMetricsResponse, error) { // 构造 PromQL 查询（返回多个查询以支持多系列） queries := s.buildPromQLQueries(agentID, metricType, interfaceName) if len(queries) == 0 { @@ -305,7 +223,7 @@ func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType stri // 执行查询并转换结果 // step 设为 0，让 VictoriaMetrics 自动选择合适的步长 - var series []MetricSeries + var series []metric.Series for _, q := range queries { result, err := s.vmClient.QueryRange(ctx, q.Query, @@ -324,7 +242,7 @@ func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType stri series = append(series, convertedSeries...) } - return &GetMetricsResponse{ + return &metric.GetMetricsResponse{ AgentID: agentID, Type: metricType, Range: fmt.Sprintf("%d-%d", start, end), @@ -332,24 +250,31 @@ func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType stri }, nil } -// alignTimeRangeToBucket 将时间范围对齐到桶边界，确保不同时间框架的桶数一致 -func alignTimeRangeToBucket(start, end int64, bucketMs int64) (int64, int64) { - if bucketMs <= 0 { - return start, end +// updateMonitorCache 更新监控数据缓存 +func (s *MetricService) updateMonitorCache(agentID string, monitorData *protocol.MonitorData, timestamp int64) { + monitorID := monitorData.MonitorId + + // 获取或创建监控缓存 + latestMetrics, ok := s.monitorLatestCache.Get(monitorID) + if !ok { + latestMetrics = &metric.LatestMonitorMetrics{ + MonitorID: monitorID, + Agents: make(map[string]*protocol.MonitorData), + } } - alignedStart := (start / bucketMs) * bucketMs - endBucket := ((end - 1) / bucketMs) * bucketMs - alignedEnd := endBucket + bucketMs - 1 - if alignedEnd < alignedStart { - alignedEnd = alignedStart - } - return alignedStart, alignedEnd + + // 更新探针数据 + latestMetrics.Agents[agentID] = monitorData + latestMetrics.UpdatedAt = timestamp + + // 保存到缓存（5分钟过期） + s.monitorLatestCache.Set(monitorID, latestMetrics, 5*time.Minute) } // GetLatestMetrics 获取最新指标 -func (s *MetricService) GetLatestMetrics(ctx context.Context, agentID string) (*LatestMetrics, error) { - metrics, _ := s.latestCache.Get(agentID) - return metrics, nil +func (s *MetricService) GetLatestMetrics(agentID string) (*metric.LatestMetrics, bool) { + metrics, ok := s.latestCache.Get(agentID) + return metrics, ok } // DeleteAgentMetrics 删除探针的所有指标数据 @@ -415,143 +340,25 @@ func (s *MetricService) GetAvailableNetworkInterfaces(ctx context.Context, agent return interfaces, nil } -// ===== 内存缓存使用的本地模型定义 ===== -// 注意：这些模型仅用于内存缓存，不再保存到 PostgreSQL - -// CPUMetric CPU指标（内存缓存） -type CPUMetric struct { - AgentID string `json:"agentId"` - UsagePercent float64 `json:"usagePercent"` - LogicalCores int `json:"logicalCores"` - PhysicalCores int `json:"physicalCores"` - ModelName string `json:"modelName"` - Timestamp int64 `json:"timestamp"` -} - -// MemoryMetric 内存指标（内存缓存） -type MemoryMetric struct { - AgentID string `json:"agentId"` - Total uint64 `json:"total"` - Used uint64 `json:"used"` - Free uint64 `json:"free"` - Available uint64 `json:"available"` - UsagePercent float64 `json:"usagePercent"` - SwapTotal uint64 `json:"swapTotal"` - SwapUsed uint64 `json:"swapUsed"` - SwapFree uint64 `json:"swapFree"` - Timestamp int64 `json:"timestamp"` -} - -// NetworkConnectionMetric 网络连接统计指标（内存缓存） -type NetworkConnectionMetric struct { - AgentID string `json:"agentId"` - Established uint32 `json:"established"` - SynSent uint32 `json:"synSent"` - SynRecv uint32 `json:"synRecv"` - FinWait1 uint32 `json:"finWait1"` - FinWait2 uint32 `json:"finWait2"` - TimeWait uint32 `json:"timeWait"` - Close uint32 `json:"close"` - CloseWait uint32 `json:"closeWait"` - LastAck uint32 `json:"lastAck"` - Listen uint32 `json:"listen"` - Closing uint32 `json:"closing"` - Total uint32 `json:"total"` - Timestamp int64 `json:"timestamp"` -} - -// GPUMetric GPU指标（内存缓存） -type GPUMetric struct { - AgentID string `json:"agentId"` - Index int `json:"index"` - Name string `json:"name"` - Utilization float64 `json:"utilization"` - MemoryTotal uint64 `json:"memoryTotal"` - MemoryUsed uint64 `json:"memoryUsed"` - MemoryFree uint64 `json:"memoryFree"` - Temperature float64 `json:"temperature"` - PowerDraw float64 `json:"powerDraw"` - FanSpeed float64 `json:"fanSpeed"` - PerformanceState string `json:"performanceState"` - Timestamp int64 `json:"timestamp"` -} - -// TemperatureMetric 温度指标（内存缓存） -type TemperatureMetric struct { - AgentID string `json:"agentId"` - SensorKey string `json:"sensorKey"` - SensorLabel string `json:"sensorLabel"` - Temperature float64 `json:"temperature"` - Timestamp int64 `json:"timestamp"` -} - -// MonitorMetric 监控指标 -type MonitorMetric struct { - ID uint `json:"id"` - AgentId string `json:"agentId"` - MonitorId string `json:"monitorId"` - Type string `json:"type"` - Target string `json:"target"` - Status string `json:"status"` - StatusCode int `json:"statusCode"` - ResponseTime int64 `json:"responseTime"` - Error string `json:"error"` - Message string `json:"message"` - ContentMatch bool `json:"contentMatch"` - CertExpiryTime int64 `json:"certExpiryTime"` - CertDaysLeft int `json:"certDaysLeft"` - Timestamp int64 `json:"timestamp"` -} - -// DiskSummary 磁盘汇总数据 -type DiskSummary struct { - UsagePercent float64 `json:"usagePercent"` // 平均使用率 - 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 *CPUMetric `json:"cpu,omitempty"` - Memory *MemoryMetric `json:"memory,omitempty"` - Disk *DiskSummary `json:"disk,omitempty"` - Network *NetworkSummary `json:"network,omitempty"` - NetworkConnection *NetworkConnectionMetric `json:"networkConnection,omitempty"` - Host *models.HostMetric `json:"host,omitempty"` - GPU []GPUMetric `json:"gpu,omitempty"` - Temp []TemperatureMetric `json:"temperature,omitempty"` -} - // buildPromQLQueries 构造 PromQL 查询列表（支持多系列） -func (s *MetricService) buildPromQLQueries(agentID, metricType string, interfaceName string) []QueryDefinition { - var queries []QueryDefinition +func (s *MetricService) buildPromQLQueries(agentID, metricType string, interfaceName string) []metric.QueryDefinition { + var queries []metric.QueryDefinition switch metricType { case "cpu": - queries = []QueryDefinition{{ + queries = []metric.QueryDefinition{{ Name: "usage", Query: fmt.Sprintf(`pika_cpu_usage_percent{agent_id="%s"}`, agentID), }} case "memory": - queries = []QueryDefinition{{ + queries = []metric.QueryDefinition{{ Name: "usage", Query: fmt.Sprintf(`pika_memory_usage_percent{agent_id="%s"}`, agentID), }} case "disk": - queries = []QueryDefinition{{ + queries = []metric.QueryDefinition{{ Name: "usage", Query: fmt.Sprintf(`pika_disk_usage_percent{agent_id="%s",mount_point=""}`, agentID), }} @@ -560,7 +367,7 @@ func (s *MetricService) buildPromQLQueries(agentID, metricType string, interface // 网络流量：上行和下行 if interfaceName != "" && interfaceName != "all" { // 指定网卡 - queries = []QueryDefinition{ + queries = []metric.QueryDefinition{ { Name: "upload", Query: fmt.Sprintf(`pika_network_sent_bytes_rate{agent_id="%s",interface="%s"}`, agentID, interfaceName), @@ -574,7 +381,7 @@ func (s *MetricService) buildPromQLQueries(agentID, metricType string, interface } } else { // 所有网卡汇总 - queries = []QueryDefinition{ + queries = []metric.QueryDefinition{ { Name: "upload", Query: fmt.Sprintf(`sum(pika_network_sent_bytes_rate{agent_id="%s"}) by (agent_id)`, agentID), @@ -588,7 +395,7 @@ func (s *MetricService) buildPromQLQueries(agentID, metricType string, interface case "network_connection": // 网络连接统计：多个状态 - queries = []QueryDefinition{ + queries = []metric.QueryDefinition{ {Name: "established", Query: fmt.Sprintf(`pika_network_conn_established{agent_id="%s"}`, agentID)}, {Name: "time_wait", Query: fmt.Sprintf(`pika_network_conn_time_wait{agent_id="%s"}`, agentID)}, {Name: "close_wait", Query: fmt.Sprintf(`pika_network_conn_close_wait{agent_id="%s"}`, agentID)}, @@ -597,14 +404,14 @@ func (s *MetricService) buildPromQLQueries(agentID, metricType string, interface case "disk_io": // 磁盘 IO：读和写 - queries = []QueryDefinition{ + queries = []metric.QueryDefinition{ {Name: "read", Query: fmt.Sprintf(`pika_disk_read_bytes_rate{agent_id="%s"}`, agentID)}, {Name: "write", Query: fmt.Sprintf(`pika_disk_write_bytes_rate{agent_id="%s"}`, agentID)}, } case "gpu": // GPU：利用率和温度（按 GPU 分组） - queries = []QueryDefinition{ + queries = []metric.QueryDefinition{ { Name: "utilization", Query: fmt.Sprintf(`pika_gpu_utilization_percent{agent_id="%s"}`, agentID), @@ -617,7 +424,7 @@ func (s *MetricService) buildPromQLQueries(agentID, metricType string, interface case "temperature": // 温度：按传感器类型分组 - queries = []QueryDefinition{{ + queries = []metric.QueryDefinition{{ Name: "temperature", Query: fmt.Sprintf(`pika_temperature_celsius{agent_id="%s"}`, agentID), }} @@ -627,17 +434,17 @@ func (s *MetricService) buildPromQLQueries(agentID, metricType string, interface } // convertQueryResultToSeries 将 VictoriaMetrics 查询结果转换为 MetricSeries -func (s *MetricService) convertQueryResultToSeries(result *vmclient.QueryResult, seriesName string, extraLabels map[string]string) []MetricSeries { +func (s *MetricService) convertQueryResultToSeries(result *vmclient.QueryResult, seriesName string, extraLabels map[string]string) []metric.Series { if result == nil || len(result.Data.Result) == 0 { - return []MetricSeries{} + return nil } - var allSeries []MetricSeries + var allSeries []metric.Series // 遍历所有时间序列 for _, timeSeries := range result.Data.Result { // 提取数据点 - var dataPoints []MetricDataPoint + var dataPoints []metric.DataPoint for _, valueArray := range timeSeries.Values { if len(valueArray) != 2 { continue @@ -654,7 +461,7 @@ func (s *MetricService) convertQueryResultToSeries(result *vmclient.QueryResult, } value, _ := strconv.ParseFloat(valueStr, 64) - dataPoints = append(dataPoints, MetricDataPoint{ + dataPoints = append(dataPoints, metric.DataPoint{ Timestamp: int64(timestamp * 1000), // 转换为毫秒 Value: value, }) @@ -683,7 +490,7 @@ func (s *MetricService) convertQueryResultToSeries(result *vmclient.QueryResult, delete(labels, "gpu_index") } - allSeries = append(allSeries, MetricSeries{ + allSeries = append(allSeries, metric.Series{ Name: finalName, Labels: labels, Data: dataPoints, @@ -693,174 +500,84 @@ func (s *MetricService) convertQueryResultToSeries(result *vmclient.QueryResult, return allSeries } -// ===== 监控查询相关 ===== - -// 监控查询类型常量 -const ( - MonitorQueryTypeCurrent = "current" // 当前状态 - MonitorQueryTypeStats24h = "stats_24h" // 24小时统计 - MonitorQueryTypeStats7d = "stats_7d" // 7天统计 - MonitorQueryTypeHistory = "history" // 历史趋势 -) - -// MonitorStatsResult 监控统计结果（所有探针的聚合数据） -type MonitorStatsResult struct { - Status string `json:"status"` // 聚合状态（up/down/unknown） - ResponseTime int64 `json:"responseTime"` // 当前平均响应时间(ms) - CertExpiryDate int64 `json:"certExpiryDate,omitempty"` // 最早过期的证书时间(毫秒时间戳) - CertExpiryDays int `json:"certExpiryDays,omitempty"` // 证书剩余天数 - AgentCount int `json:"agentCount"` // 探针数量 - LastCheckTime int64 `json:"lastCheckTime"` // 最后检测时间(毫秒时间戳) -} - -// AgentMonitorStat 单个探针的监控统计 -type AgentMonitorStat struct { - AgentID string `json:"agentID,omitempty"` - Status string `json:"status,omitempty"` // up/down/unknown - ResponseTime int64 `json:"responseTime,omitempty"` // 当前响应时间(ms) - LastCheckTime int64 `json:"lastCheckTime,omitempty"` // 最后检测时间(毫秒时间戳) - CertExpiryDate int64 `json:"certExpiryDate,omitempty"` // 证书过期时间(毫秒时间戳) - CertExpiryDays int `json:"certExpiryDays,omitempty"` // 证书剩余天数 -} - -// buildMonitorPromQLQueries 构建监控查询的 PromQL 语句（简化版：只查询当前状态） -func (s *MetricService) buildMonitorPromQLQueries(monitorID string, queryType string) []QueryDefinition { - var queries []QueryDefinition - - switch queryType { - case MonitorQueryTypeCurrent: - // 当前状态查询（即时查询）- 简化为只查询必要的指标 - queries = []QueryDefinition{ - {Name: "response_time", Query: fmt.Sprintf(`pika_monitor_response_time_ms{monitor_id="%s"}`, monitorID)}, - {Name: "status", Query: fmt.Sprintf(`pika_monitor_status{monitor_id="%s"}`, monitorID)}, - {Name: "cert_days", Query: fmt.Sprintf(`pika_monitor_cert_days_left{monitor_id="%s"}`, monitorID)}, - {Name: "cert_expiry", Query: fmt.Sprintf(`pika_monitor_cert_expiry_timestamp_ms{monitor_id="%s"}`, monitorID)}, - } - - case MonitorQueryTypeHistory: - // 历史趋势查询（范围查询） - queries = []QueryDefinition{ - {Name: "response_time", Query: fmt.Sprintf(`pika_monitor_response_time_ms{monitor_id="%s"}`, monitorID)}, - } +// buildMonitorPromQLQueries 构建监控查询的 PromQL 语句 +func (s *MetricService) buildMonitorPromQLQueries(monitorID string) []metric.QueryDefinition { + var queries = []metric.QueryDefinition{ + {Name: "response_time", Query: fmt.Sprintf(`pika_monitor_response_time_ms{monitor_id="%s"}`, monitorID)}, } - return queries } -// GetMonitorStats 获取监控任务的聚合统计数据（简化版：只查询当前状态） -func (s *MetricService) GetMonitorStats(ctx context.Context, monitorID string) (*MonitorStatsResult, error) { - result := &MonitorStatsResult{ +// GetMonitorHistory 获取监控任务的历史趋势数据 +func (s *MetricService) GetMonitorHistory(ctx context.Context, monitorID string, start, end int64) (*metric.GetMetricsResponse, error) { + queries := s.buildMonitorPromQLQueries(monitorID) + + var series []metric.Series + for _, q := range queries { + result, err := s.vmClient.QueryRange( + ctx, + q.Query, + time.UnixMilli(start), + time.UnixMilli(end), + 0, // 自动步长 + ) + if err != nil { + s.logger.Warn("查询历史趋势失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + convertedSeries := s.convertQueryResultToSeries(result, q.Name, q.Labels) + series = append(series, convertedSeries...) + } + + return &metric.GetMetricsResponse{ + AgentID: "", // 监控查询不限定单个agent + Type: "monitor", + Range: fmt.Sprintf("%d-%d", start, end), + Series: series, + }, nil +} + +// GetMonitorAgentStats 获取监控任务各探针的统计数据（只从缓存读取） +func (s *MetricService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]protocol.MonitorData, bool) { + // 从缓存读取监控数据 + latestMetrics, ok := s.monitorLatestCache.Get(monitorID) + if !ok { + // 缓存不存在，返回空列表 + return nil, false + } + + // 转换为数组 + result := make([]protocol.MonitorData, 0, len(latestMetrics.Agents)) + for _, stat := range latestMetrics.Agents { + result = append(result, *stat) + } + + return result, true +} + +// GetMonitorStats 获取监控任务的聚合统计数据（只从缓存读取） +func (s *MetricService) GetMonitorStats(ctx context.Context, monitorID string) (*metric.MonitorStatsResult, error) { + // 从缓存读取监控数据 + latestMetrics, ok := s.monitorLatestCache.Get(monitorID) + if !ok { + // 缓存不存在，返回默认值 + return &metric.MonitorStatsResult{ + Status: "unknown", + }, nil + } + + // 聚合各探针数据 + return s.aggregateMonitorStats(latestMetrics), nil +} + +// aggregateMonitorStats 聚合各探针的监控数据 +func (s *MetricService) aggregateMonitorStats(latestMetrics *metric.LatestMonitorMetrics) *metric.MonitorStatsResult { + result := &metric.MonitorStatsResult{ Status: "unknown", } - // 查询当前状态（只需4个查询） - queries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeCurrent) - queryData := make(map[string]*vmclient.QueryResult) - for _, q := range queries { - queryResult, err := s.vmClient.Query(ctx, q.Query) - if err != nil { - s.logger.Warn("查询监控状态失败", zap.String("query", q.Name), zap.Error(err)) - continue - } - queryData[q.Name] = queryResult - } - - // 聚合各探针的数据 - agentStats := make(map[string]*AgentMonitorStat) - - // 处理响应时间 - if respResult, ok := queryData["response_time"]; ok && respResult != nil { - for _, ts := range respResult.Data.Result { - agentID := ts.Metric["agent_id"] - if agentID == "" { - continue - } - if _, exists := agentStats[agentID]; !exists { - agentStats[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - timestamp, _ := lastValue[0].(float64) - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - agentStats[agentID].ResponseTime = int64(value) - agentStats[agentID].LastCheckTime = int64(timestamp * 1000) - } - } - } - } - - // 处理状态 - if statusResult, ok := queryData["status"]; ok && statusResult != nil { - for _, ts := range statusResult.Data.Result { - agentID := ts.Metric["agent_id"] - status := ts.Metric["status"] - if agentID == "" { - continue - } - if _, exists := agentStats[agentID]; !exists { - agentStats[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - if status != "" { - agentStats[agentID].Status = status - } - } - } - } - - // 处理证书信息 - if certDaysResult, ok := queryData["cert_days"]; ok && certDaysResult != nil { - for _, ts := range certDaysResult.Data.Result { - agentID := ts.Metric["agent_id"] - if agentID == "" { - continue - } - if _, exists := agentStats[agentID]; !exists { - agentStats[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - agentStats[agentID].CertExpiryDays = int(value) - } - } - } - } - - if certExpiryResult, ok := queryData["cert_expiry"]; ok && certExpiryResult != nil { - for _, ts := range certExpiryResult.Data.Result { - agentID := ts.Metric["agent_id"] - if agentID == "" { - continue - } - if _, exists := agentStats[agentID]; !exists { - agentStats[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - agentStats[agentID].CertExpiryDate = int64(value) - } - } - } - } - - // 聚合所有探针的数据 - if len(agentStats) == 0 { - return result, nil + if len(latestMetrics.Agents) == 0 { + return result } var totalResponseTime int64 @@ -871,11 +588,11 @@ func (s *MetricService) GetMonitorStats(ctx context.Context, monitorID string) ( var minCertExpiryDate int64 var minCertExpiryDays int - for _, stat := range agentStats { + for _, stat := range latestMetrics.Agents { totalResponseTime += stat.ResponseTime - if stat.LastCheckTime > lastCheckTime { - lastCheckTime = stat.LastCheckTime + if stat.Timestamp > lastCheckTime { + lastCheckTime = stat.Timestamp } if stat.Status == "up" { @@ -884,16 +601,16 @@ func (s *MetricService) GetMonitorStats(ctx context.Context, monitorID string) ( hasDown = true } - if stat.CertExpiryDate > 0 { - if !hasCert || stat.CertExpiryDate < minCertExpiryDate { - minCertExpiryDate = stat.CertExpiryDate - minCertExpiryDays = stat.CertExpiryDays + if stat.CertExpiryTime > 0 { + if !hasCert || stat.CertExpiryTime < minCertExpiryDate { + minCertExpiryDate = stat.CertExpiryTime + minCertExpiryDays = stat.CertDaysLeft hasCert = true } } } - count := len(agentStats) + count := len(latestMetrics.Agents) result.AgentCount = count if count > 0 { result.ResponseTime = totalResponseTime / int64(count) @@ -912,271 +629,5 @@ func (s *MetricService) GetMonitorStats(ctx context.Context, monitorID string) ( result.CertExpiryDays = minCertExpiryDays } - return result, nil -} - -// GetMonitorAgentStats 获取监控任务各探针的统计数据（简化版：只查询当前状态） -func (s *MetricService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]AgentMonitorStat, error) { - // 查询当前状态（只需4个查询） - queries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeCurrent) - queryData := make(map[string]*vmclient.QueryResult) - for _, q := range queries { - queryResult, err := s.vmClient.Query(ctx, q.Query) - if err != nil { - s.logger.Warn("查询监控状态失败", zap.String("query", q.Name), zap.Error(err)) - continue - } - queryData[q.Name] = queryResult - } - - // 提取每个探针的数据 - agentStatsMap := make(map[string]*AgentMonitorStat) - - // 处理响应时间 - if respResult, ok := queryData["response_time"]; ok && respResult != nil { - for _, ts := range respResult.Data.Result { - agentID := ts.Metric["agent_id"] - if agentID == "" { - continue - } - if _, exists := agentStatsMap[agentID]; !exists { - agentStatsMap[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - timestamp, _ := lastValue[0].(float64) - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - agentStatsMap[agentID].ResponseTime = int64(value) - agentStatsMap[agentID].LastCheckTime = int64(timestamp * 1000) - } - } - } - } - - // 处理状态 - if statusResult, ok := queryData["status"]; ok && statusResult != nil { - for _, ts := range statusResult.Data.Result { - agentID := ts.Metric["agent_id"] - status := ts.Metric["status"] - if agentID == "" { - continue - } - if _, exists := agentStatsMap[agentID]; !exists { - agentStatsMap[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - if status != "" { - agentStatsMap[agentID].Status = status - } - } - } - } - - // 处理证书信息 - if certDaysResult, ok := queryData["cert_days"]; ok && certDaysResult != nil { - for _, ts := range certDaysResult.Data.Result { - agentID := ts.Metric["agent_id"] - if agentID == "" { - continue - } - if _, exists := agentStatsMap[agentID]; !exists { - agentStatsMap[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - agentStatsMap[agentID].CertExpiryDays = int(value) - } - } - } - } - - if certExpiryResult, ok := queryData["cert_expiry"]; ok && certExpiryResult != nil { - for _, ts := range certExpiryResult.Data.Result { - agentID := ts.Metric["agent_id"] - if agentID == "" { - continue - } - if _, exists := agentStatsMap[agentID]; !exists { - agentStatsMap[agentID] = &AgentMonitorStat{AgentID: agentID} - } - - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - agentStatsMap[agentID].CertExpiryDate = int64(value) - } - } - } - } - - // 转换为数组 - result := make([]AgentMonitorStat, 0, len(agentStatsMap)) - for _, stat := range agentStatsMap { - result = append(result, *stat) - } - - return result, nil -} - -// GetMonitorHistory 获取监控任务的历史趋势数据 -func (s *MetricService) GetMonitorHistory(ctx context.Context, monitorID string, start, end int64) (*GetMetricsResponse, error) { - queries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeHistory) - - var series []MetricSeries - for _, q := range queries { - result, err := s.vmClient.QueryRange( - ctx, - q.Query, - time.UnixMilli(start), - time.UnixMilli(end), - 0, // 自动步长 - ) - if err != nil { - s.logger.Warn("查询历史趋势失败", zap.String("query", q.Name), zap.Error(err)) - continue - } - convertedSeries := s.convertQueryResultToSeries(result, q.Name, q.Labels) - series = append(series, convertedSeries...) - } - - return &GetMetricsResponse{ - AgentID: "", // 监控查询不限定单个agent - Type: "monitor", - Range: fmt.Sprintf("%d-%d", start, end), - Series: series, - }, nil -} - -// GetLatestMonitorMetricsByType 获取指定类型的最新监控指标（用于告警检查） -func (s *MetricService) GetLatestMonitorMetricsByType(ctx context.Context, monitorType string) ([]repo.MonitorMetric, error) { - // 查询最新的监控状态（按 monitor_type 过滤） - query := fmt.Sprintf(`pika_monitor_status{monitor_type="%s"}`, monitorType) - result, err := s.vmClient.Query(ctx, query) - if err != nil { - return nil, fmt.Errorf("query latest monitor metrics failed: %w", err) - } - - metrics := make([]repo.MonitorMetric, 0) - for _, ts := range result.Data.Result { - metric := s.convertToMonitorMetric(ts) - metrics = append(metrics, metric) - } - - return metrics, nil -} - -// GetAllLatestMonitorMetrics 获取所有最新监控指标（用于告警检查） -func (s *MetricService) GetAllLatestMonitorMetrics(ctx context.Context) ([]repo.MonitorMetric, error) { - // 查询所有最新的监控状态 - query := `pika_monitor_status` - result, err := s.vmClient.Query(ctx, query) - if err != nil { - return nil, fmt.Errorf("query all latest monitor metrics failed: %w", err) - } - - metrics := make([]repo.MonitorMetric, 0) - for _, ts := range result.Data.Result { - metric := s.convertToMonitorMetric(ts) - metrics = append(metrics, metric) - } - - return metrics, nil -} - -// convertToMonitorMetric 将 VictoriaMetrics 查询结果转换为 MonitorMetric -func (s *MetricService) convertToMonitorMetric(ts vmclient.Result) repo.MonitorMetric { - metric := repo.MonitorMetric{ - AgentId: ts.Metric["agent_id"], - MonitorId: ts.Metric["monitor_id"], - Type: ts.Metric["monitor_type"], - Target: ts.Metric["target"], - Status: ts.Metric["status"], - } - - // 获取最新值（timestamp 和 status value） - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - timestamp, _ := lastValue[0].(float64) - metric.Timestamp = int64(timestamp * 1000) // 转换为毫秒 - } - } - - // 查询其他相关指标（响应时间、状态码、证书信息等） - // 这里需要额外的查询来填充完整的 MonitorMetric 数据 - ctx := context.Background() - - // 查询响应时间 - respQuery := fmt.Sprintf(`pika_monitor_response_time_ms{agent_id="%s",monitor_id="%s"}`, metric.AgentId, metric.MonitorId) - if respResult, err := s.vmClient.Query(ctx, respQuery); err == nil && len(respResult.Data.Result) > 0 { - ts := respResult.Data.Result[0] - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - metric.ResponseTime = int64(value) - } - } - } - - // 查询状态码 - statusCodeQuery := fmt.Sprintf(`pika_monitor_status_code{agent_id="%s",monitor_id="%s"}`, metric.AgentId, metric.MonitorId) - if statusCodeResult, err := s.vmClient.Query(ctx, statusCodeQuery); err == nil && len(statusCodeResult.Data.Result) > 0 { - ts := statusCodeResult.Data.Result[0] - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - metric.StatusCode = int(value) - } - } - } - - // 查询证书信息 - certExpiryQuery := fmt.Sprintf(`pika_monitor_cert_expiry_timestamp_ms{agent_id="%s",monitor_id="%s"}`, metric.AgentId, metric.MonitorId) - if certResult, err := s.vmClient.Query(ctx, certExpiryQuery); err == nil && len(certResult.Data.Result) > 0 { - ts := certResult.Data.Result[0] - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - metric.CertExpiryTime = int64(value) - } - } - } - - certDaysQuery := fmt.Sprintf(`pika_monitor_cert_days_left{agent_id="%s",monitor_id="%s"}`, metric.AgentId, metric.MonitorId) - if certDaysResult, err := s.vmClient.Query(ctx, certDaysQuery); err == nil && len(certDaysResult.Data.Result) > 0 { - ts := certDaysResult.Data.Result[0] - if len(ts.Values) > 0 { - lastValue := ts.Values[len(ts.Values)-1] - if len(lastValue) >= 2 { - valueStr, _ := lastValue[1].(string) - var value float64 - fmt.Sscanf(valueStr, "%f", &value) - metric.CertDaysLeft = int(value) - } - } - } - - return metric + return result } diff --git a/internal/service/monitor_service.go b/internal/service/monitor_service.go index c6b3b33..05fa8e3 100644 --- a/internal/service/monitor_service.go +++ b/internal/service/monitor_service.go @@ -7,6 +7,7 @@ import ( "strings" "time" + "github.com/dushixiang/pika/internal/metric" "github.com/dushixiang/pika/internal/models" "github.com/dushixiang/pika/internal/protocol" "github.com/dushixiang/pika/internal/repo" @@ -264,7 +265,7 @@ func (s *MonitorService) ListByAuth(ctx context.Context, isAuthenticated bool) ( } if len(monitors) == 0 { - emptyResult := []PublicMonitorOverview{} + var emptyResult []PublicMonitorOverview // 缓存空结果 s.overviewCache.Set(cacheKey, emptyResult, 5*time.Minute) return emptyResult, nil @@ -280,7 +281,7 @@ func (s *MonitorService) ListByAuth(ctx context.Context, isAuthenticated bool) ( s.logger.Error("查询 VictoriaMetrics 失败", zap.String("monitorID", monitor.ID), zap.Error(err)) - stats = &MonitorStatsResult{} + stats = &metric.MonitorStatsResult{} } // 将 MonitorStatsResult 转换为 monitorOverviewSummary @@ -495,7 +496,7 @@ func (s *MonitorService) GetMonitorStatsByID(ctx context.Context, monitorID stri if err != nil { s.logger.Error("查询 VictoriaMetrics 失败", zap.String("monitorID", monitorID), zap.Error(err)) // 失败时返回默认值，不中断 - stats = &MonitorStatsResult{} + stats = &metric.MonitorStatsResult{} } // 转换为 monitorOverviewSummary 格式 @@ -516,7 +517,7 @@ func (s *MonitorService) GetMonitorStatsByID(ctx context.Context, monitorID stri // GetMonitorAgentStats 获取监控任务各探针的统计数据（详细列表） // 直接返回 VictoriaMetrics 查询结果，无需额外转换 -func (s *MonitorService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]AgentMonitorStat, error) { +func (s *MonitorService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]protocol.MonitorData, bool) { // 直接返回 VictoriaMetrics 查询结果 return s.metricService.GetMonitorAgentStats(ctx, monitorID) } @@ -524,7 +525,7 @@ func (s *MonitorService) GetMonitorAgentStats(ctx context.Context, monitorID str // GetMonitorHistory 获取监控任务的历史时序数据 // 直接返回 VictoriaMetrics 的原始时序数据，包含所有探针的独立序列 // 支持时间范围：15m, 30m, 1h, 3h, 6h, 12h, 1d, 3d, 7d -func (s *MonitorService) GetMonitorHistory(ctx context.Context, monitorID, timeRange string) (*GetMetricsResponse, error) { +func (s *MonitorService) GetMonitorHistory(ctx context.Context, monitorID, timeRange string) (*metric.GetMetricsResponse, error) { // 计算时间范围 var duration time.Duration switch timeRange { @@ -586,3 +587,42 @@ func (s *MonitorService) clearCache(monitorID string) { s.overviewCache.Delete("overview:public") s.overviewCache.Delete("overview:private") } + +// GetLatestMonitorMetricsByType 获取指定类型的最新监控指标（用于告警检查） +func (s *MonitorService) GetLatestMonitorMetricsByType(ctx context.Context, monitorType string) ([]protocol.MonitorData, error) { + // 查询数据库 + monitorTasks, err := s.FindByEnabledAndType(ctx, true, monitorType) + if err != nil { + return nil, err + } + + // 在缓存中查询最新的监控数据 + var result []protocol.MonitorData + for _, task := range monitorTasks { + monitorData, ok := s.metricService.GetMonitorAgentStats(ctx, task.ID) + if ok { + result = append(result, monitorData...) + } + } + + return result, nil +} + +// GetAllLatestMonitorMetrics 获取所有最新监控指标（用于告警检查） +func (s *MonitorService) GetAllLatestMonitorMetrics(ctx context.Context) ([]protocol.MonitorData, error) { + // 查询所有最新的监控状态 + monitorTasks, err := s.FindByEnabled(ctx, true) + if err != nil { + return nil, err + } + + // 在缓存中查询最新的监控数据 + var result []protocol.MonitorData + for _, task := range monitorTasks { + monitorData, ok := s.metricService.GetMonitorAgentStats(ctx, task.ID) + if ok { + result = append(result, monitorData...) + } + } + return result, nil +} diff --git a/internal/wire_gen.go b/internal/wire_gen.go index 9089732..4a8ca9b 100644 --- a/internal/wire_gen.go +++ b/internal/wire_gen.go @@ -47,7 +47,7 @@ func InitializeApp(logger *zap.Logger, db *gorm.DB, cfg *config.AppConfig) (*App agentHandler := handler.NewAgentHandler(logger, agentService, metricService, monitorService, tamperService, ddnsService, manager) apiKeyHandler := handler.NewApiKeyHandler(logger, apiKeyService) notifier := service.NewNotifier(logger) - alertService := service.NewAlertService(logger, db, propertyService, metricService, notifier) + alertService := service.NewAlertService(logger, db, propertyService, monitorService, notifier) alertHandler := handler.NewAlertHandler(logger, alertService) propertyHandler := handler.NewPropertyHandler(logger, propertyService, notifier) monitorHandler := handler.NewMonitorHandler(logger, monitorService, agentService) diff --git a/pkg/agent/collector/monitor.go b/pkg/agent/collector/monitor.go index b0196d7..bcd2c26 100644 --- a/pkg/agent/collector/monitor.go +++ b/pkg/agent/collector/monitor.go @@ -64,7 +64,7 @@ func (c *MonitorCollector) Collect(items []protocol.MonitorItem) []protocol.Moni result = c.checkICMP(item) default: result = protocol.MonitorData{ - ID: item.ID, + MonitorId: item.ID, Type: item.Type, Target: item.Target, Status: "down", @@ -82,7 +82,7 @@ func (c *MonitorCollector) Collect(items []protocol.MonitorItem) []protocol.Moni // checkHTTP 检查 HTTP/HTTPS 服务 func (c *MonitorCollector) checkHTTP(item protocol.MonitorItem) protocol.MonitorData { result := protocol.MonitorData{ - ID: item.ID, + MonitorId: item.ID, Type: item.Type, Target: item.Target, CheckedAt: time.Now().UnixMilli(), @@ -204,7 +204,7 @@ func (c *MonitorCollector) checkHTTP(item protocol.MonitorItem) protocol.Monitor // checkTCP 检查 TCP 端口 func (c *MonitorCollector) checkTCP(item protocol.MonitorItem) protocol.MonitorData { result := protocol.MonitorData{ - ID: item.ID, + MonitorId: item.ID, Type: item.Type, Target: item.Target, CheckedAt: time.Now().UnixMilli(), @@ -239,7 +239,7 @@ func (c *MonitorCollector) checkTCP(item protocol.MonitorItem) protocol.MonitorD // checkICMP 检查 ICMP (Ping) func (c *MonitorCollector) checkICMP(item protocol.MonitorItem) protocol.MonitorData { result := protocol.MonitorData{ - ID: item.ID, + MonitorId: item.ID, Type: item.Type, Target: item.Target, CheckedAt: time.Now().UnixMilli(),