package prometheus import ( "context" "encoding/json" "fmt" "io" "net/http" "net/url" "strconv" "time" ) // ============================================ // Prometheus HTTP API 客户端 // ============================================ // Client Prometheus API客户端 // 封装对Prometheus HTTP API的调用,支持即时查询、范围查询、指标发现等 type Client struct { baseURL string // Prometheus服务地址,如 http://localhost:9090 httpClient *http.Client // HTTP客户端 username string // 基本认证用户名 password string // 基本认证密码 } // ClientOption 客户端配置选项 type ClientOption func(*Client) // WithTimeout 设置请求超时 func WithTimeout(timeout time.Duration) ClientOption { return func(c *Client) { c.httpClient.Timeout = timeout } } // WithBasicAuth 设置基本认证 func WithBasicAuth(username, password string) ClientOption { return func(c *Client) { c.username = username c.password = password } } // NewClient 创建Prometheus客户端 // baseURL: Prometheus服务地址,如 http://localhost:9090 func NewClient(baseURL string, opts ...ClientOption) *Client { c := &Client{ baseURL: baseURL, httpClient: &http.Client{ Timeout: 30 * time.Second, }, } for _, opt := range opts { opt(c) } return c } // SetTimeout 设置请求超时 func (c *Client) SetTimeout(timeout time.Duration) { c.httpClient.Timeout = timeout } // doRequest 执行HTTP请求并解析JSON响应 func (c *Client) doRequest(ctx context.Context, path string, params url.Values, result interface{}) error { u, err := url.Parse(c.baseURL + path) if err != nil { return fmt.Errorf("invalid URL: %w", err) } u.RawQuery = params.Encode() req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) if err != nil { return fmt.Errorf("create request failed: %w", err) } if c.username != "" { req.SetBasicAuth(c.username, c.password) } resp, err := c.httpClient.Do(req) if err != nil { return fmt.Errorf("HTTP request failed: %w", err) } defer resp.Body.Close() body, err := io.ReadAll(resp.Body) if err != nil { return fmt.Errorf("read response body failed: %w", err) } if resp.StatusCode != http.StatusOK { return fmt.Errorf("unexpected status code: %d, body: %s", resp.StatusCode, string(body)) } if result != nil { if err := json.Unmarshal(body, result); err != nil { return fmt.Errorf("unmarshal response failed: %w, body: %s", err, string(body)) } } return nil } // ============================================ // 查询响应数据结构 // ============================================ // QueryResponse 查询响应 type QueryResponse struct { Status string `json:"status"` Data struct { ResultType string `json:"resultType"` Result []MetricData `json:"result"` } `json:"data"` Error string `json:"error,omitempty"` ErrorType string `json:"errorType,omitempty"` } // MetricData 指标数据(即时查询结果) type MetricData struct { Metric map[string]string `json:"metric"` // 标签集合 Value []interface{} `json:"value"` // [时间戳, 值] Values [][]interface{} `json:"values"` // 范围查询时使用 } // GetTimestamp 获取时间戳 func (m *MetricData) GetTimestamp() int64 { if len(m.Value) >= 1 { if ts, ok := m.Value[0].(float64); ok { return int64(ts) } if tsStr, ok := m.Value[0].(string); ok { if ts, err := strconv.ParseFloat(tsStr, 64); err == nil { return int64(ts) } } } return 0 } // GetValue 获取值(字符串) func (m *MetricData) GetValue() string { if len(m.Value) >= 2 { if v, ok := m.Value[1].(string); ok { return v } } return "" } // GetFloatValue 获取浮点值 func (m *MetricData) GetFloatValue() (float64, error) { v := m.GetValue() if v == "" { return 0, fmt.Errorf("empty value") } return strconv.ParseFloat(v, 64) } // MetricPoint 时序数据点 type MetricPoint struct { Timestamp int64 `json:"timestamp"` Value float64 `json:"value"` } // GetTimeSeries 将范围查询结果转换为时序数据 func (m *MetricData) GetTimeSeries() ([]MetricPoint, error) { var points []MetricPoint for _, v := range m.Values { if len(v) < 2 { continue } var ts int64 switch t := v[0].(type) { case float64: ts = int64(t) case string: if f, err := strconv.ParseFloat(t, 64); err == nil { ts = int64(f) } } var val float64 switch s := v[1].(type) { case string: if f, err := strconv.ParseFloat(s, 64); err == nil { val = f } case float64: val = s } points = append(points, MetricPoint{Timestamp: ts, Value: val}) } return points, nil } // ============================================ // 即时查询 (Instant Query) // ============================================ // Query 执行即时查询 // query: PromQL查询表达式 // ts: 可选的查询时间戳,为0时使用当前时间 // 返回查询结果列表和错误 func (c *Client) Query(ctx context.Context, query string, ts time.Time) ([]MetricData, error) { if query == "" { return nil, fmt.Errorf("query expression is required") } params := url.Values{} params.Set("query", query) if !ts.IsZero() { params.Set("time", strconv.FormatInt(ts.Unix(), 10)) } var resp QueryResponse if err := c.doRequest(ctx, "/api/v1/query", params, &resp); err != nil { return nil, fmt.Errorf("query failed: %w", err) } if resp.Status != "success" { return nil, fmt.Errorf("query error: %s", resp.Error) } return resp.Data.Result, nil } // QuerySingle 执行即时查询并返回单个结果 func (c *Client) QuerySingle(ctx context.Context, query string, ts time.Time) (*MetricData, error) { results, err := c.Query(ctx, query, ts) if err != nil { return nil, err } if len(results) == 0 { return nil, fmt.Errorf("no data returned for query: %s", query) } return &results[0], nil } // ============================================ // 范围查询 (Range Query) // ============================================ // RangeQueryOptions 范围查询选项 type RangeQueryOptions struct { Start time.Time // 开始时间 End time.Time // 结束时间 Step time.Duration // 步长 } // QueryRange 执行范围查询 // query: PromQL查询表达式 // opts: 范围查询选项 // 返回时序数据列表和错误 func (c *Client) QueryRange(ctx context.Context, query string, opts RangeQueryOptions) ([]MetricData, error) { if query == "" { return nil, fmt.Errorf("query expression is required") } params := url.Values{} params.Set("query", query) params.Set("start", strconv.FormatInt(opts.Start.Unix(), 10)) params.Set("end", strconv.FormatInt(opts.End.Unix(), 10)) params.Set("step", strconv.FormatInt(int64(opts.Step.Seconds()), 10)) var resp QueryResponse if err := c.doRequest(ctx, "/api/v1/query_range", params, &resp); err != nil { return nil, fmt.Errorf("query range failed: %w", err) } if resp.Status != "success" { return nil, fmt.Errorf("query range error: %s", resp.Error) } return resp.Data.Result, nil } // QueryRangePoints 执行范围查询并返回时序数据点 func (c *Client) QueryRangePoints(ctx context.Context, query string, opts RangeQueryOptions) (map[string][]MetricPoint, error) { results, err := c.QueryRange(ctx, query, opts) if err != nil { return nil, err } series := make(map[string][]MetricPoint) for _, r := range results { key := r.Metric["__name__"] if key == "" { // 使用所有标签拼接作为key for k, v := range r.Metric { if k != "__name__" { key += k + "=" + v + "," } } } points, err := r.GetTimeSeries() if err != nil { continue } series[key] = points } return series, nil } // ============================================ // 指标元数据查询 // ============================================ // LabelValuesResponse 标签值响应 type LabelValuesResponse struct { Status string `json:"status"` Data []string `json:"data"` Error string `json:"error,omitempty"` } // MetricsResponse 指标列表响应 type MetricsResponse struct { Status string `json:"status"` Data []string `json:"data"` Error string `json:"error,omitempty"` } // GetMetrics 获取指标名称列表 // 返回Prometheus中所有可用的指标名称 func (c *Client) GetMetrics(ctx context.Context) ([]string, error) { var resp MetricsResponse if err := c.doRequest(ctx, "/api/v1/label/__name__/values", nil, &resp); err != nil { return nil, fmt.Errorf("get metrics failed: %w", err) } if resp.Status != "success" { return nil, fmt.Errorf("get metrics error: %s", resp.Error) } return resp.Data, nil } // GetLabelValues 获取指定标签的所有值 // labelName: 标签名称,如 "instance", "job" // 返回该标签的所有不重复值 func (c *Client) GetLabelValues(ctx context.Context, labelName string, matchers ...string) ([]string, error) { if labelName == "" { return nil, fmt.Errorf("label name is required") } params := url.Values{} for _, m := range matchers { params.Add("match[]", m) } var resp LabelValuesResponse if err := c.doRequest(ctx, fmt.Sprintf("/api/v1/label/%s/values", labelName), params, &resp); err != nil { return nil, fmt.Errorf("get label values failed: %w", err) } if resp.Status != "success" { return nil, fmt.Errorf("get label values error: %s", resp.Error) } return resp.Data, nil } // ============================================ // 指标发现 (Series Discovery) // ============================================ // SeriesResponse 指标发现响应 type SeriesResponse struct { Status string `json:"status"` Data []map[string]string `json:"data"` Error string `json:"error,omitempty"` } // SeriesDiscovery 查询满足条件的指标系列 // matchers: 指标选择器,如 "up{job=\"prometheus\"}" // 返回所有匹配的时间序列的标签集合 func (c *Client) SeriesDiscovery(ctx context.Context, matchers []string, start, end time.Time) ([]map[string]string, error) { if len(matchers) == 0 { return nil, fmt.Errorf("at least one matcher is required") } params := url.Values{} for _, m := range matchers { params.Add("match[]", m) } if !start.IsZero() { params.Set("start", strconv.FormatInt(start.Unix(), 10)) } if !end.IsZero() { params.Set("end", strconv.FormatInt(end.Unix(), 10)) } var resp SeriesResponse if err := c.doRequest(ctx, "/api/v1/series", params, &resp); err != nil { return nil, fmt.Errorf("series discovery failed: %w", err) } if resp.Status != "success" { return nil, fmt.Errorf("series discovery error: %s", resp.Error) } return resp.Data, nil } // GetSeriesByMetric 通过指标名称查询所有系列 // metricName: 指标名称 func (c *Client) GetSeriesByMetric(ctx context.Context, metricName string) ([]map[string]string, error) { if metricName == "" { return nil, fmt.Errorf("metric name is required") } return c.SeriesDiscovery(ctx, []string{metricName}, time.Time{}, time.Time{}) } // ============================================ // 元数据和规则查询 // ============================================ // Target 监控目标 type Target struct { Targets []map[string]string `json:"targets"` Labels map[string]string `json:"labels"` } // TargetsResponse 目标发现响应 type TargetsResponse struct { Status string `json:"status"` Data struct { ActiveTargets []ActiveTarget `json:"activeTargets"` DroppedTargets []DroppedTarget `json:"droppedTargets"` } `json:"data"` Error string `json:"error,omitempty"` } // ActiveTarget 活跃目标 type ActiveTarget struct { DiscoveredLabels map[string]string `json:"discoveredLabels"` Labels map[string]string `json:"labels"` ScrapePool string `json:"scrapePool"` ScrapeURL string `json:"scrapeUrl"` GlobalURL string `json:"globalUrl"` LastError string `json:"lastError"` LastScrape time.Time `json:"lastScrape"` LastScrapeDuration float64 `json:"lastScrapeDuration"` Health string `json:"health"` } // DroppedTarget 丢弃的目标 type DroppedTarget struct { DiscoveredLabels map[string]string `json:"discoveredLabels"` } // GetTargets 获取监控目标列表 func (c *Client) GetTargets(ctx context.Context) (*TargetsResponse, error) { var resp TargetsResponse if err := c.doRequest(ctx, "/api/v1/targets", nil, &resp); err != nil { return nil, fmt.Errorf("get targets failed: %w", err) } if resp.Status != "success" { return nil, fmt.Errorf("get targets error: %s", resp.Error) } return &resp, nil } // ============================================ // 工具方法 // ============================================ // Health 检查Prometheus服务健康状态 func (c *Client) Health(ctx context.Context) (string, error) { u, err := url.Parse(c.baseURL + "/-/healthy") if err != nil { return "", fmt.Errorf("invalid URL: %w", err) } req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) if err != nil { return "", fmt.Errorf("create request failed: %w", err) } if c.username != "" { req.SetBasicAuth(c.username, c.password) } resp, err := c.httpClient.Do(req) if err != nil { return "", fmt.Errorf("health check failed: %w", err) } defer resp.Body.Close() body, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { return "", fmt.Errorf("prometheus unhealthy: %d - %s", resp.StatusCode, string(body)) } return string(body), nil } // BuildSelector 构建PromQL标签选择器 // metricName: 指标名称 // labels: 标签键值对 // 返回格式化后的选择器字符串,如: node_cpu_seconds_total{mode="idle",cpu="0"} func BuildSelector(metricName string, labels map[string]string) string { selector := metricName if len(labels) > 0 { selector += "{" first := true for k, v := range labels { if !first { selector += "," } selector += fmt.Sprintf("%s=\"%s\"", k, v) first = false } selector += "}" } return selector } // FormatDuration 将time.Duration格式化为Prometheus持续时间字符串 // 如: 1h30m, 5m, 10s func FormatDuration(d time.Duration) string { if d <= 0 { return "0s" } var result string if h := int(d.Hours()); h > 0 { result += fmt.Sprintf("%dh", h) d -= time.Duration(h) * time.Hour } if m := int(d.Minutes()); m > 0 { result += fmt.Sprintf("%dm", m) d -= time.Duration(m) * time.Minute } if s := int(d.Seconds()); s > 0 { result += fmt.Sprintf("%ds", s) } return result }