From d656520428f02c7a1999d1aab08f9eada7b6d34f Mon Sep 17 00:00:00 2001 From: attacker Date: Wed, 24 Jun 2026 19:18:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20v0.34=20=E5=81=A5=E5=BA=B7=E6=8E=A2?= =?UTF-8?q?=E9=92=88(Architecture=20B)=20+=20Agent=20v2.1=20+=20=E7=AB=AF?= =?UTF-8?q?=E5=8F=A3=E7=9B=91=E6=8E=A7=20+=20=E5=91=8A=E8=AD=A6=E8=81=94?= =?UTF-8?q?=E7=B3=BB=E4=BA=BA=E9=87=8D=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - v0.31: AlertContact 重构(channel_type下拉+webhook_url替代三分立字段) - v0.32: Agent ProcessTop 增强(cmdline/user/cpu_pct/mem_pct) - v0.33: 端口监控 Tab(监听端口自动发现+down联动告警) - v0.34: 健康探针(Agent本地执行HTTP/TCP/Exec→结果随心跳上报) - Agent新增: ContextSwitch/FD/HealthProbe采集器 - 后端新增: health_probe handler + Agent model扩展 - HostMonitor新增Tab: 健康探针/端口监控 --- agent/client/http.go | 140 ++++++ agent/collector/context.go | 39 ++ agent/collector/fd.go | 49 +++ agent/collector/health_probe.go | 216 ++++++++++ agent/collector/proctop.go | 297 ++++++++++--- agent/collector/util.go | 25 ++ agent/main.go | 250 ++++------- agent/model/model.go | 1 + agent/model/model_ext.go | 28 ++ agent/model/model_proctop.go | 16 +- agent/reporter/http.go | 112 +---- agent/task/client.go | 145 ++----- backend/internal/api/handlers/agent.go | 5 +- .../internal/api/handlers/alert_contacts.go | 16 +- backend/internal/api/handlers/health_probe.go | 308 +++++++++++++ backend/internal/api/routes/routes.go | 8 + backend/internal/models/agent.go | 7 +- backend/internal/models/alert.go | 41 +- backend/internal/models/monitor.go | 70 +++ backend/pkg/database/database.go | 1 + .../monitor/alert-center/AlertCenter.tsx | 24 +- .../monitor/alert-center/ContactDialog.tsx | 107 ++--- .../alert-center/alert-center-helpers.tsx | 5 +- frontend/src/pages/HostMonitor.tsx | 404 ++++++++++++++---- frontend/src/services/api.ts | 5 + 25 files changed, 1700 insertions(+), 619 deletions(-) create mode 100644 agent/client/http.go create mode 100644 agent/collector/context.go create mode 100644 agent/collector/fd.go create mode 100644 agent/collector/health_probe.go create mode 100644 agent/collector/util.go create mode 100644 backend/internal/api/handlers/health_probe.go diff --git a/agent/client/http.go b/agent/client/http.go new file mode 100644 index 0000000..1c0704e --- /dev/null +++ b/agent/client/http.go @@ -0,0 +1,140 @@ +// Package client 提供 Agent 与 AIOps 后端通信的共享 HTTP 客户端。 +// Reporter 和 TaskClient 共用此客户端，避免重复实现鉴权、重试、上下文传播。 +// +// 署名 Logan.Li +package client + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" + "time" +) + +const ( + defaultTimeout = 10 * time.Second + maxRetries = 3 + userAgent = "aiops-agent/2.1" +) + +// Client 共享 HTTP 客户端，封装鉴权、路径拼接、重试逻辑。 +type Client struct { + baseURL string + token string + httpClient *http.Client + Logger *log.Logger +} + +// New 创建客户端实例。 +func New(server, token string, logger *log.Logger) *Client { + return &Client{ + baseURL: strings.TrimRight(server, "/"), + token: token, + httpClient: &http.Client{Timeout: defaultTimeout}, + Logger: logger, + } +} + +// Do 执行 HTTP 请求并返回原始响应。调用方负责关闭 resp.Body。 +func (c *Client) Do(ctx context.Context, method, path string, body []byte) (*http.Response, error) { + url := c.baseURL + path + + var lastErr error + for attempt := 1; attempt <= maxRetries; attempt++ { + var reqBody io.Reader + if body != nil { + reqBody = bytes.NewReader(body) + } + + req, err := http.NewRequestWithContext(ctx, method, url, reqBody) + if err != nil { + return nil, fmt.Errorf("构造请求失败: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", userAgent) + if c.token != "" { + req.Header.Set("Authorization", "Bearer "+c.token) + } + + if err := ctx.Err(); err != nil { + return nil, err + } + + resp, err := c.httpClient.Do(req) + if err != nil { + lastErr = err + c.Logger.Printf("[http] 请求失败（第%d次） %s %s: %v", attempt, method, path, err) + if !sleepWithCtx(ctx, attempt) { + return nil, ctx.Err() + } + continue + } + + if resp.StatusCode >= 200 && resp.StatusCode < 300 { + return resp, nil + } + + // 读取错误响应体 + errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + resp.Body.Close() + + lastErr = fmt.Errorf("HTTP %d %s %s: %s", resp.StatusCode, method, path, string(errBody)) + + // 4xx（除 429）不重试 + if resp.StatusCode >= 400 && resp.StatusCode < 500 && resp.StatusCode != http.StatusTooManyRequests { + return nil, lastErr + } + c.Logger.Printf("[http] 响应异常（第%d次） %s %s: HTTP %d", attempt, method, path, resp.StatusCode) + if !sleepWithCtx(ctx, attempt) { + return nil, ctx.Err() + } + } + return nil, lastErr +} + +// DoJSON 执行请求并解析 JSON 响应到 result。自动关闭响应体。 +func (c *Client) DoJSON(ctx context.Context, method, path string, reqBody, result interface{}) error { + var bodyBytes []byte + if reqBody != nil { + var err error + bodyBytes, err = json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("序列化请求体失败: %w", err) + } + } + + resp, err := c.Do(ctx, method, path, bodyBytes) + if err != nil { + return err + } + defer resp.Body.Close() + + if result != nil { + body, err := io.ReadAll(resp.Body) + if err != nil { + return fmt.Errorf("读取响应体失败: %w", err) + } + if err := json.Unmarshal(body, result); err != nil { + return fmt.Errorf("解析响应失败: %w", err) + } + } + return nil +} + +// sleepWithCtx 带退避的休眠，可被 context 取消。 +func sleepWithCtx(ctx context.Context, attempt int) bool { + timer := time.NewTimer(time.Duration(attempt) * time.Second) + defer timer.Stop() + select { + case <-timer.C: + return true + case <-ctx.Done(): + return false + } +} diff --git a/agent/collector/context.go b/agent/collector/context.go new file mode 100644 index 0000000..0fca552 --- /dev/null +++ b/agent/collector/context.go @@ -0,0 +1,39 @@ +// Package collector 的上下文切换采集器。 +// 读取 /proc/stat 中的 ctxt（上下文切换）和 intr（中断）累计值。 +// 注意：此处返回累计值而非速率，后端可用两次上报差值计算。 +// +// 署名 Logan.Li +package collector + +import ( + "fmt" + "os" + + "ezops/agent/model" +) + +// ContextSwitchCollector 上下文切换采集器 +type ContextSwitchCollector struct{} + +// NewContextSwitchCollector 创建采集器实例 +func NewContextSwitchCollector() *ContextSwitchCollector { + return &ContextSwitchCollector{} +} + +// Collect 读取 /proc/stat 中的上下文切换和中断统计 +func (c *ContextSwitchCollector) Collect() (*model.ContextSwitchStat, error) { + data, err := os.ReadFile("/proc/stat") + if err != nil { + return nil, fmt.Errorf("读取 /proc/stat 失败: %w", err) + } + stat := &model.ContextSwitchStat{} + for _, line := range SplitLines(string(data)) { + if HasPrefix(line, "ctxt ") { + fmt.Sscanf(line, "ctxt %f", &stat.CTXPerSec) + } + if HasPrefix(line, "intr ") { + fmt.Sscanf(line, "intr %f", &stat.IRQPerSec) + } + } + return stat, nil +} diff --git a/agent/collector/fd.go b/agent/collector/fd.go new file mode 100644 index 0000000..7ebb9ce --- /dev/null +++ b/agent/collector/fd.go @@ -0,0 +1,49 @@ +// Package collector 的 FD 句柄采集器。 +// 读取 /proc/sys/fs/file-nr 和 /proc/sys/fs/file-max 获取系统级文件句柄使用情况。 +// +// 署名 Logan.Li +package collector + +import ( + "fmt" + "os" + + "ezops/agent/model" +) + +// FDCollector 系统文件句柄采集器 +type FDCollector struct{} + +// NewFDCollector 创建采集器实例 +func NewFDCollector() *FDCollector { + return &FDCollector{} +} + +// Collect 读取 /proc/sys/fs/file-nr 和 file-max +func (c *FDCollector) Collect() (*model.SocketFDStat, error) { + alloc, err := os.ReadFile("/proc/sys/fs/file-nr") + if err != nil { + return nil, fmt.Errorf("读取 file-nr 失败: %w", err) + } + var allocated, _unused1, _unused2 int + fmt.Sscanf(string(alloc), "%d %d %d", &allocated, &_unused1, &_unused2) + + max, err := os.ReadFile("/proc/sys/fs/file-max") + if err != nil { + return nil, fmt.Errorf("读取 file-max 失败: %w", err) + } + var maxFD int64 + fmt.Sscanf(string(max), "%d", &maxFD) + + stat := &model.SocketFDStat{ + Allocated: allocated, + Max: maxFD, + } + if maxFD > 100_000_000 { + stat.Max = -1 + stat.UsagePct = 0 + } else if maxFD > 0 { + stat.UsagePct = float64(allocated) / float64(maxFD) * 100.0 + } + return stat, nil +} diff --git a/agent/collector/health_probe.go b/agent/collector/health_probe.go new file mode 100644 index 0000000..e670feb --- /dev/null +++ b/agent/collector/health_probe.go @@ -0,0 +1,216 @@ +// Package collector 的健康探针采集器。 +// 在后端下发的探针配置驱动下，本地执行 HTTP GET / TCP 连接 / 命令执行， +// 并将结果返回供上报。 +// +// 署名 Logan.Li +package collector + +import ( + "context" + "fmt" + "io" + "net" + "net/http" + "os/exec" + "strings" + "time" + + "ezops/agent/model" +) + +// HealthProbeCollector 健康探针执行器 +type HealthProbeCollector struct { + httpClient *http.Client +} + +// NewHealthProbeCollector 创建采集器实例 +func NewHealthProbeCollector() *HealthProbeCollector { + return &HealthProbeCollector{ + httpClient: &http.Client{ + Timeout: 10 * time.Second, + Transport: &http.Transport{ + DialContext: (&net.Dialer{ + Timeout: 5 * time.Second, + }).DialContext, + }, + }, + } +} + +// Execute 执行所有探针配置，返回结果列表。 +// parentCtx 用于传播取消信号——探针执行尽量尊重主循环的生命周期。 +func (h *HealthProbeCollector) Execute(parentCtx context.Context, configs []model.ProbeConfig) []model.ProbeResult { + results := make([]model.ProbeResult, 0, len(configs)) + + for _, cfg := range configs { + // 检查父上下文是否已取消 + if err := parentCtx.Err(); err != nil { + break + } + + result := model.ProbeResult{ + ProbeID: cfg.ID, + Name: cfg.Name, + ProbeType: cfg.ProbeType, + } + + start := time.Now() + + switch cfg.ProbeType { + case "http_get": + result = h.execHTTP(parentCtx, cfg, start) + case "tcp_socket": + result = h.execTCP(cfg, start) + case "exec_command": + result = h.execCmd(parentCtx, cfg, start) + default: + result.Status = "down" + result.Message = fmt.Sprintf("unknown probe type: %s", cfg.ProbeType) + } + + results = append(results, result) + } + + return results +} + +// execHTTP 执行 HTTP 探测，接收父 context +func (h *HealthProbeCollector) execHTTP(parentCtx context.Context, cfg model.ProbeConfig, start time.Time) model.ProbeResult { + result := model.ProbeResult{ + ProbeID: cfg.ID, + Name: cfg.Name, + ProbeType: cfg.ProbeType, + } + + timeout := time.Duration(cfg.Timeout) * time.Second + if timeout <= 0 { + timeout = 10 * time.Second + } + + ctx, cancel := context.WithTimeout(parentCtx, timeout) + defer cancel() + + method := cfg.HTTPMethod + if method == "" { + method = "GET" + } + + req, err := http.NewRequestWithContext(ctx, method, cfg.HTTPURL, nil) + if err != nil { + result.Status = "down" + result.Message = fmt.Sprintf("构造请求失败: %v", err) + result.LatencyMs = int(time.Since(start).Milliseconds()) + return result + } + req.Header.Set("User-Agent", "aiops-agent-health-probe/2.1") + + resp, err := h.httpClient.Do(req) + latencyMs := int(time.Since(start).Milliseconds()) + result.LatencyMs = latencyMs + + if err != nil { + result.Status = "down" + result.Message = fmt.Sprintf("HTTP请求失败: %v", err) + return result + } + defer resp.Body.Close() + + expectCode := cfg.HTTPExpectCode + if expectCode == 0 { + expectCode = 200 + } + result.StatusCode = resp.StatusCode + + if resp.StatusCode != expectCode { + result.Status = "down" + result.Message = fmt.Sprintf("状态码不匹配: got %d, want %d", resp.StatusCode, expectCode) + return result + } + + // 可选：检查响应体 + if cfg.HTTPExpectBody != "" { + bodyBytes, err := io.ReadAll(io.LimitReader(resp.Body, 64*1024)) + if err != nil { + result.Status = "down" + result.Message = fmt.Sprintf("读取响应体失败: %v", err) + return result + } + if !strings.Contains(string(bodyBytes), cfg.HTTPExpectBody) { + result.Status = "down" + result.Message = fmt.Sprintf("响应体不包含期望字符串: %q", cfg.HTTPExpectBody) + return result + } + } + + result.Status = "up" + result.Message = fmt.Sprintf("HTTP %d OK (%dms)", resp.StatusCode, latencyMs) + return result +} + +// execTCP 执行 TCP 连接探测 +func (h *HealthProbeCollector) execTCP(cfg model.ProbeConfig, start time.Time) model.ProbeResult { + result := model.ProbeResult{ + ProbeID: cfg.ID, + Name: cfg.Name, + ProbeType: cfg.ProbeType, + } + + timeout := time.Duration(cfg.Timeout) * time.Second + if timeout <= 0 { + timeout = 10 * time.Second + } + + addr := fmt.Sprintf("%s:%d", cfg.TCPHost, cfg.TCPPort) + conn, err := net.DialTimeout("tcp", addr, timeout) + latencyMs := int(time.Since(start).Milliseconds()) + result.LatencyMs = latencyMs + + if err != nil { + result.Status = "down" + result.Message = fmt.Sprintf("TCP连接失败 %s: %v", addr, err) + return result + } + conn.Close() + + result.Status = "up" + result.Message = fmt.Sprintf("TCP %s connected (%dms)", addr, latencyMs) + return result +} + +// execCmd 执行命令探测（exit code 0 = up），接收父 context +func (h *HealthProbeCollector) execCmd(parentCtx context.Context, cfg model.ProbeConfig, start time.Time) model.ProbeResult { + result := model.ProbeResult{ + ProbeID: cfg.ID, + Name: cfg.Name, + ProbeType: cfg.ProbeType, + } + + timeout := time.Duration(cfg.Timeout) * time.Second + if timeout <= 0 { + timeout = 10 * time.Second + } + + ctx, cancel := context.WithTimeout(parentCtx, timeout) + defer cancel() + + cmd := exec.CommandContext(ctx, "sh", "-c", cfg.ExecCommand) + output, err := cmd.CombinedOutput() + latencyMs := int(time.Since(start).Milliseconds()) + result.LatencyMs = latencyMs + result.StatusCode = cmd.ProcessState.ExitCode() + + if err != nil { + result.Status = "down" + // 截断输出避免过大 + outMsg := strings.TrimSpace(string(output)) + if len(outMsg) > 200 { + outMsg = outMsg[:200] + "..." + } + result.Message = fmt.Sprintf("命令执行失败(exit=%d): %s", cmd.ProcessState.ExitCode(), outMsg) + return result + } + + result.Status = "up" + result.Message = fmt.Sprintf("命令执行成功 (%dms)", latencyMs) + return result +} diff --git a/agent/collector/proctop.go b/agent/collector/proctop.go index acfa7ef..a4b063b 100644 --- a/agent/collector/proctop.go +++ b/agent/collector/proctop.go @@ -1,6 +1,6 @@ -// Package collector — 进程 TOP3 采集器 -// 扫描 /proc/*/stat + /proc/*/status 获取 CPU/内存/句柄 TOP3 进程 -// 1000 服务器规模下的优化：只取 TOP3，不遍历全部进程的 cmdline +// Package collector — 进程 TOP 采集器 +// 扫描 /proc 获取 TOP N 进程完整信息：PID/User/CMD/CPU%/MEM%/FDs +// 阿里云 ECS 监控风格 — 直观 6 列布局 // // 署名 Logan.Li package collector @@ -8,6 +8,7 @@ package collector import ( "bufio" "os" + "os/user" "path/filepath" "sort" "strconv" @@ -16,33 +17,166 @@ import ( "ezops/agent/model" ) +const clktck = 100.0 // Linux CLK_TCK = 100 Hz + // ProcessTopCollector 进程 TOP 采集器 -type ProcessTopCollector struct{} +type ProcessTopCollector struct { + memTotalKB uint64 // 系统总内存（KB），初始化时读取 + numCPU int // CPU 核心数 +} // NewProcessTopCollector 创建进程 TOP 采集器 func NewProcessTopCollector() *ProcessTopCollector { - return &ProcessTopCollector{} + pt := &ProcessTopCollector{} + pt.memTotalKB = readMemTotalKB() + pt.numCPU = readNumCPU() + if pt.numCPU < 1 { + pt.numCPU = 1 + } + return pt } -// procSample holds key metrics for a single process -type procSample struct { - PID int - Name string - CPU float64 // CPU 使用率（需两次采样 + 系统总 jiffies 才能算，这里用 utime+stime ticks 近似） - RSSKB uint64 // 物理内存 (KB) - FDCount int // 打开的文件句柄数 - Threads int // 线程数 +// readMemTotalKB reads MemTotal from /proc/meminfo +func readMemTotalKB() uint64 { + f, err := os.Open("/proc/meminfo") + if err != nil { + return 1 + } + defer f.Close() + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.HasPrefix(sc.Text(), "MemTotal:") { + f := strings.Fields(sc.Text()) + if len(f) >= 2 { + v, _ := strconv.ParseUint(f[1], 10, 64) + return v + } + } + } + return 1 } -// Collect gathers process metrics and returns top N sorted by RSS. -// Sorting by RSS is the most stable and reliable metric without multi-sampling CPU. +// readNumCPU reads CPU count from /proc/stat (cpuN lines) +func readNumCPU() int { + f, err := os.Open("/proc/stat") + if err != nil { + return 1 + } + defer f.Close() + n := 0 + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.HasPrefix(sc.Text(), "cpu") && len(sc.Text()) > 3 && sc.Text()[3] >= '0' && sc.Text()[3] <= '9' { + n++ + } + } + return n +} + +// readProcCPU reads utime+stime from /proc//stat +func readProcCPUJiffies(pid string) uint64 { + data, err := os.ReadFile(filepath.Join("/proc", pid, "stat")) + if err != nil { + return 0 + } + // stat: pid (comm) state ppid ... utime stime ... + // We need fields 14 (utime) and 15 (stime) in 1-based indexing + // Parse: skip past the closing ')' of comm + s := string(data) + idx := strings.LastIndex(s, ")") + if idx < 0 { + return 0 + } + fields := strings.Fields(s[idx+2:]) + if len(fields) < 13 { + return 0 + } + // After ')', fields are: state ppid pgrp session tty_nr tpgid flags minflt cminflt majflt cmajflt utime stime + // So utime = fields[11], stime = fields[12] + utime, _ := strconv.ParseUint(fields[11], 10, 64) + stime, _ := strconv.ParseUint(fields[12], 10, 64) + return utime + stime +} + +// readProcUser reads the owner username from /proc//status (Uid line) +func readProcUser(pid string) string { + statusFile := filepath.Join("/proc", pid, "status") + f, err := os.Open(statusFile) + if err != nil { + return "?" + } + defer f.Close() + sc := bufio.NewScanner(f) + var uidStr string + for sc.Scan() { + line := sc.Text() + if strings.HasPrefix(line, "Uid:") { + fields := strings.Fields(line) + if len(fields) >= 2 { + uidStr = fields[1] + } + break + } + } + if uidStr == "" || uidStr == "0" { + return "root" + } + // Try to look up username + u, err := user.LookupId(uidStr) + if err == nil { + return u.Username + } + return uidStr +} + +// readProcCmdline reads /proc//cmdline, replaces null bytes with spaces +func readProcCmdline(pid string) string { + data, err := os.ReadFile(filepath.Join("/proc", pid, "cmdline")) + if err != nil || len(data) == 0 { + // Fallback to comm + data2, err2 := os.ReadFile(filepath.Join("/proc", pid, "comm")) + if err2 == nil { + return strings.TrimSpace(string(data2)) + } + return "?" + } + // Replace null bytes with spaces + out := strings.Builder{} + for _, b := range data { + if b == 0 { + out.WriteByte(' ') + } else { + out.WriteByte(b) + } + } + return strings.TrimSpace(out.String()) +} + +// Collect gathers process metrics and returns top N sorted by RSS func (p *ProcessTopCollector) Collect(topN int) ([]model.ProcessStat, error) { + uptimeJiffies := readUptimeJiffies() + if uptimeJiffies < 1 { + uptimeJiffies = 1 + } + entries, err := os.ReadDir("/proc") if err != nil { return nil, err } - var samples []procSample + type sample struct { + pid int + name string + cmdline string + user string + cpuPct float64 + memPct float64 + rssKB uint64 + fdCount int + threads int + } + var samples []sample + for _, entry := range entries { if !entry.IsDir() { continue @@ -51,32 +185,56 @@ func (p *ProcessTopCollector) Collect(topN int) ([]model.ProcessStat, error) { if err != nil { continue } + pidStr := entry.Name() - name := readProcName(entry.Name()) - // Skip kernel threads (empty cmdline + running in kernel) - if name == "" || strings.HasPrefix(name, "[") { + // Read comm to filter kernel threads + commData, err := os.ReadFile(filepath.Join("/proc", pidStr, "comm")) + if err != nil { continue } + name := strings.TrimSpace(string(commData)) + if name == "" || strings.HasPrefix(name, "[") { + continue // skip kernel threads + } - rssKB := readProcRSS(entry.Name()) - fdCount := countFD(entry.Name()) - threads := readProcThreads(entry.Name()) + // Gather all metrics + jiffies := readProcCPUJiffies(pidStr) + rssKB := readProcRSS(pidStr) + fdCount := countFD(pidStr) + threads := readProcThreads(pidStr) + userName := readProcUser(pidStr) + cmdline := readProcCmdline(pidStr) - samples = append(samples, procSample{ - PID: pid, - Name: name, - RSSKB: rssKB, - FDCount: fdCount, - Threads: threads, + // CPU% = (utime+stime) / uptime_jiffies * 100 / num_cpus + cpuPct := float64(jiffies) / float64(uptimeJiffies) * 100.0 / float64(p.numCPU) + if cpuPct > 100*float64(p.numCPU) { + cpuPct = 0 // invalid + } + + // Memory% + var memPct float64 + if p.memTotalKB > 0 { + memPct = float64(rssKB) / float64(p.memTotalKB) * 100.0 + } + + samples = append(samples, sample{ + pid: pid, + name: name, + cmdline: cmdline, + user: userName, + cpuPct: cpuPct, + memPct: memPct, + rssKB: rssKB, + fdCount: fdCount, + threads: threads, }) } // Sort by RSS descending sort.Slice(samples, func(i, j int) bool { - return samples[i].RSSKB > samples[j].RSSKB + return samples[i].rssKB > samples[j].rssKB }) - // Take top N if topN > len(samples) { topN = len(samples) } @@ -84,23 +242,41 @@ func (p *ProcessTopCollector) Collect(topN int) ([]model.ProcessStat, error) { for i := 0; i < topN; i++ { s := samples[i] result[i] = model.ProcessStat{ - PID: s.PID, - Name: s.Name, - MemoryMB: float64(s.RSSKB) / 1024.0, - FDCount: s.FDCount, - Threads: s.Threads, + PID: s.pid, + Name: s.name, + Cmdline: s.cmdline, + User: s.user, + CPUPct: s.cpuPct, + MemPct: s.memPct, + MemoryMB: float64(s.rssKB) / 1024.0, + FDCount: s.fdCount, + Threads: s.threads, } } return result, nil } -// readProcName reads /proc//comm -func readProcName(pid string) string { - data, err := os.ReadFile(filepath.Join("/proc", pid, "comm")) +// readUptimeJiffies reads /proc/uptime and converts to jiffies +func readUptimeJiffies() uint64 { + data, err := os.ReadFile("/proc/uptime") if err != nil { - return "" + return 1 } - return strings.TrimSpace(string(data)) + f := strings.Fields(string(data)) + if len(f) < 1 { + return 1 + } + sec, _ := strconv.ParseFloat(f[0], 64) + return uint64(sec * clktck) +} + +// countFD counts open file descriptors for a process +func countFD(pid string) int { + entries, err := os.ReadDir(filepath.Join("/proc", pid, "fd")) + if err != nil { + return 0 + } + return len(entries) } // readProcRSS reads VmRSS from /proc//status @@ -110,15 +286,13 @@ func readProcRSS(pid string) uint64 { return 0 } defer f.Close() - - scanner := bufio.NewScanner(f) - for scanner.Scan() { - line := scanner.Text() - if strings.HasPrefix(line, "VmRSS:") { - fields := strings.Fields(line) - if len(fields) >= 2 { - val, _ := strconv.ParseUint(fields[1], 10, 64) - return val + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.HasPrefix(sc.Text(), "VmRSS:") { + f := strings.Fields(sc.Text()) + if len(f) >= 2 { + v, _ := strconv.ParseUint(f[1], 10, 64) + return v } } } @@ -132,26 +306,15 @@ func readProcThreads(pid string) int { return 0 } defer f.Close() - - scanner := bufio.NewScanner(f) - for scanner.Scan() { - line := scanner.Text() - if strings.HasPrefix(line, "Threads:") { - fields := strings.Fields(line) - if len(fields) >= 2 { - val, _ := strconv.Atoi(fields[1]) - return val + sc := bufio.NewScanner(f) + for sc.Scan() { + if strings.HasPrefix(sc.Text(), "Threads:") { + f := strings.Fields(sc.Text()) + if len(f) >= 2 { + v, _ := strconv.Atoi(f[1]) + return v } } } return 0 } - -// countFD counts open file descriptors for a process -func countFD(pid string) int { - entries, err := os.ReadDir(filepath.Join("/proc", pid, "fd")) - if err != nil { - return 0 - } - return len(entries) -} diff --git a/agent/collector/util.go b/agent/collector/util.go new file mode 100644 index 0000000..04c62dc --- /dev/null +++ b/agent/collector/util.go @@ -0,0 +1,25 @@ +// Package collector 的通用工具函数。 +// +// 署名 Logan.Li +package collector + +// SplitLines 按换行符切割字符串 +func SplitLines(s string) []string { + var result []string + start := 0 + for i := 0; i < len(s); i++ { + if s[i] == '\n' { + result = append(result, s[start:i]) + start = i + 1 + } + } + if start < len(s) { + result = append(result, s[start:]) + } + return result +} + +// HasPrefix 检查字符串前缀 +func HasPrefix(s, prefix string) bool { + return len(s) >= len(prefix) && s[:len(prefix)] == prefix +} diff --git a/agent/main.go b/agent/main.go index 3396b0e..55445cf 100644 --- a/agent/main.go +++ b/agent/main.go @@ -1,6 +1,7 @@ // Package main 是 AIOps Agent 的入口。 // Agent 定时采集 CPU、内存、磁盘、端口、进程 TOP、Swap、FD、上下文切换等指标， // 并通过 HTTP 上报至 AIOps 后端。 +// v2.1: 共享 HTTP client + 独立探针拉取 + 配置结构体 // // 优化要点（1000 台服务器规模）： // - 启动随机 Jitter 避免惊群效应 @@ -15,7 +16,6 @@ package main import ( "context" "flag" - "fmt" "log" "math/rand" "net" @@ -25,6 +25,7 @@ import ( "syscall" "time" + agentclient "ezops/agent/client" "ezops/agent/collector" "ezops/agent/model" "ezops/agent/reporter" @@ -32,38 +33,51 @@ import ( ) var ( - version = "v2.0.0" + 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() { - server := flag.String("server", "http://localhost:8080", "AIOps 后端上报地址") - interval := flag.Int("interval", 10, "采集与上报间隔，单位：秒") - token := flag.String("token", "", "Bearer 鉴权令牌") - hostnameFlag := flag.String("hostname", "", "上报主机名，默认使用系统主机名") - topN := flag.Int("top-n", 3, "进程 TOP N 数量") + 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) - hostname := *hostnameFlag - if hostname == "" { + if cfg.Hostname == "" { if h, err := os.Hostname(); err == nil { - hostname = h + cfg.Hostname = h } else { - hostname = "unknown" + cfg.Hostname = "unknown" } } - if *server == "" { + if cfg.Server == "" { logger.Fatal("参数 -server 不能为空") } - if *interval < 5 { - logger.Fatalf("参数 -interval 必须 >= 5，当前为 %d（1000 台规模推荐 30~60s）", *interval) + 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", *server, *interval, *topN, hostname) + 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() @@ -74,7 +88,10 @@ func main() { diskIOCollector := collector.NewDiskIOCollector() procTopCollector := collector.NewProcessTopCollector() swapCollector := collector.NewSwapCollector() - rep := reporter.New(*server, *token, logger) + healthProbeCollector := collector.NewHealthProbeCollector() + fdCollector := collector.NewFDCollector() + csCollector := collector.NewContextSwitchCollector() + rep := reporter.New(httpCli, logger) // 系统静态信息（仅启动时一次） sysInfo := collector.CollectSystem() @@ -84,111 +101,97 @@ func main() { localIP := detectLocalIP() logger.Printf("探测本机 IP: %s", localIP) - sysReportCount := 0 // 前 3 次上报携带系统信息 + sysReportCount := 0 - // 任务执行器 + // 任务执行器 + 任务客户端 taskExecutor := task.New(logger) - taskClient := task.NewClient(*server, *token, hostname, logger) + taskClient := task.NewClient(httpCli, cfg.Hostname, logger) - // ═══ 启动随机 Jitter：避免 1000 台同时上报 ═══ - startupJitter := time.Duration(rand.Intn(*interval)) * time.Second + // ═══ 启动随机 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() - // Agent 自检序列号 var reportSeq uint64 + var probeConfigs []model.ProbeConfig - // collectAndReport 执行单次采集并上报 + // ═══ collectAndReport 采集 + 上报 ═══ collectAndReport := func() { report := &model.Report{ - Hostname: hostname, + Hostname: cfg.Hostname, IP: localIP, Timestamp: time.Now().UTC().Format(time.RFC3339), AgentVersion: version, } - // 系统信息（前 3 次） if sysReportCount < 3 && (sysInfo.OSVersion != "" || sysInfo.Kernel != "") { report.System = &model.SystemStat{ - OSVersion: sysInfo.OSVersion, - Kernel: sysInfo.Kernel, - Uptime: sysInfo.UptimeSeconds, + OSVersion: sysInfo.OSVersion, + Kernel: sysInfo.Kernel, + Uptime: sysInfo.UptimeSeconds, } sysReportCount++ } - // CPU + // 系统指标采集 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 } - - // Swap（仅当系统配置了 swap 才上报） 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 } - - // 磁盘 IO if dioStat, err := diskIOCollector.Collect(); err != nil { logger.Printf("磁盘IO采集异常: %v", err) } else { report.DiskIO = dioStat } - - // 进程 TOP N - if procs, err := procTopCollector.Collect(*topN); err != nil { + if procs, err := procTopCollector.Collect(cfg.TopN); err != nil { logger.Printf("进程TOP采集异常: %v", err) } else { report.ProcessTop = procs } - - // 系统 FD 句柄 - if fdStat, err := collectFD(); err != nil { + if fdStat, err := fdCollector.Collect(); err != nil { logger.Printf("FD采集异常: %v", err) } else { report.FD = fdStat } - - // 上下文切换 - if csStat, err := collectContextSwitches(); err != nil { - // 静默跳过（某些内核/容器环境不可用） + 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 @@ -196,17 +199,19 @@ func main() { report.Heartbeat = model.AgentHeartbeat{ Version: version, UptimeSec: sysInfo.UptimeSeconds, - SelfCPU: 0, // TODO: self-CPU via /proc/self/stat + 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 } - logger.Printf("上报成功: cpu=%.1f%% mem=%.1f%% swap=%.1f%% disks=%d ports=%d procs=%d fd=%d/%d cs=%.0f/s", + + 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 { @@ -215,9 +220,7 @@ func main() { } return 0 }(), - len(report.Disks), - len(report.Ports), - len(report.ProcessTop), + len(report.Disks), len(report.Ports), len(report.ProcessTop), func() int { if report.FD != nil { return report.FD.Allocated @@ -236,10 +239,11 @@ func main() { } return 0 }(), + probeCount, ) } - // pollAndExecute 任务轮询 + // ═══ pollAndExecute 任务轮询 ═══ pollAndExecute := func() { tasks, err := taskClient.Poll(ctx) if err != nil { @@ -249,46 +253,59 @@ func main() { logger.Printf("任务轮询失败: %v", err) return } - if len(tasks) == 0 { - return - } - 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 { + if len(tasks) > 0 { + logger.Printf("拉取到 %d 个待执行任务", len(tasks)) + for _, t := range tasks { if ctx.Err() != nil { return } - logger.Printf("任务 %s 结果回传失败: %v", t.TaskID, err) - } else { - logger.Printf("任务 %s 结果回传成功 status=%s", t.TaskID, result.Status) + 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) + } } } } - // ═══ 任务轮询协程（5s 间隔，不受 -interval 影响）═══ - const taskPollInterval = 5 * time.Second + // ═══ 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() - ticker := time.NewTicker(taskPollInterval) + pullProbes() + ticker := time.NewTicker(pollInterval) defer ticker.Stop() for { select { case <-ticker.C: pollAndExecute() + pullProbes() case <-ctx.Done(): return } } }() - // ═══ 主循环：启动先 sleep jitter，再按 interval 采集 ═══ + // ═══ 主循环：采集 + 上报 ═══ time.Sleep(startupJitter) collectAndReport() - ticker := time.NewTicker(time.Duration(*interval) * time.Second) + ticker := time.NewTicker(time.Duration(cfg.Interval) * time.Second) defer ticker.Stop() for { select { @@ -301,17 +318,6 @@ func main() { } } -// maskToken 对令牌做脱敏处理 -func maskToken(token string) string { - if token == "" { - return "(empty)" - } - if len(token) <= 4 { - return "****" - } - return token[:4] + "****" -} - // detectLocalIP 探测本机主 IP func detectLocalIP() string { conn, err := net.Dial("udp", "8.8.8.8:53") @@ -337,73 +343,3 @@ func detectLocalIP() string { } return "unknown" } - -// ══════════════════════════════════════════════ -// 补充采集器（内联简单实现，不单独建文件） -// ══════════════════════════════════════════════ - -// collectFD 读取 /proc/sys/fs/file-nr 和 /proc/sys/fs/file-max -func collectFD() (*model.SocketFDStat, error) { - alloc, err := os.ReadFile("/proc/sys/fs/file-nr") - if err != nil { - return nil, err - } - var allocated, _unused1, _unused2 int - fmt.Sscanf(string(alloc), "%d %d %d", &allocated, &_unused1, &_unused2) - - max, err := os.ReadFile("/proc/sys/fs/file-max") - if err != nil { - return nil, err - } - var maxFD int64 - fmt.Sscanf(string(max), "%d", &maxFD) - - stat := &model.SocketFDStat{ - Allocated: allocated, - Max: maxFD, - } - if maxFD > 0 { - stat.UsagePct = float64(allocated) / float64(maxFD) * 100.0 - } - return stat, nil -} - -// collectContextSwitches 读取 /proc/stat 中的 ctxt 和 intr 行 -// 需要两次采样才能计算速率；这里使用简单的累计值（首次返回 0） -func collectContextSwitches() (*model.ContextSwitchStat, error) { - data, err := os.ReadFile("/proc/stat") - if err != nil { - return nil, err - } - stat := &model.ContextSwitchStat{} - lines := string(data) - for _, line := range splitLines(lines) { - if hasPrefix(line, "ctxt ") { - fmt.Sscanf(line, "ctxt %f", &stat.CTXPerSec) - } - if hasPrefix(line, "intr ") { - fmt.Sscanf(line, "intr %f", &stat.IRQPerSec) - } - } - // NOTE: 这里返回的是累计值而非速率；后端可用两次上报的差值计算速率 - return stat, nil -} - -func splitLines(s string) []string { - var result []string - start := 0 - for i := 0; i < len(s); i++ { - if s[i] == '\n' { - result = append(result, s[start:i]) - start = i + 1 - } - } - if start < len(s) { - result = append(result, s[start:]) - } - return result -} - -func hasPrefix(s, prefix string) bool { - return len(s) >= len(prefix) && s[:len(prefix)] == prefix -} diff --git a/agent/model/model.go b/agent/model/model.go index ef9bb53..3cea64e 100644 --- a/agent/model/model.go +++ b/agent/model/model.go @@ -22,6 +22,7 @@ type Report struct { FD *SocketFDStat `json:"fd,omitempty"` // 系统文件句柄 ContextSwitches *ContextSwitchStat `json:"context_switches,omitempty"` // 上下文切换 Heartbeat AgentHeartbeat `json:"heartbeat"` // Agent 自检 + ProbeResults []ProbeResult `json:"probe_results,omitempty"` // v2.1 健康探针结果 } // CPUStat CPU 相关指标 diff --git a/agent/model/model_ext.go b/agent/model/model_ext.go index 60b8fd6..0a4af98 100644 --- a/agent/model/model_ext.go +++ b/agent/model/model_ext.go @@ -56,3 +56,31 @@ type SystemStat struct { UlimitNofile int64 `json:"ulimit_nofile"` // ulimit -n (最大文件描述符) UlimitNproc int64 `json:"ulimit_nproc"` // ulimit -u (最大进程数) } + +// ──── v2.1 健康探针 ──── + +// ProbeConfig 后端下发的探针配置 +type ProbeConfig struct { + ID uint `json:"id"` + Name string `json:"name"` + ProbeType string `json:"probe_type"` // http_get / tcp_socket / exec_command + HTTPURL string `json:"http_url,omitempty"` + HTTPMethod string `json:"http_method,omitempty"` + HTTPExpectCode int `json:"http_expect_code,omitempty"` + HTTPExpectBody string `json:"http_expect_body,omitempty"` + TCPHost string `json:"tcp_host,omitempty"` + TCPPort int `json:"tcp_port,omitempty"` + ExecCommand string `json:"exec_command,omitempty"` + Timeout int `json:"timeout"` +} + +// ProbeResult 探针执行结果 +type ProbeResult struct { + ProbeID uint `json:"probe_id"` + Name string `json:"name"` + ProbeType string `json:"probe_type"` + Status string `json:"status"` // up / down + LatencyMs int `json:"latency_ms"` + StatusCode int `json:"status_code,omitempty"` + Message string `json:"message,omitempty"` +} diff --git a/agent/model/model_proctop.go b/agent/model/model_proctop.go index d258ff4..1fa18ce 100644 --- a/agent/model/model_proctop.go +++ b/agent/model/model_proctop.go @@ -1,10 +1,14 @@ package model -// ProcessStat 单个进程统计（TOP N 采集） +// ProcessStat 单个进程统计（TOP N 采集）— 阿里云风格 type ProcessStat struct { - PID int `json:"pid"` - Name string `json:"name"` - MemoryMB float64 `json:"memory_mb"` // 物理内存 (MB) - FDCount int `json:"fd_count"` // 打开的文件句柄数 - Threads int `json:"threads"` // 线程数 + PID int `json:"pid"` + Name string `json:"name"` // 进程名（comm） + Cmdline string `json:"cmdline"` // 完整命令行 + User string `json:"user"` // 运行用户 + CPUPct float64 `json:"cpu_pct"` // CPU 使用率 (%) + MemPct float64 `json:"mem_pct"` // 内存使用率 (%) + MemoryMB float64 `json:"memory_mb"` // 物理内存 (MB) + FDCount int `json:"fd_count"` // 打开的文件句柄数 + Threads int `json:"threads"` // 线程数 } diff --git a/agent/reporter/http.go b/agent/reporter/http.go index b46a040..38c0eeb 100644 --- a/agent/reporter/http.go +++ b/agent/reporter/http.go @@ -1,122 +1,38 @@ // Package reporter 实现 AIOps Agent 向后端上报采集结果的能力。 -// 通过 HTTP POST 将指标 JSON 发送至 {server}/api/v1/agent/report， -// 并在请求头携带 Bearer Token 完成鉴权。失败时进行有限次指数退避重试。 +// 通过 HTTP POST 将指标 JSON 发送至 {server}/api/v1/agent/report。 +// 共享 client 处理鉴权 + 重试，Reporter 仅负责序列化和路径选择。 // // 署名 Logan.Li package reporter import ( - "bytes" "context" - "encoding/json" "fmt" - "io" "log" - "net/http" - "strings" - "time" + "ezops/agent/client" "ezops/agent/model" ) -const ( - reportPath = "/api/v1/agent/report" // 后端上报接口路径 - defaultTimeout = 10 * time.Second // 单次 HTTP 请求超时 - maxRetries = 3 // 最大重试次数（含首次） - userAgent = "aiops-agent/1.0" // 标识 Agent 版本，便于后端识别 -) +const reportPath = "/api/v1/agent/report" // Reporter 负责将采集报告上报至 AIOps 后端 type Reporter struct { - server string - token string - httpClient *http.Client - logger *log.Logger + client *client.Client + logger *log.Logger } // New 创建 Reporter 实例。 -// - server: AIOps 后端地址（如 http://localhost:8080） -// - token: Bearer 认证令牌 -func New(server, token string, logger *log.Logger) *Reporter { - return &Reporter{ - server: normalizeServer(server), - token: token, - httpClient: &http.Client{Timeout: defaultTimeout}, - logger: logger, - } +// - cli: 共享 HTTP 客户端 +// - logger: 日志输出 +func New(cli *client.Client, logger *log.Logger) *Reporter { + return &Reporter{client: cli, logger: logger} } -// normalizeServer 去除尾部斜杠，保证 URL 拼接正确 -func normalizeServer(server string) string { - return strings.TrimRight(server, "/") -} - -// Send 将报告序列化为 JSON 并 POST 到后端，失败时进行有限次重试。 -// 重试过程可被传入的上下文取消打断（如 Agent 收到退出信号时）。 +// Send 将报告 POST 到后端。纯上报，不解析响应。 func (r *Reporter) Send(ctx context.Context, report *model.Report) error { - payload, err := json.Marshal(report) - if err != nil { - return fmt.Errorf("序列化上报数据失败: %w", err) - } - - url := r.server + reportPath - - var lastErr error - for attempt := 1; attempt <= maxRetries; attempt++ { - // 每次重试都重建请求，避免请求体被消费后无法复用 - req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload)) - if err != nil { - return fmt.Errorf("构造请求失败: %w", err) - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", "Bearer "+r.token) - req.Header.Set("User-Agent", userAgent) - - // 上下文已取消则立即返回，避免无意义的网络请求 - if err := ctx.Err(); err != nil { - return err - } - - resp, err := r.httpClient.Do(req) - if err != nil { - lastErr = err - r.logger.Printf("上报请求失败（第%d次）: %v", attempt, err) - if !r.sleepWithCtx(ctx, attempt) { - return ctx.Err() - } - continue - } - - // 读取并丢弃响应体以便复用连接 - io.Copy(io.Discard, resp.Body) - resp.Body.Close() - - if resp.StatusCode >= 200 && resp.StatusCode < 300 { - return nil // 上报成功 - } - - lastErr = fmt.Errorf("上报失败，后端返回 HTTP %d", resp.StatusCode) - // 4xx（除 429 外）通常为客户端错误（如鉴权失败），重试无意义 - if resp.StatusCode >= 400 && resp.StatusCode < 500 && resp.StatusCode != http.StatusTooManyRequests { - return lastErr - } - r.logger.Printf("上报响应异常（第%d次）: HTTP %d", attempt, resp.StatusCode) - if !r.sleepWithCtx(ctx, attempt) { - return ctx.Err() - } - } - return lastErr -} - -// sleepWithCtx 带线性退避的休眠，可被上下文取消打断。 -// 返回 true 表示休眠结束可继续重试，false 表示被取消。 -func (r *Reporter) sleepWithCtx(ctx context.Context, attempt int) bool { - timer := time.NewTimer(time.Duration(attempt) * time.Second) - defer timer.Stop() - select { - case <-timer.C: - return true - case <-ctx.Done(): - return false + if err := r.client.DoJSON(ctx, "POST", reportPath, report, nil); err != nil { + return fmt.Errorf("上报失败: %w", err) } + return nil } diff --git a/agent/task/client.go b/agent/task/client.go index d3e3460..370e0f7 100644 --- a/agent/task/client.go +++ b/agent/task/client.go @@ -1,149 +1,76 @@ -// Package task 的 client.go 实现与后端任务接口的 HTTP 交互。 -// 涵盖拉取待执行任务（poll）与回传执行结果（submit result）。 -// 仅依赖 Go 标准库，失败时返回错误，由调用方决定重试策略。 +// Package task 的 client.go 实现与后端任务/探针接口的 HTTP 交互。 +// 涵盖拉取待执行任务（poll）、回传执行结果（submit result）、拉取探针配置（poll probes）。 +// 通过共享 client 处理鉴权 + 重试，本文件仅负责业务路径和数据模型。 // // 署名 Logan.Li package task import ( - "bytes" "context" - "encoding/json" "fmt" - "io" "log" - "net/http" "net/url" - "strings" - "time" + + "ezops/agent/client" + "ezops/agent/model" ) const ( - pollPath = "/api/v1/agent/tasks/poll" // 任务轮询接口 - resultPathFmt = "/api/v1/agent/tasks/%s/result" // 结果回传接口（含 task_id 占位） - httpTimeout = 10 * time.Second // 单次 HTTP 请求超时 - clientUserAgent = "aiops-agent/1.0" // 标识 Agent 版本 + pollPath = "/api/v1/agent/tasks/poll" + resultPathFmt = "/api/v1/agent/tasks/%s/result" + probePath = "/api/v1/agent/health-probes/config" ) -// Client 负责与后端任务接口通信。 +// Client 负责与后端任务/探针接口通信。 type Client struct { - server string - token string - hostname string - httpClient *http.Client - logger *log.Logger + client *client.Client + hostname string + logger *log.Logger } // NewClient 创建任务客户端实例。 -// - server: AIOps 后端地址（如 http://localhost:8080） -// - token: Bearer 认证令牌 -// - hostname: 本机主机名，用于后端筛选该主机待执行的任务 -func NewClient(server, token, hostname string, logger *log.Logger) *Client { - return &Client{ - server: strings.TrimRight(server, "/"), - token: token, - hostname: hostname, - httpClient: &http.Client{Timeout: httpTimeout}, - logger: logger, - } +func NewClient(cli *client.Client, hostname string, logger *log.Logger) *Client { + return &Client{client: cli, hostname: hostname, logger: logger} } // Poll 拉取当前主机待执行的任务列表。 -// 返回空切片表示暂无任务（后端返回空数组或 HTTP 204）。 func (c *Client) Poll(ctx context.Context) ([]Task, error) { - // 构造带 hostname 查询参数的 URL - u := c.server + pollPath + "?hostname=" + url.QueryEscape(c.hostname) - - req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) - if err != nil { - return nil, fmt.Errorf("构造轮询请求失败: %w", err) - } - c.attachHeaders(req) - - // 上下文已取消则直接返回，避免无意义请求 - if err := ctx.Err(); err != nil { - return nil, err - } - - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, fmt.Errorf("轮询请求失败: %w", err) - } - defer resp.Body.Close() - - // 204 No Content 或 200 空体均视为无任务 - if resp.StatusCode == http.StatusNoContent { - return nil, nil - } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) - return nil, fmt.Errorf("轮询响应异常: HTTP %d, body=%s", resp.StatusCode, string(body)) - } - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("读取轮询响应失败: %w", err) - } - - // 空响应体视为无任务 - if len(bytes.TrimSpace(body)) == 0 { - return nil, nil - } + path := pollPath + "?hostname=" + url.QueryEscape(c.hostname) var wrapper struct { Code int `json:"code"` Message string `json:"message"` Data []Task `json:"data"` } - if err := json.Unmarshal(body, &wrapper); err != nil { - // 兼容裸数组格式 - var tasks []Task - if err2 := json.Unmarshal(body, &tasks); err2 != nil { - return nil, fmt.Errorf("解析轮询响应失败: %w", err) - } - return tasks, nil + if err := c.client.DoJSON(ctx, "GET", path, nil, &wrapper); err != nil { + return nil, fmt.Errorf("任务轮询失败: %w", err) } return wrapper.Data, nil } // Submit 将任务执行结果回传至后端。 func (c *Client) Submit(ctx context.Context, res Result) error { - payload, err := json.Marshal(res) - if err != nil { - return fmt.Errorf("序列化任务结果失败: %w", err) - } - - u := c.server + fmt.Sprintf(resultPathFmt, url.PathEscape(res.TaskID)) - - req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(payload)) - if err != nil { - return fmt.Errorf("构造回传请求失败: %w", err) - } - c.attachHeaders(req) - - if err := ctx.Err(); err != nil { - return err - } - - resp, err := c.httpClient.Do(req) - if err != nil { - return fmt.Errorf("回传请求失败: %w", err) - } - defer resp.Body.Close() - io.Copy(io.Discard, resp.Body) - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("回传响应异常: HTTP %d", resp.StatusCode) + path := fmt.Sprintf(resultPathFmt, url.PathEscape(res.TaskID)) + if err := c.client.DoJSON(ctx, "POST", path, res, nil); err != nil { + return fmt.Errorf("任务结果回传失败: %w", err) } return nil } -// attachHeaders 统一设置公共请求头（鉴权、内容类型、UA）。 -func (c *Client) attachHeaders(req *http.Request) { - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Accept", "application/json") - req.Header.Set("User-Agent", clientUserAgent) - if c.token != "" { - req.Header.Set("Authorization", "Bearer "+c.token) +// PollProbes 拉取本机健康探针配置（v2.1）。 +func (c *Client) PollProbes(ctx context.Context) ([]model.ProbeConfig, error) { + path := probePath + "?hostname=" + url.QueryEscape(c.hostname) + + var wrapper struct { + Code int `json:"code"` + Message string `json:"message"` + Data []model.ProbeConfig `json:"data"` } + if err := c.client.DoJSON(ctx, "GET", path, nil, &wrapper); err != nil { + return nil, fmt.Errorf("探针配置拉取失败: %w", err) + } + if wrapper.Data == nil { + return []model.ProbeConfig{}, nil + } + return wrapper.Data, nil } diff --git a/backend/internal/api/handlers/agent.go b/backend/internal/api/handlers/agent.go index d40b640..0abf4b3 100644 --- a/backend/internal/api/handlers/agent.go +++ b/backend/internal/api/handlers/agent.go @@ -227,8 +227,9 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { h.writeHotMetric(c, "metrics:load1:"+req.Hostname, req.CPU.Load1, timestamp, ttl, reportTime) } - // 保存成功 - c.JSON(http.StatusOK, models.Response{Code: 200, Message: "ok"}) + // 保存成功 — 响应中携带探针配置给 Agent + resp := h.makeProbeConfigResponse(req.Hostname, 200, "ok", req.ProbeResults) + c.JSON(http.StatusOK, resp) } // writeHotMetric 将单个指标写入 Redis 有序集合并维护 TTL 和过期数据清理 diff --git a/backend/internal/api/handlers/alert_contacts.go b/backend/internal/api/handlers/alert_contacts.go index cd63c7b..f0349cd 100644 --- a/backend/internal/api/handlers/alert_contacts.go +++ b/backend/internal/api/handlers/alert_contacts.go @@ -33,12 +33,11 @@ func (h *AlertHandler) CreateContact(c *gin.Context) { return } contact := models.AlertContact{ - Name: req.Name, - Phone: req.Phone, - Email: req.Email, - Dingtalk: req.Dingtalk, - Feishu: req.Feishu, - Wechat: req.Wechat, + Name: req.Name, + Phone: req.Phone, + Email: req.Email, + ChannelType: req.ChannelType, + WebhookURL: req.WebhookURL, } if err := h.db.Create(&contact).Error; err != nil { c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建失败"}) @@ -64,9 +63,8 @@ func (h *AlertHandler) UpdateContact(c *gin.Context) { if req.Name != "" { updates["name"] = req.Name } if req.Phone != "" { updates["phone"] = req.Phone } if req.Email != "" { updates["email"] = req.Email } - if req.Dingtalk != "" { updates["dingtalk"] = req.Dingtalk } - if req.Feishu != "" { updates["feishu"] = req.Feishu } - if req.Wechat != "" { updates["wechat"] = req.Wechat } + if req.ChannelType != "" { updates["channel_type"] = req.ChannelType } + if req.WebhookURL != "" { updates["webhook_url"] = req.WebhookURL } h.db.Model(&contact).Updates(updates) h.db.First(&contact, id) c.JSON(http.StatusOK, models.Response{Code: 200, Message: "更新成功", Data: contact}) diff --git a/backend/internal/api/handlers/health_probe.go b/backend/internal/api/handlers/health_probe.go new file mode 100644 index 0000000..d4a8d5e --- /dev/null +++ b/backend/internal/api/handlers/health_probe.go @@ -0,0 +1,308 @@ +package handlers + +import ( + "net/http" + "strconv" + "time" + + "ezops/internal/models" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" +) + +// ──────────────────────────────────────── +// HealthProbe CRUD（健康探针配置管理） +// ──────────────────────────────────────── + +// ListHealthProbes GET /api/v1/agent/health-probes +func (h *AgentHandler) ListHealthProbes(c *gin.Context) { + query := h.db.Model(&models.HealthProbe{}) + + if hostname := c.Query("hostname"); hostname != "" { + query = query.Where("hostname = ?", hostname) + } + if enabled := c.Query("enabled"); enabled != "" { + if enabled == "true" { + query = query.Where("enabled = ?", true) + } else if enabled == "false" { + query = query.Where("enabled = ?", false) + } + } + + var probes []models.HealthProbe + query.Order("created_at DESC").Limit(500).Find(&probes) + + if probes == nil { + probes = []models.HealthProbe{} + } + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: "success", + Data: probes, + }) +} + +// CreateHealthProbe POST /api/v1/agent/health-probes +func (h *AgentHandler) CreateHealthProbe(c *gin.Context) { + var probe models.HealthProbe + if err := c.ShouldBindJSON(&probe); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求参数错误: " + err.Error()}) + return + } + if probe.Name == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "探针名称不能为空"}) + return + } + if probe.Hostname == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "目标主机不能为空"}) + return + } + if probe.ProbeType == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "探针类型不能为空"}) + return + } + // 验证探针类型配置 + switch probe.ProbeType { + case models.ProbeTypeHTTP: + if probe.HTTPURL == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "HTTP 探针必须填写 URL"}) + return + } + case models.ProbeTypeTCP: + if probe.TCPHost == "" || probe.TCPPort == 0 { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "TCP 探针必须填写 Host 和 Port"}) + return + } + case models.ProbeTypeExec: + if probe.ExecCommand == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "Exec 探针必须填写执行命令"}) + return + } + default: + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "不支持的探针类型: " + string(probe.ProbeType)}) + return + } + + probe.ID = 0 + probe.Enabled = true + if probe.Timeout == 0 { + probe.Timeout = 10 + } + if probe.Interval == 0 { + probe.Interval = 60 + } + + if err := h.db.Create(&probe).Error; err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "创建失败: " + err.Error()}) + return + } + + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "创建成功", Data: probe}) +} + +// UpdateHealthProbe PUT /api/v1/agent/health-probes/:id +func (h *AgentHandler) UpdateHealthProbe(c *gin.Context) { + id := c.Param("id") + var probe models.HealthProbe + if err := h.db.First(&probe, id).Error; err != nil { + c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "探针不存在"}) + return + } + + var input models.HealthProbe + if err := c.ShouldBindJSON(&input); err != nil { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()}) + return + } + + updates := map[string]interface{}{} + if input.Name != "" { + updates["name"] = input.Name + } + if input.Hostname != "" { + updates["hostname"] = input.Hostname + } + if input.ProbeType != "" { + updates["probe_type"] = input.ProbeType + } + if input.HTTPURL != "" { + updates["http_url"] = input.HTTPURL + } + if input.TCPHost != "" { + updates["tcp_host"] = input.TCPHost + } + if input.TCPPort != 0 { + updates["tcp_port"] = input.TCPPort + } + if input.ExecCommand != "" { + updates["exec_command"] = input.ExecCommand + } + if input.Timeout > 0 { + updates["timeout"] = input.Timeout + } + if input.Interval > 0 { + updates["interval"] = input.Interval + } + // enabled 布尔处理：显式传入 + updates["enabled"] = input.Enabled + if input.HTTPMethod != "" { + updates["http_method"] = input.HTTPMethod + } + if input.HTTPHeaders != "" { + updates["http_headers"] = input.HTTPHeaders + } + if input.HTTPExpectCode > 0 { + updates["http_expect_code"] = input.HTTPExpectCode + } + if input.HTTPExpectBody != "" { + updates["http_expect_body"] = input.HTTPExpectBody + } + if input.Description != "" { + updates["description"] = input.Description + } + + if err := h.db.Model(&probe).Updates(updates).Error; err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "更新失败: " + err.Error()}) + return + } + + h.db.First(&probe, id) + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "更新成功", Data: probe}) +} + +// DeleteHealthProbe DELETE /api/v1/agent/health-probes/:id +func (h *AgentHandler) DeleteHealthProbe(c *gin.Context) { + id := c.Param("id") + if err := h.db.Delete(&models.HealthProbe{}, id).Error; err != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "删除失败"}) + return + } + c.JSON(http.StatusOK, models.Response{Code: 200, Message: "删除成功"}) +} + +// GetHealthProbesForAgent Agent 拉取本机探针配置（Agent 调用，无需 JWT） +// GET /api/v1/agent/health-probes/config?hostname=xxx +func (h *AgentHandler) GetHealthProbesForAgent(c *gin.Context) { + hostname := c.Query("hostname") + if hostname == "" { + c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "hostname 不能为空"}) + return + } + + var probes []models.HealthProbe + h.db.Where("hostname = ? AND enabled = ?", hostname, true). + Order("id ASC"). + Find(&probes) + + if probes == nil { + probes = []models.HealthProbe{} + } + + c.JSON(http.StatusOK, models.Response{ + Code: 200, + Message: "success", + Data: probes, + }) +} + +// ──────────────────────────────────────── +// ReceiveReport 增强：处理探针结果 + 返回配置 +// ──────────────────────────────────────── + +// processProbeResults 处理 Agent 上报的探针结果，更新探针状态 +func (h *AgentHandler) processProbeResults(hostname string, results []models.HealthProbeResult) { + now := time.Now() + for _, r := range results { + updates := map[string]interface{}{ + "last_status": string(r.Status), + "last_checked_at": now, + "last_latency_ms": r.LatencyMs, + } + + if r.Status == models.ProbeStatusDown { + // 失败时增加连续失败计数 + h.db.Model(&models.HealthProbe{}). + Where("id = ? AND hostname = ?", r.ProbeID, hostname). + UpdateColumn("fail_count", gorm.Expr("fail_count + 1")) + } else { + // 成功时清零 + updates["fail_count"] = 0 + } + + h.db.Model(&models.HealthProbe{}). + Where("id = ? AND hostname = ?", r.ProbeID, hostname). + Updates(updates) + + // ═══ 探针告警：Agent 本地探测 down → 平台创建告警事件 ═══ + if r.Status == models.ProbeStatusDown { + // 检查是否存在未恢复的同探针告警 + var existingAlert models.AgentAlert + h.db.Where("rule_name = ? AND hostname = ? AND metric = ? AND resolved = ?", + r.Name, hostname, "health_probe", false). + First(&existingAlert) + + if existingAlert.ID == 0 { + // 获取 fail_count 判断严重级别 + var probe models.HealthProbe + if err := h.db.First(&probe, r.ProbeID).Error; err == nil { + severity := "warning" + if probe.FailCount >= 3 { + severity = "critical" + } + alert := models.AgentAlert{ + RuleName: r.Name, + Hostname: hostname, + Metric: "health_probe", + Severity: severity, + Message: "健康探针 [" + r.Name + "] 检测异常: " + r.Message + " (类型=" + r.ProbeType + ", 延迟=" + strconv.Itoa(r.LatencyMs) + "ms)", + Threshold: 0, + Resolved: false, + } + h.db.Create(&alert) + } + } + } else { + // 探针恢复正常：自动恢复告警 + h.db.Model(&models.AgentAlert{}). + Where("rule_name = ? AND hostname = ? AND metric = ? AND resolved = ?", + r.Name, hostname, "health_probe", false). + Updates(map[string]interface{}{ + "resolved": true, + "resolved_at": now, + }) + } + } +} + +// probeConfigResponse ReceiveReport 的响应结构（含探针配置） +type probeConfigResponse struct { + models.Response + ProbeConfigs []models.HealthProbe `json:"probe_configs,omitempty"` +} + +// makeProbeConfigResponse 构建带探针配置的响应 +func (h *AgentHandler) makeProbeConfigResponse(hostname string, respCode int, respMsg string, probeResults []models.HealthProbeResult) probeConfigResponse { + // 处理上报的探针结果 + if len(probeResults) > 0 { + h.processProbeResults(hostname, probeResults) + } + + // 拉取该主机的新探针配置（供 Agent 下次执行） + var probes []models.HealthProbe + h.db.Where("hostname = ? AND enabled = ?", hostname, true). + Order("id ASC"). + Find(&probes) + if probes == nil { + probes = []models.HealthProbe{} + } + + return probeConfigResponse{ + Response: models.Response{ + Code: respCode, + Message: respMsg, + }, + ProbeConfigs: probes, + } +} diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index 086a60c..d5379e2 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -674,6 +674,8 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c agentGroup.POST("/tasks/:task_id/result", taskHandler.SubmitResult) agentGroup.POST("/aggregate", agentHandler.AggregateMetrics) agentGroup.POST("/exec", agentInstallHandler.RemoteExec) + // 健康探针：Agent 拉取本机配置（无需 JWT） + agentGroup.GET("/health-probes/config", agentHandler.GetHealthProbesForAgent) } // Agent 数据查询 + 任务管理（需要 JWT） agentQuery := authorized.Group("/agent") @@ -702,6 +704,12 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c agentQuery.POST("/uninstall", agentInstallHandler.UninstallAgent) agentQuery.POST("/batch-uninstall", agentInstallHandler.BatchUninstallAgent) agentQuery.POST("/batch-upgrade", agentInstallHandler.BatchUpgradeAgent) + + // 健康探针管理（CRUD） + agentQuery.GET("/health-probes", agentHandler.ListHealthProbes) + agentQuery.POST("/health-probes", agentHandler.CreateHealthProbe) + agentQuery.PUT("/health-probes/:id", agentHandler.UpdateHealthProbe) + agentQuery.DELETE("/health-probes/:id", agentHandler.DeleteHealthProbe) } // Agent 二进制下载（无需鉴权，Agent 升级时使用） diff --git a/backend/internal/models/agent.go b/backend/internal/models/agent.go index d5805ef..0869d46 100644 --- a/backend/internal/models/agent.go +++ b/backend/internal/models/agent.go @@ -87,7 +87,8 @@ type AgentReportRequest struct { ProcessTop []ProcessMetric `json:"process_top,omitempty"` FD *FDMetric `json:"fd,omitempty"` ContextSwitches *CSMetric `json:"context_switches,omitempty"` - Heartbeat HeartbeatMetric `json:"heartbeat"` + Heartbeat HeartbeatMetric `json:"heartbeat"` + ProbeResults []HealthProbeResult `json:"probe_results,omitempty"` // v2.1 健康探针结果 } // ──── v2.0 新增上报类型 ──── @@ -101,6 +102,10 @@ type SwapMetric struct { type ProcessMetric struct { PID int `json:"pid"` Name string `json:"name"` + Cmdline string `json:"cmdline"` + User string `json:"user"` + CPUPct float64 `json:"cpu_pct"` + MemPct float64 `json:"mem_pct"` MemoryMB float64 `json:"memory_mb"` FDCount int `json:"fd_count"` Threads int `json:"threads"` diff --git a/backend/internal/models/alert.go b/backend/internal/models/alert.go index 4b0252f..f8501a4 100644 --- a/backend/internal/models/alert.go +++ b/backend/internal/models/alert.go @@ -400,16 +400,15 @@ type AcknowledgeAlertRequest struct { // ============================================ type AlertContact struct { - ID uint `gorm:"primaryKey;autoIncrement;comment:主键ID" json:"id"` - Name string `gorm:"type:varchar(128);not null;comment:联系人姓名" json:"name"` - Phone string `gorm:"type:varchar(32);comment:手机号" json:"phone,omitempty"` - Email string `gorm:"type:varchar(128);comment:邮箱" json:"email,omitempty"` - Dingtalk string `gorm:"type:varchar(256);comment:钉钉webhook地址" json:"dingtalk,omitempty"` - Feishu string `gorm:"type:varchar(256);comment:飞书webhook地址" json:"feishu,omitempty"` - Wechat string `gorm:"type:varchar(256);comment:企业微信webhook地址" json:"wechat,omitempty"` - CreatedAt time.Time `gorm:"not null;comment:创建时间" json:"created_at"` - UpdatedAt time.Time `gorm:"not null;comment:更新时间" json:"updated_at"` - DeletedAt gorm.DeletedAt `gorm:"index;comment:删除时间" json:"deleted_at,omitempty"` + ID uint `gorm:"primaryKey;autoIncrement;comment:主键ID" json:"id"` + Name string `gorm:"type:varchar(128);not null;comment:联系人姓名" json:"name"` + Phone string `gorm:"type:varchar(32);comment:手机号" json:"phone,omitempty"` + Email string `gorm:"type:varchar(128);comment:邮箱" json:"email,omitempty"` + ChannelType string `gorm:"type:varchar(32);default:'feishu';comment:渠道类型: feishu/dingtalk/wechat/webhook" json:"channel_type,omitempty"` + WebhookURL string `gorm:"type:varchar(512);comment:Webhook地址" json:"webhook_url,omitempty"` + CreatedAt time.Time `gorm:"not null;comment:创建时间" json:"created_at"` + UpdatedAt time.Time `gorm:"not null;comment:更新时间" json:"updated_at"` + DeletedAt gorm.DeletedAt `gorm:"index;comment:删除时间" json:"deleted_at,omitempty"` } func (AlertContact) TableName() string { @@ -436,22 +435,20 @@ func (AlertContactGroup) TableName() string { // CreateContactRequest 创建联系人请求 type CreateContactRequest struct { - Name string `json:"name" binding:"required"` - Phone string `json:"phone,omitempty"` - Email string `json:"email,omitempty"` - Dingtalk string `json:"dingtalk,omitempty"` - Feishu string `json:"feishu,omitempty"` - Wechat string `json:"wechat,omitempty"` + Name string `json:"name" binding:"required"` + Phone string `json:"phone,omitempty"` + Email string `json:"email,omitempty"` + ChannelType string `json:"channel_type" binding:"required,oneof=feishu dingtalk wechat webhook"` + WebhookURL string `json:"webhook_url" binding:"required"` } // UpdateContactRequest 更新联系人请求 type UpdateContactRequest struct { - Name string `json:"name,omitempty"` - Phone string `json:"phone,omitempty"` - Email string `json:"email,omitempty"` - Dingtalk string `json:"dingtalk,omitempty"` - Feishu string `json:"feishu,omitempty"` - Wechat string `json:"wechat,omitempty"` + Name string `json:"name,omitempty"` + Phone string `json:"phone,omitempty"` + Email string `json:"email,omitempty"` + ChannelType string `json:"channel_type,omitempty" binding:"omitempty,oneof=feishu dingtalk wechat webhook"` + WebhookURL string `json:"webhook_url,omitempty"` } // CreateContactGroupRequest 创建联系组请求 diff --git a/backend/internal/models/monitor.go b/backend/internal/models/monitor.go index a4cf1e6..269f683 100644 --- a/backend/internal/models/monitor.go +++ b/backend/internal/models/monitor.go @@ -353,3 +353,73 @@ type HostGroup struct { func (HostGroup) TableName() string { return "asset_host_groups" } + +// ============================================ +// HealthProbe 健康探针配置（Agent 本地执行） +// ============================================ + +// ProbeType 探针类型 +type ProbeType string + +const ( + ProbeTypeHTTP ProbeType = "http_get" // HTTP GET 探测 + ProbeTypeTCP ProbeType = "tcp_socket" // TCP 端口连接探测 + ProbeTypeExec ProbeType = "exec_command" // 命令执行探测 +) + +// ProbeStatus 探针结果状态 +type ProbeStatus string + +const ( + ProbeStatusUp ProbeStatus = "up" // 健康 + ProbeStatusDown ProbeStatus = "down" // 异常 +) + +// HealthProbe 健康探针配置表 +// 存储在平台 DB，Agent 拉取配置后在目标主机本地执行，结果随心跳上报 +type HealthProbe struct { + ID uint `gorm:"primaryKey;autoIncrement" json:"id"` + Name string `gorm:"type:varchar(128);not null;index" json:"name"` + Hostname string `gorm:"type:varchar(255);index;not null" json:"hostname"` + ProbeType ProbeType `gorm:"type:varchar(32);not null" json:"probe_type"` + // HTTP 探针配置 + HTTPURL string `gorm:"type:varchar(512)" json:"http_url,omitempty"` + HTTPMethod string `gorm:"type:varchar(10);default:GET" json:"http_method,omitempty"` + HTTPHeaders string `gorm:"type:text" json:"http_headers,omitempty"` + HTTPExpectCode int `gorm:"default:200" json:"http_expect_code,omitempty"` + HTTPExpectBody string `gorm:"type:varchar(256)" json:"http_expect_body,omitempty"` + // TCP 探针配置 + TCPHost string `gorm:"type:varchar(255)" json:"tcp_host,omitempty"` + TCPPort int `json:"tcp_port,omitempty"` + // Exec 探针配置 + ExecCommand string `gorm:"type:text" json:"exec_command,omitempty"` + // 通用配置 + Timeout int `gorm:"default:10" json:"timeout"` + Interval int `gorm:"default:60" json:"interval"` + Enabled bool `gorm:"default:true;index" json:"enabled"` + LastStatus ProbeStatus `gorm:"type:varchar(8);default:''" json:"last_status,omitempty"` + LastCheckedAt *time.Time `json:"last_checked_at,omitempty"` + LastLatencyMs int `json:"last_latency_ms"` + FailCount int `json:"fail_count"` + Description string `gorm:"type:varchar(256)" json:"description,omitempty"` + CreatedAt time.Time `gorm:"not null" json:"created_at"` + UpdatedAt time.Time `gorm:"not null" json:"updated_at"` + DeletedAt *time.Time `gorm:"index" json:"deleted_at,omitempty"` +} + +// TableName 指定表名 +func (HealthProbe) TableName() string { + return "monitor_health_probes" +} + +// HealthProbeResult Agent 上报的探针执行结果（作为心跳的一部分） +type HealthProbeResult struct { + ProbeID uint `json:"probe_id"` + Name string `json:"name"` + ProbeType string `json:"probe_type"` + Status ProbeStatus `json:"status"` // up / down + LatencyMs int `json:"latency_ms"` // 响应延迟(ms) + StatusCode int `json:"status_code,omitempty"` // HTTP 状态码 + Message string `json:"message,omitempty"` // 错误/附加信息 +} + diff --git a/backend/pkg/database/database.go b/backend/pkg/database/database.go index ee4783d..5c7d0a5 100644 --- a/backend/pkg/database/database.go +++ b/backend/pkg/database/database.go @@ -133,6 +133,7 @@ func AutoMigrate(db *gorm.DB) error { &models.AlertUpgrade{}, // 告警升级策略 // Logan.Li &models.AlertUpgradeRecord{}, // 告警升级记录 // Logan.Li &models.AlertSilence{}, // 告警静默 // Logan.Li + &models.HealthProbe{}, // 健康探针配置 // Logan.Li } var lastErr error diff --git a/frontend/src/components/monitor/alert-center/AlertCenter.tsx b/frontend/src/components/monitor/alert-center/AlertCenter.tsx index fc30426..19c42c1 100644 --- a/frontend/src/components/monitor/alert-center/AlertCenter.tsx +++ b/frontend/src/components/monitor/alert-center/AlertCenter.tsx @@ -18,9 +18,9 @@ import { useConfirm } from '@/components/ui/confirm-dialog'; import { alertCenterApi, agentApi, alertApi } from '@/services/api'; import { type Alert, type AlertRule, mapAlert, getSeverityBadge, getStatusBadge, + getChannelTypeLabel, } from './alert-center-helpers'; import { AlertDetailDialog, AlertSilenceDialog } from './AlertCenterDetail'; -import { AlertChannelsTab } from './AlertChannelsTab'; import { AlertRuleDialog } from '@/components/monitor/AlertRuleDialog'; import { ContactDialog } from './ContactDialog'; import { ContactGroupDialog } from './ContactGroupDialog'; @@ -265,7 +265,7 @@ const AlertCenter: React.FC = () => { 告警列表 告警规则 - 通知管理 + 报警联系人 报警联系组 @@ -501,11 +501,9 @@ const AlertCenter: React.FC = () => { )} - {/* 通知管理 Tab — 渠道 + 联系人合并 */} + {/* 报警联系人 Tab */} - -
-
+

