aiops/backend/pkg/aiopscontext/service.go

package aiopscontext

import (
	"context"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"ezops/model"

	"gorm.io/gorm"
)

// Service 汇聚 AIOps 平台上下文，后续可复用为 MCP Server 的数据源。 // Logan.Li
type Service struct {
	db *gorm.DB
}

// NewService 创建上下文服务。 // Logan.Li
func NewService(db *gorm.DB) *Service {
	return &Service{db: db}
}

// Summary 返回平台总览上下文。 // Logan.Li
func (s *Service) Summary(ctx context.Context) (*Summary, error) {
	out := Summary{
		HighLoadHosts: []AssetSummary{},
		RecentAlerts:  []AlertSummary{},
	}
	db := s.db.WithContext(ctx)
	db.Model(&model.Server{}).Count(&out.ServerTotal)
	db.Model(&model.Server{}).Where("agent_status = ? OR status = ?", "online", "running").Count(&out.ServerOnline)
	db.Model(&model.KubernetesCluster{}).Count(&out.ClusterTotal)
	db.Model(&model.AlertEvent{}).Where("status = ?", "firing").Count(&out.FiringAlerts)
	db.Model(&model.AgentAlert{}).Where("resolved = ?", false).Count(&out.UnresolvedAlerts)

	assets, err := s.SearchAssets(ctx, "", 8)
	if err == nil {
		for _, a := range assets {
			if a.CPUUsage >= 80 || a.MemoryUsage >= 80 || a.DiskUsage >= 85 {
				out.HighLoadHosts = append(out.HighLoadHosts, a)
			}
		}
	}
	out.RecentAlerts, _ = s.ListAlerts(ctx, AlertQuery{Limit: 10})
	out.GeneratedAt = time.Now()
	return &out, nil
}

// SearchAssets 查询资产摘要，默认按更新时间倒序返回。 // Logan.Li
func (s *Service) SearchAssets(ctx context.Context, keyword string, limit int) ([]AssetSummary, error) {
	if limit <= 0 || limit > 50 {
		limit = 10
	}
	var servers []model.Server
	q := s.db.WithContext(ctx).Model(&model.Server{})
	kw := strings.TrimSpace(keyword)
	if kw != "" {
		like := "%" + kw + "%"
		q = q.Where("hostname LIKE ? OR ip LIKE ? OR public_ip LIKE ? OR tags LIKE ? OR description LIKE ?",
			like, like, like, like, like)
	}
	if err := q.Order("updated_at DESC").Limit(limit).Find(&servers).Error; err != nil {
		return nil, err
	}
	out := make([]AssetSummary, 0, len(servers))
	for _, srv := range servers {
		out = append(out, assetFromServer(srv))
	}
	return out, nil
}

// ListAlerts 查询统一告警摘要。 // Logan.Li
func (s *Service) ListAlerts(ctx context.Context, query AlertQuery) ([]AlertSummary, error) {
	limit := query.Limit
	if limit <= 0 || limit > 100 {
		limit = 20
	}
	since := sinceByRange(query.TimeRange, 24*time.Hour)
	targetLike := "%" + strings.TrimSpace(query.Target) + "%"

	var events []model.AlertEvent
	q := s.db.WithContext(ctx).Model(&model.AlertEvent{}).Where("created_at >= ?", since)
	if query.Target != "" {
		q = q.Where("target LIKE ? OR target_name LIKE ? OR target_ip LIKE ? OR rule_name LIKE ?",
			targetLike, targetLike, targetLike, targetLike)
	}
	if query.Severity != "" {
		q = q.Where("severity = ?", query.Severity)
	}
	if query.Status != "" {
		q = q.Where("status = ?", query.Status)
	}
	if err := q.Order("created_at DESC").Limit(limit).Find(&events).Error; err != nil {
		return nil, err
	}

	out := make([]AlertSummary, 0, len(events)+limit)
	for _, e := range events {
		out = append(out, alertFromEvent(e))
	}

	var agentAlerts []model.AgentAlert
	aq := s.db.WithContext(ctx).Model(&model.AgentAlert{}).Where("created_at >= ?", since)
	if query.Target != "" {
		aq = aq.Where("hostname LIKE ? OR rule_name LIKE ?", targetLike, targetLike)
	}
	if query.Severity != "" {
		aq = aq.Where("severity = ?", query.Severity)
	}
	if query.Status == "firing" || query.Status == "active" {
		aq = aq.Where("resolved = ?", false)
	} else if query.Status == "resolved" {
		aq = aq.Where("resolved = ?", true)
	}
	if err := aq.Order("created_at DESC").Limit(limit).Find(&agentAlerts).Error; err == nil {
		for _, a := range agentAlerts {
			out = append(out, alertFromAgent(a))
		}
	}

	sort.SliceStable(out, func(i, j int) bool {
		return out[i].StartedAt.After(out[j].StartedAt)
	})
	if len(out) > limit {
		out = out[:limit]
	}
	return out, nil
}

