aiops/agent/collector/middleware.go

// 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: $<len>\r\n<data>\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 ""
}