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

重构服务监控数据

Browse Source
This commit is contained in:
dushixiang committed 2025-12-12 22:36:15 +08:00
1 parent 3207259de0
commit e2b876f82e
9 files changed
+1085 -579
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/app.go
+1 -33
View File
@@ -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("启动流量重置检查任务")
internal/models/monitor_stats.go
-30
View File
@@ -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"
}
internal/repo/metric_repo.go
+2 -52
View File
@@ -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 {
internal/repo/monitor_stats_repo.go
-66
View File
@@ -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
}
internal/service/agent_service.go
+11 -19
View File
@@ -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
internal/service/alert_service.go
+9 -9
View File
@@ -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)
}
}
internal/service/metric_service.go
+872 -13
View File
@@ -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
}
internal/service/monitor_service.go
+189 -356
View File
@@ -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
}
internal/wire_gen.go
+1 -1
View File
@@ -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)