Prometheus 数据获取技术分享

摘要

本文档详细介绍了通过 Grafana API 获取 Prometheus 时序数据的完整技术方案，深入分析了 Prometheus 查询语言（PromQL）的特点，并与 Graphite 数据获取方案进行了全面对比。该方案实现了从查询表达式构建、API 调用、响应解析到数据存储的全流程自动化处理。

1. 技术背景

1.1 Prometheus 简介

Prometheus 是一个开源的系统监控和告警工具包，最初由 SoundCloud 开发。它采用拉取（pull）模型收集指标数据，使用多维数据模型存储时序数据，并提供了强大的查询语言 PromQL。Prometheus 已经成为云原生监控领域的标准解决方案，广泛应用于 Kubernetes 集群监控、微服务监控等场景。

1.2 与 Graphite 的对比

Prometheus 和 Graphite 都是时序数据库，但在设计理念和实现方式上存在显著差异：

数据模型

• Prometheus：采用标签（label）多维数据模型，指标名称和标签组合唯一标识一个时间序列
• Graphite：采用层次化路径模型，通过点分隔的路径组织指标

查询语言

• Prometheus：使用 PromQL，支持丰富的聚合、过滤、数学运算等操作
• Graphite：使用函数式语法，通过函数嵌套实现复杂查询

数据采集

• Prometheus：主动拉取（pull）模式，通过服务发现自动发现目标
• Graphite：被动接收（push）模式，应用主动推送数据

存储方式

• Prometheus：本地时序数据库，支持高效压缩
• Graphite：基于 Whisper 文件系统存储

1.3 业务场景

在实际监控场景中，Prometheus 常用于：

• Kubernetes 集群监控
• 微服务应用监控
• 容器资源使用监控
• 业务指标监控

通过 Grafana API 获取 Prometheus 数据，可以实现：

• 批量查询多个指标
• 自动化报告生成
• 数据分析和可视化
• 告警规则验证

2. 技术架构

2.1 整体流程

Prometheus 数据获取流程分为四个主要阶段：

阶段一：查询表达式构建

根据业务需求，构建 PromQL 查询表达式。表达式可以包含指标名称、标签选择器、聚合函数、数学运算等。

阶段二：API 请求构建

通过 Grafana 的 Data Source API，构建查询请求。请求体包含查询表达式、时间范围、数据源配置等信息。

阶段三：API 调用与响应处理

发送 POST 请求到 Grafana API，获取查询结果。Grafana 返回的数据采用特定的 JSON 格式，需要进行解析和转换。

阶段四：数据提取与存储

从响应中提取时间序列数据，包括时间戳、指标值、标签信息等，最终以结构化格式（如 CSV）保存。

2.2 核心组件

查询表达式构建器（Query Expression Builder）

负责构建符合 PromQL 语法的查询表达式。需要处理指标名称、标签选择器、函数调用、运算符等元素。

Grafana API 客户端（Grafana API Client）

封装 Grafana Data Source API 的调用逻辑，处理认证、请求构建、响应解析等操作。支持多种认证方式（API Key、Cookie、Basic Auth）。

响应解析器（Response Parser）

解析 Grafana API 返回的 JSON 响应，提取时间序列数据。Grafana 返回的数据采用 Frame 格式，需要解析 schema 和 data 字段。

数据转换器（Data Transformer）

将解析后的数据转换为目标格式，包括时间戳转换、数值格式化、标签信息提取等。

3. Grafana Data Source API 详解

3.1 API 接口说明

Grafana 提供了统一的 Data Source API（/api/ds/query），用于查询各种数据源，包括 Prometheus。该接口采用 POST 请求，请求体和响应都使用 JSON 格式。

3.2 请求参数

URL 参数

• ds_type：数据源类型，对于 Prometheus 固定为 prometheus
• requestId：请求 ID，用于追踪请求，通常使用时间戳生成唯一 ID

请求体结构

请求体是一个 JSON 对象，包含以下字段：

queries（必需）

查询数组，每个元素代表一个查询。对于 Prometheus，通常只包含一个查询对象，包含以下字段：

• datasource：数据源配置对象
  • type：数据源类型，固定为 prometheus
  • uid：数据源 UID，在 Grafana 数据源配置中可以看到
• editorMode：编辑器模式，通常为 code（代码模式）
• expr：PromQL 查询表达式
• legendFormat：图例格式，__auto 表示自动生成
• range：是否为范围查询，通常为 true
• refId：查询引用 ID，用于标识查询结果，通常为 A
• exemplar：是否包含示例数据，通常为 false
• interval：采样间隔（字符串格式），可以为空
• intervalMs：采样间隔（毫秒），默认 20000（20 秒）
• maxDataPoints：最大数据点数，默认 1000