// MetricSummary 查询主机指标摘要。 // Logan.Li
func (s *Service) MetricSummary(ctx context.Context, target, metric, timeRange string, limit int) (*MetricSummary, error) {
	hostname, err := s.resolveHostname(ctx, target)
	if err != nil {
		return nil, err
	}
	if metric == "" {
		metric = "cpu_usage"
	}
	if limit <= 0 || limit > 500 {
		limit = 240
	}
	since := sinceByRange(timeRange, time.Hour)
	var reports []model.AgentReport
	if err := s.db.WithContext(ctx).Where("hostname = ? AND report_time >= ?", hostname, since).
		Order("report_time ASC").Limit(limit).Find(&reports).Error; err != nil {
		return nil, err
	}
	out := &MetricSummary{Target: target, Hostname: hostname, Metric: metric, TimeRange: normalizeRangeLabel(timeRange), Points: []MetricPoint{}}
	values := make([]float64, 0, len(reports))
	for _, r := range reports {
		v, ok := metricValue(r, metric)
		if !ok {
			continue
		}
		values = append(values, v)
		if len(out.Points) < 60 || len(reports) <= 60 {
			out.Points = append(out.Points, MetricPoint{Time: r.ReportTime, Value: round2(v)})
		}
	}
	fillMetricStats(out, values)
	return out, nil
}

