diff --git a/internal/app.go b/internal/app.go index 977ba8e..fef174f 100644 --- a/internal/app.go +++ b/internal/app.go @@ -90,9 +90,6 @@ func setup(app *orz.App) error { components.MonitorService.SetScheduler(monitorScheduler) monitorScheduler.Start(ctx) - // 启动监控统计计算任务 - go startMonitorStatsCalculation(ctx, components, app.Logger()) - // 启动流量重置检查任务(每小时检查一次) go startTrafficResetCheck(ctx, components, app.Logger()) @@ -305,7 +302,7 @@ func autoMigrate(database *gorm.DB) error { &models.AlertRecord{}, &models.AlertState{}, &models.MonitorTask{}, - &models.MonitorStats{}, + // MonitorStats 已废弃，统计数据从 VictoriaMetrics 查询 &models.TamperProtectConfig{}, &models.TamperEvent{}, &models.TamperAlert{}, @@ -420,35 +417,6 @@ func startMetricsMonitoring(ctx context.Context, components *AppComponents, logg } } -// startMonitorStatsCalculation 启动监控统计计算任务 -func startMonitorStatsCalculation(ctx context.Context, components *AppComponents, logger *zap.Logger) { - logger.Info("启动监控统计计算任务") - - ticker := time.NewTicker(5 * time.Minute) // 每5分钟计算一次统计数据 - defer ticker.Stop() - - // 首次启动时立即计算一次 - if err := components.MonitorService.CalculateMonitorStats(ctx); err != nil { - logger.Error("计算监控统计数据失败", zap.Error(err)) - } else { - logger.Info("监控统计数据计算完成") - } - - for { - select { - case <-ctx.Done(): - logger.Info("监控统计计算任务已停止") - return - case <-ticker.C: - if err := components.MonitorService.CalculateMonitorStats(ctx); err != nil { - logger.Error("计算监控统计数据失败", zap.Error(err)) - } else { - logger.Debug("监控统计数据计算完成") - } - } - } -} - // startTrafficResetCheck 启动流量重置检查定时任务 func startTrafficResetCheck(ctx context.Context, components *AppComponents, logger *zap.Logger) { logger.Info("启动流量重置检查任务") diff --git a/internal/models/monitor_stats.go b/internal/models/monitor_stats.go deleted file mode 100644 index 1e4733b..0000000 --- a/internal/models/monitor_stats.go +++ /dev/null @@ -1,30 +0,0 @@ -package models - -// MonitorStats 监控统计数据 -type MonitorStats struct { - ID string `gorm:"primaryKey" json:"id"` // ID - AgentID string `json:"agentId"` // 探针ID - AgentName string `gorm:"-" json:"agentName,omitempty"` // 探针名称（不存储在数据库，仅用于 API 返回） - MonitorId string `json:"monitorId"` // 监控项ID - MonitorName string `gorm:"-" json:"name"` // 监控项名称（不存储在数据库，仅用于 API 返回） - MonitorType string `json:"type"` // 监控类型 - Target string `json:"target"` // 目标地址 - CurrentResponse int64 `json:"currentResponse"` // 当前响应时间(ms) - AvgResponse24h int64 `json:"avgResponse24h"` // 24小时平均响应时间(ms) - Uptime24h float64 `json:"uptime24h"` // 24小时在线率(百分比) - Uptime7d float64 `json:"uptime7d"` // 7天在线率(百分比) - CertExpiryDate int64 `json:"certExpiryDate"` // 证书过期时间(毫秒时间戳)，0表示无证书 - CertExpiryDays int `json:"certExpiryDays"` // 证书剩余天数 - TotalChecks24h int64 `json:"totalChecks24h"` // 24小时总检测次数 - SuccessChecks24h int64 `json:"successChecks24h"` // 24小时成功次数 - TotalChecks7d int64 `json:"totalChecks7d"` // 7天总检测次数 - SuccessChecks7d int64 `json:"successChecks7d"` // 7天成功次数 - LastCheckTime int64 `json:"lastCheckTime"` // 最后检测时间 - LastCheckStatus string `json:"lastCheckStatus"` // 最后检测状态: up/down - LastCheckError string `json:"lastCheckError"` // 最后检测错误信息 - UpdatedAt int64 `gorm:"autoUpdateTime:milli" json:"updatedAt"` // 更新时间 -} - -func (MonitorStats) TableName() string { - return "monitor_stats" -} diff --git a/internal/repo/metric_repo.go b/internal/repo/metric_repo.go index 70fdedc..13e5054 100644 --- a/internal/repo/metric_repo.go +++ b/internal/repo/metric_repo.go @@ -28,14 +28,6 @@ func (r *MetricRepo) SaveHostMetric(ctx context.Context, metric *models.HostMetr Create(metric).Error } -// GetAvailableNetworkInterfaces 获取探针的可用网卡列表 -// TODO: 需要改为从 VictoriaMetrics 查询 labels -func (r *MetricRepo) GetAvailableNetworkInterfaces(ctx context.Context, agentID string) ([]string, error) { - // 暂时返回空列表，后续改为从 VictoriaMetrics 查询 - // 使用 GET /api/v1/label/interface/values?match[]=pika_network_sent_bytes_rate{agent_id="xxx"} - return []string{}, nil -} - // DeleteAgentMetrics 删除指定探针的所有指标数据 // TODO: 需要改为调用 VictoriaMetrics 的 DELETE API func (r *MetricRepo) DeleteAgentMetrics(ctx context.Context, agentID string) error { @@ -144,50 +136,8 @@ type AggregatedMonitorMetric struct { LastErrorMsg string `json:"lastErrorMsg"` } -// GetMonitorMetrics 获取监控指标历史数据 -// TODO: 需要改为从 VictoriaMetrics 查询 -func (r *MetricRepo) GetMonitorMetrics(ctx context.Context, agentID, monitorID string, start, end int64) ([]MonitorMetric, error) { - // 暂时返回空列表，后续改为从 VictoriaMetrics 查询 - return []MonitorMetric{}, nil -} - -// GetMonitorMetricsByName 获取指定监控项的历史数据 -// TODO: 需要改为从 VictoriaMetrics 查询 -func (r *MetricRepo) GetMonitorMetricsByName(ctx context.Context, agentID, monitorID string, start, end int64, limit int) ([]MonitorMetric, error) { - // 暂时返回空列表，后续改为从 VictoriaMetrics 查询 - return []MonitorMetric{}, nil -} - -// GetLatestMonitorMetricsByType 获取指定类型的最新监控指标（所有探针） -// TODO: 需要改为从 VictoriaMetrics 查询 -func (r *MetricRepo) GetLatestMonitorMetricsByType(ctx context.Context, monitorType string) ([]*MonitorMetric, error) { - // 暂时返回空列表，后续改为从 VictoriaMetrics 查询 - // 使用 GET /api/v1/query?query=pika_monitor_status{type="xxx"} - return []*MonitorMetric{}, nil -} - -// GetAllLatestMonitorMetrics 获取所有最新的监控指标（所有探针的所有监控项） -// TODO: 需要改为从 VictoriaMetrics 查询 -func (r *MetricRepo) GetAllLatestMonitorMetrics(ctx context.Context) ([]*MonitorMetric, error) { - // 暂时返回空列表，后续改为从 VictoriaMetrics 查询 - // 使用 GET /api/v1/query?query=pika_monitor_status - return []*MonitorMetric{}, nil -} - -// DeleteMonitorMetrics 删除指定监控任务的所有指标数据 -// TODO: 需要改为调用 VictoriaMetrics 删除 API -func (r *MetricRepo) DeleteMonitorMetrics(ctx context.Context, monitorID string) error { - // 暂时不删除数据，后续调用 VictoriaMetrics 删除 API - // DELETE /api/v1/admin/tsdb/delete_series?match[]=pika_monitor_*{monitor_id="xxx"} - return nil -} - -// GetMonitorMetricsAgg 获取聚合后的监控指标 -// TODO: 需要改为从 VictoriaMetrics 查询 -func (r *MetricRepo) GetMonitorMetricsAgg(ctx context.Context, monitorID string, start, end int64, bucketSeconds int) ([]AggregatedMonitorMetric, error) { - // 暂时返回空列表，后续从 VictoriaMetrics 查询 - return []AggregatedMonitorMetric{}, nil -} +// 注意：监控相关的查询方法已迁移到 MetricService +// DeleteMonitorMetrics 方法已在 MetricService 中实现 // MonitorMetric 监控指标（与service层定义保持一致） type MonitorMetric struct { diff --git a/internal/repo/monitor_stats_repo.go b/internal/repo/monitor_stats_repo.go deleted file mode 100644 index 094b203..0000000 --- a/internal/repo/monitor_stats_repo.go +++ /dev/null @@ -1,66 +0,0 @@ -package repo - -import ( - "context" - - "github.com/dushixiang/pika/internal/models" - "github.com/go-orz/orz" - "gorm.io/gorm" -) - -type MonitorStatsRepo struct { - orz.Repository[models.MonitorStats, string] - db *gorm.DB -} - -func NewMonitorStatsRepo(db *gorm.DB) *MonitorStatsRepo { - return &MonitorStatsRepo{ - Repository: orz.NewRepository[models.MonitorStats, string](db), - db: db, - } -} - -func (r *MonitorStatsRepo) FindByMonitorId(ctx context.Context, monitorId string) ([]models.MonitorStats, error) { - var statsList []models.MonitorStats - err := r.db.WithContext(ctx). - Where("monitor_id = ?", monitorId). - Find(&statsList).Error - return statsList, err -} - -func (r *MonitorStatsRepo) FindByMonitorIdIn(ctx context.Context, monitorIds []string) ([]models.MonitorStats, error) { - var statsList []models.MonitorStats - err := r.db.WithContext(ctx). - Where("monitor_id in ?", monitorIds). - Find(&statsList).Error - return statsList, err -} - -func (r *MonitorStatsRepo) DeleteByMonitorId(ctx context.Context, monitorId string) error { - return r.db.WithContext(ctx). - Where("monitor_id = ?", monitorId). - Delete(&models.MonitorStats{}).Error -} - -func (r *MonitorStatsRepo) DeleteByAgentId(ctx context.Context, agentId string) error { - return r.db.WithContext(ctx). - Where("agent_id = ?", agentId). - Delete(&models.MonitorStats{}).Error -} - -// FindAll 查找所有统计数据 -func (r *MonitorStatsRepo) FindAll(ctx context.Context) ([]models.MonitorStats, error) { - var statsList []models.MonitorStats - err := r.db.WithContext(ctx).Find(&statsList).Error - return statsList, err -} - -// DeleteByIDs 批量删除指定ID的统计数据 -func (r *MonitorStatsRepo) DeleteByIDs(ctx context.Context, ids []string) error { - if len(ids) == 0 { - return nil - } - return r.db.WithContext(ctx). - Where("id IN ?", ids). - Delete(&models.MonitorStats{}).Error -} diff --git a/internal/service/agent_service.go b/internal/service/agent_service.go index 52c7607..1ca6625 100644 --- a/internal/service/agent_service.go +++ b/internal/service/agent_service.go @@ -19,22 +19,20 @@ import ( type AgentService struct { logger *zap.Logger *orz.Service - AgentRepo *repo.AgentRepo - monitorStatsRepo *repo.MonitorStatsRepo - apiKeyService *ApiKeyService - metricService *MetricService - geoipService *GeoIPService + AgentRepo *repo.AgentRepo + apiKeyService *ApiKeyService + metricService *MetricService + geoipService *GeoIPService } func NewAgentService(logger *zap.Logger, db *gorm.DB, apiKeyService *ApiKeyService, metricService *MetricService, geoipService *GeoIPService) *AgentService { return &AgentService{ - logger: logger, - Service: orz.NewService(db), - AgentRepo: repo.NewAgentRepo(db), - monitorStatsRepo: repo.NewMonitorStatsRepo(db), - apiKeyService: apiKeyService, - metricService: metricService, - geoipService: geoipService, + logger: logger, + Service: orz.NewService(db), + AgentRepo: repo.NewAgentRepo(db), + apiKeyService: apiKeyService, + metricService: metricService, + geoipService: geoipService, } } @@ -350,13 +348,7 @@ func (s *AgentService) DeleteAgent(ctx context.Context, agentID string) error { return err } - // 2. 删除探针的监控统计数据 - if err := s.monitorStatsRepo.DeleteByAgentId(ctx, agentID); err != nil { - s.logger.Error("删除探针监控统计数据失败", zap.String("agentId", agentID), zap.Error(err)) - return err - } - - // 3. 删除探针的审计结果 + // 2. 删除探针的审计结果 if err := s.AgentRepo.DeleteAuditResults(ctx, agentID); err != nil { s.logger.Error("删除探针审计结果失败", zap.String("agentId", agentID), zap.Error(err)) return err diff --git a/internal/service/alert_service.go b/internal/service/alert_service.go index 50a908c..2c55e5a 100644 --- a/internal/service/alert_service.go +++ b/internal/service/alert_service.go @@ -18,19 +18,19 @@ type AlertService struct { AlertRecordRepo *repo.AlertRecordRepo AlertStateRepo *repo.AlertStateRepo agentRepo *repo.AgentRepo - metricRepo *repo.MetricRepo + metricService *MetricService propertyService *PropertyService notifier *Notifier logger *zap.Logger } -func NewAlertService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, notifier *Notifier) *AlertService { +func NewAlertService(logger *zap.Logger, db *gorm.DB, propertyService *PropertyService, metricService *MetricService, notifier *Notifier) *AlertService { return &AlertService{ Service: orz.NewService(db), AlertRecordRepo: repo.NewAlertRecordRepo(db), AlertStateRepo: repo.NewAlertStateRepo(db), agentRepo: repo.NewAgentRepo(db), - metricRepo: repo.NewMetricRepo(db), + metricService: metricService, propertyService: propertyService, notifier: notifier, logger: logger, @@ -373,7 +373,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.metricRepo.GetLatestMonitorMetricsByType(ctx, "http") + monitors, err := s.metricService.GetLatestMonitorMetricsByType(ctx, "http") if err != nil { return err } @@ -396,10 +396,10 @@ func (s *AlertService) checkCertificateAlerts(ctx context.Context, config *model // 检查证书剩余天数是否低于阈值 if certDaysLeft <= config.Rules.CertThreshold && certDaysLeft >= 0 { // 触发告警（证书告警不需要持续时间，直接触发） - s.checkCertAlert(ctx, config, &agent, monitor, certDaysLeft, now) + s.checkCertAlert(ctx, config, &agent, &monitor, certDaysLeft, now) } else { // 恢复告警（如果之前触发过） - s.resolveCertAlert(ctx, config, &agent, monitor, certDaysLeft) + s.resolveCertAlert(ctx, config, &agent, &monitor, certDaysLeft) } } @@ -536,7 +536,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.metricRepo.GetAllLatestMonitorMetrics(ctx) + monitors, err := s.metricService.GetAllLatestMonitorMetrics(ctx) if err != nil { return err } @@ -591,11 +591,11 @@ func (s *AlertService) checkServiceDownAlerts(ctx context.Context, config *model } if shouldFire { - s.fireServiceDownAlert(ctx, config, &agent, monitor, state, now) + s.fireServiceDownAlert(ctx, config, &agent, &monitor, state, now) } if shouldResolve { - s.resolveServiceDownAlert(ctx, config, &agent, monitor, state) + s.resolveServiceDownAlert(ctx, config, &agent, &monitor, state) } } diff --git a/internal/service/metric_service.go b/internal/service/metric_service.go index 737bd60..864c0a8 100644 --- a/internal/service/metric_service.go +++ b/internal/service/metric_service.go @@ -46,12 +46,11 @@ type QueryDefinition struct { // MetricService 指标服务 type MetricService struct { - logger *zap.Logger - metricRepo *repo.MetricRepo - monitorStatsRepo *repo.MonitorStatsRepo - propertyService *PropertyService - trafficService *TrafficService // 流量统计服务 - vmClient *vmclient.VMClient + logger *zap.Logger + metricRepo *repo.MetricRepo + propertyService *PropertyService + trafficService *TrafficService // 流量统计服务 + vmClient *vmclient.VMClient latestCache cache.Cache[string, *LatestMetrics] } @@ -59,13 +58,12 @@ type MetricService struct { // 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), - monitorStatsRepo: repo.NewMonitorStatsRepo(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, *LatestMetrics](time.Minute), } } @@ -709,3 +707,864 @@ 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 { + CurrentResponse int64 // 所有探针平均响应时间(ms) + AvgResponse24h int64 // 24小时平均响应时间(ms) + Uptime24h float64 // 24小时在线率(百分比) + Uptime7d float64 // 7天在线率(百分比) + CertExpiryDate int64 // 最早过期的证书时间(毫秒时间戳) + CertExpiryDays int // 证书剩余天数 + LastCheckTime int64 // 最后检测时间(毫秒时间戳) + LastCheckStatus string // 聚合状态（up/down/unknown） + LastCheckError string // 最后检测错误信息 + AgentCount int // 探针数量 +} + +// AgentMonitorStat 单个探针的监控统计 +type AgentMonitorStat struct { + AgentID string + CurrentResponse int64 + AvgResponse24h int64 + Uptime24h float64 + Uptime7d float64 + LastCheckStatus string + LastCheckError string + LastCheckTime int64 + CertExpiryDate int64 + CertExpiryDays int + TotalChecks24h int64 + SuccessChecks24h int64 + TotalChecks7d int64 + SuccessChecks7d int64 +} + +// 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 MonitorQueryTypeStats24h: + // 24小时统计查询 + queries = []QueryDefinition{ + // 24小时平均响应时间（按探针分组） + {Name: "avg_response_24h", Query: fmt.Sprintf(`avg_over_time(pika_monitor_response_time_ms{monitor_id="%s"}[24h])`, monitorID)}, + // 24小时成功次数 + {Name: "success_count_24h", Query: fmt.Sprintf(`count_over_time(pika_monitor_status{monitor_id="%s",status="up"}[24h:1m])`, monitorID)}, + // 24小时总检测次数 + {Name: "total_count_24h", Query: fmt.Sprintf(`count_over_time(pika_monitor_status{monitor_id="%s"}[24h:1m])`, monitorID)}, + } + + case MonitorQueryTypeStats7d: + // 7天统计查询 + queries = []QueryDefinition{ + // 7天成功次数 + {Name: "success_count_7d", Query: fmt.Sprintf(`count_over_time(pika_monitor_status{monitor_id="%s",status="up"}[7d:5m])`, monitorID)}, + // 7天总检测次数 + {Name: "total_count_7d", Query: fmt.Sprintf(`count_over_time(pika_monitor_status{monitor_id="%s"}[7d:5m])`, monitorID)}, + } + + case MonitorQueryTypeHistory: + // 历史趋势查询（范围查询） + queries = []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) { + // 1. 查询当前状态 + currentQueries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeCurrent) + currentData := make(map[string]*vmclient.QueryResult) + for _, q := range currentQueries { + queryResult, err := s.vmClient.Query(ctx, q.Query) + if err != nil { + s.logger.Warn("查询当前状态失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + currentData[q.Name] = queryResult + } + + // 2. 查询24小时统计 + stats24hQueries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeStats24h) + stats24hData := make(map[string]*vmclient.QueryResult) + for _, q := range stats24hQueries { + queryResult, err := s.vmClient.Query(ctx, q.Query) + if err != nil { + s.logger.Warn("查询24小时统计失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + stats24hData[q.Name] = queryResult + } + + // 3. 查询7天统计 + stats7dQueries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeStats7d) + stats7dData := make(map[string]*vmclient.QueryResult) + for _, q := range stats7dQueries { + queryResult, err := s.vmClient.Query(ctx, q.Query) + if err != nil { + s.logger.Warn("查询7天统计失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + stats7dData[q.Name] = queryResult + } + + // 4. 聚合计算 + return s.aggregateMonitorQueryResults(currentData, stats24hData, stats7dData), nil +} + +// aggregateMonitorQueryResults 聚合多个探针的查询结果 +func (s *MetricService) aggregateMonitorQueryResults( + currentData map[string]*vmclient.QueryResult, + stats24hData map[string]*vmclient.QueryResult, + stats7dData map[string]*vmclient.QueryResult, +) *MonitorStatsResult { + result := &MonitorStatsResult{ + LastCheckStatus: "unknown", + } + + // 解析当前状态数据 + agentStats := make(map[string]*AgentMonitorStat) + + // 处理响应时间 + if respResult, ok := currentData["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].CurrentResponse = int64(value) + agentStats[agentID].LastCheckTime = int64(timestamp * 1000) + } + } + } + } + + // 处理状态 + if statusResult, ok := currentData["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 { + lastValue := ts.Values[len(ts.Values)-1] + if len(lastValue) >= 2 { + valueStr, _ := lastValue[1].(string) + var value float64 + fmt.Sscanf(valueStr, "%f", &value) + + if value > 0 { + agentStats[agentID].LastCheckStatus = "up" + } else { + agentStats[agentID].LastCheckStatus = "down" + } + // 状态标签中也包含了状态信息 + if status != "" { + agentStats[agentID].LastCheckStatus = status + } + } + } + } + } + + // 处理证书信息 + if certDaysResult, ok := currentData["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 := currentData["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) + } + } + } + } + + // 处理24小时统计 + if avgResp24h, ok := stats24hData["avg_response_24h"]; ok && avgResp24h != nil { + for _, ts := range avgResp24h.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].AvgResponse24h = int64(value) + } + } + } + } + + if successCount24h, ok := stats24hData["success_count_24h"]; ok && successCount24h != nil { + for _, ts := range successCount24h.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].SuccessChecks24h = int64(value) + } + } + } + } + + if totalCount24h, ok := stats24hData["total_count_24h"]; ok && totalCount24h != nil { + for _, ts := range totalCount24h.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].TotalChecks24h = int64(value) + } + } + } + } + + // 处理7天统计 + if successCount7d, ok := stats7dData["success_count_7d"]; ok && successCount7d != nil { + for _, ts := range successCount7d.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].SuccessChecks7d = int64(value) + } + } + } + } + + if totalCount7d, ok := stats7dData["total_count_7d"]; ok && totalCount7d != nil { + for _, ts := range totalCount7d.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].TotalChecks7d = int64(value) + } + } + } + } + + // 计算在线率 + for _, stat := range agentStats { + if stat.TotalChecks24h > 0 { + stat.Uptime24h = float64(stat.SuccessChecks24h) / float64(stat.TotalChecks24h) * 100 + } + if stat.TotalChecks7d > 0 { + stat.Uptime7d = float64(stat.SuccessChecks7d) / float64(stat.TotalChecks7d) * 100 + } + } + + // 聚合所有探针的数据 + if len(agentStats) == 0 { + return result + } + + var totalCurrentResponse int64 + var totalAvgResponse24h int64 + var totalUptime24h float64 + var totalUptime7d float64 + var lastCheckTime int64 + hasUp := false + hasDown := false + hasCert := false + var minCertExpiryDate int64 + var minCertExpiryDays int + + for _, stat := range agentStats { + totalCurrentResponse += stat.CurrentResponse + totalAvgResponse24h += stat.AvgResponse24h + totalUptime24h += stat.Uptime24h + totalUptime7d += stat.Uptime7d + + if stat.LastCheckTime > lastCheckTime { + lastCheckTime = stat.LastCheckTime + } + + if stat.LastCheckStatus == "up" { + hasUp = true + } else if stat.LastCheckStatus == "down" { + hasDown = true + } + + if stat.CertExpiryDate > 0 { + if !hasCert || stat.CertExpiryDate < minCertExpiryDate { + minCertExpiryDate = stat.CertExpiryDate + minCertExpiryDays = stat.CertExpiryDays + hasCert = true + } + } + } + + count := len(agentStats) + result.AgentCount = count + if count > 0 { + result.CurrentResponse = totalCurrentResponse / int64(count) + result.AvgResponse24h = totalAvgResponse24h / int64(count) + result.Uptime24h = totalUptime24h / float64(count) + result.Uptime7d = totalUptime7d / float64(count) + } + result.LastCheckTime = lastCheckTime + + // 聚合状态：只要有一个探针 up，整体就是 up + if hasUp { + result.LastCheckStatus = "up" + } else if hasDown { + result.LastCheckStatus = "down" + } + + if hasCert { + result.CertExpiryDate = minCertExpiryDate + result.CertExpiryDays = minCertExpiryDays + } + + return result +} + +// GetMonitorAgentStats 获取监控任务各探针的统计数据 +func (s *MetricService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]AgentMonitorStat, error) { + // 复用 GetMonitorStats 的逻辑，但保留每个探针的独立数据 + // 1. 查询当前状态 + currentQueries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeCurrent) + currentData := make(map[string]*vmclient.QueryResult) + for _, q := range currentQueries { + queryResult, err := s.vmClient.Query(ctx, q.Query) + if err != nil { + s.logger.Warn("查询当前状态失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + currentData[q.Name] = queryResult + } + + // 2. 查询24小时统计 + stats24hQueries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeStats24h) + stats24hData := make(map[string]*vmclient.QueryResult) + for _, q := range stats24hQueries { + queryResult, err := s.vmClient.Query(ctx, q.Query) + if err != nil { + s.logger.Warn("查询24小时统计失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + stats24hData[q.Name] = queryResult + } + + // 3. 查询7天统计 + stats7dQueries := s.buildMonitorPromQLQueries(monitorID, MonitorQueryTypeStats7d) + stats7dData := make(map[string]*vmclient.QueryResult) + for _, q := range stats7dQueries { + queryResult, err := s.vmClient.Query(ctx, q.Query) + if err != nil { + s.logger.Warn("查询7天统计失败", zap.String("query", q.Name), zap.Error(err)) + continue + } + stats7dData[q.Name] = queryResult + } + + // 4. 提取每个探针的数据（参考 aggregateMonitorQueryResults，但不聚合） + agentStatsMap := make(map[string]*AgentMonitorStat) + + // 处理响应时间 + if respResult, ok := currentData["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].CurrentResponse = int64(value) + agentStatsMap[agentID].LastCheckTime = int64(timestamp * 1000) + } + } + } + } + + // 处理状态 + if statusResult, ok := currentData["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 { + lastValue := ts.Values[len(ts.Values)-1] + if len(lastValue) >= 2 { + valueStr, _ := lastValue[1].(string) + var value float64 + fmt.Sscanf(valueStr, "%f", &value) + + if value > 0 { + agentStatsMap[agentID].LastCheckStatus = "up" + } else { + agentStatsMap[agentID].LastCheckStatus = "down" + } + if status != "" { + agentStatsMap[agentID].LastCheckStatus = status + } + } + } + } + } + + // 处理证书信息 + if certDaysResult, ok := currentData["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 := currentData["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) + } + } + } + } + + // 处理24小时统计 + if avgResp24h, ok := stats24hData["avg_response_24h"]; ok && avgResp24h != nil { + for _, ts := range avgResp24h.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].AvgResponse24h = int64(value) + } + } + } + } + + if successCount24h, ok := stats24hData["success_count_24h"]; ok && successCount24h != nil { + for _, ts := range successCount24h.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].SuccessChecks24h = int64(value) + } + } + } + } + + if totalCount24h, ok := stats24hData["total_count_24h"]; ok && totalCount24h != nil { + for _, ts := range totalCount24h.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].TotalChecks24h = int64(value) + } + } + } + } + + // 处理7天统计 + if successCount7d, ok := stats7dData["success_count_7d"]; ok && successCount7d != nil { + for _, ts := range successCount7d.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].SuccessChecks7d = int64(value) + } + } + } + } + + if totalCount7d, ok := stats7dData["total_count_7d"]; ok && totalCount7d != nil { + for _, ts := range totalCount7d.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].TotalChecks7d = int64(value) + } + } + } + } + + // 计算在线率 + for _, stat := range agentStatsMap { + if stat.TotalChecks24h > 0 { + stat.Uptime24h = float64(stat.SuccessChecks24h) / float64(stat.TotalChecks24h) * 100 + } + if stat.TotalChecks7d > 0 { + stat.Uptime7d = float64(stat.SuccessChecks7d) / float64(stat.TotalChecks7d) * 100 + } + } + + // 转换为数组 + 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 +} diff --git a/internal/service/monitor_service.go b/internal/service/monitor_service.go index af18dff..e9564c8 100644 --- a/internal/service/monitor_service.go +++ b/internal/service/monitor_service.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "strings" "time" @@ -13,7 +14,6 @@ import ( ws "github.com/dushixiang/pika/internal/websocket" "github.com/go-orz/cache" "github.com/go-orz/orz" - "github.com/go-orz/toolkit" "github.com/google/uuid" "go.uber.org/zap" "gorm.io/datatypes" @@ -24,16 +24,13 @@ type MonitorService struct { logger *zap.Logger *repo.MonitorRepo *orz.Service - agentRepo *repo.AgentRepo - metricRepo *repo.MetricRepo - metricService *MetricService - monitorStatsRepo *repo.MonitorStatsRepo - wsManager *ws.Manager + agentRepo *repo.AgentRepo + metricRepo *repo.MetricRepo + metricService *MetricService + wsManager *ws.Manager // 监控概览缓存：缓存监控任务列表（使用不同的 key 区分 public 和 private） overviewCache cache.Cache[string, []PublicMonitorOverview] - // 监控统计缓存：缓存单个监控任务的统计数据 - statsCache cache.Cache[string, []models.MonitorStats] // 调度器引用（用于动态管理任务） scheduler MonitorScheduler @@ -48,18 +45,16 @@ type MonitorScheduler interface { func NewMonitorService(logger *zap.Logger, db *gorm.DB, metricService *MetricService, wsManager *ws.Manager) *MonitorService { return &MonitorService{ - logger: logger, - Service: orz.NewService(db), - MonitorRepo: repo.NewMonitorRepo(db), - agentRepo: repo.NewAgentRepo(db), - metricRepo: repo.NewMetricRepo(db), - metricService: metricService, - monitorStatsRepo: repo.NewMonitorStatsRepo(db), - wsManager: wsManager, + logger: logger, + Service: orz.NewService(db), + MonitorRepo: repo.NewMonitorRepo(db), + agentRepo: repo.NewAgentRepo(db), + metricRepo: repo.NewMetricRepo(db), + metricService: metricService, + wsManager: wsManager, // 缓存 5 分钟，避免频繁查询 overviewCache: cache.New[string, []PublicMonitorOverview](5 * time.Minute), - statsCache: cache.New[string, []models.MonitorStats](5 * time.Minute), } } @@ -229,12 +224,6 @@ func (s *MonitorService) DeleteMonitor(ctx context.Context, id string) error { return err } - // 删除监控统计数据 - if err := s.monitorStatsRepo.DeleteByMonitorId(ctx, id); err != nil { - s.logger.Error("删除监控统计数据失败", zap.String("monitorId", id), zap.Error(err)) - return err - } - // 删除监控指标数据（从 VictoriaMetrics） if err := s.metricService.DeleteMonitorMetrics(ctx, id); err != nil { s.logger.Error("删除监控指标数据失败", zap.String("monitorId", id), zap.Error(err)) @@ -286,29 +275,32 @@ func (s *MonitorService) ListByAuth(ctx context.Context, isAuthenticated bool) ( return emptyResult, nil } - // 提取监控任务ID列表 - monitorIds := make([]string, 0, len(monitors)) - for _, monitor := range monitors { - monitorIds = append(monitorIds, monitor.ID) - } - - // 批量获取统计数据（已经由后台任务预计算好） - statsList, err := s.monitorStatsRepo.FindByMonitorIdIn(ctx, monitorIds) - if err != nil { - return nil, err - } - - // 按监控任务ID分组统计数据 - statsMap := make(map[string][]models.MonitorStats, len(monitors)) - for _, stats := range statsList { - statsMap[stats.MonitorId] = append(statsMap[stats.MonitorId], stats) - } - // 构建监控概览列表 items := make([]PublicMonitorOverview, 0, len(monitors)) for _, monitor := range monitors { - // 直接聚合预计算的统计数据 - summary := aggregateMonitorStats(statsMap[monitor.ID]) + // 从 VictoriaMetrics 查询统计数据 + stats, err := s.metricService.GetMonitorStats(ctx, monitor.ID) + if err != nil { + // 查询失败时使用默认值 + s.logger.Error("查询 VictoriaMetrics 失败", + zap.String("monitorID", monitor.ID), + zap.Error(err)) + stats = &MonitorStatsResult{} + } + + // 将 MonitorStatsResult 转换为 monitorOverviewSummary + summary := monitorOverviewSummary{ + AgentCount: stats.AgentCount, + CurrentResponse: stats.CurrentResponse, + AvgResponse24h: stats.AvgResponse24h, + Uptime24h: stats.Uptime24h, + Uptime7d: stats.Uptime7d, + LastCheckStatus: stats.LastCheckStatus, + LastCheckError: stats.LastCheckError, + LastCheckTime: stats.LastCheckTime, + CertExpiryDate: stats.CertExpiryDate, + CertExpiryDays: stats.CertExpiryDays, + } // 构建监控概览对象 item := s.buildMonitorOverview(monitor, summary) @@ -365,88 +357,6 @@ type monitorOverviewSummary struct { LastCheckTime int64 } -func aggregateMonitorStats(stats []models.MonitorStats) monitorOverviewSummary { - summary := monitorOverviewSummary{ - LastCheckStatus: "unknown", - } - - if len(stats) == 0 { - return summary - } - - var totalCurrentResponse int64 - var totalAvgResponse24h int64 - var totalUptime24h float64 - var totalUptime7d float64 - var lastCheckTime int64 - var lastCheckError string - var certExpiryDate int64 - var certExpiryDays int - hasCert := false - hasUp := false - hasDown := false - - for _, stat := range stats { - totalCurrentResponse += stat.CurrentResponse - totalAvgResponse24h += stat.AvgResponse24h - totalUptime24h += stat.Uptime24h - totalUptime7d += stat.Uptime7d - - // 记录最新的检测时间和对应的错误信息 - if stat.LastCheckTime > lastCheckTime { - lastCheckTime = stat.LastCheckTime - // 如果这是最新的检测且状态为 down，记录错误信息 - if stat.LastCheckStatus == "down" { - lastCheckError = stat.LastCheckError - } else { - lastCheckError = "" - } - } - - switch stat.LastCheckStatus { - case "up": - hasUp = true - case "down": - hasDown = true - } - - if stat.CertExpiryDate > 0 { - if !hasCert || stat.CertExpiryDate < certExpiryDate { - certExpiryDate = stat.CertExpiryDate - certExpiryDays = stat.CertExpiryDays - hasCert = true - } - } - } - - count := len(stats) - summary.AgentCount = count - if count > 0 { - summary.CurrentResponse = totalCurrentResponse / int64(count) - summary.AvgResponse24h = totalAvgResponse24h / int64(count) - summary.Uptime24h = totalUptime24h / float64(count) - summary.Uptime7d = totalUptime7d / float64(count) - } - summary.LastCheckTime = lastCheckTime - summary.LastCheckError = lastCheckError - - switch { - case hasUp: - summary.LastCheckStatus = "up" - case hasDown: - summary.LastCheckStatus = "down" - default: - summary.LastCheckStatus = "unknown" - } - - if hasCert { - summary.CertExpiryDate = certExpiryDate - summary.CertExpiryDays = certExpiryDays - } - - return summary -} - func cloneAgentIDs(ids datatypes.JSONSlice[string]) []string { if len(ids) == 0 { return []string{} @@ -596,140 +506,6 @@ func (s *MonitorService) SendMonitorTaskToAgents(ctx context.Context, monitor mo return nil } -// CalculateMonitorStats 计算监控统计数据 -func (s *MonitorService) CalculateMonitorStats(ctx context.Context) error { - now := time.Now() - - // 获取所有启用的监控任务 - monitors, err := s.MonitorRepo.FindByEnabled(ctx, true) - if err != nil { - return err - } - - // 获取所有探针（包括在线和离线） - agents, err := s.agentRepo.FindAll(ctx) - if err != nil { - return err - } - - // 收集所有有效的监控任务ID - validStatsIDs := make(map[string]bool) - - // 为每个监控任务的每个探针计算统计数据 - for _, monitor := range monitors { - // 使用统一的方法计算目标探针 - targetAgents := s.resolveTargetAgents(monitor, agents) - - for _, agent := range targetAgents { - stats, err := s.calculateStatsForAgentMonitor(ctx, agent.ID, monitor.ID, monitor.Type, monitor.Target, now) - if err != nil { - s.logger.Error("计算监控统计失败", - zap.String("agentID", agent.ID), - zap.String("monitorName", monitor.Name), - zap.Error(err)) - continue - } - - // 记录有效的监控任务ID - validStatsIDs[stats.ID] = true - - if err := s.monitorStatsRepo.Save(ctx, stats); err != nil { - s.logger.Error("保存监控统计失败", - zap.String("agentID", agent.ID), - zap.String("monitorName", monitor.Name), - zap.Error(err)) - } - } - } - - // 清理无效监控任务的统计数据 - if err := s.cleanupInvalidStats(ctx, validStatsIDs); err != nil { - s.logger.Error("清理无效统计数据失败", zap.Error(err)) - // 不返回错误，继续运行 - } - - return nil -} - -// calculateStatsForAgentMonitor 计算单个探针单个监控任务的统计数据 -func (s *MonitorService) calculateStatsForAgentMonitor(ctx context.Context, agentID, monitorId, monitorType, target string, now time.Time) (*models.MonitorStats, error) { - stats := &models.MonitorStats{ - ID: toolkit.Sign("monitor_stats", agentID, monitorId, monitorType, target), - AgentID: agentID, - MonitorId: monitorId, - MonitorType: monitorType, - Target: target, - } - - // 计算24小时数据 - start24h := now.Add(-24 * time.Hour).UnixMilli() - end := now.UnixMilli() - metrics24h, err := s.metricRepo.GetMonitorMetrics(ctx, agentID, monitorId, start24h, end) - if err != nil { - return nil, err - } - - // 计算7天数据 - start7d := now.Add(-7 * 24 * time.Hour).UnixMilli() - metrics7d, err := s.metricRepo.GetMonitorMetrics(ctx, agentID, monitorId, start7d, end) - if err != nil { - return nil, err - } - - // 计算24小时统计 - if len(metrics24h) > 0 { - var totalResponse int64 - var successCount int64 - lastMetric := metrics24h[len(metrics24h)-1] - - for _, metric := range metrics24h { - if metric.Status == "up" { - successCount++ - totalResponse += metric.ResponseTime - } - } - - stats.TotalChecks24h = int64(len(metrics24h)) - stats.SuccessChecks24h = successCount - if successCount > 0 { - stats.AvgResponse24h = totalResponse / successCount - } - if stats.TotalChecks24h > 0 { - stats.Uptime24h = float64(successCount) / float64(stats.TotalChecks24h) * 100 - } - - // 最后一次检测数据 - stats.CurrentResponse = lastMetric.ResponseTime - stats.LastCheckTime = lastMetric.Timestamp - stats.LastCheckStatus = lastMetric.Status - stats.LastCheckError = lastMetric.Error - - // 从最新的检测结果中获取证书信息 - if lastMetric.CertExpiryTime > 0 { - stats.CertExpiryDate = lastMetric.CertExpiryTime - stats.CertExpiryDays = lastMetric.CertDaysLeft - } - } - - // 计算7天统计 - if len(metrics7d) > 0 { - var successCount int64 - for _, metric := range metrics7d { - if metric.Status == "up" { - successCount++ - } - } - - stats.TotalChecks7d = int64(len(metrics7d)) - stats.SuccessChecks7d = successCount - if stats.TotalChecks7d > 0 { - stats.Uptime7d = float64(successCount) / float64(stats.TotalChecks7d) * 100 - } - } - - return stats, nil -} - // GetMonitorStatsByID 获取监控任务的统计数据（聚合后的单个监控详情） func (s *MonitorService) GetMonitorStatsByID(ctx context.Context, monitorID string) (*PublicMonitorOverview, error) { // 查询监控任务 @@ -738,14 +514,27 @@ func (s *MonitorService) GetMonitorStatsByID(ctx context.Context, monitorID stri return nil, err } - // 获取统计数据 - statsList, err := s.monitorStatsRepo.FindByMonitorId(ctx, monitor.ID) + // 从 VictoriaMetrics 查询统计数据 + stats, err := s.metricService.GetMonitorStats(ctx, monitorID) if err != nil { - return nil, err + s.logger.Error("查询 VictoriaMetrics 失败", zap.String("monitorID", monitorID), zap.Error(err)) + // 失败时返回默认值，不中断 + stats = &MonitorStatsResult{} } - // 聚合统计数据 - summary := aggregateMonitorStats(statsList) + // 转换为 monitorOverviewSummary 格式 + summary := monitorOverviewSummary{ + AgentCount: stats.AgentCount, + LastCheckStatus: stats.LastCheckStatus, + LastCheckError: stats.LastCheckError, + CurrentResponse: stats.CurrentResponse, + AvgResponse24h: stats.AvgResponse24h, + Uptime24h: stats.Uptime24h, + Uptime7d: stats.Uptime7d, + CertExpiryDate: stats.CertExpiryDate, + CertExpiryDays: stats.CertExpiryDays, + LastCheckTime: stats.LastCheckTime, + } // 构建监控概览对象 overview := s.buildMonitorOverview(monitor, summary) @@ -753,121 +542,198 @@ func (s *MonitorService) GetMonitorStatsByID(ctx context.Context, monitorID stri return &overview, nil } +type MonitorStats struct { + ID string `gorm:"primaryKey" json:"id"` // ID + AgentID string `json:"agentId"` // 探针ID + AgentName string `gorm:"-" json:"agentName,omitempty"` // 探针名称（不存储在数据库，仅用于 API 返回） + MonitorId string `json:"monitorId"` // 监控项ID + MonitorName string `gorm:"-" json:"name"` // 监控项名称（不存储在数据库，仅用于 API 返回） + MonitorType string `json:"type"` // 监控类型 + Target string `json:"target"` // 目标地址 + CurrentResponse int64 `json:"currentResponse"` // 当前响应时间(ms) + AvgResponse24h int64 `json:"avgResponse24h"` // 24小时平均响应时间(ms) + Uptime24h float64 `json:"uptime24h"` // 24小时在线率(百分比) + Uptime7d float64 `json:"uptime7d"` // 7天在线率(百分比) + CertExpiryDate int64 `json:"certExpiryDate"` // 证书过期时间(毫秒时间戳)，0表示无证书 + CertExpiryDays int `json:"certExpiryDays"` // 证书剩余天数 + TotalChecks24h int64 `json:"totalChecks24h"` // 24小时总检测次数 + SuccessChecks24h int64 `json:"successChecks24h"` // 24小时成功次数 + TotalChecks7d int64 `json:"totalChecks7d"` // 7天总检测次数 + SuccessChecks7d int64 `json:"successChecks7d"` // 7天成功次数 + LastCheckTime int64 `json:"lastCheckTime"` // 最后检测时间 + LastCheckStatus string `json:"lastCheckStatus"` // 最后检测状态: up/down + LastCheckError string `json:"lastCheckError"` // 最后检测错误信息 + UpdatedAt int64 `gorm:"autoUpdateTime:milli" json:"updatedAt"` // 更新时间 +} + // GetMonitorAgentStats 获取监控任务各探针的统计数据（详细列表） -func (s *MonitorService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]models.MonitorStats, error) { - // 构建缓存键 - cacheKey := fmt.Sprintf("agents:%s", monitorID) - - // 尝试从缓存获取 - if cachedResult, ok := s.statsCache.Get(cacheKey); ok { - return cachedResult, nil - } - - // 缓存未命中，查询数据库 +func (s *MonitorService) GetMonitorAgentStats(ctx context.Context, monitorID string) ([]MonitorStats, error) { + // 查询监控任务 monitor, err := s.MonitorRepo.FindById(ctx, monitorID) if err != nil { return nil, err } - statsList, err := s.monitorStatsRepo.FindByMonitorId(ctx, monitor.ID) + // 从 VictoriaMetrics 查询各探针统计 + agentStats, err := s.metricService.GetMonitorAgentStats(ctx, monitorID) if err != nil { + s.logger.Error("查询 VictoriaMetrics 失败", zap.String("monitorID", monitorID), zap.Error(err)) return nil, err } - // 获取探针列表 - agents, err := s.agentRepo.FindAll(ctx) - if err != nil { - return nil, err - } - // 当前监控任务的关联的探针 - targetAgents := s.resolveTargetAgents(monitor, agents) - var targetAgentsMap = make(map[string]string) - for _, agent := range targetAgents { - targetAgentsMap[agent.ID] = agent.Name + // 填充探针名称、监控名称 + agentMap := make(map[string]*models.Agent) + if len(agentStats) > 0 { + agentIDs := make([]string, 0, len(agentStats)) + for _, stat := range agentStats { + agentIDs = append(agentIDs, stat.AgentID) + } + agents, _ := s.agentRepo.FindByIdIn(ctx, agentIDs) + for _, agent := range agents { + agentMap[agent.ID] = &agent + } } - // 填充监控名称、探针名称和隐私设置 - var filteredStatsList []models.MonitorStats - for _, stats := range statsList { - agentName, ok := targetAgentsMap[stats.AgentID] - if !ok { - continue + // 确定目标 target 值（是否隐藏） + target := monitor.Target + if !monitor.ShowTargetPublic { + target = "******" + } + + // 转换为 MonitorStats 格式 + result := make([]MonitorStats, 0, len(agentStats)) + for _, stat := range agentStats { + ms := MonitorStats{ + AgentID: stat.AgentID, + AgentName: "", + MonitorId: monitorID, + MonitorName: monitor.Name, + MonitorType: monitor.Type, + Target: target, + CurrentResponse: stat.CurrentResponse, + AvgResponse24h: stat.AvgResponse24h, + Uptime24h: stat.Uptime24h, + Uptime7d: stat.Uptime7d, + CertExpiryDate: stat.CertExpiryDate, + CertExpiryDays: stat.CertExpiryDays, + TotalChecks24h: stat.TotalChecks24h, + SuccessChecks24h: stat.SuccessChecks24h, + TotalChecks7d: stat.TotalChecks7d, + SuccessChecks7d: stat.SuccessChecks7d, + LastCheckTime: stat.LastCheckTime, + LastCheckStatus: stat.LastCheckStatus, + LastCheckError: stat.LastCheckError, } - stats.MonitorName = monitor.Name - - // 根据 ShowTargetPublic 字段决定是否隐藏 Target - if !monitor.ShowTargetPublic { - stats.Target = "******" + if agent, ok := agentMap[stat.AgentID]; ok { + ms.AgentName = agent.Name } - stats.AgentName = agentName - filteredStatsList = append(filteredStatsList, stats) + result = append(result, ms) } - // 缓存结果 - s.statsCache.Set(cacheKey, filteredStatsList, 5*time.Minute) - - return filteredStatsList, nil + return result, nil } // GetMonitorHistory 获取监控任务的历史响应时间数据 // 根据前端时间范围选择，直接使用聚合表数据 // 支持时间范围：15m, 30m, 1h, 3h, 6h, 12h, 1d, 3d, 7d func (s *MonitorService) GetMonitorHistory(ctx context.Context, monitorID, timeRange string) ([]repo.AggregatedMonitorMetric, error) { - monitor, err := s.MonitorRepo.FindById(ctx, monitorID) - if err != nil { + // 验证监控任务存在 + if _, err := s.MonitorRepo.FindById(ctx, monitorID); err != nil { return nil, err } - // 根据前端时间范围，选择最优的聚合粒度 + // 计算时间范围 var duration time.Duration - var bucketSeconds int // 使用的聚合 bucket（秒） - switch timeRange { case "15m": duration = 15 * time.Minute - bucketSeconds = 60 // 1分钟聚合，约15个点 case "30m": duration = 30 * time.Minute - bucketSeconds = 60 // 1分钟聚合，约30个点 case "1h": duration = 1 * time.Hour - bucketSeconds = 300 // 5分钟聚合，约12个点 case "3h": duration = 3 * time.Hour - bucketSeconds = 300 // 5分钟聚合，约36个点 case "6h": duration = 6 * time.Hour - bucketSeconds = 900 // 15分钟聚合，约24个点 case "12h": duration = 12 * time.Hour - bucketSeconds = 900 // 15分钟聚合，约48个点 case "1d", "24h": duration = 24 * time.Hour - bucketSeconds = 1800 // 30分钟聚合，约48个点 case "3d": duration = 3 * 24 * time.Hour - bucketSeconds = 3600 // 1小时聚合，约72个点 case "7d": duration = 7 * 24 * time.Hour - bucketSeconds = 7200 // 2小时聚合，约84个点 default: - // 默认 15 分钟 duration = 15 * time.Minute - bucketSeconds = 60 } now := time.Now() end := now.UnixMilli() start := now.Add(-duration).UnixMilli() - // 对齐时间范围到 bucket 边界 - bucketMs := int64(bucketSeconds * 1000) - start, end = alignTimeRangeToBucket(start, end, bucketMs) + // 从 VictoriaMetrics 查询历史数据 + metricsResp, err := s.metricService.GetMonitorHistory(ctx, monitorID, start, end) + if err != nil { + s.logger.Error("查询 VictoriaMetrics 失败", zap.String("monitorID", monitorID), zap.Error(err)) + return []repo.AggregatedMonitorMetric{}, nil + } - // 直接使用聚合表查询 - return s.metricRepo.GetMonitorMetricsAgg(ctx, monitor.ID, start, end, bucketSeconds) + // 转换为 AggregatedMonitorMetric 格式 + result := make([]repo.AggregatedMonitorMetric, 0) + + // 按时间戳分组，聚合多个探针的数据 + timestampMap := make(map[int64]map[string]float64) // timestamp -> agentID -> avgResponse + + for _, series := range metricsResp.Series { + // 从 series.Labels 中提取 agent_id + agentID := "" + if series.Labels != nil { + agentID = series.Labels["agent_id"] + } + if agentID == "" && len(metricsResp.Series) > 0 { + // 如果没有 agent_id 标签，尝试从 series name 中提取 + // 这是 response_time 系列，每个探针一条时间序列 + // series.Name 可能包含探针信息 + agentID = series.Name + } + + for _, dataPoint := range series.Data { + if timestampMap[dataPoint.Timestamp] == nil { + timestampMap[dataPoint.Timestamp] = make(map[string]float64) + } + timestampMap[dataPoint.Timestamp][agentID] = dataPoint.Value + } + } + + // 将 map 转换为数组，并计算每个时间点的聚合数据 + for timestamp, agentData := range timestampMap { + var totalResponse float64 + var count int + for _, avgResp := range agentData { + totalResponse += avgResp + count++ + } + + var avgResponse float64 + if count > 0 { + avgResponse = totalResponse / float64(count) + } + + result = append(result, repo.AggregatedMonitorMetric{ + Timestamp: timestamp, + AgentID: "", // 聚合数据不指定单个 agentID + AvgResponse: avgResponse, + }) + } + + // 按时间戳排序 + sort.Slice(result, func(i, j int) bool { + return result[i].Timestamp < result[j].Timestamp + }) + + return result, nil } // GetMonitorByAuth 根据认证状态获取监控任务（已登录返回全部，未登录返回公开可见） @@ -897,37 +763,4 @@ func (s *MonitorService) clearCache(monitorID string) { // 清理概览缓存 s.overviewCache.Delete("overview:public") s.overviewCache.Delete("overview:private") - - // 清理统计缓存 - s.statsCache.Delete(fmt.Sprintf("agents:%s", monitorID)) -} - -// cleanupInvalidStats 按监控任务维度清理无效的统计数据 -// 删除不在有效监控任务列表中的所有统计数据（监控任务被禁用或删除） -func (s *MonitorService) cleanupInvalidStats(ctx context.Context, validStatsIDs map[string]bool) error { - // 获取所有现有的统计数据 - allStats, err := s.monitorStatsRepo.FindAll(ctx) - if err != nil { - return err - } - - // 收集需要删除的统计数据ID - idsToDelete := make([]string, 0) - for _, stats := range allStats { - // 如果该统计数据对应的监控任务不在有效列表中，删除该统计数据 - if !validStatsIDs[stats.ID] { - idsToDelete = append(idsToDelete, stats.ID) - } - } - - // 批量删除无效的统计数据 - if len(idsToDelete) > 0 { - s.logger.Info("清理无效的监控统计数据", - zap.Int("count", len(idsToDelete))) - if err := s.monitorStatsRepo.DeleteByIDs(ctx, idsToDelete); err != nil { - return err - } - } - - return nil } diff --git a/internal/wire_gen.go b/internal/wire_gen.go index 81fbacc..9089732 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, notifier) + alertService := service.NewAlertService(logger, db, propertyService, metricService, notifier) alertHandler := handler.NewAlertHandler(logger, alertService) propertyHandler := handler.NewPropertyHandler(logger, propertyService, notifier) monitorHandler := handler.NewMonitorHandler(logger, monitorService, agentService)