Files
aiops/agent/client/http.go
T
attacker d656520428 feat: v0.34 健康探针(Architecture B) + Agent v2.1 + 端口监控 + 告警联系人重构
- 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: 健康探针/端口监控
2026-06-24 19:18:10 +08:00

141 lines
3.5 KiB
Go

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