aiops/backend/controller/middleware_monitor.go

package controller

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"ezops/model"
	"ezops/pkg/systask"

	_ "github.com/go-sql-driver/mysql"
	"github.com/gin-gonic/gin"
	"gorm.io/gorm"
)

// ============================================
// 中间件监控 (Redis / MySQL)
// 后端直连采集模式 — 无需 Agent，平台直接连接 Redis/MySQL 采集指标
// ============================================

// MiddlewareConfig 存储在 MonitorTarget.Config JSON 字段中
type MiddlewareConfig struct {
	Password  string `json:"password,omitempty"`
	Username  string `json:"username,omitempty"`  // MySQL 专用
	Database  string `json:"database,omitempty"`  // MySQL 专用
	RedisMode string `json:"redis_mode,omitempty"` // Redis: standalone / cluster
}

// parseMiddlewareConfig 解析中间件配置并补全默认值 // Logan.Li
func parseMiddlewareConfig(configJSON string) MiddlewareConfig {
	var cfg MiddlewareConfig
	if configJSON != "" {
		json.Unmarshal([]byte(configJSON), &cfg)
	}
	if cfg.RedisMode == "" {
		cfg.RedisMode = "standalone"
	}
	return cfg
}

// middlewareConfigPublic 返回给前端的脱敏配置 // Logan.Li
type middlewareConfigPublic struct {
	Username  string `json:"username,omitempty"`
	Database  string `json:"database,omitempty"`
	RedisMode string `json:"redis_mode,omitempty"`
}

// MiddlewareMonitorHandler 中间件监控 handler
type MiddlewareMonitorHandler struct {
	db     *gorm.DB
	stopCh chan struct{}
	wg     sync.WaitGroup
}

// NewMiddlewareMonitorHandler 创建 handler 并启动后台采集
func NewMiddlewareMonitorHandler(db *gorm.DB) *MiddlewareMonitorHandler {
	h := &MiddlewareMonitorHandler{
		db:     db,
		stopCh: make(chan struct{}),
	}
	return h
}

// StartCollector 启动后台指标采集 goroutine
func (h *MiddlewareMonitorHandler) StartCollector(intervalSec int) {
	if intervalSec < 10 {
		intervalSec = 30
	}
	h.wg.Add(1)
	go func() {
		defer h.wg.Done()
		ticker := time.NewTicker(time.Duration(intervalSec) * time.Second)
		defer ticker.Stop()
		log.Printf("[MiddlewareMonitor] 采集引擎启动，间隔 %ds", intervalSec)
		// 首次立即采集一次
		h.collectAllWithRecord()
		for {
			select {
			case <-ticker.C:
				h.collectAllWithRecord()
			case <-h.stopCh:
				log.Printf("[MiddlewareMonitor] 采集引擎停止")
				return
			}
		}
	}()
}

// StopCollector 停止后台采集
func (h *MiddlewareMonitorHandler) StopCollector() {
	close(h.stopCh)
	h.wg.Wait()
}

// ============================================
// CRUD 接口
// ============================================

// CreateMiddlewareTarget 创建中间件监控目标
// POST /api/v1/middleware-monitors
func (h *MiddlewareMonitorHandler) CreateMiddlewareTarget(c *gin.Context) {
	var req struct {
		Name      string `json:"name" binding:"required"`
		Type      string `json:"type" binding:"required"` // redis / mysql
		IP        string `json:"ip" binding:"required"`
		Port      int    `json:"port" binding:"required"`
		Password  string `json:"password"`
		Username  string `json:"username"`   // MySQL
		Database  string `json:"database"`   // MySQL
		RedisMode string `json:"redis_mode"` // Redis: standalone / cluster
		Interval  int    `json:"interval"`   // 采集间隔，默认 30s
	}
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()})
		return
	}

	// 验证类型
	mwType := strings.ToLower(req.Type)
	if mwType != "redis" && mwType != "mysql" {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "type 必须是 redis 或 mysql"})
		return
	}

	if req.Port == 0 {
		if mwType == "redis" {
			req.Port = 6379
		} else {
			req.Port = 3306
		}
	}
	if req.Interval == 0 {
		req.Interval = 30
	}

	redisMode := strings.ToLower(req.RedisMode)
	if mwType == "redis" {
		if redisMode != "cluster" {
			redisMode = "standalone"
		}
	} else {
		redisMode = ""
	}

	// 构建 Config JSON
	cfg := MiddlewareConfig{
		Password:  req.Password,
		Username:  req.Username,
		Database:  req.Database,
		RedisMode: redisMode,
	}
	cfgJSON, _ := json.Marshal(cfg)

	target := model.MonitorTarget{
		Name:        req.Name,
		Type:        model.TargetTypeMiddleware,
		Status:      model.TargetStatusUnknown,
		IP:          req.IP,
		Port:        req.Port,
		CollectMode: "api",
		Config:      string(cfgJSON),
		Interval:    req.Interval,
		Description: mwType, // 复用 Description 字段存储中间件子类型
	}

	// host_group_id 默认 0 会触发外键约束，创建时显式忽略 // Logan.Li
	if err := h.db.Omit("HostGroupID").Create(&target).Error; err != nil {
		c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "创建失败: " + err.Error()})
		return
	}

	// 立即采集一次
	go h.collectOne(target.ID)

	c.JSON(http.StatusOK, model.Response{Code: 200, Data: target})
}

