托管 Prometheus
托管 Prometheus
托管 Prometheus
文档指南
请输入
产品动态
Prometheus 基础知识
用户指南
工作区
告警中心
接入指南
容器与自建集群接入
自建 Kubernetes 集群接入
最佳实践
Prometheus 数据采集与接入
资源与业务监控
工作区数据写入和查询
多渠道告警通知
API 参考
工作区
告警中心
联系人
告警规则
告警静默策略
SDK 参考
指标参考
容器服务指标
常见问题
视频帮助
- 文档首页
托管 Prometheus最佳实践工作区数据写入和查询使用 Python 语言实现工作区数据写入和查询
复制全文
下载 pdf使用 Python 语言实现工作区数据写入和查询
文档反馈
问问助手本文面向首次接入 VMP 工作区的 Python 开发者，覆盖 BasicAuth 和 AK/SK 两种鉴权方式，并提供完整示例代码，为您详细介绍如何通过 Python 语言进行工作区指标数据查询或写入的方法。
前提条件
接入前，请确认以下条件均已满足：
- 已创建托管 Prometheus 工作区，并获取查询地址 query_url 与写入地址 remote_write_url，详情请参见 获取工作区地址。
- 鉴权方式，请根据实际需要选择以下任一一种鉴权方式：
- 鉴权方式适用场景说明鉴权方式适用场景说明BasicAuth适用于工作区已提供用户名/密码、需快速完成接入验证的场景，如联调测试与示例验证。用于使用 BasicAuth 鉴权访问工作区，为用户在工作区详情中配置的 BasicAuth 账号和密码。AK/SK适用于需统一使用火山引擎签名鉴权体系的正式集成场景，支持子账号及临时凭证（需额外传入 sessionToken）访问。
- 开发环境需满足以下条件：
- 依赖项用途是否必需依赖项用途是否必需Python 3.9 及以上运行示例代码是requestsHTTP 请求基础库是volcengineAK/SK 签名鉴权仅 AK/SK 模式python-snappyRemote Write 数据压缩仅写入场景protobufRemote Write 序列化仅写入场景
- 建议按场景安装依赖：
- # 仅查询场景pip install requests volcengine# 需要写入指标时，再额外安装pip install python-snappy protobuf
操作步骤
VMP 工作区访问接口封装
将以下代码保存为 client.py，该文件提供 VMP 工作区访问的最小可用封装，查询或写入示例均基于此文件进行调用，包括：
- WorkspaceConfig：用于集中管理工作区地址、鉴权方式和超时时间等配置。
- VmpWorkspaceClient：对 Prometheus Query API 和 Remote Write 能力进行了轻量封装。
#!/usr/bin/env python3
"""VMP Workspace Python SDK: Query API + Remote Write with BasicAuth / AK-SK authentication."""
from __future__ import annotations
import base64
import os
from collections import OrderedDict
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Dict, List, Optional, Sequence, Tuple
from urllib.parse import parse_qsl, urlparse
import requests
from requests.auth import AuthBase
@dataclass
class WorkspaceConfig:
"""VMP workspace access configuration."""
query_url: str = ""
remote_write_url: str = ""
auth_mode: str = "aksk"
region: Optional[str] = None
service: str = "vmp"
access_key: Optional[str] = None
secret_key: Optional[str] = None
session_token: str = ""
username: Optional[str] = None
password: Optional[str] = None
timeout_seconds: int = 30
@classmethod
def from_env(cls) -> "WorkspaceConfig":
"""Load configuration from environment variables."""
return cls(
query_url=os.environ.get("VMP_QUERY_URL", ""),
remote_write_url=os.environ.get("VMP_REMOTE_WRITE_URL", ""),
auth_mode=os.environ.get("VMP_AUTH_MODE", "aksk").lower(),
region=os.environ.get("VOLCENGINE_REGION") or os.environ.get("VMP_REGION"),
service=os.environ.get("VMP_SERVICE", "vmp"),
access_key=os.environ.get("VOLCENGINE_ACCESS_KEY") or os.environ.get("VMP_ACCESS_KEY"),
secret_key=os.environ.get("VOLCENGINE_SECRET_KEY") or os.environ.get("VMP_SECRET_KEY"),
session_token=os.environ.get("VOLCENGINE_SESSION_TOKEN", ""),
username=os.environ.get("VMP_BASIC_AUTH_USERNAME"),
password=os.environ.get("VMP_BASIC_AUTH_PASSWORD"),
timeout_seconds=int(os.environ.get("VMP_TIMEOUT_SECONDS", "30")),
)
class BasicAuth(AuthBase):
"""BasicAuth adapter."""
def __init__(self, username: str, password: str) -> None:
token = base64.b64encode(f"{username}:{password}".encode()).decode()
self._header = f"Basic {token}"
def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
request.headers["Authorization"] = self._header
return request
class VolcSigV4Auth(AuthBase):
"""Volcengine AK/SK SignV4 authentication adapter."""
def __init__(self, access_key: str, secret_key: str, region: str, service: str = "vmp", session_token: str = "") -> None:
from volcengine.Credentials import Credentials
self._credentials = Credentials(access_key, secret_key, service, region, session_token)
def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
from volcengine.auth.SignerV4 import SignerV4
from volcengine.base.Request import Request as VolcRequest
parsed = urlparse(request.url)
volc_request = VolcRequest()
volc_request.set_shema(parsed.scheme)
volc_request.set_method(request.method.upper())
volc_request.set_connection_timeout(10)
volc_request.set_socket_timeout(10)
volc_request.set_host(parsed.netloc)
volc_request.set_path(parsed.path or "/")
if parsed.query:
volc_request.set_query(OrderedDict(parse_qsl(parsed.query, keep_blank_values=True)))
# Only pass minimal required headers for signing to avoid signature mismatch
sign_headers = OrderedDict()
sign_headers["Host"] = parsed.netloc
content_type = request.headers.get("Content-Type")
if content_type:
sign_headers["Content-Type"] = content_type
volc_request.set_headers(sign_headers)
body = request.body or b""
if isinstance(body, str):
body = body.encode("utf-8")
volc_request.set_body(body)
SignerV4.sign(volc_request, self._credentials)
for key, value in volc_request.headers.items():
request.headers[key] = value
return request
def build_vmp_auth(config: WorkspaceConfig) -> AuthBase:
"""Build auth adapter from config."""
if config.auth_mode == "basic":
if not config.username or not config.password:
raise ValueError("basic auth mode requires username and password")
return BasicAuth(config.username, config.password)
if config.auth_mode == "aksk":
if not config.access_key or not config.secret_key or not config.region:
raise ValueError("aksk auth mode requires access_key, secret_key and region")
return VolcSigV4Auth(
access_key=config.access_key,
secret_key=config.secret_key,
region=config.region,
service=config.service,
session_token=config.session_token,
)
raise ValueError(f"unsupported auth mode: {config.auth_mode}")
class PrometheusHttpApiClient:
"""Generic Prometheus HTTP Query API client (auth-agnostic)."""
def __init__(self, base_url: str, session: requests.Session, timeout_seconds: int = 30) -> None:
self.base_url = base_url.rstrip("/")
self.session = session
self.timeout = timeout_seconds
def _get(self, path: str, params: Optional[List[Tuple[str, str]]] = None) -> Dict:
response = self.session.get(self.base_url + path, params=params, timeout=self.timeout)
if not response.ok:
raise requests.HTTPError(
f"GET {path} failed: HTTP {response.status_code}, response: {response.text}",
response=response,
)
return response.json()
def query(self, promql: str, ts: Optional[datetime] = None) -> Dict:
params: List[Tuple[str, str]] = [("query", promql)]
if ts is not None:
params.append(("time", _to_rfc3339(ts)))
return self._get("/api/v1/query", params)
def query_range(self, promql: str, start: datetime, end: datetime, step: str) -> Dict:
params = [("query", promql), ("start", _to_rfc3339(start)), ("end", _to_rfc3339(end)), ("step", step)]
return self._get("/api/v1/query_range", params)
def series(self, matches: Sequence[str], start: Optional[datetime] = None, end: Optional[datetime] = None) -> Dict:
if not matches:
raise ValueError("series requires at least one match selector")
params: List[Tuple[str, str]] = [("match[]", m) for m in matches]
if start:
params.append(("start", _to_rfc3339(start)))
if end:
params.append(("end", _to_rfc3339(end)))
return self._get("/api/v1/series", params)
def labels(self, start: Optional[datetime] = None, end: Optional[datetime] = None, matches: Optional[Sequence[str]] = None) -> Dict:
params: List[Tuple[str, str]] = []
if matches:
params.extend(("match[]", m) for m in matches)
if start:
params.append(("start", _to_rfc3339(start)))
if end:
params.append(("end", _to_rfc3339(end)))
return self._get("/api/v1/labels", params)
def label_values(self, label_name: str, start: Optional[datetime] = None, end: Optional[datetime] = None, matches: Optional[Sequence[str]] = None) -> Dict:
if not label_name:
raise ValueError("label_values requires a non-empty label_name")
params: List[Tuple[str, str]] = []
if matches:
params.extend(("match[]", m) for m in matches)
if start:
params.append(("start", _to_rfc3339(start)))
if end:
params.append(("end", _to_rfc3339(end)))
return self._get(f"/api/v1/label/{label_name}/values", params)
class VmpWorkspaceClient:
"""VMP workspace client: auth + Query API + Remote Write."""
def __init__(self, config: WorkspaceConfig) -> None:
self.config = config
self.session = requests.Session()
self.session.auth = build_vmp_auth(config)
self.timeout = config.timeout_seconds
self.http_api = PrometheusHttpApiClient(config.query_url, self.session, config.timeout_seconds)
def query(self, promql: str, ts: Optional[datetime] = None) -> Dict:
return self.http_api.query(promql, ts)
def query_range(self, promql: str, start: datetime, end: datetime, step: str) -> Dict:
return self.http_api.query_range(promql, start, end, step)
def series(self, matches: Sequence[str], start: Optional[datetime] = None, end: Optional[datetime] = None) -> Dict:
return self.http_api.series(matches, start=start, end=end)
def labels(self, start: Optional[datetime] = None, end: Optional[datetime] = None, matches: Optional[Sequence[str]] = None) -> Dict:
return self.http_api.labels(start=start, end=end, matches=matches)
def label_values(self, label_name: str, start: Optional[datetime] = None, end: Optional[datetime] = None, matches: Optional[Sequence[str]] = None) -> Dict:
return self.http_api.label_values(label_name, start=start, end=end, matches=matches)
def write_samples(self, series: Sequence[Tuple[Dict[str, str], float, int]]) -> Dict:
"""Write samples via Remote Write. Each element is (labels_dict, value, timestamp_ms)."""
import snappy
remote_write_pb2 = _get_remote_write_pb2()
payload = remote_write_pb2.WriteRequest()
for labels, value, timestamp_ms in series:
ts = payload.timeseries.add()
for key, label_value in labels.items():
label = ts.labels.add()
label.name = key
label.value = label_value
sample = ts.samples.add()
sample.value = float(value)
sample.timestamp = int(timestamp_ms)
body = snappy.compress(payload.SerializeToString())
headers = {
"Content-Type": "application/x-protobuf",
"Content-Encoding": "snappy",
"X-Prometheus-Remote-Write-Version": "0.1.0",
}
response = self.session.post(self.config.remote_write_url, data=body, headers=headers, timeout=self.timeout)
if not response.ok:
raise requests.HTTPError(
f"Remote Write failed: HTTP {response.status_code}, response: {response.text}",
response=response,
)
return {"status_code": response.status_code, "body": response.text}
def _to_rfc3339(value: datetime) -> str:
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
# Embedded Prometheus Remote Write protobuf definition to avoid external _pb2.py dependency
_REMOTE_WRITE_PB2 = None
def _get_remote_write_pb2():
global _REMOTE_WRITE_PB2
if _REMOTE_WRITE_PB2 is not None:
return _REMOTE_WRITE_PB2
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf.internal import builder as _builder
import types
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(
b'\n\x1dprometheus_remote_write.proto\x12\nprometheus\":\n\x0cWriteRequest\x12*\n\ntimeseries\x18\x01 \x03(\x0b\x32\x16.prometheus.TimeSeries\"T\n\nTimeSeries\x12!\n\x06labels\x18\x01 \x03(\x0b\x32\x11.prometheus.Label\x12#\n\x07samples\x18\x02 \x03(\x0b\x32\x12.prometheus.Sample\"$\n\x05Label\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t\"*\n\x06Sample\x12\r\n\x05value\x18\x01 \x01(\x01\x12\x11\n\ttimestamp\x18\x02 \x01(\x03\x62\x06proto3'
)
_globals = {}
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'prometheus_remote_write_pb2', _globals)
module = types.SimpleNamespace(**_globals)
_REMOTE_WRITE_PB2 = module
return module
查询指标数据
调用的接口说明与示例
方法 | 接口路径 | 典型用途 | 说明 |
方法 | 接口路径 | 典型用途 | 说明 |
query(promql, ts) | /api/v1/query | 即时查询 | 查询某个时间点的当前值。 |
queryRange(promql, start, end, step) | /api/v1/query_range | 范围查询 | 查询一段时间内的趋势数据。 |
series(matches, start, end) | /api/v1/series | 查询匹配的时间序列 | 适合确认某个指标或标签组合是否存在。 |
labels(start, end, matches) | /api/v1/labels | 查询标签名 | 适合排查当前工作区有哪些可用标签。 |
labelValues(labelName, start, end, matches) | /api/v1/label/<name>/values | 查询标签值 | 适合查看某个标签下已有的取值集合。 |
说明
调用查询接口后，返回结果中的关键字段说明如下：
- status：表示请求是否成功。
- data.resultType：表示结果类型。
- data.result：包含实际的指标查询结果。
使用 BasicAuth 查询指标
from datetime import datetime, timedelta, timezone
from client import WorkspaceConfig, VmpWorkspaceClient
# 1. 构造工作区配置：
# - query_url: Prometheus Query API 地址
# - auth_mode="basic": 表示使用 BasicAuth 鉴权
# - username/password: 工作区配置的 BasicAuth 用户名和密码
config = WorkspaceConfig(
query_url="<your-query-url>",
auth_mode="basic",
username="<your-username>",
password="<your-password>",
)
# 2. 创建客户端
client = VmpWorkspaceClient(config)
# 3. 即时查询：
# 查询当前时刻 up 指标的结果
result = client.query("up")
print(result)
# 4. 范围查询：
# 查询最近 5 分钟内 up 指标的变化，每 30 秒返回一个点
end = datetime.now(tz=timezone.utc)
start = end - timedelta(minutes=5)
result = client.query_range("up", start, end, "30s")
print(result)
# 5. 查询时间序列：
# 查询符合 {__name__="up"} 条件的所有时间序列
series = client.series(['{__name__=\"up\"}'], start=start, end=end)
print(series)
# 6. 查询标签名：
# 返回指定时间范围内出现过的标签名列表
labels = client.labels(start=start, end=end)
print(labels)
# 7. 查询标签值：
# 查询 job 这个标签在指定时间范围内出现过的取值
values = client.label_values("job", start=start, end=end)
print(values)
使用 AK/SK 查询指标
from datetime import datetime, timedelta, timezone
from client import WorkspaceConfig, VmpWorkspaceClient
# 1. 构造工作区配置：
# - query_url: Prometheus Query API 地址
# - auth_mode="aksk": 表示使用火山引擎 AK/SK 签名鉴权
# - region: 工作区所在地域，例如 cn-beijing
# - access_key/secret_key: 具备 VMPQueryAccess 权限的凭证
config = WorkspaceConfig(
query_url="<your-query-url>",
auth_mode="aksk",
region="cn-beijing",
access_key="<your-access-key>",
secret_key="<your-secret-key>",
)
# 2. 创建客户端
client = VmpWorkspaceClient(config)
# 3. 即时查询：
# 查询当前时刻 up 指标的结果
result = client.query("up")
print(result)
# 4. 范围查询：
# 查询最近 5 分钟内 up 指标的变化，每 30 秒返回一个点
end = datetime.now(tz=timezone.utc)
start = end - timedelta(minutes=5)
result = client.query_range("up", start, end, "30s")
print(result)
# 5. 查询时间序列：
# 查询符合 {__name__="up"} 条件的所有时间序列
series = client.series(['{__name__=\"up\"}'], start=start, end=end)
print(series)
# 6. 查询标签名：
# 返回指定时间范围内出现过的标签名列表
labels = client.labels(start=start, end=end)
print(labels)
# 7. 查询标签值：
# 查询 job 这个标签在指定时间范围内出现过的取值
values = client.label_values("job", start=start, end=end)
print(values)
写入指标数据
使用 BasicAuth 写入指标
from datetime import datetime, timezone
from client import WorkspaceConfig, VmpWorkspaceClient
# 1. 构造工作区配置：
# - remote_write_url: Remote Write 写入地址
# - auth_mode="basic": 表示使用 BasicAuth 鉴权
# - username/password: 工作区配置的 BasicAuth 用户名和密码
config = WorkspaceConfig(
remote_write_url="<your-remote-write-url>",
auth_mode="basic",
username="<your-username>",
password="<your-password>",
)
# 2. 基于配置创建客户端
client = VmpWorkspaceClient(config)
# 3. 组装待写入的样本数据并发起写入
# write_samples 的入参是一个列表，列表中每个元素为：
# (labels_dict, value, timestamp_ms)
# - labels_dict: 指标标签，必须包含 __name__ 作为指标名
# - value: 指标值
# - timestamp_ms: 毫秒级时间戳
resp = client.write_samples([
(
{
"__name__": "demo_metric", # 指标名称
"job": "example", # 业务标签，可按需扩展
"env": "test", # 环境标签
},
42.0, # 指标值
int(datetime.now(tz=timezone.utc).timestamp() * 1000), # 当前 UTC 时间，单位毫秒
)
])
# 4. 打印写入结果
# 成功时通常返回 HTTP 状态码等信息
print(resp)
使用 AK/SK 写入指标
from datetime import datetime, timezone
from client import WorkspaceConfig, VmpWorkspaceClient
# 1. 构造工作区配置：
# - remote_write_url: Remote Write 写入地址
# - auth_mode="aksk": 表示使用火山引擎 AK/SK 签名鉴权
# - region: 工作区所在地域，例如 cn-beijing
# - access_key/secret_key: 具备 VMPRemoteWriteAccess 权限的凭证
config = WorkspaceConfig(
remote_write_url="<your-remote-write-url>",
auth_mode="aksk",
region="cn-beijing",
access_key="<your-access-key>",
secret_key="<your-secret-key>",
)
# 2. 基于配置创建客户端
client = VmpWorkspaceClient(config)
# 3. 组装待写入的样本数据并发起写入
# write_samples 的入参是一个列表，列表中每个元素为：
# (labels_dict, value, timestamp_ms)
# - labels_dict: 指标标签，必须包含 __name__ 作为指标名
# - value: 指标值
# - timestamp_ms: 毫秒级时间戳
resp = client.write_samples([
(
{
"__name__": "demo_metric", # 指标名称
"job": "example", # 业务标签，可按需扩展
"env": "test", # 环境标签
},
42.0, # 指标值
int(datetime.now(tz=timezone.utc).timestamp() * 1000), # 当前 UTC 时间，单位毫秒
)
])
# 4. 打印写入结果
# 成功时通常返回 HTTP 状态码等信息
print(resp)
常见问题与排查建议
Q1：鉴权失败，返回 401 / 403
可能原因：
- BasicAuth 用户名或密码错误。
- AK / SK / Region 配置错误。
- 当前账号或凭证缺少目标工作区的访问权限。
排查建议：
- 检查鉴权方式是否与实际配置一致。
- 确认复制的接入地址是否与工作区匹配。
Q2：查询成功但结果为空
可能原因：
- PromQL 查询未匹配到相应数据。
- 查询时间范围过短或时间点不合适。
- 写入的指标数据暂未同步至查询侧。
排查建议：
- 使用更简单的指标名。
- 将时间范围由最近 5 分钟扩大到最近 30 分钟。
- 等待 5～10 分钟再查询。
Q3：Remote Write 写入失败
可能原因：
- remote_write_url 填写错误。
- 未安装 python-snappy 或 protobuf。
- 网络策略或权限限制拦截。
排查建议：
- 确认地址是否为完整写入地址，通常以 /api/v1/write 结尾。
- 确认依赖是否安装成功。
- 检查网络策略或权限限制。
Q4：AK/SK 授权失败
可能原因：
- Region 与工作区实际地域不一致。
- 使用了临时凭证但遗漏 SessionToken。
- 服务名被误改，导致签名计算不匹配。
排查建议：
- service 保持默认值 vmp。
- 若使用临时凭证，请同时提供 AccessKey、SecretKey 和 SessionToken。
最近更新时间：2026.10.14 10:03:52
这个页面对您有帮助吗？
有用
有用
无用
无用