105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
|
import requests
|
|||
|
|
from typing import Optional, Dict, Any, List
|
|||
|
|
|
|||
|
|
class PrometheusSkill:
|
|||
|
|
"""
|
|||
|
|
Prometheus API Skill
|
|||
|
|
封装 Prometheus HTTP API 所有常用接口，支持 agent/LangChain/LLM 等调用。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(self, base_url: str, bearer_token: Optional[str] = None):
|
|||
|
|
self.base_url = base_url.rstrip("/")
|
|||
|
|
self.headers = {"Authorization": f"Bearer {bearer_token}"} if bearer_token else {}
|
|||
|
|
|
|||
|
|
# ========== 1. 实时查询 ==========
|
|||
|
|
def instant_query(self, promql: str, time: Optional[float] = None) -> Dict[str, Any]:
|
|||
|
|
"""执行单点PromQL查询 /api/v1/query"""
|
|||
|
|
params = {"query": promql}
|
|||
|
|
if time:
|
|||
|
|
params["time"] = time
|
|||
|
|
return self._get("/api/v1/query", params)
|
|||
|
|
|
|||
|
|
# ========== 2. 范围查询 ==========
|
|||
|
|
def range_query(self, promql: str, start: str, end: str, step: str) -> Dict[str, Any]:
|
|||
|
|
"""执行时序PromQL查询 /api/v1/query_range"""
|
|||
|
|
params = {
|
|||
|
|
"query": promql,
|
|||
|
|
"start": start,
|
|||
|
|
"end": end,
|
|||
|
|
"step": step
|
|||
|
|
}
|
|||
|
|
return self._get("/api/v1/query_range", params)
|
|||
|
|
|
|||
|
|
# ========== 3. 获取所有 metric 名字 ==========
|
|||
|
|
def list_series(self, match: List[str], start: Optional[str] = None, end: Optional[str] = None) -> Dict[str, Any]:
|
|||
|
|
"""列出匹配的时序（可用于探查所有监控项） /api/v1/series"""
|
|||
|
|
params = [("match[]", m) for m in match]
|
|||
|
|
if start: params.append(("start", start))
|
|||
|
|
if end: params.append(("end", end))
|
|||
|
|
return self._get("/api/v1/series", params)
|
|||
|
|
|
|||
|
|
def list_labels(self, start: Optional[str] = None, end: Optional[str] = None) -> Dict[str, Any]:
|
|||
|
|
"""列出所有label名 /api/v1/labels"""
|
|||
|
|
params = {}
|
|||
|
|
if start: params["start"] = start
|
|||
|
|
if end: params["end"] = end
|
|||
|
|
return self._get("/api/v1/labels", params)
|
|||
|
|
|
|||
|
|
def label_values(self, label_name: str, start: Optional[str] = None, end: Optional[str] = None) -> Dict[str, Any]:
|
|||
|
|
"""列出某个label的所有 possible value /api/v1/label/{label_name}/values"""
|
|||
|
|
params = {}
|
|||
|
|
if start: params["start"] = start
|
|||
|
|
if end: params["end"] = end
|
|||
|
|
return self._get(f"/api/v1/label/{{label_name}}/values", params)
|
|||
|
|
|
|||
|
|
# ========== 4. 获取目标/告警 配置/状态 ==========
|
|||
|
|
def get_targets(self) -> Dict[str, Any]:
|
|||
|
|
"""获取targets（监控目标） /api/v1/targets"""
|
|||
|
|
return self._get("/api/v1/targets")
|
|||
|
|
|
|||
|
|
def get_alerts(self) -> Dict[str, Any]:
|
|||
|
|
"""Prometheus当前所有告警（由Prometheus规则产生） /api/v1/alerts"""
|
|||
|
|
return self._get("/api/v1/alerts")
|
|||
|
|
|
|||
|
|
def get_rules(self) -> Dict[str, Any]:
|
|||
|
|
"""获取当前所有Prometheus规则 /api/v1/rules"""
|
|||
|
|
return self._get("/api/v1/rules")
|
|||
|
|
|
|||
|
|
def get_active_alertmanagers(self) -> Dict[str, Any]:
|
|||
|
|
"""获取当前活跃的alertmanager列表 /api/v1/alertmanagers"""
|
|||
|
|
return self._get("/api/v1/alertmanagers")
|
|||
|
|
|
|||
|
|
def get_status(self) -> Dict[str, Any]:
|
|||
|
|
"""获取Prometheus服务状态 /api/v1/status"""
|
|||
|
|
return self._get("/api/v1/status")
|
|||
|
|
|
|||
|
|
def get_flags(self) -> Dict[str, Any]:
|
|||
|
|
"""获取Prometheus启动flags /api/v1/status/flags"""
|
|||
|
|
return self._get("/api/v1/status/flags")
|
|||
|
|
|
|||
|
|
def get_config(self) -> Dict[str, Any]:
|
|||
|
|
"""获取Prometheus配置 /api/v1/status/config"""
|
|||
|
|
return self._get("/api/v1/status/config")
|
|||
|
|
|
|||
|
|
def get_runtimeinfo(self) -> Dict[str, Any]:
|
|||
|
|
"""Prometheus运行时信息 /api/v1/status/runtimeinfo"""
|
|||
|
|
return self._get("/api/v1/status/runtimeinfo")
|
|||
|
|
|
|||
|
|
def get_buildinfo(self) -> Dict[str, Any]:
|
|||
|
|
"""Prometheus构建信息 /api/v1/status/buildinfo"""
|
|||
|
|
return self._get("/api/v1/status/buildinfo")
|
|||
|
|
|
|||
|
|
# ========== 5. 内部辅助 ==========
|
|||
|
|
def _get(self, path: str, params=None) -> Dict[str, Any]:
|
|||
|
|
url = self.base_url + path
|
|||
|
|
r = requests.get(url, headers=self.headers, params=params, timeout=15)
|
|||
|
|
r.raise_for_status()
|
|||
|
|
return r.json()
|
|||
|
|
|
|||
|
|
# ==== 用法示例 ====
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
prom = PrometheusSkill(base_url="http://localhost:9090")
|
|||
|
|
print(prom.instant_query("up"))
|
|||
|
|
print(prom.get_targets())
|
|||
|
|
print(prom.list_labels())
|
|||
|
|
print(prom.get_rules())
|