aiops/agent/main.go

// Package main 是 AIOps Agent 的入口。
// Agent 定时采集 CPU、内存、磁盘、端口、进程 TOP、Swap、FD、上下文切换等指标，
// 并通过 HTTP 上报至 AIOps 后端。
// v2.1: 共享 HTTP client + 独立探针拉取 + 配置结构体
//
// 优化要点（1000 台服务器规模）：
//   - 启动随机 Jitter 避免惊群效应
//   - 连接池复用 HTTP Keep-Alive
//   - 进程 TOP 限制为 3（可配置）
//   - Agent 自检心跳（序列号丢包检测）
//   - 端口变化率采集（增量检测，减少 payload）
//
// 署名 Logan.Li
package main

import (
	"context"
	"flag"
	"log"
	"math/rand"
	"net"
	"os"
	"os/signal"
	"runtime"
	"syscall"
	"time"

	agentclient "ezops/agent/client"
	"ezops/agent/collector"
	"ezops/agent/model"
	"ezops/agent/reporter"
	"ezops/agent/task"
)

var (
	version   = "v2.1.0"
	buildTime = "unknown"
)

// Config Agent 运行配置（v2.1 从 flag 集中解析）
type Config struct {
	Server   string
	Token    string
	Hostname string
	Interval int
	TopN     int
}

func main() {
	cfg := Config{
		Server:   *flag.String("server", "http://localhost:8080", "AIOps 后端上报地址"),
		Token:    *flag.String("token", "", "Bearer 鉴权令牌"),
		Hostname: *flag.String("hostname", "", "上报主机名，默认使用系统主机名"),
		Interval: *flag.Int("interval", 10, "采集与上报间隔，单位：秒"),
		TopN:     *flag.Int("top-n", 3, "进程 TOP N 数量"),
	}
	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
		} else {
			cfg.Hostname = "unknown"
		}
	}

	if cfg.Server == "" {
		logger.Fatal("参数 -server 不能为空")
	}
	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("配置: 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()
	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)

	// 探测本机 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)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	var reportSeq uint64
	var probeConfigs []model.ProbeConfig

	// ═══ collectAndReport 采集 + 上报 ═══
	collectAndReport := func() {
		report := &model.Report{
			Hostname:     cfg.Hostname,
			IP:           localIP,
			Timestamp:    time.Now().UTC().Format(time.RFC3339),
			AgentVersion: version,
		}

		if sysReportCount < 3 && (sysInfo.OSVersion != "" || sysInfo.Kernel != "") {
			report.System = &model.SystemStat{
				OSVersion: sysInfo.OSVersion,
				Kernel:    sysInfo.Kernel,
				Uptime:    sysInfo.UptimeSeconds,
			}
			sysReportCount++
		}

		// 系统指标采集
		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
		}

		// v2.1 健康探针（使用缓存配置）
		if len(probeConfigs) > 0 {
			report.ProbeResults = healthProbeCollector.Execute(ctx, probeConfigs)
		}

		// Agent 自检心跳
		reportSeq++
		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: 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",
			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 {
					return report.FD.Allocated
				}
				return 0
			}(),
			func() int64 {
				if report.FD != nil {
					return report.FD.Max
				}
				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 {
			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 探针配置拉取（v2.1 独立于上报）═══
	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
		}
	}

	// ═══ 任务轮询 + 探针拉取协程（5s 间隔）═══
	const pollInterval = 5 * time.Second
	go func() {
		pollAndExecute()
		pullProbes()
		ticker := time.NewTicker(pollInterval)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				pollAndExecute()
				pullProbes()
			case <-ctx.Done():
				return
			}
		}
	}()

	// ═══ 主循环：采集 + 上报 ═══
	time.Sleep(startupJitter)
	collectAndReport()
	ticker := time.NewTicker(time.Duration(cfg.Interval) * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			collectAndReport()
		case <-ctx.Done():
			logger.Println("收到退出信号，Agent 正常关闭")
			return
		}
	}
}

// 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"
}