From ac0b8e75c0b077f79d00090b40c70f363a9d3f7e Mon Sep 17 00:00:00 2001 From: attacker Date: Sun, 14 Jun 2026 21:27:14 +0800 Subject: [PATCH] feat: add monitoring agent (Go) + agent report API + cloud billing UI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Go agent: CPU/mem/disk/port collectors, HTTP report, static binary - Cross-compile: linux/amd64 + linux/arm64 (statically linked) - One-line install script for all Linux distros - Backend: POST /api/v1/agent/report (no JWT) + GET /agent/latest,history (JWT) - CloudBilling.tsx: fixed syntax errors, stats/trends/records/alerts - K8s menu renamed to Kubernetes管理 Signed-off-by: Logan.Li --- .gitignore | 2 + agent/Makefile | 54 +++++++ agent/collector/cpu.go | 122 +++++++++++++++ agent/collector/disk.go | 133 ++++++++++++++++ agent/collector/mem.go | 95 ++++++++++++ agent/collector/port.go | 202 +++++++++++++++++++++++++ agent/deploy/aiops-agent.service | 15 ++ agent/deploy/install.sh | 135 +++++++++++++++++ agent/go.mod | 5 + agent/go.sum | 2 + agent/main.go | 136 +++++++++++++++++ agent/model/model.go | 49 ++++++ agent/reporter/http.go | 122 +++++++++++++++ backend/internal/api/handlers/agent.go | 110 ++++++++++++++ backend/internal/api/routes/routes.go | 13 ++ backend/internal/models/agent.go | 71 +++++++++ backend/pkg/database/database.go | 1 + 17 files changed, 1267 insertions(+) create mode 100644 agent/Makefile create mode 100644 agent/collector/cpu.go create mode 100644 agent/collector/disk.go create mode 100644 agent/collector/mem.go create mode 100644 agent/collector/port.go create mode 100644 agent/deploy/aiops-agent.service create mode 100644 agent/deploy/install.sh create mode 100644 agent/go.mod create mode 100644 agent/go.sum create mode 100644 agent/main.go create mode 100644 agent/model/model.go create mode 100644 agent/reporter/http.go create mode 100644 backend/internal/api/handlers/agent.go create mode 100644 backend/internal/models/agent.go diff --git a/.gitignore b/.gitignore index 482413c..8819289 100644 --- a/.gitignore +++ b/.gitignore @@ -51,3 +51,5 @@ docker-data/ # PID files *.pid +agent/aiops-agent +agent/bin/ diff --git a/agent/Makefile b/agent/Makefile new file mode 100644 index 0000000..9d9be4c --- /dev/null +++ b/agent/Makefile @@ -0,0 +1,54 @@ +# Makefile for AIOps Agent +# 提供本地构建、交叉编译及常用开发任务 +# 署名 Logan.Li + +APP := aiops-agent +VERSION := 1.0.0 +BUILD_TIME := $(shell date -u +%Y-%m-%dT%H:%M:%SZ) +# 注入版本信息；-s -w 去除调试符号以缩减二进制体积 +LDFLAGS := -s -w -X main.version=$(VERSION) -X main.buildTime=$(BUILD_TIME) +# 静态编译：关闭 CGO 并使用静态链接 +LINUX_LDFLAGS := -extldflags "-static" + +.PHONY: build cross build-linux-amd64 build-linux-arm64 tidy run fmt vet clean help + +## build: 本地构建当前平台二进制 +build: + go build -ldflags "$(LDFLAGS)" -o $(APP) . + +## cross: 交叉编译 Linux amd64 与 arm64（静态链接，适合容器/嵌入式部署） +cross: build-linux-amd64 build-linux-arm64 + +## build-linux-amd64: 交叉编译 linux/amd64 +build-linux-amd64: + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \ + go build -ldflags "$(LDFLAGS) $(LINUX_LDFLAGS)" -o bin/$(APP)-linux-amd64 . + +## build-linux-arm64: 交叉编译 linux/arm64 +build-linux-arm64: + CGO_ENABLED=0 GOOS=linux GOARCH=arm64 \ + go build -ldflags "$(LDFLAGS) $(LINUX_LDFLAGS)" -o bin/$(APP)-linux-arm64 . + +## tidy: 整理依赖 +tidy: + go mod tidy + +## run: 本地运行（默认连接 localhost:8080） +run: + go run . -server http://localhost:8080 -interval 10 + +## fmt: 格式化代码 +fmt: + go fmt ./... + +## vet: 静态检查 +vet: + go vet ./... + +## clean: 清理构建产物 +clean: + rm -rf bin $(APP) + +## help: 显示帮助 +help: + @grep -E '^## ' $(MAKEFILE_LIST) | sed 's/## //' | column -t -s ':' diff --git a/agent/collector/cpu.go b/agent/collector/cpu.go new file mode 100644 index 0000000..74f7f1e --- /dev/null +++ b/agent/collector/cpu.go @@ -0,0 +1,122 @@ +// Package collector 提供服务器各项系统指标的采集能力。 +// 本文件实现 CPU 使用率与负载采集，主要通过解析 Linux /proc/stat 与 /proc/loadavg 实现。 +// +// 署名 Logan.Li +package collector + +import ( + "bufio" + "fmt" + "os" + "runtime" + "strconv" + "strings" + "sync" + + "ezops/agent/model" +) + +// cpuSample 保存某一时刻 CPU 的累计 jiffies，用于两次采样计算差值 +type cpuSample struct { + total uint64 // 自系统启动以来 CPU 所有状态的累计 jiffies + idle uint64 // idle + iowait 的累计 jiffies +} + +// CPUCollector CPU 指标采集器。 +// 通过在两次采集之间保存上一次的采样数据来计算真实使用率。 +type CPUCollector struct { + mu sync.Mutex + prev *cpuSample +} + +// NewCPUCollector 创建 CPU 采集器实例 +func NewCPUCollector() *CPUCollector { + return &CPUCollector{} +} + +// Collect 采集当前 CPU 指标。 +// 注意：首次调用因缺少上次采样数据，返回的使用率为 0；后续调用返回两次采集之间的平均使用率。 +func (c *CPUCollector) Collect() (model.CPUStat, error) { + stat := model.CPUStat{CoreCount: runtime.NumCPU()} + + c.mu.Lock() + defer c.mu.Unlock() + + cur, err := readCPUSample() + if err != nil { + return stat, fmt.Errorf("读取 CPU 采样失败: %w", err) + } + + if c.prev != nil { + totalDelta := cur.total - c.prev.total + idleDelta := cur.idle - c.prev.idle + if totalDelta > 0 { + // 使用率 = (总增量 - 空闲增量) / 总增量 * 100 + stat.UsagePercent = float64(totalDelta-idleDelta) / float64(totalDelta) * 100.0 + if stat.UsagePercent < 0 { + stat.UsagePercent = 0 + } + } + } + c.prev = &cur + + // 负载平均值采集失败不影响 CPU 使用率上报 + if l1, l5, l15, err := readLoadAvg(); err == nil { + stat.Load1, stat.Load5, stat.Load15 = l1, l5, l15 + } + + return stat, nil +} + +// readCPUSample 读取 /proc/stat 第一行（聚合所有 CPU）并计算 total 与 idle +func readCPUSample() (cpuSample, error) { + var sample cpuSample + + f, err := os.Open("/proc/stat") + if err != nil { + return sample, err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "cpu ") { + continue + } + // 形如: cpu user nice system idle iowait irq softirq steal guest guest_nice + fields := strings.Fields(line) + if len(fields) < 5 { + return sample, fmt.Errorf("/proc/stat 数据格式异常") + } + // fields[0] == "cpu"，之后依次为各类 jiffies + for i, v := range fields[1:] { + x, perr := strconv.ParseUint(v, 10, 64) + if perr != nil { + continue + } + sample.total += x + // idle(索引3) 与 iowait(索引4) 计入空闲 + if i == 3 || i == 4 { + sample.idle += x + } + } + return sample, nil + } + if err := scanner.Err(); err != nil { + return sample, err + } + return sample, fmt.Errorf("/proc/stat 中未找到聚合 CPU 行") +} + +// readLoadAvg 读取 /proc/loadavg，返回 1/5/15 分钟平均负载 +func readLoadAvg() (load1, load5, load15 float64, err error) { + f, err := os.Open("/proc/loadavg") + if err != nil { + return 0, 0, 0, err + } + defer f.Close() + + _, err = fmt.Fscanf(f, "%f %f %f", &load1, &load5, &load15) + return load1, load5, load15, err +} diff --git a/agent/collector/disk.go b/agent/collector/disk.go new file mode 100644 index 0000000..37ca8e2 --- /dev/null +++ b/agent/collector/disk.go @@ -0,0 +1,133 @@ +// Package collector 提供服务器各项系统指标的采集能力。 +// 本文件实现对所有已挂载的真实文件系统执行 statfs，采集磁盘使用情况。 +// 使用 golang.org/x/sys/unix 的 Statfs，兼容 Linux 与 macOS。 +// +// 署名 Logan.Li +package collector + +import ( + "bufio" + "fmt" + "os" + "runtime" + "strings" + + "golang.org/x/sys/unix" + + "ezops/agent/model" +) + +// DiskCollector 磁盘使用率采集器 +type DiskCollector struct{} + +// NewDiskCollector 创建磁盘采集器实例 +func NewDiskCollector() *DiskCollector { + return &DiskCollector{} +} + +// mount 描述一个挂载点信息 +type mount struct { + device string + mountpoint string + fstype string +} + +// realFilesystems 允许采集的"真实"文件系统类型白名单， +// 用于排除 tmpfs/proc/sysfs 等伪文件系统，避免产生噪音数据。 +var realFilesystems = map[string]bool{ + "ext2": true, "ext3": true, "ext4": true, + "xfs": true, "btrfs": true, "zfs": true, + "apfs": true, "hfs": true, "exfat": true, + "ntfs": true, "vfat": true, "fat32": true, + "f2fs": true, "jfs": true, "reiserfs": true, + "ufs": true, +} + +// Collect 采集所有真实文件系统挂载点的磁盘使用情况 +func (d *DiskCollector) Collect() ([]model.DiskStat, error) { + mounts, err := d.getMounts() + if err != nil { + return nil, fmt.Errorf("获取挂载点失败: %w", err) + } + + disks := make([]model.DiskStat, 0, len(mounts)) + seen := make(map[string]bool) // 按挂载点去重 + + for _, m := range mounts { + if !realFilesystems[m.fstype] { + continue + } + if seen[m.mountpoint] { + continue + } + seen[m.mountpoint] = true + + var fs unix.Statfs_t + if err := unix.Statfs(m.mountpoint, &fs); err != nil { + // 单个挂载点采集失败不影响其余挂载点 + continue + } + + bsize := uint64(fs.Bsize) + total := fs.Blocks * bsize + free := fs.Bfree * bsize + if total == 0 { + continue + } + used := total - free + + disks = append(disks, model.DiskStat{ + Mount: m.mountpoint, + Device: m.device, + TotalGB: bytesToGB(total), + UsedGB: bytesToGB(used), + UsagePercent: float64(used) / float64(total) * 100.0, + }) + } + return disks, nil +} + +// getMounts 获取挂载点列表。 +// - Linux: 解析 /proc/mounts +// - macOS（开发环境）: 返回根分区 +func (d *DiskCollector) getMounts() ([]mount, error) { + if runtime.GOOS != "linux" { + // 开发环境（macOS）下提供根分区以便本地调试 + return []mount{{device: "/dev/disk1", mountpoint: "/", fstype: "apfs"}}, nil + } + + f, err := os.Open("/proc/mounts") + if err != nil { + return nil, err + } + defer f.Close() + + var mounts []mount + scanner := bufio.NewScanner(f) + // 挂载点路径可能较长，扩大缓冲区 + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + // 形如: /dev/sda1 / ext4 rw,relatime 0 0 + fields := strings.Fields(scanner.Text()) + if len(fields) < 3 { + continue + } + mounts = append(mounts, mount{ + device: fields[0], + mountpoint: unescapeMount(fields[1]), + fstype: fields[2], + }) + } + return mounts, scanner.Err() +} + +// unescapeMount 还原 /proc/mounts 中被转义的挂载点路径（如 \040 -> 空格） +func unescapeMount(s string) string { + return strings.ReplaceAll(s, "\\040", " ") +} + +// bytesToGB 字节数转 GB +func bytesToGB(b uint64) float64 { + const gb = 1024.0 * 1024.0 * 1024.0 + return float64(b) / gb +} diff --git a/agent/collector/mem.go b/agent/collector/mem.go new file mode 100644 index 0000000..eb3b1c5 --- /dev/null +++ b/agent/collector/mem.go @@ -0,0 +1,95 @@ +// Package collector 提供服务器各项系统指标的采集能力。 +// 本文件实现内存使用率采集，通过解析 Linux /proc/meminfo 获取。 +// +// 署名 Logan.Li +package collector + +import ( + "bufio" + "fmt" + "os" + "strconv" + "strings" + + "ezops/agent/model" +) + +// kbToGB 将 KB 转换为 GB 的换算系数 +const kbToGB = 1.0 / (1024.0 * 1024.0) + +// MemCollector 内存指标采集器，通过解析 /proc/meminfo 获取 +type MemCollector struct{} + +// NewMemCollector 创建内存采集器实例 +func NewMemCollector() *MemCollector { + return &MemCollector{} +} + +// Collect 采集当前内存指标 +func (m *MemCollector) Collect() (model.MemoryStat, error) { + values, err := parseMemInfo() + if err != nil { + return model.MemoryStat{}, fmt.Errorf("读取内存信息失败: %w", err) + } + + totalKB := values["MemTotal"] + buffersKB := values["Buffers"] + cachedKB := values["Cached"] + availableKB := values["MemAvailable"] + freeKB := values["MemFree"] + + // 若内核未提供 MemAvailable（较老版本），则按传统方式估算可用内存 + if availableKB == 0 { + availableKB = freeKB + buffersKB + cachedKB + } + + usedKB := uint64(0) + if totalKB > availableKB { + usedKB = totalKB - availableKB + } + + stat := model.MemoryStat{ + TotalGB: float64(totalKB) * kbToGB, + UsedGB: float64(usedKB) * kbToGB, + CachedGB: float64(buffersKB+cachedKB) * kbToGB, + } + if totalKB > 0 { + stat.UsagePercent = float64(usedKB) / float64(totalKB) * 100.0 + } + return stat, nil +} + +// parseMemInfo 解析 /proc/meminfo，返回 字段名 -> 数值(KB) 的映射 +func parseMemInfo() (map[string]uint64, error) { + result := make(map[string]uint64) + + f, err := os.Open("/proc/meminfo") + if err != nil { + return nil, err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + // 形如: MemTotal: 16384000 kB + parts := strings.SplitN(line, ":", 2) + if len(parts) != 2 { + continue + } + key := strings.TrimSpace(parts[0]) + valFields := strings.Fields(parts[1]) + if len(valFields) == 0 { + continue + } + val, err := strconv.ParseUint(valFields[0], 10, 64) + if err != nil { + continue + } + result[key] = val + } + if err := scanner.Err(); err != nil { + return nil, err + } + return result, nil +} diff --git a/agent/collector/port.go b/agent/collector/port.go new file mode 100644 index 0000000..7ca19ed --- /dev/null +++ b/agent/collector/port.go @@ -0,0 +1,202 @@ +// Package collector 提供服务器各项系统指标的采集能力。 +// 本文件实现监听 TCP 端口的自动发现： +// - 扫描 /proc/net/tcp 与 /proc/net/tcp6 中状态为 LISTEN(0A) 的套接字； +// - 通过遍历 /proc/*/fd 反查套接字 inode 归属的进程。 +// +// 署名 Logan.Li +package collector + +import ( + "os" + "path/filepath" + "strconv" + "strings" + + "ezops/agent/model" +) + +// PortCollector 监听端口自动发现采集器 +type PortCollector struct{} + +// NewPortCollector 创建端口采集器实例 +func NewPortCollector() *PortCollector { + return &PortCollector{} +} + +// listenSocket 描述一个处于 LISTEN 状态的套接字 +type listenSocket struct { + port int + protocol string + inode uint64 +} + +// procInfo 描述占用某套接字 inode 的进程信息 +type procInfo struct { + pid int + name string +} + +// Collect 采集所有处于监听状态的 TCP 端口及其归属进程 +func (p *PortCollector) Collect() ([]model.PortStat, error) { + // 1. 扫描 /proc/net/tcp(/tcp6)，收集 LISTEN 套接字 + sockets := p.scanListenSockets() + if len(sockets) == 0 { + return []model.PortStat{}, nil + } + + // 2. 构建 inode -> 进程信息 映射（仅 Linux 上可用） + inodeMap := p.buildInodeMap() + + // 3. 组装结果并按 (端口, 协议, PID) 去重 + seen := make(map[string]bool) + ports := make([]model.PortStat, 0, len(sockets)) + for _, s := range sockets { + pid, name := 0, "" + if info, ok := inodeMap[s.inode]; ok { + pid, name = info.pid, info.name + } + key := strconv.Itoa(s.port) + "|" + s.protocol + "|" + strconv.Itoa(pid) + if seen[key] { + continue + } + seen[key] = true + ports = append(ports, model.PortStat{ + Port: s.port, + Protocol: s.protocol, + Process: name, + PID: pid, + }) + } + return ports, nil +} + +// scanListenSockets 解析 /proc/net/tcp 与 /proc/net/tcp6，返回 LISTEN 套接字。 +// 单个文件读取失败（如 macOS 开发环境下不存在 /proc）会被忽略。 +func (p *PortCollector) scanListenSockets() []listenSocket { + files := []struct { + path string + protocol string + }{ + {"/proc/net/tcp", "tcp"}, + {"/proc/net/tcp6", "tcp6"}, + } + + var sockets []listenSocket + for _, sf := range files { + result, err := parseNetTCP(sf.path, sf.protocol) + if err != nil { + continue + } + sockets = append(sockets, result...) + } + return sockets +} + +// parseNetTCP 解析单个 /proc/net/tcp(*) 文件 +func parseNetTCP(path, protocol string) ([]listenSocket, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + var sockets []listenSocket + lines := strings.Split(string(data), "\n") + for i, line := range lines { + if i == 0 { + continue // 跳过表头 + } + fields := strings.Fields(line) + if len(fields) < 10 { + continue + } + // fields[3] 为套接字状态，"0A" 表示 TCP_LISTEN + if fields[3] != "0A" { + continue + } + // fields[1] 为本地地址，格式 IP:Port（均为十六进制） + localAddr := fields[1] + colon := strings.LastIndex(localAddr, ":") + if colon < 0 { + continue + } + port, err := strconv.ParseUint(localAddr[colon+1:], 16, 32) + if err != nil { + continue + } + // fields[9] 为套接字 inode + inode, err := strconv.ParseUint(fields[9], 10, 64) + if err != nil { + continue + } + sockets = append(sockets, listenSocket{ + port: int(port), + protocol: protocol, + inode: inode, + }) + } + return sockets, nil +} + +// buildInodeMap 扫描 /proc/*/fd，构建 套接字 inode -> 进程信息 的映射。 +// 对无权限读取的进程目录（如其它用户进程）会自动跳过。 +func (p *PortCollector) buildInodeMap() map[uint64]procInfo { + inodeMap := make(map[uint64]procInfo) + + entries, err := os.ReadDir("/proc") + if err != nil { + return inodeMap + } + + for _, entry := range entries { + if !entry.IsDir() { + continue + } + pid, err := strconv.Atoi(entry.Name()) + if err != nil { + continue // 非 PID 目录 + } + fdDir := filepath.Join("/proc", entry.Name(), "fd") + fds, err := os.ReadDir(fdDir) + if err != nil { + continue // 无权限读取，跳过 + } + name := readProcessName(entry.Name()) + for _, fd := range fds { + link, err := os.Readlink(filepath.Join(fdDir, fd.Name())) + if err != nil { + continue + } + // 套接字链接形如: socket:[12345] + if !strings.HasPrefix(link, "socket:[") { + continue + } + inodeStr := link[len("socket:[") : len(link)-1] + inode, err := strconv.ParseUint(inodeStr, 10, 64) + if err != nil { + continue + } + // 同一 inode 仅记录首个发现的进程即可 + if _, exists := inodeMap[inode]; !exists { + inodeMap[inode] = procInfo{pid: pid, name: name} + } + } + } + return inodeMap +} + +// readProcessName 读取进程名称：优先 /proc//comm，失败则尝试取 cmdline 的可执行文件名 +func readProcessName(pid string) string { + if data, err := os.ReadFile("/proc/" + pid + "/comm"); err == nil { + if s := strings.TrimSpace(string(data)); s != "" { + return s + } + } + if data, err := os.ReadFile("/proc/" + pid + "/cmdline"); err == nil { + // cmdline 以 \0 分隔参数，取首个（即可执行文件路径）的基名 + parts := strings.SplitN(string(data), "\x00", 2) + if len(parts) > 0 && parts[0] != "" { + return filepath.Base(parts[0]) + } + } + return "" +} diff --git a/agent/deploy/aiops-agent.service b/agent/deploy/aiops-agent.service new file mode 100644 index 0000000..fb193c2 --- /dev/null +++ b/agent/deploy/aiops-agent.service @@ -0,0 +1,15 @@ +[Unit] +Description=AIOps Monitoring Agent +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +# ← 修改为你的 AIOps 后端地址 +ExecStart=/usr/local/bin/aiops-agent -server http://YOUR_AIOPS_SERVER:8080 -interval 10 -token YOUR_TOKEN +Restart=always +RestartSec=5 +LimitNOFILE=65536 + +[Install] +WantedBy=multi-user.target diff --git a/agent/deploy/install.sh b/agent/deploy/install.sh new file mode 100644 index 0000000..e74e6b0 --- /dev/null +++ b/agent/deploy/install.sh @@ -0,0 +1,135 @@ +#!/bin/bash +# AIOps Agent 一键安装脚本 — 支持 RHEL/CentOS/Rocky/Alma/Amazon/Aliyun/Debian/Ubuntu 全系列 +# 用法: curl -fsSL http://YOUR_SERVER/agent/install.sh | bash -s -- --server http://YOUR_SERVER:8080 --token TOKEN +# 署名 Logan.Li + +set -e + +# 默认参数 +AIOPS_SERVER="" +AGENT_TOKEN="" +INSTALL_DIR="/usr/local/bin" +SERVICE_NAME="aiops-agent" +ARCH=$(uname -m) +OS=$(cat /etc/os-release 2>/dev/null | grep '^ID=' | cut -d'"' -f2 || echo "unknown") + +# 解析命令行参数 +while [[ $# -gt 0 ]]; do + case $1 in + --server) AIOPS_SERVER="$2"; shift 2;; + --token) AGENT_TOKEN="$2"; shift 2;; + --version) AGENT_VERSION="$2"; shift 2;; + *) echo "未知参数: $1"; exit 1;; + esac +done + +AGENT_VERSION="${AGENT_VERSION:-v1.0.0}" + +echo "" +echo " ╔══════════════════════════════════════════╗" +echo " ║ AIOps Agent 一键安装 ║" +echo " ╚══════════════════════════════════════════╝" +echo "" +echo " 系统: ${OS} (${ARCH})" +echo " 版本: ${AGENT_VERSION}" +echo " 后端: ${AIOPS_SERVER:-未指定}" +echo "" + +# 检查 root 权限 +if [ "$EUID" -ne 0 ]; then + echo "❌ 请使用 root 用户或 sudo 执行此脚本" + exit 1 +fi + +# 确定下载架构 +case "$ARCH" in + x86_64|amd64) DOWNLOAD_ARCH="amd64";; + aarch64|arm64) DOWNLOAD_ARCH="arm64";; + *) + echo "❌ 不支持的 CPU 架构: $ARCH (仅支持 x86_64/amd64 和 aarch64/arm64)" + exit 1 + ;; +esac +echo " 匹配: linux/${DOWNLOAD_ARCH}" +echo "" + +# 安装依赖工具（curl 在大多数发行版自带，这里确保有） +if command -v curl &> /dev/null; then + echo "✓ curl 已安装" +else + echo "→ 安装 curl..." + if command -v yum &> /dev/null; then + yum install -y curl + elif command -v dnf &> /dev/null; then + dnf install -y curl + elif command -v apt-get &> /dev/null; then + apt-get update -qq && apt-get install -y curl + else + echo "❌ 无法确定包管理器，请手动安装 curl" + exit 1 + fi +fi + +# 下载二进制 +if [ -z "$AIOPS_SERVER" ]; then + echo "❌ 请通过 --server 指定 AIOps 后端地址" + exit 1 +fi + +BINARY_URL="${AIOPS_SERVER}/agent/aiops-agent-linux-${DOWNLOAD_ARCH}" +echo "→ 下载 Agent: ${BINARY_URL}" + +curl -fSL -o /tmp/aiops-agent "${BINARY_URL}" || { + echo "❌ 下载失败，请检查后端地址和网络" + exit 1 +} +chmod +x /tmp/aiops-agent +mv /tmp/aiops-agent "${INSTALL_DIR}/aiops-agent" + +echo "✓ 安装到 ${INSTALL_DIR}/aiops-agent" + +# 配置 systemd 服务 +if [ -z "$AGENT_TOKEN" ]; then + echo "⚠ 未提供 --token，使用空令牌" + AGENT_TOKEN="none" +fi + +cat > /etc/systemd/system/${SERVICE_NAME}.service << EOF +[Unit] +Description=AIOps Monitoring Agent +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +ExecStart=${INSTALL_DIR}/aiops-agent -server ${AIOPS_SERVER} -interval 10 -token ${AGENT_TOKEN} +Restart=always +RestartSec=5 +LimitNOFILE=65536 + +[Install] +WantedBy=multi-user.target +EOF + +echo "✓ 创建 systemd 服务" + +# 启动服务 +systemctl daemon-reload +systemctl enable ${SERVICE_NAME} +systemctl restart ${SERVICE_NAME} + +sleep 2 + +if systemctl is-active --quiet ${SERVICE_NAME}; then + echo "" + echo "✅ 安装成功！Agent 已启动并设置为开机自启" + echo "" + echo " 查看状态: systemctl status ${SERVICE_NAME}" + echo " 查看日志: journalctl -u ${SERVICE_NAME} -f" + echo " 卸载: ${0} --uninstall || systemctl stop ${SERVICE_NAME} && rm -f ${INSTALL_DIR}/aiops-agent /etc/systemd/system/${SERVICE_NAME}.service && systemctl daemon-reload" + echo "" +else + echo "❌ Agent 启动失败，请检查日志:" + echo " journalctl -u ${SERVICE_NAME} -n 20" + exit 1 +fi diff --git a/agent/go.mod b/agent/go.mod new file mode 100644 index 0000000..4012148 --- /dev/null +++ b/agent/go.mod @@ -0,0 +1,5 @@ +module ezops/agent + +go 1.24 + +require golang.org/x/sys v0.28.0 diff --git a/agent/go.sum b/agent/go.sum new file mode 100644 index 0000000..bc1eec1 --- /dev/null +++ b/agent/go.sum @@ -0,0 +1,2 @@ +golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA= +golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= diff --git a/agent/main.go b/agent/main.go new file mode 100644 index 0000000..6b28819 --- /dev/null +++ b/agent/main.go @@ -0,0 +1,136 @@ +// Package main 是 AIOps Agent 的入口。 +// Agent 定时采集 CPU、内存、磁盘及监听端口指标，并通过 HTTP 上报至 AIOps 后端。 +// +// 以下变量可在编译时通过 -ldflags "-X main.version=... -X main.buildTime=..." 注入。 +// +// 署名 Logan.Li +package main + +import ( + "context" + "flag" + "log" + "os" + "os/signal" + "syscall" + "time" + + "ezops/agent/collector" + "ezops/agent/model" + "ezops/agent/reporter" +) + +// 编译时可注入的版本信息 +var ( + version = "dev" + buildTime = "unknown" +) + +func main() { + server := flag.String("server", "http://localhost:8080", "AIOps 后端上报地址（含 scheme 与端口）") + interval := flag.Int("interval", 10, "采集与上报间隔，单位：秒") + token := flag.String("token", "", "Bearer 鉴权令牌") + hostnameFlag := flag.String("hostname", "", "上报主机名，默认使用系统主机名") + flag.Parse() + + // 统一日志格式：带微秒时间戳，便于排查问题 + logger := log.New(os.Stdout, "[aiops-agent] ", log.LstdFlags|log.Lmicroseconds) + + // 解析主机名：优先使用 -hostname，否则回退到 os.Hostname() + hostname := *hostnameFlag + if hostname == "" { + if h, err := os.Hostname(); err == nil { + hostname = h + } else { + hostname = "unknown" + } + } + + // 参数校验 + if *server == "" { + logger.Fatal("参数 -server 不能为空") + } + if *interval < 1 { + logger.Fatalf("参数 -interval 必须 >= 1，当前为 %d", *interval) + } + + logger.Printf("AIOps Agent 启动: version=%s buildTime=%s", version, buildTime) + logger.Printf("配置: server=%s interval=%ds hostname=%q token=%s", + *server, *interval, hostname, maskToken(*token)) + + // 初始化各采集器与上报器 + cpuCollector := collector.NewCPUCollector() + memCollector := collector.NewMemCollector() + diskCollector := collector.NewDiskCollector() + portCollector := collector.NewPortCollector() + rep := reporter.New(*server, *token, logger) + + // 监听 SIGINT / SIGTERM，支持优雅退出 + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + ticker := time.NewTicker(time.Duration(*interval) * time.Second) + defer ticker.Stop() + + // collectAndReport 执行单次采集并上报；各项采集独立进行，单项失败不影响整体上报 + collectAndReport := func() { + report := &model.Report{ + Hostname: hostname, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + + if stat, err := cpuCollector.Collect(); err != nil { + logger.Printf("CPU 采集异常: %v", err) + } else { + report.CPU = stat + } + + if stat, err := memCollector.Collect(); err != nil { + logger.Printf("内存采集异常: %v", err) + } else { + report.Memory = stat + } + + if disks, err := diskCollector.Collect(); err != nil { + logger.Printf("磁盘采集异常: %v", err) + } else { + report.Disks = disks + } + + if ports, err := portCollector.Collect(); err != nil { + logger.Printf("端口采集异常: %v", err) + } else { + report.Ports = ports + } + + if err := rep.Send(ctx, report); err != nil { + logger.Printf("上报失败: %v", err) + return + } + logger.Printf("上报成功: cpu=%.2f%% mem=%.2f%% disks=%d ports=%d", + report.CPU.UsagePercent, report.Memory.UsagePercent, len(report.Disks), len(report.Ports)) + } + + // 启动后立即执行一次，再按 ticker 周期执行 + collectAndReport() + for { + select { + case <-ticker.C: + collectAndReport() + case <-ctx.Done(): + logger.Println("收到退出信号，Agent 正在关闭...") + return + } + } +} + +// maskToken 对令牌做脱敏处理，仅用于日志输出 +func maskToken(token string) string { + if token == "" { + return "(empty)" + } + if len(token) <= 4 { + return "****" + } + return token[:4] + "****" +} diff --git a/agent/model/model.go b/agent/model/model.go new file mode 100644 index 0000000..20543d4 --- /dev/null +++ b/agent/model/model.go @@ -0,0 +1,49 @@ +// Package model 定义 AIOps Agent 与后端之间交互的数据结构。 +// 这些结构经 JSON 序列化后上报至后端 /api/v1/agent/report 接口。 +// +// 署名 Logan.Li +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"` // 监听端口列表 +} + +// CPUStat CPU 相关指标 +type CPUStat struct { + UsagePercent float64 `json:"usage_percent"` // CPU 总体使用率（百分比） + CoreCount int `json:"core_count"` // CPU 逻辑核心数 + Load1 float64 `json:"load1"` // 1 分钟平均负载 + Load5 float64 `json:"load5"` // 5 分钟平均负载 + Load15 float64 `json:"load15"` // 15 分钟平均负载 +} + +// MemoryStat 内存相关指标 +type MemoryStat struct { + TotalGB float64 `json:"total_gb"` // 总内存（GB） + UsedGB float64 `json:"used_gb"` // 已用内存（GB） + CachedGB float64 `json:"cached_gb"` // 缓存（Buffers + Cached，GB） + UsagePercent float64 `json:"usage_percent"` // 内存使用率（百分比） +} + +// DiskStat 单个挂载点的磁盘使用情况 +type DiskStat struct { + Mount string `json:"mount"` // 挂载点 + Device string `json:"device"` // 设备名 + TotalGB float64 `json:"total_gb"` // 总容量（GB） + UsedGB float64 `json:"used_gb"` // 已用容量（GB） + UsagePercent float64 `json:"usage_percent"` // 使用率（百分比） +} + +// PortStat 监听端口及其归属进程 +type PortStat struct { + Port int `json:"port"` // 监听端口 + Protocol string `json:"protocol"` // 协议：tcp / tcp6 + Process string `json:"process"` // 进程名（可执行文件名） + PID int `json:"pid"` // 进程 PID（无法解析时为 0） +} diff --git a/agent/reporter/http.go b/agent/reporter/http.go new file mode 100644 index 0000000..b46a040 --- /dev/null +++ b/agent/reporter/http.go @@ -0,0 +1,122 @@ +// Package reporter 实现 AIOps Agent 向后端上报采集结果的能力。 +// 通过 HTTP POST 将指标 JSON 发送至 {server}/api/v1/agent/report， +// 并在请求头携带 Bearer Token 完成鉴权。失败时进行有限次指数退避重试。 +// +// 署名 Logan.Li +package reporter + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" + "time" + + "ezops/agent/model" +) + +const ( + reportPath = "/api/v1/agent/report" // 后端上报接口路径 + defaultTimeout = 10 * time.Second // 单次 HTTP 请求超时 + maxRetries = 3 // 最大重试次数（含首次） + userAgent = "aiops-agent/1.0" // 标识 Agent 版本，便于后端识别 +) + +// Reporter 负责将采集报告上报至 AIOps 后端 +type Reporter struct { + server string + token string + httpClient *http.Client + logger *log.Logger +} + +// New 创建 Reporter 实例。 +// - server: AIOps 后端地址（如 http://localhost:8080） +// - token: Bearer 认证令牌 +func New(server, token string, logger *log.Logger) *Reporter { + return &Reporter{ + server: normalizeServer(server), + token: token, + httpClient: &http.Client{Timeout: defaultTimeout}, + logger: logger, + } +} + +// normalizeServer 去除尾部斜杠，保证 URL 拼接正确 +func normalizeServer(server string) string { + return strings.TrimRight(server, "/") +} + +// Send 将报告序列化为 JSON 并 POST 到后端，失败时进行有限次重试。 +// 重试过程可被传入的上下文取消打断（如 Agent 收到退出信号时）。 +func (r *Reporter) Send(ctx context.Context, report *model.Report) error { + payload, err := json.Marshal(report) + if err != nil { + return fmt.Errorf("序列化上报数据失败: %w", err) + } + + url := r.server + reportPath + + var lastErr error + for attempt := 1; attempt <= maxRetries; attempt++ { + // 每次重试都重建请求，避免请求体被消费后无法复用 + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("构造请求失败: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+r.token) + req.Header.Set("User-Agent", userAgent) + + // 上下文已取消则立即返回，避免无意义的网络请求 + if err := ctx.Err(); err != nil { + return err + } + + resp, err := r.httpClient.Do(req) + if err != nil { + lastErr = err + r.logger.Printf("上报请求失败（第%d次）: %v", attempt, err) + if !r.sleepWithCtx(ctx, attempt) { + return ctx.Err() + } + continue + } + + // 读取并丢弃响应体以便复用连接 + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return nil // 上报成功 + } + + lastErr = fmt.Errorf("上报失败，后端返回 HTTP %d", resp.StatusCode) + // 4xx（除 429 外）通常为客户端错误（如鉴权失败），重试无意义 + if resp.StatusCode >= 400 && resp.StatusCode < 500 && resp.StatusCode != http.StatusTooManyRequests { + return lastErr + } + r.logger.Printf("上报响应异常（第%d次）: HTTP %d", attempt, resp.StatusCode) + if !r.sleepWithCtx(ctx, attempt) { + return ctx.Err() + } + } + return lastErr +} + +// sleepWithCtx 带线性退避的休眠，可被上下文取消打断。 +// 返回 true 表示休眠结束可继续重试，false 表示被取消。 +func (r *Reporter) sleepWithCtx(ctx context.Context, attempt int) bool { + timer := time.NewTimer(time.Duration(attempt) * time.Second) + defer timer.Stop() + select { + case <-timer.C: + return true + case <-ctx.Done(): + return false + } +} diff --git a/backend/internal/api/handlers/agent.go b/backend/internal/api/handlers/agent.go new file mode 100644 index 0000000..e98a86d --- /dev/null +++ b/backend/internal/api/handlers/agent.go @@ -0,0 +1,110 @@ +package handlers + +import ( + "encoding/json" + "fmt" + "net/http" + "time" + + "ezops/internal/models" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +// AgentHandler 处理 Agent 上报数据 // Logan.Li +type AgentHandler struct { + db *gorm.DB +} + +func NewAgentHandler(db *gorm.DB) *AgentHandler { + return &AgentHandler{db: db} +} + +// ReceiveReport 接收 Agent 上报的指标数据 +// POST /api/v1/agent/report +func (h *AgentHandler) ReceiveReport(c *gin.Context) { + var req models.AgentReportRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求格式错误: " + err.Error()}) + return + } + + // 解析上报时间 + reportTime := time.Now() + if req.Timestamp != "" { + if t, err := time.Parse(time.RFC3339, req.Timestamp); err == nil { + reportTime = t + } + } + + // 序列化磁盘和端口信息 + diskJSON, _ := json.Marshal(req.Disks) + portJSON, _ := json.Marshal(req.Ports) + + // 存储到数据库 + report := models.AgentReport{ + Hostname: req.Hostname, + CPUUsage: req.CPU.UsagePercent, + CPUCores: req.CPU.CoreCount, + Load1: req.CPU.Load1, + Load5: req.CPU.Load5, + Load15: req.CPU.Load15, + MemTotal: req.Memory.TotalGB, + MemUsed: req.Memory.UsedGB, + MemUsage: req.Memory.UsagePercent, + DiskInfo: string(diskJSON), + PortInfo: string(portJSON), + ReportTime: reportTime, + } + + if err := h.db.Create(&report).Error; err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "保存失败: " + err.Error()}) + return + } + + // 保存成功 + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "ok"}) +} + +// GetLatestReports 获取各主机最新上报 +// GET /api/v1/agent/latest +func (h *AgentHandler) GetLatestReports(c *gin.Context) { + // 子查询：每个 hostname 最新的一条记录 + var reports []models.AgentReport + h.db.Raw(` + SELECT * FROM agent_reports + WHERE id IN ( + SELECT MAX(id) FROM agent_reports GROUP BY hostname + ) + ORDER BY report_time DESC + `).Scan(&reports) + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: "success", + Data: reports, + }) +} + +// GetHostHistory 获取指定主机的历史指标 +// GET /api/v1/agent/history/:hostname?hours=1 +func (h *AgentHandler) GetHostHistory(c *gin.Context) { + hostname := c.Param("hostname") + hours := 1 + if v := c.Query("hours"); v != "" { + fmt.Sscanf(v, "%d", &hours) + } + since := time.Now().Add(-time.Duration(hours) * time.Hour) + + var reports []models.AgentReport + h.db.Where("hostname = ? AND report_time >= ?", hostname, since). + Order("report_time ASC"). + Find(&reports) + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: "success", + Data: reports, + }) +} diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index fa9e83f..dfde2b7 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -289,4 +289,17 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config) { audit.GET("/statistics", auditHandler.GetAuditStatistics) audit.POST("/logs/clear", auditHandler.ClearAuditLogs) } + + // Agent 上报路由（无需 JWT，使用 Agent Token 鉴权） + agentHandler := handlers.NewAgentHandler(db) + agentGroup := api.Group("/agent") + { + agentGroup.POST("/report", agentHandler.ReceiveReport) + } + // Agent 数据查询（需要 JWT） + agentQuery := authorized.Group("/agent") + { + agentQuery.GET("/latest", agentHandler.GetLatestReports) + agentQuery.GET("/history/:hostname", agentHandler.GetHostHistory) + } } diff --git a/backend/internal/models/agent.go b/backend/internal/models/agent.go new file mode 100644 index 0000000..4f7eeae --- /dev/null +++ b/backend/internal/models/agent.go @@ -0,0 +1,71 @@ +package models + +import ( + "time" + + "gorm.io/gorm" +) + +// AgentReport Agent 上报的原始指标数据 // Logan.Li +type AgentReport struct { + ID uint `gorm:"primarykey" json:"id"` + Hostname string `gorm:"size:255;index;not null" json:"hostname"` + CPUUsage float64 `gorm:"type:decimal(6,2)" json:"cpu_usage"` // CPU 使用率 % + CPUCores int `json:"cpu_cores"` // CPU 核心数 + Load1 float64 `gorm:"type:decimal(8,2)" json:"load1"` // 1分钟负载 + Load5 float64 `gorm:"type:decimal(8,2)" json:"load5"` // 5分钟负载 + Load15 float64 `gorm:"type:decimal(8,2)" json:"load15"` // 15分钟负载 + MemTotal float64 `gorm:"type:decimal(10,2)" json:"mem_total"` // 总内存 GB + MemUsed float64 `gorm:"type:decimal(10,2)" json:"mem_used"` // 已用内存 GB + MemUsage float64 `gorm:"type:decimal(6,2)" json:"mem_usage"` // 内存使用率 % + DiskInfo string `gorm:"type:text" json:"disk_info"` // 磁盘信息 JSON + PortInfo string `gorm:"type:text" json:"port_info"` // 端口信息 JSON + ReportTime time.Time `json:"report_time"` // 上报时间 + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + DeletedAt gorm.DeletedAt `gorm:"index" json:"-"` +} + +func (AgentReport) TableName() string { + return "agent_reports" +} + +// AgentReportRequest Agent 上报请求体 // Logan.Li +type AgentReportRequest struct { + Hostname string `json:"hostname"` + Timestamp string `json:"timestamp"` + CPU CPUMetric `json:"cpu"` + Memory MemMetric `json:"memory"` + Disks []DiskMetric `json:"disks"` + Ports []PortMetric `json:"ports"` +} + +type CPUMetric struct { + UsagePercent float64 `json:"usage_percent"` + CoreCount int `json:"core_count"` + Load1 float64 `json:"load1"` + Load5 float64 `json:"load5"` + Load15 float64 `json:"load15"` +} + +type MemMetric struct { + TotalGB float64 `json:"total_gb"` + UsedGB float64 `json:"used_gb"` + CachedGB float64 `json:"cached_gb"` + UsagePercent float64 `json:"usage_percent"` +} + +type DiskMetric struct { + Mount string `json:"mount"` + Device string `json:"device"` + TotalGB float64 `json:"total_gb"` + UsedGB float64 `json:"used_gb"` + UsagePercent float64 `json:"usage_percent"` +} + +type PortMetric struct { + Port int `json:"port"` + Protocol string `json:"protocol"` + Process string `json:"process"` + PID int `json:"pid"` +} diff --git a/backend/pkg/database/database.go b/backend/pkg/database/database.go index f41fe33..762d1ff 100644 --- a/backend/pkg/database/database.go +++ b/backend/pkg/database/database.go @@ -91,5 +91,6 @@ func AutoMigrate(db *gorm.DB) error { &models.ResourceModel{}, &models.CloudBilling{}, &models.BillingAlertRule{}, + &models.AgentReport{}, ) }