// ListMiddlewareTargets 列出中间件监控目标
// GET /api/v1/middleware-monitors
func (h *MiddlewareMonitorHandler) ListMiddlewareTargets(c *gin.Context) {
	var targets []model.MonitorTarget
	q := h.db.Model(&model.MonitorTarget{}).Where("type = ?", model.TargetTypeMiddleware)

	if kw := c.Query("keyword"); kw != "" {
		q = q.Where("name LIKE ? OR ip LIKE ?", "%"+kw+"%", "%"+kw+"%")
	}
	if subType := c.Query("sub_type"); subType != "" {
		q = q.Where("description = ?", subType)
	}

	var total int64
	q.Count(&total)

	page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
	pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "100"))
	q.Offset((page - 1) * pageSize).Limit(pageSize).Order("id DESC").Find(&targets)

	if targets == nil {
		targets = []model.MonitorTarget{}
	}

	// 附带最新指标（Config 脱敏，不返回密码）
	type TargetWithMetrics struct {
		model.MonitorTarget
		SubType       string                 `json:"sub_type"`
		LatestMetrics map[string]float64     `json:"latest_metrics"`
		ConfigPublic  middlewareConfigPublic `json:"config_public"`
	}
	result := make([]TargetWithMetrics, len(targets))
	now := time.Now().Unix()
	for i, t := range targets {
		var histories []model.MetricHistory
		h.db.Where("target_id = ? AND timestamp > ?", t.ID, now-600).
			Order("timestamp DESC").Limit(20).Find(&histories)

		metrics := make(map[string]float64)
		for _, mh := range histories {
			if _, ok := metrics[mh.MetricName]; !ok {
				metrics[mh.MetricName] = mh.Value
			}
		}
		cfg := parseMiddlewareConfig(t.Config)
		// 列表响应隐藏原始 config 字段，避免泄露密码 // Logan.Li
		t.Config = ""
		result[i] = TargetWithMetrics{
			MonitorTarget: t,
			SubType:       t.Description,
			LatestMetrics: metrics,
			ConfigPublic: middlewareConfigPublic{
				Username:  cfg.Username,
				Database:  cfg.Database,
				RedisMode: cfg.RedisMode,
			},
		}
	}

	c.JSON(http.StatusOK, model.Response{Code: 200, Data: gin.H{
		"list":  result,
		"total": total,
	}})
}

// GetMiddlewareTarget 获取单个中间件监控目标
// GET /api/v1/middleware-monitors/:id
func (h *MiddlewareMonitorHandler) GetMiddlewareTarget(c *gin.Context) {
	id, err := strconv.Atoi(c.Param("id"))
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的 ID"})
		return
	}
	var target model.MonitorTarget
	if err := h.db.First(&target, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "目标不存在"})
		return
	}
	c.JSON(http.StatusOK, model.Response{Code: 200, Data: target})
}