报警联系人

)} -
{/* 报警联系组 Tab */} diff --git a/frontend/src/components/monitor/alert-center/ContactDialog.tsx b/frontend/src/components/monitor/alert-center/ContactDialog.tsx index ce40f4a..7ee6d1d 100644 --- a/frontend/src/components/monitor/alert-center/ContactDialog.tsx +++ b/frontend/src/components/monitor/alert-center/ContactDialog.tsx @@ -1,4 +1,4 @@ -// 报警联系人创建/编辑对话框 // Logan.Li +// 报警联系人创建/编辑对话框 — 渠道类型下拉 + webhook URL // Logan.Li import React, { useState, useEffect } from 'react'; import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogDescription, DialogFooter, @@ -6,10 +6,20 @@ import { import { Button } from '@/components/ui/button'; import { Input } from '@/components/ui/input'; import { Label } from '@/components/ui/label'; +import { + Select, SelectContent, SelectItem, SelectTrigger, SelectValue, +} from '@/components/ui/select'; import { toast } from 'sonner'; import { alertApi } from '@/services/api'; import type { Contact } from './alert-center-helpers'; +const CHANNEL_OPTIONS = [ + { value: 'feishu', label: '飞书', placeholder: 'https://open.feishu.cn/open-apis/bot/v2/hook/...' }, + { value: 'dingtalk', label: '钉钉', placeholder: 'https://oapi.dingtalk.com/robot/send?access_token=...' }, + { value: 'wechat', label: '企业微信', placeholder: 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=...' }, + { value: 'webhook', label: '自定义Webhook', placeholder: 'https://your-webhook-url...' }, +]; + export function ContactDialog({ open, onOpenChange, @@ -21,39 +31,37 @@ export function ContactDialog({ editingContact: Contact | null; onSaved: () => void; }) { - const [form, setForm] = useState({ - name: '', phone: '', email: '', dingtalk: '', feishu: '', wechat: '', - }); + const [channelType, setChannelType] = useState('feishu'); + const [form, setForm] = useState({ name: '', phone: '', email: '', webhook_url: '' }); const [saving, setSaving] = useState(false); useEffect(() => { if (open) { if (editingContact) { + setChannelType(editingContact.channel_type || 'feishu'); setForm({ name: editingContact.name || '', phone: editingContact.phone || '', email: editingContact.email || '', - dingtalk: editingContact.dingtalk || '', - feishu: editingContact.feishu || '', - wechat: editingContact.wechat || '', + webhook_url: editingContact.webhook_url || '', }); } else { - setForm({ name: '', phone: '', email: '', dingtalk: '', feishu: '', wechat: '' }); + setChannelType('feishu'); + setForm({ name: '', phone: '', email: '', webhook_url: '' }); } } }, [open, editingContact]); const handleSave = async () => { - if (!form.name.trim()) return; + if (!form.name.trim() || !form.webhook_url.trim()) return; setSaving(true); try { const payload = { name: form.name.trim(), phone: form.phone.trim() || undefined, email: form.email.trim() || undefined, - dingtalk: form.dingtalk.trim() || undefined, - feishu: form.feishu.trim() || undefined, - wechat: form.wechat.trim() || undefined, + channel_type: channelType, + webhook_url: form.webhook_url.trim(), }; if (editingContact) { await alertApi.updateContact(editingContact.id, payload); @@ -71,12 +79,14 @@ export function ContactDialog({ } }; + const selectedChannel = CHANNEL_OPTIONS.find(c => c.value === channelType); + return ( {editingContact ? '编辑报警联系人' : '添加报警联系人'} - 配置报警联系人信息（姓名必填，渠道选填） + 选择通知渠道并填入 Webhook 地址
@@ -87,53 +97,50 @@ export function ContactDialog({ placeholder="如：张三" />
-
- - setForm({ ...form, phone: e.target.value })} - placeholder="如：13800138000" - /> +
+
+ + setForm({ ...form, phone: e.target.value })} + placeholder="13800138000" + /> +
+
+ + setForm({ ...form, email: e.target.value })} + placeholder="zhangsan@example.com" + /> +
- - setForm({ ...form, email: e.target.value })} - placeholder="如：zhangsan@example.com" - /> + +
- + setForm({ ...form, dingtalk: e.target.value })} - placeholder="https://oapi.dingtalk.com/robot/send?access_token=..." - className="font-mono text-sm" - /> -
-
- - setForm({ ...form, feishu: e.target.value })} - placeholder="https://open.feishu.cn/open-apis/bot/v2/hook/..." - className="font-mono text-sm" - /> -
-
- - setForm({ ...form, wechat: e.target.value })} - placeholder="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=..." - className="font-mono text-sm" + value={form.webhook_url} + onChange={(e) => setForm({ ...form, webhook_url: e.target.value })} + placeholder={selectedChannel?.placeholder || 'https://...'} + className="font-mono text-xs" />
- diff --git a/frontend/src/components/monitor/alert-center/alert-center-helpers.tsx b/frontend/src/components/monitor/alert-center/alert-center-helpers.tsx index 3b7f585..7e19b02 100644 --- a/frontend/src/components/monitor/alert-center/alert-center-helpers.tsx +++ b/frontend/src/components/monitor/alert-center/alert-center-helpers.tsx @@ -60,9 +60,8 @@ export interface Contact { name: string; phone?: string; email?: string; - dingtalk?: string; - feishu?: string; - wechat?: string; + channel_type?: string; + webhook_url?: string; created_at?: string; updated_at?: string; } diff --git a/frontend/src/pages/HostMonitor.tsx b/frontend/src/pages/HostMonitor.tsx index a530d99..28f3d1a 100644 --- a/frontend/src/pages/HostMonitor.tsx +++ b/frontend/src/pages/HostMonitor.tsx @@ -10,7 +10,7 @@ import { Badge } from '@/components/ui/badge'; import { Button } from '@/components/ui/button'; import { Cpu, HardDrive, Network, Activity, Server, Loader2, - Info, Monitor, Disc, Zap, Clock, List, AlertTriangle, + Info, Monitor, Disc, Clock, List, HeartPulse, Plus, Trash2, RefreshCw, Edit, CheckCircle, XCircle, } from 'lucide-react'; import { AreaChart, Area, XAxis, YAxis, CartesianGrid, Tooltip, @@ -81,12 +81,36 @@ export function HostMonitor() { const [latest, setLatest] = useState(null); const [serverInfo, setServerInfo] = useState(null); + // 健康探针 + const [probes, setProbes] = useState([]); + const [probeDialogOpen, setProbeDialogOpen] = useState(false); + const [editingProbe, setEditingProbe] = useState(null); + const [probeForm, setProbeForm] = useState({ + name: '', hostname: '', probe_type: 'http_get', http_url: '', http_method: 'GET', + http_expect_code: 200, http_expect_body: '', + tcp_host: '', tcp_port: 3306, exec_command: '', timeout: 10, interval: 60, description: '', + }); + + // 加载探针 + + const fetchProbes = useCallback(() => { + if (!selectedHost) return; + agentApi.getHealthProbes({ hostname: selectedHost }).then((data: any) => { + if (Array.isArray(data)) setProbes(data); + else if (data?.list) setProbes(data.list); + else setProbes([]); + }).catch(() => setProbes([])); + }, [selectedHost]); + + useEffect(() => { + fetchProbes(); + }, [fetchProbes]); + const timeConfig = TIME_RANGES.find(tr => tr.hours === rangeHours) || TIME_RANGES[0]; useEffect(() => { agentApi.getLatest().then((data: any) => { const list = Array.isArray(data) ? data : (data?.data || data?.list || []); - setHosts(list); if (list.length > 0 && !selectedHost) { const m = urlHost ? list.find((h: any) => h.hostname === urlHost) : null; setSelectedHost(m?.hostname || list[0]?.hostname || ''); @@ -241,8 +265,10 @@ export function HostMonitor() { { id: 'basic', label: '基础监控', icon: }, { id: 'disk', label: '磁盘监控', icon: }, { id: 'network', label: '网络监控', icon: }, + { id: 'ports', label: '端口监控', icon: }, { id: 'process', label: '进程监控', icon: }, { id: 'system', label: '系统信息', icon: }, + { id: 'health_probes', label: '健康探针', icon: }, ].map(tab => ( {tab.icon}{tab.label} @@ -290,103 +316,135 @@ export function HostMonitor() { /> - {/* ═══ 进程监控 — v2.0 新增 ═══ */} - - {/* 进程 TOP 表格 */} + {/* ═══ 端口监控 ═══ */} +

- 进程 TOP {processTop.length} (按内存排序) + 监听端口 {portInfo.length}

- {processTop.length === 0 ? ( + {portInfo.length === 0 ? (
- Agent v2.0+ 支持，当前无数据 + 暂无端口数据
) : ( - - - - - - - - - - - - {processTop.map((p: any, i: number) => ( - - - - - - +
+
PID进程名内存 (MB)FD 数线程数
{p.pid}{p.name}{(p.memory_mb ?? 0).toFixed(0)}{p.fd_count ?? 0}{p.threads ?? 0}
+ + + + + + - ))} - -
端口协议进程状态
+ + + {portInfo.map((p: any, i: number) => ( + + {p.port} + + + {p.protocol || 'tcp'} + + + {p.process || '—'} + + + LISTEN + + + + ))} + + +
)} + - {/* Swap + FD + CS 快速卡片 */} -
- } - data={metrics.swap} color="#f43f5e" unit="%" domain={[0, 100]} aggType={aggType} - /> - } - data={metrics.fd} color="#22d3ee" unit="%" domain={[0, 100]} aggType={aggType} - /> - {/* CS 当前值卡片 */} - - -
- 上下文切换 + {/* ═══ 进程监控 — 简洁 5 列 ═══ */} + + + +

+ 进程 TOP {processTop.length} +

+ {processTop.length === 0 ? ( +
+ Agent v2.0+ 支持，等待数据上报…
-
- {csInfo?.ctx_per_sec != null ? `${(csInfo.ctx_per_sec / 1000).toFixed(1)}K/s` : '--'} + ) : ( +
+ + + + + + + + + + + + {processTop.map((p: any, i: number) => { + const cmd = p.cmdline || p.name || '?'; + const cpup = p.cpu_pct != null ? Number(p.cpu_pct) : null; + const memp = p.mem_pct != null ? Number(p.mem_pct) : null; + return ( + + + + + + + + ); + })} + +
进程名称 / CMDPID内存使用率(%)CPU使用率(%)打开文件数
+
+ + {cmd} + + +
+ 		{p.pid} + 50 ? 'text-rose-400' : 'text-cyan-400'}> + {memp != null ? memp.toFixed(2) : '--'} + + + 50 ? 'text-rose-400' : 'text-[var(--dg-text-primary)]'}> + {cpup != null ? cpup.toFixed(2) : '--'} + + {p.fd_count ?? '--'}
-
- 中断: {csInfo?.irq_per_sec != null ? `${(csInfo.irq_per_sec / 1000).toFixed(1)}K/s` : '--'} -
- - -
- - {/* FD / Swap 详情 */} -
- {fdInfo && fdInfo.max > 0 && ( - - - 文件句柄 -
- {fdInfo.allocated} - / {fdInfo.max} -
-
-
-
- - - )} - {swapInfo && swapInfo.total_gb > 0 && ( - - - Swap -
- {(swapInfo.used_gb ?? 0).toFixed(1)} - / {(swapInfo.total_gb ?? 0).toFixed(0)} GB -
-
-
50 ? 'bg-rose-500' : 'bg-amber-500'}`} - style={{ width: `${Math.min(swapInfo.usage_percent || 0, 100)}%` }} /> -
- - - )} -
+ )} + + {/* 系统信息 */} @@ -460,7 +518,183 @@ export function HostMonitor() { ); })()} + + {/* 健康探针 — Agent 本地执行 + 结果上报 */} + +
+

探针列表（Agent 在目标主机本地执行探测）

+
+ + +
+
+ {probes.length === 0 ? ( + + + 暂无健康探针，点击"添加探针"创建 + + + ) : ( +
+ {probes.map((p: any) => ( + + +
+
+ {p.last_status === 'up' ? : + p.last_status === 'down' ? : + } + {p.name} + + {p.probe_type === 'http_get' ? 'HTTP' : p.probe_type === 'tcp_socket' ? 'TCP' : 'Exec'} + +
+
+ + +
+
+
+ {p.probe_type === 'http_get' && ( +
URL: {p.http_url}
+ )} + {p.probe_type === 'tcp_socket' && ( +
{p.tcp_host}:{p.tcp_port}
+ )} + {p.probe_type === 'exec_command' && ( +
{p.exec_command}
+ )} +
+ 状态: + {p.last_status ? p.last_status.toUpperCase() : '未检测'} + + {p.last_latency_ms > 0 && 延迟: {p.last_latency_ms}ms} + {p.fail_count > 0 && 连续失败: {p.fail_count}次} + {p.last_checked_at && {new Date(p.last_checked_at).toLocaleString('zh-CN')}} +
+
+ + + ))} +
+ )} + + + {/* 添加/编辑探针对话框 */} + {probeDialogOpen && ( +
setProbeDialogOpen(false)}> +
e.stopPropagation()}> +

+ {editingProbe ? '编辑探针' : '添加探针'} +

+
+
+ + setProbeForm({ ...probeForm, name: e.target.value })} placeholder="如：cex-agent 健康检查" /> +
+
+ + setProbeForm({ ...probeForm, hostname: e.target.value })} /> +
+
+ + +
+ + {probeForm.probe_type === 'http_get' && ( + <> +
+ setProbeForm({ ...probeForm, http_url: e.target.value })} placeholder="http://localhost:8080/health" />
+
+
+ setProbeForm({ ...probeForm, http_expect_code: Number(e.target.value) })} />
+
+ setProbeForm({ ...probeForm, http_expect_body: e.target.value })} />
+
+ + )} + + {probeForm.probe_type === 'tcp_socket' && ( +
+
+ setProbeForm({ ...probeForm, tcp_host: e.target.value })} placeholder="localhost" />
+
+ setProbeForm({ ...probeForm, tcp_port: Number(e.target.value) })} />
+
+ )} + + {probeForm.probe_type === 'exec_command' && ( +
+ setProbeForm({ ...probeForm, exec_command: e.target.value })} placeholder="pgrep -x mysqld" />
+ )} + +
+
+ setProbeForm({ ...probeForm, timeout: Number(e.target.value) })} />
+
+ setProbeForm({ ...probeForm, interval: Number(e.target.value) })} />
+
+ +
+ setProbeForm({ ...probeForm, description: e.target.value })} />
+
+
+ + +
+
+
+ )}
); } diff --git a/frontend/src/services/api.ts b/frontend/src/services/api.ts index d307a5f..d271e69 100644 --- a/frontend/src/services/api.ts +++ b/frontend/src/services/api.ts @@ -885,6 +885,11 @@ export const agentApi = { batchUpgrade: (serverIds: number[]) => http.post('/agent/batch-upgrade', { server_ids: serverIds }), listAlerts: (params?: Record) => http.get('/agent/alerts', params), resolveAlert: (id: number) => http.post(`/agent/alerts/${id}/resolve`), + // 健康探针 + getHealthProbes: (params?: Record) => http.get('/agent/health-probes', params), + createHealthProbe: (data: any) => http.post('/agent/health-probes', data), + updateHealthProbe: (id: number, data: any) => http.put(`/agent/health-probes/${id}`, data), + deleteHealthProbe: (id: number) => http.delete(`/agent/health-probes/${id}`), }; // ==================== 区域管理 API ====================