feat: add monitoring agent (Go) + agent report API + cloud billing UI

- 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
This commit is contained in:
attacker committed 2026-06-14 21:27:14 +08:00
1 parent 77d13b07d5
commit ac0b8e75c0
17 files changed
+1267

No files matched your search

+2
View File
@@ -51,3 +51,5 @@ docker-data/
# PID files
*.pid
agent/aiops-agent
agent/bin/
+54
View File
@@ -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 ':'
+122
View File
@@ -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
}
+133
View File
@@ -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
}
+95
View File
@@ -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
}
+202
View File
@@ -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/<pid>/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 ""
}
+15
View File
@@ -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
+135
View File
@@ -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
+5
View File
@@ -0,0 +1,5 @@
module ezops/agent
go 1.24
require golang.org/x/sys v0.28.0
+2
View File
@@ -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=
+136
View File
@@ -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] + "****"
}
+49
View File
@@ -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)
}
+122
View File
@@ -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
}
}
+110
View File
@@ -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,
})
}
+13
View File
@@ -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)
}
}
+71
View File
@@ -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"`
}
+1
View File
@@ -91,5 +91,6 @@ func AutoMigrate(db *gorm.DB) error {
&models.ResourceModel{},
&models.CloudBilling{},
&models.BillingAlertRule{},
&models.AgentReport{},
)
}