// UpdateMiddlewareTarget 更新中间件监控目标
// PUT /api/v1/middleware-monitors/:id
func (h *MiddlewareMonitorHandler) UpdateMiddlewareTarget(c *gin.Context) {
	id, err := strconv.Atoi(c.Param("id"))
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的 ID"})
		return
	}
	var target model.MonitorTarget
	if err := h.db.First(&target, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "目标不存在"})
		return
	}

	var req struct {
		Name      *string `json:"name"`
		IP        *string `json:"ip"`
		Port      *int    `json:"port"`
		Password  *string `json:"password"`
		Username  *string `json:"username"`
		Database  *string `json:"database"`
		RedisMode *string `json:"redis_mode"`
		Interval  *int    `json:"interval"`
	}
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()})
		return
	}

	updates := map[string]interface{}{}
	if req.Name != nil {
		updates["name"] = *req.Name
	}
	if req.IP != nil {
		updates["ip"] = *req.IP
	}
	if req.Port != nil {
		updates["port"] = *req.Port
	}
	if req.Interval != nil {
		updates["interval"] = *req.Interval
	}

	// 如果有连接配置变更，更新 Config JSON
	if req.Password != nil || req.Username != nil || req.Database != nil || req.RedisMode != nil {
		cfg := parseMiddlewareConfig(target.Config)
		if req.Password != nil {
			cfg.Password = *req.Password
		}
		if req.Username != nil {
			cfg.Username = *req.Username
		}
		if req.Database != nil {
			cfg.Database = *req.Database
		}
		if req.RedisMode != nil {
			mode := strings.ToLower(*req.RedisMode)
			if mode == "cluster" {
				cfg.RedisMode = "cluster"
			} else {
				cfg.RedisMode = "standalone"
			}
		}
		cfgJSON, _ := json.Marshal(cfg)
		updates["config"] = string(cfgJSON)
	}

	if len(updates) > 0 {
		h.db.Model(&target).Updates(updates)
	}

	// 立即采集一次
	go h.collectOne(uint(id))

	c.JSON(http.StatusOK, model.Response{Code: 200, Data: target})
}

// DeleteMiddlewareTarget 删除中间件监控目标
// DELETE /api/v1/middleware-monitors/:id
func (h *MiddlewareMonitorHandler) DeleteMiddlewareTarget(c *gin.Context) {
	id, err := strconv.Atoi(c.Param("id"))
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的 ID"})
		return
	}
	if err := h.db.Delete(&model.MonitorTarget{}, id).Error; err != nil {
		c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "删除失败"})
		return
	}
	// 清理指标历史
	h.db.Where("target_id = ?", id).Delete(&model.MetricHistory{})
	c.JSON(http.StatusOK, model.Response{Code: 200, Message: "删除成功"})
}

// TestMiddlewareConnection 测试中间件连接
// POST /api/v1/middleware-monitors/:id/test
func (h *MiddlewareMonitorHandler) TestMiddlewareConnection(c *gin.Context) {
	id, err := strconv.Atoi(c.Param("id"))
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的 ID"})
		return
	}
	var target model.MonitorTarget
	if err := h.db.First(&target, id).Error; err != nil {
		c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "目标不存在"})
		return
	}

	metrics, err := h.collectTarget(target)
	if err != nil {
		c.JSON(http.StatusOK, model.Response{Code: 200, Data: gin.H{
			"success": false,
			"message": err.Error(),
		}})
		return
	}
	c.JSON(http.StatusOK, model.Response{Code: 200, Data: gin.H{
		"success": true,
		"message": fmt.Sprintf("连接成功，采集到 %d 项指标", len(metrics)),
		"metrics": metrics,
	}})
}

// CollectNow 手动触发采集
// POST /api/v1/middleware-monitors/collect
func (h *MiddlewareMonitorHandler) CollectNow(c *gin.Context) {
	count := h.collectAll()
	c.JSON(http.StatusOK, model.Response{Code: 200, Data: gin.H{
		"collected": count,
		"message":   fmt.Sprintf("采集完成，成功 %d 个目标", count),
	}})
}

