package controller import ( "context" "database/sql" "encoding/json" "fmt" "log" "net/http" "strconv" "strings" "sync" "time" "ezops/model" "ezops/pkg/systask" "github.com/gin-gonic/gin" _ "github.com/go-sql-driver/mysql" "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 }