from（必需）

查询起始时间戳（毫秒级）。

to（必需）

查询结束时间戳（毫秒级）。

请求体构建示例：

# 构建请求 URLurl = f"{GRAFANA_HOST}/api/ds/query"params = {    "ds_type": "prometheus",    "requestId": f"SQR{int(datetime.now().timestamp() * 1000)}"}# 构建请求体request_body = {    "queries": [        {            "datasource": {                "type": "prometheus",                "uid": datasource_uid            },            "editorMode": "code",            "expr": expr,  # PromQL 查询表达式            "legendFormat": "__auto",            "range": True,            "refId": "A",            "exemplar": False,            "interval": "",            "intervalMs": interval_ms or 20000,            "maxDataPoints": max_data_points or 1000        }    ],    "from": str(from_ms),  # 毫秒级时间戳    "to": str(to_ms)       # 毫秒级时间戳}# 发送 POST 请求response = requests.post(    url,    params=params,    json=request_body,    headers=headers,    timeout=TIMEOUT)
完整请求示例：
POST https://$hostname/api/ds/query?ds_type=prometheus&requestId=$requestId{  "queries": [    {      "datasource": {        "type": "prometheus",        "uid": "$datasource_uid"      },      "editorMode": "code",      "expr": "sum(kube_pod_status_ready{namespace=\"$namespace\",condition=\"$condition\"})",      "legendFormat": "__auto",      "range": true,      "refId": "A",      "exemplar": false,      "interval": "",      "intervalMs": $intervalMs,      "maxDataPoints": $maxDataPoints    }  ],  "from": "$from",  "to": "$to"}

3.3 响应格式

Grafana API 返回的响应采用 Frame 格式，结构如下：

{  "results": {    "A": {      "frames": [        {          "schema": {            "fields": [              {                "name": "Time",                "type": "time"              },              {                "name": "Value",                "type": "number",                "labels": {                  "label1": "value1",                  "label2": "value2"                }              }            ]          },          "data": {            "values": [              [timestamp1, timestamp2, ...],              [value1, value2, ...]            ]          }        }      ]    }  }}

响应字段说明

• results：查询结果对象，key 为查询的 refId
• frames：数据帧数组，每个帧代表一个时间序列
• schema：数据模式定义
  • fields：字段数组
    • 第一个字段通常是时间字段（Time）
    • 第二个字段是值字段（Value），包含 labels 对象
• data.values：数据值数组
  • 第一个子数组：时间戳数组（毫秒级）
  • 第二个子数组：指标值数组

3.4 认证方式

Grafana API 支持多种认证方式：

API Key 认证（推荐）

在请求头中添加 Authorization: Bearer <api_key>。

Cookie 认证

在请求头中添加 Cookie: <cookie_string>，从浏览器开发者工具中获取。

Basic Auth

使用 HTTP Basic Authentication，在请求中提供用户名和密码。

4. PromQL 查询语言

4.1 基本语法

PromQL 采用表达式语法，基本结构如下：

<metric_name>{<label_selectors>}[<time_range>]

示例：

kube_pod_status_ready{namespace="$namespace",condition="$condition"}

4.2 指标名称和标签

指标名称

指标名称通常采用下划线分隔的命名方式，如 kube_pod_status_ready。

标签选择器

使用花括号 {} 定义标签选择器，支持以下操作符：

• =：精确匹配
• !=：不等于
• =~：正则匹配
• !~：正则不匹配

示例：

kube_pod_status_ready{namespace="$namespace",pod=~"$pod_pattern",condition="$condition"}

4.3 聚合函数

PromQL 提供了丰富的聚合函数：

sum()
对多个时间序列求和。
示例：
sum(kube_pod_status_ready{namespace="$namespace"})
avg()
计算平均值。
示例：
avg(cpu_usage{service="$service"})
max() / min()
获取最大值或最小值。
示例：
max(response_time{service="$service"})
count()
统计时间序列数量。
示例：
count(kube_pod_status_ready{namespace="$namespace",condition="$condition"})

4.4 时间范围选择器

使用方括号 [] 指定时间范围，用于获取历史数据。

示例：

cpu_usage[5m]
获取过去 5 分钟的数据。

4.5 数学运算

PromQL 支持基本的数学运算：

算术运算

