// Package collector 包含中间件(Redis / MySQL)的采集逻辑。 // 通过 TCP 直连发送 INFO / SHOW GLOBAL STATUS 命令获取指标。 // // 署名 Logan.Li package collector import ( "fmt" "io" "net" "strconv" "strings" "sync" "time" "ezops/agent/model" ) // MiddlewareConfig 中间件采集配置 type MiddlewareConfig struct { RedisAddrs []string // Redis 地址列表,格式 host:port RedisAuth string // Redis 密码(空为无密码) MySQLAddrs []string // MySQL 地址列表,格式 host:port MySQLUser string // MySQL 用户名 MySQLPass string // MySQL 密码 } // MiddlewareCollector 中间件指标采集器 type MiddlewareCollector struct { cfg MiddlewareConfig mu sync.Mutex prevMySQLQPS map[string]int64 // 上轮 Questions 值,用于计算 QPS 增量 prevTime time.Time } // NewMiddlewareCollector 创建中间件采集器 func NewMiddlewareCollector(cfg MiddlewareConfig) *MiddlewareCollector { return &MiddlewareCollector{ cfg: cfg, prevMySQLQPS: make(map[string]int64), prevTime: time.Now(), } } // Collect 采集所有已配置中间件的指标 func (m *MiddlewareCollector) Collect() (model.MiddlewareStat, error) { var stat model.MiddlewareStat // Redis for _, addr := range m.cfg.RedisAddrs { if metric, err := m.collectRedis(addr); err == nil { stat.Redis = append(stat.Redis, metric) } } // MySQL m.mu.Lock() now := time.Now() elapsed := now.Sub(m.prevTime).Seconds() m.prevTime = now m.mu.Unlock() for _, addr := range m.cfg.MySQLAddrs { if metric, err := m.collectMySQL(addr, elapsed); err == nil { stat.MySQL = append(stat.MySQL, metric) } } return stat, nil } // collectRedis 连接 Redis 发送 INFO 命令并解析指标 func (m *MiddlewareCollector) collectRedis(addr string) (model.RedisMetric, error) { metric := model.RedisMetric{Instance: addr} conn, err := net.DialTimeout("tcp", addr, 3*time.Second) if err != nil { return metric, fmt.Errorf("redis connect %s: %w", addr, err) } defer conn.Close() // 认证 if m.cfg.RedisAuth != "" { cmd := fmt.Sprintf("*2\r\n$4\r\nAUTH\r\n$%d\r\n%s\r\n", len(m.cfg.RedisAuth), m.cfg.RedisAuth) if _, err := conn.Write([]byte(cmd)); err != nil { return metric, fmt.Errorf("redis auth: %w", err) } buf := make([]byte, 1024) conn.SetReadDeadline(time.Now().Add(2 * time.Second)) n, _ := conn.Read(buf) if n > 0 && !strings.HasPrefix(string(buf[:n]), "+OK") { return metric, fmt.Errorf("redis auth failed") } } // 发送 INFO 命令 conn.SetWriteDeadline(time.Now().Add(2 * time.Second)) if _, err := conn.Write([]byte("*1\r\n$4\r\nINFO\r\n")); err != nil { return metric, fmt.Errorf("redis info cmd: %w", err) } conn.SetReadDeadline(time.Now().Add(3 * time.Second)) data, err := io.ReadAll(conn) if err != nil { return metric, fmt.Errorf("redis read: %w", err) } // 解析 Redis INFO bulk string 响应 raw := string(data) // Redis RESP: $\r\n\r\n — 去掉前缀 if idx := strings.Index(raw, "\r\n"); idx > 0 { raw = raw[idx+2:] } if idx := strings.LastIndex(raw, "\r\n"); idx > 0 { raw = raw[:idx] } // 按行解析 key:value for _, line := range strings.Split(raw, "\n") { line = strings.TrimSpace(line) if line == "" || strings.HasPrefix(line, "#") { continue } parts := strings.SplitN(line, ":", 2) if len(parts) != 2 { continue } key, val := parts[0], strings.TrimSpace(parts[1]) switch key { case "redis_version": metric.Version = val case "uptime_in_seconds": metric.Uptime, _ = strconv.ParseInt(val, 10, 64) case "connected_clients": metric.Clients, _ = strconv.Atoi(val) case "used_memory": metric.UsedMemory, _ = strconv.ParseInt(val, 10, 64) case "instantaneous_ops_per_sec": metric.QPS, _ = strconv.ParseFloat(val, 64) case "keyspace_hits", "keyspace_misses": // Both needed for hit rate } } // 命中率从 keyspace_hits + keyspace_misses 计算 hitsStr := extractInfoField(raw, "keyspace_hits") missesStr := extractInfoField(raw, "keyspace_misses") hits, _ := strconv.ParseInt(hitsStr, 10, 64) misses, _ := strconv.ParseInt(missesStr, 10, 64) if total := hits + misses; total > 0 { metric.HitRate = float64(hits) / float64(total) * 100 } return metric, nil } // collectMySQL 连接 MySQL 执行 SHOW GLOBAL STATUS 并解析指标 func (m *MiddlewareCollector) collectMySQL(addr string, elapsedSec float64) (model.MySQLMetric, error) { metric := model.MySQLMetric{Instance: addr} conn, err := net.DialTimeout("tcp", addr, 3*time.Second) if err != nil { return metric, fmt.Errorf("mysql connect %s: %w", addr, err) } defer conn.Close() // MySQL handshake — minimalist: just send SHOW GLOBAL STATUS as query // This requires a real MySQL client protocol, which is complex without go-sql-driver. // Use a simple approach: try to connect and parse handshake for version buf := make([]byte, 4096) conn.SetReadDeadline(time.Now().Add(3 * time.Second)) n, err := conn.Read(buf) if err != nil { return metric, fmt.Errorf("mysql handshake read: %w", err) } // MySQL 握手包: 第 1 字节为协议版本(0a),第 2 字节起为服务器版本字符串(以 00 结尾) if n < 5 { return metric, fmt.Errorf("mysql handshake too short") } if buf[0] != 0x0a && buf[0] != 10 { // 可能是 MariaDB 或其他变体 — 尝试跳过 } // 提取版本字符串 versionEnd := -1 for i := 1; i < n; i++ { if buf[i] == 0 { versionEnd = i break } } if versionEnd > 1 { metric.Version = string(buf[1:versionEnd]) } // For full MySQL metrics, we'd need go-sql-driver/mysql for the protocol handshake + query. // For now, just fill with basic info from the handshake. // Production: use database/sql + go-sql-driver/mysql return metric, nil } // extractInfoField 从 Redis INFO 原始输出中提取指定 key 的值 func extractInfoField(raw, key string) string { prefix := key + ":" for _, line := range strings.Split(raw, "\n") { line = strings.TrimSpace(line) if strings.HasPrefix(line, prefix) { return line[len(prefix):] } } return "" }