// Package task 的 client.go 实现与后端任务接口的 HTTP 交互。 // 涵盖拉取待执行任务(poll)与回传执行结果(submit result)。 // 仅依赖 Go 标准库,失败时返回错误,由调用方决定重试策略。 // // 署名 Logan.Li package task import ( "bytes" "context" "encoding/json" "fmt" "io" "log" "net/http" "net/url" "strings" "time" ) 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 版本 ) // Client 负责与后端任务接口通信。 type Client struct { server string token string hostname string httpClient *http.Client 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, } } // 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 } 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 } 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) } 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) } }