// BuildPack 构建面向 LLM 的上下文包。 // Logan.Li
func (s *Service) BuildPack(ctx context.Context, req ContextPackRequest) (*ContextPack, error) {
	if req.Limit <= 0 {
		req.Limit = 10
	}
	scope := strings.TrimSpace(req.Target)
	var primaryAlert *AlertSummary
	if req.AlertID > 0 {
		alert, err := s.GetAlert(ctx, req.AlertID)
		if err == nil {
			primaryAlert = alert
			if scope == "" {
				scope = alert.Target
			}
		}
	}
	pack := &ContextPack{
		Scope: scope, TimeRange: normalizeRangeLabel(req.TimeRange),
		Assets: []AssetSummary{}, Alerts: []AlertSummary{}, Metrics: []MetricSummary{},
		GeneratedAt: time.Now(), Evidence: []EvidenceItem{},
	}
	if scope != "" {
		pack.Assets, _ = s.SearchAssets(ctx, scope, req.Limit)
		for _, asset := range pack.Assets {
			pack.Evidence = append(pack.Evidence, EvidenceItem{
				Type: "asset", Source: "asset_servers",
				Summary: fmt.Sprintf("资产 %s ip=%s protocol=%s os=%s status=%s agent=%s",
					asset.Hostname, asset.IP, asset.Protocol, asset.OSType, asset.Status, asset.AgentStatus),
				Confidence: 0.9,
			})
		}
	}
	pack.Alerts, _ = s.ListAlerts(ctx, AlertQuery{Target: scope, TimeRange: req.TimeRange, Limit: req.Limit})
	if primaryAlert != nil {
		pack.Alerts = prependAlertIfMissing(pack.Alerts, *primaryAlert)
	}
	metrics := req.Metrics
	if len(metrics) == 0 {
		metrics = []string{"cpu_usage", "mem_usage", "load1"}
	}
	for _, metric := range metrics {
		if scope == "" {
			break
		}
		ms, err := s.MetricSummary(ctx, scope, metric, req.TimeRange, 240)
		if err == nil && ms.SampleSize > 0 {
			pack.Metrics = append(pack.Metrics, *ms)
			pack.Evidence = append(pack.Evidence, EvidenceItem{
				Type: "metric", Source: "monitor_agent_reports",
				Summary: fmt.Sprintf("%s 最近 %s: latest=%.2f avg=%.2f p95=%.2f max=%.2f trend=%s",
					ms.Metric, ms.TimeRange, ms.Latest, ms.Avg, ms.P95, ms.Max, ms.Trend),
				Confidence: 0.82,
			})
		}
	}
	for _, a := range pack.Alerts {
		pack.Evidence = append(pack.Evidence, EvidenceItem{
			Type: "alert", Source: a.Source,
			Summary: fmt.Sprintf("[%s|%s] %s target=%s value=%.2f threshold=%.2f",
				a.Severity, a.Status, a.RuleName, a.Target, a.Value, a.Threshold),
			Confidence: 0.78,
		})
	}
	return pack, nil
}

// GetAlert 获取单条平台告警摘要。 // Logan.Li
func (s *Service) GetAlert(ctx context.Context, id uint) (*AlertSummary, error) {
	var event model.AlertEvent
	if err := s.db.WithContext(ctx).Take(&event, id).Error; err == nil {
		alert := alertFromEvent(event)
		return &alert, nil
	}
	var agentAlert model.AgentAlert
	if err := s.db.WithContext(ctx).Take(&agentAlert, id).Error; err != nil {
		return nil, err
	}
	alert := alertFromAgent(agentAlert)
	return &alert, nil
}

func (s *Service) resolveHostname(ctx context.Context, target string) (string, error) {
	target = strings.TrimSpace(target)
	if target == "" {
		return "", fmt.Errorf("target 不能为空")
	}
	var srv model.Server
	err := s.db.WithContext(ctx).Where("hostname = ? OR ip = ? OR public_ip = ?", target, target, target).Take(&srv).Error
	if err == nil {
		return srv.Hostname, nil
	}
	if err == gorm.ErrRecordNotFound {
		return target, nil
	}
	return "", err
}

func assetFromServer(srv model.Server) AssetSummary {
	return AssetSummary{
		ID: srv.ID, Hostname: srv.Hostname, IP: srv.IP, PublicIP: srv.PublicIP,
		Protocol: srv.Protocol, OSType: srv.OSType, OS: srv.OS, Status: srv.Status,
		AgentStatus: srv.AgentStatus, CPUUsage: srv.CPUUsage, MemoryUsage: srv.MemoryUsage,
		DiskUsage: srv.DiskUsage, Tags: srv.Tags, Description: srv.Description,
		LastCheckAt: srv.LastCheckAt, UpdatedAt: srv.UpdatedAt,
	}
}

func alertFromEvent(e model.AlertEvent) AlertSummary {
	target := e.Target
	if target == "" {
		target = e.TargetName
	}
	if target == "" {
		target = e.TargetIP
	}
	started := e.StartedAt
	if started.IsZero() {
		started = e.CreatedAt
	}
	return AlertSummary{
		ID: e.ID, Source: "monitor_alert_events", RuleName: firstNonEmpty(e.RuleName, e.Title),
		Severity: e.Severity, Status: e.Status, Target: target, Metric: e.MetricName,
		Value: e.Value, Threshold: e.Threshold, Message: e.Message, StartedAt: started,
		ResolvedAt: e.ResolvedAt,
	}
}