• +：加法
• -：减法
• *：乘法
• /：除法
• %：取模

比较运算

• ==：等于
• !=：不等于
• >、<、>=、<=：大小比较

逻辑运算

• and：逻辑与
• or：逻辑或
• unless：逻辑非

5. 数据获取流程

5.1 查询表达式构建

指标选择

根据业务需求，选择合适的指标名称。指标名称通常反映监控对象的类型和属性。

标签过滤

使用标签选择器过滤出目标时间序列。标签过滤可以精确到具体的服务、实例、环境等维度。

聚合操作

对于需要汇总的指标，使用聚合函数进行计算。可以按不同的标签维度进行分组聚合。

时间范围

对于需要历史数据的场景，使用时间范围选择器获取指定时间段的数据。

5.2 API 请求构建

数据源配置

从 Grafana 配置中获取数据源 UID，用于标识要查询的数据源。

查询对象构建

根据查询表达式，构建查询对象，设置相关参数：

• 查询表达式（expr）
• 采样间隔（intervalMs）
• 最大数据点数（maxDataPoints）

时间范围设置

将查询时间范围转换为毫秒级时间戳，设置 from 和 to 参数。

请求 ID 生成

使用时间戳生成唯一的请求 ID，用于请求追踪。

5.3 API 调用

认证处理

根据配置的认证方式，在请求头中添加相应的认证信息。

HTTP 请求发送

发送 POST 请求到 Grafana API，设置适当的超时时间。

响应处理

检查 HTTP 状态码，处理网络异常和服务器错误。解析 JSON 响应，提取查询结果。

5.4 数据解析与存储

Frame 解析

从响应中提取 frames 数组，遍历每个数据帧。

Schema 解析

从 schema.fields 中提取字段定义，获取时间字段和值字段的配置，以及标签信息。

数据提取

从 data.values 中提取时间戳数组和指标值数组，进行配对处理。

Frame 解析实现：

def parse_prometheus_response(response: Dict) -> List[Dict]:    results = []    if "results" not in response:        return results    # 遍历查询结果（按 refId 分组）    for ref_id, result in response["results"].items():        if "frames" not in result:            continue        # 遍历每个数据帧        for frame in result["frames"]:            data = frame.get("data", {})            values = data.get("values", [])            if len(values) < 2:                continue            # 提取时间戳数组和指标值数组            timestamps = values[0]  # 时间戳数组（毫秒）            metric_values = values[1]  # 值数组            # 获取标签信息            schema = frame.get("schema", {})            fields = schema.get("fields", [])            metric_labels = {}            if len(fields) > 1:                labels = fields[1].get("labels", {})                metric_labels = labels or {}            # 配对时间戳和值            for ts, val in zip(timestamps, metric_values):                results.append({                    "timestamp_ms": ts,                    "timestamp": ts / 1000,  # 转换为秒                    "value": val,                    "metric": metric_labels,                    "ref_id": ref_id                })    return results

时间戳转换

将毫秒级时间戳转换为秒级时间戳和可读的日期时间格式。

时间戳转换实现：

# 毫秒级时间戳转换为秒级timestamp_seconds = timestamp_ms / 1000# 转换为可读格式try:    timestamp_str = datetime.fromtimestamp(timestamp_seconds).strftime('%Y-%m-%d %H:%M:%S')except:    timestamp_str = str(timestamp_seconds)

标签信息提取

从值字段的 labels 对象中提取标签信息，用于后续的数据分析和过滤。

标签信息提取实现：

# 从 schema 中提取标签信息schema = frame.get("schema", {})fields = schema.get("fields", [])if len(fields) > 1:    # 值字段包含标签信息    value_field = fields[1]    labels = value_field.get("labels", {})    # 标签信息是一个字典，例如：    # {    #   "namespace": "$namespace",    #   "pod": "$pod_name",    #   "condition": "$condition"    # }    # 将标签序列化为 JSON 字符串，便于存储    labels_json = json.dumps(labels, ensure_ascii=False)

CSV 写入

将解析后的数据写入 CSV 文件，包含以下字段：

• 时间戳（毫秒）
• 时间戳（秒）
• 时间戳（可读格式）
• 指标值
• 标签信息（JSON 格式）
• 查询引用 ID

CSV 写入实现：