// GetMiddlewareMetrics 获取中间件历史指标
// GET /api/v1/middleware-monitors/:id/metrics?metric=redis_qps&start=...&end=...
func (h *MiddlewareMonitorHandler) GetMiddlewareMetrics(c *gin.Context) {
	id, err := strconv.Atoi(c.Param("id"))
	if err != nil {
		c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的 ID"})
		return
	}

	metricName := c.Query("metric")
	now := time.Now().Unix()
	startTime := now - 3600 // 默认 1h
	endTime := now

	if h := c.Query("hours"); h != "" {
		if hours, err := strconv.Atoi(h); err == nil && hours > 0 {
			if hours > 168 {
				hours = 168
			}
			startTime = now - int64(hours)*3600
		}
	}
	if s := c.Query("start"); s != "" {
		if v, err := strconv.ParseInt(s, 10, 64); err == nil {
			startTime = v
		}
	}
	if e := c.Query("end"); e != "" {
		if v, err := strconv.ParseInt(e, 10, 64); err == nil {
			endTime = v
		}
	}

	q := h.db.Model(&model.MetricHistory{}).
		Where("target_id = ? AND timestamp >= ? AND timestamp <= ?", id, startTime, endTime)
	if metricName != "" {
		q = q.Where("metric_name = ?", metricName)
	}

	var histories []model.MetricHistory
	q.Order("timestamp ASC").Limit(2000).Find(&histories)

	c.JSON(http.StatusOK, model.Response{Code: 200, Data: gin.H{
		"list":  histories,
		"total": len(histories),
	}})
}

// ============================================
// 指标采集引擎
// ============================================

// collectAllWithRecord 采集并记录系统任务状态 // Logan.Li
func (h *MiddlewareMonitorHandler) collectAllWithRecord() int {
	start := time.Now()
	count := h.collectAll()
	systask.Record("middleware_monitor", "success", fmt.Sprintf("采集 %d 个目标", count), time.Since(start))
	return count
}

// collectAll 采集所有启用的中间件目标
func (h *MiddlewareMonitorHandler) collectAll() int {
	var targets []model.MonitorTarget
	h.db.Where("type = ? AND collect_mode = ?", model.TargetTypeMiddleware, "api").Find(&targets)

	count := 0
	for _, target := range targets {
		metrics, err := h.collectTarget(target)
		if err != nil {
			h.db.Model(&model.MonitorTarget{}).Where("id = ?", target.ID).
				Update("status", model.TargetStatusOffline)
			continue
		}
		// 存储指标
		h.saveMetrics(target.ID, metrics)
		// 更新状态
		now := time.Now()
		h.db.Model(&model.MonitorTarget{}).Where("id = ?", target.ID).
			Updates(map[string]interface{}{
				"status":       model.TargetStatusOnline,
				"last_seen_at": &now,
			})
		count++
	}
	return count
}

// collectOne 采集单个目标
func (h *MiddlewareMonitorHandler) collectOne(targetID uint) {
	var target model.MonitorTarget
	if err := h.db.First(&target, targetID).Error; err != nil {
		return
	}
	metrics, err := h.collectTarget(target)
	if err != nil {
		h.db.Model(&model.MonitorTarget{}).Where("id = ?", target.ID).
			Update("status", model.TargetStatusOffline)
		return
	}
	h.saveMetrics(target.ID, metrics)
	now := time.Now()
	h.db.Model(&model.MonitorTarget{}).Where("id = ?", target.ID).
		Updates(map[string]interface{}{
			"status":       model.TargetStatusOnline,
			"last_seen_at": &now,
		})
}

// collectTarget 连接到中间件并采集指标
func (h *MiddlewareMonitorHandler) collectTarget(target model.MonitorTarget) (map[string]float64, error) {
	mwType := strings.ToLower(target.Description) // Description 存储子类型
	if mwType == "" {
		// 尝试从名称推断
		if strings.Contains(strings.ToLower(target.Name), "redis") {
			mwType = "redis"
		} else if strings.Contains(strings.ToLower(target.Name), "mysql") {
			mwType = "mysql"
		}
	}

	cfg := parseMiddlewareConfig(target.Config)

	addr := fmt.Sprintf("%s:%d", target.IP, target.Port)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	switch mwType {
	case "redis":
		return h.collectRedis(ctx, addr, cfg.Password, cfg.RedisMode)
	case "mysql":
		return h.collectMySQL(ctx, addr, cfg)
	default:
		return nil, fmt.Errorf("未知中间件类型: %s", mwType)
	}
}

