attacker
ac0b8e75c0
- Go agent: CPU/mem/disk/port collectors, HTTP report, static binary - Cross-compile: linux/amd64 + linux/arm64 (statically linked) - One-line install script for all Linux distros - Backend: POST /api/v1/agent/report (no JWT) + GET /agent/latest,history (JWT) - CloudBilling.tsx: fixed syntax errors, stats/trends/records/alerts - K8s menu renamed to Kubernetes管理 Signed-off-by: Logan.Li
123 lines
3.6 KiB
Go
123 lines
3.6 KiB
Go
// Package reporter 实现 AIOps Agent 向后端上报采集结果的能力。
|
||
// 通过 HTTP POST 将指标 JSON 发送至 {server}/api/v1/agent/report，
|
||
// 并在请求头携带 Bearer Token 完成鉴权。失败时进行有限次指数退避重试。
|
||
//
|
||
// 署名 Logan.Li
|
||
package reporter
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"ezops/agent/model"
|
||
)
|
||
|
||
const (
|
||
reportPath = "/api/v1/agent/report" // 后端上报接口路径
|
||
defaultTimeout = 10 * time.Second // 单次 HTTP 请求超时
|
||
maxRetries = 3 // 最大重试次数（含首次）
|
||
userAgent = "aiops-agent/1.0" // 标识 Agent 版本，便于后端识别
|
||
)
|
||
|
||
// Reporter 负责将采集报告上报至 AIOps 后端
|
||
type Reporter struct {
|
||
server string
|
||
token string
|
||
httpClient *http.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,
|
||
}
|
||
}
|
||
|
||
// normalizeServer 去除尾部斜杠，保证 URL 拼接正确
|
||
func normalizeServer(server string) string {
|
||
return strings.TrimRight(server, "/")
|
||
}
|
||
|
||
// Send 将报告序列化为 JSON 并 POST 到后端，失败时进行有限次重试。
|
||
// 重试过程可被传入的上下文取消打断（如 Agent 收到退出信号时）。
|
||
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
|
||
}
|
||
}
|