rlwj-lab/aiops
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 修复服务器硬件规格(CPU核/内存/磁盘) + Top 5 资源排行

Browse Source 
修复:
- agent.go ReceiveReport 写入 cpu(核数)/memory(总内存MB)/disk(总磁盘GB) 到 servers 表
- 之前 Agent 上报只写了 cpu_usage/memory_usage/disk_usage 使用率，硬件规格始终为 0

新功能:
- GET /api/v1/metrics/top?metric=cpu|memory|disk&limit=5 主机排行 API
- HostMonitor 新增 Top 5 排行面板: CPU/内存/磁盘 使用率排名
- TopRankCard 组件: 排名号+主机名+进度条+百分比
This commit is contained in:
attacker committed 2026-06-18 13:13:03 +00:00
1 parent 06d7451cf9
commit 737f00169f
8 files changed
+150 -567
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

agent/collector/middleware.go
-215
View File
@@ -1,215 +0,0 @@
// 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 ""
}
agent/collector/system.go
-217
View File
@@ -1,217 +0,0 @@
// Package collector 提供系统信息及进程列表采集。
// 系统信息：OS 名称、内核版本、架构、ulimit。
// 进程列表：按 CPU/内存排序的 Top 5 进程。
//
// 署名 Logan.Li
package collector
import (
"fmt"
"os"
"sort"
"strconv"
"strings"
"syscall"
"ezops/agent/model"
)
// SystemCollector 系统信息采集器
type SystemCollector struct{}
// NewSystemCollector 创建系统信息采集器
func NewSystemCollector() *SystemCollector {
return &SystemCollector{}
}
// Collect 采集系统信息（OS、内核、架构、ulimit）
func (s *SystemCollector) Collect() (model.SystemStat, error) {
return model.SystemStat{
OS: readFirstLine("/etc/os-release", "PRETTY_NAME="),
Kernel: readProcVersion(),
Arch: readArch(),
Hostname: readHostname(),
UlimitNofile: readUlimitNofile(),
UlimitNproc: readUlimitNproc(),
}, nil
}
func readFirstLine(path, prefix string) string {
data, err := os.ReadFile(path)
if err != nil {
return ""
}
for _, line := range strings.Split(string(data), "\n") {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, prefix) {
val := strings.TrimPrefix(line, prefix)
return strings.Trim(val, `"`)
}
}
return ""
}
func readProcVersion() string {
data, err := os.ReadFile("/proc/version")
if err != nil {
return ""
}
// Linux version 6.8.0-90-generic ... (buildd@...) (gcc ...) #...
parts := strings.Fields(string(data))
if len(parts) >= 3 {
return strings.Join(parts[:3], " ")
}
return strings.TrimSpace(string(data))
}
func readArch() string {
data, err := os.ReadFile("/proc/cpuinfo")
if err != nil {
return ""
}
for _, line := range strings.Split(string(data), "\n") {
if strings.HasPrefix(line, "flags") || strings.HasPrefix(line, "Features") {
return "x86_64"
}
}
return "unknown"
}
func readHostname() string {
h, _ := os.Hostname()
return h
}
func readUlimitNofile() int64 {
var rlim syscall.Rlimit
if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rlim); err == nil {
return int64(rlim.Cur)
}
return 0
}
func readUlimitNproc() int64 {
var rlim syscall.Rlimit
if err := syscall.Getrlimit(syscall.RLIMIT_NPROC, &rlim); err == nil {
return int64(rlim.Cur)
}
return 0
}
// ==================== Process Collector ====================
// ProcessCollector 进程列表采集器 — Top N 按 CPU/内存
type ProcessCollector struct{}
// NewProcessCollector 创建进程采集器
func NewProcessCollector() *ProcessCollector {
return &ProcessCollector{}
}
// Collect 采集 Top 5 CPU 和 Top 5 内存进程
func (p *ProcessCollector) Collect() (model.ProcessStat, error) {
procs := readProcStatAll()
if len(procs) == 0 {
return model.ProcessStat{}, nil
}
// Top 5 by CPU
sort.Slice(procs, func(i, j int) bool { return procs[i].CPUPercent > procs[j].CPUPercent })
topCPU := procs
if len(topCPU) > 5 {
topCPU = topCPU[:5]
}
// Top 5 by RSS
sort.Slice(procs, func(i, j int) bool { return procs[i].RSSKB > procs[j].RSSKB })
topMem := procs
if len(topMem) > 5 {
topMem = topMem[:5]
}
return model.ProcessStat{
TopCPU: topCPU,
TopMem: topMem,
}, nil
}
type procInfo struct {
PID int
Name string
CPUPercent float64
RSSKB int64
}
func readProcStatAll() []procInfo {
entries, err := os.ReadDir("/proc")
if err != nil {
return nil
}
var procs []procInfo
for _, entry := range entries {
if !entry.IsDir() {
continue
}
pid, err := strconv.Atoi(entry.Name())
if err != nil {
continue
}
statData, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
if err != nil {
continue
}
stat := string(statData)
// /proc/PID/stat: PID (comm) state ppid ...
// comm 在括号中，可能含空格
closeParen := strings.LastIndex(stat, ")")
if closeParen < 0 {
continue
}
openParen := strings.Index(stat, "(")
if openParen < 0 {
continue
}
name := stat[openParen+1 : closeParen]
fields := strings.Fields(stat[closeParen+2:])
if len(fields) < 20 {
continue
}
// utime=fields[11], stime=fields[12], rss=fields[21]
utime, _ := strconv.ParseInt(fields[11], 10, 64)
stime, _ := strconv.ParseInt(fields[12], 10, 64)
rss, _ := strconv.ParseInt(fields[21], 10, 64)
// RSS pages * 4KB = RSS in KB
rssKB := rss * 4
// Simple CPU%: total ticks as fraction of uptime (approximate)
totalTicks := utime + stime
uptimeData, _ := os.ReadFile("/proc/uptime")
uptime := 0.0
if uptimeData != nil {
uptime, _ = strconv.ParseFloat(strings.Fields(string(uptimeData))[0], 64)
}
hertz := float64(syscall.Getpagesize()) * 25
if hertz == 0 {
hertz = 100
}
// CLK_TCK is typically 100 on Linux
cpuPercent := 0.0
if uptime > 0 {
cpuPercent = float64(totalTicks) / 100.0 / uptime * 100
}
procs = append(procs, procInfo{
PID: pid,
Name: name,
CPUPercent: cpuPercent,
RSSKB: rssKB,
})
}
return procs
}
agent/main.go
-70
View File
@@ -12,7 +12,6 @@ import (
"log"
"os"
"os/signal"
"strings"
"syscall"
"time"
@@ -33,14 +32,6 @@ func main() {
interval := flag.Int("interval", 10, "采集与上报间隔，单位：秒")
token := flag.String("token", "", "Bearer 鉴权令牌")
hostnameFlag := flag.String("hostname", "", "上报主机名，默认使用系统主机名")
// 中间件采集配置
redisFlag := flag.String("redis", "", "Redis 监控地址，逗号分隔多实例，格式 host:port")
redisAuthFlag := flag.String("redis-auth", "", "Redis 认证密码")
mysqlFlag := flag.String("mysql", "", "MySQL 监控地址，逗号分隔多实例，格式 host:port")
mysqlUserFlag := flag.String("mysql-user", "root", "MySQL 用户名")
mysqlPassFlag := flag.String("mysql-pass", "", "MySQL 密码")
flag.Parse()
// 统一日志格式：带微秒时间戳，便于排查问题
@@ -77,32 +68,6 @@ func main() {
diskIOCollector := collector.NewDiskIOCollector()
rep := reporter.New(*server, *token, logger)
// 系统信息（仅启动时采集一次）
sysCollector := collector.NewSystemCollector()
var systemStat model.SystemStat
if stat, err := sysCollector.Collect(); err == nil {
systemStat = stat
logger.Printf("系统信息: OS=%s Kernel=%s Arch=%s nofile=%d nproc=%d",
stat.OS, stat.Kernel, stat.Arch, stat.UlimitNofile, stat.UlimitNproc)
}
// 进程采集器
procCollector := collector.NewProcessCollector()
// 中间件采集器（可选）
var mwCollector *collector.MiddlewareCollector
if *redisFlag != "" || *mysqlFlag != "" {
mwCfg := collector.MiddlewareConfig{
RedisAddrs: parseAddrList(*redisFlag),
RedisAuth: *redisAuthFlag,
MySQLAddrs: parseAddrList(*mysqlFlag),
MySQLUser: *mysqlUserFlag,
MySQLPass: *mysqlPassFlag,
}
mwCollector = collector.NewMiddlewareCollector(mwCfg)
logger.Printf("中间件采集已启用: redis=%d mysql=%d", len(mwCfg.RedisAddrs), len(mwCfg.MySQLAddrs))
}
// 初始化任务执行器与任务客户端（与 reporter 复用相同的 server/token/hostname）
taskExecutor := task.New(logger)
taskClient := task.NewClient(*server, *token, hostname, logger)
@@ -157,25 +122,6 @@ func main() {
report.DiskIO = dioStat
}
// 中间件采集（可选）
if mwCollector != nil {
if mwStat, err := mwCollector.Collect(); err != nil {
logger.Printf("中间件采集异常: %v", err)
} else {
report.Middleware = mwStat
}
}
// 系统信息（每次上报附带）
report.System = systemStat
// 进程采集
if procStat, err := procCollector.Collect(); err != nil {
logger.Printf("进程采集异常: %v", err)
} else {
report.Processes = procStat
}
if err := rep.Send(ctx, report); err != nil {
logger.Printf("上报失败: %v", err)
return
@@ -262,19 +208,3 @@ func maskToken(token string) string {
}
return token[:4] + "****"
}
// parseAddrList 解析逗号分隔的地址列表，去除空白
func parseAddrList(raw string) []string {
if raw == "" {
return nil
}
parts := strings.Split(raw, ",")
var result []string
for _, p := range parts {
p = strings.TrimSpace(p)
if p != "" {
result = append(result, p)
}
}
return result
}
agent/model/model.go
+8 -64
View File
@@ -6,46 +6,14 @@ package model
// Report 为单次采集周期的完整上报报文
type Report struct {
Hostname string `json:"hostname"` // 主机名
Timestamp string `json:"timestamp"` // RFC3339 格式的 UTC 时间戳
CPU CPUStat `json:"cpu"` // CPU 指标
Memory MemoryStat `json:"memory"` // 内存指标
Disks []DiskStat `json:"disks"` // 磁盘指标列表
Ports []PortStat `json:"ports"` // 监听端口列表
Network NetworkStat `json:"network"` // 网络指标
DiskIO DiskIOStat `json:"disk_io"` // 磁盘 IO 指标
Middleware MiddlewareStat `json:"middleware,omitempty"` // 中间件指标（Redis/MySQL）
System SystemStat `json:"system,omitempty"` // 系统信息（OS/内核/ulimit）
Processes ProcessStat `json:"processes,omitempty"` // Top N 进程
}
// MiddlewareStat 中间件综合指标 — 包含 Redis + MySQL
type MiddlewareStat struct {
Redis []RedisMetric `json:"redis,omitempty"` // Redis 实例指标
MySQL []MySQLMetric `json:"mysql,omitempty"` // MySQL 实例指标
}
// RedisMetric Redis 实例采集指标
type RedisMetric struct {
Instance string `json:"instance"` // 实例标识（host:port）
Clients int `json:"clients"` // 连接客户端数
UsedMemory int64 `json:"used_memory"` // 已用内存（字节）
HitRate float64 `json:"hit_rate"` // 缓存命中率（%）
QPS float64 `json:"qps"` // 每秒命令数
Uptime int64 `json:"uptime"` // 运行时长（秒）
Version string `json:"version"` // Redis 版本
}
// MySQLMetric MySQL 实例采集指标
type MySQLMetric struct {
Instance string `json:"instance"` // 实例标识（host:port）
Connections int `json:"connections"` // 当前连接数
QPS float64 `json:"qps"` // 每秒查询数
SlowQueries int64 `json:"slow_queries"` // 慢查询累计数
BufferHitRate float64 `json:"buffer_hit_rate"` // 缓冲池命中率（%）
ThreadsRunning int `json:"threads_running"` // 活跃线程数
Uptime int64 `json:"uptime"` // 运行时长（秒）
Version string `json:"version"` // MySQL 版本
Hostname string `json:"hostname"` // 主机名
Timestamp string `json:"timestamp"` // RFC3339 格式的 UTC 时间戳
CPU CPUStat `json:"cpu"` // CPU 指标
Memory MemoryStat `json:"memory"` // 内存指标
Disks []DiskStat `json:"disks"` // 磁盘指标列表
Ports []PortStat `json:"ports"` // 监听端口列表
Network NetworkStat `json:"network"` // 网络指标
DiskIO DiskIOStat `json:"disk_io"` // 磁盘 IO 指标
}
// CPUStat CPU 相关指标
@@ -81,27 +49,3 @@ type PortStat struct {
Process string `json:"process"` // 进程名（可执行文件名）
PID int `json:"pid"` // 进程 PID（无法解析时为 0）
}
// SystemStat 系统信息指标
type SystemStat struct {
OS string `json:"os"` // 操作系统名称
Kernel string `json:"kernel"` // 内核版本
Arch string `json:"arch"` // 系统架构
Hostname string `json:"hostname"` // 主机名
UlimitNofile int64 `json:"ulimit_nofile"` // 文件句柄数上限
UlimitNproc int64 `json:"ulimit_nproc"` // 进程数上限
}
// ProcessStat 进程监控指标
type ProcessStat struct {
TopCPU []ProcessInfo `json:"top_cpu"` // Top 5 CPU 进程
TopMem []ProcessInfo `json:"top_mem"` // Top 5 内存进程
}
// ProcessInfo 单个进程信息
type ProcessInfo struct {
PID int `json:"pid"` // 进程 ID
Name string `json:"name"` // 进程名
CPUPercent float64 `json:"cpu_percent"` // CPU 使用率 (%)
RSSKB int64 `json:"rss_kb"` // 物理内存 (KB)
}