def save_to_csv(results: List[Dict], filename: str = "prometheus_results.csv"):    with open(filename, "w", newline="", encoding="utf-8-sig") as f:        writer = csv.writer(f)        # 写入表头        writer.writerow([            "timestamp_ms",            "timestamp",            "timestamp_str",            "value",            "metric_labels",            "ref_id"        ])        # 写入数据行        for r in results:            ts_ms = r["timestamp_ms"]            ts = r["timestamp"]            val = r["value"]            metric = json.dumps(r["metric"], ensure_ascii=False)            ref_id = r.get("ref_id", "")            # 转换时间戳为可读格式            try:                ts_str = datetime.fromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S')            except:                ts_str = str(ts)            writer.writerow([ts_ms, int(ts), ts_str, val, metric, ref_id])

6. Prometheus 与 Graphite 对比

6.1 数据模型对比

6.2 查询语言对比

6.3 API 接口对比

6.4 数据获取流程对比

6.5 适用场景对比

6.6 性能对比

7. 关键技术点

7.1 时间戳处理

时间格式转换

Prometheus 查询使用毫秒级时间戳，而系统内部可能使用秒级时间戳。需要进行正确的转换：

• API 请求：使用毫秒级时间戳
• 数据存储：可以同时保存毫秒级和秒级时间戳
• 数据展示：转换为可读的日期时间格式

7.2 标签信息处理

标签提取

从 Frame 的 schema 中提取标签信息，标签信息存储在值字段的 labels 对象中。

标签序列化

将标签对象序列化为 JSON 字符串，便于在 CSV 中存储和后续解析。

标签过滤

在数据提取阶段，可以根据标签信息进行过滤，只保留符合条件的数据。

7.3 错误处理

网络异常

API 调用可能因网络问题失败，需要实现重试机制或错误记录。

数据异常

API 可能返回空数据或异常数据，需要识别这些情况并做相应处理。

表达式错误

PromQL 表达式可能存在语法错误，需要在构建阶段进行验证。

7.4 性能优化

批量查询

支持批量查询多个指标，提高处理效率。

数据点限制

使用 maxDataPoints 参数限制返回的数据点数量，避免响应过大。

采样间隔

合理设置采样间隔（intervalMs），平衡数据精度和查询性能。

8. 应用场景

8.1 Kubernetes 集群监控

通过 Prometheus 查询 Kubernetes 集群的 Pod 状态、资源使用等指标，用于集群健康监控和容量规划。

示例查询：

sum(kube_pod_status_ready{namespace="$namespace",condition="$condition"})

8.2 微服务监控

查询微服务的性能指标，如响应时间、错误率、吞吐量等，用于服务性能分析和优化。

8.3 容器资源监控

查询容器的 CPU、内存、网络等资源使用情况，用于资源使用分析和容量规划。

8.4 业务指标监控

查询业务相关的指标，如订单量、用户数、收入等，用于业务分析和决策支持。

9. 最佳实践

9.1 查询表达式优化

使用标签过滤

充分利用标签选择器进行过滤，减少查询的数据量。

合理使用聚合

根据业务需求选择合适的聚合函数，避免过度聚合导致信息丢失。

时间范围选择

合理设置查询时间范围，避免查询过大的时间范围导致性能问题。

9.2 API 调用优化

请求频率控制

在批量查询时，控制请求频率，避免对 Grafana 服务器造成过大压力。

超时设置

设置适当的请求超时时间，避免长时间等待。

错误重试

实现错误重试机制，提高查询的可靠性。

9.3 数据处理优化

数据过滤

在数据存储阶段，可以过滤掉无效数据，减少存储空间。

数据压缩

对于大量数据，可以考虑压缩存储，节省存储空间。

增量更新

对于定期查询的场景，可以实现增量更新，只查询新增的数据。

10. 总结

本文档详细介绍了通过 Grafana API 获取 Prometheus 数据的完整技术方案，并与 Graphite 数据获取方案进行了全面对比。

关键技术点：

• Grafana Data Source API 的使用方法
• PromQL 查询语言的语法和特点
• Frame 格式响应的解析方法
• 时间戳和标签信息的处理

与 Graphite 的主要区别：

• 数据模型：Prometheus 采用标签模型，Graphite 采用路径模型
• 查询语言：PromQL 采用声明式语法，Graphite 采用函数式语法
• API 接口：Prometheus 通过 Grafana API（POST），Graphite 使用 Render API（GET）
• 响应格式：Prometheus 返回 Frame 格式，Graphite 返回简单数组格式

适用场景：

• Prometheus 更适合 Kubernetes 监控、微服务监控等多维度查询场景
• Graphite 更适合传统应用监控、固定结构指标组织场景

通过该方案，可以高效地获取 Prometheus 时序数据，为监控、分析和决策提供数据支持。