Files
aiops/agent/main.go
T
Logan.Li 59320aa410 perf: 修复 Agent 校验 CPU 热点并实现指标分层聚合与频率下发
CPU 根因:每个 Agent 请求都做一次 bcrypt 校验(实测 72ms),
30 台主机 18 req/s × 72ms ≈ 1.35 核,实测容器 CPU 135%。

- agentauth: 校验结果按 Token 哈希缓存(TTL 5min,轮换即失效),
  last_used_at 写入按 60s 节流。实测 CPU 135% → 1.6%,请求延迟 103ms → 1.6ms
- metricagg: 新增后台逐级降采样聚合 raw→1m/5m→1h→6h/1d,取代前端手动触发
- metrics_aggregate: 补 uniq_metric_agg 唯一索引(原 idx_agg 非唯一导致 upsert 插重复行)
- agent 合并配置接口 GET /agent/config:一次返回任务+探针+端口策略+频率,
  配置轮询 5s→30s;服务端可下发 report_interval,Agent 热更新 ticker 无需重装
- retention: agent_reports 保留期 7d→1d(趋势由聚合表承担)
- redis: 增加 maxmemory 512mb + volatile-lru
- fix(upgrade): 升级脚本改为 fail-fast,修复下载失败仍报 UPGRADE_OK 的静默谎报

// Logan.Li
2026-09-14 08:23:31 +00:00

