feat(monitor): M15 中间件监控 — Redis/MySQL 后端直连采集引擎

后端:
- 新增 controller/middleware_monitor.go (500+ 行) — CRUD + 测试连接 + 采集引擎
- 后台 goroutine 每 30s 自动连接 Redis/MySQL 采集指标
- Redis: go-redis v9 INFO+DBSIZE 命令采集 12 项指标
- MySQL: database/sql SHOW GLOBAL STATUS 采集 12 项指标
- 指标写入 MetricHistory 表,7 天自动清理
- 8 个 RESTful 路由注册到 routes.go

前端:
- MiddlewareMonitor.tsx 从只读卡片升级为完整管理页面
- 添加监控弹框(Redis/MySQL 类型选择+连接配置表单)
- 实例卡片操作栏(测试/编辑/删除) + 工具栏(立即采集/刷新)
- middlewareMonitorApi 8 个方法

验证: go build✅ npm build✅ 17/17 回归通过 Redis 12 指标✅ MySQL 12 指标✅
This commit is contained in:
attacker committed 2026-06-29 23:24:05 +08:00
1 parent 186ece6495
commit ed7a5c07a7
8 files changed
+1211 -132

No files matched your search

+658
View File
@@ -0,0 +1,658 @@
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
}