Logan.Li
9360ebefd8
配置轮询每 30s 都会收到相同的 report_interval 并触发 ticker.Reset， 而 Reset 会重新计时 —— 30s 的 ticker 永远等不到触发， 实际上报被无形拖长到 60~90s（实测 17 次/60s，应为 52 次）。 改为记录已生效的间隔，仅在值真正变化时才 Reset。 配置轮询间隔同样处理，否则每次轮询都 Reset 会让 ticker 永不触发。 验证：升级后上报 53 次/60s，Redis 采样时间戳精确 30s 间隔。 // Logan.Li
621 lines
18 KiB
Go
621 lines
18 KiB
Go
// 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 = "v1.0.2"
|
||
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)
|
||
// appliedReportInterval 记录已生效的上报间隔（秒）。
|
||
// 只在值真正变化时才下发重置信号：否则配置轮询每 30s 推一次相同值，
|
||
// ticker.Reset 会不断重新计时，导致 30s 的 ticker 永远等不到触发、
|
||
// 上报被无形拖长到 60~90s。 // Logan.Li
|
||
appliedReportInterval := cfg.Interval
|
||
appliedConfigPollInterval := int(fallbackConfigPollInterval / time.Second)
|
||
|
||
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 && merged.ReportInterval != appliedReportInterval {
|
||
next := time.Duration(merged.ReportInterval) * time.Second
|
||
if next < minInterval {
|
||
next = minInterval
|
||
}
|
||
appliedReportInterval = merged.ReportInterval
|
||
select {
|
||
case intervalUpdates <- next:
|
||
default:
|
||
}
|
||
}
|
||
// 服务端下发的配置轮询间隔 → 同样仅在值变化时重置，
|
||
// 否则每次轮询都 Reset 会让 ticker 永远不触发。 // Logan.Li
|
||
if merged.ConfigPollInterval > 0 && merged.ConfigPollInterval != appliedConfigPollInterval {
|
||
next := time.Duration(merged.ConfigPollInterval) * time.Second
|
||
if next >= minInterval {
|
||
appliedConfigPollInterval = merged.ConfigPollInterval
|
||
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"
|
||
}
|