From ad714c49f1ad5a7e9829be4c679dc361186beb6a Mon Sep 17 00:00:00 2001
From: attacker <attacker@gitee.com>
Date: Wed, 24 Jun 2026 14:01:52 +0000
Subject: [PATCH] =?UTF-8?q?refactor(agent):=20=E7=AE=80=E5=8C=96CLI?=
 =?UTF-8?q?=E2=80=94=E2=80=94=E5=8E=BB=E6=8E=89-p/-D=20flag=EF=BC=8C?=
 =?UTF-8?q?=E6=94=B9=E7=94=A8=E5=AD=90=E5=91=BD=E4=BB=A4=20output/discover?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- ./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 容器
---
 agent/collector/network.go |  19 +++
 agent/main.go              | 269 +++++++++++++++++++++++++++++++------
 agent/model/model_ext.go   |   1 +
 3 files changed, 249 insertions(+), 40 deletions(-)

diff --git a/agent/collector/network.go b/agent/collector/network.go
index 18d7aef..0cebdeb 100644
--- a/agent/collector/network.go
+++ b/agent/collector/network.go
@@ -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
+}
diff --git a/agent/main.go b/agent/main.go
index 55445cf..416cbed 100644
--- a/agent/main.go
+++ b/agent/main.go
@@ -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")
diff --git a/agent/model/model_ext.go b/agent/model/model_ext.go
index 0a4af98..fec7487 100644
--- a/agent/model/model_ext.go
+++ b/agent/model/model_ext.go
@@ -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)