package controller import ( "context" "database/sql" "encoding/json" "fmt" "log" "net/http" "strconv" "strings" "sync" "time" "ezops/model" _ "github.com/go-sql-driver/mysql" "github.com/gin-gonic/gin" "github.com/redis/go-redis/v9" "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 专用 } // 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.collectAll() for { select { case <-ticker.C: h.collectAll() 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 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 } // 构建 Config JSON cfg := MiddlewareConfig{ Password: req.Password, Username: req.Username, Database: req.Database, } 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 字段存储中间件子类型 } if err := h.db.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{} } // 附带最新指标 type TargetWithMetrics struct { model.MonitorTarget SubType string `json:"sub_type"` LatestMetrics map[string]float64 `json:"latest_metrics"` } 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 } } result[i] = TargetWithMetrics{ MonitorTarget: t, SubType: t.Description, LatestMetrics: metrics, } } 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"` 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 { var cfg MiddlewareConfig if target.Config != "" { json.Unmarshal([]byte(target.Config), &cfg) } if req.Password != nil { cfg.Password = *req.Password } if req.Username != nil { cfg.Username = *req.Username } if req.Database != nil { cfg.Database = *req.Database } 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 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), }}) } // ============================================ // 指标采集引擎 // ============================================ // 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" } } // 解析配置 var cfg MiddlewareConfig if target.Config != "" { json.Unmarshal([]byte(target.Config), &cfg) } 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) case "mysql": return h.collectMySQL(ctx, addr, cfg) default: return nil, fmt.Errorf("未知中间件类型: %s", mwType) } } // collectRedis 采集 Redis 指标 func (h *MiddlewareMonitorHandler) collectRedis(ctx context.Context, addr, password string) (map[string]float64, error) { rdb := redis.NewClient(&redis.Options{ Addr: addr, Password: password, DB: 0, }) defer rdb.Close() // PING 检测连接 if err := rdb.Ping(ctx).Err(); err != nil { return nil, fmt.Errorf("Redis 连接失败: %v", err) } result := make(map[string]float64) // INFO 命令获取详细信息 info, err := rdb.Info(ctx).Result() if err == nil { infoMap := parseRedisInfo(info) result["redis_connected_clients"] = parseFloat(infoMap["connected_clients"]) result["redis_used_memory"] = parseFloat(infoMap["used_memory"]) result["redis_used_memory_peak"] = parseFloat(infoMap["used_memory_peak"]) result["redis_total_commands_processed"] = parseFloat(infoMap["total_commands_processed"]) result["redis_uptime_in_seconds"] = parseFloat(infoMap["uptime_in_seconds"]) result["redis_keyspace_hits"] = parseFloat(infoMap["keyspace_hits"]) result["redis_keyspace_misses"] = parseFloat(infoMap["keyspace_misses"]) result["redis_evicted_keys"] = parseFloat(infoMap["evicted_keys"]) result["redis_blocked_clients"] = parseFloat(infoMap["blocked_clients"]) // 计算命中率 hits := result["redis_keyspace_hits"] misses := result["redis_keyspace_misses"] if hits+misses > 0 { result["redis_hit_rate"] = hits / (hits + misses) } else { result["redis_hit_rate"] = 0 } // 计算 QPS uptime := result["redis_uptime_in_seconds"] cmds := result["redis_total_commands_processed"] if uptime > 0 { result["redis_qps"] = cmds / uptime } } // DBSIZE dbSize, err := rdb.DBSize(ctx).Result() if err == nil { result["redis_db_size"] = float64(dbSize) } return result, nil } // 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 } 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 }