refactor(agent): 简化CLI——去掉-p/-D flag,改用子命令 output/discover

- ./aiops-agent output  → Zabbix风格输出全部上报指标(与API一致,含Heartbeat)
- ./aiops-agent discover → Zabbix LLD端口发现JSON
- ./aiops-agent -server URL → 正常采集模式(不变)
- runOutput 构建与 collectAndReport 完全相同的 Report 结构体
- 新增 net.tcp.close_wait 采集(解析/proc/net/tcp state=08)
- 静态编译 (CGO_ENABLED=0) 兼容 Alpine 容器
This commit is contained in:
attacker committed 2026-06-24 14:01:52 +00:00
1 parent d656520428
commit ad714c49f1
3 files changed
+249 -40

No files matched your search

+19
View File
@@ -44,6 +44,7 @@ func (c *NetworkCollector) Collect() (model.NetworkStat, error) {
stat.TCPTimeWait = tcpStat.timeWait
stat.TCPTotal = tcpStat.total
}
stat.TCPCloseWait = countCloseWait()
// 2. 采集网卡流量(取所有非 lo 网卡合计)
curRx, curTx, nicName, err := readNetDevTotal()
@@ -169,3 +170,21 @@ func readNetDevTotal() (rxBytes, txBytes uint64, nicName string, err error) {
return rxBytes, txBytes, nicName, scanner.Err()
}
// countCloseWait 从 /proc/net/tcp 统计 CLOSE_WAIT 连接数
func countCloseWait() int {
count := 0
for _, path := range []string{"/proc/net/tcp", "/proc/net/tcp6"} {
data, err := os.ReadFile(path)
if err != nil {
continue
}
for _, line := range strings.Split(string(data), "\n") {
fields := strings.Fields(line)
if len(fields) >= 4 && fields[3] == "08" {
count++
}
}
}
return count
}
+229 -40
View File
@@ -3,25 +3,27 @@
// 并通过 HTTP 上报至 AIOps 后端。
// v2.1: 共享 HTTP client + 独立探针拉取 + 配置结构体
//
// 优化要点(1000 台服务器规模):
// - 启动随机 Jitter 避免惊群效应
// - 连接池复用 HTTP Keep-Alive
// - 进程 TOP 限制为 3(可配置)
// - Agent 自检心跳(序列号丢包检测)
// - 端口变化率采集(增量检测,减少 payload)
// 用法:
//
// ./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"
@@ -37,7 +39,7 @@ var (
buildTime = "unknown"
)
// Config Agent 运行配置(v2.1 从 flag 集中解析)
// Config Agent 运行配置
type Config struct {
Server string
Token string
@@ -48,7 +50,7 @@ type Config struct {
func main() {
cfg := Config{
Server: *flag.String("server", "http://localhost:8080", "AIOps 后端上报地址"),
Server: *flag.String("server", "", "AIOps 后端上报地址(output/discover 模式不需要)"),
Token: *flag.String("token", "", "Bearer 鉴权令牌"),
Hostname: *flag.String("hostname", "", "上报主机名,默认使用系统主机名"),
Interval: *flag.Int("interval", 10, "采集与上报间隔,单位:秒"),
@@ -56,8 +58,6 @@ func main() {
}
flag.Parse()
logger := log.New(os.Stdout, "[aiops-agent] ", log.LstdFlags|log.Lmicroseconds)
if cfg.Hostname == "" {
if h, err := os.Hostname(); err == nil {
cfg.Hostname = h
@@ -66,20 +66,35 @@ func main() {
}
}
// 子命令:第一个非 flag 参数
cmd := "run"
if flag.NArg() > 0 {
cmd = flag.Arg(0)
}
switch cmd {
case "output":
runOutput(cfg)
return
case "discover":
runDiscover()
return
}
// ── 正常采集上报模式 ──
logger := log.New(os.Stderr, "[aiops-agent] ", log.LstdFlags|log.Lmicroseconds)
if cfg.Server == "" {
logger.Fatal("参数 -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 v%s (built %s) go=%s", version, buildTime, runtime.Version())
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)
// ═══ 共享 HTTP 客户端(Reporter + TaskClient 共用)═══
httpCli := agentclient.New(cfg.Server, cfg.Token, logger)
// ═══ 初始化采集器 ═══
cpuCollector := collector.NewCPUCollector()
memCollector := collector.NewMemCollector()
diskCollector := collector.NewDiskCollector()
@@ -93,21 +108,17 @@ func main() {
csCollector := collector.NewContextSwitchCollector()
rep := reporter.New(httpCli, logger)
// 系统静态信息(仅启动时一次)
sysInfo := collector.CollectSystem()
logger.Printf("系统信息: OS=%q Kernel=%q", sysInfo.OSVersion, sysInfo.Kernel)
// 探测本机 IP
localIP := detectLocalIP()
logger.Printf("探测本机 IP: %s", localIP)
sysReportCount := 0
// 任务执行器 + 任务客户端
taskExecutor := task.New(logger)
taskClient := task.NewClient(httpCli, cfg.Hostname, logger)
// ═══ 启动随机 Jitter ═══
startupJitter := time.Duration(rand.Intn(cfg.Interval)) * time.Second
logger.Printf("启动 Jitter: %v(防止惊群效应)", startupJitter)
@@ -117,7 +128,6 @@ func main() {
var reportSeq uint64
var probeConfigs []model.ProbeConfig
// ═══ collectAndReport 采集 + 上报 ═══
collectAndReport := func() {
report := &model.Report{
Hostname: cfg.Hostname,
@@ -135,7 +145,6 @@ func main() {
sysReportCount++
}
// 系统指标采集
if stat, err := cpuCollector.Collect(); err != nil {
logger.Printf("CPU 采集异常: %v", err)
} else {
@@ -187,12 +196,10 @@ func main() {
report.ContextSwitches = csStat
}
// v2.1 健康探针(使用缓存配置)
if len(probeConfigs) > 0 {
report.ProbeResults = healthProbeCollector.Execute(ctx, probeConfigs)
}
// Agent 自检心跳
reportSeq++
var memStats runtime.MemStats
runtime.ReadMemStats(&memStats)
@@ -204,22 +211,15 @@ func main() {
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%% swap=%.1f%% disks=%d ports=%d procs=%d fd=%d/%d cs=%.0f/s probes=%d",
logger.Printf("上报成功: cpu=%.1f%% mem=%.1f%% disks=%d ports=%d procs=%d fd=%d/%d probes=%d",
report.CPU.UsagePercent,
report.Memory.UsagePercent,
func() float64 {
if report.Swap != nil {
return report.Swap.UsagePercent
}
return 0
}(),
len(report.Disks), len(report.Ports), len(report.ProcessTop),
func() int {
if report.FD != nil {
@@ -233,17 +233,10 @@ func main() {
}
return 0
}(),
func() float64 {
if report.ContextSwitches != nil {
return report.ContextSwitches.CTXPerSec
}
return 0
}(),
probeCount,
)
}
// ═══ pollAndExecute 任务轮询 ═══
pollAndExecute := func() {
tasks, err := taskClient.Poll(ctx)
if err != nil {
@@ -269,7 +262,6 @@ func main() {
}
}
// ═══ pullProbes 探针配置拉取(v2.1 独立于上报)═══
pullProbes := func() {
configs, err := taskClient.PollProbes(ctx)
if err != nil {
@@ -284,7 +276,6 @@ func main() {
}
}
// ═══ 任务轮询 + 探针拉取协程(5s 间隔)═══
const pollInterval = 5 * time.Second
go func() {
pollAndExecute()
@@ -302,7 +293,6 @@ func main() {
}
}()
// ═══ 主循环:采集 + 上报 ═══
time.Sleep(startupJitter)
collectAndReport()
ticker := time.NewTicker(time.Duration(cfg.Interval) * time.Second)
@@ -318,6 +308,205 @@ func main() {
}
}
// ── 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("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)
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.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)
}
// 端口 (监听)
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))
// 网络 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)
pd("net.rx_bps", report.Network.NetRxBPS)
pd("net.tx_bps", report.Network.NetTxBPS)
// 磁盘 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)
// 进程 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))
}
// ── runDiscover: Zabbix LLD 端口发现 ──
func runDiscover() {
port := collector.NewPortCollector()
ports, err := port.Collect()
if err != nil {
fmt.Println(`{"data":[]}`)
return
}
type lldItem struct {
PName string `json:"{#PNAME}"`
PPort string `json:"{#PPORT}"`
}
items := make([]lldItem, 0, len(ports))
for _, p := range ports {
items = append(items, lldItem{
PName: p.Process,
PPort: strconv.Itoa(p.Port),
})
}
out, _ := json.Marshal(map[string]interface{}{"data": items})
fmt.Println(string(out))
}
// detectLocalIP 探测本机主 IP
func detectLocalIP() string {
conn, err := net.Dial("udp", "8.8.8.8:53")
+1
View File
@@ -4,6 +4,7 @@ package model
type NetworkStat struct {
TCPEstablished int `json:"tcp_established"` // ESTABLISHED 连接数
TCPTimeWait int `json:"tcp_time_wait"` // TIME_WAIT 连接数
TCPCloseWait int `json:"tcp_close_wait"` // CLOSE_WAIT 连接数
TCPTotal int `json:"tcp_total"` // TCP 总连接数
NicName string `json:"nic_name"` // 主网卡名
NetRxBPS float64 `json:"net_rx_bps"` // 网卡接收速率 (bytes/s)