// collectMySQL 采集 MySQL 指标
func (h *MiddlewareMonitorHandler) collectMySQL(ctx context.Context, addr string, cfg MiddlewareConfig) (map[string]float64, error) {
	dbName := cfg.Database
	if dbName == "" {
		dbName = "information_schema"
	}
	user := cfg.Username
	if user == "" {
		user = "root"
	}

	dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?timeout=5s&readTimeout=3s", user, cfg.Password, addr, dbName)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, fmt.Errorf("MySQL 连接配置错误: %v", err)
	}
	defer db.Close()

	// Ping 检测连接
	if err := db.PingContext(ctx); err != nil {
		return nil, fmt.Errorf("MySQL 连接失败: %v", err)
	}

	result := make(map[string]float64)

	// SHOW GLOBAL STATUS
	rows, err := db.QueryContext(ctx, "SHOW GLOBAL STATUS")
	if err != nil {
		return nil, fmt.Errorf("MySQL 查询状态失败: %v", err)
	}
	defer rows.Close()

	statusMap := make(map[string]string)
	for rows.Next() {
		var name, val string
		if err := rows.Scan(&name, &val); err != nil {
			continue
		}
		statusMap[name] = val
	}

	result["mysql_connections"] = parseFloat(statusMap["Threads_connected"])
	result["mysql_threads_running"] = parseFloat(statusMap["Threads_running"])
	result["mysql_slow_queries"] = parseFloat(statusMap["Slow_queries"])
	result["mysql_questions"] = parseFloat(statusMap["Questions"])
	result["mysql_uptime"] = parseFloat(statusMap["Uptime"])
	result["mysql_bytes_sent"] = parseFloat(statusMap["Bytes_sent"])
	result["mysql_bytes_received"] = parseFloat(statusMap["Bytes_received"])
	result["mysql_buffer_pool_pages_total"] = parseFloat(statusMap["Innodb_buffer_pool_pages_total"])
	result["mysql_buffer_pool_read_requests"] = parseFloat(statusMap["Innodb_buffer_pool_read_requests"])
	result["mysql_buffer_pool_reads"] = parseFloat(statusMap["Innodb_buffer_pool_reads"])

	// 计算 QPS
	uptime := result["mysql_uptime"]
	questions := result["mysql_questions"]
	if uptime > 0 {
		result["mysql_qps"] = questions / uptime
	}

	// 计算缓冲池命中率
	readReqs := result["mysql_buffer_pool_read_requests"]
	reads := result["mysql_buffer_pool_reads"]
	if readReqs > 0 {
		result["mysql_buffer_hit_rate"] = (readReqs - reads) / readReqs
	} else {
		result["mysql_buffer_hit_rate"] = 0
	}

	// SHOW GLOBAL VARIABLES — 连接上限等 // Logan.Li
	varRows, err := db.QueryContext(ctx, "SHOW GLOBAL VARIABLES LIKE 'max_connections'")
	if err == nil {
		defer varRows.Close()
		for varRows.Next() {
			var name, val string
			if varRows.Scan(&name, &val) == nil && name == "max_connections" {
				result["mysql_max_connections"] = parseFloat(val)
			}
		}
	}

	return result, nil
}

// saveMetrics 将采集到的指标存入 MetricHistory
func (h *MiddlewareMonitorHandler) saveMetrics(targetID uint, metrics map[string]float64) {
	if len(metrics) == 0 {
		return
	}
	now := time.Now().Unix()
	var records []model.MetricHistory
	for name, value := range metrics {
		records = append(records, model.MetricHistory{
			TargetID:   targetID,
			MetricName: name,
			Value:      value,
			Timestamp:  now,
		})
	}
	// 批量插入
	if err := h.db.CreateInBatches(records, 100).Error; err != nil {
		log.Printf("[MiddlewareMonitor] 存储指标失败 target=%d: %v", targetID, err)
	}

	// 清理 7 天前的历史数据 (每次采集后异步清理)
	go func() {
		cutoff := time.Now().Unix() - 7*24*3600
		h.db.Where("target_id = ? AND timestamp < ?", targetID, cutoff).Delete(&model.MetricHistory{})
	}()
}

// ============================================
// 辅助函数
// ============================================

// parseRedisInfo 解析 Redis INFO 命令输出
func parseRedisInfo(info string) map[string]string {
	result := make(map[string]string)
	for _, line := range strings.Split(info, "\n") {
		line = strings.TrimSpace(line)
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		parts := strings.SplitN(line, ":", 2)
		if len(parts) == 2 {
			result[strings.TrimSpace(parts[0])] = strings.TrimSpace(parts[1])
		}
	}
	return result
}

// parseFloat 安全解析浮点数
func parseFloat(s string) float64 {
	if s == "" {
		return 0
	}
	var f float64
	fmt.Sscanf(s, "%f", &f)
	return f
}