func alertFromAgent(a model.AgentAlert) AlertSummary {
	status := "firing"
	if a.Resolved {
		status = "resolved"
	}
	return AlertSummary{
		ID: a.ID, Source: "monitor_agent_alerts", RuleName: a.RuleName,
		Severity: a.Severity, Status: status, Target: a.Hostname, Metric: a.Metric,
		Value: a.Value, Threshold: a.Threshold, Message: a.Message,
		StartedAt: a.CreatedAt, ResolvedAt: a.ResolvedAt,
	}
}

func prependAlertIfMissing(alerts []AlertSummary, alert AlertSummary) []AlertSummary {
	for _, item := range alerts {
		if item.Source == alert.Source && item.ID == alert.ID {
			return alerts
		}
	}
	return append([]AlertSummary{alert}, alerts...)
}

func metricValue(r model.AgentReport, metric string) (float64, bool) {
	switch metric {
	case "cpu", "cpu_usage":
		return r.CPUUsage, true
	case "cpu_iowait":
		return r.CPUIOwait, true
	case "mem", "memory", "mem_usage", "memory_usage":
		return r.MemUsage, true
	case "mem_available":
		return r.MemAvailable, true
	case "swap", "mem_swap_used":
		return r.MemSwapUsed, true
	case "load1":
		return r.Load1, true
	case "load5":
		return r.Load5, true
	case "load15":
		return r.Load15, true
	default:
		return 0, false
	}
}

func fillMetricStats(out *MetricSummary, values []float64) {
	out.SampleSize = len(values)
	if len(values) == 0 {
		out.Trend = "unknown"
		return
	}
	minV, maxV, sum := values[0], values[0], 0.0
	for _, v := range values {
		if v < minV {
			minV = v
		}
		if v > maxV {
			maxV = v
		}
		sum += v
	}
	sorted := append([]float64(nil), values...)
	sort.Float64s(sorted)
	p95Idx := int(math.Ceil(float64(len(sorted))*0.95)) - 1
	if p95Idx < 0 {
		p95Idx = 0
	}
	out.Latest = round2(values[len(values)-1])
	out.Min = round2(minV)
	out.Max = round2(maxV)
	out.Avg = round2(sum / float64(len(values)))
	out.P95 = round2(sorted[p95Idx])
	out.Trend = trendOf(values)
}

func trendOf(values []float64) string {
	if len(values) < 4 {
		return "stable"
	}
	n := len(values)
	head := avg(values[:n/3])
	tail := avg(values[n-n/3:])
	if tail-head > 5 {
		return "up"
	}
	if head-tail > 5 {
		return "down"
	}
	return "stable"
}

func avg(values []float64) float64 {
	if len(values) == 0 {
		return 0
	}
	sum := 0.0
	for _, v := range values {
		sum += v
	}
	return sum / float64(len(values))
}

func sinceByRange(label string, fallback time.Duration) time.Time {
	switch strings.ToLower(strings.TrimSpace(label)) {
	case "15m":
		return time.Now().Add(-15 * time.Minute)
	case "30m":
		return time.Now().Add(-30 * time.Minute)
	case "2h":
		return time.Now().Add(-2 * time.Hour)
	case "6h":
		return time.Now().Add(-6 * time.Hour)
	case "24h", "1d":
		return time.Now().Add(-24 * time.Hour)
	case "7d":
		return time.Now().Add(-7 * 24 * time.Hour)
	default:
		return time.Now().Add(-fallback)
	}
}

func normalizeRangeLabel(label string) string {
	if strings.TrimSpace(label) == "" {
		return "1h"
	}
	return label
}

func firstNonEmpty(values ...string) string {
	for _, v := range values {
		if strings.TrimSpace(v) != "" {
			return v
		}
	}
	return ""
}

func round2(v float64) float64 {
	return math.Round(v*100) / 100
}