diff --git a/internal/handler/monitor_handler.go b/internal/handler/monitor_handler.go index f18c28d..29ca6fb 100644 --- a/internal/handler/monitor_handler.go +++ b/internal/handler/monitor_handler.go @@ -202,7 +202,7 @@ func (h *MonitorHandler) GetHistoryByID(c echo.Context) error { return orz.NewError(400, err.Error()) } - history, err := h.monitorService.GetMonitorHistory(ctx, id, start, end, aggregation) + history, err := h.metricService.GetMonitorHistory(ctx, id, start, end, aggregation) if err != nil { return err } diff --git a/internal/models/monitor_task.go b/internal/models/monitor_task.go index 2d9f18f..be38bba 100644 --- a/internal/models/monitor_task.go +++ b/internal/models/monitor_task.go @@ -18,7 +18,6 @@ type MonitorTask struct { Interval int `json:"interval"` // 检测频率（秒），默认 60 AgentIds datatypes.JSONSlice[string] `json:"agentIds"` // 指定的探针 ID 列表（JSON 数组） AgentNames []string `gorm:"-" json:"agentNames"` // 指定的探针名称列表 - Tags datatypes.JSONSlice[string] `json:"tags"` // 指定的标签列表（JSON 数组），拥有这些标签的探针都会执行此监控 HTTPConfig datatypes.JSONType[protocol.HTTPMonitorConfig] `json:"httpConfig"` // HTTP 监控配置 TCPConfig datatypes.JSONType[protocol.TCPMonitorConfig] `json:"tcpConfig"` // TCP 监控配置 ICMPConfig datatypes.JSONType[protocol.ICMPMonitorConfig] `json:"icmpConfig"` // ICMP 监控配置 diff --git a/internal/service/metric_service.go b/internal/service/metric_service.go index bede616..2afa171 100644 --- a/internal/service/metric_service.go +++ b/internal/service/metric_service.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "slices" "strconv" "time" @@ -296,6 +297,38 @@ func (s *MetricService) GetMetrics(ctx context.Context, agentID, metricType stri }, nil } +// CleanMonitorCache 清理监控任务缓存中不再关联的探针数据 +func (s *MetricService) CleanMonitorCache(ctx context.Context, monitorID string) error { + // 从缓存读取监控数据 + latestMetrics, ok := s.monitorLatestCache.Get(monitorID) + if !ok { + // 缓存不存在，无需清理 + return nil + } + + // 查询监控任务配置 + monitorTask, err := s.monitorRepo.FindById(ctx, monitorID) + if err != nil { + return err + } + + // 只在有过滤条件（指定了 AgentIds）时清理缓存 + if len(monitorTask.AgentIds) > 0 { + // 遍历缓存中的探针，移除不再关联的探针数据 + for agentId := range latestMetrics.Agents.Keys() { + if !slices.Contains(monitorTask.AgentIds, agentId) { + // 该探针已不再关联到此监控任务，从缓存中移除 + latestMetrics.Agents.Delete(agentId) + s.logger.Debug("从监控缓存中移除探针", + zap.String("monitorID", monitorID), + zap.String("agentID", agentId)) + } + } + } + + return nil +} + // updateMonitorCache 更新监控数据缓存 func (s *MetricService) updateMonitorCache(agentID string, monitorData *protocol.MonitorData, timestamp int64) { monitorID := monitorData.MonitorId @@ -572,6 +605,13 @@ func (s *MetricService) buildMonitorPromQLQueries(monitorID string, aggregation // GetMonitorHistory 获取监控任务的历史趋势数据 func (s *MetricService) GetMonitorHistory(ctx context.Context, monitorID string, start, end int64, aggregation string) (*metric.GetMetricsResponse, error) { + // 查询监控任务配置 + monitorTask, err := s.monitorRepo.FindById(ctx, monitorID) + if err != nil { + s.logger.Error("查询监控任务失败", zap.String("monitorID", monitorID), zap.Error(err)) + return nil, err + } + step := vmclient.AutoStep(time.UnixMilli(start), time.UnixMilli(end)) queries := s.buildMonitorPromQLQueries(monitorID, aggregation, step) @@ -592,11 +632,26 @@ func (s *MetricService) GetMonitorHistory(ctx context.Context, monitorID string, series = append(series, convertedSeries...) } - // 收集所有 agent_id + // 过滤掉已取消关联的 agent 数据（仅在有过滤条件时） agentIdSet := make(map[string]struct{}) - for _, s := range series { - if agentId, ok := s.Labels["agent_id"]; ok { - agentIdSet[agentId] = struct{}{} + if len(monitorTask.AgentIds) > 0 { + // 有过滤条件，只保留当前关联的 agent 数据 + filteredSeries := make([]metric.Series, 0) + for _, s := range series { + if agentId, ok := s.Labels["agent_id"]; ok { + if slices.Contains(monitorTask.AgentIds, agentId) { + filteredSeries = append(filteredSeries, s) + agentIdSet[agentId] = struct{}{} + } + } + } + series = filteredSeries + } else { + // 无过滤条件，收集所有 agent_id + for _, s := range series { + if agentId, ok := s.Labels["agent_id"]; ok { + agentIdSet[agentId] = struct{}{} + } } } @@ -645,14 +700,31 @@ func (s *MetricService) GetMonitorAgentStats(monitorID string) []protocol.Monito return []protocol.MonitorData{} } - // 收集所有 agentId - agentIds := make([]string, 0, latestMetrics.Agents.Len()) - for agentId := range latestMetrics.Agents.Keys() { - agentIds = append(agentIds, agentId) + // 查询监控任务配置 + ctx := context.Background() + monitorTask, err := s.monitorRepo.FindById(ctx, monitorID) + if err != nil { + s.logger.Error("查询监控任务失败", zap.String("monitorID", monitorID), zap.Error(err)) + return []protocol.MonitorData{} + } + + // 收集所有当前关联的 agentId（从缓存中过滤） + agentIds := make([]string, 0) + if len(monitorTask.AgentIds) > 0 { + // 有过滤条件，只保留匹配的 agent + for agentId := range latestMetrics.Agents.Keys() { + if slices.Contains(monitorTask.AgentIds, agentId) { + agentIds = append(agentIds, agentId) + } + } + } else { + // 无过滤条件，返回所有缓存中的 agent + for agentId := range latestMetrics.Agents.Keys() { + agentIds = append(agentIds, agentId) + } } // 查询 agent 信息 - ctx := context.Background() agents, err := s.agentRepo.FindByIdIn(ctx, agentIds) if err != nil { s.logger.Error("查询 agent 信息失败", zap.Error(err)) @@ -665,10 +737,20 @@ func (s *MetricService) GetMonitorAgentStats(monitorID string) []protocol.Monito } // 转换为数组并填充 agent 名称 - result := make([]protocol.MonitorData, 0, latestMetrics.Agents.Len()) + result := make([]protocol.MonitorData, 0, len(agentIds)) for stat := range latestMetrics.Agents.Values() { - stat.AgentName = agentNameMap[stat.AgentId] // 填充 agent 名称 - result = append(result, *stat) + // 根据过滤条件决定是否包含该 agent + if len(monitorTask.AgentIds) > 0 { + // 有过滤条件，只返回当前关联的 agent 数据 + if slices.Contains(monitorTask.AgentIds, stat.AgentId) { + stat.AgentName = agentNameMap[stat.AgentId] // 填充 agent 名称 + result = append(result, *stat) + } + } else { + // 无过滤条件，返回所有 agent 数据 + stat.AgentName = agentNameMap[stat.AgentId] // 填充 agent 名称 + result = append(result, *stat) + } } return result @@ -685,12 +767,22 @@ func (s *MetricService) GetMonitorStats(monitorID string) *metric.MonitorStatsRe } } + // 查询监控任务配置 + ctx := context.Background() + monitorTask, err := s.monitorRepo.FindById(ctx, monitorID) + if err != nil { + s.logger.Error("查询监控任务失败", zap.String("monitorID", monitorID), zap.Error(err)) + return &metric.MonitorStatsResult{ + Status: "unknown", + } + } + // 聚合各探针数据 - return s.aggregateMonitorStats(latestMetrics) + return s.aggregateMonitorStats(latestMetrics, monitorTask.AgentIds) } // aggregateMonitorStats 聚合各探针的监控数据 -func (s *MetricService) aggregateMonitorStats(latestMetrics *metric.LatestMonitorMetrics) *metric.MonitorStatsResult { +func (s *MetricService) aggregateMonitorStats(latestMetrics *metric.LatestMonitorMetrics, agentIds []string) *metric.MonitorStatsResult { result := &metric.MonitorStatsResult{ Status: "unknown", } @@ -704,11 +796,21 @@ func (s *MetricService) aggregateMonitorStats(latestMetrics *metric.LatestMonito var maxResponseTime int64 var lastCheckTime int64 var upCount, downCount, unknownCount int + var validCount int // 实际聚合的探针数量 hasCert := false var minCertExpiryTime int64 var minCertDaysLeft int for stat := range latestMetrics.Agents.Values() { + // 根据过滤条件决定是否聚合该探针 + if len(agentIds) > 0 { + // 有过滤条件，只聚合当前关联的探针数据 + if !slices.Contains(agentIds, stat.AgentId) { + continue + } + } + + validCount++ totalResponseTime += stat.ResponseTime // 计算响应时间的最小值和最大值 @@ -742,10 +844,9 @@ func (s *MetricService) aggregateMonitorStats(latestMetrics *metric.LatestMonito } } - count := latestMetrics.Agents.Len() - result.AgentCount = count - if count > 0 { - result.ResponseTime = totalResponseTime / int64(count) + result.AgentCount = validCount + if validCount > 0 { + result.ResponseTime = totalResponseTime / int64(validCount) } result.ResponseTimeMin = minResponseTime result.ResponseTimeMax = maxResponseTime diff --git a/internal/service/monitor_service.go b/internal/service/monitor_service.go index b84eaac..9f45a83 100644 --- a/internal/service/monitor_service.go +++ b/internal/service/monitor_service.go @@ -68,7 +68,6 @@ type MonitorTaskRequest struct { TCPConfig protocol.TCPMonitorConfig `json:"tcpConfig,omitempty"` ICMPConfig protocol.ICMPMonitorConfig `json:"icmpConfig,omitempty"` AgentIds []string `json:"agentIds,omitempty"` - Tags []string `json:"tags"` } func (s *MonitorService) CreateMonitor(ctx context.Context, req *MonitorTaskRequest) (*models.MonitorTask, error) { @@ -95,7 +94,6 @@ func (s *MonitorService) CreateMonitor(ctx context.Context, req *MonitorTaskRequ Visibility: visibility, Interval: interval, AgentIds: datatypes.JSONSlice[string](req.AgentIds), - Tags: datatypes.JSONSlice[string](req.Tags), HTTPConfig: datatypes.NewJSONType(req.HTTPConfig), TCPConfig: datatypes.NewJSONType(req.TCPConfig), ICMPConfig: datatypes.NewJSONType(req.ICMPConfig), @@ -136,7 +134,6 @@ func (s *MonitorService) UpdateMonitor(ctx context.Context, id string, req *Moni task.Description = req.Description task.ShowTargetPublic = req.ShowTargetPublic task.Visibility = req.Visibility - task.Tags = req.Tags // 更新检测频率 interval := req.Interval @@ -154,6 +151,16 @@ func (s *MonitorService) UpdateMonitor(ctx context.Context, id string, req *Moni return nil, err } + // 清理监控缓存中不再关联的探针数据 + if s.metricService != nil { + if err := s.metricService.CleanMonitorCache(ctx, id); err != nil { + s.logger.Warn("清理监控缓存失败", + zap.String("monitorID", id), + zap.Error(err)) + // 不返回错误，继续执行 + } + } + // 更新调度器 if s.scheduler != nil { // 如果从禁用变为启用，或者间隔时间改变 @@ -255,57 +262,6 @@ func (s *MonitorService) buildMonitorOverview(monitor models.MonitorTask, stats return overview } -// resolveTargetAgents 计算监控任务对应的目标探针范围 -// 规则： -// 1. 如果既没有指定 AgentIds 也没有指定 Tags，返回所有传入的探针（全部节点） -// 2. 如果指定了 AgentIds 或 Tags（或两者都指定），则返回匹配的探针（自动去重） -// - AgentIds: 直接匹配探针 ID -// - Tags: 匹配探针标签中包含任意一个指定标签的探针 -// - 两者结果取并集 -func (s *MonitorService) resolveTargetAgents(monitor models.MonitorTask, availableAgents []models.Agent) []models.Agent { - // 如果既没有指定 AgentIds 也没有指定 Tags，使用所有可用探针 - if len(monitor.AgentIds) == 0 && len(monitor.Tags) == 0 { - return availableAgents - } - - // 使用 map 来去重 - targetAgentIDSet := make(map[string]struct{}) - - // 1. 处理通过 AgentIds 指定的探针 - if len(monitor.AgentIds) > 0 { - for _, agentID := range monitor.AgentIds { - targetAgentIDSet[agentID] = struct{}{} - } - } - - // 2. 处理通过 Tags 指定的探针 - if len(monitor.Tags) > 0 { - for _, agent := range availableAgents { - if agent.Tags != nil && len(agent.Tags) > 0 { - // 检查探针的标签中是否包含任何一个指定的标签 - for _, agentTag := range agent.Tags { - for _, monitorTag := range monitor.Tags { - if agentTag == monitorTag { - targetAgentIDSet[agent.ID] = struct{}{} - break - } - } - } - } - } - } - - // 3. 根据去重后的 ID 集合筛选探针 - targetAgents := make([]models.Agent, 0, len(targetAgentIDSet)) - for _, agent := range availableAgents { - if _, ok := targetAgentIDSet[agent.ID]; ok { - targetAgents = append(targetAgents, agent) - } - } - - return targetAgents -} - // sendMonitorConfigToAgent 向指定探针发送监控配置（内部方法） func (s *MonitorService) sendMonitorConfigToAgent(agentID string, payload protocol.MonitorConfigPayload) error { msgData, err := json.Marshal(protocol.OutboundMessage{ @@ -321,25 +277,17 @@ func (s *MonitorService) sendMonitorConfigToAgent(agentID string, payload protoc // SendMonitorTaskToAgents 向指定探针发送单个监控任务（公开方法） func (s *MonitorService) SendMonitorTaskToAgents(ctx context.Context, monitor models.MonitorTask) error { - // 实时获取所有在线探针，避免依赖数据库状态 - onlineIDs := s.wsManager.GetAllClients() - if len(onlineIDs) == 0 { - return nil + // 确定目标探针 ID 列表 + var targetAgentIDs []string + if len(monitor.AgentIds) == 0 { + // 没有指定探针，向所有在线探针发送 + targetAgentIDs = s.wsManager.GetAllClients() + } else { + // 指定了探针 + targetAgentIDs = monitor.AgentIds } - // 查询在线探针的详细信息 - onlineAgents, err := s.agentRepo.ListByIDs(ctx, onlineIDs) - if err != nil { - s.logger.Error("获取在线探针信息失败", zap.Error(err)) - return err - } - if len(onlineAgents) == 0 { - return nil - } - - // 使用统一的方法计算目标探针 - targetAgents := s.resolveTargetAgents(monitor, onlineAgents) - if len(targetAgents) == 0 { + if len(targetAgentIDs) == 0 { return nil } @@ -368,12 +316,12 @@ func (s *MonitorService) SendMonitorTaskToAgents(ctx context.Context, monitor mo } // 向每个目标探针发送 - for _, agent := range targetAgents { - if err := s.sendMonitorConfigToAgent(agent.ID, payload); err != nil { + for _, agentID := range targetAgentIDs { + if err := s.sendMonitorConfigToAgent(agentID, payload); err != nil { s.logger.Error("发送监控配置失败", zap.String("taskID", monitor.ID), zap.String("taskName", monitor.Name), - zap.String("agentID", agent.ID), + zap.String("agentID", agentID), zap.Error(err)) } } @@ -397,12 +345,6 @@ func (s *MonitorService) GetMonitorStatsByID(ctx context.Context, monitorID stri return &overview, nil } -// GetMonitorHistory 获取监控任务的历史时序数据 -// 直接返回 VictoriaMetrics 的原始时序数据，包含所有探针的独立序列 -func (s *MonitorService) GetMonitorHistory(ctx context.Context, monitorID string, start, end int64, aggregation string) (*metric.GetMetricsResponse, error) { - return s.metricService.GetMonitorHistory(ctx, monitorID, start, end, aggregation) -} - // GetMonitorByAuth 根据认证状态获取监控任务（已登录返回全部，未登录返回公开可见） func (s *MonitorService) GetMonitorByAuth(ctx context.Context, id string, isAuthenticated bool) (*models.MonitorTask, error) { if isAuthenticated {