612 lines
17 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// Package main 是 AIOps Agent 的入口。
// Agent 定时采集 CPU、内存、磁盘、端口、进程 TOP、Swap、FD、上下文切换等指标,
// 并通过 HTTP 上报至 AIOps 后端。
// v2.1: 共享 HTTP client + 独立探针拉取 + 配置结构体
//
// 用法:
//
// ./aiops-agent output # 打印全部待上报指标(Zabbix 风格 key [type|value])
// ./aiops-agent discover # 端口发现(Zabbix LLD JSON)
// ./aiops-agent -server URL -token T # 正常采集上报模式
//
// 署名 Logan.Li
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"log"
"math/rand"
"net"
"os"
"os/signal"
"runtime"
"strconv"
"syscall"
"time"
agentclient "ezops/agent/client"
"ezops/agent/collector"
"ezops/agent/model"
"ezops/agent/reporter"
"ezops/agent/task"
)
var (
version = "v2.2.0"
buildTime = "unknown"
)
// Config Agent 运行配置
type Config struct {
Server string
Token string
Hostname string
Interval int
TopN int
}
func main() {
serverFlag := flag.String("server", "", "AIOps 后端上报地址(output/discover 模式不需要)")
tokenFlag := flag.String("token", "", "Bearer 鉴权令牌")
hostnameFlag := flag.String("hostname", "", "上报主机名,默认使用系统主机名")
intervalFlag := flag.Int("interval", 10, "采集与上报间隔,单位:秒")
topNFlag := flag.Int("top-n", 3, "进程 TOP N 数量")
flag.Parse()
cfg := Config{
Server: *serverFlag,
Token: *tokenFlag,
Hostname: *hostnameFlag,
Interval: *intervalFlag,
TopN: *topNFlag,
}
if cfg.Hostname == "" {
if h, err := os.Hostname(); err == nil {
cfg.Hostname = h
} else {
cfg.Hostname = "unknown"
}
}
// 子命令:第一个非 flag 参数
cmd := "run"
if flag.NArg() > 0 {
cmd = flag.Arg(0)
}
switch cmd {
case "version", "v", "-v", "--version":
fmt.Printf("AIOps Agent %s (built %s) go=%s\n", version, buildTime, runtime.Version())
return
case "output":
runOutput(cfg)
return
}
// ── 正常采集上报模式 ──
logger := log.New(os.Stderr, "[aiops-agent] ", log.LstdFlags|log.Lmicroseconds)
if cfg.Server == "" {
logger.Fatal("参数 -server 不能为空(调试请用: ./aiops-agent output)")
}
if cfg.Interval < 5 {
logger.Fatalf("参数 -interval 必须 >= 5,当前为 %d(1000 台规模推荐 30~60s)", cfg.Interval)
}
logger.Printf("AIOps Agent %s (built %s) go=%s", version, buildTime, runtime.Version())
logger.Printf("配置: server=%s interval=%ds top-n=%d hostname=%q", cfg.Server, cfg.Interval, cfg.TopN, cfg.Hostname)
httpCli := agentclient.New(cfg.Server, cfg.Token, logger)
cpuCollector := collector.NewCPUCollector()
memCollector := collector.NewMemCollector()
diskCollector := collector.NewDiskCollector()
portCollector := collector.NewPortCollector()
netCollector := collector.NewNetworkCollector()
diskIOCollector := collector.NewDiskIOCollector()
procTopCollector := collector.NewProcessTopCollector()
swapCollector := collector.NewSwapCollector()
healthProbeCollector := collector.NewHealthProbeCollector()
fdCollector := collector.NewFDCollector()
csCollector := collector.NewContextSwitchCollector()
rep := reporter.New(httpCli, logger)
sysInfo := collector.CollectSystem()
logger.Printf("系统信息: OS=%q Kernel=%q", sysInfo.OSVersion, sysInfo.Kernel)
agentStartedAt := time.Now()
localIP := detectLocalIP()
logger.Printf("探测本机 IP: %s", localIP)
taskExecutor := task.New(logger)
taskClient := task.NewClient(httpCli, cfg.Hostname, logger)
startupJitter := time.Duration(rand.Intn(cfg.Interval)) * time.Second
logger.Printf("启动 Jitter: %v(防止惊群效应)", startupJitter)
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
var reportSeq uint64
var probeConfigs []model.ProbeConfig
collectAndReport := func() {
currentUptime := sysInfo.UptimeSeconds + int64(time.Since(agentStartedAt).Seconds())
report := &model.Report{
Hostname: cfg.Hostname,
IP: localIP,
Timestamp: time.Now().UTC().Format(time.RFC3339),
AgentVersion: version,
System: &model.SystemStat{
OSVersion: sysInfo.OSVersion,
Kernel: sysInfo.Kernel,
Uptime: currentUptime,
},
}
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 swap, err := swapCollector.Collect(); err != nil {
logger.Printf("Swap 采集异常: %v", err)
} else if swap != nil {
report.Swap = swap
}
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 netStat, err := netCollector.Collect(); err != nil {
logger.Printf("网络采集异常: %v", err)
} else {
report.Network = netStat
}
if dioStat, err := diskIOCollector.Collect(); err != nil {
logger.Printf("磁盘IO采集异常: %v", err)
} else {
report.DiskIO = dioStat
}
if procs, err := procTopCollector.Collect(cfg.TopN); err != nil {
logger.Printf("进程TOP采集异常: %v", err)
} else {
report.ProcessTop = procs
}
if fdStat, err := fdCollector.Collect(); err != nil {
logger.Printf("FD采集异常: %v", err)
} else {
report.FD = fdStat
}
if csStat, err := csCollector.Collect(); err != nil {
// 静默跳过
} else {
report.ContextSwitches = csStat
}
if len(probeConfigs) > 0 {
report.ProbeResults = healthProbeCollector.Execute(ctx, probeConfigs)
}
reportSeq++
var memStats runtime.MemStats
runtime.ReadMemStats(&memStats)
report.Heartbeat = model.AgentHeartbeat{
Version: version,
UptimeSec: currentUptime,
SelfCPU: 0,
SelfRSSMB: float64(memStats.Alloc) / 1024.0 / 1024.0,
ReportSeq: reportSeq,
}
if err := rep.Send(ctx, report); err != nil {
logger.Printf("上报失败: %v", err)
return
}
probeCount := len(report.ProbeResults)
logger.Printf("上报成功: cpu=%.1f%% mem=%.1f%% disks=%d ports=%d procs=%d fd=%d/%d probes=%d",
report.CPU.UsagePercent,
report.Memory.UsagePercent,
len(report.Disks), len(report.Ports), len(report.ProcessTop),
func() int {
if report.FD != nil {
return report.FD.Allocated
}
return 0
}(),
func() int64 {
if report.FD != nil {
return report.FD.Max
}
return 0
}(),
probeCount,
)
}
pollAndExecute := func() {
tasks, err := taskClient.Poll(ctx)
if err != nil {
if ctx.Err() != nil {
return
}
logger.Printf("任务轮询失败: %v", err)
return
}
if len(tasks) > 0 {
logger.Printf("拉取到 %d 个待执行任务", len(tasks))
for _, t := range tasks {
if ctx.Err() != nil {
return
}
result := taskExecutor.ExecuteTask(ctx, t)
if err := taskClient.Submit(ctx, result); err != nil {
logger.Printf("任务 %s 结果回传失败: %v", t.TaskID, err)
} else {
logger.Printf("任务 %s 结果回传成功 status=%s", t.TaskID, result.Status)
}
}
}
}
pullProbes := func() {
configs, err := taskClient.PollProbes(ctx)
if err != nil {
if ctx.Err() != nil {
return
}
logger.Printf("探针配置拉取失败: %v", err)
return
}
if len(configs) > 0 {
probeConfigs = configs
}
}
pullPortPolicy := func() {
policy, err := taskClient.PollPortPolicy(ctx)
if err != nil {
if ctx.Err() != nil {
return
}
// 静默忽略 — 端口策略为可选配置
return
}
if policy != nil {
portCollector.SetPolicy(policy)
}
}
// ── 配置轮询 + 上报循环(v2.2)────────────────────────────────
//
// 旧版每 5 秒分别拉取 tasks/poll、health-probes/config、port-policy 三个接口,
// 配置类请求量是上报本身的 6 倍;现改为单次 GET /agent/config,
// 并由服务端下发 report_interval,Agent 热更新 ticker、无需重装。 // Logan.Li
const minInterval = 5 * time.Second
const fallbackConfigPollInterval = 30 * time.Second
intervalUpdates := make(chan time.Duration, 1)
configTicker := time.NewTicker(fallbackConfigPollInterval)
go func() {
defer configTicker.Stop()
runConfig := func() {
merged, err := taskClient.PollConfig(ctx)
if err != nil {
if ctx.Err() != nil {
return
}
// 服务端尚未升级(或暂时失败)时回退旧的分接口模式,保证兼容。 // Logan.Li
pollAndExecute()
pullProbes()
pullPortPolicy()
return
}
// 执行服务端下发的任务。 // Logan.Li
for _, t := range merged.Tasks {
if ctx.Err() != nil {
return
}
result := taskExecutor.ExecuteTask(ctx, t)
if err := taskClient.Submit(ctx, result); err != nil {
logger.Printf("任务 %s 结果回传失败: %v", t.TaskID, err)
}
}
if len(merged.HealthProbes) > 0 {
probeConfigs = merged.HealthProbes
}
if merged.PortPolicy != nil {
portCollector.SetPolicy(merged.PortPolicy)
}
// 服务端下发的上报间隔 → 通知主循环热更新 ticker。 // Logan.Li
if merged.ReportInterval > 0 {
next := time.Duration(merged.ReportInterval) * time.Second
if next < minInterval {
next = minInterval
}
select {
case intervalUpdates <- next:
default:
}
}
// 服务端下发的配置轮询间隔 → 调整本循环节奏。 // Logan.Li
if merged.ConfigPollInterval > 0 {
next := time.Duration(merged.ConfigPollInterval) * time.Second
if next >= minInterval && next != fallbackConfigPollInterval {
configTicker.Reset(next)
}
}
}
runConfig()
for {
select {
case <-configTicker.C:
runConfig()
case <-ctx.Done():
return
}
}
}()
// 上报循环:间隔可被服务端热更新。 // Logan.Li
time.Sleep(startupJitter)
collectAndReport()
reportTicker := time.NewTicker(time.Duration(cfg.Interval) * time.Second)
defer reportTicker.Stop()
for {
select {
case <-reportTicker.C:
collectAndReport()
case next := <-intervalUpdates:
// 重新抖动,避免多台主机在服务端下发同值后被对齐成同一时刻上报。 // Logan.Li
reportTicker.Reset(next)
logger.Printf("服务端下发新的上报间隔: %s", next)
case <-ctx.Done():
logger.Println("收到退出信号,Agent 正常关闭")
return
}
}
}
// ── runOutput: 构建与 daemon 模式完全一致的 Report,Zabbix-agent -p 风格输出 ──
func runOutput(cfg Config) {
cpu := collector.NewCPUCollector()
mem := collector.NewMemCollector()
disk := collector.NewDiskCollector()
port := collector.NewPortCollector()
netc := collector.NewNetworkCollector()
dio := collector.NewDiskIOCollector()
ptop := collector.NewProcessTopCollector()
swap := collector.NewSwapCollector()
fd := collector.NewFDCollector()
cs := collector.NewContextSwitchCollector()
sysInfo := collector.CollectSystem()
localIP := detectLocalIP()
// 构建与 daemon collectAndReport 完全一致的 Report
report := &model.Report{
Hostname: cfg.Hostname,
IP: localIP,
Timestamp: time.Now().UTC().Format(time.RFC3339),
AgentVersion: version,
System: &model.SystemStat{
OSVersion: sysInfo.OSVersion,
Kernel: sysInfo.Kernel,
Uptime: sysInfo.UptimeSeconds,
},
}
if stat, err := cpu.Collect(); err == nil {
report.CPU = stat
}
if stat, err := mem.Collect(); err == nil {
report.Memory = stat
}
if stat, err := swap.Collect(); err == nil && stat != nil {
report.Swap = stat
}
if disks, err := disk.Collect(); err == nil {
report.Disks = disks
}
if ports, err := port.Collect(); err == nil {
report.Ports = ports
}
if netStat, err := netc.Collect(); err == nil {
report.Network = netStat
}
if dioStat, err := dio.Collect(); err == nil {
report.DiskIO = dioStat
}
if procs, err := ptop.Collect(cfg.TopN); err == nil {
report.ProcessTop = procs
}
if fdStat, err := fd.Collect(); err == nil {
report.FD = fdStat
}
if csStat, err := cs.Collect(); err == nil {
report.ContextSwitches = csStat
}
var memStats runtime.MemStats
runtime.ReadMemStats(&memStats)
report.Heartbeat = model.AgentHeartbeat{
Version: version,
UptimeSec: sysInfo.UptimeSeconds,
SelfCPU: 0,
SelfRSSMB: float64(memStats.Alloc) / 1024.0 / 1024.0,
ReportSeq: 1,
}
// ── Zabbix 风格打印 ──
p := func(key, typ, val string) { fmt.Printf("%-48s [%s|%s]\n", key, typ, val) }
ps := func(key, val string) { p(key, "s", val) }
pd := func(key string, val float64) { p(key, "d", fmt.Sprintf("%.4f", val)) }
pu := func(key string, val int) { p(key, "u", strconv.Itoa(val)) }
pu64 := func(key string, val int64) { p(key, "u", strconv.FormatInt(val, 10)) }
// Agent 身份
ps("agent.hostname", report.Hostname)
ps("system.hostname", report.Hostname)
ps("agent.version", report.AgentVersion)
ps("agent.ip", report.IP)
// 系统
if report.System != nil {
ps("system.os", report.System.OSVersion)
ps("system.kernel", report.System.Kernel)
pu64("system.uptime", report.System.Uptime)
}
// CPU
pd("cpu.usage_percent", report.CPU.UsagePercent)
pd("cpu.iowait", report.CPU.IOWait)
pu("cpu.core_count", report.CPU.CoreCount)
pd("cpu.load1", report.CPU.Load1)
pd("cpu.load5", report.CPU.Load5)
pd("cpu.load15", report.CPU.Load15)
// 内存
pd("memory.total_gb", report.Memory.TotalGB)
pd("memory.used_gb", report.Memory.UsedGB)
pd("memory.cached_gb", report.Memory.CachedGB)
pd("memory.available_gb", report.Memory.AvailableGB)
pd("memory.swap_used_gb", report.Memory.SwapUsedGB)
pd("memory.usage_percent", report.Memory.UsagePercent)
// Swap
if report.Swap != nil {
pd("swap.total_gb", report.Swap.TotalGB)
pd("swap.used_gb", report.Swap.UsedGB)
pd("swap.usage_percent", report.Swap.UsagePercent)
}
// 磁盘
for _, d := range report.Disks {
prefix := "disk[" + d.Mount + "]"
pd(prefix+".total_gb", d.TotalGB)
pd(prefix+".used_gb", d.UsedGB)
pd(prefix+".usage_percent", d.UsagePercent)
pd(prefix+".inode_usage_pct", d.InodeUsagePct)
}
// 端口 (监听)
for _, pt := range report.Ports {
ps("net.listen["+strconv.Itoa(pt.Port)+"].process", pt.Process)
ps("net.listen["+strconv.Itoa(pt.Port)+"].protocol", pt.Protocol)
}
pu("net.listen.count", len(report.Ports))
// 端口发现 Zabbix LLD JSON(合并为一行)
if len(report.Ports) > 0 {
type lldItem struct {
PName string `json:"{#PNAME}"`
PPort string `json:"{#PPORT}"`
}
items := make([]lldItem, 0, len(report.Ports))
for _, pt := range report.Ports {
items = append(items, lldItem{PName: pt.Process, PPort: strconv.Itoa(pt.Port)})
}
if jsonBytes, err := json.Marshal(map[string]interface{}{"data": items}); err == nil {
ps("ports.discovery[*]", string(jsonBytes))
}
}
// 网络 TCP
pu("net.tcp.established", report.Network.TCPEstablished)
pu("net.tcp.time_wait", report.Network.TCPTimeWait)
pu("net.tcp.close_wait", report.Network.TCPCloseWait)
pu("net.tcp.total", report.Network.TCPTotal)
pu("net.connection.total", report.Network.TCPEstablished+report.Network.TCPTimeWait+report.Network.TCPCloseWait)
pd("net.rx_bps", report.Network.NetRxBPS)
pd("net.tx_bps", report.Network.NetTxBPS)
pu64("net.rx_errors", report.Network.NetRxErrors)
pu64("net.tx_errors", report.Network.NetTxErrors)
// 磁盘 IO
pd("disk_io.read_bps", report.DiskIO.ReadBPS)
pd("disk_io.write_bps", report.DiskIO.WriteBPS)
pd("disk_io.read_iops", report.DiskIO.ReadIOPS)
pd("disk_io.write_iops", report.DiskIO.WriteIOPS)
pd("disk_io.util", report.DiskIO.Util)
// 进程 TOP
for i, pr := range report.ProcessTop {
prefix := fmt.Sprintf("proc.top[%d]", i+1)
ps(prefix+".name", pr.Name)
if pr.Cmdline != "" {
ps(prefix+".cmdline", pr.Cmdline)
}
pu(prefix+".pid", pr.PID)
ps(prefix+".user", pr.User)
pd(prefix+".cpu_pct", pr.CPUPct)
pd(prefix+".mem_pct", pr.MemPct)
pd(prefix+".memory_mb", pr.MemoryMB)
pu(prefix+".fd_count", pr.FDCount)
pu(prefix+".threads", pr.Threads)
}
// FD
if report.FD != nil {
pu("fd.allocated", report.FD.Allocated)
pu64("fd.max", report.FD.Max)
pd("fd.usage_pct", report.FD.UsagePct)
}
// 上下文切换
if report.ContextSwitches != nil {
pd("ctx.ctx_per_sec", report.ContextSwitches.CTXPerSec)
pd("ctx.irq_per_sec", report.ContextSwitches.IRQPerSec)
}
// Heartbeat(Agent 自检)
ps("agent.heartbeat.version", report.Heartbeat.Version)
pu64("agent.heartbeat.uptime_sec", report.Heartbeat.UptimeSec)
pd("agent.heartbeat.self_rss_mb", report.Heartbeat.SelfRSSMB)
pu64("agent.heartbeat.report_seq", int64(report.Heartbeat.ReportSeq))
}
// detectLocalIP 探测本机主 IP
func detectLocalIP() string {
conn, err := net.Dial("udp", "8.8.8.8:53")
if err == nil {
defer conn.Close()
if addr, ok := conn.LocalAddr().(*net.UDPAddr); ok && addr.IP != nil {
return addr.IP.String()
}
}
ifaces, err := net.Interfaces()
if err == nil {
for _, iface := range ifaces {
if iface.Flags&net.FlagLoopback != 0 || iface.Flags&net.FlagUp == 0 {
continue
}
addrs, _ := iface.Addrs()
for _, addr := range addrs {
if ipnet, ok := addr.(*net.IPNet); ok && ipnet.IP.To4() != nil {
return ipnet.IP.String()
}
}
}
}
return "unknown"
}